amadan

carlos / platform Public

merge fix M2: a run belongs to the generation it was admitted against

A run used to capture the serving build at its last check and start its
unit outside the lock, while an overlap swap could replace and retire that
generation without looking at the run. The command could start against a
retired generation, outlive a restart that should end it, and lose its
build's pin in the generations file.

- Each run records its generation (number and exe pin) under a.mu as it
  starts, and execs that generation's build.
- Admission and the swap share the lock. An overlap swap is not admitted
  (errSwapBusy, abandoned, the scheduler retries) while a run's unit is
  starting, and the switch checks again. RunCmd starts no unit while a
  live successor is in the operation set: it waits for the switch, or for
  the successor's end, and runs whatever is then current.
- The switch interrupts every run admitted against the retiring
  generation. The run is stopped and settled, and reads the attempt's
  cause through stopWhyFor: the instance moved to a new release, or was
  restarted (RUN-C2/C3, exit 125 per RUN-C75). A hold swap's stop now
  also gives a restart or config attempt the restart reason.
- The generations file names a run's generation until the run is proven
  settled, and is written again when that settle frees the pin.

Tests: a run start in flight refuses an overlap swap, which is admitted
once the start returns; a run arriving with a successor pending waits and
runs the successor's build; a running command on a retired generation is
interrupted with the cause's reason; a retired run's build stays pinned
until it settles.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 088cb85856de12c0a7554c17d25cc365d610bd07 parent db2ccdd
6 files changed, +399 −11
  • internal/activator/cmd.go +117 −8
  • internal/activator/cmd_swap_test.go +237 −0
  • internal/activator/generations_file.go +9 −1
  • internal/activator/swap.go +21 −0
  • internal/activator/swap_fixture_test.go +12 −1
  • internal/activator/swaphold.go +3 −1
diff --git a/internal/activator/cmd.go b/internal/activator/cmd.go
index fde992335..e52f8d6e0 100644
--- a/internal/activator/cmd.go
+++ b/internal/activator/cmd.go
@@ -112,6 +112,18 @@ type cmdHold struct {
// the run's last word while the instance's ownership is given up.
// Under mu.
proven bool
+ // 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
+ // build; a swap or a scheduled restart that retires gen interrupts
+ // the run (interruptCmdsLocked), and the generations file names gen's
+ // build until the run is proven settled (entryGenerationsLocked).
+ gen Gen
+ // onProven, when set, runs once a settle has proven the unit gone,
+ // with no lock held: RunCmd's republication of the generations file,
+ // whose pin for gen may have been this run's alone. Set before the
+ // unit starts and never written again.
+ onProven func()
}
// cmdSettle is one settle attempt: done closes when it has ended, and err
@@ -174,6 +186,9 @@ func (h *cmdHold) settle(at *cmdSettle) {
h.settling = nil
h.mu.Unlock()
close(at.done)
+ if err == nil && h.onProven != nil {
+ h.onProven()
+ }
}
// provenSettled reports whether the hold leaves nothing running: the start
@@ -265,6 +280,63 @@ func cmdHoldsLocked(e *actEntry) []*cmdHold {
return holds
}
+// cmdStartInFlightLocked reports whether e's run has passed RunCmd's last
+// check and its StartCmd has not returned: a unit may be starting against
+// e's current generation right now. An overlap swap is not admitted, and
+// does not switch, while one is (merge ruling M2). The caller holds a.mu.
+func cmdStartInFlightLocked(e *actEntry) bool {
+ if e.cmd == nil || !e.cmd.starting {
+ return false
+ }
+ select {
+ case <-e.cmd.started:
+ return false
+ default:
+ return true
+ }
+}
+
+// switchPendingLocked is a live overlap successor of e: a member spawning
+// or health-checking whose owner has not given up on it (an unproven
+// member's attempt is over). While there is one, e's current generation
+// may retire at any moment, so RunCmd starts no unit against it, and waits
+// for the switch or the successor's end (merge ruling M2). nil when there
+// is none. The caller holds a.mu.
+func switchPendingLocked(e *actEntry) *genMember {
+ for _, m := range e.ops.members {
+ if (m.role == roleSpawning || m.role == roleSuccessor) && !m.unproven {
+ return m
+ }
+ }
+ return nil
+}
+
+// cmdSwitchPoll is how often RunCmd re-checks a pending switch it waits on,
+// besides the successor's own done.
+var cmdSwitchPoll = 100 * time.Millisecond
+
+// interruptCmdsLocked interrupts every run of e admitted against
+// generation n, which is retiring (merge ruling M2): each unit is stopped
+// and settled off a.mu, and the run learns why, which the member reads as
+// interrupted (exit 125, RUN-C75). A settle that fails leaves the run
+// unproven: its own Stop or the instance's next stop tries again, and the
+// generations file keeps naming n's build until one succeeds. The caller
+// holds a.mu.
+func (a *Service) interruptCmdsLocked(e *actEntry, n uint64, why string) {
+ for _, h := range cmdHoldsLocked(e) {
+ if h.gen.N != n {
+ continue
+ }
+ go func(h *cmdHold) {
+ 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)
+ }
+ }(h)
+ }
+}
+
// 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
@@ -371,24 +443,61 @@ func (a *Service) RunCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (
prepErr = spec.Prepare(pctx)
}
a.mu.Lock()
- if e.state != actAwake {
- why := e.stopWhy
+ for {
+ if e.state != actAwake {
+ why := e.stopWhy
+ a.mu.Unlock()
+ return abandon(&CmdStoppedError{Reason: why})
+ }
+ if prepErr != nil {
+ a.mu.Unlock()
+ return abandon(prepErr)
+ }
+ // 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 {
+ break
+ }
a.mu.Unlock()
- return abandon(&CmdStoppedError{Reason: why})
+ select {
+ case <-m.done:
+ case <-time.After(cmdSwitchPoll):
+ case <-pctx.Done():
+ }
+ a.mu.Lock()
+ if pctx.Err() != nil && e.state == actAwake {
+ a.mu.Unlock()
+ return abandon(pctx.Err())
+ }
}
- if prepErr != nil {
+ // Admitted against the current generation, under the lock a swap's
+ // admission and switch take (merge ruling M2): no successor is live,
+ // and a swap admitted from here on is refused until StartCmd returns
+ // (cmdStartInFlightLocked).
+ h.starting = true
+ h.gen = e.gen
+ h.onProven = func() {
+ // The pin for h.gen is this run's alone once that generation is
+ // no longer the entry's current one.
+ a.mu.Lock()
+ stale := a.entries[e.host] != e || e.gen.N != h.gen.N
a.mu.Unlock()
- return abandon(prepErr)
+ if stale {
+ _ = a.persistGenerations()
+ }
}
- h.starting = true
// What the command runs is what the instance serves now: with
// generations on, the current generation's staged build, never the
// live link, which a repoint or a fallback can move off it (zero-
// downtime spec §7.5, as for the legacy sidecar). A generation woken on
// the live link has no staged build, and the run takes the live link.
spec.serving = nil
- if a.gens && e.gen.StagedPath != "" {
- spec.serving = &instancectl.StagedExe{Path: e.gen.StagedPath, Version: e.gen.Build, Digest: e.gen.Digest}
+ if a.gens && h.gen.StagedPath != "" {
+ spec.serving = &instancectl.StagedExe{Path: h.gen.StagedPath, Version: h.gen.Build, Digest: h.gen.Digest}
}
a.mu.Unlock()
sctx, scancel := context.WithTimeout(ctx, cmdStartBound)
diff --git a/internal/activator/cmd_swap_test.go b/internal/activator/cmd_swap_test.go
new file mode 100644
index 000000000..2d9d8250b
--- /dev/null
+++ b/internal/activator/cmd_swap_test.go
@@ -0,0 +1,237 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "slices"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// cmdSwapBackend is the swap fixture's backend with the run capability, on
+// the nested layout (merge ruling M2).
+type cmdSwapBackend struct {
+ *fakeSwapBackend
+ cmu sync.Mutex
+ unit *fakeCmdUnit
+ specs []CmdSpec
+ gate chan struct{} // when set, StartCmd blocks until it closes
+ entered chan struct{} // when set, closed as StartCmd is entered
+}
+
+func (b *cmdSwapBackend) StartCmd(_ context.Context, _ registry.Route, spec CmdSpec) (CmdUnit, error) {
+ b.cmu.Lock()
+ gate, entered := b.gate, b.entered
+ b.entered = nil
+ b.cmu.Unlock()
+ if entered != nil {
+ close(entered)
+ }
+ if gate != nil {
+ <-gate
+ }
+ b.cmu.Lock()
+ defer b.cmu.Unlock()
+ b.specs = append(b.specs, spec)
+ return b.unit, nil
+}
+
+func (b *cmdSwapBackend) started() []CmdSpec {
+ b.cmu.Lock()
+ defer b.cmu.Unlock()
+ return slices.Clone(b.specs)
+}
+
+func newCmdSwapFixture(t *testing.T) (*swapFixture, *cmdSwapBackend) {
+ t.Helper()
+ var cb *cmdSwapBackend
+ f := newSwapFixtureOn(t, func(b *fakeSwapBackend) Backend {
+ cb = &cmdSwapBackend{fakeSwapBackend: b, unit: &fakeCmdUnit{}}
+ return cb
+ })
+ f.b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
+ return f, cb
+}
+
+func (a *Service) runGenOf(h CmdHandle) Gen {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ return h.(*cmdRun).h.gen
+}
+
+// A run whose unit is starting refuses an overlap swap's admission: the
+// attempt is abandoned with nothing spawned, never admitted to retire the
+// generation that run is starting against. Once the start has returned,
+// the same attempt is admitted (merge ruling M2).
+func TestARunStartInFlightRefusesAnOverlapSwap(t *testing.T) {
+ f, cb := newCmdSwapFixture(t)
+ cb.gate, cb.entered = make(chan struct{}), make(chan struct{})
+ entered := cb.entered
+ type ran struct {
+ h CmdHandle
+ err error
+ }
+ done := make(chan ran, 1)
+ go func() {
+ h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"})
+ done <- ran{h, err}
+ }()
+ <-entered
+ res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapBusy.Error()) || res.Executed {
+ t.Fatalf("Swap during the run's start = %+v, want abandoned as busy with nothing run", res)
+ }
+ if gens, raises, _ := f.b.calls(); gens != 0 || raises != 0 {
+ t.Fatalf("the refused attempt spawned %d generations and raised %d times", gens, raises)
+ }
+ close(cb.gate)
+ r := <-done
+ if r.err != nil {
+ t.Fatal(r.err)
+ }
+ if g := f.a.runGenOf(r.h); g.N != f.incumbent.N {
+ t.Fatalf("the run was admitted against generation %d, want the incumbent %d", g.N, f.incumbent.N)
+ }
+ f.rev.Add(1)
+ if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone {
+ t.Fatalf("Swap after the start = %+v", res)
+ }
+}
+
+// A run that arrives while an overlap successor is spawning starts no unit
+// against the incumbent, which is about to retire: it waits for the switch
+// and then runs the successor's build, admitted against the successor's
+// generation (merge ruling M2).
+func TestARunWaitsForAPendingSwitch(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
+ })
+ type ran struct {
+ h CmdHandle
+ err error
+ }
+ done := make(chan ran, 1)
+ go func() {
+ h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"})
+ done <- ran{h, err}
+ }()
+ time.Sleep(3 * cmdSwitchPoll)
+ if n := len(cb.started()); n != 0 {
+ t.Fatalf("%d units started while a successor was pending", n)
+ }
+ f.b.gmu.Lock()
+ close(f.b.genGate)
+ f.b.gmu.Unlock()
+ res := <-swapped
+ if res.Outcome != SwapDone {
+ t.Fatalf("Swap = %+v", res)
+ }
+ r := <-done
+ if r.err != nil {
+ t.Fatal(r.err)
+ }
+ defer r.h.Release()
+ if g := f.a.runGenOf(r.h); g.N != res.Gen.N {
+ t.Fatalf("the run was admitted against generation %d, want the successor %d", g.N, res.Gen.N)
+ }
+ specs := cb.started()
+ if len(specs) != 1 || specs[0].serving == nil || specs[0].serving.Path != f.staged {
+ t.Fatalf("the run started %+v, want one unit on the successor's build %s", specs, f.staged)
+ }
+ if n, _ := cb.unit.stopped(); n != 0 {
+ t.Fatal("a run on the current generation was interrupted")
+ }
+}
+
+// A running command on a generation a swap retires is interrupted and
+// settled, and the run reads why: a new release for a release attempt, a
+// restart for a scheduled restart or config change (exit 125, RUN-C75;
+// merge ruling M2).
+func TestARunOnARetiredGenerationIsInterrupted(t *testing.T) {
+ for _, c := range []struct {
+ cause Cause
+ why string
+ }{{CauseRelease, StopRelease}, {CauseRestart, StopRestart}, {CauseConfig, StopRestart}} {
+ t.Run(string(c.cause), func(t *testing.T) {
+ f, cb := newCmdSwapFixture(t)
+ h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if g := f.a.runGenOf(h); g.N != f.incumbent.N || g.StagedPath != f.incumbent.StagedPath {
+ t.Fatalf("the run was admitted against %+v, want the incumbent", g)
+ }
+ at := f.attempt(ModeOverlap)
+ at.Cause = c.cause
+ if res := f.a.Swap(context.Background(), at, f.newest); res.Outcome != SwapDone {
+ t.Fatalf("Swap = %+v", res)
+ }
+ waitFor(t, "the run interrupted and settled", func() bool {
+ n, _ := cb.unit.stopped()
+ return n == 1 && h.(*cmdRun).h.provenSettled()
+ })
+ if why, at := h.Interrupted(); why != c.why || at.IsZero() {
+ t.Fatalf("Interrupted() = %q at %v, want %q", why, at, c.why)
+ }
+ if f.a.entryCount() != 1 {
+ t.Fatal("interrupting the run stopped the instance")
+ }
+ h.Release()
+ })
+ }
+}
+
+// A run on a retired generation keeps that generation's build in the
+// generations file until the run is proven settled, so root neither
+// restages nor sweeps the bytes the command runs; the settle that proves
+// it releases the pin (merge ruling M2).
+func TestARetiredRunKeepsItsBuildPinnedUntilItSettles(t *testing.T) {
+ f, cb := newCmdSwapFixture(t)
+ path := release.GenerationsPath(t.TempDir())
+ f.a.GenerationsFile = path
+ cb.unit.setSettleErr(errors.New("processes remained"))
+ h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest)
+ if res.Outcome != SwapDone {
+ 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)
+ }
+ if err := f.a.persistGenerations(); err != nil {
+ t.Fatal(err)
+ }
+ _, members := f.a.genStateForTest(f.host)
+ if len(members) != 0 {
+ t.Fatalf("the retiring generation is still a member: %v", members)
+ }
+ if !pinned() {
+ t.Fatal("the retired build went from the generations file while its run was unsettled")
+ }
+ cb.unit.setSettleErr(nil)
+ if err := h.Stop(context.Background()); err != nil {
+ t.Fatalf("Stop: %v", err)
+ }
+ waitFor(t, "the pin released", func() bool { return !pinned() })
+ h.Release()
+}
diff --git a/internal/activator/generations_file.go b/internal/activator/generations_file.go
index 5d8080466..2a07eaabe 100644
--- a/internal/activator/generations_file.go
+++ b/internal/activator/generations_file.go
@@ -143,7 +143,9 @@ func (a *Service) generationsSnapshotLocked() release.Generations {
// - a hold's pinned generation, from the moment its waking entry exists;
// - a planned wake's build and its fallback, from the moment the plan is
// stored, so both are on disk before either execs;
-// - a socketless attempt's build and fallback (Task 11).
+// - a socketless attempt's build and fallback (Task 11);
+// - the generation every run not yet proven settled was admitted
+// against (merge ruling M2).
//
// The plan's builds stay listed while the entry lives, as e.plan does
// (afterWake reads it after the wake): a few more pins than the processes
@@ -177,6 +179,12 @@ func (a *Service) entryGenerationsLocked(e *actEntry, add func(release.StagedIde
for _, v := range e.attemptBuilds {
add(release.StagedIdentity{Version: v})
}
+ // A run's generation, until the run is proven settled (merge ruling
+ // M2): its command execs that build, and may outlive the generation's
+ // own process once a swap retires it.
+ for _, h := range cmdHoldsLocked(e) {
+ gen(h.gen)
+ }
}
// genIdentity is g as the generations file names it, and whether it is
diff --git a/internal/activator/swap.go b/internal/activator/swap.go
index 0142fc7e8..23313e089 100644
--- a/internal/activator/swap.go
+++ b/internal/activator/swap.go
@@ -411,6 +411,15 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n
if len(e.ops.members) > 0 || e.ops.raised {
return errSwapBusy
}
+ // A run's unit may be starting against the incumbent right now
+ // (merge ruling M2). Admitting now could retire the incumbent under
+ // a command that has not even started; the attempt is abandoned
+ // instead, and the scheduler tries again once the start has
+ // returned. From here on RunCmd waits for this successor rather
+ // than starting a unit (switchPendingLocked).
+ if cmdStartInFlightLocked(e) {
+ return errSwapBusy
+ }
a.addMemberLocked(e, m)
rt, incumbent = e.route, e.gen
return nil
@@ -550,6 +559,13 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n
if m.proc == nil || m.proc.Exited() {
return errSuccessorExited
}
+ // Admission refused a run's start in flight and RunCmd starts none
+ // while this successor is live, so none can be in flight here
+ // (merge ruling M2). Checked again because a switch over one would
+ // retire the generation that run is starting against.
+ if cmdStartInFlightLocked(e) {
+ return errSwapBusy
+ }
// The canonical symlink, the purge and the socket mode first. If the
// link cannot move, nothing has switched yet. This runs under a.mu,
// in the same critical section that sets e.gen below: plan 1's lock
@@ -564,6 +580,11 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n
e.gen, e.proc = m.gen, m.proc
e.spawnVersion, e.staleSince, e.spawnedAt = m.gen.Build, time.Time{}, a.now()
a.noteCurrent(e.host, m.gen.N) // plan 1's cache generation (C4)
+ // Runs admitted against the retiring generation are interrupted
+ // with it, as a scheduled restart's stop interrupts them (merge
+ // ruling M2): a command never outlives the build it was started
+ // from as that build retires.
+ a.interruptCmdsLocked(e, retiring.gen.N, stopWhyFor(at.Cause))
switched = true
return nil
})
diff --git a/internal/activator/swap_fixture_test.go b/internal/activator/swap_fixture_test.go
index a13ad344a..a3b2d25a1 100644
--- a/internal/activator/swap_fixture_test.go
+++ b/internal/activator/swap_fixture_test.go
@@ -221,6 +221,13 @@ type swapFixture struct {
}
func newSwapFixture(t *testing.T) *swapFixture {
+ t.Helper()
+ return newSwapFixtureOn(t, nil)
+}
+
+// newSwapFixtureOn is newSwapFixture with the Service's backend made by
+// wrap from the swap fake, for a test that needs one more capability.
+func newSwapFixtureOn(t *testing.T, wrap func(*fakeSwapBackend) Backend) *swapFixture {
t.Helper()
b := &fakeSwapBackend{fakeBackend: &fakeBackend{}}
clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
@@ -228,7 +235,11 @@ func newSwapFixture(t *testing.T) *swapFixture {
if err != nil {
t.Fatal(err)
}
- a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg)
+ var be Backend = b
+ if wrap != nil {
+ be = wrap(b)
+ }
+ a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, be, reg)
a.gens = true // plan 1's switch, as plan 5 sets it from CARLOS_SWAP
a.now = clk.now
a.genHealth = func(context.Context, Gen) error { return nil }
diff --git a/internal/activator/swaphold.go b/internal/activator/swaphold.go
index e69307765..8695a6e9c 100644
--- a/internal/activator/swaphold.go
+++ b/internal/activator/swaphold.go
@@ -126,7 +126,9 @@ func (a *Service) swapHold(ctx context.Context, at Attempt, newest func(string)
}
e.holdNext = h
held, stopAt = e, time.Now()
- a.beginStopLocked(e, StopRelease)
+ // A run the stop interrupts reads the attempt's cause: a restart or
+ // a config change is a restart, not a new release (stopWhyFor).
+ a.beginStopLocked(e, stopWhyFor(at.Cause))
return nil
})
if err != nil {