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 remove_rotation_candidate_on(
138    tx: &rusqlite::Transaction<'_>,
139    intent_hash: ObjectHash,
140    generation: u64,
141) -> Result<(), DbError> {
142    let existing = load_rotation_gate_on(tx)?.ok_or_else(|| {
143        DbError::Message("rotation gate is absent during candidate loss".to_string())
144    })?;
145    let next = existing
146        .1
147        .clone()
148        .remove_candidate(generation, intent_hash)
149        .map_err(DbError::from)?;
150    replace_rotation_gate_on(tx, Some(&existing), next, "candidate loss")
151}
152
153pub(super) fn commit_rotation_candidate_on(
154    tx: &rusqlite::Transaction<'_>,
155    intent_hash: ObjectHash,
156    generation: u64,
157) -> Result<(), DbError> {
158    let existing = load_rotation_gate_on(tx)?.ok_or_else(|| {
159        DbError::Message("rotation gate is absent during candidate activation".to_string())
160    })?;
161    let gate = RotationGate::commit_candidate(Some(existing.1.clone()), generation, intent_hash)
162        .map_err(DbError::from)?;
163    replace_rotation_gate_on(tx, Some(&existing), Some(gate), "membership activation")
164}
165
166fn load_rotation_gate_on(
167    connection: &rusqlite::Connection,
168) -> Result<Option<(String, RotationGate)>, DbError> {
169    let key = coven_protocol::objects::ROTATION_GATE_STATE_KEY;
170    crate::get_protocol_state_on(connection, key)?
171        .map(|encoded| {
172            let gate = serde_json::from_str::<RotationGate>(&encoded)
173                .map_err(|error| DbError::context("parse rotation gate", error))?;
174            Ok((encoded, gate))
175        })
176        .transpose()
177}
178
179fn replace_rotation_gate_on(
180    tx: &rusqlite::Transaction<'_>,
181    expected: Option<&(String, RotationGate)>,
182    next: Option<RotationGate>,
183    operation: &'static str,
184) -> Result<(), DbError> {
185    let key = coven_protocol::objects::ROTATION_GATE_STATE_KEY;
186    let changed = match (expected, next) {
187        (Some((expected, _)), Some(next)) => {
188            let encoded = serde_json::to_string(&next).map_err(|error| {
189                DbError::context(format!("serialize rotation gate during {operation}"), error)
190            })?;
191            tx.execute(
192                "UPDATE protocol_state SET value = ?1 WHERE key = ?2 AND value = ?3",
193                (&encoded, key, expected),
194            )
195            .map_err(DbError::from)?
196        }
197        (Some((expected, _)), None) => tx
198            .execute(
199                "DELETE FROM protocol_state WHERE key = ?1 AND value = ?2",
200                (key, expected),
201            )
202            .map_err(DbError::from)?,
203        (None, Some(next)) => {
204            let encoded = serde_json::to_string(&next).map_err(|error| {
205                DbError::context(format!("serialize rotation gate during {operation}"), error)
206            })?;
207            tx.execute(
208                "INSERT INTO protocol_state (key, value) VALUES (?1, ?2)",
209                (key, &encoded),
210            )
211            .map_err(DbError::from)?
212        }
213        (None, None) => return Ok(()),
214    };
215    if changed != 1 {
216        return Err(DbError::Message(format!(
217            "rotation gate changed during {operation}"
218        )));
219    }
220    Ok(())
221}
coven_database/store/store_session/membership_rotation.rs

coven_database/store/store_session/
membership_rotation.rs
