alpha2_native/
recovery.rs

1//! Bounded recovery shared by software and hardware-backed native decoders.
2use crate::convert::adapter_error;
3use alpha2_media::{CancelToken, DecodeError};
4use rsmpeg::{error::RsmpegError, ffi};
5use std::time::{Duration, Instant};
6
7/// Damaged input handling. Initialization and resource errors are always fatal.
8#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
9pub enum DecodePolicy {
10    /// Fail on the first reported corruption.
11    #[default]
12    Strict,
13    /// Skip damaged input, retaining diagnostics; fail when recovery is unbounded.
14    Tolerant,
15}
16
17impl DecodePolicy {
18    /// The policy's name, as `ALPHA2_DECODE_POLICY` and the `alpha2` CLI's
19    /// `--decode-policy` spell it: `strict` or `tolerant`.
20    #[must_use]
21    pub const fn as_str(self) -> &'static str {
22        match self {
23            Self::Strict => "strict",
24            Self::Tolerant => "tolerant",
25        }
26    }
27
28    /// The policy `value` names ([`Self::as_str`]); any other value is none.
29    #[must_use]
30    pub fn parse(value: &str) -> Option<Self> {
31        [Self::Strict, Self::Tolerant]
32            .into_iter()
33            .find(|policy| policy.as_str() == value)
34    }
35
36    /// The policy `ALPHA2_DECODE_POLICY` selects: Strict when it is unset.
37    ///
38    /// # Errors
39    /// Any other value, including non-Unicode, is an error rather than a
40    /// silent fallback.
41    pub fn from_env() -> Result<Self, String> {
42        match std::env::var("ALPHA2_DECODE_POLICY") {
43            Err(std::env::VarError::NotPresent) => Some(Self::Strict),
44            Ok(value) => Self::parse(&value),
45            Err(std::env::VarError::NotUnicode(_)) => None,
46        }
47        .ok_or_else(|| "ALPHA2_DECODE_POLICY must be strict or tolerant".to_owned())
48    }
49}
50
51/// What decoders reported, counted: the summary a caller gives of the damage
52/// Tolerant decoding skipped and the silence it inserted. Samples are at the
53/// reporting decoder's output rate.
54#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
55pub struct DecodeDiagnosticCounts {
56    /// Corruption observations ([`DecodeDiagnostics::corrupt_packets`]): one
57    /// damaged packet can be observed at more than one stage.
58    pub corrupt_packets: u64,
59    /// Runs of dropped video between healthy frames, an unrecovered tail included.
60    pub video_gaps: u64,
61    /// Audio samples decoded as silence in place of damaged input.
62    pub concealed_audio_samples: u64,
63    /// Audio samples decoded as silence where a clean stream's timestamps jump.
64    pub timestamp_gap_samples: u64,
65    /// Audio samples dropped because their timestamps repeated samples already decoded.
66    pub dropped_audio_samples: u64,
67}
68
69impl DecodeDiagnosticCounts {
70    /// Counts one report.
71    pub fn record(&mut self, diagnostic: &DecodeDiagnostic) {
72        let samples = |start: i64, end: i64| u64::try_from(end.saturating_sub(start)).unwrap_or(0);
73        match diagnostic {
74            DecodeDiagnostic::Corruption(_) => {
75                self.corrupt_packets = self.corrupt_packets.saturating_add(1);
76            }
77            DecodeDiagnostic::VideoGap(_) => self.video_gaps = self.video_gaps.saturating_add(1),
78            DecodeDiagnostic::AudioGap(gap) => {
79                let count = samples(gap.start_sample, gap.end_sample);
80                let total = if gap.reason == "timestamp gap" {
81                    &mut self.timestamp_gap_samples
82                } else {
83                    &mut self.concealed_audio_samples
84                };
85                *total = total.saturating_add(count);
86            }
87            DecodeDiagnostic::AudioOverlap(overlap) => {
88                self.dropped_audio_samples = self
89                    .dropped_audio_samples
90                    .saturating_add(samples(overlap.start_sample, overlap.end_sample));
91            }
92        }
93    }
94
95    /// Whether nothing was reported.
96    #[must_use]
97    pub fn is_empty(&self) -> bool {
98        *self == Self::default()
99    }
100}
101
102/// One observed corruption. Timestamps use the selected stream's time base.
103#[derive(Clone, Debug, PartialEq, Eq)]
104pub struct CorruptPacket {
105    /// Pipeline stage reporting damage (may be a frame or demux error).
106    pub stage: &'static str,
107    /// Best available timestamp, absent when `FFmpeg` cannot identify it.
108    pub pts: Option<i64>,
109    /// Numerator and denominator of the timestamp time base.
110    pub time_base: (i32, i32),
111}
112
113/// The most audio gaps and overlaps one decoder records.
114const MAX_AUDIO_RECORDS: usize = 1024;
115
116/// An interval decoded as silence, at the decoder's output sample rate:
117/// where damaged input was dropped, or where the stream's timestamps jump.
118#[derive(Clone, Debug, PartialEq, Eq)]
119pub struct ConcealedAudioGap {
120    /// Inclusive source sample position.
121    pub start_sample: i64,
122    /// Exclusive source sample position.
123    pub end_sample: i64,
124    /// Output samples per second.
125    pub sample_rate: u32,
126    /// Why silence was inserted: `"corrupt input"` or `"timestamp gap"`.
127    pub reason: &'static str,
128}
129
130/// Samples whose timestamps put them over samples already decoded, dropped,
131/// at the decoder's output sample rate.
132#[derive(Clone, Debug, PartialEq, Eq)]
133pub struct DroppedAudioOverlap {
134    /// Inclusive source sample position.
135    pub start_sample: i64,
136    /// Exclusive source sample position.
137    pub end_sample: i64,
138    /// Output samples per second.
139    pub sample_rate: u32,
140}
141
142/// Healthy video boundaries surrounding dropped data. Unknown endpoints remain absent.
143#[derive(Clone, Debug, PartialEq, Eq)]
144pub struct VideoRecoveryGap {
145    /// Last healthy frame before damage; may be held by a containing-frame reader.
146    pub previous_pts: Option<i64>,
147    /// First healthy frame after damage, absent at EOF or seek reset.
148    pub recovered_pts: Option<i64>,
149    /// Numerator and denominator of both timestamps' time base.
150    pub time_base: (i32, i32),
151}
152
153/// Cumulative diagnostics. Records are bounded; overflow fails decoding.
154#[derive(Clone, Debug, Default, PartialEq, Eq)]
155pub struct DecodeDiagnostics {
156    /// Corruption observations, not necessarily distinct encoded packets.
157    pub corrupt_packets: Vec<CorruptPacket>,
158    /// Concealed audio intervals with original source positions.
159    pub audio_gaps: Vec<ConcealedAudioGap>,
160    /// Repeated audio intervals dropped, with original source positions.
161    pub audio_overlaps: Vec<DroppedAudioOverlap>,
162    /// Healthy boundaries around dropped video data; not a guessed CFR interval.
163    pub video_gaps: Vec<VideoRecoveryGap>,
164}
165
166/// A recovery observation emitted synchronously before decoding continues.
167#[derive(Clone, Debug, PartialEq, Eq)]
168pub enum DecodeDiagnostic {
169    /// Damage reported by the demuxer or decoder.
170    Corruption(CorruptPacket),
171    /// Explicitly inserted silence at source sample positions.
172    AudioGap(ConcealedAudioGap),
173    /// Repeated samples dropped at source sample positions.
174    AudioOverlap(DroppedAudioOverlap),
175    /// Video recovery boundaries without rewriting frame PTS.
176    VideoGap(VideoRecoveryGap),
177}
178
179#[derive(Default)]
180pub(crate) struct Recovery {
181    pub policy: DecodePolicy,
182    pub reporter: Option<Box<dyn Fn(DecodeDiagnostic) + Send + Sync>>,
183    pub diagnostics: DecodeDiagnostics,
184    consecutive: usize,
185    last_pts: Option<i64>,
186    stalled: usize,
187    video_damage: bool,
188    time_base: (i32, i32),
189    healthy_frames: usize,
190    /// The decoder is a hardware session (see [`Self::error`]).
191    hardware: bool,
192}
193impl Recovery {
194    pub fn new(time_base: rsmpeg::avutil::AVRational) -> Self {
195        Self {
196            time_base: (time_base.num, time_base.den),
197            ..Self::default()
198        }
199    }
200    /// Recovery for a hardware decode session.
201    // Only a hardware session, and the tests, open one.
202    #[cfg_attr(not(feature = "hwdecode"), allow(dead_code))]
203    pub fn hardware(time_base: rsmpeg::avutil::AVRational) -> Self {
204        Self {
205            hardware: true,
206            ..Self::new(time_base)
207        }
208    }
209    pub fn finish_video_gap(&mut self, recovered_pts: Option<i64>) {
210        if !self.video_damage {
211            return;
212        }
213        self.video_damage = false;
214        let gap = VideoRecoveryGap {
215            previous_pts: self.last_pts,
216            recovered_pts,
217            time_base: self.time_base,
218        };
219        self.diagnostics.video_gaps.push(gap.clone());
220        if let Some(report) = &self.reporter {
221            report(DecodeDiagnostic::VideoGap(gap));
222        }
223    }
224
225    pub fn error(
226        &mut self,
227        error: RsmpegError,
228        stage: &'static str,
229        pts: Option<i64>,
230    ) -> Result<(), DecodeError> {
231        let damaged = match error.raw_error() {
232            Some(ffi::AVERROR_INVALIDDATA) => true,
233            // A hardware session reports an undecodable picture with its hwaccel's error, never
234            // INVALIDDATA (VideoToolbox: AVERROR_UNKNOWN; Vulkan, D3D12: AVERROR_EXTERNAL).
235            // Tolerant skips it up to the corruption limit; Strict fails, which lets a host retry
236            // in software.
237            Some(ffi::AVERROR_UNKNOWN | ffi::AVERROR_EXTERNAL) => {
238                self.hardware && self.policy == DecodePolicy::Tolerant
239            }
240            _ => false,
241        };
242        if !damaged {
243            return Err(adapter_error(error));
244        }
245        self.corrupt(stage, pts)
246    }
247    pub fn corrupt(&mut self, stage: &'static str, pts: Option<i64>) -> Result<(), DecodeError> {
248        if self.diagnostics.corrupt_packets.len() >= 1024 {
249            return Err(DecodeError::Adapter(
250                "corruption diagnostic budget exceeded".into(),
251            ));
252        }
253        self.diagnostics.corrupt_packets.push(CorruptPacket {
254            stage,
255            pts: pts.filter(|pts| *pts != ffi::AV_NOPTS_VALUE),
256            time_base: self.time_base,
257        });
258        if let Some(report) = &self.reporter {
259            report(DecodeDiagnostic::Corruption(
260                self.diagnostics.corrupt_packets.last().unwrap().clone(),
261            ));
262        }
263        self.video_damage |= stage.starts_with("video");
264        self.consecutive += 1;
265        if self.policy == DecodePolicy::Strict {
266            return Err(DecodeError::Adapter(format!(
267                "corrupt input at {stage} (strict decode policy)"
268            )));
269        }
270        if self.consecutive > 32 {
271            return Err(DecodeError::Adapter(
272                "consecutive corruption budget exceeded".into(),
273            ));
274        }
275        Ok(())
276    }
277    pub fn video_eof(&mut self) -> Result<(), DecodeError> {
278        self.finish_video_gap(None);
279        if self.consecutive != 0 {
280            self.damaged_tail("video")?;
281        }
282        Ok(())
283    }
284    /// The stream ended in damage. It ends at its last healthy frame, unless
285    /// none was decoded since the open or seek: damaged input cannot succeed
286    /// with empty output.
287    pub fn damaged_tail(&self, kind: &str) -> Result<(), DecodeError> {
288        if self.healthy_frames == 0 {
289            return Err(DecodeError::Adapter(format!(
290                "damaged {kind} has no recoverable frames"
291            )));
292        }
293        Ok(())
294    }
295    pub fn progress(&mut self, pts: i64) -> Result<(), DecodeError> {
296        self.healthy_frames = self.healthy_frames.saturating_add(1);
297        self.finish_video_gap((pts != ffi::AV_NOPTS_VALUE).then_some(pts));
298        self.consecutive = 0;
299        if pts != ffi::AV_NOPTS_VALUE && self.last_pts.is_none_or(|previous| pts > previous) {
300            self.last_pts = Some(pts);
301            self.stalled = 0;
302        } else {
303            self.stalled += 1;
304        }
305        if self.stalled > 256 {
306            return Err(DecodeError::Adapter(
307                "decoder PTS progress budget exceeded".into(),
308            ));
309        }
310        Ok(())
311    }
312    pub fn reset_progress(&mut self) {
313        self.finish_video_gap(None);
314        self.last_pts = None;
315        self.healthy_frames = 0;
316        self.stalled = 0;
317        self.consecutive = 0;
318    }
319    pub fn audio_gap(
320        &mut self,
321        start_sample: i64,
322        end_sample: i64,
323        sample_rate: u32,
324        reason: &'static str,
325    ) -> Result<(), DecodeError> {
326        if self.diagnostics.audio_gaps.len() >= MAX_AUDIO_RECORDS {
327            return Err(DecodeError::Adapter(
328                "audio gap diagnostic budget exceeded".into(),
329            ));
330        }
331        self.timestamp_audio_gap(start_sample, end_sample, sample_rate, reason);
332        Ok(())
333    }
334    /// A gap the stream's own timestamps leave: not damage, so never fatal.
335    /// Past the record budget it is still reported, not recorded.
336    pub fn timestamp_gap(&mut self, start_sample: i64, end_sample: i64, sample_rate: u32) {
337        self.timestamp_audio_gap(start_sample, end_sample, sample_rate, "timestamp gap");
338    }
339    fn timestamp_audio_gap(
340        &mut self,
341        start_sample: i64,
342        end_sample: i64,
343        sample_rate: u32,
344        reason: &'static str,
345    ) {
346        let gap = ConcealedAudioGap {
347            start_sample,
348            end_sample,
349            sample_rate,
350            reason,
351        };
352        if let Some(report) = &self.reporter {
353            report(DecodeDiagnostic::AudioGap(gap.clone()));
354        }
355        if self.diagnostics.audio_gaps.len() < MAX_AUDIO_RECORDS {
356            self.diagnostics.audio_gaps.push(gap);
357        }
358    }
359    /// Samples a frame repeated over ones already placed, which were dropped.
360    pub fn timestamp_overlap(&mut self, start_sample: i64, end_sample: i64, sample_rate: u32) {
361        let overlap = DroppedAudioOverlap {
362            start_sample,
363            end_sample,
364            sample_rate,
365        };
366        if let Some(report) = &self.reporter {
367            report(DecodeDiagnostic::AudioOverlap(overlap.clone()));
368        }
369        if self.diagnostics.audio_overlaps.len() < MAX_AUDIO_RECORDS {
370            self.diagnostics.audio_overlaps.push(overlap);
371        }
372    }
373}
374
375/// The longest a decoder operation (a seek, the next frame, an open's first
376/// frame) may go without producing a frame before it fails with "decode
377/// progress budget exceeded": the time since it started or since its last
378/// decoded frame, between the `FFmpeg` calls it makes. A seek that decodes
379/// from a distant keyframe at a steady pace takes as long as it needs; a
380/// loop that decodes nothing does not.
381///
382/// Time spent inside one call is not counted: `FFmpeg` gives no way to stop a
383/// call that is blocked, which is why a caller that watches decode for stalls
384/// allows this plus its own GPU wait limit, from the last
385/// [`watch_decode_progress`] report.
386pub const DECODE_WORK_LIMIT: Duration = Duration::from_secs(10);
387
388/// The most loop steps a decoder operation takes without producing a frame.
389const DECODE_WORK_STEPS: usize = 65_536;
390
391thread_local! {
392    static DECODE_PROGRESS_WATCH: std::cell::RefCell<Option<Box<dyn Fn()>>> =
393        const { std::cell::RefCell::new(None) };
394}
395
396#[cfg(test)]
397thread_local! {
398    /// [`DECODE_WORK_LIMIT`] as a test on this thread sets it, else as
399    /// `ALPHA2_TEST_DECODE_WORK_LIMIT_MS` sets it for a whole run: a run
400    /// under valgrind decodes a frame many times slower and sets a longer
401    /// one, which no production build reads.
402    pub(crate) static WORK_LIMIT: std::cell::Cell<Duration> =
403        std::cell::Cell::new(test_work_limit(std::env::var("ALPHA2_TEST_DECODE_WORK_LIMIT_MS").ok().as_deref()));
404    /// [`DECODE_WORK_STEPS`] as a test on this thread sets it.
405    pub(crate) static WORK_STEPS: std::cell::Cell<usize> =
406        const { std::cell::Cell::new(DECODE_WORK_STEPS) };
407    /// How long a seek on this thread takes after each frame it decodes, as
408    /// a slow decoder takes: slept, or, with [`VIRTUAL_CLOCK`] set, added
409    /// to it.
410    pub(crate) static FRAME_DELAY: std::cell::Cell<Duration> =
411        const { std::cell::Cell::new(Duration::ZERO) };
412    /// When set, the clock a work budget on this thread measures, which only
413    /// [`FRAME_DELAY`] advances: a test of how long an operation may take
414    /// between frames then depends on no real decoding speed, a release
415    /// build's or one under valgrind.
416    pub(crate) static VIRTUAL_CLOCK: std::cell::Cell<Option<Duration>> =
417        const { std::cell::Cell::new(None) };
418}
419
420/// The test builds' default [`WORK_LIMIT`], `setting` milliseconds when it
421/// is a positive count.
422#[cfg(test)]
423fn test_work_limit(setting: Option<&str>) -> Duration {
424    setting
425        .and_then(|value| value.parse::<u64>().ok())
426        .filter(|millis| *millis > 0)
427        .map_or(DECODE_WORK_LIMIT, Duration::from_millis)
428}
429
430/// What a decoder operation on this thread takes after a frame in a test
431/// ([`FRAME_DELAY`]).
432#[cfg(test)]
433pub(crate) fn take_frame_delay() {
434    let delay = FRAME_DELAY.with(std::cell::Cell::get);
435    match VIRTUAL_CLOCK.with(std::cell::Cell::get) {
436        Some(now) => VIRTUAL_CLOCK.with(|clock| clock.set(Some(now.saturating_add(delay)))),
437        None => std::thread::sleep(delay),
438    }
439}
440
441/// Calls `progressed` on this thread each time a decoder operation running
442/// on it produces a frame, or a packet copy's plan reads a source packet,
443/// until the guard drops.
444///
445/// An operation such as a seek can decode many frames before it returns, and
446/// a caller that watches decode for stalls from outside (a host's watchdog)
447/// takes this as the operation being alive rather than blocked.
448#[must_use = "the watch ends when the guard drops"]
449pub fn watch_decode_progress(progressed: impl Fn() + 'static) -> DecodeProgressWatch {
450    let previous = DECODE_PROGRESS_WATCH.with(|watch| watch.replace(Some(Box::new(progressed))));
451    DecodeProgressWatch {
452        previous,
453        thread: std::marker::PhantomData,
454    }
455}
456
457/// Tells this thread's [`watch_decode_progress`] watch, if any, that the
458/// operation in progress produced something: a frame, or a packet of a read
459/// that goes through a whole file.
460pub(crate) fn report_decode_progress() {
461    DECODE_PROGRESS_WATCH.with(|watch| {
462        if let Some(progressed) = watch.borrow().as_ref() {
463            progressed();
464        }
465    });
466}
467
468/// Ends a [`watch_decode_progress`] watch, restoring the one before it.
469pub struct DecodeProgressWatch {
470    previous: Option<Box<dyn Fn()>>,
471    /// The watch belongs to the thread that set it.
472    thread: std::marker::PhantomData<*const ()>,
473}
474
475impl Drop for DecodeProgressWatch {
476    fn drop(&mut self) {
477        let previous = self.previous.take();
478        DECODE_PROGRESS_WATCH.with(|watch| {
479            watch.replace(previous);
480        });
481    }
482}
483
484/// The time and loop steps one decoder operation has spent since it started or
485/// last produced a frame, checked against [`DECODE_WORK_LIMIT`] and
486/// [`DECODE_WORK_STEPS`].
487pub(crate) struct WorkBudget {
488    started: Instant,
489    /// Where [`VIRTUAL_CLOCK`] stood at `started`, when it is set.
490    #[cfg(test)]
491    started_virtual: Option<Duration>,
492    steps: usize,
493}
494impl WorkBudget {
495    pub fn new() -> Self {
496        Self {
497            started: Instant::now(),
498            #[cfg(test)]
499            started_virtual: VIRTUAL_CLOCK.with(std::cell::Cell::get),
500            steps: 0,
501        }
502    }
503
504    /// How long since the operation started or last produced a frame.
505    fn elapsed(&self) -> Duration {
506        #[cfg(test)]
507        if let (Some(now), Some(started)) = (
508            VIRTUAL_CLOCK.with(std::cell::Cell::get),
509            self.started_virtual,
510        ) {
511            return now.saturating_sub(started);
512        }
513        self.started.elapsed()
514    }
515    pub fn step(&mut self, cancel: &CancelToken) -> Result<(), DecodeError> {
516        if cancel.is_cancelled() {
517            return Err(DecodeError::Cancelled);
518        }
519        self.steps += 1;
520        #[cfg(test)]
521        let (limit, steps) = (
522            WORK_LIMIT.with(std::cell::Cell::get),
523            WORK_STEPS.with(std::cell::Cell::get),
524        );
525        #[cfg(not(test))]
526        let (limit, steps) = (DECODE_WORK_LIMIT, DECODE_WORK_STEPS);
527        if self.steps > steps || self.elapsed() > limit {
528            return Err(DecodeError::Adapter(
529                "decode progress budget exceeded".into(),
530            ));
531        }
532        Ok(())
533    }
534    /// The operation produced a frame: its limits start again from now.
535    pub fn progressed(&mut self) {
536        self.started = Instant::now();
537        #[cfg(test)]
538        {
539            self.started_virtual = VIRTUAL_CLOCK.with(std::cell::Cell::get);
540        }
541        self.steps = 0;
542        report_decode_progress();
543    }
544}
545
546/// The closure always refers to the same packet until send accepts it. `true`
547/// requests receive/drain, `false` requests send; its result means progress.
548pub(crate) fn submit_packet(
549    mut step: impl FnMut(bool) -> Result<bool, DecodeError>,
550    budget: &mut WorkBudget,
551    cancel: &CancelToken,
552) -> Result<(), DecodeError> {
553    for _ in 0..64 {
554        budget.step(cancel)?;
555        if step(false)? {
556            return Ok(());
557        }
558        budget.step(cancel)?;
559        if !step(true)? {
560            return Err(DecodeError::Adapter(
561                "decoder send/receive made no progress".into(),
562            ));
563        }
564    }
565    Err(DecodeError::Adapter("packet retry budget exceeded".into()))
566}
567
568#[cfg(test)]
569mod tests {
570    use super::*;
571
572    /// The policy names are the host's, both ways, and nothing else reads as one.
573    #[test]
574    fn a_policy_is_named_as_the_host_names_it() {
575        for policy in [DecodePolicy::Strict, DecodePolicy::Tolerant] {
576            assert_eq!(DecodePolicy::parse(policy.as_str()), Some(policy));
577        }
578        assert_eq!(DecodePolicy::Tolerant.as_str(), "tolerant");
579        assert_eq!(DecodePolicy::parse("Tolerant"), None);
580        assert_eq!(DecodePolicy::parse(""), None);
581    }
582
583    /// Counts sort audio silence by why it was inserted.
584    #[test]
585    fn counts_tell_concealed_damage_from_a_timestamp_gap() {
586        let mut counts = DecodeDiagnosticCounts::default();
587        assert!(counts.is_empty());
588        let gap = |start_sample, end_sample, reason| {
589            DecodeDiagnostic::AudioGap(ConcealedAudioGap {
590                start_sample,
591                end_sample,
592                sample_rate: 48_000,
593                reason,
594            })
595        };
596        for diagnostic in [
597            DecodeDiagnostic::Corruption(CorruptPacket {
598                stage: "audio send",
599                pts: None,
600                time_base: (1, 48_000),
601            }),
602            gap(1024, 2048, "corrupt input"),
603            gap(4096, 4196, "timestamp gap"),
604            DecodeDiagnostic::AudioOverlap(DroppedAudioOverlap {
605                start_sample: 10,
606                end_sample: 15,
607                sample_rate: 48_000,
608            }),
609            DecodeDiagnostic::VideoGap(VideoRecoveryGap {
610                previous_pts: Some(1),
611                recovered_pts: None,
612                time_base: (1, 10),
613            }),
614        ] {
615            counts.record(&diagnostic);
616        }
617        assert_eq!(
618            counts,
619            DecodeDiagnosticCounts {
620                corrupt_packets: 1,
621                video_gaps: 1,
622                concealed_audio_samples: 1024,
623                timestamp_gap_samples: 100,
624                dropped_audio_samples: 5,
625            }
626        );
627        assert!(!counts.is_empty());
628    }
629    /// Under Tolerant a hardware session's failure to decode one picture
630    /// is damage, as `INVALIDDATA` is for software: a hardware session never
631    /// reports `INVALIDDATA`, so Tolerant could not otherwise skip its damage.
632    #[test]
633    fn a_hardware_sessions_picture_failure_is_damage_only_when_tolerant() {
634        let base = rsmpeg::avutil::ra(1, 30);
635        for code in [ffi::AVERROR_UNKNOWN, ffi::AVERROR_EXTERNAL] {
636            let mut tolerant = Recovery::hardware(base);
637            tolerant.policy = DecodePolicy::Tolerant;
638            assert!(
639                tolerant
640                    .error(RsmpegError::AVError(code), "video send", Some(3))
641                    .is_ok()
642            );
643            assert_eq!(tolerant.diagnostics.corrupt_packets.len(), 1);
644            let mut strict = Recovery::hardware(base);
645            assert!(
646                strict
647                    .error(RsmpegError::AVError(code), "video send", None)
648                    .is_err()
649            );
650            let mut software = Recovery::new(base);
651            software.policy = DecodePolicy::Tolerant;
652            assert!(
653                software
654                    .error(RsmpegError::AVError(code), "video send", None)
655                    .is_err()
656            );
657        }
658    }
659
660    #[test]
661    fn strict_tolerant_and_fatal_errors() {
662        let mut state = Recovery::default();
663        assert!(state.corrupt("send", Some(12)).is_err());
664        state.policy = DecodePolicy::Tolerant;
665        state.reset_progress();
666        assert!(
667            state
668                .error(
669                    RsmpegError::AVError(ffi::AVERROR_INVALIDDATA),
670                    "send",
671                    Some(13)
672                )
673                .is_ok()
674        );
675        assert!(
676            state
677                .error(RsmpegError::AVError(-12), "send", None)
678                .is_err()
679        );
680        assert_eq!(state.diagnostics.corrupt_packets.len(), 2);
681        for _ in 0..31 {
682            state.corrupt("receive", None).unwrap();
683        }
684        assert!(state.corrupt("receive", None).is_err());
685    }
686    #[test]
687    fn progress_and_cancellation_are_bounded() {
688        let mut state = Recovery::default();
689        state.progress(10).unwrap();
690        for _ in 0..256 {
691            state.progress(10).unwrap();
692        }
693        assert!(state.progress(10).is_err());
694        state.reset_progress();
695        state.progress(0).unwrap();
696        let cancel = CancelToken::default();
697        let mut budget = WorkBudget::new();
698        for _ in 0..65_536 {
699            budget.step(&cancel).unwrap();
700        }
701        assert!(budget.step(&cancel).is_err());
702        // The production limit, whatever ALPHA2_TEST_DECODE_WORK_LIMIT_MS gives this run.
703        WORK_LIMIT.with(|limit| limit.set(DECODE_WORK_LIMIT));
704        let mut budget = WorkBudget::new();
705        budget.started -= Duration::from_secs(11);
706        assert!(budget.step(&cancel).is_err());
707        cancel.cancel();
708        assert_eq!(WorkBudget::new().step(&cancel), Err(DecodeError::Cancelled));
709    }
710    #[test]
711    fn eagain_drains_and_retries_the_same_packet_until_accepted() {
712        let mut actions = Vec::new();
713        let mut sends = 0;
714        submit_packet(
715            |drain| {
716                actions.push(drain);
717                if drain {
718                    return Ok(true);
719                }
720                sends += 1;
721                Ok(sends == 3)
722            },
723            &mut WorkBudget::new(),
724            &CancelToken::default(),
725        )
726        .unwrap();
727        assert_eq!(actions, [false, true, false, true, false]);
728        assert!(
729            submit_packet(
730                |_| Ok(false),
731                &mut WorkBudget::new(),
732                &CancelToken::default()
733            )
734            .is_err()
735        );
736        assert!(submit_packet(Ok, &mut WorkBudget::new(), &CancelToken::default()).is_err());
737        let cancel = CancelToken::default();
738        let result = submit_packet(
739            |drain| {
740                if drain {
741                    cancel.cancel();
742                }
743                Ok(drain)
744            },
745            &mut WorkBudget::new(),
746            &cancel,
747        );
748        assert_eq!(result, Err(DecodeError::Cancelled));
749    }
750}
alpha2_native/recovery.rs

alpha2_native/
recovery.rs
