amadan

carlos / platform Public

zdd-5 final review: report whether an attempt's build executed (F8, activator half)

SwapResult.Executed is the attempt's execution evidence: an overlap
successor's process, a hold's pinned wake (pinExec, or a healthy wake),
and a worker respawn's attemptExec (F2). The scheduler re-arms a host
retry allowance only on an abandonment with nothing executed. Every
existing SwapResult literal still compiles: the field defaults to false.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude fa954e608f243ff2f5abfaea3303e6d151e15c43 parent 4dc989d
4 files changed, +157 −5
  • internal/activator/swap.go +15 −1
  • internal/activator/swap_executed_test.go +128 −0
  • internal/activator/swaphold.go +6 −2
  • internal/activator/workerswap.go +8 −2
diff --git a/internal/activator/swap.go b/internal/activator/swap.go
index 83b9f1fac..9d9046073 100644
--- a/internal/activator/swap.go
+++ b/internal/activator/swap.go
@@ -78,6 +78,14 @@ type SwapResult struct {
// hold (spec §2.9): the scheduler records a failure under this mode, so
// a failed hold is held down (spec §6.4, §8.3). Swap sets it.
Mode SwapMode
+ // Executed is the attempt's execution evidence (final review F8): true
+ // once a process of the attempt's build started, whatever happened to
+ // it after (an overlap successor, a hold's pinned wake, a worker's
+ // respawn child, installed or shed). False means nothing of the build
+ // ran, so an abandoned attempt spent nothing of what it was granted:
+ // the scheduler re-arms a `carlos instances retry` allowance on it.
+ // Swap and ReplaceWorker set it; every other result leaves it false.
+ Executed bool
}
var (
@@ -333,7 +341,12 @@ func (a *Service) spawnSuccessor(ctx context.Context, gb SwapBackend, e *actEntr
// leaveSet restores a raised parent when the last member leaves. Nothing
// here blocks under a.mu except GenSwitched, which plan 1's lock contract
// puts in the same critical section as e.gen.
-func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, newest func(string) uint64) SwapResult {
+func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, newest func(string) uint64) (out SwapResult) {
+ // The execution evidence (SwapResult.Executed): the backend returned a
+ // successor process, so the attempt's build ran, whatever ends the
+ // attempt after that.
+ var executed bool
+ defer func() { out.Executed = executed }()
// The bytes first. An artifact that is not what the attempt chose never
// reaches admission, let alone exec (spec §4.2, §7.5 step 3).
if err := verifyStaged(at.StagedPath, at.Digest); err != nil {
@@ -466,6 +479,7 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n
// Step 3: exec the attempt's staged build by path. The backend reports
// what it ran, and anything but the attempt's choice is refused.
g, proc, spawnErr := a.spawnSuccessor(mctx, gb, e, rt, work, m.gen, gate)
+ executed = proc != nil
if spawnErr == nil && (g.N != m.gen.N || g.Build != m.gen.Build || g.StagedPath != m.gen.StagedPath) {
spawnErr = fmt.Errorf("backend ran generation %d of %s from %q; the attempt chose %d of %s from %q",
g.N, g.Build, g.StagedPath, m.gen.N, m.gen.Build, m.gen.StagedPath)
diff --git a/internal/activator/swap_executed_test.go b/internal/activator/swap_executed_test.go
new file mode 100644
index 000000000..7bc4f060a
--- /dev/null
+++ b/internal/activator/swap_executed_test.go
@@ -0,0 +1,128 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// Final review F8: SwapResult.Executed says whether a process of the
+// attempt's build started. The scheduler re-arms a `carlos instances retry`
+// allowance only for an attempt that ended before that (an abandonment with
+// Executed false); an attempt whose build ran spends it, whatever its
+// outcome. These cases cover each mode's refusals before the exec, its
+// abandonments after the exec, and its executed outcomes.
+func TestSwapResultReportsWhetherTheAttemptsBuildExecuted(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ mode SwapMode
+ arrange func(f *swapFixture)
+ outcome SwapOutcome
+ executed bool
+ }{
+ {"overlap, refused at the exec", ModeOverlap, func(f *swapFixture) {
+ f.b.onRaise = func() { f.rev.Add(1) }
+ }, SwapAbandoned, false},
+ {"overlap, the spawn fails", ModeOverlap, func(f *swapFixture) {
+ f.b.genErr = errors.New("exec format error")
+ }, SwapFailed, false},
+ {"overlap, superseded at the switch", ModeOverlap, func(f *swapFixture) {
+ f.a.genHealth = func(context.Context, Gen) error { f.rev.Add(1); return nil }
+ }, SwapAbandoned, true},
+ {"overlap, done", ModeOverlap, func(*swapFixture) {}, SwapDone, true},
+ {"hold, superseded before the exec", ModeHold, func(f *swapFixture) {
+ var once sync.Once
+ f.a.backend = &acquireHookBackend{fakeSwapBackend: f.b, hook: func() { once.Do(func() { f.rev.Add(1) }) }}
+ }, SwapAbandoned, false},
+ {"hold, superseded before the result", ModeHold, func(f *swapFixture) {
+ f.a.genHealth = func(context.Context, Gen) error { f.rev.Add(1); return nil }
+ }, SwapAbandoned, true},
+ {"hold, unhealthy", ModeHold, func(f *swapFixture) {
+ f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") }
+ }, SwapDown, true},
+ {"hold, done", ModeHold, func(*swapFixture) {}, SwapDone, true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ f := newSwapFixture(t)
+ tc.arrange(f)
+ res := f.a.Swap(context.Background(), f.attempt(tc.mode), f.newest)
+ if res.Outcome != tc.outcome {
+ t.Fatalf("Swap = %+v, want %s", res, tc.outcome)
+ }
+ if res.Executed != tc.executed {
+ t.Fatalf("Swap = %+v: Executed %v, want %v", res, res.Executed, tc.executed)
+ }
+ })
+ }
+}
+
+// The same for a worker's same-entry replacement, from the respawn's
+// execution evidence (final review F2's attemptExec).
+func TestReplaceWorkerReportsWhetherTheAttemptsBuildExecuted(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ mode SwapMode
+ arrange func(t *testing.T, a *Service, b *stagedFake, rev *atomic.Uint64)
+ outcome SwapOutcome
+ executed bool
+ }{
+ {"hold, superseded at its exec gate", ModeHold, func(_ *testing.T, _ *Service, b *stagedFake, rev *atomic.Uint64) {
+ var hooked atomic.Bool
+ b.mu.Lock()
+ b.admitHook = func() {
+ if hooked.CompareAndSwap(false, true) {
+ rev.Store(2)
+ }
+ }
+ b.mu.Unlock()
+ }, SwapAbandoned, false},
+ {"hold, superseded after its exec", ModeHold, func(_ *testing.T, _ *Service, b *stagedFake, rev *atomic.Uint64) {
+ var hooked atomic.Bool
+ b.mu.Lock()
+ b.workerStartHook = func(registry.Route) {
+ if hooked.CompareAndSwap(false, true) {
+ rev.Store(2)
+ }
+ }
+ b.mu.Unlock()
+ }, SwapAbandoned, true},
+ {"refused for its publication", ModeOverlap, func(t *testing.T, a *Service, _ *stagedFake, _ *atomic.Uint64) {
+ blocker := filepath.Join(t.TempDir(), "not-a-dir")
+ if err := os.WriteFile(blocker, nil, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ a.mu.Lock()
+ a.GenerationsFile = filepath.Join(blocker, "generations.json")
+ a.mu.Unlock()
+ }, SwapAbandoned, false},
+ {"done", ModeOverlap, func(*testing.T, *Service, *stagedFake, *atomic.Uint64) {}, SwapDone, true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ var rev atomic.Uint64
+ rev.Store(1)
+ tc.arrange(t, a, b, &rev)
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, tc.mode, fakeIdentity("v2")), nil,
+ func(string) uint64 { return rev.Load() })
+ if res.Outcome != tc.outcome {
+ t.Fatalf("ReplaceWorker = %+v, want %s", res, tc.outcome)
+ }
+ if res.Executed != tc.executed {
+ t.Fatalf("ReplaceWorker = %+v: Executed %v, want %v", res, res.Executed, tc.executed)
+ }
+ waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) })
+ })
+ }
+}
diff --git a/internal/activator/swaphold.go b/internal/activator/swaphold.go
index 2e6591912..90ad768ed 100644
--- a/internal/activator/swaphold.go
+++ b/internal/activator/swaphold.go
@@ -179,8 +179,11 @@ func (a *Service) withdrawOrTake(e *actEntry, h *holdSpec, why error) (holdWoke,
return <-h.woke, nil
}
-// holdResult follows a hold's pinned wake to its outcome.
-func (a *Service) holdResult(sctx context.Context, at Attempt, newest func(string) uint64, hw holdWoke, drain time.Duration) SwapResult {
+// holdResult follows a hold's pinned wake to its outcome. Its execution
+// evidence is the pinned wake's (pinExec, set from the spawn's result), and
+// a wake that came up healthy ran the build: SwapResult.Executed (final
+// review F8).
+func (a *Service) holdResult(sctx context.Context, at Attempt, newest func(string) uint64, hw holdWoke, drain time.Duration) (out SwapResult) {
if hw.err != nil {
return abandon(hw.err)
}
@@ -188,6 +191,7 @@ func (a *Service) holdResult(sctx context.Context, at Attempt, newest func(strin
a.mu.Lock()
g, werr, proc, execed := hw.next.gen, hw.next.err, hw.next.proc, hw.next.pinExec
a.mu.Unlock()
+ defer func() { out.Executed = execed || werr == nil }()
switch {
case werr != nil && !execed:
// The pinned wake failed before its build's exec: the lease, the
diff --git a/internal/activator/workerswap.go b/internal/activator/workerswap.go
index 7a646c1a3..970b539bd 100644
--- a/internal/activator/workerswap.go
+++ b/internal/activator/workerswap.go
@@ -36,7 +36,12 @@ import (
// The scheduler runs the cohort-failure transition from the result, then
// calls PublishHeldServing for a fallback's identity. newest is called with
// a.mu held, as Swap's is.
-func (a *Service) ReplaceWorker(ctx context.Context, at Attempt, fallback *release.StagedIdentity, newest func(host string) uint64) SwapResult {
+func (a *Service) ReplaceWorker(ctx context.Context, at Attempt, fallback *release.StagedIdentity, newest func(host string) uint64) (out SwapResult) {
+ // The execution evidence (SwapResult.Executed, final review F8): the
+ // attempt's respawn started a process of its build (attemptExec),
+ // installed or shed. Read once the respawn is over; false before.
+ var executed bool
+ defer func() { out.Executed = executed }()
if newest == nil {
return abandon(errSwapSuperseded)
}
@@ -97,7 +102,8 @@ func (a *Service) ReplaceWorker(ctx context.Context, at Attempt, fallback *relea
return abandon(errSwapStopped)
}
a.mu.Lock()
- veto, executed := e.attemptVeto, e.attemptExec
+ veto := e.attemptVeto
+ executed = e.attemptExec
e.attemptVeto, e.attemptExec = nil, false
if a.entries[at.Host] != e || e.state != actAwake {
a.mu.Unlock()