Refactor EVM node synchronization (#958)

Author: droserasprout

CHANGELOG.md

@@ -4,6 +4,17 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog], and this project adheres to [Semantic Versioning].
 
+## [Unreleased]
+
+### Fixed
+
+- evm.node: Fixed default ratelimit sleep time being too high.
+- evm.subsquid.transactions: Fixed issue with `node_only` flag ignored.
+
+### Performance
+
+- evm.subsquid: Dynamically adjust the batch size when syncing with node.
+
 ## [7.5.0] - 2024-03-08
 
 ### Added

Makefile

@@ -31,7 +31,7 @@ black:          ## Format with black
 	black ${SOURCE}
 
 ruff:           ## Lint with ruff
-	ruff check --fix ${SOURCE}
+	ruff check --fix --unsafe-fixes ${SOURCE}
 
 mypy:           ## Lint with mypy
 	mypy ${SOURCE}

src/dipdup/datasources/evm_node.py

@@ -88,8 +88,8 @@ async def wait_level(self) -> None:
 
 class EvmNodeDatasource(IndexDatasource[EvmNodeDatasourceConfig]):
     _default_http_config = HttpConfig(
-        batch_size=32,
-        ratelimit_sleep=30,
+        batch_size=10,
+        ratelimit_sleep=1,
         polling_interval=1.0,
     )
 

src/dipdup/fetcher.py

@@ -136,6 +136,8 @@ async def fetch(self) -> None:
 
 
 class DataFetcher(ABC, Generic[FetcherBufferT]):
+    """Fetches contract data from REST API, merges them and yields by level."""
+
     def __init__(
         self,
         datasource: IndexDatasource[Any],
@@ -149,4 +151,9 @@ def __init__(
         self._head = 0
 
     @abstractmethod
-    def fetch_by_level(self) -> AsyncIterator[tuple[int, tuple[FetcherBufferT, ...]]]: ...
+    def fetch_by_level(self) -> AsyncIterator[tuple[int, tuple[FetcherBufferT, ...]]]:
+        """Iterate over events data from REST.
+
+        Resulting data is splitted by level, deduped, sorted and ready to be processed by TzktEventsIndex.
+        """
+        ...

src/dipdup/indexes/evm_node.py

@@ -0,0 +1,89 @@
+import asyncio
+import random
+from abc import ABC
+from collections import defaultdict
+from collections import deque
+from typing import Any
+from typing import Generic
+
+from dipdup.datasources.evm_node import EvmNodeDatasource
+from dipdup.exceptions import FrameworkException
+from dipdup.fetcher import DataFetcher
+from dipdup.fetcher import FetcherBufferT
+
+EVM_NODE_READAHEAD_LIMIT = 5000
+MIN_BATCH_SIZE = 10
+MAX_BATCH_SIZE = 10000
+BATCH_SIZE_UP = 1.1
+BATCH_SIZE_DOWN = 0.5
+
+
+class EvmNodeFetcher(Generic[FetcherBufferT], DataFetcher[FetcherBufferT], ABC):
+    def __init__(
+        self,
+        datasources: tuple[EvmNodeDatasource, ...],
+        first_level: int,
+        last_level: int,
+    ) -> None:
+        super().__init__(datasources[0], first_level, last_level)
+        self._datasources = datasources
+
+    def get_next_batch_size(self, batch_size: int, ratelimited: bool) -> int:
+        if ratelimited:
+            batch_size = int(batch_size * BATCH_SIZE_DOWN)
+        else:
+            batch_size = int(batch_size * BATCH_SIZE_UP)
+
+        batch_size = min(MAX_BATCH_SIZE, batch_size)
+        batch_size = max(MIN_BATCH_SIZE, batch_size)
+        return int(batch_size)
+
+    def get_random_node(self) -> EvmNodeDatasource:
+        if not self._datasources:
+            raise FrameworkException('A node datasource requested, but none attached to this index')
+        return random.choice(self._datasources)
+
+    async def get_blocks_batch(
+        self,
+        levels: set[int],
+        full_transactions: bool = False,
+        node: EvmNodeDatasource | None = None,
+    ) -> dict[int, dict[str, Any]]:
+        tasks: deque[asyncio.Task[Any]] = deque()
+        blocks: dict[int, Any] = {}
+        node = node or self.get_random_node()
+
+        async def _fetch(level: int) -> None:
+            blocks[level] = await node.get_block_by_level(
+                block_number=level,
+                full_transactions=full_transactions,
+            )
+
+        for level in levels:
+            tasks.append(
+                asyncio.create_task(
+                    _fetch(level),
+                    name=f'get_block_range:{level}',
+                ),
+            )
+
+        await asyncio.gather(*tasks)
+        return blocks
+
+    async def get_logs_batch(
+        self,
+        first_level: int,
+        last_level: int,
+        node: EvmNodeDatasource | None = None,
+    ) -> dict[int, list[dict[str, Any]]]:
+        grouped_logs: defaultdict[int, list[dict[str, Any]]] = defaultdict(list)
+        node = node or self.get_random_node()
+        logs = await node.get_logs(
+            {
+                'fromBlock': hex(first_level),
+                'toBlock': hex(last_level),
+            },
+        )
+        for log in logs:
+            grouped_logs[int(log['blockNumber'], 16)].append(log)
+        return grouped_logs

src/dipdup/indexes/evm_subsquid.py

@@ -1,9 +1,6 @@
-import asyncio
 import random
 from abc import ABC
 from abc import abstractmethod
-from collections import defaultdict
-from collections import deque
 from typing import Any
 from typing import Generic
 from typing import TypeVar
@@ -27,6 +24,8 @@
 from dipdup.package import DipDupPackage
 from dipdup.prometheus import Metrics
 
+SUBSQUID_READAHEAD_LIMIT = 5000
+
 IndexConfigT = TypeVar('IndexConfigT', bound=SubsquidIndexConfigU)
 DatasourceT = TypeVar('DatasourceT', bound=SubsquidDatasource)
 
@@ -86,11 +85,6 @@ def node_datasources(self) -> tuple[EvmNodeDatasource, ...]:
     def datasources(self) -> tuple[IndexDatasource[Any], ...]:
         return (self.datasource, *self.node_datasources)
 
-    def get_random_node(self) -> EvmNodeDatasource:
-        if not self._node_datasources:
-            raise FrameworkException('A node datasource requested, but none attached to this index')
-        return random.choice(self._node_datasources)
-
     def get_sync_level(self) -> int:
         """Get level index needs to be synchronized to depending on its subscription status"""
         sync_levels = set()
@@ -108,51 +102,6 @@ def get_sync_level(self) -> int:
         # NOTE: Choose the highest level; outdated realtime messages will be dropped from the queue anyway.
         return max(cast(set[int], sync_levels))
 
-    async def get_blocks_batch(
-        self,
-        levels: set[int],
-        full_transactions: bool = False,
-        node: EvmNodeDatasource | None = None,
-    ) -> dict[int, dict[str, Any]]:
-        tasks: deque[asyncio.Task[Any]] = deque()
-        blocks: dict[int, Any] = {}
-        node = node or self.get_random_node()
-
-        async def _fetch(level: int) -> None:
-            blocks[level] = await node.get_block_by_level(
-                block_number=level,
-                full_transactions=full_transactions,
-            )
-
-        for level in levels:
-            tasks.append(
-                asyncio.create_task(
-                    _fetch(level),
-                    name=f'get_block_range:{level}',
-                ),
-            )
-
-        await asyncio.gather(*tasks)
-        return blocks
-
-    async def get_logs_batch(
-        self,
-        first_level: int,
-        last_level: int,
-        node: EvmNodeDatasource | None = None,
-    ) -> dict[int, list[dict[str, Any]]]:
-        grouped_logs: defaultdict[int, list[dict[str, Any]]] = defaultdict(list)
-        node = node or self.get_random_node()
-        logs = await node.get_logs(
-            {
-                'fromBlock': hex(first_level),
-                'toBlock': hex(last_level),
-            },
-        )
-        for log in logs:
-            grouped_logs[int(log['blockNumber'], 16)].append(log)
-        return grouped_logs
-
     async def _get_node_sync_level(
         self,
         subsquid_level: int,
@@ -161,7 +110,7 @@ async def _get_node_sync_level(
     ) -> int | None:
         if not self.node_datasources:
             return None
-        node = node or self.get_random_node()
+        node = node or random.choice(self.node_datasources)
 
         node_sync_level = await node.get_head_level()
         subsquid_lag = abs(node_sync_level - subsquid_level)

src/dipdup/indexes/evm_subsquid_events/fetcher.py

@@ -1,14 +1,20 @@
+import random
+import time
 from collections.abc import AsyncIterator
 
+from dipdup.datasources.evm_node import EvmNodeDatasource
 from dipdup.datasources.evm_subsquid import SubsquidDatasource
 from dipdup.fetcher import DataFetcher
 from dipdup.fetcher import readahead_by_level
+from dipdup.indexes.evm_node import EVM_NODE_READAHEAD_LIMIT
+from dipdup.indexes.evm_node import MIN_BATCH_SIZE
+from dipdup.indexes.evm_node import EvmNodeFetcher
+from dipdup.indexes.evm_subsquid import SUBSQUID_READAHEAD_LIMIT
+from dipdup.models.evm_node import EvmNodeLogData
 from dipdup.models.evm_subsquid import SubsquidEventData
 
 
-class EventLogFetcher(DataFetcher[SubsquidEventData]):
-    """Fetches contract events from REST API, merges them and yields by level."""
-
+class SubsquidEventFetcher(DataFetcher[SubsquidEventData]):
     _datasource: SubsquidDatasource
 
     def __init__(
@@ -22,14 +28,73 @@ def __init__(
         self._topics = topics
 
     async def fetch_by_level(self) -> AsyncIterator[tuple[int, tuple[SubsquidEventData, ...]]]:
-        """Iterate over events fetched fetched from REST.
-
-        Resulting data is splitted by level, deduped, sorted and ready to be processed by TzktEventsIndex.
-        """
         event_iter = self._datasource.iter_event_logs(
             self._topics,
             self._first_level,
             self._last_level,
         )
-        async for level, batch in readahead_by_level(event_iter, limit=5_000):
+        async for level, batch in readahead_by_level(event_iter, limit=SUBSQUID_READAHEAD_LIMIT):
+            yield level, batch
+
+
+class EvmNodeEventFetcher(EvmNodeFetcher[EvmNodeLogData]):
+    _datasource: EvmNodeDatasource
+
+    async def fetch_by_level(self) -> AsyncIterator[tuple[int, tuple[EvmNodeLogData, ...]]]:
+        event_iter = self._fetch_by_level()
+        async for level, batch in readahead_by_level(event_iter, limit=EVM_NODE_READAHEAD_LIMIT):
             yield level, batch
+
+    async def _fetch_by_level(self) -> AsyncIterator[tuple[EvmNodeLogData, ...]]:
+        batch_size = MIN_BATCH_SIZE
+        batch_first_level = self._first_level
+        ratelimited: bool = False
+
+        while batch_first_level <= self._last_level:
+            node = random.choice(self._datasources)
+            batch_size = self.get_next_batch_size(batch_size, ratelimited)
+            ratelimited = False
+
+            started = time.time()
+
+            batch_last_level = min(
+                batch_first_level + batch_size,
+                self._last_level,
+            )
+            log_batch = await self.get_logs_batch(
+                batch_first_level,
+                batch_last_level,
+                node,
+            )
+
+            finished = time.time()
+            if finished - started >= node._http_config.ratelimit_sleep:
+                ratelimited = True
+
+            timestamps: dict[int, int] = {}
+            log_levels = list(log_batch.keys())
+
+            # NOTE: Split log_levels to chunks of batch_size
+            log_level_batches = [set(log_levels[i : i + batch_size]) for i in range(0, len(log_levels), batch_size)]
+
+            for log_level_batch in log_level_batches:
+
+                started = time.time()
+
+                block_batch = await self.get_blocks_batch(log_level_batch)
+                for level, block in block_batch.items():
+                    timestamps[level] = int(block['timestamp'], 16)
+
+                finished = time.time()
+                if finished - started >= node._http_config.ratelimit_sleep:
+                    ratelimited = True
+
+            for level, level_logs in log_batch.items():
+                if not level_logs:
+                    continue
+
+                parsed_level_logs = tuple(EvmNodeLogData.from_json(log, timestamps[level]) for log in level_logs)
+
+                yield parsed_level_logs
+
+            batch_first_level = batch_last_level + 1