Розробка користувацького індексатора блокчейну

Розробка кастомізованого блокчейн-індексатора

The Graph вирішує 80% завдань індексації. Решта 20% — це коли потрібна складна агрегація даних на льоту, кросчейн індексація з об'єднанням стану, доступ до даних які не потрапили в события (storage slots, trace calls), субсекундна задержка, або повний контроль над інфраструктурою без vendor lock-in. Саме для цих випадків будується кастомний індексатор.

Архітектурні рішення перед початком

Перш ніж писати код, потрібно відповісти на три питання:

Джерело даних. Logs/события — найдешевший спосіб, але тільки те що контракт явно емітує. Traces (внутрішні транзакції) — потрібен archive node з trace_ namespace або Erigon з --tracing. Storage proofs — для стану який ніколи не емітувався в события. Вибір джерела визначає вимоги до ноди та складність парсингу.

Модель консистентності. Потрібна точна консистентність (обробка реорганізацій) або eventual достатньо? Для фінансових даних реорганізація — не рідкість. На Ethereum mainnet реорги глибиною 1-2 блоки трапляються кілька разів на день. Глибина фіналізації для безпечного підтвердження — 12-15 блоків на PoS Ethereum.

Вимоги до задержки. Реалтайм (< 1 сек від блоку) — потрібен WebSocket + streaming. Аналітика (хвилини/години) — batch processing достатньо.

Компоненти системи

Шар ингестии даних (Data Ingestion Layer)

Три паттерни отримання даних з ноди:

JSON-RPC polling — найпростіший варіант. eth_getLogs з фільтром по адресі та topics, eth_getBlockByNumber. Задержка = polling interval (зазвичай 500мс-2сек). Проблема: при високому RPS нода почне throttle.

WebSocket subscriptions — eth_subscribe("newHeads") та eth_subscribe("logs", filter). Задержка близька до часу блоку. Проблема: при реконнекті можна пропустити блоки, потрібна логіка catch-up.

Direct P2P — підключення до Ethereum P2P мережі через devp2p/libp2p, отримання блоків напрямки без RPC. Мінімальна задержка, але висока складність реалізації. Практично тільки якщо індексатор фізично близько до валідаторів.

Для production рекомендую WebSocket + catch-up механізм:

async def subscribe_blocks(ws_url: str, from_block: int):
    # Спочатку наганяємо до поточного блоку
    current = await rpc.eth_block_number()
    for block_num in range(from_block, current):
        block = await rpc.eth_get_block_by_number(block_num)
        await process_block(block)
    
    # Потім підписуємося на нові
    async with websockets.connect(ws_url) as ws:
        await ws.send(json.dumps({
            "method": "eth_subscribe",
            "params": ["newHeads"]
        }))
        async for message in ws:
            head = json.loads(message)
            await process_new_head(head)

Декодування ABI

Raw log — це масив topics (bytes32) та data (bytes). Декодування через ABI:

import { decodeEventLog, parseAbi } from 'viem';

const abi = parseAbi([
  'event Transfer(address indexed from, address indexed to, uint256 value)'
]);

const decoded = decodeEventLog({
  abi,
  data: log.data,
  topics: log.topics,
});
// decoded.args.from, decoded.args.to, decoded.args.value

Indexed параметри кодуються в topics (topic[0] = keccak256 сигнатури, topic[1..3] = indexed args). Non-indexed — в data через ABI encoding.

Проблеми при декодуванні:

• Anonymous eventos — немає topic[0], matching тільки по адресі
• Proxy контракти — ABI implementation, а не proxy. Потрібно резолвити через implementation() slot (EIP-1967: slot 0x360894...)
• Upgrade события — після апгрейду ABI змінюється, потрібна версіонованість

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

Це найбільш неприємна частина кастомного індексатора. Реорг означає що блоки які ви вже обробили — більш не каноничні.

Паттерн: кожна запис в базі містить block_hash та block_number. При обробці нового блоку перевіряємо батька:

-- Обнаруження реорга
SELECT block_hash, block_number 
FROM processed_blocks 
WHERE block_number = $1 AND block_hash != $2;
-- Якщо є рядки — це реорг

При обнаруженні реорга:

1. Знайти точку розходження (спільний предок)
2. Відкатити всі записи від точки розходження до поточного блоку (DELETE WHERE block_number >= fork_point)
3. Переобробити канонічні блоки

Для цього потрібна атомарна обробка блоку — всі зміни від одного блоку застосовуються в одній транзакції БД з block_hash як ідентифікатором.

BEGIN;
  DELETE FROM events WHERE block_hash = $orphaned_hash;
  DELETE FROM processed_blocks WHERE block_hash = $orphaned_hash;
  -- вставка канонічних даних
  INSERT INTO processed_blocks (block_number, block_hash, ...) VALUES (...);
  INSERT INTO events (...) VALUES (...);
COMMIT;

Шар зберігання

Вибір БД визначається паттернами запитів:

Для більшості DeFi індексаторів: PostgreSQL для основних даних + Redis для hot cache.

API шар

GraphQL через Hasura (автогенерація з PostgreSQL схеми) або вручну через Apollo Server. REST для простих випадків.

Критична оптимізація: DataLoader для батчингу запитів. Якщо GraphQL запит просить transfers { from { balance } } — без DataLoader отримаємо N+1 запитів до БД. DataLoader групує запити за один tick event loop.

Subscriptions для реалтайм даних: PostgreSQL LISTEN/NOTIFY → WebSocket → GraphQL subscription.

Продуктивність та масштабування

Вузькі місця в порядку частоти:

Паралельна обробка блоків. Блоки незалежні якщо немає cross-block стану (зазвичай немає). Worker pool по N потокам, кожен обробляє свій діапазон блоків. Обережно: порядок запису в БД повинен бути детермінованим.

Batch insert. Не INSERT кожну подію окремо. PostgreSQL COPY або INSERT ... VALUES (…),(…),(…) — різниця в 10-50x по throughput.

# Погано: N окремих INSERT
for event in events:
    await db.execute("INSERT INTO events VALUES ($1, $2, ...)", event)

# Добре: один batch INSERT
await db.executemany(
    "INSERT INTO events VALUES ($1, $2, ...)",
    [(e.block, e.tx_hash, ...) for e in events]
)

Індекси vs insert speed. Кожен індекс сповільнює INSERT. Для історичної синхронізації: створити таблицю без індексів, завантажити дані, потім CREATE INDEX CONCURRENTLY. Прискорення в 3-10x порівняно з індексами під час завантаження.

Технологічний стек

Rust (alloy crate) дає найкращу продуктивність для високонавантажених індексаторів: парсинг ABI, десеріалізація блоків, робота з bytes — все це швидше чим у Node.js в 5-20x.

Мониторинг та операційка

Ключові метрики:

• Indexer lag — різниця між latest_block в БД та eth_blockNumber. Алерт при > 10 блоків.
• Reorg count — кількість реорганізацій за період. Різкий ріст = проблеми з нодою або RPC.
• Events per block — аномалії вказують на нестандартну активність або баги в парсингу.
• DB write latency — деградація означає потрібен vacuum, bloat, або sharding.

# Приклад Prometheus alert
- alert: IndexerLagHigh
  expr: eth_latest_block - indexer_processed_block > 50
  for: 2m
  annotations:
    summary: "Indexer is falling behind by {{ $value }} blocks"

Процес роботи

Проектування (3-5 днів). Визначення джерел даних, схеми БД, вимог до реалтайму. Вибір між кастомною розробкою та розширенням існуючих рішень (Ponder, Substreams).

Розробка ядра (5-10 днів). Ингестия + декодування + обробка реорганізацій + зберігання. Це критичний шлях, тестується на історичних даних.

API та інтеграції (3-5 днів). GraphQL/REST схема, subscriptions, документація.

Нагрузочне тестування та оптимізація (2-3 дні). Синхронізація з genesis, нагрузочне тестування API, настройка пулів з'єднань, індексів.

Деплой та мониторинг (1-2 дні). Docker Compose / Kubernetes, настройка алертів, runbook для дежурних.

Усього: 1-2 тижні для індексатора одного протоколу на одній мережі. Мультичейн з агрегацією — ближче до 3-4 тижнів.