Geyser gRPC - Thực hành tốt nhất

Bắt đầu

• Bắt đầu đơn giản: hãy bắt đầu với slot subscription để làm quen với streaming trước khi thêm các filter phức tạp.
• Sử dụng filter một cách khôn ngoan: chỉ subscribe những gì bạn cần để giảm băng thông/CPU. Vote transaction chiếm ~70% lưu lượng — hãy loại bỏ chúng nếu bạn không cần.

SubscribeRequestFilterTransactions {
    vote: Some(false),   // Loại trừ ~70% vote txs
    failed: Some(false), // Loại trừ failed txs
    ..Default::default()
}

SubscribeRequestFilterTransactions {
    vote: Some(false),   // Loại trừ ~70% vote txs
    failed: Some(false), // Loại trừ failed txs
    ..Default::default()
}

Quản lý kết nối

• Tự động kết nối lại: hãy chuẩn bị cho sự cố mạng; kết nối lại với exponential backoff.
• Xử lý ping/pong: Yellowstone gRPC server gửi ping. Luôn phản hồi bằng pong nếu không server có thể đóng kết nối (nguyên nhân phổ biến của việc ngắt kết nối ~30 giây).

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?;
}

• Khôi phục khoảng trống: sử dụng from_slot sau khi kết nối lại để tránh mất dữ liệu (dữ liệu trùng lặp là chấp nhận được).

subscribe_request.from_slot = if tracked_slot > 0 {
    Some(tracked_slot) // Có thể trừ một bộ đệm nhỏ để tránh reorg
} else {
    None
};

subscribe_request.from_slot = if tracked_slot > 0 {
    Some(tracked_slot) // Có thể trừ một bộ đệm nhỏ để tránh reorg
} else {
    None
};

Pattern kiến trúc

• Tách ingress và processing bằng channel để tách biệt network IO khỏi business logic và thêm backpressure.

let (tx, rx) = mpsc::channel::<SubscribeUpdate>(10_000);
// ingress task đọc stream và tx.send(...)
// processing task tiêu thụ rx và xử lý business logic

let (tx, rx) = mpsc::channel::<SubscribeUpdate>(10_000);
// ingress task đọc stream và tx.send(...)
// processing task tiêu thụ rx và xử lý business logic

• Sử dụng bounded channel: chọn dung lượng dựa trên tốc độ xử lý và mức chấp nhận mất dữ liệu (nhỏ hơn = ít bộ nhớ, nhiều drop hơn; lớn hơn = nhiều bộ nhớ, ít drop hơn).

Hiệu suất và khả năng phục hồi

• Giám sát độ trễ xử lý và log khi vượt ngưỡng.
• Ghi DB theo batch; flush theo kích thước batch hoặc khoảng thời gian.
• Sử dụng async IO cho các cuộc gọi bên ngoài; chuyển tính toán nặng sang worker task/thread.
• Tái sử dụng subscription request để giảm cấp phát bộ nhớ.

Xử lý lỗi

• Phân biệt loại lỗi (stream vs processing vs channel).
• Exponential backoff khi kết nối lại (bắt đầu nhỏ, giới hạn ở mức hợp lý).
• Log/metric các update bị drop để theo dõi tình trạng slow-consumer.

Quản lý dữ liệu

• Xử lý trùng lặp khi sử dụng from_slot (cache có giới hạn thời gian hoặc ràng buộc DB).
• Chọn commitment theo trường hợp sử dụng:
  • processed: nhanh nhất cho dashboard, có thể bị rollback
  • confirmed: mặc định tốt cho hầu hết ứng dụng/indexer
  • finalized: khi cần đảm bảo tuyệt đối

Subscription động

Danh sách kiểm tra production

• ✅ Tự động kết nối lại với exponential backoff
• ✅ Khôi phục khoảng trống sử dụng from_slot
• ✅ Xử lý ping/pong (tránh ngắt kết nối 30 giây)
• ✅ Tách ingress và processing với bounded channel/backpressure
• ✅ Log lỗi và metric/alerting
• ✅ Theo dõi độ trễ xử lý
• ✅ Xử lý trùng lặp và tối ưu filter
• ✅ Ghi theo batch (nếu lưu trữ)
• ✅ Tắt máy an toàn và health check

Tài nguyên bổ sung

• Solana Stream SDK (hỗ trợ Geyser gRPC): https://github.com/ValidatorsDAO/solana-stream