amadan

carlos / platform Public

activator: layout classification and per-host paths by layout (plan 3, task 2)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 10df49b2aa81f72e1c20dd647bed490bcef46c8d parent d86427f
5 files changed, +237 −40
  • internal/activator/activator_test.go +7 −7
  • internal/activator/backend_exec.go +21 −33
  • internal/activator/layout.go +94 −0
  • internal/activator/layout_test.go +67 −0
  • internal/activator/paths.go +48 −0

CI passed — run details

diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 5893ae761..8127d94f1 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -3260,7 +3260,7 @@ func stageSeedScript(t *testing.T, b *ExecBackend, host, ver, body string) {
func TestExecBackendRestoreWarmCacheIsNotFresh(t *testing.T) {
b := testExecBackend(t)
host := "warm.example.com"
- if err := os.WriteFile(b.defaultDBPath(host), []byte("db"), 0o600); err != nil {
+ if err := os.WriteFile(filepath.Join(b.DataDir, host+".db"), []byte("db"), 0o600); err != nil {
t.Fatal(err)
}
called := false
@@ -3307,7 +3307,7 @@ func TestExecBackendRestoreFromReplicaIsNotFresh(t *testing.T) {
func TestExecBackendRestoreCorruptReplicaFailsAndQuarantines(t *testing.T) {
b := testExecBackend(t)
host := "damaged.example.com"
- dbPath := b.defaultDBPath(host)
+ dbPath := filepath.Join(b.DataDir, host+".db")
swapRestoreReplica(t, func(_ context.Context, out, _, _, _ string) (bool, error) {
if err := os.WriteFile(out, []byte("this is not a sqlite database at all"), 0o600); err != nil {
@@ -3436,7 +3436,7 @@ func TestExecBackendSeedRunsDeclaredArgvWithDBPath(t *testing.T) {
}
gotArgv := strings.Fields(waitForFile(t, argvPath, time.Second))
- wantArgv := []string{"seed", "--fixtures", "--db", b.defaultDBPath(host)}
+ wantArgv := []string{"seed", "--fixtures", "--db", filepath.Join(b.DataDir, host+".db")}
if !slices.Equal(gotArgv, wantArgv) {
t.Fatalf("seed argv = %v, want %v", gotArgv, wantArgv)
}
@@ -3593,7 +3593,7 @@ func TestExecBackendSeedFailureRemovesHalfBuiltDB(t *testing.T) {
if err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `["seed"]`}); err == nil {
t.Fatal("a seed exiting 1 was reported as success")
}
- if _, err := os.Stat(b.defaultDBPath(host)); err == nil {
+ if _, err := os.Stat(filepath.Join(b.DataDir, host+".db")); err == nil {
t.Fatal("a failed seed left its half-built database behind — the next wake would treat it as a warm cache")
}
}
@@ -3605,14 +3605,14 @@ func TestExecBackendSeedFailureKeepsPreexistingDB(t *testing.T) {
b := testExecBackend(t)
host := "existing-db.example.com"
stageSeedScript(t, b, host, "v1", "echo nope >&2\nexit 1\n")
- if err := os.WriteFile(b.defaultDBPath(host), []byte("live data"), 0o600); err != nil {
+ if err := os.WriteFile(filepath.Join(b.DataDir, host+".db"), []byte("live data"), 0o600); err != nil {
t.Fatal(err)
}
if err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `["seed"]`}); err == nil {
t.Fatal("a seed exiting 1 was reported as success")
}
- got, err := os.ReadFile(b.defaultDBPath(host))
+ got, err := os.ReadFile(filepath.Join(b.DataDir, host+".db"))
if err != nil || string(got) != "live data" {
t.Fatalf("pre-existing database = %q, %v; want it untouched by a failed seed", got, err)
}
@@ -3637,7 +3637,7 @@ func TestExecBackendSeedSucceedsDespiteLingeringChild(t *testing.T) {
if err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `["seed"]`}); err != nil {
t.Fatalf("Seed = %v, want nil — the seed exited 0; a leaked output pipe is not a failed seed", err)
}
- got, err := os.ReadFile(b.defaultDBPath(host))
+ got, err := os.ReadFile(filepath.Join(b.DataDir, host+".db"))
if err != nil || string(got) != "seeded" {
t.Fatalf("seeded database = %q, %v; want it kept — deleting it would make the host fresh again next wake", got, err)
}
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 7cafa366e..e6c46dc00 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -106,6 +106,9 @@ 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
}
// execLogLineMax caps one teed log line (spec §5): longer is truncated with
@@ -158,9 +161,6 @@ func (b *ExecBackend) spawnInCgroup(rt registry.Route, build func() *exec.Cmd) (
func (b *ExecBackend) liveDir() string { return filepath.Join(b.OptDir, "live") }
func (b *ExecBackend) liveLink(host string) string { return filepath.Join(b.liveDir(), host) }
-func (b *ExecBackend) defaultDBPath(host string) string {
- return filepath.Join(b.DataDir, host+".db")
-}
// LiveVersion reports the version host's live symlink currently points at, or
// "" (no error) if the host hasn't been put under version management yet. It
@@ -196,7 +196,7 @@ func (b *ExecBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, er
}
return nil, err
}
- return &execLease{held: held, defaultDB: b.defaultDBPath(host)}, nil
+ return &execLease{held: held, defaultDB: b.pathsFor(rt).dbPath}, nil
}
// execLease adapts *lease.Held to the Lease seam.
@@ -219,10 +219,7 @@ func (l *execLease) Replicate(ctx context.Context, dbPath string, syncInterval t
func (l *execLease) Release(ctx context.Context) error { return l.held.Release(ctx) }
func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, error) {
- host, dbPath := rt.Host, rt.DBPath
- if dbPath == "" {
- dbPath = b.defaultDBPath(host)
- }
+ host, dbPath := rt.Host, b.pathsFor(rt).dbPath
if _, err := os.Stat(dbPath); err == nil {
return false, nil // warm cache — see doWake
} else if !os.IsNotExist(err) {
@@ -344,21 +341,19 @@ func (b *ExecBackend) Seed(ctx context.Context, rt registry.Route) error {
if err := b.ensureLiveLink(ctx, host); err != nil {
return err
}
+ paths := b.pathsFor(rt)
// Mint before the env is read, exactly as Start does — a seed is the FIRST
// thing a fresh instance runs, so this is usually where the token comes
// into existence, and instanceEnv's contract is that a seed sees the same
// environment the instance will.
- if err := EnsureAdminToken(filepath.Join(b.DataDir, host+".env")); err != nil {
+ if err := EnsureAdminToken(paths.envFile); err != nil {
return fmt.Errorf("admin token for %s: %w", host, err)
}
- fileEnv, err := b.instanceEnv(host)
+ fileEnv, err := b.instanceEnv(rt)
if err != nil {
return err
}
- dbPath := rt.DBPath
- if dbPath == "" {
- dbPath = b.defaultDBPath(host)
- }
+ dbPath := paths.dbPath
// Whether the database existed BEFORE the seed ran decides whether a
// failed seed may clean up after itself (below). On the wake path it is
@@ -505,7 +500,7 @@ func (c *capBuffer) String() string {
// same env file the spawn path hands the child (the activator holds it either
// way).
func (b *ExecBackend) NextDue(ctx context.Context, rt registry.Route) (int64, error) {
- fileEnv, err := readEnvFile(filepath.Join(b.DataDir, rt.Host+".env"))
+ fileEnv, err := readEnvFile(b.pathsFor(rt).envFile)
if err != nil {
return 0, fmt.Errorf("read env file: %w", err)
}
@@ -557,10 +552,8 @@ func (b *ExecBackend) ensureLiveLink(ctx context.Context, host string) error {
func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
host := rt.Host
- dbPath := rt.DBPath
- if dbPath == "" {
- dbPath = b.defaultDBPath(host)
- }
+ paths := b.pathsFor(rt)
+ dbPath := paths.dbPath
// Exec exactly what a systemd unit would: the per-instance live symlink
// (<OptDir>/live/<host> → a staged version), same --socket/--db flags,
@@ -577,11 +570,11 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
// instance with no token can neither answer /api/next-due nor accept a
// tick, and a wake that quietly skipped it would report a healthy instance
// that silently never runs its scheduled work — the #110 kind of lie.
- if err := EnsureAdminToken(filepath.Join(b.DataDir, host+".env")); err != nil {
+ if err := EnsureAdminToken(paths.envFile); err != nil {
return nil, "", fmt.Errorf("admin token for %s: %w", host, err)
}
- fileEnv, err := b.instanceEnv(host)
+ fileEnv, err := b.instanceEnv(rt)
if err != nil {
return nil, "", err
}
@@ -605,7 +598,7 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
// the instance serves requests either way; it just goes without its control
// verbs until the next restart, a stuck-stale cache self-heals via the
// safety-net TTL, and a missed tick is retried by the runner.
- envPath := filepath.Join(b.DataDir, host+".env")
+ envPath := paths.envFile
// Read on EVERY call, never captured: the env file can be rewritten under a
// running instance (`carlos env`, a re-mint on a later spawn), and a token
// cached at bind time would keep authorising the old secret.
@@ -783,12 +776,13 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
// belong to a serving process, not to a one-shot init. Both callers mint the
// instance's CARLOS_ADMIN_TOKEN into <host>.env first, so "the same
// environment" includes the same secret.)
-func (b *ExecBackend) instanceEnv(host string) (map[string]string, error) {
- cfgEnv, err := readEnvFile(filepath.Join(b.DataDir, host+".config.env"))
+func (b *ExecBackend) instanceEnv(rt registry.Route) (map[string]string, error) {
+ paths := b.pathsFor(rt)
+ cfgEnv, err := readEnvFile(paths.configEnv)
if err != nil && !os.IsNotExist(err) {
return nil, fmt.Errorf("read config env file: %w", err)
}
- ownEnv, err := readEnvFile(filepath.Join(b.DataDir, host+".env"))
+ ownEnv, err := readEnvFile(paths.envFile)
if err != nil && !os.IsNotExist(err) {
return nil, fmt.Errorf("read env file: %w", err)
}
@@ -863,17 +857,11 @@ func mergeEnv(base, overlay map[string]string) map[string]string {
return out
}
-// sidecarEnvPath: the host's sidecar EnvironmentFile,
-// <DataDir>/<host>-sidecar.env (Eleven's <host>-bot.env, renamed).
-func (b *ExecBackend) sidecarEnvPath(host string) string {
- return filepath.Join(b.DataDir, host+"-sidecar.env")
-}
-
// HasSidecar: a host has a sidecar iff its sidecar EnvironmentFile exists.
// Derived from the filesystem on purpose, the way systemd units would do it:
// no registry column to drift from what's actually staged.
func (b *ExecBackend) HasSidecar(rt registry.Route) bool {
- _, err := os.Stat(b.sidecarEnvPath(rt.Host))
+ _, err := os.Stat(b.pathsFor(rt).sidecarEnv)
return err == nil
}
@@ -884,7 +872,7 @@ func (b *ExecBackend) HasSidecar(rt registry.Route) bool {
// symlink exists by now: the instance spawn checked it. The ctx is unused (an
// exec doesn't block); it's here for the backend seam's symmetry.
func (b *ExecBackend) StartSidecar(_ context.Context, rt registry.Route) (Proc, error) {
- fileEnv, err := readEnvFile(b.sidecarEnvPath(rt.Host))
+ fileEnv, err := readEnvFile(b.pathsFor(rt).sidecarEnv)
if err != nil {
return nil, fmt.Errorf("read sidecar env: %w", err)
}
diff --git a/internal/activator/layout.go b/internal/activator/layout.go
new file mode 100644
index 000000000..4efc111f6
--- /dev/null
+++ b/internal/activator/layout.go
@@ -0,0 +1,94 @@
+// Ported from plan 3 task 2 (isolated-exec design §8.1): the classification
+// that decides whether a route's paths are today's flat layout, the nested
+// per-instance layout, or something an operator needs to look at.
+package activator
+
+import (
+ "fmt"
+ "path/filepath"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// Layout is the classification of an activator-managed route's paths
+// (isolated-exec design §8.1). It decides the spawn path: flat and foreign
+// rows run today's exec child; nested rows run as units; fenced and
+// conflict rows refuse to wake.
+type Layout string
+
+const (
+ LayoutFlat Layout = "flat"
+ LayoutNested Layout = "nested"
+ LayoutFenced Layout = "fenced"
+ LayoutConflict Layout = "conflict"
+ LayoutForeign Layout = "foreign"
+)
+
+// LayoutInputs are the box facts Classify needs beyond the row: where the
+// data dir is, whether a migration journal exists for a host (plan 4
+// writes them), and the other rows on the box (for identifier conflicts).
+// Nil seams mean "never fenced" and "never conflict", which is what unit
+// tests without a box want.
+type LayoutInputs struct {
+ DataDir string
+ // SocketDir is where flat sockets live and under which the nested
+ // instances directory sits. Empty means provision.SocketDir; a test
+ // points it at a temp dir so a route can classify as nested without
+ // writing to /run/carlos.
+ SocketDir string
+ Journal func(host string) bool
+ Peers func() ([]registry.Route, error)
+ // SameID reports whether two hosts share an instance identifier. Nil
+ // means provision.InstanceID equality; tests inject a collision.
+ SameID func(a, b string) bool
+}
+
+// sockets returns the flat and nested socket paths for host under the
+// configured socket directory.
+func (in LayoutInputs) sockets(host string) (flat, nested string) {
+ dir := in.SocketDir
+ if dir == "" {
+ dir = provision.SocketDir
+ }
+ return filepath.Join(dir, host+".sock"), filepath.Join(dir, "instances", provision.InstanceID(host), "app.sock")
+}
+
+// ActivatorManaged is the activator's own predicate for a route it owns
+// (activator.go's per-host gate and edge.go's proxy decision): an instance
+// route, hibernate-flagged, with a socket, not unit-backed, not
+// tunnel-served. An unflagged socket route is an externally supervised
+// process and is nobody's to migrate or isolate (design §8.2 step 1).
+func ActivatorManaged(rt registry.Route) bool {
+ return rt.Kind == "instance" && rt.Hibernate && rt.Socket != "" && rt.Backing == "" && rt.Tunnel == ""
+}
+
+// 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) {
+ if in.Journal != nil && in.Journal(rt.Host) {
+ return LayoutFenced, "layout migration in flight; see " + provision.LayoutJournalPath(in.DataDir, rt.Host)
+ }
+ if in.Peers != nil {
+ same := in.SameID
+ if same == nil {
+ same = func(a, b string) bool { return provision.InstanceID(a) == provision.InstanceID(b) }
+ }
+ peers, err := in.Peers()
+ if err == nil {
+ 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)
+ }
+ }
+ }
+ }
+ flatSock, nestedSock := in.sockets(rt.Host)
+ switch {
+ case rt.DBPath == provision.DefaultDBPath(in.DataDir, rt.Host) && rt.Socket == flatSock:
+ return LayoutFlat, ""
+ case rt.DBPath == provision.NestedDBPath(in.DataDir, rt.Host) && rt.Socket == nestedSock:
+ return LayoutNested, ""
+ }
+ return LayoutForeign, fmt.Sprintf("db %q socket %q are neither the flat nor the nested defaults", rt.DBPath, rt.Socket)
+}
diff --git a/internal/activator/layout_test.go b/internal/activator/layout_test.go
new file mode 100644
index 000000000..0cff8ec3a
--- /dev/null
+++ b/internal/activator/layout_test.go
@@ -0,0 +1,67 @@
+package activator
+
+import (
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+func managed(host, db, sock string) registry.Route {
+ return registry.Route{Host: host, Kind: "instance", Hibernate: true, DBPath: db, Socket: sock}
+}
+
+func TestClassifyFlatAndNested(t *testing.T) {
+ in := LayoutInputs{DataDir: "/data"}
+ if l, _ := Classify(managed("h.example.com", "/data/h.example.com.db", "/run/carlos/h.example.com.sock"), in); l != LayoutFlat {
+ t.Errorf("flat row = %s", l)
+ }
+ if l, _ := Classify(managed("h.example.com", provision.NestedDBPath("/data", "h.example.com"), provision.NestedSocket("h.example.com")), in); l != LayoutNested {
+ t.Errorf("nested row = %s", l)
+ }
+ if l, r := Classify(managed("h.example.com", "/data/sub/store.db", "/run/carlos/h.example.com.sock"), in); l != LayoutForeign || !strings.Contains(r, "store.db") {
+ t.Errorf("custom row = %s %q", l, r)
+ }
+ // A nested DB with a flat socket is not nested: both columns must agree.
+ if l, _ := Classify(managed("h.example.com", provision.NestedDBPath("/data", "h.example.com"), "/run/carlos/h.example.com.sock"), in); l != LayoutForeign {
+ t.Errorf("half-migrated row = %s", l)
+ }
+}
+
+func TestClassifyFencedWinsOverEverything(t *testing.T) {
+ in := LayoutInputs{DataDir: "/data", Journal: func(h string) bool { return h == "h.example.com" }}
+ if l, r := Classify(managed("h.example.com", "/data/h.example.com.db", "/run/carlos/h.example.com.sock"), in); l != LayoutFenced || r == "" {
+ t.Errorf("fenced = %s %q", l, r)
+ }
+ if l, _ := Classify(managed("o.example.com", "/data/o.example.com.db", "/run/carlos/o.example.com.sock"), in); l != LayoutFlat {
+ t.Errorf("other host = %s", l)
+ }
+}
+
+// The identifier collision is exercised through the SameID seam rather than
+// a real hash collision: two nested rows for different hosts, SameID forced
+// to report them equal, must classify as a conflict that names the other
+// host.
+func TestClassifyConflictRefusesBothHosts(t *testing.T) {
+ long := strings.Repeat("a", 60) + ".example.com"
+ twin := strings.Repeat("a", 60) + ".example.org"
+ a := managed(long, provision.NestedDBPath("/data", long), provision.NestedSocket(long))
+ b := managed(twin, provision.NestedDBPath("/data", twin), provision.NestedSocket(twin))
+ in := LayoutInputs{DataDir: "/data", Peers: func() ([]registry.Route, error) { return []registry.Route{a, b}, nil }, SameID: func(x, y string) bool { return true }}
+ if l, r := Classify(a, in); l != LayoutConflict || !strings.Contains(r, twin) {
+ t.Errorf("conflict = %s %q", l, r)
+ }
+}
+
+func TestClassifyIgnoresUnmanagedPeers(t *testing.T) {
+ a := managed("h.example.com", "/data/h.example.com.db", "/run/carlos/h.example.com.sock")
+ ext := registry.Route{Host: "x.example.com", Kind: "instance", Hibernate: false, Socket: "/run/carlos/x.example.com.sock"}
+ in := LayoutInputs{DataDir: "/data", Peers: func() ([]registry.Route, error) { return []registry.Route{a, ext}, nil }, SameID: func(x, y string) bool { return true }}
+ if l, _ := Classify(a, in); l != LayoutFlat {
+ t.Errorf("an unmanaged peer must not conflict: %s", l)
+ }
+ if ActivatorManaged(ext) {
+ t.Error("unflagged socket route reported managed")
+ }
+}
diff --git a/internal/activator/paths.go b/internal/activator/paths.go
new file mode 100644
index 000000000..e1fe4a40e
--- /dev/null
+++ b/internal/activator/paths.go
@@ -0,0 +1,48 @@
+package activator
+
+import (
+ "path/filepath"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// instancePaths is every per-host path the exec backend touches, resolved
+// once per call by layout so that no method builds a path from DataDir and
+// the host on its own. Flat and foreign rows keep today's flat names; a
+// nested row's files live in its instance directory with the same names
+// (design §3.1).
+type instancePaths struct {
+ layout Layout
+ reason string
+ dataDir string // the directory that holds the four files
+ runDir string // the directory that holds app.sock and the control socket
+ dbPath string
+ envFile string
+ sidecarEnv string
+ configEnv string
+}
+
+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}
+ if l == LayoutNested {
+ p.dataDir = provision.InstanceDir(b.DataDir, rt.Host)
+ p.runDir = provision.InstanceRunDir(rt.Host)
+ }
+ p.dbPath = rt.DBPath
+ if p.dbPath == "" {
+ p.dbPath = provision.DefaultDBPath(b.DataDir, rt.Host)
+ }
+ p.envFile = filepath.Join(p.dataDir, rt.Host+".env")
+ p.sidecarEnv = filepath.Join(p.dataDir, rt.Host+"-sidecar.env")
+ p.configEnv = filepath.Join(p.dataDir, rt.Host+".config.env")
+ return p
+}
+
+// layoutInputs is the backend's LayoutInputs with DataDir filled in.
+func (b *ExecBackend) layoutInputs() LayoutInputs {
+ in := b.Layout
+ in.DataDir = b.DataDir
+ return in
+}