util: improve processor metrics

common/changes/@subsquid/util-internal-processor-tools/processor-metrics-fix_2024-12-31-13-44.json

@@ -0,0 +1,15 @@
+{
+  "changes": [
+    {
+      "packageName": "@subsquid/util-internal-processor-tools",
+      "comment": "bound processor metrics chain height with request range",
+      "type": "patch"
+    },
+    {
+      "packageName": "@subsquid/util-internal-processor-tools",
+      "comment": "allow to disable processor exit",
+      "type": "minor"
+    }
+  ],
+  "packageName": "@subsquid/util-internal-processor-tools"
+}

util/util-internal-processor-tools/src/runner.ts

@@ -43,7 +43,7 @@ export class Runner<R, S> {
         if (this.getLeftRequests(state).length == 0) {
             this.printProcessingRange()
             log.info('nothing to do')
-            return
+            return this.exit()
         }
 
         this.printProcessingMessage(state)
@@ -110,9 +110,11 @@ export class Runner<R, S> {
             state = nextState
         }
 
-        return this.processHotBlocks(state).finally(
+        await this.processHotBlocks(state).finally(
             this.chainHeightUpdateLoop(hot)
         )
+
+        return this.exit()
     }
 
     private async assertWeAreOnTheSameChain(src: DataSource<unknown, unknown>, state: HashAndHeight): Promise<void> {
@@ -306,7 +308,8 @@ export class Runner<R, S> {
 
     private async initMetrics(chainHeight: number, state: HotDatabaseState): Promise<void> {
         let initialized = this.metrics.getChainHeight() >= 0
-        this.metrics.setChainHeight(chainHeight)
+        let boundedChainHeight = Math.min(chainHeight, rangeEnd(last(this.config.requests).range))
+        this.metrics.setChainHeight(boundedChainHeight)
         if (initialized) return
         this.metrics.setLastProcessedBlock(state.height + state.top.length)
         this.metrics.updateProgress()
@@ -343,6 +346,14 @@ export class Runner<R, S> {
         this.log.info(msg)
     }
 
+    private async exit(): Promise<void> {
+        if (isProcessorExitDisabled()) {
+            while (true) {
+                await wait(5000)
+            }
+        }
+    }
+
     @def
     private async startPrometheusServer(): Promise<void> {
         let prometheusServer = await this.config.prometheus.serve()
@@ -353,3 +364,7 @@ export class Runner<R, S> {
         return this.config.log
     }
 }
+
+function isProcessorExitDisabled() {
+    return process.env.SQUID_PROCESSOR_EXIT_DISABLED === 'true'
+}