Call rollback hook instead of triggering reindex when single-level rollback has failed (#369)

* Call rollback hook instead of triggering reindex when single-level rollback has failed

* Typo in changelog

* Allow mixing oneshot and regular indexes in a single config (#367)

* Decrease the size of generic and `-pytezos` Docker images by 11% and 16%, respectively (#370)

* Fix crash with `RuntimeError` after continuous realtime connection loss (#366)

* Fix crash with `RuntimeError` after continuous realtime connection loss

* Less agressive backoff

* Lint

Author: droserasprout

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Fixed
 
 * index: Allow mixing oneshot and regular indexes in a single config.
+* index: Call rollback hook instead of triggering reindex when single-level rollback has failed.
 * index: Fixed crash with `RuntimeError` after continuous realtime connection loss.
 * index: Fixed `OperationIndexConfig.types` field being partially ignored.
 * tzkt: Fixed `origination` subscription missing when `merge_subscriptions` flag is set.
@@ -52,7 +53,7 @@
 * ci: Push `X` and `X.Y` tags to the Docker Hub on release.
 * cli: Added `config env` command to export env-file with default values.
 * cli: Show warning when running an outdated version of DipDup.
-* hooks: Added new hook `on_index_rollback` to perform per-index rollbacks.
+* hooks: Added a new hook `on_index_rollback` to perform per-index rollbacks.
 
 ### Fixed
 
@@ -70,7 +71,7 @@
 
 * exceptions: Fixed incorrect formatting and broken links in help messages.
 * index: Fixed crash when the only index in config is `head`.
-* index: Fixed fetching originations during initial sync.
+* index: Fixed fetching originations during the initial sync.
 
 ## 5.0.3 - 2022-05-04
 

poetry.lock

@@ -52,9 +52,12 @@
 from dipdup.enums import SkipHistory
 from dipdup.exceptions import ConfigInitializationException
 from dipdup.exceptions import ConfigurationError
+from dipdup.exceptions import ConflictingHooksError
 from dipdup.exceptions import IndexAlreadyExistsError
+from dipdup.exceptions import InitializationRequiredError
 from dipdup.utils import exclude_none
 from dipdup.utils import import_from
+from dipdup.utils import is_importable
 from dipdup.utils import pascal_to_snake
 from dipdup.utils import snake_to_pascal
 
@@ -1225,6 +1228,23 @@ def package_path(self) -> str:
         except ImportError:
             return os.path.join(os.getcwd(), self.package)
 
+    # TODO: Remove in 6.0
+    @cached_property
+    def per_index_rollback(self) -> bool:
+        """Check if package has `on_index_rollback` hook"""
+        new_hook = is_importable(f'{self.package}.hooks.on_index_rollback', 'on_index_rollback')
+        old_hook = is_importable(f'{self.package}.hooks.on_rollback', 'on_rollback')
+        if new_hook and new_hook:
+            raise ConflictingHooksError('on_rollback', 'on_index_rollback')
+        elif not new_hook and not old_hook:
+            raise InitializationRequiredError('none of `on_rollback` or `on_index_rollback` hooks found')
+        elif new_hook:
+            return True
+        elif old_hook:
+            return False
+        else:
+            raise RuntimeError
+
     @property
     def oneshot(self) -> bool:
         """Whether all indexes have `last_level` field set"""

src/dipdup/config.py

@@ -68,9 +68,8 @@
 
 
 class IndexDispatcher:
-    def __init__(self, ctx: DipDupContext, index_rollback: bool = True) -> None:
+    def __init__(self, ctx: DipDupContext) -> None:
         self._ctx = ctx
-        self._index_rollback = index_rollback
 
         self._logger = logging.getLogger('dipdup')
         self._indexes: Dict[str, Index] = {}
@@ -332,7 +331,7 @@ async def _on_rollback(self, datasource: IndexDatasource, type_: MessageType, fr
             self._logger.info('`%s` rollback complete', channel)
             return
 
-        if self._index_rollback:
+        if self._ctx.config.per_index_rollback:
             hook_name = 'on_index_rollback'
             for index_name in unprocessed_indexes:
                 self._logger.warning('`%s`: can\'t process, firing `%s` hook', index_name, hook_name)
@@ -562,17 +561,7 @@ async def _set_up_index_dispatcher(
         start_scheduler_event: Event,
         early_realtime: bool,
     ) -> None:
-        # NOTE: Decide how to handle rollbacks depending on hooks presence
-        # TODO: Remove in 6.0
-        old_hook, new_hook = 'on_rollback', 'on_index_rollback'
-        has_old_hook = is_importable(f'{self._config.package}.hooks.{old_hook}', old_hook)
-        has_new_hook = is_importable(f'{self._config.package}.hooks.{new_hook}', new_hook)
-        if has_old_hook and has_new_hook:
-            raise ConflictingHooksError(old_hook, new_hook)
-        elif not has_old_hook and not has_new_hook:
-            raise InitializationRequiredError('none of `on_rollback` or `on_index_rollback` hooks found')
-
-        index_dispatcher = IndexDispatcher(self._ctx, index_rollback=has_new_hook)
+        index_dispatcher = IndexDispatcher(self._ctx)
         tasks.add(
             create_task(
                 index_dispatcher.run(

src/dipdup/dipdup.py

@@ -390,11 +390,13 @@ async def _process_level_operations(self, operation_subgroups: Tuple[OperationSu
         if batch_level < index_level:
             raise RuntimeError(f'Batch level is lower than index level: {batch_level} < {index_level}')
 
+        # NOTE: Single level rollback was triggered
         if head_level := self._next_head_level:
             if head_level != index_level:
                 raise RuntimeError(f'New head level is not equal to index level: {head_level} != {index_level}')
 
             self._logger.info('Rolling back to the previous level, verifying processed operations')
+            rollback_hook_called = False
             old_head_hashes = set(self._head_hashes)
             old_head_matched_hashes = {k for k, v in self._head_hashes.items() if v}
             new_head_hashes = {s.hash for s in operation_subgroups}
@@ -404,19 +406,31 @@ async def _process_level_operations(self, operation_subgroups: Tuple[OperationSu
 
             self._logger.info('Comparing hashes: %s new, %s missing', len(unprocessed_hashes), len(missing_hashes))
             if missing_hashes:
-                self._logger.info('Some operations were backtracked, requesting reindexing')
-                await self._ctx.reindex(
-                    ReindexingReason.rollback,
-                    datasource=self._datasource.name,
-                    from_level=head_level,
-                    # NOTE: Index level is not decreased on a single-level rollback
-                    to_level=head_level - 1,
-                    missing_hashes=', '.join(missing_hashes),
-                )
+                rollback_hook_called = True
+                self._logger.info('Some operations were backtracked, calling rollback hook')
+                if self._ctx.config.per_index_rollback:
+                    hook_name = 'on_index_rollback'
+                    await self._ctx.fire_hook(
+                        hook_name,
+                        index=self,
+                        from_level=head_level,
+                        to_level=head_level - 1,
+                    )
+                else:
+                    hook_name = 'on_rollback'
+                    await self._ctx.fire_hook(
+                        hook_name,
+                        datasource=self.datasource,
+                        from_level=head_level + 1,
+                        to_level=head_level,
+                    )
 
             self._next_head_level = None
             self._head_hashes.clear()
 
+            if rollback_hook_called:
+                return
+
             operation_subgroups = tuple(filter(lambda subgroup: subgroup.hash in unprocessed_hashes, operation_subgroups))
 
         self._logger.debug('Processing %s operation subgroups of level %s', len(operation_subgroups), batch_level)

src/dipdup/index.py

@@ -25,9 +25,12 @@
 
 
 def _get_index_dispatcher() -> IndexDispatcher:
-    return IndexDispatcher(
-        ctx=Mock(spec=DipDupContext),
+    index_dispatcher = IndexDispatcher(
+        ctx=MagicMock(spec=DipDupContext),
     )
+    index_dispatcher._ctx.reindex = AsyncMock()  # type: ignore
+    index_dispatcher._ctx.config = AsyncMock()  # type: ignore
+    return index_dispatcher
 
 
 def _get_operation_index(level: int) -> OperationIndex:
@@ -62,6 +65,7 @@ def _get_operation_index(level: int) -> OperationIndex:
     index._state.save = AsyncMock()  # type: ignore
     index._call_matched_handler = AsyncMock()  # type: ignore
     index._ctx.reindex = AsyncMock()  # type: ignore
+    index._ctx.config = MagicMock()
 
     return index
 
@@ -352,12 +356,11 @@ async def test_single_level_new_head_less(self) -> None:
             assert operation_index._next_head_level is None
             assert operation_index.state.level == from_level
 
-            operation_index._ctx.reindex.assert_awaited_with(
-                ReindexingReason.rollback,
-                datasource=operation_index.datasource.name,
+            operation_index._ctx.fire_hook.assert_awaited_with(
+                'on_index_rollback',
+                index=operation_index,
                 from_level=from_level,
                 to_level=to_level,
-                missing_hashes='1',
             )
             assert operation_index._call_matched_handler.await_count == 2