coven_database/store/store_session/
owner_recovery_publication.rs

1use super::*;
2use crate::*;
3use coven_protocol::store_commit::{
4    StoreBatchCommit, StoreCommitCoord, StoreDeviceRegistration,
5    StoreDeviceRegistrationActivationRef, StoreDeviceRegistrationOrigin, VerifiedStoreBatchCommit,
6};
7use rusqlite::OptionalExtension;
8
9#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
10#[serde(deny_unknown_fields)]
11struct DurableOwnerRecoveryPublication {
12    commit: DurablePreparedProtocolObject,
13    history_evidence: coven_protocol::store_commit::RetainedMergeCommitEvidence,
14}
15
16pub(super) fn complete_owner_recovery_publication_on(
17    store: super::StoreTransaction<'_, '_>,
18    commit: &VerifiedStoreBatchCommit,
19    publication: &crate::AcceptedStoreCommitPublication,
20) -> Result<(), DbError> {
21    if complete_matching_owner_recovery_publication_on(store, commit, &publication.clone().into())?
22    {
23        return Ok(());
24    }
25    Err(DbError::Message(
26        "completed Owner recovery has no exact publication journal".into(),
27    ))
28}
29
30pub(super) fn complete_matching_owner_recovery_publication_on(
31    store: super::StoreTransaction<'_, '_>,
32    commit: &VerifiedStoreBatchCommit,
33    acceptance: &crate::AcceptedStoreCommitEvidence,
34) -> Result<bool, DbError> {
35    let transaction = store.transaction;
36    let stored: Option<(String, String)> = transaction
37        .query_row(
38            "SELECT registration_hash, publication
39             FROM local_owner_recovery_publication WHERE singleton = 1",
40            [],
41            |row| Ok((row.get(0)?, row.get(1)?)),
42        )
43        .optional()
44        .map_err(DbError::from)?;
45    let Some(stored) = stored else {
46        return Ok(false);
47    };
48    if stored.0 != commit.author_registration.registration_hash.to_string() {
49        return Ok(false);
50    }
51    let durable: DurableOwnerRecoveryPublication = serde_json::from_str(&stored.1)
52        .map_err(|error| DbError::context("parse completed Owner recovery publication", error))?;
53    let active = super::active_store_publication::load_active_store_publication_on(transaction)?
54        .ok_or_else(|| {
55            DbError::Message("completed Owner recovery has no active Store publication".into())
56        })?;
57    let attempt = active.attempt()?;
58    if durable.commit.semantic_bytes() != commit.value().to_bytes()
59        || durable.commit.prepared().reference() != &commit.reference().object
60        || acceptance.commit_ref() != commit.reference()
61        || active.owner() != &ActiveStorePublicationOwner::OwnerRecovery
62        || acceptance.exact_publication().is_some_and(|publication| {
63            attempt.entry != *publication.entry()
64                || attempt.entry_object != publication.reference().object
65        })
66    {
67        return Err(DbError::Message(
68            "completed Owner recovery differs from its exact publication journal".into(),
69        ));
70    }
71    MergeMaterializationTransaction::from_store(store).activate_store_operation_remote_objects(
72        commit.reference(),
73        &[coven_protocol::remote_object::remote_object_id(
74            &commit.reference().object,
75        )],
76    )?;
77    let deleted = transaction
78        .execute(
79            "DELETE FROM local_owner_recovery_publication
80             WHERE singleton = 1 AND registration_hash = ?1 AND publication = ?2",
81            (&stored.0, &stored.1),
82        )
83        .map_err(DbError::from)?;
84    if deleted != 1 {
85        return Err(DbError::Message(
86            "Owner recovery publication changed during completion".into(),
87        ));
88    }
89    super::active_store_publication::clear_active_store_commit_for_owner_on(
90        transaction,
91        &ActiveStorePublicationOwner::OwnerRecovery,
92        commit.reference(),
93    )?;
94    Ok(true)
95}
96
97impl DurableOwnerRecoveryPublication {
98    fn from_publication(
99        publication: OwnerRecoveryPublication,
100    ) -> Result<
101        (
102            Self,
103            coven_protocol::prepared_commit::PreparedStorePublication,
104        ),
105        DbError,
106    > {
107        if publication.commit.bytes != publication.commit.value.value().to_bytes() {
108            return Err(DbError::Message(
109                "Owner recovery publication carries noncanonical semantic bytes".into(),
110            ));
111        }
112        Ok((
113            Self {
114                commit: DurablePreparedProtocolObject::new(
115                    publication.commit.bytes,
116                    publication.commit.prepared,
117                ),
118                history_evidence: publication.history_evidence,
119            },
120            publication.publication,
121        ))
122    }
123}
124
125impl StoreSession<'_> {
126    fn verify_owner_recovery_publication(
127        &mut self,
128        durable: DurableOwnerRecoveryPublication,
129        publication: coven_protocol::prepared_commit::PreparedStorePublication,
130    ) -> Result<(OwnerRecoveryPublication, ObjectHash), DbError> {
131        let local = self.local_store_device_registration()?.ok_or_else(|| {
132            DbError::Message("Owner recovery registration journal is absent".into())
133        })?;
134        if !matches!(
135            local.state,
136            LocalDeviceRegistrationState::Created | LocalDeviceRegistrationState::Activated { .. }
137        ) {
138            return Err(DbError::Message(
139                "Owner recovery publication requires created registration objects".into(),
140            ));
141        }
142        let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
143        let root = self
144            .verified_store_authority
145            .required_root_authority_on(records)?;
146        let registration =
147            StoreDeviceRegistration::parse_at(&local.registration_bytes, &root, local.device_id)
148                .map_err(|error| DbError::context("Owner recovery local registration", error))?;
149        let registration_ref =
150            coven_protocol::store_commit::StoreDeviceRegistrationRef::from_registration(
151                &registration,
152                local.prepared.reference().clone(),
153            );
154        if registration_ref.registration_hash != local.registration_hash {
155            return Err(DbError::Message(
156                "Owner recovery local registration hash differs from its exact reference".into(),
157            ));
158        }
159        let StoreDeviceRegistrationOrigin::Recovery {
160            recovery_id,
161            recovery_slot,
162            owner_grant,
163        } = &registration.origin
164        else {
165            return Err(DbError::Message(
166                "Owner recovery publication has a non-recovery registration".into(),
167            ));
168        };
169
170        durable
171            .commit
172            .prepared()
173            .reference()
174            .verify(durable.commit.prepared().stored_bytes())
175            .map_err(|error| DbError::context("Owner recovery exact commit", error))?;
176        let decoded: StoreBatchCommit = serde_json::from_slice(durable.commit.semantic_bytes())
177            .map_err(|error| DbError::context("Owner recovery commit", error))?;
178        let stream_id = coven_protocol::store_commit::StreamActivation::device_authorized_stream_id(
179            root.store_root_hash,
180            &registration_ref,
181            coven_protocol::store_commit::StreamAnchorDomain::StoreAnnouncements,
182        );
183        let coord = StoreCommitCoord {
184            stream_id,
185            sequence: decoded.seq(),
186        };
187        let commit = VerifiedStoreBatchCommit::parse_prepared(
188            durable.commit.semantic_bytes(),
189            root.store_root_hash,
190            coord,
191            durable.commit.prepared().reference().clone(),
192            &registration,
193        )
194        .map_err(|error| DbError::context("verify Owner recovery commit", error))?;
195        let [activation] = commit.device_registrations() else {
196            return Err(DbError::Message(
197                "Owner recovery commit must carry exactly one registration activation".into(),
198            ));
199        };
200        let StoreDeviceRegistrationActivationRef::Recovery {
201            recovery_id: activation_recovery_id,
202            node,
203        } = &activation.authority
204        else {
205            return Err(DbError::Message(
206                "Owner recovery commit carries another registration authority".into(),
207            ));
208        };
209        if commit.seq() != 1
210            || commit.author_registration != registration_ref
211            || activation.registration != registration_ref
212            || activation_recovery_id != recovery_id
213            || node.object.slot() != recovery_slot
214            || &node.owner_grant != owner_grant
215            || commit.value().to_bytes() != durable.commit.semantic_bytes()
216        {
217            return Err(DbError::Message(
218                "Owner recovery commit differs from its local recovery authority".into(),
219            ));
220        }
221        let proof = durable
222            .history_evidence
223            .membership_proof
224            .as_ref()
225            .ok_or_else(|| {
226                DbError::Message("Owner recovery has no registration authority head".into())
227            })?;
228        if !matches!(&proof.entry_value.change, coven_protocol::membership::StoreAuthorityChange::DeviceRegistrationActivation { registration } if registration == activation)
229        {
230            return Err(DbError::Message(
231                "Owner recovery authority entry names another activation".into(),
232            ));
233        }
234        durable
235            .history_evidence
236            .validate_for(commit.reference(), commit.value())
237            .map_err(|error| DbError::context("Owner recovery history evidence", error))?;
238
239        publication
240            .verify_commit(&commit)
241            .map_err(|error| DbError::context("Owner recovery publication", error))?;
242        Ok((
243            OwnerRecoveryPublication {
244                commit: ExactProtocolObject {
245                    value: commit,
246                    bytes: durable.commit.semantic_bytes,
247                    prepared: durable.commit.prepared,
248                },
249                publication,
250                history_evidence: durable.history_evidence,
251            },
252            local.registration_hash,
253        ))
254    }
255
256    fn stage_owner_recovery_publication(
257        &mut self,
258        publication: OwnerRecoveryPublication,
259    ) -> Result<OwnerRecoveryPublication, DbError> {
260        let (durable, publication) =
261            DurableOwnerRecoveryPublication::from_publication(publication)?;
262        let (verified, registration_hash) =
263            self.verify_owner_recovery_publication(durable.clone(), publication)?;
264        let registration_hash = registration_hash.to_string();
265        let encoded = serde_json::to_string(&durable)
266            .map_err(|error| DbError::context("serialize Owner recovery publication", error))?;
267        let active = ActiveStorePublication::commit(
268            ActiveStorePublicationOwner::OwnerRecovery,
269            verified.commit.value.write_id.clone(),
270            verified.commit.value.author_registration.clone(),
271            verified.commit.value.reference().coord.clone(),
272            verified.publication.clone(),
273        )?;
274        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
275        let current = super::observed_store_publication::load_store_current_publication_on(&tx)?;
276        if current.record() != &verified.publication.previous
277            || current.observed_version() != Some(&verified.publication.previous_version)
278        {
279            return Err(DbError::Message(
280                "Owner recovery publication extends another accepted boundary".into(),
281            ));
282        }
283        match super::active_store_publication::claim_active_store_publication_on(&tx, &active)? {
284            super::active_store_publication::ActiveStorePublicationClaim::Acquired => {}
285            super::active_store_publication::ActiveStorePublicationClaim::AlreadyOwned => {
286                return Err(DbError::Message(
287                    "Owner recovery owns publication before its journal".into(),
288                ));
289            }
290            super::active_store_publication::ActiveStorePublicationClaim::Occupied(owner) => {
291                return Err(DbError::Message(format!(
292                    "another local Store operation owns publication: {owner:?}"
293                )));
294            }
295        }
296        for remote in verified.remote_objects()? {
297            persist_exact_remote_object_on(
298                &tx,
299                self.store_dir,
300                &remote,
301                "Owner recovery candidate authority",
302            )?;
303        }
304        crate::store::store_session::StoreRecords::new(&tx, self.store_dir)
305            .stage_owner_recovery_publication(&registration_hash, &encoded)?;
306        tx.commit().map_err(DbError::from)?;
307        Ok(verified)
308    }
309
310    fn owner_recovery_publication(&mut self) -> Result<Option<OwnerRecoveryPublication>, DbError> {
311        let stored = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
312            .owner_recovery_publication_row()?;
313        stored
314            .map(|(registration_hash, encoded)| {
315                let durable = serde_json::from_str(&encoded)
316                    .map_err(|error| DbError::context("parse Owner recovery publication", error))?;
317                let active =
318                    super::active_store_publication::load_active_store_publication_on(self.conn)?
319                        .ok_or_else(|| {
320                        DbError::Message(
321                            "Owner recovery journal has no active Store publication".into(),
322                        )
323                    })?;
324                if active.owner() != &ActiveStorePublicationOwner::OwnerRecovery {
325                    return Err(DbError::Message(
326                        "Owner recovery journal differs from the active publication owner".into(),
327                    ));
328                }
329                let (publication, local_registration_hash) =
330                    self.verify_owner_recovery_publication(durable, active.attempt()?.clone())?;
331                if registration_hash != local_registration_hash.to_string() {
332                    return Err(DbError::Message(
333                        "Owner recovery publication belongs to another local registration".into(),
334                    ));
335                }
336                Ok(publication)
337            })
338            .transpose()
339    }
340}
341
342impl StoreDatabase {
343    pub async fn stage_owner_recovery_publication(
344        &self,
345        publication: OwnerRecoveryPublication,
346    ) -> Result<OwnerRecoveryPublication, DbError> {
347        self.call_store(move |session| session.stage_owner_recovery_publication(publication))
348            .await
349    }
350
351    pub async fn owner_recovery_publication(
352        &self,
353    ) -> Result<Option<OwnerRecoveryPublication>, DbError> {
354        self.call_store(|session| session.owner_recovery_publication())
355            .await
356    }
357}
coven_database/store/store_session/owner_recovery_publication.rs

coven_database/store/store_session/
owner_recovery_publication.rs
