1use super::candidate_records::{
2 begin_candidate_nonactivation_targets_on, candidate_cleanup_targets_on,
3};
4use super::*;
5use crate::store::StoreSession;
6use crate::*;
7use coven_protocol::objects::ExactObjectRef;
8use coven_protocol::remote_object::RemoteObjectRecord;
9use coven_protocol::store_commit::{ObjectHash, StoreBatchCommitRef};
10use rusqlite::OptionalExtension;
11use std::collections::BTreeSet;
12
13impl StoreSession<'_> {
14 fn outbound_membership_mutation(
15 &mut self,
16 ) -> Result<Option<DurableMembershipMutation>, DbError> {
17 let conn = self.conn;
18 conn.query_row(
19 "SELECT intent_hash, plan_bytes, progress_bytes \
20 FROM outbound_membership_mutation WHERE singleton = 1",
21 [],
22 |row| {
23 Ok((
24 row.get::<_, String>(0)?,
25 row.get::<_, Vec<u8>>(1)?,
26 row.get::<_, Vec<u8>>(2)?,
27 ))
28 },
29 )
30 .optional()
31 .map_err(DbError::from)?
32 .map(|(hash, plan_bytes, progress_bytes)| {
33 let intent_hash: ObjectHash = hash
34 .parse()
35 .map_err(|error| DbError::context("membership intent hash", error))?;
36 if ObjectHash::digest(&plan_bytes) != intent_hash {
37 return Err(DbError::Message(
38 "membership intent hash differs from its exact plan bytes".to_string(),
39 ));
40 }
41 Ok(DurableMembershipMutation {
42 intent_hash,
43 plan_bytes,
44 progress_bytes,
45 })
46 })
47 .transpose()
48 }
49
50 fn select_causal_author_stream(
51 &mut self,
52 key: &str,
53 reusable: &std::collections::BTreeSet<coven_protocol::membership::AuthorStreamId>,
54 candidate: coven_protocol::membership::AuthorStreamId,
55 ) -> Result<coven_protocol::membership::AuthorStreamId, DbError> {
56 let conn = self.conn;
57 let existing = crate::get_protocol_state_on(conn, key)?
58 .map(|value| value.parse().map_err(DbError::from))
59 .transpose()?;
60 if let Some(existing) = existing {
61 if reusable.contains(&existing) {
62 return Ok(existing);
63 }
64 }
65 let selected = reusable.iter().next_back().copied().unwrap_or(candidate);
66 crate::set_protocol_state_on(conn, key, &selected.to_string())?;
67 Ok(selected)
68 }
69
70 fn stage_membership_mutation(
71 &mut self,
72 plan_bytes: Vec<u8>,
73 progress_bytes: Vec<u8>,
74 pending_rotation_generation: Option<u64>,
75 ) -> Result<ObjectHash, DbError> {
76 let conn = self.conn;
77 let tx = conn.unchecked_transaction().map_err(DbError::from)?;
78 let intent_hash = ObjectHash::digest(&plan_bytes);
79 let existing = tx
80 .query_row(
81 "SELECT intent_hash, plan_bytes FROM outbound_membership_mutation \
82 WHERE singleton = 1",
83 [],
84 |row| Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?)),
85 )
86 .optional()
87 .map_err(DbError::from)?;
88 if let Some((existing_hash, existing_plan)) = existing {
89 if existing_hash == intent_hash.to_string() && existing_plan == plan_bytes {
90 super::membership_rotation::stage_pending_rotation_on(
91 &tx,
92 pending_rotation_generation,
93 intent_hash,
94 )?;
95 tx.commit().map_err(DbError::from)?;
96 return Ok(intent_hash);
97 }
98 return Err(DbError::Message(
99 "a different membership mutation is already pending".to_string(),
100 ));
101 }
102 tx.execute(
103 "INSERT INTO outbound_membership_mutation \
104 (singleton, intent_hash, plan_bytes, progress_bytes) \
105 VALUES (1, ?1, ?2, ?3)",
106 rusqlite::params![intent_hash.to_string(), plan_bytes, progress_bytes],
107 )
108 .map_err(DbError::from)?;
109 super::membership_rotation::stage_pending_rotation_on(
110 &tx,
111 pending_rotation_generation,
112 intent_hash,
113 )?;
114 tx.commit().map_err(DbError::from)?;
115 Ok(intent_hash)
116 }
117
118 fn stage_membership_candidate_mutation(
119 &mut self,
120 plan_bytes: Vec<u8>,
121 progress_bytes: Vec<u8>,
122 remote_objects: Vec<coven_protocol::remote_object::ClosedRemoteObject>,
123 pending_rotation_generation: Option<u64>,
124 ) -> Result<ObjectHash, DbError> {
125 let conn = self.conn;
126 let intent_hash = ObjectHash::digest(&plan_bytes);
127 let tx = conn.unchecked_transaction().map_err(DbError::from)?;
128 let existing = tx
129 .query_row(
130 "SELECT intent_hash, plan_bytes FROM outbound_membership_mutation \
131 WHERE singleton = 1",
132 [],
133 |row| Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?)),
134 )
135 .optional()
136 .map_err(DbError::from)?;
137 if let Some((existing_hash, existing_plan)) = existing {
138 if existing_hash != intent_hash.to_string() || existing_plan != plan_bytes {
139 return Err(DbError::Message(
140 "a different membership mutation is already pending".to_string(),
141 ));
142 }
143 for remote in &remote_objects {
144 let stored = load_remote_object_on(&tx, remote.object_id())?;
145 if stored != **remote {
146 return Err(DbError::Message(
147 "persisted membership ownership differs from its durable plan".to_string(),
148 ));
149 }
150 }
151 super::membership_rotation::stage_pending_rotation_on(
152 &tx,
153 pending_rotation_generation,
154 intent_hash,
155 )?;
156 tx.commit().map_err(DbError::from)?;
157 return Ok(intent_hash);
158 }
159 if remote_objects.is_empty() {
160 return Err(DbError::Message(
161 "membership candidate mutation has no remote ownership graph".to_string(),
162 ));
163 }
164 let mut object_ids = BTreeSet::new();
165 for remote in &remote_objects {
166 if !object_ids.insert(remote.object_id()) {
167 return Err(DbError::Message(
168 "membership candidate mutation repeats a remote object".to_string(),
169 ));
170 }
171 persist_exact_remote_object_on(
172 &tx,
173 self.store_dir,
174 remote,
175 "membership candidate object",
176 )?;
177 }
178 tx.execute(
179 "INSERT INTO outbound_membership_mutation \
180 (singleton, intent_hash, plan_bytes, progress_bytes) \
181 VALUES (1, ?1, ?2, ?3)",
182 rusqlite::params![intent_hash.to_string(), plan_bytes, progress_bytes],
183 )
184 .map_err(DbError::from)?;
185 super::membership_rotation::stage_pending_rotation_on(
186 &tx,
187 pending_rotation_generation,
188 intent_hash,
189 )?;
190 tx.commit().map_err(DbError::from)?;
191 Ok(intent_hash)
192 }
193
194 fn update_membership_mutation_progress(
195 &mut self,
196 intent_hash: ObjectHash,
197 progress_bytes: Vec<u8>,
198 ) -> Result<(), DbError> {
199 let conn = self.conn;
200 let updated = conn
201 .execute(
202 "UPDATE outbound_membership_mutation SET progress_bytes = ?1 \
203 WHERE singleton = 1 AND intent_hash = ?2",
204 rusqlite::params![progress_bytes, intent_hash.to_string()],
205 )
206 .map_err(DbError::from)?;
207 if updated != 1 {
208 return Err(DbError::Message(
209 "membership mutation ownership row is absent or changed".to_string(),
210 ));
211 }
212 Ok(())
213 }
214
215 fn adopt_merge_membership_candidate_head(
216 &mut self,
217 intent_hash: ObjectHash,
218 plan_bytes: Vec<u8>,
219 previous: RemoteObjectRecord,
220 replacement: coven_protocol::remote_object::ClosedRemoteObject,
221 rotation_generation: Option<u64>,
222 replacement_hash: ObjectHash,
223 ) -> Result<ObjectHash, DbError> {
224 let conn = self.conn;
225 let tx = conn.unchecked_transaction().map_err(DbError::from)?;
226 let previous_id = previous.object_id();
227 let current = load_remote_object_on(&tx, previous_id)?;
228 if current != previous {
229 return Err(DbError::Message(
230 "Merge membership candidate head changed before receipt adoption".to_string(),
231 ));
232 }
233 if !crate::remote_object_records::delete_remote_object_on(&tx, previous_id)? {
234 return Err(DbError::Message(
235 "prepared Merge membership head disappeared during receipt adoption".to_string(),
236 ));
237 }
238 persist_exact_remote_object_on(
239 &tx,
240 self.store_dir,
241 &replacement,
242 "adopted Merge membership candidate head",
243 )?;
244 if tx
245 .execute(
246 "UPDATE outbound_membership_mutation
247 SET intent_hash = ?1, plan_bytes = ?2
248 WHERE singleton = 1 AND intent_hash = ?3",
249 rusqlite::params![
250 replacement_hash.to_string(),
251 plan_bytes,
252 intent_hash.to_string()
253 ],
254 )
255 .map_err(DbError::from)?
256 != 1
257 {
258 return Err(DbError::Message(
259 "membership mutation changed before Merge head receipt adoption".to_string(),
260 ));
261 }
262 if let Some(generation) = rotation_generation {
263 super::membership_rotation::replace_rotation_candidate_mutation_on(
264 &tx,
265 intent_hash,
266 replacement_hash,
267 generation,
268 )?;
269 }
270 tx.commit().map_err(DbError::from)?;
271 Ok(replacement_hash)
272 }
273
274 fn begin_membership_candidate_nonactivation(
275 &mut self,
276 intent_hash: ObjectHash,
277 candidate: StoreBatchCommitRef,
278 candidate_objects: Vec<ExactObjectRef>,
279 retained_authorities: Vec<ExactObjectRef>,
280 progress_bytes: Vec<u8>,
281 nonactivation: coven_protocol::remote_object::CandidateNonactivation,
282 ) -> Result<Vec<CandidateCleanupObject>, DbError> {
283 let conn = self.conn;
284 let tx = conn.unchecked_transaction().map_err(DbError::from)?;
285 let exists: bool = tx
286 .query_row(
287 "SELECT EXISTS(
288 SELECT 1 FROM outbound_membership_mutation
289 WHERE singleton = 1 AND intent_hash = ?1
290 )",
291 [intent_hash.to_string()],
292 |row| row.get(0),
293 )
294 .map_err(DbError::from)?;
295 if !exists {
296 return Err(DbError::Message(
297 "membership candidate mutation changed before nonactivation".to_string(),
298 ));
299 }
300 let owned = candidate_objects
301 .iter()
302 .chain(retained_authorities.iter())
303 .cloned()
304 .collect::<Vec<_>>();
305 let cleanup =
306 begin_candidate_nonactivation_targets_on(&tx, &candidate, &owned, &nonactivation)?;
307 let updated = tx
308 .execute(
309 "UPDATE outbound_membership_mutation SET progress_bytes = ?1 \
310 WHERE singleton = 1 AND intent_hash = ?2",
311 rusqlite::params![progress_bytes, intent_hash.to_string()],
312 )
313 .map_err(DbError::from)?;
314 if updated != 1 {
315 return Err(DbError::Message(
316 "membership candidate mutation changed during nonactivation".to_string(),
317 ));
318 }
319 tx.commit().map_err(DbError::from)?;
320 Ok(cleanup)
321 }
322
323 fn complete_nonactivating_membership_candidate_mutation(
324 &mut self,
325 intent_hash: ObjectHash,
326 candidate: StoreBatchCommitRef,
327 candidate_objects: Vec<ExactObjectRef>,
328 retained_authorities: Vec<ExactObjectRef>,
329 rotation_generation: Option<u64>,
330 ) -> Result<(), DbError> {
331 let conn = self.conn;
332 let tx = conn.unchecked_transaction().map_err(DbError::from)?;
333 let mut unique = BTreeSet::new();
334 for object in &candidate_objects {
335 let object_id = remote_object_id(object);
336 if !unique.insert(object_id) {
337 return Err(DbError::Message(
338 "nonactivating membership candidate repeats an exact object".to_string(),
339 ));
340 }
341 }
342 super::candidate_records::require_candidate_cleanup_complete_on(
343 &tx,
344 &candidate,
345 &candidate_objects,
346 "losing membership candidate cleanup is incomplete",
347 )?;
348 for object in &retained_authorities {
349 let object_id = remote_object_id(object);
350 if !unique.insert(object_id) {
351 return Err(DbError::Message(
352 "nonactivating membership authority repeats an exact object".to_string(),
353 ));
354 }
355 let remote = tx
356 .query_row(
357 "SELECT state FROM remote_objects WHERE object_id = ?1",
358 [object_id.to_string()],
359 |row| row.get::<_, String>(0),
360 )
361 .optional()
362 .map_err(DbError::from)?
363 .map(|encoded| {
364 serde_json::from_str::<RemoteObjectRecord>(&encoded).map_err(|error| {
365 DbError::context(
366 format!("parse nonactivating membership authority {object_id}"),
367 error,
368 )
369 })
370 })
371 .transpose()?;
372 match remote {
373 Some(remote) => {
374 if !remote
375 .candidate_cleanup_complete(&candidate)
376 .map_err(DbError::from)?
377 {
378 return Err(DbError::Message(format!(
379 "membership authority {object_id} still owns its losing candidate"
380 )));
381 }
382 }
383 None => {
384 let inert = load_protocol_inert_object_on(&tx, object_id)?;
385 if inert.object_id() != object_id {
386 return Err(DbError::Message(
387 "protocol-inert membership authority changed exact identity"
388 .to_string(),
389 ));
390 }
391 }
392 }
393 }
394 super::candidate_records::delete_remote_objects_on(
395 &tx,
396 candidate_objects.iter().map(remote_object_id),
397 "losing membership",
398 )?;
399 for object in retained_authorities {
400 let object_id = remote_object_id(&object);
401 let removable = tx
402 .query_row(
403 "SELECT state FROM remote_objects WHERE object_id = ?1",
404 [object_id.to_string()],
405 |row| row.get::<_, String>(0),
406 )
407 .optional()
408 .map_err(DbError::from)?
409 .map(|encoded| {
410 serde_json::from_str::<RemoteObjectRecord>(&encoded).map_err(|error| {
411 DbError::context(
412 format!("parse terminal membership authority {object_id}"),
413 error,
414 )
415 })
416 })
417 .transpose()?
418 .is_some_and(|remote| {
419 matches!(
420 remote,
421 RemoteObjectRecord::RetainedAuthority(
422 coven_protocol::remote_object::RetainedAuthorityRecord {
423 state: coven_protocol::remote_object::RetainedAuthorityObjectState::UncreatedVerified { .. },
424 ..
425 }
426 )
427 )
428 });
429 if removable {
430 crate::remote_object_records::delete_remote_object_on(&tx, object_id)?;
431 }
432 }
433 if tx
434 .execute(
435 "DELETE FROM outbound_membership_mutation \
436 WHERE singleton = 1 AND intent_hash = ?1",
437 [intent_hash.to_string()],
438 )
439 .map_err(DbError::from)?
440 != 1
441 {
442 return Err(DbError::Message(
443 "membership mutation changed during nonactivation completion".to_string(),
444 ));
445 }
446 if let Some(generation) = rotation_generation {
447 super::membership_rotation::remove_rotation_candidate_on(&tx, intent_hash, generation)?;
448 }
449 tx.commit().map_err(DbError::from)
450 }
451
452 fn membership_candidate_cleanup_targets(
453 &mut self,
454 intent_hash: ObjectHash,
455 candidate: &StoreBatchCommitRef,
456 objects: &[ExactObjectRef],
457 ) -> Result<Vec<CandidateCleanupObject>, DbError> {
458 let conn = self.conn;
459 let exists: bool = conn
460 .query_row(
461 "SELECT EXISTS(
462 SELECT 1 FROM outbound_membership_mutation
463 WHERE singleton = 1 AND intent_hash = ?1
464 )",
465 [intent_hash.to_string()],
466 |row| row.get(0),
467 )
468 .map_err(DbError::from)?;
469 if !exists {
470 return Err(DbError::Message(
471 "membership mutation changed before candidate cleanup".to_string(),
472 ));
473 }
474 candidate_cleanup_targets_on(conn, candidate, objects)
475 }
476
477 fn record_direct_revoke_activation(
478 &mut self,
479 intent_hash: ObjectHash,
480 progress_bytes: Vec<u8>,
481 generation: u64,
482 ) -> Result<(), DbError> {
483 let conn = self.conn;
484 let tx = conn.unchecked_transaction().map_err(DbError::from)?;
485 if tx
486 .execute(
487 "UPDATE outbound_membership_mutation SET progress_bytes = ?1 \
488 WHERE singleton = 1 AND intent_hash = ?2",
489 rusqlite::params![progress_bytes, intent_hash.to_string()],
490 )
491 .map_err(DbError::from)?
492 != 1
493 {
494 return Err(DbError::Message(
495 "direct revoke mutation changed during activation".to_string(),
496 ));
497 }
498 super::membership_rotation::commit_rotation_candidate_on(&tx, intent_hash, generation)?;
499 tx.commit().map_err(DbError::from)
500 }
501
502 fn complete_membership_mutation(&mut self, intent_hash: ObjectHash) -> Result<(), DbError> {
503 let conn = self.conn;
504 let deleted = conn
505 .execute(
506 "DELETE FROM outbound_membership_mutation \
507 WHERE singleton = 1 AND intent_hash = ?1",
508 [intent_hash.to_string()],
509 )
510 .map_err(DbError::from)?;
511 if deleted != 1 {
512 return Err(DbError::Message(
513 "membership mutation ownership row is absent or changed".to_string(),
514 ));
515 }
516 Ok(())
517 }
518}
519
520impl StoreDatabase {
521 pub async fn outbound_membership_mutation(
522 &self,
523 ) -> Result<Option<DurableMembershipMutation>, DbError> {
524 self.call_store(|session| session.outbound_membership_mutation())
525 .await
526 }
527
528 pub async fn select_membership_author_stream(
529 &self,
530 author_pubkey: &str,
531 author_owner_grant: &coven_protocol::membership::MembershipGrantId,
532 reusable: std::collections::BTreeSet<coven_protocol::membership::AuthorStreamId>,
533 ) -> Result<coven_protocol::membership::AuthorStreamId, DbError> {
534 self.select_causal_author_stream(
535 format!("membership_author_stream/{author_pubkey}/{author_owner_grant}"),
536 reusable,
537 )
538 .await
539 }
540
541 pub async fn select_causal_author_stream(
542 &self,
543 key: String,
544 reusable: std::collections::BTreeSet<coven_protocol::membership::AuthorStreamId>,
545 ) -> Result<coven_protocol::membership::AuthorStreamId, DbError> {
546 let candidate = coven_protocol::membership::AuthorStreamId::from_digest(
547 ObjectHash::digest(self.new_store_write_id().as_str().as_bytes()),
548 );
549 self.call_store(move |session| {
550 session.select_causal_author_stream(&key, &reusable, candidate)
551 })
552 .await
553 }
554
555 pub async fn stage_membership_mutation(
556 &self,
557 plan_bytes: Vec<u8>,
558 progress_bytes: Vec<u8>,
559 pending_rotation_generation: Option<u64>,
560 ) -> Result<ObjectHash, DbError> {
561 self.call_store(move |session| {
562 session.stage_membership_mutation(
563 plan_bytes,
564 progress_bytes,
565 pending_rotation_generation,
566 )
567 })
568 .await
569 }
570
571 pub async fn stage_membership_candidate_mutation(
572 &self,
573 plan_bytes: Vec<u8>,
574 progress_bytes: Vec<u8>,
575 remote_objects: Vec<coven_protocol::remote_object::ClosedRemoteObject>,
576 pending_rotation_generation: Option<u64>,
577 ) -> Result<ObjectHash, DbError> {
578 self.call_store(move |session| {
579 session.stage_membership_candidate_mutation(
580 plan_bytes,
581 progress_bytes,
582 remote_objects,
583 pending_rotation_generation,
584 )
585 })
586 .await
587 }
588
589 pub async fn update_membership_mutation_progress(
590 &self,
591 intent_hash: ObjectHash,
592 progress_bytes: Vec<u8>,
593 ) -> Result<(), DbError> {
594 self.call_store(move |session| {
595 session.update_membership_mutation_progress(intent_hash, progress_bytes)
596 })
597 .await
598 }
599
600 pub async fn adopt_merge_membership_candidate_head(
601 &self,
602 intent_hash: ObjectHash,
603 plan_bytes: Vec<u8>,
604 previous: RemoteObjectRecord,
605 replacement: coven_protocol::remote_object::ClosedRemoteObject,
606 rotation_generation: Option<u64>,
607 ) -> Result<ObjectHash, DbError> {
608 let (
609 RemoteObjectRecord::RetainedAuthority(previous_head),
610 RemoteObjectRecord::RetainedAuthority(replacement_head),
611 ) = (&previous, replacement.record())
612 else {
613 return Err(DbError::Message(
614 "Merge membership candidate head adoption received a non-authority object"
615 .to_string(),
616 ));
617 };
618 let (
619 coven_protocol::remote_object::RetainedAuthorityObjectDomain::DeviceHead {
620 reference: previous_ref,
621 ..
622 },
623 coven_protocol::remote_object::RetainedAuthorityObjectDomain::DeviceHead {
624 reference: replacement_ref,
625 ..
626 },
627 ) = (
628 &previous_head.identity.domain,
629 &replacement_head.identity.domain,
630 )
631 else {
632 return Err(DbError::Message(
633 "Merge membership candidate head adoption received another authority domain"
634 .to_string(),
635 ));
636 };
637 if previous_ref.object.slot() != replacement_ref.object.slot()
638 || previous_ref == replacement_ref
639 {
640 return Err(DbError::Message(
641 "adopted Merge membership head does not replace the same exact slot".to_string(),
642 ));
643 }
644 let replacement = replacement
645 .map_record(|mut record| {
646 record.mark_uploaded_verified()?;
647 Ok(record)
648 })
649 .map_err(|error| {
650 DbError::context("mark adopted Merge membership head uploaded", error)
651 })?;
652 let replacement_hash = ObjectHash::digest(&plan_bytes);
653 self.call_store(move |session| {
654 session.adopt_merge_membership_candidate_head(
655 intent_hash,
656 plan_bytes,
657 previous,
658 replacement,
659 rotation_generation,
660 replacement_hash,
661 )
662 })
663 .await
664 }
665
666 pub async fn begin_membership_candidate_nonactivation(
667 &self,
668 intent_hash: ObjectHash,
669 candidate: StoreBatchCommitRef,
670 candidate_objects: Vec<ExactObjectRef>,
671 retained_authorities: Vec<ExactObjectRef>,
672 progress_bytes: Vec<u8>,
673 nonactivation: coven_protocol::remote_object::VerifiedCandidateNonactivation,
674 ) -> Result<Vec<CandidateCleanupObject>, DbError> {
675 if nonactivation.candidate_reference().map_err(DbError::from)? != candidate {
676 return Err(DbError::Message(
677 "verified nonactivation names another membership candidate".to_string(),
678 ));
679 }
680 let nonactivation = nonactivation.into_durable();
681 self.call_store(move |session| {
682 session.begin_membership_candidate_nonactivation(
683 intent_hash,
684 candidate,
685 candidate_objects,
686 retained_authorities,
687 progress_bytes,
688 nonactivation,
689 )
690 })
691 .await
692 }
693
694 pub async fn complete_nonactivating_membership_candidate_mutation(
695 &self,
696 intent_hash: ObjectHash,
697 candidate: StoreBatchCommitRef,
698 candidate_objects: Vec<ExactObjectRef>,
699 retained_authorities: Vec<ExactObjectRef>,
700 rotation_generation: Option<u64>,
701 ) -> Result<(), DbError> {
702 self.call_store(move |session| {
703 session.complete_nonactivating_membership_candidate_mutation(
704 intent_hash,
705 candidate,
706 candidate_objects,
707 retained_authorities,
708 rotation_generation,
709 )
710 })
711 .await
712 }
713
714 pub async fn membership_candidate_cleanup_targets(
715 &self,
716 intent_hash: ObjectHash,
717 candidate: StoreBatchCommitRef,
718 objects: Vec<ExactObjectRef>,
719 ) -> Result<Vec<CandidateCleanupObject>, DbError> {
720 self.call_store(move |session| {
721 session.membership_candidate_cleanup_targets(intent_hash, &candidate, &objects)
722 })
723 .await
724 }
725
726 pub async fn record_direct_revoke_activation(
727 &self,
728 intent_hash: ObjectHash,
729 progress_bytes: Vec<u8>,
730 generation: u64,
731 ) -> Result<(), DbError> {
732 self.call_store(move |session| {
733 session.record_direct_revoke_activation(intent_hash, progress_bytes, generation)
734 })
735 .await
736 }
737
738 pub async fn complete_membership_mutation(
739 &self,
740 intent_hash: ObjectHash,
741 ) -> Result<(), DbError> {
742 self.call_store(move |session| session.complete_membership_mutation(intent_hash))
743 .await
744 }
745}
746
747#[cfg(test)]
748mod tests {
749 use super::*;
750
751 #[tokio::test]
754 async fn a_reusable_selected_stream_is_returned_again() {
755 let fixture_store_dir = crate::synthetic_store::test_store_dir();
756 let fixture = crate::synthetic_store::open_test_db(fixture_store_dir.clone());
757 let database = StoreDatabase::new(&fixture);
758 let key = "circle_roster_author_stream/reselect".to_string();
759
760 let selected = database
761 .select_causal_author_stream(key.clone(), std::collections::BTreeSet::new())
762 .await
763 .expect("mint an author stream for a key holding none");
764
765 assert_eq!(
766 database
767 .select_causal_author_stream(key, std::collections::BTreeSet::from([selected]))
768 .await
769 .expect("reselect the durable author stream"),
770 selected
771 );
772 }
773}