feat(storage): remember which server generation wrote the cached catalog
The cache was version-blind: nothing recorded which Jellyfin generation produced a row, so a server upgraded underneath the app kept serving rows parsed under the previous generation's assumptions. Migration 026 adds servers.catalog_generation and deliberately does NOT clear synced_at the way migration 025 did. The column starts NULL, which reads as "no generation recorded yet" rather than "changed", so the first connection after upgrading simply records what it finds. Invalidation happens only when the recorded generation actually changes. That distinction is the point. Treating absent information as a change would charge every existing user a full catalog re-fetch to defend against a server upgrade that has not happened — and at the time of writing, 12.0 is hours old, so essentially no installed server is on the newer generation at all. Capabilities are also wired at repository creation: the version storage already holds is read once, resolved, and handed to the online repository. A missing or unparseable version is not an error — it resolves to the older generation, whose request shapes work on both. TRACES: UR-085 | IR-035, DR-280, DR-284 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -14,6 +14,7 @@ use uuid::Uuid;
|
||||
|
||||
use crate::domain::rank_search_results;
|
||||
use crate::jellyfin::HttpClient;
|
||||
use crate::repository::capabilities::ServerCapabilities;
|
||||
use crate::repository::{
|
||||
series_progress, types::*, HybridRepository, MediaRepository, OfflineRepository,
|
||||
OnlineRepository, StreamSelection,
|
||||
@@ -63,6 +64,111 @@ impl RepositoryManager {
|
||||
/// Wrapper for Tauri state
|
||||
pub struct RepositoryManagerWrapper(pub RepositoryManager);
|
||||
|
||||
/// Read the server's reported version and resolve it into capabilities.
|
||||
///
|
||||
/// Never fails: a server row that is missing, or carries a version this build
|
||||
/// cannot parse, yields the conservative generation rather than an error. A
|
||||
/// client that refused to start because it did not recognise a version string
|
||||
/// would be the exact failure UR-085 exists to remove.
|
||||
///
|
||||
/// TRACES: UR-085 | IR-035, DR-280
|
||||
async fn server_capabilities(
|
||||
db: &Arc<crate::storage::db_service::RusqliteService>,
|
||||
server_id: &str,
|
||||
) -> ServerCapabilities {
|
||||
use crate::storage::db_service::{DatabaseService, Query, QueryParam};
|
||||
|
||||
let reported: Option<String> = db
|
||||
.query_one(
|
||||
Query::with_params(
|
||||
"SELECT version FROM servers WHERE id = ?1",
|
||||
vec![QueryParam::String(server_id.to_string())],
|
||||
),
|
||||
|row| row.get::<_, Option<String>>(0),
|
||||
)
|
||||
.await
|
||||
.ok()
|
||||
.flatten();
|
||||
|
||||
match reported {
|
||||
Some(version) => ServerCapabilities::from_reported(version.as_str()),
|
||||
None => {
|
||||
debug!("[REPO] No server version recorded for {server_id}; assuming current target");
|
||||
ServerCapabilities::assumed()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Drop the cached catalog if the server changed generation since we last looked.
|
||||
///
|
||||
/// Returns whether anything was invalidated, which is what the tests assert on.
|
||||
///
|
||||
/// The first run after this feature ships records the generation and invalidates
|
||||
/// nothing: a NULL column means "never recorded", not "changed". Making the
|
||||
/// absence of information trigger a full re-fetch would charge every existing
|
||||
/// user bandwidth for a server upgrade that has not happened.
|
||||
///
|
||||
/// TRACES: UR-085 | DR-284
|
||||
async fn invalidate_cache_on_generation_change(
|
||||
db: &Arc<crate::storage::db_service::RusqliteService>,
|
||||
server_id: &str,
|
||||
generation: crate::repository::capabilities::ServerGeneration,
|
||||
) -> bool {
|
||||
use crate::storage::db_service::{DatabaseService, Query, QueryParam};
|
||||
|
||||
let current = format!("{generation:?}");
|
||||
|
||||
let previous: Option<String> = db
|
||||
.query_one(
|
||||
Query::with_params(
|
||||
"SELECT catalog_generation FROM servers WHERE id = ?1",
|
||||
vec![QueryParam::String(server_id.to_string())],
|
||||
),
|
||||
|row| row.get::<_, Option<String>>(0),
|
||||
)
|
||||
.await
|
||||
.ok()
|
||||
.flatten();
|
||||
|
||||
let changed = matches!(previous.as_deref(), Some(prev) if prev != current);
|
||||
|
||||
if changed {
|
||||
warn!(
|
||||
"[REPO] Server generation changed ({:?} -> {}); dropping the cached catalog so it \
|
||||
is re-fetched under the new generation's shapes",
|
||||
previous, current
|
||||
);
|
||||
if let Err(e) = db
|
||||
.execute(Query::with_params(
|
||||
"UPDATE items SET synced_at = NULL WHERE server_id = ?1",
|
||||
vec![QueryParam::String(server_id.to_string())],
|
||||
))
|
||||
.await
|
||||
{
|
||||
// Not fatal: stale-but-parseable rows are better than refusing to
|
||||
// start, and the next successful sync overwrites them anyway.
|
||||
error!("[REPO] Failed to invalidate cached catalog: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
if previous.as_deref() != Some(current.as_str()) {
|
||||
if let Err(e) = db
|
||||
.execute(Query::with_params(
|
||||
"UPDATE servers SET catalog_generation = ?1 WHERE id = ?2",
|
||||
vec![
|
||||
QueryParam::String(current),
|
||||
QueryParam::String(server_id.to_string()),
|
||||
],
|
||||
))
|
||||
.await
|
||||
{
|
||||
error!("[REPO] Failed to record server generation: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
changed
|
||||
}
|
||||
|
||||
/// Create a new repository instance
|
||||
/// Returns a handle (UUID) for accessing the repository
|
||||
#[tauri::command]
|
||||
@@ -100,17 +206,6 @@ pub async fn repository_create(
|
||||
monitor.reporter()
|
||||
};
|
||||
|
||||
// Create online repository wired to connectivity reporting
|
||||
debug!("[REPO] Creating online repository...");
|
||||
let online = OnlineRepository::new(
|
||||
Arc::new(http_client),
|
||||
server_url,
|
||||
user_id.clone(),
|
||||
access_token,
|
||||
)
|
||||
.with_connectivity(connectivity_reporter);
|
||||
debug!("[REPO] Online repository created");
|
||||
|
||||
// Create offline repository with async-safe database service
|
||||
debug!("[REPO] Creating database service...");
|
||||
let db_service = {
|
||||
@@ -123,6 +218,37 @@ pub async fn repository_create(
|
||||
}; // Lock is released here
|
||||
debug!("[REPO] Database service created");
|
||||
|
||||
// Resolve what this server can do, from the version it reported at connect.
|
||||
// `AuthManager::connect_to_server` already parsed it and `storage` already
|
||||
// persisted it, so this costs one indexed read and no extra round trip.
|
||||
//
|
||||
// A missing or unreadable version is not an error: `from_reported` treats it
|
||||
// as the older generation, whose request shapes also work on the newer one.
|
||||
//
|
||||
// TRACES: UR-085 | IR-035, DR-280
|
||||
let capabilities = server_capabilities(&db_service, &server_id).await;
|
||||
info!(
|
||||
"[REPO] Server generation: {:?} (reported {:?})",
|
||||
capabilities.generation,
|
||||
capabilities.version.as_ref().map(|v| v.raw.as_str())
|
||||
);
|
||||
|
||||
// A server upgraded underneath us means the cached catalog was parsed under
|
||||
// a different generation's assumptions. TRACES: UR-085 | DR-284
|
||||
invalidate_cache_on_generation_change(&db_service, &server_id, capabilities.generation).await;
|
||||
|
||||
// Create online repository wired to connectivity reporting
|
||||
debug!("[REPO] Creating online repository...");
|
||||
let online = OnlineRepository::new(
|
||||
Arc::new(http_client),
|
||||
server_url,
|
||||
user_id.clone(),
|
||||
access_token,
|
||||
)
|
||||
.with_connectivity(connectivity_reporter)
|
||||
.with_capabilities(capabilities);
|
||||
debug!("[REPO] Online repository created");
|
||||
|
||||
debug!("[REPO] Creating offline repository...");
|
||||
let offline = OfflineRepository::new(db_service, server_id, user_id);
|
||||
debug!("[REPO] Offline repository created");
|
||||
@@ -1216,3 +1342,97 @@ mod tests {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod generation_change_tests {
|
||||
use super::*;
|
||||
use crate::repository::capabilities::ServerGeneration;
|
||||
use crate::storage::db_service::{DatabaseService, Query, QueryParam, RusqliteService};
|
||||
|
||||
async fn db_with_server() -> Arc<RusqliteService> {
|
||||
let conn = rusqlite::Connection::open_in_memory().expect("in-memory db");
|
||||
for (_, sql) in crate::storage::schema::MIGRATIONS {
|
||||
conn.execute_batch(sql).expect("migration");
|
||||
}
|
||||
let db = Arc::new(RusqliteService::new(Arc::new(std::sync::Mutex::new(conn))));
|
||||
db.execute(Query::with_params(
|
||||
"INSERT INTO servers (id, name, url, version) VALUES (?1, ?2, ?3, ?4)",
|
||||
vec![
|
||||
QueryParam::String("srv-1".into()),
|
||||
QueryParam::String("Home".into()),
|
||||
QueryParam::String("https://example.test".into()),
|
||||
QueryParam::String("10.11.5".into()),
|
||||
],
|
||||
))
|
||||
.await
|
||||
.expect("seed server");
|
||||
db
|
||||
}
|
||||
|
||||
async fn recorded(db: &Arc<RusqliteService>) -> Option<String> {
|
||||
db.query_one(
|
||||
Query::new("SELECT catalog_generation FROM servers WHERE id = 'srv-1'"),
|
||||
|row| row.get::<_, Option<String>>(0),
|
||||
)
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
}
|
||||
|
||||
/// The first look records the generation and invalidates nothing. A NULL
|
||||
/// column means "never recorded", not "changed" — treating it as a change
|
||||
/// would charge every existing user a full re-fetch on upgrade.
|
||||
///
|
||||
/// TRACES: UR-085 | DR-284
|
||||
#[tokio::test]
|
||||
async fn the_first_look_records_without_invalidating() {
|
||||
let db = db_with_server().await;
|
||||
let invalidated =
|
||||
invalidate_cache_on_generation_change(&db, "srv-1", ServerGeneration::V10_11).await;
|
||||
|
||||
assert!(!invalidated, "a first sighting is not a change");
|
||||
assert_eq!(recorded(&db).await.as_deref(), Some("V10_11"));
|
||||
}
|
||||
|
||||
/// Seeing the same generation again is not a change either.
|
||||
///
|
||||
/// TRACES: UR-085 | DR-284
|
||||
#[tokio::test]
|
||||
async fn an_unchanged_generation_does_not_invalidate() {
|
||||
let db = db_with_server().await;
|
||||
invalidate_cache_on_generation_change(&db, "srv-1", ServerGeneration::V10_11).await;
|
||||
let invalidated =
|
||||
invalidate_cache_on_generation_change(&db, "srv-1", ServerGeneration::V10_11).await;
|
||||
|
||||
assert!(!invalidated);
|
||||
assert_eq!(recorded(&db).await.as_deref(), Some("V10_11"));
|
||||
}
|
||||
|
||||
/// An actual upgrade drops the cached catalog and records the new
|
||||
/// generation, so the next browse re-fetches under the new shapes.
|
||||
///
|
||||
/// TRACES: UR-085 | DR-284
|
||||
#[tokio::test]
|
||||
async fn a_real_upgrade_invalidates_and_records() {
|
||||
let db = db_with_server().await;
|
||||
invalidate_cache_on_generation_change(&db, "srv-1", ServerGeneration::V10_11).await;
|
||||
|
||||
let invalidated =
|
||||
invalidate_cache_on_generation_change(&db, "srv-1", ServerGeneration::V12Plus).await;
|
||||
|
||||
assert!(invalidated, "10.11 -> 12.x is a generation change");
|
||||
assert_eq!(recorded(&db).await.as_deref(), Some("V12Plus"));
|
||||
}
|
||||
|
||||
/// A server row that is missing entirely must not panic or invalidate.
|
||||
///
|
||||
/// TRACES: UR-085 | DR-284
|
||||
#[tokio::test]
|
||||
async fn an_unknown_server_is_harmless() {
|
||||
let db = db_with_server().await;
|
||||
let invalidated =
|
||||
invalidate_cache_on_generation_change(&db, "no-such-server", ServerGeneration::V12Plus)
|
||||
.await;
|
||||
assert!(!invalidated);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -30,6 +30,7 @@ pub const MIGRATIONS: &[(&str, &str)] = &[
|
||||
("023_downloads_expiry", MIGRATION_023),
|
||||
("024_multi_user_profiles", MIGRATION_024),
|
||||
("025_backfill_item_library_id", MIGRATION_025),
|
||||
("026_server_catalog_generation", MIGRATION_026),
|
||||
];
|
||||
|
||||
/// Initial schema migration
|
||||
@@ -921,6 +922,25 @@ const MIGRATION_025: &str = r#"
|
||||
UPDATE items SET synced_at = NULL;
|
||||
"#;
|
||||
|
||||
/// Remember which server generation wrote the cached catalog.
|
||||
///
|
||||
/// The cache was version-blind: nothing recorded which Jellyfin generation
|
||||
/// produced a row, so a server upgraded underneath the app kept serving rows
|
||||
/// parsed under the previous generation's assumptions.
|
||||
///
|
||||
/// This deliberately does **not** clear `synced_at` the way MIGRATION_025 did.
|
||||
/// The column starts NULL, which reads as "no generation recorded yet", and the
|
||||
/// first connection after upgrading simply records what it finds. Invalidation
|
||||
/// happens only when the recorded generation actually *changes* — punishing
|
||||
/// every existing user with a full re-fetch for a server upgrade that has not
|
||||
/// happened would cost real bandwidth to defend against nothing. At the time of
|
||||
/// writing no installed server is on the newer generation at all.
|
||||
///
|
||||
/// TRACES: UR-085 | DR-284
|
||||
const MIGRATION_026: &str = r#"
|
||||
ALTER TABLE servers ADD COLUMN catalog_generation TEXT;
|
||||
"#;
|
||||
|
||||
#[cfg(test)]
|
||||
mod migration_024_tests {
|
||||
use super::*;
|
||||
|
||||
Reference in New Issue
Block a user