coven_database/store/store_session/
membership_rotation.rs

1use super::{StoreDatabase, StoreSession};
2use crate::DbError;
3use coven_protocol::objects::RotationGate;
4use coven_protocol::store_commit::ObjectHash;
5
6impl StoreSession<'_> {
7    fn load_rotation_gate(&mut self) -> Result<Option<RotationGate>, DbError> {
8        load_rotation_gate_on(self.conn).map(|gate| gate.map(|(_, gate)| gate))
9    }
10
11    fn record_peer_rotation(&mut self, generation: u64) -> Result<RotationGate, DbError> {
12        let conn = self.conn;
13        let tx = conn.unchecked_transaction().map_err(DbError::from)?;
14        let existing = load_rotation_gate_on(&tx)?;
15        let next = RotationGate::merge_peer_commit(
16            existing.as_ref().map(|(_, gate)| gate.clone()),
17            generation,
18        )
19        .map_err(DbError::from)?;
20        replace_rotation_gate_on(
21            &tx,
22            existing.as_ref(),
23            Some(next.clone()),
24            "peer rotation recording",
25        )?;
26        tx.commit().map_err(DbError::from)?;
27        Ok(next)
28    }
29
30    fn complete_peer_rotation_adoption(
31        &mut self,
32        adopted_generation: u64,
33    ) -> Result<Option<RotationGate>, DbError> {
34        let conn = self.conn;
35        let tx = conn.unchecked_transaction().map_err(DbError::from)?;
36        let existing = load_rotation_gate_on(&tx)?.ok_or_else(|| {
37            DbError::Message("rotation gate is absent during peer rotation adoption".to_string())
38        })?;
39        let next = existing
40            .1
41            .clone()
42            .complete_peer_adoption(adopted_generation)
43            .map_err(DbError::from)?;
44        replace_rotation_gate_on(&tx, Some(&existing), next.clone(), "peer rotation adoption")?;
45        tx.commit().map_err(DbError::from)?;
46        Ok(next)
47    }
48
49    fn complete_local_rotation_adoption(
50        &mut self,
51        intent_hash: ObjectHash,
52        generation: u64,
53    ) -> Result<Option<RotationGate>, DbError> {
54        let conn = self.conn;
55        let tx = conn.unchecked_transaction().map_err(DbError::from)?;
56        let existing = load_rotation_gate_on(&tx)?.ok_or_else(|| {
57            DbError::Message("rotation gate is absent during local rotation adoption".to_string())
58        })?;
59        let next = existing
60            .1
61            .clone()
62            .complete_local_adoption(generation, intent_hash)
63            .map_err(DbError::from)?;
64        if tx
65            .execute(
66                "DELETE FROM outbound_membership_mutation \
67                 WHERE singleton = 1 AND intent_hash = ?1",
68                [intent_hash.to_string()],
69            )
70            .map_err(DbError::from)?
71            != 1
72        {
73            return Err(DbError::Message(
74                "membership mutation changed during local rotation adoption".to_string(),
75            ));
76        }
77        replace_rotation_gate_on(
78            &tx,
79            Some(&existing),
80            next.clone(),
81            "local rotation adoption",
82        )?;
83        tx.commit().map_err(DbError::from)?;
84        Ok(next)
85    }
86}
87
88impl StoreDatabase {
89    pub async fn load_rotation_gate(&self) -> Result<Option<RotationGate>, DbError> {
90        self.call_store(|session| session.load_rotation_gate())
91            .await
92    }
93
94    pub async fn record_peer_rotation(&self, generation: u64) -> Result<RotationGate, DbError> {
95        self.call_store(move |session| session.record_peer_rotation(generation))
96            .await
97    }
98
99    pub async fn complete_peer_rotation_adoption(
100        &self,
101        adopted_generation: u64,
102    ) -> Result<Option<RotationGate>, DbError> {
103        self.call_store(move |session| session.complete_peer_rotation_adoption(adopted_generation))
104            .await
105    }
106
107    pub async fn complete_local_rotation_adoption(
108        &self,
109        intent_hash: ObjectHash,
110        generation: u64,
111    ) -> Result<Option<RotationGate>, DbError> {
112        self.call_store(move |session| {
113            session.complete_local_rotation_adoption(intent_hash, generation)
114        })
115        .await
116    }
117}
118
119pub(super) fn stage_pending_rotation_on(
120    tx: &rusqlite::Transaction<'_>,
121    generation: Option<u64>,
122    mutation: ObjectHash,
123) -> Result<(), DbError> {
124    let Some(generation) = generation else {
125        return Ok(());
126    };
127    let existing = load_rotation_gate_on(tx)?;
128    let gate = RotationGate::with_candidate(
129        existing.as_ref().map(|(_, gate)| gate.clone()),
130        generation,
131        mutation,
132    )
133    .map_err(DbError::from)?;
134    replace_rotation_gate_on(tx, existing.as_ref(), Some(gate), "candidate staging")
135}
136
137pub(super) fn replace_rotation_candidate_mutation_on(
138    tx: &rusqlite::Transaction<'_>,
139    previous: ObjectHash,
140    replacement: ObjectHash,
141    generation: u64,
142) -> Result<(), DbError> {
143    let existing = load_rotation_gate_on(tx)?.ok_or_else(|| {
144        DbError::Message("rotation gate is absent during candidate replacement".to_string())
145    })?;
146    let next = existing
147        .1
148        .clone()
149        .replace_candidate_mutation(generation, previous, replacement)
150        .map_err(DbError::from)?;
151    replace_rotation_gate_on(tx, Some(&existing), Some(next), "candidate replacement")
152}
153
154pub(super) fn remove_rotation_candidate_on(
155    tx: &rusqlite::Transaction<'_>,
156    intent_hash: ObjectHash,
157    generation: u64,
158) -> Result<(), DbError> {
159    let existing = load_rotation_gate_on(tx)?.ok_or_else(|| {
160        DbError::Message("rotation gate is absent during candidate loss".to_string())
161    })?;
162    let next = existing
163        .1
164        .clone()
165        .remove_candidate(generation, intent_hash)
166        .map_err(DbError::from)?;
167    replace_rotation_gate_on(tx, Some(&existing), next, "candidate loss")
168}
169
170pub(super) fn commit_rotation_candidate_on(
171    tx: &rusqlite::Transaction<'_>,
172    intent_hash: ObjectHash,
173    generation: u64,
174) -> Result<(), DbError> {
175    let existing = load_rotation_gate_on(tx)?.ok_or_else(|| {
176        DbError::Message("rotation gate is absent during candidate activation".to_string())
177    })?;
178    let gate = RotationGate::commit_candidate(Some(existing.1.clone()), generation, intent_hash)
179        .map_err(DbError::from)?;
180    replace_rotation_gate_on(tx, Some(&existing), Some(gate), "membership activation")
181}
182
183fn load_rotation_gate_on(
184    connection: &rusqlite::Connection,
185) -> Result<Option<(String, RotationGate)>, DbError> {
186    let key = coven_protocol::objects::ROTATION_GATE_STATE_KEY;
187    crate::get_protocol_state_on(connection, key)?
188        .map(|encoded| {
189            let gate = serde_json::from_str::<RotationGate>(&encoded)
190                .map_err(|error| DbError::context("parse rotation gate", error))?;
191            Ok((encoded, gate))
192        })
193        .transpose()
194}
195
196fn replace_rotation_gate_on(
197    tx: &rusqlite::Transaction<'_>,
198    expected: Option<&(String, RotationGate)>,
199    next: Option<RotationGate>,
200    operation: &'static str,
201) -> Result<(), DbError> {
202    let key = coven_protocol::objects::ROTATION_GATE_STATE_KEY;
203    let changed = match (expected, next) {
204        (Some((expected, _)), Some(next)) => {
205            let encoded = serde_json::to_string(&next).map_err(|error| {
206                DbError::context(format!("serialize rotation gate during {operation}"), error)
207            })?;
208            tx.execute(
209                "UPDATE protocol_state SET value = ?1 WHERE key = ?2 AND value = ?3",
210                (&encoded, key, expected),
211            )
212            .map_err(DbError::from)?
213        }
214        (Some((expected, _)), None) => tx
215            .execute(
216                "DELETE FROM protocol_state WHERE key = ?1 AND value = ?2",
217                (key, expected),
218            )
219            .map_err(DbError::from)?,
220        (None, Some(next)) => {
221            let encoded = serde_json::to_string(&next).map_err(|error| {
222                DbError::context(format!("serialize rotation gate during {operation}"), error)
223            })?;
224            tx.execute(
225                "INSERT INTO protocol_state (key, value) VALUES (?1, ?2)",
226                (key, &encoded),
227            )
228            .map_err(DbError::from)?
229        }
230        (None, None) => return Ok(()),
231    };
232    if changed != 1 {
233        return Err(DbError::Message(format!(
234            "rotation gate changed during {operation}"
235        )));
236    }
237    Ok(())
238}
coven_database/store/store_session/membership_rotation.rs

coven_database/store/store_session/
membership_rotation.rs
