Add the Rust I/O backend: tokio event loop + tokio sockets

BUILD.bazel

@@ -1,7 +1,7 @@
 load("@aspect_rules_js//js:defs.bzl", "js_library")
 load("@aspect_rules_js//npm:defs.bzl", "npm_link_package")
 load("@bazel_skylib//lib:selects.bzl", "selects")
-load("@bazel_skylib//rules:common_settings.bzl", "bool_flag")
+load("@bazel_skylib//rules:common_settings.bzl", "bool_flag", "string_flag")
 load("@npm//:capnp-es/package_json.bzl", capnp_es_bins = "bin")
 load("@npm//:defs.bzl", "npm_link_all_packages")
 load("//:build/wd_cc_embed.bzl", "wd_cc_embed")
@@ -128,6 +128,44 @@ bool_flag(
     build_setting_default = True,
 )
 
+# =============================================================================
+# I/O backend selection: a build/compile/link-time hermeticity guarantee.
+#
+#   --//:io_backend=rust  (default) -- the rust I/O layer: the process event loop is tokio
+#                         (kj-rs-tokio) and every socket/stream is tokio-backed (kj-rs-io); the
+#                         C++ layers above the streams (kj-http, kj-tls, capnp-rpc) are unchanged
+#                         and run over those tokio streams. In this config the forbidden concrete
+#                         C++ OS-I/O target must be ABSENT from the link, enforced by the
+#                         build-graph aspect in //build:rust_io_backend.bzl (whose forbidden
+#                         set grows as later migration stages move more of the stack to rust).
+#   --//:io_backend=cxx   -- workerd's I/O is the concrete C++ stack end to end, including the
+#                         kj OS event loop and sockets. Byte-identical to the pre-migration
+#                         build.
+#
+# Code that must differ per backend keys off the WORKERD_RUST_IO_BACKEND define (see
+# //src/workerd/util:setup-async-io) or off the config_settings below in a select().
+string_flag(
+    name = "io_backend",
+    build_setting_default = "rust",
+    values = [
+        "cxx",
+        "rust",
+    ],
+    visibility = ["//visibility:public"],
+)
+
+config_setting(
+    name = "io_backend_rust",
+    flag_values = {":io_backend": "rust"},
+    visibility = ["//visibility:public"],
+)
+
+config_setting(
+    name = "io_backend_cxx",
+    flag_values = {":io_backend": "cxx"},
+    visibility = ["//visibility:public"],
+)
+
 config_setting(
     name = "set_dead_strip",
     flag_values = {"dead_strip": "True"},

build/deps/deps.jsonc

@@ -36,7 +36,9 @@
       "type": "github_tarball",
       "owner": "capnproto",
       "repo": "capnproto",
-      "branch": "v2",
+      // Interim: v2 plus kj::Rc::disown()/reown() (mirror of kj::Arc's), needed by the Rust
+      // I/O bridge's waker. Switch back to "v2" once the upstream PR merges.
+      "branch": "dlapid/RcdisReOwn",
       "extra_strip_prefix": "/c++"
     },
     // We want to avoid version skew with v8, so we use identical versions. Keep this

build/deps/gen/deps.MODULE.bazel

@@ -27,10 +27,10 @@ bazel_dep(name = "brotli", version = "1.2.0.bcr.1")
 # capnp-cpp
 http.archive(
     name = "capnp-cpp",
-    sha256 = "6753378bd099029cb2830fecd32dd158218019e459ffd3c8e379cbf025906eb8",
-    strip_prefix = "capnproto-capnproto-a1cd1c4/c++",
+    sha256 = "afaf9a84342533e12c8f8c67adcfeb59c74ba86f04c15499736de453c3edea22",
+    strip_prefix = "capnproto-capnproto-1ac3c72/c++",
     type = "tgz",
-    url = "https://github.com/capnproto/capnproto/tarball/a1cd1c4b3d241b77478035a6ccad8b0fb587d444",
+    url = "https://github.com/capnproto/capnproto/tarball/1ac3c7259ddf3917d7c5947978e1f97bf328b699",
 )
 use_repo(http, "capnp-cpp")
 

build/kj_test.bzl

@@ -1,12 +1,14 @@
 load("@rules_cc//cc:cc_binary.bzl", "cc_binary")
 load("@rules_shell//shell:sh_test.bzl", "sh_test")
+load("//:build/rust_io_backend.bzl", "rust_io_backend_local_defines")
 
 def kj_test(
         src,
         data = [],
         deps = [],
         tags = [],
         size = "medium",
+        local_defines = [],
         **kwargs):
     test_name = src.removesuffix(".c++")
     binary_name = test_name + "_binary"
@@ -18,6 +20,11 @@ def kj_test(
             "@capnp-cpp//src/kj:kj-test",
             "//build/deps:linkopts_default",
         ] + deps,
+        # Under --//:io_backend=rust the native kj::setupAsyncIo()/UnixEventPort aren't linked, so
+        # tests that need an event loop swap to the tokio backend behind
+        # `#if WORKERD_RUST_IO_BACKEND_RUST`. Supplied to every kj_test TU here (cheap, harmless
+        # where unused) instead of repeating it per target.
+        local_defines = local_defines + rust_io_backend_local_defines(),
         linkstatic = select({
             "@platforms//os:linux": 0,
             "//conditions:default": 1,

build/rust_io_backend.bzl

@@ -0,0 +1,165 @@
+"""Build-graph hermeticity check + shared toggles for workerd's Rust I/O backend.
+
+An aspect walks the transitive dependency graph of a guarded target and, in the rust I/O
+config, fails ANALYSIS if the graph reaches a forbidden concrete C++ I/O library (the ones
+the rust layer replaces), printing one example dependency path to each offender.
+
+Why a graph assertion and not a link check: if kj-async-os sneaks back into the rust-config
+graph, the failure mode is a DOUBLE-DEFINITION (kj::setupAsyncIo and kj::UnixEventPort's
+members are defined by both the native TUs and the tokio shim), and with static archives the
+winner is link-order-dependent -- possibly a loud duplicate-symbol error, possibly the wrong
+event loop silently winning. Undefined-symbol errors only backstop the opposite (absent-lib)
+direction. This aspect catches the double-definition direction deterministically, at analysis.
+
+IMPORTANT -- the gate is opt-in: //src/workerd/server:rust-io-hermeticity is tagged `manual`,
+so `bazel build //...` never runs it. The rust-config CI lane must build it EXPLICITLY for
+the guarantee to hold.
+
+The forbidden set grows in lockstep with what `=rust` means per migration stage (v1: the
+tokio event loop + I/O; later stages append kj-http/kj-tls, then capnp-rpc). Deliberately
+ALLOWED today: @capnp-cpp//src/kj:kj-async-core and :kj-async-io (the abstract Promise and
+stream/Network layers the rust backend itself is built on), kj-http/kj-tls/capnp-rpc (still
+the only implementation in both configs; they consume abstract streams, no OS I/O), and the
+kj-gzip/kj-brotli codecs (not a transport).
+"""
+
+visibility("public")
+
+def rust_io_backend_local_defines():
+    """local_defines for TUs that `#if WORKERD_RUST_IO_BACKEND_RUST` (the declared seam points).
+
+    Kept per-target rather than a repo-global define so the seam stays enumerable: the only
+    places allowed to diverge per backend are the targets that ask for this.
+    """
+    return select({
+        "//:io_backend_rust": ["WORKERD_RUST_IO_BACKEND_RUST=1"],
+        "//conditions:default": [],
+    })
+
+# Forbidden concrete C++ I/O targets, as "//package:target" label suffixes (suffix-matched so
+# bzlmod repo-name canonicalization doesn't have to be spelled out). The single source of truth.
+_FORBIDDEN = [
+    # The kj OS event loop / socket layer (setupAsyncIo, UnixEventPort/Win32IocpEventPort),
+    # replaced by kj-rs-tokio + kj-rs-io.
+    "//src/kj:kj-async-os",
+]
+
+RustIoForbiddenInfo = provider(
+    doc = "One example dependency path to each forbidden C++ I/O target a subgraph reaches.",
+    fields = {
+        "paths": "dict of forbidden-label-suffix -> example path (list of label strings)",
+    },
+)
+
+def _forbidden_suffix(label_str):
+    for suffix in _FORBIDDEN:
+        if label_str.endswith(suffix):
+            return suffix
+    return None
+
+def _rust_io_forbidden_aspect_impl(target, ctx):
+    label_str = str(target.label)
+    paths = {}
+
+    hit = _forbidden_suffix(label_str)
+    if hit != None:
+        paths[hit] = [label_str]
+
+    # Scan ctx.rule.attr generically (attr_aspects = ["*"]) rather than enumerating
+    # deps/implementation_deps/etc. per rule kind.
+    for attr_name in dir(ctx.rule.attr):
+        value = getattr(ctx.rule.attr, attr_name)
+        dep_targets = []
+        if type(value) == "list":
+            for item in value:
+                if type(item) == "Target":
+                    dep_targets.append(item)
+        elif type(value) == "Target":
+            dep_targets.append(value)
+
+        for dep in dep_targets:
+            if RustIoForbiddenInfo in dep:
+                for forbidden, subpath in dep[RustIoForbiddenInfo].paths.items():
+                    if forbidden not in paths:
+                        paths[forbidden] = [label_str] + subpath
+
+    return [RustIoForbiddenInfo(paths = paths)]
+
+rust_io_forbidden_aspect = aspect(
+    implementation = _rust_io_forbidden_aspect_impl,
+    attr_aspects = ["*"],
+    doc = "Propagates RustIoForbiddenInfo up the dependency graph.",
+)
+
+def _rust_io_hermeticity_impl(ctx):
+    info = ctx.attr.target[RustIoForbiddenInfo]
+    paths = info.paths
+
+    report = ctx.actions.declare_file(ctx.label.name + ".txt")
+
+    if ctx.attr.enforce and len(paths) > 0:
+        lines = [
+            "",
+            "Rust-I/O hermeticity FAILED for {} in the rust I/O config.".format(
+                str(ctx.attr.target.label),
+            ),
+            "",
+            "Its transitive dependency graph reaches {} forbidden concrete C++ ".format(len(paths)) +
+            "I/O target(s) that the rust I/O layer (the tokio event loop + tokio",
+            "sockets) is meant to keep off the build. Each is a real migration item; the",
+            "example dependency edge shows one path that pulls it in:",
+            "",
+        ]
+        for forbidden in sorted(paths.keys()):
+            path = paths[forbidden]
+            lines.append("  [FORBIDDEN] {}".format(forbidden))
+            lines.append("    reached via:")
+            for i, hop in enumerate(path):
+                lines.append("      {}{}".format("  " * i, hop))
+            lines.append("")
+        lines.append(
+            "Fix by removing the offending edge (migrate the code to the tokio-backed I/O layer,",
+        )
+        lines.append(
+            "or drop the forbidden dep from that target's deps under select(io_backend_rust)),",
+        )
+        lines.append(
+            "or -- if this is a deliberate remaining kj-mode/in-process site -- adjust the",
+        )
+        lines.append(
+            "forbidden set in build/rust_io_backend.bzl (a reviewable change).",
+        )
+        fail("\n".join(lines))
+
+    # Not enforcing (cxx config), or clean: emit a report artifact and a summary.
+    if len(paths) == 0:
+        summary = "rust-io-hermeticity: OK -- {} reaches 0 forbidden C++ I/O targets.".format(
+            str(ctx.attr.target.label),
+        )
+    else:
+        summary = ("rust-io-hermeticity: {} reaches {} forbidden C++ I/O target(s) " +
+                   "(NOT enforced in this config; --//:io_backend=rust would fail): {}").format(
+            str(ctx.attr.target.label),
+            len(paths),
+            ", ".join(sorted(paths.keys())),
+        )
+    ctx.actions.write(report, summary + "\n")
+    return [DefaultInfo(files = depset([report]))]
+
+rust_io_hermeticity = rule(
+    implementation = _rust_io_hermeticity_impl,
+    doc = "Fails analysis (when enforced) if `target` transitively depends on a forbidden " +
+          "concrete C++ I/O library, naming the offending edge.",
+    attrs = {
+        "target": attr.label(
+            mandatory = True,
+            aspects = [rust_io_forbidden_aspect],
+            doc = "The target whose transitive deps are checked (e.g. the workerd binary).",
+        ),
+        "enforce": attr.bool(
+            default = False,
+            doc = "Whether reaching a forbidden target fails analysis. Set from a " +
+                  "select() on //:io_backend_rust at the instantiation site.",
+        ),
+    },
+)

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", "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"