amadan
host: every moves, offline-stage and run record re-read is the judged reader
The cut, landing, commit, install, finalisation, park, rollback, recovery and revert re-reads, the offline stage round and its deletion gate, and the run launch and cancel checks read through judgedRecord with the worker's judge; MovesRecordsFrom judges by its context's judge. A record its slot no longer vouches for reads as os.ErrNotExist; with claims on the judge is nil and each read is provision.GetRecord as before.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
12 files changed,
+22
−22
internal/host/movescut.go+1 −1internal/host/movesfinal.go+3 −3internal/host/movesland.go+1 −1internal/host/moveslandcommit.go+2 −2internal/host/moveslandinstall.go+1 −1internal/host/movesoffline_stage.go+2 −2internal/host/movesparks.go+1 −1internal/host/movespolicy.go+1 −1internal/host/movesrecover.go+2 −2internal/host/movesrevert.go+3 −3internal/host/movesrollback.go+3 −3internal/host/runexec.go+2 −2
diff --git a/internal/host/movescut.go b/internal/host/movescut.go| index f09cd89d8..815bffc28 100644 |
| --- a/internal/host/movescut.go |
| +++ b/internal/host/movescut.go |
| @@ -99,7 +99,7 @@ func (w *movesWorker) runCut(ctx context.Context, s release.Store, rec provision |
| if err != nil || !advanced { |
| return err |
| } |
| - fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| + fresh, err := judgedRecord(ctx, s, w.cfg.Judge, rec.Account, rec.App, rec.Host) |
| if err != nil { |
| return fmt.Errorf("re-read the record: %w", err) |
| } |
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go| index 711500f6f..19f866c44 100644 |
| --- a/internal/host/movesfinal.go |
| +++ b/internal/host/movesfinal.go |
| @@ -559,7 +559,7 @@ func (f *sourceFinal) parkLocal(ctx context.Context, holder hostclaim.Holder, c |
| return fmt.Errorf("%w: the inventory claims %s", errFinalChanged, p) |
| } |
| } |
| - rec, err := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host) |
| + rec, err := judgedRecord(ctx, f.s, cfg.Judge, f.rec.Account, f.rec.App, host) |
| if err != nil { |
| return fmt.Errorf("re-read the record: %w", err) |
| } |
| @@ -619,7 +619,7 @@ func (f *sourceFinal) deleteRow(ctx context.Context) error { |
| host, id := f.rec.Host, f.g.ID |
| var err error |
| provision.WithHostHoldLock(host, func() { |
| - fresh, e := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host) |
| + fresh, e := judgedRecord(ctx, f.s, cfg.Judge, f.rec.Account, f.rec.App, host) |
| if e != nil { |
| err = fmt.Errorf("re-read the record: %w", e) |
| return |
| @@ -706,7 +706,7 @@ func (f *sourceFinal) stampParked(ctx context.Context) error { |
| f.w.logTransition(ctx, f.s, pre, *after) |
| } |
| if errors.Is(err, provision.ErrMoveConflict) { |
| - fresh, gerr := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host) |
| + fresh, gerr := judgedRecord(ctx, f.s, f.w.cfg.Judge, f.rec.Account, f.rec.App, host) |
| if gerr != nil { |
| return fmt.Errorf("re-read the record: %w", gerr) |
| } |
diff --git a/internal/host/movesland.go b/internal/host/movesland.go| index d87b46eb0..43d1a981d 100644 |
| --- a/internal/host/movesland.go |
| +++ b/internal/host/movesland.go |
| @@ -185,7 +185,7 @@ func (w *movesWorker) landingPass(ctx context.Context, s release.Store, rec prov |
| if err != nil || !advanced { |
| return err |
| } |
| - fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| + fresh, err := judgedRecord(ctx, s, w.cfg.Judge, rec.Account, rec.App, rec.Host) |
| if err != nil { |
| return fmt.Errorf("re-read the record: %w", err) |
| } |
diff --git a/internal/host/moveslandcommit.go b/internal/host/moveslandcommit.go| index ec9a79d9a..548f71953 100644 |
| --- a/internal/host/moveslandcommit.go |
| +++ b/internal/host/moveslandcommit.go |
| @@ -70,7 +70,7 @@ func (l *landingRun) probed(ctx context.Context) (bool, error) { |
| return l.markRollback(ctx, provision.RollbackCommitDeadline) |
| case errors.Is(err, provision.ErrMoveConflict): |
| // A replay finds committed; anything else is another writer's. |
| - if fresh, err = provision.GetRecord(ctx, l.s, l.rec.Account, l.rec.App, l.rec.Host); err != nil { |
| + if fresh, err = judgedRecord(ctx, l.s, l.cfg().Judge, l.rec.Account, l.rec.App, l.rec.Host); err != nil { |
| return false, fmt.Errorf("re-read the record: %w", err) |
| } |
| if fresh == nil || fresh.Move == nil || fresh.Move.ID != l.g.ID || fresh.Move.State != provision.MoveCommitted { |
| @@ -194,7 +194,7 @@ func (l *landingRun) open(ctx context.Context) (stamped bool, err error) { |
| host, id := l.rec.Host, l.g.ID |
| provision.WithHostHoldLock(host, func() { |
| var fresh *provision.Record |
| - if fresh, err = provision.GetRecord(ctx, l.s, l.rec.Account, l.rec.App, host); err != nil { |
| + if fresh, err = judgedRecord(ctx, l.s, cfg.Judge, l.rec.Account, l.rec.App, host); err != nil { |
| return |
| } |
| if fresh == nil || fresh.Move == nil || fresh.Move.ID != id || fresh.Move.Incarnation != l.g.Incarnation || fresh.Move.State != provision.MoveCommitted { |
diff --git a/internal/host/moveslandinstall.go b/internal/host/moveslandinstall.go| index dcc5cf629..dac1a72d6 100644 |
| --- a/internal/host/moveslandinstall.go |
| +++ b/internal/host/moveslandinstall.go |
| @@ -89,7 +89,7 @@ func (ir *installRun) guardedWith(ctx context.Context, captured hostguard.Captur |
| if c.Holder == nil || c.Holder.Stamp != captured.Stamp { |
| return fmt.Errorf("%w: the claim's holder is not at the installation stamp %d", errLandingChanged, captured.Stamp) |
| } |
| - rec, err := provision.GetRecord(ctx, ir.s, ir.rec.Account, ir.rec.App, ir.rec.Host) |
| + rec, err := judgedRecord(ctx, ir.s, ir.w.cfg.Judge, ir.rec.Account, ir.rec.App, ir.rec.Host) |
| if err != nil { |
| return fmt.Errorf("re-read the record: %w", err) |
| } |
diff --git a/internal/host/movesoffline_stage.go b/internal/host/movesoffline_stage.go| index 53690e6ad..c3cb2767c 100644 |
| --- a/internal/host/movesoffline_stage.go |
| +++ b/internal/host/movesoffline_stage.go |
| @@ -72,7 +72,7 @@ func (w *movesWorker) offlineStageStep(ctx context.Context, s release.Store, rec |
| // The record again, now, under this box's pass lock: the pass's listing |
| // may predate the start, and once the move has left staging no round |
| // ever starts (Astra round 2, finding 1). |
| - cur, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| + cur, err := judgedRecord(ctx, s, w.cfg.Judge, rec.Account, rec.App, rec.Host) |
| if err != nil { |
| return fmt.Errorf("offline move %s: read the record before a stage round: %w", m.ID, err) |
| } |
| @@ -159,7 +159,7 @@ func (w *movesWorker) offlineStageRound(ctx context.Context, s release.Store, re |
| return stageProgress{}, err |
| } |
| gated := &stageDeleteGate{MirrorStore: mirror, check: func(ctx context.Context) error { |
| - cur, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| + cur, err := judgedRecord(ctx, s, w.cfg.Judge, rec.Account, rec.App, rec.Host) |
| if err != nil { |
| return fmt.Errorf("read the record before the delete phase: %w", err) |
| } |
diff --git a/internal/host/movesparks.go b/internal/host/movesparks.go| index 62a96eeef..47c7ab5d4 100644 |
| --- a/internal/host/movesparks.go |
| +++ b/internal/host/movesparks.go |
| @@ -59,7 +59,7 @@ import ( |
| // stamps and the claim owes nothing under the op. |
| func (w *movesWorker) closeMove(ctx context.Context, s release.Store, rec provision.Record, fault func(string) error) error { |
| host, id, inc := rec.Host, rec.Move.ID, rec.Move.Incarnation |
| - fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, host) |
| + fresh, err := judgedRecord(ctx, s, w.cfg.Judge, rec.Account, rec.App, host) |
| if err != nil { |
| return fmt.Errorf("re-read the record: %w", err) |
| } |
diff --git a/internal/host/movespolicy.go b/internal/host/movespolicy.go| index 8eed6d740..34e236361 100644 |
| --- a/internal/host/movespolicy.go |
| +++ b/internal/host/movespolicy.go |
| @@ -143,7 +143,7 @@ func MovesRecordsFrom(s release.Store) MovesRecordReader { |
| if s == nil { |
| return nil, hostwrite.ErrUnavailable |
| } |
| - return provision.GetRecord(ctx, s, acct, app, host) |
| + return judgedRecord(ctx, s, provision.JudgeFrom(ctx), acct, app, host) |
| } |
| } |
diff --git a/internal/host/movesrecover.go b/internal/host/movesrecover.go| index 3912124d4..f1c69c93e 100644 |
| --- a/internal/host/movesrecover.go |
| +++ b/internal/host/movesrecover.go |
| @@ -211,7 +211,7 @@ func (w *movesWorker) recoverSource(ctx context.Context, s release.Store, rec pr |
| if err != nil || !advanced { |
| return err |
| } |
| - fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| + fresh, err := judgedRecord(ctx, s, w.cfg.Judge, rec.Account, rec.App, rec.Host) |
| if err != nil { |
| return fmt.Errorf("re-read the record: %w", err) |
| } |
| @@ -379,7 +379,7 @@ func sourceStamp(c *hostclaim.Claim) (uint64, error) { |
| // write: the record, read fresh, must still be this rollback with the |
| // source at phase. It reads the record only. |
| func (r *recoverRun) stillAt(ctx context.Context, phase string) error { |
| - fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host) |
| + fresh, err := judgedRecord(ctx, r.s, r.cfg().Judge, r.rec.Account, r.rec.App, r.rec.Host) |
| if err != nil { |
| return fmt.Errorf("re-read the record: %w", err) |
| } |
diff --git a/internal/host/movesrevert.go b/internal/host/movesrevert.go| index 5a0af451d..ad96c50c7 100644 |
| --- a/internal/host/movesrevert.go |
| +++ b/internal/host/movesrevert.go |
| @@ -241,7 +241,7 @@ func (w *movesWorker) revertSource(ctx context.Context, s release.Store, rec pro |
| if err != nil || !advanced { |
| return err |
| } |
| - fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| + fresh, err := judgedRecord(ctx, s, w.cfg.Judge, rec.Account, rec.App, rec.Host) |
| if err != nil { |
| return fmt.Errorf("re-read the record: %w", err) |
| } |
| @@ -273,7 +273,7 @@ func (r *revertRun) fault(point string) error { |
| // stillReverting is the record check run under the host's hold lock before |
| // a row write: the record, read fresh, must still be this move reverting. |
| func (r *revertRun) stillReverting(ctx context.Context) error { |
| - fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host) |
| + fresh, err := judgedRecord(ctx, r.s, r.cfg().Judge, r.rec.Account, r.rec.App, r.rec.Host) |
| if err != nil { |
| return fmt.Errorf("re-read the record: %w", err) |
| } |
| @@ -632,7 +632,7 @@ func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool, |
| } |
| // The window's end first, against the record as it stands now (the |
| // probe may have recorded its inputs this pass), then the stuck mark. |
| - fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, host) |
| + fresh, err := judgedRecord(ctx, r.s, r.cfg().Judge, r.rec.Account, r.rec.App, host) |
| if err != nil { |
| return false, fmt.Errorf("re-read the record: %w", err) |
| } |
diff --git a/internal/host/movesrollback.go b/internal/host/movesrollback.go| index f7d6c10b7..22f587e94 100644 |
| --- a/internal/host/movesrollback.go |
| +++ b/internal/host/movesrollback.go |
| @@ -255,7 +255,7 @@ func (w *movesWorker) rollbackTarget(ctx context.Context, s release.Store, rec p |
| if err != nil || !advanced { |
| return err |
| } |
| - fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| + fresh, err := judgedRecord(ctx, s, w.cfg.Judge, rec.Account, rec.App, rec.Host) |
| if err != nil { |
| return fmt.Errorf("re-read the record: %w", err) |
| } |
| @@ -374,7 +374,7 @@ func (r *rollbackRun) stillReleasing(ctx context.Context) error { |
| // 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) |
| + fresh, err := judgedRecord(ctx, r.s, r.cfg().Judge, r.rec.Account, r.rec.App, r.rec.Host) |
| if err != nil { |
| return fmt.Errorf("re-read the record: %w", err) |
| } |
| @@ -1668,7 +1668,7 @@ func (r *rollbackRun) deleteRow(ctx context.Context, j *moves.Journal, mark func |
| host, id := r.rec.Host, r.g.ID |
| var err error |
| provision.WithHostHoldLock(host, func() { |
| - fresh, e := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, host) |
| + fresh, e := judgedRecord(ctx, r.s, r.cfg().Judge, r.rec.Account, r.rec.App, host) |
| if e != nil { |
| err = fmt.Errorf("re-read the record: %w", e) |
| return |
diff --git a/internal/host/runexec.go b/internal/host/runexec.go| index f3c0004cc..5266f47af 100644 |
| --- a/internal/host/runexec.go |
| +++ b/internal/host/runexec.go |
| @@ -315,7 +315,7 @@ func (w *runsWorker) launchable(ctx context.Context, req runs.Request, h *runHan |
| } |
| rctx, cancel := context.WithTimeout(ctx, runReadBound) |
| defer cancel() |
| - cur, err := provision.GetRecord(rctx, w.store, req.Account, req.App, req.Host) |
| + cur, err := judgedRecord(rctx, w.store, w.judge, req.Account, req.App, req.Host) |
| if errors.Is(err, os.ErrNotExist) { |
| return errRunNotOurs |
| } |
| @@ -418,7 +418,7 @@ func (w *runsWorker) follow(ctx context.Context, rec provision.Record, req runs. |
| return nil, time.Time{} |
| case <-recheck.C: |
| rctx, cancel := context.WithTimeout(ctx, runReadBound) |
| - cur, err := provision.GetRecord(rctx, w.store, rec.Account, rec.App, rec.Host) |
| + cur, err := judgedRecord(rctx, w.store, w.judge, rec.Account, rec.App, rec.Host) |
| cancel() |
| if err == nil && cur.ActiveRun != nil && cur.ActiveRun.ID == req.ID && cur.ActiveRun.State == provision.RunCancel { |
| h.requestCancel() |