Разработка системы индексации блокчейн-событий

Разработка системы индексации блокчейн-событий

Проблема появляется тогда, когда вам нужно ответить на вопрос «покажи все транзакции этого пользователя за последние 30 дней» или «какой общий объём торгов прошёл через наш DEX». Прямые вызовы к ноде через eth_getLogs с диапазоном блоков — это не архитектура, это костыль. На архивной ноде с блоком от 0 такой запрос займёт минуты, на публичном RPC — скорее всего упадёт с timeout или block range too large.

Правильная система индексации — это pipeline: нода → event listener → parser → база данных → API. Каждый компонент с независимым масштабированием и гарантией exactly-once или at-least-once семантики.

Архитектура event pipeline

Слой получения событий

Есть три подхода к получению событий из ноды, и у каждого своя цена надёжности:

Polling через eth_getLogs — самый простой и самый надёжный при правильной реализации. Воркер периодически запрашивает события за диапазон блоков, сохраняет lastIndexedBlock, при перезапуске продолжает с последнего обработанного блока.

interface IndexerState {
    lastIndexedBlock: number
    lastIndexedBlockHash: string // для детекции реорга
}

async function pollEvents(
    provider: ethers.JsonRpcProvider,
    contracts: ContractConfig[],
    state: IndexerState
): Promise<ProcessedEvent[]> {
    const currentBlock = await provider.getBlockNumber()
    const toBlock = Math.min(state.lastIndexedBlock + BATCH_SIZE, currentBlock - CONFIRMATIONS)

    if (toBlock <= state.lastIndexedBlock) return []

    // Верификация непрерывности chain
    const lastBlock = await provider.getBlock(state.lastIndexedBlock)
    if (lastBlock?.hash !== state.lastIndexedBlockHash) {
        throw new ReorgDetectedError(state.lastIndexedBlock)
    }

    const logs = await provider.getLogs({
        address: contracts.map(c => c.address),
        topics: [contracts.flatMap(c => c.topics)],
        fromBlock: state.lastIndexedBlock + 1,
        toBlock,
    })

    return logs.map(parseLog)
}

WebSocket subscriptions — низкая latency (новые события приходят немедленно), но WebSocket соединения нестабильны. Нужен automatic reconnect с backoff и синхронизация пропущенных блоков при переподключении:

async function subscribeWithFallback(provider: ethers.WebSocketProvider) {
    const filter = { address: CONTRACT, topics: [EVENT_TOPIC] }

    provider.on(filter, async (log) => {
        await processLog(log)
    })

    provider.websocket.on('close', async () => {
        console.log('WebSocket closed, syncing missed blocks...')
        await syncMissedBlocks(lastProcessedBlock)
        reconnect() // exponential backoff
    })
}

Firehose / StreamingFast — enterprise-уровень. Бинарный стриминг блоков со всем содержимым, минимальная latency, встроенная обработка реоргов. Используется как источник данных для The Graph protocol nodes. Существенно сложнее в настройке.

Обработка реорганизаций

Реорги — наиболее частый источник багов в индексерах. На Ethereum finality через PoS — после двух эпох (~12-13 минут) блок финален. На PoW сетях и молодых EVM цепочках (BSC, Polygon) реорги глубиной 3-5 блоков — норма.

Стратегия: индексировать с задержкой N блоков (safe confirmations), хранить hash каждого проиндексированного блока, при несоответствии — откат до последнего консистентного состояния.

-- Таблица состояния индексера
CREATE TABLE indexer_blocks (
    block_number    BIGINT PRIMARY KEY,
    block_hash      VARCHAR(66) NOT NULL,
    indexed_at      TIMESTAMPTZ DEFAULT NOW()
);

-- Событие с привязкой к блоку для возможности rollback
CREATE TABLE indexed_events (
    id              BIGSERIAL PRIMARY KEY,
    block_number    BIGINT NOT NULL REFERENCES indexer_blocks(block_number),
    log_index       INT NOT NULL,
    tx_hash         VARCHAR(66) NOT NULL,
    contract_addr   VARCHAR(42) NOT NULL,
    event_name      VARCHAR(100) NOT NULL,
    decoded_data    JSONB NOT NULL,
    UNIQUE(tx_hash, log_index)
);

При детекции реорга — DELETE FROM indexed_events WHERE block_number >= reorg_depth, затем переиндексация с реорговой точки.

Декодирование событий

ABI-декодирование событий — тривиальная задача для ethers.js или viem, но есть нюансы.

Indexed vs non-indexed параметры: indexed параметры попадают в topics, non-indexed — в data. Событие с 3 indexed параметрами + event signature занимает 4 topics. Декодирование topics для non-primitive типов (structs, dynamic arrays) невозможно — они хешируются через keccak256 и теряют исходные данные.

import { decodeEventLog } from 'viem'

function parseSwapEvent(log: Log, abi: Abi): SwapEvent {
    const decoded = decodeEventLog({
        abi,
        eventName: 'Swap',
        data: log.data,
        topics: log.topics,
    })

    return {
        blockNumber: log.blockNumber,
        txHash: log.transactionHash,
        sender: decoded.args.sender,
        recipient: decoded.args.recipient,
        amount0: decoded.args.amount0,
        amount1: decoded.args.amount1,
        sqrtPriceX96: decoded.args.sqrtPriceX96,
        liquidity: decoded.args.liquidity,
        tick: decoded.args.tick,
    }
}

Анонимные события (без event selector в topic0) — редкость, но встречается в старых контрактах. Декодирование требует знания структуры данных без ABI.

Proxy контракты: события эмитятся с адреса proxy, но ABI реализации. Нужно резолвить implementation address через EIP-1967 storage slot: 0x360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc.

База данных и производительность

Для хранения событий PostgreSQL — стандарт де-факто. Ключевые решения по схеме:

Партиционирование по block_number: для контрактов с высоким объёмом событий (Uniswap V3, крупные ERC-20) таблица быстро вырастает до сотен миллионов строк. Партиционирование по диапазону блоков даёт плановые сканы вместо seq scan:

CREATE TABLE swap_events (
    block_number BIGINT NOT NULL,
    -- ...остальные поля
) PARTITION BY RANGE (block_number);

CREATE TABLE swap_events_0_5m    PARTITION OF swap_events FOR VALUES FROM (0) TO (5000000);
CREATE TABLE swap_events_5m_10m  PARTITION OF swap_events FOR VALUES FROM (5000000) TO (10000000);
-- и т.д.

JSONB для decoded_data: хранение декодированных данных в JSONB позволяет добавлять новые типы событий без миграций схемы. Индексы GIN на часто запрашиваемые поля внутри JSONB. Для критичных полей — вынесение в отдельные типизированные колонки.

TimescaleDB — расширение для PostgreSQL, дающее автоматическое time-based партиционирование (hypertables), сжатие старых данных и continuous aggregates для OHLCV/метрик без фоновых джобов:

-- Continuous aggregate: volume per hour per pool
CREATE MATERIALIZED VIEW pool_hourly_volume
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', event_timestamp) AS hour,
    pool_address,
    SUM(amount_usd) AS volume_usd,
    COUNT(*) AS tx_count
FROM swap_events
GROUP BY 1, 2;

SELECT add_continuous_aggregate_policy('pool_hourly_volume',
    start_offset => INTERVAL '3 hours',
    end_offset   => INTERVAL '1 hour',
    schedule_interval => INTERVAL '1 hour'
);

Мониторинг и алерты

Индексер должен иметь метрики: indexer_lag_blocks (разрыв между head и последним проиндексированным блоком), events_per_second, reorg_count. Алерт если indexer_lag_blocks > 50 — индексер отстаёт или завис.

Деплой через Docker Compose / Kubernetes с health check endpoint, который возвращает 503 если lag превышает порог. Это позволяет load balancer-у исключить нездоровый инстанс.

Типичный стек: Go или Rust для воркера (производительность), PostgreSQL / TimescaleDB для хранения, Redis для state и очереди, Grafana + Prometheus для мониторинга.