1use petgraph::graphmap::DiGraphMap;
5use petgraph::visit::{IntoNodeIdentifiers, Topo};
6use std::collections::{HashMap, HashSet};
7use std::{fmt::Debug, marker::PhantomData};
8
9use crate::graph::{concurrent_bubbles, split_bubble};
10use crate::group::crdt::{GroupCrdtInnerError, apply_remove_unsafe};
11use crate::group::{AuthorityGraphs, GroupAction, GroupCrdtInnerState, GroupMember, apply_action};
12use crate::traits::{Conditions, IdentityHandle, Operation, OperationId, Resolver};
13
14#[derive(Clone, Debug, Default)]
49pub struct StrongRemove<ID, OP, M, C> {
50 _phantom: PhantomData<(ID, OP, M, C)>,
51}
52
53impl<ID, OP, M, C> Resolver<ID, OP, M, C> for StrongRemove<ID, OP, M, C>
54where
55 ID: IdentityHandle,
56 OP: OperationId,
57 M: Clone + Operation<ID, OP, C>,
58 C: Conditions,
59{
60 type State = GroupCrdtInnerState<ID, OP, M, C>;
61 type Error = GroupCrdtInnerError<OP>;
62
63 fn rebuild_required(y: &Self::State, operation: &M) -> Result<bool, Self::Error> {
65 let dependencies = operation.dependencies().into_iter().collect();
66 Ok(y.heads() != dependencies)
67 }
68
69 fn process(mut y: Self::State) -> Result<Self::State, Self::Error> {
72 y.ignore = HashSet::default();
74 y.mutual_removes = HashSet::default();
75
76 let bubbles = concurrent_bubbles(&y.graph);
81 let mut processed_bubbles = Vec::new();
82
83 let mut topo = Topo::new(&y.graph);
84 while let Some(operation_id) = topo.next(&y.graph) {
85 let bubble = bubbles.iter().find(|bubble| bubble.contains(&operation_id));
86
87 let Some(bubble) = bubble else {
88 y = StrongRemove::apply_operation(y, operation_id);
89 continue;
90 };
91
92 if processed_bubbles.iter().any(|b| b == &bubble) {
93 continue;
94 };
95
96 y = StrongRemove::process_bubble(y, bubble);
97 processed_bubbles.push(bubble);
98 }
99
100 Ok(y)
101 }
102}
103
104impl<ID, OP, M, C> StrongRemove<ID, OP, M, C>
105where
106 ID: IdentityHandle,
107 OP: OperationId,
108 M: Clone + Operation<ID, OP, C>,
109 C: Conditions,
110{
111 fn process_bubble(
117 mut y: GroupCrdtInnerState<ID, OP, M, C>,
118 bubble: &HashSet<OP>,
119 ) -> GroupCrdtInnerState<ID, OP, M, C> {
120 let bubble_graph = {
122 let non_bubble_operations: Vec<_> = y
123 .graph
124 .node_identifiers()
125 .filter(|n| !bubble.contains(n))
126 .collect();
127
128 let mut bubble_graph = y.graph.clone();
129 for node_id in non_bubble_operations {
130 bubble_graph.remove_node(node_id);
131 }
132 bubble_graph
133 };
134
135 let mut authority_graphs = Self::build_authority_graphs(&y.operations, &bubble_graph);
137 y = Self::compute_filter(y, bubble, &mut authority_graphs);
138
139 let mut topo = Topo::new(&bubble_graph);
146 while let Some(operation_id) = topo.next(&bubble_graph) {
147 y = Self::apply_operation(y, operation_id);
148 }
149 y
150 }
151
152 fn is_mutual_remove(operation: &M, authority_graphs: &mut AuthorityGraphs<ID, OP>) -> bool {
154 let removed = removed_or_demoted_manager(operation);
155 let added = added_or_promoted_manager(operation);
156 if removed.is_none() && added.is_none() {
157 return false;
158 }
159
160 let group_id = operation.group_id();
161 authority_graphs.is_cycle(&group_id, &operation.id())
162 }
163
164 fn is_readd(group_id: ID, removed: ID, operation: &M) -> bool {
166 if group_id != operation.group_id() {
167 return false;
168 }
169
170 let GroupAction::Add { member: added, .. } = &operation.action() else {
171 return false;
172 };
173
174 added.id() == removed
175 }
176
177 fn is_removed(group_id: ID, removed: ID, operation: &M) -> bool {
179 if group_id != operation.group_id() {
180 return false;
181 }
182
183 operation.author() == removed
184 }
185
186 fn compute_filter(
188 mut y: GroupCrdtInnerState<ID, OP, M, C>,
189 bubble: &HashSet<OP>,
190 authority_graphs: &mut AuthorityGraphs<ID, OP>,
191 ) -> GroupCrdtInnerState<ID, OP, M, C> {
192 let mut filter = HashSet::new();
194
195 let mut mutual_removes = HashSet::new();
198
199 for operation_id in bubble {
201 let operation = y
202 .operations
203 .get(operation_id)
204 .expect("all operations present in map");
205
206 let Some(removed) = removed_or_demoted_manager(operation) else {
208 continue;
209 };
210
211 if Self::is_mutual_remove(operation, authority_graphs) {
213 mutual_removes.insert(*operation_id);
214 }
215
216 let group_id = operation.group_id();
218 let (mut concurrent, ..) = split_bubble(&y.graph, bubble, *operation_id);
219 concurrent.retain(|id| {
220 let concurrent_operation =
221 y.operations.get(id).expect("all operations present in map");
222
223 let is_readd = Self::is_readd(group_id, removed, concurrent_operation);
225 let is_removed = Self::is_removed(group_id, removed, concurrent_operation);
227 is_removed || is_readd
228 });
229
230 filter.extend(concurrent.iter());
231 }
232
233 y.ignore = filter;
234 y.mutual_removes = mutual_removes;
235 y
236 }
237
238 fn build_authority_graphs(
240 operations: &HashMap<OP, M>,
241 bubble_graph: &DiGraphMap<OP, ()>,
242 ) -> AuthorityGraphs<ID, OP> {
243 let mut authority_graphs = AuthorityGraphs::new(bubble_graph.clone());
244
245 for id in bubble_graph.nodes() {
247 let operation = operations.get(&id).expect("all operations present in map");
248 let author = operation.author();
249 let group_id = operation.group_id();
250
251 if let Some(removed) = removed_or_demoted_manager(operation) {
253 authority_graphs.add_removal(group_id, author, removed, id);
254 };
255
256 if let Some(added) = added_or_promoted_manager(operation) {
258 authority_graphs.add_delegation(group_id, author, added, id);
259 };
260 }
261
262 authority_graphs
263 }
264
265 fn apply_operation(
267 mut y: GroupCrdtInnerState<ID, OP, M, C>,
268 operation_id: OP,
269 ) -> GroupCrdtInnerState<ID, OP, M, C> {
270 let operation = y
271 .operations
272 .get(&operation_id)
273 .expect("all processed operations exist");
274
275 let dependencies = HashSet::from_iter(operation.dependencies().clone());
276
277 let mut groups_y = y
278 .state_at(&dependencies)
279 .expect("all state objects to exist");
280
281 groups_y = if !y.mutual_removes.contains(&operation_id) {
282 apply_action(
283 groups_y,
284 operation.group_id(),
285 operation.id(),
286 operation.author(),
287 &operation.action(),
288 &y.ignore,
289 )
290 .state()
291 .to_owned()
292 } else {
293 let Some(removed) = removed_or_demoted_manager(operation) else {
294 unreachable!();
295 };
296
297 apply_remove_unsafe(
300 groups_y,
301 operation.group_id(),
302 GroupMember::Individual(removed),
303 )
304 };
305 y.states.insert(operation.id(), groups_y);
306 y
307 }
308}
309
310fn removed_or_demoted_manager<ID, OP, M, C>(operation: &M) -> Option<ID>
313where
314 ID: IdentityHandle,
315 OP: OperationId,
316 M: Clone + Operation<ID, OP, C>,
317 C: Conditions,
318{
319 let action = operation.action();
320 if let GroupAction::Remove { member: removed } = action {
321 return Some(removed.id());
322 }
323
324 if let GroupAction::Demote {
325 member: demoted,
326 access,
327 } = action
328 && !access.is_manage()
329 {
330 return Some(demoted.id());
331 }
332
333 None
334}
335
336fn added_or_promoted_manager<ID, OP, M, C>(operation: &M) -> Option<ID>
339where
340 ID: IdentityHandle,
341 OP: OperationId,
342 M: Clone + Operation<ID, OP, C>,
343 C: Conditions,
344{
345 let action = operation.action();
346 if let GroupAction::Add {
347 member: added,
348 access,
349 } = &action
350 && access.is_manage()
351 {
352 return Some(added.id());
353 }
354
355 if let GroupAction::Promote {
356 member: promoted,
357 access,
358 } = action
359 && access.is_manage()
360 {
361 return Some(promoted.id());
362 }
363
364 None
365}
366
367#[cfg(test)]
368mod tests {
369 use crate::Access;
370 use crate::group::GroupMember;
371 use crate::test_utils::{
372 MemberId, TestGroupState, add_member, assert_members, create_group, demote_member,
373 remove_member, sync,
374 };
375
376 use super::*;
377
378 const G0: char = '0';
379 const G1: char = '1';
380
381 const ALICE: char = 'A';
382 const BOB: char = 'B';
383 const CLAIRE: char = 'C';
384 const DAVE: char = 'D';
385 const EVE: char = 'E';
386 const FRANK: char = 'F';
387 const GRACE: char = 'G';
388
389 #[test]
390 fn mutual_removal_filter() {
391 let y = TestGroupState::new();
402
403 let op0 = create_group(
405 ALICE,
406 0,
407 G1,
408 vec![
409 (GroupMember::Individual(ALICE), Access::manage()),
410 (GroupMember::Individual(BOB), Access::manage()),
411 (GroupMember::Individual(CLAIRE), Access::manage()),
412 ],
413 vec![],
414 );
415
416 let op1 = remove_member(ALICE, 1, G1, GroupMember::Individual(BOB), vec![op0.id()]);
418
419 let op2 = remove_member(BOB, 2, G1, GroupMember::Individual(ALICE), vec![op0.id()]);
421
422 let expected = vec![(CLAIRE, Access::manage())];
423 let y_i = sync(y, &[op0, op1, op2]);
424 assert_members(&y_i, G1, &expected);
425
426 assert_eq!(y_i.inner.ignore.len(), 2);
427 assert_eq!(y_i.inner.mutual_removes.len(), 2);
428 }
429
430 #[test]
431 fn mutual_remove_cycles_detected() {
432 let y = TestGroupState::new();
445
446 let op0 = create_group(
448 ALICE,
449 0,
450 G1,
451 vec![
452 (GroupMember::Individual(ALICE), Access::manage()),
453 (GroupMember::Individual(BOB), Access::manage()),
454 (GroupMember::Individual(CLAIRE), Access::manage()),
455 ],
456 vec![],
457 );
458
459 let op1 = remove_member(ALICE, 1, G1, GroupMember::Individual(BOB), vec![op0.id()]);
461
462 let op2 = remove_member(BOB, 2, G1, GroupMember::Individual(CLAIRE), vec![op0.id()]);
464
465 let op3 = remove_member(
467 CLAIRE,
468 3,
469 G1,
470 GroupMember::Individual(ALICE),
471 vec![op0.id()],
472 );
473
474 let expected: Vec<(MemberId, Access)> = vec![];
475
476 let y_i = sync(y, &[op0, op1, op2, op3]);
477 assert_members(&y_i, G1, &expected);
478
479 assert_eq!(y_i.inner.ignore.len(), 3);
480 assert_eq!(y_i.inner.mutual_removes.len(), 3);
481 }
482
483 #[test]
484 fn mutual_remove_cycle_with_delegation() {
485 let y = TestGroupState::new();
501
502 let op0 = create_group(
504 ALICE,
505 0,
506 G1,
507 vec![
508 (GroupMember::Individual(ALICE), Access::manage()),
509 (GroupMember::Individual(BOB), Access::manage()),
510 (GroupMember::Individual(CLAIRE), Access::manage()),
511 ],
512 vec![],
513 );
514
515 let op1 = remove_member(ALICE, 1, G1, GroupMember::Individual(BOB), vec![op0.id()]);
517
518 let op2 = remove_member(BOB, 2, G1, GroupMember::Individual(CLAIRE), vec![op0.id()]);
520
521 let op3 = add_member(
523 CLAIRE,
524 3,
525 G1,
526 GroupMember::Individual(DAVE),
527 Access::manage(),
528 vec![op0.id()],
529 );
530
531 let op4 = remove_member(DAVE, 4, G1, GroupMember::Individual(ALICE), vec![op3.id()]);
533
534 let expected: Vec<(MemberId, Access)> = vec![]; let y_i = sync(y, &[op0, op1, op2, op3, op4]);
537 assert_members(&y_i, G1, &expected);
538 assert_eq!(y_i.inner.ignore.len(), 3);
539 assert_eq!(y_i.inner.mutual_removes.len(), 3);
540 }
541
542 #[test]
543 fn demote_remove_filter() {
544 let y = TestGroupState::new();
555
556 let op0 = create_group(
558 ALICE,
559 0,
560 G1,
561 vec![
562 (GroupMember::Individual(ALICE), Access::manage()),
563 (GroupMember::Individual(BOB), Access::manage()),
564 (GroupMember::Individual(CLAIRE), Access::manage()),
565 ],
566 vec![],
567 );
568
569 let op1 = demote_member(
571 ALICE,
572 1,
573 G1,
574 GroupMember::Individual(BOB),
575 Access::write(),
576 vec![op0.id()],
577 );
578
579 let op2 = remove_member(BOB, 2, G1, GroupMember::Individual(CLAIRE), vec![op0.id()]);
581
582 let expected = vec![
583 (ALICE, Access::manage()),
584 (BOB, Access::write()),
585 (CLAIRE, Access::manage()),
586 ];
587
588 let y_final = sync(y, &[op0, op1, op2]);
589 assert_members(&y_final, G1, &expected);
590 }
591
592 #[test]
593 fn demote_add_filter() {
594 let y = TestGroupState::new();
605
606 let op0 = create_group(
608 ALICE,
609 0,
610 G1,
611 vec![
612 (GroupMember::Individual(ALICE), Access::manage()),
613 (GroupMember::Individual(BOB), Access::manage()),
614 (GroupMember::Individual(CLAIRE), Access::manage()),
615 ],
616 vec![],
617 );
618
619 let op1 = demote_member(
621 ALICE,
622 1,
623 G1,
624 GroupMember::Individual(BOB),
625 Access::write(),
626 vec![op0.id()],
627 );
628
629 let op2 = add_member(
631 BOB,
632 2,
633 G1,
634 GroupMember::Individual(DAVE),
635 Access::read(),
636 vec![op0.id()],
637 );
638
639 let expected = vec![
640 (ALICE, Access::manage()),
641 (BOB, Access::write()),
642 (CLAIRE, Access::manage()),
643 ];
644
645 let y_i = sync(y, &[op0, op1, op2]);
646 assert_members(&y_i, G1, &expected);
647 }
648
649 #[test]
650 fn remove_dependencies_filter() {
651 let y = TestGroupState::new();
666
667 let op0 = create_group(
669 ALICE,
670 0,
671 G1,
672 vec![
673 (GroupMember::Individual(ALICE), Access::manage()),
674 (GroupMember::Individual(BOB), Access::manage()),
675 ],
676 vec![],
677 );
678
679 let op1 = remove_member(ALICE, 1, G1, GroupMember::Individual(BOB), vec![op0.id()]);
681
682 let op2 = add_member(
684 BOB,
685 2,
686 G1,
687 GroupMember::Individual(CLAIRE),
688 Access::manage(),
689 vec![op0.id()],
690 );
691
692 let op3 = add_member(
694 CLAIRE,
695 3,
696 G1,
697 GroupMember::Individual(DAVE),
698 Access::read(),
699 vec![op2.id()],
700 );
701
702 let expected_members = vec![(ALICE, Access::manage())];
704 let y_i = sync(y, &[op0, op1, op2, op3]);
705 assert_members(&y_i, G1, &expected_members);
706 }
707
708 #[test]
709 fn remove_readd_dependencies_filter() {
710 let y = TestGroupState::new();
728
729 let op0 = create_group(
731 ALICE,
732 0,
733 G1,
734 vec![
735 (GroupMember::Individual(ALICE), Access::manage()),
736 (GroupMember::Individual(BOB), Access::manage()),
737 (GroupMember::Individual(CLAIRE), Access::manage()),
738 ],
739 vec![],
740 );
741
742 let op1 = remove_member(ALICE, 1, G1, GroupMember::Individual(BOB), vec![op0.id()]);
744
745 let op2 = add_member(
747 ALICE,
748 2,
749 G1,
750 GroupMember::Individual(BOB),
751 Access::manage(),
752 vec![op1.id()],
753 );
754
755 let op3 = add_member(
757 BOB,
758 3,
759 G1,
760 GroupMember::Individual(DAVE),
761 Access::read(),
762 vec![op0.id()],
763 );
764
765 let op4 = add_member(
767 BOB,
768 4,
769 G1,
770 GroupMember::Individual(EVE),
771 Access::read(),
772 vec![op2.id(), op3.id()],
773 );
774
775 let expected = vec![
776 (ALICE, Access::manage()),
777 (BOB, Access::manage()),
778 (CLAIRE, Access::manage()),
779 (EVE, Access::read()),
780 ];
781
782 let y_final = sync(y, &[op0, op1, op2, op3, op4]);
783 assert_members(&y_final, G1, &expected);
784 }
785
786 #[test]
787 fn two_bubbles() {
788 let y = TestGroupState::new();
812
813 let op0 = create_group(
815 ALICE,
816 0,
817 G0,
818 vec![
819 (GroupMember::Individual(ALICE), Access::manage()),
820 (GroupMember::Individual(BOB), Access::manage()),
821 ],
822 vec![],
823 );
824
825 let op1 = remove_member(ALICE, 1, G0, GroupMember::Individual(BOB), vec![op0.id()]);
827
828 let op2 = add_member(
830 BOB,
831 2,
832 G0,
833 GroupMember::Individual(CLAIRE),
834 Access::read(),
835 vec![op0.id()],
836 );
837
838 let op3 = add_member(
840 ALICE,
841 3,
842 G0,
843 GroupMember::Individual(DAVE),
844 Access::manage(),
845 vec![op1.id(), op2.id()],
846 );
847
848 let op4 = add_member(
850 DAVE,
851 4,
852 G0,
853 GroupMember::Individual(EVE),
854 Access::read(),
855 vec![op3.id()],
856 );
857
858 let op5 = add_member(
860 ALICE,
861 5,
862 G0,
863 GroupMember::Individual(FRANK),
864 Access::manage(),
865 vec![op4.id()],
866 );
867
868 let op6 = add_member(
870 FRANK,
871 6,
872 G0,
873 GroupMember::Individual(GRACE),
874 Access::read(),
875 vec![op5.id()],
876 );
877
878 let op7 = remove_member(DAVE, 7, G0, GroupMember::Individual(ALICE), vec![op4.id()]);
880
881 let expected_members = vec![(DAVE, Access::manage()), (EVE, Access::read())];
882 let y_i = sync(y, &[op0, op1, op2, op3, op4, op5, op6, op7]);
883 assert_members(&y_i, G0, &expected_members);
884 }
885
886 #[test]
887 fn concurrent_readds_filtered() {
888 let y = TestGroupState::new();
903
904 let op0 = create_group(
906 ALICE,
907 0,
908 G1,
909 vec![
910 (GroupMember::Individual(ALICE), Access::manage()),
911 (GroupMember::Individual(BOB), Access::manage()),
912 (GroupMember::Individual(CLAIRE), Access::manage()),
913 ],
914 vec![],
915 );
916
917 let op1 = remove_member(ALICE, 1, G1, GroupMember::Individual(BOB), vec![op0.id()]);
919
920 let op2 = add_member(
922 ALICE,
923 2,
924 G1,
925 GroupMember::Individual(BOB),
926 Access::manage(),
927 vec![op1.id()],
928 );
929
930 let op3 = remove_member(CLAIRE, 3, G1, GroupMember::Individual(BOB), vec![op0.id()]);
932
933 let expected = vec![(ALICE, Access::manage()), (CLAIRE, Access::manage())];
934
935 let y_final = sync(y, &[op0, op1, op2, op3]);
936 assert_members(&y_final, G1, &expected);
937 }
938
939 #[test]
940 fn filter_only_concurrent_operations() {
941 let y = TestGroupState::new();
942
943 let op0 = create_group(
945 ALICE,
946 0,
947 G1,
948 vec![
949 (GroupMember::Individual(ALICE), Access::manage()),
950 (GroupMember::Individual(BOB), Access::manage()),
951 ],
952 vec![],
953 );
954
955 let op1 = add_member(
957 BOB,
958 1,
959 G1,
960 GroupMember::Individual(CLAIRE),
961 Access::read(),
962 vec![op0.id()],
963 );
964
965 let op2 = demote_member(
967 ALICE,
968 2,
969 G1,
970 GroupMember::Individual(BOB),
971 Access::write(),
972 vec![op1.id()],
973 );
974
975 let op3 = add_member(
977 BOB,
978 3,
979 G1,
980 GroupMember::Individual(DAVE),
981 Access::read(),
982 vec![op1.id()],
983 );
984
985 let expected = vec![
986 (ALICE, Access::manage()),
987 (BOB, Access::write()),
988 (CLAIRE, Access::read()),
989 ];
990 let y_i = sync(y, &[op0, op1, op2, op3]);
991 assert_members(&y_i, G1, &expected);
992 }
993}