Fix parsing block timestamp (#825)

Author: droserasprout

pyproject.toml

@@ -147,7 +147,7 @@ ignore = [
 target-version = "py311"
 extend-select = ["B", "C4", "FA", "G", "I", "PTH", "Q", "RET", "RUF", "TCH", "UP"]
 flake8-quotes = { inline-quotes = "single", multiline-quotes = "double" }
-isort = { force-single-line = true }
+isort = { force-single-line = true, known-first-party = ["dipdup"] }
 
 [tool.mypy]
 python_version = "3.11"

src/dipdup/datasources/evm_node.py

@@ -3,6 +3,8 @@
 from collections import defaultdict
 from collections.abc import Awaitable
 from collections.abc import Callable
+from dataclasses import dataclass
+from dataclasses import field
 from typing import Any
 from uuid import uuid4
 
@@ -43,6 +45,13 @@
 RollbackCallback = Callable[['IndexDatasource', MessageType, int, int], Awaitable[None]]
 
 
+@dataclass
+class NodeHead:
+    event: asyncio.Event = field(default_factory=asyncio.Event)
+    hash: str | None = None
+    timestamp: int | None = None
+
+
 class EvmNodeDatasource(IndexDatasource[EvmNodeDatasourceConfig]):
     # TODO: Make dynamic
     _default_http_config = HttpConfig(ratelimit_sleep=30)
@@ -54,8 +63,7 @@ def __init__(self, config: EvmNodeDatasourceConfig, merge_subscriptions: bool =
         self._realtime: asyncio.Event = asyncio.Event()
         self._requests: dict[str, tuple[asyncio.Event, Any]] = {}
         self._subscription_ids: dict[str, EvmNodeSubscription] = {}
-        self._head_hashes: dict[int, str] = {}
-        self._head_events: defaultdict[int, asyncio.Event] = defaultdict(asyncio.Event)
+        self._heads: defaultdict[int, NodeHead] = defaultdict(NodeHead)
 
         self._on_connected_callbacks: set[EmptyCallback] = set()
         self._on_disconnected_callbacks: set[EmptyCallback] = set()
@@ -267,28 +275,30 @@ async def _on_message(self, message: Message) -> None:
             raise DatasourceError(f'Unknown message: {data}', self.name)
 
     async def _handle_subscription(self, subscription: EvmNodeSubscription, data: Any) -> None:
-        msg_type = MessageType()
-
         if isinstance(subscription, EvmNodeNewHeadsSubscription):
             head = EvmNodeHeadData.from_json(data)
             level = int(head.number, 16)
 
-            known_hash = self._head_hashes.get(level, None)
-            current_level = max(self._head_hashes.keys() or (level,))
-            self._head_hashes[level] = head.hash
-            if known_hash != head.hash:
-                await self.emit_rollback(msg_type, from_level=current_level, to_level=level - 1)
+            known_hash = self._heads[level].hash
+            if known_hash not in (head.hash, None):
+                await self.emit_rollback(
+                    MessageType(),
+                    from_level=max(self._heads.keys() or (level,)),
+                    to_level=level - 1,
+                )
 
+            self._heads[level].hash = head.hash
+            self._heads[level].timestamp = int(head.timestamp, 16)
             await self.emit_head(head)
-            self._head_events[level].set()
+            self._heads[level].event.set()
 
         elif isinstance(subscription, EvmNodeLogsSubscription):
-            # FIXME: Take from the head instead.
-            timestamp = int(time.time())
+            level = int(data['blockNumber'], 16)
+            await self._heads[level].event.wait()
+            timestamp = self._heads[level].timestamp
+            if timestamp is None:
+                raise FrameworkException('Head cached but timestamp is None')
             logs = EvmNodeLogData.from_json(data, timestamp)
-            level = int(logs.block_number, 16)
-
-            await self._head_events[level].wait()
             await self.emit_logs(logs)
 
         elif isinstance(subscription, EvmNodeSyncingData):

src/dipdup/indexes/evm_subsquid_events/index.py

@@ -167,7 +167,8 @@ async def _synchronize(self, sync_level: int) -> None:
                 block = await self.random_node.get_block_by_level(level)
                 if block is None:
                     raise FrameworkException(f'Block {level} not found')
-                parsed_level_logs = tuple(EvmNodeLogData.from_json(log, int(block['timestamp'])) for log in level_logs)
+                timestamp = int(block['timestamp'], 16)
+                parsed_level_logs = tuple(EvmNodeLogData.from_json(log, timestamp) for log in level_logs)
                 await self._process_level_events(parsed_level_logs, self.topics, sync_level)
 
         else:

tests/configs/demo_evm_events.yml

@@ -0,0 +1,33 @@
+spec_version: 2.0
+package: demo_evm_events
+
+datasources:
+  ethscan:
+    kind: abi.etherscan
+
+  # mainnet_node:
+  #   kind: evm.node
+  #   url: https://mainnet.infura.io/v3/${INFURA_KEY:-''}
+  #   ws_url: wss://mainnet.infura.io/ws/v3/${INFURA_KEY:-''}
+
+  mainnet_subsquid:
+    kind: evm.subsquid
+    url: ${ARCHIVE_URL:-https://v2.archive.subsquid.io/network/ethereum-mainnet}
+    # node: mainnet_node
+
+contracts:
+  eth_usdt:
+    kind: evm
+    address: 0xdac17f958d2ee523a2206206994597c13d831ec7
+    typename: eth_usdt
+
+indexes:
+  eth_usdt_events:
+    kind: evm.subsquid.events
+    datasource: mainnet_subsquid
+    handlers:
+      - callback: on_transfer
+        contract: eth_usdt
+        name: Transfer
+    first_level: 18077421
+    last_level: 18077421

tests/test_demos.py

@@ -135,9 +135,8 @@ async def assert_init(package: str) -> None:
 
 
 async def assert_run_dex() -> None:
-    from tortoise.transactions import in_transaction
-
     import demo_dex.models
+    from tortoise.transactions import in_transaction
 
     trades = await demo_dex.models.Trade.filter().count()
     positions = await demo_dex.models.Position.filter().count()
@@ -180,6 +179,13 @@ async def assert_run_raw() -> None:
     assert migrations == 2
 
 
+async def assert_run_evm_events() -> None:
+    import demo_evm_events.models
+
+    holders = await demo_evm_events.models.Holder.filter().count()
+    assert holders == 5296
+
+
 async def assert_run_dao() -> None:
     import demo_dao.models
 
@@ -226,6 +232,8 @@ async def assert_run_dao() -> None:
     ('demo_factories.yml', 'demo_factories', 'init', partial(assert_init, 'demo_factories')),
     ('demo_raw.yml', 'demo_raw', 'run', assert_run_raw),
     ('demo_raw.yml', 'demo_raw', 'init', partial(assert_init, 'demo_raw')),
+    ('demo_evm_events.yml', 'demo_evm_events', 'run', assert_run_evm_events),
+    ('demo_evm_events.yml', 'demo_evm_events', 'init', partial(assert_init, 'demo_evm_events')),
 )
 
 

tests/test_models.py

@@ -5,8 +5,8 @@
 from typing import Any
 
 import orjson as json
-
 from demo_domains.types.name_registry.tezos_storage import NameRegistryStorage
+
 from dipdup.indexes.tezos_tzkt_operations.parser import deserialize_storage
 from dipdup.models.tezos_tzkt import TzktOperationData
 from tests.types.asdf.storage import AsdfStorage

tests/test_rollback.py

@@ -1,10 +1,10 @@
 from contextlib import AsyncExitStack
 from datetime import datetime
 
-from tortoise.expressions import F
-
 import demo_domains.models as domains_models
 import demo_nft_marketplace.models as hen_models
+from tortoise.expressions import F
+
 from dipdup.config import DipDupConfig
 from dipdup.context import HookContext
 from dipdup.dipdup import DipDup