amadan

carlos / platform Public

zdd plan 3 task 12: fix round 1: the spawning subtest uses a spawn that outlives cancellation

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude ecef92996d607e05c2915164e131f4d6327c9ab1 parent 520d267
2 files changed, +52 −5
  • internal/activator/swap_fixture_test.go +10 −0
  • internal/activator/swap_review_test.go +42 −5
diff --git a/internal/activator/swap_fixture_test.go b/internal/activator/swap_fixture_test.go
index 80a5f71b1..114884bd3 100644
--- a/internal/activator/swap_fixture_test.go
+++ b/internal/activator/swap_fixture_test.go
@@ -31,6 +31,10 @@ type fakeSwapBackend struct {
genProcs []*fakeProc
genErr error
genGate chan struct{} // when set, StartGen blocks until it closes or ctx ends
+ // genHold, when set, makes StartStaged wait for it to close IGNORING ctx,
+ // then return a live successor even though the attempt was cancelled: a
+ // spawn that outlives cancellation, which the barrier must wait for.
+ genHold chan struct{}
raises int
restores int
// genSocketDir, when set, puts generation sockets there (Task 12), and
@@ -76,6 +80,12 @@ func (g *fakeSwapBackend) StartStaged(ctx context.Context, _ registry.Route, gen
if err != nil {
return gen, nil, err
}
+ g.gmu.Lock()
+ hold := g.genHold
+ g.gmu.Unlock()
+ if hold != nil {
+ <-hold
+ }
gen.Socket = fmt.Sprintf("/run/carlos/g/test.%d.sock", gen.N)
if g.genSocketDir != "" {
gen.Socket = filepath.Join(g.genSocketDir, fmt.Sprintf("g%d.sock", gen.N))
diff --git a/internal/activator/swap_review_test.go b/internal/activator/swap_review_test.go
index a3e7c1025..d189ec2a8 100644
--- a/internal/activator/swap_review_test.go
+++ b/internal/activator/swap_review_test.go
@@ -42,23 +42,60 @@ func TestSwapWitnessAgentRestartMidSwap(t *testing.T) {
t.Run("successor spawning", func(t *testing.T) {
f := newSwapFixture(t)
- f.b.genGate = make(chan struct{}) // never opened: only cancellation ends the spawn
+ // A spawn that outlives cancellation: StartStaged ignores its ctx
+ // until the hold opens, then returns a live successor. Shutdown
+ // cancels the attempt, but only the join makes it wait for that
+ // process, so the incumbent may not stop and the lease may not go
+ // back until the successor exists and has been stopped.
+ hold := make(chan struct{})
+ f.b.genHold = hold
incumbent, rep := f.b.lastProc(), f.b.lastRep()
var leaked atomic.Bool
rep.setOnClose(func() {
- if !incumbent.Exited() {
+ if _, succ := f.b.lastGen(); succ == nil || !succ.Exited() || !incumbent.Exited() {
leaked.Store(true)
}
})
results := make(chan SwapResult, 1)
go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }()
waitFor(t, "the successor's spawn", func() bool { return indexPrefix(f.b.events(), "startGen:") >= 0 })
- ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
- f.a.Shutdown(ctx)
- cancel()
+ shut := make(chan struct{})
+ go func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ f.a.Shutdown(ctx)
+ close(shut)
+ }()
+ waitFor(t, "Shutdown to reach the entry", func() bool {
+ f.a.mu.Lock()
+ defer f.a.mu.Unlock()
+ return f.a.entries[f.host].state == actStopping
+ })
+ // The barrier holds the stop back while the spawn is in flight. As in
+ // the draining case nothing signals the barrier itself, so the
+ // negative is watched for a window; a stop that passed it would stop
+ // the incumbent at once, and the poll fails then.
+ for deadline := time.Now().Add(200 * time.Millisecond); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) {
+ if incumbent.isStopped() {
+ t.Fatal("the incumbent was stopped while a successor was still spawning")
+ }
+ if rep.isClosed() {
+ t.Fatal("the lease went back while a successor was still spawning")
+ }
+ }
+ select {
+ case <-shut:
+ t.Fatal("Shutdown returned with a successor still spawning")
+ default:
+ }
+ close(hold)
+ <-shut
if res := <-results; res.Outcome != SwapAbandoned {
t.Fatalf("Swap = %+v, want abandoned by the shutdown", res)
}
+ if _, succ := f.b.lastGen(); succ == nil || !succ.Exited() {
+ t.Fatal("the successor that outlived the cancellation was never stopped")
+ }
if !rep.isClosed() || leaked.Load() {
t.Fatalf("lease closed %v, released while a generation ran %v", rep.isClosed(), leaked.Load())
}