spb/coinexplorer Public MIT

1# Author: Simon-Pierre Boucher2# Mail: contact@spboucher.ai3"""Per-chain EVM ingestion loop.45Head tailing:  poll head → compute safe head (head - confirmations) →6verify the chain still links to our cursor (parent-hash check) → sweep7forward in eth_getLogs ranges → decode → insert → advance cursor.89Backfill:      after the head is caught up, grow history BACKWARDS one10slice per cycle (head tailing always has priority) until coverage reaches11`backfill_days` (chain config) / EXPLORER_BACKFILL_DAYS (env). The12`backfill_block` cursor is the lowest indexed block.1314Reorg safety:  two independent checks — (1) each cycle re-fetches the15cursor block and compares its hash; (2) each advance fetches the next16block's header and verifies parentHash == our stored cursor hash. On17mismatch: delete the last 2×confirmations blocks of transfers and rewind.1819Range sizing is adaptive: free RPCs cap getLogs responses, so a semantic20refusal halves the range (down to 2); refusal at minimum range benches the21endpoint via pool.rotate(). Ten clean sweeps grow the range back.22"""2324import logging25import time2627from . import config, db28from .decode import TRANSFER_TOPIC, decode_transfer29from .rpc import RpcError, RpcPool3031log = logging.getLogger("ingest")323334class ChainIndexer:35    def __init__(self, chain, cfg, tokens, db_path):36        self.chain = chain37        self.cfg = cfg38        self.db_path = db_path39        self.conn = None  # opened in run(): sqlite conns are thread-bound40        self.pool = RpcPool(cfg["rpcs"])41        self.tokens = {t["address"].lower(): t for t in tokens}42        self.max_range = cfg.get("max_range", 100)43        self.range = self.max_range44        self.backfill_blocks = config.backfill_blocks(cfg)45        self.native = cfg.get("native")  # {symbol, min, decimals} — whale-only46        self._clean_sweeps = 04748    # -- RPC helpers ---------------------------------------------------4950    def head(self):51        return int(self.pool.call("eth_blockNumber"), 16)5253    def block_header(self, number):54        return self.pool.call("eth_getBlockByNumber", [hex(number), False])5556    def fetch_logs(self, frm, to):57        return self.pool.call(58            "eth_getLogs",59            [{60                "fromBlock": hex(frm),61                "toBlock": hex(to),62                "address": list(self.tokens.keys()),63                "topics": [TRANSFER_TOPIC],64            }],65        )6667    def block_timestamps(self, numbers):68        """Batch-fetch headers just for blocks that actually had transfers."""69        out = {}70        nums = sorted(numbers)71        for i in range(0, len(nums), 20):  # small chunks: free nodes cap batches72            chunk = nums[i : i + 20]73            results = self.pool.batch(74                [("eth_getBlockByNumber", [hex(n), False]) for n in chunk]75            )76            for n, blk in zip(chunk, results):77                if blk:78                    out[n] = int(blk["timestamp"], 16)79        # batches drop entries under rate limits — retry stragglers one by one80        for n in nums:81            if n not in out:82                try:83                    blk = self.block_header(n)84                    if blk:85                        out[n] = int(blk["timestamp"], 16)86                except (RpcError, RuntimeError):87                    pass88        return out8990    # -- pipeline ------------------------------------------------------9192    def process_range(self, frm, to):93        logs = self.fetch_logs(frm, to)94        rows = []95        if logs:96            ts = self.block_timestamps({int(l["blockNumber"], 16) for l in logs})97            for l in logs:98                if l.get("removed"):99                    continue100                meta = self.tokens.get(l["address"].lower())101                if meta is None or len(l.get("topics", [])) != 3:102                    continue  # not one of ours / non-standard Transfer103                rows.append(104                    decode_transfer(self.chain, l, meta, ts.get(int(l["blockNumber"], 16)))105                )106        db.insert_transfers(self.conn, rows)107        return len(rows)108109    def try_range(self, frm, to, stop):110        """One adaptive sweep starting at `frm`. Returns (end, n): the last111        block actually covered (may be < `to` after halving) and row count."""112        while not stop.is_set():113            end = min(frm + self.range - 1, to)114            try:115                n = self.process_range(frm, end)116            except RpcError as e:117                if self.range > 2:118                    self.range = max(2, self.range // 2)119                    self._clean_sweeps = 0120                    log.info("%s: getLogs refused (%s) — range now %d",121                             self.chain, e.message[:80], self.range)122                    continue123                # refused even at minimum range: this provider just won't124                # serve getLogs — bench it and let selection move on125                log.warning("%s: getLogs refused at min range (%s) — rotating "126                            "away from %s", self.chain, e.message[:80],127                            self.pool.current_url)128                self.pool.rotate()129                stop.wait(5)130                continue131            self._clean_sweeps += 1132            if self._clean_sweeps >= 10 and self.range < self.max_range:133                self.range = min(self.max_range, self.range * 2)134                self._clean_sweeps = 0135            return end, n136        return frm - 1, 0137138    # -- reorg checks ----------------------------------------------------139140    def process_native(self, frm, to):141        """Whale-only native-coin transfers: full blocks are expensive, so we142        only capture value >= min (config) and only while head-tailing over a143        bounded range — never during backfill. A skipped stretch is logged."""144        cap = 60145        if to - frm + 1 > cap:146            log.info("%s: native capture skipped for %d..%d (catch-up burst)",147                     self.chain, frm, to - cap)148            frm = to - cap + 1149        min_wei = int(self.native["min"] * 10 ** self.native["decimals"])150        rows = []151        nums = list(range(frm, to + 1))152        for i in range(0, len(nums), 10):153            chunk = nums[i : i + 10]154            blocks = self.pool.batch(155                [("eth_getBlockByNumber", [hex(n), True]) for n in chunk]156            )157            for blk in blocks:158                if not blk:159                    continue160                ts = int(blk["timestamp"], 16)161                for ti, t in enumerate(blk.get("transactions") or []):162                    if not isinstance(t, dict):163                        continue164                    val = int(t.get("value", "0x0"), 16)165                    if val < min_wei:166                        continue167                    rows.append({168                        "chain": self.chain,169                        "block": int(blk["number"], 16),170                        "block_hash": blk["hash"].lower(),171                        "tx_hash": t["hash"].lower(),172                        "log_index": 100000 + ti,  # never collides with log indexes173                        "timestamp": ts,174                        "token": "native",175                        "symbol": self.native["symbol"],176                        "from": (t.get("from") or "").lower() or None,177                        "to": (t.get("to") or "").lower() or None,178                        "amount": str(val),179                        "decimals": self.native["decimals"],180                    })181        db.insert_transfers(self.conn, rows)182        if rows:183            log.info("%s: %d native %s whale transfers in ..%d",184                     self.chain, len(rows), self.native["symbol"], to)185186    def reorged(self, cursor_block, cursor_hash):187        """True if the block we last indexed is no longer canonical."""188        if not cursor_hash:189            return False190        blk = self.block_header(cursor_block)191        return blk is not None and blk["hash"].lower() != cursor_hash.lower()192193    def links_to_cursor(self, cursor_block, cursor_hash):194        """Parent-hash link: next block's parentHash must be our cursor hash."""195        if not cursor_hash:196            return True197        nxt = self.block_header(cursor_block + 1)198        return nxt is None or nxt["parentHash"].lower() == cursor_hash.lower()199200    # -- main loop -----------------------------------------------------201202    def run(self, stop):203        self.conn = db.connect(self.db_path)204        cur = db.get_cursor(self.conn, self.chain)205        confirmations = self.cfg.get("confirmations", 6)206        while not stop.is_set():207            try:208                head = self.head()209                safe = head - confirmations210                if cur is None:211                    start = max(0, safe - self.cfg.get("start_offset", 300))212                    cur = (start, None, start)213                    db.set_cursor(self.conn, self.chain, start, None)214                    db.set_backfill(self.conn, self.chain, start)215                    log.info("%s: fresh start at block %d (head %d)", self.chain, start, head)216                cursor_block, cursor_hash, backfill_block = cur217                if backfill_block is None:  # DB predates backfill support218                    backfill_block = cursor_block219                    db.set_backfill(self.conn, self.chain, backfill_block)220221                if self.reorged(cursor_block, cursor_hash):222                    to_block = cursor_block - 2 * confirmations223                    log.warning("%s: reorg at %d — rolling back to %d",224                                self.chain, cursor_block, to_block)225                    db.rollback(self.conn, self.chain, to_block)226                    cursor_block, cursor_hash = to_block, None227228                # ---- head tailing (always first priority) ----229                while cursor_block < safe and not stop.is_set():230                    if not self.links_to_cursor(cursor_block, cursor_hash):231                        to_block = cursor_block - 2 * confirmations232                        log.warning("%s: parent-hash mismatch after %d — rolling back to %d",233                                    self.chain, cursor_block, to_block)234                        db.rollback(self.conn, self.chain, to_block)235                        cursor_block, cursor_hash = to_block, None236                        continue237                    end, n = self.try_range(cursor_block + 1, safe, stop)238                    if end < cursor_block + 1:239                        break  # stopped mid-sweep240                    if self.native:241                        try:242                            self.process_native(cursor_block + 1, end)243                        except (RpcError, RuntimeError) as e:244                            log.warning("%s: native capture failed (%s) — "245                                        "continuing", self.chain, e)246                    end_blk = self.block_header(end)247                    cursor_block = end248                    cursor_hash = end_blk["hash"].lower() if end_blk else None249                    db.set_cursor(self.conn, self.chain, cursor_block, cursor_hash, head=head)250                    if n:251                        log.info("%s: %d transfers in blocks ..%d (lag %d)",252                                 self.chain, n, end, head - end)253254                # ---- backfill: one slice per cycle, backwards ----255                if self.backfill_blocks and not stop.is_set():256                    target = max(0, safe - self.backfill_blocks)257                    if backfill_block > target:258                        frm = max(target, backfill_block - self.range)259                        covered, total = frm - 1, 0260                        while covered < backfill_block - 1 and not stop.is_set():261                            end, n = self.try_range(covered + 1, backfill_block - 1, stop)262                            if end <= covered:263                                break264                            covered, total = end, total + n265                        if covered >= backfill_block - 1:  # slice fully covered266                            db.set_backfill(self.conn, self.chain, frm)267                            backfill_block = frm268                            if total:269                                log.info("%s: backfill %d transfers, floor now %d (target %d)",270                                         self.chain, total, frm, target)271272                cur = (cursor_block, cursor_hash, backfill_block)273                db.save_rpc_health(self.conn, self.chain, self.pool.stats(), int(time.time()))274                stop.wait(max(float(self.cfg.get("block_time", 12)), 2.0))275            except Exception as e:276                log.error("%s: %s — retrying in 10s", self.chain, e)277                stop.wait(10)278