+108 -30

crates/scrobbler/src/scrobbler.rs

··· 19 19 xata::user::User, 20 20 }; 21 21 22 22 + const MAX_SPOTIFY_RETRIES: u32 = 3; 23 23 + const INITIAL_RETRY_DELAY_MS: u64 = 1000; 24 24 + 22 25 fn parse_batch(form: &BTreeMap<String, String>) -> Result<Vec<Scrobble>, Error> { 23 26 let mut result = vec![]; 24 27 let mut index = 0; ··· 203 206 let spotify_token = refresh_token(&spotify_token, &client_id, &client_secret).await?; 204 207 let spotify_client = SpotifyClient::new(&spotify_token.access_token); 205 208 206 206 - let result = spotify_client 207 207 - .search(&format!( 208 208 - r#"track:"{}" artist:"{}""#, 209 209 - scrobble.track, scrobble.artist 210 210 - )) 211 211 - .await?; 209 209 + let result = retry_spotify_call( 210 210 + || async { 211 211 + spotify_client 212 212 + .search(&format!( 213 213 + r#"track:"{}" artist:"{}""#, 214 214 + scrobble.track, scrobble.artist 215 215 + )) 216 216 + .await 217 217 + }, 218 218 + "search", 219 219 + ) 220 220 + .await?; 212 221 213 222 if let Some(track) = result.tracks.items.first() { 214 223 tracing::info!(artist = %scrobble.artist, track = %scrobble.track, "Spotify (track)"); 215 224 scrobble.album = Some(track.album.name.clone()); 216 225 let mut track = track.clone(); 217 226 218 218 - if let Some(album) = spotify_client.get_album(&track.album.id).await? { 227 227 + if let Some(album) = retry_spotify_call( 228 228 + || async { spotify_client.get_album(&track.album.id).await }, 229 229 + "get_album", 230 230 + ) 231 231 + .await? 232 232 + { 219 233 track.album = album; 220 234 } 221 235 222 222 - if let Some(artist) = spotify_client 223 223 - .get_artist(&track.album.artists[0].id) 224 224 - .await? 236 236 + if let Some(artist) = retry_spotify_call( 237 237 + || async { spotify_client.get_artist(&track.album.artists[0].id).await }, 238 238 + "get_artist", 239 239 + ) 240 240 + .await? 225 241 { 226 242 track.album.artists[0] = artist; 227 243 } ··· 383 399 let spotify_token = refresh_token(&spotify_token, &client_id, &client_secret).await?; 384 400 let spotify_client = SpotifyClient::new(&spotify_token.access_token); 385 401 386 386 - let result = spotify_client 387 387 - .search(&format!( 388 388 - r#"track:"{}" artist:"{}""#, 389 389 - scrobble.track, scrobble.artist 390 390 - )) 391 391 - .await?; 402 402 + let result = retry_spotify_call( 403 403 + || async { 404 404 + spotify_client 405 405 + .search(&format!( 406 406 + r#"track:"{}" artist:"{}""#, 407 407 + scrobble.track, scrobble.artist 408 408 + )) 409 409 + .await 410 410 + }, 411 411 + "search", 412 412 + ) 413 413 + .await?; 392 414 393 415 if let Some(track) = result.tracks.items.first() { 394 416 let normalize = |s: &str| -> String { ··· 434 456 scrobble.album = Some(track.album.name.clone()); 435 457 let mut track = track.clone(); 436 458 437 437 - if let Some(album) = spotify_client.get_album(&track.album.id).await? { 459 459 + if let Some(album) = retry_spotify_call( 460 460 + || async { spotify_client.get_album(&track.album.id).await }, 461 461 + "get_album", 462 462 + ) 463 463 + .await? 464 464 + { 438 465 track.album = album; 439 466 } 440 467 441 441 - if let Some(artist) = spotify_client 442 442 - .get_artist(&track.album.artists[0].id) 443 443 - .await? 468 468 + if let Some(artist) = retry_spotify_call( 469 469 + || async { spotify_client.get_artist(&track.album.artists[0].id).await }, 470 470 + "get_artist", 471 471 + ) 472 472 + .await? 444 473 { 445 474 track.album.artists[0] = artist; 446 475 } ··· 668 697 let spotify_token = refresh_token(&spotify_token, &client_id, &client_secret).await?; 669 698 let spotify_client = SpotifyClient::new(&spotify_token.access_token); 670 699 671 671 - let result = spotify_client 672 672 - .search(&format!( 673 673 - r#"track:"{}" artist:"{}""#, 674 674 - scrobble.track, scrobble.artist 675 675 - )) 676 676 - .await?; 700 700 + let result = retry_spotify_call( 701 701 + || async { 702 702 + spotify_client 703 703 + .search(&format!( 704 704 + r#"track:"{}" artist:"{}""#, 705 705 + scrobble.track, scrobble.artist 706 706 + )) 707 707 + .await 708 708 + }, 709 709 + "search", 710 710 + ) 711 711 + .await?; 677 712 678 713 if let Some(track) = result.tracks.items.first() { 679 714 let normalize = |s: &str| -> String { ··· 712 747 scrobble.album = Some(track.album.name.clone()); 713 748 let mut track = track.clone(); 714 749 715 715 - if let Some(album) = spotify_client.get_album(&track.album.id).await? { 750 750 + if let Some(album) = retry_spotify_call( 751 751 + || async { spotify_client.get_album(&track.album.id).await }, 752 752 + "get_album", 753 753 + ) 754 754 + .await? 755 755 + { 716 756 track.album = album; 717 757 } 718 758 719 719 - if let Some(artist) = spotify_client 720 720 - .get_artist(&track.album.artists[0].id) 721 721 - .await? 759 759 + if let Some(artist) = retry_spotify_call( 760 760 + || async { spotify_client.get_artist(&track.album.artists[0].id).await }, 761 761 + "get_artist", 762 762 + ) 763 763 + .await? 722 764 { 723 765 track.album.artists[0] = artist; 724 766 } ··· 785 827 786 828 Ok(None) 787 829 } 830 830 + 831 831 + async fn retry_spotify_call<F, Fut, T>(mut f: F, operation: &str) -> Result<T, Error> 832 832 + where 833 833 + F: FnMut() -> Fut, 834 834 + Fut: std::future::Future<Output = Result<T, Error>>, 835 835 + { 836 836 + let mut last_error = None; 837 837 + 838 838 + for attempt in 0..MAX_SPOTIFY_RETRIES { 839 839 + match f().await { 840 840 + Ok(result) => return Ok(result), 841 841 + Err(e) => { 842 842 + let is_timeout = e.to_string().contains("timed out") 843 843 + || e.to_string().contains("timeout") 844 844 + || e.to_string().contains("operation timed out"); 845 845 + 846 846 + if is_timeout && attempt < MAX_SPOTIFY_RETRIES - 1 { 847 847 + let delay = INITIAL_RETRY_DELAY_MS * 2_u64.pow(attempt); 848 848 + tracing::warn!( 849 849 + attempt = attempt + 1, 850 850 + max_attempts = MAX_SPOTIFY_RETRIES, 851 851 + delay_ms = delay, 852 852 + operation = operation, 853 853 + "Spotify API timeout, retrying..." 854 854 + ); 855 855 + tokio::time::sleep(tokio::time::Duration::from_millis(delay)).await; 856 856 + last_error = Some(e); 857 857 + } else { 858 858 + return Err(e); 859 859 + } 860 860 + } 861 861 + } 862 862 + } 863 863 + 864 864 + Err(last_error.unwrap_or_else(|| Error::msg("Max retries exceeded"))) 865 865 + }

+23 -6

crates/scrobbler/src/spotify/client.rs

··· 1 1 use super::types::{Album, Artist, SearchResponse}; 2 2 use anyhow::Error; 3 3 + use std::time::Duration; 3 4 4 5 pub const BASE_URL: &str = "https://api.spotify.com/v1"; 5 6 6 7 pub struct SpotifyClient { 7 8 token: String, 9 9 + client: reqwest::Client, 8 10 } 9 11 10 12 impl SpotifyClient { 11 13 pub fn new(token: &str) -> Self { 14 14 + let client = reqwest::Client::builder() 15 15 + .timeout(Duration::from_secs(30)) 16 16 + .connect_timeout(Duration::from_secs(10)) 17 17 + .build() 18 18 + .expect("Failed to build HTTP client"); 19 19 + 12 20 SpotifyClient { 13 21 token: token.to_string(), 22 22 + client, 14 23 } 15 24 } 16 25 17 26 pub async fn search(&self, query: &str) -> Result<SearchResponse, Error> { 18 27 let url = format!("{}/search", BASE_URL); 19 19 - let client = reqwest::Client::new(); 20 20 - let response = client 28 28 + let response = self 29 29 + .client 21 30 .get(&url) 22 31 .bearer_auth(&self.token) 23 32 .query(&[("type", "track"), ("q", query)]) ··· 29 38 30 39 pub async fn get_album(&self, id: &str) -> Result<Option<Album>, Error> { 31 40 let url = format!("{}/albums/{}", BASE_URL, id); 32 32 - let client = reqwest::Client::new(); 33 33 - let response = client.get(&url).bearer_auth(&self.token).send().await?; 41 41 + let response = self 42 42 + .client 43 43 + .get(&url) 44 44 + .bearer_auth(&self.token) 45 45 + .send() 46 46 + .await?; 34 47 35 48 let headers = response.headers().clone(); 36 49 let data = response.text().await?; ··· 45 58 46 59 pub async fn get_artist(&self, id: &str) -> Result<Option<Artist>, Error> { 47 60 let url = format!("{}/artists/{}", BASE_URL, id); 48 48 - let client = reqwest::Client::new(); 49 49 - let response = client.get(&url).bearer_auth(&self.token).send().await?; 61 61 + let response = self 62 62 + .client 63 63 + .get(&url) 64 64 + .bearer_auth(&self.token) 65 65 + .send() 66 66 + .await?; 50 67 51 68 let headers = response.headers().clone(); 52 69 let data = response.text().await?;

+2

crates/webscrobbler/src/musicbrainz/client.rs

··· 41 41 let redis = client.get_multiplexed_tokio_connection().await?; 42 42 let http = reqwest::Client::builder() 43 43 .user_agent(USER_AGENT) 44 44 + .timeout(Duration::from_secs(30)) 45 45 + .connect_timeout(Duration::from_secs(10)) 44 46 .build() 45 47 .context("build http client")?; 46 48 let me = MusicbrainzClient {

+52 -5

crates/webscrobbler/src/scrobbler.rs

··· 14 14 use rand::Rng; 15 15 use sqlx::{Pool, Postgres}; 16 16 17 17 + const MAX_SPOTIFY_RETRIES: u32 = 3; 18 18 + const INITIAL_RETRY_DELAY_MS: u64 = 1000; 19 19 + 17 20 pub async fn scrobble( 18 21 pool: &Pool<Postgres>, 19 22 cache: &Cache, ··· 135 138 136 139 tracing::info!(query = %query, "Searching on Spotify"); 137 140 138 138 - let result = spotify_client.search(&query).await?; 141 141 + let result = 142 142 + retry_spotify_call(|| async { spotify_client.search(&query).await }, "search").await?; 139 143 140 144 tracing::info!(total = %result.tracks.total, "Spotify search results"); 141 145 ··· 186 190 tracing::info!("Spotify (track)"); 187 191 let mut track = track.clone(); 188 192 189 189 - if let Some(album) = spotify_client.get_album(&track.album.id).await? { 193 193 + if let Some(album) = retry_spotify_call( 194 194 + || async { spotify_client.get_album(&track.album.id).await }, 195 195 + "get_album", 196 196 + ) 197 197 + .await? 198 198 + { 190 199 track.album = album; 191 200 } 192 201 193 193 - if let Some(artist) = spotify_client 194 194 - .get_artist(&track.album.artists[0].id) 195 195 - .await? 202 202 + if let Some(artist) = retry_spotify_call( 203 203 + || async { spotify_client.get_artist(&track.album.artists[0].id).await }, 204 204 + "get_artist", 205 205 + ) 206 206 + .await? 196 207 { 197 208 track.album.artists[0] = artist; 198 209 } ··· 260 271 261 272 Ok(None) 262 273 } 274 274 + 275 275 + async fn retry_spotify_call<F, Fut, T>(mut f: F, operation: &str) -> Result<T, Error> 276 276 + where 277 277 + F: FnMut() -> Fut, 278 278 + Fut: std::future::Future<Output = Result<T, Error>>, 279 279 + { 280 280 + let mut last_error = None; 281 281 + 282 282 + for attempt in 0..MAX_SPOTIFY_RETRIES { 283 283 + match f().await { 284 284 + Ok(result) => return Ok(result), 285 285 + Err(e) => { 286 286 + let is_timeout = e.to_string().contains("timed out") 287 287 + || e.to_string().contains("timeout") 288 288 + || e.to_string().contains("operation timed out"); 289 289 + 290 290 + if is_timeout && attempt < MAX_SPOTIFY_RETRIES - 1 { 291 291 + let delay = INITIAL_RETRY_DELAY_MS * 2_u64.pow(attempt); 292 292 + tracing::warn!( 293 293 + attempt = attempt + 1, 294 294 + max_attempts = MAX_SPOTIFY_RETRIES, 295 295 + delay_ms = delay, 296 296 + operation = operation, 297 297 + "Spotify API timeout, retrying..." 298 298 + ); 299 299 + tokio::time::sleep(tokio::time::Duration::from_millis(delay)).await; 300 300 + last_error = Some(e); 301 301 + } else { 302 302 + return Err(e); 303 303 + } 304 304 + } 305 305 + } 306 306 + } 307 307 + 308 308 + Err(last_error.unwrap_or_else(|| Error::msg("Max retries exceeded"))) 309 309 + }

+23 -6

crates/webscrobbler/src/spotify/client.rs

··· 1 1 use super::types::{Album, Artist, SearchResponse}; 2 2 use anyhow::Error; 3 3 + use std::time::Duration; 3 4 4 5 pub const BASE_URL: &str = "https://api.spotify.com/v1"; 5 6 6 7 pub struct SpotifyClient { 7 8 token: String, 9 9 + client: reqwest::Client, 8 10 } 9 11 10 12 impl SpotifyClient { 11 13 pub fn new(token: &str) -> Self { 14 14 + let client = reqwest::Client::builder() 15 15 + .timeout(Duration::from_secs(30)) 16 16 + .connect_timeout(Duration::from_secs(10)) 17 17 + .build() 18 18 + .expect("Failed to build HTTP client"); 19 19 + 12 20 SpotifyClient { 13 21 token: token.to_string(), 22 22 + client, 14 23 } 15 24 } 16 25 17 26 pub async fn search(&self, query: &str) -> Result<SearchResponse, Error> { 18 27 let url = format!("{}/search", BASE_URL); 19 19 - let client = reqwest::Client::new(); 20 20 - let response = client 28 28 + let response = self 29 29 + .client 21 30 .get(&url) 22 31 .bearer_auth(&self.token) 23 32 .query(&[("type", "track"), ("q", query)]) ··· 29 38 30 39 pub async fn get_album(&self, id: &str) -> Result<Option<Album>, Error> { 31 40 let url = format!("{}/albums/{}", BASE_URL, id); 32 32 - let client = reqwest::Client::new(); 33 33 - let response = client.get(&url).bearer_auth(&self.token).send().await?; 41 41 + let response = self 42 42 + .client 43 43 + .get(&url) 44 44 + .bearer_auth(&self.token) 45 45 + .send() 46 46 + .await?; 34 47 35 48 let headers = response.headers().clone(); 36 49 let data = response.text().await?; ··· 45 58 46 59 pub async fn get_artist(&self, id: &str) -> Result<Option<Artist>, Error> { 47 60 let url = format!("{}/artists/{}", BASE_URL, id); 48 48 - let client = reqwest::Client::new(); 49 49 - let response = client.get(&url).bearer_auth(&self.token).send().await?; 61 61 + let response = self 62 62 + .client 63 63 + .get(&url) 64 64 + .bearer_auth(&self.token) 65 65 + .send() 66 66 + .await?; 50 67 51 68 let headers = response.headers().clone(); 52 69 let data = response.text().await?;