coven_database/
local_file.rs

1//! Resolve a file still on this device only to its one local byte source.
2
3use crate::{sqlite::DatabaseConnection, DbError};
4use coven_foundation::files::StoreDirAccess;
5use coven_foundation::task::{Bug, CarriesBug};
6use coven_foundation::{
7    files::{
8        FileError, FileReader, ObservationError, StagedName, StoreDir, StoreLockError,
9        StoreReadLock, UserOriginal,
10    },
11    id_source::DeviceId,
12};
13use std::{path::PathBuf, sync::Arc};
14
15/// A local file operation's typed failure.
16#[derive(Debug, thiserror::Error)]
17pub enum LocalFileError {
18    /// The recorded original is absent.
19    #[error("original {id} is missing at {}", path.display())]
20    UserFileMissing {
21        /// The file's id.
22        id: String,
23        /// Its recorded original path.
24        path: PathBuf,
25    },
26    /// The recorded original's facts changed.
27    #[error("original {id} changed at {}", path.display())]
28    UserFileChanged {
29        /// The file's id.
30        id: String,
31        /// Its recorded original path.
32        path: PathBuf,
33    },
34    /// Bytes did not match the file's content or an owned copy disappeared.
35    #[error("file {id} failed its content check")]
36    Integrity {
37        /// The file's id.
38        id: String,
39    },
40    /// The requested range exceeds the file's size.
41    #[error("file {id} range {offset}..{end} exceeds {size}")]
42    RangeOutOfBounds {
43        /// The file's id.
44        id: String,
45        /// The requested first byte.
46        offset: u64,
47        /// The requested exclusive end, saturated if addition overflows.
48        end: u64,
49        /// The file's whole size.
50        size: u64,
51    },
52    /// Looking up the file failed.
53    #[error(transparent)]
54    Database(DbError),
55    /// The filesystem failed.
56    #[error(transparent)]
57    Disk(FileError),
58    /// The store cannot be retained while opening its file.
59    #[error(transparent)]
60    Lock(StoreLockError),
61    /// Opening or reading the file panicked: a bug, which the caller reports
62    /// (E15).
63    #[error(transparent)]
64    Bug(#[from] Bug),
65}
66
67impl From<DbError> for LocalFileError {
68    fn from(error: DbError) -> Self {
69        match error {
70            DbError::Bug(bug) => Self::Bug(bug),
71            error => Self::Database(error),
72        }
73    }
74}
75
76impl From<FileError> for LocalFileError {
77    fn from(error: FileError) -> Self {
78        match error {
79            FileError::Bug(bug) => Self::Bug(bug),
80            error => Self::Disk(error),
81        }
82    }
83}
84
85impl From<StoreLockError> for LocalFileError {
86    fn from(error: StoreLockError) -> Self {
87        match error {
88            StoreLockError::Bug(bug) => Self::Bug(bug),
89            error => Self::Lock(error),
90        }
91    }
92}
93
94impl LocalFileError {
95    /// Whether this failure judges the bytes a source holds: missing, or
96    /// changed from what its record says, rather than a failure to look.
97    pub(crate) fn judges_bytes(&self) -> bool {
98        matches!(
99            self,
100            Self::UserFileMissing { .. } | Self::UserFileChanged { .. } | Self::Integrity { .. }
101        )
102    }
103}
104
105impl CarriesBug for LocalFileError {
106    fn bug(&self) -> Option<&Bug> {
107        match self {
108            Self::Bug(bug) => Some(bug),
109            Self::Database(error) => error.bug(),
110            Self::Disk(error) => error.bug(),
111            Self::Lock(error) => error.bug(),
112            _ => None,
113        }
114    }
115}
116
117/// An open local file checked against its recorded facts. Retaining it keeps
118/// the opened bytes readable across its upload completing and file
119/// unlinking, and prevents store deletion until its reads and handles have
120/// finished.
121pub struct LocalFileStream {
122    file: Arc<LocalFileReader>,
123    /// Where a panic in a read its caller stopped waiting for goes (E15).
124    bugs: coven_foundation::task::BugSinkRef,
125}
126
127struct LocalFileReader {
128    reader: FileReader,
129    // Drop the open file before allowing deletion, even on a cancelled read.
130    _lock: StoreReadLock,
131    id: String,
132    original: Option<PathBuf>,
133}
134
135impl LocalFileStream {
136    /// The complete plaintext byte count.
137    pub fn plaintext_size(&self) -> u64 {
138        self.file.plaintext_size()
139    }
140
141    /// Read a positioned range on a blocking worker, retaining this open
142    /// handle. A caller that stops waiting leaves the read to finish, and
143    /// its panic goes to the store's bug sink (E15).
144    pub async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, LocalFileError> {
145        let file = self.file.clone();
146        coven_foundation::task::blocking_completing(
147            "read the local file",
148            self.bugs.clone(),
149            move || file.read_at(offset, len),
150        )
151        .await
152        .map_err(FileError::from)?
153    }
154
155    /// Read a complete plaintext chunk and compare its first-read hash before
156    /// returning any bytes to an encryptor. Metadata checks alone cannot detect
157    /// a same-size overwrite whose modification time was restored.
158    pub async fn read_verified_at(
159        &self,
160        offset: u64,
161        len: u64,
162        expected: coven_crypto::ContentHash,
163    ) -> Result<Vec<u8>, LocalFileError> {
164        let bytes = self.read_at(offset, len).await?;
165        let mut hash = coven_crypto::ContentHasher::new();
166        hash.update(&bytes);
167        if hash.finish() != expected {
168            return Err(changed(&self.file.id, self.file.original.as_ref()));
169        }
170        Ok(bytes)
171    }
172}
173
174impl LocalFileReader {
175    fn plaintext_size(&self) -> u64 {
176        self.reader.size()
177    }
178
179    #[expect(
180        clippy::map_err_ignore,
181        reason = "a size past the target integer is TooLarge, which names the field and size"
182    )]
183    fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, LocalFileError> {
184        let end = offset.saturating_add(len);
185        if offset.checked_add(len).is_none() || end > self.reader.size() {
186            return Err(LocalFileError::RangeOutOfBounds {
187                id: self.id.clone(),
188                offset,
189                end,
190                size: self.reader.size(),
191            });
192        }
193        let length = usize::try_from(len).map_err(|_| DbError::TooLarge {
194            field: "file range",
195            actual: len,
196            maximum: usize::MAX as u64,
197        })?;
198        self.reader
199            .read_at(offset, length)
200            .map_err(|error| observation(error, &self.id, self.original.is_some()))
201    }
202}
203
204/// Where the bytes of `file` are while its record, as this read finds it,
205/// says it is on this device only: the user's original, or coven's copy.
206/// Any other file, including one this device has finished uploading since
207/// a reference to it was taken, and one whose record is gone, returns
208/// `None`: its bytes are read from the cache or storage, or it fails as its
209/// record says (§17.7). Reads only the database.
210pub(crate) fn device_only_source(
211    db: &DatabaseConnection,
212    device: DeviceId,
213    file: coven_foundation::id_source::FileId,
214) -> Result<Option<LocalSource>, DbError> {
215    match crate::file_table::Record::read(db, file)? {
216        Some(record) if record.place == crate::file_location::Place::DeviceOnly(device) => {
217            file_source(db, file)
218        }
219        _ => Ok(None),
220    }
221}
222
223/// Where a file's bytes are on this device: the user's original, as
224/// recorded, or coven's own copy.
225pub(crate) enum LocalSource {
226    Original { path: PathBuf, modified_at: std::time::SystemTime },
227    Copy(StagedName),
228}
229
230/// The bytes this device keeps for `file`, while it is on this device only;
231/// `None` for any other file. Reads only the database.
232fn file_source(
233    db: &DatabaseConnection,
234    file: coven_foundation::id_source::FileId,
235) -> Result<Option<LocalSource>, DbError> {
236    if let Some(original) = crate::user_file::read(db, file)? {
237        return Ok(Some(LocalSource::Original {
238            path: original.path,
239            modified_at: original.modified_at,
240        }));
241    }
242    db.query("SELECT path FROM _coven_device_files WHERE file=?1", [file], |row| {
243        row.get::<_, String>(0)
244    })?
245    .into_iter()
246    .next()
247    .map(|name| StagedName::parse(&name).map(LocalSource::Copy).ok_or(DbError::DamagedDatabase))
248    .transpose()
249}
250
251/// The source an upload's entry fixed at staging; `None` once it is no
252/// longer queued. Reads only the database.
253pub(crate) fn queued_source(
254    db: &DatabaseConnection,
255    upload: i64,
256) -> Result<Option<LocalSource>, DbError> {
257    let sources = db.query(
258        "SELECT original,modified_at,copy FROM _coven_file_uploads WHERE id=?1",
259        [upload],
260        |row| {
261            Ok((
262                row.get::<_, Option<Vec<u8>>>(0)?,
263                row.get::<_, Option<Vec<u8>>>(1)?,
264                row.get::<_, Option<String>>(2)?,
265            ))
266        },
267    )?;
268    let Some(source) = sources.into_iter().next() else {
269        return Ok(None);
270    };
271    match source {
272        (Some(path), Some(modified_at), None) => Ok(Some(LocalSource::Original {
273            path: crate::user_file::decode_path(path)?,
274            modified_at: crate::user_file::decode_time(&modified_at)?,
275        })),
276        (None, None, Some(name)) => {
277            Ok(Some(LocalSource::Copy(StagedName::parse(&name).ok_or(DbError::DamagedDatabase)?)))
278        }
279        _ => Err(DbError::DamagedDatabase),
280    }
281}
282
283/// Open a file's local source: it must still have the size, and an
284/// original the modification time, recorded at staging. Reads then return
285/// what's on disk; an upload checks each chunk it sends (§17.9). Touches
286/// only the disk, never the database. `id` names the file in its errors.
287pub(crate) fn open_source(
288    directory: &StoreDir,
289    id: &str,
290    size: u64,
291    source: LocalSource,
292) -> Result<LocalFileStream, LocalFileError> {
293    let id = id.to_owned();
294    let lock = directory.lock_read_only()?;
295    let (reader, original) = match source {
296        LocalSource::Original { path, modified_at } => (
297            UserOriginal::at(path.clone())
298                .open_reader(size, modified_at)
299                .map_err(|error| observation(error, &id, true))?,
300            Some(path),
301        ),
302        LocalSource::Copy(name) => {
303            let reader = directory
304                .staged_file(&name)
305                .open_reader()
306                .map_err(|error| observation(error, &id, false))?;
307            // The stream's size is its open handle's, which must be the file's.
308            if reader.size() != size {
309                return Err(LocalFileError::Integrity { id });
310            }
311            (reader, None)
312        }
313    };
314    Ok(LocalFileStream {
315        file: Arc::new(LocalFileReader { reader, _lock: lock, id, original }),
316        bugs: directory.bugs().clone(),
317    })
318}
319
320fn observation(error: ObservationError, id: &str, original: bool) -> LocalFileError {
321    match (error, original) {
322        (ObservationError::Missing(path), true) => {
323            LocalFileError::UserFileMissing { id: id.into(), path }
324        }
325        (ObservationError::Changed(path), true) => {
326            LocalFileError::UserFileChanged { id: id.into(), path }
327        }
328        (ObservationError::Missing(_) | ObservationError::Changed(_), false) => {
329            LocalFileError::Integrity { id: id.into() }
330        }
331        (ObservationError::File(error), _) => error.into(),
332    }
333}
334
335fn changed(id: &str, original: Option<&PathBuf>) -> LocalFileError {
336    match original {
337        Some(path) => LocalFileError::UserFileChanged { id: id.into(), path: path.clone() },
338        None => LocalFileError::Integrity { id: id.into() },
339    }
340}
coven_database/local_file.rs

coven_database/
local_file.rs
