amadan
reconcile and schedule honour move holds: participant rows kept, held rows never ready, timers frozen and drained (§5.2, §9.4, §9.6, K9)
provision: ReconcileConfig.Hold mints a participant's row held at flat paths (G11) and holds an unheld row of its own with SetHold after a fresh record read (G10); recovery holds only mint. KeepParticipant keeps a participant's row through removal. HoldWritten touches the stage-requested marker after every hold write. schedule: HostGate, one per-host gate for the runner, the converger and a cut's drain. The runner skips held rows with no bookkeeping, decides inside its claim and re-checks between timers; Runner.Busy. The converger leaves a frozen host's timers alone inside the gate and holds back the table-wide prune while any host is frozen. host: movesHoldView (hold, keep, frozen; no strict, G9), rebuilt each moves pass from the record list and claim reads, failing closed before its first build and on an unreadable claim; wired into reconcile, the schedule halves and the startup converge. Nothing with the switch off.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
13 files changed,
+1493
−4
approved_moves_copy_test.go+5 −0internal/host/agent.go+11 −0internal/host/moveshold.go+280 −0internal/host/moveshold_test.go+260 −0internal/host/movespass.go+9 −0internal/provision/reconcile.go+136 −0internal/provision/reconcile_test.go+203 −0internal/schedule/converge.go+40 −1internal/schedule/converge_test.go+106 −0internal/schedule/hostgate.go+86 −0internal/schedule/hostgate_test.go+74 −0internal/schedule/runner.go+72 −3internal/schedule/runner_test.go+211 −0
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go| index a2cdc2c3b..e0cebe924 100644 |
| --- a/approved_moves_copy_test.go |
| +++ b/approved_moves_copy_test.go |
| @@ -551,6 +551,7 @@ var movesScanFiles = []string{ |
| "internal/host/movesguard.go", |
| "internal/host/movesheldenv.go", |
| "internal/host/movesheldlock.go", |
| + "internal/host/moveshold.go", |
| "internal/host/movesmeasure.go", |
| "internal/host/movesmeasure_linux.go", |
| "internal/host/movesmeasure_other.go", |
| @@ -594,11 +595,15 @@ var movesScanFiles = []string{ |
| "internal/moves/switch.go", |
| "internal/moves/timers.go", |
| "internal/provision/provision.go", |
| + "internal/provision/reconcile.go", |
| "internal/registry/registry.go", |
| "internal/release/adopt.go", |
| "internal/release/materializeheld.go", |
| "internal/release/proofgate.go", |
| "internal/release/refusal.go", |
| + "internal/schedule/converge.go", |
| + "internal/schedule/hostgate.go", |
| + "internal/schedule/runner.go", |
| "internal/stopevidence/fdidentity_linux.go", |
| "internal/stopevidence/reconstruct.go", |
| "internal/stopevidence/reconstruct_linux.go", |
diff --git a/internal/host/agent.go b/internal/host/agent.go| index bd96fc692..0d1b4e502 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -519,6 +519,10 @@ func runAgent(args []string, buildClaims ClaimFactory) { |
| } |
| schedStore, schedErr := openDeployStore(context.Background()) |
| sd := armSchedule(reg, schedStore, schedErr, svc, sink, dd) |
| + // Move holds (moves spec §9.4): one gate for the runner, the converger |
| + // and a cut's drain, and the converger's freeze. Nothing with the |
| + // Phase 1 switch off. |
| + armScheduleHolds(sd, armedHoldView()) |
| if err := configureScheduleClaims(claims, sd); err != nil { |
| log.Fatalf("agent: host claims: %v", err) |
| } |
| @@ -1128,6 +1132,9 @@ func reconcileSerialized(ctx context.Context, s release.Store, reg *registry.Reg |
| return mintLayoutIntentWith(ctx, s, reg.List) |
| } |
| } |
| + // Move holds (moves spec §9.6): participant rows minted held and kept. |
| + // Unchanged with the Phase 1 switch off. |
| + cfg = withMovesHolds(cfg) |
| return errors.Join( |
| provision.Reconcile(ctx, s, reg, cfg), |
| provision.ReconcileSidecars(ctx, s, reg, cfg), |
| @@ -2189,6 +2196,10 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu |
| sidecarWakeMem := newSidecarWakeMemory() |
| sidecarWakeSeed(ctx, reg, store, sidecarWakeMem) |
| sidecarWake := wakeFn(svc) |
| + // The moves hold view, built before the converge below: that converge |
| + // runs before the first moves pass, and an unbuilt view freezes every |
| + // host (moves spec §9.4). Nothing with the Phase 1 switch off. |
| + refreshMovesHolds(ctx, store) |
| // The schedule converge's own seed, on restartSeed's precedent and for the |
| // same shape of reason: the leg below runs only on a serial MOVE, so |
| // without one pass here a box that booted after the bump would carry no |
diff --git a/internal/host/moveshold.go b/internal/host/moveshold.go| new file mode 100644 |
| index 000000000..375c0037d |
| --- /dev/null |
| +++ b/internal/host/moveshold.go |
| @@ -0,0 +1,280 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "log" |
| + "slices" |
| + "sync/atomic" |
| + |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/schedule" |
| +) |
| + |
| +// movesHoldView is what this box's other passes ask about running moves |
| +// (moves spec §5.2, §9.4, §9.6, §19 K9): reconcile mints and keeps a |
| +// participant's row through hold and keep, and the schedule converger |
| +// leaves a frozen host's timers alone through frozen. Strictness is not |
| +// here: the moves worker's setStrict is the one strict source (G9). |
| +// |
| +// It is rebuilt from every moves pass's record list, plus one claim read |
| +// per host with a Phase 1 move in flight, and read from the reconcile and |
| +// schedule loops in between. It fails closed: until the first build, keep |
| +// and frozen answer true for every host (a removal or a schedule converge |
| +// waits a pass rather than act on a host it cannot vouch for), and a host |
| +// whose claim would not read is kept and frozen until it does. hold judges |
| +// the record the caller passes, not the view's copy, so a flip the view has |
| +// not seen yet still mints the landing row held. |
| +// |
| +// Only built with the Phase 1 switch on: off, armedHoldView is nil and no |
| +// pass is given a hook. |
| +type movesHoldView struct { |
| + box, region string |
| + read func(ctx context.Context, s release.Store, host string) (hostclaim.Observed, error) |
| + // written is reconcile's HoldWritten: the stage-requested marker after |
| + // a hold write (G10). Nil writes nothing. |
| + written func(host string) |
| + // gate is the schedule wiring's HostGate, set by armScheduleHolds; nil |
| + // until then (no scheduled work armed: nothing to drain). |
| + gate atomic.Pointer[schedule.HostGate] |
| + snap atomic.Pointer[holdSnapshot] |
| +} |
| + |
| +// holdSnapshot is one build: what each in-flight move's claim said. |
| +type holdSnapshot struct { |
| + claims map[string]holdClaim // host -> the claim's op, for hosts whose claim read |
| + unknown map[string]bool // hosts whose claim would not read |
| +} |
| + |
| +type holdClaim struct { |
| + kind string |
| + participants []string |
| +} |
| + |
| +func newMovesHoldView(box, region string, read func(context.Context, release.Store, string) (hostclaim.Observed, error), written func(host string)) *movesHoldView { |
| + return &movesHoldView{box: box, region: region, read: read, written: written} |
| +} |
| + |
| +// moveKinds are the claim op kinds a move holds (§19.1). Region HA's |
| +// relocating is its own op and freezes through its own wiring. |
| +var moveKinds = map[string]bool{hostclaim.KindMoving: true, hostclaim.KindLanding: true, |
| + hostclaim.KindReverting: true, hostclaim.KindRestoring: true} |
| + |
| +// refresh rebuilds the view from recs: every host with a Phase 1 move in |
| +// flight has its claim read. Called only with a list that was read; a |
| +// failed listing leaves the last build in place. |
| +func (v *movesHoldView) refresh(ctx context.Context, s release.Store, recs []provision.Record) { |
| + next := &holdSnapshot{claims: map[string]holdClaim{}, unknown: map[string]bool{}} |
| + for _, rec := range recs { |
| + if rec.Move == nil || rec.Move.Legacy() { |
| + continue |
| + } |
| + if _, seen := next.claims[rec.Host]; seen || next.unknown[rec.Host] { |
| + continue |
| + } |
| + obs, err := v.read(ctx, s, rec.Host) |
| + if err != nil { |
| + log.Printf("moves: %s: claim unreadable (%v); its row is kept and its timers frozen until it reads", rec.Host, err) |
| + next.unknown[rec.Host] = true |
| + continue |
| + } |
| + if obs.Absent || obs.Claim == nil { |
| + continue |
| + } |
| + next.claims[rec.Host] = holdClaim{kind: obs.Claim.Op.Kind, participants: slices.Clone(obs.Claim.Op.Participants)} |
| + } |
| + v.snap.Store(next) |
| +} |
| + |
| +// keep is reconcile's KeepParticipant: this box is a participant of a |
| +// move-kind op on host. |
| +func (v *movesHoldView) keep(host string) bool { |
| + snap := v.snap.Load() |
| + if snap == nil || snap.unknown[host] { |
| + return true |
| + } |
| + c, ok := snap.claims[host] |
| + return ok && moveKinds[c.kind] && slices.Contains(c.participants, v.box) |
| +} |
| + |
| +// frozen is the converger's Frozen: a move-kind op holds host's claim. |
| +func (v *movesHoldView) frozen(host string) bool { |
| + snap := v.snap.Load() |
| + if snap == nil || snap.unknown[host] { |
| + return true |
| + } |
| + c, ok := snap.claims[host] |
| + return ok && moveKinds[c.kind] |
| +} |
| + |
| +// The roles a box plays in one move. |
| +const ( |
| + roleNone = "" |
| + roleSource = "source" |
| + roleTarget = "target" |
| +) |
| + |
| +// role is this box's part in rec's move: by the claim's participants |
| +// ({source, target}, §19.1) when the view read the claim, else by the |
| +// record's regions. A box the claim does not name plays none. |
| +func (v *movesHoldView) role(rec provision.Record) (role, claimKind string) { |
| + if snap := v.snap.Load(); snap != nil { |
| + if c, ok := snap.claims[rec.Host]; ok && moveKinds[c.kind] && len(c.participants) == 2 { |
| + switch v.box { |
| + case c.participants[0]: |
| + return roleSource, c.kind |
| + case c.participants[1]: |
| + return roleTarget, c.kind |
| + } |
| + return roleNone, c.kind |
| + } |
| + } |
| + m := rec.Move |
| + switch { |
| + case m.From == v.region && m.To != v.region: |
| + return roleSource, "" |
| + case m.To == v.region && m.From != v.region: |
| + return roleTarget, "" |
| + case m.To == v.region: |
| + // Both ends in this region and no claim to say which box is which: |
| + // the state decides, and either answer holds the row. |
| + switch m.State { |
| + case provision.MoveLanding, provision.MoveCommitted: |
| + return roleTarget, "" |
| + } |
| + return roleSource, "" |
| + } |
| + return roleNone, "" |
| +} |
| + |
| +// hold is reconcile's Hold: the status and op a participant's row is held |
| +// under, from the record's state and this box's role (§5.2, §7.4, §7.6, |
| +// §7.8, §19.1): the source moving from stopping until its row is deleted at |
| +// finalization, restoring through a revert or a rollback; the target landing |
| +// from the flip until the commit opens its row, releasing through a |
| +// rollback. A rollback's holds are under the rollback id; a rollback that |
| +// has none (malformed) holds under the move id rather than not at all. |
| +func (v *movesHoldView) hold(rec provision.Record) (status, op string, ok bool) { |
| + m := rec.Move |
| + if m == nil || m.Legacy() { |
| + return "", "", false |
| + } |
| + role, kind := v.role(rec) |
| + rollbackOp := m.ID |
| + if m.Rollback != nil && m.Rollback.ID != "" { |
| + rollbackOp = m.Rollback.ID |
| + } |
| + switch role { |
| + case roleSource: |
| + switch m.State { |
| + case provision.MoveCutting: |
| + switch m.CutPhase { |
| + case provision.CutStopping, provision.CutStopped, provision.CutStaged, provision.CutArchived, provision.CutFlipped: |
| + return "moving", m.ID, true |
| + } |
| + case provision.MoveLanding, provision.MoveCommitted: |
| + return "moving", m.ID, true |
| + case provision.MoveReverting: |
| + return "restoring", m.ID, true |
| + case provision.MoveAborting: |
| + // Aborting follows a revert (claim reverting, the row held |
| + // restoring) or comes before any cut (claim moving, the row |
| + // serving): only the first is held, and only the claim can say. |
| + if kind == hostclaim.KindReverting { |
| + return "restoring", m.ID, true |
| + } |
| + case provision.MoveRollingBack, provision.MoveStuck: |
| + return "restoring", rollbackOp, true |
| + } |
| + case roleTarget: |
| + switch m.State { |
| + case provision.MoveLanding: |
| + return "landing", m.ID, true |
| + case provision.MoveRollingBack, provision.MoveStuck: |
| + return "releasing", rollbackOp, true |
| + } |
| + } |
| + return "", "", false |
| +} |
| + |
| +// drainTimers is the cut's timer drain (moves spec §9.4 "drained", Task 5's |
| +// moveCfg.TimersDrained): after SetHold, it waits until no delivery or |
| +// convergence that could have missed the hold is still running for host. |
| +// No gate (no scheduled work armed on this box) has nothing to drain. |
| +func (v *movesHoldView) drainTimers(ctx context.Context, host string) error { |
| + return v.gate.Load().Drain(ctx, host) |
| +} |
| + |
| +// armedHoldView is the armed worker's hold view, or nil when no worker is |
| +// armed or the Phase 1 switch is off. |
| +func armedHoldView() *movesHoldView { |
| + w := armedMovesWorker.Load() |
| + if w == nil || !w.cfg.Enabled { |
| + return nil |
| + } |
| + return w.holds |
| +} |
| + |
| +// withMovesHolds gives a reconcile pass the hold view's hooks, unless the |
| +// caller set its own. With the switch off it returns cfg unchanged: main. |
| +func withMovesHolds(cfg provision.ReconcileConfig) provision.ReconcileConfig { |
| + v := armedHoldView() |
| + if v == nil || cfg.Hold != nil || cfg.KeepParticipant != nil { |
| + return cfg |
| + } |
| + cfg.Hold, cfg.KeepParticipant = v.hold, v.keep |
| + if v.written != nil && cfg.HoldWritten == nil { |
| + cfg.HoldWritten = v.written |
| + } |
| + return cfg |
| +} |
| + |
| +// armScheduleHolds wires the schedule halves to the hold view: one HostGate |
| +// shared by the runner, the converger (across both of the agent's converge |
| +// loops: there is one Converger) and the cut's drain, and the converger's |
| +// freeze. v nil (switch off) or sd nil (no scheduled work) leaves both as |
| +// main has them. |
| +func armScheduleHolds(sd *scheduleDeps, v *movesHoldView) { |
| + if sd == nil || v == nil { |
| + return |
| + } |
| + gate := &schedule.HostGate{} |
| + sd.runner.Gate = gate |
| + if sd.conv != nil { |
| + sd.conv.Gate = gate |
| + sd.conv.Frozen = v.frozen |
| + } |
| + v.gate.Store(gate) |
| +} |
| + |
| +// refreshMovesHolds builds the hold view once, outside a moves pass: the |
| +// agent's startup converge runs before the first moves pass, and an unbuilt |
| +// view freezes every host. A failed listing leaves it unbuilt; the first |
| +// moves pass that lists builds it. |
| +func refreshMovesHolds(ctx context.Context, s release.Store) { |
| + v := armedHoldView() |
| + if v == nil || s == nil { |
| + return |
| + } |
| + recs, err := provision.ListAllRecords(ctx, s) |
| + if err != nil { |
| + log.Printf("moves: hold view: list records: %v; rows are kept and timers frozen until a moves pass lists them", err) |
| + return |
| + } |
| + v.refresh(ctx, s, recs) |
| +} |
| + |
| +// holdStageRequester is the hold view's HoldWritten: the stage-requested |
| +// marker, under its own throttle, so root re-judges what it staged under |
| +// the new hold (registry.SetHold's contract, G10). |
| +func holdStageRequester(dd string) func(host string) { |
| + req := markerRequester(dd, func(host, _ string, err error) { |
| + if err != nil { |
| + log.Printf("agent: %s: held for a move, but cannot request a root stage pass: %v", host, err) |
| + return |
| + } |
| + log.Printf("agent: %s: held for a move; asked root to re-judge what it staged", host) |
| + }) |
| + return func(host string) { req(host, "", nil) } |
| +} |
diff --git a/internal/host/moveshold_test.go b/internal/host/moveshold_test.go| new file mode 100644 |
| index 000000000..3bc892d4a |
| --- /dev/null |
| +++ b/internal/host/moveshold_test.go |
| @@ -0,0 +1,260 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/schedule" |
| +) |
| + |
| +const ( |
| + mhHost = "a.go.example.com" |
| + mhID = "mv_abcdefghijklmnopqrstuvwxyz" |
| + mhRbID = "rb_abcdefghijklmnopqrstuvwxyz" |
| +) |
| + |
| +// mhRecord is a Phase 1 record mid-move from ie to uk, in state with the |
| +// given phases; region is the record's region at that state. |
| +func mhRecord(region, state, cutPhase string) provision.Record { |
| + return provision.Record{Schema: 1, Host: mhHost, App: "app", Account: "bab", Region: region, Channel: "stable", |
| + Move: &provision.MoveState{ID: mhID, From: "ie", To: "uk", State: state, CutPhase: cutPhase, |
| + Rollback: &provision.MoveRollback{ID: mhRbID}}} |
| +} |
| + |
| +// mhClaims is a claim reader answering kind with participants ie-box and |
| +// uk-box for mhHost; err makes every read fail. |
| +func mhClaims(kind string, err error) func(context.Context, release.Store, string) (hostclaim.Observed, error) { |
| + return func(_ context.Context, _ release.Store, host string) (hostclaim.Observed, error) { |
| + if err != nil { |
| + return hostclaim.Observed{}, err |
| + } |
| + return hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Host: host, |
| + Op: hostclaim.Op{Kind: kind, ID: mhID, Participants: []string{"ie-box", "uk-box"}}}}, nil |
| + } |
| +} |
| + |
| +func mhView(box, region, kind string, err error, recs ...provision.Record) *movesHoldView { |
| + v := newMovesHoldView(box, region, mhClaims(kind, err), nil) |
| + v.refresh(context.Background(), nil, recs) |
| + return v |
| +} |
| + |
| +// hold's table (moves spec §5.2, §9.6, §19.1): the target holds its row |
| +// landing after the flip, the source moving from stopping, restoring in a |
| +// revert or a rollback, and the target releasing in a rollback. A box the |
| +// claim does not name, and any state with nothing held, answers no hold. |
| +func TestMovesHoldViewHold(t *testing.T) { |
| + cases := []struct { |
| + name string |
| + box, region, kind string |
| + rec provision.Record |
| + wantStatus, wantOp string |
| + wantOK bool |
| + }{ |
| + {"source preparing", "ie-box", "ie", hostclaim.KindMoving, mhRecord("ie", provision.MovePreparing, ""), "", "", false}, |
| + {"source cut requested", "ie-box", "ie", hostclaim.KindMoving, mhRecord("ie", provision.MoveCutting, provision.CutRequested), "", "", false}, |
| + {"source cut admitted", "ie-box", "ie", hostclaim.KindMoving, mhRecord("ie", provision.MoveCutting, provision.CutAdmitted), "", "", false}, |
| + {"source stopping", "ie-box", "ie", hostclaim.KindMoving, mhRecord("ie", provision.MoveCutting, provision.CutStopping), "moving", mhID, true}, |
| + {"source archived", "ie-box", "ie", hostclaim.KindMoving, mhRecord("ie", provision.MoveCutting, provision.CutArchived), "moving", mhID, true}, |
| + {"source after the flip", "ie-box", "ie", hostclaim.KindLanding, mhRecord("uk", provision.MoveLanding, provision.CutFlipped), "moving", mhID, true}, |
| + {"target after the flip", "uk-box", "uk", hostclaim.KindLanding, mhRecord("uk", provision.MoveLanding, provision.CutFlipped), "landing", mhID, true}, |
| + {"target committed", "uk-box", "uk", hostclaim.KindLanding, mhRecord("uk", provision.MoveCommitted, provision.CutFlipped), "", "", false}, |
| + {"source reverting", "ie-box", "ie", hostclaim.KindReverting, mhRecord("ie", provision.MoveReverting, provision.CutStopped), "restoring", mhID, true}, |
| + {"source aborting a revert", "ie-box", "ie", hostclaim.KindReverting, mhRecord("ie", provision.MoveAborting, ""), "restoring", mhID, true}, |
| + {"source aborting before the cut", "ie-box", "ie", hostclaim.KindMoving, mhRecord("ie", provision.MoveAborting, ""), "", "", false}, |
| + {"source rolling back", "ie-box", "ie", hostclaim.KindRestoring, mhRecord("ie", provision.MoveRollingBack, provision.CutFlipped), "restoring", mhRbID, true}, |
| + {"target rolling back", "uk-box", "uk", hostclaim.KindRestoring, mhRecord("ie", provision.MoveRollingBack, provision.CutFlipped), "releasing", mhRbID, true}, |
| + {"target stuck", "uk-box", "uk", hostclaim.KindRestoring, mhRecord("ie", provision.MoveStuck, provision.CutFlipped), "releasing", mhRbID, true}, |
| + {"not a participant", "uk-box-2", "uk", hostclaim.KindLanding, mhRecord("uk", provision.MoveLanding, provision.CutFlipped), "", "", false}, |
| + } |
| + for _, c := range cases { |
| + t.Run(c.name, func(t *testing.T) { |
| + v := mhView(c.box, c.region, c.kind, nil, c.rec) |
| + status, op, ok := v.hold(c.rec) |
| + if status != c.wantStatus || op != c.wantOp || ok != c.wantOK { |
| + t.Fatalf("hold = (%q, %q, %v), want (%q, %q, %v)", status, op, ok, c.wantStatus, c.wantOp, c.wantOK) |
| + } |
| + }) |
| + } |
| + // No move, or a legacy one: nothing held. |
| + v := mhView("uk-box", "uk", hostclaim.KindLanding, nil) |
| + if _, _, ok := v.hold(provision.Record{Host: mhHost}); ok { |
| + t.Fatal("a record at rest is held") |
| + } |
| + if _, _, ok := v.hold(provision.Record{Host: mhHost, Move: &provision.MoveState{To: "uk", State: provision.MoveCutting}}); ok { |
| + t.Fatal("a legacy move is held") |
| + } |
| +} |
| + |
| +// Without a claim answer for the host (the view was built before the move, |
| +// or its claim would not read) hold falls back to the record's regions, and |
| +// never answers "no hold" for a state that holds: the row is never minted |
| +// ready. A rollback with no rollback id holds under the move id. |
| +func TestMovesHoldViewHoldFailsClosedWithoutTheClaim(t *testing.T) { |
| + stale := mhView("uk-box", "uk", hostclaim.KindLanding, nil) // built from a list with no move in it |
| + rec := mhRecord("uk", provision.MoveLanding, provision.CutFlipped) |
| + if status, op, ok := stale.hold(rec); status != "landing" || op != mhID || !ok { |
| + t.Fatalf("stale view: hold = (%q, %q, %v), want landing by the move", status, op, ok) |
| + } |
| + unread := mhView("ie-box", "ie", "", errors.New("store down"), mhRecord("ie", provision.MoveCutting, provision.CutStopping)) |
| + if status, _, ok := unread.hold(mhRecord("ie", provision.MoveCutting, provision.CutStopping)); status != "moving" || !ok { |
| + t.Fatalf("unreadable claim: hold = (%q, %v), want moving", status, ok) |
| + } |
| + noRb := mhRecord("ie", provision.MoveRollingBack, provision.CutFlipped) |
| + noRb.Move.Rollback = nil |
| + if status, op, ok := stale.hold(noRb); status != "releasing" || op != mhID || !ok { |
| + t.Fatalf("rollback with no id: hold = (%q, %q, %v), want releasing by the move", status, op, ok) |
| + } |
| + never := newMovesHoldView("uk-box", "uk", mhClaims(hostclaim.KindLanding, nil), nil) // never built |
| + if status, _, ok := never.hold(rec); status != "landing" || !ok { |
| + t.Fatalf("unbuilt view: hold = (%q, %v), want landing", status, ok) |
| + } |
| +} |
| + |
| +// keep and frozen: a move-kind op freezes the host, and keeps it on a box |
| +// the op names as a participant. A claim that is serving does neither. Fail |
| +// closed: before the first build, and for a host whose claim would not read, |
| +// both answer true. |
| +func TestMovesHoldViewKeepAndFrozen(t *testing.T) { |
| + rec := mhRecord("uk", provision.MoveLanding, provision.CutFlipped) |
| + cases := []struct { |
| + name, box, kind string |
| + err error |
| + keep, frozen bool |
| + }{ |
| + {"moving, source", "ie-box", hostclaim.KindMoving, nil, true, true}, |
| + {"landing, source", "ie-box", hostclaim.KindLanding, nil, true, true}, |
| + {"landing, target", "uk-box", hostclaim.KindLanding, nil, true, true}, |
| + {"reverting, target", "uk-box", hostclaim.KindReverting, nil, true, true}, |
| + {"restoring, target", "uk-box", hostclaim.KindRestoring, nil, true, true}, |
| + {"landing, not a participant", "uk-box-2", hostclaim.KindLanding, nil, false, true}, |
| + {"serving", "ie-box", hostclaim.KindServing, nil, false, false}, |
| + {"relocating is region HA's", "ie-box", hostclaim.KindRelocating, nil, false, false}, |
| + {"unreadable claim", "ie-box", "", errors.New("store down"), true, true}, |
| + } |
| + for _, c := range cases { |
| + t.Run(c.name, func(t *testing.T) { |
| + v := mhView(c.box, "", c.kind, c.err, rec) |
| + if got := v.keep(mhHost); got != c.keep { |
| + t.Errorf("keep = %v, want %v", got, c.keep) |
| + } |
| + if got := v.frozen(mhHost); got != c.frozen { |
| + t.Errorf("frozen = %v, want %v", got, c.frozen) |
| + } |
| + if v.keep("other.go.example.com") || v.frozen("other.go.example.com") { |
| + t.Error("a host with no move is kept or frozen") |
| + } |
| + }) |
| + } |
| + never := newMovesHoldView("ie-box", "ie", mhClaims(hostclaim.KindServing, nil), nil) |
| + if !never.keep("any.go.example.com") || !never.frozen("any.go.example.com") { |
| + t.Fatal("an unbuilt view does not fail closed") |
| + } |
| + // A later build that cannot list keeps the last answer: refresh is only |
| + // called with a list, so a failed list is no refresh at all. |
| + built := mhView("ie-box", "ie", hostclaim.KindMoving, nil, mhRecord("ie", provision.MoveCutting, provision.CutStopping)) |
| + if !built.keep(mhHost) { |
| + t.Fatal("the participant is not kept") |
| + } |
| + built.refresh(context.Background(), nil, nil) |
| + if built.keep(mhHost) || built.frozen(mhHost) { |
| + t.Fatal("a host whose move ended is still kept or frozen") |
| + } |
| +} |
| + |
| +// The view against a real claim in a store, through hostclaim.Read: the |
| +// door's moving claim names ie-box and uk-box. |
| +func TestMovesHoldViewReadsTheClaim(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + created := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) |
| + inc, err := hostclaim.Incarnation("bab", "app", mhHost, created) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + obs, err := hostclaim.Update(ctx, s, hostclaim.Observed{Host: mhHost, Absent: true}, func(c *hostclaim.Claim) error { |
| + c.Holder = &hostclaim.Holder{Family: "instance", Key: provision.RecordKey("bab", "app", mhHost), Account: "bab", App: "app", |
| + Incarnation: inc, CreatedAt: created, Location: "instances/" + mhHost + "/" + inc, Generation: "cur", Stamp: 1} |
| + c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "ie-box"} |
| + c.Touched = hostclaim.Touched{Complete: true, Regions: []string{"ie"}, Boxes: []string{"ie-box"}} |
| + return nil |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := hostclaim.Update(ctx, s, obs, hostclaim.BeginMove(mhID, "op", "ie-box", "enr-ie", "uk-box", "uk", created)); err != nil { |
| + t.Fatal(err) |
| + } |
| + rec := mhRecord("ie", provision.MoveCutting, provision.CutStopping) |
| + ie := newMovesHoldView("ie-box", "ie", hostclaim.Read, nil) |
| + ie.refresh(ctx, s, []provision.Record{rec}) |
| + if !ie.keep(mhHost) || !ie.frozen(mhHost) { |
| + t.Fatal("the source is not kept and frozen under the door's moving claim") |
| + } |
| + if status, op, ok := ie.hold(rec); status != "moving" || op != mhID || !ok { |
| + t.Fatalf("hold = (%q, %q, %v), want moving by the move", status, op, ok) |
| + } |
| + other := newMovesHoldView("ie-box-2", "ie", hostclaim.Read, nil) |
| + other.refresh(ctx, s, []provision.Record{rec}) |
| + if other.keep(mhHost) { |
| + t.Fatal("a box the claim does not name is kept") |
| + } |
| +} |
| + |
| +// The drain reaches the schedule wiring's gate once armScheduleHolds has |
| +// run, and is a no-op before (no scheduled work armed). |
| +func TestMovesHoldViewDrainsTheScheduleGate(t *testing.T) { |
| + v := newMovesHoldView("ie-box", "ie", mhClaims(hostclaim.KindMoving, nil), nil) |
| + if err := v.drainTimers(context.Background(), mhHost); err != nil { |
| + t.Fatalf("drain with no schedule armed = %v", err) |
| + } |
| + sd := &scheduleDeps{runner: &schedule.Runner{}, conv: &schedule.Converger{}} |
| + armScheduleHolds(sd, v) |
| + if sd.runner.Gate == nil || sd.conv.Gate != sd.runner.Gate || sd.conv.Frozen == nil { |
| + t.Fatalf("the schedule halves do not share one gate and the freeze: %+v %+v", sd.runner, sd.conv) |
| + } |
| + if v.gate.Load() != sd.runner.Gate { |
| + t.Fatal("the view does not drain the runner's gate") |
| + } |
| + if err := v.drainTimers(context.Background(), mhHost); err != nil { |
| + t.Fatalf("drain = %v", err) |
| + } |
| +} |
| + |
| +// The switch off (moves.Enabled false) is main: no hold view, so reconcile |
| +// gets no Hold, KeepParticipant or HoldWritten, and the schedule halves no |
| +// gate and no freeze. |
| +func TestMovesHoldsAreMainWithTheSwitchOff(t *testing.T) { |
| + prev := armedMovesWorker.Load() |
| + t.Cleanup(func() { armedMovesWorker.Store(prev) }) |
| + off := &movesWorker{cfg: moveCfg{Enabled: false}, holds: newMovesHoldView("ie-box", "ie", hostclaim.Read, nil)} |
| + armedMovesWorker.Store(off) |
| + if armedHoldView() != nil { |
| + t.Fatal("a hold view with the switch off") |
| + } |
| + cfg := withMovesHolds(provision.ReconcileConfig{}) |
| + if cfg.Hold != nil || cfg.KeepParticipant != nil || cfg.HoldWritten != nil { |
| + t.Fatal("reconcile hooks with the switch off") |
| + } |
| + sd := &scheduleDeps{runner: &schedule.Runner{}, conv: &schedule.Converger{}} |
| + armScheduleHolds(sd, armedHoldView()) |
| + if sd.runner.Gate != nil || sd.conv.Gate != nil || sd.conv.Frozen != nil { |
| + t.Fatal("schedule hooks with the switch off") |
| + } |
| + armedMovesWorker.Store(nil) |
| + if cfg := withMovesHolds(provision.ReconcileConfig{}); cfg.Hold != nil { |
| + t.Fatal("reconcile hooks with no worker armed") |
| + } |
| + |
| + on := &movesWorker{cfg: moveCfg{Enabled: true}, holds: newMovesHoldView("ie-box", "ie", hostclaim.Read, func(string) {})} |
| + armedMovesWorker.Store(on) |
| + cfg = withMovesHolds(provision.ReconcileConfig{}) |
| + if cfg.Hold == nil || cfg.KeepParticipant == nil || cfg.HoldWritten == nil { |
| + t.Fatal("no reconcile hooks with the switch on") |
| + } |
| +} |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index 045b90c84..2fb108a30 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -17,6 +17,7 @@ import ( |
| "github.com/carlosframework/platform/internal/activator" |
| "github.com/carlosframework/platform/internal/appconfig" |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| "github.com/carlosframework/platform/internal/layout" |
| "github.com/carlosframework/platform/internal/lease" |
| "github.com/carlosframework/platform/internal/moves" |
| @@ -84,6 +85,10 @@ type movesWorker struct { |
| measured moves.Measurements |
| notices sync.Map // key → time.Time of the last line |
| strict atomic.Pointer[map[string]bool] |
| + // holds is what reconcile and the schedule halves ask about this |
| + // box's running moves (moveshold.go), rebuilt every pass. Nil with the |
| + // switch off. |
| + holds *movesHoldView |
| } |
| // staging is the memoized staging writer for a target bucket region. |
| @@ -221,6 +226,7 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, |
| w := &movesWorker{mirrors: map[string]lease.MirrorStore{}} |
| if cfg.Enabled { |
| armPhase1(&cfg, w, hib, dataDir, reg, svc, eb, os.Getenv) |
| + w.holds = newMovesHoldView(cfg.Box, cfg.Region, hostclaim.Read, holdStageRequester(dataDir)) |
| } |
| w.cfg = cfg |
| armedMovesWorker.Store(w) |
| @@ -466,6 +472,9 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error { |
| } |
| if cfg.Enabled { |
| w.setStrict(recs) |
| + if w.holds != nil { |
| + w.holds.refresh(ctx, s, recs) |
| + } |
| } |
| for _, rec := range recs { |
| if rec.Move != nil && !rec.Move.Legacy() { |
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go| index 75d64b1f5..96300a8ca 100644 |
| --- a/internal/provision/reconcile.go |
| +++ b/internal/provision/reconcile.go |
| @@ -99,6 +99,27 @@ type ReconcileConfig struct { |
| // wake-time gate (activator.ExecBackend.ConfigGate) is what stops one |
| // starting. Nil: no gate, every record mints as before. |
| ConfigGate func(ctx context.Context, rt registry.Route) error |
| + // Hold answers whether a record's row on this box belongs to a running |
| + // move (moves spec §5.2, §9.6, §19 K9) and, if so, the hold status and |
| + // the op holding it. When ok, a row minted for rec is minted held |
| + // (status, HoldOp op) at flat paths, never ready and never nested |
| + // (Phase 1 lands flat, G11); an unheld row of this pass's own is held |
| + // with SetHold (Upsert never changes a hold on conflict, G10) once a |
| + // fresh read of the record still asks for it, except under a recovery |
| + // hold (restoring, releasing), which only mints (holdExisting); a row |
| + // already held, by this op or another, is left as it is. Nil: no move |
| + // holds anything, main's pass exactly. |
| + Hold func(rec Record) (status, op string, ok bool) |
| + // KeepParticipant answers whether a host's claim names this box as a |
| + // participant of a running operation (claim spec §8, moves §9.6): the |
| + // removal phase never removes such a row, whatever the record's region |
| + // says. The operation removes it at its own finalization. Nil: none is. |
| + KeepParticipant func(host string) bool |
| + // HoldWritten is called after every hold this pass writes (a held mint |
| + // or a SetHold), so the caller can touch the stage-requested marker: |
| + // root re-judges what it staged under the new hold (registry.SetHold's |
| + // contract, G10). Nil: nobody is told. |
| + HoldWritten func(host string) |
| } |
| // InboundAny and InboundNone are the closed vocabulary ReconcileConfig.Inbound |
| @@ -1048,6 +1069,18 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| // clobber, and a later pass converges it once the manifest reads. |
| seed, seedIsKnown := seedFor(rec.Account, rec.App) |
| existing, ok := byHost[rec.Host] |
| + if cfg.Hold != nil { |
| + if status, op, held := cfg.Hold(rec); held { |
| + if !ok { |
| + mintHeld(ctx, reg, cfg, rec, seed, status, op) |
| + continue |
| + } |
| + // An existing row is held in place, then converged as |
| + // usual: the owned branch's narrow setters never touch |
| + // status or hold_op. |
| + holdExisting(ctx, s, reg, cfg, rec, existing, status, op) |
| + } |
| + } |
| switch { |
| case !ok: |
| // A box that declared CARLOS_INBOUND=none (design §8) has told |
| @@ -1330,6 +1363,14 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| log.Printf("provision: not removing %s: its record exists but could not be read cleanly this pass", rt.Host) |
| continue |
| } |
| + // A participant of a running operation keeps its row whatever |
| + // the record's region says (moves spec §9.6, claim spec §8): a |
| + // move's source after the flip, its target during a rollback. |
| + // The operation deletes the row at its own finalization. |
| + if cfg.KeepParticipant != nil && cfg.KeepParticipant(rt.Host) { |
| + log.Printf("provision: not removing %s (%s): participant of a running operation", rt.Host, why) |
| + continue |
| + } |
| if cfg.DryRun { |
| log.Printf("provision: dry-run: would remove retired %s route %s (%s)", rt.Kind, rt.Host, why) |
| continue |
| @@ -1344,6 +1385,101 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| return nil |
| } |
| +// mintHeld mints a move participant's row held (moves spec §9.6 "mint"): the |
| +// record's own route at flat paths whatever the box's layout intent (Phase 1 |
| +// lands flat, G11), with status and hold_op set by the INSERT, so it is never |
| +// ready for even one pass. The refusals a ready mint has (no-inbound box, |
| +// the config gate, a dry run) apply unchanged. A status that is not a hold, |
| +// or an empty op, mints nothing: a participant's row is never minted ready. |
| +func mintHeld(ctx context.Context, reg *registry.Registry, cfg ReconcileConfig, rec Record, seed, status, op string) { |
| + if !registry.IsHold(status) || op == "" { |
| + log.Printf("provision: %s: not minting: the move's hold %q by %q is not a hold", rec.Host, status, op) |
| + return |
| + } |
| + if cfg.Inbound == InboundNone { |
| + log.Printf("provision: refusing to mint held route %s (app %q acct %q): this box declares "+ |
| + "CARLOS_INBOUND=none (no-inbound) and cannot serve routed instance traffic", rec.Host, rec.App, rec.Account) |
| + return |
| + } |
| + nr := routeFor(rec, cfg.DataDir, seed, false) |
| + nr.Status, nr.HoldOp = status, op |
| + if cfg.ConfigGate != nil { |
| + if err := cfg.ConfigGate(ctx, nr); err != nil { |
| + return |
| + } |
| + } |
| + if cfg.DryRun { |
| + log.Printf("provision: dry-run: would mint %s route %s held %s by %s (app %q acct %q channel %q)", |
| + nr.Kind, rec.Host, status, op, rec.App, rec.Account, rec.Channel) |
| + return |
| + } |
| + if err := reg.Upsert(nr); err != nil { |
| + log.Printf("provision: mint held %s: %v", rec.Host, err) |
| + return |
| + } |
| + log.Printf("provision: minted %s route %s held %s by %s (app %q acct %q)", nr.Kind, rec.Host, status, op, rec.App, rec.Account) |
| + if cfg.HoldWritten != nil { |
| + cfg.HoldWritten(rec.Host) |
| + } |
| +} |
| + |
| +// holdExisting holds an existing row for a move (G10): Upsert never changes |
| +// a hold on conflict, so an unheld row of this pass's own is held with |
| +// SetHold. A row already held by op is left as is; a row held by another op |
| +// is that op's, and is left alone; a row this pass did not mint (not |
| +// Provisioned, or another app's) is not taken into a move. |
| +// |
| +// Two rules keep a hold write from closing a row an operation has opened: |
| +// |
| +// - a recovery hold (restoring, releasing) is never written over an |
| +// unheld row. The recovering operation holds its own row as it enters |
| +// the recovery, and opens it while the record still names the |
| +// recovery (moves spec §7.4 reverting `probed`, §7.8 `opened`), so an |
| +// unheld row there is one it opened. Only a missing row is minted. |
| +// - the record is read again just before the write, and the hold is |
| +// asked again of the fresh record: this pass listed the records before |
| +// the routes, and a target opens its row only after its commit, so a |
| +// fresh record still at landing means the row has not been opened. |
| +// |
| +// SetHold can wait out root's activation guard and fail (G30): logged, and |
| +// the next pass tries again. |
| +func holdExisting(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, rec Record, existing registry.Route, status, op string) { |
| + switch { |
| + case !registry.IsHold(status) || op == "": |
| + log.Printf("provision: %s: not holding: the move's hold %q by %q is not a hold", rec.Host, status, op) |
| + return |
| + case existing.HoldOp == op: |
| + return |
| + case existing.HoldOp != "": |
| + log.Printf("provision: %s: held %s by %s, not by %s; leaving it", rec.Host, existing.Status, existing.HoldOp, op) |
| + return |
| + case status == "restoring" || status == "releasing": |
| + return |
| + case !existing.Provisioned || existing.App != rec.App || existing.Account != rec.Account: |
| + log.Printf("provision: %s: the row is not this pass's (app %q acct %q); not holding it for %s", rec.Host, existing.App, existing.Account, op) |
| + return |
| + case cfg.DryRun: |
| + log.Printf("provision: dry-run: would hold %s %s by %s", rec.Host, status, op) |
| + return |
| + } |
| + fresh, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| + if err != nil { |
| + log.Printf("provision: %s: not holding this pass: re-read the record: %v", rec.Host, err) |
| + return |
| + } |
| + if st, o, ok := cfg.Hold(*fresh); !ok || st != status || o != op { |
| + return // the move moved on since the listing; the next pass judges it afresh |
| + } |
| + if err := reg.SetHold(rec.Host, status, op); err != nil { |
| + log.Printf("provision: hold %s %s by %s: %v", rec.Host, status, op, err) |
| + return |
| + } |
| + log.Printf("provision: held %s %s by %s (was %s)", rec.Host, status, op, existing.Status) |
| + if cfg.HoldWritten != nil { |
| + cfg.HoldWritten(rec.Host) |
| + } |
| +} |
| + |
| // gateWord renders a gate column for a log line: "open" or "set", never the |
| // hash — it has no business in a journal. |
| func gateWord(hash string) string { |
diff --git a/internal/provision/reconcile_test.go b/internal/provision/reconcile_test.go| index 624517255..6a4ff8e4d 100644 |
| --- a/internal/provision/reconcile_test.go |
| +++ b/internal/provision/reconcile_test.go |
| @@ -3150,3 +3150,206 @@ func TestReconcileAdoptsAHandAddedNestedRoute(t *testing.T) { |
| t.Errorf("adoption rewrote the nested paths: %+v", got) |
| } |
| } |
| + |
| +// ---- move holds (moves spec §5.2, §9.6, §19 K9) ---- |
| + |
| +// movingRecord is acme with a Phase 1 move from ie to uk after the flip: the |
| +// record's region is already the target's. |
| +func movingRecord(sqid string) Record { |
| + rec := acmeRecord(sqid) |
| + rec.Region = "uk" |
| + rec.Move = &MoveState{ID: "mv_acme", From: "ie", To: "uk", State: MoveLanding, LandingPhase: LandChecking, |
| + RequestedAt: time.Now(), UpdatedAt: time.Now()} |
| + return rec |
| +} |
| + |
| +// holdLog records the hold writes a pass reports through HoldWritten. |
| +type holdLog struct{ hosts []string } |
| + |
| +func (h *holdLog) written(host string) { h.hosts = append(h.hosts, host) } |
| + |
| +// Review Focus 5: on the target the landing row is minted held and never |
| +// ready, at flat paths whatever the box's layout intent says (G11); on the |
| +// source, a participant's moving row survives the record naming another |
| +// region, and without the participant answer it is removed as today. |
| +func TestReconcileHoldsParticipantRows(t *testing.T) { |
| + s, sqid := setup(t) |
| + rec := movingRecord(sqid) |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + hold := func(r Record) (string, string, bool) { |
| + if r.Host == rec.Host && r.Move != nil { |
| + return "landing", r.Move.ID, true |
| + } |
| + return "", "", false |
| + } |
| + |
| + // The uk box: minted held, twice over, never ready; flat paths. |
| + uk := regWith(t) |
| + var wrote holdLog |
| + ukCfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data", Hold: hold, HoldWritten: wrote.written, |
| + Layout: func(context.Context, release.Store) (string, error) { return release.LayoutModeNested, nil }} |
| + for pass := 1; pass <= 2; pass++ { |
| + if err := Reconcile(ctx, s, uk, ukCfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + got := mustGet(t, uk, rec.Host) |
| + if got.Status != "landing" || got.HoldOp != "mv_acme" { |
| + t.Fatalf("pass %d: the uk row is %q held by %q, want landing held by mv_acme", pass, got.Status, got.HoldOp) |
| + } |
| + if got.DBPath != DefaultDBPath("/data", rec.Host) || got.Socket != DefaultSocket(rec.Host) { |
| + t.Fatalf("pass %d: the held row is not flat: db %q socket %q", pass, got.DBPath, got.Socket) |
| + } |
| + } |
| + if len(wrote.hosts) != 1 || wrote.hosts[0] != rec.Host { |
| + t.Fatalf("hold writes reported = %v, want the one mint", wrote.hosts) |
| + } |
| + |
| + // The ie box: the source's moving row, held by the move. |
| + ieRow := routeFor(acmeRecord(sqid), "/data", `["seed"]`, false) |
| + ieRow.Status, ieRow.HoldOp = "moving", "mv_acme" |
| + ie := regWith(t, ieRow) |
| + keep := true |
| + ieCfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", |
| + KeepParticipant: func(host string) bool { return keep && host == rec.Host }} |
| + if err := Reconcile(ctx, s, ie, ieCfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := mustGet(t, ie, rec.Host); got.Status != "moving" || got.HoldOp != "mv_acme" { |
| + t.Fatalf("the participant's row changed: %+v", got) |
| + } |
| + keep = false |
| + if err := Reconcile(ctx, s, ie, ieCfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if rt, err := ie.Get(rec.Host); err != nil || rt != nil { |
| + t.Fatalf("without the participant answer the row was not removed: %+v %v", rt, err) |
| + } |
| +} |
| + |
| +// G10: Upsert cannot hold an existing row, so Hold on an unheld row of this |
| +// pass's own is SetHold, and reported; a row already held by the same op is |
| +// left as is, unreported; a row held by another op is left alone; a row this |
| +// pass does not own is not taken into a move. |
| +func TestReconcileHoldsAnExistingRow(t *testing.T) { |
| + s, sqid := setup(t) |
| + rec := movingRecord(sqid) |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + owned := routeFor(rec, "/data", `["seed"]`, false) |
| + cases := []struct { |
| + name string |
| + row registry.Route |
| + wantStatus, wantOp string |
| + reported bool |
| + }{ |
| + {"unheld, owned", owned, "landing", "mv_acme", true}, |
| + {"held by this op", withHold(owned, "landing", "mv_acme"), "landing", "mv_acme", false}, |
| + {"held by another op", withHold(owned, "restoring", "rb_other"), "restoring", "rb_other", false}, |
| + {"not this pass's row", registry.Route{Host: rec.Host, Kind: "instance", App: "other", Status: "ready", Addr: "127.0.0.1:9"}, "ready", "", false}, |
| + // A recovery hold is written by the operation itself, which opens its |
| + // own row while the record still names the recovery (§7.4 reverting |
| + // probed, §7.8 opened): an unheld row there is one it opened. |
| + {"unheld, a recovery hold", owned, "ready", "", false}, |
| + } |
| + for _, c := range cases { |
| + t.Run(c.name, func(t *testing.T) { |
| + reg := regWith(t, c.row) |
| + var wrote holdLog |
| + status := "landing" |
| + if strings.Contains(c.name, "recovery") { |
| + status = "restoring" |
| + } |
| + cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data", HoldWritten: wrote.written, |
| + Hold: func(r Record) (string, string, bool) { return status, r.Move.ID, r.Move.State == MoveLanding }} |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + got := mustGet(t, reg, rec.Host) |
| + if got.Status != c.wantStatus || got.HoldOp != c.wantOp { |
| + t.Fatalf("row is %q held by %q, want %q held by %q", got.Status, got.HoldOp, c.wantStatus, c.wantOp) |
| + } |
| + if reported := len(wrote.hosts) > 0; reported != c.reported { |
| + t.Fatalf("hold write reported = %v, want %v", wrote.hosts, c.reported) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func withHold(rt registry.Route, status, op string) registry.Route { |
| + rt.Status, rt.HoldOp = status, op |
| + return rt |
| +} |
| + |
| +// A dry run reports a held mint and writes nothing. |
| +func TestReconcileDryRunWritesNoHold(t *testing.T) { |
| + s, sqid := setup(t) |
| + rec := movingRecord(sqid) |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + reg := regWith(t) |
| + var wrote holdLog |
| + cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data", DryRun: true, HoldWritten: wrote.written, |
| + Hold: func(Record) (string, string, bool) { return "landing", "mv_acme", true }} |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if rt, _ := reg.Get(rec.Host); rt != nil || len(wrote.hosts) != 0 { |
| + t.Fatalf("a dry run wrote: %+v %v", rt, wrote.hosts) |
| + } |
| +} |
| + |
| +// changingStore answers the record's key with first, then with every later |
| +// read with later: the record moved on between the pass's listing and its |
| +// hold write. |
| +type changingStore struct { |
| + release.Store |
| + key string |
| + reads int |
| + first, later []byte |
| +} |
| + |
| +func (c *changingStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if key == c.key { |
| + c.reads++ |
| + body := c.later |
| + if c.reads == 1 { |
| + body = c.first |
| + } |
| + return body, fmt.Sprintf("etag-%d", c.reads), nil |
| + } |
| + return c.Store.Get(ctx, key) |
| +} |
| + |
| +// A hold write on an existing row re-reads the record first: the target |
| +// opens its row right after its commit, and the pass's listing (read before |
| +// the routes) can still say landing. A record that has moved on is not held |
| +// again. |
| +func TestReconcileHoldRereadsTheRecordBeforeHoldingARow(t *testing.T) { |
| + s, sqid := setup(t) |
| + rec := movingRecord(sqid) |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + committed := rec |
| + m := *rec.Move |
| + m.State = MoveCommitted |
| + committed.Move = &m |
| + cs := &changingStore{Store: s, key: RecordKey(rec.Account, rec.App, rec.Host), first: mustJSON(t, rec), later: mustJSON(t, committed)} |
| + reg := regWith(t, routeFor(rec, "/data", `["seed"]`, false)) // the target's row, opened |
| + var wrote holdLog |
| + cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data", HoldWritten: wrote.written, |
| + Hold: func(r Record) (string, string, bool) { return "landing", r.Move.ID, r.Move.State == MoveLanding }} |
| + if err := Reconcile(ctx, cs, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if cs.reads < 2 { |
| + t.Fatalf("the record was read %d time(s); the hold write did not re-read it", cs.reads) |
| + } |
| + if got := mustGet(t, reg, rec.Host); got.Status != "ready" || got.HoldOp != "" || len(wrote.hosts) != 0 { |
| + t.Fatalf("an opened row was held again from a stale listing: %+v %v", got, wrote.hosts) |
| + } |
| +} |
diff --git a/internal/schedule/converge.go b/internal/schedule/converge.go| index 687eaf201..534f1da0e 100644 |
| --- a/internal/schedule/converge.go |
| +++ b/internal/schedule/converge.go |
| @@ -37,6 +37,18 @@ type Converger struct { |
| Report func(rt registry.Route, st Status) // called for every host whose rows changed or that is unsupported; nil = no reports |
| Idle time.Duration // the activator's idle window; zero when no hibernation is armed on this box |
| HostWrites hostwrite.Config |
| + // Frozen answers whether a move holds host's claim (moves spec §9.4, |
| + // §19 K9): its timers are neither upserted nor pruned. Nil: none is. |
| + Frozen func(host string) bool |
| + // Gate is the per-host gate shared with the runner and a move's drain; |
| + // convergeHost holds it across its Frozen check and its writes. Nil: |
| + // no gate, main's converger. |
| + Gate *HostGate |
| + |
| + // afterFrozenCheck is the tests' pause inside the gate, after the |
| + // Frozen check (TestConvergerCannotWriteAfterTheBarrier). Nil in the |
| + // agent. |
| + afterFrozenCheck func(host string) |
| declEtag map[[2]string]string // (acct, app) → last etag seen |
| declDoc map[[2]string]Declaration |
| @@ -94,10 +106,20 @@ func (c *Converger) Pass(ctx context.Context) error { |
| return err |
| } |
| now := c.Now() |
| + // alive is every row, held ones included, so PruneTimers never treats a |
| + // frozen host's rows as orphans. frozenAny holds back the table-wide |
| + // prune itself while a move has frozen any host here: its retention leg |
| + // would otherwise delete a frozen host's finished once-rows. The |
| + // runner's daily sweep still runs (it cannot be frozen; the timers |
| + // digest leaves that history out, joint N14), so nothing accumulates. |
| alive := map[string]bool{} |
| + frozenAny := false |
| byApp := map[[2]string][]registry.Route{} |
| for _, rt := range routes { |
| alive[rt.Host] = true |
| + if c.Frozen != nil && !frozenAny && c.Frozen(rt.Host) { |
| + frozenAny = true |
| + } |
| if rt.Kind != "instance" { |
| continue |
| } |
| @@ -113,7 +135,7 @@ func (c *Converger) Pass(ctx context.Context) error { |
| errs = append(errs, fmt.Errorf("%s/%s: %w", k[0], k[1], err)) |
| } |
| } |
| - if !c.HostWrites.Enabled { |
| + if !c.HostWrites.Enabled && !frozenAny { |
| if n, err := c.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil { |
| errs = append(errs, err) |
| } else if n > 0 { |
| @@ -209,7 +231,24 @@ func (c *Converger) convergeApp(ctx context.Context, acct, app string, rts []reg |
| // convergeHost returns whether anything about the host's rows changed (or |
| // the host is unsupported and should be reported so). |
| +// |
| +// A host a move has frozen (Frozen, moves spec §9.4 and §19 K9) is left |
| +// exactly as it is: nothing upserted, nothing pruned, a run signal not |
| +// fired. The check and every write below happen inside the host's gate, so |
| +// the cut's drain (HostGate.Drain after the hold) waits for a convergence |
| +// that checked before the freeze, and one that starts after the drain sees |
| +// it. A host the gate refuses is being drained: skipped the same way. |
| func (c *Converger) convergeHost(ctx context.Context, rt registry.Route, decl Declaration, fireName string, now time.Time) (bool, error) { |
| + if !c.Gate.tryEnter(rt.Host) { |
| + return false, nil |
| + } |
| + defer c.Gate.leave(rt.Host) |
| + if c.Frozen != nil && c.Frozen(rt.Host) { |
| + return false, nil |
| + } |
| + if c.afterFrozenCheck != nil { |
| + c.afterFrozenCheck(rt.Host) |
| + } |
| var before []registry.Timer |
| if c.HostWrites.Enabled { |
| err := withTimerWork(ctx, func() error { |
diff --git a/internal/schedule/converge_test.go b/internal/schedule/converge_test.go| index 95603d9e4..6c8bb2694 100644 |
| --- a/internal/schedule/converge_test.go |
| +++ b/internal/schedule/converge_test.go |
| @@ -606,3 +606,109 @@ func scheduleTimerRegistration(t *testing.T, reg *registry.Registry, host, name |
| } |
| return row.Registration |
| } |
| + |
| +// The converger and the cut's barrier: a convergeHost paused after its |
| +// Frozen check, inside the gate, holds the drain until it has finished |
| +// writing; a converger that starts after the drain sees Frozen and writes |
| +// nothing, so the timers snapshot taken after the drain equals the table. |
| +func TestConvergerCannotWriteAfterTheBarrier(t *testing.T) { |
| + reg, st := convergeFixture(t) |
| + ctx := context.Background() |
| + now := at("2026-08-23T13:07:00Z") |
| + const host = "a.bab.oncarlos.com" |
| + UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now) |
| + gate := &HostGate{} |
| + var frozen sync.Map |
| + c := NewConverger(reg, st, "box1", nil) |
| + c.Now = func() time.Time { return now } |
| + c.Gate = gate |
| + c.Frozen = func(h string) bool { _, ok := frozen.Load(h); return ok } |
| + reached, resume := make(chan struct{}), make(chan struct{}) |
| + var once sync.Once |
| + c.afterFrozenCheck = func(h string) { |
| + if h == host { |
| + once.Do(func() { close(reached); <-resume }) |
| + } |
| + } |
| + passed := make(chan error, 1) |
| + go func() { passed <- c.Pass(ctx) }() |
| + <-reached |
| + |
| + // The cut: hold, freeze, drain. The paused convergence holds the drain. |
| + if err := reg.SetHold(host, "moving", "mv_a"); err != nil { |
| + t.Fatal(err) |
| + } |
| + frozen.Store(host, true) |
| + drained := make(chan error, 1) |
| + go func() { drained <- gate.Drain(ctx, host) }() |
| + select { |
| + case err := <-drained: |
| + t.Fatalf("the drain returned (%v) while a convergence was inside the gate", err) |
| + case <-time.After(50 * time.Millisecond): |
| + } |
| + close(resume) |
| + if err := <-drained; err != nil { |
| + t.Fatal(err) |
| + } |
| + snap, err := reg.TimersFor(host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := <-passed; err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(snap) != 1 || snap[0].Name != "sync" { |
| + t.Fatalf("the paused convergence's write is not in the snapshot: %+v", snap) |
| + } |
| + |
| + // A later pass, with the declaration changed: frozen, it writes nothing. |
| + UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "nightly", Every: "24h", Path: "/jobs/nightly"}, now) |
| + if _, _, err := RemoveSchedule(ctx, st, "bab", "app", "sync", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := c.Pass(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + got, err := reg.TimersFor(host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(got) != len(snap) || got[0] != snap[0] { |
| + t.Fatalf("the table moved after the barrier:\n got %+v\nwant %+v", got, snap) |
| + } |
| + // The other host of the app, not frozen, converged as usual. |
| + if _, err := reg.GetTimer("b.bab.oncarlos.com", "nightly"); err != nil { |
| + t.Fatalf("an unfrozen host stopped converging: %v", err) |
| + } |
| +} |
| + |
| +// While any routed host is frozen, Pass leaves the table-wide prune alone, |
| +// so a frozen host's finished once-rows stay as the snapshot saw them; the |
| +// runner's own daily sweep is the one prune that is not frozen (joint N14). |
| +func TestConvergerSkipsTheTableWidePruneWhileAHostIsFrozen(t *testing.T) { |
| + reg, st := convergeFixture(t) |
| + ctx := context.Background() |
| + now := at("2026-08-23T13:07:00Z") |
| + const host = "a.bab.oncarlos.com" |
| + old := now.Add(-8 * 24 * time.Hour).Unix() |
| + reg.UpsertTimer(registry.Timer{Host: host, Name: "done", Kind: KindOnce, Spec: "at 1", Path: "/d", NextAt: 0, LastAt: old}) |
| + if _, err := reg.MarkTimerFired(host, "done", scheduleTimerRegistration(t, reg, host, "done"), old, "204", 1, 0, 0, 0); err != nil { |
| + t.Fatal(err) |
| + } |
| + c := NewConverger(reg, st, "box1", nil) |
| + c.Now = func() time.Time { return now } |
| + c.Frozen = func(h string) bool { return h == host } |
| + if err := c.Pass(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + if tm, _ := reg.GetTimer(host, "done"); tm == nil { |
| + t.Fatal("the frozen host's finished once-row was pruned") |
| + } |
| + c.Frozen = func(string) bool { return false } |
| + if err := c.Pass(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + if tm, _ := reg.GetTimer(host, "done"); tm != nil { |
| + t.Fatalf("the prune did not resume after the freeze: %+v", tm) |
| + } |
| +} |
diff --git a/internal/schedule/hostgate.go b/internal/schedule/hostgate.go| new file mode 100644 |
| index 000000000..9ed8cea0d |
| --- /dev/null |
| +++ b/internal/schedule/hostgate.go |
| @@ -0,0 +1,86 @@ |
| +package schedule |
| + |
| +import ( |
| + "context" |
| + "sync" |
| +) |
| + |
| +// HostGate is the one per-host gate shared by the three writers of a host's |
| +// timer rows (moves spec §9.4, round-2 finding 21): the runner holds a host's |
| +// read side for the whole of a delivery, the converger holds it across its |
| +// Frozen check and its writes, and a move's drain takes the write side once |
| +// the row is held. A drain therefore returns only when no delivery or |
| +// convergence that could have missed the hold is still running, and from |
| +// then on the hold check inside the gate refuses both. |
| +// |
| +// Entry is TryRLock, never RLock: Go's RWMutex blocks new readers once a |
| +// writer waits, and a drain can wait for a delivery of up to DeliveryTimeout. |
| +// A blocking entry would park the tick loop, and with it every other host's |
| +// timers, or the converger's pass, behind one held host. A refused entry is |
| +// a host being drained, which is a host about to be frozen: it is skipped |
| +// this tick or pass, with no bookkeeping. |
| +// |
| +// A nil *HostGate is inert (every entry succeeds, a drain returns at once): |
| +// the runner and converger with moves off are main's. |
| +type HostGate struct { |
| + mu sync.Mutex |
| + hosts map[string]*sync.RWMutex |
| +} |
| + |
| +func (g *HostGate) lock(host string) *sync.RWMutex { |
| + g.mu.Lock() |
| + defer g.mu.Unlock() |
| + if g.hosts == nil { |
| + g.hosts = map[string]*sync.RWMutex{} |
| + } |
| + l := g.hosts[host] |
| + if l == nil { |
| + l = &sync.RWMutex{} |
| + g.hosts[host] = l |
| + } |
| + return l |
| +} |
| + |
| +// tryEnter takes host's read side without waiting. false: a drain holds the |
| +// host or waits for it. |
| +func (g *HostGate) tryEnter(host string) bool { |
| + if g == nil { |
| + return true |
| + } |
| + return g.lock(host).TryRLock() |
| +} |
| + |
| +// leave ends an entry tryEnter granted. |
| +func (g *HostGate) leave(host string) { |
| + if g == nil { |
| + return |
| + } |
| + g.lock(host).RUnlock() |
| +} |
| + |
| +// Drain waits until every delivery and convergence in flight for host has |
| +// finished, then lets the host go: the cut's drain, called after SetHold |
| +// (moves spec §9.4 "drained"). Work that starts afterwards re-reads the row |
| +// or the freeze inside the gate and stops. ctx bounds the wait; an ended |
| +// context returns its error, and the pending drain still completes in the |
| +// background (until then the host's new work is refused, which is what a |
| +// held host wants anyway). |
| +func (g *HostGate) Drain(ctx context.Context, host string) error { |
| + if g == nil { |
| + return nil |
| + } |
| + l := g.lock(host) |
| + done := make(chan struct{}) |
| + go func() { |
| + // The barrier is the acquisition itself: nothing runs inside it. |
| + l.Lock() |
| + l.Unlock() |
| + close(done) |
| + }() |
| + select { |
| + case <-done: |
| + return nil |
| + case <-ctx.Done(): |
| + return ctx.Err() |
| + } |
| +} |
diff --git a/internal/schedule/hostgate_test.go b/internal/schedule/hostgate_test.go| new file mode 100644 |
| index 000000000..58e141783 |
| --- /dev/null |
| +++ b/internal/schedule/hostgate_test.go |
| @@ -0,0 +1,74 @@ |
| +package schedule |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// A nil gate is main's runner and converger: every host enters, and a drain |
| +// has nothing to wait for. |
| +func TestNilHostGateIsInert(t *testing.T) { |
| + var g *HostGate |
| + if !g.tryEnter("a") { |
| + t.Fatal("a nil gate refused a host") |
| + } |
| + g.leave("a") |
| + if err := g.Drain(context.Background(), "a"); err != nil { |
| + t.Fatalf("a nil gate's drain = %v", err) |
| + } |
| +} |
| + |
| +// The drain waits for the host's in-flight work, refuses new work while it |
| +// waits, and leaves the host open once it returns. Other hosts are never |
| +// held up. |
| +func TestHostGateDrainWaitsForInFlightWork(t *testing.T) { |
| + g := &HostGate{} |
| + if !g.tryEnter("a") { |
| + t.Fatal("first entry refused") |
| + } |
| + done := make(chan error, 1) |
| + go func() { done <- g.Drain(context.Background(), "a") }() |
| + select { |
| + case err := <-done: |
| + t.Fatalf("the drain returned (%v) while work on the host was in flight", err) |
| + case <-time.After(50 * time.Millisecond): |
| + } |
| + // A drain is waiting: new work on the host is refused, other hosts enter. |
| + deadline := time.Now().Add(time.Second) |
| + for g.tryEnter("a") { |
| + g.leave("a") |
| + if time.Now().After(deadline) { |
| + t.Fatal("work entered the host while a drain was waiting") |
| + } |
| + time.Sleep(time.Millisecond) |
| + } |
| + if !g.tryEnter("b") { |
| + t.Fatal("a drain on one host refused another host") |
| + } |
| + g.leave("b") |
| + g.leave("a") |
| + if err := <-done; err != nil { |
| + t.Fatalf("drain = %v", err) |
| + } |
| + if !g.tryEnter("a") { |
| + t.Fatal("the host stayed closed after the drain") |
| + } |
| + g.leave("a") |
| +} |
| + |
| +// A drain whose context ends returns the context's error; the host's work |
| +// is unaffected. |
| +func TestHostGateDrainHonoursItsContext(t *testing.T) { |
| + g := &HostGate{} |
| + if !g.tryEnter("a") { |
| + t.Fatal("entry refused") |
| + } |
| + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) |
| + defer cancel() |
| + if err := g.Drain(ctx, "a"); !errors.Is(err, context.DeadlineExceeded) { |
| + t.Fatalf("drain = %v, want the deadline", err) |
| + } |
| + g.leave("a") |
| +} |
diff --git a/internal/schedule/runner.go b/internal/schedule/runner.go| index 23b3594d1..e6631c955 100644 |
| --- a/internal/schedule/runner.go |
| +++ b/internal/schedule/runner.go |
| @@ -62,6 +62,14 @@ type Runner struct { |
| Build string |
| Idle time.Duration // the activator's idle window; zero when no hibernation is armed on this box |
| HostWrites hostwrite.Config |
| + // Gate is the per-host gate shared with the converger and a move's |
| + // drain (HostGate): claim enters it for the whole delivery. Nil: no |
| + // gate, main's runner. |
| + Gate *HostGate |
| + |
| + // beforeClaim is the tests' pause between a tick's route read and its |
| + // claim (TestDrainRaceCannotAdvanceBookkeeping). Nil in the agent. |
| + beforeClaim func(host string) |
| // Per-host work is DETACHED from the tick: Tick launches one goroutine |
| // per host with due rows and returns without waiting for any of them. |
| @@ -164,8 +172,37 @@ func (r *Runner) Tick(ctx context.Context) { |
| // Converge prunes these; until it does, they are not ours to fire. |
| continue |
| } |
| + // A held row (moves spec §5.2, §9.4 "frozen") is skipped with no |
| + // bookkeeping at all: before this, its delivery was refused by the |
| + // activator and recorded as a retry, so a held row's attempts |
| + // advanced while its instance was moving. This read is only the |
| + // cheap pre-filter; the decision is the read inside the claim below. |
| + if registry.IsHold(rt.Status) { |
| + continue |
| + } |
| + if r.beforeClaim != nil { |
| + r.beforeClaim(host) |
| + } |
| if !r.claim(host) { |
| - continue // a previous tick's delivery to this host is still running |
| + continue // a previous tick's delivery to this host is still running, or a move is draining it |
| + } |
| + // The decision read, inside the claim. A move holds the row and then |
| + // drains the host (HostGate.Drain), and the drain is sound only if |
| + // every tick that looks at the row after the hold does so while it |
| + // holds the claim: a tick that read the row ready above and was |
| + // descheduled before claim must not slip past the drain and record |
| + // an attempt. |
| + err = withTimerWork(hctx, func() error { |
| + var readErr error |
| + rt, readErr = r.Reg.Get(host) |
| + return readErr |
| + }) |
| + if err != nil || rt == nil || registry.IsHold(rt.Status) { |
| + if err != nil { |
| + log.Printf("schedule: %s: registry read inside the claim: %v. skipping this tick", host, err) |
| + } |
| + r.release(host) |
| + continue |
| } |
| // The slot is taken HERE, before the goroutine exists, so the bound |
| // is on hosts in flight and never on goroutines waiting to be. Hosts |
| @@ -182,7 +219,13 @@ func (r *Runner) Tick(ctx context.Context) { |
| defer r.inflight.Done() |
| defer func() { <-r.slots() }() |
| defer r.release(host) |
| - for _, t := range timers { |
| + for i, t := range timers { |
| + // A hold written while this host's earlier timers were |
| + // delivering stops the rest: each later timer re-reads the |
| + // row first. The drain waits for this goroutine either way. |
| + if i > 0 && r.held(ctx, host) { |
| + break |
| + } |
| r.fire(ctx, rt, t) |
| } |
| if r.Report != nil { |
| @@ -336,13 +379,17 @@ type SocketRetainer interface { |
| RetainSockets(live map[string]bool) |
| } |
| -// claim marks host in flight; false if it already was. |
| +// claim marks host in flight and enters its gate; false if it already was in |
| +// flight, or a move's drain holds or waits for the host (HostGate). |
| func (r *Runner) claim(host string) bool { |
| r.mu.Lock() |
| defer r.mu.Unlock() |
| if r.busy[host] { |
| return false |
| } |
| + if !r.Gate.tryEnter(host) { |
| + return false |
| + } |
| if r.busy == nil { |
| r.busy = map[string]bool{} |
| } |
| @@ -350,12 +397,34 @@ func (r *Runner) claim(host string) bool { |
| return true |
| } |
| +// Busy reports whether a delivery to host is in flight: the per-host claim |
| +// (moves spec §9.4 "drained"). A snapshot: the cut's drain is |
| +// Gate.Drain, which also waits for a claim taken after this answer. |
| +func (r *Runner) Busy(host string) bool { |
| + r.mu.Lock() |
| + defer r.mu.Unlock() |
| + return r.busy[host] |
| +} |
| + |
| func (r *Runner) release(host string) { |
| r.mu.Lock() |
| delete(r.busy, host) |
| + r.Gate.leave(host) |
| r.mu.Unlock() |
| } |
| +// held reports whether host's row is held by an operation now, or cannot be |
| +// read: either way the rest of the host's timers wait for the next tick. |
| +func (r *Runner) held(ctx context.Context, host string) bool { |
| + var rt *registry.Route |
| + err := withTimerWork(ctx, func() error { |
| + var readErr error |
| + rt, readErr = r.Reg.Get(host) |
| + return readErr |
| + }) |
| + return err != nil || rt == nil || registry.IsHold(rt.Status) |
| +} |
| + |
| // slots is the cross-tick semaphore, made on first use so a zero Runner |
| // literal (every test, and the agent's) needs no constructor. |
| func (r *Runner) slots() chan struct{} { |
diff --git a/internal/schedule/runner_test.go b/internal/schedule/runner_test.go| index 3602c996f..e4cfcac63 100644 |
| --- a/internal/schedule/runner_test.go |
| +++ b/internal/schedule/runner_test.go |
| @@ -14,6 +14,7 @@ import ( |
| "time" |
| "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| ) |
| type fakeDeliverer struct { |
| @@ -976,3 +977,213 @@ func TestNonpositiveOccurrenceDeliveredAndRetriedWithOriginalTimestamp(t *testin |
| }) |
| } |
| } |
| + |
| +// ---- move holds (moves spec §9.4) ---- |
| + |
| +// holdHost is the runner fixture's host, held by a move in these tests. |
| +const holdHost = "a.bab.oncarlos.com" |
| + |
| +// timersOf is the host's whole timers table, for before/after comparisons. |
| +func timersOf(t *testing.T, reg *registry.Registry, host string) []registry.Timer { |
| + t.Helper() |
| + rows, err := reg.TimersFor(host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return rows |
| +} |
| + |
| +func sameTimers(a, b []registry.Timer) bool { |
| + if len(a) != len(b) { |
| + return false |
| + } |
| + for i := range a { |
| + if a[i] != b[i] { |
| + return false |
| + } |
| + } |
| + return true |
| +} |
| + |
| +// Witness wTimerFreeze: a held row's due timers are not delivered and their |
| +// bookkeeping does not move (before this, a refused delivery advanced |
| +// attempts); a frozen host's declared schedule is neither upserted nor |
| +// pruned by Converger.Pass; once the hold lifts, both resume. |
| +func TestHeldRowsFreezeTimers(t *testing.T) { |
| + d := &fakeDeliverer{status: 503} // a delivery would retry and advance attempts |
| + reg, r := runnerFixture(t, d) |
| + now := at("2026-08-23T13:07:00Z") |
| + r.Now = func() time.Time { return now } |
| + r.Gate = &HostGate{} |
| + reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1, DueAt: 1}) |
| + reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "stale", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/s", NextAt: now.Unix() - 5}) |
| + if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil { |
| + t.Fatal(err) |
| + } |
| + before := timersOf(t, reg, holdHost) |
| + r.tick() |
| + if len(d.calls) != 0 { |
| + t.Fatalf("a held row's timers were delivered %d time(s)", len(d.calls)) |
| + } |
| + if got := timersOf(t, reg, holdHost); !sameTimers(got, before) { |
| + t.Fatalf("a held row's bookkeeping moved:\n got %+v\nwant %+v", got, before) |
| + } |
| + if r.Busy(holdHost) { |
| + t.Fatal("the runner kept a claim on a held host") |
| + } |
| + |
| + // The converger, with the host frozen: the declaration is not upserted |
| + // and the undeclared "stale" schedule is not pruned. |
| + st := release.NewDirStore(t.TempDir()) |
| + ctx := context.Background() |
| + UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now) |
| + frozen := true |
| + c := NewConverger(reg, st, "box1", nil) |
| + c.Now = func() time.Time { return now } |
| + c.Gate = r.Gate |
| + c.Frozen = func(host string) bool { return host == holdHost && frozen } |
| + if err := c.Pass(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := timersOf(t, reg, holdHost); !sameTimers(got, before) { |
| + t.Fatalf("Converger.Pass wrote a frozen host's timers:\n got %+v\nwant %+v", got, before) |
| + } |
| + |
| + // The hold lifts: the runner delivers and the converger converges. |
| + if ok, err := reg.ReleaseHold(holdHost, "mv_a"); err != nil || !ok { |
| + t.Fatalf("release hold = %v %v", ok, err) |
| + } |
| + frozen = false |
| + r.tick() |
| + if len(d.calls) == 0 { |
| + t.Fatal("nothing was delivered after the hold lifted") |
| + } |
| + if err := c.Pass(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := reg.GetTimer(holdHost, "sync"); err != nil { |
| + t.Fatalf("the declaration did not converge after the freeze: %v", err) |
| + } |
| + if tm, _ := reg.GetTimer(holdHost, "stale"); tm != nil { |
| + t.Fatalf("the undeclared schedule survived the converge after the freeze: %+v", tm) |
| + } |
| +} |
| + |
| +// A hold written while a host's goroutine is delivering stops the rest of |
| +// that host's due timers: the next timer re-reads the row first. |
| +func TestHoldMidDeliveryStopsTheRest(t *testing.T) { |
| + reg, r := runnerFixture(t, nil) |
| + now := at("2026-08-23T13:07:00Z") |
| + r.Now = func() time.Time { return now } |
| + var delivered []string |
| + r.Deliver = deliverFunc(func(rt registry.Route, req *http.Request) (int, error) { |
| + delivered = append(delivered, req.Header.Get("X-Carlos-Schedule")) |
| + if err := reg.SetHold(rt.Host, "moving", "mv_a"); err != nil { |
| + t.Error(err) |
| + } |
| + return 204, nil |
| + }) |
| + reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "first", Kind: KindOnce, Spec: "at 1", Path: "/1", NextAt: 1}) |
| + reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "second", Kind: KindOnce, Spec: "at 2", Path: "/2", NextAt: 2}) |
| + r.tick() |
| + if len(delivered) != 1 || delivered[0] != "first" { |
| + t.Fatalf("delivered %v, want only the first", delivered) |
| + } |
| + second, _ := reg.GetTimer(holdHost, "second") |
| + if second.NextAt != 2 || second.Attempts != 0 || second.LastAt != 0 { |
| + t.Fatalf("the second timer's bookkeeping moved under the hold: %+v", second) |
| + } |
| +} |
| + |
| +type deliverFunc func(rt registry.Route, req *http.Request) (int, error) |
| + |
| +func (f deliverFunc) Deliver(_ context.Context, rt registry.Route, req *http.Request) (int, error) { |
| + return f(rt, req) |
| +} |
| + |
| +// The cut's drain race: a tick reads the route while it is ready and pauses |
| +// before its claim; the cut sets the hold and finds the host not busy, and |
| +// its drain returns; the tick resumes, claims, re-reads the row inside the |
| +// claim, sees the hold and lets go. Nothing is delivered and attempts does |
| +// not move. |
| +func TestDrainRaceCannotAdvanceBookkeeping(t *testing.T) { |
| + d := &fakeDeliverer{status: 503} |
| + reg, r := runnerFixture(t, d) |
| + now := at("2026-08-23T13:07:00Z") |
| + r.Now = func() time.Time { return now } |
| + r.Gate = &HostGate{} |
| + reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1}) |
| + before := timersOf(t, reg, holdHost) |
| + reached, resume := make(chan struct{}), make(chan struct{}) |
| + r.beforeClaim = func(host string) { |
| + if host == holdHost { |
| + close(reached) |
| + <-resume |
| + } |
| + } |
| + ticked := make(chan struct{}) |
| + go func() { r.Tick(context.Background()); close(ticked) }() |
| + <-reached |
| + // The cut: hold, then drain. |
| + if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if r.Busy(holdHost) { |
| + t.Fatal("the paused tick already held the claim") |
| + } |
| + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer cancel() |
| + if err := r.Gate.Drain(ctx, holdHost); err != nil { |
| + t.Fatalf("drain = %v", err) |
| + } |
| + close(resume) |
| + <-ticked |
| + r.wait() |
| + if len(d.calls) != 0 { |
| + t.Fatalf("the paused tick delivered %d time(s) after the drain", len(d.calls)) |
| + } |
| + if got := timersOf(t, reg, holdHost); !sameTimers(got, before) { |
| + t.Fatalf("the paused tick moved bookkeeping after the drain:\n got %+v\nwant %+v", got, before) |
| + } |
| + if r.Busy(holdHost) { |
| + t.Fatal("the tick kept its claim after seeing the hold") |
| + } |
| +} |
| + |
| +// A delivery in flight when the hold is written holds up the drain until it |
| +// has recorded its outcome: the snapshot after the drain sees the |
| +// delivery's bookkeeping, never a write that lands after it. |
| +func TestDrainWaitsForTheDeliveryInFlight(t *testing.T) { |
| + d := &blockingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, slow: holdHost, gate: make(chan struct{})} |
| + reg, r := runnerFixture(t, d) |
| + now := at("2026-08-23T13:07:00Z") |
| + r.Now = func() time.Time { return now } |
| + r.Gate = &HostGate{} |
| + reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1}) |
| + r.Tick(context.Background()) |
| + if !r.Busy(holdHost) { |
| + t.Fatal("the delivery is not in flight") |
| + } |
| + if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil { |
| + t.Fatal(err) |
| + } |
| + drained := make(chan error, 1) |
| + go func() { drained <- r.Gate.Drain(context.Background(), holdHost) }() |
| + select { |
| + case err := <-drained: |
| + t.Fatalf("the drain returned (%v) with a delivery in flight", err) |
| + case <-time.After(50 * time.Millisecond): |
| + } |
| + close(d.gate) |
| + if err := <-drained; err != nil { |
| + t.Fatal(err) |
| + } |
| + snap := timersOf(t, reg, holdHost) |
| + r.wait() |
| + if snap[0].LastStatus != "204" || snap[0].NextAt != 0 { |
| + t.Fatalf("the drain returned before the delivery recorded: %+v", snap[0]) |
| + } |
| + if got := timersOf(t, reg, holdHost); !sameTimers(got, snap) { |
| + t.Fatalf("the table changed after the drain:\n got %+v\nwant %+v", got, snap) |
| + } |
| +} |