amadan

carlos / platform Public

Sidecar isolation Task 9 fix round 1: a refused respawn stands down; D16 witnessed through the real leg; aliased os/exec imports enumerated

A worker respawn the flat-exec policy refuses (crash restart, max_run
recycle, converge or a replacement attempt) stands the entry down through
the normal proven-stop path, releasing its lease, instead of leaving the
dead child to the sweep's backoff; the refusal is logged once per host per
hour. ReplaceWorker reports the policy as the abandonment reason and
restores nothing. Tests drive ordinary supervision and a replacement. The
D16 nudge is witnessed through newLayoutLeg over runLayoutLeg, a real
layout.Pass and registry: a refusal landing mid-pass is kept and moves the
flat worker row nested; a foreign row and a reported spawn schedule
nothing. The exec.Command enumeration resolves os/exec's local import name.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude f672e3acfd1cf831e9eb9d8780dbc5c898f8c4ac parent f434efe
5 files changed, +324 −6
  • internal/activator/activator.go +16 −0
  • internal/activator/flatexec_test.go +94 −0
  • internal/activator/spawn_enumeration_test.go +46 −6
  • internal/activator/workerswap.go +9 −0
  • internal/host/flatexec_nudge_linux_test.go +159 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index c4dc469ba..738b9a6c9 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -3813,6 +3813,22 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
a.beginStopLocked(e, StopRestart)
a.mu.Unlock()
log.Printf("activator: worker %s NOT respawned (%s): %v — standing down and releasing its lease", host, why, refused)
+ case errors.Is(err, ErrFlatExecRefused):
+ // The flat-exec policy refused the exec (sidecar isolation D13):
+ // this box does not start the worker in the host namespace, so
+ // no retry of the same spawn can succeed, and leaving the dead
+ // child to the sweep's crash restart would be a backoff loop
+ // that never ends. Stand down as a config refusal does: a normal
+ // stop releases the lease through the proven-stop path, and the
+ // next wake (a timer, a request, the always-on sweep) meets the
+ // policy again, by which time D16's nudge may have moved the row
+ // nested. An attempt's veto was recorded above, so ReplaceWorker
+ // names the policy. Logged once per host per hour, as the
+ // policy's own line is: a tenant whose row stays flat would
+ // otherwise earn a line at every wake.
+ a.beginStopLocked(e, StopRestart)
+ a.mu.Unlock()
+ logPerFile("flat-exec-respawn", host, fmt.Sprintf("activator: worker %s NOT respawned (%s): %v; standing down and releasing its lease until its row runs nested or is allowlisted", host, why, err))
case verdictOf(err) != gateProceed:
// The exec's own ownership check refused (startWorker's gate),
// and ownership came back before the install check (a fence
diff --git a/internal/activator/flatexec_test.go b/internal/activator/flatexec_test.go
index c88225ab5..0429d346c 100644
--- a/internal/activator/flatexec_test.go
+++ b/internal/activator/flatexec_test.go
@@ -1,11 +1,14 @@
package activator
import (
+ "context"
"errors"
"fmt"
"os/exec"
"path/filepath"
+ "strings"
"testing"
+ "time"
"github.com/carlosframework/platform/internal/cgroup"
"github.com/carlosframework/platform/internal/hostbox"
@@ -206,3 +209,94 @@ func TestSwapGateClassifiesFlatExecRefused(t *testing.T) {
t.Fatalf("verdictOf(claimAdmission(ErrFlatExecRefused)) = %v, want gateNoOwnership", v)
}
}
+
+// D13's "recorded veto, not a crash loop", for ordinary supervision: a
+// worker whose crash restart or max_run recycle the flat-exec policy refuses
+// stands down (the entry goes, its lease is released through the normal
+// stop, replication closes) instead of leaving a dead child for the sweep to
+// retry under backoff. The next wake meets the policy again.
+func TestWorkerRespawnRefusedByFlatExecStandsDown(t *testing.T) {
+ t.Parallel()
+ for _, why := range []string{"crash", "max_run"} {
+ t.Run(why, func(t *testing.T) {
+ a, b, clk, _ := testActivator(t, time.Hour)
+ host := "bot.x.internal"
+ rt := alwaysOnWorkerRoute(host)
+ rt.MaxRunSeconds = 30
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ if b.workerStarts() != 1 {
+ t.Fatalf("workerStarts = %d, want 1", b.workerStarts())
+ }
+ first := b.lastWorker()
+
+ // The operator set CARLOS_FLAT_EXEC=refuse and restarted the
+ // agent: the backend now refuses every host-namespace exec.
+ b.mu.Lock()
+ b.workerErr = fmt.Errorf("%s: %w: layout flat", host, ErrFlatExecRefused)
+ b.mu.Unlock()
+
+ if why == "crash" {
+ first.kill()
+ } else {
+ clk.advance(31 * time.Second)
+ }
+ a.sweepOnce()
+
+ waitFor(t, "the refused worker to stand down", func() bool { return a.entryCount() == 0 })
+ if got := b.lastWorker(); got != first {
+ t.Fatalf("a refused respawn installed a child: %+v", got)
+ }
+ if !first.isStopped() && why == "max_run" {
+ t.Error("the old child was left running")
+ }
+ if n := b.releaseCount(); n != 1 {
+ t.Fatalf("lease releases = %d, want 1: a refused worker must give its lease back", n)
+ }
+ if !b.lastRep().isClosed() {
+ t.Error("replication was left open on a refused worker")
+ }
+ // And it does not loop: further sweeps start nothing.
+ for i := 0; i < 3; i++ {
+ clk.advance(time.Minute)
+ a.sweepOnce()
+ }
+ if n := b.workerStarts(); n != 2 {
+ t.Fatalf("workerStarts = %d, want 2 (the wake and the one refused exec): the refusal looped", n)
+ }
+ })
+ }
+}
+
+// And for a replacement attempt: the respawn's refusal is the attempt's
+// recorded veto, ReplaceWorker reports abandonment naming the policy, the
+// replaced build is not restored (it would be refused for the same host),
+// and the entry stands down.
+func TestReplaceWorkerRefusedByFlatExecStandsDown(t *testing.T) {
+ shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ refusal := fmt.Errorf("%s: %w: layout flat", host, ErrFlatExecRefused)
+ b.smu.Lock()
+ b.failNextStaged = refusal
+ b.smu.Unlock()
+ b.mu.Lock()
+ b.workerErr = refusal // a restore would be refused too
+ b.mu.Unlock()
+ good := fakeIdentity("v1")
+ res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), &good, one)
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, ErrFlatExecRefused.Error()) || res.Executed {
+ t.Fatalf("ReplaceWorker = %+v, want abandoned naming the flat-exec refusal, not executed", res)
+ }
+ waitFor(t, "the refused worker to stand down", func() bool { return a.entryCount() == 0 })
+ if calls := b.calls(); len(calls) != 2 || calls[1] == nil || calls[1].Version != "v2" {
+ t.Fatalf("spawns %v, want the wake and the refused attempt only: nothing restored", calls)
+ }
+ if n := b.releaseCount(); n != 1 {
+ t.Fatalf("lease releases = %d, want 1", n)
+ }
+}
diff --git a/internal/activator/spawn_enumeration_test.go b/internal/activator/spawn_enumeration_test.go
index cc3b4bbe8..5d03e9207 100644
--- a/internal/activator/spawn_enumeration_test.go
+++ b/internal/activator/spawn_enumeration_test.go
@@ -55,6 +55,12 @@ func TestEveryExecCommandReachesASeam(t *testing.T) {
if err != nil {
t.Fatal(err)
}
+ // The local name of os/exec in this file: an aliased import must
+ // not hide a spawn. A file that does not import it has no spawn.
+ execName, imported := execImportName(f)
+ if !imported {
+ continue
+ }
// A stack of the enclosing call expressions, so an exec.Command
// inside a function literal can ask what that literal was passed to.
var stack []ast.Node
@@ -68,16 +74,12 @@ func TestEveryExecCommandReachesASeam(t *testing.T) {
if !ok {
return true
}
- sel, ok := call.Fun.(*ast.SelectorExpr)
+ fn, ok := execCall(call, execName)
if !ok {
return true
}
- pkg, ok := sel.X.(*ast.Ident)
- if !ok || pkg.Name != "exec" || (sel.Sel.Name != "Command" && sel.Sel.Name != "CommandContext") {
- return true
- }
where := fset.Position(call.Pos()).String()
- if lit, ok := firstArgLiteral(call, sel.Sel.Name); ok && platformOwnCommands[lit] {
+ if lit, ok := firstArgLiteral(call, fn); ok && platformOwnCommands[lit] {
return true
}
if seamOf(stack[:len(stack)-1]) != "" {
@@ -97,6 +99,44 @@ func TestEveryExecCommandReachesASeam(t *testing.T) {
}
}
+// execImportName is the name os/exec goes by in f: its alias when the
+// import has one, "exec" otherwise, and "." for a dot import, whose
+// Command and CommandContext are bare identifiers. imported is false when
+// f does not import os/exec at all.
+func execImportName(f *ast.File) (name string, imported bool) {
+ for _, imp := range f.Imports {
+ path, err := strconv.Unquote(imp.Path.Value)
+ if err != nil || path != "os/exec" {
+ continue
+ }
+ if imp.Name != nil {
+ return imp.Name.Name, true
+ }
+ return "exec", true
+ }
+ return "", false
+}
+
+// execCall reports whether call is exec.Command or exec.CommandContext
+// under the file's local name for os/exec, and which.
+func execCall(call *ast.CallExpr, execName string) (fn string, ok bool) {
+ isFn := func(n string) bool { return n == "Command" || n == "CommandContext" }
+ switch f := call.Fun.(type) {
+ case *ast.SelectorExpr:
+ pkg, ok := f.X.(*ast.Ident)
+ if !ok || pkg.Name != execName || execName == "." || !isFn(f.Sel.Name) {
+ return "", false
+ }
+ return f.Sel.Name, true
+ case *ast.Ident:
+ if execName != "." || !isFn(f.Name) {
+ return "", false
+ }
+ return f.Name, true
+ }
+ return "", false
+}
+
// firstArgLiteral is the command's executable when it is a string literal:
// the first argument of exec.Command, the second of exec.CommandContext.
func firstArgLiteral(call *ast.CallExpr, fn string) (string, bool) {
diff --git a/internal/activator/workerswap.go b/internal/activator/workerswap.go
index f3d9083aa..ed120bbfe 100644
--- a/internal/activator/workerswap.go
+++ b/internal/activator/workerswap.go
@@ -116,6 +116,15 @@ func (a *Service) ReplaceWorker(ctx context.Context, at Attempt, fallback *relea
veto := e.attemptVeto
executed = e.attemptExec
e.attemptVeto, e.attemptExec = nil, false
+ if errors.Is(veto, ErrFlatExecRefused) {
+ // The flat-exec policy refused the attempt's exec (sidecar
+ // isolation D13): nothing of its build ran, and the respawn stood
+ // the entry down, since the build it replaced would be refused for
+ // the same host. Nothing is restored; the result names the policy
+ // rather than the stop it caused.
+ a.mu.Unlock()
+ return abandon(fmt.Errorf("re-checked at the spawn: %w", veto))
+ }
if a.entries[at.Host] != e || e.state != actAwake {
a.mu.Unlock()
return abandon(errSwapStopped)
diff --git a/internal/host/flatexec_nudge_linux_test.go b/internal/host/flatexec_nudge_linux_test.go
new file mode 100644
index 000000000..bdbcfc79d
--- /dev/null
+++ b/internal/host/flatexec_nudge_linux_test.go
@@ -0,0 +1,159 @@
+//go:build linux
+
+package host
+
+import (
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/layout"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// D16 through the production wiring: a refusal of a flat worker row on a
+// box whose effective intent is nested kicks the real layout leg, whose
+// run is runLayoutLeg over a real layout.Pass and a real registry, and the
+// pass moves the row nested. The refusal lands while a pass is already
+// running, so it also pins that the kick is kept (the leg's dirty flag)
+// and the pass runs once more after the current one. A foreign row's
+// refusal schedules nothing and its row stays as it was; under report
+// nothing is nudged either.
+func TestRefusedManagedRowNudgesTheRealLayoutLeg(t *testing.T) {
+ resetFlatExec(t)
+ resetLayoutGlobals(t)
+ ctx, cancel := context.WithCancel(context.Background())
+ t.Cleanup(cancel)
+
+ dd := t.TempDir()
+ // The socket dir is short on purpose: a nested run dir must fit a unix
+ // socket address, which the pass's preflight enforces.
+ rd, err := os.MkdirTemp("", "fx")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(rd) })
+ reg := restartTestRegistry(t)
+ flat := "job.shop.abc123" + provision.SidecarSuffix
+ foreign := "edited.shop.abc123" + provision.SidecarSuffix
+ flatRow := registry.Route{Host: flat, Kind: "sidecar", Hibernate: true, Status: "ready", Args: `["work"]`, App: "shop", Account: "acct", DBPath: provision.DefaultDBPath(dd, flat)}
+ foreignRow := registry.Route{Host: foreign, Kind: "sidecar", Hibernate: true, Status: "ready", Args: `["work"]`, App: "shop", Account: "acct", DBPath: filepath.Join(dd, "tenant-wrote-this.db")}
+ for _, rt := range []registry.Route{flatRow, foreignRow} {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ for path, body := range map[string]string{
+ provision.DefaultDBPath(dd, flat): "db " + flat,
+ provision.DefaultDBPath(dd, flat) + "-wal": "wal",
+ filepath.Join(dd, flat+".env"): "CARLOS_ADMIN_TOKEN=t\n",
+ filepath.Join(dd, flat+".config.env"): "LOG_LEVEL=debug\n",
+ foreignRow.DBPath: "db " + foreign,
+ } {
+ if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ pass := &layout.Pass{
+ Registry: reg, DataDir: dd, SocketDir: rd, SocketsGID: -1, Now: time.Now, SweptAtStart: true, Logf: t.Logf,
+ WorkersReady: func() bool { return true },
+ Replicator: func(string) (bool, string) { return false, "" },
+ }
+ armedLayoutPass.Store(pass)
+
+ // The leg as fastConvergeLoop builds it, over a real intent store. The
+ // first run (the arming kick) finds no intent and a flat deployment
+ // default, converges nothing, and is then held so a refusal can land
+ // while the leg is running.
+ t.Setenv(layoutDefaultEnv, "flat")
+ store := release.NewDirStore(t.TempDir())
+ var runs atomic.Int32
+ firstDone, release1 := make(chan struct{}), make(chan struct{})
+ leg := newLayoutLeg(ctx, func(ctx context.Context) bool {
+ retry := runLayoutLeg(ctx, store, "box-1")
+ if runs.Add(1) == 1 {
+ close(firstDone)
+ <-release1
+ }
+ return retry
+ })
+ kick := func() { leg.kick(ctx) }
+ layoutNudge.Store(&kick)
+ <-firstDone
+ if rt, _ := reg.Get(flat); rt.DBPath != flatRow.DBPath {
+ t.Fatalf("the arming run moved the row with no intent recorded: %+v", rt)
+ }
+
+ // The fleet records a nested intent for this box (the leg has not read
+ // it yet: it is held), and the agent's published intent is nested.
+ appendIntent(t, store, release.LayoutModeNested, "box-1")
+ noteLayoutIntent(release.LayoutModeNested)
+
+ // Under refuse, the agent's policy refuses the flat worker's wake while
+ // the pass is still running: the nudge must be kept, not lost.
+ policy := armFlatExec(envOf(map[string]string{"CARLOS_FLAT_EXEC": "refuse"}), nudgeLayoutLeg)
+ if err := policy.Admit(flatRow, activator.LayoutFlat); !errors.Is(err, activator.ErrFlatExecRefused) {
+ t.Fatalf("Admit(flat) = %v", err)
+ }
+ if err := policy.Admit(foreignRow, activator.LayoutForeign); !errors.Is(err, activator.ErrFlatExecRefused) {
+ t.Fatalf("Admit(foreign) = %v", err)
+ }
+ close(release1)
+ leg.wait()
+ if got := runs.Load(); got != 2 {
+ t.Fatalf("passes = %d, want 2: the arming run and the one the refusal scheduled while it ran", got)
+ }
+ rt, err := reg.Get(flat)
+ if err != nil || rt == nil || rt.DBPath != provision.NestedDBPath(dd, flat) {
+ t.Fatalf("flat worker row after the nudged pass = %+v, %v; want it moved to %s", rt, err, provision.NestedDBPath(dd, flat))
+ }
+ if _, err := os.Lstat(provision.DefaultDBPath(dd, flat)); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("database still at the flat path: %v", err)
+ }
+ if l, _ := pass.Classify(*rt, []registry.Route{*rt}); l != activator.LayoutNested {
+ t.Fatalf("moved row classifies %s", l)
+ }
+ fr, _ := reg.Get(foreign)
+ if fr == nil || fr.DBPath != foreignRow.DBPath {
+ t.Fatalf("foreign row was touched by the pass: %+v", fr)
+ }
+
+ // A foreign row's refusal alone schedules nothing (the pass does not
+ // move foreign rows; the repair is carlos add).
+ before := runs.Load()
+ if err := policy.Admit(foreignRow, activator.LayoutForeign); !errors.Is(err, activator.ErrFlatExecRefused) {
+ t.Fatalf("Admit(foreign) = %v", err)
+ }
+ leg.wait()
+ if got := runs.Load(); got != before {
+ t.Fatalf("a foreign row's refusal ran the pass (%d -> %d)", before, got)
+ }
+ // Nor does a would-be refusal under report.
+ armedFlatExec.Store(nil)
+ reporting := armFlatExec(envOf(map[string]string{"CARLOS_FLAT_EXEC": "report"}), nudgeLayoutLeg)
+ if err := reporting.Admit(flatRow, activator.LayoutFlat); err != nil {
+ t.Fatalf("Admit under report = %v", err)
+ }
+ leg.wait()
+ if got := runs.Load(); got != before {
+ t.Fatalf("a reported spawn ran the pass (%d -> %d)", before, got)
+ }
+ // The refused flat row is the only worker the spawn record names as
+ // refused, with the closed word.
+ var refusedFlat bool
+ for _, r := range armedFlatExec.Load().refusals() {
+ if r.Host == flat && r.Reason == activator.FlatExecFlatReported {
+ refusedFlat = true
+ }
+ }
+ if !refusedFlat {
+ t.Fatalf("spawn record = %+v", armedFlatExec.Load().refusals())
+ }
+}