chore: added timeout to ws connection

Author: andreabadesso

packages/event-downloader/src/config.ts

@@ -27,6 +27,7 @@ export const USE_SSL = process.env.USE_SSL === 'true';
 export const BATCH_SIZE = parseInt(process.env.BATCH_SIZE ?? '5000', 10);
 export const PARALLEL_CONNECTIONS = parseInt(process.env.PARALLEL_CONNECTIONS ?? '5', 10);
 export const WINDOW_SIZE = parseInt(process.env.WINDOW_SIZE ?? '100', 10);
+export const CONNECTION_TIMEOUT_MS = parseInt(process.env.CONNECTION_TIMEOUT_MS ?? '60000', 10);
 
 // Database configuration
 export const DB_PATH = process.env.DB_PATH ?? './events.sqlite';
@@ -41,6 +42,7 @@ export const getConfig = () => {
     PARALLEL_CONNECTIONS,
     WINDOW_SIZE,
     DB_PATH,
+    CONNECTION_TIMEOUT_MS,
   };
 };
 

packages/event-downloader/src/worker.ts

@@ -8,7 +8,7 @@
 import { WebSocket, MessageEvent, ErrorEvent } from 'ws';
 import { bigIntUtils } from '@hathor/wallet-lib';
 import { FullNodeEvent, FullNodeEventSchema, WebSocketSendEvent } from './types';
-import { FULLNODE_HOST, USE_SSL, WINDOW_SIZE } from './config';
+import { FULLNODE_HOST, USE_SSL, WINDOW_SIZE, CONNECTION_TIMEOUT_MS } from './config';
 
 export interface BatchConfig {
   batchStart: number;
@@ -43,6 +43,28 @@ export function createWorker(config: BatchConfig, callbacks: WorkerCallbacks): W
   let isRunning = false;
   let eventsSinceLastAck = 0;
   let lastReceivedEventId = 0;
+  let activityTimeout: ReturnType<typeof setTimeout> | null = null;
+
+  const resetActivityTimeout = (): void => {
+    if (activityTimeout) {
+      clearTimeout(activityTimeout);
+    }
+    if (isRunning && CONNECTION_TIMEOUT_MS > 0) {
+      activityTimeout = setTimeout(() => {
+        if (isRunning) {
+          onError(new Error(`Connection timeout: no activity for ${CONNECTION_TIMEOUT_MS}ms`));
+          stop();
+        }
+      }, CONNECTION_TIMEOUT_MS);
+    }
+  };
+
+  const clearActivityTimeout = (): void => {
+    if (activityTimeout) {
+      clearTimeout(activityTimeout);
+      activityTimeout = null;
+    }
+  };
 
   const getWsUrl = (): string => {
     const protocol = USE_SSL ? 'wss://' : 'ws://';
@@ -74,9 +96,11 @@ export function createWorker(config: BatchConfig, callbacks: WorkerCallbacks): W
         ...(lastAckEventId !== undefined && { last_ack_event_id: lastAckEventId }),
       };
       sendMessage(startMessage);
+      resetActivityTimeout();
     };
 
     socket.onmessage = (socketEvent: MessageEvent) => {
+      resetActivityTimeout();
       try {
         const rawData = bigIntUtils.JSONBigInt.parse(socketEvent.data.toString());
         const parseResult = FullNodeEventSchema.safeParse(rawData);
@@ -140,6 +164,7 @@ export function createWorker(config: BatchConfig, callbacks: WorkerCallbacks): W
 
   const stop = (): void => {
     isRunning = false;
+    clearActivityTimeout();
     if (socket) {
       socket.close();
       socket = null;

packages/event-downloader/tsconfig.tsbuildinfo

@@ -1 +1 @@
-{"root":["./src/config.ts","./src/db.ts","./src/event-parser.ts","./src/index.ts","./src/orchestrator.ts","./src/types.ts","./src/worker.ts"],"version":"5.9.3"}
+{"root":["./src/config.ts","./src/db.ts","./src/event-parser.ts","./src/index.ts","./src/orchestrator.ts","./src/types.ts","./src/worker.ts"],"version":"5.7.2"}