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 "bytes"
5 "context"
6 "database/sql"
7 "encoding/json"
8 "errors"
9 "io"
10 "os"
11 "path/filepath"
12 "strconv"
13 "strings"
14 "testing"
15 "time"
16
17 "github.com/carlosframework/platform/internal/hostclaim"
18 "github.com/carlosframework/platform/internal/hostguard"
19 "github.com/carlosframework/platform/internal/lease"
20 "github.com/carlosframework/platform/internal/moves"
21 "github.com/carlosframework/platform/internal/moves/archive"
22 "github.com/carlosframework/platform/internal/park"
23 "github.com/carlosframework/platform/internal/provision"
24 "github.com/carlosframework/platform/internal/registry"
25 "github.com/carlosframework/platform/internal/release"
26 "github.com/superfly/ltx"
27)
28
29// The rollback on the target (moves spec §7.8; Plan 4 Task 4): the decision
30// that ends a landing, then the target's release, its return lane and its
31// cleaning. The fixture is the landing harness landed through restored and
32// stepped to landing/probing on both the record and the claim (the probe is
33// Plan 3's and not on this branch), with the target box's rollback hooks
34// faked where main has no production provider in a test: the hold hand-over,
35// the final checkpoint, the local position and the row delete.
36
37type rollbackHarness struct {
38 *landingHarness
39 rw *movesWorker
40 handed []string // status:to:from of each hand-over asked for
41 rowGone bool
42 deleted []string
43 tchecks int // the target's final checkpoints
44 tpos uint64 // the target's local position
45 events []string
46 pages pagesSeen // the operator pages (moveCfg.Page), the target's and the source's
47 rbAt string
48 rbDone bool
49 rbHits []string
50 // rbHook runs at every fault point, before the crash check; beforeHand
51 // runs inside the hand-over, before its validation.
52 rbHook func(point string)
53 beforeHand func()
54}
55
56func newRollbackHarness(t *testing.T, opts ...p1Option) *rollbackHarness {
57 t.Helper()
58 h := &rollbackHarness{landingHarness: newLandingHarness(t, opts...)}
59 h.flip(t)
60 h.land(t)
61 h.installed(t)
62 h.step(t, provision.LandRestored, provision.LandProbing)
63 h.tpos = h.n
64 return h
65}
66
67// step moves the landing one phase on the record and the claim, as the
68// probe (Plan 3) would.
69func (h *rollbackHarness) step(t *testing.T, from, to string) {
70 t.Helper()
71 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
72 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveLandingPhase(g, from, to, nil)); err != nil {
73 t.Fatal(err)
74 }
75 obs, err := hostclaim.Read(context.Background(), h.s, p1Host)
76 if err != nil {
77 t.Fatal(err)
78 }
79 if _, err := hostclaim.Update(context.Background(), h.s, obs, hostclaim.SetMovePhase(p1ID, hostclaim.KindLanding, from, to)); err != nil {
80 t.Fatal(err)
81 }
82}
83
84// rollbackCfg is the target box's configuration with the rollback armed.
85func (h *rollbackHarness) rollbackCfg(t *testing.T) moveCfg {
86 t.Helper()
87 c := h.landingCfg(t)
88 c.OwnsRoute = func(host string) bool { return host == p1Host && !h.rowGone }
89 c.Route = func(host string) (*registry.Route, error) {
90 if host != p1Host || h.rowGone {
91 return nil, nil
92 }
93 rt := h.route
94 return &rt, nil
95 }
96 // The hand-over keeps handHold's contract: under the lock the row is
97 // read, then the caller's validation runs, then the hold moves.
98 c.HandHold = func(host, status, to, from string, validate func() error) (bool, error) {
99 h.handed = append(h.handed, status+":"+to+":"+from)
100 if h.rowGone {
101 return false, nil
102 }
103 if h.beforeHand != nil {
104 h.beforeHand()
105 }
106 if validate != nil {
107 if err := validate(); err != nil {
108 return false, err
109 }
110 }
111 if h.route.HoldOp == to && h.route.Status == status {
112 return true, nil
113 }
114 if h.route.HoldOp != from {
115 return false, nil
116 }
117 h.route.Status, h.route.HoldOp = status, to
118 return true, nil
119 }
120 c.FinalCheckpoint = func(ctx context.Context, host, _ string, holder hostclaim.Holder, check func(context.Context) error) error {
121 if err := check(ctx); err != nil {
122 return err
123 }
124 if host != p1Host || strings.TrimSuffix(holder.Location, "/") != cutRoot {
125 t.Errorf("checkpoint of %s at %s", host, holder.Location)
126 }
127 h.tchecks++
128 return nil
129 }
130 c.LocalPosition = func(string) (uint64, error) { return h.tpos, nil }
131 c.DeleteHeld = func(host, op string) (bool, error) {
132 h.deleted = append(h.deleted, host+"="+op)
133 if h.rowGone || h.route.HoldOp != op {
134 return false, nil
135 }
136 h.rowGone = true
137 return true, nil
138 }
139 c.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
140 h.events = append(h.events, ev.Event)
141 return nil
142 }
143 c.Page = h.pages.hook()
144 c.rbFault = func(point string) error {
145 h.rbHits = append(h.rbHits, point)
146 if h.rbHook != nil {
147 h.rbHook(point)
148 }
149 if point == h.rbAt && !h.rbDone {
150 h.rbDone = true
151 return errCrash
152 }
153 return nil
154 }
155 return c
156}
157
158// rollbackPass runs one pass as the target box, its worker kept across
159// passes.
160func (h *rollbackHarness) rollbackPass(t *testing.T) {
161 t.Helper()
162 if h.rw == nil {
163 h.rw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
164 }
165 h.rw.cfg = h.rollbackCfg(t)
166 if err := movesPass(context.Background(), h.s, h.rw); err != nil {
167 t.Fatal(err)
168 }
169}
170
171// rollBack passes the landing's deadline and runs the target's pass that
172// decides the rollback.
173func (h *rollbackHarness) rollBack(t *testing.T) *provision.MoveRollback {
174 t.Helper()
175 h.clock.Advance(cutLandingWindow + time.Minute)
176 h.rollbackPass(t)
177 m := h.read(t, p1Host).Move
178 if m.State != provision.MoveRollingBack || m.Rollback == nil {
179 t.Fatalf("no rollback: %+v", m)
180 }
181 return m.Rollback
182}
183
184func (h *rollbackHarness) rb(t *testing.T) *provision.MoveRollback {
185 t.Helper()
186 m := h.read(t, p1Host).Move
187 if m == nil || m.Rollback == nil {
188 t.Fatalf("record: %+v", m)
189 }
190 return m.Rollback
191}
192
193// uncover marks the staged seal as not covering the source (§7.4 rev 7.11):
194// the source's own replica did not reach its stopped position.
195func (h *rollbackHarness) uncover(t *testing.T) {
196 t.Helper()
197 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
198 s, err := journalledSeal(c)
199 if err != nil {
200 t.Fatal(err)
201 }
202 s.SrcCovered = false
203 c.Op.Journal[hostclaim.JournalSeal] = s.Encode()
204 })
205}
206
207// recovered drives the source's half of the record to recovered, stamping
208// RecoveredAt, as Task 5's source would.
209func (h *rollbackHarness) recovered(t *testing.T) {
210 t.Helper()
211 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
212 path := []string{provision.RbSourceWaiting, provision.RbSourceClaimed, provision.RbSourceRestoring, provision.RbSourceVerified, provision.RbSourceProbing}
213 if h.rb(t).Dirty {
214 path = []string{provision.RbSourceWaiting, provision.RbSourceClaimed, provision.RbSourceRestoring, provision.RbSourceParking, provision.RbSourceBuilding,
215 provision.RbSourcePublished, provision.RbSourceLocal, provision.RbSourceConfigured, provision.RbSourceRestored, provision.RbSourceProbing}
216 }
217 path = append(path, provision.RbSourceProbed, provision.RbSourceRecovered)
218 for i := 0; i+1 < len(path); i++ {
219 var stamp func(*provision.MoveRollback)
220 if path[i+1] == provision.RbSourceRecovered {
221 now := h.clock.Now()
222 stamp = func(rb *provision.MoveRollback) { rb.RecoveredAt = &now }
223 }
224 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRollbackSourcePhase(g, path[i], path[i+1], stamp)); err != nil {
225 t.Fatalf("%s to %s: %v", path[i], path[i+1], err)
226 }
227 }
228}
229
230// rolledBack asserts the rollback decision took the landing: the record
231// rolling_back in the source's region with reason, the landing phase kept,
232// the target releasing; the claim restoring/releasing under the rollback.
233func rolledBack(t *testing.T, h *landingHarness, reason, phase string) {
234 t.Helper()
235 rec := h.read(t, p1Host)
236 m := rec.Move
237 if rec.Region != "ie" || m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.Reason != reason ||
238 m.LandingPhase != phase || m.Rollback.TargetPhase != provision.RbTargetReleasing {
239 t.Fatalf("not rolled back for %q at %s: %s %+v %+v", reason, phase, rec.Region, m, m.Rollback)
240 }
241 if c := h.claim(t); c.Op.Kind != hostclaim.KindRestoring || c.Op.Phase != hostclaim.PhaseReleasing || c.Op.Intent["rollback"] != m.Rollback.ID {
242 t.Fatalf("the claim did not follow: %+v", c.Op)
243 }
244}
245
246func targetStoppedKey() string {
247 return moves.StoppedKey(moves.ReportPrefix("bab", "app", p1Host, p1ID), "target")
248}
249
250// laneCalls are the guarded staging requests into the source's bucket.
251func (h *rollbackHarness) laneCalls(rb string) []stagingCall {
252 var out []stagingCall
253 for _, c := range h.f.stagingCalls() {
254 if c.Bucket == "data-ie" && strings.HasPrefix(c.Key, "moves/"+p1Host+"/rb_") {
255 if !strings.HasPrefix(c.Key, moves.ReturnLane(p1Host, rb)) {
256 panic("a lane write outside the rollback's lane: " + c.Key)
257 }
258 out = append(out, c)
259 }
260 }
261 return out
262}
263
264// TestRollbackDecisionConvertsTheMark: a marked landing is rolled back with
265// the mark's reason, in the one place the mark is converted, before the
266// landing's arming gate (J10); the claim follows (RollBack, or
267// RollBackFlipped over a claim the flip's half never reached, J16), with the
268// target installation's stamp F journalled (J15).
269func TestRollbackDecisionConvertsTheMark(t *testing.T) {
270 t.Parallel()
271 t.Run("the claim at landing", func(t *testing.T) {
272 h := newRollbackHarness(t)
273 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
274 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveMarkRollback(g, provision.RollbackProbeAttempts)); err != nil {
275 t.Fatal(err)
276 }
277 f := h.claim(t).Holder.Stamp
278 // Unarmed: the decision runs all the same.
279 h.rw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
280 c := h.rollbackCfg(t)
281 c.ParkLeases = nil
282 h.rw.cfg = c
283 if err := movesPass(context.Background(), h.s, h.rw); err != nil {
284 t.Fatal(err)
285 }
286 rec := h.read(t, p1Host)
287 m := rec.Move
288 if m.State != provision.MoveRollingBack || rec.Region != "ie" || m.Rollback.Reason != provision.RollbackProbeAttempts ||
289 m.Rollback.TargetPhase != provision.RbTargetReleasing || !provision.ValidRollbackID(m.Rollback.ID) || m.LandingPhase != provision.LandProbing {
290 t.Fatalf("record: %s %+v %+v", rec.Region, m, m.Rollback)
291 }
292 cl := h.claim(t)
293 if cl.Op.Kind != hostclaim.KindRestoring || cl.Op.Phase != hostclaim.PhaseReleasing || cl.Op.Intent["rollback"] != m.Rollback.ID ||
294 cl.Op.Exec != "uk-box" || cl.Op.Journal[hostclaim.JournalTargetStamp] != strconv.FormatUint(f, 10) {
295 t.Fatalf("claim: %+v", cl.Op)
296 }
297 if len(h.events) != 1 || h.events[0] != "rolling_back" {
298 t.Fatalf("events %v", h.events)
299 }
300 })
301 t.Run("the claim still moving", func(t *testing.T) {
302 h := &rollbackHarness{landingHarness: newLandingHarness(t)}
303 h.ready(t)
304 h.crashAt = "after-record-flip"
305 h.cut(t)
306 h.crashPass(t)
307 if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Phase != hostclaim.PhaseSealed {
308 t.Fatalf("fixture claim: %+v", c.Op)
309 }
310 h.rollBack(t)
311 m := h.read(t, p1Host).Move
312 if m.Rollback.Reason != provision.RollbackLandingDeadline || m.LandingPhase != provision.LandChecking {
313 t.Fatalf("record: %+v", m.Rollback)
314 }
315 if c := h.claim(t); c.Op.Kind != hostclaim.KindRestoring || c.Op.Phase != hostclaim.PhaseReleasing || c.Op.Exec != "uk-box" ||
316 c.Op.Journal[hostclaim.JournalSrcGen] != "cur" || c.Op.Intent["rollback"] != m.Rollback.ID {
317 t.Fatalf("claim: %+v", c.Op)
318 }
319 })
320}
321
322// TestRollbackCASLosesToCommit: the rollback and the commit race on the
323// record; once the commit has won, the rollback's CAS is refused and leaves
324// no trace: the record committed with no rollback, the claim and the row as
325// they were. The open is the commit's path (Plan 3 Task 7).
326func TestRollbackCASLosesToCommit(t *testing.T) {
327 t.Parallel()
328 ctx := context.Background()
329 h := newRollbackHarness(t)
330 h.step(t, provision.LandProbing, provision.LandProbed)
331 stale := h.read(t, p1Host)
332 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
333 if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveCommit(g, h.clock.Now())); err != nil {
334 t.Fatal(err)
335 }
336 before, err := hostclaim.Read(ctx, h.s, p1Host)
337 if err != nil {
338 t.Fatal(err)
339 }
340 h.clock.Advance(cutLandingWindow + time.Minute)
341 w := &movesWorker{cfg: h.rollbackCfg(t), mirrors: map[string]lease.MirrorStore{}}
342 done, err := w.rollBackLanding(ctx, h.s, stale)
343 if err != nil || !done {
344 t.Fatalf("the decision over a stale landing: %v %v", done, err)
345 }
346 rec := h.read(t, p1Host)
347 if rec.Move.State != provision.MoveCommitted || rec.Move.Rollback != nil || rec.Region != "uk" {
348 t.Fatalf("record: %s %+v", rec.Region, rec.Move)
349 }
350 after, err := hostclaim.Read(ctx, h.s, p1Host)
351 if err != nil {
352 t.Fatal(err)
353 }
354 if after.ETag != before.ETag || after.Claim.Op.Kind != hostclaim.KindLanding {
355 t.Fatalf("the claim was written: %+v", after.Claim.Op)
356 }
357 if len(h.handed) != 0 || h.route.HoldOp != p1ID || len(h.events) != 0 {
358 t.Fatalf("the rollback left a trace: %v %+v %v", h.handed, h.route, h.events)
359 }
360 // A pass over the committed record leaves it too.
361 h.rollbackPass(t)
362 if rec := h.read(t, p1Host); rec.Move.State != provision.MoveCommitted || rec.Move.Rollback != nil {
363 t.Fatalf("after a pass: %+v", rec.Move)
364 }
365}
366
367// TestReleaseIsCleanWhenNothingChanged: the landing reached probing, but
368// nothing the target holds differs from what it installed (the database,
369// the files, the timers) and the staged seal covers the source: the release
370// is clean. Execution goes to none citing stopped-target.json, the target
371// region's whole location is queued for its park with the target's
372// park-local, and the target region's predecessors are covered.
373func TestReleaseIsCleanWhenNothingChanged(t *testing.T) {
374 t.Parallel()
375 ctx := context.Background()
376 h := newRollbackHarness(t)
377 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
378 c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Region: "uk", PredecessorRegion: "uk", Session: 3,
379 Box: "uk-box", Generation: "probe", Predecessor: "cur", Mode: "cold"})
380 })
381 f := h.claim(t).Holder.Stamp
382 rb := h.rollBack(t)
383 fences := h.fences
384 if h.route.HoldOp != p1ID || len(h.handed) != 0 {
385 t.Fatalf("the decision touched the row: %+v %v", h.route, h.handed)
386 }
387 h.rollbackPass(t)
388 rb = h.rb(t)
389 if rb.TargetPhase != provision.RbTargetReleased || rb.Dirty || rb.Evidence != targetStoppedKey() {
390 t.Fatalf("rollback: %+v", rb)
391 }
392 if h.route.Status != "releasing" || h.route.HoldOp != rb.ID || h.fences != fences+1 || h.tchecks != 1 {
393 t.Fatalf("row %+v fences %d checkpoints %d", h.route, h.fences, h.tchecks)
394 }
395 report, err := readStoppedReport(ctx, h.s, rb.Evidence)
396 if err != nil {
397 t.Fatal(err)
398 }
399 m := h.read(t, p1Host).Move
400 if report.Role != "target" || report.Box != "uk-box" || report.OpID != p1ID || report.FinalTXID != h.n ||
401 report.DBSHA256 != m.Landing.InstalledDBSHA256 || report.FilesPlain != m.Source.Files.Plain || report.TimersSHA256 != m.Source.TimersSHA256 {
402 t.Fatalf("report %+v", report)
403 }
404 c := h.claim(t)
405 if c.Op.Phase != hostclaim.PhaseReleased || c.Op.Exec != "" || c.Op.Journal[hostclaim.JournalEvidenceTarget] != rb.Evidence {
406 t.Fatalf("claim: %+v", c.Op)
407 }
408 var location, local int
409 for _, o := range c.Obligations {
410 switch {
411 case o.Kind == "park-replica" && o.Region == "uk" && o.Park.Location == cutRoot:
412 location++
413 if !o.Park.EvidenceAt.Equal(report.CheckedAt) || o.Park.RequestLimit != cutRequestLimit || o.Park.Incarnation != h.inc {
414 t.Fatalf("the location park's scope: %+v", o.Park)
415 }
416 case o.Kind == "park-local" && o.Box == "uk-box":
417 local++
418 default:
419 t.Fatalf("an obligation the release did not owe: %+v", o)
420 }
421 }
422 if location != 1 || local != 1 {
423 t.Fatalf("obligations: %+v", c.Obligations)
424 }
425 if !hostclaim.InventoryHasStamp(c, "uk-box", f) {
426 t.Fatal("the fixture's inventory does not list F")
427 }
428 for _, a := range c.Holder.Admissions {
429 if a.PredecessorRegion == "uk" && !a.Covered {
430 t.Fatalf("the target region's predecessor is not covered: %+v", a)
431 }
432 }
433 // Nothing goes to the lane, and the target waits for the source.
434 if calls := h.laneCalls(rb.ID); len(calls) != 0 {
435 t.Fatalf("a clean release wrote the lane: %+v", calls)
436 }
437 h.rollbackPass(t)
438 if got := h.rb(t); got.TargetPhase != provision.RbTargetReleased {
439 t.Fatalf("a clean release moved on before the source recovered: %+v", got)
440 }
441 if strings.Join(h.events, " ") != "rolling_back released" {
442 t.Fatalf("events %v", h.events)
443 }
444}
445
446// A landing rolled back before probing is clean when the seal covers the
447// source, whatever its digests: no writer ever ran on it. Nothing is
448// checkpointed.
449func TestReleaseBeforeProbingNeedsNoDigests(t *testing.T) {
450 t.Parallel()
451 h := &rollbackHarness{landingHarness: newLandingHarness(t)}
452 h.flip(t)
453 h.landAt = "after-build"
454 h.crashLand(t)
455 h.rollBack(t)
456 h.rollbackPass(t)
457 rb := h.rb(t)
458 if rb.TargetPhase != provision.RbTargetReleased || rb.Dirty || h.tchecks != 0 {
459 t.Fatalf("rollback %+v checkpoints %d", rb, h.tchecks)
460 }
461 if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandBuilding {
462 t.Fatalf("landing phase %s", m.LandingPhase)
463 }
464}
465
466// TestReleaseIsDirtyWithoutSrcCovered (Review Focus 1, witness wSealReturn):
467// the digests are equal but the staged seal does not cover the source, so
468// the source cannot trust its own copy: the release is dirty, and the
469// return lane gets the staged seal's exact versions, as given.
470func TestReleaseIsDirtyWithoutSrcCovered(t *testing.T) {
471 t.Parallel()
472 ctx := context.Background()
473 h := newRollbackHarness(t)
474 h.uncover(t)
475 staged := h.seal(t)
476 h.rollBack(t)
477 h.rollbackPass(t)
478 rb := h.rb(t)
479 if !rb.Dirty || rb.TargetPhase != provision.RbTargetReturned {
480 t.Fatalf("rollback %+v", rb)
481 }
482 c := h.claim(t)
483 seal, err := moves.DecodeSeal(c.Op.Journal[moves.JournalReturnSeal])
484 if err != nil {
485 t.Fatal(err)
486 }
487 if !seal.SrcCovered || !seal.FromDB || seal.N != staged.N || !seal.Bound().Equal(h.lease.until) {
488 t.Fatalf("return seal %+v (lease until %s)", seal, h.lease.until)
489 }
490 lane, from := moves.ReturnLane(p1Host, rb.ID), moves.StagingPrefix(p1Host, p1ID)
491 got := map[string]moves.SealedObject{}
492 for _, o := range append(append(append([]moves.SealedObject(nil), seal.Replica...), seal.Files...), seal.Timers...) {
493 got[o.Key] = o
494 }
495 for _, o := range append(append(append([]moves.SealedObject(nil), staged.Replica...), staged.Files...), staged.Timers...) {
496 key := lane + strings.TrimPrefix(o.Key, from)
497 r, ok := got[key]
498 if !ok {
499 t.Fatalf("the return seal does not name %s", key)
500 }
501 want, err := h.vs.GetVersion(ctx, "data-uk", o.Key, o.VersionID)
502 if err != nil {
503 t.Fatal(err)
504 }
505 body, err := h.vs.GetVersion(ctx, "data-ie", key, r.VersionID)
506 if err != nil {
507 t.Fatalf("%s at %s: %v", key, r.VersionID, err)
508 }
509 if !bytes.Equal(body, want) || r.Size != o.Size {
510 t.Fatalf("%s is not the staged version of %s", key, o.Key)
511 }
512 delete(got, key)
513 }
514 if len(got) != 0 {
515 t.Fatalf("the return seal names more than the staged seal: %v", got)
516 }
517 // The baseline is the source's own, under the target's evidence.
518 base, err := moves.DecodeReturnBaseline(c.Op.Journal[moves.JournalReturnBaseline])
519 if err != nil {
520 t.Fatal(err)
521 }
522 src := h.read(t, p1Host).Move.Source
523 if base.FinalTXID != staged.N || *base.Files != *src.Files || base.TimersSHA256 != src.TimersSHA256 || base.Evidence != rb.Evidence {
524 t.Fatalf("baseline %+v, source %+v", base, src)
525 }
526 var mj moves.MoveJSON
527 if body := h.f.buckets["data-ie"][moves.MoveJSONKey(p1Host, rb.ID)].body; json.Unmarshal([]byte(body), &mj) != nil || mj.Incarnation != h.inc || mj.MoveID != rb.ID {
528 t.Fatalf("move.json %q", body)
529 }
530 // Every lane object went through the guard, under the source's lease.
531 if calls := h.laneCalls(rb.ID); len(calls) != 1+len(staged.Replica)+len(staged.Files)+len(staged.Timers) {
532 t.Fatalf("lane calls %+v", calls)
533 }
534 if h.lease.acquired == 0 || h.lease.acquired != h.lease.released {
535 t.Fatalf("the staging lease: acquired %d released %d", h.lease.acquired, h.lease.released)
536 }
537 if strings.Join(h.events, " ") != "rolling_back released returning returned" {
538 t.Fatalf("events %v", h.events)
539 }
540}
541
542// decodeLTXDatabase writes the pages of an LTX snapshot into a database file.
543func decodeLTXDatabase(t *testing.T, body []byte, path string) ltx.Header {
544 t.Helper()
545 dec := ltx.NewDecoder(bytes.NewReader(body))
546 if err := dec.DecodeHeader(); err != nil {
547 t.Fatal(err)
548 }
549 hdr := dec.Header()
550 f, err := os.Create(path)
551 if err != nil {
552 t.Fatal(err)
553 }
554 defer f.Close()
555 page := make([]byte, hdr.PageSize)
556 for {
557 var ph ltx.PageHeader
558 if err := dec.DecodePage(&ph, page); errors.Is(err, io.EOF) {
559 break
560 } else if err != nil {
561 t.Fatal(err)
562 }
563 if _, err := f.WriteAt(page, int64(ph.Pgno-1)*int64(hdr.PageSize)); err != nil {
564 t.Fatal(err)
565 }
566 }
567 if err := dec.Close(); err != nil {
568 t.Fatal(err)
569 }
570 if err := f.Truncate(int64(hdr.Commit) * int64(hdr.PageSize)); err != nil {
571 t.Fatal(err)
572 }
573 return hdr
574}
575
576// TestDirtyReturnSnapshotsTheStoppedDatabase (witness wDirtyReturn): the
577// target served and wrote, so its state is not the staged seal's. The lane
578// carries one snapshot [1, N'] of the stopped database, its files archive
579// and its timers; the snapshot verifies and restores the target's rows, and
580// the return seal names exactly it.
581func TestDirtyReturnSnapshotsTheStoppedDatabase(t *testing.T) {
582 t.Parallel()
583 ctx := context.Background()
584 h := newRollbackHarness(t)
585 dbPath := provision.DefaultDBPath(h.tdir, p1Host)
586 d, err := sql.Open("sqlite", dbPath)
587 if err != nil {
588 t.Fatal(err)
589 }
590 for i := 0; i < 3; i++ {
591 if _, err := d.Exec(`INSERT INTO t (body) VALUES ('written on the target')`); err != nil {
592 t.Fatal(err)
593 }
594 }
595 d.Close()
596 if err := os.WriteFile(filepath.Join(h.tdir, p1Host, "new.txt"), []byte("uploaded on the target"), 0o644); err != nil {
597 t.Fatal(err)
598 }
599 h.tpos = h.n + 3
600 h.rollBack(t)
601 h.rollbackPass(t)
602 rb := h.rb(t)
603 if !rb.Dirty || rb.TargetPhase != provision.RbTargetReturned {
604 t.Fatalf("rollback %+v", rb)
605 }
606 c := h.claim(t)
607 seal, err := moves.DecodeSeal(c.Op.Journal[moves.JournalReturnSeal])
608 if err != nil {
609 t.Fatal(err)
610 }
611 lane := moves.ReturnLane(p1Host, rb.ID)
612 want := moves.SnapshotKey(lane+"replica/", h.tpos)
613 if seal.N != h.tpos || len(seal.Replica) != 1 || seal.Replica[0].Key != want || !seal.ReachesN() || !seal.FromDB || !seal.SrcCovered {
614 t.Fatalf("return seal %+v, want the snapshot %s", seal, want)
615 }
616 body, err := h.vs.GetVersion(ctx, "data-ie", want, seal.Replica[0].VersionID)
617 if err != nil {
618 t.Fatal(err)
619 }
620 if err := ltx.NewDecoder(bytes.NewReader(body)).Verify(); err != nil {
621 t.Fatalf("the snapshot does not verify: %v", err)
622 }
623 restored := filepath.Join(t.TempDir(), "restored.db")
624 hdr := decodeLTXDatabase(t, body, restored)
625 if hdr.MinTXID != 1 || uint64(hdr.MaxTXID) != h.tpos {
626 t.Fatalf("snapshot header %+v", hdr)
627 }
628 if got := countRowsForTest(t, restored); got != h.rows+3 {
629 t.Fatalf("the snapshot holds %d rows, the target %d", got, h.rows+3)
630 }
631 // The files archive is the target's, bound to the move id (the
632 // installer checks it), and the baseline is the target's state.
633 base, err := moves.DecodeReturnBaseline(c.Op.Journal[moves.JournalReturnBaseline])
634 if err != nil {
635 t.Fatal(err)
636 }
637 plain, _, _, err := archive.PlainDigest(filepath.Join(h.tdir, p1Host), filepath.Join(h.tdir, p1Host+".env"))
638 if err != nil {
639 t.Fatal(err)
640 }
641 dbSHA, err := fileSHA256(dbPath)
642 if err != nil {
643 t.Fatal(err)
644 }
645 if base.FinalTXID != h.tpos || base.DBSHA256 != dbSHA || base.Files.Plain != plain || base.Files.Sealed == "" || base.Evidence != rb.Evidence {
646 t.Fatalf("baseline %+v", base)
647 }
648 if !sealNamesArchive(seal, base.Files.Sealed) {
649 t.Fatalf("the seal's files are not the baseline's archive: %+v", seal.Files)
650 }
651 for _, o := range seal.Files {
652 if !strings.HasSuffix(o.Key, ".json") {
653 continue
654 }
655 raw, err := h.vs.GetVersion(ctx, "data-ie", o.Key, o.VersionID)
656 if err != nil {
657 t.Fatal(err)
658 }
659 var am archive.Manifest
660 if err := json.Unmarshal(raw, &am); err != nil || am.Op != p1ID || am.Incarnation != h.inc || am.Plain != plain {
661 t.Fatalf("archive manifest %+v %v", am, err)
662 }
663 }
664 if len(seal.Timers) != 1 || !strings.HasPrefix(seal.Timers[0].Key, lane+"timers/") {
665 t.Fatalf("timers %+v", seal.Timers)
666 }
667}
668
669// TestReturnLaneWriterIsGuarded: the return lane is a staging lane. A writer
670// whose lease is no longer safe for a signature's life sends nothing more; a
671// writer that finishes records its lease's bound as the seal's, after every
672// request it signed can still land, and the lane's park acknowledges at
673// LeaseUntil + H, not before.
674func TestReturnLaneWriterIsGuarded(t *testing.T) {
675 t.Parallel()
676 ctx := context.Background()
677 h := newRollbackHarness(t)
678 h.uncover(t)
679 h.rollBack(t)
680 // The lease goes unsafe after the first request is signed (a writer
681 // paused past its bound): nothing after it is sent, and no seal.
682 h.lease.mu.Lock()
683 h.lease.reads, h.lease.unsafeAfter = 0, 1
684 h.lease.mu.Unlock()
685 h.rollbackPass(t)
686 rb := h.rb(t)
687 if rb.TargetPhase != provision.RbTargetReturning {
688 t.Fatalf("rollback %+v", rb)
689 }
690 if calls := h.laneCalls(rb.ID); len(calls) != 1 {
691 t.Fatalf("an unsafe writer sent %d requests: %+v", len(calls), calls)
692 }
693 if _, ok := h.claim(t).Op.Journal[moves.JournalReturnSeal]; ok {
694 t.Fatal("a return seal without the lane")
695 }
696 if h.lease.acquired != h.lease.released {
697 t.Fatalf("the lease is still held: %d %d", h.lease.acquired, h.lease.released)
698 }
699 // A safe lease: the lane is written and sealed with the lease's bound.
700 h.lease.mu.Lock()
701 h.lease.unsafeAfter = 0
702 h.lease.mu.Unlock()
703 h.rollbackPass(t)
704 if rb = h.rb(t); rb.TargetPhase != provision.RbTargetReturned {
705 t.Fatalf("rollback %+v", rb)
706 }
707 seal, err := moves.DecodeSeal(h.claim(t).Op.Journal[moves.JournalReturnSeal])
708 if err != nil {
709 t.Fatal(err)
710 }
711 if !seal.Bound().Equal(h.lease.until) {
712 t.Fatalf("bound %s, lease until %s", seal.Bound(), h.lease.until)
713 }
714 for _, c := range h.laneCalls(rb.ID) {
715 if c.SignAt.Add(moves.SignatureLife + moves.DefaultSkew).After(seal.Bound()) {
716 t.Fatalf("%s signed at %s can land after the bound %s", c.Key, c.SignAt, seal.Bound())
717 }
718 }
719 // The lane's park (Plan 5): queued at the source's probed with the
720 // seal's bound as its horizon, acknowledged only after it.
721 for _, step := range []func(*hostclaim.Claim) error{
722 hostclaim.ReturnToSource(p1ID, "ie-box", false, hostclaim.ReturnFence{}),
723 hostclaim.SetRestoringPhase(p1ID, hostclaim.PhaseClaimed, hostclaim.PhaseProbing),
724 hostclaim.SetRestoringPhase(p1ID, hostclaim.PhaseProbing, hostclaim.PhaseProbed),
725 hostclaim.QueueReturnLanePark(p1ID, "ie", hostclaim.ParkScope{Location: moves.ScopeLocation(p1Host, rb.ID), Incarnation: h.inc,
726 EvidenceAt: seal.Bound(), RequestLimit: cutRequestLimit}, obligationID),
727 } {
728 obs, err := hostclaim.Read(ctx, h.s, p1Host)
729 if err != nil {
730 t.Fatal(err)
731 }
732 if _, err := hostclaim.Update(ctx, h.s, obs, step); err != nil {
733 t.Fatal(err)
734 }
735 }
736 var chore hostclaim.Obligation
737 for _, o := range h.claim(t).Obligations {
738 if o.Kind == "park-replica" && o.Region == "ie" {
739 chore = o
740 }
741 }
742 e := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-ie", Region: "ie", Leases: h.leases, Now: h.clock.Now, NewID: obligationID,
743 // Plan 5's view stands in here; a lane is never the holder's
744 // location, so nothing in it is protected either way.
745 View: func(_ context.Context, obs hostclaim.Observed) (park.View, error) {
746 return park.View{ClaimETag: obs.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true,
747 GenerationOf: func(park.Ref) (string, bool) { return "", true }}, nil
748 },
749 Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
750 work := park.Work{Host: p1Host, ID: chore.ID, Op: p1ID}
751 horizon := seal.Bound().Add(park.SignatureAge + cutRequestLimit)
752 h.clock.mu.Lock()
753 h.clock.now = horizon.Add(-time.Second)
754 h.clock.mu.Unlock()
755 if err := e.Run(ctx, work); !errors.Is(err, park.ErrPending) {
756 t.Fatalf("before LeaseUntil + H: %v", err)
757 }
758 if parked, err := e.Parked(ctx, work); err != nil || !parked {
759 t.Fatalf("the lane's archive: %v %v", parked, err)
760 }
761 h.clock.Advance(2 * time.Second)
762 if err := e.Run(ctx, work); err != nil {
763 t.Fatalf("after LeaseUntil + H: %v", err)
764 }
765 for _, o := range h.claim(t).Obligations {
766 if o.ID == chore.ID {
767 t.Fatalf("the lane's park was not acknowledged: %+v", o)
768 }
769 }
770}
771
772// TestTargetCleaningWaitsForRecoveredAt (Review Focus 4): the target cleans
773// only once the source has recovered; until then it keeps everything it has,
774// at released (clean) or returned (dirty).
775func TestTargetCleaningWaitsForRecoveredAt(t *testing.T) {
776 t.Parallel()
777 for _, dirty := range []bool{false, true} {
778 t.Run("dirty "+strconv.FormatBool(dirty), func(t *testing.T) {
779 h := newRollbackHarness(t)
780 if dirty {
781 h.uncover(t)
782 }
783 h.rollBack(t)
784 want := provision.RbTargetReleased
785 if dirty {
786 want = provision.RbTargetReturned
787 }
788 for i := 0; i < 3; i++ {
789 h.rollbackPass(t)
790 if rb := h.rb(t); rb.TargetPhase != want || rb.TargetParkedAt != nil {
791 t.Fatalf("pass %d: %+v", i, rb)
792 }
793 }
794 if h.rowGone || len(h.deleted) != 0 {
795 t.Fatal("the row went before the source recovered")
796 }
797 if _, err := os.Stat(provision.DefaultDBPath(h.tdir, p1Host)); err != nil {
798 t.Fatalf("the database went before the source recovered: %v", err)
799 }
800 h.recovered(t)
801 h.rollbackPass(t)
802 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaned || rb.TargetParkedAt == nil {
803 t.Fatalf("after the recovery: %+v", rb)
804 }
805 })
806 }
807}
808
809// TestTargetCleaningLeavesNoLiveCopy (INV-10): after cleaned nothing of the
810// host is live on the target, in its bucket under instances/<host>/ or
811// moves/<host>/<mv>/, or on its disk; everything is archived under
812// parked/<inc>/<host>/<retirement>/ (and .parked/ locally), the row and its
813// timers are gone, and the park-local chore is complete.
814func TestTargetCleaningLeavesNoLiveCopy(t *testing.T) {
815 t.Parallel()
816 h := newRollbackHarness(t)
817 db := provision.DefaultDBPath(h.tdir, p1Host)
818 // A WAL the inventory does not list: the moves-owned rename parks it.
819 if err := os.WriteFile(db+"-wal", []byte("a wal"), 0o600); err != nil {
820 t.Fatal(err)
821 }
822 h.rollBack(t)
823 h.rollbackPass(t)
824 rb := h.rb(t)
825 h.recovered(t)
826 h.rollbackPass(t)
827 if rb = h.rb(t); rb.TargetPhase != provision.RbTargetCleaned || rb.TargetParkedAt == nil {
828 t.Fatalf("rollback %+v", rb)
829 }
830 var live, parked []string
831 for k := range h.f.buckets["data-uk"] {
832 switch {
833 case strings.HasPrefix(k, cutRoot+"/") && !strings.HasSuffix(k, "/lock.json"),
834 strings.HasPrefix(k, moves.StagingPrefix(p1Host, p1ID)):
835 live = append(live, k)
836 case strings.HasPrefix(k, "parked/"+h.inc+"/"+p1Host+"/"):
837 parked = append(parked, k)
838 }
839 }
840 if len(live) != 0 || len(parked) == 0 {
841 t.Fatalf("live %v parked %d", live, len(parked))
842 }
843 for _, p := range sourceArtifactPaths(p1Host) {
844 if _, err := os.Lstat(filepath.Join(h.tdir, p)); !errors.Is(err, os.ErrNotExist) {
845 t.Fatalf("%s is still live: %v", p, err)
846 }
847 }
848 if _, err := os.Lstat(moves.JournalDir(h.tdir, p1Host, p1ID)); !errors.Is(err, os.ErrNotExist) {
849 t.Fatalf("the op journal is still live: %v", err)
850 }
851 var wal, dbs int
852 filepath.WalkDir(filepath.Join(h.tdir, ".parked"), func(p string, d os.DirEntry, err error) error {
853 if err == nil && !d.IsDir() {
854 switch filepath.Base(p) {
855 case p1Host + ".db-wal":
856 wal++
857 case p1Host + ".db":
858 dbs++
859 }
860 }
861 return nil
862 })
863 if wal != 1 || dbs != 1 {
864 t.Fatalf("parked locally: wal %d db %d", wal, dbs)
865 }
866 if !h.rowGone || len(h.deleted) != 1 || h.deleted[0] != p1Host+"="+rb.ID {
867 t.Fatalf("the row: %v", h.deleted)
868 }
869 c := h.claim(t)
870 for _, o := range c.Obligations {
871 if o.Kind == "park-local" {
872 t.Fatalf("the park-local is still owed: %+v", o)
873 }
874 }
875 if len(c.Inventory["uk-box"].Paths) != 0 {
876 t.Fatalf("inventory %+v", c.Inventory["uk-box"])
877 }
878 if captured, err := hostguard.Capture(h.tdir, p1Host); err != nil || captured.Stamp != 0 {
879 t.Fatalf("the installation record: %+v %v", captured, err)
880 }
881 // A later pass has nothing to route to.
882 if role := movesParticipantRecord(h.rollbackCfg(t), h.read(t, p1Host)); role != roleNone {
883 t.Fatalf("a cleaned target still routes as %s", role)
884 }
885 if !strings.HasSuffix(strings.Join(h.events, " "), "cleaning cleaned") {
886 t.Fatalf("events %v", h.events)
887 }
888}
889
890// Every release and return step is replay-safe: a crash after the record's
891// released leaves the claim at releasing, and the next pass writes it; a
892// crash after the return seal's claim write leaves the record at returning,
893// and the next pass finishes without writing the lane again.
894func TestTargetRollbackReplays(t *testing.T) {
895 t.Parallel()
896 t.Run("after-record-released", func(t *testing.T) {
897 h := newRollbackHarness(t)
898 h.rollBack(t)
899 h.rbAt = "after-record-released"
900 h.rollbackPass(t)
901 if !h.rbDone || h.claim(t).Op.Phase != hostclaim.PhaseReleasing {
902 t.Fatalf("the crash: %v %+v", h.rbHits, h.claim(t).Op)
903 }
904 h.rollbackPass(t)
905 if c := h.claim(t); c.Op.Phase != hostclaim.PhaseReleased || c.Op.Journal[hostclaim.JournalEvidenceTarget] != targetStoppedKey() {
906 t.Fatalf("claim %+v", c.Op)
907 }
908 })
909 t.Run("after-return-seal", func(t *testing.T) {
910 h := newRollbackHarness(t)
911 h.uncover(t)
912 h.rollBack(t)
913 h.rbAt = "after-return-seal"
914 h.rollbackPass(t)
915 rb := h.rb(t)
916 if !h.rbDone || rb.TargetPhase != provision.RbTargetReturning {
917 t.Fatalf("the crash: %v %+v", h.rbHits, rb)
918 }
919 calls := len(h.laneCalls(rb.ID))
920 h.rollbackPass(t)
921 if rb = h.rb(t); rb.TargetPhase != provision.RbTargetReturned || len(h.laneCalls(rb.ID)) != calls {
922 t.Fatalf("the replay: %+v, %d lane calls after %d", rb, len(h.laneCalls(rb.ID)), calls)
923 }
924 })
925}
926
927// The release waits while the target cannot prove its instance stopped,
928// and touches nothing the evidence would vouch for.
929func TestTargetReleaseWaitsForStopEvidence(t *testing.T) {
930 t.Parallel()
931 h := newRollbackHarness(t)
932 h.rollBack(t)
933 h.ev.stopped = false
934 h.rollbackPass(t)
935 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReleasing || h.tchecks != 0 {
936 t.Fatalf("rollback %+v checkpoints %d", rb, h.tchecks)
937 }
938 if c := h.claim(t); c.Op.Exec != "uk-box" || c.Op.Phase != hostclaim.PhaseReleasing {
939 t.Fatalf("claim %+v", c.Op)
940 }
941 if _, _, err := h.s.Get(context.Background(), targetStoppedKey()); err == nil {
942 t.Fatal("a stopped report without stop evidence")
943 }
944 h.ev.stopped = true
945 h.rollbackPass(t)
946 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReleased {
947 t.Fatalf("rollback %+v", rb)
948 }
949}
950
951// Production wires the hold hand-over and leaves the rollback unarmed with
952// the park engine's hooks (ruling J18), as the landing is.
953func TestTargetRollbackUnarmedInProduction(t *testing.T) {
954 t.Parallel()
955 h := newRollbackHarness(t)
956 c := h.rollbackCfg(t)
957 if !rollbackTargetArmed(c) {
958 t.Fatal("the test configuration is not armed")
959 }
960 c.ParkLeases = nil
961 if rollbackTargetArmed(c) {
962 t.Fatal("armed without park leases")
963 }
964 h.rollBack(t)
965 h.rw.cfg = c
966 if err := movesPass(context.Background(), h.s, h.rw); err != nil {
967 t.Fatal(err)
968 }
969 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReleasing || len(h.handed) != 0 {
970 t.Fatalf("an unarmed rollback moved: %+v %v", rb, h.handed)
971 }
972}
973
974// Fix round 1, ruling 1 (J3): the release re-reads its record under the hold
975// lock at the hand-over. A record that changed between the pass's
976// observation and the hand-over (here a declared loss took the target) keeps
977// the row held by the move, and nothing after the hand-over runs.
978func TestTargetReleaseRechecksTheRecordAtTheHandOver(t *testing.T) {
979 t.Parallel()
980 h := newRollbackHarness(t)
981 h.rollBack(t)
982 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
983 h.beforeHand = func() {
984 h.beforeHand = nil
985 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRollbackTargetLost(g, "control/losses/x.json", h.clock.Now())); err != nil {
986 t.Fatal(err)
987 }
988 }
989 calls := h.ev.calls
990 h.rollbackPass(t)
991 if h.route.Status != "landing" || h.route.HoldOp != p1ID || len(h.handed) != 1 {
992 t.Fatalf("the hold moved under a changed record: %+v %v", h.route, h.handed)
993 }
994 if h.ev.calls != calls || h.tchecks != 0 {
995 t.Fatalf("the release went on: evidence %d checkpoints %d", h.ev.calls-calls, h.tchecks)
996 }
997 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetLost {
998 t.Fatalf("rollback %+v", rb)
999 }
1000 if _, _, err := h.s.Get(context.Background(), targetStoppedKey()); err == nil {
1001 t.Fatal("a stopped report after a refused hand-over")
1002 }
1003}
1004
1005// targetChanged makes the target's state differ from what it installed: rows
1006// written by the app on the target, so the return snapshots it.
1007func (h *rollbackHarness) targetChanged(t *testing.T) {
1008 t.Helper()
1009 d, err := sql.Open("sqlite", provision.DefaultDBPath(h.tdir, p1Host))
1010 if err != nil {
1011 t.Fatal(err)
1012 }
1013 if _, err := d.Exec(`INSERT INTO t (body) VALUES ('written on the target')`); err != nil {
1014 t.Fatal(err)
1015 }
1016 d.Close()
1017 h.tpos = h.n + 1
1018}
1019
1020// stuckAtReturning asserts the record stopped for an operator at the
1021// target's returning with reason, the target keeping its phase, its row and
1022// what it holds, and no return seal.
1023func (h *rollbackHarness) stuckAtReturning(t *testing.T, reason string) {
1024 t.Helper()
1025 m := h.read(t, p1Host).Move
1026 if m.State != provision.MoveStuck || m.StuckReason != reason || m.Rollback.TargetPhase != provision.RbTargetReturning ||
1027 !m.Rollback.Dirty || m.Rollback.StuckAt != provision.RbSourceWaiting {
1028 t.Fatalf("not stuck at returning for %q: %+v %+v", reason, m, m.Rollback)
1029 }
1030 if h.route.Status != "releasing" || h.route.HoldOp != m.Rollback.ID || h.rowGone {
1031 t.Fatalf("the row: %+v gone %v", h.route, h.rowGone)
1032 }
1033 if _, err := os.Stat(filepath.Join(h.tdir, p1Host, "upload.txt")); err != nil {
1034 t.Fatalf("the target's files: %v", err)
1035 }
1036 if _, ok := h.claim(t).Op.Journal[moves.JournalReturnSeal]; ok {
1037 t.Fatal("a return seal on a stuck return")
1038 }
1039 if h.lease.acquired != h.lease.released {
1040 t.Fatalf("the staging lease: %d %d", h.lease.acquired, h.lease.released)
1041 }
1042}
1043
1044func (h *rollbackHarness) retry(t *testing.T) {
1045 t.Helper()
1046 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
1047 // The claim-backed retry (claim Plan 10): it advances the recovery retry
1048 // epoch, so the move log keys the next stuck as a new entry (Plan 6
1049 // Task 3, preflight L11).
1050 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRetryRecovery(g)); err != nil {
1051 t.Fatal(err)
1052 }
1053}
1054
1055// Fix round 1, ruling 2: a dirty return that cannot be built from what the
1056// target holds stops for an operator with a recorded reason, the target at
1057// returning with everything kept; a retry runs the return again.
1058func TestDirtyReturnGoesStuck(t *testing.T) {
1059 t.Parallel()
1060 t.Run("the database is missing", func(t *testing.T) {
1061 h := newRollbackHarness(t)
1062 db := provision.DefaultDBPath(h.tdir, p1Host)
1063 if err := os.Rename(db, db+".aside"); err != nil {
1064 t.Fatal(err)
1065 }
1066 h.rollBack(t)
1067 h.rollbackPass(t)
1068 h.stuckAtReturning(t, provision.StuckReturnTargetDatabase)
1069 // A stuck record is not the target's to work on.
1070 h.rollbackPass(t)
1071 h.stuckAtReturning(t, provision.StuckReturnTargetDatabase)
1072 // A retry runs the return again; still missing, it stops again.
1073 before := len(h.events)
1074 h.retry(t)
1075 h.rollbackPass(t)
1076 h.stuckAtReturning(t, provision.StuckReturnTargetDatabase)
1077 if got := strings.Join(h.events[before:], " "); got != "stuck" {
1078 t.Fatalf("the retry did not run the return: events %q", got)
1079 }
1080 })
1081 t.Run("the files cannot be archived", func(t *testing.T) {
1082 h := newRollbackHarness(t)
1083 h.targetChanged(t)
1084 link := filepath.Join(h.tdir, p1Host, "link")
1085 h.rbHook = func(point string) {
1086 if point == "after-record-returning" {
1087 if err := os.Symlink("/etc/passwd", link); err != nil {
1088 t.Fatal(err)
1089 }
1090 }
1091 }
1092 h.rollBack(t)
1093 h.rollbackPass(t)
1094 h.stuckAtReturning(t, provision.StuckReturnTargetFiles)
1095 // The operator removes the link and retries: the return completes.
1096 h.rbHook = nil
1097 if err := os.Remove(link); err != nil {
1098 t.Fatal(err)
1099 }
1100 h.retry(t)
1101 h.rollbackPass(t)
1102 if rb := h.rb(t); h.read(t, p1Host).Move.State != provision.MoveRollingBack || rb.TargetPhase != provision.RbTargetReturned {
1103 t.Fatalf("after the retry: %+v", rb)
1104 }
1105 })
1106 t.Run("a storage error is retried, not stuck", func(t *testing.T) {
1107 h := newRollbackHarness(t)
1108 h.targetChanged(t)
1109 fail := true
1110 h.f.failAt = func(op, key string) error {
1111 if fail && strings.Contains(key, "/replica/") {
1112 return errors.New("503 SlowDown")
1113 }
1114 return nil
1115 }
1116 h.rollBack(t)
1117 h.rollbackPass(t)
1118 if m := h.read(t, p1Host).Move; m.State != provision.MoveRollingBack || m.Rollback.TargetPhase != provision.RbTargetReturning {
1119 t.Fatalf("a storage error: %+v %+v", m, m.Rollback)
1120 }
1121 fail = false
1122 h.rollbackPass(t)
1123 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReturned {
1124 t.Fatalf("after the storage recovered: %+v", rb)
1125 }
1126 })
1127}
1128
1129// closeRolledBack is the source's close of the rollback, as Task 5 will
1130// write it: opened, then MoveRolledBackClose.
1131func (h *rollbackHarness) closeRolledBack(t *testing.T) {
1132 t.Helper()
1133 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
1134 now := h.clock.Now()
1135 for _, step := range []func(*provision.Record) error{
1136 provision.MoveRollbackSourcePhase(g, provision.RbSourceRecovered, provision.RbSourceOpened, func(rb *provision.MoveRollback) { rb.OpenedAt = &now }),
1137 provision.MoveRolledBackClose(g, now),
1138 } {
1139 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, step); err != nil {
1140 t.Fatal(err)
1141 }
1142 }
1143}
1144
1145// Fix round 1, ruling 3: the source closes the record right after the
1146// target's cleaned CAS, before the target's worker re-reads it, and the
1147// worker restarts. No record routes the target back, so the local journal
1148// sweep parks the op's journal (which holds the built files archive) on the
1149// close's evidence.
1150func TestTargetJournalSweptAfterTheClose(t *testing.T) {
1151 t.Parallel()
1152 h := newRollbackHarness(t)
1153 h.targetChanged(t)
1154 h.rollBack(t)
1155 h.rollbackPass(t)
1156 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReturned {
1157 t.Fatalf("rollback %+v", rb)
1158 }
1159 adir := archiveDir(h.tdir, p1Host, p1ID)
1160 built, _ := filepath.Glob(filepath.Join(adir, "files-*"))
1161 if len(built) == 0 {
1162 t.Fatal("the dirty return built no archive in the journal")
1163 }
1164 h.recovered(t)
1165 h.rbHook = func(point string) {
1166 if point == "after-record-cleaned" {
1167 h.closeRolledBack(t)
1168 }
1169 }
1170 h.rbAt = "after-record-cleaned"
1171 h.rollbackPass(t)
1172 if !h.rbDone || h.read(t, p1Host).Move != nil {
1173 t.Fatalf("the fixture: crashed %v, record %+v", h.rbDone, h.read(t, p1Host).Move)
1174 }
1175 jdir := moves.JournalDir(h.tdir, p1Host, p1ID)
1176 if _, err := os.Stat(jdir); err != nil {
1177 t.Fatalf("the fixture left no journal to sweep: %v", err)
1178 }
1179 // A restart: a fresh worker, and the next pass's sweep.
1180 h.rw, h.rbHook = nil, nil
1181 h.leases.expire()
1182 h.rollbackPass(t)
1183 if _, err := os.Lstat(jdir); !errors.Is(err, os.ErrNotExist) {
1184 t.Fatalf("the op journal is still live: %v", err)
1185 }
1186 var parkedArchive int
1187 filepath.WalkDir(filepath.Join(h.tdir, ".parked"), func(p string, d os.DirEntry, err error) error {
1188 if err == nil && !d.IsDir() && strings.Contains(p, filepath.Join(".moves", p1Host, p1ID, "archive", "files-")) {
1189 parkedArchive++
1190 }
1191 return nil
1192 })
1193 if parkedArchive == 0 {
1194 t.Fatal("the built archive was not parked with the journal")
1195 }
1196}
1197
1198// The journal sweep leaves a rollback journal alone while its record still
1199// carries the op without target_parked_at: the cleaning is the pass's.
1200func TestJournalSweepWaitsForTheTargetPark(t *testing.T) {
1201 t.Parallel()
1202 h := newRollbackHarness(t)
1203 h.rollBack(t)
1204 h.rollbackPass(t)
1205 h.recovered(t)
1206 // Crash right after the moves-owned rename, before the row and the CAS.
1207 h.rbAt = "after-" + rbStepLocal
1208 h.rollbackPass(t)
1209 if !h.rbDone {
1210 t.Fatalf("never reached %s: %v", h.rbAt, h.rbHits)
1211 }
1212 w := &movesWorker{cfg: h.rollbackCfg(t), mirrors: map[string]lease.MirrorStore{}}
1213 w.sweepFinalJournals(context.Background(), h.s, []provision.Record{h.read(t, p1Host)})
1214 if _, err := os.Stat(moves.JournalDir(h.tdir, p1Host, p1ID)); err != nil {
1215 t.Fatalf("the sweep parked a journal whose cleaning is under way: %v", err)
1216 }
1217}
1218
1219// Fix round 1, ruling 4: a crash at each durable step of the cleaning, then a
1220// fresh worker, finishes it with nothing live left.
1221func TestTargetCleaningReplays(t *testing.T) {
1222 slots := make(chan struct{}, 4)
1223 for _, point := range []string{
1224 "before-rb-local-retire", "after-rb-local-retire", "after-" + rbStepInstallation,
1225 "after-" + rbStepLocal + "-plan", "after-rename:" + p1Host + ".db-wal", "after-" + rbStepLocal,
1226 "after-rb-row-pending", "after-rb-row-delete", "after-" + rbStepRowDeleted,
1227 } {
1228 t.Run(point, func(t *testing.T) {
1229 t.Parallel()
1230 slots <- struct{}{}
1231 defer func() { <-slots }()
1232 h := newRollbackHarness(t)
1233 if err := os.WriteFile(provision.DefaultDBPath(h.tdir, p1Host)+"-wal", []byte("a wal"), 0o600); err != nil {
1234 t.Fatal(err)
1235 }
1236 h.rollBack(t)
1237 h.rollbackPass(t)
1238 h.recovered(t)
1239 h.rbAt = point
1240 h.rollbackPass(t)
1241 if !h.rbDone {
1242 t.Fatalf("never reached %s: %v", point, h.rbHits)
1243 }
1244 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaning {
1245 t.Fatalf("after the crash: %+v", rb)
1246 }
1247 h.rw = nil
1248 h.leases.expire()
1249 h.rollbackPass(t)
1250 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaned || rb.TargetParkedAt == nil {
1251 t.Fatalf("after the replay: %+v", rb)
1252 }
1253 for _, p := range sourceArtifactPaths(p1Host) {
1254 if _, err := os.Lstat(filepath.Join(h.tdir, p)); !errors.Is(err, os.ErrNotExist) {
1255 t.Fatalf("%s is still live: %v", p, err)
1256 }
1257 }
1258 if _, err := os.Lstat(moves.JournalDir(h.tdir, p1Host, p1ID)); !errors.Is(err, os.ErrNotExist) {
1259 t.Fatalf("the op journal is still live: %v", err)
1260 }
1261 if !h.rowGone {
1262 t.Fatal("the row is still there")
1263 }
1264 if c := h.claim(t); len(c.Inventory["uk-box"].Paths) != 0 {
1265 t.Fatalf("inventory %+v", c.Inventory["uk-box"])
1266 }
1267 })
1268 }
1269}
1270