1use rusqlite::{Connection, OptionalExtension};
2
3use super::*;
4use crate::mark_remote_object_uploaded_on;
5use crate::store::StoreSession;
6use crate::ActiveStorePublication;
7use coven_protocol::device_exclusion_journal::{
8 DurableStoreDeviceExclusionObject, DurableStoreDeviceExclusionOperation,
9 StoreDeviceExclusionCompletion, StoreDeviceExclusionJournalError,
10};
11use coven_protocol::remote_object::{
12 ClosedRemoteObject, RemoteObjectRecord, RetainedAuthorityObjectState,
13};
14use coven_protocol::store_commit::ObjectHash;
15
16pub(crate) fn store_device_exclusion_journal_error(
17 error: StoreDeviceExclusionJournalError,
18) -> DbError {
19 DbError::from(error)
20}
21
22pub(crate) fn parse_store_device_exclusion_operation(
23 operation_id: ObjectHash,
24 raw: &str,
25) -> Result<DurableStoreDeviceExclusionOperation, DbError> {
26 let operation: DurableStoreDeviceExclusionOperation =
27 serde_json::from_str(raw).map_err(|error| {
28 DbError::context(
29 format!(
30 "Store-device exclusion operation {operation_id} has invalid durable state"
31 ),
32 error,
33 )
34 })?;
35 operation
36 .validate()
37 .map_err(store_device_exclusion_journal_error)?;
38 if operation.operation_id() != operation_id {
39 return Err(DbError::Message(format!(
40 "Store-device exclusion operation key {operation_id} differs from its signed object {}",
41 operation.operation_id()
42 )));
43 }
44 Ok(operation)
45}
46
47pub(crate) fn load_store_device_exclusion_on(
48 conn: &Connection,
49 operation_id: ObjectHash,
50) -> Result<Option<DurableStoreDeviceExclusionOperation>, DbError> {
51 conn.query_row(
52 "SELECT state FROM outbound_store_device_exclusion WHERE operation_id = ?1",
53 [operation_id.to_string()],
54 |row| row.get::<_, String>(0),
55 )
56 .optional()
57 .map_err(DbError::from)?
58 .map(|raw| parse_store_device_exclusion_operation(operation_id, &raw))
59 .transpose()
60}
61
62pub(crate) fn load_active_store_device_exclusion_on(
63 conn: &Connection,
64) -> Result<Option<DurableStoreDeviceExclusionOperation>, DbError> {
65 conn.query_row(
66 "SELECT operation_id, state FROM outbound_store_device_exclusion WHERE active_key = 1",
67 [],
68 |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
69 )
70 .optional()
71 .map_err(DbError::from)?
72 .map(|(raw_id, raw)| {
73 let operation_id = raw_id
74 .parse::<ObjectHash>()
75 .map_err(|error| DbError::context("Store-device exclusion operation id", error))?;
76 let operation = parse_store_device_exclusion_operation(operation_id, &raw)?;
77 if operation.is_completed() {
78 return Err(DbError::Message(
79 "completed Store-device exclusion remains active".to_string(),
80 ));
81 }
82 Ok(operation)
83 })
84 .transpose()
85}
86
87pub(crate) fn insert_store_device_exclusion_on(
88 conn: &Connection,
89 operation: &DurableStoreDeviceExclusionOperation,
90 active: bool,
91) -> Result<(), DbError> {
92 operation
93 .validate()
94 .map_err(store_device_exclusion_journal_error)?;
95 if active == operation.is_completed() {
96 return Err(DbError::Message(
97 "Store-device exclusion active marker differs from its closed state".to_string(),
98 ));
99 }
100 let encoded = serde_json::to_string(operation)
101 .map_err(|error| DbError::context("serialize Store-device exclusion operation", error))?;
102 conn.execute(
103 "INSERT INTO outbound_store_device_exclusion (operation_id, active_key, state)
104 VALUES (?1, ?2, ?3)",
105 rusqlite::params![
106 operation.operation_id().to_string(),
107 active.then_some(1_i64),
108 encoded,
109 ],
110 )
111 .map(|_| ())
112 .map_err(DbError::from)
113}
114
115pub(crate) fn require_store_device_exclusion_transition_on(
116 conn: &Connection,
117 expected: &DurableStoreDeviceExclusionOperation,
118 next: &DurableStoreDeviceExclusionOperation,
119) -> Result<(), DbError> {
120 if !expected.allows_transition_to(next) {
121 return Err(DbError::Message(
122 "invalid Store-device exclusion journal transition".to_string(),
123 ));
124 }
125 let expected_state = serde_json::to_string(expected)
126 .map_err(|error| DbError::context("serialize expected Store-device exclusion", error))?;
127 let current = conn
128 .query_row(
129 "SELECT state FROM outbound_store_device_exclusion WHERE operation_id = ?1",
130 [expected.operation_id().to_string()],
131 |row| row.get::<_, String>(0),
132 )
133 .optional()
134 .map_err(DbError::from)?
135 .ok_or_else(|| {
136 DbError::Message("Store-device exclusion journal disappeared".to_string())
137 })?;
138 if current != expected_state {
139 return Err(DbError::Message(
140 "Store-device exclusion journal changed during transition".to_string(),
141 ));
142 }
143 Ok(())
144}
145
146pub(crate) fn update_store_device_exclusion_on(
147 conn: &Connection,
148 expected: &DurableStoreDeviceExclusionOperation,
149 next: &DurableStoreDeviceExclusionOperation,
150 active: bool,
151) -> Result<(), DbError> {
152 require_store_device_exclusion_transition_on(conn, expected, next)?;
153 if active == next.is_completed() {
154 return Err(DbError::Message(
155 "Store-device exclusion active marker differs from its next state".to_string(),
156 ));
157 }
158 let expected_state = serde_json::to_string(expected)
159 .map_err(|error| DbError::context("serialize expected Store-device exclusion", error))?;
160 let next_state = serde_json::to_string(next)
161 .map_err(|error| DbError::context("serialize next Store-device exclusion", error))?;
162 let updated = conn
163 .execute(
164 "UPDATE outbound_store_device_exclusion
165 SET active_key = ?3, state = ?4
166 WHERE operation_id = ?1 AND state = ?2",
167 rusqlite::params![
168 expected.operation_id().to_string(),
169 expected_state,
170 active.then_some(1_i64),
171 next_state,
172 ],
173 )
174 .map_err(DbError::from)?;
175 if updated != 1 {
176 return Err(DbError::Message(
177 "Store-device exclusion journal disappeared during transition".to_string(),
178 ));
179 }
180 Ok(())
181}
182
183pub(crate) fn complete_store_device_exclusion_activation_on(
184 conn: &Connection,
185 expected: &DurableStoreDeviceExclusionOperation,
186 acceptance: &crate::AcceptedStoreCommitEvidence,
187) -> Result<(), DbError> {
188 let next = expected
189 .activated()
190 .map_err(store_device_exclusion_journal_error)?;
191 let candidate = expected
192 .candidate()
193 .expect("validated pending exclusion has a candidate");
194 if &candidate.reference != acceptance.commit_ref() {
195 return Err(DbError::Message(
196 "Store-device exclusion completion names another accepted candidate".into(),
197 ));
198 }
199 update_store_device_exclusion_on(conn, expected, &next, false)?;
200 super::active_store_publication::clear_active_store_commit_for_owner_on(
201 conn,
202 &crate::ActiveStorePublicationOwner::DeviceExclusion(expected.operation_id()),
203 &candidate.reference,
204 )
205}
206
207impl StoreSession<'_> {
208 fn begin_outbound_store_device_exclusion(
209 &mut self,
210 operation: DurableStoreDeviceExclusionOperation,
211 remotes: Vec<ClosedRemoteObject>,
212 ) -> Result<DurableStoreDeviceExclusionOperation, DbError> {
213 let conn = self.conn;
214 let tx = conn.unchecked_transaction().map_err(DbError::from)?;
215 if let Some(active) = load_active_store_device_exclusion_on(&tx)? {
216 if active.operation_id() != operation.operation_id() {
217 return Err(DbError::Message(format!(
218 "Store-device exclusion operation {} remains active",
219 active.operation_id()
220 )));
221 }
222 return Ok(active);
223 }
224 let operation_id = operation.operation_id();
225 if let Some(existing) = load_store_device_exclusion_on(&tx, operation_id)? {
226 if existing != operation || !existing.is_completed() {
227 return Err(DbError::Message(format!(
228 "Store-device exclusion operation {operation_id} already has different durable state"
229 )));
230 }
231 return Ok(existing);
232 }
233 let candidate = operation.candidate().ok_or_else(|| {
234 DbError::Message(
235 "active Store-device exclusion has no publication candidate".to_string(),
236 )
237 })?;
238 let active_publication = ActiveStorePublication::for_commit(
239 crate::ActiveStorePublicationOwner::DeviceExclusion(operation_id),
240 candidate,
241 )?;
242 match super::active_store_publication::claim_active_store_publication_on(
243 &tx,
244 &active_publication,
245 )? {
246 super::active_store_publication::ActiveStorePublicationClaim::Acquired => {}
247 super::active_store_publication::ActiveStorePublicationClaim::AlreadyOwned => {
248 return Err(DbError::Message(
249 "Store-device exclusion candidate already owns publication before its journal"
250 .to_string(),
251 ));
252 }
253 super::active_store_publication::ActiveStorePublicationClaim::Occupied(source) => {
254 return Err(DbError::Message(format!(
255 "another local Store operation owns publication: {source:?}"
256 )));
257 }
258 }
259 for remote in &remotes {
260 persist_exact_remote_object_on(
261 &tx,
262 self.store_dir,
263 remote,
264 "Store-device exclusion candidate object",
265 )?;
266 }
267 insert_store_device_exclusion_on(&tx, &operation, true)?;
268 tx.commit().map_err(DbError::from)?;
269 Ok(operation)
270 }
271
272 fn active_outbound_store_device_exclusion(
273 &mut self,
274 ) -> Result<Option<DurableStoreDeviceExclusionOperation>, DbError> {
275 load_active_store_device_exclusion_on(self.conn)
276 }
277
278 fn complete_outbound_store_device_exclusion_slot_loss(
279 &mut self,
280 expected: DurableStoreDeviceExclusionOperation,
281 next: DurableStoreDeviceExclusionOperation,
282 remotes: Vec<ClosedRemoteObject>,
283 ) -> Result<DurableStoreDeviceExclusionOperation, DbError> {
284 let conn = self.conn;
285 let tx = conn.unchecked_transaction().map_err(DbError::from)?;
286 require_store_device_exclusion_transition_on(&tx, &expected, &next)?;
287 for remote in &remotes {
288 let object_id = remote.object_id();
289 let current = load_remote_object_on(&tx, object_id)?;
290 let unuploaded = matches!(
291 ¤t,
292 RemoteObjectRecord::CandidateCommit(record)
293 if matches!(record.state, coven_protocol::remote_object::CandidateCommitState::Prepared)
294 ) || matches!(
295 ¤t,
296 RemoteObjectRecord::CandidateExclusive(record)
297 if matches!(
298 record.state,
299 coven_protocol::remote_object::CandidateObjectState::Prepared { .. }
300 )
301 ) || matches!(
302 ¤t,
303 RemoteObjectRecord::RetainedAuthority(record)
304 if matches!(
305 record.state,
306 coven_protocol::remote_object::RetainedAuthorityObjectState::Prepared { .. }
307 )
308 );
309 if current != **remote || !unuploaded {
310 return Err(DbError::Message(format!(
311 "outcome-slot loss cannot discard uploaded exclusion object {object_id}"
312 )));
313 }
314 if !crate::remote_object_records::delete_remote_object_on(&tx, object_id)? {
315 return Err(DbError::Message(format!(
316 "unuploaded exclusion object {object_id} disappeared during slot resolution"
317 )));
318 }
319 }
320 let candidate = expected.candidate().ok_or_else(|| {
321 DbError::Message("Store-device exclusion slot loss has no candidate".to_string())
322 })?;
323 let active_publication = ActiveStorePublication::for_commit(
324 crate::ActiveStorePublicationOwner::DeviceExclusion(expected.operation_id()),
325 candidate,
326 )?;
327 update_store_device_exclusion_on(&tx, &expected, &next, false)?;
328 super::active_store_publication::clear_active_store_publication_on(
329 &tx,
330 &active_publication,
331 )?;
332 tx.commit().map_err(DbError::from)?;
333 Ok(next)
334 }
335
336 fn mark_store_device_exclusion_authority_uploaded(
337 &mut self,
338 expected: ClosedRemoteObject,
339 candidate: StoreBatchCommitRef,
340 ) -> Result<(), DbError> {
341 let conn = self.conn;
342 let object_id = expected.object_id();
343 let current = load_remote_object_on(conn, object_id)?;
344 let (
345 RemoteObjectRecord::RetainedAuthority(expected_record),
346 RemoteObjectRecord::RetainedAuthority(current_record),
347 ) = (expected.record(), ¤t)
348 else {
349 return Err(DbError::Message(
350 "Store-device exclusion authority is not retained authority".to_string(),
351 ));
352 };
353 if expected_record.identity != current_record.identity
354 || expected_record.payloads != current_record.payloads
355 {
356 return Err(DbError::Message(
357 "Store-device exclusion authority changed before upload completion".to_string(),
358 ));
359 }
360 match ¤t_record.state {
361 RetainedAuthorityObjectState::Prepared { ownership }
362 if ownership.pending.contains(&candidate) =>
363 {
364 mark_remote_object_uploaded_on(conn, current)?;
365 }
366 RetainedAuthorityObjectState::UploadedVerified { ownership }
367 if ownership.pending.contains(&candidate) => {}
368 _ => {
369 return Err(DbError::Message(
370 "Store-device exclusion authority does not belong to its current candidate"
371 .to_string(),
372 ));
373 }
374 }
375 Ok(())
376 }
377
378 #[cfg(any(test, feature = "test-utils"))]
379 fn outbound_store_device_exclusion_operations(
380 &mut self,
381 ) -> Result<Vec<DurableStoreDeviceExclusionOperation>, DbError> {
382 let conn = self.conn;
383 let mut statement = conn
384 .prepare(
385 "SELECT operation_id, state
386 FROM outbound_store_device_exclusion
387 ORDER BY operation_id",
388 )
389 .map_err(DbError::from)?;
390 let operations = statement
391 .query_map([], |row| {
392 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
393 })
394 .map_err(DbError::from)?
395 .map(|row| {
396 let (raw_id, raw) = row.map_err(DbError::from)?;
397 let operation_id = raw_id.parse::<ObjectHash>().map_err(|error| {
398 DbError::context("Store-device exclusion operation id", error)
399 })?;
400 parse_store_device_exclusion_operation(operation_id, &raw)
401 })
402 .collect();
403 operations
404 }
405}
406
407impl StoreDatabase {
408 pub async fn begin_outbound_store_device_exclusion(
409 &self,
410 operation: DurableStoreDeviceExclusionOperation,
411 ) -> Result<DurableStoreDeviceExclusionOperation, DbError> {
412 operation
413 .validate()
414 .map_err(store_device_exclusion_journal_error)?;
415 if !matches!(
416 operation,
417 DurableStoreDeviceExclusionOperation::CandidatePrepared { .. }
418 ) {
419 return Err(DbError::Message(
420 "a new Store-device exclusion journal must own its exact activation candidate"
421 .to_string(),
422 ));
423 }
424 let remotes = operation
425 .remote_objects()
426 .map_err(store_device_exclusion_journal_error)?;
427 Box::pin(self.call_store(move |session| {
428 session.begin_outbound_store_device_exclusion(operation, remotes)
429 }))
430 .await
431 }
432
433 pub async fn active_outbound_store_device_exclusion(
434 &self,
435 ) -> Result<Option<DurableStoreDeviceExclusionOperation>, DbError> {
436 Box::pin(self.call_store(|session| session.active_outbound_store_device_exclusion())).await
437 }
438
439 pub async fn complete_outbound_store_device_exclusion_slot_loss(
440 &self,
441 expected: DurableStoreDeviceExclusionOperation,
442 winner: DurableStoreDeviceExclusionObject,
443 ) -> Result<DurableStoreDeviceExclusionOperation, DbError> {
444 let next = DurableStoreDeviceExclusionOperation::Completed(
445 StoreDeviceExclusionCompletion::OutcomeSlotOccupied {
446 intended: expected.object().clone(),
447 winner,
448 },
449 );
450 next.validate()
451 .map_err(store_device_exclusion_journal_error)?;
452 let remotes = expected
453 .remote_objects()
454 .map_err(store_device_exclusion_journal_error)?;
455 Box::pin(self.call_store(move |session| {
456 session.complete_outbound_store_device_exclusion_slot_loss(expected, next, remotes)
457 }))
458 .await
459 }
460
461 pub async fn mark_store_device_exclusion_authority_uploaded(
462 &self,
463 operation: DurableStoreDeviceExclusionOperation,
464 ) -> Result<(), DbError> {
465 let expected = operation
466 .authority_remote_object()
467 .map_err(store_device_exclusion_journal_error)?;
468 let candidate = operation
469 .candidate()
470 .ok_or_else(|| {
471 DbError::Message(
472 "Store-device exclusion authority has no current candidate".to_string(),
473 )
474 })?
475 .reference
476 .clone();
477 self.call_store(move |session| {
478 session.mark_store_device_exclusion_authority_uploaded(expected, candidate)
479 })
480 .await
481 }
482
483 #[cfg(any(test, feature = "test-utils"))]
484 pub async fn outbound_store_device_exclusion_operations(
485 &self,
486 ) -> Result<Vec<DurableStoreDeviceExclusionOperation>, DbError> {
487 Box::pin(self.call_store(|session| session.outbound_store_device_exclusion_operations()))
488 .await
489 }
490}