Get next query improvements - #364
Conversation
| let rec getKnownBlockWithBackoff = async (~provider, ~blockNumber, ~backoffMsOnFailure) => | ||
| switch await getKnownBlock(provider, blockNumber) { | ||
| | exception err => | ||
| Logging.warn({ | ||
| "err": err, | ||
| "msg": `Issue while running fetching batch of events from the RPC. Will wait ${backoffMsOnFailure->Belt.Int.toString}ms and try again.`, | ||
| "type": "EXPONENTIAL_BACKOFF", | ||
| }) | ||
| await Time.resolvePromiseAfterDelay(~delayMilliseconds=backoffMsOnFailure) | ||
| await getUnwrappedBlockWithBackoff( | ||
| await getKnownBlockWithBackoff( |
There was a problem hiding this comment.
Changed the naming a little bit
| let getNextQueries = (self: t, ~maxPerChainQueueSize) => { | ||
| self.fetchState->PartitionedFetchState.getNextQueries( |
There was a problem hiding this comment.
This doesn't interact with currentBlockHeight anymore. I think it made things easier.
| exception FromBlockIsHigherThanToBlock(int, int) //from and to block respectively | ||
|
|
||
| let isGreaterThanOpt: (int, option<int>) => bool = (a: int, b: option<int>) => { | ||
| switch b { | ||
| | Some(b) => a > b | ||
| | None => false | ||
| } | ||
| } | ||
|
|
||
| let rec getEndBlock = (self: register) => { | ||
| switch self.registerType { | ||
| | RootRegister({endBlock}) => endBlock | ||
| | DynamicContractRegister({nextRegister}) => nextRegister->getEndBlock | ||
| } | ||
| } |
| type nextQueryOrWaitForBlock = | ||
| type nextQueryOrDone = | ||
| | NextQuery(nextQuery) | ||
| | WaitForNewBlock |
There was a problem hiding this comment.
Moved all the interactions with global state to the global state module. Here we simply prepare a query and don't care how it'll be executed.
| /** | ||
| Gets the next query either with a to block of the current height if it is the root node. | ||
| Or with a toBlock of the nextRegistered latestBlockNumber to catch up and merge with the next regisetered. | ||
|
|
||
| Errors if nextRegistered dynamic contract has a lower latestFetchedBlock than the current as this would be | ||
| an invalid state. | ||
| */ | ||
| let getNextQuery = (self: t, ~currentBlockHeight, ~partitionId) => { |
There was a problem hiding this comment.
Splitted it into two functions for reading simplicity
| toBlockExclusive: ?switch toBlockInclusive { | ||
| | Some(toBlockInclusive) => Some(toBlockInclusive + 1) | ||
| | None => None | ||
| }, |
There was a problem hiding this comment.
I tried to compare the latest envio version with this one. Looked at TUI of the Fuel testnet indexer. I didn't really notice any difference, but I think it'll be more noticeable when HyperSync/network is lagging and takes more time to respond.
There was a problem hiding this comment.
Not related but maybe it would be worth firing that hook on a throttler as well. Makes the TUI less jittery
There was a problem hiding this comment.
Probably firing the hook as we do and throttle the render itself.
| switch toBlock { | ||
| | Some(toBlock) if toBlock < fromBlock => | ||
| //This is an invalid case. We should never arrive at this match arm but it would be | ||
| //detrimental if it were the case. | ||
| FromBlockIsHigherThanToBlock({fromBlock, toBlock})->ErrorHandling.mkLogAndRaise( |
There was a problem hiding this comment.
Moved the validation here. Now I don't really see a reason for this 😅
| switch self.stateUpdatedHook { | ||
| | Some(hook) => hook(nextState) | ||
| | None => () | ||
| | Some(hook) if self.state !== nextState => hook(nextState) |
There was a problem hiding this comment.
Trigger the hook only when there's a change
|
@JonoPrest Assigned you for an early review. The most concerning part is the potential stop of the indexer #364 (comment) . Let's sync when you have time, if my explanation doesn't make sense. Also, I've realised that my solution will cause triggering the new block polling multiple times, I'd like to fix it. |
| registerType, | ||
| latestFetchedBlock: { | ||
| blockTimestamp: 0, | ||
| // Here's a bug that startBlock: 1 won't work |
There was a problem hiding this comment.
Ah I see, maybe we just allow a negative number here
| let includesWaitForNewBlock = ref(false) | ||
| let getNextQueries = (self: t, ~maxPerChainQueueSize, ~partitionsCurrentlyFetching) => { | ||
| let nextQueries = [] | ||
| let updatedPartitions = Js.Dict.empty() |
There was a problem hiding this comment.
I don't really see the purpose of returning this dict? Why not return the fully updated PartitionedFetchState?
There was a problem hiding this comment.
No particular reason, but it allows to skip the state update when there are no merges
| { | ||
| ...state, | ||
| chainManager: { | ||
| ...state.chainManager, | ||
| chainFetchers: state.chainManager.chainFetchers->ChainMap.update(chain, chainFetcher => { | ||
| ...chainFetcher->updateChainFetcherCurrentBlockHeight(~currentBlockHeight), | ||
| isWaitingForNewBlock: false, | ||
| }), | ||
| }, | ||
| }, |
There was a problem hiding this comment.
Looks like this can use the updateChainFetcher helper like above. In fact looks like these two actions could be combined to pass "isWaitingForNewBlock" and just set that value.
There was a problem hiding this comment.
Nvmnd, see it includes current blockheight 👍🏼 but can still use the helper fn
…balStateManager.res Co-authored-by: Jono Prest <65739024+JonoPrest@users.noreply.github.com>
JonoPrest
left a comment
JonoPrest
left a comment
There was a problem hiding this comment.
Great I like the changes!
…hainWorkers/ChainWorker.res Co-authored-by: Jono Prest <65739024+JonoPrest@users.noreply.github.com>
2 participants
This may improve latency a little bit.