host: a command that finished before an instance stop keeps its exit (final review M4)
An instance stop landing between the unit's main process exiting and the run's own handle.Stop recorded its reason first, and the finished command was reported interrupted (125). The activator's hold now records when the first stop was asked (CmdHandle.Interrupted returns the time too), the run stamps when the done status arrived, and a stop asked after it is ignored: the done status wins.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4 files changed,
+80
−25
internal/activator/cmd.go+15 −11internal/activator/cmd_test.go+25 −3internal/host/runexec.go+21 −10internal/host/runexec_linux_test.go+19 −1
CI passed — run details
diff --git a/internal/activator/cmd.go b/internal/activator/cmd.go| index 382a2e9ee..df9c298ca 100644 |
| --- a/internal/activator/cmd.go |
| +++ b/internal/activator/cmd.go |
| @@ -74,8 +74,9 @@ type CmdHandle interface { |
| Invocation() string |
| Poll(ctx context.Context) (instancectl.CmdStatus, error) |
| // Interrupted is the instance stop's reason when one stopped the unit, |
| - // "" otherwise. |
| - Interrupted() string |
| + // "" otherwise, and when that stop was asked for: a command whose done |
| + // status the caller had before then finished on its own. |
| + Interrupted() (reason string, at time.Time) |
| // Stop stops the unit and waits for it to settle and its streams to |
| // end, leaving the instance running: a cancel, a timeout, or the end |
| // of a command that exited. Safe to call more than once. |
| @@ -96,21 +97,24 @@ type cmdHold struct { |
| cancelPrepare context.CancelFunc |
| started chan struct{} // closed once StartCmd returned, either way |
| unit CmdUnit // written before started is closed; nil when no unit can be running |
| + now func() time.Time |
| mu sync.Mutex |
| asked bool // a stop has been asked for; under mu |
| reason string |
| + askedAt time.Time // when the first stop was asked for; under mu |
| stopOnce sync.Once |
| settled chan struct{} |
| stopErr error |
| } |
| -func newCmdHold(runID string, cancelPrepare context.CancelFunc) *cmdHold { |
| - return &cmdHold{runID: runID, cancelPrepare: cancelPrepare, started: make(chan struct{}), settled: make(chan struct{})} |
| +func newCmdHold(runID string, cancelPrepare context.CancelFunc, now func() time.Time) *cmdHold { |
| + return &cmdHold{runID: runID, cancelPrepare: cancelPrepare, now: now, started: make(chan struct{}), settled: make(chan struct{})} |
| } |
| -// stop waits for the start to return, records reason if it is the first |
| -// stop asked for (the run's own Stop passes "", so an instance stop after |
| -// it is not what ended the run), and stops and settles the unit once. |
| +// stop waits for the start to return, records reason and the time if it is |
| +// the first stop asked for (the run's own Stop passes "", so an instance |
| +// stop after it is not what ended the run), and stops and settles the unit |
| +// once. |
| func (h *cmdHold) stop(ctx context.Context, reason string) error { |
| select { |
| case <-h.started: |
| @@ -120,7 +124,7 @@ func (h *cmdHold) stop(ctx context.Context, reason string) error { |
| h.mu.Lock() |
| if !h.asked { |
| h.asked = true |
| - h.reason = reason |
| + h.reason, h.askedAt = reason, h.now() |
| } |
| h.mu.Unlock() |
| if h.unit == nil { |
| @@ -172,10 +176,10 @@ func (r *cmdRun) Invocation() string { return r.h.unit.Invocation() } |
| func (r *cmdRun) Poll(ctx context.Context) (instancectl.CmdStatus, error) { |
| return r.h.unit.Poll(ctx) |
| } |
| -func (r *cmdRun) Interrupted() string { |
| +func (r *cmdRun) Interrupted() (string, time.Time) { |
| r.h.mu.Lock() |
| defer r.h.mu.Unlock() |
| - return r.h.reason |
| + return r.h.reason, r.h.askedAt |
| } |
| func (r *cmdRun) Stop(ctx context.Context) error { return r.h.stop(ctx, "") } |
| @@ -258,7 +262,7 @@ func (a *Service) RunCmd(ctx context.Context, rt registry.Route, spec CmdSpec) ( |
| } |
| pctx, pcancel := context.WithCancel(ctx) |
| defer pcancel() |
| - h := newCmdHold(spec.RunID, pcancel) |
| + h := newCmdHold(spec.RunID, pcancel, a.now) |
| a.mu.Lock() |
| e := a.entries[rt.Host] |
| switch { |
diff --git a/internal/activator/cmd_test.go b/internal/activator/cmd_test.go| index a7f61c5b7..131804679 100644 |
| --- a/internal/activator/cmd_test.go |
| +++ b/internal/activator/cmd_test.go |
| @@ -254,13 +254,35 @@ func TestEveryStopPathSettlesTheRunBeforeReplicationCloses(t *testing.T) { |
| if !cb.lastRep().isClosed() { |
| t.Fatal("replication never closed") |
| } |
| - if got := h.Interrupted(); got != c.why { |
| - t.Fatalf("Interrupted() = %q, want %q", got, c.why) |
| + if got, at := h.Interrupted(); got != c.why || at.IsZero() { |
| + t.Fatalf("Interrupted() = %q at %v, want %q and when", got, at, c.why) |
| } |
| }) |
| } |
| } |
| +// The hold stamps the first stop asked for with the activator's clock, so |
| +// the run can tell an instance stop that came after its command finished. |
| +func TestRunCmdInterruptedSaysWhenTheStopWasAsked(t *testing.T) { |
| + a, _, clk := testCmdActivator(t, Config{}) |
| + rt := testRoute("c.example.com") |
| + h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + clk.advance(5 * time.Minute) |
| + asked := clk.now() |
| + a.Bounce(rt.Host) |
| + waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 }) |
| + clk.advance(time.Minute) |
| + if err := h.Stop(context.Background()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if why, at := h.Interrupted(); why != StopRestart || !at.Equal(asked) { |
| + t.Fatalf("Interrupted() = %q at %v, want %q at %v (the first stop's)", why, at, StopRestart, asked) |
| + } |
| +} |
| + |
| func TestRunCmdNeverStartsUnderAStoppingEntry(t *testing.T) { |
| a, cb, _ := testCmdActivator(t, Config{}) |
| rt := testRoute("c.example.com") |
| @@ -367,7 +389,7 @@ func TestAReleasedRunThatNeverSettledStillHoldsTheStop(t *testing.T) { |
| if got := a.StopUncertain(rt.Host); got != c.uncertain { |
| t.Fatalf("StopUncertain = %v, want %v", got, c.uncertain) |
| } |
| - if got := h.Interrupted(); got != "" && c.stopFirst { |
| + if got, _ := h.Interrupted(); got != "" && c.stopFirst { |
| t.Fatalf("Interrupted() = %q after the run's own stop", got) |
| } |
| }) |
diff --git a/internal/host/runexec.go b/internal/host/runexec.go| index 10f22c2b2..c55ffe885 100644 |
| --- a/internal/host/runexec.go |
| +++ b/internal/host/runexec.go |
| @@ -176,7 +176,7 @@ func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs |
| } |
| log.Printf("agent: runs: %s: run %s started (%s)", rec.Host, req.ID, handle.Invocation()) |
| res := runs.Result{Started: w.now().UTC()} |
| - status := w.follow(ctx, rec, req, handle, h, &res) |
| + status, doneAt := w.follow(ctx, rec, req, handle, h, &res) |
| bound := time.AfterFunc(runOutputGrace, endOutput) |
| defer bound.Stop() |
| // The unit is stopped and settled after every ending, done included |
| @@ -197,7 +197,13 @@ func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs |
| if status != nil && status.State == instancectl.CmdDone { |
| res.Exit, res.Signal = exitOf(*status) |
| } |
| - if reason := handle.Interrupted(); reason != "" { |
| + reason, askedAt := handle.Interrupted() |
| + if reason != "" && status != nil && status.State == instancectl.CmdDone && !doneAt.IsZero() && askedAt.After(doneAt) { |
| + // The instance stop was asked for after the done status arrived: |
| + // the command had finished on its own, and the done status wins. |
| + reason = "" |
| + } |
| + if reason != "" { |
| res.Interrupted = reason |
| } else if status != nil && status.State == instancectl.CmdGone && !res.Cancelled && !res.TimedOut { |
| res.Interrupted = runs.InterruptedGone |
| @@ -286,11 +292,16 @@ func cancelRequested(h *runHandle) bool { |
| // follow long-polls the unit until it is done or gone, the run is |
| // cancelled, or its timeout passes. It returns the final status when the |
| -// unit reported one, and marks a cancel or a timeout on res. |
| -func (w *runsWorker) follow(ctx context.Context, rec provision.Record, req runs.Request, handle activator.CmdHandle, h *runHandle, res *runs.Result) *instancectl.CmdStatus { |
| +// unit reported one, with when the poll answered it, and marks a cancel or |
| +// a timeout on res. |
| +func (w *runsWorker) follow(ctx context.Context, rec provision.Record, req runs.Request, handle activator.CmdHandle, h *runHandle, res *runs.Result) (*instancectl.CmdStatus, time.Time) { |
| pctx, stopPolling := context.WithCancel(ctx) |
| defer stopPolling() |
| - statuses := make(chan instancectl.CmdStatus, 1) |
| + type polled struct { |
| + st instancectl.CmdStatus |
| + at time.Time |
| + } |
| + statuses := make(chan polled, 1) |
| go func() { |
| for { |
| st, err := handle.Poll(pctx) |
| @@ -306,7 +317,7 @@ func (w *runsWorker) follow(ctx context.Context, rec provision.Record, req runs. |
| continue |
| } |
| if st.State != instancectl.CmdRunning { |
| - statuses <- st |
| + statuses <- polled{st, w.now()} |
| return |
| } |
| } |
| @@ -323,14 +334,14 @@ func (w *runsWorker) follow(ctx context.Context, rec provision.Record, req runs. |
| defer recheck.Stop() |
| for { |
| select { |
| - case st := <-statuses: |
| - return &st |
| + case p := <-statuses: |
| + return &p.st, p.at |
| case <-h.cancel: |
| res.Cancelled = true |
| - return nil |
| + return nil, time.Time{} |
| case <-timeout.C: |
| res.TimedOut = true |
| - return nil |
| + return nil, time.Time{} |
| case <-recheck.C: |
| rctx, cancel := context.WithTimeout(ctx, runReadBound) |
| cur, err := provision.GetRecord(rctx, w.store, rec.Account, rec.App, rec.Host) |
diff --git a/internal/host/runexec_linux_test.go b/internal/host/runexec_linux_test.go| index 6749ce62c..032b83237 100644 |
| --- a/internal/host/runexec_linux_test.go |
| +++ b/internal/host/runexec_linux_test.go |
| @@ -26,6 +26,7 @@ type fakeHandle struct { |
| stops int |
| released bool |
| interrupted string |
| + askedAt time.Time // when the interrupting stop was asked for; zero is "long ago" |
| afterStop *instancectl.CmdStatus |
| } |
| @@ -48,7 +49,11 @@ func (h *fakeHandle) Poll(ctx context.Context) (instancectl.CmdStatus, error) { |
| return instancectl.CmdStatus{}, ctx.Err() |
| } |
| } |
| -func (h *fakeHandle) Interrupted() string { h.mu.Lock(); defer h.mu.Unlock(); return h.interrupted } |
| +func (h *fakeHandle) Interrupted() (string, time.Time) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + return h.interrupted, h.askedAt |
| +} |
| func (h *fakeHandle) Stop(context.Context) error { |
| h.mu.Lock() |
| h.stops++ |
| @@ -427,6 +432,19 @@ func TestRunInterruptedByAnInstanceStop(t *testing.T) { |
| } |
| } |
| +// An instance stop that lands after the command's done status arrived, but |
| +// before the run's own stop, did not end the command: the result is the |
| +// command's exit, not interrupted. |
| +func TestRunFinishedBeforeAnInstanceStopKeepsItsExit(t *testing.T) { |
| + s := release.NewDirStore(t.TempDir()) |
| + rec, req := pickedRun(t, s, 60) |
| + h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1), interrupted: activator.StopRestart, askedAt: time.Now().Add(time.Hour)} |
| + h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Status: "0", Code: "1"} |
| + if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Interrupted != "" || res.Exit != 0 { |
| + t.Fatalf("result %+v, want exit 0 and not interrupted", res) |
| + } |
| +} |
| + |
| func TestRunWhoseUnitWentAway(t *testing.T) { |
| s := release.NewDirStore(t.TempDir()) |
| rec, req := pickedRun(t, s, 60) |