From feec393eee400eb4cba424af8382dbcf680e0850 Mon Sep 17 00:00:00 2001 From: DarkBreakpoint Date: Tue, 1 Sep 2026 11:35:01 -0500 Subject: [PATCH] fix(dvr): finalizing a transfer twice completes it instead of failing it `finalize_http_transfer` links the staged `.partial` to its final path and then removes the partial. Linking is the one step that is not naturally repeatable: `hard_link` answers `AlreadyExists`. A crash between those two steps leaves both files. The task was not yet marked Completed -- that happens after finalization -- so it is requeued on restart, resumes, finds the partial already complete, takes the `Unsatisfiable { complete: true }` branch, and finalizes again. That second call returned `AlreadyExists`, which the caller turns into `Failed`. A recording whose bytes are entirely and correctly on disk was reported as failed, and because retry re-runs the same path it failed that way every time. Finalization is now idempotent. An existing file at the final path is this task's own earlier link -- path reservation gives it sole claim -- so the transfer completes. If the sizes disagree that reasoning does not hold, and it errors loudly instead of publishing bytes nothing checked. The signature narrows from `&RecordingTask` to the two paths it actually used, which is what makes the behaviour testable at all. Also pins a claim I made and had wrong: retention keys on (owner, channel), so two users sharing one file are already in separate groups and each keeps their own entry under keep_last_per_channel. There was nothing to fix; there is now a test saying so. --- .../dvr/src/recording/recording_retention.rs | 15 +++ .../dvr/src/recording/recording_transfer.rs | 111 ++++++++++++++++-- 2 files changed, 116 insertions(+), 10 deletions(-) diff --git a/backend/dvr/src/recording/recording_retention.rs b/backend/dvr/src/recording/recording_retention.rs index fe4ffbcfe..4d0538b18 100644 --- a/backend/dvr/src/recording/recording_retention.rs +++ b/backend/dvr/src/recording/recording_retention.rs @@ -524,6 +524,21 @@ mod tests { assert_eq!(out[0].uuid, "exact"); } + #[test] + fn two_users_sharing_one_file_each_keep_their_own_entry() { + // A shared file is reachable from one entry per user, and each entry + // belongs to its owner's library. `keep_last_per_channel` is a + // per-library budget, so one user's entry must never consume the + // other's allowance. + let config = RetentionConfig { keep_last_per_channel: Some(1), delete_after_days: None }; + let tasks = vec![ + completed("alice-entry", RecordingOwner::User(UserId::from("web:alice")), Some("c1"), Some("Alpha"), 1_000), + completed("bob-entry", RecordingOwner::User(UserId::from("web:bob")), Some("c1"), Some("Alpha"), 1_000), + ]; + let candidates = compute_candidates(&tasks, &config, 2_000); + assert_eq!(candidates.len(), 0, "each library holds one recording and is allowed one"); + } + #[test] fn shared_owner_groups_only_by_channel() { // Two shared recordings on the same channel — count diff --git a/backend/dvr/src/recording/recording_transfer.rs b/backend/dvr/src/recording/recording_transfer.rs index c4f086834..65fc93649 100644 --- a/backend/dvr/src/recording/recording_transfer.rs +++ b/backend/dvr/src/recording/recording_transfer.rs @@ -306,18 +306,40 @@ fn http_transfer_path(task: &RecordingTask) -> std::path::PathBuf { } } -async fn finalize_http_transfer(task: &RecordingTask, transfer_path: &std::path::Path) -> std::io::Result<()> { - if transfer_path == task.file_path { +/// Publish a completed transfer at its final path. +/// +/// Idempotent: linking is the only step that is not naturally repeatable, and a +/// crash between the link and the partial's removal used to leave a recording +/// whose bytes are complete on disk reported as failed, permanently -- a retry +/// re-ran the same path and failed the same way. +async fn finalize_http_transfer(final_path: &std::path::Path, transfer_path: &std::path::Path) -> std::io::Result<()> { + if transfer_path == final_path { return Ok(()); } fs::File::open(transfer_path).await?.sync_all().await?; - fs::hard_link(transfer_path, &task.file_path).await?; + match fs::hard_link(transfer_path, final_path).await { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + // Path reservation gives this recording sole claim on `final_path`, + // so something already there is this task's own earlier link. A + // size mismatch means that reasoning does not hold; say so rather + // than publish bytes that were never checked. + let published = fs::metadata(final_path).await?.len(); + let staged = fs::metadata(transfer_path).await?.len(); + if published != staged { + return Err(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + format!( + "{} already exists with {published} bytes, but the transfer staged {staged}", + final_path.display() + ), + )); + } + } + Err(error) => return Err(error), + } if let Err(error) = fs::remove_file(transfer_path).await { - warn!( - "Finalized {} but could not remove partial {}: {error}", - task.file_path.display(), - transfer_path.display() - ); + warn!("Finalized {} but could not remove partial {}: {error}", final_path.display(), transfer_path.display()); } Ok(()) } @@ -364,7 +386,7 @@ async fn download_file( Ok(()) => {} Err(ResumeValidationError::IgnoredRange) => existing_size = 0, Err(ResumeValidationError::Unsatisfiable { complete: true }) => { - return match finalize_http_transfer(&file_download, &file_path).await { + return match finalize_http_transfer(&file_download.file_path, &file_path).await { Ok(()) => DownloadExecutionResult::Completed, Err(error) => DownloadExecutionResult::Failed(format!( "Could not finalize {}: {error}", @@ -687,7 +709,12 @@ async fn download_file( if let Err(err) = buf_writer.shutdown().await { return DownloadExecutionResult::Failed(err.to_string()); } - return match finalize_http_transfer(&file_download, &file_path).await { + return match finalize_http_transfer( + &file_download.file_path, + &file_path, + ) + .await + { Ok(()) => DownloadExecutionResult::Completed, Err(error) => DownloadExecutionResult::Failed(format!( "Could not finalize {}: {error}", @@ -1755,3 +1782,67 @@ fn start_recording_scheduler( } }); } + +#[cfg(test)] +mod tests { + use super::finalize_http_transfer; + use tempfile::TempDir; + + #[tokio::test] + async fn finalizing_publishes_the_staged_bytes_and_clears_the_partial() { + let dir = TempDir::new().expect("tempdir"); + let partial = dir.path().join("film.mp4.partial"); + let final_path = dir.path().join("film.mp4"); + std::fs::write(&partial, b"complete recording").expect("write"); + + finalize_http_transfer(&final_path, &partial).await.expect("finalize"); + + assert!(final_path.exists()); + assert!(!partial.exists(), "the partial is cleared once published"); + } + + #[tokio::test] + async fn finalizing_twice_succeeds_instead_of_failing_a_complete_recording() { + // A crash between the link and the partial's removal leaves both files. + // The resumed transfer sees a complete file, finalizes again, and used + // to get AlreadyExists back -- reporting a recording whose bytes are + // entirely on disk as failed, on every retry. + let dir = TempDir::new().expect("tempdir"); + let partial = dir.path().join("film.mp4.partial"); + let final_path = dir.path().join("film.mp4"); + std::fs::write(&partial, b"complete recording").expect("write"); + std::fs::hard_link(&partial, &final_path).expect("simulate the interrupted finalize"); + + finalize_http_transfer(&final_path, &partial).await.expect("finalizing again must succeed"); + + assert!(final_path.exists()); + assert!(!partial.exists(), "the second attempt finishes the job the first did not"); + } + + #[tokio::test] + async fn a_different_file_at_the_final_path_is_not_published_over() { + // Path reservation should make this unreachable. If it ever happens the + // staged bytes were never verified against what is already there, so + // the transfer must not claim success. + let dir = TempDir::new().expect("tempdir"); + let partial = dir.path().join("film.mp4.partial"); + let final_path = dir.path().join("film.mp4"); + std::fs::write(&partial, b"the recording we just made").expect("write"); + std::fs::write(&final_path, b"something else entirely, of another size").expect("write"); + + let error = finalize_http_transfer(&final_path, &partial).await.expect_err("must refuse"); + assert_eq!(error.kind(), std::io::ErrorKind::AlreadyExists); + assert!(partial.exists(), "the staged bytes are kept for inspection"); + } + + #[tokio::test] + async fn a_transfer_written_straight_to_its_final_path_needs_no_finalization() { + let dir = TempDir::new().expect("tempdir"); + let final_path = dir.path().join("stream.ts"); + std::fs::write(&final_path, b"live capture").expect("write"); + + finalize_http_transfer(&final_path, &final_path).await.expect("finalize"); + + assert!(final_path.exists()); + } +}