coven_database/store/store_session/
pending_publication.rs

1use super::{
2    candidate_records::parse_prepared_merge_candidate_parts_on,
3    publication_state::PreparedStoreWriteState, StoreDatabase, StoreSession,
4};
5use crate::{
6    load_prepared_audience_objects_on, DbError, ExactProtocolObject, PreparedStoreWriteCommit,
7    StoreWriteBase,
8};
9use coven_protocol::membership::AuthorStreamId;
10use coven_protocol::store_commit::{
11    CommitFrontier, StoreBatchCommit, StoreBatchCommitRef, StoreCommitCoord, StoreDeviceHead,
12    StoreDeviceRegistrationRef, VerifiedStoreBatchCommit,
13};
14use coven_protocol::write::WriteId;
15use rusqlite::OptionalExtension;
16use std::collections::BTreeMap;
17
18/// One reading of the local ledger: the author's own latest position, the
19/// materialized frontier it belongs to, and this device's turn to author the
20/// commit that extends it. See [`StoreDatabase::local_commit_base`].
21///
22/// The turn is part of the reading rather than something a caller remembers to
23/// take: the position is only true for as long as no other local writer can
24/// take it. Hold this value until the commit composed from it has published its
25/// head, or until the candidate is durably persisted for a later publisher to
26/// activate.
27pub struct LocalCommitBase {
28    authorship: super::OwnStreamAuthorship,
29    predecessor: Option<StoreBatchCommitRef>,
30    frontier: BTreeMap<String, StoreBatchCommitRef>,
31}
32
33impl LocalCommitBase {
34    pub fn into_parts(
35        self,
36    ) -> (
37        super::OwnStreamAuthorship,
38        Option<StoreBatchCommitRef>,
39        BTreeMap<String, StoreBatchCommitRef>,
40    ) {
41        (self.authorship, self.predecessor, self.frontier)
42    }
43}
44
45impl StoreSession<'_> {
46    fn local_commit_ledger_base(
47        &self,
48        stream_id: &AuthorStreamId,
49    ) -> Result<
50        (
51            Option<StoreBatchCommitRef>,
52            BTreeMap<String, StoreBatchCommitRef>,
53        ),
54        DbError,
55    > {
56        let stream_id = stream_id.to_string();
57        Ok((
58            crate::store::materialized_commit_index::latest_position_for_device_on(
59                self.conn, &stream_id,
60            )?,
61            crate::store::materialized_commit_index::materialized_frontier_on(self.conn, None)?,
62        ))
63    }
64
65    fn latest_local_store_position(
66        &self,
67        stream_id: &str,
68    ) -> Result<Option<StoreBatchCommitRef>, DbError> {
69        crate::store::materialized_commit_index::latest_position_for_device_on(self.conn, stream_id)
70    }
71
72    fn oldest_prepared_store_write(&mut self) -> Result<Option<PreparedStoreWriteCommit>, DbError> {
73        let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
74        let row = self
75            .conn
76            .query_row(
77                "SELECT write_id, base, prepared FROM store_writes
78                 WHERE prepared IS NOT NULL
79                   AND status = '\"publishing\"'
80                 ORDER BY ordinal LIMIT 1",
81                [],
82                |row| {
83                    Ok((
84                        row.get::<_, String>(0)?,
85                        row.get::<_, Option<String>>(1)?,
86                        row.get::<_, String>(2)?,
87                    ))
88                },
89            )
90            .optional()
91            .map_err(DbError::from)?;
92        row.map(|(write_id, base, prepared)| {
93            let base = base.ok_or_else(|| {
94                DbError::Message(format!(
95                    "publishing write {write_id} carries no commit base"
96                ))
97            })?;
98            let prepared: PreparedStoreWriteState = serde_json::from_str(&prepared)
99                .map_err(|error| DbError::context("prepared Store write", error))?;
100            let (commit, head, graph_commit) = match &prepared {
101                PreparedStoreWriteState::Publication { commit, head, .. } => (commit, head, None),
102                PreparedStoreWriteState::MergeAbandonment {
103                    candidate_commit,
104                    authority_commit,
105                    authority_head,
106                    ..
107                } => (authority_commit, authority_head, Some(candidate_commit)),
108            };
109            let write_id = WriteId::from_generated(write_id);
110            let unverified_commit: StoreBatchCommit =
111                serde_json::from_slice(commit.semantic_bytes())
112                    .map_err(|error| DbError::context("prepared Store commit", error))?;
113            if unverified_commit.write_id != write_id {
114                return Err(DbError::Message(
115                    "prepared write id differs from signed commit".to_string(),
116                ));
117            }
118            let registration_ref = &unverified_commit.author_registration;
119            let stored_registration_ref: String = self
120                .conn
121                .query_row(
122                    "SELECT registration_object \
123                         FROM store_device_registration_activations \
124                         WHERE device_id = ?1 AND registration_hash = ?2",
125                    (
126                        registration_ref.device_id.to_string(),
127                        registration_ref.registration_hash.to_string(),
128                    ),
129                    |row| row.get(0),
130                )
131                .map_err(DbError::from)?;
132            let stored_registration_ref: StoreDeviceRegistrationRef =
133                serde_json::from_str(&stored_registration_ref)
134                    .map_err(|error| DbError::context("prepared write registration ref", error))?;
135            if stored_registration_ref != *registration_ref {
136                return Err(DbError::Message(
137                    "prepared commit registration differs from its activation".to_string(),
138                ));
139            }
140            let authority = self.activated_registration(registration_ref)?;
141            let registration = authority.value();
142            let root = &registration.store_root;
143            let stream_id =
144                coven_protocol::store_commit::StreamActivation::device_authorized_stream_id(
145                    root.store_root_hash,
146                    registration_ref,
147                    coven_protocol::store_commit::StreamAnchorDomain::StoreAnnouncements,
148                );
149            let coord = StoreCommitCoord {
150                stream_id,
151                sequence: unverified_commit.seq(),
152            };
153            let commit_value = VerifiedStoreBatchCommit::parse_prepared(
154                commit.semantic_bytes(),
155                root.store_root_hash,
156                coord,
157                commit.prepared().reference().clone(),
158                registration,
159            )
160            .map_err(|error| DbError::context("verify prepared Store commit", error))?;
161            let commit_ref = commit_value.reference().clone();
162            let head_value = StoreDeviceHead::parse_at(
163                head.semantic_bytes(),
164                root.store_root_hash,
165                registration,
166                &commit_ref,
167            )
168            .map_err(|error| DbError::context("verify prepared Store head", error))?;
169            let base: StoreWriteBase = serde_json::from_str(&base)
170                .map_err(|error| DbError::context("prepared write base", error))?;
171            let mut dependencies = CommitFrontier::from_refs(base.dependencies)
172                .map_err(|error| DbError::context("prepared dependency frontier", error))?;
173            let observed_predecessor = dependencies.0.remove(&stream_id);
174            if dependencies.commits() != commit_value.merge_dependencies() {
175                return Err(DbError::Message(
176                    "prepared commit differs from its write dependency frontier".to_string(),
177                ));
178            }
179            if observed_predecessor.as_ref().is_some_and(|captured| {
180                commit_value.order.predecessor().is_none_or(|current| {
181                    current.coord.sequence() < captured.coord.sequence()
182                        || current.coord.sequence() == captured.coord.sequence()
183                            && current != captured
184                })
185            }) {
186                return Err(DbError::Message(
187                    "prepared commit predecessor does not cover its write capture frontier"
188                        .to_string(),
189                ));
190            }
191            let partitions = records.store_write_partitions(write_id.as_str())?;
192            let audiences =
193                load_prepared_audience_objects_on(self.conn, self.store_dir, &write_id)?;
194            let graph_commit = match graph_commit {
195                Some(graph_commit) => {
196                    let candidate_head = match &prepared {
197                        PreparedStoreWriteState::MergeAbandonment { candidate_head, .. } => {
198                            candidate_head
199                        }
200                        _ => unreachable!("matched Merge abandonment"),
201                    };
202                    let candidate = parse_prepared_merge_candidate_parts_on(
203                        records,
204                        self.verified_store_authority,
205                        graph_commit.semantic_bytes(),
206                        graph_commit.prepared().reference(),
207                        candidate_head.semantic_bytes(),
208                        candidate_head.prepared().reference(),
209                    )?;
210                    candidate.commit
211                }
212                None => commit_value.clone(),
213            };
214            let expected_package_count = usize::from(graph_commit.store_package().is_some())
215                .checked_add(graph_commit.circle_packages().len())
216                .ok_or_else(|| DbError::Message("package count overflow".to_string()))?;
217            if audiences.packages.len() != expected_package_count
218                || audiences.packages.len()
219                    != usize::from(partitions.store.is_some()) + partitions.circles.len()
220            {
221                return Err(DbError::Message(
222                    "prepared package indexes do not exactly cover commit audiences".to_string(),
223                ));
224            }
225            for package in &audiences.packages {
226                let value = package.package();
227                if value.write_id() != &write_id
228                    || value.commit_coord() != &commit_ref.coord
229                    || value.candidate_family() != commit_value.candidate_family()
230                {
231                    return Err(DbError::Message(
232                        "indexed audience package differs from its exact commit".to_string(),
233                    ));
234                }
235                let expected_object = match value.audience() {
236                    coven_protocol::audience_package::PackageAudience::Store => {
237                        graph_commit
238                            .verify_store_package(package.semantic_bytes())
239                            .map_err(DbError::from)?;
240                        &graph_commit
241                            .store_package()
242                            .as_ref()
243                            .expect("verified present")
244                            .object
245                    }
246                    coven_protocol::audience_package::PackageAudience::Circle {
247                        circle_id, ..
248                    } => {
249                        graph_commit
250                            .verify_circle_package(*circle_id, package.semantic_bytes())
251                            .map_err(DbError::from)?;
252                        &graph_commit
253                            .circle_packages()
254                            .iter()
255                            .find(|entry| entry.circle_id == *circle_id)
256                            .expect("verified present")
257                            .package
258                            .object
259                    }
260                };
261                if package.object() != expected_object {
262                    return Err(DbError::Message(
263                        "indexed audience package exact object differs from its commit".to_string(),
264                    ));
265                }
266            }
267            for package in &audiences.packages {
268                let audience = package.package().audience().remote_audience();
269                for binding in package.package().blob_bindings() {
270                    if !audiences
271                        .blobs
272                        .iter()
273                        .any(|blob| blob.audience() == &audience && blob.blob() == binding.blob())
274                    {
275                        return Err(DbError::Message(
276                            "prepared package blob binding has no exact blob index".to_string(),
277                        ));
278                    }
279                }
280            }
281            for blob in &audiences.blobs {
282                if !audiences.packages.iter().any(|package| {
283                    package.package().audience().remote_audience() == *blob.audience()
284                        && package
285                            .package()
286                            .blob_bindings()
287                            .iter()
288                            .any(|binding| binding.blob() == blob.blob())
289                }) {
290                    return Err(DbError::Message(
291                        "prepared blob index has no exact package binding".to_string(),
292                    ));
293                }
294            }
295            Ok(PreparedStoreWriteCommit {
296                audiences,
297                commit: ExactProtocolObject {
298                    value: commit_value,
299                    bytes: commit.semantic_bytes().to_vec(),
300                    prepared: commit.prepared().clone(),
301                },
302                head: ExactProtocolObject {
303                    value: head_value,
304                    bytes: head.semantic_bytes().to_vec(),
305                    prepared: head.prepared().clone(),
306                },
307            })
308        })
309        .transpose()
310    }
311}
312
313impl StoreDatabase {
314    pub async fn oldest_prepared_store_write(
315        &self,
316    ) -> Result<Option<PreparedStoreWriteCommit>, DbError> {
317        let loaded = self
318            .call_store(move |session| session.oldest_prepared_store_write())
319            .await?;
320        if let Some(batch) = &loaded {
321            for blob in &batch.audiences.blobs {
322                if let Some(spool_path) = blob.spool_path() {
323                    {
324                        let (size, digest) = coven_foundation::local_file::file_facts(spool_path)
325                            .await
326                            .map_err(|error| DbError::context("prepared blob spool", error))?;
327                        blob.blob()
328                            .object()
329                            .verify_stored_facts(
330                                spool_path,
331                                size,
332                                coven_protocol::store_commit::ObjectHash::from_digest(digest),
333                            )
334                            .map_err(|error| DbError::context("prepared blob spool", error))?;
335                    }
336                }
337            }
338        }
339        Ok(loaded)
340    }
341
342    /// The local device's own latest position and the materialized frontier
343    /// that position belongs to, read as one state of the ledger.
344    ///
345    /// A commit order names one history, and both halves of it come from the
346    /// same table. Reading them separately lets one of this device's own
347    /// activations land in between, which leaves its own stream in the frontier
348    /// one commit ahead of the position it extends. Such an order has no
349    /// predecessor cut at all — the cut is the frontier with the predecessor
350    /// inserted, and those two then contradict each other on the author's own
351    /// stream — so every operation composed from it is refused. The device
352    /// driving an operation also runs its sync loop, so that is the ordinary
353    /// case rather than a hostile one.
354    ///
355    /// Taking this device's turn to author its own stream is part of the read:
356    /// the position returned stays this device's next position for as long as
357    /// the returned `LocalCommitBase` is held.
358    pub async fn local_commit_base(
359        &self,
360        stream_id: AuthorStreamId,
361    ) -> Result<LocalCommitBase, DbError> {
362        let authorship = self.author_own_stream().await;
363        let (predecessor, frontier) = self
364            .call_store(move |session| session.local_commit_ledger_base(&stream_id))
365            .await?;
366        Ok(LocalCommitBase {
367            authorship,
368            predecessor,
369            frontier,
370        })
371    }
372
373    pub async fn latest_local_store_position(
374        &self,
375        stream_id: AuthorStreamId,
376    ) -> Result<Option<StoreBatchCommitRef>, DbError> {
377        self.call_store(move |session| session.latest_local_store_position(&stream_id.to_string()))
378            .await
379    }
380}
coven_database/store/store_session/pending_publication.rs

coven_database/store/store_session/
pending_publication.rs
