| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "sync" |
| 10 | "sync/atomic" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/cgroup" |
| 15 | "github.com/carlosframework/platform/internal/instancectl" |
| 16 | "github.com/carlosframework/platform/internal/registry" |
| 17 | ) |
| 18 | |
| 19 | // Final review F8: SwapResult.Executed says whether a process of the |
| 20 | // attempt's build started. The scheduler re-arms a `carlos instances retry` |
| 21 | // allowance only for an attempt that ended before that (an abandonment with |
| 22 | // Executed false); an attempt whose build ran spends it, whatever its |
| 23 | // outcome. These cases cover each mode's refusals before the exec, its |
| 24 | // abandonments after the exec, and its executed outcomes. |
| 25 | func TestSwapResultReportsWhetherTheAttemptsBuildExecuted(t *testing.T) { |
| 26 | t.Parallel() |
| 27 | for _, tc := range []struct { |
| 28 | name string |
| 29 | mode SwapMode |
| 30 | arrange func(f *swapFixture) |
| 31 | outcome SwapOutcome |
| 32 | executed bool |
| 33 | }{ |
| 34 | {"overlap, refused at the exec", ModeOverlap, func(f *swapFixture) { |
| 35 | f.b.onRaise = func() { f.rev.Add(1) } |
| 36 | }, SwapAbandoned, false}, |
| 37 | {"overlap, the spawn fails", ModeOverlap, func(f *swapFixture) { |
| 38 | f.b.genErr = errors.New("exec format error") |
| 39 | }, SwapFailed, false}, |
| 40 | // Amadan #272: a parent that holds a legacy process directly refuses |
| 41 | // the generation start before any exec. That is the box's condition, |
| 42 | // not a failure of the build, so the attempt is abandoned unspent. |
| 43 | {"overlap, the host's parent holds a process", ModeOverlap, func(f *swapFixture) { |
| 44 | f.b.genErr = fmt.Errorf("h.example.com: %w: holds 1 process(es)", cgroup.ErrLegacyParentPopulated) |
| 45 | }, SwapAbandoned, false}, |
| 46 | {"hold, the host's parent holds a process", ModeHold, func(f *swapFixture) { |
| 47 | f.b.genErr = fmt.Errorf("h.example.com: %w: holds 1 process(es)", cgroup.ErrLegacyParentPopulated) |
| 48 | }, SwapAbandoned, false}, |
| 49 | // Amadan #272, Task 19: the cgroup the successor would spawn into |
| 50 | // took cgroup.kill and could not be made fresh. The manager refuses |
| 51 | // before it builds the command: the box's condition, nothing ran. |
| 52 | {"overlap, a killed cgroup would not go", ModeOverlap, func(f *swapFixture) { |
| 53 | f.b.genErr = fmt.Errorf("%w: carlos-exec@h.example.com/g2: device or resource busy", cgroup.ErrKilledCgroupKept) |
| 54 | }, SwapAbandoned, false}, |
| 55 | {"hold, a killed cgroup would not go", ModeHold, func(f *swapFixture) { |
| 56 | f.b.genErr = fmt.Errorf("%w: carlos-exec@h.example.com: device or resource busy", cgroup.ErrKilledCgroupKept) |
| 57 | }, SwapAbandoned, false}, |
| 58 | // Task 7 (readiness round 2): the successor's limits could not be |
| 59 | // applied, so the manager refused before it built the command: the |
| 60 | // box's condition, nothing ran, the allowance is not spent. |
| 61 | {"overlap, the generation's limits were not applied", ModeOverlap, func(f *swapFixture) { |
| 62 | f.b.genErr = fmt.Errorf("%w: g2: memory.max: is a directory", cgroup.ErrLimitNotApplied) |
| 63 | }, SwapAbandoned, false}, |
| 64 | {"hold, the generation's limits were not applied", ModeHold, func(f *swapFixture) { |
| 65 | f.b.genErr = fmt.Errorf("%w: h.example.com: the placement shim is not available", cgroup.ErrLimitNotApplied) |
| 66 | }, SwapAbandoned, false}, |
| 67 | {"overlap, superseded at the switch", ModeOverlap, func(f *swapFixture) { |
| 68 | f.a.genHealth = func(context.Context, Gen) error { f.rev.Add(1); return nil } |
| 69 | }, SwapAbandoned, true}, |
| 70 | {"overlap, done", ModeOverlap, func(*swapFixture) {}, SwapDone, true}, |
| 71 | {"hold, superseded before the exec", ModeHold, func(f *swapFixture) { |
| 72 | var once sync.Once |
| 73 | f.a.backend = &acquireHookBackend{fakeSwapBackend: f.b, hook: func() { once.Do(func() { f.rev.Add(1) }) }} |
| 74 | }, SwapAbandoned, false}, |
| 75 | {"hold, superseded before the result", ModeHold, func(f *swapFixture) { |
| 76 | f.a.genHealth = func(context.Context, Gen) error { f.rev.Add(1); return nil } |
| 77 | }, SwapAbandoned, true}, |
| 78 | {"hold, unhealthy", ModeHold, func(f *swapFixture) { |
| 79 | f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") } |
| 80 | }, SwapDown, true}, |
| 81 | {"hold, done", ModeHold, func(*swapFixture) {}, SwapDone, true}, |
| 82 | } { |
| 83 | t.Run(tc.name, func(t *testing.T) { |
| 84 | f := newSwapFixture(t) |
| 85 | tc.arrange(f) |
| 86 | res := f.a.Swap(context.Background(), f.attempt(tc.mode), f.newest) |
| 87 | if res.Outcome != tc.outcome { |
| 88 | t.Fatalf("Swap = %+v, want %s", res, tc.outcome) |
| 89 | } |
| 90 | if res.Executed != tc.executed { |
| 91 | t.Fatalf("Swap = %+v: Executed %v, want %v", res, res.Executed, tc.executed) |
| 92 | } |
| 93 | }) |
| 94 | } |
| 95 | } |
| 96 | |
| 97 | // The same for a worker's same-entry replacement, from the respawn's |
| 98 | // execution evidence (final review F2's attemptExec). |
| 99 | func TestReplaceWorkerReportsWhetherTheAttemptsBuildExecuted(t *testing.T) { |
| 100 | for _, tc := range []struct { |
| 101 | name string |
| 102 | mode SwapMode |
| 103 | arrange func(t *testing.T, a *Service, b *stagedFake, rev *atomic.Uint64) |
| 104 | outcome SwapOutcome |
| 105 | executed bool |
| 106 | }{ |
| 107 | {"hold, superseded at its exec gate", ModeHold, func(_ *testing.T, _ *Service, b *stagedFake, rev *atomic.Uint64) { |
| 108 | var hooked atomic.Bool |
| 109 | b.mu.Lock() |
| 110 | b.admitHook = func() { |
| 111 | if hooked.CompareAndSwap(false, true) { |
| 112 | rev.Store(2) |
| 113 | } |
| 114 | } |
| 115 | b.mu.Unlock() |
| 116 | }, SwapAbandoned, false}, |
| 117 | {"hold, superseded after its exec", ModeHold, func(_ *testing.T, _ *Service, b *stagedFake, rev *atomic.Uint64) { |
| 118 | var hooked atomic.Bool |
| 119 | b.mu.Lock() |
| 120 | b.workerStartHook = func(registry.Route) { |
| 121 | if hooked.CompareAndSwap(false, true) { |
| 122 | rev.Store(2) |
| 123 | } |
| 124 | } |
| 125 | b.mu.Unlock() |
| 126 | }, SwapAbandoned, true}, |
| 127 | {"refused for its publication", ModeOverlap, func(t *testing.T, a *Service, _ *stagedFake, _ *atomic.Uint64) { |
| 128 | blocker := filepath.Join(t.TempDir(), "not-a-dir") |
| 129 | if err := os.WriteFile(blocker, nil, 0o644); err != nil { |
| 130 | t.Fatal(err) |
| 131 | } |
| 132 | a.mu.Lock() |
| 133 | a.GenerationsFile = filepath.Join(blocker, "generations.json") |
| 134 | a.mu.Unlock() |
| 135 | }, SwapAbandoned, false}, |
| 136 | {"done", ModeOverlap, func(*testing.T, *Service, *stagedFake, *atomic.Uint64) {}, SwapDone, true}, |
| 137 | } { |
| 138 | t.Run(tc.name, func(t *testing.T) { |
| 139 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 140 | a, b, _, _, _ := stagedActivator(t) |
| 141 | host := "bot.x.internal" |
| 142 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 143 | t.Fatal(err) |
| 144 | } |
| 145 | var rev atomic.Uint64 |
| 146 | rev.Store(1) |
| 147 | tc.arrange(t, a, b, &rev) |
| 148 | res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, tc.mode, fakeIdentity("v2")), nil, |
| 149 | func(string) uint64 { return rev.Load() }) |
| 150 | if res.Outcome != tc.outcome { |
| 151 | t.Fatalf("ReplaceWorker = %+v, want %s", res, tc.outcome) |
| 152 | } |
| 153 | if res.Executed != tc.executed { |
| 154 | t.Fatalf("ReplaceWorker = %+v: Executed %v, want %v", res, res.Executed, tc.executed) |
| 155 | } |
| 156 | waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) }) |
| 157 | }) |
| 158 | } |
| 159 | } |
| 160 | |
| 161 | // Final review wave 2 (N4): Executed is false only when it is PROVEN that |
| 162 | // nothing of the attempt's build ran. A nested start that root was asked |
| 163 | // for and that failed with no process (the unit exited before it was |
| 164 | // observed, or root's start answered an error), or whose kill is |
| 165 | // unproven, may have run the build: executed or unknown. Only root's |
| 166 | // refusal of the staged build before it started anything (ErrStagedExe) |
| 167 | // is proven unexecuted. Before the fix both modes read the nil handle of |
| 168 | // a kill-proven failure as "never ran", and an overlap read an unproven |
| 169 | // kill the same way, so an abandoned forced attempt regained an allowance |
| 170 | // it may have spent. |
| 171 | func TestASwapWhoseNestedStartIsUncertainReportsExecution(t *testing.T) { |
| 172 | t.Parallel() |
| 173 | unitFailed := errors.New("the unit failed to start") |
| 174 | for _, tc := range []struct { |
| 175 | name string |
| 176 | mode SwapMode |
| 177 | exeErr error |
| 178 | stopErr error |
| 179 | executed bool |
| 180 | }{ |
| 181 | {"overlap, root's start failed, kill proven", ModeOverlap, unitFailed, nil, true}, |
| 182 | {"overlap, root's start failed, kill unproven", ModeOverlap, unitFailed, errors.New("stop timed out"), true}, |
| 183 | {"overlap, root refused the staged build", ModeOverlap, fmt.Errorf("%w: digest mismatch", instancectl.ErrStagedExe), nil, false}, |
| 184 | {"hold, root's start failed, kill proven", ModeHold, unitFailed, nil, true}, |
| 185 | {"hold, root refused the staged build", ModeHold, fmt.Errorf("%w: digest mismatch", instancectl.ErrStagedExe), nil, false}, |
| 186 | } { |
| 187 | t.Run(tc.name, func(t *testing.T) { |
| 188 | f := newSwapFixture(t) |
| 189 | b := newLaunchBackend(t, "nested", f.b) |
| 190 | b.ctl.exeErr, b.ctl.stopGenErr = tc.exeErr, tc.stopErr |
| 191 | f.a.backend = b |
| 192 | id := markerScript(t, t.TempDir(), "v2", filepath.Join(t.TempDir(), "v2.ran")) |
| 193 | at := f.attempt(tc.mode) |
| 194 | at.StagedPath, at.Digest = id.StagedPath, id.Digest |
| 195 | res := f.a.Swap(context.Background(), at, f.newest) |
| 196 | if res.Outcome == SwapDone { |
| 197 | t.Fatalf("Swap = %+v, want it not done", res) |
| 198 | } |
| 199 | b.ctl.mu.Lock() |
| 200 | asked := len(b.ctl.got) |
| 201 | b.ctl.mu.Unlock() |
| 202 | if asked != 1 { |
| 203 | t.Fatalf("root was asked to start %d units, want 1", asked) |
| 204 | } |
| 205 | if res.Executed != tc.executed { |
| 206 | t.Fatalf("Swap = %+v: Executed %v, want %v", res, res.Executed, tc.executed) |
| 207 | } |
| 208 | }) |
| 209 | } |
| 210 | } |
| 211 | |
| 212 | // Astra R3 ruling 206 (zdd-readiness Task 6): a hold's nested start that |
| 213 | // root accepted and whose outcome is uncertain may have run its build, so |
| 214 | // the hold is down (SwapDown: done() holds the build down and halts the |
| 215 | // cohort), never an abandonment with Executed, whose deferral would let the |
| 216 | // scheduler run the build again. That holds for a unit that exited before |
| 217 | // it was observed (errStartedThenExited) and for any other uncertain start, |
| 218 | // whether the backend's kill proved the generation gone or not. Only root's |
| 219 | // refusal of the staged build before it started anything is still an |
| 220 | // abandonment, with nothing executed. Before the fix the kill-proven cases |
| 221 | // were abandoned with Executed true. |
| 222 | func TestAHoldWhoseNestedStartIsUncertainIsDown(t *testing.T) { |
| 223 | t.Parallel() |
| 224 | unitFailed := errors.New("the unit failed to start") |
| 225 | exitedUnobserved := func(instancectl.StagedExe) (string, error) { return "", nil } |
| 226 | rootFailed := func(instancectl.StagedExe) (string, error) { return "", unitFailed } |
| 227 | rootRefused := func(instancectl.StagedExe) (string, error) { |
| 228 | return "", fmt.Errorf("%w: digest mismatch", instancectl.ErrStagedExe) |
| 229 | } |
| 230 | for _, tc := range []struct { |
| 231 | name string |
| 232 | answer func(instancectl.StagedExe) (string, error) |
| 233 | stopErr error |
| 234 | outcome SwapOutcome |
| 235 | executed bool |
| 236 | }{ |
| 237 | {"the unit exited before it was observed, kill proven", exitedUnobserved, nil, SwapDown, true}, |
| 238 | {"the unit exited before it was observed, kill unproven", exitedUnobserved, errors.New("stop timed out"), SwapDown, true}, |
| 239 | {"root's start failed, kill proven", rootFailed, nil, SwapDown, true}, |
| 240 | {"root's start failed, kill unproven", rootFailed, errors.New("stop timed out"), SwapDown, true}, |
| 241 | {"root refused the staged build", rootRefused, nil, SwapAbandoned, false}, |
| 242 | } { |
| 243 | t.Run(tc.name, func(t *testing.T) { |
| 244 | f := newSwapFixture(t) |
| 245 | b := newLaunchBackend(t, "nested", f.b) |
| 246 | b.ctl.exeAnswer, b.ctl.stopGenErr = tc.answer, tc.stopErr |
| 247 | f.a.backend = b |
| 248 | id := markerScript(t, t.TempDir(), "v2", filepath.Join(t.TempDir(), "v2.ran")) |
| 249 | at := f.attempt(ModeHold) |
| 250 | at.StagedPath, at.Digest = id.StagedPath, id.Digest |
| 251 | res := f.a.Swap(context.Background(), at, f.newest) |
| 252 | if res.Outcome != tc.outcome || res.Executed != tc.executed { |
| 253 | t.Fatalf("Swap = %+v, want %s with Executed %v", res, tc.outcome, tc.executed) |
| 254 | } |
| 255 | b.ctl.mu.Lock() |
| 256 | asked := len(b.ctl.got) |
| 257 | b.ctl.mu.Unlock() |
| 258 | if asked != 1 { |
| 259 | t.Fatalf("root was asked to start %d units, want 1", asked) |
| 260 | } |
| 261 | }) |
| 262 | } |
| 263 | } |
| 264 | |
| 265 | // Final review wave 2 (N4): ReplaceWorker learns what its attempt's |
| 266 | // respawn did before it answers, even when its caller stops waiting. The |
| 267 | // respawn runs on (it is the entry's, on the service lifetime), so it is |
| 268 | // joined, bounded by respawnJoinBound, and one still running past that is |
| 269 | // reported executed-or-unknown. Before the fix a cancelled ReplaceWorker |
| 270 | // returned at once with Executed false, while the respawn went on to exec |
| 271 | // the attempt's build. |
| 272 | func TestReplaceWorkerLearnsItsRespawnsExecutionWhenItsCallerLeaves(t *testing.T) { |
| 273 | for _, tc := range []struct { |
| 274 | name string |
| 275 | // supersede: the attempt is superseded while the respawn is held, so |
| 276 | // its exec gate refuses: proven unexecuted. |
| 277 | supersede bool |
| 278 | // bound shrinks respawnJoinBound below the hold: the join gives up. |
| 279 | bound time.Duration |
| 280 | executed bool |
| 281 | }{ |
| 282 | {"the respawn execs after the caller left", false, 0, true}, |
| 283 | {"the respawn is refused at its exec after the caller left", true, 0, false}, |
| 284 | {"the respawn outlasts the join bound", false, 150 * time.Millisecond, true}, |
| 285 | } { |
| 286 | t.Run(tc.name, func(t *testing.T) { |
| 287 | if tc.bound > 0 { |
| 288 | defer func(d time.Duration) { respawnJoinBound = d }(respawnJoinBound) |
| 289 | respawnJoinBound = tc.bound |
| 290 | } |
| 291 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 292 | a, b, _, _, _ := stagedActivator(t) |
| 293 | host := "bot.x.internal" |
| 294 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 295 | t.Fatal(err) |
| 296 | } |
| 297 | var rev atomic.Uint64 |
| 298 | rev.Store(1) |
| 299 | entered, release := make(chan struct{}), make(chan struct{}) |
| 300 | var releaseOnce sync.Once |
| 301 | releaseAll := func() { releaseOnce.Do(func() { close(release) }) } |
| 302 | t.Cleanup(releaseAll) |
| 303 | var hooked atomic.Bool |
| 304 | b.mu.Lock() |
| 305 | b.admitHook = func() { |
| 306 | if hooked.CompareAndSwap(false, true) { |
| 307 | close(entered) |
| 308 | <-release |
| 309 | } |
| 310 | } |
| 311 | b.mu.Unlock() |
| 312 | ctx, cancel := context.WithCancel(context.Background()) |
| 313 | defer cancel() |
| 314 | out := make(chan SwapResult, 1) |
| 315 | go func() { |
| 316 | out <- a.ReplaceWorker(ctx, workerAttempt(CauseRelease, ModeHold, fakeIdentity("v2")), nil, |
| 317 | func(string) uint64 { return rev.Load() }) |
| 318 | }() |
| 319 | select { |
| 320 | case <-entered: |
| 321 | case <-time.After(5 * time.Second): |
| 322 | t.Fatal("the attempt's respawn never reached its preparation") |
| 323 | } |
| 324 | cancel() |
| 325 | if tc.bound > 0 { |
| 326 | // The respawn stays held past the bound: the join gives up. |
| 327 | select { |
| 328 | case res := <-out: |
| 329 | if res.Outcome != SwapAbandoned || res.Executed != tc.executed { |
| 330 | t.Fatalf("ReplaceWorker = %+v: want abandoned, Executed %v", res, tc.executed) |
| 331 | } |
| 332 | case <-time.After(5 * time.Second): |
| 333 | t.Fatal("ReplaceWorker never answered: the join is not bounded") |
| 334 | } |
| 335 | releaseAll() |
| 336 | waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(host) }) |
| 337 | return |
| 338 | } |
| 339 | select { |
| 340 | case res := <-out: |
| 341 | t.Fatalf("ReplaceWorker = %+v answered while its respawn was still preparing", res) |
| 342 | case <-time.After(200 * time.Millisecond): |
| 343 | } |
| 344 | if tc.supersede { |
| 345 | rev.Store(2) |
| 346 | } |
| 347 | releaseAll() |
| 348 | select { |
| 349 | case res := <-out: |
| 350 | if res.Outcome != SwapAbandoned || res.Executed != tc.executed { |
| 351 | t.Fatalf("ReplaceWorker = %+v: want abandoned, Executed %v", res, tc.executed) |
| 352 | } |
| 353 | case <-time.After(5 * time.Second): |
| 354 | t.Fatal("ReplaceWorker never answered after its respawn finished") |
| 355 | } |
| 356 | waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) }) |
| 357 | }) |
| 358 | } |
| 359 | } |
| 360 | |
amadan