Track datasource level by channel, ignore head rollbacks (#148)

* Track datasource level by channel, ignore head rollbacks

* Docs

* Typo

* Changelog, fix level check

* Rollback tests WIP

* Single queue WIP

* Refactor OperationIndex

* Fix tests

* Refactoring

* Refactor `_extract_message_data` method

* Allow "zero level" rollbacks, refactor `_on_rollback` callback

* Zero level rollbacks special case, more refactoring

* Refactoring, docs, fix integrity checks

* Replace some lists with deques

* test_rollback WIP

* Another test

* Another test

* Fix realtime atomicity

* More tests

* Update changelog

* Disable logger

* Warn on rollbacks

* Optimizations

* Moar

* Changelog

* Ability to pass additional context on reindexing

* Fix tests, enqueue single level rollback

* InitializationRequiredError on create_package failure

* Final touches, changelog

Author: droserasprout

CHANGELOG.md

@@ -2,15 +2,25 @@
 
 ## [unreleased]
 
+### Improved
+
+* A significant increase in indexing speed.
+
 ### Fixed
 
-* Removed unnecessary file IO calls, improved logging
+* Fixed unexpected reindexing caused by the bug in processing zero- and single-level rollbacks.
+* Removed unnecessary file IO calls that could cause `PermissionError` exception in Docker environments.
+* Fixed possible violation of block-level atomicity during real-time indexing.
+
+### Changes
+
+* Public methods of `TzktDatasource` now return immutable sequences.
 
 ## 3.0.3 - 2021-10-01
 
 ### Fixed
 
-* Fixed processing of single level rollbacks emitted before rolled back head
+* Fixed processing of single-level rollbacks emitted before rolled back head.
 
 ## 3.0.2 - 2021-09-30
 

src/dipdup/cli.py

@@ -20,7 +20,7 @@
 from dipdup.codegen import DEFAULT_DOCKER_ENV_FILE, DEFAULT_DOCKER_IMAGE, DEFAULT_DOCKER_TAG, DipDupCodeGenerator
 from dipdup.config import DipDupConfig, LoggingConfig, PostgresDatabaseConfig
 from dipdup.dipdup import DipDup
-from dipdup.exceptions import ConfigurationError, DeprecatedHandlerError, DipDupError, MigrationRequiredError
+from dipdup.exceptions import ConfigurationError, DeprecatedHandlerError, DipDupError, InitializationRequiredError, MigrationRequiredError
 from dipdup.hasura import HasuraGateway
 from dipdup.migrations import DipDupMigrationManager, deprecated_handlers
 from dipdup.utils.database import set_decimal_context, tortoise_wrapper
@@ -113,7 +113,10 @@ async def cli(ctx, config: List[str], env_file: List[str], logging_config: str):
     _config = DipDupConfig.load(config)
     init_sentry(_config)
 
-    await DipDupCodeGenerator(_config, {}).create_package()
+    try:
+        await DipDupCodeGenerator(_config, {}).create_package()
+    except Exception as e:
+        raise InitializationRequiredError from e
 
     if _config.spec_version not in spec_version_mapping:
         raise ConfigurationError(f'Unknown `spec_version`, correct ones: {", ".join(spec_version_mapping)}')

src/dipdup/context.py

@@ -90,22 +90,26 @@ async def restart(self) -> None:
             sys.argv.remove('--reindex')
         os.execl(sys.executable, sys.executable, *sys.argv)
 
-    async def reindex(self, reason: Optional[Union[str, ReindexingReason]] = None) -> None:
+    async def reindex(self, reason: Optional[Union[str, ReindexingReason]] = None, **context) -> None:
         """Drop all tables or whole database and restart with the same CLI arguments"""
-        reason_str = reason.value if isinstance(reason, ReindexingReason) else 'unknown'
-        self.logger.warning('Reindexing initialized, reason: %s', reason_str)
-
-        if not reason or isinstance(reason, str):
+        if not reason:
+            reason = ReindexingReason.MANUAL
+        elif isinstance(reason, str):
+            context['message'] = reason
             reason = ReindexingReason.MANUAL
 
+        reason_str = reason.value + f' ({context["message"]})' if "message" in context else ''
+        self.logger.warning('Reindexing initialized, reason: %s', reason_str)
+        self.logger.info('Additional context: %s', context)
+
         if forbid_reindexing:
             schema = await Schema.filter().get()
             if schema.reindex:
-                raise ReindexingRequiredError(schema.reindex)
+                raise ReindexingRequiredError(schema.reindex, context)
 
             schema.reindex = reason
             await schema.save()
-            raise ReindexingRequiredError(schema.reindex)
+            raise ReindexingRequiredError(schema.reindex, context)
 
         database_config = self.config.database
         if isinstance(database_config, PostgresDatabaseConfig):

src/dipdup/datasources/datasource.py

@@ -1,6 +1,6 @@
 import logging
 from abc import abstractmethod
-from typing import Awaitable, Callable, List, Set
+from typing import Awaitable, Callable, Set, Tuple
 
 from dipdup.config import HTTPConfig
 from dipdup.http import HTTPGateway
@@ -11,8 +11,8 @@
 
 
 HeadCallbackT = Callable[['IndexDatasource', HeadBlockData], Awaitable[None]]
-OperationsCallbackT = Callable[['IndexDatasource', List[OperationData]], Awaitable[None]]
-BigMapsCallbackT = Callable[['IndexDatasource', List[BigMapData]], Awaitable[None]]
+OperationsCallbackT = Callable[['IndexDatasource', Tuple[OperationData, ...]], Awaitable[None]]
+BigMapsCallbackT = Callable[['IndexDatasource', Tuple[BigMapData, ...]], Awaitable[None]]
 RollbackCallbackT = Callable[['IndexDatasource', int, int], Awaitable[None]]
 
 
@@ -57,11 +57,11 @@ async def emit_head(self, head: HeadBlockData) -> None:
         for fn in self._on_head:
             await fn(self, head)
 
-    async def emit_operations(self, operations: List[OperationData]) -> None:
+    async def emit_operations(self, operations: Tuple[OperationData, ...]) -> None:
         for fn in self._on_operations:
             await fn(self, operations)
 
-    async def emit_big_maps(self, big_maps: List[BigMapData]) -> None:
+    async def emit_big_maps(self, big_maps: Tuple[BigMapData, ...]) -> None:
         for fn in self._on_big_maps:
             await fn(self, big_maps)