kj-rs: arm the poll event directly on a same-thread stored-waker wake by danlapid · Pull Request #7349 · cloudflare/workerd

danlapid · 2026-09-13T22:37:20Z

kj-rs: arm the poll event directly on a same-thread stored-waker wake

A std::task::Waker that Rust clones out of a poll and parks -- in a tokio
oneshot, a channel, a Notify -- is an ArcWaker. Waking it fulfilled a
CrossThreadPromiseFulfiller even when the wake came from the very thread that
owns the future, and KJ dispatches a cross-thread fulfillment only when the
event port's wait()/poll() reports it: after every already-runnable event has
run. So a bridged future woken from another KJ event resumed one loop-idle
later than a coroutine waiting on a kj::PromiseFulfiller would, and anything
queued between the wake and that idle ran first. Consumers that hand values
between bridged futures through stored wakers (kj-rs-io's DNS and pump
handoffs, the Rust HTTP layer's WebSocket pipe and serve-side rendezvous)
saw handed-over messages torn down by later-queued continuations before the
receiver was re-polled.

Now ArcWaker knows the FuturePollEvent of the poll that created it (the
PollScope hands it over in LazyArcWaker::clone(), when cloned on the owning
thread) and wake_by_ref() on that thread arms it directly
(Event::armDepthFirst()), in KJ event order; every other case still goes
through the fulfiller. The reference is revoked at the rendezvous that already
existed for retained wakers, ArcWakerPromiseNode::destroy() -- the next poll,
or the FuturePollEvent's destruction -- so a waker Rust keeps past that point
falls back to the abandoned fulfiller exactly as before and can never arm a
freed event. Arming during a poll re-polls once more; fire() already returns
without polling once the future is done.

Two tests pin the guarantee -- a stored waker woken from another KJ event
re-polls the future before a KJ event queued after the wake runs -- on a bare
kj::EventLoop and under the TokioEventPort; both fail on the previous waker.

🤖 Generated with Claude Code

…overridden Second capability stage of the Rust I/O backend (after the tokio event loop + sockets): under --//:io_backend=rust, all HTTP/1.1 client and server work, WebSockets, CONNECT tunnels and TLS are done by hyper/rustls in Rust. The kj HTTP *interfaces* (kj::HttpClient, HttpService, HttpServer, WebSocket, HttpHeaders) stay the vocabulary workerd is written against; their implementations are replaced by symbol override, the same pattern the I/O layer uses, so product code is unchanged. kj-http split (capnproto side, companion commit): kj-http-types (the header table/object model, interface defaults, url) vs kj-http-impl (kj's HTTP codec, client, server, WebSocketImpl, pipes, adapters). Under rust, kj-http-impl is excluded from the link and the rust-io-hermeticity aspect forbids it; src/workerd/util/kj-http-tokio.c++ defines the kj:: symbols over: - kj-hyper (src/rust/cxx/kj-hyper): hyper client (host:port with TLS and peer filter, and a single-connection client over any kj stream), hyper server fused inline on the KJ loop (one fixed-point poll per connection, no cross-task channels), WebSocket sessions with permessage-deflate, an in-memory WebSocket pipe (kj::newWebSocketPipe) with kj's exact rendezvous/close/abort/pump-adoption semantics, CONNECT tunnels, Upgrade requests (used by the Docker container path), rustls client/server TLS, and the I/O-stall watchdog. - kj-rs-http (src/rust/cxx/kj-rs-http): the off-wire kj vocabulary codec (HttpHeaders serialize/tryParse, method/range parsing) hand-ported for byte parity with kj's storage scheme, plus the kj stream/service bridges. - The shim also carries kj-shape ports of the pooled network clients (per-address idle pool with idleTimeout, per-host cache over the restricted kj::Network, startTls), the client<->service shape adapters, and the WebSocket::pumpTo defaults. server/ collapses back to upstream form: the fd-inspection listener split, external/network channel dichotomy and http-client-backend seam are gone; server.c++ differs from upstream only by the tls-network seam (rustls-backed kj::SecureNetworkWrapper for listeners and outbound). container-client and fallback-service run on hyper with no C++ fallback. Serialization parity with kj (asserted byte-for-byte by server-test): title-case header names, original spellings via a hyper HeaderCaseMap patch, framing header in kj's header-table position, ordered 101 heads, kj's drain rule (idle+clean closes, partial request is served with Connection: close), Connection: close on error-handler responses, no entity-body on bodyless GET/HEAD. Bridge fixes found along the way: serve-side rendezvous cells (response head, streaming body ring) now register wakers so heads/chunks produced by the service on its own KJ event reach the wire without socket activity; stream-client byte pumps are owned by a kj::Promise member of the C++ client rather than a detached task (fixes a use-after-free write through the borrowed stream and armed events leaking past EventLoop teardown); the port's tokio event_interval is pinned below poll()'s yield budget so every WaitScope::poll() turns the reactor. The crate patch annotations (http/httparse lenient header values) lost in the carve are restored, plus patches/rust/hyper-public-header-case-map.patch. State: cxx server-test fully passes; rust server-test 88/90 (the two remaining share one cause: ready work in flight through tokio hops is invisible to kj's idle detection, so Worker hang-abort can fire mid-operation); hermeticity OK. Requires the companion capnproto kj-http split (kj-http-types target). Hot promises at the kj seam: on this kj-rs a bridged promise is cold (the Rust future runs only once the promise is first polled), while every kj interface these wrappers implement is hot -- kj-side callers (WebSocket::couple, the api WebSocket state machine, the hibernation manager, kj's header queue) sequence their own teardown on the operation being in flight when the method returns. hot() (hyper-http.h) starts each bridged future where it crosses a kj interface: the WebSocket pipe ends and hyper sessions (send/close/whenAborted/ pumpTo/tryPumpFrom), body/tunnel/sink streams (read/write/ whenWriteDisconnected), HyperHttpClient::request, HyperHttpServer::serve, the default WebSocket pump, and the held stream-client pump task. Left cold, the extra event hop reorders the hibernatable WebSocket Close hand-off enough that websocket-hibernation aborts in the DO's close handler on most runs (libmalloc reports the Close reason buffer as not allocated; no double free or write-after-free was attributable with zone hooks, watchpoints, guard malloc or zero-on-free checks; the C++ backend and the same binary under mimalloc pass). Once kj-rs promises are eager by default (#7010), hot() is a no-op to sweep. Rebased onto the re-cut Part 2 (3278b32): tokio resources go through kj_rs_tokio::LoopRuntime (the DNS blocking task and the fused serve driver's per-poll runtime context), matching Part 2's clippy wall against tokio's raw entry points; the kj-hyper test harness follows TokioAsyncIoContext's new by-value provider; kj-rs-http is clean under --config=lint. Serve tier: the native-serving entry points Part 2 no longer ships (serve_kj_stream / take_kj_socket, ServeIo, the duplex pump) live here now as kj-hyper's serve module, with their C++-driven tests (kj-hyper/tests:serve-test) and Rust echo fixtures. kj-rs-io keeps only what needs its crate-private state: the unwrap fast path (isTokioStream / unwrapTokioStream, TokioStream::into_socket over a wrapper that goes hollow on take, refusing while an operation still holds a share) and the refcounted read/write ends that let a foreign kj stream be pumped through two exclusive borrows (kj-rs-io/bridge.h). kj-hyper's bridge aliases kj::AsyncIoStream and kj_rs_io::TokioStream directly, so the alias transmute between the two crates' stream types is gone. Depends on kj-rs arming a poll event directly when a stored waker is woken on its own thread (#7349): the WebSocket pipe and the serve-side head/body rendezvous hand values between bridged futures through stored wakers and rely on the receiver being re-polled before a KJ event queued after the wake runs. On the previous ArcWaker that wake was a cross-thread fulfillment dispatched only once the loop went idle, and websocket-hibernation and the stream-client server-tests failed deterministically. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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. Clean under --config=asan and --config=tsan-macos. Co-Authored-By: Harris Hancock <harris@cloudflare.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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>

harrishancock

GPT found a UaF which I don't think is reachable from production ... but now that I think about it, I'm not 100% sure, ~~so holding off approving for the time being~~.

I agree with the behavior change, but I also would like to understand more about how the test in question fails. I don't think the current ordering is technically a correctness bug, though it's definitely sub-optimal.

LLM-generated review comment (Codex).

The same-thread ordering change works, and both new ordering tests fail
without it. I found one remaining blocker in its interaction with the
existing waker machinery:

[HIGH] ArcWaker borrows its executor without retaining it. A Rust waker
kept after cancellation and event-loop destruction can call
executor.isCurrent() on a freed object. A small test using the existing
retained-Rust-waker helpers passes on the parent and fails under UBSAN
on this head.

I also found that direct arming exposes the pre-existing
enterPollScope() race. That issue is fixed by
PR 7372; rebasing this PR
onto PR 7372 picks up the fix and its joined-readiness test.

This cherry-pickable fix
retains the executor while a waker can use it. It applies unchanged either
directly to this PR or after rebasing this PR onto PR 7372. The combined local
stack passes the kj-rs and kj-rs-tokio suites, the retained-waker test under
RTTI-enabled UBSAN, and PR 7372's joined-readiness test under TSAN.

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>

A std::task::Waker that Rust clones out of a poll and parks -- in a tokio oneshot, a channel, a Notify -- is an ArcWaker. Waking it fulfilled a CrossThreadPromiseFulfiller even when the wake came from the very thread that owns the future, and KJ dispatches a cross-thread fulfillment only when the event port's wait()/poll() reports it: after every already-runnable event has run. So a bridged future woken from another KJ event resumed one loop-idle later than a coroutine waiting on a kj::PromiseFulfiller would, and anything queued between the wake and that idle ran first. Consumers that hand values between bridged futures through stored wakers (kj-rs-io's DNS and pump handoffs, the Rust HTTP layer's WebSocket pipe and serve-side rendezvous) saw handed-over messages torn down by later-queued continuations before the receiver was re-polled. Now ArcWaker knows the FuturePollEvent of the poll that created it (the PollScope hands it over in LazyArcWaker::clone(), when cloned on the owning thread) and wake_by_ref() on that thread arms it directly (Event::armDepthFirst()), in KJ event order; every other case still goes through the fulfiller. The reference is revoked at the rendezvous that already existed for retained wakers, ArcWakerPromiseNode::destroy() -- the next poll, or the FuturePollEvent's destruction -- so a waker Rust keeps past that point falls back to the abandoned fulfiller exactly as before and can never arm a freed event. Arming during a poll re-polls once more; fire() already returns without polling once the future is done. Two tests pin the guarantee -- a stored waker woken from another KJ event re-polls the future before a KJ event queued after the wake runs -- on a bare kj::EventLoop and under the TokioEventPort; both fail on the previous waker. ArcWaker owns its executor (kj::Own<const kj::Executor> via addRef()) rather than borrowing it: Rust may retain a cloned waker past the future and the event loop, and wake_by_ref() asks the executor whether its loop is current. With the loop gone, isCurrent() reports false and the wake falls back to the abandoned fulfiller; two retained-waker tests cover the canceled and fulfilled paths (they fail under RTTI-enabled UBSAN with a borrowed executor). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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>

…overridden Second capability stage of the Rust I/O backend (after the tokio event loop + sockets): under --//:io_backend=rust, all HTTP/1.1 client and server work, WebSockets, CONNECT tunnels and TLS are done by hyper/rustls in Rust. The kj HTTP *interfaces* (kj::HttpClient, HttpService, HttpServer, WebSocket, HttpHeaders) stay the vocabulary workerd is written against; their implementations are replaced by symbol override, the same pattern the I/O layer uses, so product code is unchanged. kj-http split (capnproto side, companion commit): kj-http-types (the header table/object model, interface defaults, url) vs kj-http-impl (kj's HTTP codec, client, server, WebSocketImpl, pipes, adapters). Under rust, kj-http-impl is excluded from the link and the rust-io-hermeticity aspect forbids it; src/workerd/util/kj-http-tokio.c++ defines the kj:: symbols over: - kj-hyper (src/rust/cxx/kj-hyper): hyper client (host:port with TLS and peer filter, and a single-connection client over any kj stream), hyper server fused inline on the KJ loop (one fixed-point poll per connection, no cross-task channels), WebSocket sessions with permessage-deflate, an in-memory WebSocket pipe (kj::newWebSocketPipe) with kj's exact rendezvous/close/abort/pump-adoption semantics, CONNECT tunnels, Upgrade requests (used by the Docker container path), rustls client/server TLS, and the I/O-stall watchdog. - kj-rs-http (src/rust/cxx/kj-rs-http): the off-wire kj vocabulary codec (HttpHeaders serialize/tryParse, method/range parsing) hand-ported for byte parity with kj's storage scheme, plus the kj stream/service bridges. - The shim also carries kj-shape ports of the pooled network clients (per-address idle pool with idleTimeout, per-host cache over the restricted kj::Network, startTls), the client<->service shape adapters, and the WebSocket::pumpTo defaults. server/ collapses back to upstream form: the fd-inspection listener split, external/network channel dichotomy and http-client-backend seam are gone; server.c++ differs from upstream only by the tls-network seam (rustls-backed kj::SecureNetworkWrapper for listeners and outbound). container-client and fallback-service run on hyper with no C++ fallback. Serialization parity with kj (asserted byte-for-byte by server-test): title-case header names, original spellings via a hyper HeaderCaseMap patch, framing header in kj's header-table position, ordered 101 heads, kj's drain rule (idle+clean closes, partial request is served with Connection: close), Connection: close on error-handler responses, no entity-body on bodyless GET/HEAD. Bridge fixes found along the way: serve-side rendezvous cells (response head, streaming body ring) now register wakers so heads/chunks produced by the service on its own KJ event reach the wire without socket activity; stream-client byte pumps are owned by a kj::Promise member of the C++ client rather than a detached task (fixes a use-after-free write through the borrowed stream and armed events leaking past EventLoop teardown); the port's tokio event_interval is pinned below poll()'s yield budget so every WaitScope::poll() turns the reactor. The crate patch annotations (http/httparse lenient header values) lost in the carve are restored, plus patches/rust/hyper-public-header-case-map.patch. State: cxx server-test fully passes; rust server-test 88/90 (the two remaining share one cause: ready work in flight through tokio hops is invisible to kj's idle detection, so Worker hang-abort can fire mid-operation); hermeticity OK. Requires the companion capnproto kj-http split (kj-http-types target). Hot promises at the kj seam: on this kj-rs a bridged promise is cold (the Rust future runs only once the promise is first polled), while every kj interface these wrappers implement is hot -- kj-side callers (WebSocket::couple, the api WebSocket state machine, the hibernation manager, kj's header queue) sequence their own teardown on the operation being in flight when the method returns. hot() (hyper-http.h) starts each bridged future where it crosses a kj interface: the WebSocket pipe ends and hyper sessions (send/close/whenAborted/ pumpTo/tryPumpFrom), body/tunnel/sink streams (read/write/ whenWriteDisconnected), HyperHttpClient::request, HyperHttpServer::serve, the default WebSocket pump, and the held stream-client pump task. Left cold, the extra event hop reorders the hibernatable WebSocket Close hand-off enough that websocket-hibernation aborts in the DO's close handler on most runs (libmalloc reports the Close reason buffer as not allocated; no double free or write-after-free was attributable with zone hooks, watchpoints, guard malloc or zero-on-free checks; the C++ backend and the same binary under mimalloc pass). Once kj-rs promises are eager by default (#7010), hot() is a no-op to sweep. Rebased onto main with Part 2 (#7013) and the same-thread waker arm (#7349) merged and capnproto repinned to d1ebde0 (the kj-http split rebased onto it as dlapid/kjHttpSplit 61c00a8c), under the re-cut Part 3 (rust backend default, hermeticity gate + link check). kj-rs-io's stream is an Arc with a runtime-id owner now, so the take primitive is a Mutex<Option<Arc>> that goes hollow; its unwrap hook and the foreign-stream ends moved to kj-rs-io/unwrap.h so bridge.h stays free of <kj/async-io.h> (Windows: windows.h vs kj/compat/http.h); PeerFilter is atomically refcounted (kj::Arc); kj-hyper is back on tokio::task::spawn_blocking / current_handle().enter() since LoopRuntime was dropped; main's new TcpListener::run goes through the tls-network seam like HttpListener::run. kj-rs-http is clean under --config=lint. Serve tier: the native-serving entry points Part 2 no longer ships (serve_kj_stream / take_kj_socket, ServeIo, the duplex pump) live here now as kj-hyper's serve module, with their C++-driven tests (kj-hyper/tests:serve-test) and Rust echo fixtures. kj-rs-io keeps only what needs its crate-private state: the unwrap fast path (isTokioStream / unwrapTokioStream, TokioStream::into_socket over a wrapper that goes hollow on take, refusing while an operation still holds a share) and the refcounted read/write ends that let a foreign kj stream be pumped through two exclusive borrows (kj-rs-io/bridge.h). kj-hyper's bridge aliases kj::AsyncIoStream and kj_rs_io::TokioStream directly, so the alias transmute between the two crates' stream types is gone. Depends on kj-rs arming a poll event directly when a stored waker is woken on its own thread (#7349): the WebSocket pipe and the serve-side head/body rendezvous hand values between bridged futures through stored wakers and rely on the receiver being re-polled before a KJ event queued after the wake runs. On the previous ArcWaker that wake was a cross-thread fulfillment dispatched only once the loop went idle, and websocket-hibernation and the stream-client server-tests failed deterministically. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

…overridden Second capability stage of the Rust I/O backend (after the tokio event loop + sockets): under --//:io_backend=rust, all HTTP/1.1 client and server work, WebSockets, CONNECT tunnels and TLS are done by hyper/rustls in Rust. The kj HTTP *interfaces* (kj::HttpClient, HttpService, HttpServer, WebSocket, HttpHeaders) stay the vocabulary workerd is written against; their implementations are replaced by symbol override, the same pattern the I/O layer uses, so product code is unchanged. kj-http split (capnproto side, companion commit): kj-http-types (the header table/object model, interface defaults, url) vs kj-http-impl (kj's HTTP codec, client, server, WebSocketImpl, pipes, adapters). Under rust, kj-http-impl is excluded from the link and the rust-io-hermeticity aspect forbids it; src/workerd/util/kj-http-tokio.c++ defines the kj:: symbols over: - kj-hyper (src/rust/cxx/kj-hyper): hyper client (host:port with TLS and peer filter, and a single-connection client over any kj stream), hyper server fused inline on the KJ loop (one fixed-point poll per connection, no cross-task channels), WebSocket sessions with permessage-deflate, an in-memory WebSocket pipe (kj::newWebSocketPipe) with kj's exact rendezvous/close/abort/pump-adoption semantics, CONNECT tunnels, Upgrade requests (used by the Docker container path), rustls client/server TLS, and the I/O-stall watchdog. - kj-rs-http (src/rust/cxx/kj-rs-http): the off-wire kj vocabulary codec (HttpHeaders serialize/tryParse, method/range parsing) hand-ported for byte parity with kj's storage scheme, plus the kj stream/service bridges. - The shim also carries kj-shape ports of the pooled network clients (per-address idle pool with idleTimeout, per-host cache over the restricted kj::Network, startTls), the client<->service shape adapters, and the WebSocket::pumpTo defaults. server/ collapses back to upstream form: the fd-inspection listener split, external/network channel dichotomy and http-client-backend seam are gone; server.c++ differs from upstream only by the tls-network seam (rustls-backed kj::SecureNetworkWrapper for listeners and outbound). container-client and fallback-service run on hyper with no C++ fallback. Serialization parity with kj (asserted byte-for-byte by server-test): title-case header names, original spellings via a hyper HeaderCaseMap patch, framing header in kj's header-table position, ordered 101 heads, kj's drain rule (idle+clean closes, partial request is served with Connection: close), Connection: close on error-handler responses, no entity-body on bodyless GET/HEAD. Bridge fixes found along the way: serve-side rendezvous cells (response head, streaming body ring) now register wakers so heads/chunks produced by the service on its own KJ event reach the wire without socket activity; stream-client byte pumps are owned by a kj::Promise member of the C++ client rather than a detached task (fixes a use-after-free write through the borrowed stream and armed events leaking past EventLoop teardown); the port's tokio event_interval is pinned below poll()'s yield budget so every WaitScope::poll() turns the reactor. The crate patch annotations (http/httparse lenient header values) lost in the carve are restored, plus patches/rust/hyper-public-header-case-map.patch. State: cxx server-test fully passes; rust server-test 88/90 (the two remaining share one cause: ready work in flight through tokio hops is invisible to kj's idle detection, so Worker hang-abort can fire mid-operation); hermeticity OK. Requires the companion capnproto kj-http split (kj-http-types target). Hot promises at the kj seam: on this kj-rs a bridged promise is cold (the Rust future runs only once the promise is first polled), while every kj interface these wrappers implement is hot -- kj-side callers (WebSocket::couple, the api WebSocket state machine, the hibernation manager, kj's header queue) sequence their own teardown on the operation being in flight when the method returns. hot() (hyper-http.h) starts each bridged future where it crosses a kj interface: the WebSocket pipe ends and hyper sessions (send/close/whenAborted/ pumpTo/tryPumpFrom), body/tunnel/sink streams (read/write/ whenWriteDisconnected), HyperHttpClient::request, HyperHttpServer::serve, the default WebSocket pump, and the held stream-client pump task. Left cold, the extra event hop reorders the hibernatable WebSocket Close hand-off enough that websocket-hibernation aborts in the DO's close handler on most runs (libmalloc reports the Close reason buffer as not allocated; no double free or write-after-free was attributable with zone hooks, watchpoints, guard malloc or zero-on-free checks; the C++ backend and the same binary under mimalloc pass). Once kj-rs promises are eager by default (#7010), hot() is a no-op to sweep. Rebased onto main with Part 2 (#7013) and the same-thread waker arm (#7349) merged and capnproto repinned to d1ebde0 (the kj-http split rebased onto it as dlapid/kjHttpSplit 61c00a8c), under the re-cut Part 3 (rust backend default, hermeticity gate + link check). kj-rs-io's stream is an Arc with a runtime-id owner now, so the take primitive is a Mutex<Option<Arc>> that goes hollow; its unwrap hook and the foreign-stream ends moved to kj-rs-io/unwrap.h so bridge.h stays free of <kj/async-io.h> (Windows: windows.h vs kj/compat/http.h); PeerFilter is atomically refcounted (kj::Arc); kj-hyper is back on tokio::task::spawn_blocking / current_handle().enter() since LoopRuntime was dropped; main's new TcpListener::run goes through the tls-network seam like HttpListener::run. kj-rs-http is clean under --config=lint. Serve tier: the native-serving entry points Part 2 no longer ships (serve_kj_stream / take_kj_socket, ServeIo, the duplex pump) live here as kj-hyper's serve module, with their C++-driven tests (kj-hyper/tests:serve-test) and Rust echo fixtures. The handoff is consuming: kj-hyper downcasts the owned kj::AsyncIoStream, releases the kj-rs-io wrapper's native stream and destroys the wrapper (kj-stream.h), and TokioStream::into_socket takes the socket or hands the stream back -- re-wrapped, untouched -- while an operation still holds a share of it. kj-rs-io grows only that into_socket, a release() on the C++ wrapper and the Socket re-export; no hollow state, no lock in the stream's hot path. The refcounted read/write ends that let a foreign kj stream be pumped through two exclusive borrows are kj-hyper's too. kj-hyper's bridge aliases kj::AsyncIoStream and kj_rs_io::TokioStream directly, so the alias transmute between the two crates' stream types is gone. Depends on kj-rs arming a poll event directly when a stored waker is woken on its own thread (#7349): the WebSocket pipe and the serve-side head/body rendezvous hand values between bridged futures through stored wakers and rely on the receiver being re-polled before a KJ event queued after the wake runs. On the previous ArcWaker that wake was a cross-thread fulfillment dispatched only once the loop went idle, and websocket-hibernation and the stream-client server-tests failed deterministically. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

…overridden Second capability stage of the Rust I/O backend (after the tokio event loop + sockets): under --//:io_backend=rust, all HTTP/1.1 client and server work, WebSockets, CONNECT tunnels and TLS are done by hyper/rustls in Rust. The kj HTTP *interfaces* (kj::HttpClient, HttpService, HttpServer, WebSocket, HttpHeaders) stay the vocabulary workerd is written against; their implementations are replaced by symbol override, the same pattern the I/O layer uses, so product code is unchanged. kj-http split (capnproto side, companion commit): kj-http-types (the header table/object model, interface defaults, url) vs kj-http-impl (kj's HTTP codec, client, server, WebSocketImpl, pipes, adapters). Under rust, kj-http-impl is excluded from the link and the rust-io-hermeticity aspect forbids it; src/workerd/util/kj-http-tokio.c++ defines the kj:: symbols over: - kj-hyper (src/rust/cxx/kj-hyper): hyper client (host:port with TLS and peer filter, and a single-connection client over any kj stream), hyper server fused inline on the KJ loop (one fixed-point poll per connection, no cross-task channels), WebSocket sessions with permessage-deflate, an in-memory WebSocket pipe (kj::newWebSocketPipe) with kj's exact rendezvous/close/abort/pump-adoption semantics, CONNECT tunnels, Upgrade requests (used by the Docker container path), rustls client/server TLS, and the I/O-stall watchdog. - kj-rs-http (src/rust/cxx/kj-rs-http): the off-wire kj vocabulary codec (HttpHeaders serialize/tryParse, method/range parsing) hand-ported for byte parity with kj's storage scheme, plus the kj stream/service bridges. - The shim also carries kj-shape ports of the pooled network clients (per-address idle pool with idleTimeout, per-host cache over the restricted kj::Network, startTls), the client<->service shape adapters, and the WebSocket::pumpTo defaults. server/ collapses back to upstream form: the fd-inspection listener split, external/network channel dichotomy and http-client-backend seam are gone; server.c++ differs from upstream only by the tls-network seam (rustls-backed kj::SecureNetworkWrapper for listeners and outbound). container-client and fallback-service run on hyper with no C++ fallback. Serialization parity with kj (asserted byte-for-byte by server-test): title-case header names, original spellings via a hyper HeaderCaseMap patch, framing header in kj's header-table position, ordered 101 heads, kj's drain rule (idle+clean closes, partial request is served with Connection: close), Connection: close on error-handler responses, no entity-body on bodyless GET/HEAD. Bridge fixes found along the way: serve-side rendezvous cells (response head, streaming body ring) now register wakers so heads/chunks produced by the service on its own KJ event reach the wire without socket activity; stream-client byte pumps are owned by a kj::Promise member of the C++ client rather than a detached task (fixes a use-after-free write through the borrowed stream and armed events leaking past EventLoop teardown); the port's tokio event_interval is pinned below poll()'s yield budget so every WaitScope::poll() turns the reactor. The crate patch annotations (http/httparse lenient header values) lost in the carve are restored, plus patches/rust/hyper-public-header-case-map.patch. State: cxx server-test fully passes; rust server-test 88/90 (the two remaining share one cause: ready work in flight through tokio hops is invisible to kj's idle detection, so Worker hang-abort can fire mid-operation); hermeticity OK. Requires the companion capnproto kj-http split (kj-http-types target). Hot promises at the kj seam: on this kj-rs a bridged promise is cold (the Rust future runs only once the promise is first polled), while every kj interface these wrappers implement is hot -- kj-side callers (WebSocket::couple, the api WebSocket state machine, the hibernation manager, kj's header queue) sequence their own teardown on the operation being in flight when the method returns. hot() (hyper-http.h) starts each bridged future where it crosses a kj interface: the WebSocket pipe ends and hyper sessions (send/close/whenAborted/ pumpTo/tryPumpFrom), body/tunnel/sink streams (read/write/ whenWriteDisconnected), HyperHttpClient::request, HyperHttpServer::serve, the default WebSocket pump, and the held stream-client pump task. Left cold, the extra event hop reorders the hibernatable WebSocket Close hand-off enough that websocket-hibernation aborts in the DO's close handler on most runs (libmalloc reports the Close reason buffer as not allocated; no double free or write-after-free was attributable with zone hooks, watchpoints, guard malloc or zero-on-free checks; the C++ backend and the same binary under mimalloc pass). Once kj-rs promises are eager by default (#7010), hot() is a no-op to sweep. Rebased onto main with Part 2 (#7013) and the same-thread waker arm (#7349) merged and capnproto repinned to d1ebde0 (the kj-http split rebased onto it as dlapid/kjHttpSplit 61c00a8c), under the re-cut Part 3 (rust backend default, hermeticity gate + link check). kj-rs-io's stream is an Arc with a runtime-id owner now, so the take primitive is a Mutex<Option<Arc>> that goes hollow; its unwrap hook and the foreign-stream ends moved to kj-rs-io/unwrap.h so bridge.h stays free of <kj/async-io.h> (Windows: windows.h vs kj/compat/http.h); PeerFilter is atomically refcounted (kj::Arc); kj-hyper is back on tokio::task::spawn_blocking / current_handle().enter() since LoopRuntime was dropped; main's new TcpListener::run goes through the tls-network seam like HttpListener::run. kj-rs-http is clean under --config=lint. Serve tier: the native-serving entry points Part 2 no longer ships (serve_kj_stream / take_kj_socket, ServeIo, the stream pump) live here as kj-hyper's serve module, with their C++-driven tests (kj-hyper/tests:serve-test) and Rust echo fixtures. The handoff is consuming: kj-hyper downcasts the owned kj::AsyncIoStream, releases the kj-rs-io wrapper's native stream and destroys the wrapper (kj-stream.h), and TokioStream::into_socket takes the socket or hands the stream back -- re-wrapped, untouched -- while an operation still holds a share of it. kj-rs-io grows only that into_socket, a release() on the C++ wrapper and the Socket re-export; no hollow state, no lock in the stream's hot path. The refcounted read/write ends that let a foreign kj stream be pumped through two exclusive borrows are kj-hyper's too. kj-hyper's bridge aliases kj::AsyncIoStream and kj_rs_io::TokioStream directly, so the alias transmute between the two crates' stream types is gone. The pump for foreign streams is a rendezvous, not a buffer: a consumer write completes only once the pump has written it into the kj stream, a consumer shutdown becomes shutdownWrite() after the last write, and dropping the consumer end cancels an in-flight kj write -- the semantics of writing to a kj stream directly. tokio::io::duplex let writes complete into a buffer, so a kj::WebSocket over a pumped stream reported sends done that the peer never saw and the pump kept writing after its owner was destroyed (server-test "blockConcurrencyWhile throws after send"). Depends on kj-rs arming a poll event directly when a stored waker is woken on its own thread (#7349): the WebSocket pipe and the serve-side head/body rendezvous hand values between bridged futures through stored wakers and rely on the receiver being re-polled before a KJ event queued after the wake runs. On the previous ArcWaker that wake was a cross-thread fulfillment dispatched only once the loop went idle, and websocket-hibernation and the stream-client server-tests failed deterministically. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

danlapid requested review from a team as code owners September 13, 2026 22:37

danlapid force-pushed the dlapid/rustIoWakerCell branch from c0148c1 to 3dd9867 Compare September 13, 2026 22:50

danlapid requested review from a team as code owners September 13, 2026 22:50

danlapid requested review from jamesopstad and removed request for a team September 13, 2026 22:50

danlapid changed the base branch from dlapid/rustIoPart2 to main September 13, 2026 22:50

danlapid force-pushed the dlapid/rustIoWakerCell branch from 3dd9867 to f7844f8 Compare September 14, 2026 00:00

danlapid changed the title ~~kj-rs: same-thread waker cells (subset of #7010)~~ kj-rs: arm the poll event directly on a same-thread stored-waker wake Sep 14, 2026

danlapid force-pushed the dlapid/rustIoWakerCell branch 2 times, most recently from 4d73e10 to cf44c68 Compare September 15, 2026 01:56

danlapid mentioned this pull request Sep 15, 2026

kj-rs-io: KJ async I/O interfaces backed by tokio #7013

Merged

danlapid force-pushed the dlapid/rustIoWakerCell branch from cf44c68 to 3b427bc Compare September 15, 2026 02:19

jasnell approved these changes Sep 15, 2026

View reviewed changes

jamesopstad requested review from NuroDev and removed request for jamesopstad September 15, 2026 13:45

harrishancock reviewed Sep 15, 2026

View reviewed changes

Comment thread src/rust/cxx/kj-rs/waker.h Outdated

harrishancock approved these changes Sep 15, 2026

View reviewed changes

danlapid force-pushed the dlapid/rustIoWakerCell branch from 3b427bc to 6cc18f8 Compare September 15, 2026 18:27

danlapid force-pushed the dlapid/rustIoWakerCell branch from 6cc18f8 to 71c96cb Compare September 15, 2026 18:49

danlapid merged commit 355237e into main Sep 15, 2026
45 of 48 checks passed

danlapid deleted the dlapid/rustIoWakerCell branch September 15, 2026 20:21

Navigation Menu

kj-rs: arm the poll event directly on a same-thread stored-waker wake - #7349

kj-rs: arm the poll event directly on a same-thread stored-waker wake#7349
danlapid merged 1 commit into
mainfrom
dlapid/rustIoWakerCell

danlapid commented Sep 13, 2026 •

edited

Loading

Uh oh!

harrishancock left a comment •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

danlapid commented Sep 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

harrishancock left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

danlapid commented Sep 13, 2026 •

edited

Loading

harrishancock left a comment •

edited

Loading