fetch: release the FetchTasklet through its raw pointer, not a &mut receiver

src/runtime/dispatch.rs

@@ -349,7 +349,7 @@ pub(crate) fn run_task(
 
         // ── fetch / S3 ───────────────────────────────────────────────────
         task_tag::FetchTasklet => {
-            cast!(FetchTasklet).on_progress_update()?;
+            FetchTasklet::on_progress_update(cast_ptr!(FetchTasklet))?;
         }
         task_tag::FetchTaskletDeinit => {
             // SAFETY: posted by `deref_from_thread` with the last ref.

src/runtime/webcore.rs

@@ -413,9 +413,10 @@ impl SinkHandle {
         match *self {
             SinkHandle::None => {}
             SinkHandle::ServerResponse(any) => any.end_chunk(err.as_ref()),
-            // SAFETY: live backref; ByteStream clears sink before free.
-            SinkHandle::FetchRequestBody(mut p) => unsafe { p.get_mut() }.end_from_stream(err),
-            // Raw-ptr dispatch: may re-borrow and free the sink (see its doc).
+            // Raw-ptr dispatch for both: the call may free the sink (see their docs).
+            SinkHandle::FetchRequestBody(p) => {
+                fetch::FetchRequestBodySink::end_from_stream(p.as_ptr(), err)
+            }
             SinkHandle::S3Upload(p) => streams::NetworkSink::end_from_stream(p.as_ptr(), err),
             SinkHandle::FileSink(p) => p.end_from_stream(err),
             SinkHandle::HTMLRewriter(p) => p.end_from_stream(err),

src/runtime/webcore/fetch/FetchRequestBodySink.rs

@@ -1,5 +1,5 @@
 use bun_collections::ByteVecExt;
-use bun_ptr::BackRef;
+use bun_ptr::{BackRef, ScopedRef};
 use bun_sys::Error as SysError;
 
 use crate::webcore::blob::SizeType as BlobSizeType;
@@ -199,17 +199,37 @@ impl FetchRequestBodySink {
         ))
     }
 
+    /// JS entry: only a JS-pump sink has a JS object, and its release belongs
+    /// to the pump promise, so there is never a ref to release here.
     pub fn end(&mut self, err: Option<SysError>) -> bun_sys::Result<()> {
-        self.end_from_stream(err.map(StreamError::Error));
+        let release = self.end_and_take_task(err.map(StreamError::Error));
+        debug_assert!(release.is_none());
         bun_sys::Result::Ok(())
     }
 
     /// Native-path terminator called from `SinkHandle::end`. Carries the full
     /// `StreamError` so a JS-valued upstream error (e.g. fetch reset) reaches
     /// `write_end_request(Some(js))` instead of being silently dropped to EOF.
-    pub fn end_from_stream(&mut self, err: Option<StreamError>) {
-        if self.ended {
+    /// Raw pointer like `NetworkSink::end_from_stream`: the release may free
+    /// the tasklet and, through `clear_sink`, this sink.
+    #[allow(clippy::not_unsafe_ptr_arg_deref)]
+    pub fn end_from_stream(this: *mut Self, err: Option<StreamError>) {
+        // SAFETY: `this` is the live sink behind the handle; the borrow is
+        // scoped to this call and nothing touches `*this` after it.
+        let Some((task, err_js)) = (unsafe { (*this).end_and_take_task(err) }) else {
             return;
+        };
+        FetchTasklet::write_end_request(task.as_ptr(), err_js);
+    }
+
+    /// Marks the sink ended; for a native source, hands back the tasklet ref
+    /// `start_request_stream` took and the error to end the request with.
+    fn end_and_take_task(
+        &mut self,
+        err: Option<StreamError>,
+    ) -> Option<(BackRef<FetchTasklet, bun_ptr::Mut>, Option<JSValue>)> {
+        if self.ended {
+            return None;
         }
         self.ended = true;
         if matches!(
@@ -220,22 +240,16 @@ impl FetchRequestBodySink {
             // field; detach (not cancel) so we don't re-enter the source while
             // it is still on the stack (FileReader.on_reader_error ref-leak).
             self.source.clear();
-            if let Some(mut task) = self.task.take() {
-                let err_js = err.map(|e| e.to_js(&task.global_this));
-                // SAFETY: the `+1` taken in `start_request_stream` keeps the
-                // tasklet live while `task` was `Some`; `write_end_request` is
-                // the balancing release and may free `*self` via `clear_sink`,
-                // so do not touch `self` afterwards.
-                unsafe { task.get_mut() }.write_end_request(err_js);
-            }
-            return;
+            let task = self.task.take()?;
+            let err_js = err.map(|e| e.to_js(&task.global_this));
+            return Some((task, err_js));
         }
-        // JS pump path: the assign_to_stream result handler is the single balancing release.
         let sys_err = match err {
             Some(StreamError::Error(e)) => Some(e),
             _ => None,
         };
         self.source.close(sys_err);
+        None
     }
 
     pub fn end_from_js(&mut self, _global_this: &JSGlobalObject) -> bun_sys::Result<JSValue> {
@@ -253,7 +267,9 @@ impl FetchRequestBodySink {
         if let Some(task) = task {
             // Balances the `ref_()` taken in `start_request_stream` when the
             // assign_to_stream-result handler never ran to release it.
-            FetchTasklet::deref(task.as_ptr());
+            // SAFETY: that handler clears `task` before releasing, so `task`
+            // being `Some` means the ref is still held and the tasklet live.
+            drop(unsafe { ScopedRef::<FetchTasklet>::adopt(task.as_ptr()) });
         }
     }