diff --git a/packages/relay/src/cloudflare-adapter.ts b/packages/relay/src/cloudflare-adapter.ts index 73fa82fdb..f24fdafe3 100644 --- a/packages/relay/src/cloudflare-adapter.ts +++ b/packages/relay/src/cloudflare-adapter.ts @@ -121,7 +121,12 @@ export class RelayDurableObject { try { ws.send(text); } catch { - // ignore + // If the control socket is dead, close it so the daemon can reconnect. + try { + ws.close(1011, "Control send failed"); + } catch { + // ignore + } } } } @@ -237,7 +242,21 @@ export class RelayDurableObject { const { role, clientId } = attachment; if (!clientId) { - // Control channel: ignore payloads (daemon can use it for pings later). + // Control channel: support simple app-level keepalive. + if (typeof message === "string") { + try { + const parsed = JSON.parse(message) as any; + if (parsed?.type === "ping") { + try { + ws.send(JSON.stringify({ type: "pong", ts: Date.now() })); + } catch { + // ignore + } + } + } catch { + // ignore non-JSON control payloads + } + } return; } diff --git a/packages/server/src/server/relay-transport.ts b/packages/server/src/server/relay-transport.ts index 892b0bd36..2ae27d705 100644 --- a/packages/server/src/server/relay-transport.ts +++ b/packages/server/src/server/relay-transport.ts @@ -33,7 +33,9 @@ type RelaySocketLike = { type ControlMessage = | { type: "sync"; clientIds: string[] } | { type: "client_connected"; clientId: string } - | { type: "client_disconnected"; clientId: string }; + | { type: "client_disconnected"; clientId: string } + | { type: "ping" } + | { type: "pong" }; function tryParseControlMessage(raw: unknown): ControlMessage | null { try { @@ -41,6 +43,8 @@ function tryParseControlMessage(raw: unknown): ControlMessage | null { typeof raw === "string" ? raw : Buffer.isBuffer(raw) ? raw.toString("utf8") : String(raw); const parsed = JSON.parse(text) as any; if (!parsed || typeof parsed !== "object") return null; + if (parsed.type === "ping") return { type: "ping" }; + if (parsed.type === "pong") return { type: "pong" }; if (parsed.type === "sync" && Array.isArray(parsed.clientIds)) { const clientIds = parsed.clientIds.filter((id: unknown) => typeof id === "string" && id.trim().length > 0); return { type: "sync", clientIds }; @@ -71,6 +75,8 @@ export function startRelayTransport({ let reconnectTimeout: ReturnType | null = null; let reconnectAttempt = 0; const dataSockets = new Map(); // clientId -> ws + let controlKeepaliveInterval: ReturnType | null = null; + let controlLastSeenAt = 0; const stop = async (): Promise => { stopped = true; @@ -78,6 +84,10 @@ export function startRelayTransport({ clearTimeout(reconnectTimeout); reconnectTimeout = null; } + if (controlKeepaliveInterval) { + clearInterval(controlKeepaliveInterval); + controlKeepaliveInterval = null; + } if (controlWs) { try { controlWs.close(); @@ -104,11 +114,46 @@ export function startRelayTransport({ serverId, role: "server", }); - const socket = new WebSocket(url); + const socket = new WebSocket(url, { handshakeTimeout: 10_000, perMessageDeflate: false }); controlWs = socket; socket.on("open", () => { reconnectAttempt = 0; + controlLastSeenAt = Date.now(); + if (controlKeepaliveInterval) { + clearInterval(controlKeepaliveInterval); + controlKeepaliveInterval = null; + } + controlKeepaliveInterval = setInterval(() => { + if (stopped) return; + if (controlWs !== socket) return; + if (socket.readyState !== WebSocket.OPEN) return; + + const now = Date.now(); + const staleForMs = now - controlLastSeenAt; + // If the control socket is half-open or silently dropped, ws may never emit "close". + // Use app-level ping/pong to detect staleness and force a reconnect. + if (staleForMs > 90_000) { + relayLogger.warn({ url, staleForMs }, "relay_control_stale_terminating"); + try { + socket.terminate(); + } catch { + // ignore + } + return; + } + + try { + socket.send(JSON.stringify({ type: "ping", ts: now })); + } catch (error) { + relayLogger.warn({ err: error, url }, "relay_control_ping_send_failed"); + try { + socket.terminate(); + } catch { + // ignore + } + } + }, 20_000); relayLogger.info({ url }, "relay_control_connected"); }); @@ -117,6 +162,13 @@ export function startRelayTransport({ { code, reason: reason?.toString?.(), url }, "relay_control_disconnected" ); + if (controlWs === socket) { + controlWs = null; + } + if (controlKeepaliveInterval) { + clearInterval(controlKeepaliveInterval); + controlKeepaliveInterval = null; + } scheduleReconnect(); }); @@ -126,8 +178,18 @@ export function startRelayTransport({ }); socket.on("message", (data) => { + controlLastSeenAt = Date.now(); const msg = tryParseControlMessage(data); if (!msg) return; + if (msg.type === "ping") { + try { + socket.send(JSON.stringify({ type: "pong", ts: Date.now() })); + } catch { + // ignore + } + return; + } + if (msg.type === "pong") return; if (msg.type === "sync") { for (const clientId of msg.clientIds) { ensureClientDataSocket(clientId); @@ -175,12 +237,23 @@ export function startRelayTransport({ role: "server", clientId, }); - const socket = new WebSocket(url); + const socket = new WebSocket(url, { handshakeTimeout: 10_000, perMessageDeflate: false }); dataSockets.set(clientId, socket); let attached = false; + const openTimeout = setTimeout(() => { + if (stopped) return; + if (socket.readyState === WebSocket.OPEN) return; + relayLogger.warn({ url, clientId }, "relay_data_open_timeout_terminating"); + try { + socket.terminate(); + } catch { + // ignore + } + }, 15_000); socket.on("open", () => { + clearTimeout(openTimeout); relayLogger.info({ url, clientId }, "relay_data_connected"); if (attached) return; attached = true; @@ -192,6 +265,7 @@ export function startRelayTransport({ }); socket.on("close", (code, reason) => { + clearTimeout(openTimeout); relayLogger.warn( { code, reason: reason?.toString?.(), url, clientId }, "relay_data_disconnected"