coven_database/store/store_session/
device_join_publication.rs

1use super::{StoreDatabase, StoreSession};
2use crate::{
3    persist_exact_remote_object_on, ActiveStorePublication, ActiveStorePublicationOwner, DbError,
4};
5use coven_protocol::store_commit::device_join_journal::{
6    DeviceJoinJournalRecord, DeviceJoinRole, DeviceJoinRoleProgress, OwnerJoinProgress,
7    PreparedOwnerJoinPublication,
8};
9
10impl StoreSession<'_> {
11    fn prepare_owner_device_join_publication(
12        &mut self,
13        previous: DeviceJoinJournalRecord,
14        prepared: PreparedOwnerJoinPublication,
15    ) -> Result<DeviceJoinJournalRecord, DbError> {
16        if previous.progress.role() != DeviceJoinRole::Owner {
17            return Err(DbError::Message(
18                "device join publication belongs to a non-Owner journal".to_string(),
19            ));
20        }
21        prepared
22            .validate_for(previous.attempt_id)
23            .map_err(|error| DbError::context("prepared owner device join publication", error))?;
24        let next = DeviceJoinJournalRecord {
25            attempt_id: previous.attempt_id,
26            progress: Box::new(DeviceJoinRoleProgress::Owner(
27                OwnerJoinProgress::StorePublicationPrepared(prepared.clone()),
28            )),
29        };
30        crate::store::device_join_journal::validate_successor(&previous, &next).map_err(
31            |error| DbError::Message(format!("prepare owner device join publication: {error}")),
32        )?;
33        let active = ActiveStorePublication::for_commit(
34            ActiveStorePublicationOwner::DeviceJoin(previous.attempt_id),
35            &prepared.candidate,
36        )?;
37        let transaction = self.conn.unchecked_transaction().map_err(DbError::from)?;
38        let key = previous.store_key();
39        let expected = serde_json::to_string(&previous)
40            .map_err(|error| DbError::context("serialize device join predecessor", error))?;
41        let actual = crate::required_protocol_state_on(&transaction, &key)?;
42        if actual != expected {
43            return Err(DbError::Message(
44                "device join journal changed before publication preparation".to_string(),
45            ));
46        }
47        match super::active_store_publication::claim_active_store_publication_on(
48            &transaction,
49            &active,
50        )? {
51            super::active_store_publication::ActiveStorePublicationClaim::Acquired => {}
52            super::active_store_publication::ActiveStorePublicationClaim::AlreadyOwned => {
53                return Err(DbError::Message(
54                    "device join publication was active before its journal was prepared"
55                        .to_string(),
56                ));
57            }
58            super::active_store_publication::ActiveStorePublicationClaim::Occupied(owner) => {
59                return Err(DbError::Message(format!(
60                    "another local Store operation owns publication: {owner:?}"
61                )));
62            }
63        }
64        for remote in prepared
65            .remote_objects(previous.attempt_id)
66            .map_err(|error| DbError::context("prepare device join remote graph", error))?
67        {
68            persist_exact_remote_object_on(
69                &transaction,
70                self.store_dir,
71                &remote,
72                "device join candidate object",
73            )?;
74        }
75        let next_value = serde_json::to_string(&next)
76            .map_err(|error| DbError::context("serialize prepared device join", error))?;
77        if transaction
78            .execute(
79                "UPDATE protocol_state SET value = ?1 WHERE key = ?2 AND value = ?3",
80                (&next_value, &key, &expected),
81            )
82            .map_err(DbError::from)?
83            != 1
84        {
85            return Err(DbError::Message(
86                "device join journal changed during publication preparation".to_string(),
87            ));
88        }
89        transaction.commit().map_err(DbError::from)?;
90        Ok(next)
91    }
92}
93
94pub(crate) fn complete_owner_device_join_publication_on(
95    transaction: &rusqlite::Transaction<'_>,
96    candidate: &coven_protocol::store_commit::StoreBatchCommitRef,
97    completion: Option<&coven_protocol::membership_mutation::StoreMembershipJournalCompletion>,
98) -> Result<(), DbError> {
99    let active = super::active_store_publication::load_active_store_publication_on(transaction)?
100        .ok_or_else(|| {
101            DbError::Message(format!(
102                "accepted device join candidate {candidate:?} has no active publication"
103            ))
104        })?;
105    let ActiveStorePublicationOwner::DeviceJoin(attempt_id) = active.owner() else {
106        return Ok(());
107    };
108    if !matches!(
109        &active.attempt()?.entry.payload,
110        coven_protocol::store_commit::StorePublicationPayload::Commit(reference)
111            if reference == candidate
112    ) {
113        return Err(DbError::Message(
114            "accepted device join candidate differs from its active publication".to_string(),
115        ));
116    }
117    let key = DeviceJoinJournalRecord::store_key_for(*attempt_id, DeviceJoinRole::Owner);
118    let current_value = crate::required_protocol_state_on(transaction, &key)?;
119    let current: DeviceJoinJournalRecord = serde_json::from_str(&current_value)
120        .map_err(|error| DbError::context("prepared device join journal", error))?;
121    let DeviceJoinRoleProgress::Owner(OwnerJoinProgress::StorePublicationPrepared(prepared)) =
122        &*current.progress
123    else {
124        return Err(DbError::Message(
125            "active device join publication has no prepared owner journal".to_string(),
126        ));
127    };
128    if prepared.candidate.reference != *candidate
129        || active.commit_reservation()
130            != Some((
131                &prepared.candidate.commit.write_id,
132                &prepared.candidate.commit.author_registration,
133                &candidate.coord,
134            ))
135    {
136        return Err(DbError::Message(
137            "active device join publication differs from its prepared journal".to_string(),
138        ));
139    }
140    let winning = active.attempt()?;
141    if matches!(
142        prepared.operation,
143        coven_protocol::store_commit::device_join_journal::OwnerJoinPublication::SamePrincipalActivation { .. }
144    ) {
145        let Some(coven_protocol::membership_mutation::StoreMembershipJournalCompletion::DeviceJoin {
146            remote_objects,
147        }) = completion else {
148            return Err(DbError::Message(
149                "same-principal handoff has no finalized registration authority".into(),
150            ));
151        };
152        let proof = prepared.candidate.history_evidence.membership_proof.as_ref()
153            .ok_or_else(|| DbError::Message("same-principal handoff has no exact authority head".into()))?;
154        let results = remote_objects.iter().filter_map(|record| {
155            let coven_protocol::remote_object::RemoteObjectRecord::RetainedAuthority(value) = record else {
156                return None;
157            };
158            match &value.identity.domain {
159                coven_protocol::remote_object::RetainedAuthorityObjectDomain::MembershipHeadAcceptance { head, publication }
160                    if head == &proof.head => Some(publication),
161                _ => None,
162            }
163        }).collect::<Vec<_>>();
164        if results.as_slice() != [&winning.reference()?] {
165            return Err(DbError::Message(
166                "same-principal handoff differs from its exact winning publication".into(),
167            ));
168        }
169    }
170    let accepted = DeviceJoinJournalRecord {
171        attempt_id: *attempt_id,
172        progress: Box::new(DeviceJoinRoleProgress::Owner(
173            prepared
174                .accepted_progress(*attempt_id, winning.replacement.clone())
175                .map_err(|error| DbError::context("accepted device join progress", error))?,
176        )),
177    };
178    crate::store::device_join_journal::validate_successor(&current, &accepted).map_err(
179        |error| DbError::Message(format!("complete owner device join publication: {error}")),
180    )?;
181    let accepted_value = serde_json::to_string(&accepted)
182        .map_err(|error| DbError::context("serialize accepted device join", error))?;
183    if transaction
184        .execute(
185            "UPDATE protocol_state SET value = ?1 WHERE key = ?2 AND value = ?3",
186            (&accepted_value, &key, &current_value),
187        )
188        .map_err(DbError::from)?
189        != 1
190    {
191        return Err(DbError::Message(
192            "device join journal changed during publication completion".to_string(),
193        ));
194    }
195    super::active_store_publication::clear_active_store_publication_on(transaction, &active)
196}
197
198impl StoreDatabase {
199    pub async fn prepare_owner_device_join_publication(
200        &self,
201        previous: DeviceJoinJournalRecord,
202        prepared: PreparedOwnerJoinPublication,
203    ) -> Result<DeviceJoinJournalRecord, crate::store::device_join_journal::DeviceJoinJournalError>
204    {
205        self.call_store(move |session| {
206            session.prepare_owner_device_join_publication(previous, prepared)
207        })
208        .await
209        .map_err(crate::store::device_join_journal::DeviceJoinJournalError::Database)
210    }
211}
coven_database/store/store_session/device_join_publication.rs

coven_database/store/store_session/
device_join_publication.rs
