amadan
host: the hand-over's record check, a stuck dirty return, the target journal sweep (Task 4 fix round 1)
handHold takes an optional validation that runs under the hold lock after the row is re-read; the release re-reads its record there (J3). A dirty return that cannot be built from what the target holds (data gone, state changed, files not archivable) marks the record stuck and keeps everything; storage errors still retry. The local journal sweep parks a target's rollback journal once the journal says cleaned, the record shows target_parked_at, or no record carries the op (MoveRolledBackClose requires target_parked_at); cleaning parks it in the same call. Fault points around the local retirement and the row delete, with fresh-worker replays.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
6 files changed,
+557
−27
internal/host/movesfinal.go+5 −2internal/host/moveshold.go+13 −2internal/host/moveshold_test.go+56 −4internal/host/movespass.go+1 −1internal/host/movesrollback.go+178 −17internal/host/movesrollback_test.go+304 −1
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go| index 99c979804..10706c861 100644 |
| --- a/internal/host/movesfinal.go |
| +++ b/internal/host/movesfinal.go |
| @@ -630,7 +630,8 @@ func parkFinalJournal(dataDir, host, op string, j *moves.Journal, mark func(step |
| } |
| // sweepFinalJournals is the local sweep of this box's op journals (fix round |
| -// 1, ruling 1), independent of any active move: a crash after the stamp and |
| +// 1, ruling 1; a rollback target's journals too, sweepRollbackJournal), |
| +// independent of any active move: a crash after the stamp and |
| // before the journal's park, then the close, leaves a journal no record |
| // routes back here (movesParticipantRecord answers nothing for a record with |
| // no move). Each journal under <DataDir>/.moves/<host>/<op>/ that has reached |
| @@ -696,7 +697,9 @@ func (w *movesWorker) sweepFinalJournal(dataDir, host, op string, recs []provisi |
| } |
| stamp, reached := j.Steps[finalStepParked] |
| if !reached { |
| - return nil |
| + // Not a finalization's: a rollback's target journal, perhaps |
| + // (movesrollback.go, Task 4 fix round 1). |
| + return w.sweepRollbackJournal(dataDir, host, op, j, recs) |
| } |
| if stamp != "done" { |
| if len(recs) == 0 { |
diff --git a/internal/host/moveshold.go b/internal/host/moveshold.go| index 3e791cf20..3b9680953 100644 |
| --- a/internal/host/moveshold.go |
| +++ b/internal/host/moveshold.go |
| @@ -292,11 +292,17 @@ func holdStageRequester(dd string) func(host string) { |
| // nothing. Every caller hands a hold over through this wrapper, never through |
| // registry.HandHold directly, which cannot reach the marker. |
| // |
| +// validate, when not nil, is the caller's own fresh check (Task 4 fix round |
| +// 1, ruling 1; J3's record check): it runs under the hold lock, after the |
| +// row is re-read and before anything is decided, and an error from it moves |
| +// nothing and is returned. It runs inside WithHostHoldLock, so it must not |
| +// take that lock again; a provision record read is what it is for. |
| +// |
| // true means the row is held status under toOp: the hand-over landed, or a |
| // replay found it already done. false with no error means nothing matched: |
| // the row is held by some other op, unheld, or absent. |
| -func handHold(reg *registry.Registry, written func(host string)) func(host, status, toOp, fromOp string) (bool, error) { |
| - return func(host, status, toOp, fromOp string) (bool, error) { |
| +func handHold(reg *registry.Registry, written func(host string)) func(host, status, toOp, fromOp string, validate func() error) (bool, error) { |
| + return func(host, status, toOp, fromOp string, validate func() error) (bool, error) { |
| var handed bool |
| var err error |
| provision.WithHostHoldLock(host, func() { |
| @@ -304,6 +310,11 @@ func handHold(reg *registry.Registry, written func(host string)) func(host, stat |
| if rt, err = reg.Get(host); err != nil || rt == nil { |
| return |
| } |
| + if validate != nil { |
| + if err = validate(); err != nil { |
| + return |
| + } |
| + } |
| if rt.HoldOp == toOp && rt.Status == status && toOp != "" { |
| handed = true |
| return |
diff --git a/internal/host/moveshold_test.go b/internal/host/moveshold_test.go| index 3ac4b21eb..0a1aaa213 100644 |
| --- a/internal/host/moveshold_test.go |
| +++ b/internal/host/moveshold_test.go |
| @@ -6,6 +6,7 @@ import ( |
| "net/http" |
| "os" |
| "path/filepath" |
| + "strings" |
| "testing" |
| "time" |
| @@ -342,20 +343,20 @@ func TestHandHoldTouchesTheStageMarkerOnlyWhenItLands(t *testing.T) { |
| // A refusal: the row is held by p1ID, not p1ID2. Nothing moves, no marker. |
| refused := t.TempDir() |
| - if ok, err := handHold(reg, holdStageRequester(refused))(p1Host, "releasing", rb, p1ID2); err != nil || ok || marker(refused) { |
| + if ok, err := handHold(reg, holdStageRequester(refused))(p1Host, "releasing", rb, p1ID2, nil); err != nil || ok || marker(refused) { |
| t.Fatalf("refused hand-over: %v %v, marker %v", ok, err, marker(refused)) |
| } |
| if rt, _ := reg.Get(p1Host); rt.Status != "landing" || rt.HoldOp != p1ID { |
| t.Fatalf("row changed by a refusal: %+v", rt) |
| } |
| // An absent row: no marker. |
| - if ok, err := handHold(reg, holdStageRequester(refused))("absent.example.com", "releasing", rb, p1ID); err != nil || ok || marker(refused) { |
| + if ok, err := handHold(reg, holdStageRequester(refused))("absent.example.com", "releasing", rb, p1ID, nil); err != nil || ok || marker(refused) { |
| t.Fatalf("absent row: %v %v, marker %v", ok, err, marker(refused)) |
| } |
| // The hand-over: the row is the rollback's, and the marker is written. |
| landed := t.TempDir() |
| - if ok, err := handHold(reg, holdStageRequester(landed))(p1Host, "releasing", rb, p1ID); err != nil || !ok || !marker(landed) { |
| + if ok, err := handHold(reg, holdStageRequester(landed))(p1Host, "releasing", rb, p1ID, nil); err != nil || !ok || !marker(landed) { |
| t.Fatalf("hand-over: %v %v, marker %v", ok, err, marker(landed)) |
| } |
| if rt, _ := reg.Get(p1Host); rt.Status != "releasing" || rt.HoldOp != rb { |
| @@ -364,7 +365,58 @@ func TestHandHoldTouchesTheStageMarkerOnlyWhenItLands(t *testing.T) { |
| // A replay finds the row already the rollback's: done, nothing written. |
| var touched []string |
| - if ok, err := handHold(reg, func(h string) { touched = append(touched, h) })(p1Host, "releasing", rb, p1ID); err != nil || !ok || len(touched) != 0 { |
| + if ok, err := handHold(reg, func(h string) { touched = append(touched, h) })(p1Host, "releasing", rb, p1ID, nil); err != nil || !ok || len(touched) != 0 { |
| t.Fatalf("replay: %v %v, marker touched %d times", ok, err, len(touched)) |
| } |
| } |
| + |
| +// Task 4 fix round 1, ruling 1: the hand-over's caller validates its own |
| +// record under the hold lock, after the row is re-read and before the hold |
| +// moves. A validation that refuses moves nothing and writes no marker; one |
| +// that passes is asked once, and the hold moves. The lock is the wrapper's |
| +// alone: the callback runs inside it and must not take it again. |
| +func TestHandHoldValidatesUnderTheLock(t *testing.T) { |
| + const rb = "rb_abcdefghijklmnopqrstuvwxyz" |
| + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer reg.Close() |
| + if err := reg.Upsert(registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Status: "ready", Provisioned: true}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.SetHold(p1Host, "landing", p1ID); err != nil { |
| + t.Fatal(err) |
| + } |
| + changed := errors.New("the record is no longer this rollback's release") |
| + var seen []string |
| + refused := t.TempDir() |
| + ok, err := handHold(reg, holdStageRequester(refused))(p1Host, "releasing", rb, p1ID, func() error { |
| + rt, _ := reg.Get(p1Host) |
| + seen = append(seen, rt.Status+"/"+rt.HoldOp) |
| + return changed |
| + }) |
| + if !errors.Is(err, changed) || ok { |
| + t.Fatalf("a refused validation: %v %v", ok, err) |
| + } |
| + if _, err := os.Stat(filepath.Join(refused, stageRequestMarker)); err == nil { |
| + t.Fatal("a refused validation wrote the marker") |
| + } |
| + if rt, _ := reg.Get(p1Host); rt.Status != "landing" || rt.HoldOp != p1ID { |
| + t.Fatalf("the hold moved under a refused validation: %+v", rt) |
| + } |
| + landed := t.TempDir() |
| + if ok, err := handHold(reg, holdStageRequester(landed))(p1Host, "releasing", rb, p1ID, func() error { |
| + rt, _ := reg.Get(p1Host) |
| + seen = append(seen, rt.Status+"/"+rt.HoldOp) |
| + return nil |
| + }); err != nil || !ok { |
| + t.Fatalf("a passing validation: %v %v", ok, err) |
| + } |
| + if rt, _ := reg.Get(p1Host); rt.Status != "releasing" || rt.HoldOp != rb { |
| + t.Fatalf("after the hand-over: %+v", rt) |
| + } |
| + if strings.Join(seen, " ") != "landing/"+p1ID+" landing/"+p1ID { |
| + t.Fatalf("validations saw %v, want the row before the hand-over, once each", seen) |
| + } |
| +} |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index 30b8b5c2e..c7888a25b 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -571,7 +571,7 @@ type moveCfg struct { |
| // half also uses the cut's stop hooks, the landing's park hooks and |
| // DeleteHeld above; ParkLeases nil leaves it unarmed in production |
| // (rollbackTargetArmed). rbFault is its test seam, nil in production. |
| - HandHold func(host, status, toOp, fromOp string) (bool, error) |
| + HandHold func(host, status, toOp, fromOp string, validate func() error) (bool, error) |
| LogEvent func(ctx context.Context, rec provision.Record, event string) |
| rbFault func(point string) error |
| } |
diff --git a/internal/host/movesrollback.go b/internal/host/movesrollback.go| index a374a4b2f..d2c1e447f 100644 |
| --- a/internal/host/movesrollback.go |
| +++ b/internal/host/movesrollback.go |
| @@ -106,6 +106,7 @@ const ( |
| rbStepInstallation = "rb-target-park-installation" // retired, completed or unregistered |
| rbStepLocal = "rb-target-park-local" // parkLocalArtifacts' step: the retirement id |
| rbStepRowDeleted = "rb-target-row-deleted" // "pending" before the delete, "done" after it |
| + rbStepCleaned = "rb-target-cleaned" // "done": the record says cleaned, only the journal's park is left |
| ) |
| // stoppedTargetRole is the target's StoppedReport role. |
| @@ -357,6 +358,23 @@ func (r *rollbackRun) stillReleasing(ctx context.Context) error { |
| return nil |
| } |
| +// stillReleasingRecord is the hand-over's record check (J3, fix round 1, |
| +// ruling 1), run by handHold under the host's hold lock: the record, read |
| +// fresh, must still be this move and incarnation, rolling back under this |
| +// rollback, with the target releasing. It reads the record only: the lock is |
| +// the wrapper's, and nothing here takes it again. |
| +func (r *rollbackRun) stillReleasingRecord(ctx context.Context) error { |
| + fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host) |
| + if err != nil { |
| + return fmt.Errorf("re-read the record: %w", err) |
| + } |
| + if m := fresh.Move; m == nil || m.ID != r.g.ID || m.Incarnation != r.g.Incarnation || m.State != provision.MoveRollingBack || |
| + m.Rollback == nil || m.Rollback.ID != r.rb || m.Rollback.TargetPhase != provision.RbTargetReleasing { |
| + return fmt.Errorf("move %s: the record is no longer rollback %s releasing its target", r.g.ID, r.rb) |
| + } |
| + return nil |
| +} |
| + |
| // reachedProbing reports whether the landing reached probing: a writer may |
| // have run on this box. |
| func (r *rollbackRun) reachedProbing() bool { |
| @@ -400,7 +418,7 @@ func (r *rollbackRun) releasing(ctx context.Context) (bool, error) { |
| if err != nil { |
| return false, err |
| } |
| - handed, err := cfg.HandHold(host, "releasing", r.rb, id) |
| + handed, err := cfg.HandHold(host, "releasing", r.rb, id, func() error { return r.stillReleasingRecord(ctx) }) |
| if err != nil { |
| return false, fmt.Errorf("hand the row's hold to the rollback: %w", err) |
| } |
| @@ -630,6 +648,10 @@ func (r *rollbackRun) returning(ctx context.Context) (bool, error) { |
| } else { |
| lane, err = r.snapshotStopped(ctx, gm, report) |
| } |
| + var refused *returnRefusal |
| + if errors.As(err, &refused) { |
| + return r.stuck(ctx, refused) |
| + } |
| if err != nil { |
| return false, err |
| } |
| @@ -792,40 +814,106 @@ func downloadExact(ctx context.Context, vs lease.VersionedStore, bucket, key, ve |
| return nil |
| } |
| +// returnRefusal is a dirty return that cannot be built from what the target |
| +// holds (fix round 1, ruling 2): its database data is missing, it changed |
| +// after it stopped, or its files cannot be archived. No retry by itself can |
| +// help, so the record stops for an operator with reason. Storage, network |
| +// and local read errors are not refusals: they stay ordinary errors, and the |
| +// next pass retries. |
| +type returnRefusal struct { |
| + reason string |
| + err error |
| +} |
| + |
| +func (e *returnRefusal) Error() string { return e.err.Error() } |
| +func (e *returnRefusal) Unwrap() error { return e.err } |
| + |
| +func refuseReturn(reason string, err error) error { return &returnRefusal{reason: reason, err: err} } |
| + |
| +// databaseGone reports whether err says the database's data is not there to |
| +// read (absent, short, or not a database), as opposed to a read that failed: |
| +// an *os.PathError other than "does not exist" is the disk's, and transient. |
| +func databaseGone(err error) bool { |
| + if errors.Is(err, os.ErrNotExist) { |
| + return true |
| + } |
| + var pe *os.PathError |
| + return !errors.As(err, &pe) |
| +} |
| + |
| +// stuck stops the record for an operator (MoveMarkStuck) at the target's |
| +// returning: the phase, the row and everything the target holds are kept, |
| +// and move --retry runs the return again. |
| +func (r *rollbackRun) stuck(ctx context.Context, refused *returnRefusal) (bool, error) { |
| + host := r.rec.Host |
| + log.Printf("moves: %s: the return of %s cannot be built (%v); it stops for an operator: %s", host, r.rb, refused.err, refused.reason) |
| + if _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveMarkStuck(r.g, refused.reason)); err != nil { |
| + if errors.Is(err, provision.ErrMoveConflict) { |
| + log.Printf("moves: %s: the stuck mark of %s refused (%v); the record moved on", host, r.rb, err) |
| + return false, nil |
| + } |
| + return false, fmt.Errorf("mark the return stuck: %w", err) |
| + } |
| + r.w.logEvent(ctx, r.rec, "target-stuck") |
| + nudgeSerial(ctx, r.s) |
| + return true, nil |
| +} |
| + |
| // snapshotStopped is the return of a target that served and changed: one |
| // snapshot [1, N'] of the stopped database (moves.EncodeFull, in memory, the |
| // Phase 1 size limit), the target's files archive (manifest bound to the move |
| -// id) and its timers. Each must still be what the stopped report recorded. |
| +// id) and its timers. Each must still be what the stopped report recorded: |
| +// data that is gone (StuckReturnTargetDatabase), a state that differs from |
| +// the report (StuckReturnTargetChanged) and files that cannot be archived |
| +// (StuckReturnTargetFiles) are refusals; anything else is retried. |
| func (r *rollbackRun) snapshotStopped(ctx context.Context, gm *guardedMirror, report moves.StoppedReport) (returnedLane, error) { |
| cfg := r.cfg() |
| host, id := r.rec.Host, r.g.ID |
| dbPath := r.dbPath() |
| sha, err := fileSHA256(dbPath) |
| if err != nil { |
| - return returnedLane{}, fmt.Errorf("the target's stopped database: %w", err) |
| + err = fmt.Errorf("the target's stopped database: %w", err) |
| + if errors.Is(err, os.ErrNotExist) { |
| + err = refuseReturn(provision.StuckReturnTargetDatabase, err) |
| + } |
| + return returnedLane{}, err |
| } |
| if sha != report.DBSHA256 { |
| - return returnedLane{}, fmt.Errorf("move %s: the target's database changed since its release", id) |
| + return returnedLane{}, refuseReturn(provision.StuckReturnTargetChanged, fmt.Errorf("move %s: the target's database changed since its release", id)) |
| } |
| n, err := cfg.LocalPosition(dbPath) |
| if err != nil { |
| return returnedLane{}, fmt.Errorf("read the stopped database's position: %w", err) |
| } |
| - if n != report.FinalTXID || n == 0 { |
| - return returnedLane{}, fmt.Errorf("move %s: the stopped database is at TXID %d, the report at %d", id, n, report.FinalTXID) |
| + if n == 0 { |
| + return returnedLane{}, refuseReturn(provision.StuckReturnTargetDatabase, fmt.Errorf("move %s: the stopped database has no position", id)) |
| + } |
| + if n != report.FinalTXID { |
| + return returnedLane{}, refuseReturn(provision.StuckReturnTargetChanged, fmt.Errorf("move %s: the stopped database is at TXID %d, the report at %d", id, n, report.FinalTXID)) |
| } |
| pageSize, pageCount, err := moves.ReadDBHeader(dbPath) |
| if err != nil { |
| + if databaseGone(err) { |
| + err = refuseReturn(provision.StuckReturnTargetDatabase, err) |
| + } |
| return returnedLane{}, err |
| } |
| f, err := os.Open(dbPath) |
| if err != nil { |
| + if errors.Is(err, os.ErrNotExist) { |
| + err = refuseReturn(provision.StuckReturnTargetDatabase, err) |
| + } |
| return returnedLane{}, err |
| } |
| data, _, err := moves.EncodeFull(f, pageSize, pageCount, 1, ltx.TXID(n), report.CheckedAt.UnixMilli(), 0) |
| f.Close() |
| if err != nil { |
| - return returnedLane{}, fmt.Errorf("snapshot the stopped database: %w", err) |
| + err = fmt.Errorf("snapshot the stopped database: %w", err) |
| + if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { |
| + // The header names pages the file does not hold. |
| + err = refuseReturn(provision.StuckReturnTargetDatabase, err) |
| + } |
| + return returnedLane{}, err |
| } |
| var out returnedLane |
| key := moves.SnapshotKey(moves.ReplicaPrefix(host, r.rb), n) |
| @@ -842,11 +930,11 @@ func (r *rollbackRun) snapshotStopped(ctx context.Context, gm *guardedMirror, re |
| case err != nil: |
| return out, fmt.Errorf("archive the target's files: %w", err) |
| case why != "": |
| - return out, fmt.Errorf("move %s: the target's files cannot be archived (%s)", id, why) |
| + return out, refuseReturn(provision.StuckReturnTargetFiles, fmt.Errorf("move %s: the target's files cannot be archived (%s)", id, why)) |
| case retry: |
| return out, errors.New("a stopped file changed while it was archived") |
| case a.Plain != report.FilesPlain: |
| - return out, fmt.Errorf("move %s: the target's files are %s, the report %s", id, a.Plain, report.FilesPlain) |
| + return out, refuseReturn(provision.StuckReturnTargetChanged, fmt.Errorf("move %s: the target's files are %s, the report %s", id, a.Plain, report.FilesPlain)) |
| } |
| ref, err := a.stage(ctx, gm, gm.dst, host, r.rb) |
| if err != nil { |
| @@ -866,7 +954,7 @@ func (r *rollbackRun) snapshotStopped(ctx context.Context, gm *guardedMirror, re |
| return out, fmt.Errorf("read the timers: %w", err) |
| } |
| if moves.TimersDigest(rows) != report.TimersSHA256 { |
| - return out, fmt.Errorf("move %s: the target's timers changed since its release", id) |
| + return out, refuseReturn(provision.StuckReturnTargetChanged, fmt.Errorf("move %s: the target's timers changed since its release", id)) |
| } |
| body, err := moves.TimersSnapshot{V: 1, Host: host, Rows: rows, TakenAt: report.CheckedAt}.Encode() |
| if err != nil { |
| @@ -959,10 +1047,20 @@ func (r *rollbackRun) cleaning(ctx context.Context) (bool, error) { |
| if err := r.deleteRow(ctx, j, mark); err != nil { |
| return false, err |
| } |
| - return r.phase(ctx, provision.RbTargetCleaning, provision.RbTargetCleaned, func(rb *provision.MoveRollback) { |
| + ok, err := r.phase(ctx, provision.RbTargetCleaning, provision.RbTargetCleaned, func(rb *provision.MoveRollback) { |
| t := r.now().UTC() |
| rb.TargetParkedAt = &t |
| }, "target-cleaned") |
| + if !ok || err != nil { |
| + return ok, err |
| + } |
| + // The journal is parked in this same call: once target_parked_at is |
| + // stamped the source may close the record before this worker reads it |
| + // again, and then only the local sweep finds the journal. |
| + if err := mark(rbStepCleaned, "done"); err != nil { |
| + return true, err |
| + } |
| + return true, parkRollbackJournal(cfg.DataDir, host, id, j) |
| } |
| // remote parks the target region's location and the move's lane in it: the |
| @@ -1132,9 +1230,15 @@ func (r *rollbackRun) installation(ctx context.Context, ours bool) (string, erro |
| return r.authorizeChore(c, o) |
| } |
| } |
| + if err := r.fault("before-rb-local-retire"); err != nil { |
| + return "", err |
| + } |
| if err := local.RetireInstallation(ctx, host, hostguard.Captured{Incarnation: captured.Incarnation, Stamp: f}, chore.ID, id); err != nil { |
| return "", fmt.Errorf("retire the target's installation: %w", err) |
| } |
| + if err := r.fault("after-rb-local-retire"); err != nil { |
| + return "", err |
| + } |
| log.Printf("moves: %s: the target's registered installation (stamp %d) is retired", host, f) |
| return "retired", nil |
| case chore != nil: |
| @@ -1232,7 +1336,11 @@ func (r *rollbackRun) deleteRow(ctx context.Context, j *moves.Journal, mark func |
| return |
| } |
| if j.Steps[rbStepRowDeleted] != "pending" { |
| - if err = mark(rbStepRowDeleted, "pending"); err != nil { |
| + j.Mark(rbStepRowDeleted, "pending") |
| + if err = j.Write(cfg.DataDir); err != nil { |
| + return |
| + } |
| + if err = r.fault("after-rb-row-pending"); err != nil { |
| return |
| } |
| } |
| @@ -1244,13 +1352,16 @@ func (r *rollbackRun) deleteRow(ctx context.Context, j *moves.Journal, mark func |
| if deleted { |
| log.Printf("moves: %s: the target's row and its timers are deleted", host) |
| } |
| + if err = r.fault("after-rb-row-delete"); err != nil { |
| + return |
| + } |
| err = mark(rbStepRowDeleted, "done") |
| }) |
| return err |
| } |
| -// cleaned parks the op's journal into the local retirement, after the record |
| -// says cleaned: until then it is what routes a replay back here. |
| +// cleaned parks the op's journal into the local retirement once the record |
| +// says cleaned, if a crash or a quick close left it. |
| func (r *rollbackRun) cleaned() error { |
| cfg := r.cfg() |
| host, id := r.rec.Host, r.g.ID |
| @@ -1261,17 +1372,67 @@ func (r *rollbackRun) cleaned() error { |
| if err != nil { |
| return err |
| } |
| + if j.Steps[rbStepCleaned] != "done" { |
| + j.Mark(rbStepCleaned, "done") |
| + if err := j.Write(cfg.DataDir); err != nil { |
| + return err |
| + } |
| + } |
| + return parkRollbackJournal(cfg.DataDir, host, id, j) |
| +} |
| + |
| +// parkRollbackJournal parks the target's op journal (its steps, the landing's |
| +// build and extract scratch, and the built files archive of a dirty return) |
| +// into the retirement its own moves-owned rename recorded. |
| +func parkRollbackJournal(dataDir, host, op string, j *moves.Journal) error { |
| retirement := j.Steps[rbStepLocal] |
| if retirement == "" { |
| - return fmt.Errorf("move %s: the journal records no local retirement for the target's park", id) |
| + return fmt.Errorf("move %s: the journal records no local retirement for the target's park", op) |
| } |
| plan, err := journalledLocalPlan(j, rbStepLocal) |
| if err != nil { |
| return err |
| } |
| - if err := parkOpJournal(cfg.DataDir, plan.Incarnation, host, id, retirement); err != nil { |
| + if err := parkOpJournal(dataDir, plan.Incarnation, host, op, retirement); err != nil { |
| return fmt.Errorf("park the op journal: %w", err) |
| } |
| - log.Printf("moves: %s: the target's half of rollback %s is complete", host, r.rb) |
| + log.Printf("moves: %s: the target's half of the rollback of %s is complete", host, op) |
| return nil |
| } |
| + |
| +// sweepRollbackJournal is the local journal sweep's rollback half |
| +// (sweepFinalJournals, fix round 1, ruling 3): a target's journal whose |
| +// cleaning reached its moves-owned rename (rbStepLocal) is parked once its |
| +// completion is shown, independent of any record that still routes here: |
| +// - the journal records cleaned; |
| +// - a listed record carries the op and its rollback shows target_parked_at |
| +// (the cleaned CAS landed); |
| +// - no listed record carries the op: the only close of a rolling_back move, |
| +// provision.MoveRolledBackClose, refuses unless target_parked_at is |
| +// stamped, and a journal at rbStepLocal is past every declared-loss |
| +// window, so the cleaned CAS landed before the close. |
| +// |
| +// A listed record that carries the op without target_parked_at is the |
| +// cleaning's own pass, and a host the listing does not name is left. A journal |
| +// that never reached rbStepLocal is never touched. |
| +func (w *movesWorker) sweepRollbackJournal(dataDir, host, op string, j *moves.Journal, recs []provision.Record) error { |
| + if _, ok := j.Steps[rbStepLocal]; !ok { |
| + return nil |
| + } |
| + if j.Steps[rbStepCleaned] != "done" { |
| + if len(recs) == 0 { |
| + w.notice(host+"#rb-sweep-"+op, "moves: %s: the target's journal of %s reached its park, and no listed record names the host; it is left", host, op) |
| + return nil |
| + } |
| + for _, rec := range recs { |
| + if m := rec.Move; m != nil && m.ID == op && (m.Rollback == nil || m.Rollback.TargetParkedAt == nil) { |
| + return nil |
| + } |
| + } |
| + j.Mark(rbStepCleaned, "done") |
| + if err := j.Write(dataDir); err != nil { |
| + return err |
| + } |
| + } |
| + return parkRollbackJournal(dataDir, host, op, j) |
| +} |
diff --git a/internal/host/movesrollback_test.go b/internal/host/movesrollback_test.go| index 3eecb7359..022a5461f 100644 |
| --- a/internal/host/movesrollback_test.go |
| +++ b/internal/host/movesrollback_test.go |
| @@ -45,6 +45,10 @@ type rollbackHarness struct { |
| rbAt string |
| rbDone bool |
| rbHits []string |
| + // rbHook runs at every fault point, before the crash check; beforeHand |
| + // runs inside the hand-over, before its validation. |
| + rbHook func(point string) |
| + beforeHand func() |
| } |
| func newRollbackHarness(t *testing.T) *rollbackHarness { |
| @@ -87,11 +91,21 @@ func (h *rollbackHarness) rollbackCfg(t *testing.T) moveCfg { |
| rt := h.route |
| return &rt, nil |
| } |
| - c.HandHold = func(host, status, to, from string) (bool, error) { |
| + // The hand-over keeps handHold's contract: under the lock the row is |
| + // read, then the caller's validation runs, then the hold moves. |
| + c.HandHold = func(host, status, to, from string, validate func() error) (bool, error) { |
| h.handed = append(h.handed, status+":"+to+":"+from) |
| if h.rowGone { |
| return false, nil |
| } |
| + if h.beforeHand != nil { |
| + h.beforeHand() |
| + } |
| + if validate != nil { |
| + if err := validate(); err != nil { |
| + return false, err |
| + } |
| + } |
| if h.route.HoldOp == to && h.route.Status == status { |
| return true, nil |
| } |
| @@ -123,6 +137,9 @@ func (h *rollbackHarness) rollbackCfg(t *testing.T) moveCfg { |
| c.LogEvent = func(_ context.Context, _ provision.Record, event string) { h.events = append(h.events, event) } |
| c.rbFault = func(point string) error { |
| h.rbHits = append(h.rbHits, point) |
| + if h.rbHook != nil { |
| + h.rbHook(point) |
| + } |
| if point == h.rbAt && !h.rbDone { |
| h.rbDone = true |
| return errCrash |
| @@ -935,3 +952,289 @@ func TestTargetRollbackUnarmedInProduction(t *testing.T) { |
| t.Fatalf("an unarmed rollback moved: %+v %v", rb, h.handed) |
| } |
| } |
| + |
| +// Fix round 1, ruling 1 (J3): the release re-reads its record under the hold |
| +// lock at the hand-over. A record that changed between the pass's |
| +// observation and the hand-over (here a declared loss took the target) keeps |
| +// the row held by the move, and nothing after the hand-over runs. |
| +func TestTargetReleaseRechecksTheRecordAtTheHandOver(t *testing.T) { |
| + h := newRollbackHarness(t) |
| + h.rollBack(t) |
| + g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc} |
| + h.beforeHand = func() { |
| + h.beforeHand = nil |
| + if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRollbackTargetLost(g, "control/losses/x.json", h.clock.Now())); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + calls := h.ev.calls |
| + h.rollbackPass(t) |
| + if h.route.Status != "landing" || h.route.HoldOp != p1ID || len(h.handed) != 1 { |
| + t.Fatalf("the hold moved under a changed record: %+v %v", h.route, h.handed) |
| + } |
| + if h.ev.calls != calls || h.tchecks != 0 { |
| + t.Fatalf("the release went on: evidence %d checkpoints %d", h.ev.calls-calls, h.tchecks) |
| + } |
| + if rb := h.rb(t); rb.TargetPhase != provision.RbTargetLost { |
| + t.Fatalf("rollback %+v", rb) |
| + } |
| + if _, _, err := h.s.Get(context.Background(), targetStoppedKey()); err == nil { |
| + t.Fatal("a stopped report after a refused hand-over") |
| + } |
| +} |
| + |
| +// targetChanged makes the target's state differ from what it installed: rows |
| +// written by the app on the target, so the return snapshots it. |
| +func (h *rollbackHarness) targetChanged(t *testing.T) { |
| + t.Helper() |
| + d, err := sql.Open("sqlite", provision.DefaultDBPath(h.tdir, p1Host)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := d.Exec(`INSERT INTO t (body) VALUES ('written on the target')`); err != nil { |
| + t.Fatal(err) |
| + } |
| + d.Close() |
| + h.tpos = h.n + 1 |
| +} |
| + |
| +// stuckAtReturning asserts the record stopped for an operator at the |
| +// target's returning with reason, the target keeping its phase, its row and |
| +// what it holds, and no return seal. |
| +func (h *rollbackHarness) stuckAtReturning(t *testing.T, reason string) { |
| + t.Helper() |
| + m := h.read(t, p1Host).Move |
| + if m.State != provision.MoveStuck || m.StuckReason != reason || m.Rollback.TargetPhase != provision.RbTargetReturning || |
| + !m.Rollback.Dirty || m.Rollback.StuckAt != provision.RbSourceWaiting { |
| + t.Fatalf("not stuck at returning for %q: %+v %+v", reason, m, m.Rollback) |
| + } |
| + if h.route.Status != "releasing" || h.route.HoldOp != m.Rollback.ID || h.rowGone { |
| + t.Fatalf("the row: %+v gone %v", h.route, h.rowGone) |
| + } |
| + if _, err := os.Stat(filepath.Join(h.tdir, p1Host, "upload.txt")); err != nil { |
| + t.Fatalf("the target's files: %v", err) |
| + } |
| + if _, ok := h.claim(t).Op.Journal[moves.JournalReturnSeal]; ok { |
| + t.Fatal("a return seal on a stuck return") |
| + } |
| + if h.lease.acquired != h.lease.released { |
| + t.Fatalf("the staging lease: %d %d", h.lease.acquired, h.lease.released) |
| + } |
| +} |
| + |
| +func (h *rollbackHarness) retry(t *testing.T) { |
| + t.Helper() |
| + g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc} |
| + if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRetry(g)); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// Fix round 1, ruling 2: a dirty return that cannot be built from what the |
| +// target holds stops for an operator with a recorded reason, the target at |
| +// returning with everything kept; a retry runs the return again. |
| +func TestDirtyReturnGoesStuck(t *testing.T) { |
| + t.Run("the database is missing", func(t *testing.T) { |
| + h := newRollbackHarness(t) |
| + db := provision.DefaultDBPath(h.tdir, p1Host) |
| + if err := os.Rename(db, db+".aside"); err != nil { |
| + t.Fatal(err) |
| + } |
| + h.rollBack(t) |
| + h.rollbackPass(t) |
| + h.stuckAtReturning(t, provision.StuckReturnTargetDatabase) |
| + // A stuck record is not the target's to work on. |
| + h.rollbackPass(t) |
| + h.stuckAtReturning(t, provision.StuckReturnTargetDatabase) |
| + // A retry runs the return again; still missing, it stops again. |
| + before := len(h.events) |
| + h.retry(t) |
| + h.rollbackPass(t) |
| + h.stuckAtReturning(t, provision.StuckReturnTargetDatabase) |
| + if got := strings.Join(h.events[before:], " "); got != "target-stuck" { |
| + t.Fatalf("the retry did not run the return: events %q", got) |
| + } |
| + }) |
| + t.Run("the files cannot be archived", func(t *testing.T) { |
| + h := newRollbackHarness(t) |
| + h.targetChanged(t) |
| + link := filepath.Join(h.tdir, p1Host, "link") |
| + h.rbHook = func(point string) { |
| + if point == "after-record-returning" { |
| + if err := os.Symlink("/etc/passwd", link); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + } |
| + h.rollBack(t) |
| + h.rollbackPass(t) |
| + h.stuckAtReturning(t, provision.StuckReturnTargetFiles) |
| + // The operator removes the link and retries: the return completes. |
| + h.rbHook = nil |
| + if err := os.Remove(link); err != nil { |
| + t.Fatal(err) |
| + } |
| + h.retry(t) |
| + h.rollbackPass(t) |
| + if rb := h.rb(t); h.read(t, p1Host).Move.State != provision.MoveRollingBack || rb.TargetPhase != provision.RbTargetReturned { |
| + t.Fatalf("after the retry: %+v", rb) |
| + } |
| + }) |
| + t.Run("a storage error is retried, not stuck", func(t *testing.T) { |
| + h := newRollbackHarness(t) |
| + h.targetChanged(t) |
| + fail := true |
| + h.f.failAt = func(op, key string) error { |
| + if fail && strings.Contains(key, "/replica/") { |
| + return errors.New("503 SlowDown") |
| + } |
| + return nil |
| + } |
| + h.rollBack(t) |
| + h.rollbackPass(t) |
| + if m := h.read(t, p1Host).Move; m.State != provision.MoveRollingBack || m.Rollback.TargetPhase != provision.RbTargetReturning { |
| + t.Fatalf("a storage error: %+v %+v", m, m.Rollback) |
| + } |
| + fail = false |
| + h.rollbackPass(t) |
| + if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReturned { |
| + t.Fatalf("after the storage recovered: %+v", rb) |
| + } |
| + }) |
| +} |
| + |
| +// closeRolledBack is the source's close of the rollback, as Task 5 will |
| +// write it: opened, then MoveRolledBackClose. |
| +func (h *rollbackHarness) closeRolledBack(t *testing.T) { |
| + t.Helper() |
| + g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc} |
| + now := h.clock.Now() |
| + for _, step := range []func(*provision.Record) error{ |
| + provision.MoveRollbackSourcePhase(g, provision.RbSourceRecovered, provision.RbSourceOpened, func(rb *provision.MoveRollback) { rb.OpenedAt = &now }), |
| + provision.MoveRolledBackClose(g, now), |
| + } { |
| + if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, step); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| +} |
| + |
| +// Fix round 1, ruling 3: the source closes the record right after the |
| +// target's cleaned CAS, before the target's worker re-reads it, and the |
| +// worker restarts. No record routes the target back, so the local journal |
| +// sweep parks the op's journal (which holds the built files archive) on the |
| +// close's evidence. |
| +func TestTargetJournalSweptAfterTheClose(t *testing.T) { |
| + h := newRollbackHarness(t) |
| + h.targetChanged(t) |
| + h.rollBack(t) |
| + h.rollbackPass(t) |
| + if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReturned { |
| + t.Fatalf("rollback %+v", rb) |
| + } |
| + adir := archiveDir(h.tdir, p1Host, p1ID) |
| + built, _ := filepath.Glob(filepath.Join(adir, "files-*")) |
| + if len(built) == 0 { |
| + t.Fatal("the dirty return built no archive in the journal") |
| + } |
| + h.recovered(t) |
| + h.rbHook = func(point string) { |
| + if point == "after-record-cleaned" { |
| + h.closeRolledBack(t) |
| + } |
| + } |
| + h.rbAt = "after-record-cleaned" |
| + h.rollbackPass(t) |
| + if !h.rbDone || h.read(t, p1Host).Move != nil { |
| + t.Fatalf("the fixture: crashed %v, record %+v", h.rbDone, h.read(t, p1Host).Move) |
| + } |
| + jdir := moves.JournalDir(h.tdir, p1Host, p1ID) |
| + if _, err := os.Stat(jdir); err != nil { |
| + t.Fatalf("the fixture left no journal to sweep: %v", err) |
| + } |
| + // A restart: a fresh worker, and the next pass's sweep. |
| + h.rw, h.rbHook = nil, nil |
| + h.leases.expire() |
| + h.rollbackPass(t) |
| + if _, err := os.Lstat(jdir); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("the op journal is still live: %v", err) |
| + } |
| + var parkedArchive int |
| + filepath.WalkDir(filepath.Join(h.tdir, ".parked"), func(p string, d os.DirEntry, err error) error { |
| + if err == nil && !d.IsDir() && strings.Contains(p, filepath.Join(".moves", p1Host, p1ID, "archive", "files-")) { |
| + parkedArchive++ |
| + } |
| + return nil |
| + }) |
| + if parkedArchive == 0 { |
| + t.Fatal("the built archive was not parked with the journal") |
| + } |
| +} |
| + |
| +// The journal sweep leaves a rollback journal alone while its record still |
| +// carries the op without target_parked_at: the cleaning is the pass's. |
| +func TestJournalSweepWaitsForTheTargetPark(t *testing.T) { |
| + h := newRollbackHarness(t) |
| + h.rollBack(t) |
| + h.rollbackPass(t) |
| + h.recovered(t) |
| + // Crash right after the moves-owned rename, before the row and the CAS. |
| + h.rbAt = "after-" + rbStepLocal |
| + h.rollbackPass(t) |
| + if !h.rbDone { |
| + t.Fatalf("never reached %s: %v", h.rbAt, h.rbHits) |
| + } |
| + w := &movesWorker{cfg: h.rollbackCfg(t), mirrors: map[string]lease.MirrorStore{}} |
| + w.sweepFinalJournals([]provision.Record{h.read(t, p1Host)}) |
| + if _, err := os.Stat(moves.JournalDir(h.tdir, p1Host, p1ID)); err != nil { |
| + t.Fatalf("the sweep parked a journal whose cleaning is under way: %v", err) |
| + } |
| +} |
| + |
| +// Fix round 1, ruling 4: a crash at each durable step of the cleaning, then a |
| +// fresh worker, finishes it with nothing live left. |
| +func TestTargetCleaningReplays(t *testing.T) { |
| + for _, point := range []string{ |
| + "before-rb-local-retire", "after-rb-local-retire", "after-" + rbStepInstallation, |
| + "after-" + rbStepLocal + "-plan", "after-rename:" + p1Host + ".db-wal", "after-" + rbStepLocal, |
| + "after-rb-row-pending", "after-rb-row-delete", "after-" + rbStepRowDeleted, |
| + } { |
| + t.Run(point, func(t *testing.T) { |
| + h := newRollbackHarness(t) |
| + if err := os.WriteFile(provision.DefaultDBPath(h.tdir, p1Host)+"-wal", []byte("a wal"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + h.rollBack(t) |
| + h.rollbackPass(t) |
| + h.recovered(t) |
| + h.rbAt = point |
| + h.rollbackPass(t) |
| + if !h.rbDone { |
| + t.Fatalf("never reached %s: %v", point, h.rbHits) |
| + } |
| + if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaning { |
| + t.Fatalf("after the crash: %+v", rb) |
| + } |
| + h.rw = nil |
| + h.leases.expire() |
| + h.rollbackPass(t) |
| + if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaned || rb.TargetParkedAt == nil { |
| + t.Fatalf("after the replay: %+v", rb) |
| + } |
| + for _, p := range sourceArtifactPaths(p1Host) { |
| + if _, err := os.Lstat(filepath.Join(h.tdir, p)); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("%s is still live: %v", p, err) |
| + } |
| + } |
| + if _, err := os.Lstat(moves.JournalDir(h.tdir, p1Host, p1ID)); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("the op journal is still live: %v", err) |
| + } |
| + if !h.rowGone { |
| + t.Fatal("the row is still there") |
| + } |
| + if c := h.claim(t); len(c.Inventory["uk-box"].Paths) != 0 { |
| + t.Fatalf("inventory %+v", c.Inventory["uk-box"]) |
| + } |
| + }) |
| + } |
| +} |