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

src/runtime/webcore/Blob.rs

@@ -118,6 +118,39 @@ const SERIALIZATION_VERSION: u8 = 4;
 
 pub use bun_jsc::generated::JSBlob as js;
 
+/// Claimed in `do_read_file`, released in `NewReadFileHandler::run`.
+static STDIN_BLOB_READ_IN_FLIGHT: core::sync::atomic::AtomicBool =
+    core::sync::atomic::AtomicBool::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.store(false, core::sync::atomic::Ordering::SeqCst);
+}
+
+fn locked_stdin_stream(
+    blob: &Blob,
+    this_value: JSValue,
+    global: &JSGlobalObject,
+) -> Option<JSValue> {
+    if !is_stdin_fd_store(blob) {
+        return None;
+    }
+    let stream = js::stream_get_cached(this_value)?;
+    ReadableStream::is_locked_value(stream, global).then_some(stream)
+}
+
 // ──────────────────────────────────────────────────────────────────────────
 
 // is_s3: defined once above (near is_bun_file); duplicate removed to fix E0034.
@@ -441,6 +474,17 @@ 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.swap(true, core::sync::atomic::Ordering::SeqCst)
+        {
+            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 +1262,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) = locked_stdin_stream(self, callframe.this(), global_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 +1276,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) = locked_stdin_stream(self, callframe.this(), global_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 +1298,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) = locked_stdin_stream(self, callframe.this(), global_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 +1316,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) = locked_stdin_stream(self, callframe.this(), global_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/ReadableStream.rs

@@ -266,6 +266,10 @@ impl ReadableStream {
         ReadableStream__isLocked(self.value, global_object)
     }
 
+    pub fn is_locked_value(value: JSValue, global_object: &JSGlobalObject) -> bool {
+        ReadableStream__isLocked(value, global_object)
+    }
+
     /// A pure `dynamicDowncast<JSReadableStream>` type test: no tagging, no conversion.
     pub fn is_readable_stream(value: JSValue) -> bool {
         ReadableStream__is(value)

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,99 @@
+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", "json"] 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() still works after process.stdin was referenced without reading",
+  async () => {
+    const { stdout, stderr, exitCode } = await run(`
+    void process.stdin.isTTY;
+    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);
+});