fix(runtime): stop WASM pipe reads from stalling for maxBlockingReadMs (#1959)

A WASM pipeline stage read from a pipe via a synchronous, blocking process.fd_read on the sidecar reactor thread. While parked in that kernel read, the reactor could not run the upstream stage to completion, so the writer never closed its pipe end; the reader only unblocked when the maxBlockingReadMs watchdog fired — once per stage. A low cap turned that into a guest-visible `Would block`/EAGAIN with a nonzero exit; a high cap into a multi-second-per-stage stall (e.g. `env | sort | sed` ~20-30s).

Make WASM process.fd_read non-blocking (Some(Duration::ZERO)), mirroring the sibling process.fd_write path: the runner poll+retries a logically-blocking fd on EAGAIN, the reactor stays free to run the writer, and the reader observes EOF promptly. Pipeline latency no longer depends on maxBlockingReadMs.

Add a regression test asserting a low-cap pipeline succeeds without watchdog EAGAIN (with exact-output checks) and that three-stage pipeline latency is independent of the cap.

Author: abcxff

crates/native-sidecar/src/execution/javascript/rpc.rs

@@ -1181,17 +1181,31 @@ where
             .map_err(|_| SidecarError::InvalidState("fd_read length is too large".into()))?;
             let timeout_ms =
                 javascript_sync_rpc_arg_u64_optional(&request.args, 2, "fd_read timeout ms")?;
-            match timeout_ms {
-                Some(timeout_ms) => kernel
+            // Read non-blocking for WASM so the reactor is never parked in a
+            // pipe read; the runner poll+retries on EAGAIN (#1959).
+            if process.runtime == GuestRuntimeKind::WebAssembly {
+                kernel
                     .fd_read_with_timeout_result(
                         EXECUTION_DRIVER_NAME,
                         process.kernel_pid,
                         fd,
                         length,
-                        Some(Duration::from_millis(timeout_ms)),
+                        Some(Duration::ZERO),
                     )
-                    .map(Option::unwrap_or_default),
-                None => kernel.fd_read(EXECUTION_DRIVER_NAME, process.kernel_pid, fd, length),
+                    .map(Option::unwrap_or_default)
+            } else {
+                match timeout_ms {
+                    Some(timeout_ms) => kernel
+                        .fd_read_with_timeout_result(
+                            EXECUTION_DRIVER_NAME,
+                            process.kernel_pid,
+                            fd,
+                            length,
+                            Some(Duration::from_millis(timeout_ms)),
+                        )
+                        .map(Option::unwrap_or_default),
+                    None => kernel.fd_read(EXECUTION_DRIVER_NAME, process.kernel_pid, fd, length),
+                }
             }
             .map(|bytes| javascript_sync_rpc_bytes_value(&bytes))
             .map_err(kernel_error)

packages/core/tests/issue-1959-pipe-stall.repro.test.ts

@@ -0,0 +1,97 @@
+// Regression for https://github.com/rivet-dev/agentos/issues/1959
+//
+// A WASM pipeline stage reading from a pipe used to block the sidecar reactor
+// thread inside a synchronous kernel read (the `process.fd_read` sync RPC).
+// While parked, the reactor could not run the upstream stage to completion, so
+// the writer never closed its pipe end and the reader only unblocked when the
+// `maxBlockingReadMs` watchdog fired — once per pipeline stage. A low cap
+// turned that stall into a guest-visible `Would block` / EAGAIN and a nonzero
+// exit; a high cap turned it into a multi-second-per-stage stall.
+//
+// The fix makes WASM `process.fd_read` non-blocking (mirroring the sibling
+// `process.fd_write` path): the reactor stays free to run the writer, the
+// reader observes EOF promptly, and the pipeline completes with correct output
+// whose latency no longer depends on `maxBlockingReadMs`.
+import { describe, expect, test } from "vitest";
+import { AgentOs } from "../src/index.js";
+
+describe("issue-1959: pipe reader no longer stalls for maxBlockingReadMs", () => {
+	// A low cap is the sharpest probe: before the fix this stage read raced the
+	// watchdog and surfaced `sed: Would block` (rc=1). After the fix the cap is
+	// irrelevant — the read waits for the writer and observes EOF cleanly.
+	test("a fast pipeline succeeds under a low maxBlockingReadMs (no watchdog EAGAIN)", async () => {
+		const vm = await AgentOs.create({
+			limits: { resources: { maxBlockingReadMs: 500 } },
+		});
+		try {
+			const envSortSed = await vm.exec("env | sort | sed -n '1,2p'", {
+				cwd: "/workspace",
+				timeoutMs: 60_000,
+			});
+			expect(
+				envSortSed.exitCode,
+				`env|sort|sed stderr=${envSortSed.stderr}`,
+			).toBe(0);
+			expect(envSortSed.stderr.toLowerCase()).not.toMatch(
+				/would block|temporarily unavailable/,
+			);
+			expect(envSortSed.stdout.length).toBeGreaterThan(0);
+
+			// Exact outputs prove no bytes are dropped across the pipe.
+			const seq = await vm.exec("seq 1 5 | paste -sd,", {
+				cwd: "/workspace",
+				timeoutMs: 60_000,
+			});
+			expect(seq.exitCode, `seq stderr=${seq.stderr}`).toBe(0);
+			expect(seq.stdout).toBe("1,2,3,4,5\n");
+
+			const large = await vm.exec("seq 1 5000 | tail -n 1", {
+				cwd: "/workspace",
+				timeoutMs: 60_000,
+			});
+			expect(large.exitCode, `large stderr=${large.stderr}`).toBe(0);
+			expect(large.stdout).toBe("5000\n");
+		} finally {
+			await vm.dispose();
+		}
+	}, 120_000);
+
+	// The defining property of the fix: pipeline latency does not scale with
+	// maxBlockingReadMs. Before the fix, raising the cap made a three-stage
+	// pipeline pay the cap once per stage (~3x). After it, the two runs are
+	// close regardless of the cap.
+	test("three-stage pipeline latency is independent of maxBlockingReadMs", async () => {
+		const timePipeline = async (maxBlockingReadMs: number) => {
+			const vm = await AgentOs.create({
+				limits: { resources: { maxBlockingReadMs } },
+			});
+			try {
+				const startedAt = Date.now();
+				const r = await vm.exec("env | sort | sed -n '1,2p'", {
+					cwd: "/workspace",
+					timeoutMs: 60_000,
+				});
+				expect(r.exitCode, `cap=${maxBlockingReadMs} stderr=${r.stderr}`).toBe(
+					0,
+				);
+				return Date.now() - startedAt;
+			} finally {
+				await vm.dispose();
+			}
+		};
+
+		const lowCapMs = 500;
+		const highCapMs = 8000;
+		const lowElapsed = await timePipeline(lowCapMs);
+		const highElapsed = await timePipeline(highCapMs);
+
+		// If the per-stage stall were still present, the high-cap run would pay
+		// roughly an extra stage-worth of watchdog time. Allow generous slack for
+		// unoptimized (debug) builds and VM warmup while still catching a stall:
+		// a single extra stage at the 8s cap would blow past this bound.
+		expect(
+			highElapsed,
+			`low(${lowCapMs}ms cap)=${lowElapsed}ms high(${highCapMs}ms cap)=${highElapsed}ms`,
+		).toBeLessThan(lowElapsed + highCapMs);
+	}, 180_000);
+});