Bun.stdin: reject concurrent text/bytes/arrayBuffer/json reads instead of racing on fd 0

src/runtime/webcore/Blob.rs

@@ -118,6 +118,46 @@ const SERIALIZATION_VERSION: u8 = 4;
 
 pub use bun_jsc::generated::JSBlob as js;
 
+thread_local! {
+    /// Set while a `ReadFile` spawned by one of the `Bun.stdin` Blob read
+    /// helpers (`text()/json()/arrayBuffer()/bytes()`) is in flight. A second
+    /// concurrent call checks this and rejects with `ERR_INVALID_STATE`
+    /// instead of registering a second poll on fd 0 (which the kernel answers
+    /// with `EEXIST` from `epoll_ctl` after the loser has already consumed
+    /// bytes). Claimed in `do_read_file`; released in `NewReadFileHandler::run`.
+    static STDIN_BLOB_READ_IN_FLIGHT: Cell<bool> = const { Cell::new(false) };
+}
+
+pub(crate) fn is_stdin_fd_store(blob: &Blob) -> bool {
+    matches!(
+        blob.store.get().as_deref(),
+        Some(s) if matches!(
+            &s.data,
+            store::Data::File(f) if matches!(
+                f.pathlike,
+                PathOrFileDescriptor::Fd(fd) if matches!(fd.stdio_tag(), Some(bun_sys::Stdio::StdIn))
+            )
+        )
+    )
+}
+
+pub(crate) fn release_stdin_blob_read_claim() {
+    STDIN_BLOB_READ_IN_FLIGHT.set(false);
+}
+
+/// For an fd-backed Blob, return the `ReadableStream` cached on the JS wrapper
+/// by `get_stream_with_cache`, if one has been materialised.
+fn fd_cached_stream(blob: &Blob, this_value: JSValue) -> Option<JSValue> {
+    let store = blob.store.get().as_deref()?;
+    let store::Data::File(f) = &store.data else {
+        return None;
+    };
+    if !matches!(f.pathlike, PathOrFileDescriptor::Fd(_)) {
+        return None;
+    }
+    js::stream_get_cached(this_value)
+}
+
 // ──────────────────────────────────────────────────────────────────────────
 
 // is_s3: defined once above (near is_bun_file); duplicate removed to fix E0034.
@@ -441,6 +481,15 @@ impl BlobExt for Blob {
     fn do_read_file<F: read_file::ReadFileToJs>(&self, global: &JSGlobalObject) -> JSValue {
         debug!("doReadFile");
 
+        if is_stdin_fd_store(self) && STDIN_BLOB_READ_IN_FLIGHT.replace(true) {
+            return global
+                .err(
+                    jsc::ErrorCode::INVALID_STATE,
+                    format_args!("stdin is already being read by another Bun.stdin consumer"),
+                )
+                .reject();
+        }
+
         type Handler<'a, F> = read_file::NewReadFileHandler<'a, F>;
 
         // The callback may read context.content_type (e.g. to_form_data_with_bytes),
@@ -1218,7 +1267,12 @@ impl BlobExt for Blob {
 
         Ok(stream)
     }
-    fn get_text(&self, global_this: &JSGlobalObject, _: &CallFrame) -> JsResult<JSValue> {
+    fn get_text(&self, global_this: &JSGlobalObject, callframe: &CallFrame) -> JsResult<JSValue> {
+        if let Some(stream) = fd_cached_stream(self, callframe.this()) {
+            return bun_jsc::from_js_host_call(global_this, || {
+                global_this.readable_stream_to_text(stream)
+            });
+        }
         Ok(self.get_text_clone(global_this)?)
     }
 
@@ -1227,7 +1281,12 @@ impl BlobExt for Blob {
         JSPromise::wrap(global_object, |g| self.to_string(g, Lifetime::Clone))
     }
 
-    fn get_json(&self, global_this: &JSGlobalObject, _: &CallFrame) -> JsResult<JSValue> {
+    fn get_json(&self, global_this: &JSGlobalObject, callframe: &CallFrame) -> JsResult<JSValue> {
+        if let Some(stream) = fd_cached_stream(self, callframe.this()) {
+            return bun_jsc::from_js_host_call(global_this, || {
+                global_this.readable_stream_to_json(stream)
+            });
+        }
         Ok(self.get_json_share(global_this)?)
     }
 
@@ -1244,7 +1303,16 @@ impl BlobExt for Blob {
         JSPromise::wrap(global_this, |g| self.to_array_buffer(g, Lifetime::Clone))
     }
 
-    fn get_array_buffer(&self, global_this: &JSGlobalObject, _: &CallFrame) -> JsResult<JSValue> {
+    fn get_array_buffer(
+        &self,
+        global_this: &JSGlobalObject,
+        callframe: &CallFrame,
+    ) -> JsResult<JSValue> {
+        if let Some(stream) = fd_cached_stream(self, callframe.this()) {
+            return bun_jsc::from_js_host_call(global_this, || {
+                global_this.readable_stream_to_array_buffer(stream)
+            });
+        }
         Ok(self.get_array_buffer_clone(global_this)?)
     }
 
@@ -1253,7 +1321,12 @@ impl BlobExt for Blob {
         JSPromise::wrap(global_this, |g| self.to_uint8_array(g, Lifetime::Clone))
     }
 
-    fn get_bytes(&self, global_this: &JSGlobalObject, _: &CallFrame) -> JsResult<JSValue> {
+    fn get_bytes(&self, global_this: &JSGlobalObject, callframe: &CallFrame) -> JsResult<JSValue> {
+        if let Some(stream) = fd_cached_stream(self, callframe.this()) {
+            return bun_jsc::from_js_host_call(global_this, || {
+                global_this.readable_stream_to_bytes(stream)
+            });
+        }
         Ok(self.get_bytes_clone(global_this)?)
     }
 

src/runtime/webcore/blob/read_file.rs

@@ -101,6 +101,9 @@ impl<'a, F: ReadFileToJs> ReadFileCompletion for NewReadFileHandler<'a, F> {
         let blob = core::mem::take(&mut handler.context);
         let global_this = handler.global_this;
         drop(handler);
+        if crate::webcore::blob::is_stdin_fd_store(&blob) {
+            crate::webcore::blob::release_stdin_blob_read_claim();
+        }
         match maybe_bytes {
             ReadFileResultType::Result(result) => {
                 let bytes = result.buf;

test/js/bun/util/bun-stdin-concurrent-read.test.ts

@@ -0,0 +1,94 @@
+import { describe, expect, test } from "bun:test";
+import { bunEnv, bunExe } from "harness";
+
+// `Bun.stdin.text()/bytes()/arrayBuffer()/json()` read fd 0 via a `ReadFile`
+// task that owns its own io poll. A second concurrent `ReadFile` on the same
+// fd used to issue a second `EPOLL_CTL_ADD`, so one call rejected with the raw
+// `EEXIST: file already exists, epoll_ctl` after already having consumed bytes
+// (those bytes were dropped, so the other call resolved short). The helpers now
+// reject `ERR_INVALID_STATE` up front for a second concurrent consumer instead,
+// and route through the cached `ReadableStream` when `process.stdin` (or a
+// manual `Bun.stdin.stream().getReader()`) already holds it.
+
+const SIZE = 1024 * 1024;
+const payload = Buffer.alloc(SIZE, "abcdefghij");
+
+async function run(src: string) {
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "-e", src],
+    env: bunEnv,
+    stdin: "pipe",
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  proc.stdin.write(payload);
+  await proc.stdin.end();
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  return { stdout, stderr, exitCode };
+}
+
+test.concurrent("two concurrent Bun.stdin.text() calls: second rejects ERR_INVALID_STATE, first reads every byte", async () => {
+  const { stdout, stderr, exitCode } = await run(`
+    const wrap = p => p.then(
+      v => ({ state: "resolved", len: v.length }),
+      e => ({ state: "rejected", code: e?.code, name: e?.name }),
+    );
+    const [a, b] = await Promise.all([wrap(Bun.stdin.text()), wrap(Bun.stdin.text())]);
+    process.stdout.write(JSON.stringify({ a, b }));
+  `);
+  expect(stderr).toBe("");
+  expect(JSON.parse(stdout)).toEqual({
+    a: { state: "resolved", len: SIZE },
+    b: { state: "rejected", code: "ERR_INVALID_STATE", name: "Error" },
+  });
+  expect(exitCode).toBe(0);
+});
+
+describe.each(["text", "arrayBuffer", "bytes"] as const)("Bun.stdin.%s()", method => {
+  test.concurrent(`rejects when process.stdin holds the reader; process.stdin receives every byte`, async () => {
+    const { stdout, stderr, exitCode } = await run(`
+      let n = 0;
+      process.stdin.on("data", c => (n += c.length));
+      let blob = { state: "pending" };
+      Bun.stdin.${method}().then(
+        v => { blob = { state: "resolved" }; },
+        e => { blob = { state: "rejected", code: e?.code }; },
+      );
+      await new Promise(r => process.stdin.once("end", r));
+      await Promise.resolve();
+      process.stdout.write(JSON.stringify({ n, blob }));
+    `);
+    expect(stderr).toBe("");
+    expect(JSON.parse(stdout)).toEqual({
+      n: SIZE,
+      blob: { state: "rejected", code: "ERR_INVALID_STATE" },
+    });
+    expect(exitCode).toBe(0);
+  });
+});
+
+test.concurrent("sequential Bun.stdin.text() after the first read completes does not reject", async () => {
+  const { stdout, stderr, exitCode } = await run(`
+    const a = await Bun.stdin.text();
+    const b = await Bun.stdin.text().then(
+      v => ({ state: "resolved", len: v.length }),
+      e => ({ state: "rejected", code: e?.code }),
+    );
+    process.stdout.write(JSON.stringify({ aLen: a.length, b }));
+  `);
+  expect(stderr).toBe("");
+  expect(JSON.parse(stdout)).toEqual({
+    aLen: SIZE,
+    b: { state: "resolved", len: 0 },
+  });
+  expect(exitCode).toBe(0);
+});
+
+test.concurrent("Bun.stdin.text() with no other consumer reads every byte", async () => {
+  const { stdout, stderr, exitCode } = await run(
+    `process.stdout.write(String((await Bun.stdin.text()).length));`,
+  );
+  expect(stderr).toBe("");
+  expect(stdout).toBe(String(SIZE));
+  expect(exitCode).toBe(0);
+});