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

deps/rust/Cargo.toml

@@ -32,6 +32,7 @@ syn = { version = "2", features = ["full"] }
 ada-url = { version = "4", default-features = false, features = ["std"] }
 anyhow = "1"
 async-trait = { version = "0", default-features = false }
+bytes = "1"
 capnp = "0"
 capnpc = "0"
 capnp-rpc = "0"
@@ -48,9 +49,9 @@ ruff_python_parser = { git = "https://github.com/astral-sh/ruff", tag = "0.12.1"
 # param_extractor depends on unbounded_depth feature
 serde_json = { version = "1", features = ["unbounded_depth"] }
 serde = { version = "1", features = ["derive"] }
+socket2 = "0.6"
 thiserror = "2"
-# tokio is huge, let's enable only features when we actually need them.
-tokio = { version = "1", default-features = false, features = ["net", "rt", "rt-multi-thread", "sync", "time"] }
+tokio = { version = "1", default-features = false, features = ["io-util", "macros", "net", "rt", "rt-multi-thread", "signal", "sync", "time"] }
 tracing = { version = "0", default-features = false, features = ["std"] }
 swc_common = "25"
 swc_ts_fast_strip = "57"

src/rust/cxx/AGENTS.md

@@ -36,6 +36,10 @@ Bazel module, Cargo workspace, toolchain configuration, or external `workerd-cxx
 - `kj-rs-tokio/` — `TokioEventPort`: a `kj::EventPort` backed by a per-thread tokio
   `current_thread` runtime, plus `setupTokioAsyncIo()` (no I/O providers) and
   `kj_rs_tokio::spawn()`
+- `kj-rs-io/` — KJ async I/O interfaces over tokio (`kj::AsyncIoStream`, listeners,
+  `kj::Network`, providers, signals, file watching), `kj_rs_io::setupTokioAsyncIo()` as the
+  drop-in `kj::setupAsyncIo()` replacement, the stream unwrap fast path, and
+  `serve_kj_stream()` for Rust servers consuming KJ streams
 - `tests/` and `kj-rs/tests/` — Rust and C++ bridge integration tests
 - `tools/bazel/` — Bazel bridge-generation macro used by this component's tests
 

src/rust/cxx/kj-rs-io/BUILD.bazel

@@ -0,0 +1,79 @@
+load("@rules_rust//rust:defs.bzl", "rust_library", "rust_test")
+load("//:build/wd_cc_library.bzl", "wd_cc_library")
+load("//src/rust/cxx/tools/bazel:rust_cxx_bridge.bzl", "rust_cxx_bridge")
+
+wd_cc_library(
+    name = "kj-rs-io-lib",
+    srcs = glob(["*.c++"]),
+    hdrs = glob(["*.h"]),
+    include_prefix = "kj-rs-io",
+    linkstatic = select({
+        "@platforms//os:windows": True,
+        "//conditions:default": False,
+    }),
+    strip_include_prefix = "/src/rust/cxx/kj-rs-io",
+    visibility = ["//visibility:public"],
+    deps = [
+        ":bridge",
+        "//src/rust/cxx/kj-rs-tokio:kj-rs-tokio-lib",
+    ],
+)
+
+rust_library(
+    name = "kj-rs-io",
+    srcs = glob(["*.rs"]),
+    compile_data = glob(["*.h"]),
+    edition = "2024",
+    link_deps = [
+        ":bridge",
+        ":kj-rs-io-lib",
+    ],
+    target_compatible_with = select({
+        "@//build/config:no_build": ["@platforms//:incompatible"],
+        "//conditions:default": [],
+    }),
+    visibility = ["//visibility:public"],
+    deps = [
+        "//src/rust/cxx",
+        "//src/rust/cxx/kj-rs",
+        "//src/rust/cxx/kj-rs-tokio",
+        "@crates_vendor//:bytes",
+        "@crates_vendor//:libc",
+        "@crates_vendor//:socket2",
+        "@crates_vendor//:tokio",
+    ],
+)
+
+rust_test(
+    name = "kj-rs-io_test",
+    crate = "kj-rs-io",
+    edition = "2024",
+    target_compatible_with = select({
+        "@//build/config:no_build": ["@platforms//:incompatible"],
+        "//conditions:default": [],
+    }),
+)
+
+rust_cxx_bridge(
+    name = "bridge",
+    src = "ffi.rs",
+    hdrs = ["unwrap.h"],
+    include_prefix = "kj-rs-io",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//src/rust/cxx/kj-rs",
+        "@capnp-cpp//src/kj:kj",
+        # kj-rs-io IS the kj-side implementation of the rust I/O backend: it derives from the
+        # abstract kj::AsyncIoStream / kj::Network / kj::LowLevelAsyncIoProvider interfaces and
+        # reuses portable helpers (kj::newOneWayPipe, kj::CidrRange, the AsyncInputStream /
+        # AsyncOutputStream default-method vtable slots). Those live in the ABSTRACT layers
+        # :kj-async-core (Promise machinery) + :kj-async-io (abstract streams / Network / CIDR;
+        # no event loop). It does NOT call kj::setupAsyncIo / kj::UnixEventPort, so it never
+        # needs :kj-async-os (the concrete OS event loop) -- the tokio-backed EventPort
+        # (kj-rs-tokio) replaces it. A stray setupAsyncIo would be an undefined-symbol link
+        # error, the intended backstop.
+        "@capnp-cpp//src/kj:kj-async-core",
+        "@capnp-cpp//src/kj:kj-async-io",
+        "//src/rust/cxx:core",
+    ],
+)