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

alpha2_native/
recovery.rs
