Allow using evm.node as index datasource (#997)

CHANGELOG.md

@@ -4,6 +4,12 @@ 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]
+
+### Added
+
+- evm.subsquid: `evm.node` datasources can be used as index datasources.
+
 ## [7.5.4] - 2024-04-09
 
 ### Fixed

docs/7.references/2.config.md

@@ -219,12 +219,12 @@ description: "Config file reference"
 <dt class="field-odd" style="color: var(--txt-primary);">Parameters<span class="colon">:</span></dt>
 <dd class="field-odd"><ul class="simple">
 <li><p><strong>kind</strong> (<em>Literal</em><em>[</em><em>'evm.subsquid.events'</em><em>]</em>) – Always ‘evm.subsquid.events’</p></li>
-<li><p><strong>datasource</strong> (<em>str</em><em> | </em><a class="reference internal" href="#dipdupconfigevm_subsquidsubsquiddatasourceconfig" title="dipdup.config.evm_subsquid.SubsquidDatasourceConfig" target="_self"><em>SubsquidDatasourceConfig</em></a>) – Subsquid datasource</p></li>
+<li><p><strong>datasource</strong> (<em>str</em><em> | </em><a class="reference internal" href="#dipdupconfigevm_subsquidsubsquiddatasourceconfig" title="dipdup.config.evm_subsquid.SubsquidDatasourceConfig" target="_self"><em>SubsquidDatasourceConfig</em></a><em> | </em><a class="reference internal" href="#dipdupconfigevm_nodeevmnodedatasourceconfig" title="dipdup.config.evm_node.EvmNodeDatasourceConfig" target="_self"><em>EvmNodeDatasourceConfig</em></a>) – Subsquid datasource</p></li>
 <li><p><strong>handlers</strong> (<em>tuple</em><em>[</em><a class="reference internal" href="#dipdupconfigevm_subsquid_eventssubsquideventshandlerconfig" title="dipdup.config.evm_subsquid_events.SubsquidEventsHandlerConfig" target="_self"><em>SubsquidEventsHandlerConfig</em></a><em>, </em><em>...</em><em>]</em>) – Event handlers</p></li>
 <li><p><strong>abi</strong> (<a class="reference internal" href="#dipdupconfigabidatasourceconfig" title="dipdup.config.AbiDatasourceConfig" target="_self"><em>AbiDatasourceConfig</em></a><em> | </em><em>tuple</em><em>[</em><a class="reference internal" href="#dipdupconfigabidatasourceconfig" title="dipdup.config.AbiDatasourceConfig" target="_self"><em>AbiDatasourceConfig</em></a><em>, </em><em>...</em><em>] </em><em>| </em><em>None</em>) – One or more <cite>evm.abi</cite> datasource(s) for the same network</p></li>
-<li><p><strong>node_only</strong> (<em>bool</em>) – Don’t use Subsquid Network API (dev only)</p></li>
 <li><p><strong>first_level</strong> (<em>int</em>) – Level to start indexing from</p></li>
 <li><p><strong>last_level</strong> (<em>int</em>) – Level to stop indexing and disable this index</p></li>
+<li><p><strong>node_only</strong> (<em>bool</em>) – </p></li>
 </ul>
 </dd>
 </dl>
@@ -258,7 +258,7 @@ description: "Config file reference"
 <dt class="field-odd" style="color: var(--txt-primary);">Parameters<span class="colon">:</span></dt>
 <dd class="field-odd"><ul class="simple">
 <li><p><strong>kind</strong> (<em>str</em>) – starts with ‘evm.subsquid’</p></li>
-<li><p><strong>datasource</strong> (<em>str</em><em> | </em><a class="reference internal" href="#dipdupconfigevm_subsquidsubsquiddatasourceconfig" title="dipdup.config.evm_subsquid.SubsquidDatasourceConfig" target="_self"><em>SubsquidDatasourceConfig</em></a>) – Subsquid datasource config</p></li>
+<li><p><strong>datasource</strong> (<em>str</em><em> | </em><a class="reference internal" href="#dipdupconfigevm_subsquidsubsquiddatasourceconfig" title="dipdup.config.evm_subsquid.SubsquidDatasourceConfig" target="_self"><em>SubsquidDatasourceConfig</em></a><em> | </em><a class="reference internal" href="#dipdupconfigevm_nodeevmnodedatasourceconfig" title="dipdup.config.evm_node.EvmNodeDatasourceConfig" target="_self"><em>EvmNodeDatasourceConfig</em></a>) – Subsquid datasource config</p></li>
 </ul>
 </dd>
 </dl>
@@ -293,12 +293,12 @@ description: "Config file reference"
 <dt class="field-odd" style="color: var(--txt-primary);">Parameters<span class="colon">:</span></dt>
 <dd class="field-odd"><ul class="simple">
 <li><p><strong>kind</strong> (<em>Literal</em><em>[</em><em>'evm.subsquid.transactions'</em><em>]</em>) – always ‘evm.subsquid.transactions’</p></li>
-<li><p><strong>datasource</strong> (<em>str</em><em> | </em><a class="reference internal" href="#dipdupconfigevm_subsquidsubsquiddatasourceconfig" title="dipdup.config.evm_subsquid.SubsquidDatasourceConfig" target="_self"><em>SubsquidDatasourceConfig</em></a>) – Subsquid datasource config</p></li>
+<li><p><strong>datasource</strong> (<em>str</em><em> | </em><a class="reference internal" href="#dipdupconfigevm_subsquidsubsquiddatasourceconfig" title="dipdup.config.evm_subsquid.SubsquidDatasourceConfig" target="_self"><em>SubsquidDatasourceConfig</em></a><em> | </em><a class="reference internal" href="#dipdupconfigevm_nodeevmnodedatasourceconfig" title="dipdup.config.evm_node.EvmNodeDatasourceConfig" target="_self"><em>EvmNodeDatasourceConfig</em></a>) – Subsquid datasource config</p></li>
 <li><p><strong>handlers</strong> (<em>tuple</em><em>[</em><a class="reference internal" href="#dipdupconfigevm_subsquid_transactionssubsquidtransactionshandlerconfig" title="dipdup.config.evm_subsquid_transactions.SubsquidTransactionsHandlerConfig" target="_self"><em>SubsquidTransactionsHandlerConfig</em></a><em>, </em><em>...</em><em>]</em>) – Transaction handlers</p></li>
 <li><p><strong>abi</strong> (<a class="reference internal" href="#dipdupconfigabidatasourceconfig" title="dipdup.config.AbiDatasourceConfig" target="_self"><em>AbiDatasourceConfig</em></a><em> | </em><em>tuple</em><em>[</em><a class="reference internal" href="#dipdupconfigabidatasourceconfig" title="dipdup.config.AbiDatasourceConfig" target="_self"><em>AbiDatasourceConfig</em></a><em>, </em><em>...</em><em>] </em><em>| </em><em>None</em>) – One or many ABI datasource(s)</p></li>
-<li><p><strong>node_only</strong> (<em>bool</em>) – Don’t use Subsquid Network, only node RPC</p></li>
 <li><p><strong>first_level</strong> (<em>int</em>) – Level to start indexing from</p></li>
 <li><p><strong>last_level</strong> (<em>int</em>) – Level to stop indexing at</p></li>
+<li><p><strong>node_only</strong> (<em>bool</em>) – </p></li>
 </ul>
 </dd>
 </dl>

schema.json

@@ -2003,6 +2003,13 @@
                   "$ref": "#/$defs/SubsquidDatasourceConfig"
                 }
               ]
+            },
+            {
+              "allOf": [
+                {
+                  "$ref": "#/$defs/EvmNodeDatasourceConfig"
+                }
+              ]
             }
           ]
         },
@@ -2092,6 +2099,13 @@
                   "$ref": "#/$defs/SubsquidDatasourceConfig"
                 }
               ]
+            },
+            {
+              "allOf": [
+                {
+                  "$ref": "#/$defs/EvmNodeDatasourceConfig"
+                }
+              ]
             }
           ]
         },
@@ -2228,6 +2242,13 @@
                   "$ref": "#/$defs/SubsquidDatasourceConfig"
                 }
               ]
+            },
+            {
+              "allOf": [
+                {
+                  "$ref": "#/$defs/EvmNodeDatasourceConfig"
+                }
+              ]
             }
           ]
         },

src/dipdup/config/__init__.py

@@ -934,12 +934,16 @@ def _resolve_index_links(self, index_config: ResolvedIndexConfigU) -> None:
         """
         handler_config: HandlerConfig
 
-        # NOTE: Each index must have a corresponding (currently) TzKT datasource
+        # NOTE: Each index must have a corresponding index datasource
         if isinstance(index_config.datasource, str):
-            if 'tzkt' in index_config.kind:
-                index_config.datasource = self.get_tzkt_datasource(index_config.datasource)
-            elif 'subsquid' in index_config.kind:
-                index_config.datasource = self.get_subsquid_datasource(index_config.datasource)
+            name = index_config.datasource
+            if index_config.kind.startswith('tezos.tzkt'):
+                index_config.datasource = self.get_tzkt_datasource(name)
+            elif index_config.kind.startswith('evm.subsquid'):
+                try:
+                    index_config.datasource = self.get_subsquid_datasource(name)
+                except ConfigurationError:
+                    index_config.datasource = self.get_evm_node_datasource(name)
             else:
                 raise FrameworkException(f'Unknown datasource type for index `{index_config.name}`')
 
@@ -1150,6 +1154,7 @@ def _patch_annotations(replace_table: dict[str, str]) -> None:
 _original_to_aliased = {
     'TzktDatasourceConfig': 'str | TzktDatasourceConfig',
     'SubsquidDatasourceConfig': 'str | SubsquidDatasourceConfig',
+    'SubsquidDatasourceConfig | EvmNodeDatasourceConfig': 'str | SubsquidDatasourceConfig | EvmNodeDatasourceConfig',
     'ContractConfig': 'str | ContractConfig',
     'ContractConfig | None': 'str | ContractConfig | None',
     'TezosContractConfig': 'str | TezosContractConfig',

src/dipdup/config/evm_subsquid.py

@@ -60,4 +60,4 @@ class SubsquidIndexConfig(IndexConfig, ABC):
     :param datasource: Subsquid datasource config
     """
 
-    datasource: SubsquidDatasourceConfig
+    datasource: SubsquidDatasourceConfig | EvmNodeDatasourceConfig

src/dipdup/config/evm_subsquid_events.py

@@ -11,6 +11,7 @@
 from dipdup.config import AbiDatasourceConfig
 from dipdup.config import HandlerConfig
 from dipdup.config.evm import EvmContractConfig
+from dipdup.config.evm_node import EvmNodeDatasourceConfig
 from dipdup.config.evm_subsquid import SubsquidDatasourceConfig
 from dipdup.config.evm_subsquid import SubsquidIndexConfig
 from dipdup.models.evm_node import EvmNodeHeadSubscription
@@ -60,13 +61,12 @@ class SubsquidEventsIndexConfig(SubsquidIndexConfig):
     :param datasource: Subsquid datasource
     :param handlers: Event handlers
     :param abi: One or more `evm.abi` datasource(s) for the same network
-    :param node_only: Don't use Subsquid Network API (dev only)
     :param first_level: Level to start indexing from
     :param last_level: Level to stop indexing and disable this index
     """
 
     kind: Literal['evm.subsquid.events']
-    datasource: SubsquidDatasourceConfig
+    datasource: SubsquidDatasourceConfig | EvmNodeDatasourceConfig
     handlers: tuple[SubsquidEventsHandlerConfig, ...] = field(default_factory=tuple)
     abi: AbiDatasourceConfig | tuple[AbiDatasourceConfig, ...] | None = None
     node_only: bool = False

src/dipdup/config/evm_subsquid_traces.py

@@ -9,6 +9,7 @@
 
 from dipdup.config import AbiDatasourceConfig
 from dipdup.config import HandlerConfig
+from dipdup.config.evm_node import EvmNodeDatasourceConfig
 from dipdup.config.evm_subsquid import SubsquidDatasourceConfig
 from dipdup.config.evm_subsquid import SubsquidIndexConfig
 
@@ -21,7 +22,7 @@ class SubsquidTracesHandlerConfig(HandlerConfig): ...
 class SubsquidTracesIndexConfig(SubsquidIndexConfig):
     kind: Literal['evm.subsquid.traces']
 
-    datasource: SubsquidDatasourceConfig
+    datasource: SubsquidDatasourceConfig | EvmNodeDatasourceConfig
     handlers: tuple[SubsquidTracesHandlerConfig, ...] = field(default_factory=tuple)
     abi: AbiDatasourceConfig | tuple[AbiDatasourceConfig, ...] | None = None
     node_only: bool = False

src/dipdup/config/evm_subsquid_transactions.py

@@ -12,6 +12,7 @@
 from dipdup.config import CodegenMixin
 from dipdup.config import HandlerConfig
 from dipdup.config.evm import EvmContractConfig
+from dipdup.config.evm_node import EvmNodeDatasourceConfig
 from dipdup.config.evm_subsquid import SubsquidDatasourceConfig
 from dipdup.config.evm_subsquid import SubsquidIndexConfig
 from dipdup.models.evm_node import EvmNodeHeadSubscription
@@ -76,14 +77,13 @@ class SubsquidTransactionsIndexConfig(SubsquidIndexConfig):
     :param datasource: Subsquid datasource config
     :param handlers: Transaction handlers
     :param abi: One or many ABI datasource(s)
-    :param node_only: Don't use Subsquid Network, only node RPC
     :param first_level: Level to start indexing from
     :param last_level: Level to stop indexing at
     """
 
     kind: Literal['evm.subsquid.transactions']
 
-    datasource: SubsquidDatasourceConfig
+    datasource: SubsquidDatasourceConfig | EvmNodeDatasourceConfig
     handlers: tuple[SubsquidTransactionsHandlerConfig, ...] = field(default_factory=tuple)
     abi: AbiDatasourceConfig | tuple[AbiDatasourceConfig, ...] | None = None
     node_only: bool = False

src/dipdup/context.py

@@ -24,6 +24,8 @@
 from dipdup.config import HookConfig
 from dipdup.config import ResolvedIndexConfigU
 from dipdup.config.evm import EvmContractConfig
+from dipdup.config.evm_node import EvmNodeDatasourceConfig
+from dipdup.config.evm_subsquid import SubsquidDatasourceConfig
 from dipdup.config.evm_subsquid_events import SubsquidEventsIndexConfig
 from dipdup.config.evm_subsquid_traces import SubsquidTracesIndexConfig
 from dipdup.config.evm_subsquid_transactions import SubsquidTransactionsIndexConfig
@@ -76,7 +78,6 @@
     from collections.abc import Iterator
     from types import ModuleType
 
-    from dipdup.config.evm_node import EvmNodeDatasourceConfig
     from dipdup.package import DipDupPackage
     from dipdup.transactions import TransactionManager
 
@@ -320,8 +321,7 @@ async def _spawn_index(self, name: str, state: Index | None = None) -> Any:
         )
 
         datasource_name = index_config.datasource.name
-        datasource: TzktDatasource | SubsquidDatasource
-        node_configs: tuple[EvmNodeDatasourceConfig, ...] = ()
+        datasource: TzktDatasource | SubsquidDatasource | EvmNodeDatasource
 
         if isinstance(index_config, TzktOperationsIndexConfig | TzktOperationsUnfilteredIndexConfig):
             datasource = self.get_tzkt_datasource(datasource_name)
@@ -342,26 +342,33 @@ async def _spawn_index(self, name: str, state: Index | None = None) -> Any:
             datasource = self.get_tzkt_datasource(datasource_name)
             index = TzktEventsIndex(self, index_config, datasource)
         elif isinstance(index_config, SubsquidEventsIndexConfig):
-            datasource = self.get_subsquid_datasource(datasource_name)
-            node_field = index_config.datasource.node
-            if node_field:
-                node_configs = node_configs + node_field if isinstance(node_field, tuple) else (node_field,)
+            datasource_config = index_config.datasource
+            if isinstance(datasource_config, SubsquidDatasourceConfig):
+                datasource = self.get_subsquid_datasource(datasource_name)
+            elif isinstance(datasource_config, EvmNodeDatasourceConfig):
+                datasource = self.get_evm_node_datasource(datasource_name)
+            else:
+                raise NotImplementedError
             index = SubsquidEventsIndex(self, index_config, datasource)
+            for node_datasource in index.node_datasources:
+                node_datasource.add_index(index_config)
         elif isinstance(index_config, SubsquidTracesIndexConfig):
             raise NotImplementedError
         elif isinstance(index_config, SubsquidTransactionsIndexConfig):
-            datasource = self.get_subsquid_datasource(datasource_name)
-            node_field = index_config.datasource.node
-            if node_field:
-                node_configs = node_configs + node_field if isinstance(node_field, tuple) else (node_field,)
+            datasource_config = index_config.datasource
+            if isinstance(datasource_config, SubsquidDatasourceConfig):
+                datasource = self.get_subsquid_datasource(datasource_name)
+            elif isinstance(datasource_config, EvmNodeDatasourceConfig):
+                datasource = self.get_evm_node_datasource(datasource_name)
+            else:
+                raise NotImplementedError
             index = SubsquidTransactionsIndex(self, index_config, datasource)
+            for node_datasource in index.node_datasources:
+                node_datasource.add_index(index_config)
         else:
             raise NotImplementedError
 
         datasource.add_index(index_config)
-        for node_config in node_configs:
-            node_datasource = self.get_evm_node_datasource(node_config.name)
-            node_datasource.add_index(index_config)
 
         handlers = (
             (index_config.handler_config,)

src/dipdup/indexes/evm_subsquid.py

@@ -11,6 +11,7 @@
 from dipdup.config import SubsquidIndexConfigU
 from dipdup.config.evm import EvmContractConfig
 from dipdup.config.evm_node import EvmNodeDatasourceConfig
+from dipdup.config.evm_subsquid import SubsquidDatasourceConfig
 from dipdup.context import DipDupContext
 from dipdup.datasources import IndexDatasource
 from dipdup.datasources.evm_node import NODE_LAST_MILE
@@ -27,7 +28,7 @@
 SUBSQUID_READAHEAD_LIMIT = 10000
 
 IndexConfigT = TypeVar('IndexConfigT', bound=SubsquidIndexConfigU)
-DatasourceT = TypeVar('DatasourceT', bound=SubsquidDatasource)
+DatasourceT = TypeVar('DatasourceT', bound=SubsquidDatasource | EvmNodeDatasource)
 
 
 _sighashes: dict[str, str] = {}
@@ -62,14 +63,19 @@ def __init__(
     ) -> None:
         super().__init__(ctx, config, datasource)
 
-        node_field = self._config.datasource.node
-        if node_field is None:
-            node_field = ()
-        elif isinstance(node_field, EvmNodeDatasourceConfig):
-            node_field = (node_field,)
-        self._node_datasources = tuple(
-            self._ctx.get_evm_node_datasource(node_config.name) for node_config in node_field
-        )
+        if isinstance(datasource, SubsquidDatasource) and isinstance(config.datasource, SubsquidDatasourceConfig):
+            node_field = config.datasource.node
+            if node_field is None:
+                node_field = ()
+            elif isinstance(node_field, EvmNodeDatasourceConfig):
+                node_field = (node_field,)
+            self._node_datasources = tuple(
+                self._ctx.get_evm_node_datasource(node_config.name) for node_config in node_field
+            )
+        elif isinstance(datasource, EvmNodeDatasource) and isinstance(config.datasource, EvmNodeDatasourceConfig):
+            self._node_datasources = (datasource,)
+        else:
+            raise FrameworkException('Invalid datasource type')
 
     @abstractmethod
     async def _synchronize_subsquid(self, sync_level: int) -> None: ...
@@ -133,8 +139,12 @@ async def _synchronize(self, sync_level: int) -> None:
         if levels_left <= 0:
             return
 
-        subsquid_sync_level = await self.datasource.get_head_level()
-        Metrics.set_sqd_processor_chain_height(subsquid_sync_level)
+        if isinstance(self.datasource, SubsquidDatasource):
+            subsquid_sync_level = await self.datasource.get_head_level()
+            Metrics.set_sqd_processor_chain_height(subsquid_sync_level)
+        else:
+            subsquid_sync_level = 0
+
         node_sync_level = await self._get_node_sync_level(subsquid_sync_level, index_level)
 
         # NOTE: Fetch last blocks from node if there are not enough realtime messages in queue