Skip to main content

reddb_server/cluster/
ownership.rs

1//! The global shard ownership catalog (issue #989, PRD #987, ADR 0037).
2//!
3//! The shard ownership catalog is the **source of truth for range routing and
4//! failover** in a multi-writer cluster. Per the glossary it is *"explicit,
5//! versioned RedDB catalog state that records shard/range bounds, current writer
6//! owner, replicas, and ownership epoch/version"* and — crucially — it is
7//! *"special global control-plane state replicated to all data members rather
8//! than sharded like ordinary user collections"*.
9//!
10//! That last point shapes the whole module. The catalog is the thing that tells
11//! you *where* a collection's data lives, so it cannot itself be located by the
12//! same user-data sharding it describes — that would be circular. Instead every
13//! data member holds a full [`ShardOwnershipCatalog`] replica and routes against
14//! it locally ([`ShardOwnershipCatalog::route`]). Replication is modelled as
15//! shipping versioned [`RangeOwnership`] entries that each member applies through
16//! the same [`apply_update`](ShardOwnershipCatalog::apply_update) path that the
17//! Supervisor leader writes through — and that path rejects stale versions, so a
18//! late or out-of-order replica update can never overwrite newer ownership.
19//!
20//! ## What the catalog records
21//!
22//! One [`RangeOwnership`] entry per owned shard/range carries everything routing
23//! and fencing need:
24//!
25//! * [`CollectionId`] + [`RangeId`] — *which* range of *which* collection.
26//! * [`ShardKeyMode`] — [`Hash`](ShardKeyMode::Hash) (the default, for uniform
27//!   distribution) or [`Ordered`](ShardKeyMode::Ordered) (declared when range
28//!   locality and ordered scans matter more than hotspot resistance).
29//! * [`RangeBounds`] — the half-open `[lower, upper)` partition this entry owns.
30//! * `owner` ([`NodeIdentity`]) + `replicas` — the current single writer for the
31//!   range and its read/catch-up copies.
32//! * [`OwnershipEpoch`] + [`CatalogVersion`] — the fencing epoch (bumped on owner
33//!   change so a stale old owner is fenced) and the monotonic write version
34//!   (bumped on *every* accepted update so stale writes are rejected).
35//! * [`PlacementMetadata`] — replication factor and free-form placement
36//!   attributes (region/zone/weight) the rebalancer reads.
37//!
38//! Ownership changes are produced as *transitions* — new entries built with
39//! [`RangeOwnership::transfer_to`] / [`update_replicas`](RangeOwnership::update_replicas)
40//! / [`update_placement`](RangeOwnership::update_placement) — never arbitrary row
41//! edits, matching ADR 0037's "transitions, not arbitrary row edits".
42//!
43//! Everything here is a pure data model with no I/O, so the routing, versioning,
44//! and replication story is exercised deterministically.
45
46use std::collections::{BTreeMap, BTreeSet};
47
48use super::identity::NodeIdentity;
49use super::slot::hash_shard_key_to_range_key;
50
51/// A user collection's stable identity, as recorded in the catalog.
52///
53/// The catalog is keyed by collection (and range within it); this is the
54/// collection's own name, not a shard-routed handle. Resolving a
55/// [`CollectionId`] to its ranges needs only the catalog itself — no user-data
56/// sharding — which is what lets the catalog be the thing that *bootstraps*
57/// routing.
58#[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    /// Build a collection id from a non-empty name.
74    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/// Stable identity for the collection group that one Placement Authority owns.
94///
95/// A group is the control-plane sharding unit for placement decisions. Related
96/// collections can share a group so they stay in one authority domain; a large
97/// collection can instead be the group's only member when it needs isolated
98/// placement policy.
99#[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    /// Build a collection group id from a non-empty name.
115    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/// One Placement Authority assignment for a non-overlapping collection-catalog
139/// slice.
140#[derive(Debug, Clone, PartialEq, Eq)]
141pub struct CollectionGroupAuthority {
142    group: CollectionGroupId,
143    authority: NodeIdentity,
144    collections: BTreeSet<CollectionId>,
145}
146
147impl CollectionGroupAuthority {
148    /// Assign `collections` to one group owned by `authority`.
149    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/// Why a Placement Authority assignment was rejected.
186#[derive(Debug, Clone, PartialEq, Eq)]
187pub enum PlacementAuthorityError {
188    /// A group assignment must publish at least one collection in its catalog
189    /// slice.
190    EmptyCatalogSlice { group: CollectionGroupId },
191    /// The group already has exactly one Placement Authority.
192    DuplicateCollectionGroup { group: CollectionGroupId },
193    /// Two Placement Authority assignments would own the same collection-catalog
194    /// slice.
195    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/// Placement Authority assignments keyed by Collection group.
229#[derive(Debug, Clone, Default)]
230pub struct PlacementAuthorityCatalog {
231    assignments: BTreeMap<CollectionGroupId, CollectionGroupAuthority>,
232}
233
234impl PlacementAuthorityCatalog {
235    /// An empty Placement Authority catalog.
236    pub fn new() -> Self {
237        Self::default()
238    }
239
240    /// Publish one group-owned, non-overlapping collection-catalog slice.
241    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/// Stable identity of the Placement Authority that publishes a collection group.
282#[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/// Stable identifier for one shard/range within a collection.
321///
322/// Ranges are owned at sub-collection granularity (ADR 0037), so a collection
323/// can have many of these — each is one independently-owned, independently-routed
324/// partition. The id is stable across ownership transitions: moving a range to a
325/// new owner keeps its [`RangeId`] and bumps its epoch/version, it does not mint
326/// a new range.
327#[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/// Collection-level partitioning mode for shard/range ownership.
347///
348/// Per the glossary, *"Hash mode is the default for uniform distribution;
349/// ordered mode is declared when range locality and ordered scans matter more
350/// than automatic hotspot resistance."* The mode is fixed per collection: every
351/// range of a collection shares its mode, and an entry whose mode disagrees with
352/// its collection's declared mode is rejected
353/// ([`CatalogError::ShardKeyModeMismatch`]).
354#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
355pub enum ShardKeyMode {
356    /// Uniform hash distribution — the default. Bounds are over hash tokens.
357    #[default]
358    Hash,
359    /// Ordered key ranges, declared for range locality / ordered scans. Bounds
360    /// are over the ordered shard key itself.
361    Ordered,
362}
363
364/// One edge of a [`RangeBounds`].
365///
366/// Bounds are byte strings so the same type serves both shard key modes: a
367/// [`Hash`](ShardKeyMode::Hash) range bounds hash-token bytes, an
368/// [`Ordered`](ShardKeyMode::Ordered) range bounds the ordered key bytes
369/// directly. [`Min`](RangeBound::Min)/[`Max`](RangeBound::Max) are the open ends
370/// of the keyspace.
371#[derive(Debug, Clone, PartialEq, Eq)]
372pub enum RangeBound {
373    /// The open low end of the keyspace (everything is `>= Min`).
374    Min,
375    /// A concrete boundary key.
376    Key(Vec<u8>),
377    /// The open high end of the keyspace (everything is `< Max`).
378    Max,
379}
380
381impl RangeBound {
382    /// A boundary at the given key bytes.
383    pub fn key(bytes: impl Into<Vec<u8>>) -> Self {
384        RangeBound::Key(bytes.into())
385    }
386
387    /// Total order over keyspace positions: `Min < every Key < Max`, with keys
388    /// compared lexicographically. This is what makes both `contains` and
389    /// `overlaps` plain comparisons.
390    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/// The half-open `[lower, upper)` partition a range owns.
407///
408/// Half-open bounds tile the keyspace without gaps or double-cover: adjacent
409/// ranges share a boundary key that belongs to exactly one of them. `lower` must
410/// be strictly below `upper`, so an empty or inverted range cannot be
411/// constructed.
412#[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    /// Bounds from an explicit `[lower, upper)` pair. Errors if the range would
434    /// be empty or inverted (`lower >= upper`).
435    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    /// The whole keyspace, `[Min, Max)` — a single range covering a collection.
443    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    /// Does `key` fall inside this half-open range? Lower bound inclusive, upper
459    /// bound exclusive — the routing predicate.
460    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    /// Do these two ranges share any key? Used to keep a collection's ranges
466    /// non-overlapping so routing resolves to exactly one owner.
467    pub fn overlaps(&self, other: &RangeBounds) -> bool {
468        self.lower.position() < other.upper.position()
469            && other.lower.position() < self.upper.position()
470    }
471
472    /// Split this `[lower, upper)` range at `at` into a lower child
473    /// `[lower, at)` and an upper child `[at, upper)`. The split point must fall
474    /// **strictly inside** the range (`lower < at < upper`); a point at or
475    /// outside a bound would carve off an empty child and is rejected with
476    /// [`RangeBoundsError`]. The two children tile the original exactly — no gap,
477    /// no overlap — which is what lets a [split-and-move](super::move_range) shrink
478    /// the retained child and create the moved child without making routing
479    /// ambiguous.
480    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/// Monotonic write version of a single catalog entry.
498///
499/// Every accepted update to a range bumps its version. An update that does not
500/// strictly advance the version is stale and is rejected — this is the
501/// compare-and-advance rule that makes catalog replication safe regardless of
502/// delivery order.
503#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
504pub struct CatalogVersion(u64);
505
506impl CatalogVersion {
507    /// The version a range is created at.
508    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/// Fencing epoch for a range's write authority.
528///
529/// Distinct from [`CatalogVersion`]: the version advances on *any* catalog edit,
530/// but the epoch advances only when **write authority moves** (a new owner). A
531/// WAL/logical record stamped with an epoch older than the catalog's current
532/// epoch is from a fenced old owner and must be rejected (ADR 0037, "fencing is
533/// enforced below routing").
534#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
535pub struct OwnershipEpoch(u64);
536
537impl OwnershipEpoch {
538    /// The epoch a range is created at.
539    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/// Placement metadata the rebalancer reads when planning transitions.
559///
560/// The MVP carries the range's replication factor plus a free-form attribute map
561/// for region/zone/operator-weight hints. It is descriptive control-plane data,
562/// not an authorization source.
563#[derive(Debug, Clone, Default, PartialEq, Eq)]
564pub struct PlacementMetadata {
565    replication_factor: usize,
566    attributes: BTreeMap<String, String>,
567}
568
569impl PlacementMetadata {
570    /// Placement with a target replication factor and no attributes.
571    pub fn with_replication_factor(replication_factor: usize) -> Self {
572        Self {
573            replication_factor,
574            attributes: BTreeMap::new(),
575        }
576    }
577
578    /// Attach a placement attribute (e.g. `region` → `us-east-1`).
579    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/// One owned shard/range: the catalog's unit of routing and fencing.
594///
595/// An entry is self-describing — it carries its collection, range id, mode,
596/// bounds, owner, replicas, epoch, version, and placement — so a data member
597/// that receives it by replication can route and fence without consulting any
598/// other state. New ownership states are produced as *transitions* off an
599/// existing entry ([`transfer_to`](Self::transfer_to),
600/// [`update_replicas`](Self::update_replicas),
601/// [`update_placement`](Self::update_placement)), each of which advances the
602/// version — and, for an owner change, the fencing epoch.
603#[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    /// The initial ownership state for a freshly created range: version and
669    /// epoch both at their [`initial`](CatalogVersion::initial) values.
670    #[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    /// Promotable hot mirrors for this range. These are caught-up copies that
730    /// may become owner through a fenced promotion; they are not write
731    /// authorities while listed here.
732    pub fn hot_mirror_replicas(&self) -> &[NodeIdentity] {
733        &self.replicas
734    }
735
736    /// Restore-only compressed archive replicas. These copies are retained for
737    /// recovery and must not be treated as promotion candidates until a later
738    /// validation step proves them usable.
739    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    /// The catalog key for this entry: `(collection, range_id)`.
756    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    /// A transition that moves write authority to `new_owner` with `new_replicas`.
771    /// Advances **both** the version (it is a catalog write) and the ownership
772    /// epoch (write authority moved, so any old owner is fenced).
773    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    /// Promote a hot mirror only if it is a current replica and its durable
788    /// frontier covers the range commit watermark. The accepted path delegates
789    /// to [`transfer_to`](Self::transfer_to), so publishing the promotion bumps
790    /// the ownership epoch before the new owner can pass the write gate.
791    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    /// A transition that changes only the replica set. Advances the version but
825    /// **not** the epoch: write authority did not move, so no owner is fenced.
826    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    /// A transition that changes the restore-only archive replica set. Advances
835    /// the version but not the epoch because write authority did not move.
836    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    /// A transition that changes both non-owner replica roles together.
848    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    /// A transition that changes only placement metadata. Advances the version
862    /// but not the epoch.
863    pub fn update_placement(&self, placement: PlacementMetadata) -> Self {
864        Self {
865            placement,
866            version: self.version.next(),
867            ..self.clone()
868        }
869    }
870
871    /// A transition that **re-bounds** the range without moving write authority —
872    /// the retained-child step of a range split, which narrows this entry to the
873    /// keys its owner keeps while a sibling entry takes the carved-off subrange.
874    /// Advances the version but **not** the epoch: the same owner keeps writing
875    /// the retained keys, so no one is fenced.
876    pub fn with_bounds(&self, bounds: RangeBounds) -> Self {
877        Self {
878            bounds,
879            version: self.version.next(),
880            ..self.clone()
881        }
882    }
883
884    /// This node's [`RangeRole`] for *this* range (issue #990).
885    ///
886    /// A data member is the single writer ([`Owner`](RangeRole::Owner)) of a
887    /// range, holds a read/catch-up copy ([`Replica`](RangeRole::Replica)), or
888    /// holds no copy at all ([`NoCopy`](RangeRole::NoCopy)). The role is
889    /// per-range, not a global node role: the same node can be owner of one
890    /// range, replica of another, and uninvolved in a third — which is why this
891    /// is the input to the ownership-aware public-write gate rather than the
892    /// instance-wide [`WriteGate`](crate::runtime::write_gate::WriteGate).
893    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    /// This node's non-owner replica role for this range, if it has one.
910    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/// The explicit role of a non-owner range replica.
947#[derive(Debug, Clone, Copy, PartialEq, Eq)]
948pub enum ReplicaRole {
949    /// A caught-up copy that can be considered for fenced promotion. It is not a
950    /// write authority until promotion installs it as owner.
951    HotMirror,
952    /// A compressed restore copy. It is restore-only until separate validation
953    /// proves it can safely rejoin the hot set.
954    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/// A data member's role for one specific range (issue #990, PRD #987).
968///
969/// Distinguishes the three positions a node can hold relative to a range, which
970/// the ownership-aware write gate
971/// ([`admit_public_write`](ShardOwnershipCatalog::admit_public_write)) turns
972/// into an allow/reject decision: only the current [`Owner`](Self::Owner) may
973/// take a *public* write for the range; a [`Replica`](Self::Replica) and a
974/// [`NoCopy`](Self::NoCopy) node both reject it and the caller must route to the
975/// owner. (A replica still applies the owner's changes through the privileged
976/// internal apply path — that path is gated by the range-authority fence from
977/// issue #991, not by this public gate.)
978#[derive(Debug, Clone, Copy, PartialEq, Eq)]
979pub enum RangeRole {
980    /// The current single writer for the range — the only role a public write
981    /// may land on.
982    Owner,
983    /// Holds a read/catch-up copy but is not the writer. Public writes are
984    /// rejected and routed to the owner; replicated changes still flow in via
985    /// the privileged internal apply path.
986    Replica,
987    /// Holds a compressed archive copy. Archive replicas are recovery sources
988    /// only after restore, checksum validation, and watermark validation.
989    ArchiveReplica,
990    /// Holds no copy of the range at all.
991    NoCopy,
992}
993
994impl RangeRole {
995    /// Whether this role may accept a *public* write for the range. Only the
996    /// owner may; replica and no-copy may not.
997    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/// Whether an accepted update created a new range or advanced an existing one.
1016#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1017pub enum UpdateOutcome {
1018    /// The range did not exist and was created at this version.
1019    Created,
1020    /// An existing range advanced to a newer version.
1021    Updated,
1022}
1023
1024/// Why a catalog update was rejected.
1025#[derive(Debug, Clone, PartialEq, Eq)]
1026pub enum CatalogError {
1027    /// The update's version did not strictly advance the range's current
1028    /// version — a stale or out-of-order write. Carries both versions so the
1029    /// caller (or a replica) can see how far behind it was.
1030    StaleVersion {
1031        collection: CollectionId,
1032        range_id: RangeId,
1033        current: CatalogVersion,
1034        attempted: CatalogVersion,
1035    },
1036    /// The entry's shard key mode disagrees with the collection's declared mode.
1037    /// A collection is hash- *or* ordered-partitioned, never both.
1038    ShardKeyModeMismatch {
1039        collection: CollectionId,
1040        declared: ShardKeyMode,
1041        attempted: ShardKeyMode,
1042    },
1043    /// Creating this range would overlap an existing range of the same
1044    /// collection, which would make routing ambiguous.
1045    OverlappingRange {
1046        collection: CollectionId,
1047        existing: RangeId,
1048        attempted: RangeId,
1049    },
1050    /// The owner was also listed as a non-owner replica, or one node appeared in
1051    /// more than one replica role. That would blur the single-writer invariant.
1052    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/// Why an ownership-aware *public* write was rejected (issue #990).
1100///
1101/// This is a **routing/ownership** error, deliberately distinct from the
1102/// instance-wide read-only rejection raised by
1103/// [`WriteGate`](crate::runtime::write_gate::WriteGate): a replica/non-holder
1104/// rejecting a public write is not "this node is read-only", it is "this node is
1105/// not the authority for *this range* — route to the owner". Crucially, none of
1106/// these rejections fall back to the privileged internal replica-apply path; a
1107/// public write that is not for this node's owned range never reaches storage.
1108#[derive(Debug, Clone, PartialEq, Eq)]
1109pub enum RangeWriteReject {
1110    /// No range of the collection covers the routed key, so the write cannot be
1111    /// placed. The caller must (re)resolve routing against a fresher catalog.
1112    NoRange { collection: CollectionId },
1113    /// This node holds the range but is not its owner (a [`Replica`]), or holds
1114    /// no copy at all ([`NoCopy`]). Either way a public write must be routed to
1115    /// `owner`, never applied locally.
1116    ///
1117    /// [`Replica`]: RangeRole::Replica
1118    /// [`NoCopy`]: RangeRole::NoCopy
1119    NotOwner {
1120        collection: CollectionId,
1121        range_id: RangeId,
1122        role: RangeRole,
1123        owner: NodeIdentity,
1124    },
1125    /// This node *is* the range owner, but the write was authorised under an
1126    /// ownership epoch that no longer matches the catalog — a write fenced out
1127    /// because ownership has since moved (its epoch advanced). Carries both
1128    /// epochs so the caller can see how far the routing decision was behind.
1129    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/// The global shard ownership catalog held by every data member.
1170///
1171/// This single type plays both roles in ADR 0037's model: it is the authoritative
1172/// state the Cluster Supervisor leader writes through, and it is the replica each
1173/// data member holds and routes against. Both write through
1174/// [`apply_update`](Self::apply_update), so the stale-version rejection that
1175/// makes leader writes versioned is the *same* rule that makes replica
1176/// application order-independent. Nothing here needs user-data sharding to find
1177/// an entry: ranges are addressed directly by `(collection, range_id)`, and
1178/// routing ([`route`](Self::route)) is a local scan of the replica.
1179#[derive(Debug, Clone, Default)]
1180pub struct ShardOwnershipCatalog {
1181    /// Declared shard key mode per collection. A collection is recorded here the
1182    /// moment its first range is created (or via [`declare_collection`]).
1183    ///
1184    /// [`declare_collection`]: Self::declare_collection
1185    collections: BTreeMap<CollectionId, ShardKeyMode>,
1186    ranges: BTreeMap<(CollectionId, RangeId), RangeOwnership>,
1187}
1188
1189impl ShardOwnershipCatalog {
1190    /// An empty catalog — a cluster with no collections placed yet.
1191    pub fn new() -> Self {
1192        Self::default()
1193    }
1194
1195    /// Declare a collection's shard key mode up front. Hash is the default, so
1196    /// this is mainly how an operator opts a collection into
1197    /// [`Ordered`](ShardKeyMode::Ordered) mode before any range exists. Declaring
1198    /// the same mode twice is idempotent; redeclaring a different mode for a
1199    /// collection that already has a mode is a [`ShardKeyModeMismatch`].
1200    ///
1201    /// [`ShardKeyModeMismatch`]: CatalogError::ShardKeyModeMismatch
1202    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    /// The declared shard key mode of `collection`, if it has any ranges or was
1221    /// explicitly declared.
1222    pub fn shard_key_mode(&self, collection: &CollectionId) -> Option<ShardKeyMode> {
1223        self.collections.get(collection).copied()
1224    }
1225
1226    /// Apply a versioned ownership update — the single write path for both leader
1227    /// writes and replica application.
1228    ///
1229    /// Creation (the range does not yet exist) auto-declares the collection's
1230    /// mode from the entry and checks the new range does not overlap a sibling.
1231    /// Updating an existing range requires the entry's version to **strictly
1232    /// advance** the current version; anything else is a
1233    /// [`StaleVersion`](CatalogError::StaleVersion) rejection that leaves the
1234    /// catalog untouched. Either way the entry's mode must match the collection's
1235    /// declared mode.
1236    pub fn apply_update(&mut self, entry: RangeOwnership) -> Result<UpdateOutcome, CatalogError> {
1237        entry.validate_replica_roles()?;
1238
1239        // Mode must agree with the collection (auto-declared on first range).
1240        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                // Creating a range: it must not overlap any sibling range of the
1276                // same collection, or routing would be ambiguous.
1277                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    /// The current ownership of one range, addressed directly by identity — no
1301    /// routing required, because the catalog is what routing is built on.
1302    pub fn range(&self, collection: &CollectionId, range_id: RangeId) -> Option<&RangeOwnership> {
1303        self.ranges.get(&(collection.clone(), range_id))
1304    }
1305
1306    /// Every range of `collection`, in range-id order.
1307    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    /// Route a normalized range key to the range that owns it — the catalog read
1319    /// every routing decision makes. Returns the owning [`RangeOwnership`] (whose
1320    /// `owner`, `epoch`, and `replicas` the caller uses to send and fence the
1321    /// write), or `None` if no range covers the key yet.
1322    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    /// Route a logical shard key according to the collection's declared mode.
1328    ///
1329    /// Ordered collections route the shard key bytes directly. Hash collections
1330    /// first map the shard key into a stable hash slot and route that slot's
1331    /// range key through the same `collection -> range` catalog.
1332    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    /// This node's [`RangeRole`] for a directly-addressed range (issue #990).
1347    /// Returns `None` when no such range exists in the catalog — distinct from
1348    /// [`NoCopy`](RangeRole::NoCopy), which means the range exists but this node
1349    /// holds no copy of it.
1350    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    /// Ownership-aware gate for a **public** write (issue #990, PRD #987).
1361    ///
1362    /// Routes `key` to its range, then admits the write only when `node` is the
1363    /// range's current [`Owner`](RangeRole::Owner) **and** `expected_epoch`
1364    /// matches the range's current ownership epoch. On success returns the owned
1365    /// [`RangeOwnership`] (so the caller can proceed with the write against the
1366    /// authoritative epoch); otherwise a [`RangeWriteReject`] explaining why.
1367    ///
1368    /// This is the public surface's gate — the counterpart of the instance-wide
1369    /// [`WriteGate`](crate::runtime::write_gate::WriteGate) for multi-writer,
1370    /// per-range ownership. The internal replica-apply path does **not** consult
1371    /// it: replicated changes flow into a replica through the privileged apply
1372    /// path (fenced by issue #991's range-authority watermark), so a node that
1373    /// rejects a *public* write here can still legitimately apply the owner's
1374    /// replicated changes for the very same range.
1375    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    /// Total number of owned ranges across all collections.
1408    pub fn range_count(&self) -> usize {
1409        self.ranges.len()
1410    }
1411
1412    /// All ranges, in `(collection, range_id)` order — the full catalog content
1413    /// a joining member adopts as its starting replica
1414    /// (see [`ControlPlaneSnapshot`](super::join::ControlPlaneSnapshot)).
1415    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    /// A hash range over `[lower, Max)` owned by `owner`.
1437    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(), &current);
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        // The first range of a collection auto-declares its mode; a range built
1559        // with the default mode lands the collection in Hash mode.
1560        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        // Two hash token ranges split at 0x80.
1576        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        // Routing reads expose the owner for a key without any user-data sharding.
1594        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        // The routing read also exposes replicas and fencing epoch.
1607        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        // Ordered ranges bound the ordered key itself: [a, m) and [m, z).
1642        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        // A key outside every declared range routes nowhere.
1674        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        // Redeclaring the same mode is fine.
1685        catalog
1686            .declare_collection(events.clone(), ShardKeyMode::Ordered)
1687            .unwrap();
1688        // A different mode is a mismatch.
1689        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        // And a range whose mode disagrees with the declared collection is rejected.
1701        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        // Owner transfer advances both version and fencing epoch.
1720        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); // old owner is now fenced
1728
1729        // A replica-set change advances the version but NOT the epoch.
1730        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); // write authority did not move
1735        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        // Advance to v2.
1748        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        // Re-applying the original v1 entry (and even a fresh v1-versioned write)
1757        // is stale: version 1 does not advance past the current version 2.
1758        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        // The stale write did not roll ownership back.
1769        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        // A new range overlapping [0x00, 0x80) is ambiguous for routing.
1796        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        // Leader writes the catalog; a data member holds its own replica and
1858        // applies the same versioned updates — no user-data sharding involved.
1859        let orders = collection("orders");
1860        let mut leader = ShardOwnershipCatalog::new();
1861        let mut data_member = ShardOwnershipCatalog::new();
1862
1863        // Leader creates a range; ship the entry to the data member.
1864        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        // The data member can route locally to the same owner the leader has.
1872        assert_eq!(
1873            data_member.route(&orders, b"any-key").unwrap().owner(),
1874            &ident("CN=node-a")
1875        );
1876
1877        // Leader transfers ownership; replicate the v2 entry.
1878        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        // Out-of-order / duplicate replication: re-delivering v2 after it is
1893        // applied is stale on the replica and rejected, so it stays consistent.
1894        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        // Related collections intentionally share one authority domain.
1960        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        // A large collection can be isolated as its own group.
1976        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    // ---------------------------------------------------------------
2020    // Issue #990 — per-range role model and ownership-aware write gate.
2021    // ---------------------------------------------------------------
2022
2023    /// A range owned by `owner` with an explicit replica set.
2024    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        // node-a owns range 1 and is only a replica of range 2 — the same node
2107        // holds different roles for different ranges of the same collection.
2108        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        // A range that does not exist is None, not NoCopy.
2139        assert_eq!(catalog.role_at(&node_a, &orders, RangeId::new(99)), None);
2140        // And a collection nobody placed yet routes nowhere.
2141        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        // node-b holds a copy but is a replica — a public write must be routed
2208        // to the owner, not applied locally.
2209        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        // The rejection names the owner so the caller can re-route.
2227        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        // node-c holds no copy of the range at all.
2245        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        // Ownership moved a→b and back to a, advancing the epoch twice. A write
2262        // the routing layer authorised under the original epoch must be fenced
2263        // even though node-a is, once again, the current owner.
2264        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        // node-a is the current owner again, but at a newer epoch.
2276        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        // The same owner at the *current* epoch is admitted.
2290        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        // A node that rejects a *public* write because it is only a replica must
2313        // still admit the owner's replicated changes through the privileged
2314        // internal apply path — that path is gated by issue #991's range
2315        // authority fence, not by this public ownership gate.
2316        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        // Public gate: node-b is a replica → rejected, never reaches storage.
2331        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        // Internal apply: the owner's replicated change for the same range is
2347        // admitted by the range-authority fence on the replica.
2348        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}