From 5b339aa9219132eff02b855c896017e0d78339cd Mon Sep 17 00:00:00 2001 From: Ultradesu Date: Mon, 20 Jul 2026 18:40:32 +0300 Subject: [PATCH] Fixed content_id persistence --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/federation/mod.rs | 141 +++++++++++++++++++++++++------------- src/federation/serve.rs | 25 ++----- src/federation/storage.rs | 8 +++ 5 files changed, 110 insertions(+), 68 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4f2eb2e..d7ba90c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1844,7 +1844,7 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" [[package]] name = "furumusic" -version = "0.6.4-fd" +version = "0.6.5-fd" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 8bf9b82..019b509 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "furumusic" -version = "0.6.5-fd" +version = "0.6.6-fd" edition = "2024" description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL" diff --git a/src/federation/mod.rs b/src/federation/mod.rs index 7a8e43e..3e86a33 100644 --- a/src/federation/mod.rs +++ b/src/federation/mod.rs @@ -36,9 +36,6 @@ pub use serve::{AUDIO_ALPN, CATALOG_ALPN}; /// How often the published library is re-synchronized with the database. const SYNC_INTERVAL: Duration = Duration::from_secs(60); -/// Content IDs are file hashes and can be expensive to warm for large -/// libraries. Keep DHT publishing responsive and let later syncs fill them in. -const MAX_CONTENT_HASH_JOBS_PER_SYNC: usize = 512; struct Running { service: Arc, @@ -46,6 +43,12 @@ struct Running { tasks: Vec>, } +struct ContentHashJob { + media_file_id: i64, + sha256_hash: String, + file_path: String, +} + pub struct Federation { /// Transport data directory; server-side DHT state and identity live in PostgreSQL. data_dir: PathBuf, @@ -440,31 +443,36 @@ impl Federation { let tracks = sqlx::query( "SELECT t.id, t.title, COALESCE(t.year, r.year), t.duration_seconds, r.title, r.release_type, t.track_number, t.disc_number, - t.audio_file_id, m.file_path, m.sha256_hash + t.audio_file_id, m.file_path, m.sha256_hash, c.content_id FROM furumusic__track t JOIN furumusic__release r ON r.id = t.release_id JOIN furumusic__media_file m ON m.id = t.audio_file_id + LEFT JOIN furumusic__federation_content_id_cache c + ON c.media_file_id = m.id AND c.sha256_hash = m.sha256_hash WHERE t.is_hidden = false AND r.is_hidden = false", ) .fetch_all(&pool) .await?; let storage_dir = lock(&self.storage_dir).clone(); - let mut hash_jobs_scheduled = 0usize; + let mut content_hash_jobs = Vec::new(); for row in &tracks { let id: i64 = row.get(0); let duration: f64 = row.get(3); let media_file_id: i64 = row.get(8); let file_path: String = row.get(9); let sha256_hash: String = row.get(10); - let (content_id, scheduled_hash) = self.content_id_for_media( - media_file_id, - sha256_hash, - file_path, - &storage_dir, - hash_jobs_scheduled < MAX_CONTENT_HASH_JOBS_PER_SYNC, - ); - if scheduled_hash { - hash_jobs_scheduled += 1; + let cached_content_id: Option = row.get(11); + let content_id = cached_content_id + .or_else(|| self.cached_content_id_for_media(media_file_id, &sha256_hash)); + if content_id.is_none() + && !storage_dir.trim().is_empty() + && self.mark_content_hash_pending(media_file_id) + { + content_hash_jobs.push(ContentHashJob { + media_file_id, + sha256_hash, + file_path, + }); } specs.push(ItemSpec { local_key: format!("track:{id}"), @@ -481,52 +489,67 @@ impl Federation { content_id, }); } + self.spawn_content_warmer(pool.clone(), storage_dir, content_hash_jobs); Ok(specs) } - fn content_id_for_media( - self: &Arc, - media_file_id: i64, - sha256_hash: String, - file_path: String, - storage_dir: &str, - schedule_missing: bool, - ) -> (Option, bool) { - if storage_dir.trim().is_empty() { - return (None, false); - } + fn cached_content_id_for_media(&self, media_file_id: i64, sha256_hash: &str) -> Option { if let Some((cached_hash, content_id)) = lock(&self.content_cache).get(&media_file_id) - && cached_hash == &sha256_hash + && cached_hash == sha256_hash { - return (Some(content_id.clone()), false); - } - if !schedule_missing { - return (None, false); + return Some(content_id.clone()); } + None + } - { - let mut pending = lock(&self.content_pending); - if !pending.insert(media_file_id) { - return (None, false); - } - } + fn mark_content_hash_pending(&self, media_file_id: i64) -> bool { + lock(&self.content_pending).insert(media_file_id) + } - let storage_dir = storage_dir.to_string(); + fn spawn_content_warmer( + self: &Arc, + pool: PgPool, + storage_dir: String, + jobs: Vec, + ) { + if jobs.is_empty() { + return; + } let fed = Arc::clone(self); tokio::spawn(async move { - let content_id = - tokio::task::spawn_blocking(move || audio_content_id(&storage_dir, &file_path)) - .await - .ok() - .flatten(); - lock(&fed.content_pending).remove(&media_file_id); - if let Some(content_id) = content_id { - lock(&fed.content_cache).insert(media_file_id, (sha256_hash, content_id)); + let total = jobs.len(); + let mut stored = 0usize; + for job in jobs { + let job_storage_dir = storage_dir.clone(); + let job_file_path = job.file_path.clone(); + let content_id = tokio::task::spawn_blocking(move || { + audio_content_id(&job_storage_dir, &job_file_path) + }) + .await + .ok() + .flatten(); + lock(&fed.content_pending).remove(&job.media_file_id); + if let Some(content_id) = content_id { + lock(&fed.content_cache).insert( + job.media_file_id, + (job.sha256_hash.clone(), content_id.clone()), + ); + if let Err(err) = + persist_content_id(&pool, job.media_file_id, &job.sha256_hash, &content_id) + .await + { + tracing::warn!( + media_file_id = job.media_file_id, + "federation content-id cache write failed: {err:#}" + ); + } else { + stored += 1; + } + } } + tracing::info!(total, stored, "federation content-id cache warm finished"); }); - - (None, true) } /// Live status for the admin page. @@ -586,6 +609,30 @@ impl Federation { } } +async fn persist_content_id( + pool: &PgPool, + media_file_id: i64, + sha256_hash: &str, + content_id: &str, +) -> Result<()> { + sqlx::query( + "INSERT INTO furumusic__federation_content_id_cache + (media_file_id, sha256_hash, content_id, updated_at) + VALUES ($1, $2, $3, $4) + ON CONFLICT (media_file_id) DO UPDATE SET + sha256_hash = EXCLUDED.sha256_hash, + content_id = EXCLUDED.content_id, + updated_at = EXCLUDED.updated_at", + ) + .bind(media_file_id) + .bind(sha256_hash) + .bind(content_id) + .bind(now_iso()) + .execute(pool) + .await?; + Ok(()) +} + fn audio_content_id(storage_dir: &str, file_path: &str) -> Option { if storage_dir.trim().is_empty() { return None; diff --git a/src/federation/serve.rs b/src/federation/serve.rs index 4bb6f0e..7d4b400 100644 --- a/src/federation/serve.rs +++ b/src/federation/serve.rs @@ -522,7 +522,7 @@ async fn serve_catalog_one( match request.want.as_deref() { None | Some("catalog") => { - let response = match build_catalog(&pool, &own, &storage_dir, &request.artist).await { + let response = match build_catalog(&pool, &own, &request.artist).await { Ok(response) => response, Err(err) => CatalogResponse { ok: false, @@ -582,12 +582,7 @@ async fn serve_catalog_one( Ok(()) } -async fn build_catalog( - pool: &PgPool, - own: &EndpointId, - storage_dir: &str, - artist: &str, -) -> Result { +async fn build_catalog(pool: &PgPool, own: &EndpointId, artist: &str) -> Result { let Some(artist_row) = sqlx::query( "SELECT id, name FROM furumusic__artist WHERE LOWER(name) = LOWER($1) AND is_hidden = false @@ -620,9 +615,11 @@ async fn build_catalog( let release_id: i64 = release_row.get(0); let track_rows = sqlx::query( "SELECT t.id, t.title, t.track_number, t.disc_number, t.duration_seconds, - m.file_path + c.content_id FROM furumusic__track t JOIN furumusic__media_file m ON m.id = t.audio_file_id + LEFT JOIN furumusic__federation_content_id_cache c + ON c.media_file_id = m.id AND c.sha256_hash = m.sha256_hash WHERE t.release_id = $1 AND t.is_hidden = false ORDER BY t.disc_number NULLS FIRST, t.track_number NULLS LAST, t.title", ) @@ -633,14 +630,12 @@ async fn build_catalog( for row in track_rows { let track_id: i64 = row.get(0); let duration: f64 = row.get(4); - let file_path: String = row.get(5); - let content_id = catalog_content_id(storage_dir, file_path).await; tracks.push(CatalogTrack { title: row.get(1), track_number: row.get(2), disc_number: row.get(3), duration_seconds: (duration > 0.0).then_some(duration), - content_id, + content_id: row.get(5), item_id: item_id_of(own, track_id), }); } @@ -662,14 +657,6 @@ async fn build_catalog( }) } -async fn catalog_content_id(storage_dir: &str, file_path: String) -> Option { - let storage_dir = storage_dir.to_string(); - tokio::task::spawn_blocking(move || super::audio_content_id(&storage_dir, &file_path)) - .await - .ok() - .flatten() -} - async fn artist_image_by_name(pool: &PgPool, artist: &str) -> Result> { let row = sqlx::query( "SELECT m.file_path, m.mime_type FROM furumusic__artist a diff --git a/src/federation/storage.rs b/src/federation/storage.rs index bc3cdf9..070e8d8 100644 --- a/src/federation/storage.rs +++ b/src/federation/storage.rs @@ -44,6 +44,14 @@ const SCHEMA: &[&str] = &[ ticket TEXT NOT NULL, last_seen_ms BIGINT NOT NULL )", + "CREATE TABLE IF NOT EXISTS furumusic__federation_content_id_cache ( + media_file_id BIGINT PRIMARY KEY, + sha256_hash TEXT NOT NULL, + content_id TEXT NOT NULL, + updated_at TEXT NOT NULL + )", + "CREATE INDEX IF NOT EXISTS idx_furumusic_federation_content_id_cache_content_id + ON furumusic__federation_content_id_cache(content_id)", ]; #[derive(Debug, Clone)]