zencodec/
source.rs

1//! Random-access byte sources for pull-based decoding.
2//!
3//! [`ReadAt`] is the source a pull-native decoder reads from in straight-line
4//! code: a file (`pread`), an object store (HTTP Range), a host callback
5//! across FFI, or a slice. The codec reads exactly the ranges it needs, at
6//! offsets it chooses, and never learns which transport served them.
7//!
8//! A decoder takes a source through
9//! [`DecodeJob::push_decoder_from_source`](crate::decode::DecodeJob::push_decoder_from_source).
10//! The object-safe form every job receives is [`DynReadAt`]; a concrete
11//! [`ReadAt`] becomes one with [`ReadAtShim`] or `Box::<dyn DynReadAt>::from`.
12//!
13//! Drivers that hold a sequential stream, or run on an async runtime,
14//! manufacture a source with the adapters in `zencodec-io`; the contract
15//! here is deliberately the minimum a codec needs.
16
17use alloc::boxed::Box;
18use core::fmt;
19
20use crate::traits::BoxedError;
21
22/// How a decoder reads its input, declared so drivers can size a buffer or
23/// choose a fetch strategy for unseekable streams.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
25#[non_exhaustive]
26pub enum AccessPattern {
27    /// Reads move forward; any backtrack is incidental and tiny.
28    #[default]
29    Sequential,
30    /// Reads may revisit at most `window_bytes` behind the furthest read.
31    BoundedBacktrack {
32        /// Maximum bytes behind the read watermark a read may target.
33        window_bytes: u64,
34    },
35    /// Reads may target any offset at any time (directories, tile tables,
36    /// trailing indexes).
37    RandomAccess,
38}
39
40impl AccessPattern {
41    /// The more demanding of two patterns (`RandomAccess` over
42    /// `BoundedBacktrack` over `Sequential`; the larger window between two
43    /// bounded ones).
44    pub const fn most_demanding(self, other: Self) -> Self {
45        match (self, other) {
46            (Self::RandomAccess, _) | (_, Self::RandomAccess) => Self::RandomAccess,
47            (
48                Self::BoundedBacktrack { window_bytes: a },
49                Self::BoundedBacktrack { window_bytes: b },
50            ) => Self::BoundedBacktrack {
51                window_bytes: if a > b { a } else { b },
52            },
53            (Self::BoundedBacktrack { window_bytes }, _)
54            | (_, Self::BoundedBacktrack { window_bytes }) => {
55                Self::BoundedBacktrack { window_bytes }
56            }
57            _ => Self::Sequential,
58        }
59    }
60}
61
62/// Failure modes of a byte source.
63#[derive(Debug, Clone, PartialEq, Eq)]
64#[non_exhaustive]
65pub enum SourceError {
66    /// A read targeted bytes already evicted from a windowed buffer.
67    BeyondWindow {
68        /// Offset the codec asked for.
69        offset: u64,
70        /// Oldest offset still retained.
71        oldest_retained: u64,
72    },
73    /// A buffering policy's byte cap would be exceeded.
74    BeyondLimit {
75        /// Bytes the operation would have retained.
76        needed: u64,
77        /// Configured cap.
78        limit: u64,
79    },
80    /// A read ran past the end of the source.
81    UnexpectedEof {
82        /// Offset the codec asked for.
83        offset: u64,
84        /// Bytes requested.
85        len: u64,
86        /// Total source length, if known.
87        total: Option<u64>,
88    },
89    /// The underlying transport or filesystem failed.
90    Io {
91        /// Driver-supplied description of the failure.
92        message: alloc::string::String,
93    },
94}
95
96impl fmt::Display for SourceError {
97    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
98        match self {
99            Self::BeyondWindow {
100                offset,
101                oldest_retained,
102            } => write!(
103                f,
104                "source read at {offset} is behind the buffered window (oldest retained: {oldest_retained})"
105            ),
106            Self::BeyondLimit { needed, limit } => write!(
107                f,
108                "source buffering would retain {needed} bytes, over the {limit}-byte limit"
109            ),
110            Self::UnexpectedEof { offset, len, total } => write!(
111                f,
112                "source read of {len} bytes at {offset} ran past end of source (total: {total:?})"
113            ),
114            Self::Io { message } => write!(f, "source I/O error: {message}"),
115        }
116    }
117}
118
119impl core::error::Error for SourceError {}
120
121/// Random-access byte source. Modelled on imij's `ReadAt`, so an imij-style
122/// codec bridges to this trait by renaming.
123///
124/// `read_many_at` exists so a source that can serve ranges concurrently
125/// (HTTP Range, object stores, parallel `pread`) overlaps their round
126/// trips. Codecs pass every range they can ask for at once.
127///
128/// One decoder operation must observe one coherent byte representation. Bytes
129/// already supplied and any known final length must remain consistent across
130/// reads. A source over a growing stream may discover more bytes while keeping
131/// the supplied prefix unchanged. Replacing an object requires a new operation;
132/// it cannot silently change the meaning of an existing offset or image ID.
133///
134/// Establish this in the source/driver: immutable owned bytes, an immutable
135/// object version, conditional reads that fail on replacement, or a stable-file
136/// ownership/locking contract. Positional filesystem I/O alone is not snapshot
137/// isolation. Transport validators stay outside this trait; source-specific
138/// errors retain the cause when that contract cannot be maintained.
139pub trait ReadAt {
140    /// The source-specific error type.
141    type Error: core::error::Error + Send + Sync + 'static;
142
143    /// Total length in bytes. May be expensive: a buffered stream source
144    /// consumes to the end of the stream to answer it. Callers that only
145    /// read forward use [`read_at`](Self::read_at), which reports the end
146    /// by returning fewer bytes, and never need this.
147    fn len(&mut self) -> Result<u64, Self::Error>;
148
149    /// Total length if it is known without reading ahead. Default `None`.
150    fn len_hint(&self) -> Option<u64> {
151        None
152    }
153
154    /// Whether the source is empty.
155    fn is_empty(&mut self) -> Result<bool, Self::Error> {
156        Ok(self.len()? == 0)
157    }
158
159    /// Fill `out` with the bytes at `offset`. Fails rather than returning
160    /// a short read.
161    fn read_exact_at(&mut self, offset: u64, out: &mut [u8]) -> Result<(), Self::Error>;
162
163    /// Read up to `out.len()` bytes at `offset`; returns the count, which is
164    /// smaller only where the source ends (zero at or past the end).
165    ///
166    /// The default asks [`len`](Self::len) to bound the read; sources over
167    /// streams override it to read forward without consuming to the end.
168    fn read_at(&mut self, offset: u64, out: &mut [u8]) -> Result<usize, Self::Error> {
169        let len = self.len()?;
170        let n = len.saturating_sub(offset).min(out.len() as u64) as usize;
171        self.read_exact_at(offset, &mut out[..n])?;
172        Ok(n)
173    }
174
175    /// Fill each buffer from its offset. The default serves requests in
176    /// the order given; concurrent sources override to overlap them.
177    fn read_many_at(&mut self, reqs: &mut [(u64, &mut [u8])]) -> Result<(), Self::Error> {
178        for (offset, out) in reqs.iter_mut() {
179            self.read_exact_at(*offset, out)?;
180        }
181        Ok(())
182    }
183
184    /// The whole source as one slice, when it is already in memory. Drivers
185    /// and codecs use it to read without copying. Default `None`.
186    fn as_slice(&self) -> Option<&[u8]> {
187        None
188    }
189}
190
191/// [`ReadAt`] over an in-memory slice: the zero-copy source for tests,
192/// embedded data and already-buffered input.
193#[derive(Debug, Clone)]
194pub struct SliceReadAt<'a> {
195    bytes: &'a [u8],
196}
197
198impl<'a> SliceReadAt<'a> {
199    /// Create a source over `bytes`.
200    pub const fn new(bytes: &'a [u8]) -> Self {
201        Self { bytes }
202    }
203
204    fn eof(&self, offset: u64, len: usize) -> SourceError {
205        SourceError::UnexpectedEof {
206            offset,
207            len: len as u64,
208            total: Some(self.bytes.len() as u64),
209        }
210    }
211}
212
213impl ReadAt for SliceReadAt<'_> {
214    type Error = SourceError;
215
216    fn len(&mut self) -> Result<u64, Self::Error> {
217        Ok(self.bytes.len() as u64)
218    }
219
220    fn len_hint(&self) -> Option<u64> {
221        Some(self.bytes.len() as u64)
222    }
223
224    fn read_at(&mut self, offset: u64, out: &mut [u8]) -> Result<usize, Self::Error> {
225        let start = usize::try_from(offset)
226            .unwrap_or(usize::MAX)
227            .min(self.bytes.len());
228        let n = (self.bytes.len() - start).min(out.len());
229        out[..n].copy_from_slice(&self.bytes[start..start + n]);
230        Ok(n)
231    }
232
233    fn as_slice(&self) -> Option<&[u8]> {
234        Some(self.bytes)
235    }
236
237    fn read_exact_at(&mut self, offset: u64, out: &mut [u8]) -> Result<(), Self::Error> {
238        let start = usize::try_from(offset).map_err(|_| self.eof(offset, out.len()))?;
239        let end = start
240            .checked_add(out.len())
241            .ok_or_else(|| self.eof(offset, out.len()))?;
242        let data = self
243            .bytes
244            .get(start..end)
245            .ok_or_else(|| self.eof(offset, out.len()))?;
246        out.copy_from_slice(data);
247        Ok(())
248    }
249}
250
251/// Object-safe [`ReadAt`]. Jobs receive this; codecs wrap its boxed errors
252/// with their own source-error constructor.
253///
254/// A synchronous operation may borrow a thread-local source. Request
255/// `dyn DynReadAt + Send` when the erased source itself crosses threads;
256/// erasure does not move it. Async drivers can instead construct a local
257/// reader on their compute thread and keep transport ownership on the runtime.
258pub trait DynReadAt {
259    /// Total length in bytes (may be expensive; see [`ReadAt::len`]).
260    fn len(&mut self) -> Result<u64, BoxedError>;
261    /// Total length if known without reading ahead.
262    fn len_hint(&self) -> Option<u64> {
263        None
264    }
265    /// Whether the source holds no bytes.
266    fn is_empty(&mut self) -> Result<bool, BoxedError> {
267        Ok(self.len()? == 0)
268    }
269    /// Fill `out` with the bytes at `offset`.
270    fn read_exact_at(&mut self, offset: u64, out: &mut [u8]) -> Result<(), BoxedError>;
271    /// Read up to `out.len()` bytes at `offset`; see [`ReadAt::read_at`].
272    fn read_at(&mut self, offset: u64, out: &mut [u8]) -> Result<usize, BoxedError> {
273        let len = self.len()?;
274        let n = len.saturating_sub(offset).min(out.len() as u64) as usize;
275        self.read_exact_at(offset, &mut out[..n])?;
276        Ok(n)
277    }
278    /// Fill each buffer from its offset, concurrently where possible.
279    fn read_many_at(&mut self, reqs: &mut [(u64, &mut [u8])]) -> Result<(), BoxedError> {
280        for (offset, out) in reqs.iter_mut() {
281            self.read_exact_at(*offset, out)?;
282        }
283        Ok(())
284    }
285    /// The whole source as one slice, when already in memory.
286    fn as_slice(&self) -> Option<&[u8]> {
287        None
288    }
289}
290
291impl<S: DynReadAt + ?Sized> DynReadAt for Box<S> {
292    fn len(&mut self) -> Result<u64, BoxedError> {
293        (**self).len()
294    }
295
296    fn read_exact_at(&mut self, offset: u64, out: &mut [u8]) -> Result<(), BoxedError> {
297        (**self).read_exact_at(offset, out)
298    }
299
300    fn read_many_at(&mut self, reqs: &mut [(u64, &mut [u8])]) -> Result<(), BoxedError> {
301        (**self).read_many_at(reqs)
302    }
303
304    fn len_hint(&self) -> Option<u64> {
305        (**self).len_hint()
306    }
307
308    fn read_at(&mut self, offset: u64, out: &mut [u8]) -> Result<usize, BoxedError> {
309        (**self).read_at(offset, out)
310    }
311
312    fn as_slice(&self) -> Option<&[u8]> {
313        (**self).as_slice()
314    }
315}
316
317impl<S: DynReadAt + ?Sized> DynReadAt for &mut S {
318    fn len(&mut self) -> Result<u64, BoxedError> {
319        (**self).len()
320    }
321
322    fn read_exact_at(&mut self, offset: u64, out: &mut [u8]) -> Result<(), BoxedError> {
323        (**self).read_exact_at(offset, out)
324    }
325
326    fn read_many_at(&mut self, reqs: &mut [(u64, &mut [u8])]) -> Result<(), BoxedError> {
327        (**self).read_many_at(reqs)
328    }
329
330    fn len_hint(&self) -> Option<u64> {
331        (**self).len_hint()
332    }
333
334    fn read_at(&mut self, offset: u64, out: &mut [u8]) -> Result<usize, BoxedError> {
335        (**self).read_at(offset, out)
336    }
337
338    fn as_slice(&self) -> Option<&[u8]> {
339        (**self).as_slice()
340    }
341}
342
343/// Shim presenting a concrete [`ReadAt`] as a [`DynReadAt`].
344pub struct ReadAtShim<S>(pub S);
345
346impl<S: ReadAt> DynReadAt for ReadAtShim<S> {
347    fn len(&mut self) -> Result<u64, BoxedError> {
348        self.0.len().map_err(|e| Box::new(e) as BoxedError)
349    }
350
351    fn read_exact_at(&mut self, offset: u64, out: &mut [u8]) -> Result<(), BoxedError> {
352        self.0
353            .read_exact_at(offset, out)
354            .map_err(|e| Box::new(e) as BoxedError)
355    }
356
357    fn read_many_at(&mut self, reqs: &mut [(u64, &mut [u8])]) -> Result<(), BoxedError> {
358        self.0
359            .read_many_at(reqs)
360            .map_err(|e| Box::new(e) as BoxedError)
361    }
362
363    fn len_hint(&self) -> Option<u64> {
364        self.0.len_hint()
365    }
366
367    fn read_at(&mut self, offset: u64, out: &mut [u8]) -> Result<usize, BoxedError> {
368        self.0
369            .read_at(offset, out)
370            .map_err(|e| Box::new(e) as BoxedError)
371    }
372
373    fn as_slice(&self) -> Option<&[u8]> {
374        self.0.as_slice()
375    }
376}
377
378impl<'a, S: ReadAt + 'a> From<S> for Box<dyn DynReadAt + 'a> {
379    fn from(source: S) -> Self {
380        Box::new(ReadAtShim(source))
381    }
382}
383
384impl<'a, S: ReadAt + Send + 'a> From<S> for Box<dyn DynReadAt + Send + 'a> {
385    fn from(source: S) -> Self {
386        Box::new(ReadAtShim(source))
387    }
388}
389
390#[cfg(test)]
391mod tests {
392    use super::*;
393
394    #[test]
395    fn slice_reads_and_fails_past_end() {
396        let data = [1u8, 2, 3, 4, 5];
397        let mut src = SliceReadAt::new(&data);
398        assert_eq!(src.len().unwrap(), 5);
399        let mut two = [0u8; 2];
400        src.read_exact_at(3, &mut two).unwrap();
401        assert_eq!(two, [4, 5]);
402        let mut three = [0u8; 3];
403        assert!(matches!(
404            src.read_exact_at(3, &mut three),
405            Err(SourceError::UnexpectedEof {
406                offset: 3,
407                len: 3,
408                total: Some(5)
409            })
410        ));
411        assert!(src.read_exact_at(u64::MAX, &mut two).is_err());
412    }
413
414    #[test]
415    fn shim_and_box_dispatch() {
416        static DATA: [u8; 3] = [9, 8, 7];
417        let mut boxed: Box<dyn DynReadAt> = SliceReadAt::new(&DATA).into();
418        let mut a = [0u8; 1];
419        let mut b = [0u8; 2];
420        boxed
421            .read_many_at(&mut [(2, &mut a[..]), (0, &mut b[..])])
422            .unwrap();
423        assert_eq!((a, b), ([7], [9, 8]));
424        let by_ref: &mut dyn DynReadAt = &mut *boxed;
425        assert_eq!(by_ref.len().unwrap(), 3);
426    }
427}
zencodec/source.rs

zencodec/
source.rs
