diff --git a/packages/javascript-api/src/lib/services/graphql/__tests__/graphql-subscriptions.spec.ts b/packages/javascript-api/src/lib/services/graphql/__tests__/graphql-subscriptions.spec.ts index 51c0a92f..88ed8840 100644 --- a/packages/javascript-api/src/lib/services/graphql/__tests__/graphql-subscriptions.spec.ts +++ b/packages/javascript-api/src/lib/services/graphql/__tests__/graphql-subscriptions.spec.ts @@ -376,6 +376,79 @@ describe('GraphQL subscriptions', () => { subscription.unsubscribe(); }); + describe('WebSocket readyState guards', () => { + it('triggers reconnection when socket is not in OPEN state during sendMessage', async () => { + const service = fixture.graphqlService as any; + const handleDropSpy = jest.spyOn(service, 'handleConnectionDrop'); + + const sub = fixture.triggerSubscription(); + await fixture.handleConnectionInit(); + await fixture.consumeSubscribeMessage(); + + // mock-socket doesn't support simulating a half-closed socket, + // so we override readyState directly to test the guard + Object.defineProperty(service.socket, 'readyState', { + value: 0, + writable: true, + }); + service.connectionStatus = ConnectionStatus.CONNECTED; + + service.sendMessage('99', 'start', { query: 'subscription { test }' }); + + expect(handleDropSpy).toHaveBeenCalled(); + expect(fixture.server.messagesToConsume.pendingItems).toHaveLength(0); + + sub.unsubscribe(); + }); + + it('sendPing skips sending when socket is not OPEN but still sets pong timeout', () => { + const service = fixture.graphqlService as any; + service.socket = { readyState: 0, send: jest.fn() }; + service.pongTimeout = null; + + service.sendPing(); + + expect(service.pongTimeout).not.toBeNull(); + expect(service.socket.send).not.toHaveBeenCalled(); + + clearTimeout(service.pongTimeout); + }); + + it('triggers reconnection when re-subscription fails during connection_ack', async () => { + const service = fixture.graphqlService as any; + const handleDropSpy = jest.spyOn(service, 'handleConnectionDrop'); + const loggerWarnSpy = jest.spyOn(service.logger, 'warn'); + + const sub1 = fixture.triggerSubscription('subscription { first }'); + await fixture.handleConnectionInit(); + await fixture.consumeSubscribeMessage('subscription { first }'); + + await fixture.closeWithCode(1001); + fixture.openServer(); + await fixture.waitForConnection(); + await fixture.consumeInitMessage(); + + // mock-socket doesn't support simulating a half-closed socket, + // so we override readyState directly to test the guard + Object.defineProperty(service.socket, 'readyState', { + value: 0, + writable: true, + }); + fixture.sendMessageToClient({ type: 'connection_ack' }); + + // Allow mock-socket message delivery to settle + await new Promise((r) => setTimeout(r, 10)); + + const resubWarnings = loggerWarnSpy.mock.calls.filter((call) => + String(call[0]).includes('Failed to re-subscribe'), + ); + expect(resubWarnings).toHaveLength(1); + expect(handleDropSpy).toHaveBeenCalled(); + + sub1.unsubscribe(); + }); + }); + function useFakeSetInterval() { jest.useFakeTimers({ doNotFake: [ diff --git a/packages/javascript-api/src/lib/services/graphql/graphql.service.ts b/packages/javascript-api/src/lib/services/graphql/graphql.service.ts index 62052145..87458fff 100644 --- a/packages/javascript-api/src/lib/services/graphql/graphql.service.ts +++ b/packages/javascript-api/src/lib/services/graphql/graphql.service.ts @@ -374,21 +374,32 @@ export class GraphqlService { case MessageType.GQL_CONNECTION_KEEP_ALIVE: break; - case MessageType.GQL_CONNECTION_ACK: + case MessageType.GQL_CONNECTION_ACK: { this.connectionAttemptsCount = 0; this.setConnectionStatus(ConnectionStatus.CONNECTED); this.logger.info('Connected to websocket'); this.startConnectionMonitoring(); - this.subscriptions.forEach((subscription) => { + let resubscriptionFailed = false; + for (const subscription of this.subscriptions) { const payload = { query: subscription.query }; const msg = JSON.stringify({ id: subscription.id, type: MessageType.GQL_START, payload, }); - this.sendRawMessage(msg); - }); + if (!this.sendRawMessage(msg)) { + this.logger.warn( + `Failed to re-subscribe ${this.subscriptions.length} subscription(s): WebSocket not open`, + ); + resubscriptionFailed = true; + break; + } + } + if (resubscriptionFailed) { + this.handleConnectionDrop(); + } break; + } case MessageType.GQL_DATA: this.subscriptionObserverMap[message.id]?.next( @@ -439,14 +450,21 @@ export class GraphqlService { private sendMessage(id: string, type: MessageType, payload: any) { if (this.connectionStatus === ConnectionStatus.CONNECTED) { - this.sendRawMessage(JSON.stringify({ id, type, payload })); + if (!this.sendRawMessage(JSON.stringify({ id, type, payload }))) { + this.logger.warn('Message dropped: WebSocket is not in OPEN state'); + this.handleConnectionDrop(); + } } else { this.openSocket(); } } - private sendRawMessage(message: any) { - this.socket.send(message); + private sendRawMessage(message: string): boolean { + if (this.socket?.readyState === WebSocket.OPEN) { + this.socket.send(message); + return true; + } + return false; } private generateOperationId(): string { @@ -484,9 +502,7 @@ export class GraphqlService { this.handleConnectionDropWithThisBound, PONG_TIMEOUT_IN_MS, ); - if (this.socket) { - this.sendRawMessage(JSON.stringify({ type: MessageType.GQL_PING })); - } + this.sendRawMessage(JSON.stringify({ type: MessageType.GQL_PING })); } private handleConnectionDrop(): void {