amadan

carlos / platform Public

agent: the layout pass at start and on every serial move

At start, after the instance sweep and before the activator starts, the
agent recovers every layout journal under the intent it reads once for
this box's status label, and only then publishes the pass. A failed
sweep still recovers, and the pass is told the sweep failed. An intent
the box cannot read (no store, a failed read, an unknown mode) holds a
completed revert fenced.

On a serial move a leg after the schedule leg starts a Converge in the
background: one pass at a time, and one more for any move that lands
mid-pass. Each run reads the intent again; none recorded or none
readable moves nothing. The pass defers awake hosts, holds the config
sync lock from journal write to journal removal, shares the exec
backend's isolation gate, and reloads the edge through a handle the
edge's ready hook publishes. SyncExec's one caller skips a host with a
journal, and the activator's journal check is the pass's own Lstat.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 24240be256213a30b32673dfe1be7ed1f396da7f parent cfb0d5b
3 files changed, +740 −16
  • internal/host/agent.go +57 −16
  • internal/host/agent_layout_test.go +420 −0
  • internal/host/layoutleg.go +263 −0

CI passed — run details

diff --git a/internal/host/agent.go b/internal/host/agent.go
index af51537a3..b25217229 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -149,7 +149,6 @@ import (
"flag"
"fmt"
"io"
- "io/fs"
"log"
"math/rand/v2"
"net"
@@ -180,6 +179,7 @@ import (
"github.com/carlosframework/platform/internal/fleetapi"
"github.com/carlosframework/platform/internal/instancectl"
"github.com/carlosframework/platform/internal/instancefs"
+ "github.com/carlosframework/platform/internal/layout"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
@@ -320,13 +320,14 @@ func RunAgent(args []string) {
// 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,
- }
+ // ErrNotExist counts as "no journal". It is the pass's own
+ // predicate, an Lstat, so a symlink planted at the journal path
+ // fences the host rather than being followed.
+ Journal: func(host string) bool { return layout.JournalPresent(dd, host) },
+ Peers: reg.List,
+ }
+ // The layout pass is handed this same gate below, so it never makes
+ // a nested tenant this backend would refuse to start.
eb.NestedOK = nestedGate(instancefs.MountNoSymfollow, eb.Control.Ready)
svc = activator.New(*cfg, &activator.SplitBackend{
Exec: eb,
@@ -364,12 +365,30 @@ func RunAgent(args []string) {
// channel that never answers logs and moves on, and the box is no
// worse off than before this existed (a leftover unit that keeps
// running is exactly what an agent restart left before this pass).
- if sw, err := eb.Control.Sweep(context.Background()); err != nil {
- log.Printf("agent: instance sweep: %v (nested wakes refuse until the channel answers)", err)
- } else if len(sw.Units) > 0 {
- log.Printf("agent: swept %d instance unit(s) left from a previous agent: %v", len(sw.Units), sw.Units)
- }
- svc.Start()
+ //
+ // Then layout recovery (design §8.2, "Recovery is the journal"):
+ // every journal a previous agent left is completed, and the host
+ // converged to the fleet's intent, before Start can wake anything.
+ // The pass is published only after that, so the serial leg never
+ // runs beside recovery.
+ startActivator(context.Background(),
+ func(ctx context.Context) error {
+ sw, err := eb.Control.Sweep(ctx)
+ if err != nil {
+ log.Printf("agent: instance sweep: %v (nested wakes refuse until the channel answers)", err)
+ } else if len(sw.Units) > 0 {
+ log.Printf("agent: swept %d instance unit(s) left from a previous agent: %v", len(sw.Units), sw.Units)
+ }
+ return err
+ },
+ func(ctx context.Context, swept bool) {
+ pass := newLayoutPass(reg, svc, dd, eb.NestedOK, swept)
+ intentStore, storeErr := openDeployStore(ctx)
+ intent := startupLayoutIntent(ctx, intentStore, storeErr, metricsBoxLabel(os.Getenv, readMachineID()))
+ recoverLayoutAtStart(ctx, pass, intent)
+ armedLayoutPass.Store(pass)
+ },
+ svc.Start)
act = svc
log.Printf("agent: hibernation activator armed (bucket %s, idle %s)", cfg.Bucket, cfg.Idle)
@@ -515,6 +534,10 @@ func RunAgent(args []string) {
// the loop always sees the CURRENT probe set: Reload replaces probers
// behind this handle without the heartbeat needing to be re-wired.
setUpstreamSource(e.UpstreamStates)
+ // The layout pass reloads routing through this as soon as a move
+ // commits, instead of waiting for the 2s registry poll.
+ reload := func() { _ = e.Reload() }
+ layoutReload.Store(&reload)
m := edge.NewUsageMeter()
e.EnableUsage(m)
// Same arming point as the meter: the edge observes an upstream 5xx
@@ -918,7 +941,8 @@ var configSyncMu sync.Mutex
func syncExecSerialized(ctx context.Context, reg *registry.Registry, store release.Store, priv *ecdh.PrivateKey, dd string, svc *activator.Service) error {
configSyncMu.Lock()
defer configSyncMu.Unlock()
- return appconfig.SyncExec(ctx, reg, store, priv, dd, bounce(svc), os.Getenv("PLATFORM_APPS_DOMAIN"), configAckFunc(ctx, store, "agent"))
+ return appconfig.SyncExec(ctx, reg, store, priv, dd, bounce(svc), os.Getenv("PLATFORM_APPS_DOMAIN"), configAckFunc(ctx, store, "agent"),
+ appconfig.SkipFenced(func(host string) bool { return layout.JournalPresent(dd, host) }))
}
// configAckFunc builds the callback config sync's two passes (the root
@@ -1370,6 +1394,7 @@ type fastDeps struct {
restartLeg func(ctx context.Context) // the `carlos restart` check (issue #113), or a test hook
wakeLeg func(ctx context.Context) // the sidecar wake check (design: sidecar members), or a test hook
scheduleLeg func(ctx context.Context) // the schedule converge (spec §6.2), or a test hook; nil = not armed
+ layoutLeg func(ctx context.Context) // the layout pass (isolated-exec design §8.2), or a test hook; nil = not armed
systemLeg *systemLeg // the platform self-update leg (spec §4); nil = not armed
lastSerial int64
seeded bool
@@ -1613,7 +1638,9 @@ func (d *fastDeps) observeRestart(restart func(string) error) func(string) error
// provisioning change and a wake request together, and it must run on
// a box with no CARLOS_SECRETS_KEY too.
// The schedule converge (schedule spec §6.2) is a further leg on that same
-// move, last, for both of the same reasons — see the call site.
+// move, for both of the same reasons; see the call site. The layout
+// leg (isolated-exec design §8.2) follows it and only starts a pass in
+// the background; the system leg's observe comes last.
// 2. Fast sweep: fastMembers, and only when non-empty one PassFilter over
// exactly those hosts — an empty sweep costs zero store reads. After
// the sweep, members still unstaged back off; staged members clear
@@ -1657,6 +1684,15 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
if d.scheduleLeg != nil {
d.scheduleLeg(ctx)
}
+ // The layout pass converges to the fleet's recorded layout on
+ // every move, after the pass set has minted and staged whatever
+ // the move carried, so a freshly minted row is seen. It only
+ // starts the pass: the pass runs on its own goroutine, because
+ // it can wait on stops, and nothing above or below may wait on
+ // it.
+ if d.layoutLeg != nil {
+ d.layoutLeg(ctx)
+ }
// The platform self-update leg rides the move too, last: a
// `carlos system update` is a console action like any other, and
// this is what makes a promote noticed in seconds instead of at
@@ -1837,6 +1873,11 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
// declarations. Nil when the box has no scheduled work armed, which
// fastTick tolerates.
d.scheduleLeg = scheduleLegFor(sd)
+ // The layout leg: this box's label is the one its status publishes
+ // under, which is the label an operator scopes a layout request to.
+ layoutBox := metricsBoxLabel(os.Getenv, readMachineID())
+ layoutPasses := &layoutLeg{run: func(ctx context.Context) { runLayoutLeg(ctx, store, layoutBox) }}
+ d.layoutLeg = layoutPasses.kick
// The platform self-update leg (spec §4): the sandboxed half, which only
// ever fetches into /data and writes a request. It lives here rather than
// in a loop of its own because the spread deadline needs a seconds
diff --git a/internal/host/agent_layout_test.go b/internal/host/agent_layout_test.go
new file mode 100644
index 000000000..8d9aab777
--- /dev/null
+++ b/internal/host/agent_layout_test.go
@@ -0,0 +1,420 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "reflect"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/layout"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// resetLayoutGlobals puts the package's published layout state back the
+// way a fresh process has it, before and after a test that touches it.
+func resetLayoutGlobals(t *testing.T) {
+ t.Helper()
+ clear := func() {
+ armedLayoutPass.Store(nil)
+ layoutReload.Store(nil)
+ layoutIntentSeen.Store(nil)
+ }
+ clear()
+ t.Cleanup(clear)
+}
+
+func appendIntent(t *testing.T, s release.Store, mode, box string) {
+ t.Helper()
+ if _, err := release.AppendLayoutIntent(context.Background(), s, release.LayoutIntent{Mode: mode, Scope: release.LayoutScope{Box: box}}); err != nil {
+ t.Fatalf("AppendLayoutIntent: %v", err)
+ }
+}
+
+func TestLayoutLegConvergesOnlyWhenTheIntentAppliesToThisBox(t *testing.T) {
+ resetLayoutGlobals(t)
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ var calls []layout.Mode
+ converge := func(_ context.Context, mode layout.Mode) []layout.Result {
+ calls = append(calls, mode)
+ return nil
+ }
+
+ layoutLegWith(ctx, s, "kass-1", converge)
+ if len(calls) != 0 {
+ t.Fatalf("converged with no intent: %v", calls)
+ }
+ if got := lastLayoutIntent(); got != "" {
+ t.Fatalf("intent seen with none recorded = %q", got)
+ }
+
+ appendIntent(t, s, release.LayoutModeNested, "kass-2")
+ layoutLegWith(ctx, s, "kass-1", converge)
+ if len(calls) != 0 {
+ t.Fatalf("converged for another box's intent: %v", calls)
+ }
+
+ appendIntent(t, s, release.LayoutModeFlat, "")
+ layoutLegWith(ctx, s, "kass-1", converge)
+ if !reflect.DeepEqual(calls, []layout.Mode{layout.ModeFlat}) {
+ t.Fatalf("calls = %v, want [flat]", calls)
+ }
+ if got := lastLayoutIntent(); got != release.LayoutModeFlat {
+ t.Fatalf("intent seen = %q, want flat", got)
+ }
+
+ appendIntent(t, s, release.LayoutModeNested, "kass-1")
+ layoutLegWith(ctx, s, "kass-1", converge)
+ if !reflect.DeepEqual(calls, []layout.Mode{layout.ModeFlat, layout.ModeNested}) {
+ t.Fatalf("calls = %v, want [flat nested]", calls)
+ }
+ if got := lastLayoutIntent(); got != release.LayoutModeNested {
+ t.Fatalf("intent seen = %q, want nested", got)
+ }
+}
+
+// An intent this binary cannot read, or a request whose mode it does not
+// know, moves nothing, and the last intent it did read stands.
+func TestLayoutLegDoesNothingWhenTheIntentIsUnreadable(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ body string
+ }{
+ {"newer schema", `{"schema":2,"requests":[{"id":"a","mode":"nested"}]}`},
+ {"unknown mode", `{"schema":1,"requests":[{"id":"a","mode":"sideways"}]}`},
+ {"not json", `{`},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ resetLayoutGlobals(t)
+ noteLayoutIntent(release.LayoutModeFlat)
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ if _, err := s.Put(ctx, release.LayoutIntentKey(), []byte(tc.body), ""); err != nil {
+ t.Fatal(err)
+ }
+ called := false
+ layoutLegWith(ctx, s, "kass-1", func(context.Context, layout.Mode) []layout.Result {
+ called = true
+ return nil
+ })
+ if called {
+ t.Fatal("converged under an intent it could not read")
+ }
+ if got := lastLayoutIntent(); got != release.LayoutModeFlat {
+ t.Fatalf("intent seen = %q, want the last one read (flat)", got)
+ }
+ })
+ }
+}
+
+// With no pass armed (no activator on this box) the leg reads nothing.
+func TestLayoutLegWithNoPassArmedReadsNothing(t *testing.T) {
+ resetLayoutGlobals(t)
+ runLayoutLeg(context.Background(), refusingStore{t: t}, "kass-1")
+}
+
+// refusingStore fails the test on a read; every other method is the nil
+// embedded Store's, which panics, which fails it too.
+type refusingStore struct {
+ release.Store
+ t *testing.T
+}
+
+func (r refusingStore) Get(context.Context, string) ([]byte, string, error) {
+ r.t.Fatal("the store was read")
+ return nil, "", nil
+}
+
+// The leg never runs two passes at once, never makes the serial tick wait
+// for one, and a move that lands mid-pass costs exactly one more pass
+// however many moves there were.
+func TestLayoutLegRunsOneAtATimeAndOnceMoreForAMoveMidPass(t *testing.T) {
+ started := make(chan struct{}, 8)
+ release1 := make(chan struct{})
+ var runs, inFlight, most atomic.Int32
+ leg := &layoutLeg{run: func(context.Context) {
+ n := inFlight.Add(1)
+ for {
+ m := most.Load()
+ if n <= m || most.CompareAndSwap(m, n) {
+ break
+ }
+ }
+ if runs.Add(1) == 1 {
+ started <- struct{}{}
+ <-release1
+ }
+ inFlight.Add(-1)
+ }}
+ ctx := context.Background()
+
+ returned := make(chan struct{})
+ go func() {
+ leg.kick(ctx)
+ close(returned)
+ }()
+ select {
+ case <-returned:
+ case <-time.After(5 * time.Second):
+ t.Fatal("kick waited for the pass")
+ }
+ select {
+ case <-started:
+ case <-time.After(5 * time.Second):
+ t.Fatal("the first pass never started")
+ }
+ for i := 0; i < 3; i++ {
+ leg.kick(ctx)
+ }
+ close(release1)
+ leg.wait()
+ if got := runs.Load(); got != 2 {
+ t.Fatalf("passes = %d, want 2 (the first, and one for the moves mid-pass)", got)
+ }
+ if got := most.Load(); got != 1 {
+ t.Fatalf("%d passes ran at once", got)
+ }
+
+ // Idle again: the next move runs one pass.
+ leg.kick(ctx)
+ leg.wait()
+ if got := runs.Load(); got != 3 {
+ t.Fatalf("passes after an idle kick = %d, want 3", got)
+ }
+}
+
+// A kick that lands between the runner's last look at the dirty flag and
+// its going idle is taken back up by that runner, not lost.
+func TestLayoutLegTakesUpAKickThatLandsAsItGoesIdle(t *testing.T) {
+ leg := &layoutLeg{run: func(context.Context) { t.Fatal("kick started a second runner") }}
+ leg.running.Store(true) // a runner past its last look at dirty
+ leg.kick(context.Background())
+ if !leg.goIdle() {
+ t.Fatal("the runner went idle over a pending kick")
+ }
+ if !leg.running.Load() {
+ t.Fatal("the runner going again is not marked running")
+ }
+ leg.dirty.Store(false)
+ if leg.goIdle() {
+ t.Fatal("the runner went again with nothing pending")
+ }
+ if leg.running.Load() {
+ t.Fatal("an idle leg is marked running")
+ }
+}
+
+// Kicks from several goroutines at once, many times over: every one is
+// answered and the leg ends idle with nothing pending (run under -race).
+func TestLayoutLegNeverLosesAKick(t *testing.T) {
+ for i := 0; i < 200; i++ {
+ var runs atomic.Int32
+ leg := &layoutLeg{run: func(context.Context) { runs.Add(1) }}
+ var wg sync.WaitGroup
+ for k := 0; k < 4; k++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ leg.kick(context.Background())
+ }()
+ }
+ wg.Wait()
+ leg.wait()
+ if runs.Load() == 0 {
+ t.Fatal("four kicks ran no pass")
+ }
+ if leg.dirty.Load() {
+ t.Fatalf("iteration %d: a kick was left pending with the leg idle", i)
+ }
+ }
+}
+
+// Recovery runs after the startup unit sweep and before the activator
+// starts, and it runs even when the sweep failed, told so.
+func TestStartActivatorRecoversAfterTheSweepAndBeforeStart(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ sweepErr error
+ swept bool
+ }{
+ {"sweep ok", nil, true},
+ {"sweep failed", errors.New("channel down"), false},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ var events []string
+ var gotSwept bool
+ startActivator(context.Background(),
+ func(context.Context) error { events = append(events, "sweep"); return tc.sweepErr },
+ func(_ context.Context, swept bool) { events = append(events, "recover"); gotSwept = swept },
+ func() { events = append(events, "start") })
+ if want := []string{"sweep", "recover", "start"}; !reflect.DeepEqual(events, want) {
+ t.Fatalf("order = %v, want %v", events, want)
+ }
+ if gotSwept != tc.swept {
+ t.Fatalf("swept = %v, want %v", gotSwept, tc.swept)
+ }
+ })
+ }
+}
+
+// The intent recovery runs under at start: "" when none applies, the
+// recorded mode when one does, and IntentUnreadable whenever the box
+// cannot be sure, which holds a completed revert fenced.
+func TestStartupLayoutIntent(t *testing.T) {
+ ctx := context.Background()
+ resetLayoutGlobals(t)
+ if got := startupLayoutIntent(ctx, nil, errors.New("no bucket"), "kass-1"); got != layout.IntentUnreadable {
+ t.Fatalf("no store: %q", got)
+ }
+ s := release.NewDirStore(t.TempDir())
+ if got := startupLayoutIntent(ctx, s, nil, "kass-1"); got != "" {
+ t.Fatalf("no object: %q", got)
+ }
+ appendIntent(t, s, release.LayoutModeNested, "kass-2")
+ if got := startupLayoutIntent(ctx, s, nil, "kass-1"); got != "" {
+ t.Fatalf("another box's intent: %q", got)
+ }
+ appendIntent(t, s, release.LayoutModeNested, "")
+ if got := startupLayoutIntent(ctx, s, nil, "kass-1"); got != layout.ModeNested {
+ t.Fatalf("fleet intent: %q", got)
+ }
+ if got := lastLayoutIntent(); got != release.LayoutModeNested {
+ t.Fatalf("intent seen = %q", got)
+ }
+ for _, body := range []string{
+ `{"schema":2,"requests":[]}`,
+ `{"schema":1,"requests":[{"id":"a","mode":"sideways"}]}`,
+ } {
+ bad := release.NewDirStore(t.TempDir())
+ if _, err := bad.Put(ctx, release.LayoutIntentKey(), []byte(body), ""); err != nil {
+ t.Fatal(err)
+ }
+ if got := startupLayoutIntent(ctx, bad, nil, "kass-1"); got != layout.IntentUnreadable {
+ t.Fatalf("%s: %q", body, got)
+ }
+ }
+}
+
+// The layout leg rides the serial move after the schedule leg, and like
+// every other leg it does not run on the boot seed or an unchanged serial.
+func TestFastTickRunsTheLayoutLegAfterTheScheduleLeg(t *testing.T) {
+ env := newFastFixture(t)
+ var events []string
+ d := env.deps(func() { events = append(events, "pass") })
+ d.scheduleLeg = func(context.Context) { events = append(events, "schedule") }
+ d.layoutLeg = func(context.Context) { events = append(events, "layout") }
+ now := time.Unix(1754640000, 0)
+ ctx := context.Background()
+ if err := release.BumpSerial(ctx, env.store, now); err != nil {
+ t.Fatal(err)
+ }
+ fastTick(ctx, d, now)
+ fastTick(ctx, d, now.Add(10*time.Second))
+ if len(events) != 0 {
+ t.Fatalf("seed or unchanged serial ran %v", events)
+ }
+ if err := release.BumpSerial(ctx, env.store, now); err != nil {
+ t.Fatal(err)
+ }
+ fastTick(ctx, d, now.Add(20*time.Second))
+ if want := []string{"pass", "schedule", "layout"}; !reflect.DeepEqual(events, want) {
+ t.Fatalf("legs = %v, want %v", events, want)
+ }
+}
+
+// The pass the agent builds: moves serialize with config sync, the probe
+// gate is the one handed in, routing reloads through the published edge
+// (a no-op before the edge is up), awake hosts are deferred, and with no
+// activator the fence is a nil interface rather than a nil pointer in one.
+func TestNewLayoutPassWiring(t *testing.T) {
+ resetLayoutGlobals(t)
+ reg := restartTestRegistry(t)
+ dd := t.TempDir()
+ var gateCalls int
+ gate := func() error { gateCalls++; return nil }
+ p := newLayoutPass(reg, nil, dd, gate, true)
+
+ if p.Fence != nil {
+ t.Fatalf("Fence = %#v with no activator, want a nil interface", p.Fence)
+ }
+ if !p.SweptAtStart || !p.DeferAwake || p.Registry != reg || p.DataDir != dd || p.Logf == nil {
+ t.Fatalf("pass = %+v", p)
+ }
+ if err := p.NestedOK(); err != nil || gateCalls != 1 {
+ t.Fatalf("NestedOK is not the gate handed in: err %v, calls %d", err, gateCalls)
+ }
+
+ held := false
+ p.Serialize(func() { held = !configSyncMu.TryLock() })
+ if !held {
+ t.Fatal("Serialize ran without the config-sync lock")
+ }
+ if !configSyncMu.TryLock() {
+ t.Fatal("Serialize left the config-sync lock held")
+ }
+ configSyncMu.Unlock()
+
+ p.Reload() // nothing published yet: must not panic
+ reloads := 0
+ f := func() { reloads++ }
+ layoutReload.Store(&f)
+ p.Reload()
+ if reloads != 1 {
+ t.Fatalf("reloads = %d, want 1", reloads)
+ }
+
+ if q := newLayoutPass(reg, nil, dd, gate, false); q.SweptAtStart {
+ t.Fatal("SweptAtStart set after a failed sweep")
+ }
+}
+
+// The one production SyncExec caller passes the journal predicate: a host
+// with a layout journal gets no bundle, and its sibling does.
+func TestSyncExecSerializedSkipsAHostWithALayoutJournal(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ reg := restartTestRegistry(t)
+ dd := t.TempDir()
+ privText, _, err := appconfig.GenerateKey()
+ if err != nil {
+ t.Fatal(err)
+ }
+ priv, err := appconfig.ParsePrivateKey(privText, "test")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := appconfig.SetEnv(ctx, s, "bab", "eleven", map[string]string{"LOG_LEVEL": "debug"}); err != nil {
+ t.Fatal(err)
+ }
+ held, sibling := "held.example.com", "sib.example.com"
+ for _, h := range []string{held, sibling} {
+ if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", App: "eleven", Account: "bab",
+ Status: "ready", Socket: "/run/carlos/" + h + ".sock"}); err != nil {
+ t.Fatal(err)
+ }
+ }
+ journal := provision.LayoutJournalPath(dd, held)
+ if err := os.MkdirAll(filepath.Dir(journal), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(journal, []byte("{}"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if err := syncExecSerialized(ctx, reg, s, priv, dd, nil); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := os.Lstat(filepath.Join(dd, held+".config.env")); !os.IsNotExist(err) {
+ t.Fatalf("a held host got a bundle (err %v)", err)
+ }
+ if b, err := os.ReadFile(filepath.Join(dd, sibling+".config.env")); err != nil || string(b) != "LOG_LEVEL=debug\n" {
+ t.Fatalf("sibling bundle = %q, %v", b, err)
+ }
+}
diff --git a/internal/host/layoutleg.go b/internal/host/layoutleg.go
new file mode 100644
index 000000000..cd018655c
--- /dev/null
+++ b/internal/host/layoutleg.go
@@ -0,0 +1,263 @@
+package host
+
+// The layout pass on the box (isolated-exec design §8.2): recovery at agent
+// start, before the activator can wake anything, and convergence to the
+// fleet's recorded layout intent on every serial move after that. The pass
+// itself lives in internal/layout; this file is how the agent builds it,
+// publishes it to the loops that run on other goroutines, and keeps it off
+// the serial loop's own goroutine.
+
+import (
+ "context"
+ "log"
+ "os/user"
+ "strconv"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/layout"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// armedLayoutPass is the box's one layout pass. RunAgent stores it once,
+// after startup recovery and before the activator starts, and the serial
+// leg (fastConvergeLoop's goroutine) and the status snapshot read it. Nil
+// means no activator is armed, so there is nothing for a pass to move.
+var armedLayoutPass atomic.Pointer[layout.Pass]
+
+// layoutReload is the live edge's Reload, published by the edge's ready
+// hook, which runs after the pass is built. Until then the pass's reload is
+// a no-op and the edge's own 2s registry poll picks up a flipped row.
+var layoutReload atomic.Pointer[func()]
+
+// layoutIntentSeen is the last layout intent mode the agent read for this
+// box: "" when none is recorded or none addresses it. A read that fails
+// leaves it as it was, because a failed read says nothing about the intent.
+var layoutIntentSeen atomic.Pointer[string]
+
+// lastLayoutIntent is the mode layoutIntentSeen holds, "" before any read.
+func lastLayoutIntent() string {
+ if m := layoutIntentSeen.Load(); m != nil {
+ return *m
+ }
+ return ""
+}
+
+func noteLayoutIntent(mode string) { layoutIntentSeen.Store(&mode) }
+
+// reloadRouting asks the live edge to rebuild its routes now, if there is
+// one yet.
+func reloadRouting() {
+ if f := layoutReload.Load(); f != nil {
+ (*f)()
+ }
+}
+
+// serializeWithConfigSync runs f under the lock every SyncExec pass holds,
+// so config sync never renders a host's bundle into a directory the pass
+// is moving the host out of. The pass holds nothing but its own fence when
+// it asks, and SyncExec under this lock never waits on a fence or on the
+// pass (its bounce only starts a stop), so the two cannot wait on each
+// other.
+func serializeWithConfigSync(f func()) {
+ configSyncMu.Lock()
+ defer configSyncMu.Unlock()
+ f()
+}
+
+// socketsGID is the carlos-sockets group's id, which a nested host's run
+// directory is given so the edge can reach its socket, or -1 to leave the
+// group alone where the group does not exist (a dev machine).
+func socketsGID() int {
+ g, err := user.LookupGroup(socketsGroup)
+ if err != nil {
+ return -1
+ }
+ n, err := strconv.Atoi(g.Gid)
+ if err != nil {
+ return -1
+ }
+ return n
+}
+
+// newLayoutPass builds the pass from what only RunAgent has in hand.
+// nestedOK must be the same gate the exec backend refuses nested wakes
+// with, so the pass never makes a nested tenant the box would refuse to
+// start. swept is whether the startup unit sweep succeeded; without it the
+// pass never trusts an unproven stop.
+//
+// DeferAwake is set because the only Converge caller is the serial leg,
+// which runs while tenants serve; startup recovery never reaches the code
+// that reads it.
+func newLayoutPass(reg *registry.Registry, svc *activator.Service, dd string, nestedOK func() error, swept bool) *layout.Pass {
+ p := &layout.Pass{
+ Registry: reg,
+ DataDir: dd,
+ SocketsGID: socketsGID(),
+ Reload: reloadRouting,
+ Logf: log.Printf,
+ SweptAtStart: swept,
+ NestedOK: nestedOK,
+ Serialize: serializeWithConfigSync,
+ DeferAwake: true,
+ }
+ // Assigned only when non-nil: a nil *Service inside the interface would
+ // not be a nil Fence, and the pass would call through it.
+ if svc != nil {
+ p.Fence = svc
+ }
+ return p
+}
+
+// startActivator is the order the activator comes up in. The startup unit
+// sweep first, so nothing an earlier agent left is running. Then layout
+// recovery, which runs even when the sweep failed (completing a journal
+// is exact) and is told whether it succeeded. Then Start, so no host is
+// woken while a journal says its files are between two directories.
+func startActivator(ctx context.Context, sweep func(context.Context) error, recoverLayout func(ctx context.Context, swept bool), start func()) {
+ err := sweep(ctx)
+ recoverLayout(ctx, err == nil)
+ start()
+}
+
+// layoutIntentTimeout bounds the one intent read at startup, which the
+// activator's start waits on.
+const layoutIntentTimeout = 30 * time.Second
+
+// startupLayoutIntent is the intent startup recovery runs under. It is ""
+// when no intent is recorded or none addresses this box. It is
+// IntentUnreadable when the box cannot be sure: no deploy store, a failed
+// read, or a request whose mode this binary does not know. Under that
+// intent recovery holds a completed revert fenced rather than release a
+// host flat that the fleet may want nested.
+func startupLayoutIntent(ctx context.Context, store release.Store, storeErr error, boxLabel string) layout.Mode {
+ if storeErr != nil {
+ log.Printf("layout: intent -> unreadable: deploy store: %v", storeErr)
+ return layout.IntentUnreadable
+ }
+ ctx, cancel := context.WithTimeout(ctx, layoutIntentTimeout)
+ defer cancel()
+ in, found, err := release.LoadLayoutIntent(ctx, store, boxLabel)
+ switch {
+ case err != nil:
+ log.Printf("layout: intent -> unreadable: %v", err)
+ return layout.IntentUnreadable
+ case !found:
+ noteLayoutIntent(release.LayoutModeNone)
+ return ""
+ case !release.ValidLayoutMode(in.Mode):
+ log.Printf("layout: intent %s -> unreadable: mode %q is not flat or nested", in.ID, in.Mode)
+ return layout.IntentUnreadable
+ }
+ noteLayoutIntent(in.Mode)
+ return layout.Mode(in.Mode)
+}
+
+// recoverLayoutAtStart runs the pass's recovery under the startup intent
+// and logs what it did.
+func recoverLayoutAtStart(ctx context.Context, pass *layout.Pass, intent layout.Mode) {
+ logLayoutResults(pass.Recover(ctx, intent))
+}
+
+// layoutLeg runs the layout pass off the serial loop. A pass can wait up to
+// a minute per host for a stop, and the restart, wake, schedule and system
+// legs and the fast sweep must not wait behind it. One pass runs at a time;
+// a serial move that lands while one is running marks the leg dirty, and
+// the leg runs exactly once more after it, however many moves landed.
+type layoutLeg struct {
+ run func(context.Context)
+ running atomic.Bool
+ dirty atomic.Bool
+ wg sync.WaitGroup
+}
+
+// kick asks for a pass and returns at once.
+func (l *layoutLeg) kick(ctx context.Context) {
+ l.dirty.Store(true)
+ if !l.running.CompareAndSwap(false, true) {
+ return
+ }
+ l.wg.Add(1)
+ go func() {
+ defer l.wg.Done()
+ for {
+ for l.dirty.Swap(false) {
+ l.run(ctx)
+ }
+ if !l.goIdle() {
+ return
+ }
+ }
+ }()
+}
+
+// goIdle marks the runner idle and reports whether it must run again. A
+// kick that landed after the runner's last look at dirty saw the leg
+// running and left its request there; the runner takes it back up, unless
+// that kick has already started a runner of its own.
+func (l *layoutLeg) goIdle() bool {
+ l.running.Store(false)
+ return l.dirty.Load() && l.running.CompareAndSwap(false, true)
+}
+
+// wait blocks until no pass is running or pending.
+func (l *layoutLeg) wait() { l.wg.Wait() }
+
+// runLayoutLeg is one run of the serial leg: the published pass, converged
+// to this box's intent. With no pass armed there is nothing to move, and
+// the intent is not read.
+func runLayoutLeg(ctx context.Context, store release.Store, boxLabel string) {
+ pass := armedLayoutPass.Load()
+ if pass == nil {
+ return
+ }
+ layoutLegWith(ctx, store, boxLabel, pass.Converge)
+}
+
+// layoutLegWith is the leg's decision: read the intent, and converge only
+// when a request addresses this box with a mode this binary knows. An
+// intent that cannot be read moves nothing; the next serial move reads it
+// again.
+func layoutLegWith(ctx context.Context, store release.Store, boxLabel string, converge func(context.Context, layout.Mode) []layout.Result) {
+ in, found, err := release.LoadLayoutIntent(ctx, store, boxLabel)
+ switch {
+ case err != nil:
+ log.Printf("layout: intent -> unreadable: %v; not converging this move", err)
+ return
+ case !found:
+ noteLayoutIntent(release.LayoutModeNone)
+ return
+ case !release.ValidLayoutMode(in.Mode):
+ log.Printf("layout: intent %s -> unreadable: mode %q is not flat or nested; not converging this move", in.ID, in.Mode)
+ return
+ }
+ noteLayoutIntent(in.Mode)
+ logLayoutResults(converge(ctx, layout.Mode(in.Mode)))
+}
+
+// logLayoutResults logs every result that changed or withheld something.
+// Hosts deferred because they are awake are one line together: under a
+// standing intent every serial move meets them until they sleep.
+func logLayoutResults(rs []layout.Result) {
+ var awake []string
+ for _, r := range rs {
+ switch {
+ case r.Outcome == "unchanged":
+ case r.Outcome == "deferred" && r.Reason == "awake":
+ awake = append(awake, r.Host)
+ case r.Host == "":
+ log.Printf("layout: %s -> %s", r.Outcome, r.Reason)
+ case r.Reason == "":
+ log.Printf("layout: %s: %s", r.Host, r.Outcome)
+ default:
+ log.Printf("layout: %s: %s -> %s", r.Host, r.Outcome, r.Reason)
+ }
+ }
+ if len(awake) > 0 {
+ log.Printf("layout: deferred until asleep -> %s", strings.Join(awake, ", "))
+ }
+}