diff --git a/pmoqobuz/src/config_ext.rs b/pmoqobuz/src/config_ext.rs index f1535883..fb96f767 100644 --- a/pmoqobuz/src/config_ext.rs +++ b/pmoqobuz/src/config_ext.rs @@ -245,6 +245,15 @@ pub trait QobuzConfigExt { /// Persiste la version du bundle après une extraction réussie. fn set_qobuz_bundle_version(&self, version: &str) -> Result<()>; + + /// Nombre de workers concurrents pour l'enregistrement des tracks en cache. + /// + /// Contrôle le semaphore dans `register_tracks_lazy` : plus la valeur est + /// haute, plus les covers sont téléchargées en parallèle, mais plus la + /// contention sur le mutex SQLite est forte. + /// + /// Défaut : 4 (adapté à une machine sous contrainte mémoire / Docker). + fn get_qobuz_register_concurrency(&self) -> usize; } impl QobuzConfigExt for Config { @@ -504,4 +513,13 @@ impl QobuzConfigExt for Config { Value::String(version.to_string()), ) } + + fn get_qobuz_register_concurrency(&self) -> usize { + match self.get_value(&["accounts", "qobuz", "register_concurrency"]) { + Ok(Value::Number(n)) if n.as_u64().unwrap_or(0) >= 1 => { + n.as_u64().unwrap() as usize + } + _ => 4, + } + } } diff --git a/pmoqobuz/src/source.rs b/pmoqobuz/src/source.rs index 905d70bd..c8c21a40 100644 --- a/pmoqobuz/src/source.rs +++ b/pmoqobuz/src/source.rs @@ -115,6 +115,9 @@ struct QobuzSourceInner { /// Base URL for streaming server (e.g., "http://192.168.0.138:8080") base_url: String, + /// Nombre de workers concurrents pour register_tracks_lazy (configurable) + register_concurrency: usize, + /// Update tracking update_counter: tokio::sync::RwLock, last_change: tokio::sync::RwLock, @@ -142,15 +145,18 @@ impl QobuzSource { /// Returns an error if the caches are not initialized in the registry #[cfg(feature = "server")] pub fn from_registry(client: QobuzClient, base_url: impl Into) -> Result { + use crate::config_ext::QobuzConfigExt; let cache_manager = SourceCacheManager::from_registry("qobuz".to_string())?; let client = Arc::new(client); cache_manager.register_lazy_provider(Arc::new(QobuzLazyProvider::new(client.clone()))); + let register_concurrency = pmoconfig::get_config().get_qobuz_register_concurrency(); Ok(Self { inner: Arc::new(QobuzSourceInner { client, cache_manager, base_url: base_url.into(), + register_concurrency, update_counter: tokio::sync::RwLock::new(0), last_change: tokio::sync::RwLock::new(SystemTime::now()), }), @@ -180,6 +186,7 @@ impl QobuzSource { client, cache_manager, base_url: base_url.into(), + register_concurrency: 4, update_counter: tokio::sync::RwLock::new(0), last_change: tokio::sync::RwLock::new(SystemTime::now()), }), @@ -1062,10 +1069,10 @@ impl QobuzSource { /// Pour chaque track : cache la cover, enregistre la lazy entry, stocke les métadonnées. /// Retourne la liste des lazy PKs enregistrés avec succès. async fn register_tracks_lazy(&self, tracks: &[crate::models::Track]) -> Vec { - // Limite la concurrence pour ne pas saturer l'API Qobuz ni la connexion réseau. - // Les covers déjà cachées sont retournées immédiatement (pas d'HTTP), donc même - // 600 tracks ne génèrent que ~N_albums_uniques téléchargements réels. - let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(16)); + // Configurable via accounts.qobuz.register_concurrency (défaut 4). + // SQLite sérialise les écritures — au-delà de ~4 workers on accumule + // des threads en attente du mutex DB sans gain de débit. + let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(self.inner.register_concurrency)); // On attache l'index original à chaque future pour pouvoir retrier dans l'ordre // d'origine après complétion parallèle (JoinSet retourne dans l'ordre de fin).