zencodec/traits/

incremental.rs

//! Incremental (sans-I/O) decoding: the contract between a resumable
//! decoder and whatever holds the bytes.
//!
//! A decoder performs no I/O. A driver offers it bytes it already holds and
//! steps it; each step reports how many of those bytes the decoder took and
//! one [`Event`]. The driver owns the input buffer: bytes the decoder did not
//! take are still the driver's, to offer again (with more appended) or to
//! drop. Rows go straight into the driver's [`DecodeRowSink`]. The same
//! decoder runs under a synchronous file loop, an async network task, an FFI
//! host that pushes buffers, or a single-threaded WASM event loop.
//!
//! ```text
//! let mut dec = job.dyn_incremental_decoder(&[])?;
//! let mut input = Input::empty();
//! loop {
//!     let p = dec.step(input, &mut sink, Some(&pulse))?;
//!     // drop p.consumed bytes from the front of what was offered
//!     match p.event {
//!         Event::Need => { /* satisfy one of dec.requested(), build the next Input */ }
//!         Event::Done(info) => break,
//!         _ => { /* Rows, Metadata, or a newer informational event: step again */ }
//!     }
//! }
//! ```
//!
//! # The rules
//!
//! **Input.** [`Input`] is a run of bytes at an absolute `offset`, possibly
//! empty, plus the total source length once the driver knows it. A decoder
//! takes a prefix of the input and reports its length in
//! [`Progress::consumed`]: the bytes it decoded in place or copied into its
//! own state, plus any bytes before them it does not need (a gap the driver
//! read across to batch two ranges). Consumed bytes are never offered again.
//! A decoder never fails because it was offered bytes it did not ask for: it
//! takes none of them, or skips them as part of the prefix.
//!
//! **Requests.** After [`Event::Need`], [`IncrementalDecode::requested`]
//! lists what the decoder needs, never empty. The driver satisfies any one
//! request and steps again:
//! - [`InputRequest::AtLeast`]: an input starting at `offset` holding at least
//!   `min_len` bytes, or every remaining byte with `total_len` set if the
//!   source ends first. More is welcome.
//! - [`InputRequest::Range`]: an input holding exactly that span (or a larger
//!   input starting at its offset).
//! - [`InputRequest::Len`]: any input (even empty) carrying `total_len`.
//!
//! A decoder never requests bytes it knows do not exist: once `total_len` is
//! known, a request that cannot be met is a [`StepError::Codec`] error
//! (truncated input), not another `Need`. A decoder makes progress whenever
//! its input satisfies one of its requests; returning `Need` with nothing
//! consumed for an input that satisfied a request is a protocol violation,
//! and drivers report it instead of looping.
//!
//! **Output.** [`begin`](DecodeRowSink::begin) comes before the first rows
//! and [`finish`](DecodeRowSink::finish) before [`Event::Done`]. Every row is
//! written once in its final form through
//! [`provide_next_buffer`](DecodeRowSink::provide_next_buffer), in
//! increasing `y`. A progressive decoder may also write provisional rows
//! (a pass a later one refines) through
//! [`provide_pass_buffer`](DecodeRowSink::provide_pass_buffer), only when
//! the sink's [`accepts_provisional`](DecodeRowSink::accepts_provisional)
//! returned `true` at `begin`. A decoder with no provisional output ignores
//! both.
//!
//! **Termination.** After [`Event::Done`], every step returns `Done` again
//! with nothing consumed. After any error, every step returns
//! [`StepError::Finished`]. `pulse` is checked at least once per step; a
//! tripped pulse is [`StepError::Cancelled`], and terminal. [`Fused`] enforces
//! all three for any decoder, and
//! [`erase_incremental`](crate::decode::erase_incremental) applies it.
//!
//! **Events.** [`Event`] is non-exhaustive. Variants added later are
//! informational: a driver that does not recognise one steps again. Anything
//! that needs the driver to act is expressed as `Need` plus a [`InputRequest`].

use alloc::boxed::Box;
use core::fmt;

use enough::StopReason;

use super::control::{
    BoundaryAction, BoundaryId, ControlError, ControlledIncrementalDecode, ControlledProgress,
    DecodeBoundary, GuardedSink, InterpretationFacts,
};
use crate::sink::{DecodeRowSink, SinkError};
use crate::{ImageInfo, OutputInfo};

/// A byte span within an image file or stream.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
#[non_exhaustive]
pub struct ByteRange {
    /// Starting byte offset from the beginning of the stream.
    pub offset: u64,
    /// Number of bytes in this range.
    pub length: u64,
}

impl ByteRange {
    /// Create a new byte range.
    #[inline]
    pub const fn new(offset: u64, length: u64) -> Self {
        Self { offset, length }
    }

    /// The ending byte offset (exclusive), saturating.
    #[inline]
    pub const fn end(&self) -> u64 {
        self.offset.saturating_add(self.length)
    }

    /// Returns `true` if this range has zero length.
    #[inline]
    pub const fn is_empty(&self) -> bool {
        self.length == 0
    }

    /// Returns `true` if this range contains `target_offset`.
    #[inline]
    pub const fn contains(&self, target_offset: u64) -> bool {
        target_offset >= self.offset && target_offset < self.end()
    }
}

/// What an incremental decoder needs before it can continue.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum InputRequest {
    /// Bytes from `offset` on: an input starting there with at least
    /// `min_len` bytes, or every remaining byte (with `total_len` set) when
    /// the source ends first. More is welcome. A sequential decoder asks
    /// this way; `min_len` is the contiguous span it needs to make progress
    /// without copying (one worst-case row, one marker segment).
    AtLeast {
        /// Absolute offset of the first byte needed.
        offset: u64,
        /// Minimum contiguous bytes needed in one input.
        min_len: usize,
    },
    /// Exactly this span.
    Range(ByteRange),
    /// The total source length, supplied as [`Input::total_len`].
    Len,
}

/// Legacy ambiguous name for a byte acquisition request.
#[deprecated(note = "use InputRequest")]
pub type Request = InputRequest;

impl InputRequest {
    /// A request for at least `min_len` bytes from `offset`.
    #[inline]
    pub const fn at_least(offset: u64, min_len: usize) -> Self {
        Self::AtLeast { offset, min_len }
    }

    /// A request for exactly `length` bytes at `offset`.
    #[inline]
    pub const fn range(offset: u64, length: u64) -> Self {
        Self::Range(ByteRange::new(offset, length))
    }

    /// The offset this request starts at, if it asks for bytes.
    #[inline]
    pub const fn offset(&self) -> Option<u64> {
        match self {
            Self::AtLeast { offset, .. } => Some(*offset),
            Self::Range(r) => Some(r.offset),
            Self::Len => None,
        }
    }

    /// Whether `input` satisfies this request.
    pub fn is_satisfied_by(&self, input: &Input<'_>) -> bool {
        match *self {
            Self::AtLeast { offset, min_len } => {
                input.offset == offset
                    && (input.bytes.len() >= min_len || input.total_len == Some(input.end()))
            }
            Self::Range(r) => {
                input.offset == r.offset
                    && (input.bytes.len() as u64 >= r.length
                        || input.total_len == Some(input.end()))
            }
            Self::Len => input.total_len.is_some(),
        }
    }
}

/// Bytes offered to one [`step`](IncrementalDecode::step).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct Input<'a> {
    /// Absolute offset of `bytes[0]` in the source.
    pub offset: u64,
    /// The bytes. May be empty.
    pub bytes: &'a [u8],
    /// Total source length, once the driver knows it.
    pub total_len: Option<u64>,
}

impl<'a> Input<'a> {
    /// `bytes` located at `offset`, total length unknown.
    #[inline]
    pub const fn new(offset: u64, bytes: &'a [u8]) -> Self {
        Self {
            offset,
            bytes,
            total_len: None,
        }
    }

    /// No bytes: step without new input.
    #[inline]
    pub const fn empty() -> Self {
        Self {
            offset: 0,
            bytes: &[],
            total_len: None,
        }
    }

    /// The same input, carrying the total source length.
    #[inline]
    pub const fn with_total_len(mut self, total_len: u64) -> Self {
        self.total_len = Some(total_len);
        self
    }

    /// The same input, carrying the total source length if known.
    #[inline]
    pub const fn with_total_len_opt(mut self, total_len: Option<u64>) -> Self {
        self.total_len = total_len;
        self
    }

    /// Absolute offset one past the last byte.
    #[inline]
    pub const fn end(&self) -> u64 {
        self.offset.saturating_add(self.bytes.len() as u64)
    }

    /// Whether these bytes run to the end of the source.
    #[inline]
    pub fn reaches_end(&self) -> bool {
        self.total_len == Some(self.end())
    }
}

/// Metadata update reported by [`Event::Metadata`].
///
/// Image containers deliver metadata across stages: the dimensions header,
/// then colour signalling, then supplementary or trailing payloads.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum MetadataUpdate {
    /// Dimensions and format header are available in
    /// [`info()`](IncrementalDecode::info).
    HeaderReady,
    /// Colour profile or primaries (ICC / CICP) are available or refined.
    ColorReady,
    /// Supplementary or trailing metadata (EXIF, XMP, gain map) was
    /// discovered or updated.
    Updated,
}

impl MetadataUpdate {
    /// Returns `true` if basic image dimensions and format are ready.
    #[inline]
    pub const fn is_header_ready(&self) -> bool {
        matches!(self, Self::HeaderReady | Self::ColorReady | Self::Updated)
    }
}

/// Metadata describing a batch of decoded scanlines.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub struct BatchMeta {
    /// Starting row index (top row of this batch).
    pub y: u32,
    /// Progressive pass index (0 for baseline / non-progressive decode).
    pub pass: u16,
    /// Whether this batch is the final form of these scanlines.
    pub is_final: bool,
}

impl BatchMeta {
    /// Batch metadata with baseline defaults (`pass = 0`).
    #[inline]
    pub const fn new(y: u32, is_final: bool) -> Self {
        Self {
            y,
            pass: 0,
            is_final,
        }
    }

    /// Final rows starting at `y` (`pass = 0`, `is_final = true`).
    #[inline]
    pub const fn final_rows(y: u32) -> Self {
        Self::new(y, true)
    }

    /// Specify the progressive pass index.
    #[inline]
    pub const fn with_pass(mut self, pass: u16) -> Self {
        self.pass = pass;
        self
    }
}

/// What one [`step`](IncrementalDecode::step) produced.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum Event {
    /// The decoder needs bytes or the source length; see
    /// [`requested`](IncrementalDecode::requested).
    Need,
    /// `rows` rows starting at `meta.y` were written into the sink.
    Rows {
        /// Row position, pass and finality of the rows just written.
        meta: BatchMeta,
        /// Number of rows written.
        rows: u32,
    },
    /// Metadata became available or was refined.
    Metadata(MetadataUpdate),
    /// Decoding finished; the sink has been told to `finish`.
    Done(OutputInfo),
}

impl Event {
    /// Returns `true` for [`Event::Done`].
    #[inline]
    pub const fn is_done(&self) -> bool {
        matches!(self, Self::Done(_))
    }
}

/// The result of one [`step`](IncrementalDecode::step).
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct Progress {
    /// Bytes taken from the front of the offered input.
    pub consumed: usize,
    /// What happened.
    pub event: Event,
}

impl Progress {
    /// A step that took `consumed` bytes and produced `event`.
    #[inline]
    pub const fn new(consumed: usize, event: Event) -> Self {
        Self { consumed, event }
    }

    /// A step that needs more input and took `consumed` bytes.
    #[inline]
    pub const fn need(consumed: usize) -> Self {
        Self::new(consumed, Event::Need)
    }
}

/// Why a [`step`](IncrementalDecode::step) failed. Every error is terminal.
#[derive(Debug)]
#[non_exhaustive]
pub enum StepError<E> {
    /// The decoder rejected the input (malformed, truncated, unsupported,
    /// over a limit).
    Codec(E),
    /// The sink refused rows or failed.
    Sink(SinkError),
    /// The pulse token tripped.
    Cancelled(StopReason),
    /// The decoder was stepped again after an error.
    Finished,
    /// A mandatory control boundary was unhandled or violated.
    Control(ControlError),
}

impl<E> StepError<E> {
    /// Map the codec error, keeping the other variants.
    pub fn map_codec<F>(self, f: impl FnOnce(E) -> F) -> StepError<F> {
        match self {
            Self::Codec(e) => StepError::Codec(f(e)),
            Self::Sink(e) => StepError::Sink(e),
            Self::Cancelled(r) => StepError::Cancelled(r),
            Self::Finished => StepError::Finished,
            Self::Control(e) => StepError::Control(e),
        }
    }
}

impl<E> From<StopReason> for StepError<E> {
    fn from(r: StopReason) -> Self {
        Self::Cancelled(r)
    }
}

impl<E: fmt::Display> fmt::Display for StepError<E> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Codec(e) => write!(f, "{e}"),
            Self::Sink(e) => write!(f, "row sink: {e}"),
            Self::Cancelled(r) => write!(f, "decode cancelled: {r:?}"),
            Self::Finished => write!(f, "decoder stepped after it failed"),
            Self::Control(e) => e.fmt(f),
        }
    }
}

impl<E: core::error::Error + 'static> core::error::Error for StepError<E> {
    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
        match self {
            Self::Codec(e) => Some(e),
            Self::Sink(e) => Some(e.as_ref()),
            Self::Control(e) => Some(e),
            _ => None,
        }
    }
}

/// A resumable decoder driven by whoever holds the bytes. See the
/// contract in `docs/range-decoding-contract.md` for the rules.
///
/// The trait is object-safe: [`DynIncrementalDecoder`] is
/// `dyn IncrementalDecode<Error = ErasedError> + Send`.
pub trait IncrementalDecode {
    /// The codec-specific error type.
    type Error: core::error::Error + Send + Sync + 'static;

    /// Take bytes from the front of `input` and advance by one event.
    ///
    /// `None` adds no per-call control. A decoder must still honor its retained
    /// operation pulse. Forwarding adapters preserve absence; they must not
    /// replace it with a new no-op control that overrides the retained pulse.
    fn step(
        &mut self,
        input: Input<'_>,
        sink: &mut dyn DecodeRowSink,
        pulse: Option<&dyn crate::Pulse>,
    ) -> Result<Progress, StepError<Self::Error>>;

    /// What the decoder needs, valid after a step returned [`Event::Need`]
    /// (never empty then). Empty otherwise.
    fn requested(&self) -> &[InputRequest];

    /// Image metadata, once the header has been parsed.
    fn info(&self) -> Option<&ImageInfo>;

    /// Take the owned details of a successfully completed decode, at most once.
    ///
    /// Before `Done`, after failure/cancellation, after a previous take, or when
    /// unsupported, return `None`. A decoder returning `Some` must include all
    /// result details produced by this configured operation, including typed
    /// extras and source-encoding details; a header snapshot is not a substitute.
    /// This does not certify complete container inventory or absence of PII.
    ///
    /// Consumers requiring owned details must reject `None`, not manufacture an
    /// empty result. The default preserves compatibility with existing decoders.
    fn take_details(&mut self) -> Option<crate::decode::DecodeDetails> {
        None
    }

    /// Output pixel descriptor, once the header has been parsed.
    fn output_descriptor(&self) -> Option<zenpixels::PixelDescriptor> {
        None
    }

    /// How this decoder reads its input, so a driver can choose a buffering
    /// strategy. May change once the format is known.
    fn access_pattern(&self) -> crate::source::AccessPattern {
        crate::source::AccessPattern::Sequential
    }
}

/// A codec error with its type erased: the error type of
/// [`DynIncrementalDecoder`].
///
/// `Box<dyn Error>` does not itself implement `Error`, so the erased
/// decoder needs this wrapper. It displays as the inner error and reports it
/// as its [`source`](core::error::Error::source), so cause-chain lookups
/// (such as [`CodecErrorExt`](crate::CodecErrorExt)) reach the original.
pub struct ErasedError(crate::traits::BoxedError);

impl ErasedError {
    /// Wrap a boxed error.
    pub fn new(err: crate::traits::BoxedError) -> Self {
        Self(err)
    }

    /// The boxed error.
    pub fn into_inner(self) -> crate::traits::BoxedError {
        self.0
    }

    /// The error, borrowed.
    pub fn get_ref(&self) -> &(dyn core::error::Error + Send + Sync + 'static) {
        &*self.0
    }
}

impl fmt::Debug for ErasedError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt::Debug::fmt(&self.0, f)
    }
}

impl fmt::Display for ErasedError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt::Display::fmt(&self.0, f)
    }
}

impl core::error::Error for ErasedError {
    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
        Some(&*self.0)
    }
}

impl StepError<ErasedError> {
    /// Flatten into a boxed error: a codec error unwraps to the original,
    /// anything else is boxed as is.
    pub fn into_boxed(self) -> crate::traits::BoxedError {
        match self {
            StepError::Codec(c) => c.0,
            other => Box::new(other),
        }
    }
}

/// The object-safe form every erased incremental decoder takes.
pub type DynIncrementalDecoder = dyn IncrementalDecode<Error = ErasedError> + Send;

impl<D: IncrementalDecode + ?Sized> IncrementalDecode for Box<D> {
    type Error = D::Error;

    fn step(
        &mut self,
        input: Input<'_>,
        sink: &mut dyn DecodeRowSink,
        pulse: Option<&dyn crate::Pulse>,
    ) -> Result<Progress, StepError<D::Error>> {
        (**self).step(input, sink, pulse)
    }

    fn requested(&self) -> &[InputRequest] {
        (**self).requested()
    }

    fn info(&self) -> Option<&ImageInfo> {
        (**self).info()
    }

    fn take_details(&mut self) -> Option<crate::decode::DecodeDetails> {
        (**self).take_details()
    }

    fn output_descriptor(&self) -> Option<zenpixels::PixelDescriptor> {
        (**self).output_descriptor()
    }

    fn access_pattern(&self) -> crate::source::AccessPattern {
        (**self).access_pattern()
    }
}

impl<D: IncrementalDecode + ?Sized> IncrementalDecode for &mut D {
    type Error = D::Error;

    fn step(
        &mut self,
        input: Input<'_>,
        sink: &mut dyn DecodeRowSink,
        pulse: Option<&dyn crate::Pulse>,
    ) -> Result<Progress, StepError<D::Error>> {
        (**self).step(input, sink, pulse)
    }

    fn requested(&self) -> &[InputRequest] {
        (**self).requested()
    }

    fn info(&self) -> Option<&ImageInfo> {
        (**self).info()
    }

    fn take_details(&mut self) -> Option<crate::decode::DecodeDetails> {
        (**self).take_details()
    }

    fn output_descriptor(&self) -> Option<zenpixels::PixelDescriptor> {
        (**self).output_descriptor()
    }

    fn access_pattern(&self) -> crate::source::AccessPattern {
        (**self).access_pattern()
    }
}

/// Enforces the termination rules for any decoder: steps after `Done`
/// return `Done` again, steps after an error return
/// [`StepError::Finished`], and `pulse` is checked before every step, a
/// tripped pulse being terminal.
///
/// Wrap a hand-written decoder in `Fused` and its own state machine need
/// not track any of this.
#[derive(Debug)]
pub struct Fused<D> {
    inner: D,
    state: FusedState,
    controlled: bool,
    interpretation: Option<InterpretationFacts>,
    details_taken: bool,
}

#[derive(Debug)]
enum FusedState {
    Running,
    Suspended(DecodeBoundary),
    Done(OutputInfo),
    Failed,
}

impl<D> Fused<D> {
    /// Wrap `inner`.
    pub const fn new(inner: D) -> Self {
        Self {
            inner,
            state: FusedState::Running,
            controlled: false,
            interpretation: None,
            details_taken: false,
        }
    }

    pub(super) fn controlled(inner: D) -> Self {
        Self {
            controlled: true,
            ..Self::new(inner)
        }
    }

    /// The wrapped decoder.
    pub fn get_ref(&self) -> &D {
        &self.inner
    }

    /// Unwrap.
    pub fn into_inner(self) -> D {
        self.inner
    }
}

impl<D: IncrementalDecode> IncrementalDecode for Fused<D> {
    type Error = D::Error;

    fn step(
        &mut self,
        input: Input<'_>,
        sink: &mut dyn DecodeRowSink,
        pulse: Option<&dyn crate::Pulse>,
    ) -> Result<Progress, StepError<D::Error>> {
        match &self.state {
            FusedState::Done(info) => return Ok(Progress::new(0, Event::Done(info.clone()))),
            FusedState::Failed => return Err(StepError::Finished),
            FusedState::Suspended(_) => {
                self.state = FusedState::Failed;
                return Err(StepError::Control(ControlError::UnhandledBoundary));
            }
            FusedState::Running => {}
        }
        if self.controlled {
            self.state = FusedState::Failed;
            return Err(StepError::Control(ControlError::UnhandledBoundary));
        }
        if let Err(reason) = pulse.unwrap_or(&crate::NoPulse).check() {
            self.state = FusedState::Failed;
            return Err(StepError::Cancelled(reason));
        }
        self.state = FusedState::Failed;
        match self.inner.step(input, sink, pulse) {
            Ok(p) => {
                if let Event::Done(info) = &p.event {
                    self.state = FusedState::Done(info.clone());
                } else {
                    self.state = FusedState::Running;
                }
                Ok(p)
            }
            Err(e) => {
                self.state = FusedState::Failed;
                Err(e)
            }
        }
    }

    fn requested(&self) -> &[InputRequest] {
        match self.state {
            FusedState::Running => self.inner.requested(),
            _ => &[],
        }
    }

    fn info(&self) -> Option<&ImageInfo> {
        self.inner.info()
    }

    fn take_details(&mut self) -> Option<crate::decode::DecodeDetails> {
        if !matches!(self.state, FusedState::Done(_)) || self.details_taken {
            return None;
        }
        // Mark before delegation, including a native panic or unavailable result.
        self.details_taken = true;
        self.inner.take_details()
    }

    fn output_descriptor(&self) -> Option<zenpixels::PixelDescriptor> {
        self.inner.output_descriptor()
    }

    fn access_pattern(&self) -> crate::source::AccessPattern {
        self.inner.access_pattern()
    }
}

impl<D: ControlledIncrementalDecode> ControlledIncrementalDecode for Fused<D> {
    fn step_controlled(
        &mut self,
        input: Input<'_>,
        sink: &mut dyn DecodeRowSink,
        pulse: Option<&dyn crate::Pulse>,
    ) -> Result<ControlledProgress, StepError<D::Error>> {
        self.controlled = true;
        match &self.state {
            FusedState::Done(info) => return Ok(Progress::new(0, Event::Done(info.clone())).into()),
            FusedState::Failed => return Err(StepError::Finished),
            _ => {}
        }
        if let Err(reason) = pulse.unwrap_or(&crate::NoPulse).check() {
            self.state = FusedState::Failed;
            return Err(StepError::Cancelled(reason));
        }
        if let FusedState::Suspended(boundary) = self.state {
            return Ok(ControlledProgress::Suspended {
                consumed: 0,
                boundary,
            });
        }
        // Caught unwinds cannot resume a half-mutated native decoder.
        self.state = FusedState::Failed;
        let mut guarded = GuardedSink::new(sink, self.interpretation);
        let result = self.inner.step_controlled(input, &mut guarded, pulse);
        if let Some(error) = guarded.refused {
            return Err(StepError::Control(error));
        }
        if let Some(error) = guarded.take_sink_error() {
            return Err(StepError::Sink(error));
        }
        let progress = result?;
        if self.interpretation.is_none()
            && matches!(
                &progress,
                ControlledProgress::Decode(Progress {
                    event: Event::Done(_) | Event::Rows { .. },
                    ..
                })
            )
        {
            return Err(StepError::Control(ControlError::OutputBeforeInterpretation));
        }
        if let ControlledProgress::Suspended { boundary, .. } = &progress
            && self
                .interpretation
                .is_some_and(|facts| facts != boundary.facts())
        {
            return Err(StepError::Control(ControlError::FinalInterpretationChanged));
        }
        self.state = match &progress {
            ControlledProgress::Decode(Progress {
                event: Event::Done(info),
                ..
            }) => {
                if self.interpretation.is_some_and(|facts| {
                    facts.width != info.width
                        || facts.height != info.height
                        || facts.descriptor != Some(info.native_format)
                }) {
                    return Err(StepError::Control(ControlError::InconsistentFacts));
                }
                FusedState::Done(info.clone())
            }
            ControlledProgress::Suspended { boundary, .. } => FusedState::Suspended(*boundary),
            _ => FusedState::Running,
        };
        Ok(progress)
    }

    fn resume(
        &mut self,
        id: BoundaryId,
        action: BoundaryAction,
        pulse: Option<&dyn crate::Pulse>,
    ) -> Result<(), StepError<D::Error>> {
        let state = core::mem::replace(&mut self.state, FusedState::Failed);
        let FusedState::Suspended(boundary) = state else {
            return Err(StepError::Finished);
        };
        pulse.unwrap_or(&crate::NoPulse).check()?;
        if boundary.id() != id || !action.permits(boundary.facts()) {
            return Err(StepError::Control(ControlError::InvalidReply));
        }
        self.inner.resume(id, action, pulse)?;
        if action == BoundaryAction::AdoptInterpretation {
            self.interpretation = Some(boundary.facts());
        }
        self.state = FusedState::Running;
        Ok(())
    }
}

/// Shim erasing a decoder's error type to [`ErasedError`].
pub struct IncrementalDecoderShim<D>(pub D);

impl<D: IncrementalDecode> IncrementalDecode for IncrementalDecoderShim<D> {
    type Error = ErasedError;

    fn step(
        &mut self,
        input: Input<'_>,
        sink: &mut dyn DecodeRowSink,
        pulse: Option<&dyn crate::Pulse>,
    ) -> Result<Progress, StepError<ErasedError>> {
        self.0
            .step(input, sink, pulse)
            .map_err(|e| e.map_codec(|c| ErasedError(Box::new(c))))
    }

    fn requested(&self) -> &[InputRequest] {
        self.0.requested()
    }

    fn info(&self) -> Option<&ImageInfo> {
        self.0.info()
    }

    fn take_details(&mut self) -> Option<crate::decode::DecodeDetails> {
        self.0.take_details()
    }

    fn output_descriptor(&self) -> Option<zenpixels::PixelDescriptor> {
        self.0.output_descriptor()
    }

    fn access_pattern(&self) -> crate::source::AccessPattern {
        self.0.access_pattern()
    }
}

/// Erase `decoder` for [`DecodeJob::dyn_incremental_decoder`](crate::decode::DecodeJob::dyn_incremental_decoder),
/// wrapped in [`Fused`] so the termination rules hold by construction.
pub fn erase_incremental<D>(decoder: D) -> Box<DynIncrementalDecoder>
where
    D: IncrementalDecode + Send + 'static,
{
    Box::new(IncrementalDecoderShim(Fused::new(decoder)))
}

#[cfg(test)]
mod tests {
    use super::*;
    use zenpixels::{PixelDescriptor, PixelSliceMut};

    struct NullSink;
    impl DecodeRowSink for NullSink {
        fn provide_next_buffer(
            &mut self,
            _y: u32,
            _h: u32,
            _w: u32,
            _d: PixelDescriptor,
        ) -> Result<PixelSliceMut<'_>, SinkError> {
            Err("null sink".into())
        }
    }

    /// Fails on its second step, then would happily continue.
    struct Flaky(u32);
    impl IncrementalDecode for Flaky {
        type Error = core::fmt::Error;
        fn step(
            &mut self,
            _input: Input<'_>,
            _sink: &mut dyn DecodeRowSink,
            _pulse: Option<&dyn crate::Pulse>,
        ) -> Result<Progress, StepError<core::fmt::Error>> {
            self.0 += 1;
            if self.0 == 2 {
                Err(StepError::Codec(core::fmt::Error))
            } else {
                Ok(Progress::need(0))
            }
        }
        fn requested(&self) -> &[InputRequest] {
            &[InputRequest::Len]
        }
        fn info(&self) -> Option<&ImageInfo> {
            None
        }
    }

    #[test]
    fn fused_makes_errors_and_cancellation_terminal() {
        let mut d = Fused::new(Flaky(0));
        let pulse = &crate::how_far::NoPulse;
        assert!(d.step(Input::empty(), &mut NullSink, Some(&pulse)).is_ok());
        assert!(matches!(
            d.step(Input::empty(), &mut NullSink, Some(&pulse)),
            Err(StepError::Codec(_))
        ));
        assert!(matches!(
            d.step(Input::empty(), &mut NullSink, Some(&pulse)),
            Err(StepError::Finished)
        ));
        assert!(d.requested().is_empty());

        struct Tripped;
        impl enough::Stop for Tripped {
            fn check(&self) -> Result<(), StopReason> {
                Err(StopReason::Cancelled)
            }
        }
        let mut d = Fused::new(Flaky(0));
        assert!(matches!(
            d.step(
                Input::empty(),
                &mut NullSink,
                Some(&how_far::StopOnly::borrowed(&Tripped))
            ),
            Err(StepError::Cancelled(StopReason::Cancelled))
        ));
        assert!(matches!(
            d.step(Input::empty(), &mut NullSink, Some(&pulse)),
            Err(StepError::Finished)
        ));
    }

    struct DetailSource {
        calls: alloc::sync::Arc<core::sync::atomic::AtomicUsize>,
        fail: bool,
    }
    impl IncrementalDecode for DetailSource {
        type Error = core::fmt::Error;
        fn step(
            &mut self,
            _: Input<'_>,
            _: &mut dyn DecodeRowSink,
            _: Option<&dyn crate::Pulse>,
        ) -> Result<Progress, StepError<Self::Error>> {
            if self.fail {
                return Err(StepError::Codec(core::fmt::Error));
            }
            Ok(Progress::new(
                0,
                Event::Done(OutputInfo::full_decode(1, 1, PixelDescriptor::GRAY8_SRGB)),
            ))
        }
        fn requested(&self) -> &[InputRequest] {
            &[]
        }
        fn info(&self) -> Option<&ImageInfo> {
            None
        }
        // Deliberately eager and repeatable: Fused must enforce the protocol.
        fn take_details(&mut self) -> Option<crate::decode::DecodeDetails> {
            self.calls
                .fetch_add(1, core::sync::atomic::Ordering::SeqCst);
            Some(crate::decode::DecodeDetails::new(ImageInfo::new(
                1,
                1,
                crate::ImageFormat::Jpeg,
            )))
        }
    }

    #[test]
    fn fused_details_require_success_and_allow_one_take_through_erasure() {
        use core::sync::atomic::{AtomicUsize, Ordering};
        let calls = alloc::sync::Arc::new(AtomicUsize::new(0));
        let mut decoder = erase_incremental(DetailSource {
            calls: calls.clone(),
            fail: false,
        });
        assert!(decoder.take_details().is_none());
        assert_eq!(calls.load(Ordering::SeqCst), 0);
        decoder.step(Input::empty(), &mut NullSink, None).unwrap();
        // Exercise Box forwarding, &mut forwarding, erasure and Fused together.
        let mut borrowed = &mut decoder;
        assert_eq!(
            IncrementalDecode::take_details(&mut borrowed)
                .unwrap()
                .info
                .width,
            1
        );
        assert!(decoder.take_details().is_none());
        decoder.step(Input::empty(), &mut NullSink, None).unwrap();
        assert!(decoder.take_details().is_none());
        assert_eq!(calls.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn fused_details_never_call_native_after_failure_or_cancellation() {
        use core::sync::atomic::{AtomicUsize, Ordering};
        struct Cancel;
        impl enough::Stop for Cancel {
            fn check(&self) -> Result<(), StopReason> {
                Err(StopReason::Cancelled)
            }
        }
        for cancel in [false, true] {
            let calls = alloc::sync::Arc::new(AtomicUsize::new(0));
            let mut decoder = erase_incremental(DetailSource {
                calls: calls.clone(),
                fail: !cancel,
            });
            let pulse = how_far::StopOnly::borrowed(&Cancel);
            assert!(
                decoder
                    .step(
                        Input::empty(),
                        &mut NullSink,
                        cancel.then_some(&pulse as &dyn crate::Pulse)
                    )
                    .is_err()
            );
            assert!(decoder.take_details().is_none());
            assert_eq!(calls.load(Ordering::SeqCst), 0);
        }
    }

    #[test]
    fn requests_know_what_satisfies_them() {
        let bytes = [0u8; 10];
        let r = InputRequest::at_least(4, 8);
        assert!(!r.is_satisfied_by(&Input::new(4, &bytes[..6])));
        assert!(r.is_satisfied_by(&Input::new(4, &bytes[..8])));
        assert!(r.is_satisfied_by(&Input::new(4, &bytes[..6]).with_total_len(10)));
        assert!(!r.is_satisfied_by(&Input::new(3, &bytes)));
        let r = InputRequest::range(2, 3);
        assert!(r.is_satisfied_by(&Input::new(2, &bytes[..3])));
        assert!(!r.is_satisfied_by(&Input::new(2, &bytes[..2])));
        assert!(InputRequest::Len.is_satisfied_by(&Input::empty().with_total_len(10)));
        assert!(!InputRequest::Len.is_satisfied_by(&Input::empty()));
    }
}

#[cfg(test)]
mod optional_pulse_tests;