feat: Multi-Durable Object Support

docs/2.adapters/cloudflare.md

@@ -79,12 +79,44 @@ new_classes = ["$DurableObject"]
 See [`test/fixture/cloudflare-durable.ts`](https://github.com/h3js/crossws/blob/main/test/fixture/cloudflare-durable.ts) for demo and [`src/adapters/cloudflare.ts`](https://github.com/h3js/crossws/blob/main/src/adapters/cloudflare.ts) for implementation.
 ::
 
-### Adapter options
+## Durable Objects
+By default, the cloudflare adapter uses a single Durable Object (DO) to handle ALL requests. This behavior will create a bottleneck since DOs are design to scale horizontally and only handle about 1000 requests/s see [DO message throughput limits](https://developers.cloudflare.com/durable-objects/best-practices/rules-of-durable-objects/#message-throughput-limits). To address this, you can use the `useNamespaceAsId` option along with the [`upgrade() hook`](/guide/hooks) to control which DO handle which request.
+
+There are two scenarios for this:
+
+1. You only enable `useNamespaceAsId`. In this case, a DO will be created to handle the request based on the `URL().pathname`.
+
+2. You only enable `useNamespaceAsId` and use the upgrade hook in your route to return an object with a `namespace` property. Here you gain full control over the DO creation. The namespace to get the DO instance id.
+
+> [!NOTE]
+> When you enable the `useNamespaceAsId` option, your `upgrade()` hook will run twice!. First it will run on the worker to check if you have returned a `namespace`. Then it will run on the DO once the connection is passed to it. You can use the second argument of the `upgrade()` hook which contains the upgrade context. 
+>```ts
+>type UpgradeContext = {
+>  cf?: {
+>    runtime: "worker" | "DO";
+>  };
+>};
+>```
+
+> [!WARNING] 
+> When using Durable Objects, the adapter's `peers` property never contains the Durable Object peers (it only tracks the in-Worker fallback path). If you need access to the list of connected peers within a DO use the `getDurablePeers()` function instead. `getDurablePeers()` can only be used inside the `$DurableObject` class since it requires the DO instance.
+>```ts
+>export class $DurableObject extends DurableObject {
+>  // Pass `this` since it requires the Durable Object instance.
+>  // Optionally pass a topic to only list peers subscribed to it.
+>  listPeers() {
+>    return ws.getDurablePeers(this).map((peer) => peer.id);
+>  }
+>}
+>```
+
+## Adapter options
 
 > [!NOTE]
 > By default, crossws uses the durable object class `$DurableObject` from `env` with an instance named `crossws`.
 > You can customize this behavior by providing `resolveDurableStub` option.
 
 - `bindingName`: Durable Object binding name from environment (default: `$DurableObject`).
 - `instanceName`: Durable Object instance name (default: `crossws`).
-- `resolveDurableStub`: Custom function that resolves Durable Object binding to handle the WebSocket upgrade. This option will override `bindingName` and `instanceName`.
+- `useNamespaceAsId`: When set to `true`, each peer namespace gets its own Durable Object (default: `false`).
+- `resolveDurableStub`: Custom function that resolves Durable Object binding to handle the WebSocket upgrade. This option will override `bindingName`, `useNamespaceAsId`, and `instanceName`.

src/adapters/cloudflare.ts

@@ -4,7 +4,7 @@ import type { AdapterOptions, AdapterInstance, Adapter } from "../adapter.ts";
 import type * as web from "../../types/web.ts";
 import { env as cfGlobalEnv } from "cloudflare:workers";
 import { toBufferLike } from "../utils.ts";
-import { adapterUtils, getPeers } from "../adapter.ts";
+import { getPeers } from "../adapter.ts";
 import { AdapterHookable } from "../hooks.ts";
 import { Message } from "../message.ts";
 import { Peer, type PeerContext } from "../peer.ts";
@@ -19,6 +19,7 @@ type ResolveDurableStub = (
   req: CF.Request | undefined,
   env: unknown,
   context: CF.ExecutionContext | undefined,
+  namespace?: string,
 ) => WSDurableObjectStub | undefined | Promise<WSDurableObjectStub | undefined>;
 
 export interface CloudflareOptions extends AdapterOptions {
@@ -40,10 +41,21 @@ export interface CloudflareOptions extends AdapterOptions {
    */
   instanceName?: string;
 
+  /**
+   * Create durable object for each namespace.
+   *
+   * **Note:** This option will be ignored if `resolveDurableStub` is provided.
+   *
+   * **Note:** This option will cause the upgrade hook to run twice!.
+   *
+   * @default false
+   */
+  useNamespaceAsId?: boolean;
+
   /**
    * Custom function that resolves Durable Object binding to handle the WebSocket upgrade.
    *
-   * **Note:** This option will override `bindingName` and `instanceName`.
+   * **Note:** This option will override `bindingName`, `instanceName` and `useNamespaceAsId`.
    */
   resolveDurableStub?: ResolveDurableStub;
 }
@@ -52,23 +64,52 @@ export interface CloudflareOptions extends AdapterOptions {
 
 const cloudflareAdapter: Adapter<CloudflareDurableAdapter, CloudflareOptions> = (opts = {}) => {
   const hooks = new AdapterHookable(opts);
-  const globalPeers = new Map<string, Set<CloudflareDurablePeer | CloudflareFallbackPeer>>();
 
-  const resolveDurableStub: ResolveDurableStub =
-    opts.resolveDurableStub ||
-    ((_req, env: any, _context): WSDurableObjectStub | undefined => {
-      const bindingName = opts.bindingName || "$DurableObject";
-      const binding = (env || cfGlobalEnv)[bindingName] as CF.DurableObjectNamespace;
-      if (binding) {
-        const instanceId = binding.idFromName(opts.instanceName || "crossws");
-        return binding.get(instanceId);
+  // Tracks peers for the in-Worker fallback path only (single isolate, no
+  // Durable Object). Durable Object peers are intentionally NOT tracked here:
+  // holding peer references across requests/DOs on Cloudflare triggers I/O
+  // errors. Use `getDurablePeers()` from within the Durable Object instead.
+  const globalPeers = new Map<string, Set<CloudflareFallbackPeer>>();
+
+  const defaultDurableStubResolver: ResolveDurableStub = async (
+    req,
+    env: any,
+    _context,
+    explicitNamespace,
+  ) => {
+    const bindingName = opts.bindingName || "$DurableObject";
+    const binding = (env || cfGlobalEnv)[bindingName] as CF.DurableObjectNamespace;
+
+    if (!binding) {
+      return undefined;
+    }
+
+    // Determine the ID name logic:
+    // 1. Use explicitNamespace if provided (e.g. from publish(..., { namespace }))
+    // 2. If useNamespaceAsId is true and we have a request, run the upgrade hook
+    // 3. Fallback to instanceName
+    let instanceName = explicitNamespace || opts.instanceName || "crossws";
+
+    if (!explicitNamespace && opts.useNamespaceAsId && req) {
+      const { namespace } = await hooks.upgrade(req as unknown as Request, {
+        cf: { runtime: "worker" },
+      });
+      if (namespace) {
+        instanceName = namespace;
       }
-    });
+    }
 
-  const { publish: durablePublish, ...utils } = adapterUtils(globalPeers);
+    return binding.get(binding.idFromName(instanceName));
+  };
+
+  const resolveDurableStub: ResolveDurableStub =
+    opts.resolveDurableStub || defaultDurableStubResolver;
 
   return {
-    ...utils,
+    // Only the in-Worker fallback peers are exposed here. Durable Object peers
+    // are not (see `globalPeers` above); use `getDurablePeers()` for those.
+    peers: globalPeers,
+    getDurablePeers,
     handleUpgrade: async (request, cfEnv, cfCtx) => {
       // Upgrade request with Durable Object binding
       const stub = await resolveDurableStub(request as CF.Request, cfEnv, cfCtx);
@@ -79,11 +120,13 @@ const cloudflareAdapter: Adapter<CloudflareDurableAdapter, CloudflareOptions> =
       // [Fallback] Upgrade request in same Worker
       const { upgradeHeaders, endResponse, context, namespace } = await hooks.upgrade(
         request as unknown as Request,
+        { cf: { runtime: "worker" } },
       );
       if (endResponse) {
         return endResponse as unknown as Response;
       }
-      const peers = getPeers(globalPeers, namespace) as Set<CloudflareFallbackPeer>;
+
+      const peers = getPeers(globalPeers, namespace);
       const pair = new WebSocketPair() as unknown as [CF.WebSocket, CF.WebSocket];
       const client = pair[0];
       const server = pair[1];
@@ -127,13 +170,13 @@ const cloudflareAdapter: Adapter<CloudflareDurableAdapter, CloudflareOptions> =
       // placeholder
     },
     handleDurableUpgrade: async (obj, request) => {
-      const { upgradeHeaders, endResponse, namespace } = await hooks.upgrade(request as Request);
+      const { upgradeHeaders, endResponse, namespace } = await hooks.upgrade(request as Request, {
+        cf: { runtime: "DO" },
+      });
       if (endResponse) {
         return endResponse;
       }
 
-      const peers = getPeers(globalPeers, namespace);
-
       const pair = new WebSocketPair();
       const client = pair[0];
       const server = pair[1];
@@ -143,7 +186,7 @@ const cloudflareAdapter: Adapter<CloudflareDurableAdapter, CloudflareOptions> =
         request,
         namespace,
       );
-      peers.add(peer);
+
       (obj as DurableObjectPub).ctx.acceptWebSocket(server);
       await hooks.callHook("open", peer);
 
@@ -159,16 +202,24 @@ const cloudflareAdapter: Adapter<CloudflareDurableAdapter, CloudflareOptions> =
     },
     handleDurableClose: async (obj, ws, code, reason, wasClean) => {
       const peer = CloudflareDurablePeer._restore(obj, ws as CF.WebSocket);
-      const peers = getPeers(globalPeers, peer.namespace);
-      peers.delete(peer);
       const details = { code, reason, wasClean };
       await hooks.callHook("close", peer, details);
     },
     handleDurablePublish: async (_obj, topic, data, opts) => {
-      return durablePublish(topic, data, opts);
+      const peers = getDurablePeers(_obj as DurableObjectPub, topic);
+      for (const peer of peers) {
+        // When a namespace is given, scope the publish to a single namespace
+        // (single Durable Object hosting multiple namespaces). Without it,
+        // publish to every namespace subscribed to the topic.
+        if (opts?.namespace && peer.namespace !== opts.namespace) {
+          continue;
+        }
+        peer.send(data);
+      }
     },
     publish: async (topic, data, opts) => {
-      const stub = await resolveDurableStub(undefined, cfGlobalEnv, undefined);
+      const stub = await resolveDurableStub(undefined, cfGlobalEnv, undefined, opts?.namespace);
+
       if (!stub) {
         throw new Error("[crossws] Durable Object binding cannot be resolved.");
       }
@@ -188,6 +239,22 @@ export default cloudflareAdapter;
 
 // --- peer ---
 
+function getDurablePeers(obj: DurableObjectPub, topic?: string): CloudflareDurablePeer[] {
+  const peers: CloudflareDurablePeer[] = [];
+
+  const websockets = obj.ctx.getWebSockets() as unknown as AugmentedWebSocket[];
+  for (const ws of websockets) {
+    const state = getAttachedState(ws);
+    if (topic && !state.t?.has(topic)) {
+      continue;
+    }
+
+    const peer = CloudflareDurablePeer._restore(obj, ws);
+    peers.push(peer);
+  }
+  return peers;
+}
+
 class CloudflareDurablePeer extends Peer<{
   ws: AugmentedWebSocket;
   request: Request;
@@ -226,10 +293,10 @@ class CloudflareDurablePeer extends Peer<{
     }
     const dataBuff = toBufferLike(data);
     for (const ws of websockets) {
-      if (ws === this._internal.ws) {
+      const state = getAttachedState(ws);
+      if (state.i === this.id) {
         continue;
       }
-      const state = getAttachedState(ws);
       if (state.t?.has(topic)) {
         ws.send(dataBuff);
       }
@@ -251,10 +318,11 @@ class CloudflareDurablePeer extends Peer<{
       return peer;
     }
     const state = (ws.deserializeAttachment() || {}) as AttachedState;
+    const peerNamespace = namespace || state.n || ""; /* later throws error if empty */
     peer = ws._crosswsPeer = new CloudflareDurablePeer({
       ws: ws as CF.WebSocket,
       request: (request as Request | undefined) || new StubRequest(state.u || ""),
-      namespace: namespace || state.n || "" /* later throws error if empty */,
+      namespace: peerNamespace,
       durable: durable as DurableObjectPub,
     });
     if (state.i) {
@@ -264,7 +332,7 @@ class CloudflareDurablePeer extends Peer<{
       state.u = request.url;
     }
     state.i = peer.id;
-    state.n = peer.namespace;
+    state.n = peerNamespace;
     setAttachedState(ws, state);
     return peer;
   }
@@ -331,11 +399,22 @@ type AttachedState = {
   i?: string;
   /** Request url */
   u?: string;
-  /** Connection namespace */
-  n?: string;
+  /** Connection namespace mandatory! */
+  n: string;
 };
 
 export interface CloudflareDurableAdapter extends AdapterInstance {
+  /**
+   * List the peers connected to a Durable Object instance, optionally filtered
+   * by `topic`.
+   *
+   * **Note:** Must be called from within the `$DurableObject` class (e.g.
+   * `ws.getDurablePeers(this)`) since it relies on the Durable Object context.
+   * The adapter-level `peers` map only tracks the in-Worker fallback path and
+   * never contains Durable Object peers.
+   */
+  getDurablePeers(obj: DurableObject, topic?: string): Peer[];
+
   handleUpgrade(
     req: Request | CF.Request,
     env: unknown,
@@ -356,7 +435,7 @@ export interface CloudflareDurableAdapter extends AdapterInstance {
     obj: DurableObject,
     topic: string,
     data: unknown,
-    opts: any,
+    opts: Record<string, any> & { namespace?: string },
   ) => Promise<void>;
 
   handleDurableClose(

src/hooks.ts

@@ -39,7 +39,10 @@ export class AdapterHookable {
     }) as Promise<any>;
   }
 
-  async upgrade(request: Request & { readonly context?: Record<string, unknown> }): Promise<{
+  async upgrade(
+    request: Request & { readonly context?: Record<string, unknown> },
+    upgradeContext?: UpgradeContext,
+  ): Promise<{
     context: PeerContext;
     namespace: string;
     upgradeHeaders?: HeadersInit;
@@ -51,7 +54,11 @@ export class AdapterHookable {
     const context = request.context || {};
 
     try {
-      const res = await this.callHook("upgrade", request as Request & { context?: PeerContext });
+      const res = await this.callHook(
+        "upgrade",
+        request as Request & { context?: PeerContext },
+        upgradeContext,
+      );
       if (!res) {
         return { context, namespace };
       }
@@ -106,6 +113,12 @@ export type MaybePromise<T> = T | Promise<T>;
 
 export type UpgradeError = Response | { readonly response: Response };
 
+export type UpgradeContext = {
+  cf?: {
+    runtime: "worker" | "DO";
+  };
+};
+
 export interface Hooks {
   /**
    * Upgrading a request to a WebSocket connection.
@@ -126,6 +139,7 @@ export interface Hooks {
     request: Request & {
       readonly context?: Record<string, unknown>;
     },
+    context?: UpgradeContext,
   ) => MaybePromise<
     | {
         headers?: HeadersInit;

test/fixture/cloudflare-durable.ts

@@ -11,6 +11,13 @@ export default {
     env: Record<string, any>,
     context: ExecutionContext,
   ): Promise<Response> {
+    // The adapter-level `peers` map is always empty on Cloudflare; peers live
+    // inside the Durable Object and must be enumerated from within it.
+    if (new URL(request.url).pathname === "/peers") {
+      const stub = env.$DurableObject.get(env.$DurableObject.idFromName("crossws"));
+      return Response.json({ peers: await stub.webSocketPeers() });
+    }
+
     const response = handleDemoRoutes(ws, request);
     if (response) {
       return response;
@@ -40,6 +47,10 @@ export class $DurableObject extends DurableObject {
     return ws.handleDurablePublish(this, topic, message, opts);
   }
 
+  webSocketPeers() {
+    return ws.getDurablePeers(this).map((peer) => `${peer.namespace}:${peer.id}`);
+  }
+
   override async webSocketMessage(client: WebSocket, message: ArrayBuffer | string): Promise<void> {
     return ws.handleDurableMessage(this, client, message);
   }