amadan

carlos / platform Public

isolated exec: consumers follow the layout; the agent wires the control client, the gate and the startup sweep (plan 3, task 6)

Config delivery (syncexec.go): ConfigEnvPath resolves a nested row's
rendered bundle inside its own instance directory (isolated-exec
design §7) instead of the flat <dataDir>/<host>.config.env; the
writer uses instancefs.WriteAtomic for a nested path (tenant-owned
ground, §3.2) and the plain temp+rename it always used for a flat
one. reapConfigEnv now also walks <dataDir>/instances one level down
to reap stale nested files.

Schedule token (runner.go, schedule_wiring.go): TokenFromEnvPaths
takes the env file's location as a func, and TokenFromEnvFile is now
that with the flat resolver baked in. scheduleToken gains a registry
so envPathFor can classify a host through activator.Classify and
point the runner's Token and the EnsureAdminToken prepare step at the
nested layout's own <host>.env when the row is nested.

Moves pass (movespass.go): moveCfg gains Route, wired from reg.Get; a
cutting move now fences on provision.LayoutJournalPath (the same
fence Classify applies to a wake) and checkpoints the row's own
DBPath rather than guessing the legacy flat path.

Litestream glob (litestream.go): FleetDBsWithSkipped folds in a
second glob under <dataDir>/instances/*/*.db so a nested row's
database is discovered for replication.

Agent (agent.go): wires eb.Control (instancectl.NewClient over
controlsock.Path), eb.Layout (journal check + reg.List as Peers),
eb.NestedOK (nestedGate: the three nosymfollow mounts then /ready),
and svc.Routes = reg; runs the startup instance sweep before
svc.Start().

Provision (provision.go): dropped its internal/appconfig import
(appconfig now imports provision for the nested layout paths, which
would otherwise cycle) by duplicating ValidEnvironmentName's small
regex check locally as validEnvironmentName — same shape, same
behaviour, no shared import.

Tests: TestSyncExecWritesANestedRouteIntoItsInstanceDir,
TestReapConfigEnvWalksInstanceDirs, TestFleetDBsDiscoversNestedDatabases,
TestMovesPassCutFencedByLayoutJournal, TestMovesPassCutUsesRouteDBPath,
TestNestedGateRefusesOnAMissingMount, TestNestedGateRefusesOnAMountCheckError,
TestNestedGatePassesMountsThenChecksReady.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 66b33fce2fb2daebbf6bd9a0a55b0734ddc4fa7f parent d278783
13 files changed, +518 −26
  • internal/appconfig/syncexec.go +97 −11
  • internal/appconfig/syncexec_test.go +91 −0
  • internal/host/agent.go +55 −0
  • internal/host/agent_isolated_test.go +61 −0
  • internal/host/movespass.go +26 −0
  • internal/host/movespass_test.go +70 −0
  • internal/host/schedule_wake_test.go +2 −2
  • internal/host/schedule_wiring.go +31 −4
  • internal/host/schedule_wiring_test.go +3 −3
  • internal/ops/litestream.go +11 −0
  • internal/ops/ops_test.go +36 −0
  • internal/provision/provision.go +20 −2
  • internal/schedule/runner.go +15 −4

CI cancelled — superseded by a newer push — run details

diff --git a/internal/appconfig/syncexec.go b/internal/appconfig/syncexec.go
index 0b9ae9d48..4d38d8bfc 100644
--- a/internal/appconfig/syncexec.go
+++ b/internal/appconfig/syncexec.go
@@ -44,6 +44,8 @@ import (
"strings"
"time"
+ "github.com/carlosframework/platform/internal/instancefs"
+ "github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
@@ -52,6 +54,26 @@ import (
// with this suffix belongs to sync; nothing else in dataDir does.
const configEnvSuffix = ".config.env"
+// nestedRoute reports whether rt's DB and socket are exactly the nested
+// layout's defaults (isolated-exec design §8.1) — the same two-column test
+// activator.Classify makes, spelled out here rather than imported so this
+// package need not take a dependency on internal/activator for one
+// comparison.
+func nestedRoute(dataDir string, rt registry.Route) bool {
+ return rt.DBPath == provision.NestedDBPath(dataDir, rt.Host) && rt.Socket == provision.NestedSocket(rt.Host)
+}
+
+// ConfigEnvPath is where a route's rendered bundle lands: inside the
+// instance directory for a nested row (isolated-exec design §7), flat for
+// everything else, including opted-in addr routes whose supervisor reads
+// the flat path.
+func ConfigEnvPath(dataDir string, rt registry.Route) string {
+ if nestedRoute(dataDir, rt) {
+ return filepath.Join(provision.InstanceDir(dataDir, rt.Host), rt.Host+configEnvSuffix)
+ }
+ return filepath.Join(dataDir, rt.Host+configEnvSuffix)
+}
+
// SyncExec resolves config once per distinct (Account, App, Environment)
// among the box's exec-syncable routes and converges every such route's
// <dataDir>/<host>.config.env, calling bounce(host) whenever a file's
@@ -462,7 +484,7 @@ func SyncExec(ctx context.Context, reg *registry.Registry, s release.Store, priv
if c.v || !c.known {
continue
}
- path := filepath.Join(dataDir, rt.Host+configEnvSuffix)
+ path := ConfigEnvPath(dataDir, rt)
if _, err := os.Stat(path); err != nil {
continue // nothing was ever staged for this route: nothing to warn about
}
@@ -869,7 +891,7 @@ func syncExecHost(rt registry.Route, m *Manifest, key string, priv *ecdh.Private
fmt.Printf(" %q: not a bare hostname, refusing to derive a config file path from it\n", rt.Host)
return
}
- path := filepath.Join(dataDir, rt.Host+configEnvSuffix)
+ path := ConfigEnvPath(dataDir, rt)
if m == nil {
if err := os.Remove(path); err != nil {
if !os.IsNotExist(err) {
@@ -918,15 +940,29 @@ func syncExecHost(rt registry.Route, m *Manifest, key string, priv *ecdh.Private
if existed && bytes.Equal(cur, rendered) {
return // up to date — stay quiet on the periodic tick
}
- tmp := path + ".tmp"
- if err := os.WriteFile(tmp, rendered, 0o600); err != nil {
- fmt.Printf(" %s: write %s: %v\n", rt.Host, path, err)
- return
- }
- if err := os.Rename(tmp, path); err != nil {
- os.Remove(tmp) // don't leave rendered secrets behind
- fmt.Printf(" %s: rename %s: %v\n", rt.Host, path, err)
- return
+ // A nested row's config file lives inside its own instance directory —
+ // tenant-owned ground (isolated-exec design §3.2) — so it is written
+ // through instancefs.WriteAtomic's O_EXCL temp name and openat2
+ // resolution rather than a predictable path.tmp: the same reasoning
+ // backend_exec.go's own writes into that directory follow. The flat
+ // directory is agent-owned throughout, so it keeps the plain
+ // temp-then-rename it always used.
+ if nestedRoute(dataDir, rt) {
+ if err := instancefs.WriteAtomic(filepath.Dir(path), filepath.Base(path), rendered, 0o600); err != nil {
+ fmt.Printf(" %s: write %s: %v\n", rt.Host, path, err)
+ return
+ }
+ } else {
+ tmp := path + ".tmp"
+ if err := os.WriteFile(tmp, rendered, 0o600); err != nil {
+ fmt.Printf(" %s: write %s: %v\n", rt.Host, path, err)
+ return
+ }
+ if err := os.Rename(tmp, path); err != nil {
+ os.Remove(tmp) // don't leave rendered secrets behind
+ fmt.Printf(" %s: rename %s: %v\n", rt.Host, path, err)
+ return
+ }
}
fmt.Printf(" %s: config updated\n", rt.Host)
bounceErr := bounceLogged(rt.Host, bounce)
@@ -1021,4 +1057,54 @@ func reapConfigEnv(routes []registry.Route, dataDir string) {
}
fmt.Printf(" %s: no route on this box, removed its config file\n", host)
}
+ reapNestedConfigEnv(live, dataDir)
+}
+
+// reapNestedConfigEnv is reapConfigEnv's nested half (isolated-exec design
+// §7): a nested row's rendered bundle lives inside its own instance
+// directory, one level under <dataDir>/instances, so a stale file there is
+// found by walking that tree rather than dataDir's own listing. The flat
+// directory's <host>.config / <host>.config.env name collision (reapConfigEnv's
+// own ambiguity guard, above) cannot happen inside an instance directory —
+// nothing in there is named after some OTHER host — so no such guard is
+// needed here.
+func reapNestedConfigEnv(live map[string]bool, dataDir string) {
+ root := filepath.Join(dataDir, "instances")
+ dirs, err := os.ReadDir(root)
+ if err != nil {
+ return // no nested layout on this box (or nothing has migrated yet)
+ }
+ for _, d := range dirs {
+ if !d.IsDir() {
+ continue
+ }
+ instDir := filepath.Join(root, d.Name())
+ ents, err := os.ReadDir(instDir)
+ if err != nil {
+ continue
+ }
+ for _, e := range ents {
+ if e.IsDir() {
+ continue
+ }
+ if strings.HasSuffix(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)
+ }
+ continue
+ }
+ if !strings.HasSuffix(e.Name(), configEnvSuffix) {
+ continue
+ }
+ host := strings.TrimSuffix(e.Name(), configEnvSuffix)
+ if live[host] {
+ continue
+ }
+ if err := os.Remove(filepath.Join(instDir, e.Name())); err != nil {
+ fmt.Printf(" %s: remove orphaned config file: %v\n", host, err)
+ continue
+ }
+ fmt.Printf(" %s: no route on this box, removed its config file\n", host)
+ }
+ }
}
diff --git a/internal/appconfig/syncexec_test.go b/internal/appconfig/syncexec_test.go
index b580080f9..ee6c48fbb 100644
--- a/internal/appconfig/syncexec_test.go
+++ b/internal/appconfig/syncexec_test.go
@@ -12,6 +12,7 @@ import (
"testing"
"time"
+ "github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/relay"
"github.com/carlosframework/platform/internal/release"
@@ -121,6 +122,96 @@ func TestSyncExecWritesConfigFileAndBouncesOnChange(t *testing.T) {
}
}
+// TestSyncExecWritesANestedRouteIntoItsInstanceDir pins the isolated-exec
+// half of config delivery (design §7): a route whose DB and socket are the
+// nested layout's defaults gets its rendered bundle inside its OWN instance
+// directory, never at the flat <dataDir>/<host>.config.env path a foreign or
+// flat row would use.
+func TestSyncExecWritesANestedRouteIntoItsInstanceDir(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)
+ }
+ host := "inst.example.com"
+ // The nested layout's directory pre-exists here exactly as it would on
+ // a real box: the migration pass that flips a row's DBPath/Socket to
+ // the nested defaults is what creates it, before any route can
+ // classify as nested at all.
+ if err := os.MkdirAll(provision.InstanceDir(dd, host), 0o755); err != nil {
+ t.Fatalf("mkdir instance dir: %v", err)
+ }
+ if err := reg.Upsert(registry.Route{
+ Host: host, Kind: "instance", App: "eleven", Account: "bab",
+ Status: "ready", DBPath: provision.NestedDBPath(dd, host), Socket: provision.NestedSocket(host),
+ }); err != nil {
+ t.Fatalf("Upsert: %v", err)
+ }
+ var bounced []string
+ bounce := func(h string) error { bounced = append(bounced, h); return nil }
+ if err := SyncExec(ctx, reg, s, pk, dd, bounce, "", nil); err != nil {
+ t.Fatalf("SyncExec: %v", err)
+ }
+ nestedPath := filepath.Join(provision.InstanceDir(dd, host), host+".config.env")
+ got, err := os.ReadFile(nestedPath)
+ if err != nil {
+ t.Fatalf("read nested config env: %v", err)
+ }
+ if string(got) != "LOG_LEVEL=debug\n" {
+ t.Fatalf("config env = %q, want LOG_LEVEL=debug", got)
+ }
+ if fi, _ := os.Stat(nestedPath); fi.Mode().Perm() != 0o600 {
+ t.Fatalf("nested config env mode = %v, want 0600", fi.Mode().Perm())
+ }
+ if _, err := os.Stat(filepath.Join(dd, host+".config.env")); !os.IsNotExist(err) {
+ t.Fatalf("a nested route's config also landed at the flat path, stat err = %v", err)
+ }
+ if len(bounced) != 1 || bounced[0] != host {
+ t.Fatalf("bounced = %v, want exactly [%s]", bounced, host)
+ }
+}
+
+// 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.
+func TestReapConfigEnvWalksInstanceDirs(t *testing.T) {
+ dd := t.TempDir()
+ reg := testRegistry(t)
+ live, gone := "live.example.com", "gone.example.com"
+ for _, host := range []string{live, gone} {
+ if err := os.MkdirAll(provision.InstanceDir(dd, host), 0o755); err != nil {
+ t.Fatalf("mkdir instance dir for %s: %v", host, err)
+ }
+ p := filepath.Join(provision.InstanceDir(dd, host), host+configEnvSuffix)
+ if err := os.WriteFile(p, []byte("A=1\n"), 0o600); err != nil {
+ t.Fatalf("stage %s: %v", host, err)
+ }
+ }
+ if err := reg.Upsert(registry.Route{
+ Host: live, Kind: "instance", App: "eleven", Account: "bab",
+ Status: "ready", DBPath: provision.NestedDBPath(dd, live), Socket: provision.NestedSocket(live),
+ }); err != nil {
+ t.Fatalf("Upsert: %v", err)
+ }
+ routes, err := reg.List()
+ if err != nil {
+ t.Fatalf("List: %v", err)
+ }
+ reapConfigEnv(routes, dd)
+
+ livePath := filepath.Join(provision.InstanceDir(dd, live), live+configEnvSuffix)
+ if _, err := os.Stat(livePath); err != nil {
+ t.Fatalf("live nested route's config was reaped: %v", err)
+ }
+ gonePath := filepath.Join(provision.InstanceDir(dd, gone), gone+configEnvSuffix)
+ if _, err := os.Stat(gonePath); !os.IsNotExist(err) {
+ t.Fatalf("stale nested config survived reap, stat err = %v", err)
+ }
+}
+
func TestSyncExecRemovesFileWhenManifestDeleted(t *testing.T) {
ctx, s, reg, dd, priv, _ := execKit(t)
pk, _ := ParsePrivateKey(priv, "test")
diff --git a/internal/host/agent.go b/internal/host/agent.go
index f7ecf08d5..1dbc325f7 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -171,10 +171,13 @@ import (
"github.com/carlosframework/platform/internal/applog"
"github.com/carlosframework/platform/internal/caps"
"github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/controlsock"
"github.com/carlosframework/platform/internal/domains"
"github.com/carlosframework/platform/internal/edge"
"github.com/carlosframework/platform/internal/edgesfile"
"github.com/carlosframework/platform/internal/fleetapi"
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/instancefs"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
@@ -304,10 +307,26 @@ func RunAgent(args []string) {
if logSink != nil { // a typed-nil in the interface would NOT be nil
eb.Logs = logSink
}
+ // The root instance-control channel (design §5.3) and the box facts
+ // a nested wake needs beyond the row (design §8.1, Task 6): the
+ // journal check and the peer list both read this SAME registry
+ // 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{
+ DataDir: dd,
+ Journal: func(host string) bool { _, err := os.Stat(provision.LayoutJournalPath(dd, host)); return err == nil },
+ Peers: reg.List,
+ }
+ eb.NestedOK = nestedGate(instancefs.MountNoSymfollow, eb.Control.Ready)
svc = activator.New(*cfg, &activator.SplitBackend{
Exec: eb,
Unit: &activator.UnitBackend{OptDir: release.OptDir()},
}, reg) // the registry satisfies activator.DueStore directly
+ // doWake's own re-read of the committed row before it spawns
+ // (design §6.8) — the registry satisfies RouteLookup with its own
+ // Get, exactly as DueStore above.
+ svc.Routes = reg
if fleetStore != nil {
// MUST be assigned before Start() below spawns the sweeper
// goroutine, which reads svc.OnWake on every successful wake from
@@ -329,6 +348,18 @@ func RunAgent(args []string) {
eb.Mirror = m
}
armMovesPass(*cfg, dd, reg, svc)
+ // The startup sweep (design §6.7): a leftover instance unit from a
+ // previous agent process — one this box's registry no longer holds
+ // a live wake for — is stopped before this process starts spawning
+ // its own. Best-effort, like every other startup duty here: a
+ // 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()
act = svc
log.Printf("agent: hibernation activator armed (bucket %s, idle %s)", cfg.Bucket, cfg.Idle)
@@ -494,6 +525,30 @@ func RunAgent(args []string) {
stopHousekeeper()
}
+// nestedGate is what an isolated wake requires of the box (design §8.5,
+// §8.6): the three nosymfollow mounts a nested instance's tree depends on,
+// and a control service that reports ready. The answer is recomputed on
+// every wake (the mounts are cheap to read; /ready is one short request),
+// so a box that becomes ready needs no agent restart and a box that stops
+// being ready refuses the very next wake rather than the stale verdict a
+// once-at-startup check would give.
+func nestedGate(mounts func(string) (bool, error), ready func(context.Context) error) func() error {
+ return func() error {
+ for _, m := range []string{"/data/instances", "/run/carlos/instances", "/run/carlos/logs"} {
+ ok, err := mounts(m)
+ if err != nil {
+ return fmt.Errorf("%s: %v", m, err)
+ }
+ if !ok {
+ return fmt.Errorf("%s is not a nosymfollow mount", m)
+ }
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineReady)
+ defer cancel()
+ return ready(ctx)
+ }
+}
+
// peekRegistryPath pre-parses the agent args for -registry (default
// <dataDir>/registry.db) without disturbing them — edge.RunWithActivator does
// the authoritative parse. The set mirrors edge.Run's flags so a full command
diff --git a/internal/host/agent_isolated_test.go b/internal/host/agent_isolated_test.go
new file mode 100644
index 000000000..f54d110c7
--- /dev/null
+++ b/internal/host/agent_isolated_test.go
@@ -0,0 +1,61 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "strings"
+ "testing"
+)
+
+// TestNestedGateRefusesOnAMissingMount pins the first half of the isolated
+// wake gate (design §8.5): the three nosymfollow mounts are checked in
+// order, and the first one that is missing (or errors) names itself in the
+// refusal rather than the gate silently moving on to /ready.
+func TestNestedGateRefusesOnAMissingMount(t *testing.T) {
+ mounts := func(mountpoint string) (bool, error) {
+ if mountpoint == "/run/carlos/instances" {
+ return false, nil
+ }
+ return true, nil
+ }
+ readyCalled := false
+ ready := func(context.Context) error { readyCalled = true; return nil }
+ err := nestedGate(mounts, ready)()
+ if err == nil || !strings.Contains(err.Error(), "/run/carlos/instances") {
+ t.Fatalf("nestedGate() = %v, want an error naming /run/carlos/instances", err)
+ }
+ if readyCalled {
+ t.Fatal("nestedGate called /ready after a mount already refused the wake")
+ }
+}
+
+// TestNestedGateRefusesOnAMountCheckError pins the same refusal for a
+// mounts func that itself errors (an unreadable /proc/self/mountinfo, in
+// production), rather than nestedGate treating an error as "not mounted"
+// silently or panicking on it.
+func TestNestedGateRefusesOnAMountCheckError(t *testing.T) {
+ wantErr := errors.New("read mountinfo: permission denied")
+ mounts := func(string) (bool, error) { return false, wantErr }
+ err := nestedGate(mounts, func(context.Context) error { return nil })()
+ if err == nil || !strings.Contains(err.Error(), wantErr.Error()) {
+ t.Fatalf("nestedGate() = %v, want it to carry %v", err, wantErr)
+ }
+}
+
+// TestNestedGatePassesMountsThenChecksReady pins the second half: once
+// every mount checks out, the gate's answer is exactly whatever /ready
+// says — refused when the control service is not ready, allowed when it is.
+func TestNestedGatePassesMountsThenChecksReady(t *testing.T) {
+ mounts := func(string) (bool, error) { return true, nil }
+
+ wantErr := errors.New("instance control: box not ready")
+ gate := nestedGate(mounts, func(context.Context) error { return wantErr })
+ if err := gate(); err != wantErr {
+ t.Fatalf("nestedGate() = %v, want the /ready error %v", err, wantErr)
+ }
+
+ gate = nestedGate(mounts, func(context.Context) error { return nil })
+ if err := gate(); err != nil {
+ t.Fatalf("nestedGate() = %v, want nil once every mount and /ready both pass", err)
+ }
+}
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index c3d195093..b4fda635f 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -97,6 +97,7 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry,
rt, err := reg.Get(host)
return err == nil && rt != nil && rt.Provisioned
},
+ Route: reg.Get,
Checkpoint: func(ctx context.Context, dbPath string) error {
host := strings.TrimSuffix(filepath.Base(dbPath), ".db")
rep, err := lease.ReplicateWithLease(ctx, dbPath, hib.Bucket, activator.HibernatePrefix(hib.Prefix, host), hib.Region)
@@ -139,6 +140,12 @@ type moveCfg struct {
// each box has its OWN registry, so the sibling's Quiesced is
// trivially, wrongly true. The box that owns the route owns the move.
OwnsRoute func(host string) bool
+ // Route reads the row itself, so a cutting move can checkpoint the
+ // row's OWN DBPath (isolated-exec design §9) instead of guessing the
+ // legacy <DataDir>/<host>.db — the guess is wrong for a nested row,
+ // whose DB lives under provision.InstanceDir. Nil (a test with no
+ // registry in hand) keeps the guess, unchanged.
+ Route func(host string) (*registry.Route, error)
// Checkpoint re-replicates the LOCAL DB to the own bucket and waits
// (review round 1): the activator's stop logs — and swallows — a failed
// final checkpoint, because pre-move the local warm cache made that
@@ -276,13 +283,32 @@ func runMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec prov
}
return nil // the retire is in flight; the nudged pass re-enters here
}
+ // A host whose layout migration is in flight (isolated-exec design
+ // §8.2) is fenced here exactly as Classify fences it for a wake: its
+ // DB may be mid-copy between the flat and nested paths, and neither
+ // location is safe to checkpoint or purge until the journal clears.
+ // Retried next pass, like every other not-yet-ready state above.
+ if _, err := os.Stat(provision.LayoutJournalPath(cfg.DataDir, rec.Host)); err == nil {
+ log.Printf("moves: %s: layout migration in flight; leaving the move for the next pass", rec.Host)
+ return nil
+ }
// Step 3: prove the own replica carries the tail. The activator's
// stop SWALLOWS a failed final checkpoint (safe pre-move: the warm
// cache stood behind it; fatal here: step 5 deletes that cache).
// Re-checkpointing an already-synced DB is cheap and idempotent; a
// held lease (the child not fully released) or an S3 blip returns
// an error and the pass retries.
+ //
+ // dbPath defaults to the legacy flat guess and is overridden by the
+ // row's own DBPath when Route is wired (isolated-exec design §9): a
+ // nested row's DB never lived at the flat guess at all, so
+ // checkpointing and purging (step 5, below) must follow the row.
dbPath := filepath.Join(cfg.DataDir, rec.Host+".db")
+ if cfg.Route != nil {
+ if rt, err := cfg.Route(rec.Host); err == nil && rt != nil && rt.DBPath != "" {
+ dbPath = rt.DBPath
+ }
+ }
if cfg.Checkpoint != nil {
if _, statErr := os.Stat(dbPath); statErr == nil {
if err := cfg.Checkpoint(ctx, dbPath); err != nil {
diff --git a/internal/host/movespass_test.go b/internal/host/movespass_test.go
index 4cd60ecd7..2426c692e 100644
--- a/internal/host/movespass_test.go
+++ b/internal/host/movespass_test.go
@@ -10,6 +10,7 @@ import (
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
@@ -242,6 +243,75 @@ func TestMovesPassSkipsWithoutTargetCoords(t *testing.T) {
}
}
+// A cutting move for a host whose layout migration is in flight (the
+// journal at provision.LayoutJournalPath) does nothing this round: its DB
+// may be mid-copy between the flat and nested paths, so neither checkpoint
+// nor the warm-cache purge is safe until the journal clears — the same fence
+// Classify applies to a wake, applied here to a cut.
+func TestMovesPassCutFencedByLayoutJournal(t *testing.T) {
+ h := newMoveHarness(t)
+ host := "a.go.example.com"
+ h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
+ h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
+ journal := provision.LayoutJournalPath(h.cfg.DataDir, host)
+ if err := os.MkdirAll(filepath.Dir(journal), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := writeFileForTest(journal, "{}"); err != nil {
+ t.Fatal(err)
+ }
+ h.quiet = true
+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
+ t.Fatal(err)
+ }
+ if len(h.f.ops) != 0 {
+ t.Fatalf("cut proceeded past the layout fence: %v", h.f.ops)
+ }
+ if got := h.read(t, host); got.Region != "ie" || got.Move.State != provision.MoveCutting {
+ t.Fatalf("cut state advanced past the layout fence: %+v", got)
+ }
+}
+
+// A cutting move for a nested row checkpoints the row's OWN DBPath, not the
+// legacy <DataDir>/<host>.db guess: a nested row's DB lives under
+// provision.InstanceDir and the flat guess names a file that was never
+// written for it.
+func TestMovesPassCutUsesRouteDBPath(t *testing.T) {
+ h := newMoveHarness(t)
+ host := "a.go.example.com"
+ h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
+ h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
+ nested := provision.NestedDBPath(h.cfg.DataDir, host)
+ if err := os.MkdirAll(filepath.Dir(nested), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := writeFileForTest(nested, "stale"); err != nil {
+ t.Fatal(err)
+ }
+ h.cfg.Route = func(string) (*registry.Route, error) {
+ return &registry.Route{Host: host, DBPath: nested}, nil
+ }
+ h.quiet = true
+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
+ t.Fatal(err)
+ }
+ found := false
+ for _, op := range h.f.ops {
+ if op == "checkpoint "+filepath.Base(nested) {
+ found = true
+ }
+ }
+ if !found {
+ t.Fatalf("checkpoint did not use the route's own DBPath: %v", h.f.ops)
+ }
+ if fileExistsForTest(nested) {
+ t.Fatal("the route's own DB survived the cut — the warm-cache trap did not follow DBPath")
+ }
+ if got := h.read(t, host); got.Region != "uk" || got.Move != nil {
+ t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
+ }
+}
+
func writeFileForTest(path, content string) error { return os.WriteFile(path, []byte(content), 0o600) }
func fileExistsForTest(path string) bool { _, err := os.Stat(path); return err == nil }
diff --git a/internal/host/schedule_wake_test.go b/internal/host/schedule_wake_test.go
index 691c80e62..fb6e48293 100644
--- a/internal/host/schedule_wake_test.go
+++ b/internal/host/schedule_wake_test.go
@@ -238,7 +238,7 @@ func TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain(t *testing.T) {
now := time.Now().UTC()
occurrence := now.Unix() - 5
r := &schedule.Runner{
- Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd, svc),
+ Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd, svc, reg),
Now: func() time.Time { return now }, Timeout: 30 * time.Second,
Box: "box1", Build: "test",
}
@@ -257,7 +257,7 @@ func TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain(t *testing.T) {
case <-time.After(30 * time.Second):
t.Fatal("the handler never ran — the tick did not wake the sleeping instance")
}
- tok, err := scheduleToken(dd, svc)(host)
+ tok, err := scheduleToken(dd, svc, reg)(host)
if err != nil {
t.Fatal(err)
}
diff --git a/internal/host/schedule_wiring.go b/internal/host/schedule_wiring.go
index 6774ca4c1..da721303e 100644
--- a/internal/host/schedule_wiring.go
+++ b/internal/host/schedule_wiring.go
@@ -15,6 +15,7 @@ import (
"github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/schedule"
@@ -84,7 +85,7 @@ func newScheduleDeps(reg *registry.Registry, store release.Store, svc *activator
runner: &schedule.Runner{
Reg: reg,
Deliver: &activatorDeliverer{svc: svc},
- Token: scheduleToken(dd, svc),
+ Token: scheduleToken(dd, svc, reg),
Event: event,
Box: box,
Build: version.Build,
@@ -157,10 +158,11 @@ func armSchedule(reg *registry.Registry, store release.Store, storeErr error, sv
// The error is redacted: it reaches the row's last_status and from there the
// bucket status object and `carlos schedule ls`, and a member must not learn
// the box's filesystem layout from it.
-func scheduleToken(dd string, svc *activator.Service) func(host string) (string, error) {
- read := schedule.TokenFromEnvFile(dd)
+func scheduleToken(dd string, svc *activator.Service, reg *registry.Registry) func(host string) (string, error) {
+ paths := envPathFor(dd, reg)
+ read := schedule.TokenFromEnvPaths(paths)
prepare := func(host string) (string, error) {
- if err := activator.EnsureAdminToken(filepath.Join(dd, host+".env")); err != nil {
+ if err := activator.EnsureAdminToken(paths(host)); err != nil {
return "", errors.New(activator.RedactPaths(err.Error()))
}
tok, err := read(host)
@@ -179,6 +181,31 @@ func scheduleToken(dd string, svc *activator.Service) func(host string) (string,
}
}
+// envPathFor is scheduleToken's route to an instance's own env file: flat by
+// default, and the nested layout's own <host>.env (design §7) when this
+// box's registry classifies the host as a nested row — the SAME
+// classification the exec backend used to decide where it spawned the
+// instance in the first place (activator.Classify), so the runner and the
+// token-ensure step ask the file the backend actually wrote rather than
+// guessing the flat path for every host. reg nil (a test with no registry
+// in hand) is every host flat, byte-identical to before this existed.
+func envPathFor(dd string, reg *registry.Registry) func(host string) string {
+ flat := func(host string) string { return filepath.Join(dd, host+".env") }
+ if reg == nil {
+ return flat
+ }
+ return func(host string) string {
+ rt, err := reg.Get(host)
+ if err != nil || rt == nil {
+ return flat(host)
+ }
+ if layout, _ := activator.Classify(*rt, activator.LayoutInputs{DataDir: dd}); layout == activator.LayoutNested {
+ return filepath.Join(provision.InstanceDir(dd, host), host+".env")
+ }
+ return flat(host)
+ }
+}
+
// convergePassTimeout bounds ONE converge pass. The agent's three call sites
// (the seed and the serial leg in fastConvergeLoop, the hourly floor in
// adoptLoop) all hand this context.Background() — an unbounded context — and
diff --git a/internal/host/schedule_wiring_test.go b/internal/host/schedule_wiring_test.go
index ea89a293e..0b2c134ef 100644
--- a/internal/host/schedule_wiring_test.go
+++ b/internal/host/schedule_wiring_test.go
@@ -121,7 +121,7 @@ func TestScheduleTokenMintsForASleepingInstance(t *testing.T) {
if _, err := os.Stat(envPath); !os.IsNotExist(err) {
t.Fatalf("fixture: env file already exists (%v)", err)
}
- tok, err := scheduleToken(dd, nil)(host)
+ tok, err := scheduleToken(dd, nil, nil)(host)
if err != nil {
t.Fatalf("Token on a host with no env file: %v", err)
}
@@ -135,7 +135,7 @@ func TestScheduleTokenMintsForASleepingInstance(t *testing.T) {
if err := activator.EnsureAdminToken(envPath); err != nil {
t.Fatal(err)
}
- again, err := scheduleToken(dd, nil)(host)
+ again, err := scheduleToken(dd, nil, nil)(host)
if err != nil || again != tok {
t.Fatalf("second read = %q (%v), want the first token %q", again, err, tok)
}
@@ -150,7 +150,7 @@ func TestScheduleTokenMintsForASleepingInstance(t *testing.T) {
// filesystem layout.
func TestScheduleTokenErrorCarriesNoAbsolutePath(t *testing.T) {
dd := filepath.Join(t.TempDir(), "missing-data-dir")
- _, err := scheduleToken(dd, nil)("x.bab.oncarlos.com")
+ _, err := scheduleToken(dd, nil, nil)("x.bab.oncarlos.com")
if err == nil {
t.Fatal("Token succeeded with no data dir")
}
diff --git a/internal/ops/litestream.go b/internal/ops/litestream.go
index 6319d3a53..6b14b482f 100644
--- a/internal/ops/litestream.go
+++ b/internal/ops/litestream.go
@@ -355,6 +355,17 @@ func FleetDBsWithSkipped(reg *registry.Registry, dataDir string) (dbs []string,
if err != nil {
return nil, nil, err
}
+ // A nested row's DB lives one level down, under
+ // <dataDir>/instances/<id>/<host>.db (isolated-exec design §7): the flat
+ // glob above never walks into that directory, so its matches are folded
+ // in here before the claimed-files gate below, which already knows how
+ // to accept or skip each one — a nested row's DBPath was added to
+ // claimed by the very same loop that adds a flat row's.
+ nested, err := filepath.Glob(filepath.Join(dataDir, "instances", "*", "*.db"))
+ if err != nil {
+ return nil, nil, err
+ }
+ matches = append(matches, nested...)
// THE GLOB PROPOSES, THE REGISTRY DISPOSES. A *.db under dataDir is
// replicated when a route claims it, or when it is the box's own registry
// or console. Anything else is somebody's leftover, and streaming leftovers
diff --git a/internal/ops/ops_test.go b/internal/ops/ops_test.go
index 1cb67afc4..9bd1f4780 100644
--- a/internal/ops/ops_test.go
+++ b/internal/ops/ops_test.go
@@ -9,6 +9,7 @@ import (
"testing"
"time"
+ "github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
_ "modernc.org/sqlite"
@@ -84,6 +85,41 @@ func TestConfigGenSkipsHibernatingRoutes(t *testing.T) {
}
}
+// TestFleetDBsDiscoversNestedDatabases pins the isolated-exec companion to
+// the flat glob above (design §7): a route's DBPath being "claimed" only
+// FILTERS what the glob already found — for a non-unit route it never adds
+// a path by itself — so a nested row's database, under
+// <dataDir>/instances/<id>/<host>.db, would stay invisible to replication
+// forever without the nested glob to find it in the first place.
+func TestFleetDBsDiscoversNestedDatabases(t *testing.T) {
+ dir := t.TempDir()
+ reg, err := registry.Open(filepath.Join(dir, "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ host := "nested.example.com"
+ nested := provision.NestedDBPath(dir, host)
+ if err := os.MkdirAll(provision.InstanceDir(dir, host), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(nested, []byte("x"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ // Not hibernating: a resident, activator-managed nested row is exactly
+ // as replicable by this box-wide pass as a flat one.
+ if err := reg.Upsert(registry.Route{Host: host, DBPath: nested, Socket: provision.NestedSocket(host)}); err != nil {
+ t.Fatal(err)
+ }
+
+ dbs, err := FleetDBs(reg, dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !slices.Contains(dbs, nested) {
+ t.Fatalf("nested database missing from FleetDBs: %v, want %s", dbs, nested)
+ }
+}
+
// TestFleetDBsUnitTenantWithoutDBPathIsInvisible pins the honest half of the
// I1 fix: a BackingUnit route added without -db has no discoverable database
// path (the platform enforces no filename convention inside a unit tenant's
diff --git a/internal/provision/provision.go b/internal/provision/provision.go
index 525201223..9f5117c8f 100644
--- a/internal/provision/provision.go
+++ b/internal/provision/provision.go
@@ -25,7 +25,6 @@ import (
"golang.org/x/crypto/bcrypt"
- "github.com/carlosframework/platform/internal/appconfig"
"github.com/carlosframework/platform/internal/controlsock"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
@@ -420,7 +419,7 @@ func ValidateShape(rec Record) error {
if rec.Unit != "" && !registry.ValidUnitName(rec.Unit) {
return fmt.Errorf("unit %q: want a systemd service unit name (e.g. console.service)", rec.Unit)
}
- if rec.Environment != "" && !appconfig.ValidEnvironmentName(rec.Environment) {
+ if rec.Environment != "" && !validEnvironmentName(rec.Environment) {
return fmt.Errorf("environment %q: want [a-z0-9-]{1,32}", rec.Environment)
}
if rec.KindOrDefault() == "static" {
@@ -552,6 +551,25 @@ var regionPoolRe = regexp.MustCompile(`^[a-z0-9-]{0,16}$`)
func ValidRegion(region string) bool { return regionPoolRe.MatchString(region) }
func ValidPool(pool string) bool { return regionPoolRe.MatchString(pool) }
+// envNameRe is appconfig.ValidEnvironmentName's own shape (pipelines design
+// §4a), spelled out again here rather than imported: appconfig already
+// imports this package (LoadMerged reads a route's environment column),
+// so the reverse import would be a cycle. A record's Environment column
+// and a bundle's environment name are the SAME vocabulary; if the shape
+// in appconfig/environments.go ever changes, this copy must change with
+// it — it exists only because of the import direction, not because the
+// rule is provision's to define.
+var envNameRe = regexp.MustCompile(`^[a-z0-9-]{1,32}$`)
+
+// validEnvironmentName is appconfig.ValidEnvironmentName's check, duplicated
+// for the reason envNameRe's comment gives: "default" names the implicit
+// base bundle, never a named overlay, and "" is not a record's problem to
+// validate here (the field is optional; the caller above already gates on
+// rec.Environment != "").
+func validEnvironmentName(name string) bool {
+ return name != "default" && envNameRe.MatchString(name)
+}
+
// GetManifest reads app's instances manifest. Not-found stays
// os.ErrNotExist-wrapped (absence IS the "instances disabled" answer, not
// an error condition the caller must special-case beyond errors.Is). A
diff --git a/internal/schedule/runner.go b/internal/schedule/runner.go
index e4b5c1e7b..b13b08001 100644
--- a/internal/schedule/runner.go
+++ b/internal/schedule/runner.go
@@ -454,11 +454,16 @@ func BuildRequest(rt registry.Route, t registry.Timer, token, build string) (*ht
return req, 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) {
+// TokenFromEnvPaths is the production Token with the env file's location
+// taken from envPath rather than assumed flat: isolated-exec's nested rows
+// 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.
+func TokenFromEnvPaths(envPath func(host string) string) func(host string) (string, error) {
return func(host string) (string, error) {
- env, err := readEnvFile(filepath.Join(dataDir, host+".env"))
+ env, err := readEnvFile(envPath(host))
if err != nil {
return "", err
}
@@ -468,3 +473,9 @@ func TokenFromEnvFile(dataDir string) func(host string) (string, error) {
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") })
+}