amadan

carlos / platform Public

zdd-readiness: worker gates after cgroup preparation, managed revalidation at dispatch, no uncaptured live-link recycle

Astra R3 §5 item 4 (ruling 206, park worker launch hook). A worker start's ownership gate rides the context as the exec gate and is the cgroup manager's before-start hook (launchGate), asked after the cgroup's preparation and before each start attempt, the PID-move retry included; the before-exec hook keeps only the build's publication. SplitBackend asks the gate before a non-staged StartWorker. scheduler.execute revalidates the row it reads (Managed, account, app) before the good-build gate and the intent: a row that left the scheduler is abandoned unexecuted. With CoalesceRecycle set, a worker with no captured build is never respawned from the live link: a live child keeps running, a dead one stays down, logged once per host.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 7daf32acbb4649fb36cc1030185d4bd9128c9d9c parent c453a79
10 files changed, +358 −30
  • internal/activator/activator.go +11 −4
  • internal/activator/backend_exec.go +13 −3
  • internal/activator/backend_unit.go +7 −1
  • internal/activator/exec_gate_test.go +98 −0
  • internal/activator/genbackend.go +13 −10
  • internal/activator/worker_identity_test.go +73 −0
  • internal/activator/worker_staged_test.go +5 −0
  • internal/activator/workerbuild.go +43 −12
  • internal/rollout/manager_test.go +75 −0
  • internal/rollout/scheduler.go +20 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index f5246fcb1..1abf6561e 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -956,6 +956,12 @@ type Service struct {
// converges (recycleWorkerLocked), so the two share one pace. Under a.mu.
convergeBudget int
+ // uncapturedSaid is the hosts whose uncaptured worker a recycle has
+ // declined to respawn from the live link (recycleWorkerLocked), so the
+ // agent says so once per host and not at every backoff or sweep. Under
+ // a.mu.
+ uncapturedSaid map[string]bool
+
fences map[string]uint64
tokenPreparations map[string]chan struct{}
stopHistory map[string]*stopObservation
@@ -2819,10 +2825,11 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
// F1): the config admission, the materialization and the
// build's publication all come between the check above and the
// exec, and the lease can be lost, a fence set or the entry
- // stopped in that time. startWorker composes it into the
- // before-exec hook, after the publication. An attempt's checks
- // ride on it (check): a supersession in that time refuses the
- // exec too.
+ // stopped in that time. startWorker hands it to the backend as
+ // the exec gate, asked after the publication and the cgroup's
+ // preparation and before each start attempt (Astra R3 §5 item
+ // 4). An attempt's checks ride on it (check): a supersession in
+ // that time refuses the exec too.
gate := func() error {
if err := ctx.Err(); err != nil {
return err
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 67a0ae771..2da660a85 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -2111,7 +2111,11 @@ func (b *ExecBackend) StartWorker(ctx context.Context, rt registry.Route, extra
//
// Between resolving that identity and the exec it runs ctx's before-exec hook
// (runBeforeExec), through which the activator publishes the build first; a
-// hook error means nothing runs (ruling R19 item 3).
+// hook error means nothing runs (ruling R19 item 3). ctx's exec gate
+// (withExecGate), the caller's ownership check, is the cgroup manager's
+// before-start hook (launchGate), as a flat instance spawn's is: asked after
+// the cgroup's preparation and before each start attempt, and its refusal
+// starts nothing and comes back as it is (Astra R3 §5 item 4).
func (b *ExecBackend) StartWorkerStaged(ctx context.Context, rt registry.Route, extra map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) {
var argv []string
if err := json.Unmarshal([]byte(rt.Args), &argv); err != nil || len(argv) == 0 {
@@ -2154,12 +2158,18 @@ func (b *ExecBackend) StartWorkerStaged(ctx context.Context, rt registry.Route,
// of a tenant's app like any other. Bypassing it compiles and silently
// ships an uncapped process — the exact failure spawnInCgroup's own
// comment names.
- cmd, settle, err := b.spawnInCgroup(rt, func() *exec.Cmd {
+ //
+ // Gated (spawnInCgroupGated): ownership can be lost while the cgroup is
+ // prepared or between a failed clone3 start and its PID-move retry, so
+ // the caller's gate is asked at each start, not once before all of it.
+ // With no gate on ctx (an unplanned wake's spawn passes none) launchGate
+ // is nil and this is spawnInCgroup exactly.
+ cmd, settle, err := b.spawnInCgroupGated(rt, func() *exec.Cmd {
cmd := exec.Command(id.StagedPath, argv...)
cmd.Env = childEnv(fileEnv)
b.teeStd(cmd, rt)
return cmd
- })
+ }, b.launchGate(ctx, rt.Host))
if err != nil {
return nil, release.StagedIdentity{}, err
}
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go
index 57a073eb7..8c905f2a4 100644
--- a/internal/activator/backend_unit.go
+++ b/internal/activator/backend_unit.go
@@ -385,11 +385,17 @@ func (b *SplitBackend) StartWorker(ctx context.Context, rt registry.Route, env m
// StartWorkerStaged forwards spec §7.9's staged spawn to the backend the
// route picks. One without the capability (UnitBackend, which refuses
// workers anyway) gets today's StartWorker, and its identity is the version
-// alone, which no recycle can be verified against.
+// alone, which no recycle can be verified against. That one cannot ask ctx's
+// exec gate at its start, so it is asked here, immediately before
+// StartWorker, as startWorker asks it for a backend without the capability
+// (Astra R3 §5 item 4).
func (b *SplitBackend) StartWorkerStaged(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) {
if s, ok := b.pick(rt).(stagedWorkerStarter); ok {
return s.StartWorkerStaged(ctx, rt, env, onto)
}
+ if err := runExecGate(ctx); err != nil {
+ return nil, release.StagedIdentity{}, err
+ }
p, ver, err := b.pick(rt).StartWorker(ctx, rt, env)
return p, release.StagedIdentity{Version: ver}, err
}
diff --git a/internal/activator/exec_gate_test.go b/internal/activator/exec_gate_test.go
index b3c44f110..6919c6ab2 100644
--- a/internal/activator/exec_gate_test.go
+++ b/internal/activator/exec_gate_test.go
@@ -406,6 +406,104 @@ func gatedCgroups(t *testing.T, b *ExecBackend, host string, flip func()) string
return parent
}
+// workerGateService is a Service over a REAL ExecBackend worker spawn
+// (realWorkerBackend) whose cgroup manager is gatedCgroups' temp-dir
+// subtree: a worker start runs the production preparation (the build's
+// publication, the cgroup and its limits) and its start attempts. It
+// returns the worker's row, the file its build writes its $0 to the moment
+// it runs, and the host's cgroup directory.
+func workerGateService(t *testing.T, flip func()) (*Service, *ExecBackend, registry.Route, string, string) {
+ t.Helper()
+ eb := testExecBackend(t)
+ host := "bot.example.com"
+ out := filepath.Join(t.TempDir(), "argv0")
+ stageArgv0Worker(t, eb, host, "v1", out)
+ dir := gatedCgroups(t, eb, host, flip)
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, &realWorkerBackend{fakeBackend: &fakeBackend{}, exec: eb}, reg)
+ a.GenerationsFile = release.GenerationsPath(t.TempDir())
+ t.Cleanup(func() { a.Shutdown(context.Background()) })
+ return a, eb, workerRoute(host), out, dir
+}
+
+// Astra R3 §5 item 4 (ruling 206, "park worker launch hook"): a worker
+// start's ownership gate is the cgroup manager's before-start hook, as an
+// instance's is, so it is asked AFTER the cgroup preparation (the host's
+// cgroup and its limits) and before the start. Ownership lost during that
+// preparation starts nothing: no process, no $0 written, nothing placed in
+// the cgroup, the gate's own error back, and the build's pending
+// publication dropped. Before the fix the gate rode the before-exec hook,
+// asked once before the cgroup preparation, and the worker started.
+func TestAWorkerStartIsGatedAfterItsCgroupPreparation(t *testing.T) {
+ var refuse atomic.Bool
+ a, _, rt, out, dir := workerGateService(t, func() { refuse.Store(true) })
+ lost := errors.New("lease renewal failed")
+ var asked atomic.Int32
+ gate := func() error {
+ asked.Add(1)
+ if refuse.Load() {
+ return lost
+ }
+ return nil
+ }
+ p, _, _, err := a.startWorker(context.Background(), rt, nil, nil, gate)
+ if p != nil {
+ _ = p.Stop(context.Background())
+ }
+ if !errors.Is(err, lost) || p != nil {
+ t.Fatalf("startWorker = %v, %v; want no process and the gate's refusal", p, err)
+ }
+ if !refuse.Load() || asked.Load() == 0 {
+ t.Fatalf("the cgroup preparation ran: %v; the gate was asked %d times", refuse.Load(), asked.Load())
+ }
+ time.Sleep(200 * time.Millisecond)
+ if _, err := os.Stat(out); err == nil {
+ t.Fatal("the worker ran after its ownership was lost during the cgroup preparation")
+ }
+ if b, err := os.ReadFile(filepath.Join(dir, "cgroup.procs")); err == nil && len(b) != 0 {
+ t.Fatalf("the worker's cgroup was given pid %s after a refusal at its start", b)
+ }
+ if got := readGenerations(t, a.GenerationsFile); len(got[rt.Host]) != 0 {
+ t.Fatalf("generations file %v names a build that never ran", got)
+ }
+}
+
+// The same gate is asked again before the PID-move retry that follows a
+// failed clone3 start (cgroup.StartGated's contract), so ownership lost
+// between the two starts the worker on neither. The temp-dir subtree is no
+// cgroupfs: the clone3-into-cgroup start fails and the manager retries.
+// Before the fix the worker's gate was never asked at a start attempt, and
+// the retry started it.
+func TestAWorkerStartRetryReasksTheGate(t *testing.T) {
+ a, eb, rt, out, _ := workerGateService(t, nil)
+ var asked atomic.Int32
+ eb.launchHook = func(string) { asked.Add(1) }
+ superseded := errors.New("the attempt was superseded")
+ gate := func() error {
+ if asked.Load() >= 2 {
+ return superseded
+ }
+ return nil
+ }
+ p, _, _, err := a.startWorker(context.Background(), rt, nil, nil, gate)
+ if p != nil {
+ _ = p.Stop(context.Background())
+ }
+ if n := asked.Load(); n != 2 {
+ t.Fatalf("the gate was asked %d times at the starts, want twice: the clone3 start and its retry", n)
+ }
+ if !errors.Is(err, superseded) || p != nil {
+ t.Fatalf("startWorker = %v, %v; want no process and the gate's refusal", p, err)
+ }
+ time.Sleep(200 * time.Millisecond)
+ if _, err := os.Stat(out); err == nil {
+ t.Fatal("the retry started the worker after its gate refused")
+ }
+}
+
// Final review wave 2 (F1 completion, Astra round 2 finding 1): a flat
// staged spawn's exec gate is the cgroup manager's before-start hook, so
// it is asked AFTER the cgroup preparation (the parent's tier, the child
diff --git a/internal/activator/genbackend.go b/internal/activator/genbackend.go
index 02aadb39d..fd9e0fd51 100644
--- a/internal/activator/genbackend.go
+++ b/internal/activator/genbackend.go
@@ -62,19 +62,21 @@ type SwapBackend interface {
GenSwitched(rt registry.Route, g Gen, p Proc) error
}
-// execGateKey carries a staged instance spawn's exec gate on its context
+// execGateKey carries a staged spawn's exec gate on its context
// (withExecGate).
type execGateKey struct{}
// withExecGate returns ctx carrying gate, the caller's composed ownership
-// and attempt checks for a staged instance spawn (final review F1): a
-// planned wake's B or its fallback, a hold's pinned build, an overlap
-// successor. The backend asks it at the very end of its preparation, right
-// before the exec (flat) or the root start call (nested), so a lease lost,
-// a fence set or an attempt superseded while the backend read config,
-// bound sockets or wrote launch files execs nothing. It rides on the
-// context, as withBeforeExec does for workers, so StartStaged's signature
-// and every forwarding backend (SplitBackend) stay as they are.
+// and attempt checks for a staged spawn (final review F1): a planned wake's
+// B or its fallback, a hold's pinned build, an overlap successor, and every
+// worker start that has a gate (startWorker; Astra R3 §5 item 4). The
+// backend asks it at the very end of its preparation, right before the exec
+// (flat, and a worker) or the root start call (nested), so a lease lost, a
+// fence set or an attempt superseded while the backend read config, bound
+// sockets, wrote launch files or prepared the cgroup execs nothing. It rides
+// on the context, as withBeforeExec does, so StartStaged's and
+// StartWorkerStaged's signatures and every forwarding backend
+// (SplitBackend) stay as they are.
func withExecGate(ctx context.Context, gate func() error) context.Context {
if gate == nil {
return ctx
@@ -97,7 +99,8 @@ func runExecGate(ctx context.Context) error {
return gate()
}
-// launchGate is a flat staged spawn's before-start hook for the cgroup
+// launchGate is a flat staged spawn's (an instance generation's or a
+// worker's) before-start hook for the cgroup
// manager (final review F1): launchHook (the test seam), then ctx's exec
// gate, asked immediately before each start attempt, after the cgroup's
// preparation. Nil when ctx carries no gate and no test seam is set, so a
diff --git a/internal/activator/worker_identity_test.go b/internal/activator/worker_identity_test.go
index bb48fc049..eef3d1d80 100644
--- a/internal/activator/worker_identity_test.go
+++ b/internal/activator/worker_identity_test.go
@@ -3,7 +3,10 @@ package activator
import (
"errors"
"fmt"
+ "log"
+ "os"
"path/filepath"
+ "strings"
"sync/atomic"
"testing"
"time"
@@ -431,6 +434,76 @@ func TestWorkerRecycleNeverConvergesWhenTheSchedulerIsPresent(t *testing.T) {
}
}
+// Astra R3 §5 item 4: on a box whose scheduler (or its refusal) owns
+// recycles (CoalesceRecycle set), a worker with no captured build is never
+// respawned from the live link, which names whatever root installed last:
+// a build no admission chose. A live child whose max_run falls due keeps
+// running; a crashed one stays down, and the agent says so once for the
+// host, not at every backoff. A release attempt (ReplaceWorker) or a wake
+// is what starts it. Before the fix both respawned from live/<host>.
+func TestAnArmedBoxNeverRecyclesAnUncapturedWorkerFromTheLiveLink(t *testing.T) {
+ const said = "has no captured build, and the release scheduler owns its recycles"
+ capture := func(t *testing.T) *syncLog {
+ logs := &syncLog{}
+ flags := log.Flags()
+ log.SetOutput(logs)
+ t.Cleanup(func() { log.SetOutput(os.Stderr); log.SetFlags(flags) })
+ return logs
+ }
+ t.Run("crashed", func(t *testing.T) {
+ a, b, clk, _ := testActivator(t, time.Hour) // no staged capability: nothing is captured
+ host := "bot.x.internal"
+ a.CoalesceRecycle = func(string) bool { return false }
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ if got := a.entryServing(host); got.StagedPath != "" {
+ t.Fatalf("serving %+v, want no captured build", got)
+ }
+ logs := capture(t)
+ b.lastWorker().kill()
+ for range 3 {
+ a.sweepOnce()
+ clk.advance(sidecarRestartMax + time.Second)
+ }
+ time.Sleep(100 * time.Millisecond)
+ if n := b.workerStarts(); n != 1 || a.entryRespawning(host) {
+ t.Fatalf("workerStarts = %d (respawning %v): an uncaptured worker was respawned from the live link", n, a.entryRespawning(host))
+ }
+ if !a.Owns(host) {
+ t.Fatal("the entry went: it stays, down, for a release attempt or a wake to start")
+ }
+ if n := strings.Count(logs.String(), said); n != 1 {
+ t.Fatalf("said so %d times in %q, want once for the host", n, logs.String())
+ }
+ })
+ t.Run("max_run", func(t *testing.T) {
+ a, b, clk, _ := testActivator(t, time.Hour)
+ host := "bot.x.internal"
+ a.CoalesceRecycle = func(string) bool { return false }
+ rt := alwaysOnWorkerRoute(host)
+ rt.MaxRunSeconds = 30
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ logs := capture(t)
+ for range 3 {
+ clk.advance(31 * time.Second)
+ a.sweepOnce()
+ }
+ time.Sleep(100 * time.Millisecond)
+ if n := b.workerStarts(); n != 1 || a.entryRespawning(host) {
+ t.Fatalf("workerStarts = %d (respawning %v): an uncaptured worker was recycled from the live link", n, a.entryRespawning(host))
+ }
+ if b.lastWorker().Exited() {
+ t.Fatal("the running child was stopped by a recycle that could not respawn it")
+ }
+ if n := strings.Count(logs.String(), said); n != 1 {
+ t.Fatalf("said so %d times in %q, want once for the host", n, logs.String())
+ }
+ })
+}
+
// Ruling R13: an exited worker skips the sweep's drift check, so the crash
// restart measures the drift itself, and a crash loop converges once the
// grace has passed.
diff --git a/internal/activator/worker_staged_test.go b/internal/activator/worker_staged_test.go
index a1ee44bb4..830629f63 100644
--- a/internal/activator/worker_staged_test.go
+++ b/internal/activator/worker_staged_test.go
@@ -58,6 +58,11 @@ func (b *stagedFake) StartWorkerStaged(ctx context.Context, rt registry.Route, e
if err := runBeforeExec(ctx, id); err != nil {
return nil, release.StagedIdentity{}, err
}
+ // And the caller's exec gate at the start, after the publication (Astra
+ // R3 §5 item 4): ExecBackend asks it at the cgroup manager's start.
+ if err := runExecGate(ctx); err != nil {
+ return nil, release.StagedIdentity{}, err
+ }
p, ver, err := b.fakeBackend.StartWorker(ctx, rt, env)
if err != nil {
return p, release.StagedIdentity{}, err
diff --git a/internal/activator/workerbuild.go b/internal/activator/workerbuild.go
index df2758e4e..9c1e75127 100644
--- a/internal/activator/workerbuild.go
+++ b/internal/activator/workerbuild.go
@@ -18,6 +18,10 @@ import (
// captured identity to recycle onto. An implementation runs ctx's
// before-exec hook (runBeforeExec) once it has resolved the identity and
// before it execs, and does not exec when the hook fails (ruling R19 item 3).
+// It asks ctx's exec gate (withExecGate) as StartStaged does: after all of
+// its preparation, the cgroup's included, immediately before each start
+// attempt, and when the gate refuses starts nothing and returns its error
+// (Astra R3 §5 item 4).
type stagedWorkerStarter interface {
StartWorkerStaged(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error)
}
@@ -56,14 +60,20 @@ type beforeExecKey struct{}
// between resolving the identity it will exec and the exec itself (ruling R19
// item 3). It rides on the context so StartWorkerStaged's signature, and the
// stagedWorkerStarter interface, stay as they are.
+//
+// The hook is the build's publication only. The caller's ownership gate is
+// not in it: it rides the context as an instance spawn's does
+// (withExecGate), because the hook runs before the cgroup's preparation and
+// once, where ownership has to be asked after that preparation and before
+// every start attempt (Astra R3 §5 item 4, ruling 206).
func withBeforeExec(ctx context.Context, f func(release.StagedIdentity) error) context.Context {
return context.WithValue(ctx, beforeExecKey{}, f)
}
// runBeforeExec runs ctx's before-exec hook, if any, with the identity about
// to be exec'd. A backend that implements stagedWorkerStarter calls it after
-// resolving that identity and before the exec, and does not exec when it
-// returns an error.
+// resolving that identity and before its spawn's preparation, and does not
+// exec when it returns an error.
func runBeforeExec(ctx context.Context, id release.StagedIdentity) error {
if f, ok := ctx.Value(beforeExecKey{}).(func(release.StagedIdentity) error); ok && f != nil {
return f(id)
@@ -84,11 +94,16 @@ func runBeforeExec(ctx context.Context, id release.StagedIdentity) error {
// identity, or after shedding the child, and then publishes again.
//
// gate, when set, is the caller's ownership check at the exec (final review
-// F1): composed into the before-exec hook after the publication, so it is
-// asked once the preparation (materialization, the publication) is done
-// and immediately before the exec, or, on a backend without the staged
-// capability, immediately before StartWorker. Its refusal execs nothing
-// and is returned as is.
+// F1). Through the staged capability it rides the context as the exec gate
+// (withExecGate), and the backend asks it as an instance spawn's: after the
+// materialization, the publication and the cgroup's preparation, and before
+// EACH start attempt, the PID-move retry after a failed clone3 start
+// included (Astra R3 §5 item 4, ruling 206). In the before-exec hook, as it
+// was, it was asked once and before the cgroup was prepared, so ownership
+// lost during that preparation, or between a failed start and its retry,
+// still started the worker. On a backend without the staged capability it
+// is asked immediately before StartWorker. Its refusal execs nothing and is
+// returned as is.
func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity, gate func() error) (Proc, release.StagedIdentity, pendingGeneration, error) {
s, ok := a.backend.(stagedWorkerStarter)
if !ok {
@@ -112,12 +127,9 @@ func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[st
log.Printf("activator: worker %s not spawned: the build it would exec (%s %s) could not be published first: %v", rt.Host, id.Version, id.StagedPath, err)
return fmt.Errorf("%w: %v", ErrBuildNotPublished, err)
}
- if gate != nil {
- return gate()
- }
return nil
}
- p, id, err := s.StartWorkerStaged(withBeforeExec(ctx, hook), rt, env, onto)
+ p, id, err := s.StartWorkerStaged(withBeforeExec(withExecGate(ctx, gate), hook), rt, env, onto)
if err != nil && pend != 0 {
a.mu.Lock()
a.dropPendingGenerationLocked(pend)
@@ -137,7 +149,13 @@ func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[st
// CoalesceRecycle is asked first. When it says a queued release attempt is
// admissible now, that attempt makes the one spawn and the recycle is
// skipped. A worker spawned by a backend without the staged capability has
-// no captured build, and keeps today's respawn from the live link.
+// no captured build. With no scheduler (CoalesceRecycle nil) it keeps
+// today's respawn from the live link. With one, armed or refusing, it never
+// does (Astra R3 §5 item 4): the live link names whatever root installed
+// last, a build no admission chose, and on a refusing box possibly one the
+// unreadable failure book holds down. So a live child (stopOld, a max_run
+// recycle) is left running, and a dead one stays down until a release
+// attempt or a wake starts it. The agent says so once per host.
//
// Then the drift (ruling R13). Until plan 5 queues it as a release attempt,
// the converge is the one path that moves a worker onto an installed build,
@@ -181,6 +199,19 @@ func (a *Service) recycleWorkerLocked(e *actEntry, stopOld Proc, why string) {
}
id := e.workerServing
onto = &id
+ } else if a.CoalesceRecycle != nil {
+ if !a.uncapturedSaid[e.host] {
+ if a.uncapturedSaid == nil {
+ a.uncapturedSaid = map[string]bool{}
+ }
+ a.uncapturedSaid[e.host] = true
+ left := "the running child keeps running"
+ if stopOld == nil {
+ left = "it stays down until a release attempt or a wake starts it"
+ }
+ log.Printf("activator: worker %s %s; it has no captured build, and the release scheduler owns its recycles, so it is not respawned from live/%s: %s", e.host, why, e.host, left)
+ }
+ return
} else {
log.Printf("activator: worker %s %s; no captured build to recycle onto, so it respawns from live/%s", e.host, why, e.host)
}
diff --git a/internal/rollout/manager_test.go b/internal/rollout/manager_test.go
index 78b4d9f26..6ea31e715 100644
--- a/internal/rollout/manager_test.go
+++ b/internal/rollout/manager_test.go
@@ -2,6 +2,8 @@ package rollout
import (
"errors"
+ "fmt"
+ "strings"
"sync"
"sync/atomic"
"testing"
@@ -935,6 +937,79 @@ func TestMemberEventsCarryOnlyTheReasonWord(t *testing.T) {
}
}
+// Astra R3 §5 item 4 (execution admission): an attempt is admitted from the
+// pass's view of a row, and the row may change before the scheduler runs it.
+// execute revalidates the row it reads: a row that has become a named-unit
+// route, or that now belongs to another app, is no longer this scheduler's,
+// so the attempt ends abandoned with nothing run (no Swap, no ReplaceWorker,
+// no record). Before the fix the attempt went ahead on the changed row.
+func TestARowThatLeavesTheSchedulerBeforeDispatchIsAbandoned(t *testing.T) {
+ for _, worker := range []bool{false, true} {
+ for name, change := range map[string]func(*registry.Route){
+ "a named-unit route": func(r *registry.Route) { r.Unit = "blog.service" },
+ "another app": func(r *registry.Route) { r.App = "shop" },
+ } {
+ kind := "socket instance"
+ if worker {
+ kind = "worker"
+ }
+ t.Run(kind+"/"+name, func(t *testing.T) {
+ var h *harness
+ if worker {
+ h = workerHarness(t, "h")
+ } else {
+ h = newHarness(t, "h")
+ }
+ var (
+ mu sync.Mutex
+ abandoned []string
+ )
+ h.m.d.Logf = func(format string, args ...any) {
+ if line := fmt.Sprintf(format, args...); strings.Contains(line, "abandoned") {
+ mu.Lock()
+ abandoned = append(abandoned, line)
+ mu.Unlock()
+ }
+ }
+ h.stage("v2")
+ h.install("v2", "h")
+ // The row changes between the pass that queued the attempt and
+ // the scheduler's read of it.
+ h.rows.mu.Lock()
+ h.rows.getHook = func(host string) {
+ r := h.rows.get(host)
+ change(&r)
+ h.rows.put(r)
+ }
+ h.rows.mu.Unlock()
+ moves := func() int {
+ calls, _, _ := h.act.snapshot()
+ return len(calls) + len(h.act.replacements())
+ }
+ h.pass()
+ waitFor(t, "the attempt to end", func() bool {
+ mu.Lock()
+ n := len(abandoned)
+ mu.Unlock()
+ return (n > 0 || moves() > 0) && !h.m.attemptRunning("h")
+ })
+ if n := moves(); n != 0 {
+ t.Fatalf("%d Swap/ReplaceWorker calls on a row that left the scheduler, want none", n)
+ }
+ mu.Lock()
+ got := abandoned[0]
+ mu.Unlock()
+ if !strings.Contains(got, "route no longer managed") {
+ t.Fatalf("abandoned as %q, want the route no longer managed", got)
+ }
+ if recs := h.book.List(); len(recs) != 0 {
+ t.Fatalf("records %+v for an attempt that never ran", recs)
+ }
+ })
+ }
+ }
+}
+
func TestReasonWords(t *testing.T) {
for text, want := range map[string]string{
"health check failed": status.ReasonHealth,
diff --git a/internal/rollout/scheduler.go b/internal/rollout/scheduler.go
index 37e442a3c..4d102226f 100644
--- a/internal/rollout/scheduler.go
+++ b/internal/rollout/scheduler.go
@@ -9,6 +9,7 @@ import (
"github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
)
// scheduler is one app's goroutine (spec §6.1). It runs off the agent's
@@ -103,6 +104,25 @@ func (s *scheduler) execute(ctx context.Context, d dispatch) {
s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: "route not readable"}, "")
return
}
+ if !Managed(*rt) || release.RouteAccount(*rt) != s.key.acct || rt.App != s.key.app {
+ // Execution admission (Astra R3 §5 item 4): the attempt was queued
+ // from the pass's view of the row, and the row this execution runs
+ // on may have changed since. One that has become a named-unit or
+ // unit-backed route belongs to root's adoption or to systemd, and
+ // one that now names another app or account is another scheduler's.
+ // Either way it is not this attempt's to run. Managed also wants the
+ // row ready, so one passing through draining, moving or landing is
+ // not moved now either; the queue keeps it (heldOnBox) and derives
+ // the attempt again after the deferral. Checked before the
+ // good-build gate, the staging read and the write-ahead intent, so
+ // nothing is written and nothing runs: abandoned unexecuted, and
+ // done() defers the host and re-arms a retry allowance. A row that
+ // stays unmanaged, or another app's, leaves this queue at the next
+ // pass (forget).
+ m.d.Logf("rollout: %s: its %s attempt onto %s is no longer this scheduler's (%s/%s): the route is not a managed route of the app now; not running it", d.ID.Host, d.ID.Cause, d.ID.Build, s.key.acct, s.key.app)
+ s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: "route no longer managed"}, "")
+ return
+ }
if movesBuild(d.ID.Cause) && rt.GoodVersion != "" && d.ID.Build == rt.GoodVersion {
if mark, ok := holdRanFor(m.d.Book, s.key.acct, s.key.app, d.ID.Host); ok && mark.Build != d.ID.Build {
// Ruling 205 at the scheduler (Task 3 fix round 1): a move onto