amadan

carlos / platform Public

zdd-5 final review: carry the wake's token on every GenEvent (F4, activator half)

WakePlan gains Token and ConfigRev, both the planner's: Token is unique
per plan it makes (a replan is a new plan), assigned before it registers
anything for the wake; ConfigRev is the delivered revision it
snapshotted for the wake. The activator never makes or changes a token
and does not itself know the revision a wake started with. GenEvent
gains the same two fields, and every planned-wake event goes through
emitPlanned, which copies them from the plan the wake ran: afterWake's
(healthy, fellback, failed, untried, a worker's settled) and the
first-wake observer's (settled, failed, cancelled). internal/rollout is
unchanged (it compiles; part A2 sets and matches the token).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 47df2919a1037de5203e04a3f9f0a58442a4a88c parent 9f15c77
2 files changed, +189 −8
  • internal/activator/wake_token_test.go +152 −0
  • internal/activator/wakeplan.go +37 −8
diff --git a/internal/activator/wake_token_test.go b/internal/activator/wake_token_test.go
new file mode 100644
index 000000000..aac2a7e45
--- /dev/null
+++ b/internal/activator/wake_token_test.go
@@ -0,0 +1,152 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// Final review F4 (activator half): every GenEvent of a planned wake carries
+// that wake's token and the config revision it started with, as the plan
+// handed them in (WakePlan.Token, WakePlan.ConfigRev), so the scheduler can
+// match an outcome to the wake that produced it and an old event can never
+// clear or activate a newer wake's state. Covered: afterWake's events
+// (healthy, fellback, failed, untried, a worker's settled) and the
+// observer's (settled, failed, cancelled), and a replanned wake, whose
+// events carry the replan's token.
+func TestEveryGenEventCarriesItsWakesToken(t *testing.T) {
+ const token, rev = uint64(0x5eed), "cfg-7"
+ for _, tc := range []struct {
+ name string
+ run func(t *testing.T) []GenEvent
+ }{
+ {"healthy, then settled by the observer", func(t *testing.T) []GenEvent {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
+ good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2")
+ a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: token, ConfigRev: rev})
+ mustServe(t, a)
+ evs.wait(t, GenSettled)
+ return evs.all()
+ }},
+ {"fell back", func(t *testing.T) []GenEvent {
+ good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2")
+ a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: token, ConfigRev: rev})
+ a.genHealth = failBuild("v2")
+ mustServe(t, a)
+ evs.wait(t, GenFellBack)
+ return evs.all()
+ }},
+ {"failed in hold mode", func(t *testing.T) []GenEvent {
+ next := stagedID(t, t.TempDir(), "v2")
+ a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeHold, FirstWake: true, Token: token, ConfigRev: rev})
+ a.genHealth = failBuild("v2")
+ _, _ = a.beginRequest(context.Background(), testRoute("one.example.com"))
+ evs.wait(t, GenFailed)
+ return evs.all()
+ }},
+ {"untried under the verdict floor", func(t *testing.T) []GenEvent {
+ shrinkVerdictFloor(t, time.Hour)
+ good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2")
+ a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: token, ConfigRev: rev})
+ mustServe(t, a)
+ evs.wait(t, GenHealthy)
+ return evs.all()
+ }},
+ {"failed by the observer", func(t *testing.T) []GenEvent {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second)
+ good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2")
+ a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: token, ConfigRev: rev})
+ mustServe(t, a)
+ evs.wait(t, GenHealthy)
+ b.procOf(b.started()[0].N).crash()
+ evs.wait(t, GenFailed)
+ return evs.all()
+ }},
+ {"cancelled by an intentional stop", func(t *testing.T) []GenEvent {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second)
+ good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2")
+ a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: token, ConfigRev: rev})
+ mustServe(t, a)
+ evs.wait(t, GenHealthy)
+ a.BounceFor("one.example.com", CauseConfig)
+ evs.wait(t, GenCancelled)
+ return evs.all()
+ }},
+ {"a worker's first wake, settled at its readiness", func(t *testing.T) []GenEvent {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
+ a, _, _, _, _ := stagedActivator(t)
+ a.gens = true
+ evs := &genLog{}
+ a.OnGen = evs.add
+ good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2")
+ a.Planner = func(context.Context, registry.Route) WakePlan {
+ return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: token, ConfigRev: rev}
+ }
+ if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
+ t.Fatal(err)
+ }
+ evs.wait(t, GenSettled)
+ return evs.all()
+ }},
+ {"replanned at the exec: the replan's token", func(t *testing.T) []GenEvent {
+ good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2")
+ a, _, evs := planActivator(t, WakePlan{})
+ var plans atomic.Int32
+ held := func(build string) error {
+ if build == "v2" {
+ return errors.New("v2 failed on a sibling")
+ }
+ return nil
+ }
+ a.Planner = func(context.Context, registry.Route) WakePlan {
+ if plans.Add(1) == 1 {
+ return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Check: held, Token: token - 1, ConfigRev: "cfg-6"}
+ }
+ return WakePlan{Boot: good, Mode: ModeOverlap, Check: held, Token: token, ConfigRev: rev}
+ }
+ mustServe(t, a)
+ evs.wait(t, GenHealthy)
+ return evs.all()
+ }},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ evs := tc.run(t)
+ if len(evs) == 0 {
+ t.Fatal("no events")
+ }
+ for _, ev := range evs {
+ if ev.Token != token || ev.ConfigRev != rev {
+ t.Fatalf("%s event carries token %#x, config %q; want the wake's %#x, %q (all: %+v)", ev.Kind, ev.Token, ev.ConfigRev, token, rev, evs)
+ }
+ }
+ })
+ }
+}
+
+func mustServe(t *testing.T, a *Service) {
+ t.Helper()
+ done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ done()
+}
+
+func failBuild(build string) func(context.Context, Gen) error {
+ return func(_ context.Context, g Gen) error {
+ if g.Build == build {
+ return errors.New("GET / answered 500")
+ }
+ return nil
+ }
+}
+
+func (l *genLog) all() []GenEvent {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ return append([]GenEvent(nil), l.evs...)
+}
diff --git a/internal/activator/wakeplan.go b/internal/activator/wakeplan.go
index 2b4f702ea..56e9198c1 100644
--- a/internal/activator/wakeplan.go
+++ b/internal/activator/wakeplan.go
@@ -43,6 +43,21 @@ type WakePlan struct {
// good build always passes. Nil: nothing to re-check. It is called
// without a.mu held.
Check func(build string) error
+ // Token identifies this planned wake (final review F4). The planner
+ // assigns it, unique per plan it makes (a replan, checkPlan's, is a new
+ // plan with a new token), before it registers anything for the wake.
+ // The activator never makes or changes one: it copies the token of the
+ // plan the wake ran, unchanged, into every GenEvent of that wake,
+ // afterWake's and the first-wake observer's, so the scheduler can act
+ // only on the wake it registered. Zero: the planner matches nothing by
+ // token.
+ Token uint64
+ // ConfigRev is the config revision the planner snapshotted for this
+ // wake: the delivered revision to activate once the wake serves. The
+ // activator does not know which revision a wake started with (it reads
+ // the instance's env file at the spawn), so it carries this one,
+ // unchanged, into every GenEvent of the wake, beside Token.
+ ConfigRev string
}
func (p WakePlan) planned() bool { return p.Boot.StagedPath != "" }
@@ -181,6 +196,11 @@ type GenEvent struct {
// FailedUnder is the plan's FailedUnder (failed, fellback): the
// failing pointer's PromotedAt, read before the wake ran.
FailedUnder time.Time
+ // Token and ConfigRev are those of the plan the wake ran
+ // (WakePlan.Token, WakePlan.ConfigRev; final review F4). Every event of
+ // one planned wake carries the same pair, on every kind.
+ Token uint64
+ ConfigRev string
}
// The first-wake budget (spec §8.2). Vars so tests can shrink them.
@@ -568,6 +588,15 @@ func (a *Service) outcomeOwedLocked(e *actEntry, w *wokenInstance) bool {
return a.OnGen != nil && w != nil && w.failed != nil && e.plan.planned()
}
+// emitPlanned is emitGen for an event of a wake that ran plan: it carries
+// the plan's token and config revision (final review F4), so every event of
+// one planned wake, on every path, names the same wake. Never called with
+// a.mu held (emitGen's rule).
+func (a *Service) emitPlanned(plan WakePlan, ev GenEvent) {
+ ev.Token, ev.ConfigRev = plan.Token, plan.ConfigRev
+ a.emitGen(ev)
+}
+
// emitGen hands ev to OnGen. It is never called with a.mu held: OnGen
// reaches the scheduler's served, publishServing and writeGood, which take
// its goodMu and call back into Current.
@@ -588,25 +617,25 @@ func (a *Service) afterWake(e *actEntry, w *wokenInstance, err error) {
switch {
case err != nil:
if w.failed != nil {
- a.emitGen(GenEvent{Kind: GenFailed, Route: e.route, Failed: *w.failed, Mode: plan.Mode, FirstWake: true, Reason: w.failReason, At: now, FailedUnder: plan.FailedUnder})
+ a.emitPlanned(plan, GenEvent{Kind: GenFailed, Route: e.route, Failed: *w.failed, Mode: plan.Mode, FirstWake: true, Reason: w.failReason, At: now, FailedUnder: plan.FailedUnder})
}
case w.fellBack:
- a.emitGen(GenEvent{Kind: GenFellBack, Route: e.route, Gen: w.planned, Failed: *w.failed, Mode: plan.Mode, FirstWake: true, Reason: w.failReason, At: now, FailedUnder: plan.FailedUnder})
+ a.emitPlanned(plan, GenEvent{Kind: GenFellBack, Route: e.route, Gen: w.planned, Failed: *w.failed, Mode: plan.Mode, FirstWake: true, Reason: w.failReason, At: now, FailedUnder: plan.FailedUnder})
case w.untried:
// B was never exec'd (too little time for a verdict) and the good
// build serves: reported exactly as a healthy wake of the good build
// that is not a first wake, so the scheduler records it as served
// and B stays a candidate for the next wake.
- a.emitGen(GenEvent{Kind: GenHealthy, Route: e.route, Gen: w.planned, Mode: plan.Mode, At: now})
+ a.emitPlanned(plan, GenEvent{Kind: GenHealthy, Route: e.route, Gen: w.planned, Mode: plan.Mode, At: now})
default:
- a.emitGen(GenEvent{Kind: GenHealthy, Route: e.route, Gen: w.planned, Mode: plan.Mode, FirstWake: plan.FirstWake, At: now})
+ a.emitPlanned(plan, GenEvent{Kind: GenHealthy, Route: e.route, Gen: w.planned, Mode: plan.Mode, FirstWake: plan.FirstWake, At: now})
switch {
case plan.FirstWake && e.worker:
// A socketless first wake's readiness window (plannedWorkerReady,
// spec §6.5) was its settle: it is settled now. A second window
// would fail a build that was already ready, where §8.6 calls for
// ordinary crash recovery.
- a.emitGen(GenEvent{Kind: GenSettled, Route: e.route, Gen: w.planned, Mode: plan.Mode, FirstWake: true, At: now})
+ a.emitPlanned(plan, GenEvent{Kind: GenSettled, Route: e.route, Gen: w.planned, Mode: plan.Mode, FirstWake: true, At: now})
case plan.FirstWake:
a.watchFirstWake(e, w.planned, plan)
}
@@ -641,7 +670,7 @@ func (a *Service) watchFirstWake(e *actEntry, g Gen, plan WakePlan) {
// Every observation GenHealthy opened is closed by an event; this
// one says "no verdict", so the scheduler stops showing the host
// as settling and records nothing (spec §6.4, §9.1).
- a.emitGen(GenEvent{Kind: GenCancelled, Route: rt, Gen: g, Mode: plan.Mode, FirstWake: true, At: time.Now()})
+ a.emitPlanned(plan, GenEvent{Kind: GenCancelled, Route: rt, Gen: g, Mode: plan.Mode, FirstWake: true, At: time.Now()})
}
// The observation was registered in the awake transition
// (armFirstWatchLocked), so a stop that began meanwhile has already
@@ -685,12 +714,12 @@ func (a *Service) watchFirstWake(e *actEntry, g Gen, plan WakePlan) {
// Ended first: the scheduler records a settled build as good
// only while Current shows it (rollout's acknowledged).
end()
- a.emitGen(GenEvent{Kind: GenSettled, Route: rt, Gen: g, Mode: plan.Mode, FirstWake: true, At: time.Now()})
+ a.emitPlanned(plan, GenEvent{Kind: GenSettled, Route: rt, Gen: g, Mode: plan.Mode, FirstWake: true, At: time.Now()})
case exited:
// Delivered first, so nothing sees the failed generation as
// current (and admits an attempt onto it) before the failure is
// recorded.
- a.emitGen(GenEvent{Kind: GenFailed, Route: rt, Failed: plan.Boot, Mode: plan.Mode, FirstWake: true,
+ a.emitPlanned(plan, GenEvent{Kind: GenFailed, Route: rt, Failed: plan.Boot, Mode: plan.Mode, FirstWake: true,
Reason: "exited within " + firstWakeSettle.String() + " of becoming healthy", At: time.Now(), FailedUnder: plan.FailedUnder})
end()
default: // an intentional stop cancelled it