fix(dvr): one entry's deletion no longer destroys another's recording

The previous commit made one file reachable from several library
entries. It did not tell the deletion path. `DeletionTarget` still
documented "the single file this deletion owns" and unlinked
unconditionally, so Alice removing her entry deleted the bytes out from
under Bob's -- his entry survived pointing at a path that no longer
existed, and his playback started 404ing on a recording he never
touched.

The repository has carried reference counting since the split, but
nothing consulted it on the way to `unlink`. This is the same shape as
every other defect in this series: a guard that exists and is tested,
wired to nothing.

`begin_deletion_authorized` now counts the other entries holding the
same media inside the boundary that stamps the task, and
`path_to_unlink` returns `None` when there are any. The bytes go when
the last entry does.

Entries already mid-deletion are deliberately not counted as holders.
Two concurrent deletions that each saw the other would both decline to
unlink and strand the file with nothing pointing at it, which is the one
outcome worse than deleting too eagerly -- it is invisible.

Not changed: private quota is still charged per entry at full size, so
two users sharing a 4 GB film are each charged 4 GB. That is deliberate.
A user's limit must not depend on what strangers happen to record, and
the shared pool cannot double-count because two shared entries on one
programme land in the same pool and are already deduplicated there.

backend/dvr/src/recording/recording_deletion.rs

@@ -120,19 +120,26 @@ pub struct DeletionTarget {
     pub uuid: String,
     pub file_path: PathBuf,
     pub previous_state: DeletionPreviousState,
+    /// Another library entry still points at this file, so removing this
+    /// entry must leave the bytes alone.
+    pub still_referenced: bool,
 }
 
 impl DeletionTarget {
-    /// The single file this deletion owns. `Completed` recordings own
-    /// their final file; `Failed` / `Cancelled` ones never reached
-    /// finalization, so they own the `.partial`.
-    pub fn path_to_unlink(&self) -> PathBuf {
-        match self.previous_state {
+    /// The file this deletion may unlink, or `None` when the entry is one
+    /// of several holding it. `Completed` recordings own their final file;
+    /// `Failed` / `Cancelled` ones never reached finalization, so they own
+    /// the `.partial`.
+    pub fn path_to_unlink(&self) -> Option<PathBuf> {
+        if self.still_referenced {
+            return None;
+        }
+        Some(match self.previous_state {
             DeletionPreviousState::Completed => self.file_path.clone(),
             DeletionPreviousState::Failed | DeletionPreviousState::Cancelled => {
                 crate::recording_worker::recording_partial_path(&self.file_path)
             }
-        }
+        })
     }
 }
 
@@ -182,8 +189,12 @@ where
         // present. Measured/reserved bytes are kept as-is so quota
         // accounting survives a failed or interrupted deletion; they are
         // released when the task is removed in `finalize_deletion`.
-        let target =
-            DeletionTarget { uuid: uuid.to_string(), file_path: task.file_path.clone(), previous_state: prior };
+        let target = DeletionTarget {
+            uuid: uuid.to_string(),
+            file_path: task.file_path.clone(),
+            previous_state: prior,
+            still_referenced: crate::recording::recording_queue::media_is_still_referenced(candidate, uuid),
+        };
         let mut new_meta = meta;
         new_meta.deleting_previous_state = Some(prior);
         apply_meta(candidate, bucket, idx, new_meta);
@@ -340,7 +351,10 @@ pub async fn file_path_for_deletion(download: &RecordingTask, _recording_root: O
 /// success. Returns the path that was unlinked, or `None` if no
 /// physical file was present.
 pub async fn execute_deletion_target(target: &DeletionTarget) -> Result<Option<PathBuf>, DeletionError> {
-    unlink_owned_file(&target.path_to_unlink()).await
+    let Some(path) = target.path_to_unlink() else {
+        return Ok(None);
+    };
+    unlink_owned_file(&path).await
 }
 
 async fn unlink_owned_file(path: &Path) -> Result<Option<PathBuf>, DeletionError> {
@@ -666,6 +680,94 @@ mod tests {
         assert!(queue.revision.load(Ordering::SeqCst) > prior, "revision must advance");
     }
 
+    /// Seed two entries onto the same programme, so they resolve to one
+    /// media identity and one file, and delete the first.
+    async fn delete_one_of_two_entries_sharing(dir: &TempDir) -> (RecordingQueue, PathBuf, DeletionTarget) {
+        let queue = RecordingQueue::new_persistent(dir.path(), dir.path()).expect("open recording repository");
+        let shared_file = dir.path().join("programme.ts");
+        std::fs::write(&shared_file, b"recorded bytes").expect("write file");
+        for (uuid, owner) in [("alice-entry", "web:alice"), ("bob-entry", "web:bob")] {
+            let mut task = finished_with_state(uuid, RecordingTaskState::Completed, None);
+            task.file_path.clone_from(&shared_file);
+            task.recording.owner = RecordingOwner::User(UserId::from(owner));
+            let persisted = RecordingQueue::to_persisted(&task);
+            mutate(&queue, |c| {
+                c.finished.push(persisted);
+                Ok(())
+            })
+            .await
+            .expect("seed");
+        }
+        let target = begin_deletion(&queue, "alice-entry").await.expect("begin");
+        (queue, shared_file, target)
+    }
+
+    #[tokio::test]
+    async fn deleting_one_of_two_entries_leaves_the_shared_file_alone() {
+        // Alice and Bob hold the same recording. Alice removing hers must not
+        // take Bob's copy with it.
+        let dir = TempDir::new().expect("tempdir");
+        let (queue, shared_file, target) = delete_one_of_two_entries_sharing(&dir).await;
+        assert!(target.still_referenced, "Bob still holds this file");
+        assert_eq!(target.path_to_unlink(), None);
+        let unlinked = execute_deletion_target(&target).await.expect("execute");
+        assert_eq!(unlinked, None, "nothing was unlinked");
+        assert!(shared_file.exists(), "Bob's recording must survive Alice's deletion");
+        finalize_deletion(&queue, "alice-entry").await.expect("finalize");
+        let remaining = queue.finished.read().await.clone();
+        assert_eq!(remaining.len(), 1);
+        assert_eq!(remaining[0].uuid, "bob-entry");
+    }
+
+    #[tokio::test]
+    async fn the_last_entry_to_be_deleted_removes_the_file() {
+        let dir = TempDir::new().expect("tempdir");
+        let (queue, shared_file, first) = delete_one_of_two_entries_sharing(&dir).await;
+        execute_deletion_target(&first).await.expect("execute");
+        finalize_deletion(&queue, "alice-entry").await.expect("finalize");
+
+        let second = begin_deletion(&queue, "bob-entry").await.expect("begin");
+        assert!(!second.still_referenced, "nobody else holds it now");
+        let unlinked = execute_deletion_target(&second).await.expect("execute");
+        assert!(unlinked.is_some());
+        assert!(!shared_file.exists(), "the bytes go once the last entry does");
+    }
+
+    #[tokio::test]
+    async fn a_concurrent_deletion_does_not_strand_the_file() {
+        // Both entries are stamped before either unlinks. If each counted the
+        // other as a holder, the file would outlive every entry pointing at it.
+        let dir = TempDir::new().expect("tempdir");
+        let (queue, shared_file, first) = delete_one_of_two_entries_sharing(&dir).await;
+        let second = begin_deletion(&queue, "bob-entry").await.expect("begin");
+        assert!(first.still_referenced);
+        assert!(!second.still_referenced, "an entry already deleting is not a holder");
+        execute_deletion_target(&first).await.expect("execute");
+        execute_deletion_target(&second).await.expect("execute");
+        assert!(!shared_file.exists(), "the file must not be left with nothing pointing at it");
+    }
+
+    #[tokio::test]
+    async fn a_sole_entry_still_removes_its_file() {
+        let dir = TempDir::new().expect("tempdir");
+        let queue = RecordingQueue::new_persistent(dir.path(), dir.path()).expect("open recording repository");
+        let file = dir.path().join("only.ts");
+        std::fs::write(&file, b"bytes").expect("write");
+        let mut task = finished_with_state("only", RecordingTaskState::Completed, None);
+        task.file_path.clone_from(&file);
+        let persisted = RecordingQueue::to_persisted(&task);
+        mutate(&queue, |c| {
+            c.finished.push(persisted);
+            Ok(())
+        })
+        .await
+        .expect("seed");
+        let target = begin_deletion(&queue, "only").await.expect("begin");
+        assert!(!target.still_referenced);
+        execute_deletion_target(&target).await.expect("execute");
+        assert!(!file.exists());
+    }
+
     #[tokio::test]
     async fn begin_deletion_authorized_rejects_when_the_permit_declines() {
         // Authorization runs inside the same mutation boundary that stamps

backend/dvr/src/recording/recording_queue.rs

@@ -429,6 +429,30 @@ fn same_media(left: &PersistedRecordingTask, right: &PersistedRecordingTask) ->
     !left.media_identity.is_empty() && left.media_identity == right.media_identity
 }
 
+/// Every task in the candidate, whichever partition it sits in.
+fn all_tasks(candidate: &PersistedRecordingQueue) -> impl Iterator<Item = &PersistedRecordingTask> {
+    candidate
+        .queue
+        .iter()
+        .chain(candidate.scheduled.iter())
+        .chain(candidate.active.iter())
+        .chain(candidate.finished.iter())
+}
+
+/// Whether an entry other than `uuid` still holds the same file.
+///
+/// Entries already mid-deletion are not counted. Two concurrent deletions
+/// that each saw the other as a holder would both decline to unlink and
+/// leave the file behind with nothing pointing at it.
+pub fn media_is_still_referenced(candidate: &PersistedRecordingQueue, uuid: &str) -> bool {
+    let Some(subject) = all_tasks(candidate).find(|task| task.uuid == uuid) else {
+        return false;
+    };
+    all_tasks(candidate).any(|other| {
+        other.uuid != uuid && other.recording.deleting_previous_state.is_none() && same_media(other, subject)
+    })
+}
+
 pub fn promotion_decision(candidate: &PersistedRecordingQueue, task: &PersistedRecordingTask) -> PromotionDecision {
     if candidate.active.as_ref().is_some_and(|active| same_media(active, task)) {
         return PromotionDecision::Wait;