fix(relay): order replacement sockets explicitly

Wall-clock timestamps can tie during rapid daemon socket replacement. Persist a per-connection generation so forwarding ownership follows acceptance order, including when replacing pre-generation sockets during rollout.

packages/relay/src/cloudflare-adapter.test.ts

@@ -230,6 +230,74 @@ describe("RelayDurableObject control nudge/reset behavior", () => {
   });
 });
 
+describe("RelayDurableObject data forwarding", () => {
+  it("forwards a client frame only to the newest daemon socket during replacement", () => {
+    const connectionId = "clt_replacing_daemon";
+    const client = createMockSocket({
+      version: "2",
+      role: "client",
+      connectionId,
+      serverId: "srv_test",
+      createdAt: 1,
+    });
+    const daemonSockets = Array.from({ length: 6 }, (_, index) =>
+      createMockSocket({
+        version: "2",
+        role: "server",
+        connectionId,
+        serverId: "srv_test",
+        generation: index + 1,
+        createdAt: 1,
+      }),
+    );
+    const { state, setTagSockets } = createMockState();
+    setTagSockets(`server:${connectionId}`, daemonSockets);
+
+    const relay = new RelayDurableObject(state as unknown as DurableObjectStateArg);
+    relay.webSocketMessage(client as unknown as WebSocket, "create_agent_request");
+
+    for (const staleSocket of daemonSockets.slice(0, -1)) {
+      expect(staleSocket.send).not.toHaveBeenCalled();
+    }
+    expect(daemonSockets.at(-1)?.send).toHaveBeenCalledOnce();
+    expect(daemonSockets.at(-1)?.send).toHaveBeenCalledWith("create_agent_request");
+  });
+
+  it("prefers a generated replacement over rollout sockets without generations", () => {
+    const connectionId = "clt_rollout";
+    const client = createMockSocket({
+      version: "2",
+      role: "client",
+      connectionId,
+      serverId: "srv_test",
+      createdAt: 1,
+    });
+    const staleSocket = createMockSocket({
+      version: "2",
+      role: "server",
+      connectionId,
+      serverId: "srv_test",
+      createdAt: 2,
+    });
+    const replacementSocket = createMockSocket({
+      version: "2",
+      role: "server",
+      connectionId,
+      serverId: "srv_test",
+      generation: 1,
+      createdAt: 2,
+    });
+    const { state, setTagSockets } = createMockState();
+    setTagSockets(`server:${connectionId}`, [staleSocket, replacementSocket]);
+
+    const relay = new RelayDurableObject(state as unknown as DurableObjectStateArg);
+    relay.webSocketMessage(client as unknown as WebSocket, "create_agent_request");
+
+    expect(staleSocket.send).not.toHaveBeenCalled();
+    expect(replacementSocket.send).toHaveBeenCalledWith("create_agent_request");
+  });
+});
+
 describe("relay worker endpoint routing", () => {
   it("routes missing v to legacy v1 isolated DO ids", async () => {
     const fetch = vi.fn(

packages/relay/src/cloudflare-adapter.ts

@@ -86,6 +86,29 @@ function getString(record: Record<string, unknown>, key: string): string | undef
   return typeof value === "string" ? value : undefined;
 }
 
+function getSocketGeneration(ws: WebSocket): number {
+  const attachment = deserializeAttachment(ws);
+  if (!isRecord(attachment)) return 0;
+  return typeof attachment.generation === "number" ? attachment.generation : 0;
+}
+
+function selectNewestSocket(sockets: WebSocket[]): WebSocket | null {
+  let newest: WebSocket | null = null;
+  let newestGeneration = Number.NEGATIVE_INFINITY;
+  for (const socket of sockets) {
+    const generation = getSocketGeneration(socket);
+    if (generation > newestGeneration) {
+      newest = socket;
+      newestGeneration = generation;
+    }
+  }
+  return newest;
+}
+
+function nextSocketGeneration(sockets: WebSocket[]): number {
+  return sockets.reduce((highest, socket) => Math.max(highest, getSocketGeneration(socket)), 0) + 1;
+}
+
 function getGlobalWebSocketPair(): (new () => WebSocketPair) | undefined {
   // Access WebSocketPair from global scope (Cloudflare Workers runtime)
   // Use Reflect to access global property without type assertions
@@ -352,6 +375,9 @@ export class RelayDurableObject {
 
     const isServerControl = role === "server" && !resolvedConnectionId;
     const isServerData = role === "server" && !!resolvedConnectionId;
+    const serverDataGeneration = isServerData
+      ? nextSocketGeneration(this.state.getWebSockets(`server:${resolvedConnectionId}`))
+      : undefined;
 
     // Close any existing server-side connection with the same identity.
     // - server-control: single per serverId
@@ -377,6 +403,7 @@ export class RelayDurableObject {
       role,
       version: CURRENT_RELAY_VERSION,
       connectionId: resolvedConnectionId || null,
+      ...(serverDataGeneration === undefined ? {} : { generation: serverDataGeneration }),
       createdAt: Date.now(),
     };
     serializeAttachment(server, attachment);
@@ -484,12 +511,11 @@ export class RelayDurableObject {
         this.bufferFrame(connectionId, message);
         return;
       }
-      for (const target of servers) {
+      const target = selectNewestSocket(servers);
-        try {
+      try {
-          target.send(message);
+        target?.send(message);
-        } catch (error) {
+      } catch (error) {
-          console.error(`[Relay DO] Failed to forward client->server(${connectionId}):`, error);
+        console.error(`[Relay DO] Failed to forward client->server(${connectionId}):`, error);
-        }
       }
       return;
     }

packages/relay/src/types.ts

@@ -24,5 +24,7 @@ export interface RelaySessionAttachment {
    * independent socket + E2EE channel per connected connection.
    */
   connectionId?: string | null;
+  /** Monotonic acceptance order for v2 daemon data sockets sharing a connectionId. */
+  generation?: number;
   createdAt: number;
 }