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 "errors"
6 "fmt"
7 "net/http"
8 "net/http/httptest"
9 "os"
10 "path/filepath"
11 "strings"
12 "sync"
13 "testing"
14 "time"
15
16 "github.com/carlosframework/platform/internal/registry"
17 "github.com/carlosframework/platform/internal/release"
18)
19
20type fakeDeliverer struct {
21 mu sync.Mutex
22 status int
23 err error
24 calls []*http.Request
25 hosts []string
26}
27
28func (f *fakeDeliverer) Deliver(_ context.Context, rt registry.Route, req *http.Request) (int, error) {
29 f.mu.Lock()
30 defer f.mu.Unlock()
31 f.calls = append(f.calls, req)
32 f.hosts = append(f.hosts, rt.Host)
33 return f.status, f.err
34}
35
36// testClock is the Now seam for the tests that advance time WHILE a delivery
37// goroutine is running: the plain `now := …; r.Now = func() …` shape those
38// tests would otherwise use is a data race under -race, and the runner reads
39// its clock from inside every host goroutine.
40type testClock struct {
41 mu sync.Mutex
42 t time.Time
43}
44
45func (c *testClock) now() time.Time {
46 c.mu.Lock()
47 defer c.mu.Unlock()
48 return c.t
49}
50
51func (c *testClock) advance(d time.Duration) {
52 c.mu.Lock()
53 defer c.mu.Unlock()
54 c.t = c.t.Add(d)
55}
56
57// firedHosts counts deliveries per host — the counting assertion the
58// parallelism and exactly-once tests need.
59func (f *fakeDeliverer) firedHosts() map[string]int {
60 f.mu.Lock()
61 defer f.mu.Unlock()
62 n := map[string]int{}
63 for _, h := range f.hosts {
64 n[h]++
65 }
66 return n
67}
68
69func runnerFixture(t *testing.T, d Deliverer) (*registry.Registry, *Runner) {
70 t.Helper()
71 reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
72 if err != nil {
73 t.Fatal(err)
74 }
75 t.Cleanup(func() { reg.Close() })
76 if err := reg.Upsert(registry.Route{Host: "a.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true}); err != nil {
77 t.Fatal(err)
78 }
79 r := &Runner{Reg: reg, Deliver: d, Timeout: time.Minute, Box: "box1", Build: "test",
80 Token: func(string) (string, error) { return "tok", nil }}
81 return reg, r
82}
83
84func TestClassify(t *testing.T) {
85 cases := []struct {
86 status int
87 err error
88 want Outcome
89 }{
90 {204, nil, Outcome{Status: "204", Retry: false}},
91 {200, nil, Outcome{Status: "200", Retry: false}},
92 {404, nil, Outcome{Status: "404 (app refused)", Retry: false}},
93 {403, nil, Outcome{Status: "403 (app refused)", Retry: false}},
94 {302, nil, Outcome{Status: "302 (app refused)", Retry: false}},
95 {500, nil, Outcome{Status: "500", Retry: true}},
96 {503, nil, Outcome{Status: "503", Retry: true}},
97 {429, nil, Outcome{Status: "429", Retry: true}},
98 {408, nil, Outcome{Status: "408", Retry: true}},
99 {0, errors.New("instance failed to wake"), Outcome{Status: "503: instance failed to wake", Retry: true}},
100 {0, ErrTimeout, Outcome{Status: "timeout after 1m0s", Retry: false}},
101 }
102 for _, c := range cases {
103 if got := classify(c.status, c.err, time.Minute); got != c.want {
104 t.Errorf("classify(%d, %v) = %+v, want %+v", c.status, c.err, got, c.want)
105 }
106 }
107}
108
109func TestBuildRequestCarriesTheContract(t *testing.T) {
110 rt := registry.Route{Host: "a.bab.oncarlos.com"}
111 tm := registry.Timer{Name: "sync", Kind: KindSchedule, Path: "/jobs/sync?x=1", NextAt: 1787004060, DueAt: 1787004000}
112 req, err := BuildRequest(rt, tm, "tok", "abc")
113 if err != nil {
114 t.Fatal(err)
115 }
116 if req.Method != http.MethodPost || req.Host != rt.Host || req.URL.Path != "/jobs/sync" || req.URL.RawQuery != "x=1" {
117 t.Fatalf("req = %s %s %s", req.Method, req.Host, req.URL)
118 }
119 h := req.Header
120 if h.Get("Authorization") != "Bearer tok" || h.Get("X-Carlos-Schedule") != "sync" || h.Get("X-Carlos-Schedule-Kind") != "schedule" ||
121 h.Get("X-Carlos-Schedule-At") != "1787004000" || h.Get("User-Agent") != "carlos-schedule/abc" || req.ContentLength != 0 {
122 t.Fatalf("headers = %v", h)
123 }
124 // A row from before due_at existed carries 0 there: the header falls
125 // back to next_at, which at attempt 0 is the occurrence.
126 tm.DueAt = 0
127 if req, _ := BuildRequest(rt, tm, "tok", "abc"); req.Header.Get("X-Carlos-Schedule-At") != "1787004060" {
128 t.Fatalf("legacy row At = %q", req.Header.Get("X-Carlos-Schedule-At"))
129 }
130 // A path the parser refuses is an error, never a POST to "/".
131 tm.Path = "/jobs/%zz"
132 if _, err := BuildRequest(rt, tm, "tok", "abc"); err == nil {
133 t.Fatal("bad path built a request")
134 }
135}
136
137func TestTickFiresDueAdvancesScheduleAndFinishesOnce(t *testing.T) {
138 d := &fakeDeliverer{status: 204}
139 reg, r := runnerFixture(t, d)
140 now := at("2026-08-23T13:07:00Z")
141 r.Now = func() time.Time { return now }
142 reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix() - 5})
143 reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1})
144 reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "later", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/l", NextAt: now.Unix() + 60})
145 var events []string
146 r.Event = func(_ registry.Route, level, msg string) { events = append(events, level+": "+msg) }
147 var reports int
148 r.Report = func(registry.Route, Status) { reports++ }
149 r.tick()
150 if len(d.calls) != 2 {
151 t.Fatalf("delivered %d, want 2 (not the future one)", len(d.calls))
152 }
153 sync, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
154 if sync.LastStatus != "204" || sync.LastAt != now.Unix() || sync.NextAt != at("2026-08-23T18:00:00Z").Unix() || sync.Attempts != 0 {
155 t.Fatalf("sync = %+v", sync)
156 }
157 once, _ := reg.GetTimer("a.bab.oncarlos.com", "remind")
158 if once.NextAt != 0 || once.LastStatus != "204" {
159 t.Fatalf("once = %+v", once)
160 }
161 if len(events) != 2 || events[0][:5] != "info:" {
162 t.Fatalf("events = %v", events)
163 }
164 if reports != 1 { // one report per host per tick, not per timer
165 t.Fatalf("reports = %d", reports)
166 }
167}
168
169func TestTickRetriesWithBackoffThenGivesUp(t *testing.T) {
170 d := &fakeDeliverer{status: 503}
171 reg, r := runnerFixture(t, d)
172 now := at("2026-08-23T13:07:00Z")
173 r.Now = func() time.Time { return now }
174 reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix()})
175 wantBackoff := []time.Duration{time.Minute, 2 * time.Minute, 4 * time.Minute, 8 * time.Minute}
176 for i, b := range wantBackoff {
177 r.tick()
178 tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
179 if tm.Attempts != i+1 || tm.NextAt != now.Add(b).Unix() || tm.LastStatus != "503" {
180 t.Fatalf("attempt %d: %+v (want next=+%s)", i+1, tm, b)
181 }
182 now = now.Add(b)
183 }
184 r.tick() // 5th failure: give up, advance to the natural next
185 tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
186 if tm.Attempts != 0 || tm.LastStatus != "failed after 5 attempts: 503" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
187 t.Fatalf("after giving up: %+v", tm)
188 }
189}
190
191func TestBackoffNeverPassesTheNaturalNext(t *testing.T) {
192 d := &fakeDeliverer{status: 503}
193 reg, r := runnerFixture(t, d)
194 now := at("2026-08-23T13:59:30Z") // next natural 1m fire is 14:00:00, 30s away
195 r.Now = func() time.Time { return now }
196 reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "m", Kind: KindSchedule, Spec: "every 1m0s", Path: "/m", NextAt: now.Unix()})
197 r.tick()
198 tm, _ := reg.GetTimer("a.bab.oncarlos.com", "m")
199 if tm.NextAt != at("2026-08-23T14:00:00Z").Unix() {
200 t.Fatalf("next = %d, want the natural next, not now+1m", tm.NextAt)
201 }
202}
203
204func TestAppRefusalDoesNotRetry(t *testing.T) {
205 d := &fakeDeliverer{status: 404}
206 reg, r := runnerFixture(t, d)
207 now := at("2026-08-23T13:07:00Z")
208 r.Now = func() time.Time { return now }
209 reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "o", Kind: KindOnce, Spec: "at 1", Path: "/o", NextAt: 1})
210 var levels []string
211 r.Event = func(_ registry.Route, level, _ string) { levels = append(levels, level) }
212 r.tick()
213 tm, _ := reg.GetTimer("a.bab.oncarlos.com", "o")
214 if tm.NextAt != 0 || tm.LastStatus != "404 (app refused)" || tm.Attempts != 0 {
215 t.Fatalf("%+v", tm)
216 }
217 if len(levels) != 1 || levels[0] != "warn" {
218 t.Fatalf("levels = %v", levels)
219 }
220}
221
222func TestTimeoutCountsAsDelivered(t *testing.T) {
223 d := &fakeDeliverer{err: ErrTimeout}
224 reg, r := runnerFixture(t, d)
225 now := at("2026-08-23T13:07:00Z")
226 r.Now = func() time.Time { return now }
227 reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: now.Unix()})
228 r.tick()
229 tm, _ := reg.GetTimer("a.bab.oncarlos.com", "s")
230 if tm.Attempts != 0 || tm.LastStatus != "timeout after 1m0s" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
231 t.Fatalf("%+v", tm)
232 }
233}
234
235func TestNoTokenIsRecordedNotDelivered(t *testing.T) {
236 d := &fakeDeliverer{status: 204}
237 reg, r := runnerFixture(t, d)
238 r.Token = func(string) (string, error) { return "", errors.New("no CARLOS_ADMIN_TOKEN") }
239 now := at("2026-08-23T13:07:00Z")
240 r.Now = func() time.Time { return now }
241 reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: now.Unix()})
242 r.tick()
243 if len(d.calls) != 0 {
244 t.Fatal("delivered without a token")
245 }
246 tm, _ := reg.GetTimer("a.bab.oncarlos.com", "s")
247 if tm.Attempts != 1 || tm.LastStatus != "no instance token: no CARLOS_ADMIN_TOKEN" {
248 t.Fatalf("%+v", tm)
249 }
250}
251
252func TestTickSkipsRowsWhoseRouteIsGone(t *testing.T) {
253 d := &fakeDeliverer{status: 204}
254 reg, r := runnerFixture(t, d)
255 reg.UpsertTimer(registry.Timer{Host: "ghost.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: 1})
256 r.tick()
257 if len(d.calls) != 0 {
258 t.Fatal("delivered to a host with no route")
259 }
260}
261
262// X-Carlos-Schedule-At is the OCCURRENCE, the key an app dedupes on (spec
263// §6.3, §7.1): a retry re-sends the same value, a fire that advances the row
264// moves it to the new natural occurrence, and a `run` makes now the
265// occurrence. next_at is none of those things after the first retry.
266func TestScheduleAtHeaderIsTheOccurrenceNotNextAt(t *testing.T) {
267 d := &fakeDeliverer{status: 503}
268 reg, r := runnerFixture(t, d)
269 now := at("2026-08-23T13:07:00Z")
270 r.Now = func() time.Time { return now }
271 occurrence := now.Unix() - 5
272 reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: occurrence, DueAt: occurrence})
273 atHeader := func(i int) string { return d.calls[i].Header.Get("X-Carlos-Schedule-At") }
274
275 r.tick() // fails → retry in 1m
276 tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
277 if tm.NextAt != now.Add(time.Minute).Unix() || tm.DueAt != occurrence {
278 t.Fatalf("after first failure: %+v", tm)
279 }
280 now = now.Add(time.Minute)
281 r.tick() // the retry: same occurrence on the wire
282 if len(d.calls) != 2 || atHeader(0) != itoa(occurrence) || atHeader(1) != itoa(occurrence) {
283 t.Fatalf("retry header = %q, first = %q, want both %d", atHeader(1), atHeader(0), occurrence)
284 }
285
286 d.status = 204
287 now = now.Add(2 * time.Minute)
288 r.tick() // succeeds → advances
289 tm, _ = reg.GetTimer("a.bab.oncarlos.com", "sync")
290 next := at("2026-08-23T18:00:00Z").Unix()
291 if tm.NextAt != next || tm.DueAt != next || atHeader(2) != itoa(occurrence) {
292 t.Fatalf("after advance: %+v (header %q)", tm, atHeader(2))
293 }
294
295 // `run`: FireTimerNow makes now the occurrence, and the tick sends it.
296 now = now.Add(time.Minute)
297 reg.FireTimerNow("a.bab.oncarlos.com", "sync", now.Unix())
298 r.tick()
299 if len(d.calls) != 4 || atHeader(3) != itoa(now.Unix()) {
300 t.Fatalf("run header = %q, want %d", atHeader(3), now.Unix())
301 }
302}
303
304// A row whose path the URL parser refuses is recorded, not retried and not
305// delivered anywhere else: ValidPath runs the same parser at declaration, so
306// such a row was written some other way and is exactly as bad next time.
307func TestBadPathIsRecordedNotDelivered(t *testing.T) {
308 d := &fakeDeliverer{status: 204}
309 reg, r := runnerFixture(t, d)
310 now := at("2026-08-23T13:07:00Z")
311 r.Now = func() time.Time { return now }
312 reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/%zz", NextAt: now.Unix()})
313 r.tick()
314 if len(d.calls) != 0 {
315 t.Fatal("delivered a request the parser refused")
316 }
317 tm, _ := reg.GetTimer("a.bab.oncarlos.com", "s")
318 if tm.Attempts != 0 || !strings.HasPrefix(tm.LastStatus, "bad path: ") || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
319 t.Fatalf("%+v", tm)
320 }
321}
322
323// tick is Tick plus a wait for every host goroutine it launched — the shape
324// every test here asserts against, since Tick itself returns as soon as the
325// work is in flight.
326func (r *Runner) tick() {
327 r.Tick(context.Background())
328 r.wait()
329}
330
331// blockingDeliverer answers 204 at once for every host except `slow`, whose
332// delivery parks on `gate` until the test closes it — a 20-minute handler,
333// in miniature.
334type blockingDeliverer struct {
335 fakeDeliverer
336 slow string
337 gate chan struct{}
338}
339
340func (b *blockingDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) {
341 if rt.Host == b.slow {
342 <-b.gate
343 }
344 return b.fakeDeliverer.Deliver(ctx, rt, req)
345}
346
347// One long handler must not hold every other host's timers: the tick
348// launches per-host work and returns; while one host is parked, the others
349// all fire; and a later tick does NOT start a second delivery to the parked
350// host (its rows are still due — per-host serialization spans ticks).
351func TestOneSlowHostDoesNotBlockTheOthersOrRefire(t *testing.T) {
352 d := &blockingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, slow: "slow.bab.oncarlos.com", gate: make(chan struct{})}
353 reg, r := runnerFixture(t, d)
354 now := at("2026-08-23T13:07:00Z")
355 clock := &testClock{t: now}
356 r.Now = clock.now
357 hosts := []string{"slow.bab.oncarlos.com", "b.bab.oncarlos.com", "c.bab.oncarlos.com", "d.bab.oncarlos.com", "e.bab.oncarlos.com", "f.bab.oncarlos.com", "g.bab.oncarlos.com"}
358 for i, h := range hosts {
359 if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil {
360 t.Fatal(err)
361 }
362 // The slow host is due FIRST, so it takes a slot before anyone else.
363 reg.UpsertTimer(registry.Timer{Host: h, Name: "j", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 100 + int64(i)})
364 }
365 if len(hosts) <= MaxParallelHosts {
366 t.Fatalf("fixture has %d hosts; the test needs more than MaxParallelHosts (%d) so slots recycle under the parked one", len(hosts), MaxParallelHosts)
367 }
368
369 done := make(chan struct{})
370 go func() { r.Tick(context.Background()); close(done) }()
371 select {
372 case <-done:
373 case <-time.After(10 * time.Second):
374 t.Fatal("Tick did not return while one host was parked")
375 }
376 // Every other host fires AND is recorded while slow is parked — record
377 // writes the row after Deliver returns, so poll the recorded state
378 // itself rather than the delivery count (which races ahead of it).
379 deadline := time.Now().Add(10 * time.Second)
380 for {
381 recorded := 0
382 for _, h := range hosts[1:] {
383 if tm, _ := reg.GetTimer(h, "j"); tm != nil && tm.LastStatus != "" {
384 recorded++
385 }
386 }
387 if recorded == len(hosts)-1 && d.firedHosts()[d.slow] == 0 {
388 break
389 }
390 if time.Now().After(deadline) {
391 t.Fatalf("fired = %v, want every host but %s recorded", d.firedHosts(), d.slow)
392 }
393 time.Sleep(5 * time.Millisecond)
394 }
395 for _, h := range hosts[1:] {
396 if tm, _ := reg.GetTimer(h, "j"); tm.LastStatus != "204" {
397 t.Fatalf("%s not recorded while slow was parked: %+v", h, tm)
398 }
399 }
400
401 // The parked host's row is still due; a second tick must skip it.
402 clock.advance(time.Minute)
403 r.Tick(context.Background())
404 close(d.gate)
405 r.wait()
406 if n := d.firedHosts()[d.slow]; n != 1 {
407 t.Fatalf("slow host delivered %d times, want exactly 1 — a busy host was re-fired", n)
408 }
409 if tm, _ := reg.GetTimer(d.slow, "j"); tm.LastStatus != "204" || tm.NextAt != at("2026-08-23T14:00:00Z").Unix() {
410 t.Fatalf("slow row after release: %+v", tm)
411 }
412 // And once released, it is fair game again.
413 clock.advance(at("2026-08-23T14:00:01Z").Sub(clock.now()))
414 r.tick()
415 if n := d.firedHosts()[d.slow]; n != 2 {
416 t.Fatalf("released host delivered %d times, want 2", n)
417 }
418}
419
420// The prune leg is the RUNNER's, not the converger's, because a box with the
421// activator armed and no deploy store has no converger at all (spec §6.2) —
422// and it is exactly the box whose rows arrive on the control socket. Without
423// this leg its finished once-rows accumulate forever.
424func TestRunnerPrunesWithoutAConverger(t *testing.T) {
425 d := &fakeDeliverer{status: 204}
426 reg, r := runnerFixture(t, d)
427 if r.Report != nil {
428 t.Fatal("fixture has a reporter — this test is the storeless box")
429 }
430 now := at("2026-08-23T13:07:00Z")
431 r.Now = func() time.Time { return now }
432 const host = "a.bab.oncarlos.com"
433 old := now.Add(-8 * 24 * time.Hour).Unix()
434 reg.UpsertTimer(registry.Timer{Host: host, Name: "done-long-ago", Kind: KindOnce, Spec: "at 1", Path: "/o", NextAt: 0, LastAt: old})
435 reg.UpsertTimer(registry.Timer{Host: host, Name: "done-just-now", Kind: KindOnce, Spec: "at 1", Path: "/o", NextAt: 0, LastAt: now.Add(-time.Hour).Unix()})
436 reg.UpsertTimer(registry.Timer{Host: "ghost.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/s", NextAt: 0, LastAt: now.Unix()})
437 r.tick()
438 if tm, _ := reg.GetTimer(host, "done-long-ago"); tm != nil {
439 t.Fatalf("a finished once-row 8 days old survived the runner's prune: %+v", tm)
440 }
441 if tm, _ := reg.GetTimer("ghost.bab.oncarlos.com", "s"); tm != nil {
442 t.Fatalf("a row for a host with no route survived: %+v", tm)
443 }
444 if tm, _ := reg.GetTimer(host, "done-just-now"); tm == nil {
445 t.Fatal("a finished once-row inside the retention window was pruned")
446 }
447
448 // Daily, not every tick: the stamp is what keeps a 15s tick from running
449 // a DELETE 5,760 times a day.
450 reg.UpsertTimer(registry.Timer{Host: "ghost2.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/s", NextAt: 0, LastAt: now.Unix()})
451 now = now.Add(TickEvery)
452 r.tick()
453 if tm, _ := reg.GetTimer("ghost2.bab.oncarlos.com", "s"); tm == nil {
454 t.Fatal("the prune leg ran again 15s later — the stamp is not holding")
455 }
456 now = now.Add(pruneEvery)
457 r.tick()
458 if tm, _ := reg.GetTimer("ghost2.bab.oncarlos.com", "s"); tm != nil {
459 t.Fatalf("the prune leg never came round again: %+v", tm)
460 }
461}
462
463// retainingDeliverer is a Deliverer that also keeps per-socket state — the
464// agent's does (one connection pool per socket).
465type retainingDeliverer struct {
466 fakeDeliverer
467 mu sync.Mutex
468 calls []map[string]bool
469}
470
471func (d *retainingDeliverer) RetainSockets(live map[string]bool) {
472 d.mu.Lock()
473 defer d.mu.Unlock()
474 d.calls = append(d.calls, live)
475}
476
477func (d *retainingDeliverer) sweeps() []map[string]bool {
478 d.mu.Lock()
479 defer d.mu.Unlock()
480 return append([]map[string]bool(nil), d.calls...)
481}
482
483// The prune leg doubles as the delivery leg's eviction point: it has already
484// read every route, so it hands the live SOCKET set to a Deliverer that keeps
485// per-socket state. A retired route can be reclaimed no other way — nothing
486// delivers to it, so nothing it holds can error its own way out.
487func TestPruneSweepHandsTheLiveSocketsToTheDeliverer(t *testing.T) {
488 d := &retainingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}}
489 reg, r := runnerFixture(t, d)
490 now := at("2026-08-23T13:07:00Z")
491 r.Now = func() time.Time { return now }
492 if err := reg.Upsert(registry.Route{Host: "a.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Socket: "/run/carlos/a.sock"}); err != nil {
493 t.Fatal(err)
494 }
495 if err := reg.Upsert(registry.Route{Host: "site.bab.oncarlos.com", Kind: "static", Account: "bab", App: "site", Status: "ready"}); err != nil {
496 t.Fatal(err)
497 }
498 r.tick()
499 sweeps := d.sweeps()
500 if len(sweeps) != 1 {
501 t.Fatalf("%d sweeps, want the one the prune leg runs", len(sweeps))
502 }
503 if !sweeps[0]["/run/carlos/a.sock"] || len(sweeps[0]) != 1 {
504 t.Fatalf("live sockets = %v, want exactly the routed instance's (a socketless route contributes nothing)", sweeps[0])
505 }
506 // Daily, with the prune itself: not once per tick.
507 now = now.Add(TickEvery)
508 r.tick()
509 if got := len(d.sweeps()); got != 1 {
510 t.Fatalf("%d sweeps after a second tick, want still 1", got)
511 }
512}
513
514// A busy host must not crowd another host's due timer out of the batch. The
515// runner skips a host whose delivery is still running, so its rows are dead
516// weight in a capped query: the exclusion belongs in DueTimers, before the
517// LIMIT. Sized to the real batch so the test fails if dueBatch and the
518// fixture ever drift apart.
519func TestBusyHostDoesNotStarveAnotherHostsTimer(t *testing.T) {
520 d := &blockingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, slow: "slow.bab.oncarlos.com", gate: make(chan struct{})}
521 reg, r := runnerFixture(t, d)
522 clk := &testClock{t: at("2026-08-23T13:07:00Z")}
523 now := clk.now()
524 r.Now = clk.now
525 for _, h := range []string{d.slow, "quiet.bab.oncarlos.com"} {
526 if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil {
527 t.Fatal(err)
528 }
529 }
530 // A full batch of rows on the parked host, every one of them due BEFORE
531 // the quiet host's single row.
532 for i := 0; i < dueBatch; i++ {
533 reg.UpsertTimer(registry.Timer{Host: d.slow, Name: fmt.Sprintf("j%02d", i), Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 1000 + int64(i)})
534 }
535 reg.UpsertTimer(registry.Timer{Host: "quiet.bab.oncarlos.com", Name: "j", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 1})
536
537 // First tick: the parked host claims a slot and stops there. Tick claims
538 // synchronously, so the host is busy the moment it returns.
539 r.Tick(context.Background())
540 if busy := r.busyHosts(); len(busy) != 1 || busy[0] != d.slow {
541 t.Fatalf("after the first tick busy = %v, want just the parked host", busy)
542 }
543
544 // Second tick, while it is still parked: the quiet host's row is due and
545 // must be delivered and recorded.
546 clk.advance(TickEvery)
547 r.Tick(context.Background())
548 deadline := time.Now().Add(10 * time.Second)
549 for time.Now().Before(deadline) {
550 if tm, _ := reg.GetTimer("quiet.bab.oncarlos.com", "j"); tm != nil && tm.LastStatus == "204" {
551 break
552 }
553 time.Sleep(time.Millisecond)
554 }
555 if tm, _ := reg.GetTimer("quiet.bab.oncarlos.com", "j"); tm == nil || tm.LastStatus != "204" {
556 t.Fatalf("the quiet host's timer never fired behind a full batch of a busy host's rows: %+v", tm)
557 }
558 close(d.gate)
559 r.wait()
560}
561
562// countingDeliverer parks every delivery on a gate the test opens, and
563// signals two things about how many ran at once: reached (exactly
564// MaxParallelHosts in flight) and over (MORE than that — the regression).
565type countingDeliverer struct {
566 fakeDeliverer
567 gate chan struct{}
568 mu2 sync.Mutex
569 cur int
570 max int
571 reached chan struct{}
572 over chan struct{}
573 onceFull sync.Once
574 onceOver sync.Once
575}
576
577func (c *countingDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) {
578 c.mu2.Lock()
579 c.cur++
580 if c.cur > c.max {
581 c.max = c.cur
582 }
583 full, over := c.cur == MaxParallelHosts, c.cur > MaxParallelHosts
584 c.mu2.Unlock()
585 if full {
586 c.onceFull.Do(func() { close(c.reached) })
587 }
588 if over {
589 c.onceOver.Do(func() { close(c.over) })
590 }
591 <-c.gate
592 c.mu2.Lock()
593 c.cur--
594 c.mu2.Unlock()
595 return c.fakeDeliverer.Deliver(ctx, rt, req)
596}
597
598func (c *countingDeliverer) peak() int {
599 c.mu2.Lock()
600 defer c.mu2.Unlock()
601 return c.max
602}
603
604// boundWindow is how long the bound is watched with every delivery parked.
605// It is a floor on how long we LOOK, never a deadline anything must meet:
606// both assertions inside it are negative (a fifth delivery must not start,
607// Tick must not return), so a slow machine can only make the test weaker,
608// never redder. A widened bound trips both within microseconds.
609const boundWindow = 500 * time.Millisecond
610
611// The ≤4-parallel path: with more due hosts than MaxParallelHosts, exactly
612// four deliveries run at once — no fewer (they are genuinely parallel, not
613// one host at a time) and no more (the semaphore is the bound) — and every
614// host still completes once the slots recycle.
615func TestDeliveriesAreParallelToTheBoundAndAllComplete(t *testing.T) {
616 d := &countingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, gate: make(chan struct{}), reached: make(chan struct{}), over: make(chan struct{})}
617 reg, r := runnerFixture(t, d)
618 now := at("2026-08-23T13:07:00Z")
619 r.Now = func() time.Time { return now }
620 const hosts = 2 * MaxParallelHosts
621 var names []string
622 for i := 0; i < hosts; i++ {
623 h := fmt.Sprintf("h%d.bab.oncarlos.com", i)
624 names = append(names, h)
625 if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil {
626 t.Fatal(err)
627 }
628 reg.UpsertTimer(registry.Timer{Host: h, Name: "j", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 100 + int64(i)})
629 }
630
631 tickDone := make(chan struct{})
632 go func() { r.Tick(context.Background()); close(tickDone) }()
633 select {
634 case <-d.reached:
635 case <-time.After(10 * time.Second):
636 t.Fatalf("only %d deliveries ever ran at once, want %d in parallel", d.peak(), MaxParallelHosts)
637 }
638 // The BOUND, watched while the gate is still shut and all four slots are
639 // held. Two independent ways a widened semaphore shows itself, and the
640 // test must be red on either:
641 //
642 // - a fifth delivery starts (`over`);
643 // - Tick RETURNS. With 8 hosts and 4 slots it cannot: it takes the slot
644 // before launching each host's goroutine, so it is parked on the
645 // semaphore until a delivery finishes — and none can while the gate
646 // is shut.
647 //
648 // Watching the peak after the fact cannot do this job: opening the gate
649 // drains `cur` faster than a fifth delivery would ever be observed, which
650 // is why the earlier shape passed 25/25 with the semaphore quadrupled.
651 select {
652 case <-d.over:
653 t.Fatalf("%d deliveries in flight at once, want at most MaxParallelHosts (%d)", d.peak(), MaxParallelHosts)
654 case <-tickDone:
655 t.Fatalf("Tick returned with every delivery still parked — it launched more than MaxParallelHosts (%d) hosts (peak %d)", MaxParallelHosts, d.peak())
656 case <-time.After(boundWindow):
657 }
658
659 close(d.gate) // every parked delivery finishes; the waiting hosts take the slots
660 select {
661 case <-tickDone:
662 case <-time.After(10 * time.Second):
663 t.Fatal("Tick never returned after the gate opened")
664 }
665 r.wait()
666 if got := d.peak(); got > MaxParallelHosts { // belt-and-braces; the window above is the real assertion
667 t.Fatalf("%d deliveries in flight at once, want at most MaxParallelHosts (%d)", got, MaxParallelHosts)
668 }
669 for _, h := range names {
670 tm, _ := reg.GetTimer(h, "j")
671 if tm == nil || tm.LastStatus != "204" {
672 t.Fatalf("%s did not complete: %+v", h, tm)
673 }
674 }
675 if fired := d.firedHosts(); len(fired) != hosts {
676 t.Fatalf("delivered to %d hosts, want %d", len(fired), hosts)
677 }
678}
679
680type timerMutationDeliverer struct {
681 mutate func()
682 status int
683}
684
685func (d timerMutationDeliverer) Deliver(context.Context, registry.Route, *http.Request) (int, error) {
686 d.mutate()
687 return d.status, nil
688}
689
690func TestDeliveryCompletionPreservesReplacementRegistration(t *testing.T) {
691 for _, mutation := range []string{"identical", "timing", "path", "cancel", "recreate", "manual"} {
692 for _, status := range []int{204, 503} {
693 t.Run(fmt.Sprintf("%s/%d", mutation, status), func(t *testing.T) {
694 reg, runner := runnerFixture(t, nil)
695 now := at("2026-08-23T13:07:00Z")
696 runner.Now = func() time.Time { return now }
697 const host = "a.bab.oncarlos.com"
698 mux := http.NewServeMux()
699 Mount(mux, host, reg, func() (string, error) { return "tok", nil }, runner.Now)
700 put := func(when time.Time, path string) {
701 req := httptest.NewRequest(http.MethodPut, "/schedule/remind", strings.NewReader(fmt.Sprintf(`{"at":%q,"path":%q}`, when.Format(time.RFC3339), path)))
702 req.Header.Set("Authorization", "Bearer tok")
703 w := httptest.NewRecorder()
704 mux.ServeHTTP(w, req)
705 if w.Code != http.StatusOK {
706 t.Errorf("PUT: %d %s", w.Code, w.Body.String())
707 }
708 }
709 put(now, "/jobs/remind")
710 before, err := reg.GetTimer(host, "remind")
711 if err != nil {
712 t.Fatal(err)
713 }
714 var replacement *registry.Timer
715 var events []string
716 runner.Event = func(_ registry.Route, _, msg string) { events = append(events, msg) }
717 runner.Deliver = timerMutationDeliverer{status: status, mutate: func() {
718 switch mutation {
719 case "identical":
720 put(now, "/jobs/remind")
721 case "timing":
722 put(now.Add(time.Hour), "/jobs/remind")
723 case "path":
724 put(now, "/jobs/changed")
725 case "cancel", "recreate":
726 req := httptest.NewRequest(http.MethodDelete, "/schedule/remind", nil)
727 req.Header.Set("Authorization", "Bearer tok")
728 w := httptest.NewRecorder()
729 mux.ServeHTTP(w, req)
730 if w.Code != http.StatusNoContent {
731 t.Errorf("DELETE: %d", w.Code)
732 }
733 if mutation == "recreate" {
734 put(now, "/jobs/remind")
735 }
736 case "manual":
737 if err := reg.FireTimerNow(host, "remind", now.Unix()); err != nil {
738 t.Error(err)
739 }
740 }
741 replacement, err = reg.GetTimer(host, "remind")
742 if mutation != "cancel" && err != nil {
743 t.Error(err)
744 }
745 }}
746 runner.tick()
747 if len(events) != 1 || strings.Contains(events[0], "retrying at") {
748 t.Fatalf("superseded delivery event: %v", events)
749 }
750 after, err := reg.GetTimer(host, "remind")
751 if mutation == "cancel" {
752 if !errors.Is(err, os.ErrNotExist) {
753 t.Fatalf("cancelled timer: %+v, %v", after, err)
754 }
755 return
756 }
757 if err != nil {
758 t.Fatal(err)
759 }
760 if replacement == nil {
761 t.Fatal("replacement missing")
762 }
763 if after.Registration == before.Registration || after.Registration == "" {
764 t.Fatalf("registration did not change: %+v", after)
765 }
766 if *after != *replacement {
767 t.Fatalf("completion changed replacement: got %+v, want %+v", after, replacement)
768 }
769 runner.Deliver = &fakeDeliverer{status: 204}
770 now = time.Unix(after.NextAt, 0)
771 runner.tick()
772 finished, err := reg.GetTimer(host, "remind")
773 if err != nil {
774 t.Fatal(err)
775 }
776 if finished.NextAt != 0 || finished.LastStatus != "204" {
777 t.Fatalf("replacement did not fire: %+v", finished)
778 }
779 })
780 }
781 }
782}
783
784func TestDeliveryCompletionSurvivesUnchangedDeclaration(t *testing.T) {
785 for _, status := range []int{204, 503} {
786 t.Run(fmt.Sprint(status), func(t *testing.T) {
787 reg, runner := runnerFixture(t, nil)
788 now := at("2026-08-23T13:07:00Z")
789 runner.Now = func() time.Time { return now }
790 tm := registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix(), DueAt: now.Unix()}
791 if err := reg.UpsertTimer(tm); err != nil {
792 t.Fatal(err)
793 }
794 before, err := reg.GetTimer(tm.Host, tm.Name)
795 if err != nil {
796 t.Fatal(err)
797 }
798 tm.NextAt, tm.DueAt = 0, 0
799 runner.Deliver = timerMutationDeliverer{status: status, mutate: func() {
800 if err := reg.UpsertTimer(tm); err != nil {
801 t.Error(err)
802 }
803 }}
804 runner.tick()
805 after, err := reg.GetTimer(tm.Host, tm.Name)
806 if err != nil {
807 t.Fatal(err)
808 }
809 if after.Registration != before.Registration || after.LastStatus != fmt.Sprint(status) {
810 t.Fatalf("completion missing: %+v", after)
811 }
812 if status == 503 && (after.Attempts != 1 || after.DueAt != before.DueAt || after.NextAt != now.Add(time.Minute).Unix()) {
813 t.Fatalf("retry changed: %+v", after)
814 }
815 if err := reg.UpsertTimer(tm); err != nil {
816 t.Fatal(err)
817 }
818 converged, err := reg.GetTimer(tm.Host, tm.Name)
819 if err != nil {
820 t.Fatal(err)
821 }
822 if *converged != *after {
823 t.Fatalf("converge changed completed state: got %+v, want %+v", converged, after)
824 }
825 })
826 }
827}
828
829func TestInvalidTimerSnapshotCannotDeleteReplacement(t *testing.T) {
830 reg, runner := runnerFixture(t, &fakeDeliverer{status: 204})
831 tm := registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "invalid", Path: "/jobs/remind", NextAt: 100}
832 if err := reg.UpsertTimer(tm); err != nil {
833 t.Fatal(err)
834 }
835 old, err := reg.GetTimer(tm.Host, tm.Name)
836 if err != nil {
837 t.Fatal(err)
838 }
839 tm.Spec = "at 100"
840 if err := reg.UpsertTimer(tm); err != nil {
841 t.Fatal(err)
842 }
843 replacement, err := reg.GetTimer(tm.Host, tm.Name)
844 if err != nil {
845 t.Fatal(err)
846 }
847 runner.fire(context.Background(), registry.Route{Host: tm.Host}, *old)
848 after, err := reg.GetTimer(tm.Host, tm.Name)
849 if err != nil {
850 t.Fatal(err)
851 }
852 if *after != *replacement {
853 t.Fatalf("invalid snapshot changed replacement: got %+v, want %+v", after, replacement)
854 }
855}
856
857func TestBatchSkipsTimerChangedByEarlierDelivery(t *testing.T) {
858 for _, mutation := range []string{"cancel", "rearm"} {
859 t.Run(mutation, func(t *testing.T) {
860 reg, runner := runnerFixture(t, nil)
861 now := at("2026-08-23T13:07:00Z")
862 runner.Now = func() time.Time { return now }
863 const host = "a.bab.oncarlos.com"
864 for _, name := range []string{"first", "later"} {
865 if err := reg.UpsertTimer(registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: "at 100", Path: "/jobs/" + name, NextAt: 100}); err != nil {
866 t.Fatal(err)
867 }
868 }
869 calls := 0
870 runner.Deliver = timerMutationDeliverer{status: 204, mutate: func() {
871 calls++
872 if mutation == "cancel" {
873 if err := reg.DeleteTimer(host, "later"); err != nil {
874 t.Error(err)
875 }
876 } else {
877 if err := reg.FireTimerNow(host, "later", now.Add(time.Hour).Unix()); err != nil {
878 t.Error(err)
879 }
880 }
881 }}
882 runner.tick()
883 if calls != 1 {
884 t.Fatalf("delivered %d timers, want first only", calls)
885 }
886 })
887 }
888}
889
890func TestConcurrentTicksSkipSnapshotChangedDuringTokenLookup(t *testing.T) {
891 for _, mutation := range []string{"cancel", "rearm", "retry", "retry-still-due", "done"} {
892 t.Run(mutation, func(t *testing.T) {
893 deliver := &fakeDeliverer{status: 204}
894 reg, runner := runnerFixture(t, deliver)
895 now := at("2026-08-23T13:07:00Z")
896 runner.Now = func() time.Time { return now }
897 tm := registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "at 100", Path: "/jobs/remind", NextAt: 100}
898 if err := reg.UpsertTimer(tm); err != nil {
899 t.Fatal(err)
900 }
901 before, err := reg.GetTimer(tm.Host, tm.Name)
902 if err != nil {
903 t.Fatal(err)
904 }
905 entered, release := make(chan struct{}), make(chan struct{})
906 var tokenOnce sync.Once
907 runner.Token = func(string) (string, error) { tokenOnce.Do(func() { close(entered) }); <-release; return "tok", nil }
908 runner.Tick(context.Background())
909 <-entered
910 switch mutation {
911 case "cancel":
912 err = reg.DeleteTimer(tm.Host, tm.Name)
913 case "rearm":
914 err = reg.FireTimerNow(tm.Host, tm.Name, now.Add(time.Hour).Unix())
915 case "retry":
916 _, err = reg.MarkTimerFired(tm.Host, tm.Name, before.Registration, now.Unix(), "503", 0, 1, now.Add(time.Minute).Unix(), 100)
917 case "retry-still-due":
918 _, err = reg.MarkTimerFired(tm.Host, tm.Name, before.Registration, now.Unix(), "503", 0, 1, now.Unix()-1, 100)
919 case "done":
920 _, err = reg.MarkTimerFired(tm.Host, tm.Name, before.Registration, now.Unix(), "204", 0, 0, 0, 0)
921 }
922 if err != nil {
923 close(release)
924 runner.wait()
925 t.Fatal(err)
926 }
927 runner.Tick(context.Background())
928 close(release)
929 runner.wait()
930 runner.fire(context.Background(), registry.Route{Host: tm.Host}, *before)
931 if len(deliver.calls) != 0 {
932 t.Fatalf("delivered stale snapshot %d times", len(deliver.calls))
933 }
934 })
935 }
936}
937
938func TestNonpositiveOccurrenceDeliveredAndRetriedWithOriginalTimestamp(t *testing.T) {
939 for _, requested := range []int64{0, -1, -62135596800} {
940 t.Run(fmt.Sprint(requested), func(t *testing.T) {
941 deliver := &fakeDeliverer{status: 503}
942 reg, runner := runnerFixture(t, deliver)
943 now := at("2026-08-23T13:07:00Z")
944 runner.Now = func() time.Time { return now }
945 mux := http.NewServeMux()
946 Mount(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil }, runner.Now)
947 rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"`+time.Unix(requested, 0).UTC().Format(time.RFC3339)+`","path":"/jobs/remind"}`)
948 if rec.Code != http.StatusOK {
949 t.Fatalf("PUT: %d %s", rec.Code, rec.Body.String())
950 }
951 runner.tick()
952 row, err := reg.GetTimer("a.bab.oncarlos.com", "remind")
953 if err != nil {
954 t.Fatal(err)
955 }
956 if row.Attempts != 1 || row.NextAt != now.Add(time.Minute).Unix() || row.DueAt != requested {
957 t.Fatalf("retry: %+v", row)
958 }
959 now = now.Add(time.Minute)
960 deliver.status = 204
961 runner.tick()
962 if len(deliver.calls) != 2 {
963 t.Fatalf("deliveries: %d", len(deliver.calls))
964 }
965 for _, req := range deliver.calls {
966 if got := req.Header.Get("X-Carlos-Schedule-At"); got != fmt.Sprint(requested) {
967 t.Fatalf("occurrence: %q, want %d", got, requested)
968 }
969 }
970 row, err = reg.GetTimer("a.bab.oncarlos.com", "remind")
971 if err != nil {
972 t.Fatal(err)
973 }
974 if row.NextAt != 0 || row.LastStatus != "204" {
975 t.Fatalf("completed: %+v", row)
976 }
977 })
978 }
979}
980
981// ---- move holds (moves spec §9.4) ----
982
983// holdHost is the runner fixture's host, held by a move in these tests.
984const holdHost = "a.bab.oncarlos.com"
985
986// timersOf is the host's whole timers table, for before/after comparisons.
987func timersOf(t *testing.T, reg *registry.Registry, host string) []registry.Timer {
988 t.Helper()
989 rows, err := reg.TimersFor(host)
990 if err != nil {
991 t.Fatal(err)
992 }
993 return rows
994}
995
996func sameTimers(a, b []registry.Timer) bool {
997 if len(a) != len(b) {
998 return false
999 }
1000 for i := range a {
1001 if a[i] != b[i] {
1002 return false
1003 }
1004 }
1005 return true
1006}
1007
1008// Witness wTimerFreeze: a held row's due timers are not delivered and their
1009// bookkeeping does not move (before this, a refused delivery advanced
1010// attempts); a frozen host's declared schedule is neither upserted nor
1011// pruned by Converger.Pass; once the hold lifts, both resume.
1012func TestHeldRowsFreezeTimers(t *testing.T) {
1013 d := &fakeDeliverer{status: 503} // a delivery would retry and advance attempts
1014 reg, r := runnerFixture(t, d)
1015 now := at("2026-08-23T13:07:00Z")
1016 r.Now = func() time.Time { return now }
1017 r.Gate = &HostGate{}
1018 reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1, DueAt: 1})
1019 reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "stale", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/s", NextAt: now.Unix() - 5})
1020 if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
1021 t.Fatal(err)
1022 }
1023 before := timersOf(t, reg, holdHost)
1024 r.tick()
1025 if len(d.calls) != 0 {
1026 t.Fatalf("a held row's timers were delivered %d time(s)", len(d.calls))
1027 }
1028 if got := timersOf(t, reg, holdHost); !sameTimers(got, before) {
1029 t.Fatalf("a held row's bookkeeping moved:\n got %+v\nwant %+v", got, before)
1030 }
1031 if r.Busy(holdHost) {
1032 t.Fatal("the runner kept a claim on a held host")
1033 }
1034
1035 // The converger, with the host frozen: the declaration is not upserted
1036 // and the undeclared "stale" schedule is not pruned.
1037 st := release.NewDirStore(t.TempDir())
1038 ctx := context.Background()
1039 UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
1040 frozen := true
1041 c := NewConverger(reg, st, "box1", nil)
1042 c.Now = func() time.Time { return now }
1043 c.Gate = r.Gate
1044 c.Frozen = func(host string) bool { return host == holdHost && frozen }
1045 if err := c.Pass(ctx); err != nil {
1046 t.Fatal(err)
1047 }
1048 if got := timersOf(t, reg, holdHost); !sameTimers(got, before) {
1049 t.Fatalf("Converger.Pass wrote a frozen host's timers:\n got %+v\nwant %+v", got, before)
1050 }
1051
1052 // The hold lifts: the runner delivers and the converger converges.
1053 if ok, err := reg.ReleaseHold(holdHost, "mv_a"); err != nil || !ok {
1054 t.Fatalf("release hold = %v %v", ok, err)
1055 }
1056 frozen = false
1057 r.tick()
1058 if len(d.calls) == 0 {
1059 t.Fatal("nothing was delivered after the hold lifted")
1060 }
1061 if err := c.Pass(ctx); err != nil {
1062 t.Fatal(err)
1063 }
1064 if _, err := reg.GetTimer(holdHost, "sync"); err != nil {
1065 t.Fatalf("the declaration did not converge after the freeze: %v", err)
1066 }
1067 if tm, _ := reg.GetTimer(holdHost, "stale"); tm != nil {
1068 t.Fatalf("the undeclared schedule survived the converge after the freeze: %+v", tm)
1069 }
1070}
1071
1072// A hold written while a host's goroutine is delivering stops the rest of
1073// that host's due timers: the next timer re-reads the row first.
1074func TestHoldMidDeliveryStopsTheRest(t *testing.T) {
1075 reg, r := runnerFixture(t, nil)
1076 now := at("2026-08-23T13:07:00Z")
1077 r.Now = func() time.Time { return now }
1078 r.Gate = &HostGate{} // the hold wiring armed; off, both would be delivered (TestRunnerWithoutTheGateIsMain)
1079 var delivered []string
1080 r.Deliver = deliverFunc(func(rt registry.Route, req *http.Request) (int, error) {
1081 delivered = append(delivered, req.Header.Get("X-Carlos-Schedule"))
1082 if err := reg.SetHold(rt.Host, "moving", "mv_a"); err != nil {
1083 t.Error(err)
1084 }
1085 return 204, nil
1086 })
1087 reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "first", Kind: KindOnce, Spec: "at 1", Path: "/1", NextAt: 1})
1088 reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "second", Kind: KindOnce, Spec: "at 2", Path: "/2", NextAt: 2})
1089 r.tick()
1090 if len(delivered) != 1 || delivered[0] != "first" {
1091 t.Fatalf("delivered %v, want only the first", delivered)
1092 }
1093 second, _ := reg.GetTimer(holdHost, "second")
1094 if second.NextAt != 2 || second.Attempts != 0 || second.LastAt != 0 {
1095 t.Fatalf("the second timer's bookkeeping moved under the hold: %+v", second)
1096 }
1097}
1098
1099type deliverFunc func(rt registry.Route, req *http.Request) (int, error)
1100
1101func (f deliverFunc) Deliver(_ context.Context, rt registry.Route, req *http.Request) (int, error) {
1102 return f(rt, req)
1103}
1104
1105// The cut's drain race: a tick reads the route while it is ready and pauses
1106// before its claim; the cut sets the hold and finds the host not busy, and
1107// its drain returns; the tick resumes, claims, re-reads the row inside the
1108// claim, sees the hold and lets go. Nothing is delivered and attempts does
1109// not move.
1110func TestDrainRaceCannotAdvanceBookkeeping(t *testing.T) {
1111 d := &fakeDeliverer{status: 503}
1112 reg, r := runnerFixture(t, d)
1113 now := at("2026-08-23T13:07:00Z")
1114 r.Now = func() time.Time { return now }
1115 r.Gate = &HostGate{}
1116 reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1})
1117 before := timersOf(t, reg, holdHost)
1118 reached, resume := make(chan struct{}), make(chan struct{})
1119 r.beforeClaim = func(host string) {
1120 if host == holdHost {
1121 close(reached)
1122 <-resume
1123 }
1124 }
1125 ticked := make(chan struct{})
1126 go func() { r.Tick(context.Background()); close(ticked) }()
1127 <-reached
1128 // The cut: hold, then drain.
1129 if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
1130 t.Fatal(err)
1131 }
1132 if r.Busy(holdHost) {
1133 t.Fatal("the paused tick already held the claim")
1134 }
1135 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
1136 defer cancel()
1137 if err := r.Gate.Drain(ctx, holdHost); err != nil {
1138 t.Fatalf("drain = %v", err)
1139 }
1140 close(resume)
1141 <-ticked
1142 r.wait()
1143 if len(d.calls) != 0 {
1144 t.Fatalf("the paused tick delivered %d time(s) after the drain", len(d.calls))
1145 }
1146 if got := timersOf(t, reg, holdHost); !sameTimers(got, before) {
1147 t.Fatalf("the paused tick moved bookkeeping after the drain:\n got %+v\nwant %+v", got, before)
1148 }
1149 if r.Busy(holdHost) {
1150 t.Fatal("the tick kept its claim after seeing the hold")
1151 }
1152}
1153
1154// A delivery in flight when the hold is written holds up the drain until it
1155// has recorded its outcome: the snapshot after the drain sees the
1156// delivery's bookkeeping, never a write that lands after it.
1157func TestDrainWaitsForTheDeliveryInFlight(t *testing.T) {
1158 d := &blockingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, slow: holdHost, gate: make(chan struct{})}
1159 reg, r := runnerFixture(t, d)
1160 now := at("2026-08-23T13:07:00Z")
1161 r.Now = func() time.Time { return now }
1162 r.Gate = &HostGate{}
1163 reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1})
1164 r.Tick(context.Background())
1165 if !r.Busy(holdHost) {
1166 t.Fatal("the delivery is not in flight")
1167 }
1168 if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
1169 t.Fatal(err)
1170 }
1171 drained := make(chan error, 1)
1172 go func() { drained <- r.Gate.Drain(context.Background(), holdHost) }()
1173 select {
1174 case err := <-drained:
1175 t.Fatalf("the drain returned (%v) with a delivery in flight", err)
1176 case <-time.After(50 * time.Millisecond):
1177 }
1178 close(d.gate)
1179 if err := <-drained; err != nil {
1180 t.Fatal(err)
1181 }
1182 snap := timersOf(t, reg, holdHost)
1183 r.wait()
1184 if snap[0].LastStatus != "204" || snap[0].NextAt != 0 {
1185 t.Fatalf("the drain returned before the delivery recorded: %+v", snap[0])
1186 }
1187 if got := timersOf(t, reg, holdHost); !sameTimers(got, snap) {
1188 t.Fatalf("the table changed after the drain:\n got %+v\nwant %+v", got, snap)
1189 }
1190}
1191
1192// Fix round 1 (task-6-review, Important 1): a held host's rows must not
1193// consume the due batch. DueTimers applies its LIMIT before the runner's
1194// hold filter, so a full batch of held rows due before an unrelated host's
1195// row would be selected on every tick and that row never seen. With the
1196// hold wiring armed the query itself leaves held hosts out, and the
1197// unrelated timer runs this tick.
1198func TestHeldHostsDoNotStarveAnotherHostsTimer(t *testing.T) {
1199 d := &fakeDeliverer{status: 204}
1200 reg, r := runnerFixture(t, d)
1201 now := at("2026-08-23T13:07:00Z")
1202 r.Now = func() time.Time { return now }
1203 r.Gate = &HostGate{}
1204 const quiet = "quiet.bab.oncarlos.com"
1205 if err := reg.Upsert(registry.Route{Host: quiet, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil {
1206 t.Fatal(err)
1207 }
1208 // A full batch of rows on the held host, every one due BEFORE the quiet
1209 // host's single row.
1210 for i := 0; i < dueBatch; i++ {
1211 reg.UpsertTimer(registry.Timer{Host: holdHost, Name: fmt.Sprintf("j%02d", i), Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 1000 + int64(i)})
1212 }
1213 reg.UpsertTimer(registry.Timer{Host: quiet, Name: "j", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 1})
1214 if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
1215 t.Fatal(err)
1216 }
1217 before := timersOf(t, reg, holdHost)
1218 r.tick()
1219 if tm, _ := reg.GetTimer(quiet, "j"); tm == nil || tm.LastStatus != "204" {
1220 t.Fatalf("the quiet host's timer did not fire this tick behind a full batch of a held host's rows: %+v", tm)
1221 }
1222 for _, h := range d.hosts {
1223 if h != quiet {
1224 t.Fatalf("delivered to %s, a held host", h)
1225 }
1226 }
1227 if got := timersOf(t, reg, holdHost); !sameTimers(got, before) {
1228 t.Fatalf("the held host's bookkeeping moved:\n got %+v\nwant %+v", got, before)
1229 }
1230}
1231
1232// Fix round 1 (task-6-review, Important 3): with no gate (the Phase 1
1233// switch off, or no hold wiring on this box) the runner is main's code
1234// path. A row whose status is a hold is delivered like any other and its
1235// bookkeeping moves: the batch includes it, the pre-filter and the decision
1236// read do not look at the hold, and the between-timer check does not run.
1237func TestRunnerWithoutTheGateIsMain(t *testing.T) {
1238 d := &fakeDeliverer{status: 503}
1239 reg, r := runnerFixture(t, d)
1240 now := at("2026-08-23T13:07:00Z")
1241 r.Now = func() time.Time { return now }
1242 reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "first", Kind: KindOnce, Spec: "at 1", Path: "/1", NextAt: 1, DueAt: 1})
1243 reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "second", Kind: KindOnce, Spec: "at 2", Path: "/2", NextAt: 2, DueAt: 2})
1244 if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
1245 t.Fatal(err)
1246 }
1247 r.tick()
1248 if len(d.calls) != 2 {
1249 t.Fatalf("with the hold wiring off a held row's due timers were delivered %d time(s), want both", len(d.calls))
1250 }
1251 for _, name := range []string{"first", "second"} {
1252 tm, err := reg.GetTimer(holdHost, name)
1253 if err != nil {
1254 t.Fatal(err)
1255 }
1256 if tm.Attempts != 1 || tm.LastStatus != "503" || tm.NextAt != now.Add(backoffBase).Unix() {
1257 t.Fatalf("%s: bookkeeping did not advance as main's does: %+v", name, tm)
1258 }
1259 }
1260}
1261
1262// ---- the import gate (seeded-instances spec §5.4) ----
1263
1264const gateSeedID = "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa"
1265
1266// gateForImport swaps host's row for a gated import row in one journal
1267// transaction, the way the swap does.
1268func gateForImport(t *testing.T, reg *registry.Registry, host string) {
1269 t.Helper()
1270 stored, err := reg.Get(host)
1271 if err != nil || stored == nil {
1272 t.Fatalf("get %s: %v %v", host, stored, err)
1273 }
1274 next := *stored
1275 next.Provisioned = true
1276 next.ImportGate = strings.Repeat("a", 64)
1277 next.SeedSource = `{"id":"` + gateSeedID + `","kind":"import","state":"swapped"}`
1278 if err := reg.SwapForImport(registry.ImportJournal{Host: host, SeedID: gateSeedID, Original: *stored, At: 1,
1279 RecordKey: "apps/bab/app/instances/" + host + ".json"}, next); err != nil {
1280 t.Fatal(err)
1281 }
1282}
1283
1284// A gated row gets no ticks, whether or not the hold wiring is armed, and
1285// its bookkeeping does not move; commit opens it.
1286func TestTickSkipsGatedRows(t *testing.T) {
1287 for _, armed := range []bool{false, true} {
1288 t.Run(fmt.Sprintf("holds armed %v", armed), func(t *testing.T) {
1289 reg, r := runnerFixture(t, nil)
1290 now := at("2026-08-23T13:07:00Z")
1291 r.Now = func() time.Time { return now }
1292 if armed {
1293 r.Gate = &HostGate{}
1294 }
1295 delivered := 0
1296 r.Deliver = deliverFunc(func(registry.Route, *http.Request) (int, error) { delivered++; return 204, nil })
1297 gateForImport(t, reg, holdHost)
1298 reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1})
1299 before := timersOf(t, reg, holdHost)
1300 r.tick()
1301 if delivered != 0 {
1302 t.Fatalf("delivered %d tick(s) to a gated row", delivered)
1303 }
1304 if got := timersOf(t, reg, holdHost); !sameTimers(got, before) {
1305 t.Fatalf("a gated row's bookkeeping moved:\n got %+v\nwant %+v", got, before)
1306 }
1307 if r.Busy(holdHost) {
1308 t.Fatal("the runner kept a claim on a gated host")
1309 }
1310 if err := reg.CommitImport(holdHost, gateSeedID); err != nil {
1311 t.Fatal(err)
1312 }
1313 r.tick()
1314 if delivered != 1 {
1315 t.Fatalf("after commit delivered %d, want 1", delivered)
1316 }
1317 })
1318 }
1319}
1320
1321// M14: the swap lands between the tick's pre-filter read and its claim. The
1322// decision read inside the claim sees the gate and lets go: nothing is
1323// delivered and the bookkeeping does not move.
1324func TestGateLandingBeforeTheClaimGetsNoTick(t *testing.T) {
1325 d := &fakeDeliverer{status: 204}
1326 reg, r := runnerFixture(t, d)
1327 now := at("2026-08-23T13:07:00Z")
1328 r.Now = func() time.Time { return now }
1329 r.Gate = &HostGate{}
1330 reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1})
1331 before := timersOf(t, reg, holdHost)
1332 swapped := false
1333 r.beforeClaim = func(host string) {
1334 if host == holdHost && !swapped {
1335 swapped = true
1336 gateForImport(t, reg, holdHost)
1337 }
1338 }
1339 r.tick()
1340 if !swapped {
1341 t.Fatal("test bug: the tick never reached its claim")
1342 }
1343 if len(d.calls) != 0 {
1344 t.Fatalf("a row gated before the claim got %d tick(s)", len(d.calls))
1345 }
1346 if got := timersOf(t, reg, holdHost); !sameTimers(got, before) {
1347 t.Fatalf("the gated row's bookkeeping moved:\n got %+v\nwant %+v", got, before)
1348 }
1349 if r.Busy(holdHost) {
1350 t.Fatal("the tick kept its claim after seeing the gate")
1351 }
1352}
1353
1354// Final review F5: 64 gated due timers (four uncommitted imports with 16
1355// each, due first) and one ordinary due timer on another host. The batch
1356// leaves the gated rows out before its limit, so the ordinary timer is
1357// delivered on the first tick, and the gated rows' bookkeeping does not
1358// move.
1359func TestGatedTimersDoNotStarveAnotherHost(t *testing.T) {
1360 for _, armed := range []bool{false, true} {
1361 t.Run(fmt.Sprintf("holds armed %v", armed), func(t *testing.T) {
1362 reg, r := runnerFixture(t, nil)
1363 now := at("2026-08-23T13:07:00Z")
1364 r.Now = func() time.Time { return now }
1365 if armed {
1366 r.Gate = &HostGate{}
1367 }
1368 var delivered []string
1369 r.Deliver = deliverFunc(func(rt registry.Route, _ *http.Request) (int, error) {
1370 delivered = append(delivered, rt.Host)
1371 return 204, nil
1372 })
1373 var gated []string
1374 for i := range 4 {
1375 host := fmt.Sprintf("g%d.bab.oncarlos.com", i)
1376 gated = append(gated, host)
1377 if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true}); err != nil {
1378 t.Fatal(err)
1379 }
1380 gateForImport(t, reg, host)
1381 for n := range dueBatch / 4 {
1382 reg.UpsertTimer(registry.Timer{Host: host, Name: fmt.Sprintf("t%d", n), Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: int64(1 + n)})
1383 }
1384 }
1385 reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 100})
1386 before := map[string][]registry.Timer{}
1387 for _, h := range gated {
1388 before[h] = timersOf(t, reg, h)
1389 }
1390 r.tick()
1391 r.wait()
1392 if len(delivered) != 1 || delivered[0] != holdHost {
1393 t.Fatalf("delivered %q, want the ordinary host's timer alone", delivered)
1394 }
1395 for _, h := range gated {
1396 if got := timersOf(t, reg, h); !sameTimers(got, before[h]) {
1397 t.Fatalf("%s's gated rows moved:\n got %+v\nwant %+v", h, got, before[h])
1398 }
1399 }
1400 })
1401 }
1402}
1403
1404// Skip (harness Run 9, defect 8): a host the agent says not to touch, here a
1405// held row with the move-hold wiring off (an offline move's source after
1406// open), gets no token, no delivery and no bookkeeping. And a hold written
1407// between the tick's route read and the fire is seen before the token, the
1408// one write a tick makes on the instance's own ground.
1409func TestSkipMintsNoTokenAndRecordsNothing(t *testing.T) {
1410 d := &fakeDeliverer{status: 204}
1411 reg, r := runnerFixture(t, d)
1412 now := at("2026-10-20T17:27:04Z")
1413 r.Now = func() time.Time { return now }
1414 tokens := 0
1415 r.Token = func(string) (string, error) { tokens++; return "tok", nil }
1416 r.Skip = func(rt registry.Route) bool { return registry.IsHold(rt.Status) }
1417 if err := reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "every-minute", Kind: KindSchedule, Spec: "every 1m0s", Path: "/tick", NextAt: now.Unix() - 5}); err != nil {
1418 t.Fatal(err)
1419 }
1420 if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
1421 t.Fatal(err)
1422 }
1423 before := timersOf(t, reg, holdHost)
1424 r.tick()
1425 if tokens != 0 || len(d.calls) != 0 || !sameTimers(timersOf(t, reg, holdHost), before) || r.Busy(holdHost) {
1426 t.Fatalf("a skipped host: %d tokens, %d deliveries, timers %+v (were %+v)", tokens, len(d.calls), timersOf(t, reg, holdHost), before)
1427 }
1428
1429 // The row reads ready at the tick's read; the hold lands before the claim.
1430 if ok, err := reg.ReleaseHold(holdHost, "mv_a"); err != nil || !ok {
1431 t.Fatalf("release hold = %v %v", ok, err)
1432 }
1433 r.beforeClaim = func(host string) {
1434 if err := reg.SetHold(host, "moving", "mv_b"); err != nil {
1435 t.Error(err)
1436 }
1437 }
1438 r.tick()
1439 if tokens != 0 || len(d.calls) != 0 || !sameTimers(timersOf(t, reg, holdHost), before) {
1440 t.Fatalf("a hold written before the fire: %d tokens, %d deliveries", tokens, len(d.calls))
1441 }
1442
1443 // Ready again, nothing skipped: the tick fires as it always has.
1444 if ok, err := reg.ReleaseHold(holdHost, "mv_b"); err != nil || !ok {
1445 t.Fatalf("release hold = %v %v", ok, err)
1446 }
1447 r.beforeClaim = nil
1448 r.tick()
1449 if tokens != 1 || len(d.calls) != 1 {
1450 t.Fatalf("a ready host: %d tokens, %d deliveries", tokens, len(d.calls))
1451 }
1452}
1453
1454// Astra's review of moves-offline-fixes, finding 1: with the hold wiring off
1455// (Gate nil, as on an offline move's boxes), a host Skip names must not fill
1456// the due batch. 64 earlier due rows of a skipped host, held or not, and one
1457// later row of an ordinary host: the ordinary one fires on the first tick.
1458func TestSkippedHostsDoNotStarveAnotherHostsTimer(t *testing.T) {
1459 for _, held := range []bool{true, false} {
1460 d := &fakeDeliverer{status: 204}
1461 reg, r := runnerFixture(t, d)
1462 now := at("2026-10-20T17:27:04Z")
1463 r.Now = func() time.Time { return now }
1464 const other = "b.bab.oncarlos.com"
1465 if err := reg.Upsert(registry.Route{Host: other, Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true}); err != nil {
1466 t.Fatal(err)
1467 }
1468 if held {
1469 if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
1470 t.Fatal(err)
1471 }
1472 r.Skip = func(rt registry.Route) bool { return registry.IsHold(rt.Status) }
1473 } else {
1474 // Ready, but named elsewhere (an offline move's record names
1475 // another region): only Skip knows.
1476 r.Skip = func(rt registry.Route) bool { return rt.Host == holdHost }
1477 }
1478 for i := range dueBatch {
1479 if err := reg.UpsertTimer(registry.Timer{Host: holdHost, Name: fmt.Sprintf("t%02d", i), Kind: KindSchedule, Spec: "every 1m0s", Path: "/tick", NextAt: now.Unix() - 1000 + int64(i)}); err != nil {
1480 t.Fatal(err)
1481 }
1482 }
1483 if err := reg.UpsertTimer(registry.Timer{Host: other, Name: "later", Kind: KindSchedule, Spec: "every 1m0s", Path: "/tick", NextAt: now.Unix() - 5}); err != nil {
1484 t.Fatal(err)
1485 }
1486 if r.Gate != nil {
1487 t.Fatal("fixture: the hold wiring is armed")
1488 }
1489 r.tick()
1490 if len(d.calls) != 1 || d.hosts[0] != other {
1491 t.Fatalf("held %v: deliveries to %v, want exactly one to %s", held, d.hosts, other)
1492 }
1493 }
1494}
1495