JSSink: release native backing in prototype close()

src/codegen/generate-jssink.ts

@@ -526,8 +526,18 @@ 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 run __finalize for
+    // this ptr. Release the wrapper's ownership here (unconditionally: even if
+    // __close set an exception) so the native backing is freed rather than
+    // leaked. Fire the destroy callback first, same as the destructor does:
+    // Subprocess holds a weak back-pointer that must be cleared before
+    // __finalize can drop the last ref on the sink.
+    if (auto destroy = std::exchange(sink->m_onDestroy, 0)) {
+        Bun__onSinkDestroyed(destroy, ptr);
+    }
+    ${name}__finalize(ptr);
+    RETURN_IF_EXCEPTION(scope, {});
     return JSC::JSValue::encode(JSC::jsUndefined());
 }
 

src/runtime/webcore/FileSink.rs

@@ -961,8 +961,16 @@ 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`).
-        self.readable_stream.set(readable_stream::Strong::default());
-        self.pending.set(streams::WritablePending::default());
+        //
+        // `finalize` is reachable from the prototype `.close()` as well as the
+        // C++ destructor (see `${name}__doClose` in generate-jssink.ts), so it
+        // must not tear down state that in-flight IO still needs: `pending`
+        // may hold a backpressured `write()` promise that `run_pending` will
+        // settle once the writer drains, and `readable_stream` may still be
+        // driving this sink as a spawn stdin. Both are released by `deinit`
+        // (Box drop) once `must_be_kept_alive_until_eof` and the
+        // assignToStream ref are gone. `js_sink_ref` roots the wrapper itself,
+        // so it is safe (and necessary) to release here.
         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`.

test/js/bun/util/arraybuffersink.test.ts

@@ -1,6 +1,7 @@
 import { ArrayBufferSink } from "bun";
 import { describe, expect, it } from "bun:test";
-import { bunEnv, bunExe, withoutAggressiveGC } from "harness";
+import { bunEnv, bunExe, isASAN, withoutAggressiveGC } from "harness";
+import { join } from "node:path";
 
 describe("ArrayBufferSink", () => {
   const fixtures = [
@@ -99,4 +100,59 @@ describe("ArrayBufferSink", () => {
     expect(JSON.parse(stdout)).toEqual({ out: "hello" });
     expect({ exitCode, signalCode: proc.signalCode }).toEqual({ exitCode: 0, signalCode: null });
   });
+
+  // The generated ${name}__doClose detached the JS wrapper (nulling m_sinkPtr)
+  // and then called __close, but never __finalize. The wrapper's destructor
+  // skips __finalize when m_sinkPtr is null, so every close() leaked the boxed
+  // ArrayBufferSink plus its Vec<u8> buffer. The repro runs off a setImmediate
+  // so the allocation stack does not fall under the module-loader suppression.
+  // LSAN symbolization of the leak stacks can take several seconds on its own,
+  // hence the explicit per-test timeout.
+  it.skipIf(!isASAN)(
+    "close() does not leak the native sink (LSAN)",
+    async () => {
+      const src = `
+        await new Promise(resolve => setImmediate(resolve));
+        for (let i = 0; i < 4; i++) {
+          const s = new Bun.ArrayBufferSink();
+          s.start({ stream: true, asUint8Array: true });
+          s.write(Buffer.alloc(4096, 0x61).toString());
+          s.close();
+        }
+        Bun.gc(true);
+        console.log("done");
+      `;
+      await using proc = Bun.spawn({
+        cmd: [bunExe(), "-e", src],
+        env: {
+          ...bunEnv,
+          ASAN_OPTIONS: "detect_leaks=1",
+          LSAN_OPTIONS: `suppressions=${join(import.meta.dirname, "../../../leaksan.supp")}`,
+        },
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+      expect(stdout.trim()).toBe("done");
+      const summary = /SUMMARY: AddressSanitizer: (\d+) byte\(s\) leaked/.exec(stderr);
+      const leaked = summary ? Number(summary[1]) : 0;
+      // Before the fix each iteration leaked the ~48-byte struct and the
+      // 4 KiB write buffer (>16 KiB total for 4 iterations).
+      expect({ leaked, exitCode }).toEqual({ leaked: 0, exitCode: 0 });
+    },
+    30_000,
+  );
+
+  it("close() followed by further calls does not crash", () => {
+    const s = new ArrayBufferSink();
+    s.write("hello");
+    s.close();
+    // After close() the wrapper is detached; every method that needs the
+    // native backing throws the "already been closed" error rather than
+    // dereferencing a freed pointer.
+    expect(() => s.write("x")).toThrow(/already been closed/);
+    expect(() => s.flush()).toThrow(/already been closed/);
+    expect(() => s.end()).toThrow(/already been closed/);
+    expect(s.close()).toBeUndefined();
+  });
 });

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

@@ -323,12 +323,6 @@ it.skipIf(!isPosix)(
 // run_pending, so a backpressured write()'s promise was left pending forever
 // while close() threw. Now close() routes the error to that promise and
 // returns undefined.
-//
-// Runs in a subprocess because sink.close() on a Blob-created FileSink
-// currently leaks the native FileSink (doClose detaches m_sinkPtr so the
-// wrapper's +1 never reaches finalize); running it in-process would abort the
-// whole file under detect_leaks=1. That leak is pre-existing on main and
-// tracked separately.
 it.skipIf(!isPosix)(
   "close() after a backpressured write() with the reader gone rejects the write's promise with EPIPE",
   async () => {
@@ -348,12 +342,7 @@ it.skipIf(!isPosix)(
     `;
     await using proc = Bun.spawn({
       cmd: [bunExe(), "-e", src],
-      env: {
-        ...bunEnv,
-        // Pre-existing leak in sink.close() (see comment above); don't let the
-        // child's LSAN abort hide the actual assertion we're testing.
-        ASAN_OPTIONS: "allow_user_segv_handler=1:disable_coredump=0:detect_leaks=0",
-      },
+      env: bunEnv,
       stderr: "pipe",
     });
     const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
@@ -524,6 +513,49 @@ it.skipIf(!isPosix)("does not leak native FileSink when a pending write fails (E
   expect(fileSinkInternals.liveCount()).toBeLessThanOrEqual(baseline + 1);
 });
 
+// The generated ${name}__doClose detached m_sinkPtr and then called __close,
+// so the wrapper's destructor skipped __finalize and the wrapper's +1 on the
+// native FileSink was never released.
+it("close() does not leak the native FileSink", async () => {
+  const dir = tmpdirSync();
+  const baseline = fileSinkInternals.liveCount();
+  const iterations = 8;
+  for (let i = 0; i < iterations; i++) {
+    const writer = Bun.file(join(dir, `close-leak-${i}.txt`)).writer();
+    writer.write("hi");
+    writer.close();
+  }
+  for (let i = 0; i < 50; i++) {
+    Bun.gc(true);
+    if (fileSinkInternals.liveCount() <= baseline) break;
+    await Bun.sleep(10);
+  }
+  expect(fileSinkInternals.liveCount()).toBeLessThanOrEqual(baseline + 1);
+});
+
+// Now that __doClose runs finalize(), finalize() must not tear down state an
+// in-flight write still needs: clearing `pending` here would drop the
+// backpressure promise's Strong before on_write can settle it.
+it.skipIf(isWindows)("close() while a write() promise is pending still settles it", async () => {
+  await using child = Bun.spawn({
+    cmd: [bunExe(), "-e", "for await (const _ of process.stdin) {}"],
+    env: bunEnv,
+    stdin: "pipe",
+    stdout: "ignore",
+    stderr: "pipe",
+  });
+  const writer = child.stdin;
+  // 4 MiB overflows the default pipe capacity on Linux/macOS so write()
+  // returns a promise.
+  const p = writer.write(Buffer.alloc(4 * 1024 * 1024, 0x61));
+  expect(p).toBeInstanceOf(Promise);
+  writer.close();
+  await expect(p).resolves.toBeGreaterThanOrEqual(0);
+  const [stderr, exitCode] = await Promise.all([child.stderr.text(), child.exited]);
+  if (exitCode !== 0) expect(stderr).toBe("");
+  expect(exitCode).toBe(0);
+});
+
 it("start() without path/fd on an already-open writer does not crash", async () => {
   const path = join(tmpdirSync(), "filesink-restart.txt");
   const writer = Bun.file(path).writer();