amadan
zdd-readiness: final review round 1, execution identities carry the manager's incarnation
Finding 1 (Astra, High): attempt and wake counters restart at zero in a new agent process while the markers and intents they named survive in the book, so the first wake after a restart could be w1 again and markLocked kept the old execution's Seq and At. Every execution identity now carries 8 random bytes drawn once per NewManager: a<inc>-<n> and w<inc>-<token>. wakeExecID becomes a Manager method; the a/w prefix Begin's wake rule reads is kept.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
9 files changed,
+189
−28
internal/rollout/bookseq_test.go+1 −1internal/rollout/failures.go+12 −7internal/rollout/finalreview1_test.go+128 −0internal/rollout/intent_test.go+2 −2internal/rollout/manager.go+36 −9internal/rollout/queue.go+4 −2internal/rollout/scheduler.go+1 −2internal/rollout/sim_test.go+1 −1internal/rollout/wakeplan.go+4 −4
diff --git a/internal/rollout/bookseq_test.go b/internal/rollout/bookseq_test.go| index c1f5d704f..2fddabc24 100644 |
| --- a/internal/rollout/bookseq_test.go |
| +++ b/internal/rollout/bookseq_test.go |
| @@ -295,7 +295,7 @@ func TestASettleDecidedBeforeAHoldRanKeepsItsMarker(t *testing.T) { |
| // an adoption of h (ErrIntentStands), so this ordering needs the intent |
| // gone. It is removed by hand: the marker rule must hold whatever lets a |
| // hold run between a plan and its settle. |
| - if err := h.book.End("h", wakeExecID(p.Token)); err != nil { |
| + if err := h.book.End("h", h.m.wakeExecID(p.Token)); err != nil { |
| t.Fatal(err) |
| } |
diff --git a/internal/rollout/failures.go b/internal/rollout/failures.go| index 970363636..35a3d5b0e 100644 |
| --- a/internal/rollout/failures.go |
| +++ b/internal/rollout/failures.go |
| @@ -126,9 +126,13 @@ type HoldRan struct { |
| Account string `json:"account"` |
| App string `json:"app"` |
| Build string `json:"build"` |
| - // Exec is the identity of the execution that set the marker: "a<n>" for |
| - // a scheduled attempt (dispatch.ExecID), "w<token>" for a planned wake. |
| - // A marker belongs to one execution, not to its build (plan review |
| + // Exec is the identity of the execution that set the marker: |
| + // "a<incarnation>-<n>" for a scheduled attempt (dispatch.ExecID), |
| + // "w<incarnation>-<token>" for a planned wake (Manager.wakeExecID). The |
| + // incarnation is the agent process's, so a run after a restart is never |
| + // taken for one before it (final review round 1, finding 1); a file an |
| + // earlier agent wrote holds "a<n>" or "w<token>", which no identity |
| + // handed out now equals. A marker belongs to one execution, not to its build (plan review |
| // finding 4): another run of the same build renews the marker's Seq, so |
| // a good write decided between the two runs clears nothing. Empty when |
| // the caller could not tell, which counts as a new execution every time. |
| @@ -171,9 +175,9 @@ type Intent struct { |
| // runs as a hold, plan review finding 2), so a reconciled hold intent |
| // holds the instance down. |
| Mode string `json:"mode"` |
| - // ID is the execution's identity (HoldRan.Exec): "a<n>" for an attempt |
| - // (dispatch.ExecID), "w<token>" for a planned wake (wakeExecID). A |
| - // reconciled hold intent's marker carries it. |
| + // ID is the execution's identity (HoldRan.Exec): "a<incarnation>-<n>" |
| + // for an attempt (dispatch.ExecID), "w<incarnation>-<token>" for a |
| + // planned wake (Manager.wakeExecID). A reconciled hold intent's marker carries it. |
| ID string `json:"id"` |
| // Start is when the execution started (dispatch.Started, a wake's plan |
| // time): a reconciled hold intent's marker carries it as its At (ruling |
| @@ -181,7 +185,8 @@ type Intent struct { |
| Start time.Time `json:"start"` |
| } |
| -// wakeIntent reports whether id is a planned wake's intent ("w<token>"). |
| +// wakeIntent reports whether id is a planned wake's intent: "w" leads a |
| +// wake's identity, in this agent's form and an earlier one's alike. |
| func wakeIntent(id string) bool { return strings.HasPrefix(id, "w") } |
| // ErrIntentStands is Begin's refusal while another execution's intent stands |
diff --git a/internal/rollout/finalreview1_test.go b/internal/rollout/finalreview1_test.go| new file mode 100644 |
| index 000000000..f28b27b00 |
| --- /dev/null |
| +++ b/internal/rollout/finalreview1_test.go |
| @@ -0,0 +1,128 @@ |
| +package rollout |
| + |
| +import ( |
| + "context" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/appconfig" |
| +) |
| + |
| +// restartOn is an agent restart under the harness: a fresh Manager over |
| +// book, its counters (attempt numbers, wake tokens) starting again from |
| +// zero, as a new process's do. The old manager's schedulers stop with ctx. |
| +func (h *harness) restartOn(book *FailureBook) { |
| + h.t.Helper() |
| + book.now = h.clock |
| + h.book = book |
| + 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", |
| + 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) {}, |
| + }) |
| +} |
| + |
| +// cancelHoldWake plans host's first wake (a hold build) and ends its |
| +// observation cancelled after the build ran, which leaves the host's |
| +// hold-ran marker and no failure record. |
| +func (h *harness) cancelHoldWake(host, build string) activator.WakePlan { |
| + h.t.Helper() |
| + h.wakeHost(host) |
| + p := h.m.Plan(context.Background(), h.rows.get(host)) |
| + if !p.FirstWake || p.Boot.Version != build || p.Mode != activator.ModeHold { |
| + h.t.Fatalf("plan %+v: want a hold first wake of %s", p, build) |
| + } |
| + g := activator.Gen{N: 7, Build: build, StagedPath: p.Boot.StagedPath, Digest: p.Boot.Digest} |
| + h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get(host), Gen: g, Mode: p.Mode, FirstWake: true, Token: p.Token}) |
| + h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get(host), Gen: g, Mode: p.Mode, FirstWake: true, |
| + Token: p.Token, Executed: true}) |
| + return p |
| +} |
| + |
| +// Final review round 1, finding 1 (Astra): execution identities are unique |
| +// across agent restarts. B runs on h and leaves a marker with no failure |
| +// record; the agent restarts (a new Manager, its counters at zero again, |
| +// over the same book on disk) and its first wake runs B again. The second |
| +// execution is a new one: the marker gets a fresh Seq and At, so the retry |
| +// grant the console signed for the first execution, and a rollback pointer |
| +// signed between the two executions, clear nothing of it. |
| +func TestAnExecutionAfterARestartIsANewExecution(t *testing.T) { |
| + dir := t.TempDir() |
| + h := newHarness(t, "s") |
| + book, err := OpenFailureBook(dir) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + h.restartOn(book) |
| + h.swap = appconfig.SwapHold |
| + h.stage("v2") |
| + h.install("v2", "s") |
| + h.sleepHost("s") |
| + h.pass() |
| + h.cancelHoldWake("s", "v2") |
| + first, ok := h.book.HoldRanOn("s") |
| + if !ok || first.Build != "v2" || !first.At.Equal(t0) { |
| + t.Fatalf("first marker %+v, %v: want v2's at %v", first, ok, t0) |
| + } |
| + staleGrant := h.book.RestrictionEpoch("bab", "blog", "v2", "s") |
| + |
| + // The agent restarts: the book reopens from disk, the manager's |
| + // counters start again. |
| + reopened, err := OpenFailureBook(dir) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + h.restartOn(reopened) |
| + h.advance(time.Minute) |
| + second := h.clock() |
| + h.pass() |
| + h.cancelHoldWake("s", "v2") |
| + got, ok := h.book.HoldRanOn("s") |
| + if !ok || got.Build != "v2" { |
| + t.Fatalf("marker after the second execution %+v, %v", got, ok) |
| + } |
| + if got.Exec == first.Exec { |
| + t.Fatalf("the second execution reused the first's identity %q", got.Exec) |
| + } |
| + if got.Seq <= first.Seq || !got.At.Equal(second) { |
| + t.Fatalf("marker %+v after the second execution: want a fresh Seq past %d and At %v", got, first.Seq, second) |
| + } |
| + |
| + // The first execution's grant and a rollback signed between the two runs |
| + // clear nothing of the second's marker. |
| + if rec, mark, _, _ := h.book.RetryHostAt("bab", "blog", "v2", "s", staleGrant); rec || mark { |
| + t.Fatal("a retry grant signed for the first execution cleared the second's marker") |
| + } |
| + if ok, _ := h.book.ClearHoldRanOnPointer("s", "bab", "blog", t0.Add(30*time.Second)); ok { |
| + t.Fatal("a rollback signed between the two executions cleared the second's marker") |
| + } |
| + if _, ok := h.book.HoldRanOn("s"); !ok { |
| + t.Fatal("the second execution's marker is gone") |
| + } |
| +} |
| + |
| +// Finding 1's identities: an attempt's and a wake's carry the manager's |
| +// incarnation, so two managers (two agent processes) never hand out the same |
| +// one, and the a/w prefix Begin's wake rule reads is kept. |
| +func TestExecutionIdentitiesCarryTheManagersIncarnation(t *testing.T) { |
| + m1 := NewManager(context.Background(), Deps{}) |
| + m2 := NewManager(context.Background(), Deps{}) |
| + a1, a2 := m1.nextExecID(), m2.nextExecID() |
| + if a1 == a2 || !strings.HasPrefix(a1, "a") || !strings.HasPrefix(a2, "a") || wakeIntent(a1) { |
| + t.Fatalf("attempt identities %q and %q: want distinct, both attempts", a1, a2) |
| + } |
| + w1, w2 := m1.wakeExecID(1), m2.wakeExecID(1) |
| + if w1 == w2 || !wakeIntent(w1) || !wakeIntent(w2) { |
| + t.Fatalf("wake identities %q and %q: want distinct, both wakes", w1, w2) |
| + } |
| + if m1.wakeExecID(0) != "" { |
| + t.Fatal("token zero has an identity") |
| + } |
| + if again := m1.nextExecID(); again == a1 { |
| + t.Fatalf("the same manager handed out %q twice", a1) |
| + } |
| +} |
diff --git a/internal/rollout/intent_test.go b/internal/rollout/intent_test.go| index 45b0d3927..5af4d9cdd 100644 |
| --- a/internal/rollout/intent_test.go |
| +++ b/internal/rollout/intent_test.go |
| @@ -452,7 +452,7 @@ func TestAPlannedWakesIntentEndsWithItsOutcome(t *testing.T) { |
| t.Fatalf("plan %+v: want a first wake of v2", p) |
| } |
| in, ok := fileOf(t, dir).Intents["a"] |
| - if !ok || in.ID != wakeExecID(p.Token) || in.Build != "v2" || in.Mode != string(p.Mode) || !in.Start.Equal(t0) { |
| + if !ok || in.ID != h.m.wakeExecID(p.Token) || in.Build != "v2" || in.Mode != string(p.Mode) || !in.Start.Equal(t0) { |
| t.Fatalf("intent on disk %+v, %v: want the plan's", in, ok) |
| } |
| return h, dir, p |
| @@ -491,7 +491,7 @@ func TestAPlannedWakesIntentEndsWithItsOutcome(t *testing.T) { |
| h, dir, old := setup(t) |
| cur := h.m.Plan(ctx, h.rows.get("a")) |
| h.m.OnGen(ev(old, h, activator.GenCancelled, true)) |
| - if in := fileOf(t, dir).Intents["a"]; in.ID != wakeExecID(cur.Token) { |
| + if in := fileOf(t, dir).Intents["a"]; in.ID != h.m.wakeExecID(cur.Token) { |
| t.Fatalf("intent %+v, want the newer wake's", in) |
| } |
| }) |
diff --git a/internal/rollout/manager.go b/internal/rollout/manager.go| index 9cc32eed3..ebd926869 100644 |
| --- a/internal/rollout/manager.go |
| +++ b/internal/rollout/manager.go |
| @@ -2,6 +2,8 @@ package rollout |
| import ( |
| "context" |
| + "crypto/rand" |
| + "encoding/hex" |
| "errors" |
| "log" |
| "os" |
| @@ -156,9 +158,19 @@ type Manager struct { |
| // start. An entry goes with its wake's last event, with the host |
| // (forget), or when the host has wakeStartsKept newer ones. |
| wakeAt map[string][]wakeStart |
| - // execCount numbers the scheduled attempts' executions (nextExec, |
| + // execCount numbers the scheduled attempts' executions (nextExecID, |
| // dispatch.ExecID). |
| execCount uint64 |
| + // inc is this manager's incarnation: 8 random bytes, hex, drawn once per |
| + // process start (NewManager). Every execution identity carries it |
| + // (nextExecID, wakeExecID), because the counters above start again at |
| + // zero in a new process while the markers and intents they named |
| + // survive in the book: without it the first wake after a restart could |
| + // be "w1" again, and markLocked would take the new execution for the |
| + // old one and keep its Seq and At, so a grant or a rollback that matched |
| + // the old execution would clear the new one (final review round 1, |
| + // finding 1, Astra). |
| + inc string |
| // ackSeq numbers the config acks the tracks hand back, under mu, in the |
| // critical section of the transition that releases each one: the |
| @@ -197,7 +209,7 @@ func NewManager(ctx context.Context, d Deps) *Manager { |
| if d.Logf == nil { |
| d.Logf = log.Printf |
| } |
| - m := &Manager{d: d, ctx: ctx, |
| + m := &Manager{d: d, ctx: ctx, inc: newIncarnation(), |
| apps: map[appKey]*scheduler{}, hostApp: map[string]appKey{}, cfg: map[string]*configTrack{}, |
| retried: map[string]string{}, pointerRead: map[appKey]time.Time{}, nudging: map[string]bool{}, observing: map[string]uint64{}, |
| planned: map[string]plannedWake{}, wakeTok: map[string]uint64{}, wakeAt: map[string][]wakeStart{}} |
| @@ -887,21 +899,36 @@ func (m *Manager) newWakeToken(host string) (uint64, time.Time) { |
| return m.tokenCount, now |
| } |
| -// nextExec numbers a scheduled attempt's execution (dispatch.ExecID). |
| -func (m *Manager) nextExec() uint64 { |
| +// newIncarnation draws a manager's incarnation (Manager.inc). crypto/rand |
| +// does not fail on the platforms the agent runs on; if it ever did, the |
| +// clock's nanoseconds still tell one process start from the next. |
| +func newIncarnation() string { |
| + var b [8]byte |
| + if _, err := rand.Read(b[:]); err != nil { |
| + return strconv.FormatInt(time.Now().UnixNano(), 16) |
| + } |
| + return hex.EncodeToString(b[:]) |
| +} |
| + |
| +// nextExecID is a scheduled attempt's execution identity (dispatch.ExecID): |
| +// "a<incarnation>-<n>", n from this manager's counter. The "a" is what |
| +// Begin's wake rule reads (wakeIntent). |
| +func (m *Manager) nextExecID() string { |
| m.mu.Lock() |
| defer m.mu.Unlock() |
| m.execCount++ |
| - return m.execCount |
| + return "a" + m.inc + "-" + strconv.FormatUint(m.execCount, 10) |
| } |
| -// wakeExecID is a planned wake's execution identity (HoldRan.Exec): "w" |
| -// and its token. Zero, an event of no known plan, has none. |
| -func wakeExecID(tok uint64) string { |
| +// wakeExecID is a planned wake's execution identity (HoldRan.Exec): |
| +// "w<incarnation>-<token>". Zero, an event of no known plan, has none. The |
| +// incarnation is fixed for the manager's life, so every event of one wake |
| +// names the identity its plan's intent was written with. |
| +func (m *Manager) wakeExecID(tok uint64) string { |
| if tok == 0 { |
| return "" |
| } |
| - return "w" + strconv.FormatUint(tok, 10) |
| + return "w" + m.inc + "-" + strconv.FormatUint(tok, 10) |
| } |
| // wakeStart is a planned wake's start, by its token (Manager.wakeAt): its |
diff --git a/internal/rollout/queue.go b/internal/rollout/queue.go| index 5233c6c03..63a6caae8 100644 |
| --- a/internal/rollout/queue.go |
| +++ b/internal/rollout/queue.go |
| @@ -57,8 +57,10 @@ type dispatch struct { |
| // before it (Astra R3 N6), and its failure is stale after a retry since |
| // (ruling 191). Zero until then. |
| BookSeq uint64 |
| - // ExecID is this execution's identity, "a<n>" from a Manager-wide counter |
| - // (Manager.nextExec), set with Started. A hold-ran marker it leaves |
| + // ExecID is this execution's identity, "a<incarnation>-<n>" from a |
| + // Manager-wide counter (Manager.nextExecID), set with Started. The |
| + // incarnation keeps it unique across agent restarts (final review round |
| + // 1, finding 1). A hold-ran marker it leaves |
| // carries it (HoldRan.Exec). SchedRev cannot serve: an abandoned |
| // attempt is derived again with the same revision. |
| ExecID string |
diff --git a/internal/rollout/scheduler.go b/internal/rollout/scheduler.go| index 21ed9dccb..409650d3c 100644 |
| --- a/internal/rollout/scheduler.go |
| +++ b/internal/rollout/scheduler.go |
| @@ -3,7 +3,6 @@ package rollout |
| import ( |
| "context" |
| "errors" |
| - "strconv" |
| "sync" |
| "time" |
| @@ -178,7 +177,7 @@ func (s *scheduler) execute(ctx context.Context, d dispatch) { |
| // this execution for a marker it leaves (plan review finding 4). |
| d.Started = m.d.Now() |
| d.BookSeq = m.d.Book.Snapshot() |
| - d.ExecID = "a" + strconv.FormatUint(m.nextExec(), 10) |
| + d.ExecID = m.nextExecID() |
| if d.Class == classAdopt { |
| // Write-ahead (Task 5, Astra R1 #5): the adoption's intent is on disk |
| // before the activator stops or starts anything, so a failure whose |
diff --git a/internal/rollout/sim_test.go b/internal/rollout/sim_test.go| index 038f265ea..8c50c1490 100644 |
| --- a/internal/rollout/sim_test.go |
| +++ b/internal/rollout/sim_test.go |
| @@ -1400,7 +1400,7 @@ func (s *sim) checkIntents() { |
| s.book.mu.Unlock() |
| for _, host := range sortedKeys(intents) { |
| in, h := intents[host], s.byName[host] |
| - inFlight := h != nil && (h.phase == simWaking || h.phase == simObserving) && h.plan.FirstWake && in.ID == wakeExecID(h.plan.Token) |
| + inFlight := h != nil && (h.phase == simWaking || h.phase == simObserving) && h.plan.FirstWake && in.ID == s.m.wakeExecID(h.plan.Token) |
| if !inFlight { |
| violate("I6", "%s's intent %+v outlived its execution", host, in) |
| } |
diff --git a/internal/rollout/wakeplan.go b/internal/rollout/wakeplan.go| index b516ec3f1..bf4544ea1 100644 |
| --- a/internal/rollout/wakeplan.go |
| +++ b/internal/rollout/wakeplan.go |
| @@ -308,7 +308,7 @@ func (m *Manager) beginWake(rt registry.Route, plan activator.WakePlan, tok uint |
| return plan |
| } |
| in := Intent{Account: release.RouteAccount(rt), App: rt.App, Build: plan.Boot.Version, Mode: string(plan.Mode), |
| - ID: wakeExecID(tok), Start: planAt} |
| + ID: m.wakeExecID(tok), Start: planAt} |
| err := m.d.Book.Begin(rt.Host, in) |
| if err == nil { |
| // The intent is on disk: Close now waits for the terminal event |
| @@ -525,7 +525,7 @@ func (m *Manager) OnGen(ev activator.GenEvent) { |
| if started.at.IsZero() { |
| m.d.Logf("rollout: %s: the start of the wake that ran %s is not known; its hold-ran marker is stamped now", rt.Host, ev.Gen.Build) |
| } |
| - mark := HoldRan{Account: release.RouteAccount(rt), App: rt.App, Build: ev.Gen.Build, Exec: wakeExecID(ev.Token), At: started.at} |
| + mark := HoldRan{Account: release.RouteAccount(rt), App: rt.App, Build: ev.Gen.Build, Exec: m.wakeExecID(ev.Token), At: started.at} |
| if err := m.d.Book.MarkHoldRan(rt.Host, mark); err != nil { |
| m.d.Logf("rollout: %s: write hold-ran marker for %s: %v (held in memory)", rt.Host, mark.Build, err) |
| } |
| @@ -542,7 +542,7 @@ func (m *Manager) OnGen(ev activator.GenEvent) { |
| // carries its own token, so it never ends a newer wake's intent. Not |
| // under activator.Service.mu (emitGen's rule). |
| if ended { |
| - if err := m.d.Book.End(rt.Host, wakeExecID(ev.Token)); err != nil { |
| + if err := m.d.Book.End(rt.Host, m.wakeExecID(ev.Token)); err != nil { |
| m.d.Logf("rollout: %s: end the intent of its wake: %v (it stays on disk until a save lands)", rt.Host, err) |
| } |
| // The wake's count in Close's join goes with the End (beginWake |
| @@ -590,7 +590,7 @@ func (m *Manager) buildFailed(rt registry.Route, ev activator.GenEvent, started |
| PromotedAt: ev.FailedUnder, At: m.d.Now(), |
| // The wake's start for a stale failure's marker: Record sets one in |
| // place of a hold-down when the failure is pre-retry (ruling 191). |
| - Seq: started.snap, HoldRanAt: started.at, HoldRanExec: wakeExecID(ev.Token)} |
| + Seq: started.snap, HoldRanAt: started.at, HoldRanExec: m.wakeExecID(ev.Token)} |
| if rec.PromotedAt.IsZero() { |
| // No pointer named the build when the wake was planned: the |
| // failure's own time, so only a pointer signed after it clears the |