s3: free the NetworkSink behind writer() via intrusive refcount

src/codegen/generate-jssink.ts

@@ -285,6 +285,7 @@ using namespace JSC;
 
 ${classes.map(name => `extern "C" size_t ${name}__memoryCost(void* sinkPtr);`).join("\n")}
 ${classes.map(name => `extern "C" void ${name}__controllerDetached(void* sinkPtr, JSC::EncodedJSValue controllerValue);`).join("\n")}
+${classes.map(name => `extern "C" void ${name}__wrapperDetached(void* sinkPtr);`).join("\n")}
 `;
   var templ = head;
 
@@ -513,6 +514,9 @@ JSC_DEFINE_HOST_FUNCTION(${name}__doClose, (JSC::JSGlobalObject * lexicalGlobalO
     sink->detach();
     RETURN_IF_EXCEPTION(scope, {});
     ${name}__close(lexicalGlobalObject, ptr);
+    // detach() nulled m_sinkPtr so ~${className} will not reach __finalize;
+    // release the wrapper's ref via a non-GC path.
+    ${name}__wrapperDetached(ptr);
     return JSC::JSValue::encode(JSC::jsUndefined());
 }
 
@@ -1115,6 +1119,16 @@ pub extern "C" fn ${name}__finalize(this: &mut ${name}) {
     ${JSSinkT}::js_finalize(this)
 }
 
+`;
+
+    // extern "C" void ${name}__wrapperDetached(void* sinkPtr) — called from
+    // ${name}__doClose after it nulls m_sinkPtr. C++ caller null-checks `ptr`.
+    symbols.push(`${name}__wrapperDetached`);
+    templ += `#[unsafe(no_mangle)]
+pub extern "C" fn ${name}__wrapperDetached(this: &mut ${name}) {
+    ${JSSinkT}::js_wrapper_detached(this)
+}
+
 `;
 
     // extern "C" void ${name}__controllerDetached(void* sinkPtr, JSC::EncodedJSValue)

src/runtime/webcore/FileSink.rs

@@ -924,19 +924,18 @@ impl FileSink {
             }
         }
 
-        // Per-wrapper accounting is on `ref_count` directly: each path that
-        // hands `self` to C++ (`to_js` / `to_js_with_destructor`) takes a +1
-        // via `self.ref_()`, and `finalize`'s `deref()` below releases it.
-        // `JsSinkType::construct` allocates with `ref_count=1` and that +1
-        // belongs to the wrapper it's about to be stored in, so no extra
-        // `ref_()` there. Callers that allocate via `init`/`create` and then
-        // `to_js()` must `deref()` once to release init's +1 (see
-        // `Blob::get_writer`).
-        self.readable_stream.set(readable_stream::Strong::default());
         self.pending.set(streams::WritablePending::default());
+        self.release_wrapper_ref();
+    }
+
+    // Each C++ wrapper holds one +1 (taken in `to_js`/`to_js_with_destructor`,
+    // released here). `construct`'s initial rc=1 belongs to its wrapper; callers
+    // using `init`/`create` then `to_js()` release init's +1 (see Blob::get_writer).
+    fn release_wrapper_ref(&mut self) {
+        self.readable_stream.set(readable_stream::Strong::default());
         self.js_sink_ref.with_mut(|r| r.deinit());
         // SAFETY: `&mut self` carries write provenance over the whole
-        // allocation; this is the last use of `self` in `finalize`.
+        // allocation; this is the last use of `self`.
         unsafe { FileSink::deref(std::ptr::from_mut::<Self>(self)) };
     }
 
@@ -1198,6 +1197,12 @@ impl crate::webcore::sink::JsSinkType for FileSink {
     fn finalize(&mut self) {
         Self::finalize(self)
     }
+    fn wrapper_detached(&mut self) {
+        // `.close()` may run while a backpressured write promise is still
+        // awaited; leave `pending` for `run_pending` and skip the GC-sweep
+        // shutdown cleanup.
+        Self::release_wrapper_ref(self)
+    }
     fn construct(this: &mut core::mem::MaybeUninit<Self>) {
         // `Self::construct()` allocates with `ref_count=1`; that +1 belongs to
         // the C++ `JSFileSink` wrapper `js_construct` is about to create.

src/runtime/webcore/Sink.rs

@@ -749,6 +749,11 @@ pub trait JsSinkType: Sized {
 
     fn memory_cost(&self) -> usize;
     fn finalize(&mut self);
+    /// `.close()` nulled `m_sinkPtr`; release the wrapper's ref. Runs outside
+    /// GC sweep, so overrides may differ from `finalize`. Default: `finalize`.
+    fn wrapper_detached(&mut self) {
+        self.finalize();
+    }
     fn write_bytes(&mut self, data: &streams::Result) -> streams::result::Writable;
     fn write_utf16(&mut self, data: &streams::Result) -> streams::result::Writable;
     fn write_latin1(&mut self, data: &streams::Result) -> streams::result::Writable;
@@ -1032,6 +1037,12 @@ impl<T: JsSinkType + JsSinkAbi> JSSink<T> {
         this.finalize();
     }
 
+    /// `${abi_name}__wrapperDetached` body.
+    #[inline]
+    pub fn js_wrapper_detached(this: &mut T) {
+        this.wrapper_detached();
+    }
+
     /// `${abi_name}__controllerDetached` body — called from
     /// `JSReadable*Controller::detach()` (controller `.end()`/`.close()` host
     /// fns) and from the controller's destructor, i.e. whenever the

src/runtime/webcore/s3/client.rs

@@ -431,6 +431,15 @@ pub(crate) fn writable_stream(
 ) -> JsResult<JSValue> {
     // Local callback wrapper
     fn wrapper_callback(result: S3UploadResult, sink: &mut NetworkSink) -> JsTerminatedResult<()> {
+        // Release the sink's +1 on the task (so `poll_ref` unrefs synchronously)
+        // and `callback_context`'s +1 on the sink, even on `JsTerminated` below.
+        let _teardown = scopeguard::guard(core::ptr::from_mut(sink), |s| {
+            // SAFETY: `s` is the heap-allocated `callback_context`, still live
+            // for the duration of this callback; last use here.
+            unsafe { &mut *s }.detach_writable();
+            // SAFETY: releases `callback_context`'s +1.
+            unsafe { NetworkSink::deref(s) };
+        });
         // `global_this` is a `BackRef` set at construction; copy it so the
         // re-borrow does not hold `&sink` across the `&mut sink` calls below.
         let global = sink
@@ -467,7 +476,6 @@ pub(crate) fn writable_stream(
                 }
             }
         }
-        sink.finalize();
         Ok(())
     }
 
@@ -538,6 +546,7 @@ pub(crate) fn writable_stream(
     // compatible (`{ sink: NetworkSink }`) so the cast in `to_sink()` is just a pointer reinterpret.
     let response_stream: *mut NetworkSink =
         bun_core::heap::into_raw(NetworkSink::new(NetworkSink {
+            ref_count: core::cell::Cell::new(2), // +1 for callback_context
             task: NonNull::new(task_ptr).map(bun_ptr::BackRef::from),
             global_this: Some(bun_ptr::BackRef::new(global_this)),
             high_water_mark: part_size as BlobSizeType,

src/runtime/webcore/streams.rs

@@ -2133,7 +2133,13 @@ pub type H3ResponseSink = HTTPServerWritable<true, true>;
 // NetworkSink
 // ──────────────────────────────────────────────────────────────────────────
 
+// Two intrusive-rc owners: the `JSNetworkSink` wrapper (`m_sinkPtr`) and
+// `MultiPartUpload.callback_context`; each `deref()`s exactly once. `deinit`
+// releases the sink's counted ref on the task before freeing.
+#[derive(bun_ptr::CellRefCounted)]
+#[ref_count(destroy = Self::deinit)]
 pub struct NetworkSink {
+    pub ref_count: core::cell::Cell<u32>,
     // Stored as `BackRef`
     // (set-once); while `Some` the sink holds a counted ref on the intrusively
     // ref-counted `MultiPartUpload`, released in `detach_writable`.
@@ -2152,6 +2158,7 @@ pub struct NetworkSink {
 impl Default for NetworkSink {
     fn default() -> Self {
         Self {
+            ref_count: core::cell::Cell::new(1),
             task: None,
             signal: Signal::default(),
             global_this: None,
@@ -2240,10 +2247,21 @@ impl NetworkSink {
     }
 
     pub fn finalize(&mut self) {
-        self.detach_writable();
+        // SAFETY: `&mut self` carries write provenance over the whole
+        // allocation; this is the last use of `self`.
+        unsafe { NetworkSink::deref(core::ptr::from_mut::<Self>(self)) };
+    }
+
+    /// Runs once when the refcount hits zero. `Drop for MultiPartUpload` does
+    /// not enter JS, so this is safe when reached from a GC-sweep `finalize`.
+    unsafe fn deinit(this: *mut Self) {
+        // SAFETY: caller contract — sole owner of a heap-allocated `Self`.
+        unsafe { &mut *this }.detach_writable();
+        // SAFETY: `this` was allocated via `heap::into_raw` in `writable_stream`.
+        drop(unsafe { bun_core::heap::take(this) });
     }
 
-    fn detach_writable(&mut self) {
+    pub fn detach_writable(&mut self) {
         if let Some(task) = self.task.take() {
             // task is ref-counted; deref releases our ref
             bun_s3::MultiPartUpload::deref_(task.as_ptr());
@@ -2305,26 +2323,12 @@ impl NetworkSink {
         ))
     }
 
-    /// # Safety
-    /// `this` must be a valid, uniquely-owned heap pointer to `Self` produced
-    /// by `bun_core::heap::into_raw`; the caller transfers ownership.
-    // Forwards `this` to `bun_core::heap::take` without dereferencing it here;
-    // not_unsafe_ptr_arg_deref is a false positive on opaque-token forwarding.
-    #[allow(clippy::not_unsafe_ptr_arg_deref)]
-    pub fn finalize_and_destroy(this: *mut Self) {
-        // SAFETY: this was heap-allocated; reclaim sole ownership before
-        // touching fields so no `&mut *this` is live alongside the Box.
-        let mut this = unsafe { bun_core::heap::take(this) };
-        this.finalize();
-        drop(this);
-    }
-
     pub fn abort(&mut self) {
         self.ended = true;
         self.done = true;
         self.signal.close(None);
         self.cancel = true;
-        self.finalize();
+        self.detach_writable();
     }
 
     pub fn write(&mut self, data: &StreamResult) -> Writable {

test/js/bun/s3/s3-networksink-leak.test.ts

@@ -0,0 +1,92 @@
+import { describe, expect, test } from "bun:test";
+import { bunEnv, bunExe, isASAN } from "harness";
+
+// LSAN byte-count diff between N and N+20 writers: symbolize=0 keeps each
+// subprocess fast, and one-time at-exit allocations cancel out in the diff.
+// https://github.com/oven-sh/bun/pull/34999
+async function runWriters(count: number, fail: boolean, finish: "end" | "close") {
+  const script = `
+    const server = Bun.serve({
+      port: 0,
+      async fetch(req) {
+        await req.arrayBuffer();
+        ${
+          fail
+            ? `return new Response(
+                 '<?xml version="1.0" encoding="UTF-8"?><Error><Code>AccessDenied</Code><Message>nope</Message></Error>',
+                 { status: 403 },
+               );`
+            : `return new Response("", { status: 200, headers: { etag: '"e"' } });`
+        }
+      },
+    });
+    server.unref();
+    const s3 = new Bun.S3Client({
+      accessKeyId: "k",
+      secretAccessKey: "s",
+      bucket: "b",
+      endpoint: \`http://127.0.0.1:\${server.port}\`,
+    });
+    process.once("beforeExit", () => { Bun.gc(true); console.log("done"); });
+    for (let i = 0; i < ${count}; i++) {
+      const w = s3.file("key-" + i).writer({ retry: 0 });
+      w.write("hello");
+      ${finish === "end" ? `try { await w.end(); } catch {}` : `w.close();`}
+    }
+  `;
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "-e", script],
+    env: {
+      ...bunEnv,
+      ASAN_OPTIONS: [bunEnv.ASAN_OPTIONS, "detect_leaks=1", "symbolize=0"].filter(Boolean).join(":"),
+      // The S3 client does not honor NO_PROXY for writer(), so an inherited
+      // proxy would hijack the request to the in-process mock server.
+      http_proxy: undefined,
+      HTTP_PROXY: undefined,
+      https_proxy: undefined,
+      HTTPS_PROXY: undefined,
+    },
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  const [stdout, stderr] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect(stdout.trim()).toBe("done");
+  return Number(/SUMMARY: AddressSanitizer: (\d+) byte\(s\) leaked/.exec(stderr)?.[1] ?? 0);
+}
+
+describe.skipIf(!isASAN)("S3 writer() NetworkSink is freed", () => {
+  for (const fail of [true, false]) {
+    for (const finish of ["end", "close"] as const) {
+      test.concurrent(`via .${finish}() when the upload ${fail ? "fails" : "succeeds"}`, async () => {
+        const small = await runWriters(2, fail, finish);
+        const large = await runWriters(22, fail, finish);
+        // Before the fix the diff is >= 20 * sizeof(NetworkSink) ~= 1600.
+        expect(large - small).toBeLessThan(400);
+      });
+    }
+  }
+});
+
+test("S3 writer() unrefs the event loop once end() resolves, even if the writer is retained", async () => {
+  const script = `
+    const server = Bun.serve({
+      port: 0,
+      async fetch(req) { await req.arrayBuffer(); return new Response("", { status: 200, headers: { etag: '"e"' } }); },
+    });
+    server.unref();
+    const s3 = new Bun.S3Client({ accessKeyId: "k", secretAccessKey: "s", bucket: "b", endpoint: \`http://127.0.0.1:\${server.port}\` });
+    const w = s3.file("k").writer({ retry: 0 });
+    w.write("hi");
+    await w.end();
+    globalThis.keep = w;
+    process.once("beforeExit", () => console.log("beforeExit"));
+  `;
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "-e", script],
+    env: { ...bunEnv, http_proxy: undefined, HTTP_PROXY: undefined, https_proxy: undefined, HTTPS_PROXY: undefined },
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect({ stdout: stdout.trim(), exitCode }).toEqual({ stdout: "beforeExit", exitCode: 0 });
+});

test/js/bun/util/filesink.test.ts

@@ -207,6 +207,34 @@ it("write result is not cumulative", async () => {
   await util.promisify(fs.close)(fd);
 });
 
+it.skipIf(!isPosix)("a backpressured write() promise settles after .close()", async () => {
+  const [readFd, writeFd] = createSocketPair();
+  const sink = Bun.file(writeFd).writer();
+  const size = 4 * 1024 * 1024;
+
+  const pending = sink.write(Buffer.alloc(size, 0x61));
+  expect(pending).toBeInstanceOf(Promise);
+
+  // .close() detaches m_sinkPtr and releases the wrapper's ref via
+  // wrapper_detached, which must leave `pending` alone so run_pending can
+  // still settle the promise once the writer drains.
+  sink.close();
+
+  let received = 0;
+  const reader = (async () => {
+    for await (const part of Bun.file(readFd).stream()) received += part.byteLength;
+  })();
+
+  try {
+    expect(await pending).toBe(size);
+  } finally {
+    fs.closeSync(writeFd);
+    await reader;
+    fs.closeSync(readFd);
+  }
+  expect(received).toBe(size);
+});
+
 // A backpressured write buffers everything `write(2)` would not take, so the
 // Promise it returns has to resolve with the chunk's own byte count. It used to
 // resolve with the partial `write(2)` return instead.