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);
   }