diff --git a/go/internal/runner/agent_exec.go b/go/internal/runner/agent_exec.go index e0e5bdfa4..16d842126 100644 --- a/go/internal/runner/agent_exec.go +++ b/go/internal/runner/agent_exec.go @@ -226,10 +226,28 @@ func (s *AgentStream) endDrains() { } // isDeliberateKill reports whether err is the exit of a process we SIGKILLed on -// purpose — an *exec.ExitError whose wait status is "terminated by SIGKILL". -// That is exactly the outcome Terminate produces on the deliberate-teardown -// path, so Stop treats it as success while still surfacing any other failure. +// purpose, so Stop treats it as success while still surfacing any other +// failure. Two backends produce that outcome in two shapes: +// +// - The microVM GuestExec ChildHandle waitFunc returns a portable +// *runtime.ExitStatusError; a remote guest child's exit cannot be reported +// as an *exec.ExitError (it embeds an unforgeable *os.ProcessState), so the +// portable type is checked FIRST — a deliberate signal counts as a kill. +// - The podman backend's Wait returns an *exec.ExitError whose wait status is +// "terminated by SIGKILL"; that branch is unchanged, so the podman +// byte-path is byte-identical (OQ-G/U3b). func isDeliberateKill(err error) bool { + var exitStatus *runtime.ExitStatusError + if errors.As(err, &exitStatus) { + // The two branches are deliberately asymmetric (OQ-G): the portable + // branch counts ANY signalled exit as a kill, while the podman branch + // below pins SIGKILL. That is intentional — the guest reports a + // deliberate teardown as SIGKILL (Kill) or SIGTERM (Stop), and OQ-G + // blessed Signal!=0 rather than enumerating signals. Do NOT "align" the + // two by narrowing this to SIGKILL: the microVM path has no os.ProcessState + // to inspect, and Stop's SIGTERM teardown must still classify as a kill. + return exitStatus.Signal != 0 + } var exitErr *exec.ExitError if !errors.As(err, &exitErr) { return false diff --git a/go/internal/runner/deliberate_kill_test.go b/go/internal/runner/deliberate_kill_test.go new file mode 100644 index 000000000..ebccb31ed --- /dev/null +++ b/go/internal/runner/deliberate_kill_test.go @@ -0,0 +1,89 @@ +//go:build unix + +package runner + +// isDeliberateKill's widened taxonomy (U3b/OQ-G): it accepts both a real +// *exec.ExitError from a SIGKILLed local child (the podman byte-path, unchanged) +// and the portable *runtime.ExitStatusError a remote (microVM) waitFunc +// constructs, and rejects a non-signal exit and an unrelated error. Hermetic: +// the podman-path case SIGKILLs a real short-lived child, the rest are +// constructed errors; no KVM, no backend. + +import ( + "errors" + "os/exec" + "syscall" + "testing" + + "github.com/RigelBuild/compass/go/internal/runtime" +) + +// sigkilledExitError runs a trivial child and SIGKILLs it, returning the +// *exec.ExitError its Wait yields — the exact shape the podman ChildHandle.Wait +// produces on a deliberate teardown, so the regression guard exercises a real +// wait status rather than a hand-built one. +func sigkilledExitError(t *testing.T) error { + t.Helper() + cmd := exec.Command("sleep", "60") + if err := cmd.Start(); err != nil { + t.Fatalf("starting child: %v", err) + } + if err := cmd.Process.Signal(syscall.SIGKILL); err != nil { + t.Fatalf("sigkilling child: %v", err) + } + err := cmd.Wait() + if err == nil { + t.Fatal("expected a non-nil wait error for a SIGKILLed child") + } + var exitErr *exec.ExitError + if !errors.As(err, &exitErr) { + t.Fatalf("wait error is %T, want *exec.ExitError", err) + } + return err +} + +func TestIsDeliberateKill(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + { + name: "podman-path *exec.ExitError SIGKILL", + err: sigkilledExitError(t), + want: true, + }, + { + name: "portable ExitStatusError SIGKILL", + err: &runtime.ExitStatusError{Signal: syscall.SIGKILL}, + want: true, + }, + { + name: "portable ExitStatusError SIGTERM is still a deliberate signal", + err: &runtime.ExitStatusError{Signal: syscall.SIGTERM}, + want: true, + }, + { + name: "portable ExitStatusError non-signal exit is not a kill", + err: &runtime.ExitStatusError{Code: 1}, + want: false, + }, + { + name: "unrelated error", + err: errors.New("connection reset"), + want: false, + }, + { + name: "nil error", + err: nil, + want: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := isDeliberateKill(tt.err); got != tt.want { + t.Fatalf("isDeliberateKill(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} diff --git a/go/internal/runtime/child_handle_test.go b/go/internal/runtime/child_handle_test.go new file mode 100644 index 000000000..9a375b041 --- /dev/null +++ b/go/internal/runtime/child_handle_test.go @@ -0,0 +1,95 @@ +package runtime + +// newChildHandleFuncs (U3): the funcs-backed ChildHandle the microVM GuestExec +// adaptation consumes (U4 wires the real kill/wait pair onto a GuestStream). +// These hermetic unit tests pin the exported Kill/Wait/Terminate surface over a +// kill/wait function pair — no *exec.Cmd, no backend — including that a +// signalled-exit waitFunc returning *ExitStatusError flows through Wait +// unchanged, so the runner's isDeliberateKill can recognize it. + +import ( + "errors" + "syscall" + "testing" +) + +func TestNewChildHandleFuncs_Kill(t *testing.T) { + killed := false + h := newChildHandleFuncs( + func() error { killed = true; return nil }, + func() error { return nil }, + ) + if err := h.Kill(); err != nil { + t.Fatalf("Kill: %v", err) + } + if !killed { + t.Fatal("Kill did not invoke killFunc") + } +} + +func TestNewChildHandleFuncs_KillError(t *testing.T) { + want := errors.New("transport wedged") + h := newChildHandleFuncs( + func() error { return want }, + func() error { return nil }, + ) + if err := h.Kill(); !errors.Is(err, want) { + t.Fatalf("Kill error = %v, want %v", err, want) + } +} + +func TestNewChildHandleFuncs_WaitExitZero(t *testing.T) { + h := newChildHandleFuncs( + func() error { return nil }, + func() error { return nil }, + ) + if err := h.Wait(); err != nil { + t.Fatalf("Wait on exit 0 = %v, want nil", err) + } +} + +func TestNewChildHandleFuncs_WaitSignalledExit(t *testing.T) { + // A signalled guest exit: waitFunc returns the portable *ExitStatusError, + // which Wait must surface unchanged so isDeliberateKill recognizes it. + h := newChildHandleFuncs( + func() error { return nil }, + func() error { return &ExitStatusError{Signal: syscall.SIGKILL} }, + ) + err := h.Wait() + var exitStatus *ExitStatusError + if !errors.As(err, &exitStatus) { + t.Fatalf("Wait error = %T, want *ExitStatusError", err) + } + if exitStatus.Signal != syscall.SIGKILL { + t.Fatalf("signal = %v, want SIGKILL", exitStatus.Signal) + } +} + +func TestNewChildHandleFuncs_TerminateReturnsWaitError(t *testing.T) { + // Terminate is Kill then Wait; the wait error (the exit status) is what a + // caller distinguishing crash-from-teardown needs, so it wins over the kill + // error. + killErr := errors.New("signal RPC timed out") + h := newChildHandleFuncs( + func() error { return killErr }, + func() error { return &ExitStatusError{Signal: syscall.SIGKILL} }, + ) + err := h.Terminate() + var exitStatus *ExitStatusError + if !errors.As(err, &exitStatus) { + t.Fatalf("Terminate error = %T, want *ExitStatusError (the wait error)", err) + } +} + +func TestNewChildHandleFuncs_TerminateExitZero(t *testing.T) { + // Clean exit with a kill error: Terminate returns the kill error, since the + // wait error is nil. + killErr := errors.New("signal RPC timed out") + h := newChildHandleFuncs( + func() error { return killErr }, + func() error { return nil }, + ) + if err := h.Terminate(); !errors.Is(err, killErr) { + t.Fatalf("Terminate error = %v, want the kill error %v", err, killErr) + } +} diff --git a/go/internal/runtime/exit_status_error.go b/go/internal/runtime/exit_status_error.go new file mode 100644 index 000000000..180d09046 --- /dev/null +++ b/go/internal/runtime/exit_status_error.go @@ -0,0 +1,38 @@ +package runtime + +import ( + "fmt" + "syscall" +) + +// ExitStatusError is a backend-portable process-exit error the runner's +// isDeliberateKill recognizes, so a remote (microVM) guest exit can be told +// from a crash without fabricating an *exec.ExitError. +// +// The podman backend reports a deliberate SIGKILL teardown as an +// *exec.ExitError whose syscall.WaitStatus is Signaled()+SIGKILL +// (agent_exec.go isDeliberateKill). A remote exec (the microVM GuestExec +// ChildHandle waitFunc) cannot construct an *exec.ExitError: it embeds +// *os.ProcessState, which has unexported fields and no public constructor, so +// a waitFunc reporting a guest child's exit has no way to forge one. This +// exported concrete type is the portable stand-in — a plain (code, signal) +// pair the microVM waitFunc returns and isDeliberateKill matches with +// errors.As, alongside the existing *exec.ExitError branch so the podman +// byte-path is unchanged (OQ-G, design §(e)). It is a concrete struct rather +// than an interface: it is the simplest errors.As target and no caller needs +// the abstraction today. +type ExitStatusError struct { + // Code is the child's exit code, meaningful when Signal == 0. + Code int + // Signal is the terminating signal, non-zero when the child died by signal + // (e.g. syscall.SIGKILL on a deliberate Kill). + Signal syscall.Signal +} + +// Error describes the exit as either a signal death or a non-zero exit code. +func (e *ExitStatusError) Error() string { + if e.Signal != 0 { + return fmt.Sprintf("process terminated by signal %d (%s)", int(e.Signal), e.Signal) + } + return fmt.Sprintf("process exited with code %d", e.Code) +} diff --git a/go/internal/runtime/microvm/exec.go b/go/internal/runtime/microvm/exec.go new file mode 100644 index 000000000..d820b18c1 --- /dev/null +++ b/go/internal/runtime/microvm/exec.go @@ -0,0 +1,405 @@ +//go:build unix + +package microvm + +// exec.go is the host-side exec layer over the U1 GuestControl client: a +// GuestExec wrapping the GuestControl Connect client (dial.go GuestClient) with +// a one-shot Exec and a streaming ExecStream that turns the bidi frame protocol +// into live io.Pipe stdio plus a kill/wait handle (design §(c), record §Plan +// U3). +// +// Package boundary: this layer produces plain structs mirroring the proto +// (ExecCall/ExecResult/StreamCall/ExitStatus) rather than go/internal/runtime +// types. runtime's MicroVMRuntime (U4) consumes GuestExec, so runtime imports +// microvm; microvm importing runtime would cycle. The runtime-side adaptation +// (spec -> ExecCall, ExitStatus -> ExecOutput/*runtime.ExitStatusError, the +// newChildHandleFuncs kill/wait pair) is U4's, kept out of this package. + +import ( + "context" + "errors" + "fmt" + "io" + "syscall" + "time" + + "connectrpc.com/connect" + + compassv1 "github.com/RigelBuild/compass/go/internal/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/gen/compass/v1/compassv1internalconnect" +) + +// sigKill is the signal reported for a deliberate teardown (a ctx-cancelled +// stream with no exit frame): guestd SIGKILLs the child bound to the broken +// stream, so the host reports SIGKILL, which the runtime waitFunc maps to the +// portable deliberate-kill error. +const sigKill = syscall.SIGKILL + +// killSignalTimeout bounds the Signal RPC a Kill issues so a wedged transport +// cannot block the teardown path: podman's Kill is an instantaneous local +// cancel, so the microVM Kill must not stall on the wire. On timeout the error +// is returned to the caller, which ignores it (the VMM-kill escalation in Stop +// is the backstop) — Wait still returns via the demux goroutine, so a Kill RPC +// that never lands does not wedge teardown (design §(c)). +const killSignalTimeout = 5 * time.Second + +// stdinChunk is the stdin pump's read buffer size; a larger chunk is more +// frames of the same bytes, so it affects only framing overhead, not +// correctness. +const stdinChunk = 32 * 1024 + +// TimeoutError is a one-shot exec that overran its per-command wall-clock cap +// and was aborted rather than left to block the caller. It mirrors the +// discipline of runtime.TimeoutError without leaking that type across the +// package boundary; MicroVMRuntime.Exec (U4) translates it to the runtime type. +type TimeoutError struct { + Timeout time.Duration +} + +func (e *TimeoutError) Error() string { + return fmt.Sprintf("guest exec timed out after %ds", int(e.Timeout.Seconds())) +} + +// ExecCall is a one-shot exec request, mirroring ExecRequest field-for-field so +// runtime types don't leak into this package. +type ExecCall struct { + Command []string + // UID is the exec user; nil uses the session default set by Provision. UID 0 + // is refused guest-side. + UID *uint32 + // Workdir is the working directory; nil uses the child's default. + Workdir *string + // Env is merged over the session base env. + Env map[string]string + // Stdin is fed to the child's stdin over the wire — never the argv — so a + // script body never appears in the guest process list. + Stdin []byte + // TimeoutSeconds bounds the command; 0 leaves it to the caller's ctx. It is + // enforced host-side (a ctx deadline) and mirrored guest-side. + TimeoutSeconds uint32 +} + +// ExecResult is a completed one-shot exec, mirroring ExecResponse. A non-zero +// ExitCode is a successful result, not an error. +type ExecResult struct { + Stdout []byte + Stderr []byte + ExitCode int +} + +// StreamCall is a streaming exec request, mirroring StartExec. Unlike ExecCall +// it carries no stdin body or timeout: a streaming exec keeps a live stdin pipe +// for its whole life and is long-lived by design. +type StreamCall struct { + Command []string + UID *uint32 + Workdir *string + Env map[string]string +} + +// ExitStatus is how a streaming exec ended: a non-zero Signal means the child +// died by signal (e.g. SIGKILL on a Kill), otherwise Code is the exit code. +type ExitStatus struct { + Code int + Signal int +} + +// GuestExec is the host-side exec layer over a GuestControl client. +type GuestExec struct { + client compassv1internalconnect.GuestControlClient +} + +// NewGuestExec wraps a GuestControl client (dial.go GuestClient) in the exec +// layer. +func NewGuestExec(client compassv1internalconnect.GuestControlClient) *GuestExec { + return &GuestExec{client: client} +} + +// Exec runs one command to completion over the Exec RPC. A non-zero exit is a +// successful ExecResult with a non-zero ExitCode, NEVER an error; a gate-closed +// or uid-0 refusal or a transport failure is an error. A per-command timeout +// (ExecCall.TimeoutSeconds) is enforced host-side as a ctx deadline and mapped +// to a *TimeoutError, mirroring runtime.PodmanCLI's TimeoutError discipline. +func (g *GuestExec) Exec(ctx context.Context, call ExecCall) (ExecResult, error) { + var timeout time.Duration + if call.TimeoutSeconds > 0 { + timeout = time.Duration(call.TimeoutSeconds) * time.Second + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, timeout) + defer cancel() + } + + req := &compassv1.ExecRequest{ + Command: call.Command, + Uid: call.UID, + Workdir: call.Workdir, + Env: call.Env, + Stdin: call.Stdin, + TimeoutSeconds: call.TimeoutSeconds, + } + resp, err := g.client.Exec(ctx, connect.NewRequest(req)) + if err != nil { + if timeout > 0 && isDeadline(err) { + return ExecResult{}, &TimeoutError{Timeout: timeout} + } + return ExecResult{}, fmt.Errorf("guest exec: %w", err) + } + return ExecResult{ + Stdout: resp.Msg.GetStdout(), + Stderr: resp.Msg.GetStderr(), + ExitCode: int(resp.Msg.GetExitCode()), + }, nil +} + +// isDeadline reports whether err is a host-side deadline: a wrapped +// context.DeadlineExceeded or a Connect deadline-exceeded code. +func isDeadline(err error) bool { + return errors.Is(err, context.DeadlineExceeded) || + connect.CodeOf(err) == connect.CodeDeadlineExceeded +} + +// ExecStream opens the bidi stream, sends the StartExec frame, and AWAITS the +// ExecStarted frame before returning — so a spawn failure surfaces as the +// returned error rather than a broken pipe on first read. It returns a +// GuestStream whose io.Pipe stdio is pumped by a per-exec goroutine: the Stdin +// pipe is framed as stdin frames (Close -> stdin_close), and stdout/stderr +// response frames are demuxed onto the read pipes, which close when the +// ExecExit frame arrives. +func (g *GuestExec) ExecStream(ctx context.Context, call StreamCall) (_ *GuestStream, retErr error) { + stream := g.client.ExecStream(ctx) + // Until the pumps below take ownership of the stream, any early-error return + // must reap both halves itself — the bidi stream is already open, so a bare + // `return nil, err` leaks its response-body reader / makeRequest goroutine + // for a failed spawn. The pumps own the reap once they start (return gs, nil + // leaves retErr nil, so this is a no-op on the success path). + defer func() { + if retErr != nil { + _ = stream.CloseRequest() + _ = stream.CloseResponse() + } + }() + start := &compassv1.ExecStreamRequest{ + Frame: &compassv1.ExecStreamRequest_Start{ + Start: &compassv1.StartExec{ + Command: call.Command, + Uid: call.UID, + Workdir: call.Workdir, + Env: call.Env, + }, + }, + } + if err := stream.Send(start); err != nil { + return nil, fmt.Errorf("guest exec stream: sending start: %w", err) + } + + // Await the ExecStarted frame: only then is the spawn known to have + // succeeded, so a spawn failure is the returned error, not a first-read + // broken pipe. + first, err := stream.Receive() + if err != nil { + return nil, fmt.Errorf("guest exec stream: awaiting started: %w", err) + } + started := first.GetStarted() + if started == nil { + return nil, fmt.Errorf("guest exec stream: first frame was not started: %T", first.GetFrame()) + } + + stdinR, stdinW := io.Pipe() + stdoutR, stdoutW := io.Pipe() + stderrR, stderrW := io.Pipe() + + gs := &GuestStream{ + ctx: ctx, + client: g.client, + stream: stream, + execID: started.GetExecId(), + Stdin: stdinW, + Stdout: stdoutR, + Stderr: stderrR, + stdinR: stdinR, + stdoutW: stdoutW, + stderrW: stderrW, + done: make(chan struct{}), + } + go gs.pumpStdin() + go gs.pumpResponses() + return gs, nil +} + +// GuestStream is a live streaming exec: its stdio pipes plus a kill/wait +// surface over the guest child. Stdin/Stdout/Stderr are the caller's ends of +// io.Pipes pumped by pumpStdin/pumpResponses. +type GuestStream struct { + // ctx is the stream-lifetime context (not a per-request ctx): Kill's Signal + // RPC deadline derives from it, and its cancellation is the deliberate + // teardown pumpResponses maps to SIGKILL. The per-call ctx is the wrong + // lifetime here. + //nolint:containedctx // stream-lifetime scope for Kill's Signal RPC + teardown detection; a per-request ctx cannot carry it (see field doc) + ctx context.Context + client compassv1internalconnect.GuestControlClient + stream *connect.BidiStreamForClient[compassv1.ExecStreamRequest, compassv1.ExecStreamResponse] + execID string + + // Stdin/Stdout/Stderr are the caller's pipe ends. + Stdin io.WriteCloser + Stdout io.ReadCloser + Stderr io.ReadCloser + + // stdinR/stdoutW/stderrW are the pump's ends of the same pipes. + stdinR *io.PipeReader + stdoutW *io.PipeWriter + stderrW *io.PipeWriter + + // done is closed once pumpResponses observes the exit frame or a stream + // break; status is valid to read only after done is closed. + done chan struct{} + status ExitStatus +} + +// Kill delivers sig to the guest child over the Signal RPC, bounded by +// killSignalTimeout so a wedged transport never blocks the caller past it. The +// exit frame that follows a SIGKILL arrives on the response stream and unblocks +// Wait; a Signal RPC that never lands is not fatal — the response pump still +// unblocks Wait on the eventual stream break, and the VMM-kill escalation is +// the backstop. The returned error is informational (the caller's teardown path +// ignores it). +func (s *GuestStream) Kill(sig int) error { + ctx, cancel := context.WithTimeout(s.ctx, killSignalTimeout) + defer cancel() + _, err := s.client.Signal(ctx, connect.NewRequest(&compassv1.SignalRequest{ + ExecId: s.execID, + Signal: int32(sig), //nolint:gosec // G115: sig is a small signal number (syscall.Signal), never overflows int32 + })) + if err != nil { + return fmt.Errorf("guest exec kill: signal %d: %w", sig, err) + } + return nil +} + +// Wait blocks until the response pump observes the terminal exit frame (or a +// stream break) and returns how the exec ended. It never blocks past the +// child's exit: a Kill's exit frame, a clean EOF, or a broken stream all close +// done. +func (s *GuestStream) Wait() ExitStatus { + <-s.done + return s.status +} + +// pumpStdin reads the caller's Stdin pipe and frames it as stdin frames; when +// the caller closes Stdin (pipe EOF) it sends a stdin_close half-close and +// closes the request direction of the stream. Kill rides the separate Signal +// RPC, not this stream, so closing the request side here never races a +// teardown. +func (s *GuestStream) pumpStdin() { + buf := make([]byte, stdinChunk) + for { + n, readErr := s.stdinR.Read(buf) + if n > 0 { + frame := make([]byte, n) + copy(frame, buf[:n]) + if sendErr := s.stream.Send(&compassv1.ExecStreamRequest{ + Frame: &compassv1.ExecStreamRequest_Stdin{Stdin: frame}, + }); sendErr != nil { + // The stream is gone; the response pump observes the same break + // and unblocks Wait. Nothing further to do on the stdin side. + return + } + } + if readErr != nil { + // EOF is the caller closing Stdin (the common path); any other read + // error means the pipe was closed with an error. Either way, signal + // the guest to half-close the child's stdin, then close the request + // direction. + if sendErr := s.stream.Send(&compassv1.ExecStreamRequest{ + Frame: &compassv1.ExecStreamRequest_StdinClose{StdinClose: &compassv1.StdinClose{}}, + }); sendErr != nil { + return + } + // Half-close the request side; the response side stays open to carry + // stdout/stderr and the terminal exit frame. A CloseRequest error is + // not actionable here — the response pump surfaces any real break. + _ = s.stream.CloseRequest() + return + } + } +} + +// pumpResponses demuxes stdout/stderr frames onto the read pipes and, on the +// terminal exit frame (or a stream break), records the exit status, closes both +// read pipes, and closes done to unblock Wait. Writing to a pipe blocks until +// the caller reads, so a full pipe applies backpressure rather than dropping +// bytes — matching the runner's continuous-drain model. +func (s *GuestStream) pumpResponses() { + defer close(s.done) + // On any terminal exit, also reap the stdin pump and free the response + // half: pumpStdin may be parked in stdinR.Read (a caller that never wrote + // or closed Stdin), and the muxed response half must not leak per exec. + // This defer runs before close(s.done), so by the time Wait returns the + // pump's reader end is closed. + defer s.reapStdinPump() + for { + resp, err := s.stream.Receive() + if err != nil { + // Stream ended without a terminal exit frame: EOF is a clean close, + // anything else (ctx cancel, transport break) is a broken stream. A + // ctx cancel is a deliberate teardown, which SIGKILLs the guest child + // (guestd binds the child to the stream ctx), so report SIGKILL so + // the runtime waitFunc recognizes a deliberate kill; a non-cancel + // break with no exit frame is reported as a non-zero code. + if s.ctx.Err() != nil { + s.status = ExitStatus{Signal: int(sigKill)} + } else if !errors.Is(err, io.EOF) { + s.status = ExitStatus{Code: -1} + } + s.closePipes(err) + return + } + switch frame := resp.GetFrame().(type) { + case *compassv1.ExecStreamResponse_Stdout: + // A write error means the caller's read end is gone; the guest child + // is still reaped on stream teardown, so stop pumping this pipe. + if _, werr := s.stdoutW.Write(frame.Stdout); werr != nil { + s.closePipes(werr) + return + } + case *compassv1.ExecStreamResponse_Stderr: + if _, werr := s.stderrW.Write(frame.Stderr); werr != nil { + s.closePipes(werr) + return + } + case *compassv1.ExecStreamResponse_Exit: + s.status = ExitStatus{ + Code: int(frame.Exit.GetExitCode()), + Signal: int(frame.Exit.GetSignal()), + } + s.closePipes(io.EOF) + return + default: + // A duplicate started frame or an unknown frame: ignore and keep + // reading toward the terminal exit frame. + } + } +} + +// closePipes closes both read-pipe writer ends with cause, so the caller's +// Stdout/Stderr reads observe EOF (or the cause). CloseWithError(io.EOF) yields +// a plain EOF to the reader. +func (s *GuestStream) closePipes(cause error) { + _ = s.stdoutW.CloseWithError(cause) // signalled to the caller's Stdout reader + _ = s.stderrW.CloseWithError(cause) // signalled to the caller's Stderr reader +} + +// reapStdinPump unblocks and terminates pumpStdin and frees the response half +// on a stream's terminal exit. pumpStdin blocks in stdinR.Read until the caller +// writes or closes Stdin; on a clean child exit (or a break) where the caller +// did neither, nothing else wakes it, so closing the pump's reader end makes +// that Read return and the goroutine exit. It touches only the pipe, never the +// stream send half, so it does not race pumpStdin's own Send/CloseRequest (the +// send half stays single-owner). CloseResponse frees the receive half for +// connection reuse; it is safe here because pumpResponses is the sole Receiver +// and has already stopped. A caller that closed Stdin already drove pumpStdin's +// own CloseRequest, so the extra reader close is a harmless idempotent no-op. +func (s *GuestStream) reapStdinPump() { + _ = s.stdinR.CloseWithError(io.ErrClosedPipe) + _ = s.stream.CloseResponse() +} diff --git a/go/internal/runtime/microvm/exec_test.go b/go/internal/runtime/microvm/exec_test.go new file mode 100644 index 000000000..9251ae53c --- /dev/null +++ b/go/internal/runtime/microvm/exec_test.go @@ -0,0 +1,452 @@ +//go:build unix + +package microvm + +// Hermetic round-trip for the host exec layer (RIG-2588 U3): a fake +// GuestControl server on a unix listener speaking h2c Connect, dialed by a real +// GuestControlClient — no KVM, no vsock muxer. The fake's ExecStream handler is +// scripted per test so the pump's demux, stdin framing, kill/exit unblocking, +// ctx-cancel teardown, and one-shot timeout are each exercised against a real +// bidi stream. + +import ( + "context" + "errors" + "io" + "net" + "net/http" + "path/filepath" + "sync" + "testing" + "time" + + "connectrpc.com/connect" + + compassv1 "github.com/RigelBuild/compass/go/internal/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/gen/compass/v1/compassv1internalconnect" +) + +// fakeGuest is a scriptable GuestControl server. Each RPC delegates to a func +// field so a test supplies only the behavior it exercises; an unset field +// embeds UnimplementedGuestControlHandler's CodeUnimplemented. +type fakeGuest struct { + compassv1internalconnect.UnimplementedGuestControlHandler + execFn func(context.Context, *connect.Request[compassv1.ExecRequest]) (*connect.Response[compassv1.ExecResponse], error) + execStreamFn func(context.Context, *connect.BidiStream[compassv1.ExecStreamRequest, compassv1.ExecStreamResponse]) error + signalFn func(context.Context, *connect.Request[compassv1.SignalRequest]) (*connect.Response[compassv1.SignalResponse], error) +} + +func (f *fakeGuest) Exec(ctx context.Context, req *connect.Request[compassv1.ExecRequest]) (*connect.Response[compassv1.ExecResponse], error) { + return f.execFn(ctx, req) +} + +func (f *fakeGuest) ExecStream(ctx context.Context, stream *connect.BidiStream[compassv1.ExecStreamRequest, compassv1.ExecStreamResponse]) error { + return f.execStreamFn(ctx, stream) +} + +func (f *fakeGuest) Signal(ctx context.Context, req *connect.Request[compassv1.SignalRequest]) (*connect.Response[compassv1.SignalResponse], error) { + return f.signalFn(ctx, req) +} + +// serveFakeGuest binds a unix listener, serves fake over h2c Connect, and +// returns a GuestExec whose client dials that listener. Everything is torn down +// via t.Cleanup. +func serveFakeGuest(t *testing.T, fake *fakeGuest) *GuestExec { + t.Helper() + path := filepath.Join(t.TempDir(), "guest.sock") + + ln, err := net.Listen("unix", path) + if err != nil { + t.Fatalf("binding fake guest: %v", err) + } + t.Cleanup(func() { _ = ln.Close() }) // listener teardown + + mux := http.NewServeMux() + mux.Handle(compassv1internalconnect.NewGuestControlHandler(fake)) + protocols := new(http.Protocols) + protocols.SetUnencryptedHTTP2(true) + srv := &http.Server{Handler: mux, Protocols: protocols} + t.Cleanup(func() { _ = srv.Close() }) // server teardown + go func() { _ = srv.Serve(ln) }() // errors on Close + + clientProtocols := new(http.Protocols) + clientProtocols.SetUnencryptedHTTP2(true) + httpClient := &http.Client{ + Transport: &http.Transport{ + Protocols: clientProtocols, + DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + var d net.Dialer + return d.DialContext(ctx, "unix", path) + }, + }, + } + client := compassv1internalconnect.NewGuestControlClient(httpClient, "http://guest") + return NewGuestExec(client) +} + +func TestGuestExec_OneShot_NonZeroExitIsSuccess(t *testing.T) { + ge := serveFakeGuest(t, &fakeGuest{ + execFn: func(_ context.Context, req *connect.Request[compassv1.ExecRequest]) (*connect.Response[compassv1.ExecResponse], error) { + // Echo the command back so we also prove the spec mapped through. + if got := req.Msg.GetCommand(); len(got) != 1 || got[0] != "false" { + t.Errorf("command = %v, want [false]", got) + } + return connect.NewResponse(&compassv1.ExecResponse{ + Stdout: []byte("out"), + Stderr: []byte("err"), + ExitCode: 3, + }), nil + }, + }) + + res, err := ge.Exec(context.Background(), ExecCall{Command: []string{"false"}}) + if err != nil { + t.Fatalf("Exec: %v", err) + } + if res.ExitCode != 3 { + t.Fatalf("ExitCode = %d, want 3", res.ExitCode) + } + if string(res.Stdout) != "out" || string(res.Stderr) != "err" { + t.Fatalf("stdout/stderr = %q/%q, want out/err", res.Stdout, res.Stderr) + } +} + +func TestGuestExec_OneShot_RefusalIsError(t *testing.T) { + ge := serveFakeGuest(t, &fakeGuest{ + execFn: func(context.Context, *connect.Request[compassv1.ExecRequest]) (*connect.Response[compassv1.ExecResponse], error) { + return nil, connect.NewError(connect.CodeFailedPrecondition, errors.New("uid 0 refused")) + }, + }) + + _, err := ge.Exec(context.Background(), ExecCall{Command: []string{"whoami"}}) + if err == nil { + t.Fatal("expected an error for a guest-side refusal") + } + if connect.CodeOf(err) != connect.CodeFailedPrecondition { + t.Fatalf("code = %v, want FailedPrecondition", connect.CodeOf(err)) + } +} + +func TestGuestExec_OneShot_Timeout(t *testing.T) { + ge := serveFakeGuest(t, &fakeGuest{ + execFn: func(ctx context.Context, _ *connect.Request[compassv1.ExecRequest]) (*connect.Response[compassv1.ExecResponse], error) { + <-ctx.Done() // outlive the host deadline + return nil, connect.NewError(connect.CodeDeadlineExceeded, ctx.Err()) + }, + }) + + _, err := ge.Exec(context.Background(), ExecCall{Command: []string{"sleep"}, TimeoutSeconds: 1}) + var timeoutErr *TimeoutError + if !errors.As(err, &timeoutErr) { + t.Fatalf("error = %v (%T), want *TimeoutError", err, err) + } +} + +// scriptStream is a small helper wiring the guest-side of ExecStream: it sends +// the ExecStarted frame, then runs body with the stream. body returns the exit +// frame to emit (or nil to end without one). +func startStream(t *testing.T, stream *connect.BidiStream[compassv1.ExecStreamRequest, compassv1.ExecStreamResponse], execID string) { + t.Helper() + if err := stream.Send(&compassv1.ExecStreamResponse{ + Frame: &compassv1.ExecStreamResponse_Started{Started: &compassv1.ExecStarted{ExecId: execID}}, + }); err != nil { + t.Errorf("sending started: %v", err) + } +} + +func TestGuestExec_Stream_PumpDemuxAndExit(t *testing.T) { + ge := serveFakeGuest(t, &fakeGuest{ + execStreamFn: func(_ context.Context, stream *connect.BidiStream[compassv1.ExecStreamRequest, compassv1.ExecStreamResponse]) error { + // First frame must be the start. + first, err := stream.Receive() + if err != nil { + return err + } + if first.GetStart() == nil { + t.Errorf("first request frame was not start: %T", first.GetFrame()) + } + startStream(t, stream, "exec-1") + // Interleave stdout/stderr, then exit. + _ = stream.Send(&compassv1.ExecStreamResponse{Frame: &compassv1.ExecStreamResponse_Stdout{Stdout: []byte("o1")}}) + _ = stream.Send(&compassv1.ExecStreamResponse{Frame: &compassv1.ExecStreamResponse_Stderr{Stderr: []byte("e1")}}) + _ = stream.Send(&compassv1.ExecStreamResponse{Frame: &compassv1.ExecStreamResponse_Stdout{Stdout: []byte("o2")}}) + return stream.Send(&compassv1.ExecStreamResponse{ + Frame: &compassv1.ExecStreamResponse_Exit{Exit: &compassv1.ExecExit{ExitCode: 0}}, + }) + }, + }) + + gs, err := ge.ExecStream(context.Background(), StreamCall{Command: []string{"echo"}}) + if err != nil { + t.Fatalf("ExecStream: %v", err) + } + + // Close stdin immediately (no input); the pump half-closes. + if err := gs.Stdin.Close(); err != nil { + t.Fatalf("closing stdin: %v", err) + } + + // Both pipes must be drained concurrently: the pump blocks writing a stderr + // frame until the caller reads it, so a serial read (all stdout, then + // stderr) would deadlock against an interleaved stream — the same + // continuous-drain property the runner relies on. + var stdout, stderr []byte + var rerr, eerr error + var wg sync.WaitGroup + wg.Add(2) + go func() { defer wg.Done(); stdout, rerr = io.ReadAll(gs.Stdout) }() + go func() { defer wg.Done(); stderr, eerr = io.ReadAll(gs.Stderr) }() + wg.Wait() + if rerr != nil { + t.Fatalf("reading stdout: %v", rerr) + } + if eerr != nil { + t.Fatalf("reading stderr: %v", eerr) + } + if string(stdout) != "o1o2" { + t.Fatalf("stdout = %q, want o1o2", stdout) + } + if string(stderr) != "e1" { + t.Fatalf("stderr = %q, want e1", stderr) + } + if got := gs.Wait(); got.Code != 0 || got.Signal != 0 { + t.Fatalf("exit status = %+v, want {Code:0}", got) + } +} + +func TestGuestExec_Stream_StdinFramingAndClose(t *testing.T) { + gotStdin := make(chan []byte, 1) + gotClose := make(chan struct{}, 1) + ge := serveFakeGuest(t, &fakeGuest{ + execStreamFn: func(_ context.Context, stream *connect.BidiStream[compassv1.ExecStreamRequest, compassv1.ExecStreamResponse]) error { + if _, err := stream.Receive(); err != nil { // start + return err + } + startStream(t, stream, "exec-1") + var buf []byte + for { + req, err := stream.Receive() + if err != nil { + return err + } + switch f := req.GetFrame().(type) { + case *compassv1.ExecStreamRequest_Stdin: + buf = append(buf, f.Stdin...) + case *compassv1.ExecStreamRequest_StdinClose: + gotStdin <- buf + gotClose <- struct{}{} + return stream.Send(&compassv1.ExecStreamResponse{ + Frame: &compassv1.ExecStreamResponse_Exit{Exit: &compassv1.ExecExit{ExitCode: 0}}, + }) + } + } + }, + }) + + gs, err := ge.ExecStream(context.Background(), StreamCall{Command: []string{"cat"}}) + if err != nil { + t.Fatalf("ExecStream: %v", err) + } + if _, err := gs.Stdin.Write([]byte("hello ")); err != nil { + t.Fatalf("write stdin: %v", err) + } + if _, err := gs.Stdin.Write([]byte("world")); err != nil { + t.Fatalf("write stdin: %v", err) + } + if err := gs.Stdin.Close(); err != nil { + t.Fatalf("close stdin: %v", err) + } + + select { + case got := <-gotStdin: + if string(got) != "hello world" { + t.Fatalf("guest stdin = %q, want %q", got, "hello world") + } + case <-time.After(testTimeout): + t.Fatal("timed out waiting for stdin bytes") + } + select { + case <-gotClose: + case <-time.After(testTimeout): + t.Fatal("timed out waiting for stdin_close frame") + } + gs.Wait() +} + +func TestGuestExec_Stream_KillIssuesSignalAndUnblocksWait(t *testing.T) { + var mu sync.Mutex + var gotSignal int32 + signalled := make(chan struct{}) + ge := serveFakeGuest(t, &fakeGuest{ + execStreamFn: func(ctx context.Context, stream *connect.BidiStream[compassv1.ExecStreamRequest, compassv1.ExecStreamResponse]) error { + if _, err := stream.Receive(); err != nil { // start + return err + } + startStream(t, stream, "exec-kill") + // Wait for the kill signal to be observed, then emit a signalled + // exit frame — the guest's own reap of the SIGKILLed child. + <-signalled + return stream.Send(&compassv1.ExecStreamResponse{ + Frame: &compassv1.ExecStreamResponse_Exit{Exit: &compassv1.ExecExit{Signal: 9}}, + }) + }, + signalFn: func(_ context.Context, req *connect.Request[compassv1.SignalRequest]) (*connect.Response[compassv1.SignalResponse], error) { + mu.Lock() + gotSignal = req.Msg.GetSignal() + mu.Unlock() + if req.Msg.GetExecId() != "exec-kill" { + t.Errorf("signal exec_id = %q, want exec-kill", req.Msg.GetExecId()) + } + close(signalled) + return connect.NewResponse(&compassv1.SignalResponse{}), nil + }, + }) + + gs, err := ge.ExecStream(context.Background(), StreamCall{Command: []string{"sleep"}}) + if err != nil { + t.Fatalf("ExecStream: %v", err) + } + if err := gs.Kill(9); err != nil { + t.Fatalf("Kill: %v", err) + } + + got := gs.Wait() + if got.Signal != 9 { + t.Fatalf("exit status = %+v, want Signal 9", got) + } + mu.Lock() + defer mu.Unlock() + if gotSignal != 9 { + t.Fatalf("guest saw signal %d, want 9", gotSignal) + } +} + +func TestGuestExec_Stream_CtxCancelBreaksStream(t *testing.T) { + streamBroke := make(chan struct{}) + ge := serveFakeGuest(t, &fakeGuest{ + execStreamFn: func(ctx context.Context, stream *connect.BidiStream[compassv1.ExecStreamRequest, compassv1.ExecStreamResponse]) error { + if _, err := stream.Receive(); err != nil { // start + return err + } + startStream(t, stream, "exec-cancel") + // Block reading; the host cancel breaks the stream and Receive here + // returns an error the fake observes. + _, err := stream.Receive() + close(streamBroke) + return err + }, + }) + + ctx, cancel := context.WithCancel(context.Background()) + gs, err := ge.ExecStream(ctx, StreamCall{Command: []string{"sleep"}}) + if err != nil { + t.Fatalf("ExecStream: %v", err) + } + + cancel() + + // Wait unblocks on the broken stream and reports SIGKILL (deliberate + // teardown reaps the guest child). + got := gs.Wait() + if got.Signal != int(sigKill) { + t.Fatalf("exit status = %+v, want Signal SIGKILL after ctx cancel", got) + } + select { + case <-streamBroke: + case <-time.After(testTimeout): + t.Fatal("fake server never observed the stream break") + } +} + +func TestGuestExec_Stream_CleanExitReapsStdinPump(t *testing.T) { + // Regression for the stdin-pump leak: on a clean child exit where the caller + // never closes Stdin, pumpResponses must still reap pumpStdin (otherwise it + // is parked forever in stdinR.Read). reapStdinPump closes the pump's reader + // end via a defer that runs BEFORE close(s.done), so by the time Wait() + // returns the stdin pipe is closed at its read end — a caller Write then + // fails with the pipe-closed error. Before the fix the pump held the pipe + // open and the Write would block/succeed, so this is a true regression gate + // with no goroutine-count heuristic and no sleep. + ge := serveFakeGuest(t, &fakeGuest{ + execStreamFn: func(_ context.Context, stream *connect.BidiStream[compassv1.ExecStreamRequest, compassv1.ExecStreamResponse]) error { + if _, err := stream.Receive(); err != nil { // start + return err + } + startStream(t, stream, "exec-reap") + // Exit immediately with no output; the caller never touches Stdin. + return stream.Send(&compassv1.ExecStreamResponse{ + Frame: &compassv1.ExecStreamResponse_Exit{Exit: &compassv1.ExecExit{ExitCode: 0}}, + }) + }, + }) + + gs, err := ge.ExecStream(context.Background(), StreamCall{Command: []string{"true"}}) + if err != nil { + t.Fatalf("ExecStream: %v", err) + } + // Deliberately do NOT close gs.Stdin: the leak this guards is exactly the + // caller that leaves Stdin open on a clean exit. + if got := gs.Wait(); got.Code != 0 || got.Signal != 0 { + t.Fatalf("exit status = %+v, want {Code:0}", got) + } + // After Wait returns, reapStdinPump has closed the pump's read end, so the + // caller's write end is broken — the deterministic proof the pump unblocked. + if _, werr := gs.Stdin.Write([]byte("x")); werr == nil { + t.Fatal("Stdin.Write succeeded after a clean exit; pumpStdin was not reaped (goroutine leak)") + } +} + +func TestGuestExec_Stream_TransportBreakIsFailureExit(t *testing.T) { + // A non-EOF, non-cancel Receive error (a mid-stream transport/handler break + // with no exit frame) must classify as a failure exit (Code -1, no signal), + // NOT a clean exit or a deliberate kill — so isDeliberateKill treats it as a + // real failure. + ge := serveFakeGuest(t, &fakeGuest{ + execStreamFn: func(_ context.Context, stream *connect.BidiStream[compassv1.ExecStreamRequest, compassv1.ExecStreamResponse]) error { + if _, err := stream.Receive(); err != nil { // start + return err + } + startStream(t, stream, "exec-break") + // Return an error with no exit frame: the client's Receive surfaces a + // connect error (not io.EOF), and the ctx is never cancelled. + return errors.New("simulated mid-stream transport break") + }, + }) + + gs, err := ge.ExecStream(context.Background(), StreamCall{Command: []string{"sleep"}}) + if err != nil { + t.Fatalf("ExecStream: %v", err) + } + got := gs.Wait() + if got.Code != -1 || got.Signal != 0 { + t.Fatalf("exit status = %+v, want {Code:-1} (transport break is a failure, not a kill or clean exit)", got) + } +} + +func TestGuestExec_Stream_FirstFrameNotStartedIsError(t *testing.T) { + // A spawn failure surfaces as a non-Started first frame: the guest sends an + // Exit (or any non-Started) frame instead of ExecStarted. ExecStream must + // return an error rather than a live GuestStream — and reap both stream + // halves on that early-error return (the defer'd CloseRequest/CloseResponse), + // so a failed spawn leaks neither the response reader nor its goroutine. + ge := serveFakeGuest(t, &fakeGuest{ + execStreamFn: func(_ context.Context, stream *connect.BidiStream[compassv1.ExecStreamRequest, compassv1.ExecStreamResponse]) error { + if _, err := stream.Receive(); err != nil { // start + return err + } + // Skip the ExecStarted frame the client awaits: send an exit frame + // first, simulating a guest that failed to spawn the child. + return stream.Send(&compassv1.ExecStreamResponse{ + Frame: &compassv1.ExecStreamResponse_Exit{Exit: &compassv1.ExecExit{ExitCode: 127}}, + }) + }, + }) + + gs, err := ge.ExecStream(context.Background(), StreamCall{Command: []string{"nonexistent"}}) + if err == nil { + t.Fatalf("ExecStream with a non-started first frame = %+v, nil error; want an error", gs) + } + if gs != nil { + t.Fatalf("ExecStream returned a non-nil stream (%+v) alongside an error", gs) + } +} diff --git a/go/internal/runtime/podman.go b/go/internal/runtime/podman.go index 67d386cfd..08e5ca3e0 100644 --- a/go/internal/runtime/podman.go +++ b/go/internal/runtime/podman.go @@ -217,12 +217,38 @@ type StreamingIO struct { type ChildHandle struct { cmd *exec.Cmd cancel context.CancelFunc + // killFunc/waitFunc back a handle over a remote exec (the microVM + // GuestExec) rather than a local *exec.Cmd. When killFunc is non-nil this + // handle uses the funcs path; otherwise it drives cmd/cancel as before, so + // the podman path is untouched. See newChildHandleFuncs. + killFunc func() error + waitFunc func() error +} + +// newChildHandleFuncs builds a ChildHandle over a kill/wait function pair +// rather than a local *exec.Cmd — the adaptation for a remote streaming exec +// (the microVM GuestExec, design §(c)), whose child lives in the guest and has +// no host-side *exec.Cmd. The exported Kill/Wait/Terminate surface is +// unchanged; only the backing differs. +// +// kill must not block the caller past a short internal deadline — the microVM +// Kill issues a Signal RPC and the teardown path cannot stall on a wedged +// transport (the VMM-kill escalation is the backstop), matching podman's Kill +// being an instantaneous local cancel. wait blocks until the exec's exit is +// observed (the demux goroutine sees the exit frame), returning nil on exit 0 +// and otherwise an error carrying the code/signal (a *runtime.ExitStatusError +// for a signalled exit, so isDeliberateKill recognizes a deliberate kill). +func newChildHandleFuncs(kill func() error, wait func() error) *ChildHandle { + return &ChildHandle{killFunc: kill, waitFunc: wait} } // Kill signals the exec (SIGKILL) via its context, without reaping — the reap // happens in Wait. The session manager kills a container's exec on a deliberate // stop/teardown; killing an already-exited process is not an error. func (h *ChildHandle) Kill() error { + if h.killFunc != nil { + return h.killFunc() + } h.cancel() return nil } @@ -232,6 +258,9 @@ func (h *ChildHandle) Kill() error { // watches this to tell an unexpected agent exit (crash) from a deliberate // stop/teardown kill. func (h *ChildHandle) Wait() error { + if h.waitFunc != nil { + return h.waitFunc() + } return h.cmd.Wait() } @@ -243,6 +272,14 @@ func (h *ChildHandle) Wait() error { // observe the exit between the two — e.g. distinguishing a crash from a // deliberate teardown; Terminate is the safe default for the abandon path. func (h *ChildHandle) Terminate() error { + if h.killFunc != nil { + killErr := h.killFunc() + waitErr := h.waitFunc() + if waitErr != nil { + return waitErr + } + return killErr + } h.cancel() return h.cmd.Wait() }