Bun.serve: share a route's payload with the Response's unread .body stream

src/runtime/server/FileRoute.rs

@@ -59,6 +59,20 @@ fn sp_slice<'a>(ptr: StringPointer, buf: &'a [u8]) -> &'a [u8] {
     &buf[ptr.offset as usize..][..ptr.length as usize]
 }
 
+/// `blob` is a file (`needs_to_read_file()`), so its `store` is Some and its `data` is `File`.
+fn refuse_file_descriptor(global: &JSGlobalObject, blob: &Blob) -> JsResult<()> {
+    let is_fd = matches!(
+        blob.store.get().as_ref().unwrap().data,
+        StoreData::File(ref f) if matches!(f.pathlike, PathOrFileDescriptor::Fd(_))
+    );
+    if is_fd {
+        return Err(global.throw_todo(
+            b"Support serving files from a file descriptor. Please pass a path instead.",
+        ));
+    }
+    Ok(())
+}
+
 /// What `FileRoute::serve` left for `on()` to do with the open fd.
 enum Serve {
     /// The response is finished; close the fd and complete.
@@ -138,43 +152,52 @@ impl FileRoute {
         // `status_code` all take `&self`.
         if let Some(response) = argument.as_class_ref::<Response>() {
             let body_value = response.get_body_value();
-            body_value.to_blob_if_possible();
-            let needs_read = matches!(body_value, BodyValue::Blob(b) if b.needs_to_read_file());
-            if needs_read {
-                // `needs_to_read_file()` ⇒ `store` is Some and `data` is `File`.
-                let is_fd = matches!(
-                    body_value,
-                    BodyValue::Blob(b)
-                        if matches!(
-                            b.store.get().as_ref().unwrap().data,
-                            StoreData::File(ref f)
-                                if matches!(f.pathlike, PathOrFileDescriptor::Fd(_))
-                        )
-                );
-                if is_fd {
-                    return Err(global.throw_todo(
-                        b"Support serving files from a file descriptor. Please pass a path instead.",
-                    ));
+            // Share a file behind an unread `.body` stream: a lift would spend the stream of a Response that stays unused.
+            let peeked = match body_value {
+                BodyValue::Locked(locked) => locked.peek_blob_of_unread_stream(),
+                _ => None,
+            };
+            let blob = 'blob: {
+                if let Some(blob) = peeked {
+                    if !blob.needs_to_read_file() {
+                        // In memory: `StaticRoute` shares it the same way.
+                        return Ok(None);
+                    }
+                    refuse_file_descriptor(global, &blob)?;
+                    // A second reader of a pipe would compete with the stream, so that is lifted.
+                    if !crate::webcore::blob::store_yields_bytes_once(
+                        blob.store.get().as_ref().unwrap(),
+                    ) {
+                        break 'blob blob;
+                    }
+                }
+                body_value.to_blob_if_possible();
+                let BodyValue::Blob(file) = &*body_value else {
+                    return Ok(None);
+                };
+                if !file.needs_to_read_file() {
+                    return Ok(None);
                 }
+                refuse_file_descriptor(global, file)?;
 
                 let blob = body_value.use_();
-
-                blob.global_this.set(std::ptr::from_ref(global));
                 debug_assert!(
                     !blob.is_heap_allocated(),
                     "expected blob not to be heap-allocated"
                 );
                 *body_value = BodyValue::Blob(blob.dupe());
-                let headers = headers_from(response.get_init_headers(), &blob);
-                let status_code = response.status_code();
-
-                return Ok(Some(RefPtr::new(FileRoute::new(
-                    blob,
-                    headers,
-                    None,
-                    status_code,
-                ))));
-            }
+                blob
+            };
+            blob.global_this.set(std::ptr::from_ref(global));
+            let headers = headers_from(response.get_init_headers(), &blob);
+            let status_code = response.status_code();
+
+            return Ok(Some(RefPtr::new(FileRoute::new(
+                blob,
+                headers,
+                None,
+                status_code,
+            ))));
         }
         if let Some(blob) = argument.as_class_ref::<Blob>() {
             if blob.needs_to_read_file() {

src/runtime/server/StaticRoute.rs

@@ -160,9 +160,21 @@ impl StaticRoute {
             // Let's let them do that.
             let body_value = response.get_body_value();
             let was_string = body_value.was_string();
-            body_value.to_blob_if_possible();
+            // Share a payload behind an unread `.body` stream: a lift would spend the stream of a Response that stays unused.
+            let shared = match body_value {
+                BodyValue::Locked(locked) => locked
+                    .peek_blob_of_unread_stream()
+                    .filter(|blob| !blob.needs_to_read_file()),
+                _ => None,
+            };
+            if shared.is_none() {
+                body_value.to_blob_if_possible();
+            }
 
             let blob: AnyBlob = 'brk: {
+                if let Some(blob) = shared {
+                    break 'brk AnyBlob::Blob(blob);
+                }
                 match body_value {
                     BodyValue::Used => {
                         return Err(global_this.throw_invalid_arguments(format_args!(

src/runtime/webcore/Body.rs

@@ -398,6 +398,14 @@ impl PendingValue {
         Some(blob)
     }
 
+    /// What [`Self::to_any_blob`] would lift, as a Blob of its own. Nothing is taken: the stream stays the body.
+    pub(crate) fn peek_blob_of_unread_stream(&self) -> Option<Blob> {
+        if self.promise.is_some() || self.on_receive_value.is_some() {
+            return None;
+        }
+        self.readable.get()?.peek_blob(self.global())
+    }
+
     fn set_promise(
         &mut self,
         global_this: &JSGlobalObject,

src/runtime/webcore/ByteBlobLoader.rs

@@ -171,6 +171,18 @@ impl ByteBlobLoader {
         Some(blob::Any::Blob(blob))
     }
 
+    /// The Blob this loader has left to read, sharing its store. The loader keeps the store.
+    pub(crate) fn peek_blob(&self, global: &JSGlobalObject) -> Option<Blob> {
+        let blob = Blob::init_with_store(self.store.clone()?, global);
+        blob.offset.set(self.offset);
+        blob.size.set(self.remain);
+        if !self.content_type.is_empty() {
+            blob.content_type.set(self.content_type.clone());
+            blob.content_type_was_set.set(true);
+        }
+        Some(blob)
+    }
+
     pub(crate) fn detach_store(&mut self) -> Option<RefPtr<Store>> {
         if let Some(store) = self.store.take() {
             self.done = true;

src/runtime/webcore/ByteStream.rs

@@ -108,6 +108,11 @@ impl Buffered {
         bytes
     }
 
+    /// A copy of the untaken bytes. The buffer keeps them.
+    fn to_vec(&self) -> Vec<u8> {
+        self.bytes.get()[self.consumed.get()..].to_vec()
+    }
+
     fn extend(&self, chunk: &[u8]) {
         self.bytes.with_mut(|b| b.extend_from_slice(chunk));
     }
@@ -983,6 +988,12 @@ impl ByteStream {
         None
     }
 
+    /// A copy of what [`Self::to_any_blob`] would take. The stream keeps its bytes.
+    pub(crate) fn peek_bytes(&self) -> Option<Vec<u8>> {
+        (self.has_received_last_chunk.get() && !self.has_pending_error())
+            .then(|| self.buffered.to_vec())
+    }
+
     fn to_buffered_value(
         &self,
         cx: &bun_jsc::JsThread<'_>,

src/runtime/webcore/ReadableStream.rs

@@ -206,25 +206,7 @@ impl ReadableStream {
                 let blobby = unsafe { &mut *blobby };
                 blobby.to_any_blob(global_this)?
             }
-            Source::File(_) => {
-                // BACKREF: see `Source::file()` — payload valid while stream alive.
-                // R-2: `lazy`/`started` are `JsCell`/`Cell`; shared borrow suffices.
-                let blobby = self.ptr.file().expect("matched File");
-                let webcore::file_reader::Lazy::Blob(store) = blobby.lazy.get() else {
-                    return None;
-                };
-                let blob = Blob::init_with_store(store.clone(), global_this);
-                // The window `from_blob_copy_ref` moved onto the reader.
-                if let Some(offset) = blobby.start_offset {
-                    blob.offset.set(offset as webcore::blob::SizeType);
-                }
-                if let Some(size) = blobby.max_size {
-                    blob.size.set(size as webcore::blob::SizeType);
-                }
-                // it should be lazy, file shouldn't have opened yet.
-                debug_assert!(!blobby.started.get());
-                webcore::blob::Any::Blob(blob)
-            }
+            Source::File(_) => webcore::blob::Any::Blob(self.file_blob(global_this)?),
             Source::Bytes(_) => {
                 // BACKREF: see `Source::bytes()` — payload valid while stream alive.
                 let bytes = self.ptr.bytes().expect("matched Bytes");
@@ -244,6 +226,48 @@ impl ReadableStream {
         Some(blob)
     }
 
+    /// What [`Self::to_any_blob`] would lift, as a Blob of its own. Nothing is taken: the stream stays readable.
+    pub fn peek_blob(&self, global_this: &JSGlobalObject) -> Option<Blob> {
+        if self.is_disturbed(global_this) || self.is_locked(global_this) {
+            return None;
+        }
+        let stream = ReadableStream::from_js_direct(self.value)?;
+        match stream.ptr {
+            Source::Blob(loader) => {
+                // SAFETY: ptr came from ReadableStreamTag__tagged; valid while stream alive. Read-only.
+                unsafe { &*loader }.peek_blob(global_this)
+            }
+            Source::File(_) => stream.file_blob(global_this),
+            Source::Bytes(_) => {
+                let bytes = stream.ptr.bytes()?;
+                Some(Blob::init(bytes.peek_bytes()?, global_this))
+            }
+            Source::JavaScript if ReadableStream__isClosedUnread(self.value, global_this) => {
+                Some(Blob::init_empty(global_this))
+            }
+            Source::JavaScript | Source::Invalid => None,
+        }
+    }
+
+    /// The file a lazy, not yet started file stream reads, with the stream's window.
+    fn file_blob(&self, global_this: &JSGlobalObject) -> Option<Blob> {
+        let reader = self.ptr.file()?;
+        let webcore::file_reader::Lazy::Blob(store) = reader.lazy.get() else {
+            return None;
+        };
+        let blob = Blob::init_with_store(store.clone(), global_this);
+        // The window `from_blob_copy_ref` moved onto the reader.
+        if let Some(offset) = reader.start_offset {
+            blob.offset.set(offset as webcore::blob::SizeType);
+        }
+        if let Some(size) = reader.max_size {
+            blob.size.set(size as webcore::blob::SizeType);
+        }
+        // it should be lazy, file shouldn't have opened yet.
+        debug_assert!(!reader.started.get());
+        Some(blob)
+    }
+
     pub fn done(&self) {
         // done is called when we are done consuming the stream
         // cancel actually mark the stream source as done