fetch: encode FetchTasklet's cross-thread ownership in the type system

src/http/HTTPThread.rs

@@ -1067,7 +1067,16 @@ impl HttpThread {
                 if let Some(ctx) = client.custom_ssl_ctx.take() {
                     ctx.deref();
                 }
-                drop(core::mem::take(&mut client.state));
+                let mut state = core::mem::take(&mut client.state);
+                // A streaming request body holds the HTTP-side ref on the
+                // `ThreadSafeStreamBuffer` (`Stream` has no `Drop` — the body
+                // is bitwise-shared with the JS-thread original). Normal
+                // teardown releases it in `InternalState::reset`; mirror that
+                // here or the buffer outlives both of its owners' releases
+                // and LSan reports it at exit. Idempotent: `detach` takes the
+                // `Option`.
+                state.original_request_body.deinit();
+                drop(state);
                 if let Some(f) = release.release_at_shutdown {
                     f(release.ctx);
                 }

src/jsc/Weak.rs

@@ -115,14 +115,36 @@ impl<T> Weak<T> {
         global_this: &JSGlobalObject,
         ref_type: WeakRefType,
         ctx: &mut T,
+    ) -> Self {
+        // SAFETY: `ctx` is derived from a live `&mut T`, so it satisfies the
+        // liveness contract at creation; the caller's `&mut T` is the same
+        // proof `create` has always demanded.
+        unsafe { Self::create_ptr(value, global_this, ref_type, NonNull::from(ctx)) }
+    }
+
+    /// Like [`create`](Self::create), but takes `ctx` as a raw pointer for
+    /// callers that cannot form `&mut T` (e.g. while disjoint field borrows of
+    /// the owner are live).
+    ///
+    /// # Safety
+    ///
+    /// `ctx` must point to a live `T` and remain valid for as long as the
+    /// weak ref's finalize callback can fire — the GC finalizer dereferences
+    /// it. (`create` enforces this via `&mut T`; here the caller must prove
+    /// liveness instead.)
+    pub unsafe fn create_ptr(
+        value: JSValue,
+        global_this: &JSGlobalObject,
+        ref_type: WeakRefType,
+        ctx: NonNull<T>,
     ) -> Self {
         if !value.is_empty() {
             return Self {
                 r#ref: Some(WeakImpl::init(
                     global_this,
                     value,
                     ref_type,
-                    Some(NonNull::from(ctx).cast::<c_void>()),
+                    Some(ctx.cast::<c_void>()),
                 )),
                 global_this: Some(global_this.into()),
                 _ctx: PhantomData,

src/runtime/dispatch.rs

@@ -1184,16 +1184,30 @@
     use bun_event_loop::task_tag;
     match task.tag {
         // `callback` (HTTP thread) won the `has_schedule_callback` CAS and
-        // posted this entry, then deref'd its own +1 if final; the JS-side
-        // +1 it expected `on_progress_update` to drop is the one we release
-        // here. Runs on the JS thread, so the plain `deref` (→ `deinit` on
-        // 1→0) is the right teardown path; the HTTP daemon is already
-        // parked (`shutdown_for_exit` precedes `destroy`), so the
-        // `Box<AsyncHTTP>` and any `metadata` it owns are exclusively ours.
+        // posted this entry, then released its own ref if final; the JS-side
+        // ref it expected `on_progress_update` to drop is the one we release
+        // here. Runs on the JS thread, so plain derefs (→ `deinit` on 1→0)
+        // are the right teardown path; the HTTP daemon is already parked
+        // (`shutdown_for_exit` precedes `destroy`), so the `Box<AsyncHTTP>`
+        // and any `metadata` it owns are exclusively ours.
         task_tag::FetchTasklet => {
+            let tasklet = task.ptr.cast::<FetchTasklet>();
+            // If a *final* `callback` landed in the exit window while this
+            // node was queued (lost the `has_schedule_callback` CAS), the
+            // entry left `in_flight` with only its HTTP-side ref released —
+            // neither `release_at_shutdown` nor `callback`'s shutdown branch
+            // balanced a streaming upload's sink/drain refs. Take them here;
+            // the take is a no-op when another exit path already claimed
+            // them (or there was no streaming upload).
+            for _ in 0..FetchTasklet::take_streaming_refs_for_exit(tasklet) {
+                // SAFETY: `tasklet` is the live heap `FetchTasklet`; the taken
+                // markers prove the refs are still held, and the HTTP daemon
+                // is parked so we release them exclusively.
+                FetchTasklet::deref(tasklet);
+            }
             // SAFETY: `task.ptr` is the live heap `FetchTasklet`; HTTP daemon is
             // already parked so we hold the sole reference.
-            FetchTasklet::deref(task.ptr.cast::<FetchTasklet>());
+            FetchTasklet::release_js_ref(tasklet);
             true
         }
         // `AsyncFSTask`s are `Box::leak`'d in `create()` and freed by