diff --git a/src/runtime/webcore/FileReader.rs b/src/runtime/webcore/FileReader.rs index 6af15b077aa0..723b23e85f20 100644 --- a/src/runtime/webcore/FileReader.rs +++ b/src/runtime/webcore/FileReader.rs @@ -49,8 +49,9 @@ pub struct FileReader { pub(crate) fd: Cell, /// Read-only after construction (set via struct literal in `from_blob_*`). pub(crate) start_offset: Option, - /// 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, + /// Bytes delivered so far; never exceeds `max_size`. pub(crate) total_readed: Cell, pub(crate) started: Cell, pub(crate) waiting_for_on_reader_done: Cell, @@ -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 { + 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 { diff --git a/test/js/bun/util/bun-stdin-slice.test.ts b/test/js/bun/util/bun-stdin-slice.test.ts index 8fad44a3d093..6794188257a4 100644 --- a/test/js/bun/util/bun-stdin-slice.test.ts +++ b/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 }); + }, + ); +}); diff --git a/test/js/web/fetch/blob.test.ts b/test/js/web/fetch/blob.test.ts index fc9b4500bcfc..d3ec3fb404c3 100644 --- a/test/js/web/fetch/blob.test.ts +++ b/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): Promise { + 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