ZzyzxLabs
diff --git a/‎src/chainscope/providers/jsonrpc.py‎
Lines changed: 118 additions & 38 deletions b/‎src/chainscope/providers/jsonrpc.py‎
Lines changed: 118 additions & 38 deletions
diff --git a/‎tests/unit/test_the_scan_covers_what_it_was_given.py‎
Lines changed: 134 additions & 0 deletions b/‎tests/unit/test_the_scan_covers_what_it_was_given.py‎
Lines changed: 134 additions & 0 deletions
@@ -68,23 +68,37 @@
 _SPAN_START = 5_000
 _SPAN_FLOOR = 500
 
+#: How wide a chunk may grow once the endpoint has proved it can take it.
+#:
+#: Starting narrow and widening on *success* costs nothing when it fails to
+#: widen, where starting wide and narrowing on failure costs a failed request
+#: per step --- and a failed request that is a timeout is indistinguishable
+#: from an unwell host, which is what opened the circuit breaker mid-fetch.
+#: So the first chunk is cheap and cautious and the rest ride on what it
+#: learned.
+_SPAN_CEILING = 40_000
+
 #: How far back a scan reaches when the caller does not say.
 #:
 #: Not genesis. A full-history scan is hundreds of thousands of requests, and
 #: issuing it because somebody left a default alone is not a thing this should
 #: do quietly. The window is stated in the truncation message whenever it does
 #: not reach the requested start, so a short answer is never mistaken for a
 #: complete one.
-#: Roughly a day and a half of BSC, and a few days of a slower chain.
+#: How far back a scan reaches when the caller does not say.
 #:
-#: Halved from 400,000 after measuring what it costs: at 5,000 blocks a
-#: request and four in flight, 400,000 blocks in both directions is 160 round
-#: trips and several minutes of somebody watching a spinner. The window is
-#: named in the `ResultTruncated` message whenever it does not reach the
-#: requested start, so a short answer still says how short --- which is the
-#: part that matters. Somebody who needs deeper history passes `start_block`
-#: and accepts the wait deliberately.
-_DEFAULT_WINDOW = 120_000
+#: Briefly 120,000, which was a mistake made while chasing timeouts: the
+#: address it was being tested against had its whole history 300,000 blocks
+#: back, so the shallower window returned three address-poisoning dust
+#: transfers and none of the case. It said so --- `capped`, with the range ---
+#: which is the difference between a wrong answer and a short one, but short
+#: was still useless. The span now widens on success instead, so the deeper
+#: window costs round trips rather than minutes.
+#:
+#: Still not genesis. A full-history scan is hundreds of thousands of
+#: requests, and the window is named in the `ResultTruncated` message whenever
+#: it does not reach the requested start.
+_DEFAULT_WINDOW = 400_000
 
 #: Chunks in flight at once. Small on purpose --- see `_scan`.
 _SCAN_WORKERS = 4
@@ -144,10 +158,26 @@ def __init__(
         #: Token contract to (symbol, decimals). Per instance, so a scan asks
         #: each contract once however many transfers it produced.
         self._meta: dict[str, tuple[str, int]] = {}
+        #: Block span this endpoint has proved it will serve. Starts cautious,
+        #: doubles on success, halves on refusal. Shared across the chunks of a
+        #: scan so the limit is discovered once rather than per chunk.
+        self._span = _SPAN_START
+        #: Widest span this endpoint has been *seen to refuse*, minus room.
+        #:
+        #: Without it, widening undoes narrowing: succeed at 1,250, double to
+        #: 2,500, get refused, halve to 1,250, succeed, double again --- a
+        #: wasted failed request every other round, for ever. A refusal is
+        #: information about the endpoint's limit and has to be remembered as
+        #: such, not just reacted to.
+        self._span_cap = _SPAN_CEILING
 
     @classmethod
     def from_settings(cls, settings: Any, chain: ChainId, client: Any = None) -> list[Provider]:
-        """The configured endpoint for this chain, if there is one.
+        """The configured endpoints for this chain, in preference order.
+
+        ``CHAINSCOPE_RPC_<NAME>`` accepts a comma-separated list. One endpoint
+        was the original design and it made a single flaky public node a single
+        point of failure for a whole chain.
 
         Endpoints are keyed by short name (``CHAINSCOPE_RPC_ETHEREUM``), so the
         lookup goes through the alias table rather than CAIP-2 --- a user types
@@ -171,17 +201,36 @@ def from_settings(cls, settings: Any, chain: ChainId, client: Any = None) -> lis
             return []
         names = [n for n, c in ALIASES.items() if c == chain]
         for name in sorted(names, key=len, reverse=True):
-            url = settings.rpc.get(name)
-            if url:
-                return [
-                    cls(
-                        url,
-                        chain,
-                        client=client,
-                        native_symbol=native_symbol(chain, "ETH"),
-                        archive=settings.rpc_archive.get(name, False),
-                    )
-                ]
+            configured = settings.rpc.get(name)
+            if not configured:
+                continue
+            # Comma-separated, and each one becomes its own provider.
+            #
+            # A single endpoint per chain means one flaky free public node is
+            # the difference between the tool working and not working at all.
+            # Measured while chasing exactly that: of eleven public BSC
+            # endpoints, one served `eth_getLogs` over 5,000 blocks, five
+            # capped at a smaller range, two refused the method, two returned
+            # 401/403 and one was gone --- and the one that worked timed out
+            # under load an hour later, which took the whole capability down
+            # because there was nothing behind it.
+            #
+            # The router already falls over between providers and the circuit
+            # breaker already skips an unwell host. Neither could help while
+            # there was only ever one. Order is preference order: the first is
+            # tried first, and the rest exist for when it is not answering.
+            urls = [u.strip() for u in str(configured).split(",") if u.strip()]
+            archive = settings.rpc_archive.get(name, False)
+            return [
+                cls(
+                    url,
+                    chain,
+                    client=client,
+                    native_symbol=native_symbol(chain, "ETH"),
+                    archive=archive,
+                )
+                for url in urls
+            ]
         return []
 
     @property
@@ -417,14 +466,21 @@ def topics_for(way: str) -> list[Any]:
         # The cap is on the total because that is what the endpoint sees.
         order = sorted(ways)
         with ThreadPoolExecutor(max_workers=_SCAN_WORKERS) as pool:
-            plans = [
-                (way, lo_, min(lo_ + _SPAN_START - 1, hi))
-                for way in order
-                for lo_ in range(lo, hi + 1, _SPAN_START)
-            ]
+            # Chunked at the ceiling and let `_chunk` narrow to whatever the
+            # endpoint will actually take. Planning at `_SPAN_START` instead
+            # would fix the request count at the cautious width and never
+            # benefit from an endpoint that turns out to be generous.
+            plans = [(way, lo_) for way in order for lo_ in range(lo, hi + 1, _SPAN_CEILING)]
             jobs = [
-                pool.submit(self._chunk, chain, lo_, hi_, topics_for(way), contract)
-                for way, lo_, hi_ in plans
+                pool.submit(
+                    self._chunk,
+                    chain,
+                    lo_,
+                    min(lo_ + _SPAN_CEILING - 1, hi),
+                    topics_for(way),
+                    contract,
+                )
+                for way, lo_ in plans
             ]
             found = [job.result() for job in jobs]
 
@@ -467,7 +523,13 @@ def _chunk(
         topics: list[Any],
         contract: str | None,
     ) -> list[dict[str, Any]]:
-        """One block range, narrowing itself until the endpoint accepts it.
+        """One block range, covered in whatever width the endpoint will take.
+
+        **Covers the whole range it was given.** An earlier version fetched a
+        single `span`-wide request and returned, silently dropping the rest of
+        its assignment --- a gap in the middle of a scan that nothing
+        downstream could have detected, since fewer logs is exactly what an
+        address with less activity looks like.
 
         Endpoints cap `eth_getLogs` by range, by result count, or by wall
         clock, and report each differently --- measured against four BSC nodes,
@@ -480,20 +542,38 @@ def _chunk(
         both directions: a pool here as well as there multiplied the
         concurrency and tripped the circuit breaker on a healthy node.
         """
-        span = hi - lo + 1
-        while True:
+        out: list[dict[str, Any]] = []
+        at = lo
+        while at <= hi:
+            # Re-read each time round: another chunk running concurrently may
+            # have already learned that this endpoint is more or less generous
+            # than we thought.
+            span = max(_SPAN_FLOOR, min(self._span, hi - at + 1))
             try:
-                return self.get_logs(
-                    chain,
-                    address=contract,
-                    topics=topics,
-                    from_block=lo,
-                    to_block=min(lo + span - 1, hi),
+                out.extend(
+                    self.get_logs(
+                        chain,
+                        address=contract,
+                        topics=topics,
+                        from_block=at,
+                        to_block=min(at + span - 1, hi),
+                    )
                 )
             except ProviderError:
                 if span <= _SPAN_FLOOR:
                     raise
-                span = max(_SPAN_FLOOR, span // 2)
+                # Narrower for everyone from here on. Rediscovering the same
+                # limit on every chunk is what made narrowing expensive.
+                narrower = max(_SPAN_FLOOR, span // 2)
+                self._span_cap = min(self._span_cap, narrower)
+                self._span = narrower
+                continue
+            at += span
+            # It worked, so try more next time. Doubling rather than jumping
+            # to the ceiling, so an endpoint that is merely slow gets to refuse
+            # once rather than being asked for eight times what it just served.
+            self._span = min(self._span_cap, max(self._span, span * 2))
+        return out
 
     def _token_meta(self, token: str) -> tuple[str, int]:
         """``(symbol, decimals)`` for a token contract, asked once each.
 
@@ -0,0 +1,134 @@
+"""A chunk must cover its whole assignment, and the span must adapt upward.
+
+Two bugs, one after the other, both mine, both in the shape this package is
+supposed to catch.
+
+`_chunk` was handed a 40,000-block range and fetched a single `span`-wide
+request from the start of it, then returned. The remaining 35,000 blocks were
+never read and nothing downstream could tell: fewer logs is exactly what an
+address with less activity looks like.
+
+And the span started wide and narrowed on failure, which costs a failed request
+per step. A failed request that is a timeout is indistinguishable from an unwell
+host, so four concurrent chunks each halving once opened the circuit breaker on
+an endpoint that was answering. Starting narrow and widening on *success* costs
+nothing when it declines to widen.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from chainscope.core.chainid import ChainId
+from chainscope.providers.base import ProviderError
+from chainscope.providers.jsonrpc import (
+    _SPAN_CEILING,
+    _SPAN_FLOOR,
+    _SPAN_START,
+    JsonRpcProvider,
+)
+
+CHAIN = ChainId.evm(56)
+
+
+class _Node:
+    """Records every range asked for, and optionally refuses wide ones."""
+
+    def __init__(self, max_span: int | None = None) -> None:
+        self.ranges: list[tuple[int, int]] = []
+        self.max_span = max_span
+
+    def rpc(self, url: str, method: str, params: Any = None, **_: Any) -> Any:
+        if method != "eth_getLogs":
+            return "0x0"
+        lo = int(params[0]["fromBlock"], 16)
+        hi = int(params[0]["toBlock"], 16)
+        if self.max_span is not None and (hi - lo + 1) > self.max_span:
+            raise ProviderError("query returned more than 10000 results")
+        self.ranges.append((lo, hi))
+        return []
+
+
+def provider(node: _Node) -> JsonRpcProvider:
+    return JsonRpcProvider("https://node.example", CHAIN, client=node)  # type: ignore[arg-type]
+
+
+def covered(ranges: list[tuple[int, int]]) -> set[int]:
+    seen: set[int] = set()
+    for lo, hi in ranges:
+        seen.update(range(lo, hi + 1))
+    return seen
+
+
+def test_a_chunk_covers_every_block_it_was_given() -> None:
+    """The gap bug. A 40,000-block assignment at a 5,000 span is eight
+    requests, not one."""
+    node = _Node()
+    rpc = provider(node)
+    rpc._chunk(CHAIN, 1_000_000, 1_039_999, ["0xtopic"], None)
+    assert covered(node.ranges) == set(range(1_000_000, 1_040_000))
+
+
+def test_no_range_is_requested_twice() -> None:
+    node = _Node()
+    provider(node)._chunk(CHAIN, 1_000_000, 1_039_999, ["0xtopic"], None)
+    blocks = [b for lo, hi in node.ranges for b in range(lo, hi + 1)]
+    assert len(blocks) == len(set(blocks))
+
+
+def test_the_first_request_is_cautious() -> None:
+    """Starting wide costs a failure to discover the limit; starting narrow
+    costs nothing when the endpoint turns out to be generous."""
+    node = _Node()
+    provider(node)._chunk(CHAIN, 1_000_000, 1_039_999, ["0xtopic"], None)
+    first_lo, first_hi = node.ranges[0]
+    assert first_hi - first_lo + 1 == _SPAN_START
+
+
+def test_the_span_widens_after_success() -> None:
+    node = _Node()
+    provider(node)._chunk(CHAIN, 1_000_000, 1_039_999, ["0xtopic"], None)
+    widths = [hi - lo + 1 for lo, hi in node.ranges]
+    assert widths[1] > widths[0], "a working endpoint should be asked for more"
+    assert max(widths) <= _SPAN_CEILING
+
+
+def test_the_span_narrows_on_refusal_and_stays_narrow() -> None:
+    """Rediscovering the same limit on every chunk is what made narrowing
+    expensive enough to trip the breaker."""
+    node = _Node(max_span=2_000)
+    rpc = provider(node)
+    rpc._chunk(CHAIN, 1_000_000, 1_019_999, ["0xtopic"], None)
+    assert covered(node.ranges) == set(range(1_000_000, 1_020_000))
+    assert all(hi - lo + 1 <= 2_000 for lo, hi in node.ranges)
+    assert rpc._span <= 2_000
+
+
+def test_an_endpoint_that_refuses_everything_raises_rather_than_looping() -> None:
+    node = _Node(max_span=1)
+    with pytest.raises(ProviderError):
+        provider(node)._chunk(CHAIN, 1_000_000, 1_019_999, ["0xtopic"], None)
+    assert all(hi - lo + 1 >= _SPAN_FLOOR for lo, hi in node.ranges) or not node.ranges
+
+
+def test_a_single_block_range_is_one_request() -> None:
+    node = _Node()
+    provider(node)._chunk(CHAIN, 1_000_000, 1_000_000, ["0xtopic"], None)
+    assert node.ranges == [(1_000_000, 1_000_000)]
+
+
+def test_widening_does_not_undo_narrowing() -> None:
+    """Without a remembered cap the span oscillates: succeed at 1,250, double
+    to 2,500, get refused, halve back --- a wasted failed request every other
+    round, for ever. Counted here rather than asserted in prose."""
+    node = _Node(max_span=2_000)
+    rpc = provider(node)
+    rpc._chunk(CHAIN, 1_000_000, 1_079_999, ["0xtopic"], None)
+    assert rpc._span <= 2_000
+    assert rpc._span_cap <= 2_000
+    # Every request that was *served* is inside the limit, and the served
+    # widths stop shrinking once the cap is known.
+    widths = {hi - lo + 1 for lo, hi in node.ranges}
+    assert max(widths) <= 2_000