amadan
schedule: occurrence key on the wire, detached per-host delivery, runner minors
I3: X-Carlos-Schedule-At now carries the OCCURRENCE, not next_at. Additive registry column timers.due_at (ALTER guarded like the routes list, idempotent on re-open; pre-column rows read 0 and fall back to next_at). UpsertTimer moves due_at with next_at; MarkTimerFired takes an explicit due (the new occurrence on advance, the current one on a retry); FireTimerNow makes now the occurrence. TimerStatus gains due_at. I2: Tick launches one goroutine per host with due rows and returns without waiting; a busy map keeps per-host serialization across ticks, and the MaxParallelHosts semaphore bounds hosts in flight across ALL ticks. wait() is the tests' seam. Covered by a 7-host test with one parked host, under -race. Minors: WHY on the retry clamp; a registry error in Tick is logged, not read as "route gone"; BuildRequest refuses a path the parser refuses (the runner records "bad path: …", no retry) and ValidPath runs the same parser; a cron that never matches is refused by Schedule.Validate and, if one slips past, parked at next_at=0 with last_status "never fires" instead of a row due at -62135596800; `carlos schedule ls` labels once rows "(once)" and shows a parked row's condition.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
13 files changed,
+584
−67
internal/registry/registry.go+7 −0internal/registry/timers.go+36 −13internal/registry/timers_test.go+89 −1internal/schedule/control.go+1 −1internal/schedule/converge.go+25 −3internal/schedule/converge_test.go+32 −3internal/schedule/declaration.go+25 −6internal/schedule/declaration_test.go+29 −0internal/schedule/runner.go+121 −22internal/schedule/runner_test.go+186 −11internal/schedule/status.go+1 −0schedule.go+19 −5schedule_test.go+13 −2
diff --git a/internal/registry/registry.go b/internal/registry/registry.go| index 1796467ae..0b79bd5b5 100644 |
| --- a/internal/registry/registry.go |
| +++ b/internal/registry/registry.go |
| @@ -435,6 +435,13 @@ CREATE TABLE IF NOT EXISTS wildcard_domains ( |
| CREATE INDEX IF NOT EXISTS timers_due ON timers (next_at) WHERE next_at > 0;`); err != nil { |
| return err |
| } |
| + // Additive, same rule as the routes list above: a registry created by a |
| + // build before due_at existed gains the column on open, and re-opening |
| + // swallows "duplicate column". Rows carry 0 until the runner next |
| + // advances them; readers fall back to next_at (registry.Timer.DueAt). |
| + if _, err := r.sql.Exec(`ALTER TABLE timers ADD COLUMN due_at INTEGER NOT NULL DEFAULT 0`); err != nil && !strings.Contains(err.Error(), "duplicate column") { |
| + return err |
| + } |
| if err := r.migrateAccounts(); err != nil { |
| return err |
| } |
diff --git a/internal/registry/timers.go b/internal/registry/timers.go| index d1641457d..f8293daa7 100644 |
| --- a/internal/registry/timers.go |
| +++ b/internal/registry/timers.go |
| @@ -21,27 +21,35 @@ type Timer struct { |
| LastMs int64 |
| Attempts int |
| CreatedAt int64 |
| + // DueAt is the natural occurrence the row is (or was last) delivering — |
| + // the value X-Carlos-Schedule-At carries, and the key an app dedupes |
| + // on. It is NOT next_at: after a retry next_at is the retry instant and |
| + // after `run` it is "now", while the occurrence stays put. Moves only |
| + // when the row advances to a new occurrence (MarkTimerFired with a new |
| + // due, FireTimerNow, or an upsert that moves next_at). Zero on rows |
| + // written before the column existed; readers fall back to NextAt. |
| + DueAt int64 |
| } |
| var ErrTimerKind = errors.New("registry: timer exists with a different kind") |
| -const timerCols = `host, name, kind, spec, path, next_at, last_at, last_status, last_ms, attempts, created_at` |
| +const timerCols = `host, name, kind, spec, path, next_at, last_at, last_status, last_ms, attempts, created_at, due_at` |
| func scanTimer(row interface{ Scan(...any) error }) (Timer, error) { |
| var t Timer |
| - err := row.Scan(&t.Host, &t.Name, &t.Kind, &t.Spec, &t.Path, &t.NextAt, &t.LastAt, &t.LastStatus, &t.LastMs, &t.Attempts, &t.CreatedAt) |
| + err := row.Scan(&t.Host, &t.Name, &t.Kind, &t.Spec, &t.Path, &t.NextAt, &t.LastAt, &t.LastStatus, &t.LastMs, &t.Attempts, &t.CreatedAt, &t.DueAt) |
| return t, err |
| } |
| // UpsertTimer inserts t, or updates an existing row's spec/path — moving |
| -// next_at only when spec or path changed or t.NextAt is non-zero — and |
| -// never touches the runner's columns on update. |
| +// next_at (and due_at with it) only when spec or path changed or t.NextAt |
| +// is non-zero — and never touches the runner's columns on update. |
| func (r *Registry) UpsertTimer(t Timer) error { |
| existing, err := r.GetTimer(t.Host, t.Name) |
| switch { |
| case errors.Is(err, os.ErrNotExist): |
| - _, err := r.sql.Exec(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?)`, |
| - t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt) |
| + _, err := r.sql.Exec(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)`, |
| + t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt) |
| return err |
| case err != nil: |
| return err |
| @@ -49,11 +57,11 @@ func (r *Registry) UpsertTimer(t Timer) error { |
| if existing.Kind != t.Kind { |
| return ErrTimerKind |
| } |
| - next := existing.NextAt |
| + next, due := existing.NextAt, existing.DueAt |
| if existing.Spec != t.Spec || existing.Path != t.Path || t.NextAt != 0 { |
| - next = t.NextAt |
| + next, due = t.NextAt, t.DueAt |
| } |
| - _, err = r.sql.Exec(`UPDATE timers SET spec=?, path=?, next_at=? WHERE host=? AND name=?`, t.Spec, t.Path, next, t.Host, t.Name) |
| + _, err = r.sql.Exec(`UPDATE timers SET spec=?, path=?, next_at=?, due_at=? WHERE host=? AND name=?`, t.Spec, t.Path, next, due, t.Host, t.Name) |
| return err |
| } |
| @@ -98,14 +106,29 @@ 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) |
| } |
| +// FireTimerNow makes the row due at now, and now IS the occurrence being |
| +// delivered (a `carlos schedule run` is its own occurrence, not an early |
| +// delivery of the next natural one). |
| func (r *Registry) FireTimerNow(host, name string, now int64) error { |
| - _, err := r.sql.Exec(`UPDATE timers SET next_at=? WHERE host=? AND name=?`, now, host, name) |
| + _, err := r.sql.Exec(`UPDATE timers SET next_at=?, due_at=? WHERE host=? AND name=?`, now, now, host, name) |
| return err |
| } |
| -func (r *Registry) MarkTimerFired(host, name string, at int64, status string, ms int64, attempts int, next int64) error { |
| - _, err := r.sql.Exec(`UPDATE timers SET last_at=?, last_status=?, last_ms=?, attempts=?, next_at=? WHERE host=? AND name=?`, |
| - at, status, ms, attempts, next, host, name) |
| +// MarkTimerFired records one delivery. next is when the row is due again; |
| +// due is the occurrence THAT next delivery is for: the same value as next |
| +// when the row advances, and the row's current DueAt when next is a retry |
| +// of the occurrence just attempted — the runner decides which. |
| +func (r *Registry) MarkTimerFired(host, name string, at int64, status string, ms int64, attempts int, next, due int64) error { |
| + _, err := r.sql.Exec(`UPDATE timers SET last_at=?, last_status=?, last_ms=?, attempts=?, next_at=?, due_at=? WHERE host=? AND name=?`, |
| + at, status, ms, attempts, next, due, host, name) |
| + return err |
| +} |
| + |
| +// SetTimerStatus writes last_status alone — converge's one word on a row |
| +// it parked (schedule.StatusNeverFires); last_at stays as it was, so the |
| +// status reads as a condition, not a fire. |
| +func (r *Registry) SetTimerStatus(host, name, status string) error { |
| + _, err := r.sql.Exec(`UPDATE timers SET last_status=? WHERE host=? AND name=?`, status, host, name) |
| return err |
| } |
diff --git a/internal/registry/timers_test.go b/internal/registry/timers_test.go| index 15c098540..3d200c72e 100644 |
| --- a/internal/registry/timers_test.go |
| +++ b/internal/registry/timers_test.go |
| @@ -1,6 +1,7 @@ |
| package registry |
| import ( |
| + "database/sql" |
| "errors" |
| "os" |
| "path/filepath" |
| @@ -22,7 +23,7 @@ func TestTimersUpsertPreservesRunnerFields(t *testing.T) { |
| if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "schedule", Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: 100, CreatedAt: 1}); err != nil { |
| t.Fatal(err) |
| } |
| - if err := r.MarkTimerFired("h", "sync", 100, "204", 300, 0, 21700); err != nil { |
| + if err := r.MarkTimerFired("h", "sync", 100, "204", 300, 0, 21700, 21700); err != nil { |
| t.Fatal(err) |
| } |
| // Converge sees the same declaration again: spec/path unchanged, next_at |
| @@ -122,3 +123,90 @@ func TestTimersPruneAndExcept(t *testing.T) { |
| t.Fatal("delete not idempotent") |
| } |
| } |
| + |
| +// due_at is the occurrence a row is delivering (X-Carlos-Schedule-At). It |
| +// rides next_at on an upsert that moves it, follows the runner's word on |
| +// MarkTimerFired, and IS "now" after FireTimerNow — and a registry written |
| +// before the column existed gains it on open, with 0 on the old rows. |
| +func TestTimersDueAtFollowsTheOccurrence(t *testing.T) { |
| + r := openTimersReg(t) |
| + if err := r.UpsertTimer(Timer{Host: "h", Name: "s", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 3600, DueAt: 3600}); err != nil { |
| + t.Fatal(err) |
| + } |
| + // A retry: next moves to the retry instant, due stays the occurrence. |
| + if err := r.MarkTimerFired("h", "s", 3600, "503", 1, 1, 3660, 3600); err != nil { |
| + t.Fatal(err) |
| + } |
| + got, _ := r.GetTimer("h", "s") |
| + if got.NextAt != 3660 || got.DueAt != 3600 { |
| + t.Fatalf("after retry: %+v", got) |
| + } |
| + // An advance: both move together. |
| + if err := r.MarkTimerFired("h", "s", 3660, "204", 1, 0, 7200, 7200); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got, _ = r.GetTimer("h", "s"); got.NextAt != 7200 || got.DueAt != 7200 { |
| + t.Fatalf("after advance: %+v", got) |
| + } |
| + // An unchanged upsert (NextAt 0) leaves both alone; a spec change moves both. |
| + if err := r.UpsertTimer(Timer{Host: "h", Name: "s", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got, _ = r.GetTimer("h", "s"); got.NextAt != 7200 || got.DueAt != 7200 { |
| + t.Fatalf("after unchanged upsert: %+v", got) |
| + } |
| + if err := r.UpsertTimer(Timer{Host: "h", Name: "s", Kind: "schedule", Spec: "every 2h0m0s", Path: "/p", NextAt: 9000, DueAt: 9000}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got, _ = r.GetTimer("h", "s"); got.NextAt != 9000 || got.DueAt != 9000 { |
| + t.Fatalf("after spec change: %+v", got) |
| + } |
| + // `run`: now is its own occurrence. |
| + if err := r.FireTimerNow("h", "s", 8000); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got, _ = r.GetTimer("h", "s"); got.NextAt != 8000 || got.DueAt != 8000 { |
| + t.Fatalf("after FireTimerNow: %+v", got) |
| + } |
| +} |
| + |
| +func TestTimersPreDueAtRegistryGainsTheColumn(t *testing.T) { |
| + path := filepath.Join(t.TempDir(), "registry.db") |
| + reg, err := Open(path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + old, err := sql.Open("sqlite", path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := old.Exec(`ALTER TABLE timers DROP COLUMN due_at`); err != nil { |
| + t.Fatalf("simulate a pre-due_at registry: %v", err) |
| + } |
| + if _, err := old.Exec(`INSERT INTO timers (host, name, kind, spec, path, next_at, created_at) VALUES ('h', 'legacy', 'schedule', 'every 1h0m0s', '/p', 3600, 1)`); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := old.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + // Open twice: the ALTER must be idempotent, not a one-shot. |
| + for i := 0; i < 2; i++ { |
| + reg, err = Open(path) |
| + if err != nil { |
| + t.Fatalf("open %d of a pre-due_at registry: %v", i+1, err) |
| + } |
| + got, err := reg.GetTimer("h", "legacy") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if got.NextAt != 3600 || got.DueAt != 0 { |
| + t.Fatalf("migrated row = %+v, want next_at kept and due_at 0", got) |
| + } |
| + if err := reg.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| +} |
diff --git a/internal/schedule/control.go b/internal/schedule/control.go| index 5a5e249dc..7f37a1c2a 100644 |
| --- a/internal/schedule/control.go |
| +++ b/internal/schedule/control.go |
| @@ -92,7 +92,7 @@ func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (stri |
| } |
| } |
| sp := Spec{Kind: KindOnce, At: atT.UTC()} |
| - t := registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: sp.String(), Path: body.Path, NextAt: atT.Unix(), CreatedAt: now().Unix()} |
| + t := registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: sp.String(), Path: body.Path, NextAt: atT.Unix(), DueAt: atT.Unix(), CreatedAt: now().Unix()} |
| if err := store.UpsertTimer(t); err != nil { |
| http.Error(w, "registry error", http.StatusInternalServerError) |
| return |
diff --git a/internal/schedule/converge.go b/internal/schedule/converge.go| index 8dd839dcd..5fb9d0051 100644 |
| --- a/internal/schedule/converge.go |
| +++ b/internal/schedule/converge.go |
| @@ -216,10 +216,21 @@ func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName s |
| prev := findTimer(before, sc.Name, KindSchedule) |
| wouldChange := prev == nil || prev.Spec != sp.String() || prev.Path != sc.Path |
| var next int64 |
| + neverFires := false |
| if wouldChange { |
| - next = sp.Next(now).Unix() |
| + // Schedule.Validate refuses a spec whose first Next is zero at |
| + // `set` time; one written some other way must not become a row |
| + // due at the zero time's unix value (-62135596800, i.e. now). |
| + // It is parked at next_at=0 and told why, below. |
| + if n := sp.Next(now); !n.IsZero() { |
| + next = n.Unix() |
| + } else { |
| + neverFires = true |
| + } |
| } |
| - err = c.Reg.UpsertTimer(registry.Timer{Host: rt.Host, Name: sc.Name, Kind: KindSchedule, Spec: sp.String(), Path: sc.Path, NextAt: next, CreatedAt: now.Unix()}) |
| + // A new occurrence is due whenever next moves (DueAt rides NextAt — |
| + // UpsertTimer leaves both alone when NextAt is 0 and nothing changed). |
| + err = c.Reg.UpsertTimer(registry.Timer{Host: rt.Host, Name: sc.Name, Kind: KindSchedule, Spec: sp.String(), Path: sc.Path, NextAt: next, DueAt: next, CreatedAt: now.Unix()}) |
| if errors.Is(err, registry.ErrTimerKind) { |
| // Nothing was written — a collision must not pin `changed`, |
| // or a permanently-colliding name reports every single pass. |
| @@ -229,6 +240,12 @@ func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName s |
| if err != nil { |
| return changed, err |
| } |
| + if neverFires { |
| + log.Printf("schedule: %s: schedule %q never fires (no occurrence within the search window) — parked", rt.Host, sc.Name) |
| + if err := c.Reg.SetTimerStatus(rt.Host, sc.Name, StatusNeverFires); err != nil { |
| + return changed, err |
| + } |
| + } |
| if wouldChange { |
| changed = true |
| } |
| @@ -280,6 +297,11 @@ func findTimer(rows []registry.Timer, name, kind string) *registry.Timer { |
| const UnsupportedUnit = "unit-backed instance — scheduled work delivers to exec-backed instances in this release" |
| +// StatusNeverFires is the last_status of a parked row: a declared cron |
| +// expression with no occurrence inside CronExpr.Next's window (Validate |
| +// refuses these at `set`; this is the defence for one that slipped past it). |
| +const StatusNeverFires = "never fires" |
| + |
| // StatusFor builds the host's report from its rows. |
| func StatusFor(reg *registry.Registry, rt registry.Route, box string, now time.Time) (Status, error) { |
| st := Status{Schema: 1, Host: rt.Host, Box: box, ReportedAt: now.UTC().Format(time.RFC3339), Timers: []TimerStatus{}} |
| @@ -301,5 +323,5 @@ func StatusFor(reg *registry.Registry, rt registry.Route, box string, now time.T |
| // schedule ls` reader / the runner's own report path) reuses this so the |
| // row-to-status mapping is written once. |
| func toStatus(t registry.Timer) TimerStatus { |
| - return TimerStatus{Name: t.Name, Kind: t.Kind, Spec: t.Spec, Path: t.Path, NextAt: t.NextAt, LastAt: t.LastAt, LastStatus: t.LastStatus, LastMs: t.LastMs, Attempts: t.Attempts} |
| + return TimerStatus{Name: t.Name, Kind: t.Kind, Spec: t.Spec, Path: t.Path, NextAt: t.NextAt, DueAt: t.DueAt, LastAt: t.LastAt, LastStatus: t.LastStatus, LastMs: t.LastMs, Attempts: t.Attempts} |
| } |
diff --git a/internal/schedule/converge_test.go b/internal/schedule/converge_test.go| index 177c24b3d..2b61969d5 100644 |
| --- a/internal/schedule/converge_test.go |
| +++ b/internal/schedule/converge_test.go |
| @@ -75,7 +75,7 @@ func TestConvergeMaterialisesDeclarationPerExecInstance(t *testing.T) { |
| // A second pass with nothing changed: no new reports, next_at untouched. |
| reports = nil |
| - reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 10, 0, 999) |
| + reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 10, 0, 999, 999) |
| if err := c.Pass(ctx); err != nil { |
| t.Fatal(err) |
| } |
| @@ -134,7 +134,7 @@ func TestConvergeRunSignalFiresOnceAcrossInstances(t *testing.T) { |
| t.Fatalf("%s not fired: %+v", h, tm) |
| } |
| } |
| - reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 1, 0, 777) |
| + reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 1, 0, 777, 777) |
| c2.Pass(ctx) // same etag: must not re-fire |
| if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 { |
| t.Fatal("run signal replayed on an unchanged etag") |
| @@ -337,7 +337,7 @@ func TestConvergePassIsSafeUnderConcurrentCallers(t *testing.T) { |
| } |
| } |
| // 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) |
| + reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 1, 0, 777, 777) |
| if err := c.Pass(ctx); err != nil { |
| t.Fatal(err) |
| } |
| @@ -348,3 +348,32 @@ func TestConvergePassIsSafeUnderConcurrentCallers(t *testing.T) { |
| t.Fatal("no reports at all — the fixture converged nothing") |
| } |
| } |
| + |
| +// A declaration whose cron never matches (written past Validate — an older |
| +// console, or by hand) must not become a row due at the zero time's unix |
| +// value: it is parked at next_at=0 with last_status "never fires", and the |
| +// runner therefore never sees it. |
| +func TestConvergeParksAScheduleThatNeverFires(t *testing.T) { |
| + reg, st := convergeFixture(t) |
| + ctx := context.Background() |
| + now := at("2026-08-23T13:07:00Z") |
| + raw := []byte(`{"schema":1,"schedules":[{"name":"never","cron":"0 0 30 2 *","path":"/x"}]}`) |
| + if _, err := st.Put(ctx, DeclarationKey("bab", "app"), raw, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + c := NewConverger(reg, st, "box1", nil) |
| + c.Now = func() time.Time { return now } |
| + if err := c.Pass(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + tm, err := reg.GetTimer("a.bab.oncarlos.com", "never") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + 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 { |
| + t.Fatalf("a parked row is due: %+v", due) |
| + } |
| +} |
diff --git a/internal/schedule/declaration.go b/internal/schedule/declaration.go| index 4477a6af3..74355ff4d 100644 |
| --- a/internal/schedule/declaration.go |
| +++ b/internal/schedule/declaration.go |
| @@ -8,6 +8,7 @@ import ( |
| "errors" |
| "fmt" |
| "io" |
| + "net/url" |
| "os" |
| "regexp" |
| "sort" |
| @@ -23,9 +24,11 @@ var nameRE = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,31}$`) |
| func ValidName(name string) bool { return nameRE.MatchString(name) } |
| -// ValidPath: absolute, no scheme/host, no "..", short. A path is what the |
| -// tick POSTs to on the instance's own socket — nothing here can reach |
| -// another host, but a bad one should fail at the terminal, not on the box. |
| +// ValidPath: absolute, no scheme/host, no "..", short, and parseable by |
| +// the same parser the runner builds the tick with (url.ParseRequestURI) — |
| +// so a path accepted here is a path BuildRequest can send, and a bad one |
| +// fails at the terminal, not on the box. A path is what the tick POSTs to |
| +// on the instance's own socket; nothing here can reach another host. |
| func ValidPath(p string) error { |
| switch { |
| case p == "" || p[0] != '/': |
| @@ -37,7 +40,14 @@ func ValidPath(p string) error { |
| case strings.ContainsAny(p, " \t\r\n"): |
| return errors.New("path must not contain whitespace") |
| } |
| - for _, seg := range strings.Split(strings.SplitN(p, "?", 2)[0], "/") { |
| + u, err := url.ParseRequestURI(p) |
| + if err != nil { |
| + return fmt.Errorf("path: %w", err) |
| + } |
| + if u.Scheme != "" || u.Host != "" || u.User != nil { |
| + return errors.New("path must not carry a scheme or host") |
| + } |
| + for _, seg := range strings.Split(u.Path, "/") { |
| if seg == ".." { |
| return errors.New("path must not contain ..") |
| } |
| @@ -89,8 +99,17 @@ func (s Schedule) Validate() error { |
| if err := ValidPath(s.Path); err != nil { |
| return err |
| } |
| - _, err := s.Spec() |
| - return err |
| + sp, err := s.Spec() |
| + if err != nil { |
| + return err |
| + } |
| + // A cron expression can parse and still match nothing ("0 0 30 2 *"): |
| + // its Next is the zero time, which a box must never store as a due |
| + // instant. Refuse it here, where the member can read the answer. |
| + if sp.Next(time.Now()).IsZero() { |
| + return errors.New("this expression never fires (no matching minute within the next 5 years)") |
| + } |
| + return nil |
| } |
| // Declaration is apps/<acct>/<app>/schedules.json (spec §4.1). |
diff --git a/internal/schedule/declaration_test.go b/internal/schedule/declaration_test.go| index 0bcee8212..5734179d2 100644 |
| --- a/internal/schedule/declaration_test.go |
| +++ b/internal/schedule/declaration_test.go |
| @@ -112,3 +112,32 @@ func TestTouchRunMovesTheEtag(t *testing.T) { |
| t.Fatalf("etag did not move: %q → %q", e1, e2) |
| } |
| } |
| + |
| +// A cron expression can parse and still never match ("0 0 30 2 *"): its |
| +// Next is the zero time, and a box that stored that would have a row due |
| +// at -62135596800. Validate refuses it where the member can read why. |
| +func TestScheduleValidateRefusesAnExpressionThatNeverFires(t *testing.T) { |
| + err := Schedule{Name: "never", Cron: "0 0 30 2 *", Path: "/x"}.Validate() |
| + if err == nil || !strings.Contains(err.Error(), "never fires") { |
| + t.Fatalf("Validate(never-matching cron) = %v, want a 'never fires' refusal", err) |
| + } |
| + if err := (Schedule{Name: "leap", Cron: "0 0 29 2 *", Path: "/x"}).Validate(); err != nil { |
| + t.Fatalf("Feb 29 is a real (if rare) minute: %v", err) |
| + } |
| +} |
| + |
| +// ValidPath runs the parser BuildRequest builds the tick with, so a path it |
| +// accepts is a path the runner can send: a bad escape, a scheme or a |
| +// host smuggled in front of the path all fail at the terminal. |
| +func TestValidPathAgreesWithTheRequestParser(t *testing.T) { |
| + for _, bad := range []string{"/jobs/%zz", "/jobs/%", "http://x/y", "//x/y"} { |
| + if err := ValidPath(bad); err == nil { |
| + t.Errorf("ValidPath(%q) = ok, want a refusal", bad) |
| + } |
| + } |
| + for _, ok := range []string{"/jobs/sync", "/a%20b", "/x?y=1&z=2", "/x#frag"} { |
| + if err := ValidPath(ok); err != nil { |
| + t.Errorf("ValidPath(%q) = %v", ok, err) |
| + } |
| + } |
| +} |
diff --git a/internal/schedule/runner.go b/internal/schedule/runner.go| index 3c53f7d76..c366ba1ac 100644 |
| --- a/internal/schedule/runner.go |
| +++ b/internal/schedule/runner.go |
| @@ -37,7 +37,7 @@ type Deliverer interface { |
| // Runner fires due timers: at-least-once, serialized per host, bounded |
| // parallelism across hosts (spec §6.3). Every seam is a field so a tick is |
| -// testable with no goroutines, sockets or clocks. |
| +// testable with no sockets or clocks. |
| type Runner struct { |
| Reg *registry.Registry |
| Deliver Deliverer |
| @@ -48,6 +48,21 @@ type Runner struct { |
| Report func(rt registry.Route, st Status) |
| Box string |
| Build string |
| + |
| + // Per-host work is DETACHED from the tick: Tick launches one goroutine |
| + // per host with due rows and returns without waiting for any of them. |
| + // A handler is allowed to run for up to DeliveryTimeout (30m), and a |
| + // tick that waited on it would hold every other host's due timers |
| + // hostage for that long. busy is what keeps "one delivery at a time per |
| + // host" true across ticks: a host whose goroutine is still running is |
| + // skipped, and its rows (still due — nothing has been recorded yet) are |
| + // picked up by the first tick after it finishes. sem bounds hosts in |
| + // flight across ALL ticks, not per tick. inflight is for tests, which |
| + // need "everything this tick launched has finished" — see wait. |
| + mu sync.Mutex |
| + busy map[string]bool |
| + sem chan struct{} |
| + inflight sync.WaitGroup |
| } |
| // Start runs Tick every TickEvery until ctx ends. |
| @@ -80,8 +95,10 @@ func (r *Runner) timeout() time.Duration { |
| return DeliveryTimeout |
| } |
| -// Tick is one pass: everything due, grouped by host, hosts in parallel up to |
| -// MaxParallelHosts, timers within a host one after another. |
| +// Tick is one pass: everything due, grouped by host; each host not already |
| +// in flight gets a goroutine (at most MaxParallelHosts running across all |
| +// 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) |
| if err != nil { |
| @@ -99,19 +116,36 @@ func (r *Runner) Tick(ctx context.Context) { |
| } |
| byHost[t.Host] = append(byHost[t.Host], t) |
| } |
| - sem := make(chan struct{}, MaxParallelHosts) |
| - var wg sync.WaitGroup |
| for _, host := range order { |
| rt, err := r.Reg.Get(host) |
| - if err != nil || rt == nil { |
| + if err != nil { |
| + // A registry error is not "route gone": say so, or a broken |
| + // registry reads as a fleet of retired routes. |
| + log.Printf("schedule: %s: registry read: %v — skipping this tick", host, err) |
| + continue |
| + } |
| + if rt == nil { |
| // Converge prunes these; until it does, they are not ours to fire. |
| continue |
| } |
| - wg.Add(1) |
| - sem <- struct{}{} |
| - go func(rt registry.Route, timers []registry.Timer) { |
| - defer wg.Done() |
| - defer func() { <-sem }() |
| + if !r.claim(host) { |
| + continue // a previous tick's delivery to this host is still running |
| + } |
| + // The slot is taken HERE, before the goroutine exists, so the bound |
| + // is on hosts in flight and never on goroutines waiting to be. Hosts |
| + // are in due order, so when every slot is busy the earliest-due host |
| + // is the one waiting for the next free one. |
| + select { |
| + case r.slots() <- struct{}{}: |
| + case <-ctx.Done(): |
| + r.release(host) |
| + return |
| + } |
| + r.inflight.Add(1) |
| + go func(host string, rt registry.Route, timers []registry.Timer) { |
| + defer r.inflight.Done() |
| + defer func() { <-r.slots() }() |
| + defer r.release(host) |
| for _, t := range timers { |
| r.fire(ctx, rt, t) |
| } |
| @@ -120,11 +154,46 @@ func (r *Runner) Tick(ctx context.Context) { |
| r.Report(rt, st) |
| } |
| } |
| - }(*rt, byHost[host]) |
| + }(host, *rt, byHost[host]) |
| } |
| - wg.Wait() |
| } |
| +// claim marks host in flight; false if it already was. |
| +func (r *Runner) claim(host string) bool { |
| + r.mu.Lock() |
| + defer r.mu.Unlock() |
| + if r.busy[host] { |
| + return false |
| + } |
| + if r.busy == nil { |
| + r.busy = map[string]bool{} |
| + } |
| + r.busy[host] = true |
| + return true |
| +} |
| + |
| +func (r *Runner) release(host string) { |
| + r.mu.Lock() |
| + delete(r.busy, host) |
| + r.mu.Unlock() |
| +} |
| + |
| +// slots is the cross-tick semaphore, made on first use so a zero Runner |
| +// literal (every test, and the agent's) needs no constructor. |
| +func (r *Runner) slots() chan struct{} { |
| + r.mu.Lock() |
| + defer r.mu.Unlock() |
| + if r.sem == nil { |
| + r.sem = make(chan struct{}, MaxParallelHosts) |
| + } |
| + return r.sem |
| +} |
| + |
| +// wait blocks until every host goroutine launched so far has finished — |
| +// the tests' seam, so a Tick's post-conditions can be asserted without |
| +// sleeping. Nothing in the agent calls it. |
| +func (r *Runner) wait() { r.inflight.Wait() } |
| + |
| func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer) { |
| sp, err := ParseSpec(t.Spec) |
| if err != nil { |
| @@ -138,8 +207,15 @@ func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer) |
| r.record(rt, t, sp, start, Outcome{Status: "no instance token: " + err.Error(), Retry: true}, 0) |
| return |
| } |
| + req, err := BuildRequest(rt, t, token, r.Build) |
| + if err != nil { |
| + // Not retried: the path is the row's, and it will be exactly as bad |
| + // next time. The row advances (or finishes) with the reason on it. |
| + r.record(rt, t, sp, start, Outcome{Status: "bad path: " + err.Error()}, 0) |
| + return |
| + } |
| dctx, cancel := context.WithTimeout(ctx, r.timeout()) |
| - req := BuildRequest(rt, t, token, r.Build).WithContext(dctx) |
| + req = req.WithContext(dctx) |
| status, derr := r.Deliver.Deliver(dctx, rt, req) |
| cancel() |
| if errors.Is(derr, context.DeadlineExceeded) { |
| @@ -158,6 +234,10 @@ func (r *Runner) record(rt registry.Route, t registry.Timer, sp Spec, at time.Ti |
| if natural.IsZero() { |
| next = 0 |
| } |
| + // due is the occurrence the NEXT delivery is for (X-Carlos-Schedule-At, |
| + // the app's dedupe key): the new natural occurrence when the row |
| + // advances, and the one just attempted when it is retried. |
| + due := next |
| level := "info" |
| if o.Retry { |
| attempts = t.Attempts + 1 |
| @@ -171,16 +251,21 @@ func (r *Runner) record(rt registry.Route, t registry.Timer, sp Spec, at time.Ti |
| backoff = backoffCap |
| } |
| retryAt := at.Add(backoff) |
| + // Never let a retry outlive the next natural occurrence: past it |
| + // the retry and the occurrence would be the same delivery, and a |
| + // 15m backoff on a 1m schedule would skip 14 fires. A once row |
| + // has no natural next (zero), so its retries are never clamped. |
| if !natural.IsZero() && retryAt.After(natural) { |
| retryAt = natural |
| } |
| next = retryAt.Unix() |
| + due = dueOf(t) |
| level = "warn" |
| } |
| } else if !isSuccess(o.Status) { |
| level = "warn" |
| } |
| - if err := r.Reg.MarkTimerFired(rt.Host, t.Name, at.Unix(), status, took.Milliseconds(), attempts, next); err != nil { |
| + if err := r.Reg.MarkTimerFired(rt.Host, t.Name, at.Unix(), status, took.Milliseconds(), attempts, next, due); err != nil { |
| log.Printf("schedule: %s/%s: record fire: %v", rt.Host, t.Name, err) |
| } |
| if r.Event != nil { |
| @@ -218,23 +303,37 @@ func classify(status int, err error, timeout time.Duration) Outcome { |
| // Classify is the exported policy for the host package's tests. |
| func Classify(status int, err error) Outcome { return classify(status, err, DeliveryTimeout) } |
| +// dueOf is the occurrence a row is delivering: DueAt, or NextAt for a row |
| +// written before due_at existed (whose NextAt, at attempt 0, IS the |
| +// occurrence — and once pinned by a retry it stays the occurrence). |
| +func dueOf(t registry.Timer) int64 { |
| + if t.DueAt != 0 { |
| + return t.DueAt |
| + } |
| + return t.NextAt |
| +} |
| + |
| // BuildRequest is the tick (spec §6.3's header table). Content-Length 0 and |
| -// no body: a tick carries no data, by design. |
| -func BuildRequest(rt registry.Route, t registry.Timer, token, build string) *http.Request { |
| +// no body: a tick carries no data, by design. A path the URL parser refuses |
| +// is an error, never a quiet POST somewhere else — ValidPath runs the same |
| +// parser at declaration, so this only fires for a row written some other |
| +// way, and the runner records it on the row rather than delivering. |
| +func BuildRequest(rt registry.Route, t registry.Timer, token, build string) (*http.Request, error) { |
| + if err := ValidPath(t.Path); err != nil { |
| + return nil, err |
| + } |
| req, err := http.NewRequest(http.MethodPost, "http://"+rt.Host+t.Path, http.NoBody) |
| if err != nil { |
| - // Path was validated at declaration; a failure here is a bug, not |
| - // a runtime condition. Fall back to the root so the app still hears. |
| - req, _ = http.NewRequest(http.MethodPost, "http://"+rt.Host+"/", http.NoBody) |
| + return nil, err |
| } |
| req.Host = rt.Host |
| req.ContentLength = 0 |
| req.Header.Set("Authorization", "Bearer "+token) |
| req.Header.Set("X-Carlos-Schedule", t.Name) |
| req.Header.Set("X-Carlos-Schedule-Kind", t.Kind) |
| - req.Header.Set("X-Carlos-Schedule-At", strconv.FormatInt(t.NextAt, 10)) |
| + req.Header.Set("X-Carlos-Schedule-At", strconv.FormatInt(dueOf(t), 10)) |
| req.Header.Set("User-Agent", "carlos-schedule/"+build) |
| - return req |
| + return req, nil |
| } |
| // TokenFromEnvFile is the production Token: CARLOS_ADMIN_TOKEN from |
diff --git a/internal/schedule/runner_test.go b/internal/schedule/runner_test.go| index 073e0e4b4..ce92cea36 100644 |
| --- a/internal/schedule/runner_test.go |
| +++ b/internal/schedule/runner_test.go |
| @@ -5,6 +5,7 @@ import ( |
| "errors" |
| "net/http" |
| "path/filepath" |
| + "strings" |
| "sync" |
| "testing" |
| "time" |
| @@ -28,7 +29,7 @@ func (f *fakeDeliverer) Deliver(_ context.Context, rt registry.Route, req *http. |
| return f.status, f.err |
| } |
| -func runnerFixture(t *testing.T, d *fakeDeliverer) (*registry.Registry, *Runner) { |
| +func runnerFixture(t *testing.T, d Deliverer) (*registry.Registry, *Runner) { |
| t.Helper() |
| reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db")) |
| if err != nil { |
| @@ -70,8 +71,11 @@ func TestClassify(t *testing.T) { |
| func TestBuildRequestCarriesTheContract(t *testing.T) { |
| rt := registry.Route{Host: "a.bab.oncarlos.com"} |
| - tm := registry.Timer{Name: "sync", Kind: KindSchedule, Path: "/jobs/sync?x=1", NextAt: 1787004000} |
| - req := BuildRequest(rt, tm, "tok", "abc") |
| + tm := registry.Timer{Name: "sync", Kind: KindSchedule, Path: "/jobs/sync?x=1", NextAt: 1787004060, DueAt: 1787004000} |
| + req, err := BuildRequest(rt, tm, "tok", "abc") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| if req.Method != http.MethodPost || req.Host != rt.Host || req.URL.Path != "/jobs/sync" || req.URL.RawQuery != "x=1" { |
| t.Fatalf("req = %s %s %s", req.Method, req.Host, req.URL) |
| } |
| @@ -80,6 +84,17 @@ func TestBuildRequestCarriesTheContract(t *testing.T) { |
| h.Get("X-Carlos-Schedule-At") != "1787004000" || h.Get("User-Agent") != "carlos-schedule/abc" || req.ContentLength != 0 { |
| t.Fatalf("headers = %v", h) |
| } |
| + // A row from before due_at existed carries 0 there: the header falls |
| + // back to next_at, which at attempt 0 is the occurrence. |
| + tm.DueAt = 0 |
| + if req, _ := BuildRequest(rt, tm, "tok", "abc"); req.Header.Get("X-Carlos-Schedule-At") != "1787004060" { |
| + t.Fatalf("legacy row At = %q", req.Header.Get("X-Carlos-Schedule-At")) |
| + } |
| + // A path the parser refuses is an error, never a POST to "/". |
| + tm.Path = "/jobs/%zz" |
| + if _, err := BuildRequest(rt, tm, "tok", "abc"); err == nil { |
| + t.Fatal("bad path built a request") |
| + } |
| } |
| func TestTickFiresDueAdvancesScheduleAndFinishesOnce(t *testing.T) { |
| @@ -94,7 +109,7 @@ func TestTickFiresDueAdvancesScheduleAndFinishesOnce(t *testing.T) { |
| r.Event = func(_ registry.Route, level, msg string) { events = append(events, level+": "+msg) } |
| var reports int |
| r.Report = func(registry.Route, Status) { reports++ } |
| - r.Tick(context.Background()) |
| + r.tick() |
| if len(d.calls) != 2 { |
| t.Fatalf("delivered %d, want 2 (not the future one)", len(d.calls)) |
| } |
| @@ -122,14 +137,14 @@ func TestTickRetriesWithBackoffThenGivesUp(t *testing.T) { |
| reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix()}) |
| wantBackoff := []time.Duration{time.Minute, 2 * time.Minute, 4 * time.Minute, 8 * time.Minute} |
| for i, b := range wantBackoff { |
| - r.Tick(context.Background()) |
| + r.tick() |
| tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync") |
| if tm.Attempts != i+1 || tm.NextAt != now.Add(b).Unix() || tm.LastStatus != "503" { |
| t.Fatalf("attempt %d: %+v (want next=+%s)", i+1, tm, b) |
| } |
| now = now.Add(b) |
| } |
| - r.Tick(context.Background()) // 5th failure: give up, advance to the natural next |
| + r.tick() // 5th failure: give up, advance to the natural next |
| tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync") |
| if tm.Attempts != 0 || tm.LastStatus != "failed after 5 attempts: 503" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() { |
| t.Fatalf("after giving up: %+v", tm) |
| @@ -142,7 +157,7 @@ func TestBackoffNeverPassesTheNaturalNext(t *testing.T) { |
| now := at("2026-08-23T13:59:30Z") // next natural 1m fire is 14:00:00, 30s away |
| r.Now = func() time.Time { return now } |
| reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "m", Kind: KindSchedule, Spec: "every 1m0s", Path: "/m", NextAt: now.Unix()}) |
| - r.Tick(context.Background()) |
| + r.tick() |
| tm, _ := reg.GetTimer("a.bab.oncarlos.com", "m") |
| if tm.NextAt != at("2026-08-23T14:00:00Z").Unix() { |
| t.Fatalf("next = %d, want the natural next, not now+1m", tm.NextAt) |
| @@ -157,7 +172,7 @@ func TestAppRefusalDoesNotRetry(t *testing.T) { |
| reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "o", Kind: KindOnce, Spec: "at 1", Path: "/o", NextAt: 1}) |
| var levels []string |
| r.Event = func(_ registry.Route, level, _ string) { levels = append(levels, level) } |
| - r.Tick(context.Background()) |
| + r.tick() |
| tm, _ := reg.GetTimer("a.bab.oncarlos.com", "o") |
| if tm.NextAt != 0 || tm.LastStatus != "404 (app refused)" || tm.Attempts != 0 { |
| t.Fatalf("%+v", tm) |
| @@ -173,7 +188,7 @@ func TestTimeoutCountsAsDelivered(t *testing.T) { |
| now := at("2026-08-23T13:07:00Z") |
| r.Now = func() time.Time { return now } |
| reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: now.Unix()}) |
| - r.Tick(context.Background()) |
| + r.tick() |
| tm, _ := reg.GetTimer("a.bab.oncarlos.com", "s") |
| if tm.Attempts != 0 || tm.LastStatus != "timeout after 1m0s" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() { |
| t.Fatalf("%+v", tm) |
| @@ -187,7 +202,7 @@ func TestNoTokenIsRecordedNotDelivered(t *testing.T) { |
| now := at("2026-08-23T13:07:00Z") |
| r.Now = func() time.Time { return now } |
| reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: now.Unix()}) |
| - r.Tick(context.Background()) |
| + r.tick() |
| if len(d.calls) != 0 { |
| t.Fatal("delivered without a token") |
| } |
| @@ -201,8 +216,168 @@ func TestTickSkipsRowsWhoseRouteIsGone(t *testing.T) { |
| d := &fakeDeliverer{status: 204} |
| reg, r := runnerFixture(t, d) |
| reg.UpsertTimer(registry.Timer{Host: "ghost.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: 1}) |
| - r.Tick(context.Background()) |
| + r.tick() |
| if len(d.calls) != 0 { |
| t.Fatal("delivered to a host with no route") |
| } |
| } |
| + |
| +// X-Carlos-Schedule-At is the OCCURRENCE, the key an app dedupes on (spec |
| +// §6.3, §7.1): a retry re-sends the same value, a fire that advances the row |
| +// moves it to the new natural occurrence, and a `run` makes now the |
| +// occurrence. next_at is none of those things after the first retry. |
| +func TestScheduleAtHeaderIsTheOccurrenceNotNextAt(t *testing.T) { |
| + d := &fakeDeliverer{status: 503} |
| + reg, r := runnerFixture(t, d) |
| + now := at("2026-08-23T13:07:00Z") |
| + r.Now = func() time.Time { return now } |
| + occurrence := now.Unix() - 5 |
| + reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: occurrence, DueAt: occurrence}) |
| + atHeader := func(i int) string { return d.calls[i].Header.Get("X-Carlos-Schedule-At") } |
| + |
| + r.tick() // fails → retry in 1m |
| + tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync") |
| + if tm.NextAt != now.Add(time.Minute).Unix() || tm.DueAt != occurrence { |
| + t.Fatalf("after first failure: %+v", tm) |
| + } |
| + now = now.Add(time.Minute) |
| + r.tick() // the retry: same occurrence on the wire |
| + if len(d.calls) != 2 || atHeader(0) != itoa(occurrence) || atHeader(1) != itoa(occurrence) { |
| + t.Fatalf("retry header = %q, first = %q, want both %d", atHeader(1), atHeader(0), occurrence) |
| + } |
| + |
| + d.status = 204 |
| + now = now.Add(2 * time.Minute) |
| + r.tick() // succeeds → advances |
| + tm, _ = reg.GetTimer("a.bab.oncarlos.com", "sync") |
| + next := at("2026-08-23T18:00:00Z").Unix() |
| + if tm.NextAt != next || tm.DueAt != next || atHeader(2) != itoa(occurrence) { |
| + t.Fatalf("after advance: %+v (header %q)", tm, atHeader(2)) |
| + } |
| + |
| + // `run`: FireTimerNow makes now the occurrence, and the tick sends it. |
| + now = now.Add(time.Minute) |
| + reg.FireTimerNow("a.bab.oncarlos.com", "sync", now.Unix()) |
| + r.tick() |
| + if len(d.calls) != 4 || atHeader(3) != itoa(now.Unix()) { |
| + t.Fatalf("run header = %q, want %d", atHeader(3), now.Unix()) |
| + } |
| +} |
| + |
| +// A row whose path the URL parser refuses is recorded, not retried and not |
| +// delivered anywhere else: ValidPath runs the same parser at declaration, so |
| +// such a row was written some other way and is exactly as bad next time. |
| +func TestBadPathIsRecordedNotDelivered(t *testing.T) { |
| + d := &fakeDeliverer{status: 204} |
| + reg, r := runnerFixture(t, d) |
| + now := at("2026-08-23T13:07:00Z") |
| + r.Now = func() time.Time { return now } |
| + reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/%zz", NextAt: now.Unix()}) |
| + r.tick() |
| + if len(d.calls) != 0 { |
| + t.Fatal("delivered a request the parser refused") |
| + } |
| + tm, _ := reg.GetTimer("a.bab.oncarlos.com", "s") |
| + if tm.Attempts != 0 || !strings.HasPrefix(tm.LastStatus, "bad path: ") || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() { |
| + t.Fatalf("%+v", tm) |
| + } |
| +} |
| + |
| +// tick is Tick plus a wait for every host goroutine it launched — the shape |
| +// every test here asserts against, since Tick itself returns as soon as the |
| +// work is in flight. |
| +func (r *Runner) tick() { |
| + r.Tick(context.Background()) |
| + r.wait() |
| +} |
| + |
| +// blockingDeliverer answers 204 at once for every host except `slow`, whose |
| +// delivery parks on `gate` until the test closes it — a 20-minute handler, |
| +// in miniature. |
| +type blockingDeliverer struct { |
| + fakeDeliverer |
| + slow string |
| + gate chan struct{} |
| +} |
| + |
| +func (b *blockingDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) { |
| + if rt.Host == b.slow { |
| + <-b.gate |
| + } |
| + 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 |
| +// host (its rows are still due — per-host serialization spans ticks). |
| +func TestOneSlowHostDoesNotBlockTheOthersOrRefire(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 } |
| + hosts := []string{"slow.bab.oncarlos.com", "b.bab.oncarlos.com", "c.bab.oncarlos.com", "d.bab.oncarlos.com", "e.bab.oncarlos.com", "f.bab.oncarlos.com", "g.bab.oncarlos.com"} |
| + for i, h := range hosts { |
| + if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + // The slow host is due FIRST, so it takes a slot before anyone else. |
| + reg.UpsertTimer(registry.Timer{Host: h, Name: "j", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 100 + int64(i)}) |
| + } |
| + if len(hosts) <= MaxParallelHosts { |
| + t.Fatalf("fixture has %d hosts; the test needs more than MaxParallelHosts (%d) so slots recycle under the parked one", len(hosts), MaxParallelHosts) |
| + } |
| + |
| + done := make(chan struct{}) |
| + go func() { r.Tick(context.Background()); close(done) }() |
| + select { |
| + case <-done: |
| + case <-time.After(10 * time.Second): |
| + t.Fatal("Tick did not return while one host was parked") |
| + } |
| + // Every other host fires while slow is parked. |
| + deadline := time.Now().Add(10 * time.Second) |
| + for { |
| + fired := d.firedHosts() |
| + if len(fired) == len(hosts)-1 && fired[d.slow] == 0 { |
| + break |
| + } |
| + if time.Now().After(deadline) { |
| + t.Fatalf("fired = %v, want every host but %s", fired, d.slow) |
| + } |
| + time.Sleep(5 * time.Millisecond) |
| + } |
| + for _, h := range hosts[1:] { |
| + if tm, _ := reg.GetTimer(h, "j"); tm.LastStatus != "204" { |
| + t.Fatalf("%s not recorded while slow was parked: %+v", h, tm) |
| + } |
| + } |
| + |
| + // The parked host's row is still due; a second tick must skip it. |
| + now = now.Add(time.Minute) |
| + r.Tick(context.Background()) |
| + close(d.gate) |
| + r.wait() |
| + if n := d.firedHosts()[d.slow]; n != 1 { |
| + t.Fatalf("slow host delivered %d times, want exactly 1 — a busy host was re-fired", n) |
| + } |
| + if tm, _ := reg.GetTimer(d.slow, "j"); tm.LastStatus != "204" || tm.NextAt != at("2026-08-23T14:00:00Z").Unix() { |
| + t.Fatalf("slow row after release: %+v", tm) |
| + } |
| + // And once released, it is fair game again. |
| + now = at("2026-08-23T14:00:01Z") |
| + r.tick() |
| + if n := d.firedHosts()[d.slow]; n != 2 { |
| + t.Fatalf("released host delivered %d times, want 2", n) |
| + } |
| +} |
diff --git a/internal/schedule/status.go b/internal/schedule/status.go| index d691e4377..df933dc3c 100644 |
| --- a/internal/schedule/status.go |
| +++ b/internal/schedule/status.go |
| @@ -20,6 +20,7 @@ type TimerStatus struct { |
| Spec string `json:"spec"` |
| Path string `json:"path"` |
| NextAt int64 `json:"next_at"` |
| + DueAt int64 `json:"due_at"` // the occurrence next_at delivers (X-Carlos-Schedule-At); 0 from a pre-due_at box |
| LastAt int64 `json:"last_at"` |
| LastStatus string `json:"last_status"` |
| LastMs int64 `json:"last_ms"` |
diff --git a/schedule.go b/schedule.go| index 251718d78..f92b5d0a7 100644 |
| --- a/schedule.go |
| +++ b/schedule.go |
| @@ -317,19 +317,33 @@ func renderScheduleInstance(w io.Writer, inst schedule.Status, now time.Time) { |
| return |
| } |
| fmt.Fprintf(w, " %s (box %s, reported %s ago)\n", inst.Host, inst.Box, reportedAgo(inst.ReportedAt, now)) |
| + // A once-timer is the instance's own (registered off its control |
| + // socket) and may share a name with a declared schedule, so it is |
| + // labelled — a member reading two "remind" lines must be able to tell |
| + // which is which. |
| + labels := make([]string, len(inst.Timers)) |
| nameW := 0 |
| - for _, t := range inst.Timers { |
| - if n := len(t.Name); n > nameW { |
| + for i, t := range inst.Timers { |
| + labels[i] = t.Name |
| + if t.Kind == schedule.KindOnce { |
| + labels[i] += " (once)" |
| + } |
| + if n := len(labels[i]); n > nameW { |
| nameW = n |
| } |
| } |
| - for _, t := range inst.Timers { |
| + for i, t := range inst.Timers { |
| next := relFire(t.NextAt, now) |
| last := relFire(t.LastAt, now) |
| - if t.LastAt != 0 && t.LastStatus != "" { |
| + switch { |
| + case t.LastAt != 0 && t.LastStatus != "": |
| last += fmt.Sprintf(" → %s in %.1fs", t.LastStatus, float64(t.LastMs)/1000) |
| + case t.LastStatus != "": |
| + // A status with no fire behind it is a condition the box put on |
| + // the row ("never fires" — a parked cron), not an outcome. |
| + last += " (" + t.LastStatus + ")" |
| } |
| - fmt.Fprintf(w, " %-*snext %s last %s\n", nameW+2, t.Name, next, last) |
| + fmt.Fprintf(w, " %-*snext %s last %s\n", nameW+2, labels[i], next, last) |
| } |
| } |
diff --git a/schedule_test.go b/schedule_test.go| index d7b3c2f7a..60b540e68 100644 |
| --- a/schedule_test.go |
| +++ b/schedule_test.go |
| @@ -243,6 +243,15 @@ func TestRenderScheduleList(t *testing.T) { |
| Name: "reminders", |
| NextAt: now.Add(9*time.Hour + 53*time.Minute).Unix(), |
| }, |
| + { |
| + Name: "reminders", |
| + Kind: schedule.KindOnce, |
| + NextAt: now.Add(30 * time.Second).Unix(), |
| + }, |
| + { |
| + Name: "leapless", |
| + LastStatus: schedule.StatusNeverFires, |
| + }, |
| }, |
| }, |
| { |
| @@ -262,8 +271,10 @@ func TestRenderScheduleList(t *testing.T) { |
| "\n" + |
| "instances\n" + |
| " a.bab.oncarlos.com (box carlos-ie-1, reported 12s ago)\n" + |
| - " sync next in 4h52m last 1h08m ago → 204 in 3.2s\n" + |
| - " reminders next in 9h53m last —\n" + |
| + " sync next in 4h52m last 1h08m ago → 204 in 3.2s\n" + |
| + " reminders next in 9h53m last —\n" + |
| + " reminders (once) next in 30s last —\n" + |
| + " leapless next — last — (never fires)\n" + |
| " u.bab.oncarlos.com unit-backed instance — scheduled work delivers to exec-backed instances in this release\n" |
| if got := buf.String(); got != want { |
| t.Fatalf("renderScheduleList =\n%s\nwant\n%s", got, want) |