amadan

carlos / platform Public

merge fix M1: a run's stop is proven before the instance lets go of its database

An instance stop whose run did not settle used to mark the stop uncertain
and carry on: replication closed, the lease went and the entry was
deleted while the command might still be writing. Now the run is part of
the termination proof.

- releasable (zdd's retain-until-proven rule) requires every run of the
  entry to be proven stopped: the cmd unit stopped and its cgroup empty.
- finishStop checks the same before replication closes, so no stop path
  can skip it. A run that is not proven keeps the stop pending, with the
  entry, replication, the lease and the run's handle all kept.
- A failed settle is no longer cached. cmdHold and cmdUnit retry on the
  next stop, and the sweep's pending-stop pass (finishPendingStops) makes
  that retry with generations on or off.
- Streams are drained only once the unit is proven gone.

Tests: a run whose stop can't be proven keeps replication open and the
lease held across sweeps, with generations off and on; a later retry that
proves it releases both. The released-run, failed-start and lost-response
tests now expect the stop to wait for the proof instead of finishing
uncertain.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude db2ccdd30edbff9c3ffa3130d627dff3070dfff9 parent 6432b36
5 files changed, +276 −106
  • internal/activator/activator.go +22 −24
  • internal/activator/cmd.go +100 −26
  • internal/activator/cmd_test.go +118 −39
  • internal/activator/cmdunit.go +28 −13
  • internal/activator/opset.go +8 −4
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 75efd6871..8d7d2a5e9 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -414,7 +414,8 @@ type actEntry struct {
// cmd is the run attached to this instance (run spec §5.3), written
// under a.mu. While it is set the idle sweep never stops the entry;
// every other stop stops and settles its unit in finishStop, before
- // replication closes.
+ // replication closes, and a unit not proven settled keeps the stop
+ // pending with replication and the lease held (merge ruling M1).
cmd *cmdHold
// cmdLeft holds runs released before their unit was proven settled
// (cmdRun.Release): their Stop failed or never came. finishStop settles
@@ -3429,10 +3430,15 @@ func (a *Service) finishPendingStops() {
continue
}
if e.proc == nil {
- // Not reached: only the barrier and releasable leave a stop
- // pending without a process, and both need generations, which
- // the arm above takes. Kept as the guard this loop always had,
- // so a broken invariant holds the lease rather than panicking.
+ // With generations off, only a run that is not proven stopped
+ // leaves a stop pending without a process (finishStop, merge
+ // ruling M1): the retain-until-proven rule finishes it, as the
+ // arm above does with generations on. That rule's proof of a
+ // missing process is trivial, so what it waits on is the run.
+ if !e.genReaping {
+ e.genReaping = true
+ go a.reapCurrentGen(e)
+ }
continue
}
if e.proc.Exited() {
@@ -3547,25 +3553,17 @@ func (a *Service) finishStop(e *actEntry, outcome *stopObservation) {
// A run attached to this instance is stopped and settled before
// replication closes (run spec §5.3, codex 1): everything it committed
// is in the final checkpoint, and nothing writes the database after.
- // A run that will not settle leaves the stop unproven.
- a.mu.Lock()
- holds := append([]*cmdHold(nil), e.cmdLeft...)
- // A run not yet starting is not waited for: RunCmd's last awake check
- // sees this stop and abandons it before any unit starts.
- if e.cmd != nil && e.cmd.starting {
- holds = append(holds, e.cmd)
- }
- why := e.stopWhy
- a.mu.Unlock()
- if len(holds) > 0 {
- cctx, ccancel := context.WithTimeout(context.Background(), cmdStopBound)
- for _, h := range holds {
- if err := h.stop(cctx, why); err != nil {
- log.Printf("activator: %s: run %s did not settle before the final checkpoint: %v", e.host, h.runID, err)
- outcome.uncertain = true
- }
- }
- ccancel()
+ // A run whose stop is not proven keeps the entry, replication, the
+ // lease and its handle (merge ruling M1): the stop stays pending, as
+ // for any process whose termination is unproven, and the sweep tries
+ // the run's settle again (finishPendingStops). The check lives here, so
+ // no stop path can skip it.
+ if !a.settleCmds(e) {
+ log.Printf("activator: stop %s: a run is not proven stopped; keeping replication and the lease until it is", e.host)
+ a.mu.Lock()
+ a.keepPendingLocked(e, outcome)
+ a.mu.Unlock()
+ return
}
// Close = final sync + lease release. After a lost lease the release part
// fails harmlessly (someone else owns the lock); the sync still preserves
diff --git a/internal/activator/cmd.go b/internal/activator/cmd.go
index d96b59586..fde992335 100644
--- a/internal/activator/cmd.go
+++ b/internal/activator/cmd.go
@@ -3,6 +3,7 @@ package activator
import (
"context"
"errors"
+ "log"
"sync"
"time"
@@ -102,19 +103,34 @@ type cmdHold struct {
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
+ // settling is the settle attempt in flight, nil when none is. A stop
+ // joins it rather than starting a second one. Under mu.
+ settling *cmdSettle
+ // proven: a settle attempt succeeded, so nothing of the unit runs. It
+ // is never unset. A failed attempt leaves it false, and the next stop
+ // tries again (merge ruling M1): a failed settle is never cached as
+ // the run's last word while the instance's ownership is given up.
+ // Under mu.
+ proven bool
+}
+
+// cmdSettle is one settle attempt: done closes when it has ended, and err
+// is its result, written before done closes.
+type cmdSettle struct {
+ done chan struct{}
+ err error
}
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{})}
+ return &cmdHold{runID: runID, cancelPrepare: cancelPrepare, now: now, started: make(chan struct{})}
}
// 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.
+// stop after it is not what ended the run), and stops and settles the unit.
+// A unit already proven settled returns nil at once. Otherwise stop joins
+// the attempt in flight, or starts one, and returns that attempt's result:
+// an attempt that failed is tried again by the next stop.
func (h *cmdHold) stop(ctx context.Context, reason string) error {
select {
case <-h.started:
@@ -126,27 +142,43 @@ func (h *cmdHold) stop(ctx context.Context, reason string) error {
h.asked = true
h.reason, h.askedAt = reason, h.now()
}
- h.mu.Unlock()
- if h.unit == nil {
+ if h.unit == nil || h.proven {
+ h.mu.Unlock()
return nil
}
- h.stopOnce.Do(func() {
- go func() {
- h.stopErr = h.unit.StopAndSettle(context.Background())
- close(h.settled)
- }()
- })
+ at := h.settling
+ if at == nil {
+ at = &cmdSettle{done: make(chan struct{})}
+ h.settling = at
+ go h.settle(at)
+ }
+ h.mu.Unlock()
select {
- case <-h.settled:
- return h.stopErr
+ case <-at.done:
+ return at.err
case <-ctx.Done():
return ctx.Err()
}
}
+// settle is one attempt: the unit's StopAndSettle, unbounded by any
+// caller's context (StopAndSettle bounds itself), so a caller that gives
+// up waiting never cuts a settle short.
+func (h *cmdHold) settle(at *cmdSettle) {
+ err := h.unit.StopAndSettle(context.Background())
+ h.mu.Lock()
+ at.err = err
+ if err == nil {
+ h.proven = true
+ }
+ h.settling = nil
+ h.mu.Unlock()
+ close(at.done)
+}
+
// provenSettled reports whether the hold leaves nothing running: the start
-// returned no unit, or the unit settled without error. A failed start
-// that reached the control service returns a unit, since the unit may be
+// 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
// running.
func (h *cmdHold) provenSettled() bool {
select {
@@ -157,12 +189,9 @@ func (h *cmdHold) provenSettled() bool {
if h.unit == nil {
return true
}
- select {
- case <-h.settled:
- return h.stopErr == nil
- default:
- return false
- }
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ return h.proven
}
type cmdRun struct {
@@ -186,8 +215,9 @@ func (r *cmdRun) Stop(ctx context.Context) error { return r.h.stop(ctx, "") }
// Release clears the hold and stamps activity, so the instance idles a
// full window after a run rather than hibernating at the next sweep. A
// unit not proven settled (its Stop failed or never came) stays the
-// instance's: the next stop settles it before replication closes, or is
-// uncertain (run spec §5.3, codex 1).
+// instance's: the next stop settles it before replication closes, and
+// keeps replication and the lease until it has (run spec §5.3, codex 1;
+// merge ruling M1).
func (r *cmdRun) Release() {
r.once.Do(func() {
r.a.mu.Lock()
@@ -217,6 +247,50 @@ func cmdBusyLocked(e *actEntry) bool {
return e.cmd != nil || len(left) > 0
}
+// cmdHoldsLocked is every run of e whose unit may still be running: the
+// released runs not proven settled, and the attached run once it is
+// starting. An attached run not yet starting is not one: RunCmd's last
+// awake check sees a stop and abandons it before any unit starts. The
+// caller holds a.mu.
+func cmdHoldsLocked(e *actEntry) []*cmdHold {
+ var holds []*cmdHold
+ for _, h := range e.cmdLeft {
+ if !h.provenSettled() {
+ holds = append(holds, h)
+ }
+ }
+ if e.cmd != nil && e.cmd.starting && !e.cmd.provenSettled() {
+ holds = append(holds, e.cmd)
+ }
+ return holds
+}
+
+// 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
+// may be running now. Proven means the unit stopped and its cgroup is empty
+// (cmdUnit.StopAndSettle). A settle that fails is not given up on: the stop
+// stays pending with replication and the lease held, and the sweep's next
+// pass calls this again (finishPendingStops). Never call it with a.mu held.
+func (a *Service) settleCmds(e *actEntry) bool {
+ a.mu.Lock()
+ holds, why := cmdHoldsLocked(e), e.stopWhy
+ a.mu.Unlock()
+ if len(holds) == 0 {
+ return true
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), cmdStopBound)
+ defer cancel()
+ proven := true
+ for _, h := range holds {
+ if err := h.stop(ctx, why); err != nil {
+ log.Printf("activator: %s: run %s is not proven stopped: %v", e.host, h.runID, err)
+ proven = false
+ }
+ }
+ return proven
+}
+
// RunCmd runs one command against rt's instance (run spec §5.2 and §5.3):
// refused with claim control armed, for anything but an instance, or on a
// layout other than nested; the instance woken through the ordinary wake
diff --git a/internal/activator/cmd_test.go b/internal/activator/cmd_test.go
index 037822d71..384e79f10 100644
--- a/internal/activator/cmd_test.go
+++ b/internal/activator/cmd_test.go
@@ -64,6 +64,12 @@ func (u *fakeCmdUnit) StopAndSettle(context.Context) error {
}
return u.settleErr
}
+func (u *fakeCmdUnit) setSettleErr(err error) {
+ u.mu.Lock()
+ defer u.mu.Unlock()
+ u.settleErr = err
+}
+
func (u *fakeCmdUnit) stopped() (int, bool) {
u.mu.Lock()
defer u.mu.Unlock()
@@ -333,42 +339,79 @@ func TestAStopDuringTheStartWaitsForIt(t *testing.T) {
}
}
-func TestAnUnsettledRunMakesTheStopUncertain(t *testing.T) {
- for _, c := range []struct {
- settleErr error
- uncertain bool
- }{{nil, false}, {errors.New("processes remained"), true}} {
- a, cb, _ := testCmdActivator(t, Config{})
- cb.unit.settleErr = c.settleErr
- rt := testRoute("c.example.com")
- if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); err != nil {
- t.Fatal(err)
- }
- a.Bounce(rt.Host)
- waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 })
- if got := a.StopUncertain(rt.Host); got != c.uncertain {
- t.Fatalf("settle error %v: StopUncertain = %v, want %v", c.settleErr, got, c.uncertain)
- }
+// A run whose stop cannot be proven keeps the instance's stop pending:
+// the entry, replication and the lease stay, and the run's handle with
+// them. The sweep tries the settle again, and the retry that proves it
+// closes replication and releases the lease (merge ruling M1). That holds
+// with generations off and on.
+func TestAnUnprovenRunStopKeepsReplicationAndTheLease(t *testing.T) {
+ for _, gens := range []bool{false, true} {
+ t.Run(map[bool]string{false: "generations off", true: "generations on"}[gens], func(t *testing.T) {
+ a, cb, _ := testCmdActivator(t, Config{})
+ a.gens = gens
+ cb.unit.setSettleErr(errors.New("processes remained"))
+ rt := testRoute("c.example.com")
+ if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); err != nil {
+ t.Fatal(err)
+ }
+ a.Bounce(rt.Host)
+ waitFor(t, "the first settle", func() bool { n, _ := cb.unit.stopped(); return n >= 1 })
+ // Every sweep retries, and every retry fails: nothing is given up.
+ for i := 0; i < 3; i++ {
+ before, _ := cb.unit.stopped()
+ a.sweepOnce()
+ waitFor(t, "a retried settle", func() bool { n, _ := cb.unit.stopped(); return n > before })
+ if a.entryCount() != 1 {
+ t.Fatal("the entry went while its run was not proven stopped")
+ }
+ if cb.lastRep().isClosed() {
+ t.Fatal("replication closed, and the lease went, while the run was not proven stopped")
+ }
+ }
+ a.mu.Lock()
+ held := a.entries[rt.Host] != nil && len(cmdHoldsLocked(a.entries[rt.Host])) == 1
+ a.mu.Unlock()
+ if !held {
+ t.Fatal("the entry let go of the run's handle")
+ }
+ if !a.StopUncertain(rt.Host) {
+ t.Fatal("StopUncertain = false with the run unproven")
+ }
+ cb.unit.setSettleErr(nil)
+ waitFor(t, "the retry that proves the run stopped", func() bool {
+ a.sweepOnce()
+ return a.entryCount() == 0
+ })
+ if _, repWasClosed := cb.unit.stopped(); repWasClosed {
+ t.Fatal("replication closed before the run was proven stopped")
+ }
+ if !cb.lastRep().isClosed() {
+ t.Fatal("replication never closed, so the lease was never released")
+ }
+ if a.StopUncertain(rt.Host) {
+ t.Fatal("StopUncertain = true after the run was proven stopped")
+ }
+ })
}
}
// A run released without a proven settle (its Stop failed, or was never
// called) still belongs to the instance's next stop: that stop settles it
-// before replication closes, or is uncertain.
+// before replication closes, and holds replication and the lease until
+// it has.
func TestAReleasedRunThatNeverSettledStillHoldsTheStop(t *testing.T) {
for _, c := range []struct {
name string
stopFirst bool
settleErr error
- uncertain bool
}{
- {"stopped and settled", true, nil, false},
- {"never stopped", false, nil, false},
- {"stop failed", true, errors.New("processes remained"), true},
+ {"stopped and settled", true, nil},
+ {"never stopped", false, nil},
+ {"stop failed", true, errors.New("processes remained")},
} {
t.Run(c.name, func(t *testing.T) {
a, cb, clk := testCmdActivator(t, Config{})
- cb.unit.settleErr = c.settleErr
+ cb.unit.setSettleErr(c.settleErr)
rt := testRoute("c.example.com")
h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
if err != nil {
@@ -382,12 +425,26 @@ func TestAReleasedRunThatNeverSettledStillHoldsTheStop(t *testing.T) {
h.Release()
clk.advance(time.Hour)
a.sweepOnce() // the release ended the idle hold
- waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 })
- if stops, repWasClosed := cb.unit.stopped(); stops != 1 || repWasClosed {
- t.Fatalf("stops %d, replication closed first %v", stops, repWasClosed)
+ if c.settleErr != nil {
+ waitFor(t, "the stop's own settle", func() bool { n, _ := cb.unit.stopped(); return n >= 2 })
+ a.sweepOnce()
+ if a.entryCount() != 1 || cb.lastRep().isClosed() {
+ t.Fatal("the stop finished over a run that is not proven stopped")
+ }
+ cb.unit.setSettleErr(nil)
}
- if got := a.StopUncertain(rt.Host); got != c.uncertain {
- t.Fatalf("StopUncertain = %v, want %v", got, c.uncertain)
+ waitFor(t, "the instance stopped", func() bool { a.sweepOnce(); return a.entryCount() == 0 })
+ stops, repWasClosed := cb.unit.stopped()
+ // The failing case: the run's own Stop, the instance stop's
+ // settle, and at least one retry, the last of which proved it.
+ if c.settleErr == nil && stops != 1 || c.settleErr != nil && stops < 3 {
+ t.Fatalf("stops %d", stops)
+ }
+ if repWasClosed {
+ t.Fatal("replication closed before the run settled")
+ }
+ if a.StopUncertain(rt.Host) {
+ t.Fatal("StopUncertain = true after every run was proven stopped")
}
if got, _ := h.Interrupted(); got != "" && c.stopFirst {
t.Fatalf("Interrupted() = %q after the run's own stop", got)
@@ -456,7 +513,7 @@ func TestAFailedStartThatMayHaveRunIsSettled(t *testing.T) {
t.Run(c.name, func(t *testing.T) {
a, cb, _ := testCmdActivator(t, Config{})
cb.startErr, cb.errWithUnit = errors.New("502 mid-roll"), true
- cb.unit.settleErr = c.settleErr
+ cb.unit.setSettleErr(c.settleErr)
rt := testRoute("c.example.com")
if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); !errors.Is(err, cb.startErr) {
t.Fatalf("err = %v, want the start's error", err)
@@ -469,12 +526,20 @@ func TestAFailedStartThatMayHaveRunIsSettled(t *testing.T) {
t.Fatalf("second run: err = %v, busy = %v", err, busy)
}
a.Bounce(rt.Host)
- waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 })
+ if c.settleErr != nil {
+ waitFor(t, "the stop's settle", func() bool { n, _ := cb.unit.stopped(); return n >= 2 })
+ a.sweepOnce()
+ if a.entryCount() != 1 || cb.lastRep().isClosed() {
+ t.Fatal("the stop finished over a unit that is not proven stopped")
+ }
+ cb.unit.setSettleErr(nil)
+ }
+ waitFor(t, "the instance stopped", func() bool { a.sweepOnce(); return a.entryCount() == 0 })
if _, repWasClosed := cb.unit.stopped(); repWasClosed {
t.Fatal("replication closed before the unit settled")
}
- if got := a.StopUncertain(rt.Host); got != (c.settleErr != nil) {
- t.Fatalf("StopUncertain = %v", got)
+ if a.StopUncertain(rt.Host) {
+ t.Fatal("StopUncertain = true after the unit was proven stopped")
}
})
}
@@ -482,8 +547,9 @@ func TestAFailedStartThatMayHaveRunIsSettled(t *testing.T) {
// The real ExecBackend: POST /cmd starts the unit and then fails (a lost
// response). The cmd role is stopped and its cgroup waited for before
-// replication closes; a cgroup that cannot be read empty leaves the stop
-// uncertain.
+// replication closes. A cgroup that cannot be read empty keeps the stop
+// pending with replication open, and the retry once it reads empty
+// finishes it (merge ruling M1).
func TestALostCmdResponseStopsTheCmdRoleBeforeReplicationCloses(t *testing.T) {
for _, c := range []struct {
name string
@@ -496,9 +562,10 @@ func TestALostCmdResponseStopsTheCmdRoleBeforeReplicationCloses(t *testing.T) {
unit.set("active", "inv1", true, true) // the POST reached systemd
b, rt := nestedBackend(t, ctl, unit)
b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
- var repClosedAtEmpty atomic.Bool
+ var repClosedAtEmpty, unreadable atomic.Bool
+ unreadable.Store(c.cgroupErr != nil)
b.CgroupEvents = func(slice, u string) (bool, bool, error) {
- if c.cgroupErr != nil {
+ if unreadable.Load() {
return true, true, c.cgroupErr
}
exists, populated, err := unit.events(slice, u)
@@ -521,15 +588,27 @@ func TestALostCmdResponseStopsTheCmdRoleBeforeReplicationCloses(t *testing.T) {
t.Fatalf("err = %v, want the lost response", err)
}
a.Bounce(rt.Host)
- waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 })
+ if c.cgroupErr != nil {
+ waitFor(t, "the stop kept pending", func() bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[rt.Host]
+ return e != nil && e.pendingStop
+ })
+ if cb.lastRep().isClosed() {
+ t.Fatal("replication closed while the cmd cgroup could not be read empty")
+ }
+ unreadable.Store(false)
+ }
+ waitFor(t, "the instance stopped", func() bool { a.sweepOnce(); return a.entryCount() == 0 })
if !strings.Contains(ctl.joined(), "stop "+rt.Host+" cmd") {
t.Fatalf("the cmd role was never stopped:\n%s", ctl.joined())
}
- if c.cgroupErr == nil && repClosedAtEmpty.Load() {
+ if repClosedAtEmpty.Load() {
t.Fatal("replication closed before the cmd cgroup emptied")
}
- if got := a.StopUncertain(rt.Host); got != (c.cgroupErr != nil) {
- t.Fatalf("StopUncertain = %v", got)
+ if a.StopUncertain(rt.Host) {
+ t.Fatal("StopUncertain = true after the cmd role was proven stopped")
}
})
}
diff --git a/internal/activator/cmdunit.go b/internal/activator/cmdunit.go
index 27a9c863b..91bc76cf1 100644
--- a/internal/activator/cmdunit.go
+++ b/internal/activator/cmdunit.go
@@ -75,7 +75,9 @@ type CmdUnit interface {
Poll(ctx context.Context) (instancectl.CmdStatus, error)
// StopAndSettle stops the unit (SIGTERM, then SIGKILL after 20 s),
// proves its cgroup empty and reads both streams to their end. It is
- // safe to call more than once; later calls return the first's error.
+ // safe to call more than once: once a call has succeeded, later ones
+ // return nil at once, and a call after a failure tries again (merge
+ // ruling M1).
StopAndSettle(ctx context.Context) error
}
@@ -205,8 +207,8 @@ type cmdUnit struct {
rt registry.Route
inv string
out, errs *cmdStream
- once sync.Once
- err error
+ mu sync.Mutex // serialises StopAndSettle's attempts
+ proven bool // an attempt succeeded; under mu
}
func (u *cmdUnit) Invocation() string { return u.inv }
@@ -215,17 +217,30 @@ func (u *cmdUnit) Poll(ctx context.Context) (instancectl.CmdStatus, error) {
return u.b.Control.CmdStatus(ctx, u.rt.Host, u.inv)
}
+// StopAndSettle is one attempt at the proof: the cmd role stopped, then
+// its cgroup waited empty, then both streams read to their end. The
+// streams are drained only once the unit is proven gone, so a unit that
+// outlives a failed attempt keeps its output flowing to the next one.
func (u *cmdUnit) StopAndSettle(ctx context.Context) error {
- u.once.Do(func() {
- sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl)
- stopErr := u.b.Control.Stop(sctx, u.rt.Host, instancectl.RoleCmd)
- cancel()
- wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout)
- emptyErr := u.b.waitRoleCgroupEmpty(wctx, u.rt, instancectl.RoleCmd)
- wcancel()
- u.err = errors.Join(stopErr, emptyErr, u.out.Drain(cmdDrainBound), u.errs.Drain(cmdDrainBound))
- })
- return u.err
+ u.mu.Lock()
+ defer u.mu.Unlock()
+ if u.proven {
+ return nil
+ }
+ sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl)
+ stopErr := u.b.Control.Stop(sctx, u.rt.Host, instancectl.RoleCmd)
+ cancel()
+ wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout)
+ emptyErr := u.b.waitRoleCgroupEmpty(wctx, u.rt, instancectl.RoleCmd)
+ wcancel()
+ if err := errors.Join(stopErr, emptyErr); err != nil {
+ return err
+ }
+ if err := errors.Join(u.out.Drain(cmdDrainBound), u.errs.Drain(cmdDrainBound)); err != nil {
+ return err
+ }
+ u.proven = true
+ return nil
}
// cmdStream is the agent's end of one cmd stream (run spec §5.5): the
diff --git a/internal/activator/opset.go b/internal/activator/opset.go
index d0fb79bf5..2e257d804 100644
--- a/internal/activator/opset.go
+++ b/internal/activator/opset.go
@@ -465,9 +465,13 @@ func (a *Service) reapExited(e *actEntry, m *genMember) {
// §7.4): a stop finishes, and its lease is released, only when every
// member has left the operation set, no parent restore retry is in flight
// (final review R2-2), no stray sidecar of the host is tracked (Task 7),
-// and the current generation has passed genTerminated (its leader exited
-// and its child, or a worker's host cgroup, empty). It runs without a.mu
-// held; genTerminated may take DeadlineControl.
+// the current generation has passed genTerminated (its leader exited and
+// its child, or a worker's host cgroup, empty), and every run of the
+// instance is proven stopped, its unit stopped and its cgroup empty
+// (settleCmds; run spec §5.3, merge ruling M1). A run writes the database
+// as the serving process does, so it is held to the same rule. It runs
+// without a.mu held; genTerminated may take DeadlineControl, and a run's
+// settle up to cmdStopBound.
func (a *Service) releasable(e *actEntry) bool {
a.mu.Lock()
members, strays, restoring := len(e.ops.members), len(a.straySidecars[e.host]), e.ops.restoring
@@ -476,7 +480,7 @@ func (a *Service) releasable(e *actEntry) bool {
if members > 0 || strays > 0 || restoring {
return false
}
- return a.genTerminated(rt, g, p)
+ return a.genTerminated(rt, g, p) && a.settleCmds(e)
}
// keepPendingLocked leaves a stop pending with its lease held: the entry