amadan

carlos / platform Public

zdd plan 3 final fix: barrier and settle tests wait on positive signals, not sleeps (T2, T5)

The barrier witness waits for the stop to reach the barrier through a
test seam (Service.joinWaiting, called by joinOps once it is about to
wait on a non-empty set), or for the incumbent's stop or the lease
release if something skipped the barrier, instead of a 50 ms sleep. The
settle tests place an exit or a cancel at the observation's own poll
(fakeProc.exitOnPoll, cancelOnPoll) and start the intentional stop once
the settle has polled the successor, instead of sleeping.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 25a338d510bf5c0b21dedafc2e69c9228a5a6a06 parent 65375b0
5 files changed, +80 −10
  • internal/activator/activator.go +5 −0
  • internal/activator/activator_test.go +23 −0
  • internal/activator/opset.go +4 −0
  • internal/activator/opset_test.go +17 −5
  • internal/activator/settle_test.go +31 −5
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index dfae1aab2..a984c8415 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -943,6 +943,11 @@ 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)
// 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).
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index f396d12c8..f6ea62021 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -129,6 +129,11 @@ type fakeProc struct {
// the co-spawn/co-stop tests can assert interleavings.
label string
rec *fakeBackend
+ // polls counts Exited() calls; exitAt, when > 0, makes the process exit
+ // on that call: a death a test places inside an observation (a settle
+ // window) by the observation's own polling, with no sleep.
+ polls int
+ exitAt int
}
func (p *fakeProc) Stop(context.Context) error {
@@ -167,9 +172,27 @@ func (p *fakeProc) setStopGate(gate chan struct{}) {
func (p *fakeProc) Exited() bool {
p.mu.Lock()
defer p.mu.Unlock()
+ p.polls++
+ if p.exitAt > 0 && p.polls >= p.exitAt {
+ p.dead = true
+ }
return p.dead
}
+// exitOnPoll makes the process exit on its n-th Exited() call from now.
+func (p *fakeProc) exitOnPoll(n int) {
+ p.mu.Lock()
+ p.exitAt = p.polls + n
+ p.mu.Unlock()
+}
+
+// pollCount is how many times Exited() has been asked.
+func (p *fakeProc) pollCount() int {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.polls
+}
+
// Crashed models the exec child's own answer (backend_exec.go's childProc):
// only a definitely non-zero exit is a crash. kill() therefore models a CLEAN
// exit — a bot that finished its work — and crash() models a child that fell
diff --git a/internal/activator/opset.go b/internal/activator/opset.go
index 268c53f4e..0d17afe9c 100644
--- a/internal/activator/opset.go
+++ b/internal/activator/opset.go
@@ -116,10 +116,14 @@ func (a *Service) joinOps(e *actEntry) bool {
}
waits = append(waits, m.done)
}
+ hook := a.joinWaiting
a.mu.Unlock()
if len(waits) == 0 {
return true
}
+ if hook != nil {
+ hook(e.host)
+ }
bound := time.NewTimer(opJoinBound)
defer bound.Stop()
for _, w := range waits {
diff --git a/internal/activator/opset_test.go b/internal/activator/opset_test.go
index 2862871ec..933c76dcb 100644
--- a/internal/activator/opset_test.go
+++ b/internal/activator/opset_test.go
@@ -106,6 +106,11 @@ func TestSwapWitnessBarrierJoinsBeforeTheLease(t *testing.T) {
x := awakeForBarrier(t, true)
sig := &stopSignal{began: make(chan struct{})}
x.a.Logs = sig
+ joining := make(chan struct{})
+ var joinOnce sync.Once
+ x.a.mu.Lock()
+ x.a.joinWaiting = func(string) { joinOnce.Do(func() { close(joining) }) }
+ x.a.mu.Unlock()
gate := make(chan struct{})
member := &fakeProc{label: "gen:9", rec: x.b, stopGate: gate}
plantMember(t, x.a, x.host, role, member, 9)
@@ -116,11 +121,18 @@ func TestSwapWitnessBarrierJoinsBeforeTheLease(t *testing.T) {
case <-time.After(5 * time.Second):
t.Fatal("the stop never began")
}
- // The stop has begun; nothing between its first line and the
- // barrier blocks. There is no signal at the barrier itself, so
- // this short wait covers those few instructions.
- time.Sleep(50 * time.Millisecond)
- if rep.isClosed() || incumbent.isStopped() {
+ // The stop has begun. Wait for it to reach the barrier (the
+ // join seam's positive signal), or, if something skipped the
+ // barrier, for the incumbent's stop or the lease release.
+ waitFor(t, "the stop at the barrier", func() bool {
+ select {
+ case <-joining:
+ return true
+ default:
+ }
+ return rep.isClosed() || incumbent.stopEntered() > 0
+ })
+ if rep.isClosed() || incumbent.stopEntered() > 0 {
t.Fatal("the stop went past the barrier while a generation was still draining")
}
close(gate)
diff --git a/internal/activator/settle_test.go b/internal/activator/settle_test.go
index b70505e95..207943785 100644
--- a/internal/activator/settle_test.go
+++ b/internal/activator/settle_test.go
@@ -12,14 +12,18 @@ func TestObserveSettle(t *testing.T) {
if r := observeSettle(context.Background(), &fakeProc{}, 20*time.Millisecond); r != settleOK {
t.Fatalf("a process that stayed up: %v", r)
}
+ // A process that dies inside the window: alive at the first check, dead
+ // at the observation's next poll (no sleep places the death).
p := &fakeProc{}
- go func() { time.Sleep(5 * time.Millisecond); p.exit() }()
+ p.exitOnPoll(2)
if r := observeSettle(context.Background(), p, time.Second); r != settleExited {
t.Fatalf("a process that died: %v", r)
}
+ // An observation cancelled inside the window: the cancel happens at its
+ // first poll after the opening check.
ctx, cancel := context.WithCancel(context.Background())
- go func() { time.Sleep(5 * time.Millisecond); cancel() }()
- if r := observeSettle(ctx, &fakeProc{}, time.Second); r != settleCancelled {
+ defer cancel()
+ if r := observeSettle(ctx, &cancelOnPoll{at: 2, cancel: cancel}, time.Second); r != settleCancelled {
t.Fatalf("a cancelled observation: %v", r)
}
// Cancellation wins over an exit seen at the same moment (spec §6.4).
@@ -32,6 +36,23 @@ func TestObserveSettle(t *testing.T) {
}
}
+// cancelOnPoll is a live process whose at-th Exited() call cancels the
+// observation watching it.
+type cancelOnPoll struct {
+ polls int
+ at int
+ cancel context.CancelFunc
+}
+
+func (p *cancelOnPoll) Stop(context.Context) error { return nil }
+func (p *cancelOnPoll) Exited() bool {
+ p.polls++
+ if p.polls == p.at {
+ p.cancel()
+ }
+ return false
+}
+
func TestSwapCanarySettle(t *testing.T) {
old := swapSettle
t.Cleanup(func() { swapSettle = old })
@@ -40,7 +61,9 @@ func TestSwapCanarySettle(t *testing.T) {
swapSettle = 2 * time.Second
f := newSwapFixture(t)
f.a.genHealth = func(context.Context, Gen) error {
- go func() { time.Sleep(20 * time.Millisecond); _, p := f.b.lastGen(); p.exit() }()
+ // Alive at the settle's opening check, dead at its next poll.
+ _, p := f.b.lastGen()
+ p.exitOnPoll(2)
return nil
}
at := f.attempt(ModeOverlap)
@@ -116,7 +139,10 @@ func TestSwapIntentionalStopDuringSettleIsNotAFailure(t *testing.T) {
results := make(chan SwapResult, 1)
go func() { results <- f.a.Swap(context.Background(), at, f.newest) }()
<-healthy
- time.Sleep(30 * time.Millisecond)
+ // The settle has begun once it has polled the successor (nothing
+ // else asks it between the health check and the settle).
+ _, succ := f.b.lastGen()
+ waitFor(t, "the settle's first poll", func() bool { return succ.pollCount() > 0 })
tc.stop(f)
var res SwapResult
select {