molframe_core/io/input/
spill_source.rs

1//! Random-access windows over fixed-record execution-owned spill storage.
2
3use super::{ByteWindow, SourceBytes};
4use crate::execution::spill::{FILE_HEADER_BYTES, RECORD_HEADER_DISK_BYTES, checksum};
5use crate::{Code, Diagnostic, ExecutionContext, SpillArtifact};
6use std::fs::File;
7use std::io::{Read, Seek, SeekFrom};
8
9const HEADER_BYTES: usize = 16;
10
11/// A bounded source backed by an immutable, automatically deleted spill file.
12#[derive(Debug)]
13pub struct SpillWindowedFile {
14    file: File,
15    artifact: SpillArtifact,
16    record_bytes: usize,
17    length: u64,
18    output: Vec<u8>,
19    record: Vec<u8>,
20    cached_record: Option<u64>,
21    _reservation: crate::MemoryReservation,
22}
23
24impl SpillWindowedFile {
25    /// Opens fixed-size records as one logical byte stream.
26    ///
27    /// # Errors
28    ///
29    /// Returns a resource diagnostic if buffers, the artifact, or its record
30    /// geometry are invalid.
31    pub fn open(
32        artifact: SpillArtifact,
33        record_bytes: usize,
34        length: u64,
35        max_window_bytes: usize,
36        context: &ExecutionContext,
37    ) -> Result<Self, Diagnostic> {
38        if record_bytes == 0 || max_window_bytes == 0 {
39            return Err(resource_error("spill source buffers must be non-zero"));
40        }
41        let retained = record_bytes
42            .checked_add(max_window_bytes)
43            .ok_or_else(|| resource_error("spill source buffer size overflow"))?;
44        let reservation = context
45            .try_reserve(retained)
46            .map_err(|error| resource_error(error.to_string()))?;
47        let file = File::open(artifact.path()).map_err(|error| io_error("open", &error))?;
48        let mut output = Vec::new();
49        output
50            .try_reserve_exact(max_window_bytes)
51            .map_err(|error| resource_error(error.to_string()))?;
52        let mut record = Vec::new();
53        record
54            .try_reserve_exact(record_bytes)
55            .map_err(|error| resource_error(error.to_string()))?;
56        Ok(Self {
57            file,
58            artifact,
59            record_bytes,
60            length,
61            output,
62            record,
63            cached_record: None,
64            _reservation: reservation,
65        })
66    }
67
68    fn load_record(&mut self, index: u64) -> Result<(), Diagnostic> {
69        if self.cached_record == Some(index) {
70            return Ok(());
71        }
72        let stride = u64::try_from(self.record_bytes)
73            .map_err(|_| resource_error("spill record size exceeds u64"))?
74            .checked_add(RECORD_HEADER_DISK_BYTES)
75            .ok_or_else(|| resource_error("spill record stride overflow"))?;
76        let offset = index
77            .checked_mul(stride)
78            .and_then(|value| value.checked_add(FILE_HEADER_BYTES))
79            .ok_or_else(|| resource_error("spill record offset overflow"))?;
80        self.file
81            .seek(SeekFrom::Start(offset))
82            .map_err(|error| io_error("seek", &error))?;
83        let mut header = [0_u8; HEADER_BYTES];
84        self.file
85            .read_exact(&mut header)
86            .map_err(|error| io_error("read header", &error))?;
87        let length = u64::from_le_bytes(copy_u64(&header[..8]));
88        let expected = u64::from_le_bytes(copy_u64(&header[8..]));
89        let maximum = u64::try_from(self.record_bytes)
90            .map_err(|_| resource_error("spill record size exceeds u64"))?;
91        if length > maximum {
92            return Err(resource_error("spill record exceeds configured width"));
93        }
94        let length =
95            usize::try_from(length).map_err(|_| resource_error("spill record exceeds usize"))?;
96        self.record.resize(length, 0);
97        self.file
98            .read_exact(&mut self.record)
99            .map_err(|error| io_error("read payload", &error))?;
100        if checksum(&self.record) != expected {
101            return Err(resource_error("spill record checksum mismatch"));
102        }
103        self.cached_record = Some(index);
104        Ok(())
105    }
106}
107
108impl SourceBytes for SpillWindowedFile {
109    fn window(&mut self, start: u64, len: usize) -> Result<ByteWindow<'_>, Diagnostic> {
110        if len > self.output.capacity() {
111            return Err(resource_error("requested spill window exceeds capacity"));
112        }
113        if start >= self.length || len == 0 {
114            return Ok(ByteWindow::new(start, &[]));
115        }
116        let remaining = match usize::try_from(self.length - start) {
117            Ok(length) => length,
118            Err(_) => usize::MAX,
119        };
120        let target = len.min(remaining);
121        self.output.clear();
122        let record_bytes = u64::try_from(self.record_bytes)
123            .map_err(|_| resource_error("spill record size exceeds u64"))?;
124        let mut position = start;
125        while self.output.len() < target {
126            let index = position / record_bytes;
127            let local = usize::try_from(position % record_bytes)
128                .map_err(|_| resource_error("spill record offset exceeds usize"))?;
129            self.load_record(index)?;
130            let available = self.record.len().saturating_sub(local);
131            if available == 0 {
132                return Err(resource_error(
133                    "spill source ended before its logical length",
134                ));
135            }
136            let count = available.min(target.saturating_sub(self.output.len()));
137            let end = local.saturating_add(count);
138            let bytes = self
139                .record
140                .get(local..end)
141                .ok_or_else(|| resource_error("spill record range is invalid"))?;
142            self.output.extend_from_slice(bytes);
143            position = position
144                .checked_add(
145                    u64::try_from(count).map_err(|_| resource_error("spill copy exceeds u64"))?,
146                )
147                .ok_or_else(|| resource_error("spill source offset overflow"))?;
148        }
149        Ok(ByteWindow::new(start, &self.output))
150    }
151
152    fn len_hint(&self) -> Option<u64> {
153        Some(self.length)
154    }
155}
156
157impl SpillWindowedFile {
158    /// Physical bytes retained by the execution-owned artifact.
159    #[must_use]
160    pub const fn spill_bytes(&self) -> u64 {
161        self.artifact.bytes()
162    }
163}
164
165fn copy_u64(bytes: &[u8]) -> [u8; 8] {
166    let mut value = [0_u8; 8];
167    value.copy_from_slice(bytes);
168    value
169}
170
171fn io_error(operation: &'static str, error: &std::io::Error) -> Diagnostic {
172    resource_error(format!("spill source {operation} failed: {error}"))
173}
174
175fn resource_error(reason: impl Into<Box<str>>) -> Diagnostic {
176    Diagnostic::new(Code::E1902).with_message(reason)
177}
molframe_core/io/input/spill_source.rs

molframe_core/io/input/
spill_source.rs
