From 05729f5dfef45babd5f67d091f8a66ec08436c79 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Thu, 25 Jun 2026 23:39:03 -0700 Subject: [PATCH] =?UTF-8?q?fix(libfreemkv):=20rc6=20hardening=20pass=20?= =?UTF-8?q?=E2=80=94=20mux=20timeline/colour/PCR,=20demux=20panic=20sentin?= =?UTF-8?q?el,=20parser=20robustness=20+=20doc=20accuracy?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Surgical fixes (each with a regression test that fails without the change): mux/mkv.rs, mux/demux_sink.rs: drive the clip-boundary timeline epoch off the resolved PRIMARY VIDEO track, not the literal stream index 0. An M2TS/PMT title can list an audio ES before video, so streams[0] may be audio; a non-video epoch driver ratchets the frontier and inflates the timeline. mkv cluster-opening falls back to track 0 for audio-only titles so they still open clusters. mux/codec/ac3.rs: correct ACMOD_CHANNELS — acmod=5 (3/1) is 4 channels, not 3 (was undercounting a 3/1 stream); fix the A/52 Table 5.8 doc. disc/mod.rs: HDMV coding_type 0x91 (Interactive Graphics / menus) no longer maps to PGS subtitle — it falls through to Unknown so the PMT/STN walker drops it instead of surfacing a bogus subtitle track. mux/videomap.rs + mux/mkv.rs: FVI colour now mirrors the MKV muxer's CICP precedence (measured CICP authoritative; HDR-driven PQ/HLG transfer override) via a shared cicp_for_video helper, so the two sinks can't disagree (HDR10 BT.2020 no longer emits SDR transfer 14). mux/mkvstream.rs: saturating_add on cluster_ts + rel_ts so an adversarial CLUSTER_TIMESTAMP near i64::MAX can't overflow/panic before the existing saturating_mul. mux/timeline.rs: tighten the tail-straggler clamp so a normal new-epoch non-video frame leading the sparse video frontier by >3s is not demoted into the previous clip's epoch. mux/m2ts_mux/mod.rs: re-stamp PCR per video TS packet (mid-PES), not only at PES boundaries, so a large UHD I-frame can't open a multi-second PCR gap; modular 33-bit PTS rebasing so a real 90 kHz clock wrap is not collapsed to PTS 0 (pre-base frames still floor to 0). io/byte_prefetcher.rs, sector/prefetched.rs: wrap the producer feed loop in catch_unwind and emit a typed error sentinel on panic, so a mid-stream producer panic is not read as a clean EOF at the demux boundary (which would silently truncate the mux). mux/codec/h264.rs: extend HIGH_PROFILES to the full ISO/IEC 14496-15 set that mandates the avcC chroma/bit-depth extension (adds 244 et al.). Doc/comment accuracy: css/mod.rs (50000 sectors, not scrambled-sectors), aacs/decrypt.rs (decrypt_unit already-clear path), ifo.rs (TT_SRPT at 0xC4), css/lfsr.rs (LFSR0 24-bit; TAB1-then-XOR cipher; real scramble-flag predicate), disc/read_error.rs (for_sweep does bounded transient retries). Skipped: keydb.rs SSRF guard (low/latent, no live caller) — a hard loopback block breaks an existing behavioral test that exercises the header-EOF path over a loopback server; a clean fix needs a resolver test seam beyond this surgical pass. The sibling keydb_fetch.rs comment fix is out of scope (freemkv crate). --- src/aacs/decrypt.rs | 5 +- src/css/lfsr.rs | 19 +-- src/css/mod.rs | 2 +- src/disc/mod.rs | 21 ++- src/disc/read_error.rs | 8 +- src/ifo.rs | 6 +- src/io/byte_prefetcher.rs | 181 ++++++++++++++++--------- src/mux/codec/ac3.rs | 24 ++-- src/mux/codec/h264.rs | 47 ++++++- src/mux/demux_sink.rs | 62 ++++++++- src/mux/m2ts_mux/mod.rs | 188 +++++++++++++++++++++----- src/mux/mkv.rs | 201 ++++++++++++++++++++-------- src/mux/mkvstream.rs | 23 +++- src/mux/timeline.rs | 59 +++++++- src/mux/videomap.rs | 84 +++++++++++- src/sector/prefetched.rs | 274 ++++++++++++++++++++------------------ 16 files changed, 876 insertions(+), 328 deletions(-) diff --git a/src/aacs/decrypt.rs b/src/aacs/decrypt.rs index 54ace1b..4917207 100644 --- a/src/aacs/decrypt.rs +++ b/src/aacs/decrypt.rs @@ -156,7 +156,10 @@ fn verify_ts(unit: &[u8]) -> bool { } /// Decrypt one AACS aligned unit (6144 bytes) in-place. -/// Returns true if decryption succeeded (verified by TS sync bytes). +/// Returns true if the unit is now clear MPEG-TS: either it was already +/// unscrambled (returned untouched, no key used) or it was decrypted and +/// verified by its TS sync bytes. Returns false only when the unit was +/// scrambled and this key failed verification. /// /// Algorithm: /// 1. AES-128-ECB encrypt first 16 bytes with unit_key → derived diff --git a/src/css/lfsr.rs b/src/css/lfsr.rs index 2ca5fdb..36d6ce6 100644 --- a/src/css/lfsr.rs +++ b/src/css/lfsr.rs @@ -3,10 +3,13 @@ //! The CSS cipher uses two table-driven feedback circuits: //! - LFSR1: 17-bit state (9-bit lo + 8-bit hi register, seeded from //! key[0..2]), driven by TAB2/TAB3 -//! - LFSR0: 32-bit state, driven by a feedback polynomial through TAB4 +//! - LFSR0: 24-bit feedback register (seeded from key[2..5] XOR seed[2..5], +//! masked to 0xFFFFFF), driven by a feedback polynomial through TAB4 //! //! The keystream is the bytewise sum (with carry) of both LFSR outputs. -//! Content descrambling XORs this keystream with the encrypted sector data. +//! Content descrambling computes plain = TAB1[cipher] ^ keystream — a TAB1 +//! substitution of each ciphertext byte followed by an XOR with the keystream +//! (NOT a plain XOR; the cipher is not its own inverse). //! //! Algorithm: Frank A. Stevenson's divide-and-conquer attack (1999). //! Tables: CSS specification constants. @@ -324,15 +327,15 @@ mod tests { // ── scramble-flag detection (byte 0x14, bits 4-5) ────────────────────── /// Only bits 4-5 of byte 0x14 are the CSS scramble flag: the code reads - /// `(sector[0x14] >> 4) & 0x03`. Bit 6 (0x40) and bit 7 (0x80) are NOT - /// part of the flag, so a sector with 0x14 == 0x40 or 0x80 must be treated - /// as UNSCRAMBLED and left byte-for-byte unchanged. This guards against a + /// `sector[0x14] & 0x30 == 0` (bits 6-7, i.e. 0x40/0x80, are masked out by + /// 0x30). A sector with 0x14 == 0x40 or 0x80 must therefore be treated as + /// UNSCRAMBLED and left byte-for-byte unchanged. This guards against a /// too-wide mask silently "descrambling" (and thus corrupting) clear data. /// /// Grounding: CSS sector header byte 0x14 — copyright/scramble bits live - /// in bits 4-5; the 2-bit value 0 means not scrambled. - /// Mutation: change `(sector[0x14] >> 4) & 0x03` to `& 0x07` or drop the - /// shift -> 0x40 would be seen as scrambled and the body would change. + /// in bits 4-5; the masked value 0 means not scrambled. + /// Mutation: widen the mask `0x30` to `0x70`/`0xF0` -> 0x40/0x80 would be + /// seen as scrambled and the body would change. #[test] fn descramble_treats_high_bits_of_0x14_as_clear() { let key = [0x01, 0x02, 0x03, 0x04, 0x05]; diff --git a/src/css/mod.rs b/src/css/mod.rs index 10f5097..d291b23 100644 --- a/src/css/mod.rs +++ b/src/css/mod.rs @@ -51,7 +51,7 @@ pub struct CssState { /// (bytes 0x00..0x80) often ends in a short-period repeating run (stuffing / /// constant fill); the attack assumes that run continues across the 0x80 /// boundary into the encrypted region, giving the known plaintext the 2^16 -/// LFSR recovery needs. We scan up to 50000 scrambled sectors across the +/// LFSR recovery needs. We scan up to 50000 sectors across the /// extents and return the first sector that yields a key — no player keys, no /// disc-key crack. Works on a live drive (after bus-auth unlocks reads) and on /// disc images alike. diff --git a/src/disc/mod.rs b/src/disc/mod.rs index 325d9e5..0d48c86 100644 --- a/src/disc/mod.rs +++ b/src/disc/mod.rs @@ -603,7 +603,12 @@ impl Codec { // lossless pair, parallel to 0x81/0xA1 for AC-3. 0xA2 is // lossless MA, not lossy HR. 0xA2 => Codec::DtsHdMa, - 0x90 | 0x91 => Codec::Pgs, + // 0x90 = Presentation Graphics (PG / subtitles). 0x91 = Interactive + // Graphics (IG / menus) and 0x92 = Text subtitles are distinct HDMV + // coding types and are NOT PG subtitle streams; only 0x90 maps to + // Pgs. IG (0x91) falls through to Unknown so the PMT/STN walker drops + // it rather than surfacing a bogus PGS subtitle track for a menu ES. + 0x90 => Codec::Pgs, ct => Codec::Unknown(ct), } } @@ -5776,6 +5781,20 @@ mod tests { assert_eq!(Codec::from_coding_type(0x86), Codec::DtsHdMa); } + /// HDMV coding_type 0x90 = Presentation Graphics (PG / subtitles) → Pgs, + /// but 0x91 = Interactive Graphics (IG / menus) is NOT a subtitle stream. + /// It must NOT map to Pgs (whose kind() is Subtitle), else a menu ES would + /// surface as a bogus PGS subtitle track. 0x91 falls through to Unknown so + /// the PMT/STN walker drops it. + #[test] + fn coding_type_ig_0x91_is_not_pgs_subtitle() { + assert_eq!(Codec::from_coding_type(0x90), Codec::Pgs); + assert_eq!(Codec::from_coding_type(0x90).kind(), CodecKind::Subtitle); + // IG must not be a PGS subtitle. + assert_eq!(Codec::from_coding_type(0x91), Codec::Unknown(0x91)); + assert_ne!(Codec::from_coding_type(0x91).kind(), CodecKind::Subtitle); + } + /// chapter_name emits a bare 1-based ordinal (no localized prose). #[test] fn chapter_name_is_bare_ordinal() { diff --git a/src/disc/read_error.rs b/src/disc/read_error.rs index aa4967f..53624b5 100644 --- a/src/disc/read_error.rs +++ b/src/disc/read_error.rs @@ -156,8 +156,12 @@ impl ReadCtx { /// outer-batch failure — the user's wedge-prevention principle /// (2026-05-11): once the drive returns ANY recoverable error, /// retrying the same LBA quickly is what triggers the firmware - /// fast-fail transition. Jump immediately, never retry in Pass 1. - /// Pass N owns retries — it gets per-sector timeouts that don't + /// fast-fail transition. On the damage-jump and marginal paths Pass 1 + /// jumps immediately rather than grinding the same LBA. Transient errors + /// (NOT_READY, bridge degradation) are still retried a small bounded + /// number of times (`NOT_READY_MAX_RETRIES` / `BRIDGE_DEGRADATION_MAX_RETRIES`) + /// in both passes before falling through to the skip path. + /// Pass N owns the heavy retries — it gets per-sector timeouts that don't /// hammer the firmware the same way. pub fn for_sweep(batch: u16) -> Self { Self { diff --git a/src/ifo.rs b/src/ifo.rs index 79cfad9..2b8f5d8 100644 --- a/src/ifo.rs +++ b/src/ifo.rs @@ -340,10 +340,8 @@ pub fn parse_vmg(reader: &mut dyn SectorSource, udf: &UdfFs) -> Result return Err(Error::IfoParse); } - // TT_SRPT sector pointer at bytes 0xC4 (offset 196, documented as bytes 62-65 - // in some references, but the canonical IFO spec uses 0xC4). - // NOTE: The user spec says bytes 62-65, which is offset 0x3E. - // Let's use the value from the spec provided. + // The TT_SRPT pointer is at 0xC4 per the DVD-Video VMGI spec. + // (An informal '62-65' / 0x3E note seen elsewhere is wrong — do not use it.) let tt_srpt_sector = be_u32(&vmg_data, 0xC4)?; // Read TT_SRPT — it's at the given sector offset relative to the start of VIDEO_TS.IFO. diff --git a/src/io/byte_prefetcher.rs b/src/io/byte_prefetcher.rs index e86a39b..839932f 100644 --- a/src/io/byte_prefetcher.rs +++ b/src/io/byte_prefetcher.rs @@ -94,76 +94,93 @@ impl BytePrefetcher { let producer = std::thread::Builder::new() .name("freemkv-byte-prefetch".into()) .spawn(move || { - let cancelled = || halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false); - // Liveness heartbeat: the producer blocks on the recycle and - // forward channels; a stalled consumer or a wedged reader shows - // up as the beat going silent. Total is unknown, so `pos` is - // cumulative bytes read. - let mut hb = crate::progress::Heartbeat::new("byte_prefetch"); - let mut produced_bytes: u64 = 0; - loop { - hb.tick(produced_bytes, 0); - if cancelled() { - return; - } - // Park on the recycle channel, but re-poll halt - // every POLL_INTERVAL: a pure-AtomicBool Halt does - // not disconnect the channel, so a blocking recv() - // would never re-reach the cancel check. - let mut buf = loop { - match recycle_rx.recv_timeout(POLL_INTERVAL) { - Ok(b) => break b, - Err(RecvTimeoutError::Timeout) => { - if cancelled() { - return; - } - } - // Consumer dropped both channels. - Err(RecvTimeoutError::Disconnected) => return, - } - }; - // Re-expose the full extent. After a short read the - // prior iteration truncated to n < chunk_bytes, so - // this regrows the length back to chunk_bytes - // without reallocating (capacity was fixed at - // construction and never shrinks). - if buf.len() < chunk_bytes { - buf.resize(chunk_bytes, 0); - } else { - // SAFETY: capacity is at least chunk_bytes - // after construction. - unsafe { buf.set_len(chunk_bytes) }; - } - // Read up to one full chunk. Short reads are - // valid and common — pipe `truncate` so the - // consumer sees only the bytes that arrived. - let n = match reader.read(&mut buf[..]) { - Ok(0) => return, // EOF — drop tx, consumer sees RecvError - Ok(n) => n, - Err(e) => { - let _ = tx.send(Err(e)); + // Wrap the feed loop in catch_unwind so a panic in the inner + // `reader.read` (e.g. a decrypt-on-read slice/arith bug) is NOT + // indistinguishable from a clean finish at the demux boundary. A + // clean exit (EOF, halt, consumer disconnect) returns and drops + // `tx` → the demux loop reads RecvError as EOF (correct). A PANIC + // sends an explicit error sentinel first so the demux loop's + // `Ok(Err(_))` arm fires and propagates a typed error instead of + // converting the dropped channel into a clean `DemuxBatch::Eof` + // that would finalize a TRUNCATED mux while reporting success. + let body = std::panic::AssertUnwindSafe(|| { + let cancelled = || halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false); + // Liveness heartbeat: the producer blocks on the recycle and + // forward channels; a stalled consumer or a wedged reader shows + // up as the beat going silent. Total is unknown, so `pos` is + // cumulative bytes read. + let mut hb = crate::progress::Heartbeat::new("byte_prefetch"); + let mut produced_bytes: u64 = 0; + loop { + hb.tick(produced_bytes, 0); + if cancelled() { return; } - }; - produced_bytes += n as u64; - buf.truncate(n); - // Hand off the filled buffer, re-polling halt on - // each timeout slice so a cancel can interrupt a - // producer parked on a saturated forward channel. - let mut pending = Ok(buf); - loop { - match tx.send_timeout(pending, POLL_INTERVAL) { - Ok(()) => break, - Err(SendTimeoutError::Timeout(returned)) => { - if cancelled() { - return; + // Park on the recycle channel, but re-poll halt + // every POLL_INTERVAL: a pure-AtomicBool Halt does + // not disconnect the channel, so a blocking recv() + // would never re-reach the cancel check. + let mut buf = loop { + match recycle_rx.recv_timeout(POLL_INTERVAL) { + Ok(b) => break b, + Err(RecvTimeoutError::Timeout) => { + if cancelled() { + return; + } } - pending = returned; + // Consumer dropped both channels. + Err(RecvTimeoutError::Disconnected) => return, + } + }; + // Re-expose the full extent. After a short read the + // prior iteration truncated to n < chunk_bytes, so + // this regrows the length back to chunk_bytes + // without reallocating (capacity was fixed at + // construction and never shrinks). + if buf.len() < chunk_bytes { + buf.resize(chunk_bytes, 0); + } else { + // SAFETY: capacity is at least chunk_bytes + // after construction. + unsafe { buf.set_len(chunk_bytes) }; + } + // Read up to one full chunk. Short reads are + // valid and common — pipe `truncate` so the + // consumer sees only the bytes that arrived. + let n = match reader.read(&mut buf[..]) { + Ok(0) => return, // EOF — drop tx, consumer sees RecvError + Ok(n) => n, + Err(e) => { + let _ = tx.send(Err(e)); + return; + } + }; + produced_bytes += n as u64; + buf.truncate(n); + // Hand off the filled buffer, re-polling halt on + // each timeout slice so a cancel can interrupt a + // producer parked on a saturated forward channel. + let mut pending = Ok(buf); + loop { + match tx.send_timeout(pending, POLL_INTERVAL) { + Ok(()) => break, + Err(SendTimeoutError::Timeout(returned)) => { + if cancelled() { + return; + } + pending = returned; + } + // Consumer dropped. + Err(SendTimeoutError::Disconnected(_)) => return, } - // Consumer dropped. - Err(SendTimeoutError::Disconnected(_)) => return, } } + }); + if std::panic::catch_unwind(body).is_err() { + // Producer panicked mid-stream — surface a typed terminal + // error so the demux thread does NOT read the dropped channel + // as a clean EOF and truncate output. + let _ = tx.send(Err(crate::error::Error::DemuxThreadPanicked.into())); } })?; @@ -422,6 +439,42 @@ mod tests { }); } + /// PANIC propagation: a reader that PANICS mid-stream must NOT be read as a + /// clean EOF at the demux boundary. The producer's catch_unwind sends an + /// explicit `Err` sentinel before the thread unwinds, so the consumer sees + /// the good bytes followed by an error batch — never a silent truncation. + /// Without the catch_unwind the panic would just drop `tx`, the consumer + /// would see RecvError (== clean EOF) and the partial output would be + /// finalized as if complete. + #[test] + fn read_panic_surfaces_as_err_batch_not_clean_eof() { + within(10, || { + struct OneThenPanic { + served: bool, + } + impl Read for OneThenPanic { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + if !self.served { + self.served = true; + let n = buf.len().min(8); + buf[..n].fill(0x22); + Ok(n) + } else { + panic!("synthetic mid-stream reader panic"); + } + } + } + let pf = BytePrefetcher::new(OneThenPanic { served: false }, 8, None).expect("spawn"); + let (got, err) = drain_to_vec(pf); + assert_eq!(got, vec![0x22; 8], "good chunk lost before the panic"); + assert!( + err.is_some(), + "a mid-stream producer PANIC must surface as an Err batch, \ + not a clean EOF (which would silently truncate the mux)" + ); + }); + } + /// Recycle-buffer reuse must NOT leak stale bytes between chunks of /// different lengths. After a full chunk, a short read reuses the /// same recycled buffer; lines 123-129 regrow it to chunk_bytes diff --git a/src/mux/codec/ac3.rs b/src/mux/codec/ac3.rs index 4935bd8..6ae1b35 100644 --- a/src/mux/codec/ac3.rs +++ b/src/mux/codec/ac3.rs @@ -269,12 +269,12 @@ fn frame_duration_ns(data: &[u8], bsid: u8) -> u64 { /// Index is the 3-bit acmod value; add 1 when `lfeon` is set. /// /// ```text -/// 0 = 1+1 (Ch1, Ch2) -> 2 4 = 3/0 (L,C,R) -> 3 -/// 1 = 1/0 (C, mono) -> 1 5 = 2/1 (L,R,S) -> 3 -/// 2 = 2/0 (L, R) -> 2 6 = 3/1 (L,C,R,S) -> 4 +/// 0 = 1+1 (Ch1, Ch2) -> 2 4 = 2/1 (L,R,S) -> 3 +/// 1 = 1/0 (C, mono) -> 1 5 = 3/1 (L,C,R,S) -> 4 +/// 2 = 2/0 (L, R) -> 2 6 = 2/2 (L,R,SL,SR) -> 4 /// 3 = 3/0 (L,C,R) -> 3 7 = 3/2 (L,C,R,SL,SR) -> 5 /// ``` -const ACMOD_CHANNELS: [u8; 8] = [2, 1, 2, 3, 3, 3, 4, 5]; +const ACMOD_CHANNELS: [u8; 8] = [2, 1, 2, 3, 3, 4, 4, 5]; /// Decode the channel count of an (E-)AC-3 frame from its bitstream `acmod` and /// `lfeon`, starting at the 0x0B77 syncword. Returns `None` when the frame is @@ -1129,12 +1129,18 @@ mod tests { #[test] fn acmod_channels_3_0_and_2_1() { - // acmod=4 (3/0 L,C,R) → 3 (exercises cmixlev present, surmixlev absent). + // Per A/52 Table 5.8: acmod 4 = 2/1, 5 = 3/1, 6 = 2/2. + // acmod=4 (2/1 L,R,S) → 3 (surmixlev present, no centre → no cmixlev). assert_eq!(acmod_channels(&make_bsi(4, false)), Some(3)); - // acmod=5 (2/1 L,R,S) → 3 (surmixlev present, no centre). - assert_eq!(acmod_channels(&make_bsi(5, false)), Some(3)); - // acmod=6 (3/1) + LFE → 5; lfeon position shifts after both - // cmixlev (centre) and surmixlev (surround) 2-bit fields. + // acmod=5 (3/1 L,C,R,S) → 4 (centre → cmixlev present, surround → + // surmixlev present). This is the regression case: index 5 was wrongly + // 3 in ACMOD_CHANNELS, undercounting a 3/1 stream by one channel. + assert_eq!(acmod_channels(&make_bsi(5, false)), Some(4)); + // acmod=5 (3/1) + LFE → 5; lfeon position shifts after both cmixlev + // (centre) and surmixlev (surround) 2-bit fields. + assert_eq!(acmod_channels(&make_bsi(5, true)), Some(5)); + // acmod=6 (2/2 L,R,SL,SR) → 4 (surmixlev present, no centre); +LFE → 5. + assert_eq!(acmod_channels(&make_bsi(6, false)), Some(4)); assert_eq!(acmod_channels(&make_bsi(6, true)), Some(5)); } diff --git a/src/mux/codec/h264.rs b/src/mux/codec/h264.rs index 5b5bed2..8f7bb00 100644 --- a/src/mux/codec/h264.rs +++ b/src/mux/codec/h264.rs @@ -295,13 +295,18 @@ impl CodecParser for H264Parser { record.push(pps.len() as u8); record.extend_from_slice(pps); - // ISO 14496-15 §5.3.3.1.2: for High-Profile and related profiles - // (profile_idc 100, 110, 122, 144) the record has 4 trailing extension - // bytes carrying chroma_format_idc and bit depths. Older parsers expect - // the record to END after the PPS for Baseline/Main/Extended — do NOT - // append for those (strict parsers reject the extra bytes). + // ISO 14496-15 §5.3.3.1.2: for High-Profile and the related + // chroma/bit-depth-extended profiles the record has 4 trailing extension + // bytes carrying chroma_format_idc and the luma/chroma bit depths. The + // full set that mandates the extension is profile_idc ∈ {100, 110, 122, + // 144, 244 (High 4:4:4 Predictive), 44, 83, 86, 118, 128, 138, 139, 134, + // 135}. Older parsers expect the record to END after the PPS for + // Baseline/Main/Extended — do NOT append for those (strict parsers + // reject the extra bytes). let profile_idc = sps[1]; - const HIGH_PROFILES: [u8; 4] = [100, 110, 122, 144]; + const HIGH_PROFILES: [u8; 14] = [ + 100, 110, 122, 144, 244, 44, 83, 86, 118, 128, 138, 139, 134, 135, + ]; if HIGH_PROFILES.contains(&profile_idc) { if let Some((chroma_fmt, depth_luma, depth_chroma)) = parse_sps_high_profile_ext(sps) { // byte 0: 111111xx — reserved(6) + chroma_format_idc(2) @@ -1464,6 +1469,36 @@ mod tests { ); } + /// ISO 14496-15 §5.3.3.1.2 regression: profile_idc=244 (High 4:4:4 + /// Predictive) ALSO mandates the chroma/bit-depth extension. It was missing + /// from HIGH_PROFILES, so a 244 stream took the Baseline/Main path and + /// emitted an avcC with NO extension bytes — non-conforming, and strict + /// parsers then assume 8-bit 4:2:0. The extension must be appended. + #[test] + fn avcc_profile_244_appends_extension_bytes() { + // profile_idc=244, chroma_format_idc=3 (4:4:4), depths both 4 (12-bit). + let sps = build_high_profile_sps(244, 3, 4, 4); + let mut parser = H264Parser::new(); + feed_sps_pps(&mut parser, &sps); + + let cp = parser.codec_private().expect("avcC must be present"); + let ext_off = sps.len() + 14; + assert_eq!( + cp.len(), + ext_off + 4, + "profile 244 avcC must have the 4 extension bytes (len={}, expected {})", + cp.len(), + ext_off + 4 + ); + assert_eq!(cp[ext_off] & 0x03, 3, "chroma_format_idc must be 3 (4:4:4)"); + assert_eq!(cp[ext_off + 1] & 0x07, 4, "bit_depth_luma_minus8 must be 4"); + assert_eq!( + cp[ext_off + 2] & 0x07, + 4, + "bit_depth_chroma_minus8 must be 4" + ); + } + /// ISO 14496-15 §5.3.3.1.2 regression: a Main-Profile SPS (profile_idc=77) /// must NOT have the extension bytes — strict parsers reject trailing bytes /// for Baseline/Main/Extended profiles. diff --git a/src/mux/demux_sink.rs b/src/mux/demux_sink.rs index 1c03638..91d28cf 100644 --- a/src/mux/demux_sink.rs +++ b/src/mux/demux_sink.rs @@ -781,9 +781,15 @@ impl Stream for DemuxSink { } fn write(&mut self, frame: &PesFrame) -> io::Result<()> { - // Track 0 (primary video) drives epoch decisions; every other track is a - // passive rider on the same global offset — see `TimelineContinuity`. - let pts = self.timeline.adjust(frame.pts, frame.track == 0); + // The PRIMARY VIDEO track (`ref_video_track`, the first DiscStream::Video) + // drives epoch decisions; every other track is a passive rider on the same + // global offset — see `TimelineContinuity`. Drive epochs off the same + // dynamically-resolved video reference used for the audio-delay + // computation, NOT the literal stream index 0 — an M2TS/PMT title can list + // an audio ES before the video ES, in which case track 0 is audio and a + // non-video epoch driver would ratchet the frontier on sparse/lagging PTS. + let drives = Some(frame.track) == self.ref_video_track; + let pts = self.timeline.adjust(frame.pts, drives); if let Some(Some(t)) = self.tracks.get_mut(frame.track) { t.first_pts_ns.get_or_insert(pts); t.writer.write_frame(&mut t.w, frame, pts)?; @@ -1172,6 +1178,56 @@ mod tests { assert_eq!(a, out); } + /// Regression for the hardcoded `frame.track == 0` epoch driver: on an + /// M2TS/PMT title the PMT can list an AUDIO ES before the VIDEO ES, so the + /// video lands at stream index 1. The sink already resolves the video + /// reference dynamically (`ref_video_track`); the epoch driver must use that + /// SAME reference, not the literal 0. Here track 0 is audio and track 1 is + /// video. A video-only clip boundary (track 1) must open a new epoch (bump + /// `offset_ns`). With the bug (only `frame.track == 0` drives epochs) the + /// video back-jump would be treated as a passive rider and `offset_ns` would + /// stay 0 — corrupting every track's rebased timeline. + #[test] + fn epoch_driver_follows_ref_video_not_track_zero() { + let dir = tempdir(); + // Audio FIRST (index 0), video SECOND (index 1). + let title = title_with( + vec![audio_stream(Codec::Ac3, "eng"), video_stream(Codec::H264)], + vec![None, None], + ); + let mut sink = DemuxSink::create(&dir, &title, &DemuxOptions::default()).unwrap(); + assert_eq!( + sink.ref_video_track, + Some(1), + "video reference must be the first VIDEO stream (index 1), not 0" + ); + + let vid = |pts: i64, data: u8| PesFrame { + coding: None, + source: None, + track: 1, // VIDEO is track 1 here + pts, + keyframe: true, + data: vec![0x00, 0x00, 0x00, 0x01, data], + duration_ns: None, + }; + // Clip 1 video: 0s then 10s — advances the frontier. + sink.write(&vid(0, 0xAA)).unwrap(); + sink.write(&vid(10_000_000_000, 0xBB)).unwrap(); + // Clip 2 seam: video PTS jumps back to ~0 (> 3s back) → NEW epoch. The + // video track (index 1) must drive this, bumping the offset. + sink.write(&vid(0, 0xCC)).unwrap(); + + assert!( + sink.timeline.offset_ns >= 10_000_000_000, + "video (track 1) must drive the epoch: offset_ns should have advanced \ + past the previous high, got {}", + sink.timeline.offset_ns + ); + sink.finish().unwrap(); + let _ = std::fs::remove_dir_all(&dir); + } + // ── End-to-end sink ────────────────────────────────────────────────────── #[test] diff --git a/src/mux/m2ts_mux/mod.rs b/src/mux/m2ts_mux/mod.rs index c4ae4f8..447c745 100644 --- a/src/mux/m2ts_mux/mod.rs +++ b/src/mux/m2ts_mux/mod.rs @@ -239,7 +239,19 @@ impl M2tsMux { self.base_pts_90k.get_or_insert(raw_90k); } let base = self.base_pts_90k.unwrap_or(raw_90k); - raw_90k.saturating_sub(base) + // Modular 33-bit subtraction. The 90 kHz PTS clock is a 33-bit field + // that wraps every 2^33 ticks (~26.5 h). A plain `saturating_sub` would + // collapse ANY frame whose (33-bit-masked) tick lands below `base` to + // PTS 0 — including a frame across a legitimate clock wrap (raw wraps + // past 0 and lands far below base), flat-lining timing for that span. + // Wrap the difference into the 33-bit range, then interpret it as a + // signed 33-bit delta: a small magnitude in the LOWER half is genuine + // forward progression (incl. across a wrap) and is kept; a value in the + // UPPER half means the frame is truly BEFORE the base (a small backward + // step — e.g. a leading audio frame ahead of the first video keyframe), + // which still floors to 0 per the documented behavior. + let delta = raw_90k.wrapping_sub(base) & 0x1_FFFF_FFFF; + if delta > (1 << 32) { 0 } else { delta } } /// Emit one PES payload as a chain of TS packets on `pid`. If `pcr` @@ -273,11 +285,16 @@ impl M2tsMux { while offset < pes.len() { self.maybe_emit_psi()?; - // Force a PCR on the FIRST video PES (PAT+PMT precede it, so - // `packets_written` is never 0 here) so a receiver tuning at - // stream start has the clock reference the PMT promises. - let attach_pcr = first - && (pid == PID_VIDEO) + // PCR cadence is enforced per VIDEO TS packet, NOT per PES. A single + // UHD HEVC I-frame is one PES spanning thousands of TS packets; if + // PCR could only ride the PES's first packet, the clock would go + // un-restamped for the whole frame — a multi-second gap far beyond + // the 40-packet / ~100 ms bound, which strict T-STD validators treat + // as a clock discontinuity. So re-stamp whenever the per-packet + // counter reaches the interval (or on the very first video packet of + // the stream), regardless of whether this is the PES start. Re-using + // the PES's own `pcr` for a mid-PES packet keeps the gap bounded. + let attach_pcr = (pid == PID_VIDEO) && (pcr.is_some()) && (!self.first_video_written || self.video_packets_since_pcr >= PCR_INTERVAL_PACKETS); @@ -841,20 +858,23 @@ mod tests { mux.finish().unwrap(); drop(mux); - // The first PUSI video packet carries PCR + RAI (keyframe). - // A later video PUSI packet with AF + PCR but NOT keyframe must - // have RAI clear. - let video_pusi: Vec<&[u8]> = sink + // The first video packet carries PCR + RAI (keyframe PES start). RAI + // rides only the FIRST packet of a KEYFRAME PES; every OTHER PCR-bearing + // video packet — the mid-PES re-stamps and the non-keyframe PES starts — + // must have RAI clear. Skip the very first video packet (the keyframe + // RAI carrier) and assert the first remaining PCR-bearing packet is RAI + // clear. + let video_pkts: Vec<&[u8]> = sink .chunks(188) - .filter(|p| u16::from_be_bytes([p[1] & 0x1F, p[2]]) == PID_VIDEO && (p[1] & 0x40) != 0) + .filter(|p| u16::from_be_bytes([p[1] & 0x1F, p[2]]) == PID_VIDEO) .collect(); assert!( - video_pusi.len() >= 2, - "expected ≥2 video PES starts, got {}", - video_pusi.len() + video_pkts.len() >= 2, + "expected ≥2 video packets, got {}", + video_pkts.len() ); // Find a later one with AF that carries PCR (flags & 0x10 set). - let later_pcr = video_pusi + let later_pcr = video_pkts .iter() .skip(1) .find_map(|p| { @@ -865,49 +885,115 @@ mod tests { None } }) - .expect("later PCR-bearing PUSI exists"); + .expect("a later PCR-bearing video packet exists"); assert_eq!( later_pcr[0] & 0x40, 0, - "RAI must be clear on non-keyframe PCR packet" + "RAI must be clear on a non-keyframe-start PCR packet" + ); + } + + /// Regression: PCR must be re-stamped MID-PES, not only at PES boundaries. + /// A single large video frame (one PES) spans far more than + /// PCR_INTERVAL_PACKETS TS packets — a UHD I-frame. PCR-bearing video + /// packets must recur at least every PCR_INTERVAL_PACKETS video packets + /// across that one PES; before the fix only the PES's first packet carried + /// PCR, leaving a multi-second clock gap for the whole frame. + #[test] + fn pcr_restamped_mid_pes_within_interval() { + let mut sink: Vec = Vec::new(); + let mut mux = M2tsMux::new(&mut sink); + // ONE big frame → ONE PES spanning ~330 packets (≫ 40). + let big: Vec = (0..(60 * 1024)).map(|i| (i & 0xff) as u8).collect(); + let mut frame = Vec::new(); + frame.extend_from_slice(&(big.len() as u32).to_be_bytes()); + frame.extend_from_slice(&big); + mux.write_video(0, true, &frame).unwrap(); + mux.finish().unwrap(); + drop(mux); + + assert_ts_well_formed(&sink); + + // Walk every video TS packet in order; record which ones carry a PCR + // (AF present with PCR_flag 0x10). The packet INDEX (among video + // packets) of consecutive PCR carriers must never advance by more than + // PCR_INTERVAL_PACKETS. + let mut video_idx = 0usize; + let mut pcr_indices: Vec = Vec::new(); + let mut total_video = 0usize; + for pkt in sink.chunks(188) { + let pid = u16::from_be_bytes([pkt[1] & 0x1F, pkt[2]]); + if pid != PID_VIDEO { + continue; + } + total_video += 1; + if let Some(af) = af_body(pkt) { + if !af.is_empty() && (af[0] & 0x10) != 0 { + pcr_indices.push(video_idx); + } + } + video_idx += 1; + } + assert!( + total_video > PCR_INTERVAL_PACKETS as usize, + "test needs a PES spanning more than one PCR interval, got {total_video} video packets" + ); + // More than one PCR across the single PES (the whole point of the fix). + assert!( + pcr_indices.len() >= 2, + "PCR must be re-stamped mid-PES, but only {} PCR-bearing packet(s) \ + appeared across {} video packets of one PES", + pcr_indices.len(), + total_video + ); + // First PCR is on the first video packet. + assert_eq!(pcr_indices[0], 0, "first video packet must carry PCR"); + // No gap between consecutive PCRs exceeds the interval. The counter is + // post-incremented and PCR attaches on `>= PCR_INTERVAL_PACKETS`, so the + // packet index gap is `PCR_INTERVAL_PACKETS + 1` (40 packets carrying no + // PCR, then the re-stamp packet) — the spec "every 40 packets" bound. + let max_gap = PCR_INTERVAL_PACKETS + 1; + for w in pcr_indices.windows(2) { + assert!( + (w[1] - w[0]) as u64 <= max_gap, + "PCR gap {} exceeds the {}-packet bound", + w[1] - w[0], + max_gap + ); + } + let tail = total_video - 1 - *pcr_indices.last().unwrap(); + assert!( + tail as u64 <= max_gap, + "trailing run after the last PCR ({tail}) exceeds the {max_gap}-packet bound" ); } #[test] fn keyframe_video_with_pcr_combines_flags() { - // The first video PES carries a PCR (and RAI) and resets the PCR - // counter. After that, PCR re-attaches only when - // video_packets_since_pcr >= PCR_INTERVAL_PACKETS (40). We push: - // keyframe (PCR+RAI, counter reset) → many non-key (drives the - // counter past the interval) → second keyframe whose PUSI combines - // RAI (keyframe) and PCR (counter exceeded). + // The FIRST video PES is always a keyframe carrying BOTH a PCR (forced + // at stream start so the receiver has the clock the PMT promises) AND a + // RAI (keyframe) — exercising the flag-OR path that combines RAI into the + // PCR adaptation-field flags byte (0x10 | 0x40 = 0x50). (PCR cadence is + // now enforced per video packet, NOT per PES boundary, so a LATER + // keyframe PES start no longer deterministically lands on a PCR-due + // packet; the combine path is pinned here on the guaranteed first PES.) let mut sink: Vec = Vec::new(); let mut mux = M2tsMux::new(&mut sink); let mut small = Vec::new(); small.extend_from_slice(&4u32.to_be_bytes()); small.extend_from_slice(&[0x40, 0x01, 0x0C, 0x01]); mux.write_video(0, true, &small).unwrap(); - // ~50 KB ≈ 270 packets — well over PCR_INTERVAL_PACKETS. - let big: Vec = (0..(50 * 1024)).map(|i| (i & 0xff) as u8).collect(); - let mut big_frame = Vec::new(); - big_frame.extend_from_slice(&(big.len() as u32).to_be_bytes()); - big_frame.extend_from_slice(&big); - mux.write_video(40_000_000, false, &big_frame).unwrap(); - // Now a second keyframe — must combine RAI (keyframe) and PCR - // (counter exceeded). - mux.write_video(80_000_000, true, &small).unwrap(); mux.finish().unwrap(); drop(mux); - // Collect video PUSI packets and find the third (second keyframe). let video_pusi: Vec<&[u8]> = sink .chunks(188) .filter(|p| u16::from_be_bytes([p[1] & 0x1F, p[2]]) == PID_VIDEO && (p[1] & 0x40) != 0) .collect(); - assert!(video_pusi.len() >= 3, "three video PES starts expected"); - let af = af_body(video_pusi[2]).expect("AF present"); + assert!(!video_pusi.is_empty(), "a video PES start exists"); + let af = af_body(video_pusi[0]).expect("AF present on first keyframe PES"); assert!(!af.is_empty(), "AF flags byte present"); - assert_eq!(af[0], 0x50, "flags == RAI | PCR"); + assert_eq!(af[0], 0x50, "flags == RAI | PCR on the first keyframe PES"); } // ════════════════════════════════════════════════════════════════════ @@ -1152,6 +1238,36 @@ mod tests { assert!(pts < (1u64 << 33), "PTS stays within the 33-bit field"); } + #[test] + fn base_relative_pts_wraps_across_33bit_clock_rollover() { + // Regression: a real 90 kHz clock wrap must NOT collapse to PTS 0. + // Seed base near the top of the 33-bit range; a later frame whose tick + // has wrapped past 0 lands far below base. The OLD `saturating_sub` + // returned 0 (flat-lining timing); modular subtraction must return the + // true small forward delta. + let mut sink: Vec = Vec::new(); + let mut mux = M2tsMux::new(&mut sink); + // Force the base to 2^33 - 100 directly (a value reachable only after + // ~26.5 h of stream; set it rather than ripping that long). + mux.base_pts_90k = Some((1u64 << 33) - 100); + // pts_ns = 1ms → raw_90k = 1_000_000 * 9 / 100_000 = 90 ticks (wrapped + // past 0, far below the near-max base). + let pts_ns = 1_000_000i64; + let rel = mux.base_relative_pts(pts_ns, /* may_seed_base */ false); + // 90 - (2^33 - 100) mod 2^33 = 190 ticks forward across the wrap (NOT 0). + assert_eq!( + rel, 190, + "a 33-bit clock wrap must produce the true forward delta, not 0" + ); + + // And a frame genuinely a little BEFORE the base still floors to 0 + // (documented pre-base behavior — e.g. leading audio). base = 200 ticks, + // frame at 90 ticks (< base) → backward step → floor to 0. + mux.base_pts_90k = Some(200); + let rel0 = mux.base_relative_pts(1_000_000i64, false); // raw_90k = 90 < 200 + assert_eq!(rel0, 0, "a frame before the base must still floor to 0"); + } + #[test] fn negative_pts_ns_encodes_zero() { // base_relative_pts treats pts_ns <= 0 as raw 0. A negative input diff --git a/src/mux/mkv.rs b/src/mux/mkv.rs index b007e5f..1c409df 100644 --- a/src/mux/mkv.rs +++ b/src/mux/mkv.rs @@ -60,6 +60,63 @@ const COLOUR_RANGE_LIMITED: u8 = 1; /// Vision configuration record (RFC 9559 + Dolby Vision-in-Matroska spec). const BLOCK_ADD_ID_TYPE_DVCC: u64 = 0x6476_6343; +/// Resolve a video stream's CICP colour code points — `(matrix, transfer, +/// primaries, range)`, ITU-T H.273 — using a single precedence so EVERY sink +/// (the MKV muxer here AND the FVI sidecar in `videomap.rs`) agrees and can +/// never drift: +/// +/// 1. **Measured CICP** read from the bitstream (HEVC/H.264 VUI +/// `colour_description` or MPEG-2 `sequence_display_extension`) is +/// AUTHORITATIVE — copied through verbatim when present. +/// 2. Otherwise fall back to the coarse `color_space` enum (a playlist nibble / +/// PAL-NTSC guess), THEN apply the HDR-driven transfer override: BT.2020 only +/// appears on HDR UHD, where the real transfer is PQ (16) for +/// HDR10/HDR10+/DV or HLG (18) for HLG — never the SDR transfer 14 the enum +/// alone would emit. +pub(crate) fn cicp_for_video(v: &VideoStream) -> (u8, u8, u8, u8) { + if let Some(c) = v.measured_cicp { + return (c.matrix, c.transfer, c.primaries, c.range); + } + let (m, t, p, r) = match v.color_space { + ColorSpace::Bt2020 => ( + CICP_MATRIX_BT2020NC, + CICP_TRANSFER_PQ, + CICP_PRIMARIES_BT2020, + COLOUR_RANGE_LIMITED, + ), + ColorSpace::Bt709 => ( + CICP_MATRIX_BT709, + CICP_TRANSFER_BT709, + CICP_PRIMARIES_BT709, + COLOUR_RANGE_LIMITED, + ), + // PAL SD: BT.470 System B/G matrix/transfer/primaries. + ColorSpace::Bt470bg => ( + CICP_MATRIX_BT470BG, + CICP_TRANSFER_BT470BG, + CICP_PRIMARIES_BT470BG, + COLOUR_RANGE_LIMITED, + ), + // NTSC SD: SMPTE 170M / BT.601-525. + ColorSpace::Smpte170m => ( + CICP_MATRIX_BT601_525, + CICP_TRANSFER_BT601_525, + CICP_PRIMARIES_BT601_525, + COLOUR_RANGE_LIMITED, + ), + ColorSpace::Unknown => (0, 0, 0, 0), + }; + // Override the transfer for HDR signalled by the HdrFormat (the coarse enum + // can't express PQ/HLG). Only applies on the enum fallback; a measured CICP + // already carries the real transfer and returned above. + let t = match v.hdr { + HdrFormat::Hdr10 | HdrFormat::Hdr10Plus | HdrFormat::DolbyVision => CICP_TRANSFER_PQ, + HdrFormat::Hlg => CICP_TRANSFER_HLG, + _ => t, + }; + (m, t, p, r) +} + /// MKV track definition (built from disc stream metadata). pub struct MkvTrack { pub track_type: u64, // 1=video, 2=audio, 17=subtitle @@ -225,58 +282,9 @@ impl MkvTrack { 0 }; // CICP (matrix, transfer, primaries, range) — ITU-T H.273 code points. - // - // Prefer MEASURED CICP read from the bitstream (HEVC/H.264 VUI - // colour_description or MPEG-2 sequence_display_extension) when the - // stream states it: those are authoritative. Fall back to the coarse - // `color_space` enum (a playlist nibble / PAL-NTSC guess) only when no - // measured triplet is present, so the container stops ASSUMING a colour - // space the bitstream may contradict. - let (matrix, transfer, primaries, range) = match v.measured_cicp { - Some(c) => (c.matrix, c.transfer, c.primaries, c.range), - None => { - let (m, t, p, r) = match v.color_space { - ColorSpace::Bt2020 => ( - CICP_MATRIX_BT2020NC, - CICP_TRANSFER_PQ, - CICP_PRIMARIES_BT2020, - COLOUR_RANGE_LIMITED, - ), - ColorSpace::Bt709 => ( - CICP_MATRIX_BT709, - CICP_TRANSFER_BT709, - CICP_PRIMARIES_BT709, - COLOUR_RANGE_LIMITED, - ), - // PAL SD: BT.470 System B/G matrix/transfer/primaries. - ColorSpace::Bt470bg => ( - CICP_MATRIX_BT470BG, - CICP_TRANSFER_BT470BG, - CICP_PRIMARIES_BT470BG, - COLOUR_RANGE_LIMITED, - ), - // NTSC SD: SMPTE 170M / BT.601-525. - ColorSpace::Smpte170m => ( - CICP_MATRIX_BT601_525, - CICP_TRANSFER_BT601_525, - CICP_PRIMARIES_BT601_525, - COLOUR_RANGE_LIMITED, - ), - ColorSpace::Unknown => (0, 0, 0, 0), - }; - // Override the transfer for non-PQ HDR signalled by the HdrFormat - // (the enum can't express HLG). Only applies on the enum - // fallback; a measured CICP already carries the real transfer. - let t = match v.hdr { - HdrFormat::Hdr10 | HdrFormat::Hdr10Plus | HdrFormat::DolbyVision => { - CICP_TRANSFER_PQ - } - HdrFormat::Hlg => CICP_TRANSFER_HLG, - _ => t, - }; - (m, t, p, r) - } - }; + // Derived by the single shared resolver so every sink (this muxer, the + // FVI sidecar in `videomap.rs`) reports identical code points. + let (matrix, transfer, primaries, range) = cicp_for_video(v); // Display dimensions. For square-pixel video (HD/UHD/BD) the display // aspect equals the pixel grid, so display == pixel. For anamorphic // content (DVD: 720x480/576 pixels shown as 16:9 or 4:3) the coded @@ -498,6 +506,13 @@ pub struct MkvMuxer { /// non-monotonic. Keying the exemption on track type (not index) keeps that /// EL's true PTS instead of clobbering it to prev+1ms. track_is_video: Vec, + /// Index of the PRIMARY video track — the first track whose type is video. + /// This (not the literal index 0) is the clip-boundary epoch driver: the + /// M2TS/PMT path orders streams by PMT declaration order and may list an + /// audio ES before the video ES, so `streams[0]` is not guaranteed to be + /// the primary video. `None` when the title has no video track (no track + /// drives epochs). + primary_video_track: Option, /// Cross-clip timeline-continuity corrector (clip-boundary PTS rebasing). continuity: TimelineContinuity, cues: Vec, @@ -961,6 +976,9 @@ impl MkvMuxer { .iter() .map(|t| t.track_type == ebml::TRACK_TYPE_VIDEO) .collect(), + primary_video_track: tracks + .iter() + .position(|t| t.track_type == ebml::TRACK_TYPE_VIDEO), continuity: TimelineContinuity::new(), cues: Vec::new(), frame_count: 0, @@ -1026,7 +1044,9 @@ impl MkvMuxer { let is_video = self.track_is_video.get(track_idx).copied().unwrap_or(false); // The clip-boundary epoch decision is driven by the PRIMARY video track - // ONLY (track 0). A title can carry a SECOND video track — a Dolby Vision + // ONLY (the first video track, NOT the literal index 0 — the M2TS/PMT + // path can list an audio ES before the video ES, so streams[0] may be + // audio). A title can carry a SECOND video track — a Dolby Vision // enhancement layer — whose PTS runs on its OWN timeline, interleaved // with the base layer's. The two video PTS sequences overlap, so the EL's // frames look like multi-second backward jumps against the base layer's @@ -1034,7 +1054,7 @@ impl MkvMuxer { // ratchet that inflated Top Gun's 1-clip timeline to ~7 h). Only the base // video layer establishes/advances the frontier and opens epochs; the EL // — like audio and subtitles — rides the current offset. - let drives_epoch = track_idx == 0; + let drives_epoch = Some(track_idx) == self.primary_video_track; // Map the raw PES PTS onto the continuous output timeline FIRST, before // any base/cluster math: at a non-seamless clip / layer-break boundary @@ -1047,8 +1067,13 @@ impl MkvMuxer { let raw_ticks = pts_ns / TIMESTAMP_SCALE_NS; // Cluster boundaries normally coincide with a video keyframe so every - // Cues entry resolves to a seekable IDR at the cluster start. - let is_video_key = keyframe && track_idx == 0; + // Cues entry resolves to a seekable IDR at the cluster start. Keyed on + // the primary video track (first video index) when the title HAS video; + // for an audio-only / subtitle-only title (no video track) fall back to + // the first track (index 0) so its keyframes still open clusters — + // otherwise no cluster would ever open and every frame would be dropped. + let cluster_driver = self.primary_video_track.unwrap_or(0); + let is_video_key = keyframe && track_idx == cluster_driver; // Derive the timestamp base from the first *kept* keyframe (the frame // that opens the first cluster), NOT the first frame merely seen. The @@ -1905,6 +1930,70 @@ mod tests { ); } + /// Regression for the hardcoded-`track 0` epoch driver: on an M2TS/PMT + /// title the PMT may list an AUDIO ES before the VIDEO ES, so `streams[0]` + /// is audio and the video is at index 1. The epoch driver must follow the + /// PRIMARY VIDEO track (first video index), not the literal 0. If audio + /// (index 0) drove epochs, its sparse/lagging PTS would ratchet the frontier + /// and false-trigger boundary resets, inflating the timeline. This drives + /// the muxer with audio=track0 / video=track1 and asserts cluster timestamps + /// stay monotonic and the timeline does NOT ratchet past the real span. + #[test] + fn epoch_driver_follows_primary_video_not_index_zero() { + // Audio FIRST (index 0), video SECOND (index 1) — the M2TS/PMT ordering. + let tracks = [make_audio_track(), make_video_track()]; + // The muxer must pick track 1 (first video) as the epoch driver. + { + let buf = Cursor::new(Vec::new()); + let mux = MkvMuxer::new(buf, &tracks, None, 0.0, &[]).unwrap(); + assert_eq!( + mux.primary_video_track, + Some(1), + "primary video must be the first VIDEO track (index 1), not 0" + ); + } + let ms = |m: i64| m * 1_000_000; + // Track 0 = AUDIO, track 1 = VIDEO. Same clip-boundary + straggler shape + // as clip_boundary_with_straggler_yields_monotonic_clusters, but with the + // video at index 1. + let frames: Vec<(usize, i64, bool, Vec)> = vec![ + (1, ms(0), true, vec![0x01; 16]), // video kf 0s + (0, ms(0), true, vec![0xA0; 8]), // audio 0s + (1, ms(600_000), true, vec![0x02; 16]), // video kf 600s + (0, ms(600_000), true, vec![0xA1; 8]), // audio 600s + // Clip 2: video keyframe RESETS to 0 (the -600s boundary). + (1, ms(0), true, vec![0x03; 16]), + // Straggler: clip 1's tail audio arrives interleaved after the reset. + (0, ms(599_500), true, vec![0xA2; 8]), + (0, ms(0), true, vec![0xA3; 8]), // clip2 audio at 0 + (1, ms(5_000), true, vec![0x04; 16]), // clip2 + 5s video kf + ]; + let (data, frame_count) = mux_to_bytes(&tracks, &[], &frames); + assert_eq!(frame_count, 8, "all frames written (none dropped)"); + + let tick = |ms: i64| ms * 1_000_000 / TIMESTAMP_SCALE_NS; + let clusters = find_clusters(&data); + let ts: Vec = clusters.iter().map(|&(_, _, t)| t).collect(); + assert!(!ts.is_empty(), "expected clusters"); + assert!( + ts.windows(2).all(|w| w[1] >= w[0]), + "cluster timestamps must be monotonic, got {ts:?}" + ); + let max = *ts.iter().max().unwrap() as i64; + // Timeline reaches past the boundary (clip 2 present): ≥ ~600s. + assert!( + max >= tick(600_000), + "timeline must span past the boundary, got {max} ticks" + ); + // Must NOT ratchet far beyond clip1+clip2 (~605s). With the bug (audio + // index 0 driving epochs) the lagging-audio straggler ratchets the + // frontier and inflates the timeline well past this bound. + assert!( + max < tick(1_000_000), + "no ratchet: max cluster ts {max} ticks must stay near 605s" + ); + } + #[test] fn mkv_multiple_tracks() { let buf = Cursor::new(Vec::new()); diff --git a/src/mux/mkvstream.rs b/src/mux/mkvstream.rs index db31fb7..ec0dfae 100644 --- a/src/mux/mkvstream.rs +++ b/src/mux/mkvstream.rs @@ -815,7 +815,11 @@ fn parse_block( let rel_ts = i16::from_be_bytes([block[vl], block[vl + 1]]); let keyframe = block[vl + 2] & 0x80 != 0; let data = block[vl + 3..].to_vec(); - let pts_ticks = cluster_ts_ticks + rel_ts as i64; + // saturating_add: a hostile CLUSTER_TIMESTAMP near i64::MAX plus a positive + // rel_ts would overflow this add (panic in debug/test, wrap to a large + // negative PTS in release) — one operation BEFORE the saturating_mul below. + // rel_ts as i64 is exact, so this fully bounds the sum on adversarial input. + let pts_ticks = cluster_ts_ticks.saturating_add(rel_ts as i64); let track_idx = (track as usize) - 1; // track >= 1 checked above // Skip blocks for non-existent tracks. @@ -1553,6 +1557,23 @@ mod tests { assert_eq!(f.pts, i64::MAX, "ticks→ns must saturate, not wrap/panic"); } + #[test] + fn parse_block_cluster_ts_plus_rel_ts_saturates_no_overflow() { + // Regression: a hostile CLUSTER_TIMESTAMP near i64::MAX plus a POSITIVE + // rel_ts overflows the `cluster_ts + rel_ts` ADD — one step before the + // saturating_mul. With a plain `+` this panics in debug/test (overflow + // checks on) and silently wraps to a large negative PTS in release. + // rel_ts = +0x7FFF = 32767 (max positive signed 16-bit). + let block = [0x81u8, 0x7F, 0xFF, 0x80, 0xAA]; + let f = parse_block(&block, i64::MAX, 1_000_000, 1, None).unwrap(); + // The add saturates at i64::MAX, then the mul saturates too. + assert_eq!( + f.pts, + i64::MAX, + "cluster_ts + rel_ts must saturate, not panic/wrap" + ); + } + // ============================================================ // ts_pid_for_track — mid-range mapping (the existing test covers the // edges; this fills in a representative middle value to lock the diff --git a/src/mux/timeline.rs b/src/mux/timeline.rs index 64bf0f1..f323b0a 100644 --- a/src/mux/timeline.rs +++ b/src/mux/timeline.rs @@ -118,16 +118,21 @@ impl TimelineContinuity { // frontier and would force a forward-dated split cluster (breaking // cluster monotonicity). Such a straggler is recognised precisely: its // current-offset mapping lands more than a backstep PAST the frontier - // AND its PREVIOUS-offset mapping lands at/below the frontier (i.e. it - // belongs to the prior epoch). Remap it with the previous offset so - // it lands at its true seam position. This is what distinguishes a - // tail straggler from a frame that legitimately runs ahead of the - // (base-video-only) frontier — a long audio-only tail, a sparse - // subtitle, or an EL frame — which is left on the current offset. + // AND its PREVIOUS-offset mapping lands in the seam TAIL — at/below the + // frontier but no more than one backstep below it (i.e. it ended just + // before the seam, in the prior epoch). The lower bound is essential: + // a NORMAL new-epoch frame that merely leads the sparse (video-only) + // frontier by >3s ALSO has `prev_mapped <= high` (its prev-offset + // mapping lands ~a whole clip below the frontier), and clamping it + // would demote it into the just-ended clip's epoch, mis-timing that + // audio/subtitle by a whole clip. Requiring `prev_mapped` to sit + // within a backstep below the frontier keeps the remap to genuine + // tail stragglers; a long audio-only tail, a sparse subtitle, or an + // EL frame that simply runs ahead is left on the current offset. if let Some(high) = self.high_ns { if mapped > high + DISCONTINUITY_BACKSTEP_NS { let prev_mapped = raw_pts_ns.saturating_add(self.prev_offset_ns); - if prev_mapped <= high { + if prev_mapped <= high && prev_mapped >= high - DISCONTINUITY_BACKSTEP_NS { return prev_mapped; } } @@ -418,4 +423,44 @@ mod tests { let normal = adj_other(&mut tc, S); assert_eq!(normal, S + 600 * S + DISCONTINUITY_GAP_NS); } + + /// Regression for the over-eager straggler clamp: a NORMAL new-epoch + /// non-video frame that leads the (sparse, video-only) frontier by MORE than + /// one backstep must ride the CURRENT offset — it must NOT be demoted into + /// the just-ended clip's epoch. Such a frame satisfies BOTH of the old + /// discriminator's conditions (current-map > frontier+backstep AND + /// prev-map <= frontier), so the old `prev_mapped <= high` test wrongly + /// clamped it back ~a whole clip. The tightened lower bound + /// (`prev_mapped >= high - backstep`) fixes it. + #[test] + fn normal_new_epoch_frame_leading_frontier_is_not_clamped() { + let mut tc = TimelineContinuity::new(); + // Clip1 video rises to 600s, then clip2 resets to 0 → boundary. + for i in 0..=600 { + adj_video(&mut tc, i * S); + } + let frontier = tc.high_ns.unwrap(); + assert_eq!(frontier, 600 * S); + let c2 = adj_video(&mut tc, 0); + assert_eq!(c2, 600 * S + DISCONTINUITY_GAP_NS); + + // A NORMAL clip-2 audio frame at raw ~5s. Current-offset mapping is + // ~605s, which IS more than a backstep (3s) past the 600s frontier — but + // its previous-offset mapping (~5s) lands ~595s BELOW the frontier, far + // outside the seam tail. It is a legitimate new-epoch frame, NOT a tail + // straggler, and must ride the current offset. + let raw = 5 * S; + let out = adj_other(&mut tc, raw); + assert_eq!( + out, + raw + 600 * S + DISCONTINUITY_GAP_NS, + "a normal new-epoch frame leading the frontier by >3s must ride the \ + current offset, not be clamped back into the previous clip" + ); + // And it must NOT have been demoted near the previous clip's tail (~5s). + assert!( + out > frontier, + "frame must stay in the new epoch (> frontier), got {out}" + ); + } } diff --git a/src/mux/videomap.rs b/src/mux/videomap.rs index 146f20b..5533082 100644 --- a/src/mux/videomap.rs +++ b/src/mux/videomap.rs @@ -65,8 +65,27 @@ pub struct Colour { } impl Colour { - /// Map the title's [`ColorSpace`] to CICP code points. Unknown colorimetry - /// maps to code point 2 ("unspecified"), the CICP convention. + /// Derive the FVI CICP code points from a full [`VideoStream`], using the + /// SAME precedence as the MKV muxer ([`crate::mux::mkv::cicp_for_video`]): + /// measured CICP (authoritative) → coarse `color_space` enum + HDR-driven + /// transfer override. This is what the sidecar must use so it never reports + /// an SDR transfer (14) for an HDR10 BT.2020 title while the MKV container + /// reports PQ (16) — the two sinks of one title must agree. + pub fn from_video(v: &VideoStream) -> Self { + let (matrix, transfer, primaries, range) = crate::mux::mkv::cicp_for_video(v); + Self { + primaries, + transfer, + matrix, + // Matroska/MeasuredCicp Range: 1 = limited (disc norm), 2 = full. + full_range: range == 2, + } + } + + /// Map the title's [`ColorSpace`] alone to CICP code points (no HDR/measured + /// context). Retained for the no-video header fallback and unit coverage; + /// the title path uses [`Colour::from_video`]. Unknown colorimetry maps to + /// code point 2 ("unspecified"), the CICP convention. pub fn from_color_space(cs: ColorSpace) -> Self { // (primaries, transfer, matrix) per ITU-T H.273. let (p, t, m) = match cs { @@ -217,7 +236,7 @@ impl MapHeader { } else { Scan::Progressive }, - colour: Colour::from_color_space(v.color_space), + colour: Colour::from_video(v), } } None => StreamInfo { @@ -475,6 +494,65 @@ mod tests { assert_eq!(Colour::from_color_space(ColorSpace::Unknown).primaries, 2); } + /// Regression: the FVI sidecar must mirror the MKV muxer's colour precedence, + /// not blindly map `color_space` → the SDR transfer 14 for BT.2020. An HDR10 + /// BT.2020 title's real transfer is PQ (16); a measured CICP triplet is + /// authoritative and copied through verbatim. Before the fix the FVI Colour + /// reported transfer=14 while the MKV container reported 16 — two sinks of + /// one title disagreeing on the colour code points. + #[test] + fn fvi_colour_follows_hdr_and_measured_cicp() { + use crate::disc::MeasuredCicp; + let mk = |hdr: HdrFormat, cs: ColorSpace, cicp: Option| VideoStream { + pid: 0x1011, + codec: Codec::Hevc, + resolution: Resolution::R2160p, + frame_rate: FrameRate::F23_976, + hdr, + color_space: cs, + display_aspect: None, + secondary: false, + label: String::new(), + measured_cicp: cicp, + }; + + // HDR10 BT.2020 with NO measured CICP → PQ transfer (16), NOT SDR 14. + let c = Colour::from_video(&mk(HdrFormat::Hdr10, ColorSpace::Bt2020, None)); + assert_eq!( + c, + Colour { + primaries: 9, + transfer: 16, // PQ — not the SDR 14 the enum alone would give + matrix: 9, + full_range: false, + } + ); + + // HLG BT.2020 → transfer 18. + let c = Colour::from_video(&mk(HdrFormat::Hlg, ColorSpace::Bt2020, None)); + assert_eq!(c.transfer, 18, "HLG transfer must be 18"); + + // Measured CICP is authoritative — copied through verbatim, incl. full + // range (2 → full_range = true), ignoring the coarse enum/HDR guess. + let measured = MeasuredCicp { + matrix: 9, + transfer: 16, + primaries: 9, + range: 2, + }; + let c = Colour::from_video(&mk(HdrFormat::Sdr, ColorSpace::Bt709, Some(measured))); + assert_eq!( + c, + Colour { + primaries: 9, + transfer: 16, + matrix: 9, + full_range: true, + }, + "measured CICP must override the coarse color_space enum" + ); + } + #[test] fn type_label_full_and_codec_agnostic_fallback() { // coding present: full I/P/B from the agnostic coding_type(). diff --git a/src/sector/prefetched.rs b/src/sector/prefetched.rs index ff22686..0425af4 100644 --- a/src/sector/prefetched.rs +++ b/src/sector/prefetched.rs @@ -168,136 +168,158 @@ impl PrefetchedSectorSource { let producer = std::thread::Builder::new() .name("freemkv-prefetch".into()) .spawn(move || { - let mut ext_idx = 0usize; - let mut offset: u32 = 0; - let mut bytes_read_total: u64 = 0; - while ext_idx < extents.len() { - if halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false) { - return; - } - let extent = &extents[ext_idx]; - // AACS aligned units are anchored at THIS extent's start - // LBA, so tell the decrypt-on-read source to gate relative - // to it (clip-anchored), not absolute disc LBA 0. A no-op - // for non-decrypting / CSS / None sources. - reader.set_unit_base(extent.start_lba); - let remaining = extent.sector_count.saturating_sub(offset); - if remaining == 0 { - ext_idx += 1; - offset = 0; - continue; - } - // The AACS aligned unit is SECTOR_ALIGNMENT (3) - // sectors / 6144 bytes; the decrypt step only - // processes full units and silently leaves a - // shorter trailing chunk encrypted. So a batch must - // be a whole number of units — except for the final - // batch of an extent whose `sector_count` is itself - // unit-aligned (then the remaining tail is exactly - // 0 mod 3 and forms full units on its own). - // - // If the trailing sectors of this extent cannot fill - // a complete unit (`remaining < SECTOR_ALIGNMENT` - // with nothing more to read, or a 1-2 sector - // leftover after the last full unit), there is no - // way to hand the decrypt step an aligned chunk — - // surface a typed error instead of emitting - // still-encrypted bytes. - if remaining % unit_align as u32 != 0 && remaining < unit_align as u32 { - let _ = tx.send(Err(crate::error::Error::ExtentNotUnitAligned.into())); - return; - } - let mut sectors = remaining.min(batch_sectors as u32) as u16; - // Trim to a whole number of units. Once trimmed to 0 - // here it means `remaining >= SECTOR_ALIGNMENT` but - // the *batch window* landed on a sub-unit boundary — - // never the trailing-tail case, which the guard - // above already rejected. Clamp to one unit so we - // always make forward progress. - if sectors >= unit_align { - sectors -= sectors % unit_align; - } else { - sectors = unit_align; - } - let bytes = sectors as usize * 2048; - // Park on the recycle channel, but re-poll halt every - // POLL_INTERVAL: a pure-AtomicBool Halt does not - // disconnect the channel, so a blocking recv() would - // never re-reach the cancel check at the loop top. - // Mirrors the BytePrefetcher pattern exactly. - let mut buf = loop { - match recycle_rx.recv_timeout(POLL_INTERVAL) { - Ok(b) => break b, - Err(RecvTimeoutError::Timeout) => { - if halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false) { - return; - } - } - // Consumer dropped both channels. - Err(RecvTimeoutError::Disconnected) => return, - } - }; - if bytes <= buf.capacity() { - // Re-expose `bytes` without zero-filling pages that - // `read_sectors` is about to overwrite. The enclosing - // capacity guard makes the `set_len` provably sound even - // if a recycled buffer ever comes back smaller than the - // `vec![0u8; batch_bytes]` it was born with. - debug_assert!(bytes <= buf.capacity(), "set_len exceeds capacity"); - unsafe { buf.set_len(bytes) }; - } else { - buf.resize(bytes, 0); - } - // `start_lba + offset` derives from untrusted extent - // data — saturate rather than wrap/panic on a - // hostile start_lba near u32::MAX. - let lba = extent.start_lba.saturating_add(offset); - match reader.read_sectors(lba, sectors, &mut buf[..bytes], false) { - Ok(n) => { - // A short read must not silently desync the - // stream: advance the extent cursor by the - // sectors actually read, not the requested - // count, and reject a byte count that isn't a - // whole number of sectors (it would split a - // sector and leave the decrypt step a partial - // unit). The sole production inner source - // (FileSectorSource) read_exact's the full - // request, so this is belt-and-braces against - // a future short-reading source. - if n % 2048 != 0 { - let _ = - tx.send(Err(crate::error::Error::ExtentNotUnitAligned.into())); - return; - } - let sectors_read = (n / 2048) as u32; - buf.truncate(n); - bytes_read_total = bytes_read_total.saturating_add(n as u64); - if let Some(ref f) = event_fn { - f(Event { - kind: EventKind::BytesRead { - bytes: bytes_read_total, - total: bytes_total_extents, - }, - }); - } - if tx.send(Ok(buf)).is_err() { - return; // consumer dropped - } - // A genuine zero-byte read with no error would - // otherwise spin this loop forever; treat it - // as end-of-source. - if sectors_read == 0 { - return; - } - offset = offset.saturating_add(sectors_read); - } - Err(e) => { - let _ = tx.send(Err(e.into())); + // Wrap the whole feed loop in catch_unwind so a panic inside a + // decrypt-on-read SectorSource, the read path, or a panicking + // `event_fn` callback is NOT indistinguishable from a clean + // finish at the demux boundary. On a clean exit (input + // exhausted, halt cancelled, consumer disconnect) the body + // returns and `tx` is dropped → the demux loop reads RecvError + // as EOF, which is correct. On a PANIC we send an explicit error + // sentinel FIRST so the demux loop's `Ok(Err(_))` arm fires and + // propagates a typed error instead of converting the dropped + // channel into a clean `DemuxBatch::Eof` (which would finalize a + // TRUNCATED mux while reporting success). `tx`/`reader`/locals + // are only touched on this thread, so AssertUnwindSafe is sound. + let body = std::panic::AssertUnwindSafe(|| { + let mut ext_idx = 0usize; + let mut offset: u32 = 0; + let mut bytes_read_total: u64 = 0; + while ext_idx < extents.len() { + if halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false) { return; } + let extent = &extents[ext_idx]; + // AACS aligned units are anchored at THIS extent's start + // LBA, so tell the decrypt-on-read source to gate relative + // to it (clip-anchored), not absolute disc LBA 0. A no-op + // for non-decrypting / CSS / None sources. + reader.set_unit_base(extent.start_lba); + let remaining = extent.sector_count.saturating_sub(offset); + if remaining == 0 { + ext_idx += 1; + offset = 0; + continue; + } + // The AACS aligned unit is SECTOR_ALIGNMENT (3) + // sectors / 6144 bytes; the decrypt step only + // processes full units and silently leaves a + // shorter trailing chunk encrypted. So a batch must + // be a whole number of units — except for the final + // batch of an extent whose `sector_count` is itself + // unit-aligned (then the remaining tail is exactly + // 0 mod 3 and forms full units on its own). + // + // If the trailing sectors of this extent cannot fill + // a complete unit (`remaining < SECTOR_ALIGNMENT` + // with nothing more to read, or a 1-2 sector + // leftover after the last full unit), there is no + // way to hand the decrypt step an aligned chunk — + // surface a typed error instead of emitting + // still-encrypted bytes. + if remaining % unit_align as u32 != 0 && remaining < unit_align as u32 { + let _ = tx.send(Err(crate::error::Error::ExtentNotUnitAligned.into())); + return; + } + let mut sectors = remaining.min(batch_sectors as u32) as u16; + // Trim to a whole number of units. Once trimmed to 0 + // here it means `remaining >= SECTOR_ALIGNMENT` but + // the *batch window* landed on a sub-unit boundary — + // never the trailing-tail case, which the guard + // above already rejected. Clamp to one unit so we + // always make forward progress. + if sectors >= unit_align { + sectors -= sectors % unit_align; + } else { + sectors = unit_align; + } + let bytes = sectors as usize * 2048; + // Park on the recycle channel, but re-poll halt every + // POLL_INTERVAL: a pure-AtomicBool Halt does not + // disconnect the channel, so a blocking recv() would + // never re-reach the cancel check at the loop top. + // Mirrors the BytePrefetcher pattern exactly. + let mut buf = loop { + match recycle_rx.recv_timeout(POLL_INTERVAL) { + Ok(b) => break b, + Err(RecvTimeoutError::Timeout) => { + if halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false) { + return; + } + } + // Consumer dropped both channels. + Err(RecvTimeoutError::Disconnected) => return, + } + }; + if bytes <= buf.capacity() { + // Re-expose `bytes` without zero-filling pages that + // `read_sectors` is about to overwrite. The enclosing + // capacity guard makes the `set_len` provably sound even + // if a recycled buffer ever comes back smaller than the + // `vec![0u8; batch_bytes]` it was born with. + debug_assert!(bytes <= buf.capacity(), "set_len exceeds capacity"); + unsafe { buf.set_len(bytes) }; + } else { + buf.resize(bytes, 0); + } + // `start_lba + offset` derives from untrusted extent + // data — saturate rather than wrap/panic on a + // hostile start_lba near u32::MAX. + let lba = extent.start_lba.saturating_add(offset); + match reader.read_sectors(lba, sectors, &mut buf[..bytes], false) { + Ok(n) => { + // A short read must not silently desync the + // stream: advance the extent cursor by the + // sectors actually read, not the requested + // count, and reject a byte count that isn't a + // whole number of sectors (it would split a + // sector and leave the decrypt step a partial + // unit). The sole production inner source + // (FileSectorSource) read_exact's the full + // request, so this is belt-and-braces against + // a future short-reading source. + if n % 2048 != 0 { + let _ = tx.send(Err( + crate::error::Error::ExtentNotUnitAligned.into() + )); + return; + } + let sectors_read = (n / 2048) as u32; + buf.truncate(n); + bytes_read_total = bytes_read_total.saturating_add(n as u64); + if let Some(ref f) = event_fn { + f(Event { + kind: EventKind::BytesRead { + bytes: bytes_read_total, + total: bytes_total_extents, + }, + }); + } + if tx.send(Ok(buf)).is_err() { + return; // consumer dropped + } + // A genuine zero-byte read with no error would + // otherwise spin this loop forever; treat it + // as end-of-source. + if sectors_read == 0 { + return; + } + offset = offset.saturating_add(sectors_read); + } + Err(e) => { + let _ = tx.send(Err(e.into())); + return; + } + } } + // Drop tx implicitly — consumer sees RecvError → EOF. + }); + if std::panic::catch_unwind(body).is_err() { + // Producer panicked mid-stream — surface a typed terminal + // error so the demux thread does NOT read the dropped channel + // as a clean EOF and truncate output. Ignore the send result: + // if the consumer is already gone there is nothing to report. + let _ = tx.send(Err(crate::error::Error::DemuxThreadPanicked.into())); } - // Drop tx implicitly — consumer sees RecvError → EOF. }) .map_err(|e| crate::error::Error::IoError { source: e })?;