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

@@ -184,6 +184,21 @@ pub trait VersionFenceStore: Send + Sync {
 
     async fn set_version_at_least(&self, domain: &CacheDomain, version: i64) -> Result<i64>;
 
+    /// 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<bool> {
+        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<bool> {
+        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<bool> {
+        static REPLACE_IF_NO_PENDING: std::sync::LazyLock<redis::Script> =
+            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::<i64>(&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::<i64>(&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<i64> {
@@ -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());

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) => {

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

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())
     );
 }
 

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,

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::<Value>(&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 =

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<u64>,
     pub size: Option<u64>,
     pub mtim: Option<i64>,
     pub btim: Option<i64>,