coven_database/
upload.rs

1//! Original authored plaintext, with format and keys fixed by its first upload attempt.
2
3use crate::internal_schema::WRITE_DEVICE_SQL;
4use crate::sqlite::DatabaseConnection;
5use crate::write_encoding::{counter, encoded};
6use crate::DbError;
7use coven_format::chunks::CHUNK_SIZE;
8use coven_format::sealed_write::WriteObjectPrefix;
9use coven_format::write_stream::{PartHeader, WriteHeaderFrame};
10use coven_foundation::id_source::DeviceId;
11use coven_merge::WriteId;
12use rusqlite::params;
13use std::rc::Rc;
14
15/// A waiting write's plaintext, fixed at commit, read under one committed
16/// reader transaction.
17pub struct UploadPlaintext<'a> {
18    /// Identity, causality and the audience stream boundaries.
19    pub header: WriteHeaderFrame,
20    /// The exact plaintext header frame, sealed as one chunk.
21    pub header_frame: Vec<u8>,
22    /// Each audience's record stream, in header order, cut into 64 KiB chunks.
23    pub parts: UploadParts<'a>,
24}
25
26/// A captured write ready to seal. Sealing and transport belong to sync;
27/// these are its input streams and the keys it seals with.
28pub struct WaitingUpload<'a> {
29    /// The plaintext fixed at commit.
30    pub plaintext: UploadPlaintext<'a>,
31    /// Format and keys fixed by the first attempt.
32    pub keys: WriteObjectPrefix,
33}
34
35/// A captured queue prefix whose writes all have their first-attempt keys
36/// fixed, in queue order. Only [`crate::Database::prepare_uploads`] makes
37/// one, and [`crate::Database::read_uploads`] reads its writes' plaintext.
38pub struct PreparedUploads {
39    writes: Vec<PreparedUpload>,
40}
41
42/// One captured write: its queue row and the keys it seals with.
43struct PreparedUpload {
44    rowid: i64,
45    write: WriteId,
46    keys: WriteObjectPrefix,
47}
48
49impl PreparedUploads {
50    /// How many writes the prefix captured.
51    pub fn len(&self) -> usize {
52        self.writes.len()
53    }
54
55    /// Whether the prefix captured no write.
56    pub fn is_empty(&self) -> bool {
57        self.writes.is_empty()
58    }
59}
60
61/// A queue read failed in the database or in its consumer.
62#[derive(Debug, thiserror::Error)]
63#[cfg(any(test, feature = "test-utils"))]
64pub enum UploadReadError<E> {
65    /// Reading the committed queue state failed.
66    #[error(transparent)]
67    Database(#[from] DbError),
68    /// The consumer failed while reading or using the stream.
69    #[error("upload consumer: {0}")]
70    Consumer(E),
71}
72
73#[cfg(any(test, feature = "test-utils"))]
74impl<E: coven_foundation::task::CarriesBug> coven_foundation::task::CarriesBug
75    for UploadReadError<E>
76{
77    fn bug(&self) -> Option<&coven_foundation::task::Bug> {
78        match self {
79            Self::Database(error) => error.bug(),
80            Self::Consumer(error) => error.bug(),
81        }
82    }
83}
84
85/// An incremental SQLite BLOB reader. It never collects the waiting write.
86#[derive(Clone)]
87pub struct UploadBytes<'a> {
88    blob: Rc<rusqlite::blob::Blob<'a>>,
89    offset: usize,
90    end: usize,
91}
92
93impl<'a> UploadBytes<'a> {
94    pub(crate) fn new(blob: rusqlite::blob::Blob<'a>) -> Self {
95        let end = blob.len();
96        Self { blob: Rc::new(blob), offset: 0, end }
97    }
98
99    fn read(&self, offset: usize, length: usize) -> Result<Vec<u8>, DbError> {
100        self.check_length(offset, length)?;
101        let mut bytes = vec![0; length];
102        self.blob.read_at_exact(&mut bytes, offset)?;
103        Ok(bytes)
104    }
105
106    fn check_length(&self, offset: usize, length: usize) -> Result<(), DbError> {
107        if offset > self.end || length > self.end - offset {
108            return Err(DbError::Invariant {
109                what: "upload reads stay inside the queued plaintext",
110            });
111        }
112        Ok(())
113    }
114}
115
116impl Iterator for UploadBytes<'_> {
117    type Item = Result<Vec<u8>, DbError>;
118
119    fn next(&mut self) -> Option<Self::Item> {
120        if self.offset == self.end {
121            return None;
122        }
123        let length = (self.end - self.offset).min(CHUNK_SIZE);
124        let result = self.read(self.offset, length);
125        self.offset = if result.is_ok() { self.offset + length } else { self.end };
126        Some(result)
127    }
128}
129
130/// Audience stream descriptors paired with their bounded plaintext readers.
131#[derive(Clone)]
132pub struct UploadParts<'a> {
133    bytes: UploadBytes<'a>,
134    headers: std::vec::IntoIter<PartHeader>,
135}
136
137impl<'a> Iterator for UploadParts<'a> {
138    type Item = (PartHeader, UploadBytes<'a>);
139
140    fn next(&mut self) -> Option<Self::Item> {
141        let header = self.headers.next()?;
142        let offset = self.bytes.offset;
143        self.bytes.offset += header.plaintext_length as usize;
144        Some((
145            header,
146            UploadBytes { blob: self.bytes.blob.clone(), offset, end: self.bytes.offset },
147        ))
148    }
149}
150
151impl UploadParts<'_> {
152    /// Each part's authorship root (D11), read from the queued plaintext
153    /// without consuming these streams. `hasher` gives the keyed leaf hasher
154    /// of a part, by its index in header order, under its first-attempt key.
155    pub fn roots<E: From<DbError> + From<coven_format::Error>>(
156        &self,
157        mut hasher: impl FnMut(usize, &PartHeader) -> Result<coven_crypto::AuthorshipHasher, E>,
158    ) -> Result<Vec<[u8; 32]>, E> {
159        let mut roots = Vec::new();
160        for (index, (header, mut bytes)) in self.clone().enumerate() {
161            let hasher = hasher(index, &header)?;
162            let mut builder = coven_format::authorship::RootBuilder::new(&hasher);
163            crate::download_stream::visit_part::<E>(
164                header,
165                |chunk| {
166                    let plain = bytes.next().ok_or(DbError::DamagedDatabase)??;
167                    if plain.len() != chunk.len() {
168                        return Err(DbError::DamagedDatabase.into());
169                    }
170                    chunk.copy_from_slice(&plain);
171                    Ok(())
172                },
173                |frame| {
174                    let record = match &frame {
175                        coven_format::dismissal::WriteFrame::Change(row) => {
176                            coven_format::write::PartRecord::Change(row)
177                        }
178                        coven_format::dismissal::WriteFrame::Execution(execution) => {
179                            coven_format::write::PartRecord::Execution(execution)
180                        }
181                        coven_format::dismissal::WriteFrame::Dismissal(_) => return Ok(()),
182                    };
183                    for leaf in coven_format::authorship::record_leaves(record) {
184                        builder.push(&leaf)?;
185                    }
186                    Ok(())
187                },
188            )?;
189            roots.push(builder.finish());
190        }
191        Ok(roots)
192    }
193}
194
195/// A write waiting in `_coven_uploads`: its row, its id, and the keys its
196/// first attempt fixed, when that is known.
197struct QueueEntry {
198    rowid: i64,
199    write: WriteId,
200    keys: Option<Vec<u8>>,
201}
202
203impl QueueEntry {
204    /// An entry selected as `rowid,device,number` from `_coven_uploads`,
205    /// then its upload keys' `sealing_keys`, or NULL for none known.
206    fn read(row: &rusqlite::Row<'_>) -> rusqlite::Result<Self> {
207        Ok(Self {
208            rowid: row.get(0)?,
209            write: WriteId { device: row.get::<_, DeviceId>(1)?, number: counter(row.get(2)?)? },
210            keys: row.get(3)?,
211        })
212    }
213}
214
215#[cfg(any(test, feature = "test-utils"))]
216fn oldest(database: &DatabaseConnection) -> Result<Option<QueueEntry>, DbError> {
217    Ok(database
218        .query(
219            "SELECT u.rowid,u.device,u.number,k.sealing_keys FROM _coven_uploads u LEFT JOIN _coven_upload_keys k ON k.device=u.device AND k.number=u.number ORDER BY u.rowid LIMIT 1",
220            [],
221            QueueEntry::read,
222        )?
223        .pop())
224}
225
226/// The oldest waiting write and the keys its first attempt fixed, if any.
227#[cfg(test)]
228pub(crate) fn oldest_keys(
229    database: &DatabaseConnection,
230) -> Result<Option<(WriteId, Option<WriteObjectPrefix>)>, DbError> {
231    let Some(entry) = oldest(database)? else {
232        return Ok(None);
233    };
234    let keys =
235        entry.keys.as_deref().map(WriteObjectPrefix::decode).transpose().map_err(DbError::from)?;
236    Ok(Some((entry.write, keys)))
237}
238
239fn plaintext<'a>(
240    database: &'a DatabaseConnection,
241    rowid: i64,
242    write: WriteId,
243) -> Result<(WriteHeaderFrame, Vec<u8>, UploadBytes<'a>), DbError> {
244    let mut bytes = database.upload_plaintext(rowid)?;
245    let prefix = bytes.read(0, coven_format::FRAME_PREFIX_LEN)?;
246    let length = encoded(coven_format::frame_length(&prefix))?;
247    let header_frame = bytes.read(0, length)?;
248    let header = encoded(WriteHeaderFrame::decode(&header_frame))?;
249    let total =
250        header.parts.iter().try_fold(length as u64, |n, p| n.checked_add(p.plaintext_length));
251    if total != Some(bytes.end as u64) || header.header.position != write {
252        return Err(DbError::DamagedDatabase);
253    }
254    bytes.offset = length;
255    Ok((header, header_frame, bytes))
256}
257
258#[cfg(any(test, feature = "test-utils"))]
259pub(crate) fn read<R, E>(
260    database: &DatabaseConnection,
261    consume: impl FnOnce(UploadPlaintext<'_>) -> Result<R, E>,
262) -> Result<Option<R>, UploadReadError<E>> {
263    let Some(entry) = oldest(database)? else {
264        return Ok(None);
265    };
266    consume(read_plaintext(database, entry.rowid, entry.write)?)
267        .map(Some)
268        .map_err(UploadReadError::Consumer)
269}
270
271fn read_plaintext(
272    database: &DatabaseConnection,
273    rowid: i64,
274    write: WriteId,
275) -> Result<UploadPlaintext<'_>, DbError> {
276    let (header, header_frame, bytes) = plaintext(database, rowid, write)?;
277    let parts = UploadParts { bytes, headers: header.parts.clone().into_iter() };
278    Ok(UploadPlaintext { header, header_frame, parts })
279}
280
281fn batch(
282    database: &DatabaseConnection,
283    device: DeviceId,
284    through: u64,
285) -> Result<Vec<QueueEntry>, DbError> {
286    database.query(
287        "SELECT u.rowid,u.device,u.number,k.sealing_keys FROM _coven_uploads u
288        LEFT JOIN _coven_upload_keys k ON k.device=u.device AND k.number=u.number
289        WHERE u.device=?1 AND u.number<=?2 ORDER BY u.number",
290        params![device, through.to_be_bytes()],
291        QueueEntry::read,
292    )
293}
294
295/// Fix the keys of every write in `device`'s queue through `through` that
296/// has none yet, keeping the keys earlier attempts fixed.
297pub(crate) fn prepare_batch<E: From<DbError> + From<coven_format::Error>>(
298    database: &mut DatabaseConnection,
299    device: DeviceId,
300    through: u64,
301    mut select: impl FnMut(&WriteHeaderFrame) -> Result<WriteObjectPrefix, E>,
302) -> Result<PreparedUploads, E> {
303    let mut writes = Vec::new();
304    let mut selected = Vec::new();
305    for entry in batch(database, device, through)? {
306        let keys = match entry.keys {
307            Some(keys) => WriteObjectPrefix::decode(&keys).map_err(DbError::from)?,
308            None => {
309                let (header, _, _) = plaintext(database, entry.rowid, entry.write)?;
310                let keys = select(&header)?;
311                selected.push((entry.write, keys.encode()?));
312                keys
313            }
314        };
315        writes.push(PreparedUpload { rowid: entry.rowid, write: entry.write, keys });
316    }
317    if !selected.is_empty() {
318        database.stream_transaction(|database| {
319            for (write, keys) in selected {
320                keep_keys(database, write, &keys)?;
321            }
322            Ok(())
323        })?;
324    }
325    Ok(PreparedUploads { writes })
326}
327
328/// Fix the keys a queued write is sealed with, at its first attempt.
329fn keep_keys(database: &DatabaseConnection, write: WriteId, keys: &[u8]) -> Result<(), DbError> {
330    database.internal_execute(
331        "INSERT INTO _coven_upload_keys(device,number,sealing_keys) VALUES(?1,?2,?3)",
332        params![write.device, write.number.to_be_bytes(), keys],
333    )?;
334    Ok(())
335}
336
337/// Read each prepared write's plaintext, with the keys it seals with.
338pub(crate) fn read_batch<E: From<DbError>>(
339    database: &DatabaseConnection,
340    prepared: PreparedUploads,
341    mut consume: impl FnMut(WaitingUpload<'_>) -> Result<(), E>,
342) -> Result<(), E> {
343    for PreparedUpload { rowid, write, keys } in prepared.writes {
344        consume(WaitingUpload { plaintext: read_plaintext(database, rowid, write)?, keys })?;
345    }
346    Ok(())
347}
348
349/// Queue this device's write `write`, its plaintext `record` fixed at
350/// commit, to wait for its upload, with the rows its dismissals name,
351/// `dismissed`.
352pub(crate) fn queue<'a>(
353    database: &DatabaseConnection,
354    write: WriteId,
355    record: &[u8],
356    dismissed: impl IntoIterator<Item = &'a coven_merge::RowId>,
357) -> Result<(), DbError> {
358    let number = write.number.to_be_bytes();
359    database.internal_execute(
360        "INSERT INTO _coven_uploads(device,number,record) VALUES(?1,?2,?3)",
361        params![write.device, number.as_slice(), record],
362    )?;
363    for row in dismissed {
364        database.internal_execute(
365            "INSERT INTO _coven_queued_dismissals(device,number,table_name,key,audience) VALUES(?1,?2,?3,?4,?5) ON CONFLICT DO NOTHING",
366            params![write.device, number.as_slice(), row.table, row.key, &row.audience],
367        )?;
368    }
369    Ok(())
370}
371
372/// The rows the dismissals of this device's queued write `write` name.
373pub(crate) fn dismissed(
374    database: &DatabaseConnection,
375    write: WriteId,
376) -> Result<Vec<coven_merge::RowId>, DbError> {
377    database.query(
378        "SELECT table_name,key,audience FROM _coven_queued_dismissals WHERE device=?1 AND number=?2",
379        params![write.device, write.number.to_be_bytes().as_slice()],
380        crate::row_queries::read_identity,
381    )
382}
383
384/// The writes `device`'s landed claim `claim` carried leave the queue.
385pub(crate) fn forget_claimed(
386    database: &DatabaseConnection,
387    device: DeviceId,
388    claim: u64,
389) -> Result<(), DbError> {
390    database.internal_execute(
391        "DELETE FROM _coven_uploads WHERE device=?1 AND number IN
392         (SELECT number FROM _coven_upload_claims WHERE device=?1 AND claim=?2)",
393        params![device, claim.to_be_bytes().as_slice()],
394    )?;
395    Ok(())
396}
397
398/// The header of this device's write `write` while it waits in the queue.
399pub(crate) fn queued(
400    database: &DatabaseConnection,
401    write: WriteId,
402) -> Result<Option<WriteHeaderFrame>, DbError> {
403    let Some(rowid) = database
404        .query(
405            "SELECT rowid FROM _coven_uploads WHERE device=?1 AND number=?2",
406            params![write.device, write.number.to_be_bytes()],
407            |r| r.get::<_, i64>(0),
408        )?
409        .pop()
410    else {
411        return Ok(None);
412    };
413    Ok(Some(plaintext(database, rowid, write)?.0))
414}
415
416/// Read every waiting plaintext, including attempted entries.
417/// The caller holds a reader transaction for a consistent queue view.
418pub(crate) fn each_plaintext(
419    database: &DatabaseConnection,
420    mut consume: impl FnMut(WriteHeaderFrame, UploadParts<'_>) -> Result<(), DbError>,
421) -> Result<(), DbError> {
422    let entries = database.query(
423        "SELECT rowid,device,number,NULL FROM _coven_uploads ORDER BY rowid",
424        [],
425        QueueEntry::read,
426    )?;
427    for entry in entries {
428        let UploadPlaintext { header, parts, .. } =
429            read_plaintext(database, entry.rowid, entry.write)?;
430        consume(header, parts)?;
431    }
432    Ok(())
433}
434
435#[cfg(test)]
436pub(crate) fn prepare(
437    database: &mut DatabaseConnection,
438    log: &crate::StoreState,
439    select: impl FnOnce(
440        &crate::StoreState,
441        u32,
442        &WriteHeaderFrame,
443    ) -> Result<WriteObjectPrefix, DbError>,
444) -> Result<Option<WriteId>, DbError> {
445    database.stream_transaction(|database| {
446        let Some(entry) = oldest(database)? else {
447            return Ok(None);
448        };
449        if entry.keys.is_some() {
450            return Ok(Some(entry.write));
451        }
452        let (header, _, _) = plaintext(database, entry.rowid, entry.write)?;
453        let keys = select(log, database.schema_version()?, &header)?;
454        if keys.part_keys.len() != header.parts.len() {
455            return Err(DbError::Invariant { what: "sealing selects one key per write part" });
456        }
457        keep_keys(database, entry.write, &keys.encode()?)?;
458        Ok(Some(entry.write))
459    })
460}
461
462/// The newest of this device's waiting writes it may send now: none past a
463/// raise's captured prefix, none from the first write ahead of an audience's
464/// version (§18.5), and none from the first write a bound upload holds on
465/// (§17.4). `None` with no write waiting. The caller holds
466/// one committed read, so a write is never sent past a hold it was queued
467/// with.
468pub(crate) fn sendable_through(
469    db: &DatabaseConnection,
470    device: DeviceId,
471) -> Result<Option<u64>, DbError> {
472    let newest =
473        db.query_row("SELECT max(number) FROM _coven_uploads WHERE device=?1", [device], |r| {
474            r.get::<_, Option<Vec<u8>>>(0)
475        })?;
476    let Some(newest) = newest.map(counter).transpose()? else {
477        return Ok(None);
478    };
479    let captured = crate::raise_capture::captured_through(db)?;
480    let ahead = crate::raise_capture::first_ahead(db, device)?.map(|(first, _)| first - 1);
481    let held = crate::file_queue::first_held(db)?.map(|first| first - 1);
482    Ok(Some([captured, ahead, held].into_iter().flatten().fold(newest, u64::min)))
483}
484
485/// The first own queued write at or before `through` not yet confirmed
486/// stored. A snapshot covering own writes through `through` is sent only
487/// once this is none: capture may precede confirmation, but neither settles
488/// a queue slot (§16.1).
489pub(crate) fn unconfirmed(
490    db: &DatabaseConnection,
491    through: WriteId,
492) -> Result<Option<WriteId>, DbError> {
493    Ok(db
494        .query(
495            &format!("SELECT u.number FROM _coven_uploads u JOIN _coven_writes w ON {WRITE_DEVICE_SQL}=u.device AND w.number=u.number WHERE u.device=?1 AND u.number<=?2 AND w.slot IS NULL ORDER BY u.number LIMIT 1"),
496            params![
497                through.device,
498                through.number.to_be_bytes().as_slice()
499            ],
500            |r| {
501                Ok(WriteId {
502                    device: through.device,
503                    number: counter(r.get(0)?)?,
504                })
505            },
506        )?
507        .pop())
508}
509
510#[cfg(test)]
511#[path = "upload_tests.rs"]
512pub(crate) mod tests;
coven_database/upload.rs

coven_database/
upload.rs
