From bc975ea89094ac53644c912eaa9881a6db2f038b Mon Sep 17 00:00:00 2001 From: Morgan Mccauley Date: Mon, 13 Nov 2023 13:11:32 +1300 Subject: [PATCH] fix: Ensure historical invocations are configured correctly --- runner/src/stream-handler/worker.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/runner/src/stream-handler/worker.ts b/runner/src/stream-handler/worker.ts index ae0d29c56..785c755a9 100644 --- a/runner/src/stream-handler/worker.ts +++ b/runner/src/stream-handler/worker.ts @@ -105,6 +105,7 @@ async function blockQueueConsumer (workerContext: WorkerContext, streamKey: stri const startTime = performance.now(); const blockStartTime = startTime; const block = queueMessage.block; + const isHistorical = workerContext.streamType === 'historical'; streamMessageId = queueMessage.streamMessageId; if (block === undefined || block.blockHeight == null) { @@ -112,14 +113,14 @@ async function blockQueueConsumer (workerContext: WorkerContext, streamKey: stri continue; } METRICS.BLOCK_WAIT_DURATION.labels({ indexer: indexerName, type: workerContext.streamType }).set(performance.now() - blockStartTime); - await indexer.runFunctions(block, functions, false, { provision: true }); + await indexer.runFunctions(block, functions, isHistorical, { provision: true }); await workerContext.redisClient.deleteStreamMessage(streamKey, streamMessageId); await workerContext.queue.shift(); METRICS.EXECUTION_DURATION.labels({ indexer: indexerName, type: workerContext.streamType }).observe(performance.now() - startTime); - if (workerContext.streamType === 'historical') { + if (isHistorical) { METRICS.LAST_PROCESSED_BLOCK_HEIGHT.labels({ indexer: indexerName, type: workerContext.streamType }).set(block.blockHeight); }