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.into(),
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            activation_price: None,
967            trigger_price: None,
968            trigger_type: None,
969            contingency_type: None,
970            expire_time: None,
971            order_list_id: None,
972            venue_position_id: None,
973            linked_order_ids: None,
974            parent_order_id: None,
975            display_qty: None,
976            limit_offset: None,
977            trailing_offset: None,
978            trailing_offset_type: None,
979            post_only: false,
980            reduce_only: false,
981            cancel_reason: None,
982            ts_triggered: None,
983        }
984    }
985
986    fn create_stub_transport<F, Fut>(client_id: &str, handler: F) -> TransportClient
987    where
988        F: Fn() -> Fut + Send + Sync + 'static,
989        Fut: Future<Output = anyhow::Result<OrderStatusReport>> + Send + 'static,
990    {
991        let executor = MockExecutor::new(handler);
992        TransportClient::new(executor, client_id.to_string())
993    }
994
995    #[tokio::test]
996    async fn test_broadcast_submit_immediate_success() {
997        let report = create_test_report("ORDER-1");
998        let report_clone = report.clone();
999
1000        let transports = vec![
1001            create_stub_transport("client-0", move || {
1002                let report = report_clone.clone();
1003                async move { Ok(report) }
1004            }),
1005            create_stub_transport("client-1", || async {
1006                tokio::time::sleep(Duration::from_secs(10)).await;
1007                anyhow::bail!("Should be aborted")
1008            }),
1009        ];
1010
1011        let config = SubmitBroadcasterConfig::default();
1012        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1013
1014        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1015        let result = broadcaster
1016            .broadcast_submit(
1017                instrument_id,
1018                ClientOrderId::from("O-123"),
1019                OrderSide::Buy,
1020                OrderType::Limit,
1021                Quantity::new(100.0, 0),
1022                TimeInForce::Gtc,
1023                Some(Price::new(50000.0, 2)),
1024                None,
1025                None,
1026                None,
1027                None,
1028                None,
1029                false,
1030                false,
1031                None,
1032                None,
1033                None,
1034                None,
1035                None,
1036            )
1037            .await;
1038
1039        assert!(result.is_ok());
1040        let returned_report = result.unwrap();
1041        assert_eq!(returned_report.venue_order_id, report.venue_order_id);
1042
1043        let metrics = broadcaster.get_metrics_async().await;
1044        assert_eq!(metrics.successful_submits, 1);
1045        assert_eq!(metrics.failed_submits, 0);
1046        assert_eq!(metrics.total_submits, 1);
1047    }
1048
1049    #[tokio::test]
1050    async fn test_broadcast_submit_duplicate_clordid_expected() {
1051        let transports = vec![
1052            create_stub_transport("client-0", || async { anyhow::bail!("Duplicate clOrdID") }),
1053            create_stub_transport("client-1", || async { anyhow::bail!("Connection timeout") }),
1054        ];
1055
1056        let config = SubmitBroadcasterConfig::default();
1057        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1058
1059        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1060        let result = broadcaster
1061            .broadcast_submit(
1062                instrument_id,
1063                ClientOrderId::from("O-123"),
1064                OrderSide::Buy,
1065                OrderType::Limit,
1066                Quantity::new(100.0, 0),
1067                TimeInForce::Gtc,
1068                Some(Price::new(50000.0, 2)),
1069                None,
1070                None,
1071                None,
1072                None,
1073                None,
1074                false,
1075                false,
1076                None,
1077                None,
1078                None,
1079                None,
1080                None,
1081            )
1082            .await;
1083
1084        assert!(result.is_err());
1085
1086        let metrics = broadcaster.get_metrics_async().await;
1087        assert_eq!(metrics.expected_rejects, 1);
1088        assert_eq!(metrics.successful_submits, 0);
1089        assert_eq!(metrics.failed_submits, 1);
1090    }
1091
1092    #[tokio::test]
1093    async fn test_broadcast_submit_all_failures() {
1094        let transports = vec![
1095            create_stub_transport("client-0", || async { anyhow::bail!("502 Bad Gateway") }),
1096            create_stub_transport("client-1", || async { anyhow::bail!("Connection refused") }),
1097        ];
1098
1099        let config = SubmitBroadcasterConfig::default();
1100        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1101
1102        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1103        let result = broadcaster
1104            .broadcast_submit(
1105                instrument_id,
1106                ClientOrderId::from("O-456"),
1107                OrderSide::Sell,
1108                OrderType::Market,
1109                Quantity::new(50.0, 0),
1110                TimeInForce::Ioc,
1111                None,
1112                None,
1113                None,
1114                None,
1115                None,
1116                None,
1117                false,
1118                false,
1119                None,
1120                None,
1121                None,
1122                None,
1123                None,
1124            )
1125            .await;
1126
1127        assert!(result.is_err());
1128        assert!(
1129            result
1130                .unwrap_err()
1131                .to_string()
1132                .contains("All submit requests failed")
1133        );
1134
1135        let metrics = broadcaster.get_metrics_async().await;
1136        assert_eq!(metrics.failed_submits, 1);
1137        assert_eq!(metrics.successful_submits, 0);
1138    }
1139
1140    #[tokio::test]
1141    async fn test_broadcast_submit_all_bitmex_refusals_preserves_definitive_outcome() {
1142        let transports = vec![
1143            create_stub_transport("client-0", || async {
1144                Err(anyhow::Error::new(BitmexHttpError::BitmexError {
1145                    error_name: "HTTPError".to_string(),
1146                    message: "Invalid price".to_string(),
1147                }))
1148            }),
1149            create_stub_transport("client-1", || async {
1150                Err(anyhow::Error::new(BitmexHttpError::BitmexError {
1151                    error_name: "HTTPError".to_string(),
1152                    message: "Invalid price".to_string(),
1153                }))
1154            }),
1155        ];
1156
1157        let config = SubmitBroadcasterConfig::default();
1158        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1159
1160        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1161        let result = broadcaster
1162            .broadcast_submit(
1163                instrument_id,
1164                ClientOrderId::from("O-REFUSED"),
1165                OrderSide::Sell,
1166                OrderType::Limit,
1167                Quantity::new(50.0, 0),
1168                TimeInForce::Gtc,
1169                Some(Price::new(50000.0, 2)),
1170                None,
1171                None,
1172                None,
1173                None,
1174                None,
1175                false,
1176                false,
1177                None,
1178                None,
1179                None,
1180                None,
1181                None,
1182            )
1183            .await;
1184
1185        let err = result.unwrap_err().to_string();
1186
1187        assert!(err.starts_with(DEFINITIVE_SUBMIT_REJECTION));
1188
1189        let metrics = broadcaster.get_metrics_async().await;
1190        assert_eq!(metrics.failed_submits, 1);
1191        assert_eq!(metrics.successful_submits, 0);
1192    }
1193
1194    #[tokio::test]
1195    async fn test_broadcast_submit_mixed_refusal_and_network_failure_stays_ambiguous() {
1196        let transports = vec![
1197            create_stub_transport("client-0", || async {
1198                Err(anyhow::Error::new(BitmexHttpError::BitmexError {
1199                    error_name: "HTTPError".to_string(),
1200                    message: "Invalid price".to_string(),
1201                }))
1202            }),
1203            create_stub_transport("client-1", || async { anyhow::bail!("Connection refused") }),
1204        ];
1205
1206        let config = SubmitBroadcasterConfig::default();
1207        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1208
1209        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1210        let result = broadcaster
1211            .broadcast_submit(
1212                instrument_id,
1213                ClientOrderId::from("O-MIXED-FAILURE"),
1214                OrderSide::Sell,
1215                OrderType::Limit,
1216                Quantity::from("50"),
1217                TimeInForce::Gtc,
1218                Some(Price::from("50000.00")),
1219                None,
1220                None,
1221                None,
1222                None,
1223                None,
1224                false,
1225                false,
1226                None,
1227                None,
1228                None,
1229                None,
1230                None,
1231            )
1232            .await;
1233
1234        let err = result.unwrap_err().to_string();
1235
1236        assert!(err.starts_with("All submit requests failed"));
1237        assert!(!err.starts_with(DEFINITIVE_SUBMIT_REJECTION));
1238
1239        let metrics = broadcaster.get_metrics_async().await;
1240        assert_eq!(metrics.failed_submits, 1);
1241        assert_eq!(metrics.successful_submits, 0);
1242    }
1243
1244    #[tokio::test]
1245    async fn test_broadcast_submit_no_healthy_clients() {
1246        let transport =
1247            create_stub_transport("client-0", || async { Ok(create_test_report("ORDER-1")) });
1248        transport.healthy.store(false, Ordering::Relaxed);
1249
1250        let config = SubmitBroadcasterConfig::default();
1251        let broadcaster = SubmitBroadcaster::new_with_transports(config, vec![transport]);
1252
1253        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1254        let result = broadcaster
1255            .broadcast_submit(
1256                instrument_id,
1257                ClientOrderId::from("O-789"),
1258                OrderSide::Buy,
1259                OrderType::Limit,
1260                Quantity::new(100.0, 0),
1261                TimeInForce::Gtc,
1262                Some(Price::new(50000.0, 2)),
1263                None,
1264                None,
1265                None,
1266                None,
1267                None,
1268                false,
1269                false,
1270                None,
1271                None,
1272                None,
1273                None,
1274                None,
1275            )
1276            .await;
1277
1278        assert!(result.is_err());
1279        assert!(
1280            result
1281                .unwrap_err()
1282                .to_string()
1283                .contains("No healthy transport clients available")
1284        );
1285
1286        let metrics = broadcaster.get_metrics_async().await;
1287        assert_eq!(metrics.failed_submits, 1);
1288    }
1289
1290    #[tokio::test]
1291    async fn test_default_config() {
1292        let report = create_test_report("ORDER-1");
1293        let transports: Vec<TransportClient> = (0..3)
1294            .map(|i| {
1295                let r = report.clone();
1296                create_stub_transport(&format!("client-{i}"), move || {
1297                    let r = r.clone();
1298                    async move { Ok(r) }
1299                })
1300            })
1301            .collect();
1302
1303        let config = SubmitBroadcasterConfig::default();
1304        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1305        let metrics = broadcaster.get_metrics_async().await;
1306
1307        assert_eq!(metrics.total_clients, 3);
1308    }
1309
1310    #[tokio::test]
1311    async fn test_broadcaster_lifecycle() {
1312        let report = create_test_report("ORDER-1");
1313        let transports: Vec<TransportClient> = (0..2)
1314            .map(|i| {
1315                let r = report.clone();
1316                create_stub_transport(&format!("client-{i}"), move || {
1317                    let r = r.clone();
1318                    async move { Ok(r) }
1319                })
1320            })
1321            .collect();
1322
1323        let config = SubmitBroadcasterConfig::default();
1324        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1325
1326        // Should not be running initially
1327        assert!(!broadcaster.running.load(Ordering::Relaxed));
1328
1329        // Start broadcaster
1330        let start_result = broadcaster.start().await;
1331        assert!(start_result.is_ok());
1332        assert!(broadcaster.running.load(Ordering::Relaxed));
1333
1334        // Starting again should be idempotent
1335        let start_again = broadcaster.start().await;
1336        assert!(start_again.is_ok());
1337
1338        // Stop broadcaster
1339        broadcaster.stop().await;
1340        assert!(!broadcaster.running.load(Ordering::Relaxed));
1341
1342        // Stopping again should be safe
1343        broadcaster.stop().await;
1344        assert!(!broadcaster.running.load(Ordering::Relaxed));
1345    }
1346
1347    #[tokio::test]
1348    async fn test_broadcast_submit_metrics_increment() {
1349        let report = create_test_report("ORDER-1");
1350        let report_clone = report.clone();
1351
1352        let transports = vec![create_stub_transport("client-0", move || {
1353            let report = report_clone.clone();
1354            async move { Ok(report) }
1355        })];
1356
1357        let config = SubmitBroadcasterConfig::default();
1358        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1359
1360        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1361        let _ = broadcaster
1362            .broadcast_submit(
1363                instrument_id,
1364                ClientOrderId::from("O-123"),
1365                OrderSide::Buy,
1366                OrderType::Limit,
1367                Quantity::new(100.0, 0),
1368                TimeInForce::Gtc,
1369                Some(Price::new(50000.0, 2)),
1370                None,
1371                None,
1372                None,
1373                None,
1374                None,
1375                false,
1376                false,
1377                None,
1378                None,
1379                None,
1380                None,
1381                None,
1382            )
1383            .await;
1384
1385        let metrics = broadcaster.get_metrics_async().await;
1386        assert_eq!(metrics.total_submits, 1);
1387        assert_eq!(metrics.successful_submits, 1);
1388        assert_eq!(metrics.failed_submits, 0);
1389    }
1390
1391    #[tokio::test]
1392    async fn test_broadcaster_creation_with_pool() {
1393        let report = create_test_report("ORDER-1");
1394        let transports: Vec<TransportClient> = (0..4)
1395            .map(|i| {
1396                let r = report.clone();
1397                create_stub_transport(&format!("client-{i}"), move || {
1398                    let r = r.clone();
1399                    async move { Ok(r) }
1400                })
1401            })
1402            .collect();
1403
1404        let config = SubmitBroadcasterConfig::default();
1405        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1406        let metrics = broadcaster.get_metrics_async().await;
1407        assert_eq!(metrics.total_clients, 4);
1408    }
1409
1410    #[tokio::test]
1411    async fn test_client_stats_collection() {
1412        // Both clients fail so broadcast waits for all of them (no early abort on success).
1413        // This ensures both clients execute and record their stats before the function returns.
1414        let transports = vec![
1415            create_stub_transport("client-0", || async { anyhow::bail!("Timeout error") }),
1416            create_stub_transport("client-1", || async { anyhow::bail!("Connection error") }),
1417        ];
1418
1419        let config = SubmitBroadcasterConfig::default();
1420        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1421
1422        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1423        let _ = broadcaster
1424            .broadcast_submit(
1425                instrument_id,
1426                ClientOrderId::from("O-123"),
1427                OrderSide::Buy,
1428                OrderType::Limit,
1429                Quantity::new(100.0, 0),
1430                TimeInForce::Gtc,
1431                Some(Price::new(50000.0, 2)),
1432                None,
1433                None,
1434                None,
1435                None,
1436                None,
1437                false,
1438                false,
1439                None,
1440                None,
1441                None,
1442                None,
1443                None,
1444            )
1445            .await;
1446
1447        let stats = broadcaster.get_client_stats_async().await;
1448        assert_eq!(stats.len(), 2);
1449
1450        let client0 = stats.iter().find(|s| s.client_id == "client-0").unwrap();
1451        assert_eq!(client0.submit_count, 1);
1452        assert_eq!(client0.error_count, 1);
1453
1454        let client1 = stats.iter().find(|s| s.client_id == "client-1").unwrap();
1455        assert_eq!(client1.submit_count, 1);
1456        assert_eq!(client1.error_count, 1);
1457    }
1458
1459    #[tokio::test]
1460    async fn test_testnet_config_sets_base_url() {
1461        let config = SubmitBroadcasterConfig {
1462            pool_size: 1,
1463            api_key: Some("test_key".to_string()),
1464            api_secret: Some("test_secret".to_string()),
1465            environment: BitmexEnvironment::Testnet,
1466            base_url: None,
1467            ..Default::default()
1468        };
1469
1470        let broadcaster = SubmitBroadcaster::new(config);
1471        assert!(broadcaster.is_ok());
1472    }
1473
1474    #[tokio::test]
1475    async fn test_constructor_honors_default_pool_size() {
1476        let config = SubmitBroadcasterConfig {
1477            api_key: Some("test_key".to_string()),
1478            api_secret: Some("test_secret".to_string()),
1479            base_url: Some("http://127.0.0.1:19999".to_string()),
1480            ..Default::default()
1481        };
1482
1483        let expected_pool = config.pool_size;
1484        let broadcaster = SubmitBroadcaster::new(config).unwrap();
1485        let metrics = broadcaster.get_metrics_async().await;
1486
1487        assert_eq!(metrics.total_clients, expected_pool);
1488    }
1489
1490    #[tokio::test]
1491    async fn test_clone_for_async() {
1492        let report = create_test_report("ORDER-1");
1493        let transports = vec![create_stub_transport("client-0", move || {
1494            let r = report.clone();
1495            async move { Ok(r) }
1496        })];
1497
1498        let config = SubmitBroadcasterConfig::default();
1499        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1500        let cloned = broadcaster.clone_for_async();
1501
1502        // Verify they share the same atomics
1503        broadcaster.total_submits.fetch_add(1, Ordering::Relaxed);
1504        assert_eq!(cloned.total_submits.load(Ordering::Relaxed), 1);
1505    }
1506
1507    #[tokio::test]
1508    async fn test_pattern_matching() {
1509        let config = SubmitBroadcasterConfig {
1510            expected_reject_patterns: vec![
1511                "Duplicate clOrdID".to_string(),
1512                "Order already exists".to_string(),
1513            ],
1514            ..Default::default()
1515        };
1516
1517        let broadcaster = SubmitBroadcaster::new_with_transports(config, vec![]);
1518
1519        assert!(broadcaster.is_expected_reject("Error: Duplicate clOrdID for order"));
1520        assert!(broadcaster.is_expected_reject("Order already exists in system"));
1521        assert!(!broadcaster.is_expected_reject("Rate limit exceeded"));
1522        assert!(!broadcaster.is_expected_reject("Internal server error"));
1523    }
1524
1525    #[tokio::test]
1526    async fn test_submit_metrics_with_mixed_responses() {
1527        let report = create_test_report("ORDER-1");
1528        let report_clone = report.clone();
1529
1530        let transports = vec![
1531            create_stub_transport("client-0", move || {
1532                let report = report_clone.clone();
1533                async move { Ok(report) }
1534            }),
1535            create_stub_transport("client-1", || async { anyhow::bail!("Timeout") }),
1536        ];
1537
1538        let config = SubmitBroadcasterConfig::default();
1539        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1540
1541        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1542        let result = broadcaster
1543            .broadcast_submit(
1544                instrument_id,
1545                ClientOrderId::from("O-123"),
1546                OrderSide::Buy,
1547                OrderType::Limit,
1548                Quantity::new(100.0, 0),
1549                TimeInForce::Gtc,
1550                Some(Price::new(50000.0, 2)),
1551                None,
1552                None,
1553                None,
1554                None,
1555                None,
1556                false,
1557                false,
1558                None,
1559                None,
1560                None,
1561                None,
1562                None,
1563            )
1564            .await;
1565
1566        assert!(result.is_ok());
1567
1568        let metrics = broadcaster.get_metrics_async().await;
1569        assert_eq!(metrics.total_submits, 1);
1570        assert_eq!(metrics.successful_submits, 1);
1571        assert_eq!(metrics.failed_submits, 0);
1572    }
1573
1574    #[tokio::test]
1575    async fn test_metrics_initialization_and_health() {
1576        let report = create_test_report("ORDER-1");
1577        let transports: Vec<TransportClient> = (0..2)
1578            .map(|i| {
1579                let r = report.clone();
1580                create_stub_transport(&format!("client-{i}"), move || {
1581                    let r = r.clone();
1582                    async move { Ok(r) }
1583                })
1584            })
1585            .collect();
1586
1587        let config = SubmitBroadcasterConfig::default();
1588        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1589        let metrics = broadcaster.get_metrics_async().await;
1590
1591        assert_eq!(metrics.total_submits, 0);
1592        assert_eq!(metrics.successful_submits, 0);
1593        assert_eq!(metrics.failed_submits, 0);
1594        assert_eq!(metrics.expected_rejects, 0);
1595        assert_eq!(metrics.total_clients, 2);
1596        assert_eq!(metrics.healthy_clients, 2);
1597    }
1598
1599    #[tokio::test]
1600    async fn test_health_check_task_lifecycle() {
1601        let report = create_test_report("ORDER-1");
1602        let transports: Vec<TransportClient> = (0..2)
1603            .map(|i| {
1604                let r = report.clone();
1605                create_stub_transport(&format!("client-{i}"), move || {
1606                    let r = r.clone();
1607                    async move { Ok(r) }
1608                })
1609            })
1610            .collect();
1611
1612        let config = SubmitBroadcasterConfig::default();
1613        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1614
1615        // Start should spawn health check task
1616        broadcaster.start().await.unwrap();
1617        assert!(broadcaster.running.load(Ordering::Relaxed));
1618        assert!(
1619            broadcaster
1620                .health_check_task
1621                .read()
1622                .await
1623                .as_ref()
1624                .is_some()
1625        );
1626
1627        // Stop should clean up task
1628        broadcaster.stop().await;
1629        assert!(!broadcaster.running.load(Ordering::Relaxed));
1630    }
1631
1632    #[tokio::test]
1633    async fn test_expected_reject_pattern_comprehensive() {
1634        let transports = vec![
1635            create_stub_transport("client-0", || async {
1636                anyhow::bail!("Duplicate clOrdID: O-123 already exists")
1637            }),
1638            create_stub_transport("client-1", || async { anyhow::bail!("Connection timeout") }),
1639        ];
1640
1641        let config = SubmitBroadcasterConfig::default();
1642        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1643
1644        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1645        let result = broadcaster
1646            .broadcast_submit(
1647                instrument_id,
1648                ClientOrderId::from("O-123"),
1649                OrderSide::Buy,
1650                OrderType::Limit,
1651                Quantity::new(100.0, 0),
1652                TimeInForce::Gtc,
1653                Some(Price::new(50000.0, 2)),
1654                None,
1655                None,
1656                None,
1657                None,
1658                None,
1659                false,
1660                false,
1661                None,
1662                None,
1663                None,
1664                None,
1665                None,
1666            )
1667            .await;
1668
1669        // All failed with expected reject
1670        assert!(result.is_err());
1671
1672        let metrics = broadcaster.get_metrics_async().await;
1673        assert_eq!(metrics.expected_rejects, 1);
1674        assert_eq!(metrics.failed_submits, 1);
1675        assert_eq!(metrics.successful_submits, 0);
1676    }
1677
1678    #[tokio::test]
1679    async fn test_client_order_id_suffix_for_multiple_clients() {
1680        use std::sync::Arc;
1681
1682        use parking_lot::Mutex;
1683
1684        #[derive(Clone)]
1685        struct CaptureExecutor {
1686            captured_ids: Arc<Mutex<Vec<String>>>,
1687            barrier: Arc<tokio::sync::Barrier>,
1688            report: OrderStatusReport,
1689        }
1690
1691        impl SubmitExecutor for CaptureExecutor {
1692            fn health_check(
1693                &self,
1694            ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + '_>> {
1695                Box::pin(async { Ok(()) })
1696            }
1697
1698            fn submit_order(
1699                &self,
1700                _instrument_id: InstrumentId,
1701                client_order_id: ClientOrderId,
1702                _order_side: OrderSide,
1703                _order_type: OrderType,
1704                _quantity: Quantity,
1705                _time_in_force: TimeInForce,
1706                _price: Option<Price>,
1707                _trigger_price: Option<Price>,
1708                _trigger_type: Option<TriggerType>,
1709                _trailing_offset: Option<f64>,
1710                _trailing_offset_type: Option<TrailingOffsetType>,
1711                _display_qty: Option<Quantity>,
1712                _post_only: bool,
1713                _reduce_only: bool,
1714                _order_list_id: Option<OrderListId>,
1715                _contingency_type: Option<ContingencyType>,
1716                _peg_price_type: Option<BitmexPegPriceType>,
1717                _peg_offset_value: Option<f64>,
1718            ) -> Pin<Box<dyn Future<Output = anyhow::Result<OrderStatusReport>> + Send + '_>>
1719            {
1720                // Capture the client_order_id
1721                self.captured_ids
1722                    .lock()
1723                    .push(client_order_id.as_str().to_string());
1724                let report = self.report.clone();
1725                let barrier = Arc::clone(&self.barrier);
1726                // Wait for all tasks to capture their IDs before any completes
1727                // (with concurrent execution, first success aborts others)
1728                Box::pin(async move {
1729                    barrier.wait().await;
1730                    Ok(report)
1731                })
1732            }
1733
1734            fn add_instrument(&self, _instrument: InstrumentAny) {}
1735        }
1736
1737        let captured_ids = Arc::new(Mutex::new(Vec::new()));
1738        let barrier = Arc::new(tokio::sync::Barrier::new(3));
1739        let report = create_test_report("ORDER-1");
1740
1741        let transports = vec![
1742            TransportClient::new(
1743                CaptureExecutor {
1744                    captured_ids: Arc::clone(&captured_ids),
1745                    barrier: Arc::clone(&barrier),
1746                    report: report.clone(),
1747                },
1748                "client-0".to_string(),
1749            ),
1750            TransportClient::new(
1751                CaptureExecutor {
1752                    captured_ids: Arc::clone(&captured_ids),
1753                    barrier: Arc::clone(&barrier),
1754                    report: report.clone(),
1755                },
1756                "client-1".to_string(),
1757            ),
1758            TransportClient::new(
1759                CaptureExecutor {
1760                    captured_ids: Arc::clone(&captured_ids),
1761                    barrier: Arc::clone(&barrier),
1762                    report: report.clone(),
1763                },
1764                "client-2".to_string(),
1765            ),
1766        ];
1767
1768        let config = SubmitBroadcasterConfig::default();
1769        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1770
1771        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1772        let result = broadcaster
1773            .broadcast_submit(
1774                instrument_id,
1775                ClientOrderId::from("O-123"),
1776                OrderSide::Buy,
1777                OrderType::Limit,
1778                Quantity::new(100.0, 0),
1779                TimeInForce::Gtc,
1780                Some(Price::new(50000.0, 2)),
1781                None,
1782                None,
1783                None,
1784                None,
1785                None,
1786                false,
1787                false,
1788                None,
1789                None,
1790                None,
1791                None,
1792                None,
1793            )
1794            .await;
1795
1796        assert!(result.is_ok());
1797
1798        // All transports receive the same client_order_id (no suffixing)
1799        let ids = captured_ids.lock();
1800        assert_eq!(ids.len(), 3);
1801        assert!(ids.iter().all(|id| id == "O-123")); // All clients get the same ID
1802    }
1803
1804    #[tokio::test]
1805    async fn test_client_order_id_suffix_with_partial_failure() {
1806        use std::sync::Arc;
1807
1808        use parking_lot::Mutex;
1809
1810        #[derive(Clone)]
1811        struct CaptureAndFailExecutor {
1812            captured_ids: Arc<Mutex<Vec<String>>>,
1813            barrier: Arc<tokio::sync::Barrier>,
1814            should_succeed: bool,
1815        }
1816
1817        impl SubmitExecutor for CaptureAndFailExecutor {
1818            fn health_check(
1819                &self,
1820            ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + '_>> {
1821                Box::pin(async { Ok(()) })
1822            }
1823
1824            fn submit_order(
1825                &self,
1826                _instrument_id: InstrumentId,
1827                client_order_id: ClientOrderId,
1828                _order_side: OrderSide,
1829                _order_type: OrderType,
1830                _quantity: Quantity,
1831                _time_in_force: TimeInForce,
1832                _price: Option<Price>,
1833                _trigger_price: Option<Price>,
1834                _trigger_type: Option<TriggerType>,
1835                _trailing_offset: Option<f64>,
1836                _trailing_offset_type: Option<TrailingOffsetType>,
1837                _display_qty: Option<Quantity>,
1838                _post_only: bool,
1839                _reduce_only: bool,
1840                _order_list_id: Option<OrderListId>,
1841                _contingency_type: Option<ContingencyType>,
1842                _peg_price_type: Option<BitmexPegPriceType>,
1843                _peg_offset_value: Option<f64>,
1844            ) -> Pin<Box<dyn Future<Output = anyhow::Result<OrderStatusReport>> + Send + '_>>
1845            {
1846                // Capture the client_order_id
1847                self.captured_ids
1848                    .lock()
1849                    .push(client_order_id.as_str().to_string());
1850                let barrier = Arc::clone(&self.barrier);
1851                let should_succeed = self.should_succeed;
1852                // Wait for all tasks to capture their IDs before any completes
1853                // (with concurrent execution, first success aborts others)
1854                Box::pin(async move {
1855                    barrier.wait().await;
1856
1857                    if should_succeed {
1858                        Ok(create_test_report("ORDER-1"))
1859                    } else {
1860                        anyhow::bail!("Network error")
1861                    }
1862                })
1863            }
1864
1865            fn add_instrument(&self, _instrument: InstrumentAny) {}
1866        }
1867
1868        let captured_ids = Arc::new(Mutex::new(Vec::new()));
1869        let barrier = Arc::new(tokio::sync::Barrier::new(2));
1870
1871        let transports = vec![
1872            TransportClient::new(
1873                CaptureAndFailExecutor {
1874                    captured_ids: Arc::clone(&captured_ids),
1875                    barrier: Arc::clone(&barrier),
1876                    should_succeed: false,
1877                },
1878                "client-0".to_string(),
1879            ),
1880            TransportClient::new(
1881                CaptureAndFailExecutor {
1882                    captured_ids: Arc::clone(&captured_ids),
1883                    barrier: Arc::clone(&barrier),
1884                    should_succeed: true,
1885                },
1886                "client-1".to_string(),
1887            ),
1888        ];
1889
1890        let config = SubmitBroadcasterConfig::default();
1891        let broadcaster = SubmitBroadcaster::new_with_transports(config, transports);
1892
1893        let instrument_id = InstrumentId::from_str("XBTUSD.BITMEX").unwrap();
1894        let result = broadcaster
1895            .broadcast_submit(
1896                instrument_id,
1897                ClientOrderId::from("O-456"),
1898                OrderSide::Sell,
1899                OrderType::Market,
1900                Quantity::new(50.0, 0),
1901                TimeInForce::Ioc,
1902                None,
1903                None,
1904                None,
1905                None,
1906                None,
1907                None,
1908                false,
1909                false,
1910                None,
1911                None,
1912                None,
1913                None,
1914                None,
1915            )
1916            .await;
1917
1918        assert!(result.is_ok());
1919
1920        // All transports receive the same client_order_id (no suffixing)
1921        let ids = captured_ids.lock();
1922        assert_eq!(ids.len(), 2);
1923        assert!(ids.iter().all(|id| id == "O-456")); // All clients get the same ID
1924    }
1925
1926    #[tokio::test]
1927    async fn test_proxy_urls_populated_from_config() {
1928        let config = SubmitBroadcasterConfig {
1929            pool_size: 3,
1930            api_key: Some("test_key".to_string()),
1931            api_secret: Some("test_secret".to_string()),
1932            proxy_urls: vec![
1933                Some("http://proxy1:8080".to_string()),
1934                Some("http://proxy2:8080".to_string()),
1935                Some("http://proxy3:8080".to_string()),
1936            ],
1937            ..Default::default()
1938        };
1939
1940        assert_eq!(config.proxy_urls.len(), 3);
1941        assert_eq!(config.proxy_urls[0], Some("http://proxy1:8080".to_string()));
1942        assert_eq!(config.proxy_urls[1], Some("http://proxy2:8080".to_string()));
1943        assert_eq!(config.proxy_urls[2], Some("http://proxy3:8080".to_string()));
1944    }
1945}
nautilus_bitmex/broadcast/submitter.rs

nautilus_bitmex/broadcast/
submitter.rs
