amadan

carlos / platform Public

registry: timers table for scheduled work

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev bf081873891cf091d9e8ac84baf74ad65618a622 parent 852fa70
3 files changed, +300 −0
  • internal/registry/registry.go +17 −0
  • internal/registry/timers.go +159 −0
  • internal/registry/timers_test.go +124 −0
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index 08ac9554b..1796467ae 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -418,6 +418,23 @@ CREATE TABLE IF NOT EXISTS wildcard_domains (
);`); err != nil {
return err
}
+ if _, err := r.sql.Exec(`CREATE TABLE IF NOT EXISTS timers (
+ host TEXT NOT NULL,
+ name TEXT NOT NULL,
+ kind TEXT NOT NULL,
+ spec TEXT NOT NULL,
+ path TEXT NOT NULL,
+ next_at INTEGER NOT NULL,
+ last_at INTEGER NOT NULL DEFAULT 0,
+ last_status TEXT NOT NULL DEFAULT '',
+ last_ms INTEGER NOT NULL DEFAULT 0,
+ attempts INTEGER NOT NULL DEFAULT 0,
+ created_at INTEGER NOT NULL,
+ PRIMARY KEY (host, name)
+);
+CREATE INDEX IF NOT EXISTS timers_due ON timers (next_at) WHERE next_at > 0;`); err != nil {
+ return err
+ }
if err := r.migrateAccounts(); err != nil {
return err
}
diff --git a/internal/registry/timers.go b/internal/registry/timers.go
new file mode 100644
index 000000000..d1641457d
--- /dev/null
+++ b/internal/registry/timers.go
@@ -0,0 +1,159 @@
+// The timers table: scheduled work materialized per (host, name) — spec
+// docs/superpowers/specs/2026-08-23-scheduled-work-design.md §5. Rows come
+// from two writers with disjoint columns: converge (schedule.Converge) and
+// the control socket own spec/path/kind; the runner owns next_at after the
+// first fire, last_*, attempts. UpsertTimer encodes that split so neither
+// writer can clobber the other. Deliberately NOT in Fingerprint(): a timer
+// changes nothing about routing.
+package registry
+
+import (
+ "database/sql"
+ "errors"
+ "os"
+ "strings"
+)
+
+type Timer struct {
+ Host, Name, Kind, Spec, Path string
+ NextAt, LastAt int64
+ LastStatus string
+ LastMs int64
+ Attempts int
+ CreatedAt 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`
+
+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)
+ 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.
+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)
+ return err
+ case err != nil:
+ return err
+ }
+ if existing.Kind != t.Kind {
+ return ErrTimerKind
+ }
+ next := existing.NextAt
+ if existing.Spec != t.Spec || existing.Path != t.Path || t.NextAt != 0 {
+ next = t.NextAt
+ }
+ _, err = r.sql.Exec(`UPDATE timers SET spec=?, path=?, next_at=? WHERE host=? AND name=?`, t.Spec, t.Path, next, t.Host, t.Name)
+ return err
+}
+
+func (r *Registry) GetTimer(host, name string) (*Timer, error) {
+ t, err := scanTimer(r.sql.QueryRow(`SELECT `+timerCols+` FROM timers WHERE host=? AND name=?`, host, name))
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, os.ErrNotExist
+ }
+ if err != nil {
+ return nil, err
+ }
+ return &t, nil
+}
+
+func (r *Registry) DeleteTimer(host, name string) error {
+ _, err := r.sql.Exec(`DELETE FROM timers WHERE host=? AND name=?`, host, name)
+ return err
+}
+
+func (r *Registry) queryTimers(q string, args ...any) ([]Timer, error) {
+ rows, err := r.sql.Query(q, args...)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []Timer
+ for rows.Next() {
+ t, err := scanTimer(rows)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, t)
+ }
+ return out, rows.Err()
+}
+
+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)
+}
+
+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)
+ 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)
+ return err
+}
+
+func (r *Registry) CountTimers(host, kind string) (int, error) {
+ var n int
+ err := r.sql.QueryRow(`SELECT COUNT(*) FROM timers WHERE host=? AND kind=?`, host, kind).Scan(&n)
+ return n, err
+}
+
+// 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.
+func (r *Registry) PruneTimers(alive map[string]bool, doneBefore int64) (int, error) {
+ hosts := make([]any, 0, len(alive))
+ for h := range alive {
+ hosts = append(hosts, h)
+ }
+ q := `DELETE FROM timers WHERE (kind='once' AND next_at=0 AND last_at < ?)`
+ args := []any{doneBefore}
+ if len(hosts) == 0 {
+ q += ` OR 1=1`
+ } else {
+ q += ` OR host NOT IN (?` + strings.Repeat(",?", len(hosts)-1) + `)`
+ args = append(args, hosts...)
+ }
+ res, err := r.sql.Exec(q, args...)
+ if err != nil {
+ return 0, err
+ }
+ n, _ := res.RowsAffected()
+ return int(n), nil
+}
+
+func (r *Registry) DeleteTimersExcept(host, kind string, keep map[string]bool) (int, error) {
+ names := make([]any, 0, len(keep))
+ for n := range keep {
+ names = append(names, n)
+ }
+ q := `DELETE FROM timers WHERE host=? AND kind=?`
+ args := []any{host, kind}
+ if len(names) > 0 {
+ q += ` AND name NOT IN (?` + strings.Repeat(",?", len(names)-1) + `)`
+ args = append(args, names...)
+ }
+ res, err := r.sql.Exec(q, args...)
+ if err != nil {
+ return 0, err
+ }
+ n, _ := res.RowsAffected()
+ return int(n), nil
+}
diff --git a/internal/registry/timers_test.go b/internal/registry/timers_test.go
new file mode 100644
index 000000000..15c098540
--- /dev/null
+++ b/internal/registry/timers_test.go
@@ -0,0 +1,124 @@
+package registry
+
+import (
+ "errors"
+ "os"
+ "path/filepath"
+ "testing"
+)
+
+func openTimersReg(t *testing.T) *Registry {
+ t.Helper()
+ r, err := Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { r.Close() })
+ return r
+}
+
+func TestTimersUpsertPreservesRunnerFields(t *testing.T) {
+ r := openTimersReg(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 {
+ t.Fatal(err)
+ }
+ // Converge sees the same declaration again: spec/path unchanged, next_at
+ // passed along as "whatever it is now" (0 here means "don't touch").
+ if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "schedule", Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: 0, CreatedAt: 99}); err != nil {
+ t.Fatal(err)
+ }
+ got, err := r.GetTimer("h", "sync")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.LastAt != 100 || got.LastStatus != "204" || got.LastMs != 300 || got.NextAt != 21700 || got.CreatedAt != 1 {
+ t.Fatalf("runner fields clobbered: %+v", got)
+ }
+ // A changed spec DOES move next_at.
+ if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "schedule", Spec: "every 12h0m0s", Path: "/jobs/sync", NextAt: 43200}); err != nil {
+ t.Fatal(err)
+ }
+ got, _ = r.GetTimer("h", "sync")
+ if got.NextAt != 43200 || got.Spec != "every 12h0m0s" || got.LastStatus != "204" {
+ t.Fatalf("after spec change: %+v", got)
+ }
+ if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "once", Spec: "at 5", Path: "/p", NextAt: 5}); !errors.Is(err, ErrTimerKind) {
+ t.Fatalf("kind change = %v, want ErrTimerKind", err)
+ }
+}
+
+func TestTimersDueOrderingAndFireNow(t *testing.T) {
+ r := openTimersReg(t)
+ for _, tm := range []Timer{
+ {Host: "b", Name: "late", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 500},
+ {Host: "a", Name: "soon", Kind: "once", Spec: "at 10", Path: "/p", NextAt: 10},
+ {Host: "a", Name: "done", Kind: "once", Spec: "at 1", Path: "/p", NextAt: 0},
+ {Host: "c", Name: "future", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 9999},
+ } {
+ if err := r.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ }
+ due, err := r.DueTimers(600, 10)
+ 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 {
+ 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 {
+ t.Fatalf("after FireTimerNow: %d due", len(due))
+ }
+ if err := r.FireTimerNow("nope", "x", 600); err != nil {
+ t.Fatalf("FireTimerNow on absent = %v, want nil", err)
+ }
+}
+
+func TestTimersPruneAndExcept(t *testing.T) {
+ r := openTimersReg(t)
+ for _, tm := range []Timer{
+ {Host: "alive", Name: "s1", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 1},
+ {Host: "alive", Name: "s2", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 1},
+ {Host: "alive", Name: "o1", Kind: "once", Spec: "at 1", Path: "/p", NextAt: 0, LastAt: 10},
+ {Host: "alive", Name: "o2", Kind: "once", Spec: "at 1", Path: "/p", NextAt: 0, LastAt: 500},
+ {Host: "gone", Name: "s1", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 1},
+ } {
+ if err := r.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ }
+ // o1 has last_at=10 via insert — MarkTimerFired is the real path but insert suffices to seed.
+ n, err := r.PruneTimers(map[string]bool{"alive": true}, 100)
+ if err != nil || n != 2 { // gone/s1 and alive/o1
+ t.Fatalf("prune = %d %v", n, err)
+ }
+ n, err = r.DeleteTimersExcept("alive", "schedule", map[string]bool{"s2": true})
+ if err != nil || n != 1 {
+ t.Fatalf("except = %d %v", n, err)
+ }
+ left, _ := r.TimersFor("alive")
+ if len(left) != 2 || left[0].Name != "o2" || left[1].Name != "s2" {
+ t.Fatalf("left = %+v", left)
+ }
+ if c, _ := r.CountTimers("alive", "once"); c != 1 {
+ t.Fatalf("count = %d", c)
+ }
+ if _, err := r.GetTimer("gone", "s1"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("gone = %v", err)
+ }
+ if err := r.DeleteTimer("alive", "s2"); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.DeleteTimer("alive", "s2"); err != nil {
+ t.Fatal("delete not idempotent")
+ }
+}