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#[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 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#[derive(Clone, Debug)]
62pub struct TransactionConfig {
63 pub enabled: bool,
64 pub rpc_url: Option<String>,
65 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 max_body_bytes: 32 * 1024,
95 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 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#[derive(Clone, Debug, PartialEq, Eq)]
224pub struct HttpLimitsConfig {
225 pub max_body_bytes: usize,
227 pub header_read_timeout: Duration,
230 pub body_read_timeout: Duration,
233 pub upstream_connect_timeout: Duration,
235 pub upstream_timeout: Duration,
237 pub max_upstream_response_bytes: usize,
240}
241
242impl Default for HttpLimitsConfig {
243 fn default() -> Self {
244 Self {
245 max_body_bytes: 1024 * 1024,
246 header_read_timeout: Duration::from_secs(10),
247 body_read_timeout: Duration::from_secs(30),
248 upstream_connect_timeout: Duration::from_secs(5),
249 upstream_timeout: Duration::from_secs(30),
250 max_upstream_response_bytes: 64 * 1024 * 1024,
252 }
253 }
254}
255
256impl HttpLimitsConfig {
257 pub fn from_env() -> Result<Self> {
258 let mut config = Self::default();
259 config.max_body_bytes =
260 env_parse("ARETE_HTTP_MAX_BODY_BYTES")?.unwrap_or(config.max_body_bytes);
261 config.header_read_timeout =
262 env_millis("ARETE_HTTP_HEADER_TIMEOUT_MS")?.unwrap_or(config.header_read_timeout);
263 config.body_read_timeout =
264 env_millis("ARETE_HTTP_BODY_TIMEOUT_MS")?.unwrap_or(config.body_read_timeout);
265 config.upstream_connect_timeout = env_millis("ARETE_READ_RPC_CONNECT_TIMEOUT_MS")?
266 .unwrap_or(config.upstream_connect_timeout);
267 config.upstream_timeout =
268 env_millis("ARETE_READ_RPC_TIMEOUT_MS")?.unwrap_or(config.upstream_timeout);
269 config.max_upstream_response_bytes = env_parse("ARETE_READ_RPC_MAX_RESPONSE_BYTES")?
270 .unwrap_or(config.max_upstream_response_bytes);
271 config.validate()?;
272 Ok(config)
273 }
274
275 pub fn validate(&self) -> Result<()> {
276 if self.max_body_bytes == 0 || self.max_upstream_response_bytes == 0 {
277 bail!("HTTP body and read RPC response limits must be greater than zero");
278 }
279 if [
280 self.header_read_timeout,
281 self.body_read_timeout,
282 self.upstream_connect_timeout,
283 self.upstream_timeout,
284 ]
285 .contains(&Duration::ZERO)
286 {
287 bail!("HTTP and read RPC timeouts must be greater than zero");
288 }
289 Ok(())
290 }
291}
292
293#[derive(Clone, Debug, PartialEq, Eq)]
295pub struct WebSocketLimitsConfig {
296 pub max_message_bytes: usize,
299 pub max_frame_bytes: usize,
301 pub handshake_timeout: Duration,
304 pub max_pending_handshakes: usize,
307}
308
309impl Default for WebSocketLimitsConfig {
310 fn default() -> Self {
311 Self {
312 max_message_bytes: 1024 * 1024,
314 max_frame_bytes: 1024 * 1024,
315 handshake_timeout: Duration::from_secs(10),
316 max_pending_handshakes: 1024,
317 }
318 }
319}
320
321impl WebSocketLimitsConfig {
322 pub fn from_env() -> Result<Self> {
323 let mut config = Self::default();
324 config.max_message_bytes =
325 env_parse("ARETE_WS_MAX_MESSAGE_BYTES")?.unwrap_or(config.max_message_bytes);
326 config.max_frame_bytes =
327 env_parse("ARETE_WS_MAX_FRAME_BYTES")?.unwrap_or(config.max_frame_bytes);
328 config.handshake_timeout =
329 env_millis("ARETE_WS_HANDSHAKE_TIMEOUT_MS")?.unwrap_or(config.handshake_timeout);
330 config.max_pending_handshakes =
331 env_parse("ARETE_WS_MAX_PENDING_HANDSHAKES")?.unwrap_or(config.max_pending_handshakes);
332 config.validate()?;
333 Ok(config)
334 }
335
336 pub fn validate(&self) -> Result<()> {
337 if self.max_message_bytes == 0
338 || self.max_frame_bytes == 0
339 || self.max_pending_handshakes == 0
340 || self.handshake_timeout.is_zero()
341 {
342 bail!(
343 "WebSocket size, handshake and pending-handshake limits must be greater than zero"
344 );
345 }
346 Ok(())
347 }
348}
349
350fn env_millis(key: &str) -> Result<Option<Duration>> {
351 Ok(env_parse::<u64>(key)?.map(Duration::from_millis))
352}
353
354#[derive(Clone, Debug)]
356pub struct ReconnectionConfig {
357 pub initial_delay: Duration,
359 pub max_delay: Duration,
361 pub max_attempts: Option<u32>,
363 pub backoff_multiplier: f64,
365 pub http2_keep_alive_interval: Option<Duration>,
367 pub live_fallback_attempts: Option<u32>,
374}
375
376impl Default for ReconnectionConfig {
377 fn default() -> Self {
378 Self {
379 initial_delay: Duration::from_millis(100),
380 max_delay: Duration::from_secs(60),
381 max_attempts: None, backoff_multiplier: 2.0,
383 http2_keep_alive_interval: Some(Duration::from_secs(30)),
384 live_fallback_attempts: Some(3),
385 }
386 }
387}
388
389impl ReconnectionConfig {
390 pub fn new() -> Self {
391 Self::default()
392 }
393
394 pub fn with_initial_delay(mut self, delay: Duration) -> Self {
395 self.initial_delay = delay;
396 self
397 }
398
399 pub fn with_max_delay(mut self, delay: Duration) -> Self {
400 self.max_delay = delay;
401 self
402 }
403
404 pub fn with_max_attempts(mut self, attempts: u32) -> Self {
405 self.max_attempts = Some(attempts);
406 self
407 }
408
409 pub fn with_backoff_multiplier(mut self, multiplier: f64) -> Self {
410 self.backoff_multiplier = multiplier;
411 self
412 }
413
414 pub fn with_http2_keep_alive_interval(mut self, interval: Duration) -> Self {
415 self.http2_keep_alive_interval = Some(interval);
416 self
417 }
418
419 pub fn fail_closed(mut self) -> Self {
422 self.live_fallback_attempts = None;
423 self
424 }
425
426 pub fn next_backoff(&self, current: Duration) -> Duration {
428 let next_secs = current.as_secs_f64() * self.backoff_multiplier;
429 let capped_secs = next_secs.min(self.max_delay.as_secs_f64());
430 Duration::from_secs_f64(capped_secs)
431 }
432}
433
434#[derive(Clone, Debug)]
436pub struct WebSocketConfig {
437 pub bind_address: SocketAddr,
438}
439
440#[derive(Clone, Debug, PartialEq, Eq)]
445pub struct WebSocketDeliveryConfig {
446 pub list_bus_capacity: usize,
448 pub collection_coalesce_ms: Option<u64>,
451}
452
453impl Default for WebSocketDeliveryConfig {
454 fn default() -> Self {
455 Self {
456 list_bus_capacity: 8 * 1024,
458 collection_coalesce_ms: None,
459 }
460 }
461}
462
463impl WebSocketDeliveryConfig {
464 pub fn from_env() -> anyhow::Result<Self> {
465 let mut config = Self::default();
466 if let Ok(value) = std::env::var("ARETE_WS_LIST_BUS_CAPACITY") {
467 config.list_bus_capacity = value.parse().map_err(|_| {
468 anyhow::anyhow!("ARETE_WS_LIST_BUS_CAPACITY must be a positive integer")
469 })?;
470 }
471 if let Ok(value) = std::env::var("ARETE_WS_COLLECTION_COALESCE_MS") {
472 let milliseconds: u64 = value.parse().map_err(|_| {
473 anyhow::anyhow!("ARETE_WS_COLLECTION_COALESCE_MS must be a non-negative integer")
474 })?;
475 config.collection_coalesce_ms = (milliseconds > 0).then_some(milliseconds);
476 }
477 config.validate()?;
478 Ok(config)
479 }
480
481 pub fn validate(&self) -> anyhow::Result<()> {
482 anyhow::ensure!(
483 self.list_bus_capacity > 0,
484 "WebSocket list bus capacity must be greater than zero"
485 );
486 anyhow::ensure!(
487 self.collection_coalesce_ms
488 .is_none_or(|milliseconds| milliseconds <= 60_000),
489 "WebSocket collection coalescing must not exceed 60000ms"
490 );
491 Ok(())
492 }
493}
494
495pub const WEBSOCKET_PROTOCOL_VERSION: u8 = crate::websocket::subscription::PROTOCOL_VERSION;
497
498impl Default for WebSocketConfig {
499 fn default() -> Self {
500 Self {
501 bind_address: "[::]:8877".parse().expect("valid socket address"),
502 }
503 }
504}
505
506impl WebSocketConfig {
507 pub fn new(bind_address: impl Into<SocketAddr>) -> Self {
508 Self {
509 bind_address: bind_address.into(),
510 }
511 }
512}
513
514#[derive(Clone, Debug)]
516pub struct YellowstoneConfig {
517 pub endpoint: String,
518 pub x_token: Option<String>,
519}
520
521impl YellowstoneConfig {
522 pub fn new(endpoint: impl Into<String>) -> Self {
523 Self {
524 endpoint: endpoint.into(),
525 x_token: None,
526 }
527 }
528
529 pub fn with_token(mut self, token: impl Into<String>) -> Self {
530 self.x_token = Some(token.into());
531 self
532 }
533}
534
535#[derive(Clone, Debug, Default)]
537pub struct ServerConfig {
538 pub runtime_plan: RuntimePlan,
539 pub websocket: Option<WebSocketConfig>,
540 pub yellowstone: Option<YellowstoneConfig>,
541 pub health: Option<HealthConfig>,
542 pub http_health: Option<HttpHealthConfig>,
543 pub reconnection: Option<ReconnectionConfig>,
544 pub transactions: Option<TransactionConfig>,
545 pub solana_gateway_target_id: Option<String>,
546 pub program_read_binding_target_id: Option<String>,
547 pub snapshots: Option<crate::snapshot::SnapshotConfig>,
549 pub journal: Option<crate::journal::JournalConfig>,
555 pub websocket_delivery: Option<WebSocketDeliveryConfig>,
558 pub websocket_limits: Option<WebSocketLimitsConfig>,
561 pub http_limits: Option<HttpLimitsConfig>,
564}
565
566impl ServerConfig {
567 pub fn new() -> Self {
568 Self::default()
569 }
570
571 pub fn with_websocket(mut self, config: WebSocketConfig) -> Self {
572 self.websocket = Some(config);
573 self.runtime_plan.websocket = true;
574 self.runtime_plan.live_runtime = true;
575 self
576 }
577
578 pub fn with_websocket_delivery(mut self, config: WebSocketDeliveryConfig) -> Self {
579 self.websocket_delivery = Some(config);
580 self.runtime_plan.live_runtime = true;
581 self
582 }
583
584 pub fn with_websocket_limits(mut self, config: WebSocketLimitsConfig) -> Self {
585 self.websocket_limits = Some(config);
586 self
587 }
588
589 pub fn with_http_limits(mut self, config: HttpLimitsConfig) -> Self {
590 self.http_limits = Some(config);
591 self
592 }
593
594 pub fn with_yellowstone(mut self, config: YellowstoneConfig) -> Self {
595 self.yellowstone = Some(config);
596 self.runtime_plan.live_runtime = true;
597 self
598 }
599
600 pub fn with_health(mut self, config: HealthConfig) -> Self {
601 self.health = Some(config);
602 self.runtime_plan.health = true;
603 self
604 }
605
606 pub fn with_http_health(mut self, config: HttpHealthConfig) -> Self {
607 self.http_health = Some(config);
608 self.runtime_plan.health = true;
609 self.runtime_plan.chain_reads = true;
610 self.runtime_plan.program_reads = true;
611 self.runtime_plan.stack_queries = true;
612 self
613 }
614
615 pub fn with_reconnection(mut self, config: ReconnectionConfig) -> Self {
616 self.reconnection = Some(config);
617 self
618 }
619
620 pub fn with_transactions(mut self, config: TransactionConfig) -> Self {
621 self.runtime_plan.transactions = config.enabled;
622 self.transactions = Some(config);
623 self
624 }
625
626 pub fn with_runtime_plan(mut self, runtime_plan: RuntimePlan) -> Self {
627 self.runtime_plan = runtime_plan;
628 self
629 }
630
631 pub fn with_snapshots(mut self, config: crate::snapshot::SnapshotConfig) -> Self {
632 self.snapshots = Some(config);
633 self
634 }
635}