amadan

carlos / platform Public

zdd plan 1 task 2: exec backend spawns a generation on its own socket

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
claude a744e0c8b21d290b9251aeec058bba0555298844 parent bd5bcb5
5 files changed, +417 −29
  • internal/activator/backend_exec.go +103 −25
  • internal/activator/backend_unit.go +23 −0
  • internal/activator/generations.go +20 −0
  • internal/activator/generations_backend_test.go +259 −0
  • internal/activator/prepare.go +12 −4
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 4ae1ea1e7..1bedddea7 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -1063,6 +1063,14 @@ func (b *ExecBackend) waitSeedCgroupEmpty(ctx context.Context, rt registry.Route
// same env file the spawn path hands the child (the activator holds it either
// way).
func (b *ExecBackend) NextDue(ctx context.Context, rt registry.Route) (int64, error) {
+ return b.NextDueVia(ctx, rt, rt.Socket)
+}
+
+// NextDueVia is NextDue dialling socket instead of rt.Socket: the activator
+// passes the instance's current generation socket (AppSocket, zero-downtime
+// spec §7.1), because with generations on nothing listens on the canonical
+// path. With them off the activator passes rt.Socket, which is NextDue.
+func (b *ExecBackend) NextDueVia(ctx context.Context, rt registry.Route, socket string) (int64, error) {
paths := b.pathsFor(rt)
// A nested host's env file lives on tenant-writable ground (design
// §3.2), the same as instanceEnv's read of it: go through instancefs
@@ -1087,7 +1095,7 @@ func (b *ExecBackend) NextDue(ctx context.Context, rt registry.Route) (int64, er
return 0, err
}
req.Header.Set("Authorization", "Bearer "+token)
- resp, err := socketClient(rt.Socket).Do(req)
+ resp, err := socketClient(socket).Do(req)
if err != nil {
return 0, err
}
@@ -1131,17 +1139,87 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
p := b.pathsFor(rt)
switch p.layout {
case LayoutNested:
- return b.startUnit(ctx, rt)
+ return b.startUnitAt(ctx, rt, rt.Socket)
case LayoutFenced, LayoutConflict:
return nil, "", fmt.Errorf("%s: refusing to wake: %s", rt.Host, p.reason)
}
- return b.startChild(ctx, rt)
+ return b.startChildAt(ctx, rt, rt.Socket)
+}
+
+// StartGen is Start for generation n (zero-downtime spec §7.1). perGen false
+// is Start exactly: the canonical socket, the canonical control socket, the
+// same argv. perGen true binds the generation's own socket: GenSocketPath
+// for a flat row, app.<n>.sock in the instance run dir for a nested one. A
+// foreign row keeps its canonical socket either way: it is outside the two
+// layouts the spec gives generation paths to, and keeping it is the
+// conservative reading (plan 1, "Spec questions").
+//
+// The stale-file rule (spec §7.1): before the app binds, the file at THIS
+// generation's own path is removed (left there by an earlier agent process,
+// whose counter started at 1 like this one's). A sibling generation's path
+// is never touched.
+func (b *ExecBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) {
+ p := b.pathsFor(rt)
+ sock := rt.Socket
+ if perGen {
+ switch p.layout {
+ case LayoutFlat:
+ sock = b.flatGenSocket(rt.Host, n)
+ case LayoutNested:
+ sock = nestedGenSocketIn(p.runDir, n)
+ }
+ }
+ var (
+ proc Proc
+ ver string
+ err error
+ )
+ switch p.layout {
+ case LayoutNested:
+ // prepareAt removes the stale file at sock for the nested layout.
+ proc, ver, err = b.startUnitAt(ctx, rt, sock)
+ case LayoutFenced, LayoutConflict:
+ return nil, Gen{}, fmt.Errorf("%s: refusing to wake: %s", rt.Host, p.reason)
+ default:
+ if sock != rt.Socket {
+ if err := clearOwnGenSocket(sock); err != nil {
+ return nil, Gen{}, fmt.Errorf("%s: generation socket: %w", rt.Host, err)
+ }
+ }
+ proc, ver, err = b.startChildAt(ctx, rt, sock)
+ }
+ if err != nil {
+ return proc, Gen{}, err
+ }
+ return proc, Gen{N: n, Socket: sock, Control: sock + ".control", Build: ver}, nil
+}
+
+// flatGenSocket is GenSocketPath, under RunDirOverride/g in a test.
+func (b *ExecBackend) flatGenSocket(host string, n uint64) string {
+ if b.RunDirOverride != "" {
+ return genSocketPathIn(filepath.Join(b.RunDirOverride, "g"), host, n)
+ }
+ return GenSocketPath(host, n)
+}
+
+// clearOwnGenSocket makes the flat generation directory and removes a stale
+// file at sock, and only at sock. The directory inherits /run/carlos's
+// setgid carlos-sockets group, as the nested run dirs do.
+func clearOwnGenSocket(sock string) error {
+ if err := os.MkdirAll(filepath.Dir(sock), 0o750|os.ModeSetgid); err != nil {
+ return err
+ }
+ if err := os.Remove(sock); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ return nil
}
-// startChild is Start's exec-child body (flat and foreign layouts):
+// startChildAt is Start's exec-child body (flat and foreign layouts):
// exec of the versioned binary directly, exactly as before nested layouts
-// existed.
-func (b *ExecBackend) startChild(ctx context.Context, rt registry.Route) (Proc, string, error) {
+// existed. sock is the app socket this spawn binds: rt.Socket, or a
+// generation socket (StartGen).
+func (b *ExecBackend) startChildAt(ctx context.Context, rt registry.Route, sock string) (Proc, string, error) {
host := rt.Host
paths := b.pathsFor(rt)
dbPath := paths.dbPath
@@ -1176,7 +1254,7 @@ func (b *ExecBackend) startChild(ctx context.Context, rt registry.Route) (Proc,
// 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.
- controlSocket := rt.Socket + ".control"
+ controlSocket := sock + ".control"
// Bind the control channel BEFORE the spawn, so CARLOS_CONTROL_SOCKET can
// follow the OUTCOME rather than the intention. It used to be set from
// "b.Cache != nil" and the listener started afterwards; now that the
@@ -1240,7 +1318,7 @@ func (b *ExecBackend) startChild(ctx context.Context, rt registry.Route) (Proc,
// the rollout ordering harmless. It does not replace the socket channel;
// both are live for exec-backed instances, and converging them is deferred
// (spec §10.3).
- childVars := b.agentOwnedVars(rt, fileEnv, ctl != nil)
+ childVars := b.agentOwnedVars(rt, controlSocket, fileEnv, ctl != nil)
// Record the build we're about to exec — read the symlink immediately before
// the start so it names the version the kernel will resolve. Reading it here,
// rather than back in wake() after the health wait, closes a real window: a
@@ -1258,7 +1336,7 @@ func (b *ExecBackend) startChild(ctx context.Context, rt registry.Route) (Proc,
// twice — the manager's clone3-refusal fallback — so everything about
// the Cmd is configured inside it.
cmd, settle, err := b.spawnInCgroup(rt, func() *exec.Cmd {
- cmd := exec.Command(bin, "--socket", rt.Socket, "--db", dbPath)
+ cmd := exec.Command(bin, "--socket", sock, "--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
@@ -1308,13 +1386,13 @@ func (b *ExecBackend) startChild(ctx context.Context, rt registry.Route) (Proc,
close(p.done)
}()
- if err := waitHealthyCtx(ctx, rt.Socket, spawnHealthTimeout); err != nil {
+ if err := waitHealthyCtx(ctx, sock, spawnHealthTimeout); err != nil {
stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
defer cancel()
_ = p.Stop(stopCtx)
return nil, "", err
}
- b.normalizeSocketMode(rt, p)
+ b.normalizeSocketMode(rt, sock, p)
return p, spawnVer, nil
}
@@ -1325,10 +1403,10 @@ func (b *ExecBackend) startChild(ctx context.Context, rt registry.Route) (Proc,
// (design §7.2): haveControl is whether this call's control listener
// actually bound. childVars must follow the OUTCOME of the bind, not the
// intention, per the long comment this replaced, still true here.
-func (b *ExecBackend) agentOwnedVars(rt registry.Route, fileEnv map[string]string, haveControl bool) map[string]string {
+func (b *ExecBackend) agentOwnedVars(rt registry.Route, control string, fileEnv map[string]string, haveControl bool) map[string]string {
childVars := fileEnv
if haveControl {
- childVars = withVar(childVars, "CARLOS_CONTROL_SOCKET", rt.Socket+".control")
+ childVars = withVar(childVars, "CARLOS_CONTROL_SOCKET", control)
} else {
childVars = withoutVars(childVars, "CARLOS_CONTROL_SOCKET")
}
@@ -1369,30 +1447,30 @@ func (b *ExecBackend) agentOwnedVars(rt registry.Route, fileEnv map[string]strin
// anything. All three branches log and never fail the wake: a box with
// no consumers must not lose a healthy wake to a normalization it
// doesn't need.
-func (b *ExecBackend) normalizeSocketMode(rt registry.Route, proc Proc) {
+func (b *ExecBackend) normalizeSocketMode(rt registry.Route, sock string, proc Proc) {
host := rt.Host
if proc.Exited() {
log.Printf("activator: %s: child exited during the health wait (external process serving its socket?) — not normalizing the socket mode", host)
- } else if fi, statErr := os.Lstat(rt.Socket); statErr != nil || fi.Mode()&os.ModeSocket == 0 {
- log.Printf("activator: %s: %s is not a socket we can normalize (stat: %v)", host, rt.Socket, statErr)
- } else if err := os.Chmod(rt.Socket, 0o660); err != nil {
+ } else if fi, statErr := os.Lstat(sock); statErr != nil || fi.Mode()&os.ModeSocket == 0 {
+ log.Printf("activator: %s: %s is not a socket we can normalize (stat: %v)", host, sock, statErr)
+ } else if err := os.Chmod(sock, 0o660); err != nil {
log.Printf("activator: normalize socket mode for %s: %v (enrolled consumers would get permission denied)", host, err)
}
}
-// startUnit is Start's nested body (design §6): the same admin token, env
+// startUnitAt is Start's nested body (design §6): the same admin token, env
// composition and health gate as startChild, but the process itself is
// started by the root instance-control channel from a launch file rather
// than exec'd directly. The agent has no privilege to enter the tenant's
// namespace, only the control service does.
-func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, string, error) {
+func (b *ExecBackend) startUnitAt(ctx context.Context, rt registry.Route, sock string) (Proc, string, error) {
if err := b.nestedGate(rt); err != nil {
return nil, "", err
}
if err := b.ensureLiveLink(ctx, rt.Host); err != nil {
return nil, "", err
}
- p, ctl, logs, fileEnv, err := b.prepare(rt, instancectl.RoleRun)
+ p, ctl, logs, fileEnv, err := b.prepareAt(rt, instancectl.RoleRun, sock)
if err != nil {
return nil, "", err
}
@@ -1403,8 +1481,8 @@ func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, s
logs.Close()
return nil, "", err
}
- childVars := b.agentOwnedVars(rt, fileEnv, ctl != nil)
- argv := []string{b.liveLink(rt.Host), "--socket", rt.Socket, "--db", p.dbPath}
+ childVars := b.agentOwnedVars(rt, sock+".control", fileEnv, ctl != nil)
+ argv := []string{b.liveLink(rt.Host), "--socket", sock, "--db", p.dbPath}
if err := b.writeLaunch(p, instancectl.RoleRun, rt.Host, argv, childEnv(childVars)); err != nil {
return fail(err)
}
@@ -1431,7 +1509,7 @@ func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, s
log.Printf("activator: purged %d cached response(s) for %s on spawn", n, rt.Host)
}
}
- if err := waitHealthyCtx(ctx, rt.Socket, spawnHealthTimeout); err != nil {
+ if err := waitHealthyCtx(ctx, sock, spawnHealthTimeout); err != nil {
// Set before the Stop call below, closing the same race Stop's own
// internal set leaves open by a few instructions: observe runs
// concurrently from the moment newUnitProc started it, and setting
@@ -1452,7 +1530,7 @@ func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, s
}
return nil, "", err
}
- b.normalizeSocketMode(rt, up)
+ b.normalizeSocketMode(rt, sock, up)
return up, spawnVer, nil
}
@@ -1552,7 +1630,7 @@ func (b *ExecBackend) LaunchEnvironment(rt registry.Route) (map[string]string, [
sort.Strings(shadowed)
fileEnv := mergeEnv(cfgEnv, ownEnv)
out := map[string]string{}
- for _, kv := range childEnv(b.agentOwnedVars(rt, fileEnv, true)) {
+ for _, kv := range childEnv(b.agentOwnedVars(rt, rt.Socket+".control", fileEnv, true)) {
if i := strings.IndexByte(kv, '='); i > 0 {
out[kv[:i]] = kv[i+1:]
}
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go
index 137f63de0..9beabd4a5 100644
--- a/internal/activator/backend_unit.go
+++ b/internal/activator/backend_unit.go
@@ -322,6 +322,29 @@ func (b *SplitBackend) NextDue(ctx context.Context, rt registry.Route) (int64, e
}
func (b *SplitBackend) HasSidecar(rt registry.Route) bool { return b.pick(rt).HasSidecar(rt) }
+// StartGen dispatches like Start. A side without generation support (the
+// unit side: systemd binds a unit tenant's canonical socket) is started
+// with Start and gets the canonical handle.
+func (b *SplitBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) {
+ if g, ok := b.pick(rt).(GenBackend); ok {
+ return g.StartGen(ctx, rt, n, perGen)
+ }
+ proc, ver, err := b.pick(rt).Start(ctx, rt)
+ if err != nil {
+ return proc, Gen{}, err
+ }
+ return proc, canonicalGen(rt, n, ver), nil
+}
+
+// NextDueVia dispatches like NextDue; a side without generation support
+// answers on rt's own socket.
+func (b *SplitBackend) NextDueVia(ctx context.Context, rt registry.Route, socket string) (int64, error) {
+ if g, ok := b.pick(rt).(GenBackend); ok {
+ return g.NextDueVia(ctx, rt, socket)
+ }
+ return b.pick(rt).NextDue(ctx, rt)
+}
+
// Layout implements Classifier for the composed backend: a unit-backed row
// is always today's flat unit shape (the exec-only layouts do not apply to
// it), and an exec row defers to the exec side's own classification.
diff --git a/internal/activator/generations.go b/internal/activator/generations.go
index 15c83b055..2de681242 100644
--- a/internal/activator/generations.go
+++ b/internal/activator/generations.go
@@ -5,6 +5,7 @@
package activator
import (
+ "context"
"crypto/sha256"
"encoding/hex"
"path/filepath"
@@ -98,3 +99,22 @@ func canonicalGen(rt registry.Route, n uint64, build string) Gen {
}
return g
}
+
+// GenBackend is the optional Backend capability behind generations (spec
+// §7.1). ExecBackend and SplitBackend implement it. A backend that does not
+// (UnitBackend, every test double that predates this plan) is spawned with
+// Start and gets canonicalGen: one generation per entry, on the canonical
+// socket.
+type GenBackend interface {
+ // StartGen spawns generation n of rt's instance and returns its handle.
+ // perGen false must behave exactly as Start; true binds the generation's
+ // own socket where the layout has one.
+ StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error)
+ // NextDueVia is NextDue dialling socket rather than rt.Socket.
+ NextDueVia(ctx context.Context, rt registry.Route, socket string) (int64, error)
+}
+
+var (
+ _ GenBackend = (*ExecBackend)(nil)
+ _ GenBackend = (*SplitBackend)(nil)
+)
diff --git a/internal/activator/generations_backend_test.go b/internal/activator/generations_backend_test.go
new file mode 100644
index 000000000..d1cec5b28
--- /dev/null
+++ b/internal/activator/generations_backend_test.go
@@ -0,0 +1,259 @@
+package activator
+
+import (
+ "context"
+ "encoding/json"
+ "net"
+ "net/http"
+ "os"
+ "path/filepath"
+ "slices"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// shortRunDir is a RunDirOverride short enough for sockets under it: the
+// test name is not part of the path (t.TempDir's would be), so a long test
+// name cannot push a socket past sockaddr_un's limit. Run socket tests with
+// TMPDIR=/tmp/k.
+func shortRunDir(t *testing.T) string {
+ t.Helper()
+ dir, err := os.MkdirTemp("", "g")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ return dir
+}
+
+// flatGenRoute is a route that classifies as FLAT against b: the default
+// db path and <RunDirOverride>/<host>.sock.
+func flatGenRoute(b *ExecBackend, host string) registry.Route {
+ return registry.Route{Host: host, Kind: "instance", Hibernate: true,
+ DBPath: provision.DefaultDBPath(b.DataDir, host),
+ Socket: filepath.Join(b.RunDirOverride, host+".sock")}
+}
+
+func isSocket(path string) bool {
+ fi, err := os.Lstat(path)
+ return err == nil && fi.Mode()&os.ModeSocket != 0
+}
+
+func isRegular(path string) bool {
+ fi, err := os.Lstat(path)
+ return err == nil && fi.Mode().IsRegular()
+}
+
+// Flag off: StartGen(perGen=false) is Start, byte for byte. The child gets
+// the canonical socket in argv and the canonical control socket in env.
+func TestStartGenWithGenerationsOffIsStart(t *testing.T) {
+ b := testExecBackend(t)
+ host := "gen-off.example.com"
+ argvPath, envPath := stageRecordingLive(t, b, host, "v1")
+ rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(shortRunDir(t), "instance.sock")}
+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+ defer cancel()
+ done := make(chan struct{})
+ var proc Proc
+ go func() {
+ proc, _, _ = b.StartGen(ctx, rt, 5, false) // the recorder never binds; a health error is expected
+ close(done)
+ }()
+ 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, host+".db")}
+ if !slices.Equal(gotArgv, wantArgv) {
+ t.Fatalf("argv = %v, want %v (today's argv, unchanged)", gotArgv, wantArgv)
+ }
+ if gotEnv != rt.Socket+".control" {
+ t.Fatalf("CARLOS_CONTROL_SOCKET = %q, want %q", gotEnv, rt.Socket+".control")
+ }
+}
+
+// Flag off, healthy: the handle names the canonical socket the app bound.
+func TestStartGenWithGenerationsOffBindsTheCanonicalSocket(t *testing.T) {
+ requireUnixSockets(t)
+ b := testExecBackend(t)
+ b.RunDirOverride = shortRunDir(t)
+ host := "canon.example.com"
+ stageServingLive(t, b, host)
+ rt := flatGenRoute(b, host)
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ proc, g, err := b.StartGen(ctx, rt, 11, false)
+ if err != nil {
+ t.Fatalf("StartGen: %v", err)
+ }
+ defer proc.Stop(context.Background())
+ if g.N != 11 || g.Socket != rt.Socket || g.Control != rt.Socket+".control" {
+ t.Fatalf("handle = %+v, want the canonical socket and control", g)
+ }
+ if !isSocket(rt.Socket) {
+ t.Fatalf("%s is not a bound socket", rt.Socket)
+ }
+ if _, err := os.Stat(filepath.Join(b.RunDirOverride, "g")); !os.IsNotExist(err) {
+ t.Fatalf("the generation directory exists with generations off (stat: %v)", err)
+ }
+}
+
+// Flag on, flat: a 120-byte host binds the bounded generation socket and
+// its control socket, and only the generation's OWN stale file is removed.
+// The canonical path for this host would not even fit a socket address.
+func TestStartGenFlatBindsItsOwnGenerationSocket(t *testing.T) {
+ requireUnixSockets(t)
+ b := testExecBackend(t)
+ b.RunDirOverride = shortRunDir(t)
+ host := strings.Repeat("b", 113) + ".bab.io" // 120 bytes
+ stageServingLive(t, b, host)
+ rt := flatGenRoute(b, host)
+ if l, _ := b.Layout(rt); l != LayoutFlat {
+ t.Fatalf("fixture classifies %q, want flat", l)
+ }
+ own := genSocketPathIn(filepath.Join(b.RunDirOverride, "g"), host, 9)
+ sibling := genSocketPathIn(filepath.Join(b.RunDirOverride, "g"), host, 8)
+ if err := os.MkdirAll(filepath.Dir(own), 0o750); err != nil {
+ t.Fatal(err)
+ }
+ for _, p := range []string{own, sibling} {
+ if err := os.WriteFile(p, []byte("stale"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ proc, g, err := b.StartGen(ctx, rt, 9, true)
+ if err != nil {
+ t.Fatalf("StartGen: %v", err)
+ }
+ defer proc.Stop(context.Background())
+ if g.N != 9 || g.Socket != own || g.Control != own+".control" {
+ t.Fatalf("handle = %+v, want socket %s", g, own)
+ }
+ if !isSocket(own) {
+ t.Fatalf("own generation path %s is not a bound socket: the stale file was not removed", own)
+ }
+ if !isSocket(g.Control) {
+ t.Fatalf("control socket %s was not bound", g.Control)
+ }
+ if !isRegular(sibling) {
+ t.Fatalf("sibling generation's file %s was touched; only the own path may be removed", sibling)
+ }
+ if _, err := os.Lstat(rt.Socket); !os.IsNotExist(err) {
+ t.Fatalf("canonical path %s exists with generations on (lstat: %v)", rt.Socket, err)
+ }
+}
+
+// Flag on, nested: the launch file names app.<n>.sock, the control env
+// follows it, and prepare removes only that generation's stale file.
+func TestStartGenNestedUsesAppNSock(t *testing.T) {
+ ctl := &fakeControl{inv: "inv1"}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ runDir := filepath.Dir(rt.Socket)
+ own := filepath.Join(runDir, "app.9.sock")
+ sibling := filepath.Join(runDir, "app.8.sock")
+ if err := os.MkdirAll(runDir, 0o750); err != nil {
+ t.Fatal(err)
+ }
+ for _, p := range []string{rt.Socket, sibling, own} {
+ if err := os.WriteFile(p, []byte("stale"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ var stopServing func()
+ ctl.onStart = func() { stopServing = serveOK(t, own) }
+ defer func() {
+ if stopServing != nil {
+ stopServing()
+ }
+ }()
+ unit.set("active", "inv1", true, true)
+ p, g, err := b.StartGen(context.Background(), rt, 9, true)
+ if err != nil {
+ t.Fatalf("StartGen: %v", err)
+ }
+ if g.Socket != own || g.Control != own+".control" {
+ t.Fatalf("handle = %+v, want %s", g, own)
+ }
+ raw, err := os.ReadFile(filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), "launch.run.json"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var l instancectl.Launch
+ if err := json.Unmarshal(raw, &l); err != nil {
+ t.Fatal(err)
+ }
+ if len(l.Argv) != 5 || l.Argv[1] != "--socket" || l.Argv[2] != own {
+ t.Fatalf("launch argv = %v, want --socket %s", l.Argv, own)
+ }
+ if !slices.Contains(l.Env, "CARLOS_CONTROL_SOCKET="+own+".control") {
+ t.Fatalf("launch env lacks CARLOS_CONTROL_SOCKET=%s.control: %v", own, l.Env)
+ }
+ if !isRegular(rt.Socket) || !isRegular(sibling) {
+ t.Fatal("prepare removed a file that is not this generation's own socket")
+ }
+ unit.set("inactive", "inv1", false, false)
+ waitFor(t, "the unit to exit", func() bool { return p.Exited() })
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ _ = p.Stop(ctx)
+}
+
+// NextDueVia dials the socket it is given, not rt.Socket.
+func TestNextDueViaDialsTheGivenSocket(t *testing.T) {
+ requireUnixSockets(t)
+ b := testExecBackend(t)
+ host := "due.example.com"
+ if err := os.WriteFile(filepath.Join(b.DataDir, host+".env"), []byte("CARLOS_ADMIN_TOKEN=tok\n"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ sock := tmpSock(t)
+ ln, err := net.Listen("unix", sock)
+ if err != nil {
+ t.Fatal(err)
+ }
+ srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path != "/api/next-due" || r.Header.Get("Authorization") != "Bearer tok" {
+ http.Error(w, "no", http.StatusForbidden)
+ return
+ }
+ w.Write([]byte(`{"due":42}`))
+ })}
+ go srv.Serve(ln)
+ defer srv.Close()
+ rt := registry.Route{Host: host, Kind: "instance", Socket: "/nonexistent/canonical.sock"}
+ due, err := b.NextDueVia(context.Background(), rt, sock)
+ if err != nil || due != 42 {
+ t.Fatalf("NextDueVia = %d, %v; want 42 from the generation socket", due, err)
+ }
+ if _, err := b.NextDue(context.Background(), rt); err == nil {
+ t.Fatal("NextDue reached something at the canonical path; the fixture has nothing there")
+ }
+}
+
+// A unit-backed route has no generation sockets: SplitBackend starts it
+// with Start and hands back the canonical handle, whatever perGen says.
+func TestSplitBackendStartGenKeepsUnitTenantsCanonical(t *testing.T) {
+ unitSide := &fakeBackend{execVerFn: func() string { return "v3" }}
+ sb := &SplitBackend{Exec: &fakeBackend{}, Unit: unitSide}
+ rt := registry.Route{Host: "u.example.com", Kind: "instance", Backing: registry.BackingUnit, Socket: "/run/carlos/u.example.com.sock"}
+ _, g, err := sb.StartGen(context.Background(), rt, 4, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if g != canonicalGen(rt, 4, "v3") {
+ t.Fatalf("handle = %+v, want the canonical one", g)
+ }
+ if unitSide.starts != 1 {
+ t.Fatalf("unit side Start calls = %d, want 1", unitSide.starts)
+ }
+}
diff --git a/internal/activator/prepare.go b/internal/activator/prepare.go
index e3a9f4205..f8ac05fcb 100644
--- a/internal/activator/prepare.go
+++ b/internal/activator/prepare.go
@@ -31,6 +31,14 @@ import (
// and stale records cleared. It returns the listeners so the caller can
// hand them to the proc or close them on a failed start.
func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instancePaths, *edge.ControlListener, *logListener, map[string]string, error) {
+ return b.prepareAt(rt, role, rt.Socket)
+}
+
+// prepareAt is prepare for a run role whose app socket is sock: rt.Socket,
+// or a generation socket (StartGen, zero-downtime spec §7.1). Only the
+// file at sock is removed, never a sibling generation's, and the control
+// listener binds sock + ".control". The seed and sidecar roles ignore sock.
+func (b *ExecBackend) prepareAt(rt registry.Route, role instancectl.Role, sock string) (instancePaths, *edge.ControlListener, *logListener, map[string]string, error) {
p := b.pathsFor(rt)
if err := os.MkdirAll(p.dataDir, 0o700); err != nil {
return p, nil, nil, nil, err
@@ -88,8 +96,8 @@ func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instanc
// no member-facing diagnosis at all, unlike the flat path's #129
// handling, which at least turns the bind failure into a named
// error (memberSafeWakeReason).
- _ = os.Remove(rt.Socket)
- ctl = b.listenControl(rt, p.envFile)
+ _ = os.Remove(sock)
+ ctl = b.listenControl(rt, sock+".control", p.envFile)
}
logs, err := listenLogs(b.logSocket(rt.Host, role), b.logEmitter(rt))
if err != nil {
@@ -106,7 +114,7 @@ func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instanc
// reading beneath the instance directory rather than following a plain
// path, design §3.2); a bind failure is logged and the instance runs
// without its control verbs, as today.
-func (b *ExecBackend) listenControl(rt registry.Route, envPath string) *edge.ControlListener {
+func (b *ExecBackend) listenControl(rt registry.Route, control, envPath string) *edge.ControlListener {
tokenReader := func() (string, error) {
env, err := readEnvFileBeneath(filepath.Dir(envPath), filepath.Base(envPath))
if err != nil {
@@ -118,7 +126,7 @@ func (b *ExecBackend) listenControl(rt registry.Route, envPath string) *edge.Con
if b.Timers != nil {
mount = func(mux *http.ServeMux) { schedule.Mount(mux, rt.Host, b.Timers, tokenReader, time.Now) }
}
- ctl, err := edge.ListenControl(rt.Socket+".control", rt.Host, b.Cache, mount)
+ ctl, err := edge.ListenControl(control, rt.Host, b.Cache, mount)
if err != nil {
log.Printf("activator: control listener for %s: %v (continuing without it)", rt.Host, err)
return nil