coven_database/
input_bytes.rs

1//! Local bounded reads of retained provider bytes during an apply transaction.
2
3use super::{read_local, sql_offset, InputConnection, InputRead, RANGES};
4use std::io::{self, Read, Seek, SeekFrom};
5use std::sync::Arc;
6
7/// Independent cursor over one pinned input. Closing the database invalidates
8/// every cursor; an absent cached range returns an I/O error, never a network read.
9#[derive(Clone)]
10pub struct RetainedBytes {
11    reader: InputConnection,
12    reading: Arc<InputRead>,
13    position: u64,
14    length: u64,
15}
16
17impl RetainedBytes {
18    pub(super) fn new(reader: InputConnection, reading: Arc<InputRead>, length: u64) -> Self {
19        Self { reader, reading, position: 0, length }
20    }
21
22    /// The received object's declared length.
23    pub fn len(&self) -> u64 {
24        self.length
25    }
26
27    /// Whether this object has no bytes.
28    pub fn is_empty(&self) -> bool {
29        self.length == 0
30    }
31}
32
33impl Read for RetainedBytes {
34    fn read(&mut self, output: &mut [u8]) -> io::Result<usize> {
35        let length = self
36            .length
37            .saturating_sub(self.position)
38            .min(output.len() as u64)
39            .min(coven_format::chunks::CHUNK_SIZE as u64);
40        if length == 0 {
41            return Ok(0);
42        }
43        let start = self.position;
44        let end = start + length;
45        let ranges = read_local(&self.reader, |db| {
46            db.query(RANGES, (self.reading.id, sql_offset(start)?, sql_offset(end)?), |row| {
47                Ok((row.get::<_, i64>(0)? as u64, row.get::<_, Vec<u8>>(1)?))
48            })
49        })
50        .map_err(io::Error::other)?;
51        let mut copied = 0;
52        for (offset, bytes) in ranges {
53            if offset != start + copied as u64 {
54                break;
55            }
56            output[copied..copied + bytes.len()].copy_from_slice(&bytes);
57            copied += bytes.len();
58        }
59        if copied == 0 {
60            return Err(io::Error::new(
61                io::ErrorKind::UnexpectedEof,
62                "provider bytes have not been received",
63            ));
64        }
65        #[cfg(any(test, feature = "test-utils"))]
66        coven_foundation::test_work::retained_read(copied);
67        self.position += copied as u64;
68        Ok(copied)
69    }
70}
71
72impl Seek for RetainedBytes {
73    fn seek(&mut self, from: SeekFrom) -> io::Result<u64> {
74        let position = match from {
75            SeekFrom::Start(position) => Some(position),
76            SeekFrom::End(offset) => self.length.checked_add_signed(offset),
77            SeekFrom::Current(offset) => self.position.checked_add_signed(offset),
78        }
79        .ok_or_else(|| {
80            io::Error::new(io::ErrorKind::InvalidInput, "input position is outside u64")
81        })?;
82        self.position = position;
83        Ok(position)
84    }
85    fn stream_position(&mut self) -> io::Result<u64> {
86        Ok(self.position)
87    }
88}
coven_database/input_bytes.rs

coven_database/
input_bytes.rs
