coven_database/
cache_progress.rs

1//! Pin progress as a query over committed cache coverage (§17.8, E8).
2//! Nothing here counts bytes in flight.
3
4use crate::{sqlite::DatabaseConnection, Database, DbError, FileRef, LiveQuery};
5use coven_foundation::id_source::{DeviceId, FileId};
6
7/// Verified cache coverage of requested files (§17.8, E8).
8#[derive(Clone, Debug, PartialEq, Eq)]
9pub struct PinProgress {
10    /// Files whose bytes have all been kept.
11    pub files_completed: u64,
12    /// Files requested, including those already present.
13    pub files_total: u64,
14    /// Verified plaintext bytes present in the committed cache for the
15    /// requested files.
16    pub bytes_cached: u64,
17    /// Total plaintext size of the requested files, including cached bytes.
18    pub bytes_total: u64,
19}
20
21/// Verified plaintext bytes of `file` the cache holds: a checked whole file,
22/// or each authenticated chunk without its tag. A file only on this device
23/// is wholly local; one only on another device has none here.
24#[expect(
25    clippy::map_err_ignore,
26    reason = "a stored value that won't convert or decode is DamagedDatabase"
27)]
28pub(crate) fn covered(
29    db: &DatabaseConnection,
30    file: &FileRef,
31    device: DeviceId,
32) -> Result<u64, DbError> {
33    match file.place() {
34        crate::FilePlace::DeviceOnly { device: on } if on == device => return Ok(file.size()),
35        crate::FilePlace::DeviceOnly { .. } => return Ok(0),
36        crate::FilePlace::Cloud { .. } => {}
37    }
38    let object = crate::database::file_database::object_path_of(file)?;
39    if db.query_row(
40        "SELECT EXISTS(SELECT 1 FROM _coven_cache
41         WHERE namespace=?1 AND object_path=?2 AND chunk=?3)",
42        (file.namespace(), &object, crate::internal_schema::CACHE_COMPLETE_FILE),
43        |r| r.get::<_, bool>(0),
44    )? {
45        return Ok(file.size());
46    }
47    let cached = db.query_row(
48        "SELECT coalesce(sum(size-chunks*?3),0) FROM _coven_cache
49         WHERE namespace=?1 AND object_path=?2 AND chunk>=0",
50        (file.namespace(), &object, coven_crypto::FILE_CHUNK_TAG_LEN as i64),
51        |r| r.get::<_, i64>(0),
52    )?;
53    let cached = u64::try_from(cached).map_err(|_| DbError::DamagedDatabase)?;
54    if cached > file.size() {
55        return Err(DbError::DamagedDatabase);
56    }
57    Ok(cached)
58}
59
60/// Requested files' verified coverage. A pin completes with its checked
61/// whole file; a file only on this device is already whole here.
62pub(crate) fn pin_progress(
63    db: &DatabaseConnection,
64    files: &[FileRef],
65    device: DeviceId,
66) -> Result<PinProgress, DbError> {
67    let mut progress = PinProgress {
68        files_completed: 0,
69        files_total: files.len() as u64,
70        bytes_cached: 0,
71        bytes_total: 0,
72    };
73    let too_large =
74        || DbError::TooLarge { field: "requested file bytes", actual: u64::MAX, maximum: u64::MAX };
75    for file in files {
76        let cached = covered(db, file, device)?;
77        progress.bytes_cached = progress.bytes_cached.checked_add(cached).ok_or_else(too_large)?;
78        progress.bytes_total =
79            progress.bytes_total.checked_add(file.size()).ok_or_else(too_large)?;
80        let whole = match file.place() {
81            crate::FilePlace::DeviceOnly { device: on } => on == device,
82            crate::FilePlace::Cloud { .. } => crate::database::file_database::cache_kept(db, file)?,
83        };
84        if whole {
85            progress.files_completed += 1;
86        }
87    }
88    Ok(progress)
89}
90
91/// Pin answers in order; a file this device has no record of is `None`.
92/// The answer is whether `coven_pins` lists the file, so the app's SQL and
93/// this call agree. It reads no merge state, so a live query of it reruns
94/// only for those tables' changes.
95pub(crate) fn files_pinned(
96    db: &DatabaseConnection,
97    files: &[FileId],
98) -> Result<Vec<Option<bool>>, DbError> {
99    files
100        .iter()
101        .map(|file| {
102            let (known, pinned): (bool, bool) = db.query_row(
103                "SELECT EXISTS(SELECT 1 FROM main.coven_files WHERE id=?1),
104                        EXISTS(SELECT 1 FROM coven_pins WHERE file=?1)",
105                [file],
106                |r| Ok((r.get(0)?, r.get(1)?)),
107            )?;
108            Ok(known.then_some(pinned))
109        })
110        .collect()
111}
112
113impl Database {
114    /// Verified cache coverage for these fixed references, in one snapshot.
115    pub async fn pin_progress(&self, files: Vec<FileRef>) -> Result<PinProgress, DbError> {
116        let device = self.local_device()?;
117        self.read(move |sql| sql.pin_progress(&files, device)).await
118    }
119    /// The same pre-built cache query with the same fixed references.
120    pub fn subscribe_pin_progress<E>(&self, files: Vec<FileRef>) -> LiveQuery<PinProgress, E>
121    where
122        E: From<DbError> + coven_foundation::task::CarriesBug + Send + 'static,
123    {
124        let device = self.local_device().ok();
125        self.subscribe(move |sql| {
126            sql.pin_progress(&files, device.ok_or(DbError::StoreClosed)?).map_err(E::from)
127        })
128    }
129    /// Whether each file is pinned, one answer per file in order.
130    pub async fn files_pinned(&self, files: Vec<FileId>) -> Result<Vec<Option<bool>>, DbError> {
131        self.read(move |sql| sql.files_pinned(&files)).await
132    }
133    /// The same internal query, live.
134    pub fn subscribe_files_pinned<E>(&self, files: Vec<FileId>) -> LiveQuery<Vec<Option<bool>>, E>
135    where
136        E: From<DbError> + coven_foundation::task::CarriesBug + Send + 'static,
137    {
138        self.subscribe(move |sql| sql.files_pinned(&files).map_err(E::from))
139    }
140}
141
142#[cfg(test)]
143#[path = "cache_progress_tests.rs"]
144mod tests;
coven_database/cache_progress.rs

coven_database/
cache_progress.rs
