amadan

carlos / platform Public

Consume a fired due time on the wake it caused, not at stop

A worker's stop deliberately writes no next_due: it has no socket for the
next-due probe, and the probe's failure path records 0, which would erase
the schedule that is a sleeping sidecar's only way back up. But that left
nothing to give a FIRED timer its one-shot semantics — the instance path
gets that for free by re-recording next_due on every stop. So once a
sidecar's next_due had passed, DueInstances returned the row on every
sweep: wake, restore, idle, park, re-wake within 30s, indefinitely, with
one lease acquire and one S3 restore/park per cycle and nothing in the
logs that reads as a failure.

The wake that a due time fires now consumes it (registry.ClearNextDueIf),
and only that wake: a worker that goes to sleep BEFORE its timer fires
keeps its schedule, which is the distinction between fixing this and
deleting the feature. The clear is conditional on the row still holding
the value that fired, so a child that posted its next due time over the
agent socket during its own settle window keeps it. A failed wake
consumes nothing — the index stays the retry, exactly as the due sweep's
own comment describes.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014WvcQVWggXkE5BAkjRDgWF
Paul Campbell pushed by paul@keymail.dev ddee127d04c4348701794a6acc9dd9aee2c5d809 parent 988cc95
4 files changed, +235 −0
  • internal/activator/activator.go +51 −0
  • internal/activator/activator_test.go +132 −0
  • internal/registry/registry.go +24 −0
  • internal/registry/registry_test.go +28 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index fd3bdc753..bc474152a 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -210,6 +210,10 @@ type Backend interface {
// that doesn't exist anywhere it's used. NoteWorkerReady is its one caller.
type DueStore interface {
SetNextDue(host string, due int64) error
+ // ClearNextDueIf consumes a FIRED due time — see the registry's own doc
+ // comment. Conditional on the row still holding the value that fired, so
+ // a child that posted its next due time during this very wake keeps it.
+ ClearNextDueIf(host string, due int64) error
DueInstances(now int64) ([]registry.Route, error)
SetVersions(host, ver, prev string) error
}
@@ -341,6 +345,14 @@ type actEntry struct {
readyc chan struct{} // closed when the child reports ready (ends the settle window)
readyNoted bool // guards readyc's one close
+ // dueFired is the next_due value that TRIGGERED this wake, and only for a
+ // worker woken by the due-time sweep (0 everywhere else). A fired timer is
+ // one-shot: the wake consumes the schedule it acted on (consumeDueTime),
+ // which is what stops a hibernating sidecar waking, idling, parking and
+ // re-waking on every sweep forever. Set before the wake goroutine starts
+ // and never written again, like the other construction-time worker fields.
+ dueFired int64
+
workerToken string // the per-wake agent-socket token
spawnedAt time.Time // when this child spawned — the max_run and healthy-run clock
workerVersion string // the build the child reported over the agent socket
@@ -718,6 +730,27 @@ func (a *Service) releaseFunc(e *actEntry) func() {
}
}
+// consumeDueTime clears the due time that fired THIS wake, so a fired timer
+// wakes its worker exactly once. A no-op for every other wake — an operator
+// command, the boot pass, an instance's request wake — which is what leaves a
+// schedule that has not yet come due completely untouched.
+//
+// Conditional on the row still holding the fired value (ClearNextDueIf): the
+// child spawned by this very wake may already have called /v1/next-due during
+// its settle window, and that newer schedule must survive the consumption of
+// the older one.
+func (a *Service) consumeDueTime(e *actEntry) {
+ if !e.worker || e.dueFired <= 0 {
+ return
+ }
+ if err := a.due.ClearNextDueIf(e.host, e.dueFired); err != nil {
+ // Loud, but not a failed wake: the child is up and doing its work. The
+ // cost of the miss is one extra wake on a later sweep, which is
+ // idempotent — the opposite trade to failing a wake that succeeded.
+ log.Printf("activator: consume fired due time for %s: %v — it may wake once more", e.host, err)
+ }
+}
+
// wake drives asleep→awake: lease, restore-if-missing, spawn, replicate,
// co-spawn the sidecar. It runs on the first requester's goroutine but under
// its own generous timeout, detached from that request's context — queued
@@ -728,6 +761,19 @@ func (a *Service) wake(e *actEntry) {
defer cancel()
w, err := a.doWake(ctx, e)
+ // A due-fired wake CONSUMES the schedule it acted on — here, on success,
+ // and never at stop. The distinction is the whole fix: a worker that goes
+ // to sleep BEFORE its timer fires must keep its schedule (that row is its
+ // only way back up), while one whose timer has already fired must not be
+ // handed the same fired time again on the next sweep. Ordered before the
+ // awake transition below so nothing can observe this entry awake with the
+ // spent schedule still standing. A FAILED wake deliberately consumes
+ // nothing: the index is then the retry, exactly as the due sweep's own
+ // comment describes.
+ if err == nil {
+ a.consumeDueTime(e)
+ }
+
// The member-facing record of this wake (spec §5): composed here, where
// the route and the whole error are in hand — the journal line below the
// backends already write stays exactly as it was.
@@ -1879,6 +1925,11 @@ func (a *Service) sweepOnce() {
var e *actEntry
if rt.IsSidecar() {
e = a.newWorkerEntry(rt)
+ // The value that fired, carried into the wake so a SUCCESSFUL one
+ // can consume exactly it (consumeDueTime) — a worker's stop writes
+ // no next_due at all, so without this the row keeps its passed
+ // value and this same sweep re-wakes it every tick.
+ e.dueFired = rt.NextDue
} else {
e = &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})}
}
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 651a4d026..b20a447f8 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -216,6 +216,11 @@ type fakeBackend struct {
// from then on: its Stop blocks until the gate closes, so a test can hold
// a child mid-drain and watch what happens around it.
workerStopGate chan struct{}
+ // workerStartHook, when set, runs INSIDE StartWorker (off b.mu), so a test
+ // can model what the child itself does while its wake is still resolving —
+ // notably posting its next due time over the agent socket during the
+ // settle window.
+ workerStartHook func(rt registry.Route)
// order interleaves backend calls and proc stops (start:instance,
// start:sidecar, nextDue, stop:sidecar, stop:instance) for the ordering
// assertions.
@@ -337,6 +342,7 @@ func (b *fakeBackend) setSidecarErr(err error) {
func (b *fakeBackend) StartWorker(ctx context.Context, rt registry.Route, env map[string]string) (Proc, string, error) {
b.mu.Lock()
+ hook := b.workerStartHook
delay := b.workerStartDelay
b.workerEnterN++ // counted BEFORE the delay: "the exec has begun"
b.mu.Unlock()
@@ -351,6 +357,9 @@ func (b *fakeBackend) StartWorker(ctx context.Context, rt registry.Route, env ma
return nil, "", ctx.Err()
}
}
+ if hook != nil {
+ hook(rt) // the child, acting while its wake is still in flight
+ }
b.mu.Lock()
defer b.mu.Unlock()
b.workerStartN++
@@ -2503,6 +2512,129 @@ func TestDueSweepWakesASidecarAsAWorker(t *testing.T) {
}
}
+// A FIRED timer is one-shot. The row's next_due is not rewritten by a
+// worker's stop (it has no socket to probe), so unless the wake itself
+// consumes the value that fired, DueInstances hands the same row back on
+// EVERY sweep: wake, restore, idle, park, re-wake, forever — one lease
+// acquire and one S3 restore/park per cycle, and nothing in the logs that
+// reads as a failure.
+func TestDueFiredWorkerWakeHappensExactlyOnce(t *testing.T) {
+ a, b, clk, reg := testActivator(t, 10*time.Second)
+ rt := testWorkerRoute("bot.x.internal")
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ if err := reg.SetNextDue(rt.Host, clk.now().Unix()-1); err != nil {
+ t.Fatalf("set next due: %v", err)
+ }
+
+ a.sweepOnce()
+ waitFor(t, "the due-fired wake", func() bool { return a.entryAwake(rt.Host) })
+ got, err := reg.Get(rt.Host)
+ if err != nil {
+ t.Fatalf("get: %v", err)
+ }
+ if got.NextDue != 0 {
+ t.Fatalf("next_due = %d after the wake it fired, want 0 — a fired timer must be consumed by the wake it caused", got.NextDue)
+ }
+
+ // It idles out, parks, and STAYS parked: several more sweeps, no second
+ // wake, no churn.
+ clk.advance(11 * time.Second)
+ a.sweepOnce()
+ waitFor(t, "the worker to park", func() bool { return a.entryCount() == 0 })
+ for i := 0; i < 5; i++ {
+ clk.advance(11 * time.Second)
+ a.sweepOnce()
+ }
+ if n := b.workerStarts(); n != 1 {
+ t.Fatalf("workerStarts = %d, want 1 — one due time woke the sidecar more than once (wake/park churn, every sweep, forever)", n)
+ }
+ if n := b.releaseCount(); n > 1 {
+ t.Fatalf("leases released %d times for one due time — the sidecar is churning", n)
+ }
+}
+
+// The other half of the same distinction: a sidecar that goes to sleep BEFORE
+// its timer fires must keep its schedule. Consuming at STOP (rather than on
+// the wake the timer caused) would delete the feature — a sleeping sidecar's
+// next_due is its only way back up, since nothing can dial one.
+func TestWorkerSleepingBeforeItsDueTimeStillWakesForIt(t *testing.T) {
+ a, b, clk, reg := testActivator(t, 10*time.Second)
+ rt := testWorkerRoute("bot.x.internal")
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ if err := a.WakeWorker(rt); err != nil { // an operator wake, not a due-fired one
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ due := clk.now().Add(time.Minute).Unix()
+ if err := a.SetWorkerNextDue(rt.Host, due); err != nil { // "wake me in a minute"
+ t.Fatalf("SetWorkerNextDue: %v", err)
+ }
+
+ clk.advance(11 * time.Second) // idles out well before the due time
+ a.sweepOnce()
+ waitFor(t, "the worker to park", func() bool { return a.entryCount() == 0 })
+ got, err := reg.Get(rt.Host)
+ if err != nil {
+ t.Fatalf("get: %v", err)
+ }
+ if got.NextDue != due {
+ t.Fatalf("next_due = %d after a sleep BEFORE the due time, want %d — a schedule that has not fired must survive the sleep", got.NextDue, due)
+ }
+
+ clk.advance(time.Minute)
+ a.sweepOnce()
+ waitFor(t, "the scheduled wake", func() bool { return b.workerStarts() == 2 })
+}
+
+// The consumption is conditional on the row still holding the value that
+// fired: the child spawned by this very wake may post its NEXT due time over
+// the agent socket during the settle window, and an unconditional clear would
+// eat it — a bot that ran once and never again.
+func TestDueFiredWakeKeepsANewNextDuePostedByTheChild(t *testing.T) {
+ a, b, clk, reg := testActivator(t, 10*time.Second)
+ rt := testWorkerRoute("bot.x.internal")
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ if err := reg.SetNextDue(rt.Host, clk.now().Unix()-1); err != nil {
+ t.Fatalf("set next due: %v", err)
+ }
+ next := clk.now().Add(time.Minute).Unix()
+ b.mu.Lock()
+ b.workerStartHook = func(r registry.Route) {
+ // The child's own /v1/next-due, landing while the wake is still in
+ // flight — before anything could consume the time that fired.
+ if err := a.SetWorkerNextDue(r.Host, next); err != nil {
+ t.Errorf("SetWorkerNextDue: %v", err)
+ }
+ }
+ b.mu.Unlock()
+
+ a.sweepOnce()
+ waitFor(t, "the due-fired wake", func() bool { return a.entryAwake(rt.Host) })
+ got, err := reg.Get(rt.Host)
+ if err != nil {
+ t.Fatalf("get: %v", err)
+ }
+ if got.NextDue != next {
+ t.Fatalf("next_due = %d, want %d — consuming the FIRED time must not eat the one the child posted during its wake", got.NextDue, next)
+ }
+
+ // And that schedule is honoured: park, then wake again when it arrives.
+ b.mu.Lock()
+ b.workerStartHook = nil
+ b.mu.Unlock()
+ clk.advance(11 * time.Second)
+ a.sweepOnce()
+ waitFor(t, "the worker to park", func() bool { return a.entryCount() == 0 })
+ clk.advance(time.Minute)
+ a.sweepOnce()
+ waitFor(t, "the child's own scheduled wake", func() bool { return b.workerStarts() == 2 })
+}
+
// --- the settle window --------------------------------------------------------
// noteReadyAsChild reports readiness the way the real child must: presenting
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index 5c1dd4e80..2aed644dc 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -1303,6 +1303,30 @@ func (r *Registry) SetNextDue(host string, due int64) error {
return err
}
+// ClearNextDueIf CONSUMES a fired due time: it clears routes.next_due for
+// host, but only while the row still holds due — the exact value that fired.
+//
+// A fired timer is one-shot, and this is what makes it so. The instance path
+// gets that for free: an instance re-records next_due from its own probe on
+// every stop, so a passed due time is overwritten the moment the host sleeps
+// again. A WORKER's stop deliberately writes nothing (it has no socket to
+// probe, and a failed probe records 0 — erasing the schedule that is a
+// sleeping sidecar's only way back up), so without this the row keeps its
+// passed value and DueInstances returns it on EVERY sweep: wake, restore,
+// idle, park, re-wake, forever, with nothing in the logs that reads as a
+// failure.
+//
+// Conditional, not a bare clear, because the child that this wake just
+// spawned may already have posted its NEXT due time over the agent socket
+// (SetWorkerNextDue) before the wake resolved. Clearing unconditionally would
+// erase a schedule that was set after the one being consumed; the WHERE
+// next_due = ? is what keeps consumption from eating it. A row that moved
+// under us is simply left alone.
+func (r *Registry) ClearNextDueIf(host string, due int64) error {
+ _, err := r.sql.Exec(`UPDATE routes SET next_due = 0 WHERE host = ? AND next_due = ?`, host, due)
+ return err
+}
+
// SetSeed stamps a route's first-boot init argv as a narrow UPDATE — same
// rationale as SetAccount (a read-modify-write Upsert racing a deploy's
// SetVersions or the activator's SetNextDue would clobber it). Unlike
diff --git a/internal/registry/registry_test.go b/internal/registry/registry_test.go
index 189291ead..d17816b7a 100644
--- a/internal/registry/registry_test.go
+++ b/internal/registry/registry_test.go
@@ -2404,6 +2404,34 @@ func TestDueInstancesIncludesSidecars(t *testing.T) {
}
}
+// TestClearNextDueIfConsumesOnlyTheValueThatFired pins both halves of the
+// one-shot rule a sidecar's timer depends on: the fired value is consumed,
+// and a value written since (the child's own /v1/next-due, landing while its
+// wake was still resolving) is left standing.
+func TestClearNextDueIfConsumesOnlyTheValueThatFired(t *testing.T) {
+ r := openTestRegistry(t)
+ host := "bot.eleven.bdm.sidecar.internal"
+ if err := r.Upsert(Route{Host: host, Kind: "sidecar", DBPath: "/tmp/bot.db", Hibernate: true, Status: "ready"}); err != nil {
+ t.Fatalf("upsert sidecar: %v", err)
+ }
+ if err := r.SetNextDue(host, 100); err != nil {
+ t.Fatalf("SetNextDue: %v", err)
+ }
+ // A different value: not the one that fired, so it must survive.
+ if err := r.ClearNextDueIf(host, 99); err != nil {
+ t.Fatalf("ClearNextDueIf: %v", err)
+ }
+ if rt, err := r.Get(host); err != nil || rt.NextDue != 100 {
+ t.Fatalf("next_due = %+v (%v), want 100 — a row that moved since the wake fired must be left alone", rt, err)
+ }
+ if err := r.ClearNextDueIf(host, 100); err != nil {
+ t.Fatalf("ClearNextDueIf: %v", err)
+ }
+ if rt, err := r.Get(host); err != nil || rt.NextDue != 0 {
+ t.Fatalf("next_due = %+v (%v), want 0 — a fired due time must be consumed by the wake it caused", rt, err)
+ }
+}
+
// TestArgsRoundTrips checks Args survives Upsert and can be repointed
// narrowly by SetArgs, the door the sidecar reconcile pass uses when the
// app's manifest argv changes.