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.

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

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

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