amadan

carlos / platform Public

zdd-5 task 10: register a planned build before the planner's slow reads

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude caefa087ea06d90bd48a28956662fc378a51f3e5 parent 16d54ab
4 files changed, +199 −3
  • internal/rollout/fake_test.go +9 −0
  • internal/rollout/manager.go +16 −0
  • internal/rollout/wakeplan.go +44 −3
  • internal/rollout/wakeplan_test.go +130 −0
diff --git a/internal/rollout/fake_test.go b/internal/rollout/fake_test.go
index 3a96488b6..e0f9cf5df 100644
--- a/internal/rollout/fake_test.go
+++ b/internal/rollout/fake_test.go
@@ -311,6 +311,9 @@ type harness struct {
events *eventLog
// pointerHook, when set, runs at the start of every pointer read.
pointerHook func()
+ // demandHook, when set, runs at the start of every demand read, with no
+ // lock held: a test's window into the planner's slow store read.
+ demandHook func(ver string)
}
func newHarness(t *testing.T, hosts ...string) *harness {
@@ -385,6 +388,12 @@ func (h *harness) stage(vers ...string) {
// error, and noDigest is a store that could not compare, both of which must
// end as a hold attempt.
func (h *harness) demand(_ context.Context, _, _, ver, _ string) (string, string, error) {
+ h.mu.Lock()
+ hook := h.demandHook
+ h.mu.Unlock()
+ if hook != nil {
+ hook(ver)
+ }
h.mu.Lock()
defer h.mu.Unlock()
if h.mismatch[ver] {
diff --git a/internal/rollout/manager.go b/internal/rollout/manager.go
index 4f0b269be..ca077a7d8 100644
--- a/internal/rollout/manager.go
+++ b/internal/rollout/manager.go
@@ -565,6 +565,22 @@ func (m *Manager) notePlanned(rt registry.Route, plan activator.WakePlan) {
m.planned[rt.Host] = plannedWake{acct: release.RouteAccount(rt), app: rt.App, build: plan.Boot.Version, seq: m.plannedSeq}
}
+// notePlanning registers rt's installed build as host's in-flight planned
+// build while its plan is still being made (Plan, planRow), so a Pass during
+// the planner's slow reads keeps the build's failure record. A plan's end
+// (notePlanned) replaces or clears the entry, and a newer seq keeps a Pass
+// that snapshotted the old one (dropUnownedWakes) from dropping it.
+func (m *Manager) notePlanning(rt registry.Route) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ if rt.Version == "" {
+ delete(m.planned, rt.Host)
+ return
+ }
+ m.plannedSeq++
+ m.planned[rt.Host] = plannedWake{acct: release.RouteAccount(rt), app: rt.App, build: rt.Version, seq: m.plannedSeq}
+}
+
// dropUnownedWakes forgets the planned wakes of hosts the activator no
// longer owns: a wake that ended with no event (refused at its lease,
// abandoned at its exec) leaves its entry behind, and only the entry's own
diff --git a/internal/rollout/wakeplan.go b/internal/rollout/wakeplan.go
index 869444b3c..74b7417ff 100644
--- a/internal/rollout/wakeplan.go
+++ b/internal/rollout/wakeplan.go
@@ -94,7 +94,18 @@ func decideWake(in wakeInputs) (activator.WakePlan, error) {
//
// Every plan for a managed host, refused or not, replaces the host's entry
// in Manager.planned, so prune keeps the failure record of a first wake's
-// build until that wake is over.
+// build until that wake is over. The installed build is registered there
+// before the planner's slow reads (planRow), so a Pass that runs while they
+// do keeps a record they could no longer see.
+//
+// The row is read again after those reads. A Pass whose snapshot preceded
+// the registration could have pruned the build's record only if no route
+// installed the build when that Pass listed them, so this host's row had
+// already moved off it, and the read after the registration sees that. Such
+// a plan is made once more against the fresh row. A row that moves again
+// during the replan refuses the wake rather than chase the row: two installs
+// inside one plan are installs still landing, and the activator's next wake
+// plans against whatever settled.
func (m *Manager) Plan(ctx context.Context, rt registry.Route) (plan activator.WakePlan) {
if !Managed(rt) {
return activator.WakePlan{}
@@ -129,6 +140,37 @@ func (m *Manager) Plan(ctx context.Context, rt registry.Route) (plan activator.W
}
rt = *fresh
}
+ for again := false; ; again = true {
+ plan = m.planRow(ctx, rt)
+ fresh, err := m.d.Routes.Get(rt.Host)
+ if err != nil || fresh == nil {
+ // Nothing confirms the row the plan was made against, so nothing
+ // may boot from it: the same answer as an unreadable row after a
+ // cold install.
+ if err != nil {
+ m.d.Logf("rollout: %s: read the row after its plan: %v", rt.Host, err)
+ }
+ return activator.WakePlan{Refuse: "no build is installed for this instance yet"}
+ }
+ if fresh.Version == rt.Version {
+ return plan
+ }
+ if again {
+ m.d.Logf("rollout: %s: its installed build moved twice while its wake was planned (%s, now %s); refusing the wake", rt.Host, rt.Version, fresh.Version)
+ return activator.WakePlan{Refuse: "no build is installed for this instance yet"}
+ }
+ m.d.Logf("rollout: %s: its installed build moved from %s to %s while its wake was planned; planning again", rt.Host, rt.Version, fresh.Version)
+ rt = *fresh
+ }
+}
+
+// planRow is one plan against rt (Plan's body, after the cold install). It
+// registers rt's installed build as the host's in-flight planned build
+// before any slow read (the policy, the book's reading in decideWake, the
+// canonical demand, the failing pointer); Plan's deferred notePlanned
+// replaces or clears that entry with what the plan finally boots.
+func (m *Manager) planRow(ctx context.Context, rt registry.Route) activator.WakePlan {
+ m.notePlanning(rt)
acct, app := release.RouteAccount(rt), rt.App
in := wakeInputs{Route: rt, OptDir: m.d.OptDir, Policy: m.policies.get(ctx, acct, app), Book: m.d.Book,
Demand: func(ver, path string) (string, string, error) {
@@ -140,8 +182,7 @@ func (m *Manager) Plan(ctx context.Context, rt registry.Route) (plan activator.W
t := m.track(rt.Host)
t.wakeRev = t.delivered
m.mu.Unlock()
- var err error
- plan, err = decideWake(in)
+ plan, err := decideWake(in)
if err != nil {
m.stagingRefused(rt, rt.Version, err)
}
diff --git a/internal/rollout/wakeplan_test.go b/internal/rollout/wakeplan_test.go
index b64f89735..d3ac6f23c 100644
--- a/internal/rollout/wakeplan_test.go
+++ b/internal/rollout/wakeplan_test.go
@@ -632,3 +632,133 @@ func TestAPlannedWakeKeepsItsBuildsFailureRecordThroughAPrune(t *testing.T) {
pruned(t, h, "the activator no longer owns the host")
})
}
+
+// blockDemand makes the demand read for ver block (once per call to it)
+// until the returned release is called. blocked closes when the read starts.
+func (h *harness) blockDemand(ver string) (blocked <-chan struct{}, release func()) {
+ in, out := make(chan struct{}), make(chan struct{})
+ var n atomic.Int32
+ h.mu.Lock()
+ h.demandHook = func(v string) {
+ if v == ver && n.Add(1) == 1 {
+ close(in)
+ <-out
+ }
+ }
+ h.mu.Unlock()
+ var once atomic.Bool
+ return in, func() {
+ if once.CompareAndSwap(false, true) {
+ close(out)
+ }
+ }
+}
+
+// planAsync runs Plan for host on its own goroutine, as the activator's wake
+// does, and returns its plan on the channel.
+func (h *harness) planAsync(rt registry.Route) <-chan activator.WakePlan {
+ done := make(chan activator.WakePlan, 1)
+ go func() { done <- h.m.Plan(context.Background(), rt) }()
+ return done
+}
+
+func waitClosed(t *testing.T, what string, c <-chan struct{}) {
+ t.Helper()
+ select {
+ case <-c:
+ case <-time.After(5 * time.Second):
+ t.Fatalf("timed out waiting for %s", what)
+ }
+}
+
+// I1, the planner's own window (Task 10 round 3): Plan reads the book, then
+// makes its slow reads (the canonical demand, the failing pointer, the
+// policy) before it returns. A sibling's failure of B recorded while Plan is
+// in those reads, then a promotion that moves the box off B, then a Pass:
+// had Plan registered B only at its return, the Pass would prune B's record
+// (no route installs B, nothing tracks it) and the exec would boot the failed
+// B. Plan registers B before its reads, so the record stands, and it reads
+// the row again after them, so a wake whose row moved is planned against the
+// fresh row.
+func TestAPlanInItsSlowReadsKeepsItsBuildsFailureRecord(t *testing.T) {
+ sibling := FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "b", At: t0}
+ notBootable := func(t *testing.T, p activator.WakePlan) {
+ t.Helper()
+ if p.Boot.Version == "v2" && (p.Check == nil || p.Check("v2") == nil) {
+ t.Fatalf("plan %+v: the exec may boot v2, which failed on a sibling during the plan", p)
+ }
+ }
+
+ t.Run("the route moves during the reads", func(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2", "v3")
+ h.install("v2", "a")
+ h.wakeHost("a")
+ blocked, release := h.blockDemand("v2")
+ defer release()
+ done := h.planAsync(h.rows.get("a"))
+ waitClosed(t, "Plan to read v2's demand", blocked)
+ if _, err := h.book.Record(sibling); err != nil {
+ t.Fatal(err)
+ }
+ h.install("v3", "a") // the promotion: no route on the box installs v2 now
+ h.pass()
+ _, kept := h.book.Lookup("bab", "blog", "v2")
+ release()
+ p := <-done
+ notBootable(t, p)
+ if !kept {
+ t.Fatal("the Pass pruned v2's record while Plan was still planning v2")
+ }
+ if !p.FirstWake || p.Boot.Version != "v3" {
+ t.Fatalf("plan %+v: the row moved to v3 during the plan, want a first wake onto v3", p)
+ }
+ })
+
+ t.Run("the row was stale before the plan", func(t *testing.T) {
+ // The activator read the row (v2) before a promotion landed; the
+ // sibling's record was pruned before Plan began. Plan's registration
+ // comes too late for that prune, and the read of the row after the
+ // slow reads catches it.
+ h := newHarness(t, "a")
+ h.stage("v2", "v3")
+ h.install("v2", "a")
+ h.wakeHost("a")
+ stale := h.rows.get("a")
+ if _, err := h.book.Record(sibling); err != nil {
+ t.Fatal(err)
+ }
+ h.install("v3", "a")
+ h.pass()
+ if _, ok := h.book.Lookup("bab", "blog", "v2"); ok {
+ t.Fatal("setup: nothing installs or plans v2, so the Pass should have pruned it")
+ }
+ p := h.m.Plan(context.Background(), stale)
+ if p.Boot.Version != "v3" {
+ t.Fatalf("plan %+v from a stale row: want the fresh row's v3", p)
+ }
+ })
+
+ t.Run("the route moves twice", func(t *testing.T) {
+ // The row moves during the first plan and again during the replan:
+ // the plan refuses rather than chase the row.
+ h := newHarness(t, "a")
+ h.stage("v2", "v3", "v4")
+ h.install("v2", "a")
+ h.wakeHost("a")
+ blocked, release := h.blockDemand("v2")
+ defer release()
+ done := h.planAsync(h.rows.get("a"))
+ waitClosed(t, "Plan to read v2's demand", blocked)
+ h.install("v3", "a")
+ blocked3, release3 := h.blockDemand("v3")
+ defer release3()
+ release()
+ waitClosed(t, "the replan to read v3's demand", blocked3)
+ h.install("v4", "a")
+ release3()
+ if p := <-done; p.Refuse == "" || p.Boot.StagedPath != "" {
+ t.Fatalf("plan %+v: the row moved twice during the plan, want refused", p)
+ }
+ })
+}