1use std::sync::Arc;
2use std::time::Duration;
3
4use http::Method;
5use polyoxide_rt::Runtime;
6use tokio::sync::{OwnedSemaphorePermit, Semaphore};
7use url::Url;
8
9use http::header::RETRY_AFTER;
10
11use crate::error::ApiError;
12use crate::hooks::{
13 DefaultRetryPolicy, DynRetryPolicy, DynThrottle, NoThrottle, RequestParts, RetryPolicy,
14 Throttle,
15};
16use crate::rate_limit::{RateLimiter, RetryConfig};
17use crate::transport::{default_transport, HttpTransport};
18
19pub fn retry_after_header(headers: &http::HeaderMap) -> Option<String> {
21 headers.get(RETRY_AFTER)?.to_str().ok().map(String::from)
22}
23
24pub const DEFAULT_TIMEOUT_MS: u64 = 30_000;
26pub const DEFAULT_POOL_SIZE: usize = 10;
28
29#[derive(Clone)]
35pub struct HttpClient {
36 pub(crate) transport: Arc<dyn HttpTransport>,
39 pub(crate) timeout: Duration,
41 pub base_url: Url,
43 pub(crate) throttle: Arc<DynThrottle<'static>>,
44 pub(crate) policy: Arc<DynRetryPolicy<'static>>,
45 pub(crate) retry_config: RetryConfig,
46 pub(crate) runtime: Arc<dyn Runtime>,
47 concurrency_limiter: Option<Arc<Semaphore>>,
48}
49
50impl std::fmt::Debug for HttpClient {
53 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
54 f.debug_struct("HttpClient")
55 .field("transport", &self.transport)
56 .field("base_url", &self.base_url)
57 .field("retry_config", &self.retry_config)
58 .field("runtime", &self.runtime)
59 .field("concurrency_limiter", &self.concurrency_limiter)
60 .finish_non_exhaustive()
61 }
62}
63
64impl HttpClient {
65 pub fn with_base_url(&self, base_url: &str) -> Result<Self, ApiError> {
93 Ok(Self {
94 base_url: Url::parse(base_url)?,
95 ..self.clone()
96 })
97 }
98
99 pub fn runtime(&self) -> &Arc<dyn Runtime> {
103 &self.runtime
104 }
105
106 pub async fn acquire_concurrency(&self) -> Option<OwnedSemaphorePermit> {
118 let sem = self.concurrency_limiter.as_ref()?;
119 Some(
120 sem.clone()
121 .acquire_owned()
122 .await
123 .expect("concurrency semaphore is never closed"),
124 )
125 }
126
127 pub async fn get_bytes(
140 &self,
141 path: &str,
142 query: &[(String, String)],
143 ) -> Result<Vec<u8>, ApiError> {
144 let mut parts = RequestParts::new(Method::GET, path);
145 parts.query = query.to_vec();
146 let response = self.send(parts, &[], None).await?;
147
148 if !response.status().is_success() {
150 return Err(ApiError::from_response(response).await);
151 }
152
153 let bytes = response.bytes().await?;
154 Ok(bytes.to_vec())
155 }
156}
157
158pub struct HttpClientBuilder {
175 base_url: String,
176 timeout_ms: u64,
177 pool_size: usize,
178 throttle: Option<Arc<DynThrottle<'static>>>,
179 policy: Option<Arc<DynRetryPolicy<'static>>>,
180 retry_config: RetryConfig,
181 max_concurrent: Option<usize>,
182 runtime: Option<Arc<dyn Runtime>>,
183 gzip: Option<bool>,
184 http_transport: Option<Arc<dyn HttpTransport>>,
185}
186
187impl HttpClientBuilder {
188 pub fn new(base_url: impl Into<String>) -> Self {
190 Self {
191 base_url: base_url.into(),
192 timeout_ms: DEFAULT_TIMEOUT_MS,
193 pool_size: DEFAULT_POOL_SIZE,
194 throttle: None,
195 policy: None,
196 retry_config: RetryConfig::default(),
197 max_concurrent: None,
198 runtime: None,
199 gzip: None,
200 http_transport: None,
201 }
202 }
203
204 pub fn timeout_ms(mut self, timeout: u64) -> Self {
208 self.timeout_ms = timeout;
209 self
210 }
211
212 pub fn pool_size(mut self, size: usize) -> Self {
216 self.pool_size = size;
217 self
218 }
219
220 pub fn with_rate_limiter(self, limiter: RateLimiter) -> Self {
223 self.with_throttle(limiter)
224 }
225
226 pub fn with_throttle(mut self, throttle: impl Throttle + 'static) -> Self {
232 self.throttle = Some(DynThrottle::new_arc(throttle));
233 self
234 }
235
236 pub fn with_retry_policy(mut self, policy: impl RetryPolicy + 'static) -> Self {
241 self.policy = Some(DynRetryPolicy::new_arc(policy));
242 self
243 }
244
245 pub fn with_retry_config(mut self, config: RetryConfig) -> Self {
247 self.retry_config = config;
248 self
249 }
250
251 pub fn with_max_concurrent(mut self, max: usize) -> Self {
256 self.max_concurrent = Some(max);
257 self
258 }
259
260 pub fn runtime(mut self, runtime: Arc<dyn Runtime>) -> Self {
268 self.runtime = Some(runtime);
269 self
270 }
271
272 pub fn gzip(mut self, enabled: bool) -> Self {
280 self.gzip = Some(enabled);
281 self
282 }
283
284 pub fn http_transport(mut self, transport: Arc<dyn HttpTransport>) -> Self {
292 self.http_transport = Some(transport);
293 self
294 }
295
296 pub fn build(self) -> Result<HttpClient, ApiError> {
298 let timeout = Duration::from_millis(self.timeout_ms);
299 let transport = match self.http_transport {
300 Some(transport) => transport,
301 None => default_transport(timeout, self.pool_size, self.gzip)?,
302 };
303
304 let base_url = Url::parse(&self.base_url)?;
305
306 Ok(HttpClient {
307 transport,
308 timeout,
309 base_url,
310 throttle: self
311 .throttle
312 .unwrap_or_else(|| DynThrottle::new_arc(NoThrottle)),
313 policy: self
314 .policy
315 .unwrap_or_else(|| DynRetryPolicy::new_arc(DefaultRetryPolicy)),
316 retry_config: self.retry_config,
317 runtime: self.runtime.unwrap_or_else(polyoxide_rt::default_runtime),
318 concurrency_limiter: self.max_concurrent.map(|n| Arc::new(Semaphore::new(n))),
319 })
320 }
321}
322
323impl Default for HttpClientBuilder {
324 fn default() -> Self {
325 Self::new(String::new())
326 }
327}
328
329#[cfg(test)]
330mod tests {
331 use super::*;
332 use http::StatusCode;
333
334 #[test]
335 fn test_should_retry_uses_retry_after_header() {
336 let config = RetryConfig::default();
337 let d = config.retry_delay(0, Some("2"));
338 assert_eq!(d, Duration::from_millis(2000));
339 }
340
341 #[test]
342 fn test_should_retry_retry_after_fractional_seconds() {
343 let config = RetryConfig::default();
344 let d = config.retry_delay(0, Some("1.5"));
349 assert_eq!(d, Duration::from_millis(1500));
350 }
351
352 #[test]
353 fn retry_after_below_our_own_backoff_does_not_shorten_the_wait() {
354 let config = RetryConfig::default();
355 for header in ["0", "0.0", "-1", "-30", "0.0001"] {
362 let d = config.retry_delay(0, Some(header));
363 assert!(
364 d >= Duration::from_millis(375),
365 "Retry-After: {header:?} produced a {d:?} sleep; the floor is the \
366 client's own attempt-0 backoff, >=375ms after jitter"
367 );
368 }
369 }
370
371 #[test]
372 fn retry_after_zero_still_backs_off_exponentially_across_attempts() {
373 let config = RetryConfig::default();
374 for (attempt, min_ms) in [(0u32, 375u64), (1, 750), (2, 1_500)] {
378 let d = config.retry_delay(attempt, Some("0"));
379 assert!(
380 d >= Duration::from_millis(min_ms),
381 "attempt {attempt} with Retry-After: 0 slept {d:?}, expected >={min_ms}ms"
382 );
383 }
384 }
385
386 #[test]
387 fn test_should_retry_retry_after_clamped_to_max_backoff() {
388 let config = RetryConfig::default();
389 let d = config.retry_delay(0, Some("60"));
391 assert_eq!(d, Duration::from_millis(10_000));
392 }
393
394 #[test]
395 fn test_should_retry_retry_after_invalid_falls_back() {
396 let config = RetryConfig::default();
397 let d = config.retry_delay(0, Some("Wed, 21 Oct 2025 07:28:00 GMT"));
399 let ms = d.as_millis() as u64;
401 assert!(
402 (375..=625).contains(&ms),
403 "expected fallback backoff in [375, 625], got {ms}"
404 );
405 }
406
407 #[derive(Default)]
411 struct SignedAt(std::sync::Mutex<Option<std::time::Instant>>);
412
413 impl crate::hooks::Authenticator for SignedAt {
414 async fn sign(&self, _parts: &mut RequestParts, _attempt: u32) -> Result<(), ApiError> {
415 *self.0.lock().unwrap() = Some(std::time::Instant::now());
416 Ok(())
417 }
418 }
419
420 async fn time_to_sign(builder: HttpClientBuilder, server: &mockito::ServerGuard) -> Duration {
423 let client = builder.build().unwrap();
424 let client = client.with_base_url(&server.url()).unwrap();
425 let signed = SignedAt::default();
426 let start = std::time::Instant::now();
427 client
428 .send(
429 RequestParts::new(Method::POST, "/order"),
430 &[],
431 Some(crate::hooks::DynAuthenticator::from_ref(&signed)),
432 )
433 .await
434 .unwrap();
435 let at = signed.0.lock().unwrap().expect("the request was signed");
436 at - start
437 }
438
439 #[tokio::test]
440 async fn test_builder_without_rate_limiter() {
441 let mut server = mockito::Server::new_async().await;
442 let _order = server.mock("POST", "/order").create_async().await;
443 let builder = HttpClientBuilder::new("https://example.com");
444 let elapsed = time_to_sign(builder, &server).await;
445 assert!(elapsed < Duration::from_millis(10));
446 }
447
448 #[tokio::test]
451 async fn test_acquire_concurrency_none_when_not_configured() {
452 let client = HttpClientBuilder::new("https://example.com")
453 .build()
454 .unwrap();
455 assert!(client.acquire_concurrency().await.is_none());
456 }
457
458 #[tokio::test]
459 async fn test_acquire_concurrency_returns_permit() {
460 let client = HttpClientBuilder::new("https://example.com")
461 .with_max_concurrent(2)
462 .build()
463 .unwrap();
464 let permit = client.acquire_concurrency().await;
465 assert!(permit.is_some());
466 }
467
468 #[tokio::test]
469 async fn test_concurrency_shared_across_clones() {
470 let client = HttpClientBuilder::new("https://example.com")
471 .with_max_concurrent(1)
472 .build()
473 .unwrap();
474 let clone = client.clone();
475
476 let _permit = client.acquire_concurrency().await.unwrap();
478
479 let result =
481 tokio::time::timeout(Duration::from_millis(50), clone.acquire_concurrency()).await;
482 assert!(result.is_err(), "clone should block when permit is held");
483 }
484
485 #[tokio::test]
486 async fn test_concurrency_limits_parallel_tasks() {
487 let client = HttpClientBuilder::new("https://example.com")
488 .with_max_concurrent(2)
489 .build()
490 .unwrap();
491
492 let start = std::time::Instant::now();
493 let mut handles = Vec::new();
494 for _ in 0..4 {
495 let c = client.clone();
496 handles.push(tokio::spawn(async move {
497 let _permit = c.acquire_concurrency().await;
498 tokio::time::sleep(Duration::from_millis(50)).await;
499 }));
500 }
501 for h in handles {
502 h.await.unwrap();
503 }
504 assert!(
506 start.elapsed() >= Duration::from_millis(90),
507 "expected ~100ms, got {:?}",
508 start.elapsed()
509 );
510 }
511
512 #[tokio::test]
513 async fn test_builder_with_max_concurrent() {
514 let client = HttpClientBuilder::new("https://example.com")
515 .with_max_concurrent(5)
516 .build()
517 .unwrap();
518 let mut permits = Vec::new();
520 for _ in 0..5 {
521 permits.push(client.acquire_concurrency().await);
522 }
523 assert!(permits.iter().all(|p| p.is_some()));
524
525 let result =
527 tokio::time::timeout(Duration::from_millis(50), client.acquire_concurrency()).await;
528 assert!(result.is_err());
529 }
530
531 #[tokio::test]
534 async fn test_get_bytes_returns_body_verbatim() {
535 let mut server = mockito::Server::new_async().await;
536 let body: Vec<u8> = vec![0x50, 0x4B, 0x03, 0x04, 0x00, 0xFF, 0xFE, 0x42];
538 let mock = server
539 .mock("GET", "/v1/accounting/snapshot")
540 .match_query(mockito::Matcher::UrlEncoded("user".into(), "0xabc".into()))
541 .with_status(200)
542 .with_header("content-type", "application/zip")
543 .with_body(body.clone())
544 .create_async()
545 .await;
546
547 let client = HttpClientBuilder::new(server.url()).build().unwrap();
548 let out = client
549 .get_bytes(
550 "/v1/accounting/snapshot",
551 &[("user".to_string(), "0xabc".to_string())],
552 )
553 .await
554 .unwrap();
555 assert_eq!(out, body);
556 mock.assert_async().await;
557 }
558
559 #[tokio::test]
560 async fn test_get_bytes_maps_non_2xx_to_api_error() {
561 let mut server = mockito::Server::new_async().await;
562 let mock = server
563 .mock("GET", "/does-not-exist")
564 .with_status(404)
565 .with_header("content-type", "application/json")
566 .with_body(r#"{"error": "not found"}"#)
567 .create_async()
568 .await;
569
570 let client = HttpClientBuilder::new(server.url()).build().unwrap();
571 let err = client.get_bytes("/does-not-exist", &[]).await.unwrap_err();
572 match err {
573 ApiError::Response(response) => {
574 assert_eq!(response.status, StatusCode::NOT_FOUND);
575 assert_eq!(response.message, "not found");
576 }
577 other => panic!("expected ApiError::Response, got {other:?}"),
578 }
579 mock.assert_async().await;
580 }
581
582 #[tokio::test]
583 async fn test_get_bytes_no_query_params() {
584 let mut server = mockito::Server::new_async().await;
585 let mock = server
586 .mock("GET", "/raw")
587 .with_status(200)
588 .with_body(&b"hello"[..])
589 .create_async()
590 .await;
591
592 let client = HttpClientBuilder::new(server.url()).build().unwrap();
593 let out = client.get_bytes("/raw", &[]).await.unwrap();
594 assert_eq!(out, b"hello");
595 mock.assert_async().await;
596 }
597
598 #[tokio::test]
601 async fn gzip_follows_reqwest_unless_set() {
602 let mut server = mockito::Server::new_async().await;
606 let mock = server
607 .mock("GET", "/plain")
608 .match_header("accept-encoding", mockito::Matcher::Regex("gzip".into()))
609 .with_body("ok")
610 .create_async()
611 .await;
612
613 let client = HttpClientBuilder::new(server.url()).build().unwrap();
614 assert_eq!(client.get_bytes("/plain", &[]).await.unwrap(), b"ok");
615 mock.assert_async().await;
616 }
617
618 #[tokio::test]
619 async fn gzip_false_asks_for_no_compression() {
620 let mut server = mockito::Server::new_async().await;
621 let mock = server
622 .mock("GET", "/plain")
623 .match_header("accept-encoding", mockito::Matcher::Missing)
624 .with_body("ok")
625 .create_async()
626 .await;
627
628 let client = HttpClientBuilder::new(server.url())
629 .gzip(false)
630 .build()
631 .unwrap();
632 assert_eq!(client.get_bytes("/plain", &[]).await.unwrap(), b"ok");
633 mock.assert_async().await;
634 }
635
636 #[tokio::test]
637 async fn gzip_asks_for_and_decodes_a_compressed_body() {
638 const GZIPPED: &[u8] = &[
640 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xab, 0x56, 0x2a, 0x4e,
641 0x2d, 0x2a, 0x4b, 0x2d, 0x0a, 0xc9, 0xcc, 0x4d, 0x55, 0xb2, 0x32, 0xac, 0x05, 0x00,
642 0xe2, 0x1d, 0x3e, 0x1a, 0x10, 0x00, 0x00, 0x00,
643 ];
644 let mut server = mockito::Server::new_async().await;
645 let mock = server
646 .mock("GET", "/time")
647 .match_header("accept-encoding", mockito::Matcher::Regex("gzip".into()))
648 .with_header("content-encoding", "gzip")
649 .with_body(GZIPPED)
650 .create_async()
651 .await;
652
653 let client = HttpClientBuilder::new(server.url())
654 .gzip(true)
655 .build()
656 .unwrap();
657 let body = client.get_bytes("/time", &[]).await.unwrap();
658 assert_eq!(body, br#"{"serverTime":1}"#);
659 mock.assert_async().await;
660 }
661
662 #[test]
663 fn with_base_url_retargets_and_shares_transport() {
664 let client = HttpClientBuilder::new("https://api.example.com")
665 .with_max_concurrent(4)
666 .build()
667 .unwrap();
668 let sibling = client.with_base_url("https://sibling.example.com").unwrap();
669
670 assert_eq!(sibling.base_url.host_str(), Some("sibling.example.com"));
671 assert_eq!(client.base_url.host_str(), Some("api.example.com"));
673 assert!(sibling.concurrency_limiter.is_some());
676 }
677
678 #[test]
679 fn with_base_url_rejects_a_malformed_url() {
680 let client = HttpClientBuilder::new("https://api.example.com")
681 .build()
682 .unwrap();
683 assert!(client.with_base_url("not-a-url").is_err());
684 }
685
686 #[test]
687 fn base_url_path_prefixes_are_dropped() {
688 let client = HttpClientBuilder::new("http://localhost:8080/proxy")
692 .build()
693 .unwrap();
694 assert_eq!(
695 client.base_url.join("/user-pnl").unwrap().as_str(),
696 "http://localhost:8080/user-pnl",
697 "a path prefix in the base URL is not preserved"
698 );
699
700 let with_slash = client
701 .with_base_url("http://localhost:8080/proxy/")
702 .unwrap();
703 assert_eq!(
704 with_slash.base_url.join("/user-pnl").unwrap().as_str(),
705 "http://localhost:8080/user-pnl",
706 "a trailing slash does not preserve the prefix either"
707 );
708 }
709}