Anchor expansion on the case, and measure what log scanning actually costs

Expanding a counterparty has a cheaper question available than "everything
this address ever did", and the case already contains it: when a seed's
transfers span 113.61M to 113.92M, its counterparties are interesting around
*that*. `_known_range` reads the oldest block the case knows and starts there,
less a margin --- the margin because what a counterparty did *before* the
money arrived is frequently the point, and the LpdFi attacker's funding sat
two thousand blocks ahead of the exploit.

`_fetch_into` and `_fetch_page` take `start_block`; `expand` passes it.

**And the measurement that says this is not enough.** The endpoint costs about
a millisecond per block scanned, flat, whatever the chunk width --- 500 blocks
in 1.0s, 5,000 in 5.2s, 20,000 in 20.7s. Widening buys fewer requests and no
time at all, so the only lever is scanning fewer blocks, and anchoring saved
10%: the case's oldest block already sat near the window start.

  seed, 400,000 blocks, two directions   828s of server time, 207s at 4 concurrent
  anchored expansion, 358,466 blocks     742s, 186s
  three addresses                        9.3 minutes

That is the honest cost of reconstructing transfers from `eth_getLogs` on a
free public node, and it is not a bug to be tuned away. A log scan is the
fallback that makes an unindexed chain readable at all; it is not a substitute
for an indexer, and multi-address expansion is where the difference stops
being academic. Recorded here so the next person does not spend an afternoon
rediscovering it.

`_known_range` degrades to None on a store that cannot answer, because it is
an optimisation and must not break a fetch for the sake of making it cheaper.
Two test doubles caught that and the argument being threaded through.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FXWQ25VgFeMuuMhGWjvdof

Author: Toby1009

src/chainscope/server/local.py

@@ -499,7 +499,12 @@ def _fetch_one(self, key: str, chain: ChainId) -> tuple[int, bool, str | None]:
         """
         store = self._store(create=True)
         try:
-            fetched, complete = _fetch_into(store, key, chain)
+            # Anchored on what the case already knows, when it knows anything.
+            # A counterparty is interesting around the money that reached it,
+            # not for four hundred thousand blocks back from the chain head.
+            fetched, complete = _fetch_into(
+                store, key, chain, start_block=_known_range(store, chain) or 0
+            )
             return fetched, complete, None
         except (ProviderError, OSError) as exc:
             return 0, False, f"{type(exc).__name__}: {exc}"
@@ -1202,7 +1207,12 @@ def _sift(
 
 
 def _fetch_page(
-    provider: Any, chain: ChainId, address: str, page: int
+    provider: Any,
+    chain: ChainId,
+    address: str,
+    page: int,
+    *,
+    start_block: int | str = 0,
 ) -> tuple[list[Any], bool, bool]:
     """One page. Returns ``(rows, ended)`` and never raises for an empty listing.
 
@@ -1248,7 +1258,14 @@ def note(outcome: Any, rows: int, detail: str = "") -> None:
 
     try:
         rows = list(
-            provider.asset_transfers(chain, address, direction="all", limit=1000, page=page)
+            provider.asset_transfers(
+                chain,
+                address,
+                direction="all",
+                limit=1000,
+                page=page,
+                start_block=start_block,
+            )
         )
         note("ok" if rows else "empty", len(rows))
         return rows, False, False
@@ -1281,8 +1298,55 @@ def note(outcome: Any, rows: int, detail: str = "") -> None:
         raise
 
 
+def _known_range(store: SqliteStore, chain: ChainId, margin: int = 50_000) -> int | None:
+    """The oldest block the case already knows about, less a margin.
+
+    Expansion has a cheaper question available than "everything this address
+    ever did", and the case already contains it. When a seed's transfers span
+    blocks 113.61M to 113.92M, its counterparties are interesting *around
+    that*, and scanning from there rather than from a fixed window back from
+    the chain head is the difference between reading 100,000 blocks and
+    400,000.
+
+    Measured on the endpoint this was failing against: response time is about
+    a millisecond per block scanned, flat, whatever the chunk width --- 500
+    blocks in 1.0s, 5,000 in 5.2s, 20,000 in 20.7s. Widening buys fewer
+    requests and no time at all, so the only lever that matters is scanning
+    fewer blocks. A 400,000-block window over two directions is roughly 800
+    seconds of somebody else's server, per address, which does not survive
+    expanding three of them.
+
+    The margin exists because the interesting thing about a counterparty is
+    frequently what it did *before* the money arrived --- the funding of the
+    LpdFi attacker sat two thousand blocks ahead of the exploit, and a range
+    starting exactly at the earliest known transfer would have missed it.
+
+    Returns None when the case is empty, which means the caller has nothing
+    better to go on and should use its own default.
+    """
+    # Guarded, because this is an optimisation and not a requirement. A store
+    # that cannot answer means "no better idea than the default", which is
+    # exactly what None already communicates --- failing here would break a
+    # fetch for the sake of making it cheaper.
+    reader = getattr(store, "transfers", None)
+    if reader is None:
+        return None
+    oldest: int | None = None
+    for row in reader(Query(chain=chain, limit=20_000)):
+        block = getattr(row, "block", None)
+        if block and (oldest is None or block < oldest):
+            oldest = block
+    return max(0, oldest - margin) if oldest else None
+
+
 def _fetch_into(
-    store: SqliteStore, address: str, chain: ChainId, *, max_pages: int = 15, width: int = 4
+    store: SqliteStore,
+    address: str,
+    chain: ChainId,
+    *,
+    max_pages: int = 15,
+    width: int = 4,
+    start_block: int | str = 0,
 ) -> tuple[int, bool]:
     """Pull an address's transfers into the store, paging until they run out.
 
@@ -1341,7 +1405,9 @@ def absorb(batch: list[Any]) -> int:
     refused: Exception | None = None
     for candidate in providers:
         try:
-            first, ended, short = _fetch_page(candidate, chain, address, 1)
+            first, ended, short = _fetch_page(
+                candidate, chain, address, 1, start_block=start_block
+            )
             provider = candidate
             break
         except ProviderError as exc:
@@ -1368,7 +1434,10 @@ def absorb(batch: list[Any]) -> int:
             window = list(range(page, min(page + span, max_pages + 1)))
             with ThreadPoolExecutor(max_workers=len(window)) as pool:
                 jobs = {
-                    n: pool.submit(_fetch_page, provider, chain, address, n) for n in window
+                    n: pool.submit(
+                        _fetch_page, provider, chain, address, n, start_block=start_block
+                    )
+                    for n in window
                 }
                 # Absorbed in page order regardless of completion order, so the
                 # rows land in the same sequence a serial read would produce

tests/unit/test_paging_widens_only_after_it_must.py

@@ -30,7 +30,7 @@ def __init__(self, pages: list[list[object]], *, delay: float = 0.0):
         self.asked: list[int] = []
         self._lock = threading.Lock()
 
-    def asset_transfers(self, chain, address, *, direction, limit, page):
+    def asset_transfers(self, chain, address, *, direction, limit, page, **_):
         with self._lock:
             self.asked.append(page)
         if self._delay:
@@ -115,7 +115,7 @@ def test_rows_land_in_page_order_whatever_the_network_did(monkeypatch) -> None:
     """Completion order must not decide the stored sequence."""
 
     class Jittery(FakeProvider):
-        def asset_transfers(self, chain, address, *, direction, limit, page):
+        def asset_transfers(self, chain, address, *, direction, limit, page, **_):
             # Later pages return sooner, so completion order is reversed.
             time.sleep(max(0.0, 0.3 - 0.1 * page))
             return FakeProvider.asset_transfers(