1use super::identity::NodeIdentity;
77use super::ownership::{
78 CatalogVersion, CollectionId, OwnershipEpoch, RangeId, RangeOwnership, RangeRole,
79 ShardOwnershipCatalog,
80};
81
82#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
90pub struct SupervisorTerm(u64);
91
92impl SupervisorTerm {
93 pub fn genesis() -> Self {
95 Self(1)
96 }
97
98 pub fn new(value: u64) -> Self {
99 Self(value)
100 }
101
102 pub fn value(self) -> u64 {
103 self.0
104 }
105
106 pub fn next(self) -> Self {
108 Self(self.0 + 1)
109 }
110}
111
112impl std::fmt::Display for SupervisorTerm {
113 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
114 write!(f, "{}", self.0)
115 }
116}
117
118#[derive(Debug, Clone, Copy, PartialEq, Eq)]
125pub struct LeaseClockSample {
126 monotonic_ms: u64,
127 wall_ms: u64,
128}
129
130impl LeaseClockSample {
131 pub fn new(monotonic_ms: u64, wall_ms: u64) -> Self {
132 Self {
133 monotonic_ms,
134 wall_ms,
135 }
136 }
137
138 pub fn monotonic(monotonic_ms: u64) -> Self {
139 Self {
140 monotonic_ms,
141 wall_ms: 0,
142 }
143 }
144
145 pub fn monotonic_ms(self) -> u64 {
146 self.monotonic_ms
147 }
148
149 pub fn wall_ms(self) -> u64 {
150 self.wall_ms
151 }
152}
153
154#[derive(Debug, Clone, Copy, PartialEq, Eq)]
161pub struct OwnershipLeaseClockConfig {
162 lease_ttl_ms: u64,
163 safety_margin_ms: u64,
164 max_clock_drift_ms: u64,
165}
166
167impl OwnershipLeaseClockConfig {
168 pub fn new(
169 lease_ttl_ms: u64,
170 safety_margin_ms: u64,
171 max_clock_drift_ms: u64,
172 ) -> Result<Self, LeaseClockConfigError> {
173 if safety_margin_ms < max_clock_drift_ms {
174 return Err(LeaseClockConfigError::SafetyMarginBelowMaxDrift {
175 safety_margin_ms,
176 max_clock_drift_ms,
177 });
178 }
179 Ok(Self {
180 lease_ttl_ms,
181 safety_margin_ms,
182 max_clock_drift_ms,
183 })
184 }
185
186 pub fn lease_ttl_ms(self) -> u64 {
187 self.lease_ttl_ms
188 }
189
190 pub fn safety_margin_ms(self) -> u64 {
191 self.safety_margin_ms
192 }
193
194 pub fn max_clock_drift_ms(self) -> u64 {
195 self.max_clock_drift_ms
196 }
197}
198
199#[derive(Debug, Clone, Copy, PartialEq, Eq)]
200pub enum LeaseClockConfigError {
201 SafetyMarginBelowMaxDrift {
202 safety_margin_ms: u64,
203 max_clock_drift_ms: u64,
204 },
205}
206
207impl std::fmt::Display for LeaseClockConfigError {
208 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
209 match self {
210 Self::SafetyMarginBelowMaxDrift {
211 safety_margin_ms,
212 max_clock_drift_ms,
213 } => write!(
214 f,
215 "ownership lease safety margin {safety_margin_ms} ms is below configured max clock drift {max_clock_drift_ms} ms"
216 ),
217 }
218 }
219}
220
221impl std::error::Error for LeaseClockConfigError {}
222
223#[derive(Debug, Clone, PartialEq, Eq)]
234pub struct OwnershipLease {
235 supervisor_term: SupervisorTerm,
236 collection: CollectionId,
237 range_id: RangeId,
238 owner: NodeIdentity,
239 epoch: OwnershipEpoch,
240 granted_at_monotonic_ms: u64,
241 expires_at_monotonic_ms: u64,
242}
243
244impl OwnershipLease {
245 #[allow(clippy::too_many_arguments)]
249 pub fn grant(
250 supervisor_term: SupervisorTerm,
251 collection: CollectionId,
252 range_id: RangeId,
253 owner: NodeIdentity,
254 epoch: OwnershipEpoch,
255 granted_at_ms: u64,
256 ttl_ms: u64,
257 ) -> Self {
258 Self::grant_at_clock(
259 supervisor_term,
260 collection,
261 range_id,
262 owner,
263 epoch,
264 LeaseClockSample::monotonic(granted_at_ms),
265 ttl_ms,
266 )
267 }
268
269 #[allow(clippy::too_many_arguments)]
273 pub fn grant_at_clock(
274 supervisor_term: SupervisorTerm,
275 collection: CollectionId,
276 range_id: RangeId,
277 owner: NodeIdentity,
278 epoch: OwnershipEpoch,
279 granted_at: LeaseClockSample,
280 ttl_ms: u64,
281 ) -> Self {
282 let granted_at_monotonic_ms = granted_at.monotonic_ms();
283 Self {
284 supervisor_term,
285 collection,
286 range_id,
287 owner,
288 epoch,
289 granted_at_monotonic_ms,
290 expires_at_monotonic_ms: granted_at_monotonic_ms.saturating_add(ttl_ms),
291 }
292 }
293
294 pub fn supervisor_term(&self) -> SupervisorTerm {
295 self.supervisor_term
296 }
297
298 pub fn collection(&self) -> &CollectionId {
299 &self.collection
300 }
301
302 pub fn range_id(&self) -> RangeId {
303 self.range_id
304 }
305
306 pub fn owner(&self) -> &NodeIdentity {
307 &self.owner
308 }
309
310 pub fn epoch(&self) -> OwnershipEpoch {
311 self.epoch
312 }
313
314 pub fn granted_at_ms(&self) -> u64 {
315 self.granted_at_monotonic_ms
316 }
317
318 pub fn expires_at_ms(&self) -> u64 {
319 self.expires_at_monotonic_ms
320 }
321
322 pub fn is_expired(&self, now_ms: u64) -> bool {
326 now_ms >= self.expires_at_monotonic_ms
327 }
328
329 pub fn is_expired_at(&self, now: LeaseClockSample) -> bool {
330 self.is_expired(now.monotonic_ms())
331 }
332
333 pub fn remaining_ms(&self, now_ms: u64) -> u64 {
336 self.expires_at_monotonic_ms.saturating_sub(now_ms)
337 }
338
339 pub fn remaining_at(&self, now: LeaseClockSample) -> u64 {
340 self.remaining_ms(now.monotonic_ms())
341 }
342
343 fn covers(&self, collection: &CollectionId, range_id: RangeId, owner: &NodeIdentity) -> bool {
347 self.collection == *collection && self.range_id == range_id && self.owner == *owner
348 }
349}
350
351#[derive(Debug, Clone, PartialEq, Eq)]
356pub enum FenceReason {
357 Unleased,
360 Revoked,
363 TermSuperseded {
366 lease_term: SupervisorTerm,
367 current_term: SupervisorTerm,
368 },
369 EpochSuperseded {
372 lease_epoch: OwnershipEpoch,
373 current_epoch: OwnershipEpoch,
374 },
375 Expired { now_ms: u64, expires_at_ms: u64 },
379}
380
381impl std::fmt::Display for FenceReason {
382 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
383 match self {
384 Self::Unleased => write!(f, "owner holds no ownership lease"),
385 Self::Revoked => write!(f, "ownership lease was revoked"),
386 Self::TermSuperseded {
387 lease_term,
388 current_term,
389 } => write!(
390 f,
391 "ownership lease granted under supervisor term {lease_term} is behind current term {current_term}"
392 ),
393 Self::EpochSuperseded {
394 lease_epoch,
395 current_epoch,
396 } => write!(
397 f,
398 "ownership lease epoch {lease_epoch} no longer matches current ownership epoch {current_epoch}"
399 ),
400 Self::Expired {
401 now_ms,
402 expires_at_ms,
403 } => write!(
404 f,
405 "ownership lease expired at {expires_at_ms} ms (now {now_ms} ms)"
406 ),
407 }
408 }
409}
410
411impl std::error::Error for FenceReason {}
412
413#[derive(Debug, Clone, PartialEq, Eq)]
415pub enum OwnerWriteMode {
416 Durable,
418 Fenced(FenceReason),
422}
423
424impl OwnerWriteMode {
425 pub fn may_write_durable(&self) -> bool {
427 matches!(self, OwnerWriteMode::Durable)
428 }
429
430 pub fn is_fenced(&self) -> bool {
432 matches!(self, OwnerWriteMode::Fenced(_))
433 }
434}
435
436#[derive(Debug, Clone, Copy, PartialEq, Eq)]
442pub enum RangeRequest {
443 DurableWrite,
445 StaleRead,
447 ReplicationCatchUp,
451}
452
453#[derive(Debug, Clone, PartialEq, Eq)]
455pub struct LeaseFenceRejection {
456 pub reason: FenceReason,
458}
459
460impl std::fmt::Display for LeaseFenceRejection {
461 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
462 write!(
463 f,
464 "durable write rejected: owner is self-fenced ({})",
465 self.reason
466 )
467 }
468}
469
470impl std::error::Error for LeaseFenceRejection {}
471
472#[derive(Debug, Clone, Default, PartialEq, Eq)]
482pub struct LeasedOwner {
483 lease: Option<OwnershipLease>,
484 revoked: bool,
485}
486
487impl LeasedOwner {
488 pub fn unleased() -> Self {
490 Self {
491 lease: None,
492 revoked: false,
493 }
494 }
495
496 pub fn with_lease(lease: OwnershipLease) -> Self {
498 Self {
499 lease: Some(lease),
500 revoked: false,
501 }
502 }
503
504 pub fn grant(&mut self, lease: OwnershipLease) {
507 self.lease = Some(lease);
508 self.revoked = false;
509 }
510
511 pub fn revoke(&mut self) {
515 self.revoked = true;
516 }
517
518 pub fn lease(&self) -> Option<&OwnershipLease> {
522 self.lease.as_ref()
523 }
524
525 pub fn evaluate(
536 &self,
537 current_term: SupervisorTerm,
538 current_epoch: OwnershipEpoch,
539 now_ms: u64,
540 ) -> OwnerWriteMode {
541 if self.revoked {
542 return OwnerWriteMode::Fenced(FenceReason::Revoked);
543 }
544 let Some(lease) = &self.lease else {
545 return OwnerWriteMode::Fenced(FenceReason::Unleased);
546 };
547 if lease.supervisor_term != current_term {
548 return OwnerWriteMode::Fenced(FenceReason::TermSuperseded {
549 lease_term: lease.supervisor_term,
550 current_term,
551 });
552 }
553 if lease.epoch != current_epoch {
554 return OwnerWriteMode::Fenced(FenceReason::EpochSuperseded {
555 lease_epoch: lease.epoch,
556 current_epoch,
557 });
558 }
559 if lease.is_expired(now_ms) {
560 return OwnerWriteMode::Fenced(FenceReason::Expired {
561 now_ms,
562 expires_at_ms: lease.expires_at_monotonic_ms,
563 });
564 }
565 OwnerWriteMode::Durable
566 }
567
568 pub fn evaluate_at_clock(
569 &self,
570 current_term: SupervisorTerm,
571 current_epoch: OwnershipEpoch,
572 now: LeaseClockSample,
573 ) -> OwnerWriteMode {
574 self.evaluate(current_term, current_epoch, now.monotonic_ms())
575 }
576
577 pub fn admit_request(
587 &self,
588 request: RangeRequest,
589 current_term: SupervisorTerm,
590 current_epoch: OwnershipEpoch,
591 now_ms: u64,
592 ) -> Result<(), LeaseFenceRejection> {
593 match self.evaluate(current_term, current_epoch, now_ms) {
594 OwnerWriteMode::Durable => Ok(()),
595 OwnerWriteMode::Fenced(reason) => match request {
596 RangeRequest::StaleRead | RangeRequest::ReplicationCatchUp => Ok(()),
597 RangeRequest::DurableWrite => Err(LeaseFenceRejection { reason }),
598 },
599 }
600 }
601
602 pub fn admit_request_at_clock(
603 &self,
604 request: RangeRequest,
605 current_term: SupervisorTerm,
606 current_epoch: OwnershipEpoch,
607 now: LeaseClockSample,
608 ) -> Result<(), LeaseFenceRejection> {
609 self.admit_request(request, current_term, current_epoch, now.monotonic_ms())
610 }
611}
612
613#[derive(Debug, Clone, PartialEq, Eq)]
616pub enum DurableWriteReject {
617 NoRange { collection: CollectionId },
619 StaleOwnership {
624 collection: CollectionId,
625 range_id: RangeId,
626 attempted_owner: NodeIdentity,
627 current_owner: NodeIdentity,
628 attempted_epoch: OwnershipEpoch,
629 current_epoch: OwnershipEpoch,
630 },
631 NotOwner {
634 collection: CollectionId,
635 range_id: RangeId,
636 role: RangeRole,
637 owner: NodeIdentity,
638 epoch: OwnershipEpoch,
639 version: CatalogVersion,
640 },
641 Fenced {
644 collection: CollectionId,
645 range_id: RangeId,
646 reason: FenceReason,
647 },
648}
649
650impl std::fmt::Display for DurableWriteReject {
651 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
652 match self {
653 Self::NoRange { collection } => write!(
654 f,
655 "no range of collection {collection} covers the routed key — re-resolve routing"
656 ),
657 Self::StaleOwnership {
658 collection,
659 range_id,
660 attempted_owner,
661 current_owner,
662 attempted_epoch,
663 current_epoch,
664 } => write!(
665 f,
666 "stale ownership for {collection}/{range_id}: {attempted_owner} tried epoch {attempted_epoch}, current owner is {current_owner} at epoch {current_epoch}"
667 ),
668 Self::NotOwner {
669 collection,
670 range_id,
671 owner,
672 epoch,
673 version,
674 ..
675 } => write!(
676 f,
677 "this node does not own {collection}/{range_id} — route the durable write to {owner} at epoch {epoch} (catalog version {version})"
678 ),
679 Self::Fenced {
680 collection,
681 range_id,
682 reason,
683 } => write!(
684 f,
685 "owner of {collection}/{range_id} is self-fenced and rejects the durable write: {reason}"
686 ),
687 }
688 }
689}
690
691impl std::error::Error for DurableWriteReject {}
692
693pub fn admit_durable_write<'c>(
708 catalog: &'c ShardOwnershipCatalog,
709 holder: &LeasedOwner,
710 node: &NodeIdentity,
711 collection: &CollectionId,
712 key: &[u8],
713 current_term: SupervisorTerm,
714 now_ms: u64,
715) -> Result<&'c RangeOwnership, DurableWriteReject> {
716 let range =
717 catalog
718 .route_shard_key(collection, key)
719 .ok_or_else(|| DurableWriteReject::NoRange {
720 collection: collection.clone(),
721 })?;
722
723 let role = range.role_of(node);
724 let held_lease = holder.lease().filter(|lease| {
725 lease.collection() == collection
726 && lease.range_id() == range.range_id()
727 && lease.owner() == node
728 });
729 if !role.may_write_public() {
730 if let Some(lease) = held_lease {
731 if lease.epoch() < range.epoch() {
732 return Err(DurableWriteReject::StaleOwnership {
733 collection: collection.clone(),
734 range_id: range.range_id(),
735 attempted_owner: node.clone(),
736 current_owner: range.owner().clone(),
737 attempted_epoch: lease.epoch(),
738 current_epoch: range.epoch(),
739 });
740 }
741 }
742 return Err(DurableWriteReject::NotOwner {
743 collection: collection.clone(),
744 range_id: range.range_id(),
745 role,
746 owner: range.owner().clone(),
747 epoch: range.epoch(),
748 version: range.version(),
749 });
750 }
751
752 let covered = held_lease.is_some();
756
757 let mode = if covered {
758 holder.evaluate(current_term, range.epoch(), now_ms)
759 } else {
760 OwnerWriteMode::Fenced(FenceReason::Unleased)
761 };
762
763 match mode {
764 OwnerWriteMode::Durable => Ok(range),
765 OwnerWriteMode::Fenced(reason) => Err(DurableWriteReject::Fenced {
766 collection: collection.clone(),
767 range_id: range.range_id(),
768 reason,
769 }),
770 }
771}
772
773#[cfg(test)]
774mod tests {
775 use super::*;
776 use crate::cluster::ownership::{PlacementMetadata, RangeBounds, ShardKeyMode};
777
778 fn collection(name: &str) -> CollectionId {
779 CollectionId::new(name).unwrap()
780 }
781
782 fn ident(cn: &str) -> NodeIdentity {
783 NodeIdentity::from_certificate_subject(cn).unwrap()
784 }
785
786 fn catalog_with(owner: &str, replicas: &[&str]) -> (ShardOwnershipCatalog, CollectionId) {
788 let orders = collection("orders");
789 let mut catalog = ShardOwnershipCatalog::new();
790 catalog
791 .apply_update(RangeOwnership::establish(
792 orders.clone(),
793 RangeId::new(1),
794 ShardKeyMode::Hash,
795 RangeBounds::full(),
796 ident(owner),
797 replicas.iter().map(|r| ident(r)).collect::<Vec<_>>(),
798 PlacementMetadata::with_replication_factor(3),
799 ))
800 .unwrap();
801 (catalog, orders)
802 }
803
804 fn next_epoch() -> OwnershipEpoch {
807 RangeOwnership::establish(
808 collection("orders"),
809 RangeId::new(1),
810 ShardKeyMode::Hash,
811 RangeBounds::full(),
812 ident("CN=node-a"),
813 [ident("CN=node-b")],
814 PlacementMetadata::with_replication_factor(3),
815 )
816 .transfer_to(ident("CN=node-b"), [])
817 .epoch()
818 }
819
820 fn lease_for(orders: &CollectionId, owner: &str, ttl_ms: u64) -> OwnershipLease {
823 OwnershipLease::grant(
824 SupervisorTerm::genesis(),
825 orders.clone(),
826 RangeId::new(1),
827 ident(owner),
828 OwnershipEpoch::initial(),
829 0,
830 ttl_ms,
831 )
832 }
833
834 #[test]
839 fn lease_window_is_half_open() {
840 let orders = collection("orders");
841 let lease = lease_for(&orders, "CN=node-a", 1_000);
842 assert_eq!(lease.granted_at_ms(), 0);
843 assert_eq!(lease.expires_at_ms(), 1_000);
844 assert!(!lease.is_expired(0));
845 assert!(!lease.is_expired(999));
846 assert!(lease.is_expired(1_000));
849 assert!(lease.is_expired(1_001));
850 assert_eq!(lease.remaining_ms(250), 750);
851 assert_eq!(lease.remaining_ms(1_000), 0);
852 assert_eq!(lease.remaining_ms(5_000), 0);
853 }
854
855 #[test]
856 fn lease_binds_term_range_owner_and_epoch() {
857 let orders = collection("orders");
858 let lease = lease_for(&orders, "CN=node-a", 1_000);
859 assert_eq!(lease.supervisor_term(), SupervisorTerm::genesis());
860 assert_eq!(lease.collection(), &orders);
861 assert_eq!(lease.range_id(), RangeId::new(1));
862 assert_eq!(lease.owner(), &ident("CN=node-a"));
863 assert_eq!(lease.epoch(), OwnershipEpoch::initial());
864 }
865
866 #[test]
867 fn wall_clock_adjustments_do_not_change_lease_duration() {
868 let orders = collection("orders");
869 let lease = OwnershipLease::grant_at_clock(
870 SupervisorTerm::genesis(),
871 orders,
872 RangeId::new(1),
873 ident("CN=node-a"),
874 OwnershipEpoch::initial(),
875 LeaseClockSample::new(10_000, 1_000_000),
876 1_000,
877 );
878 let owner = LeasedOwner::with_lease(lease);
879
880 let jumped_forward = LeaseClockSample::new(10_999, 86_401_000);
881 assert_eq!(
882 owner.evaluate_at_clock(
883 SupervisorTerm::genesis(),
884 OwnershipEpoch::initial(),
885 jumped_forward
886 ),
887 OwnerWriteMode::Durable
888 );
889
890 let jumped_backward = LeaseClockSample::new(11_000, 1);
891 assert!(matches!(
892 owner.evaluate_at_clock(
893 SupervisorTerm::genesis(),
894 OwnershipEpoch::initial(),
895 jumped_backward
896 ),
897 OwnerWriteMode::Fenced(FenceReason::Expired { .. })
898 ));
899 }
900
901 #[test]
902 fn lease_clock_config_requires_margin_at_least_max_drift() {
903 assert!(OwnershipLeaseClockConfig::new(10_000, 250, 250).is_ok());
904
905 let err = OwnershipLeaseClockConfig::new(10_000, 249, 250).unwrap_err();
906 assert_eq!(
907 err,
908 LeaseClockConfigError::SafetyMarginBelowMaxDrift {
909 safety_margin_ms: 249,
910 max_clock_drift_ms: 250,
911 }
912 );
913 }
914
915 #[test]
920 fn valid_lease_authorises_durable_writes() {
921 let orders = collection("orders");
922 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
923 let mode = owner.evaluate(SupervisorTerm::genesis(), OwnershipEpoch::initial(), 500);
924 assert_eq!(mode, OwnerWriteMode::Durable);
925 assert!(mode.may_write_durable());
926 assert!(!mode.is_fenced());
927 }
928
929 #[test]
930 fn unleased_owner_is_fenced() {
931 let owner = LeasedOwner::unleased();
932 let mode = owner.evaluate(SupervisorTerm::genesis(), OwnershipEpoch::initial(), 0);
933 assert_eq!(mode, OwnerWriteMode::Fenced(FenceReason::Unleased));
934 }
935
936 #[test]
937 fn expired_lease_self_fences() {
938 let orders = collection("orders");
939 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
940 let mode = owner.evaluate(SupervisorTerm::genesis(), OwnershipEpoch::initial(), 1_500);
943 match mode {
944 OwnerWriteMode::Fenced(FenceReason::Expired {
945 now_ms,
946 expires_at_ms,
947 }) => {
948 assert_eq!(now_ms, 1_500);
949 assert_eq!(expires_at_ms, 1_000);
950 }
951 other => panic!("expected Expired fence, got {other:?}"),
952 }
953 }
954
955 #[test]
956 fn epoch_mismatch_self_fences() {
957 let orders = collection("orders");
958 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
960 let current_epoch = next_epoch();
961 let mode = owner.evaluate(SupervisorTerm::genesis(), current_epoch, 500);
962 match mode {
963 OwnerWriteMode::Fenced(FenceReason::EpochSuperseded {
964 lease_epoch,
965 current_epoch: reported,
966 }) => {
967 assert_eq!(lease_epoch, OwnershipEpoch::initial());
968 assert_eq!(reported, current_epoch);
969 }
970 other => panic!("expected EpochSuperseded fence, got {other:?}"),
971 }
972 }
973
974 #[test]
975 fn supervisor_term_advance_self_fences() {
976 let orders = collection("orders");
977 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
978 let current_term = SupervisorTerm::genesis().next();
981 let mode = owner.evaluate(current_term, OwnershipEpoch::initial(), 500);
982 match mode {
983 OwnerWriteMode::Fenced(FenceReason::TermSuperseded {
984 lease_term,
985 current_term: reported,
986 }) => {
987 assert_eq!(lease_term, SupervisorTerm::genesis());
988 assert_eq!(reported, current_term);
989 }
990 other => panic!("expected TermSuperseded fence, got {other:?}"),
991 }
992 }
993
994 #[test]
995 fn revoked_lease_self_fences_before_expiry() {
996 let orders = collection("orders");
997 let mut owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
998 owner.revoke();
999 let mode = owner.evaluate(SupervisorTerm::genesis(), OwnershipEpoch::initial(), 100);
1001 assert_eq!(mode, OwnerWriteMode::Fenced(FenceReason::Revoked));
1002 }
1003
1004 #[test]
1005 fn revoke_takes_precedence_over_other_causes() {
1006 let orders = collection("orders");
1009 let mut owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
1010 owner.revoke();
1011 let mode = owner.evaluate(SupervisorTerm::genesis().next(), next_epoch(), 10_000);
1012 assert_eq!(mode, OwnerWriteMode::Fenced(FenceReason::Revoked));
1013 }
1014
1015 #[test]
1016 fn renewing_a_lease_clears_a_prior_revoke_and_extends_window() {
1017 let orders = collection("orders");
1018 let mut owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
1019 owner.revoke();
1020 assert!(owner
1021 .evaluate(SupervisorTerm::genesis(), OwnershipEpoch::initial(), 100)
1022 .is_fenced());
1023 owner.grant(OwnershipLease::grant(
1026 SupervisorTerm::genesis(),
1027 orders.clone(),
1028 RangeId::new(1),
1029 ident("CN=node-a"),
1030 OwnershipEpoch::initial(),
1031 900,
1032 1_000,
1033 ));
1034 let mode = owner.evaluate(SupervisorTerm::genesis(), OwnershipEpoch::initial(), 1_500);
1035 assert_eq!(mode, OwnerWriteMode::Durable);
1036 }
1037
1038 #[test]
1039 fn stale_renewal_racing_epoch_bump_does_not_restore_authority() {
1040 let orders = collection("orders");
1041 let mut owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
1042 let current_epoch = next_epoch();
1043
1044 owner.grant(OwnershipLease::grant(
1045 SupervisorTerm::genesis(),
1046 orders.clone(),
1047 RangeId::new(1),
1048 ident("CN=node-a"),
1049 OwnershipEpoch::initial(),
1050 900,
1051 1_000,
1052 ));
1053 let mode = owner.evaluate(SupervisorTerm::genesis(), current_epoch, 950);
1054 assert!(matches!(
1055 mode,
1056 OwnerWriteMode::Fenced(FenceReason::EpochSuperseded { .. })
1057 ));
1058
1059 owner.grant(OwnershipLease::grant(
1060 SupervisorTerm::genesis(),
1061 orders,
1062 RangeId::new(1),
1063 ident("CN=node-a"),
1064 current_epoch,
1065 1_000,
1066 1_000,
1067 ));
1068 assert_eq!(
1069 owner.evaluate(SupervisorTerm::genesis(), current_epoch, 1_500),
1070 OwnerWriteMode::Durable
1071 );
1072 }
1073
1074 #[test]
1079 fn valid_lease_admits_every_request_kind() {
1080 let orders = collection("orders");
1081 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
1082 for req in [
1083 RangeRequest::DurableWrite,
1084 RangeRequest::StaleRead,
1085 RangeRequest::ReplicationCatchUp,
1086 ] {
1087 assert!(owner
1088 .admit_request(
1089 req,
1090 SupervisorTerm::genesis(),
1091 OwnershipEpoch::initial(),
1092 500
1093 )
1094 .is_ok());
1095 }
1096 }
1097
1098 #[test]
1099 fn self_fenced_read_mode_serves_reads_and_catch_up_but_rejects_durable_writes() {
1100 let orders = collection("orders");
1101 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
1102 let now = 2_000;
1104 let term = SupervisorTerm::genesis();
1105 let epoch = OwnershipEpoch::initial();
1106
1107 assert!(owner
1109 .admit_request(RangeRequest::StaleRead, term, epoch, now)
1110 .is_ok());
1111 assert!(owner
1112 .admit_request(RangeRequest::ReplicationCatchUp, term, epoch, now)
1113 .is_ok());
1114
1115 let err = owner
1117 .admit_request(RangeRequest::DurableWrite, term, epoch, now)
1118 .unwrap_err();
1119 assert!(matches!(err.reason, FenceReason::Expired { .. }));
1120 assert!(err.to_string().contains("self-fenced"));
1121 }
1122
1123 #[test]
1124 fn unleased_owner_rejects_durable_write_but_still_catches_up() {
1125 let owner = LeasedOwner::unleased();
1126 let term = SupervisorTerm::genesis();
1127 let epoch = OwnershipEpoch::initial();
1128 assert_eq!(
1129 owner
1130 .admit_request(RangeRequest::DurableWrite, term, epoch, 0)
1131 .unwrap_err()
1132 .reason,
1133 FenceReason::Unleased
1134 );
1135 assert!(owner
1138 .admit_request(RangeRequest::ReplicationCatchUp, term, epoch, 0)
1139 .is_ok());
1140 }
1141
1142 #[test]
1147 fn durable_write_admitted_for_leased_owner() {
1148 let (catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
1149 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
1150 let range = admit_durable_write(
1151 &catalog,
1152 &owner,
1153 &ident("CN=node-a"),
1154 &orders,
1155 b"k",
1156 SupervisorTerm::genesis(),
1157 500,
1158 )
1159 .expect("leased owner at current term/epoch may write");
1160 assert_eq!(range.owner(), &ident("CN=node-a"));
1161 assert_eq!(range.range_id(), RangeId::new(1));
1162 }
1163
1164 #[test]
1165 fn durable_write_rejected_for_catalog_owner_without_a_lease() {
1166 let (catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
1169 let owner = LeasedOwner::unleased();
1170 let err = admit_durable_write(
1171 &catalog,
1172 &owner,
1173 &ident("CN=node-a"),
1174 &orders,
1175 b"k",
1176 SupervisorTerm::genesis(),
1177 0,
1178 )
1179 .unwrap_err();
1180 match err {
1181 DurableWriteReject::Fenced { reason, .. } => assert_eq!(reason, FenceReason::Unleased),
1182 other => panic!("expected Fenced(Unleased), got {other:?}"),
1183 }
1184 }
1185
1186 #[test]
1187 fn durable_write_rejected_for_non_owner_before_lease_is_even_consulted() {
1188 let (catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
1189 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-b", 1_000));
1192 let err = admit_durable_write(
1193 &catalog,
1194 &owner,
1195 &ident("CN=node-b"),
1196 &orders,
1197 b"k",
1198 SupervisorTerm::genesis(),
1199 500,
1200 )
1201 .unwrap_err();
1202 match err {
1203 DurableWriteReject::NotOwner { role, owner, .. } => {
1204 assert_eq!(role, RangeRole::Replica);
1205 assert_eq!(owner, ident("CN=node-a"));
1206 }
1207 other => panic!("expected NotOwner, got {other:?}"),
1208 }
1209 }
1210
1211 #[test]
1212 fn durable_write_rejected_when_no_range_covers_the_key() {
1213 let catalog = ShardOwnershipCatalog::new();
1214 let orders = collection("orders");
1215 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
1216 let err = admit_durable_write(
1217 &catalog,
1218 &owner,
1219 &ident("CN=node-a"),
1220 &orders,
1221 b"k",
1222 SupervisorTerm::genesis(),
1223 500,
1224 )
1225 .unwrap_err();
1226 assert!(matches!(err, DurableWriteReject::NoRange { .. }));
1227 }
1228
1229 #[test]
1230 fn durable_write_fenced_when_lease_epoch_trails_the_catalog() {
1231 let (mut catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
1235 let stale_lease = lease_for(&orders, "CN=node-a", 100_000);
1236
1237 let v1 = catalog.range(&orders, RangeId::new(1)).unwrap().clone();
1238 let v2 = v1.transfer_to(ident("CN=node-b"), [ident("CN=node-a")]);
1239 catalog.apply_update(v2.clone()).unwrap();
1240 let v3 = v2.transfer_to(ident("CN=node-a"), [ident("CN=node-b")]);
1241 catalog.apply_update(v3).unwrap();
1242
1243 let owner = LeasedOwner::with_lease(stale_lease);
1244 let current_epoch = catalog.range(&orders, RangeId::new(1)).unwrap().epoch();
1245 assert_eq!(current_epoch.value(), 3);
1246
1247 let err = admit_durable_write(
1248 &catalog,
1249 &owner,
1250 &ident("CN=node-a"),
1251 &orders,
1252 b"k",
1253 SupervisorTerm::genesis(),
1254 500,
1255 )
1256 .unwrap_err();
1257 match err {
1258 DurableWriteReject::Fenced {
1259 reason: FenceReason::EpochSuperseded { lease_epoch, .. },
1260 ..
1261 } => assert_eq!(lease_epoch, OwnershipEpoch::initial()),
1262 other => panic!("expected Fenced(EpochSuperseded), got {other:?}"),
1263 }
1264 }
1265
1266 #[test]
1267 fn stale_owner_rejects_durable_write_after_epoch_bump() {
1268 let (mut catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
1269 let old_owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 100_000));
1270
1271 let v1 = catalog.range(&orders, RangeId::new(1)).unwrap().clone();
1272 let v2 = v1.transfer_to(ident("CN=node-b"), [ident("CN=node-a")]);
1273 catalog.apply_update(v2).unwrap();
1274 let current = catalog.range(&orders, RangeId::new(1)).unwrap();
1275 assert_eq!(current.owner(), &ident("CN=node-b"));
1276 assert_eq!(current.epoch().value(), 2);
1277
1278 let new_owner = LeasedOwner::with_lease(OwnershipLease::grant(
1279 SupervisorTerm::genesis(),
1280 orders.clone(),
1281 RangeId::new(1),
1282 ident("CN=node-b"),
1283 current.epoch(),
1284 0,
1285 100_000,
1286 ));
1287
1288 let err = admit_durable_write(
1289 &catalog,
1290 &old_owner,
1291 &ident("CN=node-a"),
1292 &orders,
1293 b"k",
1294 SupervisorTerm::genesis(),
1295 500,
1296 )
1297 .unwrap_err();
1298 match err {
1299 DurableWriteReject::StaleOwnership {
1300 attempted_owner,
1301 current_owner,
1302 attempted_epoch,
1303 current_epoch,
1304 ..
1305 } => {
1306 assert_eq!(attempted_owner, ident("CN=node-a"));
1307 assert_eq!(current_owner, ident("CN=node-b"));
1308 assert_eq!(attempted_epoch, OwnershipEpoch::initial());
1309 assert_eq!(current_epoch, current.epoch());
1310 }
1311 other => panic!("expected StaleOwnership, got {other:?}"),
1312 }
1313
1314 let admitted = admit_durable_write(
1315 &catalog,
1316 &new_owner,
1317 &ident("CN=node-b"),
1318 &orders,
1319 b"k",
1320 SupervisorTerm::genesis(),
1321 500,
1322 )
1323 .expect("new owner at the current epoch may write");
1324 assert_eq!(admitted.owner(), &ident("CN=node-b"));
1325 assert_eq!(admitted.epoch(), current.epoch());
1326 }
1327
1328 #[test]
1329 fn durable_write_fenced_when_lease_is_for_a_different_range() {
1330 let (catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
1331 let wrong_range_lease = OwnershipLease::grant(
1333 SupervisorTerm::genesis(),
1334 orders.clone(),
1335 RangeId::new(2),
1336 ident("CN=node-a"),
1337 OwnershipEpoch::initial(),
1338 0,
1339 1_000,
1340 );
1341 let owner = LeasedOwner::with_lease(wrong_range_lease);
1342 let err = admit_durable_write(
1343 &catalog,
1344 &owner,
1345 &ident("CN=node-a"),
1346 &orders,
1347 b"k",
1348 SupervisorTerm::genesis(),
1349 500,
1350 )
1351 .unwrap_err();
1352 match err {
1354 DurableWriteReject::Fenced { reason, .. } => assert_eq!(reason, FenceReason::Unleased),
1355 other => panic!("expected Fenced(Unleased), got {other:?}"),
1356 }
1357 }
1358
1359 #[test]
1360 fn durable_write_rejected_after_self_fence_then_restored_on_renewal() {
1361 let (catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
1365 let mut owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
1366 let term = SupervisorTerm::genesis();
1367
1368 assert!(admit_durable_write(
1370 &catalog,
1371 &owner,
1372 &ident("CN=node-a"),
1373 &orders,
1374 b"k",
1375 term,
1376 500
1377 )
1378 .is_ok());
1379 let err = admit_durable_write(
1381 &catalog,
1382 &owner,
1383 &ident("CN=node-a"),
1384 &orders,
1385 b"k",
1386 term,
1387 2_000,
1388 )
1389 .unwrap_err();
1390 assert!(matches!(
1391 err,
1392 DurableWriteReject::Fenced {
1393 reason: FenceReason::Expired { .. },
1394 ..
1395 }
1396 ));
1397 owner.grant(OwnershipLease::grant(
1399 term,
1400 orders.clone(),
1401 RangeId::new(1),
1402 ident("CN=node-a"),
1403 OwnershipEpoch::initial(),
1404 2_000,
1405 1_000,
1406 ));
1407 assert!(admit_durable_write(
1408 &catalog,
1409 &owner,
1410 &ident("CN=node-a"),
1411 &orders,
1412 b"k",
1413 term,
1414 2_500
1415 )
1416 .is_ok());
1417 }
1418}