1use std::sync::Arc;
2use std::time::Duration;
3
4use reqwest::StatusCode;
5use tokio::sync::{OwnedSemaphorePermit, Semaphore};
6use url::Url;
7
8use reqwest::header::RETRY_AFTER;
9
10use crate::error::ApiError;
11use crate::rate_limit::{RateLimiter, RetryConfig};
12
13pub fn retry_after_header(response: &reqwest::Response) -> Option<String> {
15 response
16 .headers()
17 .get(RETRY_AFTER)?
18 .to_str()
19 .ok()
20 .map(String::from)
21}
22
23pub const DEFAULT_TIMEOUT_MS: u64 = 30_000;
25pub const DEFAULT_POOL_SIZE: usize = 10;
27
28#[derive(Debug, Clone)]
33pub struct HttpClient {
34 pub client: reqwest::Client,
36 pub base_url: Url,
38 rate_limiter: Option<RateLimiter>,
39 retry_config: RetryConfig,
40 concurrency_limiter: Option<Arc<Semaphore>>,
41}
42
43impl HttpClient {
44 pub fn with_base_url(&self, base_url: &str) -> Result<Self, ApiError> {
71 Ok(Self {
72 base_url: Url::parse(base_url)?,
73 ..self.clone()
74 })
75 }
76
77 pub async fn acquire_rate_limit(&self, path: &str, method: Option<&reqwest::Method>) {
79 if let Some(rl) = &self.rate_limiter {
80 rl.acquire(path, method).await;
81 }
82 }
83
84 pub async fn acquire_concurrency(&self) -> Option<OwnedSemaphorePermit> {
90 let sem = self.concurrency_limiter.as_ref()?;
91 Some(
92 sem.clone()
93 .acquire_owned()
94 .await
95 .expect("concurrency semaphore is never closed"),
96 )
97 }
98
99 pub fn should_retry(
127 &self,
128 status: StatusCode,
129 attempt: u32,
130 retry_after: Option<&str>,
131 ) -> Option<Duration> {
132 let retriable = status == StatusCode::TOO_MANY_REQUESTS || status == StatusCode::TOO_EARLY;
133 if !retriable || attempt >= self.retry_config.max_retries {
134 return None;
135 }
136 Some(self.retry_delay(attempt, retry_after))
137 }
138
139 fn retry_delay(&self, attempt: u32, retry_after: Option<&str>) -> Duration {
146 let computed = self.retry_config.backoff(attempt);
147 let requested = retry_after
148 .and_then(|v| v.parse::<f64>().ok())
149 .filter(|secs| secs.is_finite() && *secs > 0.0)
150 .map(|secs| {
151 let ms = (secs * 1000.0) as u64;
152 Duration::from_millis(ms.min(self.retry_config.max_backoff_ms))
153 });
154 requested.map_or(computed, |r| r.max(computed))
155 }
156
157 pub fn note_rate_limited(&self, status: StatusCode, retry_after: Option<&str>) {
172 if status != StatusCode::TOO_MANY_REQUESTS {
173 return;
174 }
175 if let Some(rl) = &self.rate_limiter {
176 rl.begin_cooldown(self.retry_delay(0, retry_after));
177 }
178 }
179
180 pub async fn get_bytes(
195 &self,
196 path: &str,
197 query: &[(String, String)],
198 ) -> Result<Vec<u8>, ApiError> {
199 let url = self.base_url.join(path)?;
200 let mut attempt = 0u32;
201
202 loop {
203 let _permit = self.acquire_concurrency().await;
204 self.acquire_rate_limit(path, None).await;
205
206 let mut request = self.client.get(url.clone());
207 if !query.is_empty() {
208 request = request.query(query);
209 }
210
211 let response = request.send().await?;
212 let status = response.status();
213 let retry_after = retry_after_header(&response);
214
215 self.note_rate_limited(status, retry_after.as_deref());
216
217 if let Some(backoff) = self.should_retry(status, attempt, retry_after.as_deref()) {
218 attempt += 1;
219 tracing::warn!(
220 "Retriable status {} on {}, retry {} after {}ms",
221 status,
222 path,
223 attempt,
224 backoff.as_millis()
225 );
226 drop(_permit);
227 tokio::time::sleep(backoff).await;
228 continue;
229 }
230
231 if !status.is_success() {
232 return Err(ApiError::from_response(response).await);
233 }
234
235 let bytes = response.bytes().await?;
236 return Ok(bytes.to_vec());
237 }
238 }
239}
240
241pub struct HttpClientBuilder {
258 base_url: String,
259 timeout_ms: u64,
260 pool_size: usize,
261 rate_limiter: Option<RateLimiter>,
262 retry_config: RetryConfig,
263 max_concurrent: Option<usize>,
264 gzip: bool,
265}
266
267impl HttpClientBuilder {
268 pub fn new(base_url: impl Into<String>) -> Self {
270 Self {
271 base_url: base_url.into(),
272 timeout_ms: DEFAULT_TIMEOUT_MS,
273 pool_size: DEFAULT_POOL_SIZE,
274 rate_limiter: None,
275 retry_config: RetryConfig::default(),
276 max_concurrent: None,
277 gzip: false,
278 }
279 }
280
281 pub fn timeout_ms(mut self, timeout: u64) -> Self {
285 self.timeout_ms = timeout;
286 self
287 }
288
289 pub fn pool_size(mut self, size: usize) -> Self {
293 self.pool_size = size;
294 self
295 }
296
297 pub fn with_rate_limiter(mut self, limiter: RateLimiter) -> Self {
299 self.rate_limiter = Some(limiter);
300 self
301 }
302
303 pub fn with_retry_config(mut self, config: RetryConfig) -> Self {
305 self.retry_config = config;
306 self
307 }
308
309 pub fn with_max_concurrent(mut self, max: usize) -> Self {
314 self.max_concurrent = Some(max);
315 self
316 }
317
318 pub fn gzip(mut self, enabled: bool) -> Self {
326 self.gzip = enabled;
327 self
328 }
329
330 pub fn build(self) -> Result<HttpClient, ApiError> {
332 let client = reqwest::Client::builder()
333 .gzip(self.gzip)
334 .timeout(Duration::from_millis(self.timeout_ms))
335 .connect_timeout(Duration::from_secs(10))
336 .redirect(reqwest::redirect::Policy::none())
337 .pool_max_idle_per_host(self.pool_size)
338 .build()?;
339
340 let base_url = Url::parse(&self.base_url)?;
341
342 Ok(HttpClient {
343 client,
344 base_url,
345 rate_limiter: self.rate_limiter,
346 retry_config: self.retry_config,
347 concurrency_limiter: self.max_concurrent.map(|n| Arc::new(Semaphore::new(n))),
348 })
349 }
350}
351
352impl Default for HttpClientBuilder {
353 fn default() -> Self {
354 Self {
355 base_url: String::new(),
356 timeout_ms: DEFAULT_TIMEOUT_MS,
357 pool_size: DEFAULT_POOL_SIZE,
358 rate_limiter: None,
359 retry_config: RetryConfig::default(),
360 max_concurrent: None,
361 gzip: false,
362 }
363 }
364}
365
366#[cfg(test)]
367mod tests {
368 use super::*;
369
370 #[test]
373 fn test_should_retry_429_under_max() {
374 let client = HttpClientBuilder::new("https://example.com")
375 .build()
376 .unwrap();
377 assert!(client
379 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, None)
380 .is_some());
381 assert!(client
382 .should_retry(StatusCode::TOO_MANY_REQUESTS, 2, None)
383 .is_some());
384 }
385
386 #[test]
387 fn test_should_retry_429_at_max() {
388 let client = HttpClientBuilder::new("https://example.com")
389 .build()
390 .unwrap();
391 assert!(client
393 .should_retry(StatusCode::TOO_MANY_REQUESTS, 3, None)
394 .is_none());
395 }
396
397 #[test]
398 fn test_should_retry_425_under_max() {
399 let client = HttpClientBuilder::new("https://example.com")
400 .build()
401 .unwrap();
402 assert!(client
404 .should_retry(StatusCode::TOO_EARLY, 0, None)
405 .is_some());
406 assert!(client
407 .should_retry(StatusCode::TOO_EARLY, 2, None)
408 .is_some());
409 }
410
411 #[test]
412 fn test_should_retry_425_at_max() {
413 let client = HttpClientBuilder::new("https://example.com")
414 .build()
415 .unwrap();
416 assert!(client
417 .should_retry(StatusCode::TOO_EARLY, 3, None)
418 .is_none());
419 }
420
421 #[test]
422 fn test_should_retry_ignores_other_statuses() {
423 let client = HttpClientBuilder::new("https://example.com")
424 .build()
425 .unwrap();
426 for status in [
427 StatusCode::OK,
428 StatusCode::INTERNAL_SERVER_ERROR,
429 StatusCode::BAD_GATEWAY,
430 StatusCode::BAD_REQUEST,
431 StatusCode::FORBIDDEN,
432 StatusCode::NOT_FOUND,
433 StatusCode::SERVICE_UNAVAILABLE,
437 ] {
438 assert!(
439 client.should_retry(status, 0, None).is_none(),
440 "expected None for {status}"
441 );
442 }
443 }
444
445 #[test]
446 fn test_should_retry_5xx_not_retried_despite_being_is_retriable() {
447 let client = HttpClientBuilder::new("https://example.com")
450 .build()
451 .unwrap();
452 assert!(client
453 .should_retry(StatusCode::INTERNAL_SERVER_ERROR, 0, None)
454 .is_none());
455 assert!(ApiError::Api {
456 status: 500,
457 message: String::new()
458 }
459 .is_retriable());
460 }
461
462 #[test]
463 fn test_should_retry_custom_config() {
464 let client = HttpClientBuilder::new("https://example.com")
465 .with_retry_config(RetryConfig {
466 max_retries: 1,
467 ..RetryConfig::default()
468 })
469 .build()
470 .unwrap();
471 assert!(client
472 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, None)
473 .is_some());
474 assert!(client
475 .should_retry(StatusCode::TOO_MANY_REQUESTS, 1, None)
476 .is_none());
477 }
478
479 #[test]
480 fn test_should_retry_uses_retry_after_header() {
481 let client = HttpClientBuilder::new("https://example.com")
482 .build()
483 .unwrap();
484 let d = client
485 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, Some("2"))
486 .unwrap();
487 assert_eq!(d, Duration::from_millis(2000));
488 }
489
490 #[test]
491 fn test_should_retry_retry_after_fractional_seconds() {
492 let client = HttpClientBuilder::new("https://example.com")
493 .build()
494 .unwrap();
495 let d = client
500 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, Some("1.5"))
501 .unwrap();
502 assert_eq!(d, Duration::from_millis(1500));
503 }
504
505 #[test]
506 fn retry_after_below_our_own_backoff_does_not_shorten_the_wait() {
507 let client = HttpClientBuilder::new("https://example.com")
508 .build()
509 .unwrap();
510 for header in ["0", "0.0", "-1", "-30", "0.0001"] {
517 let d = client
518 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, Some(header))
519 .unwrap();
520 assert!(
521 d >= Duration::from_millis(375),
522 "Retry-After: {header:?} produced a {d:?} sleep; the floor is the \
523 client's own attempt-0 backoff, >=375ms after jitter"
524 );
525 }
526 }
527
528 #[test]
529 fn retry_after_zero_still_backs_off_exponentially_across_attempts() {
530 let client = HttpClientBuilder::new("https://example.com")
531 .build()
532 .unwrap();
533 for (attempt, min_ms) in [(0u32, 375u64), (1, 750), (2, 1_500)] {
537 let d = client
538 .should_retry(StatusCode::TOO_MANY_REQUESTS, attempt, Some("0"))
539 .unwrap();
540 assert!(
541 d >= Duration::from_millis(min_ms),
542 "attempt {attempt} with Retry-After: 0 slept {d:?}, expected >={min_ms}ms"
543 );
544 }
545 }
546
547 #[test]
548 fn test_should_retry_retry_after_clamped_to_max_backoff() {
549 let client = HttpClientBuilder::new("https://example.com")
550 .build()
551 .unwrap();
552 let d = client
554 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, Some("60"))
555 .unwrap();
556 assert_eq!(d, Duration::from_millis(10_000));
557 }
558
559 #[test]
560 fn test_should_retry_retry_after_invalid_falls_back() {
561 let client = HttpClientBuilder::new("https://example.com")
562 .build()
563 .unwrap();
564 let d = client
566 .should_retry(
567 StatusCode::TOO_MANY_REQUESTS,
568 0,
569 Some("Wed, 21 Oct 2025 07:28:00 GMT"),
570 )
571 .unwrap();
572 let ms = d.as_millis() as u64;
574 assert!(
575 (375..=625).contains(&ms),
576 "expected fallback backoff in [375, 625], got {ms}"
577 );
578 }
579
580 #[tokio::test]
583 async fn test_builder_with_rate_limiter() {
584 let client = HttpClientBuilder::new("https://example.com")
585 .with_rate_limiter(RateLimiter::clob_default())
586 .build()
587 .unwrap();
588 let start = std::time::Instant::now();
589 client
590 .acquire_rate_limit("/order", Some(&reqwest::Method::POST))
591 .await;
592 assert!(start.elapsed() < Duration::from_millis(50));
593 }
594
595 #[tokio::test]
596 async fn test_builder_without_rate_limiter() {
597 let client = HttpClientBuilder::new("https://example.com")
598 .build()
599 .unwrap();
600 let start = std::time::Instant::now();
601 client
602 .acquire_rate_limit("/order", Some(&reqwest::Method::POST))
603 .await;
604 assert!(start.elapsed() < Duration::from_millis(10));
605 }
606
607 #[tokio::test]
610 async fn test_acquire_concurrency_none_when_not_configured() {
611 let client = HttpClientBuilder::new("https://example.com")
612 .build()
613 .unwrap();
614 assert!(client.acquire_concurrency().await.is_none());
615 }
616
617 #[tokio::test]
618 async fn test_acquire_concurrency_returns_permit() {
619 let client = HttpClientBuilder::new("https://example.com")
620 .with_max_concurrent(2)
621 .build()
622 .unwrap();
623 let permit = client.acquire_concurrency().await;
624 assert!(permit.is_some());
625 }
626
627 #[tokio::test]
628 async fn test_concurrency_shared_across_clones() {
629 let client = HttpClientBuilder::new("https://example.com")
630 .with_max_concurrent(1)
631 .build()
632 .unwrap();
633 let clone = client.clone();
634
635 let _permit = client.acquire_concurrency().await.unwrap();
637
638 let result =
640 tokio::time::timeout(Duration::from_millis(50), clone.acquire_concurrency()).await;
641 assert!(result.is_err(), "clone should block when permit is held");
642 }
643
644 #[tokio::test]
645 async fn test_concurrency_limits_parallel_tasks() {
646 let client = HttpClientBuilder::new("https://example.com")
647 .with_max_concurrent(2)
648 .build()
649 .unwrap();
650
651 let start = std::time::Instant::now();
652 let mut handles = Vec::new();
653 for _ in 0..4 {
654 let c = client.clone();
655 handles.push(tokio::spawn(async move {
656 let _permit = c.acquire_concurrency().await;
657 tokio::time::sleep(Duration::from_millis(50)).await;
658 }));
659 }
660 for h in handles {
661 h.await.unwrap();
662 }
663 assert!(
665 start.elapsed() >= Duration::from_millis(90),
666 "expected ~100ms, got {:?}",
667 start.elapsed()
668 );
669 }
670
671 #[tokio::test]
672 async fn test_builder_with_max_concurrent() {
673 let client = HttpClientBuilder::new("https://example.com")
674 .with_max_concurrent(5)
675 .build()
676 .unwrap();
677 let mut permits = Vec::new();
679 for _ in 0..5 {
680 permits.push(client.acquire_concurrency().await);
681 }
682 assert!(permits.iter().all(|p| p.is_some()));
683
684 let result =
686 tokio::time::timeout(Duration::from_millis(50), client.acquire_concurrency()).await;
687 assert!(result.is_err());
688 }
689
690 #[tokio::test]
693 async fn test_get_bytes_returns_body_verbatim() {
694 let mut server = mockito::Server::new_async().await;
695 let body: Vec<u8> = vec![0x50, 0x4B, 0x03, 0x04, 0x00, 0xFF, 0xFE, 0x42];
697 let mock = server
698 .mock("GET", "/v1/accounting/snapshot")
699 .match_query(mockito::Matcher::UrlEncoded("user".into(), "0xabc".into()))
700 .with_status(200)
701 .with_header("content-type", "application/zip")
702 .with_body(body.clone())
703 .create_async()
704 .await;
705
706 let client = HttpClientBuilder::new(server.url()).build().unwrap();
707 let out = client
708 .get_bytes(
709 "/v1/accounting/snapshot",
710 &[("user".to_string(), "0xabc".to_string())],
711 )
712 .await
713 .unwrap();
714 assert_eq!(out, body);
715 mock.assert_async().await;
716 }
717
718 #[tokio::test]
719 async fn test_get_bytes_maps_non_2xx_to_api_error() {
720 let mut server = mockito::Server::new_async().await;
721 let mock = server
722 .mock("GET", "/does-not-exist")
723 .with_status(404)
724 .with_header("content-type", "application/json")
725 .with_body(r#"{"error": "not found"}"#)
726 .create_async()
727 .await;
728
729 let client = HttpClientBuilder::new(server.url()).build().unwrap();
730 let err = client.get_bytes("/does-not-exist", &[]).await.unwrap_err();
731 match err {
732 ApiError::Api { status, message } => {
733 assert_eq!(status, 404);
734 assert_eq!(message, "not found");
735 }
736 other => panic!("expected ApiError::Api, got {other:?}"),
737 }
738 mock.assert_async().await;
739 }
740
741 #[tokio::test]
742 async fn test_get_bytes_no_query_params() {
743 let mut server = mockito::Server::new_async().await;
744 let mock = server
745 .mock("GET", "/raw")
746 .with_status(200)
747 .with_body(&b"hello"[..])
748 .create_async()
749 .await;
750
751 let client = HttpClientBuilder::new(server.url()).build().unwrap();
752 let out = client.get_bytes("/raw", &[]).await.unwrap();
753 assert_eq!(out, b"hello");
754 mock.assert_async().await;
755 }
756
757 #[tokio::test]
760 async fn gzip_is_off_unless_asked_for() {
761 let mut server = mockito::Server::new_async().await;
765 let mock = server
766 .mock("GET", "/plain")
767 .match_header("accept-encoding", mockito::Matcher::Missing)
768 .with_body("ok")
769 .create_async()
770 .await;
771
772 let client = HttpClientBuilder::new(server.url()).build().unwrap();
773 assert_eq!(client.get_bytes("/plain", &[]).await.unwrap(), b"ok");
774 mock.assert_async().await;
775 }
776
777 #[tokio::test]
778 async fn gzip_asks_for_and_decodes_a_compressed_body() {
779 const GZIPPED: &[u8] = &[
781 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xab, 0x56, 0x2a, 0x4e,
782 0x2d, 0x2a, 0x4b, 0x2d, 0x0a, 0xc9, 0xcc, 0x4d, 0x55, 0xb2, 0x32, 0xac, 0x05, 0x00,
783 0xe2, 0x1d, 0x3e, 0x1a, 0x10, 0x00, 0x00, 0x00,
784 ];
785 let mut server = mockito::Server::new_async().await;
786 let mock = server
787 .mock("GET", "/time")
788 .match_header("accept-encoding", mockito::Matcher::Regex("gzip".into()))
789 .with_header("content-encoding", "gzip")
790 .with_body(GZIPPED)
791 .create_async()
792 .await;
793
794 let client = HttpClientBuilder::new(server.url())
795 .gzip(true)
796 .build()
797 .unwrap();
798 let body = client.get_bytes("/time", &[]).await.unwrap();
799 assert_eq!(body, br#"{"serverTime":1}"#);
800 mock.assert_async().await;
801 }
802
803 #[test]
804 fn with_base_url_retargets_and_shares_transport() {
805 let client = HttpClientBuilder::new("https://data-api.polymarket.com")
806 .with_max_concurrent(4)
807 .build()
808 .unwrap();
809 let sibling = client
810 .with_base_url("https://user-pnl-api.polymarket.com")
811 .unwrap();
812
813 assert_eq!(
814 sibling.base_url.host_str(),
815 Some("user-pnl-api.polymarket.com")
816 );
817 assert_eq!(client.base_url.host_str(), Some("data-api.polymarket.com"));
819 assert!(sibling.concurrency_limiter.is_some());
822 }
823
824 #[test]
825 fn with_base_url_rejects_a_malformed_url() {
826 let client = HttpClientBuilder::new("https://data-api.polymarket.com")
827 .build()
828 .unwrap();
829 assert!(client.with_base_url("not-a-url").is_err());
830 }
831
832 #[test]
833 fn base_url_path_prefixes_are_dropped() {
834 let client = HttpClientBuilder::new("http://localhost:8080/proxy")
838 .build()
839 .unwrap();
840 assert_eq!(
841 client.base_url.join("/user-pnl").unwrap().as_str(),
842 "http://localhost:8080/user-pnl",
843 "a path prefix in the base URL is not preserved"
844 );
845
846 let with_slash = client
847 .with_base_url("http://localhost:8080/proxy/")
848 .unwrap();
849 assert_eq!(
850 with_slash.base_url.join("/user-pnl").unwrap().as_str(),
851 "http://localhost:8080/user-pnl",
852 "a trailing slash does not preserve the prefix either"
853 );
854 }
855}