amadan

carlos / platform Public

zdd plan 3 task 2: fix round 1: wake releases under a live context, a placed failed spawn proves its child, the unproven-wake refusal reads as the fence

I1: every failed-wake lease release (doWake's bail-outs, cleanupFailedWake,
the unproven-wake reaper) goes through releaseLease, which gives the lease
back under context.WithoutCancel(ctx) with the five-second budget
lease.Replicator.Close uses. A fence or Shutdown cancels the wake's
context, and a real release under it fails at once and spends the handle,
leaving the lease held for its TTL. Not gated on generations.

I2: startChildAt's failed health wait under GenContainment kills g<n>
after the leader's stop and, when that kill fails, returns the process
with an error wrapping errGenUnproven, so the wake keeps its lease until
the sweep proves the child empty. Containment off is unchanged.

I3: memberSafeWakeReason renders errWakeUnproven as the maintenance
fence's existing text, so no new member-facing string.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude aeff21c2b84560f5338190549060bfa8440b556d parent b1c3519
8 files changed, +242 −12
  • internal/activator/activator.go +24 −8
  • internal/activator/activator_test.go +14 −2
  • internal/activator/backend_exec.go +28 −0
  • internal/activator/logsink.go +9 −0
  • internal/activator/logsink_test.go +11 −0
  • internal/activator/maintenance.go +1 −1
  • internal/activator/opset.go +1 −1
  • internal/activator/opset_test.go +154 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index d98cc7bcd..59c2775a7 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -233,6 +233,22 @@ type RouteLookup interface {
Get(host string) (*registry.Route, error)
}
+// leaseReleaseBudget bounds a wake bail-out's lease release: one small S3
+// DELETE, the same five seconds lease.Replicator.Close gives its own.
+const leaseReleaseBudget = 5 * time.Second
+
+// releaseLease gives a failed wake's lease back under a budget of its own,
+// never the wake's context. A fence or Shutdown cancels that context, and
+// the real release (an S3 DeleteObject) fails at once on a dead one while
+// spending the handle anyway, which leaves the lease held for its whole TTL
+// and the host refusing wakes until then. WithoutCancel keeps the
+// context's values and drops only its cancellation and deadline.
+func releaseLease(ctx context.Context, lease Lease) error {
+ rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), leaseReleaseBudget)
+ defer cancel()
+ return lease.Release(rctx)
+}
+
// Lease is a held-but-not-yet-replicating lease (wake in progress; Eleven's
// actLease).
type Lease interface {
@@ -1273,7 +1289,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
if err != nil {
if lease != nil {
w.stopOutcome.uncertain = true
- w.stopOutcome.leaseReleased = lease.Release(ctx) == nil
+ w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil
}
var held *errLeaseHeld
if errors.As(err, &held) {
@@ -1284,7 +1300,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
}
if ctx.Err() != nil || !a.wakeAllowed(host) {
- w.stopOutcome.leaseReleased = lease.Release(ctx) == nil
+ w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil
return w, ErrFenced
}
// 2. Restore the DB from S3 when the local copy is gone — the restore must
@@ -1298,7 +1314,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
// and no marker is ever read to skip that.
fresh, err := a.backend.Restore(ctx, rt)
if err != nil {
- w.stopOutcome.leaseReleased = lease.Release(ctx) == nil
+ w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil
return w, fmt.Errorf("restore %s: %w", host, err)
}
@@ -1311,7 +1327,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
// wake (the Start-failure precedent: release the lease, surface the
// error) — a half-seeded instance must not serve.
if ctx.Err() != nil || !a.wakeAllowed(host) {
- w.stopOutcome.leaseReleased = lease.Release(ctx) == nil
+ w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil
return w, ErrFenced
}
if fresh && rt.Seed != "" {
@@ -1325,7 +1341,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
// the same way stop() leaves ErrStillRunning.
log.Printf("activator: %s: %v; leaving the lease to its TTL", host, err)
} else {
- w.stopOutcome.leaseReleased = lease.Release(ctx) == nil
+ w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil
}
return w, fmt.Errorf("seed %s: %w", host, err)
}
@@ -1358,7 +1374,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
token string
)
if ctx.Err() != nil || !a.wakeAllowed(host) {
- w.stopOutcome.leaseReleased = lease.Release(ctx) == nil
+ w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil
return w, ErrFenced
}
w.stopOutcome.uncertain = true
@@ -1369,7 +1385,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
// is a wake failure like any other: release and report.
token, err = newWorkerToken()
if err != nil {
- _ = lease.Release(ctx)
+ _ = releaseLease(ctx, lease)
return nil, fmt.Errorf("start %s: %w", host, err)
}
// Published onto the entry NOW, not at the awake transition: the
@@ -1423,7 +1439,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
if proc != nil || w.genUnproven {
a.cleanupFailedWake(ctx, rt, w, lease)
} else {
- w.stopOutcome.leaseReleased = lease.Release(ctx) == nil
+ w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil
}
return w, fmt.Errorf("start %s: %w", host, err)
}
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index f47851376..201938592 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -202,11 +202,14 @@ func (l *fakeLease) Replicate(_ context.Context, _ string, syncInterval time.Dur
// 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.
-func (l *fakeLease) Release(context.Context) error {
+// the calls the wake made. It also records its context's error: the real
+// release is an S3 DeleteObject(ctx), which fails at once on a dead context
+// and spends the handle anyway.
+func (l *fakeLease) Release(ctx context.Context) error {
if l.b != nil {
l.b.mu.Lock()
l.b.releases++
+ l.b.releaseCtxErrs = append(l.b.releaseCtxErrs, ctx.Err())
l.b.mu.Unlock()
}
return nil
@@ -221,6 +224,8 @@ type fakeBackend struct {
seeds int
releases int // lease releases — the failure paths' assertion
acquireErr error
+ // releaseCtxErrs is ctx.Err() at each fakeLease.Release, in order.
+ releaseCtxErrs []error
// admitErr is what AdmitConfig answers: the config gate a retained-lease
// respawn must ask again (amadan #213).
admitErr error
@@ -578,6 +583,13 @@ func (b *fakeBackend) releaseCount() int {
return b.releases
}
+// releaseCtxErrors is ctx.Err() as each fakeLease.Release saw it.
+func (b *fakeBackend) releaseCtxErrors() []error {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return slices.Clone(b.releaseCtxErrs)
+}
+
func (b *fakeBackend) lastProc() *fakeProc {
b.mu.Lock()
defer b.mu.Unlock()
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 0a7c545be..86ce93b63 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -192,6 +192,10 @@ type ExecBackend struct {
// and a nil Cgroups under Strict is ErrStrictRequired. Tests replace
// this seam, never the Cgroups field.
prepareStrict func(host string, limit cgroup.Limit) (strictProc, error)
+ // killGen is the spawn path's own generation kill (startChildAt's
+ // failed health wait). Nil: KillGen. Tests replace this seam, never the
+ // Cgroups field.
+ killGen func(ctx context.Context, rt registry.Route, n uint64) error
// strictPreflight is NextSpawnStrict's seam for PrepareStrict's runtime
// preflight. Nil: Cgroups.StrictPreflight, the very function
// PrepareStrict runs, so S2 and the spawn cannot disagree.
@@ -1260,12 +1264,36 @@ func (b *ExecBackend) startChildAt(ctx context.Context, rt registry.Route, sock
stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
defer cancel()
_ = p.Stop(stopCtx)
+ if b.GenContainment && n != 0 {
+ // The spawn was placed in g<n>, and Stop signals only the
+ // leader: a process it forked can live on there (spec §7.3).
+ // The generation's own kill ends it and proves the child
+ // empty. Unproven, the process goes back with an error that
+ // says so, and the caller keeps the lease until the sweep
+ // proves it (P2-13, as startUnitAt does).
+ if kerr := b.killGenAfterFailedStart(ctx, rt, n); kerr != nil {
+ log.Printf("activator: %s: kill generation %d after a failed health wait: %v", rt.Host, n, kerr)
+ return p, "", fmt.Errorf("%w; %w: kill generation %d after a failed health wait: %v", err, errGenUnproven, n, kerr)
+ }
+ }
return nil, "", err
}
b.normalizeSocketMode(rt, sock, p)
return p, spawnVer, nil
}
+// killGenAfterFailedStart is KillGen for a spawn path's own cleanup, on a
+// budget the caller's context cannot cancel: a wake cancelled by a fence or
+// Shutdown still proves what it started. nil is that proof.
+func (b *ExecBackend) killGenAfterFailedStart(ctx context.Context, rt registry.Route, n uint64) error {
+ kctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl)
+ defer cancel()
+ if b.killGen != nil {
+ return b.killGen(kctx, rt, n)
+ }
+ return b.KillGen(kctx, rt, n)
+}
+
// spawnChildFrom is the flat spawn with the executable and the tier named,
// and nothing after the exec: it returns the moment the process runs. Empty
// exe is the live link, which is every wake (startChildAt adds today's
diff --git a/internal/activator/logsink.go b/internal/activator/logsink.go
index 6cfd20f78..08ebaed60 100644
--- a/internal/activator/logsink.go
+++ b/internal/activator/logsink.go
@@ -7,6 +7,7 @@
package activator
import (
+ "errors"
"regexp"
"strings"
@@ -47,6 +48,14 @@ var absPathToken = regexp.MustCompile(`(^|[\s"(=])(/[\w@.\-/]+)`)
// text itself originates in our own code (backends, materialize, lease) —
// this is a rendering pass, not a secrecy boundary.
func memberSafeWakeReason(err error) string {
+ if errors.Is(err, errWakeUnproven) {
+ // A wake refused while the host's last failed wake is not proven
+ // stopped (zero-downtime spec §7.4). Members get the maintenance
+ // fence's existing text, the state that is like it: the platform
+ // holds the host closed until its own work is done, and it reopens
+ // by itself. Internal detail about generations stays in the log.
+ err = ErrFenced
+ }
s := err.Error()
switch {
case strings.Contains(s, "address already in use"):
diff --git a/internal/activator/logsink_test.go b/internal/activator/logsink_test.go
index 85215a6c6..e892e6f68 100644
--- a/internal/activator/logsink_test.go
+++ b/internal/activator/logsink_test.go
@@ -3,6 +3,7 @@ package activator
import (
"context"
"errors"
+ "fmt"
"strings"
"testing"
"time"
@@ -52,6 +53,16 @@ func TestMemberSafeWakeReason(t *testing.T) {
[]string{"wake failed", "permission denied", "ledgerpay"},
[]string{"/opt/carlos"},
},
+ {
+ // A wake refused while the last failed wake's generation is
+ // unproven (zero-downtime spec §7.4) reads as the fence's
+ // existing text: the platform holds the host closed until its
+ // own work is done. No new member-facing text.
+ "a wake refused over an unproven generation",
+ fmt.Errorf("h.example.com: refusing to wake: %w", errWakeUnproven),
+ []string{memberSafeWakeReason(ErrFenced)},
+ []string{"generation", "proven"},
+ },
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
diff --git a/internal/activator/maintenance.go b/internal/activator/maintenance.go
index aecb05bdf..4d7801e79 100644
--- a/internal/activator/maintenance.go
+++ b/internal/activator/maintenance.go
@@ -292,5 +292,5 @@ func (a *Service) cleanupFailedWake(ctx context.Context, rt registry.Route, w *w
w.stopOutcome.leaseReleased = w.stopOutcome.replicaClosed
return
}
- w.stopOutcome.leaseReleased = lease.Release(ctx) == nil
+ w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil
}
diff --git a/internal/activator/opset.go b/internal/activator/opset.go
index adad828ef..75519b7b1 100644
--- a/internal/activator/opset.go
+++ b/internal/activator/opset.go
@@ -476,7 +476,7 @@ func (a *Service) reapUnprovenWake(host string, u *unprovenWake) {
err = u.rep.Close(ctx)
outcome.replicaClosed = err == nil
} else if u.lease != nil {
- err = u.lease.Release(ctx)
+ err = releaseLease(ctx, u.lease)
}
cancel()
outcome.leaseReleased = err == nil
diff --git a/internal/activator/opset_test.go b/internal/activator/opset_test.go
index 9a13fa134..ef9ca4e7f 100644
--- a/internal/activator/opset_test.go
+++ b/internal/activator/opset_test.go
@@ -522,6 +522,55 @@ func TestSwapWitnessShutdownJoinsAWakingEntry(t *testing.T) {
if got := b.releaseCount(); got != 1 {
t.Fatalf("lease releases = %d, want 1", got)
}
+ // Shutdown cancelled the wake's context. The release must not run under
+ // it: a real release is an S3 DeleteObject that would fail at once and
+ // leave the lease held for its whole TTL.
+ for _, err := range b.releaseCtxErrors() {
+ if err != nil {
+ t.Fatalf("the lease was released under a dead context (%v)", err)
+ }
+ }
+}
+
+// Every doWake bail-out gives the lease back under a live context, even
+// after the wake's own context is gone (a fence, a Shutdown). Here a fence
+// lands while Start is held, so the wake's context is cancelled before it
+// releases. Flag off: the fix is not gated on generations.
+func TestAFencedWakeReleasesUnderALiveContext(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ gate := make(chan struct{})
+ b.startGate = gate
+ host := "fenced-wake.example.com"
+ failed := make(chan error, 1)
+ go func() {
+ release, err := a.beginRequest(context.Background(), testRoute(host))
+ if err == nil {
+ release()
+ }
+ failed <- err
+ }()
+ waitFor(t, "the wake to reach Start", func() bool { return b.startCount() == 1 })
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ fenceDone := make(chan struct{})
+ go func() {
+ a.FenceAndStop(ctx, host, 1) //nolint:errcheck
+ close(fenceDone)
+ }()
+ waitFor(t, "the fence", func() bool { return a.Fenced(host) })
+ close(gate)
+ if err := <-failed; err == nil {
+ t.Fatal("a fenced wake succeeded")
+ }
+ <-fenceDone
+ if got := b.releaseCount(); got != 1 {
+ t.Fatalf("lease releases = %d, want 1", got)
+ }
+ for _, err := range b.releaseCtxErrors() {
+ if err != nil {
+ t.Fatalf("the lease was released under a dead context (%v)", err)
+ }
+ }
}
// Flag off: Shutdown keeps today's rule and does not wait for a waking entry.
@@ -554,6 +603,111 @@ func TestShutdownLeavesAWakingEntryWithSwapOff(t *testing.T) {
waitFor(t, "cleanup", func() bool { return a.entryCount() == 0 })
}
+// execGenBackend wakes through a real ExecBackend's StartGen (a placed flat
+// spawn under GenContainment, in a delegated cgroup tree of plain files)
+// while the fake keeps the lease, restore and replica. Both kills, the
+// spawn path's own (ExecBackend.killGen) and the reaper's (KillGen), answer
+// the embedded killGenBackend's err.
+type execGenBackend struct {
+ *killGenBackend
+ x *ExecBackend
+ gmu sync.Mutex
+ gens []uint64
+}
+
+func (g *execGenBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) {
+ g.gmu.Lock()
+ g.gens = append(g.gens, n)
+ g.gmu.Unlock()
+ return g.x.StartGen(ctx, rt, n, perGen)
+}
+
+func (g *execGenBackend) NextDueVia(ctx context.Context, rt registry.Route, _ string) (int64, error) {
+ return g.fakeBackend.NextDue(ctx, rt)
+}
+
+func (g *execGenBackend) firstGen() uint64 {
+ g.gmu.Lock()
+ defer g.gmu.Unlock()
+ return g.gens[0]
+}
+
+// A placed flat generation whose health wait fails is stopped by its
+// leader only, and a process it forked can live on in g<n> (spec §7.3).
+// startChildAt kills g<n> after the leader's stop. When that kill fails,
+// the process goes back with an error wrapping errGenUnproven and the wake
+// keeps its lease until the sweep proves the child empty. When it
+// succeeds, nothing is left and the lease goes back at once. With
+// containment off nothing is placed, no kill is made, and the lease goes
+// back as before.
+func TestAPlacedSpawnWithAFailedHealthWaitProvesItsChild(t *testing.T) {
+ old := spawnHealthTimeout
+ spawnHealthTimeout = 300 * time.Millisecond
+ t.Cleanup(func() { spawnHealthTimeout = old })
+ rows := []struct {
+ name string
+ containment bool
+ killErr error
+ wantKept bool
+ }{
+ {"placed, kill fails", true, errors.New("g still holds a process"), true},
+ {"placed, kill proves the child empty", true, nil, false},
+ {"containment off", false, nil, false},
+ }
+ for _, row := range rows {
+ t.Run(row.name, func(t *testing.T) {
+ host := "placed.example.com"
+ x, _ := delegatedBackend(t, host)
+ x.GenContainment = row.containment
+ x.RunDirOverride = shortRunDir(t)
+ stageLive(t, x, host, "v1") // exits at once without binding: the health wait fails
+ a, b, _, _ := testActivator(t, time.Hour)
+ a.gens = true
+ kb := &killGenBackend{fakeBackend: b, err: row.killErr}
+ gb := &execGenBackend{killGenBackend: kb, x: x}
+ x.killGen = kb.KillGen
+ a.backend = gb
+ rt := flatGenRoute(x, host)
+
+ _, err := a.beginRequest(context.Background(), rt)
+ if err == nil {
+ t.Fatal("a wake whose health wait failed succeeded")
+ }
+ n := gb.firstGen()
+ a.mu.Lock()
+ _, kept := a.unprovenWakes[host]
+ a.mu.Unlock()
+ if kept != row.wantKept {
+ t.Fatalf("retained = %v, want %v (err %v)", kept, row.wantKept, err)
+ }
+ if !row.containment {
+ if k := kb.killed(); len(k) != 0 {
+ t.Fatalf("KillGen calls = %v with containment off, want none", k)
+ }
+ if got := b.releaseCount(); got != 1 {
+ t.Fatalf("lease releases = %d, want 1", got)
+ }
+ return
+ }
+ if k := kb.killed(); len(k) == 0 || k[0] != n {
+ t.Fatalf("KillGen calls = %v, want generation %d's first", k, n)
+ }
+ if !row.wantKept {
+ if got := b.releaseCount(); got != 1 {
+ t.Fatalf("lease releases = %d after a proven kill, want 1", got)
+ }
+ return
+ }
+ if got := b.releaseCount(); got != 0 {
+ t.Fatalf("lease releases = %d while g%d is unproven, want 0", got, n)
+ }
+ kb.setErr(nil)
+ a.sweepOnce()
+ waitFor(t, "the lease release", func() bool { return b.releaseCount() == 1 })
+ })
+ }
+}
+
// unprovenStartBackend is a GenBackend whose next StartGen fails the way a
// generation start fails when its kill cannot be proven (plan 2,
// startUnitAt): fail, while set, answers that one call. Every later call