amadan

carlos / platform Public

zdd plan 3 task 6: fix round 1: a pinned wake refused before its exec abandons the hold and emits no member event

holdResult reads a pinned wake that failed before its build's exec
(lease, restore, row, staged bytes, claims admission) as abandonment,
never SwapDown: startPinnedAt marks the entry (pinExec) immediately
before StartStaged. wake() emits no member event for a pinned wake's
pre-exec, stale or cancelled refusal or a mismatched spawn (logged
instead); a pinned build that ran and failed keeps the existing
wording. Unpinned wakes are unchanged.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 5f752be24fac47373499567eba83265743632c08 parent 1e12711
3 files changed, +196 −4
  • internal/activator/activator.go +8 −1
  • internal/activator/swap_hold_test.go +130 −0
  • internal/activator/swaphold.go +58 −3
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 304b6c2b2..83db92298 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -442,6 +442,11 @@ 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 bool
// holdNext is a hold swap waiting for this entry's stop to finish.
// finishStop installs its pinned wake. Written under a.mu.
holdNext *holdSpec
@@ -1079,7 +1084,9 @@ func (a *Service) wake(e *actEntry) {
// the route and the whole error are in hand — the journal line below the
// backends already write stays exactly as it was.
if err != nil {
- emitEvent(a.Logs, e.route, "error", memberSafeWakeReason(err))
+ if msg, ok := a.wakeFailureEvent(e, err); ok {
+ emitEvent(a.Logs, e.route, "error", msg)
+ }
} else {
emitEvent(a.Logs, e.route, "info", wakeEventMessage(w, e.worker))
}
diff --git a/internal/activator/swap_hold_test.go b/internal/activator/swap_hold_test.go
index ab342ae2d..71cf9afbb 100644
--- a/internal/activator/swap_hold_test.go
+++ b/internal/activator/swap_hold_test.go
@@ -3,6 +3,7 @@ package activator
import (
"context"
"errors"
+ "os"
"os/exec"
"slices"
"strings"
@@ -157,6 +158,135 @@ func TestSwapHoldFailureIsDown(t *testing.T) {
}
}
+// A pinned wake that fails before it reaches the exec ran nothing of the
+// build: the hold is abandoned, never down (index P5/P6, spec §6.4: only a
+// hold build that executed is held down). Reported down, plan 5 records a
+// failure of a build that never ran and halts its cohort.
+func TestSwapHoldRefusedBeforeTheExecIsAbandoned(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ arrange func(f *swapFixture)
+ want string
+ }{
+ {"the lease is held elsewhere after the stop", func(f *swapFixture) {
+ f.b.mu.Lock()
+ f.b.acquireErr = &errLeaseHeld{holder: "box-b", retryAfter: time.Minute}
+ f.b.mu.Unlock()
+ }, "instance lease held"},
+ {"the restore fails", func(f *swapFixture) {
+ f.b.mu.Lock()
+ f.b.restoreErr = errors.New("replica unreadable")
+ f.b.mu.Unlock()
+ }, "replica unreadable"},
+ {"the staged build changed after the stop", func(f *swapFixture) {
+ f.b.mu.Lock()
+ f.b.restoreErr = nil
+ f.b.mu.Unlock()
+ f.a.backend = &acquireHookBackend{fakeSwapBackend: f.b, hook: func() {
+ if err := os.WriteFile(f.staged, []byte("not build v2"), 0o755); err != nil {
+ panic(err)
+ }
+ }}
+ }, "staging"},
+ } {
+ 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 != SwapAbandoned || !strings.Contains(res.Reason, tc.want) {
+ t.Fatalf("Swap = %+v, want abandoned: %s", res, tc.want)
+ }
+ if gens, _, _ := f.b.calls(); gens != 0 {
+ t.Fatalf("StartGen calls = %d, want 0", gens)
+ }
+ })
+ }
+}
+
+// lockedSink is an EventSink safe for the wake goroutine and the test.
+type lockedSink struct {
+ mu sync.Mutex
+ events []eventRec
+}
+
+func (s *lockedSink) App(string, string, string, string) {}
+func (s *lockedSink) Event(acct, app, host, level, msg string) {
+ s.mu.Lock()
+ s.events = append(s.events, eventRec{acct, app, host, level, msg})
+ s.mu.Unlock()
+}
+
+func (s *lockedSink) errors() []string {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ var out []string
+ for _, e := range s.events {
+ if e.level == "error" {
+ out = append(out, e.msg)
+ }
+ }
+ return out
+}
+
+// A pinned wake's refusal is the attempt's business, not a member's: an
+// abandoned hold emits no "wake failed" event, and no swap text reaches the
+// member's timeline (plan 5's EvSwapDown carries a failed hold's line). A
+// pinned build that ran and failed is reported in the wording any failed
+// wake already uses.
+func TestSwapHoldWakeEventsCarryNoSwapText(t *testing.T) {
+ t.Run("superseded before the exec emits nothing", func(t *testing.T) {
+ f := newSwapFixture(t)
+ sink := &lockedSink{}
+ f.a.Logs = sink
+ var once sync.Once
+ f.a.backend = &acquireHookBackend{fakeSwapBackend: f.b, hook: func() { once.Do(func() { f.rev.Add(1) }) }}
+ if res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest); res.Outcome != SwapAbandoned {
+ t.Fatalf("Swap = %+v, want abandoned", res)
+ }
+ if got := sink.errors(); len(got) != 0 {
+ t.Fatalf("member error events %q, want none for an abandoned hold", got)
+ }
+ })
+ t.Run("a mismatched spawn's text stays in the log", func(t *testing.T) {
+ f := newSwapFixture(t)
+ sink := &lockedSink{}
+ f.a.Logs = sink
+ f.a.backend = &mismatchSwapBackend{fakeSwapBackend: f.b}
+ if res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest); res.Outcome != SwapDown {
+ t.Fatalf("Swap = %+v, want down: the mismatched process ran", res)
+ }
+ for _, msg := range sink.errors() {
+ if strings.Contains(msg, "the hold chose") || strings.Contains(msg, "swap:") {
+ t.Fatalf("member event %q carries swap text", msg)
+ }
+ }
+ })
+ t.Run("a pinned build that failed its health uses the existing wording", func(t *testing.T) {
+ f := newSwapFixture(t)
+ sink := &lockedSink{}
+ f.a.Logs = sink
+ f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") }
+ if res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest); res.Outcome != SwapDown {
+ t.Fatalf("Swap = %+v, want down", res)
+ }
+ got := sink.errors()
+ if len(got) != 1 || !strings.HasPrefix(got[0], "wake failed: ") || strings.Contains(got[0], "swap:") {
+ t.Fatalf("member error events %q, want one existing \"wake failed: …\" line", got)
+ }
+ })
+}
+
+// mismatchSwapBackend runs a different build from the one it was asked for.
+type mismatchSwapBackend struct {
+ *fakeSwapBackend
+}
+
+func (b *mismatchSwapBackend) StartStaged(ctx context.Context, rt registry.Route, g Gen) (Gen, Proc, error) {
+ ran, p, err := b.fakeSwapBackend.StartStaged(ctx, rt, g)
+ ran.Build = "v9"
+ return ran, p, err
+}
+
// acquireHookBackend runs hook inside Acquire: the window between a hold's
// re-check at its stop and its pinned wake's exec.
type acquireHookBackend struct {
diff --git a/internal/activator/swaphold.go b/internal/activator/swaphold.go
index 298f0efa1..e20917caa 100644
--- a/internal/activator/swaphold.go
+++ b/internal/activator/swaphold.go
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
+ "log"
"time"
"github.com/carlosframework/platform/internal/cgroup"
@@ -17,6 +18,10 @@ var (
// bound a hold holds requests for (spec §7.8). The hold is withdrawn
// before its stop finishes, so nothing of the attempt's build runs.
errHoldStopSlow = errors.New("swap: the old generation's stop outlasted the hold's bound; the hold is withdrawn")
+ // errPinMismatch: the backend ran something other than the hold's
+ // choice. A process ran, so the hold is down, but the text is the
+ // attempt's diagnostic and never a member's event.
+ errPinMismatch = errors.New("swap: the backend ran a different generation from the hold's choice")
)
// holdSpec is a hold swap waiting for its entry's stop (spec §7.8).
@@ -175,9 +180,15 @@ func (a *Service) holdResult(sctx context.Context, at Attempt, newest func(strin
}
<-hw.next.ready
a.mu.Lock()
- g, werr, proc := hw.next.gen, hw.next.err, hw.next.proc
+ g, werr, proc, execed := hw.next.gen, hw.next.err, hw.next.proc, hw.next.pinExec
a.mu.Unlock()
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.
+ return abandon(werr)
case werr == nil:
if at.Canary {
// A canary must survive the settle before it counts (spec §6.2,
@@ -248,7 +259,42 @@ func (a *Service) startPinnedChecked(ctx context.Context, e *actEntry, rt regist
if err != nil {
return nil, "", pin, err
}
- return a.startPinned(ctx, rt, pin)
+ return a.startPinnedAt(ctx, rt, pin, func() { a.markPinExec(e) })
+}
+
+// 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.
+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.
+func (a *Service) pinExecReached(e *actEntry) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ return e.pinExec
+}
+
+// wakeFailureEvent is the member's event for a failed wake, and whether
+// there is one. An unpinned wake's is what it always was. A hold's pinned
+// wake is the attempt's business (plan 5's swap events carry its member
+// line): a refusal before the exec, a stale or cancelled one, or a spawn
+// that ran the wrong build is logged only, so no swap text reaches the
+// member's timeline. A pinned build that ran and failed gets the wording
+// every failed wake gets.
+func (a *Service) wakeFailureEvent(e *actEntry, err error) (string, bool) {
+ if e.pin == nil {
+ return memberSafeWakeReason(err), true
+ }
+ if !a.pinExecReached(e) || classifyStep(nil, err) != stepFailed || errors.Is(err, errPinMismatch) {
+ log.Printf("activator: %s: hold's pinned wake refused or abandoned: %v", e.host, err)
+ return "", false
+ }
+ return memberSafeWakeReason(err), true
}
// startPinned is a hold's spawn: the attempt's staged build, verified and
@@ -259,6 +305,12 @@ func (a *Service) startPinnedChecked(ctx context.Context, e *actEntry, rt regist
// records the outcome, and a caller that reads a nil proc as "nothing ran"
// is right to.
func (a *Service) startPinned(ctx context.Context, rt registry.Route, pin Gen) (Proc, string, Gen, error) {
+ 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) {
gb, ok := a.backend.(SwapBackend)
if !ok {
return nil, "", pin, errSwapNoGens
@@ -270,12 +322,15 @@ func (a *Service) startPinned(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()
+ }
g, proc, err := gb.StartStaged(ctx, rt, pin)
if err != nil {
return proc, "", pin, err
}
if g.N != pin.N || g.Build != pin.Build || g.StagedPath != pin.StagedPath {
- return proc, "", pin, fmt.Errorf("backend ran generation %d of %s from %q; the hold chose %s from %q", g.N, g.Build, g.StagedPath, pin.Build, pin.StagedPath)
+ return proc, "", pin, fmt.Errorf("%w: generation %d of %s from %q; the hold chose %s from %q", errPinMismatch, g.N, g.Build, g.StagedPath, pin.Build, pin.StagedPath)
}
if err := a.genHealthy(ctx, g); err != nil {
return proc, "", g, err