Geyser gRPC - Best Practices

Erste Schritte

• Starten Sie einfach: beginnen Sie mit Slot-Abonnements, um sich mit Streaming vertraut zu machen, bevor Sie komplexe Filter hinzufügen.
• Verwenden Sie Filter sinnvoll: abonnieren Sie nur, was Sie benötigen, um Bandbreite und CPU-Last zu reduzieren. Vote-Transaktionen sind ~70% des Verkehrs – verwerfen Sie sie, wenn Sie sie nicht brauchen.

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()
}

Verbindungsmanagement

• Automatische Wiederverbindung: Rechnen Sie mit Netzwerkproblemen und verbinden Sie mit exponentiellem Backoff.
• Ping/Pong-Handling: Yellowstone gRPC Server senden ping. Beantworten Sie immer mit Pong andernfalls kann der Server die Verbindung schließen (häufige Ursache von ~30-Sekunden-Trennungen).

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: Verwendung from_slot nach dem Wiederverbinden, um Datenverlust zu vermeiden (Duplikate sind akzeptabel).

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

Architekturmuster

• Ingress und Verarbeitung trennen mit Kanälen, um das Netzwerk-I/O von der Geschäftslogik zu entkoppeln und Backpressure einzubauen.

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

• Verwenden Sie begrenzte Kanäle: Wählen Sie die Kapazität basierend auf Verarbeitungsgeschwindigkeit und Datenverlusttoleranz (kleiner = weniger Speicher, mehr verworfene Updates; größer = mehr Speicher, weniger verworfene Updates).

Leistung und Widerstandsfähigkeit

• Überwachen Sie die Verarbeitungslatenz und loggen Sie Überschreitungen von Schwellenwerten.
• DB-Schreibvorgänge bündeln; nach Batchgröße oder Intervall flushen.
• Verwenden Sie asynchrones I/O für externe Anrufe; lagern Sie rechenintensive Aufgaben an Worker-Tasks/-Threads.
• Subscription-Requests wiederverwenden zur Verringerung der Zuweisungen.

Fehlerbehebung

• Fehlertypen unterscheiden (stream vs processing vs channel).
• Exponentieller Backoff beim Reconnect (klein beginnen, mit einer sinnvollen Obergrenze arbeiten.
• Verworfene Updates in Logs/Metriken erfassen, um langsame Verbraucherbedingungen zu verfolgen.

Datenverwaltung

• Duplikate behandeln bei Verwendung from_slot (zeitgebundener Cache oder DB-Constraints).
• Commitment je nach Anwendungsfall wählen:
• processed: schnellste Dashboards, kann zurückrollen
• confirmed: guter Standard für die meisten Apps/Indexer
• finalized: wenn maximale Sicherheit erforderlich ist

Dynamische Abonnements

Produktionsprüfliste

• ✅ Auto-Reconnection mit exponentiellem Backoff
• ✅ Gap-Recovery mit from_slot
• ✅ Ping/Pong-Handling (verhindert 30-Sekunden-Trennungen)
• Ingress und Verarbeitung trennen mit begrenzten Kanälen/Backpressure
• Fehlerbehandlung und Metriken/Alerting
• ✅ Verarbeitungslatenzverfolgung
• ✅ Duplikatbehandlung und Filteroptimierung
• ✅ Batch-Schreibvorgänge (bei Persistenz)
• ✅ Graceful Shutdown und Gesundheitskontrollen

Weitere Ressourcen

• Solana Stream SDK (Geyser gRPC-Unterstützung): https://github.com/ValidatorsDAO/solana-stream