fix: Write UNPROCESSED_STREAM_MESSAGES metric on both failure/succe…

…ss (#352)
near · Nov 1, 2023 · 2096c39 · 2096c39
1 parent 50a3d6c
commit 2096c39
Showing 1 changed file with 9 additions and 9 deletions.
diff --git a/runner/src/stream-handler/worker.ts b/runner/src/stream-handler/worker.ts
@@ -19,11 +19,11 @@ void (async function main () {
   console.log('Started processing stream: ', streamKey);
 
   let indexerName = '';
+  const streamType = redisClient.getStreamType(streamKey);
 
   while (true) {
     try {
       const startTime = performance.now();
-      const streamType = redisClient.getStreamType(streamKey);
 
       const messages = await redisClient.getNextStreamMessage(streamKey);
       const indexerConfig = await redisClient.getStreamStorage(streamKey);
@@ -52,14 +52,6 @@ void (async function main () {
 
       await redisClient.deleteStreamMessage(streamKey, id);
 
-      const unprocessedMessages = await redisClient.getUnprocessedStreamMessages(streamKey);
-
-      parentPort?.postMessage({
-        type: 'UNPROCESSED_STREAM_MESSAGES',
-        labels: { indexer: indexerName, type: streamType },
-        value: unprocessedMessages?.length ?? 0,
-      } satisfies Message);
-
       parentPort?.postMessage({
         type: 'EXECUTION_DURATION',
         labels: { indexer: indexerName, type: streamType },
@@ -70,6 +62,14 @@ void (async function main () {
     } catch (err) {
       await sleep(10000);
       console.log(`Failed: ${indexerName}`, err);
+    } finally {
+      const unprocessedMessages = await redisClient.getUnprocessedStreamMessages(streamKey);
+
+      parentPort?.postMessage({
+        type: 'UNPROCESSED_STREAM_MESSAGES',
+        labels: { indexer: indexerName, type: streamType },
+        value: unprocessedMessages?.length ?? 0,
+      } satisfies Message);
     }
   }
 })();