amadan

carlos / platform Public

zdd-5 task 8: the manager and the per-app scheduler

The rollout Manager and one scheduler goroutine per app: feeds (Pass,
Nudge, Adopted, Restart, ConfigDelivered, DeferAck), attempt resolution,
outcomes, good builds, the cohort-failure transition, config activation,
pointer retries and RetryHost. Tested against fakeAct until Tasks 9 and 11
give the activator Current and ReplaceWorker.

Carried rulings: member events carry reason words, never raw reasons
(reasonWord moves forward from Task 13); a demand error other than a
digest mismatch, or no digest, runs the attempt in hold; an abandoned
attempt still newest waits before it runs again; a hold's SwapDown is held
down in the step that frees its slot; a failed config attempt's backoff is
set in that step too, with no registry read in front; RetryHost clears the
book and binds the retry flag under one scheduler lock; a halt is lifted
once its record is gone; a sleeper's pending config is active on its next
wake and its held ack goes out marked Hibernating.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude e51bd9bec9d6b8a0210de21b8d58650b287b2ea2 parent a12b53c
5 files changed, +2469 −0
  • internal/rollout/events.go +94 −0
  • internal/rollout/fake_test.go +427 −0
  • internal/rollout/manager.go +808 −0
  • internal/rollout/manager_test.go +882 −0
  • internal/rollout/scheduler.go +258 −0
diff --git a/internal/rollout/events.go b/internal/rollout/events.go
new file mode 100644
index 000000000..82a4bca82
--- /dev/null
+++ b/internal/rollout/events.go
@@ -0,0 +1,94 @@
+package rollout
+
+import (
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+)
+
+// EventCode is a rollout event's stable internal name (spec §9.4). The codes
+// are what the journal and tests key on. The member-facing texts below are
+// DRAFTS: plan 6 rewrites every one of them through copy review, and the
+// CARLOS_SWAP flag stays off outside dev until it has.
+type EventCode string
+
+const (
+ EvSwapping EventCode = "swap.started"
+ EvSwapped EventCode = "swap.done"
+ EvSwapFailed EventCode = "swap.failed"
+ EvSwapDown EventCode = "swap.down"
+ EvCohortHalted EventCode = "cohort.halted"
+ EvFallbackServed EventCode = "fallback.served"
+ EvConfigPending EventCode = "config.pending"
+ EvStagingRefused EventCode = "staging.refused"
+)
+
+// The reason words: the closed vocabulary a member-facing event or status
+// line may give for a failure. Task 13 moves them into the status package
+// (status.Reason*) and points these at them.
+const (
+ reasonHealth = "health"
+ reasonSettle = "settle"
+ reasonExit = "exit"
+ reasonStaging = "staging"
+ reasonReused = "reused"
+)
+
+// reasonWord turns an attempt's diagnostic text (SwapResult.Reason, an error)
+// into one of the reason words, so no free text, which can name paths and
+// internals, reaches a member-facing surface. The raw text goes to the
+// operator log instead. Anything unrecognised is a failed health check: the
+// swap's checks are the health check, the settle and the exit watch, and the
+// health check is the one a spawn or start error also fails.
+func reasonWord(text string) string {
+ t := strings.ToLower(text)
+ switch {
+ case strings.Contains(t, "reused"):
+ return reasonReused
+ case strings.Contains(t, "staging"), strings.Contains(t, "staged"), strings.Contains(t, "digest"):
+ return reasonStaging
+ case strings.Contains(t, "settle"):
+ return reasonSettle
+ case strings.Contains(t, "exited"):
+ return reasonExit
+ }
+ return reasonHealth
+}
+
+func textSwapping(build string, cause activator.Cause) string {
+ return "swapping to build " + build + " (" + string(cause) + ")"
+}
+
+// textSwapped carries the retiring generation's drain time (spec §9.4).
+func textSwapped(build string, drain time.Duration) string {
+ return "swapped to build " + build + ". the previous build drained in " + drain.Round(100*time.Millisecond).String()
+}
+
+// textSwapFailed's reason is a reason word (reasonWord), never raw text.
+func textSwapFailed(build, serving, reason string) string {
+ return "swap to build " + build + " failed: " + reason + ". still serving build " + serving
+}
+
+func textSwapDown(build string) string {
+ return "build " + build + " failed in hold mode. the instance stays down until it is rolled back with carlos rollback"
+}
+
+// textCohortHalted names the box (spec §9.4); the event sink already
+// carries the account, the app and the host. reason is a reason word.
+func textCohortHalted(build, box, reason string) string {
+ return "rollout of build " + build + " halted on box " + box + ": " + reason
+}
+
+func textFallback(failed, good string) string {
+ return "build " + failed + " failed to start. served build " + good + " instead"
+}
+
+// textConfigPending's reason is a reason word.
+func textConfigPending(reason string, wait time.Duration) string {
+ return "config delivered, not active: " + reason + ". trying again in " + wait.String()
+}
+
+func textStagingRefused(build string) string {
+ return "build " + build + " is not staged correctly on this box. asked for it to be staged again"
+}
diff --git a/internal/rollout/fake_test.go b/internal/rollout/fake_test.go
new file mode 100644
index 000000000..686ababe8
--- /dev/null
+++ b/internal/rollout/fake_test.go
@@ -0,0 +1,427 @@
+package rollout
+
+import (
+ "context"
+ "fmt"
+ "sort"
+ "strings"
+ "sync"
+ "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"
+)
+
+type swapCall struct {
+ a activator.Attempt
+ outcome activator.SwapOutcome
+}
+
+type replaceCall struct {
+ a activator.Attempt
+ fallback *release.StagedIdentity
+}
+
+// fakeAct models plan 3's Swap faithfully on the one point this plan relies
+// on: the re-check before the switch. A superseded attempt never switches.
+type fakeAct struct {
+ mu sync.Mutex
+ gens map[string]activator.Gen
+ owned map[string]bool
+ n uint64
+ gates map[string]chan struct{}
+ outcome func(a activator.Attempt) activator.SwapOutcome
+ started []string
+ calls []swapCall
+ inFlight int
+ maxFlight int
+ replaced []replaceCall
+ worker func(a activator.Attempt, fallback *release.StagedIdentity) activator.SwapResult
+ // mode, when set, is the effective mode Swap reports (SwapResult.Mode):
+ // plan 3 runs an overlap attempt on a foreign-layout row as a hold.
+ mode activator.SwapMode
+ // reason, when set, is a failed Swap's diagnostic text in place of
+ // "health check failed".
+ reason string
+}
+
+func newFakeAct() *fakeAct {
+ return &fakeAct{gens: map[string]activator.Gen{}, owned: map[string]bool{}, gates: map[string]chan struct{}{}}
+}
+
+func (f *fakeAct) Current(host string) (activator.Gen, bool) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ g, ok := f.gens[host]
+ return g, ok
+}
+
+func (f *fakeAct) Owns(host string) bool {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ _, ok := f.gens[host]
+ return ok || f.owned[host]
+}
+
+func (f *fakeAct) serve(host, build string) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.n++
+ f.gens[host] = activator.Gen{N: f.n, Build: build, StagedPath: "/opt/" + build, Digest: "sha-" + build}
+}
+
+func (f *fakeAct) hold(host string) chan struct{} {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ ch := make(chan struct{})
+ f.gates[host] = ch
+ return ch
+}
+
+func (f *fakeAct) Swap(_ context.Context, a activator.Attempt, newest func(string) uint64) activator.SwapResult {
+ f.mu.Lock()
+ f.inFlight++
+ f.maxFlight = max(f.maxFlight, f.inFlight)
+ f.started = append(f.started, a.Host+":"+a.Build)
+ gate := f.gates[a.Host]
+ delete(f.gates, a.Host)
+ f.mu.Unlock()
+ if gate != nil {
+ <-gate
+ }
+ out := activator.SwapDone
+ if newest(a.Host) != a.SchedRev {
+ out = activator.SwapAbandoned
+ } else if f.outcome != nil {
+ out = f.outcome(a)
+ }
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.inFlight--
+ f.calls = append(f.calls, swapCall{a: a, outcome: out})
+ if out != activator.SwapDone {
+ reason := "health check failed"
+ if f.reason != "" {
+ reason = f.reason
+ }
+ return activator.SwapResult{Outcome: out, Reason: reason, Mode: f.mode}
+ }
+ f.n++
+ g := activator.Gen{N: f.n, Build: a.Build, StagedPath: a.StagedPath, Digest: a.Digest, Demand: a.Demand}
+ f.gens[a.Host] = g
+ return activator.SwapResult{Outcome: out, Gen: g}
+}
+
+// ReplaceWorker models plan 5's same-entry socketless attempt (Task 11):
+// worker decides the outcome, and every call is recorded with the fallback
+// it was handed. The default is a clean replacement onto the attempt's build.
+func (f *fakeAct) ReplaceWorker(_ context.Context, a activator.Attempt, fallback *release.StagedIdentity, newest func(string) uint64) activator.SwapResult {
+ f.mu.Lock()
+ f.replaced = append(f.replaced, replaceCall{a: a, fallback: fallback})
+ decide := f.worker
+ f.mu.Unlock()
+ if newest(a.Host) != a.SchedRev {
+ return activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: "superseded"}
+ }
+ res := activator.SwapResult{Outcome: activator.SwapDone}
+ if decide != nil {
+ res = decide(a, fallback)
+ }
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ switch {
+ case res.Outcome == activator.SwapDone:
+ f.n++
+ res.Gen = activator.Gen{N: f.n, Build: a.Build, StagedPath: a.StagedPath, Digest: a.Digest, Demand: a.Demand}
+ f.gens[a.Host] = res.Gen
+ case res.Outcome == activator.SwapFailed && fallback != nil:
+ f.n++
+ res.Gen = activator.Gen{N: f.n, Build: fallback.Version, StagedPath: fallback.StagedPath, Digest: fallback.Digest, Demand: fallback.Demand}
+ f.gens[a.Host] = res.Gen
+ case res.Outcome == activator.SwapDown:
+ delete(f.gens, a.Host)
+ }
+ return res
+}
+
+func (f *fakeAct) replacements() []replaceCall {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return append([]replaceCall(nil), f.replaced...)
+}
+
+func (f *fakeAct) snapshot() ([]swapCall, []string, int) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return append([]swapCall(nil), f.calls...), append([]string(nil), f.started...), f.maxFlight
+}
+
+type fakeRoutes struct {
+ mu sync.Mutex
+ rows map[string]registry.Route
+ // getHook, when set, runs at the start of every Get, with no lock held:
+ // a test's window into the moment the scheduler reads a row. getErr,
+ // when set and answering an error, fails that Get.
+ getHook func(host string)
+ getErr func(host string) error
+ // setGoodHook, when set, runs at the start of every SetGoodBuild, with
+ // no lock held. setServingHook does the same for SetServingVersion.
+ setGoodHook func(host string)
+ setServingHook func(host string)
+}
+
+func (f *fakeRoutes) Get(host string) (*registry.Route, error) {
+ f.mu.Lock()
+ hook, fail := f.getHook, f.getErr
+ f.mu.Unlock()
+ if hook != nil {
+ hook(host)
+ }
+ if fail != nil {
+ if err := fail(host); err != nil {
+ return nil, err
+ }
+ }
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ r, ok := f.rows[host]
+ if !ok {
+ return nil, nil
+ }
+ return &r, nil
+}
+
+func (f *fakeRoutes) List() ([]registry.Route, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ out := make([]registry.Route, 0, len(f.rows))
+ for _, r := range f.rows {
+ out = append(out, r)
+ }
+ sort.Slice(out, func(i, j int) bool { return out[i].Host < out[j].Host })
+ return out, nil
+}
+
+func (f *fakeRoutes) SetGoodBuild(host string, expect, next registry.GoodBuild) error {
+ f.mu.Lock()
+ hook := f.setGoodHook
+ f.mu.Unlock()
+ if hook != nil {
+ hook(host)
+ }
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ r := f.rows[host]
+ if r.Good() != expect {
+ return registry.ErrGoodBuildChanged
+ }
+ r.GoodVersion, r.GoodDigest, r.GoodDemand = next.Version, next.Digest, next.Demand
+ f.rows[host] = r
+ return nil
+}
+
+func (f *fakeRoutes) SetServingVersion(host, ver string) error {
+ f.mu.Lock()
+ hook := f.setServingHook
+ f.mu.Unlock()
+ if hook != nil {
+ hook(host)
+ }
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ r := f.rows[host]
+ r.ServingVersion = ver
+ f.rows[host] = r
+ return nil
+}
+
+func (f *fakeRoutes) put(r registry.Route) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.rows[r.Host] = r
+}
+
+func (f *fakeRoutes) get(host string) registry.Route {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.rows[host]
+}
+
+type eventLog struct {
+ mu sync.Mutex
+ lines []string
+}
+
+func (e *eventLog) App(string, string, string, string) {}
+
+func (e *eventLog) Event(_, _, host, _, msg string) {
+ e.mu.Lock()
+ defer e.mu.Unlock()
+ e.lines = append(e.lines, host+": "+msg)
+}
+
+func (e *eventLog) has(substr string) bool {
+ e.mu.Lock()
+ defer e.mu.Unlock()
+ for _, l := range e.lines {
+ if strings.Contains(l, substr) {
+ return true
+ }
+ }
+ return false
+}
+
+// harness is a manager over the fakes, for app "blog" of account "bab",
+// every host awake on v1, its good build.
+type harness struct {
+ t *testing.T
+ act *fakeAct
+ rows *fakeRoutes
+ book *FailureBook
+ m *Manager
+ opt, data string
+ mu sync.Mutex
+ now time.Time
+ concurrency int
+ swap string
+ demands map[string]string
+ mismatch map[string]bool
+ demandErr map[string]error
+ noDigest map[string]bool
+ pointers map[string]release.ChannelPointer // channel -> its signed pointer
+ failing atomic.Bool
+ touched atomic.Int32
+ events *eventLog
+ // pointerHook, when set, runs at the start of every pointer read.
+ pointerHook func()
+}
+
+func newHarness(t *testing.T, hosts ...string) *harness {
+ t.Helper()
+ h := &harness{t: t, act: newFakeAct(), rows: &fakeRoutes{rows: map[string]registry.Route{}}, book: &FailureBook{},
+ opt: t.TempDir(), data: t.TempDir(), now: t0, concurrency: 1,
+ demands: map[string]string{}, mismatch: map[string]bool{}, demandErr: map[string]error{}, noDigest: map[string]bool{},
+ pointers: map[string]release.ChannelPointer{},
+ events: &eventLog{}}
+ h.stage("v1")
+ for _, host := range hosts {
+ h.rows.put(registry.Route{Host: host, Kind: "instance", App: "blog", Account: "bab", Status: "ready",
+ Hibernate: true, Version: "v1", GoodVersion: "v1", GoodDigest: "sha-v1"})
+ h.act.serve(host, "v1")
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ t.Cleanup(cancel)
+ h.m = NewManager(ctx, Deps{
+ Act: h.act, Routes: h.rows, Book: h.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) {},
+ })
+ return h
+}
+
+// pointer signs channel's pointer onto ver at promoted, as `carlos promote`
+// (with or without --retry) would.
+func (h *harness) pointer(channel, ver string, promoted time.Time) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ h.pointers[channel] = release.ChannelPointer{Schema: 1, Account: "bab", App: "blog", Channel: channel, Version: ver, PromotedAt: promoted}
+}
+
+func (h *harness) pointerFor(_ context.Context, _, _, channel string) (*release.ChannelPointer, error) {
+ h.mu.Lock()
+ hook := h.pointerHook
+ h.mu.Unlock()
+ if hook != nil {
+ hook()
+ }
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ p, ok := h.pointers[channel]
+ if !ok {
+ return nil, nil
+ }
+ return &p, nil
+}
+
+func (h *harness) clock() time.Time {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ return h.now
+}
+
+func (h *harness) advance(d time.Duration) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ h.now = h.now.Add(d)
+}
+
+func (h *harness) stage(vers ...string) {
+ for _, v := range vers {
+ if err := release.StageArtifact(h.opt, "bab", v, "blog", []byte(v)); err != nil {
+ h.t.Fatal(err)
+ }
+ }
+}
+
+// demand is release.CanonicalDemand's contract: a mismatch wraps
+// release.ErrStagedDigestMismatch (a staging refusal); demandErr is any other
+// error, and noDigest is a store that could not compare, both of which must
+// end as a hold attempt.
+func (h *harness) demand(_ context.Context, _, _, ver, _ string) (string, string, error) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ if h.mismatch[ver] {
+ return "", "", fmt.Errorf("%w: the staged executable's digest differs from the canonical manifest", release.ErrStagedDigestMismatch)
+ }
+ if err := h.demandErr[ver]; err != nil {
+ return "", "", err
+ }
+ if h.noDigest[ver] {
+ return "", "", nil
+ }
+ return h.demands[ver], "sha-" + ver, nil
+}
+
+func (h *harness) policy(context.Context, string, string) (*appconfig.Manifest, error) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ return &appconfig.Manifest{Schema: 1, App: "blog", Swap: h.swap, SwapConcurrency: h.concurrency}, nil
+}
+
+func (h *harness) install(ver string, hosts ...string) {
+ for _, host := range hosts {
+ r := h.rows.get(host)
+ r.Version = ver
+ h.rows.put(r)
+ }
+}
+
+func (h *harness) pass() {
+ rs, _ := h.rows.List()
+ h.m.Pass(rs)
+}
+
+func (h *harness) servingAll(build string, hosts ...string) bool {
+ for _, host := range hosts {
+ if g, ok := h.act.Current(host); !ok || g.Build != build {
+ return false
+ }
+ }
+ return true
+}
+
+func waitFor(t *testing.T, what string, cond func() bool) {
+ t.Helper()
+ deadline := time.Now().Add(5 * time.Second)
+ for time.Now().Before(deadline) {
+ if cond() {
+ return
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ t.Fatalf("timed out waiting for %s", what)
+}
diff --git a/internal/rollout/manager.go b/internal/rollout/manager.go
new file mode 100644
index 000000000..11b4165ba
--- /dev/null
+++ b/internal/rollout/manager.go
@@ -0,0 +1,808 @@
+package rollout
+
+import (
+ "context"
+ "errors"
+ "log"
+ "os"
+ "strings"
+ "sync"
+ "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"
+)
+
+// Activator is the part of *activator.Service the scheduler drives. Swap is
+// plan 3's, for instances; ReplaceWorker is Task 11's same-entry
+// stop-then-start for workers and native sidecar members (spec §7.9).
+type Activator interface {
+ Current(host string) (activator.Gen, bool)
+ Owns(host string) bool
+ Swap(ctx context.Context, a activator.Attempt, newest func(host string) uint64) activator.SwapResult
+ ReplaceWorker(ctx context.Context, a activator.Attempt, fallback *release.StagedIdentity, newest func(host string) uint64) activator.SwapResult
+}
+
+// Routes is the part of *registry.Registry the scheduler reads and writes.
+type Routes interface {
+ Get(host string) (*registry.Route, error)
+ SetGoodBuild(host string, expect, next registry.GoodBuild) error
+ SetServingVersion(host, ver string) error
+}
+
+// Deps is everything the manager needs from the box. Its funcs are always
+// called without the manager's locks held.
+type Deps struct {
+ Act Activator
+ Routes Routes
+ Book *FailureBook
+ OptDir string
+ DataDir string
+ // Box is this box's label (host.BoxLabel), named by the cohort-halted
+ // event (spec §9.4: app, box, build, reason).
+ Box string
+ // Demand is release.CanonicalDemand bound to the deploy store and this
+ // box's platform (spec §4.2).
+ Demand func(ctx context.Context, acct, app, version, stagedPath string) (demand, digest string, err error)
+ // Policy reads the app's default config bundle (spec §4.1).
+ Policy func(ctx context.Context, acct, app string) (*appconfig.Manifest, error)
+ // Pointer reads a channel's signed pointer, verified (release.Channel).
+ // It is the retry signal (spec rev 14 §8.4) and the PromotedAt a failure
+ // record keeps. Nil: records clear only by a newer installed build. It
+ // is the raw signed read, not the per-box digest ratchet: only its
+ // Version and PromotedAt are used, to clear failure records, and root's
+ // install, which the ratchet guards, decides which build is installed.
+ Pointer func(ctx context.Context, acct, app, channel string) (*release.ChannelPointer, error)
+ Events activator.EventSink
+ // Touch asks root for a stage pass (the stage relay, spec §7.7).
+ Touch func()
+ // Materialize installs and stamps a cold managed host's build (the exec
+ // backend's Materialize hook) so a wake is planned against it (Task 10,
+ // spec §6.4, §8.4). Nil: a cold managed host's wake is refused.
+ Materialize func(ctx context.Context, host string) error
+ Now func() time.Time
+ Logf func(format string, args ...any)
+}
+
+// pointerReadEvery bounds how often an app with a standing failure record
+// has its channel pointers read for a retry. An app with no record costs no
+// read at all.
+const pointerReadEvery = 30 * time.Second
+
+// stagingDeferral is how long an attempt refused as a staging error waits:
+// one stage-relay window, for root to restage (spec §4.2, §7.7).
+const stagingDeferral = 15 * time.Second
+
+// abandonDeferral is how long a host waits after an attempt was abandoned
+// while it was still the host's newest: a hold whose stop withdrew, a lost
+// lease, a re-check that refused. Running the same attempt again at once
+// would only repeat the stop cycle (plan 3's deferral to this plan). An
+// attempt abandoned because a newer desire superseded it does not wait: the
+// newer attempt goes as soon as it is admitted.
+const abandonDeferral = 15 * time.Second
+
+// Manager is the box's rollout scheduler: one goroutine per (account, app),
+// fed by the agent pass, the adopt pass, `carlos restart` and config sync.
+//
+// Lock order: activator.Service.mu, then Manager.mu, then scheduler.mu, then
+// FailureBook.mu. The activator (Current, Owns, Swap, ReplaceWorker) is never
+// called while Manager.mu or a scheduler.mu is held.
+type Manager struct {
+ d Deps
+ ctx context.Context
+ policies *policyCache
+
+ mu sync.Mutex
+ apps map[appKey]*scheduler
+ hostApp map[string]appKey
+ 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
+
+ // goodMu serialises every serving- and good-build write with its
+ // acknowledgement check (served, publishServing, writeGood), so an older
+ // generation's write can never land after a newer one's (spec §8.5).
+ goodMu sync.Mutex
+}
+
+// NewManager builds the manager. Its schedulers run until ctx ends.
+func NewManager(ctx context.Context, d Deps) *Manager {
+ if d.Now == nil {
+ d.Now = time.Now
+ }
+ if d.Logf == nil {
+ d.Logf = log.Printf
+ }
+ 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{}}
+ m.policies = &policyCache{load: d.Policy, ttl: 30 * time.Second, now: d.Now, m: map[appKey]policyEntry{}}
+ return m
+}
+
+func (m *Manager) scheduler(k appKey) *scheduler {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ s := m.apps[k]
+ if s == nil {
+ book := m.d.Book
+ s = &scheduler{m: m, key: k, kick: make(chan struct{}, 1),
+ q: newAppQueue(k.acct, k.app, func(build, host string, served bool) bool {
+ return book.Blocks(k.acct, k.app, build, host, served)
+ })}
+ m.apps[k] = s
+ go s.loop(m.ctx)
+ }
+ return s
+}
+
+func (m *Manager) schedulerOf(host string) *scheduler {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ k, ok := m.hostApp[host]
+ if !ok {
+ return nil
+ }
+ return m.apps[k]
+}
+
+// track is host's config track; the caller holds m.mu.
+func (m *Manager) track(host string) *configTrack {
+ t := m.cfg[host]
+ if t == nil {
+ t = &configTrack{}
+ m.cfg[host] = t
+ }
+ return t
+}
+
+// Pass is the agent pass of spec §6.1: each managed route's installed build,
+// stamped by root, against the build its current generation serves. routes
+// must be the box's whole list, because Pass also prunes failure records no
+// route installs any more, and lifts the halts those records held.
+func (m *Manager) Pass(routes []registry.Route) {
+ keys := m.observe(routes)
+ m.applyPointers(routes, keys)
+ m.prune(routes)
+ m.liftClearedHalts()
+}
+
+// Nudge re-reads one host and wakes its scheduler. The activator sweep's
+// drift arm calls it (spec §10).
+func (m *Manager) Nudge(host string) {
+ rt, err := m.d.Routes.Get(host)
+ if err != nil || rt == nil {
+ return
+ }
+ m.observe([]registry.Route{*rt})
+}
+
+// Adopted replaces BounceAdopted as the adopt pass's restart callback when
+// the flag is on: the pass has repointed the live symlink, and the scheduler
+// takes the move from here, paced. It reports whether an awake instance will
+// be attempted, so makeRestartRoute does not fall through to systemctl.
+func (m *Manager) Adopted(host string) bool {
+ if _, awake := m.d.Act.Current(host); !awake {
+ return false
+ }
+ m.Nudge(host)
+ return true
+}
+
+func (m *Manager) observe(routes []registry.Route) []appKey {
+ type seen struct {
+ rt registry.Route
+ serving string
+ known bool // false while waking or stopping: the slot is left alone
+ }
+ byApp := map[appKey][]seen{}
+ for _, rt := range routes {
+ if !Managed(rt) {
+ continue
+ }
+ s := seen{rt: rt, known: true}
+ if g, ok := m.d.Act.Current(rt.Host); ok && g.Build != "" {
+ s.serving = g.Build
+ } else if m.d.Act.Owns(rt.Host) {
+ s.known = false
+ }
+ k := appKey{release.RouteAccount(rt), rt.App}
+ byApp[k] = append(byApp[k], s)
+ }
+ now := m.d.Now()
+ keys := make([]appKey, 0, len(byApp))
+ for k, list := range byApp {
+ keys = append(keys, k)
+ policy := m.policies.get(m.ctx, k.acct, k.app)
+ s := m.scheduler(k)
+ retried := map[string]string{}
+ m.mu.Lock()
+ for _, x := range list {
+ m.hostApp[x.rt.Host] = k
+ if b, ok := m.retried[x.rt.Host]; ok {
+ retried[x.rt.Host] = b
+ }
+ }
+ m.mu.Unlock()
+ s.mu.Lock()
+ s.q.concurrency = policy.Concurrency
+ for _, x := range list {
+ if x.known {
+ // Serving state only from a stable entry. The desired build
+ // and its revision are updated either way (spec §6.3): a
+ // build root installs while a hold drains or a wake runs
+ // supersedes the attempt's, so its gates (pinCheck, the
+ // result check) refuse the older build. Admission still
+ // waits for a stable entry (admit's view leaves a waking or
+ // stopping host out).
+ s.q.observe(x.rt.Host, x.serving, x.rt.GoodVersion, now)
+ }
+ cause := activator.CauseRelease
+ if retried[x.rt.Host] == x.rt.Version {
+ cause = activator.CauseRetry
+ }
+ s.q.wantBuild(x.rt.Host, cause, x.rt.Version, now)
+ }
+ s.mu.Unlock()
+ s.poke()
+ }
+ return keys
+}
+
+// routeChannel is the channel a route follows: "" means stable, as
+// release's own (unexported) routeChannel reads it.
+func routeChannel(rt registry.Route) string {
+ if rt.Channel == "" {
+ return "stable"
+ }
+ return rt.Channel
+}
+
+// applyPointers is the retry signal of spec rev 14 §8.4. For each app with a
+// standing failure record, at most every pointerReadEvery, it reads the
+// signed pointer of every channel the app's managed routes on this box
+// follow, and clears a record when a pointer names its build with a newer
+// PromotedAt (`carlos promote --retry`). The pointer the build failed under,
+// read again, clears nothing. A second retry of the same build is the same
+// rule again: the retry's own failure recorded the newer PromotedAt, so only
+// a pointer signed after it clears the record.
+func (m *Manager) applyPointers(routes []registry.Route, keys []appKey) {
+ if m.d.Pointer == nil {
+ return
+ }
+ now := m.d.Now()
+ for _, k := range keys {
+ recs := m.d.Book.ForApp(k.acct, k.app)
+ if len(recs) == 0 {
+ continue
+ }
+ m.mu.Lock()
+ due := now.Sub(m.pointerRead[k]) >= pointerReadEvery
+ if due {
+ m.pointerRead[k] = now
+ }
+ m.mu.Unlock()
+ if !due {
+ continue
+ }
+ channels := map[string]bool{}
+ for _, rt := range routes {
+ if Managed(rt) && release.RouteAccount(rt) == k.acct && rt.App == k.app {
+ channels[routeChannel(rt)] = true
+ }
+ }
+ for ch := range channels {
+ ptr, err := m.d.Pointer(m.ctx, k.acct, k.app, ch)
+ if err != nil {
+ m.d.Logf("rollout: %s/%s channel %s: %v (retry not checked this pass)", k.acct, k.app, ch, err)
+ continue
+ }
+ if ptr == nil {
+ continue
+ }
+ for _, r := range recs {
+ if ptr.Version != r.Build {
+ continue
+ }
+ ok, err := m.d.Book.ClearOnPointer(k.acct, k.app, r.Build, ptr.PromotedAt)
+ if err != nil {
+ m.d.Logf("rollout: %s/%s clear record for %s: %v (cleared in memory)", k.acct, k.app, r.Build, err)
+ }
+ if ok {
+ m.retryBuild(k, r.Build, "")
+ }
+ }
+ }
+ }
+}
+
+// retryBuild follows a cleared record: the next attempt onto build carries
+// cause retry and runs as a fresh canary. host == "" is an app-wide retry,
+// which reopens the halted cohort: its next attempt is a canary whatever
+// wantBuild reports, because unhalt starts the cohort again. A host names
+// `carlos instances retry`, whose attempt is that host's own canary;
+// RetryHost has already bound the host's retry flag, in the step that
+// cleared the book, and the flag is not set again here, where a forced
+// attempt may already have started and spent it.
+func (m *Manager) retryBuild(k appKey, build, host string) {
+ s := m.scheduler(k)
+ now := m.d.Now()
+ s.mu.Lock()
+ var hosts []string
+ for name, h := range s.q.hosts {
+ if h.wantBuild == build && (host == "" || name == host) {
+ hosts = append(hosts, name)
+ s.q.wantBuild(name, activator.CauseRetry, build, now)
+ }
+ }
+ if host == "" {
+ s.q.unhalt(build)
+ }
+ s.mu.Unlock()
+ m.mu.Lock()
+ for _, name := range hosts {
+ m.retried[name] = build
+ }
+ m.mu.Unlock()
+ m.d.Logf("rollout: %s/%s build %s retried (%v)", k.acct, k.app, build, hosts)
+ s.poke()
+}
+
+// RetryHost is the box half of `carlos instances retry <host>` (spec §8.4):
+// the record holding host's installed build is cleared for host alone, and
+// its next attempt is a canary. How the request reaches the box is plan 6's.
+//
+// The book entry is cleared and the host's retry flag bound in one hold of
+// the scheduler's lock (scheduler.mu before FailureBook.mu), the book first.
+// Admission cannot run between the two, so the adoption onto the cleared
+// build is the attempt the flag lets through, as a canary, and a restart or
+// config attempt cannot start in between and spend the flag.
+func (m *Manager) RetryHost(host string) bool {
+ m.mu.Lock()
+ k, ok := m.hostApp[host]
+ m.mu.Unlock()
+ if !ok {
+ return false
+ }
+ s := m.scheduler(k)
+ s.mu.Lock()
+ build := ""
+ if h := s.q.hosts[host]; h != nil {
+ build = h.wantBuild
+ }
+ var (
+ cleared bool
+ err error
+ )
+ if build != "" {
+ cleared, err = m.d.Book.RetryHost(k.acct, k.app, build, host)
+ }
+ if cleared {
+ s.q.wantBuild(host, activator.CauseRetry, build, m.d.Now())
+ s.q.retryHost(host)
+ }
+ s.mu.Unlock()
+ if err != nil {
+ m.d.Logf("rollout: retry %s on %s: %v (cleared in memory)", build, host, err)
+ }
+ if cleared {
+ m.retryBuild(k, build, host)
+ }
+ return cleared
+}
+
+// failedUnder is the PromotedAt a failure record keeps (spec rev 14 §8.4):
+// the pointer that named the build when it failed. If that pointer cannot be
+// read, or already names another build, it is zero, and the failure's own
+// time stands in when the record is written: only a pointer signed after the
+// failure can then clear the record, which is the safe direction.
+func (m *Manager) failedUnder(rt registry.Route, build string) time.Time {
+ if m.d.Pointer != nil {
+ ptr, err := m.d.Pointer(m.ctx, release.RouteAccount(rt), rt.App, routeChannel(rt))
+ if err == nil && ptr != nil && ptr.Version == build {
+ return ptr.PromotedAt
+ }
+ }
+ return time.Time{}
+}
+
+// prune drops failure records for builds no managed route installs. A
+// held-down instance's route still names its failed build, so its record
+// stays. An empty route list is never authority to drop anything:
+// registry.Open creates an empty database on a wrong path. The installed
+// callback runs under the book's lock and reads only this map.
+func (m *Manager) prune(routes []registry.Route) {
+ if len(routes) == 0 {
+ return
+ }
+ installed := map[[3]string]bool{}
+ for _, rt := range routes {
+ if Managed(rt) && rt.Version != "" {
+ installed[[3]string{release.RouteAccount(rt), rt.App, rt.Version}] = true
+ }
+ }
+ if err := m.d.Book.Prune(func(acct, app, build string) bool { return installed[[3]string{acct, app, build}] }); err != nil {
+ m.d.Logf("rollout: prune failure records: %v", err)
+ }
+}
+
+// liftClearedHalts reopens every halted adoption cohort whose failure record
+// no longer stands (spec §8.2, §8.4: a newer build, a retry or an install by
+// hand lifts the hold). A halt lives in memory beside its record, so a record
+// pruned after an install by hand would otherwise leave the build refused for
+// good if it were installed again. failCohort writes the record and the halt
+// under the same scheduler lock this reads them under.
+func (m *Manager) liftClearedHalts() {
+ m.mu.Lock()
+ scheds := make([]*scheduler, 0, len(m.apps))
+ for _, s := range m.apps {
+ scheds = append(scheds, s)
+ }
+ m.mu.Unlock()
+ for _, s := range scheds {
+ var lifted []string
+ s.mu.Lock()
+ for key, c := range s.q.cohorts {
+ build, adopt := strings.CutPrefix(key, "adopt:")
+ if !adopt || !c.halted {
+ continue
+ }
+ if _, held := m.d.Book.Lookup(s.key.acct, s.key.app, build); !held {
+ s.q.unhalt(build)
+ lifted = append(lifted, build)
+ }
+ }
+ s.mu.Unlock()
+ if len(lifted) > 0 {
+ m.d.Logf("rollout: %s/%s: halt lifted for %v, their failure records are gone", s.key.acct, s.key.app, lifted)
+ s.poke()
+ }
+ }
+}
+
+// Bootstrap is spec §8.5's first start: every awake instance records the
+// build it serves as good. Nothing is inferred from Version or PrevVersion.
+func (m *Manager) Bootstrap(routes []registry.Route) {
+ for _, rt := range routes {
+ if !Managed(rt) || rt.GoodVersion != "" {
+ continue
+ }
+ if g, ok := m.d.Act.Current(rt.Host); ok && g.Build != "" {
+ m.writeGood(rt.Host, g)
+ }
+ }
+}
+
+// Restart is a `carlos restart` request for host (spec §10). id is the
+// request's identity: the restart generation's etag, or a narrowed request's
+// id. A sleeping instance is never woken. One that is waking, stopping or
+// inside its first-wake observation keeps the request, which runs once it
+// is stable (spec §6.2).
+func (m *Manager) Restart(host, id string) bool {
+ s := m.schedulerOf(host)
+ if s == nil {
+ return false
+ }
+ if !m.d.Act.Owns(host) {
+ return false
+ }
+ s.mu.Lock()
+ changed := s.q.wantRestart(host, id, m.d.Now())
+ s.mu.Unlock()
+ if changed {
+ s.poke()
+ }
+ return changed
+}
+
+// ConfigDelivered is config sync's report that host's config file now holds
+// revision rev (spec §8.1). An instance with a process (awake, waking or
+// stopping) gets a pending config attempt; if its wake turns out to have read
+// rev, the wake's activation clears the attempt before it runs. A sleeping
+// one needs none, because its next wake starts with the file: rev counts as
+// what that wake will run, and config sync's ack goes out at once with
+// today's hibernating acknowledgement (configTrack.deferAck). A host this
+// manager does not manage is not tracked, so its ack goes out as before.
+func (m *Manager) ConfigDelivered(host, rev string) {
+ s := m.schedulerOf(host)
+ if s == nil {
+ return
+ }
+ // Asked before m.mu: the activator is never called under the manager's
+ // lock (the lock order on Manager).
+ awake := m.d.Act.Owns(host)
+ m.mu.Lock()
+ t := m.track(host)
+ t.delivered, t.fails, t.lastErr = rev, 0, ""
+ t.sleeping = !awake
+ if !awake {
+ t.active = rev
+ }
+ m.mu.Unlock()
+ if !awake {
+ return
+ }
+ s.mu.Lock()
+ changed := s.q.wantConfig(host, rev, m.d.Now())
+ s.mu.Unlock()
+ if changed {
+ s.poke()
+ }
+}
+
+// DeferAck wraps config sync's ack writer: a revision is acknowledged only
+// once it is active (spec §8.1).
+func (m *Manager) DeferAck(write func(appconfig.ConfigAck)) func(appconfig.ConfigAck) {
+ return func(a appconfig.ConfigAck) {
+ m.mu.Lock()
+ now := m.track(a.Host).deferAck(a, write)
+ m.mu.Unlock()
+ if now != nil {
+ now()
+ }
+ }
+}
+
+// resolve turns a dispatch into the Attempt Swap runs: its artifact and mode.
+// Recovery onto the serving build uses the generation's captured identity,
+// and a move back onto the good build uses the retained one; neither is
+// checked against the canonical manifest (spec §4.2). An adoption checks the
+// staged file exists and then the canonical demand. A missing staged file,
+// or a staged file that is not the release's canonical artifact
+// (release.ErrStagedDigestMismatch), is a staging refusal, never a build
+// failure. Any other trouble reading the demand runs the attempt in hold:
+// missing or unverifiable policy data never permits an overlap.
+func (m *Manager) resolve(ctx context.Context, d dispatch, rt registry.Route) (activator.Attempt, error) {
+ acct, app := release.RouteAccount(rt), rt.App
+ policy := m.policies.get(ctx, acct, app)
+ a := activator.Attempt{Host: d.ID.Host, Cause: d.ID.Cause, Build: d.ID.Build, ConfigRev: d.ID.ConfigRev,
+ RestartID: d.ID.RestartID, SchedRev: d.SchedRev, Canary: d.Canary}
+ if g, ok := m.d.Act.Current(rt.Host); ok && g.Build == d.ID.Build {
+ a.StagedPath, a.Digest, a.Demand = g.StagedPath, g.Digest, g.Demand
+ a.Mode = RecoveryMode(policy, g.Demand)
+ return a, nil
+ }
+ path := release.VersionPath(m.d.OptDir, acct, d.ID.Build, app)
+ if d.ID.Build == rt.GoodVersion && rt.GoodDigest != "" {
+ a.StagedPath, a.Digest, a.Demand = path, rt.GoodDigest, rt.GoodDemand
+ a.Mode = RecoveryMode(policy, rt.GoodDemand)
+ return a, nil
+ }
+ if _, err := os.Stat(path); err != nil {
+ return a, err
+ }
+ demand, digest, err := m.d.Demand(ctx, acct, app, d.ID.Build, path)
+ if errors.Is(err, release.ErrStagedDigestMismatch) {
+ return a, 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)
+ 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
+ // here, because plan 3's verifyStaged refuses an empty digest.
+ demand = release.DemandHold
+ if digest, err = release.FileSHA256(path); err != nil {
+ return a, err
+ }
+ }
+ a.StagedPath, a.Digest, a.Demand = path, digest, demand
+ a.Mode = AttemptMode(policy, demand)
+ return a, nil
+}
+
+// served records a generation that became current after its checks: the
+// serving build, and the good build (spec §8.5). Both only while g is the
+// generation the activator serves (acknowledged).
+func (m *Manager) served(host string, g activator.Gen) {
+ if g.Build == "" {
+ return
+ }
+ m.goodMu.Lock()
+ defer m.goodMu.Unlock()
+ if !m.acknowledged(host, g) {
+ m.d.Logf("rollout: %s: generation %d (%s) is no longer served; not recording it", host, g.N, g.Build)
+ return
+ }
+ m.setServingLocked(host, g)
+ m.writeGoodLocked(host, g)
+}
+
+// setServingLocked writes an instance's serving build. A worker's is the
+// activator's alone (Task 11, publishWorkerServing, ordered by incarnation),
+// so two writers never race on one row. The caller holds m.goodMu.
+func (m *Manager) setServingLocked(host string, g activator.Gen) {
+ if g.Worker {
+ return
+ }
+ if err := m.d.Routes.SetServingVersion(host, g.Build); err != nil {
+ m.d.Logf("rollout: %s: record serving build %s: %v", host, g.Build, err)
+ }
+}
+
+// publishServing records the serving build alone (a healthy first wake, a
+// fallback), on the same terms as served.
+func (m *Manager) publishServing(host string, g activator.Gen) {
+ if g.Build == "" {
+ return
+ }
+ m.goodMu.Lock()
+ defer m.goodMu.Unlock()
+ if !m.acknowledged(host, g) {
+ return
+ }
+ m.setServingLocked(host, g)
+}
+
+// acknowledged reports whether g is the generation the activator serves
+// for host right now (spec §8.5: the good build is written expecting the
+// current one). A late completion or event of an older generation is not,
+// so it can never overwrite what a newer one wrote. Generation numbers
+// decide when both sides have one; a worker's (no number) is compared by
+// its staged artifact.
+func (m *Manager) acknowledged(host string, g activator.Gen) bool {
+ cur, ok := m.d.Act.Current(host)
+ if !ok {
+ return false
+ }
+ if cur.N != 0 && g.N != 0 {
+ return cur.N == g.N
+ }
+ return cur.StagedPath == g.StagedPath && cur.Digest == g.Digest
+}
+
+// writeGood is writeGoodLocked for a caller outside served: a settled first
+// wake, a healthy planned wake. Same terms: g must be acknowledged.
+func (m *Manager) writeGood(host string, g activator.Gen) {
+ m.goodMu.Lock()
+ defer m.goodMu.Unlock()
+ if !m.acknowledged(host, g) {
+ return
+ }
+ m.writeGoodLocked(host, g)
+}
+
+// writeGoodLocked replaces host's good build with g's identity,
+// conditionally on the row still holding what was just read. goodMu
+// serialises every writer and each has just proven g current, so a retry
+// only covers a write from outside this manager; a second loss is logged
+// and the next serve tries again. The caller holds m.goodMu.
+func (m *Manager) writeGoodLocked(host string, g activator.Gen) {
+ if g.Build == "" || g.Digest == "" {
+ return
+ }
+ next := registry.GoodBuild{Version: g.Build, Digest: g.Digest, Demand: g.Demand}
+ for range 2 {
+ cur, err := m.d.Routes.Get(host)
+ if err != nil || cur == nil {
+ return
+ }
+ if cur.Good() == next {
+ m.observeGood(host, next.Version)
+ return
+ }
+ err = m.d.Routes.SetGoodBuild(host, cur.Good(), next)
+ if err == nil {
+ m.observeGood(host, next.Version)
+ return
+ }
+ if !errors.Is(err, registry.ErrGoodBuildChanged) {
+ m.d.Logf("rollout: %s: record good build %s: %v", host, g.Build, err)
+ return
+ }
+ }
+ m.d.Logf("rollout: %s: good build moved twice under this write, leaving it", host)
+}
+
+func (m *Manager) observeGood(host, ver string) {
+ s := m.schedulerOf(host)
+ if s == nil {
+ return
+ }
+ s.mu.Lock()
+ if h := s.q.hosts[host]; h != nil {
+ h.good = ver
+ }
+ s.mu.Unlock()
+}
+
+func (m *Manager) configActivated(host, rev string) {
+ if rev == "" {
+ return
+ }
+ m.mu.Lock()
+ write, _ := m.track(host).activate(rev)
+ m.mu.Unlock()
+ if s := m.schedulerOf(host); s != nil {
+ s.mu.Lock()
+ s.q.configActive(host, rev)
+ s.mu.Unlock()
+ }
+ if write != nil {
+ write()
+ }
+}
+
+// configSlept is configActivated for an instance seen asleep with rev
+// pending (spec §8.1): its next wake starts with the delivered file, so rev
+// is what that wake runs and no config attempt is left to make. A held ack
+// goes out as a sleeper's does at delivery, marked Hibernating. rev was read
+// before the activator reported the instance asleep, so the file already
+// held it when the instance stopped: this can only under-claim.
+func (m *Manager) configSlept(host, rev string) {
+ m.mu.Lock()
+ t := m.track(host)
+ if t.ack != nil && t.ackRev == rev {
+ t.ack.Hibernating, t.ack.Restarted = true, false
+ }
+ write, _ := t.activate(rev)
+ m.mu.Unlock()
+ if s := m.schedulerOf(host); s != nil {
+ s.mu.Lock()
+ s.q.configActive(host, rev)
+ s.mu.Unlock()
+ }
+ if write != nil {
+ write()
+ }
+}
+
+// configFailure counts a failed attempt that carried config revision rev
+// and returns its backoff. Only a failure of the delivered revision counts:
+// one of an older revision says nothing about the newer one's attempts. The
+// caller sets the queue's backoff (appQueue.configFailed) in the step that
+// frees the attempt's slot.
+func (m *Manager) configFailure(host, rev, reason string) (time.Duration, bool) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ t := m.track(host)
+ if rev == "" || rev != t.delivered {
+ return 0, false
+ }
+ t.fails++
+ t.lastErr = reason
+ return configBackoff(t.fails), true
+}
+
+// cohortFailed runs the cohort-failure transition for s's app.
+func (m *Manager) cohortFailed(s *scheduler, rt registry.Route, rec FailureRecord) {
+ s.mu.Lock()
+ added, superseded, err := failCohort(m.d.Book, s.q, rec)
+ s.mu.Unlock()
+ if err != nil {
+ m.d.Logf("rollout: write failure record %s/%s %s: %v (held in memory)", rec.Account, rec.App, rec.Build, err)
+ }
+ if len(superseded) > 0 {
+ m.d.Logf("rollout: %s/%s build %s failed: superseding attempts past spawn on %v", rec.Account, rec.App, rec.Build, superseded)
+ }
+ if added {
+ m.d.Logf("rollout: %s/%s build %s failed on %s: %s", rec.Account, rec.App, rec.Build, rec.Host, rec.Reason)
+ m.emit(rt, EvCohortHalted, "warn", textCohortHalted(rec.Build, m.d.Box, reasonWord(rec.Reason)))
+ }
+ s.poke()
+}
+
+func (m *Manager) stagingRefused(rt registry.Route, build string, err error) {
+ m.d.Logf("rollout: %s: build %s refused as a staging error: %v", rt.Host, build, err)
+ m.emit(rt, EvStagingRefused, "warn", textStagingRefused(build))
+ if m.d.Touch != nil {
+ m.d.Touch()
+ }
+}
+
+// emit logs an event and publishes it to the member's log. text is built
+// from build names and reason words only; RedactPaths stays as a second
+// guard against a box path in a build name.
+func (m *Manager) emit(rt registry.Route, code EventCode, level, text string) {
+ m.d.Logf("rollout: %s %s: %s", code, rt.Host, text)
+ if m.d.Events != nil {
+ m.d.Events.Event(release.RouteAccount(rt), rt.App, rt.Host, level, activator.RedactPaths(text))
+ }
+}
diff --git a/internal/rollout/manager_test.go b/internal/rollout/manager_test.go
new file mode 100644
index 000000000..6d4c4a166
--- /dev/null
+++ b/internal/rollout/manager_test.go
@@ -0,0 +1,882 @@
+package rollout
+
+import (
+ "errors"
+ "sync"
+ "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 TestAReleaseRollsOutCanaryFirstThenPaced(t *testing.T) {
+ h := newHarness(t, "a", "b", "c")
+ h.stage("v2")
+ h.install("v2", "a", "b", "c")
+ h.pass()
+ waitFor(t, "every host on v2", func() bool { return h.servingAll("v2", "a", "b", "c") })
+ calls, _, maxFlight := h.act.snapshot()
+ if len(calls) != 3 || !calls[0].a.Canary || calls[1].a.Canary {
+ t.Fatalf("calls = %+v, want the canary first and then two plain swaps", calls)
+ }
+ if maxFlight > 1 {
+ t.Fatalf("%d swaps ran at once with swap_concurrency 1", maxFlight)
+ }
+ for _, host := range []string{"a", "b", "c"} {
+ if r := h.rows.get(host); r.GoodVersion != "v2" || r.GoodDigest != "sha-v2" || r.ServingVersion != "v2" {
+ t.Fatalf("%s row = %+v, want good and serving v2", host, r)
+ }
+ }
+ if !h.events.has("swapped to build v2") {
+ t.Fatal("no swapped event")
+ }
+}
+
+// Review focus 1, the index's cross-plan item.
+func TestTwoReleasesTwentySecondsApartNeverSwitchToTheFirst(t *testing.T) {
+ h := newHarness(t, "a", "b", "c")
+ h.stage("v2", "v3")
+ gate := h.act.hold("a") // the canary onto v2 sits in its settle
+ h.install("v2", "a", "b", "c")
+ h.pass()
+ waitFor(t, "the v2 canary to start", func() bool { _, started, _ := h.act.snapshot(); return len(started) == 1 })
+ h.advance(20 * time.Second)
+ h.install("v3", "a", "b", "c")
+ h.pass()
+ close(gate)
+ waitFor(t, "every host on v3", func() bool { return h.servingAll("v3", "a", "b", "c") })
+ calls, _, maxFlight := h.act.snapshot()
+ for _, c := range calls {
+ if c.a.Build == "v2" && c.outcome == activator.SwapDone {
+ t.Fatalf("switched %s to the superseded v2", c.a.Host)
+ }
+ }
+ if calls[0].a.Build != "v2" || calls[0].outcome != activator.SwapAbandoned {
+ t.Fatalf("first call = %+v, want the v2 canary abandoned at its re-check", calls[0])
+ }
+ if len(h.book.List()) != 0 {
+ t.Fatal("an abandoned attempt is not a build failure")
+ }
+ if maxFlight > 1 {
+ t.Fatal("swap_concurrency 1 exceeded")
+ }
+}
+
+func TestAFailedCanaryHaltsTheCohort(t *testing.T) {
+ h := newHarness(t, "a", "b", "c")
+ h.stage("v2")
+ h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
+ if a.Build == "v2" {
+ return activator.SwapFailed
+ }
+ return activator.SwapDone
+ }
+ h.install("v2", "a", "b", "c")
+ h.pass()
+ waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
+ h.pass()
+ time.Sleep(50 * time.Millisecond)
+ if calls, _, _ := h.act.snapshot(); len(calls) != 1 {
+ t.Fatalf("%d attempts onto v2, want the halt at the first failure", len(calls))
+ }
+ if r := h.rows.get("a"); r.GoodVersion != "v1" {
+ t.Fatalf("good build moved to %q on a failure", r.GoodVersion)
+ }
+ if !h.events.has("rollout of build v2 halted") {
+ t.Fatal("no cohort-halted event")
+ }
+}
+
+// Spec §8.1: the failure transition (the record and the halt) happens in
+// the same step that frees the failed attempt's slot. The hook looks at the
+// queue at every moment done() reads a row after the failure: whenever the
+// failed attempt's slot is free, the record must already be in the book
+// (which wake plans read) and admission must start nothing onto the build.
+// Halting in that step but writing the record after the row read let a
+// wake select the failed build in between.
+func TestAFailureIsRecordedAndHaltedInTheStepThatFreesItsSlot(t *testing.T) {
+ h := newHarness(t, "a", "b")
+ h.stage("v2")
+ var failed atomic.Bool
+ h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
+ if a.Build != "v2" {
+ return activator.SwapDone
+ }
+ failed.Store(true)
+ return activator.SwapFailed
+ }
+ var bad atomic.Int32
+ h.rows.mu.Lock()
+ h.rows.getHook = func(host string) {
+ if !failed.Load() {
+ return
+ }
+ s := h.m.schedulerOf(host)
+ s.mu.Lock()
+ hs := s.q.hosts[host]
+ free := hs != nil && hs.running == nil
+ var ds []dispatch
+ if free {
+ ds = s.q.admit(currentView(s.q), h.clock())
+ }
+ s.mu.Unlock()
+ if !free {
+ return
+ }
+ if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {
+ bad.Add(1)
+ }
+ for _, d := range ds {
+ if d.ID.Build == "v2" {
+ bad.Add(1)
+ }
+ }
+ }
+ h.rows.mu.Unlock()
+ h.install("v2", "a", "b")
+ h.pass()
+ waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
+ h.pass() // reads the rows again, with the slot free
+ if n := bad.Load(); n != 0 {
+ t.Fatalf("%d moment(s) with the slot free and the record missing or a v2 attempt admissible", n)
+ }
+}
+
+// Spec §8.5: a successful attempt keeps its host's slot until its state
+// (serving and good build) is published, so no newer attempt can be
+// admitted and complete first and then be overwritten by this one. Freeing
+// the slot before the write let that happen.
+func TestASuccessKeepsItsSlotUntilItIsPublished(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ var freed atomic.Int32
+ h.rows.mu.Lock()
+ h.rows.setGoodHook = func(host string) {
+ s := h.m.schedulerOf(host)
+ s.mu.Lock()
+ if hs := s.q.hosts[host]; hs != nil && hs.running == nil {
+ freed.Add(1)
+ }
+ s.mu.Unlock()
+ }
+ h.rows.mu.Unlock()
+ h.install("v2", "a")
+ h.pass()
+ waitFor(t, "the good build moved to v2", func() bool { return h.rows.get("a").GoodVersion == "v2" })
+ if n := freed.Load(); n != 0 {
+ t.Fatalf("%d good-build write(s) made after the attempt's slot was freed", n)
+ }
+}
+
+// A failure is published with no blocking read between the failure and the
+// record (spec §8.1): the pointer the build failed under is read before the
+// attempt runs. Reading it in done(), before the record, left every sibling's
+// wake free to plan the failed build for as long as that read took.
+func TestNoPointerReadSitsBetweenAFailureAndItsRecord(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ var failed, late atomic.Bool
+ h.act.outcome = func(activator.Attempt) activator.SwapOutcome {
+ failed.Store(true)
+ return activator.SwapFailed
+ }
+ h.mu.Lock()
+ h.pointerHook = func() {
+ if failed.Load() {
+ if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {
+ late.Store(true)
+ }
+ }
+ }
+ h.mu.Unlock()
+ h.install("v2", "a")
+ h.pass()
+ waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
+ if late.Load() {
+ t.Fatal("a pointer read ran after the failure and before its record was published")
+ }
+}
+
+// Spec §6.3: a build installed while the entry is unstable (a hold
+// draining, a wake running) still supersedes the attempt in flight, so its
+// gates refuse the older build. Skipping wantBuild for such a host left the
+// revision on the older build, and a hold's pinned wake exec'd it. Nothing
+// is admitted onto the host until it is stable again.
+func TestAnInstallDuringAnUnstableEntryStillSupersedes(t *testing.T) {
+ h := newHarness(t, "a")
+ h.pass()
+ s := h.m.schedulerOf("a")
+ h.act.mu.Lock()
+ delete(h.act.gens, "a")
+ h.act.owned["a"] = true // stopping or waking: a process, no current generation
+ h.act.mu.Unlock()
+ s.mu.Lock()
+ before := s.q.newest("a")
+ s.mu.Unlock()
+ h.stage("v2")
+ h.install("v2", "a")
+ h.pass()
+ s.mu.Lock()
+ after := s.q.newest("a")
+ s.mu.Unlock()
+ if after == before {
+ t.Fatal("an install during an unstable entry left the attempt revision unchanged")
+ }
+ time.Sleep(50 * time.Millisecond)
+ if calls, _, _ := h.act.snapshot(); len(calls) != 0 {
+ t.Fatalf("attempts %+v admitted onto an unstable entry", calls)
+ }
+}
+
+// A row that cannot be read at the failure still gets its record: the
+// failure is the build's, not the row's (spec §8.1). Skipping the write on
+// a failed registry read left the failed build selectable by every wake.
+func TestAFailureIsRecordedWhenItsRowIsUnreadable(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ var failed atomic.Bool
+ h.act.outcome = func(activator.Attempt) activator.SwapOutcome {
+ failed.Store(true)
+ return activator.SwapFailed
+ }
+ h.rows.mu.Lock()
+ h.rows.getErr = func(string) error {
+ if failed.Load() {
+ return errors.New("registry: database is locked")
+ }
+ return nil
+ }
+ h.rows.mu.Unlock()
+ h.install("v2", "a")
+ h.pass()
+ waitFor(t, "a failure record despite the unreadable row", func() bool {
+ _, ok := h.book.Lookup("bab", "blog", "v2")
+ return ok
+ })
+}
+
+// Review focus 4.
+func TestStagingRefusalWritesNoFailureRecord(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.mismatch["v2"] = true
+ h.install("v2", "a")
+ h.pass()
+ waitFor(t, "the stage relay touch", func() bool { return h.touched.Load() > 0 })
+ if len(h.book.List()) != 0 {
+ t.Fatal("a staging error wrote a failure record against the build")
+ }
+ if calls, _, _ := h.act.snapshot(); len(calls) != 0 {
+ t.Fatal("a refused attempt reached Swap")
+ }
+ if g, _ := h.act.Current("a"); g.Build != "v1" {
+ t.Fatal("the instance left its serving build")
+ }
+ if !h.events.has("not staged correctly") {
+ t.Fatal("no staging event")
+ }
+}
+
+func TestAnUnstagedBuildIsAStagingRefusal(t *testing.T) {
+ h := newHarness(t, "a")
+ h.install("v9", "a") // never staged on this box
+ h.pass()
+ waitFor(t, "the stage relay touch", func() bool { return h.touched.Load() > 0 })
+ if len(h.book.List()) != 0 {
+ t.Fatal("a missing staged file wrote a failure record")
+ }
+}
+
+func TestRestartGoesAheadForAnInstanceServingAFailedBuild(t *testing.T) {
+ h := newHarness(t, "d")
+ h.stage("v2")
+ h.act.serve("d", "v2")
+ r := h.rows.get("d")
+ r.Version, r.GoodVersion, r.GoodDigest = "v2", "v2", "sha-v2"
+ h.rows.put(r)
+ h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "e", At: t0})
+ h.pass()
+ if !h.m.Restart("d", "gen:etag-1") {
+ t.Fatal("the restart was not queued")
+ }
+ waitFor(t, "the restart", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
+ calls, _, _ := h.act.snapshot()
+ if c := calls[0].a; c.Cause != activator.CauseRestart || c.Build != "v2" || c.RestartID != "gen:etag-1" {
+ t.Fatalf("attempt = %+v", c)
+ }
+}
+
+func TestRecoveryUsesTheRetainedIdentityAndTheCurrentPolicy(t *testing.T) {
+ h := newHarness(t, "a")
+ h.swap = "hold" // switched to hold after v1 passed under overlap
+ h.pass()
+ h.m.Restart("a", "gen:etag-1")
+ waitFor(t, "the restart", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
+ calls, _, _ := h.act.snapshot()
+ g, _ := h.act.Current("a")
+ if c := calls[0].a; c.Mode != activator.ModeHold || c.StagedPath != "/opt/v1" || c.Digest != "sha-v1" {
+ t.Fatalf("attempt = %+v, want hold, by the generation's captured path and digest (current gen %+v)", c, g)
+ }
+}
+
+func TestAReleaseOntoTheServingBuildMakesNoSwap(t *testing.T) {
+ h := newHarness(t, "a")
+ h.pass()
+ time.Sleep(50 * time.Millisecond)
+ if calls, _, _ := h.act.snapshot(); len(calls) != 0 {
+ t.Fatalf("calls = %+v", calls)
+ }
+}
+
+// Review focus 5.
+func TestConfigDeliveredMidSwapIsAckedOnlyWhenActive(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ gate := h.act.hold("a")
+ h.install("v2", "a")
+ h.pass()
+ waitFor(t, "the release swap to start", func() bool { _, s, _ := h.act.snapshot(); return len(s) == 1 })
+ // The ack is written from the scheduler's goroutine: counted atomically.
+ var acks atomic.Int32
+ ack := h.m.DeferAck(func(appconfig.ConfigAck) { acks.Add(1) })
+ h.m.ConfigDelivered("a", "sha256:c1")
+ ack(appconfig.ConfigAck{Host: "a"})
+ if acks.Load() != 0 {
+ t.Fatal("acknowledged a config no generation has started with")
+ }
+ close(gate)
+ waitFor(t, "the config attempt", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 2 })
+ calls, _, _ := h.act.snapshot()
+ if c := calls[1].a; c.Cause != activator.CauseConfig || c.ConfigRev != "sha256:c1" || c.Build != "v2" {
+ t.Fatalf("second attempt = %+v, want a config attempt onto v2", c)
+ }
+ waitFor(t, "the ack", func() bool { return acks.Load() == 1 })
+}
+
+// Spec §8.1: a successful config attempt's activation is published before
+// its slot frees, and with no registry read in front of it. Published after
+// the slot and the read, a newer config attempt could complete first and be
+// overwritten (active went back to the older revision), and an unreadable
+// row lost the activation, leaving delivered and active apart for good.
+func TestConfigActivationIsPublishedBeforeTheSlotFrees(t *testing.T) {
+ h := newHarness(t, "a")
+ h.pass()
+ var acks, freed atomic.Int32
+ ack := h.m.DeferAck(func(appconfig.ConfigAck) {
+ acks.Add(1)
+ s := h.m.schedulerOf("a")
+ s.mu.Lock()
+ if hs := s.q.hosts["a"]; hs != nil && hs.running == nil {
+ freed.Add(1)
+ }
+ s.mu.Unlock()
+ })
+ var swapped atomic.Bool
+ h.act.outcome = func(activator.Attempt) activator.SwapOutcome {
+ swapped.Store(true)
+ return activator.SwapDone
+ }
+ h.rows.mu.Lock()
+ h.rows.getErr = func(string) error {
+ if swapped.Load() {
+ return errors.New("registry: database is locked")
+ }
+ return nil
+ }
+ h.rows.mu.Unlock()
+ // The config attempt waits at the gate until the ack is held, so the ack
+ // is released by the activation and not written at once by a config sync
+ // call that came after it.
+ gate := h.act.hold("a")
+ h.m.ConfigDelivered("a", "sha256:c1")
+ ack(appconfig.ConfigAck{Host: "a"})
+ if acks.Load() != 0 {
+ t.Fatal("acknowledged a config no generation has started with")
+ }
+ close(gate)
+ waitFor(t, "the config activated and acknowledged", func() bool { return acks.Load() == 1 })
+ if n := freed.Load(); n != 0 {
+ t.Fatal("the activation was published after the attempt's slot was freed")
+ }
+}
+
+func TestAFailedConfigAttemptRetriesWithBackoff(t *testing.T) {
+ h := newHarness(t, "a")
+ h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
+ if a.Cause == activator.CauseConfig {
+ return activator.SwapFailed
+ }
+ return activator.SwapDone
+ }
+ h.pass()
+ h.m.ConfigDelivered("a", "sha256:c1")
+ waitFor(t, "the first config attempt", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
+ waitFor(t, "the pending event", func() bool { return h.events.has("config delivered, not active") })
+ h.advance(59 * time.Second)
+ h.pass()
+ time.Sleep(50 * time.Millisecond)
+ if calls, _, _ := h.act.snapshot(); len(calls) != 1 {
+ t.Fatal("retried inside the one-minute backoff")
+ }
+ h.advance(2 * time.Second)
+ h.pass()
+ waitFor(t, "the retry after the backoff", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 2 })
+ if len(h.book.List()) != 0 {
+ t.Fatal("a failed config attempt wrote a failure record")
+ }
+}
+
+func TestBootstrapRecordsAwakeInstancesOnly(t *testing.T) {
+ h := newHarness(t, "awake", "asleep")
+ for _, host := range []string{"awake", "asleep"} {
+ r := h.rows.get(host)
+ r.GoodVersion, r.GoodDigest = "", ""
+ h.rows.put(r)
+ }
+ h.act.mu.Lock()
+ delete(h.act.gens, "asleep")
+ h.act.mu.Unlock()
+ rs, _ := h.rows.List()
+ h.m.Bootstrap(rs)
+ if r := h.rows.get("awake"); r.Good() != (registry.GoodBuild{Version: "v1", Digest: "sha-v1"}) {
+ t.Fatalf("awake row good = %+v", r.Good())
+ }
+ if r := h.rows.get("asleep"); r.GoodVersion != "" {
+ t.Fatal("nothing may be inferred for a sleeping instance (spec §8.5)")
+ }
+}
+
+// `carlos promote --retry` re-signs the channel pointer with a fresh
+// PromotedAt. The record keeps the PromotedAt it failed under, survives that
+// same pointer being read again, and clears on the re-signed one.
+func TestAReSignedPointerClearsTheRecord(t *testing.T) {
+ h := newHarness(t, "a", "b")
+ h.stage("v2")
+ h.failing.Store(true)
+ h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
+ if a.Build == "v2" && h.failing.Load() {
+ return activator.SwapFailed
+ }
+ return activator.SwapDone
+ }
+ h.pointer("stable", "v2", promotedAt)
+ h.install("v2", "a", "b")
+ h.pass()
+ waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
+ if r, _ := h.book.Lookup("bab", "blog", "v2"); !r.PromotedAt.Equal(promotedAt) {
+ t.Fatalf("record keeps PromotedAt %v, want the pointer's %v", r.PromotedAt, promotedAt)
+ }
+ h.advance(pointerReadEvery)
+ h.pass()
+ if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok {
+ t.Fatal("the pointer the build failed under cleared its record")
+ }
+ h.failing.Store(false)
+ h.pointer("stable", "v2", promotedAt.Add(time.Minute))
+ h.advance(pointerReadEvery)
+ h.pass()
+ waitFor(t, "every host on v2", func() bool { return h.servingAll("v2", "a", "b") })
+ calls, _, _ := h.act.snapshot()
+ if c := calls[1].a; c.Cause != activator.CauseRetry || !c.Canary {
+ t.Fatalf("first attempt after the retry = %+v, want a retry canary", c)
+ }
+}
+
+// `carlos instances retry <host>`: that host alone, as a canary.
+func TestRetryHostClearsOneInstance(t *testing.T) {
+ h := newHarness(t, "a", "b")
+ h.stage("v2")
+ h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "c", PromotedAt: promotedAt, At: t0})
+ h.install("v2", "a", "b")
+ h.pass()
+ if !h.m.RetryHost("b") {
+ t.Fatal("RetryHost did not clear b")
+ }
+ waitFor(t, "b on v2", func() bool { return h.servingAll("v2", "b") })
+ if h.servingAll("v2", "a") {
+ t.Fatal("a host retry moved another host")
+ }
+}
+
+// Carried from Task 3: a canonical manifest that cannot be read, or a store
+// that cannot compare digests, must reach AttemptMode as a hold demand, never
+// as "". Only release.ErrStagedDigestMismatch is a staging refusal.
+func TestAnUnreadableCanonicalManifestRunsInHold(t *testing.T) {
+ for name, set := range map[string]func(h *harness){
+ "demand error": func(h *harness) { h.demandErr["v2"] = errors.New("store: GetObject: connection reset") },
+ "no digest": func(h *harness) { h.noDigest["v2"] = true },
+ } {
+ t.Run(name, func(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ set(h)
+ h.install("v2", "a")
+ h.pass()
+ waitFor(t, "the attempt", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
+ want, err := release.FileSHA256(release.VersionPath(h.opt, "bab", "v2", "blog"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ calls, _, _ := h.act.snapshot()
+ if c := calls[0].a; c.Mode != activator.ModeHold || c.Demand != release.DemandHold || c.Digest != want {
+ t.Fatalf("attempt = %+v, want hold on the staged file's own digest %s", c, want)
+ }
+ if h.touched.Load() != 0 || len(h.book.List()) != 0 {
+ t.Fatal("an unreadable manifest is neither a staging refusal nor a build failure")
+ }
+ })
+ }
+}
+
+// Carried from Task 4: a record keeps a real PromotedAt. With no pointer to
+// read, the failure's own time stands in, never a zero time.
+func TestAFailureWithNoReadablePointerKeepsItsOwnTime(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.act.outcome = func(activator.Attempt) activator.SwapOutcome { return activator.SwapFailed }
+ h.install("v2", "a")
+ h.pass()
+ waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
+ r, _ := h.book.Lookup("bab", "blog", "v2")
+ if r.PromotedAt.IsZero() || !r.PromotedAt.Equal(r.At) {
+ t.Fatalf("record %+v: want the failure's own time as PromotedAt", r)
+ }
+}
+
+// Carried from Task 5: a second `carlos promote --retry` of the same build,
+// after the retry failed too, clears the record again and the next attempt is
+// a canary again. Nothing may rest on wantBuild reporting a change.
+func TestASecondRetryOfTheSameBuildIsACanaryAgain(t *testing.T) {
+ h := newHarness(t, "a", "b")
+ h.stage("v2")
+ h.failing.Store(true)
+ h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
+ if a.Build == "v2" && h.failing.Load() {
+ return activator.SwapFailed
+ }
+ return activator.SwapDone
+ }
+ h.pointer("stable", "v2", promotedAt)
+ h.install("v2", "a", "b")
+ h.pass()
+ waitFor(t, "the first failure", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
+ waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
+
+ h.pointer("stable", "v2", promotedAt.Add(time.Minute))
+ h.advance(pointerReadEvery)
+ h.pass()
+ waitFor(t, "the retry to fail too", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 2 })
+ waitFor(t, "the record again", func() bool {
+ r, ok := h.book.Lookup("bab", "blog", "v2")
+ return ok && r.PromotedAt.Equal(promotedAt.Add(time.Minute))
+ })
+
+ h.failing.Store(false)
+ h.pointer("stable", "v2", promotedAt.Add(2*time.Minute))
+ h.advance(pointerReadEvery)
+ h.pass()
+ waitFor(t, "every host on v2", func() bool { return h.servingAll("v2", "a", "b") })
+ calls, _, _ := h.act.snapshot()
+ for i, c := range calls[:3] {
+ if c.a.Host != "a" || !c.a.Canary {
+ t.Fatalf("call %d = %+v, want a canary on a", i, c.a)
+ }
+ }
+ if c := calls[2].a; c.Cause != activator.CauseRetry {
+ t.Fatalf("second retry's attempt = %+v, want cause retry", c)
+ }
+}
+
+// Carried from Task 6 and plan 3 (O7): a hold attempt that fails is held
+// down in the book before its slot frees, so no wake planned in between can
+// boot the failed hold build. The mode is the one Swap ran (a foreign-layout
+// overlap runs as a hold, SwapResult.Mode), and the record's build comes from
+// the dispatch: a failed hold returns no generation.
+func TestAFailedHoldIsHeldDownBeforeItsSlotFrees(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.act.mode = activator.ModeHold
+ var failed atomic.Bool
+ h.act.outcome = func(activator.Attempt) activator.SwapOutcome {
+ failed.Store(true)
+ return activator.SwapDown
+ }
+ var bad, seen atomic.Int32
+ h.rows.mu.Lock()
+ h.rows.getHook = func(host string) {
+ if !failed.Load() {
+ return
+ }
+ s := h.m.schedulerOf(host)
+ s.mu.Lock()
+ free := s.q.hosts[host] != nil && s.q.hosts[host].running == nil
+ s.mu.Unlock()
+ if !free {
+ return
+ }
+ seen.Add(1)
+ if !h.book.HeldDown("bab", "blog", "v2", host) {
+ bad.Add(1)
+ }
+ }
+ h.rows.mu.Unlock()
+ h.install("v2", "a")
+ h.pass()
+ waitFor(t, "the held-down record", func() bool { return h.book.HeldDown("bab", "blog", "v2", "a") })
+ waitFor(t, "the down event", func() bool { return h.events.has("failed in hold mode") })
+ if bad.Load() != 0 || seen.Load() == 0 {
+ t.Fatalf("%d of %d look(s) with the slot free found the host not held down", bad.Load(), seen.Load())
+ }
+ r, _ := h.book.Lookup("bab", "blog", "v2")
+ if r.Mode != string(activator.ModeHold) || r.Build != "v2" || r.Host != "a" {
+ t.Fatalf("record = %+v, want build v2 held down in the mode Swap ran", r)
+ }
+}
+
+// Carried from Task 4: every Pass prunes, a held-down instance whose route
+// still names the failed build keeps the record, and a record no route
+// installs any more goes.
+func TestPassPrunesOnlyRecordsNoRouteInstalls(t *testing.T) {
+ h := newHarness(t, "a", "b")
+ h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a", HeldDown: []string{"a"}, PromotedAt: promotedAt, At: t0})
+ h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v0", Host: "b", PromotedAt: promotedAt, At: t0})
+ h.install("v2", "a")
+ h.pass()
+ if _, ok := h.book.Lookup("bab", "blog", "v0"); ok {
+ t.Fatal("a record no route installs survived the pass")
+ }
+ if !h.book.HeldDown("bab", "blog", "v2", "a") {
+ t.Fatal("the pass pruned a held-down instance's record while its route still names the build")
+ }
+ h.install("v1", "a") // rolled back by hand
+ h.pass()
+ if len(h.book.List()) != 0 {
+ t.Fatalf("records = %+v after the last route left v2", h.book.List())
+ }
+}
+
+// Carried from Task 6, spec §8.2/§8.4: an install by hand that clears the
+// record lifts the halt with it. A halt is in memory and outlived its record,
+// so installing the build again later was refused for good.
+func TestAnInstallByHandThatClearsTheRecordLiftsTheHalt(t *testing.T) {
+ h := newHarness(t, "a", "b")
+ h.stage("v2")
+ h.failing.Store(true)
+ h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
+ if a.Build == "v2" && h.failing.Load() {
+ return activator.SwapFailed
+ }
+ return activator.SwapDone
+ }
+ h.install("v2", "a", "b")
+ h.pass()
+ waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
+ h.install("v1", "a", "b")
+ h.pass()
+ if len(h.book.List()) != 0 {
+ t.Fatal("the record outlived every route installing the build")
+ }
+ h.failing.Store(false)
+ h.install("v2", "a", "b")
+ h.pass()
+ waitFor(t, "every host on v2", func() bool { return h.servingAll("v2", "a", "b") })
+ calls, _, _ := h.act.snapshot()
+ if c := calls[1].a; !c.Canary {
+ t.Fatalf("first attempt after the hold lifted = %+v, want a fresh canary", c)
+ }
+}
+
+// Carried from Task 6: `carlos instances retry` runs that host as a canary
+// through its cohort's halt, and only that host.
+func TestRetryHostRunsACanaryThroughAHalt(t *testing.T) {
+ h := newHarness(t, "a", "b")
+ h.stage("v2")
+ h.failing.Store(true)
+ h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
+ if a.Build == "v2" && h.failing.Load() {
+ return activator.SwapFailed
+ }
+ return activator.SwapDone
+ }
+ h.install("v2", "a", "b")
+ h.pass()
+ waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
+ h.failing.Store(false)
+ if !h.m.RetryHost("b") {
+ t.Fatal("RetryHost did not clear b")
+ }
+ waitFor(t, "b on v2", func() bool { return h.servingAll("v2", "b") })
+ calls, _, _ := h.act.snapshot()
+ if c := calls[1].a; c.Host != "b" || !c.Canary || c.Cause != activator.CauseRetry {
+ t.Fatalf("attempt after the host retry = %+v, want b's retry canary", c)
+ }
+ h.pass()
+ time.Sleep(50 * time.Millisecond)
+ if h.servingAll("v2", "a") {
+ t.Fatal("a host retry lifted the halt for another host")
+ }
+}
+
+// Carried from Task 6: a failed config attempt backs off even when its row
+// cannot be read afterwards; the backoff is set in the step that frees the
+// slot, so nothing can re-admit the attempt before it. The row is unreadable
+// only for the read that follows the failure: the next attempt's own read
+// succeeds, so only the backoff can keep it from running.
+func TestAFailedConfigAttemptBacksOffWithItsRowUnreadable(t *testing.T) {
+ h := newHarness(t, "a")
+ var failed, refused atomic.Bool
+ h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
+ if a.Cause == activator.CauseConfig {
+ failed.Store(true)
+ return activator.SwapFailed
+ }
+ return activator.SwapDone
+ }
+ h.rows.mu.Lock()
+ h.rows.getErr = func(string) error {
+ if failed.Load() && refused.CompareAndSwap(false, true) {
+ return errors.New("registry: database is locked")
+ }
+ return nil
+ }
+ h.rows.mu.Unlock()
+ h.pass()
+ h.m.ConfigDelivered("a", "sha256:c1")
+ waitFor(t, "the first config attempt", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) >= 1 })
+ h.pass()
+ time.Sleep(50 * time.Millisecond)
+ if calls, _, _ := h.act.snapshot(); len(calls) != 1 {
+ t.Fatalf("%d config attempts with no backoff after an unreadable row", len(calls))
+ }
+}
+
+// Carried from Task 7: activating an older revision than the delivered one
+// leaves the delivered revision's failure count, and so its backoff, alone.
+func TestAnOlderActivationKeepsTheConfigFailureCount(t *testing.T) {
+ h := newHarness(t, "a")
+ h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
+ if a.Cause == activator.CauseConfig {
+ return activator.SwapFailed
+ }
+ return activator.SwapDone
+ }
+ h.pass()
+ h.m.ConfigDelivered("a", "sha256:c2")
+ waitFor(t, "the pending event", func() bool { return h.events.has("config delivered, not active") })
+ h.m.configActivated("a", "sha256:c1")
+ h.m.mu.Lock()
+ fails := h.m.cfg["a"].fails
+ h.m.mu.Unlock()
+ if fails != 1 {
+ t.Fatalf("fails = %d after an older revision activated, want 1", fails)
+ }
+}
+
+// Carried from Task 6, spec §8.1: an instance that goes to sleep with a
+// config attempt pending needs no attempt any more. Its next wake starts with
+// the delivered file, so the revision is active, and the held ack goes out as
+// a sleeper's does, marked Hibernating. After it wakes, no config attempt
+// restarts a generation that already runs the revision.
+func TestASleepersPendingConfigIsNoAttemptAfterItWakes(t *testing.T) {
+ h := newHarness(t, "a")
+ h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
+ if a.Cause == activator.CauseConfig {
+ return activator.SwapFailed
+ }
+ return activator.SwapDone
+ }
+ h.pass()
+ var mu sync.Mutex
+ var acks []appconfig.ConfigAck
+ ack := h.m.DeferAck(func(a appconfig.ConfigAck) {
+ mu.Lock()
+ defer mu.Unlock()
+ acks = append(acks, a)
+ })
+ h.m.ConfigDelivered("a", "sha256:c1")
+ ack(appconfig.ConfigAck{Host: "a", Restarted: true})
+ waitFor(t, "the failed config attempt", func() bool { return h.events.has("config delivered, not active") })
+ h.act.mu.Lock()
+ delete(h.act.gens, "a") // hibernated inside its backoff
+ h.act.mu.Unlock()
+ h.pass()
+ waitFor(t, "the held ack", func() bool { mu.Lock(); defer mu.Unlock(); return len(acks) == 1 })
+ mu.Lock()
+ if !acks[0].Hibernating || acks[0].Restarted {
+ t.Fatalf("ack = %+v, want a sleeper's ack, Hibernating", acks[0])
+ }
+ mu.Unlock()
+ h.act.serve("a", "v1") // woke, with the delivered file
+ h.advance(2 * time.Minute)
+ h.pass()
+ time.Sleep(50 * time.Millisecond)
+ if calls, _, _ := h.act.snapshot(); len(calls) != 1 {
+ t.Fatalf("calls = %+v: a woken sleeper ran its old pending config attempt", calls)
+ }
+}
+
+// Plan 3's deferral (O5): an attempt abandoned while it is still the host's
+// newest (a hold whose stop withdrew, a lost re-check) waits before it runs
+// again, rather than looping stop cycles. One abandoned by a newer desire
+// does not wait (TestTwoReleasesTwentySecondsApartNeverSwitchToTheFirst).
+func TestAnAbandonedAttemptStillNewestWaitsBeforeItRunsAgain(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.act.outcome = func(activator.Attempt) activator.SwapOutcome { return activator.SwapAbandoned }
+ h.install("v2", "a")
+ h.pass()
+ waitFor(t, "the first attempt", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) >= 1 })
+ h.pass()
+ time.Sleep(50 * time.Millisecond)
+ if calls, _, _ := h.act.snapshot(); len(calls) != 1 {
+ t.Fatalf("%d attempts: an abandoned attempt ran again at once", len(calls))
+ }
+ h.advance(abandonDeferral + time.Second)
+ h.pass()
+ waitFor(t, "the attempt after the wait", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 2 })
+ if len(h.book.List()) != 0 {
+ t.Fatal("an abandoned attempt wrote a failure record")
+ }
+}
+
+// PF5 (plan 3's O8): member-facing events carry the reason word, never the
+// swap's diagnostic text, which can name paths and internals.
+func TestMemberEventsCarryOnlyTheReasonWord(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.act.reason = "spawn: fork/exec /opt/carlos/versions/bab/v2/blog: permission denied"
+ h.act.outcome = func(activator.Attempt) activator.SwapOutcome { return activator.SwapFailed }
+ h.install("v2", "a")
+ h.pass()
+ waitFor(t, "the halt event", func() bool { return h.events.has("rollout of build v2 halted") })
+ waitFor(t, "the failed event", func() bool { return h.events.has("swap to build v2 failed") })
+ for _, raw := range []string{"permission denied", "fork/exec", "blog:"} {
+ if h.events.has(raw) {
+ t.Fatalf("a member event carries the raw reason %q", raw)
+ }
+ }
+ if !h.events.has("swap to build v2 failed: health. still serving build v1") ||
+ !h.events.has("halted on box box-1: health") {
+ t.Fatal("the events do not carry the reason word")
+ }
+}
+
+func TestReasonWords(t *testing.T) {
+ for text, want := range map[string]string{
+ "health check failed": reasonHealth,
+ "swap: successor exited during its settle": reasonSettle,
+ "exited within 10s of becoming healthy": reasonExit,
+ "staging: digest mismatch": reasonStaging,
+ "version name reused with different content, ship under a new name": reasonReused,
+ "GET / answered 500": reasonHealth,
+ "": reasonHealth,
+ } {
+ if got := reasonWord(text); got != want {
+ t.Errorf("reasonWord(%q) = %q, want %q", text, got, want)
+ }
+ }
+}
diff --git a/internal/rollout/scheduler.go b/internal/rollout/scheduler.go
new file mode 100644
index 000000000..1ec45b540
--- /dev/null
+++ b/internal/rollout/scheduler.go
@@ -0,0 +1,258 @@
+package rollout
+
+import (
+ "context"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// scheduler is one app's goroutine (spec §6.1). It runs off the agent's
+// tick, so a slow rollout never delays the tick or another app.
+type scheduler struct {
+ m *Manager
+ key appKey
+ mu sync.Mutex
+ q *appQueue
+ kick chan struct{}
+}
+
+func (s *scheduler) poke() {
+ select {
+ case s.kick <- struct{}{}:
+ default:
+ }
+}
+
+// newest is handed to Swap for its re-checks (spec §6.3).
+func (s *scheduler) newest(host string) uint64 {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.q.newest(host)
+}
+
+func (s *scheduler) loop(ctx context.Context) {
+ timer := time.NewTimer(time.Hour)
+ defer timer.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-s.kick:
+ case <-timer.C:
+ }
+ view, pendingConfig := s.view()
+ for host, rev := range pendingConfig {
+ if serving, seen := view[host]; seen && serving == "" {
+ s.m.configSlept(host, rev)
+ }
+ }
+ now := s.m.d.Now()
+ s.mu.Lock()
+ ds := s.q.admit(view, now)
+ next := s.q.nextDue(now)
+ s.mu.Unlock()
+ for _, d := range ds {
+ go s.execute(ctx, d)
+ }
+ wait := time.Hour
+ if !next.IsZero() {
+ wait = max(next.Sub(now), time.Millisecond)
+ }
+ timer.Reset(wait)
+ }
+}
+
+// view asks the activator, with no lock held, what each queued host serves:
+// a build when awake, "" when asleep, absent while waking or stopping. It
+// also returns each host's pending config revision as it stood BEFORE the
+// activator was asked: a host then reported asleep stopped with that file
+// already delivered, so its next wake runs it (configSlept).
+func (s *scheduler) view() (map[string]string, map[string]string) {
+ s.mu.Lock()
+ hosts := make([]string, 0, len(s.q.hosts))
+ pendingConfig := map[string]string{}
+ for name, h := range s.q.hosts {
+ hosts = append(hosts, name)
+ if h.wantConfig != "" {
+ pendingConfig[name] = h.wantConfig
+ }
+ }
+ s.mu.Unlock()
+ out := make(map[string]string, len(hosts))
+ for _, h := range hosts {
+ if g, ok := s.m.d.Act.Current(h); ok && g.Build != "" {
+ out[h] = g.Build
+ } else if !s.m.d.Act.Owns(h) {
+ out[h] = ""
+ }
+ }
+ return out, pendingConfig
+}
+
+// execute runs one attempt to its outcome.
+func (s *scheduler) execute(ctx context.Context, d dispatch) {
+ m := s.m
+ rt, err := m.d.Routes.Get(d.ID.Host)
+ if err != nil || rt == nil {
+ m.d.Logf("rollout: %s: route not readable for its %s attempt onto %s: %v", d.ID.Host, d.ID.Cause, d.ID.Build, err)
+ s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: "route not readable"}, "")
+ return
+ }
+ a, err := m.resolve(ctx, d, *rt)
+ if err != nil {
+ m.stagingRefused(*rt, d.ID.Build, err)
+ s.mu.Lock()
+ s.q.deferHost(d.ID.Host, m.d.Now().Add(stagingDeferral))
+ s.mu.Unlock()
+ s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: err.Error()}, "")
+ return
+ }
+ // The failing pointer's PromotedAt, read now, while nothing is at stake:
+ // done() publishes a failure without a blocking read (spec §8.1).
+ if d.Class == classAdopt {
+ d.FailedUnder = m.failedUnder(*rt, d.ID.Build)
+ }
+ m.emit(*rt, EvSwapping, "info", textSwapping(a.Build, a.Cause))
+ s.done(d, m.d.Act.Swap(ctx, a, s.newest), a.Mode)
+}
+
+// done records an attempt's outcome and acts on it. mode is the attempt's
+// swap mode, "" when no Swap ran.
+func (s *scheduler) done(d dispatch, res activator.SwapResult, mode activator.SwapMode) {
+ m := s.m
+ if res.Mode != "" {
+ // The mode the attempt actually ran in: an overlap attempt on a
+ // foreign-layout row runs as a hold (spec §2.9), and a failed hold
+ // is held down (§6.4, §8.3).
+ mode = res.Mode
+ }
+ failed := res.Outcome == activator.SwapFailed || res.Outcome == activator.SwapDown
+ // A failure is published with no blocking read before it (spec §8.1):
+ // the failing pointer's PromotedAt was read before the attempt ran
+ // (d.FailedUnder), and the record is written in the same critical
+ // section that frees the slot and halts the cohort. No admission, no
+ // newest re-check and no wake plan (which reads the book) sees the slot
+ // free before the record. The record's build and host are the
+ // dispatch's, never res.Gen's: a failed hold returns no generation.
+ var rec *FailureRecord
+ if failed {
+ r := FailureRecord{Account: s.key.acct, App: s.key.app, Build: d.ID.Build, Host: d.ID.Host,
+ Cause: string(d.ID.Cause), Mode: string(mode), Reason: res.Reason, PromotedAt: d.FailedUnder, At: m.d.Now()}
+ if r.PromotedAt.IsZero() {
+ // No pointer named the build when the attempt started: the
+ // failure's own time, so only a pointer signed after it clears
+ // the record.
+ r.PromotedAt = r.At
+ }
+ if res.Outcome == activator.SwapDown && mode == activator.ModeHold {
+ // A hold build executed against this instance's database:
+ // nothing falls back on it automatically (spec §6.4, §8.3).
+ // Held down in the same step that frees the slot, so no wake
+ // planned in between can boot the failed hold build.
+ r.HeldDown = []string{d.ID.Host}
+ }
+ rec = &r
+ }
+ // A failed attempt that carried a config revision leaves it delivered and
+ // not active (spec §8.1). Its backoff is counted here, before the slot
+ // frees, and set on the queue in the same step that frees it, so the
+ // loop cannot re-admit the attempt in between. A failed adoption goes
+ // through the cohort-failure transition instead, and the next attempt
+ // for the host is its recovery onto the serving build with the config.
+ var (
+ cfgWait time.Duration
+ cfgCounted bool
+ )
+ if failed && d.Class == classRecover && d.ID.ConfigRev != "" {
+ cfgWait, cfgCounted = m.configFailure(d.ID.Host, d.ID.ConfigRev, res.Reason)
+ }
+ if res.Outcome == activator.SwapDone {
+ // The config the attempt started with is active, published under the
+ // same rule as the serving state (spec §8.1): before the slot frees,
+ // so a newer config attempt cannot complete first and be overwritten
+ // here, and with no registry read in front of it, so an unreadable
+ // row loses nothing.
+ m.configActivated(d.ID.Host, d.ID.ConfigRev)
+ // Published BEFORE the slot is freed (spec §8.5): while the attempt
+ // still holds its host's slot no newer attempt can be admitted and
+ // complete first, so this write can never land on top of a newer
+ // one. writeGood also requires the generation to be the one the
+ // activator serves (acknowledged).
+ m.served(d.ID.Host, res.Gen)
+ }
+ now := m.d.Now()
+ s.mu.Lock()
+ transition := s.q.finish(d, res.Outcome)
+ var (
+ added bool
+ superseded []string
+ recErr error
+ )
+ if transition && rec != nil {
+ // The whole cohort-failure transition (spec §8.1 steps 1 and 2): the
+ // record (in memory at once, on disk as it can) and the halt.
+ added, superseded, recErr = failCohort(m.d.Book, s.q, *rec)
+ }
+ h := s.q.hosts[d.ID.Host]
+ if cfgCounted && h != nil && h.wantConfig == d.ID.ConfigRev {
+ s.q.configFailed(d.ID.Host, now.Add(cfgWait))
+ }
+ if res.Outcome == activator.SwapAbandoned && h != nil && h.buildRev == d.SchedRev && h.notBefore.Before(now.Add(abandonDeferral)) {
+ // Abandoned while still the host's newest: the same attempt would be
+ // derived again at once (plan 3's deferral). A staging refusal has
+ // already deferred the host by as long.
+ s.q.deferHost(d.ID.Host, now.Add(abandonDeferral))
+ }
+ serving := ""
+ if h != nil {
+ serving = h.serving
+ }
+ s.mu.Unlock()
+ defer s.poke()
+ if recErr != nil {
+ m.d.Logf("rollout: write failure record %s/%s %s: %v (held in memory)", rec.Account, rec.App, rec.Build, recErr)
+ }
+ if len(superseded) > 0 {
+ m.d.Logf("rollout: %s/%s build %s failed: superseding attempts past spawn on %v", rec.Account, rec.App, rec.Build, superseded)
+ }
+ if failed {
+ // The raw reason is for the operator; members get the reason word.
+ m.d.Logf("rollout: %s: %s attempt onto %s %s (%s mode): %s", d.ID.Host, d.ID.Cause, d.ID.Build, res.Outcome, mode, res.Reason)
+ }
+ rt := s.route(d.ID.Host)
+ if added {
+ m.emit(rt, EvCohortHalted, "warn", textCohortHalted(rec.Build, m.d.Box, reasonWord(rec.Reason)))
+ }
+ switch res.Outcome {
+ case activator.SwapDone:
+ m.mu.Lock()
+ if m.retried[d.ID.Host] == d.ID.Build {
+ delete(m.retried, d.ID.Host)
+ }
+ m.mu.Unlock()
+ m.emit(rt, EvSwapped, "info", textSwapped(d.ID.Build, res.Drain))
+ case activator.SwapFailed, activator.SwapDown:
+ if res.Outcome == activator.SwapDown {
+ m.emit(rt, EvSwapDown, "error", textSwapDown(d.ID.Build))
+ } else {
+ m.emit(rt, EvSwapFailed, "warn", textSwapFailed(d.ID.Build, serving, reasonWord(res.Reason)))
+ }
+ if cfgCounted {
+ m.emit(rt, EvConfigPending, "warn", textConfigPending(reasonWord(res.Reason), cfgWait))
+ }
+ case activator.SwapAbandoned:
+ m.d.Logf("rollout: %s: %s attempt onto %s abandoned: %s", d.ID.Host, d.ID.Cause, d.ID.Build, res.Reason)
+ }
+}
+
+// route is host's row for an event. A row that cannot be read still gets its
+// events, addressed by what the scheduler knows: its app and the host.
+func (s *scheduler) route(host string) registry.Route {
+ if rt, err := s.m.d.Routes.Get(host); err == nil && rt != nil {
+ return *rt
+ }
+ return registry.Route{Host: host, Account: s.key.acct, App: s.key.app}
+}