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/// Explicit configuration for the fixed transaction relay routes.
12#[derive(Clone, Debug)]
13pub struct TransactionConfig {
14    pub enabled: bool,
15    pub rpc_url: Option<String>,
16    /// Development-only: allow relay requests when no auth plugin is configured.
17    pub allow_unauthenticated: bool,
18    pub max_body_bytes: usize,
19    pub max_transaction_bytes: usize,
20    pub inspect_timeout: Duration,
21    pub send_timeout: Duration,
22    pub status_timeout: Duration,
23    pub inspect_concurrency: usize,
24    pub send_concurrency: usize,
25    pub inspect_requests_per_minute: u32,
26    pub send_requests_per_minute: u32,
27    pub status_requests_per_minute: u32,
28    pub trusted_proxy_cidrs: Vec<IpNet>,
29    pub usage_enabled: bool,
30    pub usage_endpoint: Option<String>,
31    pub usage_token: Option<String>,
32    pub usage_spool_capacity: usize,
33}
34
35impl Default for TransactionConfig {
36    fn default() -> Self {
37        Self {
38            enabled: false,
39            rpc_url: None,
40            allow_unauthenticated: false,
41            max_body_bytes: 4 * 1024,
42            max_transaction_bytes: 1232,
43            inspect_timeout: Duration::from_secs(10),
44            send_timeout: Duration::from_secs(15),
45            status_timeout: Duration::from_secs(10),
46            inspect_concurrency: 64,
47            send_concurrency: 16,
48            inspect_requests_per_minute: 600,
49            send_requests_per_minute: 60,
50            status_requests_per_minute: 600,
51            trusted_proxy_cidrs: Vec::new(),
52            usage_enabled: false,
53            usage_endpoint: None,
54            usage_token: None,
55            usage_spool_capacity: 1_000,
56        }
57    }
58}
59
60impl TransactionConfig {
61    /// Load transaction settings. Routes remain disabled unless explicitly enabled.
62    pub fn from_env() -> Result<Self> {
63        let mut config = Self::default();
64        config.enabled = env_bool("ARETE_TRANSACTIONS_ENABLED")?.unwrap_or(false);
65        config.rpc_url =
66            first_nonempty(&["ARETE_TRANSACTION_RPC_URL", "SOLANA_RPC_URL", "RPC_URL"]);
67        config.allow_unauthenticated =
68            env_bool("ARETE_TRANSACTIONS_ALLOW_UNAUTHENTICATED")?.unwrap_or(false);
69        config.max_body_bytes =
70            env_parse("ARETE_TRANSACTION_MAX_BODY_BYTES")?.unwrap_or(config.max_body_bytes);
71        config.max_transaction_bytes =
72            env_parse("ARETE_TRANSACTION_MAX_BYTES")?.unwrap_or(config.max_transaction_bytes);
73        config.inspect_timeout = Duration::from_millis(
74            env_parse("ARETE_TRANSACTION_INSPECT_TIMEOUT_MS")?
75                .unwrap_or(config.inspect_timeout.as_millis() as u64),
76        );
77        config.send_timeout = Duration::from_millis(
78            env_parse("ARETE_TRANSACTION_SEND_TIMEOUT_MS")?
79                .unwrap_or(config.send_timeout.as_millis() as u64),
80        );
81        config.status_timeout = Duration::from_millis(
82            env_parse("ARETE_TRANSACTION_STATUS_TIMEOUT_MS")?
83                .unwrap_or(config.status_timeout.as_millis() as u64),
84        );
85        config.inspect_concurrency = env_parse("ARETE_TRANSACTION_INSPECT_CONCURRENCY")?
86            .unwrap_or(config.inspect_concurrency);
87        config.send_concurrency =
88            env_parse("ARETE_TRANSACTION_SEND_CONCURRENCY")?.unwrap_or(config.send_concurrency);
89        config.inspect_requests_per_minute =
90            env_parse("ARETE_TRANSACTION_INSPECT_REQUESTS_PER_MINUTE")?
91                .unwrap_or(config.inspect_requests_per_minute);
92        config.send_requests_per_minute = env_parse("ARETE_TRANSACTION_SEND_REQUESTS_PER_MINUTE")?
93            .unwrap_or(config.send_requests_per_minute);
94        config.status_requests_per_minute =
95            env_parse("ARETE_TRANSACTION_STATUS_REQUESTS_PER_MINUTE")?
96                .unwrap_or(config.status_requests_per_minute);
97        config.trusted_proxy_cidrs = std::env::var("ARETE_TRUSTED_PROXY_CIDRS")
98            .ok()
99            .filter(|value| !value.trim().is_empty())
100            .map(|value| {
101                value
102                    .split(',')
103                    .map(|cidr| cidr.trim().parse().context("invalid trusted proxy CIDR"))
104                    .collect::<Result<Vec<_>>>()
105            })
106            .transpose()?
107            .unwrap_or_default();
108        config.usage_enabled = env_bool("ARETE_TRANSACTION_USAGE_ENABLED")?.unwrap_or(false);
109        config.usage_endpoint = first_nonempty(&["ARETE_TRANSACTION_USAGE_ENDPOINT"]);
110        config.usage_token = first_nonempty(&["ARETE_TRANSACTION_USAGE_TOKEN"]);
111        config.usage_spool_capacity = env_parse("ARETE_TRANSACTION_USAGE_SPOOL_CAPACITY")?
112            .unwrap_or(config.usage_spool_capacity);
113        config.validate()?;
114        Ok(config)
115    }
116
117    pub fn validate(&self) -> Result<()> {
118        if self.enabled && self.rpc_url.as_deref().unwrap_or_default().is_empty() {
119            bail!("transactions are enabled but no transaction RPC URL is configured");
120        }
121        if self.max_body_bytes == 0
122            || self.max_transaction_bytes == 0
123            || self.inspect_concurrency == 0
124            || self.send_concurrency == 0
125        {
126            bail!("transaction size and concurrency limits must be greater than zero");
127        }
128        if self.usage_enabled && (self.usage_endpoint.is_none() || self.usage_token.is_none()) {
129            bail!("transaction usage requires an endpoint and token");
130        }
131        Ok(())
132    }
133}
134
135fn first_nonempty(keys: &[&str]) -> Option<String> {
136    keys.iter()
137        .find_map(|key| std::env::var(key).ok())
138        .filter(|value| !value.trim().is_empty())
139}
140
141fn env_parse<T>(key: &str) -> Result<Option<T>>
142where
143    T: std::str::FromStr,
144    T::Err: std::error::Error + Send + Sync + 'static,
145{
146    std::env::var(key)
147        .ok()
148        .map(|value| value.parse().with_context(|| format!("invalid {key}")))
149        .transpose()
150}
151
152fn env_bool(key: &str) -> Result<Option<bool>> {
153    let Some(value) = std::env::var(key).ok() else {
154        return Ok(None);
155    };
156    match value.to_ascii_lowercase().as_str() {
157        "true" | "1" | "yes" => Ok(Some(true)),
158        "false" | "0" | "no" => Ok(Some(false)),
159        _ => bail!("invalid {key}; expected true or false"),
160    }
161}
162
163/// Configuration for gRPC stream reconnection with exponential backoff
164#[derive(Clone, Debug)]
165pub struct ReconnectionConfig {
166    /// Initial delay before first reconnection attempt
167    pub initial_delay: Duration,
168    /// Maximum delay between reconnection attempts
169    pub max_delay: Duration,
170    /// Maximum number of reconnection attempts (None = infinite)
171    pub max_attempts: Option<u32>,
172    /// Multiplier for exponential backoff (typically 2.0)
173    pub backoff_multiplier: f64,
174    /// HTTP/2 keep-alive interval to prevent silent disconnects
175    pub http2_keep_alive_interval: Option<Duration>,
176}
177
178impl Default for ReconnectionConfig {
179    fn default() -> Self {
180        Self {
181            initial_delay: Duration::from_millis(100),
182            max_delay: Duration::from_secs(60),
183            max_attempts: None, // Infinite retries by default
184            backoff_multiplier: 2.0,
185            http2_keep_alive_interval: Some(Duration::from_secs(30)),
186        }
187    }
188}
189
190impl ReconnectionConfig {
191    pub fn new() -> Self {
192        Self::default()
193    }
194
195    pub fn with_initial_delay(mut self, delay: Duration) -> Self {
196        self.initial_delay = delay;
197        self
198    }
199
200    pub fn with_max_delay(mut self, delay: Duration) -> Self {
201        self.max_delay = delay;
202        self
203    }
204
205    pub fn with_max_attempts(mut self, attempts: u32) -> Self {
206        self.max_attempts = Some(attempts);
207        self
208    }
209
210    pub fn with_backoff_multiplier(mut self, multiplier: f64) -> Self {
211        self.backoff_multiplier = multiplier;
212        self
213    }
214
215    pub fn with_http2_keep_alive_interval(mut self, interval: Duration) -> Self {
216        self.http2_keep_alive_interval = Some(interval);
217        self
218    }
219
220    /// Calculate the next backoff duration given the current one
221    pub fn next_backoff(&self, current: Duration) -> Duration {
222        let next_secs = current.as_secs_f64() * self.backoff_multiplier;
223        let capped_secs = next_secs.min(self.max_delay.as_secs_f64());
224        Duration::from_secs_f64(capped_secs)
225    }
226}
227
228/// WebSocket server configuration
229#[derive(Clone, Debug)]
230pub struct WebSocketConfig {
231    pub bind_address: SocketAddr,
232}
233
234/// The only wire protocol accepted by the WebSocket server.
235pub const WEBSOCKET_PROTOCOL_VERSION: u8 = crate::websocket::subscription::PROTOCOL_VERSION;
236
237impl Default for WebSocketConfig {
238    fn default() -> Self {
239        Self {
240            bind_address: "[::]:8877".parse().expect("valid socket address"),
241        }
242    }
243}
244
245impl WebSocketConfig {
246    pub fn new(bind_address: impl Into<SocketAddr>) -> Self {
247        Self {
248            bind_address: bind_address.into(),
249        }
250    }
251}
252
253/// Yellowstone gRPC configuration
254#[derive(Clone, Debug)]
255pub struct YellowstoneConfig {
256    pub endpoint: String,
257    pub x_token: Option<String>,
258}
259
260impl YellowstoneConfig {
261    pub fn new(endpoint: impl Into<String>) -> Self {
262        Self {
263            endpoint: endpoint.into(),
264            x_token: None,
265        }
266    }
267
268    pub fn with_token(mut self, token: impl Into<String>) -> Self {
269        self.x_token = Some(token.into());
270        self
271    }
272}
273
274/// Main server configuration
275#[derive(Clone, Debug, Default)]
276pub struct ServerConfig {
277    pub websocket: Option<WebSocketConfig>,
278    pub yellowstone: Option<YellowstoneConfig>,
279    pub health: Option<HealthConfig>,
280    pub http_health: Option<HttpHealthConfig>,
281    pub reconnection: Option<ReconnectionConfig>,
282    pub transactions: Option<TransactionConfig>,
283}
284
285impl ServerConfig {
286    pub fn new() -> Self {
287        Self::default()
288    }
289
290    pub fn with_websocket(mut self, config: WebSocketConfig) -> Self {
291        self.websocket = Some(config);
292        self
293    }
294
295    pub fn with_yellowstone(mut self, config: YellowstoneConfig) -> Self {
296        self.yellowstone = Some(config);
297        self
298    }
299
300    pub fn with_health(mut self, config: HealthConfig) -> Self {
301        self.health = Some(config);
302        self
303    }
304
305    pub fn with_http_health(mut self, config: HttpHealthConfig) -> Self {
306        self.http_health = Some(config);
307        self
308    }
309
310    pub fn with_reconnection(mut self, config: ReconnectionConfig) -> Self {
311        self.reconnection = Some(config);
312        self
313    }
314
315    pub fn with_transactions(mut self, config: TransactionConfig) -> Self {
316        self.transactions = Some(config);
317        self
318    }
319}
arete_server/config.rs

arete_server/
config.rs
