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 identity instead of the signer's own key.
174    ///
175    /// The challenge signature is still produced by the configured signer.
176    /// Required with [`Self::new_async`] when the signer holds a delegated
177    /// auth key: the server expects the maker owner's public key here.
178    #[must_use]
179    pub fn with_auth_pubkey(mut self, pubkey: impl Into<String>) -> Self {
180        self.auth_pubkey = pubkey.into();
181        self
182    }
183
184    #[must_use]
185    pub const fn with_endpoint(mut self, endpoint: MakerWsEndpoint) -> Self {
186        self.endpoint = endpoint;
187        self
188    }
189
190    #[must_use]
191    pub fn with_initial_subscribe(mut self, subscribe: SubscribeData) -> Self {
192        self.initial_subscribe = Some(subscribe);
193        self
194    }
195
196    /// Validate task capacities and deadlines before spawning the session.
197    pub fn validate(&self) -> Result<(), ManagedWsConfigError> {
198        if self.url.trim().is_empty() {
199            return Err(ManagedWsConfigError::EmptyUrl);
200        }
201        for (field, capacity) in [
202            ("command_buffer", self.command_buffer),
203            ("broadcast_buffer", self.broadcast_buffer),
204            ("max_pending_awaits", self.max_pending_awaits),
205            ("max_batch_quotes", self.max_batch_quotes),
206            ("max_outbound_message_size", self.max_outbound_message_size),
207        ] {
208            if capacity == 0 {
209                return Err(ManagedWsConfigError::ZeroCapacity { field });
210            }
211        }
212        for (field, duration) in [
213            ("reconnect_delay", self.reconnect_delay),
214            ("max_reconnect_delay", self.max_reconnect_delay),
215            ("connect_timeout", self.connect_timeout),
216            ("ping_interval", self.ping_interval),
217            ("auth_timeout", self.auth_timeout),
218            ("readiness_timeout", self.readiness_timeout),
219            ("write_timeout", self.write_timeout),
220            ("pong_timeout", self.pong_timeout),
221        ] {
222            if duration.is_zero() {
223                return Err(ManagedWsConfigError::ZeroDuration { field });
224            }
225        }
226        if self
227            .ws_read_timeout
228            .is_some_and(|duration| duration.is_zero())
229        {
230            return Err(ManagedWsConfigError::ZeroDuration {
231                field: "ws_read_timeout",
232            });
233        }
234        if self.reconnect_delay > self.max_reconnect_delay {
235            return Err(ManagedWsConfigError::InvalidReconnectRange);
236        }
237        if !(0.0..=1.0).contains(&self.reconnect_jitter_ratio) {
238            return Err(ManagedWsConfigError::InvalidReconnectJitter);
239        }
240        validate_server_limit(
241            "max_batch_quotes",
242            self.max_batch_quotes,
243            SERVER_MAX_BATCH_QUOTES,
244        )?;
245        validate_server_limit(
246            "max_outbound_message_size",
247            self.max_outbound_message_size,
248            SERVER_MAX_CLIENT_MESSAGE_BYTES,
249        )?;
250        if self.endpoint == MakerWsEndpoint::Data && self.initial_subscribe.is_some() {
251            return Err(ManagedWsConfigError::InitialSubscribeOnDataEndpoint);
252        }
253        if let Some(subscribe) = &self.initial_subscribe {
254            let message = ClientMessage::Subscribe(subscribe.clone());
255            let prepared = PreparedClientMessage::new(&message)
256                .map_err(|_| ManagedWsConfigError::InitialSubscribeNotSerializable)?;
257            if prepared.len() > self.max_outbound_message_size {
258                return Err(ManagedWsConfigError::InitialSubscribeTooLarge {
259                    actual: prepared.len(),
260                    limit: self.max_outbound_message_size,
261                });
262            }
263        }
264        self.effective_transport().validate()?;
265        validate_slot_cap(
266            "broadcast_buffer",
267            self.broadcast_buffer,
268            MAX_INBOUND_RING_SLOTS,
269        )?;
270        validate_memory_envelope(
271            "outbound command queue",
272            self.command_buffer,
273            self.max_outbound_message_size,
274            MAX_OUTBOUND_QUEUE_WIRE_BYTES,
275        )
276    }
277
278    pub(super) const fn effective_transport(&self) -> WsTransportConfig {
279        let mut transport = self.transport;
280        transport.connect_timeout = self.connect_timeout;
281        transport
282    }
283}
284
285fn validate_server_limit(
286    field: &'static str,
287    configured: usize,
288    limit: usize,
289) -> Result<(), ManagedWsConfigError> {
290    if configured > limit {
291        return Err(ManagedWsConfigError::ProtocolLimitExceeded {
292            field,
293            configured,
294            limit,
295        });
296    }
297    Ok(())
298}
299
300fn validate_slot_cap(
301    field: &'static str,
302    capacity: usize,
303    limit: usize,
304) -> Result<(), ManagedWsConfigError> {
305    if capacity > limit {
306        return Err(ManagedWsConfigError::CapacityTooLarge {
307            field,
308            configured: capacity,
309            limit,
310        });
311    }
312    Ok(())
313}
314
315/// Every outbound message is rejected above `max_outbound_message_size` before
316/// it is queued, so unlike the inbound ring this product is a true bound on the
317/// bytes the queue can hold.
318fn validate_memory_envelope(
319    queue: &'static str,
320    capacity: usize,
321    max_message_size: usize,
322    limit: usize,
323) -> Result<(), ManagedWsConfigError> {
324    let configured = capacity.saturating_mul(max_message_size);
325    if configured > limit {
326        return Err(ManagedWsConfigError::MemoryEnvelopeTooLarge {
327            queue,
328            configured,
329            limit,
330        });
331    }
332    Ok(())
333}
334
335#[derive(Debug, thiserror::Error, Clone, Copy, PartialEq, Eq)]
336#[non_exhaustive]
337pub enum ManagedWsConfigError {
338    #[error("managed websocket URL must not be empty")]
339    EmptyUrl,
340    #[error("managed websocket capacity `{field}` must be non-zero")]
341    ZeroCapacity { field: &'static str },
342    #[error("managed websocket duration `{field}` must be non-zero")]
343    ZeroDuration { field: &'static str },
344    #[error("maximum reconnect delay must not be shorter than the initial reconnect delay")]
345    InvalidReconnectRange,
346    #[error("reconnect jitter ratio must be in the inclusive range 0.0..=1.0")]
347    InvalidReconnectJitter,
348    #[error("initial subscriptions are not supported by the maker data endpoint")]
349    InitialSubscribeOnDataEndpoint,
350    #[error("initial subscription is {actual} bytes, maximum is {limit}")]
351    InitialSubscribeTooLarge { actual: usize, limit: usize },
352    #[error("{field} is {configured}, server maximum is {limit}")]
353    ProtocolLimitExceeded {
354        field: &'static str,
355        configured: usize,
356        limit: usize,
357    },
358    #[error("initial subscribe cannot be serialized")]
359    InitialSubscribeNotSerializable,
360    #[error("{field} is {configured}, maximum is {limit}")]
361    CapacityTooLarge {
362        field: &'static str,
363        configured: usize,
364        limit: usize,
365    },
366    #[error("{queue} wire envelope is {configured} bytes, maximum is {limit}")]
367    MemoryEnvelopeTooLarge {
368        queue: &'static str,
369        configured: usize,
370        limit: usize,
371    },
372    #[error(transparent)]
373    Transport(#[from] WsTransportConfigError),
374}
acta_maker_sdk/ws/managed/config.rs

acta_maker_sdk/ws/managed/
config.rs
