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

src/codegen/generate-jssink.ts

@@ -513,6 +513,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 here instead.
+    ${name}__finalize(ptr);
     return JSC::JSValue::encode(JSC::jsUndefined());
 }
 

src/runtime/webcore/FileSink.rs

@@ -932,8 +932,12 @@ impl FileSink {
         // `ref_()` there. Callers that allocate via `init`/`create` and then
         // `to_js()` must `deref()` once to release init's +1 (see
         // `Blob::get_writer`).
+        //
+        // `pending` is NOT cleared here: `.close()` reaches `finalize` via
+        // `doClose` while a backpressured write may still be awaiting its
+        // promise; `run_pending` settles it when the writer drains, and
+        // `deinit()` drops the field once the refcount hits zero.
         self.readable_stream.set(readable_stream::Strong::default());
-        self.pending.set(streams::WritablePending::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`.

src/runtime/webcore/s3/client.rs

@@ -538,6 +538,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 owners hold a raw `*mut NetworkSink`: the `JSNetworkSink` wrapper
+// (`m_sinkPtr`, released via the generated `__finalize`) and the
+// `MultiPartUpload.callback_context` (released in `wrapper_callback`). Both
+// routes end in `finalize()`, so the allocation is freed by intrusive refcount.
+#[derive(bun_ptr::CellRefCounted)]
 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,
@@ -2241,6 +2248,9 @@ 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)) };
     }
 
     fn detach_writable(&mut self) {
@@ -2305,26 +2315,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,77 @@
+import { describe, expect, test } from "bun:test";
+import { bunEnv, bunExe, isASAN } from "harness";
+
+// The NetworkSink behind s3.file(k).writer() is heap-allocated and has two
+// owners: the JS wrapper (m_sinkPtr) and the MultiPartUpload's callback_context.
+// Both release paths routed through NetworkSink::finalize(), which only
+// detached the upload task and never reclaimed the Box. Every writer() leaked
+// one ~80-byte NetworkSink on both the success and failure completion paths,
+// whether finished via .end() or .close().
+//
+// This test spawns two subprocesses under detect_leaks=1 with N and N+20
+// writers and asserts the extra 20 writers do not add a proportional number of
+// leaked bytes. symbolize=0 keeps each run under a second; one-time at-exit
+// allocations are constant between the two runs and cancel out in the diff.
+
+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);
+      });
+    }
+  }
+});