diff --git a/src/workerd/api/streams/internal.c++ b/src/workerd/api/streams/internal.c++ index 272e0f9ab6a..2245801fc1d 100644 --- a/src/workerd/api/streams/internal.c++ +++ b/src/workerd/api/streams/internal.c++ @@ -2332,19 +2332,26 @@ jsg::Promise WritableStreamInternalController::Pipe::pipeLoop(jsg::Lock& j if (handle.isArrayBuffer() || handle.isSharedArrayBuffer() || handle.isArrayBufferView() || handle.isString()) { + // addFunctor so the Rc (whose `owner` ref strongly pins the + // destination stream) is dropped at IoContext teardown even if the + // write promise never settles; a bare capture would otherwise retain + // both streams on the isolate heap until isolate death. + auto& ioContext = IoContext::current(); return state->write(js, handle) .then(js, - [state = state.addRef()]( - jsg::Lock& js) mutable -> jsg::Promise { return state->pipeLoop(js); }, - [state = state.addRef()]( - jsg::Lock& js, jsg::Value reason) mutable -> jsg::Promise { + ioContext.addFunctor( + [state = state.addRef()](jsg::Lock& js) mutable -> jsg::Promise { + return state->pipeLoop(js); + }), + ioContext.addFunctor([state = state.addRef()](jsg::Lock& js, + jsg::Value reason) mutable -> jsg::Promise { if (state->isAborted() || state->isSourceReleased()) { return js.resolvedPromise(); } auto error = jsg::JsValue(reason.getHandle(js)); state->tryErrorParent(js, error); return state->pipeLoop(js); - }); + })); } } // Undefined and null are perfectly valid values to pass through a ReadableStream, diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index bd744bc83f0..8a975054c09 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -238,6 +238,11 @@ wd_test( data = ["errorinfo-tail-test.js"], ) +wd_test( + src = "tail-cf-gc-test.wd-test", + data = ["tail-cf-gc-test.js"], +) + wd_test( src = "errorinfo-stw-test.wd-test", args = ["--experimental"], diff --git a/src/workerd/api/tests/tail-cf-gc-test.js b/src/workerd/api/tests/tail-cf-gc-test.js new file mode 100644 index 00000000000..9e6c6b1d9c6 --- /dev/null +++ b/src/workerd/api/tests/tail-cf-gc-test.js @@ -0,0 +1,83 @@ +// Copyright (c) 2026 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 +// +// Regression test for a GC leak in tail events: TraceItem's fetch Request +// held its `cf` object in an untraced strong root, and getCf() returns that +// same mutable object. A tail handler that made `cf` reference the request +// (or anything reaching its wrapper) closed an uncollectable JS<->C++ cycle: +// cf root -> cf -> Request wrapper -> C++ Request -> Detail -> cf root. +// The fix traces `cf` from Request::visitForGc so V8 can collect the cycle. +import * as assert from 'node:assert'; + +// Module-level state may only hold WeakRefs, or it would pin the objects +// under test. +const cfRefs = []; +const reqRefs = []; + +export default { + async fetch(request, env) { + return new Response('ok'); + }, + + tail(events) { + for (const event of events) { + const req = event.event?.request; + if (!req?.cf) continue; + // Proves the cf blob propagated through the service binding into the + // trace; without this the GC assertions would vacuously pass. + assert.strictEqual(req.cf.marker, 'tail-cf-gc'); + // Mutate the shared cf object to reference the request, closing the + // would-be-uncollectable cycle through the untraced strong root. + req.cf.self = req; + cfRefs.push(new WeakRef(req.cf)); + reqRefs.push(new WeakRef(req)); + } + }, +}; + +async function awaitGc() { + // Multiple GC passes with yields between them; gives the cycle collector + // room to reclaim and avoids the conservative stack scanner pinning the + // most recent allocation. + for (let i = 0; i < 4; i++) { + await scheduler.wait(0); + globalThis.gc(); + } +} + +export const tailRequestCfCollects = { + async test(ctrl, env) { + for (let i = 0; i < 8; i++) { + const res = await env.SERVICE.fetch('http://example.com/', { + cf: { marker: 'tail-cf-gc' }, + }); + assert.strictEqual(await res.text(), 'ok'); + } + + // Tail events are delivered asynchronously after each invocation; poll + // until they have all arrived. + for (let i = 0; i < 100 && cfRefs.length < 8; i++) { + await scheduler.wait(50); + } + assert.strictEqual( + cfRefs.length, + 8, + `expected 8 traced requests with cf, got ${cfRefs.length}` + ); + + await awaitGc(); + let alive = 0; + for (const ref of [...cfRefs, ...reqRefs]) { + if (ref.deref() !== undefined) alive++; + } + // Allow a couple of stragglers: the conservative stack scanner can keep + // the most recently touched pair rooted for an extra cycle. The leak + // under test would keep all of them alive. + assert.ok( + alive <= 2, + `expected traced request cf cycles to be collected, ` + + `${alive} of ${cfRefs.length + reqRefs.length} still alive` + ); + }, +}; diff --git a/src/workerd/api/tests/tail-cf-gc-test.wd-test b/src/workerd/api/tests/tail-cf-gc-test.wd-test new file mode 100644 index 00000000000..8da2ceaf3fc --- /dev/null +++ b/src/workerd/api/tests/tail-cf-gc-test.wd-test @@ -0,0 +1,21 @@ +using Workerd = import "/workerd/workerd.capnp"; + +const unitTests :Workerd.Config = ( + v8Flags = ["--expose-gc"], + services = [ + (name = "tail-cf-gc-test", worker = .tailCfGcWorker), + ], +); + +const tailCfGcWorker :Workerd.Worker = ( + modules = [ + (name = "worker", esModule = embed "tail-cf-gc-test.js") + ], + bindings = [ + (name = "SERVICE", service = "tail-cf-gc-test"), + ], + compatibilityFlags = ["nodejs_compat", "enable_weak_ref"], + # Self-tail: the fetch handler's trace events are delivered to the tail + # handler in this same module. + tails = ["tail-cf-gc-test"], +); diff --git a/src/workerd/api/trace.h b/src/workerd/api/trace.h index c533224b4ad..f4da8b3a318 100644 --- a/src/workerd/api/trace.h +++ b/src/workerd/api/trace.h @@ -311,6 +311,14 @@ class TraceItem::FetchEventInfo::Request final: public jsg::Object { tracker.trackField("detail", detail); } + // getCf() hands out the same mutable cf object, so user code can make it + // reference this Request's wrapper; without tracing, that cycle would be + // uncollectable. Detail is shared only between Request instances, all of + // which visit it here, so the handle is re-traced on every GC cycle. + void visitForGc(jsg::GcVisitor& visitor) { + visitor.visit(detail->cf); + } + private: bool redacted = true; kj::Own detail;