amadan
host: the offline open and revert release with the observed record's pair; claims-off holds write theirs
withOfflineHolds sets ReconcileConfig.PairOf to (rec.Incarnation, rec.Move.Seq) under the box judge, so reconcile's hold writes the pair with the status (claims on: nil, SetHoldIf as before). moveCfg gains ReleaseHoldSeq, wired by armOffline to releaseHeldSeq (registry ReleaseHoldSeq, then the stage-requested marker when a row changed). The target's open and the source's revert, under reconcileMu and the host's hold lock, observe the record through the box judge and release only while it still carries this move at offline-open or offline-reverting, with that record's pair; otherwise the pass skips the release, logged. The landing tests' box overrides gain ReleaseHoldSeq and the target's moves config carries the world's judge.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
5 files changed,
+109
−7
internal/host/movesoffline.go+76 −2internal/host/movesoffline_land.go+11 −3internal/host/movesoffline_land_test.go+6 −1internal/host/movesoffline_test.go+1 −0internal/host/movespass.go+15 −1
diff --git a/internal/host/movesoffline.go b/internal/host/movesoffline.go| index d14c807a3..aff34ded4 100644 |
| --- a/internal/host/movesoffline.go |
| +++ b/internal/host/movesoffline.go |
| @@ -110,6 +110,20 @@ func offlineHold(box moveCfg) func(rec provision.Record) (status, op string, ok |
| } |
| } |
| +// releaseHeldSeq is the offline release in production: registry. |
| +// ReleaseHoldSeq, then written(host) only when the release changed a row, as |
| +// releaseHeld does, so root re-judges the now-ready row (registry.SetHold's |
| +// contract, G10). The caller holds provision.WithHostHoldLock. |
| +func releaseHeldSeq(reg *registry.Registry, written func(host string)) func(host, op string, pair registry.Pair) (bool, error) { |
| + return func(host, op string, pair registry.Pair) (bool, error) { |
| + released, err := reg.ReleaseHoldSeq(host, op, pair) |
| + if err == nil && released && written != nil { |
| + written(host) |
| + } |
| + return released, err |
| + } |
| +} |
| + |
| // offlineView is what reconcile's KeepParticipant asks about offline moves: |
| // the offline moves in flight, as the last listing read them. Only keep |
| // reads it; the hold never does (offlineHold). |
| @@ -273,6 +287,10 @@ func armedOfflineCapable() bool { return armedOfflineView() != nil } |
| // withOfflineHolds gives a reconcile pass the offline hooks, only when the |
| // caller and Phase 1's view set none (withMovesHolds runs first): the two |
| // are never armed together, since a box with Phase 1 on is not capable. |
| +// Under a judge (claims off) the hold also writes its lifetime pair |
| +// (self-serve moves plan 1 Task 7): the record's incarnation and the move's |
| +// seq, the move_seq of the transition that asked for the hold. Claims on |
| +// (no judge), PairOf stays nil and the hold is SetHoldIf, the pair untouched. |
| func withOfflineHolds(cfg provision.ReconcileConfig, hold func(provision.Record) (string, string, bool), v *offlineView) provision.ReconcileConfig { |
| if hold == nil || v == nil || cfg.Hold != nil || cfg.KeepParticipant != nil { |
| return cfg |
| @@ -281,9 +299,58 @@ func withOfflineHolds(cfg provision.ReconcileConfig, hold func(provision.Record) |
| if cfg.HoldWritten == nil && v.written != nil { |
| cfg.HoldWritten = v.written |
| } |
| + if cfg.PairOf == nil && cfg.Judge != nil { |
| + cfg.PairOf = offlinePair |
| + } |
| return cfg |
| } |
| +// offlinePair is the lifetime pair an offline hold or release names: the |
| +// record's incarnation and its move's seq (0 with no move). Nil-safe. |
| +func offlinePair(rec provision.Record) registry.Pair { |
| + p := registry.Pair{Incarnation: rec.Incarnation} |
| + if rec.Move != nil { |
| + p.ServingSeq = rec.Move.Seq |
| + } |
| + return p |
| +} |
| + |
| +// releaseObserved is an offline release under reconcileMu and the host's |
| +// hold lock (the caller holds both): the record is observed afresh through |
| +// the box's judge, and the row is released for the move only while that |
| +// record still carries this move in state want, with the observed record's |
| +// pair (the source's revert: the abort's; the target's open: open's). A |
| +// record that moved on, or that is absent, stale or retired, is not |
| +// released for this pass (logged; false, nil); a record that cannot be |
| +// judged is an error the pass reports. true means the release ran (its own |
| +// answer is the row's to tell: the caller re-reads it). |
| +func (w *movesWorker) releaseObserved(ctx context.Context, s release.Store, rec provision.Record, want string) (bool, error) { |
| + cfg, host, id := w.cfg, rec.Host, rec.Move.ID |
| + if cfg.beforeOfflineRelease != nil { |
| + cfg.beforeOfflineRelease(host) |
| + } |
| + obs := cfg.Judge.Observe(ctx, s, rec.Account, rec.App, host) |
| + switch { |
| + case obs.Verdict == provision.LifetimeUnreadable || obs.Verdict == provision.LifetimeCorrupt: |
| + return false, fmt.Errorf("offline move %s: %s: read the record before the release: %w", id, host, obs.Err) |
| + case !obs.Present || obs.Verdict != provision.LifetimeValid || obs.Rec == nil || obs.Rec.Retired: |
| + log.Printf("moves: %s: offline move %s: not releasing the hold this pass: the record is gone or its lifetime is retired", host, id) |
| + return false, nil |
| + } |
| + if m := obs.Rec.Move; m == nil || m.ID != id || m.State != want { |
| + state := "at rest" |
| + if m != nil { |
| + state = m.ID + " " + m.State |
| + } |
| + log.Printf("moves: %s: offline move %s: not releasing the hold this pass: the record is %s, not %s", host, id, state, want) |
| + return false, nil |
| + } |
| + if _, err := cfg.ReleaseHoldSeq(host, id, offlinePair(*obs.Rec)); err != nil { |
| + return false, err |
| + } |
| + return true, nil |
| +} |
| + |
| // newMovesWorker is the worker for cfg: armMovesPass's, and each test box's. |
| func newMovesWorker(cfg moveCfg) *movesWorker { |
| return &movesWorker{cfg: cfg, mirrors: map[string]lease.MirrorStore{}} |
| @@ -1073,18 +1140,25 @@ func (w *movesWorker) offlineSourceGone(ctx context.Context, s release.Store, re |
| func (w *movesWorker) offlineSourceRevert(ctx context.Context, s release.Store, rec provision.Record) error { |
| cfg := w.cfg |
| m, host := rec.Move, rec.Host |
| - if cfg.ReleaseHold == nil { |
| + if cfg.ReleaseHoldSeq == nil { |
| return errors.New("offline move: no registry to release the hold through") |
| } |
| + // The release names the abort's pair, from the record observed under |
| + // the locks (releaseObserved): a record no longer reverting this move |
| + // releases nothing this pass. |
| + var released bool |
| var relErr error |
| reconcileMu.Lock() |
| provision.WithHostHoldLock(host, func() { |
| - _, relErr = cfg.ReleaseHold(host, m.ID) |
| + released, relErr = w.releaseObserved(ctx, s, rec, provision.OfflineReverting) |
| }) |
| reconcileMu.Unlock() |
| if relErr != nil { |
| return relErr |
| } |
| + if !released { |
| + return nil |
| + } |
| w.offlineLiftFence(host, m) |
| rt, err := w.offlineRow(host) |
| if err != nil { |
diff --git a/internal/host/movesoffline_land.go b/internal/host/movesoffline_land.go| index 97cea3d12..6eaf1f6ab 100644 |
| --- a/internal/host/movesoffline_land.go |
| +++ b/internal/host/movesoffline_land.go |
| @@ -123,7 +123,8 @@ func (w *movesWorker) landCrash(point string) error { |
| // still held by the move, every landing item is ensured and verified against |
| // the seal once more, and the verified report is written; only when every |
| // item is OK is the hold released (under reconcileMu and the host's hold |
| -// lock, as the source's revert releases its own). The row then reads ready: |
| +// lock, as the source's revert releases its own), with open's lifetime pair |
| +// from the record observed there (releaseObserved). The row then reads ready: |
| // ordinary adoption stages the channel's build, and the first request |
| // promotes moves/<host>/. Then the move's archive prefix is deleted from this |
| // box's bucket and the box reports opened. A pass after the release finds the |
| @@ -156,18 +157,25 @@ func (w *movesWorker) offlineTargetOpen(ctx context.Context, s release.Store, re |
| return fmt.Errorf("offline move %s: %s: not opening: %s not OK on this box", m.ID, host, strings.Join(bad, ", ")) |
| } |
| items = r.Items |
| - if cfg.ReleaseHold == nil { |
| + if cfg.ReleaseHoldSeq == nil { |
| return errors.New("offline move: no registry to release the hold through") |
| } |
| + // The release names open's pair, from the record observed under the |
| + // locks (releaseObserved): a record no longer open for this move |
| + // releases nothing this pass. |
| + var released bool |
| var relErr error |
| reconcileMu.Lock() |
| provision.WithHostHoldLock(host, func() { |
| - _, relErr = cfg.ReleaseHold(host, m.ID) |
| + released, relErr = w.releaseObserved(ctx, s, rec, provision.OfflineOpen) |
| }) |
| reconcileMu.Unlock() |
| if relErr != nil { |
| return relErr |
| } |
| + if !released { |
| + return nil |
| + } |
| if rt, err = w.offlineRow(host); err != nil { |
| return err |
| } |
diff --git a/internal/host/movesoffline_land_test.go b/internal/host/movesoffline_land_test.go| index c261ffd7c..a8dbc8507 100644 |
| --- a/internal/host/movesoffline_land_test.go |
| +++ b/internal/host/movesoffline_land_test.go |
| @@ -86,6 +86,10 @@ type landOpts struct { |
| // judged arms every box with a lifetime judge, as a claims-off agent |
| // arms newBoxJudge's (the end-to-end test's claims-off variant). |
| judged bool |
| + // lifetime is judged with the record created through its lifetime: the |
| + // host's incarnation slot at 1, live, then the record at incarnation 1 |
| + // (the create protocol's order, self-serve moves plan 1 Task 3). |
| + lifetime bool |
| } |
| // landWorld is offHost at home (ie-a, data-ie) moving to uk (uk-box, |
| @@ -199,7 +203,7 @@ func newLandWorldWith(t *testing.T, opts landOpts) *landWorld { |
| func (lw *landWorld) arm(b *offBox) { |
| bucket, region := b.getenv("CARLOS_HIBERNATE_BUCKET"), b.getenv("CARLOS_HIBERNATE_REGION") |
| b.svc = activator.New(activator.Config{Bucket: bucket, Idle: time.Hour}, &fakeWakeBackend{}, b.reg) |
| - cfg := moveCfg{Region: b.region, Box: b.name, DataDir: b.dd, OptDir: b.opt, Quiesced: b.svc.Quiesced, Route: b.reg.Get, Now: b.wd.clock, |
| + cfg := moveCfg{Judge: b.wd.judge, Region: b.region, Box: b.name, DataDir: b.dd, OptDir: b.opt, Quiesced: b.svc.Quiesced, Route: b.reg.Get, Now: b.wd.clock, |
| OwnBucket: bucket, OwnBucketRegion: region, OwnPrefix: sealPrefix, |
| Mirror: func(ctx context.Context, dst string) (lease.MirrorStore, error) { |
| return lease.NewS3Mirror(ctx, region, dst) |
| @@ -207,6 +211,7 @@ func (lw *landWorld) arm(b *offBox) { |
| armOffline(&cfg, b.dd, b.reg, b.svc, nil, hostwrite.Config{}, b.getenv) |
| cfg.Offline = newOfflineView(cfg, nil) |
| cfg.ReleaseHold = releaseHeld(b.reg, nil) |
| + cfg.ReleaseHoldSeq = releaseHeldSeq(b.reg, nil) |
| cfg.Procs = func(host string) (int, error) { return cgroup.HostProcs(b.cg, host, "") } |
| cfg.LaunchLayers = lw.ukEB.LaunchLayers |
| b.w = newMovesWorker(cfg) |
diff --git a/internal/host/movesoffline_test.go b/internal/host/movesoffline_test.go| index b965d8e38..45cbea708 100644 |
| --- a/internal/host/movesoffline_test.go |
| +++ b/internal/host/movesoffline_test.go |
| @@ -208,6 +208,7 @@ func (b *offBox) restart() { |
| // timer would outlive the test's directories. |
| cfg.Offline = newOfflineView(cfg, nil) |
| cfg.ReleaseHold = releaseHeld(b.reg, nil) |
| + cfg.ReleaseHoldSeq = releaseHeldSeq(b.reg, nil) |
| cfg.Procs = func(host string) (int, error) { return cgroup.HostProcs(b.cg, host, "") } |
| b.w = newMovesWorker(cfg) |
| } |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index 725e17975..e4e76452a 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -370,6 +370,9 @@ func armOffline(cfg *moveCfg, dataDir string, reg *registry.Registry, svc *activ |
| // landed touches the stage-requested marker, so root re-judges the |
| // now-ready row (registry.SetHold's contract, G10). |
| cfg.ReleaseHold = releaseHeld(reg, holdStageRequester(dataDir)) |
| + // The offline open and revert release with the pair of the record |
| + // they observe (releaseObserved), touching the marker the same way. |
| + cfg.ReleaseHoldSeq = releaseHeldSeq(reg, holdStageRequester(dataDir)) |
| // The row and the timers the move landed on it, in one |
| // transaction, only while op holds the row. |
| cfg.DeleteIfHold = reg.DeleteIfHoldWithTimers |
| @@ -912,7 +915,11 @@ type moveCfg struct { |
| // landFault and parkFault are test seams: the landing's crash points |
| // and the park engine's (park.Engine.After). Nil in production. |
| landFault func(point string) error |
| - parkFault func(step string) error |
| + // beforeOfflineRelease is the tests' pause under reconcileMu and the |
| + // host's hold lock, just before an offline release observes the record |
| + // (releaseObserved). Nil in production. |
| + beforeOfflineRelease func(host string) |
| + parkFault func(step string) error |
| // The source's finalization (§7.9 step 3, movesfinal.go). DeleteHeld |
| // deletes this box's row and its timers rows in one transaction, only |
| @@ -1007,6 +1014,13 @@ type moveCfg struct { |
| // hold the row. The caller holds provision.WithHostHoldLock. |
| ReleaseHold func(host, op string) (bool, error) |
| DeleteIfHold func(host, op string) (bool, error) |
| + // ReleaseHoldSeq opens a row held by op and writes the releasing |
| + // transition's lifetime pair with the ready status |
| + // (registry.ReleaseHoldSeq through releaseHeldSeq, touching the |
| + // stage-requested marker when it landed): the offline open and revert |
| + // release through it from the record they observe (self-serve moves |
| + // plan 1 Task 7). The caller holds provision.WithHostHoldLock. |
| + ReleaseHoldSeq func(host, op string, pair registry.Pair) (bool, error) |
| // Registry is this box's registry: TimersFor, ReplaceTimers, |
| // SetNextDue, SetReplication and Get. |
| Registry *registry.Registry |