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

uqa_storage/
temporary_file.rs
