carlos / platform Public

activator: unit-backed tenants get a purge listener too

Unit-backed tenants (carlos-app@ systemd units, backend_unit.go) had no
purge path: the edge cached their routes same as any other, but nothing
could ever invalidate an entry early — staleness was bounded only by
the 5-minute TTL ceiling. Mirrors ExecBackend.Start's wiring: a Cache
field (nil = disabled, same convention), a purge listener at
rt.Socket + ".control" started right before the unit comes up and
closed on Stop, and the same purge-on-spawn host cache drop ExecBackend
does (a unit start might be new code, same as a fresh exec'd child).
Only wired when the route carries a Socket — an Addr-only tenant has no
socket-adjacent path to hang a control socket off of.

The harder part the task flagged as murky: getting
CARLOS_CONTROL_SOCKET into the unit's actual environment, since a unit
tenant is never exec'd by this process — there's no cmd.Env to hand it
through directly, only what systemd reads from EnvironmentFile
directives at start time. Investigated whether a clean path exists
without touching appconfig (root-run, owns /etc/carlos/apps/<host>.env)
or any root-owned file: bootstrap.go's tmpfilesConf already makes
/run/carlos/apps carlos-owned (0750) — the same directory the
carlos-app@.socket unit binds each tenant's public socket under, so the
agent can already write there with no privilege change. UnitBackend.Start
now stages a second env file there (<ControlEnvDir>/<host>.env,
CARLOS_CONTROL_SOCKET=<path> only) before calling systemctl start, and
carlos-app@.service gets a second, LATER EnvironmentFile=- line reading
it — later so this platform-owned value always wins over anything an
app's own config might set under the same name. Both EnvironmentFile
lines keep the "-" missing-file-is-fine prefix, so a box whose installed
unit predates this line, or a tenant with no Cache wired, starts exactly
as before.

Caveat worth stating plainly: this requires `carlos bootstrap` to
re-run on already-provisioned boxes before the second EnvironmentFile
line takes effect (systemd only reads the unit file installed on disk),
same as any other unit-template change in this repo — not a new kind of
rollout step.

Tests mirror activator_test.go's ExecBackend Cache/control-socket/purge
suite: control env written and purge listener started (and in that
order relative to systemctl start, since systemd reads its
EnvironmentFile at start time), nil Cache skips both, an Addr-only
route skips both, Stop closes the listener, and starting purges the
host's stale cache entries.
Paul Campbell 7b6584d1d7b659bb1034d86a315048c684225a69 parent 1b16157
4 files changed, +365 −5
  • internal/activator/backend_unit.go +107 −4
  • internal/activator/backend_unit_test.go +236 −0
  • internal/host/agent.go +9 −1
  • internal/host/units/carlos-app@.service +13 −0
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go
index a9b9a6c37..b0c341979 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
@@ -49,12 +50,16 @@ import (
"context"
"errors"
"fmt"
+ "log"
"net/http"
+ "os"
"os/exec"
+ "path/filepath"
"strings"
"sync"
"time"
+ "github.com/carlosframework/platform/internal/edge"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
@@ -78,6 +83,20 @@ type UnitBackend struct {
// (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
+ // Cache, when non-nil, gets a purge listener started for each started
+ // unit tenant (at rt.Socket + ".control", same convention
+ // ExecBackend.Start uses) so it can invalidate its own cached responses
+ // right after a write. Nil disables this — no purge listener, no error —
+ // matching ExecBackend.Cache's convention exactly. Only wired when the
+ // route also carries a Socket: an Addr-only tenant has no socket-adjacent
+ // path to hang a control socket off of.
+ Cache *edge.Cache
+ // ControlEnvDir is where writeControlEnv stages a started unit tenant's
+ // control-socket env snippet — /run/carlos/apps in production (see
+ // writeControlEnv's doc comment for why that directory and not
+ // /etc/carlos/apps), a temp dir in tests. Required whenever Cache is
+ // non-nil; unused otherwise.
+ ControlEnvDir string
}
var _ Backend = (*UnitBackend)(nil)
@@ -128,16 +147,94 @@ func (b *UnitBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
// (one extra convergence cycle), never the newer (a stale child the
// sweep would think converged).
spawnVer, _ := b.LiveVersion(rt.Host)
+
+ // Wire the purge channel BEFORE systemctl start: systemd reads a unit's
+ // EnvironmentFile directives at start time, so the control env snippet
+ // must already be on disk, and the listener must already be bound, before
+ // the tenant can possibly try to dial it. See writeControlEnv's doc
+ // comment for why this exists as a second, agent-writable file rather
+ // than going through appconfig.
+ var pl *edge.PurgeListener
+ if b.Cache != nil && rt.Socket != "" {
+ // A unit coming up is the same kind of moment ExecBackend.Start's
+ // spawn is: the code behind the host may have changed while it was
+ // stopped, so whatever's cached for it is suspect. Drop it before the
+ // unit serves anything.
+ if n := b.Cache.PurgeHost(rt.Host); n > 0 {
+ log.Printf("activator: purged %d cached response(s) for %s on unit start", n, rt.Host)
+ }
+ controlSocket := rt.Socket + ".control"
+ if err := b.writeControlEnv(rt.Host, controlSocket); err != nil {
+ // Not fatal to the start — the unit still starts and serves
+ // requests via its ordinary appconfig env, it just won't learn
+ // the control socket until a start that manages to write this
+ // file. A cache that never hears a purge still self-heals via
+ // the TTL ceiling, same fallback ExecBackend.Start leans on for a
+ // failed ListenPurge.
+ log.Printf("activator: control env for %s: %v (continuing without purge wiring)", rt.Host, err)
+ } else if p, err := edge.ListenPurge(controlSocket, rt.Host, b.Cache); err != nil {
+ log.Printf("activator: purge listener for %s: %v (continuing without it)", rt.Host, err)
+ } else {
+ pl = p
+ }
+ }
+
if err := b.systemctl(ctx, "start", unit); err != nil {
+ if pl != nil {
+ pl.Close()
+ }
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)
+ if pl != nil {
+ pl.Close()
+ }
return nil, "", err
}
- return &unitProc{b: b, unit: unit}, spawnVer, nil
+ return &unitProc{b: b, unit: unit, purge: pl}, spawnVer, nil
+}
+
+// writeControlEnv stages host's control-socket env snippet at
+// <ControlEnvDir>/<host>.env — the single line CARLOS_CONTROL_SOCKET=
+// controlSocket. It exists because a unit tenant, unlike an ExecBackend
+// child, is never exec'd by this process: there is no cmd.Env to hand the
+// variable through directly, only whatever systemd reads out of the unit's
+// EnvironmentFile directives at start time.
+//
+// This deliberately does NOT go through the appconfig/adopt machinery
+// (internal/appconfig) that owns /etc/carlos/apps/<host>.env: that path is
+// root-run (see RunAdopt's doc comment — "the ONLY place config sync may
+// run, because it writes /etc/carlos/apps and restarts units"), and
+// entangling this platform-internal wiring with the operator-facing secrets
+// system would mean either running the agent as root too or teaching adopt
+// about a concern that isn't app configuration at all. Instead this writes a
+// SECOND, independent file under ControlEnvDir (/run/carlos/apps in
+// production), which bootstrap.go's tmpfilesConf already makes carlos-owned
+// (0750) — the same directory the carlos-app@.socket unit binds each
+// tenant's public socket under, so the agent (running as that same carlos
+// user) can already write there with no privilege change.
+//
+// For this to actually reach the unit, carlos-app@.service needs a second,
+// LATER EnvironmentFile=-/run/carlos/apps/%i.env line (after the existing
+// appconfig one, so this platform-owned value always wins if an app's own
+// env ever collided on the same key — it has no legitimate reason to). The
+// leading "-" on both EnvironmentFile lines means a missing file is not a
+// unit failure, so this is a harmless no-op on any box whose installed unit
+// predates that line (Cache would also be nil there in that case, since
+// nothing exists yet to hand a working control socket to).
+func (b *UnitBackend) writeControlEnv(host, controlSocket string) error {
+ if err := os.MkdirAll(b.ControlEnvDir, 0o750); err != nil {
+ return err
+ }
+ path := filepath.Join(b.ControlEnvDir, host+".env")
+ tmp := path + ".tmp"
+ if err := os.WriteFile(tmp, []byte("CARLOS_CONTROL_SOCKET="+controlSocket+"\n"), 0o640); err != nil {
+ return err
+ }
+ return os.Rename(tmp, path)
}
func (b *UnitBackend) waitHealthy(ctx context.Context, rt registry.Route) error {
@@ -171,16 +268,22 @@ func (b *UnitBackend) StartSidecar(context.Context, registry.Route) (Proc, error
// unitProc adapts the unit to the Proc seam.
type unitProc struct {
- b *UnitBackend
- unit string
+ b *UnitBackend
+ unit string
+ purge *edge.PurgeListener // nil if b.Cache was nil, rt.Socket was empty, or ListenPurge failed
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.
+// the ctx bounds how long we wait for the job — plus closing the purge
+// listener, mirroring childProc.Stop's purge.Close() for exec children, so a
+// stopped unit tenant leaves no stray listening socket behind.
func (p *unitProc) Stop(ctx context.Context) error {
+ if p.purge != nil {
+ p.purge.Close()
+ }
err := p.b.systemctl(ctx, "stop", p.unit)
p.mu.Lock()
p.stopped = true
diff --git a/internal/activator/backend_unit_test.go b/internal/activator/backend_unit_test.go
index b985d7f4c..942f8c84a 100644
--- a/internal/activator/backend_unit_test.go
+++ b/internal/activator/backend_unit_test.go
@@ -16,10 +16,12 @@ import (
"net/http/httptest"
"os"
"path/filepath"
+ "strings"
"sync"
"testing"
"time"
+ "github.com/carlosframework/platform/internal/edge"
"github.com/carlosframework/platform/internal/registry"
)
@@ -400,3 +402,237 @@ func TestUnitTenantWakeServesHeldRequest(t *testing.T) {
t.Fatalf("systemctl calls = %v, want %v", got, want)
}
}
+
+// shortSockDir returns a fresh temp directory NOT nested under t.Name() —
+// unlike t.TempDir(), whose path embeds the full (sometimes long) test
+// function name. A unix socket path is capped at ~108 bytes on Linux
+// (sun_path); appending "<host>.sock.control" to a t.TempDir() path under
+// one of this file's longer test names overflows that limit and bind fails
+// with EINVAL. Only needed for paths a socket actually binds to.
+func shortSockDir(t *testing.T) string {
+ t.Helper()
+ dir, err := os.MkdirTemp("", "carlos-unit-")
+ if err != nil {
+ t.Fatalf("MkdirTemp: %v", err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ return dir
+}
+
+// Unit tenants and the cache-before-wake purge channel: mirrors
+// ExecBackend.Start's Cache wiring (activator_test.go's
+// TestExecBackendStartPassesControlSocketInEnvAndStartsPurgeListener and
+// siblings), except the child isn't exec'd so there's no environment to hand
+// a var through directly — the control socket instead reaches the unit via
+// a second EnvironmentFile the agent (running as the unprivileged carlos
+// user) writes to ControlEnvDir before starting the unit, which the unit
+// template's own second EnvironmentFile=- line then reads. See
+// backend_unit.go's writeControlEnv doc comment for why this particular
+// channel doesn't entangle with the root-owned appconfig machinery.
+
+// A Cache-armed UnitBackend writes the control env file, starts a purge
+// listener at rt.Socket + ".control" (the same convention ExecBackend
+// uses), and only THEN starts the unit — the env file must already exist
+// when systemd reads it at start time.
+func TestUnitBackendStartWritesControlEnvAndStartsPurgeListener(t *testing.T) {
+ cache := edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
+ runDir := t.TempDir()
+ sc := &fakeSystemctl{}
+ b := &UnitBackend{
+ Systemctl: sc.run,
+ OptDir: t.TempDir(),
+ ControlEnvDir: runDir,
+ Cache: cache,
+ WaitHealthy: func(context.Context, registry.Route) error { return nil },
+ }
+ rt := registry.Route{Host: "unit-cache.example.com", Kind: "instance",
+ Socket: filepath.Join(shortSockDir(t), "unit-cache.example.com.sock")}
+ wantControlSocket := rt.Socket + ".control"
+
+ proc, _, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("Start: %v", err)
+ }
+ t.Cleanup(func() { proc.Stop(context.Background()) })
+
+ envContents, err := os.ReadFile(filepath.Join(runDir, rt.Host+".env"))
+ if err != nil {
+ t.Fatalf("read control env file: %v", err)
+ }
+ if got := strings.TrimSpace(string(envContents)); got != "CARLOS_CONTROL_SOCKET="+wantControlSocket {
+ t.Fatalf("control env file = %q, want CARLOS_CONTROL_SOCKET=%s", got, wantControlSocket)
+ }
+
+ if _, err := os.Stat(wantControlSocket); err != nil {
+ t.Fatalf("expected purge listener socket %s to exist: %v", wantControlSocket, err)
+ }
+
+ if got := sc.recorded(); len(got) != 1 || got[0] != "start "+UnitName(rt.Host) {
+ t.Fatalf("systemctl calls = %v, want just [start %s]", got, UnitName(rt.Host))
+ }
+}
+
+// The control env file must exist BEFORE systemctl start runs: systemd
+// reads a unit's EnvironmentFile at start time, so writing it after the
+// unit is already starting would race a real boot.
+func TestUnitBackendStartWritesControlEnvBeforeSystemctlStart(t *testing.T) {
+ cache := edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
+ runDir := t.TempDir()
+ host := "order.example.com"
+ envPath := filepath.Join(runDir, host+".env")
+
+ var sawEnvAtStart bool
+ sc := &fakeSystemctl{}
+ sc.onStart = func() {
+ _, err := os.Stat(envPath)
+ sawEnvAtStart = err == nil
+ }
+ b := &UnitBackend{
+ Systemctl: sc.run,
+ OptDir: t.TempDir(),
+ ControlEnvDir: runDir,
+ Cache: cache,
+ WaitHealthy: func(context.Context, registry.Route) error { return nil },
+ }
+ rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(shortSockDir(t), "order.sock")}
+
+ proc, _, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("Start: %v", err)
+ }
+ t.Cleanup(func() { proc.Stop(context.Background()) })
+
+ if !sawEnvAtStart {
+ t.Fatal("control env file must exist before systemctl start runs — the unit reads its EnvironmentFile at start time")
+ }
+}
+
+// A nil UnitBackend.Cache disables the whole channel — no control env file,
+// no purge listener — matching ExecBackend's nil-Cache convention exactly.
+func TestUnitBackendStartWithNilCacheSkipsControlEnvAndPurgeListener(t *testing.T) {
+ runDir := t.TempDir()
+ sc := &fakeSystemctl{}
+ b := &UnitBackend{
+ Systemctl: sc.run,
+ OptDir: t.TempDir(),
+ ControlEnvDir: runDir, // Cache left nil
+ WaitHealthy: func(context.Context, registry.Route) error { return nil },
+ }
+ rt := registry.Route{Host: "no-cache.example.com", Kind: "instance",
+ Socket: filepath.Join(t.TempDir(), "no-cache.sock")}
+
+ proc, _, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("Start: %v", err)
+ }
+ t.Cleanup(func() { proc.Stop(context.Background()) })
+
+ if _, err := os.Stat(filepath.Join(runDir, rt.Host+".env")); err == nil {
+ t.Fatal("expected no control env file when UnitBackend.Cache is nil")
+ }
+ if _, err := os.Stat(rt.Socket + ".control"); err == nil {
+ t.Fatal("expected no purge listener socket when UnitBackend.Cache is nil")
+ }
+}
+
+// An Addr-only unit route (no unix socket at all) has no rt.Socket to build
+// a ".control" path from, so the purge channel is skipped rather than
+// wired to a bogus path — the same "no meaningful socket-adjacent path"
+// gap TCP-upstream tenants already have elsewhere in this package.
+func TestUnitBackendStartWithAddrOnlyRouteSkipsPurgeListener(t *testing.T) {
+ cache := edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
+ runDir := t.TempDir()
+ sc := &fakeSystemctl{}
+ b := &UnitBackend{
+ Systemctl: sc.run,
+ OptDir: t.TempDir(),
+ ControlEnvDir: runDir,
+ Cache: cache,
+ WaitHealthy: func(context.Context, registry.Route) error { return nil },
+ }
+ rt := registry.Route{Host: "addr-only.example.com", Kind: "instance", Addr: "127.0.0.1:1"} // no Socket
+
+ proc, _, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("Start: %v", err)
+ }
+ t.Cleanup(func() { proc.Stop(context.Background()) })
+
+ if _, err := os.Stat(filepath.Join(runDir, rt.Host+".env")); err == nil {
+ t.Fatal("expected no control env file for an Addr-only route (no socket to build a control path from)")
+ }
+}
+
+// Stop closes the purge listener — mirroring childProc.Stop's purge.Close()
+// — so a stopped unit tenant leaves no stray listening socket behind.
+func TestUnitBackendStopClosesPurgeListener(t *testing.T) {
+ cache := edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
+ runDir := t.TempDir()
+ sc := &fakeSystemctl{}
+ b := &UnitBackend{
+ Systemctl: sc.run,
+ OptDir: t.TempDir(),
+ ControlEnvDir: runDir,
+ Cache: cache,
+ WaitHealthy: func(context.Context, registry.Route) error { return nil },
+ }
+ rt := registry.Route{Host: "stop-closes.example.com", Kind: "instance",
+ Socket: filepath.Join(shortSockDir(t), "stop.sock")}
+
+ proc, _, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("Start: %v", err)
+ }
+ if _, err := os.Stat(rt.Socket + ".control"); err != nil {
+ t.Fatalf("expected purge listener socket to exist after Start: %v", err)
+ }
+
+ if err := proc.Stop(context.Background()); err != nil {
+ t.Fatalf("Stop: %v", err)
+ }
+ if _, err := os.Stat(rt.Socket + ".control"); err == nil {
+ t.Fatal("expected the purge listener socket removed after Stop")
+ }
+}
+
+// Starting a unit tenant is the same kind of moment ExecBackend.Start's spawn
+// is: whatever's cached for this host might be stale (the unit may have come
+// back up on new code), so Start drops the host's cached entries before the
+// unit is asked to run — mirrors
+// TestExecBackendStartPurgesHostCacheOnSpawn.
+func TestUnitBackendStartPurgesHostCacheOnSpawn(t *testing.T) {
+ cache := edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
+ runDir := t.TempDir()
+ sc := &fakeSystemctl{}
+ b := &UnitBackend{
+ Systemctl: sc.run,
+ OptDir: t.TempDir(),
+ ControlEnvDir: runDir,
+ Cache: cache,
+ WaitHealthy: func(context.Context, registry.Route) error { return nil },
+ }
+ rt := registry.Route{Host: "unit-deploy.example.com", Kind: "instance",
+ Socket: filepath.Join(shortSockDir(t), "unit-deploy.sock")}
+
+ stale := edge.CacheKey{Host: rt.Host, Method: "GET", Path: "/event"}
+ otherHost := edge.CacheKey{Host: "bystander.example.com", Method: "GET", Path: "/event"}
+ for _, k := range []edge.CacheKey{stale, otherHost} {
+ cache.Put(k, "", &edge.CachedResponse{
+ StatusCode: 200, Header: http.Header{}, Body: []byte("old html"),
+ StoredAt: time.Now(), TTL: time.Minute,
+ })
+ }
+
+ proc, _, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("Start: %v", err)
+ }
+ t.Cleanup(func() { proc.Stop(context.Background()) })
+
+ if _, ok := cache.Get(stale); ok {
+ t.Fatal("starting a unit tenant must drop that host's cached responses")
+ }
+ if _, ok := cache.Get(otherHost); !ok {
+ t.Fatal("another host's cached responses must survive a unit start")
+ }
+}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 07a1d6f55..78605e609 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -135,7 +135,15 @@ func RunAgent(args []string) {
DataDir: dd,
Cache: cache,
},
- Unit: &activator.UnitBackend{OptDir: release.OptDir()},
+ Unit: &activator.UnitBackend{
+ OptDir: release.OptDir(),
+ // Same shared cache the exec side gets, and the same
+ // purge-listener convention (rt.Socket + ".control") — see
+ // UnitBackend.Start / writeControlEnv for how the control
+ // socket path actually reaches a unit tenant's process.
+ Cache: cache,
+ ControlEnvDir: "/run/carlos/apps",
+ },
}, reg) // the registry satisfies activator.DueStore directly
svc.Start()
act = svc
diff --git a/internal/host/units/carlos-app@.service b/internal/host/units/carlos-app@.service
index 2fc9dde79..612eb4444 100644
--- a/internal/host/units/carlos-app@.service
+++ b/internal/host/units/carlos-app@.service
@@ -26,6 +26,18 @@
# gated, when a request arrives. Restart=always still owns crashes.
#
# Apps needing different arguments override ExecStart via a drop-in.
+#
+# Two EnvironmentFile lines, deliberately in this order: the operator-facing
+# app config appconfig/adopt owns (root-written, /etc/carlos/apps/%i.env),
+# then the platform's own control-socket wiring
+# (/run/carlos/apps/%i.env — carlos-owned, written by the agent's
+# UnitBackend.Start right before each start; see backend_unit.go's
+# writeControlEnv). Later EnvironmentFile lines win on a key collision, so
+# this order means the platform-owned CARLOS_CONTROL_SOCKET always wins over
+# anything an app's own config might (illegitimately) set under the same
+# name. Both use the "-" no-such-file-is-fine prefix: a box whose agent
+# predates the second file, or a tenant with no Cache wired at all, starts
+# exactly as it did before either file existed.
[Unit]
Description=CARLOS app %i
@@ -36,6 +48,7 @@ DynamicUser=yes
StateDirectory=carlos-app/%i
ExecStart=/opt/carlos/live/%i serve
EnvironmentFile=-/etc/carlos/apps/%i.env
+EnvironmentFile=-/run/carlos/apps/%i.env
Restart=always
NoNewPrivileges=yes
ProtectSystem=strict