amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package activator
2
3import (
4 "context"
5 "errors"
6 "os/exec"
7 "reflect"
8 "strings"
9 "sync"
10 "testing"
11 "time"
12
13 "github.com/carlosframework/platform/internal/hostclaim"
14 "github.com/carlosframework/platform/internal/hostwrite"
15 "github.com/carlosframework/platform/internal/moves"
16 "github.com/carlosframework/platform/internal/provision"
17 "github.com/carlosframework/platform/internal/registry"
18)
19
20const (
21 pwHost = "shop.example.com"
22 pwOp = "mv_probewake"
23)
24
25type pwRoutes struct {
26 mu sync.Mutex
27 rt registry.Route
28}
29
30func (r *pwRoutes) Get(string) (*registry.Route, error) {
31 r.mu.Lock()
32 defer r.mu.Unlock()
33 rt := r.rt
34 return &rt, nil
35}
36
37// pwRuntime is a claim runtime that records what the activator asked of it.
38type pwRuntime struct {
39 mu sync.Mutex
40 observed hostclaim.Observed
41 calls []string
42 policies []startPolicy
43 failedFor []hostclaim.Observed
44 stop error
45 stopped int
46}
47
48func (r *pwRuntime) note(call string, p startPolicy) {
49 r.mu.Lock()
50 defer r.mu.Unlock()
51 r.calls = append(r.calls, call)
52 r.policies = append(r.policies, p)
53}
54
55func (r *pwRuntime) admitted() hostclaim.Observed {
56 o := r.observed
57 c := *o.Claim
58 h := *c.Holder
59 h.Session, h.Generation, h.WriterBox = h.Session+5, pwOp, "box"
60 c.Holder = &h
61 o.Claim, o.ETag = &c, "admitted"
62 return o
63}
64
65func (r *pwRuntime) runtime(t *testing.T) *ClaimRuntime {
66 work, _ := (hostwrite.Config{}).Select(t.Context(), pwHost, "spawn")
67 return &ClaimRuntime{Box: "box",
68 Read: func(_ context.Context, host string) (hostclaim.Observed, error) {
69 r.mu.Lock()
70 defer r.mu.Unlock()
71 if len(r.calls) > 1 {
72 return r.admitted(), nil
73 }
74 return r.observed, nil
75 },
76 StopForeign: func(context.Context, string) error { r.mu.Lock(); r.stopped++; r.mu.Unlock(); return nil },
77 Prerequisites: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) error {
78 r.note("prerequisites", p)
79 return nil
80 },
81 Admit: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) (hostclaim.Observed, error) {
82 r.note("admit", p)
83 return r.admitted(), nil
84 },
85 Prepare: func(_ context.Context, _ registry.Route, got hostclaim.Observed, p startPolicy) (ClaimStart, error) {
86 r.note("prepare", p)
87 return ClaimStart{RoleLauncher: &RoleLaunch{}, CurrentAttempt: func() hostclaim.WakeRetryAttempt { return hostclaim.WakeRetryAttempt{} }, Host: pwHost, Work: work, Identity: hostwrite.Identity{Config: "c", Release: "r"},
88 OpenProbe: func(context.Context, time.Time) (time.Time, error) { return time.Now(), nil },
89 Restore: func(context.Context) error { r.note("restore", p); return nil },
90 Materialize: func(context.Context) error { r.note("materialize", p); return r.stop },
91 Build: func() *exec.Cmd { return nil }, Ready: func(context.Context, *exec.Cmd) error { return nil },
92 Reserve: func(_ context.Context, n int) error { r.note("reserve", p); return nil },
93 Failed: func(context.Context, int) error { r.note("failed", p); return nil },
94 Healthy: func(context.Context, int, *exec.Cmd) error { return nil },
95 Replicate: func(context.Context, string) (Rep, error) { return nil, nil },
96 Publish: func(_ context.Context, o hostclaim.Observed) (hostclaim.Observed, error) { return o, nil },
97 Retain: func(Proc, Rep, error) {}}, nil
98 },
99 FailedAdmission: func(_ context.Context, o hostclaim.Observed, _ error) error {
100 r.mu.Lock()
101 r.failedFor = append(r.failedFor, o)
102 r.mu.Unlock()
103 return nil
104 }}
105}
106
107func pwFlip() hostclaim.Observed {
108 return hostclaim.Observed{Host: pwHost, ETag: "flip", Claim: &hostclaim.Claim{Host: pwHost,
109 Holder: &hostclaim.Holder{Incarnation: "inc", Location: "instances/" + pwHost, Stamp: 18, Session: 15, Generation: "src", WriterBox: "ie-box"},
110 Op: hostclaim.Op{Kind: hostclaim.KindLanding, ID: pwOp, Phase: hostclaim.PhaseLandingProbing, Exec: "box"}}}
111}
112
113func pwService(t *testing.T, r *pwRuntime) (*Service, *pwRoutes) {
114 t.Helper()
115 routes := &pwRoutes{rt: registry.Route{Host: pwHost, Kind: "instance", Status: "landing", HoldOp: pwOp, Socket: "/run/x.sock", DBPath: "/data/x.db"}}
116 a := New(Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: r.runtime(t)}, &fakeBackend{}, nil)
117 a.Routes = routes
118 return a, routes
119}
120
121func pwConfig() ProbeConfig {
122 return ProbeConfig{Host: pwHost, Op: pwOp, HoldOp: pwOp, Deadline: time.Now().Add(time.Minute),
123 Prepared: &moves.InstallManifest{Op: pwOp, Host: pwHost, N: 42}, Inputs: &provision.MoveInputs{Release: "v1"}}
124}
125
126// ProbeWake is the claim path with the probe's policy (Plan 3 Task 5a, H2):
127// the claim's own Prerequisites, Admit and admitted start sequence, each
128// given the probe policy (five starts, the 30 s window, the deadline, the
129// prepared manifest and the pins). It adds no admission or restore of its
130// own. A failure after the admission is ErrProbeFailed, the admitted tree is
131// settled through FailedAdmission, and the entry is gone.
132func TestProbeWakeRunsTheClaimPathWithTheProbePolicy(t *testing.T) {
133 stop := errors.New("materialize failed")
134 r := &pwRuntime{observed: pwFlip(), stop: stop}
135 a, _ := pwService(t, r)
136 cfg := pwConfig()
137 _, err := a.ProbeWake(t.Context(), cfg)
138 if !errors.Is(err, ErrProbeFailed) || !errors.Is(err, stop) || errors.Is(err, ErrProbeRefused) {
139 t.Fatalf("err = %v", err)
140 }
141 if got := strings.Join(r.calls, ","); got != "prerequisites,admit,prepare,restore,reserve,materialize,failed" {
142 t.Fatalf("calls: %s", got)
143 }
144 want := startPolicy{MaxStarts: 5, Window: 30 * time.Second, Deadline: cfg.Deadline, Probe: true, Prepared: cfg.Prepared, Inputs: cfg.Inputs}
145 for i, p := range r.policies {
146 if p.Gate != nil {
147 t.Fatalf("%s ran a probe under a pin gate", r.calls[i])
148 }
149 p.Gate = nil
150 if !reflect.DeepEqual(p, want) {
151 t.Fatalf("%s ran under %+v", r.calls[i], p)
152 }
153 }
154 if len(r.failedFor) != 1 || r.failedFor[0].ETag != "admitted" {
155 t.Fatalf("FailedAdmission: %+v", r.failedFor)
156 }
157 if a.Owns(pwHost) {
158 t.Fatal("a failed probe left its entry")
159 }
160}
161
162// A probe is never planned (merge of the zero-downtime scheduler, plan 5):
163// with the scheduler armed, ProbeWake still runs the landing's prepared
164// build under its own policy. The planner is not asked, nothing is pinned,
165// published or fenced for a plan, and no plan gate, selection check or
166// fallback reaches the probe's starts.
167func TestProbeWakeIsNeverPlanned(t *testing.T) {
168 stop := errors.New("materialize failed")
169 r := &pwRuntime{observed: pwFlip(), stop: stop}
170 a, _ := pwService(t, r)
171 planned := 0
172 a.Planner = func(context.Context, registry.Route) WakePlan {
173 planned++
174 return WakePlan{Refuse: "planner consulted for a probe"}
175 }
176 cfg := pwConfig()
177 _, err := a.ProbeWake(t.Context(), cfg)
178 if !errors.Is(err, ErrProbeFailed) || !errors.Is(err, stop) {
179 t.Fatalf("err = %v", err)
180 }
181 if planned != 0 {
182 t.Fatalf("the planner was consulted %d times for a probe", planned)
183 }
184 want := startPolicy{MaxStarts: 5, Window: 30 * time.Second, Deadline: cfg.Deadline, Probe: true, Prepared: cfg.Prepared, Inputs: cfg.Inputs}
185 for i, p := range r.policies {
186 if p.Gate != nil || p.PinCheck != nil || p.Fallback != nil || p.Pin != nil || !p.HealthBy.IsZero() {
187 t.Fatalf("%s ran a probe under a plan's policy: %+v", r.calls[i], p)
188 }
189 if !reflect.DeepEqual(p, want) {
190 t.Fatalf("%s ran under %+v", r.calls[i], p)
191 }
192 }
193}
194
195// TestProbeWakeRefusesUnheldRowOrForeignExec (the activator's half; the
196// owners' is TestProbeRefusesUnheldRowOrForeignExec in internal/host): the
197// row must be held with hold_op == HoldOp and the claim must name this box
198// op.exec under op Op at probing, else ErrProbeRefused before any admission.
199func TestProbeWakeRefusesUnheldRowOrForeignExec(t *testing.T) {
200 for name, tc := range map[string]struct {
201 route func(*registry.Route)
202 claim func(*hostclaim.Claim)
203 cfg func(*ProbeConfig)
204 }{
205 "a ready row": {route: func(r *registry.Route) { r.Status, r.HoldOp = "ready", "" }},
206 "held by another op": {route: func(r *registry.Route) { r.HoldOp = "mv_other" }},
207 "a foreign executor": {claim: func(c *hostclaim.Claim) { c.Op.Exec = "other-box" }},
208 "another op": {claim: func(c *hostclaim.Claim) { c.Op.ID = "mv_other" }},
209 "not probing": {claim: func(c *hostclaim.Claim) { c.Op.Phase = hostclaim.PhaseLandingCommitted }},
210 "a serving claim": {claim: func(c *hostclaim.Claim) { c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"} }},
211 "another hold op": {cfg: func(c *ProbeConfig) { c.HoldOp = "rb_other" }},
212 } {
213 t.Run(name, func(t *testing.T) {
214 r := &pwRuntime{observed: pwFlip()}
215 if tc.claim != nil {
216 tc.claim(r.observed.Claim)
217 }
218 a, routes := pwService(t, r)
219 if tc.route != nil {
220 tc.route(&routes.rt)
221 }
222 cfg := pwConfig()
223 if tc.cfg != nil {
224 tc.cfg(&cfg)
225 }
226 _, err := a.ProbeWake(t.Context(), cfg)
227 if !errors.Is(err, ErrProbeRefused) {
228 t.Fatalf("err = %v", err)
229 }
230 if len(r.calls) != 0 {
231 t.Fatalf("the claim path ran: %v", r.calls)
232 }
233 if a.Owns(pwHost) {
234 t.Fatal("a refused probe left an entry")
235 }
236 })
237 }
238 // A live entry for the host (an earlier probe's process, or a wake) is
239 // never replaced by a probe.
240 r := &pwRuntime{observed: pwFlip()}
241 a, _ := pwService(t, r)
242 a.mu.Lock()
243 a.entries[pwHost] = &actEntry{host: pwHost, state: actWaking, ready: make(chan struct{})}
244 a.mu.Unlock()
245 if _, err := a.ProbeWake(t.Context(), pwConfig()); !errors.Is(err, ErrProbeRefused) || len(r.calls) != 0 {
246 t.Fatalf("over a live entry: %v %v", err, r.calls)
247 }
248 // Malformed configurations never reach the claim.
249 for _, mutate := range []func(*ProbeConfig){
250 func(c *ProbeConfig) { c.Host = "" }, func(c *ProbeConfig) { c.Op = "" }, func(c *ProbeConfig) { c.HoldOp = "" },
251 func(c *ProbeConfig) { c.Deadline = time.Time{} },
252 } {
253 cfg := pwConfig()
254 mutate(&cfg)
255 if _, err := a.ProbeWake(t.Context(), cfg); !errors.Is(err, hostwrite.ErrUnavailable) {
256 t.Fatalf("malformed config %+v: %v", cfg, err)
257 }
258 }
259}
260
261// The cut deadline is checked before the admission, with the window not yet
262// open (brief step 4): a deadline already past is ErrAttemptOver and nothing
263// reaches the claim. A deadline that passes during the admitted sequence
264// ends it with ErrAttemptOver through ErrProbeFailed (startAdmitted's loop,
265// claimloop_linux_test.go). The 30 s window itself opens only inside
266// RoleLaunch.Start, which main still refuses
267// (TestStartAdmittedLoopWindowAndReadinessAtTheBoundary).
268func TestProbeWakeEndsAtTheCutDeadline(t *testing.T) {
269 t.Parallel()
270 r := &pwRuntime{observed: pwFlip()}
271 a, _ := pwService(t, r)
272 cfg := pwConfig()
273 cfg.Deadline = time.Now().Add(-time.Second)
274 if _, err := a.ProbeWake(t.Context(), cfg); !errors.Is(err, ErrAttemptOver) || errors.Is(err, ErrProbeFailed) || len(r.calls) != 0 {
275 t.Fatalf("err = %v calls = %v", err, r.calls)
276 }
277 // Past the deadline once admitted: the admitted sequence ends at once.
278 cfg.Deadline = time.Now().Add(100 * time.Millisecond)
279 r.stop = errors.New("unused")
280 slow := r.runtime(t)
281 admit := slow.Admit
282 slow.Admit = func(ctx context.Context, rt registry.Route, o hostclaim.Observed, p startPolicy) (hostclaim.Observed, error) {
283 time.Sleep(150 * time.Millisecond)
284 return admit(ctx, rt, o, p)
285 }
286 a.cfg.Claims = slow
287 _, err := a.ProbeWake(t.Context(), cfg)
288 if !errors.Is(err, ErrProbeFailed) || !errors.Is(err, ErrAttemptOver) {
289 t.Fatalf("err = %v", err)
290 }
291 for _, call := range r.calls {
292 if call == "reserve" {
293 t.Fatal("a start was reserved past the deadline")
294 }
295 }
296}
297
298// TestProbeAttemptEndsAtTheWindow (brief): blocked on the claim's isolated
299// root launcher, as the loop test it would extend is.
300func TestProbeAttemptEndsAtTheWindow(t *testing.T) {
301 t.Skip("blocked: RoleLaunch.Start and ClaimRoleAssembly.Preflight return ErrClaimIsolationUnavailable on main " +
302 "(claim Plan 9 Group D did not replace them), so OpenProbe and Ready never run through startAdmitted. " +
303 "When they do: OpenProbe answering a time 31 s ago, readiness answering after it, gives ErrProbeFailed and ErrAttemptOver.")
304}
305
306// The probe's entry stays held: an edge reload (RetireExcept) never stops a
307// probe entry while the registry row is still held by the probe's hold op,
308// and does once it is not (H18).
309func TestRetireExceptKeepsTheHeldProbeEntry(t *testing.T) {
310 t.Parallel()
311 a, _, _, reg := testActivator(t, time.Hour)
312 a.Routes = reg
313 held := registry.Route{Host: "one.example.com", Kind: "instance", Socket: "/tmp/nope.sock", DBPath: "/tmp/one.example.com.db", Hibernate: true}
314 if err := reg.Upsert(held); err != nil {
315 t.Fatal(err)
316 }
317 release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
318 if err != nil {
319 t.Fatal(err)
320 }
321 release()
322 if err := reg.SetHold("one.example.com", "landing", pwOp); err != nil {
323 t.Fatal(err)
324 }
325 a.mu.Lock()
326 a.entries["one.example.com"].probe = &probeEntry{policy: startPolicy{Probe: true}, op: pwOp, holdOp: pwOp}
327 a.mu.Unlock()
328
329 a.RetireExcept(map[string]bool{})
330 time.Sleep(50 * time.Millisecond)
331 if !a.Owns("one.example.com") {
332 t.Fatal("an edge reload retired the held probe's entry")
333 }
334 if err := reg.SetHold("one.example.com", "landing", "mv_other"); err != nil {
335 t.Fatal(err)
336 }
337 a.RetireExcept(map[string]bool{})
338 waitFor(t, "the entry no longer held for its probe to retire", func() bool { return !a.Owns("one.example.com") })
339}
340
341// Finding 6: a probe entry published after RetireExcept's snapshot of the
342// held probes, and before its retirement pass, is not retired: only entries
343// the snapshot validated are.
344func TestRetireExceptKeepsAProbeInsertedBetweenTheLocks(t *testing.T) {
345 a, _, _, reg := testActivator(t, time.Hour)
346 a.Routes = reg
347 held := registry.Route{Host: "two.example.com", Kind: "instance", Socket: "/tmp/nope.sock", DBPath: "/tmp/two.example.com.db", Hibernate: true}
348 if err := reg.Upsert(held); err != nil {
349 t.Fatal(err)
350 }
351 if err := reg.SetHold("two.example.com", "landing", pwOp); err != nil {
352 t.Fatal(err)
353 }
354 inserted := &actEntry{host: "two.example.com", route: held, state: actAwake, ready: make(chan struct{}), proc: &fakeProc{}, rep: &fakeRep{},
355 probe: &probeEntry{policy: startPolicy{Probe: true}, op: pwOp, holdOp: pwOp}}
356 close(inserted.ready)
357 retireBetweenLocks = func() {
358 a.mu.Lock()
359 a.entries[inserted.host] = inserted
360 a.mu.Unlock()
361 }
362 t.Cleanup(func() { retireBetweenLocks = nil })
363 a.RetireExcept(map[string]bool{})
364 retireBetweenLocks = nil
365 a.mu.Lock()
366 state := inserted.state
367 a.mu.Unlock()
368 if state != actAwake || !a.Owns(inserted.host) {
369 t.Fatal("a held probe inserted between the locks was retired")
370 }
371 // The next reload validates it like any other.
372 a.RetireExcept(map[string]bool{})
373 if !a.Owns(inserted.host) {
374 t.Fatal("a held probe was retired on the next reload")
375 }
376}
377
378// Finding 7: a cancel that lands after the probe's entry is published and
379// before its wake starts still stops the wake: nothing is admitted.
380func TestProbeWakeCancelBeforeTheWakeStarts(t *testing.T) {
381 r := &pwRuntime{observed: pwFlip()}
382 a, _ := pwService(t, r)
383 rt := a.cfg.Claims
384 prereq := rt.Prerequisites
385 rt.Prerequisites = func(ctx context.Context, route registry.Route, o hostclaim.Observed, p startPolicy) error {
386 if err := ctx.Err(); err != nil {
387 return err
388 }
389 return prereq(ctx, route, o, p)
390 }
391 ctx, cancel := context.WithCancel(t.Context())
392 probeWakePublished = cancel
393 t.Cleanup(func() { probeWakePublished = nil })
394 _, err := a.ProbeWake(ctx, pwConfig())
395 if !errors.Is(err, context.Canceled) {
396 t.Fatalf("err = %v", err)
397 }
398 for _, call := range r.calls {
399 if call == "admit" || call == "prerequisites" {
400 t.Fatalf("the cancelled probe ran %v", r.calls)
401 }
402 }
403 if a.Owns(pwHost) {
404 t.Fatal("the cancelled probe left its entry")
405 }
406}
407
408// Task 6 fix round 1: LiveProbe answers the awake entry a passing probe left,
409// with its op and admitted session; nothing for a waking entry, an ordinary
410// wake, or a probe that never passed.
411func TestLiveProbeAnswersThePassingProbeEntry(t *testing.T) {
412 a, _, _, _ := testActivator(t, time.Hour)
413 admitted := hostclaim.Observed{Host: pwHost, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Session: 20, Stamp: 18, Generation: pwOp}}}
414 probe := &probeEntry{op: pwOp, holdOp: pwOp, admitted: &admitted}
415 a.mu.Lock()
416 a.entries[pwHost] = &actEntry{host: pwHost, state: actAwake, probe: probe}
417 a.entries["plain.example.com"] = &actEntry{host: "plain.example.com", state: actAwake}
418 a.entries["waking.example.com"] = &actEntry{host: "waking.example.com", state: actWaking, probe: &probeEntry{op: pwOp, admitted: &admitted}}
419 a.entries["unpassed.example.com"] = &actEntry{host: "unpassed.example.com", state: actAwake, probe: &probeEntry{op: pwOp}}
420 a.mu.Unlock()
421 op, res, ok := a.LiveProbe(pwHost)
422 if !ok || op != pwOp || res != (ProbeResult{Session: 20, Stamp: 18, Generation: pwOp}) {
423 t.Fatalf("the passing probe: %q %+v %v", op, res, ok)
424 }
425 for _, host := range []string{"plain.example.com", "waking.example.com", "unpassed.example.com", "absent.example.com"} {
426 if _, _, ok := a.LiveProbe(host); ok {
427 t.Fatalf("%s answered", host)
428 }
429 }
430}
431
432// pwHeldEntry is an awake probe entry for one.example.com, held by pwOp.
433func pwHeldEntry(t *testing.T, a *Service) *actEntry {
434 t.Helper()
435 admitted := hostclaim.Observed{Host: "one.example.com", Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Session: 20, Stamp: 18, Generation: pwOp}}}
436 e := &actEntry{host: "one.example.com", route: testRoute("one.example.com"), state: actAwake, ready: make(chan struct{}), proc: &fakeProc{}, rep: &fakeRep{},
437 probe: &probeEntry{policy: startPolicy{Probe: true}, op: pwOp, holdOp: pwOp, admitted: &admitted}}
438 close(e.ready)
439 a.mu.Lock()
440 a.entries[e.host] = e
441 a.mu.Unlock()
442 return e
443}
444
445// Task 7: the target's open marks the live probe entry for its op open;
446// nothing else is marked.
447func TestOpenProbeEntryMarksTheLiveProbe(t *testing.T) {
448 a, _, _, _ := testActivator(t, time.Hour)
449 e := pwHeldEntry(t, a)
450 if a.OpenProbeEntry("one.example.com", "mv_other") || a.OpenProbeEntry("absent.example.com", pwOp) {
451 t.Fatal("another op's or an absent entry was marked")
452 }
453 if !a.OpenProbeEntry("one.example.com", pwOp) {
454 t.Fatal("the probe's entry was not marked")
455 }
456 a.mu.Lock()
457 phase := e.probe.phase
458 a.mu.Unlock()
459 if phase != probeOpening {
460 t.Fatal("not marked opening")
461 }
462}
463
464// Task 7, the open race (deterministic): an edge reload that read the routes
465// before the hold was released, and retires after it, keeps the probe entry
466// the open marked: its row is ready, and the next reload manages it. Once
467// the edge manages it, it is an ordinary serving entry.
468func TestRetireExceptKeepsTheOpenedProbeAcrossTheOpen(t *testing.T) {
469 t.Parallel()
470 a, _, _, reg := testActivator(t, time.Hour)
471 a.Routes = reg
472 if err := reg.Upsert(testRoute("one.example.com")); err != nil {
473 t.Fatal(err)
474 }
475 if err := reg.SetHold("one.example.com", "landing", pwOp); err != nil {
476 t.Fatal(err)
477 }
478 e := pwHeldEntry(t, a)
479 // The reload's managed set was read while the row was held.
480 managed := map[string]bool{}
481 if !a.OpenProbeEntry("one.example.com", pwOp) {
482 t.Fatal("not marked")
483 }
484 if ok, err := reg.ReleaseHold("one.example.com", pwOp); err != nil || !ok {
485 t.Fatalf("release: %v %v", ok, err)
486 }
487 a.RetireExcept(managed)
488 a.mu.Lock()
489 state := e.state
490 a.mu.Unlock()
491 if state != actAwake || !a.Owns("one.example.com") {
492 t.Fatal("the opened probe's entry was retired across the open")
493 }
494 // The next reload manages the ready row: the entry is ordinary from here.
495 a.RetireExcept(map[string]bool{"one.example.com": true})
496 a.mu.Lock()
497 taken := e.probe.managed
498 a.mu.Unlock()
499 if !taken {
500 t.Fatal("a managed entry is still a probe")
501 }
502 // An entry held by the probe but never opened, whose row is ready, is
503 // not the probe's: it is retired as before.
504 a2, _, _, reg2 := testActivator(t, time.Hour)
505 a2.Routes = reg2
506 if err := reg2.Upsert(testRoute("one.example.com")); err != nil {
507 t.Fatal(err)
508 }
509 pwHeldEntry(t, a2)
510 a2.RetireExcept(map[string]bool{})
511 waitFor(t, "the unopened probe on a ready row to retire", func() bool { return !a2.Owns("one.example.com") })
512}
513
514// Task 7, the idle stop: a held probe entry gets no traffic, so the idle
515// sweep must not stop it before the open; once opened it idles as any entry.
516func TestIdleSweepSparesAHeldProbe(t *testing.T) {
517 t.Parallel()
518 a, _, clk, _ := testActivator(t, time.Minute)
519 e := pwHeldEntry(t, a)
520 a.mu.Lock()
521 e.lastActive = clk.now().Add(-time.Hour)
522 a.mu.Unlock()
523 a.sweepOnce()
524 a.mu.Lock()
525 state := e.state
526 a.mu.Unlock()
527 if state != actAwake {
528 t.Fatal("the idle sweep stopped a held probe")
529 }
530 if !a.OpenProbeEntry("one.example.com", pwOp) || !a.ConfirmProbeOpened("one.example.com", pwOp) {
531 t.Fatal("not opened")
532 }
533 a.sweepOnce()
534 waitFor(t, "the opened probe to idle out", func() bool { return !a.Owns("one.example.com") })
535}
536
537// Task 7 fix round 1, finding 2: the open lands between RetireExcept's probe
538// snapshot (not opened) and its registry read (ready, unheld). The
539// registry's answer is older than the entry's open, so the entry is kept,
540// not retired as a probe that never opened.
541func TestRetireExceptRevalidatesTheOpenUnderTheFinalLock(t *testing.T) {
542 a, _, _, reg := testActivator(t, time.Hour)
543 a.Routes = reg
544 if err := reg.Upsert(testRoute("one.example.com")); err != nil {
545 t.Fatal(err)
546 }
547 if err := reg.SetHold("one.example.com", "landing", pwOp); err != nil {
548 t.Fatal(err)
549 }
550 e := pwHeldEntry(t, a)
551 retireBeforeRegistryRead = func() {
552 if !a.OpenProbeEntry("one.example.com", pwOp) {
553 t.Error("not marked")
554 }
555 if ok, err := reg.ReleaseHold("one.example.com", pwOp); err != nil || !ok {
556 t.Errorf("release: %v %v", ok, err)
557 }
558 }
559 t.Cleanup(func() { retireBeforeRegistryRead = nil })
560 a.RetireExcept(map[string]bool{})
561 retireBeforeRegistryRead = nil
562 a.mu.Lock()
563 state := e.state
564 a.mu.Unlock()
565 if state != actAwake || !a.Owns("one.example.com") {
566 t.Fatal("the probe opened during the retirement was retired")
567 }
568}
569
570// Final review, ruling 2: two reloads overlap. The older pass snapshots the
571// held probe, the open lands, and a newer pass, whose managed set includes
572// the host, takes the entry over as ordinary serving, all before the older
573// pass reads the registry. Neither pass panics, and the older one does not
574// retire the serving entry.
575func TestRetireExceptOverlappingReloadsAcrossTheOpen(t *testing.T) {
576 a, _, _, reg := testActivator(t, time.Hour)
577 a.Routes = reg
578 if err := reg.Upsert(testRoute("one.example.com")); err != nil {
579 t.Fatal(err)
580 }
581 if err := reg.SetHold("one.example.com", "landing", pwOp); err != nil {
582 t.Fatal(err)
583 }
584 e := pwHeldEntry(t, a)
585 retireBeforeRegistryRead = func() {
586 retireBeforeRegistryRead = nil // the newer pass runs without the pause
587 if !a.OpenProbeEntry("one.example.com", pwOp) {
588 t.Error("not marked")
589 }
590 if ok, err := reg.ReleaseHold("one.example.com", pwOp); err != nil || !ok {
591 t.Errorf("release: %v %v", ok, err)
592 }
593 a.ConfirmProbeOpened("one.example.com", pwOp)
594 a.RetireExcept(map[string]bool{"one.example.com": true})
595 }
596 t.Cleanup(func() { retireBeforeRegistryRead = nil })
597 a.RetireExcept(map[string]bool{})
598 a.mu.Lock()
599 state := e.state
600 a.mu.Unlock()
601 if state != actAwake || !a.Owns("one.example.com") {
602 t.Fatal("the older reload retired the serving probe")
603 }
604}
605
606// Final review, ruling 3: the mark is "opening"; idle protection ends only
607// once the release is confirmed. A failed release leaves the entry
608// protected through an idle sweep; a retried open that succeeds ends it.
609func TestIdleProtectionHoldsUntilTheReleaseIsConfirmed(t *testing.T) {
610 t.Parallel()
611 a, _, clk, _ := testActivator(t, time.Minute)
612 e := pwHeldEntry(t, a)
613 a.mu.Lock()
614 e.lastActive = clk.now().Add(-time.Hour)
615 a.mu.Unlock()
616 if !a.OpenProbeEntry("one.example.com", pwOp) {
617 t.Fatal("not marked")
618 }
619 // The release failed: no confirmation.
620 a.sweepOnce()
621 a.mu.Lock()
622 state := e.state
623 a.mu.Unlock()
624 if state != actAwake {
625 t.Fatal("an opening probe whose release failed was idled")
626 }
627 // The retried open: marked again, released, confirmed.
628 if !a.OpenProbeEntry("one.example.com", pwOp) || !a.ConfirmProbeOpened("one.example.com", pwOp) {
629 t.Fatal("the retried open")
630 }
631 a.sweepOnce()
632 waitFor(t, "the opened probe to idle out", func() bool { return !a.Owns("one.example.com") })
633}
634