amadan

carlos / platform Public

zdd-5 final review w2: Executed means executed or unknown (N4, activator)

SwapResult.Executed is false only when it is proven that nothing of the
attempt's build ran. A nested start root was asked for that failed with
no process (kill proven or not) is marked startUncertain, a transparent
wrapper (same text, same errors.Is); startMayHaveRun reads it beside the
unproven-start sentinels. Swap's overlap and hold results and the worker
respawn's attemptExec use it. ReplaceWorker whose caller stops waiting
now joins the attempt's respawn (bounded by respawnJoinBound) before it
answers, and reports executed-or-unknown if the join gives up.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 4c71a7f0c67f3327c1d47a6dc019f0f9185f0ec0 parent 6d0a154
8 files changed, +249 −22
  • internal/activator/activator.go +4 −3
  • internal/activator/backend_exec.go +8 −4
  • internal/activator/containment.go +32 −0
  • internal/activator/genbackend.go +4 −1
  • internal/activator/swap.go +16 −10
  • internal/activator/swap_executed_test.go +148 −0
  • internal/activator/swaphold.go +8 −3
  • internal/activator/workerswap.go +29 −1
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 9dc67c9d6..625bbba7f 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -2821,9 +2821,10 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
e.attemptVeto = err
}
// The execution evidence (final review F2): a process of the
- // attempt's build was started, installed below or shed. Without
- // it, nothing of the build ran.
- e.attemptExec = proc != nil
+ // attempt's build was started, installed below or shed, or the
+ // start's error says it may have been (startMayHaveRun; final
+ // review wave 2, N4). Without it, nothing of the build ran.
+ e.attemptExec = startMayHaveRun(proc, err)
}
switch {
case owned && !retired && live:
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index cedac13fa..96e66227d 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -1677,6 +1677,10 @@ func (b *ExecBackend) spawnUnitFrom(ctx context.Context, rt registry.Route, sock
return fail(err)
}
inv, err := b.startRole(ctx, rt, instancectl.RoleRun, gen, exe, tier)
+ // Past this call root may have started the unit: every failure below
+ // but ErrStagedExe (root's refusal before it laid out or started
+ // anything) is marked startUncertain, since the build may have run
+ // before it failed or exited (final review wave 2, N4).
if errors.Is(err, instancectl.ErrStagedExe) {
// Root refused the staged build before it laid out or started
// anything, so there is nothing to kill and nothing unproven. The
@@ -1695,19 +1699,19 @@ func (b *ExecBackend) spawnUnitFrom(ctx context.Context, rt registry.Route, sock
}
if kerr := b.killFailedGen(ctx, rt, gen); kerr != nil {
// Nothing is released: the generation may still be running.
- return nil, "", fmt.Errorf("%s: start unit: %w; %w: %v", rt.Host, err, errGenUnproven, kerr)
+ return nil, "", markStartUncertain(fmt.Errorf("%s: start unit: %w; %w: %v", rt.Host, err, errGenUnproven, kerr))
}
_ = os.Remove(filepath.Join(p.dataDir, instancectl.GenLaunchFileName(gen)))
- return fail(fmt.Errorf("%s: start unit: %w", rt.Host, err))
+ return fail(markStartUncertain(fmt.Errorf("%s: start unit: %w", rt.Host, err)))
}
if err != nil {
- return fail(fmt.Errorf("%s: start unit: %w", rt.Host, err))
+ return fail(markStartUncertain(fmt.Errorf("%s: start unit: %w", rt.Host, err)))
}
if inv == "" {
// The server read the invocation after the start and found none:
// the unit already exited (plan 2 final review). Nothing is
// running, so there is nothing to stop; the wake failed.
- return fail(fmt.Errorf("%s: %s exited before it was observed", rt.Host, unitName(instancectl.RoleRun, rt.Host, gen)))
+ return fail(markStartUncertain(fmt.Errorf("%s: %s exited before it was observed", rt.Host, unitName(instancectl.RoleRun, rt.Host, gen))))
}
started := tier
if started == (cgroup.Limit{}) {
diff --git a/internal/activator/containment.go b/internal/activator/containment.go
index 536ca8b7b..fdae43c47 100644
--- a/internal/activator/containment.go
+++ b/internal/activator/containment.go
@@ -21,6 +21,38 @@ var (
errGenUnproven = errors.New("activator: generation start failed and its termination is unproven")
)
+// startUncertain marks a start's error when the start was handed to
+// something that may have run it before failing (final review wave 2, N4):
+// root's instance-control start call answered an error other than
+// instancectl.ErrStagedExe, or no invocation. Root may have started the
+// unit, and the build may have run, whatever became of it after. The
+// wrapper changes neither the text nor errors.Is, so every existing
+// classification and member text is the same; only startMayHaveRun reads
+// it.
+type startUncertain struct{ err error }
+
+func (e startUncertain) Error() string { return e.err.Error() }
+func (e startUncertain) Unwrap() error { return e.err }
+
+// markStartUncertain wraps err as startUncertain.
+func markStartUncertain(err error) error { return startUncertain{err: err} }
+
+// startMayHaveRun is a spawn's execution evidence, tri-state collapsed to
+// what an attempt may rely on (final review wave 2, N4): false only when it
+// is PROVEN that nothing of the build ran (no process came back, and the
+// error neither says the start's kill is unproven nor that the start was
+// handed off with an uncertain outcome); true when it ran or may have. A
+// flat start that failed is proven: the exec never happened. Every refusal
+// before the start (a gate, a preparation step, root's ErrStagedExe) is
+// proven too.
+func startMayHaveRun(proc Proc, err error) bool {
+ if proc != nil || genStartUnproven(err) {
+ return true
+ }
+ var u startUncertain
+ return errors.As(err, &u)
+}
+
// spawnKind is which child shape a spawn is, for placeFor.
type spawnKind int
diff --git a/internal/activator/genbackend.go b/internal/activator/genbackend.go
index 8c8b3a7b1..02aadb39d 100644
--- a/internal/activator/genbackend.go
+++ b/internal/activator/genbackend.go
@@ -30,7 +30,10 @@ type SwapBackend interface {
// returns as soon as the process runs, with g's Socket and Control
// filled in, and never waits for health: the caller checks health
// through that handle under its own deadline, and stops (or kills) a
- // process that fails. A nil Proc means nothing ran.
+ // process that fails. A nil Proc means nothing ran, unless the error
+ // says otherwise (startMayHaveRun; final review wave 2, N4): a start
+ // whose kill is unproven (errGenUnproven), or one handed to root whose
+ // outcome is uncertain (startUncertain).
//
// ctx may carry the caller's exec gate (withExecGate; final review F1).
// An implementation asks it (runExecGate) at its final launch point,
diff --git a/internal/activator/swap.go b/internal/activator/swap.go
index 9d9046073..0142fc7e8 100644
--- a/internal/activator/swap.go
+++ b/internal/activator/swap.go
@@ -78,13 +78,19 @@ 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 is the attempt's execution evidence (final review F8), one
+ // side of a tri-state (final review wave 2, N4): false means it is
+ // PROVEN that nothing of the attempt's build ran; true means it ran OR
+ // it is not known whether it ran. True once a process of the build
+ // started, whatever happened to it after (an overlap successor, a
+ // hold's pinned wake, a worker's respawn child, installed or shed), and
+ // also when a start was handed off and failed with its outcome
+ // uncertain (startMayHaveRun) or when ReplaceWorker could not learn
+ // what its respawn did. Only false lets an abandoned attempt count as
+ // having 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, and each of those returns
+ // before anything of the build could start.
Executed bool
}
@@ -343,8 +349,8 @@ func (a *Service) spawnSuccessor(ctx context.Context, gb SwapBackend, e *actEntr
// puts in the same critical section as e.gen.
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.
+ // successor process, or its error says the build may have run
+ // (startMayHaveRun), 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
@@ -479,7 +485,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
+ executed = startMayHaveRun(proc, spawnErr)
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
index 7bc4f060a..d1fef93b7 100644
--- a/internal/activator/swap_executed_test.go
+++ b/internal/activator/swap_executed_test.go
@@ -3,6 +3,7 @@ package activator
import (
"context"
"errors"
+ "fmt"
"os"
"path/filepath"
"sync"
@@ -10,6 +11,7 @@ import (
"testing"
"time"
+ "github.com/carlosframework/platform/internal/instancectl"
"github.com/carlosframework/platform/internal/registry"
)
@@ -126,3 +128,149 @@ func TestReplaceWorkerReportsWhetherTheAttemptsBuildExecuted(t *testing.T) {
})
}
}
+
+// Final review wave 2 (N4): Executed is false only when it is PROVEN that
+// nothing of the attempt's build ran. A nested start that root was asked
+// for and that failed with no process (the unit exited before it was
+// observed, or root's start answered an error), or whose kill is
+// unproven, may have run the build: executed or unknown. Only root's
+// refusal of the staged build before it started anything (ErrStagedExe)
+// is proven unexecuted. Before the fix both modes read the nil handle of
+// a kill-proven failure as "never ran", and an overlap read an unproven
+// kill the same way, so an abandoned forced attempt regained an allowance
+// it may have spent.
+func TestASwapWhoseNestedStartIsUncertainReportsExecution(t *testing.T) {
+ unitFailed := errors.New("the unit failed to start")
+ for _, tc := range []struct {
+ name string
+ mode SwapMode
+ exeErr error
+ stopErr error
+ executed bool
+ }{
+ {"overlap, root's start failed, kill proven", ModeOverlap, unitFailed, nil, true},
+ {"overlap, root's start failed, kill unproven", ModeOverlap, unitFailed, errors.New("stop timed out"), true},
+ {"overlap, root refused the staged build", ModeOverlap, fmt.Errorf("%w: digest mismatch", instancectl.ErrStagedExe), nil, false},
+ {"hold, root's start failed, kill proven", ModeHold, unitFailed, nil, true},
+ {"hold, root refused the staged build", ModeHold, fmt.Errorf("%w: digest mismatch", instancectl.ErrStagedExe), nil, false},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ f := newSwapFixture(t)
+ b := newLaunchBackend(t, "nested", f.b)
+ b.ctl.exeErr, b.ctl.stopGenErr = tc.exeErr, tc.stopErr
+ f.a.backend = b
+ id := markerScript(t, t.TempDir(), "v2", filepath.Join(t.TempDir(), "v2.ran"))
+ at := f.attempt(tc.mode)
+ at.StagedPath, at.Digest = id.StagedPath, id.Digest
+ res := f.a.Swap(context.Background(), at, f.newest)
+ if res.Outcome == SwapDone {
+ t.Fatalf("Swap = %+v, want it not done", res)
+ }
+ b.ctl.mu.Lock()
+ asked := len(b.ctl.got)
+ b.ctl.mu.Unlock()
+ if asked != 1 {
+ t.Fatalf("root was asked to start %d units, want 1", asked)
+ }
+ if res.Executed != tc.executed {
+ t.Fatalf("Swap = %+v: Executed %v, want %v", res, res.Executed, tc.executed)
+ }
+ })
+ }
+}
+
+// Final review wave 2 (N4): ReplaceWorker learns what its attempt's
+// respawn did before it answers, even when its caller stops waiting. The
+// respawn runs on (it is the entry's, on the service lifetime), so it is
+// joined, bounded by respawnJoinBound, and one still running past that is
+// reported executed-or-unknown. Before the fix a cancelled ReplaceWorker
+// returned at once with Executed false, while the respawn went on to exec
+// the attempt's build.
+func TestReplaceWorkerLearnsItsRespawnsExecutionWhenItsCallerLeaves(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ // supersede: the attempt is superseded while the respawn is held, so
+ // its exec gate refuses: proven unexecuted.
+ supersede bool
+ // bound shrinks respawnJoinBound below the hold: the join gives up.
+ bound time.Duration
+ executed bool
+ }{
+ {"the respawn execs after the caller left", false, 0, true},
+ {"the respawn is refused at its exec after the caller left", true, 0, false},
+ {"the respawn outlasts the join bound", false, 150 * time.Millisecond, true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ if tc.bound > 0 {
+ defer func(d time.Duration) { respawnJoinBound = d }(respawnJoinBound)
+ respawnJoinBound = tc.bound
+ }
+ 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)
+ entered, release := make(chan struct{}), make(chan struct{})
+ var releaseOnce sync.Once
+ releaseAll := func() { releaseOnce.Do(func() { close(release) }) }
+ t.Cleanup(releaseAll)
+ var hooked atomic.Bool
+ b.mu.Lock()
+ b.admitHook = func() {
+ if hooked.CompareAndSwap(false, true) {
+ close(entered)
+ <-release
+ }
+ }
+ b.mu.Unlock()
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ out := make(chan SwapResult, 1)
+ go func() {
+ out <- a.ReplaceWorker(ctx, workerAttempt(CauseRelease, ModeHold, fakeIdentity("v2")), nil,
+ func(string) uint64 { return rev.Load() })
+ }()
+ select {
+ case <-entered:
+ case <-time.After(5 * time.Second):
+ t.Fatal("the attempt's respawn never reached its preparation")
+ }
+ cancel()
+ if tc.bound > 0 {
+ // The respawn stays held past the bound: the join gives up.
+ select {
+ case res := <-out:
+ if res.Outcome != SwapAbandoned || res.Executed != tc.executed {
+ t.Fatalf("ReplaceWorker = %+v: want abandoned, Executed %v", res, tc.executed)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("ReplaceWorker never answered: the join is not bounded")
+ }
+ releaseAll()
+ waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(host) })
+ return
+ }
+ select {
+ case res := <-out:
+ t.Fatalf("ReplaceWorker = %+v answered while its respawn was still preparing", res)
+ case <-time.After(200 * time.Millisecond):
+ }
+ if tc.supersede {
+ rev.Store(2)
+ }
+ releaseAll()
+ select {
+ case res := <-out:
+ if res.Outcome != SwapAbandoned || res.Executed != tc.executed {
+ t.Fatalf("ReplaceWorker = %+v: want abandoned, Executed %v", res, tc.executed)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("ReplaceWorker never answered after its respawn finished")
+ }
+ 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 90ad768ed..e569eeb16 100644
--- a/internal/activator/swaphold.go
+++ b/internal/activator/swaphold.go
@@ -182,7 +182,10 @@ func (a *Service) withdrawOrTake(e *actEntry, h *holdSpec, why error) (holdWoke,
// 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).
+// review F8). A pinned start handed to root whose outcome is uncertain
+// (startUncertain) is executed-or-unknown too (final review wave 2, N4),
+// though without a process or an unproven kill it is still abandoned
+// rather than read as the build's failure.
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)
@@ -191,13 +194,15 @@ 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 }()
+ defer func() { out.Executed = execed || werr == nil || startMayHaveRun(nil, werr) }()
switch {
case werr != nil && !execed:
// The pinned wake failed before its build's exec: the lease, the
// restore, the row, the staged bytes, the claims admission. Nothing
// of the build ran, so this is no failure of it and nothing is held
- // down (spec §6.4, §8.3; index P5/P6). The attempt is abandoned.
+ // down (spec §6.4, §8.3; index P5/P6). The attempt is abandoned. (A
+ // nested start root answered with an uncertain outcome lands here
+ // too, reported Executed: see above.)
return abandon(werr)
case werr == nil:
if at.Canary {
diff --git a/internal/activator/workerswap.go b/internal/activator/workerswap.go
index 970b539bd..f48a59180 100644
--- a/internal/activator/workerswap.go
+++ b/internal/activator/workerswap.go
@@ -5,6 +5,7 @@ import (
"errors"
"fmt"
"log"
+ "time"
"github.com/carlosframework/platform/internal/release"
)
@@ -39,7 +40,9 @@ import (
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.
+ // installed or shed. Read once the respawn is over; false before it
+ // starts. A caller that stops waiting still gets it (final review wave
+ // 2, N4): the respawn is joined, or reported executed-or-unknown.
var executed bool
defer func() { out.Executed = executed }()
if newest == nil {
@@ -99,6 +102,7 @@ func (a *Service) ReplaceWorker(ctx context.Context, at Attempt, fallback *relea
}()
if !a.awaitRespawn(ctx, done) {
+ executed = a.joinAttemptRespawn(e, done)
return abandon(errSwapStopped)
}
a.mu.Lock()
@@ -320,6 +324,30 @@ func sameArtifact(got, want release.StagedIdentity) bool {
return got.StagedPath == want.StagedPath && got.Digest == want.Digest
}
+// joinAttemptRespawn is ReplaceWorker's answer for SwapResult.Executed when
+// its caller stopped waiting before the attempt's respawn finished (final
+// review wave 2, N4). The respawn runs on regardless: it is the entry's, on
+// the service lifetime, and may still exec the attempt's build. So it is
+// joined, bounded by its own worst case end to end (respawnJoinBound, the
+// bound a terminal stop's join uses), and its evidence read as on the
+// normal path. One still running past the bound is reported executed or
+// unknown (true): only proof that nothing ran may count as unexecuted.
+func (a *Service) joinAttemptRespawn(e *actEntry, done chan struct{}) bool {
+ bound := time.NewTimer(respawnJoinBound)
+ defer bound.Stop()
+ select {
+ case <-done:
+ case <-bound.C:
+ log.Printf("activator: worker %s: the attempt's respawn did not finish within %v of its caller leaving; reporting its build as possibly executed", e.host, respawnJoinBound)
+ return true
+ }
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ executed := e.attemptExec
+ e.attemptVeto, e.attemptExec = nil, false
+ return executed
+}
+
// awaitRespawn waits for a respawn's goroutine to finish with every child it
// made (respawnDone closes last). False when ctx ended first.
func (a *Service) awaitRespawn(ctx context.Context, done chan struct{}) bool {