eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

debugage lazy cache

Browse Source
This commit is contained in:
Eric Coissac
2025-12-15 15:05:35 +01:00
parent d2d8111668
commit 68e6f528e5
18 changed files with 641 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
pmocache/Cargo.toml
View File

@@ -18,6 +18,7 @@ hex = "0.4"
# Utilitaires
anyhow = "1.0"
async-trait = "0.1"
chrono = "0.4"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"

226
pmocache/src/cache.rs
View File

@@ -8,17 +8,23 @@ use crate::db::DB;
use crate::download::{
download_with_transformer, ingest_with_transformer, Download, StreamTransformer,
};
use crate::lazy::{lazy_prefix_from_pk, LazyEntryRemoteData, LazyProvider};
use anyhow::{anyhow, bail, Result};
use serde_json::{Number, Value};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::sync::{Arc, RwLock as StdRwLock};
use tokio::io::{AsyncRead, AsyncReadExt};
use tokio::sync::{broadcast, RwLock};
use tracing;
enum FinalizeMode<'a> {
InsertNew,
ConvertLazy { lazy_pk: &'a str },
}
// ============================================================================
// LAZY PK SUPPORT
// ============================================================================
@@ -41,7 +47,11 @@ pub fn generate_lazy_pk(url: &str) -> String {
/// Vérifie si un PK est en mode lazy
pub fn is_lazy_pk(pk: &str) -> bool {
pk.starts_with(LAZY_PK_PREFIX)
if pk.starts_with(LAZY_PK_PREFIX) {
return true;
}
lazy_prefix_from_pk(pk).is_some()
}
/// Events émis par le cache pour notifier les changements d'état
@@ -136,6 +146,8 @@ pub struct Cache<C: CacheConfig> {
min_prebuffer_size: u64,
/// LAZY PK SUPPORT: Channel pour broadcaster les events (lazy downloads, etc.)
served_tx: Option<broadcast::Sender<CacheEvent>>,
/// Providers responsables de préfixes lazy spécifiques
lazy_providers: StdRwLock<HashMap<String, Arc<dyn LazyProvider>>>,
/// Phantom data pour le type de configuration
_phantom: std::marker::PhantomData<C>,
}
@@ -242,6 +254,7 @@ impl<C: CacheConfig> Cache<C> {
download: Arc<Download>,
collection: Option<&str>,
origin_url: Option<&str>,
mode: FinalizeMode<'_>,
) -> Result<String> {
// Attendre le prébuffering (pour le cache progressif)
if self.min_prebuffer_size > 0 {
@@ -256,10 +269,17 @@ impl<C: CacheConfig> Cache<C> {
);
}
// Ajouter à la DB une fois le prébuffer terminé
self.db.add(pk, None, collection)?;
if let Some(url) = origin_url {
self.db.set_origin_url(pk, url)?;
// Ajouter ou commuter la DB selon le mode
match mode {
FinalizeMode::InsertNew => {
self.db.add(pk, None, collection)?;
if let Some(url) = origin_url {
self.db.set_origin_url(pk, url)?;
}
}
FinalizeMode::ConvertLazy { lazy_pk } => {
self.db.update_lazy_to_downloaded(lazy_pk, pk)?;
}
}
// Sauvegarder les métadonnées techniques du transformer
@@ -400,6 +420,7 @@ impl<C: CacheConfig> Cache<C> {
transformer_factory,
min_prebuffer_size: DEFAULT_PREBUFFER_SIZE,
served_tx: Some(served_tx),
lazy_providers: StdRwLock::new(HashMap::new()),
_phantom: std::marker::PhantomData,
})
}
@@ -432,6 +453,34 @@ impl<C: CacheConfig> Cache<C> {
self.min_prebuffer_size
}
/// Enregistre un provider responsable d'un préfixe de lazy PK.
pub fn register_lazy_provider(&self, provider: Arc<dyn LazyProvider>) {
let prefix = provider.lazy_prefix().to_string();
let mut guard = self
.lazy_providers
.write()
.expect("lazy provider registry poisoned");
guard.insert(prefix, provider);
}
/// Désenregistre un provider à partir de son préfixe.
pub fn unregister_lazy_provider(&self, prefix: &str) {
let mut guard = self
.lazy_providers
.write()
.expect("lazy provider registry poisoned");
guard.remove(prefix);
}
fn provider_for_lazy_pk(&self, lazy_pk: &str) -> Option<Arc<dyn LazyProvider>> {
let prefix = lazy_prefix_from_pk(lazy_pk)?;
let guard = self
.lazy_providers
.read()
.expect("lazy provider registry poisoned");
guard.get(prefix).cloned()
}
/// S'abonne aux diffusions HTTP pour un `pk` donné.
///
/// La callback est appelée à chaque fois qu'un élément est servi avec succès via les routes
@@ -611,10 +660,63 @@ impl<C: CacheConfig> Cache<C> {
}
// Finaliser avec prébuffering et nettoyage
self.finalize_download(&pk, download, collection, Some(url))
self.finalize_download(&pk, download, collection, Some(url), FinalizeMode::InsertNew)
.await
}
/// Télécharge un fichier lazy et commute l'entrée existante
pub async fn download_lazy_from_url(
&self,
lazy_pk: &str,
url: &str,
collection: Option<&str>,
) -> Result<String> {
// Si déjà converti, retourner directement
if let Ok(Some(real_pk)) = self.db.get_pk_by_lazy_pk(lazy_pk) {
return Ok(real_pk);
}
// Si l'URL pointe déjà vers un fichier complet, commuter sans re-télécharger
if let Ok(Some(existing_pk)) = self.db.get_pk_by_origin_url(url) {
if existing_pk != lazy_pk && self.check_cached_and_complete(&existing_pk).await? {
self.db.update_lazy_to_downloaded(lazy_pk, &existing_pk)?;
return Ok(existing_pk);
}
}
// 1. Télécharger les 2048 premiers octets pour calculer le pk
let header = crate::download::peek_header(url, 2048)
.await
.map_err(|e| anyhow!("Failed to peek header: {}", e))?;
let pk = crate::cache_trait::pk_from_content_header(&header);
// 2. Si déjà en cache (complet), commuter directement
if self.check_cached_and_complete(&pk).await? {
self.db.update_lazy_to_downloaded(lazy_pk, &pk)?;
return Ok(pk);
}
// 3. Lancer le téléchargement complet
tracing::debug!("Starting lazy download for pk {} (lazy {})", pk, lazy_pk);
let file_path = self.get_file_path(&pk);
let transformer = self.transformer_factory.as_ref().map(|f| f());
let download = download_with_transformer(&file_path, url, transformer);
{
let mut downloads = self.downloads.write().await;
downloads.insert(pk.clone(), download.clone());
}
self.finalize_download(
&pk,
download,
collection,
Some(url),
FinalizeMode::ConvertLazy { lazy_pk },
)
.await
}
/// Ajoute un fichier à partir d'un flux asynchrone.
///
/// Cette méthode utilise le même système d'identifiants basé sur le contenu que `add_from_url`.
@@ -725,7 +827,7 @@ impl<C: CacheConfig> Cache<C> {
}
// Finaliser avec prébuffering et nettoyage
self.finalize_download(&pk, download, collection, source_uri)
self.finalize_download(&pk, download, collection, source_uri, FinalizeMode::InsertNew)
.await
}
@@ -1297,71 +1399,95 @@ impl<C: CacheConfig> Cache<C> {
}
}
/// Ajoute une URL en mode deferred (pas de download immédiat)
///
/// Vérifie d'abord si l'URL existe déjà en DB :
/// - Si eager (déjà téléchargé) : retourne le lazy_pk si existe, sinon le real pk
/// - Si lazy (pas encore téléchargé) : retourne le lazy pk existant
/// - Sinon : crée nouvelle entry lazy
///
/// # Arguments
///
/// * `url` - URL à cacher
/// * `collection` - Collection optionnelle
///
/// # Returns
///
/// PK (lazy ou real selon l'état)
///
/// # Example
///
/// ```rust,no_run
/// let cache = Cache::<AudioConfig>::new("./cache", 1000)?;
/// let lazy_pk = cache.add_from_url_deferred("https://example.com/track.mp3", Some("qobuz")).await?;
/// // → Returns "L:abc123..." (lazy PK)
/// // Fichier pas encore téléchargé, juste métadonnées en DB
/// ```
/// Garantit l'existence d'une entrée lazy spécifique.
pub async fn ensure_lazy_entry(
&self,
lazy_pk: &str,
collection: Option<&str>,
origin_url: Option<&str>,
) -> Result<()> {
if let Ok(true) = self.db.has_lazy_entry(lazy_pk) {
self.db.update_hit_by_lazy_pk(lazy_pk)?;
} else {
self.db.add_lazy(lazy_pk, None, collection)?;
}
if let Some(url) = origin_url {
self.db.set_origin_url_for_lazy(lazy_pk, url)?;
}
Ok(())
}
/// Récupère auprès du provider les métadonnées/couvertures associées.
pub async fn fetch_lazy_provider_data(&self, lazy_pk: &str) -> Result<LazyEntryRemoteData> {
if let Some(provider) = self.provider_for_lazy_pk(lazy_pk) {
let metadata = provider.metadata(lazy_pk).await?;
let cover_url = provider.cover_url(lazy_pk).await?;
Ok(LazyEntryRemoteData { metadata, cover_url })
} else {
Ok(LazyEntryRemoteData::default())
}
}
/// Résout l'URL d'origine pour un lazy PK, via la DB ou un provider.
pub async fn resolve_lazy_url(&self, lazy_pk: &str) -> Result<String> {
if let Ok(Some(url)) = self.db.get_origin_url(lazy_pk) {
return Ok(url);
}
if let Some(provider) = self.provider_for_lazy_pk(lazy_pk) {
return provider.get_url(lazy_pk).await;
}
bail!("No origin URL or lazy provider registered for {}", lazy_pk);
}
/// Télécharge un lazy PK en résolvant automatiquement son URL.
pub async fn download_lazy(&self, lazy_pk: &str, collection: Option<&str>) -> Result<String> {
let (existing_pk, _lazy_sec, existing_collection) = self
.db
.get_entry_by_pk_or_lazy_pk(lazy_pk)?
.ok_or_else(|| anyhow!("Lazy pk {} not found in DB", lazy_pk))?;
let url = self.resolve_lazy_url(lazy_pk).await?;
let collection = collection.or(existing_collection.as_deref());
if let Some(real_pk) = existing_pk {
if real_pk != lazy_pk && self.check_cached_and_complete(&real_pk).await? {
return Ok(real_pk);
}
}
self.download_lazy_from_url(lazy_pk, &url, collection).await
}
/// Ajoute une URL sans lancer immédiatement le téléchargement.
pub async fn add_from_url_deferred(
&self,
url: &str,
collection: Option<&str>,
) -> Result<String> {
// 1. Vérifier si URL déjà en cache
if let Ok(Some((pk_opt, lazy_pk_opt))) = self.db.get_entry_by_url(url) {
// URL existe déjà
if let Some(pk) = pk_opt {
// Fichier déjà téléchargé (eager ou lazy→eager)
tracing::debug!("URL {} already downloaded with pk {}", url, pk);
self.db.update_hit(&pk)?;
// Retourner lazy_pk si existe (pour compatibilité Control Point)
// sinon retourner pk
if let Some(lpk) = lazy_pk_opt {
return Ok(lpk);
}
return Ok(pk);
} else if let Some(lpk) = lazy_pk_opt {
// Entry lazy existante (pas encore téléchargé)
tracing::debug!("URL {} already in lazy mode with pk {}", url, lpk);
self.db.update_hit_by_lazy_pk(&lpk)?;
return Ok(lpk);
}
}
// 2. URL inconnue → créer nouvelle entry lazy
let lazy_pk = generate_lazy_pk(url);
// Vérifier si ce lazy_pk existe déjà (collision improbable mais...)
let lazy_pk = format!("L:{}", generate_lazy_pk(url));
if let Ok(true) = self.db.has_lazy_entry(&lazy_pk) {
bail!("Lazy PK collision for URL: {}", url);
}
// 3. Ajouter en DB
self.db.add_lazy(&lazy_pk, None, collection)?;
self.db.set_origin_url_for_lazy(&lazy_pk, url)?;
self.ensure_lazy_entry(&lazy_pk, collection, Some(url)).await?;
tracing::debug!("Created new lazy pk {} for URL {}", lazy_pk, url);
Ok(lazy_pk)
}
}

73
pmocache/src/db.rs
View File

@@ -111,6 +111,7 @@ impl DB {
/// ```
pub fn init(path: &Path) -> Result<Self, rusqlite::Error> {
let conn = Connection::open(path)?;
conn.execute("PRAGMA foreign_keys = ON", [])?;
conn.execute(
"CREATE TABLE IF NOT EXISTS asset (
@@ -130,9 +131,10 @@ impl DB {
value_type TEXT NOT NULL CHECK (value_type IN ('string','number','boolean','null')),
value TEXT,
PRIMARY KEY (pk, key),
FOREIGN KEY (pk) REFERENCES asset (pk) ON DELETE CASCADE
)"
, [])?;
FOREIGN KEY (pk) REFERENCES asset (pk) ON DELETE CASCADE ON UPDATE CASCADE
)",
[],
)?;
// Créer un index sur la collection pour les requêtes rapides
conn.execute(
@@ -856,11 +858,9 @@ impl DB {
/// * `real_pk` - Le real PK calculé après téléchargement
pub fn update_lazy_to_downloaded(&self, lazy_pk: &str, real_pk: &str) -> rusqlite::Result<()> {
let mut conn = self.lock_conn("update_lazy_to_downloaded");
let tx = conn.transaction()?;
// 1. Récupérer l'entry lazy (pk = lazy_pk tant que pas téléchargé)
let (old_pk, collection, id, hits): (String, Option<String>, Option<String>, i32) = tx
let (current_pk, collection, id, hits): (String, Option<String>, Option<String>, i32) = tx
.query_row(
"SELECT pk, collection, id, hits FROM asset WHERE lazy_pk = ?1",
[lazy_pk],
@@ -869,35 +869,40 @@ impl DB {
.optional()?
.ok_or_else(|| Error::QueryReturnedNoRows)?;
if old_pk == real_pk {
// Rien à faire si déjà commuté
if current_pk == real_pk {
return Ok(());
}
let now = Utc::now().to_rfc3339();
let hits_to_add = if hits > 0 { hits } else { 1 };
// 2. Créer/mettre à jour l'entry avec le real pk
tx.execute(
"INSERT INTO asset (pk, lazy_pk, collection, id, hits, last_used)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)
ON CONFLICT(pk) DO UPDATE SET
lazy_pk = excluded.lazy_pk,
collection = COALESCE(excluded.collection, collection),
id = COALESCE(excluded.id, id),
hits = hits + excluded.hits,
last_used = excluded.last_used",
params![real_pk, lazy_pk, collection, id, hits_to_add, now],
// Supprimer d'éventuelles métadonnées résiduelles associées au futur pk réel
// (peut arriver si un ancien téléchargement a laissé des traces sans asset correspondant).
tx.execute("DELETE FROM metadata WHERE pk = ?1", [real_pk])?;
let updated = tx.execute(
"UPDATE asset
SET pk = ?1,
lazy_pk = ?2,
collection = COALESCE(?3, collection),
id = COALESCE(?4, id),
hits = hits + ?5,
last_used = ?6
WHERE lazy_pk = ?7",
params![
real_pk,
lazy_pk,
collection,
id,
hits_to_add,
now,
lazy_pk
],
)?;
// 3. Re-pointer les métadonnées vers le real pk
tx.execute(
"UPDATE metadata SET pk = ?1 WHERE pk = ?2",
params![real_pk, old_pk],
)?;
// 4. Supprimer l'ancienne entry lazy
tx.execute("DELETE FROM asset WHERE pk = ?1", [old_pk])?;
if updated == 0 {
return Err(Error::QueryReturnedNoRows);
}
tx.commit()
}
@@ -952,6 +957,20 @@ impl DB {
Ok(result)
}
/// Retourne une entry à partir d'un pk ou lazy_pk.
pub fn get_entry_by_pk_or_lazy_pk(
&self,
value: &str,
) -> rusqlite::Result<Option<(Option<String>, Option<String>, Option<String>)>> {
let conn = self.lock_conn("get_entry_by_pk_or_lazy_pk");
conn.query_row(
"SELECT pk, lazy_pk, collection FROM asset WHERE pk = ?1 OR lazy_pk = ?1 LIMIT 1",
[value],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.optional()
}
/// Met à jour le compteur d'accès pour une entry lazy (pk = NULL)
///
/// # Arguments

42
pmocache/src/lazy.rs Normal file
View File

@@ -0,0 +1,42 @@
use anyhow::Result;
use async_trait::async_trait;
use serde_json::Value;
/// Retourne le préfixe d'un lazy PK (`PREFIX:VALUE`)
pub fn lazy_prefix_from_pk(lazy_pk: &str) -> Option<&str> {
lazy_pk.split_once(':').map(|(prefix, _)| prefix)
}
/// Données optionnelles pouvant être fournies par un [`LazyProvider`]
#[derive(Debug, Clone, Default)]
pub struct LazyEntryRemoteData {
pub metadata: Option<Value>,
pub cover_url: Option<String>,
}
/// Trait générique décrivant un fournisseur de lazy PK.
///
/// Chaque implémentation est responsable d'un préfixe particulier (ex: `QOBUZ`).
/// Lorsque le cache rencontre un lazy PK dont le préfixe correspond,
/// il délègue au provider pour résoudre l'URL et récupérer les informations
/// nécessaires (métadonnées, couverture, etc.).
#[async_trait]
pub trait LazyProvider: Send + Sync {
/// Préfixe associé (sans le `:` final).
fn lazy_prefix(&self) -> &'static str;
/// Retourne l'URL de téléchargement actuelle pour ce lazy PK.
async fn get_url(&self, lazy_pk: &str) -> Result<String>;
/// Métadonnées optionnelles à associer immédiatement à l'entrée lazy.
async fn metadata(&self, lazy_pk: &str) -> Result<Option<Value>> {
let _ = lazy_pk;
Ok(None)
}
/// URL de couverture éventuelle pour permettre un cache eager des jaquettes.
async fn cover_url(&self, lazy_pk: &str) -> Result<Option<String>> {
let _ = lazy_pk;
Ok(None)
}
}

2
pmocache/src/lib.rs
View File

@@ -115,6 +115,7 @@ pub mod cache;
pub mod cache_trait;
pub mod db;
pub mod download;
pub mod lazy;
pub mod metadata_macros;
#[cfg(feature = "pmoserver")]
@@ -134,6 +135,7 @@ pub use cache::{
CacheSubscription,
};
pub use cache_trait::{pk_from_content_header, FileCache};
pub use lazy::{lazy_prefix_from_pk, LazyEntryRemoteData, LazyProvider};
pub use db::{CacheEntry, DB};
pub use download::{
download, download_with_transformer, ingest_with_transformer, peek_header, peek_reader_header,

110
pmocache/src/pmoserver_ext.rs
View File

@@ -110,15 +110,38 @@ async fn get_file_with_param<C: CacheConfig + 'static>(
serve_file_with_streaming(&cache, &pk, &param, content_type, param_generator).await
}
#[cfg(feature = "pmoserver")]
async fn serve_finalized_pk<C: CacheConfig>(
cache: &Arc<Cache<C>>,
pk: &str,
param: &str,
content_type: &'static str,
) -> Response {
let qualifier = param.to_string();
let file_path = cache.get_file_path_with_qualifier(pk, param);
if let Err(e) = cache.db.update_hit(pk) {
warn!("Error updating hit count for {}: {}", pk, e);
}
let response = serve_complete_file(file_path, content_type).await;
if response.status().is_success() {
cache.notify_broadcast(pk, &qualifier).await;
}
response
}
/// Handler spécifique pour les lazy PK
///
/// Gère le téléchargement on-demand des fichiers lazy :
/// 1. Fast path : vérifie si déjà téléchargé
/// 2. Récupère l'origin_url depuis la DB
/// 2. Résout l'URL via la DB ou un provider
/// 3. Lance le téléchargement et calcule le real pk
/// 4. Met à jour la DB (lazy → downloaded)
/// 5. Broadcast l'event pour PK switching
/// 6. Redirige vers le real pk
/// 6. Sert directement le fichier téléchargé
#[cfg(feature = "pmoserver")]
async fn serve_lazy_audio_file<C: CacheConfig>(
cache: &Arc<Cache<C>>,
@@ -126,54 +149,20 @@ async fn serve_lazy_audio_file<C: CacheConfig>(
param: &str,
content_type: &'static str,
) -> Response {
use axum::response::Redirect;
tracing::info!("Lazy download triggered for pk: {}", lazy_pk);
// 1. Vérifier si déjà téléchargé (fast path)
if let Ok(Some(real_pk)) = cache.db.get_pk_by_lazy_pk(lazy_pk) {
tracing::debug!(
"Lazy PK {} already downloaded as {}, redirecting",
"Lazy PK {} already downloaded as {}, serving immediately",
lazy_pk,
real_pk
);
// Construire l'URL de redirection
let redirect_url = if param == C::default_param() {
format!(
"/{}/{}/{}",
C::cache_name(),
C::cache_type(),
real_pk
)
} else {
format!(
"/{}/{}/{}/{}",
C::cache_name(),
C::cache_type(),
real_pk,
param
)
};
return Redirect::temporary(&redirect_url).into_response();
return serve_finalized_pk(cache, &real_pk, param, content_type).await;
}
// 2. Récupérer origin_url
let origin_url = match cache.db.get_origin_url(lazy_pk) {
Ok(Some(url)) => url,
Ok(None) => {
tracing::error!("Lazy PK {} has no origin_url", lazy_pk);
return (StatusCode::NOT_FOUND, "Origin URL not found").into_response();
}
Err(e) => {
tracing::error!("Error getting origin_url for {}: {}", lazy_pk, e);
return (StatusCode::INTERNAL_SERVER_ERROR, "Database error").into_response();
}
};
// 3. Lancer download complet (cela calcule le VRAI pk basé sur content[512:2048])
let real_pk = match cache.add_from_url(&origin_url, None).await {
// 2. Télécharger en résolvant l'URL via la DB ou un provider
let real_pk = match cache.download_lazy(lazy_pk, None).await {
Ok(pk) => pk,
Err(e) => {
tracing::error!("Failed to download lazy file: {}", e);
@@ -185,46 +174,11 @@ async fn serve_lazy_audio_file<C: CacheConfig>(
}
};
// 4. Mettre à jour DB : lazy_pk → real_pk mapping
if let Err(e) = cache.db.update_lazy_to_downloaded(lazy_pk, &real_pk) {
tracing::error!(
"Failed to update DB for lazy transition {} → {}: {}",
lazy_pk,
real_pk,
e
);
// Continuer quand même, le fichier est téléchargé
}
// 5. Broadcast event pour prefetch ET commutation PK
// 4. Broadcast event pour prefetch ET commutation PK
cache.broadcast_lazy_downloaded(lazy_pk, &real_pk).await;
// 6. Rediriger vers la vraie URL
let redirect_url = if param == C::default_param() {
format!(
"/{}/{}/{}",
C::cache_name(),
C::cache_type(),
real_pk
)
} else {
format!(
"/{}/{}/{}/{}",
C::cache_name(),
C::cache_type(),
real_pk,
param
)
};
tracing::debug!(
"Lazy PK {} downloaded as {}, redirecting to {}",
lazy_pk,
real_pk,
redirect_url
);
Redirect::temporary(&redirect_url).into_response()
// 5. Servir directement le fichier téléchargé
serve_finalized_pk(cache, &real_pk, param, content_type).await
}
/// Fonction utilitaire pour servir un fichier avec streaming progressif