Account for decrypt-time loss so partial AACS/CSS failures can't pass as a perfect rip

When a scrambled AACS unit fails to decrypt under every available key
(a missing/wrong CPS sub-key, or a marginal unit that fails the TS-sync
verify), decrypt_sectors restored the original encrypted bytes and
returned Ok with no signal. Those still-encrypted bytes flowed to the TS
assembler, which silently dropped the non-syncing packets with no loss
counter. The only loss accounting was DiscStream's read-error zero-fill
path, so mux reported lost_video_secs=0 for decrypt-dropped content and
the abort gate accepted the rip even under abort_on_lost_secs=0. A rip
missing real video/audio segments was published as a perfect success.

decrypt_sectors now returns the number of bytes in scrambled units that
no key could decrypt. DecryptingSectorSource accumulates that into a
shared counter exposed via decrypt_loss(); both mux pipelines fold it
into lost_bytes() — the inline DiscStream path directly, and the
file-backed highway via PipelinedPesStream sharing the producer's
counter. Restore-to-original is unchanged, so clear nav-files are never
corrupted; metadata-probe callers that don't read the counter are
unaffected. Adds regression tests at the decrypt and decorator layers.

src/mux/disc.rs

@@ -811,7 +811,17 @@ impl crate::pes::Stream for DiscStream {
     }
 
     fn lost_bytes(&self) -> u64 {
-        self.lost_bytes
+        // Read-error zero-fill loss (counted in fill_extents) PLUS decrypt-time
+        // loss — bytes of scrambled AACS units the decorator could not decrypt
+        // and passed through still encrypted (the TS assembler silently drops
+        // them). Both are real missing content the abort gate must see; without
+        // the decrypt term a partial key failure reports lost_bytes=0 and a rip
+        // missing segments passes even under abort_on_lost_secs=0.
+        self.lost_bytes.saturating_add(
+            self.reader
+                .decrypt_loss()
+                .load(std::sync::atomic::Ordering::Relaxed),
+        )
     }
 }
 

src/mux/pipelined_stream.rs

@@ -56,6 +56,14 @@ pub struct PipelinedPesStream {
     /// `std::env::var_os` takes a process-wide lock, so the per-batch /
     /// per-poll reads it replaces were needless hot-path overhead.
     skip_parse: bool,
+    /// Cumulative bytes of scrambled AACS units the producer's decrypt step
+    /// could not decrypt — silent decrypt loss the demux drops without a sync.
+    /// Shared with the producer thread's [`DecryptingSectorSource`]
+    /// (`crate::sector::DecryptingSectorSource::decrypt_loss`). Surfaced through
+    /// [`Stream::lost_bytes`] so the file-backed mux abort gate sees a partial
+    /// decrypt failure instead of reporting a perfect rip. `None` for pipelines
+    /// with no AACS decrypt step (e.g. the M2TS byte-stream path).
+    decrypt_loss: Option<std::sync::Arc<std::sync::atomic::AtomicU64>>,
 }
 
 impl PipelinedPesStream {
@@ -84,9 +92,24 @@ impl PipelinedPesStream {
             pending_frames: std::collections::VecDeque::new(),
             eof: false,
             skip_parse: std::env::var_os("FREEMKV_SKIP_PARSE").is_some(),
+            decrypt_loss: None,
         }
     }
 
+    /// Attach the producer's decrypt-loss counter so [`Stream::lost_bytes`]
+    /// reports bytes of scrambled AACS units that could not be decrypted (and
+    /// were therefore silently dropped downstream). Obtained from the
+    /// producer's `DecryptingSectorSource::decrypt_loss()` before it is moved
+    /// into the prefetch thread. The M2TS / no-decrypt pipelines leave this
+    /// unset.
+    pub(crate) fn with_decrypt_loss(
+        mut self,
+        loss: std::sync::Arc<std::sync::atomic::AtomicU64>,
+    ) -> Self {
+        self.decrypt_loss = Some(loss);
+        self
+    }
+
     /// Pull one batch of `PesPacket`s from the demux thread, run
     /// codec parse on each, enqueue resulting `PesFrame`s on
     /// `pending_frames`. Returns Ok(true) on success, Ok(false) on
@@ -263,6 +286,18 @@ impl Stream for PipelinedPesStream {
             .find(|(p, _)| *p == pid)
             .and_then(|(_, parser)| parser.codec_private())
     }
+
+    fn lost_bytes(&self) -> u64 {
+        // The file-backed highway has no read-error zero-fill term (resolve
+        // tracks read loss separately), but the producer's decrypt step can
+        // pass scrambled units through undecrypted — silent loss the demux
+        // drops. Surface that so the mux abort gate sees a partial AACS/CSS
+        // decrypt failure rather than reporting a perfect rip.
+        self.decrypt_loss
+            .as_ref()
+            .map(|c| c.load(std::sync::atomic::Ordering::Relaxed))
+            .unwrap_or(0)
+    }
 }
 
 #[cfg(test)]

src/mux/resolve.rs

@@ -549,6 +549,12 @@ pub fn build_iso_pipeline<S: SectorSource + Send + 'static>(
     };
     let decrypting =
         crate::sector::DecryptingSectorSource::new(Box::new(reader) as Box<dyn SectorSource>, keys);
+    // Grab the decrypt-loss counter before the decorator is moved into the
+    // producer thread. It tracks bytes of scrambled AACS units no key could
+    // decrypt — silent loss the demux drops; the consuming stream surfaces it
+    // through `lost_bytes()` so the mux abort gate sees a partial decrypt
+    // failure rather than a clean rip.
+    let decrypt_loss = decrypting.decrypt_loss();
     let prefetched = crate::sector::PrefetchedSectorSource::new_with_events(
         decrypting,
         extents,
@@ -564,13 +570,10 @@ pub fn build_iso_pipeline<S: SectorSource + Send + 'static>(
     let (demux_thread, demux_rx) =
         super::demux_thread::DemuxThread::spawn_zero_copy(rx, recycle_tx, shell, halt, ts, ps)
             .map_err(|e| -> io::Error { e.into() })?;
-    Ok(PipelinedPesStream::new(
-        demux_thread,
-        demux_rx,
-        title,
-        parsers,
-        pid_to_track,
-    ))
+    Ok(
+        PipelinedPesStream::new(demux_thread, demux_rx, title, parsers, pid_to_track)
+            .with_decrypt_loss(decrypt_loss),
+    )
 }
 
 /// Assemble the M2TS file mux pipeline (read → demux → parse) for a