From 17318f682c8303b85ee2b485f719ef96788876ea Mon Sep 17 00:00:00 2001 From: FedyaLight Date: Wed, 11 Nov 2026 20:48:39 +0300 Subject: [PATCH] Keep damaged audio packets from stopping synchronization Symphonia's AAC decoder reports a truncated or corrupted packet as an I/O error ("unexpected end of bitstream"). Align treated it as fatal, so one damaged packet in one camera file stopped the whole run without naming the file (#16). A rejected packet now becomes silence of its recorded duration, so the rest of the file is still used and keeps its timing. Symphonia is chosen only when the first decodable packet does not yield far fewer samples than its container records; HE-AAC, whose SBR layer Symphonia lacks, yields half and goes to FFmpeg. M4A is decoded like MP4, with its edit list. When the in-process decoder (Symphonia, or AVFoundation on macOS) still fails on a file during analysis, FFmpeg decodes it again from the start, also for a mix of all streams. Only a file that neither can read is reported, in a warning that names it, and the other files still synchronize. When only some streams of a file can be read, refinement uses the stream analysis kept. Cancellation ends the run, and so does a missing FFmpeg except on macOS, where Align does not ship it. Symphonia 0.5 ignores the probe hint and scans the first megabyte of any file for MP3 or ADTS sync words. In MXF video it finds them and decodes video data as audio, and when the reader rejects the match it can read the whole file. Containers are now opened by the reader their extension names; MXF, MTS and R3D go straight to FFmpeg. Fingerprints cached by the previous decoder are not reused (portable4). The status switches to Analyzing when audio analysis starts, instead of showing Reading metadata until the first file finishes, and analysis progress counts are published in order. --- crates/align-decode/src/backend.rs | 2 +- crates/align-decode/src/mix.rs | 82 ++-- crates/align-decode/src/pipeline.rs | 606 +++++++++++++++++++++++----- crates/align-decode/src/sym.rs | 524 ++++++++++++++++++------ docs/formats.md | 12 +- 5 files changed, 981 insertions(+), 245 deletions(-) diff --git a/crates/align-decode/src/backend.rs b/crates/align-decode/src/backend.rs index da83e81..567c67b 100644 --- a/crates/align-decode/src/backend.rs +++ b/crates/align-decode/src/backend.rs @@ -35,7 +35,7 @@ impl BackendKind { /// must not assume it). pub fn cache_tag(&self) -> &'static str { match self { - BackendKind::Portable => "portable3", + BackendKind::Portable => "portable4", BackendKind::AppleNative => "apple3", } } diff --git a/crates/align-decode/src/mix.rs b/crates/align-decode/src/mix.rs index 4a36f4e..3f6f173 100644 --- a/crates/align-decode/src/mix.rs +++ b/crates/align-decode/src/mix.rs @@ -11,17 +11,37 @@ use crate::backend::MediaBackend; const SAMPLE_RATE_16K: f64 = 16_000.0; const IO_SAMPLES: usize = 32_768; +pub(crate) type Consumer<'a> = &'a mut dyn FnMut(&[f32]) -> Result<(), DecodeError>; + pub(crate) fn decode_all_streams_mono_8k( backend: &dyn MediaBackend, path: &Path, stream_count: usize, - consume: &mut dyn FnMut(&[f32]) -> Result<(), DecodeError>, + consume: Consumer, +) -> Result<(), DecodeError> { + mix_streams_mono_8k( + path, + stream_count, + &|index, consume| { + backend.decode_mono_8k(path, AudioAnalysisSource::MixedStream(index), consume) + }, + consume, + ) +} + +/// Mixes every stream, each decoded to 8 kHz mono with its channels mixed by +/// `decode_stream`. +pub(crate) fn mix_streams_mono_8k( + path: &Path, + stream_count: usize, + decode_stream: &dyn Fn(usize, Consumer) -> Result<(), DecodeError>, + consume: Consumer, ) -> Result<(), DecodeError> { if stream_count == 0 { return Err(DecodeError::NoAudio(path.display().to_string())); } if stream_count == 1 { - return backend.decode_mono_8k(path, AudioAnalysisSource::MixedStream(0), consume); + return decode_stream(0, consume); } // Decode streams sequentially so one clip still occupies one media-reader @@ -31,38 +51,34 @@ pub(crate) fn decode_all_streams_mono_8k( for index in 0..stream_count { let mut cursor = 0_u64; let mut bytes = Vec::new(); - backend.decode_mono_8k( - path, - AudioAnalysisSource::MixedStream(index), - &mut |samples| { - if samples.is_empty() { - return Ok(()); - } - bytes.resize(samples.len() * 4, 0); - sum.seek(SeekFrom::Start(cursor * 4))?; - let mut read = 0; - while read < bytes.len() { - match sum.read(&mut bytes[read..])? { - 0 => break, - count => read += count, - } + decode_stream(index, &mut |samples| { + if samples.is_empty() { + return Ok(()); + } + bytes.resize(samples.len() * 4, 0); + sum.seek(SeekFrom::Start(cursor * 4))?; + let mut read = 0; + while read < bytes.len() { + match sum.read(&mut bytes[read..])? { + 0 => break, + count => read += count, } - sum.seek(SeekFrom::Start(cursor * 4))?; - for (sample_index, sample) in samples.iter().enumerate() { - let offset = sample_index * 4; - let previous = if offset + 4 <= read { - f32::from_le_bytes(bytes[offset..offset + 4].try_into().unwrap()) - } else { - 0.0 - }; - bytes[offset..offset + 4].copy_from_slice(&(previous + sample).to_le_bytes()); - } - sum.write_all(&bytes)?; - cursor += samples.len() as u64; - max_samples = max_samples.max(cursor); - Ok(()) - }, - )?; + } + sum.seek(SeekFrom::Start(cursor * 4))?; + for (sample_index, sample) in samples.iter().enumerate() { + let offset = sample_index * 4; + let previous = if offset + 4 <= read { + f32::from_le_bytes(bytes[offset..offset + 4].try_into().unwrap()) + } else { + 0.0 + }; + bytes[offset..offset + 4].copy_from_slice(&(previous + sample).to_le_bytes()); + } + sum.write_all(&bytes)?; + cursor += samples.len() as u64; + max_samples = max_samples.max(cursor); + Ok(()) + })?; } sum.seek(SeekFrom::Start(0))?; diff --git a/crates/align-decode/src/pipeline.rs b/crates/align-decode/src/pipeline.rs index 2af4ab5..cdea230 100644 --- a/crates/align-decode/src/pipeline.rs +++ b/crates/align-decode/src/pipeline.rs @@ -122,6 +122,27 @@ impl WaveformPreview { } } +struct FingerprintRun { + features: Vec, + selected_sources: HashMap, + /// Clips left out because their audio could not be read. + warnings: Vec, +} + +/// Errors that end the whole run rather than one file's analysis: a +/// cancellation, or a missing decoder that every such file needs. +fn ends_run(error: &PipelineError) -> bool { + matches!( + error, + PipelineError::Cancelled + | PipelineError::Decode( + crate::DecodeError::Cancelled + | crate::DecodeError::FfmpegMissing(_) + | crate::DecodeError::BackendUnavailable(_) + ) + ) +} + struct FingerprintRunOptions<'a> { accuracy: align_core::SearchAccuracy, overrides: &'a HashMap, @@ -518,7 +539,11 @@ impl Pipeline { .entry(clip.id.clone()) .or_insert(options.audio_source); } - let (features, selected_sources) = self.fingerprints( + let FingerprintRun { + features, + selected_sources, + warnings: unreadable, + } = self.fingerprints( &usable, &audio_sources, FingerprintRunOptions { @@ -529,6 +554,7 @@ impl Pipeline { progress, cancel, )?; + warnings.extend(unreadable); let mut source_map = audio_sources; source_map.extend(selected_sources); @@ -871,6 +897,82 @@ impl Pipeline { }) } + /// Decodes one audio source to 8 kHz mono into state that `start` creates. + /// When the in-process decoder (Symphonia, or AVFoundation on macOS) + /// fails on the file's content, FFmpeg decodes it again into fresh state, + /// so a file is reported unreadable only when FFmpeg cannot read it + /// either. + fn decode_mono_8k_into( + &self, + clip: &Clip, + source: AudioAnalysisSource, + cancel: &std::sync::atomic::AtomicBool, + start: impl Fn() -> S, + feed: impl Fn(&mut S, &[f32]), + ) -> Result { + use crate::DecodeError; + use crate::mix::Consumer; + let run = |decode: &dyn Fn(Consumer) -> Result<(), DecodeError>| { + let mut state = start(); + decode(&mut |block| { + if cancelled(cancel) { + return Err(DecodeError::Cancelled); + } + feed(&mut state, block); + Ok(()) + }) + .map(|()| state) + }; + let error = match run(&|consume| self.backend.decode_mono_8k(&clip.url, source, consume)) { + Ok(state) => return Ok(state), + Err(error @ (DecodeError::Symphonia(_) | DecodeError::Apple(_))) => error, + Err(DecodeError::Cancelled) => return Err(PipelineError::Cancelled), + Err(error) => return Err(PipelineError::Decode(error)), + }; + let channels = |index: usize| { + clip.audio + .get(index) + .map(|stream| stream.channels) + .ok_or_else(|| DecodeError::NoAudio(clip.url.display().to_string())) + }; + let ffmpeg_stream = |index: usize, mix: bool, channel: Option, consume: Consumer| { + crate::ff::decode_mono_8k(&clip.url, index, channels(index)?, channel, mix, consume) + }; + let ffmpeg = if source == AudioAnalysisSource::AllMixed { + run(&|consume| { + crate::mix::mix_streams_mono_8k( + &clip.url, + clip.audio.len(), + &|index, consume| ffmpeg_stream(index, true, None, consume), + consume, + ) + }) + } else { + run(&|consume| { + ffmpeg_stream( + source.stream_index(), + source.mixes_channels(), + source.selected_channel(), + consume, + ) + }) + }; + match ffmpeg { + Ok(state) => Ok(state), + Err(DecodeError::Cancelled) => Err(PipelineError::Cancelled), + // FFmpeg ships with Align except on macOS, where AVFoundation + // decodes and FFmpeg is optional: there the in-process decoder + // has the only verdict. Elsewhere a missing FFmpeg is a broken + // installation and ends the run. + Err(DecodeError::FfmpegMissing(_)) + if self.backend.kind() == BackendKind::AppleNative => + { + Err(PipelineError::Decode(error)) + } + Err(ffmpeg) => Err(PipelineError::Decode(ffmpeg)), + } + } + fn fingerprints( &self, clips: &[&Clip], @@ -878,50 +980,47 @@ impl Pipeline { options: FingerprintRunOptions<'_>, progress: Option<&(dyn Fn(PipelineProgress) + Send + Sync)>, cancel: &std::sync::atomic::AtomicBool, - ) -> Result< - ( - Vec, - HashMap, - ), - PipelineError, - > { + ) -> Result { use align_core::Fingerprint; - use std::sync::atomic::Ordering; struct Variant { source: AudioAnalysisSource, fingerprints: Vec, waveform: Option>, } let total = clips.len(); - let done = std::sync::atomic::AtomicUsize::new(0); + // Counting and publishing under one lock keeps the completed count + // monotonic for listeners such as the progress bar. + let done = std::sync::Mutex::new(0_usize); let accuracy = options.accuracy; let overrides = options.overrides; let generate_waveform_previews = options.generate_waveform_previews; - // Four streaming readers avoid saturating external media while - // variants of one clip stay sequential and results keep input order. - let all_variants: Vec, PipelineError>> = - align_core::par_map(clips, 4, |clip| { - if cancelled(cancel) { - return Err(PipelineError::Cancelled); - } - let accuracy = overrides.get(&clip.id).copied().unwrap_or(accuracy); - let req = requested - .get(&clip.id) - .copied() - .unwrap_or(AudioAnalysisSource::Automatic); - let sources: Vec = - if req == AudioAnalysisSource::Automatic && clip.audio.len() > 1 { - (0..clip.audio.len()) - .map(|idx| AudioAnalysisSource::Stream { - index: idx, - channel: None, - }) - .collect() - } else { - vec![req] - }; - let mut variants = Vec::with_capacity(sources.len()); - for source in sources { + if let Some(p) = &progress { + p(PipelineProgress::basic(Phase::Fingerprint, 0, total, None)); + } + let analyse = |clip: &&Clip| -> Result, PipelineError> { + if cancelled(cancel) { + return Err(PipelineError::Cancelled); + } + let accuracy = overrides.get(&clip.id).copied().unwrap_or(accuracy); + let req = requested + .get(&clip.id) + .copied() + .unwrap_or(AudioAnalysisSource::Automatic); + let sources: Vec = + if req == AudioAnalysisSource::Automatic && clip.audio.len() > 1 { + (0..clip.audio.len()) + .map(|idx| AudioAnalysisSource::Stream { + index: idx, + channel: None, + }) + .collect() + } else { + vec![req] + }; + let mut variants = Vec::with_capacity(sources.len()); + let mut failure = None; + for source in sources { + let variant = (|| -> Result { let cache_key = accuracy.cache_key(&source.cache_key()); if let Some(cached) = self.cache.load(&clip.url, &cache_key) { let waveform = if generate_waveform_previews { @@ -929,19 +1028,13 @@ impl Pipeline { { Some(waveform) } else { - let mut preview = WaveformPreview::new(clip.duration.as_seconds()); - self.backend - .decode_mono_8k(&clip.url, source, &mut |block| { - if cancelled(cancel) { - return Err(crate::DecodeError::Cancelled); - } - preview.consume(block); - Ok(()) - }) - .map_err(|error| match error { - crate::DecodeError::Cancelled => PipelineError::Cancelled, - other => PipelineError::Decode(other), - })?; + let preview = self.decode_mono_8k_into( + clip, + source, + cancel, + || WaveformPreview::new(clip.duration.as_seconds()), + |preview, block| preview.consume(block), + )?; let waveform = preview.finish(); self.cache.save_waveform(&waveform, &clip.url, &cache_key); Some(waveform) @@ -949,31 +1042,30 @@ impl Pipeline { } else { None }; - variants.push(Variant { + return Ok(Variant { source, fingerprints: cached, waveform, }); - continue; } - let mut extractor = FingerprintExtractor::with_accuracy(accuracy); - let mut preview = generate_waveform_previews - .then(|| WaveformPreview::new(clip.duration.as_seconds())); - self.backend - .decode_mono_8k(&clip.url, source, &mut |block| { - if cancelled(cancel) { - return Err(crate::DecodeError::Cancelled); - } + let (extractor, preview) = self.decode_mono_8k_into( + clip, + source, + cancel, + || { + ( + FingerprintExtractor::with_accuracy(accuracy), + generate_waveform_previews + .then(|| WaveformPreview::new(clip.duration.as_seconds())), + ) + }, + |(extractor, preview), block| { extractor.consume(block); - if let Some(preview) = &mut preview { + if let Some(preview) = preview { preview.consume(block); } - Ok(()) - }) - .map_err(|error| match error { - crate::DecodeError::Cancelled => PipelineError::Cancelled, - other => PipelineError::Decode(other), - })?; + }, + )?; if cancelled(cancel) { return Err(PipelineError::Cancelled); } @@ -983,21 +1075,42 @@ impl Pipeline { if let Some(waveform) = &waveform { self.cache.save_waveform(waveform, &clip.url, &cache_key); } - variants.push(Variant { + Ok(Variant { source, fingerprints, waveform, - }); + }) + })(); + // A stream that cannot be read leaves the others; the + // clip fails only when none of its streams can be read. + match variant { + Ok(variant) => variants.push(variant), + Err(error) if ends_run(&error) => return Err(error), + Err(error) => { + failure.get_or_insert(error); + } } - if let Some(p) = &progress { - let completed = done.fetch_add(1, Ordering::Relaxed) + 1; + } + match failure { + Some(error) if variants.is_empty() => Err(error), + _ => Ok(variants), + } + }; + // Four streaming readers avoid saturating external media while + // variants of one clip stay sequential and results keep input order. + let outcomes: Vec, PipelineError>> = + align_core::par_map(clips, 4, |clip| { + let outcome = analyse(clip); + if let Some(p) = &progress + && !outcome.as_ref().is_err_and(ends_run) + { let mut event = PipelineProgress::basic( Phase::Fingerprint, - completed, + 0, total, Some(clip.url.clone()), ); - if generate_waveform_previews { + if generate_waveform_previews && let Ok(variants) = &outcome { let mut waveform = vec![0.0_f32; WAVEFORM_PREVIEW_BINS]; for preview in variants .iter() @@ -1009,15 +1122,29 @@ impl Pipeline { } event.waveform = Some((clip.id.clone(), waveform)); } + let mut done = done.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + *done += 1; + event.completed = *done; p(event); } - Ok(variants) + outcome }); - let mut collected = Vec::with_capacity(clips.len()); - for variants in all_variants { - collected.push(variants?); + // One unreadable file is reported and left unmatched; the others + // still synchronize. Results stay in input order. + let mut analysed = Vec::with_capacity(clips.len()); + let mut warnings = Vec::new(); + for (clip, outcome) in clips.iter().zip(outcomes) { + match outcome { + Ok(variants) => analysed.push((*clip, variants)), + Err(error) if ends_run(&error) => return Err(error), + Err(error) => warnings.push(SyncWarning { + url: clip.url.clone(), + message: format!( + "The audio could not be read, so this file was left out of synchronization: {error}" + ), + }), + } } - let all_variants = collected; if let Some(p) = &progress { p(PipelineProgress::basic( Phase::Fingerprint, @@ -1027,25 +1154,40 @@ impl Pipeline { )); } - if all_variants.iter().all(|variants| variants.len() == 1) { - return Ok(( - clips - .iter() - .zip(all_variants) - .map(|(clip, mut variants)| align_core::ClipFingerprints { - clip_id: clip.id.clone(), - fingerprints: variants.pop().unwrap().fingerprints, - }) - .collect(), - HashMap::new(), - )); + // Refinement decodes the requested source again, so a stream chosen + // here must be recorded, including the only stream of several that + // could be read. + let requested_source = |clip: &Clip| { + requested + .get(&clip.id) + .copied() + .unwrap_or(AudioAnalysisSource::Automatic) + }; + if analysed.iter().all(|(_, variants)| variants.len() == 1) { + let mut selected = HashMap::new(); + let mut features = Vec::with_capacity(analysed.len()); + for (clip, mut variants) in analysed { + let variant = variants.pop().unwrap(); + if variant.source != requested_source(clip) { + selected.insert(clip.id.clone(), variant.source); + } + features.push(align_core::ClipFingerprints { + clip_id: clip.id.clone(), + fingerprints: variant.fingerprints, + }); + } + return Ok(FingerprintRun { + features, + selected_sources: selected, + warnings, + }); } // Automatic multi-stream selection: most shared hashes wins. // Ties select the lowest channel index. let mut first_owner: HashMap = HashMap::new(); let mut shared: HashSet = HashSet::new(); - for (clip, variants) in clips.iter().zip(all_variants.iter()) { + for (clip, variants) in &analysed { let hashes: HashSet = variants .iter() .flat_map(|v| v.fingerprints.iter().map(|f| f.hash)) @@ -1064,7 +1206,7 @@ impl Pipeline { } let mut features = Vec::with_capacity(clips.len()); let mut selected = HashMap::new(); - for (clip, variants) in clips.iter().zip(all_variants) { + for (clip, variants) in analysed { let mut best = None::<(usize, usize)>; for (idx, v) in variants.iter().enumerate() { let score = v @@ -1079,7 +1221,7 @@ impl Pipeline { if let Some((idx, _)) = best { let mut variants = variants; let v = variants.swap_remove(idx); - if !variants.is_empty() { + if v.source != requested_source(clip) { selected.insert(clip.id.clone(), v.source); } features.push(align_core::ClipFingerprints { @@ -1088,7 +1230,11 @@ impl Pipeline { }); } } - Ok((features, selected)) + Ok(FingerprintRun { + features, + selected_sources: selected, + warnings, + }) } } @@ -1568,6 +1714,282 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// A file whose audio cannot be read used to stop the whole run with a + /// bare decoder message. It is now a warning naming the file, and the + /// other files still synchronize. + #[test] + fn unreadable_audio_is_reported_and_the_rest_still_synchronizes() { + let dir = fixture_dir("unreadable-audio"); + let body = noise(12 * 44100, 0x5EED); + let a = write_delayed_wav(&dir, "a.wav", &body, 0.0); + let b = write_delayed_wav(&dir, "b.wav", &body, 1.024); + let Some(broken) = crate::sym::test_media::aac_noise(&dir, "broken.mp4", 9) else { + return; + }; + // The movie header survives; the audio it describes does not, so + // neither Symphonia nor FFmpeg can decode it. + let data = std::fs::read(&broken).unwrap(); + let mdat = data.windows(4).position(|w| w == b"mdat").unwrap(); + std::fs::write(&broken, &data[..mdat + 64]).unwrap(); + let pipe = Pipeline::new_in(BackendKind::Portable, Some(dir.join(".cache"))); + let result = pipe + .synchronize( + &[ + PipelineInput::Media(a), + PipelineInput::Media(b), + PipelineInput::Media(broken.clone()), + ], + &[], + &PipelineOptions::default(), + None, + &std::sync::atomic::AtomicBool::new(false), + ) + .expect("one unreadable file does not stop synchronization"); + let warning = result + .project + .warnings + .iter() + .find(|warning| warning.url == broken) + .expect("the unreadable file is named"); + assert!( + warning.message.contains("left out of synchronization"), + "{}", + warning.message + ); + let broken_id = &result + .project + .clips + .iter() + .find(|clip| clip.url == broken) + .unwrap() + .id; + assert!(result.unmatched.contains(broken_id)); + assert_eq!(result.matches.len(), 1, "the readable files match"); + let _ = std::fs::remove_dir_all(&dir); + } + + /// The portable backend, except that analysing chosen files and sources + /// fails like a decoder part way through the audio, after passing some + /// samples on. + struct FaultyAnalysis { + fails: fn(&Path, AudioAnalysisSource) -> Option, + /// Whether refinement windows of those sources fail too. + windows_fail: bool, + } + + impl MediaBackend for FaultyAnalysis { + fn kind(&self) -> BackendKind { + BackendKind::Portable + } + + fn inspect(&self, path: &Path) -> Result { + crate::portable::PortableBackend.inspect(path) + } + + fn decode_mono_8k( + &self, + path: &Path, + source: AudioAnalysisSource, + consume: &mut dyn FnMut(&[f32]) -> Result<(), crate::DecodeError>, + ) -> Result<(), crate::DecodeError> { + if let Some(error) = (self.fails)(path, source) { + consume(&noise(8000, 99))?; + return Err(error); + } + crate::portable::PortableBackend.decode_mono_8k(path, source, consume) + } + + fn decode_window_16k( + &self, + path: &Path, + start: f64, + duration: f64, + source: AudioAnalysisSource, + ) -> Result<(f64, Vec), crate::DecodeError> { + match (self.fails)(path, source) { + Some(error) if self.windows_fail => Err(error), + _ => crate::portable::PortableBackend + .decode_window_16k(path, start, duration, source), + } + } + + fn decode_native( + &self, + path: &Path, + stream: usize, + range: Option<(f64, Option)>, + consume: &mut dyn FnMut(crate::backend::NativeBlock) -> Result<(), crate::DecodeError>, + ) -> Result { + crate::portable::PortableBackend.decode_native(path, stream, range, consume) + } + + fn decode_native_i32( + &self, + path: &Path, + stream: usize, + range: Option<(f64, Option)>, + consume: &mut dyn FnMut( + crate::backend::NativeBlockI32, + ) -> Result<(), crate::DecodeError>, + ) -> Result { + crate::portable::PortableBackend.decode_native_i32(path, stream, range, consume) + } + } + + fn faulty_pipeline(dir: &Path, backend: FaultyAnalysis) -> Pipeline { + Pipeline { + backend: Box::new(backend), + cache: FingerprintCache::with_backend(Some(dir.join(".cache")), "faulty-test"), + } + } + + fn named(path: &Path, name: &str) -> bool { + path.file_name().is_some_and(|file| file == name) + } + + /// When the in-process decoder fails, FFmpeg decodes the file from the + /// start into fresh state, for one stream and for a mix of all streams. + /// Only one of the two files fails: samples it passed on before failing + /// would shift it against the other. + #[test] + fn in_process_decode_failures_are_retried_with_ffmpeg() { + if crate::ff::ffmpeg_bin().is_none() { + return; + } + let dir = fixture_dir("ffmpeg-retry"); + let body = noise(12 * 44100, 0xFA11); + write_delayed_wav(&dir, "a.wav", &body, 0.0); + write_delayed_wav(&dir, "b.wav", &body, 1.024); + for audio_source in [ + AudioAnalysisSource::Automatic, + AudioAnalysisSource::AllMixed, + ] { + let _ = std::fs::remove_dir_all(dir.join(".cache")); + let pipe = faulty_pipeline( + &dir, + FaultyAnalysis { + fails: |path, _| { + named(path, "a.wav").then(|| { + crate::DecodeError::Symphonia("unexpected end of bitstream".into()) + }) + }, + windows_fail: false, + }, + ); + let options = PipelineOptions { + audio_source, + ..Default::default() + }; + let result = pipe + .synchronize( + &[PipelineInput::Media(dir.clone())], + &[], + &options, + None, + &std::sync::atomic::AtomicBool::new(false), + ) + .unwrap(); + assert!( + result.project.warnings.is_empty(), + "{audio_source:?}: {:?}", + result.project.warnings + ); + assert_eq!(result.matches.len(), 1, "{audio_source:?}"); + let offset = result.matches[0].offset.as_seconds().abs(); + assert!((offset - 1.024).abs() < 0.001, "{audio_source:?}: {offset}"); + } + let _ = std::fs::remove_dir_all(&dir); + } + + /// Automatic analysis of a file with several streams keeps the streams + /// it can read. Refinement must decode the stream that was kept, not the + /// first stream again. + #[test] + fn refinement_uses_the_stream_that_could_be_read() { + let dir = fixture_dir("surviving-stream"); + let body = noise(12 * 44100, 0x57E4); + write_delayed_wav(&dir, "a.wav", &body, 0.0); + let delayed = write_delayed_wav(&dir, "b.wav", &body, 1.024); + let Some(_) = ffmpeg_clip( + &dir, + "two-streams.mov", + &[ + "-i", + &delayed.to_string_lossy(), + "-i", + &delayed.to_string_lossy(), + "-map", + "0:a", + "-map", + "1:a", + "-c:a", + "pcm_s16le", + ], + ) else { + return; + }; + std::fs::remove_file(&delayed).unwrap(); + let pipe = faulty_pipeline( + &dir, + FaultyAnalysis { + fails: |path, source| { + (named(path, "two-streams.mov") && source.stream_index() == 0) + .then(|| crate::DecodeError::Incomplete("stream 0 is unreadable".into())) + }, + windows_fail: true, + }, + ); + let result = pipe + .synchronize( + &[PipelineInput::Media(dir.clone())], + &[], + &PipelineOptions::default(), + None, + &std::sync::atomic::AtomicBool::new(false), + ) + .unwrap(); + assert!( + result.project.warnings.is_empty(), + "{:?}", + result.project.warnings + ); + assert_eq!(result.matches.len(), 1); + assert!((result.matches[0].offset.as_seconds().abs() - 1.024).abs() < 0.001); + let _ = std::fs::remove_dir_all(&dir); + } + + /// Clips finish analysis in any order; the completed count the progress + /// bar shows must still only grow. The listener holds odd counts for a + /// moment: if two workers could publish at once, the next count would + /// overtake it. + #[test] + fn analysis_progress_never_goes_backwards() { + let dir = fixture_dir("analysis-progress"); + for index in 0..16 { + write_mono_wav(&dir, &format!("{index:03}.wav"), &noise(4_800, index)); + } + let pipe = Pipeline::new_in(BackendKind::Portable, Some(dir.join(".cache"))); + let counts = std::sync::Mutex::new(Vec::new()); + let progress = |event: PipelineProgress| { + if event.phase == Phase::Fingerprint && event.current.is_some() { + if event.completed % 2 == 1 { + std::thread::sleep(std::time::Duration::from_millis(20)); + } + counts.lock().unwrap().push(event.completed); + } + }; + pipe.synchronize( + &[PipelineInput::Media(dir.clone())], + &[], + &PipelineOptions::default(), + Some(&progress), + &std::sync::atomic::AtomicBool::new(false), + ) + .unwrap(); + assert_eq!(counts.into_inner().unwrap(), (1..=16).collect::>()); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn import_and_relink_warnings_reach_project() { // draft.warnings (import findings + relink notes) used to die in diff --git a/crates/align-decode/src/sym.rs b/crates/align-decode/src/sym.rs index 2a6a8f3..7f74f76 100644 --- a/crates/align-decode/src/sym.rs +++ b/crates/align-decode/src/sym.rs @@ -1,14 +1,16 @@ -//! Symphonia tier of the portable backend: probe + streaming decode for -//! everything Symphonia demuxes (WAV/AIFF/MP3/M4A-ALAC-AAC/OGG/FLAC/MKV…). -//! Containers it cannot handle (MTS/MXF/R3D, odd MP4s) fall through to the -//! FFmpeg pipe in [`crate::ff`]. +//! Symphonia tier of the portable backend: streaming decode of WAV, AIFF, +//! MP3, ADTS, and MOV/MP4/M4A (AAC, ALAC, PCM). Other containers +//! (MTS/MXF/R3D) and MP4s Symphonia rejects fall through to the FFmpeg pipe +//! in [`crate::ff`]. //! //! Audio-only: only tracks for which a decoder constructs are opened — //! video tracks never decode (Symphonia 0.5 has no video decoders, so //! `make()` failing is the filter). Sample flow per packet: //! planar `AudioBuffer` → selected discrete channel → [`crate::mono`]. //! Full files are never resident: packets stream through a 128 KiB -//! resample block. +//! resample block. A packet the decoder rejects (FFmpeg skips such packets +//! too) becomes silence of the same duration, so one damaged packet neither +//! stops the file nor shifts the audio after it. use std::fs::File; use std::io::{Read, Seek, SeekFrom}; @@ -28,33 +30,73 @@ use crate::backend::AudioStreamProbe; use crate::pcm::PcmTrack; use align_core::AudioAnalysisSource; +fn symphonia_error(error: SymphoniaError) -> DecodeError { + DecodeError::Symphonia(error.to_string()) +} + +fn extension(path: &Path) -> String { + path.extension() + .and_then(|e| e.to_str()) + .unwrap_or_default() + .to_ascii_lowercase() +} + +/// QuickTime and ISO media: Symphonia's MP4 reader, plus the movie header for +/// what that reader leaves out (AAC channel counts, edit lists). +fn is_iso_media(path: &Path) -> bool { + matches!(extension(path).as_str(), "mov" | "mp4" | "m4a" | "m4v") +} + +/// The container reader the file extension names. Symphonia 0.5 ignores the +/// probe hint and scans up to 1 MiB of any file for an MP3 or ADTS sync word; +/// inside MXF or MTS video it finds one and reports noise as audio, and when +/// the reader then rejects the match it may search to the end of the file. +/// Only elementary MP3 and ADTS streams, which can start with an ID3 tag, +/// are probed; every other container goes to FFmpeg. +fn container_reader(path: &Path, file: File) -> Result, DecodeError> { + use symphonia::default::formats::{AiffReader, IsoMp4Reader, WavReader}; + let extension = extension(path); + let mss = MediaSourceStream::new(Box::new(file), Default::default()); + let options = FormatOptions::default(); + if is_iso_media(path) { + return Ok(Box::new( + IsoMp4Reader::try_new(mss, &options).map_err(symphonia_error)?, + )); + } + Ok(match extension.as_str() { + "wav" | "wave" | "bwf" => { + Box::new(WavReader::try_new(mss, &options).map_err(symphonia_error)?) + } + "aif" | "aiff" | "aifc" => { + Box::new(AiffReader::try_new(mss, &options).map_err(symphonia_error)?) + } + "mp3" | "aac" => { + symphonia::default::get_probe() + .format(&Hint::new(), mss, &options, &MetadataOptions::default()) + .map_err(symphonia_error)? + .format + } + _ => { + return Err(DecodeError::Symphonia(format!( + "{extension:?} files are decoded by FFmpeg" + ))); + } + }) +} + /// Open + demux, returning the reader and the audio-track shortlist. /// A track counts as audio iff a decoder constructs for it (video tracks /// fail here and are never touched). fn open(path: &Path) -> Result<(Box, Vec), DecodeError> { let mut file = File::open(path).map_err(DecodeError::Io)?; - // Symphonia 0.5 recognizes RIFF WAVE but not RF64/BW64; its generic - // signature search would scan a multi-gigabyte file. Those files are + // Symphonia 0.5 recognizes RIFF WAVE but not RF64/BW64. Those files are // read by `crate::pcm` instead. let mut signature = [0u8; 4]; if file.read_exact(&mut signature).is_ok() && matches!(&signature, b"RF64" | b"BW64") { return Err(DecodeError::Symphonia("RF64/BW64 is read directly".into())); } file.seek(SeekFrom::Start(0)).map_err(DecodeError::Io)?; - let mss = MediaSourceStream::new(Box::new(file), Default::default()); - let mut hint = Hint::new(); - if let Some(ext) = path.extension().and_then(|e| e.to_str()) { - hint.with_extension(ext); - } - let probed = symphonia::default::get_probe() - .format( - &hint, - mss, - &FormatOptions::default(), - &MetadataOptions::default(), - ) - .map_err(|e| DecodeError::Symphonia(e.to_string()))?; - let format = probed.format; + let format = container_reader(path, file)?; let mut audio = Vec::new(); for track in format.tracks() { if track.codec_params.codec == CODEC_TYPE_NULL { @@ -180,6 +222,83 @@ struct Stream { scratch_spec: Option, scratch_cap: usize, scratch: AudioBuffer, + /// Whether a decoded packet has been compared with the length its + /// container records. PCM needs no comparison. + timing_checked: bool, + /// Shape of the last decoded packet: the frames of a damaged packet whose + /// container records no duration, and the channels of its silence. + last_frames: usize, + last_channels: usize, +} + +/// Outcome of reading the next packet of the selected track. +enum Step { + /// Read the samples with `stream.decoder.last_decoded()`. + Decoded, + /// A damaged packet: this many frames of silence keep later audio at its + /// recorded time. + Silence(usize), + End, +} + +/// Frames covered by a packet duration in the track's time base. +fn packet_frames(time_base: TimeBase, sample_rate: u32, duration: u64) -> usize { + let time = time_base.calc_time(duration); + ((time.seconds as f64 + time.frac) * sample_rate as f64).round() as usize +} + +fn decode_next(stream: &mut Stream) -> Result { + loop { + let packet = match stream.format.next_packet() { + Ok(p) => p, + Err(SymphoniaError::IoError(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => { + return Ok(Step::End); + } + Err(e) => return Err(symphonia_error(e)), + }; + if packet.track_id() != stream.track_id { + continue; + } + let recorded = (packet.dur() > 0) + .then(|| packet_frames(stream.time_base, stream.sample_rate, packet.dur())); + // Decoders read only the packet, so every error here describes this + // packet: a truncated or corrupted bitstream ("unexpected end of + // bitstream"), invalid data, or an element the decoder lacks. The + // packet becomes silence of its recorded length. + match stream.decoder.decode(&packet) { + Ok(decoded) => { + let frames = decoded.frames(); + stream.last_frames = frames; + stream.last_channels = decoded.spec().channels.count(); + // A decoder that yields far fewer samples than the packet + // spans is decoding another stream than the file holds: + // HE-AAC gives half. Final packets may differ slightly either + // way (PCM bookkeeping, AAC padding), which is fine. + if let (false, Some(recorded)) = (stream.timing_checked, recorded) { + if frames * 4 < recorded * 3 { + return Err(DecodeError::Symphonia(format!( + "decoded {frames} of the {recorded} samples a packet records" + ))); + } + stream.timing_checked = true; + } + return Ok(Step::Decoded); + } + Err(SymphoniaError::ResetRequired) => { + stream.decoder.reset(); + continue; + } + Err(_) => return Ok(Step::Silence(recorded.unwrap_or(stream.last_frames))), + } + } +} + +fn silence(stream: &Stream, frames: usize, zero: T) -> Vec> { + let channels = match stream.last_channels { + 0 => stream.channels, + decoded => decoded, + }; + vec![vec![zero; frames]; channels] } fn open_stream(path: &Path, stream_index: usize) -> Result { @@ -198,7 +317,7 @@ fn open_stream(path: &Path, stream_index: usize) -> Result // Symphonia's MP4 reader leaves the channel count of AAC tracks to the // decoder and ignores edit lists. The movie header (track ids are movie // track indices) supplies both, keeping camera AAC in-process. - let movie_track = crate::isobmff::is_candidate(path) + let movie_track = is_iso_media(path) .then(|| crate::isobmff::read(path)) .flatten() .and_then(|movie| movie.tracks.into_iter().nth(track.id as usize)); @@ -233,19 +352,39 @@ fn open_stream(path: &Path, stream_index: usize) -> Result scratch_spec: None, scratch_cap: 0, scratch: AudioBuffer::unused(), + timing_checked: is_pcm, + last_frames: 0, + last_channels: 0, }) } -/// Header-only check: can Symphonia demux this stream? Used to pick the -/// decode tier *before* any sample flows, so the FFmpeg fallback never -/// double-emits after a partial Symphonia stream. +/// Can Symphonia decode this stream? Used to pick the decode tier *before* +/// any sample flows, so the FFmpeg fallback never double-emits after a +/// partial Symphonia stream. Opening it is not enough: the first decodable +/// packet must also yield the samples its container records. HE-AAC does +/// not, since Symphonia decodes only its AAC core (half the samples), so +/// FFmpeg decodes it. pub fn can_decode(path: &Path, stream_index: usize) -> bool { - PcmTrack::open(path, stream_index).is_some() || open_stream(path, stream_index).is_ok() + if PcmTrack::open(path, stream_index).is_some() { + return true; + } + let Ok(mut stream) = open_stream(path, stream_index) else { + return false; + }; + // A few damaged packets at the start do not decide the tier. + for _ in 0..16 { + match decode_next(&mut stream) { + Ok(Step::Decoded) => return true, + Ok(Step::Silence(_)) => {} + Ok(Step::End) | Err(_) => return false, + } + } + false } /// Stream a whole file to 8 kHz mono through `consume` (fingerprint path). -/// Call [`can_decode`] first: mid-stream errors return `Err` *after* partial -/// output reached `consume`, so the tier must be fixed upfront. +/// Call [`can_decode`] first: an error can return *after* partial output +/// reached `consume`, so the tier must be fixed upfront. pub fn decode_mono_8k( path: &Path, source: AudioAnalysisSource, @@ -362,37 +501,24 @@ pub fn decode_window( /// allocation-free. fn decode_packet_planes(stream: &mut Stream) -> Result>>, DecodeError> { use symphonia::core::audio::Signal; - loop { - let packet = match stream.format.next_packet() { - Ok(p) => p, - Err(SymphoniaError::IoError(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => { - return Ok(None); + match decode_next(stream)? { + Step::End => Ok(None), + Step::Silence(frames) => Ok(Some(silence(stream, frames, 0.0))), + Step::Decoded => { + let decoded = stream.decoder.last_decoded(); + let spec = *decoded.spec(); + if stream.scratch_spec != Some(spec) || stream.scratch_cap != decoded.capacity() { + stream.scratch = decoded.make_equivalent::(); + stream.scratch_spec = Some(spec); + stream.scratch_cap = decoded.capacity(); } - Err(e) => return Err(DecodeError::Symphonia(e.to_string())), - }; - if packet.track_id() != stream.track_id { - continue; + decoded.convert(&mut stream.scratch); + Ok(Some( + (0..spec.channels.count()) + .map(|ch| stream.scratch.chan(ch).to_vec()) + .collect(), + )) } - let decoded = match stream.decoder.decode(&packet) { - Ok(d) => d, - Err(SymphoniaError::ResetRequired) => { - stream.decoder.reset(); - continue; - } - Err(SymphoniaError::DecodeError(_)) => continue, - Err(e) => return Err(DecodeError::Symphonia(e.to_string())), - }; - let spec = *decoded.spec(); - if stream.scratch_spec != Some(spec) || stream.scratch_cap != decoded.capacity() { - stream.scratch = decoded.make_equivalent::(); - stream.scratch_spec = Some(spec); - stream.scratch_cap = decoded.capacity(); - } - decoded.convert(&mut stream.scratch); - let planes: Vec> = (0..spec.channels.count()) - .map(|ch| stream.scratch.chan(ch).to_vec()) - .collect(); - return Ok(Some(planes)); } } @@ -523,50 +649,37 @@ pub fn decode_native_i32( if limit_frames.is_some_and(|lim| emitted >= lim) { break; } - let packet = match stream.format.next_packet() { - Ok(p) => p, - Err(SymphoniaError::IoError(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => { - break; - } - Err(e) => return Err(DecodeError::Symphonia(e.to_string())), - }; - if packet.track_id() != stream.track_id { - continue; - } - let decoded = match stream.decoder.decode(&packet) { - Ok(d) => d, - Err(SymphoniaError::ResetRequired) => { - stream.decoder.reset(); - continue; - } - Err(SymphoniaError::DecodeError(_)) => continue, - Err(e) => return Err(DecodeError::Symphonia(e.to_string())), - }; - let spec = *decoded.spec(); - let nch = spec.channels.count(); - let mut planes: Vec> = match &decoded { - AudioBufferRef::S32(buf) => (0..nch).map(|ch| buf.chan(ch).to_vec()).collect(), - _ => { - if scratch_f32.spec() != decoded.spec() - || scratch_f32.capacity() != decoded.capacity() - { - scratch_f32 = decoded.make_equivalent::(); - } - decoded.convert(&mut scratch_f32); - (0..nch) - .map(|ch| { - scratch_f32 - .chan(ch) - .iter() - .map(|&v| { - (v as f64 * 2_147_483_648.0) - .round() - .clamp(-2_147_483_648.0, 2_147_483_647.0) - as i32 + let mut planes: Vec> = match decode_next(&mut stream)? { + Step::End => break, + Step::Silence(frames) => silence(&stream, frames, 0), + Step::Decoded => { + let decoded = stream.decoder.last_decoded(); + let nch = decoded.spec().channels.count(); + match &decoded { + AudioBufferRef::S32(buf) => (0..nch).map(|ch| buf.chan(ch).to_vec()).collect(), + _ => { + if scratch_f32.spec() != decoded.spec() + || scratch_f32.capacity() != decoded.capacity() + { + scratch_f32 = decoded.make_equivalent::(); + } + decoded.convert(&mut scratch_f32); + (0..nch) + .map(|ch| { + scratch_f32 + .chan(ch) + .iter() + .map(|&v| { + (v as f64 * 2_147_483_648.0) + .round() + .clamp(-2_147_483_648.0, 2_147_483_647.0) + as i32 + }) + .collect() }) .collect() - }) - .collect() + } + } } }; trim_native_packet( @@ -646,27 +759,25 @@ mod edit_list_tests { samples.iter().position(|s| s.abs() > 0.3) } - /// AAC in MP4 carries encoder priming in the edit list. In-process - /// decoding must place audio exactly where FFmpeg (which applies the - /// edit list) does, both from the head and after a seek. + /// AAC in MP4 and M4A carries encoder priming in the edit list. + /// In-process decoding must place audio exactly where FFmpeg (which + /// applies the edit list) does, both from the head and after a seek. #[test] fn aac_priming_matches_ffmpeg_timing() { - let Some(ffmpeg) = crate::ff::ffmpeg_bin() else { - return; - }; + for name in ["camera.mp4", "voice.m4a"] { + aac_priming_case(name); + } + } + + fn aac_priming_case(name: &str) { let dir = tempfile::tempdir().unwrap(); - let clip = dir.path().join("camera.mp4"); - let ok = std::process::Command::new(ffmpeg) - .args(["-v", "error", "-y", "-f", "lavfi", "-i"]) - .arg("aevalsrc='if(between(t,1.5,1.502),0.9,0)':s=48000:d=4") - .args(["-ac", "2", "-c:a", "aac"]) - .arg(&clip) - .status() - .is_ok_and(|s| s.success()); - if !ok { - eprintln!("SKIP: no AAC encoder"); + let Some(clip) = super::test_media::aac_clip( + dir.path(), + name, + "aevalsrc='if(between(t,1.5,1.502),0.9,0)':s=48000:d=4", + ) else { return; - } + }; assert!(can_decode(&clip, 0), "AAC must decode in-process"); let reference = { let mut out = Vec::new(); @@ -699,3 +810,184 @@ mod edit_list_tests { assert!((click - 1500.0).abs() < 0.3, "window click at {click} ms"); } } + +/// AAC media made with FFmpeg, and damaged packets made by editing its +/// sample size table. Shared with the pipeline tests. +#[cfg(test)] +pub(crate) mod test_media { + use std::path::{Path, PathBuf}; + + /// Stereo AAC in MP4 from an FFmpeg `lavfi` source, or `None` without an + /// AAC encoder. + pub(crate) fn aac_clip(dir: &Path, name: &str, source: &str) -> Option { + let ffmpeg = crate::ff::ffmpeg_bin()?; + let clip = dir.join(name); + let ok = std::process::Command::new(ffmpeg) + .args(["-v", "error", "-y", "-f", "lavfi", "-i", source]) + .args(["-movflags", "+faststart"]) + // Without noise substitution, whose generator state would let + // one skipped packet change every later packet. + .args(["-ac", "2", "-c:a", "aac", "-b:a", "192k", "-aac_pns", "0"]) + .arg(&clip) + .status() + .is_ok_and(|s| s.success()); + if !ok { + eprintln!("SKIP: no AAC encoder"); + } + ok.then_some(clip) + } + + /// Four seconds of pink noise. + pub(crate) fn aac_noise(dir: &Path, name: &str, seed: u32) -> Option { + aac_clip( + dir, + name, + &format!("anoisesrc=d=4:c=pink:r=48000:a=0.4:seed={seed}"), + ) + } + + /// Rewrites the `stsz` entries of a single-track file. + pub(crate) fn edit_sample_sizes(path: &Path, edit: impl FnOnce(&mut [u32])) { + let mut data = std::fs::read(path).unwrap(); + let at = data.windows(4).position(|w| w == b"stsz").unwrap(); + let word = |data: &[u8], i: usize| u32::from_be_bytes(data[i..i + 4].try_into().unwrap()); + assert_eq!(word(&data, at + 8), 0, "variable sample sizes"); + let count = word(&data, at + 12) as usize; + let first = at + 16; + let mut sizes: Vec = (0..count).map(|i| word(&data, first + 4 * i)).collect(); + edit(&mut sizes); + for (i, size) in sizes.iter().enumerate() { + data[first + 4 * i..first + 4 * i + 4].copy_from_slice(&size.to_be_bytes()); + } + std::fs::write(path, data).unwrap(); + } + + /// Doubles every `stts` duration of a single-track file. + pub(crate) fn double_packet_durations(path: &Path) { + let mut data = std::fs::read(path).unwrap(); + let at = data.windows(4).position(|w| w == b"stts").unwrap(); + let word = |data: &[u8], i: usize| u32::from_be_bytes(data[i..i + 4].try_into().unwrap()); + for entry in 0..word(&data, at + 8) as usize { + let delta = at + 16 + 8 * entry; + let doubled = word(&data, delta) * 2; + data[delta..delta + 4].copy_from_slice(&doubled.to_be_bytes()); + } + std::fs::write(path, data).unwrap(); + } + + /// Cuts packet `k` in half and gives the cut bytes to packet `k + 1`: + /// both become unreadable, later packets keep their file offsets. + pub(crate) fn truncate_packet(path: &Path, k: usize) { + edit_sample_sizes(path, |sizes| { + let cut = sizes[k] / 2; + sizes[k] -= cut; + sizes[k + 1] += cut; + }); + } +} + +#[cfg(test)] +mod damaged_packet_tests { + use super::test_media::{aac_noise, double_packet_durations, truncate_packet}; + use super::*; + + fn decode_all(path: &Path) -> Result, DecodeError> { + let mut samples = Vec::new(); + decode_native(path, 0, 0.0, None, &mut |block| { + samples.extend_from_slice(&block.frames[0]); + Ok(()) + })?; + Ok(samples) + } + + /// A truncated AAC packet makes Symphonia's decoder report "unexpected + /// end of bitstream". It used to stop the whole file; it is silence of + /// the packet's duration now, so later audio keeps its time. + #[test] + fn damaged_packet_becomes_silence_of_its_duration() { + let dir = tempfile::tempdir().unwrap(); + let Some(clip) = aac_noise(dir.path(), "camera.mp4", 3) else { + return; + }; + let clean = decode_all(&clip).unwrap(); + let k = 80; + truncate_packet(&clip, k); + let damaged = decode_all(&clip).expect("damaged packets are skipped"); + assert_eq!(damaged.len(), clean.len(), "timeline length is kept"); + // Packets hold 1024 frames; the edit list trims 1024 priming frames. + // The decoder's overlap mixes each packet with its predecessor. + let before = (k - 2) * 1024; + let after = (k + 4) * 1024; + assert_eq!(damaged[..before], clean[..before]); + assert_eq!(damaged[after..], clean[after..]); + assert!( + damaged[(k - 1) * 1024..k * 1024].iter().all(|s| *s == 0.0), + "the truncated packet is silent" + ); + } + + /// HE-AAC decodes to half the samples each packet records, because + /// Symphonia has no SBR. Doubling the packet durations of AAC-LC gives + /// Symphonia the same mismatch without an HE-AAC encoder. + #[test] + fn streams_decoded_shorter_than_their_packets_go_to_ffmpeg() { + let dir = tempfile::tempdir().unwrap(); + let Some(clip) = aac_noise(dir.path(), "camera.mp4", 4) else { + return; + }; + assert!(can_decode(&clip, 0)); + double_packet_durations(&clip); + assert!(!can_decode(&clip, 0), "FFmpeg decodes it"); + let error = decode_window(&clip, 1.0, 1.0, AudioAnalysisSource::default()).unwrap_err(); + assert!( + error.to_string().contains("samples a packet records"), + "{error}" + ); + } + + /// Symphonia's AIFF reader can record a couple more frames for the final + /// packet than it decodes. That is bookkeeping, not a wrong decoder, so + /// exporting the last milliseconds must still work in process. + #[test] + fn short_final_pcm_packets_still_decode() { + let Some(ffmpeg) = crate::ff::ffmpeg_bin() else { + return; + }; + let dir = tempfile::tempdir().unwrap(); + let aiff = dir.path().join("recorder.aiff"); + let ok = std::process::Command::new(ffmpeg) + .args(["-v", "error", "-y", "-f", "lavfi", "-i"]) + .arg("anoisesrc=d=1:r=48000:a=0.3") + .args(["-ac", "2", "-c:a", "pcm_s16be"]) + .arg(&aiff) + .status() + .is_ok_and(|s| s.success()); + if !ok { + return; + } + let mut frames = 0; + decode_native(&aiff, 0, 0.995, None, &mut |block| { + frames += block.frames[0].len(); + Ok(()) + }) + .expect("the final packet decodes"); + assert_eq!(frames, 240); + } + + /// Symphonia's probe ignores the extension and accepts MP3 sync words + /// anywhere in the first megabyte, as compressed video essence contains. + /// Only the containers the extension names are opened in process. + #[test] + fn mp3_frames_inside_other_containers_are_not_decoded_as_audio() { + let dir = tempfile::tempdir().unwrap(); + // MPEG-1 Layer III, 128 kbit/s, 44.1 kHz: 417-byte silent frames. + let frame = [[0xFF, 0xFB, 0x90, 0x00].as_slice(), &[0; 413]].concat(); + let frames = frame.repeat(100); + let mp3 = dir.path().join("tone.mp3"); + std::fs::write(&mp3, &frames).unwrap(); + assert!(can_decode(&mp3, 0), "an MP3 file decodes in process"); + let mxf = dir.path().join("A001C001.MXF"); + std::fs::write(&mxf, [vec![0; 1000], frames].concat()).unwrap(); + assert!(!can_decode(&mxf, 0), "MXF goes to FFmpeg"); + } +} diff --git a/docs/formats.md b/docs/formats.md index 6e570fc..ddec8a5 100644 --- a/docs/formats.md +++ b/docs/formats.md @@ -4,12 +4,18 @@ Align discovers WAV, AIF/AIFF, M4A, MP3, M4V, MOV, MP4, MTS, MXF, and R3D paths. An accepted extension does not guarantee that the installed decoder -supports the file's codec. Unsupported or unreadable media produces a warning or -an error; inspect the result before exporting. +supports the file's codec. When the built-in decoder fails on a file's audio +during analysis, FFmpeg decodes it again where FFmpeg is available. A file is +named in a warning and left out of synchronization only when its media or +audio still cannot be read; the other files are still synchronized. Inspect +the warnings before exporting. On macOS, the default backend uses AVFoundation. The portable backend reads MOV/MP4/M4V headers natively, decodes WAV, AIFF, MP3, AAC, ALAC, and PCM with -Symphonia, and uses FFmpeg for other containers (MTS, MXF, R3D) and codecs. +Symphonia, and uses FFmpeg for other containers (MTS, MXF, R3D) and for +streams Symphonia decodes incorrectly, such as HE-AAC. When Symphonia decodes, +a damaged audio packet becomes silence of the same length, so the rest of the +file keeps its timing. Set `ALIGN_BACKEND=portable` to select it on macOS. Video metadata is inspected for timing; synchronization does not decode video frames.