sol_shred_sdk/shredstream/
config.rs

1use std::net::SocketAddr;
2use std::time::Duration;
3
4use crate::shred::RawShredConfig;
5
6/// Jito-style ShredStream gRPC source configuration.
7#[derive(Debug, Clone)]
8pub struct JitoShredStreamConfig {
9    pub endpoint: String,
10    pub channel_size: usize,
11}
12
13impl JitoShredStreamConfig {
14    pub fn new(endpoint: impl Into<String>) -> Self {
15        Self {
16            endpoint: endpoint.into(),
17            channel_size: 1_000,
18        }
19    }
20
21    pub fn with_channel_size(mut self, channel_size: usize) -> Self {
22        self.channel_size = channel_size.max(1);
23        self
24    }
25}
26
27/// Input and shred/entry decode mode.
28#[derive(Debug, Clone)]
29pub enum ShredDecodeMode {
30    /// Native raw Solana UDP shreds decoded locally with `solana-ledger`.
31    RawUdp(RawShredConfig),
32    /// Jito-style ShredStream gRPC entries. This source is retained for
33    /// compatibility; it still feeds the same transaction/event parser.
34    JitoGrpc(JitoShredStreamConfig),
35}
36
37impl Default for ShredDecodeMode {
38    fn default() -> Self {
39        Self::RawUdp(RawShredConfig::default())
40    }
41}
42
43impl ShredDecodeMode {
44    pub fn raw_udp(config: RawShredConfig) -> Self {
45        Self::RawUdp(config)
46    }
47
48    pub fn jito_grpc(endpoint: impl Into<String>) -> Self {
49        Self::JitoGrpc(JitoShredStreamConfig::new(endpoint))
50    }
51
52    pub fn name(&self) -> &'static str {
53        match self {
54            Self::RawUdp(_) => "raw-udp",
55            Self::JitoGrpc(_) => "jito-grpc",
56        }
57    }
58}
59
60/// Multi-source ShredStream client configuration.
61#[derive(Debug, Clone)]
62pub struct ShredStreamConfig {
63    pub decode_mode: ShredDecodeMode,
64    pub event_queue_capacity: usize,
65    pub reconnect_delay_ms: u64,
66    /// Maximum restart attempts after a receive-loop error. `0` means forever.
67    pub max_reconnect_attempts: u32,
68}
69
70impl Default for ShredStreamConfig {
71    fn default() -> Self {
72        Self {
73            decode_mode: ShredDecodeMode::default(),
74            event_queue_capacity: 100_000,
75            reconnect_delay_ms: 250,
76            max_reconnect_attempts: 0,
77        }
78    }
79}
80
81impl ShredStreamConfig {
82    pub fn low_latency() -> Self {
83        Self {
84            decode_mode: ShredDecodeMode::RawUdp(RawShredConfig {
85                reassembly_gap_timeout: Duration::from_millis(250),
86                max_tracked_slots: 48,
87                ..RawShredConfig::default()
88            }),
89            event_queue_capacity: 100_000,
90            reconnect_delay_ms: 50,
91            max_reconnect_attempts: 0,
92        }
93    }
94
95    pub fn high_throughput() -> Self {
96        Self {
97            decode_mode: ShredDecodeMode::RawUdp(RawShredConfig {
98                udp_recv_buffer_bytes: 128 * 1024 * 1024,
99                max_tracked_slots: 128,
100                reassembly_gap_timeout: Duration::from_millis(600),
101                ..RawShredConfig::default()
102            }),
103            event_queue_capacity: 500_000,
104            reconnect_delay_ms: 250,
105            max_reconnect_attempts: 0,
106        }
107    }
108
109    pub fn jito_grpc(endpoint: impl Into<String>) -> Self {
110        Self {
111            decode_mode: ShredDecodeMode::jito_grpc(endpoint),
112            event_queue_capacity: 100_000,
113            reconnect_delay_ms: 250,
114            max_reconnect_attempts: 0,
115        }
116    }
117
118    pub fn with_decode_mode(mut self, decode_mode: ShredDecodeMode) -> Self {
119        self.decode_mode = decode_mode;
120        self
121    }
122
123    pub fn with_udp_bind(mut self, udp_bind: SocketAddr) -> Self {
124        if let ShredDecodeMode::RawUdp(raw) = &mut self.decode_mode {
125            raw.udp_bind = udp_bind;
126        }
127        self
128    }
129}
130
131#[cfg(test)]
132mod tests {
133    use super::*;
134
135    #[test]
136    fn default_decode_mode_is_raw_udp() {
137        let config = ShredStreamConfig::default();
138        assert!(matches!(config.decode_mode, ShredDecodeMode::RawUdp(_)));
139    }
140
141    #[test]
142    fn jito_decode_mode_keeps_endpoint() {
143        let config = ShredStreamConfig::jito_grpc("http://127.0.0.1:10000");
144        match config.decode_mode {
145            ShredDecodeMode::JitoGrpc(jito) => {
146                assert_eq!(jito.endpoint, "http://127.0.0.1:10000");
147            }
148            ShredDecodeMode::RawUdp(_) => panic!("expected jito grpc mode"),
149        }
150    }
151
152    #[test]
153    fn udp_bind_only_applies_to_raw_mode() {
154        let udp_bind: SocketAddr = "127.0.0.1:9001".parse().unwrap();
155        let raw = ShredStreamConfig::default().with_udp_bind(udp_bind);
156        match raw.decode_mode {
157            ShredDecodeMode::RawUdp(raw) => assert_eq!(raw.udp_bind, udp_bind),
158            ShredDecodeMode::JitoGrpc(_) => panic!("expected raw udp mode"),
159        }
160
161        let jito = ShredStreamConfig::jito_grpc("http://127.0.0.1:10000").with_udp_bind(udp_bind);
162        match jito.decode_mode {
163            ShredDecodeMode::JitoGrpc(jito) => {
164                assert_eq!(jito.endpoint, "http://127.0.0.1:10000");
165            }
166            ShredDecodeMode::RawUdp(_) => panic!("expected jito grpc mode"),
167        }
168    }
169}
sol_shred_sdk/shredstream/config.rs

sol_shred_sdk/shredstream/
config.rs
