# Author: Simon-Pierre Boucher # Mail: contact@spboucher.ai """Per-chain EVM ingestion loop. Head tailing: poll head → compute safe head (head - confirmations) → verify the chain still links to our cursor (parent-hash check) → sweep forward in eth_getLogs ranges → decode → insert → advance cursor. Backfill: after the head is caught up, grow history BACKWARDS one slice per cycle (head tailing always has priority) until coverage reaches `backfill_days` (chain config) / EXPLORER_BACKFILL_DAYS (env). The `backfill_block` cursor is the lowest indexed block. Reorg safety: two independent checks — (1) each cycle re-fetches the cursor block and compares its hash; (2) each advance fetches the next block's header and verifies parentHash == our stored cursor hash. On mismatch: delete the last 2×confirmations blocks of transfers and rewind. Range sizing is adaptive: free RPCs cap getLogs responses, so a semantic refusal halves the range (down to 2); refusal at minimum range benches the endpoint via pool.rotate(). Ten clean sweeps grow the range back. """ import logging import time from . import config, db from .decode import TRANSFER_TOPIC, decode_transfer from .rpc import RpcError, RpcPool log = logging.getLogger("ingest") class ChainIndexer: def __init__(self, chain, cfg, tokens, db_path): self.chain = chain self.cfg = cfg self.db_path = db_path self.conn = None # opened in run(): sqlite conns are thread-bound self.pool = RpcPool(cfg["rpcs"]) self.tokens = {t["address"].lower(): t for t in tokens} self.max_range = cfg.get("max_range", 100) self.range = self.max_range self.backfill_blocks = config.backfill_blocks(cfg) self.native = cfg.get("native") # {symbol, min, decimals} — whale-only self._clean_sweeps = 0 # -- RPC helpers --------------------------------------------------- def head(self): return int(self.pool.call("eth_blockNumber"), 16) def block_header(self, number): return self.pool.call("eth_getBlockByNumber", [hex(number), False]) def fetch_logs(self, frm, to): return self.pool.call( "eth_getLogs", [{ "fromBlock": hex(frm), "toBlock": hex(to), "address": list(self.tokens.keys()), "topics": [TRANSFER_TOPIC], }], ) def block_timestamps(self, numbers): """Batch-fetch headers just for blocks that actually had transfers.""" out = {} nums = sorted(numbers) for i in range(0, len(nums), 20): # small chunks: free nodes cap batches chunk = nums[i : i + 20] results = self.pool.batch( [("eth_getBlockByNumber", [hex(n), False]) for n in chunk] ) for n, blk in zip(chunk, results): if blk: out[n] = int(blk["timestamp"], 16) # batches drop entries under rate limits — retry stragglers one by one for n in nums: if n not in out: try: blk = self.block_header(n) if blk: out[n] = int(blk["timestamp"], 16) except (RpcError, RuntimeError): pass return out # -- pipeline ------------------------------------------------------ def process_range(self, frm, to): logs = self.fetch_logs(frm, to) rows = [] if logs: ts = self.block_timestamps({int(l["blockNumber"], 16) for l in logs}) for l in logs: if l.get("removed"): continue meta = self.tokens.get(l["address"].lower()) if meta is None or len(l.get("topics", [])) != 3: continue # not one of ours / non-standard Transfer rows.append( decode_transfer(self.chain, l, meta, ts.get(int(l["blockNumber"], 16))) ) db.insert_transfers(self.conn, rows) return len(rows) def try_range(self, frm, to, stop): """One adaptive sweep starting at `frm`. Returns (end, n): the last block actually covered (may be < `to` after halving) and row count.""" while not stop.is_set(): end = min(frm + self.range - 1, to) try: n = self.process_range(frm, end) except RpcError as e: if self.range > 2: self.range = max(2, self.range // 2) self._clean_sweeps = 0 log.info("%s: getLogs refused (%s) — range now %d", self.chain, e.message[:80], self.range) continue # refused even at minimum range: this provider just won't # serve getLogs — bench it and let selection move on log.warning("%s: getLogs refused at min range (%s) — rotating " "away from %s", self.chain, e.message[:80], self.pool.current_url) self.pool.rotate() stop.wait(5) continue self._clean_sweeps += 1 if self._clean_sweeps >= 10 and self.range < self.max_range: self.range = min(self.max_range, self.range * 2) self._clean_sweeps = 0 return end, n return frm - 1, 0 # -- reorg checks ---------------------------------------------------- def process_native(self, frm, to): """Whale-only native-coin transfers: full blocks are expensive, so we only capture value >= min (config) and only while head-tailing over a bounded range — never during backfill. A skipped stretch is logged.""" cap = 60 if to - frm + 1 > cap: log.info("%s: native capture skipped for %d..%d (catch-up burst)", self.chain, frm, to - cap) frm = to - cap + 1 min_wei = int(self.native["min"] * 10 ** self.native["decimals"]) rows = [] nums = list(range(frm, to + 1)) for i in range(0, len(nums), 10): chunk = nums[i : i + 10] blocks = self.pool.batch( [("eth_getBlockByNumber", [hex(n), True]) for n in chunk] ) for blk in blocks: if not blk: continue ts = int(blk["timestamp"], 16) for ti, t in enumerate(blk.get("transactions") or []): if not isinstance(t, dict): continue val = int(t.get("value", "0x0"), 16) if val < min_wei: continue rows.append({ "chain": self.chain, "block": int(blk["number"], 16), "block_hash": blk["hash"].lower(), "tx_hash": t["hash"].lower(), "log_index": 100000 + ti, # never collides with log indexes "timestamp": ts, "token": "native", "symbol": self.native["symbol"], "from": (t.get("from") or "").lower() or None, "to": (t.get("to") or "").lower() or None, "amount": str(val), "decimals": self.native["decimals"], }) db.insert_transfers(self.conn, rows) if rows: log.info("%s: %d native %s whale transfers in ..%d", self.chain, len(rows), self.native["symbol"], to) def reorged(self, cursor_block, cursor_hash): """True if the block we last indexed is no longer canonical.""" if not cursor_hash: return False blk = self.block_header(cursor_block) return blk is not None and blk["hash"].lower() != cursor_hash.lower() def links_to_cursor(self, cursor_block, cursor_hash): """Parent-hash link: next block's parentHash must be our cursor hash.""" if not cursor_hash: return True nxt = self.block_header(cursor_block + 1) return nxt is None or nxt["parentHash"].lower() == cursor_hash.lower() # -- main loop ----------------------------------------------------- def run(self, stop): self.conn = db.connect(self.db_path) cur = db.get_cursor(self.conn, self.chain) confirmations = self.cfg.get("confirmations", 6) while not stop.is_set(): try: head = self.head() safe = head - confirmations if cur is None: start = max(0, safe - self.cfg.get("start_offset", 300)) cur = (start, None, start) db.set_cursor(self.conn, self.chain, start, None) db.set_backfill(self.conn, self.chain, start) log.info("%s: fresh start at block %d (head %d)", self.chain, start, head) cursor_block, cursor_hash, backfill_block = cur if backfill_block is None: # DB predates backfill support backfill_block = cursor_block db.set_backfill(self.conn, self.chain, backfill_block) if self.reorged(cursor_block, cursor_hash): to_block = cursor_block - 2 * confirmations log.warning("%s: reorg at %d — rolling back to %d", self.chain, cursor_block, to_block) db.rollback(self.conn, self.chain, to_block) cursor_block, cursor_hash = to_block, None # ---- head tailing (always first priority) ---- while cursor_block < safe and not stop.is_set(): if not self.links_to_cursor(cursor_block, cursor_hash): to_block = cursor_block - 2 * confirmations log.warning("%s: parent-hash mismatch after %d — rolling back to %d", self.chain, cursor_block, to_block) db.rollback(self.conn, self.chain, to_block) cursor_block, cursor_hash = to_block, None continue end, n = self.try_range(cursor_block + 1, safe, stop) if end < cursor_block + 1: break # stopped mid-sweep if self.native: try: self.process_native(cursor_block + 1, end) except (RpcError, RuntimeError) as e: log.warning("%s: native capture failed (%s) — " "continuing", self.chain, e) end_blk = self.block_header(end) cursor_block = end cursor_hash = end_blk["hash"].lower() if end_blk else None db.set_cursor(self.conn, self.chain, cursor_block, cursor_hash, head=head) if n: log.info("%s: %d transfers in blocks ..%d (lag %d)", self.chain, n, end, head - end) # ---- backfill: one slice per cycle, backwards ---- if self.backfill_blocks and not stop.is_set(): target = max(0, safe - self.backfill_blocks) if backfill_block > target: frm = max(target, backfill_block - self.range) covered, total = frm - 1, 0 while covered < backfill_block - 1 and not stop.is_set(): end, n = self.try_range(covered + 1, backfill_block - 1, stop) if end <= covered: break covered, total = end, total + n if covered >= backfill_block - 1: # slice fully covered db.set_backfill(self.conn, self.chain, frm) backfill_block = frm if total: log.info("%s: backfill %d transfers, floor now %d (target %d)", self.chain, total, frm, target) cur = (cursor_block, cursor_hash, backfill_block) db.save_rpc_health(self.conn, self.chain, self.pool.stats(), int(time.time())) stop.wait(max(float(self.cfg.get("block_time", 12)), 2.0)) except Exception as e: log.error("%s: %s — retrying in 10s", self.chain, e) stop.wait(10)