retry connection lost

package.json

@@ -83,7 +83,7 @@
     "dotenv": "^16.3.1",
     "eciesjs": "^0.4.5",
     "eth-crypto": "^2.6.0",
-    "ethers": "^6.8.1",
+    "ethers": "^6.16.0",
     "express": "^4.21.1",
     "humanhash": "^1.0.4",
     "hyperdiff": "^2.0.16",

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
   }
@@ -157,11 +215,10 @@ export class OceanIndexer {
 
   async retryCrawlerWithDelay(
     blockchain: Blockchain,
-    interval: number = 5000 // in milliseconds, default 5 secs
+    interval: number = 5000 // in milliseconds, default 2 secs
   ): Promise<boolean> {
     try {
       const retryInterval = Math.max(blockchain.getKnownRPCs().length * 3000, interval) // give 2 secs per each one
-      // try
       const result = await this.startCrawler(blockchain)
       const dbActive = this.getDatabase()
       if (!dbActive || !(await isReachableConnection(dbActive.getConfig().url))) {

src/components/c2d/compute_engine_docker.ts

@@ -135,39 +135,41 @@ export class C2DEngineDocker extends C2DEngine {
         supportedChains.push(parseInt(chain))
       }
     }
-    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) {
-                envConfig.fees[feeChain][i].feeToken = tokenAddress
-                tmpFees.push(envConfig.fees[feeChain][i])
+    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) {
+                  envConfig.fees[feeChain][i].feeToken = tokenAddress
+                  tmpFees.push(envConfig.fees[feeChain][i])
+                } else {
+                  CORE_LOGGER.error(
+                    `Unable to find Ocean token address for chain ${feeChain} and no custom token provided`
+                  )
+                }
               } else {
-                CORE_LOGGER.error(
-                  `Unable to find Ocean token address for chain ${feeChain} and no custom token provided`
-                )
+                tmpFees.push(envConfig.fees[feeChain][i])
               }
             } else {
-              tmpFees.push(envConfig.fees[feeChain][i])
+              CORE_LOGGER.error(
+                `Unable to find prices for fee ${JSON.stringify(
+                  envConfig.fees[feeChain][i]
+                )} on chain ${feeChain}`
+              )
             }
-          } else {
-            CORE_LOGGER.error(
-              `Unable to find prices for fee ${JSON.stringify(
-                envConfig.fees[feeChain][i]
-              )} on chain ${feeChain}`
-            )
           }
+          fees[feeChain] = tmpFees
         }
-        fees[feeChain] = tmpFees
       }
 
       /* for (const chain of Object.keys(config.supportedNetworks)) {
@@ -436,7 +438,7 @@ export class C2DEngineDocker extends C2DEngine {
 
       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
         )