carlos / platform Public

zdd-5 final review: act on a wake's state only for its own token (F4, rollout half)

Plan takes a fresh token per plan (Manager.wakeTok, one counter) before it
registers anything, and snapshots the delivered config revision into
WakePlan.ConfigRev. OnGen clears a planned entry and closes an observation
only for the token they were made for, and opens an observation or
activates a config revision (the one the event carries) only for the
host's newest wake. A late event of an older wake no longer clears a newer
wake's registration, so a Pass keeps the failure record its check needs.
configTrack.wakeRev is gone.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 2 weeks ago 4dc989d4d1562a08d3d337dab8053b7c89d03544 parent 47df291

8 files changed, +277 −77

internal/rollout/config.go +0 −5
internal/rollout/manager.go +61 −24
internal/rollout/sim_test.go +8 −8
internal/rollout/status.go +1 −1
internal/rollout/status_test.go +5 −2
internal/rollout/wake_token_test.go +135 −0
internal/rollout/wakeplan.go +61 −33
internal/rollout/wakeplan_test.go +6 −4

diff --git a/internal/rollout/config.go b/internal/rollout/config.go

index b84048bba..cd3c14d32 100644

--- a/internal/rollout/config.go

+++ b/internal/rollout/config.go

@@ -12,11 +12,6 @@ import (

type configTrack struct {

delivered string

active string

- // wakeRev is the delivered revision when the wake now in flight was

- // planned. The wake's spawn reads the file after that, so the revision it

- // runs is at least this one: activating wakeRev can only under-claim,

- // which costs one more config attempt, never a missed one.

- wakeRev string

// sleeping: the instance had no process when delivered was written

// (spec §8.1). Its next wake starts with that file, so no attempt is

// pending and the ack goes out at once, marked Hibernating.

diff --git a/internal/rollout/manager.go b/internal/rollout/manager.go

index ca077a7d8..5b0ba5ee1 100644

--- a/internal/rollout/manager.go

+++ b/internal/rollout/manager.go

@@ -119,19 +119,27 @@ type Manager struct {

// once more for a nudge that arrived meanwhile.

nudging map[string]bool

// observing holds the hosts whose first wake is inside its 10 s

- // observation: opened by a healthy first wake, closed by whatever ends

- // it (OnGen). Status reports them as settling.

- observing map[string]bool

+ // observation, each with the token of the wake it observes: opened by

+ // that wake's healthy first-wake event, closed by an event carrying the

+ // same token (OnGen). Status reports them as settling.

+ observing map[string]uint64

// planned holds each host whose planned first wake is in flight, and the

- // build it boots: from its Plan to its outcome (OnGen), its next Plan, or

- // the activator letting the host go (a Pass that finds Owns false). prune

- // keeps those builds' failure records, because a wake's last selection

- // check (planCheck) reads only the book: a record pruned after the plan,

- // once a promotion moved every route off the build, would let the wake

- // boot a build that failed on this box. plannedSeq tells one plan's entry

- // from the next for that Pass, which asks Owns off every lock.

+ // build it boots: from its Plan to its outcome (an OnGen event carrying

+ // the plan's token), its next Plan, or the activator letting the host go

+ // (a Pass that finds Owns false). prune keeps those builds' failure

+ // records, because a wake's last selection check (planCheck) reads only

+ // the book: a record pruned after the plan, once a promotion moved every

+ // route off the build, would let the wake boot a build that failed on

+ // this box. plannedSeq tells one registration from the next for that

+ // Pass, which asks Owns off every lock.

planned map[string]plannedWake

plannedSeq uint64

+ // wakeTok is each host's newest wake token: the token of the last plan

+ // made for it (final review F4). Tokens come from one counter, so they

+ // are unique and only grow. An event carrying any other token is an older

+ // wake's, and acts on nothing that belongs to the newer one.

+ wakeTok map[string]uint64

+ tokenCount uint64

// goodMu serialises every serving- and good-build write with its

// acknowledgement check (served, publishServing, writeGood), so an older

@@ -149,8 +157,8 @@ func NewManager(ctx context.Context, d Deps) *Manager {

}

m := &Manager{d: d, ctx: ctx,

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]bool{},

- planned: map[string]plannedWake{}}

+ 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{}}

m.policies = &policyCache{load: d.Policy, ttl: 30 * time.Second, now: d.Now, m: map[appKey]policyEntry{}}

return m

}

@@ -266,6 +274,9 @@ func (m *Manager) forget(routes []registry.Route) {

delete(m.cfg, host)

delete(m.retried, host)

delete(m.observing, host)

+ // A later event of the host's last wake then matches no token: the

+ // host is gone, and there is nothing of its wake left to act on.

+ delete(m.wakeTok, host)

}

m.mu.Unlock()

if len(gone) > 0 {

@@ -545,46 +556,72 @@ func (m *Manager) failedUnder(rt registry.Route, build string) time.Time {

return time.Time{}

}

-// plannedWake is a planned first wake in flight (Manager.planned).

+// plannedWake is a planned first wake in flight (Manager.planned). token is

+// its plan's (WakePlan.Token): only an event carrying it ends the entry.

type plannedWake struct {

acct, app, build string

seq uint64

+ token uint64

+// newWakeToken starts a plan for host: a fresh token, recorded as the

+// host's newest. It is taken before the plan registers anything (Plan), so

+// the plan's registration and every event of its wake carry it.

+func (m *Manager) newWakeToken(host string) uint64 {

+ m.mu.Lock()

+ defer m.mu.Unlock()

+ m.tokenCount++

+ m.wakeTok[host] = m.tokenCount

+ return m.tokenCount

}

// notePlanned records what host's newest plan boots: an entry for a first

// wake, none for anything else. Either way it replaces the host's previous

-// plan, whose wake is over or is being planned again.

-func (m *Manager) notePlanned(rt registry.Route, plan activator.WakePlan) {

+// plan, whose wake is over or is being planned again. A plan that is no

+// longer the host's newest (tok) changes nothing: the newer plan owns the

+// entry.

+func (m *Manager) notePlanned(rt registry.Route, plan activator.WakePlan, tok uint64) {

m.mu.Lock()

defer m.mu.Unlock()

+ if m.wakeTok[rt.Host] != tok {

+ return

+ }

if !plan.FirstWake || plan.Boot.Version == "" || plan.Refuse != "" {

delete(m.planned, rt.Host)

return

}

m.plannedSeq++

- m.planned[rt.Host] = plannedWake{acct: release.RouteAccount(rt), app: rt.App, build: plan.Boot.Version, seq: m.plannedSeq}

+ m.planned[rt.Host] = plannedWake{acct: release.RouteAccount(rt), app: rt.App, build: plan.Boot.Version,

+ seq: m.plannedSeq, token: tok}

}

// 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) {

+// build while its plan (tok) 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, tok uint64) {

m.mu.Lock()

defer m.mu.Unlock()

+ if m.wakeTok[rt.Host] != tok {

+ return

+ }

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}

+ m.planned[rt.Host] = plannedWake{acct: release.RouteAccount(rt), app: rt.App, build: rt.Version,

+ seq: m.plannedSeq, token: tok}

}

// 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

-// end may let the record go. Owns is asked with no lock held (the lock order

+// abandoned at its exec) sends no event carrying its token and leaves its

+// entry behind, and only the entry's own end may let the record go. (A

+// replan's first plan gets no event either; the replan, a newer Plan,

+// replaces its entry.) Owns is asked with no lock held (the lock order

// on Manager), and an entry is dropped only if no newer plan replaced it

// meanwhile.

func (m *Manager) dropUnownedWakes() {

diff --git a/internal/rollout/sim_test.go b/internal/rollout/sim_test.go

index 3511a6b5e..1a54d18f1 100644

--- a/internal/rollout/sim_test.go

+++ b/internal/rollout/sim_test.go

@@ -802,7 +802,7 @@ func (s *sim) wakeExec() {

h.phase, h.gen = simAwake, s.newGen(good)

h.served[good.Version] = true

s.logf("wake %s: %s untried, booted the good build %s", h.name, plan.Boot.Version, good.Version)

- s.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: h.gen, Mode: plan.Mode, At: s.now})

+ s.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: h.gen, Mode: plan.Mode, At: s.now, Token: plan.Token, ConfigRev: plan.ConfigRev})

return

}

boot := plan.Boot

@@ -814,7 +814,7 @@ func (s *sim) wakeExec() {

h.phase, h.gen = simAwake, s.newGen(boot)

h.good, h.served[boot.Version] = boot.Version, true

s.logf("wake %s: booted %s", h.name, boot.Version)

- s.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: h.gen, Mode: plan.Mode, At: s.now})

+ s.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: h.gen, Mode: plan.Mode, At: s.now, Token: plan.Token, ConfigRev: plan.ConfigRev})

return

}

b := s.builds[boot.Version]

@@ -829,20 +829,20 @@ func (s *sim) wakeExec() {

case !fails:

h.phase, h.gen = simObserving, s.newGen(boot)

s.logf("wake %s: first wake of %s healthy, observing", h.name, boot.Version)

- s.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: h.gen, Mode: plan.Mode, FirstWake: true, At: s.now})

+ s.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: h.gen, Mode: plan.Mode, FirstWake: true, At: s.now, Token: plan.Token, ConfigRev: plan.ConfigRev})

case canFall:

s.failure(boot.Version, h.name, h.stamp, false)

h.phase, h.gen = simAwake, s.newGen(*plan.Fallback)

s.logf("wake %s: first wake of %s failed, fell back to %s", h.name, boot.Version, plan.Fallback.Version)

s.m.OnGen(activator.GenEvent{Kind: activator.GenFellBack, Route: rt, Gen: h.gen, Failed: boot, Mode: plan.Mode,

- FirstWake: true, Reason: "health check failed", At: s.now, FailedUnder: plan.FailedUnder})

+ FirstWake: true, Reason: "health check failed", At: s.now, FailedUnder: plan.FailedUnder, Token: plan.Token, ConfigRev: plan.ConfigRev})

default:

hold := plan.Mode == activator.ModeHold

s.failure(boot.Version, h.name, h.stamp, hold)

h.phase, h.gen = simAsleep, activator.Gen{}

s.logf("wake %s: first wake of %s failed, nothing fell back (mode %s)", h.name, boot.Version, plan.Mode)

s.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: rt, Failed: boot, Mode: plan.Mode,

- FirstWake: true, Reason: "health check failed", At: s.now, FailedUnder: plan.FailedUnder})

+ FirstWake: true, Reason: "health check failed", At: s.now, FailedUnder: plan.FailedUnder, Token: plan.Token, ConfigRev: plan.ConfigRev})

if h.good != "" && !hold {

violate("I1", "%s (good build %s) failed a first wake of %s in overlap mode and nothing fell back", h.name, h.good, boot.Version)

}

@@ -861,13 +861,13 @@ func (s *sim) observeEnd() {

h.phase, h.gen = simAsleep, activator.Gen{}

s.logf("wake %s: %s exited inside its observation", h.name, plan.Boot.Version)

s.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: rt, Failed: plan.Boot, Mode: plan.Mode, FirstWake: true,

- Reason: "exited within 10s of becoming healthy", At: s.now, FailedUnder: plan.FailedUnder})

+ Reason: "exited within 10s of becoming healthy", At: s.now, FailedUnder: plan.FailedUnder, Token: plan.Token, ConfigRev: plan.ConfigRev})

return

}

h.phase = simAwake

h.good, h.served[h.gen.Build] = h.gen.Build, true

s.logf("wake %s: %s settled", h.name, h.gen.Build)

- s.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: rt, Gen: h.gen, Mode: plan.Mode, FirstWake: true, At: s.now})

+ s.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: rt, Gen: h.gen, Mode: plan.Mode, FirstWake: true, At: s.now, Token: plan.Token, ConfigRev: plan.ConfigRev})

}

// sleep hibernates an instance with no attempt in Swap: at once, or through a

@@ -888,7 +888,7 @@ func (s *sim) sleep() {

}

s.logf("sleep %s (stopping phase %v, observation cancelled %v)", h.name, h.phase == simStopping, observing)

if observing {

- s.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.planRoute, Gen: g, Mode: h.plan.Mode, FirstWake: true, At: s.now})

+ s.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.planRoute, Gen: g, Mode: h.plan.Mode, FirstWake: true, At: s.now, Token: h.plan.Token, ConfigRev: h.plan.ConfigRev})

}

diff --git a/internal/rollout/status.go b/internal/rollout/status.go

index 5e938b5f9..775b8e0e6 100644

--- a/internal/rollout/status.go

+++ b/internal/rollout/status.go

@@ -23,7 +23,7 @@ func (m *Manager) Status() status.RolloutReport {

scheds = append(scheds, s)

}

observing := make(map[string]bool, len(m.observing))

- for h := range m.observing {

+ for h := range m.observing { // whichever wake's token: the host is settling

observing[h] = true

}

type cfgView struct {

diff --git a/internal/rollout/status_test.go b/internal/rollout/status_test.go

index 5a5f3c89b..42a9a5a9e 100644

--- a/internal/rollout/status_test.go

+++ b/internal/rollout/status_test.go

@@ -1,6 +1,7 @@

package rollout

import (

+ "context"

"sync/atomic"

"testing"

@@ -87,12 +88,14 @@ func TestACancelledObservationStopsSettlingAndRecordsNothing(t *testing.T) {

h.stage("v2")

h.install("v2", "a")

h.pass()

+ // The wake's plan: its events carry the plan's token (final review F4).

+ p := h.m.Plan(context.Background(), h.rows.get("a"))

g := activator.Gen{N: 9, Build: "v2", Digest: "sha-v2"}

- h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get("a"), Gen: g, FirstWake: true})

+ h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get("a"), Gen: g, FirstWake: true, Token: p.Token})

if got := h.m.Status().Hosts["a"]; got.Attempt != status.AttemptSettling {

t.Fatalf("under observation: a = %+v, want settling", got)

}

- h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("a"), Gen: g, FirstWake: true})

+ h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("a"), Gen: g, FirstWake: true, Token: p.Token})

if got := h.m.Status().Hosts["a"]; got.Attempt == status.AttemptSettling {

t.Fatalf("after the cancelled observation: a = %+v, still settling", got)

}

diff --git a/internal/rollout/wake_token_test.go b/internal/rollout/wake_token_test.go

new file mode 100644

index 000000000..4c241c864

--- /dev/null

+++ b/internal/rollout/wake_token_test.go

@@ -0,0 +1,135 @@

+package rollout

+import (

+ "context"

+ "testing"

+ "github.com/carlosframework/platform/internal/activator"

+ "github.com/carlosframework/platform/internal/appconfig"

+ "github.com/carlosframework/platform/internal/status"

+// Final review F4: every plan carries a token of its own, never zero, and

+// the delivered config revision at the moment it was planned.

+func TestEveryPlanCarriesItsOwnTokenAndConfigRevision(t *testing.T) {

+ ctx := context.Background()

+ h := newHarness(t, "a")

+ h.stage("v2")

+ h.registerThenInstall("v2", "a")

+ h.wakeHost("a")

+ h.m.ConfigDelivered("a", "sha256:c1")

+ p1 := h.m.Plan(ctx, h.rows.get("a"))

+ h.m.ConfigDelivered("a", "sha256:c2")

+ p2 := h.m.Plan(ctx, h.rows.get("a"))

+ if p1.Token == 0 || p2.Token == 0 || p1.Token == p2.Token {

+ t.Fatalf("tokens %d and %d: want two distinct, nonzero tokens", p1.Token, p2.Token)

+ }

+ if p1.ConfigRev != "sha256:c1" || p2.ConfigRev != "sha256:c2" {

+ t.Fatalf("config revisions %q and %q: want each plan's delivered revision", p1.ConfigRev, p2.ConfigRev)

+ }

+// Final review F4, the reviewer's sequence: an old first wake's observer

+// delivers its GenCancelled late, after the host's new first wake of B has

+// registered. A sibling then fails B, and root moves the box's routes on to

+// C. The late event belongs to the old wake: it must not clear the new

+// wake's registration, so the Pass keeps B's record and the new wake's

+// check refuses B. Keyed by host alone, the old event deleted the newer

+// entry, the Pass pruned the record, and the check let the failed B boot.

+func TestAStaleOutcomeLeavesANewerWakesRegistration(t *testing.T) {

+ ctx := context.Background()

+ sibling := FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "b", At: t0}

+ h := newHarness(t, "a")

+ h.stage("v2", "v3")

+ h.install("v2", "a")

+ h.wakeHost("a")

+ old := h.m.Plan(ctx, h.rows.get("a"))

+ g := activator.Gen{N: 7, Build: "v2", Digest: "sha-v2"}

+ h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get("a"), Gen: g, FirstWake: true,

+ Token: old.Token, ConfigRev: old.ConfigRev})

+ // The instance stopped inside its observation and woke again: a new first

+ // wake of v2, planned while the old observer's outcome is still on its way.

+ h.wakeHost("a")

+ cur := h.m.Plan(ctx, h.rows.get("a"))

+ if !cur.FirstWake || cur.Boot.Version != "v2" || cur.Check == nil {

+ t.Fatalf("plan %+v: want a new first wake onto v2 with its check", cur)

+ }

+ h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("a"), Gen: g, FirstWake: true,

+ Token: old.Token, ConfigRev: old.ConfigRev})

+ if _, err := h.book.Record(sibling); err != nil {

+ t.Fatal(err)

+ }

+ h.install("v3", "a") // no route on the box installs v2 now

+ h.pass()

+ if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {

+ t.Fatal("the Pass pruned v2's record: the old wake's late event cleared the new wake's registration")

+ }

+ if err := cur.Check("v2"); err == nil {

+ t.Fatal("the new wake's check let v2 boot after it failed on a sibling")

+ }

+ // The new wake's own outcome does let the record go.

+ h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("a"), FirstWake: true,

+ Token: cur.Token, ConfigRev: cur.ConfigRev})

+ h.pass()

+ if _, ok := h.book.Lookup("bab", "blog", "v2"); ok {

+ t.Fatal("the new wake reported its outcome, and the next Pass still kept v2's record")

+ }

+// Final review F4: an event of an older wake does nothing to a newer

+// wake's state. A late GenHealthy of the older wake neither opens an

+// observation nor activates a config revision: with the revision read from

+// the host (the newest plan's snapshot), it acknowledged c2 while the old

+// wake's process ran c1. A late GenCancelled of it does not close the newer

+// wake's observation. The newer wake's own events activate the revision

+// they carry.

+func TestAStaleOutcomeLeavesANewerWakesObservationAndConfig(t *testing.T) {

+ ctx := context.Background()

+ h := newHarness(t, "a")

+ h.stage("v2")

+ h.registerThenInstall("v2", "a")

+ h.wakeHost("a")

+ var acks []appconfig.ConfigAck

+ ack := h.m.DeferAck(func(a appconfig.ConfigAck) { acks = append(acks, a) })

+ h.m.ConfigDelivered("a", "sha256:c1")

+ old := h.m.Plan(ctx, h.rows.get("a"))

+ // The older wake's plan was replaced (a replan's first token never gets

+ // an event) after config sync delivered c2.

+ h.m.ConfigDelivered("a", "sha256:c2")

+ ack(appconfig.ConfigAck{Host: "a"})

+ cur := h.m.Plan(ctx, h.rows.get("a"))

+ g := activator.Gen{N: 7, Build: "v2", Digest: "sha-v2"}

+ h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get("a"), Gen: g, FirstWake: true,

+ Token: old.Token, ConfigRev: old.ConfigRev})

+ if len(acks) != 0 {

+ t.Fatalf("acks %+v: an older wake's late event acknowledged a revision", acks)

+ }

+ if got := h.m.Status().Hosts["a"]; got.Attempt == status.AttemptSettling || !got.ConfigPending {

+ t.Fatalf("a = %+v: an older wake's late event opened an observation or activated a config", got)

+ }

+ h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get("a"), Gen: g, FirstWake: true,

+ Token: cur.Token, ConfigRev: cur.ConfigRev})

+ if len(acks) != 1 {

+ t.Fatalf("acks %+v after the current wake's healthy event, want c2's", acks)

+ }

+ if got := h.m.Status().Hosts["a"]; got.Attempt != status.AttemptSettling || got.ConfigPending {

+ t.Fatalf("a = %+v: want settling with c2 active", got)

+ }

+ h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("a"), Gen: g, FirstWake: true,

+ Token: old.Token, ConfigRev: old.ConfigRev})

+ if got := h.m.Status().Hosts["a"]; got.Attempt != status.AttemptSettling {

+ t.Fatalf("a = %+v: an older wake's late GenCancelled closed the newer wake's observation", got)

+ }

+ h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("a"), Gen: g, FirstWake: true,

+ Token: cur.Token, ConfigRev: cur.ConfigRev})

+ if got := h.m.Status().Hosts["a"]; got.Attempt == status.AttemptSettling {

+ t.Fatalf("a = %+v: the wake's own GenCancelled left it settling", got)

+ }

diff --git a/internal/rollout/wakeplan.go b/internal/rollout/wakeplan.go

index aa44d1421..54cbbe283 100644

--- a/internal/rollout/wakeplan.go

+++ b/internal/rollout/wakeplan.go

@@ -85,9 +85,18 @@ func decideWake(in wakeInputs) (activator.WakePlan, error) {

// Plan is the activator's Planner (spec §8.2). It is called without

// activator.Service.mu held, before the wake's lease. It also snapshots the

-// instance's delivered config revision (configTrack.wakeRev), which the

-// wake's spawn reads after this moment, so activating that revision when

-// the wake is healthy (OnGen) can only under-claim.

+// instance's delivered config revision into the plan (WakePlan.ConfigRev).

+// The wake's spawn reads the config file after this moment, so activating

+// that revision when the wake is healthy (OnGen, from the revision its

+// event carries) can only under-claim, which costs one more config attempt,

+// never a missed one.

+//

+// Every plan gets a token of its own (WakePlan.Token, final review F4),

+// taken before anything is registered for it. The activator carries it

+// onto every event of the wake, and OnGen acts on a host's wake state

+// (Manager.planned, observing, the config activation) only for the token

+// that state belongs to, so a late event of an older wake never touches a

+// newer one's.

// A Refuse text reaches the member's "wake failed" event (the activator's

// wake error), so it names builds only, never an error's text.

@@ -110,7 +119,11 @@ func (m *Manager) Plan(ctx context.Context, rt registry.Route) (plan activator.W

if !Managed(rt) {

return activator.WakePlan{}

}

- defer func() { m.notePlanned(rt, plan) }()

+ tok := m.newWakeToken(rt.Host)

+ defer func() {

+ plan.Token = tok

+ m.notePlanned(rt, plan, tok)

+ }()

if m.attemptRunning(rt.Host) {

// The instance's attempt has not reported yet (spec §6.4, §8.1): a

// hold that went down, or a swap whose entry stopped under it, is

@@ -141,7 +154,7 @@ 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)

+ plan = m.planRow(ctx, rt, tok)

fresh, err := m.d.Routes.Get(rt.Host)

if err != nil || fresh == nil {

// Nothing confirms the row the plan was made against, so nothing

@@ -164,13 +177,14 @@ func (m *Manager) Plan(ctx context.Context, rt registry.Route) (plan activator.W

}

-// 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)

+// planRow is one plan against rt (Plan's body, after the cold install), for

+// the plan whose token is tok. 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, tok uint64) activator.WakePlan {

+ m.notePlanning(rt, tok)

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) {

@@ -178,11 +192,14 @@ func (m *Manager) planRow(ctx context.Context, rt registry.Route) activator.Wake

// staging fault is an error, and anything unverifiable is hold.

return m.stagedDemand(ctx, rt, ver, path)

}}

+ // The delivered revision now, before the wake's spawn reads the file.

+ // It travels with the plan and comes back on the wake's events, so no

+ // later plan for the host can change what this wake activates.

m.mu.Lock()

- t := m.track(rt.Host)

- t.wakeRev = t.delivered

+ rev := m.track(rt.Host).delivered

m.mu.Unlock()

plan, err := decideWake(in)

+ plan.ConfigRev = rev

if err != nil {

m.stagingRefused(rt, rt.Version, err)

}

@@ -241,21 +258,40 @@ func (m *Manager) attemptRunning(host string) bool {

// transition, held down when it was a hold build. It is never called with

// activator.Service.mu held (emitGen), which served and publishServing need:

// they take goodMu and call back into Current.

+//

+// The host's wake state is the event's only when the token matches (final

+// review F4): the planned entry and the observation each end only by an

+// event carrying their own token, and only the host's newest wake (the

+// token of its last plan) opens an observation or activates a config

+// revision, the one its event carries. A late event of an older wake (an

+// observer's delayed GenCancelled, say) leaves a newer wake's registration,

+// observation and config alone. What an event says about its build is

+// still acted on whoever's it is: a failure is recorded, and a serving or

+// good build is published only while its generation is the one the

+// activator serves (acknowledged). A token of zero matches nothing.

func (m *Manager) OnGen(ev activator.GenEvent) {

rt := ev.Route

m.mu.Lock()

+ newest := ev.Token != 0 && m.wakeTok[rt.Host] == ev.Token

// The wake has an outcome: its build is no longer only planned, so its

// failure record need not outlive the install for its sake.

- delete(m.planned, rt.Host)

+ if w, ok := m.planned[rt.Host]; ok && ev.Token != 0 && w.token == ev.Token {

+ delete(m.planned, rt.Host)

+ }

switch {

case ev.Kind == activator.GenHealthy && ev.FirstWake:

- m.observing[rt.Host] = true

+ if newest {

+ m.observing[rt.Host] = ev.Token

+ }

case ev.Kind != activator.GenHealthy:

// GenSettled, GenFailed, GenFellBack and GenCancelled (an

// intentional stop ended the observation with no verdict) all close

- // it. The switch below has no recording arm for GenCancelled, so a

- // cancelled observation records nothing (spec §6.4).

- delete(m.observing, rt.Host)

+ // the observation their wake opened. The switch below has no

+ // recording arm for GenCancelled, so a cancelled observation records

+ // nothing (spec §6.4).

+ if tok, ok := m.observing[rt.Host]; ok && ev.Token != 0 && tok == ev.Token {

+ delete(m.observing, rt.Host)

+ }

}

m.mu.Unlock()

defer m.changed() // after the event's own work, with no lock held

@@ -269,7 +305,9 @@ func (m *Manager) OnGen(ev activator.GenEvent) {

}

// A first wake is published when it settles: until then the

// activator hides it (Current), so it is not yet acknowledged.

- m.wakeConfigActive(rt.Host)

+ if newest {

+ m.configActivated(rt.Host, ev.ConfigRev)

+ }

case activator.GenSettled:

m.served(rt.Host, ev.Gen)

case activator.GenFellBack:

@@ -278,7 +316,9 @@ func (m *Manager) OnGen(ev activator.GenEvent) {

// failed build is held.

m.buildFailed(rt, ev)

m.publishServing(rt.Host, ev.Gen)

- m.wakeConfigActive(rt.Host)

+ if newest {

+ m.configActivated(rt.Host, ev.ConfigRev)

+ }

m.emit(rt, EvFallbackServed, "warn", textFallback(ev.Failed.Version, ev.Gen.Build))

case activator.GenFailed:

m.buildFailed(rt, ev)

@@ -288,18 +328,6 @@ func (m *Manager) OnGen(ev activator.GenEvent) {

}

-// wakeConfigActive activates the config revision the wake now serving

-// started with: the delivered revision Plan snapshotted (wakeRev).

-func (m *Manager) wakeConfigActive(host string) {

- m.mu.Lock()

- rev := ""

- if t := m.cfg[host]; t != nil {

- rev = t.wakeRev

- }

- m.mu.Unlock()

- m.configActivated(host, rev)

// buildFailed is a failed first wake's cohort-failure transition (spec

// §8.1, §8.2 step 4).

func (m *Manager) buildFailed(rt registry.Route, ev activator.GenEvent) {

diff --git a/internal/rollout/wakeplan_test.go b/internal/rollout/wakeplan_test.go

index d3ac6f23c..6c68657ce 100644

--- a/internal/rollout/wakeplan_test.go

+++ b/internal/rollout/wakeplan_test.go

@@ -269,8 +269,10 @@ func TestAWakeActivatesTheConfigItStartedWith(t *testing.T) {

t.Fatal("acknowledged before any generation started with the config")

}

rt := h.rows.get("a")

- h.m.Plan(context.Background(), rt)

- h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: activator.Gen{Build: "v1", Digest: "sha-v1"}})

+ p := h.m.Plan(context.Background(), rt)

+ // The activator carries the plan's token and revision onto the event.

+ h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: activator.Gen{Build: "v1", Digest: "sha-v1"},

+ Token: p.Token, ConfigRev: p.ConfigRev})

if acks != 1 {

t.Fatalf("acks = %d after the wake, want 1", acks)

}

@@ -613,8 +615,8 @@ func TestAPlannedWakeKeepsItsBuildsFailureRecordThroughAPrune(t *testing.T) {

}

t.Run("the wake's outcome", func(t *testing.T) {

- h, _ := setup(t)

- h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("a"), FirstWake: true})

+ h, p := setup(t)

+ h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("a"), FirstWake: true, Token: p.Token})

pruned(t, h, "the wake reported its outcome")

})

t.Run("a newer plan", func(t *testing.T) {