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.

Make workerd's entry point Rust workerd now starts from a Rust main(). Rust owns the process: the command line, where the config comes from, --watch and the reload it triggers, and the compiled-binary format. C++ keeps the two things only C++ can do -- compiling a schema-file config with capnp::SchemaParser and running V8 and the Server -- behind a small bridge, and both halves live together in src/workerd/server. The C++ that remains is the Server's own surface; going lower means changing the Server. Compiled binaries embed the config as a standard encoded message rather than a flat single-segment one; a compiled binary always carries the workerd that reads it, so no deployed binary is affected. Two command-line corners change: the accidental `--p` spelling of --perfetto-trace is gone, and naming one socket in both --socket-addr and --socket-fd is an error instead of whichever came last winning. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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>

kj-rs-io: "loopback:" addresses, used by `workerd test` The "loopback:" addresses `workerd test` uses to exercise the network stack inside one process move from the CLI's own kj::Network wrapper into kj-rs-io, alongside the rest of the tokio I/O layer. A loopback address names a queue: connect() makes a socket pair, queues one end and returns the other, and a receiver's accept() takes from the queue. Every TokioNetwork carries a LoopbackRegistry, shared down its restrictPeers() chain and off unless enableLoopback() is called, so the production address grammar is unchanged; a name belongs to the loop that first parsed it. `workerd test` enables it on the provider's network, and the wrapper goes. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

kj-rs: drop a KjOwn<T> through kj::Own<T>'s own destructor Rust dropped every KjOwn<T> as a kj::Own<void>. kj::Own<T>::~Own() hands the disposer the address of the complete object (dynamic_cast<void*> for a polymorphic T); kj::Own<void> passes the T subobject's address unchanged. When T is a base at a nonzero offset of the object it points into (multiple inheritance: RequestObserverWithTracer's WorkerInterface base, kj::NullStream's AsyncOutputStream base), the object was deleted at the wrong address. The cxx bridge now generates a per-type drop, as it does for UniquePtr<T>: a C++ function running kj::Own<T>::~Own() and a Rust OwnTarget implementation calling it, for every T the bridge declares and holds in a KjOwn. A bridge that only aliases a type asks its declaring bridge for the implementation with `impl KjOwn<T> {}`. The primitive types a KjOwn may hold are implemented in kj-rs itself. The new own-test reproduces the miscomputed address with a two-base class handed to Rust as a kj::Own of its second base. 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>

Make Bonk reviews reliable and context-aware About 20% of automatic Bonk reviews hung until the 30-minute job timeout and then posted "Bonk workflow was cancelled". opencode was waiting on a permission prompt for a path outside the checkout, which nothing can answer in CI. Separately, pushes after the first review were never re-reviewed, and manual /bonk re-reviews started from scratch each time. Reliability: - Deny reads outside the checkout instead of prompting. - Give ask-bonk a 25-minute timeout, inside the job timeout, so a stuck run finishes on its own. - Rename bigbonk.yml's workflow so it no longer shares Bonk's concurrency group and replaces its pending runs. Re-reviews: - Run ask-bonk from a SHA-pinned fork (danlapid/ask-bonk@4768a8b) until Cloudflare-Studio/ask-bonk#228 lands upstream. - Review every push. Re-reviews see Bonk's previous review, its threads, and only the author's changes since then. - Keep one Bonk summary comment per PR, edited in place on every review, with findings posted inline. Bonk replies to and resolves its own threads once they are fixed or declined. - Start every review with a verdict line, and hand findings to ask-bonk instead of posting them from the model. Review quality: - Run specialist reviewers in parallel before the main review, which then acts as judge. .github/bonk/specialists/ adds workerd's own: security, memory and thread safety, API compatibility, compat flags, JSG/GC, KJ style, design simplicity and Rust-first. - Tag findings by severity. Only blocking and warning findings go inline; the rest are listed in the summary. - Raise the job timeout to 45 minutes to fit the specialists, and skip types/generated-snapshot/. Cleanup: - Use the opencode nightly (opencode_dev), and drop the version pin it overrode along with the weekly updater for that pin. - Trigger on /bonk in review summaries, fetch full history for /bonk, review drafts once they are ready, and skip Dependabot PRs. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Make Bonk reviews precise Tune Bonk from its first live reviews and from maintainer pushback, so that a finding reads as "oh yeah, that's right" rather than noise. - Keep the rules every review follows in one place, .github/bonk/specialists/SHARED.md: never report what CI checks (builds, lints, formatting, tests) or performance micro-costs, hold test code only to "can it pass while the code is broken, is it flaky", and mark hypothetical or follow-up issues info at most. - Turn off the built-in performance specialist, and add workerd versions of the correctness, tests and docs specialists. - Stop asking for compatibility flags where they are not needed: code behind $experimental flags, Node.js and spec conformance fixes, unobservable changes, and additive exports of import-only modules. - Tighten specialist calibration from the first reviews: copyright headers only where siblings have them, compare with kj before flagging protocol deviations, test-only unsafety capped at info, and verify design suggestions before raising them. - Drop the jokes and boilerplate from review summaries. - Correct src/rust/AGENTS.md: a C++ exception through an infallible extern "C++" shim becomes a Rust panic, not an abort. - Pin danlapid/ask-bonk@4f53fb3 (Cloudflare-Studio/ask-bonk#228), which fixes the pipeline issues those reviews exposed and adds SHARED.md and disabling built-in specialists. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Defer in-memory alarm task cancellation With in-memory Durable Object storage, alarms are run by a task owned by the actor's hooks. The alarm handler can update the alarm, and runAlarm() clears it after the handler finishes, both of which call updateAlarmInMemory() from within that task. Replacing the task there destroyed it while one of its callbacks was still running, failing the "Promise callback destroyed itself" assertion and aborting the process. Move the displaced task into a TaskSet that drops it on the next turn of the event loop instead. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>