| 1 | package registry |
| 2 | |
| 3 | import ( |
| 4 | "database/sql" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "reflect" |
| 10 | "testing" |
| 11 | ) |
| 12 | |
| 13 | func openTimersReg(t *testing.T) *Registry { |
| 14 | t.Helper() |
| 15 | r, err := Open(filepath.Join(t.TempDir(), "registry.db")) |
| 16 | if err != nil { |
| 17 | t.Fatal(err) |
| 18 | } |
| 19 | t.Cleanup(func() { r.Close() }) |
| 20 | return r |
| 21 | } |
| 22 | |
| 23 | func TestTimersUpsertPreservesRunnerFields(t *testing.T) { |
| 24 | r := openTimersReg(t) |
| 25 | if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "schedule", Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: 100, CreatedAt: 1}); err != nil { |
| 26 | t.Fatal(err) |
| 27 | } |
| 28 | if _, err := r.MarkTimerFired("h", "sync", timerRegistration(t, r, "h", "sync"), 100, "204", 300, 0, 21700, 21700); err != nil { |
| 29 | t.Fatal(err) |
| 30 | } |
| 31 | // Converge sees the same declaration again: spec/path unchanged, next_at |
| 32 | // passed along as "whatever it is now" (0 here means "don't touch"). |
| 33 | if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "schedule", Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: 0, CreatedAt: 99}); err != nil { |
| 34 | t.Fatal(err) |
| 35 | } |
| 36 | got, err := r.GetTimer("h", "sync") |
| 37 | if err != nil { |
| 38 | t.Fatal(err) |
| 39 | } |
| 40 | if got.LastAt != 100 || got.LastStatus != "204" || got.LastMs != 300 || got.NextAt != 21700 || got.CreatedAt != 1 { |
| 41 | t.Fatalf("runner fields clobbered: %+v", got) |
| 42 | } |
| 43 | // A changed spec DOES move next_at. |
| 44 | if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "schedule", Spec: "every 12h0m0s", Path: "/jobs/sync", NextAt: 43200}); err != nil { |
| 45 | t.Fatal(err) |
| 46 | } |
| 47 | got, _ = r.GetTimer("h", "sync") |
| 48 | if got.NextAt != 43200 || got.Spec != "every 12h0m0s" || got.LastStatus != "204" { |
| 49 | t.Fatalf("after spec change: %+v", got) |
| 50 | } |
| 51 | if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "once", Spec: "at 5", Path: "/p", NextAt: 5}); !errors.Is(err, ErrTimerKind) { |
| 52 | t.Fatalf("kind change = %v, want ErrTimerKind", err) |
| 53 | } |
| 54 | } |
| 55 | |
| 56 | func TestTimersDueOrderingAndFireNow(t *testing.T) { |
| 57 | r := openTimersReg(t) |
| 58 | for _, tm := range []Timer{ |
| 59 | {Host: "b", Name: "late", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 500}, |
| 60 | {Host: "a", Name: "soon", Kind: "once", Spec: "at 10", Path: "/p", NextAt: 10}, |
| 61 | {Host: "a", Name: "done", Kind: "once", Spec: "at 1", Path: "/p", NextAt: 0}, |
| 62 | {Host: "c", Name: "future", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 9999}, |
| 63 | } { |
| 64 | if err := r.UpsertTimer(tm); err != nil { |
| 65 | t.Fatal(err) |
| 66 | } |
| 67 | } |
| 68 | due, err := r.DueTimers(600, 10, nil) |
| 69 | if err != nil { |
| 70 | t.Fatal(err) |
| 71 | } |
| 72 | if len(due) != 2 || due[0].Name != "soon" || due[1].Name != "late" { |
| 73 | t.Fatalf("due = %+v", due) |
| 74 | } |
| 75 | if due, _ := r.DueTimers(600, 1, nil); len(due) != 1 { |
| 76 | t.Fatal("limit ignored") |
| 77 | } |
| 78 | if err := r.FireTimerNow("c", "future", 600); err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | if due, _ := r.DueTimers(600, 10, []string{}); len(due) != 3 { // an empty exclude set is no filter |
| 82 | t.Fatalf("after FireTimerNow: %d due", len(due)) |
| 83 | } |
| 84 | if err := r.FireTimerNow("nope", "x", 600); err != nil { |
| 85 | t.Fatalf("FireTimerNow on absent = %v, want nil", err) |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | func TestTimersPruneAndExcept(t *testing.T) { |
| 90 | r := openTimersReg(t) |
| 91 | for _, tm := range []Timer{ |
| 92 | {Host: "alive", Name: "s1", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 1}, |
| 93 | {Host: "alive", Name: "s2", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 1}, |
| 94 | {Host: "alive", Name: "o1", Kind: "once", Spec: "at 1", Path: "/p", NextAt: 0, LastAt: 10}, |
| 95 | {Host: "alive", Name: "o2", Kind: "once", Spec: "at 1", Path: "/p", NextAt: 0, LastAt: 500}, |
| 96 | {Host: "gone", Name: "s1", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 1}, |
| 97 | } { |
| 98 | if err := r.UpsertTimer(tm); err != nil { |
| 99 | t.Fatal(err) |
| 100 | } |
| 101 | } |
| 102 | // o1 has last_at=10 via insert — MarkTimerFired is the real path but insert suffices to seed. |
| 103 | n, err := r.PruneTimers(map[string]bool{"alive": true}, 100) |
| 104 | if err != nil || n != 2 { // gone/s1 and alive/o1 |
| 105 | t.Fatalf("prune = %d %v", n, err) |
| 106 | } |
| 107 | n, err = r.DeleteTimersExcept("alive", "schedule", map[string]bool{"s2": true}) |
| 108 | if err != nil || n != 1 { |
| 109 | t.Fatalf("except = %d %v", n, err) |
| 110 | } |
| 111 | left, _ := r.TimersFor("alive") |
| 112 | if len(left) != 2 || left[0].Name != "o2" || left[1].Name != "s2" { |
| 113 | t.Fatalf("left = %+v", left) |
| 114 | } |
| 115 | if c, _ := r.CountTimers("alive", "once"); c != 1 { |
| 116 | t.Fatalf("count = %d", c) |
| 117 | } |
| 118 | if _, err := r.GetTimer("gone", "s1"); !errors.Is(err, os.ErrNotExist) { |
| 119 | t.Fatalf("gone = %v", err) |
| 120 | } |
| 121 | if err := r.DeleteTimer("alive", "s2"); err != nil { |
| 122 | t.Fatal(err) |
| 123 | } |
| 124 | if err := r.DeleteTimer("alive", "s2"); err != nil { |
| 125 | t.Fatal("delete not idempotent") |
| 126 | } |
| 127 | } |
| 128 | |
| 129 | // due_at is the occurrence a row is delivering (X-Carlos-Schedule-At). It |
| 130 | // rides next_at on an upsert that moves it, follows the runner's word on |
| 131 | // MarkTimerFired, and IS "now" after FireTimerNow — and a registry written |
| 132 | // before the column existed gains it on open, with 0 on the old rows. |
| 133 | func TestTimersDueAtFollowsTheOccurrence(t *testing.T) { |
| 134 | r := openTimersReg(t) |
| 135 | if err := r.UpsertTimer(Timer{Host: "h", Name: "s", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 3600, DueAt: 3600}); err != nil { |
| 136 | t.Fatal(err) |
| 137 | } |
| 138 | // A retry: next moves to the retry instant, due stays the occurrence. |
| 139 | if _, err := r.MarkTimerFired("h", "s", timerRegistration(t, r, "h", "s"), 3600, "503", 1, 1, 3660, 3600); err != nil { |
| 140 | t.Fatal(err) |
| 141 | } |
| 142 | got, _ := r.GetTimer("h", "s") |
| 143 | if got.NextAt != 3660 || got.DueAt != 3600 { |
| 144 | t.Fatalf("after retry: %+v", got) |
| 145 | } |
| 146 | // An advance: both move together. |
| 147 | if _, err := r.MarkTimerFired("h", "s", timerRegistration(t, r, "h", "s"), 3660, "204", 1, 0, 7200, 7200); err != nil { |
| 148 | t.Fatal(err) |
| 149 | } |
| 150 | if got, _ = r.GetTimer("h", "s"); got.NextAt != 7200 || got.DueAt != 7200 { |
| 151 | t.Fatalf("after advance: %+v", got) |
| 152 | } |
| 153 | // An unchanged upsert (NextAt 0) leaves both alone; a spec change moves both. |
| 154 | if err := r.UpsertTimer(Timer{Host: "h", Name: "s", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p"}); err != nil { |
| 155 | t.Fatal(err) |
| 156 | } |
| 157 | if got, _ = r.GetTimer("h", "s"); got.NextAt != 7200 || got.DueAt != 7200 { |
| 158 | t.Fatalf("after unchanged upsert: %+v", got) |
| 159 | } |
| 160 | if err := r.UpsertTimer(Timer{Host: "h", Name: "s", Kind: "schedule", Spec: "every 2h0m0s", Path: "/p", NextAt: 9000, DueAt: 9000}); err != nil { |
| 161 | t.Fatal(err) |
| 162 | } |
| 163 | if got, _ = r.GetTimer("h", "s"); got.NextAt != 9000 || got.DueAt != 9000 { |
| 164 | t.Fatalf("after spec change: %+v", got) |
| 165 | } |
| 166 | // `run`: now is its own occurrence. |
| 167 | if err := r.FireTimerNow("h", "s", 8000); err != nil { |
| 168 | t.Fatal(err) |
| 169 | } |
| 170 | if got, _ = r.GetTimer("h", "s"); got.NextAt != 8000 || got.DueAt != 8000 { |
| 171 | t.Fatalf("after FireTimerNow: %+v", got) |
| 172 | } |
| 173 | } |
| 174 | |
| 175 | func TestTimersPreDueAtRegistryGainsTheColumn(t *testing.T) { |
| 176 | path := filepath.Join(t.TempDir(), "registry.db") |
| 177 | reg, err := Open(path) |
| 178 | if err != nil { |
| 179 | t.Fatal(err) |
| 180 | } |
| 181 | if err := reg.Close(); err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | old, err := sql.Open("sqlite", path) |
| 185 | if err != nil { |
| 186 | t.Fatal(err) |
| 187 | } |
| 188 | if _, err := old.Exec(`ALTER TABLE timers DROP COLUMN due_at`); err != nil { |
| 189 | t.Fatalf("simulate a pre-due_at registry: %v", err) |
| 190 | } |
| 191 | if _, err := old.Exec(`INSERT INTO timers (host, name, kind, spec, path, next_at, created_at) VALUES ('h', 'legacy', 'schedule', 'every 1h0m0s', '/p', 3600, 1)`); err != nil { |
| 192 | t.Fatal(err) |
| 193 | } |
| 194 | if err := old.Close(); err != nil { |
| 195 | t.Fatal(err) |
| 196 | } |
| 197 | // Open twice: the ALTER must be idempotent, not a one-shot. |
| 198 | for i := 0; i < 2; i++ { |
| 199 | reg, err = Open(path) |
| 200 | if err != nil { |
| 201 | t.Fatalf("open %d of a pre-due_at registry: %v", i+1, err) |
| 202 | } |
| 203 | got, err := reg.GetTimer("h", "legacy") |
| 204 | if err != nil { |
| 205 | t.Fatal(err) |
| 206 | } |
| 207 | if got.NextAt != 3600 || got.DueAt != 0 { |
| 208 | t.Fatalf("migrated row = %+v, want next_at kept and due_at 0", got) |
| 209 | } |
| 210 | if err := reg.Close(); err != nil { |
| 211 | t.Fatal(err) |
| 212 | } |
| 213 | } |
| 214 | } |
| 215 | |
| 216 | // The batch is capped, and the runner skips hosts whose delivery is still |
| 217 | // running: without the exclude filter a busy host with a batch's worth of due |
| 218 | // rows takes every slot of every batch and another host's due timer is never |
| 219 | // SEEN, let alone fired. The filter belongs in the query, before the LIMIT. |
| 220 | func TestDueTimersExcludesBusyHostsBeforeTheLimit(t *testing.T) { |
| 221 | r := openTimersReg(t) |
| 222 | for i := 0; i < 8; i++ { |
| 223 | if err := r.UpsertTimer(Timer{Host: "busy", Name: "j" + string(rune('a'+i)), Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 100}); err != nil { |
| 224 | t.Fatal(err) |
| 225 | } |
| 226 | } |
| 227 | // Due LATER than every one of the busy host's rows, so ordering alone |
| 228 | // never rescues it. |
| 229 | if err := r.UpsertTimer(Timer{Host: "quiet", Name: "j", Kind: "schedule", Spec: "every 1h0m0s", Path: "/p", NextAt: 200}); err != nil { |
| 230 | t.Fatal(err) |
| 231 | } |
| 232 | due, err := r.DueTimers(600, 4, nil) |
| 233 | if err != nil { |
| 234 | t.Fatal(err) |
| 235 | } |
| 236 | for _, tm := range due { |
| 237 | if tm.Host == "quiet" { |
| 238 | t.Fatalf("fixture: the quiet host must be crowded out without the filter, got %+v", due) |
| 239 | } |
| 240 | } |
| 241 | due, err = r.DueTimers(600, 4, []string{"busy"}) |
| 242 | if err != nil { |
| 243 | t.Fatal(err) |
| 244 | } |
| 245 | if len(due) != 1 || due[0].Host != "quiet" { |
| 246 | t.Fatalf("excluding the busy host = %+v, want the quiet host's one row", due) |
| 247 | } |
| 248 | // Every host excluded is an empty batch, not a filter that silently |
| 249 | // stops applying. |
| 250 | if due, _ := r.DueTimers(600, 4, []string{"busy", "quiet"}); len(due) != 0 { |
| 251 | t.Fatalf("all hosts excluded = %+v, want none", due) |
| 252 | } |
| 253 | } |
| 254 | |
| 255 | func TestEarliestTimersPerHost(t *testing.T) { |
| 256 | r := openTestRegistry(t) |
| 257 | rows := []Timer{ |
| 258 | // Two rows for one host: the earliest wins. |
| 259 | {Host: "a.example", Name: "late", Kind: "schedule", Spec: "every 1h", Path: "/x", NextAt: 5000, DueAt: 5000}, |
| 260 | {Host: "a.example", Name: "soon", Kind: "schedule", Spec: "every 5m", Path: "/y", NextAt: 1200, DueAt: 1200}, |
| 261 | // A second host is reported independently. |
| 262 | {Host: "b.example", Name: "only", Kind: "schedule", Spec: "every 5m", Path: "/z", NextAt: 3000, DueAt: 3000}, |
| 263 | // next_at == 0 means "not armed" and must never appear. |
| 264 | {Host: "c.example", Name: "parked", Kind: "schedule", Spec: "every 5m", Path: "/p", NextAt: 0, DueAt: 0}, |
| 265 | } |
| 266 | for _, tm := range rows { |
| 267 | if err := r.UpsertTimer(tm); err != nil { |
| 268 | t.Fatalf("UpsertTimer %s/%s: %v", tm.Host, tm.Name, err) |
| 269 | } |
| 270 | } |
| 271 | |
| 272 | got, err := r.EarliestTimers() |
| 273 | if err != nil { |
| 274 | t.Fatalf("EarliestTimers: %v", err) |
| 275 | } |
| 276 | want := map[string]int64{"a.example": 1200, "b.example": 3000} |
| 277 | if !reflect.DeepEqual(got, want) { |
| 278 | t.Errorf("EarliestTimers = %v, want %v", got, want) |
| 279 | } |
| 280 | } |
| 281 | |
| 282 | func TestEarliestTimersEmpty(t *testing.T) { |
| 283 | r := openTestRegistry(t) |
| 284 | got, err := r.EarliestTimers() |
| 285 | if err != nil { |
| 286 | t.Fatalf("EarliestTimers: %v", err) |
| 287 | } |
| 288 | if len(got) != 0 { |
| 289 | t.Errorf("EarliestTimers on an empty table = %v, want empty", got) |
| 290 | } |
| 291 | } |
| 292 | |
| 293 | func timerRegistration(t *testing.T, r *Registry, host, name string) string { |
| 294 | t.Helper() |
| 295 | row, err := r.GetTimer(host, name) |
| 296 | if err != nil { |
| 297 | t.Fatal(err) |
| 298 | } |
| 299 | return row.Registration |
| 300 | } |
| 301 | |
| 302 | func TestTimerRegistrationProtectsMutations(t *testing.T) { |
| 303 | for _, mutation := range []string{"identical", "spec", "path", "manual", "recreate"} { |
| 304 | t.Run(mutation, func(t *testing.T) { |
| 305 | r := openTimersReg(t) |
| 306 | tm := Timer{Host: "h", Name: "s", Kind: "schedule", Spec: "every 1h0m0s", Path: "/jobs/sync", NextAt: 100, DueAt: 100, Attempts: 4, LastAt: 80, LastStatus: "503", LastMs: 1} |
| 307 | if err := r.UpsertTimer(tm); err != nil { |
| 308 | t.Fatal(err) |
| 309 | } |
| 310 | before, err := r.GetTimer(tm.Host, tm.Name) |
| 311 | if err != nil { |
| 312 | t.Fatal(err) |
| 313 | } |
| 314 | switch mutation { |
| 315 | case "spec": |
| 316 | tm.Spec, tm.NextAt, tm.DueAt = "every 2h0m0s", 0, 0 |
| 317 | case "path": |
| 318 | tm.Path, tm.NextAt, tm.DueAt = "/jobs/other", 0, 0 |
| 319 | case "recreate": |
| 320 | if err := r.DeleteTimer(tm.Host, tm.Name); err != nil { |
| 321 | t.Fatal(err) |
| 322 | } |
| 323 | tm.Attempts = 0 |
| 324 | } |
| 325 | if mutation == "manual" { |
| 326 | err = r.FireTimerNow(tm.Host, tm.Name, 100) |
| 327 | } else { |
| 328 | err = r.UpsertTimer(tm) |
| 329 | } |
| 330 | if err != nil { |
| 331 | t.Fatal(err) |
| 332 | } |
| 333 | replacement, err := r.GetTimer(tm.Host, tm.Name) |
| 334 | if err != nil { |
| 335 | t.Fatal(err) |
| 336 | } |
| 337 | if replacement.Registration == "" || replacement.Registration == before.Registration || replacement.Attempts != 0 { |
| 338 | t.Fatalf("new registration: %+v", replacement) |
| 339 | } |
| 340 | if _, err := r.MarkTimerFired(tm.Host, tm.Name, before.Registration, 100, "204", 3, 0, 3700, 3700); err != nil { |
| 341 | t.Fatal(err) |
| 342 | } |
| 343 | if err := r.DeleteTimerRegistration(tm.Host, tm.Name, before.Registration); err != nil { |
| 344 | t.Fatal(err) |
| 345 | } |
| 346 | after, err := r.GetTimer(tm.Host, tm.Name) |
| 347 | if err != nil { |
| 348 | t.Fatal(err) |
| 349 | } |
| 350 | if *after != *replacement { |
| 351 | t.Fatalf("stale mutation changed registration: %+v -> %+v", replacement, after) |
| 352 | } |
| 353 | if err := r.DeleteTimerRegistration(tm.Host, tm.Name, after.Registration); err != nil { |
| 354 | t.Fatal(err) |
| 355 | } |
| 356 | if _, err := r.GetTimer(tm.Host, tm.Name); !errors.Is(err, os.ErrNotExist) { |
| 357 | t.Fatalf("current deletion: %v", err) |
| 358 | } |
| 359 | }) |
| 360 | } |
| 361 | } |
| 362 | |
| 363 | func TestTimerRegistrationLegacyMigration(t *testing.T) { |
| 364 | path := filepath.Join(t.TempDir(), "registry.db") |
| 365 | r, err := Open(path) |
| 366 | if err != nil { |
| 367 | t.Fatal(err) |
| 368 | } |
| 369 | if _, err := r.sql.Exec(`ALTER TABLE timers DROP COLUMN registration`); err != nil { |
| 370 | t.Fatal(err) |
| 371 | } |
| 372 | if _, err := r.sql.Exec(`INSERT INTO timers (host,name,kind,spec,path,next_at,created_at) VALUES ('h','s','once','at 100','/jobs/sync',100,1)`); err != nil { |
| 373 | t.Fatal(err) |
| 374 | } |
| 375 | if err := r.Close(); err != nil { |
| 376 | t.Fatal(err) |
| 377 | } |
| 378 | for i := 0; i < 2; i++ { |
| 379 | r, err = Open(path) |
| 380 | if err != nil { |
| 381 | t.Fatal(err) |
| 382 | } |
| 383 | row, err := r.GetTimer("h", "s") |
| 384 | if err != nil { |
| 385 | t.Fatal(err) |
| 386 | } |
| 387 | if row.Registration != "" { |
| 388 | t.Fatalf("legacy registration: %q", row.Registration) |
| 389 | } |
| 390 | if _, err := r.MarkTimerFired("h", "s", row.Registration, 100, "503", 3, 1, 160, 100); err != nil { |
| 391 | t.Fatal(err) |
| 392 | } |
| 393 | if err := r.Close(); err != nil { |
| 394 | t.Fatal(err) |
| 395 | } |
| 396 | } |
| 397 | r, err = Open(path) |
| 398 | if err != nil { |
| 399 | t.Fatal(err) |
| 400 | } |
| 401 | defer r.Close() |
| 402 | row, err := r.GetTimer("h", "s") |
| 403 | if err != nil { |
| 404 | t.Fatal(err) |
| 405 | } |
| 406 | if row.NextAt != 160 || row.DueAt != 100 || row.Attempts != 1 { |
| 407 | t.Fatalf("legacy completion: %+v", row) |
| 408 | } |
| 409 | if err := r.FireTimerNow("h", "s", 100); err != nil { |
| 410 | t.Fatal(err) |
| 411 | } |
| 412 | if _, err := r.MarkTimerFired("h", "s", "", 100, "204", 3, 0, 0, 0); err != nil { |
| 413 | t.Fatal(err) |
| 414 | } |
| 415 | row, err = r.GetTimer("h", "s") |
| 416 | if err != nil { |
| 417 | t.Fatal(err) |
| 418 | } |
| 419 | if row.Registration == "" || row.NextAt != 100 || row.Attempts != 0 { |
| 420 | t.Fatalf("rearmed legacy: %+v", row) |
| 421 | } |
| 422 | } |
| 423 | |
| 424 | func TestUnchangedTimerUpsertConcurrentWithCompletion(t *testing.T) { |
| 425 | r := openTimersReg(t) |
| 426 | tm := Timer{Host: "h", Name: "s", Kind: "schedule", Spec: "every 1h0m0s", Path: "/jobs/sync", NextAt: 100, DueAt: 100} |
| 427 | if err := r.UpsertTimer(tm); err != nil { |
| 428 | t.Fatal(err) |
| 429 | } |
| 430 | registration := timerRegistration(t, r, tm.Host, tm.Name) |
| 431 | tm.NextAt, tm.DueAt = 0, 0 |
| 432 | for i := int64(1); i <= 100; i++ { |
| 433 | start := make(chan struct{}) |
| 434 | errs := make(chan error, 9) |
| 435 | for n := 0; n < 8; n++ { |
| 436 | go func() { <-start; errs <- r.UpsertTimer(tm) }() |
| 437 | } |
| 438 | go func() { |
| 439 | <-start |
| 440 | _, err := r.MarkTimerFired(tm.Host, tm.Name, registration, i, "503", 1, 1, 100+i, 100) |
| 441 | errs <- err |
| 442 | }() |
| 443 | close(start) |
| 444 | for n := 0; n < 9; n++ { |
| 445 | if err := <-errs; err != nil { |
| 446 | t.Fatal(err) |
| 447 | } |
| 448 | } |
| 449 | row, err := r.GetTimer(tm.Host, tm.Name) |
| 450 | if err != nil { |
| 451 | t.Fatal(err) |
| 452 | } |
| 453 | if row.NextAt != 100+i || row.DueAt != 100 || row.LastAt != i || row.Attempts != 1 || row.Registration != registration { |
| 454 | t.Fatalf("completion overwritten at iteration %d: %+v", i, row) |
| 455 | } |
| 456 | } |
| 457 | } |
| 458 | |
| 459 | func TestReplaceTimersSwapsAHostsRowsExactly(t *testing.T) { |
| 460 | r := openTimersReg(t) |
| 461 | for _, tm := range []Timer{ |
| 462 | {Host: "a.test", Name: "old", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 100, CreatedAt: 1}, |
| 463 | {Host: "b.test", Name: "other", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 100, CreatedAt: 1}, |
| 464 | } { |
| 465 | if err := r.UpsertTimer(tm); err != nil { |
| 466 | t.Fatal(err) |
| 467 | } |
| 468 | } |
| 469 | rows := []Timer{ |
| 470 | {Host: "a.test", Name: "nightly", Kind: "schedule", Spec: "0 3 * * *", Path: "/n", NextAt: 5000, LastAt: 4000, |
| 471 | LastStatus: "ok", LastMs: 12, Attempts: 2, CreatedAt: 10, DueAt: 4900, Registration: "reg-nightly"}, |
| 472 | {Host: "a.test", Name: "once-1", Kind: "once", Spec: "2026-10-06T00:00:00Z", Path: "/o", NextAt: 6000, CreatedAt: 11, DueAt: 6000, Registration: "reg-once"}, |
| 473 | } |
| 474 | if err := r.ReplaceTimers("a.test", rows); err != nil { |
| 475 | t.Fatal(err) |
| 476 | } |
| 477 | got, err := r.TimersFor("a.test") |
| 478 | if err != nil { |
| 479 | t.Fatal(err) |
| 480 | } |
| 481 | if len(got) != 2 { |
| 482 | t.Fatalf("a.test has %d timers, want 2: %+v", len(got), got) |
| 483 | } |
| 484 | n := got[0] // ordered by name: nightly, once-1 |
| 485 | if n.Name != "nightly" || n.NextAt != 5000 || n.LastAt != 4000 || n.LastStatus != "ok" || n.LastMs != 12 || |
| 486 | n.Attempts != 2 || n.CreatedAt != 10 || n.DueAt != 4900 || n.Registration != "reg-nightly" { |
| 487 | t.Fatalf("bookkeeping not preserved: %+v", n) |
| 488 | } |
| 489 | if other, _ := r.TimersFor("b.test"); len(other) != 1 { |
| 490 | t.Fatalf("b.test has %d timers, want 1: another host's rows were touched", len(other)) |
| 491 | } |
| 492 | } |
| 493 | |
| 494 | func TestReplaceTimersIsAllOrNothing(t *testing.T) { |
| 495 | r := openTimersReg(t) |
| 496 | if err := r.UpsertTimer(Timer{Host: "a.test", Name: "keep", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 100, CreatedAt: 1}); err != nil { |
| 497 | t.Fatal(err) |
| 498 | } |
| 499 | // Two rows with one name: the delete and the first insert succeed, the |
| 500 | // second insert breaks the primary key. Without a transaction the host |
| 501 | // would be left with "dup" and without "keep". |
| 502 | dup := []Timer{ |
| 503 | {Host: "a.test", Name: "dup", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 1, CreatedAt: 1}, |
| 504 | {Host: "a.test", Name: "dup", Kind: "schedule", Spec: "@every 2m", Path: "/t", NextAt: 1, CreatedAt: 1}, |
| 505 | } |
| 506 | if err := r.ReplaceTimers("a.test", dup); err == nil { |
| 507 | t.Fatal("two rows with one name were accepted") |
| 508 | } |
| 509 | got, _ := r.TimersFor("a.test") |
| 510 | if len(got) != 1 || got[0].Name != "keep" { |
| 511 | t.Fatalf("a failed replace changed the rows: %+v", got) |
| 512 | } |
| 513 | } |
| 514 | |
| 515 | func TestReplaceTimersRefusesARowForAnotherHost(t *testing.T) { |
| 516 | r := openTimersReg(t) |
| 517 | rows := []Timer{{Host: "elsewhere.test", Name: "y", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 1, CreatedAt: 1}} |
| 518 | if err := r.ReplaceTimers("a.test", rows); err == nil { |
| 519 | t.Fatal("a row for another host was accepted") |
| 520 | } |
| 521 | } |
| 522 | |
| 523 | func TestReplaceTimersWithNoRowsClearsTheHost(t *testing.T) { |
| 524 | r := openTimersReg(t) |
| 525 | if err := r.UpsertTimer(Timer{Host: "a.test", Name: "gone", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 100, CreatedAt: 1}); err != nil { |
| 526 | t.Fatal(err) |
| 527 | } |
| 528 | if err := r.ReplaceTimers("a.test", nil); err != nil { |
| 529 | t.Fatal(err) |
| 530 | } |
| 531 | if got, _ := r.TimersFor("a.test"); len(got) != 0 { |
| 532 | t.Fatalf("rows left: %+v", got) |
| 533 | } |
| 534 | } |
| 535 | |
| 536 | // Moves Task 6 fix round 1: DueUnheldTimers leaves out every row whose |
| 537 | // route is held (moves spec §5.2) BEFORE the limit, so a held host's rows |
| 538 | // cannot fill the batch and hide another host's due timer. DueTimers itself |
| 539 | // is unchanged: with the hold wiring off the runner still asks it. |
| 540 | func TestDueUnheldTimersExcludesHeldHostsBeforeTheLimit(t *testing.T) { |
| 541 | r := openTimersReg(t) |
| 542 | for _, h := range []string{"held.test", "free.test"} { |
| 543 | if err := r.Upsert(Route{Host: h, Kind: "instance", Socket: "/s"}); err != nil { |
| 544 | t.Fatal(err) |
| 545 | } |
| 546 | } |
| 547 | if err := r.SetHold("held.test", "moving", holdOpA); err != nil { |
| 548 | t.Fatal(err) |
| 549 | } |
| 550 | for i, name := range []string{"t0", "t1", "t2"} { |
| 551 | if err := r.UpsertTimer(Timer{Host: "held.test", Name: name, Kind: "schedule", Spec: "every 1h0m0s", Path: "/j", NextAt: int64(10 + i)}); err != nil { |
| 552 | t.Fatal(err) |
| 553 | } |
| 554 | } |
| 555 | // Due after every held row; and a row whose host has no route at all, |
| 556 | // which neither query drops (the converger prunes those). |
| 557 | if err := r.UpsertTimer(Timer{Host: "free.test", Name: "t", Kind: "schedule", Spec: "every 1h0m0s", Path: "/j", NextAt: 50}); err != nil { |
| 558 | t.Fatal(err) |
| 559 | } |
| 560 | if err := r.UpsertTimer(Timer{Host: "gone.test", Name: "t", Kind: "schedule", Spec: "every 1h0m0s", Path: "/j", NextAt: 60}); err != nil { |
| 561 | t.Fatal(err) |
| 562 | } |
| 563 | due, err := r.DueTimers(100, 2, nil) |
| 564 | if err != nil { |
| 565 | t.Fatal(err) |
| 566 | } |
| 567 | if len(due) != 2 || due[0].Host != "held.test" || due[1].Host != "held.test" { |
| 568 | t.Fatalf("DueTimers = %+v, want the two earliest held rows (main's answer)", due) |
| 569 | } |
| 570 | un, err := r.DueUnheldTimers(100, 2, nil) |
| 571 | if err != nil { |
| 572 | t.Fatal(err) |
| 573 | } |
| 574 | if len(un) != 2 || un[0].Host != "free.test" || un[1].Host != "gone.test" { |
| 575 | t.Fatalf("DueUnheldTimers = %+v, want free.test then gone.test", un) |
| 576 | } |
| 577 | // The busy-host exclusion still applies on top. |
| 578 | un, err = r.DueUnheldTimers(100, 2, []string{"free.test"}) |
| 579 | if err != nil { |
| 580 | t.Fatal(err) |
| 581 | } |
| 582 | if len(un) != 1 || un[0].Host != "gone.test" { |
| 583 | t.Fatalf("DueUnheldTimers excluding free.test = %+v", un) |
| 584 | } |
| 585 | // Once the hold lifts the rows are due again, in order. |
| 586 | if ok, _ := r.ReleaseHold("held.test", holdOpA); !ok { |
| 587 | t.Fatal("release") |
| 588 | } |
| 589 | un, _ = r.DueUnheldTimers(100, 2, nil) |
| 590 | if len(un) != 2 || un[0].Host != "held.test" { |
| 591 | t.Fatalf("after the release DueUnheldTimers = %+v", un) |
| 592 | } |
| 593 | } |
| 594 | |
| 595 | // Final review F5: both due queries leave out every row of a gated route |
| 596 | // (an uncommitted import's) before the limit. The runner skips such a row |
| 597 | // without advancing it, so 64 of them, due first, would otherwise fill the |
| 598 | // runner's whole batch on every tick and hide another host's due timer for |
| 599 | // ever. A row whose host has no route is still returned. |
| 600 | func TestDueTimersExcludeGatedRoutesBeforeTheLimit(t *testing.T) { |
| 601 | r := openTimersReg(t) |
| 602 | const batch = 64 |
| 603 | for i := range 4 { |
| 604 | host := fmt.Sprintf("gated%d.test", i) |
| 605 | if err := r.Upsert(Route{Host: host, Kind: "instance", Socket: "/s", ImportGate: ImportGateClosed}); err != nil { |
| 606 | t.Fatal(err) |
| 607 | } |
| 608 | for n := range batch / 4 { |
| 609 | if err := r.UpsertTimer(Timer{Host: host, Name: fmt.Sprintf("t%d", n), Kind: "schedule", Spec: "every 1h0m0s", Path: "/j", |
| 610 | NextAt: int64(10 + n)}); err != nil { |
| 611 | t.Fatal(err) |
| 612 | } |
| 613 | } |
| 614 | } |
| 615 | if err := r.Upsert(Route{Host: "free.test", Kind: "instance", Socket: "/s"}); err != nil { |
| 616 | t.Fatal(err) |
| 617 | } |
| 618 | if err := r.UpsertTimer(Timer{Host: "free.test", Name: "t", Kind: "schedule", Spec: "every 1h0m0s", Path: "/j", NextAt: 90}); err != nil { |
| 619 | t.Fatal(err) |
| 620 | } |
| 621 | if err := r.UpsertTimer(Timer{Host: "gone.test", Name: "t", Kind: "schedule", Spec: "every 1h0m0s", Path: "/j", NextAt: 95}); err != nil { |
| 622 | t.Fatal(err) |
| 623 | } |
| 624 | for name, due := range map[string]func(int64, int, []string) ([]Timer, error){ |
| 625 | "DueTimers": r.DueTimers, "DueUnheldTimers": r.DueUnheldTimers, |
| 626 | } { |
| 627 | got, err := due(100, batch, nil) |
| 628 | if err != nil { |
| 629 | t.Fatal(err) |
| 630 | } |
| 631 | var hosts []string |
| 632 | for _, tm := range got { |
| 633 | hosts = append(hosts, tm.Host) |
| 634 | } |
| 635 | if len(hosts) != 2 || hosts[0] != "free.test" || hosts[1] != "gone.test" { |
| 636 | t.Fatalf("%s hosts = %q, want free.test then gone.test", name, hosts) |
| 637 | } |
| 638 | } |
| 639 | } |
| 640 | |
amadan