amadan

carlos / platform Public

activator: the held probe's idle exemption and the open race (Plan 3 Task 7)

OpenProbeEntry marks the probe's live entry open before the target's open
releases the hold. The idle sweep spares a probe entry until it is
opened (a held row gets no traffic). RetireExcept keeps an opened probe
entry whose row the open made ready, so an edge reload that read the
routes before the release keeps it; once the edge manages the row the
entry is an ordinary serving one.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 580a677680132089524bcc7ec968c9efc541833d parent 70f9ea6
3 files changed, +143 −2
  • internal/activator/activator.go +11 −0
  • internal/activator/probewake.go +29 −2
  • internal/activator/probewake_test.go +103 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 5f2f780e7..8439c71cf 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -2653,6 +2653,12 @@ func (a *Service) sweepOnce() {
// An instance is never in this arm: unflagged instances don't enter
// the activator.
hibernates := !e.worker || e.route.Hibernate
+ // A held probe entry (ProbeWake) gets no traffic: the edge routes no
+ // held row. It is not idle until the target's open marks it open
+ // (OpenProbeEntry, Plan 3 Task 7).
+ if e.probe != nil && !e.probe.opened {
+ hibernates = false
+ }
idleWindow := a.cfg.Idle
if e.workerIdle > 0 {
idleWindow = e.workerIdle
@@ -2957,6 +2963,11 @@ func (a *Service) RetireExcept(keep map[string]bool) {
a.mu.Lock()
var waking []*actEntry
for _, e := range a.entries {
+ if keep[e.host] && e.probe != nil && e.probe.opened {
+ // The edge manages the opened probe's ready row: from here it is
+ // an ordinary serving entry (Plan 3 Task 7).
+ e.probe = nil
+ }
if keep[e.host] || held[e] || (e.probe != nil && !seen[e]) {
continue
}
diff --git a/internal/activator/probewake.go b/internal/activator/probewake.go
index a301deba2..80448dc70 100644
--- a/internal/activator/probewake.go
+++ b/internal/activator/probewake.go
@@ -107,6 +107,25 @@ type probeEntry struct {
// it, so a cancel is in force before the entry is published or its
// goroutine starts (fix round 1, finding 7).
parent context.Context
+ // opened is set by OpenProbeEntry before the target's open releases the
+ // hold (Plan 3 Task 7).
+ opened bool
+}
+
+// OpenProbeEntry marks host's awake probe entry for op open, before the
+// target's open releases the row's hold (Plan 3 Task 7). From then on the
+// entry idles as any other, and RetireExcept keeps it while its row is ready
+// until the edge manages it (the open race). false when host has no awake
+// probe entry of op: the next request then wakes the host under the commit.
+func (a *Service) OpenProbeEntry(host, op string) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ if e == nil || e.state != actAwake || e.probe == nil || e.probe.op != op || e.probe.admitted == nil {
+ return false
+ }
+ e.probe.opened = true
+ return true
}
// ProbeWake runs one probe attempt for cfg.Host and waits for it. On
@@ -178,7 +197,9 @@ func probeRefused(e *actEntry, err error) error {
// 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
+// row is still held by the probe's hold op, and opened ones (OpenProbeEntry)
+// whose row the open made ready, which the edge's next reload manages (Plan
+// 3 Task 7, the open race). 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
@@ -186,9 +207,11 @@ func probeRefused(e *actEntry, err error) error {
func (a *Service) heldProbes(keep map[string]bool) (seen, held map[*actEntry]bool) {
a.mu.Lock()
var probes []*actEntry
+ opened := map[*actEntry]bool{}
for _, e := range a.entries {
if e.probe != nil && !keep[e.host] {
probes = append(probes, e)
+ opened[e] = e.probe.opened
}
}
a.mu.Unlock()
@@ -206,7 +229,11 @@ func (a *Service) heldProbes(keep map[string]bool) (seen, held map[*actEntry]boo
held[e] = true
continue
}
- if rt != nil && probeHeld(*rt, e.probe.holdOp) {
+ switch {
+ case rt == nil:
+ case probeHeld(*rt, e.probe.holdOp):
+ held[e] = true
+ case opened[e] && rt.Status == "ready" && rt.HoldOp == "":
held[e] = true
}
}
diff --git a/internal/activator/probewake_test.go b/internal/activator/probewake_test.go
index 62822a543..d3192602a 100644
--- a/internal/activator/probewake_test.go
+++ b/internal/activator/probewake_test.go
@@ -388,3 +388,106 @@ func TestLiveProbeAnswersThePassingProbeEntry(t *testing.T) {
}
}
}
+
+// pwHeldEntry is an awake probe entry for one.example.com, held by pwOp.
+func pwHeldEntry(t *testing.T, a *Service) *actEntry {
+ t.Helper()
+ admitted := hostclaim.Observed{Host: "one.example.com", Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Session: 20, Stamp: 18, Generation: pwOp}}}
+ e := &actEntry{host: "one.example.com", route: testRoute("one.example.com"), state: actAwake, ready: make(chan struct{}), proc: &fakeProc{}, rep: &fakeRep{},
+ probe: &probeEntry{policy: startPolicy{Probe: true}, op: pwOp, holdOp: pwOp, admitted: &admitted}}
+ close(e.ready)
+ a.mu.Lock()
+ a.entries[e.host] = e
+ a.mu.Unlock()
+ return e
+}
+
+// Task 7: the target's open marks the live probe entry for its op open;
+// nothing else is marked.
+func TestOpenProbeEntryMarksTheLiveProbe(t *testing.T) {
+ a, _, _, _ := testActivator(t, time.Hour)
+ e := pwHeldEntry(t, a)
+ if a.OpenProbeEntry("one.example.com", "mv_other") || a.OpenProbeEntry("absent.example.com", pwOp) {
+ t.Fatal("another op's or an absent entry was marked")
+ }
+ if !a.OpenProbeEntry("one.example.com", pwOp) {
+ t.Fatal("the probe's entry was not marked")
+ }
+ a.mu.Lock()
+ opened := e.probe.opened
+ a.mu.Unlock()
+ if !opened {
+ t.Fatal("not marked open")
+ }
+}
+
+// Task 7, the open race (deterministic): an edge reload that read the routes
+// before the hold was released, and retires after it, keeps the probe entry
+// the open marked: its row is ready, and the next reload manages it. Once
+// the edge manages it, it is an ordinary serving entry.
+func TestRetireExceptKeepsTheOpenedProbeAcrossTheOpen(t *testing.T) {
+ a, _, _, reg := testActivator(t, time.Hour)
+ a.Routes = reg
+ if err := reg.Upsert(testRoute("one.example.com")); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.SetHold("one.example.com", "landing", pwOp); err != nil {
+ t.Fatal(err)
+ }
+ e := pwHeldEntry(t, a)
+ // The reload's managed set was read while the row was held.
+ managed := map[string]bool{}
+ if !a.OpenProbeEntry("one.example.com", pwOp) {
+ t.Fatal("not marked")
+ }
+ if ok, err := reg.ReleaseHold("one.example.com", pwOp); err != nil || !ok {
+ t.Fatalf("release: %v %v", ok, err)
+ }
+ a.RetireExcept(managed)
+ a.mu.Lock()
+ state := e.state
+ a.mu.Unlock()
+ if state != actAwake || !a.Owns("one.example.com") {
+ t.Fatal("the opened probe's entry was retired across the open")
+ }
+ // The next reload manages the ready row: the entry is ordinary from here.
+ a.RetireExcept(map[string]bool{"one.example.com": true})
+ a.mu.Lock()
+ probe := e.probe
+ a.mu.Unlock()
+ if probe != nil {
+ t.Fatal("a managed entry is still a probe")
+ }
+ // An entry held by the probe but never opened, whose row is ready, is
+ // not the probe's: it is retired as before.
+ a2, _, _, reg2 := testActivator(t, time.Hour)
+ a2.Routes = reg2
+ if err := reg2.Upsert(testRoute("one.example.com")); err != nil {
+ t.Fatal(err)
+ }
+ pwHeldEntry(t, a2)
+ a2.RetireExcept(map[string]bool{})
+ waitFor(t, "the unopened probe on a ready row to retire", func() bool { return !a2.Owns("one.example.com") })
+}
+
+// Task 7, the idle stop: a held probe entry gets no traffic, so the idle
+// sweep must not stop it before the open; once opened it idles as any entry.
+func TestIdleSweepSparesAHeldProbe(t *testing.T) {
+ a, _, clk, _ := testActivator(t, time.Minute)
+ e := pwHeldEntry(t, a)
+ a.mu.Lock()
+ e.lastActive = clk.now().Add(-time.Hour)
+ a.mu.Unlock()
+ a.sweepOnce()
+ a.mu.Lock()
+ state := e.state
+ a.mu.Unlock()
+ if state != actAwake {
+ t.Fatal("the idle sweep stopped a held probe")
+ }
+ if !a.OpenProbeEntry("one.example.com", pwOp) {
+ t.Fatal("not marked")
+ }
+ a.sweepOnce()
+ waitFor(t, "the opened probe to idle out", func() bool { return !a.Owns("one.example.com") })
+}