| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "os" |
| 7 | "path/filepath" |
| 8 | "slices" |
| 9 | "strings" |
| 10 | "sync" |
| 11 | "sync/atomic" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/registry" |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | ) |
| 18 | |
| 19 | func workerAttempt(cause Cause, mode SwapMode, onto release.StagedIdentity) Attempt { |
| 20 | return Attempt{Host: "bot.x.internal", Cause: cause, Build: onto.Version, StagedPath: onto.StagedPath, |
| 21 | Digest: onto.Digest, Demand: onto.Demand, Mode: mode, SchedRev: 1} |
| 22 | } |
| 23 | |
| 24 | func one(string) uint64 { return 1 } |
| 25 | |
| 26 | // waitForWorkerStarts waits until the backend has made n worker children: |
| 27 | // a spawn call is recorded before its exec, so b.calls() alone runs ahead |
| 28 | // of the child a test wants to act on. |
| 29 | func waitForWorkerStarts(t *testing.T, b *stagedFake, n int) { |
| 30 | t.Helper() |
| 31 | waitFor(t, "a worker child", func() bool { return b.workerStarts() == n }) |
| 32 | } |
| 33 | |
| 34 | // Spec §7.9: the attempt respawns on the SAME entry. The lease is never |
| 35 | // released or re-taken, and the new child execs the attempt's staged build. |
| 36 | func TestReplaceWorkerKeepsTheEntryAndItsLease(t *testing.T) { |
| 37 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 38 | a, b, _, _, _ := stagedActivator(t) |
| 39 | host := "bot.x.internal" |
| 40 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 41 | t.Fatalf("WakeWorker: %v", err) |
| 42 | } |
| 43 | acquires, _, _ := b.counts() |
| 44 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one) |
| 45 | if res.Outcome != SwapDone || res.Gen.Build != "v2" || res.Gen.Digest != fakeIdentity("v2").Digest || !res.Gen.Worker { |
| 46 | t.Fatalf("ReplaceWorker = %+v", res) |
| 47 | } |
| 48 | if got, _, _ := b.counts(); got != acquires { |
| 49 | t.Fatalf("lease acquires %d -> %d: a socketless attempt must keep the lease (spec §7.9)", acquires, got) |
| 50 | } |
| 51 | if calls := b.calls(); len(calls) != 2 || calls[1] == nil || *calls[1] != fakeIdentity("v2") { |
| 52 | t.Fatalf("spawns %v, want the wake, then v2 by its staged path", calls) |
| 53 | } |
| 54 | if got := a.entryServing(host); got != fakeIdentity("v2") { |
| 55 | t.Fatalf("serving %+v", got) |
| 56 | } |
| 57 | if _, ok := a.Current(host); !ok { |
| 58 | t.Fatal("Current hides the worker after its attempt finished") |
| 59 | } |
| 60 | } |
| 61 | |
| 62 | // Spec §6.5: a failed release in overlap mode respawns the good build in |
| 63 | // its place, on the same entry, and reports the failure. |
| 64 | func TestReplaceWorkerFallsBackToTheGoodBuild(t *testing.T) { |
| 65 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond) |
| 66 | a, b, _, _, _ := stagedActivator(t) |
| 67 | host := "bot.x.internal" |
| 68 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 69 | t.Fatal(err) |
| 70 | } |
| 71 | good := fakeIdentity("v1") |
| 72 | go func() { |
| 73 | waitForWorkerStarts(t, b, 2) |
| 74 | time.Sleep(50 * time.Millisecond) |
| 75 | b.lastWorker().crash() |
| 76 | }() |
| 77 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), &good, one) |
| 78 | if res.Outcome != SwapFailed || res.Gen.Build != "v1" { |
| 79 | t.Fatalf("ReplaceWorker = %+v, want failed with the good v1 serving", res) |
| 80 | } |
| 81 | if calls := b.calls(); len(calls) != 3 || *calls[2] != good { |
| 82 | t.Fatalf("spawns %v, want the good build last", calls) |
| 83 | } |
| 84 | if got := a.entryServing(host); got != good { |
| 85 | t.Fatalf("serving %+v, want the good build", got) |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | // Carried to Task 14 (from Task 11): a worker crash inside ReplaceWorker's |
| 90 | // observation is the attempt's to read. The sweep sees the dead child first |
| 91 | // and asks CoalesceRecycle, which the scheduler answers true while the |
| 92 | // host's attempt runs (rollout.Manager.CoalesceRecycle). So the sweep's |
| 93 | // crash restart does not respawn the attempt's build behind it, and the |
| 94 | // observation reads the crash as the build's failure and falls back to the |
| 95 | // good build. Without the hook the sweep respawns v2 and the attempt never |
| 96 | // sees its child fail. |
| 97 | func TestACrashInsideAReplaceWorkerObservationIsCoalesced(t *testing.T) { |
| 98 | // A 2 s window polls every 250 ms (settlePoll), so the sweep below runs |
| 99 | // well before the observation's next look at the child. |
| 100 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second) |
| 101 | a, b, _, _, _ := stagedActivator(t) |
| 102 | host := "bot.x.internal" |
| 103 | var running atomic.Bool |
| 104 | var asked atomic.Int32 |
| 105 | a.CoalesceRecycle = func(h string) bool { |
| 106 | if h == host { |
| 107 | asked.Add(1) |
| 108 | } |
| 109 | return running.Load() |
| 110 | } |
| 111 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 112 | t.Fatal(err) |
| 113 | } |
| 114 | good := fakeIdentity("v1") |
| 115 | go func() { |
| 116 | waitForWorkerStarts(t, b, 2) |
| 117 | time.Sleep(100 * time.Millisecond) |
| 118 | b.lastWorker().crash() |
| 119 | a.sweepOnce() // the crash restart's arm, inside the observation |
| 120 | }() |
| 121 | running.Store(true) |
| 122 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), &good, one) |
| 123 | running.Store(false) |
| 124 | if asked.Load() == 0 { |
| 125 | t.Fatal("the sweep never asked CoalesceRecycle about the crashed child") |
| 126 | } |
| 127 | if res.Outcome != SwapFailed || res.Gen.Build != "v1" { |
| 128 | t.Fatalf("ReplaceWorker = %+v, want failed with the good v1 serving", res) |
| 129 | } |
| 130 | calls := b.calls() |
| 131 | if len(calls) != 3 || *calls[1] != fakeIdentity("v2") || *calls[2] != good { |
| 132 | t.Fatalf("spawns %v, want the wake, v2 once, then the good build: the sweep respawned behind the attempt", calls) |
| 133 | } |
| 134 | if got := a.entryServing(host); got != good { |
| 135 | t.Fatalf("serving %+v, want the good build", got) |
| 136 | } |
| 137 | } |
| 138 | |
| 139 | // Spec §6.4, §8.3: a hold build that ran never falls back. The instance |
| 140 | // stays down. |
| 141 | func TestReplaceWorkerInHoldModeStaysDown(t *testing.T) { |
| 142 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond) |
| 143 | a, b, _, _, _ := stagedActivator(t) |
| 144 | host := "bot.x.internal" |
| 145 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 146 | t.Fatal(err) |
| 147 | } |
| 148 | good := fakeIdentity("v1") |
| 149 | go func() { |
| 150 | waitForWorkerStarts(t, b, 2) |
| 151 | time.Sleep(50 * time.Millisecond) |
| 152 | b.lastWorker().crash() |
| 153 | }() |
| 154 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeHold, fakeIdentity("v2")), &good, one) |
| 155 | if res.Outcome != SwapDown { |
| 156 | t.Fatalf("ReplaceWorker = %+v, want down", res) |
| 157 | } |
| 158 | if len(b.calls()) != 2 { |
| 159 | t.Fatalf("spawns %v: a hold build fell back", b.calls()) |
| 160 | } |
| 161 | waitFor(t, "the entry to stop", func() bool { return !a.Owns(host) }) |
| 162 | } |
| 163 | |
| 164 | // Spec §6.5: a failed restart respawns the build it replaced, and writes |
| 165 | // nothing against it. |
| 166 | func TestReplaceWorkerRestartFailureRespawnsTheServingBuild(t *testing.T) { |
| 167 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond) |
| 168 | a, b, _, _, _ := stagedActivator(t) |
| 169 | host := "bot.x.internal" |
| 170 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 171 | t.Fatal(err) |
| 172 | } |
| 173 | serving := a.entryServing(host) |
| 174 | go func() { |
| 175 | waitForWorkerStarts(t, b, 2) |
| 176 | time.Sleep(50 * time.Millisecond) |
| 177 | b.lastWorker().crash() |
| 178 | }() |
| 179 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRestart, ModeOverlap, serving), nil, one) |
| 180 | if res.Outcome != SwapFailed { |
| 181 | t.Fatalf("ReplaceWorker = %+v", res) |
| 182 | } |
| 183 | waitFor(t, "the serving build back", func() bool { return len(b.calls()) == 3 && !a.entryRespawning(host) }) |
| 184 | if calls := b.calls(); *calls[2] != serving { |
| 185 | t.Fatalf("respawned %+v, want the serving build", calls[2]) |
| 186 | } |
| 187 | } |
| 188 | |
| 189 | // Spec §6.5: a clean exit inside the 10 s is a one-shot's success. |
| 190 | func TestReplaceWorkerOneShotCleanExitIsReady(t *testing.T) { |
| 191 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond) |
| 192 | a, b, _, _, _ := stagedActivator(t) |
| 193 | if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| 194 | t.Fatal(err) |
| 195 | } |
| 196 | go func() { |
| 197 | waitForWorkerStarts(t, b, 2) |
| 198 | b.lastWorker().exit() |
| 199 | }() |
| 200 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one) |
| 201 | if res.Outcome != SwapDone { |
| 202 | t.Fatalf("ReplaceWorker = %+v, want done", res) |
| 203 | } |
| 204 | if !res.CleanExit { |
| 205 | t.Fatalf("ReplaceWorker = %+v: a one-shot that exited cleanly must say so, so the scheduler records its build good (spec 6.5, 8.5)", res) |
| 206 | } |
| 207 | } |
| 208 | |
| 209 | // Spec §6.5, §8.5: only a build that exited cleanly is a one-shot's success. |
| 210 | // One still running after the settle window is up and serving (done, but |
| 211 | // the scheduler records it good when the activator acknowledges it, not |
| 212 | // here), and a crash is a failure. |
| 213 | func TestReplaceWorkerRunningChildIsNotACleanExit(t *testing.T) { |
| 214 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 215 | a, _, _, _, _ := stagedActivator(t) |
| 216 | if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| 217 | t.Fatal(err) |
| 218 | } |
| 219 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one) |
| 220 | if res.Outcome != SwapDone || res.CleanExit { |
| 221 | t.Fatalf("ReplaceWorker = %+v, want done and not a clean exit", res) |
| 222 | } |
| 223 | } |
| 224 | |
| 225 | // Astra R3 flag triage T10: a first-wake worker whose build fails and whose |
| 226 | // fallback boots only after the ordinary wake bound is a success for |
| 227 | // WakeWorker, as it is for a request: the wait extends once to arrival + |
| 228 | // firstWakeBudget (spec 8.2). Without the extension the caller was told |
| 229 | // "timed out" for a wake that went on to succeed. A wake that is not a |
| 230 | // first wake still gives up at the ordinary bound. |
| 231 | func TestWakeWorkerWaitsOutAFirstWakeFallback(t *testing.T) { |
| 232 | old := wakeWaitBound |
| 233 | wakeWaitBound = 100 * time.Millisecond |
| 234 | t.Cleanup(func() { wakeWaitBound = old }) |
| 235 | for _, tc := range []struct { |
| 236 | name string |
| 237 | firstWake bool |
| 238 | wantErr bool |
| 239 | }{ |
| 240 | {"first wake", true, false}, |
| 241 | {"not a first wake", false, true}, |
| 242 | } { |
| 243 | t.Run(tc.name, func(t *testing.T) { |
| 244 | shrinkFirstWake(t, 900*time.Millisecond, 200*time.Millisecond, 50*time.Millisecond) |
| 245 | a, b, _, _, _ := stagedActivator(t) |
| 246 | a.gens = true |
| 247 | a.backend = &plannedHostKill{stagedFake: b} // the failed child's host cgroup proves empty |
| 248 | dir := t.TempDir() |
| 249 | good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| 250 | var starts atomic.Int32 |
| 251 | a.Planner = func(context.Context, registry.Route) WakePlan { |
| 252 | if !tc.firstWake { |
| 253 | // An ordinary wake of the good build; its start is slow. |
| 254 | return WakePlan{Boot: good, Mode: ModeOverlap} |
| 255 | } |
| 256 | return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good} |
| 257 | } |
| 258 | b.mu.Lock() |
| 259 | b.workerStartHook = func(registry.Route) { |
| 260 | n := starts.Add(1) |
| 261 | switch { |
| 262 | case tc.firstWake && n == 1: |
| 263 | // The failing build dies inside its settle window. |
| 264 | time.AfterFunc(20*time.Millisecond, func() { b.lastWorker().crash() }) |
| 265 | case tc.firstWake && n == 2, !tc.firstWake && n == 1: |
| 266 | time.Sleep(250 * time.Millisecond) // the good build boots slowly |
| 267 | } |
| 268 | } |
| 269 | b.mu.Unlock() |
| 270 | err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")) |
| 271 | if tc.wantErr { |
| 272 | if !errors.Is(err, errWakeTimeout) { |
| 273 | t.Fatalf("WakeWorker = %v, want errWakeTimeout at the ordinary bound", err) |
| 274 | } |
| 275 | return |
| 276 | } |
| 277 | if err != nil { |
| 278 | t.Fatalf("WakeWorker = %v: a first wake's successful fallback was reported as a failure", err) |
| 279 | } |
| 280 | }) |
| 281 | } |
| 282 | } |
| 283 | |
| 284 | // Spec §6.3, §7.4, §8.1: the attempt's checks run again at the actual |
| 285 | // spawn, after the old child has retired, and again before its result is |
| 286 | // accepted. A newer release (or a sibling's failure, whose halt bumps the |
| 287 | // revision) that lands during the retire means the attempt's build never |
| 288 | // execs: the build it replaced comes back instead. One that lands during |
| 289 | // the observation means the result is abandoned, never done, so the build |
| 290 | // is not recorded good. Checked only before the retire, both slip through. |
| 291 | func TestReplaceWorkerReChecksAtTheSpawnAndBeforeTheResult(t *testing.T) { |
| 292 | t.Run("superseded while the old child retires", func(t *testing.T) { |
| 293 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 294 | a, b, _, _, _ := stagedActivator(t) |
| 295 | host := "bot.x.internal" |
| 296 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 297 | t.Fatal(err) |
| 298 | } |
| 299 | gate := make(chan struct{}) |
| 300 | b.lastWorker().setStopGate(gate) |
| 301 | var rev atomic.Uint64 |
| 302 | rev.Store(1) |
| 303 | results := make(chan SwapResult, 1) |
| 304 | go func() { |
| 305 | results <- a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, |
| 306 | func(string) uint64 { return rev.Load() }) |
| 307 | }() |
| 308 | waitFor(t, "the retire to begin", func() bool { return a.entryRespawning(host) }) |
| 309 | rev.Store(2) // a newer release lands while the old child drains |
| 310 | close(gate) |
| 311 | res := <-results |
| 312 | if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) { |
| 313 | t.Fatalf("ReplaceWorker = %+v, want abandoned: superseded", res) |
| 314 | } |
| 315 | for _, c := range b.calls() { |
| 316 | if c != nil && *c == fakeIdentity("v2") { |
| 317 | t.Fatalf("spawns %v: the superseded build was exec'd", b.calls()) |
| 318 | } |
| 319 | } |
| 320 | waitFor(t, "a child back on the entry", func() bool { return len(b.calls()) == 2 && !a.entryRespawning(host) }) |
| 321 | if got := a.entryServing(host); got != fakeIdentity("v1") { |
| 322 | t.Fatalf("serving %+v, want the build the attempt replaced", got) |
| 323 | } |
| 324 | }) |
| 325 | t.Run("superseded during the observation", func(t *testing.T) { |
| 326 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond) |
| 327 | a, b, _, _, _ := stagedActivator(t) |
| 328 | host := "bot.x.internal" |
| 329 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 330 | t.Fatal(err) |
| 331 | } |
| 332 | var rev atomic.Uint64 |
| 333 | rev.Store(1) |
| 334 | go func() { |
| 335 | waitForWorkerStarts(t, b, 2) |
| 336 | rev.Store(2) |
| 337 | }() |
| 338 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, |
| 339 | func(string) uint64 { return rev.Load() }) |
| 340 | if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) { |
| 341 | t.Fatalf("ReplaceWorker = %+v, want abandoned: a superseded result is never done", res) |
| 342 | } |
| 343 | }) |
| 344 | } |
| 345 | |
| 346 | // PF11: the attempt's checks also run in main's exec-time gate, inside |
| 347 | // startWorker's before-exec hook, after the preparation (here the config |
| 348 | // admission) and the publication. A newer release that lands after the |
| 349 | // respawn's first check but before the exec means the attempt's build never |
| 350 | // execs, and the build it replaced comes back. |
| 351 | func TestReplaceWorkerReChecksAtTheExecItself(t *testing.T) { |
| 352 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 353 | a, b, _, _, _ := stagedActivator(t) |
| 354 | host := "bot.x.internal" |
| 355 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 356 | t.Fatal(err) |
| 357 | } |
| 358 | var rev atomic.Uint64 |
| 359 | rev.Store(1) |
| 360 | b.mu.Lock() |
| 361 | b.admitHook = func() { rev.Store(2) } // lands during the respawn's preparation |
| 362 | b.mu.Unlock() |
| 363 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, |
| 364 | func(string) uint64 { return rev.Load() }) |
| 365 | if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) { |
| 366 | t.Fatalf("ReplaceWorker = %+v, want abandoned: superseded at the exec", res) |
| 367 | } |
| 368 | waitFor(t, "the restore", func() bool { return b.workerStarts() == 2 && !a.entryRespawning(host) }) |
| 369 | if got := a.entryServing(host); got != fakeIdentity("v1") { |
| 370 | t.Fatalf("serving %+v, want the build the attempt replaced (v2 must never have exec'd)", got) |
| 371 | } |
| 372 | } |
| 373 | |
| 374 | // PF11: and once more before the child is installed. A newer release that |
| 375 | // lands while the attempt's child is being exec'd sheds that child; the |
| 376 | // build it replaced comes back, and the result is abandoned, never done. |
| 377 | func TestReplaceWorkerReChecksBeforeInstalling(t *testing.T) { |
| 378 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 379 | a, b, _, _, _ := stagedActivator(t) |
| 380 | host := "bot.x.internal" |
| 381 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 382 | t.Fatal(err) |
| 383 | } |
| 384 | var rev atomic.Uint64 |
| 385 | rev.Store(1) |
| 386 | var hooked atomic.Bool |
| 387 | b.mu.Lock() |
| 388 | b.workerStartHook = func(registry.Route) { |
| 389 | if hooked.CompareAndSwap(false, true) { |
| 390 | rev.Store(2) // lands after the exec gate, before the install check |
| 391 | } |
| 392 | } |
| 393 | b.mu.Unlock() |
| 394 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, |
| 395 | func(string) uint64 { return rev.Load() }) |
| 396 | if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) { |
| 397 | t.Fatalf("ReplaceWorker = %+v, want abandoned: superseded before the install", res) |
| 398 | } |
| 399 | waitFor(t, "the restore", func() bool { return b.workerStarts() == 3 && !a.entryRespawning(host) }) |
| 400 | b.mu.Lock() |
| 401 | shed := b.workers[1] |
| 402 | b.mu.Unlock() |
| 403 | if !shed.isStopped() { |
| 404 | t.Fatal("the superseded attempt's child was left running") |
| 405 | } |
| 406 | if got := a.entryServing(host); got != fakeIdentity("v1") { |
| 407 | t.Fatalf("serving %+v, want the build the attempt replaced", got) |
| 408 | } |
| 409 | } |
| 410 | |
| 411 | // PF11 (plan 3's termination proof is never dropped): a respawn whose old |
| 412 | // child is not proven gone execs nothing, the attempt's build and a restore |
| 413 | // alike, and the attempt is abandoned. |
| 414 | func TestReplaceWorkerNeverExecsBesideAnUnprovenOldChild(t *testing.T) { |
| 415 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 416 | a, b, _, _, _ := stagedActivator(t) |
| 417 | host := "bot.x.internal" |
| 418 | a.gens = true |
| 419 | hk := &plannedHostKill{stagedFake: b} |
| 420 | a.backend = hk |
| 421 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 422 | t.Fatal(err) |
| 423 | } |
| 424 | hk.bad.Store(true) // the host cgroup still holds a process |
| 425 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one) |
| 426 | if res.Outcome != SwapAbandoned { |
| 427 | t.Fatalf("ReplaceWorker = %+v, want abandoned", res) |
| 428 | } |
| 429 | waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) }) |
| 430 | time.Sleep(50 * time.Millisecond) |
| 431 | if n := b.workerStarts(); n != 1 { |
| 432 | t.Fatalf("%d worker children: something exec'd beside an unproven old child", n) |
| 433 | } |
| 434 | } |
| 435 | |
| 436 | // Spec §7.4: a respawn whose gate finds the host no longer owned (here the |
| 437 | // lease lost while the old child drained) spawns nothing at all, the |
| 438 | // replaced build included. Turning every refusal into a restore ran the old |
| 439 | // build against a database the box no longer owned. |
| 440 | func TestReplaceWorkerWithoutOwnershipSpawnsNothing(t *testing.T) { |
| 441 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 442 | a, b, _, _, _ := stagedActivator(t) |
| 443 | host := "bot.x.internal" |
| 444 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 445 | t.Fatal(err) |
| 446 | } |
| 447 | gate := make(chan struct{}) |
| 448 | b.lastWorker().setStopGate(gate) |
| 449 | results := make(chan SwapResult, 1) |
| 450 | go func() { |
| 451 | results <- a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one) |
| 452 | }() |
| 453 | waitFor(t, "the retire to begin", func() bool { return a.entryRespawning(host) }) |
| 454 | b.lastRep().setErr(errors.New("lease renewal failed")) // ownership lost mid-retire |
| 455 | close(gate) |
| 456 | if res := <-results; res.Outcome != SwapAbandoned { |
| 457 | t.Fatalf("ReplaceWorker = %+v, want abandoned", res) |
| 458 | } |
| 459 | waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(host) }) |
| 460 | if calls := b.calls(); len(calls) != 1 { |
| 461 | t.Fatalf("spawns %v: something exec'd after the lease was lost", calls) |
| 462 | } |
| 463 | } |
| 464 | |
| 465 | // A superseded attempt touches nothing. |
| 466 | func TestReplaceWorkerSupersededSpawnsNothing(t *testing.T) { |
| 467 | t.Parallel() |
| 468 | a, b, _, _, _ := stagedActivator(t) |
| 469 | if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| 470 | t.Fatal(err) |
| 471 | } |
| 472 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, func(string) uint64 { return 2 }) |
| 473 | if res.Outcome != SwapAbandoned || len(b.calls()) != 1 { |
| 474 | t.Fatalf("ReplaceWorker = %+v, spawns %v", res, b.calls()) |
| 475 | } |
| 476 | } |
| 477 | |
| 478 | // PF9: an attempt whose build could not be published before its exec |
| 479 | // (ErrBuildNotPublished) ran nothing of the build. It is abandonment, a |
| 480 | // staging refusal, never a failure of the build. |
| 481 | func TestReplaceWorkerRefusedForPublicationIsAbandoned(t *testing.T) { |
| 482 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 483 | a, b, _, _, _ := stagedActivator(t) |
| 484 | host := "bot.x.internal" |
| 485 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 486 | t.Fatal(err) |
| 487 | } |
| 488 | blocker := filepath.Join(t.TempDir(), "not-a-dir") |
| 489 | if err := os.WriteFile(blocker, nil, 0o644); err != nil { |
| 490 | t.Fatal(err) |
| 491 | } |
| 492 | a.mu.Lock() |
| 493 | a.GenerationsFile = filepath.Join(blocker, "generations.json") // unwritable from here on |
| 494 | a.mu.Unlock() |
| 495 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one) |
| 496 | if res.Outcome != SwapAbandoned { |
| 497 | t.Fatalf("ReplaceWorker = %+v, want abandoned: an unpublished build never ran", res) |
| 498 | } |
| 499 | waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) }) |
| 500 | if n := b.workerStarts(); n != 1 { |
| 501 | t.Fatalf("%d worker children: an unpublished build was exec'd", n) |
| 502 | } |
| 503 | } |
| 504 | |
| 505 | // Carried from Task 9: e.unexpectedExit is never cleared by the arms that |
| 506 | // set it. A stale mark from an earlier crash must not turn an intentional |
| 507 | // stop during the attempt's observation into a verdict: the stop cancels |
| 508 | // the observation, and the attempt is abandoned, never done. |
| 509 | func TestReplaceWorkerAnIntentionalStopDuringTheObservationIsNoVerdict(t *testing.T) { |
| 510 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second) |
| 511 | a, b, _, _, _ := stagedActivator(t) |
| 512 | host := "bot.x.internal" |
| 513 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 514 | t.Fatal(err) |
| 515 | } |
| 516 | a.mu.Lock() |
| 517 | a.entries[host].unexpectedExit = true // left by an earlier crash restart |
| 518 | a.mu.Unlock() |
| 519 | go func() { |
| 520 | waitForWorkerStarts(t, b, 2) |
| 521 | waitFor(t, "the observation", func() bool { |
| 522 | a.mu.Lock() |
| 523 | defer a.mu.Unlock() |
| 524 | e := a.entries[host] |
| 525 | return e != nil && e.firstWatch != nil |
| 526 | }) |
| 527 | a.mu.Lock() |
| 528 | a.beginStopLocked(a.entries[host], StopIdle) // an operator's stop |
| 529 | a.mu.Unlock() |
| 530 | }() |
| 531 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one) |
| 532 | if res.Outcome != SwapAbandoned { |
| 533 | t.Fatalf("ReplaceWorker = %+v, want abandoned: an intentional stop records nothing", res) |
| 534 | } |
| 535 | } |
| 536 | |
| 537 | // Spec §7.9: with CARLOS_SWAP unset (no generations, no rollout manager), a |
| 538 | // worker's captured serving identity is still published on its wake and on |
| 539 | // every respawn, so `carlos sidecars ls` shows what runs. Only the rollout |
| 540 | // manager wrote it, and a flag-off box showed "unknown", or a stale build |
| 541 | // after the flag was turned off. |
| 542 | func TestAWorkersServingIdentityIsPublishedWithTheFlagOff(t *testing.T) { |
| 543 | t.Parallel() |
| 544 | a, b, _, _, _ := stagedActivator(t) |
| 545 | var mu sync.Mutex |
| 546 | var got []string |
| 547 | a.OnWorkerServing = func(host string, id release.StagedIdentity) { |
| 548 | mu.Lock() |
| 549 | got = append(got, host+"="+id.Version) |
| 550 | mu.Unlock() |
| 551 | } |
| 552 | published := func() int { mu.Lock(); defer mu.Unlock(); return len(got) } |
| 553 | host := "bot.x.internal" |
| 554 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 555 | t.Fatal(err) |
| 556 | } |
| 557 | waitFor(t, "the wake's identity", func() bool { return published() == 1 }) |
| 558 | b.lastWorker().crash() |
| 559 | a.sweepOnce() // the always-on worker respawns |
| 560 | waitFor(t, "the respawn's identity", func() bool { return published() == 2 }) |
| 561 | mu.Lock() |
| 562 | defer mu.Unlock() |
| 563 | if want := host + "=" + a.entryServing(host).Version; got[1] != want { |
| 564 | t.Fatalf("published %v, want the respawned child's %q last", got, want) |
| 565 | } |
| 566 | } |
| 567 | |
| 568 | // Spec §7.9: a delayed publication of an older worker incarnation never |
| 569 | // overwrites a newer one. Here the wake's publication (incarnation 1) is |
| 570 | // replayed after the respawn published incarnation 2: it is dropped. An |
| 571 | // unconditional write left the older version on display indefinitely. |
| 572 | func TestAnOlderWorkerPublicationIsDropped(t *testing.T) { |
| 573 | t.Parallel() |
| 574 | a, b, _, _, _ := stagedActivator(t) |
| 575 | var mu sync.Mutex |
| 576 | var got []string |
| 577 | a.OnWorkerServing = func(_ string, id release.StagedIdentity) { |
| 578 | mu.Lock() |
| 579 | got = append(got, id.Version) |
| 580 | mu.Unlock() |
| 581 | } |
| 582 | host := "bot.x.internal" |
| 583 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 584 | t.Fatal(err) |
| 585 | } |
| 586 | a.mu.Lock() |
| 587 | e := a.entries[host] |
| 588 | firstID, firstInc := e.workerServing, e.workerInc |
| 589 | a.mu.Unlock() |
| 590 | b.lastWorker().crash() |
| 591 | a.sweepOnce() |
| 592 | waitFor(t, "the respawn's publication", func() bool { mu.Lock(); defer mu.Unlock(); return len(got) == 2 }) |
| 593 | a.publishWorkerServing(e, firstID, firstInc) // the older one, delayed |
| 594 | mu.Lock() |
| 595 | defer mu.Unlock() |
| 596 | if len(got) != 2 { |
| 597 | t.Fatalf("published %v: an older incarnation's publication was delivered after a newer one", got) |
| 598 | } |
| 599 | } |
| 600 | |
| 601 | // Spec §7.9 (P5 review finding 8): a child exec'd from an external live-link |
| 602 | // target has no captured version. Its publication still goes out, empty, |
| 603 | // so the row stops naming the build the previous child ran. Skipping empty |
| 604 | // versions left ServingVersion=v1 on display indefinitely. |
| 605 | func TestAnUnknownVersionClearsTheServingVersion(t *testing.T) { |
| 606 | t.Parallel() |
| 607 | a, _, _, _, _ := stagedActivator(t) |
| 608 | var mu sync.Mutex |
| 609 | var got []string |
| 610 | a.OnWorkerServing = func(_ string, id release.StagedIdentity) { |
| 611 | mu.Lock() |
| 612 | got = append(got, id.Version) |
| 613 | mu.Unlock() |
| 614 | } |
| 615 | host := "bot.x.internal" |
| 616 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 617 | t.Fatal(err) |
| 618 | } |
| 619 | waitFor(t, "the wake's identity", func() bool { mu.Lock(); defer mu.Unlock(); return len(got) == 1 }) |
| 620 | a.mu.Lock() |
| 621 | e := a.entries[host] |
| 622 | e.workerInc++ // a replacement exec'd from an external live-link target |
| 623 | inc := e.workerInc |
| 624 | a.mu.Unlock() |
| 625 | a.publishWorkerServing(e, release.StagedIdentity{StagedPath: "/srv/external/app"}, inc) |
| 626 | mu.Lock() |
| 627 | defer mu.Unlock() |
| 628 | if len(got) != 2 || got[1] != "" { |
| 629 | t.Fatalf("published %q, want the empty version last: the old one must be cleared", got) |
| 630 | } |
| 631 | } |
| 632 | |
| 633 | // Spec §7.4 (P5 review finding 1): every respawn passes the ownership check |
| 634 | // at its exec, a fallback included. Here B execs, the lease is lost, and |
| 635 | // then B crashes: the good build is NOT respawned. Without the check the |
| 636 | // fallback ran A against a database the box no longer owned. |
| 637 | func TestAWorkerFallbackWithoutOwnershipSpawnsNothing(t *testing.T) { |
| 638 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond) |
| 639 | a, b, _, _, _ := stagedActivator(t) |
| 640 | host := "bot.x.internal" |
| 641 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 642 | t.Fatal(err) |
| 643 | } |
| 644 | good := fakeIdentity("v1") |
| 645 | go func() { |
| 646 | waitForWorkerStarts(t, b, 2) |
| 647 | b.lastRep().setErr(errors.New("lease renewal failed")) // ownership lost after B's exec |
| 648 | time.Sleep(50 * time.Millisecond) |
| 649 | b.lastWorker().crash() |
| 650 | }() |
| 651 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), &good, one) |
| 652 | if res.Outcome != SwapFailed || res.Gen.Build != "" { |
| 653 | t.Fatalf("ReplaceWorker = %+v, want failed with nothing serving", res) |
| 654 | } |
| 655 | if calls := b.calls(); len(calls) != 2 { |
| 656 | t.Fatalf("spawns %v: the fallback exec'd after the lease was lost", calls) |
| 657 | } |
| 658 | } |
| 659 | |
| 660 | // Spec §6.5, §8.1 (P5 review finding 5): a restart counts only once a NEW |
| 661 | // child is installed. Here the restart's exec fails after the old child |
| 662 | // stopped cleanly: the entry still holds that child, with the same |
| 663 | // artifact, and its clean exit read as a one-shot's success, so the |
| 664 | // scheduler acknowledged config that never ran. |
| 665 | func TestAFailedRestartExecIsNeverDone(t *testing.T) { |
| 666 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 667 | a, b, _, _, _ := stagedActivator(t) |
| 668 | host := "bot.x.internal" |
| 669 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 670 | t.Fatal(err) |
| 671 | } |
| 672 | serving := a.entryServing(host) |
| 673 | b.mu.Lock() |
| 674 | b.workerErr = errors.New("exec: text file busy") |
| 675 | b.mu.Unlock() |
| 676 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRestart, ModeOverlap, serving), nil, one) |
| 677 | if res.Outcome == SwapDone { |
| 678 | t.Fatalf("ReplaceWorker = %+v: a restart whose child never ran was done", res) |
| 679 | } |
| 680 | if res.Outcome != SwapAbandoned { |
| 681 | t.Fatalf("ReplaceWorker = %+v: a failed exec is a staging refusal, never a failure of the build", res) |
| 682 | } |
| 683 | waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) }) |
| 684 | } |
| 685 | |
| 686 | // Spec §8.1 (P5 review finding 7): a ReplaceWorker fallback holds its |
| 687 | // publication until the scheduler has recorded B's failure and calls |
| 688 | // PublishHeldServing. Published from the respawn, a blocked callback sat |
| 689 | // in front of the SwapFailed result, and so of the failure record. |
| 690 | func TestAWorkerFallbackPublishesOnlyWhenReleased(t *testing.T) { |
| 691 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond) |
| 692 | a, b, _, _, _ := stagedActivator(t) |
| 693 | var mu sync.Mutex |
| 694 | var got []string |
| 695 | block := make(chan struct{}) |
| 696 | a.OnWorkerServing = func(_ string, id release.StagedIdentity) { |
| 697 | mu.Lock() |
| 698 | got = append(got, id.Version) |
| 699 | n := len(got) |
| 700 | mu.Unlock() |
| 701 | if n > 2 { |
| 702 | <-block // a slow registry write for the fallback's identity (the wake's v1, then B's v2, come first) |
| 703 | } |
| 704 | } |
| 705 | host := "bot.x.internal" |
| 706 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 707 | t.Fatal(err) |
| 708 | } |
| 709 | good := fakeIdentity("v1") |
| 710 | go func() { |
| 711 | waitForWorkerStarts(t, b, 2) |
| 712 | time.Sleep(50 * time.Millisecond) |
| 713 | b.lastWorker().crash() |
| 714 | }() |
| 715 | results := make(chan SwapResult, 1) |
| 716 | go func() { |
| 717 | results <- a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), &good, one) |
| 718 | }() |
| 719 | select { |
| 720 | case res := <-results: |
| 721 | if res.Outcome != SwapFailed || res.Gen.Build != "v1" { |
| 722 | t.Fatalf("ReplaceWorker = %+v", res) |
| 723 | } |
| 724 | case <-time.After(5 * time.Second): |
| 725 | close(block) |
| 726 | t.Fatal("the failure result waited on the fallback's publication") |
| 727 | } |
| 728 | mu.Lock() |
| 729 | n := len(got) |
| 730 | mu.Unlock() |
| 731 | if n != 2 { |
| 732 | t.Fatalf("published %v before the failure was recorded", got) |
| 733 | } |
| 734 | close(block) |
| 735 | a.PublishHeldServing(host) // the scheduler, after done() |
| 736 | mu.Lock() |
| 737 | defer mu.Unlock() |
| 738 | if len(got) != 3 || got[2] != "v1" { |
| 739 | t.Fatalf("published %v, want the fallback's v1 once released", got) |
| 740 | } |
| 741 | } |
| 742 | |
| 743 | // Spec §8.1 (P5 review finding 7): on a planned worker wake that falls back, |
| 744 | // the failure transition (GenFellBack, whose OnGen records B's failure) |
| 745 | // completes before the fallback's identity is published, and a blocked |
| 746 | // publication does not hold it up. Publishing before afterWake left B |
| 747 | // unheld while the callback blocked. |
| 748 | func TestAFallbackWakeRecordsTheFailureBeforeItPublishes(t *testing.T) { |
| 749 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond) |
| 750 | a, b, _, _, _ := stagedActivator(t) |
| 751 | a.gens = true |
| 752 | a.backend = &plannedHostKill{stagedFake: b} // the failed child's host cgroup proves empty |
| 753 | evs := &genLog{} |
| 754 | a.OnGen = evs.add |
| 755 | dir := t.TempDir() |
| 756 | good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| 757 | a.Planner = func(context.Context, registry.Route) WakePlan { |
| 758 | return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good} |
| 759 | } |
| 760 | block := make(chan struct{}) |
| 761 | var sawFailure, called atomic.Bool |
| 762 | a.OnWorkerServing = func(string, release.StagedIdentity) { |
| 763 | sawFailure.Store(slices.Contains(evs.kinds(), GenFellBack)) |
| 764 | called.Store(true) |
| 765 | <-block |
| 766 | } |
| 767 | go func() { |
| 768 | waitForWorkerStarts(t, b, 1) |
| 769 | time.Sleep(50 * time.Millisecond) |
| 770 | b.lastWorker().crash() |
| 771 | }() |
| 772 | go a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")) //nolint:errcheck |
| 773 | evs.wait(t, GenFellBack) // arrives while the publication is still blocked |
| 774 | waitFor(t, "the publication", func() bool { return called.Load() }) |
| 775 | close(block) |
| 776 | if !sawFailure.Load() { |
| 777 | t.Fatal("the fallback's identity was published before its failure transition") |
| 778 | } |
| 779 | } |
| 780 | |
| 781 | // Final review F2 (spec §§6.4/8.3, P5's hold-everywhere invariant): "not |
| 782 | // installed" is not evidence that a build never ran. A hold attempt whose |
| 783 | // build was exec'd, and then superseded before its install (the install |
| 784 | // check sheds the child), may have migrated the database: the build it |
| 785 | // replaced is NOT restored, the entry stops, and the attempt is abandoned, |
| 786 | // not a failure of the build. Superseded at its exec gate instead, the hold |
| 787 | // build provably never ran, so the replaced build comes back as before. An |
| 788 | // overlap attempt superseded after its exec still restores, as before. |
| 789 | func TestASupersededWorkerAttemptRestoresOnlyWhatProvablyNeverRan(t *testing.T) { |
| 790 | for _, tc := range []struct { |
| 791 | name string |
| 792 | mode SwapMode |
| 793 | executed bool // superseded after the exec (true) or at the exec gate |
| 794 | restores bool |
| 795 | }{ |
| 796 | {"hold, superseded after its exec", ModeHold, true, false}, |
| 797 | {"hold, superseded at its exec gate", ModeHold, false, true}, |
| 798 | {"overlap, superseded after its exec", ModeOverlap, true, true}, |
| 799 | } { |
| 800 | t.Run(tc.name, func(t *testing.T) { |
| 801 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 802 | a, b, _, _, _ := stagedActivator(t) |
| 803 | host := "bot.x.internal" |
| 804 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 805 | t.Fatal(err) |
| 806 | } |
| 807 | var rev atomic.Uint64 |
| 808 | rev.Store(1) |
| 809 | var hooked atomic.Bool |
| 810 | b.mu.Lock() |
| 811 | if tc.executed { |
| 812 | b.workerStartHook = func(registry.Route) { |
| 813 | if hooked.CompareAndSwap(false, true) { |
| 814 | rev.Store(2) // after the exec, before the install check |
| 815 | } |
| 816 | } |
| 817 | } else { |
| 818 | b.admitHook = func() { |
| 819 | if hooked.CompareAndSwap(false, true) { |
| 820 | rev.Store(2) // during the preparation, before the exec gate |
| 821 | } |
| 822 | } |
| 823 | } |
| 824 | b.mu.Unlock() |
| 825 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, tc.mode, fakeIdentity("v2")), nil, |
| 826 | func(string) uint64 { return rev.Load() }) |
| 827 | if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) { |
| 828 | t.Fatalf("ReplaceWorker = %+v, want abandoned: superseded (never a failure of the build)", res) |
| 829 | } |
| 830 | wantStarts := 1 // the wake |
| 831 | if tc.executed { |
| 832 | wantStarts++ // the attempt's child, shed |
| 833 | } |
| 834 | if tc.restores { |
| 835 | wantStarts++ // the replaced build, back |
| 836 | waitFor(t, "the restore", func() bool { return b.workerStarts() == wantStarts && !a.entryRespawning(host) }) |
| 837 | if got := a.entryServing(host); got != fakeIdentity("v1") { |
| 838 | t.Fatalf("serving %+v, want the build the attempt replaced", got) |
| 839 | } |
| 840 | return |
| 841 | } |
| 842 | waitFor(t, "the entry to stop", func() bool { return !a.Owns(host) }) |
| 843 | time.Sleep(100 * time.Millisecond) |
| 844 | if n := b.workerStarts(); n != wantStarts { |
| 845 | t.Fatalf("%d worker children, want %d: the replaced build was restored after the hold build ran", n, wantStarts) |
| 846 | } |
| 847 | for _, c := range b.calls()[wantStarts-1:] { |
| 848 | if c != nil && *c == fakeIdentity("v1") { |
| 849 | t.Fatalf("spawns %v: the replaced build was respawned after the hold build ran", b.calls()) |
| 850 | } |
| 851 | } |
| 852 | }) |
| 853 | } |
| 854 | } |
| 855 | |
| 856 | // Round 2 Task 6 (recon B §1): a one-shot's clean exit inside its |
| 857 | // observation is a good build (spec §6.5) only when the sweep parked it |
| 858 | // after its own fence and lease checks. Any other stop of the exited child |
| 859 | // (a lost lease, a fence) leaves the entry not awake, and that answer was |
| 860 | // forgiven for every exited one-shot, so the build was recorded good with no |
| 861 | // ownership behind it. Those stops are abandoned; the sweep's park is done. |
| 862 | func TestAnExitedOneShotIsGoodOnlyAfterACleanPark(t *testing.T) { |
| 863 | for _, tc := range []struct { |
| 864 | name string |
| 865 | stop func(t *testing.T, a *Service, b *stagedFake, host string) |
| 866 | clean bool |
| 867 | }{ |
| 868 | {"parked by the sweep", func(t *testing.T, a *Service, b *stagedFake, host string) { |
| 869 | a.sweepOnce() |
| 870 | }, true}, |
| 871 | {"lease lost, then the sweep", func(t *testing.T, a *Service, b *stagedFake, host string) { |
| 872 | b.lastRep().setErr(errors.New("renewal failed")) |
| 873 | a.sweepOnce() |
| 874 | }, false}, |
| 875 | {"fenced", func(t *testing.T, a *Service, b *stagedFake, host string) { |
| 876 | ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| 877 | defer cancel() |
| 878 | if _, err := a.FenceAndStop(ctx, host, 1); err != nil { |
| 879 | t.Errorf("FenceAndStop: %v", err) |
| 880 | } |
| 881 | }, false}, |
| 882 | {"parked, then fenced before the result", func(t *testing.T, a *Service, b *stagedFake, host string) { |
| 883 | a.sweepOnce() |
| 884 | ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| 885 | defer cancel() |
| 886 | if _, err := a.FenceAndStop(ctx, host, 1); err != nil { |
| 887 | t.Errorf("FenceAndStop: %v", err) |
| 888 | } |
| 889 | }, false}, |
| 890 | } { |
| 891 | t.Run(tc.name, func(t *testing.T) { |
| 892 | // A 2 s window polls every 250 ms (settlePoll): the stop below |
| 893 | // lands well before the observation's first look at the exit. |
| 894 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second) |
| 895 | a, b, _, _, _ := stagedActivator(t) |
| 896 | host := "bot.x.internal" |
| 897 | if err := a.WakeWorker(testWorkerRoute(host)); err != nil { |
| 898 | t.Fatal(err) |
| 899 | } |
| 900 | done := make(chan struct{}) |
| 901 | go func() { |
| 902 | defer close(done) |
| 903 | waitForWorkerStarts(t, b, 2) |
| 904 | waitFor(t, "the observation", func() bool { |
| 905 | a.mu.Lock() |
| 906 | defer a.mu.Unlock() |
| 907 | e := a.entries[host] |
| 908 | return e != nil && e.firstWatch != nil |
| 909 | }) |
| 910 | b.lastWorker().exit() // status 0: the one-shot finished |
| 911 | tc.stop(t, a, b, host) |
| 912 | }() |
| 913 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one) |
| 914 | <-done |
| 915 | if tc.clean { |
| 916 | if res.Outcome != SwapDone || !res.CleanExit { |
| 917 | t.Fatalf("ReplaceWorker = %+v, want done with a clean exit: the sweep parked a finished one-shot", res) |
| 918 | } |
| 919 | return |
| 920 | } |
| 921 | if res.Outcome != SwapAbandoned || res.CleanExit { |
| 922 | t.Fatalf("ReplaceWorker = %+v, want abandoned: no clean park with no fence stands behind the exit, so its build must not be recorded good", res) |
| 923 | } |
| 924 | }) |
| 925 | } |
| 926 | } |
| 927 | |
| 928 | // awaitExtensionSeam installs wakeBoundExtended for one test: the first |
| 929 | // extension of a waiter's bound lets the wake fail (release) and waits until |
| 930 | // that failed wake has deleted its entry, the window in which a waiter that |
| 931 | // re-read the entries map started a second wake. |
| 932 | func awaitExtensionSeam(t *testing.T, a *Service, release chan struct{}) *atomic.Int32 { |
| 933 | t.Helper() |
| 934 | var calls atomic.Int32 |
| 935 | old := wakeBoundExtended |
| 936 | wakeBoundExtended = func() { |
| 937 | if calls.Add(1) != 1 { |
| 938 | return |
| 939 | } |
| 940 | close(release) |
| 941 | waitFor(t, "the failed wake to delete its entry", func() bool { return a.entryCount() == 0 }) |
| 942 | } |
| 943 | t.Cleanup(func() { wakeBoundExtended = old }) |
| 944 | return &calls |
| 945 | } |
| 946 | |
| 947 | // Round 2 Task 6 (recon B §2): a first wake's waiter extends its bound once |
| 948 | // (spec §8.2) and then waits again on the SAME ready. It used to loop back to |
| 949 | // the entries map: a wake that failed inside the extension had deleted its |
| 950 | // entry, so the waiter found nothing and started a second wake, a second |
| 951 | // lease acquire and a second spawn (#184's shape). |
| 952 | func TestWakeWorkerAWakeFailingInsideTheExtensionStartsNoSecond(t *testing.T) { |
| 953 | old := wakeWaitBound |
| 954 | wakeWaitBound = 50 * time.Millisecond |
| 955 | t.Cleanup(func() { wakeWaitBound = old }) |
| 956 | a, b, _, _, _ := stagedActivator(t) |
| 957 | a.gens = true |
| 958 | a.backend = &plannedHostKill{stagedFake: b} |
| 959 | next := stagedID(t, t.TempDir(), "v2") |
| 960 | a.Planner = func(context.Context, registry.Route) WakePlan { |
| 961 | return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true} |
| 962 | } |
| 963 | boom := errors.New("exec refused") |
| 964 | release := make(chan struct{}) |
| 965 | calls := awaitExtensionSeam(t, a, release) |
| 966 | b.mu.Lock() |
| 967 | b.workerStartHook = func(registry.Route) { |
| 968 | <-release |
| 969 | b.mu.Lock() |
| 970 | b.workerErr = boom |
| 971 | b.mu.Unlock() |
| 972 | } |
| 973 | b.mu.Unlock() |
| 974 | err := a.WakeWorker(testWorkerRoute("bot.x.internal")) |
| 975 | if calls.Load() == 0 { |
| 976 | t.Fatal("the waiter never extended its bound: the test did not reach the window") |
| 977 | } |
| 978 | if !errors.Is(err, boom) { |
| 979 | t.Fatalf("WakeWorker = %v, want the wake's own error", err) |
| 980 | } |
| 981 | if acquires, _, _ := b.counts(); acquires != 1 { |
| 982 | t.Fatalf("%d lease acquires, want 1: a waiter whose wake failed inside the extension started a second wake", acquires) |
| 983 | } |
| 984 | } |
| 985 | |
| 986 | // The same window for a request (begin): the failed wake reports to its |
| 987 | // waiter, and nothing wakes the instance a second time. |
| 988 | func TestBeginAWakeFailingInsideTheExtensionStartsNoSecond(t *testing.T) { |
| 989 | old := wakeWaitBound |
| 990 | wakeWaitBound = 50 * time.Millisecond |
| 991 | t.Cleanup(func() { wakeWaitBound = old }) |
| 992 | next := stagedID(t, t.TempDir(), "v2") |
| 993 | a, b, _ := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true}) |
| 994 | boom := errors.New("GET / answered 500") |
| 995 | release := make(chan struct{}) |
| 996 | calls := awaitExtensionSeam(t, a, release) |
| 997 | a.genHealth = func(context.Context, Gen) error { |
| 998 | <-release |
| 999 | return boom |
| 1000 | } |
| 1001 | _, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| 1002 | if calls.Load() == 0 { |
| 1003 | t.Fatal("the waiter never extended its bound: the test did not reach the window") |
| 1004 | } |
| 1005 | if err == nil || errors.Is(err, errWakeTimeout) { |
| 1006 | t.Fatalf("beginRequest = %v, want the failed wake's error", err) |
| 1007 | } |
| 1008 | if acquires, _, _ := b.counts(); acquires != 1 { |
| 1009 | t.Fatalf("%d lease acquires, want 1: a waiter whose wake failed inside the extension started a second wake", acquires) |
| 1010 | } |
| 1011 | if gens := b.started(); len(gens) != 1 { |
| 1012 | t.Fatalf("spawned %+v, want one generation", gens) |
| 1013 | } |
| 1014 | } |
| 1015 | |
amadan