amadan

carlos / platform Public

zdd plan 3 final fix 2: a parent restore retry is joined by its entry's stop (R2-2)

The sweep's parent restore retry is now a lifecycle operation of its
entry. It is launched only for an awake entry that is still the host's
(a stopping entry launches none), it carries opSet.restoreDone, and the
barrier (joinOps) waits on it like a member, so a terminal stop, a
hold's included, cannot release the lease, forget the entry or install a
pinned replacement while it is in flight. releasable also refuses while
it runs. The backend call stays off a.mu.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 4a134a2b2e5f000b8231a4d7cc347a56fb8669b4 parent 895fc86
3 files changed, +256 −16
  • internal/activator/opset.go +39 −16
  • internal/activator/swap_fixture_test.go +18 −0
  • internal/activator/swap_parent_retry_join_test.go +199 −0
diff --git a/internal/activator/opset.go b/internal/activator/opset.go
index 5156eac36..f1809303b 100644
--- a/internal/activator/opset.go
+++ b/internal/activator/opset.go
@@ -75,11 +75,14 @@ type straySidecar struct {
// leave restores it; a restore that fails keeps raised, and the sweep
// retries it (retryParentRestoreLocked, final review F11). While raised is
// set with no member, no overlap is admitted. restoring marks a retry in
-// flight.
+// flight, and restoreDone closes when it has finished: the retry is a
+// lifecycle operation of the entry, joined by the barrier like a member
+// (final review R2-2).
type opSet struct {
- members map[uint64]*genMember
- raised bool
- restoring bool
+ members map[uint64]*genMember
+ raised bool
+ restoring bool
+ restoreDone chan struct{}
}
// opJoinBound bounds the barrier's wait. A member's owner needs at most a
@@ -104,18 +107,24 @@ func (a *Service) isMember(e *actEntry, m *genMember) bool {
// joinOps is the barrier (spec §7.4). It cancels every member that is still
// an attempt (spawning or successor) and waits, bounded by opJoinBound, for
-// every member to leave the set. It reports false when a member's
-// termination was not proven in time. stop() then keeps the lease, exactly
-// as it does for ErrStillRunning. An empty set returns true at once.
+// every member to leave the set, and for a parent restore retry in flight
+// to finish (final review R2-2): its write is keyed by host, so once the
+// entry is gone it could lower a later entry's raised parent. It reports
+// false when a member's termination was not proven, or the retry did not
+// finish, in time. stop() then keeps the lease, exactly as it does for
+// ErrStillRunning. An empty set returns true at once.
func (a *Service) joinOps(e *actEntry) bool {
a.mu.Lock()
- waits := make([]chan struct{}, 0, len(e.ops.members))
+ waits := make([]chan struct{}, 0, len(e.ops.members)+1)
for _, m := range e.ops.members {
if m.cancel != nil {
m.cancel()
}
waits = append(waits, m.done)
}
+ if e.ops.restoreDone != nil {
+ waits = append(waits, e.ops.restoreDone)
+ }
hook := a.joinWaiting
a.mu.Unlock()
if len(waits) == 0 {
@@ -359,20 +368,33 @@ func parentUntouched(err error) bool {
// that is gone takes its obligation with it: the next cold spawn under an
// empty parent writes the tier afresh (cgroup.ensureParent). The caller
// holds a.mu.
+//
+// The retry is a lifecycle operation of the entry (final review R2-2). Only
+// an awake entry that is still the host's launches one, so a stopping entry
+// launches none; and a terminal stop, a hold's included, joins one in
+// flight (joinOps, through restoreDone) before it releases the lease,
+// forgets the entry or installs a pinned replacement. The launch and the
+// state change to stopping are both under a.mu, so a stop either sees the
+// retry or the retry is never launched.
func (a *Service) retryParentRestoreLocked(e *actEntry) {
if !e.ops.raised || e.ops.restoring || len(e.ops.members) > 0 {
return
}
- e.ops.restoring = true
+ if a.entries[e.host] != e || e.state != actAwake {
+ return
+ }
+ done := make(chan struct{})
+ e.ops.restoring, e.ops.restoreDone = true, done
rt, tier := e.route, e.gen.Tier
go func() {
ok := a.restoreParent(rt, tier)
a.mu.Lock()
- e.ops.restoring = false
+ e.ops.restoring, e.ops.restoreDone = false, nil
if ok && len(e.ops.members) == 0 {
e.ops.raised = false
}
a.mu.Unlock()
+ close(done)
}()
}
@@ -441,16 +463,17 @@ func (a *Service) reapExited(e *actEntry, m *genMember) {
// releasable is THE retain-until-proven rule for an entry's lease (spec
// §7.4): a stop finishes, and its lease is released, only when every
-// member has left the operation set, 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.
+// 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.
func (a *Service) releasable(e *actEntry) bool {
a.mu.Lock()
- members, strays := len(e.ops.members), len(a.straySidecars[e.host])
+ members, strays, restoring := len(e.ops.members), len(a.straySidecars[e.host]), e.ops.restoring
rt, g, p := e.route, e.gen, e.proc
a.mu.Unlock()
- if members > 0 || strays > 0 {
+ if members > 0 || strays > 0 || restoring {
return false
}
return a.genTerminated(rt, g, p)
diff --git a/internal/activator/swap_fixture_test.go b/internal/activator/swap_fixture_test.go
index 228562d7c..e8beea1e7 100644
--- a/internal/activator/swap_fixture_test.go
+++ b/internal/activator/swap_fixture_test.go
@@ -67,6 +67,12 @@ type fakeSwapBackend struct {
raiseErr error
// restoreFails is how many more RestoreGenParent calls fail.
restoreFails int
+ // restoreGate, when set, holds every RestoreGenParent call before its
+ // write until it closes; restoreEntered, when set, is sent to
+ // (non-blocking) as each call reaches the gate. A held call is counted
+ // and recorded only once it writes.
+ restoreGate chan struct{}
+ restoreEntered chan struct{}
}
func (g *fakeSwapBackend) StartStaged(ctx context.Context, _ registry.Route, gen Gen) (Gen, Proc, error) {
@@ -136,6 +142,18 @@ func (g *fakeSwapBackend) RaiseGenParent(_ context.Context, _ registry.Route, in
}
func (g *fakeSwapBackend) RestoreGenParent(_ context.Context, _ registry.Route, tier cgroup.Limit) error {
+ g.gmu.Lock()
+ gate, entered := g.restoreGate, g.restoreEntered
+ g.gmu.Unlock()
+ if gate != nil {
+ if entered != nil {
+ select {
+ case entered <- struct{}{}:
+ default:
+ }
+ }
+ <-gate
+ }
g.record(fmt.Sprintf("restore:%d", tier.MemoryMax))
g.gmu.Lock()
defer g.gmu.Unlock()
diff --git a/internal/activator/swap_parent_retry_join_test.go b/internal/activator/swap_parent_retry_join_test.go
new file mode 100644
index 000000000..6cae14009
--- /dev/null
+++ b/internal/activator/swap_parent_retry_join_test.go
@@ -0,0 +1,199 @@
+package activator
+
+import (
+ "context"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+)
+
+// restoreInFlightForTest reports whether host's entry has a parent restore
+// retry in flight.
+func (a *Service) restoreInFlightForTest(host string) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ return e != nil && e.ops.restoring
+}
+
+// owedRestore leaves f's entry with a parent restore owed: an overlap whose
+// last member's restore failed (F11), so the parent stays marked raised.
+func owedRestore(t *testing.T, f *swapFixture) {
+ t.Helper()
+ f.b.gmu.Lock()
+ f.b.restoreFails = 1
+ f.b.gmu.Unlock()
+ if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone {
+ t.Fatalf("Swap = %+v", res)
+ }
+ if !f.a.parentRaisedForTest(f.host) {
+ t.Fatal("fixture: the failed restore should leave the parent marked raised")
+ }
+}
+
+// gateRestores holds every later RestoreGenParent before its write and
+// returns the gate and the channel each held call signals on.
+func gateRestores(f *swapFixture) (gate, entered chan struct{}) {
+ gate, entered = make(chan struct{}), make(chan struct{}, 8)
+ f.b.gmu.Lock()
+ f.b.restoreGate, f.b.restoreEntered = gate, entered
+ f.b.gmu.Unlock()
+ return gate, entered
+}
+
+// joinSignal installs the barrier's join hook and returns the channel it
+// signals on when a stop waits on the operation set.
+func joinSignal(f *swapFixture) chan struct{} {
+ joining := make(chan struct{}, 1)
+ f.a.mu.Lock()
+ f.a.joinWaiting = func(_, what string) {
+ if what == "operation set" {
+ select {
+ case joining <- struct{}{}:
+ default:
+ }
+ }
+ }
+ f.a.mu.Unlock()
+ return joining
+}
+
+// Final review R2-2: a parent restore retry is a lifecycle operation of its
+// entry. A terminal stop joins one in flight before its lease release; a
+// stopping entry launches none; and a hold's stop joins it before its
+// pinned replacement is installed, so the old retry can never write the
+// old entry's single-generation tier after a later overlap raised the
+// host's parent for two children.
+func TestAParentRestoreRetryIsJoinedByTheEntrysStop(t *testing.T) {
+ t.Run("a terminal stop joins the retry in flight", func(t *testing.T) {
+ f := newSwapFixture(t)
+ owedRestore(t, f)
+ gate, entered := gateRestores(f)
+ f.a.sweepOnce()
+ <-entered // the retry is past its launch, before its write
+ joining := joinSignal(f)
+ rep := f.b.lastRep()
+ f.a.mu.Lock()
+ f.a.beginStopLocked(f.a.entries[f.host])
+ f.a.mu.Unlock()
+ waitFor(t, "the stop to join the retry or finish", func() bool {
+ select {
+ case <-joining:
+ return true
+ default:
+ }
+ return !f.a.entryPresent(f.host)
+ })
+ if !f.a.entryPresent(f.host) || rep.isClosed() {
+ t.Errorf("the stop finished (entry present %v, lease released %v) while the parent restore retry was in flight", f.a.entryPresent(f.host), rep.isClosed())
+ }
+ close(gate)
+ waitFor(t, "the stop to finish once the retry is done", func() bool { return !f.a.entryPresent(f.host) && rep.isClosed() })
+ })
+
+ t.Run("a stopping entry launches no retry", func(t *testing.T) {
+ f := newSwapFixture(t)
+ owedRestore(t, f)
+ gate, _ := gateRestores(f)
+ defer close(gate)
+ stopGate := make(chan struct{})
+ f.a.mu.Lock()
+ e := f.a.entries[f.host]
+ cur := e.proc.(*fakeProc)
+ f.a.mu.Unlock()
+ cur.setStopGate(stopGate)
+ f.a.mu.Lock()
+ f.a.beginStopLocked(e)
+ f.a.mu.Unlock()
+ // Past the barrier, inside the current generation's Stop.
+ waitFor(t, "the stop to reach the current generation", func() bool { return cur.stopEntered() > 0 })
+ f.a.sweepOnce()
+ if f.a.restoreInFlightForTest(f.host) {
+ t.Error("the sweep launched a parent restore retry for a stopping entry")
+ }
+ close(stopGate)
+ waitFor(t, "the stop to finish", func() bool { return !f.a.entryPresent(f.host) })
+ })
+
+ t.Run("a hold, its replacement and a later overlap", func(t *testing.T) {
+ f := newSwapFixture(t)
+ owedRestore(t, f)
+ gate, entered := gateRestores(f)
+ f.a.sweepOnce()
+ <-entered
+ joining := joinSignal(f)
+ f.a.mu.Lock()
+ oldE := f.a.entries[f.host]
+ f.a.mu.Unlock()
+ f.rev.Add(1)
+ results := make(chan SwapResult, 1)
+ go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest) }()
+ joined := false
+ waitFor(t, "the hold's stop to join the retry or install its replacement", func() bool {
+ select {
+ case <-joining:
+ joined = true
+ return true
+ default:
+ }
+ f.a.mu.Lock()
+ defer f.a.mu.Unlock()
+ return f.a.entries[f.host] != oldE
+ })
+ var once sync.Once
+ release := func() { once.Do(func() { close(gate) }) }
+ defer release()
+ if !joined {
+ t.Error("the hold's stop installed its pinned replacement while the old entry's parent restore retry was in flight")
+ } else {
+ release()
+ }
+ select {
+ case res := <-results:
+ if res.Outcome != SwapDone {
+ t.Fatalf("hold Swap = %+v", res)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatal("the hold never finished")
+ }
+ waitFor(t, "the replacement to be awake", func() bool { return f.a.entryAwake(f.host) })
+ // A later overlap on the replacement raises the host's parent for two
+ // children. An old retry still held lands now, if it can.
+ _, _, before := f.b.calls()
+ f.b.gmu.Lock()
+ f.b.onRaise = func() {
+ if joined {
+ return
+ }
+ release()
+ deadline := time.Now().Add(5 * time.Second)
+ for time.Now().Before(deadline) {
+ if _, _, n := f.b.calls(); n > before {
+ return
+ }
+ time.Sleep(time.Millisecond)
+ }
+ }
+ f.b.gmu.Unlock()
+ f.rev.Add(1)
+ if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone {
+ t.Fatalf("later overlap Swap = %+v", res)
+ }
+ ev := f.b.events()
+ raise := -1
+ for i, x := range ev {
+ if strings.HasPrefix(x, "raise:") {
+ raise = i
+ }
+ }
+ for _, x := range ev[raise+1:] {
+ if strings.HasPrefix(x, "switch:") {
+ break
+ }
+ if strings.HasPrefix(x, "restore:") {
+ t.Fatalf("a parent restore (%s) was written after the later overlap raised the parent and before its switch, while two generations ran: %v", x, ev[raise:])
+ }
+ }
+ })
+}