1use super::identity::NodeIdentity;
71use super::ownership::{
72 CatalogVersion, CollectionId, OwnershipEpoch, RangeId, RangeOwnership, RangeRole,
73 ShardOwnershipCatalog,
74};
75
76#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
84pub struct SupervisorTerm(u64);
85
86impl SupervisorTerm {
87 pub fn genesis() -> Self {
89 Self(1)
90 }
91
92 pub fn new(value: u64) -> Self {
93 Self(value)
94 }
95
96 pub fn value(self) -> u64 {
97 self.0
98 }
99
100 pub fn next(self) -> Self {
102 Self(self.0 + 1)
103 }
104}
105
106impl std::fmt::Display for SupervisorTerm {
107 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
108 write!(f, "{}", self.0)
109 }
110}
111
112#[derive(Debug, Clone, PartialEq, Eq)]
123pub struct OwnershipLease {
124 supervisor_term: SupervisorTerm,
125 collection: CollectionId,
126 range_id: RangeId,
127 owner: NodeIdentity,
128 epoch: OwnershipEpoch,
129 granted_at_ms: u64,
130 expires_at_ms: u64,
131}
132
133impl OwnershipLease {
134 #[allow(clippy::too_many_arguments)]
138 pub fn grant(
139 supervisor_term: SupervisorTerm,
140 collection: CollectionId,
141 range_id: RangeId,
142 owner: NodeIdentity,
143 epoch: OwnershipEpoch,
144 granted_at_ms: u64,
145 ttl_ms: u64,
146 ) -> Self {
147 Self {
148 supervisor_term,
149 collection,
150 range_id,
151 owner,
152 epoch,
153 granted_at_ms,
154 expires_at_ms: granted_at_ms.saturating_add(ttl_ms),
155 }
156 }
157
158 pub fn supervisor_term(&self) -> SupervisorTerm {
159 self.supervisor_term
160 }
161
162 pub fn collection(&self) -> &CollectionId {
163 &self.collection
164 }
165
166 pub fn range_id(&self) -> RangeId {
167 self.range_id
168 }
169
170 pub fn owner(&self) -> &NodeIdentity {
171 &self.owner
172 }
173
174 pub fn epoch(&self) -> OwnershipEpoch {
175 self.epoch
176 }
177
178 pub fn granted_at_ms(&self) -> u64 {
179 self.granted_at_ms
180 }
181
182 pub fn expires_at_ms(&self) -> u64 {
183 self.expires_at_ms
184 }
185
186 pub fn is_expired(&self, now_ms: u64) -> bool {
190 now_ms >= self.expires_at_ms
191 }
192
193 pub fn remaining_ms(&self, now_ms: u64) -> u64 {
196 self.expires_at_ms.saturating_sub(now_ms)
197 }
198
199 fn covers(&self, collection: &CollectionId, range_id: RangeId, owner: &NodeIdentity) -> bool {
203 self.collection == *collection && self.range_id == range_id && self.owner == *owner
204 }
205}
206
207#[derive(Debug, Clone, PartialEq, Eq)]
212pub enum FenceReason {
213 Unleased,
216 Revoked,
219 TermSuperseded {
222 lease_term: SupervisorTerm,
223 current_term: SupervisorTerm,
224 },
225 EpochSuperseded {
228 lease_epoch: OwnershipEpoch,
229 current_epoch: OwnershipEpoch,
230 },
231 Expired { now_ms: u64, expires_at_ms: u64 },
235}
236
237impl std::fmt::Display for FenceReason {
238 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
239 match self {
240 Self::Unleased => write!(f, "owner holds no ownership lease"),
241 Self::Revoked => write!(f, "ownership lease was revoked"),
242 Self::TermSuperseded {
243 lease_term,
244 current_term,
245 } => write!(
246 f,
247 "ownership lease granted under supervisor term {lease_term} is behind current term {current_term}"
248 ),
249 Self::EpochSuperseded {
250 lease_epoch,
251 current_epoch,
252 } => write!(
253 f,
254 "ownership lease epoch {lease_epoch} no longer matches current ownership epoch {current_epoch}"
255 ),
256 Self::Expired {
257 now_ms,
258 expires_at_ms,
259 } => write!(
260 f,
261 "ownership lease expired at {expires_at_ms} ms (now {now_ms} ms)"
262 ),
263 }
264 }
265}
266
267impl std::error::Error for FenceReason {}
268
269#[derive(Debug, Clone, PartialEq, Eq)]
271pub enum OwnerWriteMode {
272 Durable,
274 Fenced(FenceReason),
278}
279
280impl OwnerWriteMode {
281 pub fn may_write_durable(&self) -> bool {
283 matches!(self, OwnerWriteMode::Durable)
284 }
285
286 pub fn is_fenced(&self) -> bool {
288 matches!(self, OwnerWriteMode::Fenced(_))
289 }
290}
291
292#[derive(Debug, Clone, Copy, PartialEq, Eq)]
298pub enum RangeRequest {
299 DurableWrite,
301 StaleRead,
303 ReplicationCatchUp,
307}
308
309#[derive(Debug, Clone, PartialEq, Eq)]
311pub struct LeaseFenceRejection {
312 pub reason: FenceReason,
314}
315
316impl std::fmt::Display for LeaseFenceRejection {
317 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
318 write!(
319 f,
320 "durable write rejected: owner is self-fenced ({})",
321 self.reason
322 )
323 }
324}
325
326impl std::error::Error for LeaseFenceRejection {}
327
328#[derive(Debug, Clone, Default, PartialEq, Eq)]
338pub struct LeasedOwner {
339 lease: Option<OwnershipLease>,
340 revoked: bool,
341}
342
343impl LeasedOwner {
344 pub fn unleased() -> Self {
346 Self {
347 lease: None,
348 revoked: false,
349 }
350 }
351
352 pub fn with_lease(lease: OwnershipLease) -> Self {
354 Self {
355 lease: Some(lease),
356 revoked: false,
357 }
358 }
359
360 pub fn grant(&mut self, lease: OwnershipLease) {
363 self.lease = Some(lease);
364 self.revoked = false;
365 }
366
367 pub fn revoke(&mut self) {
371 self.revoked = true;
372 }
373
374 pub fn lease(&self) -> Option<&OwnershipLease> {
378 self.lease.as_ref()
379 }
380
381 pub fn evaluate(
392 &self,
393 current_term: SupervisorTerm,
394 current_epoch: OwnershipEpoch,
395 now_ms: u64,
396 ) -> OwnerWriteMode {
397 if self.revoked {
398 return OwnerWriteMode::Fenced(FenceReason::Revoked);
399 }
400 let Some(lease) = &self.lease else {
401 return OwnerWriteMode::Fenced(FenceReason::Unleased);
402 };
403 if lease.supervisor_term != current_term {
404 return OwnerWriteMode::Fenced(FenceReason::TermSuperseded {
405 lease_term: lease.supervisor_term,
406 current_term,
407 });
408 }
409 if lease.epoch != current_epoch {
410 return OwnerWriteMode::Fenced(FenceReason::EpochSuperseded {
411 lease_epoch: lease.epoch,
412 current_epoch,
413 });
414 }
415 if lease.is_expired(now_ms) {
416 return OwnerWriteMode::Fenced(FenceReason::Expired {
417 now_ms,
418 expires_at_ms: lease.expires_at_ms,
419 });
420 }
421 OwnerWriteMode::Durable
422 }
423
424 pub fn admit_request(
434 &self,
435 request: RangeRequest,
436 current_term: SupervisorTerm,
437 current_epoch: OwnershipEpoch,
438 now_ms: u64,
439 ) -> Result<(), LeaseFenceRejection> {
440 match self.evaluate(current_term, current_epoch, now_ms) {
441 OwnerWriteMode::Durable => Ok(()),
442 OwnerWriteMode::Fenced(reason) => match request {
443 RangeRequest::StaleRead | RangeRequest::ReplicationCatchUp => Ok(()),
444 RangeRequest::DurableWrite => Err(LeaseFenceRejection { reason }),
445 },
446 }
447 }
448}
449
450#[derive(Debug, Clone, PartialEq, Eq)]
453pub enum DurableWriteReject {
454 NoRange { collection: CollectionId },
456 StaleOwnership {
461 collection: CollectionId,
462 range_id: RangeId,
463 attempted_owner: NodeIdentity,
464 current_owner: NodeIdentity,
465 attempted_epoch: OwnershipEpoch,
466 current_epoch: OwnershipEpoch,
467 },
468 NotOwner {
471 collection: CollectionId,
472 range_id: RangeId,
473 role: RangeRole,
474 owner: NodeIdentity,
475 epoch: OwnershipEpoch,
476 version: CatalogVersion,
477 },
478 Fenced {
481 collection: CollectionId,
482 range_id: RangeId,
483 reason: FenceReason,
484 },
485}
486
487impl std::fmt::Display for DurableWriteReject {
488 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
489 match self {
490 Self::NoRange { collection } => write!(
491 f,
492 "no range of collection {collection} covers the routed key — re-resolve routing"
493 ),
494 Self::StaleOwnership {
495 collection,
496 range_id,
497 attempted_owner,
498 current_owner,
499 attempted_epoch,
500 current_epoch,
501 } => write!(
502 f,
503 "stale ownership for {collection}/{range_id}: {attempted_owner} tried epoch {attempted_epoch}, current owner is {current_owner} at epoch {current_epoch}"
504 ),
505 Self::NotOwner {
506 collection,
507 range_id,
508 owner,
509 epoch,
510 version,
511 ..
512 } => write!(
513 f,
514 "this node does not own {collection}/{range_id} — route the durable write to {owner} at epoch {epoch} (catalog version {version})"
515 ),
516 Self::Fenced {
517 collection,
518 range_id,
519 reason,
520 } => write!(
521 f,
522 "owner of {collection}/{range_id} is self-fenced and rejects the durable write: {reason}"
523 ),
524 }
525 }
526}
527
528impl std::error::Error for DurableWriteReject {}
529
530pub fn admit_durable_write<'c>(
545 catalog: &'c ShardOwnershipCatalog,
546 holder: &LeasedOwner,
547 node: &NodeIdentity,
548 collection: &CollectionId,
549 key: &[u8],
550 current_term: SupervisorTerm,
551 now_ms: u64,
552) -> Result<&'c RangeOwnership, DurableWriteReject> {
553 let range =
554 catalog
555 .route_shard_key(collection, key)
556 .ok_or_else(|| DurableWriteReject::NoRange {
557 collection: collection.clone(),
558 })?;
559
560 let role = range.role_of(node);
561 let held_lease = holder.lease().filter(|lease| {
562 lease.collection() == collection
563 && lease.range_id() == range.range_id()
564 && lease.owner() == node
565 });
566 if !role.may_write_public() {
567 if let Some(lease) = held_lease {
568 if lease.epoch() < range.epoch() {
569 return Err(DurableWriteReject::StaleOwnership {
570 collection: collection.clone(),
571 range_id: range.range_id(),
572 attempted_owner: node.clone(),
573 current_owner: range.owner().clone(),
574 attempted_epoch: lease.epoch(),
575 current_epoch: range.epoch(),
576 });
577 }
578 }
579 return Err(DurableWriteReject::NotOwner {
580 collection: collection.clone(),
581 range_id: range.range_id(),
582 role,
583 owner: range.owner().clone(),
584 epoch: range.epoch(),
585 version: range.version(),
586 });
587 }
588
589 let covered = held_lease.is_some();
593
594 let mode = if covered {
595 holder.evaluate(current_term, range.epoch(), now_ms)
596 } else {
597 OwnerWriteMode::Fenced(FenceReason::Unleased)
598 };
599
600 match mode {
601 OwnerWriteMode::Durable => Ok(range),
602 OwnerWriteMode::Fenced(reason) => Err(DurableWriteReject::Fenced {
603 collection: collection.clone(),
604 range_id: range.range_id(),
605 reason,
606 }),
607 }
608}
609
610#[cfg(test)]
611mod tests {
612 use super::*;
613 use crate::cluster::ownership::{PlacementMetadata, RangeBounds, ShardKeyMode};
614
615 fn collection(name: &str) -> CollectionId {
616 CollectionId::new(name).unwrap()
617 }
618
619 fn ident(cn: &str) -> NodeIdentity {
620 NodeIdentity::from_certificate_subject(cn).unwrap()
621 }
622
623 fn catalog_with(owner: &str, replicas: &[&str]) -> (ShardOwnershipCatalog, CollectionId) {
625 let orders = collection("orders");
626 let mut catalog = ShardOwnershipCatalog::new();
627 catalog
628 .apply_update(RangeOwnership::establish(
629 orders.clone(),
630 RangeId::new(1),
631 ShardKeyMode::Hash,
632 RangeBounds::full(),
633 ident(owner),
634 replicas.iter().map(|r| ident(r)).collect::<Vec<_>>(),
635 PlacementMetadata::with_replication_factor(3),
636 ))
637 .unwrap();
638 (catalog, orders)
639 }
640
641 fn next_epoch() -> OwnershipEpoch {
644 RangeOwnership::establish(
645 collection("orders"),
646 RangeId::new(1),
647 ShardKeyMode::Hash,
648 RangeBounds::full(),
649 ident("CN=node-a"),
650 [ident("CN=node-b")],
651 PlacementMetadata::with_replication_factor(3),
652 )
653 .transfer_to(ident("CN=node-b"), [])
654 .epoch()
655 }
656
657 fn lease_for(orders: &CollectionId, owner: &str, ttl_ms: u64) -> OwnershipLease {
660 OwnershipLease::grant(
661 SupervisorTerm::genesis(),
662 orders.clone(),
663 RangeId::new(1),
664 ident(owner),
665 OwnershipEpoch::initial(),
666 0,
667 ttl_ms,
668 )
669 }
670
671 #[test]
676 fn lease_window_is_half_open() {
677 let orders = collection("orders");
678 let lease = lease_for(&orders, "CN=node-a", 1_000);
679 assert_eq!(lease.granted_at_ms(), 0);
680 assert_eq!(lease.expires_at_ms(), 1_000);
681 assert!(!lease.is_expired(0));
682 assert!(!lease.is_expired(999));
683 assert!(lease.is_expired(1_000));
686 assert!(lease.is_expired(1_001));
687 assert_eq!(lease.remaining_ms(250), 750);
688 assert_eq!(lease.remaining_ms(1_000), 0);
689 assert_eq!(lease.remaining_ms(5_000), 0);
690 }
691
692 #[test]
693 fn lease_binds_term_range_owner_and_epoch() {
694 let orders = collection("orders");
695 let lease = lease_for(&orders, "CN=node-a", 1_000);
696 assert_eq!(lease.supervisor_term(), SupervisorTerm::genesis());
697 assert_eq!(lease.collection(), &orders);
698 assert_eq!(lease.range_id(), RangeId::new(1));
699 assert_eq!(lease.owner(), &ident("CN=node-a"));
700 assert_eq!(lease.epoch(), OwnershipEpoch::initial());
701 }
702
703 #[test]
708 fn valid_lease_authorises_durable_writes() {
709 let orders = collection("orders");
710 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
711 let mode = owner.evaluate(SupervisorTerm::genesis(), OwnershipEpoch::initial(), 500);
712 assert_eq!(mode, OwnerWriteMode::Durable);
713 assert!(mode.may_write_durable());
714 assert!(!mode.is_fenced());
715 }
716
717 #[test]
718 fn unleased_owner_is_fenced() {
719 let owner = LeasedOwner::unleased();
720 let mode = owner.evaluate(SupervisorTerm::genesis(), OwnershipEpoch::initial(), 0);
721 assert_eq!(mode, OwnerWriteMode::Fenced(FenceReason::Unleased));
722 }
723
724 #[test]
725 fn expired_lease_self_fences() {
726 let orders = collection("orders");
727 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
728 let mode = owner.evaluate(SupervisorTerm::genesis(), OwnershipEpoch::initial(), 1_500);
731 match mode {
732 OwnerWriteMode::Fenced(FenceReason::Expired {
733 now_ms,
734 expires_at_ms,
735 }) => {
736 assert_eq!(now_ms, 1_500);
737 assert_eq!(expires_at_ms, 1_000);
738 }
739 other => panic!("expected Expired fence, got {other:?}"),
740 }
741 }
742
743 #[test]
744 fn epoch_mismatch_self_fences() {
745 let orders = collection("orders");
746 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
748 let current_epoch = next_epoch();
749 let mode = owner.evaluate(SupervisorTerm::genesis(), current_epoch, 500);
750 match mode {
751 OwnerWriteMode::Fenced(FenceReason::EpochSuperseded {
752 lease_epoch,
753 current_epoch: reported,
754 }) => {
755 assert_eq!(lease_epoch, OwnershipEpoch::initial());
756 assert_eq!(reported, current_epoch);
757 }
758 other => panic!("expected EpochSuperseded fence, got {other:?}"),
759 }
760 }
761
762 #[test]
763 fn supervisor_term_advance_self_fences() {
764 let orders = collection("orders");
765 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
766 let current_term = SupervisorTerm::genesis().next();
769 let mode = owner.evaluate(current_term, OwnershipEpoch::initial(), 500);
770 match mode {
771 OwnerWriteMode::Fenced(FenceReason::TermSuperseded {
772 lease_term,
773 current_term: reported,
774 }) => {
775 assert_eq!(lease_term, SupervisorTerm::genesis());
776 assert_eq!(reported, current_term);
777 }
778 other => panic!("expected TermSuperseded fence, got {other:?}"),
779 }
780 }
781
782 #[test]
783 fn revoked_lease_self_fences_before_expiry() {
784 let orders = collection("orders");
785 let mut owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
786 owner.revoke();
787 let mode = owner.evaluate(SupervisorTerm::genesis(), OwnershipEpoch::initial(), 100);
789 assert_eq!(mode, OwnerWriteMode::Fenced(FenceReason::Revoked));
790 }
791
792 #[test]
793 fn revoke_takes_precedence_over_other_causes() {
794 let orders = collection("orders");
797 let mut owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
798 owner.revoke();
799 let mode = owner.evaluate(SupervisorTerm::genesis().next(), next_epoch(), 10_000);
800 assert_eq!(mode, OwnerWriteMode::Fenced(FenceReason::Revoked));
801 }
802
803 #[test]
804 fn renewing_a_lease_clears_a_prior_revoke_and_extends_window() {
805 let orders = collection("orders");
806 let mut owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
807 owner.revoke();
808 assert!(owner
809 .evaluate(SupervisorTerm::genesis(), OwnershipEpoch::initial(), 100)
810 .is_fenced());
811 owner.grant(OwnershipLease::grant(
814 SupervisorTerm::genesis(),
815 orders.clone(),
816 RangeId::new(1),
817 ident("CN=node-a"),
818 OwnershipEpoch::initial(),
819 900,
820 1_000,
821 ));
822 let mode = owner.evaluate(SupervisorTerm::genesis(), OwnershipEpoch::initial(), 1_500);
823 assert_eq!(mode, OwnerWriteMode::Durable);
824 }
825
826 #[test]
831 fn valid_lease_admits_every_request_kind() {
832 let orders = collection("orders");
833 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
834 for req in [
835 RangeRequest::DurableWrite,
836 RangeRequest::StaleRead,
837 RangeRequest::ReplicationCatchUp,
838 ] {
839 assert!(owner
840 .admit_request(
841 req,
842 SupervisorTerm::genesis(),
843 OwnershipEpoch::initial(),
844 500
845 )
846 .is_ok());
847 }
848 }
849
850 #[test]
851 fn self_fenced_read_mode_serves_reads_and_catch_up_but_rejects_durable_writes() {
852 let orders = collection("orders");
853 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
854 let now = 2_000;
856 let term = SupervisorTerm::genesis();
857 let epoch = OwnershipEpoch::initial();
858
859 assert!(owner
861 .admit_request(RangeRequest::StaleRead, term, epoch, now)
862 .is_ok());
863 assert!(owner
864 .admit_request(RangeRequest::ReplicationCatchUp, term, epoch, now)
865 .is_ok());
866
867 let err = owner
869 .admit_request(RangeRequest::DurableWrite, term, epoch, now)
870 .unwrap_err();
871 assert!(matches!(err.reason, FenceReason::Expired { .. }));
872 assert!(err.to_string().contains("self-fenced"));
873 }
874
875 #[test]
876 fn unleased_owner_rejects_durable_write_but_still_catches_up() {
877 let owner = LeasedOwner::unleased();
878 let term = SupervisorTerm::genesis();
879 let epoch = OwnershipEpoch::initial();
880 assert_eq!(
881 owner
882 .admit_request(RangeRequest::DurableWrite, term, epoch, 0)
883 .unwrap_err()
884 .reason,
885 FenceReason::Unleased
886 );
887 assert!(owner
890 .admit_request(RangeRequest::ReplicationCatchUp, term, epoch, 0)
891 .is_ok());
892 }
893
894 #[test]
899 fn durable_write_admitted_for_leased_owner() {
900 let (catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
901 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
902 let range = admit_durable_write(
903 &catalog,
904 &owner,
905 &ident("CN=node-a"),
906 &orders,
907 b"k",
908 SupervisorTerm::genesis(),
909 500,
910 )
911 .expect("leased owner at current term/epoch may write");
912 assert_eq!(range.owner(), &ident("CN=node-a"));
913 assert_eq!(range.range_id(), RangeId::new(1));
914 }
915
916 #[test]
917 fn durable_write_rejected_for_catalog_owner_without_a_lease() {
918 let (catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
921 let owner = LeasedOwner::unleased();
922 let err = admit_durable_write(
923 &catalog,
924 &owner,
925 &ident("CN=node-a"),
926 &orders,
927 b"k",
928 SupervisorTerm::genesis(),
929 0,
930 )
931 .unwrap_err();
932 match err {
933 DurableWriteReject::Fenced { reason, .. } => assert_eq!(reason, FenceReason::Unleased),
934 other => panic!("expected Fenced(Unleased), got {other:?}"),
935 }
936 }
937
938 #[test]
939 fn durable_write_rejected_for_non_owner_before_lease_is_even_consulted() {
940 let (catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
941 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-b", 1_000));
944 let err = admit_durable_write(
945 &catalog,
946 &owner,
947 &ident("CN=node-b"),
948 &orders,
949 b"k",
950 SupervisorTerm::genesis(),
951 500,
952 )
953 .unwrap_err();
954 match err {
955 DurableWriteReject::NotOwner { role, owner, .. } => {
956 assert_eq!(role, RangeRole::Replica);
957 assert_eq!(owner, ident("CN=node-a"));
958 }
959 other => panic!("expected NotOwner, got {other:?}"),
960 }
961 }
962
963 #[test]
964 fn durable_write_rejected_when_no_range_covers_the_key() {
965 let catalog = ShardOwnershipCatalog::new();
966 let orders = collection("orders");
967 let owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
968 let err = admit_durable_write(
969 &catalog,
970 &owner,
971 &ident("CN=node-a"),
972 &orders,
973 b"k",
974 SupervisorTerm::genesis(),
975 500,
976 )
977 .unwrap_err();
978 assert!(matches!(err, DurableWriteReject::NoRange { .. }));
979 }
980
981 #[test]
982 fn durable_write_fenced_when_lease_epoch_trails_the_catalog() {
983 let (mut catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
987 let stale_lease = lease_for(&orders, "CN=node-a", 100_000);
988
989 let v1 = catalog.range(&orders, RangeId::new(1)).unwrap().clone();
990 let v2 = v1.transfer_to(ident("CN=node-b"), [ident("CN=node-a")]);
991 catalog.apply_update(v2.clone()).unwrap();
992 let v3 = v2.transfer_to(ident("CN=node-a"), [ident("CN=node-b")]);
993 catalog.apply_update(v3).unwrap();
994
995 let owner = LeasedOwner::with_lease(stale_lease);
996 let current_epoch = catalog.range(&orders, RangeId::new(1)).unwrap().epoch();
997 assert_eq!(current_epoch.value(), 3);
998
999 let err = admit_durable_write(
1000 &catalog,
1001 &owner,
1002 &ident("CN=node-a"),
1003 &orders,
1004 b"k",
1005 SupervisorTerm::genesis(),
1006 500,
1007 )
1008 .unwrap_err();
1009 match err {
1010 DurableWriteReject::Fenced {
1011 reason: FenceReason::EpochSuperseded { lease_epoch, .. },
1012 ..
1013 } => assert_eq!(lease_epoch, OwnershipEpoch::initial()),
1014 other => panic!("expected Fenced(EpochSuperseded), got {other:?}"),
1015 }
1016 }
1017
1018 #[test]
1019 fn stale_owner_rejects_durable_write_after_epoch_bump() {
1020 let (mut catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
1021 let old_owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 100_000));
1022
1023 let v1 = catalog.range(&orders, RangeId::new(1)).unwrap().clone();
1024 let v2 = v1.transfer_to(ident("CN=node-b"), [ident("CN=node-a")]);
1025 catalog.apply_update(v2).unwrap();
1026 let current = catalog.range(&orders, RangeId::new(1)).unwrap();
1027 assert_eq!(current.owner(), &ident("CN=node-b"));
1028 assert_eq!(current.epoch().value(), 2);
1029
1030 let new_owner = LeasedOwner::with_lease(OwnershipLease::grant(
1031 SupervisorTerm::genesis(),
1032 orders.clone(),
1033 RangeId::new(1),
1034 ident("CN=node-b"),
1035 current.epoch(),
1036 0,
1037 100_000,
1038 ));
1039
1040 let err = admit_durable_write(
1041 &catalog,
1042 &old_owner,
1043 &ident("CN=node-a"),
1044 &orders,
1045 b"k",
1046 SupervisorTerm::genesis(),
1047 500,
1048 )
1049 .unwrap_err();
1050 match err {
1051 DurableWriteReject::StaleOwnership {
1052 attempted_owner,
1053 current_owner,
1054 attempted_epoch,
1055 current_epoch,
1056 ..
1057 } => {
1058 assert_eq!(attempted_owner, ident("CN=node-a"));
1059 assert_eq!(current_owner, ident("CN=node-b"));
1060 assert_eq!(attempted_epoch, OwnershipEpoch::initial());
1061 assert_eq!(current_epoch, current.epoch());
1062 }
1063 other => panic!("expected StaleOwnership, got {other:?}"),
1064 }
1065
1066 let admitted = admit_durable_write(
1067 &catalog,
1068 &new_owner,
1069 &ident("CN=node-b"),
1070 &orders,
1071 b"k",
1072 SupervisorTerm::genesis(),
1073 500,
1074 )
1075 .expect("new owner at the current epoch may write");
1076 assert_eq!(admitted.owner(), &ident("CN=node-b"));
1077 assert_eq!(admitted.epoch(), current.epoch());
1078 }
1079
1080 #[test]
1081 fn durable_write_fenced_when_lease_is_for_a_different_range() {
1082 let (catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
1083 let wrong_range_lease = OwnershipLease::grant(
1085 SupervisorTerm::genesis(),
1086 orders.clone(),
1087 RangeId::new(2),
1088 ident("CN=node-a"),
1089 OwnershipEpoch::initial(),
1090 0,
1091 1_000,
1092 );
1093 let owner = LeasedOwner::with_lease(wrong_range_lease);
1094 let err = admit_durable_write(
1095 &catalog,
1096 &owner,
1097 &ident("CN=node-a"),
1098 &orders,
1099 b"k",
1100 SupervisorTerm::genesis(),
1101 500,
1102 )
1103 .unwrap_err();
1104 match err {
1106 DurableWriteReject::Fenced { reason, .. } => assert_eq!(reason, FenceReason::Unleased),
1107 other => panic!("expected Fenced(Unleased), got {other:?}"),
1108 }
1109 }
1110
1111 #[test]
1112 fn durable_write_rejected_after_self_fence_then_restored_on_renewal() {
1113 let (catalog, orders) = catalog_with("CN=node-a", &["CN=node-b"]);
1117 let mut owner = LeasedOwner::with_lease(lease_for(&orders, "CN=node-a", 1_000));
1118 let term = SupervisorTerm::genesis();
1119
1120 assert!(admit_durable_write(
1122 &catalog,
1123 &owner,
1124 &ident("CN=node-a"),
1125 &orders,
1126 b"k",
1127 term,
1128 500
1129 )
1130 .is_ok());
1131 let err = admit_durable_write(
1133 &catalog,
1134 &owner,
1135 &ident("CN=node-a"),
1136 &orders,
1137 b"k",
1138 term,
1139 2_000,
1140 )
1141 .unwrap_err();
1142 assert!(matches!(
1143 err,
1144 DurableWriteReject::Fenced {
1145 reason: FenceReason::Expired { .. },
1146 ..
1147 }
1148 ));
1149 owner.grant(OwnershipLease::grant(
1151 term,
1152 orders.clone(),
1153 RangeId::new(1),
1154 ident("CN=node-a"),
1155 OwnershipEpoch::initial(),
1156 2_000,
1157 1_000,
1158 ));
1159 assert!(admit_durable_write(
1160 &catalog,
1161 &owner,
1162 &ident("CN=node-a"),
1163 &orders,
1164 b"k",
1165 term,
1166 2_500
1167 )
1168 .is_ok());
1169 }
1170}