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#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
14#[non_exhaustive]
15pub enum GapPolicy {
16    /// Reconnect quote sessions; only surface gaps on data sessions.
17    #[default]
18    EndpointDefault,
19    Surface,
20    Reconnect,
21}
22
23const MAX_INBOUND_RING_SLOTS: usize = 65_536;
24const MAX_OUTBOUND_QUEUE_WIRE_BYTES: usize = 256 * 1024 * 1024;
25const SERVER_MAX_BATCH_QUOTES: usize = 50;
26const SERVER_MAX_CLIENT_MESSAGE_BYTES: usize = 32 * 1024;
27
28/// How the session proves its identity after `Welcome`.
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub(crate) enum AuthFlow {
31    /// `StartAuth` / `ResumeAuth` as a maker.
32    Maker,
33    /// `StartVaultAuth` as a vault operator. The server issues no resumable
34    /// session, so every reconnect re-runs the challenge.
35    #[cfg_attr(not(feature = "vault"), allow(dead_code))]
36    Vault { vault_pda: String },
37}
38
39#[derive(Clone)]
40pub struct ManagedWsConfig {
41    pub url: String,
42    pub endpoint: MakerWsEndpoint,
43    pub(super) auth_pubkey: String,
44    pub(super) challenge_signing: ChallengeSigning,
45    pub(crate) auth_flow: AuthFlow,
46    pub(super) initial_subscribe: Option<SubscribeData>,
47    pub reconnect_delay: Duration,
48    pub max_reconnect_delay: Duration,
49    /// Symmetric reconnect jitter ratio in the inclusive range 0.0..=1.0.
50    pub reconnect_jitter_ratio: f64,
51    /// Reconnects after the initial attempt. Zero means retry forever.
52    pub max_reconnect_attempts: u64,
53    /// Managed handshake deadline. Overrides `transport.connect_timeout` for this runtime.
54    pub connect_timeout: Duration,
55    pub ping_interval: Duration,
56    pub auth_timeout: Duration,
57    /// Deadline for initial subscription acknowledgement and recovery reads.
58    /// For live subscription changes, starts after the frame is written.
59    pub readiness_timeout: Duration,
60    pub write_timeout: Duration,
61    pub command_buffer: usize,
62    pub broadcast_buffer: usize,
63    /// Maximum number of quotes accepted in one outbound batch.
64    pub max_batch_quotes: usize,
65    /// Maximum serialized size of one outbound client message.
66    pub max_outbound_message_size: usize,
67    pub ws_read_timeout: Option<Duration>,
68    pub pong_timeout: Duration,
69    pub transport: WsTransportConfig,
70    pub max_pending_awaits: usize,
71    /// Ask the server to cancel this session's open quotes when it
72    /// disconnects; the session terminates if the server does not enable it.
73    /// Applies only to the quote endpoint.
74    pub cancel_on_disconnect: bool,
75    pub gap_policy: GapPolicy,
76    pub(super) hello: HelloData,
77}
78
79impl ManagedWsConfig {
80    /// Authenticate with a local key. The signer's own public key is the maker
81    /// identity unless [`Self::with_auth_pubkey`] names a delegating owner.
82    #[must_use]
83    pub fn new(
84        url: impl Into<String>,
85        hello: HelloData,
86        signer: Arc<dyn SignerLike + Send + Sync>,
87    ) -> Self {
88        let auth_pubkey = signer.pubkey_base58();
89        Self::new_with_signing(url, hello, auth_pubkey, ChallengeSigning::Local(signer))
90    }
91
92    pub(crate) fn new_with_signing(
93        url: impl Into<String>,
94        hello: HelloData,
95        auth_pubkey: String,
96        challenge_signing: ChallengeSigning,
97    ) -> Self {
98        Self {
99            url: url.into(),
100            endpoint: MakerWsEndpoint::Quote,
101            auth_pubkey,
102            challenge_signing,
103            auth_flow: AuthFlow::Maker,
104            initial_subscribe: None,
105            reconnect_delay: Duration::from_millis(250),
106            max_reconnect_delay: Duration::from_secs(5),
107            reconnect_jitter_ratio: 0.2,
108            max_reconnect_attempts: 0,
109            connect_timeout: Duration::from_secs(10),
110            ping_interval: Duration::from_secs(30),
111            auth_timeout: Duration::from_secs(15),
112            readiness_timeout: Duration::from_secs(15),
113            write_timeout: Duration::from_secs(5),
114            command_buffer: 256,
115            broadcast_buffer: 64,
116            max_batch_quotes: SERVER_MAX_BATCH_QUOTES,
117            max_outbound_message_size: SERVER_MAX_CLIENT_MESSAGE_BYTES,
118            ws_read_timeout: None,
119            pong_timeout: Duration::from_secs(10),
120            transport: WsTransportConfig::default(),
121            max_pending_awaits: 1024,
122            cancel_on_disconnect: true,
123            gap_policy: GapPolicy::EndpointDefault,
124            hello,
125        }
126    }
127
128    /// Short liveness deadlines, larger buffers and reconnect on inbound gaps.
129    #[must_use]
130    pub const fn low_latency(mut self) -> Self {
131        self.ping_interval = Duration::from_secs(2);
132        self.pong_timeout = Duration::from_secs(2);
133        self.write_timeout = Duration::from_secs(1);
134        self.command_buffer = 1024;
135        self.broadcast_buffer = 4096;
136        self.gap_policy = GapPolicy::Reconnect;
137        self
138    }
139
140    #[must_use]
141    pub const fn with_gap_policy(mut self, policy: GapPolicy) -> Self {
142        self.gap_policy = policy;
143        self
144    }
145
146    #[must_use]
147    pub const fn with_cancel_on_disconnect(mut self, enabled: bool) -> Self {
148        self.cancel_on_disconnect = enabled;
149        self
150    }
151
152    pub(super) fn reconnect_on_gap(&self) -> bool {
153        match self.gap_policy {
154            GapPolicy::EndpointDefault => self.endpoint == MakerWsEndpoint::Quote,
155            GapPolicy::Surface => false,
156            GapPolicy::Reconnect => true,
157        }
158    }
159
160    /// `Hello` with every feature this configuration requires.
161    pub(super) fn hello(&self) -> HelloData {
162        let mut hello = self.hello.clone();
163        if self.endpoint == MakerWsEndpoint::Quote
164            && self.cancel_on_disconnect
165            && !hello
166                .features
167                .iter()
168                .any(|f| f == FEATURE_CANCEL_ON_DISCONNECT)
169        {
170            hello.features.push(FEATURE_CANCEL_ON_DISCONNECT.to_owned());
171        }
172        hello
173    }
174
175    /// A required feature the server's `Welcome` did not enable.
176    pub(super) fn missing_required_feature(&self, enabled: &[String]) -> Option<&'static str> {
177        (self.endpoint == MakerWsEndpoint::Quote
178            && self.cancel_on_disconnect
179            && !enabled.iter().any(|f| f == FEATURE_CANCEL_ON_DISCONNECT))
180        .then_some(FEATURE_CANCEL_ON_DISCONNECT)
181    }
182
183    /// Set the maker owner's identity when using a delegated auth signer.
184    /// The configured signer still signs the challenge.
185    #[must_use]
186    pub fn with_auth_pubkey(mut self, pubkey: impl Into<String>) -> Self {
187        self.auth_pubkey = pubkey.into();
188        self
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
