amadan

carlos / platform Public

activator: probe retirement and cancellation races (Plan 3 Task 5 fix round 1, findings 6, 7)

6. RetireExcept retires a probe entry only when its held-probe snapshot
   covered it; one published between the snapshot and the retirement
   pass is left for the next reload. Deterministic test through a seam
   between the two locks.
7. A probe's wake context derives from ProbeWake's, set before the entry
   is published, so a cancel at that boundary stops the wake before
   anything is admitted.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 1436fb079c2d51d71f4eebe3611d4bb3d0fa1ec0 parent 5e803ac
3 files changed, +106 −23
  • internal/activator/activator.go +11 −3
  • internal/activator/probewake.go +28 −20
  • internal/activator/probewake_test.go +67 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 7e5bdeb1e..5f2f780e7 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -991,7 +991,12 @@ func (a *Service) consumeDueTime(e *actEntry) {
// requests still need the wake to finish even if the requester who started it
// hangs up.
func (a *Service) wake(e *actEntry) {
- ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+ parent := context.Background()
+ if e.probe != nil && e.probe.parent != nil {
+ // A probe's wake ends with its caller (ProbeWake).
+ parent = e.probe.parent
+ }
+ ctx, cancel := context.WithTimeout(parent, 5*time.Minute)
defer cancel()
a.mu.Lock()
e.wakeCancel = cancel
@@ -2945,11 +2950,14 @@ func (a *Service) sweepOnce() {
func (a *Service) RetireExcept(keep map[string]bool) {
// A probe's entry (ProbeWake) is held for its move while the registry
// row is: the edge manages ready rows only, so it is never in keep.
- held := a.heldProbes(keep)
+ seen, held := a.heldProbes(keep)
+ if retireBetweenLocks != nil {
+ retireBetweenLocks()
+ }
a.mu.Lock()
var waking []*actEntry
for _, e := range a.entries {
- if keep[e.host] || held[e] {
+ if keep[e.host] || held[e] || (e.probe != nil && !seen[e]) {
continue
}
// Workers are outside this reconcile's world entirely. `keep` is the
diff --git a/internal/activator/probewake.go b/internal/activator/probewake.go
index 51a6714dc..6181ce680 100644
--- a/internal/activator/probewake.go
+++ b/internal/activator/probewake.go
@@ -74,11 +74,20 @@ type ProbeResult struct {
Generation string
}
+// Test seams: retireBetweenLocks runs in RetireExcept between the probe
+// snapshot and the retirement pass; probeWakePublished runs in ProbeWake
+// after the entry is published and before its wake starts.
+var retireBetweenLocks, probeWakePublished func()
+
// probeEntry marks an actEntry ProbeWake made.
type probeEntry struct {
policy startPolicy
op, holdOp string
admitted *hostclaim.Observed
+ // parent is ProbeWake's context: the wake's own context derives from
+ // it, so a cancel is in force before the entry is published or its
+ // goroutine starts (fix round 1, finding 7).
+ parent context.Context
}
// ProbeWake runs one probe attempt for cfg.Host and waits for it. On
@@ -104,7 +113,7 @@ func (a *Service) ProbeWake(ctx context.Context, cfg ProbeConfig) (ProbeResult,
return ProbeResult{}, errors.Join(ErrProbeRefused, errProbeRow)
}
e := &actEntry{host: cfg.Host, route: *rt, state: actWaking, ready: make(chan struct{}),
- probe: &probeEntry{op: cfg.Op, holdOp: cfg.HoldOp,
+ probe: &probeEntry{op: cfg.Op, holdOp: cfg.HoldOp, parent: ctx,
policy: startPolicy{MaxStarts: probeMaxStarts, Window: probeWindow30, Deadline: cfg.Deadline, Probe: true, Prepared: cfg.Prepared, Inputs: cfg.Inputs}}}
a.mu.Lock()
switch {
@@ -117,17 +126,11 @@ func (a *Service) ProbeWake(ctx context.Context, cfg ProbeConfig) (ProbeResult,
}
a.entries[cfg.Host] = e
a.mu.Unlock()
- go a.wake(e)
- select {
- case <-e.ready:
- case <-ctx.Done():
- a.mu.Lock()
- if e.wakeCancel != nil {
- e.wakeCancel()
- }
- a.mu.Unlock()
- <-e.ready
+ if probeWakePublished != nil {
+ probeWakePublished()
}
+ go a.wake(e)
+ <-e.ready
// e.err and e.probe are written before e.ready closes.
if e.err != nil {
return ProbeResult{}, e.err
@@ -155,12 +158,14 @@ func probeRefused(e *actEntry, err error) error {
return errors.Join(ErrProbeRefused, err)
}
-// heldProbes is the set of entries RetireExcept keeps beyond keep: probe
-// entries whose registry row is still held by the probe's hold op. The edge
-// manages only ready rows (edge.Reload), so a held probe is never in keep.
-// The rows are read outside a.mu. A row that cannot be read keeps its entry
-// for this reload: the next reload reads it again.
-func (a *Service) heldProbes(keep map[string]bool) map[*actEntry]bool {
+// heldProbes snapshots the probe entries RetireExcept may retire (seen) and
+// those of them it keeps beyond keep (held): probe entries whose registry
+// row is still held by the probe's hold op. The edge manages only ready rows
+// (edge.Reload), so a held probe is never in keep. The rows are read outside
+// a.mu, so RetireExcept retires a probe only when this snapshot covered it
+// (fix round 1, finding 6): one published after it is left for the next
+// reload. A row that cannot be read keeps its entry for this reload.
+func (a *Service) heldProbes(keep map[string]bool) (seen, held map[*actEntry]bool) {
a.mu.Lock()
var probes []*actEntry
for _, e := range a.entries {
@@ -169,9 +174,12 @@ func (a *Service) heldProbes(keep map[string]bool) map[*actEntry]bool {
}
}
a.mu.Unlock()
- held := map[*actEntry]bool{}
+ seen, held = map[*actEntry]bool{}, map[*actEntry]bool{}
+ for _, e := range probes {
+ seen[e] = true
+ }
if a.Routes == nil {
- return held
+ return seen, held
}
for _, e := range probes {
rt, err := a.Routes.Get(e.host)
@@ -184,5 +192,5 @@ func (a *Service) heldProbes(keep map[string]bool) map[*actEntry]bool {
held[e] = true
}
}
- return held
+ return seen, held
}
diff --git a/internal/activator/probewake_test.go b/internal/activator/probewake_test.go
index 1c8a0a866..17eff56aa 100644
--- a/internal/activator/probewake_test.go
+++ b/internal/activator/probewake_test.go
@@ -297,3 +297,70 @@ func TestRetireExceptKeepsTheHeldProbeEntry(t *testing.T) {
a.RetireExcept(map[string]bool{})
waitFor(t, "the entry no longer held for its probe to retire", func() bool { return !a.Owns("one.example.com") })
}
+
+// Finding 6: a probe entry published after RetireExcept's snapshot of the
+// held probes, and before its retirement pass, is not retired: only entries
+// the snapshot validated are.
+func TestRetireExceptKeepsAProbeInsertedBetweenTheLocks(t *testing.T) {
+ a, _, _, reg := testActivator(t, time.Hour)
+ a.Routes = reg
+ held := registry.Route{Host: "two.example.com", Kind: "instance", Socket: "/tmp/nope.sock", DBPath: "/tmp/two.example.com.db", Hibernate: true}
+ if err := reg.Upsert(held); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.SetHold("two.example.com", "landing", pwOp); err != nil {
+ t.Fatal(err)
+ }
+ inserted := &actEntry{host: "two.example.com", route: held, state: actAwake, ready: make(chan struct{}), proc: &fakeProc{}, rep: &fakeRep{},
+ probe: &probeEntry{policy: startPolicy{Probe: true}, op: pwOp, holdOp: pwOp}}
+ close(inserted.ready)
+ retireBetweenLocks = func() {
+ a.mu.Lock()
+ a.entries[inserted.host] = inserted
+ a.mu.Unlock()
+ }
+ t.Cleanup(func() { retireBetweenLocks = nil })
+ a.RetireExcept(map[string]bool{})
+ retireBetweenLocks = nil
+ a.mu.Lock()
+ state := inserted.state
+ a.mu.Unlock()
+ if state != actAwake || !a.Owns(inserted.host) {
+ t.Fatal("a held probe inserted between the locks was retired")
+ }
+ // The next reload validates it like any other.
+ a.RetireExcept(map[string]bool{})
+ if !a.Owns(inserted.host) {
+ t.Fatal("a held probe was retired on the next reload")
+ }
+}
+
+// Finding 7: a cancel that lands after the probe's entry is published and
+// before its wake starts still stops the wake: nothing is admitted.
+func TestProbeWakeCancelBeforeTheWakeStarts(t *testing.T) {
+ r := &pwRuntime{observed: pwFlip()}
+ a, _ := pwService(t, r)
+ rt := a.cfg.Claims
+ prereq := rt.Prerequisites
+ rt.Prerequisites = func(ctx context.Context, route registry.Route, o hostclaim.Observed, p startPolicy) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ return prereq(ctx, route, o, p)
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ probeWakePublished = cancel
+ t.Cleanup(func() { probeWakePublished = nil })
+ _, err := a.ProbeWake(ctx, pwConfig())
+ if !errors.Is(err, context.Canceled) {
+ t.Fatalf("err = %v", err)
+ }
+ for _, call := range r.calls {
+ if call == "admit" || call == "prerequisites" {
+ t.Fatalf("the cancelled probe ran %v", r.calls)
+ }
+ }
+ if a.Owns(pwHost) {
+ t.Fatal("the cancelled probe left its entry")
+ }
+}