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

uqa_storage/
temporary_file.rs
