amadan

carlos / platform Public

zdd-5 task 6: cohorts, canary gating, swap_concurrency and the cohort-failure transition

Admission enforces notBefore and configAfter itself, because pending does not
read them. failCohort writes the record and applies the halt in one call, so
the caller's single critical section also frees the failed slot.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 8eb83e60262e08aadcd85a9ad9c27039a4952b18 parent 831f257
2 files changed, +449 −2
  • internal/rollout/admit_test.go +237 −0
  • internal/rollout/queue.go +212 −2
diff --git a/internal/rollout/admit_test.go b/internal/rollout/admit_test.go
new file mode 100644
index 000000000..55ea11355
--- /dev/null
+++ b/internal/rollout/admit_test.go
@@ -0,0 +1,237 @@
+package rollout
+
+import (
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+)
+
+// currentView is every slot as the activator would report it.
+func currentView(q *appQueue) map[string]string {
+ v := map[string]string{}
+ for host, h := range q.hosts {
+ v[host] = h.serving
+ }
+ return v
+}
+
+func releaseAll(q *appQueue, build string, hosts ...string) {
+ for _, h := range hosts {
+ q.wantBuild(h, activator.CauseRelease, build, t0)
+ }
+}
+
+func TestTheCanaryRunsAloneUntilItSettles(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b", "c")
+ q.concurrency = 2
+ releaseAll(q, "v2", "a", "b", "c")
+ first := q.admit(currentView(q), t0)
+ if len(first) != 1 || !first[0].Canary {
+ t.Fatalf("admitted %+v, want exactly one canary", first)
+ }
+ if more := q.admit(currentView(q), t0); len(more) != 0 {
+ t.Fatalf("admitted %+v while the canary has not settled", more)
+ }
+ q.finish(first[0], activator.SwapDone)
+ rest := q.admit(currentView(q), t0)
+ if len(rest) != 2 || rest[0].Canary || rest[1].Canary {
+ t.Fatalf("after the settle admitted %+v, want the other two and no canary", rest)
+ }
+}
+
+func TestSwapConcurrencyIsNeverExceeded(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b", "c", "d")
+ q.concurrency = 1
+ releaseAll(q, "v2", "a", "b", "c", "d")
+ for done := 0; done < 4; {
+ ds := q.admit(currentView(q), t0)
+ if q.running > q.concurrency || len(ds) > 1 {
+ t.Fatalf("running %d, admitted %d, with concurrency %d", q.running, len(ds), q.concurrency)
+ }
+ for _, d := range ds {
+ q.finish(d, activator.SwapDone)
+ done++
+ }
+ }
+}
+
+func TestAnAbandonedCanaryIsReplaced(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b")
+ releaseAll(q, "v2", "a", "b")
+ first := q.admit(currentView(q), t0)
+ q.finish(first[0], activator.SwapAbandoned)
+ next := q.admit(currentView(q), t0)
+ if len(next) != 1 || !next[0].Canary {
+ t.Fatalf("after an abandoned canary admitted %+v, want a new canary", next)
+ }
+}
+
+func TestHaltOnTheFirstFailureSparesInstancesServingTheBuild(t *testing.T) {
+ b := &FailureBook{}
+ q := newAppQueue("bab", "blog", func(build, host string, served bool) bool {
+ return b.Blocks("bab", "blog", build, host, served)
+ })
+ for _, h := range []string{"a", "b", "c"} {
+ q.observe(h, "v1", "v1", t0)
+ }
+ q.observe("d", "v2", "v2", t0) // d served v2 before it failed elsewhere
+ q.concurrency = 3
+ releaseAll(q, "v2", "a", "b", "c", "d")
+ canary := q.admit(currentView(q), t0)
+ if len(canary) != 1 {
+ t.Fatalf("admitted %+v", canary)
+ }
+ if !q.finish(canary[0], activator.SwapFailed) {
+ t.Fatal("a failed release attempt goes through the cohort-failure transition")
+ }
+ if _, _, err := failCohort(b, q, FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: canary[0].ID.Host, At: t0}); err != nil {
+ t.Fatal(err)
+ }
+ q.wantRestart("d", "r1", t0)
+ ds := q.admit(currentView(q), t0)
+ for _, d := range ds {
+ if d.ID.Build == "v2" && d.Class == classAdopt {
+ t.Fatalf("admitted %+v after the halt", d.ID)
+ }
+ }
+ if len(ds) != 1 || ds[0].ID.Host != "d" || ds[0].ID.Cause != activator.CauseRestart {
+ t.Fatalf("admitted %+v, want only d's restart onto the v2 it serves (spec §8.4)", ds)
+ }
+}
+
+// orderBook watches the transition's order: when the record is written, the
+// attempts past spawn must not yet have been superseded.
+type orderBook struct {
+ FailureBook
+ q *appQueue
+ host string
+ atRecord uint64
+}
+
+func (o *orderBook) Record(r FailureRecord) (bool, error) {
+ o.atRecord = o.q.newest(o.host)
+ return o.FailureBook.Record(r)
+}
+
+func TestTheRecordComesBeforeTheSupersession(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b")
+ q.concurrency = 2
+ releaseAll(q, "v2", "a", "b")
+ canary := q.admit(currentView(q), t0)
+ q.finish(canary[0], activator.SwapDone)
+ running := q.admit(currentView(q), t0) // b, now past spawn
+ if len(running) != 1 {
+ t.Fatalf("admitted %+v", running)
+ }
+ before := q.newest(running[0].ID.Host)
+ ob := &orderBook{q: q, host: running[0].ID.Host}
+ _, superseded, _ := failCohort(ob, q, FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "c", At: t0})
+ if ob.atRecord != before {
+ t.Fatal("attempts were superseded before the failure record was written (spec §8.1 order)")
+ }
+ if len(superseded) != 1 || q.newest(running[0].ID.Host) == before {
+ t.Fatalf("superseded %v: an attempt past spawn onto the failed build must be abandoned at its re-check", superseded)
+ }
+}
+
+func TestAWakingInstanceKeepsItsPlaceAndASleeperLosesItsRestart(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b")
+ q.wantRestart("a", "r1", t0)
+ q.wantRestart("b", "r1", t0)
+ v := currentView(q)
+ delete(v, "a") // waking
+ v["b"] = "" // asleep
+ if ds := q.admit(v, t0); len(ds) != 0 {
+ t.Fatalf("admitted %+v", ds)
+ }
+ if q.hosts["a"].wantRestart != "r1" {
+ t.Fatal("a waking instance keeps its place (spec §6.2)")
+ }
+ if q.hosts["b"].wantRestart != "" {
+ t.Fatal("a restart never wakes a sleeper")
+ }
+}
+
+func TestCohortsAreKeyedByDesiredBuild(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b")
+ q.concurrency = 2
+ q.wantBuild("a", activator.CauseRelease, "v2", t0) // one channel
+ q.wantBuild("b", activator.CauseRelease, "v3", t0) // another
+ ds := q.admit(currentView(q), t0)
+ if len(ds) != 2 || !ds[0].Canary || !ds[1].Canary {
+ t.Fatalf("admitted %+v, want each cohort's own canary", ds)
+ }
+}
+
+func TestARetriedInstanceRunsAsACanaryThroughAHalt(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b")
+ releaseAll(q, "v2", "a", "b")
+ q.halt("v2", "health check failed")
+ if ds := q.admit(currentView(q), t0); len(ds) != 0 {
+ t.Fatalf("admitted %+v from a halted cohort", ds)
+ }
+ q.retryHost("b")
+ ds := q.admit(currentView(q), t0)
+ if len(ds) != 1 || ds[0].ID.Host != "b" || !ds[0].Canary {
+ t.Fatalf("admitted %+v, want b alone, as a canary", ds)
+ }
+ q.unhalt("v2")
+ q.finish(ds[0], activator.SwapDone)
+ if ds := q.admit(currentView(q), t0); len(ds) != 1 || ds[0].ID.Host != "a" {
+ t.Fatalf("after the app retry admitted %+v", ds)
+ }
+}
+
+func TestAFailedConfigAttemptWaitsOutItsBackoff(t *testing.T) {
+ q := awakeQueue(noBlocks, "a")
+ q.wantConfig("a", "c1", t0)
+ q.configFailed("a", t0.Add(time.Minute))
+ if ds := q.admit(currentView(q), t0); len(ds) != 0 {
+ t.Fatal("admitted a config attempt inside its backoff")
+ }
+ if got := q.nextDue(t0); !got.Equal(t0.Add(time.Minute)) {
+ t.Fatalf("nextDue = %v, want the backoff's end", got)
+ }
+ if ds := q.admit(currentView(q), t0.Add(time.Minute)); len(ds) != 1 {
+ t.Fatal("the config attempt did not come back after its backoff")
+ }
+}
+
+// Task 5's pending() does not read notBefore, so admission must: a staging
+// refusal's wait holds every attempt for the host, not only config ones.
+func TestAStagingRefusalWaitsOutItsDeferral(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b")
+ q.concurrency = 2
+ releaseAll(q, "v2", "a", "b")
+ q.deferHost("a", t0.Add(30*time.Second))
+ ds := q.admit(currentView(q), t0)
+ if len(ds) != 1 || ds[0].ID.Host != "b" {
+ t.Fatalf("admitted %+v, want only b: a is deferred", ds)
+ }
+ if got := q.nextDue(t0); !got.Equal(t0.Add(30 * time.Second)) {
+ t.Fatalf("nextDue = %v, want the deferral's end", got)
+ }
+ q.finish(ds[0], activator.SwapDone)
+ if ds := q.admit(currentView(q), t0.Add(29*time.Second)); len(ds) != 0 {
+ t.Fatalf("admitted %+v inside the deferral", ds)
+ }
+ if ds := q.admit(currentView(q), t0.Add(30*time.Second)); len(ds) != 1 || ds[0].ID.Host != "a" {
+ t.Fatalf("after the deferral admitted %+v, want a", ds)
+ }
+ if got := q.nextDue(t0.Add(30 * time.Second)); !got.IsZero() {
+ t.Fatalf("nextDue = %v, want zero once nothing waits", got)
+ }
+}
+
+// A deferred host never holds the cohort's canary slot: the next host in the
+// queue takes it, so one refusal does not stall the whole rollout.
+func TestADeferredHostDoesNotHoldTheCanarySlot(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b")
+ releaseAll(q, "v2", "a", "b")
+ q.deferHost("a", t0.Add(time.Minute))
+ ds := q.admit(currentView(q), t0)
+ if len(ds) != 1 || ds[0].ID.Host != "b" || !ds[0].Canary {
+ t.Fatalf("admitted %+v, want b as the canary", ds)
+ }
+}
diff --git a/internal/rollout/queue.go b/internal/rollout/queue.go
index 782531fb1..4db495063 100644
--- a/internal/rollout/queue.go
+++ b/internal/rollout/queue.go
@@ -66,7 +66,7 @@ type hostSlot struct {
}
// cohort is one app's attempts on this box onto one build, of one class
-// (spec §3). Task 6 gives it its admission state; the queue only keys it.
+// (spec §3).
type cohort struct {
canary string // the host of the cohort's canary, "" until one starts
settled bool // the canary passed its health check and its settle
@@ -234,4 +234,214 @@ func cohortKey(build string, class attemptClass) string {
return "recover:" + build
}
-var _ = sort.Strings // admission (Task 6) orders hosts
+func (q *appQueue) cohortFor(key string) *cohort {
+ c := q.cohorts[key]
+ if c == nil {
+ c = &cohort{}
+ q.cohorts[key] = c
+ }
+ return c
+}
+
+// ordered is the queue order: oldest desire first, host name to break ties.
+func (q *appQueue) ordered() []*hostSlot {
+ out := make([]*hostSlot, 0, len(q.hosts))
+ for _, h := range q.hosts {
+ out = append(out, h)
+ }
+ sort.Slice(out, func(i, j int) bool {
+ if !out[i].since.Equal(out[j].since) {
+ return out[i].since.Before(out[j].since)
+ }
+ return out[i].host < out[j].host
+ })
+ return out
+}
+
+// admit hands out the attempts that may start now (spec §6.2). view maps each
+// host the activator could see to the build it serves ("" when asleep); a
+// host that is waking or stopping is absent, and keeps its place. A sleeper
+// loses only what needs a running process: its restart request.
+func (q *appQueue) admit(view map[string]string, now time.Time) []dispatch {
+ var out []dispatch
+ for _, h := range q.ordered() {
+ if h.running != nil {
+ continue
+ }
+ serving, seen := view[h.host]
+ if !seen {
+ continue
+ }
+ h.serving = serving
+ if serving == "" {
+ h.wantRestart = ""
+ continue
+ }
+ d, ok := q.gate(h, now)
+ if !ok {
+ continue
+ }
+ if q.running >= q.concurrency {
+ break
+ }
+ q.start(h, d)
+ out = append(out, d)
+ }
+ return out
+}
+
+// gate is the per-host half of admit, shared with the worker recycle hook:
+// the host's pending attempt, if its cohort lets it run now. pending does not
+// read the two waits, so they are enforced here: a staging refusal holds
+// every attempt for the host until notBefore, and a failed config attempt
+// waits out its backoff. A cohort's first attempt is its canary and runs
+// alone until it settles; a halted adoption cohort admits nothing but a
+// retried host's own canary.
+func (q *appQueue) gate(h *hostSlot, now time.Time) (dispatch, bool) {
+ d, ok := q.pending(h)
+ if !ok || now.Before(h.notBefore) {
+ return dispatch{}, false
+ }
+ if d.ID.Cause == activator.CauseConfig && now.Before(h.configAfter) {
+ return dispatch{}, false
+ }
+ c := q.cohortFor(d.Cohort)
+ switch {
+ case h.forceCanary:
+ d.Canary = true
+ case c.halted && d.Class == classAdopt:
+ return dispatch{}, false
+ case c.canary == "":
+ d.Canary = true
+ case !c.settled:
+ return dispatch{}, false
+ }
+ return d, true
+}
+
+// start marks d running. A cohort's canary is claimed here, at the moment it
+// actually starts, so a host gated out by the concurrency limit never holds
+// the canary slot.
+func (q *appQueue) start(h *hostSlot, d dispatch) {
+ if d.Canary && !h.forceCanary {
+ q.cohortFor(d.Cohort).canary = h.host
+ }
+ h.forceCanary = false
+ h.running = &d
+ q.running++
+}
+
+// finish records d's outcome and reports whether it is a build failure that
+// must go through the cohort-failure transition (spec §8.1): a failed
+// adoption attempt. A failed restart or config attempt is reported, never
+// recorded.
+func (q *appQueue) finish(d dispatch, out activator.SwapOutcome) (transition bool) {
+ h := q.hosts[d.ID.Host]
+ if h == nil || h.running == nil || h.running.SchedRev != d.SchedRev || h.running.ID != d.ID {
+ return false
+ }
+ h.running = nil
+ q.running--
+ c := q.cohorts[d.Cohort]
+ unclaim := func() {
+ if c != nil && c.canary == h.host && !c.settled {
+ c.canary = ""
+ }
+ }
+ switch out {
+ case activator.SwapDone:
+ h.serving = d.ID.Build
+ if h.wantRestart == d.ID.RestartID {
+ h.wantRestart = ""
+ }
+ if d.ID.ConfigRev != "" && h.wantConfig == d.ID.ConfigRev {
+ h.wantConfig = ""
+ }
+ if c != nil && c.canary == h.host {
+ c.settled = true
+ }
+ case activator.SwapAbandoned:
+ unclaim()
+ case activator.SwapFailed, activator.SwapDown:
+ if out == activator.SwapDown {
+ h.serving = ""
+ }
+ if d.ID.Cause == activator.CauseRestart && h.wantRestart == d.ID.RestartID {
+ h.wantRestart = ""
+ }
+ unclaim()
+ return d.Class == classAdopt
+ }
+ return false
+}
+
+// halt stops build's adoption cohort, and supersedes every adoption attempt
+// onto build already past spawn, so its re-check abandons it rather than
+// switching (spec §8.1 step 2). It returns the superseded hosts.
+func (q *appQueue) halt(build, reason string) (superseded []string) {
+ c := q.cohortFor(cohortKey(build, classAdopt))
+ c.halted, c.reason = true, reason
+ for _, h := range q.hosts {
+ if h.running != nil && h.running.Class == classAdopt && h.running.ID.Build == build {
+ h.buildRev = q.nextRev()
+ superseded = append(superseded, h.host)
+ }
+ }
+ sort.Strings(superseded)
+ return superseded
+}
+
+// unhalt reopens build's adoption cohort after `carlos deploy --retry`: its
+// next attempt is a fresh canary.
+func (q *appQueue) unhalt(build string) {
+ if c := q.cohorts[cohortKey(build, classAdopt)]; c != nil {
+ *c = cohort{}
+ }
+}
+
+// retryHost is `carlos instances retry <host>`: its next attempt is a canary.
+func (q *appQueue) retryHost(host string) {
+ if h := q.hosts[host]; h != nil {
+ h.forceCanary = true
+ }
+}
+
+// nextDue is the earliest moment a deferred attempt becomes admissible, so the
+// scheduler's goroutine knows when to look again; zero when nothing waits.
+func (q *appQueue) nextDue(now time.Time) time.Time {
+ var next time.Time
+ consider := func(t time.Time) {
+ if t.After(now) && (next.IsZero() || t.Before(next)) {
+ next = t
+ }
+ }
+ for _, h := range q.hosts {
+ consider(h.notBefore)
+ if h.wantConfig != "" {
+ consider(h.configAfter)
+ }
+ }
+ return next
+}
+
+// recorder is the failure book's write half, an interface so a test can
+// watch the transition's order.
+type recorder interface {
+ Record(FailureRecord) (bool, error)
+}
+
+// failCohort is the cohort-failure transition of spec §8.1, in its order:
+// (1) the failure record, before anything else can act on the failure;
+// (2) queued attempts onto the build are cancelled, because the record now
+// blocks them, and attempts past spawn are superseded; (3) wakes of the
+// cohort boot the good build, which needs nothing here: the planner reads
+// the record, and nothing is repointed; (4) instances that have served the
+// build are untouched, because the record never blocks them. The caller
+// holds the scheduler's lock for q, so the record, the halt and the freed
+// slot (finish) are one critical section. A failed file write still returns
+// the record held in memory, so the transition never stops halfway.
+func failCohort(book recorder, q *appQueue, rec FailureRecord) (added bool, superseded []string, err error) {
+ added, err = book.Record(rec)
+ superseded = q.halt(rec.Build, rec.Reason)
+ return added, superseded, err
+}