From 28f16464610bf579f757f6a60cd018393ddcb5c5 Mon Sep 17 00:00:00 2001
From: Chris Tate <chris@ctate.dev>
Date: Tue, 20 Oct 2026 12:29:12 -0500
Subject: [PATCH] Unpipe JavaScript destinations on unpipe() without a
 destination

With checked-dynamic stream slots, readable.unpipe() also unpipes every
JavaScript destination stream.pipeDyn registered from the source, oldest
pipe first. Each destination hears 'unpipe' and the source pauses, as in
Node's unpipe(), which empties every pipe. Before, those pipes stayed
attached and later chunks still reached them.
---
 .../src/frontend/lowering/lower-stream.ts     | 38 +++++++++++++++++++
 packages/runtime/src/scr_stream_dyn.c         | 22 ++++++++---
 .../5819-node-stream-dyn-direct-pipe/main.ts  | 13 ++++++-
 3 files changed, 66 insertions(+), 7 deletions(-)

diff --git a/packages/compiler/src/frontend/lowering/lower-stream.ts b/packages/compiler/src/frontend/lowering/lower-stream.ts
index 4d729f639..44027f65c 100644
--- a/packages/compiler/src/frontend/lowering/lower-stream.ts
+++ b/packages/compiler/src/frontend/lowering/lower-stream.ts
@@ -3876,6 +3876,44 @@ export function lowerStreamMethodCall(
         loc,
       };
     }
+    // Checked-dynamic slots: unpipe() also unpipes the JavaScript
+    // destinations stream.pipeDyn registered (stream.unpipeDyn with an
+    // undefined destination), as Node's unpipe() empties every pipe.
+    if (lowerer.dynNodeStreams) {
+      const source = lowerer.declareHiddenLocal("%unpipeSource", receiver.type);
+      const sourceRef = varRef(source.id, receiver.type, loc);
+      return {
+        kind: "seqExpr",
+        stmts: [
+          { kind: "varDecl", localId: source.id, init: receiver, loc },
+          {
+            kind: "exprStmt",
+            expr: {
+              kind: "libCall",
+              fn: "readable.unpipe",
+              args: [sourceRef],
+              type: receiver.type,
+              loc,
+            },
+            loc,
+          },
+          {
+            kind: "exprStmt",
+            expr: {
+              kind: "libCall",
+              fn: "stream.unpipeDyn",
+              args: [sourceRef, dynUndefinedExpr(loc)],
+              type: VOID,
+              loc,
+            },
+            loc,
+          },
+        ],
+        result: sourceRef,
+        type: receiver.type,
+        loc,
+      };
+    }
     return { kind: "libCall", fn: "readable.unpipe", args: [receiver], type: receiver.type, loc };
   }
 
diff --git a/packages/runtime/src/scr_stream_dyn.c b/packages/runtime/src/scr_stream_dyn.c
index d230bf22e..0859c4e07 100644
--- a/packages/runtime/src/scr_stream_dyn.c
+++ b/packages/runtime/src/scr_stream_dyn.c
@@ -365,13 +365,25 @@ out:
 /* readable.unpipe(destination) for a destination that is not a native
  * stream: the pipe scr_stream_pipe_dyn registered from src into it
  * unpipes, as Node's unpipe does for a piped destination; nothing happens
- * when there is none. Borrowed. */
+ * when there is none. An undefined destination (unpipe() without one)
+ * unpipes every JavaScript destination of src, oldest pipe first.
+ * Borrowed. */
 void scr_stream_unpipe_dyn(ScrStream *src, ScrDyn *dest) {
-  for (ScrDynPipe *p = scr_dyn_pipe_live; p; p = p->live_next) {
-    if (p->src != src || p->unpiped || p->cleaned || !p->dest || !scr_dyn_strict_eq(p->dest, dest)) continue;
-    ScrDynPipe *held = scr_dyn_pipe_retain_v(p);
+  bool all = dest == NULL || dest->kind == SCR_DYN_UNDEF;
+  for (;;) {
+    /* The live list is newest first; the oldest matching pipe goes next.
+     * Unpiping runs JavaScript ('unpipe' listeners), so each round walks
+     * the list again. */
+    ScrDynPipe *match = NULL;
+    for (ScrDynPipe *p = scr_dyn_pipe_live; p; p = p->live_next) {
+      if (p->src != src || p->unpiped || p->cleaned || !p->dest) continue;
+      if (!all && !scr_dyn_strict_eq(p->dest, dest)) continue;
+      match = p;
+    }
+    if (!match) return;
+    ScrDynPipe *held = scr_dyn_pipe_retain_v(match);
     scr_dyn_pipe_unpipe(held);
     scr_dyn_pipe_release_v(held);
-    return;
+    if (!all || scr_exc_pending()) return;
   }
 }
diff --git a/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts b/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts
index c1feb8eb5..a72328d3f 100644
--- a/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts
+++ b/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts
@@ -1,7 +1,8 @@
 // A native Readable piped straight into a JavaScript sink, the only
 // untyped stream flow in the program; unpipe of a NodeJS.WritableStream
-// that holds a native Writable or a JavaScript sink; and JavaScript
-// destinations whose prependListener or writableNeedDrain getter throws.
+// that holds a native Writable or a JavaScript sink; unpipe() of every
+// JavaScript sink; and JavaScript destinations whose prependListener or
+// writableNeedDrain getter throws.
 import { Readable, Writable } from "node:stream";
 import { plainSink, throwingSink } from "./sources.js";
 
@@ -37,6 +38,14 @@ async function main(): Promise<void> {
   await new Promise<void>((resolve) => setTimeout(resolve, 10));
   console.log("second paused", second.isPaused());
 
+  const third = new Readable({ read() {} });
+  third.pipe(plainSink("first of two"));
+  third.pipe(plainSink("second of two"));
+  third.unpipe();
+  third.push("after unpipe()");
+  await new Promise<void>((resolve) => setTimeout(resolve, 10));
+  console.log("third paused", third.isPaused(), third.listenerCount("data"));
+
   for (const member of ["prependListener", "writableNeedDrain"]) {
     try {
       new Readable({ read() {} }).pipe(throwingSink(member));