euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-31 11:12:36 +01:00
Code Issues Packages Projects Releases 5 Wiki Activity

feat(dvr): record shared media once and link it twice

Browse Source 
Two users asking for the same film produced two downloads of the same
bytes. Deduplication was scoped to the requester on purpose: with a
single physical file per request, telling the second user "duplicate"
would have left them with nothing.

The split made one file reachable from several library entries, so that
constraint is gone -- but promotion still took the head of the queue
unconditionally. Two entries on one file would each have been promoted
and each run a transfer to the same path: two writers, one file.

Promotion is now materialization-aware. Before an entry runs,
`promotion_decision` asks what else holds its media:

  Execute    nothing does -- transfer it
  AttachTo   another entry already finished it -- adopt the file
  Wait       another entry is producing it -- do not open a second writer

`attach_to_completed` copies only the physical result. The owner,
visibility and quota stay with the entry: this is one user's link onto a
file another user's request happened to produce.

With that in place `candidate_has_duplicate_recording` narrows from
"anyone already asked for this" to "this quota pool already asked for
this", and `RecordingIdentity` drops its owner and visibility fields --
the same URL over the same window is the same bytes, whoever asked.

An empty media identity deliberately matches nothing, not even another
empty one. It means the identity could not be resolved, and treating two
unidentified requests as one recording would merge two different files.

Not changed: quota is still charged per entry against the full size, so
a shared file is counted once per attached entry rather than once
physically. That is the next step.
This commit is contained in:
DarkBreakpoint
2026-09-01 10:00:15 -05:00
parent 525ed829e9
commit 22379238ea
3 changed files with 197 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/dvr/src/recording/recording_queue.rs
+156
View File
@@ -405,6 +405,68 @@ fn stem_group(recording_cfg: &RecordingConfig, file_stem: &str) -> String {
CONSTANTS.re_remove_filename_ending.replace(stem, "").into_owned()
}
/// What to do with a queued task whose media another entry may already hold.
///
/// Several library entries can reference one physical file. Promoting each of
/// them would start a second transfer to the same path, so a queued entry that
/// is not the one producing the file must attach to it or wait for it.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum PromotionDecision {
/// Nothing else holds this media; run the transfer.
Execute,
/// A completed entry at this index in `finished` already holds it.
AttachTo(usize),
/// Another entry is producing it right now.
Wait,
}
/// Two tasks refer to the same media when they carry the same identity.
///
/// An empty identity never matches, including another empty one: it means the
/// identity could not be resolved, and guessing that two unidentified requests
/// are the same recording would merge two different files.
fn same_media(left: &PersistedRecordingTask, right: &PersistedRecordingTask) -> bool {
!left.media_identity.is_empty() && left.media_identity == right.media_identity
}
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;
}
if let Some(index) =
candidate.finished.iter().position(|done| done.state == RecordingTaskState::Completed && same_media(done, task))
{
return PromotionDecision::AttachTo(index);
}
PromotionDecision::Execute
}
/// Adopt an already-produced file instead of transferring it again.
///
/// Only the physical result is copied. The owner, visibility and quota belong
/// to the entry and must survive: this is one user's link to a file another
/// user's request happened to produce.
pub fn attach_to_completed(task: &mut PersistedRecordingTask, source: &PersistedRecordingTask) {
task.file_dir.clone_from(&source.file_dir);
task.file_path.clone_from(&source.file_path);
task.filename.clone_from(&source.filename);
task.size = source.size;
task.total_size = source.total_size;
task.finished = true;
task.paused = false;
task.error = None;
task.state = RecordingTaskState::Completed;
task.next_retry_at = None;
task.retry_attempts = 0;
task.recording.relative_path.clone_from(&source.recording.relative_path);
task.recording.partial_relative_path = None;
task.recording.measured_bytes = source.recording.measured_bytes;
task.recording.completed_at = source.recording.completed_at;
// The reservation is released: the bytes are already on disk and charged
// to this entry as measured, not reserved.
task.recording.reserved_bytes = 0;
}
fn generate_recording_task_id() -> String {
let now_nanos = SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |duration| duration.as_nanos());
let counter = RECORDING_TASK_ID_COUNTER.fetch_add(1, Ordering::Relaxed);
@@ -1353,6 +1415,100 @@ mod tests {
assert!(queue.finished.read().await.is_empty());
}
fn identified(uuid: &str, identity: &str, state: RecordingTaskState) -> PersistedRecordingTask {
let mut persisted = RecordingQueue::to_persisted(&task(uuid, RecordingKind::Vod, state));
persisted.media_identity = identity.to_owned();
persisted
}
#[test]
fn a_queued_entry_runs_when_nothing_else_holds_its_media() {
let candidate = PersistedRecordingQueue::default();
let queued = identified("a", "film-42", RecordingTaskState::Queued);
assert_eq!(promotion_decision(&candidate, &queued), PromotionDecision::Execute);
}
#[test]
fn a_queued_entry_waits_while_another_entry_produces_the_file() {
// Promoting it would start a second transfer to the same path.
let candidate = PersistedRecordingQueue {
active: Some(identified("a", "film-42", RecordingTaskState::Running)),
..PersistedRecordingQueue::default()
};
let queued = identified("b", "film-42", RecordingTaskState::Queued);
assert_eq!(promotion_decision(&candidate, &queued), PromotionDecision::Wait);
}
#[test]
fn a_queued_entry_attaches_to_a_file_another_entry_finished() {
let candidate = PersistedRecordingQueue {
finished: vec![identified("a", "film-42", RecordingTaskState::Completed)],
..PersistedRecordingQueue::default()
};
let queued = identified("b", "film-42", RecordingTaskState::Queued);
assert_eq!(promotion_decision(&candidate, &queued), PromotionDecision::AttachTo(0));
}
#[test]
fn a_failed_sibling_is_not_attached_to() {
// There is no file to adopt, so the entry has to run.
let candidate = PersistedRecordingQueue {
finished: vec![identified("a", "film-42", RecordingTaskState::Failed)],
..PersistedRecordingQueue::default()
};
let queued = identified("b", "film-42", RecordingTaskState::Queued);
assert_eq!(promotion_decision(&candidate, &queued), PromotionDecision::Execute);
}
#[test]
fn an_unidentified_entry_never_matches_another() {
// An empty identity means the identity could not be resolved. Treating
// two of them as the same recording would merge two different files.
let candidate = PersistedRecordingQueue {
finished: vec![identified("a", "", RecordingTaskState::Completed)],
active: Some(identified("c", "", RecordingTaskState::Running)),
..PersistedRecordingQueue::default()
};
let queued = identified("b", "", RecordingTaskState::Queued);
assert_eq!(promotion_decision(&candidate, &queued), PromotionDecision::Execute);
}
#[test]
fn different_media_does_not_attach() {
let candidate = PersistedRecordingQueue {
finished: vec![identified("a", "film-42", RecordingTaskState::Completed)],
..PersistedRecordingQueue::default()
};
let queued = identified("b", "film-99", RecordingTaskState::Queued);
assert_eq!(promotion_decision(&candidate, &queued), PromotionDecision::Execute);
}
#[test]
fn attaching_adopts_the_file_but_keeps_the_entry_its_own() {
let mut source = identified("a", "film-42", RecordingTaskState::Completed);
source.file_path = PathBuf::from("/rec/film.mp4");
source.filename = "film.mp4".to_string();
source.size = 4_096;
source.recording.measured_bytes = 4_096;
source.recording.completed_at = Some(1_700_000_000);
source.recording.relative_path = Some("film.mp4".to_string());
let mut attached = identified("b", "film-42", RecordingTaskState::Queued);
attached.recording.owner = RecordingOwner::User(UserId::from("web:bob"));
attached.recording.reserved_bytes = 9_999;
attach_to_completed(&mut attached, &source);
// The bytes are shared.
assert_eq!(attached.file_path, source.file_path);
assert_eq!(attached.recording.measured_bytes, 4_096);
assert_eq!(attached.state, RecordingTaskState::Completed);
// The link is not: this is Bob's entry onto Alice's file.
assert_eq!(attached.uuid, "b");
assert_eq!(attached.recording.owner_id().0, "web:bob");
// Nothing is reserved any more; the bytes are already on disk.
assert_eq!(attached.recording.reserved_bytes, 0);
}
#[tokio::test]
async fn persisted_queue_round_trips_and_requeues_running_downloads() {
let state_file = temp_state_file("download_state");
backend/dvr/src/recording/recording_service.rs
+18 -29
View File
@@ -1115,35 +1115,19 @@ pub enum RecordingIdentity {
/// `(rule_id, occurrence_key)` are the same recording by
/// definition, whatever their window looks like.
Occurrence { rule_id: String, occurrence_key: String },
/// A concrete programme on a concrete source, per quota pool. The
/// pool dimension is deliberate: a shared copy and a private copy of
/// the same programme are two different recordings that charge two
/// different quotas.
Programme {
target_id: String,
virtual_id: String,
program_start: i64,
program_end: i64,
owner: RecordingOwner,
visibility: RecordingVisibility,
},
/// A concrete programme on a concrete source. Deliberately free of any
/// owner: two users asking for the same programme are asking for the same
/// file, and it is recorded once and linked twice.
Programme { target_id: String, virtual_id: String, program_start: i64, program_end: i64 },
/// No programme metadata at all: every VOD and series transfer, and a
/// Live capture whose programme window is unknown. Falls back to the
/// resolved URL plus the *scheduled* (padded) window, which — unlike
/// `start_at` — is stable across requests inside a currently-airing
/// window.
///
/// Carries the quota pool for the same reason `Programme` does: until one
/// physical file can be shared by several users, deduplicating across
/// principals would mean the second user is told "duplicate" and gets
/// nothing.
Url {
url: String,
scheduled_start: Option<i64>,
scheduled_end: Option<i64>,
owner: RecordingOwner,
visibility: RecordingVisibility,
},
/// Owner-free for the same reason `Programme` is: the same URL over the
/// same window is the same bytes, whoever asked for them.
Url { url: String, scheduled_start: Option<i64>, scheduled_end: Option<i64> },
}
/// A stable, field-named key for the media a request refers to.
@@ -1170,16 +1154,12 @@ fn recording_identity(meta: &RecordingMetadata, url: &str) -> RecordingIdentity
virtual_id: meta.source.virtual_id.clone(),
program_start,
program_end,
owner: meta.owner.clone(),
visibility: meta.visibility,
};
}
RecordingIdentity::Url {
url: url.to_string(),
scheduled_start: meta.scheduled_start,
scheduled_end: meta.scheduled_end,
owner: meta.owner.clone(),
visibility: meta.visibility,
}
}
@@ -1190,12 +1170,16 @@ fn persisted_recording_identity(task: &PersistedRecordingTask) -> RecordingIdent
fn candidate_has_duplicate_recording(candidate: &PersistedRecordingQueue, task: &RecordingTask) -> bool {
let meta = &task.recording;
let identity = recording_identity(meta, task.url.as_str());
// Pending and active tasks are duplicates of anything matching.
let pool = recording_quota::quota_pool_for_task(task);
// Only the *same* principal asking twice is a duplicate. A different
// principal asking for the same media gets their own library entry, which
// attaches to the one physical file rather than producing a second.
let pending_match = candidate
.queue
.iter()
.chain(candidate.scheduled.iter())
.chain(candidate.active.iter())
.filter(|existing| recording_quota::quota_pool_for_task(*existing) == pool)
.map(persisted_recording_identity)
.any(|existing| existing == identity);
if pending_match {
@@ -1209,7 +1193,12 @@ fn candidate_has_duplicate_recording(candidate: &PersistedRecordingQueue, task:
if !matches!(identity, RecordingIdentity::Occurrence { .. }) {
return false;
}
candidate.finished.iter().map(persisted_recording_identity).any(|existing| existing == identity)
candidate
.finished
.iter()
.filter(|existing| recording_quota::quota_pool_for_task(*existing) == pool)
.map(persisted_recording_identity)
.any(|existing| existing == identity)
}
/// Where a recording lives in the candidate snapshot. The first scan
backend/dvr/src/recording/recording_transfer.rs
+23 -4
View File
@@ -967,10 +967,29 @@ async fn promote_next_download(
if candidate.active.is_some() || candidate.queue.is_empty() {
return Ok(None);
}
let next = candidate.queue.remove(0);
let promoted = (next.uuid.clone(), next.filename.clone());
candidate.active = Some(next);
Ok(Some(promoted))
// Walk the queue rather than taking the head unconditionally: an entry
// whose file another entry already produced must attach to it, and one
// whose file is being produced right now has to wait its turn.
let mut index = 0;
while index < candidate.queue.len() {
match crate::recording::recording_queue::promotion_decision(candidate, &candidate.queue[index]) {
crate::recording::recording_queue::PromotionDecision::Execute => {
let next = candidate.queue.remove(index);
let promoted = (next.uuid.clone(), next.filename.clone());
candidate.active = Some(next);
return Ok(Some(promoted));
}
crate::recording::recording_queue::PromotionDecision::AttachTo(completed) => {
let source = candidate.finished[completed].clone();
let mut attached = candidate.queue.remove(index);
crate::recording::recording_queue::attach_to_completed(&mut attached, &source);
candidate.finished.push(attached);
// The queue shrank; re-examine this position.
}
crate::recording::recording_queue::PromotionDecision::Wait => index += 1,
}
}
Ok(None)
})
.await
}