amadan

carlos / platform Public

Merge remote-tracking branch 'origin/main' into zdd-gen-socket-names

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 1be5b7e309b13d564a7a2d0f4347a8803c361029 parent 4dba454parent d2f58fd
1 file changed, +1794 −0
  • internal/host/startup_largebox_linux_test.go +1794 −0
diff --git a/internal/host/startup_largebox_linux_test.go b/internal/host/startup_largebox_linux_test.go
new file mode 100644
index 000000000..58410ee4a
--- /dev/null
+++ b/internal/host/startup_largebox_linux_test.go
@@ -0,0 +1,1794 @@
+//go:build linux
+
+package host
+
+// The large-box startup, end to end (amadan #281, ask 2). On 2026-10-13 a
+// box with 142 instances failed the 20 s update gate twice: the updater
+// restarts carlos-instance-control just before the new edge, and the
+// agent's startup sweep waited on that cold control service before the
+// edge listened. The hotfix (6dfbff1) moved the sweep, layout recovery,
+// hold restoration, Start and the key pass off the listen path, behind a
+// startup latch that instance admission waits on.
+//
+// TestStartupServesWhileSweepAndRecoveryWait proves the latch with the
+// steps stubbed. This test runs them for real on a box larger than the
+// one that failed:
+//
+// - the agent is a child process (this test binary, re-executed), so
+// what is timed is a process start up to the update gate's own probe;
+// - instance control is the real instancectl.Server over a unix socket,
+// made cold by its systemctl seam: every call waits until the test
+// warms it, then each of the box's 300 leftover units takes a moment
+// to stop, as a restarted control service does;
+// - the sweep is the real client call, layout recovery and hold
+// restoration are the real pass over real journals, hold markers and
+// files in a temp data dir, Start is the real activator's, and the
+// edge is edge.RunWithActivatorReady in its production (TLS) mode;
+// - every kind of admission the latch gates arrives during startup: an
+// instance request, a sidecar-marked request, a worker wake (from
+// runAgent's boot pass and from a direct call), and a scheduled
+// delivery through the real schedule runner and deliverer;
+// - the activator's backend is a fake that serves an app on the row's
+// socket and starts workers that report ready. It and the key pass's
+// writer are the only fakes in the agent, and neither is a startup
+// step.
+//
+// The child's startup is runAgent's legacy-box sequence: the same
+// activatorStart, with runAgent's Sweep and Recover closures. runAgent
+// itself cannot run in a test (its control socket path is a constant,
+// it builds AWS clients, and it log.Fatals), so the child assembles the
+// same pieces in the same order, and TestRunAgentStartupWiring reads
+// runAgent's source to fail if the two drift apart. Sharing one
+// orchestration function between them is a follow-up (#281). Two box
+// facts differ from production and are set on the pass: its socket
+// directory is the fixture's, not /run/carlos, and it has no
+// carlos-sockets group.
+//
+// The child reports what it does as timestamped lines in an events file;
+// the test reads them to order "a request reached the activator", "the
+// latch opened" and "the backend acquired a host" across the two
+// processes.
+
+import (
+ "bufio"
+ "context"
+ "crypto/ecdsa"
+ "crypto/elliptic"
+ "crypto/rand"
+ "crypto/tls"
+ "crypto/x509"
+ "crypto/x509/pkix"
+ "encoding/json"
+ "encoding/pem"
+ "errors"
+ "fmt"
+ "go/ast"
+ "go/parser"
+ "go/token"
+ "go/types"
+ "io"
+ "log"
+ "math/big"
+ "net"
+ "net/http"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "runtime"
+ "strconv"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "syscall"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/instancefs"
+ "github.com/carlosframework/platform/internal/layout"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/schedule"
+ "github.com/carlosframework/platform/internal/version"
+)
+
+const (
+ // largeBoxAgentEnv names the child's config file. Set, the test binary
+ // is the agent and nothing else.
+ largeBoxAgentEnv = "CARLOS_TEST_LARGEBOX_AGENT"
+
+ largeBoxRoutes = 300 // instance routes; the box that failed had 142
+ largeBoxJournals = 8 // hosts 0..7: a layout move was in flight at the restart
+ largeBoxHolds = 4 // hosts 8..11: held down after a refused move
+ largeBoxWakeHost = 12 // an ordinary sleeping instance, woken by a request
+ largeBoxDueHost = 13 // a sleeping instance with a scheduled delivery due
+ largeBoxSidecarHost = 14 // a sleeping instance its sidecar calls
+ largeBoxWorker = "worker-000.example.com"
+ largeBoxDuePath = "/jobs/tick"
+
+ // largeBoxListenBound is how soon the gate's probe must be answered
+ // after the agent process starts. The gate allows 20 s; the edge's own
+ // startup on this box takes a fraction of a second.
+ largeBoxListenBound = 3 * time.Second
+ // largeBoxCold is how long instance control stays cold in the scenario
+ // where it warms: longer than the listen bound, so the gate is answered
+ // while the sweep is provably still waiting.
+ largeBoxCold = 4 * time.Second
+ // largeBoxUnitStop is one leftover unit's stop once control is warm.
+ largeBoxUnitStop = 5 * time.Millisecond
+ // largeBoxLatchBound is the activator's wakeWaitBound, which the
+ // startup latch shares: a wake held longer is answered 503.
+ largeBoxLatchBound = 30 * time.Second
+ // largeBoxJoinBound is runAgent's join on the startup work after the
+ // edge returns, and largeBoxDrain the edge's drain of in-flight
+ // requests before it returns.
+ largeBoxJoinBound = 5 * time.Second
+ largeBoxDrain = 10 * time.Second
+ // largeBoxBindAttempts is how often the agent is started on fresh ports
+ // when one it was handed is taken before it binds (see startAgent).
+ largeBoxBindAttempts = 3
+)
+
+func largeBoxHost(i int) string { return fmt.Sprintf("app-%03d.example.com", i) }
+
+// largeBoxConfig is what the child agent is told.
+type largeBoxConfig struct {
+ DataDir, SocketDir, StoreDir, AcmeDir string
+ Control, Events, HTTP, HTTPS string
+}
+
+func TestStartupLargeBoxEndToEnd(t *testing.T) {
+ if path := os.Getenv(largeBoxAgentEnv); path != "" {
+ os.Exit(runLargeBoxAgent(path))
+ }
+ t.Run("cold instance control", func(t *testing.T) {
+ t.Parallel()
+ testLargeBoxColdControl(t)
+ })
+ t.Run("startup overruns the latch", func(t *testing.T) {
+ t.Parallel()
+ testLargeBoxOverrun(t)
+ })
+ t.Run("shutdown mid-startup", func(t *testing.T) {
+ t.Parallel()
+ testLargeBoxShutdownMidStartup(t)
+ })
+}
+
+// testLargeBoxColdControl is the 2026-10-13 shape: control restarted just
+// before the agent, and slow to answer. The edge must answer the gate's
+// probe while the sweep waits, hold every admission at the latch until
+// sweep, recovery and hold restoration are done, and then serve each host
+// only where it is safe to.
+func testLargeBoxColdControl(t *testing.T) {
+ b := newLargeBox(t)
+ a := b.startAgent(t)
+ t.Logf("gate probe answered %s after the agent started, %d routes, instance control cold", a.listen, largeBoxRoutes)
+ if _, ok := a.event("ready"); ok {
+ t.Fatal("the startup latch opened while instance control was still cold")
+ }
+
+ wake, held, moved, due, sidecar := largeBoxHost(largeBoxWakeHost), largeBoxHost(largeBoxJournals), largeBoxHost(0), largeBoxHost(largeBoxDueHost), largeBoxHost(largeBoxSidecarHost)
+ flatMoved := filepath.Join(b.socketDir, moved+".sock")
+ wakeAns, heldAns, movedAns := b.fire(wake, false), b.fire(held, false), b.fire(moved, false)
+ sidecarAns := b.fire(sidecar, true)
+ enterWake := a.waitEvent(t, 10*time.Second, "enter", wake)
+ a.waitEvent(t, 10*time.Second, "enter", held)
+ enterSidecar := a.waitEvent(t, 10*time.Second, "enter", sidecar)
+ enterMoved := a.waitEvent(t, 10*time.Second, "enter", moved)
+ if enterMoved.arg(1) != flatMoved {
+ t.Fatalf("the edge routed %s to %q before recovery, want its pre-move socket %q", moved, enterMoved.arg(1), flatMoved)
+ }
+ // The agent's own admissions: a direct worker wake and the scheduled
+ // tick, both started before the edge listened.
+ workerAsked := a.waitEvent(t, 10*time.Second, "worker-direct-begin")
+ ticked := a.waitEvent(t, 10*time.Second, "tick-done")
+ // Every admission is now waiting, and startup is not done.
+ time.Sleep(200 * time.Millisecond)
+ for host, ch := range map[string]chan largeBoxAnswer{wake: wakeAns, held: heldAns, moved: movedAns, sidecar: sidecarAns} {
+ select {
+ case got := <-ch:
+ t.Fatalf("%s answered before startup finished: %d %q %v", host, got.status, got.body, got.err)
+ default:
+ }
+ }
+ for _, kind := range []string{"acquire", "start", "worker-start", "served", "worker-direct", "worker-boot", "activator-start"} {
+ if evs := a.events(kind); len(evs) != 0 {
+ t.Fatalf("%s while instance control was cold: %v", kind, evs)
+ }
+ }
+
+ // Instance control warms: the sweep stops the 300 leftovers, then
+ // recovery, holds, Start, and the latch.
+ if d := time.Until(a.began.Add(largeBoxCold)); d > 0 {
+ time.Sleep(d)
+ }
+ close(b.control.warm)
+ ready := a.waitEvent(t, 30*time.Second, "ready")
+ t.Logf("startup latch opened %s after the agent started (sweep stopped %d units)", ready.at.Sub(a.began), b.control.stops.Load())
+ if got := b.control.stops.Load(); got != largeBoxRoutes {
+ t.Errorf("the sweep stopped %d units, want %d", got, largeBoxRoutes)
+ }
+ for what, at := range map[string]time.Time{"wake": enterWake.at, "sidecar request": enterSidecar.at, "worker wake": workerAsked.at, "scheduled tick": ticked.at} {
+ if !at.Before(ready.at) {
+ t.Fatalf("the %s did not arrive before the latch opened; the test proves nothing about its latch", what)
+ }
+ }
+
+ // Start ran only once the held hosts were fenced again and every
+ // journal was complete.
+ st := a.waitEvent(t, time.Second, "activator-start")
+ if st.arg(0) != "held="+strconv.Itoa(largeBoxHolds) || st.arg(1) != "fenced="+strconv.Itoa(largeBoxHolds) || st.arg(2) != "journals=0" || st.at.After(ready.at) {
+ t.Fatalf("at Start: %v", st)
+ }
+
+ // The wake was held, then served.
+ got := b.answer(t, wakeAns)
+ if got.status != http.StatusOK || got.body != "host="+wake+" socket="+filepath.Join(b.socketDir, wake+".sock") {
+ t.Fatalf("held wake after startup: %d %q %v", got.status, got.body, got.err)
+ }
+ a.assertAfter(t, ready, "acquire", wake)
+
+ // The sidecar's call was held too, then answered as the activator
+ // answers a sidecar whose instance sleeps: it does not wake it.
+ got = b.answer(t, sidecarAns)
+ if got.status != http.StatusServiceUnavailable || got.body != "instance is asleep; sidecar requests do not wake it\n" || got.at.Before(ready.at) {
+ t.Fatalf("sidecar request held through startup: %d %q %v at %s, latch at %s", got.status, got.body, got.err, got.at, ready.at)
+ }
+
+ // The worker: both wakes waited, then coalesced on one child.
+ a.assertAfter(t, ready, "worker-direct", largeBoxWorker, "true")
+ a.assertAfter(t, ready, "worker-boot", largeBoxWorker, "true")
+ a.assertAfter(t, ready, "worker-start", largeBoxWorker)
+ if n := len(a.events("worker-start")); n != 1 {
+ t.Errorf("worker started %d times, want once", n)
+ }
+
+ // The scheduled delivery waited, then woke its instance and ran.
+ a.assertAfter(t, ready, "served", due, largeBoxDuePath)
+ a.assertAfter(t, ready, "acquire", due)
+
+ // The held host was not started: its restored hold answers maintenance.
+ got = b.answer(t, heldAns)
+ if got.status != http.StatusServiceUnavailable || got.body != "instance failed to wake\n" || got.header.Get("Cache-Control") != "no-store" {
+ t.Fatalf("held host after startup: %d %q %v %v", got.status, got.body, got.header, got.err)
+ }
+
+ // The moved host was not served at the socket recovery moved it from:
+ // the request answers as a wake in progress, and its retry is routed
+ // afresh to where recovery put it.
+ got = b.answer(t, movedAns)
+ if got.status != http.StatusServiceUnavailable || got.body != "instance is waking; retry shortly\n" || got.header.Get("Retry-After") != "5" {
+ t.Fatalf("host whose recovery moved it, held through startup: %d %q %v", got.status, got.body, got.err)
+ }
+ nested, err := layout.PlanFor(b.dataDir, b.socketDir, moved, layout.ModeNested)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got = b.get(moved, false)
+ if got.status != http.StatusOK || got.body != "host="+moved+" socket="+nested.To.Socket {
+ t.Fatalf("retry of the moved host: %d %q %v", got.status, got.body, got.err)
+ }
+ got = b.get(held, false)
+ if got.status != http.StatusServiceUnavailable || got.body != "instance failed to wake\n" {
+ t.Fatalf("retry of the held host: %d %q %v", got.status, got.body, got.err)
+ }
+
+ // The background key pass ran after Start, and its slot refused the
+ // held hosts.
+ a.waitEvent(t, 30*time.Second, "startup-done")
+ keys := a.events("key")
+ if len(keys) != largeBoxRoutes-largeBoxHolds {
+ t.Errorf("key pass wrote %d rows, want %d", len(keys), largeBoxRoutes-largeBoxHolds)
+ }
+ for _, e := range keys {
+ if e.at.Before(ready.at) {
+ t.Errorf("key pass reached %s before Start", e.arg(0))
+ }
+ }
+
+ b.assertNoWrongService(t, a)
+ b.assertRecovered(t)
+
+ a.signal(t, syscall.SIGTERM)
+ a.waitExit(t, largeBoxJoinBound+3*time.Second)
+ a.assertCleanShutdown(t)
+}
+
+// testLargeBoxOverrun: instance control never warms, so startup never
+// finishes. A wake must be answered with the wake-timeout 503 at the
+// latch's bound: not an error, not a hang, and not a wake.
+func testLargeBoxOverrun(t *testing.T) {
+ b := newLargeBox(t)
+ a := b.startAgent(t)
+ t.Logf("gate probe answered %s after the agent started, instance control never warms", a.listen)
+
+ wake := largeBoxHost(largeBoxWakeHost)
+ began := time.Now()
+ got := b.get(wake, false)
+ took := time.Since(began)
+ t.Logf("wake held %s by a startup that does not finish", took)
+ if got.err != nil {
+ t.Fatalf("wake during an overrunning startup: %v", got.err)
+ }
+ if got.status != http.StatusServiceUnavailable || got.body != "instance is waking; retry shortly\n" || got.header.Get("Retry-After") != "5" {
+ t.Fatalf("wake during an overrunning startup: %d %q %v", got.status, got.body, got.header)
+ }
+ if took < largeBoxLatchBound-time.Second || took > largeBoxLatchBound+5*time.Second {
+ t.Fatalf("wake answered after %s, want the latch bound %s", took, largeBoxLatchBound)
+ }
+ a.assertNeverStarted(t)
+ b.assertNoWrongService(t, a)
+
+ // Shutting down with the sweep still waiting: nothing in flight, so the
+ // edge returns at once and runAgent's join bounds the rest.
+ at := time.Now()
+ a.signal(t, syscall.SIGTERM)
+ a.waitExit(t, largeBoxJoinBound+3*time.Second)
+ t.Logf("agent exited %s after SIGTERM with the sweep still waiting", time.Since(at))
+ a.assertNeverStarted(t)
+ a.assertCleanShutdown(t)
+}
+
+// testLargeBoxShutdownMidStartup: SIGTERM with startup still waiting on
+// the sweep and a wake held at the latch. The held request is answered
+// 503 (never proxied), the background work is joined within runAgent's
+// bound, and nothing of startup outlives it.
+func testLargeBoxShutdownMidStartup(t *testing.T) {
+ b := newLargeBox(t)
+ a := b.startAgent(t)
+
+ wake := largeBoxHost(largeBoxWakeHost)
+ ans := b.fire(wake, false)
+ a.waitEvent(t, 10*time.Second, "enter", wake)
+
+ at := time.Now()
+ a.signal(t, syscall.SIGTERM)
+ // The edge drains in-flight requests for up to largeBoxDrain before it
+ // returns, and the held wake is in flight until the activator shuts
+ // down after the join.
+ a.waitExit(t, largeBoxDrain+largeBoxJoinBound+3*time.Second)
+ t.Logf("agent exited %s after SIGTERM with a wake held at the latch", time.Since(at))
+ got := b.answer(t, ans)
+ if got.err != nil || got.status != http.StatusServiceUnavailable {
+ t.Fatalf("wake held through shutdown: %d %q %v", got.status, got.body, got.err)
+ }
+ a.assertNeverStarted(t)
+ b.assertNoWrongService(t, a)
+ a.assertCleanShutdown(t)
+}
+
+// --- runAgent's wiring ------------------------------------------------------------
+
+// TestRunAgentStartupWiring reads the startup's source and fails if it
+// drifts from what runLargeBoxAgent rebuilds, or if anything on runAgent's
+// path to the edge waits for startup to finish. Calls are matched on the
+// callee's own name (a selector's Sel), and order means execution order on
+// one goroutine: a step inside a go or defer statement, or inside a
+// function literal that is not invoked on the spot, does not count as
+// taken. Where the hotfix does mean a step to run in the background, the
+// one goroutine it runs on is found structurally and its order checked
+// inside it.
+func TestRunAgentStartupWiring(t *testing.T) {
+ fset := token.NewFileSet()
+ decls := map[string]*ast.FuncDecl{} // runAgent, startActivator, recoverLayoutAtStart, activatorStart.Run
+ files, err := filepath.Glob("*.go")
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, name := range files {
+ if strings.HasSuffix(name, "_test.go") {
+ continue
+ }
+ f, err := parser.ParseFile(fset, name, nil, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, d := range f.Decls {
+ fd, ok := d.(*ast.FuncDecl)
+ if !ok {
+ continue
+ }
+ key := fd.Name.Name
+ if fd.Recv != nil && len(fd.Recv.List) == 1 {
+ key = types.ExprString(fd.Recv.List[0].Type) + "." + key
+ }
+ switch key {
+ case "runAgent", "startActivator", "recoverLayoutAtStart", "activatorStart.Run":
+ decls[key] = fd
+ }
+ }
+ }
+ for _, key := range []string{"runAgent", "startActivator", "recoverLayoutAtStart", "activatorStart.Run"} {
+ if decls[key] == nil {
+ t.Fatalf("%s not found in package host", key)
+ }
+ }
+ str := types.ExprString
+
+ // runAgent: the latch is set, the start sequence is handed the startup
+ // context, the edge runs, then cancel, join and Shutdown, all on
+ // runAgent's own goroutine.
+ runAgent := decls["runAgent"].Body
+ sync := syncCalls(runAgent)
+ var run, edgeCall *ast.CallExpr
+ var lit *ast.CompositeLit
+ for _, c := range sync {
+ if sel, ok := c.Fun.(*ast.SelectorExpr); ok && sel.Sel.Name == "Run" {
+ if cl, ok := sel.X.(*ast.CompositeLit); ok && str(cl.Type) == "activatorStart" {
+ run, lit = c, cl
+ }
+ }
+ if calleeName(c) == "RunWithActivatorReady" {
+ edgeCall = c
+ }
+ }
+ if run == nil || edgeCall == nil {
+ t.Fatalf("runAgent: activatorStart{...}.Run called synchronously: %v; edge.RunWithActivatorReady called synchronously: %v", run != nil, edgeCall != nil)
+ }
+ if len(run.Args) != 1 || str(run.Args[0]) != "startupCtx" {
+ t.Errorf("activatorStart.Run is given %v, want startupCtx", run.Args)
+ }
+ var latch token.Pos
+ ast.Inspect(runAgent, func(n ast.Node) bool {
+ if a, ok := n.(*ast.AssignStmt); ok && len(a.Lhs) == 1 && len(a.Rhs) == 1 && selName(a.Lhs[0]) == "Startup" && str(a.Rhs[0]) == "started" {
+ latch = a.Pos()
+ }
+ return true
+ })
+ if latch == token.NoPos || latch > run.Pos() {
+ t.Error("runAgent: svc.Startup = started is not set before activatorStart.Run")
+ }
+ after := func(name string, arg string) token.Pos {
+ for _, c := range sync {
+ if c.Pos() > edgeCall.Pos() && calleeName(c) == name && (arg == "" || (len(c.Args) > 0 && str(c.Args[len(c.Args)-1]) == arg)) {
+ return c.Pos()
+ }
+ }
+ return token.NoPos
+ }
+ prev, prevName := edgeCall.Pos(), "edge.RunWithActivatorReady"
+ for _, step := range []struct{ name, arg string }{{"cancelStartup", ""}, {"joinAgentStartup", "startupDone"}, {"Shutdown", ""}} {
+ p := after(step.name, step.arg)
+ if p == token.NoPos || p < prev {
+ t.Errorf("runAgent: %s is not called synchronously after %s", step.name, prevName)
+ continue
+ }
+ prev, prevName = p, step.name
+ }
+
+ // The activatorStart fields the child copies, and the steps inside
+ // them. Run calls each field on its background goroutine, so inside a
+ // field each step must be synchronous.
+ fields := map[string]ast.Expr{}
+ for _, el := range lit.Elts {
+ if kv, ok := el.(*ast.KeyValueExpr); ok {
+ fields[str(kv.Key)] = kv.Value
+ }
+ }
+ for field, want := range map[string]string{
+ "Start": "Start", "Publish": "Store", "Slot": "PrepareScheduledToken",
+ "Routes": "List", "Route": "Get", "Claims": "Enabled",
+ } {
+ if got := selName(fields[field]); got != want {
+ t.Errorf("activatorStart.%s is %s, want a %s method value", field, exprOrNil(fields[field]), want)
+ }
+ }
+ funcField := func(name string) ast.Node {
+ fl, _ := fields[name].(*ast.FuncLit)
+ if fl == nil {
+ t.Errorf("activatorStart.%s is not a func literal", name)
+ return nil
+ }
+ return fl.Body
+ }
+ if body := funcField("Recover"); body != nil {
+ requireSteps(t, "activatorStart.Recover", body, "newLayoutPass", "startupRecoveryIntent", "recoverLayoutAtStart")
+ }
+ if body := funcField("Sweep"); body != nil {
+ requireSteps(t, "activatorStart.Sweep", body, "Sweep")
+ }
+ if body := funcField("Ready"); body != nil {
+ closes := false
+ for _, c := range syncCalls(body) {
+ closes = closes || (calleeName(c) == "close" && len(c.Args) == 1 && str(c.Args[0]) == "started")
+ }
+ if !closes {
+ t.Error("activatorStart.Ready does not close started synchronously")
+ }
+ }
+
+ // activatorStart.Run: on a legacy box, one goroutine runs the start
+ // sequence and then the key pass, in that order; start itself is
+ // Start, then Ready.
+ runBody := decls["activatorStart.Run"].Body
+ var background []*ast.FuncLit
+ ast.Inspect(runBody, func(n ast.Node) bool {
+ if g, ok := n.(*ast.GoStmt); ok {
+ if fl, ok := g.Call.Fun.(*ast.FuncLit); ok && hasCall(syncCalls(fl.Body), "startActivator") {
+ background = append(background, fl)
+ }
+ }
+ return true
+ })
+ if len(background) != 1 {
+ t.Errorf("activatorStart.Run: %d goroutines run startActivator, want the one background start", len(background))
+ } else {
+ requireSteps(t, "activatorStart.Run's background goroutine", background[0].Body, "startActivator", "backfillInstanceKeys")
+ }
+ var startBody ast.Node
+ ast.Inspect(runBody, func(n ast.Node) bool {
+ if a, ok := n.(*ast.AssignStmt); ok && len(a.Lhs) == 1 && len(a.Rhs) == 1 && str(a.Lhs[0]) == "start" {
+ if fl, ok := a.Rhs[0].(*ast.FuncLit); ok {
+ startBody = fl.Body
+ }
+ }
+ return true
+ })
+ if startBody == nil {
+ t.Error("activatorStart.Run: no start closure")
+ } else {
+ requireSteps(t, "activatorStart.Run's start", startBody, "Start", "Ready")
+ }
+
+ // startActivator: its sweep, recoverLayout and start parameters, called
+ // in that order, synchronously (matched by position, not name).
+ sa := decls["startActivator"]
+ var params []string
+ for _, f := range sa.Type.Params.List {
+ for _, n := range f.Names {
+ params = append(params, n.Name)
+ }
+ }
+ if len(params) != 4 {
+ t.Errorf("startActivator takes %v, want (ctx, sweep, recoverLayout, start)", params)
+ } else {
+ requireSteps(t, "startActivator", sa.Body, params[1], params[2], params[3])
+ }
+
+ // recoverLayoutAtStart: journal recovery, then hold restoration.
+ requireSteps(t, "recoverLayoutAtStart", decls["recoverLayoutAtStart"].Body, "Recover", "RestoreHolds")
+
+ // The worker boot pass waits on the latch or the startup context, then
+ // lists and wakes.
+ var bootPass *ast.FuncLit
+ ast.Inspect(runAgent, func(n ast.Node) bool {
+ if g, ok := n.(*ast.GoStmt); ok {
+ if fl, ok := g.Call.Fun.(*ast.FuncLit); ok && hasCall(syncCalls(fl.Body), "runWorkerBootPass") {
+ bootPass = fl
+ }
+ }
+ return true
+ })
+ if bootPass == nil {
+ t.Error("runAgent: no goroutine runs runWorkerBootPass")
+ } else if !receivesFrom(bootPass.Body, "started") || !hasCall(syncCalls(bootPass.Body), "Done") {
+ t.Error("runAgent: the worker boot pass does not wait on started and startupCtx.Done")
+ }
+
+ // Nothing before the edge waits for startup. On runAgent's own
+ // goroutine (closures invoked on the spot included): no receive at all,
+ // no join, and a closure that waits on startup is only called directly
+ // on a claim box, where Run has already started the activator before
+ // it returns.
+ waiters := map[string]bool{}
+ ast.Inspect(runAgent, func(n ast.Node) bool {
+ if a, ok := n.(*ast.AssignStmt); ok && len(a.Lhs) == 1 && len(a.Rhs) == 1 {
+ if fl, ok := a.Rhs[0].(*ast.FuncLit); ok && (receivesFrom(fl.Body, "started") || receivesFrom(fl.Body, "startupDone") || hasCall(syncCalls(fl.Body), "joinAgentStartup")) {
+ waiters[str(a.Lhs[0])] = true
+ }
+ }
+ return true
+ })
+ var walk func(n ast.Node, claimBranch bool)
+ walk = func(n ast.Node, claimBranch bool) {
+ ast.Inspect(n, func(n ast.Node) bool {
+ if n == nil || n.Pos() >= edgeCall.Pos() {
+ return false
+ }
+ switch n := n.(type) {
+ case *ast.FuncLit, *ast.GoStmt, *ast.DeferStmt:
+ return false
+ case *ast.IfStmt:
+ if n.Init != nil {
+ walk(n.Init, claimBranch)
+ }
+ walk(n.Cond, claimBranch)
+ walk(n.Body, claimBranch || selName(n.Cond) == "Enabled")
+ if n.Else != nil {
+ walk(n.Else, claimBranch)
+ }
+ return false
+ case *ast.UnaryExpr:
+ if n.Op == token.ARROW {
+ t.Errorf("runAgent waits on %s at %s, before the edge listens", str(n.X), fset.Position(n.Pos()))
+ }
+ case *ast.CallExpr:
+ name := calleeName(n)
+ if name == "joinAgentStartup" {
+ t.Errorf("runAgent joins startup at %s, before the edge listens", fset.Position(n.Pos()))
+ }
+ if waiters[str(n.Fun)] && !claimBranch {
+ t.Errorf("runAgent calls %s, which waits on startup, at %s, before the edge listens", str(n.Fun), fset.Position(n.Pos()))
+ }
+ if fl, ok := n.Fun.(*ast.FuncLit); ok { // invoked on the spot: its body runs here
+ walk(fl.Body, claimBranch)
+ for _, a := range n.Args {
+ walk(a, claimBranch)
+ }
+ return false
+ }
+ }
+ return true
+ })
+ }
+ walk(runAgent, false)
+}
+
+// syncCalls lists, in source order, the calls body makes on its own
+// goroutine: none inside a go or defer statement, and none inside a
+// function literal unless the literal is invoked right there.
+func syncCalls(body ast.Node) []*ast.CallExpr {
+ var out []*ast.CallExpr
+ ast.Inspect(body, func(n ast.Node) bool {
+ switch n := n.(type) {
+ case *ast.GoStmt, *ast.DeferStmt, *ast.FuncLit:
+ return false
+ case *ast.CallExpr:
+ out = append(out, n)
+ if fl, ok := n.Fun.(*ast.FuncLit); ok {
+ for _, a := range n.Args {
+ out = append(out, syncCalls(a)...)
+ }
+ out = append(out, syncCalls(fl.Body)...)
+ return false
+ }
+ }
+ return true
+ })
+ return out
+}
+
+// requireSteps fails unless body calls each step synchronously, in order,
+// and calls none of them anywhere else (in a go or defer statement, or a
+// literal that is not invoked on the spot).
+func requireSteps(t *testing.T, where string, body ast.Node, steps ...string) {
+ t.Helper()
+ sync := syncCalls(body)
+ inSync := map[*ast.CallExpr]bool{}
+ for _, c := range sync {
+ inSync[c] = true
+ }
+ ast.Inspect(body, func(n ast.Node) bool {
+ if c, ok := n.(*ast.CallExpr); ok && !inSync[c] {
+ for _, s := range steps {
+ if calleeName(c) == s {
+ t.Errorf("%s: %s is called off its goroutine (go, defer, or a literal not invoked there)", where, s)
+ }
+ }
+ }
+ return true
+ })
+ i := 0
+ for _, c := range sync {
+ if i < len(steps) && calleeName(c) == steps[i] {
+ i++
+ }
+ }
+ if i < len(steps) {
+ t.Errorf("%s: want %v called in that order on its goroutine; %s is missing or out of order", where, steps, steps[i])
+ }
+}
+
+// calleeName is the name a call is matched on: the function's own name,
+// whatever the receiver or package it is reached through.
+func calleeName(c *ast.CallExpr) string {
+ switch f := c.Fun.(type) {
+ case *ast.Ident:
+ return f.Name
+ case *ast.SelectorExpr:
+ return f.Sel.Name
+ }
+ return ""
+}
+
+func selName(e ast.Expr) string {
+ switch e := e.(type) {
+ case *ast.SelectorExpr:
+ return e.Sel.Name
+ case *ast.Ident:
+ return e.Name
+ }
+ return ""
+}
+
+func hasCall(calls []*ast.CallExpr, name string) bool {
+ for _, c := range calls {
+ if calleeName(c) == name {
+ return true
+ }
+ }
+ return false
+}
+
+// receivesFrom reports whether body receives from the channel named ch
+// (in a select case or a bare receive).
+func receivesFrom(body ast.Node, ch string) bool {
+ found := false
+ ast.Inspect(body, func(n ast.Node) bool {
+ if u, ok := n.(*ast.UnaryExpr); ok && u.Op == token.ARROW && types.ExprString(u.X) == ch {
+ found = true
+ }
+ return !found
+ })
+ return found
+}
+
+func exprOrNil(e ast.Expr) string {
+ if e == nil {
+ return "unset"
+ }
+ return types.ExprString(e)
+}
+
+// --- the box ------------------------------------------------------------------
+
+type largeBox struct {
+ dataDir, socketDir, storeDir, acmeDir string
+ reg *registry.Registry
+ control *largeBoxControl
+ httpAddr, httpsAddr string
+ client *http.Client
+}
+
+// largeBoxSocketDir makes the box's run directory. It must be short: a
+// nested instance's socket is <dir>/instances/<instance id>/app.sock, the
+// activator binds a ".control" socket beside it, and the whole path must
+// fit a unix socket address (provision.MaxUnixPath). The default temp dir
+// is tried first, then a directory of the test's own directly under /tmp;
+// when neither is short enough the test is skipped, not failed, since
+// nothing about the code under test is wrong.
+func largeBoxSocketDir(t *testing.T) string {
+ t.Helper()
+ fits := func(dir string) bool {
+ return len(filepath.Join(dir, "instances", provision.InstanceID(largeBoxHost(0)), "app.sock.control")) <= provision.MaxUnixPath
+ }
+ var tried []string
+ for _, base := range []string{os.TempDir(), "/tmp"} {
+ dir, err := os.MkdirTemp(base, "lb")
+ if err != nil {
+ tried = append(tried, base+": "+err.Error())
+ continue
+ }
+ if fits(dir) {
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ return dir
+ }
+ os.RemoveAll(dir)
+ tried = append(tried, dir+": too long")
+ }
+ t.Skipf("no directory short enough for a nested instance's control socket (%d bytes at most): %s", provision.MaxUnixPath, strings.Join(tried, "; "))
+ return ""
+}
+
+// newLargeBox builds the box the agent restarts on: largeBoxRoutes
+// sleeping instances on flat paths, largeBoxJournals of them with a layout
+// move cut short at three different points, largeBoxHolds held down after
+// a refused move (the reason, a second name for the database, persists),
+// a scheduled delivery due on one instance, an always-on worker the boot
+// pass starts, a nested layout intent in the deployment store, a cached
+// certificate for every host the test asks for, and a cold instance
+// control.
+func newLargeBox(t *testing.T) *largeBox {
+ t.Helper()
+ sd := largeBoxSocketDir(t)
+ b := &largeBox{dataDir: t.TempDir(), storeDir: t.TempDir(), acmeDir: t.TempDir(), socketDir: sd}
+
+ var err error
+ b.reg, err = registry.Open(filepath.Join(b.dataDir, "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { b.reg.Close() })
+ var hosts []string
+ for i := 0; i < largeBoxRoutes; i++ {
+ h := largeBoxHost(i)
+ hosts = append(hosts, h)
+ if err := b.reg.Upsert(registry.Route{
+ Host: h, Kind: "instance", Account: "acct", App: "shop", Status: "ready", Hibernate: true,
+ Socket: filepath.Join(sd, h+".sock"), DBPath: provision.DefaultDBPath(b.dataDir, h),
+ }); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := b.reg.Upsert(registry.Route{
+ Host: largeBoxWorker, Kind: "sidecar", Account: "acct", App: "shop", Status: "ready", Args: `["serve"]`,
+ }); err != nil {
+ t.Fatal(err)
+ }
+ due := time.Now().Unix() - 5
+ if err := b.reg.UpsertTimer(registry.Timer{
+ Host: largeBoxHost(largeBoxDueHost), Name: "tick", Kind: schedule.KindSchedule, Spec: "every 6h0m0s",
+ Path: largeBoxDuePath, NextAt: due, DueAt: due,
+ }); err != nil {
+ t.Fatal(err)
+ }
+ for i := 0; i < largeBoxJournals; i++ {
+ crash := "after-journal"
+ switch {
+ case i >= 6:
+ crash = "before-remove"
+ case i >= 4:
+ crash = "mid-files"
+ }
+ b.crashedMove(t, largeBoxHost(i), crash)
+ }
+ for i := largeBoxJournals; i < largeBoxJournals+largeBoxHolds; i++ {
+ b.heldMove(t, largeBoxHost(i), i)
+ }
+ appendIntent(t, release.NewDirStore(b.storeDir), release.LayoutModeNested, "")
+
+ served := []string{largeBoxHost(largeBoxWakeHost), largeBoxHost(largeBoxSidecarHost)}
+ for i := 0; i < largeBoxJournals+largeBoxHolds; i++ {
+ served = append(served, largeBoxHost(i))
+ }
+ pool := writeLargeBoxCerts(t, b.acmeDir, served)
+
+ b.control = startLargeBoxControl(t, filepath.Join(sd, "control.sock"), hosts)
+ b.client = &http.Client{
+ Timeout: largeBoxLatchBound + 15*time.Second,
+ Transport: &http.Transport{
+ Proxy: nil,
+ DisableKeepAlives: true,
+ TLSClientConfig: &tls.Config{RootCAs: pool},
+ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, "tcp", b.httpsAddr)
+ },
+ },
+ CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
+ }
+ return b
+}
+
+func (b *largeBox) seedFlat(t *testing.T, h string) {
+ t.Helper()
+ db := provision.DefaultDBPath(b.dataDir, h)
+ for path, body := range map[string]string{db: "db " + h, db + "-wal": "wal", filepath.Join(b.dataDir, h+".env"): "CARLOS_ADMIN_TOKEN=t\n"} {
+ if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+// crashedMove leaves h as an agent killed mid-migration to nested leaves
+// it: the instance directories made and the journal written, then
+// nothing moved (after-journal), the database moved (mid-files), or
+// everything moved and the row flipped with only the journal's removal
+// left (before-remove).
+func (b *largeBox) crashedMove(t *testing.T, h, crash string) {
+ t.Helper()
+ b.seedFlat(t, h)
+ if err := layout.EnsureDirs(b.dataDir, b.socketDir, h, -1); err != nil {
+ t.Fatal(err)
+ }
+ plan, err := layout.PlanFor(b.dataDir, b.socketDir, h, layout.ModeNested)
+ if err != nil {
+ t.Fatal(err)
+ }
+ present := make([]bool, len(plan.Artifacts))
+ for i, a := range plan.Artifacts {
+ if a.Name == "socket" {
+ continue
+ }
+ _, err := os.Lstat(a.Src)
+ present[i] = err == nil
+ }
+ j := plan.Journal(present)
+ if err := layout.WriteJournal(b.dataDir, j); err != nil {
+ t.Fatal(err)
+ }
+ move := func(a layout.Artifact) {
+ if err := os.Rename(a.Src, a.Dst); err != nil {
+ t.Fatal(err)
+ }
+ }
+ switch crash {
+ case "mid-files":
+ move(plan.Artifacts[0])
+ case "before-remove":
+ for i, a := range plan.Artifacts {
+ if present[i] {
+ move(a)
+ }
+ }
+ if err := b.reg.SetInstancePaths(h, plan.From.DB, plan.From.Socket, plan.To.DB, plan.To.Socket); err != nil {
+ t.Fatal(err)
+ }
+ if err := layout.UpdatePhase(b.dataDir, j, layout.PhaseDone); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+// heldMove leaves h held down after a refused move to nested: its hold
+// marker, and the reason, which persists across the restart (a second
+// name for its database, which a flat tenant can plant).
+func (b *largeBox) heldMove(t *testing.T, h string, i int) {
+ t.Helper()
+ b.seedFlat(t, h)
+ if err := os.Link(provision.DefaultDBPath(b.dataDir, h), filepath.Join(b.dataDir, ".x"+strconv.Itoa(i))); err != nil {
+ t.Fatal(err)
+ }
+ dir := filepath.Join(b.dataDir, "layout")
+ if err := os.MkdirAll(dir, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ marker, _ := json.Marshal(map[string]any{"host": h, "reason": "db has 2 names", "since": time.Now().UTC()})
+ if err := os.WriteFile(filepath.Join(dir, h+".hold"), marker, 0o600); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// assertRecovered checks the files: every journal completed and removed,
+// each journalled host's row and database at the nested layout, and every
+// hold marker kept.
+func (b *largeBox) assertRecovered(t *testing.T) {
+ t.Helper()
+ if hosts, err := layout.JournalHosts(b.dataDir); err != nil || len(hosts) != 0 {
+ t.Errorf("journals after startup: %v %v", hosts, err)
+ }
+ for i := 0; i < largeBoxJournals; i++ {
+ h := largeBoxHost(i)
+ plan, err := layout.PlanFor(b.dataDir, b.socketDir, h, layout.ModeNested)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rt, err := b.reg.Get(h)
+ if err != nil || rt == nil || rt.DBPath != plan.To.DB || rt.Socket != plan.To.Socket {
+ t.Errorf("%s row after recovery: %+v %v", h, rt, err)
+ }
+ if body, err := os.ReadFile(plan.Artifacts[0].Dst); err != nil || string(body) != "db "+h {
+ t.Errorf("%s database after recovery: %q %v", h, body, err)
+ }
+ }
+ held, err := layout.HoldMarkers(b.dataDir)
+ if err != nil || len(held) != largeBoxHolds {
+ t.Errorf("hold markers after startup: %v %v", held, err)
+ }
+}
+
+// assertNoWrongService reads the backend's record: no wake admitted before
+// the latch opened, no held host started, and no journalled host started
+// at its pre-recovery socket.
+func (b *largeBox) assertNoWrongService(t *testing.T, a *largeBoxAgent) {
+ t.Helper()
+ if v := a.events("violation"); len(v) != 0 {
+ t.Errorf("backend violations: %v", v)
+ }
+ for _, e := range append(a.events("acquire"), a.events("start")...) {
+ for i := largeBoxJournals; i < largeBoxJournals+largeBoxHolds; i++ {
+ if e.arg(0) == largeBoxHost(i) {
+ t.Errorf("held host %s reached the backend: %v", e.arg(0), e)
+ }
+ }
+ }
+ for _, e := range a.events("start") {
+ for i := 0; i < largeBoxJournals; i++ {
+ if h := largeBoxHost(i); e.arg(0) == h && e.arg(1) == filepath.Join(b.socketDir, h+".sock") {
+ t.Errorf("%s started at its pre-recovery socket", h)
+ }
+ }
+ }
+}
+
+type largeBoxAnswer struct {
+ status int
+ body string
+ header http.Header
+ err error
+ at time.Time // when the answer was read
+}
+
+// get asks the edge for host's root over TLS. sidecar marks the request
+// as the host's sidecar's (activator.SidecarHeader).
+func (b *largeBox) get(host string, sidecar bool) largeBoxAnswer {
+ req, err := http.NewRequest(http.MethodGet, "https://"+host+"/", nil)
+ if err != nil {
+ return largeBoxAnswer{err: err, at: time.Now()}
+ }
+ if sidecar {
+ req.Header.Set(activator.SidecarHeader, "1")
+ }
+ resp, err := b.client.Do(req)
+ if err != nil {
+ return largeBoxAnswer{err: err, at: time.Now()}
+ }
+ defer resp.Body.Close()
+ body, err := io.ReadAll(resp.Body)
+ return largeBoxAnswer{status: resp.StatusCode, body: string(body), header: resp.Header, err: err, at: time.Now()}
+}
+
+func (b *largeBox) fire(host string, sidecar bool) chan largeBoxAnswer {
+ ch := make(chan largeBoxAnswer, 1)
+ go func() { ch <- b.get(host, sidecar) }()
+ return ch
+}
+
+func (b *largeBox) answer(t *testing.T, ch chan largeBoxAnswer) largeBoxAnswer {
+ t.Helper()
+ select {
+ case got := <-ch:
+ return got
+ case <-time.After(10 * time.Second):
+ t.Fatal("request did not finish")
+ return largeBoxAnswer{}
+ }
+}
+
+func freeLoopbackAddr(t *testing.T) string {
+ t.Helper()
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer ln.Close()
+ return ln.Addr().String()
+}
+
+// writeLargeBoxCerts puts one self-signed certificate for hosts into the
+// edge's autocert directory cache, under each host's name, so the edge in
+// production mode serves them TLS without ordering anything.
+func writeLargeBoxCerts(t *testing.T, dir string, hosts []string) *x509.CertPool {
+ t.Helper()
+ key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ tmpl := &x509.Certificate{
+ SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "large box"}, DNSNames: hosts,
+ NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(24 * time.Hour),
+ KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
+ ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, BasicConstraintsValid: true, IsCA: true,
+ }
+ der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ keyDER, err := x509.MarshalECPrivateKey(key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ pemBytes := append(pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})...)
+ for _, h := range hosts {
+ if err := os.WriteFile(filepath.Join(dir, h), pemBytes, 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ cert, err := x509.ParseCertificate(der)
+ if err != nil {
+ t.Fatal(err)
+ }
+ pool := x509.NewCertPool()
+ pool.AddCert(cert)
+ return pool
+}
+
+// --- instance control ---------------------------------------------------------
+
+// largeBoxControl is the real instance-control server with a cold systemd
+// behind its Systemctl seam: every call waits for warm, then the box's
+// leftover units are listed active and each stop takes largeBoxUnitStop.
+type largeBoxControl struct {
+ warm chan struct{}
+ units string
+ stops atomic.Int32
+}
+
+func startLargeBoxControl(t *testing.T, path string, hosts []string) *largeBoxControl {
+ t.Helper()
+ c := &largeBoxControl{warm: make(chan struct{})}
+ var units strings.Builder
+ for _, h := range hosts {
+ fmt.Fprintf(&units, "%s loaded active running carlos instance\n", instancectl.UnitName(instancectl.RoleRun, h))
+ }
+ c.units = units.String()
+ srv := &instancectl.Server{
+ Systemctl: c.systemctl,
+ PeerOK: func(net.Conn) error { return nil },
+ CgroupRoot: t.TempDir(),
+ Build: version.Build,
+ Logf: func(string, ...any) {},
+ }
+ ln, err := net.Listen("unix", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan struct{})
+ go func() { defer close(done); _ = srv.Serve(ctx, ln) }()
+ t.Cleanup(func() { cancel(); <-done })
+ return c
+}
+
+func (c *largeBoxControl) systemctl(ctx context.Context, args ...string) (string, error) {
+ select {
+ case <-c.warm:
+ case <-ctx.Done():
+ return "", ctx.Err()
+ }
+ switch args[0] {
+ case "list-units":
+ return c.units, nil
+ case "list-jobs":
+ return "", nil
+ case "stop":
+ select {
+ case <-time.After(largeBoxUnitStop):
+ case <-ctx.Done():
+ return "", ctx.Err()
+ }
+ c.stops.Add(1)
+ return "", nil
+ case "show":
+ return "inactive\n", nil
+ }
+ return "", fmt.Errorf("systemctl %v: not part of a sweep", args)
+}
+
+// --- the agent process ----------------------------------------------------------
+
+type largeBoxAgent struct {
+ cmd *exec.Cmd
+ began time.Time
+ listen time.Duration // process start to the gate probe's first answer
+ eventsPath string
+ log string
+ exited chan struct{} // closed once the process has been waited for
+ exitErr error // Wait's answer, read after exited closes
+ b *largeBox
+}
+
+// startAgent starts the agent and returns it once the gate's probe is
+// answered and its TLS listener accepts. The edge's -http and -https take
+// an address, and :0 would bind, but the edge never says which port it
+// got, so the test picks free ports and hands them over. A port taken in
+// between makes the edge's listener fail and the agent exit; only that is
+// retried on fresh ports (largeBoxRetryable). A deadline missed is the
+// failure this test exists to catch, and is never retried.
+func (b *largeBox) startAgent(t *testing.T) *largeBoxAgent {
+ t.Helper()
+ for attempt := 1; ; attempt++ {
+ b.httpAddr, b.httpsAddr = freeLoopbackAddr(t), freeLoopbackAddr(t)
+ a := b.launchAgent(t)
+ outcome, err := a.awaitListening()
+ if outcome == largeBoxUp {
+ outcome, err = a.awaitTLS()
+ }
+ if outcome == largeBoxUp {
+ return a
+ }
+ agentLog, _ := os.ReadFile(a.log)
+ if largeBoxRetryable(outcome, string(agentLog), b.httpAddr, b.httpsAddr) && attempt < largeBoxBindAttempts {
+ t.Logf("attempt %d: %v; a port the test handed the agent was taken before it bound it, starting again on fresh ports", attempt, err)
+ continue
+ }
+ t.Fatal(err)
+ }
+}
+
+// largeBoxOutcome is how waiting for the agent to come up ended.
+type largeBoxOutcome int
+
+const (
+ largeBoxUp largeBoxOutcome = iota
+ largeBoxLate // a deadline passed: a failure, never retried
+ largeBoxExited // the agent exited before it was up
+)
+
+// largeBoxRetryable reports whether a start may be retried on fresh
+// ports: only when the agent exited, and its log says the edge could not
+// bind one of the addresses the test handed it because it was in use.
+// Edge listener failures are logged as "edge: http listener <addr>: ..."
+// (edge.RunWithActivatorReady).
+func largeBoxRetryable(o largeBoxOutcome, agentLog string, addrs ...string) bool {
+ if o != largeBoxExited {
+ return false
+ }
+ for _, line := range strings.Split(agentLog, "\n") {
+ if !strings.Contains(line, "address already in use") {
+ continue
+ }
+ for _, addr := range addrs {
+ if strings.Contains(line, " listener "+addr+": ") {
+ return true
+ }
+ }
+ }
+ return false
+}
+
+func TestLargeBoxRetriesOnlyItsOwnBindFailure(t *testing.T) {
+ const plain, tls = "127.0.0.1:40001", "127.0.0.1:40002"
+ bind := "edge: https listener " + tls + ": listen tcp " + tls + ": bind: address already in use\n"
+ for _, c := range []struct {
+ name string
+ o largeBoxOutcome
+ log string
+ want bool
+ }{
+ {"own bind failure", largeBoxExited, "edge: serving 300 route(s)\n" + bind, true},
+ {"deadline missed, bind failure logged too", largeBoxLate, bind, false},
+ {"exited for another reason", largeBoxExited, "edge: tunnels: invalid\n", false},
+ {"someone else's address", largeBoxExited, "edge: http listener 127.0.0.1:1: listen tcp 127.0.0.1:1: bind: address already in use\n", false},
+ {"in use, but not a listener line", largeBoxExited, "agent: something: address already in use\n", false},
+ } {
+ if got := largeBoxRetryable(c.o, c.log, plain, tls); got != c.want {
+ t.Errorf("%s: retry = %v, want %v", c.name, got, c.want)
+ }
+ }
+}
+
+func (b *largeBox) launchAgent(t *testing.T) *largeBoxAgent {
+ t.Helper()
+ cfgDir := t.TempDir()
+ cfg := largeBoxConfig{
+ DataDir: b.dataDir, SocketDir: b.socketDir, StoreDir: b.storeDir, AcmeDir: b.acmeDir,
+ Control: filepath.Join(b.socketDir, "control.sock"), Events: filepath.Join(cfgDir, "events"),
+ HTTP: b.httpAddr, HTTPS: b.httpsAddr,
+ }
+ raw, err := json.Marshal(cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ cfgPath := filepath.Join(cfgDir, "agent.json")
+ if err := os.WriteFile(cfgPath, raw, 0o600); err != nil {
+ t.Fatal(err)
+ }
+ exe, err := os.Executable()
+ if err != nil {
+ t.Fatal(err)
+ }
+ a := &largeBoxAgent{eventsPath: cfg.Events, log: filepath.Join(cfgDir, "agent.log"), exited: make(chan struct{}), b: b}
+ logFile, err := os.Create(a.log)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { logFile.Close() })
+ a.cmd = exec.Command(exe, "-test.run=^TestStartupLargeBoxEndToEnd$", "-test.count=1")
+ // A clean environment: nothing of the developer's or the runner's
+ // (a deployment bucket, AWS credentials, CARLOS_*) reaches the agent.
+ a.cmd.Env = []string{"HOME=" + b.dataDir, "TMPDIR=" + os.TempDir(), "CARLOS_DATA_DIR=" + b.dataDir, largeBoxAgentEnv + "=" + cfgPath}
+ a.cmd.Stdout, a.cmd.Stderr = logFile, logFile
+ a.began = time.Now()
+ if err := a.cmd.Start(); err != nil {
+ t.Fatal(err)
+ }
+ go func() {
+ a.exitErr = a.cmd.Wait()
+ close(a.exited)
+ }()
+ t.Cleanup(func() {
+ select {
+ case <-a.exited:
+ default:
+ _ = a.cmd.Process.Kill()
+ <-a.exited
+ }
+ if t.Failed() {
+ if out, err := os.ReadFile(a.log); err == nil {
+ if len(out) > 16<<10 {
+ out = out[len(out)-16<<10:]
+ }
+ t.Logf("agent log (tail):\n%s", out)
+ }
+ }
+ })
+ return a
+}
+
+// awaitListening probes as the update gate does (systemapply.go
+// productionSysOps.probeEdge): GET / on the edge's plain-HTTP listener,
+// three-second client timeout, no redirect followed, and any HTTP answer
+// passes. It records how long after the process started the first answer
+// came; an answer past largeBoxListenBound, or none by then, is late.
+func (a *largeBoxAgent) awaitListening() (largeBoxOutcome, error) {
+ probe := &http.Client{
+ Timeout: 3 * time.Second,
+ Transport: &http.Transport{Proxy: nil, DisableKeepAlives: true},
+ CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
+ }
+ deadline := a.began.Add(largeBoxListenBound)
+ for {
+ resp, err := probe.Get("http://" + a.b.httpAddr + "/")
+ if err == nil {
+ resp.Body.Close()
+ a.listen = time.Since(a.began)
+ if a.listen > largeBoxListenBound {
+ return largeBoxLate, fmt.Errorf("gate probe answered %s after the agent started, over %s", a.listen, largeBoxListenBound)
+ }
+ return largeBoxUp, nil
+ }
+ if time.Now().After(deadline) {
+ return largeBoxLate, fmt.Errorf("gate probe unanswered %s after the agent started: %v", time.Since(a.began), err)
+ }
+ select {
+ case <-a.exited:
+ return largeBoxExited, fmt.Errorf("agent exited before listening: %v", a.exitErr)
+ default:
+ }
+ time.Sleep(20 * time.Millisecond)
+ }
+}
+
+// awaitTLS waits, boundedly, for the edge's TLS listener: the edge binds
+// it in its own goroutine, beside the plain-HTTP one the gate probes, so
+// it can come up a moment later. Not part of the gate's measurement.
+func (a *largeBoxAgent) awaitTLS() (largeBoxOutcome, error) {
+ deadline := time.Now().Add(largeBoxListenBound)
+ for {
+ c, err := net.DialTimeout("tcp", a.b.httpsAddr, time.Second)
+ if err == nil {
+ c.Close()
+ }
+ select {
+ case <-a.exited:
+ return largeBoxExited, fmt.Errorf("agent exited before its TLS listener was up: %v", a.exitErr)
+ default:
+ }
+ if err == nil {
+ return largeBoxUp, nil
+ }
+ if time.Now().After(deadline) {
+ return largeBoxLate, fmt.Errorf("TLS listener not up %s after the gate probe: %v", largeBoxListenBound, err)
+ }
+ time.Sleep(20 * time.Millisecond)
+ }
+}
+
+func (a *largeBoxAgent) signal(t *testing.T, sig os.Signal) {
+ t.Helper()
+ if err := a.cmd.Process.Signal(sig); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (a *largeBoxAgent) waitExit(t *testing.T, bound time.Duration) {
+ t.Helper()
+ select {
+ case <-a.exited:
+ if a.exitErr != nil {
+ t.Fatalf("agent exit: %v", a.exitErr)
+ }
+ case <-time.After(bound):
+ t.Fatalf("agent did not exit within %s of the signal", bound)
+ }
+}
+
+// assertNeverStarted: startup never finished, so the activator never
+// started and nothing reached the backend or ran.
+func (a *largeBoxAgent) assertNeverStarted(t *testing.T) {
+ t.Helper()
+ for _, kind := range []string{"ready", "activator-start", "acquire", "start", "worker-start", "served"} {
+ if evs := a.events(kind); len(evs) != 0 {
+ t.Errorf("%s with startup unfinished: %v", kind, evs)
+ }
+ }
+}
+
+// assertAfter waits briefly for the event kind with args and fails unless
+// it came after ref.
+func (a *largeBoxAgent) assertAfter(t *testing.T, ref largeBoxEvent, kind string, args ...string) {
+ t.Helper()
+ e := a.waitEvent(t, 10*time.Second, kind, args...)
+ if e.at.Before(ref.at) {
+ t.Errorf("%s %v came %s before %s", kind, args, ref.at.Sub(e.at), ref.kind)
+ }
+}
+
+// assertCleanShutdown reads the agent's own account of its shutdown: the
+// startup work joined within runAgent's bound, and no goroutine of the
+// startup sequence, the activator or the fake instances left after it.
+func (a *largeBoxAgent) assertCleanShutdown(t *testing.T) {
+ t.Helper()
+ j, ok := a.event("joined")
+ if !ok {
+ t.Fatal("agent did not report its startup join")
+ }
+ took, err := time.ParseDuration(j.arg(1))
+ if j.arg(0) != "true" || err != nil || took > largeBoxJoinBound {
+ t.Fatalf("startup join: %v", j)
+ }
+ t.Logf("startup work joined %s after the edge returned", took)
+ if _, ok := a.event("leak-free"); !ok {
+ t.Fatal("goroutines of the startup sequence outlived shutdown; see the agent log")
+ }
+}
+
+type largeBoxEvent struct {
+ at time.Time
+ kind string
+ args []string
+}
+
+func (e largeBoxEvent) arg(i int) string {
+ if i < len(e.args) {
+ return e.args[i]
+ }
+ return ""
+}
+
+func (a *largeBoxAgent) all() []largeBoxEvent {
+ f, err := os.Open(a.eventsPath)
+ if err != nil {
+ return nil
+ }
+ defer f.Close()
+ var out []largeBoxEvent
+ sc := bufio.NewScanner(f)
+ for sc.Scan() {
+ fields := strings.Fields(sc.Text())
+ if len(fields) < 2 {
+ continue
+ }
+ ns, err := strconv.ParseInt(fields[0], 10, 64)
+ if err != nil {
+ continue
+ }
+ out = append(out, largeBoxEvent{at: time.Unix(0, ns), kind: fields[1], args: fields[2:]})
+ }
+ return out
+}
+
+func (a *largeBoxAgent) events(kind string) []largeBoxEvent {
+ var out []largeBoxEvent
+ for _, e := range a.all() {
+ if e.kind == kind {
+ out = append(out, e)
+ }
+ }
+ return out
+}
+
+func (a *largeBoxAgent) event(kind string, args ...string) (largeBoxEvent, bool) {
+ for _, e := range a.events(kind) {
+ match := true
+ for i, want := range args {
+ if e.arg(i) != want {
+ match = false
+ }
+ }
+ if match {
+ return e, true
+ }
+ }
+ return largeBoxEvent{}, false
+}
+
+func (a *largeBoxAgent) waitEvent(t *testing.T, bound time.Duration, kind string, args ...string) largeBoxEvent {
+ t.Helper()
+ deadline := time.Now().Add(bound)
+ for {
+ if e, ok := a.event(kind, args...); ok {
+ return e
+ }
+ if time.Now().After(deadline) {
+ t.Fatalf("agent did not report %s %v within %s", kind, args, bound)
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+}
+
+// --- inside the agent process ---------------------------------------------------
+
+type largeBoxEvents struct {
+ mu sync.Mutex
+ f *os.File
+}
+
+func (e *largeBoxEvents) log(kind string, args ...any) {
+ var b strings.Builder
+ fmt.Fprintf(&b, "%d %s", time.Now().UnixNano(), kind)
+ for _, a := range args {
+ fmt.Fprintf(&b, " %v", a)
+ }
+ b.WriteByte('\n')
+ e.mu.Lock()
+ defer e.mu.Unlock()
+ _, _ = e.f.WriteString(b.String())
+}
+
+// runLargeBoxAgent is the agent: runAgent's legacy-box startup over the
+// fixture, then the edge, then runAgent's shutdown. It returns the exit
+// code.
+func runLargeBoxAgent(cfgPath string) int {
+ raw, err := os.ReadFile(cfgPath)
+ if err != nil {
+ log.Printf("largebox agent: %v", err)
+ return 2
+ }
+ var cfg largeBoxConfig
+ if err := json.Unmarshal(raw, &cfg); err != nil {
+ log.Printf("largebox agent: %v", err)
+ return 2
+ }
+ f, err := os.OpenFile(cfg.Events, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
+ if err != nil {
+ log.Printf("largebox agent: %v", err)
+ return 2
+ }
+ defer f.Close()
+ ev := &largeBoxEvents{f: f}
+
+ agentCtx, cancelAgent := context.WithCancel(context.Background())
+ defer cancelAgent()
+ startupCtx, cancelStartup := context.WithCancel(agentCtx)
+ defer cancelStartup()
+
+ dd := cfg.DataDir
+ regPath := filepath.Join(dd, "registry.db")
+ reg, err := registry.Open(regPath)
+ if err != nil {
+ log.Printf("largebox agent: %v", err)
+ return 2
+ }
+ defer reg.Close()
+
+ control := instancectl.NewClient(cfg.Control)
+ nestedOK := nestedGate(func() error { return instancefs.SelfCheck(dd) }, instancefs.MountNoSymfollow, control.Ready)
+ started := make(chan struct{})
+ backend := &largeBoxBackend{ev: ev, started: started}
+ svc := activator.New(activator.Config{Bucket: "unused", Idle: time.Hour}, backend, reg)
+ backend.svc = svc
+ svc.Routes = reg
+ svc.Startup = started
+ openStore := func(context.Context) (release.Store, error) { return release.NewDirStore(cfg.StoreDir), nil }
+
+ startupDone := activatorStart{
+ Ready: func() {
+ ev.log("ready") // before the close, so no acquire can be stamped earlier
+ close(started)
+ },
+ Sweep: func(ctx context.Context) error {
+ sw, err := control.Sweep(ctx)
+ if err != nil {
+ log.Printf("agent: instance sweep: %v (nested wakes refuse until the channel answers)", err)
+ } else if len(sw.Units) > 0 {
+ log.Printf("agent: swept %d instance unit(s) left from a previous agent", len(sw.Units))
+ }
+ ev.log("swept", len(sw.Units), err == nil)
+ return err
+ },
+ Recover: func(ctx context.Context, swept bool) *layout.Pass {
+ pass := newLayoutPass(reg, svc, dd, nestedOK, swept)
+ pass.SocketDir, pass.SocketsGID = cfg.SocketDir, -1 // the fixture's run dir; no carlos-sockets group
+ intent := startupRecoveryIntent(ctx, dd, openStore, "largebox")
+ recoverLayoutAtStart(ctx, pass, intent)
+ ev.log("recovered", intent)
+ return pass
+ },
+ Routes: reg.List,
+ Route: reg.Get,
+ Keys: largeBoxKeys{ev},
+ Slot: svc.PrepareScheduledToken,
+ Publish: armedLayoutPass.Store,
+ // The checkpoint: when Start runs, every hold marker's host must
+ // already be fenced again and every journal complete.
+ Start: func() {
+ held, _ := layout.HoldMarkers(dd)
+ fenced := 0
+ for _, h := range held {
+ if svc.FenceGeneration(h) != 0 {
+ fenced++
+ }
+ }
+ journals, _ := layout.JournalHosts(dd)
+ ev.log("activator-start", "held="+strconv.Itoa(len(held)), "fenced="+strconv.Itoa(fenced), "journals="+strconv.Itoa(len(journals)))
+ svc.Start()
+ },
+ }.Run(startupCtx)
+ go func() {
+ <-startupDone
+ ev.log("startup-done")
+ }()
+ // runAgent's worker boot pass, which waits on the same latch.
+ go func() {
+ select {
+ case <-started:
+ case <-startupCtx.Done():
+ return
+ }
+ routes, err := reg.List()
+ if err != nil {
+ return
+ }
+ runWorkerBootPass(routes, func(rt registry.Route) error {
+ err := svc.WakeWorker(rt)
+ ev.log("worker-boot", rt.Host, err == nil)
+ return err
+ })
+ }()
+
+ // Admissions from inside the agent that do not pass runAgent's own
+ // gates, so they meet the activator's latch on their own: a worker
+ // wake as an operator command or the sidecar-wake leg makes it, and
+ // one scheduler tick through the real runner and deliverer with a
+ // delivery already due.
+ if rt, err := reg.Get(largeBoxWorker); err == nil && rt != nil {
+ go func() {
+ ev.log("worker-direct-begin", rt.Host)
+ err := svc.WakeWorker(*rt)
+ ev.log("worker-direct", rt.Host, err == nil)
+ }()
+ }
+ runner := &schedule.Runner{
+ Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd, svc, reg),
+ Now: time.Now, Timeout: largeBoxLatchBound + 15*time.Second, Box: "largebox", Build: "test",
+ }
+ runner.Tick(agentCtx) // delivers in the background, and returns
+ ev.log("tick-done")
+
+ edge.RunWithActivatorReady([]string{"-registry", regPath, "-acme-cache", cfg.AcmeDir, "-http", cfg.HTTP, "-https", cfg.HTTPS},
+ largeBoxActivator{Service: svc, ev: ev}, edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes),
+ func(e *edge.Edge) {
+ reload := func() { _ = e.Reload() }
+ layoutReload.Store(&reload)
+ })
+
+ // runAgent's shutdown, in its order.
+ cancelStartup()
+ joinCtx, joinCancel := context.WithTimeout(context.Background(), largeBoxJoinBound)
+ began := time.Now()
+ joined := joinAgentStartup(joinCtx, startupDone)
+ joinCancel()
+ ev.log("joined", joined, time.Since(began))
+ ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+ svc.Shutdown(ctx)
+ cancel()
+ cancelAgent()
+
+ deadline := time.Now().Add(2 * time.Second)
+ for {
+ left := largeBoxStartupGoroutines()
+ if len(left) == 0 {
+ ev.log("leak-free")
+ return 0
+ }
+ if time.Now().After(deadline) {
+ log.Printf("largebox agent: %d goroutine(s) outlived shutdown:\n\n%s", len(left), strings.Join(left, "\n\n"))
+ ev.log("leak", len(left))
+ return 0
+ }
+ time.Sleep(20 * time.Millisecond)
+ }
+}
+
+// largeBoxStartupGoroutines lists every goroutine but this one that runs,
+// or was started by, the startup sequence, the layout pass, the control
+// client, the activator, the scheduler, or this agent's own fakes, and
+// every server connection still open (a request held at the latch is
+// answered only once the activator shuts down, after the edge has stopped
+// waiting for it). The edge's process-lifetime loops (its registry watch,
+// its SIGHUP reader) are not startup's and are left alone.
+func largeBoxStartupGoroutines() []string {
+ buf := make([]byte, 8<<20)
+ n := runtime.Stack(buf, true)
+ blocks := strings.Split(string(buf[:n]), "\n\n")
+ markers := []string{
+ "host.activatorStart", "host.startActivator", "host.backfillInstanceKeys", "host.keyBackfillInSlot",
+ "host.recoverLayoutAtStart", "host.startupRecoveryIntent", "host.runWorkerBootPass",
+ "host.runLargeBoxAgent", "host.(*largeBoxBackend)", "host.largeBoxActivator", "host.(*activatorDeliverer)",
+ "layout.(*Pass)", "instancectl.(*Client)", "activator.(*Service)", "schedule.(*Runner)",
+ "net/http.(*conn).serve",
+ }
+ var left []string
+ for _, g := range blocks[1:] { // [0] is the calling goroutine
+ for _, m := range markers {
+ if strings.Contains(g, m) {
+ left = append(left, g)
+ break
+ }
+ }
+ }
+ return left
+}
+
+// largeBoxActivator is the activator as the edge sees it, reporting each
+// instance request as it reaches it.
+type largeBoxActivator struct {
+ *activator.Service
+ ev *largeBoxEvents
+}
+
+func (a largeBoxActivator) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
+ if rt.Kind == "instance" {
+ a.ev.log("enter", rt.Host, rt.Socket)
+ }
+ return a.Service.ServeHTTP(w, r, rt, proxy)
+}
+
+// largeBoxBackend is activator.Backend with no S3 and no processes: a wake
+// serves an app on the row's socket that names the host and the socket it
+// was started at, and a worker reports ready at once, as a child does
+// over the agent socket. It records every acquire, start and request
+// served, and any acquire the startup latch should have prevented.
+type largeBoxBackend struct {
+ ev *largeBoxEvents
+ started <-chan struct{}
+ svc *activator.Service
+}
+
+func (b *largeBoxBackend) Acquire(_ context.Context, rt registry.Route) (activator.Lease, error) {
+ select {
+ case <-b.started:
+ default:
+ b.ev.log("violation", "acquire-before-ready", rt.Host)
+ }
+ b.ev.log("acquire", rt.Host)
+ return nopLease{}, nil
+}
+
+func (b *largeBoxBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil }
+func (b *largeBoxBackend) Seed(context.Context, registry.Route) error { return nil }
+
+func (b *largeBoxBackend) Start(_ context.Context, rt registry.Route) (activator.Proc, string, error) {
+ b.ev.log("start", rt.Host, rt.Socket)
+ if err := os.MkdirAll(filepath.Dir(rt.Socket), 0o750); err != nil {
+ return nil, "", err
+ }
+ ln, err := net.Listen("unix", rt.Socket)
+ if err != nil {
+ return nil, "", err
+ }
+ body := "host=" + rt.Host + " socket=" + rt.Socket
+ p := &largeBoxProc{done: make(chan struct{})}
+ p.srv = &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ b.ev.log("served", rt.Host, r.URL.Path)
+ io.WriteString(w, body)
+ })}
+ go func() { defer close(p.done); _ = p.srv.Serve(ln) }()
+ return p, "", nil
+}
+
+func (b *largeBoxBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil }
+func (b *largeBoxBackend) HasSidecar(registry.Route) bool { return false }
+func (b *largeBoxBackend) StartSidecar(context.Context, registry.Route) (activator.Proc, error) {
+ return nil, errors.New("largeBoxBackend: no co-spawned sidecars on this box")
+}
+
+func (b *largeBoxBackend) StartWorker(_ context.Context, rt registry.Route, _ map[string]string) (activator.Proc, string, error) {
+ b.ev.log("worker-start", rt.Host)
+ go b.svc.NoteWorkerReady(rt.Host, "")
+ return &largeBoxWorkerProc{}, "", nil
+}
+
+type largeBoxProc struct {
+ srv *http.Server
+ done chan struct{}
+ exited atomic.Bool
+}
+
+func (p *largeBoxProc) Stop(context.Context) error {
+ _ = p.srv.Close()
+ <-p.done
+ p.exited.Store(true)
+ return nil
+}
+
+func (p *largeBoxProc) Exited() bool { return p.exited.Load() }
+
+type largeBoxWorkerProc struct{ exited atomic.Bool }
+
+func (p *largeBoxWorkerProc) Stop(context.Context) error { p.exited.Store(true); return nil }
+func (p *largeBoxWorkerProc) Exited() bool { return p.exited.Load() }
+
+// largeBoxKeys is the exec backend's key writer as the key pass sees it:
+// every row it is handed already has its key.
+type largeBoxKeys struct{ ev *largeBoxEvents }
+
+func (k largeBoxKeys) BackfillInstanceKey(_ context.Context, rt registry.Route) (activator.InstanceKeyOutcome, error) {
+ k.ev.log("key", rt.Host)
+ return activator.InstanceKeyPresent, nil
+}