amadan
move holds, fix round 1: held hosts leave the due batch, hold writes are conditional and serialised, switch-off is main's code path (task-6-review)
schedule: with the hold wiring armed the runner asks DueUnheldTimers, so a held host's rows are excluded before the LIMIT and cannot starve another host (review Important 1). Every hold check in Tick (the pre-filter, the decision read inside the claim, the between-timer check) runs only with Gate set; nil is main's runner exactly (Important 3). The converger was already main's with nil hooks; a test now says so. registry: DueUnheldTimers (DueTimers minus held routes, in SQL) and SetHoldIf, a hold write conditional on the row's current hold_op in one statement, expecting "" or the same op. provision: holdExisting's record re-read, re-judgement and write run under WithHostHoldLock, the per-host lock the landing's open and finalisation (Task 10) must take, and the write is SetHoldIf expecting the row unheld (Important 2). A write another op pre-empted is logged and left. Tests: a full batch of held timers ahead of an unrelated due timer fires the unrelated one this tick; a nil-gate runner delivers a held row's two timers and advances attempts; the host switch-off test is behavioural (runner and converger wired with the switch off); a pause before the hold write while another op takes the row leaves that op's hold; an open racing the write waits for the lock and then releases the hold it finds under its own op; the lock is exclusive per host; six in-place mutants each fail their test. The wiring in moveshold.go is accepted (fix1 ruling 4).
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
11 files changed,
+605
−51
internal/host/moveshold.go+5 −0internal/host/moveshold_test.go+62 −2internal/provision/reconcile.go+91 −30internal/provision/reconcile_test.go+129 −0internal/registry/hold_test.go+56 −0internal/registry/registry.go+27 −0internal/registry/timers.go+27 −0internal/registry/timers_test.go+59 −0internal/schedule/converge_test.go+35 −0internal/schedule/runner.go+43 −19internal/schedule/runner_test.go+71 −0
diff --git a/internal/host/moveshold.go b/internal/host/moveshold.go| index 375c0037d..9128e77ea 100644 |
| --- a/internal/host/moveshold.go |
| +++ b/internal/host/moveshold.go |
| @@ -29,6 +29,11 @@ import ( |
| // |
| // Only built with the Phase 1 switch on: off, armedHoldView is nil and no |
| // pass is given a hook. |
| +// |
| +// Ending a hold that hold would re-assert (the landing's open, the |
| +// finalisation: Task 10, movesland.go) must happen under |
| +// provision.WithHostHoldLock, the lock reconcile's hold decision holds from |
| +// its record re-read through its conditional write. See that function. |
| type movesHoldView struct { |
| box, region string |
| read func(ctx context.Context, s release.Store, host string) (hostclaim.Observed, error) |
diff --git a/internal/host/moveshold_test.go b/internal/host/moveshold_test.go| index 3bc892d4a..f1bbe8f44 100644 |
| --- a/internal/host/moveshold_test.go |
| +++ b/internal/host/moveshold_test.go |
| @@ -3,15 +3,25 @@ package host |
| import ( |
| "context" |
| "errors" |
| + "net/http" |
| + "path/filepath" |
| "testing" |
| "time" |
| "github.com/carlosframework/platform/internal/hostclaim" |
| "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| "github.com/carlosframework/platform/internal/schedule" |
| ) |
| +// statusDeliverer is a schedule.Deliverer answering one status to every tick. |
| +type statusDeliverer int |
| + |
| +func (s statusDeliverer) Deliver(context.Context, registry.Route, *http.Request) (int, error) { |
| + return int(s), nil |
| +} |
| + |
| const ( |
| mhHost = "a.go.example.com" |
| mhID = "mv_abcdefghijklmnopqrstuvwxyz" |
| @@ -241,11 +251,61 @@ func TestMovesHoldsAreMainWithTheSwitchOff(t *testing.T) { |
| 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{}} |
| + |
| + // The schedule halves, wired as the agent wires them with the switch |
| + // off, against a row whose status is a hold: the runner delivers its |
| + // due timer and records the outcome, and the converger converges its |
| + // declaration, exactly as main does (fix round 1: behaviour, not the |
| + // absence of hooks). |
| + ctx := context.Background() |
| + reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { reg.Close() }) |
| + const host = "held.bab.oncarlos.com" |
| + if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.UpsertTimer(registry.Timer{Host: host, Name: "remind", Kind: schedule.KindOnce, Spec: "at 1", Path: "/r", NextAt: 1, DueAt: 1}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.SetHold(host, "moving", mhID); err != nil { |
| + t.Fatal(err) |
| + } |
| + now := time.Date(2026, 8, 23, 13, 7, 0, 0, time.UTC) |
| + st := release.NewDirStore(t.TempDir()) |
| + if _, err := schedule.UpsertSchedule(ctx, st, "bab", "app", schedule.Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now); err != nil { |
| + t.Fatal(err) |
| + } |
| + runner := &schedule.Runner{Reg: reg, Deliver: statusDeliverer(503), Now: func() time.Time { return now }, Timeout: time.Minute, |
| + Box: "ie-box", Build: "test", Token: func(string) (string, error) { return "tok", nil }} |
| + conv := schedule.NewConverger(reg, st, "ie-box", nil) |
| + conv.Now = func() time.Time { return now } |
| + sd := &scheduleDeps{runner: runner, conv: conv} |
| armScheduleHolds(sd, armedHoldView()) |
| - if sd.runner.Gate != nil || sd.conv.Gate != nil || sd.conv.Frozen != nil { |
| + if runner.Gate != nil || conv.Gate != nil || conv.Frozen != nil { |
| t.Fatal("schedule hooks with the switch off") |
| } |
| + runner.Tick(ctx) |
| + // Tick launches the delivery and returns: wait for the record, then |
| + // for the host's goroutine to let go. |
| + var tm *registry.Timer |
| + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(time.Millisecond) { |
| + if tm, _ = reg.GetTimer(host, "remind"); tm != nil && tm.Attempts == 1 && !runner.Busy(host) { |
| + break |
| + } |
| + } |
| + if tm == nil || tm.Attempts != 1 || tm.LastStatus != "503" { |
| + t.Fatalf("with the switch off a held row's due timer was not delivered and recorded as main does: %+v", tm) |
| + } |
| + if err := conv.Pass(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := reg.GetTimer(host, "sync"); err != nil { |
| + t.Fatalf("with the switch off the held host's declaration did not converge: %v", err) |
| + } |
| + |
| armedMovesWorker.Store(nil) |
| if cfg := withMovesHolds(provision.ReconcileConfig{}); cfg.Hold != nil { |
| t.Fatal("reconcile hooks with no worker armed") |
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go| index 96300a8ca..d7a4b696a 100644 |
| --- a/internal/provision/reconcile.go |
| +++ b/internal/provision/reconcile.go |
| @@ -104,9 +104,10 @@ type ReconcileConfig struct { |
| // 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 |
| + // with SetHoldIf (Upsert never changes a hold on conflict, G10) once a |
| + // fresh read of the record, under the host's hold lock |
| + // (WithHostHoldLock), 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) |
| @@ -116,10 +117,15 @@ type ReconcileConfig struct { |
| // 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. |
| + // or a SetHoldIf that landed), 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) |
| + |
| + // beforeHoldWrite is the tests' pause under the host's hold lock, just |
| + // before holdExisting's conditional write (TestReconcileHoldWrite...). |
| + // Nil in the agent. |
| + beforeHoldWrite func(host string) |
| } |
| // InboundAny and InboundNone are the closed vocabulary ReconcileConfig.Inbound |
| @@ -1423,13 +1429,51 @@ func mintHeld(ctx context.Context, reg *registry.Registry, cfg ReconcileConfig, |
| } |
| } |
| +// hostHoldLocks is one in-process mutex per host for the hold decision on |
| +// an existing row: see WithHostHoldLock. |
| +var hostHoldLocks struct { |
| + mu sync.Mutex |
| + hosts map[string]*sync.Mutex |
| +} |
| + |
| +// WithHostHoldLock runs fn holding host's hold lock, exclusively within this |
| +// process. Reconcile's hold decision for an existing row (holdExisting: the |
| +// record re-read, the hold asked again of it, the conditional write) runs |
| +// under it, so nothing that ends the hold can run between that read and |
| +// that write. The landing's open and its finalisation (moves Task 10, |
| +// movesland.go) MUST take this same lock around their record read and their |
| +// ReleaseHold or DeleteIfHold, and so must any later writer that ends a hold |
| +// reconcile could re-assert from the record: without it, an open that lands |
| +// in the window is re-held on the next line. Writers in other processes |
| +// (root) are met by SetHoldIf's own condition, not by this lock. |
| +// |
| +// A leaf: its holder reads one record, writes one row and takes no other |
| +// lock. Nothing takes it while holding the held-stage lock (G30), and |
| +// nothing holds it across a wait on another host. |
| +func WithHostHoldLock(host string, fn func()) { |
| + hostHoldLocks.mu.Lock() |
| + if hostHoldLocks.hosts == nil { |
| + hostHoldLocks.hosts = map[string]*sync.Mutex{} |
| + } |
| + l := hostHoldLocks.hosts[host] |
| + if l == nil { |
| + l = &sync.Mutex{} |
| + hostHoldLocks.hosts[host] = l |
| + } |
| + hostHoldLocks.mu.Unlock() |
| + l.Lock() |
| + defer l.Unlock() |
| + fn() |
| +} |
| + |
| // 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 |
| +// SetHoldIf. 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: |
| +// Three rules keep a hold write from closing a row an operation has opened, |
| +// or from landing over a hold written since the pass's listing (fix round 1): |
| // |
| // - a recovery hold (restoring, releasing) is never written over an |
| // unheld row. The recovering operation holds its own row as it enters |
| @@ -1437,12 +1481,19 @@ func mintHeld(ctx context.Context, reg *registry.Registry, cfg ReconcileConfig, |
| // 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. |
| +// asked again of the fresh record, all under the host's hold lock |
| +// (WithHostHoldLock), which the open and the finalisation take too: |
| +// 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, and under the lock it cannot be |
| +// opened before the write lands. |
| +// - the write is SetHoldIf expecting the row unheld: a hold another op |
| +// wrote since the listing (or since this pass's own route read) is |
| +// never overwritten; the row is left to it and the next pass judges |
| +// it afresh. |
| // |
| -// SetHold can wait out root's activation guard and fail (G30): logged, and |
| -// the next pass tries again. |
| +// SetHoldIf 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 == "": |
| @@ -1462,22 +1513,32 @@ func holdExisting(ctx context.Context, s release.Store, reg *registry.Registry, |
| 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) |
| - } |
| + WithHostHoldLock(rec.Host, func() { |
| + 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 cfg.beforeHoldWrite != nil { |
| + cfg.beforeHoldWrite(rec.Host) |
| + } |
| + held, err := reg.SetHoldIf(rec.Host, status, op, "") |
| + if err != nil { |
| + log.Printf("provision: hold %s %s by %s: %v", rec.Host, status, op, err) |
| + return |
| + } |
| + if !held { |
| + log.Printf("provision: %s: not held %s by %s: the row's hold changed under this pass; the next pass judges it afresh", rec.Host, status, op) |
| + 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 |
diff --git a/internal/provision/reconcile_test.go b/internal/provision/reconcile_test.go| index 6a4ff8e4d..03e5ef9bb 100644 |
| --- a/internal/provision/reconcile_test.go |
| +++ b/internal/provision/reconcile_test.go |
| @@ -3353,3 +3353,132 @@ func TestReconcileHoldRereadsTheRecordBeforeHoldingARow(t *testing.T) { |
| t.Fatalf("an opened row was held again from a stale listing: %+v %v", got, wrote.hosts) |
| } |
| } |
| + |
| +// Fix round 1 (task-6-review, Important 2): the hold write on an existing |
| +// row is SetHoldIf, conditional on the row still being unheld in the same |
| +// statement. Another op that takes the row after the pass's listing (here |
| +// while the pass is paused just before its write) keeps it: the stale write |
| +// does not land and nothing is reported. |
| +func TestReconcileHoldWriteIsRefusedWhenAnotherOpTookTheRow(t *testing.T) { |
| + s, sqid := setup(t) |
| + rec := movingRecord(sqid) |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + reg := regWith(t, routeFor(rec, "/data", `["seed"]`, false)) // unheld at the listing |
| + 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 }} |
| + paused := 0 |
| + cfg.beforeHoldWrite = func(host string) { |
| + paused++ |
| + // A rollback's hold, from a writer outside this pass's lock (root, |
| + // or another process): the row is another op's by the time the |
| + // pass writes. |
| + if err := reg.SetHold(host, "restoring", "rb_other"); err != nil { |
| + t.Error(err) |
| + } |
| + } |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if paused != 1 { |
| + t.Fatalf("the pass paused %d time(s) before a hold write, want once", paused) |
| + } |
| + if got := mustGet(t, reg, rec.Host); got.Status != "restoring" || got.HoldOp != "rb_other" { |
| + t.Fatalf("the stale hold write landed over another op's: %+v", got) |
| + } |
| + if len(wrote.hosts) != 0 { |
| + t.Fatalf("a refused hold write was reported: %v", wrote.hosts) |
| + } |
| +} |
| + |
| +// Fix round 1 (task-6-review, Important 2): the hold decision, from the |
| +// record re-read through the write, runs under the host's hold lock, the |
| +// one the landing's open and finalisation take (WithHostHoldLock). An open |
| +// that arrives inside that window waits for the write, then finds the hold |
| +// under its own op and releases it; it can never run between the re-read |
| +// and the write, so the pass cannot re-hold a row the move has opened. A |
| +// later pass sees the committed record and leaves the opened row alone. |
| +func TestReconcileHoldWriteIsSerialisedWithTheMoveOpening(t *testing.T) { |
| + s, sqid := setup(t) |
| + rec := movingRecord(sqid) |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + reg := regWith(t, routeFor(rec, "/data", `["seed"]`, false)) |
| + 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 }} |
| + opened := make(chan struct{}) |
| + cfg.beforeHoldWrite = func(host string) { |
| + // The target's open, racing the write: the commit (record CAS) and |
| + // the conditional release, under the same lock. |
| + go func() { |
| + defer close(opened) |
| + WithHostHoldLock(host, func() { |
| + fresh, etag, err := GetRecordETag(ctx, s, rec.Account, rec.App, rec.Host) |
| + if err != nil { |
| + t.Error(err) |
| + return |
| + } |
| + m := *fresh.Move |
| + m.State = MoveCommitted |
| + fresh.Move = &m |
| + if _, err := s.Put(ctx, RecordKey(rec.Account, rec.App, rec.Host), mustJSON(t, fresh), etag); err != nil { |
| + t.Error(err) |
| + } |
| + if ok, err := reg.ReleaseHold(host, rec.Move.ID); err != nil || !ok { |
| + t.Errorf("the open found no hold under its own op to release: %v %v", ok, err) |
| + } |
| + }) |
| + }() |
| + select { |
| + case <-opened: |
| + t.Error("the open ran inside the pass's hold decision") |
| + case <-time.After(100 * time.Millisecond): |
| + } |
| + } |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + select { |
| + case <-opened: |
| + case <-time.After(10 * time.Second): |
| + t.Fatal("the open never got the lock after the pass") |
| + } |
| + if got := mustGet(t, reg, rec.Host); got.Status != "ready" || got.HoldOp != "" { |
| + t.Fatalf("after the open the row is %+v, want ready and unheld", got) |
| + } |
| + if len(wrote.hosts) != 1 { |
| + t.Fatalf("hold writes reported = %v, want the pass's one write before the open", wrote.hosts) |
| + } |
| + cfg.beforeHoldWrite = func(string) { t.Error("a later pass tried to hold the opened row") } |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := mustGet(t, reg, rec.Host); got.Status != "ready" || got.HoldOp != "" || len(wrote.hosts) != 1 { |
| + t.Fatalf("a later pass re-held the opened row: %+v %v", got, wrote.hosts) |
| + } |
| +} |
| + |
| +// WithHostHoldLock is exclusive per host and independent across hosts. |
| +func TestWithHostHoldLockIsExclusivePerHost(t *testing.T) { |
| + entered, release := make(chan struct{}), make(chan struct{}) |
| + go WithHostHoldLock("a.go.tito.io", func() { close(entered); <-release }) |
| + <-entered |
| + WithHostHoldLock("b.go.tito.io", func() {}) // another host's lock is free |
| + second := make(chan struct{}) |
| + go func() { WithHostHoldLock("a.go.tito.io", func() {}); close(second) }() |
| + select { |
| + case <-second: |
| + t.Fatal("a second holder entered the host's lock") |
| + case <-time.After(50 * time.Millisecond): |
| + } |
| + close(release) |
| + select { |
| + case <-second: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("the lock was not handed on") |
| + } |
| +} |
diff --git a/internal/registry/hold_test.go b/internal/registry/hold_test.go| index ddce56e60..34d016f56 100644 |
| --- a/internal/registry/hold_test.go |
| +++ b/internal/registry/hold_test.go |
| @@ -412,3 +412,59 @@ func TestWithHoldsFrozenHoldsOffASetHold(t *testing.T) { |
| t.Fatalf("err = %v", err) |
| } |
| } |
| + |
| +// Moves Task 6 fix round 1: SetHoldIf writes a hold only while the row's |
| +// hold_op is what the caller expects, in one statement, so a hold decided |
| +// from an earlier read cannot land on a row another op has taken since. |
| +// expectOp is "" (the row was unheld) or op (the op changes its own hold's |
| +// status); taking over another op's hold is never one statement. |
| +func TestSetHoldIfIsConditionalOnTheCurrentOp(t *testing.T) { |
| + r := holdFixture(t, "h") |
| + if ok, err := r.SetHoldIf("h", "moving", holdOpA, ""); err != nil || !ok { |
| + t.Fatalf("unheld row, expecting unheld: %v %v", ok, err) |
| + } |
| + if rt, _ := r.Get("h"); rt.Status != "moving" || rt.HoldOp != holdOpA { |
| + t.Fatalf("after the first hold: %+v", rt) |
| + } |
| + // Held by A now; a writer that still thinks the row is unheld is refused |
| + // and changes nothing. |
| + if ok, err := r.SetHoldIf("h", "landing", holdOpB, ""); err != nil || ok { |
| + t.Fatalf("stale write over A's hold: %v %v", ok, err) |
| + } |
| + if rt, _ := r.Get("h"); rt.Status != "moving" || rt.HoldOp != holdOpA { |
| + t.Fatalf("the refused write changed the row: %+v", rt) |
| + } |
| + // A may change its own hold's status. |
| + if ok, err := r.SetHoldIf("h", "landing", holdOpA, holdOpA); err != nil || !ok { |
| + t.Fatalf("A changing its own hold: %v %v", ok, err) |
| + } |
| + if rt, _ := r.Get("h"); rt.Status != "landing" || rt.HoldOp != holdOpA { |
| + t.Fatalf("after A's status change: %+v", rt) |
| + } |
| + // B expecting A's hold: not a takeover this statement allows. |
| + if _, err := r.SetHoldIf("h", "moving", holdOpB, holdOpA); err == nil { |
| + t.Fatal("SetHoldIf accepted expectOp of another op") |
| + } |
| + if rt, _ := r.Get("h"); rt.HoldOp != holdOpA { |
| + t.Fatalf("the refused takeover changed the row: %+v", rt) |
| + } |
| + // SetHold's own refusals hold. |
| + if _, err := r.SetHoldIf("h", "ready", holdOpA, holdOpA); err == nil { |
| + t.Fatal("ready accepted as a hold") |
| + } |
| + if _, err := r.SetHoldIf("h", "moving", "", ""); err == nil { |
| + t.Fatal("a hold with no op was accepted") |
| + } |
| + // No such row: nothing matched, not an error (the caller re-judges |
| + // next pass either way). |
| + if ok, err := r.SetHoldIf("nope", "moving", holdOpA, ""); err != nil || ok { |
| + t.Fatalf("unknown host: %v %v", ok, err) |
| + } |
| + // After a release the row is unheld again and "" matches once more. |
| + if ok, _ := r.ReleaseHold("h", holdOpA); !ok { |
| + t.Fatal("release") |
| + } |
| + if ok, err := r.SetHoldIf("h", "restoring", holdOpB, ""); err != nil || !ok { |
| + t.Fatalf("hold after release: %v %v", ok, err) |
| + } |
| +} |
diff --git a/internal/registry/registry.go b/internal/registry/registry.go| index ecdc61823..eaf9c13c3 100644 |
| --- a/internal/registry/registry.go |
| +++ b/internal/registry/registry.go |
| @@ -1075,6 +1075,33 @@ func (r *Registry) SetHold(host, status, op string) error { |
| return nil |
| } |
| +// SetHoldIf is SetHold conditioned on the row's current hold_op, in one |
| +// statement: the row is held status under op only while its hold_op is |
| +// still expectOp, which is "" (the caller saw the row unheld) or op itself |
| +// (an op changing the status of its own hold). Any other expectOp is |
| +// refused: a row held by another op leaves that hold only through |
| +// ReleaseHold, under the op that holds it (§5.2), never by takeover. false |
| +// with no error means nothing matched (the row's hold changed since the |
| +// caller read it, or there is no row), so a hold decided from an earlier |
| +// read never lands over what is there now (moves Task 6 fix round 1). |
| +// |
| +// As with SetHold, every caller touches the stage-requested marker after a |
| +// write that landed. |
| +func (r *Registry) SetHoldIf(host, status, op, expectOp string) (bool, error) { |
| + if !IsHold(status) || op == "" { |
| + return false, fmt.Errorf("registry: %q with op %q is not a hold", status, op) |
| + } |
| + if expectOp != "" && expectOp != op { |
| + return false, fmt.Errorf("registry: hold %s by %q expecting %q: another op's hold is released by that op, not taken over", host, op, expectOp) |
| + } |
| + res, err := r.sql.Exec(`UPDATE routes SET status = ?, hold_op = ? WHERE host = ? AND hold_op = ?`, status, op, host, expectOp) |
| + if err != nil { |
| + return false, err |
| + } |
| + n, err := res.RowsAffected() |
| + return n > 0, err |
| +} |
| + |
| // ReleaseHold is §5.2's one conditional statement: a row leaves a hold only |
| // under the op that holds it. false means nothing matched (a late write from a |
| // finished operation, an unheld row, or an empty op). |
diff --git a/internal/registry/timers.go b/internal/registry/timers.go| index 89f65b2e0..a49f04981 100644 |
| --- a/internal/registry/timers.go |
| +++ b/internal/registry/timers.go |
| @@ -12,6 +12,7 @@ import ( |
| "errors" |
| "fmt" |
| "os" |
| + "sort" |
| "strings" |
| ) |
| @@ -152,8 +153,34 @@ func (r *Registry) TimersFor(host string) ([]Timer, error) { |
| // where the LIMIT is applied, instead of after it. An empty (or nil) slice |
| // is no filter, which is the ordinary case: nothing in flight. |
| func (r *Registry) DueTimers(now int64, limit int, excludeHosts []string) ([]Timer, error) { |
| + return r.dueTimers(now, limit, excludeHosts, false) |
| +} |
| + |
| +// DueUnheldTimers is DueTimers without the rows of held routes (moves spec |
| +// §5.2, HoldStatuses). The exclusion is in the query, before the LIMIT, so |
| +// a held host's rows cannot fill the batch with rows the runner would only |
| +// skip and hide another host's due timer behind them (moves Task 6 fix |
| +// round 1). A row whose host has no route is still returned, as DueTimers |
| +// returns it. The runner asks this only with its hold wiring armed; |
| +// DueTimers stays main's query. |
| +func (r *Registry) DueUnheldTimers(now int64, limit int, excludeHosts []string) ([]Timer, error) { |
| + return r.dueTimers(now, limit, excludeHosts, true) |
| +} |
| + |
| +func (r *Registry) dueTimers(now int64, limit int, excludeHosts []string, skipHeld bool) ([]Timer, error) { |
| q := `SELECT ` + timerCols + ` FROM timers WHERE next_at > 0 AND next_at <= ?` |
| args := []any{now} |
| + if skipHeld { |
| + statuses := make([]string, 0, len(HoldStatuses)) |
| + for s := range HoldStatuses { |
| + statuses = append(statuses, s) |
| + } |
| + sort.Strings(statuses) |
| + q += ` AND host NOT IN (SELECT host FROM routes WHERE status IN (?` + strings.Repeat(",?", len(statuses)-1) + `))` |
| + for _, s := range statuses { |
| + args = append(args, s) |
| + } |
| + } |
| if len(excludeHosts) > 0 { |
| q += ` AND host NOT IN (?` + strings.Repeat(",?", len(excludeHosts)-1) + `)` |
| for _, h := range excludeHosts { |
diff --git a/internal/registry/timers_test.go b/internal/registry/timers_test.go| index 2210ab0e8..808ac5a7c 100644 |
| --- a/internal/registry/timers_test.go |
| +++ b/internal/registry/timers_test.go |
| @@ -531,3 +531,62 @@ func TestReplaceTimersWithNoRowsClearsTheHost(t *testing.T) { |
| t.Fatalf("rows left: %+v", got) |
| } |
| } |
| + |
| +// Moves Task 6 fix round 1: DueUnheldTimers leaves out every row whose |
| +// route is held (moves spec §5.2) BEFORE the limit, so a held host's rows |
| +// cannot fill the batch and hide another host's due timer. DueTimers itself |
| +// is unchanged: with the hold wiring off the runner still asks it. |
| +func TestDueUnheldTimersExcludesHeldHostsBeforeTheLimit(t *testing.T) { |
| + r := openTimersReg(t) |
| + for _, h := range []string{"held.test", "free.test"} { |
| + if err := r.Upsert(Route{Host: h, Kind: "instance", Socket: "/s"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if err := r.SetHold("held.test", "moving", holdOpA); err != nil { |
| + t.Fatal(err) |
| + } |
| + for i, name := range []string{"t0", "t1", "t2"} { |
| + if err := r.UpsertTimer(Timer{Host: "held.test", Name: name, Kind: "schedule", Spec: "every 1h0m0s", Path: "/j", NextAt: int64(10 + i)}); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + // Due after every held row; and a row whose host has no route at all, |
| + // which neither query drops (the converger prunes those). |
| + if err := r.UpsertTimer(Timer{Host: "free.test", Name: "t", Kind: "schedule", Spec: "every 1h0m0s", Path: "/j", NextAt: 50}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := r.UpsertTimer(Timer{Host: "gone.test", Name: "t", Kind: "schedule", Spec: "every 1h0m0s", Path: "/j", NextAt: 60}); err != nil { |
| + t.Fatal(err) |
| + } |
| + due, err := r.DueTimers(100, 2, nil) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(due) != 2 || due[0].Host != "held.test" || due[1].Host != "held.test" { |
| + t.Fatalf("DueTimers = %+v, want the two earliest held rows (main's answer)", due) |
| + } |
| + un, err := r.DueUnheldTimers(100, 2, nil) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(un) != 2 || un[0].Host != "free.test" || un[1].Host != "gone.test" { |
| + t.Fatalf("DueUnheldTimers = %+v, want free.test then gone.test", un) |
| + } |
| + // The busy-host exclusion still applies on top. |
| + un, err = r.DueUnheldTimers(100, 2, []string{"free.test"}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(un) != 1 || un[0].Host != "gone.test" { |
| + t.Fatalf("DueUnheldTimers excluding free.test = %+v", un) |
| + } |
| + // Once the hold lifts the rows are due again, in order. |
| + if ok, _ := r.ReleaseHold("held.test", holdOpA); !ok { |
| + t.Fatal("release") |
| + } |
| + un, _ = r.DueUnheldTimers(100, 2, nil) |
| + if len(un) != 2 || un[0].Host != "held.test" { |
| + t.Fatalf("after the release DueUnheldTimers = %+v", un) |
| + } |
| +} |
diff --git a/internal/schedule/converge_test.go b/internal/schedule/converge_test.go| index 6c8bb2694..3fc2f4153 100644 |
| --- a/internal/schedule/converge_test.go |
| +++ b/internal/schedule/converge_test.go |
| @@ -712,3 +712,38 @@ func TestConvergerSkipsTheTableWidePruneWhileAHostIsFrozen(t *testing.T) { |
| t.Fatalf("the prune did not resume after the freeze: %+v", tm) |
| } |
| } |
| + |
| +// Fix round 1 (task-6-review, Important 3): with neither Frozen nor Gate |
| +// set (the Phase 1 switch off) the converger is main's. A host whose row |
| +// status is a hold has its declaration converged and its undeclared |
| +// schedule pruned, and the table-wide prune runs. |
| +func TestConvergerWithoutTheHooksIsMain(t *testing.T) { |
| + reg, st := convergeFixture(t) |
| + ctx := context.Background() |
| + now := at("2026-08-23T13:07:00Z") |
| + const host = "a.bab.oncarlos.com" |
| + if err := reg.SetHold(host, "moving", "mv_a"); err != nil { |
| + t.Fatal(err) |
| + } |
| + reg.UpsertTimer(registry.Timer{Host: host, Name: "stale", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/s", NextAt: now.Unix() - 5}) |
| + 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) |
| + } |
| + UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now) |
| + c := NewConverger(reg, st, "box1", nil) |
| + c.Now = func() time.Time { return now } |
| + if err := c.Pass(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := reg.GetTimer(host, "sync"); err != nil { |
| + t.Fatalf("with the hold wiring off the held host's declaration did not converge: %v", err) |
| + } |
| + if tm, _ := reg.GetTimer(host, "stale"); tm != nil { |
| + t.Fatalf("with the hold wiring off the undeclared schedule survived: %+v", tm) |
| + } |
| + if tm, _ := reg.GetTimer(host, "done"); tm != nil { |
| + t.Fatalf("with the hold wiring off the finished once-row survived the table-wide prune: %+v", tm) |
| + } |
| +} |
diff --git a/internal/schedule/runner.go b/internal/schedule/runner.go| index e6631c955..494d010f3 100644 |
| --- a/internal/schedule/runner.go |
| +++ b/internal/schedule/runner.go |
| @@ -133,7 +133,18 @@ func (r *Runner) Tick(ctx context.Context) { |
| // with rows this tick is about to skip, and another host's due timer |
| // would never be seen. The skip below stays as belt-and-braces — a host |
| // can go busy between the query and the claim. |
| - due, err := r.Reg.DueTimers(now.Unix(), dueBatch, r.busyHosts()) |
| + // |
| + // With the hold wiring armed, held hosts are excluded in the query for |
| + // the same reason: a held row is skipped below with no bookkeeping, so |
| + // a held host's rows would sit at the front of every tick's batch and |
| + // starve the rest. Off, the query is main's. |
| + var due []registry.Timer |
| + var err error |
| + if r.holdsArmed() { |
| + due, err = r.Reg.DueUnheldTimers(now.Unix(), dueBatch, r.busyHosts()) |
| + } else { |
| + due, err = r.Reg.DueTimers(now.Unix(), dueBatch, r.busyHosts()) |
| + } |
| if err != nil { |
| log.Printf("schedule: due query: %v", err) |
| return |
| @@ -177,7 +188,8 @@ func (r *Runner) Tick(ctx context.Context) { |
| // 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) { |
| + // Only with the hold wiring armed: off, the row is main's. |
| + if r.holdsArmed() && registry.IsHold(rt.Status) { |
| continue |
| } |
| if r.beforeClaim != nil { |
| @@ -186,23 +198,25 @@ func (r *Runner) Tick(ctx context.Context) { |
| if !r.claim(host) { |
| 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) |
| + if r.holdsArmed() { |
| + // 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 |
| } |
| - 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 |
| @@ -223,7 +237,8 @@ func (r *Runner) Tick(ctx context.Context) { |
| // 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) { |
| + // Only with the hold wiring armed. |
| + if i > 0 && r.holdsArmed() && r.held(ctx, host) { |
| break |
| } |
| r.fire(ctx, rt, t) |
| @@ -379,6 +394,15 @@ type SocketRetainer interface { |
| RetainSockets(live map[string]bool) |
| } |
| +// holdsArmed reports whether the move-hold wiring is on. The host package |
| +// sets Gate (armScheduleHolds) only with the Phase 1 switch on and |
| +// scheduled work armed; Gate nil is the switch off, and then every hold |
| +// check in Tick is skipped and the runner is main's code path exactly: the |
| +// due query is DueTimers, there is no decision read inside the claim, no |
| +// check between a host's timers, and a row whose status is a hold is |
| +// delivered and recorded like any other. |
| +func (r *Runner) holdsArmed() bool { return r.Gate != nil } |
| + |
| // 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 { |
diff --git a/internal/schedule/runner_test.go b/internal/schedule/runner_test.go| index e4cfcac63..0cf11fb0c 100644 |
| --- a/internal/schedule/runner_test.go |
| +++ b/internal/schedule/runner_test.go |
| @@ -1075,6 +1075,7 @@ func TestHoldMidDeliveryStopsTheRest(t *testing.T) { |
| reg, r := runnerFixture(t, nil) |
| now := at("2026-08-23T13:07:00Z") |
| r.Now = func() time.Time { return now } |
| + r.Gate = &HostGate{} // the hold wiring armed; off, both would be delivered (TestRunnerWithoutTheGateIsMain) |
| var delivered []string |
| r.Deliver = deliverFunc(func(rt registry.Route, req *http.Request) (int, error) { |
| delivered = append(delivered, req.Header.Get("X-Carlos-Schedule")) |
| @@ -1187,3 +1188,73 @@ func TestDrainWaitsForTheDeliveryInFlight(t *testing.T) { |
| t.Fatalf("the table changed after the drain:\n got %+v\nwant %+v", got, snap) |
| } |
| } |
| + |
| +// Fix round 1 (task-6-review, Important 1): a held host's rows must not |
| +// consume the due batch. DueTimers applies its LIMIT before the runner's |
| +// hold filter, so a full batch of held rows due before an unrelated host's |
| +// row would be selected on every tick and that row never seen. With the |
| +// hold wiring armed the query itself leaves held hosts out, and the |
| +// unrelated timer runs this tick. |
| +func TestHeldHostsDoNotStarveAnotherHostsTimer(t *testing.T) { |
| + d := &fakeDeliverer{status: 204} |
| + reg, r := runnerFixture(t, d) |
| + now := at("2026-08-23T13:07:00Z") |
| + r.Now = func() time.Time { return now } |
| + r.Gate = &HostGate{} |
| + const quiet = "quiet.bab.oncarlos.com" |
| + if err := reg.Upsert(registry.Route{Host: quiet, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + // A full batch of rows on the held host, every one due BEFORE the quiet |
| + // host's single row. |
| + for i := 0; i < dueBatch; i++ { |
| + reg.UpsertTimer(registry.Timer{Host: holdHost, Name: fmt.Sprintf("j%02d", i), Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 1000 + int64(i)}) |
| + } |
| + reg.UpsertTimer(registry.Timer{Host: quiet, Name: "j", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 1}) |
| + if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil { |
| + t.Fatal(err) |
| + } |
| + before := timersOf(t, reg, holdHost) |
| + r.tick() |
| + if tm, _ := reg.GetTimer(quiet, "j"); tm == nil || tm.LastStatus != "204" { |
| + t.Fatalf("the quiet host's timer did not fire this tick behind a full batch of a held host's rows: %+v", tm) |
| + } |
| + for _, h := range d.hosts { |
| + if h != quiet { |
| + t.Fatalf("delivered to %s, a held host", h) |
| + } |
| + } |
| + if got := timersOf(t, reg, holdHost); !sameTimers(got, before) { |
| + t.Fatalf("the held host's bookkeeping moved:\n got %+v\nwant %+v", got, before) |
| + } |
| +} |
| + |
| +// Fix round 1 (task-6-review, Important 3): with no gate (the Phase 1 |
| +// switch off, or no hold wiring on this box) the runner is main's code |
| +// path. A row whose status is a hold is delivered like any other and its |
| +// bookkeeping moves: the batch includes it, the pre-filter and the decision |
| +// read do not look at the hold, and the between-timer check does not run. |
| +func TestRunnerWithoutTheGateIsMain(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 } |
| + reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "first", Kind: KindOnce, Spec: "at 1", Path: "/1", NextAt: 1, DueAt: 1}) |
| + reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "second", Kind: KindOnce, Spec: "at 2", Path: "/2", NextAt: 2, DueAt: 2}) |
| + if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil { |
| + t.Fatal(err) |
| + } |
| + r.tick() |
| + if len(d.calls) != 2 { |
| + t.Fatalf("with the hold wiring off a held row's due timers were delivered %d time(s), want both", len(d.calls)) |
| + } |
| + for _, name := range []string{"first", "second"} { |
| + tm, err := reg.GetTimer(holdHost, name) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if tm.Attempts != 1 || tm.LastStatus != "503" || tm.NextAt != now.Add(backoffBase).Unix() { |
| + t.Fatalf("%s: bookkeeping did not advance as main's does: %+v", name, tm) |
| + } |
| + } |
| +} |