coven_database/
download_stream.rs

1//! Incremental write parts supplied by synchronization.
2
3use crate::SnapshotError;
4use coven_format::chunks::CHUNK_SIZE;
5use coven_format::dismissal::WriteFrame;
6use coven_format::write_stream::{PartDecoder, WriteHeaderFrame};
7use std::io::Read;
8
9/// A checked write header and its audience streams. The database commits only
10/// after sync confirms authentication. Parts follow the header's order and count.
11pub struct DownloadedWriteStream<R> {
12    /// Authenticated identity, causality and plaintext stream boundaries.
13    pub header: WriteHeaderFrame,
14    /// Opened plaintext or an explicit unreadable audience, in header order.
15    pub parts: Vec<DownloadedPartStream<R>>,
16    /// The key that authenticated this original in its authored view.
17    pub signer: coven_crypto::SigningKey,
18    /// The signed header evidence and each opened part's keyed leaf hasher.
19    pub authorship: WriteAuthorship,
20}
21
22/// A write's signed header and part roots, and the hashers that check each
23/// opened part against its root (D9, D11). The database checks the roots;
24/// sync checks the header signature with the rest of the object.
25#[derive(Clone)]
26pub struct WriteAuthorship {
27    /// The cleartext prefix, sealed header, roots and header signature.
28    pub evidence: crate::HeaderEvidence,
29    /// Each part's keyed leaf hasher, in header order, for the opened parts.
30    pub hashers: Vec<Option<coven_crypto::AuthorshipHasher>>,
31}
32
33impl WriteAuthorship {
34    fn root(&self, part: usize) -> Result<crate::authorship::PartRoot, crate::DbError> {
35        let hasher =
36            self.hashers.get(part).cloned().flatten().ok_or(crate::DbError::Invariant {
37                what: "an opened part has its authorship hasher",
38            })?;
39        let root =
40            *self.evidence.roots.get(part).ok_or_else(|| {
41                crate::snapshot_error::invalid("write evidence lacks a part's root")
42            })?;
43        Ok(crate::authorship::PartRoot::new(hasher, root))
44    }
45}
46
47fn hash_frame(
48    root: &mut crate::authorship::PartRoot,
49    frame: &WriteFrame,
50) -> Result<(), crate::DbError> {
51    match frame.record() {
52        Some(record) => root.record(record),
53        None => Ok(()),
54    }
55}
56
57/// Whether sync could decrypt the corresponding part in the header.
58pub enum DownloadedPartStream<R> {
59    /// A complete plaintext part stream, starting at its first byte.
60    Opened(R),
61    /// The part is outside this installation's current audience.
62    Skipped,
63}
64
65pub(crate) struct PreparedWrite<'a> {
66    header: WriteHeaderFrame,
67    parts: Vec<Option<PreparedPart<'a>>>,
68    signer: coven_crypto::SigningKey,
69    evidence: crate::HeaderEvidence,
70}
71pub(crate) struct PreparedPart<'a> {
72    records: crate::input_preparation::PreparedRecords<'a>,
73    hashes: Vec<[u8; 32]>,
74}
75impl<R: Read> DownloadedWriteStream<R> {
76    pub(crate) fn prepare<'a>(
77        self,
78        inputs: &crate::input_preparation::PreparedInputs<'a>,
79    ) -> Result<PreparedWrite<'a>, crate::DbError> {
80        if self.header.parts.len() != self.parts.len()
81            || self.authorship.evidence.roots.len() != self.parts.len()
82        {
83            return Err(
84                SnapshotError::Inconsistent("write stream count differs from its header").into()
85            );
86        }
87        let mut parts = Vec::new();
88        for (index, (header, stream)) in self.header.parts.iter().zip(self.parts).enumerate() {
89            match stream {
90                DownloadedPartStream::Skipped => parts.push(None),
91                DownloadedPartStream::Opened(input) => {
92                    let mut records = inputs.records();
93                    let mut root = self.authorship.root(index)?;
94                    read_part(input, header.clone(), |frame| {
95                        hash_frame(&mut root, &frame)?;
96                        records.push(coven_format::merge_fields::encode_write_frame(&frame)?)
97                    })?;
98                    parts.push(Some(PreparedPart { records, hashes: root.finish()? }));
99                }
100            }
101        }
102        Ok(PreparedWrite {
103            header: self.header,
104            parts,
105            signer: self.signer,
106            evidence: self.authorship.evidence,
107        })
108    }
109}
110impl PreparedWrite<'_> {
111    /// The write's header frame.
112    pub(crate) fn header(&self) -> &WriteHeaderFrame {
113        &self.header
114    }
115    /// The key the write was signed with.
116    pub(crate) fn signer(&self) -> &coven_crypto::SigningKey {
117        &self.signer
118    }
119    /// The audiences of parts this installation could not open.
120    pub(crate) fn unopened_audiences(&self) -> impl Iterator<Item = &coven_merge::Audience> {
121        self.header
122            .parts
123            .iter()
124            .zip(&self.parts)
125            .filter(|(_, part)| part.is_none())
126            .map(|(header, _)| &header.audience)
127    }
128}
129
130/// Stage one prepared original into the store's connection, inside its
131/// containing slot's transaction. The caller materializes the complete batch
132/// only after all parts authenticate. Before the write merges, or is consumed
133/// refused, `reach` puts back the kept rows it reaches beside the queue
134/// (§6.2); a reload passes none, having put back every kept row first.
135pub(crate) fn stage_selected(
136    database: &crate::sqlite::DatabaseConnection,
137    schema: &crate::write_schema::WriteSchema,
138    write: PreparedWrite<'_>,
139    slot: coven_merge::Slot,
140    refused: bool,
141    selected: impl Fn(&coven_merge::Audience) -> bool,
142    reach: Option<&mut crate::queued_writes::Reach>,
143) -> Result<(), crate::DbError> {
144    let opened: Vec<_> = write
145        .header
146        .parts
147        .iter()
148        .zip(&write.parts)
149        .filter(|(part, stream)| selected(&part.audience) && stream.is_some())
150        .map(|(part, _)| part.audience.clone())
151        .collect();
152    // A write the cut-off consumes is never judged, so what it claims to
153    // have read decides nothing (§14.1).
154    if !refused {
155        crate::write_commit::check_reads(database, &write.header.header)?;
156        if let Some(wait) = crate::download::prerequisite(database, &write.header.header, &opened)?
157        {
158            return Err(crate::DbError::WriteWaiting(wait));
159        }
160    }
161    if !refused {
162        crate::merge_store::MergeStore::from_staged(database, &schema.schema).apply(
163            &coven_format::write::WriteRecord {
164                header: write.header.header.clone(),
165                parts: vec![],
166            },
167        )?;
168    }
169    crate::forget_entries::superseded(database, write.header.header.position)?;
170    crate::write_commit::retain_original(database, &write.header.header, Some(slot))?;
171    let original_header = write.header.clone();
172    let form = crate::write_sources::KeptForm::new(schema, original_header.header.schema_version);
173    let boundaries = crate::write_boundary::WriteBoundary::judging(database)?;
174    let mut dismissed = Vec::new();
175    read_authenticated(database, write, |header, audience, frame| {
176        if !selected(audience) {
177            return Ok(());
178        }
179        if let WriteFrame::Dismissal(dismissal) = &frame {
180            dismissed.push(dismissal.row.clone());
181        }
182        if refused {
183            if let Some(record) = frame.record() {
184                crate::write_sources::retain(database, &form, header, record)?;
185                crate::write_sources::admit(
186                    database,
187                    header,
188                    record.row(),
189                    &coven_format::snapshot_rows::ChangeVerdict::Refused,
190                )?;
191            }
192            return Ok(());
193        }
194        let ignored =
195            crate::write_boundary::WriteBoundary::ignore(&boundaries, header, audience, Some(slot));
196        crate::download_stage::frame(database, &form, header, ignored, frame)
197    })?;
198    if let Some(reach) = reach {
199        crate::queued_writes::put_back(
200            database,
201            schema,
202            &original_header.header,
203            reach,
204            dismissed,
205        )?;
206    }
207    if refused {
208        crate::forget_records::none_current(
209            database,
210            original_header.header.position,
211            &opened,
212            slot,
213        )?;
214    } else {
215        crate::download_stage::stage(
216            database,
217            schema,
218            &original_header.header,
219            &opened,
220            Some(slot),
221            |audience| Ok(selected(audience)),
222        )?;
223    }
224    crate::download_stage::release(database, &original_header.header)
225}
226
227/// Retain a prepared write's authenticated source evidence in the store's
228/// connection without changing rows or admission.
229#[cfg(test)]
230pub(crate) fn retain_sources(
231    db: &crate::sqlite::DatabaseConnection,
232    schema: &crate::write_schema::WriteSchema,
233    write: PreparedWrite<'_>,
234) -> Result<(), crate::DbError> {
235    crate::download_stage::begin(db)?;
236    let header = write.header.header.clone();
237    let form = crate::write_sources::KeptForm::new(schema, header.schema_version);
238    read_authenticated(db, write, |header, _, frame| {
239        if let Some(record) = frame.record() {
240            crate::write_sources::retain(db, &form, header, record)?;
241        }
242        Ok(())
243    })?;
244    crate::download_stage::release(db, &header)
245}
246
247/// Hand each opened part's prepared records to `consume`, then keep the
248/// write's plaintext header, and, for another member's write, each part's
249/// leaf hashes and the header evidence in the store's connection
250/// (`authorship::keeps_evidence`).
251fn read_authenticated(
252    database: &crate::sqlite::DatabaseConnection,
253    write: PreparedWrite<'_>,
254    mut consume: impl FnMut(
255        &coven_format::write::WriteHeader,
256        &coven_merge::Audience,
257        WriteFrame,
258    ) -> Result<(), crate::DbError>,
259) -> Result<(), crate::DbError> {
260    let position = write.header.header.position;
261    let evidence = crate::authorship::keeps_evidence(database, position)?;
262    for (header, part) in write.header.parts.iter().zip(write.parts) {
263        let Some(part) = part else {
264            continue;
265        };
266        part.records.each(|bytes| {
267            consume(&write.header.header, &header.audience, WriteFrame::decode(&bytes)?)
268        })?;
269        if evidence {
270            crate::authorship::retain_leaves(database, position, &header.audience, &part.hashes)?;
271        }
272    }
273    crate::write_headers::retain(database, &write.header)?;
274    if evidence {
275        crate::authorship::retain_header(database, position, &write.evidence)?;
276    }
277    Ok(())
278}
279
280/// Decode and validate one part without collecting its row frames. Every
281/// consumer checks the same chunk lengths, record counts, ordering and EOF.
282pub(crate) fn read_part<R: Read, E: From<SnapshotError> + From<coven_format::Error>>(
283    mut input: R,
284    header: coven_format::write_stream::PartHeader,
285    consume: impl FnMut(WriteFrame) -> Result<(), E>,
286) -> Result<(), E> {
287    visit_part(
288        header,
289        |chunk| input.read_exact(chunk).map_err(|error| SnapshotError::Read(error).into()),
290        consume,
291    )?;
292    let mut trailing = [0];
293    loop {
294        match input.read(&mut trailing) {
295            Ok(0) => return Ok(()),
296            Ok(_) => return Err(coven_format::Error::TrailingBytes.into()),
297            Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
298            Err(error) => return Err(SnapshotError::Read(error).into()),
299        }
300    }
301}
302
303/// Visit a part from either an I/O stream or the upload queue's SQLite BLOB.
304/// The source fills each requested chunk exactly; the decoder checks framing,
305/// ordering, audience and record count before completing the part.
306pub(crate) fn visit_part<E: From<coven_format::Error>>(
307    header: coven_format::write_stream::PartHeader,
308    mut read_chunk: impl FnMut(&mut [u8]) -> Result<(), E>,
309    mut consume: impl FnMut(WriteFrame) -> Result<(), E>,
310) -> Result<(), E> {
311    let mut left = header.plaintext_length;
312    let mut decoder = PartDecoder::new(header)?;
313    let mut chunk = [0; CHUNK_SIZE];
314    while left > 0 {
315        let length = left.min(CHUNK_SIZE as u64) as usize;
316        read_chunk(&mut chunk[..length])?;
317        for frame in decoder.chunk(&chunk[..length])? {
318            consume(frame)?;
319        }
320        left -= length as u64;
321    }
322    decoder.finish()?;
323    Ok(())
324}
coven_database/download_stream.rs

coven_database/
download_stream.rs
