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.