fix(dvr): re-check quota and disk before opening a destination

Task 10 step 7. Admission runs when the request is accepted, which can
be a long time before anything is written: the recording may wait for
provider capacity, sit through a retry backoff, or be scheduled hours
ahead. The quota and free space it was admitted against are not the ones
it is about to consume.

Until now nothing looked again. The first sign that the disk had filled
while a recording waited was a write failure part-way through it --
after a provider slot had been taken and, for Live, after ffmpeg had
started against a broadcast that only happens once.

The check runs at the last point before a destination is opened. The
subject is excluded from both the quota and the disk sums and then added
back as the candidate charge, so it is not counted twice. An
unmeasurable root is not grounds to refuse, matching admission.

A refusal releases the provider slot, commits a terminal Failed naming
which resource ran out, and clears the reservation.

The test seeds a queued capture, drives headroom to zero through the
safety margin, and asserts the encoder was never spawned, no destination
file exists, the slot was returned exactly once, and no disk is still
reserved.

backend/dvr/src/recording/recording_service.rs

@@ -1015,7 +1015,7 @@ fn map_queue_error(err: &QueueMutationError) -> ServiceError {
     }
 }
 
-fn quota_limits_from_config(config: Option<&tuliprox_core::model::RecordingQuotaConfig>) -> QuotaLimits {
+pub(crate) fn quota_limits_from_config(config: Option<&tuliprox_core::model::RecordingQuotaConfig>) -> QuotaLimits {
     let mut per_user_bytes = HashMap::new();
     if let Some(config) = config {
         for (user_id, bytes) in &config.per_user_bytes {

backend/dvr/src/recording/recording_transfer.rs

@@ -1199,6 +1199,56 @@ fn preemption_reason_for(download: &RecordingTask) -> &'static str {
     }
 }
 
+pub(crate) const QUOTA_GONE_BEFORE_START: &str = "Quota was exhausted while this recording waited to start";
+pub(crate) const DISK_GONE_BEFORE_START: &str = "Disk space was exhausted while this recording waited to start";
+
+/// Re-run admission for the recording that is about to open its
+/// destination, returning the reason it can no longer start.
+///
+/// Admission happened when the request was accepted, which can be a long
+/// time before this point: the recording may have waited for provider
+/// capacity, sat through a retry backoff, or been scheduled hours ahead.
+/// The quota and the free space it was admitted against are not the ones
+/// it is about to consume. Without this the first sign of a full disk is
+/// a write failure part-way through a recording.
+///
+/// The subject is excluded from both sums and then added back as the
+/// candidate charge, so it is not counted twice.
+async fn refused_before_start(
+    download_queue: &Arc<RecordingQueue>,
+    app_config: &AppConfig,
+    uuid: &str,
+) -> Option<&'static str> {
+    let config = app_config.config.load();
+    let recording_cfg = config.recording()?;
+    let (_, tasks) = download_queue.committed_snapshot().await;
+    let subject = tasks.iter().find(|task| task.uuid == uuid)?;
+    let charge = crate::recording::recording_quota::charge_for_task(subject);
+    let others = || tasks.iter().filter(|task| task.uuid != uuid);
+
+    let limits = crate::recording::recording_service::quota_limits_from_config(recording_cfg.quota.as_ref());
+    let pool = crate::recording::recording_quota::quota_pool_for_task(subject);
+    let used = crate::recording::recording_quota::used_bytes_in_pool(others(), &pool);
+    if matches!(
+        crate::recording::recording_quota::would_exceed(&pool, used, charge, &limits),
+        crate::recording::recording_quota::AdmissionOutcome::OverLimit { .. }
+    ) {
+        return Some(QUOTA_GONE_BEFORE_START);
+    }
+
+    // An unmeasurable root is not grounds to refuse, exactly as at admission.
+    let free = crate::recording::recording_disk::free_bytes_for(Path::new(&recording_cfg.directory))?;
+    let safety = recording_cfg.disk.as_ref().and_then(|disk| disk.safety_bytes).unwrap_or(0);
+    let active = crate::recording::recording_disk::active_disk_reservations(others());
+    if matches!(
+        crate::recording::recording_disk::would_fit_on_disk(free, safety, active, charge),
+        crate::recording::recording_disk::DiskAdmission::Insufficient { .. }
+    ) {
+        return Some(DISK_GONE_BEFORE_START);
+    }
+    None
+}
+
 #[allow(clippy::too_many_lines)]
 pub async fn ensure_recording_worker_running(
     cfg: &AppConfig,
@@ -1447,6 +1497,18 @@ pub async fn ensure_recording_worker_running(
                                 capacity.release(provider_handle).await;
                                 break 'worker;
                             };
+                            // Last point before a destination is opened.
+                            if let Some(reason) = refused_before_start(&dq, &app_config, &worker_uuid).await {
+                                capacity.release(provider_handle).await;
+                                match fail_active_download(&dq, &worker_uuid, reason).await {
+                                    Ok(_) => publish_recording_change(&event_manager),
+                                    Err(err) => {
+                                        error!("Download worker commit failed: {err}");
+                                        break 'worker;
+                                    }
+                                }
+                                continue;
+                            }
                             match download.kind {
                                 RecordingKind::Vod | RecordingKind::Series => 'http_execution: {
                                     let execution_download = match recording_execution_download(&app_config, &download)
@@ -1914,7 +1976,7 @@ fn start_recording_scheduler(
 mod tests {
     use super::{
         ensure_recording_worker_running, finalize_http_transfer, http_transfer_path, recording_deadline_instant,
-        wait_for_provider_slot, LIVE_CAPACITY_WINDOW_CLOSED,
+        wait_for_provider_slot, DISK_GONE_BEFORE_START, LIVE_CAPACITY_WINDOW_CLOSED,
     };
     use crate::recording::{
         recording_capacity::{stub::StubCapacity, RecordingCapacityPort},
@@ -2370,4 +2432,92 @@ mod tests {
         assert!(!crate::recording_partial_path(&capture.file_path).exists(), "and nothing was left staged");
         assert_eq!(settled.recording.reserved_bytes, 0, "and it is not still holding disk");
     }
+
+    #[tokio::test]
+    async fn a_recording_whose_disk_filled_while_it_waited_never_opens_a_destination() {
+        // Admission happened when the request was accepted; this recording
+        // then waited. By the time it reaches the front of the queue the
+        // space it was admitted against is gone. Without the recheck the
+        // first sign of that is a write failure part-way through.
+        let dir = tempfile::TempDir::new().expect("tempdir");
+        let recordings_dir = dir.path().join("recordings");
+        std::fs::create_dir_all(&recordings_dir).expect("create recording dir");
+        let log = dir.path().join("spawns.log");
+        let script = counting_ffmpeg(dir.path(), &log);
+
+        let queue = Arc::new(RecordingQueue::new_persistent(dir.path(), dir.path()).expect("open repository"));
+        let now = chrono::Utc::now().timestamp();
+        let mut capture = scheduled_task(RecordingKind::Live, now, 1_800);
+        capture.uuid = "live".to_string();
+        capture.state = RecordingTaskState::Queued;
+        capture.input_name = Some(Arc::from("provider"));
+        capture.recording.source.virtual_id = "42".to_string();
+        capture.recording.reserved_bytes = 1_024;
+        capture.file_dir.clone_from(&recordings_dir);
+        capture.file_path = recordings_dir.join("capture.ts");
+        let persisted = RecordingQueue::to_persisted(&capture);
+        crate::recording::recording_queue::mutate(&queue, move |candidate| {
+            candidate.queue.push(persisted.clone());
+            Ok(())
+        })
+        .await
+        .expect("seed");
+
+        // The safety margin stands in for a disk that filled up: it drives
+        // headroom to zero without needing a real full filesystem.
+        let app_config = app_config_with_server();
+        let mut rec_cfg =
+            RecordingConfig::from(&shared::model::RecordingConfigDto { enabled: true, ..Default::default() });
+        rec_cfg.directory = recordings_dir.to_string_lossy().into_owned();
+        rec_cfg.disk = Some(tuliprox_core::model::RecordingDiskConfig {
+            high_water_percent: None,
+            low_water_percent: None,
+            cleanup_interval_secs: None,
+            safety_bytes: Some(u64::MAX),
+        });
+        let mut config = tuliprox_core::model::Config::clone(&app_config.config.load());
+        config.video = Some(tuliprox_core::model::VideoConfig {
+            extensions: Vec::new(),
+            web_search: None,
+            recording: Some(rec_cfg.clone()),
+        });
+        app_config.config.store(Arc::new(config));
+
+        let stub = StubCapacity::with_room();
+        let capacity: Arc<dyn RecordingCapacityPort> = Arc::clone(&stub) as Arc<dyn RecordingCapacityPort>;
+        ensure_recording_worker_running(
+            &app_config,
+            &rec_cfg,
+            &queue,
+            &Arc::new(EventManager::new()),
+            &capacity,
+            &script,
+        )
+        .await
+        .expect("worker started");
+
+        let settled = tokio::time::timeout(Duration::from_secs(10), async {
+            loop {
+                if let Some(done) = queue.finished.read().await.first().cloned() {
+                    break done;
+                }
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+        })
+        .await;
+        let Ok(settled) = settled else {
+            panic!(
+                "never settled: active={:?} spawns={}",
+                queue.active.read().await.as_ref().map(|active| (active.state, active.error.clone())),
+                spawn_count(&log),
+            );
+        };
+
+        assert_eq!(settled.state, RecordingTaskState::Failed, "{:?}", settled.error);
+        assert_eq!(settled.error.as_deref(), Some(DISK_GONE_BEFORE_START), "and it says which resource ran out");
+        assert_eq!(spawn_count(&log), 0, "the encoder was never started");
+        assert!(!capture.file_path.exists(), "and no destination was opened");
+        assert_eq!(stub.release_count(), 1, "the provider slot it had was given back");
+        assert_eq!(settled.recording.reserved_bytes, 0, "and it is not still holding disk");
+    }
 }