carlos / platform Public

schedule/host: a bound the test can actually break, and the prune leg evicts

Review fixes on the B2 wave.

I1, the one that mattered: the ≤4-parallel test proved nothing. It opened the
gate the moment four deliveries were in flight, so `cur` drained before a
fifth could ever be seen — with the semaphore quadrupled it passed 25 runs out
of 25. The bound is now watched with every delivery still parked, through two
negative waits: a fifth delivery must not start, and Tick must not return (with
8 hosts and 4 slots it is parked on the semaphore, and a widened bound is what
lets it finish). Both can only be weakened by a slow machine, never reddened.
Verified red under a 4x-wide semaphore, first run, 0.01s.

Minors from the same review:

- The delivery leg's eviction now includes the prune leg's sweep: it has
  already read every route, so it hands the live SOCKET set to the Deliverer
  (schedule.SocketRetainer). A retired route can be reclaimed no other way —
  nothing delivers to it, so nothing it holds errors its way out — and without
  it the entry waited for the 512 reset.
- Dropped proxies close their idle connections instead of waiting out the 90s
  IdleConnTimeout, on evict, on the retain sweep and on the reset. Guarded
  assertions, with a test pinning the shape edge.UnixProxy returns so that
  losing the eager close fails loudly rather than quietly.
- Comments corrected where they overclaimed: a socket that will not dial is
  ReverseProxy's own 502, not status 0 and not an error (so it is not the
  eviction trigger the comment named, and the dial retry is a fixed 13s
  window, not the caller's context); the 5m converge budget bounds queue plus
  work, since Pass takes its mutex first.
- The runner's prune log lines name the runner, so a journal on a box with
  both legs says which one swept; the runner declines the sweep outright when
  the registry has no routes, fencing PruneTimers' delete-everything arm now
  that it has two callers (whose doc now names them).
- The wake test's in-flight assertion had a ~400ms budget that a stalled
  machine could blow, with a misleading message. The handler now parks until
  the test releases it, so "still awake after several sweeps past the idle
  window" has no deadline at all. It also asserts X-Carlos-Schedule-Kind, the
  last header in spec §6.3's table it was skipping.
- Spec: §6.3's sketch carries the busy-host exclusion and the daily prune,
  §6.2 step 5 credits the runner's leg for the storeless box.

Paul Campbell pushed by paul@keymail.dev 2 months ago 907beb246ab9d37b4a94a0ba3775ef0d90194430 parent ac8b70c

6 files changed, +316 −55

docs/superpowers/specs/2026-08-23-scheduled-work-design.md +12 −2
internal/host/schedule_wake_test.go +87 −19
internal/host/schedule_wiring.go +67 −12
internal/registry/timers.go +10 −2
internal/schedule/runner.go +42 −7
internal/schedule/runner_test.go +98 −13

diff --git a/docs/superpowers/specs/2026-08-23-scheduled-work-design.md b/docs/superpowers/specs/2026-08-23-scheduled-work-design.md

index 2da60869f..0dde915fd 100644

--- a/docs/superpowers/specs/2026-08-23-scheduled-work-design.md

+++ b/docs/superpowers/specs/2026-08-23-scheduled-work-design.md

@@ -272,7 +272,10 @@ routes:

the named schedule on every route of the app. A name that is not declared

is logged and ignored.

5. `PruneTimers`: rows whose host is no longer in the registry (route

- removed) and `once` rows done (`next_at = 0`) for more than 7 days.

+ removed) and `once` rows done (`next_at = 0`) for more than 7 days. The

+ **Runner** runs the same sweep daily on its own tick (§6.3), because this

+ converger does not exist on a box with no deploy store — and that box

+ still accumulates once-rows from the control socket (§7.2).

6. For every host whose rows changed, `report(host)` — write §4.3.

Converge is idempotent and die-free (one app's failure never aborts the

@@ -285,7 +288,8 @@ does bucket reads; this does a local SQLite read — cheap enough for 15s and

the lateness budget is "≤15s + wake").

```

-DueTimers(now, 64)

+prune once a day // §6.2 step 5, runner-side: no converger on a storeless box

+DueTimers(now, 64, busyHosts) // hosts with a delivery still in flight are excluded

group by host; hosts run in parallel (≤4 at once), one timer at a time per host

for each timer:

deliver → outcome

@@ -294,6 +298,12 @@ DueTimers(now, 64)

report(host) // §4.3, best effort

```

+The batch is capped, and a host whose delivery is still running is skipped —

+so the busy set is excluded **in the query**, before the `LIMIT`. Without

+that, one host with a batch's worth of due rows (or one parked in a

+30-minute handler) fills every batch with rows the tick is about to skip and

+another host's due timer is never seen at all.

**Delivery** builds `POST <path>` with:

| Header | Value |

diff --git a/internal/host/schedule_wake_test.go b/internal/host/schedule_wake_test.go

index b0ca445f8..bd55d1fbb 100644

--- a/internal/host/schedule_wake_test.go

+++ b/internal/host/schedule_wake_test.go

@@ -16,6 +16,7 @@ import (

"errors"

"net"

"net/http"

+ "net/http/httputil"

"path/filepath"

"strconv"

"sync"

@@ -23,6 +24,7 @@ import (

"time"

"github.com/carlosframework/platform/internal/activator"

+ "github.com/carlosframework/platform/internal/edge"

"github.com/carlosframework/platform/internal/registry"

"github.com/carlosframework/platform/internal/release"

"github.com/carlosframework/platform/internal/schedule"

@@ -190,8 +192,8 @@ func wakeFixture(t *testing.T, host string, idle time.Duration, h http.HandlerFu

// 3. the app's handler runs and sees the contract: the bearer token the

// agent minted for THIS instance, and X-Carlos-Schedule-At carrying the

// occurrence (spec §6.3's header table);

-// 4. the in-flight hold is real — the handler runs LONGER than the idle

-// window and is not cut short;

+// 4. the in-flight hold is real — the handler is held open past the idle

+// window, over several sweeps, and the instance stays awake;

// 5. the delivery is recorded as a 2xx on the row; and

// 6. the instance hibernates once the tick has been idle for the window:

// the activator drops the entry and the socket goes away again.

@@ -204,16 +206,19 @@ func wakeFixture(t *testing.T, host string, idle time.Duration, h http.HandlerFu

// from a sleeping start.

func TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain(t *testing.T) {

const host = "sleepy.bab.oncarlos.com"

- // The idle window is short so the test can watch a real hibernation; the

- // handler deliberately runs past it (see the in-flight assertion below).

+ // The idle window is short so the test can watch a real hibernation.

const idle = 400 * time.Millisecond

- const handlerRuns = 3 * idle

- // 1. Asleep: no entry, and nothing listening (asserted in wakeFixture).

+ // The handler PARKS until the test releases it, rather than sleeping a

+ // fixed span. That is what takes the clock out of step 4: the instance

+ // must stay awake for as long as the handler runs, so the test can look

+ // at "still awake" whenever it likes, and a stalled machine only makes

+ // the handler run longer — it cannot turn the assertion red.

seen := make(chan *http.Request, 4)

+ release := make(chan struct{})

reg, inst, svc, dd := wakeFixture(t, host, idle, func(w http.ResponseWriter, r *http.Request) {

seen <- r.Clone(context.Background())

- time.Sleep(handlerRuns)

+ <-release

w.WriteHeader(http.StatusNoContent)

})

@@ -252,6 +257,9 @@ func TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain(t *testing.T) {

if got := req.Header.Get("X-Carlos-Schedule"); got != "sync" {

t.Fatalf("X-Carlos-Schedule = %q", got)

}

+ if got := req.Header.Get("X-Carlos-Schedule-Kind"); got != schedule.KindSchedule {

+ t.Fatalf("X-Carlos-Schedule-Kind = %q, want %q", got, schedule.KindSchedule)

+ }

if req.Method != http.MethodPost || req.URL.Path != "/jobs/sync" {

t.Fatalf("request = %s %s", req.Method, req.URL.Path)

}

@@ -259,16 +267,28 @@ func TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain(t *testing.T) {

t.Fatalf("the instance was started %d times, want exactly 1 wake", n)

}

- // 4 + 5. The handler outlives the idle window and is not cut short: the

- // activator's in-flight count is what holds the instance open (spec §6.3),

- // and the row records the 204 the handler eventually wrote.

+ // 4. The in-flight hold. The handler is parked, so the instance has now

+ // been "idle" (no activity stamp — that lands when the request finishes)

+ // for longer than the whole idle window, and the sweeper has run several

+ // times over it. It must still be awake: the activator's in-flight count

+ // is what holds it open for the length of a handler (spec §6.3), and

+ // without that hold a 30-minute job would be hibernated out from under

+ // itself. No deadline pressure here — the handler waits for the test.

+ time.Sleep(3 * idle)

+ if !svc.Owns(host) || !inst.running() {

+ t.Fatalf("the instance was hibernated with its tick still running (owns=%v running=%v) — the in-flight hold did not hold",

+ svc.Owns(host), inst.running())

+ }

+ close(release)

+ // 5. The row records the 204 the handler wrote once released, and

+ // advances to the next occurrence.

deadline := time.Now().Add(30 * time.Second)

for {

tm, _ := reg.GetTimer(host, "sync")

if tm != nil && tm.LastStatus != "" {

if tm.LastStatus != "204" {

- t.Fatalf("recorded %q, want 204 — the handler ran %s, past the %s idle window, and must not be cut short",

- tm.LastStatus, handlerRuns, idle)

+ t.Fatalf("recorded %q, want 204 — the handler ran past the %s idle window and must not be cut short", tm.LastStatus, idle)

}

if tm.NextAt <= now.Unix() {

t.Fatalf("row did not advance: %+v", tm)

@@ -280,9 +300,6 @@ func TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain(t *testing.T) {

}

time.Sleep(5 * time.Millisecond)

}

- if !svc.Owns(host) {

- t.Fatal("the instance was already hibernated when its tick finished — the in-flight hold did not hold")

- }

// 6. Idle again: the sweep hibernates it, and the socket goes with it.

for {

@@ -430,18 +447,21 @@ func TestProxyCacheIsStableAndBounded(t *testing.T) {

t.Fatalf("cache holds %d entries, want at most maxProxyCache (%d)", n, maxProxyCache)

}

- // A delivery that errors retires its socket: the next tick dials fresh.

+ // A delivery that ERRORS retires its socket: the next tick dials fresh.

+ // Reached here through the delivery's own deadline — deliberately, and

+ // not through a missing socket, which is NOT an error at this seam: the

+ // unix dialer retries a missing socket for a fixed 13s window

+ // (edge.dialUnixWithRetry, for an instance mid-spawn) and ReverseProxy

+ // then answers 502 itself, with no error for Deliver to see.

rt := registry.Route{Host: "gone.bab.oncarlos.com", Kind: "instance", Socket: filepath.Join(t.TempDir(), "missing.sock")}

req, err := schedule.BuildRequest(rt, registry.Timer{Name: "sync", Kind: schedule.KindSchedule, Path: "/jobs/sync", NextAt: 1, DueAt: 1}, "tok", "test")

if err != nil {

t.Fatal(err)

}

- // Short: the unix dialer retries a missing socket for the whole context

- // (an instance mid-spawn), and this one is never coming.

ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)

defer cancel()

if status, err := d.Deliver(ctx, rt, req.WithContext(ctx)); err == nil {

- t.Fatalf("delivery to a socket that does not exist = %d, want an error", status)

+ t.Fatalf("delivery that ran past its deadline = %d, want an error", status)

}

d.mu.Lock()

_, kept := d.proxies[rt.Socket]

@@ -450,3 +470,51 @@ func TestProxyCacheIsStableAndBounded(t *testing.T) {

t.Fatal("a failed delivery left its proxy (and its dead pool) in the cache")

}

+// The prune leg's eviction (schedule.SocketRetainer): a route that goes away

+// stops receiving deliveries, so nothing it caches can error its own way out

+// — the daily sweep's live-socket set is what reclaims it.

+func TestRetainSocketsDropsRetiredRoutes(t *testing.T) {

+ d := &activatorDeliverer{}

+ keep := d.proxyFor("/run/carlos/keep.sock")

+ d.proxyFor("/run/carlos/retired.sock")

+ d.RetainSockets(map[string]bool{"/run/carlos/keep.sock": true})

+ d.mu.Lock()

+ _, retired := d.proxies["/run/carlos/retired.sock"]

+ n := len(d.proxies)

+ d.mu.Unlock()

+ if retired || n != 1 {

+ t.Fatalf("after the sweep the cache holds %d entries (retired kept: %v), want just the live one", n, retired)

+ }

+ if again := d.proxyFor("/run/carlos/keep.sock"); again != keep {

+ t.Fatal("the sweep dropped a socket that is still routed here")

+ }

+ // An empty live set is a box with no exec routes at all: nothing to keep.

+ d.RetainSockets(map[string]bool{})

+ d.mu.Lock()

+ n = len(d.proxies)

+ d.mu.Unlock()

+ if n != 0 {

+ t.Fatalf("cache holds %d entries after an empty sweep, want 0", n)

+ }

+ // The runner must be able to find this method through the interface, or

+ // the sweep silently never runs.

+ if _, ok := any(d).(schedule.SocketRetainer); !ok {

+ t.Fatal("activatorDeliverer no longer satisfies schedule.SocketRetainer — the prune leg's eviction is dead code")

+ }

+// closeIdle returns a dropped proxy's pooled connections immediately instead

+// of waiting out the 90s IdleConnTimeout, and it can only do that through two

+// type assertions on what edge.UnixProxy builds. Pin the shape: if edge ever

+// returns something else, the eager close silently stops happening, and this

+// test is what says so.

+func TestUnixProxyKeepsTheShapeCloseIdleNeeds(t *testing.T) {

+ rp, ok := edge.UnixProxy("/run/carlos/x.sock").(*httputil.ReverseProxy)

+ if !ok {

+ t.Fatal("edge.UnixProxy no longer returns a *httputil.ReverseProxy — closeIdle is now a no-op")

+ }

+ if _, ok := rp.Transport.(*http.Transport); !ok {

+ t.Fatal("edge.UnixProxy's Transport is no longer an *http.Transport — closeIdle is now a no-op")

+ }

diff --git a/internal/host/schedule_wiring.go b/internal/host/schedule_wiring.go

index 0968e0def..a0af2ac4a 100644

--- a/internal/host/schedule_wiring.go

+++ b/internal/host/schedule_wiring.go

@@ -6,6 +6,7 @@ import (

"errors"

"log"

"net/http"

+ "net/http/httputil"

"os"

"path/filepath"

"strings"

@@ -170,6 +171,13 @@ func scheduleToken(dd string) func(host string) (string, error) {

// purpose: a pass reads one object per app plus the run signal, so minutes

// is far past any healthy bucket and still finite. The floors underneath

// (hourly, and the next serial move) are what recover a pass cut short.

+//

+// It bounds QUEUE PLUS WORK, not work alone: Pass takes the converger's

+// mutex first, so a caller behind a parked pass can spend the budget waiting

+// and then run with an expired context. Harmless — every app read fails,

+// convergeApp is die-free and leaves its rows alone, no etag is poisoned —

+// and it is the property that matters here: the leg returns, whatever the

+// bucket is doing.

const convergePassTimeout = 5 * time.Minute

// converge runs one converger pass, or nothing on a box with no deploy

@@ -222,12 +230,21 @@ type activatorDeliverer struct {

// Transport actually dials: a route re-pointed at a new socket gets a new

// entry rather than a pool aimed at the old file.

- // Bounded two ways: an entry is dropped when its delivery errors (a

- // socket that would not answer is exactly the one whose pooled

- // connections are worthless), and the whole map is dropped if it ever

- // passes maxProxyCache — this box has one socket per exec instance, so

- // the ceiling is only ever reached by a fleet far larger than one box

- // holds, and a rebuild costs one dial per live schedule.

+ // Bounded three ways, in the order they normally fire:

+ //

+ // - the runner's daily prune leg hands this the sockets still routed

+ // here (RetainSockets) and everything else goes. That is the one that

+ // reclaims a RETIRED route: it stops receiving deliveries, so it can

+ // never error its own way out.

+ // - an entry is dropped when its delivery comes back as an error — a

+ // refused wake or an expired deadline. (A socket that will not dial

+ // is NOT one of these: ReverseProxy answers that itself with a 502

+ // and no error. Nothing is pooled behind a failed dial anyway.)

+ // - the whole map is dropped if it ever passes maxProxyCache, a safety

+ // net rather than a working mechanism.

+ //

+ // Every one of those closes the dropped transports' idle connections

+ // rather than waiting out edge.UnixProxy's 90s IdleConnTimeout.

mu sync.Mutex

proxies map[string]http.Handler

}

@@ -247,6 +264,9 @@ func (d *activatorDeliverer) proxyFor(socket string) http.Handler {

if d.proxies == nil {

d.proxies = map[string]http.Handler{}

} else if len(d.proxies) >= maxProxyCache {

+ for _, p := range d.proxies {

+ closeIdle(p)

+ }

clear(d.proxies)

}

p := edge.UnixProxy(socket)

@@ -254,11 +274,44 @@ func (d *activatorDeliverer) proxyFor(socket string) http.Handler {

return p

}

-// dropProxy forgets socket's proxy, and with it its connection pool.

+// dropProxy forgets socket's proxy and closes the connections it was holding.

func (d *activatorDeliverer) dropProxy(socket string) {

d.mu.Lock()

defer d.mu.Unlock()

- delete(d.proxies, socket)

+ if p, ok := d.proxies[socket]; ok {

+ closeIdle(p)

+ delete(d.proxies, socket)

+ }

+// RetainSockets keeps only the sockets that are still routed on this box and

+// drops the rest — schedule.SocketRetainer, called by the runner's prune leg

+// with the host list it has already read. Without it a retired route's entry

+// would sit here until the maxProxyCache reset, since a route nothing

+// delivers to can never error its way out.

+func (d *activatorDeliverer) RetainSockets(live map[string]bool) {

+ d.mu.Lock()

+ defer d.mu.Unlock()

+ for socket, p := range d.proxies {

+ if !live[socket] {

+ closeIdle(p)

+ delete(d.proxies, socket)

+ }

+// closeIdle returns a dropped proxy's pooled connections now rather than at

+// the end of edge.UnixProxy's 90s IdleConnTimeout. Best-effort by design:

+// both assertions are guarded, so a change to the shape edge builds costs

+// the eager close and nothing else — the idle timeout is still the backstop.

+// (schedule_wake_test.go pins the shape, so such a change fails a test

+// rather than passing quietly.)

+func closeIdle(h http.Handler) {

+ if rp, ok := h.(*httputil.ReverseProxy); ok {

+ if tr, ok := rp.Transport.(*http.Transport); ok {

+ tr.CloseIdleConnections()

+ }

}

func (d *activatorDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) {

@@ -280,10 +333,12 @@ func (d *activatorDeliverer) Deliver(ctx context.Context, rt registry.Route, req

})

status, err := deliverVia(ctx, h, rt, req)

if err != nil {

- // Anything that did not come back as an HTTP status — a refused

- // wake, a socket that would not dial, a deadline — retires this

- // socket's pool. The next tick dials fresh, and a route that has

- // gone away leaves nothing behind.

+ // Anything that did not come back as an HTTP status — a wake the

+ // activator refused, a delivery that ran past its deadline — retires

+ // this socket's pool; the next tick dials fresh. Not the case a

+ // reader expects here: a socket that will not dial is ReverseProxy's

+ // own 502 with no error, and it has nothing pooled to retire anyway.

+ // Retired ROUTES are reclaimed by RetainSockets, not from here.

d.dropProxy(rt.Socket)

}

if err != nil && ctx.Err() == nil {

diff --git a/internal/registry/timers.go b/internal/registry/timers.go

index 311b5d65f..0d3bba0ef 100644

--- a/internal/registry/timers.go

+++ b/internal/registry/timers.go

@@ -159,8 +159,16 @@ func (r *Registry) CountTimers(host, kind string) (int, error) {

}

// PruneTimers: rows for hosts no longer routed here, and finished once-rows

-// older than doneBefore. An empty alive set deletes every row — callers pass

-// the registry's own host list, so that only happens on an empty registry.

+// older than doneBefore.

+//

+// An empty alive set deletes EVERY row. That is the right answer for a

+// genuinely empty registry — routes and timers live in the same file, so

+// every row really is orphaned — but it is a wipe, and there are two callers

+// now: schedule.Converger.Pass (every pass, on a box with a deploy store) and

+// schedule.Runner.prune (daily, on every box). Both build alive from the

+// registry's own route list; the runner declines the sweep outright when that

+// list is empty rather than hand the wipe over. A third caller should do the

+// same, or say in its own comment why the wipe is what it wants.

func (r *Registry) PruneTimers(alive map[string]bool, doneBefore int64) (int, error) {

hosts := make([]any, 0, len(alive))

for h := range alive {

diff --git a/internal/schedule/runner.go b/internal/schedule/runner.go

index 8ed11f2ba..e1e6501b8 100644

--- a/internal/schedule/runner.go

+++ b/internal/schedule/runner.go

@@ -36,10 +36,13 @@ var ErrTimeout = errors.New("schedule: delivery timed out")

// Deliverer sends one tick and reports the handler's status, or an error.

// Status 0 with an error means NO RESPONSE — the tick never got an HTTP

-// status line back (a wake the activator refused, a socket that would not

-// dial, a delivery that ran past its deadline). It does NOT mean the

-// instance is unreachable in any lasting sense: the commonest cause is a

-// wake still in progress, and the next attempt usually lands.

+// status line back. On the wake path that is a wake the activator refused

+// ("instance is waking", "failed to wake"), on any path a delivery that ran

+// past its deadline, and either way a handler that returned having written

+// nothing. It does NOT mean the instance is unreachable in any lasting

+// sense: the commonest cause is a wake still in progress, and the next

+// attempt usually lands. Nor is it every failure — a socket that will not

+// dial comes back as the proxy's own 502, a status like any other.

type Deliverer interface {

Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error)

}

@@ -223,20 +226,52 @@ func (r *Runner) prune(now time.Time) {

r.mu.Unlock()

routes, err := r.Reg.List()

if err != nil {

- log.Printf("schedule: prune: list routes: %v — leaving the rows alone", err)

+ log.Printf("schedule: runner prune: list routes: %v — leaving the rows alone", err)

return

}

alive := map[string]bool{}

+ sockets := map[string]bool{}

for _, rt := range routes {

alive[rt.Host] = true

+ if rt.Socket != "" {

+ sockets[rt.Socket] = true

+ }

+ // The sweep is also the delivery leg's eviction point: a route that goes

+ // away stops receiving deliveries, so nothing it caches for that route's

+ // socket would ever error and be dropped on its own. Optional — a

+ // Deliverer that caches nothing (every test's, and any future one) simply

+ // does not implement it.

+ if sr, ok := r.Deliver.(SocketRetainer); ok {

+ sr.RetainSockets(sockets)

+ }

+ if len(alive) == 0 {

+ // PruneTimers reads an empty alive set as "every host is gone" and

+ // deletes the whole table — correct for a genuinely empty registry

+ // (routes and timers live in the same file, so every row really is

+ // orphaned), but it is a wipe reachable from two callers now, and a

+ // caller that ever passed an empty set by mistake would take the

+ // table with it. This leg declines it explicitly: rows on a box with

+ // no routes are inert, and the next sweep after a route appears is

+ // the one that can tell orphans from arrivals.

+ log.Printf("schedule: runner prune: the registry has no routes — declining to sweep (that would delete every timer row)")

+ return

}

if n, err := r.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil {

- log.Printf("schedule: prune: %v", err)

+ log.Printf("schedule: runner prune: %v", err)

} else if n > 0 {

- log.Printf("schedule: pruned %d timer row(s)", n)

+ log.Printf("schedule: runner prune: pruned %d timer row(s)", n)

}

+// SocketRetainer is the optional half of Deliverer that keeps per-socket

+// state (a connection pool, in the agent's case). The runner's prune leg

+// hands it the set of sockets still routed on this box; everything else the

+// implementation holds is for a route that is gone.

+type SocketRetainer interface {

+ RetainSockets(live map[string]bool)

// claim marks host in flight; false if it already was.

func (r *Runner) claim(host string) bool {

r.mu.Lock()

diff --git a/internal/schedule/runner_test.go b/internal/schedule/runner_test.go

index 2f887c642..a6880a9f9 100644

--- a/internal/schedule/runner_test.go

+++ b/internal/schedule/runner_test.go

@@ -456,6 +456,57 @@ func TestRunnerPrunesWithoutAConverger(t *testing.T) {

}

+// retainingDeliverer is a Deliverer that also keeps per-socket state — the

+// agent's does (one connection pool per socket).

+type retainingDeliverer struct {

+ fakeDeliverer

+ mu sync.Mutex

+ calls []map[string]bool

+func (d *retainingDeliverer) RetainSockets(live map[string]bool) {

+ d.mu.Lock()

+ defer d.mu.Unlock()

+ d.calls = append(d.calls, live)

+func (d *retainingDeliverer) sweeps() []map[string]bool {

+ d.mu.Lock()

+ defer d.mu.Unlock()

+ return append([]map[string]bool(nil), d.calls...)

+// The prune leg doubles as the delivery leg's eviction point: it has already

+// read every route, so it hands the live SOCKET set to a Deliverer that keeps

+// per-socket state. A retired route can be reclaimed no other way — nothing

+// delivers to it, so nothing it holds can error its own way out.

+func TestPruneSweepHandsTheLiveSocketsToTheDeliverer(t *testing.T) {

+ d := &retainingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}}

+ reg, r := runnerFixture(t, d)

+ now := at("2026-08-23T13:07:00Z")

+ r.Now = func() time.Time { return now }

+ if err := reg.Upsert(registry.Route{Host: "a.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Socket: "/run/carlos/a.sock"}); err != nil {

+ t.Fatal(err)

+ }

+ if err := reg.Upsert(registry.Route{Host: "site.bab.oncarlos.com", Kind: "static", Account: "bab", App: "site", Status: "ready"}); err != nil {

+ t.Fatal(err)

+ }

+ r.tick()

+ sweeps := d.sweeps()

+ if len(sweeps) != 1 {

+ t.Fatalf("%d sweeps, want the one the prune leg runs", len(sweeps))

+ }

+ if !sweeps[0]["/run/carlos/a.sock"] || len(sweeps[0]) != 1 {

+ t.Fatalf("live sockets = %v, want exactly the routed instance's (a socketless route contributes nothing)", sweeps[0])

+ }

+ // Daily, with the prune itself: not once per tick.

+ now = now.Add(TickEvery)

+ r.tick()

+ if got := len(d.sweeps()); got != 1 {

+ t.Fatalf("%d sweeps after a second tick, want still 1", got)

+ }

// A busy host must not crowd another host's due timer out of the batch. The

// runner skips a host whose delivery is still running, so its rows are dead

// weight in a capped query: the exclusion belongs in DueTimers, before the

@@ -505,15 +556,18 @@ func TestBusyHostDoesNotStarveAnotherHostsTimer(t *testing.T) {

}

// countingDeliverer parks every delivery on a gate the test opens, and

-// records how many were in flight at once.

+// signals two things about how many ran at once: reached (exactly

+// MaxParallelHosts in flight) and over (MORE than that — the regression).

type countingDeliverer struct {

fakeDeliverer

- gate chan struct{}

- mu2 sync.Mutex

- cur int

- max int

- reached chan struct{}

- once sync.Once

+ gate chan struct{}

+ mu2 sync.Mutex

+ cur int

+ max int

+ reached chan struct{}

+ over chan struct{}

+ onceFull sync.Once

+ onceOver sync.Once

}

func (c *countingDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) {

@@ -522,10 +576,13 @@ func (c *countingDeliverer) Deliver(ctx context.Context, rt registry.Route, req

if c.cur > c.max {

c.max = c.cur

}

- full := c.cur == MaxParallelHosts

+ full, over := c.cur == MaxParallelHosts, c.cur > MaxParallelHosts

c.mu2.Unlock()

if full {

- c.once.Do(func() { close(c.reached) })

+ c.onceFull.Do(func() { close(c.reached) })

+ }

+ if over {

+ c.onceOver.Do(func() { close(c.over) })

}

<-c.gate

c.mu2.Lock()

@@ -540,12 +597,19 @@ func (c *countingDeliverer) peak() int {

return c.max

}

+// boundWindow is how long the bound is watched with every delivery parked.

+// It is a floor on how long we LOOK, never a deadline anything must meet:

+// both assertions inside it are negative (a fifth delivery must not start,

+// Tick must not return), so a slow machine can only make the test weaker,

+// never redder. A widened bound trips both within microseconds.

+const boundWindow = 500 * time.Millisecond

// The ≤4-parallel path: with more due hosts than MaxParallelHosts, exactly

-// four deliveries run at once — no more (the semaphore is the bound) and no

-// fewer (they are genuinely parallel, not one host at a time) — and every

+// four deliveries run at once — no fewer (they are genuinely parallel, not

+// one host at a time) and no more (the semaphore is the bound) — and every

// host still completes once the slots recycle.

func TestDeliveriesAreParallelToTheBoundAndAllComplete(t *testing.T) {

- d := &countingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, gate: make(chan struct{}), reached: make(chan struct{})}

+ d := &countingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, gate: make(chan struct{}), reached: make(chan struct{}), over: make(chan struct{})}

reg, r := runnerFixture(t, d)

now := at("2026-08-23T13:07:00Z")

r.Now = func() time.Time { return now }

@@ -567,6 +631,27 @@ func TestDeliveriesAreParallelToTheBoundAndAllComplete(t *testing.T) {

case <-time.After(10 * time.Second):

t.Fatalf("only %d deliveries ever ran at once, want %d in parallel", d.peak(), MaxParallelHosts)

}

+ // The BOUND, watched while the gate is still shut and all four slots are

+ // held. Two independent ways a widened semaphore shows itself, and the

+ // test must be red on either:

+ //

+ // - a fifth delivery starts (`over`);

+ // - Tick RETURNS. With 8 hosts and 4 slots it cannot: it takes the slot

+ // before launching each host's goroutine, so it is parked on the

+ // semaphore until a delivery finishes — and none can while the gate

+ // is shut.

+ //

+ // Watching the peak after the fact cannot do this job: opening the gate

+ // drains `cur` faster than a fifth delivery would ever be observed, which

+ // is why the earlier shape passed 25/25 with the semaphore quadrupled.

+ select {

+ case <-d.over:

+ t.Fatalf("%d deliveries in flight at once, want at most MaxParallelHosts (%d)", d.peak(), MaxParallelHosts)

+ case <-tickDone:

+ t.Fatalf("Tick returned with every delivery still parked — it launched more than MaxParallelHosts (%d) hosts (peak %d)", MaxParallelHosts, d.peak())

+ case <-time.After(boundWindow):

+ }

close(d.gate) // every parked delivery finishes; the waiting hosts take the slots

select {

case <-tickDone:

@@ -574,7 +659,7 @@ func TestDeliveriesAreParallelToTheBoundAndAllComplete(t *testing.T) {

t.Fatal("Tick never returned after the gate opened")

}

r.wait()

- if got := d.peak(); got > MaxParallelHosts {

+ if got := d.peak(); got > MaxParallelHosts { // belt-and-braces; the window above is the real assertion

t.Fatalf("%d deliveries in flight at once, want at most MaxParallelHosts (%d)", got, MaxParallelHosts)

}

for _, h := range names {