diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index 9fe49cbacee5..87c62675ff0e 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -146,6 +146,9 @@ pub struct PosixBufferedReader { pub handle: PollOrFd, pub _buffer: Vec, pub _offset: usize, + /// Bytes of the `start_file_offset` offset still to be read and discarded + /// because the source cannot `pread`. Always 0 for seekable sources. + pub _skip_remaining: usize, pub vtable: BufferedReaderVTable, pub flags: PosixFlags, pub count: usize, @@ -182,6 +185,7 @@ impl PosixBufferedReader { handle: PollOrFd::Closed, _buffer: Vec::new(), _offset: 0, + _skip_remaining: 0, vtable: BufferedReaderVTable::init::(), flags: PosixFlags::new(), count: 0, @@ -213,6 +217,7 @@ impl PosixBufferedReader { handle: mem::replace(&mut other.handle, PollOrFd::Closed), _buffer: mem::take(other.buffer()), _offset: other._offset, + _skip_remaining: other._skip_remaining, flags: other.flags, vtable: BufferedReaderVTable { kind, parent }, count: 0, @@ -220,6 +225,7 @@ impl PosixBufferedReader { }; other.flags.insert(PosixFlags::IS_DONE); other._offset = 0; + other._skip_remaining = 0; MaxBuf::transfer_to_pipereader(&mut other.maxbuf, &mut self.maxbuf); // Capture *mut Self before borrowing `handle` so the owner pointer // doesn't conflict with the field borrow. @@ -457,9 +463,71 @@ impl PosixBufferedReader { pub fn start_file_offset(&mut self, fd: Fd, poll: bool, offset: usize) -> sys::Result<()> { self._offset = offset; self.flags.insert(PosixFlags::USE_PREAD); + // `pread` only honors the offset on seekable sources. Pipes, sockets and + // ttys have to read those bytes out of the stream and discard them. + // `start` only ever inserts POLLABLE, so this is the file type `read` + // will go on to dispatch against. + if poll + || self + .flags + .intersects(PosixFlags::POLLABLE | PosixFlags::SOCKET) + { + self._skip_remaining = offset; + } self.start(fd, poll) } + /// Reads and discards the bytes that precede the requested start offset on a + /// source that cannot `pread`. Returns false when the caller must stop for + /// this tick: the poll was re-armed, or EOF/an error was already reported. + fn drain_skipped( + parent: &mut PosixBufferedReader, + fd: Fd, + sys_fn: &impl Fn(Fd, &mut [u8], usize) -> sys::Result, + ) -> bool { + while parent._skip_remaining > 0 { + // `sys_fn` blocks on a blocking pipe (and anywhere RWF_NOWAIT is + // unavailable), so readiness has to be confirmed before every read — + // the same discipline `read_blocking_pipe` keeps. + match bun_core::is_readable(fd) { + bun_core::Pollable::Ready | bun_core::Pollable::Hup => {} + bun_core::Pollable::NotReady => { + parent.register_poll(); + return false; + } + } + + // Per-loop scratch buffer; single-threaded event loop (see + // `EventLoopCtx::pipe_read_buffer_mut`). + let scratch = parent.vtable.event_loop().pipe_read_buffer_mut(); + let want = scratch.len().min(parent._skip_remaining); + match sys_fn(fd, &mut scratch[..want], parent._offset) { + sys::Result::Ok(0) => { + // EOF before the offset was reached: the slice is empty. + parent._skip_remaining = 0; + parent.close_without_reporting(); + if !parent.flags.contains(PosixFlags::IS_DONE) { + parent.done(); + } + return false; + } + sys::Result::Ok(bytes_read) => { + parent._skip_remaining -= bytes_read; + } + sys::Result::Err(err) => { + // Both arms are tail calls: either may free `parent`. + if err.is_retry() { + parent.register_poll(); + } else { + parent.on_error(err); + } + return false; + } + } + } + true + } + // Exists for consistently with Windows. pub fn has_pending_read(&self) -> bool { matches!(&self.handle, PollOrFd::Poll(poll) if poll.is_registered()) @@ -657,6 +725,22 @@ impl PosixBufferedReader { // touches `parent` after one can have dispatched it. let parent = unsafe { &mut *this }; let mut received_hup = received_hup_initially; + if parent._skip_remaining > 0 { + if !Self::drain_skipped(parent, fd, &|fd, buf, _| sys::read_nonblocking(fd, buf)) { + return; + } + // This function is only entered once its caller has confirmed the fd + // is readable, which is what lets the first read below be a blocking + // one. The drain just spent that readiness, so confirm it again. + match bun_core::is_readable(fd) { + bun_core::Pollable::Ready => {} + bun_core::Pollable::Hup => received_hup = true, + bun_core::Pollable::NotReady => { + parent.register_poll(); + return; + } + } + } loop { let streaming = parent.vtable.is_streaming_enabled(); let mut got_retry = false; @@ -856,6 +940,9 @@ impl PosixBufferedReader { // contract. `on_reader_error` MAY free it, so nothing below touches // `parent` after dispatching an error (see the EAGAIN arm). let parent = unsafe { &mut *this }; + if parent._skip_remaining > 0 && !Self::drain_skipped(parent, fd, &sys_fn) { + return; + } let streaming = parent.vtable.is_streaming_enabled(); if streaming { @@ -1123,6 +1210,9 @@ pub struct WindowsBufferedReader { /// It cannot change because we don't know what libuv will do with it. pub source: Option, pub _offset: usize, + /// Bytes of the `start_file_offset` offset still to be read and discarded + /// because the source is not seekable. Always 0 for file sources. + pub _skip_remaining: usize, pub _buffer: Vec, // for compatibility with Linux pub flags: WindowsFlags, @@ -1166,6 +1256,7 @@ impl WindowsBufferedReader { WindowsBufferedReader { source: None, _offset: 0, + _skip_remaining: 0, _buffer: Vec::new(), flags: WindowsFlags::new(), maxbuf: None, @@ -1189,10 +1280,12 @@ impl WindowsBufferedReader { self.flags = other.flags; self._buffer = mem::take(other.buffer()); self._offset = other._offset; + self._skip_remaining = other._skip_remaining; self.source = other.source.take(); other.flags.insert(WindowsFlags::IS_DONE); other._offset = 0; + other._skip_remaining = 0; // other._buffer / other.source already cleared by mem::take above. // The field-by-field assigns above leave `self.maxbuf` untouched, so // drop any prior owner-count first to avoid leaking a MaxBuf ref when @@ -1393,7 +1486,13 @@ impl WindowsBufferedReader { pub fn start_file_offset(&mut self, fd: Fd, poll: bool, offset: usize) -> sys::Result<()> { self._offset = offset; self.flags.insert(WindowsFlags::USE_PREAD); - self.start(fd, poll) + self.start(fd, poll)?; + // libuv only honors the offset on file sources. Pipes and ttys have to + // read those bytes out of the stream and discard them. + if !matches!(self.source, Some(Source::File(_))) { + self._skip_remaining = offset; + } + sys::Result::Ok(()) } pub fn set_raw_mode(&mut self, value: bool) -> sys::Result<()> { @@ -1878,10 +1977,27 @@ impl WindowsBufferedReader { self.on_error(err); return; } - let amount_result = match amount { + let mut amount_result = match amount { sys::Result::Ok(n) => n, sys::Result::Err(_) => unreachable!(), }; + let mut slice = slice; + + // A source that cannot seek ignored the offset libuv was handed, so the + // bytes before the requested start arrive here, at the head of the + // chunk. Shift the rest of it down over them; nothing past this point + // ever sees them because they are still in uncommitted spare capacity. + if self._skip_remaining > 0 && amount_result > 0 { + let dropped = self._skip_remaining.min(amount_result); + self._skip_remaining -= dropped; + slice.copy_within(dropped..amount_result, 0); + amount_result -= dropped; + slice = &mut core::mem::take(&mut slice)[..amount_result]; + if amount_result == 0 && has_more != ReadState::Eof { + self.flags.remove(WindowsFlags::HAS_INFLIGHT_READ); + return; + } + } #[cfg(debug_assertions)] { diff --git a/src/runtime/webcore/blob/read_file.rs b/src/runtime/webcore/blob/read_file.rs index a79cdc061347..3c27a42cdfc0 100644 --- a/src/runtime/webcore/blob/read_file.rs +++ b/src/runtime/webcore/blob/read_file.rs @@ -183,6 +183,9 @@ pub struct ReadFile { pub byte_store: ByteStore, pub store: Option, pub offset: SizeType, + /// Bytes of `offset` still to be read and discarded because the fd could + /// not be seeked. Always 0 once `lseek` applied `offset`. + pub skip_remaining: SizeType, pub max_length: SizeType, pub total_size: SizeType, pub opened_fd: Fd, @@ -348,6 +351,7 @@ impl ReadFile { byte_store: ByteStore::default(), store: Some(store), offset: off, + skip_remaining: 0, max_length: max_len, total_size: MAX_SIZE, opened_fd: Fd::INVALID, @@ -715,8 +719,11 @@ impl ReadFile { if self.offset > 0 { // We DO support offset in Bun.file() - // we ignore errors because it should continue to work even if its a pipe - let _ = bun_sys::set_file_offset(fd, self.offset); + // Pipes, sockets and ttys cannot seek, so the offset has to be read + // out of the stream and discarded instead. + if bun_sys::set_file_offset(fd, self.offset).is_err() { + self.skip_remaining = self.offset; + } } } @@ -815,13 +822,28 @@ impl ReadFile { let (buf_ptr, buf_len) = self.remaining_buffer(&mut stack_buffer); if buf_len > 0 && self.errno.is_none() && !self.read_eof { + // An unseekable fd still owes us `skip_remaining` bytes before + // the blob's own bytes start. Read them into the stack buffer + // and throw them away. + let skipping = self.skip_remaining > 0; + let (buf_ptr, buf_len) = if skipping { + ( + stack_ptr, + stack_buffer.len().min(self.skip_remaining as usize), + ) + } else { + (buf_ptr, buf_len) + }; + let mut read_amount: usize = 0; let mut retry = false; let continue_reading = self.do_read((buf_ptr, buf_len), &mut read_amount, &mut retry); // We might read into the stack buffer, so we need to copy it into the heap. - if buf_ptr == stack_ptr { + if skipping { + self.skip_remaining -= read_amount as SizeType; + } else if buf_ptr == stack_ptr { // `do_read` wrote `read_amount` initialized bytes at // `stack_buffer[..read_amount]`; the stack array is live // for this iteration. @@ -923,6 +945,9 @@ pub struct ReadFileUV<'a> { pub byte_store: ByteStore, pub store: StoreRef, pub offset: SizeType, + /// Bytes of `offset` still to be read and discarded because the handle is + /// not seekable. Always 0 when libuv can apply `offset` itself. + pub skip_remaining: SizeType, pub max_length: SizeType, pub total_size: SizeType, pub opened_fd: Fd, @@ -1069,6 +1094,7 @@ impl<'a> ReadFileUV<'a> { byte_store: ByteStore::default(), store, // store.ref() — Arc clone owned here offset: off, + skip_remaining: 0, max_length: max_len, total_size: MAX_SIZE, opened_fd: Fd::INVALID, @@ -1254,6 +1280,11 @@ impl<'a> ReadFileUV<'a> { // we ignore errors because it should continue to work even if its a pipe Err(_) | Ok(_) => {} } + // `uv_fs_read` only honors the offset on seekable handles; pipes and + // ttys ignore it, so the offset has to be consumed from the stream. + if !this.is_regular_file { + this.skip_remaining = this.offset; + } } // Special files might report a size of > 0, and be wrong. @@ -1301,7 +1332,13 @@ impl<'a> ReadFileUV<'a> { // libuv writes into spare capacity before any read; callers only need // ptr/len, so expose the spare slice directly instead of materialising // a `&mut [u8]` over uninitialized bytes. - let limit = (self.max_length.saturating_sub(self.read_off)) as usize; + // The bytes before the blob's start still have to come out of the + // stream, so while skipping they count toward what we ask libuv for — + // otherwise a tight `max_length` would shrink every skip read. + let limit = self + .max_length + .saturating_sub(self.read_off) + .saturating_add(self.skip_remaining) as usize; let spare = self.buffer.spare_capacity_mut(); let take = spare.len().min(limit); &mut spare[..take] @@ -1409,12 +1446,18 @@ impl<'a> ReadFileUV<'a> { return; } - this.read_off += SizeType::try_from(result.int()).expect("int cast"); + let read_amount = usize::try_from(result.int()).expect("int cast"); // SAFETY: libuv wrote result.int() bytes into remaining_buffer()'s spare slice. - unsafe { - this.buffer - .uv_commit(usize::try_from(result.int()).expect("int cast")) - }; + unsafe { this.buffer.uv_commit(read_amount) }; + + // Until `skip_remaining` is paid off the buffer holds nothing but the + // bytes before the blob's start, so they sit at its head. + let dropped = (this.skip_remaining as usize).min(read_amount); + if dropped > 0 { + this.buffer.drain(..dropped); + this.skip_remaining -= dropped as SizeType; + } + this.read_off += (read_amount - dropped) as SizeType; this.req.deinit(); this.queue_read(); diff --git a/test/js/bun/util/bun-stdin-slice.test.ts b/test/js/bun/util/bun-stdin-slice.test.ts index 8fad44a3d093..e196d5b4ef5e 100644 --- a/test/js/bun/util/bun-stdin-slice.test.ts +++ b/test/js/bun/util/bun-stdin-slice.test.ts @@ -1,5 +1,22 @@ import { expect, test } from "bun:test"; -import { bunEnv, bunExe, isWindows } from "harness"; +import { bunEnv, bunExe, isWindows, tempDir } from "harness"; +import { join } from "node:path"; + +async function runWithPipedStdin(script: string, input: string | Uint8Array) { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + }); + + proc.stdin.write(input); + await proc.stdin.end(); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { stdout, stderr, exitCode }; +} // Reading a sliced non-regular file blob (like stdin from a pipe) with a size // close to Blob.max_size used to overflow when computing the initial read @@ -19,7 +36,7 @@ test.skipIf(isWindows)("Bun.stdin.slice(1).text() does not crash when stdin is a const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); - expect(stdout).toBe("hello world"); + expect(stdout).toBe("ello world"); expect(exitCode).toBe(0); }); @@ -40,3 +57,134 @@ test.skipIf(isWindows)("Bun.stdin.slice(0, N).text() caps reads at N bytes", asy expect(stdout).toBe("012"); expect(exitCode).toBe(0); }); + +// A piped stdin cannot be seeked, so the start offset has to be consumed from +// the stream. It used to be dropped entirely, leaving `end` to act as a length. +test.concurrent("Bun.stdin.slice(start, end).text() honors start when stdin is a pipe", async () => { + const { stdout, exitCode } = await runWithPipedStdin( + `process.stdout.write(await Bun.stdin.slice(2, 7).text());`, + "abcdefghij", + ); + + expect(stdout).toBe("cdefg"); + expect(exitCode).toBe(0); +}); + +test.concurrent("Bun.stdin.slice(start).text() honors start when stdin is a pipe", async () => { + const { stdout, exitCode } = await runWithPipedStdin( + `process.stdout.write(await Bun.stdin.slice(3).text());`, + "abcdefghij", + ); + + expect(stdout).toBe("defghij"); + expect(exitCode).toBe(0); +}); + +test.concurrent("Bun.stdin.slice(start, end).bytes() honors start when stdin is a pipe", async () => { + const { stdout, exitCode } = await runWithPipedStdin( + `process.stdout.write(await Bun.stdin.slice(2, 7).bytes());`, + "abcdefghij", + ); + + expect(stdout).toBe("cdefg"); + expect(exitCode).toBe(0); +}); + +test.concurrent("Bun.stdin.slice(start, end).stream() honors start when stdin is a pipe", async () => { + const { stdout, exitCode } = await runWithPipedStdin( + `for await (const chunk of Bun.stdin.slice(2, 7).stream()) process.stdout.write(chunk);`, + "abcdefghij", + ); + + expect(stdout).toBe("cdefg"); + expect(exitCode).toBe(0); +}); + +test.concurrent("new Response(Bun.stdin.slice(start, end)) honors start when stdin is a pipe", async () => { + const { stdout, exitCode } = await runWithPipedStdin( + `process.stdout.write(await new Response(Bun.stdin.slice(2, 7)).text());`, + "abcdefghij", + ); + + expect(stdout).toBe("cdefg"); + expect(exitCode).toBe(0); +}); + +// The offset is larger than a single read, so it has to survive across reads +// (and across poll re-arms) before any byte is kept. +test.concurrent("Bun.stdin.slice(start) skips offsets that span many reads", async () => { + const input = Buffer.concat([Buffer.alloc(199_995, 0x61), Buffer.from("ZZZZZ")]); + const { stdout, exitCode } = await runWithPipedStdin( + `process.stdout.write(await Bun.stdin.slice(199995).text());`, + input, + ); + + expect(stdout).toBe("ZZZZZ"); + expect(exitCode).toBe(0); +}); + +test.concurrent("Bun.stdin.slice(start) past the end of a piped stdin is empty", async () => { + const { stdout, exitCode } = await runWithPipedStdin( + `const text = await Bun.stdin.slice(100).text(); process.stdout.write(JSON.stringify(text));`, + "abc", + ); + + expect(stdout).toBe(`""`); + expect(exitCode).toBe(0); +}); + +// Draining the offset empties the pipe, so it has to hand control back to the +// event loop and re-arm the poll instead of reading an empty pipe. A writer that +// trickles bytes makes the reader re-arm once per byte of the offset, and then +// again on the byte that lands exactly on the boundary. +test.concurrent("Bun.stdin.slice(start, end).stream() handles a writer that trickles bytes", async () => { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", `for await (const chunk of Bun.stdin.slice(5, 10).stream()) process.stdout.write(chunk);`], + env: bunEnv, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + }); + + for (const byte of "abcdefghij") { + proc.stdin.write(byte); + await proc.stdin.flush(); + await new Promise(resolve => setImmediate(resolve)); + } + await proc.stdin.end(); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect({ stdout, exitCode }).toEqual({ stdout: "fghij", exitCode: 0 }); + expect(stderr).not.toContain("error"); +}); + +// The writer hangs up with the offset exactly consumed and nothing left over. +test.concurrent("Bun.stdin.slice(start, end).stream() handles a writer that closes at the offset", async () => { + const { stdout, exitCode } = await runWithPipedStdin( + `let n = 0; for await (const chunk of Bun.stdin.slice(5, 10).stream()) n += chunk.length; process.stdout.write(String(n));`, + "abcde", + ); + + expect(stdout).toBe("0"); + expect(exitCode).toBe(0); +}); + +// A regular file stdin is seekable, so lseek applies the offset. Guards against +// the offset being consumed twice. +test.concurrent("Bun.stdin.slice(start, end) honors start when stdin is a regular file", async () => { + using dir = tempDir("stdin-slice-file", { "input.txt": "abcdefghij" }); + + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", `process.stdout.write(await Bun.stdin.slice(2, 7).text());`], + env: bunEnv, + stdin: Bun.file(join(String(dir), "input.txt")), + stdout: "pipe", + stderr: "pipe", + }); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect({ stdout, exitCode }).toEqual({ stdout: "cdefg", exitCode: 0 }); + expect(stderr).not.toContain("error"); +});