chore(runner): add safeguard against duplicate sockets

engine/sdks/typescript/runner/src/tunnel.ts

@@ -2,7 +2,7 @@
 import type { MessageId, RequestId } from "@rivetkit/engine-runner-protocol";
 import type { Logger } from "pino";
 import { stringify as uuidstringify, v4 as uuidv4 } from "uuid";
 import { logger } from "./log";
 import type { ActorInstance, Runner } from "./mod";
 import {
 	stringifyToClientTunnelMessageKind,
@@ -13,7 +13,7 @@
 
 const GC_INTERVAL = 60000; // 60 seconds
 const MESSAGE_ACK_TIMEOUT = 5000; // 5 seconds
 const WEBSOCKET_STATE_PERSIST_TIMEOUT = 30000; // 30 seconds
 
 interface PendingRequest {
 	resolve: (response: Response) => void;
@@ -573,6 +573,24 @@
 			return;
 		}
 
+		// Close existing WebSocket if one already exists for this request ID.
+		// There should always be a close message sent before another open
+		// message for the same message ID.
+		//
+		// This should never occur if all is functioning correctly, but this
+		// prevents any edge case that would result in duplicate WebSockets for
+		// the same request.
+		const existingAdapter = this.#actorWebSockets.get(requestIdStr);
+		if (existingAdapter) {
+			this.log?.warn({
+				msg: "closing existing websocket for duplicate open event for the same request id",
+				requestId: requestIdStr,
+			});
+			// Close without sending a message through the tunnel since the server
+			// already knows about the new connection
+			existingAdapter.__closeWithoutCallback(1000, "ws.duplicate_open");
+		}
+
 		// Track this WebSocket for the actor
 		if (actor) {
 			actor.webSockets.add(requestIdStr);
@@ -635,8 +653,8 @@
 					headerInit[k] = v;
 				}
 			}
 			headerInit["Upgrade"] = "websocket";
 			headerInit["Connection"] = "Upgrade";
 
 			const request = new Request(`http://localhost${open.path}`, {
 				method: "GET",