PNRxA commited on
Commit
65c3047
·
1 Parent(s): 5d84fe3

add service request retries

Browse files
api/src/lib.rs CHANGED
@@ -39,7 +39,7 @@ pub struct AppState {
39
  pub tmdb: TmdbClient,
40
  pub omdb: Option<OmdbClient>,
41
  pub mdblist: Option<MdblistClient>,
42
- pub http: reqwest::Client,
43
  pub font: FontArc,
44
  pub refresh_locks: moka::sync::Cache<String, ()>,
45
  pub db: DatabaseConnection,
 
39
  pub tmdb: TmdbClient,
40
  pub omdb: Option<OmdbClient>,
41
  pub mdblist: Option<MdblistClient>,
42
+
43
  pub font: FontArc,
44
  pub refresh_locks: moka::sync::Cache<String, ()>,
45
  pub db: DatabaseConnection,
api/src/main.rs CHANGED
@@ -33,7 +33,11 @@ async fn main() {
33
  .init();
34
 
35
  let config = Config::from_env();
36
- let http = reqwest::Client::new();
 
 
 
 
37
  let font = FontArc::try_from_slice(FONT_BYTES).expect("failed to load font");
38
 
39
  let omdb = config
@@ -204,10 +208,10 @@ async fn main() {
204
  let pending_last_used: Arc<DashMap<i32, ()>> = Arc::new(DashMap::new());
205
 
206
  let state = Arc::new(AppState {
207
- tmdb: TmdbClient::new(config.tmdb_api_key.clone(), http.clone()),
208
  omdb,
209
  mdblist,
210
- http,
211
  font,
212
  refresh_locks,
213
  db: database,
 
33
  .init();
34
 
35
  let config = Config::from_env();
36
+ let http = reqwest::Client::builder()
37
+ .timeout(Duration::from_secs(30))
38
+ .connect_timeout(Duration::from_secs(10))
39
+ .build()
40
+ .expect("failed to build HTTP client");
41
  let font = FontArc::try_from_slice(FONT_BYTES).expect("failed to load font");
42
 
43
  let omdb = config
 
208
  let pending_last_used: Arc<DashMap<i32, ()>> = Arc::new(DashMap::new());
209
 
210
  let state = Arc::new(AppState {
211
+ tmdb: TmdbClient::new(config.tmdb_api_key.clone(), http),
212
  omdb,
213
  mdblist,
214
+
215
  font,
216
  refresh_locks,
217
  db: database,
api/src/poster/generate.rs CHANGED
@@ -16,7 +16,7 @@ pub struct PosterParams<'a> {
16
  pub poster_path: &'a str,
17
  pub badges: &'a [RatingBadge],
18
  pub tmdb: &'a TmdbClient,
19
- pub http: &'a reqwest::Client,
20
  pub font: &'a FontArc,
21
  pub quality: u8,
22
  pub cache_dir: &'a str,
@@ -31,7 +31,6 @@ pub async fn generate_poster(params: PosterParams<'_>) -> Result<Vec<u8>, AppErr
31
  poster_path,
32
  badges,
33
  tmdb,
34
- http,
35
  font,
36
  quality,
37
  cache_dir,
@@ -47,14 +46,14 @@ pub async fn generate_poster(params: PosterParams<'_>) -> Result<Vec<u8>, AppErr
47
  let poster_cache = cache::base_poster_path(cache_dir, poster_path)?;
48
  if let Some(entry) = cache::read(&poster_cache, poster_stale_secs).await {
49
  if entry.is_stale {
50
- let bytes = tmdb.fetch_poster_bytes(poster_path, http).await?;
51
  cache::write(&poster_cache, &bytes).await?;
52
  bytes
53
  } else {
54
  entry.bytes
55
  }
56
  } else {
57
- let bytes = tmdb.fetch_poster_bytes(poster_path, http).await?;
58
  cache::write(&poster_cache, &bytes).await?;
59
  bytes
60
  }
 
16
  pub poster_path: &'a str,
17
  pub badges: &'a [RatingBadge],
18
  pub tmdb: &'a TmdbClient,
19
+
20
  pub font: &'a FontArc,
21
  pub quality: u8,
22
  pub cache_dir: &'a str,
 
31
  poster_path,
32
  badges,
33
  tmdb,
 
34
  font,
35
  quality,
36
  cache_dir,
 
46
  let poster_cache = cache::base_poster_path(cache_dir, poster_path)?;
47
  if let Some(entry) = cache::read(&poster_cache, poster_stale_secs).await {
48
  if entry.is_stale {
49
+ let bytes = tmdb.fetch_poster_bytes(poster_path).await?;
50
  cache::write(&poster_cache, &bytes).await?;
51
  bytes
52
  } else {
53
  entry.bytes
54
  }
55
  } else {
56
+ let bytes = tmdb.fetch_poster_bytes(poster_path).await?;
57
  cache::write(&poster_cache, &bytes).await?;
58
  bytes
59
  }
api/src/poster/serve.rs CHANGED
@@ -565,7 +565,6 @@ async fn generate_poster_with_source(
565
  poster_path,
566
  badges: &badges,
567
  tmdb: &state.tmdb,
568
- http: &state.http,
569
  font: &state.font,
570
  quality: state.config.poster_quality,
571
  cache_dir: &state.config.cache_dir,
 
565
  poster_path,
566
  badges: &badges,
567
  tmdb: &state.tmdb,
 
568
  font: &state.font,
569
  quality: state.config.poster_quality,
570
  cache_dir: &state.config.cache_dir,
api/src/services/fanart.rs CHANGED
@@ -1,4 +1,5 @@
1
  use crate::error::AppError;
 
2
  use serde::Deserialize;
3
 
4
  #[derive(Clone)]
@@ -60,7 +61,7 @@ impl FanartClient {
60
  "https://webservice.fanart.tv/v3/movies/{tmdb_id}?api_key={}",
61
  self.api_key
62
  );
63
- let resp = self.http.get(&url).send().await?;
64
  if resp.status() == reqwest::StatusCode::NOT_FOUND {
65
  return Ok(FanartImages { posters: vec![], logos: vec![], backdrops: vec![] });
66
  }
@@ -79,7 +80,7 @@ impl FanartClient {
79
  "https://webservice.fanart.tv/v3/tv/{id}?api_key={}",
80
  self.api_key
81
  );
82
- let resp = self.http.get(&url).send().await?;
83
  if resp.status() == reqwest::StatusCode::NOT_FOUND {
84
  return Ok(FanartImages { posters: vec![], logos: vec![], backdrops: vec![] });
85
  }
@@ -126,7 +127,10 @@ impl FanartClient {
126
  }
127
 
128
  pub async fn fetch_poster_bytes(&self, url: &str) -> Result<Vec<u8>, AppError> {
129
- let resp = self.http.get(url).send().await?.error_for_status()?;
 
 
 
130
  Ok(resp.bytes().await?.to_vec())
131
  }
132
  }
 
1
  use crate::error::AppError;
2
+ use crate::services::retry::{self, FANART_RETRY};
3
  use serde::Deserialize;
4
 
5
  #[derive(Clone)]
 
61
  "https://webservice.fanart.tv/v3/movies/{tmdb_id}?api_key={}",
62
  self.api_key
63
  );
64
+ let resp = retry::send_with_retry(&FANART_RETRY, || self.http.get(&url).send()).await?;
65
  if resp.status() == reqwest::StatusCode::NOT_FOUND {
66
  return Ok(FanartImages { posters: vec![], logos: vec![], backdrops: vec![] });
67
  }
 
80
  "https://webservice.fanart.tv/v3/tv/{id}?api_key={}",
81
  self.api_key
82
  );
83
+ let resp = retry::send_with_retry(&FANART_RETRY, || self.http.get(&url).send()).await?;
84
  if resp.status() == reqwest::StatusCode::NOT_FOUND {
85
  return Ok(FanartImages { posters: vec![], logos: vec![], backdrops: vec![] });
86
  }
 
127
  }
128
 
129
  pub async fn fetch_poster_bytes(&self, url: &str) -> Result<Vec<u8>, AppError> {
130
+ let url = url.to_owned();
131
+ let resp = retry::send_with_retry(&FANART_RETRY, || self.http.get(&url).send())
132
+ .await?
133
+ .error_for_status()?;
134
  Ok(resp.bytes().await?.to_vec())
135
  }
136
  }
api/src/services/mdblist.rs CHANGED
@@ -1,5 +1,6 @@
1
  use crate::error::AppError;
2
  use crate::id::MediaType;
 
3
  use serde::Deserialize;
4
 
5
  #[derive(Clone)]
@@ -39,13 +40,14 @@ impl MdblistClient {
39
 
40
  let url = format!("https://api.mdblist.com/imdb/{kind}/{imdb_id}");
41
 
42
- let resp = self
43
- .http
44
- .get(&url)
45
- .query(&[("apikey", &self.api_key)])
46
- .send()
47
- .await?
48
- .error_for_status()?;
 
49
 
50
  Ok(resp.json().await?)
51
  }
 
1
  use crate::error::AppError;
2
  use crate::id::MediaType;
3
+ use crate::services::retry::{self, MDBLIST_RETRY};
4
  use serde::Deserialize;
5
 
6
  #[derive(Clone)]
 
40
 
41
  let url = format!("https://api.mdblist.com/imdb/{kind}/{imdb_id}");
42
 
43
+ let resp = retry::send_with_retry(&MDBLIST_RETRY, || {
44
+ self.http
45
+ .get(&url)
46
+ .query(&[("apikey", &self.api_key)])
47
+ .send()
48
+ })
49
+ .await?
50
+ .error_for_status()?;
51
 
52
  Ok(resp.json().await?)
53
  }
api/src/services/mod.rs CHANGED
@@ -3,5 +3,6 @@ pub mod fanart;
3
  pub mod mdblist;
4
  pub mod omdb;
5
  pub mod ratings;
 
6
  pub mod tmdb;
7
  pub mod validation;
 
3
  pub mod mdblist;
4
  pub mod omdb;
5
  pub mod ratings;
6
+ pub mod retry;
7
  pub mod tmdb;
8
  pub mod validation;
api/src/services/omdb.rs CHANGED
@@ -1,4 +1,5 @@
1
  use crate::error::AppError;
 
2
  use serde::Deserialize;
3
 
4
  #[derive(Clone)]
@@ -31,13 +32,15 @@ impl OmdbClient {
31
  }
32
 
33
  pub async fn get_ratings(&self, imdb_id: &str) -> Result<OmdbResponse, AppError> {
34
- let resp = self
35
- .http
36
- .get("https://www.omdbapi.com/")
37
- .query(&[("apikey", &self.api_key), ("i", &imdb_id.to_string())])
38
- .send()
39
- .await?
40
- .error_for_status()?;
 
 
41
  Ok(resp.json().await?)
42
  }
43
  }
 
1
  use crate::error::AppError;
2
+ use crate::services::retry::{self, OMDB_RETRY};
3
  use serde::Deserialize;
4
 
5
  #[derive(Clone)]
 
32
  }
33
 
34
  pub async fn get_ratings(&self, imdb_id: &str) -> Result<OmdbResponse, AppError> {
35
+ let imdb_id = imdb_id.to_owned();
36
+ let resp = retry::send_with_retry(&OMDB_RETRY, || {
37
+ self.http
38
+ .get("https://www.omdbapi.com/")
39
+ .query(&[("apikey", &self.api_key), ("i", &imdb_id)])
40
+ .send()
41
+ })
42
+ .await?
43
+ .error_for_status()?;
44
  Ok(resp.json().await?)
45
  }
46
  }
api/src/services/retry.rs ADDED
@@ -0,0 +1,447 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ use std::future::Future;
2
+ use std::time::Duration;
3
+
4
+ use rand::Rng;
5
+ use reqwest::Response;
6
+
7
+ use crate::error::AppError;
8
+
9
+ pub struct RetryConfig {
10
+ pub max_retries: u32,
11
+ pub base_delay: Duration,
12
+ pub max_delay: Duration,
13
+ pub service_name: &'static str,
14
+ }
15
+
16
+ pub const TMDB_API_RETRY: RetryConfig = RetryConfig {
17
+ max_retries: 2,
18
+ base_delay: Duration::from_millis(500),
19
+ max_delay: Duration::from_secs(4),
20
+ service_name: "tmdb",
21
+ };
22
+
23
+ pub const TMDB_CDN_RETRY: RetryConfig = RetryConfig {
24
+ max_retries: 2,
25
+ base_delay: Duration::from_secs(1),
26
+ max_delay: Duration::from_secs(8),
27
+ service_name: "tmdb-cdn",
28
+ };
29
+
30
+ pub const FANART_RETRY: RetryConfig = RetryConfig {
31
+ max_retries: 3,
32
+ base_delay: Duration::from_secs(1),
33
+ max_delay: Duration::from_secs(8),
34
+ service_name: "fanart",
35
+ };
36
+
37
+ pub const OMDB_RETRY: RetryConfig = RetryConfig {
38
+ max_retries: 1,
39
+ base_delay: Duration::from_secs(2),
40
+ max_delay: Duration::from_secs(2),
41
+ service_name: "omdb",
42
+ };
43
+
44
+ pub const MDBLIST_RETRY: RetryConfig = RetryConfig {
45
+ max_retries: 1,
46
+ base_delay: Duration::from_secs(2),
47
+ max_delay: Duration::from_secs(2),
48
+ service_name: "mdblist",
49
+ };
50
+
51
+ /// Send an HTTP request with retry logic for transient failures.
52
+ ///
53
+ /// The closure must rebuild and send the request each time (since `RequestBuilder`
54
+ /// is not `Clone`). Returns the raw `Response` so callers keep their existing
55
+ /// status-handling logic.
56
+ pub async fn send_with_retry<F, Fut>(config: &RetryConfig, request_fn: F) -> Result<Response, AppError>
57
+ where
58
+ F: Fn() -> Fut,
59
+ Fut: Future<Output = Result<Response, reqwest::Error>>,
60
+ {
61
+ for attempt in 0..=config.max_retries {
62
+ let result = request_fn().await;
63
+
64
+ match result {
65
+ Ok(resp) => {
66
+ let status = resp.status();
67
+
68
+ if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
69
+ if attempt == config.max_retries {
70
+ tracing::warn!(
71
+ service = config.service_name,
72
+ "request failed after {} retries: 429 Too Many Requests",
73
+ config.max_retries,
74
+ );
75
+ return Ok(resp);
76
+ }
77
+
78
+ let delay = retry_after_delay(&resp, config, attempt);
79
+ tracing::warn!(
80
+ service = config.service_name,
81
+ "429 Too Many Requests, retrying (attempt {}/{}, delay {}ms)",
82
+ attempt + 1,
83
+ config.max_retries,
84
+ delay.as_millis(),
85
+ );
86
+ tokio::time::sleep(delay).await;
87
+ continue;
88
+ }
89
+
90
+ if status.is_server_error() {
91
+ if attempt == config.max_retries {
92
+ tracing::warn!(
93
+ service = config.service_name,
94
+ "request failed after {} retries: {status}",
95
+ config.max_retries,
96
+ );
97
+ return Ok(resp);
98
+ }
99
+
100
+ let delay = backoff_delay(config, attempt);
101
+ tracing::warn!(
102
+ service = config.service_name,
103
+ "{status}, retrying (attempt {}/{}, delay {}ms)",
104
+ attempt + 1,
105
+ config.max_retries,
106
+ delay.as_millis(),
107
+ );
108
+ tokio::time::sleep(delay).await;
109
+ continue;
110
+ }
111
+
112
+ // 2xx, 3xx, or non-retryable 4xx — return immediately
113
+ return Ok(resp);
114
+ }
115
+ Err(e) => {
116
+ if attempt == config.max_retries {
117
+ tracing::warn!(
118
+ service = config.service_name,
119
+ "request failed after {} retries: {e}",
120
+ config.max_retries,
121
+ );
122
+ return Err(AppError::Api(e));
123
+ }
124
+
125
+ let delay = backoff_delay(config, attempt);
126
+ tracing::warn!(
127
+ service = config.service_name,
128
+ "connection error, retrying (attempt {}/{}, delay {}ms): {e}",
129
+ attempt + 1,
130
+ config.max_retries,
131
+ delay.as_millis(),
132
+ );
133
+ tokio::time::sleep(delay).await;
134
+ }
135
+ }
136
+ }
137
+
138
+ // All paths return inside the loop; this is unreachable but satisfies the compiler.
139
+ unreachable!("retry loop always returns within max_retries iterations")
140
+ }
141
+
142
+ /// Parse `Retry-After` header as seconds, falling back to exponential backoff.
143
+ fn retry_after_delay(resp: &Response, config: &RetryConfig, attempt: u32) -> Duration {
144
+ if let Some(val) = resp.headers().get(reqwest::header::RETRY_AFTER) {
145
+ if let Ok(s) = val.to_str() {
146
+ if let Ok(secs) = s.trim().parse::<u64>() {
147
+ let capped = Duration::from_secs(secs).min(config.max_delay);
148
+ return add_jitter(capped);
149
+ }
150
+ }
151
+ }
152
+ backoff_delay(config, attempt)
153
+ }
154
+
155
+ /// Exponential backoff: `min(base_delay * 2^attempt, max_delay)` + jitter.
156
+ fn backoff_delay(config: &RetryConfig, attempt: u32) -> Duration {
157
+ let delay = config
158
+ .base_delay
159
+ .saturating_mul(1 << attempt)
160
+ .min(config.max_delay);
161
+ add_jitter(delay)
162
+ }
163
+
164
+ /// Add small random jitter (0–25% of delay).
165
+ fn add_jitter(delay: Duration) -> Duration {
166
+ let jitter_ms = rand::rng().random_range(0..=delay.as_millis() as u64 / 4);
167
+ delay + Duration::from_millis(jitter_ms)
168
+ }
169
+
170
+ #[cfg(test)]
171
+ mod tests {
172
+ use super::*;
173
+ use std::sync::atomic::{AtomicU32, Ordering};
174
+ use std::sync::Arc;
175
+
176
+ use axum::response::IntoResponse;
177
+
178
+ /// Tiny test server that returns a sequence of status codes, then 200.
179
+ /// Returns the server URL and a request counter.
180
+ async fn mock_server(statuses: Vec<u16>) -> (String, Arc<AtomicU32>) {
181
+ let counter = Arc::new(AtomicU32::new(0));
182
+ let counter_clone = counter.clone();
183
+ let statuses = Arc::new(statuses);
184
+
185
+ let app = axum::Router::new().route(
186
+ "/",
187
+ axum::routing::get(move || {
188
+ let counter = counter_clone.clone();
189
+ let statuses = statuses.clone();
190
+ async move {
191
+ let n = counter.fetch_add(1, Ordering::SeqCst) as usize;
192
+ let code = statuses.get(n).copied().unwrap_or(200);
193
+ axum::http::StatusCode::from_u16(code).unwrap_or(axum::http::StatusCode::OK)
194
+ }
195
+ }),
196
+ );
197
+
198
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
199
+ let addr = listener.local_addr().unwrap();
200
+ tokio::spawn(axum::serve(listener, app).into_future());
201
+ (format!("http://{addr}"), counter)
202
+ }
203
+
204
+ /// Server that returns a specific header along with the status code sequence.
205
+ async fn mock_server_with_headers(
206
+ statuses: Vec<(u16, Vec<(&'static str, &'static str)>)>,
207
+ ) -> (String, Arc<AtomicU32>) {
208
+ let counter = Arc::new(AtomicU32::new(0));
209
+ let counter_clone = counter.clone();
210
+ let statuses: Arc<Vec<(u16, Vec<(&'static str, &'static str)>)>> = Arc::new(statuses);
211
+
212
+ let app = axum::Router::new().route(
213
+ "/",
214
+ axum::routing::get(move || {
215
+ let counter = counter_clone.clone();
216
+ let statuses = statuses.clone();
217
+ async move {
218
+ let n = counter.fetch_add(1, Ordering::SeqCst) as usize;
219
+ let (code, headers) = statuses
220
+ .get(n)
221
+ .cloned()
222
+ .unwrap_or((200, vec![]));
223
+ let status = axum::http::StatusCode::from_u16(code)
224
+ .unwrap_or(axum::http::StatusCode::OK);
225
+ let mut resp = status.into_response();
226
+ for (k, v) in headers {
227
+ resp.headers_mut().insert(
228
+ axum::http::HeaderName::from_static(k),
229
+ axum::http::HeaderValue::from_static(v),
230
+ );
231
+ }
232
+ resp
233
+ }
234
+ }),
235
+ );
236
+
237
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
238
+ let addr = listener.local_addr().unwrap();
239
+ tokio::spawn(axum::serve(listener, app).into_future());
240
+ (format!("http://{addr}"), counter)
241
+ }
242
+
243
+ const FAST: RetryConfig = RetryConfig {
244
+ max_retries: 2,
245
+ base_delay: Duration::from_millis(1),
246
+ max_delay: Duration::from_millis(10),
247
+ service_name: "test",
248
+ };
249
+
250
+ #[test]
251
+ fn backoff_respects_max_delay() {
252
+ let config = RetryConfig {
253
+ max_retries: 5,
254
+ base_delay: Duration::from_secs(1),
255
+ max_delay: Duration::from_secs(4),
256
+ service_name: "test",
257
+ };
258
+ let delay = backoff_delay(&config, 3);
259
+ assert!(delay <= Duration::from_secs(5));
260
+ assert!(delay >= Duration::from_secs(4));
261
+ }
262
+
263
+ #[test]
264
+ fn backoff_first_attempt() {
265
+ let config = RetryConfig {
266
+ max_retries: 2,
267
+ base_delay: Duration::from_millis(500),
268
+ max_delay: Duration::from_secs(4),
269
+ service_name: "test",
270
+ };
271
+ let delay = backoff_delay(&config, 0);
272
+ assert!(delay >= Duration::from_millis(500));
273
+ assert!(delay <= Duration::from_millis(625));
274
+ }
275
+
276
+ #[tokio::test]
277
+ async fn success_on_first_try() {
278
+ let (url, counter) = mock_server(vec![200]).await;
279
+ let client = reqwest::Client::new();
280
+ let resp = send_with_retry(&FAST, || client.get(&url).send()).await.unwrap();
281
+ assert_eq!(resp.status(), 200);
282
+ assert_eq!(counter.load(Ordering::SeqCst), 1);
283
+ }
284
+
285
+ #[tokio::test]
286
+ async fn retry_429_then_success() {
287
+ let (url, counter) = mock_server(vec![429, 200]).await;
288
+ let client = reqwest::Client::new();
289
+ let resp = send_with_retry(&FAST, || client.get(&url).send()).await.unwrap();
290
+ assert_eq!(resp.status(), 200);
291
+ assert_eq!(counter.load(Ordering::SeqCst), 2);
292
+ }
293
+
294
+ #[tokio::test]
295
+ async fn retry_429_exhausted_returns_429() {
296
+ let (url, counter) = mock_server(vec![429, 429, 429]).await;
297
+ let client = reqwest::Client::new();
298
+ let resp = send_with_retry(&FAST, || client.get(&url).send()).await.unwrap();
299
+ assert_eq!(resp.status(), 429);
300
+ assert_eq!(counter.load(Ordering::SeqCst), 3); // initial + 2 retries
301
+ }
302
+
303
+ #[tokio::test]
304
+ async fn retry_500_then_success() {
305
+ let (url, counter) = mock_server(vec![500, 200]).await;
306
+ let client = reqwest::Client::new();
307
+ let resp = send_with_retry(&FAST, || client.get(&url).send()).await.unwrap();
308
+ assert_eq!(resp.status(), 200);
309
+ assert_eq!(counter.load(Ordering::SeqCst), 2);
310
+ }
311
+
312
+ #[tokio::test]
313
+ async fn retry_502_then_success() {
314
+ let (url, counter) = mock_server(vec![502, 200]).await;
315
+ let client = reqwest::Client::new();
316
+ let resp = send_with_retry(&FAST, || client.get(&url).send()).await.unwrap();
317
+ assert_eq!(resp.status(), 200);
318
+ assert_eq!(counter.load(Ordering::SeqCst), 2);
319
+ }
320
+
321
+ #[tokio::test]
322
+ async fn retry_503_then_success() {
323
+ let (url, counter) = mock_server(vec![503, 200]).await;
324
+ let client = reqwest::Client::new();
325
+ let resp = send_with_retry(&FAST, || client.get(&url).send()).await.unwrap();
326
+ assert_eq!(resp.status(), 200);
327
+ assert_eq!(counter.load(Ordering::SeqCst), 2);
328
+ }
329
+
330
+ #[tokio::test]
331
+ async fn no_retry_on_404() {
332
+ let (url, counter) = mock_server(vec![404]).await;
333
+ let client = reqwest::Client::new();
334
+ let resp = send_with_retry(&FAST, || client.get(&url).send()).await.unwrap();
335
+ assert_eq!(resp.status(), 404);
336
+ assert_eq!(counter.load(Ordering::SeqCst), 1);
337
+ }
338
+
339
+ #[tokio::test]
340
+ async fn no_retry_on_401() {
341
+ let (url, counter) = mock_server(vec![401]).await;
342
+ let client = reqwest::Client::new();
343
+ let resp = send_with_retry(&FAST, || client.get(&url).send()).await.unwrap();
344
+ assert_eq!(resp.status(), 401);
345
+ assert_eq!(counter.load(Ordering::SeqCst), 1);
346
+ }
347
+
348
+ #[tokio::test]
349
+ async fn no_retry_on_403() {
350
+ let (url, counter) = mock_server(vec![403]).await;
351
+ let client = reqwest::Client::new();
352
+ let resp = send_with_retry(&FAST, || client.get(&url).send()).await.unwrap();
353
+ assert_eq!(resp.status(), 403);
354
+ assert_eq!(counter.load(Ordering::SeqCst), 1);
355
+ }
356
+
357
+ #[tokio::test]
358
+ async fn no_retry_on_400() {
359
+ let (url, counter) = mock_server(vec![400]).await;
360
+ let client = reqwest::Client::new();
361
+ let resp = send_with_retry(&FAST, || client.get(&url).send()).await.unwrap();
362
+ assert_eq!(resp.status(), 400);
363
+ assert_eq!(counter.load(Ordering::SeqCst), 1);
364
+ }
365
+
366
+ #[tokio::test]
367
+ async fn connection_error_retries() {
368
+ // Point at a port nothing is listening on
369
+ let client = reqwest::Client::builder()
370
+ .connect_timeout(Duration::from_millis(50))
371
+ .build()
372
+ .unwrap();
373
+ let config = RetryConfig {
374
+ max_retries: 1,
375
+ base_delay: Duration::from_millis(1),
376
+ max_delay: Duration::from_millis(10),
377
+ service_name: "test",
378
+ };
379
+ let result =
380
+ send_with_retry(&config, || client.get("http://127.0.0.1:1").send()).await;
381
+ assert!(result.is_err());
382
+ }
383
+
384
+ #[tokio::test]
385
+ async fn retry_after_header_respected() {
386
+ let (url, counter) = mock_server_with_headers(vec![
387
+ (429, vec![("retry-after", "1")]),
388
+ (200, vec![]),
389
+ ])
390
+ .await;
391
+ let client = reqwest::Client::new();
392
+ let config = RetryConfig {
393
+ max_retries: 1,
394
+ base_delay: Duration::from_millis(1),
395
+ max_delay: Duration::from_secs(5),
396
+ service_name: "test",
397
+ };
398
+ let start = tokio::time::Instant::now();
399
+ let resp = send_with_retry(&config, || client.get(&url).send()).await.unwrap();
400
+ let elapsed = start.elapsed();
401
+ assert_eq!(resp.status(), 200);
402
+ assert_eq!(counter.load(Ordering::SeqCst), 2);
403
+ // Should have waited ~1s (the Retry-After value), not the 1ms base_delay
404
+ assert!(elapsed >= Duration::from_millis(900));
405
+ }
406
+
407
+ #[tokio::test]
408
+ async fn retry_after_capped_to_max_delay() {
409
+ let (url, counter) = mock_server_with_headers(vec![
410
+ (429, vec![("retry-after", "60")]),
411
+ (200, vec![]),
412
+ ])
413
+ .await;
414
+ let client = reqwest::Client::new();
415
+ let config = RetryConfig {
416
+ max_retries: 1,
417
+ base_delay: Duration::from_millis(1),
418
+ max_delay: Duration::from_millis(50),
419
+ service_name: "test",
420
+ };
421
+ let start = tokio::time::Instant::now();
422
+ let resp = send_with_retry(&config, || client.get(&url).send()).await.unwrap();
423
+ let elapsed = start.elapsed();
424
+ assert_eq!(resp.status(), 200);
425
+ assert_eq!(counter.load(Ordering::SeqCst), 2);
426
+ // 60s Retry-After capped to 50ms max_delay + up to 25% jitter
427
+ assert!(elapsed < Duration::from_millis(200));
428
+ }
429
+
430
+ #[tokio::test]
431
+ async fn mixed_5xx_then_429_then_success() {
432
+ let (url, counter) = mock_server(vec![503, 429, 200]).await;
433
+ let client = reqwest::Client::new();
434
+ let resp = send_with_retry(&FAST, || client.get(&url).send()).await.unwrap();
435
+ assert_eq!(resp.status(), 200);
436
+ assert_eq!(counter.load(Ordering::SeqCst), 3);
437
+ }
438
+
439
+ #[tokio::test]
440
+ async fn server_error_exhausted_returns_last_status() {
441
+ let (url, counter) = mock_server(vec![503, 502, 500]).await;
442
+ let client = reqwest::Client::new();
443
+ let resp = send_with_retry(&FAST, || client.get(&url).send()).await.unwrap();
444
+ assert_eq!(resp.status(), 500);
445
+ assert_eq!(counter.load(Ordering::SeqCst), 3);
446
+ }
447
+ }
api/src/services/tmdb.rs CHANGED
@@ -1,4 +1,5 @@
1
  use crate::error::AppError;
 
2
  use serde::de::DeserializeOwned;
3
 
4
  #[derive(Clone)]
@@ -18,17 +19,23 @@ impl TmdbClient {
18
  params: &[(&str, &str)],
19
  ) -> Result<T, AppError> {
20
  let url = format!("https://api.themoviedb.org/3{path}");
21
- let mut req = self.http.get(&url).query(&[("api_key", &self.api_key)]);
22
- if !params.is_empty() {
23
- req = req.query(params);
24
- }
25
- let resp = req.send().await?.error_for_status()?;
 
 
 
 
26
  Ok(resp.json().await?)
27
  }
28
 
29
- pub async fn fetch_poster_bytes(&self, poster_path: &str, http: &reqwest::Client) -> Result<Vec<u8>, AppError> {
30
  let url = format!("https://image.tmdb.org/t/p/w500{poster_path}");
31
- let resp = http.get(&url).send().await?.error_for_status()?;
 
 
32
  Ok(resp.bytes().await?.to_vec())
33
  }
34
  }
 
1
  use crate::error::AppError;
2
+ use crate::services::retry::{self, TMDB_API_RETRY, TMDB_CDN_RETRY};
3
  use serde::de::DeserializeOwned;
4
 
5
  #[derive(Clone)]
 
19
  params: &[(&str, &str)],
20
  ) -> Result<T, AppError> {
21
  let url = format!("https://api.themoviedb.org/3{path}");
22
+ let resp = retry::send_with_retry(&TMDB_API_RETRY, || {
23
+ let mut req = self.http.get(&url).query(&[("api_key", &self.api_key)]);
24
+ if !params.is_empty() {
25
+ req = req.query(params);
26
+ }
27
+ req.send()
28
+ })
29
+ .await?
30
+ .error_for_status()?;
31
  Ok(resp.json().await?)
32
  }
33
 
34
+ pub async fn fetch_poster_bytes(&self, poster_path: &str) -> Result<Vec<u8>, AppError> {
35
  let url = format!("https://image.tmdb.org/t/p/w500{poster_path}");
36
+ let resp = retry::send_with_retry(&TMDB_CDN_RETRY, || self.http.get(&url).send())
37
+ .await?
38
+ .error_for_status()?;
39
  Ok(resp.bytes().await?.to_vec())
40
  }
41
  }
api/tests/common/mod.rs CHANGED
@@ -109,7 +109,7 @@ async fn _setup_test_app(cors_origin: Option<String>, secure_cookies: bool) -> (
109
  tmdb: TmdbClient::new("test".into(), http.clone()),
110
  omdb: None,
111
  mdblist: None,
112
- http: http.clone(),
113
  font,
114
  refresh_locks,
115
  db,
 
109
  tmdb: TmdbClient::new("test".into(), http.clone()),
110
  omdb: None,
111
  mdblist: None,
112
+
113
  font,
114
  refresh_locks,
115
  db,