From 7deffa52f6ae12d3c4c8a5cd419386fed4f77998 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 11 Oct 2026 11:00:47 -0600 Subject: [PATCH 1/2] fix(runtime): unblock canceled MicroSandbox input streams --- .github/workflows/pr-ci.yml | 9 + e2e/e2e_suite_test.go | 10 + .../up/provider_microsandbox_logs_cancel.go | 590 ++++++++++++++++++ .../exec_stream_cancellation_unix_test.go | 261 ++++++++ pkg/agent/delivery/kubernetes.go | 29 +- pkg/driver/microsandbox/cliclient.go | 24 +- .../exec_cancellation_unix_test.go | 248 ++++++++ .../developing-providers/runtime-protocol.mdx | 10 +- 8 files changed, 1177 insertions(+), 4 deletions(-) create mode 100644 e2e/tests/up/provider_microsandbox_logs_cancel.go create mode 100644 pkg/agent/delivery/exec_stream_cancellation_unix_test.go create mode 100644 pkg/driver/microsandbox/exec_cancellation_unix_test.go diff --git a/.github/workflows/pr-ci.yml b/.github/workflows/pr-ci.yml index 523ef8b796..4c9a249660 100644 --- a/.github/workflows/pr-ci.yml +++ b/.github/workflows/pr-ci.yml @@ -714,6 +714,15 @@ jobs: test-timeout: 1200s job-timeout-minutes: 25 + - label: up-provider-microsandbox-logs-cancel + runner: ubuntu-latest + free-disk-space: false + install-kind: false + requires-secret: false + install-microsandbox: true + test-timeout: 1200s + job-timeout-minutes: 25 + # Snapshot tests - label: snapshot diff --git a/e2e/e2e_suite_test.go b/e2e/e2e_suite_test.go index 87d3355917..abe2f0af1c 100644 --- a/e2e/e2e_suite_test.go +++ b/e2e/e2e_suite_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "github.com/devsy-org/devsy/cmd" "github.com/devsy-org/devsy/e2e/framework" // Register tests. _ "github.com/devsy-org/devsy/e2e/tests/build" @@ -43,6 +44,15 @@ import ( "github.com/onsi/gomega" ) +func TestMain(m *testing.M) { + // The external runtime self-execs this test binary as the real supervisor. + if len(os.Args) > 2 && os.Args[1] == "internal" && os.Args[2] == "runtime-supervisor" { + cmd.Execute() + return + } + os.Exit(m.Run()) +} + // TestRunE2ETests checks configuration parameters (specified through flags) and then runs // E2E tests using the Ginkgo runner. // If a "report directory" is specified, one or more JUnit test reports will be diff --git a/e2e/tests/up/provider_microsandbox_logs_cancel.go b/e2e/tests/up/provider_microsandbox_logs_cancel.go new file mode 100644 index 0000000000..77900ed26b --- /dev/null +++ b/e2e/tests/up/provider_microsandbox_logs_cancel.go @@ -0,0 +1,590 @@ +package up + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "time" + + "al.essio.dev/pkg/shellescape" + "github.com/devsy-org/devsy/e2e/framework" + "github.com/devsy-org/devsy/pkg/agent" + pkgconfig "github.com/devsy-org/devsy/pkg/config" + "github.com/devsy-org/devsy/pkg/driver" + "github.com/devsy-org/devsy/pkg/driver/drivercreate" + "github.com/devsy-org/devsy/pkg/encoding" + "github.com/devsy-org/devsy/pkg/provider" + "github.com/onsi/ginkgo/v2" + "github.com/onsi/gomega" +) + +const ( + microsandboxCancelBound = 10 * time.Second + microsandboxCancelSentinel = "source-sentinel" + microsandboxCancelShell = "/bin/sh" +) + +type microsandboxLogCancelState struct { + Status string `json:"status"` +} + +// The child blocks on its own FIFO; the parent either waits for it or reads the +// caller's stdin. Neither process has a self-exit timer. The nonce is in both +// argv vectors, and readiness includes each process's kernel start time. +const microsandboxCancelScript = `set -eu +nonce=$1 +mode=$2 +start_time() { sed 's/^.*) //' "/proc/$1/stat" | awk '{print $20}'; } +if [ "$mode" = child ]; then + fifo="/root/$nonce.fifo" + mkfifo "$fifo" + exec 9<>"$fifo" + printf 'READY %s child %s %s\n' "$nonce" "$$" "$(start_time $$)" + read -r ignored <&9 + exit 19 +fi +/bin/sh -c "$3" "$nonce-child" "$nonce" child & +child=$! +printf 'READY %s parent %s %s\n' "$nonce" "$$" "$(start_time $$)" +if [ "$mode" = stdin ]; then + read -r ignored + exit 20 +fi +wait "$child" +` + +type microsandboxLogCancelFixture struct { + framework *framework.Framework + workspace string + sandbox string + createdAt string + bootID string + nonce string + uid string + label string + runtime driver.ArgvExecDriver +} + +var _ = ginkgo.Describe("microsandbox logs and cancellation parity", + ginkgo.Label("up-provider-microsandbox-logs-cancel"), func() { + for _, selection := range []struct{ name, source, runtime string }{ + {"builtin-logs-cancel", microsandboxResourceBuiltin, provider.MicrosandboxDriver}, + {"external-logs-cancel", microsandboxExternalProvider, provider.ExternalDriver}, + } { + ginkgo.Context(selection.name, func() { + var fixture *microsandboxLogCancelFixture + ginkgo.BeforeEach(func(ctx context.Context) { + fixture = newMicrosandboxLogCancelFixture(ctx, selection.name, selection.source) + fixture.loadRuntime(ctx, selection.runtime) + }) + ginkgo.It( + "returns actual stdout and stderr workload logs while running and stopped", + func(ctx context.Context) { fixture.assertLogs(ctx) }, + ginkgo.SpecTimeout(framework.TimeoutLong()), + ) + ginkgo.It("cancels a ready Exec parent and child without replacing the VM", + func(ctx context.Context) { fixture.assertCancellation(ctx, false) }, + ginkgo.SpecTimeout(framework.TimeoutLong()), + ) + ginkgo.It("cancels a ready Exec with blocked owned stdin without replacing the VM", + func(ctx context.Context) { fixture.assertCancellation(ctx, true) }, + ginkgo.SpecTimeout(framework.TimeoutLong()), + ) + }) + } + }) + +func newMicrosandboxLogCancelFixture( + ctx context.Context, name, source string, +) *microsandboxLogCancelFixture { + skipIfNoMicrosandbox(ctx) + ginkgo.GinkgoT().Setenv("DEVSY_HOME", ginkgo.GinkgoT().TempDir()) + ginkgo.GinkgoT().Setenv("DEVSY_CONFIG", "") + initialDir, err := os.Getwd() + framework.ExpectNoError(err) + f := framework.NewDefaultFramework(filepath.Join(initialDir, "bin")) + workspace, err := framework.CreateTempDir() + framework.ExpectNoError(err) + ginkgo.DeferCleanup(framework.CleanupTempDir, initialDir, workspace) + var random [16]byte + _, err = rand.Read(random[:]) + framework.ExpectNoError(err) + fixture := µsandboxLogCancelFixture{ + framework: f, workspace: workspace, + nonce: "exec-parity-" + hex.EncodeToString(random[:]), + uid: encoding.CreateNewUID("", filepath.Base(workspace)), + } + fixture.sandbox = "devsy-" + fixture.uid + fixture.label = "devsy.logs-cancel-parity=" + fixture.uid + assertMicrosandboxEphemeralAbsent(ctx, fixture.label) + out, err := microsandboxResourceCommand(ctx, "list", "--quiet") + framework.ExpectNoError(err) + gomega.Expect(strings.Fields(string(out))).NotTo(gomega.ContainElement(fixture.sandbox)) + framework.ExpectNoError(f.DevsyProviderAdd(ctx, source, "--name", name)) + fixture.registerCleanup(name) + framework.ExpectNoError(f.DevsyProviderUse(ctx, name)) + writeMicrosandboxResourceConfig(workspace, nil) + framework.ExpectNoError(os.WriteFile( + filepath.Join(workspace, microsandboxCancelSentinel), []byte(fixture.nonce), 0o600, + )) + fixture.up(ctx) + gomega.Expect(microsandboxResourceSandbox(ctx, f, workspace)).To(gomega.Equal(fixture.sandbox)) + fixture.createdAt = microsandboxCreationTime(ctx, fixture.sandbox) + fixture.bootID = microsandboxStorageBootID(ctx, f, workspace) + return fixture +} + +func (f *microsandboxLogCancelFixture) registerCleanup(name string) { + ginkgo.DeferCleanup(func() error { + cleanup, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + return f.framework.DevsyProviderDelete(cleanup, name) + }) + // Registered only after proving the fresh UID/name/label absent. A failed + // startup can remove its Devsy record before deleting the owned backend VM. + ginkgo.DeferCleanup(func() error { + cleanup, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + out, err := microsandboxResourceCommand(cleanup, "list", "--quiet") + if err != nil { + return err + } + for sandbox := range strings.FieldsSeq(string(out)) { + if sandbox != f.sandbox { + continue + } + out, err = microsandboxResourceCommand(cleanup, "remove", "--force", sandbox) + if err != nil { + return fmt.Errorf("remove exact test-owned sandbox %s: %w: %s", sandbox, err, out) + } + } + return nil + }) + ginkgo.DeferCleanup(f.framework.CleanupWorkspace, f.workspace) +} + +func (f *microsandboxLogCancelFixture) up(ctx context.Context) { + framework.ExpectNoError(f.framework.DevsyUp(ctx, f.workspace, + "--uid", f.uid, "--id-label", f.label, "--devcontainer", ".devcontainer.json")) +} + +func (f *microsandboxLogCancelFixture) loadRuntime(ctx context.Context, expected string) { + workspace, err := f.framework.FindWorkspace(ctx, f.workspace) + framework.ExpectNoError(err) + origin, err := provider.GetWorkspaceAgentDir(pkgconfig.DefaultContext, workspace.ID) + framework.ExpectNoError(err) + info, err := agent.ParseAgentWorkspaceInfo(filepath.Join(origin, provider.WorkspaceConfigFile)) + framework.ExpectNoError(err) + gomega.Expect(info.Workspace).NotTo(gomega.BeNil()) + gomega.Expect(string(info.Agent.Local)).To(gomega.Equal("true")) + gomega.Expect(info.Agent.Driver).To(gomega.Equal(expected)) + gomega.Expect(info.Origin).To(gomega.Equal(origin)) + gomega.Expect(info.Workspace.UID).To(gomega.Equal(workspace.UID)) + gomega.Expect(info.Workspace.UID).To(gomega.Equal(f.uid)) + gomega.Expect(info.Workspace.UID).NotTo(gomega.BeEmpty()) + if expected == provider.ExternalDriver { + binaries, resolveErr := agent.GetAgentBinariesDirFromWorkspaceDir(origin) + framework.ExpectNoError(resolveErr) + _, resolveErr = provider.ResolveExternalRuntimeBinary(info.Agent, binaries) + framework.ExpectNoError(resolveErr) + } + bundle, err := drivercreate.New(ctx, info) + framework.ExpectNoError(err) + var ok bool + f.runtime, ok = bundle.Runtime.(driver.ArgvExecDriver) + gomega.Expect(ok).To(gomega.BeTrue(), "selected runtime must implement literal argv Exec") + f.uid = info.Workspace.UID +} + +func (f *microsandboxLogCancelFixture) assertLogs(ctx context.Context) { + outMarker, errMarker := f.nonce+"-stdout", f.nonce+"-stderr" + command, cancel := context.WithTimeout(ctx, 30*time.Second) + err := f.runtime.CommandContainerArgv(command, f.uid, + []string{ + microsandboxCancelShell, "-c", "printf '%s\\n' \"$1\"; printf '%s\\n' \"$2\" >&2", + f.nonce, outMarker, errMarker, + }, + driver.Streams{Stdout: io.Discard, Stderr: io.Discard}) + cancel() + framework.ExpectNoError(err) + f.assertLogMarkers(ctx, outMarker, errMarker) + f.assertUnchanged(ctx) + framework.ExpectNoError(f.framework.DevsyStop(ctx, f.workspace)) + assertMicrosandboxEphemeralStopped(ctx, f.sandbox, f.createdAt) + f.assertLogMarkers(ctx, outMarker, errMarker) + assertMicrosandboxEphemeralStopped(ctx, f.sandbox, f.createdAt) + f.up(ctx) + gomega.Expect(microsandboxCreationTime(ctx, f.sandbox)).To(gomega.Equal(f.createdAt)) + gomega.Expect(microsandboxStorageBootID(ctx, f.framework, f.workspace)). + NotTo(gomega.Equal(f.bootID)) + f.assertSourceAndCommand(ctx) +} + +func (f *microsandboxLogCancelFixture) assertLogMarkers(ctx context.Context, markers ...string) { + logsCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + out, err := f.framework.DevsyLogs(logsCtx, f.workspace) + framework.ExpectNoError(err) + lines := strings.Split(strings.ReplaceAll(out, "\r\n", "\n"), "\n") + for _, marker := range markers { + gomega.Expect(lines).To(gomega.ContainElement(marker)) + } +} + +type microsandboxExecIdentity struct { + Role string + PID int + Start string +} + +type microsandboxExecReadiness struct { + mu sync.Mutex + pending string + nonce string + frames chan microsandboxExecIdentity + seen []microsandboxExecIdentity +} + +func (r *microsandboxExecReadiness) Write(p []byte) (int, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.pending += string(p) + if len(r.pending) > 4096 { + return 0, fmt.Errorf("unexpectedly large Exec readiness output") + } + for { + line, rest, found := strings.Cut(r.pending, "\n") + if !found { + return len(p), nil + } + r.pending = rest + identity, err := parseMicrosandboxExecIdentity(line, r.nonce) + if err != nil { + return 0, err + } + r.seen = append(r.seen, identity) + select { + case r.frames <- identity: + default: + return 0, fmt.Errorf("too many Exec readiness frames") + } + } +} + +func parseMicrosandboxExecIdentity(line, nonce string) (microsandboxExecIdentity, error) { + fields := strings.Fields(line) + if len(fields) != 5 || fields[0] != "READY" || fields[1] != nonce { + return microsandboxExecIdentity{}, fmt.Errorf("unexpected Exec readiness frame %q", line) + } + pid, err := strconv.Atoi(fields[3]) + if err != nil || pid <= 1 { + return microsandboxExecIdentity{}, fmt.Errorf("invalid Exec PID in %q", line) + } + if _, err = strconv.ParseUint(fields[4], 10, 64); err != nil { + return microsandboxExecIdentity{}, fmt.Errorf("invalid Exec start time in %q", line) + } + return microsandboxExecIdentity{Role: fields[2], PID: pid, Start: fields[4]}, nil +} + +func (r *microsandboxExecReadiness) snapshot() []microsandboxExecIdentity { + r.mu.Lock() + defer r.mu.Unlock() + return append([]microsandboxExecIdentity(nil), r.seen...) +} + +type microsandboxOwnedInput struct { + *io.PipeReader + entered chan struct{} + closed chan struct{} + readOnce sync.Once + closeOnce sync.Once +} + +func (r *microsandboxOwnedInput) Read(p []byte) (int, error) { + r.readOnce.Do(func() { close(r.entered) }) + return r.PipeReader.Read(p) +} + +func (r *microsandboxOwnedInput) Close() error { + err := r.PipeReader.Close() + r.closeOnce.Do(func() { close(r.closed) }) + return err +} + +type microsandboxExecOperation struct { + cancel context.CancelFunc + done chan struct{} + err error + input *microsandboxOwnedInput + writer *io.PipeWriter +} + +func (f *microsandboxLogCancelFixture) startExec( + ctx context.Context, blocked bool, readiness *microsandboxExecReadiness, +) *microsandboxExecOperation { + commandCtx, cancel := context.WithCancel(ctx) + op := µsandboxExecOperation{cancel: cancel, done: make(chan struct{})} + mode := "wait" + var input io.Reader + if blocked { + reader, writer := io.Pipe() + op.input = µsandboxOwnedInput{ + PipeReader: reader, entered: make(chan struct{}), closed: make(chan struct{}), + } + op.writer, input, mode = writer, op.input, "stdin" + } + ginkgo.DeferCleanup(op.cleanup) + go func() { + defer close(op.done) + op.err = f.runtime.CommandContainerArgv(commandCtx, f.uid, + []string{ + microsandboxCancelShell, "-c", microsandboxCancelScript, f.nonce, + f.nonce, mode, microsandboxCancelScript, + }, + driver.Streams{Stdin: input, Stdout: readiness, Stderr: io.Discard}) + }() + return op +} + +func (op *microsandboxExecOperation) cleanup() { + op.cancel() + if op.input != nil { + _ = op.input.Close() + _ = op.writer.Close() + } + select { + case <-op.done: + case <-time.After(microsandboxCancelBound): + ginkgo.Fail("owned Exec goroutine failed to join after cancellation and input cleanup") + } +} + +func (f *microsandboxLogCancelFixture) assertCancellation(ctx context.Context, blocked bool) { + ready := µsandboxExecReadiness{ + nonce: f.nonce, + frames: make(chan microsandboxExecIdentity, 2), + } + op := f.startExec(ctx, blocked, ready) + ginkgo.DeferCleanup(func() { f.cleanupProcesses(ready.snapshot()) }) + identities := f.awaitReadiness(ctx, ready, op) + observer, cancelObserver := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) + alive, err := f.processesAlive(observer, identities) + cancelObserver() + framework.ExpectNoError(err) + gomega.Expect(alive). + To(gomega.Equal(len(identities)), "both exact identities must be live before cancellation") + + // Both bounds start at the cancellation event, not after waiting for caller exit. + deathCtx, cancelDeath := context.WithTimeout( + context.WithoutCancel(ctx), + microsandboxCancelBound, + ) + defer cancelDeath() + op.cancel() + select { + case <-op.done: + gomega.Expect(op.err). + To(gomega.HaveOccurred(), "cancellation must not look like successful completion") + case <-deathCtx.Done(): + ginkgo.Fail("ready Runtime Exec did not return within 10s of cancellation") + } + if blocked { + select { + case <-op.input.closed: + default: + ginkgo.Fail("Runtime Exec returned without closing its blocked owned input") + } + _, err = op.writer.Write([]byte("late producer")) + gomega.Expect(err).To(gomega.MatchError(io.ErrClosedPipe)) + } + f.awaitProcessDeath(deathCtx, identities) + f.assertUnchanged(ctx) +} + +func (f *microsandboxLogCancelFixture) awaitReadiness( + ctx context.Context, ready *microsandboxExecReadiness, op *microsandboxExecOperation, +) []microsandboxExecIdentity { + readinessCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + identities := make([]microsandboxExecIdentity, 0, 2) + roles := make(map[string]bool) + for len(identities) < 2 { + select { + case identity := <-ready.frames: + gomega.Expect(identity.Role).To(gomega.BeElementOf("parent", "child")) + gomega.Expect(roles[identity.Role]).To(gomega.BeFalse(), "duplicate readiness role") + roles[identity.Role] = true + identities = append(identities, identity) + case <-op.done: + ginkgo.Fail(fmt.Sprintf("Exec exited before readiness: %v", op.err)) + case <-readinessCtx.Done(): + ginkgo.Fail("guest parent and child were not ready within 30s") + } + } + gomega.Expect(identities[0].PID).NotTo(gomega.Equal(identities[1].PID)) + op.awaitInputRead(readinessCtx) + return identities +} + +func (op *microsandboxExecOperation) awaitInputRead(ctx context.Context) { + if op.input == nil { + return + } + select { + case <-op.input.entered: + case <-op.done: + ginkgo.Fail(fmt.Sprintf("Exec exited before stdin Read entered: %v", op.err)) + case <-ctx.Done(): + ginkgo.Fail("host input Read was not entered within the readiness bound") + } +} + +// The observer is a separate SSH operation, never the canceled Exec session. +// Identity replacement or a zombie means the original process is not live; +// unexpected procfs or SSH errors fail, rather than being interpreted as death. +func (f *microsandboxLogCancelFixture) processesAlive( + ctx context.Context, identities []microsandboxExecIdentity, +) (int, error) { + out, err := f.framework.DevsySSHOnce(ctx, f.workspace, f.identityCommand(identities, false)) + if err != nil { + return 0, err + } + count, err := strconv.Atoi(strings.TrimSpace(out)) + if err != nil { + return 0, fmt.Errorf("invalid process observer response %q: %w", out, err) + } + if count < 0 || count > len(identities) { + return 0, fmt.Errorf("unexpected live process count %d", count) + } + return count, nil +} + +func (f *microsandboxLogCancelFixture) identityCommand( + identities []microsandboxExecIdentity, cleanup bool, +) string { + var command strings.Builder + command.WriteString("set -eu; alive=0; ") + for _, identity := range identities { + fmt.Fprintf( + &command, + "p=%d; expected=%s; ", + identity.PID, + shellescape.Quote(identity.Start), + ) + command.WriteString(`if [ -d "/proc/$p" ]; then ` + + `if stat=$(cat "/proc/$p/stat"); then ` + + `stat=${stat##*) }; set -- $stat; state=$1; shift 19; actual=$1; ` + + `if [ "$actual" = "$expected" ] && [ "$state" != Z ] && [ "$state" != X ]; then ` + + `if cmdline=$(tr '\000' ' ' < "/proc/$p/cmdline"); then `) + fmt.Fprintf(&command, "case \"$cmdline\" in *%s*) ", shellescape.Quote(f.nonce)) + if cleanup { + command.WriteString(`kill -KILL "$p" || { [ ! -d "/proc/$p" ] || exit 31; }; `) + } else { + command.WriteString("alive=$((alive + 1)); ") + } + if cleanup { + command.WriteString(";; *) : ;; esac; ") + } else { + command.WriteString(`;; *) +if latest=$(cat "/proc/$p/stat"); then + latest=${latest##*) }; set -- $latest; latest_state=$1; shift 19 + if [ "$1" = "$expected" ] && [ "$latest_state" != Z ] && [ "$latest_state" != X ]; then exit 34; fi +else [ ! -d "/proc/$p" ] || exit 35; fi +;; esac; `) + } + command.WriteString(`else [ ! -d "/proc/$p" ] || exit 32; fi; fi; ` + + `else [ ! -d "/proc/$p" ] || exit 33; fi; fi; `) + } + command.WriteString("printf '%s\\n' \"$alive\"") + return command.String() +} + +func (f *microsandboxLogCancelFixture) awaitProcessDeath( + ctx context.Context, identities []microsandboxExecIdentity, +) { + ticker := time.NewTicker(200 * time.Millisecond) + defer ticker.Stop() + for { + alive, err := f.processesAlive(ctx, identities) + framework.ExpectNoError(err) + if alive == 0 { + return + } + select { + case <-ticker.C: + case <-ctx.Done(): + ginkgo.Fail( + "original guest process identities remained live after the 10s cancellation bound", + ) + } + } +} + +func (f *microsandboxLogCancelFixture) cleanupProcesses(identities []microsandboxExecIdentity) { + if len(identities) == 0 { + return + } + ctx, cancel := context.WithTimeout(context.Background(), microsandboxCancelBound) + defer cancel() + _, err := f.framework.DevsySSHOnce(ctx, f.workspace, f.identityCommand(identities, true)) + framework.ExpectNoError(err) +} + +func (f *microsandboxLogCancelFixture) assertUnchanged(ctx context.Context) { + out, err := microsandboxResourceCommand( + ctx, + "inspect", + f.sandbox, + "--format", + microsandboxResourceJSON, + ) + framework.ExpectNoError(err) + var state microsandboxLogCancelState + framework.ExpectNoError(json.Unmarshal(out, &state)) + gomega.Expect(state.Status).To(gomega.Equal("Running")) + gomega.Expect(microsandboxCreationTime(ctx, f.sandbox)).To(gomega.Equal(f.createdAt)) + gomega.Expect(microsandboxStorageBootID(ctx, f.framework, f.workspace)). + To(gomega.Equal(f.bootID)) + f.assertSourceAndCommand(ctx) +} + +func (f *microsandboxLogCancelFixture) assertSourceAndCommand(ctx context.Context) { + command, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + var out strings.Builder + err := f.runtime.CommandContainerArgv( + command, + f.uid, + []string{ + microsandboxCancelShell, + "-c", + "printf '%s\\n' \"$1\"", + f.nonce, + f.nonce + "-next", + }, + driver.Streams{Stdout: &out, Stderr: io.Discard}, + ) + framework.ExpectNoError(err) + gomega.Expect(out.String()).To(gomega.Equal(f.nonce + "-next\n")) + data, err := os.ReadFile(filepath.Join(f.workspace, microsandboxCancelSentinel)) + framework.ExpectNoError(err) + gomega.Expect(string(data)).To(gomega.Equal(f.nonce)) + guestPath := filepath.Join( + "/workspaces", + filepath.Base(f.workspace), + microsandboxCancelSentinel, + ) + out.Reset() + err = f.runtime.CommandContainerArgv(command, f.uid, []string{"cat", guestPath}, + driver.Streams{Stdout: &out, Stderr: io.Discard}) + framework.ExpectNoError(err) + gomega.Expect(out.String()).To(gomega.Equal(f.nonce)) +} diff --git a/pkg/agent/delivery/exec_stream_cancellation_unix_test.go b/pkg/agent/delivery/exec_stream_cancellation_unix_test.go new file mode 100644 index 0000000000..9967166970 --- /dev/null +++ b/pkg/agent/delivery/exec_stream_cancellation_unix_test.go @@ -0,0 +1,261 @@ +//go:build linux || darwin + +package delivery + +import ( + "context" + "io" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/devsy-org/devsy/pkg/driver" + "github.com/devsy-org/devsy/pkg/driver/microsandbox" + "github.com/devsy-org/devsy/pkg/provider" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const execStreamCancellationTestBound = 3 * time.Second + +type cancellationBinary struct { + io.Reader + closer io.Closer + entered chan struct{} + once sync.Once + closes atomic.Int32 +} + +func (s *cancellationBinary) Read(p []byte) (int, error) { + s.once.Do(func() { close(s.entered) }) + return s.Reader.Read(p) +} + +func (s *cancellationBinary) Close() error { + s.closes.Add(1) + if s.closer != nil { + return s.closer.Close() + } + return nil +} + +type cancellationBinarySource struct { + mu sync.Mutex + blocked int + sources []*cancellationBinary + writers []*io.PipeWriter + acquired chan *cancellationBinary +} + +func (s *cancellationBinarySource) acquire(ctx context.Context, _ string) (io.ReadCloser, error) { + s.mu.Lock() + defer s.mu.Unlock() + // Cleanup cancels before taking this mutex; a concurrent retry must not + // allocate a new blocked reader after cleanup has closed the existing ones. + if err := ctx.Err(); err != nil { + return nil, err + } + source := &cancellationBinary{entered: make(chan struct{})} + if len(s.sources) < s.blocked { + reader, writer := io.Pipe() + source.Reader, source.closer = reader, reader + s.writers = append(s.writers, writer) + } else { + source.Reader = strings.NewReader("fresh-source-binary") + } + s.sources = append(s.sources, source) + s.acquired <- source + return source, nil +} + +func (s *cancellationBinarySource) cleanup() { + s.mu.Lock() + defer s.mu.Unlock() + for _, source := range s.sources { + if source.closer != nil { + _ = source.closer.Close() + } + } + for _, writer := range s.writers { + _ = writer.Close() + } +} + +func (s *cancellationBinarySource) assertOwnership(t *testing.T, expected int) { + t.Helper() + s.mu.Lock() + defer s.mu.Unlock() + require.Len(t, s.sources, expected, "every retry must acquire a fresh source") + for index, source := range s.sources { + select { + case <-source.entered: + default: + t.Errorf("source %d was acquired but never read", index) + } + assert.EqualValues(t, 1, source.closes.Load(), "source %d close count", index) + if index > 0 { + assert.NotSame(t, s.sources[index-1], source, "retry reused the previous source") + } + } +} + +func cancellationDeliveryRuntime(t *testing.T, blockCompletion bool) driver.ArgvExecDriver { + t.Helper() + dir := t.TempDir() + script := "#!/bin/sh\nexec /bin/cat\n" + if blockCompletion { + // A stage's small finite source drains into the pipe, but its process + // remains alive. CommandContext must kill it on the real completion timer. + script = "#!/bin/sh\ncase \"$*\" in *\"cat >\"*) exec /bin/sleep 60;; *) exit 0;; esac\n" + } + // #nosec G306 -- Fixed fake executable in the test's private temporary directory. + require.NoError(t, os.WriteFile(filepath.Join(dir, "msb"), []byte(script), 0o700)) + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + runtime, err := microsandbox.NewMicrosandboxDriver( + context.Background(), + &provider.AgentWorkspaceInfo{}, + ) + require.NoError(t, err) + argvRuntime, ok := runtime.(driver.ArgvExecDriver) + require.True(t, ok) + return argvRuntime +} + +func cancellationDeliveryExecutor( + t *testing.T, runtime driver.ArgvExecDriver, canceled chan<- error, +) func(context.Context, []string, driver.Streams) error { + return func(ctx context.Context, argv []string, streams driver.Streams) error { + if streams.Stdin != nil { + callbackDone := make(chan struct{}) + stop := context.AfterFunc(ctx, func() { + defer close(callbackDone) + select { + case canceled <- context.Cause(ctx): + default: + t.Error("unexpected extra attempt cancellation") + } + }) + defer func() { + if !stop() { + select { + case <-callbackDone: + case <-time.After(execStreamCancellationTestBound): + t.Error("attempt observer callback did not join") + } + } + }() + } + return runtime.CommandContainerArgv(ctx, "owned-delivery-test", argv, streams) + } +} + +type cancellationDeliveryCall struct { + done chan struct{} + cancel context.CancelFunc + err error +} + +func startCancellationDelivery( + t *testing.T, delivery *KubernetesDelivery, sources *cancellationBinarySource, +) (context.Context, *cancellationDeliveryCall) { + t.Helper() + outer, cancel := context.WithCancel(context.Background()) + call := &cancellationDeliveryCall{done: make(chan struct{}), cancel: cancel} + t.Cleanup(func() { + cancel() + sources.cleanup() + select { + case <-call.done: + case <-time.After(execStreamCancellationTestBound): + t.Error("delivery failed to join after explicit source cleanup") + } + }) + go func() { + defer close(call.done) + // The fake msb never interprets this stage destination or writes to it. + call.err = delivery.deliverViaExecStream( + outer, + "/tmp/owned-delivery-never-written", + PostStartOptions{ + Arch: "amd64", SkipVersionCheck: true, BinarySource: sources.acquire, + }, + ) + }() + return outer, call +} + +func awaitDeliveryRead(t *testing.T, sources *cancellationBinarySource) { + t.Helper() + select { + case source := <-sources.acquired: + select { + case <-source.entered: + case <-time.After(execStreamCancellationTestBound): + t.Fatal("actual source Read not entered") + } + case <-time.After(execStreamCancellationTestBound): + t.Fatal("source not acquired") + } +} + +func TestExecStreamAttemptCancellationOwnsSource(t *testing.T) { + for _, testCase := range []struct { + name string + blockedSources int + blockCompletion bool + cause error + result error + acquisitions int + }{ + {"retry acquires fresh source", 1, false, errExecStreamIdleTimeout, nil, 2}, + {"idle retries exhausted", 2, false, errExecStreamIdleTimeout, errExecStreamIdleTimeout, 2}, + {"completion retries exhausted", 0, true, errExecStreamCompletionTimeout, errExecStreamCompletionTimeout, 2}, + {"normal completion", 0, false, nil, nil, 1}, + } { + t.Run(testCase.name, func(t *testing.T) { + runtime := cancellationDeliveryRuntime(t, testCase.blockCompletion) + sources := &cancellationBinarySource{ + blocked: testCase.blockedSources, + acquired: make(chan *cancellationBinary, execStreamMaxAttempts), + } + canceled := make(chan error, execStreamMaxAttempts) + delivery := &KubernetesDelivery{ + execStreamIdleTimeout: 300 * time.Millisecond, + execStreamCompletionTimeout: 300 * time.Millisecond, + Exec: cancellationDeliveryExecutor(t, runtime, canceled), + } + outer, call := startCancellationDelivery(t, delivery, sources) + awaitDeliveryRead(t, sources) + if testCase.cause != nil { + select { + case cause := <-canceled: + require.ErrorIs(t, cause, testCase.cause) + case <-time.After(execStreamCancellationTestBound): + t.Fatal("actual attempt cancellation was not observed") + } + } + select { + case <-call.done: + case <-time.After(execStreamCancellationTestBound): + t.Fatal("delivery remained blocked after actual attempt cancellation") + } + if testCase.result == nil { + require.NoError(t, call.err) + } else { + require.ErrorIs( + t, + call.err, + testCase.result, + "original attempt cause must survive cleanup and retry", + ) + } + require.NoError(t, outer.Err(), "attempt timeout must not cancel the outer operation") + call.cancel() + sources.assertOwnership(t, testCase.acquisitions) + }) + } +} diff --git a/pkg/agent/delivery/kubernetes.go b/pkg/agent/delivery/kubernetes.go index 2ad79924d5..3b1ef0574e 100644 --- a/pkg/agent/delivery/kubernetes.go +++ b/pkg/agent/delivery/kubernetes.go @@ -203,9 +203,10 @@ func (d *KubernetesDelivery) deliverViaExecStream( if err != nil { return &permanentDeliveryError{fmt.Errorf("acquire binary: %w", err)} } - defer func() { _ = binary.Close() }() + owned := &execStreamSource{ReadCloser: binary} + defer func() { _ = owned.Close() }() - streamErr := d.execStreamOnce(ctx, destPath, binary, opts.SkipVersionCheck) + streamErr := d.execStreamOnce(ctx, destPath, owned, opts.SkipVersionCheck) if streamErr != nil && isTransientDeliveryError(streamErr) && attempt < execStreamMaxAttempts { log.Warnf( @@ -222,6 +223,18 @@ func (d *KubernetesDelivery) deliverViaExecStream( return err } +// execStreamSource joins cancel-time closure with the acquiring owner's cleanup. +type execStreamSource struct { + io.ReadCloser + closeOnce sync.Once + closeErr error +} + +func (s *execStreamSource) Close() error { + s.closeOnce.Do(func() { s.closeErr = s.ReadCloser.Close() }) + return s.closeErr +} + type permanentDeliveryError struct{ err error } func (e *permanentDeliveryError) Error() string { return e.err.Error() } @@ -241,6 +254,18 @@ func (d *KubernetesDelivery) execStreamOnce( attemptCtx, cancel := context.WithCancelCause(ctx) defer cancel(nil) + if source, ok := binary.(io.Closer); ok { + closed := make(chan struct{}) + stop := context.AfterFunc(attemptCtx, func() { + _ = source.Close() + close(closed) + }) + defer func() { + if !stop() { + <-closed + } + }() + } progress := newProgressReader(binary) monitorDone := make(chan struct{}) monitorStop := make(chan struct{}) diff --git a/pkg/driver/microsandbox/cliclient.go b/pkg/driver/microsandbox/cliclient.go index 4c0230b314..799f017aa2 100644 --- a/pkg/driver/microsandbox/cliclient.go +++ b/pkg/driver/microsandbox/cliclient.go @@ -153,6 +153,8 @@ func (cliClient) Remove(ctx context.Context, sandbox string) error { // Exec uses --stream (byte-faithful stdio, no PTY) because the agent binary // injection and the SSH-over-stdio tunnel require it; a PTY stalls the tunnel. +// Cancellation closes non-file ReadCloser input; Close must unblock Read. +// File input and input after normal completion remain caller-owned. func (cliClient) Exec(ctx context.Context, sandbox string, req execRequest) error { args := []string{"exec", "--stream"} if req.User != "" { @@ -169,7 +171,27 @@ func (cliClient) Exec(ctx context.Context, sandbox string, req execRequest) erro cmd.Stdin = req.Stdin cmd.Stdout = req.Stdout cmd.Stderr = req.Stderr - return cmd.Run() + input, closable := req.Stdin.(io.ReadCloser) + _, file := req.Stdin.(*os.File) + if !closable || file { + return cmd.Run() + } + if err := cmd.Start(); err != nil { + return err + } + // Non-file input is delegated for cancellation only. Keep the callback + // alive while Wait drains stdin, even if the child has already exited. + closed := make(chan struct{}) + stop := context.AfterFunc(ctx, func() { + _ = input.Close() + close(closed) + }) + defer func() { + if !stop() { + <-closed + } + }() + return cmd.Wait() } func (cliClient) Logs(ctx context.Context, sandbox string, w io.Writer) error { diff --git a/pkg/driver/microsandbox/exec_cancellation_unix_test.go b/pkg/driver/microsandbox/exec_cancellation_unix_test.go new file mode 100644 index 0000000000..9dd09a88e8 --- /dev/null +++ b/pkg/driver/microsandbox/exec_cancellation_unix_test.go @@ -0,0 +1,248 @@ +//go:build linux || darwin + +package microsandbox + +import ( + "context" + "errors" + "io" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const execCancellationTestBound = 3 * time.Second + +type cancellationInput struct { + io.Reader + closer func() error + entered chan struct{} + readOnce sync.Once + closes atomic.Int32 + closeEntered chan struct{} + closeGate <-chan struct{} +} + +func (r *cancellationInput) Read(p []byte) (int, error) { + if r.entered != nil { + r.readOnce.Do(func() { close(r.entered) }) + } + return r.Reader.Read(p) +} + +func (r *cancellationInput) Close() error { + r.closes.Add(1) + if r.closeEntered != nil { + close(r.closeEntered) + } + if r.closeGate != nil { + <-r.closeGate + } + if r.closer != nil { + return r.closer() + } + return nil +} + +func cancellationMSB(t *testing.T, exitImmediately bool) string { + t.Helper() + dir := t.TempDir() + ready := filepath.Join(dir, "ready") + script := "#!/bin/sh\nprintf '%s\\n' \"$$\" > \"$DEVSY_CANCEL_READY\"\n" + if exitImmediately { + script += "exit 0\n" + } else { + script += "exec /bin/cat\n" + } + // #nosec G306 -- This executable is a fixed test-owned fake in a private temporary directory. + require.NoError(t, os.WriteFile(filepath.Join(dir, "msb"), []byte(script), 0o700)) + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("DEVSY_CANCEL_READY", ready) + return ready +} + +func awaitCancellationPID(t *testing.T, path string) int { + t.Helper() + var pid int + require.Eventually(t, func() bool { + data, err := os.ReadFile(path) // #nosec G304 -- Only the test-owned readiness file is read. + if err != nil { + return false + } + pid, err = strconv.Atoi(strings.TrimSpace(string(data))) + return err == nil && pid > 1 + }, execCancellationTestBound, time.Millisecond, "child readiness absent") + return pid +} + +func awaitCancellationSignal(t *testing.T, signal <-chan struct{}, message string) { + t.Helper() + select { + case <-signal: + case <-time.After(execCancellationTestBound): + t.Fatal(message) + } +} + +type cancellationCall struct { + cancel context.CancelFunc + done chan struct{} + err error +} + +func startCancellationCall(t *testing.T, input io.Reader, cleanup func()) *cancellationCall { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + call := &cancellationCall{cancel: cancel, done: make(chan struct{})} + t.Cleanup(func() { + cancel() + if cleanup != nil { + cleanup() + } + select { + case <-call.done: + case <-time.After(execCancellationTestBound): + t.Error("Exec failed to join after explicit input cleanup") + } + }) + go func() { + defer close(call.done) + call.err = (cliClient{}).Exec(ctx, "owned-cancellation-test", execRequest{ + Stdin: input, Stdout: io.Discard, Stderr: io.Discard, + }) + }() + return call +} + +func TestExecCancellationClosesBlockedInput(t *testing.T) { + for _, testCase := range []struct { + name string + exitImmediately bool + }{ + {name: "running child"}, + {name: "child exited while stdin copy remains blocked", exitImmediately: true}, + } { + t.Run(testCase.name, func(t *testing.T) { + ready := cancellationMSB(t, testCase.exitImmediately) + reader, writer := io.Pipe() + input := &cancellationInput{ + Reader: reader, + closer: reader.Close, + entered: make(chan struct{}), + } + call := startCancellationCall(t, input, func() { + _ = reader.Close() + _ = writer.Close() + }) + awaitCancellationSignal(t, input.entered, "source Read not entered") + pid := awaitCancellationPID(t, ready) + if testCase.exitImmediately { + require.Eventually(t, func() bool { + return errors.Is(syscall.Kill(pid, 0), syscall.ESRCH) + }, execCancellationTestBound, time.Millisecond, "child did not exit before cancellation") + } + call.cancel() + awaitCancellationSignal(t, call.done, "Exec stayed blocked after cancellation") + if !testCase.exitImmediately { + require.Error(t, call.err, "cancellation must not report successful execution") + } + assert.EqualValues(t, 1, input.closes.Load()) + }) + } +} + +func TestExecCompletionPreservesInputOwnership(t *testing.T) { + cancellationMSB(t, false) + input := &cancellationInput{Reader: strings.NewReader("complete")} + call := startCancellationCall(t, input, nil) + awaitCancellationSignal(t, call.done, "normal execution did not finish") + require.NoError(t, call.err) + call.cancel() + assert.Never( + t, + func() bool { return input.closes.Load() != 0 }, + 50*time.Millisecond, + time.Millisecond, + "normal completion must disarm the cancellation callback", + ) +} + +func TestExecCancellationPreservesCallerFile(t *testing.T) { + ready := cancellationMSB(t, false) + reader, writer, err := os.Pipe() + require.NoError(t, err) + call := startCancellationCall(t, reader, func() { + _ = reader.Close() + _ = writer.Close() + }) + awaitCancellationPID(t, ready) + call.cancel() + awaitCancellationSignal(t, call.done, "file-backed Exec cancellation did not return") + require.Error(t, call.err) + _, err = reader.Stat() + assert.NoError(t, err, "the caller retains ownership of its file") +} + +func TestExecStartFailurePreservesInput(t *testing.T) { + dir := t.TempDir() + // #nosec G306 -- Intentionally executable fixture with an invalid interpreter, in a private directory. + require.NoError( + t, + os.WriteFile(filepath.Join(dir, "msb"), []byte("#!/nonexistent-interpreter\n"), 0o700), + ) + t.Setenv("PATH", dir) + input := &cancellationInput{Reader: strings.NewReader("unused")} + call := startCancellationCall(t, input, nil) + awaitCancellationSignal(t, call.done, "failed Start did not return") + require.Error(t, call.err) + call.cancel() + assert.Zero(t, input.closes.Load(), "failed Start must not transfer input ownership") +} + +func TestExecAlreadyCanceledPreservesInput(t *testing.T) { + cancellationMSB(t, false) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + input := &cancellationInput{Reader: strings.NewReader("unused")} + err := (cliClient{}).Exec(ctx, "owned-cancellation-test", execRequest{Stdin: input}) + require.ErrorIs(t, err, context.Canceled) + assert.Zero(t, input.closes.Load(), "pre-start cancellation must not transfer input ownership") +} + +func TestExecCancellationJoinsInputClose(t *testing.T) { + ready := cancellationMSB(t, false) + reader, writer := io.Pipe() + gate := make(chan struct{}) + var release sync.Once + input := &cancellationInput{ + Reader: reader, closer: reader.Close, entered: make(chan struct{}), + closeEntered: make(chan struct{}), closeGate: gate, + } + call := startCancellationCall(t, input, func() { + release.Do(func() { close(gate) }) + _ = reader.Close() + _ = writer.Close() + }) + awaitCancellationPID(t, ready) + awaitCancellationSignal(t, input.entered, "source Read not entered") + call.cancel() + awaitCancellationSignal(t, input.closeEntered, "cancellation did not invoke Close") + select { + case <-call.done: + t.Fatal("Exec returned before its Close callback completed") + default: + } + release.Do(func() { close(gate) }) + awaitCancellationSignal(t, call.done, "Exec did not join the completed Close callback") + require.Error(t, call.err) + assert.EqualValues(t, 1, input.closes.Load()) +} diff --git a/sites/docs-devsy-sh/content/docs/developing-providers/runtime-protocol.mdx b/sites/docs-devsy-sh/content/docs/developing-providers/runtime-protocol.mdx index 738a288350..0ba64a9bd8 100644 --- a/sites/docs-devsy-sh/content/docs/developing-providers/runtime-protocol.mdx +++ b/sites/docs-devsy-sh/content/docs/developing-providers/runtime-protocol.mdx @@ -109,6 +109,14 @@ The `up-provider-microsandbox-prebuilds` label publishes a Dockerfile with a loc DEVSY_REQUIRE_MICROSANDBOX=true task cli:test:e2e:suite -- up-provider-microsandbox-prebuilds ``` -These scenarios do not establish complete parity. Dockerless operation, logs, cancellation, and runtime compatibility failures still require coverage before replacing the built-in provider. Green lifecycle, image, mount, resource, persistent-storage, ephemeral-root, TCP-egress, and prebuild checks alone do not authorize that cutover. Egress coverage does not establish UDP, DNS, TLS, ingress, or live-policy modification behavior. +The `up-provider-microsandbox-logs-cancel` label checks completed stdout/stderr workload logs while running and stopped, direct runtime Exec cancellation of a ready parent and child, and cancellation with blocked owned stdin against each provider. A separate SSH observer verifies process identity using PID, kernel start time, and a unique command marker. Caller return and process death must both occur within ten seconds of cancellation; the VM's creation time, boot ID, and source files must remain intact. Logs do not start a stopped VM. These checks do not establish log ordering, binary log fidelity, TTY cancellation, or cancellation of arbitrary nonclosable input. + +```sh +DEVSY_REQUIRE_MICROSANDBOX=true task cli:test:e2e:suite -- up-provider-microsandbox-logs-cancel +``` + +For built-in Exec, cancellation closes non-file `io.ReadCloser` input whose `Close` unblocks `Read`; ordinary completion and caller-owned file input retain their ownership. Agent binary delivery owns its acquired source and closes it when the individual transfer attempt is canceled, before retrying with a fresh source. Cleanup joins cancellation callbacks before returning. + +These scenarios do not establish complete parity. Dockerless operation and runtime compatibility failures still require coverage before replacing the built-in provider. Green lifecycle, image, mount, resource, persistent-storage, ephemeral-root, TCP-egress, prebuild, logs, and cancellation checks alone do not authorize that cutover. Egress coverage does not establish UDP, DNS, TLS, ingress, or live-policy modification behavior. Prebuild invalidation coverage changes the Dockerfile; it does not independently establish hashing of local Feature script contents. From b4b41358e17589ceb5e5bc8d2416074dee7e405e Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 12 Oct 2026 23:33:45 -0600 Subject: [PATCH 2/2] test(runtime): frame MicroSandbox log markers explicitly --- e2e/tests/up/provider_microsandbox_logs_cancel.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/e2e/tests/up/provider_microsandbox_logs_cancel.go b/e2e/tests/up/provider_microsandbox_logs_cancel.go index 77900ed26b..41cc08ca24 100644 --- a/e2e/tests/up/provider_microsandbox_logs_cancel.go +++ b/e2e/tests/up/provider_microsandbox_logs_cancel.go @@ -206,10 +206,11 @@ func (f *microsandboxLogCancelFixture) loadRuntime(ctx context.Context, expected func (f *microsandboxLogCancelFixture) assertLogs(ctx context.Context) { outMarker, errMarker := f.nonce+"-stdout", f.nonce+"-stderr" + // Earlier Exec traffic can leave binary log bytes without a line boundary. command, cancel := context.WithTimeout(ctx, 30*time.Second) err := f.runtime.CommandContainerArgv(command, f.uid, []string{ - microsandboxCancelShell, "-c", "printf '%s\\n' \"$1\"; printf '%s\\n' \"$2\" >&2", + microsandboxCancelShell, "-c", "printf '\\n%s\\n' \"$1\"; printf '\\n%s\\n' \"$2\" >&2", f.nonce, outMarker, errMarker, }, driver.Streams{Stdout: io.Discard, Stderr: io.Discard})