carlos / platform Public

zdd plan 3 final fix: hold, sidecar-join and worker-proof tests use controlled ordering (T6, T7, T11)

T6: the hold witness learns its request is held on the stopping entry
from the request context's first Done() (doneAskedCtx), asserts it has
not been answered before the gate opens, and checks Drain against the
measured gated interval instead of a 50 ms sleep.
T7: the stop's join on an in-flight sidecar start reports itself through
the join seam (joinWaiting now names the join: "sidecar start" or
"operation set"); the test waits for it, or for the instance's stop,
instead of a 200 ms bounded look.
T11: the worker-proof tests assert the host-cgroup kill was asked, drop
the sleep (the stop pends only after its proof failed), and add a
positive control: once the proof heals, the crash restart respawns.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 2 weeks ago 89319f5d4fcc2bccd2fa1edcde9765745255ab3c parent 25a338d

6 files changed, +89 −21

internal/activator/activator.go +11 −6
internal/activator/opset.go +1 −1
internal/activator/opset_test.go +5 −1
internal/activator/swap_hold_test.go +28 −4
internal/activator/swap_sidecar_test.go +22 −7
internal/activator/worker_gen_test.go +22 −2

diff --git a/internal/activator/activator.go b/internal/activator/activator.go

index a984c8415..94bf86cd9 100644

--- a/internal/activator/activator.go

+++ b/internal/activator/activator.go

@@ -943,11 +943,13 @@ type Service struct {

// names them. Both under a.mu.

genPending map[pendingGeneration]pendingBuild

genPendingSeq uint64

- // joinWaiting, when set, is called by the barrier (joinOps) once it has

- // cancelled the attempts and is about to wait on a non-empty set. It is

- // a test seam: the positive signal that a stop has reached the barrier.

- // Set before the stop it observes; nil in production.

- joinWaiting func(host string)

+ // joinWaiting, when set, is called by a terminal stop's joins as each

+ // is about to wait: what is "operation set" (the barrier, joinOps, once

+ // it has cancelled the attempts, on a non-empty set) or "sidecar start"

+ // (stop() joining an in-flight supervised sidecar start). It is a test

+ // seam: the positive signal that a stop has reached a join. Set before

+ // the stop it observes; nil in production.

+ joinWaiting func(host, what string)

// genHealth, when set, replaces a generation's health check. It is a

// test seam. Nil is waitHealthyCtx on the generation's own socket (spec

// §7.1: health uses the handle, never AppSocket).

@@ -2764,9 +2766,12 @@ func (a *Service) stop(e *actEntry) {

// finishes, however it finishes, so this waits on that signal directly

// rather than polling sidecarStarting on a timer.

a.mu.Lock()

- startDone := e.sidecarStartDone

+ startDone, hook := e.sidecarStartDone, a.joinWaiting

a.mu.Unlock()

if startDone != nil {

+ if hook != nil {

+ hook(e.host, "sidecar start")

+ }

if a.gens {

<-startDone // the start, and every process it left, is accounted for

} else {

diff --git a/internal/activator/opset.go b/internal/activator/opset.go

index 0d17afe9c..5156eac36 100644

--- a/internal/activator/opset.go

+++ b/internal/activator/opset.go

@@ -122,7 +122,7 @@ func (a *Service) joinOps(e *actEntry) bool {

return true

}

if hook != nil {

- hook(e.host)

+ hook(e.host, "operation set")

}

bound := time.NewTimer(opJoinBound)

defer bound.Stop()

diff --git a/internal/activator/opset_test.go b/internal/activator/opset_test.go

index 933c76dcb..c4cebf1e7 100644

--- a/internal/activator/opset_test.go

+++ b/internal/activator/opset_test.go

@@ -109,7 +109,11 @@ func TestSwapWitnessBarrierJoinsBeforeTheLease(t *testing.T) {

joining := make(chan struct{})

var joinOnce sync.Once

x.a.mu.Lock()

- x.a.joinWaiting = func(string) { joinOnce.Do(func() { close(joining) }) }

+ x.a.joinWaiting = func(_, what string) {

+ if what == "operation set" {

+ joinOnce.Do(func() { close(joining) })

+ }

x.a.mu.Unlock()

gate := make(chan struct{})

member := &fakeProc{label: "gen:9", rec: x.b, stopGate: gate}

diff --git a/internal/activator/swap_hold_test.go b/internal/activator/swap_hold_test.go

index b6d41343e..fc1ce0a33 100644

--- a/internal/activator/swap_hold_test.go

+++ b/internal/activator/swap_hold_test.go

@@ -32,15 +32,25 @@ func TestSwapWitnessHoldExecsTheStagedBuild(t *testing.T) {

e := f.a.entries[f.host]

return e != nil && e.state == actStopping

})

+ stopSeen := time.Now() // the old generation's stop began before this

served := make(chan error, 1)

+ waiting := &doneAskedCtx{Context: context.Background(), asked: make(chan struct{})}

go func() {

- release, err := f.a.beginRequest(context.Background(), testRoute(f.host))

+ release, err := f.a.beginRequest(waiting, testRoute(f.host))

if err == nil {

release()

}

served <- err

}()

- time.Sleep(50 * time.Millisecond)

+ // The request has reached its wait on the stopping entry: begin asks

+ // its context's Done only there (the first select it reaches).

+ <-waiting.asked

+ select {

+ case err := <-served:

+ t.Fatalf("the request was answered (%v) while the old generation was still stopping", err)

+ default:

+ }

+ gateOpen := time.Now()

close(gate)

var res SwapResult

select {

@@ -51,8 +61,8 @@ func TestSwapWitnessHoldExecsTheStagedBuild(t *testing.T) {

if res.Outcome != SwapDone || res.Gen.Build != "v2" || res.Gen.StagedPath != f.staged {

t.Fatalf("Swap = %+v", res)

}

- if res.Drain < 40*time.Millisecond {

- t.Fatalf("Drain = %v, want the old generation's gated stop (spec §9.4)", res.Drain)

+ if held := gateOpen.Sub(stopSeen); res.Drain < held {

+ t.Fatalf("Drain = %v, want at least the old generation's gated stop, %v (spec §9.4)", res.Drain, held)

}

if err := <-served; err != nil {

t.Fatalf("the held request was refused: %v", err)

@@ -70,6 +80,20 @@ func TestSwapWitnessHoldExecsTheStagedBuild(t *testing.T) {

}

+// doneAskedCtx is a context that signals the first time its Done channel

+// is asked for: begin() asks only when it reaches a wait, so the signal is

+// the request held on the entry.

+type doneAskedCtx struct {

+ context.Context

+ once sync.Once

+ asked chan struct{}

+func (c *doneAskedCtx) Done() <-chan struct{} {

+ c.once.Do(func() { close(c.asked) })

+ return c.Context.Done()

// A hold canary is not done when it is healthy: it must survive the 10 s

// settle (spec §6.2, §8.5). Without the settle, swapHold answers SwapDone

// the moment the pinned wake is ready, and a crash a second later neither

diff --git a/internal/activator/swap_sidecar_test.go b/internal/activator/swap_sidecar_test.go

index a40779728..76548d7ae 100644

--- a/internal/activator/swap_sidecar_test.go

+++ b/internal/activator/swap_sidecar_test.go

@@ -7,6 +7,7 @@ import (

"path/filepath"

"slices"

"strings"

+ "sync"

"testing"

"time"

@@ -336,16 +337,30 @@ func TestSwapStopJoinsTheSidecarAStartLeftBehind(t *testing.T) {

f.a.sweepOnce() // the supervisor's start blocks

waitFor(t, "the supervisor's start", func() bool { return len(f.b.sidecarGenPaths()) == 2 })

rep := f.b.lastRep()

+ joining := make(chan struct{})

+ var joinOnce sync.Once

+ f.a.mu.Lock()

+ f.a.joinWaiting = func(_, what string) {

+ if what == "sidecar start" {

+ joinOnce.Do(func() { close(joining) })

+ }

+ f.a.mu.Unlock()

f.a.Bounce(f.host)

// The stop is joined on the supervisor's start: it goes no further (no

- // instance stop, so no lease release) while the start is in flight. Only

- // a bounded look can show an absence, so it fails the moment the instance

- // is stopped.

- for range 40 {

- if slices.Contains(f.b.events(), "stop:instance") {

- t.Fatal("the stop went past its join on the in-flight sidecar start")

+ // instance stop, so no lease release) while the start is in flight. Wait

+ // for the stop to reach that join (the join seam's positive signal), or,

+ // if something skipped it, for the instance's stop.

+ waitFor(t, "the stop at its sidecar-start join", func() bool {

+ select {

+ case <-joining:

+ return true

+ default:

}

- time.Sleep(5 * time.Millisecond)

+ return slices.Contains(f.b.events(), "stop:instance")

+ })

+ if slices.Contains(f.b.events(), "stop:instance") || rep.isClosed() {

+ t.Fatal("the stop went past its join on the in-flight sidecar start")

}

close(startGate) // the start lands on a stopping entry; its sidecar's Stop blocks

waitFor(t, "the left-behind sidecar's stop to be in flight", func() bool { return f.b.sidecarStoppingIn(1) })

diff --git a/internal/activator/worker_gen_test.go b/internal/activator/worker_gen_test.go

index 4cb4b9401..773fb1382 100644

--- a/internal/activator/worker_gen_test.go

+++ b/internal/activator/worker_gen_test.go

@@ -82,6 +82,12 @@ func (h *hostKillBackend) KillHost(context.Context, registry.Route) error {

return nil

}

+func (h *hostKillBackend) killCount() int {

+ h.hmu.Lock()

+ defer h.hmu.Unlock()

+ return h.kills

func (h *hostKillBackend) heal() {

h.hmu.Lock()

h.bad = false

@@ -105,13 +111,17 @@ func TestWorkerTerminationProvesItsHostCgroup(t *testing.T) {

}

rep := b.lastRep()

a.Bounce(rt.Host)

+ // The stop pends only once it has decided the lease stays: its proof

+ // (KillHost) ran and failed, and the release was not made.

waitFor(t, "the stop to pend", func() bool {

a.mu.Lock()

defer a.mu.Unlock()

e := a.entries[rt.Host]

return e != nil && e.pendingStop

})

- time.Sleep(50 * time.Millisecond)

+ if n := hk.killCount(); n == 0 {

+ t.Fatal("the stop pended without asking the host cgroup's proof")

+ }

if rep.isClosed() {

t.Fatal("the lease was released while the worker's host cgroup still held a process")

}

@@ -182,7 +192,7 @@ func TestWorkerTerminationProvesItsHostCgroup(t *testing.T) {

old := sidecarSettle

sidecarSettle = 20 * time.Millisecond

t.Cleanup(func() { sidecarSettle = old })

- a, b, _, _ := testActivator(t, time.Hour)

+ a, b, clk, _ := testActivator(t, time.Hour)

a.gens = true

hk := &hostKillBackend{fakeBackend: b}

a.backend = hk

@@ -194,11 +204,21 @@ func TestWorkerTerminationProvesItsHostCgroup(t *testing.T) {

hk.bad = true

hk.hmu.Unlock()

b.lastWorker().crash()

+ kills := hk.killCount()

a.sweepOnce() // the respawn retires the crashed child, and its proof fails

waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(rt.Host) })

+ if got := hk.killCount(); got <= kills {

+ t.Fatalf("host cgroup kills = %d (was %d): the respawn never asked the old child's proof", got, kills)

+ }

if got := b.workerStarts(); got != 1 {

t.Fatalf("worker starts = %d: a respawn exec'd beside an unproven old child", got)

}

+ // Positive control: once the proof succeeds, the next sweep's

+ // crash restart respawns the worker.

+ hk.heal()

+ clk.advance(sidecarRestartMax + time.Second) // past the crash restart's backoff

+ a.sweepOnce()

+ waitFor(t, "the respawn after the proof healed", func() bool { return b.workerStarts() == 2 && !a.entryRespawning(rt.Host) })

})

}