diff --git a/src/disc/mod.rs b/src/disc/mod.rs index ac60236..f3d53cd 100644 --- a/src/disc/mod.rs +++ b/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) -> 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) diff --git a/src/io/pipeline.rs b/src/io/pipeline.rs index d158f0c..95400b6 100644 --- a/src/io/pipeline.rs +++ b/src/io/pipeline.rs @@ -325,20 +325,21 @@ impl Pipeline { /// /// 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>(depth: usize, sink: S) -> Result { Self::spawn_named("freemkv-pipeline-consumer", depth, sink) } @@ -346,7 +347,7 @@ impl Pipeline { /// 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>( name: &str, depth: usize, @@ -631,6 +632,14 @@ impl Pipeline { /// 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 Pipeline { /// 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 { let Pipeline { tx, diff --git a/src/mux/codec/ac3.rs b/src/mux/codec/ac3.rs index e8f4281..870266a 100644 --- a/src/mux/codec/ac3.rs +++ b/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, /// 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(); diff --git a/src/mux/codec/adts.rs b/src/mux/codec/adts.rs index 6d6dd3a..d811399 100644 --- a/src/mux/codec/adts.rs +++ b/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 { diff --git a/src/session.rs b/src/session.rs index aa857d6..23246de 100644 --- a/src/session.rs +++ b/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