WebSocket connection to Ethereum node closes/becomes idle after some time and reconnection fails to listen to smart contract events

[royki]

Expected behavior

The WebSocket connection to the Ethereum node should remain active. The connection should continuously listen to smart contract events.

Actual behavior

The WebSocket connection to the Ethereum node either closes or becomes idle after some time (approximately 5-6 hours or 10 hours). When this happens, the reconnection attempts fail to listen to smart contract events.

Steps to reproduce the behavior

• Run the provided code to initialize the Web3 instance with the WebSocket provider and connect to the Ethereum node.
  Subscribe to new block headers and smart contract events.
• Keep running the app and observe the logs. After some hours (approximately 5-6 hours or 10 hours) app stops logs subscribeToEthNewBlockHeaders .
• Observe that the WebSocket connection either closes or becomes idle, and reconnection attempts fail to listen to smart contract events.

const { Web3 } = require('web3');
const connectWeb3 = async (wsUrl) => {
  // https://community.infura.io/t/web3-js-how-to-keep-websocket-subscriptions-alive-and-reconnect-on-failures/6199
  const provider = new Web3(new Web3.providers.WebsocketProvider(wsUrl, {}, {
    autoReconnect: true,
      delay: 5000,
      maxAttempts: 5,
  }));
  await new Promise((resolve, reject) => {
    provider.currentProvider.on('connect', () => {
      ethereumLogger.info(`Connected to Ethereum node at ${wsUrl}`);
      resolve();
    });
    provider.currentProvider.on('error', (error) => {
      ethereumLogger.error(`Error while connecting to Ethereum node: ${error}`);
      reject(error);
    });
  });

  return provider;
};

// Subscribe to new block headers
const subscribeToEthNewBlockHeaders = async (web3, ethNetwork) => {
  const subscription = await web3.eth.subscribe('newHeads', async (error, result) => {
    if (error) {
      ethereumLogger.error(`Error subscribing to new block headers: ${error}`);
    } else {
      ethereumLogger.info(`${ethNetwork}:=> New block header received: ${result.number}`);
    }
  });

  subscription.on("data", (blockHeader) => {
    ethereumLogger.info(`${ethNetwork}:=> New block header received: ${blockHeader.number}`);
  });

  subscription.on("changed", (newSubscription) => {
    ethereumLogger.info(`${ethNetwork}:=> Subscription changed: ${newSubscription}`);
  });

  subscription.on("error", (error) => {
    ethereumLogger.error(`Error subscribing to new block headers: ${error}`);
  });

  subscription.on("end", (error, unsubscribe) => {
    if (error) {
      ethereumLogger.error(`Error subscribing to new block headers: ${error}`);
    } else {
      ethereumLogger.info(`${ethNetwork}:=> Unsubscribed from new block headers: ${unsubscribe}`);
    }
  });
};

// src/listeners/ethListener.js

const { loadFile } = require('../utils/fileUtils');
const { checkEnvVars } = require('../utils/checkEnv');
const { ethereumLogger } = require('../utils/logger');
const { connectWeb3, ethNetworkName, subscribeToEthNewBlockHeaders } = require('../utils/ethUtils');
const { handleEthEvents } = require('../events/ethEventHandlers');

// Check for required environment variables
checkEnvVars('ABI_FILE_PATH', 'CONFIG_FILE_PATH', 'ETH_RPC_URL', 'BRIDGE_CONTRACT_ADDRESS');

// Subscribe to an event
const subscribeToEvent = async (contract, eventConfig, ethNetwork, handleEvent) => {
  if (!eventConfig.enabled) return;

  const eventName = eventConfig.name;
  const event = contract.events[eventName]();

  event.on('data', async (eventData) => {
    try {
      await handleEvent(eventName, eventData, contract._address, eventConfig);
    } catch (error) {
      ethereumLogger.error(`Error ${eventName} event: ${error.message}`);
    }
  });

  event.on('error', (error) => {
    ethereumLogger.error(`Web3 event error for ${eventName}: ${error}`);
  });

  event.on('connected', (id) => {
    ethereumLogger.info(`${eventName} subscriptions (subscription id): ${id}`);
  });

  ethereumLogger.info(`${ethNetwork}:=> Listening for ${eventName} events of contract ${contract._address}`);
};

async function listenEthereumEvents() {
  const ethWsUrl = process.env.ETH_RPC_URL;
  const contractAddress = process.env.BRIDGE_CONTRACT_ADDRESS;

  // Initialize Web3
  let web3;
  try {
    web3 = await connectWeb3(ethWsUrl);
  } catch (error) {
    ethereumLogger.error(`Error connecting to Ethereum node: ${error.message}`);
    return;
  }

  // Load ABI
  let abi;
  try {
    abi = loadFile(process.env.ABI_FILE_PATH, 'json');
  } catch (error) {
    ethereumLogger.error(`Error loading ABI: ${error.message}`);
    return;
  }

  // Load config
  let config;
  try {
    config = loadFile(process.env.CONFIG_FILE_PATH, 'yaml');
  } catch (error) {
    ethereumLogger.error(`Error loading config: ${error.message}`);
    return;
  }

  // Get the network name
  const ethNetwork = await ethNetworkName(ethWsUrl);

  // Create contract instance
  const contract = new web3.eth.Contract(abi, contractAddress);

  // Get the current block number
  const blockNumber = await web3.eth.getBlockNumber();
  ethereumLogger.info(`${ethNetwork}:=> Listening for events on block ${blockNumber}`);

  // Subscribe to events
  const eventsConfig = config.eth.events;
  for (const eventConfig of eventsConfig) {
    const eventName = eventConfig.name;
    if (eventConfig.enabled) {
      ethereumLogger.info(`Subscribing to ${eventName} event`);
      try {
        await subscribeToEvent(contract, eventConfig, ethNetwork, handleEthEvents);
      } catch (error) {
        ethereumLogger.error(`Error subscribing to ${eventName} event: ${error.message}`);
      }
    }
  }

  // Subscribe to New block header to keep the connection alive
  try {
    subscribeToEthNewBlockHeaders(web3, ethNetwork);
  } catch (error) {
    console.error(`Error subscribing to new block headers: ${error.message}`);
  }
}

module.exports = { listenEthereumEvents };

To address this issue, here are some resources to explore:

• https://docs.web3js.org/guides/web3_upgrade_guide/x/providers_migration_guide#provider-options-changes
• https://community.infura.io/t/web3-js-how-to-keep-websocket-subscriptions-alive-and-reconnect-on-failures/6199
• https://ethereum.stackexchange.com/questions/83575/websocket-provider-reconnects-every-1-minute
• Handle web socket disconnects Handle web socket disconnects #1354 (comment)

Here is code that temporarily solves the issue by incorporating some of the suggestions found above, such as pinging, set provider, and reconnect listener.

async function resetProviderAndResubscribe(web3, abi, config, ethNetwork, contractAddress) {
  // Reset provider
  web3.setProvider(new Web3.providers.WebsocketProvider(process.env.ETH_RPC_URL));
  ethereumLogger.info(`ReConnected to ${ethNetwork} node:: ${process.env.ETH_RPC_URL}`);

  const blockNumber = await web3.eth.getBlockNumber();
  // Initialize contract
  let contractInstance;
  try {
    contractInstance = new web3.eth.Contract(abi, contractAddress);
  } catch (error) {
    ethereumLogger.error(`Error initializing contract: ${error}`);
    return;
  }

  // Avoid the Websocket connection to go idle
  await subscribeToEthNewBlockHeaders(web3, ethNetwork);

  // Subscribe to events
  const eventsConfig = config.eth.events;
  for (const eventConfig of eventsConfig) {
    if (eventConfig.enabled) {
      const eventName = eventConfig.name;
      ethereumLogger.info(`${ethNetwork}:=> ReSubscribing to ${eventName} event on block ${blockNumber}`);
      try {
        await subscribeToContractEvents(contractInstance, eventConfig, ethNetwork, handleEthEvents);
      } catch (error) {
        ethereumLogger.error(`Error subscribing to ${eventName} event: ${error}`);
      }
    }
  }
}

async function listenEthereumEvents() {
  const ethWsUrl = process.env.ETH_RPC_URL;
  const contractAddress = process.env.BRIDGE_CONTRACT_ADDRESS;

  const ethNetwork = await ethNetworkName(ethWsUrl);

  // Initialize Web3
  let web3;
  try {
    web3 = new Web3(new Web3.providers.WebsocketProvider(ethWsUrl), {}, {
      delay: 500,
      autoReconnect: true,
      maxAttempts: 10,
    });
    ethereumLogger.info(`Connected to ${ethNetwork} node: ${ethWsUrl}`);
  } catch (error) {
    ethereumLogger.error(`Error connecting to Ethereum node: ${error}`);
    return;
  }

  // Load ABI
  let abi;
  try {
    abi = loadFile(process.env.ABI_FILE_PATH, 'json');
  } catch (error) {
    ethereumLogger.error(`Error loading ABI: ${error}`);
    return;
  }

  // Load config
  let config;
  try {
    config = loadFile(process.env.CONFIG_FILE_PATH, 'yaml');
  } catch (error) {
    ethereumLogger.error(`Error loading config: ${error}`);
    return;
  }

  // Initialize contract
  let contract;
  try {
    contract = new web3.eth.Contract(abi, contractAddress);
  } catch (error) {
    ethereumLogger.error(`Error initializing contract: ${error}`);
    return;
  }

  // Avoid the Websocket connection to go idle
  await subscribeToEthNewBlockHeaders(web3, ethNetwork);

  let pingInterval;
  function startWebsocketPingInterval() {
    pingInterval = setInterval(async () => {
      try {
        await web3.eth.net.isListening();
        ethereumLogger.info(`${ethNetwork}:=> Websocket connection alive (ping successful)`);
      } catch (error) {
        ethereumLogger.warn(`Ping failed, connection might be inactive, ${error}`);
        await resetProviderAndResubscribe(web3, abi, config, ethNetwork, contractAddress);
      }
    }, 5000);
  }

  startWebsocketPingInterval();

  // Get the current block number
  const blockNumber = await web3.eth.getBlockNumber();

  // Subscribe to events
  const eventsConfig = config.eth.events;
  for (const eventConfig of eventsConfig) {
    if (eventConfig.enabled) {
      const eventName = eventConfig.name;
      ethereumLogger.info(`${ethNetwork}:=> Subscribing to ${eventName} event on block ${blockNumber}`);
      try {
        await subscribeToContractEvents(contract, eventConfig, ethNetwork, handleEthEvents);
      } catch (error) {
        ethereumLogger.error(`Error subscribing to ${eventName} event: ${error}`);
      }
    }
  }

  web3.currentProvider.on('error', async error => {
    ethereumLogger.error(`Websocket Error: ${error}`);
    cleanup();
    startWebsocketPingInterval();
  });

  web3.currentProvider.on('end', async error => {
    ethereumLogger.error(`Websocket connection ended: ${error}`);
    cleanup();
    startWebsocketPingInterval();
  });

  process.stdin.resume();

  function cleanup() {
    clearInterval(pingInterval);
  }

  process.on('exit', cleanup);
  process.on('SIGINT', cleanup);
  process.on('SIGUSR1', cleanup);
  process.on('SIGUSR2', cleanup);
}

module.exports = { listenEthereumEvents };

Logs

2024-04-11T23:49:58.404Z [ERROR]: Websocket Error: PendingRequestsOnReconnectingError: CONNECTION ERROR: Provider started to reconnect before the response got received!
2024-04-11T23:49:58.406Z [WARN]: Ping failed, connection might be inactive, PendingRequestsOnReconnectingError: CONNECTION ERROR: Provider started to reconnect before the response got received!
2024-04-11T23:49:58.415Z [INFO]: ReConnected to EnergyWebChain node:: wss://xxxxxxxx/ws

Environment

• Network: Volta and EnergyWebChain
• Node: v18.12.0 LTS or higher
• NPM: v9.8.1
• web3: "^4.4.0"
• OS: Ubuntu 22.04 LTS

[SantiagoDevRel]

Hi @royki , thanks for publishing this here!
FYI @jdevcs @avkos @Muhammad-Altabba @luu-alex we tried different code samples (the ones that are in the docs as well) and it didn't work, its not reconnecting

[bugradursun]

I am facing the same issue, could you solve it?

[tuannm91]

I got the same issue. I try print the error log
ErrorEvent {Symbol(kTarget): WebSocket, Symbol(kType): 'error', Symbol(kError): RangeError: Invalid WebSocket frame: inval…06 at Receiver.controlMessage (/User…, Symbol(kMessage): 'Invalid WebSocket frame: invalid status code 1006'}

[royki]

Hi @bugradursun,

I solved it now using the startWebsocketPingInterval and resetProviderAndResubscribe functions. You need to adjust accordingly in your code.

[royki]

Hi @SantiagoDevRel @jdevcs @avkos @Muhammad-Altabba @luu-alex
any breakthrough regarding this?

[yanosh1982]

@SantiagoDevRel Thank you for linking me this issue. I've tried to implement the solution reported above adjusting the code to my needs:

import { Controller, Inject, LoggerService } from '@nestjs/common';
import { WINSTON_MODULE_PROVIDER } from 'nest-winston';
import Web3, { Contract, EventLog } from 'web3';
import { FdvDigestTrackerType } from "./types/fdv-digest-tracker.type";
import * as FdvDigestTracker from "./types/FdvDigestTracker.json";
import { EventListenerService } from './event-listener-service';


@Controller()
export class FdvDigestTrackerEventListener {

    private readonly contractAddress: string;
    private readonly contract: Contract<FdvDigestTrackerType>;
    private readonly web3: Web3
    private pingInterval: NodeJS.Timeout;
    private events: any[];

    constructor(
        @Inject(WINSTON_MODULE_PROVIDER) 
        private readonly logger: LoggerService,
        private readonly appService: EventListenerService
    ) {
        try {
            this.web3 = new Web3(new Web3.providers.WebsocketProvider(process.env.BLOCKCHAIN_NODE_URL));
            this.logger.log('info', `Connected to node: ${process.env.BLOCKCHAIN_NODE_URL}`);
        } catch (error) {
            this.logger.error(`Error connecting to Ethereum node: ${error}`);
            return;
        }
        this.contractAddress = process.env.BLOCKCHAIN_FDV_DIGEST_TRACKER_CONTRACT_ADDRESS;
        const abi: FdvDigestTrackerType = FdvDigestTracker.abi as unknown as FdvDigestTrackerType;
        this.contract = new this.web3.eth.Contract(abi, this.contractAddress);

        this.events = [
            { name: 'NewDigest', event: this.contract.events.NewDigest },
            { name: 'DigestUpdated', event: this.contract.events.DigestUpdated },
            { name: 'DigestDeleted', event: this.contract.events.DigestDeleted }
        ];

    }

    async onApplicationBootstrap() {
        this.logger.log('info', 'Bootstrapping %s', process.env.APP_NAME);

        await this.subscribeToNewBlockHeaders();

        this.startWebsocketPingInterval();

       for (const event of this.events) {
            await this.subscribe(event.name, event.event(), this.onData);
        }

        this.web3.currentProvider.on('error', async error => {
            this.logger.error(`Websocket Error: ${error}`);
            this.cleanup();
            this.startWebsocketPingInterval();
        });
    
        this.web3.currentProvider.on('end', async error => {
            this.logger.error(`Websocket connection ended: ${error}`);
            this.cleanup();
            this.startWebsocketPingInterval();
        })
    
    }

    private async startWebsocketPingInterval() {
        this.pingInterval = setInterval(async () => {
            try {
                await this.web3.eth.net.isListening();
                this.logger.log('info', `Websocket connection alive (ping successful)`);
            } catch (error) {
                this.logger.warn(`Ping failed, connection might be inactive, ${error}`);
                await this.resetProviderAndResubscribe();
            }
        }, parseInt(process.env.BLOCKCHAIN_WS_PROVIDER_MS_PING_INTERVAL));
    }

    private cleanup() {
        clearInterval(this.pingInterval);
    }

    private onData = async (event: EventLog) => {
        this.logger.log('info', '%s: Processing Event of type %s: %o', event.returnValues.eventId, event.event, event);
    
        const processResult = await this.appService.processEvent(event);
    
        if(processResult){
            this.logger.log('info', '%s: Event %s processed successfully', event.returnValues.eventId, event.returnValues.eventId);
        } else {
            this.logger.log('info', '%s: Event %s processed with errors', event.returnValues.eventId, event.returnValues.eventId);
        }
    }

    private async subscribeToNewBlockHeaders() {
        const subscription = await this.web3.eth.subscribe('newHeads', async (error, result) => {
          if (error) {
            this.logger.error(`Error subscribing to new block headers: ${error}`);
          } else {
            this.logger.log('info', `New block header received: ${result.number}`);
          }
        });
      
        subscription.on("data", (blockHeader) => {
          this.logger.log('info', `New block header received: ${blockHeader.number}`);
        });
      
        subscription.on("error", (error) => {
          this.logger.error(`Error subscribing to new block headers: ${error}`);
        });

        this.logger.log('info', 'Listening on %s events with subscription id %s', 'newHeads', subscription.id);
      
      };

      private async resetProviderAndResubscribe() {
        // Reset provider
        this.web3.setProvider(new Web3.providers.WebsocketProvider(process.env.BLOCKCHAIN_NODE_URL));
        this.logger.log('info', `ReConnected to node:: ${process.env.BLOCKCHAIN_NODE_URL}`);
      
        await this.subscribeToNewBlockHeaders();
      
        for (const event of this.events) {
            await this.subscribe(event.name, event.event(), this.onData);
        }

      }
      

    private async subscribe(eventName: string, subscription: any, onData: (event: EventLog) => void) {
        subscription.on("connected", (subscriptionId) => {
            this.logger.log('info', 'Listening on %s events with subscription id %s on contract %s', eventName, subscriptionId, process.env.BLOCKCHAIN_FDV_DIGEST_TRACKER_CONTRACT_ADDRESS);
        })
        
        subscription.on('data', onData);

        subscription.on('error', (error) => {
            this.logger.error('Error subscribing %s events: %o', eventName, error);
        });
    }
}

and now events are processed. Anyway I've tried to test locally what happens when the ping to the network fails turning off the wifi connection. The process is idle until the connection is back and then I obtain the following output:

2024-06-18 06:50:51 [info]: Bootstrapping fdv-blockchain-event-listener 
2024-06-18 06:50:51 [info]: **Listening on newHeads events with subscription id 0x106** 
[Nest] 82077  - 06/18/2024, 6:50:51 AM     LOG [NestApplication] Nest application successfully started +57ms
2024-06-18 06:50:51 [info]: Listening on NewDigest events with subscription id 0x107 on contract 0x5bcf3FB71dbc54a0eaf1F414911538b5142D9b22 
2024-06-18 06:50:51 [info]: Listening on DigestUpdated events with subscription id 0x108 on contract 0x5bcf3FB71dbc54a0eaf1F414911538b5142D9b22 
2024-06-18 06:50:51 [info]: Listening on DigestDeleted events with subscription id 0x109 on contract 0x5bcf3FB71dbc54a0eaf1F414911538b5142D9b22 
2024-06-18 06:50:52 [info]: New block header received: 7820125 
2024-06-18 06:51:14 [error]: Websocket Error: PendingRequestsOnReconnectingError: CONNECTION ERROR: Provider started to reconnect before the response got received! 
2024-06-18 06:51:14 [error]: Websocket Error: PendingRequestsOnReconnectingError: CONNECTION ERROR: Provider started to reconnect before the response got received! 
2024-06-18 06:51:14 [error]: Websocket Error: PendingRequestsOnReconnectingError: CONNECTION ERROR: Provider started to reconnect before the response got received! 
2024-06-18 06:51:14 [error]: Websocket Error: PendingRequestsOnReconnectingError: CONNECTION ERROR: Provider started to reconnect before the response got received! 
2024-06-18 06:51:14 [warn]: Ping failed, connection might be inactive, PendingRequestsOnReconnectingError: CONNECTION ERROR: Provider started to reconnect before the response got received! 
2024-06-18 06:51:14 [info]: ReConnected to node:: ws://172.24.181.127:8546 
2024-06-18 06:51:14 [warn]: Ping failed, connection might be inactive, PendingRequestsOnReconnectingError: CONNECTION ERROR: Provider started to reconnect before the response got received! 
2024-06-18 06:51:14 [info]: ReConnected to node:: ws://172.24.181.127:8546 
2024-06-18 06:51:14 [warn]: Ping failed, connection might be inactive, PendingRequestsOnReconnectingError: CONNECTION ERROR: Provider started to reconnect before the response got received! 
2024-06-18 06:51:14 [info]: ReConnected to node:: ws://172.24.181.127:8546 
2024-06-18 06:51:14 [warn]: Ping failed, connection might be inactive, PendingRequestsOnReconnectingError: CONNECTION ERROR: Provider started to reconnect before the response got received! 
2024-06-18 06:51:14 [info]: ReConnected to node:: ws://172.24.181.127:8546 
2024-06-18 06:51:14 [info]: Listening on newHeads events with subscription id 0x10a 
2024-06-18 06:51:14 [info]: Listening on newHeads events with subscription id 0x10b 
2024-06-18 06:51:14 [info]: Listening on newHeads events with subscription id 0x10d 
2024-06-18 06:51:14 [info]: Listening on newHeads events with subscription id 0x10c 
/home/node/app/node_modules/web3-core/src/web3_request_manager.ts:362
                                throw new InvalidResponseError<ErrorType, RequestType>(response, payload);
          ^
InvalidResponseError: Returned error: Subscription not found
    at Web3RequestManager._processJsonRpcResponse (/home/node/app/node_modules/web3-core/src/web3_request_manager.ts:362:11)
    at Web3RequestManager.<anonymous> (/home/node/app/node_modules/web3-core/src/web3_request_manager.ts:206:16)
    at Generator.next (<anonymous>)
    at fulfilled (/home/node/app/node_modules/web3-core/lib/commonjs/web3_request_manager.js:21:58) {
  innerError: { code: -32000, message: 'Subscription not found' },
  code: 101,
  data: undefined,
  request: {
    jsonrpc: '2.0',
    id: '2a57c01e-9633-452a-9294-7020e2e6fca1',
    method: 'eth_unsubscribe',
    **params: [ '0x106' ]**
  }
}
Waiting for the debugger to disconnect...

Please notice that logs entries are repeated a variable number of times and the process terminate trying to unsubscribe the initial newHeads subscription with id 0x106. This differs from @royki provided output that looks like the following

2024-04-11T23:49:58.404Z [ERROR]: Websocket Error: PendingRequestsOnReconnectingError: CONNECTION ERROR: Provider started to reconnect before the response got received!
2024-04-11T23:49:58.406Z [WARN]: Ping failed, connection might be inactive, PendingRequestsOnReconnectingError: CONNECTION ERROR: Provider started to reconnect before the response got received!
2024-04-11T23:49:58.415Z [INFO]: ReConnected to EnergyWebChain node:: wss://xxxxxxxx/ws

Can someone help me to solve this problem? Thank you

[luu-alex]

Hi @royki sorry for the long wait, will dive into this now

[luu-alex]

@royki @yanosh1982 this issue seems to be from an improper error handling within our ws providers. this will be fixed next release. thanks so much for the issue report.

[luu-alex]

Pr has been merged, these changes will be available in next release. You shouldn't need all this error checking now :). I'll update this comment when we do the next release. Thank you!

[yanosh1982]

Hi @luu-alex, thank you!!! Can you please provide a snippet to perform a web socket connection with reconnection options after your patch?

Regards!

[luu-alex]

@yanosh1982 yes i can! i'll do it sometime soon, the patch has been released so let me knwo if u run into other issues. thanks!