itlxrd/Psychotoxical-psysonic
1
0
Fork 0
mirror of https://github.com/Psychotoxical/psysonic.git synced 2026-07-21 14:55:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(library): bound post-sync maintenance

Browse Source
This commit is contained in:
cucadmuh
2026-07-21 00:42:51 +03:00
parent 92954fbbc3
commit ebbbc39014
18 changed files with 889 additions and 611 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/crates/psysonic-library/migrations/026_library_tag_cursor.sql
+8
View File
@@ -0,0 +1,8 @@
-- Resumable cursor for bounded post-sync library membership tagging.
CREATE TABLE IF NOT EXISTS library_tag_cursor (
server_id TEXT PRIMARY KEY,
folders_hash TEXT NOT NULL,
next_folder_id TEXT NOT NULL,
next_album_offset INTEGER NOT NULL DEFAULT 0,
updated_at INTEGER NOT NULL
);
src-tauri/crates/psysonic-library/src/commands.rs
+11 -3
View File
@@ -1467,7 +1467,7 @@ async fn library_sync_start_inner(
let _ = app_for_emit.emit(LibrarySyncProgressPayload::PROGRESS_EVENT_NAME, &payload);
}
// Wait for the runner to finish + emit sync-idle.
let outcome = match runner_handle.await {
let mut outcome = match runner_handle.await {
Ok(Ok(())) => {
LibrarySyncIdlePayload::ok(
&server_id_for_emit,
@@ -1520,6 +1520,7 @@ async fn library_sync_start_inner(
server_id_for_emit,
error
);
outcome.mark_failed(format!("identity maintenance failed: {error}"));
}
}
}
@@ -1921,6 +1922,10 @@ fn purge_server_data(
"DELETE FROM library_tag_state WHERE server_id = ?1",
params![server_id],
)?;
tx.execute(
"DELETE FROM library_tag_cursor WHERE server_id = ?1",
params![server_id],
)?;
tx.execute(
"DELETE FROM cluster.track_cluster_key WHERE server_id = ?1",
params![server_id],
@@ -2213,8 +2218,10 @@ mod tests {
VALUES ('{server_id}', '{track_id}', 'Rock', '{album_id}');
INSERT INTO artist_artwork_lookup(server_id, artist_id, surface_kind, status, updated_at)
VALUES ('{server_id}', '{artist_id}', 'fanart', 'hit', 1);
INSERT INTO library_tag_state(server_id, folders_hash, completed_at)
VALUES ('{server_id}', 'hash', 1);
INSERT INTO library_tag_state(server_id, folders_hash, completed_at)
VALUES ('{server_id}', 'hash', 1);
INSERT INTO library_tag_cursor(server_id, folders_hash, next_folder_id, updated_at)
VALUES ('{server_id}', 'hash', 'folder-1', 1);
INSERT INTO entity_user_rating(server_id, entity_kind, entity_id, rating, fetched_at)
VALUES ('{server_id}', 'track', '{track_id}', 5, 1);
INSERT INTO album_browse_projection(
@@ -2648,6 +2655,7 @@ mod tests {
("track_genre", "server_id"),
("artist_artwork_lookup", "server_id"),
("library_tag_state", "server_id"),
("library_tag_cursor", "server_id"),
("entity_user_rating", "server_id"),
("album_browse_projection", "server_id"),
("canonical_enrichment_link", "owner_server_id"),
src-tauri/crates/psysonic-library/src/identity/rebuild.rs
+68
View File
@@ -479,7 +479,22 @@ pub fn rebuild_cluster_keys(store: &LibraryStore, server_id: Option<&str>) -> Re
/// - durable invalidations are pending for this server. Initial/resync ingestion
/// records a server invalidation; ordinary mutations record exact entities.
pub fn ensure_cluster_keys_built(store: &LibraryStore, server_id: &str) -> Result<u64, String> {
let pending = match store
.with_read_conn(|conn| Ok(pending_rebuild(conn, server_id)?.is_some()))
{
Ok(pending) => pending,
// Shared-cache in-memory tests and a busy sidecar can briefly reject a
// read while another owner commits cluster metadata. Serialize through
// the writer and re-check instead of surfacing a maintenance failure.
Err(error) if error.contains("locked") => true,
Err(error) => return Err(error),
};
if !pending {
return Ok(0);
}
store.with_conn_mut("identity.ensure_cluster_keys_built", |conn| {
// Re-check under the writer lock because another maintenance owner may
// have drained the journal after the read-only preflight.
let Some(pending) = pending_rebuild(conn, server_id)? else {
return Ok(0);
};
@@ -1142,6 +1157,59 @@ mod tests {
assert_eq!(rebuilt.into_iter().sum::<u64>(), 1);
}
#[test]
fn clean_identity_ensure_does_not_wait_for_writer_lock() {
use std::sync::{mpsc, Arc};
use std::time::Duration;
let store = Arc::new(LibraryStore::open_in_memory());
TrackRepository::new(&store)
.upsert_batch(&[track_row(
"s1",
"t1",
"Title",
Some("Artist"),
"Album",
None,
180,
"lib",
)])
.unwrap();
rebuild_cluster_keys(&store, None).unwrap();
let (writer_started_tx, writer_started_rx) = mpsc::channel();
let (release_writer_tx, release_writer_rx) = mpsc::channel();
let writer_store = Arc::clone(&store);
let writer = std::thread::spawn(move || {
writer_store
.with_conn_mut("test.hold_writer", |_conn| {
writer_started_tx.send(()).unwrap();
release_writer_rx.recv().unwrap();
Ok(())
})
.unwrap();
});
writer_started_rx.recv().unwrap();
let (ensure_tx, ensure_rx) = mpsc::channel();
let ensure_store = Arc::clone(&store);
let ensure = std::thread::spawn(move || {
ensure_tx
.send(ensure_cluster_keys_built(&ensure_store, "s1"))
.unwrap();
});
let result = ensure_rx.recv_timeout(Duration::from_secs(2));
release_writer_tx.send(()).unwrap();
writer.join().unwrap();
ensure.join().unwrap();
assert_eq!(
result
.expect("clean identity preflight blocked on writer")
.unwrap(),
0
);
}
#[test]
fn repeated_forced_rebuild_skips_unchanged_derived_rows() {
let store = LibraryStore::open_in_memory();
src-tauri/crates/psysonic-library/src/payload.rs
+17
View File
@@ -138,6 +138,11 @@ impl LibrarySyncIdlePayload {
self.job_id = Some(job_id.to_string());
self
}
pub fn mark_failed(&mut self, message: impl Into<String>) {
self.ok = false;
self.error = Some(message.into());
}
}
#[cfg(test)]
@@ -175,6 +180,18 @@ mod tests {
assert!(background.get("jobId").is_none());
}
#[test]
fn idle_payload_failure_preserves_job_context() {
let mut payload =
LibrarySyncIdlePayload::ok("s1", "scope", "delta_sync", "foreground")
.with_job_id("job-1");
payload.mark_failed("identity maintenance failed");
assert!(!payload.ok);
assert_eq!(payload.error.as_deref(), Some("identity maintenance failed"));
assert_eq!(payload.job_id.as_deref(), Some("job-1"));
}
#[test]
fn phase_changed_maps_to_phase_kind() {
let p = LibrarySyncProgressPayload::from_event(
src-tauri/crates/psysonic-library/src/repos/artifact.rs
+81 -50
View File
@@ -48,51 +48,10 @@ impl<'a> ArtifactRepository<'a> {
format: Option<&str>,
now: i64,
) -> Result<Option<TrackArtifactDto>, String> {
if self.store.bulk_ingest_active() {
return self.get_readonly(
server_id,
track_id,
artifact_kind,
source_kind,
source_id,
format,
);
}
self.store
.with_conn_mut("artifact.get_gc", |conn| {
// Lazy TTL cleanup, scoped to the looked-up kind.
conn.execute(
"DELETE FROM track_artifact \
WHERE server_id = ?1 AND track_id = ?2 AND artifact_kind = ?3 \
AND expires_at IS NOT NULL AND expires_at < ?4",
params![server_id, track_id, artifact_kind, now],
)?;
Self::query_one(
conn,
server_id,
track_id,
artifact_kind,
source_kind,
source_id,
format,
)
})
.map_err(|e| e.to_string())
}
fn get_readonly(
&self,
server_id: &str,
track_id: &str,
artifact_kind: &str,
source_kind: Option<&str>,
source_id: Option<&str>,
format: Option<&str>,
) -> Result<Option<TrackArtifactDto>, String> {
self.store
let (artifact, has_expired) = self
.store
.with_read_conn(|conn| {
Self::query_one(
let artifact = Self::query_one(
conn,
server_id,
track_id,
@@ -100,11 +59,35 @@ impl<'a> ArtifactRepository<'a> {
source_kind,
source_id,
format,
)
now,
)?;
let has_expired = conn.query_row(
"SELECT EXISTS(SELECT 1 FROM track_artifact \
WHERE server_id = ?1 AND track_id = ?2 AND artifact_kind = ?3 \
AND expires_at IS NOT NULL AND expires_at < ?4)",
params![server_id, track_id, artifact_kind, now],
|row| row.get::<_, bool>(0),
)?;
Ok((artifact, has_expired))
})
.map_err(|e| e.to_string())
.map_err(|e| e.to_string())?;
if has_expired && !self.store.bulk_ingest_active() {
self.store
.with_conn_mut("artifact.get_gc", |conn| {
conn.execute(
"DELETE FROM track_artifact \
WHERE server_id = ?1 AND track_id = ?2 AND artifact_kind = ?3 \
AND expires_at IS NOT NULL AND expires_at < ?4",
params![server_id, track_id, artifact_kind, now],
)?;
Ok(())
})
.map_err(|e| e.to_string())?;
}
Ok(artifact)
}
#[allow(clippy::too_many_arguments)]
fn query_one(
conn: &rusqlite::Connection,
server_id: &str,
@@ -113,19 +96,22 @@ impl<'a> ArtifactRepository<'a> {
source_kind: Option<&str>,
source_id: Option<&str>,
format: Option<&str>,
now: i64,
) -> rusqlite::Result<Option<TrackArtifactDto>> {
let mut sql = String::from(
"SELECT server_id, track_id, artifact_kind, format, source_kind, source_id, \
language, content_text, content_bytes, not_found, content_hash, fetched_at, \
expires_at FROM track_artifact \
WHERE server_id = ?1 AND track_id = ?2 AND artifact_kind = ?3",
language, content_text, content_bytes, not_found, content_hash, fetched_at, \
expires_at FROM track_artifact \
WHERE server_id = ?1 AND track_id = ?2 AND artifact_kind = ?3 \
AND (expires_at IS NULL OR expires_at >= ?4)",
);
let mut bound: Vec<Value> = vec![
Value::Text(server_id.to_string()),
Value::Text(track_id.to_string()),
Value::Text(artifact_kind.to_string()),
Value::Integer(now),
];
let mut next = 4;
let mut next = 5;
if let Some(sk) = source_kind {
sql.push_str(&format!(" AND source_kind = ?{next}"));
bound.push(Value::Text(sk.to_string()));
@@ -256,6 +242,9 @@ const UPSERT_ARTIFACT: &str = "INSERT INTO track_artifact \
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{mpsc, Arc};
use std::thread;
use std::time::Duration;
fn artifact(
kind: &str,
@@ -324,6 +313,48 @@ mod tests {
assert_eq!(total, 0, "expired row deleted, not just filtered");
}
#[test]
fn get_without_expired_rows_does_not_wait_for_writer_lock() {
let store = Arc::new(LibraryStore::open_in_memory());
seed_track(&store, "s1", "t1");
ArtifactRepository::new(&store)
.put("s1", "t1", &artifact("lyrics", "plain", "lrclib", "lrclib"), 100)
.unwrap();
let (writer_started_tx, writer_started_rx) = mpsc::channel();
let (release_writer_tx, release_writer_rx) = mpsc::channel();
let writer_store = Arc::clone(&store);
let writer = thread::spawn(move || {
writer_store
.with_conn_mut("test.hold_writer", |_conn| {
writer_started_tx.send(()).unwrap();
release_writer_rx.recv().unwrap();
Ok(())
})
.unwrap();
});
writer_started_rx.recv().unwrap();
let (read_tx, read_rx) = mpsc::channel();
let reader_store = Arc::clone(&store);
let reader = thread::spawn(move || {
read_tx
.send(
ArtifactRepository::new(&reader_store)
.get("s1", "t1", "lyrics", None, None, None, 200),
)
.unwrap();
});
let result = read_rx.recv_timeout(Duration::from_secs(2));
release_writer_tx.send(()).unwrap();
writer.join().unwrap();
reader.join().unwrap();
let artifact = result
.expect("read-only artifact lookup blocked on writer")
.unwrap();
assert!(artifact.is_some());
}
#[test]
fn negative_cache_row_is_returned_until_it_expires() {
let store = LibraryStore::open_in_memory();
src-tauri/crates/psysonic-library/src/repos/fact.rs
+71 -26
View File
@@ -34,33 +34,34 @@ impl<'a> FactRepository<'a> {
fact_kinds: &[String],
now: i64,
) -> Result<Vec<TrackFactDto>, String> {
if self.store.bulk_ingest_active() {
return self.get_readonly(server_id, track_id, fact_kinds);
}
self.store
.with_conn_mut("fact.get_gc", |conn| {
// Lazy TTL cleanup for this track.
conn.execute(
"DELETE FROM track_fact \
let (facts, has_expired) = self
.store
.with_read_conn(|conn| {
let facts = Self::query_facts(conn, server_id, track_id, fact_kinds, now)?;
let has_expired = conn.query_row(
"SELECT EXISTS(SELECT 1 FROM track_fact \
WHERE server_id = ?1 AND track_id = ?2 \
AND expires_at IS NOT NULL AND expires_at < ?3",
AND expires_at IS NOT NULL AND expires_at < ?3)",
params![server_id, track_id, now],
|row| row.get::<_, bool>(0),
)?;
Self::query_facts(conn, server_id, track_id, fact_kinds)
Ok((facts, has_expired))
})
.map_err(|e| e.to_string())
}
fn get_readonly(
&self,
server_id: &str,
track_id: &str,
fact_kinds: &[String],
) -> Result<Vec<TrackFactDto>, String> {
self.store
.with_read_conn(|conn| Self::query_facts(conn, server_id, track_id, fact_kinds))
.map_err(|e| e.to_string())
.map_err(|e| e.to_string())?;
if has_expired && !self.store.bulk_ingest_active() {
self.store
.with_conn_mut("fact.get_gc", |conn| {
conn.execute(
"DELETE FROM track_fact \
WHERE server_id = ?1 AND track_id = ?2 \
AND expires_at IS NOT NULL AND expires_at < ?3",
params![server_id, track_id, now],
)?;
Ok(())
})
.map_err(|e| e.to_string())?;
}
Ok(facts)
}
fn query_facts(
@@ -68,16 +69,17 @@ impl<'a> FactRepository<'a> {
server_id: &str,
track_id: &str,
fact_kinds: &[String],
now: i64,
) -> rusqlite::Result<Vec<TrackFactDto>> {
if fact_kinds.is_empty() {
let mut stmt = conn.prepare(SELECT_FACTS)?;
let rows: rusqlite::Result<Vec<TrackFactDto>> = stmt
.query_map(params![server_id, track_id], row_to_fact_dto)?
.query_map(params![server_id, track_id, now], row_to_fact_dto)?
.collect();
rows
} else {
let placeholders = (0..fact_kinds.len())
.map(|i| format!("?{}", i + 3))
.map(|i| format!("?{}", i + 4))
.collect::<Vec<_>>()
.join(", ");
let sql = format!(
@@ -87,6 +89,7 @@ impl<'a> FactRepository<'a> {
let mut bound: Vec<rusqlite::types::Value> = vec![
rusqlite::types::Value::Text(server_id.to_string()),
rusqlite::types::Value::Text(track_id.to_string()),
rusqlite::types::Value::Integer(now),
];
for k in fact_kinds {
bound.push(rusqlite::types::Value::Text(k.clone()));
@@ -163,11 +166,13 @@ fn row_to_fact_dto(row: &rusqlite::Row<'_>) -> rusqlite::Result<TrackFactDto> {
const SELECT_FACTS_BASE: &str = "SELECT server_id, track_id, fact_kind, value_real, value_int, \
value_text, unit, source_kind, source_id, confidence, content_hash, fetched_at, expires_at \
FROM track_fact WHERE server_id = ?1 AND track_id = ?2";
FROM track_fact WHERE server_id = ?1 AND track_id = ?2 \
AND (expires_at IS NULL OR expires_at >= ?3)";
const SELECT_FACTS: &str = "SELECT server_id, track_id, fact_kind, value_real, value_int, \
value_text, unit, source_kind, source_id, confidence, content_hash, fetched_at, expires_at \
FROM track_fact WHERE server_id = ?1 AND track_id = ?2 \
AND (expires_at IS NULL OR expires_at >= ?3) \
ORDER BY fact_kind ASC, fetched_at DESC";
const UPSERT_FACT: &str = "INSERT INTO track_fact \
@@ -187,6 +192,9 @@ const UPSERT_FACT: &str = "INSERT INTO track_fact \
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{mpsc, Arc};
use std::thread;
use std::time::Duration;
fn fact(kind: &str, source_kind: &str, value_int: Option<i64>, expires_at: Option<i64>) -> FactInputDto {
FactInputDto {
@@ -248,6 +256,43 @@ mod tests {
assert_eq!(total, 1);
}
#[test]
fn get_without_expired_rows_does_not_wait_for_writer_lock() {
let store = Arc::new(LibraryStore::open_in_memory());
seed_track(&store, "s1", "t1");
FactRepository::new(&store)
.put("s1", "t1", &fact("bpm", "analysis", Some(120), None), 100)
.unwrap();
let (writer_started_tx, writer_started_rx) = mpsc::channel();
let (release_writer_tx, release_writer_rx) = mpsc::channel();
let writer_store = Arc::clone(&store);
let writer = thread::spawn(move || {
writer_store
.with_conn_mut("test.hold_writer", |_conn| {
writer_started_tx.send(()).unwrap();
release_writer_rx.recv().unwrap();
Ok(())
})
.unwrap();
});
writer_started_rx.recv().unwrap();
let (read_tx, read_rx) = mpsc::channel();
let reader_store = Arc::clone(&store);
let reader = thread::spawn(move || {
read_tx
.send(FactRepository::new(&reader_store).get("s1", "t1", &[], 200))
.unwrap();
});
let result = read_rx.recv_timeout(Duration::from_secs(2));
release_writer_tx.send(()).unwrap();
writer.join().unwrap();
reader.join().unwrap();
let facts = result.expect("read-only fact lookup blocked on writer").unwrap();
assert_eq!(facts.len(), 1);
}
#[test]
fn get_filters_by_kind() {
let store = LibraryStore::open_in_memory();
src-tauri/crates/psysonic-library/src/repos/track.rs
+84 -12
View File
@@ -285,9 +285,11 @@ impl<'a> TrackRepository<'a> {
params![server_id, library_scope, resync_gen, now],
)?
};
crate::browse_projection::rebuild_scope(&tx, server_id, library_scope)?;
crate::identity::prune_cluster_keys_for_scope(&tx, server_id, library_scope)?;
crate::identity::mark_cluster_keys_dirty(&tx, [server_id])?;
if changed > 0 {
crate::browse_projection::rebuild_scope(&tx, server_id, library_scope)?;
crate::identity::prune_cluster_keys_for_scope(&tx, server_id, library_scope)?;
crate::identity::mark_cluster_keys_dirty(&tx, [server_id])?;
}
tx.commit()?;
Ok(changed)
})?;
@@ -619,34 +621,62 @@ impl<'a> TrackRepository<'a> {
let tx = conn.transaction()?;
for chunk in album_ids.chunks(CHUNK) {
let placeholders = (0..chunk.len()).map(|_| "?").collect::<Vec<_>>().join(", ");
let changed_album_sql = format!(
"SELECT DISTINCT album_id FROM track \
WHERE server_id = ? AND deleted = 0 \
AND album_id IN ({placeholders}) \
AND (library_id IS NULL OR library_id = '')"
);
let mut changed_params: Vec<rusqlite::types::Value> =
vec![rusqlite::types::Value::Text(server_id.to_string())];
changed_params.extend(chunk.iter().cloned().map(Into::into));
let changed_album_ids = {
let mut statement = tx.prepare(&changed_album_sql)?;
let rows = statement
.query_map(params_from_iter(changed_params.iter()), |row| row.get(0))?
.collect::<rusqlite::Result<Vec<String>>>()?;
rows
};
if changed_album_ids.is_empty() {
continue;
}
let changed_placeholders = (0..changed_album_ids.len())
.map(|_| "?")
.collect::<Vec<_>>()
.join(", ");
let sql = format!(
"UPDATE track SET library_id = ?1 \
WHERE server_id = ?2 AND deleted = 0 \
AND album_id IN ({placeholders}) \
AND album_id IN ({changed_placeholders}) \
AND (library_id IS NULL OR library_id = '')"
);
let mut params: Vec<rusqlite::types::Value> = vec![
rusqlite::types::Value::Text(library_id.to_string()),
rusqlite::types::Value::Text(server_id.to_string()),
];
for id in chunk {
params.push(id.clone().into());
}
params.extend(changed_album_ids.iter().cloned().map(Into::into));
let n = tx.execute(&sql, params_from_iter(params.iter()))?;
total += n as u64;
tx.execute(
&format!(
"UPDATE track_genre SET library_id = ?1 \
WHERE server_id = ?2 AND track_id IN ( \
SELECT id FROM track WHERE server_id = ?2 \
AND album_id IN ({placeholders}) AND library_id = ?1 \
)"
SELECT id FROM track WHERE server_id = ?2 \
AND album_id IN ({changed_placeholders}) AND library_id = ?1 \
) AND COALESCE(library_id, '') != ?1"
),
params_from_iter(params.iter()),
)?;
crate::identity::refresh_library_ids_for_albums(&tx, server_id, chunk)?;
crate::identity::refresh_library_ids_for_albums(
&tx,
server_id,
&changed_album_ids,
)?;
crate::browse_projection::refresh_library_tagged_albums(
&tx, server_id, library_id, chunk,
&tx,
server_id,
library_id,
&changed_album_ids,
)?;
}
tx.commit()?;
@@ -1210,6 +1240,24 @@ mod tests {
assert_eq!(orphan_deleted, 1);
}
#[test]
fn resync_sweep_with_no_orphans_does_not_rewrite_derived_state() {
let store = LibraryStore::open_in_memory();
let repo = TrackRepository::new(&store);
repo.upsert_batch_initial_ingest_timed(&[row("s1", "seen", "Seen")], Some(2))
.unwrap();
let before = store
.with_conn("test.total_changes", |conn| Ok(conn.total_changes()))
.unwrap();
assert_eq!(repo.sweep_resync_orphans("s1", "", 2).unwrap(), 0);
let after = store
.with_conn("test.total_changes", |conn| Ok(conn.total_changes()))
.unwrap();
assert_eq!(after, before);
}
#[test]
fn scoped_resync_sweep_preserves_other_library_and_refreshes_derived_rows() {
let store = LibraryStore::open_in_memory();
@@ -1441,6 +1489,30 @@ mod tests {
assert_eq!(genre_tagged, 2);
}
#[test]
fn tag_library_by_album_ids_with_no_empty_rows_is_write_free() {
let store = LibraryStore::open_in_memory();
let repo = TrackRepository::new(&store);
let mut tagged = row("s1", "t1", "First");
tagged.library_id = Some("1".into());
tagged.album_id = Some("al1".into());
repo.upsert_batch(&[tagged]).unwrap();
crate::identity::rebuild_cluster_keys(&store, None).unwrap();
let before = store
.with_conn("test.total_changes", |conn| Ok(conn.total_changes()))
.unwrap();
let changed = repo
.tag_library_by_album_ids("s1", "1", &["al1".into()])
.unwrap();
let after = store
.with_conn("test.total_changes", |conn| Ok(conn.total_changes()))
.unwrap();
assert_eq!(changed, 0);
assert_eq!(after, before);
}
#[test]
fn count_untagged_tracks_excludes_deleted_and_populated_rows() {
let store = LibraryStore::open_in_memory();
src-tauri/crates/psysonic-library/src/store.rs
+45 -1
View File
@@ -12,7 +12,7 @@ use tauri::Manager;
///
/// Migration checklist (wiring, data backfill, open/swap path):
/// psysonic-workdocs `ai/agent-rules/08-library-db-migrations.md`.
pub const LIBRARY_DB_SCHEMA_VERSION: i64 = 25;
pub const LIBRARY_DB_SCHEMA_VERSION: i64 = 26;
/// One-time data repair after migration 014 (`artist.name_sort`).
pub(crate) const ARTIST_NAME_SORT_RECONCILE_ID: &str = "artist_name_sort_reconcile_v1";
@@ -86,6 +86,9 @@ pub(crate) const MIGRATION_024_COMPOSER_BROWSE_PROJECTION: &str =
/// Version 25: durable invalidation journal for incremental identity maintenance.
pub(crate) const MIGRATION_025_IDENTITY_INVALIDATION: &str =
include_str!("../migrations/025_identity_invalidation.sql");
/// Version 26: resumable cursor for bounded post-sync library tagging.
pub(crate) const MIGRATION_026_LIBRARY_TAG_CURSOR: &str =
include_str!("../migrations/026_library_tag_cursor.sql");
/// Embedded migrations. Ordered ascending by `version`; the runner sorts
/// defensively before applying so the source order can stay readable.
@@ -105,6 +108,7 @@ const MIGRATIONS: &[(i64, &str)] = &[
(23, MIGRATION_023_STARRED_BROWSE_INDEXES),
(24, MIGRATION_024_COMPOSER_BROWSE_PROJECTION),
(25, MIGRATION_025_IDENTITY_INVALIDATION),
(26, MIGRATION_026_LIBRARY_TAG_CURSOR),
];
/// Idempotent repair — also runs after the migration runner on every open so
@@ -1819,6 +1823,46 @@ mod tests {
);
}
#[test]
fn migration_026_adds_tag_cursor_without_rewriting_completion_state() {
let conn = Connection::open_in_memory().unwrap();
run_migrations_with(
&conn,
&MIGRATIONS[..MIGRATIONS.len() - 1],
LIBRARY_DB_MIN_COMPATIBLE_VERSION,
handle_breaking_schema_bump,
)
.unwrap();
conn.execute(
"INSERT INTO library_tag_state \
(server_id, folders_hash, last_untagged_count, completed_at) \
VALUES ('s1', 'folders', 7, 123)",
[],
)
.unwrap();
run_migrations(&conn).unwrap();
let state: (String, i64, i64) = conn
.query_row(
"SELECT folders_hash, last_untagged_count, completed_at \
FROM library_tag_state WHERE server_id = 's1'",
[],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.unwrap();
assert_eq!(state, ("folders".into(), 7, 123));
let cursor_table: i64 = conn
.query_row(
"SELECT COUNT(*) FROM sqlite_master \
WHERE type = 'table' AND name = 'library_tag_cursor'",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(cursor_table, 1);
}
#[test]
fn fresh_database_marks_projection_backfills_complete() {
let store = LibraryStore::open_in_memory();
src-tauri/crates/psysonic-library/src/sync/library_tag.rs
+127 -257
View File
@@ -19,6 +19,7 @@ use super::now_unix_ms;
use super::progress::{Progress, ProgressEvent};
const ALBUM_PAGE_SIZE: u32 = 500;
const MAX_ALBUM_LIST_REQUESTS_PER_PASS: u32 = 8;
/// Summary of a library-tagging pass.
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -28,6 +29,7 @@ pub struct TagReport {
pub tracks_tagged: u64,
pub untagged_remaining: u64,
pub skipped: bool,
pub completed: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -36,6 +38,13 @@ pub(crate) struct TagStateRow {
last_untagged_count: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct TagCursorRow {
folders_hash: String,
next_folder_id: String,
next_album_offset: u32,
}
/// Stable fingerprint of the server's music-folder list for gating.
pub(crate) fn folders_hash(folders: &[MusicFolder]) -> String {
let mut pairs: Vec<(String, String)> = folders
@@ -55,11 +64,15 @@ pub(crate) fn folders_hash(folders: &[MusicFolder]) -> String {
pub(crate) fn should_run_tagging_pass(
untagged: u64,
prior: Option<&TagStateRow>,
cursor_active: bool,
folders_hash: &str,
) -> bool {
if untagged == 0 {
return false;
}
if cursor_active {
return true;
}
if let Some(p) = prior {
if p.last_untagged_count == untagged && p.folders_hash == folders_hash {
return false;
@@ -68,6 +81,29 @@ pub(crate) fn should_run_tagging_pass(
true
}
fn read_tag_cursor(
store: &LibraryStore,
server_id: &str,
) -> Result<Option<TagCursorRow>, SyncError> {
store
.with_read_conn(|conn| {
conn.query_row(
"SELECT folders_hash, next_folder_id, next_album_offset \
FROM library_tag_cursor WHERE server_id = ?1",
rusqlite::params![server_id],
|row| {
Ok(TagCursorRow {
folders_hash: row.get(0)?,
next_folder_id: row.get(1)?,
next_album_offset: row.get::<_, i64>(2)?.max(0) as u32,
})
},
)
.optional()
})
.map_err(|e| SyncError::Storage(e.to_string()))
}
fn read_tag_state(store: &LibraryStore, server_id: &str) -> Result<Option<TagStateRow>, SyncError> {
store
.with_read_conn(|conn| {
@@ -86,7 +122,39 @@ fn read_tag_state(store: &LibraryStore, server_id: &str) -> Result<Option<TagSta
.map_err(|e| SyncError::Storage(e.to_string()))
}
fn write_tag_state(
fn write_tag_cursor(
store: &LibraryStore,
server_id: &str,
folders_hash: &str,
next_folder_id: &str,
next_album_offset: u32,
) -> Result<(), SyncError> {
let now = now_unix_ms();
store
.with_conn_mut("library_tag.write_cursor", |conn| {
conn.execute(
"INSERT INTO library_tag_cursor \
(server_id, folders_hash, next_folder_id, next_album_offset, updated_at) \
VALUES (?1, ?2, ?3, ?4, ?5) \
ON CONFLICT(server_id) DO UPDATE SET \
folders_hash = excluded.folders_hash, \
next_folder_id = excluded.next_folder_id, \
next_album_offset = excluded.next_album_offset, \
updated_at = excluded.updated_at",
rusqlite::params![
server_id,
folders_hash,
next_folder_id,
next_album_offset as i64,
now
],
)
})
.map_err(|e| SyncError::Storage(e.to_string()))?;
Ok(())
}
fn write_tag_completion(
store: &LibraryStore,
server_id: &str,
folders_hash: &str,
@@ -94,19 +162,25 @@ fn write_tag_state(
) -> Result<(), SyncError> {
let now = now_unix_ms();
store
.with_conn_mut("library_tag.write_state", |conn| {
conn.execute(
"INSERT INTO library_tag_state (server_id, folders_hash, last_untagged_count, completed_at) \
.with_conn_mut("library_tag.write_completion", |conn| {
let tx = conn.transaction()?;
tx.execute(
"INSERT INTO library_tag_state \
(server_id, folders_hash, last_untagged_count, completed_at) \
VALUES (?1, ?2, ?3, ?4) \
ON CONFLICT(server_id) DO UPDATE SET \
folders_hash = excluded.folders_hash, \
last_untagged_count = excluded.last_untagged_count, \
completed_at = excluded.completed_at",
rusqlite::params![server_id, folders_hash, untagged as i64, now],
)
)?;
tx.execute(
"DELETE FROM library_tag_cursor WHERE server_id = ?1",
rusqlite::params![server_id],
)?;
tx.commit()
})
.map_err(|e| SyncError::Storage(e.to_string()))?;
Ok(())
.map_err(|e| SyncError::Storage(e.to_string()))
}
fn check_cancel(cancel: Option<&Arc<AtomicBool>>) -> Result<(), SyncError> {
@@ -142,6 +216,7 @@ pub async fn tag_library_membership(
tracks_tagged: 0,
untagged_remaining: 0,
skipped: true,
completed: true,
});
}
@@ -156,18 +231,24 @@ pub async fn tag_library_membership(
tracks_tagged: 0,
untagged_remaining: untagged,
skipped: true,
completed: true,
});
}
let mut folders = folders;
folders.sort_by(|a, b| a.id.cmp(&b.id));
let hash = folders_hash(&folders);
let prior = read_tag_state(store, server_id)?;
if !should_run_tagging_pass(untagged, prior.as_ref(), &hash) {
let cursor = read_tag_cursor(store, server_id)?;
let active_cursor = cursor.as_ref().filter(|cursor| cursor.folders_hash == hash);
if !should_run_tagging_pass(untagged, prior.as_ref(), active_cursor.is_some(), &hash) {
return Ok(TagReport {
folders_processed: 0,
albums_processed: 0,
tracks_tagged: 0,
untagged_remaining: untagged,
skipped: true,
completed: true,
});
}
@@ -178,12 +259,31 @@ pub async fn tag_library_membership(
let mut folders_processed = 0u32;
let mut albums_processed = 0u32;
let mut tracks_tagged = 0u64;
let mut requests_made = 0u32;
let mut completed = true;
let (start_folder_index, start_offset) = active_cursor
.and_then(|cursor| {
folders
.iter()
.position(|folder| folder.id == cursor.next_folder_id)
.map(|index| (index, cursor.next_album_offset))
})
.unwrap_or((0, 0));
for folder in &folders {
'folders: for (folder_index, folder) in folders.iter().enumerate().skip(start_folder_index) {
check_cancel(cancel.as_ref())?;
let mut offset = 0u32;
let mut offset = if folder_index == start_folder_index {
start_offset
} else {
0
};
loop {
check_cancel(cancel.as_ref())?;
if requests_made >= MAX_ALBUM_LIST_REQUESTS_PER_PASS {
write_tag_cursor(store, server_id, &hash, &folder.id, offset)?;
completed = false;
break 'folders;
}
let page = subsonic
.get_album_list2(
"alphabeticalByName",
@@ -193,7 +293,12 @@ pub async fn tag_library_membership(
)
.await
.map_err(SyncError::from)?;
requests_made += 1;
if page.is_empty() {
folders_processed += 1;
if let Some(next_folder) = folders.get(folder_index + 1) {
write_tag_cursor(store, server_id, &hash, &next_folder.id, 0)?;
}
break;
}
let album_ids: Vec<String> = page.iter().map(|a| a.id.clone()).collect();
@@ -204,17 +309,23 @@ pub async fn tag_library_membership(
tracks_tagged += tagged;
if page.len() < ALBUM_PAGE_SIZE as usize {
folders_processed += 1;
if let Some(next_folder) = folders.get(folder_index + 1) {
write_tag_cursor(store, server_id, &hash, &next_folder.id, 0)?;
}
break;
}
offset = offset.saturating_add(ALBUM_PAGE_SIZE);
write_tag_cursor(store, server_id, &hash, &folder.id, offset)?;
}
folders_processed += 1;
}
let untagged_remaining = tracks
.count_untagged_tracks(server_id)
.map_err(SyncError::Storage)?;
write_tag_state(store, server_id, &hash, untagged_remaining)?;
if completed {
write_tag_completion(store, server_id, &hash, untagged_remaining)?;
}
Ok(TagReport {
folders_processed,
@@ -222,6 +333,7 @@ pub async fn tag_library_membership(
tracks_tagged,
untagged_remaining,
skipped: false,
completed,
})
}
@@ -247,11 +359,12 @@ pub async fn run_tag_pass_best_effort(
{
Ok(report) if !report.skipped => {
crate::app_eprintln!(
"[library-tag] server `{server_id}`: tagged {} tracks across {} folders ({} albums), {} untagged left",
"[library-tag] server `{server_id}`: tagged {} tracks across {} folders ({} albums), {} untagged left, completed={}",
report.tracks_tagged,
report.folders_processed,
report.albums_processed,
report.untagged_remaining,
report.completed,
);
}
Ok(_) => {}
@@ -265,247 +378,4 @@ pub async fn run_tag_pass_best_effort(
}
#[cfg(test)]
mod tests {
use super::*;
use crate::repos::{TrackRepository, TrackRow};
use crate::store::LibraryStore;
use psysonic_integration::subsonic::{SubsonicClient, SubsonicCredentials};
use serde_json::json;
use wiremock::matchers::{method, path, query_param};
use wiremock::{Mock, MockServer, ResponseTemplate};
fn track_row(server: &str, id: &str, album_id: &str) -> TrackRow {
TrackRow {
server_id: server.into(),
id: id.into(),
title: id.into(),
title_sort: None,
artist: Some("A".into()),
artist_id: Some("ar1".into()),
album: "Al".into(),
album_id: Some(album_id.into()),
album_artist: Some("A".into()),
duration_sec: 100,
track_number: Some(1),
disc_number: Some(1),
year: None,
genre: None,
suffix: None,
bit_rate: None,
size_bytes: None,
cover_art_id: None,
starred_at: None,
user_rating: None,
play_count: None,
played_at: None,
server_path: None,
library_id: None,
isrc: None,
mbid_recording: None,
bpm: None,
replay_gain_track_db: None,
replay_gain_album_db: None,
replay_gain_peak: None,
content_hash: None,
server_updated_at: None,
server_created_at: None,
deleted: false,
synced_at: 1,
raw_json: "{}".into(),
}
}
fn test_client(base: &str) -> SubsonicClient {
SubsonicClient::with_static_credentials(
base.to_string(),
SubsonicCredentials {
username: "u".into(),
token: "t".into(),
salt: "s".into(),
},
reqwest::Client::new(),
)
}
#[test]
fn folders_hash_is_order_independent() {
let a = vec![
MusicFolder {
id: "2".into(),
name: "B".into(),
},
MusicFolder {
id: "1".into(),
name: "A".into(),
},
];
let b = vec![
MusicFolder {
id: "1".into(),
name: "A".into(),
},
MusicFolder {
id: "2".into(),
name: "B".into(),
},
];
assert_eq!(folders_hash(&a), folders_hash(&b));
assert_eq!(folders_hash(&a), "1:A|2:B");
}
#[test]
fn should_run_tagging_pass_gates_no_progress() {
let prior = TagStateRow {
folders_hash: "1:Main".into(),
last_untagged_count: 5,
};
assert!(!should_run_tagging_pass(0, None, "1:Main"));
assert!(!should_run_tagging_pass(5, Some(&prior), "1:Main"));
assert!(should_run_tagging_pass(4, Some(&prior), "1:Main"));
assert!(should_run_tagging_pass(5, Some(&prior), "1:Other"));
}
#[tokio::test(flavor = "multi_thread")]
async fn tag_library_membership_tags_by_album_and_respects_prior_tags() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/rest/getMusicFolders.view"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"subsonic-response": {
"status": "ok",
"musicFolders": {
"musicFolder": [
{ "id": 1, "name": "Main" },
{ "id": 2, "name": "Other" }
]
}
}
})))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/rest/getAlbumList2.view"))
.and(query_param("musicFolderId", "1"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"subsonic-response": {
"status": "ok",
"albumList2": {
"album": [{ "id": "alb-a", "name": "A" }]
}
}
})))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/rest/getAlbumList2.view"))
.and(query_param("musicFolderId", "2"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"subsonic-response": {
"status": "ok",
"albumList2": {
"album": [{ "id": "alb-b", "name": "B" }]
}
}
})))
.mount(&server)
.await;
let store = LibraryStore::open_in_memory();
let mut already = track_row("srv", "t0", "alb-a");
already.library_id = Some("9".into());
TrackRepository::new(&store)
.upsert_batch(&[
track_row("srv", "t1", "alb-a"),
track_row("srv", "t2", "alb-b"),
already,
])
.unwrap();
let report = tag_library_membership(
&store,
&test_client(&server.uri()),
"srv",
None,
Arc::new(super::super::progress::NoopProgress),
false,
)
.await
.unwrap();
assert!(!report.skipped);
assert_eq!(report.folders_processed, 2);
assert_eq!(report.tracks_tagged, 2);
assert_eq!(report.untagged_remaining, 0);
let lib1: String = store
.with_read_conn(|c| {
c.query_row(
"SELECT library_id FROM track WHERE id = 't1'",
[],
|r| r.get(0),
)
})
.unwrap();
let lib2: String = store
.with_read_conn(|c| {
c.query_row(
"SELECT library_id FROM track WHERE id = 't2'",
[],
|r| r.get(0),
)
})
.unwrap();
let kept: String = store
.with_read_conn(|c| {
c.query_row(
"SELECT library_id FROM track WHERE id = 't0'",
[],
|r| r.get(0),
)
})
.unwrap();
assert_eq!(lib1, "1");
assert_eq!(lib2, "2");
assert_eq!(kept, "9");
}
#[tokio::test(flavor = "multi_thread")]
async fn tag_library_membership_skips_when_no_progress_possible() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/rest/getMusicFolders.view"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"subsonic-response": {
"status": "ok",
"musicFolders": {
"musicFolder": { "id": 1, "name": "Main" }
}
}
})))
.expect(1)
.mount(&server)
.await;
let store = LibraryStore::open_in_memory();
TrackRepository::new(&store)
.upsert_batch(&[track_row("srv", "orphan", "no-album")])
.unwrap();
write_tag_state(&store, "srv", "1:Main", 1).unwrap();
let report = tag_library_membership(
&store,
&test_client(&server.uri()),
"srv",
None,
Arc::new(super::super::progress::NoopProgress),
false,
)
.await
.unwrap();
assert!(report.skipped);
assert_eq!(report.albums_processed, 0);
assert_eq!(report.tracks_tagged, 0);
assert_eq!(report.untagged_remaining, 1);
}
}
mod tests;
src-tauri/crates/psysonic-library/src/sync/library_tag/tests.rs
+324
View File
@@ -0,0 +1,324 @@
use super::*;
use crate::repos::{TrackRepository, TrackRow};
use crate::store::LibraryStore;
use psysonic_integration::subsonic::{SubsonicClient, SubsonicCredentials};
use serde_json::json;
use wiremock::matchers::{method, path, query_param};
use wiremock::{Mock, MockServer, ResponseTemplate};
fn track_row(server: &str, id: &str, album_id: &str) -> TrackRow {
TrackRow {
server_id: server.into(),
id: id.into(),
title: id.into(),
title_sort: None,
artist: Some("A".into()),
artist_id: Some("ar1".into()),
album: "Al".into(),
album_id: Some(album_id.into()),
album_artist: Some("A".into()),
duration_sec: 100,
track_number: Some(1),
disc_number: Some(1),
year: None,
genre: None,
suffix: None,
bit_rate: None,
size_bytes: None,
cover_art_id: None,
starred_at: None,
user_rating: None,
play_count: None,
played_at: None,
server_path: None,
library_id: None,
isrc: None,
mbid_recording: None,
bpm: None,
replay_gain_track_db: None,
replay_gain_album_db: None,
replay_gain_peak: None,
content_hash: None,
server_updated_at: None,
server_created_at: None,
deleted: false,
synced_at: 1,
raw_json: "{}".into(),
}
}
fn test_client(base: &str) -> SubsonicClient {
SubsonicClient::with_static_credentials(
base.to_string(),
SubsonicCredentials {
username: "u".into(),
token: "t".into(),
salt: "s".into(),
},
reqwest::Client::new(),
)
}
#[test]
fn folders_hash_is_order_independent() {
let a = vec![
MusicFolder {
id: "2".into(),
name: "B".into(),
},
MusicFolder {
id: "1".into(),
name: "A".into(),
},
];
let b = vec![
MusicFolder {
id: "1".into(),
name: "A".into(),
},
MusicFolder {
id: "2".into(),
name: "B".into(),
},
];
assert_eq!(folders_hash(&a), folders_hash(&b));
assert_eq!(folders_hash(&a), "1:A|2:B");
}
#[test]
fn should_run_tagging_pass_gates_no_progress() {
let prior = TagStateRow {
folders_hash: "1:Main".into(),
last_untagged_count: 5,
};
assert!(!should_run_tagging_pass(0, None, false, "1:Main"));
assert!(!should_run_tagging_pass(5, Some(&prior), false, "1:Main"));
assert!(should_run_tagging_pass(5, Some(&prior), true, "1:Main"));
assert!(should_run_tagging_pass(4, Some(&prior), false, "1:Main"));
assert!(should_run_tagging_pass(5, Some(&prior), false, "1:Other"));
}
#[tokio::test(flavor = "multi_thread")]
async fn tag_library_membership_tags_by_album_and_respects_prior_tags() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/rest/getMusicFolders.view"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"subsonic-response": {
"status": "ok",
"musicFolders": {
"musicFolder": [
{ "id": 1, "name": "Main" },
{ "id": 2, "name": "Other" }
]
}
}
})))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/rest/getAlbumList2.view"))
.and(query_param("musicFolderId", "1"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"subsonic-response": {
"status": "ok",
"albumList2": { "album": [{ "id": "alb-a", "name": "A" }] }
}
})))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/rest/getAlbumList2.view"))
.and(query_param("musicFolderId", "2"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"subsonic-response": {
"status": "ok",
"albumList2": { "album": [{ "id": "alb-b", "name": "B" }] }
}
})))
.mount(&server)
.await;
let store = LibraryStore::open_in_memory();
let mut already = track_row("srv", "t0", "alb-a");
already.library_id = Some("9".into());
TrackRepository::new(&store)
.upsert_batch(&[
track_row("srv", "t1", "alb-a"),
track_row("srv", "t2", "alb-b"),
already,
])
.unwrap();
let report = tag_library_membership(
&store,
&test_client(&server.uri()),
"srv",
None,
Arc::new(super::super::progress::NoopProgress),
false,
)
.await
.unwrap();
assert!(!report.skipped);
assert_eq!(report.folders_processed, 2);
assert_eq!(report.tracks_tagged, 2);
assert_eq!(report.untagged_remaining, 0);
assert!(report.completed);
let read_library = |id: &str| -> String {
store
.with_read_conn(|conn| {
conn.query_row("SELECT library_id FROM track WHERE id = ?1", [id], |row| {
row.get(0)
})
})
.unwrap()
};
assert_eq!(read_library("t1"), "1");
assert_eq!(read_library("t2"), "2");
assert_eq!(read_library("t0"), "9");
}
#[tokio::test(flavor = "multi_thread")]
async fn tag_library_membership_skips_when_no_progress_possible() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/rest/getMusicFolders.view"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"subsonic-response": {
"status": "ok",
"musicFolders": { "musicFolder": { "id": 1, "name": "Main" } }
}
})))
.expect(1)
.mount(&server)
.await;
let store = LibraryStore::open_in_memory();
TrackRepository::new(&store)
.upsert_batch(&[track_row("srv", "orphan", "no-album")])
.unwrap();
write_tag_completion(&store, "srv", "1:Main", 1).unwrap();
let report = tag_library_membership(
&store,
&test_client(&server.uri()),
"srv",
None,
Arc::new(super::super::progress::NoopProgress),
false,
)
.await
.unwrap();
assert!(report.skipped);
assert_eq!(report.albums_processed, 0);
assert_eq!(report.tracks_tagged, 0);
assert_eq!(report.untagged_remaining, 1);
}
#[tokio::test(flavor = "multi_thread")]
async fn tag_library_membership_resumes_from_persisted_page_cursor() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/rest/getMusicFolders.view"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"subsonic-response": {
"status": "ok",
"musicFolders": { "musicFolder": { "id": 1, "name": "Main" } }
}
})))
.expect(3)
.mount(&server)
.await;
for page_index in 0..MAX_ALBUM_LIST_REQUESTS_PER_PASS {
let offset = page_index * ALBUM_PAGE_SIZE;
let albums = (0..ALBUM_PAGE_SIZE)
.map(|index| json!({ "id": format!("album-{}", offset + index), "name": "A" }))
.collect::<Vec<_>>();
Mock::given(method("GET"))
.and(path("/rest/getAlbumList2.view"))
.and(query_param("musicFolderId", "1"))
.and(query_param("offset", offset.to_string()))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"subsonic-response": {
"status": "ok",
"albumList2": { "album": albums }
}
})))
.expect(1)
.mount(&server)
.await;
}
let resume_offset = MAX_ALBUM_LIST_REQUESTS_PER_PASS * ALBUM_PAGE_SIZE;
Mock::given(method("GET"))
.and(path("/rest/getAlbumList2.view"))
.and(query_param("musicFolderId", "1"))
.and(query_param("offset", resume_offset.to_string()))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"subsonic-response": {
"status": "ok",
"albumList2": { "album": [{ "id": "last-album", "name": "Last" }] }
}
})))
.expect(1)
.mount(&server)
.await;
let store = LibraryStore::open_in_memory();
TrackRepository::new(&store)
.upsert_batch(&[track_row("srv", "orphan", "no-album")])
.unwrap();
let client = test_client(&server.uri());
let progress = Arc::new(super::super::progress::NoopProgress);
let first = tag_library_membership(
&store,
&client,
"srv",
None,
progress.clone(),
false,
)
.await
.unwrap();
assert!(!first.completed);
assert_eq!(first.albums_processed, resume_offset);
let persisted_offset: i64 = store
.with_read_conn(|conn| {
conn.query_row(
"SELECT next_album_offset FROM library_tag_cursor WHERE server_id = 'srv'",
[],
|row| row.get(0),
)
})
.unwrap();
assert_eq!(persisted_offset, i64::from(resume_offset));
let second = tag_library_membership(
&store,
&client,
"srv",
None,
progress.clone(),
false,
)
.await
.unwrap();
assert!(second.completed);
assert_eq!(second.albums_processed, 1);
let cursor_count: i64 = store
.with_read_conn(|conn| {
conn.query_row("SELECT COUNT(*) FROM library_tag_cursor", [], |row| row.get(0))
})
.unwrap();
assert_eq!(cursor_count, 0);
let third = tag_library_membership(&store, &client, "srv", None, progress, false)
.await
.unwrap();
assert!(third.skipped);
}
src-tauri/src/lib.rs
+33 -20
View File
@@ -675,7 +675,7 @@ pub fn run() {
.map_err(|e| format!("library store init failed: {e}"))?;
let runtime = psysonic_library::LibraryRuntime::new(std::sync::Arc::new(store));
app.manage(runtime);
library_identity_maintenance::setup_library_sync_idle_listener(app.handle());
library_identity_maintenance::setup_library_identity_maintenance(app.handle());
let app_for_sched = app.handle().clone();
tauri::async_runtime::spawn(async move {
@@ -779,31 +779,44 @@ pub fn run() {
Ok(report) => {
let identity_store = Arc::clone(&runtime.store);
let identity_server_id = session.server_id.clone();
match tokio::task::spawn_blocking(move || {
psysonic_library::identity::ensure_cluster_keys_built(
&identity_store,
&identity_server_id,
)
})
let identity_error = match tokio::task::spawn_blocking(
move || {
psysonic_library::identity::ensure_cluster_keys_built(
&identity_store,
&identity_server_id,
)
},
)
.await
{
Ok(Ok(_)) => {}
Ok(Err(error)) => crate::app_eprintln!(
"[library-cluster] background maintenance failed server_id={}: {}",
session.server_id,
error
),
Err(error) => crate::app_eprintln!(
"[library-cluster] background maintenance task failed server_id={}: {}",
session.server_id,
error
),
}
if let Some(payload) = scheduler_idle_payload(
Ok(Ok(_)) => None,
Ok(Err(error)) => {
crate::app_eprintln!(
"[library-cluster] background maintenance failed server_id={}: {}",
session.server_id,
error
);
Some(error)
}
Err(error) => {
crate::app_eprintln!(
"[library-cluster] background maintenance task failed server_id={}: {}",
session.server_id,
error
);
Some(error.to_string())
}
};
if let Some(mut payload) = scheduler_idle_payload(
&report,
&session.server_id,
&scope,
) {
if let Some(error) = identity_error {
payload.mark_failed(format!(
"identity maintenance failed: {error}"
));
}
let _ = app_for_session.emit(
psysonic_library::LibrarySyncProgressPayload::IDLE_EVENT_NAME,
&payload,
src-tauri/src/lib_commands/app_api/migration.rs
+7 -2
View File
@@ -23,6 +23,7 @@ const LIBRARY_TABLES: &[ScopedTable] = &[
ScopedTable { table: "track_genre", column: "server_id" },
ScopedTable { table: "artist_artwork_lookup", column: "server_id" },
ScopedTable { table: "library_tag_state", column: "server_id" },
ScopedTable { table: "library_tag_cursor", column: "server_id" },
ScopedTable { table: "entity_user_rating", column: "server_id" },
ScopedTable { table: "album_browse_projection", column: "server_id" },
ScopedTable { table: "composer_album_projection", column: "server_id" },
@@ -810,6 +811,8 @@ mod tests {
include_str!("../../../crates/psysonic-library/migrations/022_artist_name_fold.sql"),
include_str!("../../../crates/psysonic-library/migrations/023_starred_browse_indexes.sql"),
include_str!("../../../crates/psysonic-library/migrations/024_composer_browse_projection.sql"),
include_str!("../../../crates/psysonic-library/migrations/025_identity_invalidation.sql"),
include_str!("../../../crates/psysonic-library/migrations/026_library_tag_cursor.sql"),
] {
conn.execute_batch(migration).expect("apply library migration");
}
@@ -844,8 +847,10 @@ mod tests {
VALUES ('legacy-a', 'track-1', 'Rock', 'album-1');
INSERT INTO artist_artwork_lookup(server_id, artist_id, surface_kind, status, updated_at)
VALUES ('legacy-a', 'artist-1', 'fanart', 'hit', 1);
INSERT INTO library_tag_state(server_id, folders_hash, completed_at)
VALUES ('legacy-a', 'hash', 1);
INSERT INTO library_tag_state(server_id, folders_hash, completed_at)
VALUES ('legacy-a', 'hash', 1);
INSERT INTO library_tag_cursor(server_id, folders_hash, next_folder_id, updated_at)
VALUES ('legacy-a', 'hash', 'folder-1', 1);
INSERT INTO entity_user_rating(server_id, entity_kind, entity_id, rating, fetched_at)
VALUES ('legacy-a', 'track', 'track-1', 5, 1);
INSERT INTO album_browse_projection(
src-tauri/src/library_identity_maintenance.rs
+4 -39
View File
@@ -1,17 +1,10 @@
use std::sync::Arc;
use serde::Deserialize;
use tauri::{AppHandle, Listener, Manager};
use tauri::{AppHandle, Manager};
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct SyncIdlePayload {
server_id: String,
ok: bool,
}
/// Consume durable identity invalidations even when no webview page is mounted.
pub fn setup_library_sync_idle_listener(app: &AppHandle) {
/// Drain durable identity invalidations left by an interrupted prior process.
/// Active sync orchestrators own post-sync maintenance before emitting idle.
pub fn setup_library_identity_maintenance(app: &AppHandle) {
if let Some(runtime) = app.try_state::<psysonic_library::LibraryRuntime>() {
let store = Arc::clone(&runtime.store);
tauri::async_runtime::spawn_blocking(move || {
@@ -20,32 +13,4 @@ pub fn setup_library_sync_idle_listener(app: &AppHandle) {
}
});
}
let app_handle = app.clone();
let _ = app.listen(
psysonic_library::LibrarySyncProgressPayload::IDLE_EVENT_NAME,
move |event| {
let Ok(payload) = serde_json::from_str::<SyncIdlePayload>(event.payload()) else {
return;
};
if !payload.ok {
return;
}
let Some(runtime) = app_handle.try_state::<psysonic_library::LibraryRuntime>() else {
return;
};
let store = Arc::clone(&runtime.store);
tauri::async_runtime::spawn_blocking(move || {
if let Err(error) =
psysonic_library::identity::ensure_cluster_keys_built(&store, &payload.server_id)
{
crate::app_eprintln!(
"[library-cluster] sync-idle maintenance failed server_id={}: {}",
payload.server_id,
error
);
}
});
},
);
}
src/app/MainApp.tsx
-3
View File
@@ -17,7 +17,6 @@ import { initHotCachePrefetch } from '../hotCachePrefetch';
import { initLocalPlaybackInvalidation } from '../localPlaybackInvalidation';
import { initFavoritesOfflineSync } from '@/features/offline/utils/favoritesOfflineSync';
import { initPinnedOfflineSync } from '@/features/offline/utils/pinnedOfflineSync';
import { initClusterRebuildOnSync } from '@/lib/library/clusterRebuildOnSync';
import {
initResumeIncompleteOfflinePins,
scheduleResumeIncompleteOfflinePins,
@@ -130,13 +129,11 @@ export default function MainApp() {
const stopInvalidation = initLocalPlaybackInvalidation();
const stopFavoritesSync = initFavoritesOfflineSync();
const stopPinnedOfflineSync = initPinnedOfflineSync();
const stopClusterRebuild = initClusterRebuildOnSync();
const stopOfflineResume = initResumeIncompleteOfflinePins();
return () => {
stopInvalidation();
stopFavoritesSync();
stopPinnedOfflineSync();
stopClusterRebuild();
stopOfflineResume();
};
}, [migrationReady, serverIdsKey]);
src/lib/library/clusterRebuildOnSync.test.ts
-95
View File
@@ -1,95 +0,0 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import type { LibrarySyncIdlePayload } from '@/lib/api/library/dto';
const rebuildMock = vi.fn();
let idleHandler: ((payload: LibrarySyncIdlePayload) => void) | undefined;
vi.mock('@/generated/bindings', () => ({
commands: {
libraryClusterRebuild: (...args: unknown[]) => rebuildMock(...args),
},
}));
vi.mock('@/lib/api/library/events', () => ({
subscribeLibrarySyncIdle: vi.fn(async (handler: (payload: LibrarySyncIdlePayload) => void) => {
idleHandler = handler;
return () => {
idleHandler = undefined;
};
}),
}));
vi.mock('@/lib/server/serverIndexKey', () => ({
resolveIndexKey: (id: string) => `key:${id}`,
}));
import { initClusterRebuildOnSync } from './clusterRebuildOnSync';
function idlePayload(overrides: Partial<LibrarySyncIdlePayload> = {}): LibrarySyncIdlePayload {
return {
serverId: 'srv-1',
libraryScope: 'default',
kind: 'delta_sync',
ok: true,
error: null,
...overrides,
};
}
describe('initClusterRebuildOnSync', () => {
beforeEach(() => {
rebuildMock.mockReset();
rebuildMock.mockResolvedValue({ status: 'ok', data: 1 });
idleHandler = undefined;
});
it('rebuilds cluster once on ok:true sync idle', async () => {
const stop = initClusterRebuildOnSync();
await Promise.resolve();
idleHandler!(idlePayload());
await vi.waitFor(() => expect(rebuildMock).toHaveBeenCalledOnce());
expect(rebuildMock).toHaveBeenCalledWith('key:srv-1');
stop();
});
it('does not rebuild on ok:false sync idle', async () => {
const stop = initClusterRebuildOnSync();
await Promise.resolve();
idleHandler!(idlePayload({ ok: false, error: 'boom' }));
await Promise.resolve();
expect(rebuildMock).not.toHaveBeenCalled();
stop();
});
it('dedupes concurrent rebuilds for the same server', async () => {
let release!: () => void;
rebuildMock.mockImplementation(
() =>
new Promise(resolve => {
release = () => resolve({ status: 'ok', data: 1 });
}),
);
const stop = initClusterRebuildOnSync();
await Promise.resolve();
idleHandler!(idlePayload());
idleHandler!(idlePayload());
idleHandler!(idlePayload());
expect(rebuildMock).toHaveBeenCalledOnce();
release();
await Promise.resolve();
await Promise.resolve();
idleHandler!(idlePayload());
await vi.waitFor(() => expect(rebuildMock).toHaveBeenCalledTimes(2));
stop();
});
});
src/lib/library/clusterRebuildOnSync.ts
-53
View File
@@ -1,53 +0,0 @@
/**
* Rebuild library-cluster identity keys after each successful library sync so
* multi-library dedup does not use stale precomputed keys.
*/
import { commands } from '@/generated/bindings';
import { subscribeLibrarySyncIdle } from '@/lib/api/library/events';
import { resolveIndexKey } from '@/lib/server/serverIndexKey';
const inFlight = new Map<string, Promise<boolean>>();
export function rebuildClusterForIndexKey(indexKey: string): Promise<boolean> {
const existing = inFlight.get(indexKey);
if (existing) return existing;
const promise = (async () => {
try {
const res = await commands.libraryClusterRebuild(indexKey);
if (res.status === 'error') {
console.warn('[psysonic] libraryClusterRebuild failed:', indexKey, res.error);
return false;
}
return true;
} catch (err) {
console.warn('[psysonic] libraryClusterRebuild error:', indexKey, err);
return false;
}
})();
inFlight.set(indexKey, promise);
void promise.finally(() => {
if (inFlight.get(indexKey) === promise) inFlight.delete(indexKey);
});
return promise;
}
/** Subscribe globally; call the returned fn on teardown (e.g. MainApp unmount). */
export function initClusterRebuildOnSync(): () => void {
let unlisten: (() => void) | undefined;
let stopped = false;
void subscribeLibrarySyncIdle(payload => {
if (!payload.ok) return;
const indexKey = resolveIndexKey(payload.serverId);
void rebuildClusterForIndexKey(indexKey);
}).then(fn => {
if (stopped) fn();
else unlisten = fn;
});
return () => {
stopped = true;
unlisten?.();
unlisten = undefined;
};
}
src/store/offlineLocalLibrarySyncRevision.test.ts
+7 -45
View File
@@ -5,10 +5,6 @@ import { useAuthStore } from '@/store/authStore';
const syncIdleHandlerRef = vi.hoisted(() => ({
current: null as ((payload: LibrarySyncIdlePayload) => void) | null,
}));
const rebuildClusterMock = vi.hoisted(() =>
vi.fn<(indexKey: string) => Promise<boolean>>(async () => true),
);
vi.mock('@/lib/api/library/events', () => ({
subscribeLibrarySyncIdle: vi.fn(async (handler: (payload: LibrarySyncIdlePayload) => void) => {
syncIdleHandlerRef.current = handler;
@@ -18,10 +14,6 @@ vi.mock('@/lib/api/library/events', () => ({
}),
}));
vi.mock('@/lib/library/clusterRebuildOnSync', () => ({
rebuildClusterForIndexKey: (indexKey: string) => rebuildClusterMock(indexKey),
}));
import {
librarySyncRevision,
libraryScopeSyncRevision,
@@ -36,12 +28,10 @@ describe('offlineLocalLibrarySyncRevision', () => {
servers: [{ id: 'srv-a', name: 'A', url: 'https://a.test', username: 'u', password: 'p' }],
});
resetOfflineLocalLibrarySyncRevisionForTests();
rebuildClusterMock.mockReset();
rebuildClusterMock.mockResolvedValue(true);
syncIdleHandlerRef.current = null;
});
it('bumps revision after derived keys are ready for index key and profile id', async () => {
it('bumps revision after Rust publishes derived-ready sync idle', () => {
expect(offlineLocalLibrarySyncRevision('srv-a')).toBe(0);
syncIdleHandlerRef.current?.({
serverId: 'a.test',
@@ -50,8 +40,7 @@ describe('offlineLocalLibrarySyncRevision', () => {
ok: true,
error: null,
});
expect(offlineLocalLibrarySyncRevision('srv-a')).toBe(0);
await vi.waitFor(() => expect(offlineLocalLibrarySyncRevision('srv-a')).toBe(1));
expect(offlineLocalLibrarySyncRevision('srv-a')).toBe(1);
expect(offlineLocalLibrarySyncRevision('a.test')).toBe(1);
expect(librarySyncRevision()).toBe(1);
expect(libraryScopeSyncRevision(['srv-a'])).toBe(1);
@@ -70,7 +59,7 @@ describe('offlineLocalLibrarySyncRevision', () => {
expect(librarySyncRevision()).toBe(0);
});
it('changes the scoped revision when a different selected server completes', async () => {
it('changes the scoped revision when a different selected server completes', () => {
useAuthStore.setState({
servers: [
{ id: 'srv-a', name: 'A', url: 'https://a.test', username: 'u', password: 'p' },
@@ -81,47 +70,20 @@ describe('offlineLocalLibrarySyncRevision', () => {
syncIdleHandlerRef.current?.({
serverId: 'a.test', libraryScope: '', kind: 'delta_sync', ok: true,
});
await vi.waitFor(() => expect(libraryScopeSyncRevision(['srv-a', 'srv-b'])).toBe(1));
expect(libraryScopeSyncRevision(['srv-a', 'srv-b'])).toBe(1);
syncIdleHandlerRef.current?.({
serverId: 'b.test', libraryScope: '', kind: 'delta_sync', ok: true,
});
await vi.waitFor(() => expect(libraryScopeSyncRevision(['srv-a', 'srv-b'])).toBe(2));
expect(libraryScopeSyncRevision(['srv-a', 'srv-b'])).toBe(2);
});
it('bumps revision for successful background sync-idle events', async () => {
it('bumps revision for successful background sync-idle events', () => {
expect(offlineLocalLibrarySyncRevision('srv-a')).toBe(0);
syncIdleHandlerRef.current?.({
serverId: 'a.test', libraryScope: '', kind: 'delta_sync', source: 'background', ok: true,
});
await vi.waitFor(() => expect(offlineLocalLibrarySyncRevision('srv-a')).toBe(1));
expect(offlineLocalLibrarySyncRevision('srv-a')).toBe(1);
expect(librarySyncRevision()).toBe(1);
});
it('does not publish the revision while cluster maintenance is pending', async () => {
let release!: () => void;
rebuildClusterMock.mockImplementationOnce(() => new Promise(resolve => {
release = () => resolve(true);
}));
expect(offlineLocalLibrarySyncRevision('srv-a')).toBe(0);
syncIdleHandlerRef.current?.({
serverId: 'a.test', libraryScope: '', kind: 'delta_sync', ok: true,
});
await Promise.resolve();
expect(offlineLocalLibrarySyncRevision('srv-a')).toBe(0);
release();
await vi.waitFor(() => expect(offlineLocalLibrarySyncRevision('srv-a')).toBe(1));
});
it('does not publish the revision when cluster maintenance fails', async () => {
rebuildClusterMock.mockResolvedValueOnce(false);
expect(offlineLocalLibrarySyncRevision('srv-a')).toBe(0);
syncIdleHandlerRef.current?.({
serverId: 'a.test', libraryScope: '', kind: 'delta_sync', ok: true,
});
await vi.waitFor(() => expect(rebuildClusterMock).toHaveBeenCalledOnce());
await Promise.resolve();
expect(offlineLocalLibrarySyncRevision('srv-a')).toBe(0);
expect(librarySyncRevision()).toBe(0);
});
});
src/store/offlineLocalLibrarySyncRevision.ts
+2 -5
View File
@@ -1,6 +1,5 @@
import { useSyncExternalStore } from 'react';
import { subscribeLibrarySyncIdle } from '@/lib/api/library/events';
import { rebuildClusterForIndexKey } from '@/lib/library/clusterRebuildOnSync';
import { resolveServerIdForIndexKey } from '@/lib/server/serverLookup';
import { resolveIndexKey } from '@/lib/server/serverIndexKey';
@@ -80,10 +79,8 @@ function ensureOfflineLocalLibrarySyncHook(): void {
if (typeof subscribeLibrarySyncIdle !== 'function') return;
void subscribeLibrarySyncIdle(payload => {
if (payload.ok) {
const indexKey = resolveIndexKey(payload.serverId);
void rebuildClusterForIndexKey(indexKey).then(ready => {
if (ready) bumpOfflineLocalLibrarySyncRevision(payload.serverId);
});
// Rust drains identity invalidations before publishing sync-idle.
bumpOfflineLocalLibrarySyncRevision(payload.serverId);
}
});
}