coven_database/store/store_session/
materialization.rs

1use std::collections::BTreeSet;
2
3use super::{
4    MergeMaterializationTransaction, StoreDatabase, StoreSession, StoreTransactionOutcome,
5    VerifiedStoreTransaction,
6};
7use crate::{install_store_founder_state_on, DbError, VerifiedMergeMaterialization};
8use coven_protocol::circle_activation::VerifiedCircleActivations;
9use coven_protocol::objects::ExactObjectRef;
10use coven_protocol::store_commit::{
11    ActivatedStoreDeviceRegistration, StoreDeviceHead, VerifiedStoreBatchCommit,
12    VerifiedStoreDeviceOperations,
13};
14
15#[cfg(any(test, feature = "test-utils"))]
16fn reach_materialization_failure(
17    armed: &std::sync::Mutex<Option<crate::MergeMaterializationFailurePoint>>,
18    point: crate::MergeMaterializationFailurePoint,
19) -> Result<bool, DbError> {
20    let mut armed = armed
21        .lock()
22        .map_err(|_| DbError::Message("Merge materialization failure lock poisoned".to_string()))?;
23    if armed.as_ref() != Some(&point) {
24        return Ok(false);
25    }
26    armed.take();
27    Ok(true)
28}
29
30/// The bootstrap commits this database already materializes, which is to say
31/// the ones a previous run of this same installation landed before it stopped.
32///
33/// A plan carries only the history past the installed snapshot's coverage: the
34/// device that built it knew which snapshot this device installs and walked
35/// forward from that snapshot's tips. So a plan commit at or under a coverage
36/// tip is never the ordinary case — it is a plan built against a different
37/// history, a fork at that coordinate, or a snapshot that ran past the
38/// bootstrap cut. None can be installed over, so the join fails here instead of
39/// writing rows against an image that disagrees with them.
40fn device_join_bootstrap_represented_on(
41    tx: &rusqlite::Transaction<'_>,
42    commits: &[crate::DeviceJoinBootstrapCommit],
43) -> Result<BTreeSet<coven_protocol::store_commit::StoreBatchCommitRef>, DbError> {
44    let mut represented = BTreeSet::new();
45    let coverage = crate::store::materialized_commit_index::snapshot_coverage_on(tx)?;
46    for prepared in commits {
47        let stream_id = prepared.reference.coord.stream_id.to_string();
48        let sequence = prepared.reference.coord.sequence();
49        if let Some(existing) = crate::store::materialized_commit_index::materialized_commit_ref_on(
50            tx, &stream_id, sequence,
51        )? {
52            if existing != prepared.reference {
53                return Err(DbError::Message(format!(
54                    "device join bootstrap conflicts at {stream_id}/{sequence}"
55                )));
56            }
57            represented.insert(prepared.reference.clone());
58            continue;
59        }
60        if coverage
61            .get(&stream_id)
62            .is_some_and(|tip| sequence <= tip.coord.sequence())
63        {
64            return Err(DbError::Message(format!(
65                "device join bootstrap history at {stream_id}/{sequence} is not the history the \
66                 installed snapshot covers"
67            )));
68        }
69    }
70    Ok(represented)
71}
72
73impl VerifiedStoreTransaction<'_, '_, '_> {
74    fn apply_received_merge_materialization(
75        &mut self,
76        materialization: crate::PreparedMergeMaterialization,
77        retractions: Vec<coven_protocol::remote_object::VerifiedCandidateNonactivation>,
78        local_store_membership: coven_protocol::membership::LocalStoreMembership,
79        routing_key: Option<coven_protocol::circle::RowRoutingKey>,
80        receiver_wall_ms: u64,
81    ) -> Result<super::merge_materialization_transaction::AppliedMergeMaterialization, DbError>
82    {
83        #[cfg(any(test, feature = "test-utils"))]
84        let materialization_failure = self.merge_materialization_failure;
85        let authority = &mut *self.authority;
86        let blob_decls = self.blob_decls;
87        let gates = self.gates;
88        let synced_tables = self.synced_tables;
89        let root = materialization.root.clone();
90        let candidate = materialization.verified_commit.reference().clone();
91        let local_exclusions = materialization
92            .circle_activations
93            .local_exclusions()
94            .to_vec();
95        if !materialization.packages.is_empty()
96            && materialization.package_application
97                != Some(crate::RetainedPackageApplication::Received { receiver_wall_ms })
98        {
99            return Err(DbError::Message(
100                "received Merge packages carry another application timestamp".to_string(),
101            ));
102        }
103        let tx = self.store.transaction;
104        let merge_transaction = MergeMaterializationTransaction::from_store(self.store);
105        merge_transaction.record_prepared_materialization_authority(&materialization)?;
106        let retained =
107            merge_transaction.retain_prepared_merge_materialization(authority, &materialization)?;
108        authority.insert_verified(retained)?;
109        #[cfg(any(test, feature = "test-utils"))]
110        if reach_materialization_failure(
111            materialization_failure,
112            crate::MergeMaterializationFailurePoint::SummaryMaterialization,
113        )? {
114            return Err(DbError::Message(
115                "injected failure after Merge summary materialization".to_string(),
116            ));
117        }
118        for exclusion in &local_exclusions {
119            super::circle_operations::record_circle_close_exclusion_on(tx, exclusion)?;
120        }
121        let retracted = retractions
122            .iter()
123            .map(|retraction| retraction.candidate_reference().map_err(DbError::from))
124            .collect::<Result<BTreeSet<_>, _>>()?;
125        let mut write_status_notifications = Vec::new();
126        if !retractions.is_empty() {
127            write_status_notifications =
128                super::merge_materialization_transaction::retract_verified_merge_materializations(
129                    &merge_transaction,
130                    &root,
131                    authority,
132                    retractions,
133                )?;
134            #[cfg(any(test, feature = "test-utils"))]
135            if reach_materialization_failure(
136                materialization_failure,
137                crate::MergeMaterializationFailurePoint::RetractionDeletion,
138            )? {
139                return Err(DbError::Message(
140                    "injected failure after Merge retraction deletion".to_string(),
141                ));
142            }
143        }
144        let replayed = authority.replay_projection_watching_on(
145            crate::store::store_session::StoreTransaction::new(tx, self.store.store_dir),
146            blob_decls,
147            gates,
148            synced_tables,
149            routing_key.as_ref(),
150            &retracted,
151            crate::ReplayJournal::Owed,
152            local_store_membership,
153            &candidate,
154        )?;
155        let watched = replayed.watched_outcome().ok_or_else(|| {
156            DbError::Message("incoming Merge materialization was not replayed".to_string())
157        })?;
158        let max_updated_at = match watched {
159            super::WatchedReplayOutcome::Applied { max_updated_at } => max_updated_at,
160            super::WatchedReplayOutcome::Held(reason) => {
161                return Ok(
162                    super::merge_materialization_transaction::AppliedMergeMaterialization {
163                        outcome: crate::MaterializationOutcome::Held(reason),
164                        max_updated_at: None,
165                        write_status_notifications: Vec::new(),
166                    },
167                );
168            }
169        };
170        let rows = replayed.install_on(self, &root)?;
171        Ok(
172            super::merge_materialization_transaction::AppliedMergeMaterialization {
173                outcome: crate::MaterializationOutcome::Applied(rows),
174                max_updated_at,
175                write_status_notifications,
176            },
177        )
178    }
179
180    pub(super) fn install_replay_projection(
181        &self,
182        root: &coven_protocol::store_commit::StoreRootRef,
183        replay: &super::ReplayProjection,
184    ) -> Result<Vec<coven_foundation::changeset::RowChange>, DbError> {
185        let tx = self.store.transaction;
186        let mut host_changes = rusqlite::session::Session::new(tx).map_err(DbError::from)?;
187        for table in self.synced_tables {
188            host_changes
189                .attach(Some(table.name()))
190                .map_err(DbError::from)?;
191        }
192        let mut tables = crate::projection_table_names(self.gates.has_scoped_graph());
193        tables.extend(
194            self.synced_tables
195                .iter()
196                .map(|table| table.name().to_string()),
197        );
198        tables.sort();
199        tables.dedup();
200        let projected_blobs = replay
201            .publication_blobs(self.blob_decls)?
202            .into_iter()
203            .map(|publication| publication.blob)
204            .collect::<Vec<_>>();
205        let suspended_cleanup =
206            super::local_blob_cleanup::suspend_leased_blob_cleanup_for_restoration_on(
207                tx,
208                &projected_blobs,
209            )?;
210        let old_exact_bindings = super::local_blob_cleanup::exact_blob_bindings_on(tx)?;
211        tx.pragma_update(None, "defer_foreign_keys", "ON")
212            .map_err(DbError::from)?;
213        crate::store::store_session::StoreTransaction::new(tx, self.store.store_dir)
214            .replace_tables_from_projection(replay, &tables)?;
215        let violations: bool = tx
216            .query_row(
217                "SELECT EXISTS(SELECT 1 FROM pragma_foreign_key_check)",
218                [],
219                |row| row.get(0),
220            )
221            .map_err(DbError::from)?;
222        if violations {
223            let violation: (String, Option<i64>, String, i64) = tx
224                .query_row(
225                    "SELECT * FROM pragma_foreign_key_check LIMIT 1",
226                    [],
227                    |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
228                )
229                .map_err(DbError::from)?;
230            return Err(DbError::Message(format!(
231                "retained replay projection violates foreign keys: {violation:?}"
232            )));
233        }
234        let mut projection_changeset = Vec::new();
235        host_changes
236            .changeset_strm(&mut projection_changeset)
237            .map_err(DbError::from)?;
238        #[cfg(any(test, feature = "test-utils"))]
239        if reach_materialization_failure(
240            self.merge_materialization_failure,
241            crate::MergeMaterializationFailurePoint::ProjectionReplacement,
242        )? {
243            return Err(DbError::Message(
244                "injected failure after Merge projection replacement".to_string(),
245            ));
246        }
247        MergeMaterializationTransaction::from_store(self.store)
248            .replace_store_device_exclusion_freezes_from_replay(root)?;
249        let old_projection =
250            crate::walk_old_changeset(&projection_changeset).map_err(DbError::Changeset)?;
251        let new_projection =
252            crate::walk_changeset(&projection_changeset).map_err(DbError::Changeset)?;
253        for intent in crate::local_blob_cleanup_intents::intents_from_changes(
254            self.blob_decls,
255            &old_projection,
256            &new_projection,
257        )
258        .map_err(DbError::from)?
259        {
260            super::local_blob_cleanup::record_obsolete_copy_intents_from_bindings_on(
261                tx,
262                self.blob_decls,
263                &intent,
264                &old_exact_bindings,
265            )?;
266        }
267        super::local_blob_cleanup::reevaluate_suspended_blob_cleanup_on(
268            tx,
269            self.blob_decls,
270            &suspended_cleanup,
271        )?;
272        crate::Database::cancel_transitions_for_deleted_roots_on(
273            tx,
274            &super::merge_materialization_transaction::deleted_rows(&new_projection),
275        )?;
276        Ok(new_projection)
277    }
278
279    #[allow(clippy::too_many_arguments)]
280    fn materialize_published_store_operation(
281        &mut self,
282        root: coven_protocol::store_commit::StoreRootRef,
283        verified_commit: VerifiedStoreBatchCommit,
284        registrations: Vec<ActivatedStoreDeviceRegistration>,
285        device_operations: VerifiedStoreDeviceOperations,
286        circle_activations: VerifiedCircleActivations,
287        activation_head: StoreDeviceHead,
288        activation_head_object: ExactObjectRef,
289        history_evidence: coven_protocol::store_commit::RetainedMergeCommitEvidence,
290        membership_objects: Option<crate::VerifiedMergeMembershipObjects>,
291        operation_object_ids: Option<Vec<coven_protocol::store_commit::ObjectHash>>,
292        membership_completion: Option<
293            coven_protocol::membership_mutation::StoreMembershipJournalCompletion,
294        >,
295    ) -> Result<(), DbError> {
296        let reference = verified_commit.reference().clone();
297        let tx = self.store.transaction;
298        let authority = &mut *self.authority;
299        let store_transaction = MergeMaterializationTransaction::from_store(self.store);
300        if let Some(object_ids) = operation_object_ids {
301            store_transaction.activate_store_operation_remote_objects(&reference, &object_ids)?;
302        }
303        if !registrations.is_empty() {
304            super::record_activated_store_device_registrations_on(
305                tx,
306                verified_commit.value(),
307                &registrations,
308            )?;
309        }
310        let materialization = VerifiedMergeMaterialization::verify(
311            &root,
312            &verified_commit,
313            &registrations,
314            &device_operations,
315            &circle_activations,
316            &activation_head,
317            &activation_head_object,
318            &history_evidence,
319            membership_objects.as_ref(),
320            &[],
321            None,
322        )?;
323        if let Some(completion) = membership_completion {
324            store_transaction
325                .complete_membership_journal(completion, &reference)
326                .map_err(|error| DbError::context("complete exact membership journal", error))?;
327        }
328        let retained = store_transaction
329            .record_verified_merge_materialization(authority, materialization)
330            .map_err(|error| DbError::context("record exact Merge materialization", error))?;
331        authority.insert_verified(retained)?;
332        Ok(())
333    }
334
335    fn materialize_device_join_activation(
336        &mut self,
337        root: coven_protocol::store_commit::StoreRootRef,
338        verified_commit: VerifiedStoreBatchCommit,
339        registrations: Vec<ActivatedStoreDeviceRegistration>,
340        device_operations: VerifiedStoreDeviceOperations,
341        activation_head: StoreDeviceHead,
342        activation_head_object: ExactObjectRef,
343        history_evidence: coven_protocol::store_commit::RetainedMergeCommitEvidence,
344    ) -> Result<(), DbError> {
345        let expected_ref = verified_commit.reference().clone();
346        let stream_id = expected_ref.coord.stream_id.to_string();
347        let sequence = expected_ref.coord.sequence();
348        let tx = self.store.transaction;
349        if let Some(materialized) =
350            crate::store::materialized_commit_index::materialized_commit_ref_on(
351                tx, &stream_id, sequence,
352            )?
353        {
354            if materialized != expected_ref {
355                return Err(DbError::Message(format!(
356                    "device join activation coordinate {stream_id}/{sequence} is already occupied by another commit"
357                )));
358            }
359            super::record_activated_store_device_registrations_on(
360                tx,
361                verified_commit.value(),
362                &registrations,
363            )?;
364            return Ok(());
365        }
366        let authority = &mut *self.authority;
367        super::record_activated_store_device_registrations_on(
368            tx,
369            verified_commit.value(),
370            &registrations,
371        )?;
372        let circle_activations =
373            VerifiedCircleActivations::none(verified_commit.value(), verified_commit.reference())
374                .map_err(DbError::from)?;
375        let materialization = VerifiedMergeMaterialization::verify(
376            &root,
377            &verified_commit,
378            &registrations,
379            &device_operations,
380            &circle_activations,
381            &activation_head,
382            &activation_head_object,
383            &history_evidence,
384            None,
385            &[],
386            None,
387        )?;
388        let retained = MergeMaterializationTransaction::from_store(self.store)
389            .record_verified_merge_materialization(authority, materialization)?;
390        authority.insert_verified(retained)?;
391        Ok(())
392    }
393
394    fn install_device_join_bootstrap(
395        &mut self,
396        root: coven_protocol::store_commit::StoreRootRef,
397        resolved: crate::ResolvedDeviceJoinBootstrap,
398    ) -> Result<Option<coven_protocol::hlc::Timestamp>, DbError> {
399        let crate::ResolvedDeviceJoinBootstrap {
400            plan,
401            mut row_data,
402            local_store_membership,
403            routing_key,
404            receiver_wall_ms,
405        } = resolved;
406        let tx = self.store.transaction;
407        let blob_decls = self.blob_decls;
408        let gates = self.gates;
409        let synced_tables = self.synced_tables;
410        let authority = &mut *self.authority;
411        let installed_root = authority.root().clone();
412        if installed_root != root || plan.founder.store_root != root {
413            return Err(DbError::Message(
414                "device join bootstrap root differs from the installed exact root".to_string(),
415            ));
416        }
417        install_store_founder_state_on(
418            tx,
419            &root,
420            &plan.founder_reference,
421            &plan.founder,
422            &plan.founder_bytes,
423            &plan.genesis,
424        )?;
425        crate::set_protocol_state_on(
426            tx,
427            coven_protocol::membership::OWNER_PUBKEY_STATE_KEY,
428            &plan.founder.author_pubkey,
429        )?;
430        plan.membership.install_on(tx)?;
431
432        let represented = device_join_bootstrap_represented_on(tx, &plan.commits)?;
433
434        // Row data has to be present before anything advances over it. A commit
435        // that names a Store package but resolved none would otherwise leave the
436        // joining device with an advanced position and no rows.
437        for prepared in &plan.commits {
438            if represented.contains(&prepared.reference) {
439                continue;
440            }
441            let commit = prepared.commit.value();
442            let resolved = row_data.get(&prepared.reference);
443            let carries_store_package = resolved.is_some_and(|data| {
444                data.packages.iter().any(|prepared| {
445                    matches!(
446                        prepared.package.audience(),
447                        coven_protocol::audience_package::PackageAudience::Store
448                    )
449                })
450            });
451            if resolved.is_none() || (commit.store_package().is_some() && !carries_store_package) {
452                return Err(DbError::Message(format!(
453                    "device join bootstrap cannot advance over unmaterialized row data at {}/{}",
454                    prepared.reference.coord.stream_id,
455                    prepared.reference.coord.sequence()
456                )));
457            }
458        }
459
460        let mut retained_any = false;
461        for prepared in plan.commits {
462            if represented.contains(&prepared.reference) {
463                continue;
464            }
465            let stream_id = prepared.reference.coord.stream_id.to_string();
466            if let Some(existing) =
467                crate::store::materialized_commit_index::materialized_commit_ref_on(
468                    tx,
469                    &stream_id,
470                    prepared.reference.coord.sequence(),
471                )?
472            {
473                if existing != prepared.reference {
474                    return Err(DbError::Message(format!(
475                        "device join bootstrap conflicts at {stream_id}/{}",
476                        prepared.reference.coord.sequence()
477                    )));
478                }
479                continue;
480            }
481            let data = row_data.remove(&prepared.reference).ok_or_else(|| {
482                DbError::Message(format!(
483                    "device join bootstrap has no resolved row data at {stream_id}/{}",
484                    prepared.reference.coord.sequence()
485                ))
486            })?;
487            let activation = prepared.activation;
488            let materialization = crate::PreparedMergeMaterialization {
489                root: root.clone(),
490                verified_commit: prepared.commit,
491                activation_head: activation.head,
492                activation_head_object: activation.object,
493                history_evidence: activation.history_evidence,
494                membership_objects: data.membership_objects,
495                membership_remote_objects: data.membership_remote_objects,
496                registrations: prepared.registrations,
497                package_application: (!data.packages.is_empty())
498                    .then_some(crate::RetainedPackageApplication::Received { receiver_wall_ms }),
499                packages: data.packages,
500                device_operations: prepared.device_operations,
501                circle_activations: data.circle_activations,
502            };
503            let merge_transaction = MergeMaterializationTransaction::from_store(self.store);
504            merge_transaction.record_prepared_materialization_authority(&materialization)?;
505            let retained = merge_transaction
506                .retain_prepared_merge_materialization(authority, &materialization)?;
507            authority.insert_verified(retained)?;
508            retained_any = true;
509        }
510        if !row_data.is_empty() {
511            return Err(DbError::Message(
512                "device join bootstrap resolved row data outside its exact history".to_string(),
513            ));
514        }
515        if !retained_any {
516            return Ok(None);
517        }
518        let replayed = authority.replay_projection_result_on(
519            crate::store::store_session::StoreTransaction::new(tx, self.store.store_dir),
520            blob_decls,
521            gates,
522            synced_tables,
523            routing_key.as_ref(),
524            crate::ReplayJournal::Owed,
525            local_store_membership,
526        )?;
527        replayed.install_on(self, &root)?;
528        Ok(replayed.max_updated_at())
529    }
530
531    fn complete_owner_recovery(
532        &mut self,
533        verified_commit: VerifiedStoreBatchCommit,
534        activation_head: StoreDeviceHead,
535        activation_head_object: ExactObjectRef,
536        history_evidence: coven_protocol::store_commit::RetainedMergeCommitEvidence,
537        registration: ActivatedStoreDeviceRegistration,
538    ) -> Result<(), DbError> {
539        let tx = self.store.transaction;
540        let authority = &mut *self.authority;
541        let root = authority.root().clone();
542        let registrations = vec![registration];
543        let commit = verified_commit.value();
544        super::record_activated_store_device_registrations_on(tx, commit, &registrations)?;
545        let retained = MergeMaterializationTransaction::from_store(self.store)
546            .record_materialized_merge_commit(
547                authority,
548                &root,
549                &verified_commit,
550                &registrations,
551                &activation_head,
552                &activation_head_object,
553                &history_evidence,
554                &[],
555                None,
556            )?;
557        authority.insert_verified(retained)?;
558        super::owner_recovery_publication::complete_owner_recovery_publication_on(
559            tx,
560            &verified_commit,
561            &activation_head,
562            &activation_head_object,
563        )?;
564        #[cfg(any(test, feature = "test-utils"))]
565        if reach_materialization_failure(
566            self.merge_materialization_failure,
567            crate::MergeMaterializationFailurePoint::SummaryMaterialization,
568        )? {
569            return Err(DbError::Message(
570                "injected failure after Merge summary materialization".to_string(),
571            ));
572        }
573        Ok(())
574    }
575}
576
577impl StoreSession<'_> {
578    fn apply_received_merge_materialization(
579        &mut self,
580        materialization: crate::PreparedMergeMaterialization,
581        retractions: Vec<coven_protocol::remote_object::VerifiedCandidateNonactivation>,
582        local_store_membership: coven_protocol::membership::LocalStoreMembership,
583        routing_key: Option<coven_protocol::circle::RowRoutingKey>,
584        receiver_wall_ms: u64,
585    ) -> Result<super::merge_materialization_transaction::AppliedMergeMaterialization, DbError>
586    {
587        let applied = self.verified_store_transaction(move |transaction| {
588            let applied = transaction.apply_received_merge_materialization(
589                materialization,
590                retractions,
591                local_store_membership,
592                routing_key,
593                receiver_wall_ms,
594            )?;
595            if matches!(applied.outcome, crate::MaterializationOutcome::Applied(_)) {
596                Ok(StoreTransactionOutcome::Commit(applied))
597            } else {
598                Ok(StoreTransactionOutcome::Rollback(applied))
599            }
600        })?;
601        if let Some(max_applied) = applied.max_updated_at.as_ref() {
602            self.hlc.advance_past(max_applied);
603        }
604        Ok(applied)
605    }
606
607    #[allow(clippy::too_many_arguments)]
608    fn materialize_published_store_operation(
609        &mut self,
610        root: coven_protocol::store_commit::StoreRootRef,
611        verified_commit: VerifiedStoreBatchCommit,
612        registrations: Vec<ActivatedStoreDeviceRegistration>,
613        device_operations: VerifiedStoreDeviceOperations,
614        circle_activations: VerifiedCircleActivations,
615        activation_head: StoreDeviceHead,
616        activation_head_object: ExactObjectRef,
617        history_evidence: coven_protocol::store_commit::RetainedMergeCommitEvidence,
618        membership_objects: Option<crate::VerifiedMergeMembershipObjects>,
619        operation_object_ids: Option<Vec<coven_protocol::store_commit::ObjectHash>>,
620        membership_completion: Option<
621            coven_protocol::membership_mutation::StoreMembershipJournalCompletion,
622        >,
623    ) -> Result<(), DbError> {
624        self.verified_store_transaction(move |transaction| {
625            transaction.materialize_published_store_operation(
626                root,
627                verified_commit,
628                registrations,
629                device_operations,
630                circle_activations,
631                activation_head,
632                activation_head_object,
633                history_evidence,
634                membership_objects,
635                operation_object_ids,
636                membership_completion,
637            )?;
638            Ok(StoreTransactionOutcome::Commit(()))
639        })
640    }
641
642    fn materialize_device_join_activation(
643        &mut self,
644        root: coven_protocol::store_commit::StoreRootRef,
645        verified_commit: VerifiedStoreBatchCommit,
646        registrations: Vec<ActivatedStoreDeviceRegistration>,
647        device_operations: VerifiedStoreDeviceOperations,
648        activation_head: StoreDeviceHead,
649        activation_head_object: ExactObjectRef,
650        history_evidence: coven_protocol::store_commit::RetainedMergeCommitEvidence,
651    ) -> Result<(), DbError> {
652        self.verified_store_transaction(move |transaction| {
653            transaction.materialize_device_join_activation(
654                root,
655                verified_commit,
656                registrations,
657                device_operations,
658                activation_head,
659                activation_head_object,
660                history_evidence,
661            )?;
662            Ok(StoreTransactionOutcome::Commit(()))
663        })
664    }
665
666    fn unrepresented_device_join_bootstrap_commits(
667        &mut self,
668        plan: crate::DeviceJoinBootstrapPlan,
669    ) -> Result<
670        (
671            crate::DeviceJoinBootstrapPlan,
672            Vec<coven_protocol::store_commit::StoreBatchCommitRef>,
673        ),
674        DbError,
675    > {
676        self.verified_store_transaction(move |transaction| {
677            let represented =
678                device_join_bootstrap_represented_on(transaction.store.transaction, &plan.commits)?;
679            let unrepresented = plan
680                .commits
681                .iter()
682                .map(|prepared| prepared.reference.clone())
683                .filter(|reference| !represented.contains(reference))
684                .collect::<Vec<_>>();
685            Ok(StoreTransactionOutcome::Rollback((plan, unrepresented)))
686        })
687    }
688
689    fn install_device_join_bootstrap(
690        &mut self,
691        root: coven_protocol::store_commit::StoreRootRef,
692        resolved: crate::ResolvedDeviceJoinBootstrap,
693    ) -> Result<(), DbError> {
694        let max_updated_at = self.verified_store_transaction(move |transaction| {
695            let max_updated_at = transaction.install_device_join_bootstrap(root, resolved)?;
696            Ok(StoreTransactionOutcome::Commit(max_updated_at))
697        })?;
698        if let Some(max_applied) = max_updated_at.as_ref() {
699            self.hlc.advance_past(max_applied);
700        }
701        Ok(())
702    }
703
704    fn complete_owner_recovery(
705        &mut self,
706        verified_commit: VerifiedStoreBatchCommit,
707        activation_head: StoreDeviceHead,
708        activation_head_object: ExactObjectRef,
709        history_evidence: coven_protocol::store_commit::RetainedMergeCommitEvidence,
710        registration: ActivatedStoreDeviceRegistration,
711    ) -> Result<(), DbError> {
712        self.verified_store_transaction(move |transaction| {
713            transaction.complete_owner_recovery(
714                verified_commit,
715                activation_head,
716                activation_head_object,
717                history_evidence,
718                registration,
719            )?;
720            Ok(StoreTransactionOutcome::Commit(()))
721        })
722    }
723}
724
725impl StoreDatabase {
726    pub async fn apply_received_merge_materialization(
727        &self,
728        materialization: crate::PreparedMergeMaterialization,
729        retractions: Vec<coven_protocol::remote_object::VerifiedCandidateNonactivation>,
730        local_store_membership: coven_protocol::membership::LocalStoreMembership,
731        routing_key: Option<coven_protocol::circle::RowRoutingKey>,
732        receiver_wall_ms: u64,
733    ) -> Result<crate::MaterializationOutcome, DbError> {
734        let applied = self
735            .call_store(move |session| {
736                session.apply_received_merge_materialization(
737                    materialization,
738                    retractions,
739                    local_store_membership,
740                    routing_key,
741                    receiver_wall_ms,
742                )
743            })
744            .await?;
745        for (write_id, status) in applied.write_status_notifications {
746            self.notify_write_status(write_id, status);
747        }
748        Ok(applied.outcome)
749    }
750
751    #[allow(clippy::too_many_arguments)]
752    pub async fn materialize_published_store_operation(
753        &self,
754        root: coven_protocol::store_commit::StoreRootRef,
755        verified_commit: VerifiedStoreBatchCommit,
756        registrations: Vec<ActivatedStoreDeviceRegistration>,
757        device_operations: VerifiedStoreDeviceOperations,
758        circle_activations: VerifiedCircleActivations,
759        activation_head: StoreDeviceHead,
760        activation_head_object: ExactObjectRef,
761        history_evidence: coven_protocol::store_commit::RetainedMergeCommitEvidence,
762        membership_objects: Option<crate::VerifiedMergeMembershipObjects>,
763        operation_object_ids: Option<Vec<coven_protocol::store_commit::ObjectHash>>,
764        membership_completion: Option<
765            coven_protocol::membership_mutation::StoreMembershipJournalCompletion,
766        >,
767    ) -> Result<(), DbError> {
768        self.call_store(move |session| {
769            session.materialize_published_store_operation(
770                root,
771                verified_commit,
772                registrations,
773                device_operations,
774                circle_activations,
775                activation_head,
776                activation_head_object,
777                history_evidence,
778                membership_objects,
779                operation_object_ids,
780                membership_completion,
781            )
782        })
783        .await
784    }
785
786    pub async fn materialize_device_join_activation(
787        &self,
788        root: coven_protocol::store_commit::StoreRootRef,
789        verified_commit: VerifiedStoreBatchCommit,
790        registrations: Vec<ActivatedStoreDeviceRegistration>,
791        device_operations: VerifiedStoreDeviceOperations,
792        activation_head: StoreDeviceHead,
793        activation_head_object: ExactObjectRef,
794        history_evidence: coven_protocol::store_commit::RetainedMergeCommitEvidence,
795    ) -> Result<(), DbError> {
796        self.call_store(move |session| {
797            session.materialize_device_join_activation(
798                root,
799                verified_commit,
800                registrations,
801                device_operations,
802                activation_head,
803                activation_head_object,
804                history_evidence,
805            )
806        })
807        .await
808    }
809
810    /// The plan commits whose rows this database does not already materialize.
811    /// The joining device resolves row data for exactly these before installing.
812    pub async fn unrepresented_device_join_bootstrap_commits(
813        &self,
814        plan: crate::DeviceJoinBootstrapPlan,
815    ) -> Result<
816        (
817            crate::DeviceJoinBootstrapPlan,
818            Vec<coven_protocol::store_commit::StoreBatchCommitRef>,
819        ),
820        DbError,
821    > {
822        self.call_store(move |session| session.unrepresented_device_join_bootstrap_commits(plan))
823            .await
824    }
825
826    pub async fn install_device_join_bootstrap(
827        &self,
828        root: coven_protocol::store_commit::StoreRootRef,
829        resolved: crate::ResolvedDeviceJoinBootstrap,
830    ) -> Result<(), DbError> {
831        self.call_store(move |session| session.install_device_join_bootstrap(root, resolved))
832            .await
833    }
834
835    pub async fn complete_owner_recovery(
836        &self,
837        verified_commit: VerifiedStoreBatchCommit,
838        activation_head: StoreDeviceHead,
839        activation_head_object: ExactObjectRef,
840        history_evidence: coven_protocol::store_commit::RetainedMergeCommitEvidence,
841        registration: ActivatedStoreDeviceRegistration,
842    ) -> Result<(), DbError> {
843        self.call_store(move |session| {
844            session.complete_owner_recovery(
845                verified_commit,
846                activation_head,
847                activation_head_object,
848                history_evidence,
849                registration,
850            )
851        })
852        .await
853    }
854}
coven_database/store/store_session/materialization.rs

coven_database/store/store_session/
materialization.rs
