amadan

carlos / platform Public

schedule: converge.go — drop dead guard, why-comment, four edge-case tests

k[0] (release.RouteAccount(rt)) can never be "" — its floor is
release.OpsAccount — so that half of the guard was dead; drop it and
note why only k[1] is still checked.

convergeApp reads schedules.json directly instead of calling
ReadDeclaration so it can compare the etag against its cache BEFORE
deciding whether to unmarshal (an unchanged declaration costs one GET
and zero parses); add the comment explaining why it doesn't just call
the shared reader.

Add the four edge cases the reviewer named: a declaration going
absent then present again (etag cache must not stick); a run signal
going absent (must read as no-movement, not a fire, and must not
block a later real touch); a zero-route registry (Pass must not
error when byApp is empty); and a present-but-empty declaration
(RemoveSchedule rewrites, never deletes, the key).
Paul Campbell pushed by paul@keymail.dev a213774dbef06f27f54f5e30235716297fd34cf0 parent 3df1e7f
2 files changed, +146 −1
  • internal/schedule/converge.go +7 −1
  • internal/schedule/converge_test.go +139 −0
diff --git a/internal/schedule/converge.go b/internal/schedule/converge.go
index fbe383b81..c17df56bd 100644
--- a/internal/schedule/converge.go
+++ b/internal/schedule/converge.go
@@ -99,7 +99,7 @@ func (c *Converger) Pass(ctx context.Context) error {
continue
}
k := [2]string{release.RouteAccount(rt), rt.App}
- if k[0] == "" || k[1] == "" {
+ if k[1] == "" { // no k[0] check: RouteAccount's floor is release.OpsAccount, never ""
continue
}
byApp[k] = append(byApp[k], rt)
@@ -120,6 +120,12 @@ func (c *Converger) Pass(ctx context.Context) error {
func (c *Converger) convergeApp(ctx context.Context, acct, app string, rts []registry.Route, now time.Time) error {
k := [2]string{acct, app}
+ // Reads the object directly rather than calling ReadDeclaration: the
+ // etag has to be compared against declEtag[k] BEFORE deciding whether
+ // to unmarshal, so an unchanged declaration costs one GET and zero JSON
+ // parses (the struct's own doc comment's promise). ReadDeclaration
+ // always unmarshals what it reads — right for a one-off caller, wrong
+ // for a loop that runs every 2s per app.
raw, etag, err := c.Store.Get(ctx, DeclarationKey(acct, app))
var decl Declaration
switch {
diff --git a/internal/schedule/converge_test.go b/internal/schedule/converge_test.go
index 2c6d6ba4c..5f64f4e28 100644
--- a/internal/schedule/converge_test.go
+++ b/internal/schedule/converge_test.go
@@ -349,6 +349,145 @@ func TestConvergePassIsSafeUnderConcurrentCallers(t *testing.T) {
}
}
+// The declaration object itself can go away (not just move to an empty
+// schedules list — that is TestConvergeHandlesPresentButEmptyDeclaration
+// below) and later reappear: e.g. hand-deleted from the bucket, or an app
+// re-created after deletion (spec §4.1: absence is legal forever). Its
+// absence must clear a host's rows exactly like an empty declaration does,
+// and the cached "" etag from the absent read must not be mistaken for a
+// match against whatever real etag the object comes back with.
+func TestConvergeDeclarationGoingAbsentThenPresentAgain(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 }
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 {
+ t.Fatalf("setup: want one row, got %+v", rows)
+ }
+
+ if err := st.Delete(ctx, DeclarationKey("bab", "app")); err != nil {
+ t.Fatal(err)
+ }
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 0 {
+ t.Fatalf("row survived the declaration going absent: %+v", rows)
+ }
+
+ UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 {
+ t.Fatalf("row not re-materialised once the declaration came back: %+v", rows)
+ }
+}
+
+// RemoveSchedule rewrites schedules.json with an empty Schedules slice; it
+// never deletes the key (spec §4.1's "absence is legal forever" is about a
+// key that was never written, not one a member emptied out). A
+// present-but-empty declaration must prune existing rows exactly like an
+// absent one, without erroring on the zero-length Schedules slice.
+func TestConvergeHandlesPresentButEmptyDeclaration(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 }
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 {
+ t.Fatalf("setup: want one row, got %+v", rows)
+ }
+
+ if _, present, err := RemoveSchedule(ctx, st, "bab", "app", "sync", now); err != nil || !present {
+ t.Fatalf("remove: present=%v err=%v", present, err)
+ }
+ if _, etag, err := ReadDeclaration(ctx, st, "bab", "app"); err != nil || etag == "" {
+ t.Fatalf("declaration should still be present (non-empty etag) after removing its only schedule: %q %v", etag, err)
+ }
+
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 0 {
+ t.Fatalf("row survived a present-but-empty declaration: %+v", rows)
+ }
+}
+
+// The run signal object can itself go away between passes (no deletion API
+// is designed for it, but nothing stops a hand-edited bucket). Its absence
+// must read as "no movement" — not as a fire, and not as a poisoned cache
+// that can no longer recognise a later, genuinely new touch.
+func TestConvergeRunSignalGoingAbsentIsNotAFire(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 }
+ if err := c.Pass(ctx); err != nil { // seed
+ t.Fatal(err)
+ }
+ TouchRun(ctx, st, "bab", "app", "sync", now)
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 1, 0, 777, 777)
+ if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
+ t.Fatalf("setup: run signal did not fire: %+v", tm)
+ }
+
+ if err := st.Delete(ctx, RunKey("bab", "app")); err != nil {
+ t.Fatal(err)
+ }
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
+ t.Fatalf("run signal going absent misfired: %+v", tm)
+ }
+
+ // A later, genuinely new touch must still be recognised.
+ TouchRun(ctx, st, "bab", "app", "sync", now)
+ if err := c.Pass(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != now.Unix() {
+ t.Fatalf("run signal did not fire once the object came back: %+v", tm)
+ }
+}
+
+// A registry with zero routes at all (nothing adopted yet, or every route
+// gone) must not be distinguished from "an app with no instance routes":
+// byApp never gains an entry, convergeApp is never called, and Pass must
+// still return cleanly rather than erroring on an empty registry.
+func TestConvergeZeroRouteRegistryIsANoop(t *testing.T) {
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ st := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ now := at("2026-08-23T13:07:00Z")
+ // A declaration exists for an app the registry has never heard of.
+ 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 }
+ if err := c.Pass(ctx); err != nil {
+ t.Fatalf("Pass on a route-less registry: %v", err)
+ }
+}
+
// 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