Fixes for contract registration registers and partitions - #281

JonoPrest · 2024-10-23T14:18:59Z

Closes #245

The first bug as linked above was caused by the fact that I used a list to hold partitions. The partition id was it's index in this list. When adding new partitions, since it's a list it gets added to the head. Meaning that the partition ID was not static. The bug occurs when there's an inflight "fetch block range" query for a partition, during this time a new partition gets added (shifting the partition indexes along the list) and the query response gets routed to the wrong partition.

It obviously presents as a crash when there are dynamic contract registers and it can't find the unique register id on the wrong partition. But it can happen silently when they are root register queries. This is not so bad since the events are ordered but it can give us bad "latestFetchedBlock" states between partitions. Causing strange behaviours of which partitions to execute next, or incorrect UI representation of fetch state etc.

I've fixed this by changing the data structure to use a dict to map the partitions.

The second bug is subtle and never actually presented but I added a fix for it anyhow.

It occurs when

A dynamic contract fetchstate register is added upon a contract registration (say at block 100)
Next query is actioned and is inflight (say querying block 100-500)
A new fetchstate register is added between the first contract and it's upper bound query (say block 200)
A response happens from the first register up to the queried block, and merges into the new register before it has queried blocks 2-500. Resulting in a missed block range for that new register.

This has been fixed by adding an array of pending dynamic contract registerations to the fetchState and lazily registering them when the fetch state gets updated or when a "nextQuery" is requested meaning that the actual creation of the fetchState register never occurs when a query is in flight.

It may be worth reviewing this PR one commit at a time.

JonoPrest · 2024-10-23T14:19:31Z

-type dynamicContractRegistration = {
-  registeringEventBlockNumber: int,
-  registeringEventLogIndex: int,
-  registeringEventChain: ChainMap.Chain.t,
-  dynamicContracts: array<TablesStatic.DynamicContractRegistry.t>,
-}


Moved to FetchState.res to avoid cyclical dependencies

JonoPrest · 2024-10-23T14:20:35Z

+  let merge: (dict<'a>, dict<'a>) => dict<'a> = %raw(`(dictA, dictB) => ({...dictA, ...dictB})`)
+
+  let updateImmutable: (
+    dict<'a>,
+    string,
+    'a,
+  ) => dict<'a> = %raw(`(dict, key, value) => ({...dict, [key]: value})`)


I added these raw js functions as I thought that would be the most performant way to do these two operations.

You can see what merge was doing before and I think spreads would be most performant.

As I know spreads have terrible performance, so it's not guaranteed for them to be faster. But in the case probably it will be.

I haven't tested but would they be less performant than say:

let copied = Object.assign({}, dict); return Object.assign(copied, {[key]: value})

Or what method of immutable update structure would you choose? Can use a Belt.Map for eg.

My assumption was that a spread can be optimised by the js engine.

For this case, I'd just kept it as it is. I only have a comparison data with inlining the field set https://x.com/artalar_dev/status/1571839333694324736

For Object.assign need to do a benchmark, but I doubt there will be a difference which is worth to consider

JonoPrest · 2024-10-23T14:24:13Z

+type registerData = {
  latestFetchedBlock: blockNumberAndTimestamp,
  contractAddressMapping: ContractAddressingMap.mapping,
  //Events ordered from latest to earliest
  fetchedEventQueue: array<Types.eventBatchQueueItem>,
  //Used to prune dynamic contract registrations in the event
  //of a rollback.
  dynamicContracts: DynamicContractsMap.t,
-  isFetchingAtHead: bool,
  firstEventBlockNumber: option<int>,
 }

-type rec t = {
-  registerType: register,
-  ...fetchStateData,
+type rec register = {
+  registerType: registerType,
+  ...registerData,
+}


Some renaming going on over here. What was FetchState.t is now simply a "register", FetchState.t is a record that contains a "base register" and some additional state.

JonoPrest · 2024-10-23T14:25:22Z

+}
+and registerType =
+  | RootRegister({endBlock: option<int>})
+  | DynamicContractRegister({id: EventUtils.eventIndex, nextRegister: register})


I changed the payload of DynamicContractRegister to have a record argument rather than positional for more clear labelling.

JonoPrest · 2024-10-23T14:25:46Z

+type dynamicContractRegistration = {
+  registeringEventBlockNumber: int,
+  registeringEventLogIndex: int,
+  registeringEventChain: ChainMap.Chain.t,
+  dynamicContracts: array<TablesStatic.DynamicContractRegistry.t>,
+}


This is moved from EventProcessing.res to avoid cyclical deps

JonoPrest · 2024-10-23T14:26:18Z

  //Used to prune dynamic contract registrations in the event
  //of a rollback.
  dynamicContracts: DynamicContractsMap.t,
-  isFetchingAtHead: bool,


Removed isFetchingAtHead since not every register needs to have it's own value for this.

JonoPrest · 2024-10-23T14:27:29Z

+type t = {
+  baseRegister: register,
+  pendingDynamicContracts: array<dynamicContractRegistration>,
+  isFetchingAtHead: bool,


This is the new FetchState.t, it holds the baseRegister, an array of pending dynamic contracts that are to be lazily added and the "isFetchingAtHead" state is moved from each register to a single value here.

JonoPrest · 2024-10-23T14:29:50Z

+/**
+Inserts a dynamic contract register to the head of a given
+register. It will then precede the given register in the chain
+*/
+let addNewRegisterToHead = (
+  self,
+  ~registeringEventBlockNumber,
+  ~registeringEventLogIndex,
+  ~contractAddressMapping,
+) => {
+  let id: dynamicContractId = {
+    blockNumber: registeringEventBlockNumber,
+    logIndex: registeringEventLogIndex,
+  }
+  let registerType = DynamicContractRegister({id, nextRegister: self})
+
+  let dynamicContracts =
+    DynamicContractsMap.empty->DynamicContractsMap.add(
+      id,
+      contractAddressMapping->ContractAddressingMap.getAllAddresses,
+    )
+
+  {
+    registerType,
+    latestFetchedBlock: {
+      blockNumber: registeringEventBlockNumber - 1,
+      blockTimestamp: 0,
+    },
+    contractAddressMapping,
+    dynamicContracts,
+    fetchedEventQueue: [],
+    firstEventBlockNumber: None,
+  }
+}
+
+/**
+Adds a new dynamic contract registration. It inserts the registration ordered in the
+chain from earliest registered contract to latest. So if this is being called on a batch
+of registrations its best to do this in order of latest to earliest to reduce recursions
+of this function.
+*/
+let rec addDynamicContractRegister = (
+  self: register,
+  ~registeringEventBlockNumber,
+  ~registeringEventLogIndex,
+  ~dynamicContractRegistrations: array<TablesStatic.DynamicContractRegistry.t>,
+  ~parent: option<Parent.t>=?,
+) => {
+  let handleParent = updated =>
+    switch parent {
+    | Some(parent) => parent->Parent.joinChild(updated)
+    | None => updated
+    }
+
+  let addToHead = updated =>
+    updated
+    ->addNewRegisterToHead(
+      ~contractAddressMapping=dynamicContractRegistrations
+      ->Array.map(d => (d.contractAddress, (d.contractType :> string)))
+      ->ContractAddressingMap.fromArray,
+      ~registeringEventLogIndex,
+      ~registeringEventBlockNumber,
+    )
+    ->handleParent
+
+  let latestFetchedBlockNumber = registeringEventBlockNumber - 1
+
+  switch self.registerType {
+  | RootRegister(_) => self->addToHead
+  | DynamicContractRegister(_) if latestFetchedBlockNumber <= self.latestFetchedBlock.blockNumber =>
+    self->addToHead
+  | DynamicContractRegister({id: dynamicContractId, nextRegister}) =>
+    nextRegister->addDynamicContractRegister(
+      ~registeringEventBlockNumber,
+      ~registeringEventLogIndex,
+      ~dynamicContractRegistrations,
+      ~parent=self->Parent.make(~dynamicContractId, ~parent),
+    )
+  }
+}


These two functions were simply moved to be accessible to the functions below. No logic changes.

JonoPrest · 2024-10-23T14:37:13Z

+/**
+Adds a new dynamic contract registration. It appends the registration to the pending dynamic
+contract registrations. These pending registrations are applied to the base register when next
+query is called.
+*/
+let registerDynamicContract = (
+  self: t,
+  registration: dynamicContractRegistration,
+  ~isFetchingAtHead,
+) => {
+  ...self,
+  pendingDynamicContracts: self.pendingDynamicContracts->Array.concat([registration]),
+  isFetchingAtHead,
+}


When calling registerDynamicContract, it now simply appends to "pendingDynamicContracts" which will be handled later.

JonoPrest · 2024-10-23T14:42:58Z

+    let withNewDynamicContracts =
+      updatedRegister->addDynamicContractRegisters(pendingDynamicContracts)
+    let maybeMerged = withNewDynamicContracts->pruneAndMergeNextRegistered
+    {
+      baseRegister: maybeMerged->Option.getWithDefault(withNewDynamicContracts),
+      pendingDynamicContracts: [],
+      isFetchingAtHead,
+    }


Update now first:

adds the new items to the targeted register,

then registers any pending dynamic contracts

then merges registers

This ensures registers remain ordered based on latestFetched block and can't be messed around by in flight queries

JonoPrest · 2024-10-23T14:46:52Z

+  //First apply pending dynamic contracts, then try and merge
+  //These steps should only happen on getNextQuery, to avoid in between states where a
+  //query is in flight and the underlying registers are changing
+  let maybeUpdatedFetchState = switch self->applyPendingDynamicContractRegistrations {
+  | Some(updatedWithDynamicContracts) =>
+    //After adding the pending dynamic contracts, try and merge registers
+    switch updatedWithDynamicContracts->mapMaybeMerge {
+    //Pass through the merged value if it updated anything
+    | Some(merged) => Some(merged)
+    //Even if the merge returned none, the pending dynamic contracts should be applied
+    //as an updated
+    | None => Some(updatedWithDynamicContracts)
+    }
+  //If no dynamic contracts were added just try and merge
+  | None => self->mapMaybeMerge
+  }


getNextQuery now also first tries to apply pending dynamic contracts and then merge. Returning an optional updated state along with the query if anything changes.

JonoPrest · 2024-10-23T14:49:03Z

+let getLatestFullyFetchedBlock = (self: t) => {
+  //Consider pending dynamic contracts when calculating the latest fully fetched block
+  //Since they are now registered lazily on query or update of the fetchstate, not when
+  //the register function is called
+  let minPendingDynamicContracts = self.pendingDynamicContracts->Belt.Array.reduce(None, (
+    acc,
+    contract,
+  ) => {
+    let {registeringEventBlockNumber} = contract
+    minOfOption(registeringEventBlockNumber - 1, acc)->Some
+  })
+
+  switch (self.baseRegister.latestFetchedBlock, minPendingDynamicContracts) {
+  | ({blockNumber}, Some(pendingDynamicContractBlockNumber))
+    if pendingDynamicContractBlockNumber < blockNumber => {
+      blockNumber: pendingDynamicContractBlockNumber,
+      blockTimestamp: 0,
+    }
+  | (baseRegisterLatest, _) => baseRegisterLatest
+  }
+}


It used to be implicit that after a contract registeration the base register would contain the latest fetched block. But now there could be pending dynamic contracts so we need to check against those if there are any.

JonoPrest · 2024-10-23T14:53:59Z

+  newestPartitionIndex: partitionIndex,
+  partitions: dict<FetchState.t>,


Data structure changed as described in PR description.

I needed immutable operations with an int key so it might be sensible to use a belt map, but I thought we could rather stay closer to native js and just use a dict. I therefore am converting the key back and forth to string and also added a util for dict spreading.

JonoPrest · 2024-10-23T14:57:38Z

 ) => {
-  let partitions = switch partitions {
-  | list{head, ...tail} if head->FetchState.getNumContracts < maxAddrInPartition =>
+  let newestPartition = partitions->Js.Dict.unsafeGet(newestPartitionIndex->Int.toString)


Not sure if I should rather do a runtime check here and explicitly fail if the partition does not exist.

Let's do this, it won't harm.

I think unsafeGet is good, when we 100% sure that the item will be present in the context of the function. But if we're talking about the whole system, it's better to do a check. Especially when it doesn't cost us anything.

JonoPrest · 2024-10-23T15:00:46Z

-) =>
-  async chain => {
-    let chainFetcher = state.chainManager.chainFetchers->ChainMap.get(chain)
-    let {chainConfig: {chainWorker}, logger, currentBlockHeight} = chainFetcher
-
-    if !isRollingBack(state) {
-      let (nextQuery, nextStateIfChangeRequired) =
-        chainFetcher->ChainFetcher.getNextQuery(~maxPerChainQueueSize=state.maxPerChainQueueSize)
-
-      switch nextStateIfChangeRequired {


Most of these changes are just formatting. Either review with whitespace disabled or I separated the commit with the formatting changes so you should be able to see exactly what changed.

JonoPrest · 2024-10-23T15:03:37Z

+      baseRegister: {
+        registerType: RootRegister({endBlock: None}),
+        latestFetchedBlock: {
+          blockTimestamp: latestFetchedBlockTimestamp,
+          blockNumber: 0,
+        },
+        contractAddressMapping: ContractAddressingMap.make(),
+        fetchedEventQueue: item->Option.mapWithDefault([], v => [v]),
+        dynamicContracts: FetchState.DynamicContractsMap.empty,
+        firstEventBlockNumber: item->Option.map(v => v.blockNumber),


Simply fixing the test with the new data structure

JonoPrest · 2024-10-23T15:05:38Z

+  it(
+    "Adding dynamic between two registers while query is mid flight does no result in early merged registers",
+    () => {


This test validates the 2nd bug described in the PR discription

JonoPrest · 2024-10-23T15:07:33Z

    )
  })
+
+  it("Partition id never changes when adding new partitions", () => {


This test validates the first bug described is fixed

DZakh

Nice. Really liked the renamings in FetchState file. I couldn't process all of the changes.

Mostly I have 3 comments:

What about timestamp: 0
Use array for partitions
Consider logIndex for rollback of pending dynamic contracts

DZakh · 2024-10-24T13:07:59Z

+  | ({blockNumber}, Some(pendingDynamicContractBlockNumber))
+    if pendingDynamicContractBlockNumber < blockNumber => {
+      blockNumber: pendingDynamicContractBlockNumber,
+      blockTimestamp: 0,


Is it fine to have 0 here?

Yes, this is the same behaviour as before where when you create a dynamic contract register it initialises with blockTimestamp 0 since we don't know what the timestamp is for that block.

DZakh · 2024-10-24T13:10:48Z

+  let merge: (dict<'a>, dict<'a>) => dict<'a> = %raw(`(dictA, dictB) => ({...dictA, ...dictB})`)
+
+  let updateImmutable: (
+    dict<'a>,
+    string,
+    'a,
+  ) => dict<'a> = %raw(`(dict, key, value) => ({...dict, [key]: value})`)


As I know spreads have terrible performance, so it's not guaranteed for them to be faster. But in the case probably it will be.

DZakh · 2024-10-24T13:13:29Z

+  let getNewestPartitionIndex = () => {
+    switch newestPartitionIndexRef.contents {
+    | Some(newestPartitionIndex) => newestPartitionIndex
+    | None => Js.Exn.raiseError("Unexpected no part")


Should newestPartitionIndexRef.contents maybe always be int instead of option<int>?

It could be, just wasn't sure what the initial value should be. Could make it -1 for eg. to represent this case.

I think if you use an array, you won't need to have the counter anymore

DZakh · 2024-10-24T13:22:06Z

 ) => {
-  let partitions = switch partitions {
-  | list{head, ...tail} if head->FetchState.getNumContracts < maxAddrInPartition =>
+  let newestPartition = partitions->Js.Dict.unsafeGet(newestPartitionIndex->Int.toString)


Let's do this, it won't harm.

DZakh · 2024-10-24T13:23:24Z

 ) => {
-  let partitions = switch partitions {
-  | list{head, ...tail} if head->FetchState.getNumContracts < maxAddrInPartition =>
+  let newestPartition = partitions->Js.Dict.unsafeGet(newestPartitionIndex->Int.toString)


I think unsafeGet is good, when we 100% sure that the item will be present in the context of the function. But if we're talking about the whole system, it's better to do a check. Especially when it doesn't cost us anything.

DZakh · 2024-10-24T13:31:54Z


      let restartedFetchState =
-        restartedChainFetcher.fetchState.partitions->List.head->Option.getExn
+        restartedChainFetcher.fetchState.partitions->Js.Dict.values->Array.get(0)->Option.getExn


Since we have numerical ids, I think it's better to turn partitions into an array? This way we still be able to get items by id - aka hashmap + no need to always call Js.Dict.values

The newestPartitionIndex can be calculated from length

We could, the operation that I want to improve is updating a single partition in the array in an "immutable" way.

So I guess we can shallow copy the array and then set the given index on the array.

@DZakh, I just made some benchmarks here: https://jsbm.dev/Gwg2BVoQDxavT

Super interesting. Object.assign is the fastest of the 4 cases.

No, it's the slowest :)

And noticiably

Oh lol 🤣 I was reading the bars as runtime not ops/s

Sweet, I like the idea of using arrays 👍🏼

DZakh · 2024-10-24T13:35:17Z

+    self.pendingDynamicContracts->Array.keep(({registeringEventBlockNumber}) =>
+      registeringEventBlockNumber <= lastKnownValidBlock.blockNumber


Should it also consider logIndex?

No since when we rollback we always do a full block. If you look at the filters we add:

cf ->ChainFetcher.rollbackToLastBlockHashes(~rolledBackLastBlockData) ->ChainFetcher.addProcessingFilter( ~filter=eventBatchQueueItem => { let {timestamp, blockNumber} = eventBatchQueueItem //Filter out events that occur passed the block where the query starts but //are lower than the timestamp where we rolled back to (timestamp, chain->ChainMap.Chain.toChainId, blockNumber) > (lastKnownValidBlockTimestamp, rollbackChainId, lastKnownValidBlockNumber) }, ~isValid=(~fetchState) => { //Remove the event filter once the fetchState has fetched passed the //timestamp of the valid rollback block's timestamp let {blockTimestamp, blockNumber} = FetchState.getLatestFullyFetchedBlock(fetchState) (blockTimestamp, chain->ChainMap.Chain.toChainId, blockNumber) <= (lastKnownValidBlockTimestamp, rollbackChainId, lastKnownValidBlockNumber) }, )

We only care about the block number

DZakh

I left a few comments on Discord. But they are mostly small nitpicks. The PR is good to merge 👍

… inflight queries

…t added

Fix test passing new isFetchingAtHead flag

JonoPrest requested a review from DZakh October 23, 2024 14:18

JonoPrest commented Oct 23, 2024

View reviewed changes

JonoPrest force-pushed the jp/fix-register-does-not-exist branch from c7d2c4d to a77630b Compare October 23, 2024 14:40

JonoPrest commented Oct 23, 2024

View reviewed changes

JonoPrest force-pushed the jp/fix-register-does-not-exist branch 2 times, most recently from 1ffb583 to 7844914 Compare October 24, 2024 09:36

moose-code mentioned this pull request Oct 24, 2024

Skips event when using preRegisterDynamicContracts option #286

Closed

DZakh reviewed Oct 24, 2024

View reviewed changes

JonoPrest force-pushed the jp/fix-register-does-not-exist branch from 7844914 to b18f28e Compare October 24, 2024 14:51

JonoPrest requested a review from DZakh October 24, 2024 14:57

DZakh approved these changes Oct 24, 2024

View reviewed changes

JonoPrest added 3 commits October 25, 2024 10:54

Implement lazy dynamic contract registration and single register merge

539bc33

Apply formatting from project to global state

974312f

Apply updates to registers on update fn call

9640b6a

JonoPrest added 15 commits October 25, 2024 10:54

Fix tests

3229ede

Fix fetchstate test

e67ec63

Update data structure of partitions and fix wrong partition index

09b6523

Fix tests

f3de476

Add test to validate dynamic contract registration is not affected by…

1e15c77

… inflight queries

Add test to ensure partition ids do not change when new partitions ge…

7a69612

…t added

Ensure isFetchingAtHead gets handed down to fetch state

fc6fc83

Fix test passing new isFetchingAtHead flag

Move getNextQuery back to its original position in file

92b7231

Handle rollback of pending dynamic contracts

00f547b

Add unfinished error string

969de7c

Remove it_only from test

4b0c4ab

Change partitions from using a dict to an array

f4d1a9a

Fix tests

56e2e13

Add util for immutable array set index

cf2f182

Remove one iteration step from getMostBehindPartitions

fbfa346

JonoPrest force-pushed the jp/fix-register-does-not-exist branch from 4b3fcc6 to fbfa346 Compare October 25, 2024 08:54

JonoPrest enabled auto-merge (squash) October 25, 2024 08:54

JonoPrest merged commit 5fe4dbf into main Oct 25, 2024

JonoPrest deleted the jp/fix-register-does-not-exist branch October 25, 2024 09:02

		newestPartitionIndex: partitionIndex,
		partitions: dict<FetchState.t>,

		self.pendingDynamicContracts->Array.keep(({registeringEventBlockNumber}) =>
		registeringEventBlockNumber <= lastKnownValidBlock.blockNumber

Uh oh!

Conversation

JonoPrest commented Oct 23, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

DZakh Oct 24, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

DZakh left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JonoPrest commented Oct 23, 2024 •

edited

Loading

DZakh Oct 24, 2024 •

edited

Loading