From 8c0ea444f0f7fec8ba1d497940c6ac8275e56414 Mon Sep 17 00:00:00 2001 From: Aleksandr Bogomiakov Date: Tue, 6 Oct 2026 18:09:47 +0100 Subject: [PATCH] Fixed SQL insert shtorm and connected devices migration --- Cargo.toml | 2 +- src/federation/mod.rs | 40 ++++++++++++--- src/federation/receive.rs | 7 ++- src/federation/serve.rs | 103 ++++++++++++++++++++------------------ src/federation/storage.rs | 44 +++++++++++++--- src/music/mod.rs | 40 +++++++++++++++ 6 files changed, 172 insertions(+), 64 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index dc4dbf7..2b4de27 100644 --- a/Cargo.toml +++ b/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" diff --git a/src/federation/mod.rs b/src/federation/mod.rs index cf70ddb..435b019 100644 --- a/src/federation/mod.rs +++ b/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>>>, pool: tokio::sync::OnceCell, running: tokio::sync::Mutex>, + sync_lock: tokio::sync::Mutex<()>, supervisor_started: AtomicBool, recovery_count: AtomicU64, last_sync: std::sync::Mutex>, @@ -319,6 +325,7 @@ pub fn handle() -> Arc { 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) -> 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, service: &MusicDhtService) -> Result { + let _sync_guard = self.sync_lock.lock().await; + self.sync_once_locked(service).await + } + + async fn sync_once_locked(self: &Arc, service: &MusicDhtService) -> Result { 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 = 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, diff --git a/src/federation/receive.rs b/src/federation/receive.rs index c05c507..e3f7ef2 100644 --- a/src/federation/receive.rs +++ b/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:#}"); } }); diff --git a/src/federation/serve.rs b/src/federation/serve.rs index b1669c8..3fdbe3e 100644 --- a/src/federation/serve.rs +++ b/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 Result<(Vec, Vec)> { - 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::(1).as_str() { - "featuring" => featured.push(name), - "main" => artists.push(name), - _ => {} - } - } - Ok((artists, featured)) -} - async fn track_metadata(pool: &PgPool, track_id: i64) -> Result> { 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 = 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> = 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(), }); } diff --git a/src/federation/storage.rs b/src/federation/storage.rs index 7fac005..03d0e9c 100644 --- a/src/federation/storage.rs +++ b/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 { diff --git a/src/music/mod.rs b/src/music/mod.rs index 8f425fa..ea010f5 100644 --- a/src/music/mod.rs +++ b/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, ]; }