Fixed content_id persistence

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",

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"
 

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<MusicDhtService>,
@@ -46,6 +43,12 @@ struct Running {
     tasks: Vec<tokio::task::JoinHandle<()>>,
 }
 
+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<String> = 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<Self>,
-        media_file_id: i64,
-        sha256_hash: String,
-        file_path: String,
-        storage_dir: &str,
-        schedule_missing: bool,
-    ) -> (Option<String>, 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<String> {
         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<Self>,
+        pool: PgPool,
+        storage_dir: String,
+        jobs: Vec<ContentHashJob>,
+    ) {
+        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<String> {
     if storage_dir.trim().is_empty() {
         return None;

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<CatalogResponse> {
+async fn build_catalog(pool: &PgPool, own: &EndpointId, artist: &str) -> Result<CatalogResponse> {
     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<String> {
-    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<Option<(String, String)>> {
     let row = sqlx::query(
         "SELECT m.file_path, m.mime_type FROM furumusic__artist a

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)]