Make self-hosted Durable Objects scalable across a cluster.

src/workerd/io/worker-interface.capnp

@@ -888,12 +888,27 @@ interface WorkerdDebugPort {
   # purposes.
   #
   # This interface is subject to change. It is intended for use by miniflare.
+  #
+  # This is also the same interface that workerd exports on its cluster port, which other workerd
+  # instances in the same cluster may connect to. See ClusterConfig in workerd.capnp for how to
+  # configure clustering. The cluster port is only intended to be used by other instances in the
+  # cluster, and does not implement a regular two-party RPC protocol. `ClusterRegistry` implements
+  # the `VatNetwork` intended to be used for this. Nobody else should try to connect to it.
+  #
+  # TODO(clustering): Once channel tokens have been extended to support actors, consider changing
+  #   the cluster interface with one that takes a channel token. This seems a bit cleaner and safer
+  #   than letting the client specify props and such directly.
+  # TODO(clustering): Automatically use encryption on non-localhost networks. We have an X25519
+  #   keypair for each node, no need for certs!
 
   getEntrypoint @0 (service :Text, entrypoint :Text, props :Frankenvalue)
               -> (entrypoint :WorkerdBootstrap);
   # Get direct access to a stateless entrypoint.
 
-  getActor @1 (service :Text, entrypoint :Text, actorId :Text) -> (actor :WorkerdBootstrap);
+  getActor @1 (service :Text, entrypoint :Text, actorId :Text, actorName :Text)
+           -> (actor :WorkerdBootstrap);
   # Get an actor (Durable Object) stub.
   # The actorId should be a hex string for Durable Objects or a plain string for ephemeral actors.
+  # `actorName` may optionally be specified for Durable Objects, if `actorId` is derived from the
+  # name.
 }

src/workerd/server/BUILD.bazel

@@ -212,6 +212,7 @@ wd_cc_library(
         ":channel-token_capnp",
         "//src/workerd/io",
         "//src/workerd/util:entropy",
+        "@ssl",
     ],
 )
 
@@ -228,6 +229,9 @@ wd_cc_library(
         ":alarm-scheduler",
         ":channel-token",
         ":channel-token_capnp",
+        ":cluster-lock",
+        ":cluster-registry",
+        ":cluster_capnp",
         ":container-client",
         ":facet-tree-index",
         ":fallback-service",
@@ -244,11 +248,45 @@ wd_cc_library(
         "//src/workerd/jsg",
         "//src/workerd/util:perfetto",
         "//src/workerd/util:websocket-error-handler",
+        "@capnp-cpp//src/capnp:capnp-rpc",
         "@capnp-cpp//src/kj/compat:kj-gzip",
         "@capnp-cpp//src/kj/compat:kj-tls",
     ],
 )
 
+wd_capnp_library(src = "cluster.capnp")
+
+wd_cc_library(
+    name = "cluster-registry",
+    srcs = ["cluster-registry.c++"],
+    hdrs = ["cluster-registry.h"],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":cluster_capnp",
+        "//src/workerd/util:ofd-lock",
+        "@capnp-cpp//src/capnp:capnp-rpc",
+        "@capnp-cpp//src/kj",
+        "@capnp-cpp//src/kj:kj-async",
+        "@ssl",
+    ],
+)
+
+wd_cc_library(
+    name = "cluster-lock",
+    srcs = ["cluster-lock.c++"],
+    hdrs = ["cluster-lock.h"],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":cluster-registry",
+        ":cluster_capnp",
+        "//src/workerd/io:worker-interface_capnp",
+        "//src/workerd/util:ofd-lock",
+        "@capnp-cpp//src/capnp:capnp-rpc",
+        "@capnp-cpp//src/kj",
+        "@capnp-cpp//src/kj:kj-async",
+    ],
+)
+
 wd_capnp_library(src = "docker-api.capnp")
 
 wd_capnp_library(src = "log-schema.capnp")
@@ -398,6 +436,28 @@ kj_test(
     ],
 )
 
+kj_test(
+    src = "cluster-registry-test.c++",
+    deps = [
+        ":cluster-registry",
+        "@capnp-cpp//src/capnp:capnp-rpc",
+        "@capnp-cpp//src/kj",
+        "@capnp-cpp//src/kj:kj-async",
+    ],
+)
+
+kj_test(
+    src = "cluster-lock-test.c++",
+    deps = [
+        ":cluster-lock",
+        ":cluster-registry",
+        "//src/workerd/io:worker-interface_capnp",
+        "@capnp-cpp//src/capnp:capnp-rpc",
+        "@capnp-cpp//src/kj",
+        "@capnp-cpp//src/kj:kj-async",
+    ],
+)
+
 kj_test(
     src = "json-logger-test.c++",
     deps = [

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

@@ -8,6 +8,7 @@
 #include <workerd/util/entropy.h>
 
 #include <openssl/evp.h>
+#include <openssl/hmac.h>
 #include <openssl/sha.h>
 
 #include <capnp/serialize-packed.h>
@@ -21,9 +22,31 @@
 
 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) {
+    // All nodes in the cluster must derive the same key, based on `clusterKey`. We HMAC the
+    // `clusterKey` with a randomly-chosen salt to derive the channel token key.
+
+    // clang-format off
+    static constexpr byte SALT[] = {
+      0x9a, 0x24, 0x71, 0x39, 0x1b, 0x85, 0xce, 0x97,
+      0x6c, 0xf9, 0x1c, 0xdf, 0x93, 0xc0, 0xc6, 0x36,
+      0xb9, 0xd2, 0x09, 0xfe, 0x01, 0xde, 0xb1, 0x9a,
+      0xda, 0xd3, 0x8e, 0x76, 0xbb, 0xae, 0xeb, 0x89,
+    };
+    // clang-format on
+
+    uint outLen = 0;
+    KJ_ASSERT(HMAC(EVP_sha256(), SALT, sizeof(SALT), key.asBytes().begin(), key.size(), tokenKey,
+                  &outLen) != nullptr);
+    KJ_ASSERT(outLen == sizeof(tokenKey));
+  } else {
+    // Single-instance, generate a new key for every run.
+    getEntropy(tokenKey);
+  }
 
+  // Construct key ID by hashing the key.
   SHA256_CTX ctx{};
   KJ_ASSERT(SHA256_Init(&ctx));
   KJ_ASSERT(SHA256_Update(&ctx, tokenKey, sizeof(tokenKey)));

src/workerd/server/channel-token.h

@@ -34,7 +34,7 @@ class ChannelTokenHandler {
         kj::StringPtr serviceName, kj::Maybe<kj::StringPtr> entrypoint, Frankenvalue props) = 0;
   };
 
-  explicit ChannelTokenHandler(Resolver& resolver);
+  explicit ChannelTokenHandler(Resolver& resolver, kj::Maybe<kj::StringPtr> clusterKey = kj::none);
 
   // Helpers to implement `IoChannelFactory::{SubrequestChannel,ActorClassChannel}::getToken()`.
   kj::Array<byte> encodeSubrequestChannelToken(IoChannelFactory::ChannelTokenUsage usage,