Refactor Fetch State to allow dynamically manage partitions - #405

DZakh · 2024-12-26T14:10:03Z

Refactor FetchState to allow dynamically control partitions.

Fixes rollback with wildcard events.

DZakh · 2024-12-27T19:58:46Z

+          )
+        | Some(nextResponse) =>
+          switch nextResponse.reorgGuard.firstBlockParentNumberAndHash {
+          | None => Js.Exn.raiseError("Unexpected, nextResponse reorgGuard is missing firstBlockParentNumberAndHash")


Actually got the error at block 30,948,833 of Gnosis indexer.

JonoPrest · 2025-01-06T09:37:29Z

  }
+
+  @send
+  external flatten: (array<array<'a>>, @as(json`1`) _) => array<'a> = "flat"


Interesting, never done a binding like this with hardcoded params :) does @as(1) not work as well out of interest?

Yeah, it works 😁 I didn't know that it does :)

JonoPrest · 2025-01-06T09:50:46Z

+    ->Utils.Dict.dangerouslyGetNonOption(chain->ChainMap.Chain.toString)
+    ->Option.mapWithDefault(false, dcs =>
+      dcs->Array.some(dc =>
+        dc.contractType == contractType && dc.contractAddress == contractAddress


I actually think we need to remove this check. I removed it in all the other places I think this was left by mistake. We are not allowing the same address registered to multiple contracts.

Suggested change

dc.contractType == contractType && dc.contractAddress == contractAddress

dc.contractAddress == contractAddress

JonoPrest · 2025-01-06T14:14:43Z

+      fetchedEventQueue: [],
+      contractAddressMapping: restContractAddressMapping,
+      dynamicContracts: restDcs,
+      latestFetchedBlock,


Should there be any validation in this function that latestFetchedBlock is exactly the same between the partitions being merged?

I thought it's not necessary, since the function won't run when the latestFetchedBlock is not the same (according to the setQueryResponse logic). And this is checked by integration tests.

JonoPrest · 2025-01-06T15:10:51Z

+  switch query {
+  | PartitionQuery({partitionId})
+  | MergeQuery({partitionId}) =>
+    switch partitions->Array.getIndexBy(p => p.id === partitionId) {


This actually makes me think we should rather use a dict over an array. Not a big deal though is these arrays will never get massive.

Or we could always just maintain indexes by using delete on an index of the array 🤔

During the refactoring I changed it to dict, but then returned it back, because it was more annoying to write mocks in tests :)

JonoPrest · 2025-01-06T15:22:06Z

      currentBlockHeight <= latestFetchedBlock.blockNumber,
+      ~firstEventBlockNumber=switch newItems->Array.get(0) {
+      | Some(newFirstItem) =>
+        Utils.Math.minOptInt(fetchState.firstEventBlockNumber, Some(newFirstItem.blockNumber))


Looks like we can just use min and not minOptInt

@DZakh just checking if you saw this comment

JonoPrest · 2025-01-06T16:18:51Z

  {
    maxAddrInPartition,
-    partitions,
+    partitions: [partition],


This is temporary right now right?

Yes, the PartitionedFetchState is now 100% useless, I didn't remove it in the PR only to minimize already huge diff.

JonoPrest · 2025-01-06T16:21:19Z

+  let partitionId = switch query {
+  | PartitionQuery({partitionId})
+  | MergeQuery({partitionId}) => partitionId
+  }


Maybe query should keep shared fields outside of the variant constr payload

I'll consider this suggestion, when I work on partition kind.

JonoPrest · 2025-01-07T08:22:22Z

+  | PartitionQuery({partitionId, fromBlock, toBlock, contractAddressMapping}) =>
+    chainWorker->ChainWorker.fetchBlockRange(
+      ~fromBlock,
+      ~toBlock,
+      ~contractAddressMapping,
+      ~partitionId,
+      ~chain,
+      ~currentBlockHeight,
+      ~isPreRegisteringDynamicContracts,
+      ~logger,
+      //Only apply wildcards on the first partition
+      //to avoid duplicate wildcard queries
+      ~shouldApplyWildcards=partitionId === "0",
+    )
+  | MergeQuery({partitionId, fromBlock, toBlock, contractAddressMapping}) =>
+    chainWorker->ChainWorker.fetchBlockRange(
+      ~fromBlock,
+      ~toBlock=Some(toBlock),
+      ~contractAddressMapping,
+      ~partitionId,
+      ~chain,
+      ~currentBlockHeight,
+      ~isPreRegisteringDynamicContracts,
+      ~logger,
+      //Only apply wildcards on the first partition
+      //to avoid duplicate wildcard queries
+      ~shouldApplyWildcards=partitionId === "0",


These two seem so similar it looks like we should be able to have more shared data in query. Maybe even just currying the function above would look better 😝

JonoPrest · 2025-01-07T08:23:10Z

+    } = chainFetcher
+
+    await chainFetcher.sourceManager->SourceManager.fetchNext(
+      ~fetchState=partitionedFetchState.partitions->Js.Array2.unsafe_get(0),


I think lets get rid of partitionedFetchState ASAP

Let's do it in the next PR

JonoPrest · 2025-01-07T08:55:11Z

-    let mergedRegister = switch registerByLatestBlock->Utils.Dict.dangerouslyGetNonOption(key) {
-    | Some(next) => register->mergeWithNextRegister(~next)
-    | None => register
+  let firstPartition = partitions->Js.Array2.unsafe_get(0)


Just a note here @DZakh, I've been testing your branch on an indexer that reorgs a lot this morning and a reorg crashes the indexer here. First partition does not exist in the case I ran in to.

I fixed it yesterday, could you double check that you have the latest version
10fdcd9

JonoPrest · 2025-01-07T14:03:55Z

  ~chain,
  ~contractAddress,
-  ~contractType,
+  ~contractType as _,


Can we remove this field? Or add todo

I'll do in the next PR

Cool just don't want to forget about it 👍🏼

JonoPrest · 2025-01-07T14:09:30Z

+  let fetchStatesMap = fetchStatesMap->ChainMap.map(v => {
+    {
+      ...v,
+      partitionedFetchState: v.partitionedFetchState->PartitionedFetchState.syncStateOnQueueUpdate,
+    }
+  })
+


Maybe this should be done inside getFetchStateWithData function on the shouldDeepCopy flag or with a different flag.

During shouldDeepCopy, we don't know the number of events taken from the queue to batch, so we can't correctly update the state. I was actually thinking of changing the way how the batch is created so the part of the code is more reliable, but I scoped it out since this was not mandatory for the code to work.

Ah I see, well what should probably happen is the popItem function should decrement the queue size. But all good

JonoPrest

I've taken in the most I can with reviewing. Everything looks good to me. It's difficult to comprehend the whole change and everything it touches but I think we can move forward 👍🏼

Nice refactor Dmitry, lots of improvements all over.

DZakh changed the base branch from main to dz/respect-rpc-suggested-interval December 26, 2024 14:10

DZakh marked this pull request as draft December 26, 2024 14:10

Base automatically changed from dz/respect-rpc-suggested-interval to main December 27, 2024 15:04

DZakh added 4 commits December 27, 2024 19:08

Improve fethcing idempotency logic - unconnect it from register id

3f77d7a

Start implementing merge query

938903c

Make it query

2ca07c4

Handle merge query correctly with reorgs

118b5c6

DZakh force-pushed the dz/concurrent-catch-up-query branch from 4f43c5a to 118b5c6 Compare December 27, 2024 15:08

DZakh commented Dec 27, 2024

View reviewed changes

DZakh added 8 commits December 30, 2024 13:56

Redesign fetching and concurrency

b584c87

Handle all chain registrations at once

4bd1f0d

Group dc registrations by block number

7de0989

Start updating tests

2b496fb

Use the current query selection logic

7598773

Finish partition splitting logic

013d460

Start writign FetchState tests

f009dd0

Add more tests and fix some bugs

b87877a

JonoPrest reviewed Jan 6, 2025

View reviewed changes

DZakh added 2 commits January 6, 2025 14:49

Fix rollback

10fdcd9

Fix concurrency limit

efb8205

JonoPrest reviewed Jan 6, 2025

View reviewed changes

Fix stateId handling and uncomment tests

c4be087

JonoPrest reviewed Jan 6, 2025

View reviewed changes

JonoPrest reviewed Jan 7, 2025

View reviewed changes

DZakh added 3 commits January 7, 2025 15:54

Fix test_codegen tests

d884bf1

Fix the rest of tests

db7d537

Merge branch 'main' into dz/concurrent-catch-up-query

17adadb

DZakh marked this pull request as ready for review January 7, 2025 12:34

Final improvements of the prometheus logging

8001d74

DZakh requested a review from JonoPrest January 7, 2025 13:09

JonoPrest reviewed Jan 7, 2025

View reviewed changes

JonoPrest approved these changes Jan 7, 2025

View reviewed changes

DZakh added 4 commits January 7, 2025 19:00

Add failing test

d77f63b

Fix qItemLt

d624d6a

Remove partitioned fetch state (#407)

fa46f2b

Final fixes

528fb11

DZakh changed the title ~~Concurrent catch up query for registered dynamic contracts~~ Refactor Fetch State to allow dynamically manage partitions Jan 8, 2025

DZakh merged commit 97d2f22 into main Jan 8, 2025

DZakh deleted the dz/concurrent-catch-up-query branch January 8, 2025 11:22

DZakh mentioned this pull request Jan 9, 2025

Query Wildcard events concurrently with static events #409

Merged

	dc.contractType == contractType && dc.contractAddress == contractAddress
	dc.contractAddress == contractAddress

Uh oh!

Conversation

DZakh commented Dec 26, 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!

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JonoPrest left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

DZakh commented Dec 26, 2024 •

edited

Loading