coven_database/
sequence.rs

1//! One committed store prefix, including claims whose contents were refused.
2
3use crate::internal_schema::WRITE_DEVICE_SQL;
4use crate::{
5    sqlite::DatabaseConnection,
6    write_encoding::{counter, slot_counter},
7    DbError,
8};
9use coven_crypto::SigningKey;
10use coven_format::{admin::StoreChange, sequence::SlotHeader, snapshot::LastClaim};
11use coven_foundation::id_source::{DeviceId, MemberId};
12use coven_merge::{ClaimOrder, Slot, WriteId};
13use rusqlite::params;
14
15pub(crate) fn applied(db: &DatabaseConnection) -> Result<u64, DbError> {
16    db.query_row("SELECT applied FROM _coven_sequence WHERE singleton=1", [], |r| {
17        counter(r.get(0)?)
18    })
19}
20
21pub(crate) fn set_applied(db: &DatabaseConnection, slot: u64) -> Result<(), DbError> {
22    db.internal_execute(
23        "UPDATE _coven_sequence SET applied=?1 WHERE singleton=1",
24        [slot.to_be_bytes().as_slice()],
25    )?;
26    Ok(())
27}
28
29pub(crate) fn claims(db: &DatabaseConnection) -> Result<Vec<LastClaim>, DbError> {
30    db.query("SELECT device,number,digest FROM _coven_last_claims ORDER BY device", [], |r| {
31        Ok(LastClaim {
32            device: r.get::<_, DeviceId>(0)?,
33            number: counter(r.get(1)?)?,
34            digest: r.get(2)?,
35        })
36    })
37}
38
39/// Coverage orders observations from different audience snapshots and their suffixes.
40pub(crate) fn note_record(
41    db: &DatabaseConnection,
42    claim: &LastClaim,
43    through: u64,
44) -> Result<(), DbError> {
45    db.internal_execute("INSERT INTO _coven_last_claims(device,number,digest,through) VALUES(?1,?2,?3,?4) ON CONFLICT(device) DO UPDATE SET number=excluded.number,digest=excluded.digest,through=excluded.through WHERE excluded.through>=through AND excluded.number>=number",
46        params![claim.device,claim.number.to_be_bytes(),claim.digest,through.to_be_bytes()])?;
47    Ok(())
48}
49
50/// A device the store's state names starts with no claim: number zero,
51/// through slot zero. A device with a claim keeps it.
52pub(crate) fn note_device(db: &DatabaseConnection, device: DeviceId) -> Result<(), DbError> {
53    db.internal_execute("INSERT INTO _coven_last_claims(device,number,digest,through) VALUES(?1,zeroblob(8),NULL,zeroblob(8)) ON CONFLICT(device) DO NOTHING",[device])?;
54    Ok(())
55}
56
57pub(crate) fn note_claim(
58    db: &DatabaseConnection,
59    header: &SlotHeader,
60    slot: Slot,
61    digest: [u8; 32],
62) -> Result<(), DbError> {
63    note_record(
64        db,
65        &LastClaim { device: header.device, number: header.claim, digest: Some(digest) },
66        slot.get(),
67    )
68}
69
70pub(crate) fn write_slot(db: &DatabaseConnection, write: WriteId) -> Result<Option<Slot>, DbError> {
71    db.query_row(
72        &format!(
73            "SELECT (SELECT slot FROM _coven_writes WHERE {WRITE_DEVICE_SQL}=?1 AND number=?2)"
74        ),
75        params![write.device, write.number.to_be_bytes().as_slice()],
76        |r| r.get::<_, Option<Vec<u8>>>(0)?.map(slot_counter).transpose(),
77    )
78}
79
80/// Local originals have no containing slot until their batch is applied.
81pub(crate) fn write_order(db: &DatabaseConnection, write: WriteId) -> Result<ClaimOrder, DbError> {
82    Ok(match write_slot(db, write)? {
83        Some(slot) => ClaimOrder::Published { slot, number: write.number },
84        None => ClaimOrder::Local { number: write.number },
85    })
86}
87
88/// Record the irrevocable cut-off introduced by an applied admin slot.
89pub(crate) fn cutoff(
90    db: &DatabaseConnection,
91    slot: Slot,
92    change: &StoreChange,
93) -> Result<(), DbError> {
94    let (device, member, signing) = match change {
95        StoreChange::RemoveDevice { device }
96        | StoreChange::AddDevice { replaces: Some(device), .. } => (Some(*device), None, None),
97        StoreChange::RemoveMember { member } => (None, Some(*member), None),
98        StoreChange::ReplaceMemberKeys { member, expected, .. }
99        | StoreChange::RecoverMember { member, expected, .. } => {
100            (None, Some(*member), Some(expected.signing.as_bytes().to_vec()))
101        }
102        _ => return Ok(()),
103    };
104    db.internal_execute(
105        "INSERT INTO _coven_cutoffs(slot,device,member,signing) VALUES(?1,?2,?3,?4)",
106        params![slot.get().to_be_bytes().as_slice(), device, member, signing],
107    )?;
108    Ok(())
109}
110
111/// Replace every recorded cut-off with `cutoffs`, as a reload's state
112/// settles them.
113pub(crate) fn replace_cutoffs(
114    db: &DatabaseConnection,
115    cutoffs: Vec<(Slot, StoreChange)>,
116) -> Result<(), DbError> {
117    db.internal_execute("DELETE FROM _coven_cutoffs", [])?;
118    for (slot, change) in cutoffs {
119        cutoff(db, slot, &change)?;
120    }
121    Ok(())
122}
123
124/// Why a lower slot permanently excludes an original write.
125#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
126pub enum WriteCutoff {
127    /// The device was removed or replaced.
128    DeviceRemoved,
129    /// The member was removed.
130    MemberRemoved,
131    /// The member's signing and sealing keys were replaced.
132    MemberKeysReplaced,
133}
134
135/// The earliest prior removal or replacement that refuses this original.
136pub(crate) fn refusal(
137    db: &DatabaseConnection,
138    slot: Slot,
139    device: DeviceId,
140    member: MemberId,
141    signer: &SigningKey,
142) -> Result<Option<(Slot, WriteCutoff)>, DbError> {
143    Ok(db.query("SELECT slot,device,signing FROM _coven_cutoffs WHERE slot<?1 AND (device=?2 OR (member=?3 AND (signing IS NULL OR signing=?4))) ORDER BY slot LIMIT 1",
144        params![slot.get().to_be_bytes().as_slice(),device,member,signer.as_bytes().as_slice()],
145        |r| {
146            let cause = if r.get::<_,Option<Vec<u8>>>(1)?.is_some() { WriteCutoff::DeviceRemoved }
147                else if r.get::<_,Option<Vec<u8>>>(2)?.is_some() { WriteCutoff::MemberKeysReplaced }
148                else { WriteCutoff::MemberRemoved };
149            Ok((slot_counter(r.get(0)?)?, cause))
150        })?.pop())
151}
152
153pub(crate) fn refused(db: &DatabaseConnection, write: WriteId) -> Result<bool, DbError> {
154    db.query_row(&format!("SELECT EXISTS(SELECT 1 FROM _coven_writes w JOIN _coven_write_refusals r ON r.slot=w.slot WHERE {WRITE_DEVICE_SQL}=?1 AND w.number=?2)"),
155        params![write.device,write.number.to_be_bytes().as_slice()],|r|r.get(0))
156}
157
158#[cfg(test)]
159#[path = "sequence_tests.rs"]
160pub(crate) mod tests;
coven_database/sequence.rs

coven_database/
sequence.rs
