nautilus_network/socket/
config.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Static transport, framing, heartbeat, and reconnect configuration for TCP sockets.
17//!
18//! # Reconnection strategy
19//!
20//! The default configuration uses unlimited reconnection attempts (`reconnect_max_attempts: None`).
21//! This suits long-lived trading connections because:
22//!
23//! - Venues may remain unavailable for an extended period and later recover.
24//! - Exponential backoff bounds retry frequency during the outage.
25//! - Automatic recovery avoids requiring manual intervention for a transient failure.
26//!
27//! A connection active for at least 10 seconds resets the attempt count and backoff delay.
28//! Shorter-lived connections remain part of the same reconnect cycle. Use `Some(n)` primarily for
29//! tests, development, or connections that should stop retrying without intervention.
30
31use std::fmt::Debug;
32
33use nautilus_core::string::secret::REDACTED;
34use tokio_tungstenite::tungstenite::stream::Mode;
35
36use super::types::TcpMessageHandler;
37use crate::error::{NetworkConfigError, NetworkConfigResult};
38
39/// Application keepalive for a raw TCP socket.
40///
41/// A raw socket has no control frames, so [`Self::payload`] is required. WebSocket keepalives stay
42/// on [`crate::websocket::WebSocketConfig`] and may omit a payload to send a protocol Ping.
43#[derive(Clone, Debug, PartialEq, Eq)]
44pub struct SocketHeartbeat {
45    /// Interval between keepalives, in seconds.
46    ///
47    /// Must be positive.
48    pub interval_secs: u64,
49    /// Bytes sent as each heartbeat, framed with [`SocketConfig::suffix`].
50    pub payload: Vec<u8>,
51}
52
53/// Configuration for a TCP socket connection.
54#[derive(Clone, bon::Builder)]
55#[builder(finish_fn(name = build_inner, vis = ""))]
56pub struct SocketConfig {
57    /// The server address as `host:port` or a URL.
58    pub url: String,
59    /// The plain or TLS connection mode.
60    pub mode: Mode,
61    /// The byte sequence that frames messages in both directions.
62    pub suffix: Vec<u8>,
63    /// The function called for each complete incoming message.
64    pub message_handler: Option<TcpMessageHandler>,
65    /// Optional application keepalive.
66    ///
67    /// When set, the client sends [`SocketHeartbeat::payload`] every
68    /// [`SocketHeartbeat::interval_secs`] seconds. A raw socket has no Ping frames, so this type
69    /// requires a payload; the WebSocket Ping-frame case stays on
70    /// [`crate::websocket::WebSocketConfig::heartbeat_payload`].
71    ///
72    /// Each timing field carries the coarsest unit that expresses every legitimate value, and
73    /// quantities compared against each other share a unit: the interval and
74    /// [`Self::heartbeat_timeout_secs`] are bounded below by whole-second cadences, while reconnect
75    /// delays and jitter have real sub-second values and stay in milliseconds.
76    pub heartbeat: Option<SocketHeartbeat>,
77    /// The timeout (milliseconds) for establishing a usable connection. Defaults to 10 seconds.
78    ///
79    /// Bounds the initial connection attempt, each reconnect attempt, and how long a send waits for
80    /// the client to become active again. Keep it above the reconnect backoff so a send does not
81    /// give up part-way through a normal reconnect.
82    pub connect_timeout_ms: Option<u64>,
83    /// The initial reconnection delay (milliseconds) for reconnects.
84    pub reconnect_delay_initial_ms: Option<u64>,
85    /// The maximum reconnect delay (milliseconds) for exponential backoff.
86    pub reconnect_delay_max_ms: Option<u64>,
87    /// The exponential backoff factor for reconnection delays.
88    pub reconnect_backoff_factor: Option<f64>,
89    /// The maximum jitter (milliseconds) added to reconnection delays.
90    pub reconnect_jitter_ms: Option<u64>,
91    /// The maximum number of initial connection attempts. Defaults to 5.
92    pub connection_max_retries: Option<u32>,
93    /// The maximum number of reconnection attempts before closing the client.
94    ///
95    /// - `None`: Unlimited reconnection attempts (default, recommended for production).
96    /// - `Some(n)`: Transitions to CLOSED once `n` consecutive reconnect attempts have either
97    ///   failed or established connections active for less than 10 seconds.
98    pub reconnect_max_attempts: Option<u32>,
99    /// The dead-peer timeout (seconds) for the read task.
100    ///
101    /// Seconds rather than milliseconds because this is a multiple of the heartbeat interval: it
102    /// can never sensibly sit below one heartbeat cycle.
103    ///
104    /// When set, the read task stops and triggers reconnection if no bytes at all arrive within
105    /// this duration. A raw socket has no control frames, so any inbound byte refreshes it,
106    /// including the venue's reply to a heartbeat. That makes this the byte-level equivalent of
107    /// the WebSocket client's `heartbeat_timeout_secs`, not of its `idle_timeout_ms`: there is
108    /// no transport-level way to tell keepalive traffic from data here.
109    ///
110    /// `None` derives three heartbeat intervals when [`Self::heartbeat`] is set, and disables
111    /// detection otherwise. `Some(0)` is rejected. Set an explicit value above the heartbeat
112    /// interval so a healthy connection cannot trip it.
113    pub heartbeat_timeout_secs: Option<u64>,
114    /// The path to the certificates directory.
115    pub certs_dir: Option<String>,
116}
117
118impl<S: socket_config_builder::IsComplete> SocketConfigBuilder<S> {
119    /// Validates and builds the [`SocketConfig`].
120    ///
121    /// # Errors
122    ///
123    /// Returns a [`NetworkConfigError`] if any field fails validation
124    /// (see [`SocketConfig::validate`]).
125    pub fn build(self) -> NetworkConfigResult<SocketConfig> {
126        let config = self.build_inner();
127        config.validate()?;
128        Ok(config)
129    }
130}
131
132impl SocketConfig {
133    /// Checks whether all socket settings are valid.
134    ///
135    /// # Errors
136    ///
137    /// Returns a [`NetworkConfigError`] if `url` is empty, the heartbeat interval or a
138    /// reconnection timing field is not positive, `reconnect_backoff_factor` is outside
139    /// `[1.0, 100.0]`, or `reconnect_delay_initial_ms` exceeds `reconnect_delay_max_ms`.
140    pub fn validate(&self) -> NetworkConfigResult<()> {
141        let mut errors = Vec::new();
142
143        if self.url.trim().is_empty() {
144            errors.push(NetworkConfigError::invalid("url", "must not be empty"));
145        }
146
147        if let Some(heartbeat) = &self.heartbeat
148            && heartbeat.interval_secs == 0
149        {
150            errors.push(NetworkConfigError::invalid(
151                "heartbeat",
152                "interval must be positive",
153            ));
154        }
155
156        // A timeout at or below the send cadence tears every connection down before its first
157        // reply is due, so a healthy socket would reconnect forever.
158        if let (Some(heartbeat), Some(timeout_secs)) =
159            (&self.heartbeat, self.heartbeat_timeout_secs)
160            && timeout_secs <= heartbeat.interval_secs
161        {
162            errors.push(NetworkConfigError::invalid(
163                "heartbeat_timeout_secs",
164                format!(
165                    "must exceed heartbeat interval ({}s), was {timeout_secs}s",
166                    heartbeat.interval_secs
167                ),
168            ));
169        }
170
171        // `reconnect_jitter_ms` is intentionally unchecked: zero disables jitter and
172        // `ExponentialBackoff::new` accepts it.
173        for (field, value) in [
174            ("connect_timeout_ms", self.connect_timeout_ms),
175            (
176                "reconnect_delay_initial_ms",
177                self.reconnect_delay_initial_ms,
178            ),
179            ("reconnect_delay_max_ms", self.reconnect_delay_max_ms),
180            ("heartbeat_timeout_secs", self.heartbeat_timeout_secs),
181        ] {
182            if let Some(value) = value
183                && value == 0
184            {
185                errors.push(NetworkConfigError::invalid(
186                    field,
187                    format!("must be positive, was {value}"),
188                ));
189            }
190        }
191
192        if let Some(factor) = self.reconnect_backoff_factor
193            && !(1.0..=100.0).contains(&factor)
194        {
195            errors.push(NetworkConfigError::invalid(
196                "reconnect_backoff_factor",
197                format!("must be in range [1.0, 100.0], was {factor}"),
198            ));
199        }
200
201        if let (Some(initial), Some(max)) =
202            (self.reconnect_delay_initial_ms, self.reconnect_delay_max_ms)
203            && initial > max
204        {
205            errors.push(NetworkConfigError::invalid(
206                "reconnect_delay_initial_ms",
207                format!("must not exceed reconnect_delay_max_ms ({max}), was {initial}"),
208            ));
209        }
210
211        NetworkConfigError::collect(errors)
212    }
213
214    pub(crate) fn resolved_heartbeat_timeout(&self) -> Option<u64> {
215        crate::heartbeat::resolve_heartbeat_timeout(
216            self.heartbeat_timeout_secs,
217            self.heartbeat
218                .as_ref()
219                .map(|heartbeat| heartbeat.interval_secs),
220        )
221    }
222}
223
224impl Debug for SocketConfig {
225    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
226        f.debug_struct(stringify!(SocketConfig))
227            .field("url", &REDACTED)
228            .field("mode", &self.mode)
229            .field("suffix", &self.suffix)
230            .field(
231                "message_handler",
232                &self.message_handler.as_ref().map(|_| "<function>"),
233            )
234            .field("heartbeat", &self.heartbeat)
235            .field("connect_timeout_ms", &self.connect_timeout_ms)
236            .field(
237                "reconnect_delay_initial_ms",
238                &self.reconnect_delay_initial_ms,
239            )
240            .field("reconnect_delay_max_ms", &self.reconnect_delay_max_ms)
241            .field("reconnect_backoff_factor", &self.reconnect_backoff_factor)
242            .field("reconnect_jitter_ms", &self.reconnect_jitter_ms)
243            .field("connection_max_retries", &self.connection_max_retries)
244            .field("reconnect_max_attempts", &self.reconnect_max_attempts)
245            .field("heartbeat_timeout_secs", &self.heartbeat_timeout_secs)
246            .field("certs_dir", &self.certs_dir)
247            .finish()
248    }
249}
250
251#[cfg(test)]
252mod tests {
253    use rstest::rstest;
254    use tokio_tungstenite::tungstenite::stream::Mode;
255
256    use super::{SocketConfig, SocketHeartbeat};
257    use crate::error::NetworkConfigError;
258
259    fn valid_config() -> SocketConfig {
260        SocketConfig::builder()
261            .url("tcp://127.0.0.1:8080".to_string())
262            .mode(Mode::Plain)
263            .suffix(vec![b'\n'])
264            .build()
265            .expect("baseline socket config should be valid")
266    }
267
268    #[rstest]
269    fn test_builder_accepts_valid_config() {
270        let result = SocketConfig::builder()
271            .url("tcp://127.0.0.1:8080".to_string())
272            .mode(Mode::Plain)
273            .suffix(vec![b'\n'])
274            .build();
275
276        assert!(result.is_ok());
277    }
278
279    #[rstest]
280    fn test_validate_accepts_zero_jitter() {
281        let mut config = valid_config();
282        config.reconnect_jitter_ms = Some(0);
283
284        assert!(config.validate().is_ok());
285    }
286
287    #[rstest]
288    fn test_validate_accepts_heartbeat_with_payload() {
289        let mut config = valid_config();
290        config.heartbeat = Some(SocketHeartbeat {
291            interval_secs: 5,
292            payload: b"ping".to_vec(),
293        });
294
295        assert!(config.validate().is_ok());
296    }
297
298    #[rstest]
299    #[case::derived(None, Some(15))]
300    #[case::explicit_wins(Some(20), Some(20))]
301    fn test_resolve_timeout_from_socket_heartbeat(
302        #[case] timeout_secs: Option<u64>,
303        #[case] expected: Option<u64>,
304    ) {
305        let mut config = valid_config();
306        config.heartbeat = Some(SocketHeartbeat {
307            interval_secs: 5,
308            payload: b"ping".to_vec(),
309        });
310        config.heartbeat_timeout_secs = timeout_secs;
311
312        assert_eq!(config.resolved_heartbeat_timeout(), expected);
313    }
314
315    #[rstest]
316    #[case::empty_url(|c: &mut SocketConfig| c.url = String::new(), "url")]
317    #[case::heartbeat_interval(|c: &mut SocketConfig| { c.heartbeat = Some(SocketHeartbeat { interval_secs: 0, payload: vec![] }); }, "heartbeat")]
318    #[case::heartbeat_timeout_below_interval(|c: &mut SocketConfig| { c.heartbeat = Some(SocketHeartbeat { interval_secs: 5, payload: vec![b'p'] }); c.heartbeat_timeout_secs = Some(5); }, "heartbeat_timeout_secs")]
319    #[case::connect_timeout(|c: &mut SocketConfig| c.connect_timeout_ms = Some(0), "connect_timeout_ms")]
320    #[case::reconnect_delay_initial(|c: &mut SocketConfig| c.reconnect_delay_initial_ms = Some(0), "reconnect_delay_initial_ms")]
321    #[case::reconnect_delay_max(|c: &mut SocketConfig| c.reconnect_delay_max_ms = Some(0), "reconnect_delay_max_ms")]
322    #[case::heartbeat_timeout_zero(|c: &mut SocketConfig| c.heartbeat_timeout_secs = Some(0), "heartbeat_timeout_secs")]
323    fn test_validate_rejects_invalid_field(
324        #[case] mutate: fn(&mut SocketConfig),
325        #[case] expected_field: &str,
326    ) {
327        let mut config = valid_config();
328        mutate(&mut config);
329
330        let err = config
331            .validate()
332            .expect_err("invalid value should be rejected");
333
334        assert!(
335            matches!(err, NetworkConfigError::Invalid { field, .. } if field == expected_field)
336        );
337    }
338
339    #[rstest]
340    #[case::too_small(0.5)]
341    #[case::too_large(100.1)]
342    #[case::nan(f64::NAN)]
343    #[case::infinite(f64::INFINITY)]
344    fn test_validate_rejects_invalid_backoff_factor(#[case] factor: f64) {
345        let mut config = valid_config();
346        config.reconnect_backoff_factor = Some(factor);
347
348        let err = config
349            .validate()
350            .expect_err("invalid backoff factor should be rejected");
351
352        assert!(
353            matches!(err, NetworkConfigError::Invalid { field, .. } if field == "reconnect_backoff_factor")
354        );
355    }
356
357    #[rstest]
358    fn test_validate_rejects_delay_initial_exceeding_max() {
359        let mut config = valid_config();
360        config.reconnect_delay_initial_ms = Some(5_000);
361        config.reconnect_delay_max_ms = Some(1_000);
362
363        let err = config
364            .validate()
365            .expect_err("initial delay above max should be rejected");
366
367        assert!(
368            matches!(err, NetworkConfigError::Invalid { field, .. } if field == "reconnect_delay_initial_ms")
369        );
370    }
371
372    #[rstest]
373    fn test_validate_collects_multiple_errors() {
374        let mut config = valid_config();
375        config.url = String::new();
376        config.connect_timeout_ms = Some(0);
377
378        let err = config.validate().expect_err("multiple invalid fields");
379
380        match err {
381            NetworkConfigError::Multiple { errors } => assert_eq!(errors.len(), 2),
382            other @ NetworkConfigError::Invalid { .. } => {
383                panic!("expected Multiple, was {other:?}")
384            }
385        }
386    }
387
388    #[rstest]
389    fn test_debug_redacts_endpoint_credentials() {
390        const ENDPOINT_PATH_SECRET: &str = "unique-endpoint-path-secret";
391        const ENDPOINT_QUERY_SECRET: &str = "unique-endpoint-query-secret";
392        let mut config = valid_config();
393        config.url =
394            format!("wss://rpc.example.com/{ENDPOINT_PATH_SECRET}?api_key={ENDPOINT_QUERY_SECRET}");
395
396        let debug = format!("{config:?}");
397
398        assert!(debug.contains("url: \"<redacted>\""));
399        assert!(!debug.contains(ENDPOINT_PATH_SECRET));
400        assert!(!debug.contains(ENDPOINT_QUERY_SECRET));
401    }
402}
nautilus_network/socket/config.rs

nautilus_network/socket/
config.rs
