| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "slices" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/registry" |
| 13 | "github.com/carlosframework/platform/internal/release" |
| 14 | ) |
| 15 | |
| 16 | // cmdSwapBackend is the swap fixture's backend with the run capability, on |
| 17 | // the nested layout (merge ruling M2). |
| 18 | type cmdSwapBackend struct { |
| 19 | *fakeSwapBackend |
| 20 | cmu sync.Mutex |
| 21 | unit *fakeCmdUnit |
| 22 | specs []CmdSpec |
| 23 | gate chan struct{} // when set, StartCmd blocks until it closes |
| 24 | entered chan struct{} // when set, closed as StartCmd is entered |
| 25 | onStart func() // when set, runs inside StartCmd, before the unit is handed back |
| 26 | } |
| 27 | |
| 28 | func (b *cmdSwapBackend) StartCmd(_ context.Context, _ registry.Route, spec CmdSpec) (CmdUnit, error) { |
| 29 | b.cmu.Lock() |
| 30 | gate, entered, onStart := b.gate, b.entered, b.onStart |
| 31 | b.entered = nil |
| 32 | b.cmu.Unlock() |
| 33 | if entered != nil { |
| 34 | close(entered) |
| 35 | } |
| 36 | if onStart != nil { |
| 37 | onStart() |
| 38 | } |
| 39 | if gate != nil { |
| 40 | <-gate |
| 41 | } |
| 42 | b.cmu.Lock() |
| 43 | defer b.cmu.Unlock() |
| 44 | b.specs = append(b.specs, spec) |
| 45 | return b.unit, nil |
| 46 | } |
| 47 | |
| 48 | func (b *cmdSwapBackend) started() []CmdSpec { |
| 49 | b.cmu.Lock() |
| 50 | defer b.cmu.Unlock() |
| 51 | return slices.Clone(b.specs) |
| 52 | } |
| 53 | |
| 54 | func newCmdSwapFixture(t *testing.T) (*swapFixture, *cmdSwapBackend) { |
| 55 | t.Helper() |
| 56 | var cb *cmdSwapBackend |
| 57 | f := newSwapFixtureOn(t, func(b *fakeSwapBackend) Backend { |
| 58 | cb = &cmdSwapBackend{fakeSwapBackend: b, unit: &fakeCmdUnit{}} |
| 59 | return cb |
| 60 | }) |
| 61 | f.b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" } |
| 62 | return f, cb |
| 63 | } |
| 64 | |
| 65 | func (a *Service) runGenOf(h CmdHandle) Gen { |
| 66 | a.mu.Lock() |
| 67 | defer a.mu.Unlock() |
| 68 | return h.(*cmdRun).h.gen |
| 69 | } |
| 70 | |
| 71 | // A run whose unit is starting refuses an overlap swap's admission: the |
| 72 | // attempt is abandoned with nothing spawned, never admitted to retire the |
| 73 | // generation that run is starting against. Once the start has returned, |
| 74 | // the same attempt is admitted (merge ruling M2). |
| 75 | func TestARunStartInFlightRefusesAnOverlapSwap(t *testing.T) { |
| 76 | f, cb := newCmdSwapFixture(t) |
| 77 | cb.gate, cb.entered = make(chan struct{}), make(chan struct{}) |
| 78 | entered := cb.entered |
| 79 | type ran struct { |
| 80 | h CmdHandle |
| 81 | err error |
| 82 | } |
| 83 | done := make(chan ran, 1) |
| 84 | go func() { |
| 85 | h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"}) |
| 86 | done <- ran{h, err} |
| 87 | }() |
| 88 | <-entered |
| 89 | res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) |
| 90 | if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapBusy.Error()) || res.Executed { |
| 91 | t.Fatalf("Swap during the run's start = %+v, want abandoned as busy with nothing run", res) |
| 92 | } |
| 93 | if gens, raises, _ := f.b.calls(); gens != 0 || raises != 0 { |
| 94 | t.Fatalf("the refused attempt spawned %d generations and raised %d times", gens, raises) |
| 95 | } |
| 96 | close(cb.gate) |
| 97 | r := <-done |
| 98 | if r.err != nil { |
| 99 | t.Fatal(r.err) |
| 100 | } |
| 101 | if g := f.a.runGenOf(r.h); g.N != f.incumbent.N { |
| 102 | t.Fatalf("the run was admitted against generation %d, want the incumbent %d", g.N, f.incumbent.N) |
| 103 | } |
| 104 | f.rev.Add(1) |
| 105 | if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone { |
| 106 | t.Fatalf("Swap after the start = %+v", res) |
| 107 | } |
| 108 | } |
| 109 | |
| 110 | // A run that arrives while an overlap successor is spawning starts no unit |
| 111 | // against the incumbent, which is about to retire: it waits for the switch |
| 112 | // and then runs the successor's build, admitted against the successor's |
| 113 | // generation (merge ruling M2). |
| 114 | func TestARunWaitsForAPendingSwitch(t *testing.T) { |
| 115 | t.Parallel() |
| 116 | f, cb := newCmdSwapFixture(t) |
| 117 | f.b.gmu.Lock() |
| 118 | f.b.genGate = make(chan struct{}) |
| 119 | f.b.gmu.Unlock() |
| 120 | swapped := make(chan SwapResult, 1) |
| 121 | go func() { swapped <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }() |
| 122 | waitFor(t, "the successor in the operation set", func() bool { |
| 123 | _, members := f.a.genStateForTest(f.host) |
| 124 | return len(members) == 1 |
| 125 | }) |
| 126 | type ran struct { |
| 127 | h CmdHandle |
| 128 | err error |
| 129 | } |
| 130 | done := make(chan ran, 1) |
| 131 | go func() { |
| 132 | h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"}) |
| 133 | done <- ran{h, err} |
| 134 | }() |
| 135 | time.Sleep(3 * cmdSwitchPoll) |
| 136 | if n := len(cb.started()); n != 0 { |
| 137 | t.Fatalf("%d units started while a successor was pending", n) |
| 138 | } |
| 139 | f.b.gmu.Lock() |
| 140 | close(f.b.genGate) |
| 141 | f.b.gmu.Unlock() |
| 142 | res := <-swapped |
| 143 | if res.Outcome != SwapDone { |
| 144 | t.Fatalf("Swap = %+v", res) |
| 145 | } |
| 146 | r := <-done |
| 147 | if r.err != nil { |
| 148 | t.Fatal(r.err) |
| 149 | } |
| 150 | defer r.h.Release() |
| 151 | if g := f.a.runGenOf(r.h); g.N != res.Gen.N { |
| 152 | t.Fatalf("the run was admitted against generation %d, want the successor %d", g.N, res.Gen.N) |
| 153 | } |
| 154 | specs := cb.started() |
| 155 | if len(specs) != 1 || specs[0].serving == nil || specs[0].serving.Path != f.staged { |
| 156 | t.Fatalf("the run started %+v, want one unit on the successor's build %s", specs, f.staged) |
| 157 | } |
| 158 | if n, _ := cb.unit.stopped(); n != 0 { |
| 159 | t.Fatal("a run on the current generation was interrupted") |
| 160 | } |
| 161 | } |
| 162 | |
| 163 | // A running command on a generation a swap retires is interrupted and |
| 164 | // settled, and the run reads why: a new release for a release attempt, a |
| 165 | // restart for a scheduled restart or config change (exit 125, RUN-C75; |
| 166 | // merge ruling M2). |
| 167 | func TestARunOnARetiredGenerationIsInterrupted(t *testing.T) { |
| 168 | t.Parallel() |
| 169 | for _, c := range []struct { |
| 170 | cause Cause |
| 171 | why string |
| 172 | }{{CauseRelease, StopRelease}, {CauseRestart, StopRestart}, {CauseConfig, StopRestart}} { |
| 173 | t.Run(string(c.cause), func(t *testing.T) { |
| 174 | f, cb := newCmdSwapFixture(t) |
| 175 | h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"}) |
| 176 | if err != nil { |
| 177 | t.Fatal(err) |
| 178 | } |
| 179 | if g := f.a.runGenOf(h); g.N != f.incumbent.N || g.StagedPath != f.incumbent.StagedPath { |
| 180 | t.Fatalf("the run was admitted against %+v, want the incumbent", g) |
| 181 | } |
| 182 | at := f.attempt(ModeOverlap) |
| 183 | at.Cause = c.cause |
| 184 | if res := f.a.Swap(context.Background(), at, f.newest); res.Outcome != SwapDone { |
| 185 | t.Fatalf("Swap = %+v", res) |
| 186 | } |
| 187 | waitFor(t, "the run interrupted and settled", func() bool { |
| 188 | n, _ := cb.unit.stopped() |
| 189 | return n == 1 && h.(*cmdRun).h.provenSettled() |
| 190 | }) |
| 191 | if why, at := h.Interrupted(); why != c.why || at.IsZero() { |
| 192 | t.Fatalf("Interrupted() = %q at %v, want %q", why, at, c.why) |
| 193 | } |
| 194 | if f.a.entryCount() != 1 { |
| 195 | t.Fatal("interrupting the run stopped the instance") |
| 196 | } |
| 197 | h.Release() |
| 198 | }) |
| 199 | } |
| 200 | } |
| 201 | |
| 202 | // A run on a retired generation keeps that generation's build in the |
| 203 | // generations file until the run is proven settled, so root neither |
| 204 | // restages nor sweeps the bytes the command runs; the settle that proves |
| 205 | // it releases the pin (merge ruling M2). |
| 206 | func TestARetiredRunKeepsItsBuildPinnedUntilItSettles(t *testing.T) { |
| 207 | t.Parallel() |
| 208 | f, cb := newCmdSwapFixture(t) |
| 209 | path := release.GenerationsPath(t.TempDir()) |
| 210 | f.a.GenerationsFile = path |
| 211 | cb.unit.setSettleErr(errors.New("processes remained")) |
| 212 | h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"}) |
| 213 | if err != nil { |
| 214 | t.Fatal(err) |
| 215 | } |
| 216 | res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) |
| 217 | if res.Outcome != SwapDone { |
| 218 | t.Fatalf("Swap = %+v", res) |
| 219 | } |
| 220 | waitFor(t, "the interrupt's failed settle", func() bool { n, _ := cb.unit.stopped(); return n >= 1 }) |
| 221 | pinned := func() bool { return slices.Contains(readPinPaths(t, path, f.host), f.incumbent.StagedPath) } |
| 222 | if err := f.a.persistGenerations(); err != nil { |
| 223 | t.Fatal(err) |
| 224 | } |
| 225 | _, members := f.a.genStateForTest(f.host) |
| 226 | if len(members) != 0 { |
| 227 | t.Fatalf("the retiring generation is still a member: %v", members) |
| 228 | } |
| 229 | if !pinned() { |
| 230 | t.Fatal("the retired build went from the generations file while its run was unsettled") |
| 231 | } |
| 232 | cb.unit.setSettleErr(nil) |
| 233 | if err := h.Stop(context.Background()); err != nil { |
| 234 | t.Fatalf("Stop: %v", err) |
| 235 | } |
| 236 | waitFor(t, "the pin released", func() bool { return !pinned() }) |
| 237 | h.Release() |
| 238 | } |
| 239 | |
| 240 | // A run waiting for a switch carries its own deadline and cancellation: |
| 241 | // cancelled or expired while the successor is held, it starts nothing, |
| 242 | // before or after the switch, and returns its context's error. A Recheck |
| 243 | // that refuses after the wait starts nothing either (re-review 5). |
| 244 | func TestARunEndedDuringASwitchWaitNeverStarts(t *testing.T) { |
| 245 | t.Parallel() |
| 246 | for _, c := range []struct { |
| 247 | name string |
| 248 | ctx func() (context.Context, context.CancelFunc) |
| 249 | cancel bool |
| 250 | recheck error |
| 251 | want error |
| 252 | }{ |
| 253 | {name: "cancelled", ctx: func() (context.Context, context.CancelFunc) { return context.WithCancel(context.Background()) }, cancel: true, want: context.Canceled}, |
| 254 | {name: "expired", ctx: func() (context.Context, context.CancelFunc) { |
| 255 | return context.WithTimeout(context.Background(), 3*cmdSwitchPoll) |
| 256 | }, want: context.DeadlineExceeded}, |
| 257 | {name: "recheck refuses", ctx: func() (context.Context, context.CancelFunc) { return context.WithCancel(context.Background()) }, |
| 258 | recheck: errors.New("the record no longer names the run"), want: nil}, |
| 259 | } { |
| 260 | t.Run(c.name, func(t *testing.T) { |
| 261 | f, cb := newCmdSwapFixture(t) |
| 262 | f.b.gmu.Lock() |
| 263 | f.b.genGate = make(chan struct{}) |
| 264 | f.b.gmu.Unlock() |
| 265 | swapped := make(chan SwapResult, 1) |
| 266 | go func() { swapped <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }() |
| 267 | waitFor(t, "the successor in the operation set", func() bool { |
| 268 | _, members := f.a.genStateForTest(f.host) |
| 269 | return len(members) == 1 |
| 270 | }) |
| 271 | ctx, cancel := c.ctx() |
| 272 | defer cancel() |
| 273 | var rechecks int |
| 274 | spec := CmdSpec{RunID: "run_a"} |
| 275 | if c.recheck != nil { |
| 276 | spec.Recheck = func(context.Context) error { rechecks++; return c.recheck } |
| 277 | } |
| 278 | done := make(chan error, 1) |
| 279 | go func() { |
| 280 | _, err := f.a.RunCmd(ctx, testRoute(f.host), spec) |
| 281 | done <- err |
| 282 | }() |
| 283 | waitFor(t, "the run waiting", func() bool { |
| 284 | f.a.mu.Lock() |
| 285 | defer f.a.mu.Unlock() |
| 286 | e := f.a.entries[f.host] |
| 287 | return e != nil && e.cmd != nil |
| 288 | }) |
| 289 | time.Sleep(cmdSwitchPoll) |
| 290 | if c.cancel { |
| 291 | cancel() |
| 292 | } |
| 293 | if c.want != nil { |
| 294 | if err := <-done; !errors.Is(err, c.want) { |
| 295 | t.Fatalf("RunCmd = %v, want %v", err, c.want) |
| 296 | } |
| 297 | } |
| 298 | f.b.gmu.Lock() |
| 299 | close(f.b.genGate) |
| 300 | f.b.gmu.Unlock() |
| 301 | if res := <-swapped; res.Outcome != SwapDone { |
| 302 | t.Fatalf("Swap = %+v", res) |
| 303 | } |
| 304 | if c.want == nil { |
| 305 | if err := <-done; !errors.Is(err, c.recheck) || rechecks != 1 { |
| 306 | t.Fatalf("RunCmd = %v after %d rechecks, want the recheck's refusal after one", err, rechecks) |
| 307 | } |
| 308 | } |
| 309 | time.Sleep(2 * cmdSwitchPoll) |
| 310 | if n := len(cb.started()); n != 0 { |
| 311 | t.Fatalf("%d units started for a run that ended before its launch", n) |
| 312 | } |
| 313 | f.a.mu.Lock() |
| 314 | attached := f.a.entries[f.host].cmd != nil |
| 315 | f.a.mu.Unlock() |
| 316 | if attached { |
| 317 | t.Fatal("the ended run is still attached to the instance") |
| 318 | } |
| 319 | }) |
| 320 | } |
| 321 | } |
| 322 | |
| 323 | // A run on a generation a swap retires, whose interruption's stop fails |
| 324 | // once while the instance stays awake: the sweeps alone retry the settle, |
| 325 | // actStopTimeout apart, prove it, drop the retired build's pin, and the |
| 326 | // next run is admitted. The run's handle and the pin stay until then |
| 327 | // (re-review 6). |
| 328 | func TestARetiredRunsFailedStopIsSettledBySweeps(t *testing.T) { |
| 329 | t.Parallel() |
| 330 | for _, released := range []bool{false, true} { |
| 331 | t.Run(map[bool]string{false: "attached", true: "released"}[released], func(t *testing.T) { |
| 332 | f, cb := newCmdSwapFixture(t) |
| 333 | path := release.GenerationsPath(t.TempDir()) |
| 334 | f.a.GenerationsFile = path |
| 335 | cb.unit.setSettleErr(errors.New("instance control: 502 mid-roll")) |
| 336 | h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"}) |
| 337 | if err != nil { |
| 338 | t.Fatal(err) |
| 339 | } |
| 340 | if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone { |
| 341 | t.Fatalf("Swap = %+v", res) |
| 342 | } |
| 343 | waitFor(t, "the interruption's failed stop", func() bool { |
| 344 | n, _ := cb.unit.stopped() |
| 345 | return n == 1 && h.(*cmdRun).h.settlingDone() |
| 346 | }) |
| 347 | if why, _ := h.Interrupted(); why != StopRelease { |
| 348 | t.Fatalf("Interrupted() = %q, want %q", why, StopRelease) |
| 349 | } |
| 350 | if released { |
| 351 | h.Release() // the worker wrote 125 and let go, its own Stop failed too |
| 352 | } |
| 353 | if _, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_b"}); !errors.Is(err, ErrCmdBusy) { |
| 354 | t.Fatalf("a run while the retired one is unproven: %v, want ErrCmdBusy", err) |
| 355 | } |
| 356 | pinned := func() bool { |
| 357 | if err := f.a.persistGenerations(); err != nil { |
| 358 | t.Fatal(err) |
| 359 | } |
| 360 | return slices.Contains(readPinPaths(t, path, f.host), f.incumbent.StagedPath) |
| 361 | } |
| 362 | if !pinned() { |
| 363 | t.Fatal("the retired build lost its pin while its run was unproven") |
| 364 | } |
| 365 | cb.unit.setSettleErr(nil) |
| 366 | f.a.sweepOnce() // within actStopTimeout of the failure: not yet |
| 367 | time.Sleep(50 * time.Millisecond) |
| 368 | if n, _ := cb.unit.stopped(); n != 1 { |
| 369 | t.Fatalf("the sweep retried %d times inside actStopTimeout", n-1) |
| 370 | } |
| 371 | f.clk.advance(actStopTimeout) |
| 372 | f.a.sweepOnce() |
| 373 | waitFor(t, "the sweep's retry proving the run stopped", func() bool { return h.(*cmdRun).h.provenSettled() }) |
| 374 | if n, _ := cb.unit.stopped(); n != 2 { |
| 375 | t.Fatalf("stops %d, want the failed one and the sweep's", n) |
| 376 | } |
| 377 | if pinned() { |
| 378 | t.Fatal("the retired build is still pinned after its run was proven stopped") |
| 379 | } |
| 380 | if !released { |
| 381 | h.Release() |
| 382 | } |
| 383 | h2, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_b"}) |
| 384 | if err != nil { |
| 385 | t.Fatalf("the next run: %v", err) |
| 386 | } |
| 387 | h2.Release() |
| 388 | if f.a.entryCount() != 1 { |
| 389 | t.Fatal("the instance stopped") |
| 390 | } |
| 391 | }) |
| 392 | } |
| 393 | } |
| 394 | |
| 395 | // settlingDone reports whether no settle attempt is in flight. |
| 396 | func (h *cmdHold) settlingDone() bool { |
| 397 | h.mu.Lock() |
| 398 | defer h.mu.Unlock() |
| 399 | return h.settling == nil |
| 400 | } |
| 401 | |
| 402 | // readPinPaths is host's staged paths in the generations file. A read |
| 403 | // that meets a write in progress (another goroutine republishing) is |
| 404 | // tried again, as root would on its next pass. |
| 405 | func readPinPaths(t *testing.T, path, host string) []string { |
| 406 | t.Helper() |
| 407 | deadline := time.Now().Add(5 * time.Second) |
| 408 | for { |
| 409 | g, err := release.ReadGenerations(path) |
| 410 | if err == nil && g != nil { |
| 411 | return g.Paths[host] |
| 412 | } |
| 413 | if time.Now().After(deadline) { |
| 414 | t.Fatalf("ReadGenerations: %v", err) |
| 415 | } |
| 416 | time.Sleep(10 * time.Millisecond) |
| 417 | } |
| 418 | } |
| 419 | |
| 420 | // buildRecorder is a Service.BeforeExec that keeps every record in order: |
| 421 | // its last entry for a host is that host's last_executed. |
| 422 | type buildRecorder struct { |
| 423 | mu sync.Mutex |
| 424 | builds []string |
| 425 | err error |
| 426 | hook func() // runs inside the write, when set |
| 427 | } |
| 428 | |
| 429 | func (r *buildRecorder) record(_ context.Context, _, build string) error { |
| 430 | r.mu.Lock() |
| 431 | hook, err := r.hook, r.err |
| 432 | if err == nil { |
| 433 | r.builds = append(r.builds, build) |
| 434 | } |
| 435 | r.mu.Unlock() |
| 436 | if hook != nil { |
| 437 | hook() |
| 438 | } |
| 439 | return err |
| 440 | } |
| 441 | |
| 442 | func (r *buildRecorder) last() string { |
| 443 | r.mu.Lock() |
| 444 | defer r.mu.Unlock() |
| 445 | if len(r.builds) == 0 { |
| 446 | return "" |
| 447 | } |
| 448 | return r.builds[len(r.builds)-1] |
| 449 | } |
| 450 | |
| 451 | // A successor B records its build and then fails its health check, so the |
| 452 | // incumbent A goes on serving under a build state that names B. A command |
| 453 | // then runs A: its build is recorded before its unit starts, so the build |
| 454 | // state names A again, as it does for the sidecar (settings spec §4.2; |
| 455 | // second merge review, finding 1). |
| 456 | func TestARunRecordsItsServingBuildBeforeItStarts(t *testing.T) { |
| 457 | f, cb := newCmdSwapFixture(t) |
| 458 | rec := &buildRecorder{} |
| 459 | f.a.BeforeExec = rec.record |
| 460 | f.a.genHealth = func(context.Context, Gen) error { return errors.New("502 from the successor") } |
| 461 | if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapFailed || !res.Executed { |
| 462 | t.Fatalf("Swap = %+v, want the successor to have run and failed", res) |
| 463 | } |
| 464 | if got := rec.last(); got != "v2" { |
| 465 | t.Fatalf("after the failed successor the build state names %q, want its v2", got) |
| 466 | } |
| 467 | if cur, _ := f.a.genStateForTest(f.host); cur.N != f.incumbent.N { |
| 468 | t.Fatalf("generation %d is serving, want the incumbent", cur.N) |
| 469 | } |
| 470 | // What the build state names at the moment the unit starts. |
| 471 | var atStart string |
| 472 | cb.cmu.Lock() |
| 473 | cb.onStart = func() { atStart = rec.last() } |
| 474 | cb.cmu.Unlock() |
| 475 | h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"}) |
| 476 | if err != nil { |
| 477 | t.Fatal(err) |
| 478 | } |
| 479 | defer h.Release() |
| 480 | if got := rec.last(); got != f.incumbent.Build { |
| 481 | t.Fatalf("after the command the build state names %q, want the serving %q", got, f.incumbent.Build) |
| 482 | } |
| 483 | if specs := cb.started(); len(specs) != 1 || specs[0].serving == nil || specs[0].serving.Path != f.incumbent.StagedPath { |
| 484 | t.Fatalf("the command started %+v, want one unit on the incumbent's build", specs) |
| 485 | } |
| 486 | if atStart != f.incumbent.Build { |
| 487 | t.Fatalf("the unit started with the build state naming %q: the record must come first", atStart) |
| 488 | } |
| 489 | } |
| 490 | |
| 491 | // A build state that cannot be written launches no command (settings spec |
| 492 | // §4.2): the run is refused, the hold is released, and the instance and its |
| 493 | // next run are unaffected. A stop, or the run's own cancel, that lands |
| 494 | // during the write starts nothing either. |
| 495 | func TestARunWhoseBuildRecordFailsOrIsOvertakenNeverStarts(t *testing.T) { |
| 496 | t.Parallel() |
| 497 | t.Run("the write fails", func(t *testing.T) { |
| 498 | f, cb := newCmdSwapFixture(t) |
| 499 | rec := &buildRecorder{err: errors.New("bucket: 503")} |
| 500 | f.a.BeforeExec = rec.record |
| 501 | if _, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"}); !errors.Is(err, errBuildStateUnwritten) { |
| 502 | t.Fatalf("RunCmd = %v, want errBuildStateUnwritten", err) |
| 503 | } |
| 504 | if n := len(cb.started()); n != 0 { |
| 505 | t.Fatalf("%d units started over an unwritten build state", n) |
| 506 | } |
| 507 | rec.mu.Lock() |
| 508 | rec.err = nil |
| 509 | rec.mu.Unlock() |
| 510 | h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_b"}) |
| 511 | if err != nil { |
| 512 | t.Fatalf("the next run: %v (the refused one must leave no hold)", err) |
| 513 | } |
| 514 | h.Release() |
| 515 | }) |
| 516 | t.Run("a stop lands during the write", func(t *testing.T) { |
| 517 | f, cb := newCmdSwapFixture(t) |
| 518 | rec := &buildRecorder{} |
| 519 | rec.hook = func() { f.a.Bounce(f.host) } |
| 520 | f.a.BeforeExec = rec.record |
| 521 | _, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"}) |
| 522 | var se *CmdStoppedError |
| 523 | if !errors.As(err, &se) || se.Reason != StopRestart { |
| 524 | t.Fatalf("RunCmd = %v, want CmdStoppedError{%q}", err, StopRestart) |
| 525 | } |
| 526 | if n := len(cb.started()); n != 0 { |
| 527 | t.Fatalf("%d units started under a stopping instance", n) |
| 528 | } |
| 529 | rec.mu.Lock() |
| 530 | rec.hook = nil |
| 531 | rec.mu.Unlock() |
| 532 | waitFor(t, "the instance stopped", func() bool { return f.a.entryCount() == 0 }) |
| 533 | }) |
| 534 | t.Run("the run is cancelled during the write", func(t *testing.T) { |
| 535 | f, cb := newCmdSwapFixture(t) |
| 536 | ctx, cancel := context.WithCancel(context.Background()) |
| 537 | defer cancel() |
| 538 | rec := &buildRecorder{hook: cancel} |
| 539 | f.a.BeforeExec = rec.record |
| 540 | if _, err := f.a.RunCmd(ctx, testRoute(f.host), CmdSpec{RunID: "run_a"}); !errors.Is(err, context.Canceled) { |
| 541 | t.Fatalf("RunCmd = %v, want context.Canceled", err) |
| 542 | } |
| 543 | if n := len(cb.started()); n != 0 { |
| 544 | t.Fatalf("%d units started for a cancelled run", n) |
| 545 | } |
| 546 | f.a.mu.Lock() |
| 547 | attached := f.a.entries[f.host].cmd != nil |
| 548 | f.a.mu.Unlock() |
| 549 | if attached { |
| 550 | t.Fatal("the cancelled run is still attached") |
| 551 | } |
| 552 | }) |
| 553 | } |
| 554 | |
amadan