amadan

carlos / platform Public

activator: unproven stops keep the lease, sidecar restarts leave the lock, wakes run on the committed row (plan 3, task 5)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev ecb6ac8be7ed00ce18cd38ae6255c59fbe93ccdd parent 48d084c
3 files changed, +400 −34
  • internal/activator/activator.go +198 −21
  • internal/activator/activator_test.go +179 −3
  • internal/activator/maintenance_test.go +23 −10

CI passed — run details

diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index fac275fa8..e76214b69 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -55,6 +55,7 @@ import (
"time"
"github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/instancectl"
"github.com/carlosframework/platform/internal/registry"
)
@@ -179,6 +180,16 @@ type DueStore interface {
EarliestTimers() (map[string]int64, error)
}
+// RouteLookup is the registry read doWake performs before Acquire (design
+// §6.8): the edge hands a wake the route it had cached at request time, which
+// can be stale by the time the wake actually runs (a config edit landed in
+// between); doWake re-reads the committed row so it spawns against
+// whatever's really in the registry, not a snapshot. *registry.Registry
+// satisfies this directly with its own Get.
+type RouteLookup interface {
+ Get(host string) (*registry.Route, error)
+}
+
// Lease is a held-but-not-yet-replicating lease (wake in progress; Eleven's
// actLease).
type Lease interface {
@@ -292,11 +303,29 @@ type actEntry struct {
// awake. All sidecar fields are written under a.mu in the awake state only
// (wake's attach, the sweeper's restart); a stop flips the state under the
// same lock first, so stop() owns them uncontended from then on.
- sidecar Proc // the co-spawned `sidecar run` child; nil = none running
- sidecarWanted bool // the host had a sidecar staged at wake time
- sidecarStarted time.Time // when the current sidecar spawned (backoff reset)
- sidecarBackoff time.Duration // last restart delay (doubles, capped)
- sidecarRetryAt time.Time // earliest next restart attempt
+ sidecar Proc // the co-spawned `sidecar run` child; nil = none running
+ sidecarWanted bool // the host had a sidecar staged at wake time
+ sidecarStarted time.Time // when the current sidecar spawned (backoff reset)
+ sidecarBackoff time.Duration // last restart delay (doubles, capped)
+ sidecarRetryAt time.Time // earliest next restart attempt
+ sidecarStarting bool // a restart's backend.StartSidecar call is in flight, off a.mu (design §6.7)
+
+ // pendingStop marks a stop whose proc.Stop returned ErrStillRunning
+ // (design §6.3): termination is unproven, so stop() returned without
+ // releasing the lease or forgetting the entry, and the entry stays in
+ // stopping. sweepOnce polls proc.Exited() for every pendingStop entry
+ // and finishes the stop (finishStop) the moment it's proven.
+ pendingStop bool
+
+ // wakeGen counts this host's wakes, bumped each time an entry becomes
+ // awake (in wake()). A sidecar restart goroutine captures it before
+ // dropping a.mu (design §6.7) and re-checks it on the far side of the
+ // async StartSidecar call: a generation mismatch means the entry moved
+ // on (stopped and possibly rewoken) while the call was in flight, so
+ // the goroutine's result belongs to a wake that's no longer current and
+ // must be discarded (stopping whatever it started) rather than
+ // installed.
+ wakeGen uint64
}
// Sidecar restart pacing — Restart=always semantics without a supervisor: the
@@ -386,6 +415,13 @@ type Service struct {
now func() time.Time // a seam for the unit tests' clock
liveVer func(string) (string, error) // live/<host> symlink target; a seam for tests
+ // Routes, when non-nil, is read by doWake before Acquire (design §6.8)
+ // to refresh the route against the registry's committed row rather than
+ // spawning against the copy the edge cached at request time. Nil (every
+ // test that doesn't wire it, and any caller that doesn't set it) keeps
+ // today's behaviour: the wake runs on the route it was handed.
+ Routes RouteLookup
+
// Logs, when non-nil, receives member-visible platform events (wake
// outcomes, restart bounces) for `carlos logs` — set before Start, like
// the backend's own Logs field. Nil (every test, a box without the
@@ -674,6 +710,7 @@ func (a *Service) wake(e *actEntry) {
delete(a.entries, e.host) // next request retries from scratch
} else {
e.state = actAwake
+ e.wakeGen++ // this entry's sidecar-restart goroutines (design §6.7) key on this generation
e.rep, e.proc = w.rep, w.proc
e.sidecar, e.sidecarWanted = w.sidecar, w.sidecarWanted
e.sidecarUncertain = w.sidecarUncertain
@@ -776,6 +813,32 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance
return w, ErrFenced
}
+ // 0. Refresh against the registry's committed row (design §6.8): the
+ // edge cached rt at request time, and a config edit (provision, a path
+ // change, a residence move) can land between that cache and this wake
+ // actually running. Spawning against a stale DBPath/Socket would boot
+ // the instance onto the wrong place on disk or bind the wrong socket, so
+ // the wake re-reads the row it's actually about to spawn and carries
+ // that forward — every step below reads rt, not the edge's copy.
+ if a.Routes != nil {
+ fresh, err := a.Routes.Get(rt.Host)
+ if err != nil {
+ return w, fmt.Errorf("%s: route lookup: %w", rt.Host, err)
+ }
+ if fresh == nil {
+ return w, fmt.Errorf("%s: no route", rt.Host)
+ }
+ if fresh.DBPath != rt.DBPath || fresh.Socket != rt.Socket {
+ log.Printf("activator: %s: route refreshed: db %q -> %q socket %q -> %q", rt.Host, rt.DBPath, fresh.DBPath, rt.Socket, fresh.Socket)
+ }
+ rt = *fresh
+ a.mu.Lock()
+ if e := a.entries[rt.Host]; e != nil {
+ e.route = rt
+ }
+ a.mu.Unlock()
+ }
+
// 1. The lease first: everything after this line is done as the prefix's
// single writer. Contention is not retried here — on one box it means a
// previous incarnation hasn't released; tell the client to come back after
@@ -967,6 +1030,26 @@ func (a *Service) stop(e *actEntry) {
// crash.
emitEvent(a.Logs, e.route, "info", "instance stopping")
+ // A sidecar restart (sweepOnce) runs its backend.StartSidecar call off
+ // a.mu (design §6.7) and may still be in flight when a stop begins. Wait,
+ // bounded by sidecarStopTimeout, for it to land — either it installs a
+ // fresh e.sidecar in time and the co-stop below stops that one directly,
+ // or it doesn't and the restart goroutine's own generation check (it
+ // re-reads a.entries[host] on the far side of the call) finds this entry
+ // gone or stopping and stops whatever it started instead. Either way
+ // nothing is left orphaned; this wait only decides which of the two
+ // stops it.
+ deadline := time.Now().Add(sidecarStopTimeout)
+ for {
+ a.mu.Lock()
+ starting := e.sidecarStarting
+ a.mu.Unlock()
+ if !starting || time.Now().After(deadline) {
+ break
+ }
+ time.Sleep(20 * time.Millisecond)
+ }
+
// Co-stop, sidecar FIRST — before the next-due probe and the instance's
// own SIGTERM. Two reasons: the sidecar must never outlive its instance
// (an orphan's reconnect loop would be a standing wake source — the
@@ -1013,16 +1096,41 @@ func (a *Service) stop(e *actEntry) {
// Separate budgets: a SIGTERM drain that runs long must not eat the final
// checkpoint's time — the S3 sync is the part that makes sleep durable.
procCtx, cancelProc := context.WithTimeout(context.Background(), actStopTimeout)
+ defer cancelProc()
if e.proc != nil {
err = e.proc.Stop(procCtx)
+ if errors.Is(err, ErrStillRunning) {
+ // The lease is never released over an unproven termination
+ // (design §6.3, codex R2-5): stop() bails out here rather than
+ // falling through to the checkpoint and lease release below,
+ // leaving the entry in stopping with pendingStop set.
+ // sweepOnce polls proc.Exited() for it and calls finishStop —
+ // the rest of what this function would have done — the moment
+ // termination is actually proven.
+ log.Printf("activator: stop %s: %v; keeping the lease until termination is proven", e.host, err)
+ a.mu.Lock()
+ e.pendingStop = true
+ e.stopOutcome = outcome
+ a.mu.Unlock()
+ return
+ }
outcome.mainStop, outcome.mainExited = err == nil, e.proc.Exited()
} else {
err = ErrStopUnproven
}
- cancelProc()
if err != nil {
log.Printf("activator: stop %s: %v", e.host, err)
}
+ a.finishStop(e, outcome)
+}
+
+// finishStop is stop()'s tail: the final checkpoint, lease release, and
+// forgetting the entry. Split out (design §6.3, codex R2-5) so sweepOnce can
+// run it once a pendingStop entry's proc.Exited() finally proves the
+// termination that stop() itself couldn't — outcome carries forward whatever
+// stop() had already observed (the sidecar co-stop, the due-time probe)
+// before it hit ErrStillRunning and returned.
+func (a *Service) finishStop(e *actEntry, outcome *stopObservation) {
// Close = final sync + lease release. After a lost lease the release part
// fails harmlessly (someone else owns the lock); the sync still preserves
// a local checkpoint.
@@ -1033,8 +1141,8 @@ func (a *Service) stop(e *actEntry) {
repErr = e.rep.Close(ctx)
}
outcome.replicaClosed = repErr == nil
- if err := repErr; err != nil {
- log.Printf("activator: final checkpoint %s: %v", e.host, err)
+ if repErr != nil {
+ log.Printf("activator: final checkpoint %s: %v", e.host, repErr)
}
a.mu.Lock()
outcome.uncertain = !outcome.complete()
@@ -1103,6 +1211,24 @@ func (a *Service) sweepOnce() {
earliest = nil
}
a.mu.Lock()
+ // Finish stops that were left pending because proc.Stop returned
+ // ErrStillRunning (design §6.3, codex R2-5): termination wasn't proven
+ // at the time, so stop() kept the lease and returned without forgetting
+ // the entry. Checked before the idle pass below for the same reason it's
+ // a separate loop at all — a pendingStop entry is in actStopping, not
+ // actAwake, so the idle pass would never reach it.
+ for _, e := range a.entries {
+ if e.pendingStop && e.proc != nil && e.proc.Exited() {
+ e.pendingStop = false
+ outcome := e.stopOutcome
+ if outcome == nil {
+ outcome = &stopObservation{}
+ }
+ // The exit that unblocked this loop IS the proof stop() lacked.
+ outcome.mainStop, outcome.mainExited = true, true
+ go a.finishStop(e, outcome)
+ }
+ }
cycled := 0 // build-convergence cycles started this tick (rate-limited)
for _, e := range a.entries {
if a.fences[e.host] != 0 {
@@ -1193,20 +1319,71 @@ func (a *Service) sweepOnce() {
e.sidecarBackoff = min(e.sidecarBackoff*2, sidecarRestartMax)
}
e.sidecarRetryAt = now.Add(e.sidecarBackoff)
- sc, err := a.backend.StartSidecar(context.Background(), e.route)
- if err != nil {
- e.sidecarUncertain = true
- if sc == nil {
- log.Printf("activator: SIDECAR restart for %s failed: %v (next attempt in %s)", e.host, err, e.sidecarBackoff)
- } else {
- log.Printf("activator: SIDECAR restart for %s failed: %v — retaining returned process", e.host, err)
- }
- e.sidecar = sc
- continue
+ // The restart call itself runs off a.mu (design §6.7, codex
+ // R2-6): held under the lock, a stalled StartSidecar (a hung
+ // exec, a wedged control socket) would block every other host's
+ // wake and sweep behind it. sidecarStarting keeps a second
+ // attempt from launching while one is already in flight;
+ // wakeGen, re-checked on the far side, catches an entry that
+ // stopped (and possibly rewoke under a new generation) while
+ // the call was running, so a stale attempt's result is stopped
+ // rather than installed onto a wake it doesn't belong to.
+ if !e.sidecarStarting {
+ e.sidecarStarting = true
+ gen := e.wakeGen
+ host := e.host
+ route := e.route
+ backoff := e.sidecarBackoff
+ go func() {
+ ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
+ sc, err := a.backend.StartSidecar(ctx, route)
+ cancel()
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ if e == nil || e.wakeGen != gen {
+ // The entry is gone (stopped and forgotten) or this
+ // host has moved on to a fresh wake under a new
+ // generation — that entry's own sidecarStarting, if
+ // any, belongs to a different attempt and is not
+ // this goroutine's to touch. Just stop whatever
+ // came back so it doesn't outlive the wake it
+ // doesn't belong to.
+ if sc != nil {
+ go sc.Stop(context.Background()) //nolint:errcheck
+ }
+ return
+ }
+ // Still the same entry and generation: this attempt owns
+ // its sidecarStarting flag regardless of what state the
+ // entry is in now — a stop may have begun while the call
+ // was in flight (stop() itself waits, bounded, for this
+ // to clear; leaving it set here would make that wait run
+ // the full sidecarStopTimeout for nothing).
+ e.sidecarStarting = false
+ if e.state != actAwake {
+ // Stopped mid-call: don't install onto a wake that's
+ // on its way out.
+ if sc != nil {
+ go sc.Stop(context.Background()) //nolint:errcheck
+ }
+ return
+ }
+ if err != nil {
+ e.sidecarUncertain = true
+ if sc == nil {
+ log.Printf("activator: SIDECAR restart for %s failed: %v (next attempt in %s)", host, err, backoff)
+ } else {
+ log.Printf("activator: SIDECAR restart for %s failed: %v — retaining returned process", host, err)
+ }
+ e.sidecar = sc
+ return
+ }
+ log.Printf("activator: sidecar for %s was down — restarted (backoff now %s)", host, backoff)
+ e.sidecar = sc
+ e.sidecarStarted = a.now()
+ }()
}
- log.Printf("activator: sidecar for %s was down — restarted (backoff now %s)", e.host, e.sidecarBackoff)
- e.sidecar = sc
- e.sidecarStarted = now
}
}
// Due-time wakes: a request-less wake through the same singleflight state
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 0a9075823..443056eb0 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -42,6 +42,13 @@ type fakeRep struct {
mu sync.Mutex
err error
closed bool
+ // b, when set, is bumped on Close — production's Rep.Close "does the
+ // final checkpoint/sync and releases the lease" (Rep's doc comment), so
+ // a successful Close is the fake's equivalent of the lease coming free
+ // once a wake has gotten past Replicate (before that point, the raw
+ // Lease.Release calls below count it instead — the two are mutually
+ // exclusive per wake, since Replicate spends the Lease).
+ b *fakeBackend
}
func (r *fakeRep) Err() error {
@@ -56,8 +63,14 @@ func (r *fakeRep) setErr(err error) {
}
func (r *fakeRep) Close(context.Context) error {
r.mu.Lock()
- defer r.mu.Unlock()
r.closed = true
+ b := r.b
+ r.mu.Unlock()
+ if b != nil {
+ b.mu.Lock()
+ b.releases++
+ b.mu.Unlock()
+ }
return nil
}
func (r *fakeRep) isClosed() bool {
@@ -73,6 +86,13 @@ type fakeProc struct {
// stopGate, when non-nil, blocks Stop() until closed — holds an entry in
// the stopping state so a test can poke at it.
stopGate chan struct{}
+ // stuck models a Proc whose termination cannot be proven within Stop's
+ // bounded ctx (design §6.3, unitProc's ErrStillRunning): while stuck and
+ // not yet dead, Stop reports ErrStillRunning instead of succeeding; exit()
+ // clears it. stopAttempts counts every Stop() call, proven or not, so a
+ // test can see the sweeper retrying a stop it hasn't been able to finish.
+ stuck bool
+ stopAttempts int
// label + rec feed the backend's order log ("instance" / "sidecar"), so
// the co-spawn/co-stop tests can assert interleavings.
label string
@@ -87,6 +107,11 @@ func (p *fakeProc) Stop(context.Context) error {
<-gate
}
p.mu.Lock()
+ p.stopAttempts++
+ if p.stuck && !p.dead {
+ p.mu.Unlock()
+ return ErrStillRunning
+ }
p.stopped = true
p.mu.Unlock()
if p.rec != nil {
@@ -104,6 +129,19 @@ func (p *fakeProc) isStopped() bool {
defer p.mu.Unlock()
return p.stopped
}
+func (p *fakeProc) stopCalls() int {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.stopAttempts
+}
+
+// exit marks the process as having died on its own (a crash, or — for a
+// stuck proc — the delayed termination Stop couldn't prove at the time).
+func (p *fakeProc) exit() {
+ p.mu.Lock()
+ p.dead = true
+ p.mu.Unlock()
+}
type fakeLease struct {
rep *fakeRep
@@ -171,10 +209,21 @@ type fakeBackend struct {
sidecarErr error
sidecarStarts int
sidecars []*fakeProc
+ // sidecarGate, when non-nil, blocks StartSidecar() until closed — models
+ // a stalled restart call so a test can assert it doesn't hold a.mu.
+ sidecarGate chan struct{}
// order interleaves backend calls and proc stops (start:instance,
// start:sidecar, nextDue, stop:sidecar, stop:instance) for the ordering
// assertions.
order []string
+
+ // procStuck makes every Start() return a fakeProc whose Stop reports
+ // ErrStillRunning until the test calls its exit() (design §6.3).
+ procStuck bool
+ // startedRoutes records the registry.Route Start() actually received,
+ // per call — doWake's route-refresh assertion (design §6.8) needs to see
+ // what the backend was actually handed, not what the caller passed in.
+ startedRoutes []registry.Route
}
func (b *fakeBackend) record(ev string) {
@@ -193,7 +242,7 @@ func (b *fakeBackend) Acquire(context.Context, registry.Route) (Lease, error) {
b.mu.Lock()
b.acquires++
err := b.acquireErr
- rep := &fakeRep{}
+ rep := &fakeRep{b: b}
b.reps = append(b.reps, rep)
b.mu.Unlock()
if err != nil {
@@ -226,8 +275,9 @@ func (b *fakeBackend) Start(_ context.Context, rt registry.Route) (Proc, string,
gate := b.startGate
fn := b.execVerFn
startErr := b.startErr
- proc := &fakeProc{label: "instance", rec: b}
+ proc := &fakeProc{label: "instance", rec: b, stuck: b.procStuck}
b.procs = append(b.procs, proc)
+ b.startedRoutes = append(b.startedRoutes, rt)
b.mu.Unlock()
if startErr != nil {
return nil, "", startErr
@@ -259,6 +309,12 @@ func (b *fakeBackend) HasSidecar(registry.Route) bool {
}
func (b *fakeBackend) StartSidecar(context.Context, registry.Route) (Proc, error) {
+ b.mu.Lock()
+ gate := b.sidecarGate
+ b.mu.Unlock()
+ if gate != nil {
+ <-gate
+ }
b.mu.Lock()
defer b.mu.Unlock()
b.sidecarStarts++
@@ -319,6 +375,12 @@ func (b *fakeBackend) lastProc() *fakeProc {
return b.procs[len(b.procs)-1]
}
+func (b *fakeBackend) lastStartedRoute() registry.Route {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.startedRoutes[len(b.startedRoutes)-1]
+}
+
func (b *fakeBackend) lastRep() *fakeRep {
b.mu.Lock()
defer b.mu.Unlock()
@@ -430,6 +492,18 @@ func waitFor(t *testing.T, what string, cond func() bool) {
t.Fatalf("timed out waiting for %s", what)
}
+// sidecarSettled reports whether host has no sidecar restart in flight — the
+// call runs off a.mu (design §6.7), so a test driving several restarts back
+// to back (each expecting the previous one's count and install to have
+// already landed) waits on this rather than racing sweepOnce's own, now much
+// shorter, return.
+func (a *Service) sidecarSettled(host string) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ return e == nil || !e.sidecarStarting
+}
+
func (a *Service) entryCount() int {
a.mu.Lock()
defer a.mu.Unlock()
@@ -2045,7 +2119,11 @@ func TestActivatorSidecarFailureDoesNotFailWake(t *testing.T) {
// The sweeper retries: attempt 2 immediately (first retry), then gated by
// the doubling backoff; once the fault clears, a retry lands the sidecar.
+ // The restart call itself now runs off a.mu (design §6.7), so a fresh
+ // attempt lands asynchronously — waitFor gives it a moment rather than
+ // asserting on sweepOnce's own (now much shorter) return.
a.sweepOnce()
+ waitFor(t, "sidecar restart attempt 2 to land", func() bool { return b.sidecarStartCount() == 2 && a.sidecarSettled(route.Host) })
if got := b.sidecarStartCount(); got != 2 {
t.Fatalf("sidecar attempts after first sweep = %d, want 2", got)
}
@@ -2056,6 +2134,7 @@ func TestActivatorSidecarFailureDoesNotFailWake(t *testing.T) {
b.setSidecarErr(nil)
clk.advance(3 * time.Second)
a.sweepOnce()
+ waitFor(t, "sidecar restart attempt 3 to land", func() bool { return b.sidecarStartCount() == 3 && a.sidecarSettled(route.Host) })
if got := b.sidecarStartCount(); got != 3 {
t.Fatalf("sidecar attempts after fault cleared = %d, want 3", got)
}
@@ -2135,9 +2214,17 @@ func TestActivatorSidecarRestartBackoff(t *testing.T) {
sc.mu.Unlock()
}
+ // The restart call itself runs off a.mu (design §6.7): each waitFor below
+ // gives the previous attempt's goroutine a moment to land — both the
+ // count and the entry's sidecarStarting flag — before the next kill/sweep,
+ // so a slow scheduler can't make this sequence's next sweepOnce see the
+ // prior attempt still "in flight" and skip its own.
+ host := "bots.example.com"
+
// T0: first crash — restarted on the next sweep (attempt 2), gate 2s.
kill()
a.sweepOnce()
+ waitFor(t, "restart 2 to land", func() bool { return b.sidecarStartCount() == 2 && a.sidecarSettled(host) })
if got := b.sidecarStartCount(); got != 2 {
t.Fatalf("first restart: attempts = %d, want 2", got)
}
@@ -2150,6 +2237,7 @@ func TestActivatorSidecarRestartBackoff(t *testing.T) {
// T3: past the 2s gate — attempt 3, gate doubles to 4s.
clk.advance(3 * time.Second)
a.sweepOnce()
+ waitFor(t, "restart 3 to land", func() bool { return b.sidecarStartCount() == 3 && a.sidecarSettled(host) })
if got := b.sidecarStartCount(); got != 3 {
t.Fatalf("second restart: attempts = %d, want 3", got)
}
@@ -2163,6 +2251,7 @@ func TestActivatorSidecarRestartBackoff(t *testing.T) {
// T8: past it — attempt 4 (gate now 8s).
clk.advance(2 * time.Second)
a.sweepOnce()
+ waitFor(t, "restart 4 to land", func() bool { return b.sidecarStartCount() == 4 && a.sidecarSettled(host) })
if got := b.sidecarStartCount(); got != 4 {
t.Fatalf("third restart: attempts = %d, want 4", got)
}
@@ -2171,6 +2260,7 @@ func TestActivatorSidecarRestartBackoff(t *testing.T) {
clk.advance(2 * time.Minute)
kill()
a.sweepOnce()
+ waitFor(t, "restart 5 to land", func() bool { return b.sidecarStartCount() == 5 && a.sidecarSettled(host) })
if got := b.sidecarStartCount(); got != 5 {
t.Fatalf("post-healthy restart: attempts = %d, want 5", got)
}
@@ -2178,6 +2268,7 @@ func TestActivatorSidecarRestartBackoff(t *testing.T) {
kill()
clk.advance(3 * time.Second)
a.sweepOnce()
+ waitFor(t, "restart 6 to land", func() bool { return b.sidecarStartCount() == 6 && a.sidecarSettled(host) })
if got := b.sidecarStartCount(); got != 6 {
t.Fatalf("backoff not reset after a healthy run: attempts = %d, want 6", got)
}
@@ -4014,3 +4105,88 @@ func TestChildEnvKeepsOneEntryPerKeyLastWins(t *testing.T) {
t.Errorf("A = %q", seen["A"])
}
}
+
+// A Proc whose Stop reports ErrStillRunning must not have its lease
+// released: stop() keeps the entry in stopping and the sweeper finishes it
+// once the proc reports exited (design §6.3, codex R2-5).
+func TestStopDoesNotReleaseTheLeaseOverAnUnprovenStop(t *testing.T) {
+ a, b, clk, _ := testActivator(t, time.Minute)
+ b.procStuck = true // Start returns a fakeProc whose Stop reports ErrStillRunning until exit()
+ rt := testRoute("h.example.com")
+ arrangeAwake(t, a, rt) // wake rt and release the request, as the idle-hibernate test does
+
+ clk.advance(2 * time.Minute)
+ a.sweepOnce() // idle: stop() runs, Stop reports ErrStillRunning
+ p := b.lastProc()
+ waitFor(t, "the stuck stop to be attempted", func() bool { return p.stopCalls() >= 1 })
+ if n := b.releaseCount(); n != 0 {
+ t.Fatalf("lease released %d times while the stop is unproven", n)
+ }
+ if a.entryCount() != 1 {
+ t.Fatal("entry forgotten before the stop was proven")
+ }
+ p.exit()
+ a.sweepOnce()
+ waitFor(t, "the lease to release once termination is proven", func() bool { return b.releaseCount() == 1 })
+}
+
+// The sidecar restart must not hold the activator mutex across the start
+// call (design §6.7, codex R2-6): a stalled sidecar start does not block
+// another host's wake.
+func TestSidecarRestartDoesNotHoldTheLockAcrossTheStart(t *testing.T) {
+ a, b, clk, _ := testActivator(t, time.Hour)
+ b.sidecarStaged = true
+
+ // wake a.example.com first, ungated — the gate below is for the
+ // sweeper's RESTART call, not this initial co-spawn.
+ arrangeAwake(t, a, testRoute("a.example.com"))
+ b.lastSidecar().exit() // the sweeper now wants a restart
+
+ gate := make(chan struct{})
+ b.sidecarGate = gate // StartSidecar blocks on this channel when non-nil
+ clk.advance(time.Minute)
+ go a.sweepOnce() // blocks inside StartSidecar until the gate closes
+ time.Sleep(50 * time.Millisecond)
+ // host b has no sidecar of its own — HasSidecar false keeps its wake's
+ // own co-spawn (unrelated to what's under test) off the same gate.
+ b.sidecarStaged = false
+ done := make(chan struct{})
+ go func() {
+ // wake b.example.com
+ release, err := a.beginRequest(context.Background(), testRoute("b.example.com"))
+ if err != nil {
+ t.Errorf("beginRequest b.example.com: %v", err)
+ return
+ }
+ release()
+ close(done)
+ }()
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("a second host's wake blocked behind a stalled sidecar start")
+ }
+ close(gate)
+}
+
+// doWake runs on the committed row, not the copy the edge supplied
+// (design §6.8, codex R8-1).
+func TestDoWakeRefreshesTheRouteFromTheRegistry(t *testing.T) {
+ a, b, _, reg := testActivator(t, time.Hour)
+ a.Routes = reg
+ old := testRoute("h.example.com")
+ fresh := old
+ fresh.DBPath = "/data/instances/h.example.com/h.example.com.db"
+ fresh.Socket = "/run/carlos/instances/h.example.com/app.sock"
+ // insert fresh into reg the way the file's registry-backed tests do
+ if err := reg.Upsert(fresh); err != nil {
+ t.Fatalf("Upsert: %v", err)
+ }
+ arrangeAwake(t, a, old) // wake with old (the edge's stale copy)
+ if got := b.lastStartedRoute().DBPath; got != fresh.DBPath {
+ t.Fatalf("Start saw DBPath %q, want the registry's %q", got, fresh.DBPath)
+ }
+ if got := b.lastStartedRoute().Socket; got != fresh.Socket {
+ t.Fatalf("Start saw Socket %q, want the registry's %q", got, fresh.Socket)
+ }
+}
diff --git a/internal/activator/maintenance_test.go b/internal/activator/maintenance_test.go
index e90d7628f..90f2196b6 100644
--- a/internal/activator/maintenance_test.go
+++ b/internal/activator/maintenance_test.go
@@ -379,6 +379,17 @@ func TestMaintenanceFailedBackendStartRemainsUnproven(t *testing.T) {
t.Fatal("hidden cleanup failure became proof", err)
}
}
+
+// Before design §6.7 (codex R2-6) this restart ran synchronously under a.mu,
+// so a fence attempt that raced a stalled StartSidecar found the lock held
+// and came back ErrFenceNotInstalled without ever touching the entry. Task 5
+// moved the call off a.mu specifically so a stalled restart can't block
+// anything else — which includes FenceAndStop: it now installs its fence
+// immediately. What still bounds a short-context attempt is stop()'s own
+// wait (sidecarStopTimeout) for the stalled restart to land before it can
+// act on e.sidecar — so the outcome changes (ErrStopUnproven, fence
+// installed) but a stalled sidecar still keeps a checked stop from
+// completing instantly, which is the property this test exists to cover.
func TestMaintenanceFenceSerializesSidecarSweepReplacement(t *testing.T) {
a, e, _, side, rep := maintenanceAwake(t)
gate, entered := make(chan struct{}), make(chan struct{})
@@ -388,23 +399,22 @@ func TestMaintenanceFenceSerializesSidecarSweepReplacement(t *testing.T) {
a.backend = &maintenanceBackend{fakeBackend: base, side: next, sideGate: gate, sideEntered: entered, rep: rep}
side.exited.Store(true)
e.sidecarWanted = true
- sweep := make(chan struct{})
- go func() { a.sweepOnce(); close(sweep) }()
+ // sweepOnce itself now returns immediately — the restart it launches
+ // runs in its own goroutine, which is what's actually gated below.
+ a.sweepOnce()
<-entered
- if proof, err := a.FenceAndStop(maintenanceContext(t, 15*time.Millisecond), e.host, 1); err != ErrFenceNotInstalled || proof != nil {
+
+ if proof, err := a.FenceAndStop(maintenanceContext(t, 15*time.Millisecond), e.host, 1); err != ErrStopUnproven || proof != nil {
t.Fatal("blocked replacement yielded proof", err)
}
- close(gate)
- <-sweep
- if !a.wakeAllowed(e.host) {
- t.Fatal("timed-out admission attempt installed a fence")
+ if a.wakeAllowed(e.host) {
+ t.Fatal("fenced admission attempt failed to install its fence")
}
+ close(gate)
if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), e.host, 1); err != nil || !proof.Matches(e.host, 1) {
t.Fatal("replacement escaped checked stop", err)
}
- if next.calls.Load() != 1 {
- t.Fatal("replacement sidecar was not stopped")
- }
+ waitFor(t, "the discarded replacement to be stopped", func() bool { return next.calls.Load() == 1 })
before := base.sidecarStartCount()
a.sweepOnce()
if base.sidecarStartCount() != before {
@@ -519,6 +529,9 @@ func TestMaintenanceFailedSidecarHandlesRemainOwned(t *testing.T) {
old.exited.Store(true)
a.entries[rt.Host] = &actEntry{host: rt.Host, route: rt, state: actAwake, proc: main, rep: rep, sidecar: old, sidecarWanted: true, lastActive: a.now()}
a.sweepOnce()
+ // The restart call runs off a.mu (design §6.7) — give it a
+ // moment to land before inspecting what it installed.
+ waitFor(t, "sidecar restart to land", func() bool { return base.sidecarStartCount() == 1 && a.sidecarSettled(rt.Host) })
}
a.mu.Lock()
entry := a.entries[rt.Host]