amadan

carlos / platform Public

merge fix 5: a run's deadline and cancel span its whole admission

Re-review finding 5. M2's wait for a swap's switch came after Prepare,
which made the run's last cancel and ownership checks. The worker handed
RunCmd context.Background(), so a run cancelled or expired during the
wait still started once the successor switched.

- The worker runs RunCmd under a launch context: the record's run
  deadline, ended by the run's cancel. That covers the wake, Prepare,
  any wait for a switch and the last checks.
- RunCmd's admission loop checks the run's context, ownership (the
  entry is the host's, awake with its lease, no fence or shutdown) and
  that the current generation's process is running, before anything
  starts. Each ownership refusal becomes the stop reason it stands for
  (RUN-C1 to RUN-C8).
- After any wait for a switch, the new CmdSpec.Recheck reads the run's
  eligibility again: the record still names the run as this box's, not
  cancelled, the same incarnation. Then every check runs once more.
- An admission ended by the run's own cancel or deadline gets the run's
  own result, cancelled or timed out. A Prepare that failed keeps its own
  result as before.

Tests: in the activator, a run cancelled, expired, or refused by its
recheck while a successor is held starts nothing, and nothing starts
after the switch either. In the worker, a run cancelled or expired while
RunCmd waits ends cancelled or timed out, with no unit.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 6eb51e91f87cd74121cca048f0f087c568e531f2 parent 4cb605c
5 files changed, +241 −14
  • internal/activator/cmd.go +63 −12
  • internal/activator/cmd_swap_test.go +82 −0
  • internal/activator/cmdunit.go +5 −0
  • internal/host/runexec.go +38 −2
  • internal/host/runexec_linux_test.go +53 −0
diff --git a/internal/activator/cmd.go b/internal/activator/cmd.go
index 303f27724..476d7bbec 100644
--- a/internal/activator/cmd.go
+++ b/internal/activator/cmd.go
@@ -311,6 +311,27 @@ func switchPendingLocked(e *actEntry) *genMember {
return nil
}
+// cmdOwnershipLocked is why e must not start a run now, or "": the entry
+// is the host's and awake with its lease (ownedLocked), and its current
+// generation's process is running. Each refusal is the stop reason it
+// will become (re-review 5). The caller holds a.mu.
+func cmdOwnershipLocked(a *Service, e *actEntry) string {
+ switch err := a.ownedLocked(e, false); {
+ case errors.Is(err, errSwapLeaseLost):
+ return StopLeaseLost
+ case errors.Is(err, errSwapFenced):
+ return StopFenced
+ case errors.Is(err, errSwapShutdown):
+ return StopShutdown
+ case err != nil:
+ return StopIdle
+ }
+ if e.proc == nil || e.proc.Exited() {
+ return StopExited
+ }
+ return ""
+}
+
// cmdSwitchPoll is how often RunCmd re-checks a pending switch it waits on,
// besides the successor's own done.
var cmdSwitchPoll = 100 * time.Millisecond
@@ -373,6 +394,11 @@ func (a *Service) settleCmds(e *actEntry) bool {
// here (CmdStoppedError, no unit). One that begins after it waits in
// finishStop for the start to return and then stops the unit before
// replication closes.
+//
+// ctx is the run's own: its deadline and cancellation span the whole
+// admission, the wake, Prepare, any wait for a swap's switch and the last
+// checks (re-review 5). A run whose ctx ends before the start returns
+// ctx's error and starts nothing.
func (a *Service) RunCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdHandle, error) {
if a.cfg.HostWrites.Enabled {
return nil, ErrCmdClaim
@@ -443,6 +469,10 @@ func (a *Service) RunCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (
prepErr = spec.Prepare(pctx)
}
a.mu.Lock()
+ // waited: the run has waited for a switch since launch eligibility was
+ // last checked, by Prepare or by spec.Recheck. A wait can be long, so
+ // eligibility is checked again before anything starts (re-review 5).
+ waited := false
for {
if e.state != actAwake {
why := e.stopWhy
@@ -453,26 +483,47 @@ func (a *Service) RunCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (
a.mu.Unlock()
return abandon(prepErr)
}
+ // ctx carries the run's deadline and cancellation (the box's run
+ // pass derives it from the record): a run cancelled or expired at
+ // any point of its admission never starts (re-review 5).
+ if err := ctx.Err(); err != nil {
+ a.mu.Unlock()
+ return abandon(err)
+ }
+ // Ownership, and a generation that is serving: the lease is held,
+ // no fence or shutdown holds the host, and the current
+ // generation's process is running.
+ if why := cmdOwnershipLocked(a, e); why != "" {
+ a.mu.Unlock()
+ return abandon(&CmdStoppedError{Reason: why})
+ }
// An overlap successor is on its way: the current generation may
// retire before the unit is even up, so nothing starts against it.
// The run waits for the switch (and then runs the successor's
// build) or for the successor's end (merge ruling M2). A stop
- // cancels pctx, and the check above then abandons the run.
- m := switchPendingLocked(e)
- if m == nil {
+ // cancels pctx, as do the run's own cancel and deadline, and the
+ // checks above then abandon the run.
+ if m := switchPendingLocked(e); m != nil {
+ a.mu.Unlock()
+ select {
+ case <-m.done:
+ case <-time.After(cmdSwitchPoll):
+ case <-pctx.Done():
+ }
+ waited = true
+ a.mu.Lock()
+ continue
+ }
+ if !waited || spec.Recheck == nil {
break
}
+ // The run's own eligibility, which Prepare checked before the
+ // wait, read again off a.mu (the record), then every check above
+ // once more, since the entry may have moved meanwhile.
a.mu.Unlock()
- select {
- case <-m.done:
- case <-time.After(cmdSwitchPoll):
- case <-pctx.Done():
- }
+ prepErr = spec.Recheck(pctx)
+ waited = false
a.mu.Lock()
- if pctx.Err() != nil && e.state == actAwake {
- a.mu.Unlock()
- return abandon(pctx.Err())
- }
}
// Admitted against the current generation, under the lock a swap's
// admission and switch take (merge ruling M2): no successor is live,
diff --git a/internal/activator/cmd_swap_test.go b/internal/activator/cmd_swap_test.go
index 2d9d8250b..a46d11b60 100644
--- a/internal/activator/cmd_swap_test.go
+++ b/internal/activator/cmd_swap_test.go
@@ -235,3 +235,85 @@ func TestARetiredRunKeepsItsBuildPinnedUntilItSettles(t *testing.T) {
waitFor(t, "the pin released", func() bool { return !pinned() })
h.Release()
}
+
+// A run waiting for a switch carries its own deadline and cancellation:
+// cancelled or expired while the successor is held, it starts nothing,
+// before or after the switch, and returns its context's error. A Recheck
+// that refuses after the wait starts nothing either (re-review 5).
+func TestARunEndedDuringASwitchWaitNeverStarts(t *testing.T) {
+ for _, c := range []struct {
+ name string
+ ctx func() (context.Context, context.CancelFunc)
+ cancel bool
+ recheck error
+ want error
+ }{
+ {name: "cancelled", ctx: func() (context.Context, context.CancelFunc) { return context.WithCancel(context.Background()) }, cancel: true, want: context.Canceled},
+ {name: "expired", ctx: func() (context.Context, context.CancelFunc) {
+ return context.WithTimeout(context.Background(), 3*cmdSwitchPoll)
+ }, want: context.DeadlineExceeded},
+ {name: "recheck refuses", ctx: func() (context.Context, context.CancelFunc) { return context.WithCancel(context.Background()) },
+ recheck: errors.New("the record no longer names the run"), want: nil},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ f, cb := newCmdSwapFixture(t)
+ f.b.gmu.Lock()
+ f.b.genGate = make(chan struct{})
+ f.b.gmu.Unlock()
+ swapped := make(chan SwapResult, 1)
+ go func() { swapped <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }()
+ waitFor(t, "the successor in the operation set", func() bool {
+ _, members := f.a.genStateForTest(f.host)
+ return len(members) == 1
+ })
+ ctx, cancel := c.ctx()
+ defer cancel()
+ var rechecks int
+ spec := CmdSpec{RunID: "run_a"}
+ if c.recheck != nil {
+ spec.Recheck = func(context.Context) error { rechecks++; return c.recheck }
+ }
+ done := make(chan error, 1)
+ go func() {
+ _, err := f.a.RunCmd(ctx, testRoute(f.host), spec)
+ done <- err
+ }()
+ waitFor(t, "the run waiting", func() bool {
+ f.a.mu.Lock()
+ defer f.a.mu.Unlock()
+ e := f.a.entries[f.host]
+ return e != nil && e.cmd != nil
+ })
+ time.Sleep(cmdSwitchPoll)
+ if c.cancel {
+ cancel()
+ }
+ if c.want != nil {
+ if err := <-done; !errors.Is(err, c.want) {
+ t.Fatalf("RunCmd = %v, want %v", err, c.want)
+ }
+ }
+ f.b.gmu.Lock()
+ close(f.b.genGate)
+ f.b.gmu.Unlock()
+ if res := <-swapped; res.Outcome != SwapDone {
+ t.Fatalf("Swap = %+v", res)
+ }
+ if c.want == nil {
+ if err := <-done; !errors.Is(err, c.recheck) || rechecks != 1 {
+ t.Fatalf("RunCmd = %v after %d rechecks, want the recheck's refusal after one", err, rechecks)
+ }
+ }
+ time.Sleep(2 * cmdSwitchPoll)
+ if n := len(cb.started()); n != 0 {
+ t.Fatalf("%d units started for a run that ended before its launch", n)
+ }
+ f.a.mu.Lock()
+ attached := f.a.entries[f.host].cmd != nil
+ f.a.mu.Unlock()
+ if attached {
+ t.Fatal("the ended run is still attached to the instance")
+ }
+ })
+ }
+}
diff --git a/internal/activator/cmdunit.go b/internal/activator/cmdunit.go
index a116402cf..cb18b81ef 100644
--- a/internal/activator/cmdunit.go
+++ b/internal/activator/cmdunit.go
@@ -46,6 +46,11 @@ type CmdSpec struct {
// Prepare runs after the run is attached to the awake entry and before
// the unit starts (Service.RunCmd); an error ends the run before launch.
Prepare func(ctx context.Context) error
+ // Recheck, when set, is the run's launch eligibility checked again
+ // after RunCmd has waited for a swap's switch (re-review 5): Prepare's
+ // last check is that old by then. An error ends the run before launch,
+ // as Prepare's does.
+ Recheck func(ctx context.Context) error
// serving is the serving generation's staged build, set by RunCmd at its
// last awake check when generations are on and that generation has one
diff --git a/internal/host/runexec.go b/internal/host/runexec.go
index c55ffe885..9d23558e4 100644
--- a/internal/host/runexec.go
+++ b/internal/host/runexec.go
@@ -142,6 +142,9 @@ func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs
// overlapped Prepare in place of Prepare's error, and a run the record
// no longer gives this box must get no result whatever RunCmd says.
notOurs := false
+ // prepared: Prepare (or the Recheck after a switch wait) answered, so
+ // its own error, not the launch context's end, is the run's story.
+ prepared := false
spec := activator.CmdSpec{
RunID: req.ID,
Args: req.Argv,
@@ -150,11 +153,23 @@ func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs
Stderr: chunks.Stream(runs.StreamErr),
Prepare: func(ctx context.Context) error {
err := w.prepare(ctx, instDir, req, h, launchDeadline(rec, req, w.now()))
- notOurs = errors.Is(err, errRunNotOurs)
+ notOurs, prepared = errors.Is(err, errRunNotOurs), err != nil
+ return err
+ },
+ Recheck: func(ctx context.Context) error {
+ err := w.launchable(ctx, req, h)
+ notOurs, prepared = errors.Is(err, errRunNotOurs), err != nil
return err
},
}
- handle, err := w.launch.RunCmd(ctx, rt, spec)
+ // The launch runs under the run's deadline and its cancel, the whole
+ // admission through (re-review 5): the wake, Prepare, a wait for a
+ // swap's switch and the last checks. A run cancelled or expired at
+ // any of them starts nothing.
+ lctx, lcancel := w.launchContext(ctx, rec, req, h)
+ handle, err := w.launch.RunCmd(lctx, rt, spec)
+ launchEnded := lctx.Err()
+ lcancel()
if err != nil {
bound := time.AfterFunc(runOutputGrace, endOutput)
defer bound.Stop()
@@ -167,6 +182,12 @@ func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs
return
}
res := runResultForLaunch(err)
+ if launchEnded != nil && !prepared {
+ // The admission ended with the run's own cancel or deadline:
+ // the run's own result, as after launch (re-review 5).
+ cancelled := cancelRequested(h)
+ res = runs.Result{Cancelled: cancelled, TimedOut: !cancelled}
+ }
if res.Interrupted == runs.InterruptedStartFailed {
log.Printf("agent: runs: %s %s: start: %v", rec.Host, req.ID, err)
}
@@ -212,6 +233,21 @@ func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs
w.finish(ctx, rec, req, res, handle)
}
+// launchContext is ctx bounded by the run's deadline and ended by its
+// cancel: what RunCmd runs under (re-review 5). The cancel func must be
+// called once RunCmd has returned.
+func (w *runsWorker) launchContext(ctx context.Context, rec provision.Record, req runs.Request, h *runHandle) (context.Context, context.CancelFunc) {
+ lctx, cancel := context.WithDeadline(ctx, launchDeadline(rec, req, w.now()))
+ go func() {
+ select {
+ case <-h.cancel:
+ cancel()
+ case <-lctx.Done():
+ }
+ }()
+ return lctx, cancel
+}
+
// launchDeadline is the run's deadline from the record (RunPick set it to
// the pickup plus the timeout), or now plus the timeout without one.
func launchDeadline(rec provision.Record, req runs.Request, now time.Time) time.Time {
diff --git a/internal/host/runexec_linux_test.go b/internal/host/runexec_linux_test.go
index 032b83237..002789c7a 100644
--- a/internal/host/runexec_linux_test.go
+++ b/internal/host/runexec_linux_test.go
@@ -629,6 +629,59 @@ func TestRunNotOursGetsNoResultWhenAStopOverlapsPrepare(t *testing.T) {
}
}
+// waitsForSwitch is RunCmd as it waits for a swap's switch after Prepare:
+// it holds until the launch context ends, as the activator's wait does
+// under the run's deadline and cancel, and starts nothing.
+type waitsForSwitch struct {
+ waiting chan struct{}
+ units int
+}
+
+func (l *waitsForSwitch) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
+ if err := spec.Prepare(ctx); err != nil {
+ return nil, err
+ }
+ close(l.waiting)
+ <-ctx.Done()
+ return nil, ctx.Err()
+}
+
+// A run cancelled, or past its deadline, while RunCmd waits for a swap's
+// switch ends with its own cancelled or timed-out result, and no unit
+// starts (re-review 5): the launch runs under the run's deadline and
+// cancel, not the worker's background context.
+func TestRunEndedDuringASwitchWaitGetsItsOwnResult(t *testing.T) {
+ t.Run("cancelled", func(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 60)
+ l := &waitsForSwitch{waiting: make(chan struct{})}
+ h := newRunHandle()
+ go func() { <-l.waiting; h.requestCancel() }()
+ res := runOn(t, execWorker(t, s, l, "boot-1"), s, rec, req, h)
+ if !res.Cancelled || res.TimedOut || l.units != 0 || !res.Started.IsZero() {
+ t.Fatalf("result %+v", res)
+ }
+ })
+ t.Run("expired", func(t *testing.T) {
+ dir := release.NewDirStore(t.TempDir())
+ rec, id := queueRun(t, dir, time.Now().Add(time.Minute))
+ req, err := runs.ReadRequest(context.Background(), dir, rec.Account, rec.App, id)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, err := provision.SetRecordRun(context.Background(), dir, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(300*time.Millisecond)))
+ if err != nil {
+ t.Fatal(err)
+ }
+ l := &waitsForSwitch{waiting: make(chan struct{})}
+ start := time.Now()
+ res := runOn(t, execWorker(t, dir, l, "boot-1"), dir, *got, *req, newRunHandle())
+ if !res.TimedOut || res.Cancelled || l.units != 0 || time.Since(start) > 5*time.Second {
+ t.Fatalf("result %+v after %s", res, time.Since(start))
+ }
+ })
+}
+
func TestExitOfReadsNumbersAndWords(t *testing.T) {
for _, c := range []struct {
code, status string