| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "errors" |
| 5 | "fmt" |
| 6 | "log" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "strings" |
| 10 | "sync/atomic" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/registry" |
| 15 | "github.com/carlosframework/platform/internal/release" |
| 16 | ) |
| 17 | |
| 18 | // stagedActivator is shippingActivatorWithRegistry over stagedFake: ver is |
| 19 | // what live/<host> names, and every spawn reports the identity it exec'd. |
| 20 | func stagedActivator(t *testing.T) (*Service, *stagedFake, *fakeClock, *registry.Registry, *atomic.Value) { |
| 21 | t.Helper() |
| 22 | fb := &fakeBackend{} |
| 23 | b := &stagedFake{fakeBackend: fb} |
| 24 | clk := &fakeClock{t: time.Unix(1_700_000_000, 0)} |
| 25 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 26 | if err != nil { |
| 27 | t.Fatalf("open registry: %v", err) |
| 28 | } |
| 29 | a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg) |
| 30 | a.now = clk.now |
| 31 | var ver atomic.Value |
| 32 | ver.Store("v1") |
| 33 | fb.execVerFn = func() string { return ver.Load().(string) } |
| 34 | a.liveVer = func(string) (string, error) { return ver.Load().(string), nil } |
| 35 | return a, b, clk, reg, &ver |
| 36 | } |
| 37 | |
| 38 | func (a *Service) entryServing(host string) release.StagedIdentity { |
| 39 | a.mu.Lock() |
| 40 | defer a.mu.Unlock() |
| 41 | if e := a.entries[host]; e != nil { |
| 42 | return e.workerServing |
| 43 | } |
| 44 | return release.StagedIdentity{} |
| 45 | } |
| 46 | |
| 47 | // Spec §7.9: a crash recycle respawns the serving build, not the installed one. |
| 48 | func TestWorkerCrashRecyclesTheCapturedServingBuild(t *testing.T) { |
| 49 | t.Parallel() |
| 50 | a, b, _, _, ver := stagedActivator(t) |
| 51 | host := "bot.x.internal" |
| 52 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 53 | t.Fatalf("WakeWorker: %v", err) |
| 54 | } |
| 55 | if got := a.entryServing(host); got != fakeIdentity("v1") { |
| 56 | t.Fatalf("serving after wake = %+v, want the captured v1", got) |
| 57 | } |
| 58 | ver.Store("v2") // root installs v2 under the running child |
| 59 | b.lastWorker().kill() |
| 60 | a.sweepOnce() |
| 61 | waitForRestart(t, a, b.fakeBackend, host, 2) |
| 62 | calls := b.calls() |
| 63 | if len(calls) != 2 || calls[0] != nil { |
| 64 | t.Fatalf("spawns = %v; want a wake from the live link, then a recycle", calls) |
| 65 | } |
| 66 | if calls[1] == nil || *calls[1] != fakeIdentity("v1") { |
| 67 | t.Fatalf("crash recycle asked for %v; want the captured v1 build", calls[1]) |
| 68 | } |
| 69 | if got := a.entrySpawnVersion(host); got != "v1" { |
| 70 | t.Fatalf("spawn version = %q, want v1", got) |
| 71 | } |
| 72 | } |
| 73 | |
| 74 | // Spec §7.9: so does an always-on worker's max_run recycle. |
| 75 | func TestWorkerMaxRunRecyclesTheCapturedServingBuild(t *testing.T) { |
| 76 | t.Parallel() |
| 77 | a, b, clk, _, ver := stagedActivator(t) |
| 78 | host := "bot.x.internal" |
| 79 | rt := alwaysOnWorkerRoute(host) |
| 80 | rt.MaxRunSeconds = 30 |
| 81 | if err := a.WakeWorker(rt); err != nil { |
| 82 | t.Fatalf("WakeWorker: %v", err) |
| 83 | } |
| 84 | ver.Store("v2") |
| 85 | clk.advance(31 * time.Second) |
| 86 | a.sweepOnce() |
| 87 | waitForRestart(t, a, b.fakeBackend, host, 2) |
| 88 | if calls := b.calls(); len(calls) != 2 || calls[1] == nil || *calls[1] != fakeIdentity("v1") { |
| 89 | t.Fatalf("max_run recycle asked for %v; want the captured v1 build", calls) |
| 90 | } |
| 91 | } |
| 92 | |
| 93 | // Until plan 5 queues it, the drift check is the one path that moves a |
| 94 | // worker to a different installed build, and it reads the live link. |
| 95 | func TestWorkerConvergeStillMovesOntoTheInstalledBuild(t *testing.T) { |
| 96 | t.Parallel() |
| 97 | a, b, clk, _, ver := stagedActivator(t) |
| 98 | host := "bot.x.internal" |
| 99 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 100 | t.Fatalf("WakeWorker: %v", err) |
| 101 | } |
| 102 | ver.Store("v2") |
| 103 | a.sweepOnce() |
| 104 | clk.advance(ConvergeGrace + time.Minute) |
| 105 | a.sweepOnce() |
| 106 | waitFor(t, "the converge onto v2", func() bool { return a.entrySpawnVersion(host) == "v2" && !a.entryRespawning(host) }) |
| 107 | if calls := b.calls(); len(calls) != 2 || calls[1] != nil { |
| 108 | t.Fatalf("converge asked for %v; want the live link (nil)", calls) |
| 109 | } |
| 110 | if got := a.entryServing(host); got != fakeIdentity("v2") { |
| 111 | t.Fatalf("serving after converge = %+v, want v2", got) |
| 112 | } |
| 113 | } |
| 114 | |
| 115 | // Plan 5's hook: an admissible queued attempt takes the spawn. |
| 116 | func TestWorkerRecycleDefersToCoalesceRecycle(t *testing.T) { |
| 117 | t.Parallel() |
| 118 | a, b, _, _, _ := stagedActivator(t) |
| 119 | host := "bot.x.internal" |
| 120 | var asked []string |
| 121 | a.CoalesceRecycle = func(h string) bool { asked = append(asked, h); return true } |
| 122 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 123 | t.Fatalf("WakeWorker: %v", err) |
| 124 | } |
| 125 | b.lastWorker().kill() |
| 126 | a.sweepOnce() |
| 127 | if a.entryRespawning(host) || b.workerStarts() != 1 { |
| 128 | t.Fatalf("workerStarts = %d; the hook said a queued attempt makes the spawn", b.workerStarts()) |
| 129 | } |
| 130 | if len(asked) != 1 || asked[0] != host { |
| 131 | t.Fatalf("hook asked about %v, want [%s]", asked, host) |
| 132 | } |
| 133 | } |
| 134 | |
| 135 | // Review Focus 4, and spec §7.9: the report is display metadata; registry |
| 136 | // Version is root's install stamp. |
| 137 | func TestNoteWorkerReadyIsDisplayOnly(t *testing.T) { |
| 138 | t.Parallel() |
| 139 | a, b, _, reg := testActivator(t, time.Hour) |
| 140 | host := "bot.x.internal" |
| 141 | if err := reg.Upsert(testWorkerRoute(host)); err != nil { |
| 142 | t.Fatal(err) |
| 143 | } |
| 144 | if err := reg.SetVersions(host, "v1", ""); err != nil { |
| 145 | t.Fatal(err) |
| 146 | } |
| 147 | if err := a.WakeWorker(testWorkerRoute(host)); err != nil { |
| 148 | t.Fatalf("WakeWorker: %v", err) |
| 149 | } |
| 150 | noteReadyAsChild(t, a, b, host, "1.4.2-app-chosen") |
| 151 | if rt, _ := reg.Get(host); rt == nil || rt.Version != "v1" { |
| 152 | t.Fatalf("row = %+v; a worker's report must not overwrite root's stamp", rt) |
| 153 | } |
| 154 | if got := a.WorkerReportedVersion(host); got != "1.4.2-app-chosen" { |
| 155 | t.Fatalf("WorkerReportedVersion = %q", got) |
| 156 | } |
| 157 | a.NoteWorkerReady(host, "") |
| 158 | if rt, _ := reg.Get(host); rt == nil || rt.Version != "v1" { |
| 159 | t.Fatalf("row = %+v after an empty report", rt) |
| 160 | } |
| 161 | if got := a.WorkerReportedVersion("nobody.x.internal"); got != "" { |
| 162 | t.Fatalf("WorkerReportedVersion for no worker = %q", got) |
| 163 | } |
| 164 | } |
| 165 | |
| 166 | // Spec §7.9: no spawn path writes registry Version: not a wake, a crash |
| 167 | // recycle or a converge. |
| 168 | func TestWorkerSpawnsLeaveTheInstalledStampAlone(t *testing.T) { |
| 169 | t.Parallel() |
| 170 | a, b, clk, reg, ver := shippingActivatorWithRegistry(t, time.Hour) |
| 171 | host := "bot.x.internal" |
| 172 | rt := alwaysOnWorkerRoute(host) |
| 173 | if err := reg.Upsert(rt); err != nil { |
| 174 | t.Fatal(err) |
| 175 | } |
| 176 | if err := reg.SetVersions(host, "v1", "v0"); err != nil { |
| 177 | t.Fatal(err) |
| 178 | } |
| 179 | stamp := func(when string) { |
| 180 | t.Helper() |
| 181 | if got, _ := reg.Get(host); got == nil || got.Version != "v1" || got.PrevVersion != "v0" { |
| 182 | t.Fatalf("%s: row = %+v; want root's stamp v1 (prev v0) untouched", when, got) |
| 183 | } |
| 184 | } |
| 185 | if err := a.WakeWorker(rt); err != nil { |
| 186 | t.Fatalf("WakeWorker: %v", err) |
| 187 | } |
| 188 | stamp("after a wake") |
| 189 | noteReadyAsChild(t, a, b, host, "whatever-the-app-says") |
| 190 | stamp("after a ready report") |
| 191 | b.lastWorker().kill() |
| 192 | a.sweepOnce() |
| 193 | waitForRestart(t, a, b, host, 2) |
| 194 | stamp("after a crash recycle") |
| 195 | ver.Store("v2") |
| 196 | a.sweepOnce() |
| 197 | clk.advance(ConvergeGrace + time.Minute) |
| 198 | a.sweepOnce() |
| 199 | waitFor(t, "the converge onto v2", func() bool { return a.entrySpawnVersion(host) == "v2" && !a.entryRespawning(host) }) |
| 200 | stamp("after a converge") |
| 201 | } |
| 202 | |
| 203 | // Review Focus 4: an app-chosen report changes nothing about what recycles, |
| 204 | // and a new child's report starts out unknown. |
| 205 | func TestWorkerReportingAnAppChosenVersionStillRecyclesItsCapturedBuild(t *testing.T) { |
| 206 | t.Parallel() |
| 207 | a, b, _, _, _ := stagedActivator(t) |
| 208 | host := "bot.x.internal" |
| 209 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 210 | t.Fatalf("WakeWorker: %v", err) |
| 211 | } |
| 212 | noteReadyAsChild(t, a, b.fakeBackend, host, "latest") |
| 213 | b.lastWorker().kill() |
| 214 | a.sweepOnce() |
| 215 | waitForRestart(t, a, b.fakeBackend, host, 2) |
| 216 | if calls := b.calls(); len(calls) != 2 || calls[1] == nil || *calls[1] != fakeIdentity("v1") { |
| 217 | t.Fatalf("recycle asked for %v; want the captured v1, whatever the child said", calls) |
| 218 | } |
| 219 | if got := a.WorkerReportedVersion(host); got != "" { |
| 220 | t.Fatalf("WorkerReportedVersion = %q after a recycle; the new child has not reported", got) |
| 221 | } |
| 222 | } |
| 223 | |
| 224 | func (b *stagedFake) failNextStagedSpawn(err error) { |
| 225 | b.smu.Lock() |
| 226 | defer b.smu.Unlock() |
| 227 | b.failNextStaged = err |
| 228 | } |
| 229 | |
| 230 | // waitForSpawnCalls waits until the backend has been asked for n spawns and |
| 231 | // the respawn that asked has finished, failed or not. |
| 232 | func waitForSpawnCalls(t *testing.T, a *Service, b *stagedFake, host string, n int) { |
| 233 | t.Helper() |
| 234 | waitFor(t, fmt.Sprintf("spawn call #%d of %s", n, host), func() bool { |
| 235 | return len(b.calls()) == n && !a.entryRespawning(host) |
| 236 | }) |
| 237 | } |
| 238 | |
| 239 | // Ruling R17 (spec §4.2): a recycle the backend refuses because the captured |
| 240 | // copy changed stays refused. The worker stays down, it keeps its captured |
| 241 | // build, and the next crash restart asks for that same build again, never |
| 242 | // for what live/<host> names. Here live/<host> still names v1, so there is |
| 243 | // no drift that could admit a converge either. |
| 244 | func TestWorkerRecycleRefusedForAChangedCopyStaysRefused(t *testing.T) { |
| 245 | t.Parallel() |
| 246 | a, b, clk, _, _ := stagedActivator(t) |
| 247 | host := "bot.x.internal" |
| 248 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 249 | t.Fatalf("WakeWorker: %v", err) |
| 250 | } |
| 251 | changed := fmt.Errorf("worker %s: %w", host, release.ErrStagedCopyChanged) |
| 252 | b.failNextStagedSpawn(changed) |
| 253 | b.lastWorker().kill() |
| 254 | a.sweepOnce() |
| 255 | waitForSpawnCalls(t, a, b, host, 2) |
| 256 | if calls := b.calls(); calls[1] == nil || *calls[1] != fakeIdentity("v1") { |
| 257 | t.Fatalf("crash recycle asked for %v; want the captured v1 build", calls) |
| 258 | } |
| 259 | if got := a.entryServing(host); got != fakeIdentity("v1") { |
| 260 | t.Fatalf("serving after a refused recycle = %+v; want the captured v1 kept", got) |
| 261 | } |
| 262 | if b.workerStarts() != 1 || !b.lastWorker().Exited() { |
| 263 | t.Fatalf("workerStarts = %d; a refused recycle must leave the worker down", b.workerStarts()) |
| 264 | } |
| 265 | // The bytes still differ: the next attempt asks for the captured build |
| 266 | // again and is refused again. |
| 267 | b.failNextStagedSpawn(changed) |
| 268 | clk.advance(sidecarRestartMax + time.Second) |
| 269 | a.sweepOnce() |
| 270 | waitForSpawnCalls(t, a, b, host, 3) |
| 271 | if calls := b.calls(); calls[2] == nil || *calls[2] != fakeIdentity("v1") { |
| 272 | t.Fatalf("restart after the refusal asked for %v; want the captured v1 again, never the live link", calls[2]) |
| 273 | } |
| 274 | if b.workerStarts() != 1 || !b.lastWorker().Exited() { |
| 275 | t.Fatalf("workerStarts = %d; the refusal must stand while the bytes differ", b.workerStarts()) |
| 276 | } |
| 277 | if got := a.entryServing(host); got != fakeIdentity("v1") { |
| 278 | t.Fatalf("serving after a second refusal = %+v; want the captured v1 kept", got) |
| 279 | } |
| 280 | } |
| 281 | |
| 282 | // Any other spawn failure keeps the captured build too. |
| 283 | func TestWorkerRecycleFailingForAnotherReasonKeepsTheCapturedBuild(t *testing.T) { |
| 284 | t.Parallel() |
| 285 | a, b, clk, _, _ := stagedActivator(t) |
| 286 | host := "bot.x.internal" |
| 287 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 288 | t.Fatalf("WakeWorker: %v", err) |
| 289 | } |
| 290 | b.failNextStagedSpawn(errors.New("fork/exec: resource temporarily unavailable")) |
| 291 | b.lastWorker().kill() |
| 292 | a.sweepOnce() |
| 293 | waitForSpawnCalls(t, a, b, host, 2) |
| 294 | if got := a.entryServing(host); got != fakeIdentity("v1") { |
| 295 | t.Fatalf("serving after a failed recycle = %+v; want the captured v1 kept", got) |
| 296 | } |
| 297 | clk.advance(sidecarRestartMax + time.Second) |
| 298 | a.sweepOnce() |
| 299 | waitForSpawnCalls(t, a, b, host, 3) |
| 300 | if calls := b.calls(); calls[2] == nil || *calls[2] != fakeIdentity("v1") { |
| 301 | t.Fatalf("restart after the failure asked for %v; want the captured v1 again", calls[2]) |
| 302 | } |
| 303 | if b.workerStarts() != 2 { |
| 304 | t.Fatalf("workerStarts = %d; want the worker running again", b.workerStarts()) |
| 305 | } |
| 306 | } |
| 307 | |
| 308 | // Ruling R13: a max_run shorter than ConvergeGrace must not starve the |
| 309 | // converge. Recycles onto the captured build keep the drift clock, and the |
| 310 | // recycle that falls due past ConvergeGrace converges instead. |
| 311 | func TestWorkerMaxRunRecyclesConvergeOnceTheGraceHasPassed(t *testing.T) { |
| 312 | t.Parallel() |
| 313 | a, b, clk, _, ver := stagedActivator(t) |
| 314 | host := "bot.x.internal" |
| 315 | rt := alwaysOnWorkerRoute(host) |
| 316 | rt.MaxRunSeconds = 30 |
| 317 | if err := a.WakeWorker(rt); err != nil { |
| 318 | t.Fatalf("WakeWorker: %v", err) |
| 319 | } |
| 320 | ver.Store("v2") |
| 321 | converged := -1 |
| 322 | for i := 1; i <= 6 && converged < 0; i++ { |
| 323 | clk.advance(31 * time.Second) |
| 324 | a.sweepOnce() |
| 325 | waitForRestart(t, a, b.fakeBackend, host, i+1) |
| 326 | if b.calls()[i] == nil { |
| 327 | converged = i |
| 328 | } |
| 329 | } |
| 330 | calls := b.calls() |
| 331 | if converged < 0 { |
| 332 | t.Fatalf("spawns = %v; six max_run recycles never moved the worker onto v2", calls) |
| 333 | } |
| 334 | for i := 1; i < converged; i++ { |
| 335 | if calls[i] == nil || *calls[i] != fakeIdentity("v1") { |
| 336 | t.Fatalf("recycle %d asked for %v; inside the grace a recycle keeps the captured v1", i, calls[i]) |
| 337 | } |
| 338 | } |
| 339 | if elapsed := time.Duration(converged) * 31 * time.Second; elapsed < ConvergeGrace { |
| 340 | t.Fatalf("converged after %s; want the drift to have lasted ConvergeGrace first", elapsed) |
| 341 | } |
| 342 | if got := a.entryServing(host); got != fakeIdentity("v2") { |
| 343 | t.Fatalf("serving after the converge = %+v, want v2", got) |
| 344 | } |
| 345 | // No drift now: the next max_run recycles the captured v2. |
| 346 | clk.advance(31 * time.Second) |
| 347 | a.sweepOnce() |
| 348 | waitForRestart(t, a, b.fakeBackend, host, converged+2) |
| 349 | if c := b.calls()[converged+1]; c == nil || *c != fakeIdentity("v2") { |
| 350 | t.Fatalf("recycle after the converge asked for %v; want the captured v2", c) |
| 351 | } |
| 352 | } |
| 353 | |
| 354 | func (a *Service) entryStaleSince(host string) time.Time { |
| 355 | a.mu.Lock() |
| 356 | defer a.mu.Unlock() |
| 357 | if e := a.entries[host]; e != nil { |
| 358 | return e.staleSince |
| 359 | } |
| 360 | return time.Time{} |
| 361 | } |
| 362 | |
| 363 | // Ruling R17 (N2): a converge by recycle draws on the sweep's converge |
| 364 | // budget. Two workers whose drift is overdue crash together; one sweep |
| 365 | // converges at most maxConvergeCyclesPerSweep of them, the other recycles its |
| 366 | // captured build and keeps its drift age, and converges on a later sweep. |
| 367 | func TestWorkerRecycleConvergesShareTheSweepsConvergeBudget(t *testing.T) { |
| 368 | t.Parallel() |
| 369 | a, b, clk, _, ver := stagedActivator(t) |
| 370 | hosts := []string{"bot1.x.internal", "bot2.x.internal"} |
| 371 | for _, h := range hosts { |
| 372 | if err := a.WakeWorker(alwaysOnWorkerRoute(h)); err != nil { |
| 373 | t.Fatalf("WakeWorker %s: %v", h, err) |
| 374 | } |
| 375 | } |
| 376 | ver.Store("v2") |
| 377 | a.sweepOnce() // the drift is first seen here, by the sweep's own check |
| 378 | seen := a.entryStaleSince(hosts[0]) |
| 379 | if seen.IsZero() || !a.entryStaleSince(hosts[1]).Equal(seen) { |
| 380 | t.Fatalf("drift clocks = %v, %v; want both started by the sweep", seen, a.entryStaleSince(hosts[1])) |
| 381 | } |
| 382 | clk.advance(ConvergeGrace + time.Minute) |
| 383 | b.mu.Lock() |
| 384 | workers := append([]*fakeProc(nil), b.workers...) |
| 385 | b.mu.Unlock() |
| 386 | for _, w := range workers { |
| 387 | w.kill() |
| 388 | } |
| 389 | a.sweepOnce() |
| 390 | waitFor(t, "both crash restarts", func() bool { |
| 391 | return b.workerStarts() == 4 && !a.entryRespawning(hosts[0]) && !a.entryRespawning(hosts[1]) |
| 392 | }) |
| 393 | var converged, recycled []string |
| 394 | for _, h := range hosts { |
| 395 | switch a.entrySpawnVersion(h) { |
| 396 | case "v2": |
| 397 | converged = append(converged, h) |
| 398 | case "v1": |
| 399 | recycled = append(recycled, h) |
| 400 | } |
| 401 | } |
| 402 | if len(converged) != maxConvergeCyclesPerSweep || len(recycled) != len(hosts)-maxConvergeCyclesPerSweep { |
| 403 | t.Fatalf("in one sweep converged %v and recycled %v; want at most %d converge", converged, recycled, maxConvergeCyclesPerSweep) |
| 404 | } |
| 405 | if got := a.entryStaleSince(recycled[0]); !got.Equal(seen) { |
| 406 | t.Fatalf("drift clock of the worker that recycled = %v; want its age kept (%v)", got, seen) |
| 407 | } |
| 408 | a.sweepOnce() |
| 409 | waitFor(t, "the later converge", func() bool { |
| 410 | return a.entrySpawnVersion(recycled[0]) == "v2" && !a.entryRespawning(recycled[0]) |
| 411 | }) |
| 412 | } |
| 413 | |
| 414 | // Ruling R17 (N2): with plan 5's scheduler present (CoalesceRecycle set), a |
| 415 | // recycle never converges, even past ConvergeGrace: plan 5's admission owns |
| 416 | // build changes. A false answer recycles the captured build, and the drift |
| 417 | // keeps its age. |
| 418 | func TestWorkerRecycleNeverConvergesWhenTheSchedulerIsPresent(t *testing.T) { |
| 419 | t.Parallel() |
| 420 | a, b, clk, _, ver := stagedActivator(t) |
| 421 | host := "bot.x.internal" |
| 422 | a.CoalesceRecycle = func(string) bool { return false } |
| 423 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 424 | t.Fatalf("WakeWorker: %v", err) |
| 425 | } |
| 426 | ver.Store("v2") |
| 427 | b.lastWorker().kill() |
| 428 | a.sweepOnce() // the crash restart sees the drift first |
| 429 | waitForRestart(t, a, b.fakeBackend, host, 2) |
| 430 | seen := a.entryStaleSince(host) |
| 431 | if seen.IsZero() { |
| 432 | t.Fatal("the crash restart did not start the drift clock") |
| 433 | } |
| 434 | b.lastWorker().kill() |
| 435 | clk.advance(ConvergeGrace + time.Minute) |
| 436 | a.sweepOnce() |
| 437 | waitForRestart(t, a, b.fakeBackend, host, 3) |
| 438 | calls := b.calls() |
| 439 | for i := 1; i < len(calls); i++ { |
| 440 | if calls[i] == nil || *calls[i] != fakeIdentity("v1") { |
| 441 | t.Fatalf("recycle %d asked for %v; with the scheduler present a recycle keeps the captured v1", i, calls[i]) |
| 442 | } |
| 443 | } |
| 444 | if got := a.entryStaleSince(host); !got.Equal(seen) { |
| 445 | t.Fatalf("drift clock = %v after the recycle; want its age kept (%v)", got, seen) |
| 446 | } |
| 447 | } |
| 448 | |
| 449 | // Astra R3 §5 item 4: on a box whose scheduler (or its refusal) owns |
| 450 | // recycles (CoalesceRecycle set), a worker with no captured build is never |
| 451 | // respawned from the live link, which names whatever root installed last: |
| 452 | // a build no admission chose. A live child whose max_run falls due keeps |
| 453 | // running; a crashed one stays down, and the agent says so once for the |
| 454 | // host, not at every backoff. A release attempt (ReplaceWorker) or a wake |
| 455 | // is what starts it. Before the fix both respawned from live/<host>. |
| 456 | func TestAnArmedBoxNeverRecyclesAnUncapturedWorkerFromTheLiveLink(t *testing.T) { |
| 457 | const said = "has no captured build, and the release scheduler owns its recycles" |
| 458 | capture := func(t *testing.T) *syncLog { |
| 459 | logs := &syncLog{} |
| 460 | flags := log.Flags() |
| 461 | log.SetOutput(logs) |
| 462 | t.Cleanup(func() { log.SetOutput(os.Stderr); log.SetFlags(flags) }) |
| 463 | return logs |
| 464 | } |
| 465 | t.Run("crashed", func(t *testing.T) { |
| 466 | a, b, clk, _ := testActivator(t, time.Hour) // no staged capability: nothing is captured |
| 467 | host := "bot.x.internal" |
| 468 | a.CoalesceRecycle = func(string) bool { return false } |
| 469 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 470 | t.Fatalf("WakeWorker: %v", err) |
| 471 | } |
| 472 | if got := a.entryServing(host); got.StagedPath != "" { |
| 473 | t.Fatalf("serving %+v, want no captured build", got) |
| 474 | } |
| 475 | logs := capture(t) |
| 476 | b.lastWorker().kill() |
| 477 | for range 3 { |
| 478 | a.sweepOnce() |
| 479 | clk.advance(sidecarRestartMax + time.Second) |
| 480 | } |
| 481 | time.Sleep(100 * time.Millisecond) |
| 482 | if n := b.workerStarts(); n != 1 || a.entryRespawning(host) { |
| 483 | t.Fatalf("workerStarts = %d (respawning %v): an uncaptured worker was respawned from the live link", n, a.entryRespawning(host)) |
| 484 | } |
| 485 | if !a.Owns(host) { |
| 486 | t.Fatal("the entry went: it stays, down, for a release attempt or a wake to start") |
| 487 | } |
| 488 | if n := strings.Count(logs.String(), said); n != 1 { |
| 489 | t.Fatalf("said so %d times in %q, want once for the host", n, logs.String()) |
| 490 | } |
| 491 | }) |
| 492 | t.Run("max_run", func(t *testing.T) { |
| 493 | a, b, clk, _ := testActivator(t, time.Hour) |
| 494 | host := "bot.x.internal" |
| 495 | a.CoalesceRecycle = func(string) bool { return false } |
| 496 | rt := alwaysOnWorkerRoute(host) |
| 497 | rt.MaxRunSeconds = 30 |
| 498 | if err := a.WakeWorker(rt); err != nil { |
| 499 | t.Fatalf("WakeWorker: %v", err) |
| 500 | } |
| 501 | logs := capture(t) |
| 502 | for range 3 { |
| 503 | clk.advance(31 * time.Second) |
| 504 | a.sweepOnce() |
| 505 | } |
| 506 | time.Sleep(100 * time.Millisecond) |
| 507 | if n := b.workerStarts(); n != 1 || a.entryRespawning(host) { |
| 508 | t.Fatalf("workerStarts = %d (respawning %v): an uncaptured worker was recycled from the live link", n, a.entryRespawning(host)) |
| 509 | } |
| 510 | if b.lastWorker().Exited() { |
| 511 | t.Fatal("the running child was stopped by a recycle that could not respawn it") |
| 512 | } |
| 513 | if n := strings.Count(logs.String(), said); n != 1 { |
| 514 | t.Fatalf("said so %d times in %q, want once for the host", n, logs.String()) |
| 515 | } |
| 516 | }) |
| 517 | } |
| 518 | |
| 519 | // Ruling R13: an exited worker skips the sweep's drift check, so the crash |
| 520 | // restart measures the drift itself, and a crash loop converges once the |
| 521 | // grace has passed. |
| 522 | func TestWorkerCrashLoopConvergesOnceTheGraceHasPassed(t *testing.T) { |
| 523 | t.Parallel() |
| 524 | a, b, clk, _, ver := stagedActivator(t) |
| 525 | host := "bot.x.internal" |
| 526 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 527 | t.Fatalf("WakeWorker: %v", err) |
| 528 | } |
| 529 | ver.Store("v2") |
| 530 | converged := -1 |
| 531 | for i := 1; i <= 6 && converged < 0; i++ { |
| 532 | b.lastWorker().kill() |
| 533 | clk.advance(sidecarRestartMax + time.Second) |
| 534 | a.sweepOnce() |
| 535 | waitForRestart(t, a, b.fakeBackend, host, i+1) |
| 536 | if b.calls()[i] == nil { |
| 537 | converged = i |
| 538 | } |
| 539 | } |
| 540 | calls := b.calls() |
| 541 | if converged < 0 { |
| 542 | t.Fatalf("spawns = %v; a crash loop never moved the worker onto v2", calls) |
| 543 | } |
| 544 | if converged < 2 || calls[1] == nil || *calls[1] != fakeIdentity("v1") { |
| 545 | t.Fatalf("spawns = %v; the first crash after the install must recycle the captured v1 (the drift has only just been seen)", calls) |
| 546 | } |
| 547 | if a.entrySpawnVersion(host) != "v2" { |
| 548 | t.Fatalf("spawn version = %q after the converge, want v2", a.entrySpawnVersion(host)) |
| 549 | } |
| 550 | } |
| 551 | |
amadan