amadan

carlos / platform Public

feat(activator): report fresh boots and run the route's seed before first health probe

Restore widens to (fresh bool, err error) — fresh means the local DB was
absent AND no replica existed, so a restore is never mistaken for a first
boot — and the Backend seam gains Seed(ctx, rt). doWake runs the route's
seed strictly between Restore and Start (before Start's first health
probe), only when fresh and the route declares one; a seed failure fails
the whole wake and releases the lease, like a failed start.

ExecBackend.Seed execs live/<host> with the manifest's seed argv plus
--db <path>, under the instance's own merged env layers (Start's env
composition extracted into instanceEnv; the purge-channel vars stay in
Start, where a serving process needs them), bounded by seedTimeout. A
crashed seed's half-built database is removed, or the next wake would
read it as a warm cache and boot the instance on it. UnitBackend never
reports fresh — the agent cannot see inside a DynamicUser's state dir,
and "I can't tell" must not run an app's seed over live data.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell bb8f92cb0d8eba4ddce2e4f9b63d06d249baf15b parent fd9e85b
5 files changed, +770 −36
  • internal/activator/activator.go +30 −6
  • internal/activator/activator_test.go +520 −6
  • internal/activator/backend_exec.go +162 −17
  • internal/activator/backend_unit.go +18 −5
  • internal/activator/backend_unit_test.go +40 −2
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 96aa4753e..9014bfe37 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -128,8 +128,16 @@ type Backend interface {
Acquire(ctx context.Context, rt registry.Route) (Lease, error)
// Restore materializes the local DB (rt.DBPath; the backend defaults it)
// from S3 when it's absent (a no-op when the warm-cache copy is still on
- // disk).
- Restore(ctx context.Context, rt registry.Route) error
+ // disk). fresh reports that the local DB was absent AND no replica
+ // existed — a first boot, never a restore. The activator runs the route's
+ // Seed exactly then.
+ Restore(ctx context.Context, rt registry.Route) (fresh bool, err error)
+ // Seed runs the route's declared first-boot initialization (rt.Seed, a
+ // JSON argv array) against the instance's own database, before anything
+ // has spawned. Called by doWake only on a fresh boot with a seed
+ // declared, so it must be safe to assume an empty database — and its
+ // failure fails the wake, because a half-seeded instance must not serve.
+ Seed(ctx context.Context, rt registry.Route) error
// Start spawns the instance process and waits until its socket accepts. It
// also returns the build it execed (the live/<host> symlink target read at
// spawn), so the sweep can later spot a repoint under a still-running child
@@ -623,12 +631,28 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance
// S3 state written by the true holder). When it's still on disk it is the
// warm cache: on one box, with the lease held, the local copy can only be
// what we ourselves checkpointed at the last sleep.
- if err := a.backend.Restore(ctx, rt); err != nil {
+ fresh, err := a.backend.Restore(ctx, rt)
+ if err != nil {
_ = lease.Release(ctx)
return nil, fmt.Errorf("restore %s: %w", host, err)
}
- // 3. Spawn before replication starts: a fresh instance creates its own DB
+ // 3. First-boot seed, strictly between the restore and the spawn — so it
+ // runs before Start's first health probe, and a brand-new instance is born
+ // seeded rather than briefly serving an empty database. fresh is the
+ // backend's one fact worth keeping: the local DB was absent AND no replica
+ // existed, so a restore can never be mistaken for a first boot and a seed
+ // can never run over recovered data. A seed that fails fails the WHOLE
+ // wake (the Start-failure precedent: release the lease, surface the
+ // error) — a half-seeded instance must not serve.
+ if fresh && rt.Seed != "" {
+ if err := a.backend.Seed(ctx, rt); err != nil {
+ _ = lease.Release(ctx)
+ return nil, fmt.Errorf("seed %s: %w", host, err)
+ }
+ }
+
+ // 4. Spawn before replication starts: a fresh instance creates its own DB
// on boot, so waiting for the socket guarantees the file exists for
// litestream to open. Writes that land in the gap are picked up by the
// replica's first sync.
@@ -638,7 +662,7 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance
return nil, fmt.Errorf("start %s: %w", host, err)
}
- // 4. Trade the held lease for running replication, at the route's
+ // 5. Trade the held lease for running replication, at the route's
// replication-class sync interval (the per-plan durability ladder).
// The DB path rides the route; when empty the backend defaults it to
// <DataDir>/<host>.db internally (Eleven defaulted here via a package
@@ -652,7 +676,7 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance
return nil, fmt.Errorf("replicate %s: %w", host, err)
}
- // 5. Co-spawn the sidecar, last — only once the instance is healthy and
+ // 6. Co-spawn the sidecar, last — only once the instance is healthy and
// replicating, so the sidecar always boots against a live instance (its
// marked requests pass through the edge while we're awake). A spawn
// failure is loud but never fails the wake: members come before bots, and
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index eb9ad4e20..3f24130df 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -113,17 +113,38 @@ func (l *fakeLease) Replicate(_ context.Context, _ string, syncInterval time.Dur
}
return l.rep, nil
}
-func (l *fakeLease) Release(context.Context) error { return nil }
+
+// Release counts on the backend: every doWake bail-out must give the lease
+// back exactly once, and the only way to assert that from outside is to count
+// the calls the wake made.
+func (l *fakeLease) Release(context.Context) error {
+ if l.b != nil {
+ l.b.mu.Lock()
+ l.b.releases++
+ l.b.mu.Unlock()
+ }
+ return nil
+}
type fakeBackend struct {
mu sync.Mutex
acquires int
restores int
starts int
+ seeds int
+ releases int // lease releases — the failure paths' assertion
acquireErr error
- startGate chan struct{} // when non-nil, Start() blocks until closed
- reps []*fakeRep
- procs []*fakeProc
+ // restoreFresh is what Restore reports: a first boot (local DB absent AND
+ // no replica). False by default, so every pre-seed test doubles as the
+ // "a restored instance is never seeded" case.
+ restoreFresh bool
+ restoreErr error
+ seedErr error
+ seedRoutes []registry.Route
+ startErr error
+ startGate chan struct{} // when non-nil, Start() blocks until closed
+ reps []*fakeRep
+ procs []*fakeProc
// syncIntervals records, per wake, the replication-class tick doWake
// threads into Replicate() — the ladder's seam assertion.
syncIntervals []time.Duration
@@ -176,11 +197,21 @@ func (b *fakeBackend) Acquire(context.Context, registry.Route) (Lease, error) {
return &fakeLease{rep: rep, b: b}, nil
}
-func (b *fakeBackend) Restore(context.Context, registry.Route) error {
+func (b *fakeBackend) Restore(context.Context, registry.Route) (bool, error) {
b.mu.Lock()
defer b.mu.Unlock()
b.restores++
- return nil
+ b.order = append(b.order, "restore")
+ return b.restoreFresh, b.restoreErr
+}
+
+func (b *fakeBackend) Seed(_ context.Context, rt registry.Route) error {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ b.seeds++
+ b.order = append(b.order, "seed")
+ b.seedRoutes = append(b.seedRoutes, rt)
+ return b.seedErr
}
func (b *fakeBackend) Start(context.Context, registry.Route) (Proc, string, error) {
@@ -189,9 +220,13 @@ func (b *fakeBackend) Start(context.Context, registry.Route) (Proc, string, erro
b.order = append(b.order, "start:instance")
gate := b.startGate
fn := b.execVerFn
+ startErr := b.startErr
proc := &fakeProc{label: "instance", rec: b}
b.procs = append(b.procs, proc)
b.mu.Unlock()
+ if startErr != nil {
+ return nil, "", startErr
+ }
if gate != nil {
<-gate
}
@@ -261,6 +296,18 @@ func (b *fakeBackend) counts() (acquires, restores, starts int) {
return b.acquires, b.restores, b.starts
}
+func (b *fakeBackend) seedCount() int {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.seeds
+}
+
+func (b *fakeBackend) releaseCount() int {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.releases
+}
+
func (b *fakeBackend) lastProc() *fakeProc {
b.mu.Lock()
defer b.mu.Unlock()
@@ -569,6 +616,157 @@ func TestActivatorWakeThreadsReplicationClass(t *testing.T) {
}
}
+// --- first-wake seed ---------------------------------------------------------
+//
+// The seam spec §6 pins: fresh means the local DB was absent AND no replica
+// existed — a first boot, never a restore — and the route's seed runs strictly
+// between Restore and Start, i.e. before the first health probe ever reaches
+// the instance. A brand-new instance is therefore born seeded, and a restored
+// one is never re-seeded over its own data.
+
+func TestActivatorSeedsFreshInstanceBeforeStart(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Minute)
+ b.restoreFresh = true
+ rt := testRoute("fresh.example.com")
+ rt.Seed = `["seed"]`
+
+ release, err := a.beginRequest(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+
+ if got := b.seedCount(); got != 1 {
+ t.Fatalf("seed calls = %d, want 1 on a fresh boot with a seed", got)
+ }
+ ev := b.events()
+ r, s, st := indexOf(ev, "restore"), indexOf(ev, "seed"), indexOf(ev, "start:instance")
+ if r == -1 || s == -1 || st == -1 || !(r < s && s < st) {
+ t.Fatalf("seed order wrong: %v (want restore < seed < start:instance — seeded before the first health probe)", ev)
+ }
+ b.mu.Lock()
+ seededHost := b.seedRoutes[0].Host
+ b.mu.Unlock()
+ if seededHost != rt.Host {
+ t.Fatalf("seeded host = %q, want %q", seededHost, rt.Host)
+ }
+}
+
+// A restore is never a first boot: the instance's data came back from the
+// bucket, and running the seed over it is exactly the corruption this fresh
+// flag exists to prevent.
+func TestActivatorSkipsSeedWhenRestored(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Minute)
+ b.restoreFresh = false
+ rt := testRoute("restored.example.com")
+ rt.Seed = `["seed"]`
+
+ release, err := a.beginRequest(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+
+ if got := b.seedCount(); got != 0 {
+ t.Fatalf("seed calls = %d on a restored instance, want 0", got)
+ }
+}
+
+// No seed declared: a fresh boot just boots (every route added before this
+// feature, and every app that creates its own schema).
+func TestActivatorSkipsSeedWhenRouteHasNone(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Minute)
+ b.restoreFresh = true
+ rt := testRoute("noseed.example.com") // rt.Seed == ""
+
+ release, err := a.beginRequest(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+
+ if got := b.seedCount(); got != 0 {
+ t.Fatalf("seed calls = %d for a route with no seed, want 0", got)
+ }
+}
+
+// A failed seed fails the WHOLE wake — a half-seeded instance must not serve —
+// and gives the lease back exactly once, the Start-failure precedent. Nothing
+// is spawned, and the entry is gone so the next request retries from scratch.
+func TestActivatorSeedFailureFailsWakeAndReleasesLease(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Minute)
+ b.restoreFresh = true
+ b.seedErr = errors.New("seed exited 1")
+ rt := testRoute("badseed.example.com")
+ rt.Seed = `["seed"]`
+
+ _, err := a.beginRequest(context.Background(), rt)
+ if err == nil {
+ t.Fatal("wake succeeded despite a failing seed")
+ }
+ if !strings.Contains(err.Error(), "seed "+rt.Host) {
+ t.Fatalf("wake error = %q, want it to name the seed and the host", err)
+ }
+ if _, _, starts := b.counts(); starts != 0 {
+ t.Fatalf("instance was spawned after a failed seed (starts=%d)", starts)
+ }
+ if got := b.releaseCount(); got != 1 {
+ t.Fatalf("lease releases = %d after a failed seed, want exactly 1", got)
+ }
+ if a.entryCount() != 0 {
+ t.Fatal("failed wake left an entry behind")
+ }
+}
+
+// The other two doWake bail-outs release the lease exactly once too — the
+// property the seed step slots into the middle of.
+func TestActivatorWakeFailuresReleaseLeaseOnce(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ inject func(*fakeBackend)
+ want string
+ }{
+ {"restore", func(b *fakeBackend) { b.restoreErr = errors.New("s3 down") }, "restore "},
+ {"start", func(b *fakeBackend) { b.startErr = errors.New("never healthy") }, "start "},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Minute)
+ tc.inject(b)
+ rt := testRoute(tc.name + "-fail.example.com")
+
+ _, err := a.beginRequest(context.Background(), rt)
+ if err == nil {
+ t.Fatalf("wake succeeded despite a failing %s", tc.name)
+ }
+ if !strings.Contains(err.Error(), tc.want+rt.Host) {
+ t.Fatalf("wake error = %q, want it to name %q and the host", err, tc.want)
+ }
+ if got := b.releaseCount(); got != 1 {
+ t.Fatalf("lease releases = %d after a failed %s, want exactly 1", got, tc.name)
+ }
+ if a.entryCount() != 0 {
+ t.Fatal("failed wake left an entry behind")
+ }
+ })
+ }
+}
+
+// A restore failure never reaches the seed: seeding a host whose data we
+// couldn't materialize is precisely the "half-seeded instance" case.
+func TestActivatorRestoreFailureSkipsSeed(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Minute)
+ b.restoreErr = errors.New("s3 down")
+ rt := testRoute("restorefail.example.com")
+ rt.Seed = `["seed"]`
+
+ if _, err := a.beginRequest(context.Background(), rt); err == nil {
+ t.Fatal("wake succeeded despite a failing restore")
+ }
+ if got := b.seedCount(); got != 0 {
+ t.Fatalf("seed calls = %d after a failed restore, want 0", got)
+ }
+}
+
// After the idle window with nothing in flight, the sweeper hibernates the
// instance: process stopped, replication closed (final checkpoint + lease
// release), entry forgotten; the next request wakes it afresh.
@@ -2481,3 +2679,319 @@ func TestExecBackendStartPurgesHostCacheOnSpawn(t *testing.T) {
t.Fatal("another host's cached responses must survive a spawn")
}
}
+
+// --- ExecBackend: fresh detection + seed --------------------------------------
+
+// stageSeedScript stages host's live binary as a shell script with a chosen
+// body — the seed tests need exits and stderr the recording scripts above
+// don't produce. Returns nothing: each test knows the files its body writes.
+func stageSeedScript(t *testing.T, b *ExecBackend, host, ver, body string) {
+ t.Helper()
+ vdir := filepath.Join(b.OptDir, "versions", ver)
+ if err := os.MkdirAll(vdir, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ bin := filepath.Join(vdir, "app")
+ if err := os.WriteFile(bin, []byte("#!/bin/sh\n"+body), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.MkdirAll(filepath.Join(b.OptDir, "live"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ os.Remove(b.liveLink(host))
+ if err := os.Symlink(bin, b.liveLink(host)); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// The warm cache — the local DB still on disk from the last sleep — is never a
+// first boot, and the bucket isn't even consulted.
+func TestExecBackendRestoreWarmCacheIsNotFresh(t *testing.T) {
+ b := testExecBackend(t)
+ host := "warm.example.com"
+ if err := os.WriteFile(b.defaultDBPath(host), []byte("db"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ called := false
+ swapRestoreReplica(t, func(context.Context, string, string, string, string) (bool, error) {
+ called = true
+ return false, nil
+ })
+
+ fresh, err := b.Restore(context.Background(), registry.Route{Host: host, Kind: "instance"})
+ if err != nil {
+ t.Fatalf("Restore: %v", err)
+ }
+ if fresh {
+ t.Fatal("warm cache reported fresh — the seed would run over live data")
+ }
+ if called {
+ t.Fatal("warm cache consulted the bucket")
+ }
+}
+
+// DB absent but a replica exists: the data came back from S3, so this is a
+// restore, not a first boot.
+func TestExecBackendRestoreFromReplicaIsNotFresh(t *testing.T) {
+ b := testExecBackend(t)
+ swapRestoreReplica(t, func(context.Context, string, string, string, string) (bool, error) {
+ return true, nil
+ })
+
+ fresh, err := b.Restore(context.Background(), registry.Route{Host: "restored.example.com", Kind: "instance"})
+ if err != nil {
+ t.Fatalf("Restore: %v", err)
+ }
+ if fresh {
+ t.Fatal("a restored instance reported fresh — a seed here would overwrite recovered data")
+ }
+}
+
+// DB absent AND no replica in the bucket: the one true first boot.
+func TestExecBackendRestoreNoReplicaIsFresh(t *testing.T) {
+ b := testExecBackend(t)
+ swapRestoreReplica(t, func(context.Context, string, string, string, string) (bool, error) {
+ return false, nil
+ })
+
+ fresh, err := b.Restore(context.Background(), registry.Route{Host: "brand-new.example.com", Kind: "instance"})
+ if err != nil {
+ t.Fatalf("Restore: %v", err)
+ }
+ if !fresh {
+ t.Fatal("a never-replicated host did not report fresh — it would never be seeded")
+ }
+}
+
+// A restore ERROR is not a first boot: "we couldn't tell" must never be
+// reported as "there was nothing there".
+func TestExecBackendRestoreErrorIsNotFresh(t *testing.T) {
+ b := testExecBackend(t)
+ swapRestoreReplica(t, func(context.Context, string, string, string, string) (bool, error) {
+ return false, errors.New("s3: connection refused")
+ })
+
+ fresh, err := b.Restore(context.Background(), registry.Route{Host: "unreachable.example.com", Kind: "instance"})
+ if err == nil {
+ t.Fatal("Restore swallowed the replica error")
+ }
+ if fresh {
+ t.Fatal("a failed restore reported fresh")
+ }
+}
+
+// swapRestoreReplica points the replica-restore seam at a stub for one test.
+func swapRestoreReplica(t *testing.T, fn func(ctx context.Context, dbPath, bucket, prefix, region string) (bool, error)) {
+ t.Helper()
+ prev := restoreReplica
+ restoreReplica = fn
+ t.Cleanup(func() { restoreReplica = prev })
+}
+
+// The seed contract, end to end through a real exec: the app's declared argv
+// followed by --db <path>, under the instance's own merged env layers, with
+// the agent's CARLOS_* vars scrubbed and the purge/control vars (Start's, not
+// the seed's) absent.
+func TestExecBackendSeedRunsDeclaredArgvWithDBPath(t *testing.T) {
+ b := testExecBackend(t)
+ b.Cache = edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
+ host := "seeded.example.com"
+ argvPath, envPath := stageRecordingLiveFullEnv(t, b, host, "v1")
+
+ cfgEnv := "SHARED=bucket\nBUCKET_ONLY=1\n"
+ if err := os.WriteFile(filepath.Join(b.DataDir, host+".config.env"), []byte(cfgEnv), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ ownEnv := "SHARED=instance\nCARLOS_ADMIN_TOKEN=real\n"
+ if err := os.WriteFile(filepath.Join(b.DataDir, host+".env"), []byte(ownEnv), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ t.Setenv("CARLOS_HIBERNATE_BUCKET", "agent-only")
+
+ rt := registry.Route{Host: host, Kind: "instance", Seed: `["seed","--fixtures"]`}
+ if err := b.Seed(context.Background(), rt); err != nil {
+ t.Fatalf("Seed: %v", err)
+ }
+
+ gotArgv := strings.Fields(waitForFile(t, argvPath, time.Second))
+ wantArgv := []string{"seed", "--fixtures", "--db", b.defaultDBPath(host)}
+ if !slices.Equal(gotArgv, wantArgv) {
+ t.Fatalf("seed argv = %v, want %v", gotArgv, wantArgv)
+ }
+
+ child := map[string]string{}
+ for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") {
+ if k, v, ok := strings.Cut(line, "="); ok {
+ child[k] = v
+ }
+ }
+ if got := child["SHARED"]; got != "instance" {
+ t.Fatalf("seed SHARED = %q, want %q — the instance's own env must win over bucket config", got, "instance")
+ }
+ if got := child["BUCKET_ONLY"]; got != "1" {
+ t.Fatalf("seed BUCKET_ONLY = %q, want %q — bucket config must flow through", got, "1")
+ }
+ if got := child["CARLOS_ADMIN_TOKEN"]; got != "real" {
+ t.Fatalf("seed CARLOS_ADMIN_TOKEN = %q, want %q — the seed gets the instance's own env file", got, "real")
+ }
+ if _, ok := child["CARLOS_HIBERNATE_BUCKET"]; ok {
+ t.Fatal("the agent's own CARLOS_* config leaked into the seed process")
+ }
+ if _, ok := child["CARLOS_CONTROL_SOCKET"]; ok {
+ t.Fatal("seed was handed a control socket — the purge channel belongs to the serving process, not a one-shot seed")
+ }
+ if _, ok := child["CARLOS_PURGE"]; ok {
+ t.Fatal("seed was told a purge channel exists — that layering belongs to Start")
+ }
+}
+
+// The route's own DB path wins over the <DataDir>/<host>.db default, exactly
+// as it does for the spawn.
+func TestExecBackendSeedHonoursRouteDBPath(t *testing.T) {
+ b := testExecBackend(t)
+ host := "custom-db.example.com"
+ argvPath, _ := stageRecordingLiveFullEnv(t, b, host, "v1")
+ dbPath := filepath.Join(t.TempDir(), "elsewhere.db")
+
+ if err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", DBPath: dbPath, Seed: `["seed"]`}); err != nil {
+ t.Fatalf("Seed: %v", err)
+ }
+ gotArgv := strings.Fields(waitForFile(t, argvPath, time.Second))
+ wantArgv := []string{"seed", "--db", dbPath}
+ if !slices.Equal(gotArgv, wantArgv) {
+ t.Fatalf("seed argv = %v, want %v", gotArgv, wantArgv)
+ }
+}
+
+// A seed that exits non-zero fails, naming the host, the exit state, and what
+// it printed — the wake dies with it (see doWake), so this error is the whole
+// diagnosis an operator gets.
+func TestExecBackendSeedNonZeroExitFails(t *testing.T) {
+ b := testExecBackend(t)
+ host := "badseed.example.com"
+ stageSeedScript(t, b, host, "v1", "echo 'schema is corrupt' >&2\nexit 1\n")
+
+ err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `["seed"]`})
+ if err == nil {
+ t.Fatal("a seed exiting 1 was reported as success")
+ }
+ for _, want := range []string{host, "exit status 1", "schema is corrupt"} {
+ if !strings.Contains(err.Error(), want) {
+ t.Fatalf("seed error = %q, want it to contain %q", err, want)
+ }
+ }
+}
+
+// Unparseable seed argv is an ERROR, never a silent skip: a route whose seed
+// column got mangled must fail loudly at the first wake rather than boot an
+// empty instance that looks fine.
+func TestExecBackendSeedRejectsMalformedArgv(t *testing.T) {
+ b := testExecBackend(t)
+ host := "malformed.example.com"
+ argvPath, _ := stageRecordingLiveFullEnv(t, b, host, "v1")
+
+ err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `seed --fixtures`})
+ if err == nil {
+ t.Fatal("malformed seed argv was silently skipped")
+ }
+ if !strings.Contains(err.Error(), host) {
+ t.Fatalf("seed parse error = %q, want it to name the host", err)
+ }
+ if _, statErr := os.Stat(argvPath); statErr == nil {
+ t.Fatal("the binary was execed despite unparseable seed argv")
+ }
+}
+
+// An empty argv array is a no-op, not an exec of the bare binary (defensive:
+// doWake only calls Seed for a non-empty route seed).
+func TestExecBackendSeedEmptyArgvRunsNothing(t *testing.T) {
+ b := testExecBackend(t)
+ host := "empty.example.com"
+ argvPath, _ := stageRecordingLiveFullEnv(t, b, host, "v1")
+
+ if err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `[]`}); err != nil {
+ t.Fatalf("empty seed argv: %v", err)
+ }
+ if _, statErr := os.Stat(argvPath); statErr == nil {
+ t.Fatal("an empty seed argv execed the binary")
+ }
+}
+
+// A seed that never returns is bounded by seedTimeout — a wedged seed must
+// fail the wake, not hold the lease until the wake's own 5m budget expires.
+func TestExecBackendSeedIsTimeBounded(t *testing.T) {
+ b := testExecBackend(t)
+ host := "hangs.example.com"
+ stageSeedScript(t, b, host, "v1", "sleep 30\n")
+
+ prevTimeout, prevGrace := seedTimeout, seedDrainGrace
+ seedTimeout, seedDrainGrace = 200*time.Millisecond, 200*time.Millisecond
+ t.Cleanup(func() { seedTimeout, seedDrainGrace = prevTimeout, prevGrace })
+
+ start := time.Now()
+ err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `["seed"]`})
+ if err == nil {
+ t.Fatal("a hanging seed was reported as success")
+ }
+ if !strings.Contains(err.Error(), "timed out") {
+ t.Fatalf("seed error = %q, want it to say the seed timed out", err)
+ }
+ // The bound that matters: the script sleeps 30s, and a seed still running
+ // when Seed returns would be holding the lease behind a wake that already
+ // failed.
+ if elapsed := time.Since(start); elapsed > 5*time.Second {
+ t.Fatalf("seed took %s — the timeout did not bound it", elapsed)
+ }
+}
+
+// A missing live binary is the same loud failure the spawn path gives: the
+// seed goes through ensureLiveLink, so a host with nothing staged never
+// silently "seeds".
+func TestExecBackendSeedRequiresLiveBinary(t *testing.T) {
+ b := testExecBackend(t)
+ err := b.Seed(context.Background(), registry.Route{Host: "unstaged.example.com", Kind: "instance", Seed: `["seed"]`})
+ if err == nil {
+ t.Fatal("seed with no live binary reported success")
+ }
+ if !strings.Contains(err.Error(), "unstaged.example.com") {
+ t.Fatalf("seed error = %q, want it to name the host", err)
+ }
+}
+
+// A crashed seed takes its half-built database with it. Otherwise the next
+// wake stats a file that exists, calls it a warm cache, reports NOT fresh, and
+// boots the instance on a half-seeded database — the exact outcome failing the
+// wake was meant to prevent.
+func TestExecBackendSeedFailureRemovesHalfBuiltDB(t *testing.T) {
+ b := testExecBackend(t)
+ host := "crashed.example.com"
+ // argv is: seed --db <path> → $3 is the database path.
+ stageSeedScript(t, b, host, "v1", "printf half > \"$3\"\necho 'died mid-schema' >&2\nexit 1\n")
+
+ 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 {
+ t.Fatal("a failed seed left its half-built database behind — the next wake would treat it as a warm cache")
+ }
+}
+
+// The cleanup is guarded: a database that already existed when Seed was called
+// is never ours to delete (the wake path can't produce this — Restore reported
+// fresh — but a direct call must not be able to destroy live data).
+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 {
+ 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))
+ if err != nil || string(got) != "live data" {
+ t.Fatalf("pre-existing database = %q, %v; want it untouched by a failed seed", got, err)
+ }
+}
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 16ade10d1..29d931837 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -16,6 +16,7 @@ import (
"os/exec"
"path/filepath"
"strings"
+ "sync"
"syscall"
"time"
@@ -121,15 +122,15 @@ 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) error {
+func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, error) {
host, dbPath := rt.Host, rt.DBPath
if dbPath == "" {
dbPath = b.defaultDBPath(host)
}
if _, err := os.Stat(dbPath); err == nil {
- return nil // warm cache — see doWake
+ return false, nil // warm cache — see doWake
} else if !os.IsNotExist(err) {
- return err
+ return false, err
}
// The DB is gone, so any leftover litestream sidecar state (meta dirs,
// WAL/SHM) describes a database that no longer exists — litestream's own
@@ -144,21 +145,153 @@ func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) error {
_ = os.RemoveAll(dbPath + "-litestream")
_ = os.RemoveAll(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream"))
- restored, err := lease.Restore(ctx, dbPath, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region)
+ restored, err := restoreReplica(ctx, dbPath, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region)
if err != nil {
- return err
+ // "We couldn't tell" is never reported as "there was nothing there":
+ // fresh stays false on every error path, so a bucket outage can't
+ // route a live instance into the seed.
+ return false, err
}
if restored {
log.Printf("activator: restored %s from s3://%s/%s", host, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host))
} else {
// Never replicated: a brand-new instance boots fresh and creates its
// own DB — the register-only create path (provisioning as a metadata
- // write) landing on its first wake.
+ // write) landing on its first wake. This — DB absent AND no replica —
+ // is the ONLY fresh boot, and the one moment the route's seed runs.
log.Printf("activator: no replica for %s in s3://%s/%s — booting fresh", host, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host))
}
+ return !restored, nil
+}
+
+// restoreReplica is lease.Restore behind a package var so the fresh-detection
+// matrix (replica present / absent / unreachable) is testable without S3.
+var restoreReplica = lease.Restore
+
+// Seed runs the route's first-boot initialization, once, before the instance
+// has ever been spawned (doWake calls it only when Restore reported fresh).
+//
+// THE CONTRACT, FOR APP AUTHORS: the platform invokes your binary with your
+// manifest's `seed` argv followed by `--db <path>` — you create your schema
+// and starter content in that database and exit 0. (`titogo seed` is the shape
+// this was named for.) It runs under the same merged env layers the instance
+// itself gets, so the credentials and config in your env file are there; it
+// does NOT get the purge-channel vars, which describe a serving process's
+// cache and mean nothing to a one-shot init. A non-zero exit fails the wake:
+// a half-seeded instance must not serve.
+func (b *ExecBackend) Seed(ctx context.Context, rt registry.Route) error {
+ host := rt.Host
+ var seedArgs []string
+ if err := json.Unmarshal([]byte(rt.Seed), &seedArgs); err != nil {
+ // Never a silent skip: a mangled seed column must fail the wake
+ // loudly, not boot an empty instance that looks healthy.
+ return fmt.Errorf("parse seed argv for %s (%q): %w", host, rt.Seed, err)
+ }
+ if len(seedArgs) == 0 {
+ return nil // defensive: doWake only calls us for a non-empty seed
+ }
+ if err := b.ensureLiveLink(ctx, host); err != nil {
+ return err
+ }
+ fileEnv, err := b.instanceEnv(host)
+ if err != nil {
+ return err
+ }
+ dbPath := rt.DBPath
+ if dbPath == "" {
+ dbPath = b.defaultDBPath(host)
+ }
+
+ // Whether the database existed BEFORE the seed ran decides whether a
+ // failed seed may clean up after itself (below). On the wake path it never
+ // does — Restore reported fresh — so the normal answer is false.
+ _, statErr := os.Stat(dbPath)
+ dbPreexisting := statErr == nil
+
+ ctx, cancel := context.WithTimeout(ctx, seedTimeout)
+ defer cancel()
+ argv := append(seedArgs, "--db", dbPath)
+ cmd := exec.CommandContext(ctx, b.liveLink(host), argv...)
+ cmd.Env = childEnv(fileEnv)
+ // Like the instance child, the seed logs through the agent's own streams
+ // (journald under systemd) — and its stderr is ALSO teed into a bounded
+ // buffer, because the wake error this returns is the whole diagnosis an
+ // operator gets for a host that refuses to come up.
+ var stderr capBuffer
+ cmd.Stdout = os.Stdout
+ cmd.Stderr = io.MultiWriter(os.Stderr, &stderr)
+ // WaitDelay is what makes seedTimeout real. Because Stderr is a plain
+ // io.Writer (the tee), os/exec pipes it and Run waits for that pipe to
+ // close — and a seed that spawned its own grandchild holds the write end
+ // open long after the timeout killed the seed itself. Without this, a
+ // wedged seed would hold the lease until the wake's own 5m budget expired.
+ cmd.WaitDelay = seedDrainGrace
+ log.Printf("activator: seeding %s: %s %s", host, b.liveLink(host), strings.Join(argv, " "))
+ if err := cmd.Run(); err != nil {
+ if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+ err = fmt.Errorf("timed out after %s and was killed: %w", seedTimeout, err)
+ }
+ // Take the half-built database with the failure. Failing the wake is
+ // only half the promise: the file a crashed seed left behind would
+ // look like a warm cache to the NEXT wake — DB present, so not fresh,
+ // so no seed — and the instance would boot on it, which is the
+ // half-seeded instance this whole path exists to prevent. Only a
+ // database that wasn't there when we started is ours to remove, and
+ // under the lease nothing else can have written it.
+ if !dbPreexisting {
+ for _, p := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} {
+ _ = os.Remove(p)
+ }
+ }
+ if tail := strings.TrimSpace(stderr.String()); tail != "" {
+ return fmt.Errorf("seeding %s: %w: %s", host, err, tail)
+ }
+ return fmt.Errorf("seeding %s: %w", host, err)
+ }
+ log.Printf("activator: seeded %s", host)
return nil
}
+// seedTimeout bounds a first-boot seed run — a wedged seed must fail the wake
+// (which releases the lease) rather than sit on it. seedDrainGrace is how long
+// Run then waits for the killed seed's output pipes to close before giving up
+// on them (cmd.WaitDelay). Both vars so tests can shrink them.
+var (
+ seedTimeout = 2 * time.Minute
+ seedDrainGrace = 10 * time.Second
+)
+
+// capBuffer keeps at most capBufferMax bytes of what's written to it and
+// discards the rest, so quoting a child's stderr in an error can never turn a
+// chatty seed into an unbounded string. Locked because it is read after
+// cmd.Run returns while os/exec's own copy goroutine may still be writing: on
+// a WaitDelay expiry (the wedged-seed path) Run returns without waiting for
+// that copy to finish.
+type capBuffer struct {
+ mu sync.Mutex
+ b []byte
+}
+
+const capBufferMax = 4 << 10
+
+func (c *capBuffer) Write(p []byte) (int, error) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ if room := capBufferMax - len(c.b); room > 0 {
+ if len(p) < room {
+ room = len(p)
+ }
+ c.b = append(c.b, p[:room]...)
+ }
+ return len(p), nil
+}
+
+func (c *capBuffer) String() string {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return string(c.b)
+}
+
// NextDue asks the (still-running) instance for its earliest pending scheduled
// delivery over its socket. The endpoint is admin-gated like every instance
// data endpoint, so present the instance's own admin token — read from the
@@ -230,18 +363,10 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
return nil, "", err
}
- // Two env layers, both optional like the unit's "-" prefix: the
- // sync-owned bucket config underneath, the instance's own file on
- // top (it wins conflicts — see mergeEnv).
- cfgEnv, err := readEnvFile(filepath.Join(b.DataDir, host+".config.env"))
- 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"))
- if err != nil && !os.IsNotExist(err) {
- return nil, "", fmt.Errorf("read env file: %w", err)
+ fileEnv, err := b.instanceEnv(host)
+ if err != nil {
+ return nil, "", err
}
- fileEnv := mergeEnv(cfgEnv, ownEnv)
// The control socket reaches the child through its ENVIRONMENT, never as a
// CLI flag: every app binary built before this feature existed parses its
@@ -335,6 +460,26 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
return p, spawnVer, nil
}
+// instanceEnv composes the host's two EnvironmentFile layers, both optional
+// like a unit's "-" prefix: the sync-owned bucket config (<host>.config.env)
+// underneath, the instance's own file (<host>.env) on top — it wins conflicts,
+// see mergeEnv. Shared by Start and Seed precisely so a seed sees the SAME
+// environment the instance will: a seed that read different credentials, or a
+// different database URL, than the process it initializes would be a trap.
+// (Start layers its agent-owned purge vars on top of this; those belong to a
+// serving process, not to a one-shot init.)
+func (b *ExecBackend) instanceEnv(host string) (map[string]string, error) {
+ cfgEnv, err := readEnvFile(filepath.Join(b.DataDir, host+".config.env"))
+ 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"))
+ if err != nil && !os.IsNotExist(err) {
+ return nil, fmt.Errorf("read env file: %w", err)
+ }
+ return mergeEnv(cfgEnv, ownEnv), nil
+}
+
// childEnv approximates a systemd unit's environment from inside the agent:
// inherit the process env but strip every CARLOS_* var first, so the agent's
// own config (hibernation, registry, tokens) can never leak into a child;
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go
index a9b9a6c37..55caeda87 100644
--- a/internal/activator/backend_unit.go
+++ b/internal/activator/backend_unit.go
@@ -31,6 +31,7 @@
// recommended but stays the operator's unchecked call, and a route
// added without it is silently replicated by nobody. See
// ops.FleetDBs's own doc comment for the mechanics.
+//
// - There is no lease to trade: the unit exists on exactly one box (its
// state dir is that box's disk) and the box replicator never hands
// ownership off, so wake/sleep is not a writer handoff. When unit
@@ -108,10 +109,19 @@ func (b *UnitBackend) Acquire(context.Context, registry.Route) (Lease, error) {
return unitLease{}, nil
}
-// Restore is a no-op: the tenant's state dir persists on the box, and
-// disaster recovery is a litestream restore run by the operator as root —
-// the agent can't touch a DynamicUser's 0700 state dir either way.
-func (b *UnitBackend) Restore(context.Context, registry.Route) error { return nil }
+// Restore is a no-op, and NEVER fresh: the tenant's state dir persists on the
+// box and the agent can't even read it (a DynamicUser's 0700 dir), so it
+// cannot know whether a database exists in there — and "I can't tell" must
+// read as "not a first boot", because the alternative is running an app's
+// seed over its live data. Disaster recovery here is a litestream restore run
+// by the operator as root. A unit tenant that wants first-boot initialization
+// does it in its own unit (ExecStartPre), where the state dir is readable.
+func (b *UnitBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil }
+
+// Seed is unreachable for unit tenants (Restore never reports fresh) and is a
+// no-op for the same reason it has to be: the agent can neither read the
+// tenant's state dir nor run a process as its DynamicUser.
+func (b *UnitBackend) Seed(context.Context, registry.Route) error { return nil }
// Start brings the unit up and holds until its upstream answers — the
// health gate that makes the edge-mediated wake strictly better than bare
@@ -270,9 +280,12 @@ func (b *SplitBackend) pick(rt registry.Route) Backend {
func (b *SplitBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, error) {
return b.pick(rt).Acquire(ctx, rt)
}
-func (b *SplitBackend) Restore(ctx context.Context, rt registry.Route) error {
+func (b *SplitBackend) Restore(ctx context.Context, rt registry.Route) (bool, error) {
return b.pick(rt).Restore(ctx, rt)
}
+func (b *SplitBackend) Seed(ctx context.Context, rt registry.Route) error {
+ return b.pick(rt).Seed(ctx, rt)
+}
func (b *SplitBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
return b.pick(rt).Start(ctx, rt)
}
diff --git a/internal/activator/backend_unit_test.go b/internal/activator/backend_unit_test.go
index b985d7f4c..89494ab29 100644
--- a/internal/activator/backend_unit_test.go
+++ b/internal/activator/backend_unit_test.go
@@ -155,6 +155,38 @@ func TestSplitBackendExecRouteBypassesUnit(t *testing.T) {
}
}
+// Restore and Seed dispatch by Backing like every other SplitBackend method —
+// and a unit tenant, whose Restore is never fresh, never reaches a seed even
+// with one declared on its route.
+func TestSplitBackendDispatchesFreshAndSeedByBacking(t *testing.T) {
+ eb := &fakeBackend{restoreFresh: true}
+ ub := &UnitBackend{Systemctl: (&fakeSystemctl{}).run, OptDir: t.TempDir()}
+ sb := &SplitBackend{Exec: eb, Unit: ub}
+ ctx := context.Background()
+
+ if fresh, err := sb.Restore(ctx, testRoute("plain.example.com")); err != nil || !fresh {
+ t.Fatalf("exec route Restore = %v, %v; want true, nil", fresh, err)
+ }
+ unitRoute := testUnitRoute("kass.example.com")
+ unitRoute.Seed = `["seed"]`
+ if fresh, err := sb.Restore(ctx, unitRoute); err != nil || fresh {
+ t.Fatalf("unit route Restore = %v, %v; want false, nil", fresh, err)
+ }
+
+ if err := sb.Seed(ctx, testRoute("plain.example.com")); err != nil {
+ t.Fatalf("exec route Seed: %v", err)
+ }
+ if got := eb.seedCount(); got != 1 {
+ t.Fatalf("exec-side seeds = %d, want 1", got)
+ }
+ if err := sb.Seed(ctx, unitRoute); err != nil {
+ t.Fatalf("unit route Seed: %v", err)
+ }
+ if got := eb.seedCount(); got != 1 {
+ t.Fatalf("a unit tenant's seed leaked into the exec backend (seeds=%d)", got)
+ }
+}
+
// A unit that starts but never turns healthy fails the wake cleanly: the
// unit is put back down (no half-up tenant lingering behind a 503) and no
// entry is left, so the next request retries from scratch.
@@ -252,8 +284,14 @@ func TestUnitBackendInertSeams(t *testing.T) {
if err := lease.Release(ctx); err != nil {
t.Fatalf("Release: %v", err)
}
- if err := b.Restore(ctx, rt); err != nil {
- t.Fatalf("Restore: %v", err)
+ // Never fresh: the agent cannot see inside a DynamicUser's state dir, so
+ // it must never claim a unit tenant is on its first boot — that claim is
+ // what would run an app's seed over live data.
+ if fresh, err := b.Restore(ctx, rt); err != nil || fresh {
+ t.Fatalf("Restore = %v, %v; want false, nil", fresh, err)
+ }
+ if err := b.Seed(ctx, rt); err != nil {
+ t.Fatalf("Seed: %v", err)
}
due, err := b.NextDue(ctx, rt)
if err != nil || due != 0 {