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}
265
266impl HttpClientBuilder {
267 pub fn new(base_url: impl Into<String>) -> Self {
269 Self {
270 base_url: base_url.into(),
271 timeout_ms: DEFAULT_TIMEOUT_MS,
272 pool_size: DEFAULT_POOL_SIZE,
273 rate_limiter: None,
274 retry_config: RetryConfig::default(),
275 max_concurrent: None,
276 }
277 }
278
279 pub fn timeout_ms(mut self, timeout: u64) -> Self {
283 self.timeout_ms = timeout;
284 self
285 }
286
287 pub fn pool_size(mut self, size: usize) -> Self {
291 self.pool_size = size;
292 self
293 }
294
295 pub fn with_rate_limiter(mut self, limiter: RateLimiter) -> Self {
297 self.rate_limiter = Some(limiter);
298 self
299 }
300
301 pub fn with_retry_config(mut self, config: RetryConfig) -> Self {
303 self.retry_config = config;
304 self
305 }
306
307 pub fn with_max_concurrent(mut self, max: usize) -> Self {
312 self.max_concurrent = Some(max);
313 self
314 }
315
316 pub fn build(self) -> Result<HttpClient, ApiError> {
318 let client = reqwest::Client::builder()
319 .timeout(Duration::from_millis(self.timeout_ms))
320 .connect_timeout(Duration::from_secs(10))
321 .redirect(reqwest::redirect::Policy::none())
322 .pool_max_idle_per_host(self.pool_size)
323 .build()?;
324
325 let base_url = Url::parse(&self.base_url)?;
326
327 Ok(HttpClient {
328 client,
329 base_url,
330 rate_limiter: self.rate_limiter,
331 retry_config: self.retry_config,
332 concurrency_limiter: self.max_concurrent.map(|n| Arc::new(Semaphore::new(n))),
333 })
334 }
335}
336
337impl Default for HttpClientBuilder {
338 fn default() -> Self {
339 Self {
340 base_url: String::new(),
341 timeout_ms: DEFAULT_TIMEOUT_MS,
342 pool_size: DEFAULT_POOL_SIZE,
343 rate_limiter: None,
344 retry_config: RetryConfig::default(),
345 max_concurrent: None,
346 }
347 }
348}
349
350#[cfg(test)]
351mod tests {
352 use super::*;
353
354 #[test]
357 fn test_should_retry_429_under_max() {
358 let client = HttpClientBuilder::new("https://example.com")
359 .build()
360 .unwrap();
361 assert!(client
363 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, None)
364 .is_some());
365 assert!(client
366 .should_retry(StatusCode::TOO_MANY_REQUESTS, 2, None)
367 .is_some());
368 }
369
370 #[test]
371 fn test_should_retry_429_at_max() {
372 let client = HttpClientBuilder::new("https://example.com")
373 .build()
374 .unwrap();
375 assert!(client
377 .should_retry(StatusCode::TOO_MANY_REQUESTS, 3, None)
378 .is_none());
379 }
380
381 #[test]
382 fn test_should_retry_425_under_max() {
383 let client = HttpClientBuilder::new("https://example.com")
384 .build()
385 .unwrap();
386 assert!(client
388 .should_retry(StatusCode::TOO_EARLY, 0, None)
389 .is_some());
390 assert!(client
391 .should_retry(StatusCode::TOO_EARLY, 2, None)
392 .is_some());
393 }
394
395 #[test]
396 fn test_should_retry_425_at_max() {
397 let client = HttpClientBuilder::new("https://example.com")
398 .build()
399 .unwrap();
400 assert!(client
401 .should_retry(StatusCode::TOO_EARLY, 3, None)
402 .is_none());
403 }
404
405 #[test]
406 fn test_should_retry_ignores_other_statuses() {
407 let client = HttpClientBuilder::new("https://example.com")
408 .build()
409 .unwrap();
410 for status in [
411 StatusCode::OK,
412 StatusCode::INTERNAL_SERVER_ERROR,
413 StatusCode::BAD_GATEWAY,
414 StatusCode::BAD_REQUEST,
415 StatusCode::FORBIDDEN,
416 StatusCode::NOT_FOUND,
417 StatusCode::SERVICE_UNAVAILABLE,
421 ] {
422 assert!(
423 client.should_retry(status, 0, None).is_none(),
424 "expected None for {status}"
425 );
426 }
427 }
428
429 #[test]
430 fn test_should_retry_5xx_not_retried_despite_being_is_retriable() {
431 let client = HttpClientBuilder::new("https://example.com")
434 .build()
435 .unwrap();
436 assert!(client
437 .should_retry(StatusCode::INTERNAL_SERVER_ERROR, 0, None)
438 .is_none());
439 assert!(ApiError::Api {
440 status: 500,
441 message: String::new()
442 }
443 .is_retriable());
444 }
445
446 #[test]
447 fn test_should_retry_custom_config() {
448 let client = HttpClientBuilder::new("https://example.com")
449 .with_retry_config(RetryConfig {
450 max_retries: 1,
451 ..RetryConfig::default()
452 })
453 .build()
454 .unwrap();
455 assert!(client
456 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, None)
457 .is_some());
458 assert!(client
459 .should_retry(StatusCode::TOO_MANY_REQUESTS, 1, None)
460 .is_none());
461 }
462
463 #[test]
464 fn test_should_retry_uses_retry_after_header() {
465 let client = HttpClientBuilder::new("https://example.com")
466 .build()
467 .unwrap();
468 let d = client
469 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, Some("2"))
470 .unwrap();
471 assert_eq!(d, Duration::from_millis(2000));
472 }
473
474 #[test]
475 fn test_should_retry_retry_after_fractional_seconds() {
476 let client = HttpClientBuilder::new("https://example.com")
477 .build()
478 .unwrap();
479 let d = client
484 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, Some("1.5"))
485 .unwrap();
486 assert_eq!(d, Duration::from_millis(1500));
487 }
488
489 #[test]
490 fn retry_after_below_our_own_backoff_does_not_shorten_the_wait() {
491 let client = HttpClientBuilder::new("https://example.com")
492 .build()
493 .unwrap();
494 for header in ["0", "0.0", "-1", "-30", "0.0001"] {
501 let d = client
502 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, Some(header))
503 .unwrap();
504 assert!(
505 d >= Duration::from_millis(375),
506 "Retry-After: {header:?} produced a {d:?} sleep; the floor is the \
507 client's own attempt-0 backoff, >=375ms after jitter"
508 );
509 }
510 }
511
512 #[test]
513 fn retry_after_zero_still_backs_off_exponentially_across_attempts() {
514 let client = HttpClientBuilder::new("https://example.com")
515 .build()
516 .unwrap();
517 for (attempt, min_ms) in [(0u32, 375u64), (1, 750), (2, 1_500)] {
521 let d = client
522 .should_retry(StatusCode::TOO_MANY_REQUESTS, attempt, Some("0"))
523 .unwrap();
524 assert!(
525 d >= Duration::from_millis(min_ms),
526 "attempt {attempt} with Retry-After: 0 slept {d:?}, expected >={min_ms}ms"
527 );
528 }
529 }
530
531 #[test]
532 fn test_should_retry_retry_after_clamped_to_max_backoff() {
533 let client = HttpClientBuilder::new("https://example.com")
534 .build()
535 .unwrap();
536 let d = client
538 .should_retry(StatusCode::TOO_MANY_REQUESTS, 0, Some("60"))
539 .unwrap();
540 assert_eq!(d, Duration::from_millis(10_000));
541 }
542
543 #[test]
544 fn test_should_retry_retry_after_invalid_falls_back() {
545 let client = HttpClientBuilder::new("https://example.com")
546 .build()
547 .unwrap();
548 let d = client
550 .should_retry(
551 StatusCode::TOO_MANY_REQUESTS,
552 0,
553 Some("Wed, 21 Oct 2025 07:28:00 GMT"),
554 )
555 .unwrap();
556 let ms = d.as_millis() as u64;
558 assert!(
559 (375..=625).contains(&ms),
560 "expected fallback backoff in [375, 625], got {ms}"
561 );
562 }
563
564 #[tokio::test]
567 async fn test_builder_with_rate_limiter() {
568 let client = HttpClientBuilder::new("https://example.com")
569 .with_rate_limiter(RateLimiter::clob_default())
570 .build()
571 .unwrap();
572 let start = std::time::Instant::now();
573 client
574 .acquire_rate_limit("/order", Some(&reqwest::Method::POST))
575 .await;
576 assert!(start.elapsed() < Duration::from_millis(50));
577 }
578
579 #[tokio::test]
580 async fn test_builder_without_rate_limiter() {
581 let client = HttpClientBuilder::new("https://example.com")
582 .build()
583 .unwrap();
584 let start = std::time::Instant::now();
585 client
586 .acquire_rate_limit("/order", Some(&reqwest::Method::POST))
587 .await;
588 assert!(start.elapsed() < Duration::from_millis(10));
589 }
590
591 #[tokio::test]
594 async fn test_acquire_concurrency_none_when_not_configured() {
595 let client = HttpClientBuilder::new("https://example.com")
596 .build()
597 .unwrap();
598 assert!(client.acquire_concurrency().await.is_none());
599 }
600
601 #[tokio::test]
602 async fn test_acquire_concurrency_returns_permit() {
603 let client = HttpClientBuilder::new("https://example.com")
604 .with_max_concurrent(2)
605 .build()
606 .unwrap();
607 let permit = client.acquire_concurrency().await;
608 assert!(permit.is_some());
609 }
610
611 #[tokio::test]
612 async fn test_concurrency_shared_across_clones() {
613 let client = HttpClientBuilder::new("https://example.com")
614 .with_max_concurrent(1)
615 .build()
616 .unwrap();
617 let clone = client.clone();
618
619 let _permit = client.acquire_concurrency().await.unwrap();
621
622 let result =
624 tokio::time::timeout(Duration::from_millis(50), clone.acquire_concurrency()).await;
625 assert!(result.is_err(), "clone should block when permit is held");
626 }
627
628 #[tokio::test]
629 async fn test_concurrency_limits_parallel_tasks() {
630 let client = HttpClientBuilder::new("https://example.com")
631 .with_max_concurrent(2)
632 .build()
633 .unwrap();
634
635 let start = std::time::Instant::now();
636 let mut handles = Vec::new();
637 for _ in 0..4 {
638 let c = client.clone();
639 handles.push(tokio::spawn(async move {
640 let _permit = c.acquire_concurrency().await;
641 tokio::time::sleep(Duration::from_millis(50)).await;
642 }));
643 }
644 for h in handles {
645 h.await.unwrap();
646 }
647 assert!(
649 start.elapsed() >= Duration::from_millis(90),
650 "expected ~100ms, got {:?}",
651 start.elapsed()
652 );
653 }
654
655 #[tokio::test]
656 async fn test_builder_with_max_concurrent() {
657 let client = HttpClientBuilder::new("https://example.com")
658 .with_max_concurrent(5)
659 .build()
660 .unwrap();
661 let mut permits = Vec::new();
663 for _ in 0..5 {
664 permits.push(client.acquire_concurrency().await);
665 }
666 assert!(permits.iter().all(|p| p.is_some()));
667
668 let result =
670 tokio::time::timeout(Duration::from_millis(50), client.acquire_concurrency()).await;
671 assert!(result.is_err());
672 }
673
674 #[tokio::test]
677 async fn test_get_bytes_returns_body_verbatim() {
678 let mut server = mockito::Server::new_async().await;
679 let body: Vec<u8> = vec![0x50, 0x4B, 0x03, 0x04, 0x00, 0xFF, 0xFE, 0x42];
681 let mock = server
682 .mock("GET", "/v1/accounting/snapshot")
683 .match_query(mockito::Matcher::UrlEncoded("user".into(), "0xabc".into()))
684 .with_status(200)
685 .with_header("content-type", "application/zip")
686 .with_body(body.clone())
687 .create_async()
688 .await;
689
690 let client = HttpClientBuilder::new(server.url()).build().unwrap();
691 let out = client
692 .get_bytes(
693 "/v1/accounting/snapshot",
694 &[("user".to_string(), "0xabc".to_string())],
695 )
696 .await
697 .unwrap();
698 assert_eq!(out, body);
699 mock.assert_async().await;
700 }
701
702 #[tokio::test]
703 async fn test_get_bytes_maps_non_2xx_to_api_error() {
704 let mut server = mockito::Server::new_async().await;
705 let mock = server
706 .mock("GET", "/does-not-exist")
707 .with_status(404)
708 .with_header("content-type", "application/json")
709 .with_body(r#"{"error": "not found"}"#)
710 .create_async()
711 .await;
712
713 let client = HttpClientBuilder::new(server.url()).build().unwrap();
714 let err = client.get_bytes("/does-not-exist", &[]).await.unwrap_err();
715 match err {
716 ApiError::Api { status, message } => {
717 assert_eq!(status, 404);
718 assert_eq!(message, "not found");
719 }
720 other => panic!("expected ApiError::Api, got {other:?}"),
721 }
722 mock.assert_async().await;
723 }
724
725 #[tokio::test]
726 async fn test_get_bytes_no_query_params() {
727 let mut server = mockito::Server::new_async().await;
728 let mock = server
729 .mock("GET", "/raw")
730 .with_status(200)
731 .with_body(&b"hello"[..])
732 .create_async()
733 .await;
734
735 let client = HttpClientBuilder::new(server.url()).build().unwrap();
736 let out = client.get_bytes("/raw", &[]).await.unwrap();
737 assert_eq!(out, b"hello");
738 mock.assert_async().await;
739 }
740
741 #[test]
742 fn with_base_url_retargets_and_shares_transport() {
743 let client = HttpClientBuilder::new("https://data-api.polymarket.com")
744 .with_max_concurrent(4)
745 .build()
746 .unwrap();
747 let sibling = client
748 .with_base_url("https://user-pnl-api.polymarket.com")
749 .unwrap();
750
751 assert_eq!(
752 sibling.base_url.host_str(),
753 Some("user-pnl-api.polymarket.com")
754 );
755 assert_eq!(client.base_url.host_str(), Some("data-api.polymarket.com"));
757 assert!(sibling.concurrency_limiter.is_some());
760 }
761
762 #[test]
763 fn with_base_url_rejects_a_malformed_url() {
764 let client = HttpClientBuilder::new("https://data-api.polymarket.com")
765 .build()
766 .unwrap();
767 assert!(client.with_base_url("not-a-url").is_err());
768 }
769
770 #[test]
771 fn base_url_path_prefixes_are_dropped() {
772 let client = HttpClientBuilder::new("http://localhost:8080/proxy")
776 .build()
777 .unwrap();
778 assert_eq!(
779 client.base_url.join("/user-pnl").unwrap().as_str(),
780 "http://localhost:8080/user-pnl",
781 "a path prefix in the base URL is not preserved"
782 );
783
784 let with_slash = client
785 .with_base_url("http://localhost:8080/proxy/")
786 .unwrap();
787 assert_eq!(
788 with_slash.base_url.join("/user-pnl").unwrap().as_str(),
789 "http://localhost:8080/user-pnl",
790 "a trailing slash does not preserve the prefix either"
791 );
792 }
793}