T17 · Indexer

loop:
  head = rpc.blockNumber()
  cp   = db.readCheckpoint(chainId)          # 高度 + 区块哈希

  if not isStillOnChain(cp):                 # 1. 前进之前先确认自己还在这条链上
      rollbackTo(findForkPoint(cp))          #    一个事务
      continue

  from = cp.number + 1                       # 2. 这一轮的范围
  to   = min(cp.number + batchSize, head)
  if from > to: sleep(pollInterval); continue

  for n in from..to:                         # 3. 每个区块一个事务
      block = rpc.getBlockHeader(n)
      if block.parentHash != lastHash: break # 拉取途中链头变了，下一轮重判
      logs = rpc.getLogs(n, n, filters)
      db.transaction:
          insertBlock(block)
          insertLogs(logs)                   # 幂等，靠主键
          enqueueOutbox(logs, n)             # 副作用先入库，不直接发
          writeCheckpoint(n, block.hash)     # 和数据同一个事务
      lastHash = block.hash

findForkPoint(cp):
    n = cp.number
    while n > cp.number - maxRollbackDepth:
        onchain = rpc.getBlockHeader(n).hash
        local   = db.blockHash(chainId, n)
        if local is null:  return n        # 库里没有，说明这里就是边界
        if onchain == local: return n      # 第一个对得上的高度 = 分叉点
        n = n - 1
    raise DeepReorgError(n)

begin;

-- 0. 先把受影响的地址收集出来。必须在删除之前，删完就查不到了。
create temp table touched_holders on commit drop as
select distinct holder from (
  select from_addr as holder from transfer_events
   where chain_id = $1 and block_number > $fork
  union
  select to_addr   as holder from transfer_events
   where chain_id = $1 and block_number > $fork
) x;

-- 1. 事实表与区块表：按高度删
delete from transfer_events where chain_id = $1 and block_number > $fork;
delete from blocks           where chain_id = $1 and block_number > $fork;

-- 2. 还没发出去的副作用，跟着一起删掉
delete from outbox
 where chain_id = $1 and block_number > $fork and delivered_at is null;

-- 3. 派生表：受影响的 holder 清零，从剩下的事实表重算
delete from token_balances b
 where b.chain_id = $1
   and exists (select 1 from touched_holders t where t.holder = b.holder);

insert into token_balances (chain_id, contract, holder, balance, updated_at_block)
select ... from transfer_events
 where chain_id = $1 and to_addr in (select holder from touched_holders)
    or chain_id = $1 and from_addr in (select holder from touched_holders)
group by chain_id, contract, holder;

-- 4. checkpoint 退回分叉点
update indexer_checkpoints
   set last_block_number = $fork, last_block_hash = $forkHash
 where chain_id = $1;

commit;

错法一：先提交 checkpoint，再写数据
  崩溃点落在中间 → checkpoint 说「处理过了」，但数据没写
  → 永久漏数据，而且没有任何东西会发现

错法二：先写数据，再提交 checkpoint（两个事务）
  崩溃点落在中间 → 数据写了，checkpoint 没更新
  → 重启后重做这一段 → 幂等写入把它挡掉 → 无害

正确：数据和 checkpoint 在同一个事务里提交

create table backfill_ranges (
  chain_id   integer not null,
  from_block bigint  not null,
  to_block   bigint  not null,
  status     text    not null,          -- pending / running / done / failed
  attempts   integer not null default 0,
  primary key (chain_id, from_block)
);

create table outbox (
  id            bigserial   primary key,
  chain_id      integer     not null,
  block_number  bigint      not null,   -- 它属于哪个高度，用来判断够不够深
  event_key     text        not null,   -- 链上坐标，投递方去重用
  payload       jsonb       not null,
  delivered_at  timestamptz,
  unique (chain_id, event_key)
);

select * from outbox
 where chain_id = $1
   and delivered_at is null
   and block_number <= $safeHeight        -- 只发够深的
 order by block_number, id
 limit 500;