From a2a56f59d5f0a723c5bc415c4163ce3cb3c41e48 Mon Sep 17 00:00:00 2001 From: Steve James Date: Tue, 4 Aug 2026 13:02:34 +0200 Subject: [PATCH 1/2] keep whatsapp session resident --- adapters/whatsapp/README.md | 19 +- adapters/whatsapp/src/lifecycle.ts | 32 +- adapters/whatsapp/src/types.ts | 4 +- adapters/whatsapp/src/whatsapp-account.ts | 511 ++++++------------ adapters/whatsapp/test/lifecycle.test.ts | 37 +- .../whatsapp/test/whatsapp-account.test.ts | 243 ++------- docs/how-to/deploy.md | 21 +- docs/how-to/messengers.md | 17 +- docs/reference/cli-commands.md | 11 +- 9 files changed, 258 insertions(+), 637 deletions(-) diff --git a/adapters/whatsapp/README.md b/adapters/whatsapp/README.md index 8a1d9900..bf4fa18f 100644 --- a/adapters/whatsapp/README.md +++ b/adapters/whatsapp/README.md @@ -61,17 +61,14 @@ while signed in as the intended GSV user. The code expires after ten minutes. The message that requests the code is consumed by the linking flow, so send a new message after linking to start a conversation with an agent. -While connected, the outbound WhatsApp WebSocket prevents the account Durable -Object from being evicted. Cloudflare limits that keepalive effect to 15 minutes -per outbound connection; the WebSocket itself may continue after that point. -GSV therefore schedules an alarm ten minutes after each successful connection. -When the alarm fires, the account opens a replacement with its saved -linked-device credentials while the current transport keeps handling messages. -Only after the replacement authenticates does it become active and close the -old transport, establishing a fresh outbound-connection lease before the -15-minute keepalive cap. The alarm is the scheduled trigger, not the mechanism -that keeps the object alive, and this reconnect is not a WhatsApp logout or a -new pairing. +While connected, the outbound WhatsApp WebSocket initially prevents the account +Durable Object from being evicted. Cloudflare limits that keepalive effect to 15 +minutes per outbound connection; the WebSocket itself may continue after that +point. GSV keeps the same provider session and schedules a Durable Object alarm +every 30 seconds. Each alarm is an incoming event inside Cloudflare's minimum +70-second idle-eviction window, so routine residency maintenance never opens a +second WhatsApp session. Baileys separately pings WhatsApp every 30 seconds and +the ordinary reconnect path replaces a transport only when it is unhealthy. This design targets the free Workers and Durable Objects plan; it does not require Containers. The account alarm also arbitrates pairing expiry, diff --git a/adapters/whatsapp/src/lifecycle.ts b/adapters/whatsapp/src/lifecycle.ts index 22c30f42..ad27e559 100644 --- a/adapters/whatsapp/src/lifecycle.ts +++ b/adapters/whatsapp/src/lifecycle.ts @@ -1,6 +1,6 @@ import { DisconnectReason } from "@whiskeysockets/baileys"; -export const SOCKET_LEASE_REFRESH_INTERVAL_MS = 10 * 60 * 1000; +export const SOCKET_RESIDENCY_ALARM_INTERVAL_MS = 30_000; export const PAIRING_WINDOW_MS = 2 * 60 * 1000; export const INBOUND_RETRY_DELAY_MS = 10_000; export const INBOUND_RETRY_BATCH_SIZE = 25; @@ -78,36 +78,6 @@ export function nextAccountAlarmDeadline( ); } -export type SocketLeaseHealth = { - hasSocket: boolean; - stateConnected: boolean; - socketAuthenticated: boolean; - webSocketOpen: boolean; -}; - -export type SocketLeaseAction = "wait" | "refresh" | "recover"; - -/** - * An established socket is checked on every account alarm, but a healthy - * socket is replaced only when its absolute Cloudflare keepalive lease is due. - */ -export function socketLeaseAction( - refreshAt: number | undefined, - health: SocketLeaseHealth, - now = Date.now(), -): SocketLeaseAction { - if (refreshAt === undefined || !Number.isFinite(refreshAt)) return "wait"; - if ( - !health.hasSocket - || !health.stateConnected - || !health.socketAuthenticated - || !health.webSocketOpen - ) { - return "recover"; - } - return refreshAt <= now ? "refresh" : "wait"; -} - export function pairingSessionExpired( authenticated: boolean, expiresAt: number | undefined, diff --git a/adapters/whatsapp/src/types.ts b/adapters/whatsapp/src/types.ts index 0ab61659..774b232c 100644 --- a/adapters/whatsapp/src/types.ts +++ b/adapters/whatsapp/src/types.ts @@ -23,7 +23,7 @@ export type WhatsAppAccountState = { lastActivity?: number; lastError?: string; disconnectReason?: string; - leaseRefreshAt?: number; + residencyAlarmAt?: number; reconnectAt?: number; connectionDeadlineAt?: number; pairingExpiresAt?: number; @@ -63,10 +63,12 @@ export function restoreWhatsAppAccountState( if (stored?.version === 2) { const { rotationAt: _obsoleteRotationAt, + leaseRefreshAt: _obsoleteLeaseRefreshAt, lastMessageAt: _obsoleteLastMessageAt, ...current } = stored as WhatsAppAccountState & { rotationAt?: number; + leaseRefreshAt?: number; lastMessageAt?: number; }; return { ...defaultWhatsAppAccountState(), ...current }; diff --git a/adapters/whatsapp/src/whatsapp-account.ts b/adapters/whatsapp/src/whatsapp-account.ts index 9e930310..e6a4ff21 100644 --- a/adapters/whatsapp/src/whatsapp-account.ts +++ b/adapters/whatsapp/src/whatsapp-account.ts @@ -87,8 +87,7 @@ import { PAIRING_WINDOW_MS, reconnectDelayMs, restartDelayMs, - socketLeaseAction, - SOCKET_LEASE_REFRESH_INTERVAL_MS, + SOCKET_RESIDENCY_ALARM_INTERVAL_MS, SocketOperationQueue, } from "./lifecycle"; import { errorFields, errorMessage, logWhatsApp } from "./logging"; @@ -119,7 +118,6 @@ const MAX_MEDIA_ITEMS = 20; const CONNECTION_OPEN_TIMEOUT_MS = 30_000; const CONNECT_WAIT_MS = 60_000; const SOCKET_OPEN_WAIT_MS = 25_000; -const SOCKET_REPLACEMENT_WAIT_MS = 4 * 60_000; const SOCKET_CLOSE_WAIT_MS = 5_000; const TINY_JPEG_BASE64 = "/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAP//////////////////////////////////////////////////////////////////////////////////////2wBDAf//////////////////////////////////////////////////////////////////////////////////////wAARCAABAAEDASIAAhEBAxEB/8QAFQABAQAAAAAAAAAAAAAAAAAAAAX/xAAUEAEAAAAAAAAAAAAAAAAAAAAA/9oADAMBAAIQAxAAAAEf/8QAFBABAAAAAAAAAAAAAAAAAAAAAP/aAAgBAQABBQJ//8QAFBEBAAAAAAAAAAAAAAAAAAAAAP/aAAgBAwEBPwF//8QAFBEBAAAAAAAAAAAAAAAAAAAAAP/aAAgBAgEBPwF//8QAFBABAAAAAAAAAAAAAAAAAAAAAP/aAAgBAQAGPwJ//8QAFBABAAAAAAAAAAAAAAAAAAAAAP/aAAgBAQABPyF//9oADAMBAAIAAwAAABCf/8QAFBEBAAAAAAAAAAAAAAAAAAAAAP/aAAgBAwEBPxB//8QAFBEBAAAAAAAAAAAAAAAAAAAAAP/aAAgBAgEBPxB//8QAFBABAAAAAAAAAAAAAAAAAAAAAP/aAAgBAQABPxB//9k="; @@ -136,12 +134,6 @@ type InboundIdentity = { isGroup: boolean; }; -type SocketReplacement = { - socket: WASocket | null; - generation: number; - saveCreds: (() => Promise) | null; -}; - class WhatsAppPreparationError extends Error { constructor(message: string, readonly retryable: boolean) { super(message); @@ -151,7 +143,6 @@ class WhatsAppPreparationError extends Error { export class WhatsAppAccount extends DurableObject { private sock: WASocket | null = null; - private socketReplacement: SocketReplacement | null = null; private readonly authenticatedSockets = new WeakSet(); private socketGeneration = 0; private readonly socketOperations = new SocketOperationQueue(); @@ -499,26 +490,35 @@ export class WhatsAppAccount extends DurableObject { await this.failConnectionAttemptLocked("connection_timeout"); }); } - const leaseAction = this.state.desired === "connected" - ? socketLeaseAction( - this.state.leaseRefreshAt, - this.socketLeaseHealth(), - now, - ) - : "wait"; - if (leaseAction !== "wait" && this.sock) { - this.beginSocketLeaseRefresh(leaseAction); + if ( + this.state.desired === "connected" + && this.sock + && ( + !this.socketIsHealthy() + || this.state.residencyAlarmAt === undefined + || this.state.residencyAlarmAt <= now + ) + ) { + await this.socketOperations.run(async () => { + if (this.state.desired !== "connected" || !this.sock) return; + if (!this.socketIsHealthy()) { + await this.failConnectionAttemptLocked("transport_unhealthy"); + return; + } + const supersededResidencyAlarm = this.state.residencyAlarmAt; + this.state.residencyAlarmAt = Date.now() + + SOCKET_RESIDENCY_ALARM_INTERVAL_MS; + await this.persistStateAndSchedule(supersededResidencyAlarm); + }); } else if ( this.state.desired === "connected" && !this.sock - && !this.socketReplacement && (this.state.reconnectAt === undefined || this.state.reconnectAt <= now) ) { await this.socketOperations.run(async () => { if ( this.state.desired !== "connected" || this.sock - || this.socketReplacement || (this.state.reconnectAt !== undefined && this.state.reconnectAt > now) ) return; await this.startSocket("alarm"); @@ -556,7 +556,7 @@ export class WhatsAppAccount extends DurableObject { const reconnectAt = now + 1_000; this.state.connected = false; this.state.status = "reconnecting"; - this.state.leaseRefreshAt = undefined; + this.state.residencyAlarmAt = undefined; this.state.connectionDeadlineAt = undefined; const reconnectDeadline = Math.min( this.state.reconnectAt ?? reconnectAt, @@ -651,7 +651,7 @@ export class WhatsAppAccount extends DurableObject { private lifecycleDeadline(): number | undefined { return earliestDeadline( - this.state.leaseRefreshAt, + this.state.residencyAlarmAt, this.state.reconnectAt, this.state.connectionDeadlineAt, this.state.pairingExpiresAt, @@ -663,42 +663,29 @@ export class WhatsAppAccount extends DurableObject { await this.inboundDeliveries.armIfPending(Date.now() + INBOUND_RETRY_DELAY_MS); } - private async startSocket( - source: string, - replacement?: SocketReplacement, - ): Promise { - if (replacement) { - if (this.socketReplacement !== replacement || replacement.socket) return; - } else if (this.sock || this.socketReplacement) { - return; - } + private async startSocket(source: string): Promise { + if (this.sock) return; const { state: authState, saveCreds, authReset } = await useDOAuthState( this.ctx.storage, ); - if (replacement && this.socketReplacement !== replacement) return; if (authReset && this.state.authenticated) { - if (replacement) { - throw new Error("WhatsApp auth reset while preparing a replacement socket"); - } await this.advanceProviderSessionLocked(); this.clearSelfIdentity(); } const sessionEpoch = this.state.sessionEpoch; - const generation = replacement?.generation ?? ++this.socketGeneration; + const generation = ++this.socketGeneration; const now = Date.now(); - if (!replacement) { - this.qrCode = null; - this.state.connected = false; - this.state.authenticated = authState.creds.registered; - this.state.status = this.state.reconnectAttempt > 0 ? "reconnecting" : "connecting"; - this.state.reconnectAt = undefined; - this.state.leaseRefreshAt = undefined; - this.state.connectionDeadlineAt = now + CONNECTION_OPEN_TIMEOUT_MS; - this.state.pairingExpiresAt = authState.creds.registered - ? undefined - : now + PAIRING_WINDOW_MS; - await this.persistStateAndSchedule(); - } + this.qrCode = null; + this.state.connected = false; + this.state.authenticated = authState.creds.registered; + this.state.status = this.state.reconnectAttempt > 0 ? "reconnecting" : "connecting"; + this.state.reconnectAt = undefined; + this.state.residencyAlarmAt = undefined; + this.state.connectionDeadlineAt = now + CONNECTION_OPEN_TIMEOUT_MS; + this.state.pairingExpiresAt = authState.creds.registered + ? undefined + : now + PAIRING_WINDOW_MS; + await this.persistStateAndSchedule(); let socket: WASocket; try { @@ -720,7 +707,6 @@ export class WhatsAppAccount extends DurableObject { cachedGroupMetadata: async (jid) => this.groupMetadata.get(jid), }); } catch (error) { - if (replacement) throw error; this.state.connectionDeadlineAt = undefined; if (this.state.desired === "connected") { this.state.status = "reconnecting"; @@ -735,15 +721,7 @@ export class WhatsAppAccount extends DurableObject { await this.persistStateAndSchedule(); throw error; } - if (replacement) { - if (this.socketReplacement !== replacement) { - await socket.end(new Error("WhatsApp socket replacement was cancelled")); - return; - } - replacement.socket = socket; - } else { - this.sock = socket; - } + this.sock = socket; socket.ev.on("creds.update", () => { this.handleCredentialsUpdate(generation, socket, saveCreds); @@ -753,7 +731,9 @@ export class WhatsAppAccount extends DurableObject { if (update.connection === "close") this.authenticatedSockets.delete(socket); this.own( "connection_update", - this.handleConnectionUpdate(generation, socket, update), + this.socketOperations.run(() => + this.handleConnectionUpdate(generation, socket, update) + ), ); }); socket.ev.on("lid-mapping.update", (mapping) => { @@ -779,35 +759,35 @@ export class WhatsAppAccount extends DurableObject { ); }); socket.ev.on("messages.upsert", (event) => { - this.own("messages_upsert", this.handleMessagesUpsert(sessionEpoch, event)); + const receivedAt = Date.now(); + logWhatsApp("info", "inbound_upsert_received", { + generation, + messageCount: event.messages.length, + upsertType: event.type, + }); + this.own( + "messages_upsert", + this.handleMessagesUpsert(sessionEpoch, event, receivedAt), + ); }); logWhatsApp("info", "socket_started", { generation, source }); - const replacementReady = replacement - ? socket.waitForConnectionUpdate( - async (update) => update.connection === "open", - SOCKET_REPLACEMENT_WAIT_MS, - ) - : Promise.resolve(); try { - await Promise.all([ - withTimeout( - socket.waitForSocketOpen(), - SOCKET_OPEN_WAIT_MS, - "WhatsApp WebSocket upgrade timed out", - ), - replacementReady, - ]); + await withTimeout( + socket.waitForSocketOpen(), + SOCKET_OPEN_WAIT_MS, + "WhatsApp WebSocket upgrade timed out", + ); } catch (error) { const failure = toError(error, "WhatsApp WebSocket upgrade failed"); const supersededConnectionDeadline = this.state.connectionDeadlineAt; - if (!replacement && this.isCurrentSocket(generation, socket)) { + if (this.isCurrentSocket(generation, socket)) { ++this.socketGeneration; this.sock = null; this.authenticatedSockets.delete(socket); this.state.connected = false; this.state.status = this.state.desired === "connected" ? "reconnecting" : "error"; this.state.connectionDeadlineAt = undefined; - this.state.leaseRefreshAt = undefined; + this.state.residencyAlarmAt = undefined; this.state.reconnectAt = this.state.desired === "connected" ? Date.now() + reconnectDelayMs(this.state.reconnectAttempt++) : undefined; @@ -829,72 +809,9 @@ export class WhatsAppAccount extends DurableObject { socket: WASocket, update: Partial, ): Promise { - const replacement = this.socketReplacement; - if ( - replacement?.generation === generation - && replacement.socket === socket - ) { - if (update.connection === "close") { - await this.failSocketReplacement( - replacement, - update.lastDisconnect?.error ?? new Error("WhatsApp replacement socket closed"), - ); - return; - } - if (update.connection !== "open") return; - if (this.state.desired !== "connected") { - this.socketReplacement = null; - await withTimeout( - socket.end(new Error("WhatsApp socket replacement is no longer needed")), - SOCKET_CLOSE_WAIT_MS, - "WhatsApp replacement close timed out", - ).catch(() => undefined); - return; - } - - const oldSocket = this.sock; - this.sock = socket; - this.socketGeneration = generation; - this.socketReplacement = null; - if (oldSocket) this.authenticatedSockets.delete(oldSocket); - const saveReplacementCreds = replacement.saveCreds; - if (saveReplacementCreds) { - this.own( - "credentials_update", - this.sessionMutations.run(saveReplacementCreds), - ); - } - if (oldSocket && oldSocket !== socket) { - this.own( - "socket_replaced_close", - this.socketOperations.run(() => - withTimeout( - oldSocket.end(new Error("WhatsApp outbound connection lease replaced")), - SOCKET_CLOSE_WAIT_MS, - "WhatsApp replaced-socket close timed out", - ) - ), - ); - } - } if (!this.isCurrentSocket(generation, socket)) return; const statusCode = providerStatusCode(update.lastDisconnect?.error); - if ( - update.connection === "close" - && statusCode === 440 - && this.socketReplacement - ) { - this.sock = null; - this.authenticatedSockets.delete(socket); - this.state.connected = false; - this.state.status = "reconnecting"; - this.state.leaseRefreshAt = undefined; - this.state.connectionDeadlineAt = undefined; - this.state.reconnectAt = undefined; - return; - } - if (update.qr) { if (pairingSessionExpired( this.state.authenticated, @@ -918,13 +835,17 @@ export class WhatsAppAccount extends DurableObject { } if (update.connection === "open") { + if ( + !this.authenticatedSockets.has(socket) + || socket.ws.isOpen !== true + ) return; if (this.state.desired === "disconnected") { const supersededLifecycleDeadline = this.lifecycleDeadline(); ++this.socketGeneration; this.sock = null; this.state.connected = false; this.state.status = "idle"; - this.state.leaseRefreshAt = undefined; + this.state.residencyAlarmAt = undefined; this.state.reconnectAt = undefined; this.state.connectionDeadlineAt = undefined; this.state.pairingExpiresAt = undefined; @@ -950,7 +871,7 @@ export class WhatsAppAccount extends DurableObject { this.state.reconnectAt = undefined; this.state.connectionDeadlineAt = undefined; this.state.pairingExpiresAt = undefined; - this.state.leaseRefreshAt = now + SOCKET_LEASE_REFRESH_INTERVAL_MS; + this.state.residencyAlarmAt = now + SOCKET_RESIDENCY_ALARM_INTERVAL_MS; this.qrCode = null; this.state.selfJid = normalizeWhatsAppJid(socket.user?.id) ?? undefined; @@ -967,11 +888,8 @@ export class WhatsAppAccount extends DurableObject { this.own("gateway_status", this.notifyGatewayStatus()); logWhatsApp("info", "socket_open", { generation, - leaseRefreshInMs: SOCKET_LEASE_REFRESH_INTERVAL_MS, + residencyAlarmInMs: SOCKET_RESIDENCY_ALARM_INTERVAL_MS, }); - if (replacement?.generation === generation && replacement.socket === socket) { - logWhatsApp("info", "socket_lease_renewed", { generation }); - } } if (update.connection !== "close") return; @@ -980,7 +898,7 @@ export class WhatsAppAccount extends DurableObject { this.groupMetadata.clear(); this.state.connected = false; this.state.lastDisconnectedAt = Date.now(); - this.state.leaseRefreshAt = undefined; + this.state.residencyAlarmAt = undefined; this.state.connectionDeadlineAt = undefined; this.qrCode = null; @@ -1005,15 +923,11 @@ export class WhatsAppAccount extends DurableObject { if (policy.clearAuth) this.state.authenticated = false; } else { this.state.status = "reconnecting"; - if (this.socketReplacement) { - this.state.reconnectAt = undefined; - } else { - const attempt = this.state.reconnectAttempt++; - const delay = policy.action === "restart" - ? restartDelayMs(attempt) - : reconnectDelayMs(attempt); - this.state.reconnectAt = Date.now() + delay; - } + const attempt = this.state.reconnectAttempt++; + const delay = policy.action === "restart" + ? restartDelayMs(attempt) + : reconnectDelayMs(attempt); + this.state.reconnectAt = Date.now() + delay; } if (policy.clearAuth) { await this.advanceProviderSessionLocked(); @@ -1021,9 +935,6 @@ export class WhatsAppAccount extends DurableObject { this.state.authenticated = false; this.clearSelfIdentity(); } - if (this.state.desired === "disconnected") { - this.cancelSocketReplacement("WhatsApp account is no longer connected"); - } await this.persistStateAndSchedule(supersededLifecycleDeadline); if (this.state.desired === "disconnected") this.resolvePairingWaiters({}); this.own("gateway_status", this.notifyGatewayStatus()); @@ -1034,140 +945,27 @@ export class WhatsAppAccount extends DurableObject { }); } - private socketLeaseHealth(): { - hasSocket: boolean; - stateConnected: boolean; - socketAuthenticated: boolean; - webSocketOpen: boolean; - } { + private socketIsHealthy(): boolean { const socket = this.sock; - return { - hasSocket: socket !== null, - stateConnected: this.state.connected, - socketAuthenticated: socket !== null - && this.authenticatedSockets.has(socket), - webSocketOpen: socket?.ws.isOpen === true, - }; - } - - private beginSocketLeaseRefresh(action: "refresh" | "recover"): void { - if ( - this.socketReplacement - || !this.sock - || this.state.desired !== "connected" - ) return; - - const activeSocket = this.sock; - const supersededLeaseDeadline = this.state.leaseRefreshAt; - const replacement: SocketReplacement = { - socket: null, - generation: this.socketGeneration + 1, - saveCreds: null, - }; - this.socketReplacement = replacement; - this.state.leaseRefreshAt = undefined; - this.state.reconnectAt = undefined; - if (action === "recover") { - this.sock = null; - this.socketGeneration = replacement.generation; - this.authenticatedSockets.delete(activeSocket); - this.state.connected = false; - this.state.status = "reconnecting"; - this.state.connectionDeadlineAt = undefined; - this.state.lastDisconnectedAt = Date.now(); - this.own( - "unhealthy_socket_close", - withTimeout( - activeSocket.end(new Error("WhatsApp transport is unhealthy")), - SOCKET_CLOSE_WAIT_MS, - "WhatsApp unhealthy-socket close timed out", - ), - ); - } - this.own( - "socket_lease_refresh", - (async () => { - try { - await this.persistStateAndSchedule(supersededLeaseDeadline); - await this.startSocket( - action === "refresh" ? "lease_refresh" : "lease_recovery", - replacement, - ); - } catch (error) { - await this.failSocketReplacement(replacement, error); - throw error; - } - })(), - ); - logWhatsApp("info", "socket_lease_refresh_started", { - previousConnectionHealthy: action === "refresh", - }); - } - - private async failSocketReplacement( - replacement: SocketReplacement, - error: unknown, - ): Promise { - if (this.socketReplacement !== replacement) return; - this.socketReplacement = null; - if (replacement.socket) { - this.authenticatedSockets.delete(replacement.socket); - } - - if (this.state.desired !== "connected") { - this.state.leaseRefreshAt = undefined; - this.state.reconnectAt = undefined; - await this.persistStateAndSchedule(); - return; - } - - const activeSocketHealthy = this.sock !== null + return socket !== null && this.state.connected - && this.authenticatedSockets.has(this.sock) - && this.sock.ws.isOpen === true; - if (activeSocketHealthy) { - this.state.leaseRefreshAt = Date.now() - + reconnectDelayMs(this.state.reconnectAttempt++); - } else { - if (this.sock) this.authenticatedSockets.delete(this.sock); - this.sock = null; - this.state.connected = false; - this.state.status = "reconnecting"; - this.state.leaseRefreshAt = undefined; - this.state.connectionDeadlineAt = undefined; - this.state.reconnectAt = Date.now() - + reconnectDelayMs(this.state.reconnectAttempt++); - } - await this.persistStateAndSchedule(); - logWhatsApp("warn", "socket_lease_refresh_failed", errorFields(error)); - } - - private cancelSocketReplacement(reason: string): void { - const replacement = this.socketReplacement; - if (!replacement) return; - this.socketReplacement = null; - const socket = replacement.socket; - if (!socket) return; - this.authenticatedSockets.delete(socket); - this.own( - "socket_replacement_cancelled", - withTimeout( - socket.end(new Error(reason)), - SOCKET_CLOSE_WAIT_MS, - "WhatsApp replacement cancellation timed out", - ), - ); + && this.authenticatedSockets.has(socket) + && socket.ws.isOpen === true; } private async failConnectionAttemptLocked(reason: string): Promise { const oldSocket = this.sock; + const now = Date.now(); ++this.socketGeneration; this.sock = null; + if (oldSocket) this.authenticatedSockets.delete(oldSocket); this.state.connected = false; + this.state.lastDisconnectedAt = now; + this.state.residencyAlarmAt = undefined; this.state.connectionDeadlineAt = undefined; if (this.state.desired === "connected") { this.state.status = "reconnecting"; - this.state.reconnectAt = Date.now() + this.state.reconnectAt = now + reconnectDelayMs(this.state.reconnectAttempt++); } else { this.state.status = "error"; @@ -1177,9 +975,9 @@ export class WhatsAppAccount extends DurableObject { await this.persistStateAndSchedule(); if (oldSocket) { await withTimeout( - oldSocket.end(new Error("WhatsApp connection attempt timed out")), + oldSocket.end(new Error("WhatsApp transport is unhealthy")), SOCKET_CLOSE_WAIT_MS, - "WhatsApp failed-attempt close timed out", + "WhatsApp unhealthy-transport close timed out", ).catch(() => undefined); } } @@ -1195,7 +993,7 @@ export class WhatsAppAccount extends DurableObject { this.state.status = "error"; this.state.lastError = "WhatsApp QR pairing expired"; this.state.disconnectReason = "pairing_expired"; - this.state.leaseRefreshAt = undefined; + this.state.residencyAlarmAt = undefined; this.state.reconnectAt = undefined; this.state.connectionDeadlineAt = undefined; this.state.pairingExpiresAt = undefined; @@ -1214,9 +1012,13 @@ export class WhatsAppAccount extends DurableObject { private async scheduleReconnectAfterFailure(error: unknown): Promise { if (this.state.desired !== "connected") return; + if (this.sock) { + this.state.lastError = errorMessage(error); + await this.failConnectionAttemptLocked("lifecycle_failure"); + return; + } if ( - !this.sock - && this.state.status === "reconnecting" + this.state.status === "reconnecting" && this.state.reconnectAt !== undefined && this.state.reconnectAt > Date.now() ) { @@ -1227,14 +1029,13 @@ export class WhatsAppAccount extends DurableObject { this.state.connected = false; this.state.status = "reconnecting"; this.state.connectionDeadlineAt = undefined; - this.state.leaseRefreshAt = undefined; + this.state.residencyAlarmAt = undefined; this.state.reconnectAt = Date.now() + reconnectDelayMs(this.state.reconnectAttempt++); this.state.lastError = errorMessage(error); await this.persistStateAndSchedule(); } private async forceNewPairingLocked(): Promise { - this.cancelSocketReplacement("Replacing WhatsApp linked-device session"); this.state.desired = "disconnected"; const providerError = await this.detachProviderSessionLocked( "force_logout", @@ -1261,10 +1062,9 @@ export class WhatsAppAccount extends DurableObject { } private async logoutLocked(): Promise { - this.cancelSocketReplacement("GSV adapter disconnected"); const supersededLifecycleDeadline = this.lifecycleDeadline(); this.state.desired = "disconnected"; - this.state.leaseRefreshAt = undefined; + this.state.residencyAlarmAt = undefined; this.state.reconnectAt = undefined; this.state.connectionDeadlineAt = undefined; this.state.pairingExpiresAt = undefined; @@ -1354,7 +1154,7 @@ export class WhatsAppAccount extends DurableObject { const supersededLifecycleDeadline = this.lifecycleDeadline(); this.state.connected = false; - this.state.leaseRefreshAt = undefined; + this.state.residencyAlarmAt = undefined; this.state.reconnectAt = undefined; this.state.connectionDeadlineAt = undefined; this.state.pairingExpiresAt = undefined; @@ -1381,6 +1181,7 @@ export class WhatsAppAccount extends DurableObject { private async handleMessagesUpsert( expectedSessionEpoch: number, event: BaileysEventMap["messages.upsert"], + receivedAt: number, ): Promise { if (expectedSessionEpoch !== this.state.sessionEpoch) return; if (event.type !== "notify" && event.type !== "append") return; @@ -1391,40 +1192,48 @@ export class WhatsAppAccount extends DurableObject { Date.now() - APPEND_CATCH_UP_MAX_AGE_MS, ); await enqueueThenDeliverInboundBatch( - async () => this.sessionMutations.run(async () => { - if (expectedSessionEpoch !== this.state.sessionEpoch) return []; - const accepted: Array<{ deliveryId: string; sessionEpoch: number }> = []; - for (const message of messages) { - if (expectedSessionEpoch !== this.state.sessionEpoch) break; - if (message.key.fromMe || !message.key.id) continue; - const identity = await this.inboundIdentity(message); - if (expectedSessionEpoch !== this.state.sessionEpoch) break; - if (!identity) continue; - const deliveryId = await whatsAppInboundDeliveryIdForSession( - expectedSessionEpoch, - { - remoteCanonicalJid: identity.surfaceJid, - senderCanonicalJid: identity.isGroup ? identity.actorJid : undefined, - legacyRemoteJid: message.key.remoteJid, - legacyParticipantJid: message.key.participant, - providerMessageId: message.key.id, - }, - ); - if (expectedSessionEpoch !== this.state.sessionEpoch) break; - await this.recentMessages.put(message); - if (expectedSessionEpoch !== this.state.sessionEpoch) break; - await this.inboundDeliveries.enqueueAndArm( - deliveryId, - proto.WebMessageInfo.encode(message).finish(), - Date.now() + INBOUND_RETRY_DELAY_MS, - ); - accepted.push({ - deliveryId, - sessionEpoch: expectedSessionEpoch, - }); - } + async () => { + const accepted = await this.sessionMutations.run(async () => { + if (expectedSessionEpoch !== this.state.sessionEpoch) return []; + const batch: Array<{ deliveryId: string; sessionEpoch: number }> = []; + for (const message of messages) { + if (expectedSessionEpoch !== this.state.sessionEpoch) break; + if (message.key.fromMe || !message.key.id) continue; + const identity = await this.inboundIdentity(message); + if (expectedSessionEpoch !== this.state.sessionEpoch) break; + if (!identity) continue; + const deliveryId = await whatsAppInboundDeliveryIdForSession( + expectedSessionEpoch, + { + remoteCanonicalJid: identity.surfaceJid, + senderCanonicalJid: identity.isGroup ? identity.actorJid : undefined, + legacyRemoteJid: message.key.remoteJid, + legacyParticipantJid: message.key.participant, + providerMessageId: message.key.id, + }, + ); + if (expectedSessionEpoch !== this.state.sessionEpoch) break; + await this.recentMessages.put(message); + if (expectedSessionEpoch !== this.state.sessionEpoch) break; + await this.inboundDeliveries.enqueueAndArm( + deliveryId, + proto.WebMessageInfo.encode(message).finish(), + Date.now() + INBOUND_RETRY_DELAY_MS, + ); + batch.push({ + deliveryId, + sessionEpoch: expectedSessionEpoch, + }); + } + return batch; + }); + logWhatsApp("info", "inbound_batch_persisted", { + acceptedCount: accepted.length, + persistenceDelayMs: Date.now() - receivedAt, + selectedCount: messages.length, + }); return accepted; - }), + }, async (accepted) => this.deliverPendingInbound( accepted.deliveryId, accepted.sessionEpoch, @@ -1561,17 +1370,27 @@ export class WhatsAppAccount extends DurableObject { await cancelBinaryBody(media.body, "Stale WhatsApp account session"); return { terminal: true }; } - const result = await callAdapterGateway( - this.gatewayBinding(), - "adapter.inbound", - { - adapter: "whatsapp", - accountId: this.state.accountId, - deliveryId, - message: inbound, - }, - media.body, - ); + const gatewayStartedAt = Date.now(); + let result: AdapterInboundResult; + try { + result = await callAdapterGateway( + this.gatewayBinding(), + "adapter.inbound", + { + adapter: "whatsapp", + accountId: this.state.accountId, + deliveryId, + message: inbound, + }, + media.body, + ); + } catch (error) { + logWhatsApp("warn", "inbound_gateway_failed", { + durationMs: Date.now() - gatewayStartedAt, + ...errorFields(error), + }); + throw error; + } if (expectedSessionEpoch !== this.state.sessionEpoch) { return { terminal: true }; } @@ -1580,6 +1399,11 @@ export class WhatsAppAccount extends DurableObject { providerMessageId: inbound.messageId, actorId: inbound.actor?.id, }); + logWhatsApp("info", "inbound_gateway_completed", { + durationMs: Date.now() - gatewayStartedAt, + ok: result.ok, + terminal: disposition.terminal, + }); if (!disposition.terminal) return disposition; this.state.lastActivity = Date.now(); if (!result.ok) { @@ -1907,24 +1731,11 @@ export class WhatsAppAccount extends DurableObject { return generation === this.socketGeneration && socket === this.sock; } - private isReplacementSocket(generation: number, socket: WASocket): boolean { - return this.socketReplacement?.generation === generation - && this.socketReplacement.socket === socket; - } - private handleCredentialsUpdate( generation: number, socket: WASocket, saveCreds: () => Promise, ): void { - if (this.isReplacementSocket(generation, socket)) { - const current = this.socketReplacement; - if (current) current.saveCreds = saveCreds; - return; - } - // Admit the update while the socket still owns the active generation. Once - // admitted, let it drain even if replacement promotion happens while the - // mutation queue is busy. Auth resets remain fenced by the store epoch. if (!this.isCurrentSocket(generation, socket)) return; this.own("credentials_update", this.sessionMutations.run(saveCreds)); } @@ -1964,7 +1775,7 @@ export class WhatsAppAccount extends DurableObject { private adapterStatus(): AdapterAccountStatus { return { accountId: this.state.accountId, - connected: this.state.connected && this.sock !== null, + connected: this.socketIsHealthy(), authenticated: this.state.authenticated, mode: "websocket", lastActivity: this.state.lastActivity, diff --git a/adapters/whatsapp/test/lifecycle.test.ts b/adapters/whatsapp/test/lifecycle.test.ts index cdd85341..d4d835f3 100644 --- a/adapters/whatsapp/test/lifecycle.test.ts +++ b/adapters/whatsapp/test/lifecycle.test.ts @@ -12,8 +12,7 @@ import { pairingSessionExpired, reconnectDelayMs, restartDelayMs, - socketLeaseAction, - SOCKET_LEASE_REFRESH_INTERVAL_MS, + SOCKET_RESIDENCY_ALARM_INTERVAL_MS, SocketOperationQueue, } from "../src/lifecycle"; import { @@ -22,36 +21,10 @@ import { } from "../src/types"; import type { WhatsAppAccountState } from "../src/types"; -const healthySocketLease = { - hasSocket: true, - stateConnected: true, - socketAuthenticated: true, - webSocketOpen: true, -}; - describe("WhatsApp lifecycle policy", () => { - it("refreshes a healthy transport only when its lease is due", () => { - expect(SOCKET_LEASE_REFRESH_INTERVAL_MS).toBeLessThan(15 * 60 * 1_000); - expect(socketLeaseAction(60_000, healthySocketLease, 10_000)).toBe("wait"); - expect(socketLeaseAction(60_000, healthySocketLease, 60_000)).toBe("refresh"); - }); - - it("recovers an unhealthy established lease without waiting for expiry", () => { - const healthSignals = Object.keys(healthySocketLease) as Array< - keyof typeof healthySocketLease - >; - for (const field of healthSignals) { - expect(socketLeaseAction(60_000, { - ...healthySocketLease, - [field]: false, - }, 10_000)) - .toBe("recover"); - } - expect(socketLeaseAction(undefined, { - ...healthySocketLease, - webSocketOpen: false, - }, 10_000)) - .toBe("wait"); + it("schedules residency alarms before Cloudflare's minimum idle eviction", () => { + expect(SOCKET_RESIDENCY_ALARM_INTERVAL_MS).toBeGreaterThan(0); + expect(SOCKET_RESIDENCY_ALARM_INTERVAL_MS).toBeLessThan(70_000); }); it("distinguishes restart, replacement, logout, and corrupt auth", () => { @@ -178,11 +151,13 @@ describe("WhatsApp state upgrade", () => { accountId: "default", status: "logged_out" as const, rotationAt: 42_000, + leaseRefreshAt: 43_000, lastMessageAt: 41_000, }; expect(restoreWhatsAppAccountState( stored as WhatsAppAccountState & { rotationAt: number; + leaseRefreshAt: number; lastMessageAt: number; }, undefined, diff --git a/adapters/whatsapp/test/whatsapp-account.test.ts b/adapters/whatsapp/test/whatsapp-account.test.ts index 9e81f2af..ddacc906 100644 --- a/adapters/whatsapp/test/whatsapp-account.test.ts +++ b/adapters/whatsapp/test/whatsapp-account.test.ts @@ -9,18 +9,12 @@ vi.mock("cloudflare:workers", () => ({ })); import { - SOCKET_LEASE_REFRESH_INTERVAL_MS, + SOCKET_RESIDENCY_ALARM_INTERVAL_MS, SocketOperationQueue, } from "../src/lifecycle"; import { defaultWhatsAppAccountState } from "../src/types"; import { WhatsAppAccount } from "../src/whatsapp-account"; -type SocketReplacement = { - socket: WASocket | null; - generation: number; - saveCreds: (() => Promise) | null; -}; - type HandleConnectionUpdate = ( this: WhatsAppAccount, generation: number, @@ -57,277 +51,150 @@ afterEach(() => { vi.restoreAllMocks(); }); -describe("WhatsApp account socket lease", () => { - it("starts one replacement without closing or blocking the active socket", async () => { - const oldSocket = { +describe("WhatsApp account residency", () => { + it("renews residency without replacing a healthy provider session", async () => { + const now = 1_000; + vi.spyOn(Date, "now").mockReturnValue(now); + const socket = { ws: { isOpen: true }, end: vi.fn(async () => undefined), } as unknown as WASocket; + const authenticatedSockets = new WeakSet(); + authenticatedSockets.add(socket); const state = { ...defaultWhatsAppAccountState(), desired: "connected" as const, status: "connected" as const, connected: true, authenticated: true, - leaseRefreshAt: 1_000, + residencyAlarmAt: now, }; - let releasePersist: (() => void) | undefined; - const persist = new Promise((resolve) => { - releasePersist = resolve; - }); - const started = vi.fn(async () => undefined); - const owned: Promise[] = []; - const authenticatedSockets = new WeakSet(); - authenticatedSockets.add(oldSocket); - const socketOperations = { run: vi.fn() }; + const persistStateAndSchedule = vi.fn(async () => undefined); const account = fakeAccount({ - sock: oldSocket, - socketReplacement: null, + sock: socket, socketGeneration: 7, authenticatedSockets, inboundDeliveries: { armIfPending: vi.fn(async () => false) }, - socketOperations, + socketOperations: { run: (operation: () => Promise) => operation() }, state, - persistStateAndSchedule: vi.fn(() => persist), - startSocket: started, + persistStateAndSchedule, retryPendingInbound: vi.fn(async () => undefined), scheduleNextAlarm: vi.fn(async () => undefined), scheduleReconnectAfterFailure: vi.fn(async () => undefined), - own: vi.fn((_event: string, promise: Promise) => { - owned.push(promise); - }), }); - await account.alarm(); await account.alarm(); - expect(accountField(account, "sock")).toBe(oldSocket); - expect(accountField(account, "socketReplacement")).toMatchObject({ - socket: null, - generation: 8, - }); + expect(accountField(account, "sock")).toBe(socket); expect(state.connected).toBe(true); - expect(state.status).toBe("connected"); - expect(oldSocket.end).not.toHaveBeenCalled(); - expect(owned).toHaveLength(1); - expect(started).not.toHaveBeenCalled(); - expect(socketOperations.run).not.toHaveBeenCalled(); - - releasePersist?.(); - await Promise.all(owned); - expect(started).toHaveBeenCalledTimes(1); - expect(started).toHaveBeenCalledWith( - "lease_refresh", - accountField(account, "socketReplacement"), + expect(state.residencyAlarmAt).toBe( + now + SOCKET_RESIDENCY_ALARM_INTERVAL_MS, ); + expect(persistStateAndSchedule).toHaveBeenCalledWith(now); + expect(socket.end).not.toHaveBeenCalled(); }); - it("retires an unhealthy socket before starting its recovery", async () => { - const oldSocket = { + it("retires an unhealthy transport through the normal reconnect path", async () => { + const now = 1_000; + vi.spyOn(Date, "now").mockReturnValue(now); + const socket = { ws: { isOpen: false }, end: vi.fn(async () => undefined), } as unknown as WASocket; + const authenticatedSockets = new WeakSet(); + authenticatedSockets.add(socket); const state = { ...defaultWhatsAppAccountState(), desired: "connected" as const, status: "connected" as const, connected: true, authenticated: true, - leaseRefreshAt: 60_000, + residencyAlarmAt: 60_000, }; - const authenticatedSockets = new WeakSet(); - authenticatedSockets.add(oldSocket); - const started = vi.fn(async () => undefined); - const owned: Promise[] = []; const account = fakeAccount({ - sock: oldSocket, - socketReplacement: null, + sock: socket, socketGeneration: 7, authenticatedSockets, inboundDeliveries: { armIfPending: vi.fn(async () => false) }, - socketOperations: { run: vi.fn() }, + socketOperations: { run: (operation: () => Promise) => operation() }, state, persistStateAndSchedule: vi.fn(async () => undefined), - startSocket: started, retryPendingInbound: vi.fn(async () => undefined), scheduleNextAlarm: vi.fn(async () => undefined), scheduleReconnectAfterFailure: vi.fn(async () => undefined), - own: vi.fn((_event: string, promise: Promise) => { - owned.push(promise); - }), }); await account.alarm(); - const replacement = accountField( - account, - "socketReplacement", - ); expect(accountField(account, "sock")).toBeNull(); expect(accountField(account, "socketGeneration")).toBe(8); - expect(replacement).toMatchObject({ socket: null, generation: 8 }); + expect(authenticatedSockets.has(socket)).toBe(false); expect(state.connected).toBe(false); expect(state.status).toBe("reconnecting"); - expect(state.authenticated).toBe(true); - expect(authenticatedSockets.has(oldSocket)).toBe(false); - expect(oldSocket.end).toHaveBeenCalledOnce(); - - await Promise.all(owned); - expect(started).toHaveBeenCalledOnce(); - expect(started).toHaveBeenCalledWith("lease_recovery", replacement); + expect(state.residencyAlarmAt).toBeUndefined(); + expect(state.lastDisconnectedAt).toBe(now); + expect(state.disconnectReason).toBe("transport_unhealthy"); + expect(state.reconnectAt).toBeGreaterThanOrEqual(now + 2_000); + expect(state.reconnectAt).toBeLessThan(now + 3_000); + expect(socket.end).toHaveBeenCalledOnce(); }); - it("keeps the active socket when its replacement fails", async () => { + it("starts residency alarms when the provider session authenticates", async () => { const now = 1_000; vi.spyOn(Date, "now").mockReturnValue(now); vi.spyOn(console, "log").mockImplementation(() => undefined); - const oldSocket = { ws: { isOpen: true } } as unknown as WASocket; - const candidate = { ws: { isOpen: false } } as unknown as WASocket; + const socket = { + ws: { isOpen: true }, + user: {}, + } as unknown as WASocket; const authenticatedSockets = new WeakSet(); - authenticatedSockets.add(oldSocket); + authenticatedSockets.add(socket); const state = { ...defaultWhatsAppAccountState(), desired: "connected" as const, - status: "connected" as const, - connected: true, - authenticated: true, - }; - const replacement: SocketReplacement = { - socket: candidate, - generation: 8, - saveCreds: null, + connectionDeadlineAt: 30_000, }; const account = fakeAccount({ - sock: oldSocket, - socketReplacement: replacement, + sock: socket, socketGeneration: 7, authenticatedSockets, - groupMetadata: { clear: vi.fn() }, + identities: { bindLidPn: vi.fn(async () => undefined) }, state, persistStateAndSchedule: vi.fn(async () => undefined), + resolvePairingWaiters: vi.fn(), notifyGatewayStatus: vi.fn(async () => undefined), own: vi.fn((_event: string, promise: Promise) => { void promise; }), }); - await accountMethod("handleConnectionUpdate").call( - account, - 8, - candidate, - { connection: "close" }, - ); - - expect(accountField(account, "sock")).toBe(oldSocket); - expect(accountField(account, "socketReplacement")).toBeNull(); - expect(state.connected).toBe(true); - expect(state.status).toBe("connected"); - expect(state.reconnectAt).toBeUndefined(); - expect(state.leaseRefreshAt).toBeGreaterThanOrEqual(now + 2_000); - expect(state.leaseRefreshAt).toBeLessThan(now + 3_000); - - authenticatedSockets.delete(oldSocket); await accountMethod("handleConnectionUpdate").call( account, 7, - oldSocket, - { - connection: "close", - lastDisconnect: { - error: { output: { statusCode: 440 } } as unknown as Error, - date: new Date(now), - }, - }, + socket, + { connection: "open" }, ); - expect(state.desired).toBe("connected"); - expect(state.status).toBe("reconnecting"); - expect(state.reconnectAt).toBeGreaterThan(now); - }); - it("promotes the replacement and ignores the displaced socket", async () => { - const now = 1_000; - vi.spyOn(Date, "now").mockReturnValue(now); - vi.spyOn(console, "log").mockImplementation(() => undefined); - const oldSocket = { - ws: { isOpen: false }, - end: vi.fn(async () => undefined), - } as unknown as WASocket; - const candidate = { - ws: { isOpen: true }, - user: {}, - } as unknown as WASocket; - const authenticatedSockets = new WeakSet(); - authenticatedSockets.add(candidate); - const state = { - ...defaultWhatsAppAccountState(), - desired: "connected" as const, - status: "connected" as const, - connected: true, - authenticated: true, - leaseRefreshAt: now, - }; - const replacement: SocketReplacement = { - socket: candidate, - generation: 8, - saveCreds: null, - }; - const owned: Promise[] = []; - const account = fakeAccount({ - sock: oldSocket, - socketReplacement: replacement, - socketGeneration: 7, - authenticatedSockets, - groupMetadata: { clear: vi.fn() }, - identities: { bindLidPn: vi.fn(async () => undefined) }, - socketOperations: { run: (operation: () => Promise) => operation() }, - state, - persistStateAndSchedule: vi.fn(async () => undefined), - resolvePairingWaiters: vi.fn(), - notifyGatewayStatus: vi.fn(async () => undefined), - own: vi.fn((_event: string, promise: Promise) => { - owned.push(promise); - }), - }); - const handleUpdate = accountMethod("handleConnectionUpdate"); - - await handleUpdate.call(account, 8, candidate, { connection: "open" }); - expect(accountField(account, "sock")).toBe(candidate); - expect(accountField(account, "socketGeneration")).toBe(8); - expect(accountField(account, "socketReplacement")).toBeNull(); expect(state.connected).toBe(true); expect(state.status).toBe("connected"); - expect(state.leaseRefreshAt).toBe( - now + SOCKET_LEASE_REFRESH_INTERVAL_MS, + expect(state.residencyAlarmAt).toBe( + now + SOCKET_RESIDENCY_ALARM_INTERVAL_MS, ); - expect(oldSocket.end).toHaveBeenCalledTimes(1); - - const connectedState = { ...state }; - await handleUpdate.call(account, 7, oldSocket, { - connection: "close", - lastDisconnect: { - error: { output: { statusCode: 440 } } as unknown as Error, - date: new Date(now), - }, - }); - expect(state).toEqual(connectedState); - expect(accountField(account, "sock")).toBe(candidate); - await Promise.all(owned); }); - it("persists active credential updates admitted before promotion", async () => { + it("persists credential updates admitted by the current session", async () => { const sessionMutations = new SocketOperationQueue(); let releaseMutation: () => void = () => undefined; const mutationGate = new Promise((resolve) => { releaseMutation = resolve; }); const precedingMutation = sessionMutations.run(() => mutationGate); - const oldSocket = {} as WASocket; - const replacementSocket = {} as WASocket; + const socket = {} as WASocket; + const nextSocket = {} as WASocket; const saveCreds = vi.fn(async () => undefined); const owned: Promise[] = []; const account = fakeAccount({ - sock: oldSocket, - socketReplacement: null, + sock: socket, socketGeneration: 7, sessionMutations, own: vi.fn((_event: string, promise: Promise) => { @@ -338,15 +205,15 @@ describe("WhatsApp account socket lease", () => { "handleCredentialsUpdate", ); - handleCredentials.call(account, 7, oldSocket, saveCreds); + handleCredentials.call(account, 7, socket, saveCreds); expect(saveCreds).not.toHaveBeenCalled(); - Reflect.set(account, "sock", replacementSocket); + Reflect.set(account, "sock", nextSocket); Reflect.set(account, "socketGeneration", 8); releaseMutation(); await Promise.all([precedingMutation, ...owned]); expect(saveCreds).toHaveBeenCalledOnce(); - handleCredentials.call(account, 7, oldSocket, saveCreds); + handleCredentials.call(account, 7, socket, saveCreds); expect(saveCreds).toHaveBeenCalledOnce(); }); }); diff --git a/docs/how-to/deploy.md b/docs/how-to/deploy.md index 6243eb24..1b2855eb 100644 --- a/docs/how-to/deploy.md +++ b/docs/how-to/deploy.md @@ -20,13 +20,14 @@ For WhatsApp, budget the Free plan for one continuously connected account. Its [outbound WebSocket prevents account Durable Object eviction for at most 15 minutes per connection](https://developers.cloudflare.com/changelog/post/2026-06-19-outbound-connections-keep-dos-alive/). The connection itself can continue after that cap, but it stops preventing -eviction. Ten minutes after each successful connection, an alarm closes the old -transport and reconnects with the saved credentials. That reconnect establishes -a fresh outbound-connection lease before the cap; the alarm is only its -scheduled trigger and does not keep the object alive by itself. -That is roughly 144 alarm requests and writes per day, but resident duration is -the tighter limit: one continuously resident 128 MB object is about 11,060 GB-s -against Cloudflare's current 13,000 GB-s daily Free allowance. This is an +eviction. While the transport is healthy, the account schedules an alarm every +30 seconds so an incoming event reaches the Durable Object before Cloudflare's +minimum idle eviction window. Routine residency maintenance therefore keeps the +same provider session; only an unhealthy transport reconnects with the saved +credentials. +That is roughly 2,880 alarm requests and writes per day, but resident duration +is the tighter limit: one continuously resident 128 MB object is about 11,060 +GB-s against Cloudflare's current 13,000 GB-s daily Free allowance. This is an operating estimate, not a hard account-capacity guarantee, because other active Durable Objects use the same allowance. Review the current [Durable Objects pricing](https://developers.cloudflare.com/durable-objects/platform/pricing/) @@ -64,9 +65,9 @@ From the CLI: gsv infra upgrade --all ``` -Routine WhatsApp upgrades and the adapter's ten-minute lease-refresh reconnect -keep the saved linked-device authentication. They are not logout operations and -do not require scanning a new QR. +Routine WhatsApp upgrades and unhealthy-transport reconnects keep the saved +linked-device authentication. They are not logout operations and do not require +scanning a new QR. ## Remove diff --git a/docs/how-to/messengers.md b/docs/how-to/messengers.md index 43561415..603cd8e4 100644 --- a/docs/how-to/messengers.md +++ b/docs/how-to/messengers.md @@ -49,13 +49,12 @@ The adapter keeps an outbound WhatsApp WebSocket in its account Durable Object. An active outbound connection now prevents eviction, but Cloudflare limits that [keepalive effect to 15 minutes per connection](https://developers.cloudflare.com/changelog/post/2026-06-19-outbound-connections-keep-dos-alive/). The connection can continue after 15 minutes, but it no longer keeps the object -resident. GSV schedules an alarm for ten minutes after each successful -connection; the alarm closes the current transport and reconnects with the -saved linked-device credentials, establishing a fresh connection lease before -the cap. The alarm does not keep the object alive by itself. This internal -reconnect is not a WhatsApp logout and does not require another scan. An -explicit **Log out** or a forced re-pair is different and removes those -credentials. +resident. GSV therefore schedules an account alarm every 30 seconds. Each alarm +is an incoming Durable Object event inside Cloudflare's minimum 70-second idle +eviction window, so the object remains resident without periodically opening a +second WhatsApp session. Baileys pings WhatsApp separately and reconnects only +when the provider transport is unhealthy. An explicit **Log out** or a forced +re-pair is different and removes the linked-device credentials. WhatsApp uses the unofficial open-source Baileys client rather than an official WhatsApp Business API integration. WhatsApp protocol changes or linked-device @@ -71,8 +70,8 @@ running another client that repeatedly replaces the same linked session. - **GSV rejects the link code:** codes are single-use and expire after ten minutes. Send another new direct message, then enter the new code while signed in as the GSV user you want to link. CLI users can run `gsv auth link CODE`. - **The code was accepted but the original message got no agent answer:** send another message. The message that generated the code is used only for identity linking and is not replayed to an agent. - **The account was logged out or replaced:** remove stale linked-device entries in WhatsApp, then use the confirmed force re-pair flow once. -- **It reconnects every ten minutes:** the brief connection-lease refresh is expected. It should not remove the linked device or ask for a new QR. -- **Several accounts do not stay connected on Workers Free:** the limiting resource is Durable Object duration, not the roughly 144 ten-minute alarms per day. Treat one continuously connected WhatsApp account as the Free-plan baseline and use Workers Paid for more always-resident accounts. +- **It reconnects repeatedly while idle:** this is not routine maintenance. Inspect the adapter's structured `socket_closed` logs and provider status code. +- **Several accounts do not stay connected on Workers Free:** the limiting resource is Durable Object duration, not the roughly 2,880 residency alarms per day. Treat one continuously connected WhatsApp account as the Free-plan baseline and use Workers Paid for more always-resident accounts. The Workers Free plan supports the SQLite-backed Durable Objects used by GSV; Containers are not required. One continuously resident 128 MB account consumes diff --git a/docs/reference/cli-commands.md b/docs/reference/cli-commands.md index 9756b5d7..9673e677 100644 --- a/docs/reference/cli-commands.md +++ b/docs/reference/cli-commands.md @@ -337,14 +337,13 @@ Treat that QR like a password. If terminal rendering fails, the CLI hides the underlying payload. `--config-json` must be a JSON object and is passed to the adapter implementation. WhatsApp accepts `{"force":true}` only as destructive recovery: it clears the existing linked-device authentication and starts a new -QR pairing. Routine reconnects and the ten-minute connection-lease refresh do not use it. +QR pairing. Routine transport recovery does not use it. Cloudflare lets an active outbound connection prevent Durable Object eviction -for at most 15 minutes. Ten minutes after each successful WhatsApp connection, -the account alarm closes that transport and reconnects with the stored -linked-device credentials. The reconnect establishes a fresh outbound -connection lease before Cloudflare's per-connection keepalive cap; the alarm -does not extend the old lease or keep the object alive by itself. +for at most 15 minutes. The account schedules an alarm every 30 seconds so an +incoming event reaches the Durable Object before Cloudflare's minimum idle +eviction window. Routine residency maintenance therefore keeps the same +WhatsApp provider session; only an unhealthy transport reconnects. If the account is paired but a direct message gets no link-code reply, first confirm `gsv adapter status` reports connected and authenticated. Send a fresh From 9d361f38c214dc62ac47dd2090f2bb8c3ae899b1 Mon Sep 17 00:00:00 2001 From: Steve James Date: Tue, 4 Aug 2026 16:10:34 +0200 Subject: [PATCH 2/2] preserve connecting whatsapp sockets --- adapters/whatsapp/src/whatsapp-account.ts | 7 +++- .../whatsapp/test/whatsapp-account.test.ts | 41 +++++++++++++++++++ 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/adapters/whatsapp/src/whatsapp-account.ts b/adapters/whatsapp/src/whatsapp-account.ts index e6a4ff21..dcd76a59 100644 --- a/adapters/whatsapp/src/whatsapp-account.ts +++ b/adapters/whatsapp/src/whatsapp-account.ts @@ -492,6 +492,7 @@ export class WhatsAppAccount extends DurableObject { } if ( this.state.desired === "connected" + && this.state.connected && this.sock && ( !this.socketIsHealthy() @@ -500,7 +501,11 @@ export class WhatsAppAccount extends DurableObject { ) ) { await this.socketOperations.run(async () => { - if (this.state.desired !== "connected" || !this.sock) return; + if ( + this.state.desired !== "connected" + || !this.state.connected + || !this.sock + ) return; if (!this.socketIsHealthy()) { await this.failConnectionAttemptLocked("transport_unhealthy"); return; diff --git a/adapters/whatsapp/test/whatsapp-account.test.ts b/adapters/whatsapp/test/whatsapp-account.test.ts index ddacc906..a37bea21 100644 --- a/adapters/whatsapp/test/whatsapp-account.test.ts +++ b/adapters/whatsapp/test/whatsapp-account.test.ts @@ -139,6 +139,47 @@ describe("WhatsApp account residency", () => { expect(socket.end).toHaveBeenCalledOnce(); }); + it("leaves a connecting socket to its connection deadline", async () => { + const now = 10_000; + vi.spyOn(Date, "now").mockReturnValue(now); + const socket = { + ws: { isOpen: true }, + end: vi.fn(async () => undefined), + } as unknown as WASocket; + const state = { + ...defaultWhatsAppAccountState(), + desired: "connected" as const, + status: "reconnecting" as const, + connected: false, + authenticated: true, + connectionDeadlineAt: now + 20_000, + }; + const armIfPending = vi.fn(async () => true); + const retryPendingInbound = vi.fn(async () => undefined); + const account = fakeAccount({ + sock: socket, + socketGeneration: 7, + authenticatedSockets: new WeakSet(), + inboundDeliveries: { armIfPending }, + socketOperations: { run: (operation: () => Promise) => operation() }, + state, + persistStateAndSchedule: vi.fn(async () => undefined), + retryPendingInbound, + scheduleNextAlarm: vi.fn(async () => undefined), + scheduleReconnectAfterFailure: vi.fn(async () => undefined), + }); + + await account.alarm(); + + expect(armIfPending).toHaveBeenCalledWith(now + 10_000); + expect(retryPendingInbound).toHaveBeenCalledOnce(); + expect(accountField(account, "sock")).toBe(socket); + expect(accountField(account, "socketGeneration")).toBe(7); + expect(state.status).toBe("reconnecting"); + expect(state.connectionDeadlineAt).toBe(now + 20_000); + expect(socket.end).not.toHaveBeenCalled(); + }); + it("starts residency alarms when the provider session authenticates", async () => { const now = 1_000; vi.spyOn(Date, "now").mockReturnValue(now);