uqa_storage/
temporary_file.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Ephemeral authenticated byte files with independent reader positions and bounded workspace.
8
9use std::io::{self, IoSlice, Read, Seek, SeekFrom, Write};
10use std::path::{Path, PathBuf};
11use std::sync::Arc;
12
13use chacha20poly1305::{
14    aead::{AeadInPlace, KeyInit},
15    XChaCha20Poly1305, XNonce,
16};
17use parking_lot::Mutex;
18
19const BLOCK_BYTES: usize = 16 * 1024;
20const NONCE_BYTES: usize = 24;
21const LENGTH_BYTES: usize = 2;
22const TAG_BYTES: usize = 16;
23const SLOT_HEADER_BYTES: usize = NONCE_BYTES + LENGTH_BYTES + TAG_BYTES;
24#[cfg(test)]
25const RECORD_BYTES: usize = 1 + 2 * (SLOT_HEADER_BYTES + BLOCK_BYTES);
26
27/// Ordinary spill files use 16 KiB logical blocks.
28pub type TemporaryFile = BlockTemporaryFile<BLOCK_BYTES>;
29
30/// A private temporary byte file with a compile-time logical block width of 1 through 16,384 bytes. Each physical block reserves two authenticated ciphertext slots and a one-byte active selector: `1 + 2 * (24 + 2 + 16 + BYTES)` bytes. Only the populated prefix is encrypted; its two-byte length and block position are authenticated. An incomplete replacement never overwrites the active slot. Only ciphertext and its public framing reach disk; each file has a fresh random key which exists solely in this owner. Reopened handles share the owner but have independent logical positions. The last handle removes the file. Abrupt process death may leave an encrypted file whose key was never persisted; this is not a durable recovery or anti-replay format.
31pub struct BlockTemporaryFile<const BYTES: usize> {
32    owner: Arc<Mutex<Owner<BYTES>>>,
33    path: Arc<PathBuf>,
34    position: u64,
35}
36
37struct Owner<const BYTES: usize> {
38    file: tempfile::NamedTempFile,
39    cipher: XChaCha20Poly1305,
40    length: u64,
41    #[cfg(test)]
42    faults: tests::Faults,
43}
44
45pub struct TemporaryFileMetadata {
46    length: u64,
47}
48
49impl TemporaryFileMetadata {
50    pub fn len(&self) -> u64 {
51        self.length
52    }
53
54    pub fn is_empty(&self) -> bool {
55        self.length == 0
56    }
57}
58
59impl<const BYTES: usize> BlockTemporaryFile<BYTES> {
60    /// Ciphertext file length for a logical length, including both slots and complete block framing. This excludes filesystem allocation metadata.
61    pub fn physical_len_for(logical_length: u64) -> io::Result<u64> {
62        if BYTES == 0 || BYTES > BLOCK_BYTES {
63            return Err(io::Error::new(
64                io::ErrorKind::InvalidInput,
65                "invalid temporary file block width",
66            ));
67        }
68        record_offset::<BYTES>(logical_length.div_ceil(BYTES as u64))
69    }
70
71    pub fn new() -> io::Result<Self> {
72        Self::from_file(tempfile::NamedTempFile::new()?)
73    }
74
75    pub fn new_in(directory: impl AsRef<Path>) -> io::Result<Self> {
76        Self::from_file(tempfile::NamedTempFile::new_in(directory)?)
77    }
78
79    fn from_file(file: tempfile::NamedTempFile) -> io::Result<Self> {
80        Self::physical_len_for(0)?;
81        let mut key = [0_u8; 32];
82        getrandom::fill(&mut key).map_err(|error| io::Error::other(error.to_string()))?;
83        let cipher = XChaCha20Poly1305::new((&key).into());
84        key.fill(0);
85        let path = Arc::new(file.path().to_path_buf());
86        Ok(Self {
87            owner: Arc::new(Mutex::new(Owner {
88                file,
89                cipher,
90                length: 0,
91                #[cfg(test)]
92                faults: tests::Faults::default(),
93            })),
94            path,
95            position: 0,
96        })
97    }
98
99    pub fn path(&self) -> &Path {
100        &self.path
101    }
102
103    #[cfg(test)]
104    pub(crate) fn fail_write_after(&self, bytes: usize) {
105        self.owner.lock().faults.fail_after_bytes = Some(bytes);
106    }
107
108    #[cfg(test)]
109    pub(crate) fn block_io_counts(&self) -> (usize, u64) {
110        let owner = self.owner.lock();
111        (owner.faults.read_blocks, owner.faults.written_bytes)
112    }
113
114    pub fn as_file(&self) -> &Self {
115        self
116    }
117
118    pub fn as_file_mut(&mut self) -> &mut Self {
119        self
120    }
121
122    /// Write the complete concatenation of the slices, sharing one authenticated block publication across adjacent fields. Like `Write::write_all`, an error can leave a written prefix; record owners retain responsibility for rolling back incomplete records.
123    pub fn write_all_vectored(&mut self, mut input: &mut [IoSlice<'_>]) -> io::Result<()> {
124        IoSlice::advance_slices(&mut input, 0);
125        while !input.is_empty() {
126            match self.write_vectored(input) {
127                Ok(0) => {
128                    return Err(io::Error::new(
129                        io::ErrorKind::WriteZero,
130                        "incomplete temporary file write",
131                    ));
132                }
133                Ok(written) => IoSlice::advance_slices(&mut input, written),
134                Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
135                Err(error) => return Err(error),
136            }
137        }
138        Ok(())
139    }
140
141    pub fn reopen(&self) -> io::Result<Self> {
142        self.owner.lock().validate_length()?;
143        Ok(Self {
144            owner: Arc::clone(&self.owner),
145            path: Arc::clone(&self.path),
146            position: 0,
147        })
148    }
149
150    pub fn metadata(&self) -> io::Result<TemporaryFileMetadata> {
151        Ok(TemporaryFileMetadata {
152            length: self.owner.lock().length,
153        })
154    }
155
156    pub fn set_len(&self, length: u64) -> io::Result<()> {
157        record_offset::<BYTES>(length.div_ceil(BYTES as u64))?;
158        let mut owner = self.owner.lock();
159        if length >= owner.length {
160            return owner.extend_zeroed(length);
161        }
162        // Bytes beyond the new logical end are hidden, not rewritten. Growth zeroes them before publication; a failed truncate cannot corrupt the retained partial block.
163        owner.truncate_physical(record_offset::<BYTES>(length.div_ceil(BYTES as u64))?)?;
164        owner.length = length;
165        Ok(())
166    }
167}
168
169fn record_offset<const BYTES: usize>(block: u64) -> io::Result<u64> {
170    block
171        .checked_mul((1 + 2 * (SLOT_HEADER_BYTES + BYTES)) as u64)
172        .ok_or_else(|| {
173            io::Error::new(
174                io::ErrorKind::InvalidInput,
175                "temporary file offset overflow",
176            )
177        })
178}
179
180fn block_aad(block: u64, length: u16) -> [u8; 8 + LENGTH_BYTES] {
181    let mut aad = [0; 8 + LENGTH_BYTES];
182    aad[..8].copy_from_slice(&block.to_le_bytes());
183    aad[8..].copy_from_slice(&length.to_le_bytes());
184    aad
185}
186
187impl<const BYTES: usize> Owner<BYTES> {
188    fn validate_length(&self) -> io::Result<()> {
189        let expected = record_offset::<BYTES>(self.length.div_ceil(BYTES as u64))?;
190        if self.file.as_file().metadata()?.len() != expected {
191            return Err(io::Error::new(
192                io::ErrorKind::InvalidData,
193                "invalid authenticated temporary file length",
194            ));
195        }
196        Ok(())
197    }
198
199    fn active_slot(&mut self, block: u64) -> io::Result<u8> {
200        let file = self.file.as_file_mut();
201        file.seek(SeekFrom::Start(record_offset::<BYTES>(block)?))?;
202        let mut active = [0];
203        file.read_exact(&mut active)?;
204        if active[0] > 1 {
205            return Err(io::Error::new(
206                io::ErrorKind::InvalidData,
207                "invalid temporary file active slot",
208            ));
209        }
210        Ok(active[0])
211    }
212
213    fn read_block(&mut self, block: u64) -> io::Result<[u8; BYTES]> {
214        let mut plaintext = [0_u8; BYTES];
215        if block < self.length.div_ceil(BYTES as u64) {
216            #[cfg(test)]
217            {
218                self.faults.read_blocks += 1;
219            }
220            let active = self.active_slot(block)?;
221            let file = self.file.as_file_mut();
222            file.seek(SeekFrom::Start(slot_offset::<BYTES>(block, active)?))?;
223            let mut nonce = [0_u8; NONCE_BYTES];
224            let mut length = [0_u8; LENGTH_BYTES];
225            let mut tag = [0_u8; TAG_BYTES];
226            file.read_exact(&mut nonce)?;
227            file.read_exact(&mut length)?;
228            let length = u16::from_le_bytes(length);
229            let populated = usize::from(length);
230            let required = (self.length - block * BYTES as u64).min(BYTES as u64) as usize;
231            if populated > BYTES || populated < required {
232                return Err(io::Error::new(
233                    io::ErrorKind::InvalidData,
234                    "invalid authenticated temporary block length",
235                ));
236            }
237            file.read_exact(&mut tag)?;
238            file.read_exact(&mut plaintext[..populated])?;
239            self.cipher
240                .decrypt_in_place_detached(
241                    XNonce::from_slice(&nonce),
242                    &block_aad(block, length),
243                    &mut plaintext[..populated],
244                    (&tag).into(),
245                )
246                .map_err(|_| {
247                    io::Error::new(
248                        io::ErrorKind::InvalidData,
249                        "temporary file authentication failed",
250                    )
251                })?;
252        }
253        Ok(plaintext)
254    }
255
256    fn write_block(
257        &mut self,
258        block: u64,
259        ciphertext: &mut [u8; BYTES],
260        populated: usize,
261    ) -> io::Result<()> {
262        let existing = block < self.length.div_ceil(BYTES as u64);
263        let next = if existing {
264            self.active_slot(block)? ^ 1
265        } else {
266            0
267        };
268        let mut nonce = [0_u8; NONCE_BYTES];
269        getrandom::fill(&mut nonce).map_err(|error| io::Error::other(error.to_string()))?;
270        let length = u16::try_from(populated).expect("temporary blocks are at most 16 KiB");
271        let tag = self
272            .cipher
273            .encrypt_in_place_detached(
274                XNonce::from_slice(&nonce),
275                &block_aad(block, length),
276                &mut ciphertext[..populated],
277            )
278            .map_err(|_| io::Error::other("temporary file encryption failed"))?;
279        // The complete replacement reaches the inactive slot before its one-byte publication marker. Failed short writes leave the authoritative slot unchanged, including the prefix retained by append rollback.
280        self.file
281            .as_file_mut()
282            .seek(SeekFrom::Start(slot_offset::<BYTES>(block, next)?))?;
283        self.write_physical(&nonce)?;
284        self.write_physical(&length.to_le_bytes())?;
285        self.write_physical(&tag)?;
286        self.write_physical(&ciphertext[..populated])?;
287        if !existing {
288            self.truncate_physical(record_offset::<BYTES>(block + 1)?)?;
289        }
290        self.file
291            .as_file_mut()
292            .seek(SeekFrom::Start(record_offset::<BYTES>(block)?))?;
293        self.write_physical(&[next])
294    }
295
296    fn write_physical(&mut self, bytes: &[u8]) -> io::Result<()> {
297        #[cfg(test)]
298        self.faults.before_write(self.file.as_file_mut(), bytes)?;
299        self.file.as_file_mut().write_all(bytes)
300    }
301
302    fn truncate_physical(&mut self, length: u64) -> io::Result<()> {
303        #[cfg(test)]
304        self.faults.before_truncate()?;
305        self.file.as_file().set_len(length)
306    }
307
308    fn rollback_length(&mut self, length: u64, cause: &io::Error) -> io::Result<()> {
309        // Logical visibility returns to the original boundary even when the filesystem cannot remove an unpublished ciphertext tail. Existing readers keep that prefix; reopen rejects inconsistent physical length and the caller receives the cleanup error.
310        self.length = length;
311        self.truncate_physical(record_offset::<BYTES>(length.div_ceil(BYTES as u64))?)
312            .map_err(|rollback| {
313                io::Error::other(format!(
314                    "{cause}; temporary file rollback failed: {rollback}"
315                ))
316            })?;
317        Ok(())
318    }
319
320    fn extend_zeroed(&mut self, length: u64) -> io::Result<()> {
321        let original = self.length;
322        let result = (|| {
323            while self.length < length {
324                let block = self.length / BYTES as u64;
325                let offset = (self.length % BYTES as u64) as usize;
326                let mut bytes = self.read_block(block)?;
327                let count = (length - self.length).min((BYTES - offset) as u64) as usize;
328                bytes[offset..offset + count].fill(0);
329                self.write_block(block, &mut bytes, offset + count)?;
330                self.length += count as u64;
331            }
332            Ok(())
333        })();
334        if let Err(error) = result {
335            self.rollback_length(original, &error)?;
336            return Err(error);
337        }
338        Ok(())
339    }
340}
341
342fn slot_offset<const BYTES: usize>(block: u64, slot: u8) -> io::Result<u64> {
343    record_offset::<BYTES>(block)?
344        .checked_add(1 + u64::from(slot) * (SLOT_HEADER_BYTES + BYTES) as u64)
345        .ok_or_else(|| {
346            io::Error::new(
347                io::ErrorKind::InvalidInput,
348                "temporary file slot offset overflow",
349            )
350        })
351}
352
353impl<const BYTES: usize> Read for BlockTemporaryFile<BYTES> {
354    fn read(&mut self, output: &mut [u8]) -> io::Result<usize> {
355        let mut owner = self.owner.lock();
356        if self.position >= owner.length || output.is_empty() {
357            return Ok(0);
358        }
359        let block = self.position / BYTES as u64;
360        let offset = (self.position % BYTES as u64) as usize;
361        let bytes = owner.read_block(block)?;
362        let count =
363            (output.len().min(BYTES - offset) as u64).min(owner.length - self.position) as usize;
364        output[..count].copy_from_slice(&bytes[offset..offset + count]);
365        self.position += count as u64;
366        Ok(count)
367    }
368}
369
370impl<const BYTES: usize> Write for BlockTemporaryFile<BYTES> {
371    fn write(&mut self, input: &[u8]) -> io::Result<usize> {
372        self.write_vectored(&[IoSlice::new(input)])
373    }
374
375    fn write_vectored(&mut self, input: &[IoSlice<'_>]) -> io::Result<usize> {
376        if input.iter().all(|bytes| bytes.is_empty()) {
377            return Ok(0);
378        }
379        let minimum_end = self.position.checked_add(1).ok_or_else(|| {
380            io::Error::new(
381                io::ErrorKind::InvalidInput,
382                "temporary file length overflow",
383            )
384        })?;
385        record_offset::<BYTES>(minimum_end.div_ceil(BYTES as u64))?;
386        let mut owner = self.owner.lock();
387        let original = owner.length;
388        owner.extend_zeroed(self.position)?;
389        let block = self.position / BYTES as u64;
390        let offset = (self.position % BYTES as u64) as usize;
391        let mut bytes = match owner.read_block(block) {
392            Ok(bytes) => bytes,
393            Err(error) => {
394                owner.rollback_length(original, &error)?;
395                return Err(error);
396            }
397        };
398        let mut count = 0;
399        for source in input {
400            let taken = source.len().min(BYTES - offset - count);
401            bytes[offset + count..offset + count + taken].copy_from_slice(&source[..taken]);
402            count += taken;
403            if offset + count == BYTES {
404                break;
405            }
406        }
407        let next_position = self.position.checked_add(count as u64).ok_or_else(|| {
408            io::Error::new(
409                io::ErrorKind::InvalidInput,
410                "temporary file length overflow",
411            )
412        })?;
413        let populated =
414            (owner.length.max(next_position) - block * BYTES as u64).min(BYTES as u64) as usize;
415        if let Err(error) = owner.write_block(block, &mut bytes, populated) {
416            owner.rollback_length(original, &error)?;
417            return Err(error);
418        }
419        owner.length = owner.length.max(next_position);
420        self.position = next_position;
421        Ok(count)
422    }
423
424    fn flush(&mut self) -> io::Result<()> {
425        self.owner.lock().file.as_file_mut().flush()
426    }
427}
428
429impl<const BYTES: usize> Seek for BlockTemporaryFile<BYTES> {
430    fn seek(&mut self, position: SeekFrom) -> io::Result<u64> {
431        self.position = match position {
432            SeekFrom::Start(position) => position,
433            SeekFrom::End(offset) => self
434                .owner
435                .lock()
436                .length
437                .checked_add_signed(offset)
438                .ok_or_else(|| {
439                    io::Error::new(io::ErrorKind::InvalidInput, "invalid temporary file seek")
440                })?,
441            SeekFrom::Current(offset) => {
442                self.position.checked_add_signed(offset).ok_or_else(|| {
443                    io::Error::new(io::ErrorKind::InvalidInput, "invalid temporary file seek")
444                })?
445            }
446        };
447        Ok(self.position)
448    }
449}
450
451#[cfg(test)]
452mod tests;
uqa_storage/temporary_file.rs

uqa_storage/
temporary_file.rs
