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,10 @@ 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 blob's slice window, counted from `start_offset`; the
+    /// stream ends there even if the file goes on. Read-only after construction.
     pub(crate) max_size: Option<usize>,
+    /// Bytes delivered so far, charged against `max_size`; never exceeds it.
     pub(crate) total_readed: Cell<usize>,
     pub(crate) started: Cell<bool>,
     pub(crate) waiting_for_on_reader_done: Cell<bool>,
@@ -632,21 +634,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 +669,40 @@ 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` delivered bytes against the slice window. Returns `true` once the window is
+    /// used up: that is this stream's EOF whatever the file still holds, so the caller must
+    /// `end_at_window` after delivering the bytes.
+    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
+    }
+
+    /// Closes the reader as EOF would have; `on_reader_done` then ends the sink or settles a
+    /// parked read, and the next `on_pull` reports `Done`.
+    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 +851,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/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],
+      [100, 700_000], // several pulls; the last one has to be cut short
+      [size - 10, size], // ends at EOF
+      [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);
+    }
+
+    test.each(windows)("Bun.file(path).slice(%d, %d).stream()", async (start, end) => {
+      using dir = tempDir("blob-file-slice-stream", { "data.bin": data });
+      const streamed = await collect(Bun.file(`${dir}/data.bin`).slice(start, end).stream());
+      expect(streamed.length).toBe(end - start);
+      expect(streamed).toEqual(data.subarray(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 = Buffer.from(await Bun.file(`${dir}/data.bin`).slice(start, end).stream().bytes());
+      expect(bytes.length).toBe(end - start);
+      expect(bytes).toEqual(data.subarray(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 });
+      const streamed = await collect(new Response(Bun.file(`${dir}/data.bin`).slice(start, end)).body!);
+      expect(streamed.length).toBe(end - start);
+      expect(streamed).toEqual(data.subarray(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)));
+      const rewritten = Buffer.from(await response.arrayBuffer());
+      expect(rewritten.length).toBe(end - start);
+      expect(rewritten).toEqual(data.subarray(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);
+      const streamed = await collect(file.stream());
+      expect(streamed.length).toBe(size);
+      expect(streamed).toEqual(data);
+    });
+  });
 });
 
 // Blob conversion accepts every ArrayBuffer-like type, both as a direct body