amadan
schedule: converge — kind-aware baseline, drop rows on unit flip
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2 files changed,
+114
−6
internal/schedule/converge.go+33 −6internal/schedule/converge_test.go+81 −0
diff --git a/internal/schedule/converge.go b/internal/schedule/converge.go| index dea12c171..803c9102b 100644 |
| --- a/internal/schedule/converge.go |
| +++ b/internal/schedule/converge.go |
| @@ -146,9 +146,21 @@ func (c *Converger) convergeApp(ctx context.Context, acct, app string, rts []reg |
| // 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 { |
| + // A host can flip exec -> unit mid-lifetime (the app-capabilities |
| + // migration door, registry.SetBacking): drop any schedule rows it |
| + // still carries from before the flip, so "no rows" stays true for |
| + // every unit-backed host, not just ones that were always unit-backed. |
| + n, err := c.Reg.DeleteTimersExcept(rt.Host, KindSchedule, map[string]bool{}) |
| + if err != nil { |
| + return false, err |
| + } |
| + changed := n > 0 |
| // 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 |
| + if len(decl.Schedules) > 0 && c.unsupportedUnreported(rt.Host) { |
| + changed = true |
| + } |
| + return changed, nil |
| } |
| before, err := c.Reg.TimersFor(rt.Host) |
| if err != nil { |
| @@ -163,20 +175,29 @@ func (c *Converger) convergeHost(rt registry.Route, decl Declaration, fireName s |
| continue |
| } |
| keep[sc.Name] = true |
| + // prev is the ONLY-schedule-kind baseline: a same-named once-row |
| + // (the instance's own, off its control socket) is never "previous |
| + // state" for this comparison — a kind collision decides itself |
| + // below, at the write, not here. |
| + prev := findTimer(before, sc.Name, KindSchedule) |
| + wouldChange := prev == nil || prev.Spec != sp.String() || prev.Path != sc.Path |
| var next int64 |
| - prev := findTimer(before, sc.Name) |
| - if prev == nil || prev.Spec != sp.String() || prev.Path != sc.Path { |
| + if wouldChange { |
| 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) { |
| + // Nothing was written — a collision must not pin `changed`, |
| + // or a permanently-colliding name reports every single pass. |
| 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 |
| } |
| + if wouldChange { |
| + changed = true |
| + } |
| } |
| n, err := c.Reg.DeleteTimersExcept(rt.Host, KindSchedule, keep) |
| if err != nil { |
| @@ -208,9 +229,15 @@ func (c *Converger) unsupportedUnreported(host string) bool { |
| return true |
| } |
| -func findTimer(rows []registry.Timer, name string) *registry.Timer { |
| +// findTimer returns the row named name with the given kind, or nil. Kind |
| +// matters: a once-row and a declared schedule can share a name (the |
| +// once-row is the instance's own, registered off its control socket), and |
| +// only a same-kind row is ever "previous state" for the change comparison |
| +// below — a kind collision is a skip (ErrTimerKind, at the UpsertTimer |
| +// call), never a manufactured "changed". |
| +func findTimer(rows []registry.Timer, name, kind string) *registry.Timer { |
| for i := range rows { |
| - if rows[i].Name == name { |
| + if rows[i].Name == name && rows[i].Kind == kind { |
| return &rows[i] |
| } |
| } |
diff --git a/internal/schedule/converge_test.go b/internal/schedule/converge_test.go| index 305507c27..321d047f7 100644 |
| --- a/internal/schedule/converge_test.go |
| +++ b/internal/schedule/converge_test.go |
| @@ -152,3 +152,84 @@ func TestConvergePrunesRemovedHosts(t *testing.T) { |
| t.Fatal("orphan rows survived") |
| } |
| } |
| + |
| +// TestConvergeNameCollisionWithOnceRowSkippedSilently pins the fix for a |
| +// found bug: a declared schedule sharing a name with an existing once-row |
| +// (the instance's own, registered off its control socket) must never be |
| +// treated as "previous state" for the changed comparison — the once-row's |
| +// spec text ("at ...") always differs from the declared schedule's, and |
| +// comparing against it pinned `changed` (and hence a Report) on every |
| +// single pass forever, never converging quiet even though nothing was |
| +// ever actually written. |
| +func TestConvergeNameCollisionWithOnceRowSkippedSilently(t *testing.T) { |
| + reg, st := convergeFixture(t) |
| + ctx := context.Background() |
| + now := at("2026-08-23T13:07:00Z") |
| + if err := reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindOnce, Spec: "at 99999999999", Path: "/r", NextAt: 99999999999}); err != nil { |
| + t.Fatal(err) |
| + } |
| + 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) |
| + } |
| + if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 || rows[0].Kind != KindOnce || rows[0].NextAt != 99999999999 { |
| + t.Fatalf("once row not left untouched / a schedule row was created: %+v", rows) |
| + } |
| + |
| + reports = nil |
| + if err := c.Pass(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 || rows[0].Kind != KindOnce { |
| + t.Fatalf("once row disturbed on the second pass: %+v", rows) |
| + } |
| + for _, r := range reports { |
| + if r.Host == "a.bab.oncarlos.com" { |
| + t.Fatalf("second pass reported the colliding host: %+v", r) |
| + } |
| + } |
| +} |
| + |
| +// TestConvergeDropsScheduleRowsOnUnitFlip pins the fix for a found bug: the |
| +// unit-backed early return skipped TimersFor/DeleteTimersExcept entirely, |
| +// so a host that flips from exec-backed to unit-backed (registry.SetBacking |
| +// — the app-capabilities migration door) kept its old schedule rows |
| +// forever, violating "a unit-backed host has no rows". |
| +func TestConvergeDropsScheduleRowsOnUnitFlip(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: expected one schedule row before the flip, got %+v", rows) |
| + } |
| + |
| + if err := reg.SetBacking("a.bab.oncarlos.com", registry.BackingUnit); err != nil { |
| + t.Fatal(err) |
| + } |
| + var reports []Status |
| + c.Report = func(_ registry.Route, s Status) { reports = append(reports, s) } |
| + if err := c.Pass(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 0 { |
| + t.Fatalf("schedule rows survived a flip to unit-backed: %+v", rows) |
| + } |
| + var got *Status |
| + for i := range reports { |
| + if reports[i].Host == "a.bab.oncarlos.com" { |
| + got = &reports[i] |
| + } |
| + } |
| + if got == nil || got.Unsupported == "" { |
| + t.Fatalf("no unsupported report for the flipped host: %+v", reports) |
| + } |
| +} |