amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package schedule
2
3import (
4 "context"
5 "path/filepath"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/carlosframework/platform/internal/registry"
11 "github.com/carlosframework/platform/internal/release"
12)
13
14func TestHoldsAwakeWhenTheIntervalIsInsideTheIdleWindow(t *testing.T) {
15 now := time.Unix(1788120000, 0).UTC()
16 cases := []struct {
17 name string
18 kind string
19 spec string
20 idle time.Duration
21 want bool
22 }{
23 {"tighter than the window", KindSchedule, "every 5m", 15 * time.Minute, true},
24 {"exactly the window", KindSchedule, "every 15m", 15 * time.Minute, true},
25 {"wider than the window", KindSchedule, "every 1h", 15 * time.Minute, false},
26 {"tight, short window", KindSchedule, "every 5m", 90 * time.Second, false},
27 {"a once-timer never holds", KindOnce, "at 1788130000", 15 * time.Minute, false},
28 {"hibernation not armed", KindSchedule, "every 1m", 0, false},
29 {"hourly cron", KindSchedule, "cron 0 * * * *", 15 * time.Minute, false},
30 {"every-minute cron", KindSchedule, "cron * * * * *", 15 * time.Minute, true},
31 {"unparseable spec is not a claim", KindSchedule, "banana", 15 * time.Minute, false},
32 }
33 for _, tc := range cases {
34 t.Run(tc.name, func(t *testing.T) {
35 got := holdsAwake(registry.Timer{Kind: tc.kind, Spec: tc.spec}, tc.idle, now)
36 if got != tc.want {
37 t.Errorf("holdsAwake(%q, idle=%s) = %v, want %v", tc.spec, tc.idle, got, tc.want)
38 }
39 })
40 }
41}
42
43// Hibernation is armed PER ROUTE (rt.Hibernate), not just per box: a route
44// with Hibernate off was never going to sleep, so no schedule on it can
45// hold anything awake, however tight. StatusFor must zero the window before
46// holdsAwake ever sees it — a box-wide idle window alone is not enough.
47func TestStatusForNeverClaimsHoldsAwakeOffARouteWithHibernateOff(t *testing.T) {
48 reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
49 if err != nil {
50 t.Fatal(err)
51 }
52 t.Cleanup(func() { reg.Close() })
53 rt := registry.Route{Host: "resident.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: false}
54 if err := reg.Upsert(rt); err != nil {
55 t.Fatal(err)
56 }
57 now := at("2026-08-23T13:07:00Z")
58 if err := reg.UpsertTimer(registry.Timer{Host: rt.Host, Name: "sweep", Kind: KindSchedule, Spec: "every 5m", Path: "/jobs/sweep", CreatedAt: now.Unix()}); err != nil {
59 t.Fatal(err)
60 }
61 st, err := StatusFor(reg, rt, "box1", now, 15*time.Minute)
62 if err != nil {
63 t.Fatal(err)
64 }
65 if len(st.Timers) != 1 || st.Timers[0].HoldsAwake {
66 t.Fatalf("HoldsAwake = %v on a Hibernate:false route, want false regardless of the box's idle window", st.Timers[0].HoldsAwake)
67 }
68}
69
70func convergeFixture(t *testing.T) (*registry.Registry, *release.DirStore) {
71 t.Helper()
72 reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
73 if err != nil {
74 t.Fatal(err)
75 }
76 t.Cleanup(func() { reg.Close() })
77 for _, rt := range []registry.Route{
78 {Host: "a.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true},
79 {Host: "b.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true},
80 {Host: "u.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Backing: registry.BackingUnit},
81 {Host: "other.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "other", Status: "ready"},
82 {Host: "site.bab.oncarlos.com", Kind: "static", Account: "bab", App: "site", Status: "ready"},
83 } {
84 if err := reg.Upsert(rt); err != nil {
85 t.Fatal(err)
86 }
87 }
88 return reg, release.NewDirStore(t.TempDir())
89}
90
91func TestConvergeMaterialisesDeclarationPerExecInstance(t *testing.T) {
92 reg, st := convergeFixture(t)
93 ctx := context.Background()
94 now := at("2026-08-23T13:07:00Z")
95 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
96 var reports []Status
97 c := NewConverger(reg, st, "box1", func(_ registry.Route, s Status) { reports = append(reports, s) })
98 c.Now = func() time.Time { return now }
99 if err := c.Pass(ctx); err != nil {
100 t.Fatal(err)
101 }
102 for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
103 tm, err := reg.GetTimer(h, "sync")
104 if err != nil {
105 t.Fatalf("%s: %v", h, err)
106 }
107 if tm.Kind != KindSchedule || tm.Spec != "every 6h" || tm.Path != "/jobs/sync" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
108 t.Fatalf("%s: %+v", h, tm)
109 }
110 }
111 if rows, _ := reg.TimersFor("u.bab.oncarlos.com"); len(rows) != 0 {
112 t.Fatal("unit-backed host got rows")
113 }
114 if rows, _ := reg.TimersFor("other.bab.oncarlos.com"); len(rows) != 0 {
115 t.Fatal("undeclared app got rows")
116 }
117 // Reports: a and b with a timer each, u as unsupported. Nothing for site.
118 byHost := map[string]Status{}
119 for _, r := range reports {
120 byHost[r.Host] = r
121 }
122 if len(byHost["a.bab.oncarlos.com"].Timers) != 1 || byHost["a.bab.oncarlos.com"].Box != "box1" {
123 t.Fatalf("report a = %+v", byHost["a.bab.oncarlos.com"])
124 }
125 if byHost["u.bab.oncarlos.com"].Unsupported == "" {
126 t.Fatalf("unit host not reported unsupported: %+v", byHost)
127 }
128 if _, ok := byHost["site.bab.oncarlos.com"]; ok {
129 t.Fatal("static route reported")
130 }
131
132 // A second pass with nothing changed: no new reports, next_at untouched.
133 reports = nil
134 reg.MarkTimerFired("a.bab.oncarlos.com", "sync", scheduleTimerRegistration(t, reg, "a.bab.oncarlos.com", "sync"), now.Unix(), "204", 10, 0, 999, 999)
135 if err := c.Pass(ctx); err != nil {
136 t.Fatal(err)
137 }
138 if len(reports) != 0 {
139 t.Fatalf("unchanged pass reported %d hosts", len(reports))
140 }
141 if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 999 {
142 t.Fatalf("unchanged pass moved next_at: %+v", tm)
143 }
144}
145
146func TestConvergeRemovesUndeclaredAndKeepsOnce(t *testing.T) {
147 reg, st := convergeFixture(t)
148 ctx := context.Background()
149 now := at("2026-08-23T13:07:00Z")
150 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
151 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "gone", Every: "1h", Path: "/jobs/gone"}, now)
152 reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "at 99999999999", Path: "/r", NextAt: 99999999999})
153 c := NewConverger(reg, st, "box1", nil)
154 c.Now = func() time.Time { return now }
155 c.Pass(ctx)
156 RemoveSchedule(ctx, st, "bab", "app", "gone", now)
157 if err := c.Pass(ctx); err != nil {
158 t.Fatal(err)
159 }
160 rows, _ := reg.TimersFor("a.bab.oncarlos.com")
161 names := map[string]bool{}
162 for _, r := range rows {
163 names[r.Name] = true
164 }
165 if !names["sync"] || !names["remind"] || names["gone"] {
166 t.Fatalf("rows = %v", names)
167 }
168}
169
170func TestConvergeRunSignalFiresOnceAcrossInstances(t *testing.T) {
171 reg, st := convergeFixture(t)
172 ctx := context.Background()
173 now := at("2026-08-23T13:07:00Z")
174 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
175 c := NewConverger(reg, st, "box1", nil)
176 c.Now = func() time.Time { return now }
177 c.Pass(ctx)
178 // A run signal that predates this converger is history, not a request.
179 TouchRun(ctx, st, "bab", "app", "sync", now)
180 c2 := NewConverger(reg, st, "box1", nil)
181 c2.Now = func() time.Time { return now }
182 c2.Pass(ctx)
183 if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt == now.Unix() {
184 t.Fatal("a pre-existing run signal was replayed at seed")
185 }
186 TouchRun(ctx, st, "bab", "app", "sync", now)
187 c2.Pass(ctx)
188 for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
189 if tm, _ := reg.GetTimer(h, "sync"); tm.NextAt != now.Unix() {
190 t.Fatalf("%s not fired: %+v", h, tm)
191 }
192 }
193 reg.MarkTimerFired("a.bab.oncarlos.com", "sync", scheduleTimerRegistration(t, reg, "a.bab.oncarlos.com", "sync"), now.Unix(), "204", 1, 0, 777, 777)
194 c2.Pass(ctx) // same etag: must not re-fire
195 if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
196 t.Fatal("run signal replayed on an unchanged etag")
197 }
198}
199
200func TestConvergePrunesRemovedHosts(t *testing.T) {
201 reg, st := convergeFixture(t)
202 ctx := context.Background()
203 reg.UpsertTimer(registry.Timer{Host: "removed.bab.oncarlos.com", Name: "x", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/p", NextAt: 1})
204 c := NewConverger(reg, st, "box1", nil)
205 if err := c.Pass(ctx); err != nil {
206 t.Fatal(err)
207 }
208 if rows, _ := reg.TimersFor("removed.bab.oncarlos.com"); len(rows) != 0 {
209 t.Fatal("orphan rows survived")
210 }
211}
212
213// TestConvergeNameCollisionWithOnceRowSkippedSilently pins the fix for a
214// found bug: a declared schedule sharing a name with an existing once-row
215// (the instance's own, registered off its control socket) must never be
216// treated as "previous state" for the changed comparison — the once-row's
217// spec text ("at ...") always differs from the declared schedule's, and
218// comparing against it pinned `changed` (and hence a Report) on every
219// single pass forever, never converging quiet even though nothing was
220// ever actually written.
221func TestConvergeNameCollisionWithOnceRowSkippedSilently(t *testing.T) {
222 reg, st := convergeFixture(t)
223 ctx := context.Background()
224 now := at("2026-08-23T13:07:00Z")
225 if err := reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindOnce, Spec: "at 99999999999", Path: "/r", NextAt: 99999999999}); err != nil {
226 t.Fatal(err)
227 }
228 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
229 var reports []Status
230 c := NewConverger(reg, st, "box1", func(_ registry.Route, s Status) { reports = append(reports, s) })
231 c.Now = func() time.Time { return now }
232 if err := c.Pass(ctx); err != nil {
233 t.Fatal(err)
234 }
235 if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 || rows[0].Kind != KindOnce || rows[0].NextAt != 99999999999 {
236 t.Fatalf("once row not left untouched / a schedule row was created: %+v", rows)
237 }
238
239 reports = nil
240 if err := c.Pass(ctx); err != nil {
241 t.Fatal(err)
242 }
243 if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 || rows[0].Kind != KindOnce {
244 t.Fatalf("once row disturbed on the second pass: %+v", rows)
245 }
246 for _, r := range reports {
247 if r.Host == "a.bab.oncarlos.com" {
248 t.Fatalf("second pass reported the colliding host: %+v", r)
249 }
250 }
251}
252
253// TestConvergeDropsScheduleRowsOnUnitFlip pins the fix for a found bug: the
254// unit-backed early return skipped TimersFor/DeleteTimersExcept entirely,
255// so a host that flips from exec-backed to unit-backed (registry.SetBacking
256// — the app-capabilities migration door) kept its old schedule rows
257// forever, violating "a unit-backed host has no rows".
258func TestConvergeDropsScheduleRowsOnUnitFlip(t *testing.T) {
259 reg, st := convergeFixture(t)
260 ctx := context.Background()
261 now := at("2026-08-23T13:07:00Z")
262 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
263 c := NewConverger(reg, st, "box1", nil)
264 c.Now = func() time.Time { return now }
265 if err := c.Pass(ctx); err != nil {
266 t.Fatal(err)
267 }
268 if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 {
269 t.Fatalf("setup: expected one schedule row before the flip, got %+v", rows)
270 }
271
272 if err := reg.SetBacking("a.bab.oncarlos.com", registry.BackingUnit); err != nil {
273 t.Fatal(err)
274 }
275 var reports []Status
276 c.Report = func(_ registry.Route, s Status) { reports = append(reports, s) }
277 if err := c.Pass(ctx); err != nil {
278 t.Fatal(err)
279 }
280 if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 0 {
281 t.Fatalf("schedule rows survived a flip to unit-backed: %+v", rows)
282 }
283 var got *Status
284 for i := range reports {
285 if reports[i].Host == "a.bab.oncarlos.com" {
286 got = &reports[i]
287 }
288 }
289 if got == nil || got.Unsupported == "" {
290 t.Fatalf("no unsupported report for the flipped host: %+v", reports)
291 }
292}
293
294// gateStore blocks every Get until open is closed, recording the keys it was
295// asked for. It is how the two tests below observe whether a second Pass got
296// past the lock while the first one was parked mid-pass in the store.
297type gateStore struct {
298 release.Store
299 entered chan string
300 open chan struct{}
301}
302
303func (g *gateStore) Get(ctx context.Context, key string) ([]byte, string, error) {
304 g.entered <- key
305 <-g.open
306 return g.Store.Get(ctx, key)
307}
308
309// TestConvergePassHoldsTheLockForTheWholePass pins the serialization the agent
310// depends on: fastConvergeLoop (seed + serial leg) and adoptLoop (hourly floor)
311// call Pass on ONE shared Converger from two goroutines, and they genuinely
312// overlap. Without the lock the etag maps take a concurrent read and write —
313// a runtime throw that kills the agent and the activator with it — and two
314// interleaved passes can fire the same manual run twice.
315//
316// The store parks the first pass inside its very first Get, still holding the
317// lock; a second Pass must not reach the store at all until it is released.
318func TestConvergePassHoldsTheLockForTheWholePass(t *testing.T) {
319 reg, st := convergeFixture(t)
320 ctx := context.Background()
321 now := at("2026-08-23T13:07:00Z")
322 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
323 gate := &gateStore{Store: st, entered: make(chan string, 64), open: make(chan struct{})}
324 c := NewConverger(reg, gate, "box1", nil)
325 c.Now = func() time.Time { return now }
326
327 first := make(chan error, 1)
328 go func() { first <- c.Pass(ctx) }()
329 <-gate.entered // pass 1 is inside a Get, holding the lock
330
331 second := make(chan error, 1)
332 go func() { second <- c.Pass(ctx) }()
333 select {
334 case k := <-gate.entered:
335 // Only pass 2 can be here: pass 1 is parked inside its Get and cannot
336 // issue another until open closes.
337 t.Fatalf("a second Pass reached the store (%s) while the first held the lock", k)
338 case <-time.After(100 * time.Millisecond):
339 // Blocked, as it must be. A slow machine can only weaken this check,
340 // never fail it spuriously.
341 }
342 close(gate.open)
343 if err := <-first; err != nil {
344 t.Fatalf("first pass: %v", err)
345 }
346 if err := <-second; err != nil {
347 t.Fatalf("second pass: %v", err)
348 }
349 if len(gate.entered) == 0 {
350 t.Fatal("the gate store was never consulted — this test pinned nothing")
351 }
352}
353
354// TestConvergePassIsSafeUnderConcurrentCallers is the -race half of the same
355// finding, and the exactly-once half of it: four passes racing on one
356// Converger must leave the manual run fired ONCE, with the etag recorded so a
357// later pass does not replay it. The Report closure appends to a plain slice
358// on purpose — Pass calls it under the lock, so a caller's ordinary,
359// non-thread-safe callback is safe, and this would flag under -race if that
360// ever stopped being true.
361func TestConvergePassIsSafeUnderConcurrentCallers(t *testing.T) {
362 reg, st := convergeFixture(t)
363 ctx := context.Background()
364 now := at("2026-08-23T13:07:00Z")
365 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
366 var reports []Status
367 c := NewConverger(reg, st, "box1", func(_ registry.Route, s Status) { reports = append(reports, s) })
368 c.Now = func() time.Time { return now }
369 if err := c.Pass(ctx); err != nil { // seed: a run signal older than the converger is history
370 t.Fatal(err)
371 }
372 TouchRun(ctx, st, "bab", "app", "sync", now)
373
374 var wg sync.WaitGroup
375 errs := make(chan error, 8)
376 for i := 0; i < 4; i++ {
377 wg.Add(1)
378 go func() {
379 defer wg.Done()
380 if err := c.Pass(ctx); err != nil {
381 errs <- err
382 }
383 }()
384 }
385 wg.Wait()
386 close(errs)
387 for err := range errs {
388 t.Fatalf("concurrent Pass: %v", err)
389 }
390 for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
391 if tm, _ := reg.GetTimer(h, "sync"); tm.NextAt != now.Unix() {
392 t.Fatalf("%s: run signal not fired: %+v", h, tm)
393 }
394 }
395 // Exactly-once, COUNTED: a run signal that fired twice is one that got
396 // delivered twice, so the assertion runs the rows through a real Runner
397 // and counts the deliveries rather than inferring from next_at. Four
398 // racing passes, one tick each host — and a later pass, which sees the
399 // recorded etag, adds no second delivery.
400 d := &fakeDeliverer{status: 204}
401 r := &Runner{Reg: reg, Deliver: d, Timeout: time.Minute, Box: "box1", Build: "test",
402 Now: func() time.Time { return now }, Token: func(string) (string, error) { return "tok", nil }}
403 r.tick()
404 fired := d.firedHosts()
405 for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
406 if fired[h] != 1 {
407 t.Fatalf("%s delivered %d times, want exactly 1 (fired = %v)", h, fired[h], fired)
408 }
409 }
410 if err := c.Pass(ctx); err != nil {
411 t.Fatal(err)
412 }
413 r.tick()
414 if got := d.firedHosts(); len(got) != len(fired) || got["a.bab.oncarlos.com"] != 1 || got["b.bab.oncarlos.com"] != 1 {
415 t.Fatalf("the run signal replayed after the concurrent passes: deliveries now %v, were %v", got, fired)
416 }
417 if len(reports) == 0 {
418 t.Fatal("no reports at all — the fixture converged nothing")
419 }
420}
421
422// The declaration object itself can go away (not just move to an empty
423// schedules list — that is TestConvergeHandlesPresentButEmptyDeclaration
424// below) and later reappear: e.g. hand-deleted from the bucket, or an app
425// re-created after deletion (spec §4.1: absence is legal forever). Its
426// absence must clear a host's rows exactly like an empty declaration does,
427// and the cached "" etag from the absent read must not be mistaken for a
428// match against whatever real etag the object comes back with.
429func TestConvergeDeclarationGoingAbsentThenPresentAgain(t *testing.T) {
430 reg, st := convergeFixture(t)
431 ctx := context.Background()
432 now := at("2026-08-23T13:07:00Z")
433 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
434 c := NewConverger(reg, st, "box1", nil)
435 c.Now = func() time.Time { return now }
436 if err := c.Pass(ctx); err != nil {
437 t.Fatal(err)
438 }
439 if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 {
440 t.Fatalf("setup: want one row, got %+v", rows)
441 }
442
443 if err := st.Delete(ctx, DeclarationKey("bab", "app")); err != nil {
444 t.Fatal(err)
445 }
446 if err := c.Pass(ctx); err != nil {
447 t.Fatal(err)
448 }
449 if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 0 {
450 t.Fatalf("row survived the declaration going absent: %+v", rows)
451 }
452
453 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
454 if err := c.Pass(ctx); err != nil {
455 t.Fatal(err)
456 }
457 if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 {
458 t.Fatalf("row not re-materialised once the declaration came back: %+v", rows)
459 }
460}
461
462// RemoveSchedule rewrites schedules.json with an empty Schedules slice; it
463// never deletes the key (spec §4.1's "absence is legal forever" is about a
464// key that was never written, not one a member emptied out). A
465// present-but-empty declaration must prune existing rows exactly like an
466// absent one, without erroring on the zero-length Schedules slice.
467func TestConvergeHandlesPresentButEmptyDeclaration(t *testing.T) {
468 reg, st := convergeFixture(t)
469 ctx := context.Background()
470 now := at("2026-08-23T13:07:00Z")
471 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
472 c := NewConverger(reg, st, "box1", nil)
473 c.Now = func() time.Time { return now }
474 if err := c.Pass(ctx); err != nil {
475 t.Fatal(err)
476 }
477 if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 {
478 t.Fatalf("setup: want one row, got %+v", rows)
479 }
480
481 if _, present, err := RemoveSchedule(ctx, st, "bab", "app", "sync", now); err != nil || !present {
482 t.Fatalf("remove: present=%v err=%v", present, err)
483 }
484 if _, etag, err := ReadDeclaration(ctx, st, "bab", "app"); err != nil || etag == "" {
485 t.Fatalf("declaration should still be present (non-empty etag) after removing its only schedule: %q %v", etag, err)
486 }
487
488 if err := c.Pass(ctx); err != nil {
489 t.Fatal(err)
490 }
491 if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 0 {
492 t.Fatalf("row survived a present-but-empty declaration: %+v", rows)
493 }
494}
495
496// The run signal object can itself go away between passes (no deletion API
497// is designed for it, but nothing stops a hand-edited bucket). Its absence
498// must read as "no movement" — not as a fire, and not as a poisoned cache
499// that can no longer recognise a later, genuinely new touch.
500func TestConvergeRunSignalGoingAbsentIsNotAFire(t *testing.T) {
501 reg, st := convergeFixture(t)
502 ctx := context.Background()
503 now := at("2026-08-23T13:07:00Z")
504 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
505 c := NewConverger(reg, st, "box1", nil)
506 c.Now = func() time.Time { return now }
507 if err := c.Pass(ctx); err != nil { // seed
508 t.Fatal(err)
509 }
510 TouchRun(ctx, st, "bab", "app", "sync", now)
511 if err := c.Pass(ctx); err != nil {
512 t.Fatal(err)
513 }
514 reg.MarkTimerFired("a.bab.oncarlos.com", "sync", scheduleTimerRegistration(t, reg, "a.bab.oncarlos.com", "sync"), now.Unix(), "204", 1, 0, 777, 777)
515 if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
516 t.Fatalf("setup: run signal did not fire: %+v", tm)
517 }
518
519 if err := st.Delete(ctx, RunKey("bab", "app")); err != nil {
520 t.Fatal(err)
521 }
522 if err := c.Pass(ctx); err != nil {
523 t.Fatal(err)
524 }
525 if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
526 t.Fatalf("run signal going absent misfired: %+v", tm)
527 }
528
529 // A later, genuinely new touch must still be recognised.
530 TouchRun(ctx, st, "bab", "app", "sync", now)
531 if err := c.Pass(ctx); err != nil {
532 t.Fatal(err)
533 }
534 if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != now.Unix() {
535 t.Fatalf("run signal did not fire once the object came back: %+v", tm)
536 }
537}
538
539// A declaration can exist for an app the registry has never heard of
540// (nothing shipped there yet, or every instance since removed) — byApp
541// never gains an entry for it, convergeApp is never called for it, and
542// Pass must return cleanly rather than erroring. The weak version of this
543// test is a wholly empty registry, where that's true trivially because
544// NOTHING converges; the real claim is that a routeless app is skipped
545// while a SIBLING app with live routes still converges normally in the
546// same pass, so this seeds both: "ghost" (declared, zero routes) beside
547// "other" (declared here, and the fixture already gives it a real route).
548func TestConvergeZeroRouteAppDoesNotBlockASiblingApp(t *testing.T) {
549 reg, st := convergeFixture(t)
550 ctx := context.Background()
551 now := at("2026-08-23T13:07:00Z")
552 UpsertSchedule(ctx, st, "bab", "ghost", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
553 UpsertSchedule(ctx, st, "bab", "other", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
554 c := NewConverger(reg, st, "box1", nil)
555 c.Now = func() time.Time { return now }
556 if err := c.Pass(ctx); err != nil {
557 t.Fatalf("Pass with one routeless app declared alongside a routed sibling: %v", err)
558 }
559 rows, err := reg.TimersFor("other.bab.oncarlos.com")
560 if err != nil {
561 t.Fatal(err)
562 }
563 if len(rows) != 1 || rows[0].Name != "sync" || rows[0].Kind != KindSchedule {
564 t.Fatalf("the routed sibling app (other) did not converge: %+v", rows)
565 }
566 // "ghost" has no host anywhere to hold a row, so there is nothing to
567 // query for it directly — Pass returning cleanly with "other" actually
568 // converged above IS the assertion that byApp skipped "ghost" without
569 // that skip taking its sibling down with it.
570}
571
572// A declaration whose cron never matches (written past Validate — an older
573// console, or by hand) must not become a row due at the zero time's unix
574// value: it is parked at next_at=0 with last_status "never fires", and the
575// runner therefore never sees it.
576func TestConvergeParksAScheduleThatNeverFires(t *testing.T) {
577 reg, st := convergeFixture(t)
578 ctx := context.Background()
579 now := at("2026-08-23T13:07:00Z")
580 raw := []byte(`{"schema":1,"schedules":[{"name":"never","cron":"0 0 30 2 *","path":"/x"}]}`)
581 if _, err := st.Put(ctx, DeclarationKey("bab", "app"), raw, ""); err != nil {
582 t.Fatal(err)
583 }
584 c := NewConverger(reg, st, "box1", nil)
585 c.Now = func() time.Time { return now }
586 if err := c.Pass(ctx); err != nil {
587 t.Fatal(err)
588 }
589 tm, err := reg.GetTimer("a.bab.oncarlos.com", "never")
590 if err != nil {
591 t.Fatal(err)
592 }
593 if tm.NextAt != 0 || tm.DueAt != 0 || tm.LastStatus != StatusNeverFires {
594 t.Fatalf("parked row = %+v, want next_at 0 and %q", tm, StatusNeverFires)
595 }
596 if due, _ := reg.DueTimers(now.Unix(), 10, nil); len(due) != 0 {
597 t.Fatalf("a parked row is due: %+v", due)
598 }
599}
600
601func scheduleTimerRegistration(t *testing.T, reg *registry.Registry, host, name string) string {
602 t.Helper()
603 row, err := reg.GetTimer(host, name)
604 if err != nil {
605 t.Fatal(err)
606 }
607 return row.Registration
608}
609
610// The converger and the cut's barrier: a convergeHost paused after its
611// Frozen check, inside the gate, holds the drain until it has finished
612// writing; a converger that starts after the drain sees Frozen and writes
613// nothing, so the timers snapshot taken after the drain equals the table.
614func TestConvergerCannotWriteAfterTheBarrier(t *testing.T) {
615 reg, st := convergeFixture(t)
616 ctx := context.Background()
617 now := at("2026-08-23T13:07:00Z")
618 const host = "a.bab.oncarlos.com"
619 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
620 gate := &HostGate{}
621 var frozen sync.Map
622 c := NewConverger(reg, st, "box1", nil)
623 c.Now = func() time.Time { return now }
624 c.Gate = gate
625 c.Frozen = func(h string) bool { _, ok := frozen.Load(h); return ok }
626 reached, resume := make(chan struct{}), make(chan struct{})
627 var once sync.Once
628 c.afterFrozenCheck = func(h string) {
629 if h == host {
630 once.Do(func() { close(reached); <-resume })
631 }
632 }
633 passed := make(chan error, 1)
634 go func() { passed <- c.Pass(ctx) }()
635 <-reached
636
637 // The cut: hold, freeze, drain. The paused convergence holds the drain.
638 if err := reg.SetHold(host, "moving", "mv_a"); err != nil {
639 t.Fatal(err)
640 }
641 frozen.Store(host, true)
642 drained := make(chan error, 1)
643 go func() { drained <- gate.Drain(ctx, host) }()
644 select {
645 case err := <-drained:
646 t.Fatalf("the drain returned (%v) while a convergence was inside the gate", err)
647 case <-time.After(50 * time.Millisecond):
648 }
649 close(resume)
650 if err := <-drained; err != nil {
651 t.Fatal(err)
652 }
653 snap, err := reg.TimersFor(host)
654 if err != nil {
655 t.Fatal(err)
656 }
657 if err := <-passed; err != nil {
658 t.Fatal(err)
659 }
660 if len(snap) != 1 || snap[0].Name != "sync" {
661 t.Fatalf("the paused convergence's write is not in the snapshot: %+v", snap)
662 }
663
664 // A later pass, with the declaration changed: frozen, it writes nothing.
665 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "nightly", Every: "24h", Path: "/jobs/nightly"}, now)
666 if _, _, err := RemoveSchedule(ctx, st, "bab", "app", "sync", now); err != nil {
667 t.Fatal(err)
668 }
669 if err := c.Pass(ctx); err != nil {
670 t.Fatal(err)
671 }
672 got, err := reg.TimersFor(host)
673 if err != nil {
674 t.Fatal(err)
675 }
676 if len(got) != len(snap) || got[0] != snap[0] {
677 t.Fatalf("the table moved after the barrier:\n got %+v\nwant %+v", got, snap)
678 }
679 // The other host of the app, not frozen, converged as usual.
680 if _, err := reg.GetTimer("b.bab.oncarlos.com", "nightly"); err != nil {
681 t.Fatalf("an unfrozen host stopped converging: %v", err)
682 }
683}
684
685// While any routed host is frozen, Pass leaves the table-wide prune alone,
686// so a frozen host's finished once-rows stay as the snapshot saw them; the
687// runner's own daily sweep is the one prune that is not frozen (joint N14).
688func TestConvergerSkipsTheTableWidePruneWhileAHostIsFrozen(t *testing.T) {
689 reg, st := convergeFixture(t)
690 ctx := context.Background()
691 now := at("2026-08-23T13:07:00Z")
692 const host = "a.bab.oncarlos.com"
693 old := now.Add(-8 * 24 * time.Hour).Unix()
694 reg.UpsertTimer(registry.Timer{Host: host, Name: "done", Kind: KindOnce, Spec: "at 1", Path: "/d", NextAt: 0, LastAt: old})
695 if _, err := reg.MarkTimerFired(host, "done", scheduleTimerRegistration(t, reg, host, "done"), old, "204", 1, 0, 0, 0); err != nil {
696 t.Fatal(err)
697 }
698 c := NewConverger(reg, st, "box1", nil)
699 c.Now = func() time.Time { return now }
700 c.Frozen = func(h string) bool { return h == host }
701 if err := c.Pass(ctx); err != nil {
702 t.Fatal(err)
703 }
704 if tm, _ := reg.GetTimer(host, "done"); tm == nil {
705 t.Fatal("the frozen host's finished once-row was pruned")
706 }
707 c.Frozen = func(string) bool { return false }
708 if err := c.Pass(ctx); err != nil {
709 t.Fatal(err)
710 }
711 if tm, _ := reg.GetTimer(host, "done"); tm != nil {
712 t.Fatalf("the prune did not resume after the freeze: %+v", tm)
713 }
714}
715
716// Fix round 1 (task-6-review, Important 3): with neither Frozen nor Gate
717// set (the Phase 1 switch off) the converger is main's. A host whose row
718// status is a hold has its declaration converged and its undeclared
719// schedule pruned, and the table-wide prune runs.
720func TestConvergerWithoutTheHooksIsMain(t *testing.T) {
721 reg, st := convergeFixture(t)
722 ctx := context.Background()
723 now := at("2026-08-23T13:07:00Z")
724 const host = "a.bab.oncarlos.com"
725 if err := reg.SetHold(host, "moving", "mv_a"); err != nil {
726 t.Fatal(err)
727 }
728 reg.UpsertTimer(registry.Timer{Host: host, Name: "stale", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/s", NextAt: now.Unix() - 5})
729 old := now.Add(-8 * 24 * time.Hour).Unix()
730 reg.UpsertTimer(registry.Timer{Host: host, Name: "done", Kind: KindOnce, Spec: "at 1", Path: "/d", NextAt: 0, LastAt: old})
731 if _, err := reg.MarkTimerFired(host, "done", scheduleTimerRegistration(t, reg, host, "done"), old, "204", 1, 0, 0, 0); err != nil {
732 t.Fatal(err)
733 }
734 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
735 c := NewConverger(reg, st, "box1", nil)
736 c.Now = func() time.Time { return now }
737 if err := c.Pass(ctx); err != nil {
738 t.Fatal(err)
739 }
740 if _, err := reg.GetTimer(host, "sync"); err != nil {
741 t.Fatalf("with the hold wiring off the held host's declaration did not converge: %v", err)
742 }
743 if tm, _ := reg.GetTimer(host, "stale"); tm != nil {
744 t.Fatalf("with the hold wiring off the undeclared schedule survived: %+v", tm)
745 }
746 if tm, _ := reg.GetTimer(host, "done"); tm != nil {
747 t.Fatalf("with the hold wiring off the finished once-row survived the table-wide prune: %+v", tm)
748 }
749}
750