FileReader: apply the slice window on the read_into pull path and end the stream when it is used up

src/runtime/webcore/FileReader.rs

@@ -49,8 +49,9 @@ pub struct FileReader {
     pub(crate) fd: Cell<Fd>,
     /// Read-only after construction (set via struct literal in `from_blob_*`).
     pub(crate) start_offset: Option<usize>,
-    /// Read-only after construction.
+    /// Length of the slice window at `start_offset`; the stream ends there. Read-only after init.
     pub(crate) max_size: Option<usize>,
+    /// Bytes delivered so far; never exceeds `max_size`.
     pub(crate) total_readed: Cell<usize>,
     pub(crate) started: Cell<bool>,
     pub(crate) waiting_for_on_reader_done: Cell<bool>,
@@ -632,21 +633,14 @@ impl FileReader {
             self.reader().close();
             return false;
         }
-        let mut close = false;
-        let mut has_more = state != ReadState::Eof;
-        if let (Some(max_size), false) = (self.max_size, chunk.is_empty()) {
-            let total_readed = self.total_readed.get();
-            if total_readed >= max_size {
-                return false;
-            }
-            let len = (max_size - total_readed).min(chunk.len());
-            chunk.truncate(len);
-            self.total_readed.set(total_readed + len);
-            if len == 0 {
-                close = true;
-                has_more = false;
+        let window_exhausted = match self.window_remaining() {
+            Some(remaining) => {
+                chunk.truncate(remaining);
+                self.consume_window(chunk.len())
             }
-        }
+            None => false,
+        };
+        let has_more = state != ReadState::Eof && !window_exhausted;
 
         let sink = *self.sink.get();
         let keep_going = if sink.is_some() {
@@ -674,12 +668,37 @@ impl FileReader {
             }
             keep_going
         };
-        if close {
-            self.reader().close();
+        if window_exhausted {
+            // Closed only now: closing first would settle a parked read with `Done` ahead of this chunk.
+            self.end_at_window();
+            return false;
         }
         keep_going
     }
 
+    /// Bytes the blob's slice window still allows; `None` when the source is unbounded.
+    fn window_remaining(&self) -> Option<usize> {
+        Some(self.max_size? - self.total_readed.get())
+    }
+
+    /// Charges `len` bytes to the window; `true` once it is used up, which is this stream's EOF.
+    fn consume_window(&self, len: usize) -> bool {
+        let Some(max_size) = self.max_size else {
+            return false;
+        };
+        let total_readed = self.total_readed.get() + len;
+        debug_assert!(total_readed <= max_size);
+        self.total_readed.set(total_readed);
+        total_readed == max_size
+    }
+
+    /// Ends the stream the way EOF does; `on_reader_done` settles whatever is waiting.
+    fn end_at_window(&self) {
+        if !self.reader().is_done() {
+            self.reader().close();
+        }
+    }
+
     fn write_chunk_to_sink(&self, sink: SinkHandle, chunk: &[u8], has_more: bool) -> bool {
         if !chunk.is_empty() {
             let chunk = bun_ptr::RawSlice::new(chunk);
@@ -828,9 +847,17 @@ impl FileReader {
         }
 
         if !self.reader().has_pending_read() && self.flowing.get() {
+            // `read_into` does not go through `on_read_chunk`, so the slice window is applied here: the read is cut to what is left of it (an empty destination reads nothing).
+            let len = self
+                .window_remaining()
+                .map_or(buffer.len(), |remaining| remaining.min(buffer.len()));
             // SAFETY: the reader cell is live for `self`'s lifetime; `read_into` is the raw re-entrancy-safe entry (EOF/error dispatch runs user JS).
-            let (amount_read, state) = unsafe { IOReader::read_into(self.reader.get(), buffer) };
-            bun_core::scoped_log!(FileReader, "onPull({}) = {}", buffer.len(), amount_read);
+            let (amount_read, state) =
+                unsafe { IOReader::read_into(self.reader.get(), &mut buffer[..len]) };
+            bun_core::scoped_log!(FileReader, "onPull({}) = {}", len, amount_read);
+            if self.consume_window(amount_read) {
+                self.end_at_window();
+            }
             let done = state == ReadState::Eof || self.reader().is_done();
             if amount_read > 0 {
                 let into = streams::IntoArray {

test/js/bun/util/bun-stdin-slice.test.ts

@@ -1,4 +1,4 @@
-import { expect, test } from "bun:test";
+import { describe, expect, test } from "bun:test";
 import { bunEnv, bunExe, isWindows } from "harness";
 
 // Reading a sliced non-regular file blob (like stdin from a pipe) with a size
@@ -40,3 +40,58 @@ test.skipIf(isWindows)("Bun.stdin.slice(0, N).text() caps reads at N bytes", asy
   expect(stdout).toBe("012");
   expect(exitCode).toBe(0);
 });
+
+// Streaming a slice of a pipe is the POSIX path where the chunk that ends the
+// slice is handed to a read that is already waiting (FileReader::on_read_chunk
+// with a parked read); a regular file is read straight into the pull buffer
+// instead. stdin is never closed here, so only the end of the slice can end the
+// stream and let the child exit.
+describe("Bun.stdin.slice(0, N).stream() over a pipe that stays open", () => {
+  function spawnSliceEcho(n: number) {
+    return Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `for await (const chunk of Bun.stdin.slice(0, ${n}).stream()) process.stdout.write(chunk);`,
+      ],
+      env: bunEnv,
+      stdin: "pipe",
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+  }
+
+  test.concurrent.skipIf(isWindows)("ends after N bytes of a larger write", async () => {
+    await using proc = spawnSliceEcho(3);
+    proc.stdin.write("0123456789");
+    await proc.stdin.flush();
+
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout, stderr, exitCode }).toEqual({ stdout: "012", stderr: "", exitCode: 0 });
+  });
+
+  test.concurrent.skipIf(isWindows)(
+    "delivers a chunk that leaves the slice open, then ends on the one that fills it",
+    async () => {
+      await using proc = spawnSliceEcho(4);
+      const reader = proc.stdout.getReader();
+      const decoder = new TextDecoder();
+      let stdout = "";
+
+      proc.stdin.write("01");
+      await proc.stdin.flush();
+      // Wait for the first chunk to come back out so the second write is a separate read in the child.
+      for (let r; !stdout.includes("01") && !(r = await reader.read()).done; ) {
+        stdout += decoder.decode(r.value, { stream: true });
+      }
+      expect(stdout).toBe("01");
+
+      proc.stdin.write("23456");
+      await proc.stdin.flush();
+      for (let r; !(r = await reader.read()).done; ) stdout += decoder.decode(r.value, { stream: true });
+
+      const [stderr, exitCode] = await Promise.all([proc.stderr.text(), proc.exited]);
+      expect({ stdout, stderr, exitCode }).toEqual({ stdout: "0123", stderr: "", exitCode: 0 });
+    },
+  );
+});

test/js/web/fetch/blob.test.ts

@@ -688,6 +688,73 @@ describe("file-backed slice bounds are respected when streaming and serving", ()
     // Serializing resolves the original's size, clamping the window to EOF.
     expect(s.size).toBe(5);
   });
+
+  // A sliced Bun.file() is streamed by FileReader, which hands bytes out two
+  // ways: a JS pull reads straight into the pull buffer, while a native sink
+  // (HTMLRewriter here; also pollable fds and Windows) is fed from the read
+  // loop's on_read_chunk. Both have to stop at the end of the slice while the
+  // file goes on past it, and a used-up slice has to end the stream, not hang it.
+  describe("a slice of a file that continues past it", () => {
+    const size = 1024 * 1024;
+    // 61-byte period: coprime with every chunk size involved, so bytes streamed
+    // from the wrong offset compare unequal, not just a wrong number of them.
+    const data = Buffer.alloc(size, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXY");
+    const windows: [start: number, end: number][] = [
+      [0, 5], // inside the first read
+      [3, 7],
+      [0, 256 * 1024], // exactly the first pull's buffer: the window ends on a read that fills it
+      [100, 700_000], // several pulls; the last one has to be cut short
+      [size - 10, size], // ends at EOF
+      [size - 10, size + 100], // the file ends first (the slice is taken before the size is known)
+      [4096, 4096], // nothing to deliver at all
+    ];
+
+    async function collect(stream: ReadableStream<Uint8Array>): Promise<Buffer> {
+      const chunks: Uint8Array[] = [];
+      for await (const chunk of stream) chunks.push(chunk);
+      return Buffer.concat(chunks);
+    }
+
+    // `subarray` clamps at EOF like the stream has to.
+    function expectWindow(delivered: Buffer, start: number, end: number) {
+      const expected = data.subarray(start, end);
+      expect(delivered.length).toBe(expected.length);
+      expect(delivered).toEqual(expected);
+    }
+
+    test.each(windows)("Bun.file(path).slice(%d, %d).stream()", async (start, end) => {
+      using dir = tempDir("blob-file-slice-stream", { "data.bin": data });
+      expectWindow(await collect(Bun.file(`${dir}/data.bin`).slice(start, end).stream()), start, end);
+    });
+
+    // Buffered consumers size their pulls from the slice and need the stream to
+    // close once it is delivered (#18192, #31675).
+    test.each(windows)("Bun.file(path).slice(%d, %d).stream().bytes()", async (start, end) => {
+      using dir = tempDir("blob-file-slice-bytes", { "data.bin": data });
+      const bytes = await Bun.file(`${dir}/data.bin`).slice(start, end).stream().bytes();
+      expectWindow(Buffer.from(bytes), start, end);
+    });
+
+    test.each(windows)("new Response(Bun.file(path).slice(%d, %d)).body", async (start, end) => {
+      using dir = tempDir("blob-file-slice-body", { "data.bin": data });
+      expectWindow(await collect(new Response(Bun.file(`${dir}/data.bin`).slice(start, end)).body!), start, end);
+    });
+
+    test.each(windows)("HTMLRewriter.transform(new Response(Bun.file(path).slice(%d, %d)))", async (start, end) => {
+      using dir = tempDir("blob-file-slice-rewriter", { "data.bin": data });
+      const response = new HTMLRewriter().transform(new Response(Bun.file(`${dir}/data.bin`).slice(start, end)));
+      expectWindow(Buffer.from(await response.arrayBuffer()), start, end);
+    });
+
+    // Reading .size gives the unsliced file's stream a window that ends exactly
+    // where the file does; ending the stream there must not drop the last chunk.
+    test("Bun.file(path) with a resolved size still streams the whole file", async () => {
+      using dir = tempDir("blob-file-resolved-size-stream", { "data.bin": data });
+      const file = Bun.file(`${dir}/data.bin`);
+      expect(file.size).toBe(size);
+      expectWindow(await collect(file.stream()), 0, size);
+    });
+  });
 });
 
 // Blob conversion accepts every ArrayBuffer-like type, both as a direct body