T17 · 索引器

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        # DB にないので境界
        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. チェックポイントを分岐点へ戻す
update indexer_checkpoints
   set last_block_number = $fork, last_block_hash = $forkHash
 where chain_id = $1;

commit;

誤り 1：チェックポイントを先にコミットし、後でデータを書く
  途中停止 → 処理済み表示だがデータなし
  → 永久に欠落し、検出もされない

誤り 2：データを書いてから別トランザクションで進捗をコミット
  途中停止 → データはあるが進捗未更新
  → 再起動で再処理 → 冪等書き込みが重複を防ぐ → 無害

正解：データと進捗を同じトランザクションでコミット

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;