nautilus_bitmex/broadcast/
submitter.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//! Submit request broadcaster for redundant order submission.
17//!
18//! This module provides the [`SubmitBroadcaster`] which fans out submit requests
19//! to multiple HTTP clients in parallel for redundancy. The broadcaster is triggered
20//! when the `SubmitOrder` command contains `params["broadcast_submit_tries"]`.
21//!
22//! Key design patterns:
23//!
24//! - **Dependency injection via traits**: Uses `SubmitExecutor` trait to abstract
25//!   the HTTP client, enabling testing without `#[cfg(test)]` conditional compilation.
26//! - **Trait objects over generics**: Uses `Arc<dyn SubmitExecutor>` to avoid
27//!   generic type parameters on the public API (simpler Python FFI).
28//! - **Short-circuit on first success**: Aborts remaining requests once any client
29//!   succeeds, minimizing latency.
30//! - **Idempotent rejection handling**: Recognizes duplicate clOrdID as expected
31//!   rejections for debug-level logging without noise.
32
33// TODO: Replace boxed futures in `SubmitExecutor` once stable async trait object support
34// lands so we can drop the per-call heap allocation
35
36use std::{
37    fmt::Debug,
38    future::Future,
39    pin::Pin,
40    sync::{
41        Arc,
42        atomic::{AtomicBool, AtomicU64, Ordering},
43    },
44    time::Duration,
45};
46
47use futures_util::future;
48use nautilus_common::live::get_runtime;
49use nautilus_model::{
50    enums::{ContingencyType, OrderSide, OrderType, TimeInForce, TrailingOffsetType, TriggerType},
51    identifiers::{ClientOrderId, InstrumentId, OrderListId},
52    instruments::InstrumentAny,
53    reports::OrderStatusReport,
54    types::{Price, Quantity},
55};
56use tokio::{sync::RwLock, task::JoinHandle, time::interval};
57
58use crate::{
59    common::{
60        consts::BITMEX_HTTP_TESTNET_URL,
61        enums::{BitmexEnvironment, BitmexPegPriceType},
62    },
63    http::{client::BitmexHttpClient, error::BitmexHttpError},
64};
65
66pub(crate) const DEFINITIVE_SUBMIT_REJECTION: &str = "DEFINITIVE_SUBMIT_REJECTION";
67
68/// Trait for order submission operations.
69///
70/// This trait abstracts the execution layer to enable dependency injection and testing
71/// without conditional compilation. The broadcaster holds executors as `Arc<dyn SubmitExecutor>`
72/// to avoid generic type parameters that would complicate the Python FFI boundary.
73///
74/// # Thread Safety
75///
76/// All methods must be safe to call concurrently from multiple threads. Implementations
77/// should use interior mutability (e.g., `Arc<Mutex<T>>`) if mutable state is required.
78///
79/// # Error Handling
80///
81/// Methods return `anyhow::Result` for flexibility. Implementers should provide
82/// meaningful error messages that can be logged and tracked by the broadcaster.
83///
84/// # Implementation Note
85///
86/// This trait does not require `Clone` because executors are wrapped in `Arc` at the
87/// `TransportClient` level. This allows `BitmexHttpClient` (which doesn't implement
88/// `Clone`) to be used without modification.
89trait SubmitExecutor: Send + Sync {
90    /// Adds an instrument for caching.
91    fn add_instrument(&self, instrument: InstrumentAny);
92
93    /// Performs a health check on the executor.
94    fn health_check(&self) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + '_>>;
95
96    /// Submits a single order.
97    #[expect(clippy::too_many_arguments)]
98    fn submit_order(
99        &self,
100        instrument_id: InstrumentId,
101        client_order_id: ClientOrderId,
102        order_side: OrderSide,
103        order_type: OrderType,
104        quantity: Quantity,
105        time_in_force: TimeInForce,
106        price: Option<Price>,
107        trigger_price: Option<Price>,
108        trigger_type: Option<TriggerType>,
109        trailing_offset: Option<f64>,
110        trailing_offset_type: Option<TrailingOffsetType>,
111        display_qty: Option<Quantity>,
112        post_only: bool,
113        reduce_only: bool,
114        order_list_id: Option<OrderListId>,
115        contingency_type: Option<ContingencyType>,
116        peg_price_type: Option<BitmexPegPriceType>,
117        peg_offset_value: Option<f64>,
118    ) -> Pin<Box<dyn Future<Output = anyhow::Result<OrderStatusReport>> + Send + '_>>;
119}
120
121impl SubmitExecutor for BitmexHttpClient {
122    fn add_instrument(&self, instrument: InstrumentAny) {
123        Self::cache_instrument(self, instrument);
124    }
125
126    fn health_check(&self) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + '_>> {
127        Box::pin(async move {
128            Self::get_server_time(self)
129                .await
130                .map(|_| ())
131                .map_err(|e| anyhow::anyhow!("{e}"))
132        })
133    }
134
135    fn submit_order(
136        &self,
137        instrument_id: InstrumentId,
138        client_order_id: ClientOrderId,
139        order_side: OrderSide,
140        order_type: OrderType,
141        quantity: Quantity,
142        time_in_force: TimeInForce,
143        price: Option<Price>,
144        trigger_price: Option<Price>,
145        trigger_type: Option<TriggerType>,
146        trailing_offset: Option<f64>,
147        trailing_offset_type: Option<TrailingOffsetType>,
148        display_qty: Option<Quantity>,
149        post_only: bool,
150        reduce_only: bool,
151        order_list_id: Option<OrderListId>,
152        contingency_type: Option<ContingencyType>,
153        peg_price_type: Option<BitmexPegPriceType>,
154        peg_offset_value: Option<f64>,
155    ) -> Pin<Box<dyn Future<Output = anyhow::Result<OrderStatusReport>> + Send + '_>> {
156        Box::pin(async move {
157            Self::submit_order(
158                self,
159                instrument_id,
160                client_order_id,
161                order_side,
162                order_type,
163                quantity,
164                time_in_force,
165                price,
166                trigger_price,
167                trigger_type,
168                trailing_offset,
169                trailing_offset_type,
170                display_qty,
171                post_only,
172                reduce_only,
173                order_list_id,
174                contingency_type,
175                peg_price_type,
176                peg_offset_value,
177            )
178            .await
179        })
180    }
181}
182
183/// Configuration for the submit broadcaster.
184#[derive(Debug, Clone)]
185pub struct SubmitBroadcasterConfig {
186    /// Number of HTTP clients in the pool.
187    pub pool_size: usize,
188    /// BitMEX API key (None will source from environment).
189    pub api_key: Option<String>,
190    /// BitMEX API secret (None will source from environment).
191    pub api_secret: Option<String>,
192    /// Base URL for BitMEX HTTP API.
193    pub base_url: Option<String>,
194    /// BitMEX environment (mainnet or testnet).
195    pub environment: BitmexEnvironment,
196    /// Timeout in seconds for HTTP requests.
197    pub timeout_secs: u64,
198    /// Maximum number of retry attempts for failed requests.
199    pub max_retries: u32,
200    /// Initial delay in milliseconds between retry attempts.
201    pub retry_delay_ms: u64,
202    /// Maximum delay in milliseconds between retry attempts.
203    pub retry_delay_max_ms: u64,
204    /// Expiration window in milliseconds for signed requests.
205    pub recv_window_ms: u64,
206    /// Maximum REST burst rate (requests per second).
207    pub max_requests_per_second: u32,
208    /// Maximum REST rolling rate (requests per minute).
209    pub max_requests_per_minute: u32,
210    /// Interval in seconds between health check pings.
211    pub health_check_interval_secs: u64,
212    /// Timeout in seconds for health check requests.
213    pub health_check_timeout_secs: u64,
214    /// Substrings to identify expected submit rejections for debug-level logging.
215    pub expected_reject_patterns: Vec<String>,
216    /// Optional list of proxy URLs for path diversity.
217    ///
218    /// Each transport instance uses the proxy at its index. If the list is shorter
219    /// than pool_size, remaining transports will use no proxy. If longer, extra proxies
220    /// are ignored.
221    pub proxy_urls: Vec<Option<String>>,
222}
223
224impl Default for SubmitBroadcasterConfig {
225    fn default() -> Self {
226        Self {
227            pool_size: 3,
228            api_key: None,
229            api_secret: None,
230            base_url: None,
231            environment: BitmexEnvironment::Mainnet,
232            timeout_secs: 60,
233            max_retries: 3,
234            retry_delay_ms: 1_000,
235            retry_delay_max_ms: 5_000,
236            recv_window_ms: 10_000,
237            max_requests_per_second: 10,
238            max_requests_per_minute: 120,
239            health_check_interval_secs: 30,
240            health_check_timeout_secs: 5,
241            expected_reject_patterns: vec!["Duplicate clOrdID".to_string()],
242            proxy_urls: vec![],
243        }
244    }
245}
246
247/// Transport client wrapper with health monitoring.
248#[derive(Clone)]
249struct TransportClient {
250    /// Executor wrapped in Arc to enable cloning without requiring Clone on SubmitExecutor.
251    ///
252    /// BitmexHttpClient doesn't implement Clone, so we use reference counting to share
253    /// the executor across multiple TransportClient clones.
254    executor: Arc<dyn SubmitExecutor>,
255    client_id: String,
256    healthy: Arc<AtomicBool>,
257    submit_count: Arc<AtomicU64>,
258    error_count: Arc<AtomicU64>,
259}
260
261impl Debug for TransportClient {
262    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
263        f.debug_struct(stringify!(TransportClient))
264            .field("client_id", &self.client_id)
265            .field("healthy", &self.healthy)
266            .field("submit_count", &self.submit_count)
267            .field("error_count", &self.error_count)
268            .finish()
269    }
270}
271
272impl TransportClient {
273    fn new<E: SubmitExecutor + 'static>(executor: E, client_id: String) -> Self {
274        Self {
275            executor: Arc::new(executor),
276            client_id,
277            healthy: Arc::new(AtomicBool::new(true)),
278            submit_count: Arc::new(AtomicU64::new(0)),
279            error_count: Arc::new(AtomicU64::new(0)),
280        }
281    }
282
283    fn is_healthy(&self) -> bool {
284        self.healthy.load(Ordering::Relaxed)
285    }
286
287    fn mark_healthy(&self) {
288        self.healthy.store(true, Ordering::Relaxed);
289    }
290
291    fn mark_unhealthy(&self) {
292        self.healthy.store(false, Ordering::Relaxed);
293    }
294
295    fn get_submit_count(&self) -> u64 {
296        self.submit_count.load(Ordering::Relaxed)
297    }
298
299    fn get_error_count(&self) -> u64 {
300        self.error_count.load(Ordering::Relaxed)
301    }
302
303    async fn health_check(&self, timeout_secs: u64) -> bool {
304        match tokio::time::timeout(
305            Duration::from_secs(timeout_secs),
306            self.executor.health_check(),
307        )
308        .await
309        {
310            Ok(Ok(())) => {
311                self.mark_healthy();
312                true
313            }
314            Ok(Err(e)) => {
315                log::warn!("Health check failed for client {}: {e:?}", self.client_id);
316                self.mark_unhealthy();
317                false
318            }
319            Err(_) => {
320                log::warn!("Health check timeout for client {}", self.client_id);
321                self.mark_unhealthy();
322                false
323            }
324        }
325    }
326
327    #[expect(clippy::too_many_arguments)]
328    async fn submit_order(
329        &self,
330        instrument_id: InstrumentId,
331        client_order_id: ClientOrderId,
332        order_side: OrderSide,
333        order_type: OrderType,
334        quantity: Quantity,
335        time_in_force: TimeInForce,
336        price: Option<Price>,
337        trigger_price: Option<Price>,
338        trigger_type: Option<TriggerType>,
339        trailing_offset: Option<f64>,
340        trailing_offset_type: Option<TrailingOffsetType>,
341        display_qty: Option<Quantity>,
342        post_only: bool,
343        reduce_only: bool,
344        order_list_id: Option<OrderListId>,
345        contingency_type: Option<ContingencyType>,
346        peg_price_type: Option<BitmexPegPriceType>,
347        peg_offset_value: Option<f64>,
348    ) -> anyhow::Result<OrderStatusReport> {
349        self.submit_count.fetch_add(1, Ordering::Relaxed);
350
351        match self
352            .executor
353            .submit_order(
354                instrument_id,
355                client_order_id,
356                order_side,
357                order_type,
358                quantity,
359                time_in_force,
360                price,
361                trigger_price,
362                trigger_type,
363                trailing_offset,
364                trailing_offset_type,
365                display_qty,
366                post_only,
367                reduce_only,
368                order_list_id,
369                contingency_type,
370                peg_price_type,
371                peg_offset_value,
372            )
373            .await
374        {
375            Ok(report) => {
376                self.mark_healthy();
377                Ok(report)
378            }
379            Err(e) => {
380                self.error_count.fetch_add(1, Ordering::Relaxed);
381                Err(e)
382            }
383        }
384    }
385}
386
387/// Broadcasts submit requests to multiple HTTP clients for redundancy.
388///
389/// This broadcaster fans out submit requests to multiple pre-warmed HTTP clients
390/// in parallel, short-circuits when the first successful acknowledgement is received,
391/// and handles expected rejection patterns (duplicate clOrdID) with appropriate log levels.
392#[cfg_attr(feature = "python", pyo3::pyclass)]
393#[cfg_attr(
394    feature = "python",
395    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.bitmex")
396)]
397#[derive(Debug)]
398pub struct SubmitBroadcaster {
399    config: SubmitBroadcasterConfig,
400    transports: Arc<[TransportClient]>,
401    health_check_task: Arc<RwLock<Option<JoinHandle<()>>>>,
402    running: Arc<AtomicBool>,
403    total_submits: Arc<AtomicU64>,
404    successful_submits: Arc<AtomicU64>,
405    failed_submits: Arc<AtomicU64>,
406    expected_rejects: Arc<AtomicU64>,
407}
408
409impl SubmitBroadcaster {
410    /// Creates a new [`SubmitBroadcaster`] with internal HTTP client pool.
411    ///
412    /// # Errors
413    ///
414    /// Returns an error if any HTTP client fails to initialize.
415    pub fn new(config: SubmitBroadcasterConfig) -> anyhow::Result<Self> {
416        let mut transports = Vec::with_capacity(config.pool_size);
417
418        let base_url = match config.environment {
419            BitmexEnvironment::Testnet if config.base_url.is_none() => {
420                Some(BITMEX_HTTP_TESTNET_URL.to_string())
421            }
422            _ => config.base_url.clone(),
423        };
424
425        for i in 0..config.pool_size {
426            // Assign proxy from config list, or None if index exceeds list length
427            let proxy_url = config.proxy_urls.get(i).and_then(|p| p.clone());
428
429            let client = BitmexHttpClient::with_credentials(
430                config.api_key.clone(),
431                config.api_secret.clone(),
432                base_url.clone(),
433                config.timeout_secs,
434                config.max_retries,
435                config.retry_delay_ms,
436                config.retry_delay_max_ms,
437                config.recv_window_ms,
438                config.max_requests_per_second,
439                config.max_requests_per_minute,
440                proxy_url,
441            )
442            .map_err(|e| anyhow::anyhow!("Failed to create HTTP client {i}: {e}"))?;
443
444            transports.push(TransportClient::new(client, format!("bitmex-submit-{i}")));
445        }
446
447        Ok(Self {
448            config,
449            transports: Arc::from(transports),
450            health_check_task: Arc::new(RwLock::new(None)),
451            running: Arc::new(AtomicBool::new(false)),
452            total_submits: Arc::new(AtomicU64::new(0)),
453            successful_submits: Arc::new(AtomicU64::new(0)),
454            failed_submits: Arc::new(AtomicU64::new(0)),
455            expected_rejects: Arc::new(AtomicU64::new(0)),
456        })
457    }
458
459    /// Starts the broadcaster and health check loop.
460    ///
461    /// # Errors
462    ///
463    /// Returns an error if the broadcaster is already running.
464    pub async fn start(&self) -> anyhow::Result<()> {
465        if self.running.load(Ordering::Relaxed) {
466            return Ok(());
467        }
468
469        self.running.store(true, Ordering::Relaxed);
470
471        // Initial health check for all clients
472        self.run_health_checks().await;
473
474        // Start periodic health check task
475        let transports = Arc::clone(&self.transports);
476        let running = Arc::clone(&self.running);
477        let interval_secs = self.config.health_check_interval_secs;
478        let timeout_secs = self.config.health_check_timeout_secs;
479
480        let task = get_runtime().spawn(async move {
481            let mut ticker = interval(Duration::from_secs(interval_secs));
482            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
483
484            loop {
485                ticker.tick().await;
486
487                if !running.load(Ordering::Relaxed) {
488                    break;
489                }
490
491                let tasks: Vec<_> = transports
492                    .iter()
493                    .map(|t| t.health_check(timeout_secs))
494                    .collect();
495
496                let results = future::join_all(tasks).await;
497                let healthy_count = results.iter().filter(|&&r| r).count();
498
499                log::debug!(
500                    "Health check complete: {healthy_count}/{} clients healthy",
501                    results.len()
502                );
503            }
504        });
505
506        *self.health_check_task.write().await = Some(task);
507
508        log::debug!(
509            "SubmitBroadcaster started with {} clients",
510            self.transports.len()
511        );
512
513        Ok(())
514    }
515
516    /// Stops the broadcaster and health check loop.
517    pub async fn stop(&self) {
518        if !self.running.load(Ordering::Relaxed) {
519            return;
520        }
521
522        self.running.store(false, Ordering::Relaxed);
523
524        if let Some(task) = self.health_check_task.write().await.take() {
525            task.abort();
526        }
527
528        log::debug!("SubmitBroadcaster stopped");
529    }
530
531    async fn run_health_checks(&self) {
532        let tasks: Vec<_> = self
533            .transports
534            .iter()
535            .map(|t| t.health_check(self.config.health_check_timeout_secs))
536            .collect();
537
538        let results = future::join_all(tasks).await;
539        let healthy_count = results.iter().filter(|&&r| r).count();
540
541        log::debug!(
542            "Health check complete: {healthy_count}/{} clients healthy",
543            results.len()
544        );
545    }
546
547    fn is_expected_reject(&self, error_message: &str) -> bool {
548        self.config
549            .expected_reject_patterns
550            .iter()
551            .any(|pattern| error_message.contains(pattern))
552    }
553
554    /// Processes submit request results, handling success and failures.
555    ///
556    /// This helper consolidates the common error handling loop used for submit broadcasts.
557    async fn process_submit_results<T>(
558        &self,
559        mut handles: Vec<JoinHandle<(String, anyhow::Result<T>)>>,
560        operation: &str,
561        params: String,
562    ) -> anyhow::Result<T>
563    where
564        T: Send + 'static,
565    {
566        let mut errors = Vec::new();
567        let mut all_duplicate_clordid = true;
568        let mut all_definitive_refusals = true;
569
570        while !handles.is_empty() {
571            let current_handles = std::mem::take(&mut handles);
572            let (result, _idx, remaining) = future::select_all(current_handles).await;
573            handles = remaining.into_iter().collect();
574
575            match result {
576                Ok((client_id, Ok(result))) => {
577                    // First success - abort remaining handles
578                    for handle in &handles {
579                        handle.abort();
580                    }
581                    self.successful_submits.fetch_add(1, Ordering::Relaxed);
582                    log::debug!("{operation} broadcast succeeded [{client_id}] {params}",);
583                    return Ok(result);
584                }
585                Ok((client_id, Err(e))) => {
586                    let error_msg = e.to_string();
587                    let is_duplicate = error_msg.contains("Duplicate clOrdID");
588                    let is_definitive_refusal = is_definitive_submit_refusal(&e);
589
590                    if !is_duplicate {
591                        all_duplicate_clordid = false;
592                    }
593
594                    if !is_definitive_refusal {
595                        all_definitive_refusals = false;
596                    }
597
598                    if self.is_expected_reject(&error_msg) {
599                        self.expected_rejects.fetch_add(1, Ordering::Relaxed);
600                        log::debug!(
601                            "Expected {} rejection [{client_id}]: {error_msg} {params}",
602                            operation.to_lowercase(),
603                        );
604                        errors.push(error_msg);
605                    } else {
606                        log::warn!(
607                            "{operation} request failed [{client_id}]: {error_msg} {params}",
608                        );
609                        errors.push(error_msg);
610                    }
611                }
612                Err(e) => {
613                    all_duplicate_clordid = false;
614                    all_definitive_refusals = false;
615                    log::warn!("{operation} task join error: {e:?}");
616                    errors.push(format!("Task panicked: {e:?}"));
617                }
618            }
619        }
620
621        // All tasks failed
622        self.failed_submits.fetch_add(1, Ordering::Relaxed);
623
624        // If all errors were "Duplicate clOrdID", this is likely an idempotent scenario
625        // where the order exists but the success response was lost
626        if all_duplicate_clordid && !errors.is_empty() {
627            log::warn!(
628                "All {} requests returned 'Duplicate clOrdID' - order likely exists {params}",
629                operation.to_lowercase(),
630            );
631            anyhow::bail!("IDEMPOTENT_DUPLICATE: Order likely exists but confirmation was lost");
632        }
633
634        if all_definitive_refusals && !errors.is_empty() {
635            log::error!(
636                "All {} requests were refused by BitMEX: {errors:?} {params}",
637                operation.to_lowercase(),
638            );
639            anyhow::bail!(
640                "{DEFINITIVE_SUBMIT_REJECTION}: All {} requests were refused by BitMEX: {errors:?}",
641                operation.to_lowercase(),
642            );
643        }
644
645        log::error!(
646            "All {} requests failed: {errors:?} {params}",
647            operation.to_lowercase(),
648        );
649        Err(anyhow::anyhow!(
650            "All {} requests failed: {:?}",
651            operation.to_lowercase(),
652            errors
653        ))
654    }
655
656    /// Broadcasts a submit request to all healthy clients in parallel.
657    ///
658    /// # Returns
659    ///
660    /// - `Ok(report)` if successfully submitted with a report.
661    /// - `Err` if all requests failed.
662    ///
663    /// # Errors
664    ///
665    /// Returns an error if all submit requests fail or no healthy clients are available.
666    #[expect(clippy::too_many_arguments)]
667    pub async fn broadcast_submit(
668        &self,
669        instrument_id: InstrumentId,
670        client_order_id: ClientOrderId,
671        order_side: OrderSide,
672        order_type: OrderType,
673        quantity: Quantity,
674        time_in_force: TimeInForce,
675        price: Option<Price>,
676        trigger_price: Option<Price>,
677        trigger_type: Option<TriggerType>,
678        trailing_offset: Option<f64>,
679        trailing_offset_type: Option<TrailingOffsetType>,
680        display_qty: Option<Quantity>,
681        post_only: bool,
682        reduce_only: bool,
683        order_list_id: Option<OrderListId>,
684        contingency_type: Option<ContingencyType>,
685        submit_tries: Option<usize>,
686        peg_price_type: Option<BitmexPegPriceType>,
687        peg_offset_value: Option<f64>,
688    ) -> anyhow::Result<OrderStatusReport> {
689        self.total_submits.fetch_add(1, Ordering::Relaxed);
690
691        let pool_size = self.config.pool_size;
692        let actual_tries = if let Some(t) = submit_tries {
693            if t > pool_size {
694                // Use log macro for Python visibility for now
695                log::warn!("submit_tries={t} exceeds pool_size={pool_size}, capping at pool_size");
696            }
697            std::cmp::min(t, pool_size)
698        } else {
699            pool_size
700        };
701
702        log::debug!(
703            "Submit broadcast requested for client_order_id={client_order_id} (tries={actual_tries}/{pool_size})",
704        );
705
706        let healthy_transports: Vec<TransportClient> = self
707            .transports
708            .iter()
709            .filter(|t| t.is_healthy())
710            .take(actual_tries)
711            .cloned()
712            .collect();
713
714        if healthy_transports.is_empty() {
715            self.failed_submits.fetch_add(1, Ordering::Relaxed);
716            anyhow::bail!("No healthy transport clients available");
717        }
718
719        log::debug!(
720            "Broadcasting submit to {} clients: client_order_id={client_order_id}, instrument_id={instrument_id}",
721            healthy_transports.len(),
722        );
723
724        let mut handles = Vec::new();
725
726        for transport in healthy_transports {
727            // All transports use the same client_order_id. If multiple succeed,
728            // BitMEX rejects duplicates with "duplicate clOrdID" (expected rejection).
729            let handle = get_runtime().spawn(async move {
730                let client_id = transport.client_id.clone();
731                let result = transport
732                    .submit_order(
733                        instrument_id,
734                        client_order_id,
735                        order_side,
736                        order_type,
737                        quantity,
738                        time_in_force,
739                        price,
740                        trigger_price,
741                        trigger_type,
742                        trailing_offset,
743                        trailing_offset_type,
744                        display_qty,
745                        post_only,
746                        reduce_only,
747                        order_list_id,
748                        contingency_type,
749                        peg_price_type,
750                        peg_offset_value,
751                    )
752                    .await;
753                (client_id, result)
754            });
755            handles.push(handle);
756        }
757
758        self.process_submit_results(
759            handles,
760            "Submit",
761            format!("(client_order_id={client_order_id:?})"),
762        )
763        .await
764    }
765
766    /// Gets broadcaster metrics.
767    pub fn get_metrics(&self) -> BroadcasterMetrics {
768        let healthy_clients = self.transports.iter().filter(|t| t.is_healthy()).count();
769        let total_clients = self.transports.len();
770
771        BroadcasterMetrics {
772            total_submits: self.total_submits.load(Ordering::Relaxed),
773            successful_submits: self.successful_submits.load(Ordering::Relaxed),
774            failed_submits: self.failed_submits.load(Ordering::Relaxed),
775            expected_rejects: self.expected_rejects.load(Ordering::Relaxed),
776            healthy_clients,
777            total_clients,
778        }
779    }
780
781    /// Gets broadcaster metrics (async version for use within async context).
782    pub async fn get_metrics_async(&self) -> BroadcasterMetrics {
783        self.get_metrics()
784    }
785
786    /// Gets per-client statistics.
787    pub fn get_client_stats(&self) -> Vec<ClientStats> {
788        self.transports
789            .iter()
790            .map(|t| ClientStats {
791                client_id: t.client_id.clone(),
792                healthy: t.is_healthy(),
793                submit_count: t.get_submit_count(),
794                error_count: t.get_error_count(),
795            })
796            .collect()
797    }
798
799    /// Gets per-client statistics (async version for use within async context).
800    pub async fn get_client_stats_async(&self) -> Vec<ClientStats> {
801        self.get_client_stats()
802    }
803
804    /// Caches an instrument in all HTTP clients in the pool.
805    pub fn cache_instrument(&self, instrument: &InstrumentAny) {
806        for transport in self.transports.iter() {
807            transport.executor.add_instrument(instrument.clone());
808        }
809    }
810
811    #[must_use]
812    pub fn clone_for_async(&self) -> Self {
813        Self {
814            config: self.config.clone(),
815            transports: Arc::clone(&self.transports),
816            health_check_task: Arc::clone(&self.health_check_task),
817            running: Arc::clone(&self.running),
818            total_submits: Arc::clone(&self.total_submits),
819            successful_submits: Arc::clone(&self.successful_submits),
820            failed_submits: Arc::clone(&self.failed_submits),
821            expected_rejects: Arc::clone(&self.expected_rejects),
822        }
823    }
824
825    #[cfg(test)]
826    fn new_with_transports(
827        config: SubmitBroadcasterConfig,
828        transports: Vec<TransportClient>,
829    ) -> Self {
830        Self {
831            config,
832            transports: Arc::from(transports),
833            health_check_task: Arc::new(RwLock::new(None)),
834            running: Arc::new(AtomicBool::new(false)),
835            total_submits: Arc::new(AtomicU64::new(0)),
836            successful_submits: Arc::new(AtomicU64::new(0)),
837            failed_submits: Arc::new(AtomicU64::new(0)),
838            expected_rejects: Arc::new(AtomicU64::new(0)),
839        }
840    }
841}
842
843fn is_definitive_submit_refusal(err: &anyhow::Error) -> bool {
844    if err.chain().any(|cause| {
845        cause
846            .downcast_ref::<BitmexHttpError>()
847            .is_some_and(|e| matches!(e, BitmexHttpError::BitmexError { .. }))
848    }) {
849        return true;
850    }
851
852    err.to_string().starts_with("Order rejected:")
853}
854
855/// Broadcaster metrics snapshot.
856#[derive(Debug, Clone)]
857pub struct BroadcasterMetrics {
858    pub total_submits: u64,
859    pub successful_submits: u64,
860    pub failed_submits: u64,
861    pub expected_rejects: u64,
862    pub healthy_clients: usize,
863    pub total_clients: usize,
864}
865
866/// Per-client statistics.
867#[derive(Debug, Clone)]
868pub struct ClientStats {
869    pub client_id: String,
870    pub healthy: bool,
871    pub submit_count: u64,
872    pub error_count: u64,
873}
874
875#[cfg(test)]
876mod tests {
877    use std::{str::FromStr, sync::atomic::Ordering, time::Duration};
878
879    use nautilus_core::UUID4;
880    use nautilus_model::{
881        enums::{
882            ContingencyType, OrderSide, OrderStatus, OrderType, TimeInForce, TrailingOffsetType,
883        },
884        identifiers::{AccountId, ClientOrderId, InstrumentId, VenueOrderId},
885        reports::OrderStatusReport,
886        types::{Price, Quantity},
887    };
888
889    use super::*;
890
891    /// Mock executor for testing.
892    #[derive(Clone)]
893    #[expect(clippy::type_complexity)]
894    struct MockExecutor {
895        handler: Arc<
896            dyn Fn() -> Pin<Box<dyn Future<Output = anyhow::Result<OrderStatusReport>> + Send>>
897                + Send
898                + Sync,
899        >,
900    }
901
902    impl MockExecutor {
903        fn new<F, Fut>(handler: F) -> Self
904        where
905            F: Fn() -> Fut + Send + Sync + 'static,
906            Fut: Future<Output = anyhow::Result<OrderStatusReport>> + Send + 'static,
907        {
908            Self {
909                handler: Arc::new(move || Box::pin(handler())),
910            }
911        }
912    }
913
914    impl SubmitExecutor for MockExecutor {
915        fn health_check(&self) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + '_>> {
916            Box::pin(async { Ok(()) })
917        }
918
919        fn submit_order(
920            &self,
921            _instrument_id: InstrumentId,
922            _client_order_id: ClientOrderId,
923            _order_side: OrderSide,
924            _order_type: OrderType,
925            _quantity: Quantity,
926            _time_in_force: TimeInForce,
927            _price: Option<Price>,
928            _trigger_price: Option<Price>,
929            _trigger_type: Option<TriggerType>,
930            _trailing_offset: Option<f64>,
931            _trailing_offset_type: Option<TrailingOffsetType>,
932            _display_qty: Option<Quantity>,
933            _post_only: bool,
934            _reduce_only: bool,
935            _order_list_id: Option<OrderListId>,
936            _contingency_type: Option<ContingencyType>,
937            _peg_price_type: Option<BitmexPegPriceType>,
938            _peg_offset_value: Option<f64>,
939        ) -> Pin<Box<dyn Future<Output = anyhow::Result<OrderStatusReport>> + Send + '_>> {
940            (self.handler)()
941        }
942
943        fn add_instrument(&self, _instrument: InstrumentAny) {
944            // No-op for mock
945        }
946    }
947
948    fn create_test_report(venue_order_id: &str) -> OrderStatusReport {
949        OrderStatusReport {
950            account_id: AccountId::from("BITMEX-001"),
951            instrument_id: InstrumentId::from_str("XBTUSD.BITMEX").unwrap(),
952            venue_order_id: VenueOrderId::from(venue_order_id),
953            order_side: OrderSide::Buy,
954            order_type: OrderType::Limit,
955            time_in_force: TimeInForce::Gtc,
956            order_status: OrderStatus::Accepted,
957            price: Some(Price::new(50000.0, 2)),
958            quantity: Quantity::new(100.0, 0),
959            filled_qty: Quantity::new(0.0, 0),
960            report_id: UUID4::new(),
961            ts_accepted: 0.into(),
962            ts_last: 0.into(),
963            ts_init: 0.into(),
964            client_order_id: None,
965            avg_px: None,
966            trigger_price: None,
967            trigger_type: None,
968            contingency_type: ContingencyType::NoContingency,
969            expire_time: None,
970            order_list_id: None,
971            venue_position_id: None,
972            linked_order_ids: None,
973            parent_order_id: None,
974            display_qty: None,
975            limit_offset: None,
976            trailing_offset: None,
977            trailing_offset_type: TrailingOffsetType::NoTrailingOffset,
978            post_only: false,
979            reduce_only: false,
980            cancel_reason: None,
981            ts_triggered: None,
982        }
983    }
984
985    fn create_stub_transport<F, Fut>(client_id: &str, handler: F) -> TransportClient
986    where
987        F: Fn() -> Fut + Send + Sync + 'static,
988        Fut: Future<Output = anyhow::Result<OrderStatusReport>> + Send + 'static,
989    {
990        let executor = MockExecutor::new(handler);
991        TransportClient::new(executor, client_id.to_string())
992    }
993
994    #[tokio::test]
995    async fn test_broadcast_submit_immediate_success() {
996        let report = create_test_report("ORDER-1");
997        let report_clone = report.clone();
998
999        let transports = vec![
1000            create_stub_transport("client-0", move || {
1001                let report = report_clone.clone();
1002                async move { Ok(report) }
1003            }),
1004            create_stub_transport("client-1", || async {
1005                tokio::time::sleep(Duration::from_secs(10)).await;
1006                anyhow::bail!("Should be aborted")
1007            }),
1008        ];
1009
1010        let config = SubmitBroadcasterConfig::default();
1011        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1012
1013        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1014        let result = broadcaster
1015            .broadcast_submit(
1016                instrument_id,
1017                ClientOrderId::from("O-123"),
1018                OrderSide::Buy,
1019                OrderType::Limit,
1020                Quantity::new(100.0, 0),
1021                TimeInForce::Gtc,
1022                Some(Price::new(50000.0, 2)),
1023                None,
1024                None,
1025                None,
1026                None,
1027                None,
1028                false,
1029                false,
1030                None,
1031                None,
1032                None,
1033                None,
1034                None,
1035            )
1036            .await;
1037
1038        assert!(result.is_ok());
1039        let returned_report = result.unwrap();
1040        assert_eq!(returned_report.venue_order_id, report.venue_order_id);
1041
1042        let metrics = broadcaster.get_metrics_async().await;
1043        assert_eq!(metrics.successful_submits, 1);
1044        assert_eq!(metrics.failed_submits, 0);
1045        assert_eq!(metrics.total_submits, 1);
1046    }
1047
1048    #[tokio::test]
1049    async fn test_broadcast_submit_duplicate_clordid_expected() {
1050        let transports = vec![
1051            create_stub_transport("client-0", || async { anyhow::bail!("Duplicate clOrdID") }),
1052            create_stub_transport("client-1", || async {
1053                tokio::time::sleep(Duration::from_secs(10)).await;
1054                anyhow::bail!("Should be aborted")
1055            }),
1056        ];
1057
1058        let config = SubmitBroadcasterConfig::default();
1059        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1060
1061        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1062        let result = broadcaster
1063            .broadcast_submit(
1064                instrument_id,
1065                ClientOrderId::from("O-123"),
1066                OrderSide::Buy,
1067                OrderType::Limit,
1068                Quantity::new(100.0, 0),
1069                TimeInForce::Gtc,
1070                Some(Price::new(50000.0, 2)),
1071                None,
1072                None,
1073                None,
1074                None,
1075                None,
1076                false,
1077                false,
1078                None,
1079                None,
1080                None,
1081                None,
1082                None,
1083            )
1084            .await;
1085
1086        assert!(result.is_err());
1087
1088        let metrics = broadcaster.get_metrics_async().await;
1089        assert_eq!(metrics.expected_rejects, 1);
1090        assert_eq!(metrics.successful_submits, 0);
1091        assert_eq!(metrics.failed_submits, 1);
1092    }
1093
1094    #[tokio::test]
1095    async fn test_broadcast_submit_all_failures() {
1096        let transports = vec![
1097            create_stub_transport("client-0", || async { anyhow::bail!("502 Bad Gateway") }),
1098            create_stub_transport("client-1", || async { anyhow::bail!("Connection refused") }),
1099        ];
1100
1101        let config = SubmitBroadcasterConfig::default();
1102        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1103
1104        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1105        let result = broadcaster
1106            .broadcast_submit(
1107                instrument_id,
1108                ClientOrderId::from("O-456"),
1109                OrderSide::Sell,
1110                OrderType::Market,
1111                Quantity::new(50.0, 0),
1112                TimeInForce::Ioc,
1113                None,
1114                None,
1115                None,
1116                None,
1117                None,
1118                None,
1119                false,
1120                false,
1121                None,
1122                None,
1123                None,
1124                None,
1125                None,
1126            )
1127            .await;
1128
1129        assert!(result.is_err());
1130        assert!(
1131            result
1132                .unwrap_err()
1133                .to_string()
1134                .contains("All submit requests failed")
1135        );
1136
1137        let metrics = broadcaster.get_metrics_async().await;
1138        assert_eq!(metrics.failed_submits, 1);
1139        assert_eq!(metrics.successful_submits, 0);
1140    }
1141
1142    #[tokio::test]
1143    async fn test_broadcast_submit_all_bitmex_refusals_preserves_definitive_outcome() {
1144        let transports = vec![
1145            create_stub_transport("client-0", || async {
1146                Err(anyhow::Error::new(BitmexHttpError::BitmexError {
1147                    error_name: "HTTPError".to_string(),
1148                    message: "Invalid price".to_string(),
1149                }))
1150            }),
1151            create_stub_transport("client-1", || async {
1152                Err(anyhow::Error::new(BitmexHttpError::BitmexError {
1153                    error_name: "HTTPError".to_string(),
1154                    message: "Invalid price".to_string(),
1155                }))
1156            }),
1157        ];
1158
1159        let config = SubmitBroadcasterConfig::default();
1160        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1161
1162        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1163        let result = broadcaster
1164            .broadcast_submit(
1165                instrument_id,
1166                ClientOrderId::from("O-REFUSED"),
1167                OrderSide::Sell,
1168                OrderType::Limit,
1169                Quantity::new(50.0, 0),
1170                TimeInForce::Gtc,
1171                Some(Price::new(50000.0, 2)),
1172                None,
1173                None,
1174                None,
1175                None,
1176                None,
1177                false,
1178                false,
1179                None,
1180                None,
1181                None,
1182                None,
1183                None,
1184            )
1185            .await;
1186
1187        let err = result.unwrap_err().to_string();
1188
1189        assert!(err.starts_with(DEFINITIVE_SUBMIT_REJECTION));
1190
1191        let metrics = broadcaster.get_metrics_async().await;
1192        assert_eq!(metrics.failed_submits, 1);
1193        assert_eq!(metrics.successful_submits, 0);
1194    }
1195
1196    #[tokio::test]
1197    async fn test_broadcast_submit_mixed_refusal_and_network_failure_stays_ambiguous() {
1198        let transports = vec![
1199            create_stub_transport("client-0", || async {
1200                Err(anyhow::Error::new(BitmexHttpError::BitmexError {
1201                    error_name: "HTTPError".to_string(),
1202                    message: "Invalid price".to_string(),
1203                }))
1204            }),
1205            create_stub_transport("client-1", || async { anyhow::bail!("Connection refused") }),
1206        ];
1207
1208        let config = SubmitBroadcasterConfig::default();
1209        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1210
1211        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1212        let result = broadcaster
1213            .broadcast_submit(
1214                instrument_id,
1215                ClientOrderId::from("O-MIXED-FAILURE"),
1216                OrderSide::Sell,
1217                OrderType::Limit,
1218                Quantity::from("50"),
1219                TimeInForce::Gtc,
1220                Some(Price::from("50000.00")),
1221                None,
1222                None,
1223                None,
1224                None,
1225                None,
1226                false,
1227                false,
1228                None,
1229                None,
1230                None,
1231                None,
1232                None,
1233            )
1234            .await;
1235
1236        let err = result.unwrap_err().to_string();
1237
1238        assert!(err.starts_with("All submit requests failed"));
1239        assert!(!err.starts_with(DEFINITIVE_SUBMIT_REJECTION));
1240
1241        let metrics = broadcaster.get_metrics_async().await;
1242        assert_eq!(metrics.failed_submits, 1);
1243        assert_eq!(metrics.successful_submits, 0);
1244    }
1245
1246    #[tokio::test]
1247    async fn test_broadcast_submit_no_healthy_clients() {
1248        let transport =
1249            create_stub_transport("client-0", || async { Ok(create_test_report("ORDER-1")) });
1250        transport.healthy.store(false, Ordering::Relaxed);
1251
1252        let config = SubmitBroadcasterConfig::default();
1253        let broadcaster = SubmitBroadcaster::new_with_transports(config, vec![transport]);
1254
1255        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1256        let result = broadcaster
1257            .broadcast_submit(
1258                instrument_id,
1259                ClientOrderId::from("O-789"),
1260                OrderSide::Buy,
1261                OrderType::Limit,
1262                Quantity::new(100.0, 0),
1263                TimeInForce::Gtc,
1264                Some(Price::new(50000.0, 2)),
1265                None,
1266                None,
1267                None,
1268                None,
1269                None,
1270                false,
1271                false,
1272                None,
1273                None,
1274                None,
1275                None,
1276                None,
1277            )
1278            .await;
1279
1280        assert!(result.is_err());
1281        assert!(
1282            result
1283                .unwrap_err()
1284                .to_string()
1285                .contains("No healthy transport clients available")
1286        );
1287
1288        let metrics = broadcaster.get_metrics_async().await;
1289        assert_eq!(metrics.failed_submits, 1);
1290    }
1291
1292    #[tokio::test]
1293    async fn test_default_config() {
1294        let report = create_test_report("ORDER-1");
1295        let transports: Vec<TransportClient> = (0..3)
1296            .map(|i| {
1297                let r = report.clone();
1298                create_stub_transport(&format!("client-{i}"), move || {
1299                    let r = r.clone();
1300                    async move { Ok(r) }
1301                })
1302            })
1303            .collect();
1304
1305        let config = SubmitBroadcasterConfig::default();
1306        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1307        let metrics = broadcaster.get_metrics_async().await;
1308
1309        assert_eq!(metrics.total_clients, 3);
1310    }
1311
1312    #[tokio::test]
1313    async fn test_broadcaster_lifecycle() {
1314        let report = create_test_report("ORDER-1");
1315        let transports: Vec<TransportClient> = (0..2)
1316            .map(|i| {
1317                let r = report.clone();
1318                create_stub_transport(&format!("client-{i}"), move || {
1319                    let r = r.clone();
1320                    async move { Ok(r) }
1321                })
1322            })
1323            .collect();
1324
1325        let config = SubmitBroadcasterConfig::default();
1326        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1327
1328        // Should not be running initially
1329        assert!(!broadcaster.running.load(Ordering::Relaxed));
1330
1331        // Start broadcaster
1332        let start_result = broadcaster.start().await;
1333        assert!(start_result.is_ok());
1334        assert!(broadcaster.running.load(Ordering::Relaxed));
1335
1336        // Starting again should be idempotent
1337        let start_again = broadcaster.start().await;
1338        assert!(start_again.is_ok());
1339
1340        // Stop broadcaster
1341        broadcaster.stop().await;
1342        assert!(!broadcaster.running.load(Ordering::Relaxed));
1343
1344        // Stopping again should be safe
1345        broadcaster.stop().await;
1346        assert!(!broadcaster.running.load(Ordering::Relaxed));
1347    }
1348
1349    #[tokio::test]
1350    async fn test_broadcast_submit_metrics_increment() {
1351        let report = create_test_report("ORDER-1");
1352        let report_clone = report.clone();
1353
1354        let transports = vec![create_stub_transport("client-0", move || {
1355            let report = report_clone.clone();
1356            async move { Ok(report) }
1357        })];
1358
1359        let config = SubmitBroadcasterConfig::default();
1360        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1361
1362        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1363        let _ = broadcaster
1364            .broadcast_submit(
1365                instrument_id,
1366                ClientOrderId::from("O-123"),
1367                OrderSide::Buy,
1368                OrderType::Limit,
1369                Quantity::new(100.0, 0),
1370                TimeInForce::Gtc,
1371                Some(Price::new(50000.0, 2)),
1372                None,
1373                None,
1374                None,
1375                None,
1376                None,
1377                false,
1378                false,
1379                None,
1380                None,
1381                None,
1382                None,
1383                None,
1384            )
1385            .await;
1386
1387        let metrics = broadcaster.get_metrics_async().await;
1388        assert_eq!(metrics.total_submits, 1);
1389        assert_eq!(metrics.successful_submits, 1);
1390        assert_eq!(metrics.failed_submits, 0);
1391    }
1392
1393    #[tokio::test]
1394    async fn test_broadcaster_creation_with_pool() {
1395        let report = create_test_report("ORDER-1");
1396        let transports: Vec<TransportClient> = (0..4)
1397            .map(|i| {
1398                let r = report.clone();
1399                create_stub_transport(&format!("client-{i}"), move || {
1400                    let r = r.clone();
1401                    async move { Ok(r) }
1402                })
1403            })
1404            .collect();
1405
1406        let config = SubmitBroadcasterConfig::default();
1407        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1408        let metrics = broadcaster.get_metrics_async().await;
1409        assert_eq!(metrics.total_clients, 4);
1410    }
1411
1412    #[tokio::test]
1413    async fn test_client_stats_collection() {
1414        // Both clients fail so broadcast waits for all of them (no early abort on success).
1415        // This ensures both clients execute and record their stats before the function returns.
1416        let transports = vec![
1417            create_stub_transport("client-0", || async { anyhow::bail!("Timeout error") }),
1418            create_stub_transport("client-1", || async { anyhow::bail!("Connection error") }),
1419        ];
1420
1421        let config = SubmitBroadcasterConfig::default();
1422        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1423
1424        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1425        let _ = broadcaster
1426            .broadcast_submit(
1427                instrument_id,
1428                ClientOrderId::from("O-123"),
1429                OrderSide::Buy,
1430                OrderType::Limit,
1431                Quantity::new(100.0, 0),
1432                TimeInForce::Gtc,
1433                Some(Price::new(50000.0, 2)),
1434                None,
1435                None,
1436                None,
1437                None,
1438                None,
1439                false,
1440                false,
1441                None,
1442                None,
1443                None,
1444                None,
1445                None,
1446            )
1447            .await;
1448
1449        let stats = broadcaster.get_client_stats_async().await;
1450        assert_eq!(stats.len(), 2);
1451
1452        let client0 = stats.iter().find(|s| s.client_id == "client-0").unwrap();
1453        assert_eq!(client0.submit_count, 1);
1454        assert_eq!(client0.error_count, 1);
1455
1456        let client1 = stats.iter().find(|s| s.client_id == "client-1").unwrap();
1457        assert_eq!(client1.submit_count, 1);
1458        assert_eq!(client1.error_count, 1);
1459    }
1460
1461    #[tokio::test]
1462    async fn test_testnet_config_sets_base_url() {
1463        let config = SubmitBroadcasterConfig {
1464            pool_size: 1,
1465            api_key: Some("test_key".to_string()),
1466            api_secret: Some("test_secret".to_string()),
1467            environment: BitmexEnvironment::Testnet,
1468            base_url: None,
1469            ..Default::default()
1470        };
1471
1472        let broadcaster = SubmitBroadcaster::new(config);
1473        assert!(broadcaster.is_ok());
1474    }
1475
1476    #[tokio::test]
1477    async fn test_constructor_honors_default_pool_size() {
1478        let config = SubmitBroadcasterConfig {
1479            api_key: Some("test_key".to_string()),
1480            api_secret: Some("test_secret".to_string()),
1481            base_url: Some("http://127.0.0.1:19999".to_string()),
1482            ..Default::default()
1483        };
1484
1485        let expected_pool = config.pool_size;
1486        let broadcaster = SubmitBroadcaster::new(config).unwrap();
1487        let metrics = broadcaster.get_metrics_async().await;
1488
1489        assert_eq!(metrics.total_clients, expected_pool);
1490    }
1491
1492    #[tokio::test]
1493    async fn test_clone_for_async() {
1494        let report = create_test_report("ORDER-1");
1495        let transports = vec![create_stub_transport("client-0", move || {
1496            let r = report.clone();
1497            async move { Ok(r) }
1498        })];
1499
1500        let config = SubmitBroadcasterConfig::default();
1501        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1502        let cloned = broadcaster.clone_for_async();
1503
1504        // Verify they share the same atomics
1505        broadcaster.total_submits.fetch_add(1, Ordering::Relaxed);
1506        assert_eq!(cloned.total_submits.load(Ordering::Relaxed), 1);
1507    }
1508
1509    #[tokio::test]
1510    async fn test_pattern_matching() {
1511        let config = SubmitBroadcasterConfig {
1512            expected_reject_patterns: vec![
1513                "Duplicate clOrdID".to_string(),
1514                "Order already exists".to_string(),
1515            ],
1516            ..Default::default()
1517        };
1518
1519        let broadcaster = SubmitBroadcaster::new_with_transports(config, vec![]);
1520
1521        assert!(broadcaster.is_expected_reject("Error: Duplicate clOrdID for order"));
1522        assert!(broadcaster.is_expected_reject("Order already exists in system"));
1523        assert!(!broadcaster.is_expected_reject("Rate limit exceeded"));
1524        assert!(!broadcaster.is_expected_reject("Internal server error"));
1525    }
1526
1527    #[tokio::test]
1528    async fn test_submit_metrics_with_mixed_responses() {
1529        let report = create_test_report("ORDER-1");
1530        let report_clone = report.clone();
1531
1532        let transports = vec![
1533            create_stub_transport("client-0", move || {
1534                let report = report_clone.clone();
1535                async move { Ok(report) }
1536            }),
1537            create_stub_transport("client-1", || async { anyhow::bail!("Timeout") }),
1538        ];
1539
1540        let config = SubmitBroadcasterConfig::default();
1541        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1542
1543        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1544        let result = broadcaster
1545            .broadcast_submit(
1546                instrument_id,
1547                ClientOrderId::from("O-123"),
1548                OrderSide::Buy,
1549                OrderType::Limit,
1550                Quantity::new(100.0, 0),
1551                TimeInForce::Gtc,
1552                Some(Price::new(50000.0, 2)),
1553                None,
1554                None,
1555                None,
1556                None,
1557                None,
1558                false,
1559                false,
1560                None,
1561                None,
1562                None,
1563                None,
1564                None,
1565            )
1566            .await;
1567
1568        assert!(result.is_ok());
1569
1570        let metrics = broadcaster.get_metrics_async().await;
1571        assert_eq!(metrics.total_submits, 1);
1572        assert_eq!(metrics.successful_submits, 1);
1573        assert_eq!(metrics.failed_submits, 0);
1574    }
1575
1576    #[tokio::test]
1577    async fn test_metrics_initialization_and_health() {
1578        let report = create_test_report("ORDER-1");
1579        let transports: Vec<TransportClient> = (0..2)
1580            .map(|i| {
1581                let r = report.clone();
1582                create_stub_transport(&format!("client-{i}"), move || {
1583                    let r = r.clone();
1584                    async move { Ok(r) }
1585                })
1586            })
1587            .collect();
1588
1589        let config = SubmitBroadcasterConfig::default();
1590        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1591        let metrics = broadcaster.get_metrics_async().await;
1592
1593        assert_eq!(metrics.total_submits, 0);
1594        assert_eq!(metrics.successful_submits, 0);
1595        assert_eq!(metrics.failed_submits, 0);
1596        assert_eq!(metrics.expected_rejects, 0);
1597        assert_eq!(metrics.total_clients, 2);
1598        assert_eq!(metrics.healthy_clients, 2);
1599    }
1600
1601    #[tokio::test]
1602    async fn test_health_check_task_lifecycle() {
1603        let report = create_test_report("ORDER-1");
1604        let transports: Vec<TransportClient> = (0..2)
1605            .map(|i| {
1606                let r = report.clone();
1607                create_stub_transport(&format!("client-{i}"), move || {
1608                    let r = r.clone();
1609                    async move { Ok(r) }
1610                })
1611            })
1612            .collect();
1613
1614        let config = SubmitBroadcasterConfig::default();
1615        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1616
1617        // Start should spawn health check task
1618        broadcaster.start().await.unwrap();
1619        assert!(broadcaster.running.load(Ordering::Relaxed));
1620        assert!(
1621            broadcaster
1622                .health_check_task
1623                .read()
1624                .await
1625                .as_ref()
1626                .is_some()
1627        );
1628
1629        // Stop should clean up task
1630        broadcaster.stop().await;
1631        assert!(!broadcaster.running.load(Ordering::Relaxed));
1632    }
1633
1634    #[tokio::test]
1635    async fn test_expected_reject_pattern_comprehensive() {
1636        let transports = vec![
1637            create_stub_transport("client-0", || async {
1638                anyhow::bail!("Duplicate clOrdID: O-123 already exists")
1639            }),
1640            create_stub_transport("client-1", || async {
1641                tokio::time::sleep(Duration::from_secs(10)).await;
1642                anyhow::bail!("Should be aborted")
1643            }),
1644        ];
1645
1646        let config = SubmitBroadcasterConfig::default();
1647        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1648
1649        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1650        let result = broadcaster
1651            .broadcast_submit(
1652                instrument_id,
1653                ClientOrderId::from("O-123"),
1654                OrderSide::Buy,
1655                OrderType::Limit,
1656                Quantity::new(100.0, 0),
1657                TimeInForce::Gtc,
1658                Some(Price::new(50000.0, 2)),
1659                None,
1660                None,
1661                None,
1662                None,
1663                None,
1664                false,
1665                false,
1666                None,
1667                None,
1668                None,
1669                None,
1670                None,
1671            )
1672            .await;
1673
1674        // All failed with expected reject
1675        assert!(result.is_err());
1676
1677        let metrics = broadcaster.get_metrics_async().await;
1678        assert_eq!(metrics.expected_rejects, 1);
1679        assert_eq!(metrics.failed_submits, 1);
1680        assert_eq!(metrics.successful_submits, 0);
1681    }
1682
1683    #[tokio::test]
1684    async fn test_client_order_id_suffix_for_multiple_clients() {
1685        use std::sync::{Arc, Mutex};
1686
1687        #[derive(Clone)]
1688        struct CaptureExecutor {
1689            captured_ids: Arc<Mutex<Vec<String>>>,
1690            barrier: Arc<tokio::sync::Barrier>,
1691            report: OrderStatusReport,
1692        }
1693
1694        impl SubmitExecutor for CaptureExecutor {
1695            fn health_check(
1696                &self,
1697            ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + '_>> {
1698                Box::pin(async { Ok(()) })
1699            }
1700
1701            fn submit_order(
1702                &self,
1703                _instrument_id: InstrumentId,
1704                client_order_id: ClientOrderId,
1705                _order_side: OrderSide,
1706                _order_type: OrderType,
1707                _quantity: Quantity,
1708                _time_in_force: TimeInForce,
1709                _price: Option<Price>,
1710                _trigger_price: Option<Price>,
1711                _trigger_type: Option<TriggerType>,
1712                _trailing_offset: Option<f64>,
1713                _trailing_offset_type: Option<TrailingOffsetType>,
1714                _display_qty: Option<Quantity>,
1715                _post_only: bool,
1716                _reduce_only: bool,
1717                _order_list_id: Option<OrderListId>,
1718                _contingency_type: Option<ContingencyType>,
1719                _peg_price_type: Option<BitmexPegPriceType>,
1720                _peg_offset_value: Option<f64>,
1721            ) -> Pin<Box<dyn Future<Output = anyhow::Result<OrderStatusReport>> + Send + '_>>
1722            {
1723                // Capture the client_order_id
1724                self.captured_ids
1725                    .lock()
1726                    .unwrap()
1727                    .push(client_order_id.as_str().to_string());
1728                let report = self.report.clone();
1729                let barrier = Arc::clone(&self.barrier);
1730                // Wait for all tasks to capture their IDs before any completes
1731                // (with concurrent execution, first success aborts others)
1732                Box::pin(async move {
1733                    barrier.wait().await;
1734                    Ok(report)
1735                })
1736            }
1737
1738            fn add_instrument(&self, _instrument: InstrumentAny) {}
1739        }
1740
1741        let captured_ids = Arc::new(Mutex::new(Vec::new()));
1742        let barrier = Arc::new(tokio::sync::Barrier::new(3));
1743        let report = create_test_report("ORDER-1");
1744
1745        let transports = vec![
1746            TransportClient::new(
1747                CaptureExecutor {
1748                    captured_ids: Arc::clone(&captured_ids),
1749                    barrier: Arc::clone(&barrier),
1750                    report: report.clone(),
1751                },
1752                "client-0".to_string(),
1753            ),
1754            TransportClient::new(
1755                CaptureExecutor {
1756                    captured_ids: Arc::clone(&captured_ids),
1757                    barrier: Arc::clone(&barrier),
1758                    report: report.clone(),
1759                },
1760                "client-1".to_string(),
1761            ),
1762            TransportClient::new(
1763                CaptureExecutor {
1764                    captured_ids: Arc::clone(&captured_ids),
1765                    barrier: Arc::clone(&barrier),
1766                    report: report.clone(),
1767                },
1768                "client-2".to_string(),
1769            ),
1770        ];
1771
1772        let config = SubmitBroadcasterConfig::default();
1773        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1774
1775        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1776        let result = broadcaster
1777            .broadcast_submit(
1778                instrument_id,
1779                ClientOrderId::from("O-123"),
1780                OrderSide::Buy,
1781                OrderType::Limit,
1782                Quantity::new(100.0, 0),
1783                TimeInForce::Gtc,
1784                Some(Price::new(50000.0, 2)),
1785                None,
1786                None,
1787                None,
1788                None,
1789                None,
1790                false,
1791                false,
1792                None,
1793                None,
1794                None,
1795                None,
1796                None,
1797            )
1798            .await;
1799
1800        assert!(result.is_ok());
1801
1802        // All transports receive the same client_order_id (no suffixing)
1803        let ids = captured_ids.lock().unwrap();
1804        assert_eq!(ids.len(), 3);
1805        assert!(ids.iter().all(|id| id == "O-123")); // All clients get the same ID
1806    }
1807
1808    #[tokio::test]
1809    async fn test_client_order_id_suffix_with_partial_failure() {
1810        use std::sync::{Arc, Mutex};
1811
1812        #[derive(Clone)]
1813        struct CaptureAndFailExecutor {
1814            captured_ids: Arc<Mutex<Vec<String>>>,
1815            barrier: Arc<tokio::sync::Barrier>,
1816            should_succeed: bool,
1817        }
1818
1819        impl SubmitExecutor for CaptureAndFailExecutor {
1820            fn health_check(
1821                &self,
1822            ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + '_>> {
1823                Box::pin(async { Ok(()) })
1824            }
1825
1826            fn submit_order(
1827                &self,
1828                _instrument_id: InstrumentId,
1829                client_order_id: ClientOrderId,
1830                _order_side: OrderSide,
1831                _order_type: OrderType,
1832                _quantity: Quantity,
1833                _time_in_force: TimeInForce,
1834                _price: Option<Price>,
1835                _trigger_price: Option<Price>,
1836                _trigger_type: Option<TriggerType>,
1837                _trailing_offset: Option<f64>,
1838                _trailing_offset_type: Option<TrailingOffsetType>,
1839                _display_qty: Option<Quantity>,
1840                _post_only: bool,
1841                _reduce_only: bool,
1842                _order_list_id: Option<OrderListId>,
1843                _contingency_type: Option<ContingencyType>,
1844                _peg_price_type: Option<BitmexPegPriceType>,
1845                _peg_offset_value: Option<f64>,
1846            ) -> Pin<Box<dyn Future<Output = anyhow::Result<OrderStatusReport>> + Send + '_>>
1847            {
1848                // Capture the client_order_id
1849                self.captured_ids
1850                    .lock()
1851                    .unwrap()
1852                    .push(client_order_id.as_str().to_string());
1853                let barrier = Arc::clone(&self.barrier);
1854                let should_succeed = self.should_succeed;
1855                // Wait for all tasks to capture their IDs before any completes
1856                // (with concurrent execution, first success aborts others)
1857                Box::pin(async move {
1858                    barrier.wait().await;
1859
1860                    if should_succeed {
1861                        Ok(create_test_report("ORDER-1"))
1862                    } else {
1863                        anyhow::bail!("Network error")
1864                    }
1865                })
1866            }
1867
1868            fn add_instrument(&self, _instrument: InstrumentAny) {}
1869        }
1870
1871        let captured_ids = Arc::new(Mutex::new(Vec::new()));
1872        let barrier = Arc::new(tokio::sync::Barrier::new(2));
1873
1874        let transports = vec![
1875            TransportClient::new(
1876                CaptureAndFailExecutor {
1877                    captured_ids: Arc::clone(&captured_ids),
1878                    barrier: Arc::clone(&barrier),
1879                    should_succeed: false,
1880                },
1881                "client-0".to_string(),
1882            ),
1883            TransportClient::new(
1884                CaptureAndFailExecutor {
1885                    captured_ids: Arc::clone(&captured_ids),
1886                    barrier: Arc::clone(&barrier),
1887                    should_succeed: true,
1888                },
1889                "client-1".to_string(),
1890            ),
1891        ];
1892
1893        let config = SubmitBroadcasterConfig::default();
1894        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1895
1896        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1897        let result = broadcaster
1898            .broadcast_submit(
1899                instrument_id,
1900                ClientOrderId::from("O-456"),
1901                OrderSide::Sell,
1902                OrderType::Market,
1903                Quantity::new(50.0, 0),
1904                TimeInForce::Ioc,
1905                None,
1906                None,
1907                None,
1908                None,
1909                None,
1910                None,
1911                false,
1912                false,
1913                None,
1914                None,
1915                None,
1916                None,
1917                None,
1918            )
1919            .await;
1920
1921        assert!(result.is_ok());
1922
1923        // All transports receive the same client_order_id (no suffixing)
1924        let ids = captured_ids.lock().unwrap();
1925        assert_eq!(ids.len(), 2);
1926        assert!(ids.iter().all(|id| id == "O-456")); // All clients get the same ID
1927    }
1928
1929    #[tokio::test]
1930    async fn test_proxy_urls_populated_from_config() {
1931        let config = SubmitBroadcasterConfig {
1932            pool_size: 3,
1933            api_key: Some("test_key".to_string()),
1934            api_secret: Some("test_secret".to_string()),
1935            proxy_urls: vec![
1936                Some("http://proxy1:8080".to_string()),
1937                Some("http://proxy2:8080".to_string()),
1938                Some("http://proxy3:8080".to_string()),
1939            ],
1940            ..Default::default()
1941        };
1942
1943        assert_eq!(config.proxy_urls.len(), 3);
1944        assert_eq!(config.proxy_urls[0], Some("http://proxy1:8080".to_string()));
1945        assert_eq!(config.proxy_urls[1], Some("http://proxy2:8080".to_string()));
1946        assert_eq!(config.proxy_urls[2], Some("http://proxy3:8080".to_string()));
1947    }
1948}
nautilus_bitmex/broadcast/submitter.rs

nautilus_bitmex/broadcast/
submitter.rs
