amadan
moves: no merge on write; every journal change commits through UpdateJournal
moves-phase1-6 Tasks 4+5 fix round 2, ruling 2. Journal.Write and its key merge are gone. Every read-modify-write of a move journal goes through moves.UpdateJournal, which holds the journal's lock across the read, the change and the write. A writer that read the journal earlier records its own changes (Mark, the new Unmark, Phase, Attempt) and Commit applies those and nothing else, through UpdateJournal, to the journal as it is then; so a step or phase another writer advanced or deleted is never written back. Creation goes the same way. The 38 production writes and 18 test writes now Commit; the two step removals in the revert's probing are Unmarks. Tests: TestStaleJournalWriterRestoresNothing (an advanced step, a deleted step and a moved phase are never restored), the two-writer cases both ways, a deterministic interleave under the lock (TestJournalLockSerializesWriters), and TestJournalHasNoPlainWrite (Journal has no Write; the whole-file write is called only by UpdateJournal). Census: revertRun.probing's B8 digest 55e249b1... to a3751bbc....
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
22 files changed,
+322
−151
internal/host/movesabort.go+1 −1internal/host/movesarchive.go+1 −1internal/host/movescut.go+3 −3internal/host/movesfinal.go+4 −4internal/host/movesfinal_test.go+10 −4internal/host/movesinstall.go+8 −8internal/host/movesland.go+1 −1internal/host/movesland_probe_test.go+2 −2internal/host/moveslandinstall.go+2 −2internal/host/moveslandprobe.go+2 −2internal/host/movespaging_test.go+1 −1internal/host/movesprobeattempts.go+4 −4internal/host/movesprobeattempts_test.go+1 −1internal/host/movesprobeowners_test.go+5 −5internal/host/movesrecover.go+2 −2internal/host/movesrevert.go+8 −8internal/host/movesrollback.go+4 −4internal/host/movesrollback_finalfix_test.go+2 −2internal/moves/journal.go+71 −47internal/moves/journal_lock_test.go+183 −42internal/moves/journal_test.go+5 −5internal/provision/plan9_b8_census_test.go+2 −2
diff --git a/internal/host/movesabort.go b/internal/host/movesabort.go| index 5453eb862..98742ef20 100644 |
| --- a/internal/host/movesabort.go |
| +++ b/internal/host/movesabort.go |
| @@ -226,7 +226,7 @@ func (w *movesWorker) abortClaim(ctx context.Context, s release.Store, rec provi |
| bound = seal.Bound().UTC() |
| } |
| j.Mark(abortStepBound, bound.Format(time.RFC3339Nano)) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return false, err |
| } |
| if err := w.abFault("after-abort-bound"); err != nil { |
diff --git a/internal/host/movesarchive.go b/internal/host/movesarchive.go| index 28387a8b1..e7b069bef 100644 |
| --- a/internal/host/movesarchive.go |
| +++ b/internal/host/movesarchive.go |
| @@ -150,7 +150,7 @@ func (c *cutRun) cutArchive(ctx context.Context, plain string) (builtArchive, [] |
| return builtArchive{}, nil, "", err |
| } |
| j.Mark(cutStepArchive, string(body)) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return builtArchive{}, nil, "", err |
| } |
| if err := c.fault("archive-journaled"); err != nil { |
diff --git a/internal/host/movescut.go b/internal/host/movescut.go| index db322735c..b373f2e85 100644 |
| --- a/internal/host/movescut.go |
| +++ b/internal/host/movescut.go |
| @@ -437,7 +437,7 @@ func (c *cutRun) stopping(ctx context.Context) (bool, error) { |
| } |
| j.Phase = provision.CutStopping |
| j.Mark(cutStepEvidence, string(body)) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return false, err |
| } |
| if err := c.fault("after-evidence"); err != nil { |
| @@ -593,7 +593,7 @@ func (c *cutRun) giveUpExecution(ctx context.Context, obs hostclaim.Observed, st |
| return moves.StoppedReport{}, err |
| } |
| j.Mark(cutStepCoverage, string(body)) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return moves.StoppedReport{}, err |
| } |
| } |
| @@ -1226,7 +1226,7 @@ func (w *movesWorker) revertClaim(ctx context.Context, s release.Store, rec prov |
| bound = seal.Bound().UTC() |
| } |
| j.Mark(cutStepRevertBound, bound.Format(time.RFC3339Nano)) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return err |
| } |
| } |
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go| index 6c9c328b5..c6401c950 100644 |
| --- a/internal/host/movesfinal.go |
| +++ b/internal/host/movesfinal.go |
| @@ -242,7 +242,7 @@ func (f *sourceFinal) fault(point string) error { |
| // mark journals a step durably, then passes its fault point. |
| func (f *sourceFinal) mark(step, value string) error { |
| f.j.Mark(step, value) |
| - if err := f.j.Write(f.cfg().DataDir); err != nil { |
| + if err := f.j.Commit(f.cfg().DataDir); err != nil { |
| return err |
| } |
| return f.fault("after-" + step) |
| @@ -326,7 +326,7 @@ func (f *sourceFinal) run(ctx context.Context) error { |
| // (sweepFinalJournals). |
| if j.Steps[finalStepParked] != "pending" { |
| j.Mark(finalStepParked, "pending") |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return err |
| } |
| if err := f.fault("after-source-parked-pending"); err != nil { |
| @@ -628,7 +628,7 @@ func (f *sourceFinal) deleteRow(ctx context.Context) error { |
| } |
| if f.j.Steps[finalStepRowDeleted] != "pending" { |
| f.j.Mark(finalStepRowDeleted, "pending") |
| - if err = f.j.Write(cfg.DataDir); err != nil { |
| + if err = f.j.Commit(cfg.DataDir); err != nil { |
| return |
| } |
| if err = f.fault("after-row-pending"); err != nil { |
| @@ -849,7 +849,7 @@ func (w *movesWorker) sweepFinalJournal(dataDir, host, op string, recs []provisi |
| } |
| mark := func(step, value string) error { |
| j.Mark(step, value) |
| - if err := j.Write(dataDir); err != nil { |
| + if err := j.Commit(dataDir); err != nil { |
| return err |
| } |
| if fn := w.cfg.finalFault; fn != nil { |
diff --git a/internal/host/movesfinal_test.go b/internal/host/movesfinal_test.go| index c4cf9d286..c80997ab3 100644 |
| --- a/internal/host/movesfinal_test.go |
| +++ b/internal/host/movesfinal_test.go |
| @@ -734,7 +734,7 @@ func TestMovesParticipantRecord(t *testing.T) { |
| } |
| owns = false |
| j := &moves.Journal{Host: p1Host, Op: p1ID} |
| - if err := j.Write(dd); err != nil { |
| + if err := j.Commit(dd); err != nil { |
| t.Fatal(err) |
| } |
| if got := movesParticipantRecord(cfg, rec(provision.MoveCommitted, "ie", "uk")); got != roleSource { |
| @@ -836,7 +836,7 @@ func TestParkLocalArtifactsRefusals(t *testing.T) { |
| } |
| // Another plan under the same step is refused. |
| p.Paths = []string{"b"} |
| - delete(j.Steps, "s") |
| + j.Unmark("s") |
| if _, err := parkLocalArtifacts(context.Background(), j, mark, p); err == nil { |
| t.Fatal("a plan for other paths was accepted") |
| } |
| @@ -937,8 +937,14 @@ func TestSourceParkSweepLeavesUnstampedJournals(t *testing.T) { |
| t.Fatal(err) |
| } |
| otherSteps := map[string]string{finalStepLocal + "-plan": string(plan), finalStepLocal: "r0", finalStepRowDeleted: "done"} |
| - other := &moves.Journal{Host: p1Host, Op: p1ID2, Steps: otherSteps} |
| - if err := other.Write(h.cfg.DataDir); err != nil { |
| + other, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID2) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for k, v := range otherSteps { |
| + other.Mark(k, v) |
| + } |
| + if err := other.Commit(h.cfg.DataDir); err != nil { |
| t.Fatal(err) |
| } |
| h.finalAt = "after-source-parked-pending" |
diff --git a/internal/host/movesinstall.go b/internal/host/movesinstall.go| index ce72cfd46..78f2aedb2 100644 |
| --- a/internal/host/movesinstall.go |
| +++ b/internal/host/movesinstall.go |
| @@ -414,7 +414,7 @@ func (ir *installRun) parking(ctx context.Context) (bool, error) { |
| j.Phase = phase |
| j.Mark(landStepResidue, "[]") |
| j.Mark(installStepResidueKept, why) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return false, err |
| } |
| log.Printf("moves: %s: the %s parks nothing: %s", host, ir.src.Name, why) |
| @@ -459,7 +459,7 @@ func (ir *installRun) parking(ctx context.Context) (bool, error) { |
| } |
| j.Phase = phase |
| j.Mark(landStepResidue, string(body)) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return false, err |
| } |
| } |
| @@ -593,7 +593,7 @@ func (ir *installRun) building(ctx context.Context) (bool, error) { |
| } |
| j.Phase = phase |
| j.Mark(landStepInstall, string(body)) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return false, err |
| } |
| if err := ir.fault("after-build"); err != nil { |
| @@ -621,7 +621,7 @@ func (ir *installRun) published(ctx context.Context) (bool, error) { |
| } |
| j.Phase = phase |
| j.Mark(landStepPublished, ir.cfg().Now().UTC().Format(time.RFC3339Nano)) |
| - if err := j.Write(ir.cfg().DataDir); err != nil { |
| + if err := j.Commit(ir.cfg().DataDir); err != nil { |
| return false, err |
| } |
| ok, err := ir.advance(ctx, phase, next, installStamp{}) |
| @@ -676,7 +676,7 @@ func (ir *installRun) local_(ctx context.Context) (bool, error) { |
| j.Phase = phase |
| mark := func(step, value string) error { |
| j.Mark(step, value) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return err |
| } |
| return ir.fault("after-" + step) |
| @@ -1110,7 +1110,7 @@ func (ir *installRun) registerDatabase(ctx context.Context, obs hostclaim.Observ |
| } |
| j.Phase = phase |
| j.Mark(landStepInventoryDB, "") |
| - if err := j.Write(ir.cfg().DataDir); err != nil { |
| + if err := j.Commit(ir.cfg().DataDir); err != nil { |
| return err |
| } |
| return ir.fault("after-" + landStepInventoryDB) |
| @@ -1194,7 +1194,7 @@ func (ir *installRun) notePreRender(sha string) error { |
| return err |
| } |
| j.Mark(landStepPreRender, string(body)) |
| - if err := j.Write(ir.cfg().DataDir); err != nil { |
| + if err := j.Commit(ir.cfg().DataDir); err != nil { |
| return err |
| } |
| return ir.fault("after-" + landStepPreRender) |
| @@ -1212,7 +1212,7 @@ func (ir *installRun) noteRendered(sha string) error { |
| return nil |
| } |
| j.Mark(landStepRendered, want) |
| - if err := j.Write(ir.cfg().DataDir); err != nil { |
| + if err := j.Commit(ir.cfg().DataDir); err != nil { |
| return err |
| } |
| return ir.fault("after-" + landStepRendered) |
diff --git a/internal/host/movesland.go b/internal/host/movesland.go| index 089364023..c9ec9548b 100644 |
| --- a/internal/host/movesland.go |
| +++ b/internal/host/movesland.go |
| @@ -576,7 +576,7 @@ func (l *landingRun) checking(ctx context.Context) (bool, error) { |
| case !j.Done(landStepPrevious) && captured.Stamp != 0 && captured.Stamp != holder.Stamp && hostclaim.InventoryHasStamp(obs.Claim, cfg.Box, captured.Stamp): |
| j.Phase = provision.LandChecking |
| j.Mark(landStepPrevious, strconv.FormatUint(captured.Stamp, 10)) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return false, err |
| } |
| log.Printf("moves: %s: an earlier stay at stamp %d will be retired by local", host, captured.Stamp) |
diff --git a/internal/host/movesland_probe_test.go b/internal/host/movesland_probe_test.go| index 2da13677f..6896b4a21 100644 |
| --- a/internal/host/movesland_probe_test.go |
| +++ b/internal/host/movesland_probe_test.go |
| @@ -84,7 +84,7 @@ func (p *probeLog) wake(ctx context.Context, cfg activator.ProbeConfig) (activat |
| if p.openWindow { |
| j.Mark(probeStepOpened+strconv.Itoa(k), p.h.clock.Now().UTC().Format(time.RFC3339Nano)) |
| } |
| - if err := j.Write(p.h.tdir); err != nil { |
| + if err := j.Commit(p.h.tdir); err != nil { |
| return activator.ProbeResult{}, err |
| } |
| p.last = activator.ProbeResult{Session: ah.Session, Stamp: ah.Stamp, Generation: ah.Generation} |
| @@ -317,7 +317,7 @@ func TestLandingProbeCrashRestartsOnlyTheProbe(t *testing.T) { |
| h2.crashLand(t) |
| j2 := h2.landJournal(t) |
| j2.Mark("probe_opened_1", h2.clock.Now().Add(-31*time.Second).UTC().Format(time.RFC3339Nano)) |
| - if err := j2.Write(h2.tdir); err != nil { |
| + if err := j2.Commit(h2.tdir); err != nil { |
| t.Fatal(err) |
| } |
| h2.freshTarget() |
diff --git a/internal/host/moveslandinstall.go b/internal/host/moveslandinstall.go| index 6f5c56245..778cb46f7 100644 |
| --- a/internal/host/moveslandinstall.go |
| +++ b/internal/host/moveslandinstall.go |
| @@ -670,7 +670,7 @@ func (ir *installRun) notePreInstall() error { |
| return nil |
| } |
| j.Mark(landStepPreInstall, string(body)) |
| - if err := j.Write(ir.cfg().DataDir); err != nil { |
| + if err := j.Commit(ir.cfg().DataDir); err != nil { |
| return err |
| } |
| return ir.fault("after-" + landStepPreInstall) |
| @@ -740,7 +740,7 @@ func (ir *installRun) rebuild(ctx context.Context, obs hostclaim.Observed) (move |
| return m, false, err |
| } |
| j.Mark(landStepInstall, string(body)) |
| - return m, false, j.Write(cfg.DataDir) |
| + return m, false, j.Commit(cfg.DataDir) |
| } |
| // writeFileAtomic writes body to path through a temporary file in the same |
diff --git a/internal/host/moveslandprobe.go b/internal/host/moveslandprobe.go| index a379c880a..f9344a0ab 100644 |
| --- a/internal/host/moveslandprobe.go |
| +++ b/internal/host/moveslandprobe.go |
| @@ -184,7 +184,7 @@ func (l *landingRun) journalPassed(k int, res activator.ProbeResult) error { |
| return err |
| } |
| j.Mark(probeStepPassed+strconv.Itoa(k), string(raw)) |
| - return j.Write(l.cfg().DataDir) |
| + return j.Commit(l.cfg().DataDir) |
| } |
| // failAttempt journals attempt k over. |
| @@ -197,7 +197,7 @@ func (l *landingRun) failAttempt(k int) error { |
| return nil |
| } |
| j.Mark(probeStepFailed+strconv.Itoa(k), l.cfg().Now().UTC().Format(time.RFC3339Nano)) |
| - return j.Write(l.cfg().DataDir) |
| + return j.Commit(l.cfg().DataDir) |
| } |
| // passed writes the phase steps of a passing attempt: record, then claim. |
diff --git a/internal/host/movespaging_test.go b/internal/host/movespaging_test.go| index 756ac5fbc..3825ba6a8 100644 |
| --- a/internal/host/movespaging_test.go |
| +++ b/internal/host/movespaging_test.go |
| @@ -437,7 +437,7 @@ func TestPagingIsRateLimited(t *testing.T) { |
| t.Fatal(err) |
| } |
| j.Mark("x", "y") |
| - if err := j.Write(dd); err != nil { |
| + if err := j.Commit(dd); err != nil { |
| t.Fatal(err) |
| } |
| w.flushPageOutbox(ctx, nil) |
diff --git a/internal/host/movesprobeattempts.go b/internal/host/movesprobeattempts.go| index 6a1c71225..056158a31 100644 |
| --- a/internal/host/movesprobeattempts.go |
| +++ b/internal/host/movesprobeattempts.go |
| @@ -266,7 +266,7 @@ func (p *movesProbeAttempts) OpenProbe(ctx context.Context, s OperationStart, pr |
| } |
| step := probeStepOpened + startKey(s, k, recoveryKey) |
| if raw, ok := j.Steps[step]; ok { |
| - if err := j.Write(p.DataDir); err != nil { |
| + if err := j.Commit(p.DataDir); err != nil { |
| return time.Time{}, err |
| } |
| opened, err := time.Parse(time.RFC3339Nano, raw) |
| @@ -277,7 +277,7 @@ func (p *movesProbeAttempts) OpenProbe(ctx context.Context, s OperationStart, pr |
| } |
| opened := proposed.UTC() |
| j.Mark(step, opened.Format(time.RFC3339Nano)) |
| - if err := j.Write(p.DataDir); err != nil { |
| + if err := j.Commit(p.DataDir); err != nil { |
| return time.Time{}, err |
| } |
| return opened, nil |
| @@ -302,7 +302,7 @@ func writeStart(dataDir string, j *moves.Journal, key string, st probeStart) err |
| return err |
| } |
| j.Mark(probeStepStart+key, string(raw)) |
| - return j.Write(dataDir) |
| + return j.Commit(dataDir) |
| } |
| // sameStartOwner reports whether a journalled reservation is the one this |
| @@ -358,7 +358,7 @@ func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n in |
| if _, bound := j.Steps[probeStepSession+strconv.FormatUint(s.Owner.Session, 10)]; bound && st.Attempt.ID != "" { |
| // Reserve(0) validated an existing reservation; the binding it may |
| // have just written is made durable. |
| - if err := j.Write(p.DataDir); err != nil { |
| + if err := j.Commit(p.DataDir); err != nil { |
| return err |
| } |
| } |
diff --git a/internal/host/movesprobeattempts_test.go b/internal/host/movesprobeattempts_test.go| index 3c72cee6d..35a464aa2 100644 |
| --- a/internal/host/movesprobeattempts_test.go |
| +++ b/internal/host/movesprobeattempts_test.go |
| @@ -560,7 +560,7 @@ func TestRecoveryProbeAttemptsLegacyBudget(t *testing.T) { |
| t.Fatal(err) |
| } |
| j.Mark("recovery_attempts_"+string(raw), "9") |
| - if err := j.Write(f.dataDir); err != nil { |
| + if err := j.Commit(f.dataDir); err != nil { |
| t.Fatal(err) |
| } |
| if err := f.ops.Reserve(t.Context(), f.start, 0); !errors.Is(err, ErrRecoveryProbeExhausted) { |
diff --git a/internal/host/movesprobeowners_test.go b/internal/host/movesprobeowners_test.go| index 815f399cb..dccf2ec8f 100644 |
| --- a/internal/host/movesprobeowners_test.go |
| +++ b/internal/host/movesprobeowners_test.go |
| @@ -132,7 +132,7 @@ func newPOFixture(t *testing.T) *poFixture { |
| } |
| placed, _ := json.Marshal(stagedFiles{Files: filesID, Env: envID}) |
| j.Mark(landStepFilesPlaced, string(placed)) |
| - if err := j.Write(f.dataDir); err != nil { |
| + if err := j.Commit(f.dataDir); err != nil { |
| t.Fatal(err) |
| } |
| f.route = registry.Route{Host: poHost, Kind: "instance", App: "jam", Account: "bab", Status: "landing", HoldOp: poOp, Socket: filepath.Join(f.dataDir, poHost+".sock"), DBPath: filepath.Join(f.dataDir, poHost+".db")} |
| @@ -423,12 +423,12 @@ func TestMovesProbeOwnersValidateThePreparedInputs(t *testing.T) { |
| j, _ := moves.ReadJournal(f.dataDir, poHost, poOp) |
| body, _ := json.Marshal(f.manifest) |
| j.Mark(landStepInstall, string(body)) |
| - j.Write(f.dataDir) |
| + j.Commit(f.dataDir) |
| }, hostbox.ErrChanged}, |
| "no manifest": {func(f *poFixture) { |
| j, _ := moves.ReadJournal(f.dataDir, poHost, poOp) |
| - delete(j.Steps, landStepInstall) |
| - j.Write(f.dataDir) |
| + j.Unmark(landStepInstall) |
| + j.Commit(f.dataDir) |
| }, hostbox.ErrChanged}, |
| } { |
| t.Run(name, func(t *testing.T) { |
| @@ -657,7 +657,7 @@ func TestMovesProbeOwnersThroughTheClaimRuntime(t *testing.T) { |
| j, _ := moves.ReadJournal(journalDir, recoveryHost, poOp) |
| body, _ := json.Marshal(manifest) |
| j.Mark(landStepInstall, string(body)) |
| - if err := j.Write(journalDir); err != nil { |
| + if err := j.Commit(journalDir); err != nil { |
| t.Fatal(err) |
| } |
| route := registry.Route{Host: recoveryHost, Kind: "instance", App: "jam", Account: "bab", Status: "landing", HoldOp: poOp, Socket: filepath.Join(f.guard.DataDir, "app.sock"), DBPath: filepath.Join(f.guard.DataDir, "state", "data.db")} |
diff --git a/internal/host/movesrecover.go b/internal/host/movesrecover.go| index 886fc7a09..74bc04486 100644 |
| --- a/internal/host/movesrecover.go |
| +++ b/internal/host/movesrecover.go |
| @@ -891,7 +891,7 @@ func (r *recoverRun) markJournals() error { |
| continue |
| } |
| j.Mark(rcStepSource, string(body)) |
| - if err := j.Write(dd); err != nil { |
| + if err := j.Commit(dd); err != nil { |
| return err |
| } |
| } |
| @@ -922,7 +922,7 @@ func parkRecoveryJournal(dataDir, host, op string, closed bool) error { |
| } |
| write := func(step, value string) error { |
| j.Mark(step, value) |
| - return j.Write(dataDir) |
| + return j.Commit(dataDir) |
| } |
| if closed && j.Steps[rcStepClosed] != "done" { |
| if err := write(rcStepClosed, "done"); err != nil { |
diff --git a/internal/host/movesrevert.go b/internal/host/movesrevert.go| index f995e655d..0a0346de1 100644 |
| --- a/internal/host/movesrevert.go |
| +++ b/internal/host/movesrevert.go |
| @@ -423,7 +423,7 @@ func (r *revertRun) evidence(ctx context.Context, obs hostclaim.Observed) (bool, |
| return false, err |
| } |
| j.Mark(cutStepEvidence, string(body)) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return false, err |
| } |
| if err := r.fault("after-revert-evidence"); err != nil { |
| @@ -556,9 +556,9 @@ func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool, |
| return r.stuck(ctx, ex.Reason, errors.New("the probe's window ended before this pass")) |
| } |
| // Marked stuck and retried since: a new window. |
| - delete(j.Steps, rvStepProbeExhausted) |
| - delete(j.Steps, rvStepProbeSince) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + j.Unmark(rvStepProbeExhausted) |
| + j.Unmark(rvStepProbeSince) |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return false, err |
| } |
| log.Printf("moves: %s: the revert of %s was retried; its probe has a new window", host, id) |
| @@ -567,7 +567,7 @@ func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool, |
| if perr != nil { |
| since = r.now().UTC() |
| j.Mark(rvStepProbeSince, since.Format(time.RFC3339Nano)) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return false, err |
| } |
| } |
| @@ -631,7 +631,7 @@ func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool, |
| return false, err |
| } |
| j.Mark(rvStepProbeExhausted, string(body)) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return false, err |
| } |
| if err := r.fault("after-probe-exhausted"); err != nil { |
| @@ -779,7 +779,7 @@ func (r *revertRun) markJournal() error { |
| return err |
| } |
| j.Mark(rvStepSource, string(body)) |
| - return j.Write(dd) |
| + return j.Commit(dd) |
| } |
| // parkRevertJournal parks the source's journal of a closed revert: closed |
| @@ -806,7 +806,7 @@ func parkRevertJournal(dataDir, host, op string, closed bool) error { |
| } |
| write := func(step, value string) error { |
| j.Mark(step, value) |
| - return j.Write(dataDir) |
| + return j.Commit(dataDir) |
| } |
| if closed && j.Steps[rvStepClosed] != "done" { |
| if err := write(rvStepClosed, "done"); err != nil { |
diff --git a/internal/host/movesrollback.go b/internal/host/movesrollback.go| index 8d4e0775f..326a9ae91 100644 |
| --- a/internal/host/movesrollback.go |
| +++ b/internal/host/movesrollback.go |
| @@ -1027,7 +1027,7 @@ func (r *rollbackRun) cleaning(ctx context.Context) (bool, error) { |
| } |
| mark := func(step, value string) error { |
| j.Mark(step, value) |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return err |
| } |
| return r.fault("after-" + step) |
| @@ -1589,7 +1589,7 @@ func (r *rollbackRun) deleteRow(ctx context.Context, j *moves.Journal, mark func |
| } |
| if j.Steps[rbStepRowDeleted] != "pending" { |
| j.Mark(rbStepRowDeleted, "pending") |
| - if err = j.Write(cfg.DataDir); err != nil { |
| + if err = j.Commit(cfg.DataDir); err != nil { |
| return |
| } |
| if err = r.fault("after-rb-row-pending"); err != nil { |
| @@ -1626,7 +1626,7 @@ func (r *rollbackRun) cleaned() error { |
| } |
| if j.Steps[rbStepCleaned] != "done" { |
| j.Mark(rbStepCleaned, "done") |
| - if err := j.Write(cfg.DataDir); err != nil { |
| + if err := j.Commit(cfg.DataDir); err != nil { |
| return err |
| } |
| } |
| @@ -1682,7 +1682,7 @@ func (w *movesWorker) sweepRollbackJournal(dataDir, host, op string, j *moves.Jo |
| } |
| } |
| j.Mark(rbStepCleaned, "done") |
| - if err := j.Write(dataDir); err != nil { |
| + if err := j.Commit(dataDir); err != nil { |
| return err |
| } |
| } |
diff --git a/internal/host/movesrollback_finalfix_test.go b/internal/host/movesrollback_finalfix_test.go| index db865eb91..f7b3457c5 100644 |
| --- a/internal/host/movesrollback_finalfix_test.go |
| +++ b/internal/host/movesrollback_finalfix_test.go |
| @@ -495,8 +495,8 @@ func TestRollbackCleaningOwnsDatabaseSiblingsOnlyWhenRecordedAbsent(t *testing.T |
| if _, ok := j.Steps[landStepPreInstall]; !ok { |
| t.Fatalf("the install journalled no %s step: %v", landStepPreInstall, j.Steps) |
| } |
| - delete(j.Steps, landStepPreInstall) |
| - if err := j.Write(h.tdir); err != nil { |
| + j.Unmark(landStepPreInstall) |
| + if err := j.Commit(h.tdir); err != nil { |
| t.Fatal(err) |
| } |
| made := plantSiblings(t, h.tdir, "unrecorded") |
diff --git a/internal/moves/journal.go b/internal/moves/journal.go| index d0a3410a3..c8bb22ea7 100644 |
| --- a/internal/moves/journal.go |
| +++ b/internal/moves/journal.go |
| @@ -33,20 +33,29 @@ type Journal struct { |
| Attempt int `json:"attempt,omitempty"` |
| UpdatedAt time.Time `json:"updated_at"` |
| - // read is the steps this value held when it was read or last written: |
| - // a step on disk that it does not hold and did not read was another |
| - // writer's, and Write keeps it. |
| - read map[string]bool |
| + // changes are what this value changed since it was read or last |
| + // committed (Mark, Unmark), in order, and loadedPhase and loadedAttempt |
| + // the Phase and Attempt it was read with: what Commit applies. |
| + changes []journalChange |
| + loadedPhase string |
| + loadedAttempt int |
| +} |
| + |
| +// journalChange is one Mark (del false) or Unmark of a step. |
| +type journalChange struct { |
| + step, value string |
| + del bool |
| } |
| // A move's journal has more than one writer on a box: the pass that runs the |
| -// phase, the probe owners and attempts, the paging rules (Plan 6 fix round |
| -// 1, ruling 3). Every write of one journal holds that journal's lock, the one |
| -// lock every writer in this process shares (journalLock), and under it keeps |
| -// any step another writer added since this value was read, so no writer's |
| -// step is lost to another's read-modify-write. UpdateJournal holds the lock |
| -// across the read as well, for a writer whose decision depends on what is |
| -// there (a rate limit, a recorded digest). |
| +// phase, the probe owners and attempts (Plan 6 fix round 1, ruling 3). Every |
| +// read-modify-write of one journal goes through UpdateJournal, which holds |
| +// that journal's lock, the one lock every writer in this process shares |
| +// (journalLock), across the read, the change and the write (fix round 2, |
| +// ruling 2). There is no plain write: a writer that read the journal earlier |
| +// commits its own changes (Commit), which UpdateJournal applies to the |
| +// journal as it is then, so nothing the writer read and did not change is |
| +// ever written back over another writer's. |
| var journalLocks sync.Map // the journal's directory -> *sync.Mutex |
| // journalLock is the lock of the journal of (host, op) under dataDir. |
| @@ -56,8 +65,8 @@ func journalLock(dataDir, host, op string) *sync.Mutex { |
| } |
| // UpdateJournal reads the journal of (host, op), calls fn with it and writes |
| -// it, all under the journal's lock. An error from fn writes nothing and is |
| -// returned. fn must not call Write. |
| +// it, all under the journal's lock. It makes the journal when there is none. |
| +// An error from fn writes nothing and is returned. fn must not call Commit. |
| func UpdateJournal(dataDir, host, op string, fn func(*Journal) error) error { |
| if err := journalNames(host, op); err != nil { |
| return err |
| @@ -75,13 +84,41 @@ func UpdateJournal(dataDir, host, op string, fn func(*Journal) error) error { |
| return j.write(dataDir) |
| } |
| -// readSteps is the set of steps j holds now. |
| -func (j *Journal) readSteps() map[string]bool { |
| - out := make(map[string]bool, len(j.Steps)) |
| - for k := range j.Steps { |
| - out[k] = true |
| +// Commit makes j's own changes durable: through UpdateJournal, on the |
| +// journal as it is on disk now, it applies what j changed since it was read |
| +// or last committed (its Marks and Unmarks in order, and Phase and Attempt |
| +// if j changed them) and nothing else, and j becomes the journal as written. |
| +// A step another writer advanced or deleted after j was read stays as that |
| +// writer left it unless j itself marks it. |
| +func (j *Journal) Commit(dataDir string) error { |
| + var now *Journal |
| + err := UpdateJournal(dataDir, j.Host, j.Op, func(d *Journal) error { |
| + for _, c := range j.changes { |
| + if c.del { |
| + delete(d.Steps, c.step) |
| + } else { |
| + d.Steps[c.step] = c.value |
| + } |
| + } |
| + if j.Phase != j.loadedPhase { |
| + d.Phase = j.Phase |
| + } |
| + if j.Attempt != j.loadedAttempt { |
| + d.Attempt = j.Attempt |
| + } |
| + now = d |
| + return nil |
| + }) |
| + if err != nil { |
| + return err |
| + } |
| + j.V, j.Phase, j.Attempt, j.UpdatedAt = now.V, now.Phase, now.Attempt, now.UpdatedAt |
| + j.Steps = make(map[string]string, len(now.Steps)) |
| + for k, v := range now.Steps { |
| + j.Steps[k] = v |
| } |
| - return out |
| + j.changes, j.loadedPhase, j.loadedAttempt = nil, j.Phase, j.Attempt |
| + return nil |
| } |
| // JournalDir is where one operation's journal lives on the box: |
| @@ -140,39 +177,19 @@ func ReadJournal(dataDir, host, op string) (*Journal, error) { |
| if j.Steps == nil { |
| j.Steps = map[string]string{} |
| } |
| - j.read = j.readSteps() |
| + j.loadedPhase, j.loadedAttempt = j.Phase, j.Attempt |
| return &j, nil |
| } |
| -// Write stamps UpdatedAt and saves the journal: a temp file in the same |
| -// directory, fsync, rename over the old file, then fsync of the directory, |
| -// so a crash leaves either the old journal or the new one. It holds the |
| -// journal's lock, and keeps the steps other writers added since j was read. |
| -func (j *Journal) Write(dataDir string) error { |
| - if err := journalNames(j.Host, j.Op); err != nil { |
| - return err |
| - } |
| - l := journalLock(dataDir, j.Host, j.Op) |
| - l.Lock() |
| - defer l.Unlock() |
| - return j.write(dataDir) |
| -} |
| - |
| -// write is Write under the journal's lock, which the caller holds. A step on |
| -// disk that j neither holds nor read is another writer's and is kept; a step |
| -// j read and then dropped stays dropped. A journal on disk that cannot be |
| -// read is replaced, as before. |
| +// write stamps UpdatedAt and saves the journal whole: a temp file in the |
| +// same directory, fsync, rename over the old file, then fsync of the |
| +// directory, so a crash leaves either the old journal or the new one. Only |
| +// UpdateJournal calls it, with the journal's lock held since its read |
| +// (TestJournalHasNoPlainWrite). |
| func (j *Journal) write(dataDir string) error { |
| if err := journalNames(j.Host, j.Op); err != nil { |
| return err |
| } |
| - if disk, err := ReadJournal(dataDir, j.Host, j.Op); err == nil { |
| - for k, v := range disk.Steps { |
| - if _, have := j.Steps[k]; !have && !j.read[k] { |
| - j.Mark(k, v) |
| - } |
| - } |
| - } |
| if j.V == 0 { |
| j.V = journalVersion |
| } |
| @@ -224,7 +241,6 @@ func (j *Journal) write(dataDir string) error { |
| os.Remove(tmpName) |
| return err |
| } |
| - j.read = j.readSteps() |
| return syncDir(dir) |
| } |
| @@ -263,10 +279,18 @@ func (j *Journal) Done(step string) bool { |
| } |
| // Mark records step as done with value. It changes the in-memory journal |
| -// only; Write makes it durable. |
| +// only; Commit makes it durable. |
| func (j *Journal) Mark(step, value string) { |
| if j.Steps == nil { |
| j.Steps = map[string]string{} |
| } |
| j.Steps[step] = value |
| + j.changes = append(j.changes, journalChange{step: step, value: value}) |
| +} |
| + |
| +// Unmark removes step. It changes the in-memory journal only; Commit makes |
| +// it durable. |
| +func (j *Journal) Unmark(step string) { |
| + delete(j.Steps, step) |
| + j.changes = append(j.changes, journalChange{step: step, del: true}) |
| } |
diff --git a/internal/moves/journal_lock_test.go b/internal/moves/journal_lock_test.go| index dd51e629e..dbf4bf39e 100644 |
| --- a/internal/moves/journal_lock_test.go |
| +++ b/internal/moves/journal_lock_test.go |
| @@ -1,58 +1,199 @@ |
| package moves |
| import ( |
| + "go/ast" |
| + "go/parser" |
| + "go/token" |
| + "os" |
| + "path/filepath" |
| + "reflect" |
| + "strings" |
| "testing" |
| + "time" |
| ) |
| -// Fix round 1, ruling 3: a move's journal has more than one writer on a box |
| -// (the landing's pass, the probe owners, the paging rules), each a |
| -// read-modify-write. One shared per-journal lock serializes them, and a |
| -// writer whose read came before another's write keeps that write's new |
| -// steps: the landing's progress mark and paged:start_check both survive, |
| -// in either order. A step the writer itself read and deleted stays deleted. |
| +const ( |
| + tjHost = "a.go.example.com" |
| + tjOp = "mv_abcdefghijklmnopqrstuvwxyz" |
| +) |
| + |
| +func tjRead(t *testing.T, dd string) *Journal { |
| + t.Helper() |
| + j, err := ReadJournal(dd, tjHost, tjOp) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return j |
| +} |
| + |
| +func tjCommit(t *testing.T, dd string, j *Journal) { |
| + t.Helper() |
| + if err := j.Commit(dd); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// Fix round 1, ruling 3, and fix round 2, ruling 2: a move's journal has |
| +// more than one writer on a box (the landing's pass, the probe owners and |
| +// attempts), each a read-modify-write through UpdateJournal: a writer that |
| +// read before another's write commits only its own changes, so the |
| +// landing's progress mark and the other writer's step both survive, in |
| +// either order. A step the writer itself removes is removed. |
| func TestJournalWritersKeepEachOthersSteps(t *testing.T) { |
| dd := t.TempDir() |
| - read := func() *Journal { |
| - t.Helper() |
| - j, err := ReadJournal(dd, "a.go.example.com", "mv_abcdefghijklmnopqrstuvwxyz") |
| - if err != nil { |
| - t.Fatal(err) |
| + landing, other := tjRead(t, dd), tjRead(t, dd) |
| + other.Mark("files-plain-g1", "sha256:f") |
| + tjCommit(t, dd, other) |
| + landing.Phase, landing.Attempt = "probing", 1 |
| + landing.Mark("probe-admitting-1", `{"session":7}`) |
| + tjCommit(t, dd, landing) |
| + got := tjRead(t, dd) |
| + if !got.Done("files-plain-g1") || !got.Done("probe-admitting-1") || got.Phase != "probing" || got.Attempt != 1 { |
| + t.Fatalf("after the landing's later commit: %+v", got) |
| + } |
| + if !landing.Done("files-plain-g1") { |
| + t.Fatalf("a committed value is not the journal as written: %+v", landing.Steps) |
| + } |
| + |
| + // The other order: the other writer commits last. |
| + landing, other = tjRead(t, dd), tjRead(t, dd) |
| + landing.Mark("probe-passed-1", "ok") |
| + tjCommit(t, dd, landing) |
| + other.Mark("files-plain-g2", "sha256:g") |
| + tjCommit(t, dd, other) |
| + got = tjRead(t, dd) |
| + if !got.Done("probe-passed-1") || !got.Done("files-plain-g2") || !got.Done("probe-admitting-1") || !got.Done("files-plain-g1") || got.Phase != "probing" { |
| + t.Fatalf("after the other's later commit: %+v", got) |
| + } |
| + |
| + // A step the writer removes is removed; another's is not touched. |
| + j := tjRead(t, dd) |
| + j.Unmark("probe-passed-1") |
| + tjCommit(t, dd, j) |
| + if got := tjRead(t, dd); got.Done("probe-passed-1") || !got.Done("files-plain-g2") { |
| + t.Fatalf("after an Unmark: %+v", got.Steps) |
| + } |
| +} |
| + |
| +// Fix round 2, ruling 2: a stale writer never restores a value another |
| +// writer advanced or deleted after the stale writer read it. It writes only |
| +// what it changed itself. |
| +func TestStaleJournalWriterRestoresNothing(t *testing.T) { |
| + dd := t.TempDir() |
| + if err := UpdateJournal(dd, tjHost, tjOp, func(j *Journal) error { |
| + j.Mark("probe-attempt-1", "started") |
| + j.Mark("scratch", "here") |
| + j.Phase = "restored" |
| + return nil |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + stale := tjRead(t, dd) |
| + // Another writer advances one step, deletes another and moves the phase. |
| + if err := UpdateJournal(dd, tjHost, tjOp, func(j *Journal) error { |
| + j.Mark("probe-attempt-1", "passed") |
| + j.Unmark("scratch") |
| + j.Phase = "probing" |
| + return nil |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + stale.Mark("probe-admitting-2", "x") |
| + tjCommit(t, dd, stale) |
| + got := tjRead(t, dd) |
| + if got.Steps["probe-attempt-1"] != "passed" { |
| + t.Errorf("an advanced step was restored: %q", got.Steps["probe-attempt-1"]) |
| + } |
| + if got.Done("scratch") { |
| + t.Errorf("a deleted step was restored: %+v", got.Steps) |
| + } |
| + if got.Phase != "probing" { |
| + t.Errorf("a phase the stale writer did not change was restored: %q", got.Phase) |
| + } |
| + if !got.Done("probe-admitting-2") { |
| + t.Errorf("the stale writer's own change is lost: %+v", got) |
| + } |
| +} |
| + |
| +// Fix round 1, ruling 3: one lock per journal. A writer that commits while |
| +// another holds the journal in UpdateJournal (the landing's admitting mark) |
| +// waits for it, and both changes are in the journal after. |
| +func TestJournalLockSerializesWriters(t *testing.T) { |
| + dd := t.TempDir() |
| + other := tjRead(t, dd) |
| + other.Mark("files-plain-g1", "sha256:f") |
| + done := make(chan error, 1) |
| + err := UpdateJournal(dd, tjHost, tjOp, func(j *Journal) error { |
| + go func() { done <- other.Commit(dd) }() |
| + select { |
| + case err := <-done: |
| + t.Errorf("a commit wrote the journal (%v) while UpdateJournal held it", err) |
| + case <-time.After(100 * time.Millisecond): |
| } |
| - return j |
| + j.Phase, j.Attempt = "probing", 1 |
| + j.Mark("probe-admitting-1", `{"session":7}`) |
| + return nil |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := <-done; err != nil { |
| + t.Fatal(err) |
| + } |
| + got := tjRead(t, dd) |
| + if !got.Done("probe-admitting-1") || !got.Done("files-plain-g1") || got.Phase != "probing" { |
| + t.Fatalf("the journal lost a writer's change: %+v", got) |
| + } |
| +} |
| + |
| +// Fix round 2, ruling 2: there is no plain write. Journal has no exported |
| +// Write, and the whole-file write (write) is called only by UpdateJournal, |
| +// which holds the lock across the read, the change and the write; every |
| +// other writer commits its own changes through it (Commit). A production |
| +// caller cannot read a journal and then write it whole. |
| +func TestJournalHasNoPlainWrite(t *testing.T) { |
| + if _, ok := reflect.TypeOf(&Journal{}).MethodByName("Write"); ok { |
| + t.Fatal("Journal has a plain Write") |
| + } |
| + files, err := filepath.Glob("*.go") |
| + if err != nil { |
| + t.Fatal(err) |
| } |
| - write := func(j *Journal) { |
| - t.Helper() |
| - if err := j.Write(dd); err != nil { |
| + fset := token.NewFileSet() |
| + var callers []string |
| + for _, f := range files { |
| + if strings.HasSuffix(f, "_test.go") { |
| + continue |
| + } |
| + af, err := parser.ParseFile(fset, f, nil, 0) |
| + if err != nil { |
| t.Fatal(err) |
| } |
| + for _, d := range af.Decls { |
| + fd, ok := d.(*ast.FuncDecl) |
| + if !ok || fd.Body == nil { |
| + continue |
| + } |
| + ast.Inspect(fd.Body, func(n ast.Node) bool { |
| + if c, ok := n.(*ast.CallExpr); ok { |
| + if sel, ok := c.Fun.(*ast.SelectorExpr); ok && sel.Sel.Name == "write" { |
| + callers = append(callers, f+":"+fd.Name.Name) |
| + } |
| + } |
| + return true |
| + }) |
| + } |
| } |
| - landing, paging := read(), read() |
| - paging.Mark("paged:start_check", "2026-10-11T09:00:00Z") |
| - write(paging) |
| - landing.Phase, landing.Attempt = "probing", 1 |
| - landing.Mark("probe-admitting-1", `{"session":7}`) |
| - write(landing) |
| - got := read() |
| - if !got.Done("paged:start_check") || !got.Done("probe-admitting-1") || got.Phase != "probing" || got.Attempt != 1 { |
| - t.Fatalf("after the landing's later write: %+v", got) |
| + if !reflect.DeepEqual(callers, []string{"journal.go:UpdateJournal"}) { |
| + t.Fatalf("the journal's whole-file write is called by %v, want only UpdateJournal", callers) |
| } |
| - |
| - // The other order: the page's write lands last. |
| - landing, paging = read(), read() |
| - landing.Mark("probe-passed-1", "ok") |
| - write(landing) |
| - paging.Mark("paged:stray", "2026-10-11T09:01:00Z") |
| - write(paging) |
| - got = read() |
| - if !got.Done("probe-passed-1") || !got.Done("paged:stray") || !got.Done("probe-admitting-1") || !got.Done("paged:start_check") { |
| - t.Fatalf("after the page's later write: %+v", got.Steps) |
| - } |
| - |
| - // A step the writer read and deleted is not brought back. |
| - j := read() |
| - delete(j.Steps, "probe-passed-1") |
| - write(j) |
| - if got := read(); got.Done("probe-passed-1") || !got.Done("paged:stray") { |
| - t.Fatalf("a deleted step came back, or another went: %+v", got.Steps) |
| + // Every journal written in this package's tests reads back: a smoke |
| + // check that Commit makes a journal where there was none. |
| + dd := t.TempDir() |
| + j := tjRead(t, dd) |
| + j.Mark("a", "b") |
| + tjCommit(t, dd, j) |
| + if _, err := os.Stat(filepath.Join(JournalDir(dd, tjHost, tjOp), journalFile)); err != nil { |
| + t.Fatalf("Commit made no journal: %v", err) |
| } |
| } |
diff --git a/internal/moves/journal_test.go b/internal/moves/journal_test.go| index 3e6ddf3e3..6a364cf0a 100644 |
| --- a/internal/moves/journal_test.go |
| +++ b/internal/moves/journal_test.go |
| @@ -50,7 +50,7 @@ func TestJournalRoundTripAndDone(t *testing.T) { |
| if !j.Done("seal") || !j.Done("flip") || j.Done("land") { |
| t.Fatalf("Done does not reflect Mark: %+v", j.Steps) |
| } |
| - if err := j.Write(dir); err != nil { |
| + if err := j.Commit(dir); err != nil { |
| t.Fatal(err) |
| } |
| if j.UpdatedAt.IsZero() || time.Since(j.UpdatedAt) > time.Minute { |
| @@ -77,7 +77,7 @@ func TestJournalWriteLeavesOnlyTheJournal(t *testing.T) { |
| j.Mark("a", "1") |
| for i := 0; i < 3; i++ { |
| j.Attempt = i |
| - if err := j.Write(dir); err != nil { |
| + if err := j.Commit(dir); err != nil { |
| t.Fatal(err) |
| } |
| } |
| @@ -132,7 +132,7 @@ func TestJournalRejectsUnsafeNames(t *testing.T) { |
| t.Errorf("ReadJournal(%q,%q) accepted", c[0], c[1]) |
| } |
| j := &Journal{V: 1, Host: c[0], Op: c[1]} |
| - if err := j.Write(dir); err == nil { |
| + if err := j.Commit(dir); err == nil { |
| t.Errorf("Write(%q,%q) accepted", c[0], c[1]) |
| } |
| } |
| @@ -181,7 +181,7 @@ func TestJournalWriteCreatesTheChainAndReadsBack(t *testing.T) { |
| t.Fatal(err) |
| } |
| j.Mark("a", op) |
| - if err := j.Write(dir); err != nil { |
| + if err := j.Commit(dir); err != nil { |
| t.Fatalf("Write %s: %v", op, err) |
| } |
| back, err := ReadJournal(dir, "h", op) |
| @@ -204,7 +204,7 @@ func TestJournalWriteUnwritableParentErrorsCleanly(t *testing.T) { |
| } |
| defer os.Chmod(filepath.Join(dir, ".moves"), 0o700) |
| j := &Journal{Host: "h", Op: "op"} |
| - if err := j.Write(dir); err == nil { |
| + if err := j.Commit(dir); err == nil { |
| t.Fatal("Write under an unwritable parent succeeded") |
| } |
| if _, err := os.Stat(JournalDir(dir, "h", "op")); err == nil { |
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go| index 0decb99be..65e6c365f 100644 |
| --- a/internal/provision/plan9_b8_census_test.go |
| +++ b/internal/provision/plan9_b8_census_test.go |
| @@ -551,8 +551,8 @@ var plan9B8CensusInventory = []censusEntry{ |
| {"internal/host/movesrecover.go", "recoverRun.probing", "record-writer", "key:SetRecordMove", "a0144975bf52b29bbc6c2354636d3665487365d24838d3ef4baef1c665ce52ab"}, |
| // Marks recovery stuck through SetRecordMove; event and serial writes are not record objects. |
| {"internal/host/movesrecover.go", "recoverRun.stuck", "record-writer", "key:SetRecordMove", "d634bbd18794aedf081c51f6e3c16b62a9b6b82b21b24f636b949c840c8e5776"}, |
| - // Records revert inputs through SetRecordMove; stuck uses the reviewed record helper and journals are local. |
| - {"internal/host/movesrevert.go", "revertRun.probing", "record-writer", "key:SetRecordMove", "55e249b195b2325917763cdc87cfcd44563d95095918d696f14fa5cec1859898"}, |
| + // Records revert inputs through SetRecordMove; stuck uses the reviewed record helper and journals are local. Plan 6 fix round 2: its journal write is a Commit through moves.UpdateJournal, and its two step removals are Unmarks. |
| + {"internal/host/movesrevert.go", "revertRun.probing", "record-writer", "key:SetRecordMove", "a3751bbc49cfc694236c446473a4bca6c55c2152736dd05f0cdbc1d684379a2d"}, |
| // Applies revert record mutations only through SetRecordMove and its sequenced writer. |
| {"internal/host/movesrevert.go", "revertRun.record", "record-writer", "key:SetRecordMove", "f42100f118eb62ceece23b49423bda2ef357f1aa065760b288bf20340e36a920"}, |
| // Starts rollback through SetRecordMove; the companion claim update targets the separate claim namespace. |