Dan Lapid

kj-rs-io: KJ async I/O interfaces backed by tokio Final piece of the Rust I/O backend, built on the kj-rs bridge already on main: implements the abstract kj::AsyncIoStream / kj::ConnectionReceiver / kj::Network / kj::AsyncIoProvider / kj::LowLevelAsyncIoProvider interfaces over tokio, so KJ async I/O runs on the tokio-backed kj::EventPort from kj-rs-tokio instead of kj's OS event loop. - Streams, networking (one listening socket per resolved address, KJ's aggregate receiver), socket pairs for the provider's newOneWayPipe/newTwoWayPipe (real sockets, so a write into an empty pipe completes without a reader and workerd's loopback transport gets the sockets it asks for), the --watch file watcher (Rust over the notify crate), signal delivery, and SIGPIPE handling. - Addresses are typed on the bridge. A SocketAddress shared struct (family tag plus that family's fields) is the only form an address takes between Rust and C++: Rust converts it to and from std's SocketAddr / std::os::unix::net::SocketAddr (safe code; the crate views or builds no struct sockaddr bytes at all), and async-io.c++ is the only place a raw sockaddr is decoded (getSockaddr) or encoded (getsockname/getpeername, KJ's NetworkFilter), field by field, at the KJ interfaces that speak them. A caller's struct with garbage past its family's fields, an oversized addrlen, a pathname filling sun_path with no NUL, or a zero-filled sockaddr_un all decode to what KJ makes of them; short or unknown-family structs throw. ffi.rs's hand-written unsafe is fd ownership and the raw read-buffer view alone. - The C++ half is interface adaptation with KJ's own structure where KJ has one. The connect fall-through loop and the accept loop live in the adapter, as in kj/async-io-unix.c++, and apply restrictPeers() there: to each target before connect() tries it and to each accepted peer, through PeerFilter, a wrapper over KJ's own kj::_::NetworkFilter behind a kj::Arc chain (atomic because KJ allows a kj::NetworkAddress to be cloned on another thread and clone() takes a share; a kj::Rc raced under TSAN). Rust returns the peer's typed address with each accepted or connected stream and lists the targets in order; no filter object and no C++ callback crosses the bridge. Both loops own their shares (listener handle, filter), so a receiver or address destroyed with an operation pending does not dangle. KJ's parse-time rejection of a filtered literal (and getSockaddr's eager check) is not reproduced: connect() rejects the same address with the same text. - Two port checks, both two thread-local reads. ensure_loop_thread() before every registration (connect, listen, wrap, resolve, signals, the hangup watch): a call on a thread without a TokioEventPort, or under another runtime entered over the port's, fails with a kj::Exception instead of tokio's "no reactor running" panic (a process abort at the bridge). ensure_owner_loop() -- each stream and listener records the runtime it was registered with -- at the point an operation is about to wait for readiness, never on the fast path: a stream or listener carried to another loop thread fails at its first wait instead of parking on its creator's idle driver forever. Otherwise kj-rs-io is ordinary tokio code: a TokioEventPort thread is a tokio runtime thread for its whole life (kj-rs-tokio), so tokio's own constructors register with the loop's driver as they are; no wrapper runtime type, no lint. - Every adapter method starts its promise inside the call, as KJ's native streams start their operation (coroutine bodies run to their first co_await; eagerlyEvaluate -> EagerPromiseNode -> kj-rs FutureAwaiter::onReady polls on the caller's stack), so a promise that is kept but never awaited completes as the loop turns. When the syscall itself happens is deliberately tokio's semantics, not KJ's: reads and writes are tokio's try_read_buf / try_write / try_write_vectored plus readiness waits, with no direct socket2 / nix / libc syscalls, and vectored writes rely on std clamping the iovec count to IOV_MAX. KJ's AsyncStreamFd issues write(2) synchronously, so a KJ caller may drop a write's promise on the spot and the bytes still go out; under tokio that write is not sent if it is the first operation on a descriptor the driver has not yet seen ready. The long-term direction is to move workerd's I/O onto tokio, so the crate is written as the tokio program it will be part of; workerd's full test suite under the Rust backend has no fire-and-forget write of that kind. - whenWriteDisconnected() costs one dup(2)'d descriptor per stream, created on first use: tokio has one readiness registration per socket, and waiting on it for a hangup would park a concurrent writer. kj-http observes every served connection, so a workerd process holding N connections holds about 2N descriptors under this backend. That is a decision, stated in stream.rs: KJ permits a never-resolving promise (the Windows arm returns one), but early client-disconnect detection is what lets workerd stop work for clients that went away. - Ownership at the FFI boundary: raw fds/SOCKETs become owned typed handles in one `unsafe` block per (unsafe fn) bridge entry point (ffi.rs, the crate's only module allowed to write unsafe), with a bad handle reported as a kj::Exception rather than a panic; KJ read buffers, which callers may leave uninitialized, cross as pointer + length and are handled as MaybeUninit storage rather than `&mut [u8]`. Bridge declarations borrow only where the future really borrows (buffers); operations whose futures own their state are declared safe and lifetime-free, so the compiler enforces that independence. - Operations own their state: every Rust object behind a C++ wrapper is a handle to Arc-shared state and every bridged operation owns a share, so a wrapper destroyed with a read pending does not dangle -- the socket lives until the operation settles or is cancelled (the caller's buffer remains KJ's contract, as under KJ). Every handle is Send + Sync by type (Arc, atomics, Mutex; tokio's own resources are Send + Sync), asserted at compile time in lib.rs, so a rust::Box that C++ carries to another thread is never a memory-safety question. - KJ interface parity for what workerd uses: the address grammar is KJ's SocketAddress::parse for everything workerd.capnp documents for Socket.address / ExternalServer.address -- IP literals, wildcards, decimal ports, service names, hostnames (getaddrinfo with KJ's hints -- AF_UNSPEC, AI_V4MAPPED | AI_ADDRCONFIG -- so IPv6 scope IDs resolve and a host without IPv6 gets no AAAA results; the dns-lookup crate, chosen over tokio::net::lookup_host for exactly those hints and service names), unix: paths and, on Linux, unix-abstract: names (std's from_abstract_name behind tokio's bind_addr / connect_addr); abortRead() ends a pending read with EOF on every platform (the stream records the abort and wakes a parked read itself, since Windows' AFD poller reports no event for a local shutdown(SD_RECEIVE)) and performs KJ's shutdown(SHUT_RD); tryRead waits for readability on EAGAIN whatever minBytes is; address text and watched paths cross the bridge as bytes; connectAuthenticated() and acceptAuthenticated() build the peer identity from the typed peer address, with the network's or receiver's own filter chain threaded into the identity's NetworkAddress as KJ does; accept() retries KJ's set of transient per-connection failures and tolerates TCP_NODELAY failing on an already reset socket; setupTokioAsyncIo() ignores SIGPIPE once per process like kj::UnixEventPort. - Scope: this is workerd's provider, not a drop-in for every KJ program, and lib.rs ("Scope: workerd's provider") says so with the rule behind the list: no consumer in workerd's production code or its configuration surface (workerd.capnp's documented grammar counts), and hand-written libc / sockaddr / fd code to keep. Left out, documented at each site: kj's unix pipe-fd tier (wrapInputFd / wrapOutputFd take sockets only, kj's win32 definition, on every platform; newOneWayPipe is a socket pair), wrapConnectingSocketFd (UNIMPLEMENTED, like getsockopt/setsockopt and newPipeThread), wrapListenSocketFd with a caller-owned NetworkFilter (the two-argument allow-all overload workerd uses works; anything else is UNIMPLEMENTED rather than borrowed for the receiver's lifetime) -- both stubs close a TAKE_OWNERSHIP handle before throwing, since KJ's owning overloads have already released it -- KJ's strtoul(..., 0) port grammar, KJ's std::set re-sort of resolver results (getaddrinfo's RFC 6724 order is kept), the parse-time filter check, and content hashing / ctime tracking in the file watcher (metadata stamps only; the residue -- a same-length rewrite of the same inode within one kernel timestamp tick -- is documented). - The file watcher watches each file's directory (and a symlink target's directory: resolved through dangling links too, so a link whose target is created later fires, re-resolved while the target is missing, and re-registered when a retarget is reported) and judges changes by re-stamping the watched files (inode, size, mtime) whenever the backend reports anything -- an event, an overflow, an error. No event kinds or paths, no content hash, no ctime; a chmod or a replayed pre-watch event moves no stamp and does not fire. No event is stored (the producer only wakes the consumer) and no per-file watch or per-entry descriptor exists. The hand-off from notify's thread is a tokio Notify whose stored permit cannot lose a wake-up; onChange() rejects a second concurrent waiter. It has no C++ wrapper of its own: workerd's TokioFileWatcher (the io_backend change) holds the Rust watcher directly through the three bridged calls. Depends on the kj-rs same-thread waker cells change (#7349, the #7010 subset) for use in workerd: without it, kj-rs's cross-thread waker path races (FuturePollEvent::enterPollScope reading an unfulfilled waker promise, which --config=tsan reports intermittently). This backend must not be enabled by default before that change lands; until then the tokio-backed I/O is opt-in (nothing on main uses it). Dependencies: declares the socket2 (IPV6_V6ONLY, shutdown(2), and the family of a wrapped descriptor), dns-lookup (a safe getaddrinfo wrapper, called with KJ's hints on every platform), notify (file watching, default backends: FSEvents on macOS, which keeps no per-entry descriptor), nix (`fs` for fcntl, `signal` for SIGPIPE) and windows-sys (the Win32 / winsock error codes of KJ's exception-type table) crates and enables tokio's signal and io-util (try_read_buf into uninitialized buffers) features; Cargo.lock repinned accordingly. The CoreServices framework is linked on macOS for FSEvents. Tests: tokio-backed streams and networks (including SIGPIPE survival in a child process with the default disposition, a multi-address hostname listener accepting on every family, vectored writes past IOV_MAX worth of empty pieces and of non-empty pieces, unawaited (kept) writes still going out, abortRead ending a pending read, AF_UNIX pathname and -- on Linux -- abstract sockets listening, connecting, printing and identifying, socket-pair provider semantics (the Windows loopback pair accepting only its own client), decimal ports and service names with the octal/hex grammar gone, the intentional UNIMPLEMENTED / sockets-only stubs), file watching (including a symlinked file whose target lives elsewhere, a retargeted symlink whose new target directory is watched from then on, and a symlink whose target does not exist yet), connectAuthenticated identities over TCP and unix sockets (with the network's filter kept), sockaddrs with garbage past the family's fields or in their padding decoding to the same address, a sun_path-filling unterminated pathname printed whole, short or unknown-family sockaddrs rejected, restrictPeers applied at connect() and accept() (not at parse), every bridged operation refused with a kj::Exception on a thread without a TokioEventPort or under a foreign entered runtime (accept() included), reads and accepts on a different port refused at their first wait, transferred handles closed by the UNIMPLEMENTED stubs, addresses cloned concurrently on two threads (TSAN-clean, kj::Arc), HTTP over tokio-backed streams, a zero-initialized sockaddr_un through getSockaddr, an address with no socket addresses failing connect() and listen(), a zero-minimum read waiting for data, a cancelled backpressured write leaving the socket usable, Cap'n Proto RPC over tokio streams, and PeerFilter's chain ownership and atomic sharing. The C++ tests link statically on every platform: Bazel links cc_test binaries dynamically by default, and under the Linux tsan config each shared library then carries its own unwinder, so an exception thrown in libkj-rs-io-lib.so that unwinds through a frame in libkj-async-io.so (kj's inline owning wrap*Fd overloads) aborted in _Unwind_SetGR. Clean under --config=asan, --config=tsan-macos and the Linux tsan lane. Co-Authored-By: Harris Hancock <harris@cloudflare.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Implement memory cache V2 in safe Rust Introduce a Rust memory-cache core behind the MEMORY_CACHE_V2 autogate, while retaining the existing C++ implementation as the rollback path. Keep the handwritten CXX bridge thin and isolate generated unsafe code from the core, which forbids unsafe code. Store cache entries in a LinkedHashMap for LRU ordering and maintain a separate expiration index for efficient cleanup. Use a per-key Tokio mutex to coalesce concurrent misses without a custom waiter queue. Represent registry membership with weak flight references so abandoned requests do not retain cache state, while flight destruction safely removes only its own registry entry. Make leader handoff cancellation-safe through mutex ownership. Promotion follows waiter poll order rather than request creation order, avoiding eager polling and custom scheduling machinery. Track waiter cardinality with lock-free, best-effort counters so metrics do not add contention to the cache-state lock. Expose the implementation through the existing memory-cache API and add coverage for eviction, expiration, coalescing, cancellation, abandoned reads, leader promotion, weak lifetime cleanup, and the autogated C++ integration.

kj-rs-tokio: a kj::EventPort driven by a per-thread tokio runtime Piece of the Rust I/O backend split, built on the kj-rs bridge already on main: TokioEventPort implements kj::EventPort over a per-thread tokio current_thread runtime, so a KJ event loop sleeps by parking inside tokio's scheduler -- tokio tasks (and, in the next piece, tokio I/O) run whenever KJ would otherwise block. - wait()/poll() park in LocalSet::block_on on the runtime, woken by cross-thread wake(), by KJ itself when a tokio task queues KJ work, or by the next KJ timer deadline (tokio's timer wheel; ~1 ms, the same granularity as KJ's epoll-based port). - A tokio task running inside the park may use KJ freely -- fulfill a PromiseFulfiller, arm a timer, add to a TaskSet, wake a bridged future -- and the port learns of it through KJ's own hooks rather than a convention: kj::EventLoop reports setRunnable(false) right before it calls wait() (capnproto/capnproto#2814), so the first arm during the park is a setRunnable(true) edge the port turns into a notify; and the port installs itself as the kj::TimerImpl's SleepHooks for the duration of the park, so a sooner timer armed mid-park ends the park and kj::Timer::now() reads the live clock. The one thing a task must not do is re-enter promise.wait() / waitScope.poll() on the loop thread (tokio rejects the nested block_on; the panic surfaces as a kj::Exception). - The port owns its kj::EventLoop, declared last so it is destroyed first, and cancels the LocalSet's spawned tasks in its own destructor before any member goes. Teardown order is member order rather than something a context has to remember: spawned tasks routinely own KJ promises (a bridged PromiseFuture holds an OwnPromiseNode and an armed awaiter Event; a KJ timer promise holds a TimerImpl entry), and dropping them after the loop died was a use-after-free (reproduced: ~TimerPromiseAdapter dereferencing the destroyed TimerImpl). A bare port with no TokioAsyncIoContext is safe too. One port per thread is KJ_REQUIREd on the C++ side and asserted on the Rust side. - kj_rs_tokio::spawn() enqueues !Send futures onto the loop's LocalSet, pinned to the loop thread. - The loop thread stays inside the runtime's tokio context for the port's whole life (EnteredRuntime, the crate's one hand-written unsafe: tokio's EnterGuard<'a> carries its lifetime only as PhantomData, so it is extended to 'static and dropped before the runtime by a manual Drop). Bridged futures are polled outside block_on, and everything they create -- tokio sockets, AsyncFds, signal streams, timers -- must find this runtime's drivers; entering once replaces re-entering around every poll of every I/O future. TokioPort is therefore !Send by type (the context is left on the thread that entered it), which is the contract it always had. - setupTokioAsyncIo() yields the port (loop) + wait scope only; the I/O providers over this port arrive in the kj-rs-io piece. - This port relies on capnproto/capnproto#2814 (lock-free Executor::isCurrent(); setRunnable(false) before EventPort::wait()), now merged into v2. This adds tokio's `sync` feature (Notify) to the workspace; no new crates. Also folds in the #7011 review fixes: link the test bridge through link_deps; avoid tokio TLS access during late thread teardown; bound port shutdown so it does not wait on spawn_blocking tasks; keep task cancellation and thread-affine port operations on the receiver's physical owner thread (rejecting foreign-thread calls); reject tokio setup inside an already-current KJ loop before installing runtime TLS; make the async I/O context immovable; and measure timer deadlines from the precise clock. - Tests: a spawned task fulfilling a kj::PromiseFulfiller during a timed park and during a wait-forever park; a KJ timer armed by a task during both kinds of park, honored at its own (live-clock) deadline; a cross-thread bridged wake arriving while the loop is busy-polling rather than parked; spawned tasks holding a KJ timer / a bridged KJ promise across context teardown and across bare-port teardown (ASAN); one port per thread; a bridged future woken from a plain std::thread while the loop is parked; poll() budget under a yield-forever task; nested promise.wait() from a task surfacing as a kj::Exception; timer cancellation; already-due timers; a panicking spawned task surfacing as a kj::Exception and a detached task still completing; two tokio-ported loops executeAsync-ing into each other concurrently (TSAN); and the review-fix reproducers (port outliving tokio thread-locals, drop not waiting on blocking tasks, foreign-thread cancellation/operations, already-current KJ loop rejection). port.rs units: wake-latch semantics, a concurrent wake() storm from multiple threads (TSAN), cross-thread Handle::spawn, poll() advancing a ready LocalSet task, !Send spawned futures, and the bare-TokioPort Drop fallback. Clean under --config=asan and --config=tsan-macos. - Object-relationship overview in tokio-event-port.h. Co-Authored-By: Harris Hancock <harris@cloudflare.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

fix: flush pending socket writes before shutdown (#1054) A pending write retains the WritableStream writer lock. Calling close on the stream at that point rejects instead of completing the write and closing the writable side. Poll the outstanding write before starting close, returning Pending or the write error as appropriate. Tests use a controlled write promise to cover completion and failure; both fail without this change and pass with it under Node 26. Co-authored-by: Guy Bedford <gbedford@cloudflare.com>

Tag post-delivery exceptions to distinguish runtime failures Introduce WORKER_REQUEST_DELIVERED_DETAIL_ID as the authoritative boundary between startup and runtime failures. Centralize custom-event delivery and tag escaping CONNECT, alarm, custom-event, and request failures. Preserve the final output-gate failure in request metrics while still reporting cancellation during the gate wait. Add focused regression tests for each delivery boundary.

build: --//:io_backend flag and the kj::setupAsyncIo() seam Introduce the build-time I/O backend selection (--//:io_backend={cxx,rust}, defaulting to cxx until the CLI arms land) and the one seam workerd proper needs for it: //src/workerd/util:setup-async-io supplies the kj::setupAsyncIo() symbol per backend. In the cxx config it compiles to an empty TU and the dependency falls through to kj's own; in the rust config the concrete kj OS I/O layer (@capnp-cpp//src/kj:kj-async-os) is not linked at all and the TU defines a tokio-backed kj::setupAsyncIo() plus an inert kj::UnixEventPort (whose real definitions live in the unlinked async-unix.c++), so every existing call site links unchanged with no #if. An EventLoopObserver argument is refused rather than dropped. A kj_test checks that kj::setupAsyncIo() hands back the configured backend with a live event loop. The backend is a per-binary choice. Binaries link the seam -- the workerd binary directly, every kj_test and wd_cc_benchmark binary through its macro -- and libraries do not, so which event loop a binary runs on is that binary's decision alone and a downstream binary linking workerd libraries keeps its own. One exception for now: //src/workerd/tests:test-fixture keeps the dep so that downstream binaries linking it without naming a backend keep linking (TODO(cleanup) there). The per-backend dependency set lives in one place, //src/rust/cxx/kj-rs-io:active-backend (a select), which the seam library depends on. Every caller of kj::setupAsyncIo() (:server, :fallback-service and its test, //src/workerd/tests:test-fixture, incoming-request-test) now depends on :kj-async-core / :kj-async-io instead of the :kj-async umbrella: the umbrella carries kj-async-os, whose definitions would collide with the shim's (an ODR violation ASAN reports on every test binary). json-logger-test drops an unused <kj/async-unix.h> include for the same reason. WORKERD_RUST_IO_BACKEND_RUST is defined only for the TUs that read it (the seam library and its test), so the set of places allowed to diverge per backend stays enumerable. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

server: move the --watch file watcher and SIGTERM handling into cli-io-backend A pure code move, in preparation for selecting the I/O backend at build time: the inotify / kqueue / Win32 FileWatcher classes and the SIGTERM capture and wait leave workerd.c++ for src/workerd/server/cli-io-backend.{h,c++}, behind a small FileWatcher interface and three entry points (makeFileWatcher(), captureSigterm(), onSigterm()). The watcher classes are unchanged apart from implementing that interface (`KjFileWatcher final: public FileWatcher`, `override` on the three methods); workerd.c++'s call sites change only in how they obtain the watcher and the SIGTERM promise. No behavior change. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

workerd: make --//:io_backend=rust the default; cxx CI lane With kj-rs, kj-rs-tokio and kj-rs-io in place and the seams above, workerd runs on them by default: the process event loop is tokio and every socket, stream and listener is tokio-backed. Everything above the stream layer -- kj-http, kj-tls, capnp-rpc, and all of workerd -- is unchanged C++ running over those streams: same request paths, same wire bytes. --//:io_backend=cxx keeps the all-C++ build selectable as an escape hatch; a new CI lane builds everything under it and runs the tests of the two packages that have a per-backend arm (//src/workerd/server, //src/workerd/util); the rest of the tree is identical in both configs. compile_flags.txt carries the define so clangd analyzes the default backend's arms. The server and util AGENTS.md files record the seam's rules: which two libraries may read WORKERD_RUST_IO_BACKEND_RUST, the backend as a per-binary choice, never depend on the :kj-async umbrella, and how the checks enforce it. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

build: dependency-graph and link checks for the rust I/O backend Why the seam is sound only with enforcement: if kj-async-os is ever linked into a rust-config binary, the shim's definitions of kj::setupAsyncIo() and kj::UnixEventPort collide with kj's own, and with static archives the winner is link-order dependent -- a duplicate-symbol error if you are lucky, the wrong event loop silently winning if you are not (archive members are pulled lazily, so a second definition is not an error). Two checks make that deterministic: * build/rust_io_graph_check.sh: one `bazel cquery somepath(:workerd, kj-async-os)` under --//:io_backend=rust, which must come back empty; otherwise it prints the offending dependency path. Run by `just check-io-backend-graph` and by the lint CI lane. * //src/workerd/server:rust-io-link-check, an sh_test over the linked workerd binary: kj's setupAsyncIo()/OS provider absent, the tokio shim present. The mangled names are matched as strings of the binary in one `grep -a` pass -- no nm is hermetically available to a test (the C++ toolchain's is autoconfigured on the host and need not exist on a remote executor; the Rust toolchain does not export its llvm-nm). Rust config, unix and unstripped builds only -- it fails rather than passes on a stripped binary and is excluded by tag from the stripped ASAN lane; MSVC mangling is not covered. The binary is located through $(rlocationpath) so the test also works when workerd is built as an external repository. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Move actor-call detail IDs from jsg/util.h to jsg/exception.h REQUEST_NOT_DELIVERED_TO_ACTOR_DETAIL_ID and the related markers are plain kj::Exception detail IDs, like EXCEPTION_IS_USER_ERROR and the Durable Object abort IDs that already live in exception.h. Keeping them in util.h meant that code which only classifies actor-call failures had to depend on the full //src/workerd/jsg target, and with it V8. util.h includes exception.h, so existing users are unaffected.