amadan

carlos / platform Public

schedule: converge bucket declarations into timers rows

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 5628dfb482a6b392b9998cf5b18a198fc62f3735 parent bf08187
2 files changed, +398 −0
  • internal/schedule/converge.go +244 −0
  • internal/schedule/converge_test.go +154 −0
diff --git a/internal/schedule/converge.go b/internal/schedule/converge.go
new file mode 100644
index 000000000..dea12c171
--- /dev/null
+++ b/internal/schedule/converge.go
@@ -0,0 +1,244 @@
+package schedule
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// Converger turns apps/<acct>/<app>/schedules.json into timers rows for every
+// exec-backed instance this box serves, acts on schedule-run.json, and prunes
+// what no longer belongs (spec §6.2). One instance per agent; Pass is called
+// on every serial move, hourly, and once at start.
+//
+// Etag memory is what keeps an idle box cheap and a run-signal
+// exactly-once: an unchanged declaration is one GET and no writes; a run
+// signal fires when its etag MOVES, and the first sighting only seeds
+// (a signal older than this process is history — restart-gen's rule).
+type Converger struct {
+ Reg *registry.Registry
+ Store release.Store
+ Now func() time.Time
+ Box string // reported in Status.Box
+ Report func(rt registry.Route, st Status) // called for every host whose rows changed or that is unsupported; nil = no reports
+
+ declEtag map[[2]string]string // (acct, app) → last etag seen
+ declDoc map[[2]string]Declaration
+ runEtag map[[2]string]string
+
+ // reportedUnsupported tracks which unit-backed hosts have already had
+ // their one report — Report fires for them once, not every pass.
+ reportedUnsupported map[string]bool
+}
+
+const onceRetention = 7 * 24 * time.Hour
+
+func NewConverger(reg *registry.Registry, store release.Store, box string, report func(registry.Route, Status)) *Converger {
+ return &Converger{
+ Reg: reg, Store: store, Now: time.Now, Box: box, Report: report,
+ declEtag: map[[2]string]string{}, declDoc: map[[2]string]Declaration{},
+ runEtag: map[[2]string]string{}, reportedUnsupported: map[string]bool{},
+ }
+}
+
+// Pass is die-free per app: one app's unreadable declaration is logged and
+// its rows left exactly as they were (a baseline with holes manufactures
+// deletions), and every other app still converges.
+func (c *Converger) Pass(ctx context.Context) error {
+ routes, err := c.Reg.List()
+ if err != nil {
+ return err
+ }
+ now := c.Now()
+ alive := map[string]bool{}
+ byApp := map[[2]string][]registry.Route{}
+ for _, rt := range routes {
+ alive[rt.Host] = true
+ if rt.Kind != "instance" {
+ continue
+ }
+ k := [2]string{release.RouteAccount(rt), rt.App}
+ if k[0] == "" || k[1] == "" {
+ continue
+ }
+ byApp[k] = append(byApp[k], rt)
+ }
+ var errs []error
+ for k, rts := range byApp {
+ if err := c.convergeApp(ctx, k[0], k[1], rts, now); err != nil {
+ errs = append(errs, fmt.Errorf("%s/%s: %w", k[0], k[1], err))
+ }
+ }
+ if n, err := c.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil {
+ errs = append(errs, err)
+ } else if n > 0 {
+ log.Printf("schedule: pruned %d timer row(s)", n)
+ }
+ return errors.Join(errs...)
+}
+
+func (c *Converger) convergeApp(ctx context.Context, acct, app string, rts []registry.Route, now time.Time) error {
+ k := [2]string{acct, app}
+ raw, etag, err := c.Store.Get(ctx, DeclarationKey(acct, app))
+ var decl Declaration
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ etag = ""
+ case err != nil:
+ return fmt.Errorf("read schedules.json: %w", err)
+ case etag == c.declEtag[k]:
+ decl = c.declDoc[k]
+ default:
+ if err := json.Unmarshal(raw, &decl); err != nil {
+ return fmt.Errorf("parse schedules.json: %w", err)
+ }
+ }
+ c.declEtag[k], c.declDoc[k] = etag, decl
+
+ // Run signal: seed silently on first sight, fire on movement.
+ runRaw, runEtag, err := c.Store.Get(ctx, RunKey(acct, app))
+ fireName := ""
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ runEtag = ""
+ case err != nil:
+ return fmt.Errorf("read schedule-run.json: %w", err)
+ }
+ if prev, seen := c.runEtag[k]; seen && runEtag != prev && runEtag != "" {
+ var sig RunSignal
+ if err := json.Unmarshal(runRaw, &sig); err == nil {
+ if _, ok := decl.Find(sig.Name); ok {
+ fireName = sig.Name
+ } else {
+ log.Printf("schedule: %s/%s: run requested for undeclared schedule %q — ignored", acct, app, sig.Name)
+ }
+ }
+ }
+ c.runEtag[k] = runEtag
+
+ var errs []error
+ for _, rt := range rts {
+ changed, err := c.convergeHost(rt, decl, fireName, now)
+ if err != nil {
+ errs = append(errs, fmt.Errorf("%s: %w", rt.Host, err))
+ continue
+ }
+ if changed && c.Report != nil {
+ st, err := StatusFor(c.Reg, rt, c.Box, now)
+ if err != nil {
+ errs = append(errs, err)
+ continue
+ }
+ c.Report(rt, st)
+ }
+ }
+ return errors.Join(errs...)
+}
+
+// convergeHost returns whether anything about the host's rows changed (or
+// the host is unsupported and should be reported so).
+func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName string, now time.Time) (bool, error) {
+ if rt.Backing == registry.BackingUnit {
+ // Reported once per host, and only when there is something to warn
+ // about — an undeclared app on a unit-backed host is silent.
+ return len(decl.Schedules) > 0 && c.unsupportedUnreported(rt.Host), nil
+ }
+ before, err := c.Reg.TimersFor(rt.Host)
+ if err != nil {
+ return false, err
+ }
+ keep := map[string]bool{}
+ changed := false
+ for _, sc := range decl.Schedules {
+ sp, err := sc.Spec()
+ if err != nil {
+ log.Printf("schedule: %s: schedule %q has an invalid spec (%v) — skipped", rt.Host, sc.Name, err)
+ continue
+ }
+ keep[sc.Name] = true
+ var next int64
+ prev := findTimer(before, sc.Name)
+ if prev == nil || prev.Spec != sp.String() || prev.Path != sc.Path {
+ next = sp.Next(now).Unix()
+ changed = 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()})
+ if errors.Is(err, registry.ErrTimerKind) {
+ log.Printf("schedule: %s: %q is a once-timer the instance registered; the declared schedule of that name is skipped", rt.Host, sc.Name)
+ continue
+ }
+ if err != nil {
+ return changed, err
+ }
+ }
+ n, err := c.Reg.DeleteTimersExcept(rt.Host, KindSchedule, keep)
+ if err != nil {
+ return changed, err
+ }
+ if n > 0 {
+ changed = true
+ }
+ if fireName != "" && keep[fireName] {
+ if err := c.Reg.FireTimerNow(rt.Host, fireName, now.Unix()); err != nil {
+ return changed, err
+ }
+ changed = true
+ }
+ return changed, nil
+}
+
+// unsupportedUnreported reports whether host has NOT yet had its
+// unsupported-host report sent, and marks it sent — so the report is
+// written once per host's lifetime in this Converger, not every pass.
+func (c *Converger) unsupportedUnreported(host string) bool {
+ if c.reportedUnsupported == nil {
+ c.reportedUnsupported = map[string]bool{}
+ }
+ if c.reportedUnsupported[host] {
+ return false
+ }
+ c.reportedUnsupported[host] = true
+ return true
+}
+
+func findTimer(rows []registry.Timer, name string) *registry.Timer {
+ for i := range rows {
+ if rows[i].Name == name {
+ return &rows[i]
+ }
+ }
+ return nil
+}
+
+const UnsupportedUnit = "unit-backed instance — scheduled work delivers to exec-backed instances in this release"
+
+// 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{}}
+ if rt.Backing == registry.BackingUnit {
+ st.Unsupported = UnsupportedUnit
+ return st, nil
+ }
+ rows, err := reg.TimersFor(rt.Host)
+ if err != nil {
+ return st, err
+ }
+ for _, r := range rows {
+ st.Timers = append(st.Timers, toStatus(r))
+ }
+ return st, nil
+}
+
+// toStatus maps one timers row onto its wire shape. Task 5 (the `carlos
+// 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}
+}
diff --git a/internal/schedule/converge_test.go b/internal/schedule/converge_test.go
new file mode 100644
index 000000000..305507c27
--- /dev/null
+++ b/internal/schedule/converge_test.go
@@ -0,0 +1,154 @@
+package schedule
+
+import (
+ "context"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func convergeFixture(t *testing.T) (*registry.Registry, *release.DirStore) {
+ t.Helper()
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ for _, rt := range []registry.Route{
+ {Host: "a.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true},
+ {Host: "b.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true},
+ {Host: "u.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Backing: registry.BackingUnit},
+ {Host: "other.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "other", Status: "ready"},
+ {Host: "site.bab.oncarlos.com", Kind: "static", Account: "bab", App: "site", Status: "ready"},
+ } {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ return reg, release.NewDirStore(t.TempDir())
+}
+
+func TestConvergeMaterialisesDeclarationPerExecInstance(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+ UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
+ var reports []Status
+ c := NewConverger(reg, st, "box1", func(_ registry.Route, s Status) { reports = append(reports, s) })
+ c.Now = func() time.Time { return now }
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
+ tm, err := reg.GetTimer(h, "sync")
+ if err != nil {
+ t.Fatalf("%s: %v", h, err)
+ }
+ if tm.Kind != KindSchedule || tm.Spec != "every 6h0m0s" || tm.Path != "/jobs/sync" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
+ t.Fatalf("%s: %+v", h, tm)
+ }
+ }
+ if rows, _ := reg.TimersFor("u.bab.oncarlos.com"); len(rows) != 0 {
+ t.Fatal("unit-backed host got rows")
+ }
+ if rows, _ := reg.TimersFor("other.bab.oncarlos.com"); len(rows) != 0 {
+ t.Fatal("undeclared app got rows")
+ }
+ // Reports: a and b with a timer each, u as unsupported. Nothing for site.
+ byHost := map[string]Status{}
+ for _, r := range reports {
+ byHost[r.Host] = r
+ }
+ if len(byHost["a.bab.oncarlos.com"].Timers) != 1 || byHost["a.bab.oncarlos.com"].Box != "box1" {
+ t.Fatalf("report a = %+v", byHost["a.bab.oncarlos.com"])
+ }
+ if byHost["u.bab.oncarlos.com"].Unsupported == "" {
+ t.Fatalf("unit host not reported unsupported: %+v", byHost)
+ }
+ if _, ok := byHost["site.bab.oncarlos.com"]; ok {
+ t.Fatal("static route reported")
+ }
+
+ // 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)
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if len(reports) != 0 {
+ t.Fatalf("unchanged pass reported %d hosts", len(reports))
+ }
+ if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 999 {
+ t.Fatalf("unchanged pass moved next_at: %+v", tm)
+ }
+}
+
+func TestConvergeRemovesUndeclaredAndKeepsOnce(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+ UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
+ UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "gone", Every: "1h", Path: "/jobs/gone"}, now)
+ reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "at 99999999999", Path: "/r", NextAt: 99999999999})
+ c := NewConverger(reg, st, "box1", nil)
+ c.Now = func() time.Time { return now }
+ c.Pass(ctx)
+ RemoveSchedule(ctx, st, "bab", "app", "gone", now)
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ rows, _ := reg.TimersFor("a.bab.oncarlos.com")
+ names := map[string]bool{}
+ for _, r := range rows {
+ names[r.Name] = true
+ }
+ if !names["sync"] || !names["remind"] || names["gone"] {
+ t.Fatalf("rows = %v", names)
+ }
+}
+
+func TestConvergeRunSignalFiresOnceAcrossInstances(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+ UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
+ c := NewConverger(reg, st, "box1", nil)
+ c.Now = func() time.Time { return now }
+ c.Pass(ctx)
+ // A run signal that predates this converger is history, not a request.
+ TouchRun(ctx, st, "bab", "app", "sync", now)
+ c2 := NewConverger(reg, st, "box1", nil)
+ c2.Now = func() time.Time { return now }
+ c2.Pass(ctx)
+ if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt == now.Unix() {
+ t.Fatal("a pre-existing run signal was replayed at seed")
+ }
+ TouchRun(ctx, st, "bab", "app", "sync", now)
+ c2.Pass(ctx)
+ for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
+ if tm, _ := reg.GetTimer(h, "sync"); tm.NextAt != now.Unix() {
+ t.Fatalf("%s not fired: %+v", h, tm)
+ }
+ }
+ reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 1, 0, 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")
+ }
+}
+
+func TestConvergePrunesRemovedHosts(t *testing.T) {
+ reg, st := convergeFixture(t)
+ ctx := context.Background()
+ reg.UpsertTimer(registry.Timer{Host: "removed.bab.oncarlos.com", Name: "x", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/p", NextAt: 1})
+ c := NewConverger(reg, st, "box1", nil)
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := reg.TimersFor("removed.bab.oncarlos.com"); len(rows) != 0 {
+ t.Fatal("orphan rows survived")
+ }
+}