fix: [web3] only ever send RPC socket messages when the socket is open (#29195)

web3.js/rollup.config.js

@@ -110,6 +110,10 @@ function generateConfig(configType, format) {
       'jayson/lib/client/browser',
       'node-fetch',
       'rpc-websockets',
+      'rpc-websockets/dist/lib/client',
+      'rpc-websockets/dist/lib/client/client.types',
+      'rpc-websockets/dist/lib/client/websocket',
+      'rpc-websockets/dist/lib/client/websocket.browser',
       'superstruct',
     ];
   }
@@ -179,6 +183,10 @@ function generateConfig(configType, format) {
             'node-fetch',
             'react-native-url-polyfill',
             'rpc-websockets',
+            'rpc-websockets/dist/lib/client',
+            'rpc-websockets/dist/lib/client/client.types',
+            'rpc-websockets/dist/lib/client/websocket',
+            'rpc-websockets/dist/lib/client/websocket.browser',
             'superstruct',
           ];
 

web3.js/src/__forks__/browser/rpc-websocket-factory.ts

@@ -0,0 +1 @@
+export {default} from 'rpc-websockets/dist/lib/client/websocket.browser';

web3.js/src/__forks__/react-native/rpc-websocket-factory.ts

@@ -0,0 +1 @@
+export {default} from 'rpc-websockets/dist/lib/client/websocket.browser';

web3.js/src/connection.ts

@@ -26,7 +26,6 @@ import {
   any,
 } from 'superstruct';
 import type {Struct} from 'superstruct';
-import {Client as RpcWebSocketClient} from 'rpc-websockets';
 import RpcClient from 'jayson/lib/client/browser';
 import {JSONRPCError} from 'jayson';
 
@@ -36,6 +35,7 @@ import fetchImpl, {Response} from './fetch-impl';
 import {DurableNonce, NonceAccount} from './nonce-account';
 import {PublicKey} from './publickey';
 import {Signer} from './keypair';
+import RpcWebSocketClient from './rpc-websocket';
 import {MS_PER_SLOT} from './timing';
 import {
   Transaction,
@@ -5803,7 +5803,12 @@ export class Connection {
     this._rpcWebSocketConnected = true;
     this._rpcWebSocketHeartbeat = setInterval(() => {
       // Ping server every 5s to prevent idle timeouts
-      this._rpcWebSocket.notify('ping').catch(() => {});
+      (async () => {
+        try {
+          await this._rpcWebSocket.notify('ping');
+          // eslint-disable-next-line no-empty
+        } catch {}
+      })();
     }, 5000);
     this._updateSubscriptions();
   }

web3.js/src/rpc-websocket-factory.ts

@@ -0,0 +1,4 @@
+import {ICommonWebSocketFactory} from 'rpc-websockets/dist/lib/client/client.types';
+import WebsocketFactory from 'rpc-websockets/dist/lib/client/websocket';
+
+export default WebsocketFactory as ICommonWebSocketFactory;

web3.js/src/rpc-websocket.ts

@@ -0,0 +1,79 @@
+import RpcWebSocketCommonClient from 'rpc-websockets/dist/lib/client';
+import RpcWebSocketBrowserFactory from 'rpc-websockets/dist/lib/client/websocket.browser';
+import {
+  ICommonWebSocket,
+  IWSClientAdditionalOptions,
+  NodeWebSocketType,
+  NodeWebSocketTypeOptions,
+} from 'rpc-websockets/dist/lib/client/client.types';
+
+import createRpc from './rpc-websocket-factory';
+
+interface IHasReadyState {
+  readyState: WebSocket['readyState'];
+}
+
+export default class RpcWebSocketClient extends RpcWebSocketCommonClient {
+  private underlyingSocket: IHasReadyState | undefined;
+  constructor(
+    address?: string,
+    options?: IWSClientAdditionalOptions & NodeWebSocketTypeOptions,
+    generate_request_id?: (
+      method: string,
+      params: object | Array<any>,
+    ) => number,
+  ) {
+    const webSocketFactory = (url: string) => {
+      const rpc = createRpc(url, {
+        autoconnect: true,
+        max_reconnects: 5,
+        reconnect: true,
+        reconnect_interval: 1000,
+        ...options,
+      });
+      if ('socket' in rpc) {
+        this.underlyingSocket = (
+          rpc as ReturnType<typeof RpcWebSocketBrowserFactory>
+        ).socket;
+      } else {
+        this.underlyingSocket = rpc as NodeWebSocketType;
+      }
+      return rpc as ICommonWebSocket;
+    };
+    super(webSocketFactory, address, options, generate_request_id);
+  }
+  call(
+    ...args: Parameters<RpcWebSocketCommonClient['call']>
+  ): ReturnType<RpcWebSocketCommonClient['call']> {
+    const readyState = this.underlyingSocket?.readyState;
+    if (readyState === 1 /* WebSocket.OPEN */) {
+      return super.call(...args);
+    }
+    return Promise.reject(
+      new Error(
+        'Tried to call a JSON-RPC method `' +
+          args[0] +
+          '` but the socket was not `CONNECTING` or `OPEN` (`readyState` was ' +
+          readyState +
+          ')',
+      ),
+    );
+  }
+  notify(
+    ...args: Parameters<RpcWebSocketCommonClient['notify']>
+  ): ReturnType<RpcWebSocketCommonClient['notify']> {
+    const readyState = this.underlyingSocket?.readyState;
+    if (readyState === 1 /* WebSocket.OPEN */) {
+      return super.notify(...args);
+    }
+    return Promise.reject(
+      new Error(
+        'Tried to send a JSON-RPC notification `' +
+          args[0] +
+          '` but the socket was not `CONNECTING` or `OPEN` (`readyState` was ' +
+          readyState +
+          ')',
+      ),
+    );
+  }
+}

web3.js/test/connection-subscriptions.test.ts

@@ -1,5 +1,4 @@
 import chai from 'chai';
-import {Client} from 'rpc-websockets';
 import {stub, SinonStubbedInstance, SinonSpy, spy} from 'sinon';
 import sinonChai from 'sinon-chai';
 
@@ -15,6 +14,7 @@ import {
   SlotChangeCallback,
   SlotUpdateCallback,
 } from '../src';
+import type Client from '../src/rpc-websocket';
 import {url} from './url';
 
 chai.use(sinonChai);

web3.js/test/connection.test.ts

@@ -58,7 +58,7 @@ import {
   stubRpcWebSocket,
   restoreRpcWebSocket,
   mockRpcMessage,
-} from './mocks/rpc-websockets';
+} from './mocks/rpc-websocket';
 import {
   NonceInformation,
   TransactionInstruction,

web3.js/test/mocks/rpc-http.ts

@@ -2,7 +2,7 @@ import bs58 from 'bs58';
 import BN from 'bn.js';
 import * as mockttp from 'mockttp';
 
-import {mockRpcMessage} from './rpc-websockets';
+import {mockRpcMessage} from './rpc-websocket';
 import {Connection, PublicKey, Transaction, Signer} from '../../src';
 import invariant from '../../src/utils/assert';
 import type {Commitment, HttpHeaders, RpcParams} from '../../src/connection';

web3.js/test/mocks/rpc-websocket.ts

@@ -1,8 +1,8 @@
-import {Client as LiveClient} from 'rpc-websockets';
 import {expect} from 'chai';
 import {createSandbox} from 'sinon';
 
 import {Connection} from '../../src';
+import LiveClient from '../../src/rpc-websocket';
 
 type RpcRequest = {
   method: string;
@@ -55,6 +55,7 @@ export const stubRpcWebSocket = (connection: Connection) => {
     .callsFake((method: string, params: any) => {
       return mockClient.call(method, params);
     });
+  sandbox.stub(rpcWebSocket, 'notify');
 };
 
 export const restoreRpcWebSocket = (connection: Connection) => {

web3.js/test/nonce.test.ts

@@ -12,7 +12,7 @@ import {
 import {NONCE_ACCOUNT_LENGTH} from '../src/nonce-account';
 import {MOCK_PORT, url} from './url';
 import {helpers, mockRpcResponse, mockServer} from './mocks/rpc-http';
-import {stubRpcWebSocket, restoreRpcWebSocket} from './mocks/rpc-websockets';
+import {stubRpcWebSocket, restoreRpcWebSocket} from './mocks/rpc-websocket';
 
 const expectedData = (authorizedPubkey: PublicKey): [string, string] => {
   const expectedData = Buffer.alloc(NONCE_ACCOUNT_LENGTH);

web3.js/test/transaction-payer.test.ts

@@ -11,7 +11,7 @@ import {
 import invariant from '../src/utils/assert';
 import {MOCK_PORT, url} from './url';
 import {helpers, mockRpcResponse, mockServer} from './mocks/rpc-http';
-import {stubRpcWebSocket, restoreRpcWebSocket} from './mocks/rpc-websockets';
+import {stubRpcWebSocket, restoreRpcWebSocket} from './mocks/rpc-websocket';
 
 describe('Transaction Payer', () => {
   let connection: Connection;

web3.js/test/websocket.test.ts

@@ -23,7 +23,15 @@ if (process.env.TEST_LIVE) {
       expect(connection._rpcWebSocketHeartbeat).not.to.eq(null);
 
       // test if socket is open
-      await connection._rpcWebSocket.notify('ping');
+      let open = false;
+      while (!open) {
+        try {
+          await connection._rpcWebSocket.notify('ping');
+          open = true;
+        } catch {
+          continue;
+        }
+      }
 
       await connection.removeSignatureListener(id);
       expect(connection._rpcWebSocketConnected).to.eq(false);
@@ -38,7 +46,7 @@ if (process.env.TEST_LIVE) {
 
       // test if socket is closed
       await expect(connection._rpcWebSocket.notify('ping')).to.be.rejectedWith(
-        'socket not ready',
+        'Tried to send a JSON-RPC notification `ping` but the socket was not `CONNECTING` or `OPEN` (`readyState` was 3)',
       );
     });