amadan

carlos / platform Public

zdd plan 3 final fix: a hold checks its own lease and attempt at the exec; exec evidence is the spawn result (F7, F10)

startPinnedChecked's gate is the hold's pinCheck (entry, fences,
shutdown, revision) plus the liveness of the lease the pinned wake
actually holds (leaseHealth, which execLease implements over
lease.Held.Err; refusal errSwapLeaseLost, NoOwnership). It runs before
the preparation and again immediately before StartStaged, so a renewal
that fails or a fence set during Restore, the stage check or the tier
read execs nothing and the hold is abandoned.

pinExec is now set from the spawn's result, a returned process or an
unproven-start sentinel, before doWake's cleanup, not on entering
StartStaged. A StartStaged that fails before its exec returns no
process and the hold is abandoned, not down.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 967812fc4297db466db29f2e3d5e7d1b197bcc6c parent b19a0f0
5 files changed, +212 −26
  • internal/activator/activator.go +27 −5
  • internal/activator/activator_test.go +10 −0
  • internal/activator/backend_exec.go +4 −0
  • internal/activator/swap_hold_exec_test.go +127 −0
  • internal/activator/swaphold.go +44 −21
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 8e2a7a49c..91d583394 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -252,6 +252,26 @@ func releaseLease(ctx context.Context, lease Lease) error {
// Lease is a held-but-not-yet-replicating lease (wake in progress; Eleven's
// actLease).
+// leaseHealth is the OPTIONAL half of Lease: a held lease that can say
+// whether it is still held (no failed renewal, not spent). execLease has it
+// (lease.Held.Err). A hold's pinned wake asks it at its exec, before the
+// lease has become a replica (final review F7).
+type leaseHealth interface {
+ Err() error
+}
+
+// leaseLost is l's liveness as an ownership refusal: errSwapLeaseLost
+// (gateNoOwnership) when l says it is no longer held, nil when it is held
+// or cannot say.
+func leaseLost(l Lease) error {
+ if h, ok := l.(leaseHealth); ok {
+ if err := h.Err(); err != nil {
+ return fmt.Errorf("%w: %v", errSwapLeaseLost, err)
+ }
+ }
+ return nil
+}
+
type Lease interface {
// Replicate trades the lease for running replication of the (now present)
// local DB; the lease handle is spent on success. syncInterval is the
@@ -446,10 +466,12 @@ type actEntry struct {
// pinCheck is a hold's attempt checks, re-run under a.mu at the pinned
// exec. Set at construction (holdWakeLocked), like pin.
pinCheck func() error
- // pinExec is set (under a.mu) the moment a pinned wake reaches its
- // build's exec (markPinExec). Unset, the wake failed before anything of
- // the build ran, so the hold is abandoned, never down (spec §6.4), and
- // its refusal is no member's event.
+ // pinExec is the pinned wake's execution evidence, set (under a.mu,
+ // markPinExec) from its spawn's RESULT: a returned process, or an
+ // unproven-start sentinel (final review F10). It is set before any
+ // cleanup of the failed wake. Unset, nothing of the build ran, so the
+ // hold is abandoned, never down (spec §6.4), and its refusal is no
+ // member's event.
pinExec bool
// holdNext is a hold swap waiting for this entry's stop to finish.
// finishStop installs its pinned wake. Written under a.mu.
@@ -1615,7 +1637,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
}
} else if e.pin != nil { // swapsim:M3
var g Gen
- proc, spawnVer, g, err = a.startPinnedChecked(ctx, e, rt, *e.pin)
+ proc, spawnVer, g, err = a.startPinnedChecked(ctx, e, rt, *e.pin, lease)
w.gen = g
} else {
var g Gen
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index e1ef5d11e..42100f027 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -224,6 +224,16 @@ func (l *fakeLease) Replicate(_ context.Context, _ string, syncInterval time.Dur
return l.rep, nil
}
+// Err is the held lease's liveness (leaseHealth): the renewal the wake's
+// replica will inherit. A test loses the lease before Replicate by setting
+// the error on the rep this lease hands out (b.lastRep().setErr).
+func (l *fakeLease) Err() error {
+ if l.rep == nil {
+ return nil
+ }
+ return l.rep.Err()
+}
+
// Release counts on the backend: every doWake bail-out must give the lease
// back exactly once, and the only way to assert that from outside is to count
// the calls the wake made. It also records its context's error: the real
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 3a479b949..601262f98 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -630,6 +630,10 @@ func (l *execLease) Replicate(ctx context.Context, dbPath string, syncInterval t
func (l *execLease) Release(ctx context.Context) error { return l.held.Release(ctx) }
+// Err is the held lease's liveness (leaseHealth): a failed renewal, a
+// spent handle or a poisoned one.
+func (l *execLease) Err() error { return l.held.Err() }
+
func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, error) {
host, dbPath := rt.Host, b.pathsFor(rt).dbPath
if _, err := os.Stat(dbPath); err == nil {
diff --git a/internal/activator/swap_hold_exec_test.go b/internal/activator/swap_hold_exec_test.go
new file mode 100644
index 000000000..3543c97dd
--- /dev/null
+++ b/internal/activator/swap_hold_exec_test.go
@@ -0,0 +1,127 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strings"
+ "sync"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// prepHookBackend is fakeSwapBackend with hooks in the pinned wake's
+// preparation: Restore (under the wake's lease) and GenTier (the last step
+// before StartStaged).
+type prepHookBackend struct {
+ *fakeSwapBackend
+ onRestore func()
+ onTier func()
+}
+
+func (b *prepHookBackend) Restore(ctx context.Context, rt registry.Route) (bool, error) {
+ if b.onRestore != nil {
+ b.onRestore()
+ }
+ return b.fakeSwapBackend.Restore(ctx, rt)
+}
+
+func (b *prepHookBackend) GenTier(rt registry.Route, build string) cgroup.Limit {
+ if b.onTier != nil {
+ b.onTier()
+ }
+ return b.fakeSwapBackend.GenTier(rt, build)
+}
+
+// Final review F7: a hold's pinned wake checks the lease it actually holds,
+// and the attempt's checks, at the exec itself (immediately before
+// StartStaged), not before its preparation. A renewal that fails during
+// Restore or the rest of the preparation, or a fence set there, execs
+// nothing, and the attempt is abandoned (NoOwnership).
+func TestSwapHoldChecksItsLeaseAndAttemptAtTheExec(t *testing.T) {
+ lose := func(f *swapFixture) func() {
+ var once sync.Once
+ return func() {
+ once.Do(func() { f.b.lastRep().setErr(errors.New("lease renewal failed")) })
+ }
+ }
+ fence := func(f *swapFixture) func() {
+ var once sync.Once
+ return func() {
+ once.Do(func() {
+ f.a.mu.Lock()
+ if f.a.fences == nil {
+ f.a.fences = map[string]uint64{}
+ }
+ f.a.fences[f.host] = 9
+ f.a.mu.Unlock()
+ })
+ }
+ }
+ for _, tc := range []struct {
+ name string
+ arrange func(f *swapFixture, b *prepHookBackend)
+ want error
+ }{
+ {"the lease is lost during Restore", func(f *swapFixture, b *prepHookBackend) { b.onRestore = lose(f) }, errSwapLeaseLost},
+ {"the lease is lost during the preparation", func(f *swapFixture, b *prepHookBackend) { b.onTier = lose(f) }, errSwapLeaseLost},
+ {"a fence lands during the preparation", func(f *swapFixture, b *prepHookBackend) { b.onTier = fence(f) }, errSwapFenced},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ f := newSwapFixture(t)
+ b := &prepHookBackend{fakeSwapBackend: f.b}
+ tc.arrange(f, b)
+ f.a.backend = b
+ res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, tc.want.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned: %v", res, tc.want)
+ }
+ if gens, _, _ := f.b.calls(); gens != 0 {
+ t.Fatalf("StartStaged calls = %d: the pinned build was exec'd without its ownership", gens)
+ }
+ })
+ }
+}
+
+// Final review F10: a pinned wake's execution evidence is the spawn's
+// result, a returned process or an unproven-start sentinel, never having
+// entered StartStaged. A StartStaged that fails before its exec (a socket
+// directory, a token, a placement, cmd.Start) returns no process: nothing
+// of the build ran, so the hold is abandoned, not down. One that returns an
+// unproven-start sentinel, or a process that then fails its health check,
+// executed: the hold is down.
+func TestSwapHoldExecEvidenceIsTheSpawnResult(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ arrange func(f *swapFixture)
+ want SwapOutcome
+ }{
+ {"StartStaged fails before its exec", func(f *swapFixture) {
+ f.b.gmu.Lock()
+ f.b.genErr = errors.New("socket dir: permission denied")
+ f.b.gmu.Unlock()
+ }, SwapAbandoned},
+ {"StartStaged answers an unproven start", func(f *swapFixture) {
+ f.b.gmu.Lock()
+ f.b.genErr = fmt.Errorf("start g: %w", errGenUnproven)
+ f.b.gmu.Unlock()
+ }, SwapDown},
+ {"the build ran and failed its health check", func(f *swapFixture) {
+ f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") }
+ }, SwapDown},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ f := newSwapFixture(t)
+ tc.arrange(f)
+ res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
+ if res.Outcome != tc.want {
+ t.Fatalf("Swap = %+v, want %s", res, tc.want)
+ }
+ if gens, _, _ := f.b.calls(); gens != 1 {
+ t.Fatalf("StartStaged calls = %d, want 1", gens)
+ }
+ })
+ }
+}
diff --git a/internal/activator/swaphold.go b/internal/activator/swaphold.go
index 5878a98b8..46b94c060 100644
--- a/internal/activator/swaphold.go
+++ b/internal/activator/swaphold.go
@@ -252,33 +252,53 @@ func (a *Service) holdWakeLocked(e *actEntry, h *holdSpec) (*actEntry, error) {
return next, nil
}
-// startPinnedChecked is a hold's pinned spawn behind its attempt's checks,
-// run under a.mu at the exec (pinCheck, set by holdWakeLocked). A check
-// that fails spawns nothing.
-func (a *Service) startPinnedChecked(ctx context.Context, e *actEntry, rt registry.Route, pin Gen) (Proc, string, Gen, error) {
- a.mu.Lock()
- var err error
- if e.pinCheck != nil {
- err = e.pinCheck()
+// startPinnedChecked is a hold's pinned spawn behind its ownership and its
+// attempt's checks. The gate is the attempt's checks (pinCheck, set by
+// holdWakeLocked: the entry, the fences, the shutdown, the revision) plus
+// the liveness of the lease this wake actually holds (leaseLost; the entry
+// has no replica yet, so ownedLocked's e.rep check cannot see it). It runs
+// under a.mu before the preparation, and again at the exec itself,
+// immediately before StartStaged (final review F7): a renewal that failed
+// or a fence set during Restore, the stage check or the tier read execs
+// nothing. Its refusals are NoOwnership.
+//
+// The execution evidence comes from the spawn's result (final review F10):
+// a returned process, or an unproven-start sentinel, marks the wake as one
+// whose build may have run, before doWake cleans the failed wake up.
+// Anything else ran nothing of the build.
+func (a *Service) startPinnedChecked(ctx context.Context, e *actEntry, rt registry.Route, pin Gen, lease Lease) (Proc, string, Gen, error) {
+ gate := func() error {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ if e.pinCheck != nil {
+ if err := e.pinCheck(); err != nil {
+ return err
+ }
+ }
+ return leaseLost(lease)
}
- a.mu.Unlock()
- if err != nil {
+ if err := gate(); err != nil {
return nil, "", pin, err
}
- return a.startPinnedAt(ctx, rt, pin, func() { a.markPinExec(e) })
+ proc, ver, g, err := a.startPinnedAt(ctx, rt, pin, gate)
+ if proc != nil || genStartUnproven(err) {
+ a.markPinExec(e)
+ }
+ return proc, ver, g, err
}
-// markPinExec records that e's pinned wake has reached its build's exec.
-// A claims-on pinned spawn (Task 10) must call it from its exec hook, as
-// startPinnedAt does here: unmarked, a failed pinned wake reads as one that
-// ran nothing.
+// markPinExec records e's pinned wake's execution evidence (pinExec).
+// startPinnedChecked sets it from the spawn's result. A claims-on pinned
+// spawn (Task 10) sets it from its exec hook; every claims-backed hold is
+// refused before its stop on this base (claimsSwapRefusal), so that path
+// is never a hold's here.
func (a *Service) markPinExec(e *actEntry) {
a.mu.Lock()
e.pinExec = true
a.mu.Unlock()
}
-// pinExecReached reports whether e's pinned wake reached its build's exec.
+// pinExecReached reports whether e's pinned wake has execution evidence.
func (a *Service) pinExecReached(e *actEntry) bool {
a.mu.Lock()
defer a.mu.Unlock()
@@ -314,9 +334,10 @@ func (a *Service) startPinned(ctx context.Context, rt registry.Route, pin Gen) (
return a.startPinnedAt(ctx, rt, pin, nil)
}
-// startPinnedAt is startPinned with reached, when set, called immediately
-// before the exec: everything earlier refuses with nothing of the build run.
-func (a *Service) startPinnedAt(ctx context.Context, rt registry.Route, pin Gen, reached func()) (Proc, string, Gen, error) {
+// startPinnedAt is startPinned with gate, when set, asked immediately
+// before StartStaged: a refusal there returns no process, with nothing of
+// the build run.
+func (a *Service) startPinnedAt(ctx context.Context, rt registry.Route, pin Gen, gate func() error) (Proc, string, Gen, error) {
gb, ok := a.backend.(SwapBackend)
if !ok {
return nil, "", pin, errSwapNoGens
@@ -328,8 +349,10 @@ func (a *Service) startPinnedAt(ctx context.Context, rt registry.Route, pin Gen,
if pin.Tier == (cgroup.Limit{}) {
pin.Tier = gb.GenTier(rt, pin.Build) // the pinned artifact's own tier (spec §7.3)
}
- if reached != nil {
- reached()
+ if gate != nil {
+ if err := gate(); err != nil {
+ return nil, "", pin, err
+ }
}
g, proc, err := gb.StartStaged(ctx, rt, pin)
if err != nil {