Fix data loss in skip_history, fix default sqlite path (#266)

* Fix data loss in `skip_history` and default sqlite path

* Pagination fixes

* Another attempt

* similar_contracts test

* faster get_originated

* originated contracts test

* More tzkt tests

* even more tzkt tests

* more tests

* Check for offset.cr

* Cleanup

Author: droserasprout

CHANGELOG.md

@@ -2,6 +2,14 @@
 
 Please use [this](https://docs.gitlab.com/ee/development/changelog.html) document as guidelines to keep a changelog.
 
+## [unreleased]
+
+### Fixed
+
+* config: Fixed default SQLite path (`:memory:`).
+* tzkt: Fixed pagination in several getter methods.
+* tzkt: Fixed data loss when `skip_history` option is enabled.
+
 ## 5.0.0-rc2 - 2022-03-13
 
 ### Fixed

src/dipdup/config.py

@@ -62,7 +62,7 @@
 DEFAULT_POSTGRES_SCHEMA = 'public'
 DEFAULT_POSTGRES_USER = DEFAULT_POSTGRES_DATABASE = 'postgres'
 DEFAULT_POSTGRES_PORT = 5432
-DEFAULT_SQLITE_PATH = ':memory'
+DEFAULT_SQLITE_PATH = ':memory:'
 
 _logger = logging.getLogger('dipdup.config')
 

src/dipdup/datasources/tzkt/datasource.py

@@ -305,19 +305,19 @@ async def get_similar_contracts(
             'get',
             url=f'v1/contracts/{address}/{entrypoint}',
             params={
-                'select': 'id,address',
-                'offset.cr': offset,
+                'select': 'address',
+                'offset': offset,
                 'limit': limit,
             },
         )
-        return tuple(c['address'] for c in response)
+        return tuple(response)
 
     async def iter_similar_contracts(
         self,
         address: str,
         strict: bool = False,
     ) -> AsyncIterator[Tuple[str, ...]]:
-        async for batch in self._iter_batches(self.get_similar_contracts, address, strict):
+        async for batch in self._iter_batches(self.get_similar_contracts, address, strict, cursor=False):
             yield batch
 
     async def get_originated_contracts(
@@ -333,15 +333,15 @@ async def get_originated_contracts(
             'get',
             url=f'v1/accounts/{address}/contracts',
             params={
-                'select': 'id,address',
-                'offset.cr': offset,
+                'select': 'address',
+                'offset': offset,
                 'limit': limit,
             },
         )
-        return tuple(c['address'] for c in response)
+        return tuple(response)
 
     async def iter_originated_contracts(self, address: str) -> AsyncIterator[Tuple[str, ...]]:
-        async for batch in self._iter_batches(self.get_originated_contracts, address):
+        async for batch in self._iter_batches(self.get_originated_contracts, address, cursor=False):
             yield batch
 
     async def get_contract_summary(self, address: str) -> Dict[str, Any]:
@@ -386,7 +386,7 @@ async def get_big_map(
             params={
                 **kwargs,
                 'level': level,
-                'offset.cr': offset,
+                'offset': offset,
                 'limit': limit,
             },
         )
@@ -403,6 +403,7 @@ async def iter_big_map(
             big_map_id,
             level,
             active,
+            cursor=False,
         ):
             yield batch
 
@@ -413,6 +414,7 @@ async def get_contract_big_maps(
         limit: Optional[int] = None,
     ) -> Tuple[Dict[str, Any], ...]:
         offset, limit = offset or 0, limit or self.request_limit
+        # TODO: Can we cache it?
         big_maps = await self.request(
             'get',
             url=f'v1/contracts/{address}/bigmaps',
@@ -705,8 +707,9 @@ async def _on_subscribe(message: CompletionMessage) -> None:
         await event.wait()
 
     async def _iter_batches(self, fn, *args, cursor: bool = True, **kwargs) -> AsyncIterator:
-        if 'offset' in kwargs or 'limit' in kwargs:
+        if set(kwargs).intersection(('offset', 'offset.cr', 'limit')):
             raise ValueError('`offset` and `limit` arguments are not allowed')
+
         size, offset = self.request_limit, 0
         while size == self.request_limit:
             result = await fn(*args, offset=offset, **kwargs)
@@ -717,7 +720,11 @@ async def _iter_batches(self, fn, *args, cursor: bool = True, **kwargs) -> Async
 
             size = len(result)
             if cursor:
-                offset = result[-1]['id']
+                # NOTE: Guess if response is already deserialized or not
+                try:
+                    offset = result[-1]['id']
+                except TypeError:
+                    offset = result[-1].id
             else:
                 offset += self.request_limit
 

src/dipdup/index.py

@@ -5,7 +5,6 @@
 from collections import namedtuple
 from contextlib import ExitStack
 from datetime import datetime
-from typing import Any
 from typing import DefaultDict
 from typing import Deque
 from typing import Dict
@@ -635,8 +634,8 @@ async def _synchronize(self, last_level: int, cache: bool = False) -> None:
     async def _synchronize_full(self, first_level: int, last_level: int, cache: bool = False) -> None:
         self._logger.info('Fetching big map diffs from level %s to %s', first_level, last_level)
 
-        big_map_addresses = await self._get_big_map_addresses()
-        big_map_paths = await self._get_big_map_paths()
+        big_map_addresses = self._get_big_map_addresses()
+        big_map_paths = self._get_big_map_paths()
 
         fetcher = BigMapFetcher(
             datasource=self._datasource,
@@ -659,42 +658,38 @@ async def _synchronize_level(self, last_level: int, cache: bool = False) -> None
         if not self._ctx.config.advanced.early_realtime:
             raise ConfigurationError('`skip_history` requires `early_realtime` feature flag to be enabled')
 
-        big_map_addresses = await self._get_big_map_addresses()
-        big_map_paths = await self._get_big_map_paths()
-        big_map_ids: Set[Tuple[int, str]] = set()
+        big_map_pairs = self._get_big_map_pairs()
+        big_map_ids: Set[Tuple[int, str, str]] = set()
 
-        for address in big_map_addresses:
+        for address, path in big_map_pairs:
             async for contract_big_maps in self._datasource.iter_contract_big_maps(address):
                 for contract_big_map in contract_big_maps:
-                    if contract_big_map['path'] in big_map_paths:
-                        big_map_ids.add((int(contract_big_map['ptr']), contract_big_map['path']))
+                    if contract_big_map['path'] == path:
+                        big_map_ids.add((int(contract_big_map['ptr']), address, path))
 
         # NOTE: Do not use `_process_level_big_maps` here; we want to maintain transaction manually.
-        async def _process_big_map_batch(big_maps: Tuple[Dict[str, Any], ...], path: str) -> None:
-            big_map_data = tuple(
-                BigMapData(
-                    id=big_map['id'],
-                    level=last_level,
-                    operation_id=last_level,
-                    timestamp=datetime.now(),
-                    bigmap=big_map_id,
-                    contract_address=address,
-                    path=path,
-                    action=BigMapAction.ADD_KEY,
-                    active=big_map['active'],
-                    key=big_map['key'],
-                    value=big_map['value'],
-                )
-                for big_map in big_maps
-            )
-            matched_handlers = await self._match_big_maps(big_map_data)
-            for handler_config, big_map_diff in matched_handlers:
-                await self._call_matched_handler(handler_config, big_map_diff)
-
         async with in_global_transaction():
-            for big_map_id, path in big_map_ids:
-                async for big_maps in self._datasource.iter_big_map(big_map_id, last_level):
-                    await _process_big_map_batch(big_maps, path)
+            for big_map_id, address, path in big_map_ids:
+                async for big_map_keys in self._datasource.iter_big_map(big_map_id, last_level):
+                    big_map_data = tuple(
+                        BigMapData(
+                            id=big_map_key['id'],
+                            level=last_level,
+                            operation_id=last_level,
+                            timestamp=datetime.now(),
+                            bigmap=big_map_id,
+                            contract_address=address,
+                            path=path,
+                            action=BigMapAction.ADD_KEY,
+                            active=big_map_key['active'],
+                            key=big_map_key['key'],
+                            value=big_map_key['value'],
+                        )
+                        for big_map_key in big_map_keys
+                    )
+                    matched_handlers = await self._match_big_maps(big_map_data)
+                    for handler_config, big_map_diff in matched_handlers:
+                        await self._call_matched_handler(handler_config, big_map_diff)
 
             await self.state.update_status(level=last_level)
 
@@ -767,8 +762,8 @@ async def _match_big_maps(self, big_maps: Iterable[BigMapData]) -> Deque[Matched
         """Try to match big map diffs in cache with all patterns from indexes."""
         matched_handlers: Deque[MatchedBigMapsT] = deque()
 
-        for big_map in big_maps:
-            for handler_config in self._config.handlers:
+        for handler_config in self._config.handlers:
+            for big_map in big_maps:
                 big_map_matched = await self._match_big_map(handler_config, big_map)
                 if big_map_matched:
                     arg = await self._prepare_handler_args(handler_config, big_map)
@@ -789,20 +784,32 @@ async def _call_matched_handler(self, handler_config: BigMapHandlerConfig, big_m
             big_map_diff,
         )
 
-    async def _get_big_map_addresses(self) -> Set[str]:
+    def _get_big_map_addresses(self) -> Set[str]:
         """Get addresses to fetch big map diffs from during initial synchronization"""
         addresses = set()
         for handler_config in self._config.handlers:
             addresses.add(cast(ContractConfig, handler_config.contract).address)
         return addresses
 
-    async def _get_big_map_paths(self) -> Set[str]:
+    def _get_big_map_paths(self) -> Set[str]:
         """Get addresses to fetch big map diffs from during initial synchronization"""
         paths = set()
         for handler_config in self._config.handlers:
             paths.add(handler_config.path)
         return paths
 
+    def _get_big_map_pairs(self) -> Set[Tuple[str, str]]:
+        """Get address-path pairs for fetch big map diffs during sync with `skip_history`"""
+        pairs = set()
+        for handler_config in self._config.handlers:
+            pairs.add(
+                (
+                    cast(ContractConfig, handler_config.contract).address,
+                    handler_config.path,
+                )
+            )
+        return pairs
+
 
 class HeadIndex(Index):
     _config: HeadIndexConfig