amadan

carlos / platform Public

activator, host: overlapping reloads and the open's idle protection (Plan 3 final review, rulings 2 and 3)

2. RetireExcept snapshots each probe entry's immutable metadata (hold op,
   phase, seq) under a.mu and reads the registry from the snapshot only;
   the probe metadata is never cleared (managed marks the edge's
   takeover), and an entry whose seq moved since a pass's snapshot is
   kept by that pass. Overlapping reloads across the open neither panic
   nor retire the serving probe.
3. OpenProbeEntry marks opening (retirement protection); idle protection
   ends only at ConfirmProbeOpened, which the target's open calls once
   the release is proved. A failed release confirms nothing.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude ff6110f4727fa64eb9ca6016fbd7f0805a6cbe0b parent cdc3c40
8 files changed, +217 −61
  • internal/activator/activator.go +21 −12
  • internal/activator/probewake.go +80 −40
  • internal/activator/probewake_test.go +71 −7
  • internal/host/movesland_commit_test.go +25 −1
  • internal/host/movesland_test.go +9 −0
  • internal/host/moveslandcommit.go +6 −0
  • internal/host/movespass.go +4 −0
  • internal/provision/plan9_b8_census_test.go +1 −1
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 246963866..993842595 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -3376,7 +3376,7 @@ func (a *Service) sweepOnce() {
// 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 {
+ if e.probe != nil && !e.probe.managed && e.probe.phase < probeOpened {
hibernates = false
}
idleWindow := a.cfg.Idle
@@ -3721,24 +3721,33 @@ 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.
- seen, held, openedAt := 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] && 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
+ if e.probe != nil {
+ snap, covered := seen[e]
+ switch {
+ case covered && e.probe.seq != snap.seq:
+ // The probe changed since this pass's snapshot (an open, or a
+ // newer pass took it over): the registry read is older than
+ // the change and cannot judge it. Kept for this reload.
+ continue
+ case !e.probe.managed && keep[e.host] && e.probe.phase >= probeOpening:
+ // The edge manages the opened probe's ready row: from here it
+ // is an ordinary serving entry (Plan 3 Task 7). The metadata
+ // stays (ruling 2).
+ e.probe.managed = true
+ e.probe.seq++
+ continue
+ case !e.probe.managed && (!covered || held[e]):
+ continue
+ }
}
- if e.probe != nil && e.probe.opened && !openedAt[e] {
- // Opened after the snapshot: the registry read is older than the
- // open, so it cannot judge the entry. Kept for this reload.
+ if keep[e.host] {
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 a36690069..c5965f00b 100644
--- a/internal/activator/probewake.go
+++ b/internal/activator/probewake.go
@@ -107,24 +107,75 @@ 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
+ // phase is the probe's open (final review, ruling 3): held, opening
+ // once the target's open marks it (OpenProbeEntry, before it releases
+ // the hold), opened once the release is confirmed (ConfirmProbeOpened).
+ // Retirement protection holds from opening on; idle protection ends only
+ // at opened.
+ phase probePhase
+ // managed is set when the edge manages the opened probe's ready row: the
+ // entry is an ordinary serving one from then on. The probe metadata is
+ // kept, never cleared, so an older overlapping retirement pass still
+ // validates against it (ruling 2).
+ managed bool
+ // seq counts every change of phase or managed. A retirement pass keeps
+ // an entry whose seq moved since its snapshot: its registry read is
+ // older than the change.
+ seq uint64
}
-// 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.
+// probePhase is a probe entry's open state.
+type probePhase int
+
+const (
+ probeHeldPhase probePhase = iota
+ probeOpening
+ probeOpened
+)
+
+// probeSnap is the immutable probe metadata a retirement pass snapshots
+// under a.mu and reads outside it (ruling 2).
+type probeSnap struct {
+ holdOp string
+ phase probePhase
+ seq uint64
+}
+
+// ConfirmProbeOpened records that the target's open released host's hold
+// for op (final review, ruling 3): the entry, opening, is opened, and its
+// idle protection ends. false when host has no opening probe entry of op.
+func (a *Service) ConfirmProbeOpened(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.managed || e.probe.phase < probeOpening {
+ return false
+ }
+ if e.probe.phase != probeOpened {
+ e.probe.phase = probeOpened
+ e.probe.seq++
+ }
+ return true
+}
+
+// OpenProbeEntry marks host's awake probe entry for op opening, before the
+// target's open releases the row's hold (Plan 3 Task 7; final review,
+// ruling 3). RetireExcept keeps it while its row is held or, once released,
+// ready, until the edge manages it (the open race); it stays exempt from
+// idle stop until ConfirmProbeOpened. A retried open marks it again. 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 {
+ if e == nil || e.state != actAwake || e.probe == nil || e.probe.op != op || e.probe.admitted == nil || e.probe.managed {
return false
}
- e.probe.opened = true
+ if e.probe.phase < probeOpening {
+ e.probe.phase = probeOpening
+ e.probe.seq++
+ }
return true
}
@@ -195,43 +246,32 @@ func probeRefused(e *actEntry, err error) error {
return errors.Join(ErrProbeRefused, err)
}
-// 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, 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
-// reload. A row that cannot be read keeps its entry for this reload.
-//
-// openedAt is the open state each snapshotted entry had when the snapshot
-// was taken. An entry opened since then was opened after its registry
-// observation could have been made, so that observation is stale for it:
-// RetireExcept re-validates the open under its final lock and keeps the
-// entry for this reload (fix round 1, finding 2).
-func (a *Service) heldProbes(keep map[string]bool) (seen, held, openedAt map[*actEntry]bool) {
+// heldProbes snapshots, under a.mu, the immutable metadata of every probe
+// entry RetireExcept may retire (seen), then reads their registry rows
+// outside the lock from the snapshot alone, never dereferencing e.probe
+// there (final review, ruling 2). held is the entries it keeps beyond keep:
+// those whose row is still held by the probe's hold op, and opening or
+// opened ones whose row the open made ready, which the edge's next reload
+// manages (the open race). The edge manages only ready rows (edge.Reload),
+// so a held probe is never in keep. A row that cannot be read keeps its
+// entry for this reload. RetireExcept re-validates each entry's seq under
+// its final lock: one that moved since the snapshot is kept.
+func (a *Service) heldProbes(keep map[string]bool) (seen map[*actEntry]probeSnap, held map[*actEntry]bool) {
+ seen, held = map[*actEntry]probeSnap{}, 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
+ if e.probe != nil && !e.probe.managed && !keep[e.host] {
+ seen[e] = probeSnap{holdOp: e.probe.holdOp, phase: e.probe.phase, seq: e.probe.seq}
}
}
a.mu.Unlock()
- seen, held = map[*actEntry]bool{}, map[*actEntry]bool{}
- for _, e := range probes {
- seen[e] = true
- }
if retireBeforeRegistryRead != nil {
retireBeforeRegistryRead()
}
if a.Routes == nil {
- return seen, held, opened
+ return seen, held
}
- for _, e := range probes {
+ for e, snap := range seen {
rt, err := a.Routes.Get(e.host)
if err != nil {
log.Printf("activator: %s: the probe's row could not be read (%v); kept for this reload", e.host, err)
@@ -240,11 +280,11 @@ func (a *Service) heldProbes(keep map[string]bool) (seen, held, openedAt map[*ac
}
switch {
case rt == nil:
- case probeHeld(*rt, e.probe.holdOp):
+ case probeHeld(*rt, snap.holdOp):
held[e] = true
- case opened[e] && rt.Status == "ready" && rt.HoldOp == "":
+ case snap.phase >= probeOpening && rt.Status == "ready" && rt.HoldOp == "":
held[e] = true
}
}
- return seen, held, opened
+ return seen, held
}
diff --git a/internal/activator/probewake_test.go b/internal/activator/probewake_test.go
index cf809dc08..27feb4727 100644
--- a/internal/activator/probewake_test.go
+++ b/internal/activator/probewake_test.go
@@ -419,10 +419,10 @@ func TestOpenProbeEntryMarksTheLiveProbe(t *testing.T) {
t.Fatal("the probe's entry was not marked")
}
a.mu.Lock()
- opened := e.probe.opened
+ phase := e.probe.phase
a.mu.Unlock()
- if !opened {
- t.Fatal("not marked open")
+ if phase != probeOpening {
+ t.Fatal("not marked opening")
}
}
@@ -458,9 +458,9 @@ func TestRetireExceptKeepsTheOpenedProbeAcrossTheOpen(t *testing.T) {
// 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
+ taken := e.probe.managed
a.mu.Unlock()
- if probe != nil {
+ if !taken {
t.Fatal("a managed entry is still a probe")
}
// An entry held by the probe but never opened, whose row is ready, is
@@ -490,8 +490,8 @@ func TestIdleSweepSparesAHeldProbe(t *testing.T) {
if state != actAwake {
t.Fatal("the idle sweep stopped a held probe")
}
- if !a.OpenProbeEntry("one.example.com", pwOp) {
- t.Fatal("not marked")
+ if !a.OpenProbeEntry("one.example.com", pwOp) || !a.ConfirmProbeOpened("one.example.com", pwOp) {
+ t.Fatal("not opened")
}
a.sweepOnce()
waitFor(t, "the opened probe to idle out", func() bool { return !a.Owns("one.example.com") })
@@ -529,3 +529,67 @@ func TestRetireExceptRevalidatesTheOpenUnderTheFinalLock(t *testing.T) {
t.Fatal("the probe opened during the retirement was retired")
}
}
+
+// Final review, ruling 2: two reloads overlap. The older pass snapshots the
+// held probe, the open lands, and a newer pass, whose managed set includes
+// the host, takes the entry over as ordinary serving, all before the older
+// pass reads the registry. Neither pass panics, and the older one does not
+// retire the serving entry.
+func TestRetireExceptOverlappingReloadsAcrossTheOpen(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)
+ retireBeforeRegistryRead = func() {
+ retireBeforeRegistryRead = nil // the newer pass runs without the pause
+ if !a.OpenProbeEntry("one.example.com", pwOp) {
+ t.Error("not marked")
+ }
+ if ok, err := reg.ReleaseHold("one.example.com", pwOp); err != nil || !ok {
+ t.Errorf("release: %v %v", ok, err)
+ }
+ a.ConfirmProbeOpened("one.example.com", pwOp)
+ a.RetireExcept(map[string]bool{"one.example.com": true})
+ }
+ t.Cleanup(func() { retireBeforeRegistryRead = nil })
+ a.RetireExcept(map[string]bool{})
+ a.mu.Lock()
+ state := e.state
+ a.mu.Unlock()
+ if state != actAwake || !a.Owns("one.example.com") {
+ t.Fatal("the older reload retired the serving probe")
+ }
+}
+
+// Final review, ruling 3: the mark is "opening"; idle protection ends only
+// once the release is confirmed. A failed release leaves the entry
+// protected through an idle sweep; a retried open that succeeds ends it.
+func TestIdleProtectionHoldsUntilTheReleaseIsConfirmed(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()
+ if !a.OpenProbeEntry("one.example.com", pwOp) {
+ t.Fatal("not marked")
+ }
+ // The release failed: no confirmation.
+ a.sweepOnce()
+ a.mu.Lock()
+ state := e.state
+ a.mu.Unlock()
+ if state != actAwake {
+ t.Fatal("an opening probe whose release failed was idled")
+ }
+ // The retried open: marked again, released, confirmed.
+ if !a.OpenProbeEntry("one.example.com", pwOp) || !a.ConfirmProbeOpened("one.example.com", pwOp) {
+ t.Fatal("the retried open")
+ }
+ a.sweepOnce()
+ waitFor(t, "the opened probe to idle out", func() bool { return !a.Owns("one.example.com") })
+}
diff --git a/internal/host/movesland_commit_test.go b/internal/host/movesland_commit_test.go
index fc2c2d2e0..4f3729689 100644
--- a/internal/host/movesland_commit_test.go
+++ b/internal/host/movesland_commit_test.go
@@ -2,7 +2,9 @@ package host
import (
"context"
+ "errors"
"reflect"
+ "strings"
"testing"
"time"
@@ -71,7 +73,7 @@ func TestCommitThenOpenInOrder(t *testing.T) {
h, p := probedLanding(t)
h.land(t)
committedOpened(t, h, p)
- want := []string{"mark:committed/committed/landing", "release:committed/committed/landing"}
+ want := []string{"mark:committed/committed/landing", "release:committed/committed/landing", "confirm:committed/committed/ready"}
if !reflect.DeepEqual(h.openLog, want) {
t.Fatalf("the open: %v, want %v", h.openLog, want)
}
@@ -206,3 +208,25 @@ func TestCommittedOpenIsProved(t *testing.T) {
t.Fatalf("the open after the refusal: %+v %+v", m, h2.route)
}
}
+
+// Final review, ruling 3: the probe entry's idle protection ends only on a
+// proved release. A release that fails confirms nothing and stamps
+// nothing; the retried open that succeeds confirms it.
+func TestCommittedOpenConfirmsOnlyAProvedRelease(t *testing.T) {
+ h, _ := probedLanding(t)
+ h.releaseErr = errors.New("the registry is busy")
+ h.land(t)
+ for _, step := range h.openLog {
+ if strings.HasPrefix(step, "confirm:") {
+ t.Fatalf("a failed release was confirmed: %v", h.openLog)
+ }
+ }
+ if m := h.read(t, p1Host).Move; m.OpenedAt != nil {
+ t.Fatalf("a failed release was stamped: %+v", m)
+ }
+ h.releaseErr = nil
+ h.land(t)
+ if last := h.openLog[len(h.openLog)-1]; !strings.HasPrefix(last, "confirm:") || h.read(t, p1Host).Move.OpenedAt == nil {
+ t.Fatalf("the retried open: %v", h.openLog)
+ }
+}
diff --git a/internal/host/movesland_test.go b/internal/host/movesland_test.go
index f89dd3be6..7efa1d5fa 100644
--- a/internal/host/movesland_test.go
+++ b/internal/host/movesland_test.go
@@ -88,6 +88,8 @@ type landingHarness struct {
holdProbed bool
// releaseNoop makes ReleaseHeld match nothing and change nothing.
releaseNoop bool
+ // releaseErr makes ReleaseHeld fail.
+ releaseErr error
}
// fakeParkLeases refuses a second acquisition of a held location.
@@ -372,6 +374,9 @@ func (h *landingHarness) landingCfg(t *testing.T) moveCfg {
if h.releaseNoop {
return false, nil
}
+ if h.releaseErr != nil {
+ return false, h.releaseErr
+ }
if host != p1Host || !registry.IsHold(h.route.Status) || h.route.HoldOp != op {
return false, nil
}
@@ -381,6 +386,10 @@ func (h *landingHarness) landingCfg(t *testing.T) moveCfg {
if h.noRelease {
c.ReleaseHeld = nil
}
+ c.ConfirmProbeOpen = func(host, op string) bool {
+ h.openLog = append(h.openLog, "confirm:"+h.openState(t))
+ return host == p1Host && op == p1ID
+ }
c.MarkProbeOpen = func(host, op string) bool {
h.openLog = append(h.openLog, "mark:"+h.openState(t))
return host == p1Host && op == p1ID
diff --git a/internal/host/moveslandcommit.go b/internal/host/moveslandcommit.go
index 04f5bb65b..43a124116 100644
--- a/internal/host/moveslandcommit.go
+++ b/internal/host/moveslandcommit.go
@@ -222,6 +222,12 @@ func (l *landingRun) open(ctx context.Context) (stamped bool, err error) {
}
}
}
+ // The release is proved (the row is ready and unheld): the probe's
+ // entry is opened, ending its idle protection (final review, ruling
+ // 3). A failed release returned above and confirmed nothing.
+ if cfg.ConfirmProbeOpen != nil {
+ cfg.ConfirmProbeOpen(host, id)
+ }
if fresh.Move.OpenedAt == nil {
if _, err = provision.SetRecordMove(ctx, l.s, l.rec.Account, l.rec.App, host, provision.MoveOpened(l.g, cfg.Now())); err != nil {
err = fmt.Errorf("the record's opened_at: %w", err)
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 25543b042..e436542d8 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -572,6 +572,10 @@ type moveCfg struct {
// edge reload in between keeps it (Plan 3 Task 7). Nil in production
// with Probe.
MarkProbeOpen func(host, op string) bool
+ // ConfirmProbeOpen is the activator's ConfirmProbeOpened: called only
+ // once the open's release of the hold is proved, ending the probe entry's
+ // idle protection (final review, ruling 3). Nil in production with Probe.
+ ConfirmProbeOpen func(host, op string) bool
// landFault and parkFault are test seams: the landing's crash points
// and the park engine's (park.Engine.After). Nil in production.
landFault func(point string) error
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go
index b295de4d4..eb73640bf 100644
--- a/internal/provision/plan9_b8_census_test.go
+++ b/internal/provision/plan9_b8_census_test.go
@@ -504,7 +504,7 @@ var plan9B8CensusInventory = []censusEntry{
// Commits the landing through SetRecordMove (MoveCommit), its refusal the rollback mark.
{"internal/host/moveslandcommit.go", "landingRun.probed", "record-writer", "key:SetRecordMove", "6f961790d8fe169af68e2f70790c400d37c2388a4f62c26921f22b8b47c859de"},
// Stamps opened_at through SetRecordMove (MoveOpened) inside the hold lock, after the proved release.
- {"internal/host/moveslandcommit.go", "landingRun.open", "record-writer", "key:SetRecordMove", "a61665f2f91222b44a54f9550a9500b5b60ed415bf80485db8ae286a9059c7ab"},
+ {"internal/host/moveslandcommit.go", "landingRun.open", "record-writer", "key:SetRecordMove", "65e6908056c3055d5d0df2c27bb88943e31c08691a2173a090a9edb94c8c0b49"},
// Closes the rollback record through SetRecordMove; claim release and local journal parking use separate storage.
{"internal/host/movesrecover.go", "recoverRun.finalize", "record-writer", "key:SetRecordMove", "fff2e28c78db68bccb16da6ab70ca877b51243b0aa56b9cb2c513c6f5660f7f7"},
// Advances the source recovery phase through SetRecordMove and its sequenced writer.