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        if let Err(e) = crate::writer::validate_capacity(self.writer_capacity) {
156            errors.push(e);
157        }
158
159        if self.url.trim().is_empty() {
160            errors.push(NetworkConfigError::invalid("url", "must not be empty"));
161        }
162
163        if let Some(heartbeat) = &self.heartbeat
164            && heartbeat.interval_secs == 0
165        {
166            errors.push(NetworkConfigError::invalid(
167                "heartbeat",
168                "interval must be positive",
169            ));
170        }
171
172        // A timeout at or below the send cadence tears every connection down before its first
173        // reply is due, so a healthy socket would reconnect forever.
174        if let (Some(heartbeat), Some(timeout_secs)) =
175            (&self.heartbeat, self.heartbeat_timeout_secs)
176            && timeout_secs <= heartbeat.interval_secs
177        {
178            errors.push(NetworkConfigError::invalid(
179                "heartbeat_timeout_secs",
180                format!(
181                    "must exceed heartbeat interval ({}s), was {timeout_secs}s",
182                    heartbeat.interval_secs
183                ),
184            ));
185        }
186
187        // `reconnect_jitter_ms` is intentionally unchecked: zero disables jitter and
188        // `ExponentialBackoff::new` accepts it.
189        for (field, value) in [
190            ("connect_timeout_ms", self.connect_timeout_ms),
191            (
192                "reconnect_delay_initial_ms",
193                self.reconnect_delay_initial_ms,
194            ),
195            ("reconnect_delay_max_ms", self.reconnect_delay_max_ms),
196            ("heartbeat_timeout_secs", self.heartbeat_timeout_secs),
197        ] {
198            if let Some(value) = value
199                && value == 0
200            {
201                errors.push(NetworkConfigError::invalid(
202                    field,
203                    format!("must be positive, was {value}"),
204                ));
205            }
206        }
207
208        if let Some(factor) = self.reconnect_backoff_factor
209            && !(1.0..=100.0).contains(&factor)
210        {
211            errors.push(NetworkConfigError::invalid(
212                "reconnect_backoff_factor",
213                format!("must be in range [1.0, 100.0], was {factor}"),
214            ));
215        }
216
217        if let (Some(initial), Some(max)) =
218            (self.reconnect_delay_initial_ms, self.reconnect_delay_max_ms)
219            && initial > max
220        {
221            errors.push(NetworkConfigError::invalid(
222                "reconnect_delay_initial_ms",
223                format!("must not exceed reconnect_delay_max_ms ({max}), was {initial}"),
224            ));
225        }
226
227        NetworkConfigError::collect(errors)
228    }
229
230    pub(crate) fn resolved_heartbeat_timeout(&self) -> Option<u64> {
231        crate::heartbeat::resolve_heartbeat_timeout(
232            self.heartbeat_timeout_secs,
233            self.heartbeat
234                .as_ref()
235                .map(|heartbeat| heartbeat.interval_secs),
236        )
237    }
238}
239
240impl Debug for SocketConfig {
241    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
242        f.debug_struct(stringify!(SocketConfig))
243            .field("url", &REDACTED)
244            .field("mode", &self.mode)
245            .field("suffix", &self.suffix)
246            .field(
247                "message_handler",
248                &self.message_handler.as_ref().map(|_| "<function>"),
249            )
250            .field("heartbeat", &self.heartbeat)
251            .field("writer_capacity", &self.writer_capacity)
252            .field("connect_timeout_ms", &self.connect_timeout_ms)
253            .field(
254                "reconnect_delay_initial_ms",
255                &self.reconnect_delay_initial_ms,
256            )
257            .field("reconnect_delay_max_ms", &self.reconnect_delay_max_ms)
258            .field("reconnect_backoff_factor", &self.reconnect_backoff_factor)
259            .field("reconnect_jitter_ms", &self.reconnect_jitter_ms)
260            .field("connection_max_retries", &self.connection_max_retries)
261            .field("reconnect_max_attempts", &self.reconnect_max_attempts)
262            .field("heartbeat_timeout_secs", &self.heartbeat_timeout_secs)
263            .field("certs_dir", &self.certs_dir)
264            .finish()
265    }
266}
267
268#[cfg(test)]
269mod tests {
270    use rstest::rstest;
271    use tokio_tungstenite::tungstenite::stream::Mode;
272
273    use super::{SocketConfig, SocketHeartbeat};
274    use crate::error::NetworkConfigError;
275
276    fn valid_config() -> SocketConfig {
277        SocketConfig::builder()
278            .url("tcp://127.0.0.1:8080".to_string())
279            .mode(Mode::Plain)
280            .suffix(vec![b'\n'])
281            .build()
282            .expect("baseline socket config should be valid")
283    }
284
285    #[rstest]
286    fn test_builder_accepts_valid_config() {
287        let result = SocketConfig::builder()
288            .url("tcp://127.0.0.1:8080".to_string())
289            .mode(Mode::Plain)
290            .suffix(vec![b'\n'])
291            .build();
292
293        assert!(result.is_ok());
294    }
295
296    #[rstest]
297    fn test_validate_accepts_zero_jitter() {
298        let mut config = valid_config();
299        config.reconnect_jitter_ms = Some(0);
300
301        assert!(config.validate().is_ok());
302    }
303
304    #[rstest]
305    fn test_validate_accepts_heartbeat_with_payload() {
306        let mut config = valid_config();
307        config.heartbeat = Some(SocketHeartbeat {
308            interval_secs: 5,
309            payload: b"ping".to_vec(),
310        });
311
312        assert!(config.validate().is_ok());
313    }
314
315    #[rstest]
316    #[case::derived(None, Some(15))]
317    #[case::explicit_wins(Some(20), Some(20))]
318    fn test_resolve_timeout_from_socket_heartbeat(
319        #[case] timeout_secs: Option<u64>,
320        #[case] expected: Option<u64>,
321    ) {
322        let mut config = valid_config();
323        config.heartbeat = Some(SocketHeartbeat {
324            interval_secs: 5,
325            payload: b"ping".to_vec(),
326        });
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
