Fix TzKT pagination and Prometheus labels (#264)

* Fixed crash in methods that do not support cursor pagination

* Fixed invalid metric labels

* Fix pagination of gem_big_maps

* Edge case

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
+
+* tzkt: Fixed crash in methods that do not support cursor pagination.
+* prometheus: Fixed invalid metric labels. 
+
 ## 5.0.0-rc1 - 2022-03-02
 
 ### Added

src/dipdup/datasources/tzkt/datasource.py

@@ -228,19 +228,16 @@ async def fetch_big_maps_by_level(self) -> AsyncGenerator[Tuple[int, Tuple[BigMa
         Resulting data is splitted by level, deduped, sorted and ready to be processed by BigMapIndex.
         """
 
-        offset = 0
         big_maps: Tuple[BigMapData, ...] = ()
 
         # TODO: Share code between this and OperationFetcher
-        while True:
-            fetched_big_maps = await self._datasource.get_big_maps(
-                self._big_map_addresses,
-                self._big_map_paths,
-                self._first_level,
-                self._last_level,
-                cache=self._cache,
-                offset=offset,
-            )
+        big_map_iter = self._datasource.iter_big_maps(
+            self._big_map_addresses,
+            self._big_map_paths,
+            self._first_level,
+            self._last_level,
+        )
+        async for fetched_big_maps in big_map_iter:
             big_maps = big_maps + fetched_big_maps
 
             # NOTE: Yield big map slices by level except the last one
@@ -256,11 +253,6 @@ async def fetch_big_maps_by_level(self) -> AsyncGenerator[Tuple[int, Tuple[BigMa
                 else:
                     break
 
-            if len(fetched_big_maps) < self._datasource.request_limit:
-                break
-
-            offset += self._datasource.request_limit
-
         if big_maps:
             yield big_maps[0].level, big_maps
 
@@ -425,7 +417,7 @@ async def get_contract_big_maps(
             'get',
             url=f'v1/contracts/{address}/bigmaps',
             params={
-                'offset.cr': offset,
+                'offset': offset,
                 'limit': limit,
             },
         )
@@ -435,7 +427,7 @@ async def iter_contract_big_maps(
         self,
         address: str,
     ) -> AsyncIterator[Tuple[Dict[str, Any], ...]]:
-        async for batch in self._iter_batches(self.get_contract_big_maps, address):
+        async for batch in self._iter_batches(self.get_contract_big_maps, address, cursor=False):
             yield batch
 
     async def get_head_block(self) -> HeadBlockData:
@@ -583,7 +575,7 @@ async def get_big_maps(
                 "path.in": ",".join(paths),
                 "level.gt": first_level,
                 "level.le": last_level,
-                "offset.cr": offset,
+                "offset": offset,
                 "limit": limit,
             },
             cache=cache,
@@ -605,6 +597,7 @@ async def iter_big_maps(
             first_level,
             last_level,
             cache,
+            cursor=False,
         ):
             yield batch
 
@@ -711,15 +704,22 @@ async def _on_subscribe(message: CompletionMessage) -> None:
         await self._send(method, request, _on_subscribe)
         await event.wait()
 
-    async def _iter_batches(self, fn, *args, **kwargs) -> AsyncIterator:
+    async def _iter_batches(self, fn, *args, cursor: bool = True, **kwargs) -> AsyncIterator:
         if 'offset' in kwargs or 'limit' in kwargs:
             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)
+            if not result:
+                return
+
             yield result
-            offset = result[-1]['id']
+
             size = len(result)
+            if cursor:
+                offset = result[-1]['id']
+            else:
+                offset += self.request_limit
 
     def _get_ws_client(self) -> SignalRClient:
         """Create SignalR client, register message callbacks"""

src/dipdup/prometheus.py

@@ -1,5 +1,4 @@
 import time
-from collections import deque
 from contextlib import contextmanager
 
 from prometheus_client import Counter  # type: ignore
@@ -15,27 +14,24 @@
 _index_level_sync_duration = Histogram(
     'dipdup_index_level_sync_duration_seconds',
     'Duration of indexing a single level',
-    ['field'],
 )
 _index_level_realtime_duration = Histogram(
     'dipdup_index_level_realtime_duration_seconds',
     'Duration of last index syncronization',
-    ['field'],
 )
 _index_total_sync_duration = Histogram(
     'dipdup_index_total_sync_duration_seconds',
     'Duration of the last index syncronization',
-    ['field'],
 )
 _index_total_realtime_duration = Histogram(
     'dipdup_index_total_realtime_duration_seconds',
     'Duration of the last index realtime syncronization',
-    ['field'],
 )
 
 _index_levels_to_sync = Histogram(
     'dipdup_index_levels_to_sync_total',
     'Number of levels to reach synced state',
+    ['index'],
 )
 _index_levels_to_realtime = Histogram(
     'dipdup_index_levels_to_realtime_total',
@@ -70,14 +66,6 @@
 )
 
 
-@contextmanager
-def _average_duration(queue: deque):
-    start = time.perf_counter()
-    yield
-    end = time.perf_counter()
-    queue.appendleft(end - start)
-
-
 class Metrics:
     enabled = False
 
@@ -87,25 +75,25 @@ def __new__(cls):
     @classmethod
     @contextmanager
     def measure_level_sync_duration(cls):
-        with _average_duration(cls._level_sync_durations):
+        with _index_level_sync_duration.time():
             yield
 
     @classmethod
     @contextmanager
     def measure_level_realtime_duration(cls):
-        with _average_duration(cls._level_realtime_durations):
+        with _index_level_realtime_duration.time():
             yield
 
     @classmethod
     @contextmanager
     def measure_total_sync_duration(cls):
-        with _average_duration(cls._total_sync_durations):
+        with _index_total_sync_duration.time():
             yield
 
     @classmethod
     @contextmanager
     def measure_total_realtime_duration(cls):
-        with _average_duration(cls._total_realtime_durations):
+        with _index_total_realtime_duration.time():
             yield
 
     @classmethod