amadan

carlos / platform Public

activator: the control socket travels by environment, and a spawn clears the host's cache

C3 — Start passed --control-socket to every spawned instance unconditionally.
Go's flag package with ExitOnError (what app binaries use) exits 2 on a flag it
doesn't know, so every app binary built before this feature would refuse to
start the moment the agent upgraded — a fleet-wide outage. The path now reaches
the child as CARLOS_CONTROL_SOCKET in its environment, which an older binary
simply ignores, and only when there is actually a cache (and therefore a purge
listener) to talk to. The test now reads the child's real argv and environment
from a recording stub binary: argv must be exactly --socket/--db, and it fails
against the pre-fix code.

I5 — a redeploy kept serving the old process's HTML until its TTL ran out.
Spawning a fresh process for a host is precisely the moment its cached content
is suspect, so Start now calls Cache.PurgeHost for that host (and only that
host) before the new process takes traffic.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Paul Campbell 5847f37ec3a0406ea7cf7adff561c11f8447393b parent 3ece43f
2 files changed, +158 −21
  • internal/activator/activator_test.go +125 −19
  • internal/activator/backend_exec.go +33 −2
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 2a5db49b0..36fb7eebe 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -20,6 +20,7 @@ import (
"net/http/httputil"
"os"
"path/filepath"
+ "slices"
"strings"
"sync"
"sync/atomic"
@@ -1911,28 +1912,79 @@ func TestExecBackendHasSidecar(t *testing.T) {
}
}
-// Start passes a --control-socket flag and starts a purge listener bound to
-// it whenever ExecBackend.Cache is set — the wiring that lets a spawned
-// instance invalidate its own cached responses right after a write.
+// stageRecordingLive stages host's live binary as a shell script that dumps
+// its argv and its CARLOS_CONTROL_SOCKET to two files and exits — the only way
+// to see what the child was ACTUALLY handed, argv and environment both.
+// Returns the two paths.
+func stageRecordingLive(t *testing.T, b *ExecBackend, host, ver string) (argvPath, envPath string) {
+ t.Helper()
+ dir := t.TempDir()
+ argvPath = filepath.Join(dir, "argv")
+ envPath = filepath.Join(dir, "env")
+ script := "#!/bin/sh\n" +
+ "printf '%s\\n' \"$@\" > '" + argvPath + "'\n" +
+ "printf '%s\\n' \"${CARLOS_CONTROL_SOCKET-<unset>}\" > '" + envPath + "'\n"
+
+ 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(script), 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)
+ }
+ return argvPath, envPath
+}
+
+// waitForFile polls until path exists (the child writes it concurrently with
+// Start's health wait) and returns its contents.
+func waitForFile(t *testing.T, path string, within time.Duration) string {
+ t.Helper()
+ deadline := time.Now().Add(within)
+ for time.Now().Before(deadline) {
+ if b, err := os.ReadFile(path); err == nil {
+ return string(b)
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ t.Fatalf("%s never appeared within %s", path, within)
+ return ""
+}
+
+// Start hands the spawned instance its control socket through the ENVIRONMENT
+// (CARLOS_CONTROL_SOCKET), never as a CLI flag, and starts a purge listener
+// bound to it whenever ExecBackend.Cache is set.
+//
+// The argv assertion is the regression guard: a --control-socket flag makes
+// every app binary that predates this feature exit 2 at startup (Go's flag
+// package with ExitOnError, which is what app binaries use), i.e. a
+// fleet-wide outage the moment the agent upgrades. An unknown environment
+// variable is simply ignored by such a binary.
//
// Deviation from the task brief's literal test: the brief asserted the
// control socket file still exists AFTER Start returns, even on the
-// health-wait error stageLive's fake binary (`#!/bin/sh\n`, never binds
-// rt.Socket) is expected to produce here. That's not actually true of the
-// implementation the same brief specifies — Start's pre-existing
-// health-wait-failure path calls p.Stop() before returning, and Step 3's
-// own childProc.Stop change closes the purge listener there, which (per
-// Go's net.Listener.Close on a unix socket the listener itself created)
-// unlinks the socket file. So by the time Start returns with an error, the
-// file is already gone — confirmed empirically, not a guess. This test
-// instead polls for the socket WHILE Start's health-wait is still running
-// (i.e. before its internal Stop cleanup fires), which is the invariant
-// that's actually guaranteed and is what "Start wires up the purge
-// listener" means in practice.
-func TestExecBackendStartPassesControlSocketFlagAndStartsPurgeListener(t *testing.T) {
+// health-wait error the fake binary (which never binds rt.Socket) is expected
+// to produce here. That's not actually true of the implementation the same
+// brief specifies — Start's pre-existing health-wait-failure path calls
+// p.Stop() before returning, and Step 3's own childProc.Stop change closes
+// the purge listener there, which (per Go's net.Listener.Close on a unix
+// socket the listener itself created) unlinks the socket file. So by the time
+// Start returns with an error, the file is already gone — confirmed
+// empirically, not a guess. This test instead polls for the socket WHILE
+// Start's health-wait is still running (i.e. before its internal Stop cleanup
+// fires), which is the invariant that's actually guaranteed and is what
+// "Start wires up the purge listener" means in practice.
+func TestExecBackendStartPassesControlSocketInEnvAndStartsPurgeListener(t *testing.T) {
b := testExecBackend(t)
b.Cache = edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
- stageLive(t, b, "control-socket.example.com", "v1")
+ argvPath, envPath := stageRecordingLive(t, b, "control-socket.example.com", "v1")
rt := registry.Route{
Host: "control-socket.example.com",
@@ -1960,21 +2012,33 @@ func TestExecBackendStartPassesControlSocketFlagAndStartsPurgeListener(t *testin
}
time.Sleep(10 * time.Millisecond)
}
+
+ gotArgv := strings.Fields(waitForFile(t, argvPath, time.Second))
+ gotEnv := strings.TrimSpace(waitForFile(t, envPath, time.Second))
+
<-done
if proc != nil {
defer proc.Stop(context.Background())
}
+ wantArgv := []string{"--socket", rt.Socket, "--db", filepath.Join(b.DataDir, rt.Host+".db")}
+ if !slices.Equal(gotArgv, wantArgv) {
+ t.Fatalf("child argv = %v, want %v — a flag the app binary doesn't parse makes it exit 2 at startup", gotArgv, wantArgv)
+ }
+ if gotEnv != wantControlSocket {
+ t.Fatalf("child CARLOS_CONTROL_SOCKET = %q, want %q", gotEnv, wantControlSocket)
+ }
if !found {
t.Fatalf("expected control socket %s to exist while Start's health-wait ran", wantControlSocket)
}
}
// A nil ExecBackend.Cache disables the purge listener entirely — no control
-// socket, no error.
+// socket, no CARLOS_CONTROL_SOCKET in the child's environment (there is
+// nothing for it to talk to), no error.
func TestExecBackendStartWithNilCacheSkipsPurgeListener(t *testing.T) {
b := testExecBackend(t) // b.Cache left nil
- stageLive(t, b, "no-cache.example.com", "v1")
+ _, envPath := stageRecordingLive(t, b, "no-cache.example.com", "v1")
rt := registry.Route{
Host: "no-cache.example.com",
@@ -1992,4 +2056,46 @@ func TestExecBackendStartWithNilCacheSkipsPurgeListener(t *testing.T) {
if _, err := os.Stat(rt.Socket + ".control"); err == nil {
t.Fatal("expected no control socket when ExecBackend.Cache is nil")
}
+ if got := strings.TrimSpace(waitForFile(t, envPath, time.Second)); got != "<unset>" {
+ t.Fatalf("child CARLOS_CONTROL_SOCKET = %q, want it unset when there is no cache", got)
+ }
+}
+
+// Spawning a fresh process for a host is a deploy boundary: the code behind
+// the host just changed, so every response cached from the old process is
+// stale. Start drops them all before the new process serves anything —
+// without this, a redeploy keeps serving old HTML until the TTL expires.
+func TestExecBackendStartPurgesHostCacheOnSpawn(t *testing.T) {
+ b := testExecBackend(t)
+ cache := edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
+ b.Cache = cache
+ stageLive(t, b, "deploy.example.com", "v1")
+
+ stale := edge.CacheKey{Host: "deploy.example.com", 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,
+ })
+ }
+
+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+ defer cancel()
+ rt := registry.Route{
+ Host: "deploy.example.com",
+ Kind: "instance",
+ Socket: filepath.Join(t.TempDir(), "instance.sock"),
+ }
+ proc, _, _ := b.Start(ctx, rt) // health-wait failure is expected; the spawn still happened
+ if proc != nil {
+ defer proc.Stop(context.Background())
+ }
+
+ if _, ok := cache.Get(stale); ok {
+ t.Fatal("spawning a fresh process for a host must drop that host's cached responses")
+ }
+ if _, ok := cache.Get(otherHost); !ok {
+ t.Fatal("another host's cached responses must survive a spawn")
+ }
}
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index e6a52237f..f04447ee9 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -236,9 +236,21 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
return nil, "", fmt.Errorf("read env file: %w", err)
}
+ // The control socket reaches the child through its ENVIRONMENT, never as a
+ // CLI flag: every app binary built before this feature existed parses its
+ // flags with flag.ExitOnError, which exits 2 on an unrecognized flag — so
+ // passing --control-socket would refuse to start every app that hasn't
+ // also shipped support for it, a fleet-wide outage on agent upgrade. An
+ // environment variable a binary doesn't know about is simply ignored.
+ // Only set when there's actually a purge listener to talk to, so the
+ // child's environment stays honest about what exists.
controlSocket := rt.Socket + ".control"
- cmd := exec.Command(bin, "--socket", rt.Socket, "--db", dbPath, "--control-socket", controlSocket)
- cmd.Env = childEnv(fileEnv)
+ childVars := fileEnv
+ if b.Cache != nil {
+ childVars = withVar(fileEnv, "CARLOS_CONTROL_SOCKET", controlSocket)
+ }
+ cmd := exec.Command(bin, "--socket", rt.Socket, "--db", dbPath)
+ cmd.Env = childEnv(childVars)
// The child logs through the agent's own stderr/stdout (journald under
// systemd); there's no per-instance unit to own a log stream anymore.
cmd.Stdout = os.Stdout
@@ -259,6 +271,14 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
p := &childProc{cmd: cmd, done: make(chan struct{})}
if b.Cache != nil {
+ // A fresh process for this host is exactly the moment its cached
+ // content becomes suspect: a deploy/redeploy means new code and
+ // potentially new pages, and nothing else would invalidate them until
+ // their TTL ran out. Drop the whole host's entries before the new
+ // process serves its first request.
+ if n := b.Cache.PurgeHost(rt.Host); n > 0 {
+ log.Printf("activator: purged %d cached response(s) for %s on spawn", n, rt.Host)
+ }
pl, err := edge.ListenPurge(controlSocket, rt.Host, b.Cache)
if err != nil {
// Not fatal to the wake — the instance still serves requests, it
@@ -302,6 +322,17 @@ func childEnv(fileEnv map[string]string) []string {
return env
}
+// withVar returns a copy of env with key=value added — a copy, so the
+// caller's map (an EnvironmentFile's parsed contents) is never mutated.
+func withVar(env map[string]string, key, value string) map[string]string {
+ out := make(map[string]string, len(env)+1)
+ for k, v := range env {
+ out[k] = v
+ }
+ out[key] = value
+ return out
+}
+
// sidecarEnvPath: the host's sidecar EnvironmentFile,
// <DataDir>/<host>-sidecar.env (Eleven's <host>-bot.env, renamed).
func (b *ExecBackend) sidecarEnvPath(host string) string {