amadan
schedule/registry: fair due batch, a prune leg the runner owns, honest status-0
Three deferred review findings on the runner, all of them about a box the converger never runs on: - DueTimers grows an excludeHosts filter and the runner passes its busy set. The batch is capped at 64 rows and the runner skips a host whose delivery is still running, so a host parked in a long handler (or one with more due rows than the batch holds) filled every batch with rows the tick was about to skip — another host's due timer was never SEEN. The filter belongs in the query, before the LIMIT; the runner-side skip stays as belt-and-braces for a host that goes busy between the two. - The runner gets its own prune leg. The 7-day sweep of finished once-rows rode the converger, and the converger needs the deploy store — so a box with the activator armed and no store (the one whose rows arrive on the control socket, spec §7.2) never pruned at all, and its finished rows could pin an app against the 1000-per-host cap. Daily, stamped on the 15s tick rather than a second goroutine. - Logging and comments: a status report that cannot be built is logged instead of dropped, and the Deliverer contract now says what status 0 means — no response / transport error, not "the instance is unreachable". Tests: the two behaviours above both fail without their fix (checked by reverting each); the >4-due-host case pins bounded parallelism AND completion; and the concurrent-callers exactly-once test now counts deliveries through a real Runner instead of inferring from next_at.
5 files changed,
+356
−24
internal/registry/timers.go+22 −2internal/registry/timers_test.go+42 −3internal/schedule/converge_test.go+19 −5internal/schedule/runner.go+86 −4internal/schedule/runner_test.go+187 −10
diff --git a/internal/registry/timers.go b/internal/registry/timers.go| index ce4b22dc8..311b5d65f 100644 |
| --- a/internal/registry/timers.go |
| +++ b/internal/registry/timers.go |
| @@ -102,8 +102,28 @@ func (r *Registry) TimersFor(host string) ([]Timer, error) { |
| return r.queryTimers(`SELECT `+timerCols+` FROM timers WHERE host=? ORDER BY name`, host) |
| } |
| -func (r *Registry) DueTimers(now int64, limit int) ([]Timer, error) { |
| - return r.queryTimers(`SELECT `+timerCols+` FROM timers WHERE next_at > 0 AND next_at <= ? ORDER BY next_at, host, name LIMIT ?`, now, limit) |
| +// DueTimers is the runner's batch: rows due at or before now, earliest |
| +// first, at most limit of them. |
| +// |
| +// excludeHosts is what keeps the batch FAIR. The runner delivers one timer |
| +// at a time per host and skips a host whose delivery is still running, so a |
| +// host with more due rows than the batch holds — or one parked in a |
| +// 30-minute handler — would otherwise fill every slot of every batch with |
| +// rows the runner is about to skip, and another host's due timer would never |
| +// be seen at all. Passing the busy set here moves the filter into the query, |
| +// 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) { |
| + q := `SELECT ` + timerCols + ` FROM timers WHERE next_at > 0 AND next_at <= ?` |
| + args := []any{now} |
| + if len(excludeHosts) > 0 { |
| + q += ` AND host NOT IN (?` + strings.Repeat(",?", len(excludeHosts)-1) + `)` |
| + for _, h := range excludeHosts { |
| + args = append(args, h) |
| + } |
| + } |
| + q += ` ORDER BY next_at, host, name LIMIT ?` |
| + return r.queryTimers(q, append(args, limit)...) |
| } |
| // FireTimerNow makes the row due at now, and now IS the occurrence being |
diff --git a/internal/registry/timers_test.go b/internal/registry/timers_test.go| index 3d200c72e..3e68a127d 100644 |
| --- a/internal/registry/timers_test.go |
| +++ b/internal/registry/timers_test.go |
| @@ -63,20 +63,20 @@ func TestTimersDueOrderingAndFireNow(t *testing.T) { |
| t.Fatal(err) |
| } |
| } |
| - due, err := r.DueTimers(600, 10) |
| + due, err := r.DueTimers(600, 10, nil) |
| if err != nil { |
| t.Fatal(err) |
| } |
| if len(due) != 2 || due[0].Name != "soon" || due[1].Name != "late" { |
| t.Fatalf("due = %+v", due) |
| } |
| - if due, _ := r.DueTimers(600, 1); len(due) != 1 { |
| + if due, _ := r.DueTimers(600, 1, nil); len(due) != 1 { |
| t.Fatal("limit ignored") |
| } |
| if err := r.FireTimerNow("c", "future", 600); err != nil { |
| t.Fatal(err) |
| } |
| - if due, _ := r.DueTimers(600, 10); len(due) != 3 { |
| + if due, _ := r.DueTimers(600, 10, []string{}); len(due) != 3 { // an empty exclude set is no filter |
| t.Fatalf("after FireTimerNow: %d due", len(due)) |
| } |
| if err := r.FireTimerNow("nope", "x", 600); err != nil { |
| @@ -210,3 +210,42 @@ func TestTimersPreDueAtRegistryGainsTheColumn(t *testing.T) { |
| } |
| } |
| } |
| + |
| +// The batch is capped, and the runner skips hosts whose delivery is still |
| +// running: without the exclude filter a busy host with a batch's worth of due |
| +// rows takes every slot of every batch and another host's due timer is never |
| +// SEEN, let alone fired. The filter belongs in the query, before the LIMIT. |
| +func TestDueTimersExcludesBusyHostsBeforeTheLimit(t *testing.T) { |
| + r := openTimersReg(t) |
| + for i := 0; i < 8; i++ { |
| + if err := r.UpsertTimer(Timer{Host: "busy", Name: "j" + string(rune('a'+i)), Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 100}); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + // Due LATER than every one of the busy host's rows, so ordering alone |
| + // never rescues it. |
| + if err := r.UpsertTimer(Timer{Host: "quiet", Name: "j", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 200}); err != nil { |
| + t.Fatal(err) |
| + } |
| + due, err := r.DueTimers(600, 4, nil) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, tm := range due { |
| + if tm.Host == "quiet" { |
| + t.Fatalf("fixture: the quiet host must be crowded out without the filter, got %+v", due) |
| + } |
| + } |
| + due, err = r.DueTimers(600, 4, []string{"busy"}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(due) != 1 || due[0].Host != "quiet" { |
| + t.Fatalf("excluding the busy host = %+v, want the quiet host's one row", due) |
| + } |
| + // Every host excluded is an empty batch, not a filter that silently |
| + // stops applying. |
| + if due, _ := r.DueTimers(600, 4, []string{"busy", "quiet"}); len(due) != 0 { |
| + t.Fatalf("all hosts excluded = %+v, want none", due) |
| + } |
| +} |
diff --git a/internal/schedule/converge_test.go b/internal/schedule/converge_test.go| index 5f64f4e28..d3ee5093e 100644 |
| --- a/internal/schedule/converge_test.go |
| +++ b/internal/schedule/converge_test.go |
| @@ -336,13 +336,27 @@ func TestConvergePassIsSafeUnderConcurrentCallers(t *testing.T) { |
| t.Fatalf("%s: run signal not fired: %+v", h, tm) |
| } |
| } |
| - // Exactly-once: the etag was recorded, so the next pass is a no-op. |
| - reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 1, 0, 777, 777) |
| + // Exactly-once, COUNTED: a run signal that fired twice is one that got |
| + // delivered twice, so the assertion runs the rows through a real Runner |
| + // and counts the deliveries rather than inferring from next_at. Four |
| + // racing passes, one tick each host — and a later pass, which sees the |
| + // recorded etag, adds no second delivery. |
| + d := &fakeDeliverer{status: 204} |
| + r := &Runner{Reg: reg, Deliver: d, Timeout: time.Minute, Box: "box1", Build: "test", |
| + Now: func() time.Time { return now }, Token: func(string) (string, error) { return "tok", nil }} |
| + r.tick() |
| + fired := d.firedHosts() |
| + for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} { |
| + if fired[h] != 1 { |
| + t.Fatalf("%s delivered %d times, want exactly 1 (fired = %v)", h, fired[h], fired) |
| + } |
| + } |
| if err := c.Pass(ctx); err != nil { |
| t.Fatal(err) |
| } |
| - if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 { |
| - t.Fatal("the run signal replayed after the concurrent passes") |
| + r.tick() |
| + if got := d.firedHosts(); len(got) != len(fired) || got["a.bab.oncarlos.com"] != 1 || got["b.bab.oncarlos.com"] != 1 { |
| + t.Fatalf("the run signal replayed after the concurrent passes: deliveries now %v, were %v", got, fired) |
| } |
| if len(reports) == 0 { |
| t.Fatal("no reports at all — the fixture converged nothing") |
| @@ -512,7 +526,7 @@ func TestConvergeParksAScheduleThatNeverFires(t *testing.T) { |
| if tm.NextAt != 0 || tm.DueAt != 0 || tm.LastStatus != StatusNeverFires { |
| t.Fatalf("parked row = %+v, want next_at 0 and %q", tm, StatusNeverFires) |
| } |
| - if due, _ := reg.DueTimers(now.Unix(), 10); len(due) != 0 { |
| + if due, _ := reg.DueTimers(now.Unix(), 10, nil); len(due) != 0 { |
| t.Fatalf("a parked row is due: %+v", due) |
| } |
| } |
diff --git a/internal/schedule/runner.go b/internal/schedule/runner.go| index c366ba1ac..8ed11f2ba 100644 |
| --- a/internal/schedule/runner.go |
| +++ b/internal/schedule/runner.go |
| @@ -22,6 +22,10 @@ const ( |
| dueBatch = 64 |
| backoffBase = time.Minute |
| backoffCap = 15 * time.Minute |
| + // pruneEvery is the prune leg's cadence — see Runner.prune. Daily |
| + // because the thing it deletes is a 7-day-old finished row: a sweep an |
| + // hour earlier or later changes nothing a member can see. |
| + pruneEvery = 24 * time.Hour |
| ) |
| // ErrTimeout is what a Deliverer returns when the handler ran past the |
| @@ -29,8 +33,13 @@ const ( |
| // may have finished, or still be running — a retry would double it. |
| var ErrTimeout = errors.New("schedule: delivery timed out") |
| -// Deliverer sends one tick and reports the handler's status, or a |
| -// transport-level error (wake failed, connect refused, ErrTimeout). |
| +// 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. |
| type Deliverer interface { |
| Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) |
| } |
| @@ -63,6 +72,9 @@ type Runner struct { |
| busy map[string]bool |
| sem chan struct{} |
| inflight sync.WaitGroup |
| + |
| + // lastPruned is the prune leg's stamp — see prune. |
| + lastPruned time.Time |
| } |
| // Start runs Tick every TickEvery until ctx ends. |
| @@ -100,7 +112,15 @@ func (r *Runner) timeout() time.Duration { |
| // ticks), timers within a host one after another. Tick returns once every |
| // goroutine is launched — it never waits for a delivery. |
| func (r *Runner) Tick(ctx context.Context) { |
| - due, err := r.Reg.DueTimers(r.now().Unix(), dueBatch) |
| + now := r.now() |
| + r.prune(now) |
| + // Busy hosts are excluded IN THE QUERY, not after it. The batch is |
| + // capped at dueBatch rows: a host parked in a long handler (or one |
| + // holding more due rows than the batch) would otherwise fill the batch |
| + // 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()) |
| if err != nil { |
| log.Printf("schedule: due query: %v", err) |
| return |
| @@ -150,7 +170,12 @@ func (r *Runner) Tick(ctx context.Context) { |
| r.fire(ctx, rt, t) |
| } |
| if r.Report != nil { |
| - if st, err := StatusFor(r.Reg, rt, r.Box, r.now()); err == nil { |
| + // A report that cannot be built is a silent hole in `carlos |
| + // schedule ls` — the member's only view of this box. Say so: |
| + // the fires themselves still landed on the rows. |
| + if st, err := StatusFor(r.Reg, rt, r.Box, r.now()); err != nil { |
| + log.Printf("schedule: %s: build status report: %v", rt.Host, err) |
| + } else { |
| r.Report(rt, st) |
| } |
| } |
| @@ -158,6 +183,60 @@ func (r *Runner) Tick(ctx context.Context) { |
| } |
| } |
| +// busyHosts is the in-flight set as DueTimers wants it. A snapshot: a host |
| +// that goes busy after this read is caught by claim instead. |
| +func (r *Runner) busyHosts() []string { |
| + r.mu.Lock() |
| + defer r.mu.Unlock() |
| + if len(r.busy) == 0 { |
| + return nil |
| + } |
| + hosts := make([]string, 0, len(r.busy)) |
| + for h := range r.busy { |
| + hosts = append(hosts, h) |
| + } |
| + return hosts |
| +} |
| + |
| +// prune is the runner's OWN sweep of dead timer rows, and it exists because |
| +// the converger's identical sweep is not always there: the converger needs |
| +// the deploy store, and a box with the activator armed and no store still |
| +// accumulates rows — once-timers registered by instances on their control |
| +// socket (spec §7.2). Without this leg those finished rows would sit in the |
| +// table forever and could pin an app against the 1000-per-host cap on the |
| +// one kind of box that has no other way to shed them. |
| +// |
| +// Daily, stamped rather than scheduled: it rides the 15s tick so there is no |
| +// second goroutine to stop, and the first tick after start prunes once. |
| +// Cheap either way (one DELETE over an indexed table), and idempotent with |
| +// the converger's sweep on a box that has both. |
| +func (r *Runner) prune(now time.Time) { |
| + r.mu.Lock() |
| + if !r.lastPruned.IsZero() && now.Sub(r.lastPruned) < pruneEvery { |
| + r.mu.Unlock() |
| + return |
| + } |
| + // Stamped before the work, not after: a registry that cannot be read is |
| + // not worth retrying every 15s, and the rows a failed sweep would have |
| + // dropped are inert until it next runs. |
| + r.lastPruned = now |
| + r.mu.Unlock() |
| + routes, err := r.Reg.List() |
| + if err != nil { |
| + log.Printf("schedule: prune: list routes: %v — leaving the rows alone", err) |
| + return |
| + } |
| + alive := map[string]bool{} |
| + for _, rt := range routes { |
| + alive[rt.Host] = true |
| + } |
| + if n, err := r.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil { |
| + log.Printf("schedule: prune: %v", err) |
| + } else if n > 0 { |
| + log.Printf("schedule: pruned %d timer row(s)", n) |
| + } |
| +} |
| + |
| // claim marks host in flight; false if it already was. |
| func (r *Runner) claim(host string) bool { |
| r.mu.Lock() |
| @@ -290,6 +369,9 @@ func classify(status int, err error, timeout time.Duration) Outcome { |
| case errors.Is(err, ErrTimeout): |
| return Outcome{Status: "timeout after " + timeout.String()} |
| case err != nil: |
| + // No response / transport error (status 0). Recorded as 503 because |
| + // that is what a member's own client would have seen, and retried |
| + // for the same reason: nothing says the handler ran. |
| return Outcome{Status: "503: " + err.Error(), Retry: true} |
| case status >= 200 && status < 300: |
| return Outcome{Status: strconv.Itoa(status)} |
diff --git a/internal/schedule/runner_test.go b/internal/schedule/runner_test.go| index 3363ffccd..01056be33 100644 |
| --- a/internal/schedule/runner_test.go |
| +++ b/internal/schedule/runner_test.go |
| @@ -3,6 +3,7 @@ package schedule |
| import ( |
| "context" |
| "errors" |
| + "fmt" |
| "net/http" |
| "path/filepath" |
| "strings" |
| @@ -29,6 +30,18 @@ func (f *fakeDeliverer) Deliver(_ context.Context, rt registry.Route, req *http. |
| return f.status, f.err |
| } |
| +// firedHosts counts deliveries per host — the counting assertion the |
| +// parallelism and exactly-once tests need. |
| +func (f *fakeDeliverer) firedHosts() map[string]int { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + n := map[string]int{} |
| + for _, h := range f.hosts { |
| + n[h]++ |
| + } |
| + return n |
| +} |
| + |
| func runnerFixture(t *testing.T, d Deliverer) (*registry.Registry, *Runner) { |
| t.Helper() |
| reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db")) |
| @@ -307,16 +320,6 @@ func (b *blockingDeliverer) Deliver(ctx context.Context, rt registry.Route, req |
| return b.fakeDeliverer.Deliver(ctx, rt, req) |
| } |
| -func (b *blockingDeliverer) firedHosts() map[string]int { |
| - b.mu.Lock() |
| - defer b.mu.Unlock() |
| - n := map[string]int{} |
| - for _, h := range b.hosts { |
| - n[h]++ |
| - } |
| - return n |
| -} |
| - |
| // One long handler must not hold every other host's timers: the tick |
| // launches per-host work and returns; while one host is parked, the others |
| // all fire; and a later tick does NOT start a second delivery to the parked |
| @@ -388,3 +391,177 @@ func TestOneSlowHostDoesNotBlockTheOthersOrRefire(t *testing.T) { |
| t.Fatalf("released host delivered %d times, want 2", n) |
| } |
| } |
| + |
| +// The prune leg is the RUNNER's, not the converger's, because a box with the |
| +// activator armed and no deploy store has no converger at all (spec §6.2) — |
| +// and it is exactly the box whose rows arrive on the control socket. Without |
| +// this leg its finished once-rows accumulate forever. |
| +func TestRunnerPrunesWithoutAConverger(t *testing.T) { |
| + d := &fakeDeliverer{status: 204} |
| + reg, r := runnerFixture(t, d) |
| + if r.Report != nil { |
| + t.Fatal("fixture has a reporter — this test is the storeless box") |
| + } |
| + now := at("2026-08-23T13:07:00Z") |
| + r.Now = func() time.Time { return now } |
| + const host = "a.bab.oncarlos.com" |
| + old := now.Add(-8 * 24 * time.Hour).Unix() |
| + reg.UpsertTimer(registry.Timer{Host: host, Name: "done-long-ago", Kind: KindOnce, Spec: "at 1", Path: "/o", NextAt: 0, LastAt: old}) |
| + reg.UpsertTimer(registry.Timer{Host: host, Name: "done-just-now", Kind: KindOnce, Spec: "at 1", Path: "/o", NextAt: 0, LastAt: now.Add(-time.Hour).Unix()}) |
| + reg.UpsertTimer(registry.Timer{Host: "ghost.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/s", NextAt: 0, LastAt: now.Unix()}) |
| + r.tick() |
| + if tm, _ := reg.GetTimer(host, "done-long-ago"); tm != nil { |
| + t.Fatalf("a finished once-row 8 days old survived the runner's prune: %+v", tm) |
| + } |
| + if tm, _ := reg.GetTimer("ghost.bab.oncarlos.com", "s"); tm != nil { |
| + t.Fatalf("a row for a host with no route survived: %+v", tm) |
| + } |
| + if tm, _ := reg.GetTimer(host, "done-just-now"); tm == nil { |
| + t.Fatal("a finished once-row inside the retention window was pruned") |
| + } |
| + |
| + // Daily, not every tick: the stamp is what keeps a 15s tick from running |
| + // a DELETE 5,760 times a day. |
| + reg.UpsertTimer(registry.Timer{Host: "ghost2.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/s", NextAt: 0, LastAt: now.Unix()}) |
| + now = now.Add(TickEvery) |
| + r.tick() |
| + if tm, _ := reg.GetTimer("ghost2.bab.oncarlos.com", "s"); tm == nil { |
| + t.Fatal("the prune leg ran again 15s later — the stamp is not holding") |
| + } |
| + now = now.Add(pruneEvery) |
| + r.tick() |
| + if tm, _ := reg.GetTimer("ghost2.bab.oncarlos.com", "s"); tm != nil { |
| + t.Fatalf("the prune leg never came round again: %+v", tm) |
| + } |
| +} |
| + |
| +// 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 |
| +// LIMIT. Sized to the real batch so the test fails if dueBatch and the |
| +// fixture ever drift apart. |
| +func TestBusyHostDoesNotStarveAnotherHostsTimer(t *testing.T) { |
| + d := &blockingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, slow: "slow.bab.oncarlos.com", gate: make(chan struct{})} |
| + reg, r := runnerFixture(t, d) |
| + now := at("2026-08-23T13:07:00Z") |
| + r.Now = func() time.Time { return now } |
| + for _, h := range []string{d.slow, "quiet.bab.oncarlos.com"} { |
| + if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + // A full batch of rows on the parked host, every one of them due BEFORE |
| + // the quiet host's single row. |
| + for i := 0; i < dueBatch; i++ { |
| + reg.UpsertTimer(registry.Timer{Host: d.slow, 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.bab.oncarlos.com", Name: "j", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 1}) |
| + |
| + // First tick: the parked host claims a slot and stops there. Tick claims |
| + // synchronously, so the host is busy the moment it returns. |
| + r.Tick(context.Background()) |
| + if busy := r.busyHosts(); len(busy) != 1 || busy[0] != d.slow { |
| + t.Fatalf("after the first tick busy = %v, want just the parked host", busy) |
| + } |
| + |
| + // Second tick, while it is still parked: the quiet host's row is due and |
| + // must be delivered and recorded. |
| + now = now.Add(TickEvery) |
| + r.Tick(context.Background()) |
| + deadline := time.Now().Add(10 * time.Second) |
| + for time.Now().Before(deadline) { |
| + if tm, _ := reg.GetTimer("quiet.bab.oncarlos.com", "j"); tm != nil && tm.LastStatus == "204" { |
| + break |
| + } |
| + time.Sleep(time.Millisecond) |
| + } |
| + if tm, _ := reg.GetTimer("quiet.bab.oncarlos.com", "j"); tm == nil || tm.LastStatus != "204" { |
| + t.Fatalf("the quiet host's timer never fired behind a full batch of a busy host's rows: %+v", tm) |
| + } |
| + close(d.gate) |
| + r.wait() |
| +} |
| + |
| +// countingDeliverer parks every delivery on a gate the test opens, and |
| +// records how many were in flight at once. |
| +type countingDeliverer struct { |
| + fakeDeliverer |
| + gate chan struct{} |
| + mu2 sync.Mutex |
| + cur int |
| + max int |
| + reached chan struct{} |
| + once sync.Once |
| +} |
| + |
| +func (c *countingDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) { |
| + c.mu2.Lock() |
| + c.cur++ |
| + if c.cur > c.max { |
| + c.max = c.cur |
| + } |
| + full := c.cur == MaxParallelHosts |
| + c.mu2.Unlock() |
| + if full { |
| + c.once.Do(func() { close(c.reached) }) |
| + } |
| + <-c.gate |
| + c.mu2.Lock() |
| + c.cur-- |
| + c.mu2.Unlock() |
| + return c.fakeDeliverer.Deliver(ctx, rt, req) |
| +} |
| + |
| +func (c *countingDeliverer) peak() int { |
| + c.mu2.Lock() |
| + defer c.mu2.Unlock() |
| + return c.max |
| +} |
| + |
| +// 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 |
| +// 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{})} |
| + reg, r := runnerFixture(t, d) |
| + now := at("2026-08-23T13:07:00Z") |
| + r.Now = func() time.Time { return now } |
| + const hosts = 2 * MaxParallelHosts |
| + var names []string |
| + for i := 0; i < hosts; i++ { |
| + h := fmt.Sprintf("h%d.bab.oncarlos.com", i) |
| + names = append(names, h) |
| + if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + reg.UpsertTimer(registry.Timer{Host: h, Name: "j", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 100 + int64(i)}) |
| + } |
| + |
| + tickDone := make(chan struct{}) |
| + go func() { r.Tick(context.Background()); close(tickDone) }() |
| + select { |
| + case <-d.reached: |
| + case <-time.After(10 * time.Second): |
| + t.Fatalf("only %d deliveries ever ran at once, want %d in parallel", d.peak(), MaxParallelHosts) |
| + } |
| + close(d.gate) // every parked delivery finishes; the waiting hosts take the slots |
| + select { |
| + case <-tickDone: |
| + case <-time.After(10 * time.Second): |
| + t.Fatal("Tick never returned after the gate opened") |
| + } |
| + r.wait() |
| + if got := d.peak(); got > MaxParallelHosts { |
| + t.Fatalf("%d deliveries in flight at once, want at most MaxParallelHosts (%d)", got, MaxParallelHosts) |
| + } |
| + for _, h := range names { |
| + tm, _ := reg.GetTimer(h, "j") |
| + if tm == nil || tm.LastStatus != "204" { |
| + t.Fatalf("%s did not complete: %+v", h, tm) |
| + } |
| + } |
| + if fired := d.firedHosts(); len(fired) != hosts { |
| + t.Fatalf("delivered to %d hosts, want %d", len(fired), hosts) |
| + } |
| +} |