Fix inconsistent processed events with many dynamic contracts - #399
Conversation
| let newPartition = FetchState.make( | ||
| ~partitionId=partitions->Array.length, | ||
| ~startBlock, | ||
| ~startBlock=dynamicContractRegistration.registeringEventBlockNumber, |
There was a problem hiding this comment.
This is a fix for the first issue. I had to fix the second one with a huge refactoring to be able to reproduce and find the little first one 😁
There was a problem hiding this comment.
Ah man, I had a feeling there was something going wrong when creating new partitions. Good catch
| if register.latestFetchedBlock.blockNumber > nextRegister.latestFetchedBlock.blockNumber { | ||
| let logger = Logging.createChild( | ||
| ~params={ | ||
| "context": "Merging Dynamic Contract Registers", | ||
| "currentRegister": { | ||
| "id": register->getRegisterId, | ||
| "latestFetchedBlock": register.latestFetchedBlock.blockNumber, | ||
| "addresses": register.contractAddressMapping->ContractAddressingMap.getAllAddresses, | ||
| }, | ||
| "nextRegister": { | ||
| "id": nextRegister->getRegisterId, | ||
| "latestFetchedBlock": nextRegister.latestFetchedBlock.blockNumber, | ||
| "addresses": nextRegister.registerType->isRootRegister | ||
| ? "Root"->Utils.magic | ||
| : nextRegister.contractAddressMapping->ContractAddressingMap.getAllAddresses, | ||
| }, | ||
| }, | ||
| ) | ||
| NextRegisterIsLessThanCurrent->ErrorHandling.mkLogAndRaise( | ||
| ~msg="Unexpected: Dynamic contract register latest fetched block is greater than next register when it should be equal", | ||
| ~logger, |
There was a problem hiding this comment.
This is not possible anymore
| let rollbackRegister = ( | ||
| register: register, | ||
| ~lastScannedBlock, | ||
| ~firstChangeEvent: blockNumberAndLogIndex, | ||
| ~parent: option<Parent.t>=?, | ||
| ) => { | ||
| let handleParent = updated => | ||
| switch parent { | ||
| | Some(parent) => parent->Parent.joinChild(updated) | ||
| | None => updated | ||
| } | ||
|
|
||
| switch self.registerType { | ||
| //Case 1 Root register that has only fetched up to a confirmed valid block number | ||
| //Should just return itself unchanged | ||
| | RootRegister if self.latestFetchedBlock.blockNumber < firstChangeEvent.blockNumber => | ||
| self->handleParent | ||
| //Case 2 Dynamic register that has only fetched up to a confirmed valid block number | ||
| //Should just return itself, with the next register rolled back recursively | ||
| | DynamicContractRegister({id, nextRegister}) | ||
| if self.latestFetchedBlock.blockNumber < firstChangeEvent.blockNumber => | ||
| nextRegister->rollbackRegister( | ||
| ~lastScannedBlock, | ||
| ~firstChangeEvent, | ||
| ~parent=self->Parent.make(~dynamicContractId=id, ~parent), | ||
| ) | ||
|
|
||
| //Case 3 Root register that has fetched further than the confirmed valid block number | ||
| //Should prune its queue and set its latest fetched block data to the latest known confirmed block | ||
| | RootRegister => | ||
| { | ||
| ...self, | ||
| fetchedEventQueue: self.fetchedEventQueue->pruneQueueFromFirstChangeEvent(~firstChangeEvent), | ||
| latestFetchedBlock: lastScannedBlock, | ||
| } | ||
| ->pruneDynamicContractAddressesFromFirstChangeEvent(~firstChangeEvent) | ||
| ->handleParent | ||
| //Case 4 DynamicContract register that has fetched further than the confirmed valid block number | ||
| //Should prune its queue, set its latest fetched blockdata + pruned queue | ||
| //And recursivle prune the nextRegister | ||
| | DynamicContractRegister({id, nextRegister}) => | ||
| if register.latestFetchedBlock.blockNumber < firstChangeEvent.blockNumber { | ||
| Some(register) | ||
| } else { | ||
| let updatedWithRemovedDynamicContracts = | ||
| self->pruneDynamicContractAddressesFromFirstChangeEvent(~firstChangeEvent) | ||
|
|
||
| register->pruneDynamicContractAddressesFromFirstChangeEvent(~firstChangeEvent) | ||
| if updatedWithRemovedDynamicContracts.contractAddressMapping->ContractAddressingMap.isEmpty { | ||
| //If the contractAddressMapping is empty after pruning dynamic contracts, then this | ||
| //is a dead register. Simly return its next register rolled back | ||
| nextRegister->rollbackRegister(~lastScannedBlock, ~firstChangeEvent, ~parent?) | ||
| //If the contractAddressMapping is empty after pruning dynamic contracts, | ||
| // then this is a dead register. | ||
| None | ||
| } else { | ||
| //If there are still values in the contractAddressMapping, we should keep the register but | ||
| //prune queues and next register | ||
| let updated = { | ||
| //If there are still values in the contractAddressMapping, | ||
| //we should keep the register but prune queues | ||
| Some({ | ||
| ...updatedWithRemovedDynamicContracts, | ||
| fetchedEventQueue: self.fetchedEventQueue->pruneQueueFromFirstChangeEvent( | ||
| fetchedEventQueue: register.fetchedEventQueue->pruneQueueFromFirstChangeEvent( | ||
| ~firstChangeEvent, | ||
| ), | ||
| latestFetchedBlock: lastScannedBlock, | ||
| } | ||
| nextRegister->rollbackRegister( | ||
| ~lastScannedBlock, | ||
| ~firstChangeEvent, | ||
| ~parent=updated->Parent.make(~dynamicContractId=id, ~parent), | ||
| ) | ||
| }) |
There was a problem hiding this comment.
This is absolutely the same, but it is so much simpler without a recursion.
| partitionId: 1, | ||
| baseRegister: { | ||
| registerType: RootRegister, | ||
| latestFetchedBlock: {blockNumber: 0, blockTimestamp: 0}, |
There was a problem hiding this comment.
Here's the wrong test for the first issue. After the fix it became:
latestFetchedBlock: {blockNumber: 9, blockTimestamp: 0},
| // Must use the reference to copied value, so we use find | ||
| mostBehindRegister: registers | ||
| ->Js.Array2.find(r => r.id == self.mostBehindRegister.id) | ||
| ->Option.getExn, | ||
| nextMostBehindRegister: switch self.nextMostBehindRegister { | ||
| | Some(nextMostBehindRegister) => | ||
| registers->Js.Array2.find(r => r.id == nextMostBehindRegister.id)->Option.getExn->Some | ||
| | None => None | ||
| }, |
There was a problem hiding this comment.
Just a thought, maybe worth using an array index here rather than an obj reference? Not sure of the performance hit though.
Looks something like an arena/index tree data structure.
There was a problem hiding this comment.
I think a reference should be faster, since we don't need to get it every time we need it. Regarding memory, it should be the same. In any way the difference is neglectable.
Also, find looks dirty here, but I've decided to keep it as it is, since I was thinking to get rid of mostBehindRegister and nextMostBehindRegister when I turn registers into partitions.
| ~registers=fetchState.registers, | ||
| ~pendingDynamicContracts=fetchState.pendingDynamicContracts, | ||
| ~isFetchingAtHead=fetchState.isFetchingAtHead, |
There was a problem hiding this comment.
Woah, I didn't know you could default to another param in rescript 😅
| let item = ref(registers->Js.Array2.unsafe_get(0)->getEarliestEventInRegister) | ||
| for idx in 1 to registers->Js.Array2.length - 1 { | ||
| let register = registers->Js.Array2.unsafe_get(idx) | ||
| let registerItem = register->getEarliestEventInRegister | ||
| if registerItem->qItemLt(item.contents) { | ||
| item := registerItem | ||
| } | ||
| } | ||
| item.contents | ||
| } |
There was a problem hiding this comment.
Is this much faster than reduce?
If so looks like we should just make this into our own reduce util?
There was a problem hiding this comment.
I suppose the speed difference with our number of registers is neglectable, but I think regarding the indexer code we should strive for the better, especially when it doesn't cost anything.
I'm not really into the reduce util. There's only one pro in using reduce - readability, but this is a very arguable and I find the go-like style code equally readable, especially when it's scoped inside of a small function covered with tests.
There was a problem hiding this comment.
Well there's a number of places this pattern is used. If it's faster we should make a util for it try convert all other reduce use cases to this IMO.
It's not really readability since as you say procedural code often reads easier, rather composability/reusability and safety from the off by ones/bad index accessing etc. Those bugs can creep in from time to time.
| | RootRegister => accum | ||
| | DynamicContractRegister({nextRegister}) => nextRegister->getRegisterAddressMaps(~accum) | ||
| } | ||
| // FIXME: Should include pending contracts? |
There was a problem hiding this comment.
Yes it definitely should!
There was a problem hiding this comment.
I didn't fix it since I planned to change the way we split partitions
3 participants
Terminology:
We have two fetching modes:
There were two main issues:
To fix the second issue, I've change the ordered list-like structure for registers to an independent array, simplifying operations with them, and allowing to addition new dynamic registrations after the latestFetchBlock. Also, this will simplify the further improvement to fetch the dynamic contract registers in parallel. Currently, the fetch requests to catch up to the latest register are done sequentially.