cloudflare · danlapid · Aug 14, 2026
@@ -27,10 +27,10 @@ bazel_dep(name = "brotli", version = "1.2.0.bcr.1")
 # capnp-cpp
 http.archive(
     name = "capnp-cpp",
-    sha256 = "6753378bd099029cb2830fecd32dd158218019e459ffd3c8e379cbf025906eb8",
-    strip_prefix = "capnproto-capnproto-a1cd1c4/c++",
+    sha256 = "b7b6ec882c84cc513b271acb20ef7df70228f94d6b32679ec639706f886aee59",
+    strip_prefix = "capnproto-capnproto-59cc025/c++",
     type = "tgz",
-    url = "https://github.com/capnproto/capnproto/tarball/a1cd1c4b3d241b77478035a6ccad8b0fb587d444",
+    url = "https://github.com/capnproto/capnproto/tarball/59cc025fa0c04d083aafb57604a28e72eb506037",
 )
 use_repo(http, "capnp-cpp")
 

@@ -36,6 +36,19 @@ Bazel module, Cargo workspace, toolchain configuration, or external `workerd-cxx
 - `tests/` and `kj-rs/tests/` — Rust and C++ bridge integration tests
 - `tools/bazel/` — Bazel bridge-generation macro used by this component's tests
 
+## Async bridge semantics
+
+- Marking a fn `async` in `extern "Rust"` yields a `kj::Promise<T>` in C++; `async` in
+  `extern "C++"` yields an `impl Future` in Rust.
+- Bridged `kj::Promise<T>`s are **eager by default**: the Rust future is polled to its first
+  suspension point at the call (KJ code assumes hot promises), so callers never need
+  `.eagerlyEvaluate(nullptr)`. `RustFuture::lazily()` (kj-rs/future.h) is the C++-side
+  escape hatch for the rare cold case.
+- The waker bridge is single-threaded: a Rust `.await` of a KJ promise links to the
+  `FuturePollEvent` via an intrusive weak link (`RustPromiseAwaiter::link` /
+  `FuturePollEvent::leaves`), and a cloned waker is a same-thread `FutureWakerCell` that arms
+  the `FuturePollEvent` directly (no atomics, no cross-thread fulfiller).
+
 ## Conventions
 
 - Follow the parent `src/rust/AGENTS.md` and repository `AGENTS.md`.

@@ -11,7 +11,10 @@ wd_cc_library(
         "@platforms//os:windows": True,
         "//conditions:default": False,
     }),
-    visibility = ["//src/rust/cxx/tests:__pkg__"],
+    visibility = [
+        "//src/rust/cxx/kj-rs-tokio:__pkg__",
+        "//src/rust/cxx/tests:__pkg__",
+    ],
     deps = [
         ":bridge",
     ],
@@ -54,13 +57,17 @@ rust_test(
 
 rust_cxx_bridge(
     name = "bridge",
-    src = "lib.rs",
+    src = "ffi.rs",
     hdrs = glob(["*.h"]),
     include_prefix = "kj-rs",
     visibility = ["//src/rust/cxx/tests:__pkg__"],
     deps = [
         "//src/rust/cxx:core",
         "@capnp-cpp//src/kj:kj",
-        "@capnp-cpp//src/kj:kj-async",
+        # kj-rs is the base cxx<->rust Promise/Future bridge: it uses only the abstract async
+        # core (kj::Promise / kj::EventLoop via async.h), no kj OS I/O. Depending on
+        # :kj-async-core (not the :kj-async umbrella) keeps the whole kj-rs stack -- and thus
+        # kj-rs-io / kj-rs-tokio built on it -- off the concrete kj OS event loop (:kj-async-os).
+        "@capnp-cpp//src/kj:kj-async-core",
     ],
 )
@@ -1,6 +1,6 @@
 #include "awaiter.h"
 
-#include <kj-rs/lib.rs.h>
+#include <kj-rs/ffi.rs.h>
 
 #include <kj/debug.h>
 
@@ -28,24 +28,40 @@ RustPromiseAwaiter::RustPromiseAwaiter(
 }
 
 RustPromiseAwaiter::~RustPromiseAwaiter() noexcept(false) {
-  // Our `tracePromise()` implementation checks for a null `node`, so we don't have to sever our
-  // LinkedGroup relationship before destroying `node`. If our FuturePollEvent (our LinkedGroup)
-  // tries to trace us between now and our destructor completing, `tracePromise()` will ignore the
-  // null `node`.
+  // Sever our weak link to any FuturePollEvent before we go away, so it can't trace into or arm a
+  // destroyed awaiter. Our `tracePromise()` implementation also checks for a null `node`, so even
+  // between clearPollEvent() and node reset we are safe to trace.
+  clearPollEvent();
   unwindDetector.catchExceptionsIfUnwinding([this]() { node = nullptr; });
 }
 
+void RustPromiseAwaiter::setPollEvent(FuturePollEvent& futurePollEvent) {
+  KJ_IF_SOME(old, maybePollEvent) {
+    if (&old == &futurePollEvent) return;
+    old.leaves.remove(*this);
+  }
+  futurePollEvent.leaves.add(*this);
+  maybePollEvent = futurePollEvent;
+}
+
+void RustPromiseAwaiter::clearPollEvent() {
+  KJ_IF_SOME(old, maybePollEvent) {
+    old.leaves.remove(*this);
+    maybePollEvent = kj::none;
+  }
+}
+
 void RustPromiseAwaiter::fire() {
   // Safety: Our Event can only fire on the event loop which was active when our Event base class
   // was constructed. Therefore, we don't need to check that we're on the correct event loop.
 
   // Nullify our `maybeOptionWaker` to signal that we are done.
   KJ_DEFER(maybeOptionWaker = kj::none);
 
-  KJ_IF_SOME(futurePollEvent, linkedGroup().tryGet()) {
+  KJ_IF_SOME(futurePollEvent, maybePollEvent) {
     // Optimized path: we're still linked to a FuturePollEvent. Arm it directly.
     futurePollEvent.armDepthFirst();
-    linkedGroup().set(kj::none);
+    clearPollEvent();
   } else KJ_IF_SOME(optionWaker, maybeOptionWaker) {
     // We use wake_if_some() rather than an unconditional wake because the OptionWaker may be empty. This
     // happens when poll() took the optimized path (clearing the OptionWaker and linking to a
@@ -68,7 +84,7 @@ void RustPromiseAwaiter::traceEvent(kj::_::TraceBuilder& builder) {
     node->tracePromise(builder, true);
   }
   // TODO(someday): Can we add an entry for the `.await` expression in Rust here?
-  KJ_IF_SOME(futurePollEvent, linkedGroup().tryGet()) {
+  KJ_IF_SOME(futurePollEvent, maybePollEvent) {
     futurePollEvent.traceEvent(builder);
   }
 }
@@ -82,48 +98,21 @@ void RustPromiseAwaiter::tracePromise(kj::_::TraceBuilder& builder, bool stopAtN
   // TODO(someday): Can we add an entry for the `.await` expression in Rust here?
 }
 
-bool RustPromiseAwaiter::poll(const WakerRef& waker, const KjWaker* maybeKjWaker) {
+bool RustPromiseAwaiter::poll(const WakerRef& waker) {
   // TODO(perf): If `this->isNext()` is true, meaning our event is next in line to fire, can we
   //   disarm it, set `done = true`, etc.? If we can only suspend if our enclosing KJ coroutine has
-  //   suspended at least once, we may be able to check for that through LazyArcWaker, but this path
+  //   suspended at least once, we may be able to check for that through PollWaker, but this path
   //   doesn't have access to one.
 
   KJ_IF_SOME(optionWaker, maybeOptionWaker) {
     // Our Promise is not yet ready.
 
-    // Check for an optimized wake path.
-    KJ_IF_SOME(kjWaker, maybeKjWaker) {
-      KJ_IF_SOME(futurePollEvent, kjWaker.tryGetFuturePollEvent()) {
-        // Optimized path. The Future which is polling our Promise is in turn being polled by a
-        // `co_await` expression somewhere up the stack from us. We can arrange to arm the
-        // `co_await` expression's KJ Event directly when our Promise is ready.
-
-        // Drop any Waker stored in OptionWaker. We'll use the LinkedGroup to wake instead.
-        //
-        // Note: this leaves OptionWaker empty while maybeOptionWaker is still Some(ref). If the
-        // FuturePollEvent is later destroyed (severing the LinkedGroup link) before our Promise
-        // fires, fire() will find no LinkedGroup AND an empty OptionWaker. fire() handles this
-        // via wake_if_some(), which is a no-op on an empty OptionWaker.
-        optionWaker.set_none();
-
-        // Store a reference to the current `co_await` expression's Future polling Event. The
-        // reference is weak, and will be cleared if the `co_await` expression happens to end before
-        // our Promise is ready. In the more likely case that our Promise becomes ready while the
-        // `co_await` expression is still active, we'll arm its Event so it can `poll()` us again.
-        linkedGroup().set(futurePollEvent);
-
-        return false;
-      }
-    }
-
-    // Unoptimized fallback path.
-
     // Tell our OptionWaker to store a clone of whatever Waker we were given.
     optionWaker.set(waker);
 
-    // Clearing our reference to the FuturePollEvent (if we have one) tells our fire()
+    // Clearing our weak reference to the FuturePollEvent (if we have one) tells our fire()
     // implementation to use our OptionWaker to perform the wake.
-    linkedGroup().set(kj::none);
+    clearPollEvent();
 
     return false;
   } else {
@@ -132,6 +121,40 @@ bool RustPromiseAwaiter::poll(const WakerRef& waker, const KjWaker* maybeKjWaker
   }
 }
 
+bool RustPromiseAwaiter::poll(const WakerRef& waker, const PollWaker& pollWaker) {
+  KJ_IF_SOME(futurePollEvent, pollWaker.tryGetFuturePollEvent()) {
+    KJ_IF_SOME(optionWaker, maybeOptionWaker) {
+      // Our Promise is not yet ready, and we have an optimized wake path. The Future which is
+      // polling our Promise is in turn being polled by a `co_await` expression somewhere up the
+      // stack from us. We can arrange to arm the `co_await` expression's KJ Event directly when
+      // our Promise is ready.
+
+      // Drop any Waker stored in OptionWaker. We'll use our weak link to the FuturePollEvent to
+      // wake instead.
+      //
+      // Note: this leaves OptionWaker empty while maybeOptionWaker is still Some(ref). If the
+      // FuturePollEvent is later destroyed (severing our weak link) before our Promise fires,
+      // fire() will find no linked FuturePollEvent AND an empty OptionWaker. fire() handles this
+      // via wake_if_some(), which is a no-op on an empty OptionWaker.
+      optionWaker.set_none();
+
+      // Store a weak reference to the current `co_await` expression's Future polling Event. The
+      // reference is weak, and will be cleared if the `co_await` expression happens to end before
+      // our Promise is ready. In the more likely case that our Promise becomes ready while the
+      // `co_await` expression is still active, we'll arm its Event so it can `poll()` us again.
+      setPollEvent(futurePollEvent);
+
+      return false;
+    } else {
+      // Our Promise is ready.
+      return true;
+    }
+  }
+  // The PollWaker exposes no FuturePollEvent (its owning thread's kj::Executor is not ours --
+  // cannot normally happen in the single-thread world). Fall back to the generic path.
+  return poll(waker);
+}
+
 OwnPromiseNode RustPromiseAwaiter::take_own_promise_node() {
   KJ_ASSERT(maybeOptionWaker == kj::none,
       "take_own_promise_node() should only be called after poll() "
@@ -151,43 +174,29 @@ void guarded_rust_promise_awaiter_drop_in_place(GuardedRustPromiseAwaiter* ptr)
 // =======================================================================================
 // FuturePollEvent
 
-void FuturePollEvent::exitPollScope(kj::Maybe<kj::Promise<void>> maybePromise) {
-  // Await any LazyArcWaker promise that got created during the call to `poll()`. Note that if a
-  // Future returns Ready _and_ synchronously wakes its Waker, the work done to await the
-  // LazyArcWaker promise is wasted, since we will immediately tear the entire BoxFutureAwaiter<T>
-  // down. However, that's an unlikely case, and this work here isn't likely to be a significant
-  // source of overhead.
-  KJ_IF_SOME(promise, maybePromise) {
-    auto& node = arcWakerPromise.emplace(kj::_::PromiseNode::from(kj::mv(promise)));
-    node->setSelfPointer(&node);
-    node->onReady(this);
+FuturePollEvent::~FuturePollEvent() noexcept(false) {
+  // Our FutureWakerCell (if any) is neutralized by the wakerCell guard's destructor during member
+  // destruction, so any waker reference Rust retained past our lifetime observes a dead weak link
+  // on a later wake and is a safe no-op, rather than arming this freed event.
+
+  // Sever our weak links to all leaves, so a RustPromiseAwaiter that outlives us (e.g. a stashed
+  // PromiseFuture) never arms this freed event.
+  for (;;) {
+    auto it = leaves.begin();
+    if (it == leaves.end()) break;
+    auto& leaf = *it;
+    leaves.remove(leaf);
+    leaf.maybePollEvent = kj::none;
   }
 }
 
-void FuturePollEvent::enterPollScope() noexcept {
-  // Clear out any previous LazyArcWaker promise the FuturePollEvent was holding onto. Note that
-  // since there is no code path which rejects this Promise, this is not strictly required for
-  // correctness, but nevertheless serves as a useful assertion.
-  KJ_IF_SOME(node, arcWakerPromise) {
-    kj::_::ExceptionOr<kj::_::Void> output;
-
-    node->get(output);
-    KJ_IF_SOME(exception, kj::runCatchingExceptions([this]() { arcWakerPromise = kj::none; })) {
-      output.addException(kj::mv(exception));
-    }
-
-    // NOTE: `node` is now dangling.
-
-    KJ_IF_SOME(exception, output.exception) {
-      // We should only ever receive a WakeInstruction, never an exception. If we do receive an
-      // exception, it would be because our ArcWaker implementation allowed its cross-thread promise
-      // fulfiller to be destroyed without being fulfilled, or because we foolishly added an
-      // explicit call to the fulfiller's reject() function. Either way, it is a programming error,
-      // so we abort the process here by re-throwing across a noexcept boundary. This avoids having
-      // implement the ability to "reject" the Future poll() Event.
-      kj::throwFatalException(kj::mv(exception));
-    }
+kj::Rc<FutureWakerCell> FuturePollEvent::cloneWakerCell() {
+  // Lazily create the cell, bound to this event.
+  if (wakerCell.cell == nullptr) {
+    wakerCell.cell = kj::rc<FutureWakerCell>(*this);
   }
+  // Hand out a new strong reference for Rust to retain.
+  return wakerCell.cell.addRef();
 }
 
 void FuturePollEvent::tracePromise(kj::_::TraceBuilder& builder, bool stopAtNextEvent) {
@@ -199,32 +208,9 @@ void FuturePollEvent::tracePromise(kj::_::TraceBuilder& builder, bool stopAtNext
   // When tracing, we can only pick one branch to follow. Arbitrarily, I'm following the first
   // RustPromiseAwaiter branch, similar to how ExclusiveJoinPromiseNode chooses its left branch. In
   // the common case, this will be whatever OwnPromiseNode our Rust Future is currently `.await`ing.
-  auto rustPromiseAwaiters = linkedObjects();
-  if (rustPromiseAwaiters.begin() != rustPromiseAwaiters.end()) {
+  if (!leaves.empty()) {
     // Our Rust Future is awaiting an OwnPromiseNode. We'll pick the first one in our list.
-    rustPromiseAwaiters.front().tracePromise(builder, false);
-  } else KJ_IF_SOME(node, arcWakerPromise) {
-    // Our Rust Future is not awaiting any OwnPromiseNode, and instead cloned our Waker. We'll trace
-    // our ArcWaker Promise instead.
-    if (node.get() != nullptr) {
-      node->tracePromise(builder, false);
-    }
-  }
-}
-
-FuturePollEvent::PollScope::PollScope(FuturePollEvent& futurePollEvent): holder(futurePollEvent) {
-  futurePollEvent.enterPollScope();
-}
-
-FuturePollEvent::PollScope::~PollScope() noexcept(false) {
-  holder.get().futurePollEvent.exitPollScope(reset());
-}
-
-kj::Maybe<FuturePollEvent&> FuturePollEvent::PollScope::tryGetFuturePollEvent() const {
-  KJ_IF_SOME(h, holder.tryGet()) {
-    return h.futurePollEvent;
-  } else {
-    return kj::none;
+    leaves.front().tracePromise(builder, false);
   }
 }