coven_database/store/store_session/
device_streams.rs

1use crate::*;
2use coven_protocol::store_commit::{
3    StoreAck, StoreAckRef, StoreDeviceRegistration, StoreDeviceRegistrationActivation,
4};
5
6use super::device_registration_journal::LocalRegistrationRecord;
7use super::*;
8
9impl StoreDatabase {
10    /// Install an already activated recovery together with its current exact
11    /// acknowledgement. Preparation performs every remote read before this call.
12    pub async fn install_adopted_owner_recovery(
13        &self,
14        registration: ExactProtocolObject<StoreDeviceRegistration>,
15        initial_ack_ref: StoreAckRef,
16        initial_ack: ExactProtocolObject<StoreAck>,
17        activation: StoreDeviceRegistrationActivation,
18        latest_ack: (StoreAckRef, StoreAck),
19    ) -> Result<(), DbError> {
20        const SUBJECT: &str = "adopted Owner recovery registration graph";
21        let record = LocalRegistrationRecord::checked_owner_recovery(
22            registration,
23            initial_ack_ref,
24            initial_ack,
25            &activation,
26            SUBJECT,
27        )?;
28        self.call_store(move |session| {
29            session.install_adopted_owner_recovery(record, activation, latest_ack, SUBJECT)
30        })
31        .await
32    }
33}
34
35impl StoreSession<'_> {
36    fn install_adopted_owner_recovery(
37        &mut self,
38        record: LocalRegistrationRecord,
39        activation: StoreDeviceRegistrationActivation,
40        (latest_ack_ref, latest_ack): (StoreAckRef, StoreAck),
41        subject: &str,
42    ) -> Result<(), DbError> {
43        let conn = self.conn;
44        let tx = conn.unchecked_transaction()?;
45        let root = self.required_root_authority()?;
46        record.require_installed_store_root(&root, subject)?;
47        let activated = self.activated_store_device_registration_with_authority(
48            root.clone(),
49            record.reference().clone(),
50        )?;
51        if activated.value() != record.registration() || activated.activation() != &activation {
52            return Err(DbError::Message(
53                "adopted recovery differs from its installed registration authority".into(),
54            ));
55        }
56        let accepted = self.activated_store_ack(record.reference())?;
57        let expected = match &accepted {
58            Some(accepted) => &accepted.reference,
59            None => record.initial_ack_ref(),
60        };
61        if &latest_ack_ref != expected {
62            return Err(DbError::Message(
63                "adopted recovery acknowledgement changed while its exact object was prepared"
64                    .into(),
65            ));
66        }
67        StoreAck::parse_at(
68            &latest_ack.to_bytes(),
69            &root,
70            &latest_ack_ref,
71            activated.value(),
72        )
73        .map_err(DbError::from)?;
74        let recorded = load_published_store_ack_on(&tx)?;
75        if let Some(recorded) = &recorded {
76            if recorded.reference.registration == *record.reference()
77                && (recorded.reference.sequence > latest_ack_ref.sequence
78                    || (recorded.reference.sequence == latest_ack_ref.sequence
79                        && (recorded.reference != latest_ack_ref
80                            || recorded.successor_slot != latest_ack.successor.next_slot)))
81            {
82                return Err(DbError::Message(
83                    "adopted recovery acknowledgement conflicts with its existing local stream"
84                        .into(),
85                ));
86            }
87        }
88        let standing = coven_protocol::store_commit::StandingStoreAck {
89            assertion: latest_ack.assertion(),
90            activating_commit: accepted.map(|accepted| accepted.activating_commit),
91        };
92        let ack_ref = serde_json::to_string(&latest_ack_ref)
93            .map_err(|error| DbError::context("adopted acknowledgement head", error))?;
94        let successor = serde_json::to_string(&latest_ack.successor.next_slot)
95            .map_err(|error| DbError::context("adopted acknowledgement successor", error))?;
96        let standing = serde_json::to_string(&standing)
97            .map_err(|error| DbError::context("adopted standing acknowledgement", error))?;
98        record.replace_journal_on(
99            &tx,
100            LocalDeviceRegistrationState::Activated {
101                authority: activation,
102            },
103            subject,
104        )?;
105        tx.execute(
106            "INSERT INTO published_store_acks (singleton, ack_ref, successor_slot, standing) \
107             VALUES (1, ?1, ?2, ?3) \
108             ON CONFLICT (singleton) DO UPDATE SET \
109                 ack_ref = excluded.ack_ref, successor_slot = excluded.successor_slot, \
110                 standing = excluded.standing",
111            (&ack_ref, &successor, &standing),
112        )?;
113        crate::set_protocol_state_on(&tx, LOCAL_DEVICE_ID_STATE_KEY, &record.device_id())?;
114        tx.commit().map_err(DbError::from)
115    }
116}
coven_database/store/store_session/device_streams.rs

coven_database/store/store_session/
device_streams.rs
