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 "os"
6 "path/filepath"
7 "strings"
8 "testing"
9
10 "github.com/carlosframework/platform/internal/appconfig"
11 "github.com/carlosframework/platform/internal/hostbox"
12 "github.com/carlosframework/platform/internal/hostclaim"
13 "github.com/carlosframework/platform/internal/hostguard"
14 "github.com/carlosframework/platform/internal/lease"
15 "github.com/carlosframework/platform/internal/moves"
16 "github.com/carlosframework/platform/internal/park"
17 "github.com/carlosframework/platform/internal/provision"
18 "github.com/carlosframework/platform/internal/registry"
19 "github.com/carlosframework/platform/internal/release"
20)
21
22// The source's recovery after a rollback (moves spec §7.8 source table; Plan
23// 4 Task 5). The fixture is Task 4's: the landing harness landed through
24// restored and stepped to probing, then rolled back and released by the
25// target's own pass (clean, or dirty with the target's rows changed, so its
26// return lane holds a snapshot of the stopped database). The source box
27// (ie-box) then runs its own passes over the same store and buckets: its
28// row is the cut's, held moving by the move id; its warm database's marker
29// names the pre-move generation; its hold hand-over, its open and the probe
30// (Task 9, BLOCKED-ON-CLAIM C1) are faked; everything else is the real
31// recovery, the shared installer (Task 3) and provision's and hostclaim's own
32// transitions.
33
34type recoverHarness struct {
35 *rollbackHarness
36 sdir string // the source's data directory
37 lockDir string
38 srcRoute registry.Route
39 sw *movesWorker
40 srcRender appconfig.HeldRender
41 srcRends int // the installer's held renders on the source
42 srcAsked []string // the releases the source's gate was asked about
43 srcHanded []string // status:to:from of each hand-over the source asked for
44 sEvents []string
45 // probe answers each recovery probe; nil records the render's inputs
46 // and answers.
47 probe func(req recoveryProbeRequest) (recoveryProbeOutcome, error)
48 probes []recoveryProbeRequest
49 // recoveryWake, when set, replaces the fake: the production adapter
50 // runs with it as the activator's recovery probe entry (home part 2,
51 // Task 4).
52 recoveryWake ClaimRecoveryProbeWake
53 // srcPage, when set, is the source's pager instead of pages.hook().
54 srcPage pageFunc
55 rcAt string
56 rcDone bool
57 rcHits []string
58 rcHook func(point string)
59}
60
61// newRecoverHarness builds the fixture up to the rollback decision's
62// release: dirty makes the target's state differ from what it installed, so
63// its release is dirty and its return a snapshot. The caller rolls back and
64// runs the target's pass.
65func newRecoverHarness(t *testing.T, dirty bool, opts ...p1Option) *recoverHarness {
66 t.Helper()
67 h := &recoverHarness{rollbackHarness: newRollbackHarness(t, opts...)}
68 if dirty {
69 h.targetChanged(t)
70 }
71 h.sdir = h.cfg.DataDir
72 h.lockDir = t.TempDir()
73 for _, d := range []string{"hostguard", ".parked"} {
74 if err := os.MkdirAll(filepath.Join(h.sdir, d), 0o770); err != nil {
75 t.Fatal(err)
76 }
77 }
78 // The source's warm database (the cut's stopped database) carries the
79 // marker of its pre-move generation, src_gen.
80 db := provision.DefaultDBPath(h.sdir, p1Host)
81 if err := os.WriteFile(db+".gen", []byte("cur\n"), 0o644); err != nil {
82 t.Fatal(err)
83 }
84 h.srcRoute = registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Hibernate: true, Status: "moving", HoldOp: p1ID,
85 Socket: provision.DefaultSocket(p1Host), DBPath: db, Provisioned: true}
86 // The current bundle is not the pinned one: a recovery renders and
87 // records what is current, never compared with move.inputs.
88 h.srcRender = appconfig.HeldRender{Path: filepath.Join(h.sdir, p1Host+".json"), SHA256: "y", BundleETags: map[string]string{"default": "e9", "environment": ""}, OverlayETag: "o9"}
89 return h
90}
91
92// released rolls the landing back and runs the target's pass: released
93// (clean) or returned (dirty).
94func (h *recoverHarness) released(t *testing.T) *provision.MoveRollback {
95 t.Helper()
96 h.rollBack(t)
97 h.rollbackPass(t)
98 rb := h.rb(t)
99 if want := map[bool]string{false: provision.RbTargetReleased, true: provision.RbTargetReturned}[rb.Dirty]; rb.TargetPhase != want {
100 t.Fatalf("the target did not release: %+v", rb)
101 }
102 return rb
103}
104
105// sourceCfg is the source box's configuration with the recovery armed.
106func (h *recoverHarness) sourceCfg(t *testing.T) moveCfg {
107 t.Helper()
108 c := h.landingCfg(t)
109 c.Region, c.Box, c.OwnBucket, c.OwnBucketRegion = h.src, h.srcBox, h.srcBucket, h.srcAWS
110 c.ParkHomeRegion = ""
111 c.DataDir = h.sdir
112 c.OptDir = h.optDir
113 c.OwnsRoute = func(host string) bool { return host == p1Host }
114 c.Route = func(host string) (*registry.Route, error) {
115 if host != p1Host {
116 return nil, nil
117 }
118 rt := h.srcRoute
119 return &rt, nil
120 }
121 c.FilesDir = func(host string) (string, string) {
122 return filepath.Join(h.sdir, host), filepath.Join(h.sdir, host+".env")
123 }
124 c.Guard = hostguard.Guard{DataDir: h.sdir, LockDir: h.lockDir}
125 c.Local = park.Local{DataDir: h.sdir, Box: h.srcBox}
126 c.ParkView = recordView(h.s, park.PlacementRegion(h.src), roleSource)
127 c.RenderHeld = func(_ context.Context, _ release.Store, rt registry.Route, op string, _ *provision.MoveInputs, before func(string) error) (appconfig.HeldRender, error) {
128 h.srcRends++
129 if err := before(h.srcRender.SHA256); err != nil {
130 return appconfig.HeldRender{}, err
131 }
132 if rb := h.read(t, p1Host).Move.Rollback; rb == nil || op != rb.ID || rt.HoldOp != rb.ID {
133 t.Errorf("rendered for %q over a row held by %q, want the rollback", op, rt.HoldOp)
134 }
135 return h.srcRender, nil
136 }
137 c.ReleaseReady = func(_ context.Context, _ release.Store, _ provision.Record, v string) moves.Check {
138 h.srcAsked = append(h.srcAsked, v)
139 return moves.Check{OK: true}
140 }
141 c.HandHold = func(host, status, to, from string, validate func() error) (bool, error) {
142 h.srcHanded = append(h.srcHanded, status+":"+to+":"+from)
143 if validate != nil {
144 if err := validate(); err != nil {
145 return false, err
146 }
147 }
148 if h.srcRoute.HoldOp == to && h.srcRoute.Status == status {
149 return true, nil
150 }
151 if h.srcRoute.HoldOp != from {
152 return false, nil
153 }
154 h.srcRoute.Status, h.srcRoute.HoldOp = status, to
155 return true, nil
156 }
157 c.ReleaseHeld = func(host, op string) (bool, error) {
158 if op == "" || h.srcRoute.HoldOp != op {
159 return false, nil
160 }
161 h.srcRoute.Status, h.srcRoute.HoldOp = "ready", ""
162 return true, nil
163 }
164 c.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
165 h.sEvents = append(h.sEvents, ev.Event)
166 return nil
167 }
168 c.Page = h.pages.hook()
169 if h.srcPage != nil {
170 c.Page = h.srcPage
171 }
172 c.recoveryProbe = func(ctx context.Context, s release.Store, rec provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) {
173 h.probes = append(h.probes, req)
174 if h.recoveryWake != nil {
175 // The production adapter (armRecoveryProbe's) over this worker:
176 // the box's current inputs recorded, then the wake.
177 w := h.sw
178 a := recoveryProbeAdapter{Box: w.cfg.Box, Claims: true, Route: w.cfg.Route, Current: w.recoveryInputs, Wake: h.recoveryWake}
179 return a.probe(ctx, s, rec, req)
180 }
181 if h.probe != nil {
182 return h.probe(req)
183 }
184 if err := req.Record(ctx, h.renderedInputs()); err != nil {
185 return recoveryProbeWaiting, err
186 }
187 return recoveryProbeAnswered, nil
188 }
189 c.rcFault = func(point string) error {
190 h.rcHits = append(h.rcHits, point)
191 if h.rcHook != nil {
192 h.rcHook(point)
193 }
194 if point == h.rcAt && !h.rcDone {
195 h.rcDone = true
196 return errCrash
197 }
198 return nil
199 }
200 return c
201}
202
203// renderedInputs are what the source renders now: the render's etags and
204// the live release.
205func (h *recoverHarness) renderedInputs() provision.MoveInputs {
206 b := map[string]string{}
207 for k, v := range h.srcRender.BundleETags {
208 b[k] = v
209 }
210 return provision.MoveInputs{BundleETags: b, OverlayETag: h.srcRender.OverlayETag, Release: "v7"}
211}
212
213// srcPass runs one pass as the source box, its worker kept across passes.
214func (h *recoverHarness) srcPass(t *testing.T) {
215 t.Helper()
216 if h.sw == nil {
217 h.sw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
218 }
219 h.sw.cfg = h.sourceCfg(t)
220 if err := movesPass(context.Background(), h.s, h.sw); err != nil {
221 t.Fatal(err)
222 }
223}
224
225func (h *recoverHarness) sourceAt(t *testing.T, phase string) *provision.MoveRollback {
226 t.Helper()
227 m := h.read(t, p1Host).Move
228 if m == nil || m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.SourcePhase != phase {
229 if m != nil {
230 t.Fatalf("the source is not at %s: %+v %+v", phase, m, m.Rollback)
231 }
232 t.Fatalf("the source is not at %s: no move", phase)
233 }
234 return m.Rollback
235}
236
237// opened asserts the end of a recovery: the record opened with RecoveredAt
238// and OpenedAt, the claim restoring/opened with the source executing, the
239// row open.
240func (h *recoverHarness) opened(t *testing.T) *provision.MoveRollback {
241 t.Helper()
242 rb := h.sourceAt(t, provision.RbSourceOpened)
243 if rb.RecoveredAt == nil || rb.OpenedAt == nil {
244 t.Fatalf("opened without its stamps: %+v", rb)
245 }
246 c := h.claim(t)
247 if c.Op.Kind != hostclaim.KindRestoring || c.Op.Phase != hostclaim.PhaseOpened || c.Op.Exec != h.srcBox || c.Op.Intent["rollback"] != rb.ID {
248 t.Fatalf("claim: %+v", c.Op)
249 }
250 if h.srcRoute.Status != "ready" || h.srcRoute.HoldOp != "" {
251 t.Fatalf("the row: %+v", h.srcRoute)
252 }
253 return rb
254}
255
256// TestSourceWaitsForTargetEvidence: the source's recovery starts only on the
257// target's evidence (§7.8 waiting): the record released clean or returned
258// dirty, and the claim's execution holder none. Until then nothing is
259// claimed, held or probed.
260func TestSourceWaitsForTargetEvidence(t *testing.T) {
261 t.Parallel()
262 untouched := func(t *testing.T, h *recoverHarness) {
263 t.Helper()
264 h.sourceAt(t, provision.RbSourceWaiting)
265 if h.srcRoute.Status != "moving" || h.srcRoute.HoldOp != p1ID || len(h.srcHanded) != 0 || len(h.probes) != 0 {
266 t.Fatalf("the source acted: row %+v handed %v probes %d", h.srcRoute, h.srcHanded, len(h.probes))
267 }
268 if c := h.claim(t); c.Op.Exec == "ie-box" || c.Op.Phase == hostclaim.PhaseClaimed {
269 t.Fatalf("the source claimed: %+v", c.Op)
270 }
271 }
272 t.Run("the target is releasing", func(t *testing.T) {
273 h := newRecoverHarness(t, false)
274 h.rollBack(t)
275 h.srcPass(t)
276 untouched(t, h)
277 })
278 t.Run("released clean on the record, the target still executing", func(t *testing.T) {
279 h := newRecoverHarness(t, false)
280 h.rollBack(t)
281 h.rbAt = "after-record-released"
282 h.rollbackPass(t)
283 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReleased || rb.Dirty {
284 t.Fatalf("fixture: %+v", rb)
285 }
286 if c := h.claim(t); c.Op.Phase != hostclaim.PhaseReleasing || c.Op.Exec != "uk-box" {
287 t.Fatalf("fixture claim: %+v", c.Op)
288 }
289 h.srcPass(t)
290 untouched(t, h)
291 // The target's next pass writes the claim half; the source then
292 // recovers.
293 h.rollbackPass(t)
294 h.srcPass(t)
295 h.opened(t)
296 })
297 t.Run("dirty, released but not returned", func(t *testing.T) {
298 h := newRecoverHarness(t, true)
299 h.rollBack(t)
300 h.rbAt = "after-record-returning"
301 h.rollbackPass(t)
302 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReturning || !rb.Dirty {
303 t.Fatalf("fixture: %+v", rb)
304 }
305 if c := h.claim(t); c.Op.Phase != hostclaim.PhaseReleased || c.Op.Exec != "" {
306 t.Fatalf("fixture claim: %+v", c.Op)
307 }
308 h.srcPass(t)
309 untouched(t, h)
310 })
311 t.Run("unarmed in production", func(t *testing.T) {
312 h := newRecoverHarness(t, false)
313 h.released(t)
314 // The source box as production wires it: no park engine leases
315 // (ruling G17), so the recovery waits for Plan 5. h.cfg was armed
316 // for the cut guard (armStandIns), so the leases are taken away.
317 h.cfg.ParkLeases = nil
318 if recoverySourceArmed(h.cfg) {
319 t.Fatal("fixture: the source's recovery is still armed")
320 }
321 h.pass(t)
322 untouched(t, h)
323 })
324}
325
326// TestCleanReturnVerifiesAndOpens: a clean return claims the host back with
327// the source's own generation named again (holder.generation = src_gen), its
328// row handed from the move to the rollback as restoring, its files, timers
329// and warm database marker verified against what it recorded when it
330// stopped, then the probe (faked until Task 9), probed, recovered and the
331// open. No install phase runs.
332func TestCleanReturnVerifiesAndOpens(t *testing.T) {
333 t.Parallel()
334 h := newRecoverHarness(t, false)
335 pinned := h.read(t, p1Host).Move.Inputs
336 rb := h.released(t)
337 srcGen := h.claim(t).Op.Journal[hostclaim.JournalSrcGen]
338 // The landing's probe admission named the landing's generation; the
339 // return must name the source's again.
340 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Holder.Generation = p1ID })
341 var atClaimed *hostclaim.Holder
342 h.rcHook = func(point string) {
343 if point == "after-activate" {
344 c := h.claim(t)
345 hc := *c.Holder
346 atClaimed = &hc
347 }
348 }
349 h.srcPass(t)
350 got := h.opened(t)
351 if atClaimed == nil || atClaimed.Generation != srcGen || srcGen != "cur" || atClaimed.Stamp != 1 {
352 t.Fatalf("the holder after claimed: %+v (src_gen %q)", atClaimed, srcGen)
353 }
354 if captured, err := hostguard.Capture(h.sdir, p1Host); err != nil || captured.Stamp != 1 {
355 t.Fatalf("the source's installation record: %+v %v", captured, err)
356 }
357 if strings.Join(h.srcHanded, " ") != "restoring:"+rb.ID+":"+p1ID {
358 t.Fatalf("hand-overs: %v", h.srcHanded)
359 }
360 // No install phase ran: no render, no install steps in the recovery's
361 // journal (which holds only its mark, fix round 1), no generation.
362 if h.srcRends != 0 {
363 t.Fatalf("a clean return rendered through the installer %d times", h.srcRends)
364 }
365 if j, err := moves.ReadJournal(h.sdir, p1Host, rb.ID); err != nil || len(j.Steps) != 1 || !j.Done(rcStepSource) {
366 t.Fatalf("a clean return's journal: %+v %v", j, err)
367 }
368 for k := range h.f.buckets["data-ie"] {
369 if strings.HasPrefix(k, moves.GenerationPrefix(cutRoot, rb.ID)) {
370 t.Fatalf("a clean return built a generation: %s", k)
371 }
372 }
373 for _, p := range h.rcHits {
374 if strings.HasPrefix(p, "after-record-") {
375 switch strings.TrimPrefix(p, "after-record-") {
376 case provision.RbSourceParking, provision.RbSourceBuilding, provision.RbSourcePublished, provision.RbSourceLocal, provision.RbSourceConfigured, provision.RbSourceRestored:
377 t.Fatalf("a clean return reached %s", p)
378 }
379 }
380 }
381 // The probe (Task 9's): clean, coverage clean-return, nine attempts; its
382 // inputs recorded, never compared with move.inputs.
383 if len(h.probes) != 1 || !h.probes[0].Clean || h.probes[0].Coverage != "clean-return" || h.probes[0].Attempts != 9 || h.probes[0].RollbackID != rb.ID ||
384 h.probes[0].SourceGeneration != "cur" {
385 t.Fatalf("probes: %+v", h.probes)
386 }
387 m := h.read(t, p1Host).Move
388 if in := got.RecoveryInputs; in == nil || in.BundleETags["default"] != "e9" || in.OverlayETag != "o9" || in.Release != "v7" {
389 t.Fatalf("recovery inputs: %+v", got.RecoveryInputs)
390 }
391 if m.Inputs == nil || pinned == nil || m.Inputs.BundleETags["default"] != pinned.BundleETags["default"] || m.Inputs.Release != pinned.Release {
392 t.Fatalf("move.inputs changed: %+v was %+v", m.Inputs, pinned)
393 }
394 // A clean return queues no park the source owes: no return lane, no
395 // park-local of its own.
396 for _, o := range h.claim(t).Obligations {
397 if o.Region == "ie" || o.Box == "ie-box" {
398 t.Fatalf("a clean return queued %+v", o)
399 }
400 }
401 if !strings.Contains(strings.Join(h.sEvents, " "), "recovered opened") {
402 t.Fatalf("events %v", h.sEvents)
403 }
404 // The record waits for the target's cleaning: a second pass changes
405 // nothing.
406 h.srcPass(t)
407 h.opened(t)
408}
409
410// TestDirtyReturnInstallsFromTheLane: a dirty return installs the target's
411// snapshot from the return lane through the shared installer: the return
412// generation g-<rb>/ (snapshot and anchor) in the source's own bucket, the
413// database restored to what the target wrote, .gen naming the rollback;
414// recovery inputs recorded at configured and again by the probe, never
415// move.inputs (witness wRecoveryInputs: a bundle published after the
416// configured render does not block the probe); then probed, which queues
417// the return lane's park in the source's region, recovered and opened.
418func TestDirtyReturnInstallsFromTheLane(t *testing.T) {
419 t.Parallel()
420 for _, registered := range []bool{false, true} {
421 name := "unregistered"
422 if registered {
423 name = "registered"
424 }
425 t.Run(name, func(t *testing.T) {
426 h := newRecoverHarness(t, true)
427 pinned := h.read(t, p1Host).Move.Inputs
428 rb := h.released(t)
429 c := h.claim(t)
430 returnSeal, err := moves.DecodeSeal(c.Op.Journal[moves.JournalReturnSeal])
431 if err != nil {
432 t.Fatal(err)
433 }
434 if registered {
435 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
436 c.Inventory = map[string]hostclaim.InventoryEntry{"ie-box": {Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 1, p1Host + ".db.gen": 1, p1Host: 1, p1Host + ".env": 1}}}
437 })
438 }
439 var queuedLocal bool
440 var atConfigured *provision.MoveInputs
441 h.landFault = func(point string) error {
442 if point == "after-record-"+provision.RbSourceRestored {
443 atConfigured = h.read(t, p1Host).Move.Rollback.RecoveryInputs
444 // A bundle published after the configured render.
445 h.srcRender.BundleETags = map[string]string{"default": "e10", "environment": ""}
446 }
447 return nil
448 }
449 h.rcHook = func(point string) {
450 if point == "after-record-"+provision.RbSourceRestoring {
451 for _, o := range h.claim(t).Obligations {
452 if o.Kind == "park-local" && o.Box == "ie-box" {
453 queuedLocal = true
454 }
455 }
456 }
457 }
458 h.srcPass(t)
459 got := h.opened(t)
460 if queuedLocal != registered {
461 t.Fatalf("the source's park-local queued at claimed: %v, registered %v", queuedLocal, registered)
462 }
463 if atConfigured == nil || atConfigured.BundleETags["default"] != "e9" || atConfigured.Release != "v7" {
464 t.Fatalf("configured recorded %+v", atConfigured)
465 }
466 if in := got.RecoveryInputs; in == nil || in.BundleETags["default"] != "e10" || in.Release != "v7" {
467 t.Fatalf("the probe's recovery inputs: %+v", got.RecoveryInputs)
468 }
469 m := h.read(t, p1Host).Move
470 if m.Inputs == nil || m.Inputs.BundleETags["default"] != pinned.BundleETags["default"] || m.Inputs.Release != pinned.Release {
471 t.Fatalf("move.inputs changed: %+v", m.Inputs)
472 }
473 // The return generation: a snapshot and an anchor at N'.
474 prefix := moves.GenerationPrefix(cutRoot, rb.ID)
475 for _, k := range []string{moves.SnapshotKey(prefix, returnSeal.N), moves.AnchorKey(prefix, returnSeal.N)} {
476 if _, ok := h.f.buckets["data-ie"][k]; !ok {
477 t.Fatalf("the return generation has no %s", k)
478 }
479 }
480 dbPath := provision.DefaultDBPath(h.sdir, p1Host)
481 if n := countRowsForTest(t, dbPath); n != h.rows+1 {
482 t.Fatalf("the returned database holds %d rows, the target wrote %d", n, h.rows+1)
483 }
484 if gen, err := hostbox.ReadGenerationMarker(dbPath + ".gen"); err != nil || gen != rb.ID {
485 t.Fatalf(".gen %q %v", gen, err)
486 }
487 // The probe: dirty, into the return generation, coverage sealed.
488 if len(h.probes) != 1 || h.probes[0].Clean || h.probes[0].Coverage != "sealed" || h.probes[0].Generation != rb.ID || h.probes[0].Attempts != 9 {
489 t.Fatalf("probes: %+v", h.probes)
490 }
491 // probed queued the return lane's park, in the source's region,
492 // its horizon the return seal's bound.
493 c = h.claim(t)
494 lane := 0
495 for _, o := range c.Obligations {
496 if o.Kind == "park-replica" && o.Region == "ie" && o.Park != nil && o.Park.Location == moves.ScopeLocation(p1Host, rb.ID) {
497 lane++
498 if !o.Park.EvidenceAt.Equal(returnSeal.Bound()) || o.Park.RequestLimit != cutRequestLimit || o.Park.Incarnation != h.inc {
499 t.Fatalf("the lane park's scope: %+v", o.Park)
500 }
501 }
502 if o.Kind == "park-local" && o.Box == "ie-box" {
503 t.Fatalf("the source's park-local is still owed: %+v", o)
504 }
505 }
506 if lane != 1 {
507 t.Fatalf("return lane parks: %d (%+v)", lane, c.Obligations)
508 }
509 if strings.Join(h.srcHanded, " ") != "restoring:"+rb.ID+":"+p1ID {
510 t.Fatalf("hand-overs: %v", h.srcHanded)
511 }
512 if held := h.leases.holding(); len(held) != 0 {
513 t.Fatalf("the source still holds %v", held)
514 }
515 })
516 }
517}
518
519// TestVerifyMismatchIsStuck (INV-4): a clean return whose files, timers or
520// warm database marker no longer match what the source recorded when it
521// stopped stops for an operator with the baseline-mismatch sentence, at
522// verified, before any probe; the row stays held. A retry verifies again,
523// never skips it: unchanged, it stops again; repaired, it recovers.
524func TestVerifyMismatchIsStuck(t *testing.T) {
525 t.Parallel()
526 for _, c := range []struct {
527 name string
528 breakIt, mend func(h *recoverHarness)
529 }{
530 {"files", func(h *recoverHarness) {
531 os.WriteFile(filepath.Join(h.sdir, p1Host, "upload.txt"), []byte("changed on the source"), 0o644)
532 }, func(h *recoverHarness) {
533 os.WriteFile(filepath.Join(h.sdir, p1Host, "upload.txt"), []byte("hello"), 0o644)
534 }},
535 {"timers", func(h *recoverHarness) {
536 h.timers = append(append([]registry.Timer(nil), h.timers...), registry.Timer{Host: p1Host, Name: "extra", Kind: "cron", Spec: "0 4 * * *", NextAt: 2000})
537 }, func(h *recoverHarness) {
538 h.timers = h.timers[:len(h.timers)-1]
539 }},
540 {"generation marker", func(h *recoverHarness) {
541 os.WriteFile(provision.DefaultDBPath(h.sdir, p1Host)+".gen", []byte("other\n"), 0o644)
542 }, func(h *recoverHarness) {
543 os.WriteFile(provision.DefaultDBPath(h.sdir, p1Host)+".gen", []byte("cur\n"), 0o644)
544 }},
545 } {
546 t.Run(c.name, func(t *testing.T) {
547 h := newRecoverHarness(t, false)
548 rb := h.released(t)
549 c.breakIt(h)
550 stuck := func() {
551 t.Helper()
552 m := h.read(t, p1Host).Move
553 if m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch || m.Rollback.StuckAt != provision.RbSourceVerified {
554 t.Fatalf("not stuck at verified: %+v %+v", m, m.Rollback)
555 }
556 if len(h.probes) != 0 {
557 t.Fatalf("probed past a mismatch: %+v", h.probes)
558 }
559 if h.srcRoute.Status != "restoring" || h.srcRoute.HoldOp != rb.ID {
560 t.Fatalf("the row: %+v", h.srcRoute)
561 }
562 if cl := h.claim(t); cl.Op.Phase != hostclaim.PhaseClaimed || cl.Op.Exec != "ie-box" {
563 t.Fatalf("claim: %+v", cl.Op)
564 }
565 }
566 h.srcPass(t)
567 stuck()
568 h.srcPass(t)
569 stuck()
570 // A retry verifies again: still broken, stuck again.
571 h.retry(t)
572 h.srcPass(t)
573 stuck()
574 c.mend(h)
575 h.retry(t)
576 h.srcPass(t)
577 h.opened(t)
578 if !strings.Contains(strings.Join(h.sEvents, " "), "stuck") {
579 t.Fatalf("events %v", h.sEvents)
580 }
581 })
582 }
583}
584
585// TestSourceRecoveryNeverReverifiesAfterProbing (Review Focus 2; INV-2,
586// INV-7): once the record says probing, the recovery's own writes are
587// authoritative: a probe that has not answered yet, or a restart, runs only
588// the probe again, never the verification or the install, whatever the
589// files say now. Nine failed attempts stop it for an operator at probing,
590// and a retry probes again.
591func TestSourceRecoveryNeverReverifiesAfterProbing(t *testing.T) {
592 t.Parallel()
593 h := newRecoverHarness(t, false)
594 h.released(t)
595 waiting := true
596 h.probe = func(req recoveryProbeRequest) (recoveryProbeOutcome, error) {
597 if err := req.Record(context.Background(), h.renderedInputs()); err != nil {
598 return recoveryProbeWaiting, err
599 }
600 if waiting {
601 return recoveryProbeWaiting, nil
602 }
603 return recoveryProbeAnswered, nil
604 }
605 h.srcPass(t)
606 h.sourceAt(t, provision.RbSourceProbing)
607 if c := h.claim(t); c.Op.Phase != hostclaim.PhaseProbing || c.Op.Exec != "ie-box" {
608 t.Fatalf("the claim is not probing before the attempt: %+v", c.Op)
609 }
610 // The source's files and marker change under the probe (its app wrote).
611 os.WriteFile(filepath.Join(h.sdir, p1Host, "upload.txt"), []byte("written by the recovered app"), 0o644)
612 os.WriteFile(provision.DefaultDBPath(h.sdir, p1Host)+".gen", []byte("fresh\n"), 0o644)
613 h.sw = nil // a restarted agent
614 h.srcPass(t)
615 h.sourceAt(t, provision.RbSourceProbing)
616 // Nine attempts used: stuck at probing; the retry probes again.
617 h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeExhausted, nil }
618 h.srcPass(t)
619 m := h.read(t, p1Host).Move
620 if m.State != provision.MoveStuck || m.StuckReason != provision.StuckRecoveryProbeAttempts || m.Rollback.StuckAt != provision.RbSourceProbing {
621 t.Fatalf("not stuck at probing: %+v %+v", m, m.Rollback)
622 }
623 h.retry(t)
624 waiting = false
625 h.probe = func(req recoveryProbeRequest) (recoveryProbeOutcome, error) {
626 if err := req.Record(context.Background(), h.renderedInputs()); err != nil {
627 return recoveryProbeWaiting, err
628 }
629 return recoveryProbeAnswered, nil
630 }
631 h.srcPass(t)
632 h.opened(t)
633 if n := len(h.probes); n != 4 {
634 t.Fatalf("probe attempts: %d", n)
635 }
636 verified := 0
637 for _, p := range h.rcHits {
638 if p == "verify" {
639 verified++
640 }
641 }
642 if verified != 1 {
643 t.Fatalf("verified %d times (%v)", verified, h.rcHits)
644 }
645}
646
647// TestRolledBackFinalizationReleasesTheClaim (§7.8 rolled_back): once the
648// source has opened and the target has parked, and no obligation of the op
649// remains, the source writes the terminal event, closes the record with the
650// rolled_back outcome and releases the claim to serving, in that order
651// (ruling M4). Until every chore is acknowledged it waits; a crash between
652// the record's close and the release is the console's to finish
653// (TestBoxClosesAreRecordFirst).
654func TestRolledBackFinalizationReleasesTheClaim(t *testing.T) {
655 t.Parallel()
656 ctx := context.Background()
657 h := newRecoverHarness(t, false)
658 h.released(t)
659 h.srcPass(t)
660 h.opened(t)
661 // The target cleans now that the source has recovered.
662 h.rollbackPass(t)
663 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaned || rb.TargetParkedAt == nil {
664 t.Fatalf("the target did not clean: %+v", rb)
665 }
666 // The parks are acknowledged by the regional runner (C4), not here:
667 // the source waits for them.
668 h.srcPass(t)
669 h.opened(t)
670 owed := 0
671 for _, o := range h.claim(t).Obligations {
672 if o.Op == p1ID {
673 owed++
674 }
675 }
676 if owed == 0 {
677 t.Fatal("the fixture owes no chore; the wait is not exercised")
678 }
679 if strings.Contains(strings.Join(h.sEvents, " "), "rolled_back") {
680 t.Fatalf("finalized with chores owed: %v", h.sEvents)
681 }
682 for _, o := range h.claim(t).Obligations {
683 if o.Op != p1ID {
684 continue
685 }
686 obs, err := hostclaim.Read(ctx, h.s, p1Host)
687 if err != nil {
688 t.Fatal(err)
689 }
690 if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.CompleteObligation(p1ID, o.ID)); err != nil {
691 t.Fatal(err)
692 }
693 }
694 h.srcPass(t)
695 c := h.claim(t)
696 if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "ie-box" {
697 t.Fatalf("the claim was not released: %+v", c.Op)
698 }
699 if err := hostclaim.RequireReleasedMove(p1ID)(c); err != nil {
700 t.Fatal(err)
701 }
702 rec := h.read(t, p1Host)
703 if rec.Move != nil || rec.LastMove == nil || rec.LastMove.ID != p1ID || rec.LastMove.Outcome != "rolled_back" || rec.Region != "ie" {
704 t.Fatalf("not closed: region %s move %+v last %+v", rec.Region, rec.Move, rec.LastMove)
705 }
706 if n := strings.Count(strings.Join(h.sEvents, " "), "rolled_back"); n != 1 {
707 t.Fatalf("the terminal event %d times: %v", n, h.sEvents)
708 }
709 if c := h.claim(t); c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "ie-box" {
710 t.Fatalf("claim after the close: %+v", c.Op)
711 }
712 // Nothing routes to the source any more.
713 h.srcPass(t)
714 if rec := h.read(t, p1Host); rec.Move != nil {
715 t.Fatalf("record %+v", rec.Move)
716 }
717}
718
719// Fix round 1, ruling 1: an evicted warm database. A clean return whose
720// warm database and marker are both gone is restored cold from src_gen by
721// its probe (J14), so verified passes; a database with no marker is refused
722// unless the generation is the origin; a marker naming another generation
723// is refused, with or without the database.
724func TestCleanReturnWarmDatabaseMarker(t *testing.T) {
725 t.Parallel()
726 for _, c := range []struct {
727 name string
728 db, marker bool
729 markerBody string
730 origin, stuck bool
731 }{
732 {name: "database and marker both evicted", db: false, marker: false},
733 {name: "database without its marker", db: true, marker: false, stuck: true},
734 {name: "database without its marker, the origin generation", db: true, marker: false, origin: true},
735 {name: "a conflicting marker beside the database", db: true, marker: true, markerBody: "other\n", stuck: true},
736 {name: "a conflicting marker, the database evicted", db: false, marker: true, markerBody: "other\n", stuck: true},
737 } {
738 t.Run(c.name, func(t *testing.T) {
739 h := newRecoverHarness(t, false)
740 h.released(t)
741 db := provision.DefaultDBPath(h.sdir, p1Host)
742 if c.origin {
743 rewriteP1Claim(t, h.s, func(cl *hostclaim.Claim) { cl.Op.Journal[hostclaim.JournalSrcGen] = "" })
744 }
745 if !c.db {
746 if err := os.Remove(db); err != nil {
747 t.Fatal(err)
748 }
749 }
750 if !c.marker {
751 if err := os.Remove(db + ".gen"); err != nil {
752 t.Fatal(err)
753 }
754 } else if err := os.WriteFile(db+".gen", []byte(c.markerBody), 0o644); err != nil {
755 t.Fatal(err)
756 }
757 h.srcPass(t)
758 if !c.stuck {
759 h.opened(t)
760 return
761 }
762 m := h.read(t, p1Host).Move
763 if m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch || m.Rollback.StuckAt != provision.RbSourceVerified || len(h.probes) != 0 {
764 t.Fatalf("not stuck at verified: %+v %+v probes %d", m, m.Rollback, len(h.probes))
765 }
766 })
767 }
768}
769
770// parkedJournal reports whether op's journal is live under the source's
771// .moves/, and whether it was parked into a retirement's artifacts.
772func (h *recoverHarness) parkedJournal(t *testing.T, op string) (live, parked bool) {
773 t.Helper()
774 live = moveJournalPresent(h.sdir, p1Host, op)
775 suffix := filepath.Join("artifacts", ".moves", p1Host, op)
776 filepath.WalkDir(filepath.Join(h.sdir, ".parked"), func(p string, d os.DirEntry, err error) error {
777 if err == nil && d.IsDir() && strings.HasSuffix(p, suffix) {
778 if _, err := os.Stat(filepath.Join(p, "journal.json")); err == nil {
779 parked = true
780 }
781 }
782 return nil
783 })
784 return live, parked
785}
786
787// closeOut drives a recovered source and its target to the close: the
788// target cleans, the chores are acknowledged (the regional runner's, C4),
789// and the source's next pass finalizes.
790func (h *recoverHarness) closeOut(t *testing.T) {
791 t.Helper()
792 ctx := context.Background()
793 h.rollbackPass(t)
794 if rb := h.rb(t); rb.TargetParkedAt == nil {
795 t.Fatalf("the target did not park: %+v", rb)
796 }
797 for _, o := range h.claim(t).Obligations {
798 if o.Op != p1ID {
799 continue
800 }
801 obs, err := hostclaim.Read(ctx, h.s, p1Host)
802 if err != nil {
803 t.Fatal(err)
804 }
805 if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.CompleteObligation(p1ID, o.ID)); err != nil {
806 t.Fatal(err)
807 }
808 }
809 h.srcPass(t)
810}
811
812// Fix round 1, ruling 2: the source keeps two journals through a rollback,
813// the cut's under the move id and the recovery's under the rollback id (a
814// dirty return's holds the install's build and extract scratch). The pass
815// that closes the record parks both; a crash right after the close leaves
816// them to the local journal sweep, which parks them after a restart on the
817// close's evidence (no listed record carries the move). Until the close's
818// preconditions hold, the sweep leaves them.
819func TestSourceJournalsParkedAfterTheClose(t *testing.T) {
820 t.Parallel()
821 for _, dirty := range []bool{false, true} {
822 name := "clean"
823 if dirty {
824 name = "dirty"
825 }
826 t.Run(name+", in the pass that closes", func(t *testing.T) {
827 h := newRecoverHarness(t, dirty)
828 rb := h.released(t)
829 h.srcPass(t)
830 h.opened(t)
831 for _, op := range []string{p1ID, rb.ID} {
832 if live, _ := h.parkedJournal(t, op); !live {
833 t.Fatalf("the journal of %s is gone before the close's preconditions", op)
834 }
835 }
836 h.srcPass(t)
837 for _, op := range []string{p1ID, rb.ID} {
838 if live, _ := h.parkedJournal(t, op); !live {
839 t.Fatalf("the sweep parked the journal of %s while the target still holds its state", op)
840 }
841 }
842 h.closeOut(t)
843 if rec := h.read(t, p1Host); rec.Move != nil {
844 t.Fatalf("not closed: %+v", rec.Move)
845 }
846 for _, op := range []string{p1ID, rb.ID} {
847 if live, parked := h.parkedJournal(t, op); live || !parked {
848 t.Fatalf("the journal of %s: live %v parked %v", op, live, parked)
849 }
850 }
851 })
852 t.Run(name+", a crash after the close", func(t *testing.T) {
853 h := newRecoverHarness(t, dirty)
854 rb := h.released(t)
855 h.srcPass(t)
856 h.opened(t)
857 if dirty {
858 if _, err := os.Stat(filepath.Join(moves.JournalDir(h.sdir, p1Host, rb.ID), "build")); err != nil {
859 t.Fatalf("the dirty return's build scratch: %v", err)
860 }
861 }
862 // The agent dies right after the close's write: nothing else of
863 // that pass runs, the journal sweep included.
864 crashed := false
865 h.rcHook = func(point string) {
866 if point == "after-close" {
867 crashed = true
868 panic(errCrash)
869 }
870 }
871 func() {
872 defer func() {
873 if r := recover(); r != nil && r != errCrash {
874 panic(r)
875 }
876 }()
877 h.closeOut(t)
878 }()
879 h.rcHook = nil
880 if !crashed {
881 t.Fatalf("never reached the close (%v)", h.rcHits)
882 }
883 if rec := h.read(t, p1Host); rec.Move != nil {
884 t.Fatalf("not closed: %+v", rec.Move)
885 }
886 for _, op := range []string{p1ID, rb.ID} {
887 if live, _ := h.parkedJournal(t, op); !live {
888 t.Fatalf("the journal of %s went with the crash", op)
889 }
890 }
891 h.sw = nil // a restarted agent
892 h.srcPass(t)
893 for _, op := range []string{p1ID, rb.ID} {
894 if live, parked := h.parkedJournal(t, op); live || !parked {
895 t.Fatalf("the journal of %s after the sweep: live %v parked %v", op, live, parked)
896 }
897 }
898 if entries, err := os.ReadDir(filepath.Join(h.sdir, ".moves", p1Host)); err == nil && len(entries) != 0 {
899 t.Fatalf("journals left: %v", entries)
900 }
901 })
902 }
903}
904
905// Fix round 1, ruling 3: every recovery step is replay-safe. A crash after
906// each of the claim's return, the guard's activation, the hold's hand-over,
907// the claim's probing and probed, and the opening of the row and of the
908// claim (and after each record step), then a restarted worker, ends opened
909// with one hold, opened once.
910func TestSourceRecoveryReplays(t *testing.T) {
911 slots := make(chan struct{}, 4)
912 for _, point := range []string{
913 "after-record-claimed", "after-return-claimed", "after-activate", "after-record-restoring", "after-hand",
914 "verify", "after-record-verified", "after-claim-probing", "after-record-probing", "after-claim-probed",
915 "after-record-recovered", "after-row-opened", "after-claim-opened", "after-record-opened",
916 } {
917 t.Run(point, func(t *testing.T) {
918 t.Parallel()
919 slots <- struct{}{}
920 defer func() { <-slots }()
921 h := newRecoverHarness(t, false)
922 rb := h.released(t)
923 h.rcAt = point
924 h.srcPass(t)
925 if !h.rcDone {
926 t.Fatalf("never reached %s (%v)", point, h.rcHits)
927 }
928 h.sw = nil
929 h.srcPass(t)
930 h.opened(t)
931 for _, hd := range h.srcHanded {
932 if hd != "restoring:"+rb.ID+":"+p1ID {
933 t.Fatalf("hand-overs %v", h.srcHanded)
934 }
935 }
936 if captured, err := hostguard.Capture(h.sdir, p1Host); err != nil || captured.Stamp != 1 {
937 t.Fatalf("the installation record: %+v %v", captured, err)
938 }
939 })
940 }
941}
942
943// Fix round 1, ruling 3: a row held by another op is never taken: the hand-
944// over at restoring and the open at recovered each refuse, and the record
945// stays where it is.
946func TestSourceRecoveryRefusesARowHeldByAnotherOp(t *testing.T) {
947 t.Parallel()
948 t.Run("at restoring", func(t *testing.T) {
949 h := newRecoverHarness(t, false)
950 h.released(t)
951 h.srcRoute.Status, h.srcRoute.HoldOp = "restoring", "op_someoneelse"
952 h.srcPass(t)
953 h.sourceAt(t, provision.RbSourceRestoring)
954 if h.srcRoute.HoldOp != "op_someoneelse" || len(h.probes) != 0 {
955 t.Fatalf("the row %+v, probes %d", h.srcRoute, len(h.probes))
956 }
957 })
958 t.Run("at recovered", func(t *testing.T) {
959 h := newRecoverHarness(t, false)
960 rb := h.released(t)
961 h.rcHook = func(point string) {
962 if point == "after-record-recovered" {
963 h.srcRoute.Status, h.srcRoute.HoldOp = "restoring", "op_someoneelse"
964 }
965 }
966 h.srcPass(t)
967 h.sourceAt(t, provision.RbSourceRecovered)
968 if h.srcRoute.HoldOp != "op_someoneelse" {
969 t.Fatalf("the row %+v", h.srcRoute)
970 }
971 if c := h.claim(t); c.Op.Phase != hostclaim.PhaseProbed || c.Op.Intent["rollback"] != rb.ID {
972 t.Fatalf("the claim opened over another op's row: %+v", c.Op)
973 }
974 })
975}
976