Audit round: an ADTS frame that cannot hold its own CRC, and three drifts

An ADTS header that declares a CRC follows must be at least nine bytes —
seven of header plus the two the CRC occupies — because the declared frame
length counts them. The structural gate compared against a flat seven and
never read the bit that says whether a CRC is there at all, so a frame
whose own header describes something impossible was accepted and handed to
the muxer as decodable.

The pipeline's spawn doc named Sweep, and the thread Sweep would have
created, as callers to look for. Neither has been in this crate since the
recovery passes moved out. The same paragraph already records fixing this
once, for a different departed caller — it simply drifted again a sentence
later, so it now says to name callers that live here or name none.

Whether a disc is structurally AACS-encrypted was spelled out by hand in
both the fast identify and the full scan. They agreed today; nothing made
them agree tomorrow, and disagreeing would mean the same disc reported
encrypted by one path and clear by the other. There is one definition now,
and the comment that pointed at it by line number points at its name.

The AC-3 parser built a fresh buffer on every packet — of the order of a
hundred thousand times per title — to work around a borrow it cannot
avoid. The copy stays; the allocation does not. The buffer is now lent out
and handed back, and a test pins that, because reverting it would be
invisible in behaviour.

Left alone deliberately: send/send_with_halt and finish/finish_with_halt
look like one action under two names, and are not. After the consumer
fails, one must still accept items and the other must refuse them; that
difference is what stops a producer reading an entire disc for a write
that died on its first frame. Collapsing them was tried here and the
existing test caught it. Both now say so where the choice is made.

src/disc/mod.rs

@@ -479,6 +479,25 @@ pub struct Extent {
 /// Union a set of extents into sorted, merged, disjoint `(start_lba,
 /// sector_count)` ranges — the pure, testable core of
 /// [`Disc::encrypted_content_ranges`]. Reuses [`crate::udf::merge_ranges`].
+/// Is this disc structurally AACS-encrypted — i.e. does it carry an AACS
+/// directory?
+///
+/// THE ONE definition of that question. `Disc::identify` (the fast path, which
+/// only reads the filesystem) and `Disc::scan_with` (the full scan) both need
+/// it, and both used to spell the same two `find_dir` calls out by hand. They
+/// agreed today; nothing made them agree tomorrow. Adding a third AACS
+/// location, or excluding an empty placeholder directory, to one copy and not
+/// the other would silently desync the fast identify from the full scan — the
+/// same disc reported encrypted by one and clear by the other.
+///
+/// This is STRUCTURAL, not cryptographic: it says the tree looks like an
+/// encrypted disc, not that any sector actually is. A folder copied verbatim
+/// from a decrypted disc keeps its `AACS/` and answers true here (see
+/// `session::scan_dir`, which corrects for exactly that).
+pub(crate) fn aacs_dir_present(udf_fs: &crate::udf::UdfFs) -> bool {
+    udf_fs.find_dir("/AACS").is_some() || udf_fs.find_dir("/BDMV/AACS").is_some()
+}
+
 fn merged_extents<'a>(extents: impl Iterator<Item = &'a Extent>) -> Vec<(u32, u32)> {
     let mut ranges: Vec<(u32, u32)> = extents.map(|e| (e.start_lba, e.sector_count)).collect();
     ranges.sort_by_key(|r| r.0);
@@ -1628,8 +1647,7 @@ impl Disc {
         // (no titles needed: BD/UHD/FMTS come from the MKB generation). It no
         // longer defaults to BluRay or defers UHD/FMTS to the full scan.
         let format = Self::detect_disc_format(&mut buffered, &udf_fs, &[]);
-        let encrypted =
-            udf_fs.find_dir("/AACS").is_some() || udf_fs.find_dir("/BDMV/AACS").is_some();
+        let encrypted = aacs_dir_present(&udf_fs);
         let layers = if capacity > 24_000_000 { 2 } else { 1 };
 
         Ok(DiscId {
@@ -2015,8 +2033,7 @@ impl Disc {
         let scan_with_t0 = std::time::Instant::now();
         tracing::info!(target: "freemkv::scan", phase = "scan_with", "begin");
         // 2. Resolve encryption (AACS, CSS, or none)
-        let encrypted =
-            udf_fs.find_dir("/AACS").is_some() || udf_fs.find_dir("/BDMV/AACS").is_some();
+        let encrypted = aacs_dir_present(&udf_fs);
 
         let (aacs, aacs_error) = if !encrypted {
             (None, None)

src/io/pipeline.rs

@@ -325,20 +325,21 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     ///
     /// The thread is named `freemkv-pipeline-consumer` so it shows up
     /// distinctly in stack traces and `top -H`. Callers that want a
-    /// more specific name (e.g. `freemkv-sweep-consumer`) should use
-    /// [`Pipeline::spawn_named`] instead. Returns an `Error::IoError`
-    /// if the OS refuses the thread spawn (resource exhaustion);
-    /// callers already operate in fallible context, so this is
-    /// propagated rather than panicked.
+    /// more specific name should use [`Pipeline::spawn_named`] instead.
+    /// Returns an `Error::IoError` if the OS refuses the thread spawn
+    /// (resource exhaustion); callers already operate in fallible context, so
+    /// this is propagated rather than panicked.
     ///
-    /// Sweep uses [`Pipeline::spawn_named`] directly so the consumer
-    /// thread shows up as `freemkv-sweep-consumer`; mux uses
-    /// `freemkv-mux-consumer`. `Pipeline::spawn` (this function, with
-    /// the default name) is used only by the unit tests in this module.
-    /// It used to name `disc::patch` as a caller; that module does not exist
-    /// here any more — the sweep/patch recovery passes moved to freemkv-engine
-    /// in 1.6.0, so the comment sent readers hunting a caller in this crate
-    /// that had already left it.
+    /// Inside this crate the only [`Pipeline::spawn_named`] caller is the mux
+    /// driver, which names its thread `freemkv-mux-consumer`. `Pipeline::spawn`
+    /// (this function, with the default name) is used only by the unit tests in
+    /// this module.
+    ///
+    /// This paragraph twice named a caller that had left the crate: first
+    /// `disc::patch`, then Sweep and its `freemkv-sweep-consumer` thread. Both
+    /// went to freemkv-engine with the recovery passes in 1.6.0, and each in
+    /// turn sent readers hunting a component that is not here. Name callers
+    /// that live in THIS crate, or none.
     pub fn spawn<S: Sink<I, Output = R>>(depth: usize, sink: S) -> Result<Self, Error> {
         Self::spawn_named("freemkv-pipeline-consumer", depth, sink)
     }
@@ -346,7 +347,7 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// Like [`Pipeline::spawn`] but lets the caller supply the
     /// consumer thread's name. Useful when several pipelines run in
     /// the same process and stack traces / `top -H` need to tell them
-    /// apart (e.g. `freemkv-sweep-consumer`, `freemkv-mux-consumer`).
+    /// apart (e.g. `freemkv-mux-consumer`).
     pub fn spawn_named<S: Sink<I, Output = R>>(
         name: &str,
         depth: usize,
@@ -631,6 +632,14 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// wedged inside an unkillable syscall, the producer can still
     /// observe `/api/stop` and unwind within
     /// [`SEND_HALT_CHECK_INTERVAL`].
+    /// NOT a `foo_with_X` variant of [`Pipeline::send`], despite the name.
+    /// The two encode OPPOSITE policies on the same event, each with its own
+    /// test: after the consumer's `apply` has failed, `send` still succeeds
+    /// (the consumer keeps draining, so the channel accepts the item), while
+    /// this one hands the item straight back — so a producer does not read an
+    /// hour of disc for a write that died on the first frame. Collapsing them
+    /// into one Option-parameterised method deletes one of those behaviours;
+    /// it was tried and `apply_error_drains_then_propagates` caught it.
     pub fn send_with_halt(&self, item: I, halt: &Halt, deadline: Duration) -> Result<(), I> {
         use crossbeam_channel::SendTimeoutError;
         let end = Instant::now() + deadline;
@@ -742,6 +751,12 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// Plain [`Pipeline::finish`] is preserved for callers without a
     /// halt-token plumbed through; that path still blocks indefinitely
     /// on `join()`, matching pre-0.20.8 behaviour.
+    /// Also not a `foo_with_X` variant: [`Pipeline::finish`] joins and waits
+    /// however long the consumer needs, while this one gives up after
+    /// `JOIN_TIMEOUT_SECS` and reports halted. Which is right depends on
+    /// whether the caller has a user waiting to cancel — the mux driver does
+    /// and uses this; the unit tests do not and use the plain join. Merging
+    /// them means picking one of those policies for both.
     pub fn finish_with_halt(self, halt: Option<&Halt>) -> Result<R, Error> {
         let Pipeline {
             tx,

src/mux/codec/ac3.rs

@@ -67,6 +67,13 @@ pub struct Ac3Parser {
     /// began in an earlier packet takes THAT packet's source offset, not the
     /// one that happened to complete it.
     acc: super::pesbuf::PesBuf,
+    /// Reused working copy of `acc`, kept across calls so the per-PES scan does
+    /// not allocate. `parse` runs once per PES packet on an audio track — of
+    /// the order of 10^5 times per title — and previously built a fresh `Vec`
+    /// each time. The copy itself is unavoidable without restructuring the
+    /// borrow relationship (the scanner needs `&mut self.tally` while it reads
+    /// these bytes); the ALLOCATION is not.
+    scratch: Vec<u8>,
     /// PTS (ns) to stamp on the frame that begins the carry-over `buf` — i.e.
     /// the running per-frame PTS at the point the partial tail was retained.
     /// Used by `flush()` to time the final buffered frame at EOS.
@@ -95,6 +102,7 @@ impl Default for Ac3Parser {
 impl Ac3Parser {
     pub fn new() -> Self {
         Self {
+            scratch: Vec::new(),
             acc: super::pesbuf::PesBuf::with_capacity(4096),
             flush_pts_ns: 0,
             tally: super::dropgate::DropTally::new("ac3"),
@@ -500,7 +508,12 @@ impl CodecParser for Ac3Parser {
         // Copy the working bytes out so the scanner can borrow `self.tally`;
         // the buffer keeps its marks, so the unconsumed tail stays attributed
         // to the packet that carried it.
-        let buf = self.acc.as_slice().to_vec();
+        // Take the scratch OUT of `self` so the scanner can borrow `self.tally`
+        // while reading it; it is put back at the end of the call, keeping its
+        // capacity for the next PES. There is no early return after this point.
+        let mut buf = std::mem::take(&mut self.scratch);
+        buf.clear();
+        buf.extend_from_slice(self.acc.as_slice());
         let marks = self.acc.marks_snapshot();
         let data = &buf;
         let (frames, keep_from, frame_pts_ns) =
@@ -541,6 +554,9 @@ impl CodecParser for Ac3Parser {
             self.flush_pts_ns = frame_pts_ns;
         }
 
+        // Hand the working buffer back so the next PES reuses its capacity.
+        // `data` borrowed it; that borrow ends here, at its last use.
+        self.scratch = buf;
         frames
     }
 
@@ -807,6 +823,53 @@ mod tests {
         frame[n - 1] = (c & 0xFF) as u8;
     }
 
+    /// The per-PES working buffer must be REUSED, not reallocated.
+    ///
+    /// `parse` runs once per PES packet — of the order of 10^5 times on a
+    /// feature's audio track — and used to build a fresh `Vec` every call. The
+    /// copy itself cannot go without restructuring the borrow relationship, but
+    /// the allocation can: the buffer is taken out of `self`, filled, and put
+    /// back with its capacity intact.
+    ///
+    /// Asserting on capacity is deliberately white-box, because that is exactly
+    /// what regresses if someone reverts to `to_vec()` — the behaviour would be
+    /// identical and no other test would notice.
+    #[test]
+    fn the_working_buffer_is_reused_across_packets_not_reallocated() {
+        let mut parser = Ac3Parser::new();
+        let frame = make_ac3_frame(0, 0);
+
+        parser.parse(&PesPacket {
+            source: None,
+            pid: 0x1100,
+            pts: Some(90_000),
+            dts: None,
+            data: frame.clone(),
+            discontinuity: false,
+        });
+        let cap_after_first = parser.scratch.capacity();
+        assert!(
+            cap_after_first > 0,
+            "the buffer must be handed back to the parser, not dropped"
+        );
+
+        for _ in 0..8 {
+            parser.parse(&PesPacket {
+                source: None,
+                pid: 0x1100,
+                pts: None,
+                dts: None,
+                data: frame.clone(),
+                discontinuity: false,
+            });
+        }
+        assert!(
+            parser.scratch.capacity() >= cap_after_first,
+            "capacity must persist across calls; a fresh Vec each time would \
+             show it dropping back to the last packet's size"
+        );
+    }
+
     #[test]
     fn parse_empty_pes() {
         let mut parser = Ac3Parser::new();

src/mux/codec/adts.rs

@@ -50,7 +50,15 @@ fn adts_verdict(data: &[u8]) -> AdtsVerdict {
     // aac_frame_length: 13 bits = byte3[1:0] | byte4 | byte5[7:5].
     let frame_length =
         ((u32::from(data[3]) & 0x03) << 11) | (u32::from(data[4]) << 3) | (u32::from(data[5]) >> 5);
-    if frame_length < 7 {
+    // The floor is the header the frame SAYS it carries, not a constant.
+    // protection_absent (byte1 bit0) clear means a 16-bit crc_check follows the
+    // 7-byte fixed+variable header, so the frame cannot be shorter than 9 —
+    // aac_frame_length counts the header and the CRC, not just the payload.
+    // Comparing against a flat 7 let a CRC-present frame declaring 7 or 8
+    // through as structurally Valid, and the muxer then carried a frame whose
+    // own header says it is impossible.
+    let header_bytes = if data[1] & 0x01 == 0 { 9 } else { 7 };
+    if frame_length < header_bytes {
         return AdtsVerdict::Invalid;
     }
     AdtsVerdict::Valid
@@ -152,6 +160,51 @@ mod tests {
         }
     }
 
+    /// A header that CLAIMS a CRC (protection_absent = 0) but declares a
+    /// frame length too short to contain one.
+    ///
+    /// `aac_frame_length` counts the header and the CRC, not just the payload,
+    /// so with a CRC present the smallest structurally possible frame is 9
+    /// bytes: the 7-byte fixed+variable header plus the 16-bit crc_check.
+    /// The gate compared against a flat 7 and never read protection_absent at
+    /// all, so a frame whose own header says it is impossible was classified
+    /// Valid and forwarded to the muxer.
+    #[test]
+    fn a_crc_present_header_shorter_than_its_own_crc_is_invalid() {
+        for declared in [7u32, 8] {
+            let mut f = adts_frame(16);
+            f[1] = 0xF0; // sync + MPEG-4, protection_absent = 0 => CRC present
+            f[3] = (f[3] & 0xFC) | ((declared >> 11) & 0x03) as u8;
+            f[4] = ((declared >> 3) & 0xFF) as u8;
+            f[5] = (f[5] & 0x1F) | ((declared & 0x07) << 5) as u8;
+            assert!(
+                matches!(adts_verdict(&f), AdtsVerdict::Invalid),
+                "protection_absent=0 declaring {declared} bytes cannot hold its \
+                 own 7-byte header plus a 2-byte CRC"
+            );
+        }
+
+        // 9 is the smallest length that CAN hold header + CRC, so it must pass
+        // the structural gate — the floor moved, it did not become stricter
+        // than the spec.
+        let mut ok = adts_frame(16);
+        ok[1] = 0xF0;
+        let nine = 9u32;
+        ok[3] = (ok[3] & 0xFC) | ((nine >> 11) & 0x03) as u8;
+        ok[4] = ((nine >> 3) & 0xFF) as u8;
+        ok[5] = (ok[5] & 0x1F) | ((nine & 0x07) << 5) as u8;
+        assert!(matches!(adts_verdict(&ok), AdtsVerdict::Valid));
+
+        // And with NO CRC the floor is still 7, unchanged.
+        let mut no_crc = adts_frame(16);
+        no_crc[1] = 0xF1; // protection_absent = 1
+        let seven = 7u32;
+        no_crc[3] = (no_crc[3] & 0xFC) | ((seven >> 11) & 0x03) as u8;
+        no_crc[4] = ((seven >> 3) & 0xFF) as u8;
+        no_crc[5] = (no_crc[5] & 0x1F) | ((seven & 0x07) << 5) as u8;
+        assert!(matches!(adts_verdict(&no_crc), AdtsVerdict::Valid));
+    }
+
     /// A valid ADTS header (AAC-LC, 44.1 kHz, stereo) + payload, with
     /// aac_frame_length set to the total size.
     fn adts_frame(payload: usize) -> Vec<u8> {

src/session.rs

@@ -450,7 +450,7 @@ const AACS_PROBE_UNITS: usize = 8;
 ///
 /// The extra step over `scan_iso` is the encryption verdict.
 /// `Disc::scan_with` decides `encrypted` STRUCTURALLY, from the presence of an
-/// `/AACS` or `/BDMV/AACS` directory (`disc/mod.rs:1992-1993`). For the common
+/// `/AACS` or `/BDMV/AACS` directory (see `disc::aacs_dir_present`). For the common
 /// case — a MakeMKV-style backup, which strips `AACS/` — that already gives the
 /// right answer, and `DecryptKeys::None` is a pass-through. But a folder copied
 /// verbatim from a decrypted disc keeps `AACS/`, and the tree shape then claims