Fix inconsistent processed events with many dynamic contracts #399
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
DZakh
commented
Dec 24, 2024
| @@ -130,7 +130,7 @@ let registerDynamicContracts = ( | |||
| } else { | |||
| 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 😁
DZakh
commented
Dec 24, 2024
Comment on lines
-491
to
-511
| 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
Comment on lines
+753
to
+776
| 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.
| @@ -123,22 +139,9 @@ describe("PartitionedFetchState getMostBehindPartitions", () => { | |||
| fetchState0, | |||
| { | |||
| 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},
JasoonS
approved these changes
Dec 24, 2024
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.