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 maximum number of outstanding messages, including queued, in-flight, and replay messages.
115    ///
116    /// Defaults to 1,024. Sends fail immediately when full. Must be positive and no greater
117    /// than [`tokio::sync::Semaphore::MAX_PERMITS`]. This bounds message count, not payload bytes.
118    pub writer_capacity: Option<usize>,
119    /// The path to the certificates directory.
120    ///
121    /// Every certificate that parses from any file directly inside the directory, regardless of
122    /// file name, becomes a trust anchor for the connection alongside the bundled `webpki-roots`
123    /// set. No CA or self-signature checks are performed. A certificate chain matched to a
124    /// private key in the directory is used as the client certificate instead. Write access to
125    /// the directory is therefore equivalent to control over which servers the connection
126    /// trusts. Restrict its contents to intended trust anchors. Each anchor loaded from it is
127    /// logged at INFO with its SHA-256 fingerprint.
128    pub certs_dir: Option<String>,
129}
130
131impl<S: socket_config_builder::IsComplete> SocketConfigBuilder<S> {
132    /// Validates and builds the [`SocketConfig`].
133    ///
134    /// # Errors
135    ///
136    /// Returns an error if any field fails validation (see [`SocketConfig::validate`]).
137    pub fn build(self) -> NetworkConfigResult<SocketConfig> {
138        let config = self.build_inner();
139        config.validate()?;
140        Ok(config)
141    }
142}
143
144impl SocketConfig {
145    /// Checks whether all socket settings are valid.
146    ///
147    /// # Errors
148    ///
149    /// Returns an error if `url` is empty, the heartbeat interval or a reconnection timing field is
150    /// not positive, `reconnect_backoff_factor` is outside `[1.0, 100.0]`,
151    /// `reconnect_delay_initial_ms` exceeds `reconnect_delay_max_ms`, or `writer_capacity` is
152    /// outside `[1, Semaphore::MAX_PERMITS]`.
153    pub fn validate(&self) -> NetworkConfigResult<()> {
154        let mut errors = Vec::new();
155
156        if let Err(e) = crate::writer::validate_capacity(self.writer_capacity) {
157            errors.push(e);
158        }
159
160        if self.url.trim().is_empty() {
161            errors.push(NetworkConfigError::invalid("url", "must not be empty"));
162        }
163
164        if let Some(heartbeat) = &self.heartbeat
165            && heartbeat.interval_secs == 0
166        {
167            errors.push(NetworkConfigError::invalid(
168                "heartbeat",
169                "interval must be positive",
170            ));
171        }
172
173        // A timeout at or below the send cadence tears every connection down before its first
174        // reply is due, so a healthy socket would reconnect forever.
175        if let (Some(heartbeat), Some(timeout_secs)) =
176            (&self.heartbeat, self.heartbeat_timeout_secs)
177            && timeout_secs <= heartbeat.interval_secs
178        {
179            errors.push(NetworkConfigError::invalid(
180                "heartbeat_timeout_secs",
181                format!(
182                    "must exceed heartbeat interval ({}s), was {timeout_secs}s",
183                    heartbeat.interval_secs
184                ),
185            ));
186        }
187
188        // `reconnect_jitter_ms` is intentionally unchecked: zero disables jitter and
189        // `ExponentialBackoff::new` accepts it.
190        for (field, value) in [
191            ("connect_timeout_ms", self.connect_timeout_ms),
192            (
193                "reconnect_delay_initial_ms",
194                self.reconnect_delay_initial_ms,
195            ),
196            ("reconnect_delay_max_ms", self.reconnect_delay_max_ms),
197            ("heartbeat_timeout_secs", self.heartbeat_timeout_secs),
198        ] {
199            if let Some(value) = value
200                && value == 0
201            {
202                errors.push(NetworkConfigError::invalid(
203                    field,
204                    format!("must be positive, was {value}"),
205                ));
206            }
207        }
208
209        if let Some(factor) = self.reconnect_backoff_factor
210            && !(1.0..=100.0).contains(&factor)
211        {
212            errors.push(NetworkConfigError::invalid(
213                "reconnect_backoff_factor",
214                format!("must be in range [1.0, 100.0], was {factor}"),
215            ));
216        }
217
218        if let (Some(initial), Some(max)) =
219            (self.reconnect_delay_initial_ms, self.reconnect_delay_max_ms)
220            && initial > max
221        {
222            errors.push(NetworkConfigError::invalid(
223                "reconnect_delay_initial_ms",
224                format!("must not exceed reconnect_delay_max_ms ({max}), was {initial}"),
225            ));
226        }
227
228        NetworkConfigError::collect(errors)
229    }
230
231    pub(crate) fn resolved_heartbeat_timeout(&self) -> Option<u64> {
232        crate::heartbeat::resolve_heartbeat_timeout(
233            self.heartbeat_timeout_secs,
234            self.heartbeat
235                .as_ref()
236                .map(|heartbeat| heartbeat.interval_secs),
237        )
238    }
239}
240
241impl Debug for SocketConfig {
242    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
243        f.debug_struct(stringify!(SocketConfig))
244            .field("url", &REDACTED)
245            .field("mode", &self.mode)
246            .field("suffix", &self.suffix)
247            .field(
248                "message_handler",
249                &self.message_handler.as_ref().map(|_| "<function>"),
250            )
251            .field("heartbeat", &self.heartbeat)
252            .field("writer_capacity", &self.writer_capacity)
253            .field("connect_timeout_ms", &self.connect_timeout_ms)
254            .field(
255                "reconnect_delay_initial_ms",
256                &self.reconnect_delay_initial_ms,
257            )
258            .field("reconnect_delay_max_ms", &self.reconnect_delay_max_ms)
259            .field("reconnect_backoff_factor", &self.reconnect_backoff_factor)
260            .field("reconnect_jitter_ms", &self.reconnect_jitter_ms)
261            .field("connection_max_retries", &self.connection_max_retries)
262            .field("reconnect_max_attempts", &self.reconnect_max_attempts)
263            .field("heartbeat_timeout_secs", &self.heartbeat_timeout_secs)
264            .field("certs_dir", &self.certs_dir)
265            .finish()
266    }
267}
268
269#[cfg(test)]
270mod tests {
271    use rstest::rstest;
272    use tokio_tungstenite::tungstenite::stream::Mode;
273
274    use super::{SocketConfig, SocketHeartbeat};
275    use crate::error::NetworkConfigError;
276
277    fn valid_config() -> SocketConfig {
278        SocketConfig::builder()
279            .url("tcp://127.0.0.1:8080".to_string())
280            .mode(Mode::Plain)
281            .suffix(vec![b'\n'])
282            .build()
283            .expect("baseline socket config should be valid")
284    }
285
286    #[rstest]
287    fn test_builder_accepts_valid_config() {
288        let result = SocketConfig::builder()
289            .url("tcp://127.0.0.1:8080".to_string())
290            .mode(Mode::Plain)
291            .suffix(vec![b'\n'])
292            .build();
293
294        assert!(result.is_ok());
295    }
296
297    #[rstest]
298    fn test_validate_accepts_zero_jitter() {
299        let mut config = valid_config();
300        config.reconnect_jitter_ms = Some(0);
301
302        assert!(config.validate().is_ok());
303    }
304
305    #[rstest]
306    fn test_validate_accepts_heartbeat_with_payload() {
307        let mut config = valid_config();
308        config.heartbeat = Some(SocketHeartbeat {
309            interval_secs: 5,
310            payload: b"ping".to_vec(),
311        });
312
313        assert!(config.validate().is_ok());
314    }
315
316    #[rstest]
317    #[case::derived(None, Some(15))]
318    #[case::explicit_wins(Some(20), Some(20))]
319    fn test_resolve_timeout_from_socket_heartbeat(
320        #[case] timeout_secs: Option<u64>,
321        #[case] expected: Option<u64>,
322    ) {
323        let mut config = valid_config();
324        config.heartbeat = Some(SocketHeartbeat {
325            interval_secs: 5,
326            payload: b"ping".to_vec(),
327        });
328        config.heartbeat_timeout_secs = timeout_secs;
329
330        assert_eq!(config.resolved_heartbeat_timeout(), expected);
331    }
332
333    #[rstest]
334    #[case::writer_capacity_zero(|c: &mut SocketConfig| c.writer_capacity = Some(0), "writer_capacity")]
335    #[case::writer_capacity_excess(|c: &mut SocketConfig| c.writer_capacity = Some(usize::MAX), "writer_capacity")]
336    #[case::empty_url(|c: &mut SocketConfig| c.url = String::new(), "url")]
337    #[case::heartbeat_interval(|c: &mut SocketConfig| { c.heartbeat = Some(SocketHeartbeat { interval_secs: 0, payload: vec![] }); }, "heartbeat")]
338    #[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")]
339    #[case::connect_timeout(|c: &mut SocketConfig| c.connect_timeout_ms = Some(0), "connect_timeout_ms")]
340    #[case::reconnect_delay_initial(|c: &mut SocketConfig| c.reconnect_delay_initial_ms = Some(0), "reconnect_delay_initial_ms")]
341    #[case::reconnect_delay_max(|c: &mut SocketConfig| c.reconnect_delay_max_ms = Some(0), "reconnect_delay_max_ms")]
342    #[case::heartbeat_timeout_zero(|c: &mut SocketConfig| c.heartbeat_timeout_secs = Some(0), "heartbeat_timeout_secs")]
343    fn test_validate_rejects_invalid_field(
344        #[case] mutate: fn(&mut SocketConfig),
345        #[case] expected_field: &str,
346    ) {
347        let mut config = valid_config();
348        mutate(&mut config);
349
350        let err = config
351            .validate()
352            .expect_err("invalid value should be rejected");
353
354        assert!(
355            matches!(err, NetworkConfigError::Invalid { field, .. } if field == expected_field)
356        );
357    }
358
359    #[rstest]
360    #[case::too_small(0.5)]
361    #[case::too_large(100.1)]
362    #[case::nan(f64::NAN)]
363    #[case::infinite(f64::INFINITY)]
364    fn test_validate_rejects_invalid_backoff_factor(#[case] factor: f64) {
365        let mut config = valid_config();
366        config.reconnect_backoff_factor = Some(factor);
367
368        let err = config
369            .validate()
370            .expect_err("invalid backoff factor should be rejected");
371
372        assert!(
373            matches!(err, NetworkConfigError::Invalid { field, .. } if field == "reconnect_backoff_factor")
374        );
375    }
376
377    #[rstest]
378    fn test_validate_rejects_delay_initial_exceeding_max() {
379        let mut config = valid_config();
380        config.reconnect_delay_initial_ms = Some(5_000);
381        config.reconnect_delay_max_ms = Some(1_000);
382
383        let err = config
384            .validate()
385            .expect_err("initial delay above max should be rejected");
386
387        assert!(
388            matches!(err, NetworkConfigError::Invalid { field, .. } if field == "reconnect_delay_initial_ms")
389        );
390    }
391
392    #[rstest]
393    fn test_validate_collects_multiple_errors() {
394        let mut config = valid_config();
395        config.url = String::new();
396        config.connect_timeout_ms = Some(0);
397
398        let err = config.validate().expect_err("multiple invalid fields");
399
400        match err {
401            NetworkConfigError::Multiple { errors } => assert_eq!(errors.len(), 2),
402            other @ NetworkConfigError::Invalid { .. } => {
403                panic!("expected Multiple, was {other:?}")
404            }
405        }
406    }
407
408    #[rstest]
409    fn test_debug_redacts_endpoint_credentials() {
410        const ENDPOINT_PATH_SECRET: &str = "unique-endpoint-path-secret";
411        const ENDPOINT_QUERY_SECRET: &str = "unique-endpoint-query-secret";
412        let mut config = valid_config();
413        config.url =
414            format!("wss://rpc.example.com/{ENDPOINT_PATH_SECRET}?api_key={ENDPOINT_QUERY_SECRET}");
415
416        let debug = format!("{config:?}");
417
418        assert!(debug.contains("url: \"<redacted>\""));
419        assert!(!debug.contains(ENDPOINT_PATH_SECRET));
420        assert!(!debug.contains(ENDPOINT_QUERY_SECRET));
421    }
422}
nautilus_network/socket/config.rs

nautilus_network/socket/
config.rs
