Cluster workerd processes over a shared bucket

deps/rust/Cargo.toml

@@ -77,6 +77,8 @@ pico-args = "0"
 rand = { version = "0.9", default-features = false, features = ["os_rng", "thread_rng"] }
 ruff_python_ast = { git = "https://github.com/astral-sh/ruff", tag = "0.12.1" }
 ruff_python_parser = { git = "https://github.com/astral-sh/ruff", tag = "0.12.1" }
+# cluster: SigV4's SHA-256 and HMAC, and its peer MACs.
+ring = { version = "0.17", default-features = false }
 # Use rust_icu's checked-in bindings for the Chromium ICU version linked by workerd. ICU itself is
 # supplied by Bazel, so bindgen, icu-config, and rust_icu's library-linking features stay disabled.
 rust_icu_sys = { version = "5.8", default-features = false, features = ["renaming", "icu_version_in_env"] }

deps/rust/cargo-bazel-lock.json

@@ -2,7 +2,7 @@
   "binary_crates": [
     "capnpc 0.27.0"
   ],
-  "checksum": "7c291f282fd54549d51644602320eca9c686cea67d071b55fb7e6b84270a7b27",
+  "checksum": "4f7c905fb454b3d367e5ae35edc0a5d3b38300d86a842ca721465a1cb99a6cc0",
   "conditions": {
     "aarch64-apple-darwin": [
       "aarch64-apple-darwin"
@@ -3018,6 +3018,10 @@
             "id": "rand 0.9.5",
             "target": "rand"
           },
+          {
+            "id": "ring 0.17.14",
+            "target": "ring"
+          },
           {
             "id": "ruff_python_ast 0.0.0",
             "target": "ruff_python_ast"
@@ -16018,6 +16022,7 @@
     "proc-macro2 1.0.107",
     "quote 1.0.47",
     "rand 0.9.5",
+    "ring 0.17.14",
     "ruff_python_ast 0.0.0",
     "ruff_python_parser 0.0.0",
     "rust_icu_sys 5.8.0",

src/workerd/server/AGENTS.md

@@ -7,6 +7,7 @@ The `workerd` binary and its server. `:workerd` (`workerd.rs`, this package) is
 - `cli/` -- the `workerd-cli` crate (declared in this package's `BUILD.bazel`, sources globbed from `cli/`): parses the command line (clap), produces the encoded config (schema files are compiled by this package's `config-compiler.c++`), handles `compile`, `--watch` and re-exec, and hands `serve`, `test` (which `fuzzilli` and `make-pyodide-baseline-snapshot` are built on) and `bench` to the server crate's `entry` module.
 - `server/` -- the `workerd-server` crate: the single-tenant server. It interprets the config, builds the service graph, binds the sockets, runs the listeners, actors and dynamic workers, drains on SIGTERM, and runs `workerd test` and `workerd bench`. `Error` is `cxx::KjError`; `bridge.rs` is the one module that allows `unsafe`.
 - `factory/` -- C++: `WorkerFactory` (`worker-factory*`), everything that needs the isolate, driven by the server through the cxx bridge in `server/bridge.rs`; and `bootstrap.{h,c++}`, the C++ half of a command. `with_process_context` sets up the process's logging (the JSON logger, when the config asks for structured logging) on the C++ stack around the Rust command; `bootstrap` sets up perfetto, autogates and V8 on the Rust-owned event loop and builds the `WorkerFactory`, which owns them, from the encoded config and the command-line options for `server/entry.rs`, which owns the loop (`kj_rs_tokio::Runtime`) and blocks on `server_serve` / `server_test` / `server_bench`.
+- `cluster/` -- the `cluster` crate and, in `cluster/logic/`, `cluster-logic`: cluster mode (`Config.cluster`, experimental), Durable Objects whose shared state is an S3-compatible bucket. Leases, ownership, LTX capture and restore; pure Rust, driven by `server/cluster.rs`. `cluster-logic` holds decisions only, with no dependencies to do I/O with.
 
 This package keeps the config schema (`workerd.capnp`, `compiled-bindings.capnp`), the config compiler, the `WorkerdApi` and the C++ libraries the factory is built from (alarm scheduler, channel tokens, container client, facet tree index, fallback service, actor IDs, JSON logger, Pyodide, V8 platform), and their C++ unit tests.
 

src/workerd/server/channel-token-test.c++

@@ -414,6 +414,24 @@ KJ_TEST("actor channel tokens") {
       "channel token type mismatch", handler.decodeActorClassChannelToken(Usage::RPC, token));
 }
 
+KJ_TEST("instances of a cluster redeem each other's channel tokens") {
+  MockResolver senderResolver, receiverResolver, strangerResolver;
+  auto key = "test-cluster-shared-secret-of-32-bytes"_kj;
+  ChannelTokenHandler sender(senderResolver, key);
+  ChannelTokenHandler receiver(receiverResolver, key);
+  ChannelTokenHandler stranger(strangerResolver, "another-cluster-secret-of-32-bytes"_kj);
+
+  const byte actorId[] = {12, 34, 56, 78};
+  auto token = sender.encodeActorChannelToken(
+      Usage::RPC, "foo-namespace", actorId, "my-actor"_kj, Persistent::NO);
+  auto channel =
+      receiver.decodeSubrequestChannelToken(Usage::RPC, token).downcast<MockActorChannel>();
+  expectActorChannel(*channel, "foo-namespace", actorId, "my-actor"_kj);
+
+  KJ_EXPECT_THROW_MESSAGE(
+      "failed authentication", stranger.decodeSubrequestChannelToken(Usage::RPC, token));
+}
+
 KJ_TEST("channel token with nested channels (all synchronous)") {
   MockResolver resolver;
   ChannelTokenHandler handler(resolver);

src/workerd/server/channel-token.c++

@@ -9,6 +9,7 @@
 #include <workerd/util/entropy.h>
 
 #include <openssl/evp.h>
+#include <openssl/hmac.h>
 #include <openssl/sha.h>
 
 #include <capnp/serialize-packed.h>
@@ -22,8 +23,18 @@
 
 namespace workerd::server {
 
-ChannelTokenHandler::ChannelTokenHandler(Resolver& resolver): resolver(resolver) {
-  getEntropy(tokenKey);
+ChannelTokenHandler::ChannelTokenHandler(Resolver& resolver, kj::Maybe<kj::StringPtr> clusterKey)
+    : resolver(resolver) {
+  KJ_IF_SOME(key, clusterKey) {
+    // Labeled so this key never equals another one derived from the cluster key.
+    auto label = "workerd channel token v1"_kj;
+    uint outLen = 0;
+    KJ_ASSERT(HMAC(EVP_sha256(), label.begin(), label.size(), key.asBytes().begin(), key.size(),
+                  tokenKey, &outLen) != nullptr);
+    KJ_ASSERT(outLen == sizeof(tokenKey));
+  } else {
+    getEntropy(tokenKey);
+  }
 
   SHA256_CTX ctx{};
   KJ_ASSERT(SHA256_Init(&ctx));

src/workerd/server/channel-token.h

@@ -60,7 +60,8 @@ class ChannelTokenHandler: public kj::PtrTarget {
         IoChannelFactory::ChannelTokenUsage usage) = 0;
   };
 
-  explicit ChannelTokenHandler(Resolver& resolver);
+  // Instances of a cluster share `clusterKey`, so each decodes the others' tokens.
+  explicit ChannelTokenHandler(Resolver& resolver, kj::Maybe<kj::StringPtr> clusterKey = kj::none);
 
   // Helpers to implement `IoChannelFactory::{SubrequestChannel,ActorClassChannel}::getToken()`.
   //

src/workerd/server/cli/args-test.rs

@@ -65,6 +65,8 @@ fn serve_options() {
         "--async-trace=trace.ndjson",
         "--async-trace-stacks=8",
         "--async-trace-promises",
+        "--cluster-addr=0.0.0.0:7000",
+        "--cluster-advertise=10.0.0.4:7000",
     ])
     .command
     else {
@@ -85,6 +87,8 @@ fn serve_options() {
     assert_eq!(serve_or_test.async_trace_stacks, Some(8));
     assert!(serve_or_test.async_trace_promises);
     assert_eq!(serve.control_fd, Some(3));
+    assert_eq!(serve.cluster_address.as_deref(), Some("0.0.0.0:7000"));
+    assert_eq!(serve.cluster_advertise.as_deref(), Some("10.0.0.4:7000"));
     assert_eq!(
         serve.socket_addr_overrides,
         [Override {

src/workerd/server/cli/args.rs

@@ -55,7 +55,13 @@ pub enum Command {
     /// run the server
     #[command(
         long_about = "Serve requests based on a config.",
-        after_long_help = "Serves requests based on the configuration specified in <config-file>."
+        after_long_help = "Serves requests based on the configuration specified in <config-file>.\n\n\
+            Cluster mode (`Config.cluster`, with --experimental): instances that share an \
+            S3-compatible bucket serve Durable Objects together. Each `cluster.bucketOptions` \
+            setting falls back to an environment variable: AWS_ENDPOINT_URL, AWS_REGION, \
+            AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN, or \
+            AWS_CREDENTIALS_FILE. The cluster port must stay on a private network; see \
+            --cluster-addr."
     )]
     Serve {
         #[command(flatten)]
@@ -325,6 +331,21 @@ pub struct ServeArgs {
     #[arg(long, value_name = "fd", value_parser = parse_control_fd)]
     pub control_fd: Option<u32>,
 
+    /// Bind the cluster's peer port to <addr> instead of the config's `cluster.address`. The
+    /// port gives full access to every service in the process to holders of the cluster key, and
+    /// its traffic is not encrypted: only the cluster's instances should reach it, on a private
+    /// network (a VPC or Tailscale). A public IP is refused unless the config sets
+    /// `cluster.unsafePublicAddress`; a wildcard (0.0.0.0) listens on every interface, so
+    /// firewall the port.
+    #[arg(long = "cluster-addr", value_name = "addr")]
+    pub cluster_address: Option<String>,
+
+    /// Advertise <addr> to the cluster's other instances instead of the config's
+    /// `cluster.advertise` or the bound address. A public IP is refused unless the config sets
+    /// `cluster.unsafePublicAddress`.
+    #[arg(long = "cluster-advertise", value_name = "addr")]
+    pub cluster_advertise: Option<String>,
+
     /// Listen on the specified address for debug RPC connections. This exposes a privileged
     /// interface that allows access to all services in the process. For use by miniflare and local
     /// development only.

src/workerd/server/cli/lib.rs

@@ -467,6 +467,8 @@ fn serve_options<T: CommandFactory>(
         socket_fds,
         control,
         debug_port: serve.debug_port,
+        cluster_address: serve.cluster_address,
+        cluster_advertise: serve.cluster_advertise,
         ..run_options(serve_or_test)
     };
     (options, inherited)

src/workerd/server/cluster/BUILD.bazel

@@ -0,0 +1,42 @@
+load("//:build/wd_capnp_library.bzl", "wd_capnp_library")
+load("//:build/wd_rust_crate.bzl", "wd_rust_crate")
+
+# Durable Objects whose shared state is an S3-compatible bucket. Pure Rust:
+# the server (//src/workerd/server/server) drives it.
+wd_rust_crate(
+    name = "cluster",
+    proc_macro_deps = ["@crates_vendor//:async-trait"],
+    # The capture's tests drive a real SQLite.
+    test_deps = ["@sqlite3"],
+    # The capture's tests hand SQLite Rust callbacks, which a downstream
+    # build's cfi-icall rejects.
+    test_tags = ["no-downstream"],
+    visibility = ["//src/workerd/server:__subpackages__"],
+    deps = [
+        ":frames_capnp_rust",
+        "//src/workerd/server/cluster/logic:cluster-logic",
+        "@crates_vendor//:base64",
+        "@crates_vendor//:bytes",
+        "@crates_vendor//:capnp",
+        "@crates_vendor//:flate2",
+        "@crates_vendor//:futures",
+        "@crates_vendor//:http",
+        "@crates_vendor//:http-body",
+        "@crates_vendor//:httpdate",
+        "@crates_vendor//:hyper-util",
+        "@crates_vendor//:nix",
+        "@crates_vendor//:ring",
+        "@crates_vendor//:rustls",
+        "@crates_vendor//:rustls-platform-verifier",
+        "@crates_vendor//:serde",
+        "@crates_vendor//:serde_json",
+        "@crates_vendor//:thiserror",
+        "@crates_vendor//:tokio",
+        "@crates_vendor//:tokio-rustls",
+        "@crates_vendor//:tower-service",
+        "@crates_vendor//:tracing",
+    ],
+)
+
+# The format of an object's log in the bucket.
+wd_capnp_library(src = "frames.capnp")