coven_database/
database_local_bytes.rs

1//! The bytes the cache and pins keep on this device, read from the counters
2//! the cache keeps with its entries (§17.8, Appendix C.3).
3
4use super::{Database, Mode};
5use crate::{sqlite::DatabaseConnection, DbError, LiveQuery};
6use coven_foundation::task::CarriesBug;
7
8/// One namespace's cache, in bytes on disk.
9#[derive(Clone, Debug, PartialEq, Eq)]
10pub struct NamespaceBytes {
11    /// The namespace's name as declared.
12    pub namespace: String,
13    /// Its budget, declared or since set.
14    pub budget: u64,
15    /// Chunks and headers: what counts toward the budget.
16    pub cached: u64,
17    /// Pinned whole copies.
18    pub pinned: u64,
19}
20
21/// The cache's bytes: every declared namespace's, and the partial downloads
22/// pins left, which the cache neither counts nor evicts.
23#[derive(Clone, Debug, PartialEq, Eq)]
24pub struct CacheBytes {
25    /// Each declared namespace's cache, in declaration order.
26    pub namespaces: Vec<NamespaceBytes>,
27    /// Partial downloads pins left.
28    pub pinning: u64,
29}
30
31/// What a store keeps on this device, in bytes on disk (§17.8, §19.3).
32#[derive(Clone, Debug, PartialEq, Eq)]
33pub struct LocalBytes {
34    /// Each declared namespace's cache, in declaration order.
35    pub namespaces: Vec<NamespaceBytes>,
36    /// Partial downloads pins left, kept apart from the cache.
37    pub pinning: u64,
38    /// The store's database and its journals, its settings and key files, and
39    /// coven's own copies of files: every file in the store's directory but
40    /// the cache and the archives.
41    pub library: u64,
42    /// The archives resets kept, while there is one.
43    pub before_reset: Option<u64>,
44}
45
46fn unsigned(value: i64) -> Result<u64, DbError> {
47    u64::try_from(value).map_err(|_| DbError::DamagedDatabase)
48}
49
50/// Every namespace's budget and counters, by name.
51#[expect(
52    clippy::map_err_ignore,
53    reason = "a stored value that won't convert or decode is DamagedDatabase"
54)]
55pub(crate) fn budgets(db: &DatabaseConnection) -> Result<Vec<NamespaceBytes>, DbError> {
56    db.query(
57        "SELECT namespace,bytes,cached,pinned FROM _coven_cache_budgets ORDER BY namespace",
58        [],
59        |r| {
60            Ok((
61                r.get::<_, String>(0)?,
62                r.get::<_, Vec<u8>>(1)?,
63                r.get::<_, i64>(2)?,
64                r.get::<_, i64>(3)?,
65            ))
66        },
67    )?
68    .into_iter()
69    .map(|(namespace, budget, cached, pinned)| {
70        Ok(NamespaceBytes {
71            namespace,
72            budget: u64::from_be_bytes(budget.try_into().map_err(|_| DbError::DamagedDatabase)?),
73            cached: unsigned(cached)?,
74            pinned: unsigned(pinned)?,
75        })
76    })
77    .collect()
78}
79
80/// The sum of the partial downloads' sizes.
81pub(crate) fn partials(db: &DatabaseConnection) -> Result<u64, DbError> {
82    unsigned(db.query_row("SELECT coalesce(sum(size),0) FROM _coven_cache_partials", [], |r| {
83        r.get::<_, i64>(0)
84    })?)
85}
86
87impl<M: Mode> Database<M> {
88    /// The cache's counters and the partial downloads' sizes, from one
89    /// snapshot, for the namespaces declared at open, in declaration order.
90    pub async fn cache_bytes(&self) -> Result<CacheBytes, DbError> {
91        let declared: Vec<String> = self
92            .call(|open| {
93                Ok::<_, DbError>(
94                    M::access(open)
95                        .write_schema
96                        .namespaces
97                        .iter()
98                        .map(|namespace| namespace.name.clone())
99                        .collect(),
100                )
101            })
102            .await?;
103        let (mut kept, pinning) = self.read(|sql| sql.cache_bytes()).await?;
104        let namespaces = declared
105            .into_iter()
106            .filter_map(|name| {
107                kept.iter().position(|bytes| bytes.namespace == name).map(|at| kept.swap_remove(at))
108            })
109            .collect();
110        Ok(CacheBytes { namespaces, pinning })
111    }
112}
113
114impl Database {
115    /// The namespaces' budgets and counters, now and after each change to
116    /// either. Reads nothing else, so a cached read's recency or a pin
117    /// download's partial growing starts no run.
118    pub fn subscribe_cache_budgets<E>(&self) -> LiveQuery<Vec<NamespaceBytes>, E>
119    where
120        E: From<DbError> + CarriesBug + Send + 'static,
121    {
122        self.subscribe(|sql| sql.cache_budgets().map_err(E::from))
123    }
124}
125
126impl Database {
127    /// One measurement of what the store keeps on this device: the cache's
128    /// counters, budgets and partials from one snapshot, then the store's
129    /// directory walked for its own files and the archives resets kept. The
130    /// cache's folder is left out of the walk, since its entries count it.
131    pub async fn local_bytes(&self) -> Result<LocalBytes, DbError> {
132        let cache = self.cache_bytes().await?;
133        let directory = self.call(|open| Ok::<_, DbError>(open.access.directory.clone())).await?;
134        let walked = coven_foundation::task::blocking("measure the store's directory", move || {
135            directory.local_bytes()
136        })
137        .await??;
138        Ok(LocalBytes {
139            namespaces: cache.namespaces,
140            pinning: cache.pinning,
141            library: walked.library,
142            before_reset: walked.archives,
143        })
144    }
145
146    /// The first measurement at once, then a new one after each write that
147    /// changes a namespace's counters or budget, and after each
148    /// `local_bytes_changed`. Partial downloads are measured with each but
149    /// start none.
150    pub fn subscribe_local_bytes(&self) -> LocalBytesWatch {
151        let mut changed = self.writer.local_bytes_changes.subscribe();
152        changed.mark_unchanged();
153        LocalBytesWatch { database: self.clone(), budgets: self.subscribe_cache_budgets(), changed }
154    }
155
156    /// Say that the store's directory changed in a way its database doesn't
157    /// report, such as the archives resets kept going.
158    pub fn local_bytes_changed(&self) {
159        self.writer.local_bytes_changes.send_modify(|count| *count += 1);
160    }
161}
162
163/// The measurements of what the store keeps on this device, as they change.
164pub struct LocalBytesWatch {
165    database: Database,
166    budgets: LiveQuery<Vec<NamespaceBytes>, DbError>,
167    changed: tokio::sync::watch::Receiver<u64>,
168}
169
170impl LocalBytesWatch {
171    /// The first measurement at once, then the latest one after each change.
172    /// A reader slower than the changes gets one measurement of the state
173    /// now, not one per change. `None` once the store has closed.
174    pub async fn next(&mut self) -> Option<Result<LocalBytes, DbError>> {
175        tokio::select! {
176            run = self.budgets.next() => match run? {
177                Ok(_) => {}
178                Err(error) => return Some(Err(error)),
179            },
180            changed = self.changed.changed() => changed.ok()?,
181        }
182        Some(self.database.local_bytes().await)
183    }
184
185    /// Stop measuring.
186    pub fn cancel(self) {}
187}
coven_database/database_local_bytes.rs

coven_database/
database_local_bytes.rs
