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: Option<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: None,
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 = Some(enabled);
327 self
328 }
329
330 pub fn build(self) -> Result<HttpClient, ApiError> {
332 let mut builder = reqwest::Client::builder();
333 if let Some(enabled) = self.gzip {
334 builder = builder.gzip(enabled);
335 }
336 let client = builder
337 .timeout(Duration::from_millis(self.timeout_ms))
338 .connect_timeout(Duration::from_secs(10))
339 .redirect(reqwest::redirect::Policy::none())
340 .pool_max_idle_per_host(self.pool_size)
341 .build()?;
342
343 let base_url = Url::parse(&self.base_url)?;
344
345 Ok(HttpClient {
346 client,
347 base_url,
348 rate_limiter: self.rate_limiter,
349 retry_config: self.retry_config,
350 concurrency_limiter: self.max_concurrent.map(|n| Arc::new(Semaphore::new(n))),
351 })
352 }
353}
354
355impl Default for HttpClientBuilder {
356 fn default() -> Self {
357 Self {
358 base_url: String::new(),
359 timeout_ms: DEFAULT_TIMEOUT_MS,
360 pool_size: DEFAULT_POOL_SIZE,
361 rate_limiter: None,
362 retry_config: RetryConfig::default(),
363 max_concurrent: None,
364 gzip: None,
365 }
366 }
367}
368
369#[cfg(test)]
370mod tests {
371 use super::*;
372
373 #[test]
376 fn test_should_retry_429_under_max() {
377 let client = HttpClientBuilder::new("https://example.com")
378 .build()
379 .unwrap();
380 assert!(client
382 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, None)
383 .is_some());
384 assert!(client
385 .should_retry(StatusCode::TOO_MANY_REQUESTS, 2, None)
386 .is_some());
387 }
388
389 #[test]
390 fn test_should_retry_429_at_max() {
391 let client = HttpClientBuilder::new("https://example.com")
392 .build()
393 .unwrap();
394 assert!(client
396 .should_retry(StatusCode::TOO_MANY_REQUESTS, 3, None)
397 .is_none());
398 }
399
400 #[test]
401 fn test_should_retry_425_under_max() {
402 let client = HttpClientBuilder::new("https://example.com")
403 .build()
404 .unwrap();
405 assert!(client
407 .should_retry(StatusCode::TOO_EARLY, 0, None)
408 .is_some());
409 assert!(client
410 .should_retry(StatusCode::TOO_EARLY, 2, None)
411 .is_some());
412 }
413
414 #[test]
415 fn test_should_retry_425_at_max() {
416 let client = HttpClientBuilder::new("https://example.com")
417 .build()
418 .unwrap();
419 assert!(client
420 .should_retry(StatusCode::TOO_EARLY, 3, None)
421 .is_none());
422 }
423
424 #[test]
425 fn test_should_retry_ignores_other_statuses() {
426 let client = HttpClientBuilder::new("https://example.com")
427 .build()
428 .unwrap();
429 for status in [
430 StatusCode::OK,
431 StatusCode::INTERNAL_SERVER_ERROR,
432 StatusCode::BAD_GATEWAY,
433 StatusCode::BAD_REQUEST,
434 StatusCode::FORBIDDEN,
435 StatusCode::NOT_FOUND,
436 StatusCode::SERVICE_UNAVAILABLE,
440 ] {
441 assert!(
442 client.should_retry(status, 0, None).is_none(),
443 "expected None for {status}"
444 );
445 }
446 }
447
448 #[test]
449 fn test_should_retry_5xx_not_retried_despite_being_is_retriable() {
450 let client = HttpClientBuilder::new("https://example.com")
453 .build()
454 .unwrap();
455 assert!(client
456 .should_retry(StatusCode::INTERNAL_SERVER_ERROR, 0, None)
457 .is_none());
458 assert!(ApiError::Api {
459 status: 500,
460 message: String::new()
461 }
462 .is_retriable());
463 }
464
465 #[test]
466 fn test_should_retry_custom_config() {
467 let client = HttpClientBuilder::new("https://example.com")
468 .with_retry_config(RetryConfig {
469 max_retries: 1,
470 ..RetryConfig::default()
471 })
472 .build()
473 .unwrap();
474 assert!(client
475 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, None)
476 .is_some());
477 assert!(client
478 .should_retry(StatusCode::TOO_MANY_REQUESTS, 1, None)
479 .is_none());
480 }
481
482 #[test]
483 fn test_should_retry_uses_retry_after_header() {
484 let client = HttpClientBuilder::new("https://example.com")
485 .build()
486 .unwrap();
487 let d = client
488 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, Some("2"))
489 .unwrap();
490 assert_eq!(d, Duration::from_millis(2000));
491 }
492
493 #[test]
494 fn test_should_retry_retry_after_fractional_seconds() {
495 let client = HttpClientBuilder::new("https://example.com")
496 .build()
497 .unwrap();
498 let d = client
503 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, Some("1.5"))
504 .unwrap();
505 assert_eq!(d, Duration::from_millis(1500));
506 }
507
508 #[test]
509 fn retry_after_below_our_own_backoff_does_not_shorten_the_wait() {
510 let client = HttpClientBuilder::new("https://example.com")
511 .build()
512 .unwrap();
513 for header in ["0", "0.0", "-1", "-30", "0.0001"] {
520 let d = client
521 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, Some(header))
522 .unwrap();
523 assert!(
524 d >= Duration::from_millis(375),
525 "Retry-After: {header:?} produced a {d:?} sleep; the floor is the \
526 client's own attempt-0 backoff, >=375ms after jitter"
527 );
528 }
529 }
530
531 #[test]
532 fn retry_after_zero_still_backs_off_exponentially_across_attempts() {
533 let client = HttpClientBuilder::new("https://example.com")
534 .build()
535 .unwrap();
536 for (attempt, min_ms) in [(0u32, 375u64), (1, 750), (2, 1_500)] {
540 let d = client
541 .should_retry(StatusCode::TOO_MANY_REQUESTS, attempt, Some("0"))
542 .unwrap();
543 assert!(
544 d >= Duration::from_millis(min_ms),
545 "attempt {attempt} with Retry-After: 0 slept {d:?}, expected >={min_ms}ms"
546 );
547 }
548 }
549
550 #[test]
551 fn test_should_retry_retry_after_clamped_to_max_backoff() {
552 let client = HttpClientBuilder::new("https://example.com")
553 .build()
554 .unwrap();
555 let d = client
557 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, Some("60"))
558 .unwrap();
559 assert_eq!(d, Duration::from_millis(10_000));
560 }
561
562 #[test]
563 fn test_should_retry_retry_after_invalid_falls_back() {
564 let client = HttpClientBuilder::new("https://example.com")
565 .build()
566 .unwrap();
567 let d = client
569 .should_retry(
570 StatusCode::TOO_MANY_REQUESTS,
571 0,
572 Some("Wed, 21 Oct 2025 07:28:00 GMT"),
573 )
574 .unwrap();
575 let ms = d.as_millis() as u64;
577 assert!(
578 (375..=625).contains(&ms),
579 "expected fallback backoff in [375, 625], got {ms}"
580 );
581 }
582
583 #[tokio::test]
586 async fn test_builder_with_rate_limiter() {
587 let client = HttpClientBuilder::new("https://example.com")
588 .with_rate_limiter(RateLimiter::clob_default())
589 .build()
590 .unwrap();
591 let start = std::time::Instant::now();
592 client
593 .acquire_rate_limit("/order", Some(&reqwest::Method::POST))
594 .await;
595 assert!(start.elapsed() < Duration::from_millis(50));
596 }
597
598 #[tokio::test]
599 async fn test_builder_without_rate_limiter() {
600 let client = HttpClientBuilder::new("https://example.com")
601 .build()
602 .unwrap();
603 let start = std::time::Instant::now();
604 client
605 .acquire_rate_limit("/order", Some(&reqwest::Method::POST))
606 .await;
607 assert!(start.elapsed() < Duration::from_millis(10));
608 }
609
610 #[tokio::test]
613 async fn test_acquire_concurrency_none_when_not_configured() {
614 let client = HttpClientBuilder::new("https://example.com")
615 .build()
616 .unwrap();
617 assert!(client.acquire_concurrency().await.is_none());
618 }
619
620 #[tokio::test]
621 async fn test_acquire_concurrency_returns_permit() {
622 let client = HttpClientBuilder::new("https://example.com")
623 .with_max_concurrent(2)
624 .build()
625 .unwrap();
626 let permit = client.acquire_concurrency().await;
627 assert!(permit.is_some());
628 }
629
630 #[tokio::test]
631 async fn test_concurrency_shared_across_clones() {
632 let client = HttpClientBuilder::new("https://example.com")
633 .with_max_concurrent(1)
634 .build()
635 .unwrap();
636 let clone = client.clone();
637
638 let _permit = client.acquire_concurrency().await.unwrap();
640
641 let result =
643 tokio::time::timeout(Duration::from_millis(50), clone.acquire_concurrency()).await;
644 assert!(result.is_err(), "clone should block when permit is held");
645 }
646
647 #[tokio::test]
648 async fn test_concurrency_limits_parallel_tasks() {
649 let client = HttpClientBuilder::new("https://example.com")
650 .with_max_concurrent(2)
651 .build()
652 .unwrap();
653
654 let start = std::time::Instant::now();
655 let mut handles = Vec::new();
656 for _ in 0..4 {
657 let c = client.clone();
658 handles.push(tokio::spawn(async move {
659 let _permit = c.acquire_concurrency().await;
660 tokio::time::sleep(Duration::from_millis(50)).await;
661 }));
662 }
663 for h in handles {
664 h.await.unwrap();
665 }
666 assert!(
668 start.elapsed() >= Duration::from_millis(90),
669 "expected ~100ms, got {:?}",
670 start.elapsed()
671 );
672 }
673
674 #[tokio::test]
675 async fn test_builder_with_max_concurrent() {
676 let client = HttpClientBuilder::new("https://example.com")
677 .with_max_concurrent(5)
678 .build()
679 .unwrap();
680 let mut permits = Vec::new();
682 for _ in 0..5 {
683 permits.push(client.acquire_concurrency().await);
684 }
685 assert!(permits.iter().all(|p| p.is_some()));
686
687 let result =
689 tokio::time::timeout(Duration::from_millis(50), client.acquire_concurrency()).await;
690 assert!(result.is_err());
691 }
692
693 #[tokio::test]
696 async fn test_get_bytes_returns_body_verbatim() {
697 let mut server = mockito::Server::new_async().await;
698 let body: Vec<u8> = vec![0x50, 0x4B, 0x03, 0x04, 0x00, 0xFF, 0xFE, 0x42];
700 let mock = server
701 .mock("GET", "/v1/accounting/snapshot")
702 .match_query(mockito::Matcher::UrlEncoded("user".into(), "0xabc".into()))
703 .with_status(200)
704 .with_header("content-type", "application/zip")
705 .with_body(body.clone())
706 .create_async()
707 .await;
708
709 let client = HttpClientBuilder::new(server.url()).build().unwrap();
710 let out = client
711 .get_bytes(
712 "/v1/accounting/snapshot",
713 &[("user".to_string(), "0xabc".to_string())],
714 )
715 .await
716 .unwrap();
717 assert_eq!(out, body);
718 mock.assert_async().await;
719 }
720
721 #[tokio::test]
722 async fn test_get_bytes_maps_non_2xx_to_api_error() {
723 let mut server = mockito::Server::new_async().await;
724 let mock = server
725 .mock("GET", "/does-not-exist")
726 .with_status(404)
727 .with_header("content-type", "application/json")
728 .with_body(r#"{"error": "not found"}"#)
729 .create_async()
730 .await;
731
732 let client = HttpClientBuilder::new(server.url()).build().unwrap();
733 let err = client.get_bytes("/does-not-exist", &[]).await.unwrap_err();
734 match err {
735 ApiError::Api { status, message } => {
736 assert_eq!(status, 404);
737 assert_eq!(message, "not found");
738 }
739 other => panic!("expected ApiError::Api, got {other:?}"),
740 }
741 mock.assert_async().await;
742 }
743
744 #[tokio::test]
745 async fn test_get_bytes_no_query_params() {
746 let mut server = mockito::Server::new_async().await;
747 let mock = server
748 .mock("GET", "/raw")
749 .with_status(200)
750 .with_body(&b"hello"[..])
751 .create_async()
752 .await;
753
754 let client = HttpClientBuilder::new(server.url()).build().unwrap();
755 let out = client.get_bytes("/raw", &[]).await.unwrap();
756 assert_eq!(out, b"hello");
757 mock.assert_async().await;
758 }
759
760 #[tokio::test]
763 async fn gzip_follows_reqwest_unless_set() {
764 let mut server = mockito::Server::new_async().await;
768 let mock = server
769 .mock("GET", "/plain")
770 .match_header("accept-encoding", mockito::Matcher::Regex("gzip".into()))
771 .with_body("ok")
772 .create_async()
773 .await;
774
775 let client = HttpClientBuilder::new(server.url()).build().unwrap();
776 assert_eq!(client.get_bytes("/plain", &[]).await.unwrap(), b"ok");
777 mock.assert_async().await;
778 }
779
780 #[tokio::test]
781 async fn gzip_false_asks_for_no_compression() {
782 let mut server = mockito::Server::new_async().await;
783 let mock = server
784 .mock("GET", "/plain")
785 .match_header("accept-encoding", mockito::Matcher::Missing)
786 .with_body("ok")
787 .create_async()
788 .await;
789
790 let client = HttpClientBuilder::new(server.url())
791 .gzip(false)
792 .build()
793 .unwrap();
794 assert_eq!(client.get_bytes("/plain", &[]).await.unwrap(), b"ok");
795 mock.assert_async().await;
796 }
797
798 #[tokio::test]
799 async fn gzip_asks_for_and_decodes_a_compressed_body() {
800 const GZIPPED: &[u8] = &[
802 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xab, 0x56, 0x2a, 0x4e,
803 0x2d, 0x2a, 0x4b, 0x2d, 0x0a, 0xc9, 0xcc, 0x4d, 0x55, 0xb2, 0x32, 0xac, 0x05, 0x00,
804 0xe2, 0x1d, 0x3e, 0x1a, 0x10, 0x00, 0x00, 0x00,
805 ];
806 let mut server = mockito::Server::new_async().await;
807 let mock = server
808 .mock("GET", "/time")
809 .match_header("accept-encoding", mockito::Matcher::Regex("gzip".into()))
810 .with_header("content-encoding", "gzip")
811 .with_body(GZIPPED)
812 .create_async()
813 .await;
814
815 let client = HttpClientBuilder::new(server.url())
816 .gzip(true)
817 .build()
818 .unwrap();
819 let body = client.get_bytes("/time", &[]).await.unwrap();
820 assert_eq!(body, br#"{"serverTime":1}"#);
821 mock.assert_async().await;
822 }
823
824 #[test]
825 fn with_base_url_retargets_and_shares_transport() {
826 let client = HttpClientBuilder::new("https://data-api.polymarket.com")
827 .with_max_concurrent(4)
828 .build()
829 .unwrap();
830 let sibling = client
831 .with_base_url("https://user-pnl-api.polymarket.com")
832 .unwrap();
833
834 assert_eq!(
835 sibling.base_url.host_str(),
836 Some("user-pnl-api.polymarket.com")
837 );
838 assert_eq!(client.base_url.host_str(), Some("data-api.polymarket.com"));
840 assert!(sibling.concurrency_limiter.is_some());
843 }
844
845 #[test]
846 fn with_base_url_rejects_a_malformed_url() {
847 let client = HttpClientBuilder::new("https://data-api.polymarket.com")
848 .build()
849 .unwrap();
850 assert!(client.with_base_url("not-a-url").is_err());
851 }
852
853 #[test]
854 fn base_url_path_prefixes_are_dropped() {
855 let client = HttpClientBuilder::new("http://localhost:8080/proxy")
859 .build()
860 .unwrap();
861 assert_eq!(
862 client.base_url.join("/user-pnl").unwrap().as_str(),
863 "http://localhost:8080/user-pnl",
864 "a path prefix in the base URL is not preserved"
865 );
866
867 let with_slash = client
868 .with_base_url("http://localhost:8080/proxy/")
869 .unwrap();
870 assert_eq!(
871 with_slash.base_url.join("/user-pnl").unwrap().as_str(),
872 "http://localhost:8080/user-pnl",
873 "a trailing slash does not preserve the prefix either"
874 );
875 }
876}