Geyser gRPC - แนวทางปฏิบัติที่ดีที่สุด

คู่มือนี้อธิบายรูปแบบสำคัญสำหรับการสร้างแอปพลิเคชันที่ทนทานและพร้อมใช้งานระดับ production ด้วย Geyser gRPC ทั้งนี้ Solana Stream SDK รองรับ Geyser gRPC

เริ่มต้นใช้งาน

เริ่มจากสิ่งง่าย ๆ: ใช้ slot subscription ก่อน เพื่อทำความคุ้นเคยกับการ stream แล้วจึงเพิ่ม filter ที่ซับซ้อน
ใช้ filter อย่างชาญฉลาด: subscribe เฉพาะข้อมูลที่ต้องใช้ เพื่อลด bandwidth/CPU

rust

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

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

การจัดการการเชื่อมต่อ

reconnect อัตโนมัติ: ให้คาดไว้ว่าจะเกิดปัญหาเครือข่าย และ reconnect ด้วย exponential backoff
การจัดการ ping/pong: เซิร์ฟเวอร์ Yellowstone gRPC จะส่ง ping มา ให้ตอบกลับด้วย pong ทุกครั้ง มิฉะนั้นเซิร์ฟเวอร์อาจปิดการเชื่อมต่อ (สาเหตุที่พบบ่อยของการเชื่อมต่อหลุดเมื่อผ่านไปราว 30 วินาที)

rust

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 หลัง reconnect เพื่อไม่ให้ข้อมูลสูญหาย (ยอมรับข้อมูลซ้ำได้)

rust

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 ออกจากกันด้วย channel เพื่อไม่ให้ network IO ผูกติดกับ business logic และให้มีกลไก backpressure

rust

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 channel: เลือก capacity ตามความเร็วในการประมวลผลและระดับที่ยอมให้ข้อมูลสูญหายได้ (เล็กลง = ใช้ memory น้อยลงแต่ drop มากขึ้น; ใหญ่ขึ้น = ใช้ memory มากขึ้นแต่ drop น้อยลง)

ประสิทธิภาพและความทนทาน

ติดตามความหน่วงของการประมวลผล และบันทึก log เมื่อเกินค่าเกณฑ์
เขียน DB เป็น batch และ flush เมื่อถึงขนาด batch หรือครบช่วงเวลาที่กำหนด
ใช้ async IO สำหรับการเรียกภายนอก และย้ายงานคำนวณหนักไปไว้ที่ worker task/thread
ใช้ subscription request ซ้ำเพื่อลดการจัดสรรหน่วยความจำ

การจัดการ error

แยกประเภทของ error (stream, processing หรือ channel)
แก้ request ที่ไม่ถูกต้องก่อน retry และใช้ exponential backoff ที่คำนึงถึง pause เฉพาะเมื่อ response ระบุให้ retry
บันทึก log/metric ของ update ที่ถูก drop เพื่อติดตามภาวะที่ consumer ช้า

แก้ไขก่อน retry และ backoff ที่คำนึงถึง pause

เก็บ gRPC error แรกไว้ก่อนที่ reconnect loop จะเขียนทับ บันทึก status และ message ของ error นั้น รวมถึง metadata x-erpc-error-code, x-erpc-error-reason, x-erpc-doc-url, x-erpc-retry-after-seconds และ x-erpc-retry-at ที่มีอยู่ การปิด connection ก่อนกำหนดอาจดูเหมือน transport error ทั่วไป จึงอาจไม่มี trailer ครบทุกตัว

ก่อน retry ให้แก้ request ที่ไม่ถูกต้อง, filter shape ที่ถูกปฏิเสธ หรือ IP ที่ลงทะเบียนไม่ตรงกันเสียก่อน สำหรับ pause ของ ERPC ให้รอจนถึง deadline ที่ส่งกลับมา ซึ่งเจาะจงกว่า local backoff ส่วน availability failure และ response อื่นที่ retry ได้ ให้ใช้ exponential backoff แบบมีขอบเขต อย่า retry request เดิมที่ไม่ได้แก้ไขซ้ำ ๆ เพียงเพราะได้ status ที่ไม่ใช่ศูนย์

การใช้ทรัพยากรและความสามารถในการประมวลผล

การคุ้มครองของ Shared ป้องกันไม่ให้ workload หนึ่งใช้ delivery capacity ที่ลูกค้า ERPC รายอื่นต้องใช้ ทั้งนี้ stream อาจช้าลงได้เพราะ effective bandwidth, CPU, memory, consumer queue, การเขียนฐานข้อมูล หรือการเรียกภายนอก ทรัพยากรต้องเพียงพอกับ workload และการที่ receiver ช้าเพียงอย่างเดียวไม่ได้บ่งบอกว่าส่วนใดถูกจำกัด

วัด peak data rate ที่ฝั่ง consumer แปลง MB/s เป็น Mbps ด้วย MB/s × 8 แล้วเผื่อ headroom ไว้เหนือ peak ที่วัดได้ ไม่มีค่าขั้นต่ำตายตัวที่ใช้ได้โดยไม่ต้องวัด ให้งานในส่วนรับข้อมูลสั้นไว้ ส่ง update ต่อไปยัง bounded queue และหาก processing ตามไม่ทัน ให้ขยายระบบหรือลด subscription แม้ bounded queue จะจำกัดการใช้ memory ได้ แต่ก็แก้ปัญหา consumer ที่ช้าอย่างต่อเนื่องไม่ได้

การจัดการข้อมูล

จัดการข้อมูลซ้ำเมื่อใช้ from_slot (cache ที่มีอายุจำกัด หรือ constraint ของ DB)
เลือก commitment ตามกรณีการใช้งาน:
- processed: สำหรับ dashboard ที่ต้องการความเร็วสูงสุด อาจ roll back ได้
- confirmed: ค่าเริ่มต้นที่ดีสำหรับแอปและ indexer ส่วนใหญ่
- finalized: เมื่อต้องการความแน่นอนอย่างสมบูรณ์

Subscription แบบไดนามิก

ใช้ stream แบบสองทิศทาง (bidirectional) เพื่ออัปเดต subscription ระหว่างทำงาน (สลับ filter ทันที ขยายขอบเขตข้อมูล) โดยไม่ต้อง reconnect

รายการตรวจสอบสำหรับ production

✅ reconnect อัตโนมัติด้วย exponential backoff
✅ gap recovery ด้วย from_slot
✅ จัดการ ping/pong (ป้องกันการเชื่อมต่อหลุดที่ 30 วินาที)
✅ แยก ingress และ processing ด้วย bounded channel/backpressure
✅ บันทึก log ของ error และมี metric/การแจ้งเตือน
✅ ติดตามความหน่วงของการประมวลผล
✅ จัดการข้อมูลซ้ำและปรับ filter ให้เหมาะสม
✅ เขียนแบบ batch (หากมีการจัดเก็บข้อมูล)
✅ graceful shutdown และ health check

แหล่งข้อมูลเพิ่มเติม

Solana Stream SDK (รองรับ Geyser gRPC): https://github.com/ValidatorsDAO/solana-stream

ก่อนหน้าทดสอบการเชื่อมต่อ ถัดไปเปรียบเทียบความเร็ว