nautilus_betfair/http/
client.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//! Betfair HTTP client implementation.
17
18use std::{
19    collections::HashMap,
20    num::NonZeroU32,
21    sync::{
22        Arc,
23        atomic::{AtomicBool, AtomicU64, Ordering},
24    },
25};
26
27use nautilus_core::string::urlencoding;
28use nautilus_network::{
29    http::{HttpClient, Method},
30    ratelimiter::quota::Quota,
31    retry::{RetryConfig, RetryError, RetryManager},
32};
33use serde::{Deserialize, Serialize, de::DeserializeOwned};
34use tokio_util::sync::CancellationToken;
35
36use super::{
37    error::BetfairHttpError,
38    models::{LoginResponse, LoginStatus},
39};
40use crate::common::{
41    consts::{
42        BETFAIR_ACCOUNTS_URL, BETFAIR_BETTING_URL, BETFAIR_IDENTITY_LOGIN_URL,
43        BETFAIR_KEEP_ALIVE_URL, BETFAIR_NAVIGATION_URL, BETFAIR_RATE_LIMIT_DEFAULT,
44        BETFAIR_RATE_LIMIT_ORDERS, HEADER_X_APPLICATION, HEADER_X_AUTHENTICATION,
45    },
46    credential::BetfairCredential,
47};
48
49// Keep the final dispatch 15 seconds inside Betfair's 60-second customerRef window
50const ORDER_RETRY_MAX_ELAPSED_MS: u64 = 45_000;
51
52/// Betfair JSON-RPC request envelope.
53#[derive(Debug, Serialize)]
54struct JsonRpcRequest<P: Serialize> {
55    jsonrpc: &'static str,
56    method: String,
57    params: P,
58    id: u64,
59}
60
61/// Betfair JSON-RPC response envelope.
62#[derive(Debug, Deserialize)]
63struct JsonRpcResponse<T> {
64    result: Option<T>,
65    error: Option<JsonRpcError>,
66}
67
68/// JSON-RPC error object.
69#[derive(Debug, Deserialize)]
70struct JsonRpcError {
71    code: i64,
72    message: String,
73    data: Option<serde_json::Value>,
74}
75
76#[derive(Debug, Deserialize)]
77#[serde(rename_all = "camelCase")]
78struct JsonRpcApiException {
79    error_code: String,
80    error_details: Option<String>,
81}
82
83impl JsonRpcError {
84    fn api_exception(&self) -> Option<JsonRpcApiException> {
85        let data = self.data.as_ref()?;
86        let exception_name = data.get("exceptionname")?.as_str()?;
87        serde_json::from_value(data.get(exception_name)?.clone()).ok()
88    }
89}
90
91/// Betfair HTTP client for raw API operations.
92///
93/// Handles session-token authentication, JSON-RPC protocol, form-encoded
94/// identity requests, REST navigation, rate limiting, and retry logic.
95#[derive(Debug)]
96pub struct BetfairHttpClient {
97    client: HttpClient,
98    credential: BetfairCredential,
99    session_token: Arc<tokio::sync::RwLock<Option<String>>>,
100    retry_manager: RetryManager<BetfairHttpError>,
101    order_retry_manager: RetryManager<BetfairHttpError>,
102    cancellation_token: parking_lot::Mutex<CancellationToken>,
103    connect_lock: tokio::sync::Mutex<()>,
104    request_id: AtomicU64,
105    url_identity_login: String,
106    url_keep_alive: String,
107    url_betting: String,
108    url_accounts: String,
109    url_navigation: String,
110}
111
112impl BetfairHttpClient {
113    /// Creates a new [`BetfairHttpClient`].
114    ///
115    /// # Errors
116    ///
117    /// Returns an error if the HTTP client cannot be created.
118    pub fn new(
119        credential: BetfairCredential,
120        timeout_secs: Option<u64>,
121        max_retries: Option<u32>,
122        retry_delay_ms: Option<u64>,
123        proxy_url: Option<String>,
124        request_rate_per_second: Option<u32>,
125        order_request_rate_per_second: Option<u32>,
126    ) -> Result<Self, BetfairHttpError> {
127        let retry_config = RetryConfig {
128            max_retries: max_retries.unwrap_or(3),
129            initial_delay_ms: retry_delay_ms.unwrap_or(1000),
130            max_delay_ms: 10_000,
131            backoff_factor: 2.0,
132            jitter_ms: 500,
133            operation_timeout_ms: Some(30_000),
134            immediate_first: false,
135            max_elapsed_ms: Some(120_000),
136        };
137        let order_retry_config = RetryConfig {
138            max_elapsed_ms: Some(ORDER_RETRY_MAX_ELAPSED_MS),
139            ..retry_config
140        };
141
142        Ok(Self {
143            client: HttpClient::builder()
144                .keyed_quotas(Self::rate_limiter_quotas(
145                    request_rate_per_second.unwrap_or(5),
146                    order_request_rate_per_second.unwrap_or(20),
147                )?)
148                .maybe_default_quota(Self::default_quota(request_rate_per_second.unwrap_or(5))?)
149                .maybe_timeout_secs(timeout_secs)
150                .maybe_proxy_url(proxy_url)
151                .build()
152                .map_err(|e| anyhow::anyhow!("Failed to create HTTP client: {e}"))?,
153            credential,
154            session_token: Arc::new(tokio::sync::RwLock::new(None)),
155            retry_manager: RetryManager::new(retry_config),
156            order_retry_manager: RetryManager::new(order_retry_config),
157            cancellation_token: parking_lot::Mutex::new(CancellationToken::new()),
158            connect_lock: tokio::sync::Mutex::new(()),
159            request_id: AtomicU64::new(1),
160            url_identity_login: BETFAIR_IDENTITY_LOGIN_URL.to_string(),
161            url_keep_alive: BETFAIR_KEEP_ALIVE_URL.to_string(),
162            url_betting: BETFAIR_BETTING_URL.to_string(),
163            url_accounts: BETFAIR_ACCOUNTS_URL.to_string(),
164            url_navigation: BETFAIR_NAVIGATION_URL.to_string(),
165        })
166    }
167
168    /// Overrides the API base URLs (for testing with mock servers).
169    ///
170    /// The keep-alive URL is derived from `identity_login` by replacing the
171    /// path with `/keepAlive`.
172    #[must_use]
173    pub fn with_urls(
174        mut self,
175        identity_login: String,
176        betting: String,
177        accounts: String,
178        navigation: String,
179    ) -> Self {
180        // Derive keep-alive from same host as login
181        if let Some(base) = identity_login.rfind('/') {
182            self.url_keep_alive = format!("{}/keepAlive", &identity_login[..base]);
183        }
184        self.url_identity_login = identity_login;
185        self.url_betting = betting;
186        self.url_accounts = accounts;
187        self.url_navigation = navigation;
188        self
189    }
190
191    /// Returns a clone of the current cancellation token for this client.
192    ///
193    /// `disconnect()` cancels and replaces the token, so callers should fetch
194    /// a fresh clone for each operation rather than holding one long-term.
195    pub fn cancellation_token(&self) -> CancellationToken {
196        self.cancellation_token.lock().clone()
197    }
198
199    /// Returns the current session token, if authenticated.
200    pub async fn session_token(&self) -> Option<String> {
201        self.session_token.read().await.clone()
202    }
203
204    /// Runs a synchronous token publication while session mutation is serialized.
205    pub(crate) async fn with_session_token<T>(&self, publish: impl FnOnce(&str) -> T) -> Option<T> {
206        let _guard = self.connect_lock.lock().await;
207        self.session_token.read().await.as_deref().map(publish)
208    }
209
210    /// Returns whether the client has an active session.
211    pub async fn is_connected(&self) -> bool {
212        self.session_token.read().await.is_some()
213    }
214
215    /// Returns the application key.
216    #[must_use]
217    pub fn app_key(&self) -> &str {
218        self.credential.app_key()
219    }
220
221    /// Authenticates with Betfair using interactive (non-cert) login.
222    ///
223    /// Sends credentials to the Identity API and stores the returned
224    /// session token for subsequent requests.
225    ///
226    /// # Errors
227    ///
228    /// Returns an error if the login request fails or authentication
229    /// is rejected.
230    pub async fn connect(&self) -> Result<(), BetfairHttpError> {
231        // Serialize session mutations so an older keep-alive response cannot overwrite a newer
232        // full login. This matches Python's per-instance asyncio.Lock around the same path.
233        let _guard = self.connect_lock.lock().await;
234
235        if self.session_token.read().await.is_some() {
236            log::debug!("Session token exists (already connected), skipping");
237            return Ok(());
238        }
239
240        let token = self.login().await?;
241        *self.session_token.write().await = Some(token);
242        Ok(())
243    }
244
245    /// Resets the session and re-authenticates.
246    ///
247    /// # Errors
248    ///
249    /// Returns an error if re-authentication fails.
250    pub async fn reconnect(&self) -> Result<(), BetfairHttpError> {
251        self.reconnect_with_token().await.map(drop)
252    }
253
254    /// Resets the session, re-authenticates, and returns the replacement token.
255    ///
256    /// # Errors
257    ///
258    /// Returns an error if re-authentication fails.
259    pub(crate) async fn reconnect_with_token(&self) -> Result<String, BetfairHttpError> {
260        let _guard = self.connect_lock.lock().await;
261        log::info!("Betfair reconnecting...");
262        *self.session_token.write().await = None;
263        let token = self.login().await?;
264        *self.session_token.write().await = Some(token.clone());
265        Ok(token)
266    }
267
268    async fn login(&self) -> Result<String, BetfairHttpError> {
269        let form_body = format!(
270            "username={}&password={}",
271            urlencoding::encode(self.credential.username()),
272            urlencoding::encode(self.credential.password()),
273        );
274
275        let resp_bytes = self
276            .send_identity(&self.url_identity_login, form_body.into_bytes())
277            .await?;
278
279        let resp: LoginResponse = serde_json::from_slice(&resp_bytes)?;
280
281        if resp.status == LoginStatus::Success {
282            log::debug!("Betfair login successful");
283            Ok(resp.token)
284        } else {
285            Err(BetfairHttpError::LoginFailed {
286                status: resp.error.unwrap_or_else(|| format!("{:?}", resp.status)),
287            })
288        }
289    }
290
291    /// Clears the session token, cancels any in-flight retries, and primes a
292    /// fresh cancellation token for the next session.
293    pub async fn disconnect(&self) {
294        log::info!("Betfair disconnecting...");
295        let _guard = self.connect_lock.lock().await;
296        {
297            let mut guard = self.cancellation_token.lock();
298            guard.cancel();
299            *guard = CancellationToken::new();
300        }
301        *self.session_token.write().await = None;
302    }
303
304    /// Sends a keep-alive request to renew the session.
305    ///
306    /// # Errors
307    ///
308    /// Returns an error if the keep-alive request fails.
309    pub async fn keep_alive(&self) -> Result<(), BetfairHttpError> {
310        self.keep_alive_with_token().await.map(drop)
311    }
312
313    /// Renews the session and returns the current token.
314    ///
315    /// # Errors
316    ///
317    /// Returns an error if the keep-alive request fails.
318    pub(crate) async fn keep_alive_with_token(&self) -> Result<String, BetfairHttpError> {
319        let _guard = self.connect_lock.lock().await;
320        let resp_bytes = self.send_identity(&self.url_keep_alive, Vec::new()).await?;
321
322        let resp: LoginResponse = serde_json::from_slice(&resp_bytes)?;
323
324        if resp.status == LoginStatus::Success {
325            *self.session_token.write().await = Some(resp.token.clone());
326            Ok(resp.token)
327        } else {
328            Err(BetfairHttpError::LoginFailed {
329                status: resp.error.unwrap_or_else(|| format!("{:?}", resp.status)),
330            })
331        }
332    }
333
334    /// Sends a JSON-RPC request to the Betting API.
335    ///
336    /// # Errors
337    ///
338    /// Returns an error if the request fails, authentication is missing,
339    /// or the response contains a JSON-RPC error.
340    pub async fn send_betting<T, P>(&self, method: &str, params: P) -> Result<T, BetfairHttpError>
341    where
342        T: DeserializeOwned,
343        P: Serialize,
344    {
345        self.send_jsonrpc(&self.url_betting, method, params, false)
346            .await
347    }
348
349    /// Sends a JSON-RPC request to the Betting API with order rate limiting.
350    /// Ambiguous failures are retried only when the params include a request-level `customerRef`.
351    ///
352    /// # Errors
353    ///
354    /// Returns an error if the request fails, authentication is missing,
355    /// or the response contains a JSON-RPC error.
356    pub async fn send_betting_order<T, P>(
357        &self,
358        method: &str,
359        params: P,
360    ) -> Result<T, BetfairHttpError>
361    where
362        T: DeserializeOwned,
363        P: Serialize,
364    {
365        self.send_jsonrpc(&self.url_betting, method, params, true)
366            .await
367    }
368
369    /// Sends a JSON-RPC request to the Accounts API.
370    ///
371    /// # Errors
372    ///
373    /// Returns an error if the request fails, authentication is missing,
374    /// or the response contains a JSON-RPC error.
375    pub async fn send_accounts<T, P>(&self, method: &str, params: P) -> Result<T, BetfairHttpError>
376    where
377        T: DeserializeOwned,
378        P: Serialize,
379    {
380        self.send_jsonrpc(&self.url_accounts, method, params, false)
381            .await
382    }
383
384    /// Sends a GET request to the Navigation API.
385    ///
386    /// # Errors
387    ///
388    /// Returns an error if the request fails or the response cannot be parsed.
389    pub async fn send_navigation<T>(&self) -> Result<T, BetfairHttpError>
390    where
391        T: DeserializeOwned,
392    {
393        let headers = self.build_headers("application/json").await?;
394
395        let resp = self
396            .client
397            .request(
398                Method::GET,
399                self.url_navigation.clone(),
400                None,
401                Some(headers),
402                None,
403                None,
404                Some(vec![BETFAIR_RATE_LIMIT_DEFAULT.to_string()]),
405            )
406            .await
407            .map_err(|e| BetfairHttpError::NetworkError(e.to_string()))?;
408
409        if resp.status.as_u16() != 200 {
410            let body = String::from_utf8_lossy(&resp.body);
411            return Err(BetfairHttpError::UnexpectedStatus {
412                status: resp.status.as_u16(),
413                body: body.to_string(),
414            });
415        }
416
417        serde_json::from_slice(&resp.body).map_err(BetfairHttpError::from)
418    }
419
420    fn make_quota(requests_per_second: u32, label: &str) -> Result<Quota, BetfairHttpError> {
421        let rate = NonZeroU32::new(requests_per_second).ok_or_else(|| {
422            BetfairHttpError::InvalidConfiguration(format!("{label} must be greater than zero"))
423        })?;
424
425        Quota::per_second(rate).ok_or_else(|| {
426            BetfairHttpError::InvalidConfiguration(format!("Invalid {label} quota configuration"))
427        })
428    }
429
430    fn rate_limiter_quotas(
431        request_rate_per_second: u32,
432        order_request_rate_per_second: u32,
433    ) -> Result<Vec<(String, Quota)>, BetfairHttpError> {
434        Ok(vec![
435            (
436                BETFAIR_RATE_LIMIT_DEFAULT.to_string(),
437                Self::make_quota(request_rate_per_second, "request_rate_per_second")?,
438            ),
439            (
440                BETFAIR_RATE_LIMIT_ORDERS.to_string(),
441                Self::make_quota(
442                    order_request_rate_per_second,
443                    "order_request_rate_per_second",
444                )?,
445            ),
446        ])
447    }
448
449    fn default_quota(request_rate_per_second: u32) -> Result<Option<Quota>, BetfairHttpError> {
450        Ok(Some(Self::make_quota(
451            request_rate_per_second,
452            "request_rate_per_second",
453        )?))
454    }
455
456    async fn build_headers(
457        &self,
458        content_type: &str,
459    ) -> Result<HashMap<String, String>, BetfairHttpError> {
460        let token = self
461            .session_token
462            .read()
463            .await
464            .clone()
465            .ok_or(BetfairHttpError::MissingCredentials)?;
466
467        let mut headers = HashMap::new();
468        headers.insert(HEADER_X_AUTHENTICATION.to_string(), token);
469        headers.insert(
470            HEADER_X_APPLICATION.to_string(),
471            self.credential.app_key().to_string(),
472        );
473        headers.insert("Accept".to_string(), "application/json".to_string());
474        headers.insert("Content-Type".to_string(), content_type.to_string());
475        Ok(headers)
476    }
477
478    async fn send_identity(&self, url: &str, body: Vec<u8>) -> Result<Vec<u8>, BetfairHttpError> {
479        let mut headers = HashMap::new();
480        headers.insert("Accept".to_string(), "application/json".to_string());
481        headers.insert(
482            "Content-Type".to_string(),
483            "application/x-www-form-urlencoded".to_string(),
484        );
485        headers.insert(
486            HEADER_X_APPLICATION.to_string(),
487            self.credential.app_key().to_string(),
488        );
489
490        // Add session token if we have one (for keep-alive)
491        if let Some(token) = self.session_token.read().await.as_ref() {
492            headers.insert(HEADER_X_AUTHENTICATION.to_string(), token.clone());
493        }
494
495        let resp = self
496            .client
497            .request(
498                Method::POST,
499                url.to_string(),
500                None,
501                Some(headers),
502                Some(body),
503                None,
504                Some(vec![BETFAIR_RATE_LIMIT_DEFAULT.to_string()]),
505            )
506            .await
507            .map_err(|e| BetfairHttpError::NetworkError(e.to_string()))?;
508
509        if resp.status.as_u16() != 200 {
510            let body = String::from_utf8_lossy(&resp.body);
511            return Err(BetfairHttpError::UnexpectedStatus {
512                status: resp.status.as_u16(),
513                body: body.to_string(),
514            });
515        }
516
517        Ok(resp.body.to_vec())
518    }
519
520    async fn send_jsonrpc<T, P>(
521        &self,
522        base_url: &str,
523        method: &str,
524        params: P,
525        is_order: bool,
526    ) -> Result<T, BetfairHttpError>
527    where
528        T: DeserializeOwned,
529        P: Serialize,
530    {
531        let operation_id = format!("{base_url}#{method}");
532        let params_value = serde_json::to_value(&params)?;
533        let has_order_customer_ref = params_value
534            .get("customerRef")
535            .and_then(serde_json::Value::as_str)
536            .is_some_and(|customer_ref| !customer_ref.is_empty());
537        let had_ambiguous_attempt = AtomicBool::new(false);
538
539        let operation = || {
540            let params_value = params_value.clone();
541            let had_ambiguous_attempt = &had_ambiguous_attempt;
542
543            async move {
544                let result = self
545                    .send_jsonrpc_once(base_url, method, params_value, is_order)
546                    .await;
547
548                if is_order && result.as_ref().is_err_and(|e| e.is_order_ambiguous()) {
549                    had_ambiguous_attempt.store(true, Ordering::Relaxed);
550                }
551
552                result
553            }
554        };
555
556        let should_retry = |error: &BetfairHttpError| -> bool {
557            if is_order {
558                error.is_order_retryable()
559                    && (has_order_customer_ref || !error.is_order_ambiguous())
560            } else {
561                error.is_retryable()
562            }
563        };
564
565        let create_error = |error: RetryError| -> BetfairHttpError {
566            map_retry_error(error, is_order, &had_ambiguous_attempt)
567        };
568
569        // Snapshot the current token; `disconnect()` may swap it for a fresh
570        // one mid-flight, but the in-flight retry loop should observe the
571        // pre-disconnect token so a cancel actually unblocks it.
572        let token = self.cancellation_token.lock().clone();
573
574        let retry_manager = if is_order {
575            &self.order_retry_manager
576        } else {
577            &self.retry_manager
578        };
579        let result = retry_manager
580            .execute_with_retry_with_cancel(
581                &operation_id,
582                operation,
583                should_retry,
584                create_error,
585                &token,
586            )
587            .await;
588
589        let result = match result {
590            Err(e)
591                if is_order
592                    && had_ambiguous_attempt.load(Ordering::Relaxed)
593                    && !e.is_order_ambiguous() =>
594            {
595                Err(BetfairHttpError::OrderRequestAmbiguous(format!(
596                    "an earlier attempt had an unknown outcome; final error: {e}",
597                )))
598            }
599            result => result,
600        };
601
602        if let Err(ref e) = result
603            && should_retry(e)
604        {
605            log::error!("Request exhausted retries: method={method}, error={e}");
606        }
607
608        result
609    }
610
611    async fn send_jsonrpc_once<T>(
612        &self,
613        base_url: &str,
614        method: &str,
615        params: serde_json::Value,
616        is_order: bool,
617    ) -> Result<T, BetfairHttpError>
618    where
619        T: DeserializeOwned,
620    {
621        let id = self.request_id.fetch_add(1, Ordering::SeqCst);
622        let request = JsonRpcRequest {
623            jsonrpc: "2.0",
624            method: method.to_string(),
625            params,
626            id,
627        };
628
629        let body = serde_json::to_vec(&request)?;
630        let headers = self.build_headers("application/json").await?;
631        let rate_keys = if is_order {
632            vec![BETFAIR_RATE_LIMIT_ORDERS.to_string()]
633        } else {
634            vec![BETFAIR_RATE_LIMIT_DEFAULT.to_string()]
635        };
636
637        let resp = self
638            .client
639            .request(
640                Method::POST,
641                base_url.to_string(),
642                None,
643                Some(headers),
644                Some(body),
645                None,
646                Some(rate_keys),
647            )
648            .await
649            .map_err(|e| BetfairHttpError::NetworkError(e.to_string()))?;
650
651        if !resp.status.is_success() {
652            let error_body = String::from_utf8_lossy(&resp.body);
653            let preview: String = error_body.chars().take(500).collect();
654            log::warn!(
655                "HTTP error response: method={method}, status={}, body={}",
656                resp.status.as_u16(),
657                preview,
658            );
659            return Err(BetfairHttpError::UnexpectedStatus {
660                status: resp.status.as_u16(),
661                body: error_body.to_string(),
662            });
663        }
664
665        let json_value: serde_json::Value = serde_json::from_slice(&resp.body).map_err(|e| {
666            let preview: String = String::from_utf8_lossy(&resp.body)
667                .chars()
668                .take(500)
669                .collect();
670            log::warn!(
671                "Non-JSON response: method={method}, status={}, body={preview}",
672                resp.status.as_u16(),
673            );
674            BetfairHttpError::ResponseError(e.to_string())
675        })?;
676
677        let rpc_resp: JsonRpcResponse<T> = serde_json::from_value(json_value).map_err(|e| {
678            log::warn!("Failed to deserialize JSON-RPC response: method={method}, error={e}",);
679            BetfairHttpError::ResponseError(e.to_string())
680        })?;
681
682        if let Some(result) = rpc_resp.result {
683            Ok(result)
684        } else if let Some(error) = rpc_resp.error {
685            let api_exception = error.api_exception();
686            Err(BetfairHttpError::BetfairError {
687                code: error.code,
688                message: error.message,
689                api_error_code: api_exception
690                    .as_ref()
691                    .map(|exception| exception.error_code.clone()),
692                api_error_details: api_exception.and_then(|exception| exception.error_details),
693            })
694        } else {
695            Err(BetfairHttpError::ResponseError(
696                "Response contains neither result nor error".to_string(),
697            ))
698        }
699    }
700}
701
702fn map_retry_error(
703    error: RetryError,
704    is_order: bool,
705    had_ambiguous_attempt: &AtomicBool,
706) -> BetfairHttpError {
707    if is_order && matches!(&error, RetryError::OperationTimeout { .. }) {
708        had_ambiguous_attempt.store(true, Ordering::Relaxed);
709    }
710
711    match error {
712        RetryError::Canceled => {
713            BetfairHttpError::Canceled("Adapter disconnecting or shutting down".to_string())
714        }
715        error => BetfairHttpError::NetworkError(error.to_string()),
716    }
717}
718
719#[cfg(test)]
720mod tests {
721    use std::time::Duration;
722
723    use parking_lot::Mutex;
724    use proptest::prelude::*;
725    use rstest::rstest;
726
727    use super::*;
728    use crate::common::consts::{
729        BETFAIR_RATE_LIMIT_DEFAULT, BETFAIR_RATE_LIMIT_ORDERS, METHOD_LIST_MARKET_CATALOGUE,
730    };
731
732    fn json_value_strategy() -> impl Strategy<Value = serde_json::Value> {
733        let leaf = prop_oneof![
734            Just(serde_json::Value::Null),
735            any::<bool>().prop_map(serde_json::Value::Bool),
736            any::<i64>().prop_map(|value| serde_json::Value::Number(value.into())),
737            "[ -~]{0,32}".prop_map(serde_json::Value::String),
738        ];
739
740        leaf.prop_recursive(3, 64, 8, |inner| {
741            prop_oneof![
742                prop::collection::vec(inner.clone(), 0..8).prop_map(serde_json::Value::Array),
743                prop::collection::vec(("[A-Za-z0-9_]{0,16}", inner), 0..8)
744                    .prop_map(|entries| serde_json::Value::Object(entries.into_iter().collect())),
745            ]
746        })
747    }
748
749    #[rstest]
750    fn test_rate_limiter_quotas_has_expected_keys() {
751        let quotas = BetfairHttpClient::rate_limiter_quotas(5, 20).unwrap();
752        let keys: Vec<&str> = quotas.iter().map(|(k, _)| k.as_str()).collect();
753        assert!(keys.contains(&BETFAIR_RATE_LIMIT_DEFAULT));
754        assert!(keys.contains(&BETFAIR_RATE_LIMIT_ORDERS));
755    }
756
757    #[rstest]
758    fn test_default_quota_is_some() {
759        assert!(BetfairHttpClient::default_quota(5).unwrap().is_some());
760    }
761
762    #[rstest]
763    fn test_rate_limiter_quotas_reject_zero_rate_limit() {
764        let result = BetfairHttpClient::rate_limiter_quotas(0, 20);
765
766        assert!(result.is_err());
767        assert!(
768            result
769                .err()
770                .unwrap()
771                .to_string()
772                .contains("request_rate_per_second")
773        );
774    }
775
776    #[rstest]
777    fn test_json_rpc_request_serialization() {
778        let request = JsonRpcRequest {
779            jsonrpc: "2.0",
780            method: METHOD_LIST_MARKET_CATALOGUE.to_string(),
781            params: serde_json::json!({"filter": {}, "maxResults": 100}),
782            id: 1,
783        };
784
785        let json = serde_json::to_value(&request).unwrap();
786        assert_eq!(json["jsonrpc"], "2.0");
787        assert_eq!(json["method"], "SportsAPING/v1.0/listMarketCatalogue");
788        assert_eq!(json["params"]["maxResults"], 100);
789        assert_eq!(json["id"], 1);
790    }
791
792    #[rstest]
793    fn test_json_rpc_response_success() {
794        let json = include_str!(concat!(
795            env!("CARGO_MANIFEST_DIR"),
796            "/test_data/rest/betting_place_order_success.json"
797        ));
798        let resp: JsonRpcResponse<serde_json::Value> = serde_json::from_str(json).unwrap();
799        assert_eq!(
800            resp.result
801                .as_ref()
802                .and_then(|result| result["status"].as_str()),
803            Some("SUCCESS")
804        );
805        assert!(resp.error.is_none());
806    }
807
808    #[rstest]
809    fn test_json_rpc_response_error() {
810        let json = include_str!(concat!(
811            env!("CARGO_MANIFEST_DIR"),
812            "/test_data/rest/betting_jsonrpc_error_too_much_data_live.json"
813        ));
814        let resp: JsonRpcResponse<serde_json::Value> = serde_json::from_str(json).unwrap();
815        assert!(resp.result.is_none());
816        let error = resp.error.unwrap();
817        assert_eq!(error.code, -32099);
818        assert_eq!(error.message, "ANGX-0001");
819        let api_exception = error.api_exception().unwrap();
820        assert_eq!(api_exception.error_code, "TOO_MUCH_DATA");
821        assert_eq!(
822            api_exception.error_details.as_deref(),
823            Some("MaxResults must be less than or equal to 1000")
824        );
825    }
826
827    #[rstest]
828    fn test_order_operation_timeout_records_ambiguous_attempt() {
829        let had_ambiguous_attempt = AtomicBool::new(false);
830
831        let error = map_retry_error(
832            RetryError::OperationTimeout { timeout_ms: 30_000 },
833            true,
834            &had_ambiguous_attempt,
835        );
836
837        assert!(error.is_order_ambiguous());
838        assert!(had_ambiguous_attempt.load(Ordering::Relaxed));
839    }
840
841    proptest! {
842        #![proptest_config(ProptestConfig::with_cases(256))]
843
844        #[rstest]
845        fn arbitrary_json_rpc_error_data_preserves_outer_error(
846            data in json_value_strategy(),
847        ) {
848            let response: JsonRpcResponse<serde_json::Value> = serde_json::from_value(
849                serde_json::json!({
850                    "jsonrpc": "2.0",
851                    "id": 73,
852                    "error": {
853                        "code": -32099,
854                        "message": "ANGX-PROPERTY",
855                        "data": data,
856                    },
857                }),
858            )
859            .unwrap();
860            let error = response.error.unwrap();
861            let api_exception = error.api_exception();
862            let surfaced = BetfairHttpError::BetfairError {
863                code: error.code,
864                message: error.message.clone(),
865                api_error_code: api_exception
866                    .as_ref()
867                    .map(|exception| exception.error_code.clone()),
868                api_error_details: api_exception
869                    .and_then(|exception| exception.error_details),
870            };
871
872            prop_assert_eq!(error.code, -32099);
873            prop_assert_eq!(error.message, "ANGX-PROPERTY");
874
875            if matches!(
876                &surfaced,
877                BetfairHttpError::BetfairError {
878                    api_error_code: None,
879                    ..
880                }
881            ) {
882                prop_assert!(surfaced.is_order_ambiguous());
883                prop_assert!(!surfaced.is_order_retryable());
884            }
885        }
886
887        #[rstest]
888        fn unknown_api_error_code_is_ambiguous_and_not_retried(
889            api_error_code in "FUTURE_[A-Z0-9_]{1,24}",
890        ) {
891            let exception_name = "FutureAPINGException";
892            let response: JsonRpcResponse<serde_json::Value> = serde_json::from_value(
893                serde_json::json!({
894                    "jsonrpc": "2.0",
895                    "id": 74,
896                    "error": {
897                        "code": -32099,
898                        "message": "ANGX-FUTURE",
899                        "data": {
900                            "exceptionname": exception_name,
901                            (exception_name): {
902                                "errorCode": api_error_code,
903                                "errorDetails": "future wire shape",
904                            },
905                        },
906                    },
907                }),
908            )
909            .unwrap();
910            let error = response.error.unwrap();
911            let api_exception = error.api_exception().unwrap();
912            let surfaced = BetfairHttpError::BetfairError {
913                code: error.code,
914                message: error.message.clone(),
915                api_error_code: Some(api_exception.error_code.clone()),
916                api_error_details: api_exception.error_details.clone(),
917            };
918
919            prop_assert_eq!(error.code, -32099);
920            prop_assert_eq!(error.message, "ANGX-FUTURE");
921            prop_assert_eq!(api_exception.error_code, api_error_code);
922            prop_assert_eq!(
923                api_exception.error_details.as_deref(),
924                Some("future wire shape"),
925            );
926            prop_assert!(surfaced.is_order_ambiguous());
927            prop_assert!(!surfaced.is_order_retryable());
928        }
929    }
930
931    #[tokio::test(start_paused = true)]
932    async fn test_order_retry_budget_prevents_dispatch_at_45_seconds() {
933        let client = BetfairHttpClient::new(
934            BetfairCredential::new(
935                "username".to_string(),
936                "password".to_string(),
937                "app-key".to_string(),
938            ),
939            None,
940            Some(100),
941            Some(10_000),
942            None,
943            Some(5),
944            Some(20),
945        )
946        .unwrap();
947        let started_at = tokio::time::Instant::now();
948        let attempt_times = Arc::new(Mutex::new(Vec::new()));
949        let attempt_times_for_operation = Arc::clone(&attempt_times);
950
951        let result = client
952            .order_retry_manager
953            .execute_with_retry(
954                "placeOrders",
955                move || {
956                    let attempt_times = Arc::clone(&attempt_times_for_operation);
957                    async move {
958                        attempt_times.lock().push(started_at.elapsed());
959                        Err::<(), _>(BetfairHttpError::UnexpectedStatus {
960                            status: 502,
961                            body: "Bad Gateway".to_string(),
962                        })
963                    }
964                },
965                BetfairHttpError::is_order_retryable,
966                |e| BetfairHttpError::NetworkError(e.to_string()),
967            )
968            .await;
969
970        assert!(matches!(result, Err(BetfairHttpError::NetworkError(_))));
971        assert_eq!(started_at.elapsed(), Duration::from_secs(45));
972        let attempt_times = attempt_times.lock();
973        assert!(attempt_times.len() > 1);
974        assert_eq!(attempt_times[0], Duration::ZERO);
975        assert!(
976            attempt_times
977                .iter()
978                .all(|elapsed| *elapsed < Duration::from_secs(45)),
979        );
980    }
981}
nautilus_betfair/http/client.rs

nautilus_betfair/http/
client.rs
