streams: stop a stdio source when its reader is released

src/js/builtins/ConsoleObject.ts

@@ -26,6 +26,8 @@ export function asyncIterator(this: Console) {
           last = i + 1;
           i = indexOf(actualChunk, last);
         }
+        // What follows the last newline of the chunk is the start of the next line.
+        if (last < actualChunk.length) pendingChunk = actualChunk.subarray(last);
 
         for (idx++; idx < value_len; idx++) {
           actualChunk = value[idx];

src/jsc/bindings/webcore/streams/BunStreamSource.cpp

@@ -195,6 +195,23 @@ void resumeNativeSource(JSReadableStream* stream)
         Bun__FileReader__resume(file->wrapped());
 }
 
+extern "C" void Bun__FileReader__onReaderRelease(void* source);
+
+void releaseNativeSource(JSReadableStream* stream)
+{
+    if (stream->m_controllerKind != ControllerKind::Default)
+        return;
+    auto* controller = uncheckedDowncast<JSReadableStreamDefaultController>(stream->m_controller.get());
+    if (controller->m_algorithms.kind != SourceKind::Native)
+        return;
+    auto* handle = uncheckedDowncast<JSNativeStreamSourceAdapter>(controller->m_algorithms.algorithmContext.get())->handle();
+    if (!handle)
+        return;
+    // Only a file source holds an event-loop ref or reads by itself, so Blob and Bytes handles have nothing to release.
+    if (auto* file = dynamicDowncast<JSFileInternalReadableStreamSource>(handle))
+        Bun__FileReader__onReaderRelease(file->wrapped());
+}
+
 // object.<name>(...args) with a real [[Get]], as the replaced builtins did.
 static JSValue invokeMethod(JSC::VM& vm, JSGlobalObject* globalObject, JSObject* object, const Identifier& name, const MarkedArgumentBuffer& args)
 {

src/jsc/bindings/webcore/streams/ReadableStreamOperations.cpp

@@ -507,30 +507,6 @@ void readableStreamReaderGenericInitialize(JSGlobalObject* globalObject, JSReada
     stream->m_reader.set(vm, stream, reader);
 }
 
-// Bun: drop the native handle's event-loop ref when its consumer releases the lock.
-static void releaseNativeSourceRef(JSGlobalObject* globalObject, JSReadableStream* stream)
-{
-    auto& vm = getVM(globalObject);
-    auto scope = DECLARE_THROW_SCOPE(vm);
-    if (stream->m_controllerKind != ControllerKind::Default)
-        return;
-    auto* controller = defaultControllerOf(stream);
-    if (controller->m_algorithms.kind != SourceKind::Native)
-        return;
-    auto* handle = uncheckedDowncast<WebCore::JSNativeStreamSourceAdapter>(controller->m_algorithms.algorithmContext.get())->handle();
-    if (!handle)
-        return;
-    JSValue updateRef = handle->getIfPropertyExists(globalObject, builtinNames(vm).updateRefPublicName());
-    RETURN_IF_EXCEPTION(scope, void());
-    if (!updateRef || !updateRef.isCallable())
-        return;
-    MarkedArgumentBuffer args;
-    args.append(jsBoolean(false));
-    ASSERT(!args.hasOverflowed());
-    scope.release();
-    JSC::call(globalObject, updateRef, JSC::getCallData(updateRef), handle, args);
-}
-
 // ReadableStreamReaderGenericRelease(reader)
 void readableStreamReaderGenericRelease(JSGlobalObject* globalObject, JSReadableStreamReaderBase* reader)
 {
@@ -552,8 +528,7 @@ void readableStreamReaderGenericRelease(JSGlobalObject* globalObject, JSReadable
             RETURN_IF_EXCEPTION(scope, void());
         }
     }
-    releaseNativeSourceRef(globalObject, stream);
-    RETURN_IF_EXCEPTION(scope, void());
+    releaseNativeSource(stream);
 
     if (stream->m_state == ReadableStreamState::Readable)
         rejectPromiseAsHandled(globalObject, reader->m_closedPromise.get(), releaseError);

src/jsc/bindings/webcore/streams/WebStreamsInternals.h

@@ -279,7 +279,7 @@ void setUpReadableStreamDefaultReader(JSC::JSGlobalObject*, JSReadableStreamDefa
 void setUpReadableStreamBYOBReader(JSC::JSGlobalObject*, JSReadableStreamBYOBReader*, JSReadableStream*); // userJS: no — ReadableStreamOperations.cpp
 JSC::JSPromise* readableStreamReaderGenericCancel(JSC::JSGlobalObject*, JSReadableStreamReaderBase*, JSC::JSValue reason); // userJS: yes — ReadableStreamOperations.cpp
 void readableStreamReaderGenericInitialize(JSC::JSGlobalObject*, JSReadableStreamReaderBase*, JSReadableStream*); // userJS: no, never throws — ReadableStreamOperations.cpp
-void readableStreamReaderGenericRelease(JSC::JSGlobalObject*, JSReadableStreamReaderBase*); // userJS: no (also runs Bun's native-handle updateRef(false) gate) — ReadableStreamOperations.cpp
+void readableStreamReaderGenericRelease(JSC::JSGlobalObject*, JSReadableStreamReaderBase*); // userJS: no (also runs Bun's releaseNativeSource) — ReadableStreamOperations.cpp
 
 // Stream-level state ops.
 JSC::JSPromise* readableStreamCancel(JSC::JSGlobalObject*, JSReadableStream*, JSC::JSValue reason); // userJS: yes — ReadableStreamOperations.cpp
@@ -606,6 +606,8 @@ void pipeToReadRequestErrorSteps(JSC::JSGlobalObject*, JSStreamPipeToOperation*,
 void materializeNativeSource(JSC::JSGlobalObject*, JSReadableStream*); // userJS: yes — BunStreamSource.cpp
 // process.stdin pauses the file source it shares with Bun.stdin.stream(); the next reader makes it read again.
 void resumeNativeSource(JSReadableStream*); // userJS: no (a paused source reads at once; a pull that settles reacts in a microtask, as in nativeSourcePull) — BunStreamSource.cpp
+// The consumer released the lock: a stdio pipe, socket or terminal stops reading until resumeNativeSource and keeps its event-loop ref for that reader. Every other file source reads on and drops its ref.
+void releaseNativeSource(JSReadableStream*); // userJS: no (no read, and a parked pull stays parked) — BunStreamSource.cpp
 
 // The SourceKind::Native algorithm ARMS. The pull/cancel dispatch is a TOTAL
 // `switch (m_algorithms.kind)` in JSReadableStreamDefaultController.cpp (a Native source is

src/runtime/webcore/FileReader.rs

@@ -55,6 +55,8 @@ pub struct FileReader {
     /// Read-only after construction.
     pub(crate) highwater_mark: usize,
     pub(crate) flowing: Cell<bool>,
+    /// Reads fd 0, 1 or 2 as a pipe, socket or terminal. That descriptor has other readers (a child process, a synchronous read), so a reader release stops this source.
+    pub(crate) shared_stdio: Cell<bool>,
     /// Native sink attached by a hookup site (e.g. fetch request body). When
     /// set, `on_read_chunk` writes directly to it instead of the JS `pending`
     /// path; `pull_into_sink` is the drain-ack resume.
@@ -81,6 +83,7 @@ impl Default for FileReader {
             read_error: JsCell::new(None),
             highwater_mark: 16384,
             flowing: Cell::new(true),
+            shared_stdio: Cell::new(false),
             sink: JsCell::new(SinkHandle::None),
             sink_paused: Cell::new(false),
         }
@@ -98,6 +101,8 @@ pub(crate) struct OpenedFileBlob {
     pub(crate) fd: Fd,
     pub(crate) pollable: bool,
     pub(crate) nonblocking: bool,
+    /// The Blob names fd 0, 1 or 2. `fd` can be another descriptor: a terminal is read through one of its own.
+    pub(crate) is_stdio: bool,
     #[cfg(not(windows))]
     pub(crate) file_type: FileType,
 }
@@ -108,6 +113,7 @@ impl Default for OpenedFileBlob {
             fd: Fd::INVALID,
             pollable: false,
             nonblocking: true,
+            is_stdio: false,
             #[cfg(not(windows))]
             file_type: FileType::File,
         }
@@ -132,6 +138,7 @@ impl Lazy {
         let fd: Fd = match &file.pathlike {
             PathOrFileDescriptor::Fd(pl_fd) => {
                 if pl_fd.stdio_tag().is_some() {
+                    this.is_stdio = true;
                     'brk: {
                         #[cfg(unix)]
                         {
@@ -303,6 +310,7 @@ impl FileReader {
         self.reader().set_parent(self.as_ctx_ptr().cast());
         let was_lazy = !matches!(self.lazy.get(), Lazy::None);
         let mut pollable = false;
+        let mut is_stdio = false;
         #[cfg(unix)]
         let mut file_type = FileType::File;
         // R-2: move the `Lazy` out of the cell up-front (it's reset to `None`
@@ -327,6 +335,7 @@ impl FileReader {
                             debug_assert!(opened.fd.is_valid());
                             self.fd.set(opened.fd);
                             pollable = opened.pollable;
+                            is_stdio = opened.is_stdio;
                             #[cfg(unix)]
                             {
                                 file_type = opened.file_type;
@@ -404,6 +413,12 @@ impl FileReader {
                 }
                 return streams::Start::Err(e);
             }
+            // Only a pipe, socket or terminal reads with no consumer. On Windows only a pipe stops: a stop cancels a file read in flight, and libuv cancels a console line read with an Enter key of its own.
+            #[cfg(unix)]
+            self.shared_stdio.set(is_stdio && pollable);
+            #[cfg(windows)]
+            self.shared_stdio
+                .set(is_stdio && matches!(self.reader().source, Some(bun_io::Source::Pipe(_))));
         } else {
             #[cfg(unix)]
             {
@@ -741,8 +756,8 @@ impl FileReader {
         // SAFETY: see `parent()`.
         let _pin = unsafe { RefPtr::init_ref(self.parent()) };
         self.pending.with_mut(|p| p.run());
-        // Re-entrant cancel or a nested pull that read to EOF closed the reader; tell the io caller to stop so it does not re-read the captured fd.
-        ret && !self.done.get() && !self.reader().is_done()
+        // Re-entrant cancel or a nested pull that read to EOF closed the reader; tell the io caller to stop so it does not re-read the captured fd. The same for a consumer that stopped the source inside `run()`: the read loop would read once more.
+        ret && self.flowing.get() && !self.done.get() && !self.reader().is_done()
     }
 
     pub(crate) fn drain(&self) -> Vec<u8> {
@@ -901,6 +916,10 @@ impl FileReader {
         self.flowing.set(flag);
 
         if flag {
+            // With `buffered` at the highwater mark no pull is parked, and a read now would grow `buffered` once per pause and resume. The next `on_pull` unpauses and reads.
+            if self.buffered.get().len() >= self.highwater_mark {
+                return;
+            }
             self.reader().unpause();
             if !self.reader().is_done() && !self.reader().has_pending_read() {
                 // Kick off a new read if needed
@@ -922,6 +941,20 @@ pub(crate) extern "C" fn Bun__FileReader__resume(source: &Source) {
     source.context.set_flowing(true);
 }
 
+/// The reader of the stream was released. A shared stdio source stops, so input that arrives now stays on the descriptor. Stopped, it keeps nothing alive, and its ref is for the next reader. Every other source reads on and drops its ref.
+#[unsafe(no_mangle)]
+pub(crate) extern "C" fn Bun__FileReader__onReaderRelease(source: &Source) {
+    let this = &source.context;
+    if this.done.get() {
+        return;
+    }
+    if this.shared_stdio.get() {
+        this.set_flowing(false);
+    } else {
+        this.reader().update_ref(false);
+    }
+}
+
 // SAFETY: `FileReader` is always the `context` field of a heap-allocated
 // `Source`. `parent` is the `raw` arm for releasing a ref, which can free it.
 bun_core::impl_field_parent! { FileReader => Source.context; pub fn raw parent; pub fn shared parent_const; }

test/js/bun/console/console-iterator.test.ts

@@ -1,6 +1,7 @@
 import { spawn, spawnSync } from "bun";
 import { describe, expect, it } from "bun:test";
-import { bunEnv, bunExe } from "harness";
+import { bunEnv, bunExe, tempDir } from "harness";
+import { join } from "node:path";
 
 describe("should work for static input", () => {
   const inputs = [
@@ -74,3 +75,76 @@ it("can use the console iterator more than once", async () => {
   expect(await stdout.text()).toBe('["hello","world"]["another"]');
   proc.kill(0);
 });
+
+// The loop ends inside a chunk whose last line is not complete yet. The next loop has to keep that start of a line.
+it("a line that is split across two chunks survives a break", async () => {
+  await using proc = spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `
+      for await (const line of console) { console.log("first:" + line); break; }
+      for await (const line of console) console.log("next:" + line);
+      `,
+    ],
+    stdin: "pipe",
+    stdout: "pipe",
+    stderr: "pipe",
+    env: bunEnv,
+  });
+  const stderr = proc.stderr.text();
+  proc.stdin.write("one\ntw");
+  proc.stdin.flush();
+  const decoder = new TextDecoder();
+  let stdout = "";
+  let wroteRest = false;
+  for await (const chunk of proc.stdout) {
+    stdout += decoder.decode(chunk, { stream: true });
+    // The first loop took "one" and ended. The chunk it stopped in ends with "tw".
+    if (!wroteRest && stdout.includes("first:one\n")) {
+      wroteRest = true;
+      proc.stdin.write("o\nthree");
+      await proc.stdin.end();
+    }
+  }
+  expect({ stdout, stderr: await stderr, exitCode: await proc.exited }).toEqual({
+    stdout: "first:one\nnext:two\nnext:three\n",
+    stderr: "",
+    exitCode: 0,
+  });
+});
+
+// One loop per line, over a file that spans many chunks: most chunks end inside a line.
+it("one line per loop over a file keeps every line", async () => {
+  const lines = Array.from({ length: 2000 }, (_, i) => String(i).padStart(8, "0") + Buffer.alloc(391, "x").toString());
+  using dir = tempDir("console-iterator-lines", { "lines.txt": lines.join("\n") + "\n" });
+  await using proc = spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `
+      async function readLine() {
+        for await (const line of console) return line;
+      }
+      const hash = new Bun.CryptoHasher("sha1");
+      let count = 0;
+      for (let line; (line = await readLine()) !== undefined && line !== ""; count++) hash.update(line + "\\n");
+      console.log(JSON.stringify({ count, sha1: hash.digest("hex") }));
+      `,
+    ],
+    stdin: Bun.file(join(String(dir), "lines.txt")),
+    stdout: "pipe",
+    stderr: "pipe",
+    env: bunEnv,
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect({ stdout, stderr, exitCode }).toEqual({
+    stdout:
+      JSON.stringify({
+        count: lines.length,
+        sha1: new Bun.CryptoHasher("sha1").update(lines.join("\n") + "\n").digest("hex"),
+      }) + "\n",
+    stderr: "",
+    exitCode: 0,
+  });
+});