From 0bbceed985726f9f92c70abd60888e3ea16d1f75 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Wed, 29 Jul 2026 22:09:52 -0700 Subject: [PATCH] Round 4: fix 26 defects across crypto, resource use and codec paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Twenty-six confirmed findings from the fourth audit round, landed as one cluster because they were found by agents working over disjoint file sets. The one worth calling out is a pair of AACS tests that could not fail. Both asserted CBC behaviour against a hand-rolled expectation that happened to be IV-independent, so replacing AACS_IV with sixteen zero bytes left them passing — they were pinning the code's own arithmetic, not the published constant. Replaced with a literal witness of the published IV plus the NIST SP 800-38A F.2.2 CBC-AES128 vector, and verified the other way round: zeroing AACS_IV now fails three tests. The rest are allocation and correctness work on hot paths: the Annex-B writer in demux_sink allocated and freed a whole-frame Vec per frame, which for a UHD title is ~200,000 allocations over the mmap threshold plus the page faults to first-touch each one; it now reuses a buffer on the writer, and still takes the NAL prefix width from the configuration record rather than assuming four. Six findings whose real fix lives in a consumer crate are recorded for re-filing rather than patched here. --- src/aacs/content.rs | 135 +++++++++++++++++--- src/aacs/crypto.rs | 70 +++++++++-- src/decrypt.rs | 56 +++++++++ src/drive/mod.rs | 75 +++++++++++- src/io/pipeline.rs | 5 +- src/io/writeback_file/macos.rs | 22 +++- src/keysource.rs | 77 +++++++++++- src/mux/au_assembly.rs | 87 ++++++++++++- src/mux/codec/hevc.rs | 218 +++++++++++++++++++++++++++++++-- src/mux/codec/truehd.rs | 106 ++++++++++++++-- src/mux/demux_sink.rs | 16 ++- src/mux/driver.rs | 39 +++++- src/mux/m2ts_mux/mod.rs | 104 +++++++++++++--- src/mux/mkv.rs | 69 ++++++++++- src/mux/ts.rs | 96 ++++++++++++++- src/mux/tsmux.rs | 100 +++++++++++++-- src/mux/videomap.rs | 9 +- src/sector/prefetched.rs | 82 +++++++++++-- src/udf.rs | 102 ++++++++++++++- 19 files changed, 1342 insertions(+), 126 deletions(-) diff --git a/src/aacs/content.rs b/src/aacs/content.rs index 5727e47..a0efc20 100644 --- a/src/aacs/content.rs +++ b/src/aacs/content.rs @@ -147,8 +147,14 @@ pub fn aacs_unit_needs_decrypt(unit: &[u8], format: crate::disc::ContentFormat) } /// Minimum synced content packets that PROVE a key opened a unit. Four `0x47` -/// syncs ≈ 32 bits of MPEG-TS structure ≈ 1-in-4-billion that a wrong key (uniform -/// AES noise, `0x47` at 1/256 per packet) fakes it. It is an ABSOLUTE proof floor, +/// syncs are 32 bits of MPEG-TS structure, but the per-UNIT false-pass risk is +/// NOT 2^-32: `is_clean_ts` accepts ANY four of the ~31 encrypted packets in a +/// 6144-byte aligned unit, so for a wrong key (uniform AES noise, `0x47` at 1/256 +/// per packet) it is ≈ C(31,4)·256^-4 ≈ 7e-6, i.e. ~1e-5 — the figure +/// [`is_clean_ts`]'s own doc below states. 1-in-4-billion is the probability for +/// four SPECIFIC packets and overstates the margin by ~4000x; at +/// `KEY_PROOF_PACKETS = 3` the per-unit rate is ≈ C(31,3)·256^-3 ≈ 2.6e-4, so do +/// NOT lower it on the strength of slack that is not there. It is an ABSOLUTE proof floor, /// NOT a proportion — a unit the key opened but whose content is bad-encoded /// (many non-conforming packets) is proven by ANY four good packets, not rejected /// for the bad ones. @@ -172,8 +178,16 @@ pub fn is_clean(unit: &[u8], format: crate::disc::ContentFormat) -> bool { /// Structural "does this unit carry enough valid MPEG-TS to prove a key opened /// it?" — the Transport-Stream arm of [`is_clean`]. It is /// NOT a decryption verdict: [`decrypt_unit`] applies a key (that is -/// "decrypt"); whether the plaintext is clean TS is this SEPARATE question. The -/// mux never calls this — TS validity is a muxer concern, never a decrypt result. +/// "decrypt"); whether the plaintext is clean TS is this SEPARATE question. +/// +/// The mux is its PRINCIPAL consumer for `BdTs` discs, reaching it through +/// [`is_clean`]: `mux::resolve`'s multi-CPS `pick` closure selects a unit key by +/// it, `probe_index_phase` reports each FMTS index's interleave parity by it, and +/// `decrypt::decrypt_sectors_mapped` uses it as the forensic-range verify net. +/// (The doc used to say "the mux never calls this", which invited a maintainer to +/// tighten or loosen the proof rule below believing only whole-disc read +/// verification was affected — while it in fact changes which unit key a +/// multi-CPS disc muxes with and which phase an FMTS index is muxed at.) /// /// Rule — evidence is ABSOLUTE, scaled to the packets that exist. Over the /// ENCRYPTED packets (skip packet 0: its `0x47` sits in the clear 16-byte seed, so @@ -381,13 +395,20 @@ pub fn encrypt_unit(unit: &mut [u8], unit_key: &[u8; 16]) -> bool { /// Remove bus encryption from an aligned unit (AACS 2.0 / UHD). /// Bus encryption uses read_data_key, decrypting bytes 16..2048 of each 2048-byte sector. pub(crate) fn decrypt_bus(unit: &mut [u8], read_data_key: &[u8; 16]) { + // Expand the key schedule ONCE for the whole unit. `read_data_key` is + // loop-invariant here (and constant for the entire disc), but calling + // `aes_cbc_decrypt` per sector rebuilt the AES-128 schedule per sector — three + // expansions per 6144-byte aligned unit, i.e. ~29 million redundant expansions + // over a 90 GB read on a stock drive, on the per-unit decrypt hot path. + // Measured by `decrypt_bus_expands_the_read_data_key_once_per_unit`. + let cipher = crate::aacs::crypto::new_cipher_for(read_data_key); for sector_start in (0..ALIGNED_UNIT_LEN).step_by(SECTOR_BYTES) { if sector_start + SECTOR_BYTES > unit.len() { break; } // First 16 bytes of each sector are plaintext - aes_cbc_decrypt( - read_data_key, + crate::aacs::crypto::cbc_decrypt_blocks( + &cipher, &mut unit[sector_start + 16..sector_start + SECTOR_BYTES], ); } @@ -1182,21 +1203,100 @@ mod tests { assert_eq!(aes_ecb_decrypt(&key, &expected), pt); } + // ── decrypt_bus: one key schedule per unit, not one per sector ───────── + + /// MEASURED, not reasoned: `decrypt_bus` called `aes_cbc_decrypt` once per + /// 2048-byte sector, and each call built its own AES-128 key schedule, so a + /// 6144-byte aligned unit performed THREE key expansions under the same + /// loop-invariant `read_data_key`. On a 90 GB UHD read on a stock (non- + /// LibreDrive) drive — ~14.6 million aligned units — that is ~29 million + /// redundant expansions on the per-unit decrypt hot path, for a key that is + /// constant for the whole disc. The counter is incremented inside + /// `crypto::new_cipher`, the single construction site. + #[test] + fn decrypt_bus_expands_the_read_data_key_once_per_unit() { + use crate::aacs::crypto::KEY_EXPANSIONS; + let mut unit = clear_unit(); + let rdk = [0x4Eu8; 16]; + KEY_EXPANSIONS.with(|c| c.set(0)); + decrypt_bus(&mut unit, &rdk); + let n = KEY_EXPANSIONS.with(|c| c.get()); + assert_eq!( + n, 1, + "one aligned unit under one read_data_key must expand the schedule \ + exactly once, not once per 2048-byte sector" + ); + } + + /// The single-expansion refactor must be byte-identical: bus encryption + /// ([C] §4.2) covers bytes 16..2048 of every 2048-byte sector, so a + /// three-sector aligned unit round-trips through the forward direction + /// sector by sector and `decrypt_bus` must recover it exactly. + #[test] + fn decrypt_bus_roundtrips_every_sector_region() { + let rdk = [0x91u8; 16]; + let original = clear_unit(); + let mut unit = original.clone(); + // Forward direction, region by region — the inverse of decrypt_bus. + for start in (0..ALIGNED_UNIT_LEN).step_by(SECTOR_BYTES) { + crate::aacs::crypto::aes_cbc_encrypt(&rdk, &mut unit[start + 16..start + SECTOR_BYTES]); + } + assert_ne!( + &unit[16..64], + &original[16..64], + "the forward direction must have changed the bytes" + ); + decrypt_bus(&mut unit, &rdk); + assert_eq!( + unit.as_slice(), + original.as_slice(), + "decrypt_bus must invert the per-sector bus encryption exactly" + ); + } + // ── CBC decrypt: first-block uses fixed AACS IV ──────────────────────── + /// The published `iv0` bytes, INDEPENDENT of the production constant. + /// + /// [C] §2.1.2 fixes one default CBC IV for every AACS AES-CBC operation. + /// Both IV tests below used to compute their expected value from + /// `crypto::AACS_IV` itself, so the constant was asserted against itself and + /// NOTHING in the suite pinned its bytes: swapping `AACS_IV` for `[0u8; 16]` + /// left both tests passing (one builds its ciphertext with the same value and + /// the other cancels the change in a triple XOR) while every real AACS disc + /// decrypted to noise — block 0 of every 6128-byte aligned unit and of every + /// bus-encrypted sector XORed with the wrong IV. This literal is the + /// independent witness the tests assert against. + const IV0_PUBLISHED: [u8; 16] = [ + 0x0B, 0xA0, 0xF8, 0xDD, 0xFE, 0xA6, 0x1F, 0xB3, 0xD8, 0xDF, 0x9F, 0x56, 0x6A, 0x05, 0x0F, + 0x78, + ]; + + /// Pins the fixed AACS CBC IV ([C] §2.1.2 `iv0`) against a literal, so a + /// change to `crypto::AACS_IV` fails HERE rather than silently shipping. + #[test] + fn aacs_iv_matches_published_iv0() { + assert_eq!( + AACS_IV, IV0_PUBLISHED, + "the fixed AACS CBC IV must be the published iv0" + ); + } + #[test] fn cbc_decrypt_first_block_xors_aacs_iv() { // CBC: P[0] = AES-D(K, C[0]) XOR IV, and the IV is the fixed AACS - // constant (not zero). Encrypt a single block forward with IV, then - // confirm aes_cbc_decrypt recovers it — proving the IV used on block - // 0 is exactly AACS_IV. A mutation that swaps AACS_IV for [0u8;16] - // makes the recovered block wrong. + // constant (not zero). Encrypt a single block forward with the PUBLISHED + // iv0 literal, then confirm aes_cbc_decrypt recovers it — proving the IV + // the production code uses on block 0 is exactly that value. Building the + // fixture from `IV0_PUBLISHED` rather than from `AACS_IV` is what makes + // the claim real: a mutation that swaps AACS_IV for [0u8;16] now makes the + // recovered block wrong. let key = [0x24u8; 16]; let plain = [0x5Au8; 16]; // Forward CBC for one block: C = AES-E(K, P XOR IV). let mut x = plain; for j in 0..16 { - x[j] ^= AACS_IV[j]; + x[j] ^= IV0_PUBLISHED[j]; } let ct = aes_ecb_encrypt(&key, &x); let mut buf = ct; @@ -1225,10 +1325,13 @@ mod tests { // * Blocks 1..=3 are independent of the IV — they MUST equal the NIST // plaintext byte-for-byte (P[i] = AES-D(K, C[i]) XOR C[i-1]). This // pins the real reverse-order CBC chaining against a published KAT. - // * Block 0 = AES-D(K, C[0]) XOR AACS_IV = NIST_PT[0] XOR NIST_IV - // XOR AACS_IV — the documented IV substitution. Asserting this exact - // relation pins both the AES decrypt of C[0] AND that block 0 uses - // AACS_IV (a swap to [0u8;16] or a chaining bug fails it). + // * Block 0 = AES-D(K, C[0]) XOR iv0 = NIST_PT[0] XOR NIST_IV XOR iv0 — + // the documented IV substitution. Asserting this exact relation pins + // both the AES decrypt of C[0] AND that block 0 uses iv0. The expected + // value is built from the `IV0_PUBLISHED` literal, NOT from + // `crypto::AACS_IV`: computing it from the production constant made + // the change cancel out of the triple XOR, so a swap to [0u8;16] still + // passed. It now fails. let key = [ 0x2B, 0x7E, 0x15, 0x16, 0x28, 0xAE, 0xD2, 0xA6, 0xAB, 0xF7, 0x15, 0x88, 0x09, 0xCF, 0x4F, 0x3C, @@ -1268,7 +1371,7 @@ mod tests { // Block 0: NIST_PT[0] XOR NIST_IV XOR AACS_IV (the fixed-IV substitution). let mut expected_block0 = [0u8; 16]; for i in 0..16 { - expected_block0[i] = nist_plaintext[i] ^ nist_iv[i] ^ AACS_IV[i]; + expected_block0[i] = nist_plaintext[i] ^ nist_iv[i] ^ IV0_PUBLISHED[i]; } assert_eq!( &buf[0..16], diff --git a/src/aacs/crypto.rs b/src/aacs/crypto.rs index c588044..43940f5 100644 --- a/src/aacs/crypto.rs +++ b/src/aacs/crypto.rs @@ -15,6 +15,33 @@ pub(crate) const AACS_IV: [u8; 16] = [ 0x0B, 0xA0, 0xF8, 0xDD, 0xFE, 0xA6, 0x1F, 0xB3, 0xD8, 0xDF, 0x9F, 0x56, 0x6A, 0x05, 0x0F, 0x78, ]; +// Per-thread count of AES-128 key schedules built through `new_cipher`. +// Test-only instrumentation: an AES-128 key expansion is 10 round-key +// derivations, and the CBC helpers here run on the per-aligned-unit decrypt hot +// path of a whole disc read, so "how many times was the schedule built for one +// loop-invariant key" is a property worth asserting rather than reasoning about. +// THREAD-LOCAL, not a global atomic: `cargo test` runs tests concurrently, so a +// shared counter would see every other test's expansions. See +// `content::tests::decrypt_bus_expands_the_read_data_key_once_per_unit`. +#[cfg(test)] +thread_local! { + pub(crate) static KEY_EXPANSIONS: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + +/// Build an AES-128 key schedule for a caller that will drive +/// [`cbc_decrypt_blocks`] over several regions under one key. +pub(crate) fn new_cipher_for(key: &[u8; 16]) -> Aes128 { + new_cipher(key) +} + +/// Build an AES-128 key schedule. The single construction site for the CBC +/// helpers, so [`KEY_EXPANSIONS`] can count them under test. +fn new_cipher(key: &[u8; 16]) -> Aes128 { + #[cfg(test)] + KEY_EXPANSIONS.with(|c| c.set(c.get() + 1)); + Aes128::new(GenericArray::from_slice(key)) +} + /// AES-128-ECB encrypt a single 16-byte block. [C] §2.1.1 (`AES-128E`). pub(crate) fn aes_ecb_encrypt(key: &[u8; 16], data: &[u8; 16]) -> [u8; 16] { let cipher = Aes128::new(GenericArray::from_slice(key)); @@ -35,13 +62,12 @@ pub(crate) fn aes_ecb_decrypt(key: &[u8; 16], data: &[u8; 16]) -> [u8; 16] { out } -/// AES-128-CBC decrypt in-place with the fixed AACS IV. [C] §2.1.2 (`AES-128CBCD`). +/// AES-128-CBC ENCRYPT in place under the fixed [`AACS_IV`] — the forward +/// direction of [`aes_cbc_decrypt`], and its exact inverse. [C] §2.1.2 +/// (`AES-128CBCE`). /// -/// Precondition: `data.len()` is a multiple of 16. Any trailing partial -/// block is silently ignored; all callers pass aligned regions (6128 and -/// 2032 bytes), and the assert documents/enforces that contract. -/// AES-128-CBC encrypt in place under the fixed [`AACS_IV`] — the forward -/// direction of [`aes_cbc_decrypt`], and its exact inverse. +/// Precondition: `data.len()` is a multiple of 16; the assert +/// documents/enforces that contract. /// /// Constructs the cipher ONCE for the whole slice. Driving this from the /// single-block [`aes_ecb_encrypt`] instead rebuilds the AES key schedule per @@ -52,7 +78,7 @@ pub(crate) fn aes_cbc_encrypt(key: &[u8; 16], data: &mut [u8]) { data.len().is_multiple_of(16), "aes_cbc_encrypt requires a block-aligned slice" ); - let cipher = Aes128::new(GenericArray::from_slice(key)); + let cipher = new_cipher(key); let num_blocks = data.len() / 16; let mut prev = AACS_IV; // Forward order: each block is XORed with the PRECEDING ciphertext block. @@ -69,12 +95,40 @@ pub(crate) fn aes_cbc_encrypt(key: &[u8; 16], data: &mut [u8]) { } } +/// AES-128-CBC DECRYPT in-place with the fixed AACS IV. [C] §2.1.2 +/// (`AES-128CBCD`). +/// +/// Precondition: `data.len()` is a multiple of 16. Any trailing partial +/// block is silently ignored; all callers pass aligned regions (6128 and +/// 2032 bytes), and the assert documents/enforces that contract. +/// +/// (This doc block was orphaned onto `aes_cbc_encrypt` above when that function +/// was inserted directly after it with no separating blank line, so rustdoc +/// rendered the crate's only forward-direction AACS primitive as "decrypt" and +/// cited the spec's DECRYPT clause for it, while this function had no doc at +/// all. `encrypt_unit_is_the_exact_inverse_of_decrypt_unit` in `content.rs` pins +/// the directions behaviourally so a maintainer 'fixing' the contradiction by +/// swapping the two bodies fails the suite instead of shipping a second +/// decryptor behind an already-set encrypted flag.) pub(crate) fn aes_cbc_decrypt(key: &[u8; 16], data: &mut [u8]) { debug_assert!( data.len().is_multiple_of(16), "aes_cbc_decrypt requires a block-aligned slice" ); - let cipher = Aes128::new(GenericArray::from_slice(key)); + cbc_decrypt_blocks(&new_cipher(key), data); +} + +/// AES-128-CBC decrypt in place under the fixed [`AACS_IV`] with an ALREADY +/// EXPANDED key schedule. +/// +/// Split out of [`aes_cbc_decrypt`] so a caller that decrypts several regions +/// under one loop-invariant key expands the schedule once. `decrypt_bus` +/// ([`super::content::decrypt_bus`]) is that caller: bus encryption +/// ([C] §4.2 / the AACS 2.0 Read Data Key) covers bytes 16..2048 of EVERY +/// 2048-byte sector, so a 6144-byte aligned unit is three regions under one +/// `read_data_key` — three key schedules where one suffices, on the per-unit +/// decrypt hot path of a whole 90 GB read. +pub(crate) fn cbc_decrypt_blocks(cipher: &Aes128, data: &mut [u8]) { let num_blocks = data.len() / 16; // Process blocks in reverse to avoid clobbering ciphertext needed for XOR for i in (0..num_blocks).rev() { diff --git a/src/decrypt.rs b/src/decrypt.rs index 10ff1b5..94d0e99 100644 --- a/src/decrypt.rs +++ b/src/decrypt.rs @@ -1240,6 +1240,62 @@ mod tests { } } + /// The mapped descramble indexes the committed key pool POSITIONALLY + /// (`unit_keys[key_idx].1`), so the ORDER of the `Vec` a + /// `KeySource` returns is load-bearing — it is NOT "cosmetic, the decrypt path + /// strips it and tries every key", as `keysource::resolve_and_apply_traced`'s + /// doc used to claim. Trial-decrypt was deliberately deleted; nothing here + /// searches the pool. Reordering the same two keys therefore sends each range + /// to the WRONG key: the range that decrypted clean now fails the correct-phase + /// `is_clean` net loudly (or, off a forensic phase, would decrypt a whole span + /// under a neighbour's key). Pins the corrected doc. + #[test] + fn mapped_key_selection_is_positional_so_pool_order_matters() { + use crate::disc::ContentFormat; + let key_a = [0xAAu8; 16]; + let key_b = [0xBBu8; 16]; + let ul = aacs::content::ALIGNED_UNIT_LEN; + let usz = (ul / 2048) as u32; + // Unit 0 encrypted under key_a, unit 1 under key_b. + let build = || { + let mut buf = vec![0u8; 2 * ul]; + for (i, k) in [key_a, key_b].iter().enumerate() { + let mut u = clear_ts_unit(); + aacs_encrypt_unit_for_test(&mut u, k); + buf[i * ul..(i + 1) * ul].copy_from_slice(&u); + } + buf + }; + // Map: unit 0 → pool position 0, unit 1 → pool position 1. + let map = AacsKeyMap::from_ranges_phased(vec![ + (0, usz, 0, Phase::Even), + (usz, 2 * usz, 1, Phase::Even), + ]); + // Pool in CPS-unit order: each range gets its own key, both come clean. + let mut buf = build(); + let keys = DecryptKeys::Aacs { + unit_keys: vec![(0, key_a), (1, key_b)], + read_data_key: None, + format: ContentFormat::BdTs, + }; + decrypt_sectors_mapped(&mut buf, &keys, 0, &map) + .expect("pool in CPS-unit order decrypts clean"); + // SAME keys, SAME CPS-unit numbers, swapped POSITIONS. If the number were + // what mattered (or if the path searched the pool) this would be + // equivalent; positional indexing makes it decrypt both units wrong. + let mut buf = build(); + let swapped = DecryptKeys::Aacs { + unit_keys: vec![(1, key_b), (0, key_a)], + read_data_key: None, + format: ContentFormat::BdTs, + }; + assert!( + decrypt_sectors_mapped(&mut buf, &swapped, 0, &map).is_err(), + "a reordered pool must fail loud — key selection is positional, so the \ + ORDER a KeySource returns its keys in is part of the contract" + ); + } + /// The correct-phase safety `is_clean` fires loud: an even unit whose mapped /// key is wrong does NOT come clean → `DecryptFailed` (not silent corruption). #[test] diff --git a/src/drive/mod.rs b/src/drive/mod.rs index 3ca261e..e6079b3 100644 --- a/src/drive/mod.rs +++ b/src/drive/mod.rs @@ -781,9 +781,34 @@ impl Drive { // each a self-contained READ(10) with the same validation. Any // chunk error reports that chunk's LBA (more precise than the whole // request's base LBA). + let count = count as u32; + // Check the caller's buffer ONCE, up front. The chunk loop slices `buf` + // by `count * 2048`; without this an undersized buffer PANICKED ('range + // end index out of range') out of the public `read`/`read_fua`, while the + // single-chunk path above tolerates the same undersized buffer and returns + // `Err(DiscRead)` from `checked_exec`. Behaviour on an undersized buffer + // must not depend on the transport's transfer limit. + if buf.len() < count as usize * 2048 { + return Err(Error::DiscRead { + sector: lba as u64, + status: None, + sense: None, + }); + } + // The whole range must be addressable: SBC-3 READ(10) carries a 32-bit + // LOGICAL BLOCK ADDRESS, so a request whose last chunk crosses `u32::MAX` + // has no valid CDB. `lba + done` below was unchecked — a debug panic out + // of the public API, and in release a wrap to a low LBA that was read and + // returned as if it were the requested one. + if lba.checked_add(count.saturating_sub(1)).is_none() { + return Err(Error::DiscRead { + sector: lba as u64, + status: None, + sense: None, + }); + } let mut done: u32 = 0; let mut total: usize = 0; - let count = count as u32; while done < count { let chunk = (count - done).min(max_sectors); let cur_lba = lba + done; @@ -1935,6 +1960,54 @@ mod command_tests { assert_eq!(*reads.lock().unwrap(), vec![(0, 3)], "single CDB, no split"); } + /// The multi-chunk path slices the caller's buffer by `count * 2048` with no + /// length check, so an undersized `buf` PANICKED ('range end index out of + /// range') out of the public `Drive::read` / `Drive::read_fua` — while the + /// single-chunk path (`read_one` → `checked_exec`) tolerates the same + /// undersized buffer and returns `Err(DiscRead)`. The public API's behaviour + /// on an undersized buffer must not depend on the transport's transfer limit. + #[test] + fn undersized_buffer_multi_chunk_errors_not_panics() { + let ChunkingHarness { + drive: mut d, + reads: _reads, + } = chunking(4 * 2048, None); + // count (10) > max_sectors (4) → the chunk loop; buf holds only 1 sector. + let mut buf = vec![0u8; 2048]; + assert!( + matches!(d.read(0, 10, &mut buf, false), Err(Error::DiscRead { .. })), + "an undersized buffer must error, not panic" + ); + // The single-chunk path with the SAME undersized buffer already errored; + // the two paths must now agree. + let mut buf = vec![0u8; 2048]; + assert!( + matches!(d.read(0, 3, &mut buf, false), Err(Error::DiscRead { .. })), + "single-chunk path errors on an undersized buffer (unchanged)" + ); + } + + /// `Drive::read`'s chunk loop advanced the per-chunk LBA with an unchecked + /// `lba + done`. SBC-3 READ(10) `LOGICAL BLOCK ADDRESS` is a 32-bit field, so + /// a request whose last chunk crosses `u32::MAX` overflowed: debug panic out + /// of the public API, release wrap to a low LBA silently read instead. + #[test] + fn chunk_lba_near_u32_max_errors_not_overflows() { + let ChunkingHarness { + drive: mut d, + reads: _reads, + } = chunking(4 * 2048, None); + let mut buf = vec![0u8; 10 * 2048]; + // 0xFFFF_FFFE + 4 overflows on the second chunk. + assert!( + matches!( + d.read(0xFFFF_FFFE, 10, &mut buf, false), + Err(Error::DiscRead { .. }) + ), + "an LBA range past u32::MAX must error, not overflow" + ); + } + // ── find_drive media-preference selection policy ──────────────── /// Build a fake drive whose GET EVENT STATUS reply reports the given diff --git a/src/io/pipeline.rs b/src/io/pipeline.rs index 8655b34..f05655a 100644 --- a/src/io/pipeline.rs +++ b/src/io/pipeline.rs @@ -125,7 +125,10 @@ fn consumer_panicked(payload: Box) -> Error { /// caller reporting the rip as interrupted while a fully finalised MKV (Cues /// written, Segment size patched) landed on disk, indistinguishable from a /// complete one. The two transitions are therefore a single compare-exchange each, -/// out of [`ST_RUNNING`]: whoever wins decides, and the loser observes the winner. +/// out of [`state::RUNNING`]: whoever wins decides, and the loser observes the +/// winner. (`ST_RUNNING` does not exist anywhere in the crate — the constants are +/// `state::RUNNING` / `state::ABANDONED` / `state::CLOSING` below, and both +/// compare-exchange sites that must stay in step with this argument name them.) mod state { /// Consumer is running; neither side has committed yet. pub const RUNNING: u8 = 0; diff --git a/src/io/writeback_file/macos.rs b/src/io/writeback_file/macos.rs index e45a34a..1fa4ff2 100644 --- a/src/io/writeback_file/macos.rs +++ b/src/io/writeback_file/macos.rs @@ -112,8 +112,26 @@ pub(super) fn durable_sync(file: &File) -> io::Result<()> { ); Ok(()) } - Err(crate::io::bounded::BoundedError::Halted) => Ok(()), - Err(crate::io::bounded::BoundedError::WorkerLost) => Ok(()), + // Both arms below used to map to `Ok(())` with NO diagnostic at all, while + // the Linux sibling logs the identical failures (writeback_file/linux.rs). + // A lost F_FULLFSYNC worker at the end of a UHD mux therefore reported + // `completed = true` with an empty log, leaving an operator investigating a + // truncated/corrupt output file after a power loss no record that the final + // fsync never ran — on Linux the same failure is at error level. + Err(crate::io::bounded::BoundedError::Halted) => { + tracing::warn!( + target: "mux", + "WritebackFile::sync_all F_FULLFSYNC skipped (halt requested); data not durably flushed, kernel will flush on close" + ); + Ok(()) + } + Err(crate::io::bounded::BoundedError::WorkerLost) => { + tracing::error!( + target: "mux", + "WritebackFile::sync_all F_FULLFSYNC worker lost before completion; data not durably flushed, kernel will flush on close" + ); + Ok(()) + } } } diff --git a/src/keysource.rs b/src/keysource.rs index 3922807..34f0e8d 100644 --- a/src/keysource.rs +++ b/src/keysource.rs @@ -53,9 +53,30 @@ pub const MIN_SAMPLE_UNITS: usize = 8; /// units it yields); the *requested* count is a caller-side compile-time constant that /// callers pin to `MIN_SAMPLE_UNITS` (see e.g. autorip's `SAMPLE_UNITS`). Together the /// two make under-sampling unrepresentable at the request boundary. -#[derive(Debug, Clone)] +/// +/// The wrapped samples are on-disc AACS ciphertext — the same bytes the sibling +/// [`DiscInputs::samples`] redacts as key MATERIAL — so [`Debug`] is hand-written +/// and redacting; see the impl below. +#[derive(Clone)] pub struct DecodeSampleSet(Vec>); +impl std::fmt::Debug for DecodeSampleSet { + /// Prints the SHAPE only. A derived `Debug` dumped every wrapped sample + /// verbatim: a `DecodeSampleSet` carries at least [`MIN_SAMPLE_UNITS`] + /// 6144-byte aligned units (≥ 49 KiB, in practice multi-MB) of AACS + /// ciphertext plus each unit's clear 16-byte derivation seed, so one + /// `tracing::debug!("{set:?}")` on a failed `/decode` request — or an + /// `assert_eq!` whose panic message formats it — wrote all of it to the log + /// that gets attached to a bug report. Same policy and same shape as + /// [`DiscInputs`]'s impl below. + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DecodeSampleSet") + .field("units", &"") + .field("units_len", &self.0.len()) + .finish() + } +} + impl DecodeSampleSet { /// Wrap `units` iff it carries at least [`MIN_SAMPLE_UNITS`] samples; `None` /// otherwise (the caller then skips the online source rather than sending an @@ -334,8 +355,23 @@ pub fn resolve_and_apply( /// from [`crate::aacs::derive::decrypt_unit_key`]; the library's canonical CPS-unit number is /// `position + 1` (matching [`crate::aacs::inf::parse_unit_key_ro`]'s `(i + 1)`), so /// the committed `AacsState.unit_keys` is byte-identical to the library-resolved -/// path. The number is cosmetic for descramble (the decrypt path strips it and -/// tries every key) but is kept faithful to the resolver's convention. +/// path. +/// +/// The NUMBER itself is not what descramble indexes by — but the ORDER is +/// load-bearing, so a source must return its keys in CPS-unit order. Trial +/// decrypt-and-check was deliberately deleted (see +/// [`crate::decrypt::AacsKeyMap`]: decryption is driven by the disc's CPS-unit / +/// FMTS-segment structure, "never by trial-decrypt-and-check per unit"), and +/// `decrypt_sectors_mapped` indexes the committed pool POSITIONALLY — +/// `unit_keys[key_idx].1`, where `key_idx` is a POSITION in the Vec a source +/// returned, recorded by `resolve_mux_key_map_cached` / `resolve_fmts_key_map`. +/// Return the same keys in a different order and every `AacsKeyMap` points at the +/// wrong key: the whole title decrypts under a neighbour's key, or a forensic +/// range trips the `is_clean` net into `DecryptFailed`. (The doc used to say the +/// number "is cosmetic for descramble (the decrypt path strips it and tries every +/// key)", which is what the DELETED trial-decrypt path did; the only place that +/// still tries every key is `Disc::decrypt_with`'s sample VALIDATION, which does +/// not descramble content.) pub fn resolve_and_apply_traced( sources: &[Box], inputs: &DiscInputs, @@ -400,8 +436,15 @@ pub fn resolve_and_apply_traced( // boundary in `freemkv-keysources`, so a failure is reported as a // failure, with `Disc::aacs_error` as the channel the operator actually // reads. `FetchOutcome::errored` in `drive_unit_keys` / - // `drive_fmts_indexes` is dead for the same reason — it is the right - // contract, honoured by no source yet. + // `drive_fmts_indexes` never FIRES for the same reason — it is the + // right contract, honoured by no shipped source yet. It is NOT dead + // code: it is written at both `drive_*` sites and read by the + // cache-insert guard `if !keys.is_empty() || !outcome.errored`, the + // only thing that stops a transient source outage from being memoised + // permanently into the per-fingerprint key cache — pinned by + // `errored_empty_is_not_cached_and_retries_when_source_recovers`. Do + // not delete it while making a source report failures as `Err`; that + // is precisely when it starts to matter. Ok(_) | Err(_) => { trace.keys.push(KeyStep { who, @@ -1396,4 +1439,28 @@ mod tests { assert!(dbg.contains("samples_len: 1"), "{dbg}"); assert!(dbg.contains("TITLE_2024"), "{dbg}"); } + + /// `DecodeSampleSet` is public and wraps the SAME on-disc ciphertext the + /// sibling `DiscInputs` redacts, so a derived `Debug` dumped ≥ MIN_SAMPLE_UNITS + /// × 6144 bytes of verbatim AACS ciphertext (plus every unit's clear 16-byte + /// derivation seed) into any log that formatted it. Sentinel byte 0xD5 = + /// decimal 213, matching `aacs::types::redaction_tests` and the + /// `DiscInputs` test above. Mutation guard: restoring `#[derive(Debug)]` + /// fails this. + #[test] + fn decode_sample_set_debug_is_redacted() { + let set = DecodeSampleSet::new(vec![vec![0xD5; 6144]; MIN_SAMPLE_UNITS]) + .expect("MIN_SAMPLE_UNITS units is a valid set"); + let dbg = format!("{set:?}"); + assert!( + !dbg.contains("213"), + "DecodeSampleSet Debug leaked ciphertext (decimal 213): {dbg}" + ); + assert!( + dbg.contains("redacted"), + "DecodeSampleSet Debug missing redaction marker: {dbg}" + ); + // Non-secret shape stays printable for diagnostics. + assert!(dbg.contains("units_len: 8"), "{dbg}"); + } } diff --git a/src/mux/au_assembly.rs b/src/mux/au_assembly.rs index 03a23dd..2bce747 100644 --- a/src/mux/au_assembly.rs +++ b/src/mux/au_assembly.rs @@ -126,6 +126,12 @@ pub(crate) struct AuAssembler { /// so a long run of junk with no start code (hostile/corrupt input) costs /// O(bytes) total, not O(buffer) per push. Reset when `buf[0]` moves. opener_pos: usize, + /// Test-only: how many times `take_front` fell back to the COPY path. The + /// handover is the whole point of `take_front`, so "did it actually fire" is a + /// property to MEASURE, not to reason about. See + /// `handover_survives_a_large_au_instead_of_copying_every_later_one`. + #[cfg(test)] + copy_path_hits: usize, } impl AuAssembler { @@ -155,6 +161,8 @@ impl AuAssembler { scan_pos: 0, seen_unit: false, opener_pos: 0, + #[cfg(test)] + copy_path_hits: 0, } } @@ -172,6 +180,8 @@ impl AuAssembler { scan_pos: 0, seen_unit: false, opener_pos: 0, + #[cfg(test)] + copy_path_hits: 0, } } @@ -356,14 +366,38 @@ impl AuAssembler { /// (a small AU after a multi-MB one): handing over would otherwise attach an /// oversized idle allocation to a small frame for as long as the frame queues /// downstream, trading a copy for resident memory. + /// + /// That fallback must not become permanent. `buf`'s capacity used to be a + /// one-way high-water mark — the replacement buffer was created with + /// `cap.max(tail_len)`, and the copy path's `drain` also preserves `cap` — so + /// once ONE large AU had been assembled, every later smaller AU satisfied + /// `cap > 2*end` and took the copy path forever. On a UHD HEVC title the first + /// IDR grows `buf` to ~4-8 MB, after which each ~200-400 KB P/B AU paid a + /// whole-AU allocation plus a whole-AU memcpy plus a tail memmove for ~99% of + /// the ~200,000 coded pictures — tens of GB of exactly the memcpy this handover + /// exists to remove. So the copy path now also RELEASES the high-water + /// capacity, which re-arms the handover for the next AU: one copy after a size + /// step down, not one per frame forever. fn take_front(&mut self, end: usize) -> Vec { let cap = self.buf.capacity(); + let tail_len = self.buf.len() - end; if cap > end.saturating_mul(2) { + #[cfg(test)] + { + self.copy_path_hits += 1; + } let data = self.buf[..end].to_vec(); self.buf.drain(..end); + // Shrink toward what this AU actually needed (the tail plus room for + // another AU of about this size). Only the short tail is copied, and it + // brings `cap` back under the `2*end` threshold so the next AU of this + // size hands over instead of copying. + self.buf.shrink_to(end.max(tail_len)); return data; } - let mut tail = Vec::with_capacity(cap.max(self.buf.len() - end)); + // Replacement buffer: enough for the tail plus room to accumulate the next + // AU of about this size. NOT `cap`, which would re-pin the high-water mark. + let mut tail = Vec::with_capacity(end.max(tail_len)); tail.extend_from_slice(&self.buf[end..]); let mut data = std::mem::replace(&mut self.buf, tail); data.truncate(end); @@ -915,11 +949,56 @@ mod tests { before, "the emitted AU must own the buffer's allocation (no whole-frame copy)" ); - assert_eq!( + // The replacement buffer keeps room for another AU of about this size, so + // the next AU does not re-grow — but it is NOT pinned to the OLD capacity, + // which would make `buf` a permanent high-water mark and send every later + // smaller AU down the copy path (see + // `handover_survives_a_large_au_instead_of_copying_every_later_one`). + assert!( + a.buf.capacity() >= au1.len(), + "replacement buffer must fit another AU of this size: {} < {}", a.buf.capacity(), - cap_before, - "the replacement buffer keeps the capacity, so the next AU does not re-grow" + au1.len() + ); + assert!( + a.buf.capacity() <= cap_before, + "replacement buffer must never EXCEED the old capacity" ); assert_eq!(a.buf.len(), 4, "the buffer holds only AU2's delimiter tail"); } + + /// MEASURED: `take_front`'s copy fallback must not become permanent. + /// + /// `buf`'s capacity used to be a one-way high-water mark, and the copy path's + /// `drain` preserves it, so after ONE large AU every later smaller AU satisfied + /// `cap > 2*end` and copied forever. On a UHD HEVC title the first IDR grows + /// `buf` to multiple MB, after which ~99% of the ~200,000 coded pictures each + /// paid a whole-AU allocation + whole-AU memcpy + tail memmove — tens of GB of + /// exactly the copy the handover exists to remove. Counted at the copy path + /// itself: one copy is expected right after the size step down; a per-frame + /// copy is the bug. + #[test] + fn handover_survives_a_large_au_instead_of_copying_every_later_one() { + let mut a = AuAssembler::for_codec(Codec::H264); + // Production shape: BD-TS aligns one access unit per PES, so each `push` + // carries about one AU and the buffer holds ~one AU at a time. One large AU + // (the IDR) followed by a run of much smaller ones (P/B frames). Each AU is + // pushed with the NEXT AU's opener so the previous one closes. + const SMALL: usize = 64 * 1024; + let mut pending = au(0x11, 2 * 1024 * 1024); + for i in 0..20u8 { + let next = au(0x30 + i, SMALL); + // Append the next AU's 4-byte opener to close `pending`, push, and + // carry the rest of `next` forward. + pending.extend_from_slice(&next[..4]); + a.push(&pending, Some(1), None, None, false); + pending = next[4..].to_vec(); + } + let hits = a.copy_path_hits; + assert!( + hits <= 2, + "the copy fallback must re-arm the handover, not fire for every AU \ + after a large one: {hits} copies over 20 access units" + ); + } } diff --git a/src/mux/codec/hevc.rs b/src/mux/codec/hevc.rs index 69634ea..9a8fdf2 100644 --- a/src/mux/codec/hevc.rs +++ b/src/mux/codec/hevc.rs @@ -138,13 +138,28 @@ pub struct HevcParser { // is consumed (cleared) by that first CRA so only ONE CRA per boundary is // touched — never a mid-stream CRA, never an IDR, never a non-CRA NAL. // - // SAFETY: defaults to `false` and is ONLY ever set through - // `mark_clip_boundary`, which the caller invokes ONLY for a non-seamless - // (0x05/0x06) join. connection_condition 0x01 is the first-item/seamless - // case and must NOT trigger this flag. A stream with no boundary marker - // (single-clip title, or seamless-joined 0x01 UHD/BD) never has this set, - // so the rewrite branch is never reached and output is byte-identical to a - // parser without this field. + // ARMED BY TWO PATHS — do not read this flag as caller-driven only: + // + // 1. `mark_clip_boundary`, which a caller invokes only for a non-seamless + // (0x05/0x06) join. connection_condition 0x01 is the first-item/seamless + // case and must NOT trigger this flag. In practice NO caller wires this + // up: the mpls connection_condition is not plumbed through the threaded + // mux pipeline (see the note at the auto-detect site in `parse`). + // 2. The in-parser PTS-backstep AUTO-DETECTION in `parse` — a backward PES + // PTS step beyond `BACKSTEP_TICKS` sets it with no caller involvement. + // This is the path that actually fires in production, and it is the one + // the CRA→BLA rewrite exists for. + // + // So the rewrite branch is NOT dead, and output is NOT byte-identical to a + // parser without this field: any stream whose PES PTS steps backward by more + // than `BACKSTEP_TICKS` — including a damaged/rewritten PTS field on an + // untrusted disc — has its next CRA_NUT rewritten to BLA_W_LP, which makes a + // decoder discard that CRA's valid RASL leading pictures. That false-arming + // risk is held down by the `PTS_WRAP_PERIOD` unwrapping and the high-water + // watermark, not by the flag being unreachable: REMOVING either guard on the + // strength of "only `mark_clip_boundary` sets this" is a live corruption bug. + // Pinned by `cra_at_auto_detected_pts_backstep_rewritten_to_bla` and + // `cra_after_33bit_pts_wrap_not_rewritten`. pending_clip_boundary: bool, // Highest PES PTS seen on this video stream so far, on a MONOTONIC 64-bit // timeline (raw 33-bit PTS unwrapped across 2^33 wraparounds — see @@ -216,6 +231,24 @@ const _: () = assert!( "HEVC BACKSTEP_TICKS must mirror mux::timeline::DISCONTINUITY_BACKSTEP_NS" ); +// Bytes reserved at the front of every assembled access unit so the keyframe +// parameter-set re-assert can be spliced in without reallocating. A VPS + SPS + +// PPS re-assert is a few hundred bytes (each a 4-byte length prefix plus a NAL +// that is tens to low hundreds of bytes on real BD/UHD streams); 1 KiB covers it +// with margin, and costs 1 KiB of slack per in-flight frame. If a stream's +// parameter sets ever exceed this the splice still produces correct output — it +// just reallocates once, exactly as it always did. +const PARAM_REASSERT_HEADROOM: usize = 1024; + +// Per-thread count of keyframe re-asserts that had to reallocate the frame buffer. +// Test-only instrumentation: the whole point of `PARAM_REASSERT_HEADROOM` is that +// the splice is in-place, so that is MEASURED rather than reasoned about. See +// `keyframe_param_reassert_does_not_reallocate_the_frame`. +#[cfg(test)] +thread_local! { + static PARAM_REASSERT_REALLOCS: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + // The 33-bit 90 kHz PES PTS counter wraps at 2^33 ticks (~26.5 h). When the raw // PTS steps backward by approximately a full period — i.e. it landed just past // the wrap — it is a counter wraparound, NOT a clip reset: unwrap it (add 2^33) @@ -302,6 +335,16 @@ impl HevcParser { /// Unknown payload types are skipped by their size so a later HDR10 message /// in the same NAL is still reached. fn scan_sei(&mut self, nal: &[u8]) { + // Both HDR10 messages are per-stream constants and STICKY (first seen + // wins), so once both are captured every remaining match arm below + // declines and the whole scan is a guaranteed no-op. Return BEFORE + // `strip_emulation_prevention`, which allocates and byte-copies the entire + // SEI RBSP: an HDR10 UHD stream carries a prefix SEI per access unit, so + // without this the other ~200,000 access units of a title each paid one + // allocation and one copy for a result that is discarded. + if self.sei_mastering.is_some() && self.sei_content_light.is_some() { + return; + } let Some(raw) = nal.get(2..) else { return; }; @@ -532,7 +575,11 @@ impl CodecParser for HevcParser { // Pre-size: output is ~input bytes with a few 4-byte length // prefixes added. UHD frames are 150-300 KB; the unsized Vec // growth chain otherwise reallocs 5-7× per frame. - let mut frame_data = Vec::with_capacity(data.len() + 64); + // + // Plus `PARAM_REASSERT_HEADROOM` so the keyframe parameter-set re-assert + // below can be spliced in FRONT of the frame without reallocating. See + // that site. + let mut frame_data = Vec::with_capacity(data.len() + 64 + PARAM_REASSERT_HEADROOM); // Single-pass NAL scan: extract params, detect keyframes, build length-prefixed output let mut pos = 0; @@ -670,13 +717,34 @@ impl CodecParser for HevcParser { // when active == codecPrivate) so each keyframe is self-contained and a // decoder that dropped the set (CRA reset / SPS event) self-heals. if keyframe { - let mut prefix = Vec::new(); + let mut prefix = Vec::with_capacity(PARAM_REASSERT_HEADROOM); reassert_active(&mut prefix, &self.cur_vps, emitted_vps); reassert_active(&mut prefix, &self.cur_sps, emitted_sps); reassert_active(&mut prefix, &self.cur_pps, emitted_pps); if !prefix.is_empty() { - prefix.extend_from_slice(&frame_data); - frame_data = prefix; + // SPLICE the few hundred prefix bytes into the front of the + // already-assembled frame, in place. + // + // This used to be `prefix.extend_from_slice(&frame_data)` followed + // by `frame_data = prefix`: that grew `prefix` from a few hundred + // bytes to the FULL access-unit size (a fresh multi-MB allocation), + // memcpy'd the whole frame into it, and dropped the presized + // `frame_data` buffer — one extra whole-frame allocation plus one + // extra whole-frame copy per keyframe. A 2 h UHD title at 24 fps + // with a 1 s GOP is ~7,200 keyframes, i.e. ~7,200 multi-MB + // allocations and ~14-28 GB of avoidable memcpy per title. + // + // `frame_data` was reserved with `PARAM_REASSERT_HEADROOM` to spare + // precisely so this splice fits without reallocating; what remains + // is one in-place memmove inside the existing buffer. Byte-identical + // output either way. + #[cfg(test)] + let cap_before = frame_data.capacity(); + frame_data.splice(0..0, prefix); + #[cfg(test)] + if frame_data.capacity() != cap_before { + PARAM_REASSERT_REALLOCS.with(|c| c.set(c.get() + 1)); + } } } @@ -839,9 +907,22 @@ struct SpsChroma { temporal_id_nesting_flag: u8, } +// Per-thread count of `strip_emulation_prevention` calls. Test-only +// instrumentation: the function allocates and byte-copies a whole RBSP, and +// `scan_sei` used to run it for every SEI NAL of every access unit, so "how many +// copies did a stream actually cost" is worth MEASURING rather than reasoning +// about. Thread-local, not a global atomic, because `cargo test` runs tests +// concurrently. See `scan_sei_stops_copying_once_both_hdr10_messages_are_captured`. +#[cfg(test)] +thread_local! { + static RBSP_COPIES: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + /// Strip HEVC/H.264 emulation-prevention bytes (00 00 03 → 00 00) from a NAL /// RBSP so a bit reader sees the true coded values. fn strip_emulation_prevention(rbsp: &[u8]) -> Vec { + #[cfg(test)] + RBSP_COPIES.with(|c| c.set(c.get() + 1)); let mut out = Vec::with_capacity(rbsp.len()); let mut zeros = 0usize; for &b in rbsp { @@ -1169,6 +1250,67 @@ mod tests { assert_eq!(h.max_pic_average_light_level, maxfall); } + /// MEASURED, not reasoned: an HDR10 stream carries a prefix SEI per access + /// unit, and `scan_sei` allocated + byte-copied the whole SEI RBSP through + /// `strip_emulation_prevention` on EVERY one — including after both HDR10 + /// messages were already captured and every match arm was guaranteed to + /// decline. On a ~200,000-frame UHD title that is ~200,000 allocations and + /// copies for a discarded result. Counted at the single + /// `strip_emulation_prevention` site. + #[test] + fn scan_sei_stops_copying_once_both_hdr10_messages_are_captured() { + let pps = { + let mut v = vec![0x00, 0x00, 0x01]; + v.extend_from_slice(&hevc_nal_header(NAL_PPS)); + v.push(0xC0); + v + }; + let idr = { + let mut v = vec![0x00, 0x00, 0x01]; + v.extend_from_slice(&hevc_nal_header(19)); + v.push(0xEC); + v + }; + // Every AU carries BOTH HDR10 SEI messages, as a real HDR10 stream does. + let au = || { + let mut data = pps.clone(); + data.extend_from_slice(&sei_nal(&[ + sei_message( + SEI_MASTERING_DISPLAY_COLOUR_VOLUME, + &mastering_payload([1, 2, 3], [4, 5, 6], 7, 8, 9, 10), + ), + sei_message(SEI_CONTENT_LIGHT_LEVEL_INFO, &cll_payload(1000, 400)), + ])); + data.extend_from_slice(&idr); + data + }; + + let mut parser = HevcParser::new(); + // First AU: both messages captured, so this one legitimately copies. + parser.parse(&make_pes(au(), Some(0))); + assert!( + parser.sei_mastering.is_some() && parser.sei_content_light.is_some(), + "first AU must capture both HDR10 messages" + ); + // Now measure the next 50 AUs, whose SEI scan is a guaranteed no-op. + RBSP_COPIES.with(|c| c.set(0)); + for i in 0..50 { + parser.parse(&make_pes(au(), Some(3750 * (i + 1)))); + } + let copies = RBSP_COPIES.with(|c| c.get()); + assert_eq!( + copies, 0, + "SEI RBSP must not be copied once both HDR10 messages are captured; \ + {copies} copies over 50 access units" + ); + // And the captured metadata is still surfaced on those later frames. + let f = parser.parse(&make_pes(au(), Some(3750 * 51))); + assert!( + f[0].coding.unwrap().hdr10().is_some(), + "the sticky HDR10 metadata must still ride every later frame" + ); + } + /// Only the mastering-display SEI (no content-light SEI) → metadata is NOT /// surfaced. HDR10 requires BOTH; a half-populated record is never emitted. #[test] @@ -1510,6 +1652,60 @@ mod tests { ); } + /// MEASURED: the keyframe parameter-set re-assert must be spliced into the + /// front of the already-assembled access unit IN PLACE, not built as a fresh + /// full-size buffer. + /// + /// It used to `prefix.extend_from_slice(&frame_data)` and then replace + /// `frame_data` with `prefix`, which grew a few-hundred-byte `prefix` to the + /// FULL access-unit size — a fresh multi-MB allocation — memcpy'd the whole + /// frame into it, and dropped the presized buffer. One extra whole-frame + /// allocation plus one extra whole-frame copy per keyframe: a 2 h UHD title at + /// 24 fps with a 1 s GOP is ~7,200 keyframes, ~14-28 GB of avoidable memcpy per + /// title. `PARAM_REASSERT_HEADROOM` exists so the splice never reallocates; + /// this counts the reallocations that happen, which must be zero. + #[test] + fn keyframe_param_reassert_does_not_reallocate_the_frame() { + fn nal(t: u8, body: &[u8]) -> Vec { + let mut v = vec![0x00, 0x00, 0x01]; + v.extend_from_slice(&hevc_nal_header(t)); + v.extend_from_slice(body); + v + } + let sps_body = [0x01u8; 24]; + let pps_body = [0xA1u8, 0xA2, 0xA3]; + let mut parser = HevcParser::new(); + // AU1 seeds the active VPS/SPS/PPS. + let au1 = [ + nal(32, &[0xAA; 12]), + nal(33, &sps_body), + nal(34, &pps_body), + nal(19, &[0x10; 4096]), + ] + .concat(); + parser.parse(&make_pes(au1, Some(0))); + + // A run of BARE keyframes (source omits the parameter sets), each of which + // takes the re-assert path. Payload sized like a real coded picture so a + // reallocation would be the expensive one. + PARAM_REASSERT_REALLOCS.with(|c| c.set(0)); + for i in 0..30i64 { + let au = nal(19, &vec![0x11u8; 300_000]); + let f = parser.parse(&make_pes(au, Some(3600 * (i + 1)))); + // The re-assert really happened (otherwise the count is vacuously 0). + assert!( + f[0].data.len() > 300_000, + "keyframe {i} must carry the re-asserted parameter sets" + ); + } + let reallocs = PARAM_REASSERT_REALLOCS.with(|c| c.get()); + assert_eq!( + reallocs, 0, + "the parameter-set splice must fit in the reserved headroom; \ + {reallocs} of 30 keyframes reallocated the whole frame" + ); + } + /// Regression (Fight Club UHD, the real bug): id 0 is body A (→ hvcC), then /// redefined to B, then the title switches BACK to A. A streaming decoder /// (hvcC at init, in-band updates only) is sitting on B; the switch back to diff --git a/src/mux/codec/truehd.rs b/src/mux/codec/truehd.rs index d07b58e..72eea55 100644 --- a/src/mux/codec/truehd.rs +++ b/src/mux/codec/truehd.rs @@ -17,6 +17,32 @@ use super::dropgate::DropTally; use super::{CodecParser, Frame, PesPacket, pts_to_ns}; use crate::mux::timeline::DISCONTINUITY_BACKSTEP_NS; +/// Is `w` an MLP-family major sync — a random-access / decoder re-init point? +/// +/// The 24-bit signature is 0xF8726F; the following byte is the STREAM TYPE: +/// 0xBA = Dolby TrueHD (the only one Blu-ray carries), 0xBB = MLP. Both are +/// restart points, so the keyframe / re-sync decision accepts either. +fn is_mlp_major_sync(w: u32) -> bool { + (w & 0xFFFF_FFFE) == 0xF872_6FBA +} + +/// Is `w` specifically the TrueHD major sync (stream type 0xBA)? +/// +/// The 32-bit word that FOLLOWS the sync is laid out per stream type: TrueHD's +/// `format_info` carries [31..28] audio_sampling_frequency, the 5-bit 6-channel +/// and 13-bit 8-channel presentation channel-assignment masks; MLP's (0xBB) same +/// word carries quantization word lengths and the MLP group sample-rate fields +/// instead. So every site that DECODES `format_info` with the TrueHD layout must +/// require 0xBA exactly — masking the low sync bit there read a quantization code +/// as the rate nibble and pulled channel masks out of unrelated bits, giving the +/// track header a wrong `SamplingFrequency` and `truehd_au_duration_ns` a wrong +/// per-AU increment (audio drifting against video for the whole track). +/// `None` from these helpers is the safe outcome: the caller falls back to its +/// container-derived rate/channel count. +fn is_truehd_major_sync(w: u32) -> bool { + w == 0xF872_6FBA +} + /// Duration of one TrueHD access unit in nanoseconds for the 48 kHz family /// (48 / 96 / 192 kHz). `access_unit_size = 40 << (ratebits & 7)` and /// `sample_rate = 48000 << (ratebits & 7)`; the shared shift cancels in @@ -130,7 +156,13 @@ impl TrueHdParser { // The rate nibble is only trustworthy once the major sync's CRC has // validated (above), so capture format_info here and refine the PTS // cadence from it ONLY on this validated path. - if au.len() >= 12 { + // ONLY for stream type 0xBA. An MLP (0xBB) major sync's following word + // is not the TrueHD `format_info` layout, so decoding it as one gave a + // wrong rate and channel count; leaving it `None` keeps the + // container-derived rate, which is the honest answer. + if au.len() >= 12 + && is_truehd_major_sync(u32::from_be_bytes([au[4], au[5], au[6], au[7]])) + { format_info = Some(u32::from_be_bytes([au[8], au[9], au[10], au[11]])); } } @@ -445,10 +477,17 @@ impl CodecParser for TrueHdParser { break; // incomplete access unit, wait for more data } + // Restart-point question — either stream type (0xBA TrueHD, 0xBB MLP) + // is a decoder re-init point, so both count as a major sync here. + // DECODING format_info with the TrueHD layout is a separate question, + // gated on 0xBA alone in `au_check`. let is_major_sync = unit_bytes >= 8 - && (u32::from_be_bytes([self.buf[4], self.buf[5], self.buf[6], self.buf[7]]) - & 0xFFFF_FFFE) - == 0xF872_6FBA; + && is_mlp_major_sync(u32::from_be_bytes([ + self.buf[4], + self.buf[5], + self.buf[6], + self.buf[7], + ])); // Decodability gate. MLP/TrueHD decode state persists across access // units, so a corrupt AU is dropped FORWARD to the next VALIDATED @@ -597,7 +636,9 @@ pub fn truehd_channels_from_stream(data: &[u8]) -> Option { let mut p = 0; while p + 8 <= data.len() { let w = u32::from_be_bytes([data[p], data[p + 1], data[p + 2], data[p + 3]]); - if (w & 0xFFFF_FFFE) == 0xF872_6FBA { + // 0xBA only: `truehd_channels` reads the TrueHD `format_info` channel + // masks, which an MLP (0xBB) major sync does not carry. + if is_truehd_major_sync(w) { let fi = u32::from_be_bytes([data[p + 4], data[p + 5], data[p + 6], data[p + 7]]); return truehd_channels(fi); } @@ -663,7 +704,9 @@ pub fn truehd_sync_info_from_stream(data: &[u8]) -> Option { let mut p = 0; while p + 8 <= data.len() { let w = u32::from_be_bytes([data[p], data[p + 1], data[p + 2], data[p + 3]]); - if (w & 0xFFFF_FFFE) == 0xF872_6FBA { + // 0xBA only: `format_info` (and the num_substreams/Atmos nibble) are the + // TrueHD layout, not MLP's. + if is_truehd_major_sync(w) { let format_info = u32::from_be_bytes([data[p + 4], data[p + 5], data[p + 6], data[p + 7]]); // num_substreams is the top nibble of the 17th sync byte (p + 16). @@ -1447,16 +1490,53 @@ mod tests { // --- truehd_channels_from_stream: major-sync variant bit + scan --- #[test] - fn channels_from_stream_matches_variant_sync_0xfb() { - // The sync match masks the low bit: 0xF8726FBA & 0xFFFFFFFE == base, and - // 0xF8726FBB (the +1 variant) matches the same masked pattern. A stream - // carrying 0xF8726FBB must still be recognised. + fn channels_from_stream_rejects_mlp_sync_0xfb() { + // 0xF8726FBB is the MLP stream type, NOT TrueHD (0xF8726FBA). The word + // after an MLP major sync holds quantization word lengths and the MLP + // group sample-rate fields, not TrueHD's rate nibble plus the 6ch/8ch + // presentation channel-assignment masks. Decoding it with the TrueHD + // layout (the scan used to mask the low sync bit) reported channels out of + // unrelated bits; the honest answer is None so the caller keeps its + // container-derived count. Byte pattern below decodes as 8 channels ONLY + // under the TrueHD layout, so this test fails if the mask comes back. let mut data = vec![0x00]; data.extend_from_slice(&0xF872_6FBBu32.to_be_bytes()); data.extend_from_slice(&0x0000_001Fu32.to_be_bytes()); + assert_eq!( + truehd_channels_from_stream(&data), + None, + "an MLP (0xBB) major sync must not be decoded as TrueHD format_info" + ); + // The same bytes under the TrueHD stream type DO decode. + let mut data = vec![0x00]; + data.extend_from_slice(&0xF872_6FBAu32.to_be_bytes()); + data.extend_from_slice(&0x0000_001Fu32.to_be_bytes()); assert_eq!(truehd_channels_from_stream(&data), Some(8)); } + #[test] + fn mlp_sync_0xfb_yields_no_sample_rate_or_atmos() { + // Same split for the shared scan: an MLP major sync must not produce a + // TrueHD rate (bits 31..28 of an MLP header are a quantization code, not + // the rate) nor an Atmos verdict. + let mut data = vec![0x00]; + data.extend_from_slice(&0xF872_6FBBu32.to_be_bytes()); + // ratebits nibble 0x1 would decode as 96 kHz under the TrueHD layout. + data.extend_from_slice(&0x1000_001Fu32.to_be_bytes()); + data.extend_from_slice(&[0x00; 12]); + assert!( + truehd_sync_info_from_stream(&data).is_none(), + "no TrueHD sync info from an MLP major sync" + ); + assert_eq!(truehd_sample_rate_from_stream(&data), None); + // TrueHD stream type, identical trailing bytes → the rate IS decoded. + let mut data = vec![0x00]; + data.extend_from_slice(&0xF872_6FBAu32.to_be_bytes()); + data.extend_from_slice(&0x1000_001Fu32.to_be_bytes()); + data.extend_from_slice(&[0x00; 12]); + assert_eq!(truehd_sample_rate_from_stream(&data), Some(96000)); + } + #[test] fn channels_from_stream_none_without_major_sync() { // No major sync anywhere → None, no panic, scan terminates. @@ -1527,8 +1607,10 @@ mod tests { #[test] fn major_sync_variant_bit_also_keyframe() { - // The keyframe check masks the low bit (0xFFFF_FFFE), so the 0xF8726FBB - // variant must also be detected as a major sync. + // The RESTART-POINT check masks the low sync bit, so 0xF8726FBB (MLP) + // counts as a major sync too — both stream types re-init the decoder, so + // both are keyframes. (Only the format_info DECODE is 0xBA-only; see + // `channels_from_stream_rejects_mlp_sync_0xfb`.) let mut parser = TrueHdParser::new(); let mut unit = make_truehd_unit(200); unit[4..8].copy_from_slice(&0xF872_6FBBu32.to_be_bytes()); diff --git a/src/mux/demux_sink.rs b/src/mux/demux_sink.rs index a68de31..748135d 100644 --- a/src/mux/demux_sink.rs +++ b/src/mux/demux_sink.rs @@ -220,6 +220,13 @@ struct AnnexBWriter { /// avcC/hvcC may declare 1 or 2, and reading those as u32-BE parses no NALs /// at all, so the raw prefixed bytes would be emitted as if already Annex B. length_size: usize, + /// Reused length-prefixed -> Annex-B conversion buffer. `write_frame` used to + /// allocate and free a whole-frame Vec per video frame; extracting the video ES + /// of a UHD title is ~200,000 frames of 150-400 KB, every one over the + /// allocator's mmap threshold, so that was ~200,000 mmap/munmap pairs plus + /// millions of first-touch page faults of pure overhead. Kept on the writer and + /// cleared per frame instead, matching what tsmux.rs already does. + scratch: Vec, } impl AnnexBWriter { @@ -231,6 +238,7 @@ impl AnnexBWriter { params, wrote_params: false, length_size: nal_length_size(codec, codec_private), + scratch: Vec::new(), } } } @@ -251,10 +259,16 @@ impl EsWriter for AnnexBWriter { // source of truth across all muxers — see `crate::mux::hevc`). It skips // zero-length NALs and drops a truncated trailing NAL without panicking, // rather than `break`ing on the first zero-length NAL. - let mut scratch = Vec::with_capacity(f.data.len() + (f.data.len() / 32) + 4); + // Reuse the writer's buffer rather than allocating per frame; clear() + // keeps the capacity, so steady state costs no allocation at all. The + // prefix width still comes from the record, never a hardcoded 4. + self.scratch.clear(); + self.scratch.reserve(f.data.len() + (f.data.len() / 32) + 4); + let mut scratch = std::mem::take(&mut self.scratch); append_length_prefixed_as_annex_b_sized(&mut scratch, &f.data, self.length_size); w.write_all(&scratch)?; n += scratch.len(); + self.scratch = scratch; Ok(n) } } diff --git a/src/mux/driver.rs b/src/mux/driver.rs index a720266..7202bd9 100644 --- a/src/mux/driver.rs +++ b/src/mux/driver.rs @@ -825,9 +825,42 @@ fn drive_mux( } Err(e) => { // Drain + join the consumer so its output file handle is - // released, then propagate the read error. - let _ = pipe.finish_with_halt(Some(halt)); - return Err(e); + // released, then report the ROOT cause. + // + // The consumer's result used to be discarded with `let _`. A + // write-side `WriteSink::apply` failure that had ALREADY killed + // the output — the destination volume filling, say — was thrown + // away, and only the read error surfaced: the caller diagnosed a + // damaged disc and retried the rip onto the same full volume + // instead of being told it was out of space. A hard write + // failure precedes and explains the read error here (the + // producer only reaches the next `read()` because its previous + // `send` did not block on a dead consumer), so prefer it. + // Halt/join-timeout are NOT root causes — those are the clean + // operator-stop and wedge paths the finish stage below + // translates to `completed = false` — so the read error still + // wins over them. + match pipe.finish_with_halt(Some(halt)) { + Err(w @ (Error::Halted | Error::PipelineJoinTimeout)) => { + tracing::debug!( + target: "mux", + write_side = %w, + read_side = %e, + "read failed; consumer stopped for a non-root-cause reason — reporting the read error" + ); + return Err(e); + } + Err(w) => { + tracing::error!( + target: "mux", + write_side = %w, + read_side = %e, + "read failed, but the write side had already failed — reporting the write failure as the root cause" + ); + return Err(w.into()); + } + Ok(_) => return Err(e), + } } } } diff --git a/src/mux/m2ts_mux/mod.rs b/src/mux/m2ts_mux/mod.rs index b4b9593..20c3244 100644 --- a/src/mux/m2ts_mux/mod.rs +++ b/src/mux/m2ts_mux/mod.rs @@ -1151,6 +1151,15 @@ mod tests { }) } + /// Every packet on `pid` (optionally requiring PUSI), in stream order. + fn find_all_pkts(buf: &[u8], pid: u16, pusi: bool) -> Vec<&[u8]> { + buf.chunks(188) + .filter(|p| { + u16::from_be_bytes([p[1] & 0x1F, p[2]]) == pid && (!pusi || (p[1] & 0x40) != 0) + }) + .collect() + } + /// Extract a PSI section (after the pointer_field) from a PUSI PSI /// packet: payload starts at byte 4 (no AF on PSI here), first payload /// byte is pointer_field, section follows. @@ -1355,31 +1364,92 @@ mod tests { #[test] fn extreme_pts_does_not_overflow_and_clamps_to_33bit() { - // base_relative_pts widens to u128 then masks to 33 bits. An - // adversarial i64::MAX ns must not overflow and the encoded PTS must - // stay within the 33-bit field. With a single video frame the base - // is itself, so relative PTS is 0 — proving no panic on the path. + // `base_relative_pts` widens to u128 then wraps the delta into 33 bits. An + // adversarial i64::MAX ns must not overflow, and the encoded PES PTS must be + // the EXACT wrapped value. + // + // The old version of this test wrote a single video frame, which by its own + // admission rebases to relative PTS 0 — so its only assertion was + // `0 < 2^33`, which cannot fail: widening the 33-bit mask (or deleting it) + // left the one test named for 33-bit clamping green. Two frames are needed + // so the second has a non-zero delta to pin, and the expected value is + // computed here from the spec formula rather than read back from the code + // under test. + // + // NOTE on the test's name: bit 32 of the PES PTS field is UNREACHABLE from + // this path. `base_relative_pts` interprets the delta as a signed 33-bit + // value and floors the entire upper half [2^32, 2^33) to 0, so anything the + // encoder ever receives is < 2^32. "Clamping to 33 bits" is therefore a + // structural property of the delta rule, not something a test can exercise; + // what IS pinned below is the exact encoding of the largest reachable value + // and the documented i64::MAX outcome. + // Decode the 33-bit PTS out of a PUSI video packet's PES header. + let decode_pts = |pkt: &[u8]| -> u64 { + // Payload after the AF: AF area = 1 (length byte) + af_len. + let af_len = pkt[4] as usize; + let pes = &pkt[4 + 1 + af_len..]; + // PES: 00 00 01 E0 00 00 80 80 05 PTS[5]. PTS at pes[9..14]. + ((((pes[9] >> 1) & 0x07) as u64) << 30) + | ((pes[10] as u64) << 22) + | (((pes[11] >> 1) as u64) << 15) + | ((pes[12] as u64) << 7) + | ((pes[13] >> 1) as u64) + }; + let mut frame = Vec::new(); + frame.extend_from_slice(&4u32.to_be_bytes()); + frame.extend_from_slice(&[0x40, 0x01, 0x0C, 0x01]); + + // (a) A LARGE but forward delta: exact-value assertion across all 33 bits. + // + // Frame 1 at PTS 0 seeds the base to 0, so frame 2's relative PTS is its own + // tick value. 477_218_477 x 100_000 ns divides exactly by the 100_000/9 + // conversion, giving 4_294_966_293 ticks — just under 2^32, i.e. the largest + // magnitude the signed-33-bit delta rule treats as forward progression, and + // a value that needs the full 3+15+15-bit PES PTS field to survive. let mut sink: Vec = Vec::new(); { let mut mux = M2tsMux::new(&mut sink); - let mut frame = Vec::new(); - frame.extend_from_slice(&4u32.to_be_bytes()); - frame.extend_from_slice(&[0x40, 0x01, 0x0C, 0x01]); + mux.write_video(0, true, &frame).unwrap(); + mux.write_video(477_218_477 * 100_000, true, &frame) + .unwrap(); + mux.finish().unwrap(); + } + assert_ts_well_formed(&sink); + let pkts = find_all_pkts(&sink, PID_VIDEO, true); + assert_eq!(pkts.len(), 2, "one PUSI packet per video frame"); + let pts = decode_pts(pkts[1]); + assert!(pts < (1u64 << 33), "PTS stays within the 33-bit field"); + assert_eq!( + pts, 4_294_966_293, + "the encoded PES PTS must be the exact tick value, not a truncated one" + ); + + // (b) The adversarial i64::MAX: no overflow, no panic, and the documented + // signed-33-bit outcome. Its tick count masks to 6_564_084_417, which is in + // the UPPER half of the 33-bit range, so the delta rule reads it as a frame + // BEFORE the base and floors it to 0 (the same rule that floors leading + // audio). Asserted explicitly so this is a pinned decision, not the vacuous + // `0 < 2^33` the old single-frame version checked. + let mut sink: Vec = Vec::new(); + { + let mut mux = M2tsMux::new(&mut sink); + mux.write_video(0, true, &frame).unwrap(); mux.write_video(i64::MAX, true, &frame).unwrap(); mux.finish().unwrap(); } assert_ts_well_formed(&sink); - let pkt = find_pkt(&sink, PID_VIDEO, true).unwrap(); - // Reach the PES PTS: payload after AF. AF area = 1 (length) + af_len. - let af_len = pkt[4] as usize; - let pes = &pkt[4 + 1 + af_len..]; - // PES: 00 00 01 E0 00 00 80 80 05 PTS[5]. PTS at pes[9..14]. - let pts = ((((pes[9] >> 1) & 0x07) as u64) << 30) - | ((pes[10] as u64) << 22) - | (((pes[11] >> 1) as u64) << 15) - | ((pes[12] as u64) << 7) - | ((pes[13] >> 1) as u64); + let masked = (((i64::MAX as u128) * 9 / 100_000) as u64) & 0x1_FFFF_FFFF; + assert!( + masked >= 1 << 32, + "i64::MAX masks into the upper (negative) half" + ); + let pkts = find_all_pkts(&sink, PID_VIDEO, true); + let pts = decode_pts(pkts[1]); assert!(pts < (1u64 << 33), "PTS stays within the 33-bit field"); + assert_eq!( + pts, 0, + "an i64::MAX tick lands in the signed-33-bit upper half and floors to 0" + ); } #[test] diff --git a/src/mux/mkv.rs b/src/mux/mkv.rs index 9df12f5..f15c843 100644 --- a/src/mux/mkv.rs +++ b/src/mux/mkv.rs @@ -681,10 +681,18 @@ pub struct MkvMuxer { cues: Vec, frame_count: u64, /// Frames handed to `write_frame` that were dropped because no cluster was - /// open yet (a cluster only opens on a track-0 video keyframe). If this is - /// non-zero at `finish()` and not a single frame was ever written, the - /// caller produced an empty MKV — surfaced as an error rather than a - /// silently empty file. See `write_frame` for the track-0 invariant. + /// open yet (a cluster only opens on a track-0 video keyframe). See + /// `write_frame` for the track-0 invariant. + /// + /// The ALL-dropped case is surfaced as an error by `finish()`, but via + /// `frame_count == 0`, not via this counter. A PARTIAL drop — leading audio / + /// subtitle frames ahead of the first video IDR, or an M2TS whose PMT lists + /// audio before video — is normal enough not to fail the mux, but it used to + /// leave NO record anywhere: the field was incremented at two sites and read + /// nowhere (no log, no error, no accessor), so those frames vanished from the + /// output with `completed = true`, an empty `undelivered_streams`, and nothing + /// in the log. `finish()` now logs the count — that log is the field's only + /// reader, so do not delete it and turn this back into dead bookkeeping. dropped_pre_cluster: u64, seek_fixups: Vec, /// Absolute file offset of the CUES SeekHead entry (a fixed 21-byte Seek @@ -1657,6 +1665,18 @@ impl MkvMuxer { if self.frame_count == 0 { return Err(crate::error::Error::MkvInvalid.into()); } + // Partial pre-cluster drops do not fail the mux (leading audio ahead of the + // first video IDR is normal), but they MUST leave a record: this counter + // was write-only, so frames silently vanished from the output while the run + // reported success. See the field's doc. + if self.dropped_pre_cluster > 0 { + tracing::warn!( + target: "mux", + dropped = self.dropped_pre_cluster, + frames_written = self.frame_count, + "frames were discarded before the first cluster opened (no track-0 video keyframe had arrived yet); they are absent from the output" + ); + } // The source declared no duration up-front (DURATION was reserved as a // placeholder). Derive the real runtime from the muxed timeline so the // Segment declares it — and so the BPS tags below can be computed. @@ -2615,6 +2635,47 @@ mod tests { ); } + /// Frames dropped before the first cluster opens must be COUNTED, and the + /// count must survive to `finish()` so it can be reported. The counter was + /// incremented at two sites and read nowhere — no log, no error, no accessor — + /// so a partial drop (leading audio/subtitle frames ahead of the first video + /// IDR, or an M2TS whose PMT lists audio before video) silently omitted those + /// frames from the output while the run reported `completed = true` with an + /// empty `undelivered_streams` and nothing in the log. `finish()` now logs it; + /// this pins the accounting the log depends on. + #[test] + fn frames_dropped_before_first_cluster_are_counted() { + let buf = Cursor::new(Vec::new()); + let tracks = [make_video_track()]; + let mut muxer = MkvMuxer::new(buf, &tracks, None, 60.0, &[]).unwrap(); + // Two non-keyframes arrive before any track-0 keyframe: no cluster can be + // open, so both are dropped. + muxer + .write_frame(0, 0, false, &[0x01; 8], None, None) + .unwrap(); + muxer + .write_frame(0, 1_000_000, false, &[0x02; 8], None, None) + .unwrap(); + assert_eq!( + muxer.dropped_pre_cluster, 2, + "both pre-cluster frames must be counted, not silently lost" + ); + assert_eq!(muxer.frame_count, 0, "neither frame was written"); + // A keyframe then opens the cluster and is written; the drop count stands + // so `finish()` can still report it. + muxer + .write_frame(0, 2_000_000, true, &[0x03; 8], None, None) + .unwrap(); + assert_eq!(muxer.frame_count, 1); + assert_eq!( + muxer.dropped_pre_cluster, 2, + "the drop count must survive to finish(), which reports it" + ); + muxer + .finish() + .expect("a mux with one written frame succeeds"); + } + #[test] fn mkv_finish_writes_cues_element() { // finish() consumes self and flushes the writer, so use the diff --git a/src/mux/ts.rs b/src/mux/ts.rs index 79993a2..4ab238c 100644 --- a/src/mux/ts.rs +++ b/src/mux/ts.rs @@ -53,6 +53,11 @@ pub struct PesPacket { /// Per-PID PES reassembly state. struct PesAssembler { pid: u16, + /// This PID's PES-reassembly ceiling: its share of [`MAX_PES_BUFFER_TOTAL`], + /// clamped to [`MAX_PES_BUFFER`]. Resolved once by `TsDemuxer::new` so the + /// per-PID caps sum to a bounded total no matter how many streams the disc + /// declares. + cap: usize, buffer: Vec, pts: Option, dts: Option, @@ -107,10 +112,34 @@ const PES_BUFFER_INIT_CAP: usize = 16 * 1024; /// the next PUSI. const MAX_PES_BUFFER: usize = 64 * 1024 * 1024; // 64 MiB +/// AGGREGATE ceiling across every tracked PID. +/// +/// [`MAX_PES_BUFFER`] bounds each PID's buffer independently and never sees the +/// total, while the tracked-PID count comes straight off the disc: `TsDemuxer::new` +/// makes one [`PesAssembler`] per SELECTED stream, and the selection derives from +/// the MPLS STN, whose per-category counts are `u8` (up to 255 each across 8 +/// categories) bounded only by the MPLS file's own bytes. A crafted MPLS declaring +/// 100 streams on 100 distinct PIDs, plus a clip feeding each PID continuation +/// packets (no PUSI) until just under the per-PID cap, held 100 x 64 MiB = 6.4 GiB +/// of PES buffers at once; 1000 distinct PIDs — well inside the 8192-entry +/// `pid_index` table — is 64 GiB. +/// +/// So the per-PID cap is derived from this total instead: `pes_cap` (below) is +/// `MAX_PES_BUFFER_TOTAL / tracked_pids`, clamped to `MAX_PES_BUFFER`. A real title +/// selects a handful of streams and keeps the full 64 MiB each; only a stream count +/// far past anything an authored disc carries is squeezed, and even then a complete +/// HEVC/UHD access unit (1-3 MiB) still fits at ~170 PIDs. Overflow is graceful in +/// any case — the partial PES is dropped and the assembler resyncs on the next +/// PUSI, flagging a discontinuity. +const MAX_PES_BUFFER_TOTAL: usize = 512 * 1024 * 1024; // 512 MiB + impl PesAssembler { - fn new(pid: u16) -> Self { + /// `cap` is this PID's SHARE of [`MAX_PES_BUFFER_TOTAL`], resolved by + /// `TsDemuxer::new` from the tracked-PID count. + fn new(pid: u16, cap: usize) -> Self { Self { pid, + cap, buffer: Vec::with_capacity(PES_BUFFER_INIT_CAP), pts: None, dts: None, @@ -155,13 +184,14 @@ impl PesAssembler { /// Append payload data to the current PES packet. /// - /// If the buffer would exceed [`MAX_PES_BUFFER`] the partial PES is + /// If the buffer would exceed this PID's `cap` — its share of + /// [`MAX_PES_BUFFER_TOTAL`], at most [`MAX_PES_BUFFER`] — the partial PES is /// silently dropped and the assembler is reset. Normal traffic resumes /// on the next PUSI; a crafted/corrupt stream that never sends one can - /// no longer drive unbounded allocation. + /// no longer drive unbounded allocation, on this PID OR in aggregate. fn push(&mut self, data: &[u8]) { if self.active { - if self.buffer.len().saturating_add(data.len()) > MAX_PES_BUFFER { + if self.buffer.len().saturating_add(data.len()) > self.cap { tracing::trace!( target: "mux", pid = self.pid, @@ -237,9 +267,13 @@ impl TsDemuxer { let table_size = (max_pid + 1).max(8192); let mut pid_index = vec![-1i32; table_size]; let mut assemblers = Vec::with_capacity(pids.len()); + // Per-PID cap = this PID's share of the AGGREGATE ceiling. Without this the + // caps were per-PID only and never saw the total, so a disc-declared stream + // list could multiply 64 MiB by its own length. + let pes_cap = (MAX_PES_BUFFER_TOTAL / pids.len().max(1)).min(MAX_PES_BUFFER); for (i, &pid) in pids.iter().enumerate() { pid_index[pid as usize] = i as i32; - assemblers.push(PesAssembler::new(pid)); + assemblers.push(PesAssembler::new(pid, pes_cap)); } Self { assemblers, @@ -2229,6 +2263,58 @@ mod tests { // ── PES reassembly buffer cap (DoS hardening) ───────────────────────── + /// The per-PID PES cap must be a SHARE of an aggregate ceiling, not a flat + /// 64 MiB per PID that never sees the total. The tracked-PID count comes off + /// the disc (one assembler per selected stream, selection driven by the MPLS + /// STN whose per-category counts are u8), so a crafted MPLS declaring many + /// streams on distinct PIDs held `count x 64 MiB` of PES buffers at once — + /// 6.4 GiB at 100 PIDs, 64 GiB at 1000 (still inside the 8192-entry pid_index + /// table). With 64 tracked PIDs each share is 512 MiB / 64 = 8 MiB, so a PID + /// flooded with continuation packets must drop its partial PES at ~8 MiB, not + /// at 64 MiB. + #[test] + fn per_pid_pes_cap_is_a_share_of_an_aggregate_ceiling() { + let pids: Vec = (0x1000..0x1040).collect(); // 64 PIDs + assert_eq!(pids.len(), 64); + let mut demux = TsDemuxer::new(&pids); + + let expected_share = MAX_PES_BUFFER_TOTAL / 64; + assert!( + expected_share < MAX_PES_BUFFER, + "the test is only meaningful when the share is below the per-PID cap" + ); + // The caps must sum to the aggregate ceiling, never to 64 x 64 MiB. + let total: usize = demux.assemblers.iter().map(|a| a.cap).sum(); + assert!( + total <= MAX_PES_BUFFER_TOTAL, + "per-PID caps must sum within the aggregate ceiling: {total} > {MAX_PES_BUFFER_TOTAL}" + ); + + // Behavioural: flood ONE PID with continuation packets and confirm the + // partial PES is dropped at its share, not at MAX_PES_BUFFER. + let pid = pids[0]; + let mut pes_start = vec![0x00, 0x00, 0x01, 0xE0, 0x00, 0x00, 0x80, 0x00, 0x00]; + pes_start.extend_from_slice(&[0xAB; 10]); + demux.feed(&es_packet_exact(pid, true, &pes_start)); + let payload = [0xCCu8; 184]; + let cont_pkt = data_packet(pid, false, &payload); + let mut high_water = 0usize; + for _ in 0..(expected_share / 184 + 64) { + demux.feed(&cont_pkt); + let idx = demux.pid_index[pid as usize] as usize; + high_water = high_water.max(demux.assemblers[idx].buffer.len()); + } + assert!( + high_water <= expected_share, + "a flooded PID must be capped at its share ({expected_share}), \ + not at the flat per-PID cap; high water was {high_water}" + ); + assert!( + high_water > expected_share / 2, + "sanity: the flood must actually have filled the share, got {high_water}" + ); + } + #[test] fn pes_buffer_cap_resets_on_overflow_and_recovers_on_next_pusi() { // Feed continuation-only packets that would exceed MAX_PES_BUFFER if diff --git a/src/mux/tsmux.rs b/src/mux/tsmux.rs index 70b2243..eb48983 100644 --- a/src/mux/tsmux.rs +++ b/src/mux/tsmux.rs @@ -265,14 +265,22 @@ impl TsMuxer { // Video PES may be unbounded (length 0); a 0xBD private_stream_1 // PES must carry a bounded length, so split oversized audio/sub - // access units into multiple PES packets. Each emitted PES carries - // the same PTS and starts on its own PUSI packet (only the keyframe - // RAI rides the first packet of the first PES). + // access units into multiple PES packets. + // + // ONLY THE FIRST emitted PES carries the PTS. ISO/IEC 13818-1 §2.4.3.7 + // puts the PTS in the header of the PES packet containing the FIRST byte of + // the access unit; every chunk used to repeat it, so on read-back a demuxer + // (which treats each PUSI as a new access unit) received the second half of + // e.g. an oversized full-screen PGS display set as an independent segment at + // the SAME timestamp, and the display set was emitted as two blocks with + // identical timestamps instead of one. Each PES still necessarily starts on + // its own PUSI packet — that is what delimits a PES — but only the keyframe + // RAI rides the first packet of the first PES. // // The write result is held rather than `?`-propagated so the conversion // buffer goes back into `self` on every path. let res = if is_video || es_data.len() <= MAX_BD_PES_PAYLOAD { - self.write_pes_chain(track, pid, pts_90k, is_video, keyframe, es_data) + self.write_pes_chain(track, pid, Some(pts_90k), is_video, keyframe, es_data) } else { let mut first_pes = true; let mut res = Ok(()); @@ -280,7 +288,7 @@ impl TsMuxer { res = self.write_pes_chain( track, pid, - pts_90k, + first_pes.then_some(pts_90k), is_video, keyframe && first_pes, chunk, @@ -309,7 +317,7 @@ impl TsMuxer { &mut self, track: usize, pid: u16, - pts_90k: u64, + pts_90k: Option, is_video: bool, keyframe: bool, es_data: &[u8], @@ -434,7 +442,14 @@ impl TsMuxer { } /// Build a PES packet header for a BD stream. -fn build_pes_header(pid: u16, pts_90k: u64, data_len: usize) -> Vec { +/// `pts_90k` is `None` for a CONTINUATION PES packet — one carrying the rest of an +/// access unit that was too large for a single bounded-length private_stream_1 PES. +/// ISO/IEC 13818-1 §2.4.3.7 puts the PTS in the header of the PES packet that +/// contains the FIRST byte of the access unit; repeating it on the continuations +/// makes each of them look like a new access unit at the same timestamp, so a +/// demuxer re-reading the stream splits one display set into two blocks with +/// identical timestamps. +fn build_pes_header(pid: u16, pts_90k: Option, data_len: usize) -> Vec { use crate::consts::pes_stream_id; // Determine stream_id from PID range let stream_id: u8 = if is_video_pid(pid) { @@ -443,7 +458,8 @@ fn build_pes_header(pid: u16, pts_90k: u64, data_len: usize) -> Vec { pes_stream_id::PRIVATE_STREAM_1 // audio, PGS subtitle, or default }; - let pes_data_len = data_len + 8; // 3 header bytes + 5 PTS bytes + data + // 3 optional-header bytes + 5 PTS bytes (when present) + data. + let pes_data_len = data_len + if pts_90k.is_some() { 8 } else { 3 }; let mut header = Vec::with_capacity(14); // Start code: 00 00 01 stream_id @@ -465,8 +481,14 @@ fn build_pes_header(pid: u16, pts_90k: u64, data_len: usize) -> Vec { header.push(len as u8); } - // Flags: 10xx xxxx — MPEG-2, PTS present + // Flags: 10xx xxxx — MPEG-2 header.push(0x80); // marker bits + let Some(pts_90k) = pts_90k else { + // Continuation packet: PTS_DTS_flags = 00, no optional fields. + header.push(0x00); + header.push(0); + return header; + }; header.push(0x80); // PTS present // PES header data length @@ -863,10 +885,15 @@ mod tests { let mut out = Vec::new(); for p in packets.iter().filter(|p| p.pid == pid) { if p.pusi { - // Skip the 14-byte PES header (3 startcode + 1 stream_id + - // 2 length + 2 flags + 1 hdr_len + 5 PTS). - assert!(p.payload.len() >= 14, "PUSI payload holds a PES header"); - out.extend_from_slice(&p.payload[14..]); + // Read the PES header's own length rather than assuming one: + // 6 bytes (startcode + stream_id + length) + 3 optional-header + // bytes + PES_header_data_length. A CONTINUATION PES carries no PTS + // (ISO/IEC 13818-1 §2.4.3.7), so its header is 9 bytes, not 14 — + // this helper used to hardcode 14 and so silently depended on every + // split chunk repeating the PTS. + assert!(p.payload.len() >= 9, "PUSI payload holds a PES header"); + let hdr = 9 + p.payload[8] as usize; + out.extend_from_slice(&p.payload[hdr..]); } else { out.extend_from_slice(&p.payload); } @@ -874,6 +901,15 @@ mod tests { out } + /// PTS_DTS_flags of every PUSI PES header on `pid`, in order. + fn pes_pts_flags(packets: &[TsPacket], pid: u16) -> Vec { + packets + .iter() + .filter(|p| p.pid == pid && p.pusi) + .map(|p| (p.payload[7] >> 6) & 0x03) + .collect() + } + /// A non-NAL codec must pass the ES through byte-for-byte: MPEG-2 and /// VC-1 are not NAL-based, so their ES already IS the wire format and /// `length_prefixed_to_annex_b` would mangle it. @@ -1179,6 +1215,44 @@ mod tests { assert_eq!(got, big, "split audio reassembles byte-for-byte"); } + /// ISO/IEC 13818-1 §2.4.3.7: the PTS belongs in the header of the PES packet + /// that contains the FIRST byte of the access unit. An oversized + /// private_stream_1 access unit is split across several PES packets, and every + /// one of them used to carry the SAME PTS (PTS_DTS_flags = 0b10) even though + /// only the first holds the start of the AU. On read-back a demuxer treats each + /// PUSI as a new access unit, so the second half of e.g. a full-screen PGS + /// display set arrived as an independent segment at an identical timestamp and + /// the display set was emitted as TWO blocks with the same timestamp instead of + /// one. Only the first PES may carry a PTS; the continuations must set + /// PTS_DTS_flags = 0b00. + #[test] + fn split_access_unit_carries_pts_only_on_the_first_pes() { + // Three PES worth of ES so there are two continuations to check. + let big: Vec = (0..(2 * MAX_BD_PES_PAYLOAD + 3000)) + .map(|i| (i & 0xFF) as u8) + .collect(); + let mut sink: Vec = Vec::new(); + { + let mut mux = TsMuxer::new(&mut sink, &[AUDIO_PID]); + mux.write_frame(0, 1_000_000_000, false, &big).unwrap(); + mux.finish().unwrap(); + } + let packets = parse_bd_ts(&sink); + let flags = pes_pts_flags(&packets, AUDIO_PID); + assert_eq!(flags.len(), 3, "the AU must split into three PES packets"); + assert_eq!( + flags, + vec![0b10, 0b00, 0b00], + "only the PES containing the first byte of the access unit may carry a PTS" + ); + // And the split is still lossless with the shorter continuation headers. + assert_eq!( + reassemble_es(&packets, AUDIO_PID), + big, + "split audio still reassembles byte-for-byte" + ); + } + /// MEASURED: the Annex-B conversion buffer must be REUSED across video /// frames, not allocated per frame. Both the allocation's address and its /// capacity are unchanged after the second and third same-sized frames — if diff --git a/src/mux/videomap.rs b/src/mux/videomap.rs index 5c8de52..79e5ee5 100644 --- a/src/mux/videomap.rs +++ b/src/mux/videomap.rs @@ -20,7 +20,7 @@ //! different output format would be a DIFFERENT sink reusing this same model, //! not a pluggable encoder here. -use crate::disc::{ColorSpace, DiscTitle, FrameRate, Stream as DiscStream, VideoStream}; +use crate::disc::{ColorSpace, DiscTitle, Stream as DiscStream, VideoStream}; use crate::mux::codec::PictureInfo; use crate::mux::codec::coding::{CodingType, FieldOrder}; use crate::pes::{PesFrame, SourcePos}; @@ -268,13 +268,6 @@ fn display_aspect_ratio(v: &VideoStream, w: u32, h: u32) -> (u32, u32) { } } -/// The title's nominal frame rate as a fraction — the single mapping site reused -/// by the header builder. (Retained as the canonical accessor.) -#[allow(dead_code)] -fn frame_rate_fraction(fr: FrameRate) -> (u32, u32) { - fr.as_fraction() -} - /// One per-picture index record, distilled from a video [`PesFrame`] /// (`docs/FVI_FORMAT.md` §7). /// diff --git a/src/sector/prefetched.rs b/src/sector/prefetched.rs index 9b7accb..dbfc70b 100644 --- a/src/sector/prefetched.rs +++ b/src/sector/prefetched.rs @@ -403,10 +403,32 @@ impl Drop for PrefetchShell { impl Drop for PrefetchedSectorSource { fn drop(&mut self) { - // Dropping the receiver closes the channel, which makes the - // next producer `send` return Err and exits the loop. Joining - // here gives us a deterministic shutdown — no detached thread - // can outlive the source. + // Drop the channel endpoints BEFORE joining the producer. + // + // `rx` and `recycle_tx` are sibling fields, so they are dropped only + // AFTER this `Drop::drop` body returns. Joining first therefore joined + // while both endpoints were still alive: a producer parked in the plain + // blocking `tx.send(Ok(buf))` (no timeout, so the `Halt` is never + // re-polled) never observed a disconnect and never returned, and the + // dropping thread blocked in `join()` forever. Any source dropped before + // its extents were drained — an error path, an operator stop, or a + // consuming crate using the public `new` + direct `read_sectors` — hit a + // permanent two-thread deadlock. `BytePrefetcher::drop` already had this + // shape; see `drop_undrained_source_joins_cleanly`. + // + // The endpoints are moved out via `mem::replace` with already-disconnected + // stand-ins (each stand-in's peer is dropped immediately), which drops the + // real ones here and needs neither `Option` fields nor `unsafe` — and + // leaves `into_channels`'s `ptr::read` moves untouched. + let (dead_tx, dead_rx) = bounded::(0); + drop(dead_tx); + drop(std::mem::replace(&mut self.rx, dead_rx)); + let (dead_send, dead_recv) = bounded::>(0); + drop(dead_recv); + drop(std::mem::replace(&mut self.recycle_tx, dead_send)); + // Now the producer's next `send`/`recv` returns Err and its loop exits; + // joining gives a deterministic shutdown — no detached thread can outlive + // the source. if let Some(h) = self.producer.take() { let _ = h.join(); } @@ -615,6 +637,42 @@ mod tests { } } + /// Regression: dropping a `PrefetchedSectorSource` DIRECTLY — the + /// public `new` + documented direct-read path, and any error/halt + /// exit before the extents are drained — must join the producer + /// cleanly. `Drop::drop` used to `join()` while the struct still + /// held `rx` and `recycle_tx` (sibling fields drop only AFTER + /// `Drop::drop` returns), so a producer parked in the plain blocking + /// `tx.send(Ok(buf))` never saw a disconnect and the dropping thread + /// blocked in `join()` forever — a permanent two-thread deadlock that + /// cancelling the `Halt` could not escape, because that `send` has no + /// timeout and never re-polls the token. + /// + /// 300 sectors at batch=3 is 100 batches against a forward channel of + /// depth `PREFETCH_CHANNEL_DEPTH` (2), so the producer is guaranteed + /// to be blocked in `send` by the time the drop runs. Mirrors + /// `byte_prefetcher::drop_endless_prefetcher_joins_cleanly`, whose + /// `Drop` already had the correct shape. + #[test] + fn drop_undrained_source_joins_cleanly() { + with_watchdog(Duration::from_secs(10), || { + let extents = vec![Extent { + start_lba: 0, + sector_count: 300, + }]; + let halt = Halt::new(); + let pf = PrefetchedSectorSource::new( + PatternSource { capacity: 9999 }, + extents, + 3, + Some(halt.clone()), + ) + .expect("spawn"); + // Drop without draining a single batch — the old Drop deadlocked here. + drop(pf); + }); + } + /// The CRITICAL regression: after `into_channels`, dropping the /// returned forward receiver + recycle sender must let the producer /// observe disconnection and exit, so dropping the `PrefetchShell` @@ -1002,11 +1060,11 @@ mod tests { let pf = PrefetchedSectorSource::new(src, extents, 3, None).expect("spawn"); // 9 + 6 + 3 = 18, independent of inner source capacity. assert_eq!(pf.capacity_sectors(), 18); - // Release the producer without draining: peel the channels - // and drop them so the producer observes disconnection - // (dropping `pf` directly would join while still holding the - // channels → deadlock; the production drain path always uses - // into_channels). + // Release the producer without draining via the production + // zero-copy path: peel the channels and drop them so the + // producer observes disconnection. (A direct `drop(pf)` is + // also safe now — see `drop_undrained_source_joins_cleanly` + // — but the mux always uses into_channels.) let (rx, recycle_tx, shell) = pf.into_channels(); drop(rx); drop(recycle_tx); @@ -1042,9 +1100,9 @@ mod tests { u32::MAX, "summed total must saturate at u32::MAX, not wrap" ); - // Release the producer via into_channels + drop (a direct - // drop of `pf` would join while still holding the channels → - // deadlock against the still-running EndlessZeroSource). + // Release the producer via into_channels + drop, the production + // zero-copy path. (A direct `drop(pf)` also joins cleanly now — + // see `drop_undrained_source_joins_cleanly`.) let (rx, recycle_tx, shell) = pf.into_channels(); drop(rx); drop(recycle_tx); diff --git a/src/udf.rs b/src/udf.rs index 3f63ef0..9418a8a 100644 --- a/src/udf.rs +++ b/src/udf.rs @@ -1315,6 +1315,15 @@ impl BufferedSectorReader<'_> { // Cap to 8192 sectors (16 MiB) so a disc-controlled ad_len cannot // drive a multi-hundred-MiB allocation before any sectors are read. let count = count.min(8192); + // Clamp the range to the u32 LBA space. `start_lba` comes straight off + // the disc (ECMA-167 §14.1 `extent_location` is an unconstrained Uint32), + // so a metadata partition declared near the top of the space made + // `start_lba + offset` below overflow: an 'attempt to add with overflow' + // panic in debug inside the public `Disc::scan`, and in release a wrap to + // a low LBA that filled the sliding cache with a completely different + // region while `cache_start` still claimed the high one. Sectors past + // `u32::MAX` cannot be addressed at all, so dropping them loses nothing. + let count = count.min(u32::MAX - start_lba); let total = count as usize * 2048; self.cache.resize(total, 0); let mut offset = 0u32; @@ -1359,6 +1368,16 @@ impl BufferedSectorReader<'_> { let mut done: u64 = 0; let mut hb = crate::progress::Heartbeat::new("udf_prefetch"); for &(start, count) in ranges { + // Clamp each range to the u32 LBA space before walking it. Ranges come + // from `collect_file_ranges`, whose allocation descriptors are + // unconstrained ECMA-167 §14.14.1 Uint32s, and nothing bounds + // `start + count`; a range within one batch of the top of the space + // made `start + offset` and `start + offset + i` below overflow — + // a debug panic inside the public `Disc::scan`, and in release a wrap + // that seeded the PERMANENT cache with this file's bytes keyed at low + // LBAs, so every later single-sector read of those LBAs (the AVDP/VDS + // re-reads) silently parsed the wrong sector. + let count = count.min(u32::MAX - start); let mut offset = 0u32; while offset < count { hb.tick(done, total); @@ -1409,8 +1428,13 @@ impl SectorSource for BufferedSectorReader<'_> { buf[..2048].copy_from_slice(data); return Ok(2048); } - // Check sliding cache - if lba >= self.cache_start && lba < self.cache_start + self.cache_sectors { + // Check sliding cache. Tested as a DISTANCE from `cache_start`, not + // as `cache_start + cache_sectors`: `cache_start` is a disc-controlled + // LBA (and the batch-read path below sets it verbatim), so the sum + // overflowed for a window near `u32::MAX` — a debug panic inside + // `SectorSource::read_sectors`, and in release a wrap to a small value + // that silently disabled the cache. + if lba >= self.cache_start && lba - self.cache_start < self.cache_sectors { let offset = (lba - self.cache_start) as usize * 2048; buf[..2048].copy_from_slice(&self.cache[offset..offset + 2048]); return Ok(2048); @@ -1487,7 +1511,14 @@ mod tests { } for i in 0..count as u32 { let off = i as usize * 2048; - let s = self.sectors.get(&(lba + i)).copied().unwrap_or([0u8; 2048]); + // `wrapping_add`: the near-`u32::MAX` overflow regressions below + // hand this harness LBAs at the top of the space on purpose, and + // the harness's own bookkeeping must not be what panics. + let s = self + .sectors + .get(&lba.wrapping_add(i)) + .copied() + .unwrap_or([0u8; 2048]); buf[off..off + 2048].copy_from_slice(&s); } Ok(need) @@ -2332,6 +2363,71 @@ mod tests { ); } + /// `prefetch` advances the read LBA with `start_lba + offset`. A UDF whose + /// metadata descriptor declares a partition near the top of the LBA space + /// (ECMA-167 §14.1: `extent_location` is an unconstrained Uint32) drove that + /// add past `u32::MAX` — an 'attempt to add with overflow' panic in debug + /// inside the public `Disc::scan`, and in release a wrap to a low LBA that + /// filled the sliding cache with a completely different region while + /// `cache_start` still claimed the high one. + #[test] + fn prefetch_near_u32_max_does_not_overflow() { + let mut inner = MapReader::new(); + let mut br = BufferedSectorReader::new(&mut inner, 60); + // start + count = 0xFFFF_FFC0 + 200 > u32::MAX: the second batch + // iteration evaluates 0xFFFF_FFC0 + 60. + br.prefetch(0xFFFF_FFC0, 200); + // Nothing read (MapReader serves no sector here), but crucially the + // walk must never form an LBA above u32::MAX. + assert!( + br.cache_start.checked_add(br.cache_sectors).is_some(), + "cache window must stay inside the u32 LBA space: {} + {}", + br.cache_start, + br.cache_sectors + ); + } + + /// `prefetch_ranges` walks each disc-derived `(start_lba, sector_count)` + /// range with `start + offset + i`. `collect_file_ranges` permits any LBA up + /// to `u32::MAX` (ECMA-167 §14.14.1 allocation descriptors are unconstrained + /// Uint32s) and nothing bounds `start + count`, so a range within one batch + /// of the top of the space overflowed: debug panic inside `Disc::scan`, or in + /// release a wrap that seeded the PERMANENT cache with this file's bytes keyed + /// at LBA 0 — every later single-sector read of those low LBAs (the AVDP/VDS + /// re-reads) then returned the wrong sector. + #[test] + fn prefetch_ranges_near_u32_max_does_not_overflow() { + let mut inner = MapReader::new(); + let mut br = BufferedSectorReader::new(&mut inner, 60); + br.prefetch_ranges(&[(0xFFFF_FFF0, 512)]); + // Every key the permanent cache holds must be a real LBA, i.e. inside + // the declared range — never a wrapped low sector. + for &lba in br.prefetched.keys() { + assert!( + lba >= 0xFFFF_FFF0, + "prefetch_ranges wrapped a near-u32::MAX LBA to {lba}" + ); + } + } + + /// The sliding-cache hit test computed `cache_start + cache_sectors`. Once + /// `cache_start` is a disc-controlled LBA near `u32::MAX` (set by the batch + /// read below) that add overflowed: debug panic inside + /// `SectorSource::read_sectors`, release wrap to a small value that silently + /// disabled the cache. + #[test] + fn cache_hit_test_near_u32_max_does_not_overflow() { + let mut inner = MapReader::new(); + let mut br = BufferedSectorReader::new(&mut inner, 60); + let mut buf = [0u8; 2048]; + // First read seeds cache_start = 0xFFFF_FFF5, cache_sectors = 60. + br.read_sectors(0xFFFF_FFF5, 1, &mut buf, true) + .expect("seed read"); + // Second read of the same LBA evaluates the hit test: 0xFFFF_FFF5 + 60. + br.read_sectors(0xFFFF_FFF5, 1, &mut buf, true) + .expect("cache hit test must not overflow"); + } + /// Build a 2048-byte directory sector containing `count` minimal file FIDs. /// /// Each FID uses a 2-byte name (compression-id `8` + `b'A'`), so l_fi=2