acta_maker_sdk/ws/managed/
config.rs

1use std::time::Duration;
2
3use crate::ws::client::{PreparedClientMessage, WsTransportConfig};
4use crate::ws::error::WsTransportConfigError;
5use crate::ws::types::{ClientMessage, FEATURE_CANCEL_ON_DISCONNECT, HelloData, SubscribeData};
6
7use super::endpoint::MakerWsEndpoint;
8use super::signer::ChallengeSigning;
9use crate::orders::SignerLike;
10use std::sync::Arc;
11
12/// What the session does when a message subscriber falls behind the inbound
13/// ring. The subscriber always gets `Gap`; this decides whether the session
14/// also reconnects so a fresh snapshot follows.
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
16#[non_exhaustive]
17pub enum GapPolicy {
18    /// Reconnect quote sessions; only surface gaps on data sessions.
19    #[default]
20    EndpointDefault,
21    Surface,
22    Reconnect,
23}
24
25// Bounds ring slots; retained payload memory depends on traffic.
26const MAX_INBOUND_RING_SLOTS: usize = 65_536;
27const MAX_OUTBOUND_QUEUE_WIRE_BYTES: usize = 256 * 1024 * 1024;
28const SERVER_MAX_BATCH_QUOTES: usize = 50;
29const SERVER_MAX_CLIENT_MESSAGE_BYTES: usize = 32 * 1024;
30
31#[derive(Clone)]
32pub struct ManagedWsConfig {
33    pub url: String,
34    pub endpoint: MakerWsEndpoint,
35    pub(super) auth_pubkey: String,
36    pub(super) challenge_signing: ChallengeSigning,
37    pub(super) initial_subscribe: Option<SubscribeData>,
38    pub reconnect_delay: Duration,
39    pub max_reconnect_delay: Duration,
40    /// Symmetric reconnect jitter ratio in the inclusive range 0.0..=1.0.
41    pub reconnect_jitter_ratio: f64,
42    /// Reconnects after the initial attempt. Zero means retry forever.
43    pub max_reconnect_attempts: u64,
44    /// Managed handshake deadline. Overrides `transport.connect_timeout` for this runtime.
45    pub connect_timeout: Duration,
46    pub ping_interval: Duration,
47    pub auth_timeout: Duration,
48    /// Deadline for initial subscription acknowledgement and recovery reads.
49    /// For live subscription changes, starts after the frame is written.
50    pub readiness_timeout: Duration,
51    pub write_timeout: Duration,
52    pub command_buffer: usize,
53    pub broadcast_buffer: usize,
54    /// Maximum number of quotes accepted in one outbound batch.
55    pub max_batch_quotes: usize,
56    /// Maximum serialized size of one outbound client message.
57    pub max_outbound_message_size: usize,
58    pub ws_read_timeout: Option<Duration>,
59    pub pong_timeout: Duration,
60    pub transport: WsTransportConfig,
61    pub max_pending_awaits: usize,
62    /// Ask the server to cancel this session's open quotes when it
63    /// disconnects; the session terminates if the server does not enable it.
64    /// Applies only to the quote endpoint.
65    pub cancel_on_disconnect: bool,
66    pub gap_policy: GapPolicy,
67    pub(super) hello: HelloData,
68}
69
70impl ManagedWsConfig {
71    /// Authenticate with a local key. The signer's own public key is the maker
72    /// identity unless [`Self::with_auth_pubkey`] names a delegating owner.
73    #[must_use]
74    pub fn new(
75        url: impl Into<String>,
76        hello: HelloData,
77        signer: Arc<dyn SignerLike + Send + Sync>,
78    ) -> Self {
79        let auth_pubkey = signer.pubkey_base58();
80        Self::new_with_signing(url, hello, auth_pubkey, ChallengeSigning::Local(signer))
81    }
82
83    pub(super) fn new_with_signing(
84        url: impl Into<String>,
85        hello: HelloData,
86        auth_pubkey: String,
87        challenge_signing: ChallengeSigning,
88    ) -> Self {
89        Self {
90            url: url.into(),
91            endpoint: MakerWsEndpoint::Quote,
92            auth_pubkey,
93            challenge_signing,
94            initial_subscribe: None,
95            reconnect_delay: Duration::from_millis(250),
96            max_reconnect_delay: Duration::from_secs(5),
97            reconnect_jitter_ratio: 0.2,
98            max_reconnect_attempts: 0,
99            connect_timeout: Duration::from_secs(10),
100            ping_interval: Duration::from_secs(30),
101            auth_timeout: Duration::from_secs(15),
102            readiness_timeout: Duration::from_secs(15),
103            write_timeout: Duration::from_secs(5),
104            command_buffer: 256,
105            broadcast_buffer: 64,
106            max_batch_quotes: SERVER_MAX_BATCH_QUOTES,
107            max_outbound_message_size: SERVER_MAX_CLIENT_MESSAGE_BYTES,
108            ws_read_timeout: None,
109            pong_timeout: Duration::from_secs(10),
110            transport: WsTransportConfig::default(),
111            max_pending_awaits: 1024,
112            cancel_on_disconnect: true,
113            gap_policy: GapPolicy::EndpointDefault,
114            hello,
115        }
116    }
117
118    /// Shorter liveness deadlines, larger queues, and reconnect on inbound gaps.
119    #[must_use]
120    pub const fn low_latency(mut self) -> Self {
121        self.ping_interval = Duration::from_secs(2);
122        self.pong_timeout = Duration::from_secs(2);
123        self.write_timeout = Duration::from_secs(1);
124        self.command_buffer = 1024;
125        self.broadcast_buffer = 4096;
126        self.gap_policy = GapPolicy::Reconnect;
127        self
128    }
129
130    #[must_use]
131    pub const fn with_gap_policy(mut self, policy: GapPolicy) -> Self {
132        self.gap_policy = policy;
133        self
134    }
135
136    #[must_use]
137    pub const fn with_cancel_on_disconnect(mut self, enabled: bool) -> Self {
138        self.cancel_on_disconnect = enabled;
139        self
140    }
141
142    pub(super) fn reconnect_on_gap(&self) -> bool {
143        match self.gap_policy {
144            GapPolicy::EndpointDefault => self.endpoint == MakerWsEndpoint::Quote,
145            GapPolicy::Surface => false,
146            GapPolicy::Reconnect => true,
147        }
148    }
149
150    /// `Hello` with every feature this configuration requires.
151    pub(super) fn hello(&self) -> HelloData {
152        let mut hello = self.hello.clone();
153        if self.endpoint == MakerWsEndpoint::Quote
154            && self.cancel_on_disconnect
155            && !hello
156                .features
157                .iter()
158                .any(|f| f == FEATURE_CANCEL_ON_DISCONNECT)
159        {
160            hello.features.push(FEATURE_CANCEL_ON_DISCONNECT.to_owned());
161        }
162        hello
163    }
164
165    /// A required feature the server's `Welcome` did not enable.
166    pub(super) fn missing_required_feature(&self, enabled: &[String]) -> Option<&'static str> {
167        (self.endpoint == MakerWsEndpoint::Quote
168            && self.cancel_on_disconnect
169            && !enabled.iter().any(|f| f == FEATURE_CANCEL_ON_DISCONNECT))
170        .then_some(FEATURE_CANCEL_ON_DISCONNECT)
171    }
172
173    /// Authenticate as this maker owner instead of the signer's public key.
174    ///
175    /// The configured signer signs the challenge. Set this for delegated auth keys.
176    #[must_use]
177    pub fn with_auth_pubkey(mut self, pubkey: impl Into<String>) -> Self {
178        self.auth_pubkey = pubkey.into();
179        self
180    }
181
182    #[must_use]
183    pub const fn with_endpoint(mut self, endpoint: MakerWsEndpoint) -> Self {
184        self.endpoint = endpoint;
185        self
186    }
187
188    #[must_use]
189    pub fn with_initial_subscribe(mut self, subscribe: SubscribeData) -> Self {
190        self.initial_subscribe = Some(subscribe);
191        self
192    }
193
194    /// Validate task capacities and deadlines before spawning the session.
195    pub fn validate(&self) -> Result<(), ManagedWsConfigError> {
196        if self.url.trim().is_empty() {
197            return Err(ManagedWsConfigError::EmptyUrl);
198        }
199        for (field, capacity) in [
200            ("command_buffer", self.command_buffer),
201            ("broadcast_buffer", self.broadcast_buffer),
202            ("max_pending_awaits", self.max_pending_awaits),
203            ("max_batch_quotes", self.max_batch_quotes),
204            ("max_outbound_message_size", self.max_outbound_message_size),
205        ] {
206            if capacity == 0 {
207                return Err(ManagedWsConfigError::ZeroCapacity { field });
208            }
209        }
210        for (field, duration) in [
211            ("reconnect_delay", self.reconnect_delay),
212            ("max_reconnect_delay", self.max_reconnect_delay),
213            ("connect_timeout", self.connect_timeout),
214            ("ping_interval", self.ping_interval),
215            ("auth_timeout", self.auth_timeout),
216            ("readiness_timeout", self.readiness_timeout),
217            ("write_timeout", self.write_timeout),
218            ("pong_timeout", self.pong_timeout),
219        ] {
220            if duration.is_zero() {
221                return Err(ManagedWsConfigError::ZeroDuration { field });
222            }
223        }
224        if self
225            .ws_read_timeout
226            .is_some_and(|duration| duration.is_zero())
227        {
228            return Err(ManagedWsConfigError::ZeroDuration {
229                field: "ws_read_timeout",
230            });
231        }
232        if self.reconnect_delay > self.max_reconnect_delay {
233            return Err(ManagedWsConfigError::InvalidReconnectRange);
234        }
235        if !(0.0..=1.0).contains(&self.reconnect_jitter_ratio) {
236            return Err(ManagedWsConfigError::InvalidReconnectJitter);
237        }
238        validate_server_limit(
239            "max_batch_quotes",
240            self.max_batch_quotes,
241            SERVER_MAX_BATCH_QUOTES,
242        )?;
243        validate_server_limit(
244            "max_outbound_message_size",
245            self.max_outbound_message_size,
246            SERVER_MAX_CLIENT_MESSAGE_BYTES,
247        )?;
248        if self.endpoint == MakerWsEndpoint::Data && self.initial_subscribe.is_some() {
249            return Err(ManagedWsConfigError::InitialSubscribeOnDataEndpoint);
250        }
251        if let Some(subscribe) = &self.initial_subscribe {
252            let message = ClientMessage::Subscribe(subscribe.clone());
253            let prepared = PreparedClientMessage::new(&message)
254                .map_err(|_| ManagedWsConfigError::InitialSubscribeNotSerializable)?;
255            if prepared.len() > self.max_outbound_message_size {
256                return Err(ManagedWsConfigError::InitialSubscribeTooLarge {
257                    actual: prepared.len(),
258                    limit: self.max_outbound_message_size,
259                });
260            }
261        }
262        self.effective_transport().validate()?;
263        validate_slot_cap(
264            "broadcast_buffer",
265            self.broadcast_buffer,
266            MAX_INBOUND_RING_SLOTS,
267        )?;
268        validate_memory_envelope(
269            "outbound command queue",
270            self.command_buffer,
271            self.max_outbound_message_size,
272            MAX_OUTBOUND_QUEUE_WIRE_BYTES,
273        )
274    }
275
276    pub(super) const fn effective_transport(&self) -> WsTransportConfig {
277        let mut transport = self.transport;
278        transport.connect_timeout = self.connect_timeout;
279        transport
280    }
281}
282
283fn validate_server_limit(
284    field: &'static str,
285    configured: usize,
286    limit: usize,
287) -> Result<(), ManagedWsConfigError> {
288    if configured > limit {
289        return Err(ManagedWsConfigError::ProtocolLimitExceeded {
290            field,
291            configured,
292            limit,
293        });
294    }
295    Ok(())
296}
297
298fn validate_slot_cap(
299    field: &'static str,
300    capacity: usize,
301    limit: usize,
302) -> Result<(), ManagedWsConfigError> {
303    if capacity > limit {
304        return Err(ManagedWsConfigError::CapacityTooLarge {
305            field,
306            configured: capacity,
307            limit,
308        });
309    }
310    Ok(())
311}
312
313/// Every outbound message is rejected above `max_outbound_message_size` before
314/// it is queued, so unlike the inbound ring this product is a true bound on the
315/// bytes the queue can hold.
316fn validate_memory_envelope(
317    queue: &'static str,
318    capacity: usize,
319    max_message_size: usize,
320    limit: usize,
321) -> Result<(), ManagedWsConfigError> {
322    let configured = capacity.saturating_mul(max_message_size);
323    if configured > limit {
324        return Err(ManagedWsConfigError::MemoryEnvelopeTooLarge {
325            queue,
326            configured,
327            limit,
328        });
329    }
330    Ok(())
331}
332
333#[derive(Debug, thiserror::Error, Clone, Copy, PartialEq, Eq)]
334#[non_exhaustive]
335pub enum ManagedWsConfigError {
336    #[error("managed websocket URL must not be empty")]
337    EmptyUrl,
338    #[error("managed websocket capacity `{field}` must be non-zero")]
339    ZeroCapacity { field: &'static str },
340    #[error("managed websocket duration `{field}` must be non-zero")]
341    ZeroDuration { field: &'static str },
342    #[error("maximum reconnect delay must not be shorter than the initial reconnect delay")]
343    InvalidReconnectRange,
344    #[error("reconnect jitter ratio must be in the inclusive range 0.0..=1.0")]
345    InvalidReconnectJitter,
346    #[error("initial subscriptions are not supported by the maker data endpoint")]
347    InitialSubscribeOnDataEndpoint,
348    #[error("initial subscription is {actual} bytes, maximum is {limit}")]
349    InitialSubscribeTooLarge { actual: usize, limit: usize },
350    #[error("{field} is {configured}, server maximum is {limit}")]
351    ProtocolLimitExceeded {
352        field: &'static str,
353        configured: usize,
354        limit: usize,
355    },
356    #[error("initial subscribe cannot be serialized")]
357    InitialSubscribeNotSerializable,
358    #[error("{field} is {configured}, maximum is {limit}")]
359    CapacityTooLarge {
360        field: &'static str,
361        configured: usize,
362        limit: usize,
363    },
364    #[error("{queue} wire envelope is {configured} bytes, maximum is {limit}")]
365    MemoryEnvelopeTooLarge {
366        queue: &'static str,
367        configured: usize,
368        limit: usize,
369    },
370    #[error(transparent)]
371    Transport(#[from] WsTransportConfigError),
372}
acta_maker_sdk/ws/managed/config.rs

acta_maker_sdk/ws/managed/
config.rs
