coven_database/
store_ack_records.rs

1use crate::query_mapped_rows;
2use coven_protocol::store_commit::CircleAck;
3
4use super::*;
5
6pub(crate) fn verify_next_local_store_ack_on(
7    conn: &Connection,
8    authority: &coven_protocol::store_commit::ReferencedStoreDeviceRegistration,
9    bytes: &[u8],
10    prepared: &PreparedExactObject,
11) -> Result<(StoreAckRef, StoreAck), DbError> {
12    let registration_ref = authority.reference();
13    let registration = authority.value();
14    let root = &registration.store_root;
15    let unverified: StoreAck = serde_json::from_slice(bytes)
16        .map_err(|error| DbError::context("parse Store acknowledgement", error))?;
17    if &unverified.registration != registration_ref {
18        return Err(DbError::Message(
19            "Store acknowledgement author differs from local activation".to_string(),
20        ));
21    }
22    let reference = StoreAckRef {
23        registration: registration_ref.clone(),
24        sequence: unverified.sequence,
25        ack_hash: unverified.ack_hash(),
26        object: prepared.reference().clone(),
27    };
28    let ack = StoreAck::parse_at(bytes, root, &reference, registration)
29        .map_err(|error| DbError::context("verify Store acknowledgement", error))?;
30    let previous = load_published_store_ack_on(conn)?;
31    let (expected_sequence, expected_predecessor, expected_slot) = match &previous {
32        Some(previous) => (
33            previous.reference.sequence.checked_add(1).ok_or_else(|| {
34                DbError::Message("Store acknowledgement sequence overflow".to_string())
35            })?,
36            Some(previous.reference.object.clone()),
37            previous.successor_slot.clone(),
38        ),
39        None => (1, None, store_ack_first_slot(registration)?.clone()),
40    };
41    if ack.sequence != expected_sequence
42        || ack.successor.predecessor != expected_predecessor
43        || prepared.reference().slot() != &expected_slot
44    {
45        return Err(DbError::Message(
46            "Store acknowledgement does not extend the exact local stream".to_string(),
47        ));
48    }
49    let next_sequence = ack
50        .sequence
51        .checked_add(1)
52        .ok_or_else(|| DbError::Message("Store acknowledgement sequence overflow".to_string()))?;
53    if ack.successor.activation
54        != registration
55            .store_acknowledgement_activation(registration_ref)
56            .map_err(DbError::from)?
57            .activation_id()
58        || ack.successor.next_slot.logical_key()
59            != format!(
60                "{}.json",
61                ack_slot_prefix(&registration.device_id.to_string(), next_sequence)
62            )
63    {
64        return Err(DbError::Message(
65            "Store acknowledgement successor is outside its activated exact stream".to_string(),
66        ));
67    }
68    Ok((reference, ack))
69}
70
71pub(crate) fn store_ack_first_slot(
72    registration: &StoreDeviceRegistration,
73) -> Result<&coven_protocol::objects::ObjectSlot, DbError> {
74    match &registration.acknowledgements {
75        coven_protocol::store_commit::DeviceStreamAnchor::StoreAcknowledgements { first_slot } => {
76            Ok(first_slot)
77        }
78        _ => Err(DbError::Message(
79            "local Store registration has no acknowledgement stream anchor".to_string(),
80        )),
81    }
82}
83
84pub fn store_snapshot_first_slot(
85    registration: &StoreDeviceRegistration,
86) -> Result<&coven_protocol::objects::ObjectSlot, DbError> {
87    match &registration.snapshots {
88        coven_protocol::store_commit::DeviceStreamAnchor::StoreSnapshots { first_slot } => {
89            Ok(first_slot)
90        }
91        _ => Err(DbError::Message(
92            "local Store registration has no snapshot stream anchor".to_string(),
93        )),
94    }
95}
96
97pub(crate) fn load_published_store_ack_on(
98    conn: &Connection,
99) -> Result<Option<PublishedStoreAck>, DbError> {
100    conn.query_row(
101        "SELECT ack_ref, successor_slot, standing FROM published_store_acks \
102         WHERE singleton = 1",
103        [],
104        |row| {
105            Ok((
106                row.get::<_, String>(0)?,
107                row.get::<_, String>(1)?,
108                row.get::<_, Option<String>>(2)?,
109            ))
110        },
111    )
112    .optional()
113    .map_err(DbError::from)?
114    .map(|(reference, successor_slot, standing)| {
115        let reference: StoreAckRef = serde_json::from_str(&reference)
116            .map_err(|error| DbError::context("published Store acknowledgement ref", error))?;
117        if reference.sequence == 0 {
118            return Err(DbError::Message(
119                "published Store acknowledgement uses sequence zero".to_string(),
120            ));
121        }
122        Ok(PublishedStoreAck {
123            reference,
124            successor_slot: serde_json::from_str(&successor_slot).map_err(|error| {
125                DbError::context("published Store acknowledgement successor slot", error)
126            })?,
127            standing: standing
128                .map(|state| {
129                    serde_json::from_str(&state)
130                        .map_err(|error| DbError::context("standing Store acknowledgement", error))
131                })
132                .transpose()?,
133        })
134    })
135    .transpose()
136}
137
138pub(crate) fn finish_outbound_store_ack_on(
139    conn: &Connection,
140    reference: &StoreAckRef,
141    successor_slot: &coven_protocol::objects::ObjectSlot,
142    standing: &coven_protocol::store_commit::StandingStoreAck,
143) -> Result<(), DbError> {
144    let removed = conn
145        .execute(
146            "DELETE FROM outbound_store_acks WHERE singleton = 1 AND ack_ref = ?1",
147            [serde_json::to_string(reference)
148                .map_err(|error| DbError::context("serialize Store acknowledgement ref", error))?],
149        )
150        .map_err(DbError::from)?;
151    if removed != 1 {
152        return Err(DbError::Message(
153            "outbound Store acknowledgement disappeared".to_string(),
154        ));
155    }
156    let successor_slot = serde_json::to_string(successor_slot).map_err(|error| {
157        DbError::context("serialize Store acknowledgement successor slot", error)
158    })?;
159    let standing = serde_json::to_string(standing)
160        .map_err(|error| DbError::context("serialize standing Store acknowledgement", error))?;
161    conn.execute(
162        "INSERT INTO published_store_acks (singleton, ack_ref, successor_slot, standing) \
163         VALUES (1, ?1, ?2, ?3) \
164         ON CONFLICT(singleton) DO UPDATE SET \
165           ack_ref = excluded.ack_ref, successor_slot = excluded.successor_slot, \
166           standing = excluded.standing",
167        (
168            serde_json::to_string(reference).map_err(|error| {
169                DbError::context("serialize published Store acknowledgement ref", error)
170            })?,
171            successor_slot,
172            standing,
173        ),
174    )
175    .map(|_| ())
176    .map_err(DbError::from)
177}
178
179pub(crate) fn load_outbound_circle_acks_on(
180    conn: &Connection,
181    authority: &coven_protocol::store_commit::ReferencedStoreDeviceRegistration,
182) -> Result<Vec<coven_protocol::prepared_commit::CircleAckActivation>, DbError> {
183    let registration = authority.value();
184    let root = &registration.store_root;
185    let rows = query_mapped_rows(
186        conn,
187        "SELECT ack_ref, ack_bytes, prepared_object FROM outbound_circle_acks
188             ORDER BY circle_id",
189        [],
190        |row| {
191            Ok((
192                row.get::<_, String>(0)?,
193                row.get::<_, Vec<u8>>(1)?,
194                row.get::<_, String>(2)?,
195            ))
196        },
197    )?;
198    let mut activations = Vec::with_capacity(rows.len());
199    for (reference, bytes, prepared) in rows {
200        let reference: coven_protocol::store_commit::CircleAckRef =
201            serde_json::from_str(&reference)
202                .map_err(|error| DbError::context("outbound Circle acknowledgement ref", error))?;
203        let prepared: PreparedExactObject = serde_json::from_str(&prepared)
204            .map_err(|error| DbError::context("outbound prepared Circle acknowledgement", error))?;
205        if prepared.reference() != &reference.object {
206            return Err(DbError::Message(
207                "outbound Circle acknowledgement ref differs from its prepared object".to_string(),
208            ));
209        }
210        let value = CircleAck::parse_at(&bytes, root, &reference, registration)
211            .map_err(|error| DbError::context("outbound Circle acknowledgement", error))?;
212        activations.push(coven_protocol::prepared_commit::CircleAckActivation {
213            reference,
214            ack: ExactProtocolObject {
215                value,
216                bytes,
217                prepared,
218            },
219        });
220    }
221    Ok(activations)
222}
223
224/// Load the singleton outbound Store acknowledgement and require it to match
225/// `expected`; `mismatch` is the error when another acknowledgement is queued.
226pub(crate) fn load_expected_outbound_store_ack_on(
227    conn: &Connection,
228    authority: &coven_protocol::store_commit::ReferencedStoreDeviceRegistration,
229    expected: &coven_protocol::store_commit::StoreAckRef,
230    mismatch: &str,
231) -> Result<OutboundStoreAck, DbError> {
232    let outbound = load_outbound_store_ack_on(conn, authority)?
233        .ok_or_else(|| DbError::Message("outbound Store acknowledgement is absent".to_string()))?;
234    if &outbound.reference != expected {
235        return Err(DbError::Message(mismatch.to_string()));
236    }
237    Ok(outbound)
238}
239
240/// Write the singleton outbound acknowledgement's activation column, keyed by
241/// the acknowledgement reference it must still carry; `missing` is the error
242/// when that row is gone.
243pub(crate) fn set_outbound_store_ack_activation_on(
244    conn: &Connection,
245    expected: &coven_protocol::store_commit::StoreAckRef,
246    activation: &crate::OutboundStoreAckActivation,
247    missing: &str,
248) -> Result<(), DbError> {
249    let activation = serde_json::to_string(activation).map_err(|error| {
250        DbError::context("serialize Merge Store acknowledgement activation", error)
251    })?;
252    let updated = conn
253        .execute(
254            "UPDATE outbound_store_acks SET activation = ?2 \
255             WHERE singleton = 1 AND ack_ref = ?1",
256            rusqlite::params![
257                serde_json::to_string(expected).map_err(|error| DbError::context(
258                    "serialize Store acknowledgement ref",
259                    error
260                ))?,
261                activation,
262            ],
263        )
264        .map_err(DbError::from)?;
265    if updated != 1 {
266        return Err(DbError::Message(missing.to_string()));
267    }
268    Ok(())
269}
270
271pub(crate) fn load_outbound_store_ack_on(
272    conn: &Connection,
273    authority: &coven_protocol::store_commit::ReferencedStoreDeviceRegistration,
274) -> Result<Option<OutboundStoreAck>, DbError> {
275    conn.query_row(
276        "SELECT ack_ref, ack_bytes, prepared_object, activation \
277         FROM outbound_store_acks WHERE singleton = 1",
278        [],
279        |row| {
280            Ok((
281                row.get::<_, String>(0)?,
282                row.get::<_, Vec<u8>>(1)?,
283                row.get::<_, String>(2)?,
284                row.get::<_, String>(3)?,
285            ))
286        },
287    )
288    .optional()
289    .map_err(DbError::from)?
290    .map(|(reference, bytes, prepared, activation)| {
291        let reference: StoreAckRef = serde_json::from_str(&reference)
292            .map_err(|error| DbError::context("outbound Store acknowledgement ref", error))?;
293        let prepared: PreparedExactObject = serde_json::from_str(&prepared)
294            .map_err(|error| DbError::context("outbound prepared Store acknowledgement", error))?;
295        let activation: OutboundStoreAckActivation =
296            serde_json::from_str(&activation).map_err(|error| {
297                DbError::context("outbound Store acknowledgement activation", error)
298            })?;
299        if prepared.reference() != &reference.object {
300            return Err(DbError::Message(
301                "outbound Store acknowledgement ref differs from its prepared object".to_string(),
302            ));
303        }
304        let author_ref = authority.reference();
305        let author = authority.value();
306        let value = StoreAck::parse_at(&bytes, &author.store_root, &reference, author)
307            .map_err(|error| DbError::context("outbound Store acknowledgement", error))?;
308        if &value.registration != author_ref {
309            return Err(DbError::Message(
310                "outbound Store acknowledgement author differs from local activation".to_string(),
311            ));
312        }
313        Ok(OutboundStoreAck {
314            reference,
315            ack: ExactProtocolObject {
316                value,
317                bytes,
318                prepared,
319            },
320            circle_acknowledgements: load_outbound_circle_acks_on(conn, authority)?,
321            activation,
322        })
323    })
324    .transpose()
325}
coven_database/store_ack_records.rs

coven_database/
store_ack_records.rs
