amadan
Start an always-on sidecar that appeared while the agent was running
Design §7.2 says the record exists ⇒ the process runs. That held only across agent restarts: the worker boot pass runs once, at arming, and the due-time loop gates on Hibernate — so an always-on sidecar created under a running agent had no starter at all. Nothing dials one and it carries no schedule, so it simply never ran until the box's agent next restarted. The sweep grows the matching arm: any always-on sidecar row with no live entry is woken, bounded per tick the way the boot pass bounds its own fan-out, since each wake claims an S3 lease and restores a database. It also covers the other ways such a worker ends up stranded — a wake that failed, or a stop with no re-wake behind it — and it leaves hibernating rows strictly alone, which is the whole point of that class.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014WvcQVWggXkE5BAkjRDgWF
3 files changed,
+142
−0
internal/activator/activator.go+48 −0internal/activator/activator_test.go+67 −0internal/registry/registry.go+27 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index bc474152a..e33edf39d 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -215,6 +215,10 @@ type DueStore interface { |
| // 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) |
| + // AlwaysOnWorkers lists the non-hibernating sidecar rows this box carries, |
| + // so the sweep can start one that appeared after the agent's boot pass — |
| + // which runs once, at arming, and never again. |
| + AlwaysOnWorkers() ([]registry.Route, error) |
| SetVersions(host, ver, prev string) error |
| } |
| @@ -411,6 +415,13 @@ func settlePoll(window time.Duration) time.Duration { |
| return poll |
| } |
| +// maxWorkerStartsPerSweep bounds the sweep's always-on arm — how many |
| +// stranded always-on sidecars it starts per tick. The same trade host's |
| +// workerBootConcurrency makes: bounded fan-out, not one-at-a-time and not |
| +// the whole box at once, because each wake claims a lease and restores a |
| +// database. The remainder is started on the next sweep. |
| +const maxWorkerStartsPerSweep = 8 |
| + |
| // Build convergence: how long an awake child may serve a symlink-stale binary |
| // before the sweep gracefully cycles it, and how many to cycle per tick. |
| // - ConvergeGrace comfortably exceeds one sweep interval (≤30s) and a |
| @@ -1735,6 +1746,12 @@ func (a *Service) sweepOnce() { |
| if err != nil { |
| log.Printf("activator: due-time sweep: %v", err) |
| } |
| + // The always-on arm's input, read on the same off-lock terms and for the |
| + // same reason: nothing under a.mu should wait on SQLite. |
| + alwaysOn, err := a.due.AlwaysOnWorkers() |
| + if err != nil { |
| + log.Printf("activator: always-on sidecar sweep: %v", err) |
| + } |
| a.mu.Lock() |
| cycled := 0 // build-convergence cycles started this tick (rate-limited) |
| for _, e := range a.entries { |
| @@ -1936,6 +1953,37 @@ func (a *Service) sweepOnce() { |
| a.entries[rt.Host] = e |
| go a.wake(e) |
| } |
| + |
| + // The always-on arm: design §7.2 says the record exists ⇒ the process |
| + // runs, and until this existed that held only across agent RESTARTS. The |
| + // boot pass (host's runWorkerBootPass) runs once, at arming; the due loop |
| + // above gates on Hibernate; so an always-on sidecar created while the |
| + // agent was running had nothing that would ever start it — no request can |
| + // reach one, and it has no schedule. It also covers every other way an |
| + // always-on worker ends up with no live entry: a wake that failed, or a |
| + // stop with no re-wake behind it. |
| + // |
| + // Bounded per sweep for the reason host's workerBootConcurrency is |
| + // bounded: each wake claims an S3 lease and restores a database, so a box |
| + // just handed a fleet of sidecars starts them in batches. The remainder |
| + // comes up on the next sweep. |
| + started := 0 |
| + for _, rt := range alwaysOn { |
| + if a.shuttingDown { |
| + break // the agent is going away; nothing new may claim a lease |
| + } |
| + if started >= maxWorkerStartsPerSweep { |
| + break |
| + } |
| + if _, live := a.entries[rt.Host]; live { |
| + continue // awake, waking, or draining a stop — none of those is stranded |
| + } |
| + log.Printf("activator: always-on sidecar %s is not running — starting it", rt.Host) |
| + e := a.newWorkerEntry(rt) |
| + a.entries[rt.Host] = e |
| + go a.wake(e) |
| + started++ |
| + } |
| a.mu.Unlock() |
| } |
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go| index b20a447f8..7999ec627 100644 |
| --- a/internal/activator/activator_test.go |
| +++ b/internal/activator/activator_test.go |
| @@ -3043,6 +3043,73 @@ func TestAlwaysOnWorkerIgnoresIdleClockAndMaxRun(t *testing.T) { |
| } |
| } |
| +// §7.2's contract is "the record exists ⇒ the process runs", and until the |
| +// sweep grew this arm it held only across agent RESTARTS: the boot pass runs |
| +// once, at arming, and the due-time loop gates on Hibernate — so an always-on |
| +// sidecar created while the agent was running had nothing at all that would |
| +// start it. |
| +func TestSweepStartsAnAlwaysOnWorkerThatAppearedMidFlight(t *testing.T) { |
| + a, b, _, reg := testActivator(t, time.Hour) |
| + rt := alwaysOnWorkerRoute("bot.x.internal") |
| + if err := reg.Upsert(rt); err != nil { // the reconcile pass mints it, mid-flight |
| + t.Fatalf("upsert: %v", err) |
| + } |
| + |
| + a.sweepOnce() |
| + waitFor(t, "the always-on sidecar to start", func() bool { return a.entryAwake(rt.Host) }) |
| + if n := b.workerStarts(); n != 1 { |
| + t.Fatalf("workerStarts = %d, want 1", n) |
| + } |
| + |
| + // And it is a one-shot, not a per-tick respawn: an entry that is already |
| + // live is left entirely alone. |
| + for i := 0; i < 3; i++ { |
| + a.sweepOnce() |
| + } |
| + if n := b.workerStarts(); n != 1 { |
| + t.Fatalf("workerStarts = %d after three more sweeps, want 1 — a running always-on sidecar must not be started again", n) |
| + } |
| +} |
| + |
| +// The arm is for always-on rows only. A HIBERNATING sidecar exists precisely |
| +// so that nothing runs until a wake — starting one here would delete the |
| +// class. |
| +func TestSweepNeverStartsAHibernatingWorker(t *testing.T) { |
| + a, b, _, reg := testActivator(t, time.Hour) |
| + rt := testWorkerRoute("bot.x.internal") // Hibernate: true, no next_due |
| + if err := reg.Upsert(rt); err != nil { |
| + t.Fatalf("upsert: %v", err) |
| + } |
| + for i := 0; i < 3; i++ { |
| + a.sweepOnce() |
| + } |
| + if n := b.workerStarts(); n != 0 { |
| + t.Fatalf("workerStarts = %d — a hibernating sidecar must stay asleep until a wake source fires", n) |
| + } |
| + if a.entryCount() != 0 { |
| + t.Fatal("a hibernating sidecar was given a live entry by the sweep") |
| + } |
| +} |
| + |
| +// A stranded always-on worker is a bounded batch, not a stampede: each wake |
| +// claims an S3 lease and restores a database. |
| +func TestSweepBoundsAlwaysOnStartsPerTick(t *testing.T) { |
| + a, b, _, reg := testActivator(t, time.Hour) |
| + for i := 0; i < maxWorkerStartsPerSweep+3; i++ { |
| + rt := alwaysOnWorkerRoute(fmt.Sprintf("bot%02d.x.internal", i)) |
| + if err := reg.Upsert(rt); err != nil { |
| + t.Fatalf("upsert: %v", err) |
| + } |
| + } |
| + a.sweepOnce() |
| + waitFor(t, "the first batch to start", func() bool { return b.workerStarts() == maxWorkerStartsPerSweep }) |
| + if n := b.workerStarts(); n != maxWorkerStartsPerSweep { |
| + t.Fatalf("workerStarts = %d in one sweep, want the per-sweep bound %d", n, maxWorkerStartsPerSweep) |
| + } |
| + a.sweepOnce() // the remainder comes up on the next tick |
| + waitFor(t, "the remainder to start", func() bool { return b.workerStarts() == maxWorkerStartsPerSweep+3 }) |
| +} |
| + |
| // A supervised restart must not exec while holding the activator's global |
| // mutex: ExecBackend.StartWorker materializes the build from the bucket when |
| // live/<host> is dangling, and a slow fetch held across a.mu stalls every |
diff --git a/internal/registry/registry.go b/internal/registry/registry.go| index 2aed644dc..440380e12 100644 |
| --- a/internal/registry/registry.go |
| +++ b/internal/registry/registry.go |
| @@ -1327,6 +1327,33 @@ func (r *Registry) ClearNextDueIf(host string, due int64) error { |
| return err |
| } |
| +// AlwaysOnWorkers returns every non-hibernating sidecar route in a runnable |
| +// status — design §7.2's "the record exists ⇒ the process runs", asked as a |
| +// query so the activator's sweep can start one that appeared AFTER the |
| +// agent's boot pass ran (the boot pass runs once, at arming, and never |
| +// again). |
| +// |
| +// Status is filtered exactly as the due-time sweep and the boot pass filter |
| +// it (host's workersToStart): only "ready", or the legacy "" some pre-status |
| +// rows carry — a row still provisioning, or draining, must not be spawned. |
| +func (r *Registry) AlwaysOnWorkers() ([]Route, error) { |
| + rows, err := r.sql.Query(`SELECT ` + routeCols + ` FROM routes |
| +WHERE kind = 'sidecar' AND hibernate = 0 AND (status = '' OR status = 'ready') ORDER BY host`) |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer rows.Close() |
| + var out []Route |
| + for rows.Next() { |
| + t, err := scanRoute(rows.Scan) |
| + if err != nil { |
| + return nil, err |
| + } |
| + out = append(out, t) |
| + } |
| + return out, rows.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 |