Geyser gRPC - Best Practices

Начало работы

• Начинайте с простого: сначала подпишитесь на slot updates, чтобы освоить streaming, и только потом добавляйте сложные filters.
• Используйте filters осмысленно: подписывайтесь только на действительно нужные данные, чтобы снизить расход bandwidth/CPU. Vote transactions составляют примерно 70% трафика - исключайте их, если они вам не нужны.

SubscribeRequestFilterTransactions {
    vote: Some(false),   // Exclude ~70% vote txs
    failed: Some(false), // Exclude failed txs
    ..Default::default()
}

SubscribeRequestFilterTransactions {
    vote: Some(false),   // Exclude ~70% vote txs
    failed: Some(false), // Exclude failed txs
    ..Default::default()
}

Управление соединением

• Автоматический reconnection: закладывайтесь на сетевые проблемы и переподключайтесь с exponential backoff.
• Обработка ping/pong: Yellowstone gRPC servers отправляют ping. Всегда отвечайте pong, иначе сервер может закрыть соединение (это частая причина disconnect примерно через 30 секунд).

if matches!(update.update_oneof, Some(UpdateOneof::Ping(_))) {
    subscribe_tx.send(SubscribeRequest {
        ping: Some(SubscribeRequestPing { id: 1 }),
        ..Default::default()
    }).await?;
}

if matches!(update.update_oneof, Some(UpdateOneof::Ping(_))) {
    subscribe_tx.send(SubscribeRequest {
        ping: Some(SubscribeRequestPing { id: 1 }),
        ..Default::default()
    }).await?;
}

• Gap recovery: после переподключения используйте from_slot, чтобы избежать потери данных (duplicates допустимы).

subscribe_request.from_slot = if tracked_slot > 0 {
    Some(tracked_slot) // Optionally subtract a small buffer to dodge reorgs
} else {
    None
};

subscribe_request.from_slot = if tracked_slot > 0 {
    Some(tracked_slot) // Optionally subtract a small buffer to dodge reorgs
} else {
    None
};

Архитектурные паттерны

• Разделяйте ingress и processing через channels, чтобы отделить network IO от business logic и добавить backpressure.

let (tx, rx) = mpsc::channel::<SubscribeUpdate>(10_000);
// ingress task reads stream and tx.send(...)
// processing task consumes rx and does business logic

let (tx, rx) = mpsc::channel::<SubscribeUpdate>(10_000);
// ingress task reads stream and tx.send(...)
// processing task consumes rx and does business logic

• Используйте bounded channels: подбирайте capacity исходя из скорости обработки и допустимых потерь данных (меньше = меньше памяти, больше потерь; больше = больше памяти, меньше потерь).

Производительность и устойчивость

• Отслеживайте processing latency и логируйте превышение порогов.
• Батчируйте DB writes; выполняйте flush по размеру batch или по интервалу.
• Для внешних вызовов используйте async IO; тяжелые вычисления выносите в worker tasks/threads.
• Переиспользуйте subscription requests, чтобы уменьшить количество allocations.

Обработка ошибок

• Разделяйте типы ошибок (stream vs processing vs channel).
• Используйте exponential backoff при reconnect (начинайте с малого и ограничивайте разумным максимумом).
• Логируйте и отправляйте в metrics dropped updates, чтобы отслеживать slow-consumer conditions.

Работа с данными

• Обрабатывайте duplicates при использовании from_slot (time-bounded cache или DB constraints).
• Выбирайте commitment под вашу задачу:
  • processed: самые быстрые dashboards, возможен rollback
  • confirmed: хороший default для большинства apps/indexers
  • finalized: когда нужна максимальная определенность

Динамические subscriptions

Production checklist

• ✅ Auto reconnection с exponential backoff
• ✅ Gap recovery через from_slot
• ✅ Обработка ping/pong (чтобы избежать disconnect через 30 секунд)
• ✅ Разделение ingress и processing с bounded channels/backpressure
• ✅ Error logging и metrics/alerting
• ✅ Отслеживание processing latency
• ✅ Обработка duplicates и оптимизация filters
• ✅ Batch writes (если вы сохраняете данные)
• ✅ Graceful shutdown и health checks

Дополнительные ресурсы

• Solana Stream SDK (поддержка Geyser gRPC): https://github.com/ValidatorsDAO/solana-stream