arete_server/
config.rs

1use std::net::SocketAddr;
2use std::time::Duration;
3
4use anyhow::{bail, Context, Result};
5use ipnet::IpNet;
6
7pub use crate::health::HealthConfig;
8pub use crate::http_health::HttpHealthConfig;
9pub use crate::http_server::HttpServerConfig;
10
11/// Runtime capabilities selected for one server process.
12#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
13pub struct RuntimePlan {
14    pub health: bool,
15    pub chain_reads: bool,
16    pub program_reads: bool,
17    pub stack_queries: bool,
18    pub transactions: bool,
19    pub websocket: bool,
20    pub live_runtime: bool,
21}
22
23impl RuntimePlan {
24    /// Health, chain reads, and fixed transaction routes only.
25    pub const fn solana_gateway() -> Self {
26        Self {
27            health: true,
28            chain_reads: true,
29            program_reads: false,
30            stack_queries: false,
31            transactions: true,
32            websocket: false,
33            live_runtime: false,
34        }
35    }
36
37    pub fn http() -> Self {
38        Self {
39            health: true,
40            chain_reads: true,
41            program_reads: true,
42            stack_queries: true,
43            ..Self::default()
44        }
45    }
46
47    pub fn program_reads() -> Self {
48        Self {
49            health: true,
50            program_reads: true,
51            ..Self::default()
52        }
53    }
54
55    pub fn live_runtime_enabled(self) -> bool {
56        self.live_runtime || self.websocket
57    }
58}
59
60/// Explicit configuration for the fixed transaction relay routes.
61#[derive(Clone, Debug)]
62pub struct TransactionConfig {
63    pub enabled: bool,
64    pub rpc_url: Option<String>,
65    /// Development-only: allow relay requests when no auth plugin is configured.
66    pub allow_unauthenticated: bool,
67    pub max_body_bytes: usize,
68    pub max_transaction_bytes: usize,
69    pub inspect_timeout: Duration,
70    pub send_timeout: Duration,
71    pub status_timeout: Duration,
72    pub inspect_concurrency: usize,
73    pub send_concurrency: usize,
74    pub inspect_requests_per_minute: u32,
75    pub send_requests_per_minute: u32,
76    pub status_requests_per_minute: u32,
77    pub trusted_proxy_cidrs: Vec<IpNet>,
78    pub usage_enabled: bool,
79    pub usage_endpoint: Option<String>,
80    pub usage_token: Option<String>,
81    pub usage_spool_capacity: usize,
82}
83
84impl Default for TransactionConfig {
85    fn default() -> Self {
86        Self {
87            enabled: false,
88            rpc_url: None,
89            allow_unauthenticated: false,
90            // Envelope for any relay body, not the transaction itself — `max_transaction_bytes`
91            // below still bounds a submit at Solana's packet limit. Sized for the largest
92            // legitimate body, which is a 256-signature status batch at roughly 23 KiB; 4 KiB
93            // admitted only about 44 signatures and made the advertised batch unreachable.
94            max_body_bytes: 32 * 1024,
95            // SIMD-0296 raised the packet limit to 4096 bytes for v1 transactions. legacy and v0
96            // stay at 1232, enforced by the cluster; lower this to cap submits below the network.
97            max_transaction_bytes: 4096,
98            inspect_timeout: Duration::from_secs(10),
99            send_timeout: Duration::from_secs(15),
100            status_timeout: Duration::from_secs(10),
101            inspect_concurrency: 64,
102            send_concurrency: 16,
103            inspect_requests_per_minute: 600,
104            send_requests_per_minute: 60,
105            status_requests_per_minute: 600,
106            trusted_proxy_cidrs: Vec::new(),
107            usage_enabled: false,
108            usage_endpoint: None,
109            usage_token: None,
110            usage_spool_capacity: 1_000,
111        }
112    }
113}
114
115impl TransactionConfig {
116    /// Load transaction settings. Routes remain disabled unless explicitly enabled.
117    pub fn from_env() -> Result<Self> {
118        let mut config = Self::default();
119        config.enabled = env_bool("ARETE_TRANSACTIONS_ENABLED")?.unwrap_or(false);
120        config.rpc_url =
121            first_nonempty(&["ARETE_TRANSACTION_RPC_URL", "SOLANA_RPC_URL", "RPC_URL"]);
122        config.allow_unauthenticated =
123            env_bool("ARETE_TRANSACTIONS_ALLOW_UNAUTHENTICATED")?.unwrap_or(false);
124        config.max_body_bytes =
125            env_parse("ARETE_TRANSACTION_MAX_BODY_BYTES")?.unwrap_or(config.max_body_bytes);
126        config.max_transaction_bytes =
127            env_parse("ARETE_TRANSACTION_MAX_BYTES")?.unwrap_or(config.max_transaction_bytes);
128        config.inspect_timeout = Duration::from_millis(
129            env_parse("ARETE_TRANSACTION_INSPECT_TIMEOUT_MS")?
130                .unwrap_or(config.inspect_timeout.as_millis() as u64),
131        );
132        config.send_timeout = Duration::from_millis(
133            env_parse("ARETE_TRANSACTION_SEND_TIMEOUT_MS")?
134                .unwrap_or(config.send_timeout.as_millis() as u64),
135        );
136        config.status_timeout = Duration::from_millis(
137            env_parse("ARETE_TRANSACTION_STATUS_TIMEOUT_MS")?
138                .unwrap_or(config.status_timeout.as_millis() as u64),
139        );
140        config.inspect_concurrency = env_parse("ARETE_TRANSACTION_INSPECT_CONCURRENCY")?
141            .unwrap_or(config.inspect_concurrency);
142        config.send_concurrency =
143            env_parse("ARETE_TRANSACTION_SEND_CONCURRENCY")?.unwrap_or(config.send_concurrency);
144        config.inspect_requests_per_minute =
145            env_parse("ARETE_TRANSACTION_INSPECT_REQUESTS_PER_MINUTE")?
146                .unwrap_or(config.inspect_requests_per_minute);
147        config.send_requests_per_minute = env_parse("ARETE_TRANSACTION_SEND_REQUESTS_PER_MINUTE")?
148            .unwrap_or(config.send_requests_per_minute);
149        config.status_requests_per_minute =
150            env_parse("ARETE_TRANSACTION_STATUS_REQUESTS_PER_MINUTE")?
151                .unwrap_or(config.status_requests_per_minute);
152        config.trusted_proxy_cidrs = std::env::var("ARETE_TRUSTED_PROXY_CIDRS")
153            .ok()
154            .filter(|value| !value.trim().is_empty())
155            .map(|value| {
156                value
157                    .split(',')
158                    .map(|cidr| cidr.trim().parse().context("invalid trusted proxy CIDR"))
159                    .collect::<Result<Vec<_>>>()
160            })
161            .transpose()?
162            .unwrap_or_default();
163        config.usage_enabled = env_bool("ARETE_TRANSACTION_USAGE_ENABLED")?.unwrap_or(false);
164        config.usage_endpoint = first_nonempty(&["ARETE_TRANSACTION_USAGE_ENDPOINT"]);
165        config.usage_token = first_nonempty(&["ARETE_TRANSACTION_USAGE_TOKEN"]);
166        config.usage_spool_capacity = env_parse("ARETE_TRANSACTION_USAGE_SPOOL_CAPACITY")?
167            .unwrap_or(config.usage_spool_capacity);
168        config.validate()?;
169        Ok(config)
170    }
171
172    pub fn validate(&self) -> Result<()> {
173        if self.enabled && self.rpc_url.as_deref().unwrap_or_default().is_empty() {
174            bail!("transactions are enabled but no transaction RPC URL is configured");
175        }
176        if self.max_body_bytes == 0
177            || self.max_transaction_bytes == 0
178            || self.inspect_concurrency == 0
179            || self.send_concurrency == 0
180        {
181            bail!("transaction size and concurrency limits must be greater than zero");
182        }
183        if self.usage_enabled && (self.usage_endpoint.is_none() || self.usage_token.is_none()) {
184            bail!("transaction usage requires an endpoint and token");
185        }
186        Ok(())
187    }
188}
189
190fn first_nonempty(keys: &[&str]) -> Option<String> {
191    keys.iter()
192        .find_map(|key| std::env::var(key).ok())
193        .filter(|value| !value.trim().is_empty())
194}
195
196pub(crate) fn env_parse<T>(key: &str) -> Result<Option<T>>
197where
198    T: std::str::FromStr,
199    T::Err: std::error::Error + Send + Sync + 'static,
200{
201    std::env::var(key)
202        .ok()
203        .map(|value| value.parse().with_context(|| format!("invalid {key}")))
204        .transpose()
205}
206
207pub(crate) fn env_bool(key: &str) -> Result<Option<bool>> {
208    let Some(value) = std::env::var(key).ok() else {
209        return Ok(None);
210    };
211    match value.to_ascii_lowercase().as_str() {
212        "true" | "1" | "yes" => Ok(Some(true)),
213        "false" | "0" | "no" => Ok(Some(false)),
214        _ => bail!("invalid {key}; expected true or false"),
215    }
216}
217
218/// Configuration for gRPC stream reconnection with exponential backoff
219#[derive(Clone, Debug)]
220pub struct ReconnectionConfig {
221    /// Initial delay before first reconnection attempt
222    pub initial_delay: Duration,
223    /// Maximum delay between reconnection attempts
224    pub max_delay: Duration,
225    /// Maximum number of reconnection attempts (None = infinite)
226    pub max_attempts: Option<u32>,
227    /// Multiplier for exponential backoff (typically 2.0)
228    pub backoff_multiplier: f64,
229    /// HTTP/2 keep-alive interval to prevent silent disconnects
230    pub http2_keep_alive_interval: Option<Duration>,
231    /// Consecutive short-lived connections before a runtime with no snapshot
232    /// to replay gives up on its checkpoint and subscribes live.
233    ///
234    /// The fallback trades data for availability: every slot between the
235    /// checkpoint and the live tip is lost. `None` refuses that trade and
236    /// keeps retrying from the checkpoint, which is what a recorder wants.
237    pub live_fallback_attempts: Option<u32>,
238}
239
240impl Default for ReconnectionConfig {
241    fn default() -> Self {
242        Self {
243            initial_delay: Duration::from_millis(100),
244            max_delay: Duration::from_secs(60),
245            max_attempts: None, // Infinite retries by default
246            backoff_multiplier: 2.0,
247            http2_keep_alive_interval: Some(Duration::from_secs(30)),
248            live_fallback_attempts: Some(3),
249        }
250    }
251}
252
253impl ReconnectionConfig {
254    pub fn new() -> Self {
255        Self::default()
256    }
257
258    pub fn with_initial_delay(mut self, delay: Duration) -> Self {
259        self.initial_delay = delay;
260        self
261    }
262
263    pub fn with_max_delay(mut self, delay: Duration) -> Self {
264        self.max_delay = delay;
265        self
266    }
267
268    pub fn with_max_attempts(mut self, attempts: u32) -> Self {
269        self.max_attempts = Some(attempts);
270        self
271    }
272
273    pub fn with_backoff_multiplier(mut self, multiplier: f64) -> Self {
274        self.backoff_multiplier = multiplier;
275        self
276    }
277
278    pub fn with_http2_keep_alive_interval(mut self, interval: Duration) -> Self {
279        self.http2_keep_alive_interval = Some(interval);
280        self
281    }
282
283    /// Never abandon the resume checkpoint, even if the provider keeps
284    /// refusing it. Ingestion stalls rather than silently skipping slots.
285    pub fn fail_closed(mut self) -> Self {
286        self.live_fallback_attempts = None;
287        self
288    }
289
290    /// Calculate the next backoff duration given the current one
291    pub fn next_backoff(&self, current: Duration) -> Duration {
292        let next_secs = current.as_secs_f64() * self.backoff_multiplier;
293        let capped_secs = next_secs.min(self.max_delay.as_secs_f64());
294        Duration::from_secs_f64(capped_secs)
295    }
296}
297
298/// WebSocket server configuration
299#[derive(Clone, Debug)]
300pub struct WebSocketConfig {
301    pub bind_address: SocketAddr,
302}
303
304/// Buffering and latest-state delivery controls for WebSocket subscriptions.
305///
306/// These settings are separate from [`WebSocketConfig`] because embedded hosts
307/// serve accepted connections without binding a listener of their own.
308#[derive(Clone, Debug, PartialEq, Eq)]
309pub struct WebSocketDeliveryConfig {
310    /// Source frames retained for each active list-view broadcast bus.
311    pub list_bus_capacity: usize,
312    /// Default fixed flush cadence for latest-state collection subscriptions.
313    /// `None` preserves immediate delivery unless a view overrides it.
314    pub collection_coalesce_ms: Option<u64>,
315}
316
317impl Default for WebSocketDeliveryConfig {
318    fn default() -> Self {
319        Self {
320            // Shared per active view rather than allocated per client.
321            list_bus_capacity: 8 * 1024,
322            collection_coalesce_ms: None,
323        }
324    }
325}
326
327impl WebSocketDeliveryConfig {
328    pub fn from_env() -> anyhow::Result<Self> {
329        let mut config = Self::default();
330        if let Ok(value) = std::env::var("ARETE_WS_LIST_BUS_CAPACITY") {
331            config.list_bus_capacity = value.parse().map_err(|_| {
332                anyhow::anyhow!("ARETE_WS_LIST_BUS_CAPACITY must be a positive integer")
333            })?;
334        }
335        if let Ok(value) = std::env::var("ARETE_WS_COLLECTION_COALESCE_MS") {
336            let milliseconds: u64 = value.parse().map_err(|_| {
337                anyhow::anyhow!("ARETE_WS_COLLECTION_COALESCE_MS must be a non-negative integer")
338            })?;
339            config.collection_coalesce_ms = (milliseconds > 0).then_some(milliseconds);
340        }
341        config.validate()?;
342        Ok(config)
343    }
344
345    pub fn validate(&self) -> anyhow::Result<()> {
346        anyhow::ensure!(
347            self.list_bus_capacity > 0,
348            "WebSocket list bus capacity must be greater than zero"
349        );
350        anyhow::ensure!(
351            self.collection_coalesce_ms
352                .is_none_or(|milliseconds| milliseconds <= 60_000),
353            "WebSocket collection coalescing must not exceed 60000ms"
354        );
355        Ok(())
356    }
357}
358
359/// The only wire protocol accepted by the WebSocket server.
360pub const WEBSOCKET_PROTOCOL_VERSION: u8 = crate::websocket::subscription::PROTOCOL_VERSION;
361
362impl Default for WebSocketConfig {
363    fn default() -> Self {
364        Self {
365            bind_address: "[::]:8877".parse().expect("valid socket address"),
366        }
367    }
368}
369
370impl WebSocketConfig {
371    pub fn new(bind_address: impl Into<SocketAddr>) -> Self {
372        Self {
373            bind_address: bind_address.into(),
374        }
375    }
376}
377
378/// Yellowstone gRPC configuration
379#[derive(Clone, Debug)]
380pub struct YellowstoneConfig {
381    pub endpoint: String,
382    pub x_token: Option<String>,
383}
384
385impl YellowstoneConfig {
386    pub fn new(endpoint: impl Into<String>) -> Self {
387        Self {
388            endpoint: endpoint.into(),
389            x_token: None,
390        }
391    }
392
393    pub fn with_token(mut self, token: impl Into<String>) -> Self {
394        self.x_token = Some(token.into());
395        self
396    }
397}
398
399/// Main server configuration
400#[derive(Clone, Debug, Default)]
401pub struct ServerConfig {
402    pub runtime_plan: RuntimePlan,
403    pub websocket: Option<WebSocketConfig>,
404    pub yellowstone: Option<YellowstoneConfig>,
405    pub health: Option<HealthConfig>,
406    pub http_health: Option<HttpHealthConfig>,
407    pub reconnection: Option<ReconnectionConfig>,
408    pub transactions: Option<TransactionConfig>,
409    pub solana_gateway_target_id: Option<String>,
410    pub program_read_binding_target_id: Option<String>,
411    /// State snapshot settings. `None` falls back to `SnapshotConfig::from_env()`.
412    pub snapshots: Option<crate::snapshot::SnapshotConfig>,
413    /// Event journal settings. `None` falls back to `JournalConfig::from_env()`.
414    ///
415    /// Set this per runtime when several deployments share one process: the
416    /// env vars are process-wide, so they cannot enable replay for one stack
417    /// or size a busy stack differently from a quiet one.
418    pub journal: Option<crate::journal::JournalConfig>,
419    /// WebSocket buffering and latest-state delivery settings. `None` falls
420    /// back to [`WebSocketDeliveryConfig::from_env`].
421    pub websocket_delivery: Option<WebSocketDeliveryConfig>,
422}
423
424impl ServerConfig {
425    pub fn new() -> Self {
426        Self::default()
427    }
428
429    pub fn with_websocket(mut self, config: WebSocketConfig) -> Self {
430        self.websocket = Some(config);
431        self.runtime_plan.websocket = true;
432        self.runtime_plan.live_runtime = true;
433        self
434    }
435
436    pub fn with_websocket_delivery(mut self, config: WebSocketDeliveryConfig) -> Self {
437        self.websocket_delivery = Some(config);
438        self.runtime_plan.live_runtime = true;
439        self
440    }
441
442    pub fn with_yellowstone(mut self, config: YellowstoneConfig) -> Self {
443        self.yellowstone = Some(config);
444        self.runtime_plan.live_runtime = true;
445        self
446    }
447
448    pub fn with_health(mut self, config: HealthConfig) -> Self {
449        self.health = Some(config);
450        self.runtime_plan.health = true;
451        self
452    }
453
454    pub fn with_http_health(mut self, config: HttpHealthConfig) -> Self {
455        self.http_health = Some(config);
456        self.runtime_plan.health = true;
457        self.runtime_plan.chain_reads = true;
458        self.runtime_plan.program_reads = true;
459        self.runtime_plan.stack_queries = true;
460        self
461    }
462
463    pub fn with_reconnection(mut self, config: ReconnectionConfig) -> Self {
464        self.reconnection = Some(config);
465        self
466    }
467
468    pub fn with_transactions(mut self, config: TransactionConfig) -> Self {
469        self.runtime_plan.transactions = config.enabled;
470        self.transactions = Some(config);
471        self
472    }
473
474    pub fn with_runtime_plan(mut self, runtime_plan: RuntimePlan) -> Self {
475        self.runtime_plan = runtime_plan;
476        self
477    }
478
479    pub fn with_snapshots(mut self, config: crate::snapshot::SnapshotConfig) -> Self {
480        self.snapshots = Some(config);
481        self
482    }
483}
arete_server/config.rs

arete_server/
config.rs
