amadan

carlos / platform Public

activator: StartCmd, cmdVars and the two cmd streams (run spec §4, §5.4, §5.5)

Drain takes a connection still queued in the listener backlog, so the
command's last bytes reach the run's writer before StopAndSettle returns.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 056fe554157043cd10f739b080d0a5c8e40bd212 parent 71135ab
7 files changed, +537 −9
  • internal/activator/backend_exec.go +11 −1
  • internal/activator/backend_unit.go +9 −0
  • internal/activator/cmdunit.go +282 −0
  • internal/activator/cmdunit_test.go +199 −0
  • internal/activator/control.go +4 −0
  • internal/activator/prepare.go +20 −8
  • internal/activator/unitproc_test.go +12 −0
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 638c1d616..54b7655b6 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -168,6 +168,10 @@ type ExecBackend struct {
// nested run dir and log socket beneath it, so a test can exercise the
// nested layout without writing to /run/carlos.
RunDirOverride string
+
+ // layoutFn, when set, replaces Classify in StartCmd's layout check; a
+ // unit test's way to stand a temp directory in for a nested host.
+ layoutFn func(rt registry.Route) (Layout, string)
// Strict, when non-nil and true for a host, makes every spawn of that
// host strict (moves design §5.3): the child starts only inside a group
// prepareStrict prepared, which is drained (KillAndEmpty) and closed
@@ -1027,6 +1031,12 @@ func (b *ExecBackend) seedUnitFailed(ctx context.Context, rt registry.Route, dbP
// caller as a log line, not a failure: it is the deciding result's
// confidence that is at stake here, not the wake itself.
func (b *ExecBackend) waitSeedCgroupEmpty(ctx context.Context, rt registry.Route) error {
+ return b.waitRoleCgroupEmpty(ctx, rt, instancectl.RoleSeed)
+}
+
+// waitRoleCgroupEmpty polls role's unit's cgroup.events until it reports
+// empty or actStopTimeout elapses, honouring ctx and b.ObservePoll.
+func (b *ExecBackend) waitRoleCgroupEmpty(ctx context.Context, rt registry.Route, role instancectl.Role) error {
events := b.CgroupEvents
if events == nil {
events = cgroupEventsUnder(b.cgroupRoot())
@@ -1036,7 +1046,7 @@ func (b *ExecBackend) waitSeedCgroupEmpty(ctx context.Context, rt registry.Route
poll = time.Second
}
slice := instancectl.SliceName(rt.Host)
- unit := instancectl.UnitName(instancectl.RoleSeed, rt.Host)
+ unit := instancectl.UnitName(role, rt.Host)
deadline := time.Now().Add(actStopTimeout)
for {
exists, populated, err := events(slice, unit)
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go
index 137f63de0..19535c0a5 100644
--- a/internal/activator/backend_unit.go
+++ b/internal/activator/backend_unit.go
@@ -342,6 +342,15 @@ func (b *SplitBackend) StartWorker(ctx context.Context, rt registry.Route, env m
return b.pick(rt).StartWorker(ctx, rt, env)
}
+// StartCmd dispatches like every other call; a unit-backed row has no
+// nested instance to run a command in.
+func (b *SplitBackend) StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error) {
+ if c, ok := b.pick(rt).(CmdBackend); ok {
+ return c.StartCmd(ctx, rt, spec)
+ }
+ return nil, fmt.Errorf("%w: %s is unit-backed", ErrCmdNotNested, rt.Host)
+}
+
// LiveVersion forwards to the exec side's reader when it has one — both
// backings share the live/<host> symlink layout, so one reader serves the
// convergence sweep for every host. (New picks this up via the optional
diff --git a/internal/activator/cmdunit.go b/internal/activator/cmdunit.go
new file mode 100644
index 000000000..da39ce66b
--- /dev/null
+++ b/internal/activator/cmdunit.go
@@ -0,0 +1,282 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "log"
+ "net"
+ "path/filepath"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// CmdRunDirName is the directory under an instance's own data directory
+// that holds each run's --file uploads (run spec §5.2 step 3).
+const CmdRunDirName = ".carlos-run"
+
+// cmdErrSocket is the stderr listener's name; stdout's is the role's.
+const cmdErrSocket = "cmd-err"
+
+// cmdDrainBound is how long a stream may take to reach EOF once its unit
+// has stopped.
+const cmdDrainBound = 5 * time.Second
+
+// ErrCmdNotNested: the instance is not on the nested layout, so a command
+// would run with the agent's reach (run spec §5.2 step 1, refused "flat").
+var ErrCmdNotNested = errors.New("activator: the instance is not on the nested layout")
+
+// CmdSpec is one run's launch, as the box's run pass hands it over.
+type CmdSpec struct {
+ RunID string
+ // Args is exactly what followed `--`; the binary is the host's live link.
+ Args []string
+ // RunDir is CARLOS_RUN_DIR: <instance dir>/.carlos-run/<RunID>, made and
+ // filled by the caller.
+ RunDir string
+ // Stdout and Stderr receive the unit's two streams, byte for byte.
+ Stdout, Stderr io.Writer
+ // Prepare runs after the run is attached to the awake entry and before
+ // the unit starts (Service.RunCmd); an error ends the run before launch.
+ Prepare func(ctx context.Context) error
+}
+
+// CmdUnit is a started cmd unit.
+type CmdUnit interface {
+ Invocation() string
+ // Poll long-polls the invocation, at most about 30 seconds.
+ Poll(ctx context.Context) (instancectl.CmdStatus, error)
+ // StopAndSettle stops the unit (SIGTERM, then SIGKILL after 20 s),
+ // proves its cgroup empty and reads both streams to their end. It is
+ // safe to call more than once; later calls return the first's error.
+ StopAndSettle(ctx context.Context) error
+}
+
+// CmdBackend is the backend capability Service.RunCmd needs.
+type CmdBackend interface {
+ StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error)
+}
+
+var _ CmdBackend = (*ExecBackend)(nil)
+
+// cmdVars is a run's environment over fileEnv (run spec §4, codex 7): the
+// control-channel variables stripped as seedVars strips them, CARLOS_REGION
+// set or removed from the backend's region as agentOwnedVars does,
+// CARLOS_HOST from the row, and the three run variables. fileEnv carries
+// CARLOS_ADMIN_TOKEN through.
+func (b *ExecBackend) cmdVars(rt registry.Route, fileEnv map[string]string, dbPath, runID, runDir string) map[string]string {
+ v := withoutVars(fileEnv, "CARLOS_CONTROL_SOCKET", "CARLOS_PURGE")
+ if b.Region != "" {
+ v = withVar(v, "CARLOS_REGION", b.Region)
+ } else {
+ v = withoutVars(v, "CARLOS_REGION")
+ }
+ v = withVar(v, "CARLOS_HOST", rt.Host)
+ v = withVar(v, "CARLOS_DB", dbPath)
+ v = withVar(v, "CARLOS_RUN_ID", runID)
+ return withVar(v, "CARLOS_RUN_DIR", runDir)
+}
+
+func (b *ExecBackend) cmdLayout(rt registry.Route) (Layout, string) {
+ if b.layoutFn != nil {
+ return b.layoutFn(rt)
+ }
+ return b.Layout(rt)
+}
+
+// StartCmd starts one run's cmd unit (run spec §5.4): the layout check, the
+// nested gate, the live link, the instance's directories and environment,
+// the two stream listeners, launch.cmd.json, then POST /cmd. It returns
+// once the unit has an invocation. The environment is never logged.
+func (b *ExecBackend) StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error) {
+ if l, why := b.cmdLayout(rt); l != LayoutNested {
+ return nil, fmt.Errorf("%w: %s is %s (%s)", ErrCmdNotNested, rt.Host, l, why)
+ }
+ if err := b.nestedGate(rt); err != nil {
+ return nil, err
+ }
+ if err := b.ensureLiveLink(ctx, rt.Host); err != nil {
+ return nil, err
+ }
+ p, _, _, fileEnv, err := b.prepare(rt, instancectl.RoleCmd)
+ if err != nil {
+ return nil, err
+ }
+ if want := filepath.Join(p.dataDir, CmdRunDirName, spec.RunID); spec.RunDir != want {
+ return nil, fmt.Errorf("%s: run directory %q is not %q", rt.Host, spec.RunDir, want)
+ }
+ out, err := listenCmdStream(b.logSocketNamed(rt.Host, string(instancectl.RoleCmd)), spec.Stdout)
+ if err != nil {
+ return nil, fmt.Errorf("stdout listener for %s: %w", rt.Host, err)
+ }
+ errs, err := listenCmdStream(b.logSocketNamed(rt.Host, cmdErrSocket), spec.Stderr)
+ if err != nil {
+ out.Close()
+ return nil, fmt.Errorf("stderr listener for %s: %w", rt.Host, err)
+ }
+ fail := func(err error) (CmdUnit, error) {
+ out.Close()
+ errs.Close()
+ return nil, err
+ }
+ argv := append([]string{b.liveLink(rt.Host)}, spec.Args...)
+ env := childEnv(b.cmdVars(rt, fileEnv, p.dbPath, spec.RunID, spec.RunDir))
+ if err := b.writeLaunch(p, instancectl.RoleCmd, rt.Host, argv, env); err != nil {
+ return fail(err)
+ }
+ inv, err := b.Control.Cmd(ctx, rt.Host, b.tier(rt).MemoryMax)
+ if err != nil {
+ return fail(fmt.Errorf("%s: start cmd unit: %w", rt.Host, err))
+ }
+ log.Printf("activator: %s: started %s for run %s", rt.Host, instancectl.UnitName(instancectl.RoleCmd, rt.Host), spec.RunID)
+ return &cmdUnit{b: b, rt: rt, inv: inv, out: out, errs: errs}, nil
+}
+
+type cmdUnit struct {
+ b *ExecBackend
+ rt registry.Route
+ inv string
+ out, errs *cmdStream
+ once sync.Once
+ err error
+}
+
+func (u *cmdUnit) Invocation() string { return u.inv }
+
+func (u *cmdUnit) Poll(ctx context.Context) (instancectl.CmdStatus, error) {
+ return u.b.Control.CmdStatus(ctx, u.rt.Host, u.inv)
+}
+
+func (u *cmdUnit) StopAndSettle(ctx context.Context) error {
+ u.once.Do(func() {
+ sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl)
+ stopErr := u.b.Control.Stop(sctx, u.rt.Host, instancectl.RoleCmd)
+ cancel()
+ wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout)
+ emptyErr := u.b.waitRoleCgroupEmpty(wctx, u.rt, instancectl.RoleCmd)
+ wcancel()
+ u.err = errors.Join(stopErr, emptyErr, u.out.Drain(cmdDrainBound), u.errs.Drain(cmdDrainBound))
+ })
+ return u.err
+}
+
+// cmdStream is the agent's end of one cmd stream (run spec §5.5): the
+// unit's StandardOutput=file: or StandardError=file: connects once, at
+// exec, and every byte is copied to w unchanged until EOF. Nothing goes to
+// the journal or the app log. One connection is taken; a second is closed
+// unread.
+type cmdStream struct {
+ ln net.Listener
+ w io.Writer
+ accepted atomic.Bool
+ closed atomic.Bool
+ mu sync.Mutex
+ conn net.Conn
+ copied chan struct{}
+ looped chan struct{} // closed when the accept loop has returned
+ wg sync.WaitGroup
+}
+
+func listenCmdStream(path string, w io.Writer) (*cmdStream, error) {
+ if w == nil {
+ w = io.Discard
+ }
+ ln, err := edge.SecureUnixListen(path)
+ if err != nil {
+ return nil, err
+ }
+ s := &cmdStream{ln: ln, w: w, copied: make(chan struct{}), looped: make(chan struct{})}
+ go s.accept()
+ return s, nil
+}
+
+func (s *cmdStream) accept() {
+ defer close(s.looped)
+ for {
+ c, err := s.ln.Accept()
+ if err != nil {
+ return
+ }
+ s.take(c)
+ }
+}
+
+// take starts copying c if it is the stream's first connection, and closes
+// it unread otherwise.
+func (s *cmdStream) take(c net.Conn) {
+ if s.accepted.Swap(true) {
+ c.Close()
+ return
+ }
+ s.mu.Lock()
+ if s.closed.Load() {
+ s.mu.Unlock()
+ c.Close()
+ return
+ }
+ s.conn = c
+ s.wg.Add(1)
+ s.mu.Unlock()
+ go s.copy(c)
+}
+
+func (s *cmdStream) copy(c net.Conn) {
+ defer s.wg.Done()
+ defer close(s.copied)
+ defer c.Close()
+ _, _ = io.Copy(s.w, c)
+}
+
+// cmdAcceptGrace is how long Drain waits for a connection still queued in
+// the listener's backlog.
+const cmdAcceptGrace = 100 * time.Millisecond
+
+// Drain waits up to bound for the stream to reach EOF (its unit has
+// stopped, so every writer has closed), then closes the listener. The unit
+// connected at exec, so the connection may still be queued in the backlog
+// when the accept loop has not yet run: Drain stops the loop and takes any
+// queued connection itself, so the command's last bytes are never dropped.
+// A stream that never connected has nothing to drain.
+func (s *cmdStream) Drain(bound time.Duration) error {
+ if dl, ok := s.ln.(interface{ SetDeadline(time.Time) error }); ok {
+ _ = dl.SetDeadline(time.Now())
+ <-s.looped
+ if !s.accepted.Load() {
+ _ = dl.SetDeadline(time.Now().Add(cmdAcceptGrace))
+ if c, err := s.ln.Accept(); err == nil {
+ s.take(c)
+ }
+ }
+ }
+ var err error
+ if s.accepted.Load() {
+ select {
+ case <-s.copied:
+ case <-time.After(bound):
+ err = fmt.Errorf("a cmd stream did not reach its end within %s", bound)
+ }
+ }
+ s.Close()
+ return err
+}
+
+func (s *cmdStream) Close() error {
+ if s.closed.Swap(true) {
+ return nil
+ }
+ err := s.ln.Close()
+ s.mu.Lock()
+ if s.conn != nil {
+ s.conn.Close()
+ }
+ s.mu.Unlock()
+ <-s.looped
+ s.wg.Wait()
+ return err
+}
diff --git a/internal/activator/cmdunit_test.go b/internal/activator/cmdunit_test.go
new file mode 100644
index 000000000..942c2b9fe
--- /dev/null
+++ b/internal/activator/cmdunit_test.go
@@ -0,0 +1,199 @@
+package activator
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "net"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// syncBuffer is a bytes.Buffer two goroutines may write.
+type syncBuffer struct {
+ mu sync.Mutex
+ b bytes.Buffer
+}
+
+func (s *syncBuffer) Write(p []byte) (int, error) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.b.Write(p)
+}
+func (s *syncBuffer) String() string { s.mu.Lock(); defer s.mu.Unlock(); return s.b.String() }
+
+func TestCmdVarsComposeTheRunEnvironment(t *testing.T) {
+ b := &ExecBackend{Region: "ie"}
+ rt := registry.Route{Host: "h.example.com"}
+ file := map[string]string{
+ "APP_KEY": "k", "CARLOS_ADMIN_TOKEN": "tok", "CARLOS_CONTROL_SOCKET": "/x", "CARLOS_PURGE": "header",
+ "CARLOS_REGION": "us", "CARLOS_HOST": "liar.example.com", "CARLOS_RUN_ID": "planted",
+ }
+ got := b.cmdVars(rt, file, "/data/instances/h.example.com/app.db", "run_a", "/data/instances/h.example.com/.carlos-run/run_a")
+ want := map[string]string{
+ "APP_KEY": "k", "CARLOS_ADMIN_TOKEN": "tok", "CARLOS_REGION": "ie", "CARLOS_HOST": "h.example.com",
+ "CARLOS_DB": "/data/instances/h.example.com/app.db", "CARLOS_RUN_ID": "run_a",
+ "CARLOS_RUN_DIR": "/data/instances/h.example.com/.carlos-run/run_a",
+ }
+ if len(got) != len(want) {
+ t.Fatalf("env = %v", got)
+ }
+ for k, v := range want {
+ if got[k] != v {
+ t.Errorf("%s = %q, want %q", k, got[k], v)
+ }
+ }
+ b.Region = ""
+ if _, ok := b.cmdVars(rt, file, "/db", "run_a", "/d")["CARLOS_REGION"]; ok {
+ t.Error("an empty backend region must remove CARLOS_REGION (codex 7)")
+ }
+}
+
+func startTestCmd(t *testing.T) (*ExecBackend, registry.Route, *fakeControl, *fakeUnit, CmdUnit, *syncBuffer, *syncBuffer) {
+ t.Helper()
+ ctl := &fakeControl{cmdInv: "inv1"}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
+ out, errs := &syncBuffer{}, &syncBuffer{}
+ dir := filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), CmdRunDirName, "run_a")
+ u, err := b.StartCmd(context.Background(), rt, CmdSpec{RunID: "run_a", Args: []string{"import", "--dry-run"}, RunDir: dir, Stdout: out, Stderr: errs})
+ if err != nil {
+ t.Fatalf("StartCmd: %v", err)
+ }
+ return b, rt, ctl, unit, u, out, errs
+}
+
+func TestStartCmdWritesTheLaunchFileAndStartsTheUnit(t *testing.T) {
+ b, rt, ctl, _, u, _, _ := startTestCmd(t)
+ if u.Invocation() != "inv1" || !strings.Contains(ctl.joined(), "cmd "+rt.Host) {
+ t.Fatalf("invocation %q, calls\n%s", u.Invocation(), ctl.joined())
+ }
+ raw, err := os.ReadFile(filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), "launch.cmd.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) != 3 || l.Argv[0] != b.liveLink(rt.Host) || l.Argv[1] != "import" || l.Argv[2] != "--dry-run" {
+ t.Errorf("argv = %q: exactly the binary and what followed --", l.Argv)
+ }
+ env := strings.Join(l.Env, "\n")
+ for _, want := range []string{"CARLOS_RUN_ID=run_a", "CARLOS_HOST=" + rt.Host, "CARLOS_DB=" + rt.DBPath} {
+ if !strings.Contains(env, want) {
+ t.Errorf("env lacks %q", want)
+ }
+ }
+ if strings.Contains(env, "CARLOS_CONTROL_SOCKET=") {
+ t.Error("a command has no control socket")
+ }
+}
+
+func TestStartCmdRefusesAFlatInstanceAndAForeignRunDir(t *testing.T) {
+ ctl := &fakeControl{cmdInv: "inv1"}
+ b, rt := nestedBackend(t, ctl, &fakeUnit{})
+ b.layoutFn = func(registry.Route) (Layout, string) { return LayoutFlat, "test" }
+ if _, err := b.StartCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); !errors.Is(err, ErrCmdNotNested) {
+ t.Errorf("flat: %v, want ErrCmdNotNested", err)
+ }
+ b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
+ if _, err := b.StartCmd(context.Background(), rt, CmdSpec{RunID: "run_a", RunDir: "/elsewhere"}); err == nil {
+ t.Error("a run directory outside the instance's .carlos-run was accepted")
+ }
+ if strings.Contains(ctl.joined(), "cmd ") {
+ t.Errorf("a refused start reached the control service:\n%s", ctl.joined())
+ }
+}
+
+func TestStopAndSettleDrainsBothStreams(t *testing.T) {
+ b, rt, ctl, unit, u, out, errs := startTestCmd(t)
+ unit.set("active", "inv1", true, true)
+ write := func(path, s string) {
+ c, err := net.Dial("unix", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := c.Write([]byte(s)); err != nil {
+ t.Fatal(err)
+ }
+ c.Close()
+ }
+ write(b.logSocketNamed(rt.Host, string(instancectl.RoleCmd)), "out-1\nout-2")
+ write(b.logSocketNamed(rt.Host, cmdErrSocket), "err-1")
+ go func() {
+ time.Sleep(30 * time.Millisecond)
+ unit.set("inactive", "inv1", false, false)
+ }()
+ if err := u.StopAndSettle(context.Background()); err != nil {
+ t.Fatalf("StopAndSettle: %v", err)
+ }
+ if out.String() != "out-1\nout-2" || errs.String() != "err-1" {
+ t.Errorf("stdout %q stderr %q", out.String(), errs.String())
+ }
+ if !strings.Contains(ctl.joined(), "stop "+rt.Host+" cmd") {
+ t.Errorf("the cmd unit was not stopped:\n%s", ctl.joined())
+ }
+ if err := u.StopAndSettle(context.Background()); err != nil {
+ t.Errorf("a second StopAndSettle: %v", err)
+ }
+}
+
+// A connection the unit made at exec may still be queued in the listener's
+// backlog when the unit has exited: Drain must take it and deliver its
+// bytes (run spec §5.5; plan Review Focus 5).
+func TestCmdStreamDrainTakesAConnectionStillInTheBacklog(t *testing.T) {
+ dir, err := os.MkdirTemp("", "cs")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(dir)
+ for i := 0; i < 200; i++ {
+ var w syncBuffer
+ s, err := listenCmdStream(filepath.Join(dir, "s.sock"), &w)
+ if err != nil {
+ t.Fatal(err)
+ }
+ c, err := net.Dial("unix", filepath.Join(dir, "s.sock"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ c.Write([]byte("last words"))
+ c.Close()
+ if err := s.Drain(time.Second); err != nil {
+ t.Fatalf("round %d: Drain: %v", i, err)
+ }
+ if w.String() != "last words" {
+ t.Fatalf("round %d: got %q", i, w.String())
+ }
+ }
+}
+
+func TestCmdStreamDrainOfAnUnconnectedStreamIsQuick(t *testing.T) {
+ dir, err := os.MkdirTemp("", "cs")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(dir)
+ s, err := listenCmdStream(filepath.Join(dir, "s.sock"), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ start := time.Now()
+ if err := s.Drain(time.Second); err != nil {
+ t.Fatal(err)
+ }
+ if d := time.Since(start); d > time.Second {
+ t.Errorf("took %s", d)
+ }
+}
diff --git a/internal/activator/control.go b/internal/activator/control.go
index 3fe7bf79c..9eb73e035 100644
--- a/internal/activator/control.go
+++ b/internal/activator/control.go
@@ -17,6 +17,10 @@ import (
type Control interface {
Start(ctx context.Context, host string, role instancectl.Role, memoryMax int64) (string, error)
Seed(ctx context.Context, host string, memoryMax int64) (instancectl.SeedResponse, error)
+ // Cmd starts host's cmd unit and returns its invocation; CmdStatus
+ // long-polls it (run spec §5.4).
+ Cmd(ctx context.Context, host string, memoryMax int64) (string, error)
+ CmdStatus(ctx context.Context, host, invocation string) (instancectl.CmdStatus, error)
Stop(ctx context.Context, host string, role instancectl.Role) error
Reset(ctx context.Context, host string, role instancectl.Role) error
Sweep(ctx context.Context) (instancectl.SweepResponse, error)
diff --git a/internal/activator/prepare.go b/internal/activator/prepare.go
index e3a9f4205..078fd7558 100644
--- a/internal/activator/prepare.go
+++ b/internal/activator/prepare.go
@@ -42,7 +42,7 @@ func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instanc
return p, nil, nil, nil, err
}
var env map[string]string
- if role == instancectl.RoleRun || role == instancectl.RoleSeed {
+ if role == instancectl.RoleRun || role == instancectl.RoleSeed || role == instancectl.RoleCmd {
// p.dataDir is the instance directory here, tenant-writable ground
// (design §3.2), so the token file is minted and read through
// instancefs rather than a plain path-based open.
@@ -91,12 +91,18 @@ func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instanc
_ = os.Remove(rt.Socket)
ctl = b.listenControl(rt, p.envFile)
}
- logs, err := listenLogs(b.logSocket(rt.Host, role), b.logEmitter(rt))
- if err != nil {
- if ctl != nil {
- ctl.Close()
+ // The cmd role's output goes only to its own chunks, never to the
+ // journal or the app log, so it has no app-log listener.
+ var logs *logListener
+ if role != instancectl.RoleCmd {
+ var err error
+ logs, err = listenLogs(b.logSocket(rt.Host, role), b.logEmitter(rt))
+ if err != nil {
+ if ctl != nil {
+ ctl.Close()
+ }
+ return p, nil, nil, nil, fmt.Errorf("log listener for %s %s: %w", rt.Host, role, err)
}
- return p, nil, nil, nil, fmt.Errorf("log listener for %s %s: %w", rt.Host, role, err)
}
return p, ctl, logs, env, nil
}
@@ -129,10 +135,16 @@ func (b *ExecBackend) listenControl(rt registry.Route, envPath string) *edge.Con
// logSocket is the role's log socket path, honouring the test override of
// the run directory (RunDirOverride).
func (b *ExecBackend) logSocket(host string, role instancectl.Role) string {
+ return b.logSocketNamed(host, string(role))
+}
+
+// logSocketNamed is logSocket for any listener name: the cmd role has two,
+// "cmd" and "cmd-err" (run spec §5.4).
+func (b *ExecBackend) logSocketNamed(host, name string) string {
if b.RunDirOverride != "" {
- return filepath.Join(b.RunDirOverride, "logs", provision.InstanceID(host)+"."+string(role)+".sock")
+ return filepath.Join(b.RunDirOverride, "logs", provision.InstanceID(host)+"."+name+".sock")
}
- return provision.LogSocket(host, string(role))
+ return provision.LogSocket(host, name)
}
// logEmitter is the same sink teeStd feeds, or nil when Logs is unset.
diff --git a/internal/activator/unitproc_test.go b/internal/activator/unitproc_test.go
index 7b867e7c5..84b8eace0 100644
--- a/internal/activator/unitproc_test.go
+++ b/internal/activator/unitproc_test.go
@@ -39,6 +39,10 @@ type fakeControl struct {
// might have written to disk before it failed. A test uses it to give
// seedOwnsDB cleanup something real to (not) remove.
onSeed func()
+
+ cmdInv string
+ cmdErr error
+ cmdStatus instancectl.CmdStatus
}
func (f *fakeControl) rec(s string) { f.mu.Lock(); f.calls = append(f.calls, s); f.mu.Unlock() }
@@ -56,6 +60,14 @@ func (f *fakeControl) Seed(_ context.Context, host string, max int64) (instancec
}
return f.seed, nil
}
+func (f *fakeControl) Cmd(_ context.Context, host string, max int64) (string, error) {
+ f.rec("cmd " + host)
+ return f.cmdInv, f.cmdErr
+}
+func (f *fakeControl) CmdStatus(_ context.Context, host, inv string) (instancectl.CmdStatus, error) {
+ f.rec("cmdstatus " + host + " " + inv)
+ return f.cmdStatus, nil
+}
func (f *fakeControl) Stop(_ context.Context, host string, role instancectl.Role) error {
f.rec("stop " + host + " " + string(role))
if f.stopHook != nil {