Fixed SQL insert shtorm and connected devices migration

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "furumusic"
-version = "0.10.7"
+version = "0.10.8"
 edition = "2024"
 description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL"
 

src/federation/mod.rs

@@ -48,6 +48,11 @@ pub use similarity::SIMILARITY_ALPN;
 
 /// How often the published library is re-synchronized with the database.
 const SYNC_INTERVAL: Duration = Duration::from_secs(60);
+// Active records live for twelve hours. Refreshing a large server library
+// every ten minutes rewrites hundreds of thousands of PostgreSQL rows even
+// though no catalog metadata changed; hourly refreshes leave at least an
+// eleven-hour lifetime if the owner disappears between successful rounds.
+const REPUBLISH_INTERVAL: Duration = Duration::from_secs(60 * 60);
 const SUPERVISOR_INTERVAL: Duration = Duration::from_secs(15);
 const RECOVERY_COOLDOWN: Duration = Duration::from_secs(5 * 60);
 const TRANSPORT_SAMPLE_LIMIT: usize = 16;
@@ -285,6 +290,7 @@ pub struct Federation {
     download_locks: std::sync::Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
     pool: tokio::sync::OnceCell<PgPool>,
     running: tokio::sync::Mutex<Option<Running>>,
+    sync_lock: tokio::sync::Mutex<()>,
     supervisor_started: AtomicBool,
     recovery_count: AtomicU64,
     last_sync: std::sync::Mutex<Option<String>>,
@@ -319,6 +325,7 @@ pub fn handle() -> Arc<Federation> {
             download_locks: std::sync::Mutex::new(Default::default()),
             pool: tokio::sync::OnceCell::new(),
             running: tokio::sync::Mutex::new(None),
+            sync_lock: tokio::sync::Mutex::new(()),
             supervisor_started: AtomicBool::new(false),
             recovery_count: AtomicU64::new(0),
             last_sync: std::sync::Mutex::new(None),
@@ -482,6 +489,7 @@ impl Federation {
         let config = MusicDhtConfig::builder()
             .data_dir(&self.data_dir)
             .network_id(NetworkId::from_name(&network_name))
+            .republish_interval(REPUBLISH_INTERVAL)
             // Peers of the network find each other knowing only its name.
             .rendezvous(RendezvousConfig::default())
             .stream_protocol(AUDIO_ALPN)
@@ -533,6 +541,7 @@ impl Federation {
         let sync_service = Arc::clone(&service);
         let sync_task = tokio::spawn(async move {
             let mut interval = tokio::time::interval(SYNC_INTERVAL);
+            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
             loop {
                 interval.tick().await;
                 let _ = sync_self.sync_once(&sync_service).await;
@@ -686,8 +695,9 @@ impl Federation {
     }
 
     pub async fn sync_now(self: &Arc<Self>) -> Result<()> {
+        let _sync_guard = self.sync_lock.lock().await;
         let service = self.service().await?;
-        let sync_stats = self.sync_once(&service).await?;
+        let sync_stats = self.sync_once_locked(&service).await?;
         let publish_stats = match service.republish().await {
             Ok(stats) => stats,
             Err(err) => {
@@ -701,6 +711,11 @@ impl Federation {
     }
 
     async fn sync_once(self: &Arc<Self>, service: &MusicDhtService) -> Result<SyncStats> {
+        let _sync_guard = self.sync_lock.lock().await;
+        self.sync_once_locked(service).await
+    }
+
+    async fn sync_once_locked(self: &Arc<Self>, service: &MusicDhtService) -> Result<SyncStats> {
         let specs = match self.collect_specs().await {
             Ok(specs) => specs,
             Err(err) => {
@@ -964,19 +979,30 @@ impl Federation {
             Some(running) => {
                 let service = &running.service;
                 let health = service.network_health();
-                let published = service
-                    .list_local_items()
+                // The admin page polls this endpoint. Loading and decoding the
+                // complete library just to display its size used all four DB
+                // connections for large libraries and starved catalog/device
+                // requests. COUNT avoids fetching every serialized payload.
+                let pool = self.pool().await.ok();
+                let published = if let Some(pool) = &pool {
+                    sqlx::query_scalar::<_, i64>(
+                        "SELECT COUNT(*) FROM furumusic__federation_local_item
+                         WHERE deleted = false",
+                    )
+                    .fetch_one(pool)
                     .await
-                    .map(|items| items.len())
-                    .unwrap_or(0);
+                    .unwrap_or(0) as usize
+                } else {
+                    0
+                };
                 let peers: Vec<String> = service
                     .connected_peers()
                     .iter()
                     .map(|p| p.to_string())
                     .collect();
                 let mut transport = self.transport_stats.snapshot();
-                if let Ok(pool) = self.pool().await {
-                    enrich_transport_users(&pool, &mut transport).await;
+                if let Some(pool) = &pool {
+                    enrich_transport_users(pool, &mut transport).await;
                 }
                 json!({
                     "running": true,

src/federation/receive.rs

@@ -399,7 +399,12 @@ impl Federation {
         // when a federation peer is slow or offline.
         let federation = std::sync::Arc::clone(self);
         tokio::spawn(async move {
-            if let Err(err) = federation.sync_now().await {
+            let result = async {
+                let service = federation.service().await?;
+                federation.sync_once(&service).await.map(|_| ())
+            }
+            .await;
+            if let Err(err) = result {
                 tracing::warn!(track_id, "post-import federation publish failed: {err:#}");
             }
         });

src/federation/serve.rs

@@ -2,6 +2,7 @@
 //! the PostgreSQL library and the media storage directory. Wire compatible
 //! with the furumi TUI client and any other furumi peer.
 
+use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
@@ -230,31 +231,6 @@ async fn track_artist_image_file(pool: &PgPool, track_id: i64) -> Result<Option<
     Ok(row.map(|row| (row.get(0), row.get(1))))
 }
 
-async fn track_catalog_artist_names(
-    pool: &PgPool,
-    track_id: i64,
-) -> Result<(Vec<String>, Vec<String>)> {
-    let mut artists = Vec::new();
-    let mut featured = Vec::new();
-    let rows = sqlx::query(
-        "SELECT a.name, ta.role FROM furumusic__track_artist ta
-         JOIN furumusic__artist a ON a.id = ta.artist_id
-         WHERE ta.track_id = $1 ORDER BY ta.position",
-    )
-    .bind(track_id)
-    .fetch_all(pool)
-    .await?;
-    for row in rows {
-        let name: String = row.get(0);
-        match row.get::<String, _>(1).as_str() {
-            "featuring" => featured.push(name),
-            "main" => artists.push(name),
-            _ => {}
-        }
-    }
-    Ok((artists, featured))
-}
-
 async fn track_metadata(pool: &PgPool, track_id: i64) -> Result<Option<TrackMetadata>> {
     let Some(track) = sqlx::query(
         "SELECT t.title, t.track_number, t.disc_number, COALESCE(t.year, r.year),
@@ -620,43 +596,72 @@ async fn build_catalog(pool: &PgPool, own: &EndpointId, artist: &str) -> Result<
     .bind(artist_id)
     .fetch_all(pool)
     .await?;
-    let mut releases = Vec::new();
-    for release_row in release_rows {
-        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,
-                    c.content_id
+
+    // Fetch every release track and its credits in one round trip. The old
+    // implementation performed one track query per release and one artist
+    // query per track; a large discography could not finish before the
+    // client's five-second catalog deadline when PostgreSQL was busy.
+    let release_ids: Vec<i64> = release_rows.iter().map(|row| row.get(0)).collect();
+    let track_rows = if release_ids.is_empty() {
+        Vec::new()
+    } else {
+        sqlx::query(
+            "SELECT t.release_id, t.id, t.title, t.track_number, t.disc_number,
+                    t.duration_seconds, c.content_id,
+                    COALESCE(
+                        array_agg(a.name::text ORDER BY ta.position)
+                            FILTER (WHERE ta.role = 'main'),
+                        ARRAY[]::text[]
+                    ) AS artists,
+                    COALESCE(
+                        array_agg(a.name::text ORDER BY ta.position)
+                            FILTER (WHERE ta.role = 'featuring'),
+                        ARRAY[]::text[]
+                    ) AS featured_artists
              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",
+             LEFT JOIN furumusic__track_artist ta ON ta.track_id = t.id
+             LEFT JOIN furumusic__artist a ON a.id = ta.artist_id
+             WHERE t.release_id = ANY($1) AND t.is_hidden = false
+             GROUP BY t.release_id, t.id, t.title, t.track_number, t.disc_number,
+                      t.duration_seconds, c.content_id
+             ORDER BY t.release_id, t.disc_number NULLS FIRST,
+                      t.track_number NULLS LAST, t.title",
         )
-        .bind(release_id)
+        .bind(&release_ids)
         .fetch_all(pool)
-        .await?;
-        let mut tracks = Vec::with_capacity(track_rows.len());
-        for row in track_rows {
-            let track_id: i64 = row.get(0);
-            let duration: f64 = row.get(4);
-            let (artists, featured_artists) = track_catalog_artist_names(pool, track_id).await?;
-            tracks.push(CatalogTrack {
-                title: row.get(1),
-                artists,
-                featured_artists,
-                track_number: row.get(2),
-                disc_number: row.get(3),
+        .await?
+    };
+    let mut tracks_by_release: HashMap<i64, Vec<CatalogTrack>> = HashMap::new();
+    for row in track_rows {
+        let release_id: i64 = row.get(0);
+        let track_id: i64 = row.get(1);
+        let duration: f64 = row.get(5);
+        tracks_by_release
+            .entry(release_id)
+            .or_default()
+            .push(CatalogTrack {
+                title: row.get(2),
+                artists: row.get(7),
+                featured_artists: row.get(8),
+                track_number: row.get(3),
+                disc_number: row.get(4),
                 duration_seconds: (duration > 0.0).then_some(duration),
-                content_id: row.get(5),
+                content_id: row.get(6),
                 item_id: item_id_of(own, track_id),
             });
-        }
+    }
+
+    let mut releases = Vec::new();
+    for release_row in release_rows {
+        let release_id: i64 = release_row.get(0);
         releases.push(CatalogRelease {
             title: release_row.get(1),
             release_type: release_row.get(2),
             year: release_row.get(3),
-            tracks,
+            tracks: tracks_by_release.remove(&release_id).unwrap_or_default(),
         });
     }
 

src/federation/storage.rs

@@ -8,6 +8,11 @@ use music_dht::{
 use sqlx::{PgPool, Row as _};
 
 const IDENTITY_NAME: &str = "default";
+// Several republish paths can converge on the same record within seconds
+// (startup discovery, a manual sync and a periodic tick). Extending a
+// twelve-hour TTL by less than five minutes is operationally identical, but
+// rewriting the wide row and its expiry index generates substantial WAL.
+const MIN_EXPIRY_EXTENSION_MS: u64 = 5 * 60 * 1_000;
 
 const SCHEMA: &[&str] = &[
     "CREATE TABLE IF NOT EXISTS furumusic__federation_identity (
@@ -133,7 +138,18 @@ impl MusicDhtStorage for PostgresFederationStorage {
                 revision = EXCLUDED.revision,
                 deleted = EXCLUDED.deleted,
                 updated_at_ms = EXCLUDED.updated_at_ms,
-                payload = EXCLUDED.payload",
+                payload = EXCLUDED.payload
+             WHERE (furumusic__federation_local_item.normalized_name,
+                    furumusic__federation_local_item.revision,
+                    furumusic__federation_local_item.deleted,
+                    furumusic__federation_local_item.updated_at_ms,
+                    furumusic__federation_local_item.payload)
+                   IS DISTINCT FROM
+                   (EXCLUDED.normalized_name,
+                    EXCLUDED.revision,
+                    EXCLUDED.deleted,
+                    EXCLUDED.updated_at_ms,
+                    EXCLUDED.payload)",
         )
         .bind(item.id.as_bytes().as_slice())
         .bind(&item.normalized_name)
@@ -317,11 +333,17 @@ async fn store_record_in_conn(
 
     match decide_store(existing, record) {
         StoreDecision::Ignore => return Ok(false),
-        StoreDecision::Write | StoreDecision::RefreshExpiry(_) => {}
+        StoreDecision::RefreshExpiry(new_expiry) => {
+            let previous_expiry = existing.map(|(_, _, expiry)| expiry).unwrap_or(0);
+            if new_expiry.saturating_sub(previous_expiry) < MIN_EXPIRY_EXTENSION_MS {
+                return Ok(false);
+            }
+        }
+        StoreDecision::Write => {}
     }
 
     let payload = postcard::to_stdvec(record).map_err(db_error)?;
-    sqlx::query(
+    let written = sqlx::query(
         "INSERT INTO furumusic__federation_dht_record
          (dht_key, item_id, owner_peer_id, payload, revision, deleted, expires_at_ms)
          VALUES ($1, $2, $3, $4, $5, $6, $7)
@@ -329,7 +351,16 @@ async fn store_record_in_conn(
             payload = EXCLUDED.payload,
             revision = EXCLUDED.revision,
             deleted = EXCLUDED.deleted,
-            expires_at_ms = EXCLUDED.expires_at_ms",
+            expires_at_ms = EXCLUDED.expires_at_ms
+         WHERE (furumusic__federation_dht_record.payload,
+                furumusic__federation_dht_record.revision,
+                furumusic__federation_dht_record.deleted,
+                furumusic__federation_dht_record.expires_at_ms)
+               IS DISTINCT FROM
+               (EXCLUDED.payload,
+                EXCLUDED.revision,
+                EXCLUDED.deleted,
+                EXCLUDED.expires_at_ms)",
     )
     .bind(key.as_bytes().as_slice())
     .bind(record.item.id.as_bytes().as_slice())
@@ -340,8 +371,9 @@ async fn store_record_in_conn(
     .bind(record.expires_at_ms as i64)
     .execute(&mut *conn)
     .await
-    .map_err(db_error)?;
-    Ok(true)
+    .map_err(db_error)?
+    .rows_affected();
+    Ok(written > 0)
 }
 
 fn db_error(err: impl std::fmt::Display) -> MusicDhtError {

src/music/mod.rs

@@ -2784,6 +2784,45 @@ pub mod db_migrations {
             &[Operation::custom(create_youtube_import_media_links).build()];
     }
 
+    // -- M0048: playback coordination columns -----------------------------
+
+    // These columns were originally added to M0038 after that migration had
+    // already shipped. Existing databases therefore never ran the ALTERs:
+    // migration bodies are immutable once their name is recorded as applied.
+    #[cot::db::migrations::migration_op]
+    async fn add_playback_coordination_columns(
+        ctx: migrations::MigrationContext<'_>,
+    ) -> cot::db::Result<()> {
+        ctx.db
+            .raw(
+                "ALTER TABLE furumusic__fed_device_identity
+                 ADD COLUMN IF NOT EXISTS playback_coordination_json JSONB",
+            )
+            .await?;
+        ctx.db
+            .raw(
+                "ALTER TABLE furumusic__fed_device_identity
+                 ADD COLUMN IF NOT EXISTS playback_config_json JSONB",
+            )
+            .await?;
+        Ok(())
+    }
+
+    #[derive(Debug, Copy, Clone)]
+    pub struct M0048AddPlaybackCoordinationColumns;
+
+    impl migrations::Migration for M0048AddPlaybackCoordinationColumns {
+        const APP_NAME: &'static str = "furumusic";
+        const MIGRATION_NAME: &'static str = "m_0048_add_playback_coordination_columns";
+        const DEPENDENCIES: &'static [migrations::MigrationDependency] =
+            &[migrations::MigrationDependency::migration(
+                "furumusic",
+                "m_0047_create_youtube_import_media_links",
+            )];
+        const OPERATIONS: &'static [Operation] =
+            &[Operation::custom(add_playback_coordination_columns).build()];
+    }
+
     pub const MIGRATIONS: &[&SyncDynMigration] = &[
         &M0006CreateMediaFile,
         &M0007CreateArtist,
@@ -2822,5 +2861,6 @@ pub mod db_migrations {
         &M0045CreateYouTubeDownloads,
         &M0046CreateLocalUploadHistory,
         &M0047CreateYouTubeImportMediaLinks,
+        &M0048AddPlaybackCoordinationColumns,
     ];
 }