From ce513da18c354a3ff4e6321d6a6747e4c8fbc64f Mon Sep 17 00:00:00 2001 From: Intizar Date: Tue, 28 May 2024 12:54:51 +0900 Subject: [PATCH] rearrange execution sequence in state.add() --- core/src/listener/state.ts | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/core/src/listener/state.ts b/core/src/listener/state.ts index ed44582d4..3c959393f 100644 --- a/core/src/listener/state.ts +++ b/core/src/listener/state.ts @@ -182,9 +182,18 @@ export class State { await this.redisClient.set(this.stateName, JSON.stringify(updatedActiveListeners)) const contractAddress = toAddListener.address + const contract = new ethers.Contract(contractAddress, this.abi, this.provider) + this.contracts = { ...this.contracts, [contractAddress]: contract } + const latestBlock = await this.latestBlockNumber() const observedBlock = await getObservedBlock({ service: this.service }) + // fetch all unprocessed blocks and pass to the history queue + const unprocessedBlocks = await getUnprocessedBlocks({ service: this.service }) + for (const block of unprocessedBlocks) { + await this.addBlockToHistoryQueue(contractAddress, block.blockNumber) + } + // if there is no observedBlock record in the db, // use the listenerInitType to determine how to initialize the listener if (observedBlock.service === '') { @@ -201,27 +210,21 @@ export class State { default: // [block number] initialization - for (let blockNumber = this.listenerInitType; blockNumber < latestBlock; ++blockNumber) { + for (let blockNumber = this.listenerInitType; blockNumber <= latestBlock; ++blockNumber) { await this.addBlockToHistoryQueue(contractAddress, blockNumber) } - await upsertObservedBlock({ service: this.service, blockNumber: latestBlock - 1 }) + await upsertObservedBlock({ service: this.service, blockNumber: latestBlock }) break } } else { for ( let blockNumber = observedBlock.blockNumber + 1; - blockNumber < latestBlock; + blockNumber <= latestBlock; ++blockNumber ) { await this.addBlockToHistoryQueue(contractAddress, blockNumber) } - await upsertObservedBlock({ service: this.service, blockNumber: latestBlock - 1 }) - } - - // fetch all unprocessed blocks and pass to the history queue - const unprocessedBlocks = await getUnprocessedBlocks({ service: this.service }) - for (const block of unprocessedBlocks) { - await this.addBlockToHistoryQueue(contractAddress, block.blockNumber) + await upsertObservedBlock({ service: this.service, blockNumber: latestBlock }) } // Insert listener jobs @@ -236,9 +239,6 @@ export class State { } }) - const contract = new ethers.Contract(contractAddress, this.abi, this.provider) - this.contracts = { ...this.contracts, [contractAddress]: contract } - return toAddListener }