From 4f4383015579b1a3d007b6231cae3bab3cf7058e Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 5 Oct 2026 20:53:13 -0600 Subject: [PATCH 1/8] feat(driver): stream external runtime exec and logs --- pkg/driver/external/exec.go | 167 ++++++++++++ pkg/driver/external/exec_test.go | 258 ++++++++++++++++++ pkg/driver/external/host.go | 2 +- .../external/internal/testfixture/main.go | 5 +- .../external/internal/testfixture/stream.go | 123 +++++++++ pkg/driver/external/logs.go | 57 ++++ pkg/driver/external/logs_test.go | 62 +++++ pkg/driver/external/stream.go | 120 ++++++++ .../docs/developing-providers/driver.mdx | 14 +- 9 files changed, 804 insertions(+), 4 deletions(-) create mode 100644 pkg/driver/external/exec.go create mode 100644 pkg/driver/external/exec_test.go create mode 100644 pkg/driver/external/internal/testfixture/stream.go create mode 100644 pkg/driver/external/logs.go create mode 100644 pkg/driver/external/logs_test.go create mode 100644 pkg/driver/external/stream.go diff --git a/pkg/driver/external/exec.go b/pkg/driver/external/exec.go new file mode 100644 index 0000000000..fc89022a36 --- /dev/null +++ b/pkg/driver/external/exec.go @@ -0,0 +1,167 @@ +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 ( + _ 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") + } + output, flush := h.commandOutput(streams, raw) + var exit *runtimev1.ExecExit + err := h.call(ctx, "Exec", func(client runtimev1.RuntimeDriverClient) error { + var err error + exit, err = execStream(ctx, client, start, output) + return err + }) + flushErr := flush() + 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) { + closeInput := sync.OnceFunc(func() { + if closer, ok := streams.Stdin.(io.Closer); ok { + _ = closer.Close() + } + }) + stop := context.AfterFunc(ctx, closeInput) + defer stop() + inputDone := make(chan struct{}) + go func() { + defer close(inputDone) + err := sendExecInput(ctx, stream, streams.Stdin) + if err != nil && !errors.Is(err, io.EOF) { + cancel(err) + } + }() + exit, receiveErr := receiveExecOutput(stream, streams) + inputCause := context.Cause(ctx) + // Cancel before joining: Send may be flow-controlled and Read may be blocked. + cancel(nil) + closeInput() + <-inputDone + if receiveErr != nil && inputCause != nil { + return nil, inputCause + } + // Recv owns command completion, including a command that exits before stdin EOF. + return exit, receiveErr +} diff --git a/pkg/driver/external/exec_test.go b/pkg/driver/external/exec_test.go new file mode 100644 index 0000000000..49ae6bf118 --- /dev/null +++ b/pkg/driver/external/exec_test.go @@ -0,0 +1,258 @@ +package external + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "os" + "path/filepath" + "strings" + "testing/iotest" + "time" + + "github.com/devsy-org/devsy-runtime-sdk/conformance/fake" + "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" +) + +const ( + earlyExitCommand = "early-exit" + echoCommand = "echo" +) + +func (s *HostSuite) TestExecLargeDuplexBinaryStream() { + host := s.runningHost(fake.Conformance) + // Larger than the gRPC flow-control window; NUL and invalid UTF-8 stay intact. + data := bytes.Repeat([]byte{0, 255, 'a', '\n', 128}, 1<<20) + var stdout, stderr bytes.Buffer + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + err := host.CommandContainerArgv(ctx, fixtureWorkspace, []string{"duplex"}, driver.Streams{ + Stdin: bytes.NewReader(data), Stdout: &stdout, Stderr: &stderr, + }) + s.Require().NoError(err) + s.Equal(data, stdout.Bytes()) + s.Equal(data, stderr.Bytes()) +} + +func (s *HostSuite) TestExecExactArgvAndShellMetadata() { + host := s.host("stream-metadata") + argv := []string{"space name", "", "é", "$(touch nope)", "quote'\""} + var stdout bytes.Buffer + s.Require().NoError(host.CommandContainerArgv(context.Background(), fixtureWorkspace, argv, + driver.Streams{Stdout: &stdout})) + var start runtimev1.ExecStart + s.Require().NoError(json.Unmarshal(stdout.Bytes(), &start)) + s.Equal(argv, start.Argv) + s.Equal("root", start.User) + s.Equal(fixtureWorkspace, start.WorkspaceId) + s.False(start.Tty) + stdout.Reset() + command := "printf '%s' '$HOME; quoted'" + s.Require().NoError(host.CommandDevContainer(context.Background(), &driver.CommandParams{ + WorkspaceID: fixtureWorkspace, User: "guest", Command: command, Stdout: &stdout, + })) + s.Require().NoError(json.Unmarshal(stdout.Bytes(), &start)) + s.Equal([]string{"/bin/sh", "-c", command}, start.Argv) + s.Equal("guest", start.User) +} + +func (s *HostSuite) TestExecTextRedactionAndRawStdout() { + s.T().Setenv("FIXTURE_TOKEN", "private-canary-value") + host := s.host("stream-secret") + for _, raw := range []bool{false, true} { + var stdout, stderr bytes.Buffer + s.Require().NoError(host.CommandDevContainer(context.Background(), &driver.CommandParams{ + WorkspaceID: fixtureWorkspace, Command: fixtureImage, RawStdout: raw, + Stdout: &stdout, Stderr: &stderr, + })) + s.Equal(raw, strings.Contains(stdout.String(), "private-canary-value")) + s.NotContains(stderr.String(), "private-canary-value") + s.Contains(stdout.String(), "safe-tail") + s.Contains(stderr.String(), "safe-tail") + } +} + +func (s *HostSuite) TestExecEmptyInputAndExitOutcomes() { + host := s.runningHost(fake.Conformance) + for _, command := range []string{echoCommand, earlyExitCommand, "nonzero", "signal"} { + s.Run(command, func() { + err := host.CommandContainerArgv(context.Background(), fixtureWorkspace, + []string{command}, driver.Streams{}) + if command == echoCommand || command == earlyExitCommand { + s.NoError(err) + return + } + var exit *CommandExitError + s.Require().ErrorAs(err, &exit) + if command == "signal" { + s.Equal("TERM", exit.Signal) + s.Zero(exit.ExitCode()) + } else { + s.Equal(7, exit.ExitCode()) + } + }) + } + err := host.CommandContainerArgv(context.Background(), fixtureWorkspace, + []string{"error", "PERMISSION_DENIED"}, driver.Streams{}) + s.Equal(codes.PermissionDenied, status.Code(err)) + var failure *RuntimeError + s.Require().ErrorAs(err, &failure) + s.Equal(runtimev1.RuntimeErrorCode_RUNTIME_ERROR_CODE_PERMISSION_DENIED, failure.Category) +} + +func (s *HostSuite) TestExecRejectsMalformedCompletion() { + for mode, message := range map[string]string{ + "stream-missing-exit": "without a terminal exit", + "stream-duplicate-exit": "after its terminal exit", + "stream-after-exit": "after its terminal exit", + "stream-unset-frame": "invalid output frame", + "stream-empty-chunk": "1..32768 bytes", + "stream-large-chunk": "1..32768 bytes", + "stream-exit-rpc-error": "failure after exit", + } { + s.Run(mode, func() { + host := s.host(mode) + err := host.CommandContainerArgv(context.Background(), fixtureWorkspace, + []string{fixtureImage}, driver.Streams{}) + s.ErrorContains(err, message) + }) + } +} + +func (s *HostSuite) TestExecCancellationClosesInputAndReapsChild() { + host := s.host("stream-block-child") + input, producer := io.Pipe() + defer func() { _ = producer.Close() }() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + done <- host.CommandContainerArgv(ctx, fixtureWorkspace, + []string{fixtureImage}, driver.Streams{Stdin: input}) + }() + s.waitForStream(host) + cancel() + select { + case err := <-done: + s.ErrorIs(err, context.Canceled) + case <-time.After(5 * time.Second): + s.FailNow("Exec cancellation did not join its input pump") + } + _, err := producer.Write([]byte("late")) + s.ErrorIs(err, io.ErrClosedPipe) +} + +func (s *HostSuite) TestExecEarlyExitClosesBlockedInput() { + host := s.runningHost(fake.Conformance) + input, producer := io.Pipe() + defer func() { _ = producer.Close() }() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + s.NoError(host.CommandContainerArgv(ctx, fixtureWorkspace, + []string{earlyExitCommand}, driver.Streams{Stdin: input})) + _, err := producer.Write([]byte("late")) + s.ErrorIs(err, io.ErrClosedPipe) +} + +func (s *HostSuite) TestExecDeadlineAndCrash() { + for _, mode := range []string{fake.ExecSlow, fake.ExecCrash} { + s.Run(mode, func() { + host := s.runningHost(mode) + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + err := host.CommandContainerArgv( + ctx, + fixtureWorkspace, + []string{fixtureImage}, + driver.Streams{}, + ) + if mode == fake.ExecSlow { + s.ErrorIs(err, context.DeadlineExceeded) + } else { + s.Error(err) + } + }) + } +} + +func (s *HostSuite) TestExecInputAndOutputFailures() { + host := s.runningHost(fake.Conformance) + for _, streams := range []driver.Streams{ + {Stdin: iotest.ErrReader(errors.New("input failed"))}, + {Stdin: strings.NewReader("output"), Stdout: shortWriter{}}, + {Stdin: strings.NewReader("output"), Stderr: shortWriter{}}, + } { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + err := host.CommandContainerArgv(ctx, fixtureWorkspace, []string{"duplex"}, streams) + cancel() + s.Error(err) + s.NotErrorIs(err, context.DeadlineExceeded) + } +} + +func (s *HostSuite) TestExecReaderBoundarySemantics() { + host := s.runningHost(fake.Conformance) + data := []byte("final bytes\x00\xff") + for _, input := range []io.Reader{ + iotest.DataErrReader(bytes.NewReader(data)), + &emptyReadsReader{Reader: bytes.NewReader(data), remaining: 3}, + } { + var stdout bytes.Buffer + s.Require().NoError(host.CommandContainerArgv(context.Background(), fixtureWorkspace, + []string{echoCommand}, driver.Streams{Stdin: input, Stdout: &stdout})) + s.Equal(data, stdout.Bytes()) + } + err := host.CommandContainerArgv(context.Background(), fixtureWorkspace, + []string{echoCommand}, driver.Streams{Stdin: &emptyReadsReader{remaining: 100}}) + s.ErrorContains(err, io.ErrNoProgress.Error()) +} + +func (s *HostSuite) TestExecValidatesBeforeLaunch() { + host := s.host(fake.Normal) + host.config.Binaries[fixtureKey][0].Checksum = "bad" + s.ErrorContains(host.CommandDevContainer(context.Background(), nil), "parameters") + for _, argv := range [][]string{nil, {}, {""}} { + s.ErrorContains(host.CommandContainerArgv(context.Background(), fixtureWorkspace, + argv, driver.Streams{}), "nonempty executable") + } + s.ErrorContains(host.CommandContainerArgv(context.Background(), "", + []string{"exec"}, driver.Streams{}), "workspace ID") +} + +func (s *HostSuite) runningHost(mode string) *Host { + host := s.host(mode) + s.Require().NoError(host.RunImage(context.Background(), &runtimev1.RunImageRequest{ + WorkspaceId: fixtureWorkspace, Image: fixtureImage, + })) + return host +} + +func (s *HostSuite) waitForStream(host *Host) { + directory := host.config.External.Args[3] + s.Require().Eventually(func() bool { + _, err := os.Stat(filepath.Join(directory, "ready")) + return err == nil + }, 5*time.Second, 10*time.Millisecond) +} + +type shortWriter struct{} + +func (shortWriter) Write(data []byte) (int, error) { return len(data) - 1, nil } + +type emptyReadsReader struct { + io.Reader + remaining int +} + +func (r *emptyReadsReader) Read(data []byte) (int, error) { + if r.remaining > 0 { + r.remaining-- + return 0, nil + } + return r.Reader.Read(data) +} diff --git a/pkg/driver/external/host.go b/pkg/driver/external/host.go index 70c8e76c75..7b92c9f780 100644 --- a/pkg/driver/external/host.go +++ b/pkg/driver/external/host.go @@ -22,7 +22,7 @@ import ( const startupTimeout = 15 * time.Second // Host opens an owned plugin session per operation. It is not yet registered -// as a workspace driver; Exec and Logs are added in the next integration stage. +// as a workspace driver; factory integration is a separate stage. type Host struct { config provider.ProviderAgentConfig binariesDir string diff --git a/pkg/driver/external/internal/testfixture/main.go b/pkg/driver/external/internal/testfixture/main.go index dc53e7b6dc..a259c7e5d7 100644 --- a/pkg/driver/external/internal/testfixture/main.go +++ b/pkg/driver/external/internal/testfixture/main.go @@ -8,6 +8,7 @@ import ( "os/exec" "path/filepath" "strconv" + "strings" "time" "github.com/devsy-org/devsy-runtime-sdk/conformance/fake" @@ -47,7 +48,7 @@ func serve(mode, state string, delay time.Duration) { } fixtureMode := mode if fixtureMode == "blocked" || fixtureMode == "environment" || fixtureMode == "block-info" || - fixtureMode == "env-error" { + fixtureMode == "env-error" || strings.HasPrefix(fixtureMode, "stream-") { fixtureMode = fake.Normal } runtime, err := fake.New(fake.Config{StateDir: state, Mode: fixtureMode}) @@ -89,7 +90,7 @@ func (f *fixture) Preflight( ctx context.Context, request *runtimev1.PreflightRequest, ) (*runtimev1.PreflightResponse, error) { - if f.mode != "blocked" { + if f.mode != "blocked" && f.mode != "stream-block-child" { return f.Driver.Preflight(ctx, request) } executable, err := os.Executable() diff --git a/pkg/driver/external/internal/testfixture/stream.go b/pkg/driver/external/internal/testfixture/stream.go new file mode 100644 index 0000000000..1966e4ee88 --- /dev/null +++ b/pkg/driver/external/internal/testfixture/stream.go @@ -0,0 +1,123 @@ +package main + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + + "github.com/devsy-org/devsy-runtime-sdk/runtimev1" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type execStream = grpc.BidiStreamingServer[runtimev1.ExecClientMessage, runtimev1.ExecServerMessage] + +func (f *fixture) Exec(stream execStream) error { + if !strings.HasPrefix(f.mode, "stream-") { + return f.Driver.Exec(stream) + } + first, err := stream.Recv() + if err != nil { + return err + } + if f.mode == "stream-block-child" { + _, err := f.Preflight(stream.Context(), &runtimev1.PreflightRequest{}) + return err + } + frames, err := fixtureFrames(f.mode, first.GetStart()) + if err != nil { + return err + } + for _, frame := range frames { + if err := stream.Send(frame); err != nil { + return err + } + } + if f.mode == "stream-exit-rpc-error" { + return status.Error(codes.Unavailable, "failure after exit") + } + return nil +} + +func fixtureFrames( + mode string, + start *runtimev1.ExecStart, +) ([]*runtimev1.ExecServerMessage, error) { + exit := &runtimev1.ExecServerMessage{ + Payload: &runtimev1.ExecServerMessage_Exit{Exit: &runtimev1.ExecExit{}}, + } + switch mode { + case "stream-secret": + return secretFrames(exit), nil + case "stream-metadata": + data, err := json.Marshal(start) + return []*runtimev1.ExecServerMessage{stdoutFrame(data), exit}, err + default: + return malformedFrames(mode, exit), nil + } +} + +func malformedFrames( + mode string, + exit *runtimev1.ExecServerMessage, +) []*runtimev1.ExecServerMessage { + switch mode { + case "stream-missing-exit": + return nil + case "stream-duplicate-exit": + return []*runtimev1.ExecServerMessage{exit, exit} + case "stream-after-exit": + return []*runtimev1.ExecServerMessage{exit, stdoutFrame([]byte("late"))} + case "stream-unset-frame": + return []*runtimev1.ExecServerMessage{{}, exit} + case "stream-empty-chunk": + return []*runtimev1.ExecServerMessage{stdoutFrame(nil), exit} + case "stream-large-chunk": + return []*runtimev1.ExecServerMessage{ + stdoutFrame(make([]byte, runtimev1.ChunkSize+1)), + exit, + } + default: + return []*runtimev1.ExecServerMessage{exit} + } +} + +func stdoutFrame(data []byte) *runtimev1.ExecServerMessage { + return &runtimev1.ExecServerMessage{ + Payload: &runtimev1.ExecServerMessage_Stdout{Stdout: &runtimev1.OutputChunk{Data: data}}, + } +} + +func secretFrames(exit *runtimev1.ExecServerMessage) []*runtimev1.ExecServerMessage { + var frames []*runtimev1.ExecServerMessage + for _, chunk := range []string{"private-canary-", "value", "safe-tail"} { + frames = append(frames, stdoutFrame([]byte(chunk)), &runtimev1.ExecServerMessage{ + Payload: &runtimev1.ExecServerMessage_Stderr{ + Stderr: &runtimev1.OutputChunk{Data: []byte(chunk)}, + }, + }) + } + return append(frames, exit) +} + +func (f *fixture) Logs( + request *runtimev1.LogsRequest, + stream grpc.ServerStreamingServer[runtimev1.OutputChunk], +) error { + switch f.mode { + case "stream-logs-block": + if err := os.WriteFile(filepath.Join(f.directory, "ready"), nil, 0o600); err != nil { + return err + } + <-stream.Context().Done() + return stream.Context().Err() + case "stream-logs-empty": + return stream.Send(&runtimev1.OutputChunk{}) + case "stream-logs-large": + return stream.Send(&runtimev1.OutputChunk{Data: make([]byte, runtimev1.ChunkSize+1)}) + default: + return f.Driver.Logs(request, stream) + } +} diff --git a/pkg/driver/external/logs.go b/pkg/driver/external/logs.go new file mode 100644 index 0000000000..a1403add88 --- /dev/null +++ b/pkg/driver/external/logs.go @@ -0,0 +1,57 @@ +package external + +import ( + "context" + "errors" + "io" + + "github.com/devsy-org/devsy-runtime-sdk/runtimev1" + "google.golang.org/grpc/codes" +) + +// GetDevContainerLogs writes the protocol's merged binary stream to stdout. +// Stderr is unused because Runtime Protocol v1 Logs has no channel identity. +func (h *Host) GetDevContainerLogs( + ctx context.Context, + workspaceID string, + stdout, _ io.Writer, +) error { + if err := ctx.Err(); err != nil { + return err + } + if workspaceID == "" { + return errors.New("external runtime Logs requires workspace ID") + } + if !h.info.Capabilities.Logs { + return &RuntimeError{ + Operation: "Logs", Code: codes.Unimplemented, + Category: runtimev1.RuntimeErrorCode_RUNTIME_ERROR_CODE_UNSUPPORTED, + Message: "runtime does not support Logs", + } + } + if stdout == nil { + stdout = io.Discard + } + return h.call(ctx, "Logs", func(client runtimev1.RuntimeDriverClient) error { + stream, err := client.Logs(ctx, &runtimev1.LogsRequest{WorkspaceId: workspaceID}) + if err != nil { + return err + } + return receiveLogs(stream, stdout) + }) +} + +func receiveLogs(stream runtimev1.RuntimeDriver_LogsClient, stdout io.Writer) error { + for { + chunk, err := stream.Recv() + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return err + } + if err := writeOutputChunk(stdout, chunk); err != nil { + return err + } + } +} diff --git a/pkg/driver/external/logs_test.go b/pkg/driver/external/logs_test.go new file mode 100644 index 0000000000..7a5aabd9e4 --- /dev/null +++ b/pkg/driver/external/logs_test.go @@ -0,0 +1,62 @@ +package external + +import ( + "bytes" + "context" + "time" + + "github.com/devsy-org/devsy-runtime-sdk/conformance/fake" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func (s *HostSuite) TestLogsMergedBinaryOutput() { + host := s.runningHost(fake.Normal) + var stdout, stderr bytes.Buffer + s.Require(). + NoError(host.GetDevContainerLogs(context.Background(), fixtureWorkspace, &stdout, &stderr)) + s.Equal([]byte("fake stdout\x00\xff\nfake stderr\n"), stdout.Bytes()) + s.Empty(stderr.Bytes()) + s.NoError(host.GetDevContainerLogs(context.Background(), fixtureWorkspace, nil, nil)) +} + +func (s *HostSuite) TestLogsCapabilityAndValidationBeforeLaunch() { + host := s.host(fake.Normal) + host.info.Capabilities.Logs = false + host.config.Binaries[fixtureKey][0].Checksum = "bad" + err := host.GetDevContainerLogs(context.Background(), fixtureWorkspace, nil, nil) + s.Equal(codes.Unimplemented, status.Code(err)) + s.ErrorContains(host.GetDevContainerLogs(context.Background(), "", nil, nil), "workspace ID") + ctx, cancel := context.WithCancel(context.Background()) + cancel() + s.ErrorIs(host.GetDevContainerLogs(ctx, fixtureWorkspace, nil, nil), context.Canceled) +} + +func (s *HostSuite) TestLogsWriterAndProtocolFailures() { + host := s.runningHost(fake.Normal) + s.ErrorContains( + host.GetDevContainerLogs(context.Background(), fixtureWorkspace, shortWriter{}, nil), + "short write", + ) + for _, mode := range []string{"stream-logs-empty", "stream-logs-large"} { + host := s.host(mode) + s.ErrorContains(host.GetDevContainerLogs(context.Background(), fixtureWorkspace, nil, nil), + "1..32768 bytes") + } +} + +func (s *HostSuite) TestLogsCancellationReapsProcess() { + host := s.host("stream-logs-block") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- host.GetDevContainerLogs(ctx, fixtureWorkspace, nil, nil) }() + s.waitForStream(host) + cancel() + select { + case err := <-done: + s.ErrorIs(err, context.Canceled) + case <-time.After(5 * time.Second): + s.FailNow("Logs cancellation did not return") + } +} diff --git a/pkg/driver/external/stream.go b/pkg/driver/external/stream.go new file mode 100644 index 0000000000..ad103c2fcf --- /dev/null +++ b/pkg/driver/external/stream.go @@ -0,0 +1,120 @@ +package external + +import ( + "bytes" + "context" + "errors" + "io" + + "github.com/devsy-org/devsy-runtime-sdk/runtimev1" + "github.com/devsy-org/devsy/pkg/driver" +) + +func sendExecInput( + ctx context.Context, + stream runtimev1.RuntimeDriver_ExecClient, + input io.Reader, +) error { + if input != nil { + if err := sendExecData(ctx, stream, input); err != nil { + return err + } + } + if err := stream.Send(&runtimev1.ExecClientMessage{ + Payload: &runtimev1.ExecClientMessage_CloseStdin{CloseStdin: &runtimev1.CloseStdin{}}, + }); err != nil { + return err + } + return stream.CloseSend() +} + +func sendExecData( + ctx context.Context, + stream runtimev1.RuntimeDriver_ExecClient, + input io.Reader, +) error { + buffer := make([]byte, runtimev1.ChunkSize) + for ctx.Err() == nil { + n, err := readExecChunk(ctx, input, buffer) + if n > 0 { + if sendErr := stream.Send(&runtimev1.ExecClientMessage{ + Payload: &runtimev1.ExecClientMessage_Stdin{Stdin: bytes.Clone(buffer[:n])}, + }); sendErr != nil { + return sendErr + } + } + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return err + } + } + return ctx.Err() +} + +func readExecChunk(ctx context.Context, input io.Reader, buffer []byte) (int, error) { + // Match bufio's bounded tolerance of readers returning (0, nil). + for range 100 { + if err := ctx.Err(); err != nil { + return 0, err + } + n, err := input.Read(buffer) + if n != 0 || err != nil { + return n, err + } + } + return 0, io.ErrNoProgress +} + +func receiveExecOutput( + stream runtimev1.RuntimeDriver_ExecClient, + streams driver.Streams, +) (*runtimev1.ExecExit, error) { + var exit *runtimev1.ExecExit + for { + frame, err := stream.Recv() + if errors.Is(err, io.EOF) { + if exit == nil { + return nil, errors.New("runtime Exec ended without a terminal exit") + } + return exit, nil + } + if err != nil { + return nil, err + } + if exit != nil { + return nil, errors.New("runtime Exec sent a frame after its terminal exit") + } + if terminal := frame.GetExit(); terminal != nil { + exit = terminal + continue + } + if err := writeExecFrame(frame, streams); err != nil { + return nil, err + } + } +} + +func writeExecFrame(frame *runtimev1.ExecServerMessage, streams driver.Streams) error { + switch payload := frame.GetPayload().(type) { + case *runtimev1.ExecServerMessage_Stdout: + return writeOutputChunk(streams.Stdout, payload.Stdout) + case *runtimev1.ExecServerMessage_Stderr: + return writeOutputChunk(streams.Stderr, payload.Stderr) + default: + return errors.New("runtime Exec sent an invalid output frame") + } +} + +func writeOutputChunk(writer io.Writer, chunk *runtimev1.OutputChunk) error { + data := chunk.GetData() + if len(data) == 0 || len(data) > runtimev1.ChunkSize { + return errors.New("runtime output chunk must contain 1..32768 bytes") + } + n, err := writer.Write(data) + if err == nil && n != len(data) { + return io.ErrShortWrite + } + return err +} diff --git a/sites/docs-devsy-sh/content/docs/developing-providers/driver.mdx b/sites/docs-devsy-sh/content/docs/developing-providers/driver.mdx index cee4d02c07..9ebb81aeca 100644 --- a/sites/docs-devsy-sh/content/docs/developing-providers/driver.mdx +++ b/sites/docs-devsy-sh/content/docs/developing-providers/driver.mdx @@ -64,7 +64,19 @@ Find, TargetArchitecture, RunImage, Start, Stop, and Delete calls. Each operatio uses a fresh plugin process owned by the SDK supervisor, exposed through a hidden helper in the running Devsy executable. Cancellation also covers startup and terminates the leased process tree. This is a prerequisite implementation; -Exec, Logs, and workspace driver registration remain separate stages. +workspace driver registration and image-backend integration remain a separate stage. + +The host also supports binary-safe Exec with separate stdout/stderr, literal argv, +stdin closure, and terminal exit status. Shell commands use `/bin/sh -c`; text +stdout and stderr are redacted, while `RawStdout` and argv execution preserve raw +stdout for protocol consumers. Logs capability is checked before launch and its +merged binary stream is written to stdout. + +Once Exec streaming starts, the host closes closable stdin on completion or +cancellation. Blocking input must unblock when closed or when the caller context +is canceled, and output writers must return from writes. Go cannot interrupt an +arbitrary blocking reader or writer. The host joins the input pump and reaps the +owned plugin session before returning. The host inherits its environment for compatibility with proxy, certificate, HOME/XDG, Docker, and runtime CLI settings. Provider environment overrides and From 2351cf817a2e8a7615ac85af75c07b979a8986fe Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 5 Oct 2026 22:15:26 -0600 Subject: [PATCH 2/8] fix(driver): preserve secrets across runtime streams --- pkg/driver/external/exec.go | 14 +-- pkg/driver/external/exec_exchange_test.go | 41 +++++++++ pkg/driver/external/host.go | 2 + .../external/internal/testfixture/main.go | 5 ++ .../external/internal/testfixture/stream.go | 67 +++++++++++++-- pkg/driver/external/lifecycle.go | 75 ++++++++-------- pkg/driver/external/logs.go | 15 ++-- pkg/driver/external/redaction.go | 37 ++++++++ pkg/driver/external/redaction_test.go | 85 +++++++++++++++++++ pkg/secrets/redact.go | 27 ++++++ pkg/secrets/redact_boundary_test.go | 58 +++++++++++++ .../docs/developing-providers/driver.mdx | 5 ++ 12 files changed, 375 insertions(+), 56 deletions(-) create mode 100644 pkg/driver/external/exec_exchange_test.go create mode 100644 pkg/driver/external/redaction.go create mode 100644 pkg/driver/external/redaction_test.go create mode 100644 pkg/secrets/redact_boundary_test.go diff --git a/pkg/driver/external/exec.go b/pkg/driver/external/exec.go index fc89022a36..1976bff484 100644 --- a/pkg/driver/external/exec.go +++ b/pkg/driver/external/exec.go @@ -72,9 +72,10 @@ func (h *Host) exec( if start.WorkspaceId == "" || len(start.Argv) == 0 || start.Argv[0] == "" { return errors.New("external runtime Exec requires workspace ID and a nonempty executable") } - output, flush := h.commandOutput(streams, raw) + operation := h.forWorkspace(start.WorkspaceId, nil) + output, flush := operation.commandOutput(streams, raw) var exit *runtimev1.ExecExit - err := h.call(ctx, "Exec", func(client runtimev1.RuntimeDriverClient) error { + err := operation.call(ctx, "Exec", func(client runtimev1.RuntimeDriverClient) error { var err error exit, err = execStream(ctx, client, start, output) return err @@ -84,10 +85,13 @@ func (h *Host) exec( return err } if flushErr != nil { - return h.operationError(ctx, "Exec output", flushErr) + return operation.operationError(ctx, "Exec output", flushErr) } if exit.ExitCode != 0 || exit.Signal != "" { - return &CommandExitError{Code: exit.ExitCode, Signal: h.redactor.Redact(exit.Signal)} + return &CommandExitError{ + Code: exit.ExitCode, + Signal: operation.redactor.Redact(exit.Signal), + } } return nil } @@ -159,7 +163,7 @@ func exchangeExec( cancel(nil) closeInput() <-inputDone - if receiveErr != nil && inputCause != nil { + if inputCause != nil { return nil, inputCause } // Recv owns command completion, including a command that exits before stdin EOF. diff --git a/pkg/driver/external/exec_exchange_test.go b/pkg/driver/external/exec_exchange_test.go new file mode 100644 index 0000000000..bb3ba22254 --- /dev/null +++ b/pkg/driver/external/exec_exchange_test.go @@ -0,0 +1,41 @@ +package external + +import ( + "context" + "errors" + "io" + "testing/iotest" + + "github.com/devsy-org/devsy-runtime-sdk/runtimev1" + "github.com/devsy-org/devsy/pkg/driver" +) + +func (s *HostSuite) TestExecInputFailureSurvivesBufferedSuccessfulExit() { + ctx, cancel := context.WithCancelCause(context.Background()) + defer cancel(nil) + want := errors.New("stdin read failed before command completion") + stream := &completedExecStream{ctx: ctx} + 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) +} + +// Models a terminal exit already buffered when an input failure cancels the RPC. +type completedExecStream struct { + runtimev1.RuntimeDriver_ExecClient + ctx context.Context + exitSent bool +} + +func (s *completedExecStream) Recv() (*runtimev1.ExecServerMessage, error) { + if s.exitSent { + return nil, io.EOF + } + <-s.ctx.Done() + s.exitSent = true + return &runtimev1.ExecServerMessage{ + Payload: &runtimev1.ExecServerMessage_Exit{Exit: &runtimev1.ExecExit{}}, + }, nil +} diff --git a/pkg/driver/external/host.go b/pkg/driver/external/host.go index 7b92c9f780..e9c317c71b 100644 --- a/pkg/driver/external/host.go +++ b/pkg/driver/external/host.go @@ -30,6 +30,7 @@ type Host struct { supervisorArgs []string environment []string redactor *secrets.Redactor + redactions *workspaceRedactions info *runtimev1.InfoResponse timeout time.Duration } @@ -83,6 +84,7 @@ func newHost(ctx context.Context, options hostOptions) (*Host, error) { environment: slices.Clone(options.environment), timeout: options.timeout, redactor: secrets.NewEnvironmentRedactor(options.environment), + redactions: &workspaceRedactions{values: make(map[string]map[string]struct{})}, } // Freeze provider declarations so later caller mutation cannot change identity. h.config.External.Args = slices.Clone(options.config.External.Args) diff --git a/pkg/driver/external/internal/testfixture/main.go b/pkg/driver/external/internal/testfixture/main.go index a259c7e5d7..aa8ce9cda1 100644 --- a/pkg/driver/external/internal/testfixture/main.go +++ b/pkg/driver/external/internal/testfixture/main.go @@ -122,6 +122,11 @@ func (f *fixture) RunImage( ctx context.Context, request *runtimev1.RunImageRequest, ) (*runtimev1.RunImageResponse, error) { + if f.mode == "stream-workspace-secret" { + if err := f.saveEnvironment(request); err != nil { + return nil, err + } + } if f.mode != "env-error" { return f.Driver.RunImage(ctx, request) } diff --git a/pkg/driver/external/internal/testfixture/stream.go b/pkg/driver/external/internal/testfixture/stream.go index 1966e4ee88..e830f2dfd7 100644 --- a/pkg/driver/external/internal/testfixture/stream.go +++ b/pkg/driver/external/internal/testfixture/stream.go @@ -1,7 +1,9 @@ package main import ( + "crypto/sha256" "encoding/json" + "fmt" "os" "path/filepath" "strings" @@ -26,7 +28,7 @@ func (f *fixture) Exec(stream execStream) error { _, err := f.Preflight(stream.Context(), &runtimev1.PreflightRequest{}) return err } - frames, err := fixtureFrames(f.mode, first.GetStart()) + frames, err := f.fixtureFrames(first.GetStart()) if err != nil { return err } @@ -41,21 +43,24 @@ func (f *fixture) Exec(stream execStream) error { return nil } -func fixtureFrames( - mode string, +func (f *fixture) fixtureFrames( start *runtimev1.ExecStart, ) ([]*runtimev1.ExecServerMessage, error) { exit := &runtimev1.ExecServerMessage{ Payload: &runtimev1.ExecServerMessage_Exit{Exit: &runtimev1.ExecExit{}}, } - switch mode { + switch f.mode { case "stream-secret": return secretFrames(exit), nil + case "stream-repeated-secret": + return repeatedSecretFrames(exit), nil + case "stream-workspace-secret": + return f.workspaceSecretFrames(start.WorkspaceId, exit) case "stream-metadata": data, err := json.Marshal(start) return []*runtimev1.ExecServerMessage{stdoutFrame(data), exit}, err default: - return malformedFrames(mode, exit), nil + return malformedFrames(f.mode, exit), nil } } @@ -91,8 +96,15 @@ func stdoutFrame(data []byte) *runtimev1.ExecServerMessage { } func secretFrames(exit *runtimev1.ExecServerMessage) []*runtimev1.ExecServerMessage { + return splitSecretFrames("private-canary-value", exit) +} + +func splitSecretFrames( + secret string, + exit *runtimev1.ExecServerMessage, +) []*runtimev1.ExecServerMessage { var frames []*runtimev1.ExecServerMessage - for _, chunk := range []string{"private-canary-", "value", "safe-tail"} { + for _, chunk := range []string{secret[:len(secret)/2], secret[len(secret)/2:], "safe-tail"} { frames = append(frames, stdoutFrame([]byte(chunk)), &runtimev1.ExecServerMessage{ Payload: &runtimev1.ExecServerMessage_Stderr{ Stderr: &runtimev1.OutputChunk{Data: []byte(chunk)}, @@ -102,6 +114,49 @@ func secretFrames(exit *runtimev1.ExecServerMessage) []*runtimev1.ExecServerMess return append(frames, exit) } +func repeatedSecretFrames(exit *runtimev1.ExecServerMessage) []*runtimev1.ExecServerMessage { + return []*runtimev1.ExecServerMessage{ + stdoutFrame([]byte("abcabc")), + { + Payload: &runtimev1.ExecServerMessage_Stderr{ + Stderr: &runtimev1.OutputChunk{Data: []byte("abcabc")}, + }, + }, + exit, + } +} + +func (f *fixture) environmentPath(workspaceID string) string { + return filepath.Join( + f.directory, + fmt.Sprintf("%x-environment", sha256.Sum256([]byte(workspaceID))), + ) +} + +func (f *fixture) saveEnvironment(request *runtimev1.RunImageRequest) error { + data, err := json.Marshal(request.Environment) + if err != nil { + return err + } + return os.WriteFile(f.environmentPath(request.WorkspaceId), data, 0o600) +} + +func (f *fixture) workspaceSecretFrames( + workspaceID string, + exit *runtimev1.ExecServerMessage, +) ([]*runtimev1.ExecServerMessage, error) { + // #nosec G304 -- Reads a hash-named fixture file in the test-owned directory. + data, err := os.ReadFile(f.environmentPath(workspaceID)) + if err != nil { + return nil, err + } + var environment map[string]string + if err := json.Unmarshal(data, &environment); err != nil { + return nil, err + } + return splitSecretFrames(environment["CUSTOM_VALUE"], exit), nil +} + func (f *fixture) Logs( request *runtimev1.LogsRequest, stream grpc.ServerStreamingServer[runtimev1.OutputChunk], diff --git a/pkg/driver/external/lifecycle.go b/pkg/driver/external/lifecycle.go index 7c739e2b05..1fe76e6021 100644 --- a/pkg/driver/external/lifecycle.go +++ b/pkg/driver/external/lifecycle.go @@ -9,7 +9,6 @@ import ( "github.com/devsy-org/devsy-runtime-sdk/runtimev1" "github.com/devsy-org/devsy/pkg/devcontainer/config" "github.com/devsy-org/devsy/pkg/driver" - "github.com/devsy-org/devsy/pkg/secrets" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -42,7 +41,8 @@ func (h *Host) FindDevContainer( return nil, errors.New("workspace ID is required") } var container *config.ContainerDetails - err := h.call(ctx, "Find", func(client runtimev1.RuntimeDriverClient) error { + operation := h.forWorkspace(workspaceID, nil) + err := operation.call(ctx, "Find", func(client runtimev1.RuntimeDriverClient) error { response, err := client.Find(ctx, &runtimev1.FindRequest{WorkspaceId: workspaceID}) if err != nil { return err @@ -55,7 +55,7 @@ func (h *Host) FindDevContainer( } container, err = convertContainer(response.Container) if container != nil { - container.State.Error = h.redactor.Redact(container.State.Error) + container.State.Error = operation.redactor.Redact(container.State.Error) } return err }) @@ -67,20 +67,21 @@ func (h *Host) TargetArchitecture(ctx context.Context, workspaceID string) (stri return "", errors.New("workspace ID is required") } var architecture string - err := h.call(ctx, "TargetArchitecture", func(client runtimev1.RuntimeDriverClient) error { - response, err := client.TargetArchitecture( - ctx, - &runtimev1.TargetArchitectureRequest{WorkspaceId: workspaceID}, - ) - if err != nil { - return err - } - architecture = response.GetArchitecture() - if architecture != "amd64" && architecture != "arm64" { - return fmt.Errorf("invalid runtime target architecture %q", architecture) - } - return nil - }) + err := h.forWorkspace(workspaceID, nil). + call(ctx, "TargetArchitecture", func(client runtimev1.RuntimeDriverClient) error { + response, err := client.TargetArchitecture( + ctx, + &runtimev1.TargetArchitectureRequest{WorkspaceId: workspaceID}, + ) + if err != nil { + return err + } + architecture = response.GetArchitecture() + if architecture != "amd64" && architecture != "arm64" { + return fmt.Errorf("invalid runtime target architecture %q", architecture) + } + return nil + }) return architecture, err } @@ -89,12 +90,7 @@ func (h *Host) RunImage(ctx context.Context, request *runtimev1.RunImageRequest) if err := h.validateRunImage(request); err != nil { return err } - operation := *h - environment := make([]string, 0, len(request.Environment)) - for key, value := range request.Environment { - environment = append(environment, key+"="+value) - } - operation.redactor = secrets.Combine(h.redactor, secrets.NewRedactor(environment)) + operation := h.forWorkspace(request.WorkspaceId, request.Environment) return operation.call(ctx, "RunImage", func(client runtimev1.RuntimeDriverClient) error { _, err := client.RunImage(ctx, request) return err @@ -124,33 +120,36 @@ func (h *Host) StartDevContainer(ctx context.Context, workspaceID string) error if workspaceID == "" { return errors.New("workspace ID is required") } - return h.call(ctx, "Start", func(client runtimev1.RuntimeDriverClient) error { - _, err := client.Start(ctx, &runtimev1.StartRequest{WorkspaceId: workspaceID}) - return err - }) + return h.forWorkspace(workspaceID, nil). + call(ctx, "Start", func(client runtimev1.RuntimeDriverClient) error { + _, err := client.Start(ctx, &runtimev1.StartRequest{WorkspaceId: workspaceID}) + return err + }) } func (h *Host) StopDevContainer(ctx context.Context, workspaceID string) error { if workspaceID == "" { return errors.New("workspace ID is required") } - return h.call(ctx, "Stop", func(client runtimev1.RuntimeDriverClient) error { - _, err := client.Stop(ctx, &runtimev1.StopRequest{WorkspaceId: workspaceID}) - return err - }) + return h.forWorkspace(workspaceID, nil). + call(ctx, "Stop", func(client runtimev1.RuntimeDriverClient) error { + _, err := client.Stop(ctx, &runtimev1.StopRequest{WorkspaceId: workspaceID}) + return err + }) } func (h *Host) DeleteDevContainer(ctx context.Context, workspaceID string) error { if workspaceID == "" { return errors.New("workspace ID is required") } - return h.call(ctx, "Delete", func(client runtimev1.RuntimeDriverClient) error { - _, err := client.Delete(ctx, &runtimev1.DeleteRequest{WorkspaceId: workspaceID}) - if status.Code(err) == codes.NotFound { - return nil - } - return err - }) + return h.forWorkspace(workspaceID, nil). + call(ctx, "Delete", func(client runtimev1.RuntimeDriverClient) error { + _, err := client.Delete(ctx, &runtimev1.DeleteRequest{WorkspaceId: workspaceID}) + if status.Code(err) == codes.NotFound { + return nil + } + return err + }) } func convertContainer(source *runtimev1.ContainerDetails) (*config.ContainerDetails, error) { diff --git a/pkg/driver/external/logs.go b/pkg/driver/external/logs.go index a1403add88..1742c00191 100644 --- a/pkg/driver/external/logs.go +++ b/pkg/driver/external/logs.go @@ -32,13 +32,14 @@ func (h *Host) GetDevContainerLogs( if stdout == nil { stdout = io.Discard } - return h.call(ctx, "Logs", func(client runtimev1.RuntimeDriverClient) error { - stream, err := client.Logs(ctx, &runtimev1.LogsRequest{WorkspaceId: workspaceID}) - if err != nil { - return err - } - return receiveLogs(stream, stdout) - }) + return h.forWorkspace(workspaceID, nil). + call(ctx, "Logs", func(client runtimev1.RuntimeDriverClient) error { + stream, err := client.Logs(ctx, &runtimev1.LogsRequest{WorkspaceId: workspaceID}) + if err != nil { + return err + } + return receiveLogs(stream, stdout) + }) } func receiveLogs(stream runtimev1.RuntimeDriver_LogsClient, stdout io.Writer) error { diff --git a/pkg/driver/external/redaction.go b/pkg/driver/external/redaction.go new file mode 100644 index 0000000000..61ff0e20fd --- /dev/null +++ b/pkg/driver/external/redaction.go @@ -0,0 +1,37 @@ +package external + +import ( + "sync" + + "github.com/devsy-org/devsy/pkg/secrets" +) + +// Shared through a pointer so operation snapshots never copy a live mutex. +// Keep prior values for the host's lifetime: retries and backend diagnostics +// can still contain secrets from an earlier or partially failed RunImage. +type workspaceRedactions struct { + mu sync.Mutex + values map[string]map[string]struct{} +} + +func (h *Host) forWorkspace(workspaceID string, environment map[string]string) *Host { + h.redactions.mu.Lock() + defer h.redactions.mu.Unlock() + known := h.redactions.values[workspaceID] + if known == nil && len(environment) > 0 { + known = make(map[string]struct{}) + h.redactions.values[workspaceID] = known + } + for _, value := range environment { + if value != "" { + known[value] = struct{}{} + } + } + entries := make([]string, 0, len(known)) + for value := range known { + entries = append(entries, "WORKSPACE_VALUE="+value) + } + operation := *h + operation.redactor = secrets.Combine(h.redactor, secrets.NewRedactor(entries)) + return &operation +} diff --git a/pkg/driver/external/redaction_test.go b/pkg/driver/external/redaction_test.go new file mode 100644 index 0000000000..e75e5ec218 --- /dev/null +++ b/pkg/driver/external/redaction_test.go @@ -0,0 +1,85 @@ +package external + +import ( + "bytes" + "context" + "fmt" + "sync" + + "github.com/devsy-org/devsy-runtime-sdk/conformance/fake" + "github.com/devsy-org/devsy-runtime-sdk/runtimev1" + "github.com/devsy-org/devsy/pkg/driver" +) + +const workspaceSecretKey = "CUSTOM_VALUE" + +func (s *HostSuite) TestExecRepeatedSecretAtStreamEnd() { + s.T().Setenv("FIXTURE_TOKEN", "abcabc") + host := s.host("stream-repeated-secret") + var stdout, stderr bytes.Buffer + s.Require().NoError(host.CommandDevContainer(context.Background(), &driver.CommandParams{ + WorkspaceID: fixtureWorkspace, Command: fixtureImage, Stdout: &stdout, Stderr: &stderr, + })) + s.Equal("***", stdout.String()) + s.Equal("***", stderr.String()) +} + +func (s *HostSuite) TestExecMasksWorkspaceEnvironmentAcrossFreshProcesses() { + host := s.host("stream-workspace-secret") + request := &runtimev1.RunImageRequest{ + WorkspaceId: fixtureWorkspace, Image: fixtureImage, + Environment: map[string]string{workspaceSecretKey: workspaceCanary}, + } + s.Require().NoError(host.RunImage(context.Background(), request)) + // Caller mutation cannot discard the value already installed in the runtime. + request.Environment[workspaceSecretKey] = "later-caller-value" + for _, raw := range []bool{false, true} { + var stdout, stderr bytes.Buffer + s.Require().NoError(host.CommandDevContainer(context.Background(), &driver.CommandParams{ + WorkspaceID: fixtureWorkspace, Command: fixtureImage, + Stdout: &stdout, Stderr: &stderr, RawStdout: raw, + })) + if raw { + s.Equal(workspaceCanary+"safe-tail", stdout.String()) + } else { + s.Equal("***safe-tail", stdout.String()) + } + s.Equal("***safe-tail", stderr.String()) + } +} + +func (s *HostSuite) TestWorkspaceRedactionRetainsFailedRequestsAndPriorValues() { + host := s.host("env-error") + for _, value := range []string{"first-private-value", "rotated-private-value", "first-private-value"} { + s.Error(host.RunImage(context.Background(), &runtimev1.RunImageRequest{ + WorkspaceId: fixtureWorkspace, Image: fixtureImage, + Environment: map[string]string{workspaceSecretKey: value}, + })) + } + operation := host.forWorkspace(fixtureWorkspace, nil) + s.Equal("*** ***", operation.redactor.Redact("first-private-value rotated-private-value")) + s.Len(host.redactions.values[fixtureWorkspace], 2) + s.Equal( + "first-private-value", + host.forWorkspace("other", nil).redactor.Redact("first-private-value"), + ) + s.Equal("first-private-value", host.redactor.Redact("first-private-value")) +} + +func (s *HostSuite) TestConcurrentWorkspaceRedactionSnapshots() { + host := s.host(fake.Normal) + const count = 20 + var workers sync.WaitGroup + for index := range count { + workers.Go(func() { + value := fmt.Sprintf("concurrent-private-%d-value", index) + host.forWorkspace(fixtureWorkspace, map[string]string{"VALUE": value}) + _ = host.forWorkspace(fixtureWorkspace, nil).redactor.Redact(value) + }) + } + workers.Wait() + snapshot := host.forWorkspace(fixtureWorkspace, nil) + for index := range count { + s.Equal("***", snapshot.redactor.Redact(fmt.Sprintf("concurrent-private-%d-value", index))) + } +} diff --git a/pkg/secrets/redact.go b/pkg/secrets/redact.go index 1223030cf5..9852aa7b2d 100644 --- a/pkg/secrets/redact.go +++ b/pkg/secrets/redact.go @@ -144,10 +144,37 @@ func (r *StreamingRedactor) RedactChunk(chunk string) string { if formatStart := incompleteCredentialURLStart(combined); formatStart >= 0 { cut = min(cut, formatStart) } + cut = r.base.completeSecretBoundary(combined, cut) r.pending = combined[cut:] return r.base.Redact(combined[:cut]) } +// Follow the replacer's leftmost, longest-first matches. A complete secret +// straddling cut must stay whole; overlapping matches must not make buffering +// grow across an arbitrarily long repeated value. +func (r *Redactor) completeSecretBoundary(value string, cut int) int { + if len(r.values) == 0 { + return cut + } + for offset := 0; offset < cut; { + length := r.secretMatchLength(value[offset:]) + if offset+length > cut { + return offset + } + offset += max(length, 1) + } + return cut +} + +func (r *Redactor) secretMatchLength(value string) int { + for _, secret := range r.values { + if strings.HasPrefix(value, secret) { + return len(secret) + } + } + return 0 +} + // incompleteCredentialURLStart returns the start of a URL suffix that may // still contain unredacted userinfo. Waiting for the terminating '@' prevents // a password from escaping when the URL is split across writes. diff --git a/pkg/secrets/redact_boundary_test.go b/pkg/secrets/redact_boundary_test.go new file mode 100644 index 0000000000..ae207c46d7 --- /dev/null +++ b/pkg/secrets/redact_boundary_test.go @@ -0,0 +1,58 @@ +package secrets_test + +import ( + "strings" + "testing" + + "github.com/devsy-org/devsy/pkg/secrets" + "github.com/stretchr/testify/suite" +) + +type StreamingBoundarySuite struct{ suite.Suite } + +func TestStreamingBoundarySuite(t *testing.T) { suite.Run(t, new(StreamingBoundarySuite)) } + +func (s *StreamingBoundarySuite) TestKnownSecretsAcrossEverySplit() { + for _, entries := range [][]string{ + {"TOKEN=abcabc"}, + {"TOKEN=aaaaa"}, + {"TOKEN=abababa"}, + {"A=abcabcX", "B=abcabc"}, + {"A=aba", "B=bab"}, + } { + redactor := secrets.NewRedactor(entries) + var values []string + for _, entry := range entries { + _, value, _ := strings.Cut(entry, "=") + values = append(values, value) + } + input := "ordinary " + strings.Join(values, " ") + want := redactor.Redact(input) + for split := 0; split <= len(input); split++ { + stream := secrets.NewStreamingRedactor(redactor) + got := stream.RedactChunk( + input[:split], + ) + stream.RedactChunk( + input[split:], + ) + stream.Flush() + s.Equal(want, got, "entries %v, split %d", entries, split) + } + stream := secrets.NewStreamingRedactor(redactor) + var output strings.Builder + for index := range len(input) { + output.WriteString(stream.RedactChunk(input[index : index+1])) + } + output.WriteString(stream.Flush()) + s.Equal(want, output.String(), "one-byte chunks, entries %v", entries) + } +} + +func (s *StreamingBoundarySuite) TestRepeatedOutputDoesNotBufferWholeStream() { + const count = 1 << 14 + input := strings.Repeat("abcabc", count) + redactor := secrets.NewRedactor([]string{"TOKEN=abcabc"}) + stream := secrets.NewStreamingRedactor(redactor) + emitted := stream.RedactChunk(input) + s.GreaterOrEqual(len(emitted), 3*(count-1)) + s.Equal(redactor.Redact(input), emitted+stream.Flush()) +} diff --git a/sites/docs-devsy-sh/content/docs/developing-providers/driver.mdx b/sites/docs-devsy-sh/content/docs/developing-providers/driver.mdx index 9ebb81aeca..a23a203fd0 100644 --- a/sites/docs-devsy-sh/content/docs/developing-providers/driver.mdx +++ b/sites/docs-devsy-sh/content/docs/developing-providers/driver.mdx @@ -72,6 +72,11 @@ stdout and stderr are redacted, while `RawStdout` and argv execution preserve ra stdout for protocol consumers. Logs capability is checked before launch and its merged binary stream is written to stdout. +The host retains environment values supplied through RunImage for that workspace +throughout its lifetime, including prior values from retries or failed requests. +Later text output, errors, and diagnostics use that workspace's redaction snapshot; +raw stdout and merged Logs remain binary streams for their callers to handle. + Once Exec streaming starts, the host closes closable stdin on completion or cancellation. Blocking input must unblock when closed or when the caller context is canceled, and output writers must return from writes. Go cannot interrupt an From fd0a89d19f7f2037e0f0139fa2b57027f419093e Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 5 Oct 2026 23:20:05 -0600 Subject: [PATCH 3/8] fix(driver): bound streaming redaction costs --- pkg/driver/external/exec.go | 18 ++++++++-- pkg/driver/external/host.go | 2 +- pkg/driver/external/redaction.go | 52 +++++++++++++++++++-------- pkg/driver/external/redaction_test.go | 25 +++++++++++-- pkg/secrets/redact.go | 16 ++++++--- pkg/secrets/redact_boundary_test.go | 23 ++++++++++++ pkg/secrets/redact_matcher.go | 28 +++++++++++++++ 7 files changed, 138 insertions(+), 26 deletions(-) create mode 100644 pkg/secrets/redact_matcher.go diff --git a/pkg/driver/external/exec.go b/pkg/driver/external/exec.go index 1976bff484..57f6ab78e1 100644 --- a/pkg/driver/external/exec.go +++ b/pkg/driver/external/exec.go @@ -80,17 +80,29 @@ func (h *Host) exec( exit, err = execStream(ctx, client, start, output) return err }) - flushErr := flush() + return operation.commandResult(ctx, exit, err, flush()) +} + +func (h *Host) commandResult( + ctx context.Context, + exit *runtimev1.ExecExit, + err, flushErr error, +) error { if err != nil { + // Cleanup can finish after the caller deadline even when the transport + // reported its cancellation just before the local context timer fired. + if ctx.Err() != nil { + return h.operationError(ctx, "Exec", err) + } return err } if flushErr != nil { - return operation.operationError(ctx, "Exec output", flushErr) + return h.operationError(ctx, "Exec output", flushErr) } if exit.ExitCode != 0 || exit.Signal != "" { return &CommandExitError{ Code: exit.ExitCode, - Signal: operation.redactor.Redact(exit.Signal), + Signal: h.redactor.Redact(exit.Signal), } } return nil diff --git a/pkg/driver/external/host.go b/pkg/driver/external/host.go index e9c317c71b..32cb967700 100644 --- a/pkg/driver/external/host.go +++ b/pkg/driver/external/host.go @@ -84,7 +84,7 @@ func newHost(ctx context.Context, options hostOptions) (*Host, error) { environment: slices.Clone(options.environment), timeout: options.timeout, redactor: secrets.NewEnvironmentRedactor(options.environment), - redactions: &workspaceRedactions{values: make(map[string]map[string]struct{})}, + redactions: &workspaceRedactions{workspaces: make(map[string]*workspaceRedaction)}, } // Freeze provider declarations so later caller mutation cannot change identity. h.config.External.Args = slices.Clone(options.config.External.Args) diff --git a/pkg/driver/external/redaction.go b/pkg/driver/external/redaction.go index 61ff0e20fd..c8c64213af 100644 --- a/pkg/driver/external/redaction.go +++ b/pkg/driver/external/redaction.go @@ -10,28 +10,50 @@ import ( // Keep prior values for the host's lifetime: retries and backend diagnostics // can still contain secrets from an earlier or partially failed RunImage. type workspaceRedactions struct { - mu sync.Mutex - values map[string]map[string]struct{} + mu sync.Mutex + workspaces map[string]*workspaceRedaction +} + +type workspaceRedaction struct { + mu sync.Mutex + values map[string]struct{} + redactor *secrets.Redactor } func (h *Host) forWorkspace(workspaceID string, environment map[string]string) *Host { h.redactions.mu.Lock() - defer h.redactions.mu.Unlock() - known := h.redactions.values[workspaceID] - if known == nil && len(environment) > 0 { - known = make(map[string]struct{}) - h.redactions.values[workspaceID] = known + workspace := h.redactions.workspaces[workspaceID] + if workspace == nil && len(environment) > 0 { + workspace = &workspaceRedaction{values: make(map[string]struct{}), redactor: h.redactor} + h.redactions.workspaces[workspaceID] = workspace } + h.redactions.mu.Unlock() + operation := *h + if workspace != nil { + operation.redactor = workspace.snapshot(h.redactor, environment) + } + return &operation +} + +func (w *workspaceRedaction) snapshot( + base *secrets.Redactor, + environment map[string]string, +) *secrets.Redactor { + w.mu.Lock() + defer w.mu.Unlock() + changed := false for _, value := range environment { - if value != "" { - known[value] = struct{}{} + if _, known := w.values[value]; value != "" && !known { + w.values[value] = struct{}{} + changed = true } } - entries := make([]string, 0, len(known)) - for value := range known { - entries = append(entries, "WORKSPACE_VALUE="+value) + if changed { + entries := make([]string, 0, len(w.values)) + for value := range w.values { + entries = append(entries, "WORKSPACE_VALUE="+value) + } + w.redactor = secrets.Combine(base, secrets.NewRedactor(entries)) } - operation := *h - operation.redactor = secrets.Combine(h.redactor, secrets.NewRedactor(entries)) - return &operation + return w.redactor } diff --git a/pkg/driver/external/redaction_test.go b/pkg/driver/external/redaction_test.go index e75e5ec218..f515608e08 100644 --- a/pkg/driver/external/redaction_test.go +++ b/pkg/driver/external/redaction_test.go @@ -58,7 +58,7 @@ func (s *HostSuite) TestWorkspaceRedactionRetainsFailedRequestsAndPriorValues() } operation := host.forWorkspace(fixtureWorkspace, nil) s.Equal("*** ***", operation.redactor.Redact("first-private-value rotated-private-value")) - s.Len(host.redactions.values[fixtureWorkspace], 2) + s.Len(host.redactions.workspaces[fixtureWorkspace].values, 2) s.Equal( "first-private-value", host.forWorkspace("other", nil).redactor.Redact("first-private-value"), @@ -66,6 +66,27 @@ func (s *HostSuite) TestWorkspaceRedactionRetainsFailedRequestsAndPriorValues() s.Equal("first-private-value", host.redactor.Redact("first-private-value")) } +func (s *HostSuite) TestWorkspaceRedactionReusesUnchangedSnapshots() { + host := s.host(fake.Normal) + first := host.forWorkspace( + fixtureWorkspace, + map[string]string{workspaceSecretKey: workspaceCanary}, + ) + unchanged := host.forWorkspace( + fixtureWorkspace, + map[string]string{workspaceSecretKey: workspaceCanary}, + ) + s.Same(first.redactor, unchanged.redactor) + s.Same(first.redactor, host.forWorkspace(fixtureWorkspace, nil).redactor) + updated := host.forWorkspace( + fixtureWorkspace, + map[string]string{workspaceSecretKey: "rotated-private-value"}, + ) + s.NotSame(first.redactor, updated.redactor) + s.Equal("rotated-private-value", first.redactor.Redact("rotated-private-value")) + s.Equal("*** ***", updated.redactor.Redact(workspaceCanary+" rotated-private-value")) +} + func (s *HostSuite) TestConcurrentWorkspaceRedactionSnapshots() { host := s.host(fake.Normal) const count = 20 @@ -73,7 +94,7 @@ func (s *HostSuite) TestConcurrentWorkspaceRedactionSnapshots() { for index := range count { workers.Go(func() { value := fmt.Sprintf("concurrent-private-%d-value", index) - host.forWorkspace(fixtureWorkspace, map[string]string{"VALUE": value}) + host.forWorkspace(fixtureWorkspace, map[string]string{workspaceSecretKey: value}) _ = host.forWorkspace(fixtureWorkspace, nil).redactor.Redact(value) }) } diff --git a/pkg/secrets/redact.go b/pkg/secrets/redact.go index 9852aa7b2d..e50880f246 100644 --- a/pkg/secrets/redact.go +++ b/pkg/secrets/redact.go @@ -21,6 +21,7 @@ type Redactor struct { replacer *strings.Replacer values []string maxLength int + matcher *secretNode } // NewRedactor masks the values (not keys) of KEY=VALUE entries; empty values are ignored. @@ -49,6 +50,7 @@ func NewRedactor(secretsEnv []string) *Redactor { replacer: strings.NewReplacer(pairs...), values: values, maxLength: len(values[0]), + matcher: newSecretMatcher(values), } } @@ -153,7 +155,7 @@ func (r *StreamingRedactor) RedactChunk(chunk string) string { // straddling cut must stay whole; overlapping matches must not make buffering // grow across an arbitrarily long repeated value. func (r *Redactor) completeSecretBoundary(value string, cut int) int { - if len(r.values) == 0 { + if cut == len(value) || r.matcher == nil { return cut } for offset := 0; offset < cut; { @@ -167,12 +169,16 @@ func (r *Redactor) completeSecretBoundary(value string, cut int) int { } func (r *Redactor) secretMatchLength(value string) int { - for _, secret := range r.values { - if strings.HasPrefix(value, secret) { - return len(secret) + node := r.matcher + length := 0 + for index := range len(value) { + node = node.children[value[index]] + if node == nil { + break } + length = max(length, node.length) } - return 0 + return length } // incompleteCredentialURLStart returns the start of a URL suffix that may diff --git a/pkg/secrets/redact_boundary_test.go b/pkg/secrets/redact_boundary_test.go index ae207c46d7..b3f3000134 100644 --- a/pkg/secrets/redact_boundary_test.go +++ b/pkg/secrets/redact_boundary_test.go @@ -1,6 +1,7 @@ package secrets_test import ( + "fmt" "strings" "testing" @@ -19,6 +20,7 @@ func (s *StreamingBoundarySuite) TestKnownSecretsAcrossEverySplit() { {"TOKEN=abababa"}, {"A=abcabcX", "B=abcabc"}, {"A=aba", "B=bab"}, + {"TOKEN=abc\x00\xffabc"}, } { redactor := secrets.NewRedactor(entries) var values []string @@ -47,6 +49,27 @@ func (s *StreamingBoundarySuite) TestKnownSecretsAcrossEverySplit() { } } +func BenchmarkStreamingBoundaryRetainedValues(b *testing.B) { + for _, count := range []int{1, 100, 1000} { + b.Run(fmt.Sprint(count), func(b *testing.B) { + entries := make([]string, count) + for index := range count { + entries[index] = fmt.Sprintf("TOKEN=private-%d-value", index) + } + redactor := secrets.NewRedactor(entries) + input := strings.Repeat("ordinary output ", 2048) + "private-" + b.SetBytes(int64(len(input))) + b.ReportAllocs() + b.ResetTimer() + for range b.N { + stream := secrets.NewStreamingRedactor(redactor) + _ = stream.RedactChunk(input) + _ = stream.Flush() + } + }) + } +} + func (s *StreamingBoundarySuite) TestRepeatedOutputDoesNotBufferWholeStream() { const count = 1 << 14 input := strings.Repeat("abcabc", count) diff --git a/pkg/secrets/redact_matcher.go b/pkg/secrets/redact_matcher.go new file mode 100644 index 0000000000..f1617fa8b6 --- /dev/null +++ b/pkg/secrets/redact_matcher.go @@ -0,0 +1,28 @@ +package secrets + +// An immutable byte trie matches the longest known secret without scanning +// every retained value at each output position. Byte keys preserve binary data. +type secretNode struct { + children map[byte]*secretNode + length int +} + +func newSecretMatcher(values []string) *secretNode { + root := &secretNode{} + for _, value := range values { + node := root + for index := range len(value) { + if node.children == nil { + node.children = make(map[byte]*secretNode) + } + child := node.children[value[index]] + if child == nil { + child = &secretNode{} + node.children[value[index]] = child + } + node = child + } + node.length = len(value) + } + return root +} From aa502e4c8caa993c949eb5ad223c4d7f751106cc Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 5 Oct 2026 23:43:22 -0600 Subject: [PATCH 4/8] fix(driver): normalize runtime errors after cleanup --- pkg/driver/external/exec.go | 5 ----- pkg/driver/external/session.go | 14 +++++++++----- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/pkg/driver/external/exec.go b/pkg/driver/external/exec.go index 57f6ab78e1..7778fa1e5e 100644 --- a/pkg/driver/external/exec.go +++ b/pkg/driver/external/exec.go @@ -89,11 +89,6 @@ func (h *Host) commandResult( err, flushErr error, ) error { if err != nil { - // Cleanup can finish after the caller deadline even when the transport - // reported its cancellation just before the local context timer fired. - if ctx.Err() != nil { - return h.operationError(ctx, "Exec", err) - } return err } if flushErr != nil { diff --git a/pkg/driver/external/session.go b/pkg/driver/external/session.go index 5171cb2d82..ee3f7e54d2 100644 --- a/pkg/driver/external/session.go +++ b/pkg/driver/external/session.go @@ -2,6 +2,7 @@ package external import ( "context" + "errors" "fmt" "io" "os/exec" @@ -23,7 +24,7 @@ func (h *Host) call( ctx context.Context, operation string, rpc func(runtimev1.RuntimeDriverClient) error, -) error { +) (err error) { if err := ctx.Err(); err != nil { return err } @@ -31,6 +32,9 @@ func (h *Host) call( if err != nil { return fmt.Errorf("external runtime %s: %w", operation, err) } + // Normalize after owned cleanup so caller deadlines win over transport + // cancellation reported just before the local context timer fires. + defer func() { err = h.operationError(ctx, operation, err) }() output := log.Writer(log.LevelDebug) diagnostic := &diagnosticWriter{ writer: output, @@ -63,18 +67,18 @@ func (h *Host) call( started := time.Now() transport, err := client.Client() if err != nil { - return h.operationError(ctx, operation, err) + return err } instance, err := transport.Dispense(sdkplugin.Name) if err != nil { - return h.operationError(ctx, operation, err) + return err } runtimeClient, ok := instance.(runtimev1.RuntimeDriverClient) if !ok { - return fmt.Errorf("external runtime %s: invalid gRPC client", operation) + return errors.New("invalid gRPC client") } log.Debugf("External runtime %s startup completed in %s", operation, time.Since(started)) - return h.operationError(ctx, operation, rpc(runtimeClient)) + return rpc(runtimeClient) } // go-plugin holds its client lock throughout handshake, so Client.Kill cannot From 096a1d782e5a245003c3476ce6949f7be974dbb7 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 6 Oct 2026 00:43:47 -0600 Subject: [PATCH 5/8] fix(driver): retain exec input failures through cleanup --- pkg/driver/external/exec.go | 9 ++++-- pkg/driver/external/exec_exchange_test.go | 37 +++++++++++++++++++++-- 2 files changed, 40 insertions(+), 6 deletions(-) diff --git a/pkg/driver/external/exec.go b/pkg/driver/external/exec.go index 7778fa1e5e..41920a8d07 100644 --- a/pkg/driver/external/exec.go +++ b/pkg/driver/external/exec.go @@ -13,6 +13,8 @@ import ( ) var ( + errExecComplete = errors.New("external runtime Exec stream completed") + _ driver.Driver = (*Host)(nil) _ driver.ArgvExecDriver = (*Host)(nil) ) @@ -165,12 +167,13 @@ func exchangeExec( } }() exit, receiveErr := receiveExecOutput(stream, streams) - inputCause := context.Cause(ctx) + // Distinguish ordinary cleanup from an input failure that wins cancellation. // Cancel before joining: Send may be flow-controlled and Read may be blocked. - cancel(nil) + cancel(errExecComplete) closeInput() <-inputDone - if inputCause != nil { + inputCause := context.Cause(ctx) + if inputCause != nil && inputCause != errExecComplete { return nil, inputCause } // Recv owns command completion, including a command that exits before stdin EOF. diff --git a/pkg/driver/external/exec_exchange_test.go b/pkg/driver/external/exec_exchange_test.go index bb3ba22254..d747e6e2dd 100644 --- a/pkg/driver/external/exec_exchange_test.go +++ b/pkg/driver/external/exec_exchange_test.go @@ -14,7 +14,7 @@ func (s *HostSuite) TestExecInputFailureSurvivesBufferedSuccessfulExit() { ctx, cancel := context.WithCancelCause(context.Background()) defer cancel(nil) want := errors.New("stdin read failed before command completion") - stream := &completedExecStream{ctx: ctx} + stream := &completedExecStream{ready: ctx.Done()} exit, err := exchangeExec(ctx, cancel, stream, driver.Streams{ Stdin: iotest.ErrReader(want), Stdout: io.Discard, Stderr: io.Discard, }) @@ -22,10 +22,41 @@ func (s *HostSuite) TestExecInputFailureSurvivesBufferedSuccessfulExit() { s.Nil(exit) } +func (s *HostSuite) TestExecInputFailureAtCleanupBoundarySurvivesSuccessfulExit() { + 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{}) + close(ready) + inputFailed := make(chan struct{}) + cancelAtBoundary := func(cause error) { + if errors.Is(cause, want) { + cancel(cause) + close(inputFailed) + return + } + _ = producer.CloseWithError(want) + <-inputFailed + cancel(cause) + } + exit, err := exchangeExec( + ctx, + cancelAtBoundary, + &completedExecStream{ready: ready}, + driver.Streams{ + Stdin: input, Stdout: io.Discard, Stderr: io.Discard, + }, + ) + s.ErrorIs(err, want) + s.Nil(exit) +} + // Models a terminal exit already buffered when an input failure cancels the RPC. type completedExecStream struct { runtimev1.RuntimeDriver_ExecClient - ctx context.Context + ready <-chan struct{} exitSent bool } @@ -33,7 +64,7 @@ func (s *completedExecStream) Recv() (*runtimev1.ExecServerMessage, error) { if s.exitSent { return nil, io.EOF } - <-s.ctx.Done() + <-s.ready s.exitSent = true return &runtimev1.ExecServerMessage{ Payload: &runtimev1.ExecServerMessage_Exit{Exit: &runtimev1.ExecExit{}}, From a1ac956a44999982dd0b4282c3e3645f250a70f3 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 6 Oct 2026 01:54:16 -0600 Subject: [PATCH 6/8] fix(driver): preserve late stdin pump errors --- pkg/driver/external/exec.go | 16 +++- pkg/driver/external/exec_exchange_test.go | 107 ++++++++++++++++------ 2 files changed, 94 insertions(+), 29 deletions(-) diff --git a/pkg/driver/external/exec.go b/pkg/driver/external/exec.go index 41920a8d07..f101821b9b 100644 --- a/pkg/driver/external/exec.go +++ b/pkg/driver/external/exec.go @@ -10,6 +10,8 @@ import ( "github.com/devsy-org/devsy-runtime-sdk/runtimev1" "github.com/devsy-org/devsy/pkg/driver" "github.com/devsy-org/devsy/pkg/subprocess" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) var ( @@ -158,24 +160,32 @@ func exchangeExec( }) stop := context.AfterFunc(ctx, closeInput) defer stop() - inputDone := make(chan struct{}) + inputDone := make(chan error, 1) go func() { - defer close(inputDone) err := sendExecInput(ctx, stream, streams.Stdin) if err != nil && !errors.Is(err, io.EOF) { cancel(err) } + inputDone <- 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 + inputErr := <-inputDone inputCause := context.Cause(ctx) if inputCause != nil && inputCause != errExecComplete { return nil, inputCause } + if inputErr != nil && !isExecInputCleanupError(inputErr) { + return nil, inputErr + } // Recv owns command completion, including a command that exits before stdin EOF. return exit, receiveErr } + +func isExecInputCleanupError(err error) bool { + return errors.Is(err, io.EOF) || errors.Is(err, io.ErrClosedPipe) || + errors.Is(err, context.Canceled) || status.Code(err) == codes.Canceled +} diff --git a/pkg/driver/external/exec_exchange_test.go b/pkg/driver/external/exec_exchange_test.go index d747e6e2dd..c7c1dfb13e 100644 --- a/pkg/driver/external/exec_exchange_test.go +++ b/pkg/driver/external/exec_exchange_test.go @@ -8,6 +8,8 @@ import ( "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() { @@ -23,34 +25,76 @@ func (s *HostSuite) TestExecInputFailureSurvivesBufferedSuccessfulExit() { } func (s *HostSuite) TestExecInputFailureAtCleanupBoundarySurvivesSuccessfulExit() { - 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{}) - close(ready) - inputFailed := make(chan struct{}) - cancelAtBoundary := func(cause error) { - if errors.Is(cause, want) { - cancel(cause) - close(inputFailed) - return + for _, completionFirst := range []bool{false, true} { + name := "input failure wins cancellation" + if completionFirst { + name = "completion wins cancellation" } - _ = producer.CloseWithError(want) - <-inputFailed - cancel(cause) + 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"), + } { + s.Run(cleanupErr.Error(), func() { + ctx, cancel := context.WithCancelCause(context.Background()) + defer cancel(nil) + input, producer := io.Pipe() + defer func() { _ = producer.Close() }() + ready := make(chan struct{}) + cancelAtBoundary := func(cause error) { + cancel(cause) + _ = producer.CloseWithError(cleanupErr) + } + exit, err := exchangeExec( + ctx, + cancelAtBoundary, + &completedExecStream{ready: ready}, + driver.Streams{ + Stdin: &readyInputReader{Reader: input, ready: ready}, + Stdout: io.Discard, + Stderr: io.Discard, + }, + ) + s.NoError(err) + s.NotNil(exit) + }) } - exit, err := exchangeExec( - ctx, - cancelAtBoundary, - &completedExecStream{ready: ready}, - driver.Streams{ - Stdin: input, Stdout: io.Discard, Stderr: io.Discard, - }, - ) - s.ErrorIs(err, want) - s.Nil(exit) } // Models a terminal exit already buffered when an input failure cancels the RPC. @@ -70,3 +114,14 @@ func (s *completedExecStream) Recv() (*runtimev1.ExecServerMessage, error) { 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) +} From 9252223b7b43425530a934e172d490f22334be6d Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 6 Oct 2026 03:14:51 -0600 Subject: [PATCH 7/8] fix(driver): track owned stdin closure explicitly --- pkg/driver/external/exec.go | 26 ++++--------- pkg/driver/external/exec_exchange_test.go | 38 ++++++++++++------- pkg/driver/external/exec_input.go | 46 +++++++++++++++++++++++ pkg/driver/external/exec_test.go | 13 +++++++ 4 files changed, 91 insertions(+), 32 deletions(-) create mode 100644 pkg/driver/external/exec_input.go diff --git a/pkg/driver/external/exec.go b/pkg/driver/external/exec.go index f101821b9b..782c717b9d 100644 --- a/pkg/driver/external/exec.go +++ b/pkg/driver/external/exec.go @@ -10,8 +10,6 @@ import ( "github.com/devsy-org/devsy-runtime-sdk/runtimev1" "github.com/devsy-org/devsy/pkg/driver" "github.com/devsy-org/devsy/pkg/subprocess" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" ) var ( @@ -153,39 +151,31 @@ func exchangeExec( stream runtimev1.RuntimeDriver_ExecClient, streams driver.Streams, ) (*runtimev1.ExecExit, error) { - closeInput := sync.OnceFunc(func() { - if closer, ok := streams.Stdin.(io.Closer); ok { - _ = closer.Close() - } - }) + input := &execInput{reader: streams.Stdin} + closeInput := sync.OnceFunc(input.close) stop := context.AfterFunc(ctx, closeInput) defer stop() - inputDone := make(chan error, 1) + inputDone := make(chan struct{}) go func() { - err := sendExecInput(ctx, stream, streams.Stdin) + defer close(inputDone) + err := sendExecInput(ctx, stream, input) if err != nil && !errors.Is(err, io.EOF) { cancel(err) } - inputDone <- 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() - inputErr := <-inputDone + <-inputDone inputCause := context.Cause(ctx) if inputCause != nil && inputCause != errExecComplete { return nil, inputCause } - if inputErr != nil && !isExecInputCleanupError(inputErr) { - return nil, inputErr + if err := input.readError(); err != nil { + return nil, err } // Recv owns command completion, including a command that exits before stdin EOF. return exit, receiveErr } - -func isExecInputCleanupError(err error) bool { - return errors.Is(err, io.EOF) || errors.Is(err, io.ErrClosedPipe) || - errors.Is(err, context.Canceled) || status.Code(err) == codes.Canceled -} diff --git a/pkg/driver/external/exec_exchange_test.go b/pkg/driver/external/exec_exchange_test.go index c7c1dfb13e..366427ee58 100644 --- a/pkg/driver/external/exec_exchange_test.go +++ b/pkg/driver/external/exec_exchange_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "io" + "os" "testing/iotest" "github.com/devsy-org/devsy-runtime-sdk/runtimev1" @@ -70,26 +71,17 @@ func (s *HostSuite) TestExecInputFailureAtCleanupBoundarySurvivesSuccessfulExit( 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, producer := io.Pipe() - defer func() { _ = producer.Close() }() - ready := make(chan struct{}) - cancelAtBoundary := func(cause error) { - cancel(cause) - _ = producer.CloseWithError(cleanupErr) + input := &closedInputReader{ + ready: make(chan struct{}), closed: make(chan struct{}), err: cleanupErr, } exit, err := exchangeExec( - ctx, - cancelAtBoundary, - &completedExecStream{ready: ready}, - driver.Streams{ - Stdin: &readyInputReader{Reader: input, ready: ready}, - Stdout: io.Discard, - Stderr: io.Discard, - }, + ctx, cancel, &completedExecStream{ready: input.ready}, + driver.Streams{Stdin: input, Stdout: io.Discard, Stderr: io.Discard}, ) s.NoError(err) s.NotNil(exit) @@ -125,3 +117,21 @@ 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 +} diff --git a/pkg/driver/external/exec_input.go b/pkg/driver/external/exec_input.go new file mode 100644 index 0000000000..4ee6797670 --- /dev/null +++ b/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 +} diff --git a/pkg/driver/external/exec_test.go b/pkg/driver/external/exec_test.go index 49ae6bf118..39dde18f2e 100644 --- a/pkg/driver/external/exec_test.go +++ b/pkg/driver/external/exec_test.go @@ -159,6 +159,19 @@ func (s *HostSuite) TestExecEarlyExitClosesBlockedInput() { s.ErrorIs(err, io.ErrClosedPipe) } +func (s *HostSuite) TestExecEarlyExitClosesFileInput() { + host := s.runningHost(fake.Conformance) + input, producer, err := os.Pipe() + s.Require().NoError(err) + defer func() { _ = input.Close(); _ = producer.Close() }() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + s.NoError(host.CommandContainerArgv(ctx, fixtureWorkspace, + []string{earlyExitCommand}, driver.Streams{Stdin: input})) + _, err = input.Stat() + s.ErrorIs(err, os.ErrClosed) +} + func (s *HostSuite) TestExecDeadlineAndCrash() { for _, mode := range []string{fake.ExecSlow, fake.ExecCrash} { s.Run(mode, func() { From c0d692de29148e193182e1969ce6d33a8ef79b58 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 6 Oct 2026 04:34:50 -0600 Subject: [PATCH 8/8] ci: allow macOS race tests to finish within job budget --- .github/workflows/pr-ci.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/pr-ci.yml b/.github/workflows/pr-ci.yml index b9b2e9a7b9..0d16c703b6 100644 --- a/.github/workflows/pr-ci.yml +++ b/.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