Ignore indexes with different message types on rollback (#343)

Author: droserasprout

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Fixed
 
+* index: Ignore indexes with different message types on rollback.
 * metadata: Add `ithacanet` to available networks.
 
 ## 5.1.0 - 2022-05-12

src/dipdup/datasources/datasource.py

@@ -12,6 +12,7 @@
 from dipdup.datasources.subscription import HeadSubscription
 from dipdup.datasources.subscription import Subscription
 from dipdup.datasources.subscription import SubscriptionManager
+from dipdup.enums import MessageType
 from dipdup.http import HTTPGateway
 from dipdup.models import BigMapData
 from dipdup.models import HeadBlockData
@@ -24,7 +25,7 @@
 HeadCallbackT = Callable[['IndexDatasource', HeadBlockData], Awaitable[None]]
 OperationsCallbackT = Callable[['IndexDatasource', Tuple[OperationData, ...]], Awaitable[None]]
 BigMapsCallbackT = Callable[['IndexDatasource', Tuple[BigMapData, ...]], Awaitable[None]]
-RollbackCallbackT = Callable[['IndexDatasource', int, int], Awaitable[None]]
+RollbackCallbackT = Callable[['IndexDatasource', MessageType, int, int], Awaitable[None]]
 
 
 class Datasource(HTTPGateway):
@@ -114,9 +115,9 @@ async def emit_big_maps(self, big_maps: Tuple[BigMapData, ...]) -> None:
         for fn in self._on_big_maps:
             await fn(self, big_maps)
 
-    async def emit_rollback(self, from_level: int, to_level: int) -> None:
+    async def emit_rollback(self, type_: MessageType, from_level: int, to_level: int) -> None:
         for fn in self._on_rollback:
-            await fn(self, from_level, to_level)
+            await fn(self, type_, from_level, to_level)
 
     def set_network(self, network: str) -> None:
         if self._network:

src/dipdup/datasources/tzkt/datasource.py

@@ -967,7 +967,7 @@ async def _on_message(self, type_: MessageType, message: List[Dict[str, Any]]) -
             # NOTE: Try to process rollback automatically, emit if failed
             elif tzkt_type == TzktMessageType.REORG:
                 if not self._buffer.rollback(type_, channel_level, message_level):
-                    await self.emit_rollback(channel_level, message_level)
+                    await self.emit_rollback(type_, channel_level, message_level)
 
             else:
                 raise NotImplementedError(f'Unknown message type: {tzkt_type}')

src/dipdup/dipdup.py

@@ -37,6 +37,7 @@
 from dipdup.datasources.datasource import IndexDatasource
 from dipdup.datasources.factory import DatasourceFactory
 from dipdup.datasources.tzkt.datasource import TzktDatasource
+from dipdup.enums import MessageType
 from dipdup.enums import ReindexingReason
 from dipdup.exceptions import ConfigInitializationException
 from dipdup.exceptions import ConflictingHooksError
@@ -269,12 +270,13 @@ async def _on_big_maps(self, datasource: TzktDatasource, big_maps: Tuple[BigMapD
         for index in big_map_indexes:
             index.push_big_maps(big_maps)
 
-    async def _on_rollback(self, datasource: TzktDatasource, from_level: int, to_level: int) -> None:
+    async def _on_rollback(self, datasource: TzktDatasource, type_: MessageType, from_level: int, to_level: int) -> None:
         """Perform a single level rollback when possible, otherwise call `on_rollback` hook"""
         if from_level <= to_level:
             raise RuntimeError(f'Attempt to rollback forward: {from_level} -> {to_level}')
 
-        self._logger.warning('Datasource `%s` has rolled back: %s -> %s', datasource.name, from_level, to_level)
+        channel = f'{datasource.name}:{type_.value}'
+        self._logger.info('Channel `%s` has rolled back: %s -> %s', channel, from_level, to_level)
         if Metrics.enabled:
             Metrics.set_datasource_rollback(datasource.name)
 
@@ -286,12 +288,16 @@ async def _on_rollback(self, datasource: TzktDatasource, from_level: int, to_lev
         for index_name, index in self._indexes.items():
             index_level = index.state.level
 
-            if index.datasource != datasource:
-                self._logger.debug('%s: another datasource, skipping', index_name)
+            if index.message_type != type_:
+                self._logger.debug('%s: different channel, skipping', index_name)
+                ignored_indexes.add(index_name)
+
+            elif index.datasource != datasource:
+                self._logger.debug('%s: different datasource, skipping', index_name)
                 ignored_indexes.add(index_name)
 
             elif to_level >= index_level:
-                self._logger.debug('%s: not affected, skipping', index_name)
+                self._logger.debug('%s: level is too low, skipping', index_name)
                 ignored_indexes.add(index_name)
 
             elif from_level - to_level == 1:
@@ -321,7 +327,7 @@ async def _on_rollback(self, datasource: TzktDatasource, from_level: int, to_lev
             index.push_rollback(from_level)
 
         if not unprocessed_indexes:
-            self._logger.info('Rollback complete')
+            self._logger.info('`%s` rollback complete', channel)
             return
 
         if self._index_rollback:

src/dipdup/index.py

@@ -43,6 +43,7 @@
 from dipdup.datasources.tzkt.datasource import TokenTransferFetcher
 from dipdup.datasources.tzkt.datasource import TzktDatasource
 from dipdup.datasources.tzkt.models import deserialize_storage
+from dipdup.enums import MessageType
 from dipdup.exceptions import ConfigInitializationException
 from dipdup.exceptions import ConfigurationError
 from dipdup.exceptions import InvalidDataError
@@ -138,6 +139,7 @@ class Index:
     Provides common interface for managing index state and switching between sync and realtime modes.
     """
 
+    message_type: MessageType
     _queue: Deque
 
     def __init__(self, ctx: DipDupContext, config: ResolvedIndexConfigT, datasource: TzktDatasource) -> None:
@@ -272,6 +274,7 @@ def _extract_level(
 
 
 class OperationIndex(Index):
+    message_type = MessageType.operation
     _config: OperationIndexConfig
 
     def __init__(self, ctx: DipDupContext, config: OperationIndexConfig, datasource: TzktDatasource) -> None:
@@ -618,6 +621,7 @@ async def _get_contract_hashes(self, address: str) -> Tuple[int, int]:
 
 
 class BigMapIndex(Index):
+    message_type = MessageType.big_map
     _config: BigMapIndexConfig
 
     def __init__(self, ctx: DipDupContext, config: BigMapIndexConfig, datasource: TzktDatasource) -> None:
@@ -842,6 +846,7 @@ def _get_big_map_pairs(self) -> Set[Tuple[str, str]]:
 
 
 class HeadIndex(Index):
+    message_type: MessageType = MessageType.head
     _config: HeadIndexConfig
 
     def __init__(self, ctx: DipDupContext, config: HeadIndexConfig, datasource: TzktDatasource) -> None:
@@ -886,6 +891,7 @@ def push_head(self, head: HeadBlockData) -> None:
 
 
 class TokenTransferIndex(Index):
+    message_type = MessageType.token_transfer
     _config: TokenTransferIndexConfig
 
     def __init__(self, ctx: DipDupContext, config: TokenTransferIndexConfig, datasource: TzktDatasource) -> None:

tests/test_dipdup/test_rollback.py

@@ -15,6 +15,7 @@
 from dipdup.context import DipDupContext
 from dipdup.datasources.tzkt.datasource import TzktDatasource
 from dipdup.dipdup import IndexDispatcher
+from dipdup.enums import MessageType
 from dipdup.enums import ReindexingReason
 from dipdup.index import BigMapIndex
 from dipdup.index import HeadIndex
@@ -133,6 +134,7 @@ async def test_forward_rollback(self) -> None:
         with self.assertRaises(RuntimeError):
             await dispatcher._on_rollback(
                 datasource=Mock(spec=TzktDatasource),
+                type_=MessageType.operation,
                 from_level=from_level,
                 to_level=to_level,
             )
@@ -146,6 +148,7 @@ async def test_not_affected_by_level(self) -> None:
         }
         await dispatcher._on_rollback(
             datasource=operation_index.datasource,
+            type_=MessageType.operation,
             from_level=from_level,
             to_level=to_level,
         )
@@ -163,6 +166,24 @@ async def test_not_affected_by_datasource(self) -> None:
         }
         await dispatcher._on_rollback(
             datasource=other_datasource,
+            type_=MessageType.operation,
+            from_level=from_level,
+            to_level=to_level,
+        )
+        self.assertIsNone(operation_index._next_head_level)
+        dispatcher._ctx.fire_hook.assert_not_awaited()  # type: ignore
+        self.assertEqual(0, len(operation_index._queue))
+
+    async def test_not_affected_by_type(self) -> None:
+        index_level, from_level, to_level = 20, 20, 15
+        dispatcher = _get_index_dispatcher()
+        operation_index = _get_operation_index(level=index_level)
+        dispatcher._indexes = {
+            'operation': operation_index,
+        }
+        await dispatcher._on_rollback(
+            datasource=operation_index.datasource,
+            type_=MessageType.head,
             from_level=from_level,
             to_level=to_level,
         )
@@ -179,6 +200,7 @@ async def test_unprocessed_head(self) -> None:
         }
         await dispatcher._on_rollback(
             datasource=head_index.datasource,
+            type_=MessageType.head,
             from_level=from_level,
             to_level=to_level,
         )
@@ -199,6 +221,7 @@ async def test_unprocessed(self) -> None:
         }
         await dispatcher._on_rollback(
             datasource=operation_index.datasource,
+            type_=MessageType.operation,
             from_level=from_level,
             to_level=to_level,
         )
@@ -219,6 +242,7 @@ async def test_single_level_supported(self) -> None:
         }
         await dispatcher._on_rollback(
             datasource=operation_index.datasource,
+            type_=MessageType.operation,
             from_level=from_level,
             to_level=to_level,
         )
@@ -235,6 +259,7 @@ async def test_single_level_not_supported(self) -> None:
         }
         await dispatcher._on_rollback(
             datasource=big_map_index.datasource,
+            type_=MessageType.big_map,
             from_level=from_level,
             to_level=to_level,
         )
@@ -266,6 +291,7 @@ async def test_single_level_new_head_equal(self) -> None:
 
             await dispatcher._on_rollback(
                 datasource=operation_index.datasource,
+                type_=MessageType.operation,
                 from_level=from_level,
                 to_level=to_level,
             )
@@ -307,6 +333,7 @@ async def test_single_level_new_head_less(self) -> None:
 
             await dispatcher._on_rollback(
                 datasource=operation_index.datasource,
+                type_=MessageType.operation,
                 from_level=from_level,
                 to_level=to_level,
             )
@@ -354,6 +381,7 @@ async def test_single_level_new_head_less_not_matched(self) -> None:
 
             await dispatcher._on_rollback(
                 datasource=operation_index.datasource,
+                type_=MessageType.operation,
                 from_level=from_level,
                 to_level=to_level,
             )
@@ -394,6 +422,7 @@ async def test_single_level_new_head_more(self) -> None:
 
             await dispatcher._on_rollback(
                 datasource=operation_index.datasource,
+                type_=MessageType.operation,
                 from_level=from_level,
                 to_level=to_level,
             )