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

acta_maker_sdk/ws/managed/
config.rs
