feat(driver): stream external runtime exec and logs

.github/workflows/pr-ci.yml

@@ -71,7 +71,8 @@ jobs:
           - ubuntu-latest
           - macos-latest
     runs-on: ${{ matrix.runner }}
-    timeout-minutes: 30
+    # Leave room for race compilation and the existing crypto tests on macOS.
+    timeout-minutes: ${{ matrix.runner == 'macos-latest' && 45 || 30 }}
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
 

pkg/driver/external/exec.go

@@ -0,0 +1,181 @@
+package external
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+
+	"github.com/devsy-org/devsy-runtime-sdk/runtimev1"
+	"github.com/devsy-org/devsy/pkg/driver"
+	"github.com/devsy-org/devsy/pkg/subprocess"
+)
+
+var (
+	errExecComplete = errors.New("external runtime Exec stream completed")
+
+	_ driver.Driver         = (*Host)(nil)
+	_ driver.ArgvExecDriver = (*Host)(nil)
+)
+
+// CommandExitError describes an ordinary command failure, distinct from an RPC failure.
+type CommandExitError struct {
+	Code   int32
+	Signal string
+}
+
+func (e *CommandExitError) Error() string {
+	if e.Signal != "" {
+		return fmt.Sprintf("external runtime command terminated by signal %q", e.Signal)
+	}
+	return fmt.Sprintf("external runtime command exited with code %d", e.Code)
+}
+
+func (e *CommandExitError) ExitCode() int { return int(e.Code) }
+
+// CommandDevContainer uses the container shell; RawStdout preserves protocol bytes.
+// Once streaming starts, the host closes closable stdin on completion or cancellation.
+// Blocking readers must unblock on Close or caller cancellation; writers must return
+// from Write. Arbitrary io.Reader/io.Writer calls cannot be interrupted by a context.
+func (h *Host) CommandDevContainer(ctx context.Context, params *driver.CommandParams) error {
+	if params == nil {
+		return errors.New("external runtime command parameters are missing")
+	}
+	return h.exec(ctx, &runtimev1.ExecStart{
+		WorkspaceId: params.WorkspaceID,
+		Argv:        []string{"/bin/sh", "-c", params.Command},
+		User:        params.User,
+	}, driver.Streams{Stdin: params.Stdin, Stdout: params.Stdout, Stderr: params.Stderr}, params.RawStdout)
+}
+
+// CommandContainerArgv preserves literal argv and raw stdout for binary agent injection.
+// Stdin ownership and cancellation follow CommandDevContainer.
+func (h *Host) CommandContainerArgv(
+	ctx context.Context,
+	workspaceID string,
+	argv []string,
+	streams driver.Streams,
+) error {
+	return h.exec(
+		ctx,
+		&runtimev1.ExecStart{WorkspaceId: workspaceID, Argv: argv, User: "root"},
+		streams,
+		true,
+	)
+}
+
+func (h *Host) exec(
+	ctx context.Context,
+	start *runtimev1.ExecStart,
+	streams driver.Streams,
+	raw bool,
+) error {
+	if start.WorkspaceId == "" || len(start.Argv) == 0 || start.Argv[0] == "" {
+		return errors.New("external runtime Exec requires workspace ID and a nonempty executable")
+	}
+	operation := h.forWorkspace(start.WorkspaceId, nil)
+	output, flush := operation.commandOutput(streams, raw)
+	var exit *runtimev1.ExecExit
+	err := operation.call(ctx, "Exec", func(client runtimev1.RuntimeDriverClient) error {
+		var err error
+		exit, err = execStream(ctx, client, start, output)
+		return err
+	})
+	return operation.commandResult(ctx, exit, err, flush())
+}
+
+func (h *Host) commandResult(
+	ctx context.Context,
+	exit *runtimev1.ExecExit,
+	err, flushErr error,
+) error {
+	if err != nil {
+		return err
+	}
+	if flushErr != nil {
+		return h.operationError(ctx, "Exec output", flushErr)
+	}
+	if exit.ExitCode != 0 || exit.Signal != "" {
+		return &CommandExitError{
+			Code:   exit.ExitCode,
+			Signal: h.redactor.Redact(exit.Signal),
+		}
+	}
+	return nil
+}
+
+func (h *Host) commandOutput(streams driver.Streams, raw bool) (driver.Streams, func() error) {
+	if streams.Stdout == nil {
+		streams.Stdout = io.Discard
+	}
+	if streams.Stderr == nil {
+		streams.Stderr = io.Discard
+	}
+	stdout := &subprocess.StreamingRedactingWriter{Next: streams.Stdout, Redactor: h.redactor}
+	stderr := &subprocess.StreamingRedactingWriter{Next: streams.Stderr, Redactor: h.redactor}
+	if !raw {
+		streams.Stdout = stdout
+	}
+	streams.Stderr = stderr
+	return streams, func() error { return errors.Join(stdout.Flush(), stderr.Flush()) }
+}
+
+func execStream(
+	ctx context.Context,
+	client runtimev1.RuntimeDriverClient,
+	start *runtimev1.ExecStart,
+	streams driver.Streams,
+) (*runtimev1.ExecExit, error) {
+	streamCtx, cancel := context.WithCancelCause(ctx)
+	defer cancel(nil)
+	stream, err := client.Exec(streamCtx)
+	if err != nil {
+		return nil, err
+	}
+	if err := stream.Send(&runtimev1.ExecClientMessage{
+		Payload: &runtimev1.ExecClientMessage_Start{Start: start},
+	}); err != nil && !errors.Is(err, io.EOF) {
+		return nil, err
+	}
+	exit, err := exchangeExec(streamCtx, cancel, stream, streams)
+	if ctx.Err() != nil {
+		return nil, ctx.Err()
+	}
+	return exit, err
+}
+
+func exchangeExec(
+	ctx context.Context,
+	cancel context.CancelCauseFunc,
+	stream runtimev1.RuntimeDriver_ExecClient,
+	streams driver.Streams,
+) (*runtimev1.ExecExit, error) {
+	input := &execInput{reader: streams.Stdin}
+	closeInput := sync.OnceFunc(input.close)
+	stop := context.AfterFunc(ctx, closeInput)
+	defer stop()
+	inputDone := make(chan struct{})
+	go func() {
+		defer close(inputDone)
+		err := sendExecInput(ctx, stream, input)
+		if err != nil && !errors.Is(err, io.EOF) {
+			cancel(err)
+		}
+	}()
+	exit, receiveErr := receiveExecOutput(stream, streams)
+	// Distinguish ordinary cleanup from an input failure that wins cancellation.
+	// Cancel before joining: Send may be flow-controlled and Read may be blocked.
+	cancel(errExecComplete)
+	closeInput()
+	<-inputDone
+	inputCause := context.Cause(ctx)
+	if inputCause != nil && inputCause != errExecComplete {
+		return nil, inputCause
+	}
+	if err := input.readError(); err != nil {
+		return nil, err
+	}
+	// Recv owns command completion, including a command that exits before stdin EOF.
+	return exit, receiveErr
+}

pkg/driver/external/exec_exchange_test.go

@@ -0,0 +1,137 @@
+package external
+
+import (
+	"context"
+	"errors"
+	"io"
+	"os"
+	"testing/iotest"
+
+	"github.com/devsy-org/devsy-runtime-sdk/runtimev1"
+	"github.com/devsy-org/devsy/pkg/driver"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func (s *HostSuite) TestExecInputFailureSurvivesBufferedSuccessfulExit() {
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(nil)
+	want := errors.New("stdin read failed before command completion")
+	stream := &completedExecStream{ready: ctx.Done()}
+	exit, err := exchangeExec(ctx, cancel, stream, driver.Streams{
+		Stdin: iotest.ErrReader(want), Stdout: io.Discard, Stderr: io.Discard,
+	})
+	s.ErrorIs(err, want)
+	s.Nil(exit)
+}
+
+func (s *HostSuite) TestExecInputFailureAtCleanupBoundarySurvivesSuccessfulExit() {
+	for _, completionFirst := range []bool{false, true} {
+		name := "input failure wins cancellation"
+		if completionFirst {
+			name = "completion wins cancellation"
+		}
+		s.Run(name, func() {
+			ctx, cancel := context.WithCancelCause(context.Background())
+			defer cancel(nil)
+			want := errors.New("stdin failure at cleanup boundary")
+			input, producer := io.Pipe()
+			defer func() { _ = producer.Close() }()
+			ready := make(chan struct{})
+			inputFailed := make(chan struct{})
+			cancelAtBoundary := func(cause error) {
+				if errors.Is(cause, want) {
+					cancel(cause)
+					close(inputFailed)
+					return
+				}
+				if completionFirst {
+					cancel(cause)
+				}
+				_ = producer.CloseWithError(want)
+				<-inputFailed
+				cancel(cause)
+			}
+			exit, err := exchangeExec(
+				ctx,
+				cancelAtBoundary,
+				&completedExecStream{ready: ready},
+				driver.Streams{
+					Stdin:  &readyInputReader{Reader: input, ready: ready},
+					Stdout: io.Discard,
+					Stderr: io.Discard,
+				},
+			)
+			s.ErrorIs(err, want)
+			s.Nil(exit)
+		})
+	}
+}
+
+func (s *HostSuite) TestExecCleanupInputErrorsPreserveSuccessfulExit() {
+	for _, cleanupErr := range []error{
+		io.ErrClosedPipe, context.Canceled, status.Error(codes.Canceled, "stream canceled"),
+		os.ErrClosed, io.ErrUnexpectedEOF, errors.New("custom reader closed"),
+	} {
+		s.Run(cleanupErr.Error(), func() {
+			ctx, cancel := context.WithCancelCause(context.Background())
+			defer cancel(nil)
+			input := &closedInputReader{
+				ready: make(chan struct{}), closed: make(chan struct{}), err: cleanupErr,
+			}
+			exit, err := exchangeExec(
+				ctx, cancel, &completedExecStream{ready: input.ready},
+				driver.Streams{Stdin: input, Stdout: io.Discard, Stderr: io.Discard},
+			)
+			s.NoError(err)
+			s.NotNil(exit)
+		})
+	}
+}
+
+// Models a terminal exit already buffered when an input failure cancels the RPC.
+type completedExecStream struct {
+	runtimev1.RuntimeDriver_ExecClient
+	ready    <-chan struct{}
+	exitSent bool
+}
+
+func (s *completedExecStream) Recv() (*runtimev1.ExecServerMessage, error) {
+	if s.exitSent {
+		return nil, io.EOF
+	}
+	<-s.ready
+	s.exitSent = true
+	return &runtimev1.ExecServerMessage{
+		Payload: &runtimev1.ExecServerMessage_Exit{Exit: &runtimev1.ExecExit{}},
+	}, nil
+}
+
+// Starts completion only after the input pump is blocked inside Read.
+type readyInputReader struct {
+	io.Reader
+	ready chan struct{}
+}
+
+func (r *readyInputReader) Read(p []byte) (int, error) {
+	close(r.ready)
+	return r.Reader.Read(p)
+}
+
+// Models a reader whose Close unblocks Read with its own error type.
+type closedInputReader struct {
+	ready  chan struct{}
+	closed chan struct{}
+	err    error
+}
+
+func (r *closedInputReader) Read([]byte) (int, error) {
+	close(r.ready)
+	<-r.closed
+	return 0, r.err
+}
+
+func (r *closedInputReader) Close() error {
+	close(r.closed)
+	return nil
+}

pkg/driver/external/exec_input.go

@@ -0,0 +1,46 @@
+package external
+
+import (
+	"errors"
+	"io"
+	"sync"
+)
+
+// Serialize failure recording with owned closure: errors observed before Close
+// survive stream completion; errors from unblocking Read during Close do not.
+type execInput struct {
+	reader io.Reader
+	mu     sync.Mutex
+	closed bool
+	err    error
+}
+
+func (r *execInput) Read(data []byte) (int, error) {
+	if r.reader == nil {
+		return 0, io.EOF
+	}
+	n, err := r.reader.Read(data)
+	if err != nil && !errors.Is(err, io.EOF) {
+		r.mu.Lock()
+		if !r.closed {
+			r.err = err
+		}
+		r.mu.Unlock()
+	}
+	return n, err
+}
+
+func (r *execInput) close() {
+	if closer, ok := r.reader.(io.Closer); ok {
+		r.mu.Lock()
+		r.closed = true
+		r.mu.Unlock()
+		_ = closer.Close()
+	}
+}
+
+func (r *execInput) readError() error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.err
+}