amadan

carlos / platform Public

merge fix 6: the sweep retries a retired generation's run until it is proven

Re-review finding 6. When a swap retired a run's generation and the
interruption's stop failed, nothing tried again while the instance stayed
awake. The sweep only settled runs for a pending stop. The worker kept
following the command to its end or its timeout, and a released run that
never settled blocked every later run on that instance.

- Each sweep (finishPendingStops, which visits every entry) calls
  retryCmdSettlesLocked, pending stop or not. A run is retried in the
  background, one attempt at a time and actStopTimeout after the last,
  when its stop was asked for and is not proven: an interruption by a
  retired generation, or its own Stop that failed. A released run that
  never settled is retried too. A running command nobody asked to stop
  is left alone. The reason the stop recorded is kept.
- The run's handle and its generation's pin stay until the proof
  succeeds. The pin goes then, and later runs are admitted again.
- The worker's follow asks Interrupted() every 250 ms. Once the run is
  interrupted, it ends as interrupted (exit 125) within a 2 s grace for
  a done status, which still wins if the command had already finished.
  It doesn't wait for the stop, which the activator keeps retrying.

Tests: a retirement stop that fails once, with the run attached and
released, is not retried inside actStopTimeout, then is proven by the
sweeps alone. The pin stays until then and goes after, and the next run
is admitted. With the sweep's call removed, this test fails. In the
worker, a run interrupted while its stop keeps failing ends interrupted
promptly, and the handle is still stopped and released. The M2 pin tests
now retry a generations-file read that meets a concurrent republish.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 1c23afa9e094e766871c52065ca602a04e3de72d parent 6eb51e9
5 files changed, +209 −10
  • internal/activator/activator.go +3 −0
  • internal/activator/cmd.go +48 −2
  • internal/activator/cmd_swap_test.go +97 −7
  • internal/host/runexec.go +25 −0
  • internal/host/runexec_linux_test.go +36 −1
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 8d7d2a5e9..332a33be4 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -3415,6 +3415,9 @@ func (a *Service) finishPendingStops() {
// A parent restore that failed when the last member left keeps its
// obligation and is retried here (final review F11).
a.retryParentRestoreLocked(e)
+ // A run whose stop was asked for and is not proven, pending stop
+ // or not (re-review 6).
+ a.retryCmdSettlesLocked(e)
if !e.pendingStop || len(e.ops.members) > 0 || !a.respawnSettledLocked(e) {
continue
}
diff --git a/internal/activator/cmd.go b/internal/activator/cmd.go
index 476d7bbec..5855dc7da 100644
--- a/internal/activator/cmd.go
+++ b/internal/activator/cmd.go
@@ -112,6 +112,9 @@ type cmdHold struct {
// the run's last word while the instance's ownership is given up.
// Under mu.
proven bool
+ // lastTry is when the last settle attempt ended, by h.now; the sweep's
+ // retry keeps actStopTimeout after it (retrySettle). Under mu.
+ lastTry time.Time
// gen is the generation the run was admitted against (merge ruling
// M2): its number and its exe pin. RunCmd sets it under a.mu with
// starting, and it is never written again. The command execs gen's
@@ -183,7 +186,7 @@ func (h *cmdHold) settle(at *cmdSettle) {
if err == nil {
h.proven = true
}
- h.settling = nil
+ h.settling, h.lastTry = nil, h.now()
h.mu.Unlock()
close(at.done)
if err == nil && h.onProven != nil {
@@ -191,6 +194,33 @@ func (h *cmdHold) settle(at *cmdSettle) {
}
}
+// retrySettle starts one settle attempt in the background, for the sweep
+// (re-review 6), when the unit has started and is not proven settled, no
+// attempt is in flight, and the last one ended at least every ago. A hold
+// still attached to its run is retried only once a stop has been asked
+// for it (an interruption, or the run's own Stop that failed): a running
+// command nobody asked to stop is left alone. A released hold is retried
+// whatever: its run is over, and only its proof is missing. The reason a
+// stop recorded is kept. It never blocks.
+func (h *cmdHold) retrySettle(released bool, every time.Duration) {
+ select {
+ case <-h.started:
+ default:
+ return
+ }
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ if h.unit == nil || h.proven || h.settling != nil || (!released && !h.asked) || h.now().Sub(h.lastTry) < every {
+ return
+ }
+ if !h.asked {
+ h.asked, h.askedAt = true, h.now()
+ }
+ at := &cmdSettle{done: make(chan struct{})}
+ h.settling = at
+ go h.settle(at)
+}
+
// provenSettled reports whether the hold leaves nothing running: the start
// returned no unit, or a settle of the unit succeeded. A failed start that
// reached the control service returns a unit, since the unit may be
@@ -352,12 +382,28 @@ func (a *Service) interruptCmdsLocked(e *actEntry, n uint64, why string) {
ctx, cancel := context.WithTimeout(context.Background(), cmdStopBound)
defer cancel()
if err := h.stop(ctx, why); err != nil {
- log.Printf("activator: %s: run %s on retired generation %d is not proven stopped: %v", e.host, h.runID, n, err)
+ log.Printf("activator: %s: run %s on retired generation %d is not proven stopped: %v; the sweep tries again", e.host, h.runID, n, err)
}
}(h)
}
}
+// retryCmdSettlesLocked is the sweep's retry of every run of e whose stop
+// was asked for and is not proven (re-review 6): a run interrupted when a
+// swap retired its generation, a run whose own Stop failed, and a released
+// run left unsettled. It runs whatever state e is in, pending stop or
+// awake, so an instance that stays awake still gets its runs proven, their
+// pins dropped and later runs admitted. Each attempt runs off a.mu, one at
+// a time per run, actStopTimeout apart. The caller holds a.mu.
+func (a *Service) retryCmdSettlesLocked(e *actEntry) {
+ for _, h := range e.cmdLeft {
+ h.retrySettle(true, actStopTimeout)
+ }
+ if e.cmd != nil && e.cmd.starting {
+ e.cmd.retrySettle(false, actStopTimeout)
+ }
+}
+
// settleCmds is a run's part of an instance's termination proof (run spec
// §5.3; merge ruling M1): every run of e whose unit may still be running is
// stopped and settled, bounded by cmdStopBound, and it reports whether none
diff --git a/internal/activator/cmd_swap_test.go b/internal/activator/cmd_swap_test.go
index a46d11b60..39277a56a 100644
--- a/internal/activator/cmd_swap_test.go
+++ b/internal/activator/cmd_swap_test.go
@@ -211,13 +211,7 @@ func TestARetiredRunKeepsItsBuildPinnedUntilItSettles(t *testing.T) {
t.Fatalf("Swap = %+v", res)
}
waitFor(t, "the interrupt's failed settle", func() bool { n, _ := cb.unit.stopped(); return n >= 1 })
- pinned := func() bool {
- g, err := release.ReadGenerations(path)
- if err != nil || g == nil {
- t.Fatalf("ReadGenerations: %v", err)
- }
- return slices.Contains(g.Paths[f.host], f.incumbent.StagedPath)
- }
+ pinned := func() bool { return slices.Contains(readPinPaths(t, path, f.host), f.incumbent.StagedPath) }
if err := f.a.persistGenerations(); err != nil {
t.Fatal(err)
}
@@ -317,3 +311,99 @@ func TestARunEndedDuringASwitchWaitNeverStarts(t *testing.T) {
})
}
}
+
+// A run on a generation a swap retires, whose interruption's stop fails
+// once while the instance stays awake: the sweeps alone retry the settle,
+// actStopTimeout apart, prove it, drop the retired build's pin, and the
+// next run is admitted. The run's handle and the pin stay until then
+// (re-review 6).
+func TestARetiredRunsFailedStopIsSettledBySweeps(t *testing.T) {
+ for _, released := range []bool{false, true} {
+ t.Run(map[bool]string{false: "attached", true: "released"}[released], func(t *testing.T) {
+ f, cb := newCmdSwapFixture(t)
+ path := release.GenerationsPath(t.TempDir())
+ f.a.GenerationsFile = path
+ cb.unit.setSettleErr(errors.New("instance control: 502 mid-roll"))
+ h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone {
+ t.Fatalf("Swap = %+v", res)
+ }
+ waitFor(t, "the interruption's failed stop", func() bool {
+ n, _ := cb.unit.stopped()
+ return n == 1 && h.(*cmdRun).h.settlingDone()
+ })
+ if why, _ := h.Interrupted(); why != StopRelease {
+ t.Fatalf("Interrupted() = %q, want %q", why, StopRelease)
+ }
+ if released {
+ h.Release() // the worker wrote 125 and let go, its own Stop failed too
+ }
+ if _, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_b"}); !errors.Is(err, ErrCmdBusy) {
+ t.Fatalf("a run while the retired one is unproven: %v, want ErrCmdBusy", err)
+ }
+ pinned := func() bool {
+ if err := f.a.persistGenerations(); err != nil {
+ t.Fatal(err)
+ }
+ return slices.Contains(readPinPaths(t, path, f.host), f.incumbent.StagedPath)
+ }
+ if !pinned() {
+ t.Fatal("the retired build lost its pin while its run was unproven")
+ }
+ cb.unit.setSettleErr(nil)
+ f.a.sweepOnce() // within actStopTimeout of the failure: not yet
+ time.Sleep(50 * time.Millisecond)
+ if n, _ := cb.unit.stopped(); n != 1 {
+ t.Fatalf("the sweep retried %d times inside actStopTimeout", n-1)
+ }
+ f.clk.advance(actStopTimeout)
+ f.a.sweepOnce()
+ waitFor(t, "the sweep's retry proving the run stopped", func() bool { return h.(*cmdRun).h.provenSettled() })
+ if n, _ := cb.unit.stopped(); n != 2 {
+ t.Fatalf("stops %d, want the failed one and the sweep's", n)
+ }
+ if pinned() {
+ t.Fatal("the retired build is still pinned after its run was proven stopped")
+ }
+ if !released {
+ h.Release()
+ }
+ h2, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_b"})
+ if err != nil {
+ t.Fatalf("the next run: %v", err)
+ }
+ h2.Release()
+ if f.a.entryCount() != 1 {
+ t.Fatal("the instance stopped")
+ }
+ })
+ }
+}
+
+// settlingDone reports whether no settle attempt is in flight.
+func (h *cmdHold) settlingDone() bool {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ return h.settling == nil
+}
+
+// readPinPaths is host's staged paths in the generations file. A read
+// that meets a write in progress (another goroutine republishing) is
+// tried again, as root would on its next pass.
+func readPinPaths(t *testing.T, path, host string) []string {
+ t.Helper()
+ deadline := time.Now().Add(5 * time.Second)
+ for {
+ g, err := release.ReadGenerations(path)
+ if err == nil && g != nil {
+ return g.Paths[host]
+ }
+ if time.Now().After(deadline) {
+ t.Fatalf("ReadGenerations: %v", err)
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+}
diff --git a/internal/host/runexec.go b/internal/host/runexec.go
index 9d23558e4..9df72ec5a 100644
--- a/internal/host/runexec.go
+++ b/internal/host/runexec.go
@@ -24,6 +24,12 @@ import (
var (
runPollRetry = time.Second
runRecheckEvery = 30 * time.Second
+ // runInterruptEvery is how often follow asks the handle whether an
+ // instance stop or a retired generation interrupted the run, and
+ // runInterruptGrace how long it then waits for the unit's own done
+ // status, which wins if the command had already finished (re-review 6).
+ runInterruptEvery = 250 * time.Millisecond
+ runInterruptGrace = 2 * time.Second
)
// runOutputGrace bounds the output once the command has ended: the unit's
@@ -368,10 +374,29 @@ func (w *runsWorker) follow(ctx context.Context, rec provision.Record, req runs.
defer timeout.Stop()
recheck := time.NewTicker(runRecheckEvery)
defer recheck.Stop()
+ interrupts := time.NewTicker(runInterruptEvery)
+ defer interrupts.Stop()
for {
select {
case p := <-statuses:
return &p.st, p.at
+ case <-interrupts.C:
+ // An interruption ends the run as interrupted (exit 125) at
+ // once, even while the activator is still retrying the unit's
+ // stop (re-review 6): the command's result does not wait on a
+ // stop that may take a while to be proven. A done status that
+ // arrives within the grace is still read, so a command that had
+ // finished first keeps its exit.
+ if reason, _ := handle.Interrupted(); reason != "" {
+ grace := time.NewTimer(runInterruptGrace)
+ defer grace.Stop()
+ select {
+ case p := <-statuses:
+ return &p.st, p.at
+ case <-grace.C:
+ return nil, time.Time{}
+ }
+ }
case <-h.cancel:
res.Cancelled = true
return nil, time.Time{}
diff --git a/internal/host/runexec_linux_test.go b/internal/host/runexec_linux_test.go
index 002789c7a..e1e21bb91 100644
--- a/internal/host/runexec_linux_test.go
+++ b/internal/host/runexec_linux_test.go
@@ -28,6 +28,13 @@ type fakeHandle struct {
interrupted string
askedAt time.Time // when the interrupting stop was asked for; zero is "long ago"
afterStop *instancectl.CmdStatus
+ stopErr error
+}
+
+func (h *fakeHandle) interrupt(why string) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ h.interrupted, h.askedAt = why, time.Now()
}
func (h *fakeHandle) Invocation() string { return "inv1" }
@@ -57,8 +64,9 @@ func (h *fakeHandle) Interrupted() (string, time.Time) {
func (h *fakeHandle) Stop(context.Context) error {
h.mu.Lock()
h.stops++
+ err := h.stopErr
h.mu.Unlock()
- return nil
+ return err
}
func (h *fakeHandle) Release() { h.mu.Lock(); h.released = true; h.mu.Unlock() }
@@ -445,6 +453,33 @@ func TestRunFinishedBeforeAnInstanceStopKeepsItsExit(t *testing.T) {
}
}
+// A run interrupted by a retired generation while its unit is still
+// running, and whose stop keeps failing, ends interrupted (exit 125) as
+// soon as the interruption is seen: the worker does not follow the command
+// to its own end or its timeout (re-review 6).
+func TestRunInterruptedWhileItsStopIsRetriedEndsAt125(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, req := pickedRun(t, s, 3600)
+ h := &fakeHandle{polls: make(chan instancectl.CmdStatus), stopErr: errors.New("processes remained")}
+ go func() {
+ time.Sleep(100 * time.Millisecond)
+ h.interrupt(activator.StopRelease)
+ }()
+ start := time.Now()
+ res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle())
+ if res.Interrupted != activator.StopRelease || res.TimedOut || res.Cancelled {
+ t.Fatalf("result %+v, want interrupted by the retirement", res)
+ }
+ if took := time.Since(start); took > runInterruptGrace+5*time.Second {
+ t.Fatalf("the result took %s", took)
+ }
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ if h.stops == 0 || !h.released {
+ t.Fatalf("stops %d released %v: the worker must still stop and release the handle", h.stops, h.released)
+ }
+}
+
func TestRunWhoseUnitWentAway(t *testing.T) {
s := release.NewDirStore(t.TempDir())
rec, req := pickedRun(t, s, 60)