fix: prevent darwin-vz exec stream hangs (#90)

Amp-Thread-ID: https://ampcode.com/threads/T-019cd4f1-489c-7240-b1c9-edb74b908b7d

Author: lox

cmd/cleanroom-darwin-vz/main.swift

@@ -219,6 +219,7 @@ private final class GuestChannel {
     let readFD: Int32
     let writeFD: Int32
     private let closeAfterUse: Bool
+    private let lock = NSLock()
     private var closed = false
     private let onClose: () -> Void
 
@@ -236,12 +237,46 @@ private final class GuestChannel {
     }
 
     func close() {
+        lock.lock()
         if closed {
+            lock.unlock()
             return
         }
         closed = true
+        lock.unlock()
         onClose()
     }
+
+    func duplicate(closeAfterUse: Bool = true) throws -> GuestChannel {
+        let duplicatedReadFD = Darwin.dup(readFD)
+        if duplicatedReadFD < 0 {
+            throw HelperError.posix("dup(guest read fd)", errno)
+        }
+
+        let duplicatedWriteFD: Int32
+        if writeFD == readFD {
+            duplicatedWriteFD = duplicatedReadFD
+        } else {
+            duplicatedWriteFD = Darwin.dup(writeFD)
+            if duplicatedWriteFD < 0 {
+                let code = errno
+                _ = Darwin.close(duplicatedReadFD)
+                throw HelperError.posix("dup(guest write fd)", code)
+            }
+        }
+
+        return GuestChannel(
+            readFD: duplicatedReadFD,
+            writeFD: duplicatedWriteFD,
+            closeAfterUse: closeAfterUse,
+            onClose: {
+                if duplicatedWriteFD != duplicatedReadFD {
+                    _ = Darwin.close(duplicatedWriteFD)
+                }
+                _ = Darwin.close(duplicatedReadFD)
+            }
+        )
+    }
 }
 
 private final class ProxyServer {
@@ -308,7 +343,7 @@ private final class ProxyServer {
         activeChannel = guestChannel
         lock.unlock()
 
-        bridge(hostFD: hostFD, guestReadFD: guestChannel.readFD, guestWriteFD: guestChannel.writeFD)
+        bridge(hostFD: hostFD, guestChannel: guestChannel)
         guestChannel.finishSession()
 
         lock.lock()
@@ -319,7 +354,9 @@ private final class ProxyServer {
         return true
     }
 
-    private func bridge(hostFD: Int32, guestReadFD: Int32, guestWriteFD: Int32) {
+    private func bridge(hostFD: Int32, guestChannel: GuestChannel) {
+        let guestReadFD = guestChannel.readFD
+        let guestWriteFD = guestChannel.writeFD
         let group = DispatchGroup()
         let errorLock = NSLock()
         var firstError: Error?
@@ -335,6 +372,11 @@ private final class ProxyServer {
         group.enter()
         DispatchQueue.global(qos: .userInitiated).async {
             defer { group.leave() }
+            defer {
+                // Closing the guest side after host EOF prevents the bridge from
+                // stalling on long-lived transports (for example serial fallback).
+                guestChannel.close()
+            }
             do {
                 try pumpBytes(src: hostFD, dst: guestWriteFD)
             } catch {
@@ -661,7 +703,7 @@ private final class VMRuntime {
         guard let serialChannel else {
             throw HelperError.vm("guest serial channel is not available")
         }
-        return serialChannel
+        return try serialChannel.duplicate()
     }
 }
 

internal/backend/darwinvz/persistent_e2e_test.go

@@ -178,3 +178,118 @@ func TestPersistentSandboxE2E(t *testing.T) {
 		t.Fatalf("expected run after terminate to fail with unknown sandbox, got %v", err)
 	}
 }
+
+func TestPersistentSandboxE2EExecStreamingDoesNotHang(t *testing.T) {
+	if strings.TrimSpace(os.Getenv(darwinVZE2EEnvEnabled)) == "" {
+		t.Skipf("set %s=1 to run real darwin-vz persistence e2e", darwinVZE2EEnvEnabled)
+	}
+	if testing.Short() {
+		t.Skip("skipping darwin-vz e2e in short mode")
+	}
+
+	helperPath, err := resolveHelperBinaryPath()
+	if err != nil {
+		t.Fatalf("resolve helper binary: %v", err)
+	}
+	hasEntitlement, err := helperHasVirtualizationEntitlement(helperPath)
+	if err != nil {
+		t.Fatalf("verify helper entitlement: %v", err)
+	}
+	if !hasEntitlement {
+		t.Fatalf("helper %q is missing com.apple.security.virtualization entitlement", helperPath)
+	}
+	if _, _, err := New().getGuestAgentBinary(); err != nil {
+		t.Fatalf("resolve guest agent binary: %v", err)
+	}
+
+	rootFSOverride := strings.TrimSpace(os.Getenv(darwinVZE2EEnvRootFS))
+	if rootFSOverride == "" {
+		if _, err := hosttools.ResolveE2FSProgsBinary("mkfs.ext4"); err != nil {
+			t.Fatalf("resolve mkfs.ext4: %v", err)
+		}
+		if _, err := hosttools.ResolveE2FSProgsBinary("debugfs"); err != nil {
+			t.Fatalf("resolve debugfs: %v", err)
+		}
+	}
+
+	imageRef := strings.TrimSpace(os.Getenv(darwinVZE2EEnvImageRef))
+	if imageRef == "" {
+		imageRef = defaultDarwinVZE2EImageRef()
+	}
+
+	cfg := backend.FirecrackerConfig{
+		KernelImagePath: strings.TrimSpace(os.Getenv(darwinVZE2EEnvKernelImage)),
+		RootFSPath:      rootFSOverride,
+		VCPUs:           1,
+		MemoryMiB:       1024,
+		LaunchSeconds:   90,
+	}
+	compiled := &policy.CompiledPolicy{
+		Version:        1,
+		ImageRef:       imageRef,
+		NetworkDefault: "deny",
+	}
+	sandboxID := fmt.Sprintf("cr-e2e-streaming-%d", time.Now().UnixNano())
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+
+	adapter := New()
+	if err := adapter.ProvisionSandbox(ctx, backend.ProvisionRequest{
+		SandboxID:         sandboxID,
+		Policy:            compiled,
+		FirecrackerConfig: cfg,
+	}); err != nil {
+		t.Fatalf("ProvisionSandbox returned error: %v", err)
+	}
+
+	defer func() {
+		if err := adapter.TerminateSandbox(context.Background(), sandboxID); err != nil {
+			t.Fatalf("deferred TerminateSandbox returned error: %v", err)
+		}
+	}()
+
+	// A quick warm-up run that exercises the readiness/proxy path before we
+	// assert repeated non-interactive executions complete promptly.
+	warmupCtx, warmupCancel := context.WithTimeout(ctx, 60*time.Second)
+	warmup, err := adapter.RunInSandbox(warmupCtx, backend.RunRequest{
+		SandboxID: sandboxID,
+		RunID:     "run-warmup",
+		Command:   []string{"sh", "-lc", "echo warmup"},
+		Policy:    compiled,
+		FirecrackerConfig: backend.FirecrackerConfig{
+			LaunchSeconds: cfg.LaunchSeconds,
+		},
+	}, backend.OutputStream{})
+	warmupCancel()
+	if err != nil {
+		t.Fatalf("warm-up RunInSandbox returned error: %v", err)
+	}
+	if warmup.ExitCode != 0 {
+		t.Fatalf("expected warm-up exit code 0, got %d (stderr=%q)", warmup.ExitCode, warmup.Stderr)
+	}
+
+	for i := 0; i < 8; i++ {
+		runID := fmt.Sprintf("run-stream-%d", i)
+		want := fmt.Sprintf("stream-%d", i)
+		runCtx, runCancel := context.WithTimeout(ctx, 30*time.Second)
+		res, err := adapter.RunInSandbox(runCtx, backend.RunRequest{
+			SandboxID: sandboxID,
+			RunID:     runID,
+			Command:   []string{"sh", "-lc", "echo " + want},
+			Policy:    compiled,
+			FirecrackerConfig: backend.FirecrackerConfig{
+				LaunchSeconds: cfg.LaunchSeconds,
+			},
+		}, backend.OutputStream{})
+		runCancel()
+		if err != nil {
+			t.Fatalf("RunInSandbox %q returned error: %v", runID, err)
+		}
+		if res.ExitCode != 0 {
+			t.Fatalf("expected %q exit code 0, got %d (stderr=%q)", runID, res.ExitCode, res.Stderr)
+		}
+		if got := strings.TrimSpace(res.Stdout); got != want {
+			t.Fatalf("unexpected stdout for %q: got %q want %q", runID, got, want)
+		}
+	}
+}