amadan

carlos / platform Public

zdd-5 task 5: per-instance desired state, attempt identity and supersession

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 831f257cc0ba5830c17174665aa8edbe25ef7573 parent 7094629
2 files changed, +367 −0
  • internal/rollout/queue.go +237 −0
  • internal/rollout/queue_test.go +130 −0
diff --git a/internal/rollout/queue.go b/internal/rollout/queue.go
new file mode 100644
index 000000000..782531fb1
--- /dev/null
+++ b/internal/rollout/queue.go
@@ -0,0 +1,237 @@
+package rollout
+
+import (
+ "sort"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+)
+
+// Identity is an attempt's identity (spec §6.3) without the scheduler
+// revision, which travels beside it in dispatch. Two submissions with equal
+// identities are one attempt.
+type Identity struct {
+ Host string
+ Cause activator.Cause
+ Build string
+ ConfigRev string
+ RestartID string
+}
+
+// attemptClass separates the two failure policies of spec §8.1.
+type attemptClass int
+
+const (
+ // classAdopt is a release, install or retry onto a build this instance
+ // has not served: its failure writes a record and halts the cohort.
+ classAdopt attemptClass = iota
+ // classRecover is a restart or config attempt, or a move back onto a
+ // build the instance has served: its failure writes no record.
+ classRecover
+)
+
+// dispatch is one attempt the queue has handed out to run.
+type dispatch struct {
+ ID Identity
+ SchedRev uint64
+ Canary bool
+ Class attemptClass
+ Cohort string
+ // FailedUnder is the PromotedAt of the pointer the attempt's build was
+ // promoted under, read by the scheduler (Task 8) before the attempt
+ // runs, so a failure is published with no blocking read in between
+ // (spec §8.1, §8.4). Zero: the failure's own time stands in.
+ FailedUnder time.Time
+}
+
+// hostSlot is one instance's desired state. The attempt to run is derived
+// from it (pending), never stored, so supersession and coalescing are the
+// same rule: the newest desire wins.
+type hostSlot struct {
+ host string
+ since time.Time // its place in the queue: when its current desire appeared
+ serving string // the current generation's build; "" when not awake
+ good string // GoodVersion
+
+ wantBuild string
+ wantCause activator.Cause
+ wantRestart string
+ wantConfig string
+ configAfter time.Time // a failed config attempt's backoff (spec §8.1)
+ notBefore time.Time // a staging refusal's wait (spec §4.2)
+
+ running *dispatch
+ buildRev uint64 // newest(host): moved only by a change of desired build
+ forceCanary bool // `carlos instances retry`: the next attempt is a canary
+}
+
+// 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.
+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
+ halted bool
+ reason string
+}
+
+// appQueue is one app's attempts on this box. It is not safe for concurrent
+// use: the scheduler holds its lock around every call.
+type appQueue struct {
+ acct, app string
+ concurrency int
+ hosts map[string]*hostSlot
+ cohorts map[string]*cohort
+ running int
+ rev uint64
+ // blocks is the failure book's answer for this app (spec §8.4).
+ blocks func(build, host string, served bool) bool
+}
+
+func newAppQueue(acct, app string, blocks func(build, host string, served bool) bool) *appQueue {
+ return &appQueue{acct: acct, app: app, concurrency: 1, hosts: map[string]*hostSlot{}, cohorts: map[string]*cohort{}, blocks: blocks}
+}
+
+func (q *appQueue) nextRev() uint64 {
+ q.rev++
+ return q.rev
+}
+
+func (q *appQueue) slot(host string, now time.Time) *hostSlot {
+ h := q.hosts[host]
+ if h == nil {
+ h = &hostSlot{host: host, since: now}
+ q.hosts[host] = h
+ }
+ return h
+}
+
+// newest is the revision an attempt past spawn must still carry when it
+// switches (spec §6.3). Only a change of desired build moves it: a restart
+// or config request that arrives meanwhile queues behind the attempt instead
+// of abandoning it.
+func (q *appQueue) newest(host string) uint64 {
+ if h := q.hosts[host]; h != nil {
+ return h.buildRev
+ }
+ return 0
+}
+
+// observe records the agent pass's view of one host. A running attempt owns
+// the serving build until it finishes.
+func (q *appQueue) observe(host, serving, good string, now time.Time) {
+ h := q.slot(host, now)
+ if h.running == nil {
+ h.serving = serving
+ }
+ h.good = good
+}
+
+func (q *appQueue) touch(h *hostSlot, now time.Time) {
+ if _, queued := q.pending(h); !queued && h.running == nil {
+ h.since = now
+ }
+}
+
+// wantBuild sets host's desired build: its installed build (spec §3). cause
+// is release, install or retry.
+func (q *appQueue) wantBuild(host string, cause activator.Cause, build string, now time.Time) bool {
+ if build == "" {
+ return false
+ }
+ h := q.slot(host, now)
+ if h.wantBuild == build {
+ if cause == activator.CauseRetry && h.wantCause != activator.CauseRetry {
+ h.wantCause = cause
+ return true
+ }
+ return false
+ }
+ q.touch(h, now)
+ h.wantBuild, h.wantCause = build, cause
+ h.buildRev = q.nextRev()
+ return true
+}
+
+// wantRestart records a `carlos restart` request; the newest id covers the
+// older ones (spec §6.3).
+func (q *appQueue) wantRestart(host, id string, now time.Time) bool {
+ h := q.slot(host, now)
+ if id == "" || h.wantRestart == id {
+ return false
+ }
+ q.touch(h, now)
+ h.wantRestart = id
+ return true
+}
+
+// wantConfig records a delivered config revision that is not yet active. A
+// newer revision replaces the pending attempt and its backoff (spec §8.1).
+func (q *appQueue) wantConfig(host, rev string, now time.Time) bool {
+ h := q.slot(host, now)
+ if h.wantConfig == rev {
+ return false
+ }
+ q.touch(h, now)
+ h.wantConfig, h.configAfter = rev, time.Time{}
+ return true
+}
+
+// configActive clears the pending config attempt once rev is active.
+func (q *appQueue) configActive(host, rev string) {
+ if h := q.hosts[host]; h != nil && h.wantConfig == rev {
+ h.wantConfig = ""
+ }
+}
+
+// configFailed holds a config-only attempt back until its backoff ends.
+func (q *appQueue) configFailed(host string, until time.Time) {
+ if h := q.hosts[host]; h != nil {
+ h.configAfter = until
+ }
+}
+
+// deferHost holds every attempt for host back until until: a staging refusal
+// waits one stage-relay window for root to restage.
+func (q *appQueue) deferHost(host string, until time.Time) {
+ if h := q.hosts[host]; h != nil {
+ h.notBefore = until
+ }
+}
+
+// pending derives the attempt host should run next, if any. Only an awake
+// instance with nothing running has one. A desired build the failure book
+// holds falls through to the instance's restart or config work, which goes
+// ahead on the build it serves (spec §8.4).
+func (q *appQueue) pending(h *hostSlot) (dispatch, bool) {
+ if h.running != nil || h.serving == "" {
+ return dispatch{}, false
+ }
+ id := Identity{Host: h.host, ConfigRev: h.wantConfig, RestartID: h.wantRestart}
+ switch {
+ case h.wantBuild != "" && h.wantBuild != h.serving && !q.blocks(h.wantBuild, h.host, h.wantBuild == h.good):
+ id.Cause, id.Build = h.wantCause, h.wantBuild
+ case h.wantRestart != "":
+ id.Cause, id.Build = activator.CauseRestart, h.serving
+ case h.wantConfig != "":
+ id.Cause, id.Build = activator.CauseConfig, h.serving
+ default:
+ return dispatch{}, false
+ }
+ class := classRecover
+ if id.Build != h.serving && id.Build != h.good {
+ class = classAdopt
+ }
+ return dispatch{ID: id, SchedRev: h.buildRev, Class: class, Cohort: cohortKey(id.Build, class)}, true
+}
+
+// cohortKey names a cohort (spec §3): the attempts onto one build, split by
+// class so a halt of the adoption cohort never stops the restarts and config
+// changes of instances already serving the build.
+func cohortKey(build string, class attemptClass) string {
+ if class == classAdopt {
+ return "adopt:" + build
+ }
+ return "recover:" + build
+}
+
+var _ = sort.Strings // admission (Task 6) orders hosts
diff --git a/internal/rollout/queue_test.go b/internal/rollout/queue_test.go
new file mode 100644
index 000000000..3171f6d36
--- /dev/null
+++ b/internal/rollout/queue_test.go
@@ -0,0 +1,130 @@
+package rollout
+
+import (
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+)
+
+func noBlocks(string, string, bool) bool { return false }
+
+// awakeQueue is an app queue whose hosts all serve v1, their good build.
+func awakeQueue(blocks func(build, host string, served bool) bool, hosts ...string) *appQueue {
+ q := newAppQueue("bab", "blog", blocks)
+ for _, h := range hosts {
+ q.observe(h, "v1", "v1", t0)
+ }
+ return q
+}
+
+func TestAReleaseOntoTheServingBuildIsDropped(t *testing.T) {
+ q := awakeQueue(noBlocks, "a")
+ q.wantBuild("a", activator.CauseRelease, "v1", t0)
+ if d, ok := q.pending(q.hosts["a"]); ok {
+ t.Fatalf("pending = %+v, want nothing: v1 is already serving", d.ID)
+ }
+}
+
+func TestRestartAndConfigOntoTheServingBuildAreRealWork(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b")
+ q.wantBuild("a", activator.CauseRelease, "v1", t0)
+ q.wantRestart("a", "r1", t0)
+ d, ok := q.pending(q.hosts["a"])
+ if !ok || d.ID != (Identity{Host: "a", Cause: activator.CauseRestart, Build: "v1", RestartID: "r1"}) {
+ t.Fatalf("restart pending = %+v, %v", d.ID, ok)
+ }
+ if d.Class != classRecover {
+ t.Fatal("a restart is recovery: no failure record, no halt")
+ }
+ q.wantConfig("b", "c1", t0)
+ d, ok = q.pending(q.hosts["b"])
+ if !ok || d.ID != (Identity{Host: "b", Cause: activator.CauseConfig, Build: "v1", ConfigRev: "c1"}) {
+ t.Fatalf("config pending = %+v, %v", d.ID, ok)
+ }
+}
+
+func TestDedupeComparesTheWholeIdentity(t *testing.T) {
+ q := awakeQueue(noBlocks, "a")
+ if !q.wantBuild("a", activator.CauseRelease, "v2", t0) {
+ t.Fatal("the first desire of v2 must register")
+ }
+ rev := q.newest("a")
+ if q.wantBuild("a", activator.CauseRelease, "v2", t0) {
+ t.Fatal("the same desire again is the same attempt")
+ }
+ if q.newest("a") != rev {
+ t.Fatal("a duplicate moved the newest revision")
+ }
+ if !q.wantBuild("a", activator.CauseRetry, "v2", t0) {
+ t.Fatal("a retry onto the same build is a different identity")
+ }
+}
+
+func TestANewerDesiredBuildSupersedesPerInstance(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b")
+ q.wantBuild("a", activator.CauseRelease, "v2", t0)
+ q.wantBuild("b", activator.CauseRelease, "v2", t0)
+ ra := q.newest("a")
+ q.wantBuild("a", activator.CauseRelease, "v3", t0.Add(20*time.Second))
+ d, _ := q.pending(q.hosts["a"])
+ if d.ID.Build != "v3" || d.SchedRev != q.newest("a") || q.newest("a") == ra {
+ t.Fatalf("a pending %+v rev %d newest %d (was %d)", d.ID, d.SchedRev, q.newest("a"), ra)
+ }
+ if d, _ := q.pending(q.hosts["b"]); d.ID.Build != "v2" {
+ t.Fatal("supersession is per instance: b keeps v2 until its own desire moves")
+ }
+}
+
+func TestRestartAndConfigCoalesceAndDoNotSupersede(t *testing.T) {
+ q := awakeQueue(noBlocks, "a")
+ q.wantBuild("a", activator.CauseRelease, "v2", t0)
+ q.wantRestart("a", "r1", t0)
+ q.wantRestart("a", "r2", t0)
+ q.wantConfig("a", "c1", t0)
+ q.wantConfig("a", "c2", t0)
+ d, _ := q.pending(q.hosts["a"])
+ want := Identity{Host: "a", Cause: activator.CauseRelease, Build: "v2", ConfigRev: "c2", RestartID: "r2"}
+ if d.ID != want {
+ t.Fatalf("pending = %+v, want %+v: one attempt covers the newest restart and config", d.ID, want)
+ }
+ rev := q.newest("a")
+ q.wantRestart("a", "r3", t0)
+ q.wantConfig("a", "c3", t0)
+ if q.newest("a") != rev {
+ t.Fatal("a restart or config request must not abandon an attempt past spawn")
+ }
+}
+
+func TestABlockedBuildFallsThroughToRecovery(t *testing.T) {
+ held := func(build, _ string, served bool) bool { return build == "v2" && !served }
+ q := awakeQueue(held, "a")
+ q.wantBuild("a", activator.CauseRelease, "v2", t0)
+ if _, ok := q.pending(q.hosts["a"]); ok {
+ t.Fatal("a held build must not be attempted")
+ }
+ q.wantRestart("a", "r1", t0)
+ d, ok := q.pending(q.hosts["a"])
+ if !ok || d.ID.Cause != activator.CauseRestart || d.ID.Build != "v1" {
+ t.Fatalf("pending = %+v: the restart onto the serving build still goes ahead", d.ID)
+ }
+}
+
+func TestAMoveBackToTheGoodBuildIsRecovery(t *testing.T) {
+ q := newAppQueue("bab", "blog", func(_, _ string, served bool) bool { return !served })
+ q.observe("a", "v2", "v1", t0) // serving v2, which has not settled; good v1
+ q.wantBuild("a", activator.CauseRelease, "v1", t0)
+ d, ok := q.pending(q.hosts["a"])
+ if !ok || d.Class != classRecover {
+ t.Fatalf("pending = %+v %v: moving back onto a build it has served is recovery", d, ok)
+ }
+}
+
+func TestNothingIsSwappedOnASleeper(t *testing.T) {
+ q := newAppQueue("bab", "blog", noBlocks)
+ q.observe("a", "", "v1", t0)
+ q.wantBuild("a", activator.CauseRelease, "v2", t0)
+ if _, ok := q.pending(q.hosts["a"]); ok {
+ t.Fatal("a sleeping instance boots its installed build on its next wake; there is nothing to swap")
+ }
+}