Skip to main content

otf_pixels_core/
source.rs

1//! Producers that root a graph in actual pixels.
2//!
3//! [`BufferSource`] wraps pixels the caller already holds. [`DecodedSource`]
4//! bridges a streaming [`Decoder`] into the graph, deferring all pixel work
5//! until a terminal pulls.
6
7use crate::{
8    DecodeCapability, Decoder, ImageDescriptor, PixelsError, Producer, Region, Result, TileBuf,
9    TileMut, copy_region,
10};
11use std::sync::{Arc, Mutex};
12
13/// A producer over pixels already in memory.
14///
15/// The buffer is shared, not copied, so cloning is cheap and two graphs can
16/// root at the same pixels.
17#[derive(Debug, Clone)]
18pub struct BufferSource {
19    buffer: Arc<TileBuf>,
20    descriptor: ImageDescriptor,
21}
22
23impl BufferSource {
24    /// Wrap a whole-image buffer as a producer.
25    ///
26    /// # Errors
27    ///
28    /// Returns [`PixelsError::InvalidArgument`] if `buffer` does not start at
29    /// the image origin or does not match `descriptor`.
30    pub fn new(descriptor: ImageDescriptor, buffer: Arc<TileBuf>) -> Result<Self> {
31        if buffer.region() != descriptor.region() {
32            return Err(PixelsError::invalid_argument(
33                "buffer",
34                format!(
35                    "buffer covers {} but the image is {}",
36                    buffer.region(),
37                    descriptor.region()
38                ),
39            ));
40        }
41        if buffer.pixel() != descriptor.pixel {
42            return Err(PixelsError::invalid_argument(
43                "buffer",
44                format!(
45                    "buffer is {} but the image is {}",
46                    buffer.pixel(),
47                    descriptor.pixel
48                ),
49            ));
50        }
51        Ok(Self { buffer, descriptor })
52    }
53
54    /// The pixels backing this producer.
55    #[must_use]
56    pub fn buffer(&self) -> &Arc<TileBuf> {
57        &self.buffer
58    }
59}
60
61impl Producer for BufferSource {
62    fn name(&self) -> &'static str {
63        "buffer"
64    }
65
66    fn descriptor(&self) -> ImageDescriptor {
67        self.descriptor
68    }
69
70    fn capability(&self) -> DecodeCapability {
71        // The pixels are already in memory, so any region is equally cheap.
72        // This is what lets `flip` over a memory buffer stream rather than
73        // materialize (ADR-0009).
74        DecodeCapability::Regions
75    }
76
77    fn produce(&self, region: Region, output: &mut TileMut<'_>) -> Result<()> {
78        let tile = self.buffer.as_tile()?;
79        copy_region(&tile, output, region)
80    }
81}
82
83/// The decode state of a [`DecodedSource`].
84#[derive(Debug)]
85enum DecodeState {
86    /// Header parsed; `cursor` rows have been consumed from the stream.
87    Reading {
88        /// The decoder, positioned at row `cursor`.
89        decoder: Box<dyn Decoder>,
90        /// The next row the decoder will emit.
91        cursor: u32,
92    },
93    /// Decoding failed. The stream is consumed and cannot be retried, so the
94    /// failure is remembered and replayed rather than producing partial pixels.
95    Failed(String),
96}
97
98/// A producer that decodes a stream on demand, forward only.
99///
100/// Construction parses the header only — no pixel bytes are read — so
101/// [`Image::metadata`] stays free and laziness holds (SPEC §Guarantees 3).
102///
103/// # Memory
104///
105/// Constant, and independent of image height: the decoder is advanced to the
106/// requested band and exactly that band is retained as a rolling window. This
107/// is what makes SPEC §Guarantees 1 true for streaming formats — the whole
108/// image is never resident.
109///
110/// The window also absorbs *repeated* demand for the same band, which is what
111/// lets two graph branches share one source without either re-reading a stream
112/// that has already moved on.
113///
114/// # Forward only
115///
116/// A request that starts before the window is an error, not a rewind: the
117/// bytes are gone (ADR-0005). The scheduler is responsible for never asking —
118/// [`Plan`] detects non-forward demand ahead of time and materializes instead
119/// (ADR-0009). Reaching this error therefore indicates a scheduling defect, and
120/// it is reported rather than papered over.
121///
122/// [`Image::metadata`]: crate::Image::metadata
123/// [`Plan`]: crate::Plan
124#[derive(Debug)]
125pub struct DecodedSource {
126    /// Behind a lock because shrink-on-load can lower it once, before any
127    /// pixel is produced. Everything else about a source is fixed at
128    /// construction; this is the one thing the planner may still change.
129    descriptor: Mutex<ImageDescriptor>,
130    capability: DecodeCapability,
131    state: Mutex<DecodeState>,
132    /// The most recently decoded band, reused for repeated demand.
133    window: Mutex<Option<(Region, Arc<TileBuf>)>>,
134}
135
136impl DecodedSource {
137    /// Wrap a decoder whose header has been parsed.
138    ///
139    /// No pixel rows are read here.
140    #[must_use]
141    pub fn new(decoder: Box<dyn Decoder>) -> Self {
142        Self {
143            descriptor: Mutex::new(decoder.descriptor()),
144            capability: decoder.capability(),
145            state: Mutex::new(DecodeState::Reading { decoder, cursor: 0 }),
146            window: Mutex::new(None),
147        }
148    }
149
150    /// The shape currently being produced.
151    ///
152    /// A poisoned lock still holds a valid descriptor — the value is `Copy`
153    /// and is never left half-written — so the poison is stepped over rather
154    /// than turned into an error the caller cannot act on.
155    fn shape(&self) -> ImageDescriptor {
156        *self
157            .descriptor
158            .lock()
159            .unwrap_or_else(std::sync::PoisonError::into_inner)
160    }
161
162    /// Decode rows `band.y .. band.bottom()` into a fresh buffer.
163    ///
164    /// Rows before `band.y` are decoded and discarded, since a forward-only
165    /// stream has no way to skip them.
166    ///
167    /// # Overlapping bands
168    ///
169    /// Consecutive requests routinely *overlap* rather than abut: a resize
170    /// needs input rows either side of each output row for filter support, so
171    /// it asks for `[0, 224)` and then `[214, 438)`. The ten rows in common
172    /// have already been consumed from the stream and cannot be read again, so
173    /// they are carried over from the retained band instead. Without that,
174    /// every streaming decode resized to more than one tile column failed with
175    /// a rewind error — which is most real resizes.
176    fn decode_band(&self, band: Region) -> Result<Arc<TileBuf>> {
177        let mut state = self
178            .state
179            .lock()
180            .map_err(|_| PixelsError::graph("decoder state was poisoned by a panicking thread"))?;
181
182        if let DecodeState::Failed(detail) = &*state {
183            return Err(PixelsError::malformed("stream", detail.clone()));
184        }
185        let DecodeState::Reading { decoder, cursor } = &mut *state else {
186            return Err(PixelsError::graph("decoder state changed unexpectedly"));
187        };
188
189        // Rows already consumed have to come from the retained band, if it
190        // still holds them.
191        let carry = if band.y < *cursor {
192            let held = self
193                .window
194                .lock()
195                .ok()
196                .and_then(|window| window.clone())
197                .filter(|(covered, _)| covered.y <= band.y);
198            match held {
199                Some((_, buffer)) => Some(buffer),
200                None => {
201                    return Err(PixelsError::graph(format!(
202                        "source cannot rewind: row {} was requested but the stream is at row \
203                         {cursor} and the retained band no longer covers it; this pipeline \
204                         needed a materialization point",
205                        band.y
206                    )));
207                }
208            }
209        } else {
210            None
211        };
212
213        let outcome = Self::fill_band(
214            decoder.as_mut(),
215            cursor,
216            band,
217            self.shape(),
218            carry.as_deref(),
219        );
220        match outcome {
221            Ok(buffer) => Ok(Arc::new(buffer)),
222            Err(error) => {
223                *state = DecodeState::Failed(error.to_string());
224                Err(error)
225            }
226        }
227    }
228
229    /// Advance `decoder` to `band` and read it, updating `cursor`.
230    ///
231    /// Rows below `cursor` are copied from `carry` — the previously retained
232    /// band — because the stream is already past them.
233    fn fill_band(
234        decoder: &mut dyn Decoder,
235        cursor: &mut u32,
236        band: Region,
237        descriptor: ImageDescriptor,
238        carry: Option<&TileBuf>,
239    ) -> Result<TileBuf> {
240        let mut scratch = vec![0_u8; descriptor.row_bytes()];
241        while *cursor < band.y {
242            decoder.read_row(&mut scratch)?;
243            *cursor += 1;
244        }
245        let mut buffer = TileBuf::zeroed(band, descriptor.pixel)?;
246        {
247            let mut tile = buffer.as_tile_mut()?;
248            let carried = carry.map(TileBuf::as_tile).transpose()?;
249            for y in band.y..band.y.saturating_add(band.height) {
250                let row = tile.row_mut(y).ok_or_else(|| {
251                    PixelsError::malformed("stream", format!("row {y} is outside the band"))
252                })?;
253                if y < *cursor {
254                    // Already consumed from the stream; the retained band is
255                    // the only place these rows still exist.
256                    let source =
257                        carried
258                            .as_ref()
259                            .and_then(|tile| tile.row(y))
260                            .ok_or_else(|| {
261                                PixelsError::graph(format!(
262                                    "row {y} is behind the stream and outside the retained band"
263                                ))
264                            })?;
265                    let take = row.len().min(source.len());
266                    row.get_mut(..take)
267                        .zip(source.get(..take))
268                        .map(|(target, source)| target.copy_from_slice(source))
269                        .ok_or_else(|| {
270                            PixelsError::graph(format!("row {y} could not be carried over"))
271                        })?;
272                    continue;
273                }
274                decoder.read_row(row)?;
275                *cursor += 1;
276            }
277        }
278        Ok(buffer)
279    }
280
281    /// Ask the decoder for `region` directly, for a random-access decoder.
282    ///
283    /// This is the path that makes [`DecodeCapability::Regions`] mean
284    /// something. Without it a tiled TIFF would be declared random-access and
285    /// then driven row by row anyway, so a thumbnail of a huge scan would
286    /// decode every tile rather than the handful it needs.
287    ///
288    /// The rolling window is bypassed entirely: a decoder that can seek has no
289    /// use for one, and keeping a band would reintroduce the memory the
290    /// random access exists to avoid.
291    fn produce_region(&self, region: Region, output: &mut TileMut<'_>) -> Result<()> {
292        let mut state = self
293            .state
294            .lock()
295            .map_err(|_| PixelsError::graph("decoder state was poisoned by a panicking thread"))?;
296
297        if let DecodeState::Failed(detail) = &*state {
298            return Err(PixelsError::malformed("stream", detail.clone()));
299        }
300        let DecodeState::Reading { decoder, .. } = &mut *state else {
301            return Err(PixelsError::graph("decoder state changed unexpectedly"));
302        };
303
304        match decoder.read_region(region, output) {
305            Ok(()) => Ok(()),
306            Err(error) => {
307                // A region decoder that fails has no defined cursor to
308                // recover to, so the source is poisoned rather than left in a
309                // state a later pull would misread.
310                *state = DecodeState::Failed(error.to_string());
311                Err(error)
312            }
313        }
314    }
315
316    /// The band covering `region`, from the window or freshly decoded.
317    fn band_for(&self, region: Region) -> Result<Arc<TileBuf>> {
318        {
319            let window = self
320                .window
321                .lock()
322                .map_err(|_| PixelsError::graph("source window was poisoned"))?;
323            if let Some((covered, buffer)) = &*window {
324                if covered.contains(region) {
325                    return Ok(Arc::clone(buffer));
326                }
327            }
328        }
329        // Decode full-width bands: a forward-only stream produces whole rows
330        // anyway, so narrowing would discard pixels a later request may want.
331        let band = Region::new(0, region.y, self.shape().width, region.height);
332        let buffer = self.decode_band(band)?;
333        if let Ok(mut window) = self.window.lock() {
334            *window = Some((band, Arc::clone(&buffer)));
335        }
336        Ok(buffer)
337    }
338}
339
340impl Producer for DecodedSource {
341    fn name(&self) -> &'static str {
342        "decoded"
343    }
344
345    fn descriptor(&self) -> ImageDescriptor {
346        self.shape()
347    }
348
349    fn capability(&self) -> DecodeCapability {
350        self.capability
351    }
352
353    fn reduced_descriptor(&self, target: (u32, u32)) -> Option<ImageDescriptor> {
354        let state = self.state.lock().ok()?;
355        let DecodeState::Reading { decoder, cursor } = &*state else {
356            return None;
357        };
358        // Past the first row the resolution is settled: rows already handed
359        // out cannot be retracted.
360        if *cursor > 0 {
361            return None;
362        }
363        decoder.reduced_descriptor(target)
364    }
365
366    fn reduce_to(&self, descriptor: ImageDescriptor) -> Result<()> {
367        let mut state = self
368            .state
369            .lock()
370            .map_err(|_| PixelsError::graph("decoder state was poisoned by a panicking thread"))?;
371        let DecodeState::Reading { decoder, cursor } = &mut *state else {
372            return Err(PixelsError::graph("decoder state changed unexpectedly"));
373        };
374        if *cursor > 0 {
375            return Err(PixelsError::invalid_argument(
376                "descriptor",
377                format!("{cursor} rows have already been produced; the resolution is settled"),
378            ));
379        }
380        decoder.reduce_to(descriptor)?;
381
382        let mut shape = self
383            .descriptor
384            .lock()
385            .map_err(|_| PixelsError::graph("descriptor was poisoned by a panicking thread"))?;
386        *shape = descriptor;
387        // Any band cached at the old resolution describes an image that no
388        // longer exists.
389        if let Ok(mut window) = self.window.lock() {
390            *window = None;
391        }
392        Ok(())
393    }
394
395    fn produce(&self, region: Region, output: &mut TileMut<'_>) -> Result<()> {
396        // A decoder that can answer regions is asked for one. Anything else
397        // gets the forward-only rolling window.
398        if self.capability == DecodeCapability::Regions {
399            return self.produce_region(region, output);
400        }
401        let buffer = self.band_for(region)?;
402        let tile = buffer.as_tile()?;
403        copy_region(&tile, output, region)
404    }
405}
406
407#[cfg(test)]
408#[allow(
409    clippy::unwrap_used,
410    clippy::indexing_slicing,
411    reason = "tests operate on known-good values and assert shapes directly"
412)]
413mod tests {
414    use super::*;
415    use crate::{ErrorCode, PixelFormat};
416
417    /// A decoder that yields `height` rows of an increasing counter, then
418    /// optionally fails at row `fail_at`.
419    #[derive(Debug)]
420    struct StubDecoder {
421        descriptor: ImageDescriptor,
422        row: u32,
423        fail_at: Option<u32>,
424        rows_read: Arc<std::sync::atomic::AtomicU32>,
425    }
426
427    impl Decoder for StubDecoder {
428        fn descriptor(&self) -> ImageDescriptor {
429            self.descriptor
430        }
431        fn read_row(&mut self, out: &mut [u8]) -> Result<()> {
432            if Some(self.row) == self.fail_at {
433                return Err(PixelsError::malformed("stub", "corrupt row"));
434            }
435            self.rows_read
436                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
437            out.fill(self.row as u8);
438            self.row += 1;
439            Ok(())
440        }
441    }
442
443    /// A stub of `width` x `height` where every row holds its own index.
444    fn sized_stub(width: u32, height: u32) -> (DecodedSource, Arc<std::sync::atomic::AtomicU32>) {
445        let rows_read = Arc::new(std::sync::atomic::AtomicU32::new(0));
446        let decoder = StubDecoder {
447            descriptor: ImageDescriptor::new(width, height, PixelFormat::Gray8).unwrap(),
448            row: 0,
449            fail_at: None,
450            rows_read: Arc::clone(&rows_read),
451        };
452        (DecodedSource::new(Box::new(decoder)), rows_read)
453    }
454
455    /// Read `region` and return its rows as their first byte.
456    fn rows_of(source: &DecodedSource, region: Region) -> Vec<u8> {
457        let mut buffer = TileBuf::zeroed(region, PixelFormat::Gray8).unwrap();
458        {
459            let mut tile = buffer.as_tile_mut().unwrap();
460            source.produce(region, &mut tile).unwrap();
461        }
462        let tile = buffer.as_tile().unwrap();
463        (region.y..region.y + region.height)
464            .map(|y| tile.row(y).unwrap()[0])
465            .collect()
466    }
467
468    /// A resize asks for bands that overlap, because each output row needs
469    /// input rows either side of it. The overlapping rows are already past the
470    /// stream cursor, so they have to be carried over from the retained band —
471    /// and they have to be the *right* rows.
472    #[test]
473    fn overlapping_bands_carry_the_rows_already_consumed() {
474        let (source, rows_read) = sized_stub(2, 16);
475
476        assert_eq!(
477            rows_of(&source, Region::new(0, 0, 2, 6)),
478            [0, 1, 2, 3, 4, 5]
479        );
480        assert_eq!(rows_read.load(std::sync::atomic::Ordering::Relaxed), 6);
481
482        // Overlaps the first band by two rows: 4 and 5 are behind the cursor.
483        assert_eq!(
484            rows_of(&source, Region::new(0, 4, 2, 6)),
485            [4, 5, 6, 7, 8, 9],
486            "the carried rows are wrong"
487        );
488        // Only the four new rows were decoded; the overlap was not re-read,
489        // which a forward-only stream could not have done anyway.
490        assert_eq!(rows_read.load(std::sync::atomic::Ordering::Relaxed), 10);
491    }
492
493    /// Overlap is servable; going back further than the retained band is not.
494    #[test]
495    fn a_request_behind_the_retained_band_is_still_an_error() {
496        let (source, _) = sized_stub(2, 16);
497        assert_eq!(rows_of(&source, Region::new(0, 8, 2, 4)), [8, 9, 10, 11]);
498
499        let region = Region::new(0, 2, 2, 4);
500        let mut buffer = TileBuf::zeroed(region, PixelFormat::Gray8).unwrap();
501        let mut tile = buffer.as_tile_mut().unwrap();
502        let error = source.produce(region, &mut tile).unwrap_err();
503        assert_eq!(error.code(), ErrorCode::Graph, "{error}");
504    }
505
506    fn stub(fail_at: Option<u32>) -> (DecodedSource, Arc<std::sync::atomic::AtomicU32>) {
507        let rows_read = Arc::new(std::sync::atomic::AtomicU32::new(0));
508        let decoder = StubDecoder {
509            descriptor: ImageDescriptor::new(2, 3, PixelFormat::Gray8).unwrap(),
510            row: 0,
511            fail_at,
512            rows_read: Arc::clone(&rows_read),
513        };
514        (DecodedSource::new(Box::new(decoder)), rows_read)
515    }
516
517    /// A 2x64 source, tall enough that a band is much smaller than the image.
518    fn tall_stub() -> (DecodedSource, Arc<std::sync::atomic::AtomicU32>) {
519        let rows_read = Arc::new(std::sync::atomic::AtomicU32::new(0));
520        let decoder = StubDecoder {
521            descriptor: ImageDescriptor::new(2, 64, PixelFormat::Gray8).unwrap(),
522            row: 0,
523            fail_at: None,
524            rows_read: Arc::clone(&rows_read),
525        };
526        (DecodedSource::new(Box::new(decoder)), rows_read)
527    }
528
529    #[test]
530    fn construction_reads_no_pixel_rows() {
531        let (source, rows_read) = stub(None);
532        assert_eq!(source.descriptor().width, 2);
533        assert_eq!(rows_read.load(std::sync::atomic::Ordering::Relaxed), 0);
534    }
535
536    #[test]
537    fn the_stream_is_decoded_once_and_reused() {
538        let (source, rows_read) = stub(None);
539        let mut out = TileBuf::zeroed(Region::from_size(2, 3), PixelFormat::Gray8).unwrap();
540        for _ in 0..3 {
541            let mut tile = out.as_tile_mut().unwrap();
542            source.produce(Region::from_size(2, 3), &mut tile).unwrap();
543        }
544        assert_eq!(
545            rows_read.load(std::sync::atomic::Ordering::Relaxed),
546            3,
547            "decoded exactly once"
548        );
549        assert_eq!(out.bytes(), &[0, 0, 1, 1, 2, 2]);
550    }
551
552    #[test]
553    fn a_partial_region_pull_returns_just_that_region() {
554        let (source, _) = stub(None);
555        let mut out = TileBuf::zeroed(Region::new(0, 1, 2, 1), PixelFormat::Gray8).unwrap();
556        let mut tile = out.as_tile_mut().unwrap();
557        source.produce(Region::new(0, 1, 2, 1), &mut tile).unwrap();
558        assert_eq!(out.bytes(), &[1, 1]);
559    }
560
561    #[test]
562    fn a_decode_failure_is_sticky_and_never_partial() {
563        let (source, _) = stub(Some(1));
564        let mut out = TileBuf::zeroed(Region::from_size(2, 3), PixelFormat::Gray8).unwrap();
565        for attempt in 0..2 {
566            let mut tile = out.as_tile_mut().unwrap();
567            let err = source
568                .produce(Region::from_size(2, 3), &mut tile)
569                .unwrap_err();
570            assert_eq!(err.code(), ErrorCode::Malformed, "attempt {attempt}");
571        }
572    }
573
574    #[test]
575    fn decoding_advances_only_as_far_as_demanded() {
576        // The constant-memory property: asking for an early band must not
577        // decode the rest of the image.
578        let (source, rows_read) = tall_stub();
579        let mut out = TileBuf::zeroed(Region::new(0, 0, 2, 4), PixelFormat::Gray8).unwrap();
580        let mut tile = out.as_tile_mut().unwrap();
581        source.produce(Region::new(0, 0, 2, 4), &mut tile).unwrap();
582        assert_eq!(
583            rows_read.load(std::sync::atomic::Ordering::Relaxed),
584            4,
585            "decoded past the requested band"
586        );
587    }
588
589    #[test]
590    fn successive_bands_stream_forward() {
591        let (source, rows_read) = tall_stub();
592        for start in (0..64).step_by(4) {
593            let band = Region::new(0, start, 2, 4);
594            let mut out = TileBuf::zeroed(band, PixelFormat::Gray8).unwrap();
595            let mut tile = out.as_tile_mut().unwrap();
596            source.produce(band, &mut tile).unwrap();
597            // Each row of the ramp holds its own index.
598            assert_eq!(out.bytes()[0], start as u8, "band at row {start} is wrong");
599        }
600        assert_eq!(
601            rows_read.load(std::sync::atomic::Ordering::Relaxed),
602            64,
603            "rows re-read"
604        );
605    }
606
607    #[test]
608    fn repeated_demand_for_a_band_is_served_from_the_window() {
609        // Two graph branches pulling the same band must not re-read a stream
610        // that has already moved on.
611        let (source, rows_read) = tall_stub();
612        let band = Region::new(0, 0, 2, 4);
613        for _ in 0..5 {
614            let mut out = TileBuf::zeroed(band, PixelFormat::Gray8).unwrap();
615            let mut tile = out.as_tile_mut().unwrap();
616            source.produce(band, &mut tile).unwrap();
617        }
618        assert_eq!(
619            rows_read.load(std::sync::atomic::Ordering::Relaxed),
620            4,
621            "band was re-decoded"
622        );
623    }
624
625    #[test]
626    fn a_sub_band_of_the_window_is_served_without_re_reading() {
627        let (source, rows_read) = tall_stub();
628        let band = Region::new(0, 0, 2, 8);
629        let mut out = TileBuf::zeroed(band, PixelFormat::Gray8).unwrap();
630        source
631            .produce(band, &mut out.as_tile_mut().unwrap())
632            .unwrap();
633
634        let inner = Region::new(0, 2, 2, 2);
635        let mut small = TileBuf::zeroed(inner, PixelFormat::Gray8).unwrap();
636        source
637            .produce(inner, &mut small.as_tile_mut().unwrap())
638            .unwrap();
639        assert_eq!(small.bytes(), &[2, 2, 3, 3]);
640        assert_eq!(rows_read.load(std::sync::atomic::Ordering::Relaxed), 8);
641    }
642
643    #[test]
644    fn rewinding_is_a_reported_error_not_silent_corruption() {
645        // A forward-only stream cannot go back (ADR-0005). Reaching this means
646        // the plan failed to insert a materialization point (ADR-0009), so it
647        // is surfaced rather than papered over.
648        let (source, _) = tall_stub();
649        let later = Region::new(0, 16, 2, 4);
650        let mut out = TileBuf::zeroed(later, PixelFormat::Gray8).unwrap();
651        source
652            .produce(later, &mut out.as_tile_mut().unwrap())
653            .unwrap();
654
655        let earlier = Region::new(0, 0, 2, 4);
656        let mut back = TileBuf::zeroed(earlier, PixelFormat::Gray8).unwrap();
657        let err = source
658            .produce(earlier, &mut back.as_tile_mut().unwrap())
659            .unwrap_err();
660        assert_eq!(err.code(), ErrorCode::Graph);
661        assert!(err.to_string().contains("rewind"), "{err}");
662        assert!(err.to_string().contains("materialization"), "{err}");
663    }
664
665    #[test]
666    fn skipped_rows_are_consumed_not_lost() {
667        // Jumping forward past a band still has to read those bytes, because a
668        // forward-only stream cannot seek.
669        let (source, rows_read) = tall_stub();
670        let band = Region::new(0, 32, 2, 4);
671        let mut out = TileBuf::zeroed(band, PixelFormat::Gray8).unwrap();
672        source
673            .produce(band, &mut out.as_tile_mut().unwrap())
674            .unwrap();
675        assert_eq!(rows_read.load(std::sync::atomic::Ordering::Relaxed), 36);
676        assert_eq!(out.bytes()[0], 32, "landed on the wrong row");
677    }
678
679    #[test]
680    fn producers_report_their_capability() {
681        // The upstream half of ADR-0009's analysis.
682        let (source, _) = tall_stub();
683        assert_eq!(source.capability(), DecodeCapability::Sequential);
684
685        let descriptor = ImageDescriptor::new(2, 2, PixelFormat::Gray8).unwrap();
686        let buffer = Arc::new(TileBuf::for_image(&descriptor).unwrap());
687        let buffered = BufferSource::new(descriptor, buffer).unwrap();
688        assert_eq!(buffered.capability(), DecodeCapability::Regions);
689    }
690
691    #[test]
692    fn buffer_source_validates_against_its_descriptor() {
693        let desc = ImageDescriptor::new(2, 2, PixelFormat::Gray8).unwrap();
694        let good = Arc::new(TileBuf::zeroed(desc.region(), PixelFormat::Gray8).unwrap());
695        assert!(BufferSource::new(desc, good).is_ok());
696
697        let wrong_size =
698            Arc::new(TileBuf::zeroed(Region::from_size(3, 3), PixelFormat::Gray8).unwrap());
699        assert_eq!(
700            BufferSource::new(desc, wrong_size).unwrap_err().code(),
701            ErrorCode::InvalidArgument
702        );
703
704        let wrong_format = Arc::new(TileBuf::zeroed(desc.region(), PixelFormat::Rgb8).unwrap());
705        assert_eq!(
706            BufferSource::new(desc, wrong_format).unwrap_err().code(),
707            ErrorCode::InvalidArgument
708        );
709    }
710
711    #[test]
712    fn buffer_source_serves_regions() {
713        let desc = ImageDescriptor::new(2, 2, PixelFormat::Gray8).unwrap();
714        let buffer = Arc::new(
715            TileBuf::from_vec(desc.region(), PixelFormat::Gray8, vec![1, 2, 3, 4]).unwrap(),
716        );
717        let source = BufferSource::new(desc, buffer).unwrap();
718        let mut out = TileBuf::zeroed(Region::new(1, 0, 1, 2), PixelFormat::Gray8).unwrap();
719        let mut tile = out.as_tile_mut().unwrap();
720        source.produce(Region::new(1, 0, 1, 2), &mut tile).unwrap();
721        assert_eq!(out.bytes(), &[2, 4]);
722    }
723}