From e623e2888a4526d7302676f6632b32e9c9959fce Mon Sep 17 00:00:00 2001 From: Siim Raud Date: Tue, 17 Mar 2026 16:17:57 +0200 Subject: [PATCH 1/5] fix: add WebSocket readyState guard to prevent send on non-OPEN socket Closes #826 sendRawMessage() called socket.send() without checking readyState, which throws when the WebSocket is still in CONNECTING state. This could break the reconnection loop entirely, leaving the client permanently disconnected. Changes: - sendRawMessage() now checks readyState === OPEN before sending - sendPing() checks readyState before sending ping messages - Re-subscription loop in connection_ack handler logs per-subscription warnings when messages are dropped --- .../__tests__/graphql-subscriptions.spec.ts | 79 +++++++++++++++++++ .../lib/services/graphql/graphql.service.ts | 19 ++++- 2 files changed, 94 insertions(+), 4 deletions(-) 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..1cf20d2c 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,85 @@ describe('GraphQL subscriptions', () => { subscription.unsubscribe(); }); + describe('WebSocket readyState guards', () => { + it('drops messages and logs warning when socket is not in OPEN state during subscribe', async () => { + const service = fixture.graphqlService as any; + const loggerWarnSpy = jest.spyOn(service.logger, 'warn'); + + // Establish a working connection first + const sub = fixture.triggerSubscription(); + await fixture.handleConnectionInit(); + await fixture.consumeSubscribeMessage(); + + // Force socket to CONNECTING state, then attempt a new subscription + // which calls sendMessage -> sendRawMessage + Object.defineProperty(service.socket, 'readyState', { + value: 0, + writable: true, + }); + service.connectionStatus = 'CONNECTED'; // force status so sendMessage doesn't bail to openSocket + + service.sendMessage('99', 'start', { query: 'subscription { test }' }); + + expect(loggerWarnSpy).toHaveBeenCalledWith( + 'Message dropped: WebSocket is not in OPEN state', + ); + // Verify server never received the message + expect(fixture.server.messagesToConsume.pendingItems).toHaveLength(0); + + sub.unsubscribe(); + }); + + it('sendPing sets pong timeout but does not send when socket is not OPEN', () => { + const service = fixture.graphqlService as any; + const sendRawSpy = jest.spyOn(service, 'sendRawMessage'); + service.socket = { readyState: 0 }; + service.pongTimeout = null; + + service.sendPing(); + + // pongTimeout should be set (reconnection safety net) + expect(service.pongTimeout).not.toBeNull(); + // But no ping message should have been sent + expect(sendRawSpy).not.toHaveBeenCalled(); + + clearTimeout(service.pongTimeout); + }); + + it('logs warning for each failed re-subscription when socket is not open during connection_ack', async () => { + const service = fixture.graphqlService as any; + const loggerWarnSpy = jest.spyOn(service.logger, 'warn'); + + // Create a real subscription and establish connection + const sub1 = fixture.triggerSubscription('subscription { first }'); + await fixture.handleConnectionInit(); + await fixture.consumeSubscribeMessage('subscription { first }'); + + // Force disconnect and reconnect + await fixture.closeWithCode(1001); + fixture.openServer(); + await fixture.waitForConnection(); + await fixture.consumeInitMessage(); + + // Before sending connection_ack, force the socket to non-OPEN + Object.defineProperty(service.socket, 'readyState', { + value: 0, + writable: true, + }); + fixture.sendMessageToClient({ type: 'connection_ack' }); + + // Allow message processing + 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); + + 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..a63c2ffa 100644 --- a/packages/javascript-api/src/lib/services/graphql/graphql.service.ts +++ b/packages/javascript-api/src/lib/services/graphql/graphql.service.ts @@ -386,7 +386,11 @@ export class GraphqlService { type: MessageType.GQL_START, payload, }); - this.sendRawMessage(msg); + if (!this.sendRawMessage(msg)) { + this.logger.warn( + `Failed to re-subscribe subscription ${subscription.id}: WebSocket not open`, + ); + } }); break; @@ -445,8 +449,13 @@ export class GraphqlService { } } - 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; + } + this.logger.warn('Message dropped: WebSocket is not in OPEN state'); + return false; } private generateOperationId(): string { @@ -480,11 +489,13 @@ export class GraphqlService { } private sendPing(): void { + // Always set the pong timeout as a safety net: if the socket is not open, + // the timeout will fire and trigger reconnection via handleConnectionDrop. this.pongTimeout = setTimeout( this.handleConnectionDropWithThisBound, PONG_TIMEOUT_IN_MS, ); - if (this.socket) { + if (this.socket?.readyState === WebSocket.OPEN) { this.sendRawMessage(JSON.stringify({ type: MessageType.GQL_PING })); } } From cb9dd32cc4c0ffafd047175e3f76c06a4350be3c Mon Sep 17 00:00:00 2001 From: Siim Raud Date: Tue, 17 Mar 2026 16:26:03 +0200 Subject: [PATCH 2/5] chore: remove unnecessary comments from tests --- .../graphql/__tests__/graphql-subscriptions.spec.ts | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) 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 1cf20d2c..d06ba883 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 @@ -381,25 +381,21 @@ describe('GraphQL subscriptions', () => { const service = fixture.graphqlService as any; const loggerWarnSpy = jest.spyOn(service.logger, 'warn'); - // Establish a working connection first const sub = fixture.triggerSubscription(); await fixture.handleConnectionInit(); await fixture.consumeSubscribeMessage(); - // Force socket to CONNECTING state, then attempt a new subscription - // which calls sendMessage -> sendRawMessage Object.defineProperty(service.socket, 'readyState', { value: 0, writable: true, }); - service.connectionStatus = 'CONNECTED'; // force status so sendMessage doesn't bail to openSocket + service.connectionStatus = 'CONNECTED'; service.sendMessage('99', 'start', { query: 'subscription { test }' }); expect(loggerWarnSpy).toHaveBeenCalledWith( 'Message dropped: WebSocket is not in OPEN state', ); - // Verify server never received the message expect(fixture.server.messagesToConsume.pendingItems).toHaveLength(0); sub.unsubscribe(); @@ -413,9 +409,7 @@ describe('GraphQL subscriptions', () => { service.sendPing(); - // pongTimeout should be set (reconnection safety net) expect(service.pongTimeout).not.toBeNull(); - // But no ping message should have been sent expect(sendRawSpy).not.toHaveBeenCalled(); clearTimeout(service.pongTimeout); @@ -425,25 +419,21 @@ describe('GraphQL subscriptions', () => { const service = fixture.graphqlService as any; const loggerWarnSpy = jest.spyOn(service.logger, 'warn'); - // Create a real subscription and establish connection const sub1 = fixture.triggerSubscription('subscription { first }'); await fixture.handleConnectionInit(); await fixture.consumeSubscribeMessage('subscription { first }'); - // Force disconnect and reconnect await fixture.closeWithCode(1001); fixture.openServer(); await fixture.waitForConnection(); await fixture.consumeInitMessage(); - // Before sending connection_ack, force the socket to non-OPEN Object.defineProperty(service.socket, 'readyState', { value: 0, writable: true, }); fixture.sendMessageToClient({ type: 'connection_ack' }); - // Allow message processing await new Promise((r) => setTimeout(r, 10)); const resubWarnings = loggerWarnSpy.mock.calls.filter((call) => From 9ac144278ea3a283147fde80aacb819863e83f9d Mon Sep 17 00:00:00 2001 From: Siim Raud Date: Tue, 17 Mar 2026 16:34:22 +0200 Subject: [PATCH 3/5] fix: trigger immediate reconnection when sendMessage detects non-OPEN socket Instead of silently dropping messages and waiting up to 32s for the ping/pong cycle to detect the issue, sendMessage now calls handleConnectionDrop() immediately when sendRawMessage fails. --- .../graphql/__tests__/graphql-subscriptions.spec.ts | 10 ++++------ .../src/lib/services/graphql/graphql.service.ts | 4 +++- 2 files changed, 7 insertions(+), 7 deletions(-) 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 d06ba883..8ad035dd 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 @@ -377,9 +377,9 @@ describe('GraphQL subscriptions', () => { }); describe('WebSocket readyState guards', () => { - it('drops messages and logs warning when socket is not in OPEN state during subscribe', async () => { + it('triggers reconnection when socket is not in OPEN state during sendMessage', async () => { const service = fixture.graphqlService as any; - const loggerWarnSpy = jest.spyOn(service.logger, 'warn'); + const handleDropSpy = jest.spyOn(service, 'handleConnectionDrop'); const sub = fixture.triggerSubscription(); await fixture.handleConnectionInit(); @@ -393,15 +393,13 @@ describe('GraphQL subscriptions', () => { service.sendMessage('99', 'start', { query: 'subscription { test }' }); - expect(loggerWarnSpy).toHaveBeenCalledWith( - 'Message dropped: WebSocket is not in OPEN state', - ); + expect(handleDropSpy).toHaveBeenCalled(); expect(fixture.server.messagesToConsume.pendingItems).toHaveLength(0); sub.unsubscribe(); }); - it('sendPing sets pong timeout but does not send when socket is not OPEN', () => { + it('sendPing skips sending when socket is not OPEN but still sets pong timeout', () => { const service = fixture.graphqlService as any; const sendRawSpy = jest.spyOn(service, 'sendRawMessage'); service.socket = { readyState: 0 }; 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 a63c2ffa..46d32c31 100644 --- a/packages/javascript-api/src/lib/services/graphql/graphql.service.ts +++ b/packages/javascript-api/src/lib/services/graphql/graphql.service.ts @@ -443,7 +443,9 @@ 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.handleConnectionDrop(); + } } else { this.openSocket(); } From ec588069e1dc50ef10d9c8e509ee995e1df18f4f Mon Sep 17 00:00:00 2001 From: Siim Raud Date: Tue, 17 Mar 2026 16:45:15 +0200 Subject: [PATCH 4/5] refactor: make sendRawMessage a silent predicate and trigger immediate reconnect on re-subscription failure - Remove logging from sendRawMessage so callers provide context-specific messages - Remove redundant readyState guard in sendPing (sendRawMessage already checks) - Trigger handleConnectionDrop() in connection_ack when re-subscriptions fail - Use ConnectionStatus enum instead of raw string in tests --- .../graphql/__tests__/graphql-subscriptions.spec.ts | 11 ++++++----- .../src/lib/services/graphql/graphql.service.ts | 13 +++++++------ 2 files changed, 13 insertions(+), 11 deletions(-) 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 8ad035dd..94034a67 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 @@ -389,7 +389,7 @@ describe('GraphQL subscriptions', () => { value: 0, writable: true, }); - service.connectionStatus = 'CONNECTED'; + service.connectionStatus = ConnectionStatus.CONNECTED; service.sendMessage('99', 'start', { query: 'subscription { test }' }); @@ -401,20 +401,20 @@ describe('GraphQL subscriptions', () => { it('sendPing skips sending when socket is not OPEN but still sets pong timeout', () => { const service = fixture.graphqlService as any; - const sendRawSpy = jest.spyOn(service, 'sendRawMessage'); - service.socket = { readyState: 0 }; + service.socket = { readyState: 0, send: jest.fn() }; service.pongTimeout = null; service.sendPing(); expect(service.pongTimeout).not.toBeNull(); - expect(sendRawSpy).not.toHaveBeenCalled(); + expect(service.socket.send).not.toHaveBeenCalled(); clearTimeout(service.pongTimeout); }); - it('logs warning for each failed re-subscription when socket is not open during connection_ack', async () => { + 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 }'); @@ -438,6 +438,7 @@ describe('GraphQL subscriptions', () => { String(call[0]).includes('Failed to re-subscribe'), ); expect(resubWarnings).toHaveLength(1); + expect(handleDropSpy).toHaveBeenCalled(); sub1.unsubscribe(); }); 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 46d32c31..819b90ea 100644 --- a/packages/javascript-api/src/lib/services/graphql/graphql.service.ts +++ b/packages/javascript-api/src/lib/services/graphql/graphql.service.ts @@ -379,6 +379,7 @@ export class GraphqlService { this.setConnectionStatus(ConnectionStatus.CONNECTED); this.logger.info('Connected to websocket'); this.startConnectionMonitoring(); + let resubscriptionFailed = false; this.subscriptions.forEach((subscription) => { const payload = { query: subscription.query }; const msg = JSON.stringify({ @@ -390,8 +391,12 @@ export class GraphqlService { this.logger.warn( `Failed to re-subscribe subscription ${subscription.id}: WebSocket not open`, ); + resubscriptionFailed = true; } }); + if (resubscriptionFailed) { + this.handleConnectionDrop(); + } break; case MessageType.GQL_DATA: @@ -444,6 +449,7 @@ export class GraphqlService { private sendMessage(id: string, type: MessageType, payload: any) { if (this.connectionStatus === ConnectionStatus.CONNECTED) { if (!this.sendRawMessage(JSON.stringify({ id, type, payload }))) { + this.logger.warn('Message dropped: WebSocket is not in OPEN state'); this.handleConnectionDrop(); } } else { @@ -456,7 +462,6 @@ export class GraphqlService { this.socket.send(message); return true; } - this.logger.warn('Message dropped: WebSocket is not in OPEN state'); return false; } @@ -491,15 +496,11 @@ export class GraphqlService { } private sendPing(): void { - // Always set the pong timeout as a safety net: if the socket is not open, - // the timeout will fire and trigger reconnection via handleConnectionDrop. this.pongTimeout = setTimeout( this.handleConnectionDropWithThisBound, PONG_TIMEOUT_IN_MS, ); - if (this.socket?.readyState === WebSocket.OPEN) { - this.sendRawMessage(JSON.stringify({ type: MessageType.GQL_PING })); - } + this.sendRawMessage(JSON.stringify({ type: MessageType.GQL_PING })); } private handleConnectionDrop(): void { From 670d602cfe999060a48bba216fd4973d34910890 Mon Sep 17 00:00:00 2001 From: Siim Raud Date: Tue, 17 Mar 2026 16:59:24 +0200 Subject: [PATCH 5/5] refactor: break early on resubscription failure and add clarifying comments - Switch forEach to for...of with early break when first re-subscription fails (all subsequent sends would also fail on same dead socket) - Wrap connection_ack case body in braces for proper let scoping - Add comments explaining Object.defineProperty readyState overrides and mock-socket message delivery flush in tests --- .../graphql/__tests__/graphql-subscriptions.spec.ts | 5 +++++ .../src/lib/services/graphql/graphql.service.ts | 10 ++++++---- 2 files changed, 11 insertions(+), 4 deletions(-) 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 94034a67..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 @@ -385,6 +385,8 @@ describe('GraphQL subscriptions', () => { 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, @@ -426,12 +428,15 @@ describe('GraphQL subscriptions', () => { 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) => 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 819b90ea..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,13 +374,13 @@ 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(); let resubscriptionFailed = false; - this.subscriptions.forEach((subscription) => { + for (const subscription of this.subscriptions) { const payload = { query: subscription.query }; const msg = JSON.stringify({ id: subscription.id, @@ -389,15 +389,17 @@ export class GraphqlService { }); if (!this.sendRawMessage(msg)) { this.logger.warn( - `Failed to re-subscribe subscription ${subscription.id}: WebSocket not open`, + `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(