From cc4329a892d14210a39258db21b6fad2b7e23c7e Mon Sep 17 00:00:00 2001
From: Devin Ivy <devinivy@gmail.com>
Date: Sun, 19 Apr 2026 22:18:24 -0400
Subject: [PATCH] fix: plumb highWaterMark through to stream producer

dispatchStream was reading opts.highWaterMark into a local but never
forwarding it to the worker. Worker-side pipeIterable was defaulting
to its own hardcoded 16, so user-supplied highWaterMark had no effect
after the atomics refactor.

Include highWater in the dispatch message; handleStreamTask passes it
to pipeIterable's flags-branch so the producer parks at the requested
threshold.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 src/execute.ts      |  2 +-
 src/worker-entry.ts | 14 ++++++++++++--
 2 files changed, 13 insertions(+), 3 deletions(-)

diff --git a/src/execute.ts b/src/execute.ts
index 61383d0..58cc0f4 100644
--- a/src/execute.ts
+++ b/src/execute.ts
@@ -117,7 +117,7 @@ export function dispatchStream<T>(
   streamPortStack.push([]);
   const preparedArgs = extracted.args.map(prepareArg);
   const ports = streamPortStack.pop()!;
-  const msg = { id, args: preparedArgs, port: port2, flags: flags.buffer };
+  const msg = { id, args: preparedArgs, port: port2, flags: flags.buffer, highWater };
   worker.postMessage(msg, [...extracted.transfer, ...ports, port2] as any[]);
 
   let resolveDone: () => void;
diff --git a/src/worker-entry.ts b/src/worker-entry.ts
index b328722..160d3be 100644
--- a/src/worker-entry.ts
+++ b/src/worker-entry.ts
@@ -166,7 +166,14 @@ async function resolveArg(arg: unknown): Promise<unknown> {
   return deserializeArg(arg);
 }
 
-type TaskMsg = { callId?: number; id: string; args: unknown[]; port?: MessagePort; flags?: SharedArrayBuffer };
+type TaskMsg = {
+  callId?: number;
+  id: string;
+  args: unknown[];
+  port?: MessagePort;
+  flags?: SharedArrayBuffer;
+  highWater?: number;
+};
 
 function postResult(callId: number, value: unknown): void {
   const returnValue = serializeArg(value);
@@ -257,7 +264,10 @@ async function handleStreamTask(msg: TaskMsg): Promise<void> {
       resolvedArgs[i] = needsAsyncResolve(a) ? await resolveArg(a) : deserializeArg(a);
     }
     const gen = fn(...resolvedArgs) as AsyncGenerator;
-    await pipeIterable(gen, port!, { flags: flags ? pipeFlagsFromBuffer(flags) : undefined });
+    await pipeIterable(gen, port!, {
+      flags: flags ? pipeFlagsFromBuffer(flags) : undefined,
+      highWater: msg.highWater,
+    });
   } catch (err) {
     if (callId != null) postError(callId, err);
   }
-- 
2.51.2