Honor Blob slice offsets on non-seekable sources

src/io/PipeReader.rs

@@ -146,6 +146,9 @@ pub struct PosixBufferedReader {
     pub handle: PollOrFd,
     pub _buffer: Vec<u8>,
     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::<T>(),
             flags: PosixFlags::new(),
             count: 0,
@@ -213,13 +217,15 @@ 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,
             maxbuf: None,
         };
         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<usize>,
+    ) -> 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<Source>,
     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<u8>,
     // 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)]
         {

src/runtime/webcore/blob/read_file.rs

@@ -183,6 +183,9 @@ pub struct ReadFile {
     pub byte_store: ByteStore,
     pub store: Option<StoreRef>,
     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();