amadan

carlos / platform Public

activator: unit-backed hibernation — carlos-app@ tenants sleep and wake

UnitBackend drives whole-app tenants (Backing=unit) through the same
state machine as exec children: systemctl start behind a health gate on
wake (the edge holds the request), systemctl stop at the idle window; a
failed boot puts the unit back down and 503s cleanly. SplitBackend
composes exec + unit behind the one Backend seam by rt.Backing.

Lease/replication seams are inert for units by design: the agent (user
carlos) can't read a DynamicUser's 0700 state dir, so replication stays
with root carlos-litestream, which streams continuously — sleep is
durable before the stop ever runs, and a single box's unit has no
writer handoff to lease. Crash supervision stays with systemd
(Restart=always); Exited() is deliberately fork-free.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 4f2c257d1e6ec3c8e2b82fc5ef031056f502f764 parent d9026f4
3 files changed, +706 −1
  • internal/activator/activator_test.go +18 −1
  • internal/activator/backend_unit.go +294 −0
  • internal/activator/backend_unit_test.go +394 −0
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index e7b05d609..35a46dbc5 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -374,7 +374,12 @@ func (rt *testRouter) reload() error {
if hib {
managed[t.Host] = true
}
- m[t.Host] = testUpstream{proxy: testUnixProxy(t.Socket), route: t, hib: hib}
+ // Socket wins over Addr, mirroring edge.Reload's upstream choice.
+ proxy := testUnixProxy(t.Socket)
+ if t.Socket == "" && t.Addr != "" {
+ proxy = testTCPProxy(t.Addr)
+ }
+ m[t.Host] = testUpstream{proxy: proxy, route: t, hib: hib}
}
rt.mu.Lock()
rt.up = m
@@ -419,6 +424,18 @@ func testUnixProxy(socket string) http.Handler {
}
}
+// testTCPProxy mirrors edge's tcpProxy: the upstream shape for whole-app
+// tenants on a TCP Addr (the unix-socket fallback).
+func testTCPProxy(addr string) http.Handler {
+ return &httputil.ReverseProxy{
+ Rewrite: func(pr *httputil.ProxyRequest) {
+ pr.Out.URL.Scheme = "http"
+ pr.Out.URL.Host = addr
+ pr.Out.Host = pr.In.Host
+ },
+ }
+}
+
// proxyGet sends GET base+"/" with an overridden Host header and returns the
// response body + status code.
func proxyGet(t *testing.T, base, host string) (string, int) {
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go
new file mode 100644
index 000000000..79fc48490
--- /dev/null
+++ b/internal/activator/backend_unit.go
@@ -0,0 +1,294 @@
+// Unit-backed tenants: hibernation for whole-app instances whose process
+// manager stays systemd (carlos-app@<host>.service — DynamicUser,
+// StateDirectory) rather than the agent. The activator still owns WHEN the
+// process runs — stopped after the idle window, started when a request
+// arrives, the edge holding that request while the unit boots and its
+// upstream turns healthy — but the HOW is systemctl, behind the Systemctl
+// seam so the state-machine tests need no systemd.
+//
+// The lease/replication seams are satisfied by inert implementations, on
+// purpose — this is a deliberate division of labor, not a shortcut:
+//
+// - A unit tenant's data lives in its DynamicUser-owned state dir
+// (/var/lib/carlos-app/<host>, 0700) which the agent, as user carlos,
+// cannot read — so the agent can neither restore it nor run litestream
+// over it. Its replication belongs to the box-wide carlos-litestream
+// unit (root), which streams changes continuously whether the tenant is
+// awake or asleep. That IS the snapshot-on-sleep story here: a stop only
+// happens after a full idle window, so streaming replication shipped the
+// final writes long before the unit goes down. ops.FleetDBs keeps
+// unit-backed hibernating routes IN the generated config for the same
+// reason — only exec-backed hibernating hosts, whose replication the
+// activator's embedded litestream owns under the S3 lease, are excluded.
+// - 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
+// tenants go multi-box, a real lease slots in through this same seam.
+//
+// Pairing with carlos-app@.socket (see internal/host/units): with the
+// tenant's socket unit enabled, systemd holds the listener bound across
+// stops, so the edge's unix dial never hits the socket-absent window — and a
+// connection while the service is stopped starts it (wake-on-connect). The
+// activator's wake path here remains the primary one because it is
+// health-gated: a broken boot becomes a clean 503, not a connect that hangs
+// in the listen backlog.
+package activator
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "net/http"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// UnitName is the systemd unit backing a unit-backed tenant's route —
+// carlos-app@<host>.service, the same instance name the adopt pass
+// try-restarts for non-hibernating tenants.
+func UnitName(host string) string { return "carlos-app@" + host + ".service" }
+
+// UnitBackend is the Backend for routes with Backing == registry.BackingUnit.
+type UnitBackend struct {
+ // Systemctl runs one systemctl verb against a unit — the Proc-style seam
+ // wrapping systemctl that keeps hibernation tests systemd-free. Nil = the
+ // real systemctl binary.
+ Systemctl func(ctx context.Context, verb, unit string) error
+ // OptDir is the on-box install root (the live/<host> symlinks) — same
+ // layout as ExecBackend's, read for spawnVersion so build convergence
+ // cycles awake unit tenants exactly like exec children.
+ OptDir string
+ // WaitHealthy blocks until the tenant answers HTTP on its upstream
+ // (rt.Socket, else rt.Addr). Nil = the package's socket/TCP prober with
+ // the standard spawnHealthTimeout. A seam so unit tests need no listener.
+ WaitHealthy func(ctx context.Context, rt registry.Route) error
+}
+
+var _ Backend = (*UnitBackend)(nil)
+
+func (b *UnitBackend) systemctl(ctx context.Context, verb, unit string) error {
+ if b.Systemctl != nil {
+ return b.Systemctl(ctx, verb, unit)
+ }
+ out, err := exec.CommandContext(ctx, "systemctl", verb, unit).CombinedOutput()
+ if err != nil {
+ return fmt.Errorf("systemctl %s %s: %v: %s", verb, unit, err, strings.TrimSpace(string(out)))
+ }
+ return nil
+}
+
+// LiveVersion mirrors ExecBackend.LiveVersion — the unit's ExecStart is the
+// same live/<host> symlink the adopt pass repoints, so convergence reads the
+// same layout.
+func (b *UnitBackend) LiveVersion(host string) (string, error) {
+ target, err := os.Readlink(filepath.Join(b.OptDir, "live", host))
+ if err != nil {
+ if os.IsNotExist(err) {
+ return "", nil
+ }
+ return "", err
+ }
+ return filepath.Base(filepath.Dir(target)), nil
+}
+
+// Acquire returns the inert lease (see the package comment: replication
+// ownership stays with the root carlos-litestream unit, and a single box's
+// unit has no writer handoff to guard).
+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 }
+
+// Start brings the unit up and holds until its upstream answers — the
+// health gate that makes the edge-mediated wake strictly better than bare
+// socket activation (a broken boot becomes a clean 503, not a connect
+// parked in the listen backlog). A start against an already-running unit is
+// systemd's no-op, which is what lets a fresh agent re-adopt tenants left
+// awake by its previous incarnation. No live-symlink precheck: if the
+// binary is missing the unit's exec fails, the health wait times out, and
+// the stop below puts the unit back down.
+func (b *UnitBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
+ unit := UnitName(rt.Host)
+ // Read the symlink before the start, same fail-safe direction as
+ // ExecBackend: a concurrent repoint at worst records the older version
+ // (one extra convergence cycle), never the newer (a stale child the
+ // sweep would think converged).
+ spawnVer, _ := b.LiveVersion(rt.Host)
+ if err := b.systemctl(ctx, "start", unit); err != nil {
+ return nil, "", err
+ }
+ if err := b.waitHealthy(ctx, rt); err != nil {
+ stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
+ defer cancel()
+ _ = b.systemctl(stopCtx, "stop", unit)
+ return nil, "", err
+ }
+ return &unitProc{b: b, unit: unit}, spawnVer, nil
+}
+
+func (b *UnitBackend) waitHealthy(ctx context.Context, rt registry.Route) error {
+ if b.WaitHealthy != nil {
+ return b.WaitHealthy(ctx, rt)
+ }
+ switch {
+ case rt.Socket != "":
+ return waitHealthyCtx(ctx, rt.Socket, spawnHealthTimeout)
+ case rt.Addr != "":
+ return waitHealthyAddrCtx(ctx, rt.Addr, spawnHealthTimeout)
+ }
+ return fmt.Errorf("route %s has neither socket nor addr to health-check", rt.Host)
+}
+
+// NextDue reports "nothing scheduled": the due-time probe is admin-gated by
+// the instance's CARLOS_ADMIN_TOKEN, and a unit tenant's env file
+// (/etc/carlos/apps/<host>.env) is root-0600 — unreadable to the agent by
+// design. Unit tenants therefore sleep until a request; the due-time index
+// stays exec-only until the contract grows an agent-readable channel.
+func (b *UnitBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil }
+
+// HasSidecar: sidecar co-spawn is exec-child machinery (the sidecar is the
+// agent's own child, CARLOS_*-scrubbed); a unit tenant that wants a sidecar
+// runs it inside its own unit.
+func (b *UnitBackend) HasSidecar(registry.Route) bool { return false }
+
+func (b *UnitBackend) StartSidecar(context.Context, registry.Route) (Proc, error) {
+ return nil, errors.New("unit-backed tenants have no agent-managed sidecars")
+}
+
+// unitProc adapts the unit to the Proc seam.
+type unitProc struct {
+ b *UnitBackend
+ unit string
+
+ mu sync.Mutex
+ stopped bool
+}
+
+// Stop is `systemctl stop` — systemd runs the unit's own SIGTERM drain and
+// the ctx bounds how long we wait for the job.
+func (p *unitProc) Stop(ctx context.Context) error {
+ err := p.b.systemctl(ctx, "stop", p.unit)
+ p.mu.Lock()
+ p.stopped = true
+ p.mu.Unlock()
+ return err
+}
+
+// Exited reports false until Stop: crash supervision belongs to systemd
+// (Restart=always on the unit), so a crashed tenant restarts under its unit
+// without the activator noticing — deliberately, because Exited is polled
+// under a.mu on every request and sweep tick, and an exec'd `systemctl
+// is-active` there would put a fork on the hot path. The trade-off: a unit
+// wedged past its start rate limit looks awake until its idle window stops
+// it (the edge's dials 502 meanwhile) — operator territory, exactly as it
+// is for a non-hibernating unit today.
+func (p *unitProc) Exited() bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.stopped
+}
+
+// unitLease / unitRep: the inert lease (rationale in the package comment).
+type unitLease struct{}
+
+func (unitLease) Replicate(context.Context, string, time.Duration) (Rep, error) {
+ return unitRep{}, nil
+}
+func (unitLease) Release(context.Context) error { return nil }
+
+type unitRep struct{}
+
+func (unitRep) Err() error { return nil }
+func (unitRep) Close(context.Context) error { return nil }
+
+// waitHealthyAddr is waitHealthy for TCP upstreams (rt.Addr — unit tenants
+// that can't accept a unix listener yet; see the carlos-app@.socket unit's
+// fallback note).
+func waitHealthyAddr(addr string, timeout time.Duration) error {
+ deadline := time.Now().Add(timeout)
+ client := &http.Client{Timeout: 3 * time.Second}
+ var last error
+ for time.Now().Before(deadline) {
+ resp, err := client.Get("http://" + addr + "/")
+ if err == nil {
+ resp.Body.Close()
+ if resp.StatusCode < 500 {
+ return nil
+ }
+ last = fmt.Errorf("unhealthy: HTTP %d", resp.StatusCode)
+ } else {
+ last = err
+ }
+ time.Sleep(150 * time.Millisecond)
+ }
+ return fmt.Errorf("instance not healthy within %s: %v", timeout, last)
+}
+
+func waitHealthyAddrCtx(ctx context.Context, addr string, timeout time.Duration) error {
+ done := make(chan error, 1)
+ go func() { done <- waitHealthyAddr(addr, timeout) }()
+ select {
+ case err := <-done:
+ return err
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+}
+
+// SplitBackend composes the two production backends behind the one Backend
+// seam the state machine sees: each call dispatches on the route's Backing,
+// so exec children and unit tenants share the singleflight wakes, idle
+// clock, due index, and convergence sweep without forking any of it.
+type SplitBackend struct {
+ Exec Backend // rt.Backing == "" — agent-spawned children (ExecBackend)
+ Unit Backend // rt.Backing == registry.BackingUnit (UnitBackend)
+}
+
+var _ Backend = (*SplitBackend)(nil)
+
+func (b *SplitBackend) pick(rt registry.Route) Backend {
+ if rt.Backing == registry.BackingUnit {
+ return b.Unit
+ }
+ return b.Exec
+}
+
+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 {
+ return b.pick(rt).Restore(ctx, rt)
+}
+func (b *SplitBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
+ return b.pick(rt).Start(ctx, rt)
+}
+func (b *SplitBackend) NextDue(ctx context.Context, rt registry.Route) (int64, error) {
+ return b.pick(rt).NextDue(ctx, rt)
+}
+func (b *SplitBackend) HasSidecar(rt registry.Route) bool { return b.pick(rt).HasSidecar(rt) }
+func (b *SplitBackend) StartSidecar(ctx context.Context, rt registry.Route) (Proc, error) {
+ return b.pick(rt).StartSidecar(ctx, rt)
+}
+
+// LiveVersion forwards to the exec side's reader when it has one — both
+// backings share the live/<host> symlink layout, so one reader serves the
+// convergence sweep for every host. (New picks this up via the optional
+// interface, exactly as it does for a bare ExecBackend.)
+func (b *SplitBackend) LiveVersion(host string) (string, error) {
+ if lv, ok := b.Exec.(interface {
+ LiveVersion(host string) (string, error)
+ }); ok {
+ return lv.LiveVersion(host)
+ }
+ return "", nil
+}
diff --git a/internal/activator/backend_unit_test.go b/internal/activator/backend_unit_test.go
new file mode 100644
index 000000000..027417e8b
--- /dev/null
+++ b/internal/activator/backend_unit_test.go
@@ -0,0 +1,394 @@
+package activator
+
+// Unit-backed tenant tests: the same state machine (activator.go) driving a
+// carlos-app@ unit through the Systemctl seam instead of an exec'd child —
+// no systemd, no AWS. The end-to-end test at the bottom proves the edge
+// shape: a request for a sleeping unit tenant is HELD while the "unit"
+// starts and turns healthy, then forwarded.
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// fakeSystemctl records verbs and optionally runs hooks — the Proc-style
+// seam's test double. Hooks let the end-to-end test bind/close a real
+// listener exactly when the "unit" would.
+type fakeSystemctl struct {
+ mu sync.Mutex
+ calls []string
+ startErr error
+ onStart func() // runs on "start", outside the lock
+ onStop func() // runs on "stop", outside the lock
+}
+
+func (f *fakeSystemctl) run(_ context.Context, verb, unit string) error {
+ f.mu.Lock()
+ f.calls = append(f.calls, verb+" "+unit)
+ err := f.startErr
+ onStart, onStop := f.onStart, f.onStop
+ f.mu.Unlock()
+ if verb == "start" {
+ if err != nil {
+ return err
+ }
+ if onStart != nil {
+ onStart()
+ }
+ }
+ if verb == "stop" && onStop != nil {
+ onStop()
+ }
+ return nil
+}
+
+func (f *fakeSystemctl) recorded() []string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return append([]string(nil), f.calls...)
+}
+
+// testUnitRoute is testRoute's unit-backed sibling: a whole-app tenant
+// (carlos-app@<host>) flagged for hibernation.
+func testUnitRoute(host string) registry.Route {
+ return registry.Route{Host: host, Kind: "instance", Socket: "/run/carlos/apps/" + host + ".sock",
+ Hibernate: true, Backing: registry.BackingUnit}
+}
+
+// testUnitActivator wires a Service over a SplitBackend: the usual
+// fakeBackend on the exec side, a UnitBackend with the fake systemctl and an
+// instant health gate on the unit side.
+func testUnitActivator(t *testing.T, idle time.Duration) (*Service, *fakeBackend, *fakeSystemctl, *fakeClock, *registry.Registry) {
+ t.Helper()
+ eb := &fakeBackend{}
+ sc := &fakeSystemctl{}
+ ub := &UnitBackend{
+ Systemctl: sc.run,
+ OptDir: t.TempDir(),
+ WaitHealthy: func(context.Context, registry.Route) error { return nil },
+ }
+ clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatalf("open registry: %v", err)
+ }
+ a := New(Config{Bucket: "test-bucket", Idle: idle}, &SplitBackend{Exec: eb, Unit: ub}, reg)
+ a.now = clk.now
+ return a, eb, sc, clk, reg
+}
+
+// The whole-app hibernation cycle: a request starts the unit (systemctl
+// start, health-gated); the idle window stops it (systemctl stop); the next
+// request starts it again. The exec side is never touched.
+func TestUnitTenantWakeIdleSleepCycle(t *testing.T) {
+ a, eb, sc, clk, _ := testUnitActivator(t, 10*time.Second)
+ route := testUnitRoute("kass.example.com")
+ unit := UnitName(route.Host)
+
+ release, err := a.beginRequest(context.Background(), route)
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+ if got := sc.recorded(); len(got) != 1 || got[0] != "start "+unit {
+ t.Fatalf("wake systemctl calls = %v, want [start %s]", got, unit)
+ }
+
+ // Not yet idle: the unit stays up.
+ clk.advance(9 * time.Second)
+ a.sweepOnce()
+ if a.entryCount() != 1 {
+ t.Fatal("unit tenant hibernated before its idle window elapsed")
+ }
+
+ // Idle: stopped via systemctl, entry forgotten.
+ clk.advance(2 * time.Second)
+ a.sweepOnce()
+ waitFor(t, "unit hibernation", func() bool { return a.entryCount() == 0 })
+ if got := sc.recorded(); len(got) != 2 || got[1] != "stop "+unit {
+ t.Fatalf("sleep systemctl calls = %v, want [... stop %s]", got, unit)
+ }
+
+ // Re-wake: a fresh start.
+ release, err = a.beginRequest(context.Background(), route)
+ if err != nil {
+ t.Fatalf("beginRequest after sleep: %v", err)
+ }
+ release()
+ if got := sc.recorded(); len(got) != 3 || got[2] != "start "+unit {
+ t.Fatalf("re-wake systemctl calls = %v, want a second start", got)
+ }
+
+ // The exec backend was never involved: no lease claims, no spawns.
+ if acquires, restores, starts := eb.counts(); acquires != 0 || restores != 0 || starts != 0 {
+ t.Fatalf("unit tenant leaked into the exec backend: acquires=%d restores=%d starts=%d", acquires, restores, starts)
+ }
+}
+
+// SplitBackend dispatch, the other direction: an exec-backed route wakes
+// through the fake exec backend and never touches systemctl.
+func TestSplitBackendExecRouteBypassesUnit(t *testing.T) {
+ a, eb, sc, _, _ := testUnitActivator(t, time.Minute)
+
+ release, err := a.beginRequest(context.Background(), testRoute("plain.example.com"))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+ if acquires, restores, starts := eb.counts(); acquires != 1 || restores != 1 || starts != 1 {
+ t.Fatalf("exec route skipped the exec backend: acquires=%d restores=%d starts=%d", acquires, restores, starts)
+ }
+ if got := sc.recorded(); len(got) != 0 {
+ t.Fatalf("exec route touched systemctl: %v", 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.
+func TestUnitTenantFailedHealthStopsUnit(t *testing.T) {
+ route := testUnitRoute("broken.example.com")
+ unit := UnitName(route.Host)
+
+ sc := &fakeSystemctl{}
+ ub := &UnitBackend{Systemctl: sc.run, OptDir: t.TempDir(),
+ WaitHealthy: func(context.Context, registry.Route) error { return errors.New("never healthy") }}
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatalf("open registry: %v", err)
+ }
+ a := New(Config{Bucket: "test-bucket", Idle: time.Minute}, &SplitBackend{Exec: &fakeBackend{}, Unit: ub}, reg)
+
+ if _, err := a.beginRequest(context.Background(), route); err == nil {
+ t.Fatal("wake succeeded despite the health gate failing")
+ }
+ if got := sc.recorded(); len(got) != 2 || got[0] != "start "+unit || got[1] != "stop "+unit {
+ t.Fatalf("systemctl calls = %v, want [start %s, stop %s]", got, unit, unit)
+ }
+ if a.entryCount() != 0 {
+ t.Fatal("failed unit wake left an entry behind")
+ }
+}
+
+// A start that fails outright (unit file missing, systemd unreachable)
+// fails the wake with no stop attempt and no entry.
+func TestUnitTenantStartFailureFailsWake(t *testing.T) {
+ a, _, sc, _, _ := testUnitActivator(t, time.Minute)
+ sc.startErr = errors.New("Unit carlos-app@gone.example.com.service not found")
+
+ if _, err := a.beginRequest(context.Background(), testUnitRoute("gone.example.com")); err == nil {
+ t.Fatal("wake succeeded despite systemctl start failing")
+ }
+ if got := sc.recorded(); len(got) != 1 {
+ t.Fatalf("systemctl calls = %v, want just the failed start", got)
+ }
+ if a.entryCount() != 0 {
+ t.Fatal("failed start left an entry behind")
+ }
+}
+
+// Shutdown puts unit tenants down like exec children — deliberately: with
+// the tenant's carlos-app@.socket unit enabled, systemd's wake-on-connect
+// brings it back on the next connection even while the agent is away, and
+// the next agent's health-gated wake re-adopts it (start on a running unit
+// is a no-op).
+func TestUnitTenantShutdownStopsUnit(t *testing.T) {
+ a, _, sc, _, _ := testUnitActivator(t, time.Hour)
+ route := testUnitRoute("kass.example.com")
+
+ release, err := a.beginRequest(context.Background(), route)
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ a.Shutdown(ctx)
+ if a.entryCount() != 0 {
+ t.Fatal("shutdown left the unit entry behind")
+ }
+ got := sc.recorded()
+ if len(got) != 2 || got[1] != "stop "+UnitName(route.Host) {
+ t.Fatalf("systemctl calls = %v, want the shutdown stop", got)
+ }
+}
+
+// The inert lease/replication seams: everything the state machine calls on
+// them must succeed and report healthy, so the machine treats a unit tenant
+// exactly like a replicating exec child. (Replication ownership really
+// lives with the root carlos-litestream unit — see backend_unit.go.)
+func TestUnitBackendInertSeams(t *testing.T) {
+ b := &UnitBackend{Systemctl: (&fakeSystemctl{}).run, OptDir: t.TempDir()}
+ ctx := context.Background()
+ rt := testUnitRoute("inert.example.com")
+
+ lease, err := b.Acquire(ctx, rt)
+ if err != nil {
+ t.Fatalf("Acquire: %v", err)
+ }
+ rep, err := lease.Replicate(ctx, rt.DBPath, 0)
+ if err != nil {
+ t.Fatalf("Replicate: %v", err)
+ }
+ if err := rep.Err(); err != nil {
+ t.Fatalf("Rep.Err: %v", err)
+ }
+ if err := rep.Close(ctx); err != nil {
+ t.Fatalf("Rep.Close: %v", err)
+ }
+ 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)
+ }
+ due, err := b.NextDue(ctx, rt)
+ if err != nil || due != 0 {
+ t.Fatalf("NextDue = %d, %v; want 0, nil (unit tenants sleep until a request)", due, err)
+ }
+ if b.HasSidecar(rt) {
+ t.Fatal("unit tenants must not report agent-managed sidecars")
+ }
+}
+
+// LiveVersion reads the same live/<host> symlink layout as ExecBackend, so
+// build convergence cycles awake unit tenants onto shipped builds.
+func TestUnitBackendLiveVersion(t *testing.T) {
+ opt := t.TempDir()
+ b := &UnitBackend{Systemctl: (&fakeSystemctl{}).run, OptDir: opt}
+
+ if v, err := b.LiveVersion("new.example.com"); err != nil || v != "" {
+ t.Fatalf("LiveVersion(no symlink) = %q, %v; want empty, nil", v, err)
+ }
+ verDir := filepath.Join(opt, "versions", "abc1234")
+ if err := os.MkdirAll(verDir, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.MkdirAll(filepath.Join(opt, "live"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(filepath.Join(verDir, "app"), filepath.Join(opt, "live", "kass.example.com")); err != nil {
+ t.Fatal(err)
+ }
+ if v, err := b.LiveVersion("kass.example.com"); err != nil || v != "abc1234" {
+ t.Fatalf("LiveVersion = %q, %v; want abc1234", v, err)
+ }
+}
+
+// The end-to-end shape the feature promises: a request for a SLEEPING unit
+// tenant is held at the edge while the unit starts and its upstream turns
+// healthy (the real TCP prober — no injected health seam here), then
+// forwarded; after the idle window the unit is stopped for real (listener
+// closed); the next request repeats the cycle. TCP Addr keeps the test
+// sandbox-safe; the unix-socket path differs only in the dialer.
+func TestUnitTenantWakeServesHeldRequest(t *testing.T) {
+ // Pre-pick the tenant's port so the route can exist before the "unit" runs.
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatalf("pick port: %v", err)
+ }
+ addr := ln.Addr().String()
+ ln.Close()
+
+ // The "unit": the fake systemctl binds/closes a real HTTP listener on
+ // start/stop, with a deliberate boot delay so the request provably waits
+ // on the health gate.
+ var srvMu sync.Mutex
+ var srv *http.Server
+ sc := &fakeSystemctl{}
+ sc.onStart = func() {
+ time.Sleep(200 * time.Millisecond) // boot time — the edge must hold the request
+ l, err := net.Listen("tcp", addr)
+ if err != nil {
+ t.Errorf("unit listen: %v", err)
+ return
+ }
+ s := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ io.WriteString(w, "kass up")
+ })}
+ srvMu.Lock()
+ srv = s
+ srvMu.Unlock()
+ go s.Serve(l)
+ }
+ sc.onStop = func() {
+ srvMu.Lock()
+ s := srv
+ srv = nil
+ srvMu.Unlock()
+ if s != nil {
+ s.Close()
+ }
+ }
+
+ old := spawnHealthTimeout
+ spawnHealthTimeout = 5 * time.Second
+ t.Cleanup(func() { spawnHealthTimeout = old })
+
+ eb := &fakeBackend{}
+ ub := &UnitBackend{Systemctl: sc.run, OptDir: t.TempDir()} // nil WaitHealthy: the real prober
+ clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatalf("open registry: %v", err)
+ }
+ a := New(Config{Bucket: "test-bucket", Idle: 10 * time.Second}, &SplitBackend{Exec: eb, Unit: ub}, reg)
+ a.now = clk.now
+
+ route := registry.Route{Host: "kass.example.com", Kind: "instance", Addr: addr,
+ Hibernate: true, Backing: registry.BackingUnit}
+ if err := reg.Upsert(route); err != nil {
+ t.Fatal(err)
+ }
+ rt, err := newTestRouter(reg, a)
+ if err != nil {
+ t.Fatalf("new router: %v", err)
+ }
+ front := httptest.NewServer(rt)
+ t.Cleanup(front.Close)
+ t.Cleanup(func() { sc.onStop() }) // in case the test dies mid-cycle
+
+ // Asleep → the request wakes the unit and is answered by it.
+ if body, code := proxyGet(t, front.URL, "kass.example.com"); code != 200 || body != "kass up" {
+ t.Fatalf("held request = %d %q, want 200 kass up", code, body)
+ }
+
+ // Idle out → the unit is stopped for real: its listener is gone.
+ clk.advance(11 * time.Second)
+ a.sweepOnce()
+ waitFor(t, "unit hibernation", func() bool { return a.entryCount() == 0 })
+ waitFor(t, "listener closed", func() bool {
+ conn, err := net.DialTimeout("tcp", addr, 100*time.Millisecond)
+ if err == nil {
+ conn.Close()
+ return false
+ }
+ return true
+ })
+
+ // The next request wakes it again.
+ if body, code := proxyGet(t, front.URL, "kass.example.com"); code != 200 || body != "kass up" {
+ t.Fatalf("post-sleep request = %d %q, want 200 kass up", code, body)
+ }
+ want := []string{
+ "start " + UnitName(route.Host),
+ "stop " + UnitName(route.Host),
+ "start " + UnitName(route.Host),
+ }
+ if got := sc.recorded(); fmt.Sprint(got) != fmt.Sprint(want) {
+ t.Fatalf("systemctl calls = %v, want %v", got, want)
+ }
+}