get_logs now individually requests logs by block hash for unstable blocks (default last 2). requesting a filter with toBlock higher than what the node knows of just ignores the missing blocks which results in missed events.

Author: simone1999

IceCreamSwapWeb3/EthAdvanced.py

@@ -125,17 +125,18 @@ def get_logs(
             p_bar=None,
             no_retry: bool = False
     ) -> list[LogReceipt]:
-        if not self.w3.should_retry:
-            no_retry = True
+        filter_block_range = self.w3.filter_block_range
+        if filter_block_range == 0:
+            raise Exception("RPC does not support eth_getLogs")
 
-        # getting the respective block numbers, could be block hashes or strings like "latest"
-        from_block = filter_params["fromBlock"]
-        to_block = filter_params["toBlock"]
-        if not isinstance(from_block, int):
-            from_block = self.get_block(from_block)["number"]
-        if not isinstance(to_block, int):
-            to_block = self.get_block(to_block)["number"]
+        # getting logs for a single block defined by its block hash. No drama
+        if "blockHash" in filter_params:
+            assert "fromBlock" not in filter_params and "toBlock" not in filter_params
+            return self.get_logs_inner(filter_params, no_retry=no_retry)
 
+        # sanitizing block numbers, could be strings like "latest"
+        filter_params["fromBlock"] = from_block = self.get_block_number_from_identifier(filter_params["fromBlock"])
+        filter_params["toBlock"] = to_block = self.get_block_number_from_identifier(filter_params["toBlock"])
         assert to_block >= from_block, f"{from_block=}, {to_block=}"
 
         # note: fromBlock and toBlock are both inclusive. e.g. 5 to 6 are 2 blocks
@@ -147,51 +148,66 @@ def get_logs(
             from tqdm import tqdm
             p_bar = tqdm(total=num_blocks)
 
+        kwargs = dict(
+            show_progress_bar=show_progress_bar,
+            p_bar=p_bar,
+            no_retry=no_retry,
+        )
+
+        # the latest blocks might be available on some nodes but not at others.
+        # setting toBlock to a block bigger than the latest known block of the node
+        # simply ignores logs from the missing block
+        # to prevent this, we get the latest blocks individually by their hashes
+        unstable_blocks = self.w3.unstable_blocks
+        if to_block > self.w3.latest_seen_block - unstable_blocks and to_block > self.block_number - unstable_blocks:
+            results = []
+            while to_block > self.w3.latest_seen_block - unstable_blocks and to_block >= from_block:
+                single_hash_filter = {**filter_params, "blockHash": self.get_block(to_block)["hash"]}
+                del single_hash_filter["fromBlock"]
+                del single_hash_filter["toBlock"]
+                results += self.get_logs_inner(single_hash_filter, no_retry=no_retry)
+                to_block -= 1
+                if p_bar is not None:
+                    p_bar.update(1)
+            if to_block >= from_block:
+                results += self.get_logs({**filter_params, "toBlock": to_block}, **kwargs)
+            return results
+
+        # getting logs for a single block, which is not at the chain head. No drama
+        if num_blocks == 1:
+            return self.get_logs_inner(filter_params, no_retry=no_retry)
+
         # if we already know that the filter range is too large, split it
-        filter_block_range = self.w3.filter_block_range
-        if filter_block_range == 0:
-            raise Exception("RPC does not support eth_getLogs")
         if num_blocks > filter_block_range:
             results = []
             for filter_start in range(from_block, to_block + 1, filter_block_range):
-                filter_end = min(filter_start + filter_block_range - 1, to_block)
-                partial_filter = filter_params.copy()
-                partial_filter["fromBlock"] = filter_start
-                partial_filter["toBlock"] = filter_end
-                results += self.get_logs(partial_filter, show_progress_bar=show_progress_bar, p_bar=p_bar)
+                results += self.get_logs({
+                    **filter_params,
+                    "fromBlock": filter_start,
+                    "toBlock": min(filter_start + filter_block_range - 1, to_block),
+                }, **kwargs)
             return results
 
         # get logs
         try:
             events = self._get_logs(filter_params)
-        except Exception:
-            # if errors should not be retried, still do splitting but not retry if it can not be split further
-            if no_retry and num_blocks == 1:
-                raise
-        else:
             if p_bar is not None:
                 p_bar.update(num_blocks)
             return events
-
-        # if directly getting logs did not work, split the filter range and try again
-        if num_blocks > 1:
+        except Exception:
+            # split the filter range and try again
             mid_block = (from_block + to_block) // 2
-            left_filter = filter_params.copy()
-            left_filter["toBlock"] = mid_block
-            right_filter = filter_params.copy()
-            right_filter["fromBlock"] = mid_block + 1
+            left_filter = {**filter_params, "toBlock": mid_block}
+            right_filter = {**filter_params, "fromBlock": mid_block + 1}
+            return self.get_logs(left_filter, **kwargs) + self.get_logs(right_filter, **kwargs)
 
-            results = []
-            results += self.get_logs(left_filter, show_progress_bar=show_progress_bar, p_bar=p_bar)
-            results += self.get_logs(right_filter, show_progress_bar=show_progress_bar, p_bar=p_bar)
-            return results
+    def get_logs_inner(self, filter_params: FilterParams, no_retry: bool = False):
+        if not self.w3.should_retry:
+            no_retry = True
+        return exponential_retry(func_name="get_logs")(self._get_logs)(filter_params, no_retry=no_retry)
 
-        # filter is trying to get a single block, retrying till it works
-        assert from_block == to_block and num_blocks == 1, f"{from_block=}, {to_block=}, {num_blocks=}"
-        events = exponential_retry(func_name="get_logs")(self._get_logs)(filter_params)
-        if p_bar is not None:
-            p_bar.update(num_blocks)
-        return events
+    def get_block_number_from_identifier(self, block_identifier: BlockIdentifier) -> BlockNumber:
+        return block_identifier if isinstance(block_identifier, int) else self.get_block(block_identifier)["number"]
 
     def _chain_id(self):
         # usually this causes an RPC call and is used in every eth_call. Getting it once in the init and then not again.

IceCreamSwapWeb3/Web3Advanced.py

@@ -34,11 +34,13 @@ class Web3Advanced(Web3):
     def __init__(
             self,
             node_url: str,
-            should_retry=True,
+            should_retry: bool = True,
+            unstable_blocks: int = 2,  # not all nodes might have latest n blocks, these are seen as unstable
     ):
         patch_error_formatters()
         self.node_url = node_url
         self.should_retry = should_retry
+        self.unstable_blocks = unstable_blocks
 
         provider = self._construct_provider(node_url=self.node_url)