amadan

carlos / platform Public

isolated exec: plan 3 final-review fixes (regular-file opens, fence before Acquire, stop retries, listener close after done, copy)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev fcf1784940fedda107d164f614bf468e7c1051cc parent d9902bd
20 files changed, +699 −174
  • internal/activator/activator.go +102 −31
  • internal/activator/activator_test.go +155 −27
  • internal/activator/backend_exec.go +53 −43
  • internal/activator/backend_unit.go +14 −0
  • internal/activator/layout.go +18 −1
  • internal/activator/maintenance_test.go +12 −12
  • internal/activator/paths.go +8 −1
  • internal/activator/prepare.go +3 −3
  • internal/activator/unitproc.go +35 −16
  • internal/activator/unitproc_test.go +65 −19
  • internal/appconfig/syncexec.go +29 −6
  • internal/appconfig/syncexec_test.go +76 −0
  • internal/host/agent.go +11 −3
  • internal/instancectl/client.go +5 −0
  • internal/instancefs/instancefs.go +6 −0
  • internal/instancefs/instancefs_linux.go +27 −3
  • internal/instancefs/instancefs_linux_test.go +28 −0
  • internal/lease/endpoint_test.go +40 −0
  • internal/lease/lease.go +10 −0
  • internal/schedule/runner.go +2 −9

CI passed — run details

diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index b88df25d1..80f737ae2 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -322,6 +322,13 @@ type actEntry struct {
// proc.Exited() for every pendingStop entry and finishes the stop
// (finishPendingStop) the moment it's proven.
pendingStop bool
+ // stopRetrying is true while a retried proc.Stop call for a still-running
+ // pendingStop entry is in flight, so finishPendingStops does not launch a
+ // second one on top of it. stopRetriedAt is when the last retry was
+ // issued; finishPendingStops waits actStopTimeout between retries, the
+ // same cadence proc.Stop itself is bounded by.
+ stopRetrying bool
+ stopRetriedAt time.Time
}
// Sidecar restart pacing — Restart=always semantics without a supervisor: the
@@ -335,18 +342,18 @@ const (
)
// sidecarStopTimeout serves two related bounds. First, its name: the
-// sidecar's own stop at sleep time — `sidecar run` has no drain handler, so
+// sidecar's own stop at sleep time. `sidecar run` has no drain handler, so
// the SIGTERM is effectively immediate, and the SIGKILL fallback behind this
// timeout is insurance. Second (design §6.7): how long stop() waits on
// sidecarStartDone for an in-flight sidecar restart to land before giving up
// and proceeding on whatever e.sidecar currently is. The restart goroutine's
// own entry-identity check makes that second wait purely a promptness
-// trade-off, never a correctness one — nothing is orphaned or misplaced
+// trade-off, never a correctness one. Nothing is orphaned or misplaced
// whether it fires in time or times out.
const sidecarStopTimeout = 5 * time.Second
// stopDiscardedSidecar stops a sidecar process a restart attempt returned
-// but could not install — the wake it belonged to moved on (stopped, or
+// but could not install. The wake it belonged to moved on (stopped, or
// replaced by a fresh wake of the same host) while the call was in flight.
// Bounded by sidecarStopTimeout so a wedged Stop can't leak the goroutine
// forever; fire-and-forget, since nothing is waiting on this beyond the
@@ -837,7 +844,7 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance
// 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.
+ // 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 {
@@ -857,6 +864,21 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance
a.mu.Unlock()
}
+ // 0.5. Refuse a fenced or conflicting row before it ever reaches Acquire
+ // or Restore (design §6.8). The in-Start/in-Seed refusal below still
+ // exists as the second line of defense, but by the time either of those
+ // runs the lease is already held and Restore has already written to
+ // disk. For a fenced row that means restoring into a database a
+ // migration journal says is being vacated, and for a conflict row it
+ // means writing one host's database into another host's instance
+ // directory. Checking here, on the row this wake just refreshed against,
+ // closes both before either write happens.
+ if c, ok := a.backend.(Classifier); ok {
+ if layout, reason := c.Layout(rt); layout == LayoutFenced || layout == LayoutConflict {
+ return w, fmt.Errorf("%s: refusing to wake: %s", rt.Host, reason)
+ }
+ }
+
// 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
@@ -905,7 +927,16 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance
if fresh && rt.Seed != "" {
w.stopOutcome.uncertain = true
if err := a.backend.Seed(ctx, rt); err != nil {
- w.stopOutcome.leaseReleased = lease.Release(ctx) == nil
+ if errors.Is(err, ErrSeedUnproven) {
+ // A SIGKILL survivor of the seed may still be writing the
+ // database: releasing the lease here would be exactly the
+ // write-after-release hazard the lease exists to prevent
+ // (design §6.2), so it is left for its own TTL to expire,
+ // 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
+ }
return w, fmt.Errorf("seed %s: %w", host, err)
}
w.stopOutcome.uncertain = false
@@ -1050,14 +1081,14 @@ func (a *Service) stop(e *actEntry) {
// 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
+ // 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 entry-identity 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. sidecarStartDone is closed by the restart goroutine when it
- // finishes — however it finishes — so this waits on that signal directly
+ // finishes, however it finishes, so this waits on that signal directly
// rather than polling sidecarStarting on a timer.
a.mu.Lock()
startDone := e.sidecarStartDone
@@ -1123,13 +1154,19 @@ func (a *Service) stop(e *actEntry) {
// (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
+ // 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
+ // The call just above is this entry's first stop attempt, and it
+ // was already bounded by actStopTimeout. Recording it here is
+ // what makes finishPendingStops' own cadence a true actStopTimeout
+ // apart from THIS attempt, rather than firing a redundant retry
+ // the moment pendingStop becomes visible to it.
+ e.stopRetriedAt = a.now()
a.mu.Unlock()
return
}
@@ -1146,15 +1183,15 @@ func (a *Service) stop(e *actEntry) {
// finishPendingStop finishes a stop that was left pending because proc.Stop
// returned ErrStillRunning (design §6.3): now that proc.Exited() has proven
// termination, spec §6.3 calls for retrying proc.Stop once more before
-// declaring the main process closed — "sweepOnce retries proc.Stop on its
-// cadence until done closes" — so outcome.mainStop reflects an actual
-// observation rather than an assumption baked in at the ErrStillRunning
-// return. A unitProc that has already exited takes its done branch and
-// returns nil immediately; a childProc's Stop is a no-op on an
-// already-exited process either way, so retrying costs nothing once the
-// process is well and truly gone. mainExited is unconditionally true — the
-// exit that made this entry eligible for this pass IS that proof — whatever
-// this retried Stop call itself reports.
+// declaring the main process closed. Spec §6.3 says "sweepOnce retries
+// proc.Stop on its cadence until done closes", so outcome.mainStop reflects
+// an actual observation rather than an assumption baked in at the
+// ErrStillRunning return. A unitProc that has already exited takes its
+// done branch and returns nil immediately; a childProc's Stop is a no-op
+// on an already-exited process either way, so retrying costs nothing once
+// the process is well and truly gone. mainExited is unconditionally true:
+// the exit that made this entry eligible for this pass IS that proof,
+// whatever this retried Stop call itself reports.
func (a *Service) finishPendingStop(e *actEntry, outcome *stopObservation) {
ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
err := e.proc.Stop(ctx)
@@ -1169,27 +1206,61 @@ func (a *Service) finishPendingStop(e *actEntry, outcome *stopObservation) {
// finishPendingStops scans for entries left pending because proc.Stop
// returned ErrStillRunning (design §6.3, codex R2-5) and finishes every one
// whose proc has since exited, via finishPendingStop. Called from
-// sweepOnce's own tick and, once more, from Shutdown — the sweeper is
+// sweepOnce's own tick and, once more, from Shutdown. The sweeper is
// stopped by the time Shutdown runs, so without this second call a process
// that exits during shutdown would never get its final checkpoint. The
// caller holds a.mu.
func (a *Service) finishPendingStops() {
for _, e := range a.entries {
- if e.pendingStop && e.proc != nil && e.proc.Exited() {
+ if !e.pendingStop || e.proc == nil {
+ continue
+ }
+ if e.proc.Exited() {
e.pendingStop = false
outcome := e.stopOutcome
if outcome == nil {
outcome = &stopObservation{}
}
go a.finishPendingStop(e, outcome)
+ continue
}
+ // Still running: spec §6.3 says sweepOnce retries proc.Stop on its
+ // own cadence until done closes, and that applies here too, not only
+ // once termination has already been proven some other way. Without
+ // this, a stop whose first Control.Stop was itself lost (the socket
+ // refused, a 502 mid-roll) never asks again, and the entry sits in
+ // stopping holding the lease until the agent restarts.
+ if e.stopRetrying || a.now().Sub(e.stopRetriedAt) < actStopTimeout {
+ continue
+ }
+ e.stopRetrying = true
+ e.stopRetriedAt = a.now()
+ go a.retryPendingStop(e)
+ }
+}
+
+// retryPendingStop re-issues proc.Stop for a pendingStop entry that is
+// still running, off a.mu like finishPendingStop's own call. Unlike that
+// one, a return here proves nothing by itself. Done closing is still what
+// proves termination, and the next finishPendingStops pass is what notices
+// it and calls finishPendingStop. This only clears the in-flight guard so a
+// later pass can try again.
+func (a *Service) retryPendingStop(e *actEntry) {
+ ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
+ err := e.proc.Stop(ctx)
+ cancel()
+ if err != nil && !errors.Is(err, ErrStillRunning) {
+ log.Printf("activator: retry stop %s: %v", e.host, err)
}
+ a.mu.Lock()
+ e.stopRetrying = false
+ a.mu.Unlock()
}
// finishStop is stop()'s (and finishPendingStop's) tail: the final
// checkpoint, lease release, and forgetting the entry. Split out (design
// §6.3, codex R2-5) so a pendingStop entry's eventual finish shares it with
-// stop()'s own ordinary path — outcome carries forward whatever stop() had
+// stop()'s own ordinary path. outcome carries forward whatever stop() had
// already observed (the sidecar co-stop, the due-time probe) before it hit
// ErrStillRunning and returned, plus whatever finishPendingStop's own
// retried proc.Stop just added.
@@ -1278,7 +1349,7 @@ func (a *Service) sweepOnce() {
// 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 call at all — a pendingStop entry is in actStopping, not
+ // a separate call at all: a pendingStop entry is in actStopping, not
// actAwake, so the idle pass would never reach it.
a.finishPendingStops()
cycled := 0 // build-convergence cycles started this tick (rate-limited)
@@ -1378,12 +1449,12 @@ func (a *Service) sweepOnce() {
// attempt from launching while one is already in flight.
//
// The far side re-checks by ENTRY IDENTITY (e0, the *actEntry
- // itself), not a generation counter (codex review round 1,
- // Critical): a per-entry counter that starts at 0 and is bumped
- // once when an entry becomes awake reads 1 for every entry's
- // first wake, so it can never tell "this entry" apart from "a
- // completely different entry — for the same host, after a stop
- // and rewake — that also just had its first wake". A pointer
+ // itself), not a generation counter: a per-entry counter that
+ // starts at 0 and is bumped once when an entry becomes awake
+ // reads 1 for every entry's first wake, so it can never tell
+ // "this entry" apart from "a completely different entry, for
+ // the same host, after a stop and rewake, that also just had
+ // its first wake". A pointer
// comparison against a.entries[host] has no such collision: a
// fresh wake always allocates a brand new *actEntry (begin() and
// the due-wake path both do `&actEntry{...}`), so identity alone
@@ -1404,7 +1475,7 @@ func (a *Service) sweepOnce() {
a.mu.Lock()
defer a.mu.Unlock()
if a.entries[host] != e0 {
- // This host's map slot no longer holds e0 at all —
+ // This host's map slot no longer holds e0 at all:
// forgotten (stopped and deleted) or replaced by an
// entirely different entry from a fresh wake. e0 is
// unreachable from here on, so there's nothing left
@@ -1417,7 +1488,7 @@ func (a *Service) sweepOnce() {
// Still e0: this attempt owns its sidecarStarting flag
// and clears it (and the done channel already closing
// via the defer above) unconditionally, regardless of
- // what state e0 is in now — a stop may have begun
+ // what state e0 is in now: a stop may have begun
// (flipping e0 to actStopping) while the call was in
// flight, and stop()'s own wait on sidecarStartDone
// must not run the full sidecarStopTimeout for nothing.
@@ -1558,7 +1629,7 @@ func (a *Service) Shutdown(ctx context.Context) {
}
a.mu.Lock()
// The sweeper is stopped as of the line above, so nothing else will ever
- // poll a pendingStop entry (design §6.3) again on its own — run the same
+ // poll a pendingStop entry (design §6.3) again on its own. Run the same
// pass once here, and again on the ticker below while waiting, so a
// process that finally exits during shutdown still gets its final
// checkpoint instead of being abandoned mid-drain.
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index a7045124d..54217d827 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -42,11 +42,11 @@ type fakeRep struct {
mu sync.Mutex
err error
closed bool
- // b, when set, is bumped on Close — production's Rep.Close "does the
+ // 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
+ // Lease.Release calls below count it instead: the two are mutually
// exclusive per wake, since Replicate spends the Lease).
b *fakeBackend
}
@@ -135,8 +135,8 @@ func (p *fakeProc) stopCalls() int {
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).
+// 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
@@ -209,11 +209,11 @@ type fakeBackend struct {
sidecarErr error
sidecarStarts int
sidecars []*fakeProc
- // sidecarGate, when non-nil, blocks StartSidecar() until closed — models
+ // 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{}
// sidecarStartEntered, when non-nil, is closed the first time
- // StartSidecar is entered (before it waits on sidecarGate) — lets a test
+ // StartSidecar is entered (before it waits on sidecarGate). Lets a test
// wait for a restart's call to actually begin rather than sleeping a
// guessed duration.
sidecarStartEntered chan struct{}
@@ -227,9 +227,23 @@ type fakeBackend struct {
// 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
+ // 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
+ // layoutFn, when set, is what Layout reports: a test's way to exercise
+ // doWake's pre-Acquire fence check without a real ExecBackend. Nil means
+ // LayoutFlat, "": every test that never sets this field gets the
+ // behavior it always had, since a flat layout never fences a wake.
+ layoutFn func(rt registry.Route) (Layout, string)
+}
+
+// Layout is fakeBackend's Classifier implementation, always present so
+// doWake's type assertion succeeds; layoutFn (above) decides what it says.
+func (b *fakeBackend) Layout(rt registry.Route) (Layout, string) {
+ if b.layoutFn == nil {
+ return LayoutFlat, ""
+ }
+ return b.layoutFn(rt)
}
func (b *fakeBackend) record(ev string) {
@@ -502,7 +516,7 @@ 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
+// 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
@@ -893,6 +907,51 @@ func TestActivatorSeedFailureFailsWakeAndReleasesLease(t *testing.T) {
}
}
+// TestActivatorSeedUnprovenLeavesTheLeaseToItsTTL pins design §6.2: a seed
+// whose cgroup never emptied may still have a SIGKILL survivor writing the
+// database, so doWake must not release the lease over it the way it does
+// for every other seed failure (design §6.3's ErrStillRunning gets the same
+// treatment in stop()).
+func TestActivatorSeedUnprovenLeavesTheLeaseToItsTTL(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Minute)
+ b.restoreFresh = true
+ b.seedErr = fmt.Errorf("seed for badseed.example.com: %w: processes remain after a successful seed: still running", ErrSeedUnproven)
+ rt := testRoute("badseed.example.com")
+ rt.Seed = `["seed"]`
+
+ _, err := a.beginRequest(context.Background(), rt)
+ if err == nil {
+ t.Fatal("wake succeeded despite an unproven seed")
+ }
+ if !errors.Is(err, ErrSeedUnproven) {
+ t.Fatalf("wake error = %v, want it to wrap ErrSeedUnproven", err)
+ }
+ if got := b.releaseCount(); got != 0 {
+ t.Fatalf("lease releases = %d for an unproven seed, want exactly 0", got)
+ }
+}
+
+// TestActivatorWakeRefusesFencedRowBeforeAcquire pins design §6.8: a fenced
+// row is refused at the top of doWake, before the lease is ever acquired,
+// not only inside Start after the lease is already held.
+func TestActivatorWakeRefusesFencedRowBeforeAcquire(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Minute)
+ const reason = "layout migration in flight"
+ b.layoutFn = func(registry.Route) (Layout, string) { return LayoutFenced, reason }
+ rt := testRoute("fenced.example.com")
+
+ _, err := a.beginRequest(context.Background(), rt)
+ if err == nil {
+ t.Fatal("wake succeeded against a fenced row")
+ }
+ if !strings.Contains(err.Error(), reason) {
+ t.Fatalf("wake error = %q, want it to carry the fence reason %q", err, reason)
+ }
+ if acquires, _, starts := b.counts(); acquires != 0 || starts != 0 {
+ t.Fatalf("fenced row reached the backend (acquires=%d starts=%d), want neither", acquires, starts)
+ }
+}
+
// The other two doWake bail-outs release the lease exactly once too — the
// property the seed step slots into the middle of.
func TestActivatorWakeFailuresReleaseLeaseOnce(t *testing.T) {
@@ -2130,7 +2189,7 @@ 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
+ // 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) })
@@ -2225,8 +2284,8 @@ func TestActivatorSidecarRestartBackoff(t *testing.T) {
}
// 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,
+ // 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"
@@ -4139,7 +4198,7 @@ func TestStopDoesNotReleaseTheLeaseOverAnUnprovenStop(t *testing.T) {
// Drive the sweep from inside the wait itself: finishing a pendingStop
// entry now retries proc.Stop (design §6.3) on the sweeper's own
// cadence, so a single sweepOnce() right after exit() is a race against
- // exactly when the sweeper would next notice — polling sweepOnce()
+ // exactly when the sweeper would next notice. Polling sweepOnce()
// alongside the release count removes that race instead of hoping one
// call lands after exit() is visible.
waitFor(t, "the lease to release once termination is proven", func() bool {
@@ -4148,6 +4207,75 @@ func TestStopDoesNotReleaseTheLeaseOverAnUnprovenStop(t *testing.T) {
})
}
+// finishPendingStops must retry the stop itself while the process is still
+// running, not only once it has already exited (design §6.3: "sweepOnce
+// retries proc.Stop on its cadence until done closes"). Without the retry, a
+// stop whose first Control.Stop call itself failed would never ask again.
+func TestFinishPendingStopsRetriesTheStopWhileStillRunning(t *testing.T) {
+ a, b, clk, _ := testActivator(t, time.Minute)
+ b.procStuck = true // Stop reports ErrStillRunning until exit(), and this proc never exits here
+ rt := testRoute("stuck.example.com")
+ arrangeAwake(t, a, rt)
+
+ clk.advance(2 * time.Minute)
+ a.sweepOnce() // idle: stop() runs, Stop reports ErrStillRunning, entry goes pendingStop
+ p := b.lastProc()
+ waitFor(t, "the first stop attempt", func() bool { return p.stopCalls() >= 1 })
+
+ // Immediately after: no retry yet, the entry's own actStopTimeout budget
+ // has not passed since the first attempt.
+ a.sweepOnce()
+ if n := p.stopCalls(); n != 1 {
+ t.Fatalf("stop retried before actStopTimeout elapsed (stopCalls=%d)", n)
+ }
+
+ clk.advance(actStopTimeout + time.Second)
+ waitFor(t, "a second stop attempt once actStopTimeout has passed", func() bool {
+ a.sweepOnce()
+ return p.stopCalls() >= 2
+ })
+ if b.releaseCount() != 0 {
+ t.Fatal("lease released while the process still reports running")
+ }
+ if a.entryCount() != 1 {
+ t.Fatal("entry forgotten while the process still reports running")
+ }
+}
+
+// The deferred Shutdown drain (ledger T5): a pendingStop entry whose process
+// finally exits DURING Shutdown still gets its final checkpoint and lease
+// release before Shutdown returns, via the same finishPendingStops pass
+// Shutdown's own drain loop re-runs on its poll ticker.
+func TestShutdownDrainsAPendingStopThatExitsMidShutdown(t *testing.T) {
+ a, b, clk, _ := testActivator(t, time.Minute)
+ b.procStuck = true
+ rt := testRoute("shutdown-pending.example.com")
+ arrangeAwake(t, a, rt)
+
+ clk.advance(2 * time.Minute)
+ a.sweepOnce() // stop() runs, Stop reports ErrStillRunning, entry goes pendingStop
+ p := b.lastProc()
+ waitFor(t, "the stop to be attempted", func() bool { return p.stopCalls() >= 1 })
+
+ // The process exits on its own shortly after Shutdown begins draining.
+ // Shutdown's 100ms poll ticker is what has to notice it, not this test.
+ go func() {
+ time.Sleep(150 * time.Millisecond)
+ p.exit()
+ }()
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ a.Shutdown(ctx)
+
+ if got := b.releaseCount(); got != 1 {
+ t.Fatalf("lease releases after Shutdown = %d, want exactly 1", got)
+ }
+ if a.entryCount() != 0 {
+ t.Fatal("Shutdown returned with a pending-stop entry still on the books")
+ }
+}
+
// 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.
@@ -4155,7 +4283,7 @@ 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
+ // 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
@@ -4169,7 +4297,7 @@ func TestSidecarRestartDoesNotHoldTheLockAcrossTheStart(t *testing.T) {
clk.advance(time.Minute)
a.sweepOnce() // launches the restart; it blocks inside StartSidecar on gate
<-entered // wait for the restart's call to actually begin, not a guessed sleep
- // host b has no sidecar of its own — HasSidecar false keeps its wake's
+ // 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.mu.Lock()
b.sidecarStaged = false
@@ -4216,22 +4344,22 @@ func TestDoWakeRefreshesTheRouteFromTheRegistry(t *testing.T) {
}
// A restart whose StartSidecar call is still stalled must never install
-// onto a DIFFERENT wake of the same host once it finally returns (codex
-// review round 1, Critical). The bug it regression-tests: the discard check
-// used to compare a per-entry generation counter that is always 1 on an
-// entry's first wake (entries are never reused, so nothing ever bumps it a
-// second time), which meant it could never tell "this entry" apart from "a
-// completely different entry — for the same host, after a stop and rewake —
-// that also just had its own first wake". A stale restart could install its
-// result straight over a fresh wake's own sidecar. The fix (this test's
-// assertion) compares entry identity: the *actEntry pointer captured when
-// the restart launched, not a counter on it.
+// onto a DIFFERENT wake of the same host once it finally returns. The bug
+// it regression-tests: a discard check that compares a per-entry
+// generation counter is always 1 on an entry's first wake (entries are
+// never reused, so nothing ever bumps it a second time), which means it
+// can never tell "this entry" apart from "a completely different entry,
+// for the same host, after a stop and rewake, that also just had its own
+// first wake". A stale restart could install its result straight over a
+// fresh wake's own sidecar. The fix (this test's assertion) compares entry
+// identity: the *actEntry pointer captured when the restart launched, not
+// a counter on it.
//
// This test legitimately costs several seconds of wall time: stop() itself
// waits on the in-flight restart, bounded by sidecarStopTimeout, before it
// can finish taking the old entry down (that wait is a separate, already
-// covered property — see TestSidecarRestartDoesNotHoldTheLockAcrossTheStart
-// and stop()'s own doc comment) — the file has no seam to shrink
+// covered property, see TestSidecarRestartDoesNotHoldTheLockAcrossTheStart
+// and stop()'s own doc comment). The file has no seam to shrink
// sidecarStopTimeout for tests, so this accepts the wall time rather than
// adding one for a single regression test.
func TestStaleSidecarRestartNeverInstallsOntoAFreshWake(t *testing.T) {
@@ -4262,7 +4390,7 @@ func TestStaleSidecarRestartNeverInstallsOntoAFreshWake(t *testing.T) {
t.Fatal("Bounce'd entry never finished stopping (the stalled restart wedged the stop past its bound)")
}
- // A fresh wake of the same host gets its own sidecar — this is what the
+ // A fresh wake of the same host gets its own sidecar. This is what the
// stale restart, still stalled, must not be allowed to overwrite.
b.mu.Lock()
b.sidecarGate = nil // the fresh wake's own co-spawn must not block on it too
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 77b0eca5c..8e78e615e 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -35,6 +35,14 @@ import (
// socket and answer. A var so tests can shrink the failure path.
var spawnHealthTimeout = 30 * time.Second
+// ErrSeedUnproven is seedUnit's answer when a seed's cgroup never reports
+// empty (design §6.2): only a SIGKILL survivor reaches this, and while its
+// cgroup stays populated it may still be writing the database, so the
+// caller must treat this the way stop() treats ErrStillRunning and leave
+// the lease for its own TTL to expire rather than release it over a write
+// that might still be happening.
+var ErrSeedUnproven = errors.New("activator: seed unproven; processes remain")
+
// quarantineSuffix names where a restore that failed its integrity check is
// parked. It deliberately does NOT end in ".db": the box-wide litestream
// generator globs <data>/*.db, and a quarantined corpse must never be picked
@@ -108,12 +116,15 @@ type ExecBackend struct {
// feature — binds the control socket without those verbs, so an app that
// asks gets 404 rather than a socket that isn't there.
Timers schedule.TimerStore
- // Layout supplies the box facts the classification of §8.1 needs beyond
- // the row: journals and peers. Zero value: never fenced, never conflict.
- Layout LayoutInputs
+ // LayoutFacts supplies the box facts the classification of §8.1 needs
+ // beyond the row: journals and peers. Zero value: never fenced, never
+ // conflict. Named apart from the Layout method (the Classifier
+ // interface's Layout(rt) (Layout, string)) so a struct field and a
+ // method can coexist on the same type.
+ LayoutFacts LayoutInputs
// Control is the root instance-control channel a nested wake talks to
// (design §5.3). Nil refuses every nested wake with "instance control
- // channel not configured" rather than dialing nothing — a flat or
+ // channel not configured" rather than dialing nothing. A flat or
// foreign host never needs it, and a nil Control does not touch them.
Control Control
// UnitState answers `systemctl show` for the observer (design §6.5).
@@ -374,7 +385,7 @@ func (b *ExecBackend) Seed(ctx context.Context, rt registry.Route) error {
}
// seedArgv parses rt.Seed's JSON argv array and appends the --db flag, the
-// composition both spawn paths need identically (design §6.2) — an empty
+// composition both spawn paths need identically (design §6.2). An empty
// declared seed (an empty array, len 0) returns a nil argv, which both
// callers treat as "nothing to run" (doWake only calls Seed for a
// non-empty rt.Seed; this stays defensive rather than assuming that).
@@ -392,7 +403,7 @@ func seedArgv(rt registry.Route, dbPath string) ([]string, error) {
}
// removeSeedDB deletes a failed seed's own half-built database and its
-// WAL/SHM siblings — the cleanup both spawn paths run when, and only when,
+// WAL/SHM siblings: the cleanup both spawn paths run when, and only when,
// seedOwnsDB says the database is the seed's own work (seedCreatedTheDB).
// Shared so the two paths cannot drift on which files "the database" means.
func removeSeedDB(dbPath string) {
@@ -563,9 +574,9 @@ func (c *capBuffer) String() string {
return string(c.b)
}
-// seedUnit is Seed's nested body: the same contract as seedChild — the
-// merged env, the CARLOS_ADMIN_TOKEN mint, the cleanup on a failure this
-// call itself caused — run through the root instance-control channel
+// seedUnit is Seed's nested body: the same contract as seedChild, namely
+// the merged env, the CARLOS_ADMIN_TOKEN mint, and the cleanup on a failure
+// this call itself caused, run through the root instance-control channel
// instead of a direct exec (design §6.2). Argv and env come from the same
// seedArgv and withoutVars calls seedChild uses, so the two cannot drift.
func (b *ExecBackend) seedUnit(ctx context.Context, rt registry.Route) error {
@@ -595,14 +606,14 @@ func (b *ExecBackend) seedUnit(ctx context.Context, rt registry.Route) error {
return err
}
// Whether the database existed BEFORE the seed ran decides whether a
- // failed seed may clean up after itself below — seedChild's rule,
+ // failed seed may clean up after itself below, seedChild's rule,
// unchanged for the unit path.
_, statErr := os.Stat(dbPath)
seedOwnsDB := seedCreatedTheDB(statErr)
log.Printf("activator: seeding %s: %s", host, strings.Join(argv, " "))
- // The client already bounds a seed call at instancectl.DeadlineSeed —
- // wrapping ctx any tighter here (seedTimeout, at 2m, is shorter) would
+ // The client already bounds a seed call at instancectl.DeadlineSeed.
+ // Wrapping ctx any tighter here (seedTimeout, at 2m, is shorter) would
// guarantee the failure path on every seed that legitimately runs
// longer than that. Wrapping at the SAME deadline is redundant with the
// client's own timeout but harmless, and keeps this call visibly
@@ -624,19 +635,19 @@ func (b *ExecBackend) seedUnit(ctx context.Context, rt registry.Route) error {
emptyErr := b.waitSeedCgroupEmpty(wctx, rt)
wcancel()
if emptyErr != nil {
- return fmt.Errorf("seed for %s: processes remain after a successful seed: %w", host, emptyErr)
+ return fmt.Errorf("seed for %s: %w: processes remain after a successful seed: %v", host, ErrSeedUnproven, emptyErr)
}
log.Printf("activator: seeded %s", host)
return nil
}
-// seedUnitFailed is seedUnit's shared failure tail (controller ruling, fix
-// round 1): on EITHER failure shape — Control.Seed itself errored, or it
-// answered with a non-success result — the seed unit is stopped and its
-// cgroup is proven empty BEFORE any database cleanup runs, never after.
-// The earlier ordering ran the cleanup on the bare result, which could race
-// a seed process SIGKILL hadn't actually reaped yet — exactly the class of
-// bug Stop's own ErrStillRunning discipline exists to rule out elsewhere.
+// seedUnitFailed is seedUnit's shared failure tail: on EITHER failure
+// shape, whether Control.Seed itself errored or it answered with a
+// non-success result, the seed unit is stopped and its cgroup is proven
+// empty BEFORE any database cleanup runs, never after. Running the cleanup
+// on the bare result instead could race a seed process SIGKILL hadn't
+// actually reaped yet, exactly the class of bug Stop's own ErrStillRunning
+// discipline exists to rule out elsewhere.
// removeSeedDB runs only when seedOwnsDB AND the wait proved the cgroup
// empty; when it did not, nothing is removed and the returned error says
// so, because a database is not safe to delete out from under a process
@@ -654,7 +665,7 @@ func (b *ExecBackend) seedUnitFailed(ctx context.Context, rt registry.Route, dbP
emptyErr := b.waitSeedCgroupEmpty(wctx, rt)
wcancel()
if emptyErr != nil {
- return fmt.Errorf("%w (the seed's processes remain after stopping it; its database was left in place: %v)", cause, emptyErr)
+ return fmt.Errorf("%w: %w (the seed's processes remain after stopping it; its database was left in place: %v)", cause, ErrSeedUnproven, emptyErr)
}
if seedOwnsDB {
removeSeedDB(dbPath)
@@ -945,13 +956,13 @@ func (b *ExecBackend) startChild(ctx context.Context, rt registry.Route) (Proc,
return p, spawnVer, nil
}
-// agentOwnedVars composes the child's agent-owned environment variables —
-// the control socket, the purge channel, the region — on top of fileEnv,
-// exactly as startChild did inline before the unit spawn path needed the
-// same composition. Extracted so the two paths cannot drift (design §7.2):
-// haveControl is whether this call's control listener actually bound
-// (childVars must follow the OUTCOME of the bind, not the intention — see
-// the long comment this replaced, still true here).
+// agentOwnedVars composes the child's agent-owned environment variables,
+// namely the control socket, the purge channel, and the region, on top of
+// fileEnv, exactly as startChild did inline before the unit spawn path
+// needed the same composition. Extracted so the two paths cannot drift
+// (design §7.2): haveControl is whether this call's control listener
+// actually bound. childVars must follow the OUTCOME of the bind, not the
+// intention, per the long comment this replaced, still true here.
func (b *ExecBackend) agentOwnedVars(rt registry.Route, fileEnv map[string]string, haveControl bool) map[string]string {
childVars := fileEnv
if haveControl {
@@ -973,24 +984,24 @@ func (b *ExecBackend) agentOwnedVars(rt registry.Route, fileEnv map[string]strin
}
// normalizeSocketMode is the PLATFORM's 0660 on a spawned instance's socket
-// (issue #103), not the app's — extracted from startChild so the unit path
+// (issue #103), not the app's. Extracted from startChild so the unit path
// (whose own bind is the trampoline's, not this process's exec) applies the
// identical rule rather than a second copy that can drift. So a deployment's
-// enrolled cross-user consumers (the carlos-sockets group — group-inherited
+// enrolled cross-user consumers (the carlos-sockets group, group-inherited
// from the setgid /run/carlos, see host.tmpfilesConf) can dial, no matter
-// what the child's umask or its own post-bind chmod produced — titogo
+// what the child's umask or its own post-bind chmod produced. titogo
// tightened its socket to 0600 and stranded the sibling router that proxies
// to it. Called after the health wait, so the app's own bind/chmod has
// already happened and this wins.
//
// The two guards make "only a socket THIS backend's child bound" true
// rather than merely intended. waitHealthy tolerates a squatter (its own
-// contract — it only dials), so a child that lost the bind race with a
+// contract: it only dials), so a child that lost the bind race with a
// stray supervisor (EADDRINUSE, exits) can still see the health wait
// succeed off the SQUATTER's answers; a dead child means the socket is
// not ours to widen (that route is the #17 external-socket advisory's
// territory). The Lstat keeps the chmod from ever following a
-// non-socket — a symlink planted at the route's path must not become a
+// non-socket: a symlink planted at the route's path must not become a
// mode change somewhere else. On a box with no consumers group the
// socket's group is just `carlos`, so the wider mode grants nobody
// anything. All three branches log and never fail the wake: a box with
@@ -1010,7 +1021,7 @@ func (b *ExecBackend) normalizeSocketMode(rt registry.Route, proc Proc) {
// startUnit is Start's nested body (design §6): the same admin token, env
// composition and health gate as startChild, but the process itself is
// started by the root instance-control channel from a launch file rather
-// than exec'd directly — the agent has no privilege to enter the tenant's
+// than exec'd directly. The agent has no privilege to enter the tenant's
// namespace, only the control service does.
func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, string, error) {
if err := b.nestedGate(rt); err != nil {
@@ -1065,12 +1076,12 @@ func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, s
// this here rather than waiting for it to happen inside up.Stop
// means the observer can never see a foreign invocation appear and
// fire its own redundant stop for a unit we already know we are
- // about to stop (controller ruling, fix round 1).
+ // about to stop.
up.stopRequested.Store(true)
stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
defer cancel()
if serr := up.Stop(stopCtx); serr != nil {
- // The health error, not this one, is the wake's failure — the
+ // The health error, not this one, is the wake's failure. The
// observer stands down on its own once the stop actually
// lands (stopRequested above), so an ErrStillRunning here
// just means it had not landed yet within actStopTimeout, not
@@ -1083,12 +1094,11 @@ func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, s
return up, spawnVer, nil
}
-// nestedGate is the pre-flight check every nested spawn path runs first
-// (controller ruling, fix round 1): no root instance-control channel
-// configured, or the box itself is not fit to run an isolated instance
-// (NestedOK's own mounts and /ready check), refuses before anything else
-// runs. Shared so seedUnit and startSidecarUnit gate on NestedOK exactly
-// like startUnit does — before this ruling, only startUnit checked it.
+// nestedGate is the pre-flight check every nested spawn path runs first: no
+// root instance-control channel configured, or the box itself is not fit
+// to run an isolated instance (NestedOK's own mounts and /ready check),
+// refuses before anything else runs. Shared so seedUnit and
+// startSidecarUnit gate on NestedOK exactly like startUnit does.
func (b *ExecBackend) nestedGate(rt registry.Route) error {
if b.Control == nil {
return fmt.Errorf("%s: instance control channel not configured", rt.Host)
@@ -1279,7 +1289,7 @@ func (b *ExecBackend) startSidecarChild(_ context.Context, rt registry.Route) (P
}
// startSidecarUnit is StartSidecar's nested body: a sidecar has no health
-// wait either way, and no control listener of its own — sidecar env is a
+// wait either way, and no control listener of its own. Sidecar env is a
// single layer, read straight from its own file, never merged with the
// instance's two-layer config.
func (b *ExecBackend) startSidecarUnit(ctx context.Context, rt registry.Route) (Proc, error) {
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go
index 4969187c8..d1a7ddd19 100644
--- a/internal/activator/backend_unit.go
+++ b/internal/activator/backend_unit.go
@@ -306,6 +306,20 @@ func (b *SplitBackend) NextDue(ctx context.Context, rt registry.Route) (int64, e
return b.pick(rt).NextDue(ctx, rt)
}
func (b *SplitBackend) HasSidecar(rt registry.Route) bool { return b.pick(rt).HasSidecar(rt) }
+
+// Layout implements Classifier for the composed backend: a unit-backed row
+// is always today's flat unit shape (the exec-only layouts do not apply to
+// it), and an exec row defers to the exec side's own classification.
+func (b *SplitBackend) Layout(rt registry.Route) (Layout, string) {
+ if rt.Backing == registry.BackingUnit {
+ return LayoutFlat, ""
+ }
+ if c, ok := b.Exec.(Classifier); ok {
+ return c.Layout(rt)
+ }
+ return LayoutFlat, ""
+}
+
func (b *SplitBackend) StartSidecar(ctx context.Context, rt registry.Route) (Proc, error) {
return b.pick(rt).StartSidecar(ctx, rt)
}
diff --git a/internal/activator/layout.go b/internal/activator/layout.go
index 4efc111f6..61f9ddd85 100644
--- a/internal/activator/layout.go
+++ b/internal/activator/layout.go
@@ -5,6 +5,7 @@ package activator
import (
"fmt"
+ "log"
"path/filepath"
"github.com/carlosframework/platform/internal/provision"
@@ -63,6 +64,15 @@ func ActivatorManaged(rt registry.Route) bool {
return rt.Kind == "instance" && rt.Hibernate && rt.Socket != "" && rt.Backing == "" && rt.Tunnel == ""
}
+// Classifier lets doWake ask a backend for a route's layout before it does
+// anything the layout might forbid: acquiring the lease or restoring a
+// database. A backend that does not implement it (a test double, say) is
+// simply not asked, and the in-Start/in-Seed refusal still catches a fenced
+// or conflicting row after the fact.
+type Classifier interface {
+ Layout(rt registry.Route) (Layout, string)
+}
+
// Classify returns the layout and, for anything but flat and nested, the
// reason a member or operator sees.
func Classify(rt registry.Route, in LayoutInputs) (Layout, string) {
@@ -75,7 +85,14 @@ func Classify(rt registry.Route, in LayoutInputs) (Layout, string) {
same = func(a, b string) bool { return provision.InstanceID(a) == provision.InstanceID(b) }
}
peers, err := in.Peers()
- if err == nil {
+ if err != nil {
+ // Fail open, not the fenced/conflict refusal itself: a registry
+ // read that fails here would otherwise silently skip the
+ // conflict check with no trace that it never ran. Loud, so a
+ // registry problem during a wake shows up in carlos logs rather
+ // than only in a collision nobody could explain later.
+ log.Printf("activator: %s: peer list for conflict check: %v (proceeding)", rt.Host, err)
+ } else {
for _, p := range peers {
if p.Host != rt.Host && ActivatorManaged(p) && same(p.Host, rt.Host) {
return LayoutConflict, fmt.Sprintf("instance identifier collides with %s; neither host wakes until one is removed", p.Host)
diff --git a/internal/activator/maintenance_test.go b/internal/activator/maintenance_test.go
index 90f2196b6..8a78ed15f 100644
--- a/internal/activator/maintenance_test.go
+++ b/internal/activator/maintenance_test.go
@@ -380,16 +380,16 @@ func TestMaintenanceFailedBackendStartRemainsUnproven(t *testing.T) {
}
}
-// 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.
+// Before design §6.7 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. Moving the
+// call off a.mu means a stalled restart can't block anything else,
+// including 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{})
@@ -399,7 +399,7 @@ 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
- // sweepOnce itself now returns immediately — the restart it launches
+ // sweepOnce itself now returns immediately. The restart it launches
// runs in its own goroutine, which is what's actually gated below.
a.sweepOnce()
<-entered
@@ -529,7 +529,7 @@ 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
+ // 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) })
}
diff --git a/internal/activator/paths.go b/internal/activator/paths.go
index 00d207288..a3901caec 100644
--- a/internal/activator/paths.go
+++ b/internal/activator/paths.go
@@ -23,6 +23,13 @@ type instancePaths struct {
configEnv string
}
+// Layout is ExecBackend's Classifier implementation: the same
+// classification pathsFor already computes, exposed so doWake can check it
+// before Acquire or Restore ever run.
+func (b *ExecBackend) Layout(rt registry.Route) (Layout, string) {
+ return Classify(rt, b.layoutInputs())
+}
+
func (b *ExecBackend) pathsFor(rt registry.Route) instancePaths {
l, reason := Classify(rt, b.layoutInputs())
p := instancePaths{layout: l, reason: reason, dataDir: b.DataDir, runDir: provision.SocketDir}
@@ -51,7 +58,7 @@ func (b *ExecBackend) pathsFor(rt registry.Route) instancePaths {
// against, so a route built against the override dir still classifies as
// nested through Task 2's sockets() helper.
func (b *ExecBackend) layoutInputs() LayoutInputs {
- in := b.Layout
+ in := b.LayoutFacts
in.DataDir = b.DataDir
if b.RunDirOverride != "" {
in.SocketDir = b.RunDirOverride
diff --git a/internal/activator/prepare.go b/internal/activator/prepare.go
index 25cbad8f9..c776ae26d 100644
--- a/internal/activator/prepare.go
+++ b/internal/activator/prepare.go
@@ -73,7 +73,7 @@ func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instanc
if role == instancectl.RoleRun {
// Nothing is running: either this is the first invocation of this
// role for this host, or the unit is not active (Start would not
- // have reached prepare otherwise) — so any file at rt.Socket is a
+ // have reached prepare otherwise), so any file at rt.Socket is a
// leftover from an earlier invocation, most likely one that was
// SIGKILLed before its own bind was ever established. On the
// nested path the agent itself never binds this socket (the app
@@ -81,7 +81,7 @@ func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instanc
// so unlike startChild's flat spawn there is no "our own live
// process squatting on it" case this could ever race: removing it
// here is safe by construction, and leaving it in place would fail
- // EVERY later wake's health check with nothing but a dial timeout —
+ // EVERY later wake's health check with nothing but a dial timeout,
// no member-facing diagnosis at all, unlike the flat path's #129
// handling, which at least turns the bind failure into a named
// error (memberSafeWakeReason).
@@ -101,7 +101,7 @@ func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instanc
// listenControl binds the control socket exactly as startChild does for an
// exec child (the tokenReader and schedule mount are unchanged, aside from
// reading beneath the instance directory rather than following a plain
-// path — design §3.2); a bind failure is logged and the instance runs
+// path, design §3.2); a bind failure is logged and the instance runs
// without its control verbs, as today.
func (b *ExecBackend) listenControl(rt registry.Route, envPath string) *edge.ControlListener {
tokenReader := func() (string, error) {
diff --git a/internal/activator/unitproc.go b/internal/activator/unitproc.go
index 47ce5de61..480a6a3a0 100644
--- a/internal/activator/unitproc.go
+++ b/internal/activator/unitproc.go
@@ -26,7 +26,7 @@ var ErrStillRunning = errors.New("activator: stop unproven; processes remain")
// unitProc is a nested host's Proc (design §6.3): the unit the control
// service started, the listeners the agent holds for it, and the observer
// that proves termination. Named apart from backend_unit.go's svcUnitProc,
-// a whole-app tenant's own systemd unit — this one is a nested exec
+// a whole-app tenant's own systemd unit: this one is a nested exec
// instance the activator spawns through the root instance-control channel.
type unitProc struct {
b *ExecBackend
@@ -40,11 +40,18 @@ type unitProc struct {
exited atomic.Bool
done chan struct{}
closeOnce atomic.Bool
+ // listenersArmed guards the done-watcher goroutine Stop's ctx.Done
+ // branch starts: several callers bound their wait on Stop shorter than
+ // the unit may actually take to die (a sidecar co-stop, a discarded
+ // sidecar, a failed health wait) and never call Stop again once it
+ // returns ErrStillRunning, which otherwise leaves ctl and logs open
+ // for as long as the agent runs. Armed at most once per unitProc.
+ listenersArmed atomic.Bool
// stopRequested is set before the FIRST Control.Stop this proc's owner
- // ever issues for it (fix round 1, controller ruling): once true, the
- // observer stops treating a differently-invoked or not-yet-invoked unit
- // as a FOREIGN restart to stop on its own — it is instead the proof
- // that our own invocation is already over (see observe).
+ // ever issues for it: once true, the observer stops treating a
+ // differently-invoked or not-yet-invoked unit as a FOREIGN restart to
+ // stop on its own. It is instead the proof that our own invocation is
+ // already over (see observe).
stopRequested atomic.Bool
}
@@ -79,6 +86,19 @@ func (p *unitProc) Stop(ctx context.Context) error {
p.closeListeners()
return nil
case <-ctx.Done():
+ // The caller is giving up on waiting, not on the stop itself: the
+ // unit may still be mid-drain, so the listeners must stay open while
+ // a writer could still exist. But several callers never call Stop
+ // again, and closing them still has to happen once done eventually
+ // closes, or they leak for the life of the agent.
+ // One goroutine per unitProc, armed once, does that regardless of
+ // whether anyone ever retries.
+ if !p.listenersArmed.Swap(true) {
+ go func() {
+ <-p.done
+ p.closeListeners()
+ }()
+ }
return ErrStillRunning
}
}
@@ -134,14 +154,13 @@ func (b *ExecBackend) observe(p *unitProc) {
// A stop of OUR OWN invocation is outstanding (Stop was already
// called): systemd starts a fresh invocation of the same unit only
// once the previous one's stop job has fully completed, so seeing
- // anything that is not provably that same old invocation — a
+ // anything that is not provably that same old invocation, a
// different id, or none reported yet while the unit sits in a
- // non-terminal ActiveState (mid-cycle to a fresh one) — is itself
+ // non-terminal ActiveState (mid-cycle to a fresh one), is itself
// proof our invocation is over. No cgroup check is needed here:
// there is no "our process might still be running" left to
// disprove, because systemd's own state machine already disproved
- // it by allowing a new invocation to begin (controller ruling, fix
- // round 1).
+ // it by allowing a new invocation to begin.
if p.stopRequested.Load() && err == nil && !terminal && (foreign || inv == "") {
emitEvent(b.Logs, p.rt, "warn", "instance exited (status unknown: replaced after stop)")
p.exited.Store(true)
@@ -161,7 +180,7 @@ func (b *ExecBackend) observe(p *unitProc) {
if terminal || (foreign && foreignStopped) {
exists, populated, eerr := events(p.slice, p.unit)
if eerr != nil {
- // A read error is not evidence of anything — not exited,
+ // A read error is not evidence of anything. Not exited,
// not still running, just unknown. Treating it as "cgroup
// empty" would let a still-running process's exit be
// reported anyway, exactly what this proof scheme exists
@@ -206,12 +225,12 @@ func (b *ExecBackend) observe(p *unitProc) {
}
// exitLine words the event line from the advisory record observe already
-// read (fix round 1: read once, pass to both exitLine and exitLevel, so the
-// two can never disagree from reading the file twice with something
-// changing in between), the same way childProc does: a non-zero status is a
-// warn line, a clean exit is info, and no record is unknown. The record is
-// tenant-writable text (design §6.5, final review of plan 2), so every
-// field that reaches the line passes through recordToken first.
+// read once and passes to both exitLine and exitLevel, so the two can
+// never disagree from reading the file twice with something changing in
+// between, the same way childProc does: a non-zero status is a warn line,
+// a clean exit is info, and no record is unknown. The record is
+// tenant-writable text (design §6.5), so every field that reaches the line
+// passes through recordToken first.
func exitLine(rec instancectl.ExitRecord, ok bool) string {
switch {
case !ok:
diff --git a/internal/activator/unitproc_test.go b/internal/activator/unitproc_test.go
index 05374a3ee..3c445dac1 100644
--- a/internal/activator/unitproc_test.go
+++ b/internal/activator/unitproc_test.go
@@ -27,14 +27,14 @@ type fakeControl struct {
startErr error
stopHook func(host string, role instancectl.Role)
readyErr error
- // onStart runs inside Start, in the fake control's own goroutine — the
+ // onStart runs inside Start, in the fake control's own goroutine, the
// same place the real control service's trampoline would exec the app
- // and the app would bind its socket (fix round 1: prepare's stale-socket
- // removal races a test's OWN pre-bound fixture socket when that bind
- // happens before Start is even called, so tests move it here instead).
+ // and the app would bind its socket: prepare's stale-socket removal
+ // races a test's OWN pre-bound fixture socket when that bind happens
+ // before Start is even called, so tests move it here instead.
onStart func()
// onSeed runs inside Seed, standing in for whatever a real seed process
- // might have written to disk before it failed — a test uses it to give
+ // might have written to disk before it failed. A test uses it to give
// seedOwnsDB cleanup something real to (not) remove.
onSeed func()
}
@@ -134,7 +134,7 @@ func TestStartUnitWritesTheLaunchFileAndReturnsAUnitProc(t *testing.T) {
b, rt := nestedBackend(t, ctl, unit)
// The fake control does not start anything, so health must be faked:
// bind a listener at rt.Socket that answers 200, from inside Start
- // itself — where the app's own bind would actually happen — so
+ // itself, where the app's own bind would actually happen, so
// prepare's stale-socket removal (which runs before Start ever calls
// the fake control) cannot race it.
var stopServing func()
@@ -259,7 +259,7 @@ func TestStartUnitRefusesWhenTheBoxIsNotReadyOrFenced(t *testing.T) {
t.Error("a start was issued on a box that is not ready")
}
b.NestedOK = nil
- b.Layout.Journal = func(string) bool { return true }
+ b.LayoutFacts.Journal = func(string) bool { return true }
if _, _, err := b.Start(context.Background(), rt); err == nil || !strings.Contains(err.Error(), "migration in flight") {
t.Fatalf("fenced wake = %v", err)
}
@@ -319,11 +319,11 @@ func (e *eventCapture) last() (capturedEvent, bool) {
return e.events[len(e.events)-1], true
}
-// TestUnitProcExitLineBoundsAForgedExitRecord covers fix round 1's
-// controller ruling on the exit record: it is tenant-writable text, so a
-// record with fields no real trampoline would ever write must still
-// produce a bounded, single-line, member-safe event — and the record must
-// be consumed (removed) exactly once.
+// TestUnitProcExitLineBoundsAForgedExitRecord covers the exit record's
+// ruling: it is tenant-writable text, so a record with fields no real
+// trampoline would ever write must still produce a bounded, single-line,
+// member-safe event, and the record must be consumed (removed) exactly
+// once.
func TestUnitProcExitLineBoundsAForgedExitRecord(t *testing.T) {
ctl := &fakeControl{inv: "inv1"}
unit := &fakeUnit{}
@@ -383,7 +383,7 @@ func TestUnitProcExitLineBoundsAForgedExitRecord(t *testing.T) {
// TestStartUnitWithEmptyInvocationFailsCleanly covers the case the control
// service reports no invocation at all: the unit already exited before its
// start was observed (plan 2 final review). The wake must fail with a
-// diagnosis naming that, and — because nothing is running — no unitProc
+// diagnosis naming that, and, because nothing is running, no unitProc
// is ever created to observe, so Control must never see a stop.
func TestStartUnitWithEmptyInvocationFailsCleanly(t *testing.T) {
ctl := &fakeControl{inv: ""}
@@ -399,8 +399,8 @@ func TestStartUnitWithEmptyInvocationFailsCleanly(t *testing.T) {
// TestStopKeepsListenersOpenUntilTerminationIsProven covers Stop's
// ordering contract: an unproven stop (ErrStillRunning) must leave the
-// control and log listeners open — a writer that may still exist must
-// still be able to reach them — and only a stop that observes done may
+// control and log listeners open, since a writer that may still exist must
+// still be able to reach them, and only a stop that observes done may
// close them.
func TestStopKeepsListenersOpenUntilTerminationIsProven(t *testing.T) {
ctl := &fakeControl{inv: "inv1"}
@@ -444,10 +444,56 @@ func TestStopKeepsListenersOpenUntilTerminationIsProven(t *testing.T) {
}
}
+// TestStopClosesListenersAfterDoneWithNoSecondStopCall pins the fix for the
+// leak Stop's ctx.Done branch used to have: several real callers (the
+// sidecar co-stop, a discarded sidecar, a failed health wait) never call
+// Stop a second time once it returns ErrStillRunning. The listeners must
+// still close once termination is proven, with nobody asking again.
+func TestStopClosesListenersAfterDoneWithNoSecondStopCall(t *testing.T) {
+ ctl := &fakeControl{inv: "inv1"}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ var stopServing func()
+ ctl.onStart = func() { stopServing = serveOK(t, rt.Socket) }
+ defer func() {
+ if stopServing != nil {
+ stopServing()
+ }
+ }()
+ unit.set("active", "inv1", true, true)
+ p, _, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ logSocket := b.logSocket(rt.Host, instancectl.RoleRun)
+
+ ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+ defer cancel()
+ if err := p.Stop(ctx); !errors.Is(err, ErrStillRunning) {
+ t.Fatalf("Stop = %v, want ErrStillRunning", err)
+ }
+
+ unit.set("inactive", "inv1", false, false)
+ waitFor(t, "the unit to exit", func() bool { return p.Exited() })
+
+ deadline := time.Now().Add(2 * time.Second)
+ for {
+ c, err := net.Dial("unix", logSocket)
+ if err != nil {
+ break // the listener closed on its own; no second Stop call was made
+ }
+ c.Close()
+ if time.Now().After(deadline) {
+ t.Fatal("log socket still accepts a dial 2s after the unit exited, with no second Stop call")
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+}
+
// TestSeedUnitFailureStopsBeforeCleanupAndKeepsTheDBWhenProcessesRemain
-// covers the controller ruling's ordering: Control.Stop for the seed unit
-// runs, and is proven by an empty cgroup, BEFORE any database cleanup —
-// and when the cgroup never reports empty, nothing is removed at all. The
+// covers the ordering rule: Control.Stop for the seed unit runs, and is
+// proven by an empty cgroup, BEFORE any database cleanup, and when the
+// cgroup never reports empty, nothing is removed at all. The
// wait is bounded by actStopTimeout, a const this package does not let a
// test shrink, so this test genuinely waits out that bound.
func TestSeedUnitFailureStopsBeforeCleanupAndKeepsTheDBWhenProcessesRemain(t *testing.T) {
@@ -456,7 +502,7 @@ func TestSeedUnitFailureStopsBeforeCleanupAndKeepsTheDBWhenProcessesRemain(t *te
b, rt := nestedBackend(t, ctl, unit)
rt.Seed = `["seed","--fast"]`
// Stand in for a real seed process that wrote its half-built database
- // before dying — created only once Seed actually runs, so the PRE-seed
+ // before dying, created only once Seed actually runs, so the PRE-seed
// stat that decides seedOwnsDB still sees it absent (seedOwnsDB=true).
ctl.onSeed = func() {
if err := os.WriteFile(rt.DBPath, []byte("half-built"), 0o600); err != nil {
diff --git a/internal/appconfig/syncexec.go b/internal/appconfig/syncexec.go
index 9b286db07..04d9865b8 100644
--- a/internal/appconfig/syncexec.go
+++ b/internal/appconfig/syncexec.go
@@ -903,11 +903,28 @@ func syncExecHost(rt registry.Route, m *Manifest, key string, priv *ecdh.Private
bounceLogged(rt.Host, bounce)
return
}
- cur, err := os.ReadFile(path)
- existed := err == nil
- if err != nil && !os.IsNotExist(err) {
- fmt.Printf(" %s: read %s: %v\n", rt.Host, path, err)
- return
+ // A nested row's current file lives on tenant-owned ground, so its read
+ // goes through instancefs the same way its write below does: a plain
+ // os.ReadFile would block forever on a FIFO a tenant planted at this
+ // name (isolated-exec design §3.3). The flat directory is agent-owned
+ // throughout and keeps the plain read it always used.
+ var cur []byte
+ var existed bool
+ var err error
+ if nestedRoute(dataDir, rt) {
+ cur, err = instancefs.ReadBeneathLimit(filepath.Dir(path), filepath.Base(path), 1<<20)
+ existed = err == nil
+ if err != nil && !os.IsNotExist(err) {
+ fmt.Printf(" %s: read %s: %v\n", rt.Host, path, err)
+ return
+ }
+ } else {
+ cur, err = os.ReadFile(path)
+ existed = err == nil
+ if err != nil && !os.IsNotExist(err) {
+ fmt.Printf(" %s: read %s: %v\n", rt.Host, path, err)
+ return
+ }
}
// Where sync.go has its defaultMissing/lostKeys guard, this path has
// nothing — deliberately. That guard exists because /etc/carlos/apps
@@ -1087,7 +1104,13 @@ func reapNestedConfigEnv(live map[string]bool, dataDir string) {
if e.IsDir() {
continue
}
- if strings.HasSuffix(e.Name(), configEnvSuffix+".tmp") {
+ // A nested config file is written through instancefs.WriteAtomic,
+ // whose temp name is ".<name>.tmp-<hex>" beside the real name, not
+ // "<name>.tmp" the way the flat directory's plain rename leaves
+ // one: the leading dot and the "-<hex>" suffix are both load
+ // bearing here, or a crashed pass's leftover (it holds decrypted
+ // secrets) is never reaped.
+ if strings.HasPrefix(e.Name(), ".") && strings.Contains(e.Name(), configEnvSuffix+".tmp-") {
if err := os.Remove(filepath.Join(instDir, e.Name())); err != nil {
fmt.Printf(" %s: remove stale config temp file: %v\n", e.Name(), err)
}
diff --git a/internal/appconfig/syncexec_test.go b/internal/appconfig/syncexec_test.go
index ee6c48fbb..28c504177 100644
--- a/internal/appconfig/syncexec_test.go
+++ b/internal/appconfig/syncexec_test.go
@@ -12,6 +12,8 @@ import (
"testing"
"time"
+ "golang.org/x/sys/unix"
+
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/relay"
@@ -174,6 +176,80 @@ func TestSyncExecWritesANestedRouteIntoItsInstanceDir(t *testing.T) {
}
}
+// TestSyncExecFIFOAtANestedBundleDoesNotHangThePass pins the review fix: a
+// nested row's existing bundle is read through instancefs, so a FIFO a
+// tenant planted at that name (isolated-exec design §3.3, no tenant-planted
+// inode may block the agent) must fail fast rather than hang SyncExec's
+// whole pass forever behind one host's blocking read(2). A sibling host on
+// the same pass must still converge normally.
+func TestSyncExecFIFOAtANestedBundleDoesNotHangThePass(t *testing.T) {
+ ctx, s, reg, dd, priv, _ := execKit(t)
+ pk, err := ParsePrivateKey(priv, "test")
+ if err != nil {
+ t.Fatalf("ParsePrivateKey: %v", err)
+ }
+ if err := SetEnv(ctx, s, "bab", "eleven", map[string]string{"LOG_LEVEL": "debug"}); err != nil {
+ t.Fatalf("SetEnv: %v", err)
+ }
+ if err := SetEnv(ctx, s, "bab", "twelve", map[string]string{"LOG_LEVEL": "debug"}); err != nil {
+ t.Fatalf("SetEnv: %v", err)
+ }
+ okHost, fifoHost := "ok.example.com", "fifo.example.com"
+ for _, h := range []string{okHost, fifoHost} {
+ if err := os.MkdirAll(provision.InstanceDir(dd, h), 0o755); err != nil {
+ t.Fatalf("mkdir instance dir for %s: %v", h, err)
+ }
+ }
+ if err := reg.Upsert(registry.Route{
+ Host: okHost, Kind: "instance", App: "eleven", Account: "bab",
+ Status: "ready", DBPath: provision.NestedDBPath(dd, okHost), Socket: provision.NestedSocket(okHost),
+ }); err != nil {
+ t.Fatalf("Upsert %s: %v", okHost, err)
+ }
+ if err := reg.Upsert(registry.Route{
+ Host: fifoHost, Kind: "instance", App: "twelve", Account: "bab",
+ Status: "ready", DBPath: provision.NestedDBPath(dd, fifoHost), Socket: provision.NestedSocket(fifoHost),
+ }); err != nil {
+ t.Fatalf("Upsert %s: %v", fifoHost, err)
+ }
+ // Planted where syncExecHost's existed-check reads BEFORE it ever writes
+ // anything: a tenant that got here first, not a crash artifact.
+ fifoPath := filepath.Join(provision.InstanceDir(dd, fifoHost), fifoHost+".config.env")
+ if err := unix.Mkfifo(fifoPath, 0o600); err != nil {
+ t.Fatalf("Mkfifo: %v", err)
+ }
+
+ var bounced []string
+ bounce := func(h string) error { bounced = append(bounced, h); return nil }
+ done := make(chan error, 1)
+ go func() { done <- SyncExec(ctx, reg, s, pk, dd, bounce, "", nil) }()
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatalf("SyncExec: %v", err)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("SyncExec hung behind the FIFO'd host instead of failing that one host fast")
+ }
+
+ okPath := filepath.Join(provision.InstanceDir(dd, okHost), okHost+".config.env")
+ got, err := os.ReadFile(okPath)
+ if err != nil {
+ t.Fatalf("sibling host's config was not written: %v", err)
+ }
+ if string(got) != "LOG_LEVEL=debug\n" {
+ t.Fatalf("sibling config env = %q, want LOG_LEVEL=debug", got)
+ }
+ if len(bounced) != 1 || bounced[0] != okHost {
+ t.Fatalf("bounced = %v, want exactly [%s] (the FIFO'd host must not bounce)", bounced, okHost)
+ }
+ // The FIFO itself is untouched: syncExecHost's read failed before it
+ // ever reached a write.
+ if fi, err := os.Lstat(fifoPath); err != nil || fi.Mode()&os.ModeNamedPipe == 0 {
+ t.Fatalf("FIFO at %s was replaced or removed: %v %v", fifoPath, fi, err)
+ }
+}
+
// TestReapConfigEnvWalksInstanceDirs pins reapConfigEnv's nested half: a
// stale config file inside a gone host's instance directory is removed, and
// a live nested route's own file survives.
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 8f4b5e6c6..d2ff8f1c3 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -149,6 +149,7 @@ import (
"flag"
"fmt"
"io"
+ "io/fs"
"log"
"math/rand/v2"
"net"
@@ -313,10 +314,17 @@ func RunAgent(args []string) {
// handle, so a nested row's classification here always agrees with
// what the rest of this process already knows about the box.
eb.Control = instancectl.NewClient(controlsock.Path)
- eb.Layout = activator.LayoutInputs{
+ eb.LayoutFacts = activator.LayoutInputs{
DataDir: dd,
- Journal: func(host string) bool { _, err := os.Stat(provision.LayoutJournalPath(dd, host)); return err == nil },
- Peers: reg.List,
+ // A journal that cannot be read is not proof there isn't one: an
+ // EACCES or EIO on /data/layout must fence the row rather than
+ // wake it as if nothing were in flight, so only a definite
+ // ErrNotExist counts as "no journal".
+ Journal: func(host string) bool {
+ _, err := os.Stat(provision.LayoutJournalPath(dd, host))
+ return !errors.Is(err, fs.ErrNotExist)
+ },
+ Peers: reg.List,
}
eb.NestedOK = nestedGate(instancefs.MountNoSymfollow, eb.Control.Ready)
svc = activator.New(*cfg, &activator.SplitBackend{
diff --git a/internal/instancectl/client.go b/internal/instancectl/client.go
index 689b08eb0..33dfa32cc 100644
--- a/internal/instancectl/client.go
+++ b/internal/instancectl/client.go
@@ -54,6 +54,11 @@ func (e *NotReadyError) Error() string {
return "instance control: box not ready: " + strings.Join(fails, "; ")
}
+// httpClient builds a fresh Transport per call rather than caching one on c.
+// DisableKeepAlives is what makes that the right trade: a pooled idle
+// connection nothing ever reuses (each call already builds its own client)
+// would just sit open, leaking one file descriptor per call instead of
+// closing when the response is read.
func (c *Client) httpClient() *http.Client {
dial := c.Dial
if dial == nil {
diff --git a/internal/instancefs/instancefs.go b/internal/instancefs/instancefs.go
index acad5dcff..93cb41447 100644
--- a/internal/instancefs/instancefs.go
+++ b/internal/instancefs/instancefs.go
@@ -34,6 +34,12 @@ var ErrSymlink = errors.New("instancefs: path contains a symlink")
// ErrEscapes is returned for a name that is absolute or reaches above dir.
var ErrEscapes = errors.New("instancefs: path escapes its directory")
+// ErrNotRegular is returned when a read-only open resolves to anything
+// other than a regular file: a FIFO with no writer blocks open(2)
+// indefinitely, and a directory or device is never a legitimate target for
+// the files this package reads on a tenant's behalf.
+var ErrNotRegular = errors.New("instancefs: not a regular file")
+
// cleanRel validates name as a relative path that stays inside dir.
func cleanRel(name string) (string, error) {
if name == "" || filepath.IsAbs(name) {
diff --git a/internal/instancefs/instancefs_linux.go b/internal/instancefs/instancefs_linux.go
index 12fd93c5c..333d8a089 100644
--- a/internal/instancefs/instancefs_linux.go
+++ b/internal/instancefs/instancefs_linux.go
@@ -43,7 +43,14 @@ func openDir(dir string) (int, error) {
}
// OpenBeneath opens name under dir with openat2, following no symlink at
-// any component and never leaving dir.
+// any component and never leaving dir. A read-only open also carries
+// O_NONBLOCK on the wire and is then checked with fstat: a FIFO with no
+// writer blocks open(2) indefinitely, and a tenant can plant one anywhere
+// under an instance directory the agent reads by contract (its own run
+// directory, its own env files). O_NONBLOCK is a no-op for opening a
+// regular file, so this changes nothing for the read paths that are
+// already the common case; a non-regular target is refused with
+// ErrNotRegular instead of hanging the caller forever.
func OpenBeneath(dir, name string, flag int, perm os.FileMode) (*os.File, error) {
rel, err := cleanRel(name)
if err != nil {
@@ -54,8 +61,13 @@ func OpenBeneath(dir, name string, flag int, perm os.FileMode) (*os.File, error)
return nil, err
}
defer unix.Close(dfd)
+ wireFlags := uint64(flag)
+ readOnly := flag == os.O_RDONLY
+ if readOnly {
+ wireFlags |= unix.O_NONBLOCK
+ }
how := unix.OpenHow{
- Flags: uint64(flag) | unix.O_CLOEXEC,
+ Flags: wireFlags | unix.O_CLOEXEC,
Mode: uint64(perm.Perm()),
Resolve: resolveFlags,
}
@@ -69,7 +81,19 @@ func OpenBeneath(dir, name string, flag int, perm os.FileMode) (*os.File, error)
}
return nil, &os.PathError{Op: "openat2", Path: filepath.Join(dir, rel), Err: err}
}
- return os.NewFile(uintptr(fd), filepath.Join(dir, rel)), nil
+ path := filepath.Join(dir, rel)
+ if readOnly {
+ var st unix.Stat_t
+ if err := unix.Fstat(fd, &st); err != nil {
+ unix.Close(fd)
+ return nil, &os.PathError{Op: "fstat", Path: path, Err: err}
+ }
+ if st.Mode&unix.S_IFMT != unix.S_IFREG {
+ unix.Close(fd)
+ return nil, fmt.Errorf("%w: %s", ErrNotRegular, path)
+ }
+ }
+ return os.NewFile(uintptr(fd), path), nil
}
// WriteAtomic writes data to a fresh O_EXCL temp file beside name and
diff --git a/internal/instancefs/instancefs_linux_test.go b/internal/instancefs/instancefs_linux_test.go
index 674a2a73d..158f05a2e 100644
--- a/internal/instancefs/instancefs_linux_test.go
+++ b/internal/instancefs/instancefs_linux_test.go
@@ -8,6 +8,9 @@ import (
"path/filepath"
"strings"
"testing"
+ "time"
+
+ "golang.org/x/sys/unix"
)
func TestOpenBeneathReadsAPlainFile(t *testing.T) {
@@ -194,6 +197,31 @@ func TestMountOptionOfIsTheGeneralForm(t *testing.T) {
_ = got // / is not reliably ro or rw across test hosts; only the absence of an error matters here.
}
+// TestOpenBeneathRefusesAFIFOInsteadOfBlocking pins the review fix: a FIFO
+// with no writer blocks open(2) indefinitely, and a tenant can plant one at
+// any name the agent reads by contract. The read must fail fast with
+// ErrNotRegular rather than hang the caller, so this runs it in a goroutine
+// with its own timeout: a regression here must fail the test, not wedge it.
+func TestOpenBeneathRefusesAFIFOInsteadOfBlocking(t *testing.T) {
+ dir := t.TempDir()
+ if err := unix.Mkfifo(filepath.Join(dir, "h.env"), 0o600); err != nil {
+ t.Fatalf("Mkfifo: %v", err)
+ }
+ done := make(chan error, 1)
+ go func() {
+ _, err := ReadBeneath(dir, "h.env")
+ done <- err
+ }()
+ select {
+ case err := <-done:
+ if !errors.Is(err, ErrNotRegular) {
+ t.Fatalf("ReadBeneath a FIFO: err = %v, want ErrNotRegular", err)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("ReadBeneath blocked on a FIFO with no writer instead of refusing it")
+ }
+}
+
func TestReadBeneathLimitRefusesALargeFile(t *testing.T) {
dir := t.TempDir()
if err := os.WriteFile(filepath.Join(dir, "big"), make([]byte, 4096), 0o600); err != nil {
diff --git a/internal/lease/endpoint_test.go b/internal/lease/endpoint_test.go
index 935e9cb00..75fb3f9f1 100644
--- a/internal/lease/endpoint_test.go
+++ b/internal/lease/endpoint_test.go
@@ -10,9 +10,12 @@ package lease
import (
"context"
"encoding/json"
+ "fmt"
"io"
"net/http"
"net/http/httptest"
+ "os"
+ "path/filepath"
"sync"
"testing"
@@ -261,3 +264,40 @@ func TestResolveRegion_PrefersAWSRegionEnv(t *testing.T) {
t.Errorf("resolveRegion = %q, want %q (AWS_REGION must outrank the non-AWS default)", got, "nbg1")
}
}
+
+// TestRestoreCreatesTheOutputPathsParentDirectory pins the review fix: a
+// nested row's instance directory does not exist before its first wake (it
+// is created by the spawn path, which runs after Restore), so Restore must
+// create OutputPath's parent itself rather than fail with ENOENT before the
+// spawn path ever gets a chance to.
+func TestRestoreCreatesTheOutputPathsParentDirectory(t *testing.T) {
+ fakeCreds(t)
+ // Every listing this exercises, the v0.3.x/LTX format comparison and
+ // the LTX restore plan itself, is a ListObjectsV2 call; answering all
+ // of them with an empty bucket makes Restore take the "no replica"
+ // path, the one every fresh nested row actually takes on its first wake.
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Type", "application/xml")
+ fmt.Fprint(w, `<?xml version="1.0" encoding="UTF-8"?>
+<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>b</Name><KeyCount>0</KeyCount><IsTruncated>false</IsTruncated></ListBucketResult>`)
+ }))
+ defer srv.Close()
+ t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
+
+ root := t.TempDir()
+ dbPath := filepath.Join(root, "instances", "h.example.com", "h.example.com.db")
+ if _, err := os.Stat(filepath.Dir(dbPath)); !os.IsNotExist(err) {
+ t.Fatalf("test setup: parent directory already exists")
+ }
+
+ restored, err := Restore(context.Background(), dbPath, "test-bucket", "inst/h.example.com", "")
+ if err != nil {
+ t.Fatalf("Restore into a path whose parent is absent: %v", err)
+ }
+ if restored {
+ t.Fatal("restored = true against an empty bucket, want false (nothing to restore)")
+ }
+ if fi, err := os.Stat(filepath.Dir(dbPath)); err != nil || !fi.IsDir() {
+ t.Fatalf("Restore did not create the output path's parent directory: %v", err)
+ }
+}
diff --git a/internal/lease/lease.go b/internal/lease/lease.go
index 8755c2ed3..72761a941 100644
--- a/internal/lease/lease.go
+++ b/internal/lease/lease.go
@@ -15,6 +15,7 @@ import (
"context"
"errors"
"fmt"
+ "os"
"path/filepath"
"sync"
"time"
@@ -410,6 +411,15 @@ func Restore(ctx context.Context, dbPath, bucket, prefix, region string) (bool,
}
client := newReplicaClient(bucket, prefix, region)
+ // A nested row's instance directory may not exist yet the first time it
+ // wakes: the directory is created by the spawn path (isolated-exec
+ // design §3.1), which runs AFTER this restore. Without this, a fresh
+ // nested row's very first wake fails with ENOENT before it ever gets to
+ // create anything.
+ if err := os.MkdirAll(filepath.Dir(dbPath), 0o700); err != nil {
+ return false, fmt.Errorf("create %s: %w", filepath.Dir(dbPath), err)
+ }
+
// A nil *DB is litestream's own pattern for URL restores (the CLI does the
// same): Restore only needs the replica client and the output path.
r := litestream.NewReplicaWithClient(nil, client)
diff --git a/internal/schedule/runner.go b/internal/schedule/runner.go
index b13b08001..888a5a784 100644
--- a/internal/schedule/runner.go
+++ b/internal/schedule/runner.go
@@ -6,7 +6,6 @@ import (
"fmt"
"log"
"net/http"
- "path/filepath"
"strconv"
"sync"
"time"
@@ -459,8 +458,8 @@ func BuildRequest(rt registry.Route, t registry.Timer, token, build string) (*ht
// keep <host>.env inside their own instance directory (design §7), and the
// host wiring (schedule_wiring.go) resolves that through the same
// classification the exec backend used to decide where it spawned the
-// instance. TokenFromEnvFile below is this with the flat resolver baked in,
-// which is every caller that predates the nested layout.
+// instance. Every caller passes a resolver here now; a caller that predates
+// the nested layout gets one that always names the flat path.
func TokenFromEnvPaths(envPath func(host string) string) func(host string) (string, error) {
return func(host string) (string, error) {
env, err := readEnvFile(envPath(host))
@@ -473,9 +472,3 @@ func TokenFromEnvPaths(envPath func(host string) string) func(host string) (stri
return env["CARLOS_ADMIN_TOKEN"], nil
}
}
-
-// TokenFromEnvFile is the production Token: CARLOS_ADMIN_TOKEN from
-// <dataDir>/<host>.env, the file the activator owns (spec §6.4).
-func TokenFromEnvFile(dataDir string) func(host string) (string, error) {
- return TokenFromEnvPaths(func(host string) string { return filepath.Join(dataDir, host+".env") })
-}