From a6c0af59ff49dd80652499c52abf82162796581c Mon Sep 17 00:00:00 2001 From: zijiren <84728412+zijiren233@users.noreply.github.com> Date: Tue, 11 Aug 2026 21:25:31 +0800 Subject: [PATCH] fix(settings): recover stale version fences (#399) ## Changes - roll back local and Redis reservations when a settings transaction aborts - atomically reconcile stale committed fences against the database version - repair runtime-setting fences during startup and before writes - retry paths now refresh fence state after SQL optimistic-lock conflicts ## Verification - cargo test -p synctv-core --lib - cargo check --workspace - cargo clippy -p synctv-core -p synctv-api-common --all-targets -- -D warnings - real HTTP regression with Redis fence seeded above the database version --- synctv-core/src/cache/consistency.rs | 257 +++++++++++++++++- synctv-core/src/service/playback.rs | 24 +- synctv-core/src/service/room/settings.rs | 36 ++- .../src/service/room_settings/tests.rs | 28 +- synctv-core/src/service/settings.rs | 38 +++ synctv-media-providers/src/fnos/client.rs | 177 +++++++++++- synctv-media-providers/src/fnos/types.rs | 3 + 7 files changed, 533 insertions(+), 30 deletions(-) diff --git a/synctv-core/src/cache/consistency.rs b/synctv-core/src/cache/consistency.rs index 4997acd4..414665e5 100644 --- a/synctv-core/src/cache/consistency.rs +++ b/synctv-core/src/cache/consistency.rs @@ -184,6 +184,21 @@ pub trait VersionFenceStore: Send + Sync { async fn set_version_at_least(&self, domain: &CacheDomain, version: i64) -> Result; + /// Replace a committed fence version when the fence is still at the + /// expected value and no write reservation is pending. + /// + /// Implementations use an atomic compare-and-set operation. Returning + /// `false` means a concurrent writer changed the fence before the repair + /// could be applied. + async fn replace_version_if_no_pending( + &self, + _domain: &CacheDomain, + _expected_version: i64, + _version: i64, + ) -> Result { + Ok(false) + } + async fn reserve_next_after_observed_version( &self, domain: &CacheDomain, @@ -271,6 +286,7 @@ pub enum FenceRepairDecision { FinalizePending, ExpirePending, AdvanceCommitted, + ResetCommitted, Noop, } @@ -357,6 +373,26 @@ impl VersionFenceStore for LocalVersionFenceStore { Ok(committed) } + async fn replace_version_if_no_pending( + &self, + domain: &CacheDomain, + expected_version: i64, + version: i64, + ) -> Result { + let mut state = self.state.lock(); + if state.pending.contains_key(domain) { + return Ok(false); + } + let Some(current) = state.versions.get_mut(domain) else { + return Ok(expected_version == 0 && version == 0); + }; + if *current != expected_version { + return Ok(false); + } + *current = version; + Ok(true) + } + async fn reserve_next_after_observed_version( &self, domain: &CacheDomain, @@ -452,6 +488,19 @@ impl VersionFenceStore for LocalVersionFenceStore { .is_some_and(|current| current == reservation) { state.pending.remove(domain); + // A local reservation advances the committed value while the + // database transaction is in flight. Roll that reservation back + // when the transaction is aborted so the fence remains aligned + // with the last committed database version. + if state + .versions + .get(domain) + .is_some_and(|current| *current == reservation.version) + { + state + .versions + .insert(domain.clone(), reservation.version.saturating_sub(1)); + } } Ok(()) } @@ -676,6 +725,54 @@ impl VersionFenceStore for RedisVersionFenceStore { .map_err(Error::from) } + async fn replace_version_if_no_pending( + &self, + domain: &CacheDomain, + expected_version: i64, + version: i64, + ) -> Result { + static REPLACE_IF_NO_PENDING: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + redis::Script::new( + r" + local pending_version = redis.call('HGET', KEYS[2], 'version') + if pending_version ~= false then + return 0 + end + + local raw_current = redis.call('GET', KEYS[1]) + local current = tonumber(raw_current or '0') + local expected = tonumber(ARGV[1]) + local replacement = tonumber(ARGV[2]) + if not expected or not replacement or current ~= expected then + return 0 + end + + redis.call('SET', KEYS[1], replacement) + return 1 + ", + ) + }); + + let key = self.key(domain); + let pending_key = format!("{key}:pending"); + let mut conn = self.conn("replace cache version fence").await?; + let replaced = tokio::time::timeout( + self.timeout(), + REPLACE_IF_NO_PENDING + .key(key) + .key(pending_key) + .arg(expected_version) + .arg(version) + .invoke_async::(&mut conn), + ) + .await + .map_err(|_| Error::Timeout("Redis timeout: replace cache version fence".to_string()))? + .map_err(Error::from)?; + + Ok(replaced == 1) + } + async fn reserve_next_after_observed_version( &self, domain: &CacheDomain, @@ -875,6 +972,11 @@ impl VersionFenceStore for RedisVersionFenceStore { local pending_version = tonumber(pending_version_raw) if pending_version == requested_version and pending_token == requested_token then redis.call('DEL', KEYS[1]) + local committed_raw = redis.call('GET', KEYS[2]) + local committed = tonumber(committed_raw or '0') + if committed == requested_version then + redis.call('SET', KEYS[2], requested_version - 1) + end return 1 end @@ -890,6 +992,7 @@ impl VersionFenceStore for RedisVersionFenceStore { self.timeout(), ABORT_WRITE .key(pending_key) + .key(key) .arg(reservation.version) .arg(&reservation.token) .invoke_async::(&mut conn), @@ -1036,7 +1139,40 @@ impl ConsistencyCoordinator { let result = self.fence_store.begin_write(domain, observed_version).await; Self::record_result(domain, "begin_write", &result); - result.map(Some) + match result { + Ok(reservation) => Ok(Some(reservation)), + Err(error @ Error::OptimisticLockConflict) => { + // A committed fence can remain ahead of the database after an + // interrupted reservation finalization. Repair that state only + // when there is no active pending writer and use a CAS so a + // concurrent writer keeps ownership of the fence. + let Ok(Some(state)) = self.fence_store.current_state(domain).await else { + return Err(error); + }; + if state.pending_version.is_some() || state.committed_version <= observed_version { + return Err(error); + } + + let replaced = self + .fence_store + .replace_version_if_no_pending( + domain, + state.committed_version, + observed_version, + ) + .await; + match replaced { + Ok(true) => { + Self::record_success(domain, "begin_repair"); + let retry = self.fence_store.begin_write(domain, observed_version).await; + Self::record_result(domain, "begin_write_retry", &retry); + retry.map(Some) + } + Ok(false) | Err(_) => Err(error), + } + } + Err(error) => Err(error), + } } pub async fn commit_observed_write(&self, domain: &CacheDomain, version: i64) -> Result { @@ -1216,6 +1352,21 @@ impl ConsistencyCoordinator { Err(error) => Self::record_repair_error(domain, &error), } } + FenceRepairDecision::ResetCommitted => { + match self + .fence_store + .replace_version_if_no_pending( + domain, + state.committed_version, + db_version, + ) + .await + { + Ok(true) => Self::record_repair(domain, "reset_committed"), + Ok(false) => Self::record_repair(domain, "reset_skipped"), + Err(error) => Self::record_repair_error(domain, &error), + } + } FenceRepairDecision::KeepPending => { Self::record_repair(domain, "pending_still_ahead"); } @@ -1299,6 +1450,9 @@ impl ConsistencyCoordinator { if state.committed_version < db_version { return FenceRepairDecision::AdvanceCommitted; } + if state.committed_version > db_version { + return FenceRepairDecision::ResetCommitted; + } FenceRepairDecision::Noop } @@ -1696,6 +1850,51 @@ mod tests { ); } + #[tokio::test] + async fn local_version_fence_abort_restores_reserved_version() { + let store = LocalVersionFenceStore::new(); + let domain = room_settings_domain(1); + store + .set_version_at_least(&domain, 7) + .await + .checked("seed version should succeed"); + + let reservation = store + .begin_write(&domain, 7) + .await + .checked("reservation should succeed"); + assert_eq!(reservation.version, 8); + assert_eq!( + store + .current_version(&domain) + .await + .checked("version read should succeed"), + Some(7) + ); + + store + .abort_write(&domain, &reservation) + .await + .checked("abort should succeed"); + + assert_eq!( + store + .current_version(&domain) + .await + .checked("version read should succeed"), + Some(7) + ); + assert_eq!( + store + .current_state(&domain) + .await + .checked("state read should succeed") + .checked("state should exist") + .pending_version, + None + ); + } + #[tokio::test] async fn coordinator_hides_cache_fence_while_local_write_is_pending() { let store = Arc::new(LocalVersionFenceStore::new()); @@ -1899,6 +2098,35 @@ mod tests { ); } + #[tokio::test] + async fn coordinator_repair_resets_stale_committed_fence_to_database_version() { + let store = Arc::new(LocalVersionFenceStore::new()); + let coordinator = ConsistencyCoordinator::new(store.clone()); + let domain = room_settings_domain(1); + + store + .set_version_at_least(&domain, 2) + .await + .checked("stale fence should be seeded"); + coordinator.repair_after_db_read(&domain, 0).await; + + let state = coordinator + .current_state(&domain) + .await + .checked("operation should succeed") + .checked("operation should succeed"); + assert_eq!(state.committed_version, 0); + assert_eq!(state.pending_version, None); + assert_eq!( + coordinator + .begin_observed_write(&domain, 0) + .await + .checked("a repaired fence should allow the next write") + .map(|reservation| reservation.version), + Some(1) + ); + } + #[tokio::test] async fn coordinator_repair_finalizes_committed_pending_even_when_expired() { let store = Arc::new(LocalVersionFenceStore::new()); @@ -1995,6 +2223,33 @@ mod tests { assert_eq!(state.pending_token.as_deref(), Some(first.token.as_str())); } + #[tokio::test] + async fn begin_repairs_committed_fence_left_ahead_without_pending_writer() { + let store = Arc::new(LocalVersionFenceStore::new()); + let coordinator = ConsistencyCoordinator::new(store.clone()); + let domain = room_settings_domain(1); + + store + .set_version_at_least(&domain, 7) + .await + .checked("operation should succeed"); + + let reservation = coordinator + .begin_observed_write(&domain, 3) + .await + .checked("stale committed fence should be repaired") + .checked("authoritative local fence should reserve"); + assert_eq!(reservation.version, 4); + + let state = coordinator + .current_state(&domain) + .await + .checked("operation should succeed") + .checked("operation should succeed"); + assert_eq!(state.committed_version, 3); + assert_eq!(state.pending_version, Some(4)); + } + #[tokio::test] async fn local_begin_clears_pending_observed_in_database_before_next_reservation() { let store = Arc::new(LocalVersionFenceStore::new()); diff --git a/synctv-core/src/service/playback.rs b/synctv-core/src/service/playback.rs index 239bfdb5..fdc13912 100644 --- a/synctv-core/src/service/playback.rs +++ b/synctv-core/src/service/playback.rs @@ -655,7 +655,14 @@ impl PlaybackService { let new_version = reservation .as_ref() .map_or(state.version + 1, |reservation| reservation.version); - let mut tx = self.playback_repo.pool().begin().await?; + let mut tx = match self.playback_repo.pool().begin().await { + Ok(tx) => tx, + Err(error) => { + self.abort_playback_write(&state.room_id, reservation.as_ref()) + .await; + return Err(error.into()); + } + }; let result = async { if let Some(position) = previous_progress_position { self.history_repo @@ -912,7 +919,14 @@ impl PlaybackService { let new_version = reservation .as_ref() .map_or(previous.version + 1, |reservation| reservation.version); - let mut tx = self.playback_repo.pool().begin().await?; + let mut tx = match self.playback_repo.pool().begin().await { + Ok(tx) => tx, + Err(error) => { + self.abort_playback_write(&state.room_id, reservation.as_ref()) + .await; + return Err(error.into()); + } + }; let result = async { self.history_repo .save_cursor_position_on_conn(&state.room_id, previous.computed_position(), &mut tx) @@ -978,7 +992,11 @@ impl PlaybackService { .await; let (updated, chat_event) = match result { Ok(result) => { - tx.commit().await?; + if let Err(error) = tx.commit().await { + self.abort_playback_write(&state.room_id, reservation.as_ref()) + .await; + return Err(error.into()); + } result } Err(error) => { diff --git a/synctv-core/src/service/room/settings.rs b/synctv-core/src/service/room/settings.rs index ba266fa6..4db3cdc2 100644 --- a/synctv-core/src/service/room/settings.rs +++ b/synctv-core/src/service/room/settings.rs @@ -211,10 +211,18 @@ impl RoomService { ) .await? }; - let outbox_event = outbox_event_factory + let outbox_event = match outbox_event_factory .as_ref() .map(|factory| factory(settings, new_version)) - .transpose()?; + .transpose() + { + Ok(event) => event, + Err(error) => { + self.abort_room_settings_write(&domain, reservation.as_ref()) + .await; + return Err(error); + } + }; if let Err(error) = self .insert_realtime_outbox_tx(&mut tx, outbox_event.as_ref()) .await @@ -325,10 +333,18 @@ impl RoomService { ) .await? }; - let outbox_event = outbox_event_factory + let outbox_event = match outbox_event_factory .as_ref() .map(|factory| factory(&updated_settings, new_version)) - .transpose()?; + .transpose() + { + Ok(event) => event, + Err(error) => { + self.abort_room_settings_write(&domain, reservation.as_ref()) + .await; + return Err(error); + } + }; if let Err(error) = self .insert_realtime_outbox_tx(&mut tx, outbox_event.as_ref()) .await @@ -456,10 +472,18 @@ impl RoomService { ) .await? }; - let outbox_event = outbox_event_factory + let outbox_event = match outbox_event_factory .as_ref() .map(|factory| factory(&default_settings, new_version)) - .transpose()?; + .transpose() + { + Ok(event) => event, + Err(error) => { + self.abort_room_settings_write(&domain, reservation.as_ref()) + .await; + return Err(error); + } + }; if let Err(error) = self .insert_realtime_outbox_tx(&mut tx, outbox_event.as_ref()) .await diff --git a/synctv-core/src/service/room_settings/tests.rs b/synctv-core/src/service/room_settings/tests.rs index 250fa968..59276a13 100644 --- a/synctv-core/src/service/room_settings/tests.rs +++ b/synctv-core/src/service/room_settings/tests.rs @@ -325,7 +325,7 @@ async fn test_settings_write_uses_redis_allocated_version() { #[tokio::test] #[ignore = "Requires Docker"] -async fn test_settings_reserve_rejects_stale_snapshot_without_advancing_fence() { +async fn test_settings_reserve_repairs_stale_committed_fence() { let (_container, pool) = create_test_pool().await; let user_repo = UserRepository::new(pool.clone()); let room_service = ok( @@ -372,18 +372,22 @@ async fn test_settings_reserve_rejects_stale_snapshot_without_advancing_fence() "concurrent writer should advance fence", ); - let result = service.begin_write(&room.id, stale_observed_version).await; - assert!( - matches!(result, Err(Error::OptimisticLockConflict)), - "stale settings snapshots must retry before reserving a fence version; got {result:?}" - ); + let reservation = ok( + service.begin_write(&room.id, stale_observed_version).await, + "stale committed fence without a pending writer should be repaired", + ) + .expect("authoritative fence should reserve a version"); + assert_eq!(reservation.version, stale_observed_version + 1); + let state = ok( + fence.current_state(&domain).await, + "fence should be readable", + ) + .expect("fence state should exist"); + assert_eq!(state.committed_version, stale_observed_version); + assert_eq!(state.pending_version, Some(stale_observed_version + 1)); assert_eq!( - ok( - fence.current_version(&domain).await, - "fence should be readable" - ), - Some(stale_observed_version + 1), - "failed reservations must not burn additional fence versions" + state.pending_token.as_deref(), + Some(reservation.token.as_str()) ); } diff --git a/synctv-core/src/service/settings.rs b/synctv-core/src/service/settings.rs index 57460ea1..aff0d75f 100644 --- a/synctv-core/src/service/settings.rs +++ b/synctv-core/src/service/settings.rs @@ -177,6 +177,12 @@ impl SettingsService { "Loaded setting '{}.{}' = '{}'", setting.group_name, setting.key, setting.value ); + self.consistency + .repair_after_db_read( + &Self::runtime_setting_domain(&setting.key), + i64::from(setting.version), + ) + .await; self.cache.insert(setting.key.clone(), setting); } @@ -321,6 +327,12 @@ impl SettingsService { } }; let domain = Self::runtime_setting_domain(key); + if database_version.is_none() { + // An absent setting has database version zero. Keep the + // insertion path aligned with that version when an old fence + // remains from a previously removed key. + self.consistency.repair_after_db_read(&domain, 0).await; + } let observed_fence_version = if let Some(version) = database_version { i64::from(version) } else { @@ -419,6 +431,8 @@ impl SettingsService { warn!(%rollback_error, "Failed to roll back conflicting runtime settings batch"); } self.abort_reserved_settings_writes(&fences).await; + self.repair_aborted_settings_fences(repository, &fences) + .await; return Err(Error::OptimisticLockConflict); } Err(error) => { @@ -731,6 +745,30 @@ impl SettingsService { } } + async fn repair_aborted_settings_fences( + &self, + repository: &SettingsRepository, + fences: &[RuntimeSettingWriteFence], + ) { + for fence in fences { + let database_version = match repository.current_version_optional(&fence.key).await { + Ok(Some(version)) => i64::from(version), + Ok(None) => 0, + Err(error) => { + warn!( + key = %fence.key, + error = %error, + "Failed to read runtime setting version after an aborted write" + ); + continue; + } + }; + self.consistency + .repair_after_db_read(&fence.domain, database_version) + .await; + } + } + async fn finalize_committed_write_best_effort( &self, domain: &CacheDomain, diff --git a/synctv-media-providers/src/fnos/client.rs b/synctv-media-providers/src/fnos/client.rs index 6ab5df3b..60c5f3cc 100644 --- a/synctv-media-providers/src/fnos/client.rs +++ b/synctv-media-providers/src/fnos/client.rs @@ -468,19 +468,31 @@ impl FnosSession { } fn map_file(parent: &str, file: RawFile) -> FnosFile { + let RawFile { + name, + uid, + size, + mtim, + btim, + dir, + v, + } = file; let path = if parent.is_empty() { - file.name.clone() + match (v, uid) { + (Some(storage_id), Some(user_id)) => format!("vol{storage_id}/{user_id}/{name}"), + _ => name.clone(), + } } else { - format!("{parent}/{}", file.name) + format!("{parent}/{name}") }; FnosFile { - name: file.name, + name, path, - size: file.size, - modified_at: file.mtim, - created_at: file.btim, - is_dir: file.dir == Some(1), - storage_id: file.v, + size, + modified_at: mtim, + created_at: btim, + is_dir: dir == Some(1), + storage_id: v, } } @@ -618,6 +630,7 @@ mod tests { "vol1/1000/Videos", RawFile { name: "movie.mp4".to_string(), + uid: Some(1000), size: Some(42), mtim: Some(1), btim: Some(2), @@ -629,6 +642,154 @@ mod tests { assert!(!file.is_dir); } + #[test] + fn maps_root_directory_to_an_addressable_fnos_path() { + let file = map_file( + "", + RawFile { + name: "Series".to_string(), + uid: Some(1000), + size: None, + mtim: None, + btim: None, + dir: Some(1), + v: Some(2), + }, + ); + assert_eq!(file.path, "vol2/1000/Series"); + + let nested = map_file( + &file.path, + RawFile { + name: "Season 1".to_string(), + uid: None, + size: None, + mtim: None, + btim: None, + dir: Some(1), + v: None, + }, + ); + assert_eq!(nested.path, "vol2/1000/Series/Season 1"); + } + + #[tokio::test] + async fn lists_nested_directories_with_full_fnos_paths() { + use std::net::SocketAddr; + + use futures_util::{SinkExt, StreamExt}; + use serde_json::Value; + use tokio::net::TcpListener; + use tokio_tungstenite::accept_async; + use tokio_tungstenite::tungstenite::Message; + + async fn serve_connection(stream: tokio::net::TcpStream) { + let mut socket = accept_async(stream) + .await + .expect("test websocket handshake should succeed"); + while let Some(Ok(message)) = socket.next().await { + let Message::Text(text) = message else { + continue; + }; + let text = text.as_str(); + let payload = text + .find('{') + .and_then(|start| serde_json::from_str::(&text[start..]).ok()) + .expect("test request should contain JSON"); + let reqid = payload + .get("reqid") + .and_then(Value::as_str) + .expect("test request should contain reqid"); + let response = match payload.get("req").and_then(Value::as_str) { + Some("util.crypto.getRSAPub") => { + serde_json::json!({"reqid": reqid, "pub": "unused", "si": "session"}) + } + Some("appcgi.sysinfo.getHostName") => serde_json::json!({ + "reqid": reqid, + "result": "succ", + "data": {"hostName": "test-fnos"} + }), + Some("user.authToken") => { + serde_json::json!({"reqid": reqid, "result": "succ"}) + } + Some("file.ls") => { + let path = payload.get("path").and_then(Value::as_str); + let files = match path { + None => serde_json::json!([{ + "name": "Series", + "uid": 1000, + "dir": 1, + "v": 2 + }]), + Some("vol2/1000/Series") => serde_json::json!([{ + "name": "episode.mkv", + "uid": 1000, + "size": 42 + }]), + Some(other) => panic!("unexpected FNOS path {other}"), + }; + serde_json::json!({ + "reqid": reqid, + "result": "succ", + "files": files, + "uver": 1 + }) + } + request => panic!("unexpected FNOS request {request:?}"), + }; + socket + .send(Message::Text(response.to_string().into())) + .await + .expect("test response should be sent"); + } + } + + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("test websocket listener should bind"); + let address: SocketAddr = listener + .local_addr() + .expect("test websocket listener should have an address"); + let server = tokio::spawn(async move { + for _ in 0..2 { + let (stream, _) = listener + .accept() + .await + .expect("test websocket connection should be accepted"); + serve_connection(stream).await; + } + }); + + let client = FnosClient::new() + .with_ssrf_guard( + synctv_common::ssrf::SsrfGuard::builder() + .allow_private_network_targets(true) + .build(), + ) + .with_timeout(Duration::from_secs(2)); + let endpoints = FnosEndpoints::parse(&format!("ws://{address}")) + .expect("test FNOS endpoint should parse"); + let credential = FnosCredential { + username: "user".to_string(), + password: "password".to_string(), + token: "token".to_string(), + long_token: None, + secret: "c2VjcmV0".to_string(), + }; + let root = client + .list(&endpoints, &credential, None) + .await + .expect("root FNOS listing should succeed"); + assert_eq!(root.files[0].path, "vol2/1000/Series"); + let nested = client + .list(&endpoints, &credential, Some(&root.files[0].path)) + .await + .expect("nested FNOS listing should succeed"); + assert_eq!(nested.files[0].path, "vol2/1000/Series/episode.mkv"); + + server.await.expect("test websocket server should finish"); + } + #[test] fn discovers_nested_webdav_configuration() { let endpoints = diff --git a/synctv-media-providers/src/fnos/types.rs b/synctv-media-providers/src/fnos/types.rs index 262f9233..b59efde3 100644 --- a/synctv-media-providers/src/fnos/types.rs +++ b/synctv-media-providers/src/fnos/types.rs @@ -66,6 +66,9 @@ pub(crate) struct FileListResponse { pub(crate) struct RawFile { #[serde(default)] pub name: String, + /// FNOS returns the owning user id for entries in the user's root. + /// It is required together with `v` to address child directories. + pub uid: Option, pub size: Option, pub mtim: Option, pub btim: Option,