amadan

carlos / platform Public

zdd-5 task 10: the wake planner and planned-wake outcomes

Plan picks the build a wake boots (decideWake), refuses while the host's
attempt runs or its hold build went down, installs a cold managed host
first, and carries FailedUnder and a selection gate (planCheck). OnGen
publishes, activates the wake's config and runs the cohort-failure
transition, the hold before the serving build. A planned wake reads the
canonical demand as an attempt does (stagedDemand, shared with resolve),
and nudges for one host collapse.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude ad3cb1fe3e229721e5446bdc706e7ba78a542925 parent b09bccf
3 files changed, +841 −14
  • internal/rollout/manager.go +54 −14
  • internal/rollout/wakeplan.go +257 −0
  • internal/rollout/wakeplan_test.go +530 −0
diff --git a/internal/rollout/manager.go b/internal/rollout/manager.go
index 4a349ed73..43bef4ab6 100644
--- a/internal/rollout/manager.go
+++ b/internal/rollout/manager.go
@@ -106,6 +106,9 @@ type Manager struct {
cfg map[string]*configTrack
retried map[string]string // host -> build a retry cleared: its next attempt's cause is retry
pointerRead map[appKey]time.Time
+ // nudging holds the hosts a Nudge is working on; true asks it to look
+ // once more for a nudge that arrived meanwhile.
+ nudging map[string]bool
// goodMu serialises every serving- and good-build write with its
// acknowledgement check (served, publishServing, writeGood), so an older
@@ -123,7 +126,7 @@ 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{}}
+ retried: map[string]string{}, pointerRead: map[appKey]time.Time{}, nudging: map[string]bool{}}
m.policies = &policyCache{load: d.Policy, ttl: 30 * time.Second, now: d.Now, m: map[appKey]policyEntry{}}
return m
}
@@ -176,13 +179,36 @@ func (m *Manager) Pass(routes []registry.Route) {
}
// Nudge re-reads one host and wakes its scheduler. The activator sweep's
-// drift arm calls it (spec §10).
+// drift arm calls it (spec §10, OnDrift) on a goroutine of its own, for every
+// drifted entry on every sweep, with no dedupe of its own. So nudges for one
+// host collapse here: while one is reading and observing, the others only
+// ask it for one more look and return, and a slow read (the row, the
+// policy) never collects a goroutine per sweep. A nudge for the build the
+// host already wants changes nothing (wantBuild), so a queued or running
+// attempt is neither queued again nor superseded.
func (m *Manager) Nudge(host string) {
- rt, err := m.d.Routes.Get(host)
- if err != nil || rt == nil {
+ m.mu.Lock()
+ if _, busy := m.nudging[host]; busy {
+ m.nudging[host] = true
+ m.mu.Unlock()
return
}
- m.observe([]registry.Route{*rt})
+ m.nudging[host] = false
+ m.mu.Unlock()
+ for {
+ if rt, err := m.d.Routes.Get(host); err == nil && rt != nil {
+ m.observe([]registry.Route{*rt})
+ }
+ m.mu.Lock()
+ again := m.nudging[host]
+ if !again {
+ delete(m.nudging, host)
+ m.mu.Unlock()
+ return
+ }
+ m.nudging[host] = false
+ m.mu.Unlock()
+ }
}
// Adopted replaces BounceAdopted as the adopt pass's restart callback when
@@ -588,32 +614,46 @@ func (m *Manager) resolve(ctx context.Context, d dispatch, rt registry.Route) (a
a.Mode = RecoveryMode(policy, rt.GoodDemand)
return a, nil
}
- if _, err := os.Stat(path); err != nil {
+ demand, digest, err := m.stagedDemand(ctx, rt, d.ID.Build, path)
+ if err != nil {
return a, err
}
- demand, digest, err := m.d.Demand(ctx, acct, app, d.ID.Build, path)
+ a.StagedPath, a.Digest, a.Demand = path, digest, demand
+ a.Mode = AttemptMode(policy, demand)
+ return a, nil
+}
+
+// stagedDemand is the canonical demand and digest of build's staged file at
+// path, for an adoption attempt (resolve) and a first wake (Plan) alike
+// (spec §4.2). A missing staged file, or one that is not the release's
+// canonical artifact (release.ErrStagedDigestMismatch), is an error: a
+// staging refusal, never a build failure. Any other trouble reading the
+// demand is hold, on the staged bytes hashed here.
+func (m *Manager) stagedDemand(ctx context.Context, rt registry.Route, build, path string) (demand, digest string, err error) {
+ if _, err := os.Stat(path); err != nil {
+ return "", "", err
+ }
+ demand, digest, err = m.d.Demand(ctx, release.RouteAccount(rt), rt.App, build, path)
if errors.Is(err, release.ErrStagedDigestMismatch) {
- return a, err
+ return "", "", err
}
if err != nil {
// The canonical manifest could not be read or compared: hold, on
// the bytes hashed below.
- m.d.Logf("rollout: %s: canonical demand of %s: %v (attempt runs in hold)", rt.Host, d.ID.Build, err)
+ m.d.Logf("rollout: %s: canonical demand of %s: %v (runs in hold)", rt.Host, build, err)
demand, digest = release.DemandHold, ""
}
if digest == "" {
// Nothing compared the staged bytes with the manifest (plan 4:
// DemandHold with no digest), so the demand read describes nothing
- // verified: hold. The attempt still runs only on the bytes hashed
+ // verified: hold. The build still runs only on the bytes hashed
// here, because plan 3's verifyStaged refuses an empty digest.
demand = release.DemandHold
if digest, err = release.FileSHA256(path); err != nil {
- return a, err
+ return "", "", err
}
}
- a.StagedPath, a.Digest, a.Demand = path, digest, demand
- a.Mode = AttemptMode(policy, demand)
- return a, nil
+ return demand, digest, nil
}
// served records a generation that became current after its checks: the
diff --git a/internal/rollout/wakeplan.go b/internal/rollout/wakeplan.go
new file mode 100644
index 000000000..df6e0248d
--- /dev/null
+++ b/internal/rollout/wakeplan.go
@@ -0,0 +1,257 @@
+package rollout
+
+import (
+ "context"
+ "fmt"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// wakeBook is the failure book's read half the planner needs.
+type wakeBook interface {
+ Blocks(acct, app, build, host string, served bool) bool
+ HeldDown(acct, app, build, host string) bool
+}
+
+type wakeInputs struct {
+ Route registry.Route
+ OptDir string
+ Policy AppPolicy
+ Book wakeBook
+ // Demand is the canonical check for an adoption (spec §4.2), returning a
+ // digest always. An error is a staging refusal: the staged file is
+ // missing or is not the release's artifact.
+ Demand func(version, stagedPath string) (demand, digest string, err error)
+}
+
+// decideWake is spec §8.2's "when the wake picks a build", as a pure
+// function. The installed build (root's stamp) boots unless a failure record
+// holds it for this instance, in which case the good build boots from its
+// staged path with its retained identity, and nothing is repointed. A build
+// this instance has never served is a first wake; in overlap mode it carries
+// the good build as its fallback. A build that failed in hold mode on this
+// instance refuses the wake: nothing falls back automatically once a hold
+// build has run against the database (spec §6.4, §8.3). An error is a
+// staging refusal for the caller to report.
+func decideWake(in wakeInputs) (activator.WakePlan, error) {
+ rt := in.Route
+ acct, app, host := release.RouteAccount(rt), rt.App, rt.Host
+ installed := rt.Version
+ if installed == "" {
+ // Plan installs a cold host first; a plan for a host with nothing
+ // installed refuses rather than let the live link run unplanned.
+ return activator.WakePlan{Refuse: "no build is installed for this instance yet"}, nil
+ }
+ hasGood := rt.GoodVersion != "" && rt.GoodDigest != ""
+ good := release.StagedIdentity{Version: rt.GoodVersion, StagedPath: release.VersionPath(in.OptDir, acct, rt.GoodVersion, app),
+ Digest: rt.GoodDigest, Demand: rt.GoodDemand}
+ recovery := activator.WakePlan{Boot: good, Mode: RecoveryMode(in.Policy, good.Demand)}
+ if in.Book.HeldDown(acct, app, installed, host) {
+ return activator.WakePlan{Refuse: "build " + installed + " failed in hold mode on this instance; recovery is explicit"}, nil
+ }
+ if hasGood && installed == rt.GoodVersion {
+ return recovery, nil
+ }
+ if in.Book.Blocks(acct, app, installed, host, false) {
+ if !hasGood {
+ return activator.WakePlan{Refuse: "build " + installed + " failed on this box and this instance has no good build"}, nil
+ }
+ return recovery, nil
+ }
+ path := release.VersionPath(in.OptDir, acct, installed, app)
+ demand, digest, err := in.Demand(installed, path)
+ if err != nil {
+ if hasGood {
+ return recovery, err
+ }
+ return activator.WakePlan{Refuse: "build " + installed + " is not staged correctly on this box"}, err
+ }
+ mode := AttemptMode(in.Policy, demand)
+ plan := activator.WakePlan{Boot: release.StagedIdentity{Version: installed, StagedPath: path, Digest: digest, Demand: demand},
+ Mode: mode, FirstWake: true}
+ if hasGood && mode == activator.ModeOverlap {
+ plan.Fallback = &good
+ }
+ return plan, nil
+}
+
+// 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.
+//
+// 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.
+func (m *Manager) Plan(ctx context.Context, rt registry.Route) activator.WakePlan {
+ if !Managed(rt) {
+ return activator.WakePlan{}
+ }
+ 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
+ // recorded by done() before anything wakes the instance, so no wake
+ // can boot the good build ahead of a hold-down record.
+ return activator.WakePlan{Refuse: "an attempt is still running for this instance"}
+ }
+ if rt.Version == "" {
+ // A cold managed host (spec §6.4, §8.4): its build is installed and
+ // stamped first, and the wake is planned against that installed
+ // identity, with its failure records and its first-wake canary.
+ // Nothing managed ever runs through an empty plan: the live-link
+ // wake would materialize the channel head and exec it unplanned.
+ if m.d.Materialize == nil {
+ return activator.WakePlan{Refuse: "no build is installed for this instance yet"}
+ }
+ if err := m.d.Materialize(ctx, rt.Host); err != nil {
+ m.d.Logf("rollout: %s: install before its wake: %v", rt.Host, err)
+ return activator.WakePlan{Refuse: "its build could not be installed"}
+ }
+ fresh, err := m.d.Routes.Get(rt.Host)
+ if err != nil || fresh == nil || fresh.Version == "" {
+ if err != nil {
+ m.d.Logf("rollout: %s: read the row after its install: %v", rt.Host, err)
+ }
+ return activator.WakePlan{Refuse: "no build is installed for this instance yet"}
+ }
+ rt = *fresh
+ }
+ 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) {
+ // The same reading an adoption attempt gets (resolve): only a
+ // staging fault is an error, and anything unverifiable is hold.
+ return m.stagedDemand(ctx, rt, ver, path)
+ }}
+ m.mu.Lock()
+ t := m.track(rt.Host)
+ t.wakeRev = t.delivered
+ m.mu.Unlock()
+ plan, err := decideWake(in)
+ if err != nil {
+ m.stagingRefused(rt, rt.Version, err)
+ }
+ if plan.FirstWake && plan.Boot.Version != "" {
+ // Read now, before the wake runs, so a failure of the build is
+ // published with no blocking read (buildFailed, spec §8.1).
+ plan.FailedUnder = m.failedUnder(rt, plan.Boot.Version)
+ }
+ if plan.Boot.StagedPath != "" {
+ plan.Check = m.planCheck(rt, plan)
+ }
+ return plan
+}
+
+// planCheck is a plan's selection gate (WakePlan.Check), asked by the
+// activator at each planned exec, after the lease and the restore (spec
+// §7.4, §8.1–8.4): a first wake's build that a failure record now holds for
+// this instance (a sibling's failure since the plan, or a hold-down) may not
+// boot. The good build always may: it is what a held build falls back to.
+// The book is in memory; nothing here blocks.
+func (m *Manager) planCheck(rt registry.Route, plan activator.WakePlan) func(build string) error {
+ acct, app, host := release.RouteAccount(rt), rt.App, rt.Host
+ boot := plan.Boot.Version
+ return func(build string) error {
+ if !plan.FirstWake || build != boot {
+ return nil
+ }
+ if m.d.Book.HeldDown(acct, app, build, host) || m.d.Book.Blocks(acct, app, build, host, false) {
+ return fmt.Errorf("build %s is held by a failure recorded since the wake was planned", build)
+ }
+ return nil
+ }
+}
+
+// attemptRunning reports whether host has an attempt in flight: admitted
+// and not yet finished by done().
+func (m *Manager) attemptRunning(host string) bool {
+ s := m.schedulerOf(host)
+ if s == nil {
+ return false
+ }
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ h := s.q.hosts[host]
+ return h != nil && h.running != nil
+}
+
+// OnGen is the activator's OnGen: a planned wake's outcome (spec §6.4, §8.2,
+// §8.5). A healthy wake records the serving build and activates the config
+// it started with; a first wake becomes the good build only once it has
+// survived its 10 s; a failed one goes through the cohort-failure
+// 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.
+func (m *Manager) OnGen(ev activator.GenEvent) {
+ rt := ev.Route
+ switch ev.Kind {
+ case activator.GenHealthy:
+ // A wake that is not a first wake includes the good build booted
+ // untried under Task 9's verdict floor: an ordinary wake of the good
+ // build, recorded as served, with nothing against the build it skipped.
+ if !ev.FirstWake {
+ m.served(rt.Host, ev.Gen)
+ }
+ // 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)
+ case activator.GenSettled:
+ m.served(rt.Host, ev.Gen)
+ case activator.GenFellBack:
+ // The failure hold first (spec §8.1): nothing that reads the
+ // serving build or the active config may see the fallback before the
+ // failed build is held.
+ m.buildFailed(rt, ev)
+ m.publishServing(rt.Host, ev.Gen)
+ m.wakeConfigActive(rt.Host)
+ m.emit(rt, EvFallbackServed, "warn", textFallback(ev.Failed.Version, ev.Gen.Build))
+ case activator.GenFailed:
+ m.buildFailed(rt, ev)
+ case activator.GenCancelled:
+ // An intentional stop ended the first wake's observation (spec
+ // §6.4): no verdict on the build, so nothing is recorded or published.
+ }
+}
+
+// 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) {
+ // No blocking read before the record (spec §8.1): the failing pointer's
+ // PromotedAt was read when the wake was planned (WakePlan.FailedUnder).
+ rec := FailureRecord{Account: release.RouteAccount(rt), App: rt.App, Build: ev.Failed.Version, Host: rt.Host,
+ Cause: string(activator.CauseRelease), Mode: string(ev.Mode), Reason: ev.Reason,
+ PromotedAt: ev.FailedUnder, At: m.d.Now()}
+ 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
+ // record (as done() does for an attempt).
+ rec.PromotedAt = rec.At
+ }
+ if ev.Mode == activator.ModeHold {
+ // A hold build executed against this instance's database: nothing
+ // falls back on it automatically (spec §6.4, §8.3).
+ rec.HeldDown = []string{rt.Host}
+ }
+ s := m.schedulerOf(rt.Host)
+ if s == nil {
+ if _, err := m.d.Book.Record(rec); err != nil {
+ m.d.Logf("rollout: write failure record %s: %v (held in memory)", rec.Build, err)
+ }
+ return
+ }
+ m.cohortFailed(s, rt, rec)
+}
diff --git a/internal/rollout/wakeplan_test.go b/internal/rollout/wakeplan_test.go
new file mode 100644
index 000000000..ab2f50b92
--- /dev/null
+++ b/internal/rollout/wakeplan_test.go
@@ -0,0 +1,530 @@
+package rollout
+
+import (
+ "context"
+ "errors"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func wakeRoute(installed, good string) registry.Route {
+ return registry.Route{Host: "a", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Hibernate: true,
+ Version: installed, GoodVersion: good, GoodDigest: "sha-" + good}
+}
+
+func inputs(rt registry.Route, b *FailureBook, p AppPolicy) wakeInputs {
+ return wakeInputs{Route: rt, OptDir: "/opt/carlos", Policy: p, Book: b,
+ Demand: func(ver, _ string) (string, string, error) { return "", "sha-" + ver, nil }}
+}
+
+func TestDecideWake(t *testing.T) {
+ overlap := DefaultPolicy()
+ held := &FailureBook{}
+ held.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "x", At: t0})
+ down := &FailureBook{}
+ down.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a", HeldDown: []string{"a"}, At: t0})
+
+ if p, _ := decideWake(inputs(wakeRoute("", ""), &FailureBook{}, overlap)); p.Boot.StagedPath != "" || p.Refuse == "" {
+ t.Fatal("nothing stamped: today's wake, unplanned")
+ }
+ p, _ := decideWake(inputs(wakeRoute("v1", "v1"), &FailureBook{}, overlap))
+ if p.Boot.Version != "v1" || p.FirstWake || p.Boot.Digest != "sha-v1" {
+ t.Fatalf("installed is the good build: recovery by the retained identity, got %+v", p)
+ }
+ p, _ = decideWake(inputs(wakeRoute("v2", "v1"), &FailureBook{}, overlap))
+ if p.Boot.Version != "v2" || !p.FirstWake || p.Fallback == nil || p.Fallback.Version != "v1" || p.Mode != activator.ModeOverlap {
+ t.Fatalf("a new build: a first wake with the good build behind it, got %+v", p)
+ }
+ p, _ = decideWake(inputs(wakeRoute("v2", "v1"), &FailureBook{}, AppPolicy{Swap: activator.ModeHold, Concurrency: 1}))
+ if p.Mode != activator.ModeHold || p.Fallback != nil {
+ t.Fatalf("hold: no fallback, got %+v", p)
+ }
+ in := inputs(wakeRoute("v2", "v1"), &FailureBook{}, overlap)
+ in.Demand = func(ver, _ string) (string, string, error) { return "hold", "sha-" + ver, nil }
+ if p, _ := decideWake(in); p.Mode != activator.ModeHold || p.Fallback != nil {
+ t.Fatalf("the release demands hold: no fallback, got %+v", p)
+ }
+ if p, _ := decideWake(inputs(wakeRoute("v2", "v1"), held, overlap)); p.Boot.Version != "v1" || p.FirstWake {
+ t.Fatalf("installed build held by a record: boot the good build, got %+v", p)
+ }
+ if p, _ := decideWake(inputs(wakeRoute("v2", ""), held, overlap)); p.Refuse == "" {
+ t.Fatal("held, with no good build: nothing to boot")
+ }
+ if p, _ := decideWake(inputs(wakeRoute("v2", "v1"), down, overlap)); p.Refuse == "" {
+ t.Fatal("held down after a hold failure: no automatic fallback (spec §6.4)")
+ }
+ in = inputs(wakeRoute("v2", "v1"), &FailureBook{}, overlap)
+ in.Demand = func(string, string) (string, string, error) { return "", "", errors.New("digest mismatch") }
+ p, err := decideWake(in)
+ if err == nil || p.Boot.Version != "v1" || p.FirstWake {
+ t.Fatalf("a staging refusal with a good build boots the good build, got %+v, %v", p, err)
+ }
+}
+
+// Review focus 3.
+func TestPlanBootsTheGoodBuildForAHeldInstallation(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.install("v2", "a")
+ h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "b", At: t0})
+ p := h.m.Plan(context.Background(), h.rows.get("a"))
+ if p.Boot.Version != "v1" || p.FirstWake {
+ t.Fatalf("plan = %+v, want the good v1 without a repoint", p)
+ }
+}
+
+// registerThenInstall makes the scheduler know host (the pass) while every
+// host still wants v1, and only then stamps v2 on host's row. The pass sees
+// nothing to adopt, so no swap onto v2 runs beside the wake outcome a test
+// hands OnGen: these tests are about the wake, and a swap racing it would
+// change what the activator serves under them.
+func (h *harness) registerThenInstall(ver, host string) {
+ h.pass()
+ h.install(ver, host)
+}
+
+func TestAFellBackWakeRecordsTheFailureAndKeepsTheGoodBuild(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.registerThenInstall("v2", "a")
+ rt := h.rows.get("a")
+ h.act.serve("a", "v1") // the good build the fallback left serving
+ served, _ := h.act.Current("a")
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenFellBack, Route: rt, FirstWake: true, Mode: activator.ModeOverlap,
+ Gen: served, Failed: release.StagedIdentity{Version: "v2"}, Reason: "health check failed"})
+ if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {
+ t.Fatal("no failure record")
+ }
+ if r := h.rows.get("a"); r.GoodVersion != "v1" || r.ServingVersion != "v1" {
+ t.Fatalf("row = %+v", r)
+ }
+ if !h.events.has("served build v1 instead") {
+ t.Fatal("no fallback event")
+ }
+}
+
+// Spec §8.1: a fallback publishes the failure hold before the serving build
+// and the active config. Writing the serving build first showed the good
+// build serving with no hold on the failed one, a window in which a wake or
+// an admission could select the failed build again.
+func TestAFallbackHoldsTheFailedBuildBeforeItPublishesServing(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.registerThenInstall("v2", "a")
+ h.act.serve("a", "v1")
+ served, _ := h.act.Current("a")
+ var early atomic.Int32
+ h.rows.mu.Lock()
+ h.rows.setServingHook = func(string) {
+ if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {
+ early.Add(1)
+ }
+ }
+ h.rows.mu.Unlock()
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenFellBack, Route: h.rows.get("a"), FirstWake: true, Mode: activator.ModeOverlap,
+ Gen: served, Failed: release.StagedIdentity{Version: "v2"}, Reason: "health check failed"})
+ if n := early.Load(); n != 0 {
+ t.Fatal("the serving build was published before the failed build was held")
+ }
+ if r := h.rows.get("a"); r.ServingVersion != "v1" {
+ t.Fatalf("serving = %q, want the good v1", r.ServingVersion)
+ }
+}
+
+func TestOnlyASettledFirstWakeBecomesTheGoodBuild(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.registerThenInstall("v2", "a")
+ rt := h.rows.get("a")
+ h.act.serve("a", "v2") // the first wake's generation, which the activator serves
+ g, _ := h.act.Current("a")
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: g, FirstWake: true})
+ if h.rows.get("a").GoodVersion != "v1" {
+ t.Fatal("a first wake became the good build before its 10 s (spec §8.5)")
+ }
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: rt, Gen: g, FirstWake: true})
+ if r := h.rows.get("a"); r.GoodVersion != "v2" || r.ServingVersion != "v2" {
+ t.Fatalf("row = %+v: the settled first wake must be the good and the serving build", r)
+ }
+}
+
+// Task 9's verdict floor: a first wake left too little health window boots
+// the good build untried and reports an ordinary healthy wake of it
+// (FirstWake false). The scheduler records the good build as serving and
+// records nothing against the untried build, which stays installed for a
+// later wake.
+func TestAnUntriedFirstWakeIsAnOrdinaryWakeOfTheGoodBuild(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.registerThenInstall("v2", "a")
+ h.act.serve("a", "v1")
+ g, _ := h.act.Current("a")
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get("a"), Gen: g, Mode: activator.ModeOverlap})
+ if _, ok := h.book.Lookup("bab", "blog", "v2"); ok {
+ t.Fatal("the untried build was recorded as failed")
+ }
+ if r := h.rows.get("a"); r.ServingVersion != "v1" || r.GoodVersion != "v1" || r.Version != "v2" {
+ t.Fatalf("row = %+v: want v1 serving and good, v2 still installed", r)
+ }
+ if p := h.m.Plan(context.Background(), h.rows.get("a")); !p.FirstWake || p.Boot.Version != "v2" {
+ t.Fatalf("plan %+v: the untried build must stay a candidate for the next wake", p)
+ }
+}
+
+// Spec §6.4: an intentional stop during the 10 s observation records
+// nothing. GenCancelled only closes the observation.
+func TestACancelledObservationRecordsNothing(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.registerThenInstall("v2", "a")
+ h.act.serve("a", "v2")
+ g, _ := h.act.Current("a")
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("a"), Gen: g, FirstWake: true})
+ if _, ok := h.book.Lookup("bab", "blog", "v2"); ok {
+ t.Fatal("a cancelled observation recorded a failure")
+ }
+ if r := h.rows.get("a"); r.GoodVersion != "v1" || r.ServingVersion != "" {
+ t.Fatalf("row = %+v: a cancelled observation published a build", r)
+ }
+}
+
+// Spec §7.9: a worker's serving build has one writer, the activator
+// (ordered by incarnation). The scheduler records a worker's good build but
+// leaves its ServingVersion alone, so its write can never interleave with
+// the activator's and restore an older value.
+func TestTheSchedulerLeavesAWorkersServingVersionAlone(t *testing.T) {
+ h := newHarness(t, "a")
+ h.act.serve("a", "v2")
+ g, _ := h.act.Current("a")
+ g.Worker = true
+ h.m.served("a", g)
+ if r := h.rows.get("a"); r.ServingVersion != "" || r.GoodVersion != "v2" {
+ t.Fatalf("row = %+v: want the good build recorded and the serving build left to the activator", r)
+ }
+}
+
+// Spec §8.5: a good build is written expecting the current one. A late
+// success of an older generation (a completion delayed past a newer swap, a
+// stale wake event) is not the generation the activator serves, so it never
+// overwrites the newer good or serving build. Writing whatever arrived last
+// regressed both to the older build.
+func TestAStaleSuccessNeverOverwritesANewerBuild(t *testing.T) {
+ h := newHarness(t, "a")
+ h.act.serve("a", "v2")
+ older, _ := h.act.Current("a")
+ h.act.serve("a", "v3") // a newer generation took over
+ newer, _ := h.act.Current("a")
+ h.m.served("a", newer)
+ h.m.served("a", older) // the older one's completion arrives late
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: h.rows.get("a"), Gen: older, FirstWake: true})
+ if r := h.rows.get("a"); r.GoodVersion != "v3" || r.ServingVersion != "v3" {
+ t.Fatalf("row = %+v: an older generation overwrote the newer good and serving build", r)
+ }
+}
+
+func TestAFailedHoldFirstWakeHoldsTheInstanceDown(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.registerThenInstall("v2", "a")
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: h.rows.get("a"), FirstWake: true, Mode: activator.ModeHold,
+ Failed: release.StagedIdentity{Version: "v2"}, Reason: "health check failed"})
+ if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Refuse == "" {
+ t.Fatalf("plan = %+v, want a refusal: recovery after a hold failure is explicit", p)
+ }
+ // Task 4's rule: a record never carries a zero PromotedAt. No pointer was
+ // read for this wake, so the failure's own time stands in.
+ if rec, ok := h.book.Lookup("bab", "blog", "v2"); !ok || !rec.PromotedAt.Equal(t0) || !rec.At.Equal(t0) {
+ t.Fatalf("record %+v %v: want PromotedAt = At = the failure's time", rec, ok)
+ }
+ if !h.events.has("rollout of build v2 halted on box box-1: health") {
+ t.Fatal("no cohort-halted event with the reason word")
+ }
+}
+
+func TestAWakeActivatesTheConfigItStartedWith(t *testing.T) {
+ h := newHarness(t, "a")
+ h.pass()
+ h.act.mu.Lock()
+ delete(h.act.gens, "a")
+ h.act.owned["a"] = true // waking: a process, not yet current
+ h.act.mu.Unlock()
+ var acks int
+ ack := h.m.DeferAck(func(appconfig.ConfigAck) { acks++ })
+ h.m.ConfigDelivered("a", "sha256:c1")
+ ack(appconfig.ConfigAck{Host: "a"})
+ if acks != 0 {
+ 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"}})
+ if acks != 1 {
+ t.Fatalf("acks = %d after the wake, want 1", acks)
+ }
+}
+
+// Spec §8.1: a sleeping instance needs no config attempt. Its next wake
+// starts with the delivered file, and config sync's ack goes out at
+// delivery, as today, marked Hibernating.
+func TestASleepersConfigNeedsNoAttempt(t *testing.T) {
+ h := newHarness(t, "a")
+ h.pass()
+ h.act.mu.Lock()
+ delete(h.act.gens, "a") // asleep: no process at all
+ h.act.mu.Unlock()
+ var acks []appconfig.ConfigAck
+ ack := h.m.DeferAck(func(a appconfig.ConfigAck) { acks = append(acks, a) })
+ h.m.ConfigDelivered("a", "sha256:c1")
+ ack(appconfig.ConfigAck{Host: "a", Restarted: true})
+ if len(acks) != 1 || !acks[0].Hibernating || acks[0].Restarted {
+ t.Fatalf("acks = %+v, want one at delivery, Hibernating", acks)
+ }
+ h.pass()
+ time.Sleep(50 * time.Millisecond)
+ if calls, _, _ := h.act.snapshot(); len(calls) != 0 {
+ t.Fatalf("attempts %+v: a sleeper got a config attempt", calls)
+ }
+}
+
+// Spec §8.1: a failed first wake is published with no blocking read in
+// between. The failing pointer's PromotedAt is read when the wake is
+// planned and travels on the event. buildFailed read it again after the
+// failure, before the record, leaving every sibling's wake free to plan the
+// failed build meanwhile.
+func TestAFailedFirstWakeIsRecordedWithNoPointerRead(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.install("v2", "a")
+ promoted := t0.Add(-time.Hour)
+ h.pointer("stable", "v2", promoted)
+ p := h.m.Plan(context.Background(), h.rows.get("a"))
+ if !p.FirstWake || !p.FailedUnder.Equal(promoted) {
+ t.Fatalf("plan %+v: want a first wake carrying the pointer's PromotedAt", p)
+ }
+ var reads atomic.Int32
+ h.mu.Lock()
+ h.pointerHook = func() { reads.Add(1) }
+ h.mu.Unlock()
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: h.rows.get("a"), FirstWake: true, Mode: activator.ModeOverlap,
+ Failed: p.Boot, Reason: "GET / answered 500", FailedUnder: p.FailedUnder})
+ if n := reads.Load(); n != 0 {
+ t.Fatalf("%d pointer read(s) between the failure and its record", n)
+ }
+ rec, ok := h.book.Lookup("bab", "blog", "v2")
+ if !ok || !rec.PromotedAt.Equal(promoted) {
+ t.Fatalf("record %+v %v: want one carrying the planned PromotedAt", rec, ok)
+ }
+}
+
+// Spec §6.4, §8.4: a cold managed host (nothing installed yet) has its build
+// installed and stamped before the wake is planned, so the plan boots that
+// identity, as a first wake, and respects its failure records. An empty
+// plan sent the wake down the live-link path, which materialized the
+// channel head and ran it unplanned, a held build included.
+func TestAColdManagedHostIsInstalledBeforeItIsPlanned(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ cold := func() {
+ r := h.rows.get("a")
+ r.Version = ""
+ h.rows.put(r)
+ }
+ cold()
+ installs := 0
+ h.m.d.Materialize = func(_ context.Context, host string) error {
+ installs++
+ r := h.rows.get(host)
+ r.Version = "v2"
+ h.rows.put(r)
+ return nil
+ }
+ if p := h.m.Plan(context.Background(), h.rows.get("a")); installs != 1 || p.Boot.Version != "v2" || !p.FirstWake {
+ t.Fatalf("installs %d, plan %+v: want v2 installed, then planned as a first wake", installs, p)
+ }
+ if _, err := h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a", At: t0}); err != nil {
+ t.Fatal(err)
+ }
+ cold()
+ if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Boot.Version == "v2" {
+ t.Fatalf("plan %+v: a cold host was booted onto a build its app holds", p)
+ }
+ h.m.d.Materialize = nil
+ cold()
+ if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Refuse == "" || p.Boot.StagedPath != "" {
+ t.Fatalf("plan %+v: with nothing installed and no way to install, the wake must be refused", p)
+ }
+}
+
+// A refusal's text reaches the member's "wake failed" event (Task 9). A
+// failed install's error, which can name box paths, goes to the operator log
+// only.
+func TestAFailedColdInstallRefusesWithoutItsErrorText(t *testing.T) {
+ h := newHarness(t, "a")
+ r := h.rows.get("a")
+ r.Version = ""
+ h.rows.put(r)
+ h.m.d.Materialize = func(context.Context, string) error {
+ return errors.New("stage /opt/carlos/versions/bab/v2/blog: no space left on device")
+ }
+ p := h.m.Plan(context.Background(), h.rows.get("a"))
+ if p.Refuse == "" || p.Boot.StagedPath != "" {
+ t.Fatalf("plan %+v: a failed install must refuse the wake", p)
+ }
+ if strings.Contains(p.Refuse, "/opt") || strings.Contains(p.Refuse, "no space") {
+ t.Fatalf("refusal %q carries the raw error", p.Refuse)
+ }
+}
+
+// Spec §4.2 (Task 3's carry, as resolve applies it to an attempt): a first
+// wake whose canonical demand cannot be read or compared boots in hold, on
+// the staged bytes hashed here, with no fallback. Missing or unverifiable
+// data never permits an overlap. A staged file that is not the release's
+// artifact is a staging refusal: the good build boots and root is asked to
+// stage again.
+func TestAPlannedWakeReadsTheDemandAsAnAttemptDoes(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.install("v2", "a")
+ sum, err := release.FileSHA256(release.VersionPath(h.opt, "bab", "v2", "blog"))
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ h.mu.Lock()
+ h.demandErr["v2"] = errors.New("manifest unreadable")
+ h.mu.Unlock()
+ if p := h.m.Plan(context.Background(), h.rows.get("a")); !p.FirstWake || p.Mode != activator.ModeHold || p.Fallback != nil || p.Boot.Digest != sum {
+ t.Fatalf("plan %+v: an unreadable demand must boot v2 in hold on its hashed bytes", p)
+ }
+
+ h.mu.Lock()
+ delete(h.demandErr, "v2")
+ h.noDigest["v2"] = true
+ h.mu.Unlock()
+ if p := h.m.Plan(context.Background(), h.rows.get("a")); !p.FirstWake || p.Mode != activator.ModeHold || p.Fallback != nil || p.Boot.Digest != sum {
+ t.Fatalf("plan %+v: a demand nothing compared must boot v2 in hold on its hashed bytes", p)
+ }
+
+ h.mu.Lock()
+ delete(h.noDigest, "v2")
+ h.mismatch["v2"] = true
+ h.mu.Unlock()
+ p := h.m.Plan(context.Background(), h.rows.get("a"))
+ if p.FirstWake || p.Boot.Version != "v1" {
+ t.Fatalf("plan %+v: a mis-staged v2 must boot the good v1", p)
+ }
+ if h.touched.Load() != 1 || !h.events.has("build v2 is not staged correctly on this box") {
+ t.Fatal("a staging refusal must ask root to stage again and say so")
+ }
+}
+
+// Spec §7.4, §8.1–8.4 (P5 review finding 3): a plan carries its selection
+// gate. A sibling's failure recorded after the plan was made refuses the
+// planned build at the exec; the good build still passes.
+func TestAPlansCheckRefusesABuildHeldSinceThePlan(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.install("v2", "a")
+ p := h.m.Plan(context.Background(), h.rows.get("a"))
+ if !p.FirstWake || p.Boot.Version != "v2" || p.Check == nil {
+ t.Fatalf("plan %+v: want a first wake onto v2 with its check", p)
+ }
+ if err := p.Check("v2"); err != nil {
+ t.Fatalf("nothing holds v2 yet: %v", err)
+ }
+ if _, err := h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "b", At: t0}); err != nil {
+ t.Fatal(err)
+ }
+ if err := p.Check("v2"); err == nil {
+ t.Fatal("v2 failed on a sibling after the plan, and the check still let it boot")
+ }
+ if err := p.Check("v1"); err != nil {
+ t.Fatalf("the good build was refused: %v", err)
+ }
+}
+
+// Spec §2.9 (rev 17), §6.4, §8.3: an overlap attempt on a foreign-layout row
+// runs as a hold, and the activator says so (SwapResult.Mode). Its failure
+// is a hold build's failure: held down, so no wake boots the good build in
+// its place. And while an attempt runs for an instance, no wake of it is
+// planned, so a hold that went down is recorded before anything wakes it.
+// Recording under the dispatched ModeOverlap left it un-held, and the next
+// wake booted the good build against the database the hold build ran on.
+func TestAForeignRowsFailedHoldIsHeldDown(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.act.outcome = func(activator.Attempt) activator.SwapOutcome { return activator.SwapDown }
+ h.act.mode = activator.ModeHold // the effective mode, as plan 3's Swap reports it
+ gate := h.act.hold("a")
+ h.install("v2", "a")
+ h.pass()
+ waitFor(t, "the attempt to start", func() bool { _, started, _ := h.act.snapshot(); return len(started) == 1 })
+ if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Refuse == "" {
+ t.Fatalf("a wake was planned while the instance's attempt ran: %+v", p)
+ }
+ close(gate)
+ waitFor(t, "the hold-down record", func() bool { return h.book.HeldDown("bab", "blog", "v2", "a") })
+ if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Refuse == "" || p.Boot.StagedPath != "" {
+ t.Fatalf("after the hold went down the wake planned %+v, want refused (held down)", p)
+ }
+}
+
+// The activator's sweep calls OnDrift (Nudge) on a goroutine of its own for
+// every drifted entry on every sweep, with no dedupe of its own (Task 9).
+// Nudges for one host collapse: while one is reading the row, the others
+// return at once and leave it one more look, so nothing piles up behind a
+// slow read and nothing a later nudge was for is lost. And a nudge for the
+// build an attempt is already running onto neither queues it again nor
+// supersedes it.
+func TestNudgesForOneHostCollapse(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.pass()
+ release1 := make(chan struct{})
+ var gets atomic.Int32
+ h.rows.mu.Lock()
+ h.rows.getHook = func(host string) {
+ if host == "a" && gets.Add(1) == 1 {
+ <-release1
+ }
+ }
+ h.rows.mu.Unlock()
+ first := make(chan struct{})
+ go func() { h.m.Nudge("a"); close(first) }()
+ waitFor(t, "the first nudge to read the row", func() bool { return gets.Load() == 1 })
+ for range 20 {
+ h.m.Nudge("a") // returns at once: the first nudge owns this host
+ }
+ if n := gets.Load(); n != 1 {
+ t.Fatalf("%d row reads while one nudge was in flight, want 1", n)
+ }
+ close(release1)
+ <-first
+ if n := gets.Load(); n != 2 {
+ t.Fatalf("%d row reads, want 2: the in-flight nudge looks once more for the ones it absorbed", n)
+ }
+
+ h.rows.mu.Lock()
+ h.rows.getHook = nil
+ h.rows.mu.Unlock()
+ gate := h.act.hold("a")
+ h.install("v2", "a")
+ h.m.Nudge("a")
+ waitFor(t, "the attempt onto v2 to start", func() bool { _, started, _ := h.act.snapshot(); return len(started) == 1 })
+ for range 5 {
+ h.m.Nudge("a")
+ }
+ close(gate)
+ waitFor(t, "the attempt to finish", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
+ calls, started, _ := h.act.snapshot()
+ if calls[0].outcome != activator.SwapDone || len(started) != 1 {
+ t.Fatalf("calls %+v started %v: repeated nudges re-queued or superseded the running attempt", calls, started)
+ }
+}