amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "os/exec"
9 "reflect"
10 "strings"
11 "testing"
12 "time"
13
14 "github.com/carlosframework/platform/internal/activator"
15 "github.com/carlosframework/platform/internal/hostbox"
16 "github.com/carlosframework/platform/internal/hostclaim"
17 "github.com/carlosframework/platform/internal/hostguard"
18 "github.com/carlosframework/platform/internal/hostwrite"
19 "github.com/carlosframework/platform/internal/lease"
20 "github.com/carlosframework/platform/internal/moves"
21 "github.com/carlosframework/platform/internal/provision"
22 "github.com/carlosframework/platform/internal/registry"
23 "github.com/carlosframework/platform/internal/release"
24)
25
26// plan10J2AdapterInputs are what the source would render now.
27func plan10J2AdapterInputs() provision.MoveInputs {
28 return provision.MoveInputs{BundleETags: map[string]string{"default": "b1"}, OverlayETag: "o1", Release: "v7", EnvReport: "env"}
29}
30
31// plan10J2AdapterRecord is the rolling_back record of the recovery fixture's
32// claim: rollback plan10J2Rollback with the source at probing.
33func plan10J2AdapterRecord(incarnation string, dirty bool) provision.Record {
34 return provision.Record{Account: "bab", App: "jam", Host: recoveryHost, Move: &provision.MoveState{ID: "m1", Incarnation: incarnation, From: "ie", To: "uk", State: provision.MoveRollingBack,
35 Rollback: &provision.MoveRollback{ID: plan10J2Rollback, Dirty: dirty, SourcePhase: provision.RbSourceProbing}}}
36}
37
38// plan10J2AdapterRequest is the request movesrecover.go's probing builds,
39// recording into rec through the real provision transition.
40func plan10J2AdapterRequest(rec *provision.Record, srcGen string, events *[]string) recoveryProbeRequest {
41 g := provision.MoveGuard{ID: rec.Move.ID, Incarnation: rec.Move.Incarnation}
42 req := recoveryProbeRequest{RollbackID: rec.Move.Rollback.ID, Clean: !rec.Move.Rollback.Dirty, Attempts: recoveryProbeAttempts, SourceGeneration: srcGen,
43 Record: func(_ context.Context, in provision.MoveInputs) error {
44 *events = append(*events, "record")
45 return provision.MoveRecordRecoveryInputs(g, rec.Move.Rollback.ID, in)(rec)
46 }}
47 if rec.Move.Rollback.Dirty {
48 req.Coverage, req.Generation = recoveryCoverageSealed, rec.Move.Rollback.ID
49 } else {
50 req.Coverage = recoveryCoverageClean
51 }
52 return req
53}
54
55// plan10J2AdapterClaim puts the fixture's claim at restoring/probing with
56// box-1 executing and returns the incarnation.
57func plan10J2AdapterClaim(t *testing.T, f *recoveryFixture, phase, srcGen string) string {
58 t.Helper()
59 o, err := hostclaim.Update(t.Context(), f.claims, f.read(), func(c *hostclaim.Claim) error {
60 c.Op = plan10J2Op(hostclaim.KindRestoring, phase)
61 c.Op.Journal = map[string]string{hostclaim.JournalSrcGen: srcGen}
62 return nil
63 })
64 if err != nil {
65 t.Fatal(err)
66 }
67 return o.Claim.Holder.Incarnation
68}
69
70// The adapter records the recovery inputs at probing before anything is
71// rendered, then asks the activator for a probe-shaped wake carrying exactly
72// those inputs. That wake runs the real Group D path: it is admitted at
73// probing as a recovery probe, and Prepare binds the operation attempts and
74// probe readiness to the admitted session. The held render (Materialize)
75// follows the record.
76func TestPlan10J2AdapterRecordsInputsBeforeTheGroupDProbe(t *testing.T) {
77 for _, tc := range []struct{ dirty, revert bool }{{}, {dirty: true}, {revert: true}} {
78 dirty := tc.dirty
79 t.Run(fmt.Sprintf("dirty=%v/revert=%v", dirty, tc.revert), func(t *testing.T) {
80 f := newRecoveryFixture(t)
81 inc := plan10J2AdapterClaim(t, f, hostclaim.PhaseProbing, "src-gen")
82 c := f.controller("boot-1", "controller-1", 4101, "start-1")
83 ops := &plan9D2Attempts{}
84 c.wake.Inputs, c.wake.Continue, c.wake.Operations = plan9D2Inputs{recoveryInputs{f}}, hostclaim.SessionContinues, ops
85 c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }
86 rec := plan10J2AdapterRecord(inc, dirty)
87 var events []string
88 req := plan10J2AdapterRequest(&rec, "src-gen", &events)
89 kind := hostclaim.KindRestoring
90 if tc.revert {
91 kind = hostclaim.KindReverting
92 o, err := hostclaim.Update(t.Context(), f.claims, f.read(), func(c *hostclaim.Claim) error {
93 c.Op = plan10J2Op(hostclaim.KindReverting, hostclaim.PhaseProbing)
94 c.Op.Journal = map[string]string{hostclaim.JournalRevPhase: hostclaim.PhaseProbing}
95 return nil
96 })
97 if err != nil {
98 t.Fatal(err)
99 }
100 rec.Move.State, rec.Move.Rollback = provision.MoveReverting, nil
101 req = recoveryProbeRequest{Clean: true, SourceGeneration: o.Claim.Holder.Generation, Attempts: recoveryProbeAttempts, Record: func(_ context.Context, in provision.MoveInputs) error {
102 events = append(events, "record")
103 return provision.MoveRecordRevertInputs(provision.MoveGuard{ID: rec.Move.ID, Incarnation: inc}, in)(&rec)
104 }}
105 }
106 recorded := func() *provision.MoveInputs {
107 if tc.revert {
108 return rec.Move.RecoveryInputs
109 }
110 return rec.Move.Rollback.RecoveryInputs
111 }
112 f.materialize = func(context.Context, hostclaim.Observed) error {
113 events = append(events, "render")
114 if recorded() == nil {
115 t.Error("rendered before the inputs were recorded")
116 }
117 return nil
118 }
119 var policies []activator.StartPolicy
120 wake := func(ctx context.Context, rt registry.Route, p activator.StartPolicy) error {
121 events = append(events, "wake")
122 policies = append(policies, p)
123 if recorded() == nil || p.Inputs == nil || !reflect.DeepEqual(*p.Inputs, *recorded()) {
124 t.Errorf("the wake does not carry the recorded inputs: %+v / %+v", p.Inputs, recorded())
125 }
126 if err := c.wake.admitsOperation(f.read(), p, "box-1"); err != nil {
127 return err
128 }
129 // Prerequisites' S3 lease stands in, as plan10J2Recovery does.
130 live, cancel := context.WithCancel(context.Background())
131 t.Cleanup(cancel)
132 done := make(chan struct{})
133 go func() { <-live.Done(); close(done) }()
134 o := f.read()
135 c.wake.mu.Lock()
136 c.wake.pending[rt.Host] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done}
137 c.wake.mu.Unlock()
138 admitted, err := c.runtime.Admit(ctx, rt, o, p)
139 if err != nil {
140 return err
141 }
142 events = append(events, "admit")
143 s, err := c.runtime.Prepare(ctx, rt, admitted, p)
144 if err != nil {
145 return err
146 }
147 if _, err := s.OpenProbe(ctx, time.Now()); err != nil {
148 return err
149 }
150 if err := s.Reserve(ctx, 0); err != nil {
151 return err
152 }
153 return s.Materialize(ctx)
154 }
155 a := recoveryProbeAdapter{Box: "box-1", Claims: true, Wake: wake,
156 Route: func(host string) (*registry.Route, error) { return &registry.Route{Host: host}, nil },
157 Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
158 return plan10J2AdapterInputs(), nil
159 }}
160 out, err := a.probe(t.Context(), f.claims, rec, req)
161 if err != nil || out != recoveryProbeAnswered {
162 t.Fatalf("probe: %v %v", out, err)
163 }
164 if got := strings.Join(events, ","); got != "record,wake,admit,render" {
165 t.Fatalf("order: %s", got)
166 }
167 if len(policies) != 1 || !policies[0].Probe || policies[0].MaxStarts != recoveryProbeStartsPerWake || policies[0].Window != 30*time.Second || policies[0].RecoveryFence != moveFenceGeneration(rec.Move) {
168 t.Fatalf("policy: %+v", policies)
169 }
170 if got := strings.Join(ops.calls, ","); got != "open,reserve-0" {
171 t.Fatalf("attempt journal calls: %s", got)
172 }
173 for _, s := range ops.starts {
174 if !s.Policy.Probe || s.Admitted.Claim.Op.Kind != kind || s.Admitted.Claim.Op.Phase != hostclaim.PhaseProbing || !reflect.DeepEqual(*s.Policy.Inputs, plan10J2AdapterInputs()) {
175 t.Fatalf("attempt bound to %+v", s)
176 }
177 }
178 })
179 }
180}
181
182// A failed record write, inputs naming no release, or a claim that is not
183// this rollback's probe at probing by this box stops the attempt before any
184// render or wake.
185func TestPlan10J2AdapterStopsBeforeTheWake(t *testing.T) {
186 t.Parallel()
187 type tc struct {
188 claimPhase, exec, srcGen string
189 in provision.MoveInputs
190 recordErr error
191 mutateReq func(*recoveryProbeRequest)
192 mutateRec func(*provision.Record)
193 want error
194 }
195 recordFailed := errors.New("the record write failed")
196 for name, c := range map[string]tc{
197 "record write fails": {recordErr: recordFailed, want: recordFailed},
198 "no release": {in: provision.MoveInputs{OverlayETag: "o1"}, want: errRecoveryInputsNoRelease},
199 "claim at claimed": {claimPhase: hostclaim.PhaseClaimed, want: hostbox.ErrChanged},
200 "claim at probed": {claimPhase: hostclaim.PhaseProbed, want: hostbox.ErrChanged},
201 "other executor": {exec: "box-2", want: hostbox.ErrChanged},
202 "other rollback": {mutateReq: func(r *recoveryProbeRequest) { r.RollbackID = "rb_zzzzzzzzzzzzzzzzzzzzzzzzzz" }, want: errRecoveryProbeRequest},
203 "record not probing": {mutateRec: func(r *provision.Record) { r.Move.Rollback.SourcePhase = provision.RbSourceProbed }, want: errRecoveryProbeRequest},
204 "budget not nine": {mutateReq: func(r *recoveryProbeRequest) { r.Attempts = 3 }, want: errRecoveryProbeRequest},
205 "clean as sealed": {mutateReq: func(r *recoveryProbeRequest) { r.Coverage = recoveryCoverageSealed }, want: errRecoveryProbeRequest},
206 "clean with a gen": {mutateReq: func(r *recoveryProbeRequest) { r.Generation = plan10J2Rollback }, want: errRecoveryProbeRequest},
207 "src_gen differs": {srcGen: "other-gen", want: hostbox.ErrChanged},
208 "no record hook": {mutateReq: func(r *recoveryProbeRequest) { r.Record = nil }, want: errRecoveryProbeRequest},
209 } {
210 t.Run(name, func(t *testing.T) {
211 f := newRecoveryFixture(t)
212 phase := c.claimPhase
213 if phase == "" {
214 phase = hostclaim.PhaseProbing
215 }
216 srcGen := c.srcGen
217 if srcGen == "" {
218 srcGen = "src-gen"
219 }
220 inc := plan10J2AdapterClaim(t, f, phase, srcGen)
221 if c.exec != "" {
222 if _, err := hostclaim.Update(t.Context(), f.claims, f.read(), func(cl *hostclaim.Claim) error { cl.Op.Exec = c.exec; return nil }); err != nil {
223 t.Fatal(err)
224 }
225 }
226 rec := plan10J2AdapterRecord(inc, false)
227 var events []string
228 req := plan10J2AdapterRequest(&rec, "src-gen", &events)
229 if c.recordErr != nil {
230 req.Record = func(context.Context, provision.MoveInputs) error {
231 events = append(events, "record")
232 return c.recordErr
233 }
234 }
235 if c.mutateReq != nil {
236 c.mutateReq(&req)
237 }
238 if c.mutateRec != nil {
239 c.mutateRec(&rec)
240 }
241 in := c.in
242 if in.Release == "" && in.OverlayETag == "" {
243 in = plan10J2AdapterInputs()
244 }
245 a := recoveryProbeAdapter{Box: "box-1", Claims: true,
246 Wake: func(context.Context, registry.Route, activator.StartPolicy) error {
247 events = append(events, "wake")
248 return nil
249 },
250 Route: func(host string) (*registry.Route, error) { return &registry.Route{Host: host}, nil },
251 Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) { return in, nil }}
252 out, err := a.probe(t.Context(), f.claims, rec, req)
253 if !errors.Is(err, c.want) || out != recoveryProbeWaiting {
254 t.Fatalf("got %v %v, want %v", out, err, c.want)
255 }
256 for _, e := range events {
257 if e == "wake" {
258 t.Fatalf("woke: %v", events)
259 }
260 }
261 if c.recordErr == nil && len(events) != 0 {
262 t.Fatalf("recorded: %v", events)
263 }
264 })
265 }
266}
267
268// Without the host claim or the activator's recovery probe entry the
269// adapter is a typed dependency refusal: nothing is read, recorded or woken.
270func TestPlan10J2AdapterRefusesWithoutItsDependencies(t *testing.T) {
271 route := func(host string) (*registry.Route, error) { return &registry.Route{Host: host}, nil }
272 current := func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
273 return plan10J2AdapterInputs(), nil
274 }
275 wake := func(context.Context, registry.Route, activator.StartPolicy) error { return nil }
276 for name, a := range map[string]recoveryProbeAdapter{
277 "runtime unavailable": {Box: "box-1", Claims: true, Wake: wake, Route: route, Current: current, Ready: func() bool { return false }},
278 "claim off": {Box: "box-1", Claims: false, Wake: wake, Route: route, Current: current},
279 "no wake entry": {Box: "box-1", Claims: true, Route: route, Current: current},
280 "no route": {Box: "box-1", Claims: true, Wake: wake, Current: current},
281 "no inputs": {Box: "box-1", Claims: true, Wake: wake, Route: route},
282 } {
283 t.Run(name, func(t *testing.T) {
284 rec := plan10J2AdapterRecord("inc", false)
285 var events []string
286 out, err := a.probe(t.Context(), release.NewDirStore(t.TempDir()), rec, plan10J2AdapterRequest(&rec, "src-gen", &events))
287 var refusal *ClaimDependencyError
288 if !errors.As(err, &refusal) || out != recoveryProbeWaiting || len(events) != 0 || rec.Move.Rollback.RecoveryInputs != nil {
289 t.Fatalf("%v %v %v", out, err, events)
290 }
291 })
292 }
293}
294
295// The wake's outcome maps onto the recovery's: answered goes to probed, a
296// spent nine-attempt budget stops for an operator, a lapsed window or a
297// changed claim waits for the next pass, anything else is an error.
298func TestPlan10J2AdapterOutcomes(t *testing.T) {
299 t.Parallel()
300 other := errors.New("boom")
301 for name, tc := range map[string]struct {
302 wakeErr error
303 out recoveryProbeOutcome
304 err error
305 }{
306 "answered": {nil, recoveryProbeAnswered, nil},
307 "exhausted": {fmt.Errorf("reserve: %w", ErrRecoveryProbeExhausted), recoveryProbeExhausted, nil},
308 "window": {moves.ErrAttemptOver, recoveryProbeWaiting, nil},
309 "changed": {hostbox.ErrChanged, recoveryProbeWaiting, nil},
310 "other": {other, recoveryProbeWaiting, other},
311 } {
312 t.Run(name, func(t *testing.T) {
313 f := newRecoveryFixture(t)
314 rec := plan10J2AdapterRecord(plan10J2AdapterClaim(t, f, hostclaim.PhaseProbing, "src-gen"), false)
315 var events []string
316 a := recoveryProbeAdapter{Box: "box-1", Claims: true,
317 Wake: func(context.Context, registry.Route, activator.StartPolicy) error { return tc.wakeErr },
318 Route: func(host string) (*registry.Route, error) { return &registry.Route{Host: host}, nil },
319 Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
320 return plan10J2AdapterInputs(), nil
321 }}
322 out, err := a.probe(t.Context(), f.claims, rec, plan10J2AdapterRequest(&rec, "src-gen", &events))
323 if out != tc.out || (tc.err == nil) != (err == nil) || (tc.err != nil && !errors.Is(err, tc.err)) {
324 t.Fatalf("got %v %v", out, err)
325 }
326 })
327 }
328}
329
330// An unavailable wake entry keeps the recovery at probing without recording
331// inputs. The production binding itself is checked below with both flags.
332func TestPlan10J2MissingWakeKeepsRecoveryAtProbing(t *testing.T) {
333 t.Parallel()
334 cfg := moveCfg{Box: "ie-box", claimRecords: true, Route: func(string) (*registry.Route, error) { return nil, nil }}
335 w := &movesWorker{mirrors: map[string]lease.MirrorStore{}}
336 armRecoveryProbe(&cfg, w, nil, nil)
337 if cfg.recoveryProbe == nil {
338 t.Fatal("the production recovery probe hook is nil")
339 }
340 h := newRecoverHarness(t, false)
341 h.released(t)
342 h.sw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
343 for i := 0; i < 2; i++ {
344 c := h.sourceCfg(t)
345 c.claimRecords = true
346 armRecoveryProbe(&c, h.sw, nil, nil)
347 h.sw.cfg = c
348 if err := movesPass(context.Background(), h.s, h.sw); err != nil {
349 t.Fatal(err)
350 }
351 }
352 if rb := h.sourceAt(t, provision.RbSourceProbing); rb.RecoveryInputs != nil {
353 t.Fatalf("recorded without a wake entry: %+v", rb.RecoveryInputs)
354 }
355}
356
357// With a wake entry the same production adapter drives the real recovery
358// pass: each return records the inputs it renders at probing, before the
359// wake, and the recovery goes on to opened.
360// The production adapter accepts the claim's legacy encoding of a source
361// generation (whole-branch fixes, item 4): "" is a source whose sessions
362// predate admission records, which recoverRun.probing passes through
363// from the claim's journal unchanged and the owners normalise as @root.
364// A declared return and a clean return of such a source reach the wake;
365// a request naming another generation than the claim's still does not.
366func TestPlan10J2AdapterAcceptsALegacySourceGeneration(t *testing.T) {
367 ctx := context.Background()
368 for name, declared := range map[string]bool{"a clean return": false, "a declared return": true} {
369 t.Run(name, func(t *testing.T) {
370 s := release.NewDirStore(t.TempDir())
371 created := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
372 inc, err := hostclaim.Incarnation("bab", "app", p1Host, created)
373 if err != nil {
374 t.Fatal(err)
375 }
376 rec := provision.Record{Host: p1Host, Account: "bab", App: "app", Region: "ie",
377 Move: &provision.MoveState{ID: p1ID, Incarnation: inc, From: "ie", To: "uk", State: provision.MoveRollingBack,
378 Rollback: &provision.MoveRollback{ID: "rb_legacy", SourcePhase: provision.RbSourceProbing, TargetPhase: provision.RbTargetReleased}}}
379 journal := map[string]string{hostclaim.JournalSrcGen: "", hostclaim.JournalSourceStamp: "15", hostclaim.JournalCutSession: "15"}
380 if declared {
381 lossKey, err := hostclaim.LossRecordKey("l1")
382 if err != nil {
383 t.Fatal(err)
384 }
385 rec.Move.Rollback.LossRef = lossKey
386 journal["loss.l1"] = `{"kind":"move-target-lost","target":"uk-box"}`
387 }
388 c := &hostclaim.Claim{V: 2, Host: p1Host, Gen: 9,
389 Holder: &hostclaim.Holder{Family: "instance", Key: "apps/bab/app/instances/" + p1Host + ".json", Account: "bab", App: "app", Incarnation: inc, CreatedAt: created, Location: "instances/" + p1Host, Stamp: 15, Session: 20, WriterBox: "uk-box",
390 Admissions: []hostclaim.AdmissionRecord{{Session: 20, Box: "uk-box", Generation: p1ID, Region: "uk", PredecessorRegion: "ie", Mode: "cold", Coverage: "sealed"}}},
391 Op: hostclaim.Op{Kind: hostclaim.KindRestoring, ID: p1ID, Phase: hostclaim.PhaseProbing, Exec: "ie-box", Participants: []string{"ie-box", "uk-box"},
392 Intent: map[string]string{"to": "uk", "source": "ie-box", "target": "uk-box", "rollback": "rb_legacy"}, Journal: journal}}
393 if declared {
394 c.LossRefs, c.Lost = []string{"l1"}, []string{"uk-box"}
395 }
396 key, _ := hostclaim.Key(p1Host)
397 body, _ := json.Marshal(c)
398 if _, err := s.Put(ctx, key, body, ""); err != nil {
399 t.Fatal(err)
400 }
401 woke := 0
402 a := recoveryProbeAdapter{Box: "ie-box", Claims: true,
403 Route: func(string) (*registry.Route, error) { return &registry.Route{Host: p1Host, DBPath: "/data/x.db"}, nil },
404 Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
405 return provision.MoveInputs{Release: "v1"}, nil
406 },
407 Wake: func(context.Context, registry.Route, activator.StartPolicy) error { woke++; return nil }}
408 req := recoveryProbeRequest{RollbackID: "rb_legacy", Clean: true, Declared: declared, Attempts: recoveryProbeAttempts, SourceGeneration: "",
409 Record: func(context.Context, provision.MoveInputs) error { return nil }}
410 if !declared {
411 req.Coverage = recoveryCoverageClean
412 }
413 if out, err := a.probe(ctx, s, rec, req); err != nil || out != recoveryProbeAnswered || woke != 1 {
414 t.Fatalf("a legacy source's return: %v %v woke=%d", out, err, woke)
415 }
416 req.SourceGeneration = "other"
417 if _, err := a.probe(ctx, s, rec, req); !errors.Is(err, hostbox.ErrChanged) || woke != 1 {
418 t.Fatalf("another generation than the claim's: %v woke=%d", err, woke)
419 }
420 })
421 }
422}
423
424func TestPlan10J2AdapterDrivesTheRecoveryPass(t *testing.T) {
425 t.Parallel()
426 for _, dirty := range []bool{false, true} {
427 t.Run(fmt.Sprintf("dirty=%v", dirty), func(t *testing.T) {
428 h := newRecoverHarness(t, dirty)
429 h.released(t)
430 h.sw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
431 var wakes []activator.StartPolicy
432 wake := func(ctx context.Context, rt registry.Route, p activator.StartPolicy) error {
433 rb := h.read(t, p1Host).Move.Rollback
434 if rb.SourcePhase != provision.RbSourceProbing || rb.RecoveryInputs == nil || p.Inputs == nil || !reflect.DeepEqual(*rb.RecoveryInputs, *p.Inputs) {
435 t.Errorf("woke before the record: %+v / %+v", rb, p.Inputs)
436 }
437 c := h.claim(t)
438 if rt.Host != p1Host || c.Op.Kind != hostclaim.KindRestoring || c.Op.Phase != hostclaim.PhaseProbing || c.Op.Exec != "ie-box" {
439 t.Errorf("woke %s with the claim at %+v", rt.Host, c.Op)
440 }
441 wakes = append(wakes, p)
442 return nil
443 }
444 for i := 0; i < 3; i++ {
445 c := h.sourceCfg(t)
446 c.claimRecords = true
447 armRecoveryProbe(&c, h.sw, wake, nil)
448 a := c.recoveryProbe
449 c.recoveryProbe = func(ctx context.Context, s release.Store, rec provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) {
450 h.probes = append(h.probes, req)
451 return a(ctx, s, rec, req)
452 }
453 h.sw.cfg = c
454 h.sw.cfg.LaunchEnv = func(string) (map[string]string, map[string]bool, error) { return map[string]string{"A": "1"}, nil, nil }
455 h.sw.cfg.InputVersions = func(context.Context, release.Store, provision.Record) (map[string]string, string, error) {
456 in := h.renderedInputs()
457 return in.BundleETags, in.OverlayETag, nil
458 }
459 if err := movesPass(context.Background(), h.s, h.sw); err != nil {
460 t.Fatal(err)
461 }
462 }
463 h.opened(t)
464 if len(wakes) != 1 || !wakes[0].Probe || len(h.probes) != 1 {
465 t.Fatalf("wakes %+v probes %d", wakes, len(h.probes))
466 }
467 req := h.probes[0]
468 if req.Clean == dirty || req.Attempts != recoveryProbeAttempts || (dirty && (req.Coverage != recoveryCoverageSealed || req.Generation == "")) || (!dirty && (req.Coverage != recoveryCoverageClean || req.SourceGeneration != h.claim(t).Op.Journal[hostclaim.JournalSrcGen])) {
469 t.Fatalf("request %+v", req)
470 }
471 })
472 }
473}
474
475func TestPlan10J2ArmPhase1ConnectsRecoveryOnlyWithClaimFlag(t *testing.T) {
476 for _, enabled := range []bool{false, true} {
477 t.Run(fmt.Sprint(enabled), func(t *testing.T) {
478 cfg := moveCfg{claimRecords: enabled}
479 w := &movesWorker{mirrors: map[string]lease.MirrorStore{}}
480 armPhase1(&cfg, w, activator.Config{}, t.TempDir(), nil, &activator.Service{}, nil, hostwrite.Config{Enabled: enabled}, func(string) string { return "" })
481 if (cfg.recoveryProbe != nil) != enabled {
482 t.Fatal("production recovery hook does not follow claim flag")
483 }
484 })
485 }
486}
487
488func TestPlan10J2RevertRunUsesProductionAdapter(t *testing.T) {
489 t.Parallel()
490 h := newRevertHarness(t)
491 h.fromStopped(t)
492 h.arm(t)
493 wakes := 0
494 a := recoveryProbeAdapter{Box: "ie-box", Claims: true,
495 Route: func(string) (*registry.Route, error) { return &h.route, nil },
496 Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
497 return revertRendered(), nil
498 },
499 Wake: func(ctx context.Context, rt registry.Route, p activator.StartPolicy) error {
500 wakes++
501 m := h.move(t)
502 if m.RecoveryInputs == nil || !reflect.DeepEqual(*m.RecoveryInputs, *p.Inputs) {
503 t.Fatal("wake before revert inputs")
504 }
505 c := h.claim(t)
506 if c.Op.Kind != hostclaim.KindReverting || c.Op.Phase != hostclaim.PhaseProbing || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing {
507 t.Fatal("wrong claim")
508 }
509 return nil
510 }}
511 h.cfg.recoveryProbe = a.probe
512 h.pass(t)
513 if wakes != 1 {
514 t.Fatalf("wakes=%d", wakes)
515 }
516 h.reopened(t)
517}
518
519func TestPlan10J2RevertMissingProviderWaitsWithoutInputsOrWake(t *testing.T) {
520 t.Parallel()
521 h := newRevertHarness(t)
522 h.fromStopped(t)
523 h.arm(t)
524 calls := 0
525 a := recoveryProbeAdapter{Box: "ie-box", Claims: true, Ready: func() bool { return false }, Route: func(string) (*registry.Route, error) { return &h.route, nil }, Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
526 calls++
527 return revertRendered(), nil
528 }, Wake: func(context.Context, registry.Route, activator.StartPolicy) error { calls++; return nil }}
529 h.cfg.recoveryProbe = a.probe
530 h.pass(t)
531 before := h.move(t).UpdatedAt
532 h.clock.Advance(16 * time.Minute)
533 h.pass(t)
534 m := h.move(t)
535 if calls != 0 || m.RecoveryInputs != nil || m.State != provision.MoveReverting || !m.UpdatedAt.Equal(before) {
536 t.Fatalf("unbound provider changed recovery: calls=%d move=%+v", calls, m)
537 }
538}
539
540func TestPlan10J2NinthFailedStartSticksAndKeepsJournal(t *testing.T) {
541 t.Parallel()
542 h := newRevertHarness(t)
543 h.fromStopped(t)
544 h.arm(t)
545 starts, ids := 0, 0
546 var last OperationStart
547 provider := func() *movesProbeAttempts {
548 return &movesProbeAttempts{Claims: h.s, DataDir: h.cfg.DataDir,
549 Records: func(ctx context.Context, account, app, host string) (*provision.Record, error) {
550 return provision.GetRecord(ctx, h.s, account, app, host)
551 },
552 Admitted: func(context.Context, *provision.Record) (int, error) { return 9, nil },
553 NewID: func() string { ids++; return fmt.Sprintf("attempt-%d", ids) }, Now: time.Now}
554 }
555 a := recoveryProbeAdapter{Box: "ie-box", Claims: true, Ready: func() bool { return provider().RecoveryProbeAttemptsReady() }, Route: func(string) (*registry.Route, error) { return &h.route, nil }, Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
556 return revertRendered(), nil
557 },
558 Wake: func(ctx context.Context, rt registry.Route, p activator.StartPolicy) error {
559 o, err := hostclaim.Read(ctx, h.s, rt.Host)
560 if err != nil {
561 return err
562 }
563 holder := o.Claim.Holder
564 last = OperationStart{Admitted: o, Policy: p, Owner: hostguard.FenceTarget{Host: rt.Host, Incarnation: holder.Incarnation, Stamp: holder.Stamp, Session: holder.Session, Box: "ie-box"}}
565 ops := provider()
566 for i := 0; i < 3; i++ {
567 if err := ops.Reserve(ctx, last, starts); err != nil {
568 return err
569 }
570 starts++
571 if err := ops.Failed(ctx, last, starts-1); err != nil {
572 return err
573 }
574 }
575 return moves.ErrAttemptOver
576 }}
577 h.cfg.recoveryProbe = a.probe
578 for i := 0; i < 3; i++ {
579 h.pass(t)
580 }
581 m := h.move(t)
582 if starts != 9 || m.State != provision.MoveStuck || m.StuckReason != provision.StuckRecoveryProbeAttempts {
583 t.Fatalf("starts=%d move=%+v", starts, m)
584 }
585 j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
586 if err != nil {
587 t.Fatal(err)
588 }
589 if count, err := recoveryAttemptCount(j, recoveryAttemptKey(last, 0)); err != nil || count != 9 {
590 t.Fatalf("journal count=%d err=%v", count, err)
591 }
592}
593
594func TestPlan10J2RevertAdapterRejectsChangedBinding(t *testing.T) {
595 t.Parallel()
596 for _, tc := range []string{"reopened", "generation", "rev_phase", "kind", "executor"} {
597 t.Run(tc, func(t *testing.T) {
598 f := newRecoveryFixture(t)
599 o, err := hostclaim.Update(t.Context(), f.claims, f.read(), func(c *hostclaim.Claim) error {
600 c.Op = plan10J2Op(hostclaim.KindReverting, hostclaim.PhaseProbing)
601 c.Op.Journal = map[string]string{hostclaim.JournalRevPhase: hostclaim.PhaseProbing}
602 switch tc {
603 case "rev_phase":
604 c.Op.Journal[hostclaim.JournalRevPhase] = hostclaim.PhaseProbed
605 case "kind":
606 c.Op.Kind = hostclaim.KindRestoring
607 case "executor":
608 c.Op.Exec = "box-2"
609 }
610 return nil
611 })
612 if err != nil {
613 t.Fatal(err)
614 }
615 rec := provision.Record{Host: recoveryHost, Move: &provision.MoveState{ID: o.Claim.Op.ID, Incarnation: o.Claim.Holder.Incarnation, State: provision.MoveReverting}}
616 req := recoveryProbeRequest{Clean: true, SourceGeneration: o.Claim.Holder.Generation, Attempts: recoveryProbeAttempts, Record: func(context.Context, provision.MoveInputs) error { t.Fatal("recorded invalid revert"); return nil }}
617 if tc == "generation" {
618 req.SourceGeneration = "other"
619 }
620 if tc == "reopened" {
621 now := time.Now()
622 rec.Move.ReopenedAt = &now
623 }
624 a := recoveryProbeAdapter{Claims: true, Box: "box-1", Route: func(string) (*registry.Route, error) { t.Fatal("looked up invalid revert"); return nil, nil }, Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
625 t.Fatal("selected invalid revert")
626 return provision.MoveInputs{}, nil
627 }, Wake: func(context.Context, registry.Route, activator.StartPolicy) error {
628 t.Fatal("woke invalid revert")
629 return nil
630 }}
631 if _, err := a.probe(t.Context(), f.claims, rec, req); err == nil {
632 t.Fatal("invalid revert accepted")
633 }
634 })
635 }
636}
637