Fix EventSource use-after-free

src/workerd/api/eventsource.c++

@@ -17,7 +17,9 @@ namespace workerd::api {
 namespace {
 class EventSourceSink final: public WritableStreamSink {
  public:
-  EventSourceSink(EventSource& eventSource): eventSource(eventSource) {}
+  EventSourceSink(EventSource& eventSource, kj::Rc<jsg::WeakRef<EventSource>> weakEventSource)
+      : eventSource(eventSource),
+        weakEventSource(kj::mv(weakEventSource)) {}
 
   kj::Promise<void> write(kj::ArrayPtr<const kj::byte> buffer) override {
     // The event stream is a new-line delimited format where each line represents an event.
@@ -86,6 +88,7 @@ class EventSourceSink final: public WritableStreamSink {
 
  private:
   kj::Maybe<EventSource&> eventSource;
+  kj::Rc<jsg::WeakRef<EventSource>> weakEventSource;
 
   // Retained bytes to be processed in the next write.
   kj::Vector<char> kept;
@@ -181,7 +184,7 @@ class EventSourceSink final: public WritableStreamSink {
     auto pending = pendingMessages.releaseAsArray();
     // If the event source is gone, just drop the messages on the floor.
     KJ_IF_SOME(es, eventSource) {
-      es.enqueueMessages(kj::mv(pending));
+      es.enqueueMessages(kj::mv(pending), weakEventSource.addRef());
     }
   }
 
@@ -497,8 +500,11 @@ void EventSource::run(jsg::Lock& js,
   // or errored.
   context
       .awaitIo(js,
-          processBody(
-              context, readable.pumpTo(js, kj::heap<EventSourceSink>(*this), EndStream::YES)))
+          processBody(context,
+              readable.pumpTo(js,
+                  kj::heap<EventSourceSink>(
+                      *this, kj::rc<jsg::WeakRef<EventSource>>(JSG_THIS_WEAK(js))),
+                  EndStream::YES)))
       .then(js, kj::mv(onSuccess), kj::mv(onFailed));
 }
 
@@ -509,9 +515,14 @@ void EventSource::close(jsg::Lock& js) {
   readyState = State::CLOSED;
 }
 
-void EventSource::enqueueMessages(kj::Array<PendingMessage> messages) {
-  context.addTask(context.run([this, messages = kj::mv(messages)](
-                                  auto& lock) mutable { notifyMessages(lock, kj::mv(messages)); }));
+void EventSource::enqueueMessages(
+    kj::Array<PendingMessage> messages, kj::Rc<jsg::WeakRef<EventSource>> weakSelf) {
+  context.addTask(
+      context.run([weakSelf = kj::mv(weakSelf), messages = kj::mv(messages)](auto& lock) mutable {
+    KJ_IF_SOME(self, weakSelf->tryAddRef(lock)) {
+      self->notifyMessages(lock, kj::mv(messages));
+    }
+  }));
 }
 
 void EventSource::setReconnectionTime(uint32_t time) {

src/workerd/api/eventsource.h

@@ -122,7 +122,8 @@ class EventSource: public EventTarget {
 
   // Called by the internal implementation to notify the EventSource about messages
   // received from the server.
-  void enqueueMessages(kj::Array<PendingMessage> messages);
+  void enqueueMessages(
+      kj::Array<PendingMessage> messages, kj::Rc<jsg::WeakRef<EventSource>> weakSelf);
 
   // Called by the internal implementation to notify the EventSource that the server
   // has provided a new reconnection time.

src/workerd/api/tests/BUILD.bazel

@@ -555,6 +555,12 @@ wd_test(
     data = ["events-test.js"],
 )
 
+wd_test(
+    src = "eventsource-gc-test.wd-test",
+    args = ["--experimental"],
+    data = ["eventsource-gc-test.js"],
+)
+
 wd_test(
     src = "eventsource-test.wd-test",
     args = ["--experimental"],

src/workerd/api/tests/eventsource-gc-test.js

@@ -0,0 +1,41 @@
+// 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
+
+import assert from 'node:assert';
+import { DurableObject } from 'cloudflare:workers';
+
+export class EventSourceGcDo extends DurableObject {
+  async trigger() {
+    await this.ctx.blockConcurrencyWhile(async () => {
+      // EventSource delivery acquires the actor input gate, so messages stay queued until this
+      // callback returns.
+      const response = await this.env.backend.fetch('https://example.com');
+      const ended = Promise.withResolvers();
+      let source = EventSource.from(response.body);
+      source.onerror = () => ended.resolve();
+
+      await ended.promise;
+
+      const sourceRef = new WeakRef(source);
+      source = null;
+      await scheduler.wait(0);
+      gc();
+      await scheduler.wait(0);
+      gc();
+      await scheduler.wait(0);
+      assert.strictEqual(sourceRef.deref(), undefined);
+    });
+
+    await scheduler.wait(0);
+  }
+}
+
+export default {
+  async test(_controller, env) {
+    await env.ns.getByName('test').trigger();
+  },
+  fetch() {
+    return new Response('data: message\n\n');
+  },
+};

src/workerd/api/tests/eventsource-gc-test.wd-test

@@ -0,0 +1,28 @@
+using Workerd = import "/workerd/workerd.capnp";
+
+const config :Workerd.Config = (
+  v8Flags = ["--expose-gc"],
+  services = [
+    (name = "eventsource-gc-test", worker = .testWorker),
+  ],
+);
+
+const testWorker :Workerd.Worker = (
+  compatibilityFlags = [
+    "enable_weak_ref",
+    "experimental",
+    "nodejs_compat",
+    "streams_enable_constructors",
+  ],
+  modules = [
+    (name = "worker", esModule = embed "eventsource-gc-test.js"),
+  ],
+  durableObjectNamespaces = [
+    (className = "EventSourceGcDo", uniqueKey = "c9f59b7c629443cd973bd29cb3a5d5e9"),
+  ],
+  durableObjectStorage = (inMemory = void),
+  bindings = [
+    (name = "backend", service = "eventsource-gc-test"),
+    (name = "ns", durableObjectNamespace = "EventSourceGcDo"),
+  ],
+);