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,13 @@
 
 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 data loss when `skip_history` option is enabled.
+
 ## 4.2.6 - 2022-02-25
 
 ### 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

@@ -316,7 +316,7 @@ async def get_similar_contracts(self, address: str, strict: bool = False) -> Tup
                 'get',
                 url=f'v1/contracts/{address}/{entrypoint}',
                 params={
-                    'select': 'address',
+                    'select': 'id,address',
                     'limit': self.request_limit,
                     'offset': offset,
                 },
@@ -339,7 +339,7 @@ async def get_originated_contracts(self, address: str) -> Tuple[str, ...]:
                 'get',
                 url=f'v1/accounts/{address}/contracts',
                 params={
-                    'select': 'address',
+                    'select': 'id,address',
                     'limit': self.request_limit,
                     'offset': offset,
                 },

src/dipdup/index.py

@@ -593,8 +593,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,
@@ -613,37 +613,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: Tuple[Tuple[int, str], ...] = ()
-        big_map_data: Tuple[BigMapData, ...] = ()
-
-        for address in big_map_addresses:
-            contract_big_maps = await self._datasource.get_contract_big_maps(address)
-            for contract_big_map in contract_big_maps:
-                if contract_big_map['path'] in big_map_paths:
-                    big_map_ids += ((int(contract_big_map['ptr']), contract_big_map['path']),)
-
-        for bigmap_id, path in big_map_ids:
-            big_maps = await self._datasource.get_big_map(bigmap_id, last_level)
-            big_map_data = big_map_data + tuple(
-                BigMapData(
-                    id=big_map['id'],
-                    level=last_level,
-                    operation_id=last_level,
-                    timestamp=datetime.now(),
-                    bigmap=bigmap_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
-            )
+        big_map_pairs = self._get_big_map_pairs()
+        big_map_ids: Set[Tuple[int, str, str]] = set()
 
-        await self._process_level_big_maps(big_map_data)
+        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'] == 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 with in_global_transaction():
+            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)
 
     async def _process_level_big_maps(self, big_maps: Tuple[BigMapData, ...]):
         if not big_maps:
@@ -714,8 +715,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)
@@ -736,20 +737,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

tests/test_dipdup/test_datasources/test_tzkt/test_datasource.py

@@ -0,0 +1,191 @@
+from contextlib import asynccontextmanager
+from typing import AsyncIterator
+from unittest import IsolatedAsyncioTestCase
+
+from dipdup.config import HTTPConfig
+from dipdup.datasources.tzkt.datasource import TzktDatasource
+
+# from unittest import skip
+
+
+@asynccontextmanager
+async def with_tzkt(batch_size: int):
+    config = HTTPConfig(batch_size=batch_size)
+    datasource = TzktDatasource('https://api.tzkt.io', config)
+    async with datasource:
+        yield datasource
+
+
+async def take_two(iterable: AsyncIterator):
+    result = ()
+    left = 2
+    async for batch in iterable:
+        result = result + batch
+        left -= 1
+        if not left:
+            return result
+    raise RuntimeError
+
+
+class TzktDatasourceTest(IsolatedAsyncioTestCase):
+    async def test_get_similar_contracts(self) -> None:
+        async with with_tzkt(2) as tzkt:
+            contracts = await tzkt.get_similar_contracts(
+                address='KT1WBLrLE2vG8SedBqiSJFm4VVAZZBytJYHc',
+                strict=False,
+            )
+            self.assertEqual(
+                ('KT1W3VGRUjvS869r4ror8kdaxqJAZUbPyjMT', 'KT1K4EwTpbvYN9agJdjpyJm4ZZdhpUNKB3F6'),
+                contracts,
+            )
+
+            contracts = await tzkt.get_similar_contracts(
+                address='KT1WBLrLE2vG8SedBqiSJFm4VVAZZBytJYHc',
+                strict=True,
+            )
+            self.assertEqual(
+                ('KT1W3VGRUjvS869r4ror8kdaxqJAZUbPyjMT', 'KT1K4EwTpbvYN9agJdjpyJm4ZZdhpUNKB3F6'),
+                contracts,
+            )
+
+    async def test_iter_similar_contracts(self):
+        async with with_tzkt(1) as tzkt:
+            contracts = await take_two(
+                tzkt.iter_similar_contracts(
+                    address='KT1WBLrLE2vG8SedBqiSJFm4VVAZZBytJYHc',
+                    strict=False,
+                )
+            )
+            self.assertEqual(
+                ('KT1W3VGRUjvS869r4ror8kdaxqJAZUbPyjMT', 'KT1K4EwTpbvYN9agJdjpyJm4ZZdhpUNKB3F6'),
+                contracts,
+            )
+
+            contracts = await take_two(
+                tzkt.iter_similar_contracts(
+                    address='KT1WBLrLE2vG8SedBqiSJFm4VVAZZBytJYHc',
+                    strict=True,
+                )
+            )
+            self.assertEqual(
+                ('KT1W3VGRUjvS869r4ror8kdaxqJAZUbPyjMT', 'KT1K4EwTpbvYN9agJdjpyJm4ZZdhpUNKB3F6'),
+                contracts,
+            )
+
+    async def test_get_originated_contracts(self) -> None:
+        async with with_tzkt(2) as tzkt:
+            contracts = await tzkt.get_originated_contracts(
+                address='KT1Lw8hCoaBrHeTeMXbqHPG4sS4K1xn7yKcD',
+            )
+            self.assertEqual(
+                'KT1X1LgNkQShpF9nRLYw3Dgdy4qp38MX617z',
+                contracts[0]['address'],
+            )
+            self.assertEqual(
+                'KT1BgezWwHBxA9NrczwK9x3zfgFnUkc7JJ4b',
+                contracts[1]['address'],
+            )
+
+    async def iter_originated_contracts(self):
+        async with with_tzkt(1) as tzkt:
+            contracts = await take_two(
+                tzkt.iter_originated_contracts(
+                    address='KT1Lw8hCoaBrHeTeMXbqHPG4sS4K1xn7yKcD',
+                )
+            )
+            self.assertEqual(
+                'KT1X1LgNkQShpF9nRLYw3Dgdy4qp38MX617z',
+                contracts[0]['address'],
+            )
+            self.assertEqual(
+                'KT1BgezWwHBxA9NrczwK9x3zfgFnUkc7JJ4b',
+                contracts[1]['address'],
+            )
+
+    async def test_get_contract_summary(self):
+        async with with_tzkt(1) as tzkt:
+            contract = await tzkt.get_contract_summary(
+                address='KT1Lw8hCoaBrHeTeMXbqHPG4sS4K1xn7yKcD',
+            )
+            self.assertEqual(
+                'KT1Lw8hCoaBrHeTeMXbqHPG4sS4K1xn7yKcD',
+                contract['address'],
+            )
+
+    async def test_get_contract_storage(self):
+        async with with_tzkt(1) as tzkt:
+            storage = await tzkt.get_contract_storage(
+                address='KT1Lw8hCoaBrHeTeMXbqHPG4sS4K1xn7yKcD',
+            )
+            self.assertEqual(
+                1451,
+                storage['token_lambdas'],
+            )
+
+    async def test_get_jsonschemas(self):
+        async with with_tzkt(1) as tzkt:
+            jsonschemas = await tzkt.get_jsonschemas(
+                address='KT1Lw8hCoaBrHeTeMXbqHPG4sS4K1xn7yKcD',
+            )
+            self.assertEqual(
+                'string',
+                jsonschemas['storageSchema']['properties']['baker_validator']['type'],
+            )
+
+    async def test_get_big_map(self):
+        async with with_tzkt(2) as tzkt:
+            big_map_keys = await tzkt.get_big_map(
+                big_map_id=55031,
+                level=550310,
+            )
+            self.assertEqual(
+                (12392933, 12393108),
+                (big_map_keys[0]['id'], big_map_keys[1]['id']),
+            )
+
+    async def test_iter_big_map(self):
+        async with with_tzkt(1) as tzkt:
+            big_map_keys = await take_two(
+                tzkt.iter_big_map(
+                    big_map_id=55031,
+                    level=550310,
+                )
+            )
+            self.assertEqual(
+                (12392933, 12393108),
+                (big_map_keys[0]['id'], big_map_keys[1]['id']),
+            )
+
+    async def test_get_contract_big_maps(self):
+        async with with_tzkt(2) as tzkt:
+            big_maps = await tzkt.get_contract_big_maps(
+                address='KT1Lw8hCoaBrHeTeMXbqHPG4sS4K1xn7yKcD',
+            )
+            self.assertEqual(
+                ('votes', 'voters'),
+                (big_maps[0]['path'], big_maps[1]['path']),
+            )
+
+    async def test_iter_contract_big_maps(self):
+        async with with_tzkt(1) as tzkt:
+            big_maps = await take_two(
+                tzkt.iter_contract_big_maps(
+                    address='KT1Lw8hCoaBrHeTeMXbqHPG4sS4K1xn7yKcD',
+                )
+            )
+            self.assertEqual(
+                ('votes', 'voters'),
+                (big_maps[0]['path'], big_maps[1]['path']),
+            )
+
+    async def test_get_migration_originations(self):
+        async with with_tzkt(2) as tzkt:
+            originations = await tzkt.get_migration_originations()
+            self.assertEqual(67955553, originations[0].id)
+            self.assertEqual(67955554, originations[1].id)
+
+    async def test_iter_migration_originations(self):
+        async with with_tzkt(1) as tzkt:
+            originations = await take_two(tzkt.iter_migration_originations())
+            self.assertEqual(67955553, originations[0].id)
+            self.assertEqual(67955554, originations[1].id)