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

uqa_storage/
temporary_file.rs
