amadan

carlos / platform Public

registry: the due-timer queries leave out gated routes before the limit (final fix F5)

The runner skips an uncommitted import's timers without advancing them,
and only after the query's LIMIT. Four unfinished imports with 16 due
schedules each filled all 64 rows of every batch, so no other host's
timer was ever selected.

Both due-timer queries now exclude rows whose route has an import gate,
in the query, before the LIMIT. The runner keeps its own check. A row
whose host has no route is still returned.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 579c15823dfd7f67bf4397dea2a3a3063f5bd527 parent dffb1df
3 files changed, +105 −1
  • internal/registry/timers.go +8 −1
  • internal/registry/timers_test.go +47 −0
  • internal/schedule/runner_test.go +50 −0
diff --git a/internal/registry/timers.go b/internal/registry/timers.go
index 8ef7bd7cb..0da19d9b2 100644
--- a/internal/registry/timers.go
+++ b/internal/registry/timers.go
@@ -225,8 +225,15 @@ func (r *Registry) DueUnheldTimers(now int64, limit int, excludeHosts []string)
return r.dueTimers(now, limit, excludeHosts, true)
}
+// dueTimers is both queries. Neither returns the rows of a gated route (an
+// uncommitted import's, seeded-instances spec §5.4), and the exclusion is
+// in the query, before the LIMIT (final review F5): the runner skips such a
+// row without advancing it, so gated rows left in the batch would stay due
+// for ever and could fill every slot of it, hiding every other host's due
+// timer. The runner keeps its own check.
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 <= ?`
+ q := `SELECT ` + timerCols + ` FROM timers WHERE next_at > 0 AND next_at <= ?` +
+ ` AND host NOT IN (SELECT host FROM routes WHERE import_gate != '')`
args := []any{now}
if skipHeld {
statuses := make([]string, 0, len(HoldStatuses))
diff --git a/internal/registry/timers_test.go b/internal/registry/timers_test.go
index 808ac5a7c..465ff1868 100644
--- a/internal/registry/timers_test.go
+++ b/internal/registry/timers_test.go
@@ -3,6 +3,7 @@ package registry
import (
"database/sql"
"errors"
+ "fmt"
"os"
"path/filepath"
"reflect"
@@ -590,3 +591,49 @@ func TestDueUnheldTimersExcludesHeldHostsBeforeTheLimit(t *testing.T) {
t.Fatalf("after the release DueUnheldTimers = %+v", un)
}
}
+
+// Final review F5: both due queries leave out every row of a gated route
+// (an uncommitted import's) before the limit. The runner skips such a row
+// without advancing it, so 64 of them, due first, would otherwise fill the
+// runner's whole batch on every tick and hide another host's due timer for
+// ever. A row whose host has no route is still returned.
+func TestDueTimersExcludeGatedRoutesBeforeTheLimit(t *testing.T) {
+ r := openTimersReg(t)
+ const batch = 64
+ for i := range 4 {
+ host := fmt.Sprintf("gated%d.test", i)
+ if err := r.Upsert(Route{Host: host, Kind: "instance", Socket: "/s", ImportGate: ImportGateClosed}); err != nil {
+ t.Fatal(err)
+ }
+ for n := range batch / 4 {
+ if err := r.UpsertTimer(Timer{Host: host, Name: fmt.Sprintf("t%d", n), Kind: "schedule", Spec: "every 1h0m0s", Path: "/j",
+ NextAt: int64(10 + n)}); err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+ if err := r.Upsert(Route{Host: "free.test", Kind: "instance", Socket: "/s"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.UpsertTimer(Timer{Host: "free.test", Name: "t", Kind: "schedule", Spec: "every 1h0m0s", Path: "/j", NextAt: 90}); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.UpsertTimer(Timer{Host: "gone.test", Name: "t", Kind: "schedule", Spec: "every 1h0m0s", Path: "/j", NextAt: 95}); err != nil {
+ t.Fatal(err)
+ }
+ for name, due := range map[string]func(int64, int, []string) ([]Timer, error){
+ "DueTimers": r.DueTimers, "DueUnheldTimers": r.DueUnheldTimers,
+ } {
+ got, err := due(100, batch, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var hosts []string
+ for _, tm := range got {
+ hosts = append(hosts, tm.Host)
+ }
+ if len(hosts) != 2 || hosts[0] != "free.test" || hosts[1] != "gone.test" {
+ t.Fatalf("%s hosts = %q, want free.test then gone.test", name, hosts)
+ }
+ }
+}
diff --git a/internal/schedule/runner_test.go b/internal/schedule/runner_test.go
index 2e716c700..155e2cb39 100644
--- a/internal/schedule/runner_test.go
+++ b/internal/schedule/runner_test.go
@@ -1350,3 +1350,53 @@ func TestGateLandingBeforeTheClaimGetsNoTick(t *testing.T) {
t.Fatal("the tick kept its claim after seeing the gate")
}
}
+
+// Final review F5: 64 gated due timers (four uncommitted imports with 16
+// each, due first) and one ordinary due timer on another host. The batch
+// leaves the gated rows out before its limit, so the ordinary timer is
+// delivered on the first tick, and the gated rows' bookkeeping does not
+// move.
+func TestGatedTimersDoNotStarveAnotherHost(t *testing.T) {
+ for _, armed := range []bool{false, true} {
+ t.Run(fmt.Sprintf("holds armed %v", armed), func(t *testing.T) {
+ reg, r := runnerFixture(t, nil)
+ now := at("2026-08-23T13:07:00Z")
+ r.Now = func() time.Time { return now }
+ if armed {
+ r.Gate = &HostGate{}
+ }
+ var delivered []string
+ r.Deliver = deliverFunc(func(rt registry.Route, _ *http.Request) (int, error) {
+ delivered = append(delivered, rt.Host)
+ return 204, nil
+ })
+ var gated []string
+ for i := range 4 {
+ host := fmt.Sprintf("g%d.bab.oncarlos.com", i)
+ gated = append(gated, host)
+ if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true}); err != nil {
+ t.Fatal(err)
+ }
+ gateForImport(t, reg, host)
+ for n := range dueBatch / 4 {
+ reg.UpsertTimer(registry.Timer{Host: host, Name: fmt.Sprintf("t%d", n), Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: int64(1 + n)})
+ }
+ }
+ reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 100})
+ before := map[string][]registry.Timer{}
+ for _, h := range gated {
+ before[h] = timersOf(t, reg, h)
+ }
+ r.tick()
+ r.wait()
+ if len(delivered) != 1 || delivered[0] != holdHost {
+ t.Fatalf("delivered %q, want the ordinary host's timer alone", delivered)
+ }
+ for _, h := range gated {
+ if got := timersOf(t, reg, h); !sameTimers(got, before[h]) {
+ t.Fatalf("%s's gated rows moved:\n got %+v\nwant %+v", h, got, before[h])
+ }
+ }
+ })
+ }
+}