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

nautilus_network/socket/
config.rs
