carlos / platform Public

zdd-readiness: final review round 1, status probes are wired before any scheduler starts

Finding 7 (Astra, Medium, Merge): two status tests assigned h.m.d.Act after h.pass() had started scheduler goroutines that read it. The harness now hands its manager a probeAct (the fake behind an atomically swappable hook) at build, and the tests install their probe through the hook. The lock probe now makes Status ask Owns too (an asleep host) and asserts both questions were asked with no Manager or scheduler lock held. -race -count=50 clean.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude agent, operated by paul@keymail.dev ccf0d31cd361be073311d6daca9e557c04ff11d9 parent bbb7fc8
3 files changed, +68 −41
  • internal/rollout/fake_test.go +39 −1
  • internal/rollout/finalreview1_test.go +1 −1
  • internal/rollout/status_test.go +28 −39
diff --git a/internal/rollout/fake_test.go b/internal/rollout/fake_test.go
index e41b5c6f4..fa5528d89 100644
--- a/internal/rollout/fake_test.go
+++ b/internal/rollout/fake_test.go
@@ -81,6 +81,42 @@ func newFakeAct() *fakeAct {
return &fakeAct{gens: map[string]activator.Gen{}, owned: map[string]bool{}, gates: map[string]chan struct{}{}}
}
+// probeAct is the Activator the harness hands its manager: the fake, with
+// a hook a test may install at any moment, run before every Owns and
+// Current. It is wired in at harness build, before any scheduler starts,
+// and the hook is swapped atomically (final review round 1, finding 7,
+// Astra): a test that replaced Deps.Act after a Pass raced the scheduler
+// goroutines that read it.
+type probeAct struct {
+ *fakeAct
+ hook atomic.Pointer[func(method, host string)]
+}
+
+// onAsk installs f as the hook; nil removes it.
+func (p *probeAct) onAsk(f func(method, host string)) {
+ if f == nil {
+ p.hook.Store(nil)
+ return
+ }
+ p.hook.Store(&f)
+}
+
+func (p *probeAct) ask(method, host string) {
+ if f := p.hook.Load(); f != nil {
+ (*f)(method, host)
+ }
+}
+
+func (p *probeAct) Owns(host string) bool {
+ p.ask("Owns", host)
+ return p.fakeAct.Owns(host)
+}
+
+func (p *probeAct) Current(host string) (activator.Gen, bool) {
+ p.ask("Current", host)
+ return p.fakeAct.Current(host)
+}
+
func (f *fakeAct) Current(host string) (activator.Gen, bool) {
f.mu.Lock()
defer f.mu.Unlock()
@@ -331,6 +367,7 @@ func (e *eventLog) count(substr string) int {
type harness struct {
t *testing.T
act *fakeAct
+ probe *probeAct // the manager's Activator: act behind a test hook
rows *fakeRoutes
book *FailureBook
m *Manager
@@ -365,6 +402,7 @@ func newHarness(t *testing.T, hosts ...string) *harness {
pointers: map[string]release.ChannelPointer{},
events: &eventLog{}}
h.book.now = h.clock // a marker's At follows the harness clock
+ h.probe = &probeAct{fakeAct: h.act}
h.stage("v1")
for _, host := range hosts {
h.rows.put(registry.Route{Host: host, Kind: "instance", App: "blog", Account: "bab", Status: "ready",
@@ -374,7 +412,7 @@ func newHarness(t *testing.T, hosts ...string) *harness {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
h.m = NewManager(ctx, Deps{
- Act: h.act, Routes: h.rows, Book: h.book, OptDir: h.opt, DataDir: h.data, Box: "box-1",
+ Act: h.probe, Routes: h.rows, Book: h.book, OptDir: h.opt, DataDir: h.data, Box: "box-1",
Demand: h.demand, Policy: h.policy, Pointer: h.pointerFor, Events: h.events,
Touch: func() { h.touched.Add(1) }, Now: h.clock, Logf: func(string, ...any) {},
})
diff --git a/internal/rollout/finalreview1_test.go b/internal/rollout/finalreview1_test.go
index 2e7b93ed4..157ccb93f 100644
--- a/internal/rollout/finalreview1_test.go
+++ b/internal/rollout/finalreview1_test.go
@@ -21,7 +21,7 @@ func (h *harness) restartOn(book *FailureBook) {
ctx, cancel := context.WithCancel(context.Background())
h.t.Cleanup(cancel)
h.m = NewManager(ctx, Deps{
- Act: h.act, Routes: h.rows, Book: book, OptDir: h.opt, DataDir: h.data, Box: "box-1",
+ Act: h.probe, Routes: h.rows, Book: book, OptDir: h.opt, DataDir: h.data, Box: "box-1",
Demand: h.demand, Policy: h.policy, Pointer: h.pointerFor, Events: h.events,
Touch: func() { h.touched.Add(1) }, Now: h.clock, Logf: func(string, ...any) {},
})
diff --git a/internal/rollout/status_test.go b/internal/rollout/status_test.go
index d42704637..0cf4333ca 100644
--- a/internal/rollout/status_test.go
+++ b/internal/rollout/status_test.go
@@ -251,9 +251,13 @@ func TestStatusReportsAStartingOrStoppingHostAsTransient(t *testing.T) {
}
// Status asks the activator with no Manager or scheduler lock held (the lock
-// order on Manager). The probe only TryLocks, so a violation fails the test
-// instead of deadlocking it. No attempt runs (nothing is installed), so no
-// other goroutine holds these locks.
+// order on Manager), for both of its questions: Current for every host, and
+// Owns for one with no current generation (b, asleep here). The probe only
+// TryLocks, so a violation fails the test instead of deadlocking it. No
+// attempt runs (nothing is installed), so no other goroutine holds these
+// locks for long. The probe is the harness's (probeAct), wired in before
+// any scheduler started; its hook is installed atomically (final review
+// round 1, finding 7).
func TestStatusAsksTheActivatorWithNoLockHeld(t *testing.T) {
h := newHarness(t, "a", "b")
h.pass()
@@ -266,29 +270,32 @@ func TestStatusAsksTheActivatorWithNoLockHeld(t *testing.T) {
if len(scheds) == 0 {
t.Fatal("the pass built no scheduler")
}
- probe := &lockProbeAct{fakeAct: h.act, m: h.m, scheds: scheds, t: t}
- h.m.d.Act = probe
+ h.act.mu.Lock()
+ delete(h.act.gens, "b") // asleep: Status asks Owns about it
+ h.act.mu.Unlock()
+ lp := &lockProbe{m: h.m, scheds: scheds, t: t, asked: map[string]*atomic.Int64{"Owns": {}, "Current": {}}}
+ h.probe.onAsk(lp.check)
+ defer h.probe.onAsk(nil)
h.m.Status()
- if probe.asked.Load() == 0 {
- t.Fatal("Status never asked the activator")
+ if lp.asked["Current"].Load() == 0 || lp.asked["Owns"].Load() == 0 {
+ t.Fatalf("Status asked Current %d and Owns %d time(s): want both asked", lp.asked["Current"].Load(), lp.asked["Owns"].Load())
}
}
-// lockProbeAct fails the test when Owns or Current is called while the
+// lockProbe fails the test when Owns or Current is asked by Status while the
// Manager's or a scheduler's mutex is held.
-type lockProbeAct struct {
- *fakeAct
+type lockProbe struct {
m *Manager
scheds []*scheduler
t *testing.T
- asked atomic.Int64
+ asked map[string]*atomic.Int64 // by method; built before the hook is installed, never written after
}
-func (p *lockProbeAct) check() {
+func (p *lockProbe) check(method, _ string) {
if !askedByStatus() {
return // the scheduler's own goroutine asks too; only Status is under test
}
- p.asked.Add(1)
+ p.asked[method].Add(1)
// A lock that Status holds stays held for the whole ask, while a
// scheduler goroutine's own hold is brief, so a violation is a lock that
// cannot be taken for a long while. TryLock, never Lock: a violation
@@ -303,11 +310,11 @@ func (p *lockProbeAct) check() {
return false
}
if !free(&p.m.mu) {
- p.t.Error("activator asked with Manager.mu held")
+ p.t.Errorf("activator asked (%s) with Manager.mu held", method)
}
for _, s := range p.scheds {
if !free(&s.mu) {
- p.t.Error("activator asked with a scheduler.mu held")
+ p.t.Errorf("activator asked (%s) with a scheduler.mu held", method)
}
}
}
@@ -328,16 +335,6 @@ func askedByStatus() bool {
}
}
-func (p *lockProbeAct) Owns(host string) bool {
- p.check()
- return p.fakeAct.Owns(host)
-}
-
-func (p *lockProbeAct) Current(host string) (activator.Gen, bool) {
- p.check()
- return p.fakeAct.Current(host)
-}
-
// A host that joins a scheduler after Status read the names was never asked of
// the activator, so nothing says it is settled: that report calls it
// transient, and the next one asks.
@@ -350,13 +347,17 @@ func TestStatusCallsAHostThatJoinedAfterTheNamesWereReadTransient(t *testing.T)
}
h.m.mu.Unlock()
var once atomic.Bool
- h.m.d.Act = &joinProbeAct{fakeAct: h.act, onAsk: func() {
+ h.probe.onAsk(func(method, _ string) {
+ if method != "Current" || !askedByStatus() { // not the scheduler's own goroutine
+ return
+ }
if once.CompareAndSwap(false, true) { // the names are read; now a host joins
s.mu.Lock()
s.q.hosts["late"] = &hostSlot{wantBuild: "v1", good: "v1", serving: "v1"}
s.mu.Unlock()
}
- }}
+ })
+ defer h.probe.onAsk(nil)
r := h.m.Status()
if late := r.Hosts["late"]; !late.Transient || late.Serving != "" || late.Awake {
t.Fatalf("late = %+v, want transient for the report that never asked", late)
@@ -365,15 +366,3 @@ func TestStatusCallsAHostThatJoinedAfterTheNamesWereReadTransient(t *testing.T)
t.Fatalf("a = %+v, want awake on v1", a)
}
}
-
-type joinProbeAct struct {
- *fakeAct
- onAsk func()
-}
-
-func (p *joinProbeAct) Current(host string) (activator.Gen, bool) {
- if askedByStatus() { // not the scheduler's own goroutine
- p.onAsk()
- }
- return p.fakeAct.Current(host)
-}