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 "fmt"
7 "os"
8 "os/exec"
9 "path/filepath"
10 "reflect"
11 "sync"
12 "testing"
13 "time"
14
15 "github.com/carlosframework/platform/internal/hostbox"
16 "github.com/carlosframework/platform/internal/hostclaim"
17 "github.com/carlosframework/platform/internal/hostwrite"
18 "github.com/carlosframework/platform/internal/moves"
19 "github.com/carlosframework/platform/internal/provision"
20 "github.com/carlosframework/platform/internal/registry"
21)
22
23// The recovery probe entry (claim Plan 10 J2, moves Plan 4 Task 9), driven
24// through the claim runtime the host assembles: K6, Prerequisites, Admit,
25// Prepare and the start loop. RoleLaunch.Start still refuses every launch
26// (claim Plan 9 Group D's isolated root launcher), so a full run ends in that
27// refusal as an ordinary failed start; retention of a healthy start is driven
28// through finishRecoveryProbe with the instance a healthy startAdmitted yields.
29
30const probeHost = "x.example"
31
32func probeClaim(kind, phase, exec string) *hostclaim.Claim {
33 return &hostclaim.Claim{
34 Holder: &hostclaim.Holder{Incarnation: "inc", Stamp: 1, Session: 1, Generation: "one", Location: "instances/x.example/inc", WriterBox: "box"},
35 Op: hostclaim.Op{Kind: kind, ID: "move", Phase: phase, Exec: exec},
36 }
37}
38
39func probeObserved(c *hostclaim.Claim) hostclaim.Observed {
40 return hostclaim.Observed{Host: probeHost, ETag: "one", Claim: c}
41}
42
43func probeAdmitted(o hostclaim.Observed) hostclaim.Observed {
44 c := *o.Claim
45 h := *c.Holder
46 h.Session, h.Generation = h.Session+1, "two"
47 c.Holder = &h
48 return hostclaim.Observed{Host: o.Host, ETag: "two", Claim: &c}
49}
50
51func probeInputs() *provision.MoveInputs {
52 return &provision.MoveInputs{Release: "release", OverlayETag: "overlay", BundleETags: map[string]string{"main": "etag"}}
53}
54
55func probePolicy(in *provision.MoveInputs) StartPolicy {
56 return StartPolicy{MaxStarts: 1, Window: 30 * time.Second, Probe: true, Inputs: in}
57}
58
59// probeRun records what the claim runtime saw, in order.
60type probeRun struct {
61 mu sync.Mutex
62 steps []string
63 policies []StartPolicy
64 failedAdmission *hostclaim.Observed
65 reserved, failed, stopped int
66}
67
68func (r *probeRun) step(name string, p StartPolicy) {
69 r.mu.Lock()
70 defer r.mu.Unlock()
71 r.steps = append(r.steps, name)
72 r.policies = append(r.policies, p)
73}
74
75func (r *probeRun) seen() []string {
76 r.mu.Lock()
77 defer r.mu.Unlock()
78 return append([]string(nil), r.steps...)
79}
80
81// probeService wires a Service whose claim runtime reads obs, admits it to
82// the next session and prepares a start that reaches RoleLaunch.Start.
83func probeService(t *testing.T, obs hostclaim.Observed) (*Service, *ClaimRuntime, *probeRun) {
84 t.Helper()
85 a, _, _, _ := testActivator(t, time.Hour)
86 run := &probeRun{}
87 admitted := probeAdmitted(obs)
88 work, _ := (hostwrite.Config{}).Select(t.Context(), probeHost, "spawn")
89 reads := 0
90 r := &ClaimRuntime{
91 RecoveryAttemptsReady: func() bool { return true },
92 Box: "box",
93 // The wake's first read sees obs; every later one (startAdmitted's
94 // fresh read) sees the admission.
95 Read: func(_ context.Context, host string) (hostclaim.Observed, error) {
96 run.step("read", StartPolicy{})
97 run.mu.Lock()
98 defer run.mu.Unlock()
99 if reads++; reads > 1 {
100 return admitted, nil
101 }
102 return obs, nil
103 },
104 StopForeign: func(context.Context, string) error { run.step("stop-foreign", StartPolicy{}); return nil },
105 Prerequisites: func(_ context.Context, rt registry.Route, o hostclaim.Observed, p StartPolicy) error {
106 run.step("prerequisites", p)
107 return nil
108 },
109 Admit: func(_ context.Context, _ registry.Route, o hostclaim.Observed, p StartPolicy) (hostclaim.Observed, error) {
110 run.step("admit", p)
111 return admitted, nil
112 },
113 Prepare: func(_ context.Context, _ registry.Route, o hostclaim.Observed, p StartPolicy) (ClaimStart, error) {
114 run.step("prepare", p)
115 return ClaimStart{RoleLauncher: &RoleLaunch{}, CurrentAttempt: func() hostclaim.WakeRetryAttempt { return hostclaim.WakeRetryAttempt{} }, Host: probeHost, Work: work, Identity: hostwrite.Identity{Config: "c", Release: "r"},
116 OpenProbe: func(context.Context, time.Time) (time.Time, error) { return time.Now(), nil },
117 Restore: func(context.Context) error { return nil },
118 Materialize: func(context.Context) error { return nil },
119 Build: func() *exec.Cmd { return nil },
120 Ready: func(context.Context, *exec.Cmd) error { return nil },
121 Reserve: func(context.Context, int) error { run.mu.Lock(); run.reserved++; run.mu.Unlock(); return nil },
122 Failed: func(context.Context, int) error { run.mu.Lock(); run.failed++; run.mu.Unlock(); return nil },
123 Healthy: func(context.Context, int, *exec.Cmd) error { return nil },
124 Replicate: func(context.Context, string) (Rep, error) { return &fakeRep{}, nil },
125 Publish: func(_ context.Context, o hostclaim.Observed) (hostclaim.Observed, error) { return o, nil },
126 Retain: func(Proc, Rep, error) { t.Error("a refused start is not retained") },
127 }, nil
128 },
129 FailedAdmission: func(_ context.Context, o hostclaim.Observed, _ error) error {
130 run.mu.Lock()
131 run.failedAdmission = &o
132 run.mu.Unlock()
133 return nil
134 },
135 }
136 a.cfg.HostWrites = hostwrite.Config{Enabled: true}
137 a.cfg.Claims = r
138 return a, r, run
139}
140
141func TestRecoveryProbeRunsTheClaimRuntimeWithTheProbePolicyAndExactInputs(t *testing.T) {
142 t.Parallel()
143 for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
144 t.Run(kind, func(t *testing.T) {
145 obs := probeObserved(probeClaim(kind, hostclaim.PhaseProbing, "box"))
146 a, _, run := probeService(t, obs)
147 in := probeInputs()
148 want := *in
149 err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(in))
150 // The start reached RoleLaunch.Start, which refuses until Group D:
151 // an honest failed start, never a success.
152 if !errors.Is(err, ErrClaimIsolationUnavailable) {
153 t.Fatalf("err = %v, want the launcher's refusal", err)
154 }
155 steps := run.seen()
156 if len(steps) < 4 || steps[0] != "read" || steps[1] != "prerequisites" || steps[2] != "admit" || steps[len(steps)-1] != "prepare" {
157 t.Fatalf("steps = %v", steps)
158 }
159 run.mu.Lock()
160 defer run.mu.Unlock()
161 for i, p := range run.policies {
162 if run.steps[i] == "read" || run.steps[i] == "stop-foreign" {
163 continue
164 }
165 if !p.Probe || !reflect.DeepEqual(p.Inputs, in) || p.MaxStarts != 1 || p.Window != 30*time.Second || p.Prepared != nil {
166 t.Fatalf("%s saw %+v, want the caller's probe policy", run.steps[i], p)
167 }
168 }
169 if in.Release != want.Release || in.OverlayETag != want.OverlayETag || in.BundleETags["main"] != "etag" {
170 t.Fatalf("inputs changed: %+v", in)
171 }
172 if run.reserved != 1 || run.failed != 1 {
173 t.Fatalf("reserved=%d failed=%d, want one refused start", run.reserved, run.failed)
174 }
175 if run.failedAdmission == nil || run.failedAdmission.Claim.Holder.Session != 2 {
176 t.Fatalf("the failed admission was not handed back: %+v", run.failedAdmission)
177 }
178 if a.Owns(probeHost) {
179 t.Fatal("a failed probe left an entry")
180 }
181 })
182 }
183}
184
185func TestRecoveryProbeK6StopsForeignBeforeAnyAdmission(t *testing.T) {
186 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "elsewhere"))
187 a, _, run := probeService(t, obs)
188 err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(probeInputs()))
189 if !errors.Is(err, hostbox.ErrChanged) {
190 t.Fatalf("err = %v", err)
191 }
192 if got := run.seen(); len(got) != 2 || got[0] != "read" || got[1] != "stop-foreign" {
193 t.Fatalf("steps = %v, want read then stop-foreign only", got)
194 }
195 if a.Owns(probeHost) {
196 t.Fatal("entry left")
197 }
198}
199
200func TestRecoveryProbeRefusesWrongKindOrPhaseBeforePrerequisites(t *testing.T) {
201 t.Parallel()
202 for _, c := range []*hostclaim.Claim{
203 probeClaim(hostclaim.KindRestoring, hostclaim.PhaseClaimed, "box"),
204 probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbed, "box"),
205 probeClaim(hostclaim.KindReverting, hostclaim.PhaseOpened, "box"),
206 probeClaim(hostclaim.KindLanding, "probing", "box"),
207 {Holder: &hostclaim.Holder{Incarnation: "inc", Session: 1, Generation: "one", Location: "l", WriterBox: "box"}, Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}},
208 } {
209 t.Run(c.Op.Kind+"/"+c.Op.Phase, func(t *testing.T) {
210 a, _, run := probeService(t, probeObserved(c))
211 err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(probeInputs()))
212 if !errors.Is(err, hostbox.ErrChanged) {
213 t.Fatalf("err = %v", err)
214 }
215 if got := run.seen(); len(got) != 1 || got[0] != "read" {
216 t.Fatalf("steps = %v, want the read alone", got)
217 }
218 if a.Owns(probeHost) {
219 t.Fatal("entry left")
220 }
221 })
222 }
223}
224
225func TestRecoveryProbeRefusesMissingRuntimeOrPolicyBeforeAnyRead(t *testing.T) {
226 t.Parallel()
227 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
228 in := probeInputs()
229 runtimes := map[string]func(*Service, *ClaimRuntime){
230 "host writes off": func(a *Service, _ *ClaimRuntime) { a.cfg.HostWrites.Enabled = false },
231 "no claims": func(a *Service, _ *ClaimRuntime) { a.cfg.Claims = nil },
232 "no box": func(_ *Service, r *ClaimRuntime) { r.Box = "" },
233 "no stop foreign": func(_ *Service, r *ClaimRuntime) { r.StopForeign = nil },
234 "no prerequisites": func(_ *Service, r *ClaimRuntime) { r.Prerequisites = nil },
235 "no admit": func(_ *Service, r *ClaimRuntime) { r.Admit = nil },
236 "no prepare": func(_ *Service, r *ClaimRuntime) { r.Prepare = nil },
237 "no failed admission": func(_ *Service, r *ClaimRuntime) { r.FailedAdmission = nil },
238 }
239 for name, breakIt := range runtimes {
240 t.Run(name, func(t *testing.T) {
241 a, r, run := probeService(t, obs)
242 breakIt(a, r)
243 if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(in)); !errors.Is(err, hostwrite.ErrUnavailable) {
244 t.Fatalf("err = %v", err)
245 }
246 if got := run.seen(); len(got) != 0 || a.Owns(probeHost) {
247 t.Fatalf("steps = %v owns=%v", got, a.Owns(probeHost))
248 }
249 })
250 }
251 policies := map[string]StartPolicy{
252 "ordinary": {MaxStarts: 1, Window: 30 * time.Second, Inputs: in},
253 "no inputs": {MaxStarts: 1, Window: 30 * time.Second, Probe: true},
254 "prepared": {MaxStarts: 1, Window: 30 * time.Second, Probe: true, Inputs: in, Prepared: &moves.InstallManifest{Op: "move"}},
255 "no starts": {Window: 30 * time.Second, Probe: true, Inputs: in},
256 "too many": {MaxStarts: 6, Window: 30 * time.Second, Probe: true, Inputs: in},
257 "no window": {MaxStarts: 1, Probe: true, Inputs: in},
258 "window too big": {MaxStarts: 1, Window: 31 * time.Second, Probe: true, Inputs: in},
259 }
260 for name, p := range policies {
261 t.Run(name, func(t *testing.T) {
262 a, _, run := probeService(t, obs)
263 if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, p); !errors.Is(err, hostbox.ErrUnsupported) {
264 t.Fatalf("err = %v", err)
265 }
266 if got := run.seen(); len(got) != 0 || a.Owns(probeHost) {
267 t.Fatalf("steps = %v", got)
268 }
269 })
270 }
271 a, _, run := probeService(t, obs)
272 if err := a.RecoveryProbe(t.Context(), registry.Route{}, probePolicy(in)); !errors.Is(err, hostbox.ErrUnsupported) || len(run.seen()) != 0 {
273 t.Fatalf("no host: %v", err)
274 }
275}
276
277func TestRecoveryProbeRefusesAnOrdinaryEntryAndShutdown(t *testing.T) {
278 t.Parallel()
279 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
280 for _, state := range []actState{actWaking, actAwake, actStopping} {
281 a, _, run := probeService(t, obs)
282 a.entries[probeHost] = &actEntry{host: probeHost, state: state, ready: make(chan struct{}), stopped: make(chan struct{})}
283 if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(probeInputs())); !errors.Is(err, hostbox.ErrChanged) {
284 t.Fatalf("state %v: %v", state, err)
285 }
286 if got := run.seen(); len(got) != 0 {
287 t.Fatalf("state %v: steps %v", state, got)
288 }
289 }
290 a, _, run := probeService(t, obs)
291 a.shuttingDown = true
292 if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(probeInputs())); !errors.Is(err, ErrFenced) || len(run.seen()) != 0 {
293 t.Fatalf("shutdown: %v", err)
294 }
295}
296
297// One probe per host: an identical second call joins the one in flight and
298// shares its outcome; a different policy is refused; a request never joins.
299func TestRecoveryProbeCoalescesTheSamePolicyAndRefusesOthers(t *testing.T) {
300 t.Parallel()
301 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
302 a, r, run := probeService(t, obs)
303 entered, release := make(chan struct{}), make(chan struct{})
304 var once sync.Once
305 refuse := errors.New("prerequisites refused")
306 r.Prerequisites = func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p StartPolicy) error {
307 run.step("prerequisites", p)
308 once.Do(func() { close(entered) })
309 <-release
310 return refuse
311 }
312 in := probeInputs()
313 first := make(chan error, 1)
314 go func() {
315 first <- a.RecoveryProbe(context.Background(), registry.Route{Host: probeHost}, probePolicy(in))
316 }()
317 <-entered
318
319 other := probeInputs()
320 other.Release = "other"
321 if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(other)); !errors.Is(err, hostbox.ErrChanged) {
322 t.Fatalf("different inputs: %v", err)
323 }
324 longer := probePolicy(in)
325 longer.MaxStarts = 2
326 if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, longer); !errors.Is(err, hostbox.ErrChanged) {
327 t.Fatalf("different starts: %v", err)
328 }
329 if _, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost}); !errors.Is(err, ErrFenced) {
330 t.Fatalf("a request joined the probe: %v", err)
331 }
332 if err := a.BeginSidecarRequest(t.Context(), registry.Route{Host: probeHost}); !errors.Is(err, ErrFenced) {
333 t.Fatalf("a sidecar request joined the probe: %v", err)
334 }
335 if a.Bounce(probeHost) {
336 t.Fatal("bounce stopped the probe")
337 }
338
339 same := probeInputs()
340 second := make(chan error, 1)
341 go func() {
342 second <- a.RecoveryProbe(context.Background(), registry.Route{Host: probeHost}, probePolicy(same))
343 }()
344 time.Sleep(100 * time.Millisecond) // let the second call reach the entry
345 close(release)
346 if err := <-first; !errors.Is(err, refuse) {
347 t.Fatalf("first: %v", err)
348 }
349 if err := <-second; !errors.Is(err, refuse) {
350 t.Fatalf("joined: %v", err)
351 }
352 n := 0
353 for _, s := range run.seen() {
354 if s == "prerequisites" {
355 n++
356 }
357 }
358 if n != 1 || a.Owns(probeHost) {
359 t.Fatalf("prerequisites ran %d times, owns=%v", n, a.Owns(probeHost))
360 }
361}
362
363// A maintenance fence installed while the probe runs cancels it, exactly
364// as it cancels an ordinary wake.
365func TestRecoveryProbeIsCancelledByAFenceInstalledMidProbe(t *testing.T) {
366 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
367 a, r, run := probeService(t, obs)
368 entered := make(chan struct{})
369 r.Prerequisites = func(ctx context.Context, _ registry.Route, _ hostclaim.Observed, p StartPolicy) error {
370 run.step("prerequisites", p)
371 close(entered)
372 <-ctx.Done()
373 return ctx.Err()
374 }
375 done := make(chan error, 1)
376 go func() {
377 done <- a.RecoveryProbe(context.Background(), registry.Route{Host: probeHost}, probePolicy(probeInputs()))
378 }()
379 <-entered
380 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
381 defer cancel()
382 if _, err := a.FenceAndStop(ctx, probeHost, 7); err != nil && !errors.Is(err, ErrStopUnproven) {
383 t.Fatal(err)
384 }
385 if err := <-done; !errors.Is(err, context.Canceled) {
386 t.Fatalf("err = %v", err)
387 }
388 if a.Owns(probeHost) {
389 t.Fatal("entry left")
390 }
391}
392
393// heldProbe installs a probe entry as RecoveryProbe does and finishes it with
394// the instance a healthy startAdmitted yields.
395func heldProbe(t *testing.T, a *Service, fence uint64, admitted hostclaim.Observed, in *provision.MoveInputs) (*actEntry, *fakeProc, *fakeRep, error) {
396 t.Helper()
397 p := probePolicy(in)
398 p.RecoveryFence = fence
399 e := &actEntry{host: probeHost, route: registry.Route{Host: probeHost}, state: actWaking, ready: make(chan struct{}), recoveryPolicy: &p, probeHeld: true}
400 a.mu.Lock()
401 a.entries[probeHost] = e
402 a.mu.Unlock()
403 e.probeAdmitted = admitted
404 proc, rep := &fakeProc{}, &fakeRep{}
405 err := a.finishRecoveryProbe(e, fence, &wokenInstance{proc: proc, rep: rep, spawnVersion: "v", stopOutcome: &stopObservation{}}, nil)
406 return e, proc, rep, err
407}
408
409func TestRecoveryProbeRetainsTheHealthyProcessHeldUntilOpened(t *testing.T) {
410 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
411 admitted := probeAdmitted(obs)
412 a, r, _ := probeService(t, obs)
413 // The rollback's fence stays installed through the recovery.
414 a.fences = map[string]uint64{probeHost: 9}
415 in := probeInputs()
416 e, proc, _, err := heldProbe(t, a, 9, admitted, in)
417 if err != nil {
418 t.Fatal(err)
419 }
420 a.mu.Lock()
421 state, held, kept := e.state, e.probeHeld, e.proc == Proc(proc)
422 a.mu.Unlock()
423 if state != actAwake || !held || !kept || proc.isStopped() {
424 t.Fatalf("state=%v held=%v kept=%v stopped=%v", state, held, kept, proc.isStopped())
425 }
426 if _, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost}); !errors.Is(err, ErrFenced) {
427 t.Fatalf("a request reached the held probe: %v", err)
428 }
429 if a.Bounce(probeHost) || proc.isStopped() {
430 t.Fatal("bounce stopped the held probe")
431 }
432
433 // A replayed call for the same session answers from the retained process.
434 r.Read = func(context.Context, string) (hostclaim.Observed, error) { return admitted, nil }
435 if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, *e.recoveryPolicy); err != nil {
436 t.Fatalf("replay: %v", err)
437 }
438 moved := probeAdmitted(admitted)
439 r.Read = func(context.Context, string) (hostclaim.Observed, error) { return moved, nil }
440 if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, *e.recoveryPolicy); !errors.Is(err, hostbox.ErrChanged) {
441 t.Fatalf("replay on another session: %v", err)
442 }
443 proc.exit()
444 r.Read = func(context.Context, string) (hostclaim.Observed, error) { return admitted, nil }
445 if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, *e.recoveryPolicy); !errors.Is(err, hostbox.ErrChanged) {
446 t.Fatalf("replay on an exited process: %v", err)
447 }
448}
449
450func TestRecoveryProbeStopsAHealthyStartWhenTheFenceMoved(t *testing.T) {
451 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
452 for name, now := range map[string]map[string]uint64{"installed": {probeHost: 3}, "replaced": {probeHost: 12}} {
453 t.Run(name, func(t *testing.T) {
454 a, _, _ := probeService(t, obs)
455 a.fences = now
456 start := uint64(0)
457 if name != "installed" {
458 start = 9
459 }
460 e, proc, rep, err := heldProbe(t, a, start, probeAdmitted(obs), probeInputs())
461 if !errors.Is(err, ErrFenced) {
462 t.Fatalf("err = %v", err)
463 }
464 select {
465 case <-e.stopped:
466 case <-time.After(5 * time.Second):
467 t.Fatal("not stopped")
468 }
469 if !proc.isStopped() || !rep.isClosed() || a.Owns(probeHost) {
470 t.Fatalf("stopped=%v closed=%v owns=%v", proc.isStopped(), rep.isClosed(), a.Owns(probeHost))
471 }
472 })
473 }
474}
475
476func TestRecoveryHoldVerdict(t *testing.T) {
477 admitted := probeAdmitted(probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box")))
478 at := func(mutate func(*hostclaim.Claim)) hostclaim.Observed {
479 c := *admitted.Claim
480 h := *c.Holder
481 c.Holder = &h
482 mutate(&c)
483 return hostclaim.Observed{Host: probeHost, ETag: "later", Claim: &c}
484 }
485 cases := []struct {
486 name string
487 obs hostclaim.Observed
488 open bool
489 err bool
490 }{
491 {"probing", at(func(*hostclaim.Claim) {}), false, false},
492 {"probed", at(func(c *hostclaim.Claim) { c.Op.Phase = hostclaim.PhaseProbed }), false, false},
493 {"opened", at(func(c *hostclaim.Claim) { c.Op.Phase = hostclaim.PhaseOpened }), true, false},
494 {"released to serving", at(func(c *hostclaim.Claim) { c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"} }), true, false},
495 {"claimed", at(func(c *hostclaim.Claim) { c.Op.Phase = hostclaim.PhaseClaimed }), false, true},
496 {"new session", at(func(c *hostclaim.Claim) { c.Holder.Session++ }), false, true},
497 {"new generation", at(func(c *hostclaim.Claim) { c.Holder.Generation = "three" }), false, true},
498 {"other op", at(func(c *hostclaim.Claim) { c.Op.ID = "other" }), false, true},
499 {"other exec", at(func(c *hostclaim.Claim) { c.Op.Exec = "elsewhere" }), false, true},
500 {"lost", at(func(c *hostclaim.Claim) { c.Lost = []string{"box"} }), false, true},
501 {"serving elsewhere", at(func(c *hostclaim.Claim) {
502 c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "elsewhere"}
503 c.Holder.WriterBox = "elsewhere"
504 }), false, true},
505 {"absent", hostclaim.Observed{Host: probeHost, Absent: true}, false, true},
506 }
507 for _, c := range cases {
508 open, err := recoveryHoldVerdict(admitted, c.obs, "box")
509 if open != c.open || (err != nil) != c.err || (err != nil && !errors.Is(err, hostbox.ErrChanged)) {
510 t.Errorf("%s: open=%v err=%v", c.name, open, err)
511 }
512 }
513}
514
515func TestRecoveryHoldReviewOpensOrStops(t *testing.T) {
516 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
517 admitted := probeAdmitted(obs)
518 opened := probeAdmitted(obs)
519 oc := *opened.Claim
520 oc.Op.Phase = hostclaim.PhaseOpened
521 opened.Claim = &oc
522
523 a, r, _ := probeService(t, obs)
524 e, proc, _, err := heldProbe(t, a, 0, admitted, probeInputs())
525 if err != nil {
526 t.Fatal(err)
527 }
528 r.Read = func(context.Context, string) (hostclaim.Observed, error) { return admitted, nil }
529 a.reviewRecoveryHold(e)
530 a.mu.Lock()
531 held := e.probeHeld
532 a.mu.Unlock()
533 if !held {
534 t.Fatal("released at probing")
535 }
536 r.Read = func(context.Context, string) (hostclaim.Observed, error) {
537 return hostclaim.Observed{}, errors.New("offline")
538 }
539 a.reviewRecoveryHold(e)
540 a.mu.Lock()
541 held, state := e.probeHeld, e.state
542 a.mu.Unlock()
543 if !held || state != actAwake {
544 t.Fatal("an unreadable claim changed the hold")
545 }
546 r.Read = func(context.Context, string) (hostclaim.Observed, error) { return opened, nil }
547 a.reviewRecoveryHold(e)
548 a.mu.Lock()
549 held, state = e.probeHeld, e.state
550 a.mu.Unlock()
551 if held || state != actAwake || proc.isStopped() {
552 t.Fatalf("opened: held=%v state=%v", held, state)
553 }
554 release, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost})
555 if err != nil {
556 t.Fatalf("an opened probe does not serve: %v", err)
557 }
558 release()
559
560 b, rb, _ := probeService(t, obs)
561 e, proc, _, err = heldProbe(t, b, 0, admitted, probeInputs())
562 if err != nil {
563 t.Fatal(err)
564 }
565 moved := probeAdmitted(admitted)
566 rb.Read = func(context.Context, string) (hostclaim.Observed, error) { return moved, nil }
567 b.reviewRecoveryHold(e)
568 select {
569 case <-e.stopped:
570 case <-time.After(5 * time.Second):
571 t.Fatal("a changed session kept its process")
572 }
573 if !proc.isStopped() || b.Owns(probeHost) {
574 t.Fatal("not stopped")
575 }
576}
577
578// With the probe fields unset, the claim wake is main's: the ordinary
579// policy, never Probe, never Inputs.
580func TestOrdinaryClaimWakeKeepsTheServingPolicy(t *testing.T) {
581 c := &hostclaim.Claim{Holder: &hostclaim.Holder{Incarnation: "inc", Session: 1, Generation: "one", Location: "l", WriterBox: "box"}, Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}}
582 a, r, run := probeService(t, probeObserved(c))
583 stop := errors.New("stop")
584 r.Prerequisites = func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p StartPolicy) error {
585 run.step("prerequisites", p)
586 return stop
587 }
588 if _, err := a.doClaimWake(t.Context(), &actEntry{host: probeHost, route: registry.Route{Host: probeHost}}); !errors.Is(err, stop) {
589 t.Fatal(err)
590 }
591 run.mu.Lock()
592 defer run.mu.Unlock()
593 p := run.policies[len(run.policies)-1]
594 if !reflect.DeepEqual(p, StartPolicy{MaxStarts: 5, Window: 30 * time.Second}) {
595 t.Fatalf("ordinary policy %+v", p)
596 }
597}
598
599func TestRecoveryProbeOwnsItsInputs(t *testing.T) {
600 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
601 a, r, _ := probeService(t, obs)
602 in := probeInputs()
603 called := false
604 r.Prerequisites = func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p StartPolicy) error {
605 called = true
606 in.Release = "changed"
607 in.BundleETags["main"] = "changed"
608 if p.Inputs.Release != "release" || p.Inputs.BundleETags["main"] != "etag" {
609 t.Error("caller mutated captured probe inputs")
610 }
611 return hostbox.ErrChanged
612 }
613 _ = a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(in))
614 if !called {
615 t.Fatal("did not reach prerequisites")
616 }
617}
618
619func TestRecoveryProbeChangedInputsRetireHeldEntry(t *testing.T) {
620 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
621 a, _, _ := probeService(t, obs)
622 e, proc, rep, err := heldProbe(t, a, 0, probeAdmitted(obs), probeInputs())
623 if err != nil {
624 t.Fatal(err)
625 }
626 in := probeInputs()
627 in.Release = "new-release"
628 if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(in)); !errors.Is(err, hostbox.ErrChanged) {
629 t.Fatal(err)
630 }
631 a.mu.Lock()
632 stopped := e.stopped
633 state := e.state
634 a.mu.Unlock()
635 if state != actStopping && a.Owns(probeHost) {
636 t.Fatal("changed inputs stranded old held probe")
637 }
638 select {
639 case <-stopped:
640 case <-time.After(5 * time.Second):
641 t.Fatal("old probe did not stop")
642 }
643 if !proc.isStopped() || !rep.isClosed() || a.Owns(probeHost) {
644 t.Fatal("old probe retained resources")
645 }
646}
647
648func TestRecoveryProbeChangedInputsCancelWakingEntry(t *testing.T) {
649 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
650 a, _, _ := probeService(t, obs)
651 ctx, cancel := context.WithCancel(context.Background())
652 defer cancel()
653 p := probePolicy(probeInputs())
654 e := &actEntry{host: probeHost, recoveryPolicy: &p, probeHeld: true, state: actWaking, wakeCancel: cancel, ready: make(chan struct{})}
655 a.entries[probeHost] = e
656 in := probeInputs()
657 in.Release = "new"
658 if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(in)); !errors.Is(err, hostbox.ErrChanged) {
659 t.Fatal(err)
660 }
661 if ctx.Err() == nil {
662 t.Fatal("stale in-flight probe was not cancelled")
663 }
664 delete(a.entries, probeHost)
665}
666
667func TestRecoveryHoldSurvivesEdgeRetireExcept(t *testing.T) {
668 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
669 a, _, _ := probeService(t, obs)
670 e, _, _, err := heldProbe(t, a, 0, probeAdmitted(obs), probeInputs())
671 if err != nil {
672 t.Fatal(err)
673 }
674 a.RetireExcept(map[string]bool{})
675 a.mu.Lock()
676 state := e.state
677 a.mu.Unlock()
678 if state != actAwake {
679 t.Fatal("edge reload retired operation-owned recovery probe")
680 }
681}
682
683func TestRecoveryProbeLateHealthyAfterInputChangeStops(t *testing.T) {
684 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
685 a, _, _ := probeService(t, obs)
686 p := probePolicy(probeInputs())
687 ctx, cancel := context.WithCancel(context.Background())
688 defer cancel()
689 e := &actEntry{host: probeHost, recoveryPolicy: &p, probeHeld: true, state: actWaking, wakeCancel: cancel, ready: make(chan struct{}), probeAdmitted: probeAdmitted(obs)}
690 a.entries[probeHost] = e
691 in := probeInputs()
692 in.Release = "new"
693 _ = a.joinRecoveryProbe(t.Context(), e, probePolicy(in))
694 if ctx.Err() == nil {
695 t.Fatal("not cancelled")
696 }
697 proc, rep := &fakeProc{}, &fakeRep{}
698 err := a.finishRecoveryProbe(e, 0, &wokenInstance{proc: proc, rep: rep, stopOutcome: &stopObservation{}}, nil)
699 if !errors.Is(err, hostbox.ErrChanged) {
700 t.Fatal("late success kept obsolete inputs", err)
701 }
702 select {
703 case <-e.stopped:
704 case <-time.After(5 * time.Second):
705 t.Fatal("not stopped")
706 }
707 if !proc.isStopped() || !rep.isClosed() {
708 t.Fatal("obsolete process remains")
709 }
710}
711
712// Recovery must retain the handle returned by the launcher, and only publish
713// it as current after the exact admitted session opens.
714func TestRecoveryProbeGenerationRoutingAfterOpening(t *testing.T) {
715 for _, generated := range []bool{false, true} {
716 for _, phase := range []string{hostclaim.PhaseOpened, hostclaim.KindServing} {
717 t.Run(fmt.Sprint(generated)+"/"+phase, func(t *testing.T) {
718 obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
719 admitted := probeAdmitted(obs)
720 a, r, _ := probeService(t, obs)
721 a.gens = generated
722 rt := registry.Route{Host: probeHost, Socket: filepath.Join(t.TempDir(), "app.sock")}
723 p := probePolicy(probeInputs())
724 e := &actEntry{host: probeHost, route: rt, state: actWaking, ready: make(chan struct{}), recoveryPolicy: &p, probeHeld: true, probeAdmitted: admitted}
725 a.entries[probeHost] = e
726 proc, rep := &fakeProc{}, &fakeRep{}
727 w := &wokenInstance{proc: proc, rep: rep, spawnVersion: "recovered", stopOutcome: &stopObservation{}}
728 if generated {
729 w.gen = Gen{N: nextGen(), Socket: filepath.Join(filepath.Dir(rt.Socket), "generation.sock"), Build: "recovered"}
730 }
731 if err := a.finishRecoveryProbe(e, 0, w, nil); err != nil {
732 t.Fatal(err)
733 }
734 if e.gen.N == 0 {
735 t.Fatal("recovery lost its generation")
736 }
737 want := w.gen
738 if !generated {
739 want = canonicalGen(rt, e.gen.N, w.spawnVersion)
740 }
741 if !reflect.DeepEqual(e.gen, want) {
742 t.Fatalf("retained %+v, want %+v", e.gen, want)
743 }
744 if _, ok := a.CacheGen(probeHost); ok {
745 t.Fatal("held generation became current")
746 }
747 if generated {
748 if _, err := os.Lstat(rt.Socket); !errors.Is(err, os.ErrNotExist) {
749 t.Fatalf("held generation linked: %v", err)
750 }
751 }
752 opened := admitted
753 c := *admitted.Claim
754 c.Op.Phase = phase
755 if phase == hostclaim.KindServing {
756 c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
757 }
758 opened.Claim = &c
759 r.Read = func(context.Context, string) (hostclaim.Observed, error) { return opened, nil }
760 a.reviewRecoveryHold(e)
761 if e.probeHeld || e.proc != Proc(proc) || proc.isStopped() {
762 t.Fatal("opening replaced the recovered process")
763 }
764 got, ok := a.ServingGen(probeHost)
765 if !ok || got.N != want.N || got.Socket != want.Socket || got.Build != want.Build {
766 t.Fatalf("ServingGen = %+v, %v", got, ok)
767 }
768 if socket, ok := a.AppSocket(probeHost); !ok || socket != want.Socket {
769 t.Fatalf("AppSocket = %q, %v", socket, ok)
770 }
771 if n, ok := a.CacheGen(probeHost); !ok || n != want.N {
772 t.Fatalf("CacheGen = %d, %v", n, ok)
773 }
774 if generated {
775 if target, err := os.Readlink(rt.Socket); err != nil || target != want.Socket {
776 t.Fatalf("canonical link = %q, %v", target, err)
777 }
778 }
779 release, err := a.beginRequest(t.Context(), rt)
780 if err != nil {
781 t.Fatal(err)
782 }
783 release()
784 })
785 }
786 }
787}
788
789func TestRecoveryHoldSurvivesAdoptionAtProbed(t *testing.T) {
790 t.Parallel()
791 for _, kind := range []string{hostclaim.KindReverting, hostclaim.KindRestoring} {
792 for _, fenced := range []bool{false, true} {
793 for _, api := range []string{"Bounce", "BounceAdopted", "ConvergeAdopted"} {
794 t.Run(kind+"/"+fmt.Sprint(fenced)+"/"+api, func(t *testing.T) {
795 obs := probeObserved(probeClaim(kind, hostclaim.PhaseProbing, "box"))
796 admitted := probeAdmitted(obs)
797 a, r, _ := probeService(t, obs)
798 a.fences = map[string]uint64{probeHost: 9}
799 e, proc, rep, err := heldProbe(t, a, 9, admitted, probeInputs())
800 if err != nil {
801 t.Fatal(err)
802 }
803 probed := admitted
804 c := *admitted.Claim
805 c.Op.Phase = hostclaim.PhaseProbed
806 probed.Claim = &c
807 r.Read = func(context.Context, string) (hostclaim.Observed, error) { return probed, nil }
808 if !fenced {
809 delete(a.fences, probeHost)
810 }
811 a.reviewRecoveryHold(e)
812 a.liveVer = func(string) (string, error) { return "adopted", nil }
813 var bounced bool
814 switch api {
815 case "Bounce":
816 bounced = a.Bounce(probeHost)
817 case "BounceAdopted":
818 bounced = a.BounceAdopted(probeHost)
819 case "ConvergeAdopted":
820 bounced = a.ConvergeAdopted(probeHost, "adopted")
821 }
822 a.mu.Lock()
823 held, state, same := e.probeHeld, e.state, a.entries[probeHost] == e
824 a.mu.Unlock()
825 if bounced || !held || state != actAwake || !same || proc.isStopped() || rep.isClosed() {
826 t.Fatalf("adoption disturbed probed session: bounced=%v held=%v state=%v same=%v", bounced, held, state, same)
827 }
828 if err := a.RecoveryProbe(t.Context(), e.route, *e.recoveryPolicy); err != nil {
829 t.Fatalf("cannot rejoin held session: %v", err)
830 }
831 if _, err := a.beginRequest(t.Context(), e.route); !errors.Is(err, ErrFenced) {
832 t.Fatalf("probed session served: %v", err)
833 }
834 })
835 }
836 }
837 }
838}
839
840func TestRecoveryProbeUnboundAttemptsAreNotReady(t *testing.T) {
841 a, r, _ := probeService(t, probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box")))
842 r.RecoveryAttemptsReady = nil
843 if a.RecoveryProbeReady() {
844 t.Fatal("recovery probe ready without recovery attempts provider")
845 }
846}
847