retry connection lost

src/components/Indexer/index.ts

@@ -35,9 +35,10 @@ import { BlockchainRegistry } from '../BlockchainRegistry/index.js'
 import { CommandStatus, JobStatus } from '../../@types/commands.js'
 import { buildJobIdentifier, getDeployedContractBlock } from './utils.js'
 import { create256Hash } from '../../utils/crypt.js'
-import { isReachableConnection } from '../../utils/database.js'
+import { getDatabase, isReachableConnection } from '../../utils/database.js'
 import { sleep } from '../../utils/util.js'
 import { isReindexingNeeded } from './version.js'
+import { DB_EVENTS, ES_CONNECTION_EVENTS } from '../database/ElasticsearchConfigHelper.js'
 
 /**
  * Event emitter for DDO (Data Descriptor Object) events
@@ -82,6 +83,8 @@ export class OceanIndexer {
   private supportedChains: string[]
   private indexers: Map<number, ChainIndexer> = new Map()
   private MIN_REQUIRED_VERSION = '0.2.2'
+  private isDbConnected: boolean = true
+  private reconnectTimer: NodeJS.Timeout | null = null
 
   constructor(
     db: Database,
@@ -93,9 +96,64 @@ export class OceanIndexer {
     this.blockchainRegistry = blockchainRegistry
     this.supportedChains = Object.keys(supportedNetworks)
     INDEXING_QUEUE = []
+    this.setupDbConnectionListeners()
     this.startAllChainIndexers()
   }
 
+  /**
+   * Listen for Elasticsearch connection events.
+   *
+   * CONNECTION_LOST  → cancel any pending restart, stop all indexers once.
+   * CONNECTION_RESTORED → debounce restart by 5 s so rapid LOST/RESTORED cycles are a single restart.
+   */
+  private setupDbConnectionListeners(): void {
+    ES_CONNECTION_EVENTS.on(DB_EVENTS.CONNECTION_LOST, async () => {
+      if (this.reconnectTimer) {
+        clearTimeout(this.reconnectTimer)
+        this.reconnectTimer = null
+      }
+
+      if (!this.isDbConnected) {
+        return
+      }
+
+      this.isDbConnected = false
+      INDEXER_LOGGER.error(
+        'Database connection lost - stopping all chain indexers until DB is back'
+      )
+      await this.stopAllChainIndexers()
+    })
+
+    ES_CONNECTION_EVENTS.on(DB_EVENTS.CONNECTION_RESTORED, () => {
+      if (this.isDbConnected) {
+        return
+      }
+
+      if (this.reconnectTimer) {
+        clearTimeout(this.reconnectTimer)
+      }
+
+      this.reconnectTimer = setTimeout(async () => {
+        this.reconnectTimer = null
+        if (this.isDbConnected) {
+          return
+        }
+
+        this.isDbConnected = true
+        numCrawlAttempts = 0
+        INDEXER_LOGGER.info(
+          'Database connection stable - reinitialising DB and restarting all chain indexers'
+        )
+        const freshDb = await getDatabase(true)
+        if (freshDb) {
+          this.db = freshDb
+        }
+
+        await this.startAllChainIndexers()
+      }, 5000)
+    })
+  }
+
   public getSupportedNetworks(): RPCS {
     return this.networks
   }

src/components/Indexer/utils.ts

@@ -62,9 +62,17 @@ export const getDeployedContractBlock = (network: number) => {
 }
 
 export const getNetworkHeight = async (provider: FallbackProvider) => {
-  const networkHeight = await provider.getBlockNumber()
-
-  return networkHeight
+  try {
+    const result = await withRetrial(() => provider.getBlockNumber(), 3, 2000)
+    console.log(`----------> RPC NETWORK HEIGHT: ${result}`)
+    return result
+  } catch (error: unknown) {
+    const msg = (error as Error)?.message ?? String(error)
+    if (msg.includes('invalid numeric value') || msg.includes('timeout')) {
+      throw new Error(`RPC timeout fetching block number: ${msg}`)
+    }
+    throw error
+  }
 }
 
 export const retrieveChunkEvents = async (

src/components/c2d/compute_engine_docker.ts

@@ -135,17 +135,18 @@
         supportedChains.push(parseInt(chain))
       }
     }
-    for (const feeChain of Object.keys(envConfig.fees)) {
+    if (envConfig.fees && Object.keys(envConfig.fees).length > 0) {
+      for (const feeChain of Object.keys(envConfig.fees)) {
       // for (const feeConfig of envConfig.fees) {
       if (supportedChains.includes(parseInt(feeChain))) {
         if (fees === null) fees = {}
         if (!(feeChain in fees)) fees[feeChain] = []
         const tmpFees: ComputeEnvFees[] = []
         for (let i = 0; i < envConfig.fees[feeChain].length; i++) {
           if (
             envConfig.fees[feeChain][i].prices &&
             envConfig.fees[feeChain][i].prices.length > 0
           ) {
             if (!envConfig.fees[feeChain][i].feeToken) {
               const tokenAddress = getOceanTokenAddressForChain(parseInt(feeChain))
               if (tokenAddress) {
@@ -169,6 +170,7 @@
         }
         fees[feeChain] = tmpFees
       }
+    }
 
       /* for (const chain of Object.keys(config.supportedNetworks)) {
         const chainId = parseInt(chain)
@@ -436,7 +438,7 @@
 
       if (minDuration > 0) {
         // We need to claim payment
-        const fee = env.fees[job.payment.chainId]?.find(
+        const fee = env.fees?.[job.payment.chainId]?.find(
           (fee) => fee.feeToken === job.payment.token
         )