1use std::collections::{BTreeMap, BTreeSet};
47
48use super::identity::NodeIdentity;
49use super::slot::hash_shard_key_to_range_key;
50
51#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
59pub struct CollectionId(String);
60
61#[derive(Debug, Clone, PartialEq, Eq)]
62pub struct CollectionIdError;
63
64impl std::fmt::Display for CollectionIdError {
65 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
66 write!(f, "collection id is empty")
67 }
68}
69
70impl std::error::Error for CollectionIdError {}
71
72impl CollectionId {
73 pub fn new(value: impl AsRef<str>) -> Result<Self, CollectionIdError> {
75 let value = value.as_ref().trim();
76 if value.is_empty() {
77 return Err(CollectionIdError);
78 }
79 Ok(Self(value.to_string()))
80 }
81
82 pub fn as_str(&self) -> &str {
83 &self.0
84 }
85}
86
87impl std::fmt::Display for CollectionId {
88 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
89 f.write_str(&self.0)
90 }
91}
92
93#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
100pub struct CollectionGroupId(String);
101
102#[derive(Debug, Clone, PartialEq, Eq)]
103pub struct CollectionGroupIdError;
104
105impl std::fmt::Display for CollectionGroupIdError {
106 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
107 write!(f, "collection group id is empty")
108 }
109}
110
111impl std::error::Error for CollectionGroupIdError {}
112
113impl CollectionGroupId {
114 pub fn new(value: impl AsRef<str>) -> Result<Self, CollectionGroupIdError> {
116 let value = value.as_ref().trim();
117 if value.is_empty() {
118 return Err(CollectionGroupIdError);
119 }
120 Ok(Self(value.to_string()))
121 }
122
123 pub fn for_collection(collection: &CollectionId) -> Self {
124 Self(collection.as_str().to_string())
125 }
126
127 pub fn as_str(&self) -> &str {
128 &self.0
129 }
130}
131
132impl std::fmt::Display for CollectionGroupId {
133 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
134 f.write_str(&self.0)
135 }
136}
137
138#[derive(Debug, Clone, PartialEq, Eq)]
141pub struct CollectionGroupAuthority {
142 group: CollectionGroupId,
143 authority: NodeIdentity,
144 collections: BTreeSet<CollectionId>,
145}
146
147impl CollectionGroupAuthority {
148 pub fn new(
150 group: CollectionGroupId,
151 authority: NodeIdentity,
152 collections: impl IntoIterator<Item = CollectionId>,
153 ) -> Result<Self, PlacementAuthorityError> {
154 let collections = collections.into_iter().collect::<BTreeSet<_>>();
155 if collections.is_empty() {
156 return Err(PlacementAuthorityError::EmptyCatalogSlice { group });
157 }
158 Ok(Self {
159 group,
160 authority,
161 collections,
162 })
163 }
164
165 pub fn group(&self) -> &CollectionGroupId {
166 &self.group
167 }
168
169 pub fn authority(&self) -> &NodeIdentity {
170 &self.authority
171 }
172
173 pub fn collections(&self) -> impl Iterator<Item = &CollectionId> {
174 self.collections.iter()
175 }
176
177 fn overlaps(&self, other: &CollectionGroupAuthority) -> Option<CollectionId> {
178 self.collections
179 .intersection(&other.collections)
180 .next()
181 .cloned()
182 }
183}
184
185#[derive(Debug, Clone, PartialEq, Eq)]
187pub enum PlacementAuthorityError {
188 EmptyCatalogSlice { group: CollectionGroupId },
191 DuplicateCollectionGroup { group: CollectionGroupId },
193 OverlappingCatalogSlice {
196 collection: CollectionId,
197 existing_group: CollectionGroupId,
198 attempted_group: CollectionGroupId,
199 },
200}
201
202impl std::fmt::Display for PlacementAuthorityError {
203 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
204 match self {
205 Self::EmptyCatalogSlice { group } => {
206 write!(f, "collection group {group} has an empty catalog slice")
207 }
208 Self::DuplicateCollectionGroup { group } => {
209 write!(
210 f,
211 "collection group {group} already has a placement authority"
212 )
213 }
214 Self::OverlappingCatalogSlice {
215 collection,
216 existing_group,
217 attempted_group,
218 } => write!(
219 f,
220 "collection {collection} is already in group {existing_group}, so group {attempted_group} would overlap its catalog slice"
221 ),
222 }
223 }
224}
225
226impl std::error::Error for PlacementAuthorityError {}
227
228#[derive(Debug, Clone, Default)]
230pub struct PlacementAuthorityCatalog {
231 assignments: BTreeMap<CollectionGroupId, CollectionGroupAuthority>,
232}
233
234impl PlacementAuthorityCatalog {
235 pub fn new() -> Self {
237 Self::default()
238 }
239
240 pub fn assign(
242 &mut self,
243 assignment: CollectionGroupAuthority,
244 ) -> Result<(), PlacementAuthorityError> {
245 if self.assignments.contains_key(assignment.group()) {
246 return Err(PlacementAuthorityError::DuplicateCollectionGroup {
247 group: assignment.group().clone(),
248 });
249 }
250 for existing in self.assignments.values() {
251 if let Some(collection) = existing.overlaps(&assignment) {
252 return Err(PlacementAuthorityError::OverlappingCatalogSlice {
253 collection,
254 existing_group: existing.group().clone(),
255 attempted_group: assignment.group().clone(),
256 });
257 }
258 }
259 self.assignments
260 .insert(assignment.group().clone(), assignment);
261 Ok(())
262 }
263
264 pub fn authority_for_group(
265 &self,
266 group: &CollectionGroupId,
267 ) -> Option<&CollectionGroupAuthority> {
268 self.assignments.get(group)
269 }
270
271 pub fn authority_for_collection(
272 &self,
273 collection: &CollectionId,
274 ) -> Option<&CollectionGroupAuthority> {
275 self.assignments
276 .values()
277 .find(|assignment| assignment.collections.contains(collection))
278 }
279}
280
281#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
283pub struct PlacementAuthorityId(String);
284
285#[derive(Debug, Clone, PartialEq, Eq)]
286pub struct PlacementAuthorityIdError;
287
288impl std::fmt::Display for PlacementAuthorityIdError {
289 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
290 write!(f, "placement authority id is empty")
291 }
292}
293
294impl std::error::Error for PlacementAuthorityIdError {}
295
296impl PlacementAuthorityId {
297 pub fn new(value: impl AsRef<str>) -> Result<Self, PlacementAuthorityIdError> {
298 let value = value.as_ref().trim();
299 if value.is_empty() {
300 return Err(PlacementAuthorityIdError);
301 }
302 Ok(Self(value.to_string()))
303 }
304
305 pub fn default_global() -> Self {
306 Self("global-placement-authority".to_string())
307 }
308
309 pub fn as_str(&self) -> &str {
310 &self.0
311 }
312}
313
314impl std::fmt::Display for PlacementAuthorityId {
315 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
316 f.write_str(&self.0)
317 }
318}
319
320#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
328pub struct RangeId(u64);
329
330impl RangeId {
331 pub fn new(value: u64) -> Self {
332 Self(value)
333 }
334
335 pub fn value(self) -> u64 {
336 self.0
337 }
338}
339
340impl std::fmt::Display for RangeId {
341 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
342 write!(f, "{}", self.0)
343 }
344}
345
346#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
355pub enum ShardKeyMode {
356 #[default]
358 Hash,
359 Ordered,
362}
363
364#[derive(Debug, Clone, PartialEq, Eq)]
372pub enum RangeBound {
373 Min,
375 Key(Vec<u8>),
377 Max,
379}
380
381impl RangeBound {
382 pub fn key(bytes: impl Into<Vec<u8>>) -> Self {
384 RangeBound::Key(bytes.into())
385 }
386
387 fn position(&self) -> Position<'_> {
391 match self {
392 RangeBound::Min => Position::Min,
393 RangeBound::Key(k) => Position::Key(k),
394 RangeBound::Max => Position::Max,
395 }
396 }
397}
398
399#[derive(PartialEq, Eq, PartialOrd, Ord)]
400enum Position<'a> {
401 Min,
402 Key(&'a [u8]),
403 Max,
404}
405
406#[derive(Debug, Clone, PartialEq, Eq)]
413pub struct RangeBounds {
414 lower: RangeBound,
415 upper: RangeBound,
416}
417
418#[derive(Debug, Clone, PartialEq, Eq)]
419pub struct RangeBoundsError;
420
421impl std::fmt::Display for RangeBoundsError {
422 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
423 write!(
424 f,
425 "range lower bound must be strictly below the upper bound"
426 )
427 }
428}
429
430impl std::error::Error for RangeBoundsError {}
431
432impl RangeBounds {
433 pub fn new(lower: RangeBound, upper: RangeBound) -> Result<Self, RangeBoundsError> {
436 if lower.position() >= upper.position() {
437 return Err(RangeBoundsError);
438 }
439 Ok(Self { lower, upper })
440 }
441
442 pub fn full() -> Self {
444 Self {
445 lower: RangeBound::Min,
446 upper: RangeBound::Max,
447 }
448 }
449
450 pub fn lower(&self) -> &RangeBound {
451 &self.lower
452 }
453
454 pub fn upper(&self) -> &RangeBound {
455 &self.upper
456 }
457
458 pub fn contains(&self, key: &[u8]) -> bool {
461 let key = Position::Key(key);
462 self.lower.position() <= key && key < self.upper.position()
463 }
464
465 pub fn overlaps(&self, other: &RangeBounds) -> bool {
468 self.lower.position() < other.upper.position()
469 && other.lower.position() < self.upper.position()
470 }
471
472 pub fn split_at(&self, at: &[u8]) -> Result<(RangeBounds, RangeBounds), RangeBoundsError> {
481 let at_pos = Position::Key(at);
482 if at_pos <= self.lower.position() || at_pos >= self.upper.position() {
483 return Err(RangeBoundsError);
484 }
485 let lower = RangeBounds {
486 lower: self.lower.clone(),
487 upper: RangeBound::key(at.to_vec()),
488 };
489 let upper = RangeBounds {
490 lower: RangeBound::key(at.to_vec()),
491 upper: self.upper.clone(),
492 };
493 Ok((lower, upper))
494 }
495}
496
497#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
504pub struct CatalogVersion(u64);
505
506impl CatalogVersion {
507 pub fn initial() -> Self {
509 Self(1)
510 }
511
512 pub fn value(self) -> u64 {
513 self.0
514 }
515
516 fn next(self) -> Self {
517 Self(self.0 + 1)
518 }
519}
520
521impl std::fmt::Display for CatalogVersion {
522 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
523 write!(f, "{}", self.0)
524 }
525}
526
527#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
535pub struct OwnershipEpoch(u64);
536
537impl OwnershipEpoch {
538 pub fn initial() -> Self {
540 Self(1)
541 }
542
543 pub fn value(self) -> u64 {
544 self.0
545 }
546
547 fn next(self) -> Self {
548 Self(self.0 + 1)
549 }
550}
551
552impl std::fmt::Display for OwnershipEpoch {
553 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
554 write!(f, "{}", self.0)
555 }
556}
557
558#[derive(Debug, Clone, Default, PartialEq, Eq)]
564pub struct PlacementMetadata {
565 replication_factor: usize,
566 attributes: BTreeMap<String, String>,
567}
568
569impl PlacementMetadata {
570 pub fn with_replication_factor(replication_factor: usize) -> Self {
572 Self {
573 replication_factor,
574 attributes: BTreeMap::new(),
575 }
576 }
577
578 pub fn with_attribute(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
580 self.attributes.insert(key.into(), value.into());
581 self
582 }
583
584 pub fn replication_factor(&self) -> usize {
585 self.replication_factor
586 }
587
588 pub fn attribute(&self, key: &str) -> Option<&str> {
589 self.attributes.get(key).map(String::as_str)
590 }
591}
592
593#[derive(Debug, Clone, PartialEq, Eq)]
604pub struct RangeOwnership {
605 collection: CollectionId,
606 collection_group: CollectionGroupId,
607 placement_authority: PlacementAuthorityId,
608 range_id: RangeId,
609 shard_key_mode: ShardKeyMode,
610 bounds: RangeBounds,
611 owner: NodeIdentity,
612 replicas: Vec<NodeIdentity>,
613 compressed_archive_replicas: Vec<NodeIdentity>,
614 epoch: OwnershipEpoch,
615 version: CatalogVersion,
616 placement: PlacementMetadata,
617}
618
619#[derive(Debug, Clone, PartialEq, Eq)]
620pub struct HotMirrorCandidate {
621 node: NodeIdentity,
622 durable_lsn: u64,
623}
624
625impl HotMirrorCandidate {
626 pub fn new(node: NodeIdentity, durable_lsn: u64) -> Self {
627 Self { node, durable_lsn }
628 }
629
630 pub fn node(&self) -> &NodeIdentity {
631 &self.node
632 }
633
634 pub fn durable_lsn(&self) -> u64 {
635 self.durable_lsn
636 }
637}
638
639#[derive(Debug, Clone, PartialEq, Eq)]
640pub enum HotMirrorPromotionRefusal {
641 NotReplica {
642 candidate: NodeIdentity,
643 role: RangeRole,
644 },
645 WatermarkNotCovered {
646 candidate_lsn: u64,
647 watermark: u64,
648 },
649}
650
651#[derive(Debug, Clone, PartialEq, Eq)]
652pub struct HotMirrorPromotion {
653 previous: RangeOwnership,
654 promoted: RangeOwnership,
655}
656
657impl HotMirrorPromotion {
658 pub fn previous(&self) -> &RangeOwnership {
659 &self.previous
660 }
661
662 pub fn promoted(&self) -> &RangeOwnership {
663 &self.promoted
664 }
665}
666
667impl RangeOwnership {
668 #[allow(clippy::too_many_arguments)]
671 pub fn establish(
672 collection: CollectionId,
673 range_id: RangeId,
674 shard_key_mode: ShardKeyMode,
675 bounds: RangeBounds,
676 owner: NodeIdentity,
677 replicas: impl IntoIterator<Item = NodeIdentity>,
678 placement: PlacementMetadata,
679 ) -> Self {
680 let collection_group = CollectionGroupId::for_collection(&collection);
681 Self {
682 collection,
683 collection_group,
684 placement_authority: PlacementAuthorityId::default_global(),
685 range_id,
686 shard_key_mode,
687 bounds,
688 owner,
689 replicas: replicas.into_iter().collect(),
690 compressed_archive_replicas: Vec::new(),
691 epoch: OwnershipEpoch::initial(),
692 version: CatalogVersion::initial(),
693 placement,
694 }
695 }
696
697 pub fn collection(&self) -> &CollectionId {
698 &self.collection
699 }
700
701 pub fn collection_group(&self) -> &CollectionGroupId {
702 &self.collection_group
703 }
704
705 pub fn placement_authority(&self) -> &PlacementAuthorityId {
706 &self.placement_authority
707 }
708
709 pub fn range_id(&self) -> RangeId {
710 self.range_id
711 }
712
713 pub fn shard_key_mode(&self) -> ShardKeyMode {
714 self.shard_key_mode
715 }
716
717 pub fn bounds(&self) -> &RangeBounds {
718 &self.bounds
719 }
720
721 pub fn owner(&self) -> &NodeIdentity {
722 &self.owner
723 }
724
725 pub fn replicas(&self) -> &[NodeIdentity] {
726 &self.replicas
727 }
728
729 pub fn hot_mirror_replicas(&self) -> &[NodeIdentity] {
733 &self.replicas
734 }
735
736 pub fn compressed_archive_replicas(&self) -> &[NodeIdentity] {
740 &self.compressed_archive_replicas
741 }
742
743 pub fn epoch(&self) -> OwnershipEpoch {
744 self.epoch
745 }
746
747 pub fn version(&self) -> CatalogVersion {
748 self.version
749 }
750
751 pub fn placement(&self) -> &PlacementMetadata {
752 &self.placement
753 }
754
755 fn key(&self) -> (CollectionId, RangeId) {
757 (self.collection.clone(), self.range_id)
758 }
759
760 pub fn with_collection_group(mut self, collection_group: CollectionGroupId) -> Self {
761 self.collection_group = collection_group;
762 self
763 }
764
765 pub fn with_placement_authority(mut self, placement_authority: PlacementAuthorityId) -> Self {
766 self.placement_authority = placement_authority;
767 self
768 }
769
770 pub fn transfer_to(
774 &self,
775 new_owner: NodeIdentity,
776 new_replicas: impl IntoIterator<Item = NodeIdentity>,
777 ) -> Self {
778 Self {
779 owner: new_owner,
780 replicas: new_replicas.into_iter().collect(),
781 epoch: self.epoch.next(),
782 version: self.version.next(),
783 ..self.clone()
784 }
785 }
786
787 pub fn promote_hot_mirror(
792 &self,
793 candidate: &HotMirrorCandidate,
794 commit_watermark: u64,
795 ) -> Result<HotMirrorPromotion, HotMirrorPromotionRefusal> {
796 let role = self.role_of(candidate.node());
797 if role != RangeRole::Replica {
798 return Err(HotMirrorPromotionRefusal::NotReplica {
799 candidate: candidate.node().clone(),
800 role,
801 });
802 }
803 if candidate.durable_lsn() < commit_watermark {
804 return Err(HotMirrorPromotionRefusal::WatermarkNotCovered {
805 candidate_lsn: candidate.durable_lsn(),
806 watermark: commit_watermark,
807 });
808 }
809
810 let new_replicas = std::iter::once(self.owner().clone())
811 .chain(
812 self.replicas()
813 .iter()
814 .filter(|replica| *replica != candidate.node())
815 .cloned(),
816 )
817 .collect::<Vec<_>>();
818 Ok(HotMirrorPromotion {
819 previous: self.clone(),
820 promoted: self.transfer_to(candidate.node().clone(), new_replicas),
821 })
822 }
823
824 pub fn update_replicas(&self, new_replicas: impl IntoIterator<Item = NodeIdentity>) -> Self {
827 Self {
828 replicas: new_replicas.into_iter().collect(),
829 version: self.version.next(),
830 ..self.clone()
831 }
832 }
833
834 pub fn with_compressed_archive_replicas(
837 &self,
838 archive_replicas: impl IntoIterator<Item = NodeIdentity>,
839 ) -> Self {
840 Self {
841 compressed_archive_replicas: archive_replicas.into_iter().collect(),
842 version: self.version.next(),
843 ..self.clone()
844 }
845 }
846
847 pub fn update_replica_roles(
849 &self,
850 hot_mirror_replicas: impl IntoIterator<Item = NodeIdentity>,
851 compressed_archive_replicas: impl IntoIterator<Item = NodeIdentity>,
852 ) -> Self {
853 Self {
854 replicas: hot_mirror_replicas.into_iter().collect(),
855 compressed_archive_replicas: compressed_archive_replicas.into_iter().collect(),
856 version: self.version.next(),
857 ..self.clone()
858 }
859 }
860
861 pub fn update_placement(&self, placement: PlacementMetadata) -> Self {
864 Self {
865 placement,
866 version: self.version.next(),
867 ..self.clone()
868 }
869 }
870
871 pub fn with_bounds(&self, bounds: RangeBounds) -> Self {
877 Self {
878 bounds,
879 version: self.version.next(),
880 ..self.clone()
881 }
882 }
883
884 pub fn role_of(&self, node: &NodeIdentity) -> RangeRole {
894 if self.owner == *node {
895 RangeRole::Owner
896 } else if self.replicas.iter().any(|replica| replica == node) {
897 RangeRole::Replica
898 } else if self
899 .compressed_archive_replicas
900 .iter()
901 .any(|archive_replica| archive_replica == node)
902 {
903 RangeRole::ArchiveReplica
904 } else {
905 RangeRole::NoCopy
906 }
907 }
908
909 pub fn replica_role_of(&self, node: &NodeIdentity) -> Option<ReplicaRole> {
911 if self.replicas.iter().any(|replica| replica == node) {
912 Some(ReplicaRole::HotMirror)
913 } else if self
914 .compressed_archive_replicas
915 .iter()
916 .any(|replica| replica == node)
917 {
918 Some(ReplicaRole::CompressedArchive)
919 } else {
920 None
921 }
922 }
923
924 fn validate_replica_roles(&self) -> Result<(), CatalogError> {
925 let mut seen = BTreeSet::new();
926 for replica in &self.replicas {
927 if replica == &self.owner || !seen.insert(replica) {
928 return Err(CatalogError::InvalidReplicaRoles {
929 collection: self.collection.clone(),
930 range_id: self.range_id,
931 });
932 }
933 }
934 for replica in &self.compressed_archive_replicas {
935 if replica == &self.owner || !seen.insert(replica) {
936 return Err(CatalogError::InvalidReplicaRoles {
937 collection: self.collection.clone(),
938 range_id: self.range_id,
939 });
940 }
941 }
942 Ok(())
943 }
944}
945
946#[derive(Debug, Clone, Copy, PartialEq, Eq)]
948pub enum ReplicaRole {
949 HotMirror,
952 CompressedArchive,
955}
956
957impl ReplicaRole {
958 pub fn is_promotion_candidate(self) -> bool {
959 matches!(self, ReplicaRole::HotMirror)
960 }
961
962 pub fn is_restore_only(self) -> bool {
963 matches!(self, ReplicaRole::CompressedArchive)
964 }
965}
966
967#[derive(Debug, Clone, Copy, PartialEq, Eq)]
979pub enum RangeRole {
980 Owner,
983 Replica,
987 ArchiveReplica,
990 NoCopy,
992}
993
994impl RangeRole {
995 pub fn may_write_public(self) -> bool {
998 matches!(self, RangeRole::Owner)
999 }
1000
1001 pub fn is_direct_promotion_candidate(self) -> bool {
1002 matches!(self, RangeRole::Replica)
1003 }
1004
1005 fn label(self) -> &'static str {
1006 match self {
1007 RangeRole::Owner => "owner",
1008 RangeRole::Replica => "replica",
1009 RangeRole::ArchiveReplica => "archive-replica",
1010 RangeRole::NoCopy => "no-copy",
1011 }
1012 }
1013}
1014
1015#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1017pub enum UpdateOutcome {
1018 Created,
1020 Updated,
1022}
1023
1024#[derive(Debug, Clone, PartialEq, Eq)]
1026pub enum CatalogError {
1027 StaleVersion {
1031 collection: CollectionId,
1032 range_id: RangeId,
1033 current: CatalogVersion,
1034 attempted: CatalogVersion,
1035 },
1036 ShardKeyModeMismatch {
1039 collection: CollectionId,
1040 declared: ShardKeyMode,
1041 attempted: ShardKeyMode,
1042 },
1043 OverlappingRange {
1046 collection: CollectionId,
1047 existing: RangeId,
1048 attempted: RangeId,
1049 },
1050 InvalidReplicaRoles {
1053 collection: CollectionId,
1054 range_id: RangeId,
1055 },
1056}
1057
1058impl std::fmt::Display for CatalogError {
1059 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1060 match self {
1061 Self::StaleVersion {
1062 collection,
1063 range_id,
1064 current,
1065 attempted,
1066 } => write!(
1067 f,
1068 "stale catalog update for {collection}/{range_id}: current version {current}, attempted {attempted}"
1069 ),
1070 Self::ShardKeyModeMismatch {
1071 collection,
1072 declared,
1073 attempted,
1074 } => write!(
1075 f,
1076 "collection {collection} is declared {declared:?} but range uses {attempted:?}"
1077 ),
1078 Self::OverlappingRange {
1079 collection,
1080 existing,
1081 attempted,
1082 } => write!(
1083 f,
1084 "range {attempted} overlaps existing range {existing} of collection {collection}"
1085 ),
1086 Self::InvalidReplicaRoles {
1087 collection,
1088 range_id,
1089 } => write!(
1090 f,
1091 "range {range_id} of collection {collection} has invalid replica role metadata"
1092 ),
1093 }
1094 }
1095}
1096
1097impl std::error::Error for CatalogError {}
1098
1099#[derive(Debug, Clone, PartialEq, Eq)]
1109pub enum RangeWriteReject {
1110 NoRange { collection: CollectionId },
1113 NotOwner {
1120 collection: CollectionId,
1121 range_id: RangeId,
1122 role: RangeRole,
1123 owner: NodeIdentity,
1124 },
1125 StaleEpoch {
1130 collection: CollectionId,
1131 range_id: RangeId,
1132 expected: OwnershipEpoch,
1133 current: OwnershipEpoch,
1134 },
1135}
1136
1137impl std::fmt::Display for RangeWriteReject {
1138 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1139 match self {
1140 Self::NoRange { collection } => write!(
1141 f,
1142 "no range of collection {collection} covers the routed key — re-resolve routing"
1143 ),
1144 Self::NotOwner {
1145 collection,
1146 range_id,
1147 role,
1148 owner,
1149 } => write!(
1150 f,
1151 "this node is {} of {collection}/{range_id}, not its owner — route the write to {owner}",
1152 role.label()
1153 ),
1154 Self::StaleEpoch {
1155 collection,
1156 range_id,
1157 expected,
1158 current,
1159 } => write!(
1160 f,
1161 "stale ownership epoch for {collection}/{range_id}: write authorised under epoch {expected}, current is {current}"
1162 ),
1163 }
1164 }
1165}
1166
1167impl std::error::Error for RangeWriteReject {}
1168
1169#[derive(Debug, Clone, Default)]
1180pub struct ShardOwnershipCatalog {
1181 collections: BTreeMap<CollectionId, ShardKeyMode>,
1186 ranges: BTreeMap<(CollectionId, RangeId), RangeOwnership>,
1187}
1188
1189impl ShardOwnershipCatalog {
1190 pub fn new() -> Self {
1192 Self::default()
1193 }
1194
1195 pub fn declare_collection(
1203 &mut self,
1204 collection: CollectionId,
1205 mode: ShardKeyMode,
1206 ) -> Result<(), CatalogError> {
1207 match self.collections.get(&collection) {
1208 Some(&declared) if declared != mode => Err(CatalogError::ShardKeyModeMismatch {
1209 collection,
1210 declared,
1211 attempted: mode,
1212 }),
1213 _ => {
1214 self.collections.insert(collection, mode);
1215 Ok(())
1216 }
1217 }
1218 }
1219
1220 pub fn shard_key_mode(&self, collection: &CollectionId) -> Option<ShardKeyMode> {
1223 self.collections.get(collection).copied()
1224 }
1225
1226 pub fn apply_update(&mut self, entry: RangeOwnership) -> Result<UpdateOutcome, CatalogError> {
1237 entry.validate_replica_roles()?;
1238
1239 match self.collections.get(entry.collection()) {
1241 Some(&declared) if declared != entry.shard_key_mode() => {
1242 return Err(CatalogError::ShardKeyModeMismatch {
1243 collection: entry.collection().clone(),
1244 declared,
1245 attempted: entry.shard_key_mode(),
1246 });
1247 }
1248 _ => {}
1249 }
1250
1251 let key = entry.key();
1252 match self.ranges.get(&key) {
1253 Some(current) => {
1254 if entry.version() <= current.version() {
1255 return Err(CatalogError::StaleVersion {
1256 collection: entry.collection().clone(),
1257 range_id: entry.range_id(),
1258 current: current.version(),
1259 attempted: entry.version(),
1260 });
1261 }
1262 if let Some(existing) = self.overlapping_sibling(&entry) {
1263 return Err(CatalogError::OverlappingRange {
1264 collection: entry.collection().clone(),
1265 existing,
1266 attempted: entry.range_id(),
1267 });
1268 }
1269 self.collections
1270 .insert(entry.collection().clone(), entry.shard_key_mode());
1271 self.ranges.insert(key, entry);
1272 Ok(UpdateOutcome::Updated)
1273 }
1274 None => {
1275 if let Some(existing) = self.overlapping_sibling(&entry) {
1278 return Err(CatalogError::OverlappingRange {
1279 collection: entry.collection().clone(),
1280 existing,
1281 attempted: entry.range_id(),
1282 });
1283 }
1284 self.collections
1285 .insert(entry.collection().clone(), entry.shard_key_mode());
1286 self.ranges.insert(key, entry);
1287 Ok(UpdateOutcome::Created)
1288 }
1289 }
1290 }
1291
1292 fn overlapping_sibling(&self, entry: &RangeOwnership) -> Option<RangeId> {
1293 self.ranges_for(entry.collection())
1294 .find(|range| {
1295 range.range_id() != entry.range_id() && range.bounds().overlaps(entry.bounds())
1296 })
1297 .map(RangeOwnership::range_id)
1298 }
1299
1300 pub fn range(&self, collection: &CollectionId, range_id: RangeId) -> Option<&RangeOwnership> {
1303 self.ranges.get(&(collection.clone(), range_id))
1304 }
1305
1306 pub fn ranges_for<'a>(
1308 &'a self,
1309 collection: &CollectionId,
1310 ) -> impl Iterator<Item = &'a RangeOwnership> {
1311 let collection = collection.clone();
1312 self.ranges
1313 .iter()
1314 .filter(move |((c, _), _)| *c == collection)
1315 .map(|(_, r)| r)
1316 }
1317
1318 pub fn route(&self, collection: &CollectionId, key: &[u8]) -> Option<&RangeOwnership> {
1323 self.ranges_for(collection)
1324 .find(|r| r.bounds().contains(key))
1325 }
1326
1327 pub fn route_shard_key(
1333 &self,
1334 collection: &CollectionId,
1335 shard_key: &[u8],
1336 ) -> Option<&RangeOwnership> {
1337 match self.shard_key_mode(collection)? {
1338 ShardKeyMode::Ordered => self.route(collection, shard_key),
1339 ShardKeyMode::Hash => {
1340 let range_key = hash_shard_key_to_range_key(shard_key);
1341 self.route(collection, &range_key)
1342 }
1343 }
1344 }
1345
1346 pub fn role_at(
1351 &self,
1352 node: &NodeIdentity,
1353 collection: &CollectionId,
1354 range_id: RangeId,
1355 ) -> Option<RangeRole> {
1356 self.range(collection, range_id)
1357 .map(|range| range.role_of(node))
1358 }
1359
1360 pub fn admit_public_write(
1376 &self,
1377 node: &NodeIdentity,
1378 collection: &CollectionId,
1379 key: &[u8],
1380 expected_epoch: OwnershipEpoch,
1381 ) -> Result<&RangeOwnership, RangeWriteReject> {
1382 let range =
1383 self.route_shard_key(collection, key)
1384 .ok_or_else(|| RangeWriteReject::NoRange {
1385 collection: collection.clone(),
1386 })?;
1387 let role = range.role_of(node);
1388 if !role.may_write_public() {
1389 return Err(RangeWriteReject::NotOwner {
1390 collection: collection.clone(),
1391 range_id: range.range_id(),
1392 role,
1393 owner: range.owner().clone(),
1394 });
1395 }
1396 if expected_epoch != range.epoch() {
1397 return Err(RangeWriteReject::StaleEpoch {
1398 collection: collection.clone(),
1399 range_id: range.range_id(),
1400 expected: expected_epoch,
1401 current: range.epoch(),
1402 });
1403 }
1404 Ok(range)
1405 }
1406
1407 pub fn range_count(&self) -> usize {
1409 self.ranges.len()
1410 }
1411
1412 pub fn entries(&self) -> impl Iterator<Item = &RangeOwnership> {
1416 self.ranges.values()
1417 }
1418}
1419
1420#[cfg(test)]
1421mod tests {
1422 use super::*;
1423
1424 fn collection(name: &str) -> CollectionId {
1425 CollectionId::new(name).unwrap()
1426 }
1427
1428 fn ident(cn: &str) -> NodeIdentity {
1429 NodeIdentity::from_certificate_subject(cn).unwrap()
1430 }
1431
1432 fn bounds(lower: &[u8], upper: &[u8]) -> RangeBounds {
1433 RangeBounds::new(RangeBound::key(lower), RangeBound::key(upper)).unwrap()
1434 }
1435
1436 fn hash_range(coll: &CollectionId, id: u64, bnds: RangeBounds, owner: &str) -> RangeOwnership {
1438 RangeOwnership::establish(
1439 coll.clone(),
1440 RangeId::new(id),
1441 ShardKeyMode::Hash,
1442 bnds,
1443 ident(owner),
1444 [ident("CN=replica-1")],
1445 PlacementMetadata::with_replication_factor(3),
1446 )
1447 }
1448
1449 fn single_hash_slot_bounds(key: &[u8]) -> RangeBounds {
1450 let slot = super::super::slot::hash_shard_key_to_slot(key);
1451 let lower = RangeBound::key(slot.range_key());
1452 let upper = match slot.value().checked_add(1) {
1453 Some(next) if next < super::super::slot::PRODUCTION_HASH_SLOT_COUNT => {
1454 RangeBound::key(super::super::slot::HashSlot::new(next).unwrap().range_key())
1455 }
1456 _ => RangeBound::Max,
1457 };
1458 RangeBounds::new(lower, upper).unwrap()
1459 }
1460
1461 #[test]
1462 fn hot_mirror_behind_commit_watermark_cannot_be_promoted() {
1463 let orders = collection("orders");
1464 let current = hash_range(&orders, 1, RangeBounds::full(), "CN=owner-a");
1465 let candidate = HotMirrorCandidate::new(ident("CN=replica-1"), 99);
1466
1467 assert_eq!(
1468 current.promote_hot_mirror(&candidate, 100),
1469 Err(HotMirrorPromotionRefusal::WatermarkNotCovered {
1470 candidate_lsn: 99,
1471 watermark: 100,
1472 }),
1473 );
1474 }
1475
1476 #[test]
1477 fn hot_mirror_covering_watermark_can_be_selected_for_promotion() {
1478 let orders = collection("orders");
1479 let current = hash_range(&orders, 1, RangeBounds::full(), "CN=owner-a");
1480 let candidate = HotMirrorCandidate::new(ident("CN=replica-1"), 100);
1481
1482 let promotion = current.promote_hot_mirror(&candidate, 100).unwrap();
1483
1484 assert_eq!(promotion.previous(), ¤t);
1485 assert_eq!(promotion.promoted().owner(), &ident("CN=replica-1"));
1486 assert_eq!(promotion.promoted().epoch(), current.epoch().next());
1487 assert_eq!(
1488 promotion.promoted().role_of(&ident("CN=owner-a")),
1489 RangeRole::Replica,
1490 );
1491 }
1492
1493 #[test]
1494 fn hot_mirror_promotion_publishes_epoch_before_durable_writes() {
1495 let orders = collection("orders");
1496 let key = b"order-7";
1497 let current = hash_range(&orders, 1, single_hash_slot_bounds(key), "CN=owner-a");
1498 let old_epoch = current.epoch();
1499 let promotion = current
1500 .promote_hot_mirror(&HotMirrorCandidate::new(ident("CN=replica-1"), 120), 100)
1501 .unwrap();
1502 let new_epoch = promotion.promoted().epoch();
1503 let mut catalog = ShardOwnershipCatalog::new();
1504 catalog.apply_update(current).unwrap();
1505 catalog.apply_update(promotion.promoted().clone()).unwrap();
1506
1507 assert_eq!(
1508 catalog
1509 .admit_public_write(&ident("CN=replica-1"), &orders, key, old_epoch)
1510 .unwrap_err(),
1511 RangeWriteReject::StaleEpoch {
1512 collection: orders.clone(),
1513 range_id: RangeId::new(1),
1514 expected: old_epoch,
1515 current: new_epoch,
1516 },
1517 );
1518 assert!(catalog
1519 .admit_public_write(&ident("CN=replica-1"), &orders, key, new_epoch)
1520 .is_ok());
1521 }
1522
1523 #[test]
1524 fn hot_mirror_promotion_fences_old_owner_by_epoch() {
1525 let orders = collection("orders");
1526 let key = b"order-7";
1527 let current = hash_range(&orders, 1, single_hash_slot_bounds(key), "CN=owner-a");
1528 let old_epoch = current.epoch();
1529 let promotion = current
1530 .promote_hot_mirror(&HotMirrorCandidate::new(ident("CN=replica-1"), 120), 100)
1531 .unwrap();
1532 let mut catalog = ShardOwnershipCatalog::new();
1533 catalog.apply_update(current).unwrap();
1534 catalog.apply_update(promotion.promoted().clone()).unwrap();
1535
1536 assert_eq!(
1537 catalog
1538 .admit_public_write(&ident("CN=owner-a"), &orders, key, old_epoch)
1539 .unwrap_err(),
1540 RangeWriteReject::NotOwner {
1541 collection: orders,
1542 range_id: RangeId::new(1),
1543 role: RangeRole::Replica,
1544 owner: ident("CN=replica-1"),
1545 },
1546 );
1547 }
1548
1549 #[test]
1550 fn empty_catalog_creation() {
1551 let catalog = ShardOwnershipCatalog::new();
1552 assert_eq!(catalog.range_count(), 0);
1553 assert!(catalog.shard_key_mode(&collection("orders")).is_none());
1554 }
1555
1556 #[test]
1557 fn hash_is_the_default_shard_key_mode() {
1558 assert_eq!(ShardKeyMode::default(), ShardKeyMode::Hash);
1561
1562 let mut catalog = ShardOwnershipCatalog::new();
1563 let orders = collection("orders");
1564 catalog
1565 .apply_update(hash_range(&orders, 1, RangeBounds::full(), "CN=node-a"))
1566 .unwrap();
1567 assert_eq!(catalog.shard_key_mode(&orders), Some(ShardKeyMode::Hash));
1568 }
1569
1570 #[test]
1571 fn hash_range_entry_routes_to_owner() {
1572 let mut catalog = ShardOwnershipCatalog::new();
1573 let orders = collection("orders");
1574
1575 catalog
1577 .apply_update(hash_range(
1578 &orders,
1579 1,
1580 RangeBounds::new(RangeBound::Min, RangeBound::key([0x80])).unwrap(),
1581 "CN=node-a",
1582 ))
1583 .unwrap();
1584 catalog
1585 .apply_update(hash_range(
1586 &orders,
1587 2,
1588 RangeBounds::new(RangeBound::key([0x80]), RangeBound::Max).unwrap(),
1589 "CN=node-b",
1590 ))
1591 .unwrap();
1592
1593 assert_eq!(
1595 catalog.route(&orders, &[0x10]).unwrap().owner(),
1596 &ident("CN=node-a")
1597 );
1598 assert_eq!(
1599 catalog.route(&orders, &[0x80]).unwrap().owner(),
1600 &ident("CN=node-b")
1601 );
1602 assert_eq!(
1603 catalog.route(&orders, &[0xff]).unwrap().owner(),
1604 &ident("CN=node-b")
1605 );
1606 let r = catalog.route(&orders, &[0x10]).unwrap();
1608 assert_eq!(r.replicas(), &[ident("CN=replica-1")]);
1609 assert_eq!(r.epoch(), OwnershipEpoch::initial());
1610 }
1611
1612 #[test]
1613 fn hash_mode_routes_logical_shard_key_through_hash_slot() {
1614 let mut catalog = ShardOwnershipCatalog::new();
1615 let orders = collection("orders");
1616 let key = b"tenant:42";
1617 catalog
1618 .apply_update(hash_range(
1619 &orders,
1620 1,
1621 single_hash_slot_bounds(key),
1622 "CN=node-a",
1623 ))
1624 .unwrap();
1625
1626 let routed = catalog
1627 .route_shard_key(&orders, key)
1628 .expect("hash slot range covers the logical shard key");
1629 assert_eq!(routed.owner(), &ident("CN=node-a"));
1630 }
1631
1632 #[test]
1633 fn ordered_mode_can_be_declared_and_routed() {
1634 let mut catalog = ShardOwnershipCatalog::new();
1635 let events = collection("events");
1636 catalog
1637 .declare_collection(events.clone(), ShardKeyMode::Ordered)
1638 .unwrap();
1639 assert_eq!(catalog.shard_key_mode(&events), Some(ShardKeyMode::Ordered));
1640
1641 catalog
1643 .apply_update(RangeOwnership::establish(
1644 events.clone(),
1645 RangeId::new(1),
1646 ShardKeyMode::Ordered,
1647 bounds(b"a", b"m"),
1648 ident("CN=node-a"),
1649 [],
1650 PlacementMetadata::with_replication_factor(3),
1651 ))
1652 .unwrap();
1653 catalog
1654 .apply_update(RangeOwnership::establish(
1655 events.clone(),
1656 RangeId::new(2),
1657 ShardKeyMode::Ordered,
1658 bounds(b"m", b"z"),
1659 ident("CN=node-b"),
1660 [],
1661 PlacementMetadata::with_replication_factor(3),
1662 ))
1663 .unwrap();
1664
1665 assert_eq!(
1666 catalog.route(&events, b"alpha").unwrap().owner(),
1667 &ident("CN=node-a")
1668 );
1669 assert_eq!(
1670 catalog.route(&events, b"mike").unwrap().owner(),
1671 &ident("CN=node-b")
1672 );
1673 assert!(catalog.route(&events, b"zzz").is_none());
1675 }
1676
1677 #[test]
1678 fn declaring_a_conflicting_mode_is_rejected() {
1679 let mut catalog = ShardOwnershipCatalog::new();
1680 let events = collection("events");
1681 catalog
1682 .declare_collection(events.clone(), ShardKeyMode::Ordered)
1683 .unwrap();
1684 catalog
1686 .declare_collection(events.clone(), ShardKeyMode::Ordered)
1687 .unwrap();
1688 let err = catalog
1690 .declare_collection(events.clone(), ShardKeyMode::Hash)
1691 .unwrap_err();
1692 assert_eq!(
1693 err,
1694 CatalogError::ShardKeyModeMismatch {
1695 collection: events.clone(),
1696 declared: ShardKeyMode::Ordered,
1697 attempted: ShardKeyMode::Hash,
1698 }
1699 );
1700 let err = catalog
1702 .apply_update(hash_range(&events, 1, RangeBounds::full(), "CN=node-a"))
1703 .unwrap_err();
1704 assert!(matches!(err, CatalogError::ShardKeyModeMismatch { .. }));
1705 }
1706
1707 #[test]
1708 fn version_bumps_on_owner_transfer_and_epoch_fences() {
1709 let mut catalog = ShardOwnershipCatalog::new();
1710 let orders = collection("orders");
1711 catalog
1712 .apply_update(hash_range(&orders, 1, RangeBounds::full(), "CN=node-a"))
1713 .unwrap();
1714
1715 let current = catalog.range(&orders, RangeId::new(1)).unwrap();
1716 assert_eq!(current.version(), CatalogVersion::initial());
1717 assert_eq!(current.epoch(), OwnershipEpoch::initial());
1718
1719 let moved = current.transfer_to(ident("CN=node-b"), [ident("CN=node-a")]);
1721 let outcome = catalog.apply_update(moved).unwrap();
1722 assert_eq!(outcome, UpdateOutcome::Updated);
1723
1724 let after = catalog.range(&orders, RangeId::new(1)).unwrap();
1725 assert_eq!(after.owner(), &ident("CN=node-b"));
1726 assert_eq!(after.version().value(), 2);
1727 assert_eq!(after.epoch().value(), 2); let replicas_changed = after.update_replicas([ident("CN=node-c")]);
1731 catalog.apply_update(replicas_changed).unwrap();
1732 let after2 = catalog.range(&orders, RangeId::new(1)).unwrap();
1733 assert_eq!(after2.version().value(), 3);
1734 assert_eq!(after2.epoch().value(), 2); assert_eq!(after2.replicas(), &[ident("CN=node-c")]);
1736 }
1737
1738 #[test]
1739 fn stale_update_is_rejected_and_leaves_catalog_unchanged() {
1740 let mut catalog = ShardOwnershipCatalog::new();
1741 let orders = collection("orders");
1742 catalog
1743 .apply_update(hash_range(&orders, 1, RangeBounds::full(), "CN=node-a"))
1744 .unwrap();
1745
1746 let v1 = catalog.range(&orders, RangeId::new(1)).unwrap().clone();
1747 catalog
1749 .apply_update(v1.transfer_to(ident("CN=node-b"), []))
1750 .unwrap();
1751 assert_eq!(
1752 catalog.range(&orders, RangeId::new(1)).unwrap().owner(),
1753 &ident("CN=node-b")
1754 );
1755
1756 let err = catalog.apply_update(v1.clone()).unwrap_err();
1759 assert_eq!(
1760 err,
1761 CatalogError::StaleVersion {
1762 collection: orders.clone(),
1763 range_id: RangeId::new(1),
1764 current: CatalogVersion::initial().next(),
1765 attempted: CatalogVersion::initial(),
1766 }
1767 );
1768 assert_eq!(
1770 catalog.range(&orders, RangeId::new(1)).unwrap().owner(),
1771 &ident("CN=node-b")
1772 );
1773 assert_eq!(
1774 catalog
1775 .range(&orders, RangeId::new(1))
1776 .unwrap()
1777 .version()
1778 .value(),
1779 2
1780 );
1781 }
1782
1783 #[test]
1784 fn overlapping_range_creation_is_rejected() {
1785 let mut catalog = ShardOwnershipCatalog::new();
1786 let orders = collection("orders");
1787 catalog
1788 .apply_update(hash_range(
1789 &orders,
1790 1,
1791 bounds(&[0x00], &[0x80]),
1792 "CN=node-a",
1793 ))
1794 .unwrap();
1795 let err = catalog
1797 .apply_update(hash_range(
1798 &orders,
1799 2,
1800 bounds(&[0x40], &[0xc0]),
1801 "CN=node-b",
1802 ))
1803 .unwrap_err();
1804 assert_eq!(
1805 err,
1806 CatalogError::OverlappingRange {
1807 collection: orders.clone(),
1808 existing: RangeId::new(1),
1809 attempted: RangeId::new(2),
1810 }
1811 );
1812 assert_eq!(catalog.range_count(), 1);
1813 }
1814
1815 #[test]
1816 fn overlapping_range_update_is_rejected() {
1817 let mut catalog = ShardOwnershipCatalog::new();
1818 let orders = collection("orders");
1819 catalog
1820 .apply_update(hash_range(
1821 &orders,
1822 1,
1823 bounds(&[0x00], &[0x80]),
1824 "CN=node-a",
1825 ))
1826 .unwrap();
1827 catalog
1828 .apply_update(hash_range(
1829 &orders,
1830 2,
1831 RangeBounds::new(RangeBound::key([0x80]), RangeBound::Max).unwrap(),
1832 "CN=node-b",
1833 ))
1834 .unwrap();
1835
1836 let widened = catalog
1837 .range(&orders, RangeId::new(1))
1838 .unwrap()
1839 .with_bounds(bounds(&[0x00], &[0xc0]));
1840 let err = catalog.apply_update(widened).unwrap_err();
1841 assert_eq!(
1842 err,
1843 CatalogError::OverlappingRange {
1844 collection: orders.clone(),
1845 existing: RangeId::new(2),
1846 attempted: RangeId::new(1),
1847 }
1848 );
1849 assert_eq!(
1850 catalog.range(&orders, RangeId::new(1)).unwrap().bounds(),
1851 &bounds(&[0x00], &[0x80])
1852 );
1853 }
1854
1855 #[test]
1856 fn catalog_replicates_to_data_members_with_read_visibility() {
1857 let orders = collection("orders");
1860 let mut leader = ShardOwnershipCatalog::new();
1861 let mut data_member = ShardOwnershipCatalog::new();
1862
1863 let create = hash_range(&orders, 1, RangeBounds::full(), "CN=node-a");
1865 leader.apply_update(create.clone()).unwrap();
1866 assert_eq!(
1867 data_member.apply_update(create).unwrap(),
1868 UpdateOutcome::Created
1869 );
1870
1871 assert_eq!(
1873 data_member.route(&orders, b"any-key").unwrap().owner(),
1874 &ident("CN=node-a")
1875 );
1876
1877 let v2 = leader
1879 .range(&orders, RangeId::new(1))
1880 .unwrap()
1881 .transfer_to(ident("CN=node-b"), []);
1882 leader.apply_update(v2.clone()).unwrap();
1883 assert_eq!(
1884 data_member.apply_update(v2.clone()).unwrap(),
1885 UpdateOutcome::Updated
1886 );
1887 assert_eq!(
1888 data_member.route(&orders, b"any-key").unwrap().owner(),
1889 &ident("CN=node-b")
1890 );
1891
1892 let err = data_member.apply_update(v2).unwrap_err();
1895 assert!(matches!(err, CatalogError::StaleVersion { .. }));
1896 assert_eq!(
1897 data_member
1898 .range(&orders, RangeId::new(1))
1899 .unwrap()
1900 .version()
1901 .value(),
1902 2
1903 );
1904 }
1905
1906 #[test]
1907 fn collection_group_scope_is_owned_by_one_placement_authority() {
1908 let group = CollectionGroupId::new("commerce").unwrap();
1909 let assignment = CollectionGroupAuthority::new(
1910 group.clone(),
1911 ident("CN=placement-authority-a"),
1912 [collection("orders"), collection("order_items")],
1913 )
1914 .unwrap();
1915 let mut catalog = PlacementAuthorityCatalog::new();
1916
1917 catalog.assign(assignment.clone()).unwrap();
1918
1919 assert_eq!(
1920 catalog.authority_for_group(&group).unwrap().authority(),
1921 &ident("CN=placement-authority-a")
1922 );
1923 assert_eq!(
1924 catalog
1925 .authority_for_collection(&collection("orders"))
1926 .unwrap(),
1927 &assignment
1928 );
1929
1930 let err = catalog
1931 .assign(
1932 CollectionGroupAuthority::new(
1933 group.clone(),
1934 ident("CN=placement-authority-b"),
1935 [collection("invoices")],
1936 )
1937 .unwrap(),
1938 )
1939 .unwrap_err();
1940 assert_eq!(
1941 err,
1942 PlacementAuthorityError::DuplicateCollectionGroup {
1943 group: group.clone(),
1944 }
1945 );
1946 }
1947
1948 #[test]
1949 fn collection_group_catalog_slices_do_not_overlap() {
1950 let commerce = CollectionGroupId::new("commerce").unwrap();
1951 let analytics = CollectionGroupId::new("analytics-events").unwrap();
1952 let support = CollectionGroupId::new("support").unwrap();
1953 let orders = collection("orders");
1954 let order_items = collection("order_items");
1955 let events = collection("events");
1956 let tickets = collection("tickets");
1957 let mut catalog = PlacementAuthorityCatalog::new();
1958
1959 catalog
1961 .assign(
1962 CollectionGroupAuthority::new(
1963 commerce.clone(),
1964 ident("CN=placement-authority-a"),
1965 [orders.clone(), order_items],
1966 )
1967 .unwrap(),
1968 )
1969 .unwrap();
1970 assert_eq!(
1971 catalog.authority_for_collection(&orders).unwrap().group(),
1972 &commerce
1973 );
1974
1975 catalog
1977 .assign(
1978 CollectionGroupAuthority::new(
1979 analytics.clone(),
1980 ident("CN=placement-authority-b"),
1981 [events.clone()],
1982 )
1983 .unwrap(),
1984 )
1985 .unwrap();
1986 assert_eq!(
1987 catalog.authority_for_collection(&events).unwrap().group(),
1988 &analytics
1989 );
1990
1991 let err = catalog
1992 .assign(
1993 CollectionGroupAuthority::new(
1994 support.clone(),
1995 ident("CN=placement-authority-c"),
1996 [tickets, events.clone()],
1997 )
1998 .unwrap(),
1999 )
2000 .unwrap_err();
2001 assert_eq!(
2002 err,
2003 PlacementAuthorityError::OverlappingCatalogSlice {
2004 collection: events,
2005 existing_group: analytics,
2006 attempted_group: support,
2007 }
2008 );
2009 }
2010
2011 #[test]
2012 fn range_bounds_reject_empty_or_inverted() {
2013 assert!(RangeBounds::new(RangeBound::key([0x10]), RangeBound::key([0x10])).is_err());
2014 assert!(RangeBounds::new(RangeBound::key([0x20]), RangeBound::key([0x10])).is_err());
2015 assert!(RangeBounds::new(RangeBound::Max, RangeBound::Min).is_err());
2016 assert!(RangeBounds::full().contains(b"anything"));
2017 }
2018
2019 fn range_with(
2025 coll: &CollectionId,
2026 id: u64,
2027 bnds: RangeBounds,
2028 owner: &str,
2029 replicas: &[&str],
2030 ) -> RangeOwnership {
2031 RangeOwnership::establish(
2032 coll.clone(),
2033 RangeId::new(id),
2034 ShardKeyMode::Hash,
2035 bnds,
2036 ident(owner),
2037 replicas.iter().map(|r| ident(r)).collect::<Vec<_>>(),
2038 PlacementMetadata::with_replication_factor(3),
2039 )
2040 }
2041
2042 #[test]
2043 fn role_of_distinguishes_owner_replica_and_no_copy() {
2044 let orders = collection("orders");
2045 let range = range_with(&orders, 1, RangeBounds::full(), "CN=node-a", &["CN=node-b"]);
2046
2047 assert_eq!(range.role_of(&ident("CN=node-a")), RangeRole::Owner);
2048 assert_eq!(range.role_of(&ident("CN=node-b")), RangeRole::Replica);
2049 assert_eq!(range.role_of(&ident("CN=node-c")), RangeRole::NoCopy);
2050 assert!(RangeRole::Owner.may_write_public());
2051 assert!(!RangeRole::Replica.may_write_public());
2052 assert!(!RangeRole::NoCopy.may_write_public());
2053 }
2054
2055 #[test]
2056 fn replica_role_metadata_distinguishes_hot_mirror_and_archive() {
2057 let orders = collection("orders");
2058 let range = range_with(&orders, 1, RangeBounds::full(), "CN=node-a", &["CN=node-b"])
2059 .with_compressed_archive_replicas([ident("CN=node-c")]);
2060
2061 assert_eq!(range.hot_mirror_replicas(), &[ident("CN=node-b")]);
2062 assert_eq!(range.compressed_archive_replicas(), &[ident("CN=node-c")]);
2063 assert_eq!(
2064 range.replica_role_of(&ident("CN=node-b")),
2065 Some(ReplicaRole::HotMirror)
2066 );
2067 assert_eq!(
2068 range.replica_role_of(&ident("CN=node-c")),
2069 Some(ReplicaRole::CompressedArchive)
2070 );
2071 assert!(ReplicaRole::HotMirror.is_promotion_candidate());
2072 assert!(!ReplicaRole::CompressedArchive.is_promotion_candidate());
2073 assert!(ReplicaRole::CompressedArchive.is_restore_only());
2074 }
2075
2076 #[test]
2077 fn catalog_rejects_owner_or_duplicate_replica_roles() {
2078 let orders = collection("orders");
2079 let mut catalog = ShardOwnershipCatalog::new();
2080
2081 let owner_also_hot =
2082 range_with(&orders, 1, RangeBounds::full(), "CN=node-a", &["CN=node-a"]);
2083 assert!(matches!(
2084 catalog.apply_update(owner_also_hot).unwrap_err(),
2085 CatalogError::InvalidReplicaRoles { .. }
2086 ));
2087
2088 let owner_also_archive = range_with(&orders, 1, RangeBounds::full(), "CN=node-a", &[])
2089 .with_compressed_archive_replicas([ident("CN=node-a")]);
2090 assert!(matches!(
2091 catalog.apply_update(owner_also_archive).unwrap_err(),
2092 CatalogError::InvalidReplicaRoles { .. }
2093 ));
2094
2095 let duplicate_hot_and_archive =
2096 range_with(&orders, 1, RangeBounds::full(), "CN=node-a", &["CN=node-b"])
2097 .with_compressed_archive_replicas([ident("CN=node-b")]);
2098 assert!(matches!(
2099 catalog.apply_update(duplicate_hot_and_archive).unwrap_err(),
2100 CatalogError::InvalidReplicaRoles { .. }
2101 ));
2102 }
2103
2104 #[test]
2105 fn role_is_per_range_not_a_global_node_role() {
2106 let mut catalog = ShardOwnershipCatalog::new();
2109 let orders = collection("orders");
2110 catalog
2111 .apply_update(range_with(
2112 &orders,
2113 1,
2114 RangeBounds::new(RangeBound::Min, RangeBound::key([0x80])).unwrap(),
2115 "CN=node-a",
2116 &["CN=node-b"],
2117 ))
2118 .unwrap();
2119 catalog
2120 .apply_update(range_with(
2121 &orders,
2122 2,
2123 RangeBounds::new(RangeBound::key([0x80]), RangeBound::Max).unwrap(),
2124 "CN=node-b",
2125 &["CN=node-a"],
2126 ))
2127 .unwrap();
2128
2129 let node_a = ident("CN=node-a");
2130 assert_eq!(
2131 catalog.role_at(&node_a, &orders, RangeId::new(1)),
2132 Some(RangeRole::Owner)
2133 );
2134 assert_eq!(
2135 catalog.role_at(&node_a, &orders, RangeId::new(2)),
2136 Some(RangeRole::Replica)
2137 );
2138 assert_eq!(catalog.role_at(&node_a, &orders, RangeId::new(99)), None);
2140 assert_eq!(
2142 catalog.role_at(&node_a, &collection("ghost"), RangeId::new(1)),
2143 None
2144 );
2145 }
2146
2147 #[test]
2148 fn public_write_admitted_on_owner_at_matching_epoch() {
2149 let mut catalog = ShardOwnershipCatalog::new();
2150 let orders = collection("orders");
2151 catalog
2152 .apply_update(range_with(
2153 &orders,
2154 1,
2155 RangeBounds::full(),
2156 "CN=node-a",
2157 &["CN=node-b"],
2158 ))
2159 .unwrap();
2160
2161 let admitted = catalog
2162 .admit_public_write(
2163 &ident("CN=node-a"),
2164 &orders,
2165 b"k",
2166 OwnershipEpoch::initial(),
2167 )
2168 .expect("owner at current epoch may write");
2169 assert_eq!(admitted.owner(), &ident("CN=node-a"));
2170 assert_eq!(admitted.range_id(), RangeId::new(1));
2171 }
2172
2173 #[test]
2174 fn public_write_uses_hash_slot_routing_for_hash_collections() {
2175 let mut catalog = ShardOwnershipCatalog::new();
2176 let orders = collection("orders");
2177 let key = b"tenant:42";
2178 catalog
2179 .apply_update(hash_range(
2180 &orders,
2181 1,
2182 single_hash_slot_bounds(key),
2183 "CN=node-a",
2184 ))
2185 .unwrap();
2186
2187 let admitted = catalog
2188 .admit_public_write(&ident("CN=node-a"), &orders, key, OwnershipEpoch::initial())
2189 .expect("hash owner admits write routed by shard-key slot");
2190 assert_eq!(admitted.range_id(), RangeId::new(1));
2191 }
2192
2193 #[test]
2194 fn public_write_rejected_on_replica_with_routing_error() {
2195 let mut catalog = ShardOwnershipCatalog::new();
2196 let orders = collection("orders");
2197 catalog
2198 .apply_update(range_with(
2199 &orders,
2200 1,
2201 RangeBounds::full(),
2202 "CN=node-a",
2203 &["CN=node-b"],
2204 ))
2205 .unwrap();
2206
2207 let err = catalog
2210 .admit_public_write(
2211 &ident("CN=node-b"),
2212 &orders,
2213 b"k",
2214 OwnershipEpoch::initial(),
2215 )
2216 .unwrap_err();
2217 match err {
2218 RangeWriteReject::NotOwner {
2219 role, ref owner, ..
2220 } => {
2221 assert_eq!(role, RangeRole::Replica);
2222 assert_eq!(owner, &ident("CN=node-a"));
2223 }
2224 other => panic!("expected NotOwner(Replica), got {other:?}"),
2225 }
2226 assert!(err.to_string().contains("route the write to"));
2228 }
2229
2230 #[test]
2231 fn public_write_rejected_on_no_copy_holder() {
2232 let mut catalog = ShardOwnershipCatalog::new();
2233 let orders = collection("orders");
2234 catalog
2235 .apply_update(range_with(
2236 &orders,
2237 1,
2238 RangeBounds::full(),
2239 "CN=node-a",
2240 &["CN=node-b"],
2241 ))
2242 .unwrap();
2243
2244 let err = catalog
2246 .admit_public_write(
2247 &ident("CN=node-c"),
2248 &orders,
2249 b"k",
2250 OwnershipEpoch::initial(),
2251 )
2252 .unwrap_err();
2253 match err {
2254 RangeWriteReject::NotOwner { role, .. } => assert_eq!(role, RangeRole::NoCopy),
2255 other => panic!("expected NotOwner(NoCopy), got {other:?}"),
2256 }
2257 }
2258
2259 #[test]
2260 fn public_write_rejected_on_stale_ownership_epoch() {
2261 let mut catalog = ShardOwnershipCatalog::new();
2265 let orders = collection("orders");
2266 let v1 = range_with(&orders, 1, RangeBounds::full(), "CN=node-a", &["CN=node-b"]);
2267 let original_epoch = v1.epoch();
2268 catalog.apply_update(v1.clone()).unwrap();
2269
2270 let v2 = v1.transfer_to(ident("CN=node-b"), [ident("CN=node-a")]);
2271 catalog.apply_update(v2.clone()).unwrap();
2272 let v3 = v2.transfer_to(ident("CN=node-a"), [ident("CN=node-b")]);
2273 catalog.apply_update(v3.clone()).unwrap();
2274
2275 assert_ne!(original_epoch, v3.epoch());
2277 let err = catalog
2278 .admit_public_write(&ident("CN=node-a"), &orders, b"k", original_epoch)
2279 .unwrap_err();
2280 match err {
2281 RangeWriteReject::StaleEpoch {
2282 expected, current, ..
2283 } => {
2284 assert_eq!(expected, original_epoch);
2285 assert_eq!(current, v3.epoch());
2286 }
2287 other => panic!("expected StaleEpoch, got {other:?}"),
2288 }
2289 assert!(catalog
2291 .admit_public_write(&ident("CN=node-a"), &orders, b"k", v3.epoch())
2292 .is_ok());
2293 }
2294
2295 #[test]
2296 fn public_write_rejected_when_no_range_covers_the_key() {
2297 let catalog = ShardOwnershipCatalog::new();
2298 let orders = collection("orders");
2299 let err = catalog
2300 .admit_public_write(
2301 &ident("CN=node-a"),
2302 &orders,
2303 b"k",
2304 OwnershipEpoch::initial(),
2305 )
2306 .unwrap_err();
2307 assert!(matches!(err, RangeWriteReject::NoRange { .. }));
2308 }
2309
2310 #[test]
2311 fn internal_apply_path_stays_privileged_for_a_public_write_replica() {
2312 use crate::replication::cdc::{ChangeOperation, ChangeRecord, RangeAuthority};
2317
2318 let mut catalog = ShardOwnershipCatalog::new();
2319 let orders = collection("orders");
2320 catalog
2321 .apply_update(range_with(
2322 &orders,
2323 7,
2324 RangeBounds::full(),
2325 "CN=node-a",
2326 &["CN=node-b"],
2327 ))
2328 .unwrap();
2329
2330 assert!(matches!(
2332 catalog
2333 .admit_public_write(
2334 &ident("CN=node-b"),
2335 &orders,
2336 b"k",
2337 OwnershipEpoch::initial()
2338 )
2339 .unwrap_err(),
2340 RangeWriteReject::NotOwner {
2341 role: RangeRole::Replica,
2342 ..
2343 }
2344 ));
2345
2346 let record = ChangeRecord {
2349 term: 1,
2350 lsn: 1,
2351 timestamp: 0,
2352 operation: ChangeOperation::Insert,
2353 collection: orders.as_str().to_string(),
2354 entity_id: 1,
2355 entity_kind: "row".to_string(),
2356 entity_bytes: Some(vec![1]),
2357 metadata: None,
2358 refresh_records: None,
2359 range_id: None,
2360 ownership_epoch: None,
2361 }
2362 .with_range_authority(7, OwnershipEpoch::initial().value());
2363 let fence = RangeAuthority {
2364 range_id: 7,
2365 min_term: 1,
2366 min_ownership_epoch: OwnershipEpoch::initial().value(),
2367 };
2368 assert!(
2369 fence.admit(&record).is_ok(),
2370 "replica internal apply must remain privileged for the owner's changes"
2371 );
2372 }
2373}