Introduce mutable fetch state - #385

DZakh · 2024-12-12T16:41:22Z

Fix a stale fetching partition on reorg
Improve test coverage of the core fetch execution logic from ~5% up to ~90%
Potentially fixes an issue with double fetching of the same partition, which could lead to a corrupted data (not proved)
Cleans up and removes unused code

…-fetch-state

DZakh · 2024-12-13T13:19:55Z

-let applyConditionalFunction = (value: 'a, condition: bool, callback: 'a => 'b) => {
-  condition ? callback(value) : value
-}
-
-let queryEventsWithCombinedFilter = async (
-  ~contractInterfaceManager,
-  ~fromBlock,
-  ~toBlock,
-  ~minFromBlockLogIndex=0,
-  ~provider,
-  ~logger: Pino.t,
-): array<Ethers.log> => {
-  let combinedFilterRes = await makeCombinedEventFilterQuery(
-    ~provider,
-    ~contractInterfaceManager,
-    ~fromBlock,
-    ~toBlock,
-    ~logger,
-  )
-
-  combinedFilterRes->applyConditionalFunction(minFromBlockLogIndex > 0, arrLogs => {
-    arrLogs->Belt.Array.keep(log => {
-      log.blockNumber > fromBlock ||
-        (log.blockNumber == fromBlock && log.logIndex >= minFromBlockLogIndex)
-    })
-  })
-}


I've noticed that we never set minFromBlockLogIndex to something besides 0, so the filter was never used.

DZakh · 2024-12-13T13:20:27Z

-let eventIndexSchema = S.object(s => {
-  blockNumber: s.field("blockNumber", S.int),
-  logIndex: s.field("logIndex", S.int),
-})


Remove unused

DZakh · 2024-12-13T13:21:31Z

-  let (displayAddr, restCount) = addressesAll->Array.reduce(([], 0), (
-    (currentDisplay, restCount),
-    addr,
-  ) => {
-    if currentDisplay->Array.length < 3 {
-      (Array.concat(currentDisplay, [addr->Address.toString]), restCount)
-    } else {
-      (currentDisplay, restCount + 1)
-    }
-  })


The Array.reduce could actually affect peformance for indexers with many addresses. Optimised it a little bit.

DZakh · 2024-12-13T13:22:00Z

-    Logging.childTrace(
-      logger,
-      {
-        "msg": "adding contract address",
-        "contractName": contractName,
-        "address": address,
-      },
-    )


Should I keep the log? I don't see a big use in it

No I don't think so.

DZakh · 2024-12-13T13:22:29Z

-/**
-Instantiates a fetch state with root register
-*/
-let makeRoot = (~endBlock) => makeInternal(~registerType=RootRegister({endBlock: endBlock}), ...)


Combined makeInternal and makeRoot into a single make.

DZakh · 2024-12-13T13:23:15Z

-
-/**
-Retrieves an array of partitions that are most behind with a max number based on
-the max number of queries with the context of the partitions currently fetching.
-
-The array could be shorter than the max number of queries if the partitions are
-at the max queue size.
-*/
-let getMostBehindPartitions = (
-  {partitions}: t,
-  ~maxNumQueries,


Moved this to SourceMap and covered with tests

DZakh · 2024-12-13T13:24:07Z

+let toQueryId = (query: FetchState.nextQuery) => {
+  query.fetchStateRegisterId->FetchState.registerIdToString ++ "-" ++ query.fromBlock->Int.toString
+}
+
+// Can't simply store fetching partitions, since fetchBatch
+// can be called with old chainFetchers after isFetching was set to false,
+// but state isn't still updated with fetched data.
+// This is temporary until all fetching logic moved to mutable state,
+// but for now prevent fetching the same partition twice with the same query,
+// using lastFetchedQueryId (aka idempotency key)
+type partitionFetchingState = {
+  isFetching: bool,
+  lastFetchedQueryId?: string,
+}


We'll be able to get rid of these, after the code start managing fetching itself instead of being called by global state module.

DZakh · 2024-12-13T13:24:50Z

+let fetchBatch = async (
+  sourceManger: t,
+  ~allPartitions: PartitionedFetchState.allPartitions,
+  ~currentBlockHeight,
+  ~executePartitionQuery,
+  ~waitForNewBlock,
+  ~onNewBlock,
+  ~maxPerChainQueueSize,
+  ~setMergedPartitions,
+  ~stateId,
+) => {


Even though it looks complicated, but now it has almost 100% test coverage, so I find it a drastic improvement.

DZakh · 2024-12-13T13:27:20Z

+      // DynamicContract
+      // fromBlock: 0
+      // toBlock: 0
+      await dispatchAllTasks()
+      // DynamicContract
+      // fromBlock: 0
+      // toBlock: 3
+      await dispatchAllTasks()
+      // DynamicContract
+      // fromBlock: 2
+      // toBlock: 3
      await dispatchAllTasks()


Previously all 3 queries of the same fetchState somehow managed to run during a single await dispatchAllTasks() call. Now it the run one by one, thanks to the mutable fetching partitions set.

DZakh · 2024-12-13T13:27:53Z

-  it(
-    "With multiple partitions always returns the most behind partitions up to the max concurrency level",
-    () => {
-      let maxNumQueries = 3
-      let partitionsCurrentlyFetching = Set.Int.empty
-
-      let mostBehindPartitions =
-        partitionedFetchState->PartitionedFetchState.getMostBehindPartitions(
-          ~maxNumQueries,
-          ~maxPerChainQueueSize=10,
-          ~partitionsCurrentlyFetching,
-        )


Redone the tests in SourceManager_test.res

JonoPrest · 2024-12-13T14:53:29Z

+  ~stateId,
+) => {
+  if stateId < sourceManager.currentStateId {
+    ()


Maybe it's worth logging a trace or a warning for this case?

I don't think that it'll be useful. The fetchBatch function currently might be called however times you want, and it won't be an issue. Also, I hope to get rid of the check at some point, when we move all the fetching state to the module.

JonoPrest · 2024-12-13T15:01:28Z

+        switch allPartitionsFetchingState->Belt.Array.get(partitionId) {
+        // Deduplicate queries when fetchBatch is called after
+        // isFetching was set to false, but state isn't updated with fetched data
+        | Some({lastFetchedQueryId}) if lastFetchedQueryId === toQueryId(nextQuery) => None


Out of interest does === vs == have any effect on strings?

No, it's the same generated code, I just don't like using == both in Js and ReScript 😅

JonoPrest · 2024-12-13T15:13:11Z

+    | // For the case with currentBlockHeight=0 we should
+    // force getting the known chain block, even if there are no ready queries


Does this handle a chain with endblock?

Yes. It filters out all completely fetched partitions

JonoPrest · 2024-12-13T15:21:07Z

-        }
-        true
-      }
+    let {chainConfig: {chainWorker}, logger, currentBlockHeight, fetchState} = chainFetcher


should probably rename fetchState to partitionedFetchState on the chain fetcher

JonoPrest

Hey @DZakh, from a high level it looks good. I don't have the time right now to pick apart everything and I haven't had a chance to review each of your tests.

I would suggest being cautious about releasing without running it on a known indexer but also looks like you've tested the parts thoroughly

DZakh added 7 commits December 11, 2024 16:12

Improve getQueryLogger perf

7f8c918

Remove unused rpc events filter

b3545a3

Move maxPartitionConcurrency check after all the filtering is done

375ae24

Clean up maxQueriesNumber

907e769

Clean up FetchState.make function

ea0cc5d

Start moving fetching state to mutable

0d45c6a

Add mutable fetch state and fully test fetchBatch logic

6c9455d

DZakh requested a review from JonoPrest December 12, 2024 16:41

DZakh added 3 commits December 13, 2024 14:04

Fix most of the tests

936f5ff

Finish fixing tests

1d82031

Merge branch 'main' of github.com:enviodev/hyperindex into dz/mutable…

cb777c5

…-fetch-state

DZakh commented Dec 13, 2024

View reviewed changes

DZakh changed the title ~~Dz/mutable fetch state~~ Introduce mutable fetch state Dec 13, 2024

Merge branch 'main' into dz/mutable-fetch-state

1baa466

JonoPrest reviewed Dec 13, 2024

View reviewed changes

Comment thread codegenerator/cli/templates/static/codegen/src/eventFetching/ChainFetcher.res Outdated

Fix typo

aa49fce

JonoPrest reviewed Dec 13, 2024

View reviewed changes

JonoPrest approved these changes Dec 13, 2024

View reviewed changes

Merge branch 'main' into dz/mutable-fetch-state

6e9cd1b

DZakh merged commit 5605da1 into main Dec 13, 2024

DZakh deleted the dz/mutable-fetch-state branch December 13, 2024 16:15

		\| // For the case with currentBlockHeight=0 we should
		// force getting the known chain block, even if there are no ready queries

Uh oh!

Conversation

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

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 12, 2024 •

edited

Loading