amadan

carlos / platform Public

zdd-5 task 11: the scheduler's worker path and the recycle hook

execute sends a sidecar's attempt to ReplaceWorker with the good build as
its fallback, then releases the held publication after done().
CoalesceRecycle skips a recycle while the host's attempt runs, or admits
a queued release attempt when admit would start it now.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 3d7d7eadd82872c2889cc3f51a4ca7a4704cadb9 parent 0d3469a
5 files changed, +269 −1
  • internal/rollout/fake_test.go +13 −0
  • internal/rollout/manager.go +4 −1
  • internal/rollout/scheduler.go +9 −0
  • internal/rollout/worker.go +77 −0
  • internal/rollout/worker_test.go +166 −0
diff --git a/internal/rollout/fake_test.go b/internal/rollout/fake_test.go
index 686ababe8..3a96488b6 100644
--- a/internal/rollout/fake_test.go
+++ b/internal/rollout/fake_test.go
@@ -47,6 +47,19 @@ type fakeAct struct {
// reason, when set, is a failed Swap's diagnostic text in place of
// "health check failed".
reason string
+ // onHeldPublish, when set, runs on every PublishHeldServing: a test's
+ // view of the moment the scheduler releases a held worker publication.
+ onHeldPublish func(host string)
+}
+
+// PublishHeldServing runs onHeldPublish, when set.
+func (f *fakeAct) PublishHeldServing(host string) {
+ f.mu.Lock()
+ hook := f.onHeldPublish
+ f.mu.Unlock()
+ if hook != nil {
+ hook(host)
+ }
}
func newFakeAct() *fakeAct {
diff --git a/internal/rollout/manager.go b/internal/rollout/manager.go
index 43bef4ab6..c4b3997be 100644
--- a/internal/rollout/manager.go
+++ b/internal/rollout/manager.go
@@ -17,12 +17,15 @@ import (
// 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).
+// stop-then-start for workers and native sidecar members (spec §7.9), and
+// PublishHeldServing releases a worker fallback's serving identity once
+// done() has recorded the failure it falls back from (spec §8.1).
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
+ PublishHeldServing(host string)
}
// Routes is the part of *registry.Registry the scheduler reads and writes.
diff --git a/internal/rollout/scheduler.go b/internal/rollout/scheduler.go
index 1ec45b540..402be5e23 100644
--- a/internal/rollout/scheduler.go
+++ b/internal/rollout/scheduler.go
@@ -115,6 +115,15 @@ func (s *scheduler) execute(ctx context.Context, d dispatch) {
if d.Class == classAdopt {
d.FailedUnder = m.failedUnder(*rt, d.ID.Build)
}
+ if rt.IsSidecar() {
+ // Workers and native sidecar members keep stop-then-start on the
+ // same entry (spec §7.9). done() runs the cohort-failure transition
+ // synchronously; only then may a fallback's identity be published
+ // (spec §8.1), which the activator held back for this call.
+ s.done(d, s.executeWorker(ctx, *rt, a), a.Mode)
+ m.d.Act.PublishHeldServing(rt.Host)
+ return
+ }
m.emit(*rt, EvSwapping, "info", textSwapping(a.Build, a.Cause))
s.done(d, m.d.Act.Swap(ctx, a, s.newest), a.Mode)
}
diff --git a/internal/rollout/worker.go b/internal/rollout/worker.go
new file mode 100644
index 000000000..8ae9be6e7
--- /dev/null
+++ b/internal/rollout/worker.go
@@ -0,0 +1,77 @@
+package rollout
+
+import (
+ "context"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// executeWorker runs one resolved attempt on a socketless host (spec §6.5,
+// §7.9) through the activator's same-entry replacement. The host's good
+// build, by its retained identity, goes with it as the fallback for a failed
+// adoption in overlap mode; the activator decides whether it applies.
+//
+// An abandoned result (a staging refusal, a busy entry, a re-check that
+// refused) is deferred by done(), as a refused swap is: while the attempt is
+// still the host's newest it waits abandonDeferral, and one superseded by a
+// newer desire does not wait at all.
+func (s *scheduler) executeWorker(ctx context.Context, rt registry.Route, a activator.Attempt) activator.SwapResult {
+ m := s.m
+ var fallback *release.StagedIdentity
+ if g := rt.Good(); g.Version != "" && g.Digest != "" && g.Version != a.Build {
+ fallback = &release.StagedIdentity{Version: g.Version, Digest: g.Digest, Demand: g.Demand,
+ StagedPath: release.VersionPath(m.d.OptDir, release.RouteAccount(rt), g.Version, rt.App)}
+ }
+ m.emit(rt, EvSwapping, "info", textSwapping(a.Build, a.Cause))
+ return m.d.Act.ReplaceWorker(ctx, a, fallback, s.newest)
+}
+
+// admitOne is admit for one host, for the recycle hook: the host's pending
+// attempt, if it is a release-class attempt (an adoption) the queue would
+// start right now. It honours everything admit does for that host: its
+// waits, its cohort's canary and settle, a halt, and swap_concurrency. A
+// restart or config attempt is not taken: the recycle runs on the serving
+// build, and that attempt follows in its turn. The caller holds the
+// scheduler's lock.
+func (q *appQueue) admitOne(host string, now time.Time) (dispatch, bool) {
+ h := q.hosts[host]
+ if h == nil || h.running != nil || q.running >= q.concurrency {
+ return dispatch{}, false
+ }
+ d, ok := q.gate(h, now)
+ if !ok || d.Class != classAdopt {
+ return dispatch{}, false
+ }
+ q.start(h, d)
+ return d, true
+}
+
+// CoalesceRecycle is plan 4's Service.CoalesceRecycle (spec §7.9). The
+// activator calls it under its own lock (activator.Service.mu), so it calls
+// nothing in the activator and waits on nothing that could: it takes
+// Manager.mu (schedulerOf) and then the scheduler's lock, the documented
+// order, and the attempt it admits runs on its own goroutine. True means an
+// attempt makes the spawn and the recycle is skipped: one admitted now, or
+// one already running for the host (ReplaceWorker respawns that child
+// itself, and its observation reads a crash as the build's).
+func (m *Manager) CoalesceRecycle(host string) bool {
+ s := m.schedulerOf(host)
+ if s == nil {
+ return false
+ }
+ s.mu.Lock()
+ if h := s.q.hosts[host]; h != nil && h.running != nil {
+ s.mu.Unlock()
+ return true
+ }
+ d, ok := s.q.admitOne(host, m.d.Now())
+ s.mu.Unlock()
+ if !ok {
+ return false
+ }
+ go s.execute(m.ctx, d)
+ return true
+}
diff --git a/internal/rollout/worker_test.go b/internal/rollout/worker_test.go
new file mode 100644
index 000000000..7a462b266
--- /dev/null
+++ b/internal/rollout/worker_test.go
@@ -0,0 +1,166 @@
+package rollout
+
+import (
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// workerHarness turns the harness's hosts into always-on sidecars.
+func workerHarness(t *testing.T, hosts ...string) *harness {
+ h := newHarness(t, hosts...)
+ for _, host := range hosts {
+ r := h.rows.get(host)
+ r.Kind, r.Hibernate = "sidecar", false
+ h.rows.put(r)
+ }
+ return h
+}
+
+// Spec §7.9: a socketless attempt is ReplaceWorker on the same entry, never
+// Swap and never a stop and a wake. It carries the host's good build as its
+// fallback.
+func TestAWorkerReleaseReplacesOnTheSameEntry(t *testing.T) {
+ h := workerHarness(t, "bot")
+ h.stage("v2")
+ h.install("v2", "bot")
+ h.pass()
+ waitFor(t, "the good build to move to v2", func() bool { return h.rows.get("bot").GoodVersion == "v2" })
+ if calls, _, _ := h.act.snapshot(); len(calls) != 0 {
+ t.Fatal("a socketless attempt went through Swap")
+ }
+ r := h.act.replacements()
+ if len(r) != 1 || r[0].a.Build != "v2" || r[0].fallback == nil || r[0].fallback.Version != "v1" {
+ t.Fatalf("replacements %+v, want one onto v2 with the good v1 behind it", r)
+ }
+ if fb := r[0].fallback; fb.Digest != "sha-v1" || fb.StagedPath != release.VersionPath(h.opt, "bab", "v1", "blog") {
+ t.Fatalf("fallback %+v, want the good build by its retained identity", fb)
+ }
+}
+
+func TestAFailedWorkerReleaseFallsBackAndHalts(t *testing.T) {
+ h := workerHarness(t, "bot")
+ h.stage("v2")
+ h.act.mu.Lock()
+ h.act.worker = func(a activator.Attempt, fb *release.StagedIdentity) activator.SwapResult {
+ return activator.SwapResult{Outcome: activator.SwapFailed, Reason: "exited within 10s of starting"}
+ }
+ h.act.mu.Unlock()
+ h.install("v2", "bot")
+ h.pass()
+ waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
+ if r := h.rows.get("bot"); r.GoodVersion != "v1" {
+ t.Fatalf("good build = %q after a failed worker release", r.GoodVersion)
+ }
+}
+
+// Spec §8.1 (P5 review finding 7): the fallback's held publication is
+// released only after done() has recorded the failure it falls back from.
+func TestAWorkerFallbackIsPublishedAfterItsFailureIsRecorded(t *testing.T) {
+ h := workerHarness(t, "bot")
+ h.stage("v2")
+ recorded := make(chan bool, 1)
+ h.act.mu.Lock()
+ h.act.worker = func(a activator.Attempt, fb *release.StagedIdentity) activator.SwapResult {
+ return activator.SwapResult{Outcome: activator.SwapFailed, Reason: "exited within 10s of starting"}
+ }
+ h.act.onHeldPublish = func(string) {
+ _, ok := h.book.Lookup("bab", "blog", "v2")
+ select {
+ case recorded <- ok:
+ default:
+ }
+ }
+ h.act.mu.Unlock()
+ h.install("v2", "bot")
+ h.pass()
+ select {
+ case ok := <-recorded:
+ if !ok {
+ t.Fatal("the fallback's publication was released before v2's failure was recorded")
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("the held publication was never released")
+ }
+}
+
+// Spec §7.9: a recycle that falls due coalesces with a queued release attempt
+// only when the scheduler would admit that attempt now.
+func TestARecycleCoalescesOnlyWithAnAdmissibleRelease(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b")
+ releaseAll(q, "v2", "a", "b")
+ canary, ok := q.admitOne("a", t0)
+ if !ok || !canary.Canary {
+ t.Fatalf("admitOne(a) = %+v %v, want a's canary", canary, ok)
+ }
+ if _, ok := q.admitOne("b", t0); ok {
+ t.Fatal("coalesced past swap_concurrency 1 while another worker's canary runs")
+ }
+ q.finish(canary, activator.SwapDone)
+ if d, ok := q.admitOne("b", t0); !ok || d.ID.Build != "v2" {
+ t.Fatalf("admitOne(b) = %+v %v after the canary settled", d, ok)
+ }
+ r := awakeQueue(noBlocks, "c")
+ r.wantRestart("c", "r1", t0)
+ if _, ok := r.admitOne("c", t0); ok {
+ t.Fatal("a restart is not a release attempt; the recycle runs on the serving build")
+ }
+ h := newHarness(t, "a")
+ if h.m.CoalesceRecycle("nobody") {
+ t.Fatal("an unknown host coalesced")
+ }
+}
+
+// Spec §7.9: while an attempt runs for the host, the recycle is skipped:
+// that attempt's ReplaceWorker makes the spawn. A crash during its
+// observation is the build's to report, never a recycle's to paper over.
+func TestARecycleIsSkippedWhileTheHostsAttemptRuns(t *testing.T) {
+ h := workerHarness(t, "bot")
+ h.stage("v2")
+ running := make(chan struct{})
+ proceed := make(chan struct{})
+ h.act.mu.Lock()
+ h.act.worker = func(activator.Attempt, *release.StagedIdentity) activator.SwapResult {
+ close(running)
+ <-proceed
+ return activator.SwapResult{Outcome: activator.SwapDone}
+ }
+ h.act.mu.Unlock()
+ h.install("v2", "bot")
+ h.pass()
+ select {
+ case <-running:
+ case <-time.After(5 * time.Second):
+ t.Fatal("the attempt never started")
+ }
+ if !h.m.CoalesceRecycle("bot") {
+ t.Fatal("a recycle went ahead while the host's attempt was running")
+ }
+ close(proceed)
+ waitFor(t, "the attempt to finish", func() bool { return h.rows.get("bot").GoodVersion == "v2" })
+ if h.m.CoalesceRecycle("bot") {
+ t.Fatal("coalesced with nothing queued or running")
+ }
+}
+
+// Plan 4's contract: CoalesceRecycle is called under the activator's lock,
+// so it must never call back into the activator. Here the fake's own lock
+// stands in for activator.Service.mu, held across the call: a callback into
+// Current, Owns or ReplaceWorker would never return.
+func TestCoalesceRecycleNeverCallsTheActivator(t *testing.T) {
+ h := workerHarness(t, "bot")
+ h.pass() // the host is known to its scheduler
+ waitFor(t, "the host's scheduler", func() bool { return h.m.schedulerOf("bot") != nil })
+ answered := make(chan bool, 1)
+ h.act.mu.Lock()
+ go func() { answered <- h.m.CoalesceRecycle("bot") }()
+ select {
+ case <-answered:
+ case <-time.After(2 * time.Second):
+ h.act.mu.Unlock()
+ t.Fatal("CoalesceRecycle waited on the activator")
+ }
+ h.act.mu.Unlock()
+}