Skip to main content

mongreldb_cluster/
tablet.rs

1//! Tablet identity, partitioning, local layout, and tablet routing (spec
2//! sections 12.1-12.4, Stages 3B/3C/3D).
3//!
4//! Stage 3B (spec section 12.2) lands the partition model: [`Partitioning`]
5//! (hash, range, tenant, time-range) with partition-key extraction through
6//! the order-preserving [`RowKeyEncoder`], the slot mapping
7//! [`Partitioning::route`], the per-table metadata shape
8//! [`TablePartitioningRecord`] (whose [`PartitioningOrigin`] keeps automatic
9//! defaults visible in schema metadata), and the colocation declaration
10//! [`ColocatedWith`].
11//!
12//! Stage 3C (spec section 12.3) lands the tablet descriptor family of spec
13//! section 12.1 ([`TabletDescriptor`], [`ReplicaDescriptor`], [`TabletState`]
14//! with its transition graph enforced in code, [`PartitionBounds`] over typed
15//! [`Key`] bytes) and the on-node layout [`TabletLayout`]:
16//!
17//! ```text
18//! node-data/
19//!   tablets/<tablet-id>/{state,runs,indexes,temp}   + tablet.json
20//!   groups/<raft-group-id>/{raft,snapshots}
21//! ```
22//!
23//! The per-tablet metadata file is versioned, checksummed, and written
24//! atomically with the same idiom as the node identity in [`crate::node`];
25//! loading fails closed on a missing, corrupt, unknown-version, or
26//! wrong-tablet file. One tablet storage core is owned by one node process:
27//! [`TabletOwnershipRegistry`] is the process-local half of that rule (the
28//! storage core's file lease remains the cross-process half), mirroring the
29//! open-reservation concept of the Stage 1 shared-core registry (S1A-002).
30//!
31//! Stage 3D (spec section 12.4) lands pure tablet-selection helpers:
32//! [`find_tablet_for_key`] routes point reads/writes directly,
33//! [`tablets_overlapping`] fans range queries out to every overlapping
34//! tablet, and [`check_generation`] classifies a stale request generation as
35//! the typed [`RoutingError`] the gateway maps onto `TabletMoved`,
36//! `TabletSplitting`, or `StaleMetadata` before refreshing and retrying
37//! through [`crate::routing`].
38//!
39//! Stages 3E/3F (spec sections 12.5-12.6) live in [`crate::split`] and
40//! [`crate::merge`]; this file carries their descriptor-side helpers —
41//! [`PartitionBounds::split_at`], [`PartitionBounds::meets_start_of`],
42//! [`PartitionBounds::union_adjacent`],
43//! [`TabletDescriptor::published_transition`] (the one-publication/one-tick
44//! generation rules are documented on [`TabletDescriptor`]), and
45//! [`TabletLayout::teardown`] for source-replica removal. The enforced
46//! [`TabletState`] graph needs no new edges: child publication rides
47//! `Creating -> Active`, source retirement `Splitting`/`Merging ->
48//! `Retiring -> Retired`, and abort paths ride the existing
49//! `Splitting`/`Merging -> Active` and `Creating -> Retired` edges.
50
51use std::collections::{BTreeMap, HashMap};
52use std::fmt;
53use std::fs;
54use std::ops;
55use std::path::{Path, PathBuf};
56use std::sync::{Mutex, OnceLock};
57
58use mongreldb_types::errors::ErrorCategory;
59use mongreldb_types::ids::{NodeId, RaftGroupId, TableId, TabletId};
60use serde::{Deserialize, Serialize};
61use sha2::{Digest, Sha256};
62
63use crate::node::ClusterError;
64
65// ---------------------------------------------------------------------------
66// Column identifiers
67// ---------------------------------------------------------------------------
68
69/// Column identifier inside one table.
70///
71/// Mirrors the storage core's `u16` column ids (spec section 12.2 declares
72/// partition keys over `ColumnId`); the cluster crate deliberately does not
73/// depend on the core, so the newtype is declared here and must track it.
74#[derive(
75    Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize,
76)]
77#[repr(transparent)]
78pub struct ColumnId(pub u16);
79
80impl ColumnId {
81    /// Wraps a raw column id.
82    pub const fn new(id: u16) -> Self {
83        Self(id)
84    }
85
86    /// The raw column id.
87    pub const fn get(self) -> u16 {
88        self.0
89    }
90}
91
92impl fmt::Display for ColumnId {
93    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
94        self.0.fmt(f)
95    }
96}
97
98// ---------------------------------------------------------------------------
99// Typed key bytes
100// ---------------------------------------------------------------------------
101
102/// Typed key bytes produced by [`RowKeyEncoder`].
103///
104/// The encoding is order-preserving: the lexicographic byte order of two
105/// [`Key`]s matches the typed order of the values they encode, which is what
106/// lets [`PartitionBounds`] range over bytes while partitioning ranges over
107/// typed values.
108#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
109pub struct Key(Vec<u8>);
110
111impl Key {
112    /// Wraps raw encoded bytes without copying.
113    pub const fn from_bytes(bytes: Vec<u8>) -> Self {
114        Self(bytes)
115    }
116
117    /// Borrows the encoded bytes.
118    pub fn as_bytes(&self) -> &[u8] {
119        &self.0
120    }
121
122    /// Consumes the key, returning the encoded bytes.
123    pub fn into_bytes(self) -> Vec<u8> {
124        self.0
125    }
126}
127
128impl AsRef<[u8]> for Key {
129    fn as_ref(&self) -> &[u8] {
130        self.as_bytes()
131    }
132}
133
134impl fmt::Display for Key {
135    /// Lowercase hexadecimal of the encoded bytes.
136    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
137        f.write_str(&hex_encode(&self.0))
138    }
139}
140
141impl Serialize for Key {
142    /// Human-readable serializers (e.g. JSON) receive lowercase hex; binary
143    /// serializers receive the raw byte string.
144    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
145        if serializer.is_human_readable() {
146            serializer.serialize_str(&hex_encode(&self.0))
147        } else {
148            serializer.serialize_bytes(&self.0)
149        }
150    }
151}
152
153impl<'de> Deserialize<'de> for Key {
154    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
155        struct KeyVisitor;
156
157        impl<'v> serde::de::Visitor<'v> for KeyVisitor {
158            type Value = Key;
159
160            fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
161                f.write_str("a hex string or byte sequence")
162            }
163
164            fn visit_str<E: serde::de::Error>(self, text: &str) -> Result<Key, E> {
165                hex_decode(text)
166                    .map(Key)
167                    .map_err(|detail| E::custom(format!("invalid key hex: {detail}")))
168            }
169
170            fn visit_bytes<E: serde::de::Error>(self, bytes: &[u8]) -> Result<Key, E> {
171                Ok(Key(bytes.to_vec()))
172            }
173
174            fn visit_byte_buf<E: serde::de::Error>(self, bytes: Vec<u8>) -> Result<Key, E> {
175                Ok(Key(bytes))
176            }
177
178            fn visit_seq<A: serde::de::SeqAccess<'v>>(self, mut seq: A) -> Result<Key, A::Error> {
179                let mut bytes = Vec::with_capacity(seq.size_hint().unwrap_or(0));
180                while let Some(byte) = seq.next_element::<u8>()? {
181                    bytes.push(byte);
182                }
183                Ok(Key(bytes))
184            }
185        }
186
187        if deserializer.is_human_readable() {
188            deserializer.deserialize_str(KeyVisitor)
189        } else {
190            deserializer.deserialize_bytes(KeyVisitor)
191        }
192    }
193}
194
195fn hex_encode(bytes: &[u8]) -> String {
196    const HEX: &[u8; 16] = b"0123456789abcdef";
197    let mut out = String::with_capacity(bytes.len() * 2);
198    for byte in bytes {
199        out.push(HEX[(byte >> 4) as usize] as char);
200        out.push(HEX[(byte & 0x0f) as usize] as char);
201    }
202    out
203}
204
205fn hex_decode(text: &str) -> Result<Vec<u8>, String> {
206    if !text.len().is_multiple_of(2) {
207        return Err("odd number of hex digits".to_owned());
208    }
209    let bytes = text.as_bytes();
210    let mut out = Vec::with_capacity(bytes.len() / 2);
211    for pair in bytes.chunks_exact(2) {
212        let digit = |byte: u8| -> Result<u8, String> {
213            (byte as char)
214                .to_digit(16)
215                .map(|value| value as u8)
216                .ok_or_else(|| format!("invalid hex character `{}`", byte as char))
217        };
218        out.push((digit(pair[0])? << 4) | digit(pair[1])?);
219    }
220    Ok(out)
221}
222
223// ---------------------------------------------------------------------------
224// Partition bounds
225// ---------------------------------------------------------------------------
226
227/// One endpoint of a partition range.
228///
229/// Mirrors [`std::ops::Bound`] with a serde-friendly shape: the standard
230/// enum carries no `Serialize`/`Deserialize`, and tablet descriptors must
231/// persist their bounds (spec section 12.1). Declaration order is frozen;
232/// variants are never reused (spec section 4.10).
233#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
234pub enum Bound<T> {
235    /// No endpoint; the range extends without limit on this side.
236    Unbounded,
237    /// The endpoint value is part of the range.
238    Included(T),
239    /// The endpoint value is not part of the range.
240    Excluded(T),
241}
242
243impl<T: Clone> Bound<T> {
244    /// Converts from the standard-library bound.
245    pub fn from_std(bound: ops::Bound<T>) -> Self {
246        match bound {
247            ops::Bound::Unbounded => Self::Unbounded,
248            ops::Bound::Included(value) => Self::Included(value),
249            ops::Bound::Excluded(value) => Self::Excluded(value),
250        }
251    }
252}
253
254/// The key range one tablet covers: `low` to `high` over typed [`Key`] bytes
255/// (spec section 12.1).
256///
257/// Tablets partition the whole key space, so well-formed bounds never leave
258/// gaps or overlaps between routable tablets of one table;
259/// [`Self::overlaps`] and [`Self::is_adjacent_to`] are the predicates split,
260/// merge, and range routing are built on.
261#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
262#[serde(deny_unknown_fields)]
263pub struct PartitionBounds {
264    /// First key of the tablet (`Unbounded` for the first tablet of a table).
265    pub low: Bound<Key>,
266    /// End key of the tablet (`Unbounded` for the last tablet of a table).
267    pub high: Bound<Key>,
268}
269
270impl PartitionBounds {
271    /// Builds bounds, rejecting empty or inverted ranges.
272    pub fn new(low: Bound<Key>, high: Bound<Key>) -> Result<Self, TabletError> {
273        let bounds = Self { low, high };
274        bounds.validate()?;
275        Ok(bounds)
276    }
277
278    /// The whole key space.
279    pub fn unbounded() -> Self {
280        Self {
281            low: Bound::Unbounded,
282            high: Bound::Unbounded,
283        }
284    }
285
286    /// The bounds are non-empty: `low` is strictly below `high`, except that
287    /// the single-point range `[k, k]` (both endpoints included) is allowed.
288    pub fn validate(&self) -> Result<(), TabletError> {
289        if let (Some(low), Some(high)) = (bound_key(&self.low), bound_key(&self.high)) {
290            if low > high {
291                return Err(TabletError::InvalidBounds(format!(
292                    "low endpoint {low} is above high endpoint {high}"
293                )));
294            }
295            let single_point =
296                matches!(self.low, Bound::Included(_)) && matches!(self.high, Bound::Included(_));
297            if low == high && !single_point {
298                return Err(TabletError::InvalidBounds(format!(
299                    "empty range at endpoint {low}"
300                )));
301            }
302        }
303        Ok(())
304    }
305
306    /// Whether `key` lies inside the bounds.
307    pub fn contains(&self, key: &Key) -> bool {
308        let above_low = match &self.low {
309            Bound::Unbounded => true,
310            Bound::Included(low) => key >= low,
311            Bound::Excluded(low) => key > low,
312        };
313        let below_high = match &self.high {
314            Bound::Unbounded => true,
315            Bound::Included(high) => key <= high,
316            Bound::Excluded(high) => key < high,
317        };
318        above_low && below_high
319    }
320
321    /// Whether the two ranges share at least one key.
322    pub fn overlaps(&self, other: &Self) -> bool {
323        !ends_before(&self.high, &other.low) && !ends_before(&other.high, &self.low)
324    }
325
326    /// Whether the two ranges touch with no gap and no overlap: the touching
327    /// endpoints name the same key and exactly one of them includes it (for
328    /// example `[a, c)` and `[c, b)` are adjacent; `[a, c]` and `[c, b)`
329    /// overlap at `c`; `[a, c)` and `(c, b]` leave `c` uncovered).
330    pub fn is_adjacent_to(&self, other: &Self) -> bool {
331        meets(&self.high, &other.low) || meets(&other.high, &self.low)
332    }
333
334    /// Whether `self` ends exactly where `other` begins — the lower half of
335    /// an adjacent pair (spec section 12.6 merge ordering).
336    pub fn meets_start_of(&self, other: &Self) -> bool {
337        meets(&self.high, &other.low)
338    }
339
340    /// Splits the range at `key` into the lower half `[low, key)` and the
341    /// upper half `[key, high)` (spec section 12.5 step 2). The halves meet at
342    /// `key` ([`Self::meets_start_of`]), so they partition the original range
343    /// with no gap and no overlap.
344    ///
345    /// Returns `None` — never a partial split — when `key` is not contained
346    /// in the range or would leave either half empty (for example a split at
347    /// an included `low` endpoint).
348    pub fn split_at(&self, key: &Key) -> Option<(Self, Self)> {
349        if !self.contains(key) {
350            return None;
351        }
352        let lower = Self::new(self.low.clone(), Bound::Excluded(key.clone())).ok()?;
353        let upper = Self::new(Bound::Included(key.clone()), self.high.clone()).ok()?;
354        Some((lower, upper))
355    }
356
357    /// The combined bounds of two adjacent ranges: the lower range's `low`
358    /// and the upper range's `high` (spec section 12.6). Returns `None` when
359    /// the ranges are not adjacent (overlapping or disjoint) — adjacent valid
360    /// ranges always combine into a valid range.
361    pub fn union_adjacent(&self, other: &Self) -> Option<Self> {
362        let (lower, upper) = if self.meets_start_of(other) {
363            (self, other)
364        } else if other.meets_start_of(self) {
365            (other, self)
366        } else {
367            return None;
368        };
369        Self::new(lower.low.clone(), upper.high.clone()).ok()
370    }
371}
372
373/// The endpoint key of a bound, if bounded.
374fn bound_key(bound: &Bound<Key>) -> Option<&Key> {
375    match bound {
376        Bound::Unbounded => None,
377        Bound::Included(key) | Bound::Excluded(key) => Some(key),
378    }
379}
380
381/// Whether a range ending at `high` ends strictly before a range starting at
382/// `low` begins. Touching endpoints count as ending before only when at least
383/// one side excludes the shared key.
384fn ends_before(high: &Bound<Key>, low: &Bound<Key>) -> bool {
385    let (Some(high_key), Some(low_key)) = (bound_key(high), bound_key(low)) else {
386        return false;
387    };
388    if high_key != low_key {
389        return high_key < low_key;
390    }
391    // Same endpoint key: the ranges share it only when both sides include it.
392    !(matches!(high, Bound::Included(_)) && matches!(low, Bound::Included(_)))
393}
394
395/// Whether `high` and `low` meet: same key, exactly one side included.
396fn meets(high: &Bound<Key>, low: &Bound<Key>) -> bool {
397    match (high, low) {
398        (Bound::Included(high), Bound::Excluded(low))
399        | (Bound::Excluded(high), Bound::Included(low)) => high == low,
400        _ => false,
401    }
402}
403
404// ---------------------------------------------------------------------------
405// Tablet descriptors (spec section 12.1)
406// ---------------------------------------------------------------------------
407
408/// Lifecycle state of a tablet (spec sections 12.1, 12.5, 12.6). Declaration
409/// order is frozen; variants are never reused (spec section 4.10).
410#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
411pub enum TabletState {
412    /// Created as learners and catching up; never routed to (spec section
413    /// 12.5: do not expose child tablets before catch-up).
414    Creating,
415    /// Serving traffic.
416    Active,
417    /// Split in flight: the source remains the routable owner until the
418    /// children are published atomically (spec section 12.5).
419    Splitting,
420    /// Merge in flight: the sources remain routable until the hidden
421    /// replacement is published atomically (spec section 12.6).
422    Merging,
423    /// Retired from routing by an atomic publication; retained until no
424    /// old-generation requests or pins remain (spec section 12.5 step 10).
425    Retiring,
426    /// Fully retired; replicas may be removed. Terminal: tablet identifiers
427    /// are never reused (spec section 7).
428    Retired,
429}
430
431impl TabletState {
432    /// Whether the transition `self -> next` is permitted.
433    ///
434    /// The graph (enforced here, applied at quorum by the meta group):
435    ///
436    /// ```text
437    /// Creating  -> Active | Retired          (catch-up done / creation aborted)
438    /// Active    -> Splitting | Merging | Retiring
439    /// Splitting -> Active | Retiring         (split aborted / children published)
440    /// Merging   -> Active | Retiring         (merge aborted / replacement published)
441    /// Retiring  -> Retired                   (no old-generation pins remain)
442    /// Retired   -> (terminal)
443    /// ```
444    pub fn can_transition_to(self, next: Self) -> bool {
445        use TabletState::{Active, Creating, Merging, Retired, Retiring, Splitting};
446        matches!(
447            (self, next),
448            (Creating, Active)
449                | (Creating, Retired)
450                | (Active, Splitting)
451                | (Active, Merging)
452                | (Active, Retiring)
453                | (Splitting, Active)
454                | (Splitting, Retiring)
455                | (Merging, Active)
456                | (Merging, Retiring)
457                | (Retiring, Retired)
458        )
459    }
460
461    /// Whether requests may be routed to this tablet: the serving owner
462    /// (`Active`) and the sources of an in-flight split or merge, which stay
463    /// authoritative until the atomic publication flips them to
464    /// [`TabletState::Retiring`]. `Creating` tablets are never exposed (spec
465    /// section 12.5).
466    pub fn is_routable(self) -> bool {
467        matches!(self, Self::Active | Self::Splitting | Self::Merging)
468    }
469}
470
471impl fmt::Display for TabletState {
472    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
473        let name = match self {
474            Self::Creating => "Creating",
475            Self::Active => "Active",
476            Self::Splitting => "Splitting",
477            Self::Merging => "Merging",
478            Self::Retiring => "Retiring",
479            Self::Retired => "Retired",
480        };
481        f.write_str(name)
482    }
483}
484
485/// Role of one replica within its tablet's Raft group (spec section 12.1).
486/// Declaration order is frozen; variants are never reused (spec section
487/// 4.10).
488#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
489pub enum ReplicaRole {
490    /// Voting member of the group; counts toward quorum.
491    Voter,
492    /// Non-voting member receiving the log; promoted to voter during the
493    /// movement protocol (spec section 12.7).
494    Learner,
495    /// Non-voting specialized read replica (spec §13.5, Stage 4E).
496    ReadReplica,
497    /// Non-voting AI-optimized read replica (larger ANN indexes).
498    AiReadReplica,
499    /// Non-voting analytics replica (columnar projections, longer history).
500    AnalyticsReplica,
501    /// Non-voting backup replica (snapshot retention).
502    BackupReplica,
503}
504
505impl ReplicaRole {
506    /// Whether this role counts toward Raft quorum (only [`Self::Voter`]).
507    pub const fn counts_toward_quorum(self) -> bool {
508        matches!(self, Self::Voter)
509    }
510
511    /// Whether the replica applies the committed log (all roles do).
512    pub const fn applies_log(self) -> bool {
513        true
514    }
515
516    /// Whether lag-based routing may select this replica for reads.
517    pub const fn serves_reads(self) -> bool {
518        !matches!(self, Self::BackupReplica)
519    }
520}
521
522impl fmt::Display for ReplicaRole {
523    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
524        let name = match self {
525            Self::Voter => "Voter",
526            Self::Learner => "Learner",
527            Self::ReadReplica => "ReadReplica",
528            Self::AiReadReplica => "AiReadReplica",
529            Self::AnalyticsReplica => "AnalyticsReplica",
530            Self::BackupReplica => "BackupReplica",
531        };
532        f.write_str(name)
533    }
534}
535
536/// One replica of a tablet (spec section 12.1).
537///
538/// `raft_node_id` is the replica's identifier inside the tablet's Raft
539/// group: the `RaftNodeId` (`u64`) of `mongreldb-consensus`, allocated by the
540/// meta control plane and never reused within one group. The mapping between
541/// the cluster-wide [`NodeId`] and the per-group `raft_node_id` is replicated
542/// meta-group state (spec section 12.1); this descriptor carries both so
543/// routers and placement never re-resolve it.
544#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
545#[serde(deny_unknown_fields)]
546pub struct ReplicaDescriptor {
547    /// Cluster-wide identity of the node holding the replica.
548    pub node_id: NodeId,
549    /// Voter or learner.
550    pub role: ReplicaRole,
551    /// Openraft node id within the tablet's Raft group (see above).
552    pub raft_node_id: u64,
553}
554
555/// One independently replicated data partition (spec section 12.1).
556///
557/// `generation` advances at every atomic descriptor publication (split,
558/// merge, move); every request carries the generation it was routed with,
559/// and a mismatch is classified by [`check_generation`].
560///
561/// Publication generation rules (spec sections 12.5-12.6; the protocols live
562/// in [`crate::split`] and [`crate::merge`]):
563///
564/// - Split: the source is marked [`TabletState::Splitting`] at `g + 1` (`g` =
565///   the pre-split generation); the children are created `Creating` at
566///   `g + 1` and are never routed to; the atomic routing publication assigns
567///   one new generation `p = (source generation at publication) + 1`, taking
568///   the children `Creating -> Active` and the source `Splitting -> Retiring`
569///   together. A request holding `g` against the splitting source is
570///   classified [`RoutingError::TabletSplit`]; after publication any
571///   generation below `p` against the retiring source is
572///   [`RoutingError::TabletMoved`].
573/// - Merge: each source is marked [`TabletState::Merging`] at its own
574///   `g + 1`; the publication generation is
575///   `p = max(sources' generations at publication) + 1`, assigned to the
576///   replacement (`Creating -> Active`) and to both sources
577///   (`Merging -> Retiring`) in one atomic command.
578/// - Source removal publishes `Retiring -> Retired` at `p + 1` before the
579///   descriptor is deleted.
580#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
581#[serde(deny_unknown_fields)]
582pub struct TabletDescriptor {
583    /// The tablet's identity.
584    pub tablet_id: TabletId,
585    /// Table the tablet belongs to.
586    pub table_id: TableId,
587    /// Logical database owning the table (meta-resolved). When zero on a
588    /// legacy descriptor, the runtime resolves via meta
589    /// `table_id → database_id` before falling back to a deterministic
590    /// raft-group-derived id for pre-metadata tablets only.
591    #[serde(default = "crate::tablet::zero_database_id")]
592    pub database_id: mongreldb_types::ids::DatabaseId,
593    /// Raft group replicating the tablet.
594    pub raft_group_id: RaftGroupId,
595    /// Key range the tablet covers.
596    pub partition: PartitionBounds,
597    /// All replicas (voters and learners), on distinct nodes.
598    pub replicas: Vec<ReplicaDescriptor>,
599    /// Last known leader, when the meta plane has observed one.
600    pub leader_hint: Option<NodeId>,
601    /// Descriptor generation; bumped by every atomic publication.
602    pub generation: u64,
603    /// Lifecycle state; transitions go through [`Self::try_transition`].
604    pub state: TabletState,
605}
606
607fn zero_database_id() -> mongreldb_types::ids::DatabaseId {
608    mongreldb_types::ids::DatabaseId::ZERO
609}
610
611impl TabletDescriptor {
612    /// Structural validation: reserved identifiers, well-formed bounds,
613    /// distinct replica nodes and Raft ids, a leader hint that names a
614    /// replica, and at least one voter outside [`TabletState::Creating`].
615    pub fn validate(&self) -> Result<(), TabletError> {
616        if self.tablet_id == TabletId::ZERO {
617            return Err(TabletError::InvalidDescriptor(
618                "reserved all-zero tablet id".to_owned(),
619            ));
620        }
621        if self.raft_group_id == RaftGroupId::ZERO {
622            return Err(TabletError::InvalidDescriptor(
623                "reserved all-zero raft group id".to_owned(),
624            ));
625        }
626        self.partition.validate()?;
627        if self.replicas.is_empty() {
628            return Err(TabletError::InvalidDescriptor(
629                "tablet has no replicas".to_owned(),
630            ));
631        }
632        for (index, replica) in self.replicas.iter().enumerate() {
633            if self.replicas[..index]
634                .iter()
635                .any(|prior| prior.node_id == replica.node_id)
636            {
637                return Err(TabletError::InvalidDescriptor(format!(
638                    "node {} holds more than one replica",
639                    replica.node_id
640                )));
641            }
642            if self.replicas[..index]
643                .iter()
644                .any(|prior| prior.raft_node_id == replica.raft_node_id)
645            {
646                return Err(TabletError::InvalidDescriptor(format!(
647                    "raft node id {} is used by more than one replica",
648                    replica.raft_node_id
649                )));
650            }
651        }
652        if let Some(leader) = self.leader_hint {
653            if !self
654                .replicas
655                .iter()
656                .any(|replica| replica.node_id == leader)
657            {
658                return Err(TabletError::InvalidDescriptor(format!(
659                    "leader hint {leader} is not a replica of the tablet"
660                )));
661            }
662        }
663        if self.state != TabletState::Creating && self.voter_count() == 0 {
664            return Err(TabletError::InvalidDescriptor(format!(
665                "tablet in state {} has no voters",
666                self.state
667            )));
668        }
669        Ok(())
670    }
671
672    /// The voter replicas.
673    pub fn voters(&self) -> impl Iterator<Item = &ReplicaDescriptor> {
674        self.replicas
675            .iter()
676            .filter(|replica| replica.role == ReplicaRole::Voter)
677    }
678
679    /// The learner replicas.
680    pub fn learners(&self) -> impl Iterator<Item = &ReplicaDescriptor> {
681        self.replicas
682            .iter()
683            .filter(|replica| replica.role == ReplicaRole::Learner)
684    }
685
686    /// Number of voting replicas.
687    pub fn voter_count(&self) -> usize {
688        self.voters().count()
689    }
690
691    /// The replica on `node`, if any.
692    pub fn replica_on(&self, node: NodeId) -> Option<&ReplicaDescriptor> {
693        self.replicas.iter().find(|replica| replica.node_id == node)
694    }
695
696    /// Transitions to `next`, enforcing the [`TabletState`] graph. Generation
697    /// bumps are the meta group's business — this changes only the state.
698    pub fn try_transition(&mut self, next: TabletState) -> Result<(), TabletError> {
699        if !self.state.can_transition_to(next) {
700            return Err(TabletError::InvalidStateTransition {
701                tablet: self.tablet_id,
702                from: self.state,
703                to: next,
704            });
705        }
706        self.state = next;
707        Ok(())
708    }
709
710    /// A copy of the descriptor transitioned to `next` with the generation
711    /// bumped by one: the shape of one atomic descriptor publication (see the
712    /// publication generation rules on [`TabletDescriptor`]). The descriptor
713    /// itself is unchanged, so the caller stages the published form and the
714    /// meta plane applies it last-writer-wins.
715    pub fn published_transition(&self, next: TabletState) -> Result<Self, TabletError> {
716        let mut published = self.clone();
717        published.try_transition(next)?;
718        published.generation =
719            published
720                .generation
721                .checked_add(1)
722                .ok_or(TabletError::InvalidDescriptor(
723                    "descriptor generation overflows u64".to_owned(),
724                ))?;
725        Ok(published)
726    }
727}
728
729// ---------------------------------------------------------------------------
730// Errors
731// ---------------------------------------------------------------------------
732
733/// The one error type of the tablet surface: descriptors, layout, and
734/// ownership.
735#[derive(Debug, thiserror::Error)]
736pub enum TabletError {
737    /// A durable metadata file failed the node.rs idiom's checks (unknown
738    /// version, corrupt payload, I/O). Always fail closed.
739    #[error("tablet metadata operation failed: {0}")]
740    Metadata(#[from] ClusterError),
741    /// No `tablet.json` exists where a tablet was expected.
742    #[error("tablet metadata is missing at {0}")]
743    MissingMetadata(PathBuf),
744    /// The persisted metadata names a different tablet or Raft group than the
745    /// directory it lives in.
746    #[error(
747        "tablet directory {path} holds metadata for tablet {found} / group {found_group}, \
748         expected tablet {expected} / group {expected_group}"
749    )]
750    TabletMismatch {
751        /// The tablet directory.
752        path: PathBuf,
753        /// Tablet id implied by the directory name.
754        expected: TabletId,
755        /// Tablet id the metadata names.
756        found: TabletId,
757        /// Raft group id implied by the layout.
758        expected_group: RaftGroupId,
759        /// Raft group id the metadata names.
760        found_group: RaftGroupId,
761    },
762    /// `create` found existing, different metadata; never silently replaced.
763    #[error("tablet directory {0} already holds different tablet metadata")]
764    MetadataConflict(PathBuf),
765    /// The requested [`TabletState`] transition is not in the graph.
766    #[error("invalid tablet state transition for tablet {tablet}: {from} -> {to}")]
767    InvalidStateTransition {
768        /// The tablet whose state was to change.
769        tablet: TabletId,
770        /// Current state.
771        from: TabletState,
772        /// Requested target state.
773        to: TabletState,
774    },
775    /// Empty or inverted partition bounds.
776    #[error("invalid partition bounds: {0}")]
777    InvalidBounds(String),
778    /// A [`TabletDescriptor`] failed structural validation.
779    #[error("invalid tablet descriptor: {0}")]
780    InvalidDescriptor(String),
781    /// Partitioning failed.
782    #[error("partitioning failed: {0}")]
783    Partition(#[from] PartitionError),
784    /// Another tablet storage core in this process already owns the tablet
785    /// directory (spec section 4.1; mirrors S1A-002).
786    #[error("tablet storage core at {path} is already owned in this process by tablet {tablet}")]
787    AlreadyOwned {
788        /// Tablet holding the reservation.
789        tablet: TabletId,
790        /// The contested tablet directory.
791        path: PathBuf,
792    },
793}
794
795/// Partition-key extraction or slot routing failed (spec section 12.2).
796#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
797pub enum PartitionError {
798    /// The [`Partitioning`] declaration itself is malformed.
799    #[error("invalid partitioning: {0}")]
800    InvalidPartitioning(String),
801    /// A declared partition-key column was absent from the supplied values.
802    #[error("partition key column {column} is missing from the supplied values")]
803    MissingPartitionColumn {
804        /// The column that was looked up.
805        column: ColumnId,
806    },
807    /// A partition-key column carried a value of the wrong type.
808    #[error("partition key column {column} is {found}, expected {expected}")]
809    PartitionColumnType {
810        /// The column whose value failed the type check.
811        column: ColumnId,
812        /// The required value kind.
813        expected: &'static str,
814        /// The supplied value kind.
815        found: &'static str,
816    },
817    /// An encoded [`Key`] could not be decoded back into components.
818    #[error("malformed encoded key: {0}")]
819    MalformedKey(String),
820    /// A time-range timestamp before the Unix epoch maps to a negative slot.
821    #[error("time-range partition slot is negative for pre-epoch timestamp {micros} micros")]
822    NegativeSlot {
823        /// The offending timestamp, in microseconds since the Unix epoch.
824        micros: i64,
825    },
826}
827
828/// A request arrived with a tablet generation that no longer matches (spec
829/// section 12.4). The gateway refreshes metadata and retries safe operations.
830#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
831pub enum RoutingError {
832    /// The tablet moved (or retired); refresh routing metadata.
833    #[error(
834        "tablet {tablet_id} moved: request used generation {used_generation}, \
835         current generation is {current_generation}"
836    )]
837    TabletMoved {
838        /// The tablet the request targeted.
839        tablet_id: TabletId,
840        /// Generation the request was routed with.
841        used_generation: u64,
842        /// Generation the replica now holds.
843        current_generation: u64,
844    },
845    /// The tablet is mid-split; retry once the split publishes.
846    #[error(
847        "tablet {tablet_id} is splitting: request used generation {used_generation}, \
848         current generation is {current_generation}"
849    )]
850    TabletSplit {
851        /// The tablet the request targeted.
852        tablet_id: TabletId,
853        /// Generation the request was routed with.
854        used_generation: u64,
855        /// Generation the replica now holds.
856        current_generation: u64,
857    },
858    /// The request's routing metadata is stale for any other reason.
859    #[error(
860        "stale tablet metadata for tablet {tablet_id}: request used generation \
861         {used_generation}, current generation is {current_generation}"
862    )]
863    StaleMetadata {
864        /// The tablet the request targeted.
865        tablet_id: TabletId,
866        /// Generation the request was routed with.
867        used_generation: u64,
868        /// Generation the replica now holds.
869        current_generation: u64,
870    },
871}
872
873impl RoutingError {
874    /// The stable error category the gateway maps this onto (spec sections
875    /// 9.7, 12.4); all three refresh routing metadata and retry safe
876    /// operations.
877    pub fn category(&self) -> ErrorCategory {
878        match self {
879            Self::TabletMoved { .. } => ErrorCategory::TabletMoved,
880            Self::TabletSplit { .. } => ErrorCategory::TabletSplitting,
881            Self::StaleMetadata { .. } => ErrorCategory::StaleMetadata,
882        }
883    }
884}
885
886// ---------------------------------------------------------------------------
887// Row key encoding (spec section 12.2)
888// ---------------------------------------------------------------------------
889
890/// One typed key component fed to [`RowKeyEncoder`].
891///
892/// The supported type tags are deliberately lean — they cover the partition
893/// keys of spec section 12.2 (numeric, textual, timestamp). Extension rule
894/// (spec section 4.10): a new value kind is a new variant with a fresh tag
895/// byte appended at the end of the tag space; tags are never reused or
896/// reordered, because keys persist in tablet bounds.
897#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
898pub enum KeyValue {
899    /// SQL NULL; sorts before every other value.
900    Null,
901    /// Boolean; `false` sorts before `true`.
902    Bool(bool),
903    /// Signed 64-bit integer.
904    Int(i64),
905    /// Microseconds since the Unix epoch (the time-range partition type).
906    TimestampMicros(i64),
907    /// UTF-8 text.
908    Text(String),
909    /// Opaque bytes.
910    Bytes(Vec<u8>),
911}
912
913impl KeyValue {
914    /// Stable name of the value kind, for typed errors.
915    pub fn type_name(&self) -> &'static str {
916        match self {
917            Self::Null => "null",
918            Self::Bool(_) => "bool",
919            Self::Int(_) => "int",
920            Self::TimestampMicros(_) => "timestamp-micros",
921            Self::Text(_) => "text",
922            Self::Bytes(_) => "bytes",
923        }
924    }
925}
926
927// Type tags; declaration order defines the cross-type sort order and is
928// frozen (spec section 4.10).
929const TAG_NULL: u8 = 0x01;
930const TAG_BOOL: u8 = 0x02;
931const TAG_INT: u8 = 0x03;
932const TAG_TIMESTAMP_MICROS: u8 = 0x04;
933const TAG_TEXT: u8 = 0x05;
934const TAG_BYTES: u8 = 0x06;
935
936/// Order-preserving encoder for typed key components.
937///
938/// Encoding rules, component by component (one tag byte, then the payload):
939///
940/// - `Null`: tag only.
941/// - `Bool`: tag, then `0x00`/`0x01`.
942/// - `Int` and `TimestampMicros`: tag, then 8 big-endian bytes of the value
943///   with the sign bit flipped, so byte order matches signed numeric order.
944/// - `Text` and `Bytes`: tag, then the payload with every `0x00` byte
945///   escaped as `0x00 0xFF`, terminated by `0x00 0x00` — prefix-free and
946///   order-preserving.
947///
948/// Components concatenate in declared column order, so a composite key sorts
949/// by its first column, then its second, and so on.
950pub struct RowKeyEncoder;
951
952impl RowKeyEncoder {
953    /// Encodes a full key from its typed components.
954    pub fn encode_key(values: &[KeyValue]) -> Key {
955        let mut out = Vec::new();
956        for value in values {
957            Self::encode_component(&mut out, value);
958        }
959        Key::from_bytes(out)
960    }
961
962    /// Appends one component to `out`.
963    pub fn encode_component(out: &mut Vec<u8>, value: &KeyValue) {
964        match value {
965            KeyValue::Null => out.push(TAG_NULL),
966            KeyValue::Bool(bit) => {
967                out.push(TAG_BOOL);
968                out.push(u8::from(*bit));
969            }
970            KeyValue::Int(int) => {
971                out.push(TAG_INT);
972                out.extend_from_slice(&orderable_i64(*int).to_be_bytes());
973            }
974            KeyValue::TimestampMicros(micros) => {
975                out.push(TAG_TIMESTAMP_MICROS);
976                out.extend_from_slice(&orderable_i64(*micros).to_be_bytes());
977            }
978            KeyValue::Text(text) => {
979                out.push(TAG_TEXT);
980                encode_escaped(out, text.as_bytes());
981            }
982            KeyValue::Bytes(bytes) => {
983                out.push(TAG_BYTES);
984                encode_escaped(out, bytes);
985            }
986        }
987    }
988
989    /// Decodes a key back into its components; the inverse of
990    /// [`Self::encode_key`]. Malformed input fails closed.
991    pub fn decode_components(key: &Key) -> Result<Vec<KeyValue>, PartitionError> {
992        let bytes = key.as_bytes();
993        let mut values = Vec::new();
994        let mut cursor = 0;
995        while cursor < bytes.len() {
996            let tag = bytes[cursor];
997            cursor += 1;
998            let value = match tag {
999                TAG_NULL => KeyValue::Null,
1000                TAG_BOOL => {
1001                    let byte = take(bytes, &mut cursor, 1)?[0];
1002                    match byte {
1003                        0 => KeyValue::Bool(false),
1004                        1 => KeyValue::Bool(true),
1005                        other => {
1006                            return Err(PartitionError::MalformedKey(format!(
1007                                "invalid bool payload {other}"
1008                            )))
1009                        }
1010                    }
1011                }
1012                TAG_INT => {
1013                    let raw = take(bytes, &mut cursor, 8)?;
1014                    KeyValue::Int(unorderable_i64(raw))
1015                }
1016                TAG_TIMESTAMP_MICROS => {
1017                    let raw = take(bytes, &mut cursor, 8)?;
1018                    KeyValue::TimestampMicros(unorderable_i64(raw))
1019                }
1020                TAG_TEXT => {
1021                    let payload = decode_escaped(bytes, &mut cursor)?;
1022                    let text = String::from_utf8(payload)
1023                        .map_err(|error| PartitionError::MalformedKey(error.to_string()))?;
1024                    KeyValue::Text(text)
1025                }
1026                TAG_BYTES => KeyValue::Bytes(decode_escaped(bytes, &mut cursor)?),
1027                other => {
1028                    return Err(PartitionError::MalformedKey(format!(
1029                        "unknown type tag 0x{other:02x}"
1030                    )))
1031                }
1032            };
1033            values.push(value);
1034        }
1035        Ok(values)
1036    }
1037
1038    /// FNV-1a 64-bit over `bytes` — the same mixer the storage core uses to
1039    /// derive row ids from primary keys (`engine.rs`): offset basis
1040    /// `0xcbf29ce484222325`, prime `0x100000001b3`.
1041    pub fn fnv1a64(bytes: &[u8]) -> u64 {
1042        let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
1043        for byte in bytes {
1044            hash ^= u64::from(*byte);
1045            hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
1046        }
1047        hash
1048    }
1049}
1050
1051/// Maps an `i64` onto `u64` preserving order (sign-bit flip).
1052fn orderable_i64(value: i64) -> u64 {
1053    (value as u64) ^ (1 << 63)
1054}
1055
1056/// Inverse of [`orderable_i64`], from 8 big-endian bytes.
1057fn unorderable_i64(raw: &[u8]) -> i64 {
1058    let mut bytes = [0u8; 8];
1059    bytes.copy_from_slice(raw);
1060    (u64::from_be_bytes(bytes) ^ (1 << 63)) as i64
1061}
1062
1063/// Takes `count` bytes from `bytes` at `cursor`, advancing it.
1064fn take<'a>(bytes: &'a [u8], cursor: &mut usize, count: usize) -> Result<&'a [u8], PartitionError> {
1065    let end = cursor
1066        .checked_add(count)
1067        .filter(|end| *end <= bytes.len())
1068        .ok_or_else(|| PartitionError::MalformedKey("truncated payload".to_owned()))?;
1069    let slice = &bytes[*cursor..end];
1070    *cursor = end;
1071    Ok(slice)
1072}
1073
1074/// Appends `payload` with `0x00` escaped as `0x00 0xFF`, plus the `0x00 0x00`
1075/// terminator.
1076fn encode_escaped(out: &mut Vec<u8>, payload: &[u8]) {
1077    for byte in payload {
1078        out.push(*byte);
1079        if *byte == 0x00 {
1080            out.push(0xFF);
1081        }
1082    }
1083    out.extend_from_slice(&[0x00, 0x00]);
1084}
1085
1086/// Reads one escaped payload up to and past its `0x00 0x00` terminator.
1087fn decode_escaped(bytes: &[u8], cursor: &mut usize) -> Result<Vec<u8>, PartitionError> {
1088    let mut payload = Vec::new();
1089    loop {
1090        let byte = *bytes
1091            .get(*cursor)
1092            .ok_or_else(|| PartitionError::MalformedKey("unterminated payload".to_owned()))?;
1093        *cursor += 1;
1094        if byte != 0x00 {
1095            payload.push(byte);
1096            continue;
1097        }
1098        let escape = *bytes
1099            .get(*cursor)
1100            .ok_or_else(|| PartitionError::MalformedKey("unterminated payload".to_owned()))?;
1101        *cursor += 1;
1102        match escape {
1103            0x00 => return Ok(payload),
1104            0xFF => payload.push(0x00),
1105            other => {
1106                return Err(PartitionError::MalformedKey(format!(
1107                    "invalid escape byte 0x{other:02x}"
1108                )))
1109            }
1110        }
1111    }
1112}
1113
1114// ---------------------------------------------------------------------------
1115// Partitioning (spec section 12.2)
1116// ---------------------------------------------------------------------------
1117
1118/// Fixed-width time interval for time-range partitioning, in microseconds.
1119///
1120/// Calendar intervals (week, month) are deliberately out of scope: they are
1121/// timezone- and calendar-dependent, while partition boundaries must be a
1122/// pure function of the timestamp.
1123#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
1124#[serde(deny_unknown_fields)]
1125pub struct TimeInterval {
1126    micros: u64,
1127}
1128
1129impl TimeInterval {
1130    /// An interval of `micros` microseconds.
1131    pub fn micros(micros: u64) -> Result<Self, TabletError> {
1132        if micros == 0 {
1133            return Err(TabletError::InvalidDescriptor(
1134                "time interval must be positive".to_owned(),
1135            ));
1136        }
1137        Ok(Self { micros })
1138    }
1139
1140    /// An interval of `hours` hours.
1141    pub fn hours(hours: u64) -> Result<Self, TabletError> {
1142        Self::micros(
1143            hours
1144                .checked_mul(3_600_000_000)
1145                .ok_or(TabletError::InvalidDescriptor(
1146                    "time interval overflows u64 micros".to_owned(),
1147                ))?,
1148        )
1149    }
1150
1151    /// An interval of `days` days.
1152    pub fn days(days: u64) -> Result<Self, TabletError> {
1153        Self::micros(
1154            days.checked_mul(86_400_000_000)
1155                .ok_or(TabletError::InvalidDescriptor(
1156                    "time interval overflows u64 micros".to_owned(),
1157                ))?,
1158        )
1159    }
1160
1161    /// The interval width in microseconds.
1162    pub const fn as_micros(self) -> u64 {
1163        self.micros
1164    }
1165}
1166
1167/// How one table's rows map onto tablets (spec section 12.2). Declaration
1168/// order is frozen; variants are never reused (spec section 4.10).
1169#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1170pub enum Partitioning {
1171    /// Hash of the declared columns into a fixed bucket space. Tablets own
1172    /// contiguous runs of buckets (see [`Partitioning::routed_key`]).
1173    Hash {
1174        /// Partition-key columns, in hash order.
1175        columns: Vec<ColumnId>,
1176        /// Bucket count; positive.
1177        buckets: u32,
1178    },
1179    /// Lexicographic ranges over the declared columns, split at `splits`.
1180    /// `splits.len() + 1` partitions cover the key space: partition `i` owns
1181    /// `[splits[i - 1], splits[i])`, unbounded at the edges.
1182    Range {
1183        /// Partition-key columns, in sort order.
1184        columns: Vec<ColumnId>,
1185        /// Ordered, strictly increasing split points.
1186        splits: Vec<Key>,
1187    },
1188    /// One bucket space per tenant: the tenant column hashes into
1189    /// `buckets_per_tenant` buckets, so every tenant's rows spread evenly and
1190    /// tenants can be isolated onto dedicated tablets (spec section 12.5).
1191    Tenant {
1192        /// Column carrying the tenant identifier.
1193        tenant_column: ColumnId,
1194        /// Buckets each tenant hashes into; positive.
1195        buckets_per_tenant: u32,
1196    },
1197    /// Time-bucketed ranges: partition `i` owns
1198    /// `[i * interval, (i + 1) * interval)` of the timestamp column.
1199    TimeRange {
1200        /// Column carrying the row timestamp.
1201        timestamp_column: ColumnId,
1202        /// Fixed interval width.
1203        interval: TimeInterval,
1204    },
1205}
1206
1207impl Partitioning {
1208    /// Validates the declaration: non-empty column lists, positive bucket
1209    /// counts, strictly increasing splits, positive interval.
1210    pub fn validate(&self) -> Result<(), PartitionError> {
1211        match self {
1212            Self::Hash { columns, buckets } => {
1213                if columns.is_empty() {
1214                    return Err(PartitionError::InvalidPartitioning(
1215                        "hash partitioning needs at least one column".to_owned(),
1216                    ));
1217                }
1218                if *buckets == 0 {
1219                    return Err(PartitionError::InvalidPartitioning(
1220                        "hash partitioning needs at least one bucket".to_owned(),
1221                    ));
1222                }
1223            }
1224            Self::Range { columns, splits } => {
1225                if columns.is_empty() {
1226                    return Err(PartitionError::InvalidPartitioning(
1227                        "range partitioning needs at least one column".to_owned(),
1228                    ));
1229                }
1230                if splits.windows(2).any(|window| window[0] >= window[1]) {
1231                    return Err(PartitionError::InvalidPartitioning(
1232                        "range splits must be strictly increasing".to_owned(),
1233                    ));
1234                }
1235            }
1236            Self::Tenant {
1237                buckets_per_tenant, ..
1238            } => {
1239                if *buckets_per_tenant == 0 {
1240                    return Err(PartitionError::InvalidPartitioning(
1241                        "tenant partitioning needs at least one bucket per tenant".to_owned(),
1242                    ));
1243                }
1244            }
1245            Self::TimeRange { interval, .. } => {
1246                if interval.as_micros() == 0 {
1247                    return Err(PartitionError::InvalidPartitioning(
1248                        "time-range interval must be positive".to_owned(),
1249                    ));
1250                }
1251            }
1252        }
1253        Ok(())
1254    }
1255
1256    /// The declared partition-key columns, in key order (spec section 12.2:
1257    /// every table has a declared partition key).
1258    pub fn partition_columns(&self) -> Vec<ColumnId> {
1259        match self {
1260            Self::Hash { columns, .. } | Self::Range { columns, .. } => columns.clone(),
1261            Self::Tenant { tenant_column, .. } => vec![*tenant_column],
1262            Self::TimeRange {
1263                timestamp_column, ..
1264            } => vec![*timestamp_column],
1265        }
1266    }
1267
1268    /// Extracts the partition key from a row's column values: the declared
1269    /// partition-key columns, in declared order, encoded by
1270    /// [`RowKeyEncoder`]. A missing declared column fails closed.
1271    ///
1272    /// The fast path of spec section 12.2 — the primary key includes the
1273    /// partition key — means the caller can usually supply the primary-key
1274    /// components directly; this function does not care where the values came
1275    /// from.
1276    pub fn partition_key(
1277        &self,
1278        values: &BTreeMap<ColumnId, KeyValue>,
1279    ) -> Result<Key, PartitionError> {
1280        self.validate()?;
1281        let columns = self.partition_columns();
1282        let mut components = Vec::with_capacity(columns.len());
1283        for column in columns {
1284            let value = values
1285                .get(&column)
1286                .ok_or(PartitionError::MissingPartitionColumn { column })?;
1287            if let Self::TimeRange { .. } = self {
1288                if !matches!(value, KeyValue::TimestampMicros(_)) {
1289                    return Err(PartitionError::PartitionColumnType {
1290                        column,
1291                        expected: "timestamp-micros",
1292                        found: value.type_name(),
1293                    });
1294                }
1295            }
1296            components.push(value.clone());
1297        }
1298        Ok(RowKeyEncoder::encode_key(&components))
1299    }
1300
1301    /// Maps an extracted partition key onto its partition slot (spec section
1302    /// 12.2):
1303    ///
1304    /// - `Hash`: `fnv1a64(key) % buckets`.
1305    /// - `Range`: index of the first split above the key, `0..=splits.len()`.
1306    /// - `Tenant`: `fnv1a64(tenant key) % buckets_per_tenant`; the slot is
1307    ///   per tenant, so the full partition address is `(tenant, slot)` — see
1308    ///   [`Self::routed_key`].
1309    /// - `TimeRange`: `timestamp / interval` (epoch-floored); pre-epoch
1310    ///   timestamps fail closed with [`PartitionError::NegativeSlot`].
1311    pub fn route(&self, partition_key: &Key) -> Result<u64, PartitionError> {
1312        self.validate()?;
1313        match self {
1314            Self::Hash { buckets, .. } => {
1315                Ok(RowKeyEncoder::fnv1a64(partition_key.as_bytes()) % u64::from(*buckets))
1316            }
1317            Self::Range { splits, .. } => {
1318                Ok(splits.partition_point(|split| split <= partition_key) as u64)
1319            }
1320            Self::Tenant {
1321                buckets_per_tenant, ..
1322            } => {
1323                Ok(RowKeyEncoder::fnv1a64(partition_key.as_bytes())
1324                    % u64::from(*buckets_per_tenant))
1325            }
1326            Self::TimeRange { interval, .. } => {
1327                let mut components = RowKeyEncoder::decode_components(partition_key)?;
1328                if components.len() != 1 {
1329                    return Err(PartitionError::MalformedKey(format!(
1330                        "time-range partition key has {} components, expected 1",
1331                        components.len()
1332                    )));
1333                }
1334                let Some(KeyValue::TimestampMicros(micros)) = components.pop() else {
1335                    return Err(PartitionError::MalformedKey(
1336                        "time-range partition key does not start with a timestamp".to_owned(),
1337                    ));
1338                };
1339                let slot =
1340                    micros.div_euclid(i64::try_from(interval.as_micros()).map_err(|_| {
1341                        PartitionError::InvalidPartitioning(
1342                            "time-range interval exceeds i64 micros".to_owned(),
1343                        )
1344                    })?);
1345                u64::try_from(slot).map_err(|_| PartitionError::NegativeSlot { micros })
1346            }
1347        }
1348    }
1349
1350    /// The canonical routing address of a partition key — the byte string
1351    /// whose ranges the tablet [`PartitionBounds`] of this table are
1352    /// allocated over:
1353    ///
1354    /// - `Range` and `TimeRange`: the partition key itself; bounds range
1355    ///   directly over encoded key bytes.
1356    /// - `Hash`: the 8 big-endian bytes of the bucket; the meta plane
1357    ///   allocates tablet bounds as contiguous bucket runs, e.g. buckets
1358    ///   `[lo, hi)` via [`hash_slot_bounds`].
1359    /// - `Tenant`: the tenant key encoding followed by the 8 big-endian
1360    ///   bytes of the bucket, so bounds range per tenant over the composite.
1361    pub fn routed_key(&self, partition_key: &Key) -> Result<Key, PartitionError> {
1362        match self {
1363            Self::Range { .. } | Self::TimeRange { .. } => Ok(partition_key.clone()),
1364            Self::Hash { .. } => Ok(Key::from_bytes(
1365                self.route(partition_key)?.to_be_bytes().to_vec(),
1366            )),
1367            Self::Tenant { .. } => {
1368                let mut bytes = partition_key.as_bytes().to_vec();
1369                bytes.extend_from_slice(&self.route(partition_key)?.to_be_bytes());
1370                Ok(Key::from_bytes(bytes))
1371            }
1372        }
1373    }
1374
1375    /// The bounds of range partition `index` (`0..=splits.len()`); `None`
1376    /// for out-of-range indexes and non-range partitioning.
1377    pub fn range_bounds(&self, index: u64) -> Option<PartitionBounds> {
1378        let Self::Range { splits, .. } = self else {
1379            return None;
1380        };
1381        let index = usize::try_from(index).ok()?;
1382        if index > splits.len() {
1383            return None;
1384        }
1385        let low = index.checked_sub(1).map_or(Bound::Unbounded, |previous| {
1386            Bound::Included(splits[previous].clone())
1387        });
1388        let high = splits
1389            .get(index)
1390            .map_or(Bound::Unbounded, |split| Bound::Excluded(split.clone()));
1391        Some(PartitionBounds { low, high })
1392    }
1393}
1394
1395/// Tablet bounds covering hash buckets `[low_slot, high_slot_exclusive)` in
1396/// the canonical [`Partitioning::routed_key`] space of a hash-partitioned
1397/// table.
1398pub fn hash_slot_bounds(low_slot: u64, high_slot_exclusive: u64) -> PartitionBounds {
1399    PartitionBounds {
1400        low: Bound::Included(Key::from_bytes(low_slot.to_be_bytes().to_vec())),
1401        high: Bound::Excluded(Key::from_bytes(high_slot_exclusive.to_be_bytes().to_vec())),
1402    }
1403}
1404
1405/// Why a table's [`Partitioning`] is what it is (spec section 12.2: automatic
1406/// defaults must be visible in schema metadata). Declaration order is frozen;
1407/// variants are never reused (spec section 4.10).
1408#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
1409pub enum PartitioningOrigin {
1410    /// Declared explicitly by `CREATE TABLE`.
1411    Declared,
1412    /// Derived automatically (hash of the primary key); recorded so the
1413    /// default is visible in schema metadata rather than implicit.
1414    AutomaticDefault,
1415}
1416
1417/// Colocation declaration (spec section 12.2: related tables may declare
1418/// colocation). A table colocated with another shares its partition layout,
1419/// so equal partition keys land on the same tablets and local joins stay
1420/// local. Colocation requires the two tables' partitionings to agree on the
1421/// partition-key types; the meta group enforces that at declaration time.
1422#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
1423#[repr(transparent)]
1424pub struct ColocatedWith(pub TableId);
1425
1426/// The per-table partitioning metadata record (spec section 12.2). One such
1427/// record lives in the replicated schema metadata of every table; the
1428/// [`PartitioningOrigin`] keeps automatic defaults visible.
1429#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1430#[serde(deny_unknown_fields)]
1431pub struct TablePartitioningRecord {
1432    /// The table this record describes.
1433    pub table_id: TableId,
1434    /// How the table's rows map onto tablets.
1435    pub partitioning: Partitioning,
1436    /// Whether the partitioning was declared or derived by default.
1437    pub origin: PartitioningOrigin,
1438    /// Colocation declaration, when the table is colocated with another.
1439    pub colocated_with: Option<ColocatedWith>,
1440}
1441
1442impl TablePartitioningRecord {
1443    /// The automatic default (spec section 12.2): hash partitioning over the
1444    /// primary-key columns, recorded with [`PartitioningOrigin::AutomaticDefault`]
1445    /// so the default is visible in schema metadata.
1446    pub fn automatic_default(
1447        table_id: TableId,
1448        primary_key_columns: Vec<ColumnId>,
1449        buckets: u32,
1450    ) -> Self {
1451        Self {
1452            table_id,
1453            partitioning: Partitioning::Hash {
1454                columns: primary_key_columns,
1455                buckets,
1456            },
1457            origin: PartitioningOrigin::AutomaticDefault,
1458            colocated_with: None,
1459        }
1460    }
1461
1462    /// Validates the record: usable table id, well-formed partitioning, no
1463    /// self-colocation.
1464    pub fn validate(&self) -> Result<(), TabletError> {
1465        if self.table_id == TableId::ZERO {
1466            return Err(TabletError::InvalidDescriptor(
1467                "reserved zero table id".to_owned(),
1468            ));
1469        }
1470        self.partitioning.validate()?;
1471        if self
1472            .colocated_with
1473            .is_some_and(|target| target.0 == self.table_id)
1474        {
1475            return Err(TabletError::InvalidDescriptor(
1476                "a table cannot be colocated with itself".to_owned(),
1477            ));
1478        }
1479        Ok(())
1480    }
1481}
1482
1483// ---------------------------------------------------------------------------
1484// Tablet routing (spec section 12.4)
1485// ---------------------------------------------------------------------------
1486
1487/// Routes a point read/write directly: the routable tablet of `table_id`
1488/// whose bounds contain `routed_key` (see [`Partitioning::routed_key`]).
1489///
1490/// A well-formed meta plane yields exactly one match; against a stale cache
1491/// copy a miss means "refresh metadata", which is why the caller resolves a
1492/// miss to a metadata refresh, not an error.
1493pub fn find_tablet_for_key<'a>(
1494    tablets: &'a [TabletDescriptor],
1495    table_id: TableId,
1496    routed_key: &Key,
1497) -> Option<&'a TabletDescriptor> {
1498    tablets.iter().find(|tablet| {
1499        tablet.table_id == table_id
1500            && tablet.state.is_routable()
1501            && tablet.partition.contains(routed_key)
1502    })
1503}
1504
1505/// Routes a range query to every overlapping tablet (spec section 12.4),
1506/// ordered deterministically by low endpoint, then tablet id.
1507pub fn tablets_overlapping<'a>(
1508    tablets: &'a [TabletDescriptor],
1509    table_id: TableId,
1510    range: &PartitionBounds,
1511) -> Vec<&'a TabletDescriptor> {
1512    let mut matched: Vec<&TabletDescriptor> = tablets
1513        .iter()
1514        .filter(|tablet| {
1515            tablet.table_id == table_id
1516                && tablet.state.is_routable()
1517                && tablet.partition.overlaps(range)
1518        })
1519        .collect();
1520    matched.sort_by(|left, right| {
1521        cmp_low_bounds(&left.partition.low, &right.partition.low)
1522            .then_with(|| left.tablet_id.cmp(&right.tablet_id))
1523    });
1524    matched
1525}
1526
1527/// Low-endpoint order for routing output: `Unbounded` sorts first, and at
1528/// equal keys an included endpoint sorts before an excluded one.
1529fn cmp_low_bounds(left: &Bound<Key>, right: &Bound<Key>) -> std::cmp::Ordering {
1530    use std::cmp::Ordering;
1531    match (left, right) {
1532        (Bound::Unbounded, Bound::Unbounded) => Ordering::Equal,
1533        (Bound::Unbounded, _) => Ordering::Less,
1534        (_, Bound::Unbounded) => Ordering::Greater,
1535        (Bound::Included(left), Bound::Included(right))
1536        | (Bound::Excluded(left), Bound::Excluded(right)) => left.cmp(right),
1537        (Bound::Included(left), Bound::Excluded(right)) => left.cmp(right).then(Ordering::Less),
1538        (Bound::Excluded(left), Bound::Included(right)) => left.cmp(right).then(Ordering::Greater),
1539    }
1540}
1541
1542/// Verifies the generation a request was routed with against the descriptor
1543/// the receiving replica holds (spec section 12.4: every request includes
1544/// the tablet generation it used).
1545///
1546/// A match passes. A mismatch is classified from the tablet's current state:
1547/// a source in [`TabletState::Splitting`] reports [`RoutingError::TabletSplit`],
1548/// a retired or retiring tablet reports [`RoutingError::TabletMoved`], and
1549/// anything else — including a request newer than the replica's descriptor,
1550/// which means the replica is behind — reports
1551/// [`RoutingError::StaleMetadata`]. The gateway refreshes metadata and
1552/// retries safe operations on all three.
1553pub fn check_generation(
1554    descriptor: &TabletDescriptor,
1555    used_generation: u64,
1556) -> Result<(), RoutingError> {
1557    if used_generation == descriptor.generation {
1558        return Ok(());
1559    }
1560    let tablet_id = descriptor.tablet_id;
1561    let current_generation = descriptor.generation;
1562    // Review N2: a request generation *newer* than the replica's descriptor
1563    // means the replica is behind (StaleMetadata), even while Splitting —
1564    // only an *older* request against a splitting source is TabletSplit.
1565    Err(match descriptor.state {
1566        TabletState::Splitting if used_generation < current_generation => {
1567            RoutingError::TabletSplit {
1568                tablet_id,
1569                used_generation,
1570                current_generation,
1571            }
1572        }
1573        TabletState::Retiring | TabletState::Retired => RoutingError::TabletMoved {
1574            tablet_id,
1575            used_generation,
1576            current_generation,
1577        },
1578        _ => RoutingError::StaleMetadata {
1579            tablet_id,
1580            used_generation,
1581            current_generation,
1582        },
1583    })
1584}
1585
1586// ---------------------------------------------------------------------------
1587// Tablet local storage (spec section 12.3)
1588// ---------------------------------------------------------------------------
1589
1590/// Name of the per-node tablet directory root under the node data dir.
1591pub const TABLETS_DIR: &str = "tablets";
1592/// Name of the per-node Raft-group directory root under the node data dir.
1593pub const GROUPS_DIR: &str = "groups";
1594/// Subdirectories of one tablet directory.
1595pub const TABLET_STATE_DIR: &str = "state";
1596/// Sorted-run subdirectory of one tablet directory.
1597pub const TABLET_RUNS_DIR: &str = "runs";
1598/// Local index-generation subdirectory of one tablet directory.
1599pub const TABLET_INDEXES_DIR: &str = "indexes";
1600/// Temporary/spill subdirectory of one tablet directory.
1601pub const TABLET_TEMP_DIR: &str = "temp";
1602/// Raft log subdirectory of one group directory.
1603pub const GROUP_RAFT_DIR: &str = "raft";
1604/// Snapshot subdirectory of one group directory.
1605///
1606/// Historical name kept for layout docs; the consensus state machine stores
1607/// snapshots under `groups/<id>/raft/snapshot` (review **N3**). Prefer
1608/// [`TabletLayout::raft_snapshot_dir`].
1609pub const GROUP_SNAPSHOTS_DIR: &str = "snapshots";
1610/// Name of the versioned, checksummed tablet metadata file.
1611pub const TABLET_META_FILENAME: &str = "tablet.json";
1612/// The tablet-metadata format version this build writes.
1613pub const TABLET_META_FORMAT_VERSION: u32 = 1;
1614/// The oldest tablet-metadata format version this build accepts.
1615pub const MIN_SUPPORTED_TABLET_META_FORMAT_VERSION: u32 = 1;
1616
1617/// On-node directory layout of one tablet replica and its Raft group (spec
1618/// section 12.3):
1619///
1620/// ```text
1621/// node-data/
1622///   tablets/<tablet-id>/{state,runs,indexes,temp}   + tablet.json
1623///   groups/<raft-group-id>/{raft,snapshots}
1624/// ```
1625///
1626/// The applied MVCC state, sorted runs, local index generations, and
1627/// compaction state live under the tablet directory; the Raft log and
1628/// snapshots live under the group directory, so a group can outlive any
1629/// single applied-state rebuild (spec section 4.4: runs and indexes are
1630/// applied state and may be rebuilt).
1631#[derive(Clone, Debug, PartialEq, Eq)]
1632pub struct TabletLayout {
1633    node_data: PathBuf,
1634    tablet_id: TabletId,
1635    raft_group_id: RaftGroupId,
1636}
1637
1638impl TabletLayout {
1639    /// The layout of `tablet_id`/`raft_group_id` under `node_data`.
1640    pub fn new(
1641        node_data: impl Into<PathBuf>,
1642        tablet_id: TabletId,
1643        raft_group_id: RaftGroupId,
1644    ) -> Self {
1645        Self {
1646            node_data: node_data.into(),
1647            tablet_id,
1648            raft_group_id,
1649        }
1650    }
1651
1652    /// The tablet this layout belongs to.
1653    pub fn tablet_id(&self) -> TabletId {
1654        self.tablet_id
1655    }
1656
1657    /// The Raft group this layout belongs to.
1658    pub fn raft_group_id(&self) -> RaftGroupId {
1659        self.raft_group_id
1660    }
1661
1662    /// The node data root the layout lives under.
1663    pub fn node_data(&self) -> &Path {
1664        &self.node_data
1665    }
1666
1667    /// `node-data/tablets/<tablet-id>`.
1668    pub fn tablet_dir(&self) -> PathBuf {
1669        self.node_data
1670            .join(TABLETS_DIR)
1671            .join(self.tablet_id.to_hex())
1672    }
1673
1674    /// `tablets/<tablet-id>/state` (applied MVCC + compaction state).
1675    pub fn state_dir(&self) -> PathBuf {
1676        self.tablet_dir().join(TABLET_STATE_DIR)
1677    }
1678
1679    /// `tablets/<tablet-id>/runs` (sorted runs).
1680    pub fn runs_dir(&self) -> PathBuf {
1681        self.tablet_dir().join(TABLET_RUNS_DIR)
1682    }
1683
1684    /// `tablets/<tablet-id>/indexes` (local index generations).
1685    pub fn indexes_dir(&self) -> PathBuf {
1686        self.tablet_dir().join(TABLET_INDEXES_DIR)
1687    }
1688
1689    /// `tablets/<tablet-id>/temp` (spill scratch space).
1690    pub fn temp_dir(&self) -> PathBuf {
1691        self.tablet_dir().join(TABLET_TEMP_DIR)
1692    }
1693
1694    /// `node-data/groups/<raft-group-id>`.
1695    pub fn group_dir(&self) -> PathBuf {
1696        self.node_data
1697            .join(GROUPS_DIR)
1698            .join(self.raft_group_id.to_hex())
1699    }
1700
1701    /// `groups/<raft-group-id>/raft` (Raft log).
1702    pub fn raft_dir(&self) -> PathBuf {
1703        self.group_dir().join(GROUP_RAFT_DIR)
1704    }
1705
1706    /// Legacy layout path `groups/<raft-group-id>/snapshots` (unused by the
1707    /// consensus SM). Prefer [`Self::raft_snapshot_dir`] (review **N3**).
1708    pub fn snapshots_dir(&self) -> PathBuf {
1709        self.group_dir().join(GROUP_SNAPSHOTS_DIR)
1710    }
1711
1712    /// `groups/<raft-group-id>/raft/snapshot` — the path the consensus state
1713    /// machine actually writes (review **N3** reconciliation).
1714    pub fn raft_snapshot_dir(&self) -> PathBuf {
1715        self.raft_dir().join("snapshot")
1716    }
1717
1718    /// `tablets/<tablet-id>/tablet.json`.
1719    pub fn metadata_path(&self) -> PathBuf {
1720        self.tablet_dir().join(TABLET_META_FILENAME)
1721    }
1722
1723    /// Every directory spec section 12.3 requires, tablet dirs first.
1724    fn required_dirs(&self) -> [PathBuf; 6] {
1725        [
1726            self.state_dir(),
1727            self.runs_dir(),
1728            self.indexes_dir(),
1729            self.temp_dir(),
1730            self.raft_dir(),
1731            self.snapshots_dir(),
1732        ]
1733    }
1734
1735    /// Creates the directory tree and persists the initial tablet metadata.
1736    ///
1737    /// The metadata file is created atomically and only if absent (the same
1738    /// hard-link publish idiom as the node identity); a concurrent or
1739    /// repeated `create` that finds identical metadata succeeds, while
1740    /// different existing metadata fails closed with
1741    /// [`TabletError::MetadataConflict`] — a tablet directory is never
1742    /// silently repurposed.
1743    pub fn create(&self, descriptor: &TabletDescriptor) -> Result<(), TabletError> {
1744        self.check_descriptor_identity(descriptor)?;
1745        descriptor.validate()?;
1746        for dir in self.required_dirs() {
1747            fs::create_dir_all(&dir).map_err(ClusterError::Io)?;
1748        }
1749        let file = TabletMetaFile::envelope(descriptor)?;
1750        let bytes = crate::node::encode_json(TABLET_META_FILENAME, &file)?;
1751        match crate::node::create_meta_file(&self.tablet_dir(), TABLET_META_FILENAME, &bytes)
1752            .map_err(ClusterError::Io)?
1753        {
1754            true => Ok(()),
1755            // Lost a create race or re-run: accept identical metadata only.
1756            false => match self.load_metadata()? {
1757                existing if existing == *descriptor => Ok(()),
1758                _ => Err(TabletError::MetadataConflict(self.tablet_dir())),
1759            },
1760        }
1761    }
1762
1763    /// Atomically replaces the persisted metadata (used by the meta-driven
1764    /// apply path when the descriptor's generation advances). The descriptor
1765    /// is validated and must name this layout's tablet and group.
1766    pub fn store_metadata(&self, descriptor: &TabletDescriptor) -> Result<(), TabletError> {
1767        self.check_descriptor_identity(descriptor)?;
1768        descriptor.validate()?;
1769        let file = TabletMetaFile::envelope(descriptor)?;
1770        let bytes = crate::node::encode_json(TABLET_META_FILENAME, &file)?;
1771        crate::node::write_meta_atomic(&self.tablet_dir(), TABLET_META_FILENAME, &bytes)
1772            .map_err(ClusterError::Io)?;
1773        Ok(())
1774    }
1775
1776    /// Loads and verifies the persisted tablet metadata. Missing, corrupt,
1777    /// unknown-version, wrong-checksum, or wrong-tablet files all fail closed
1778    /// (spec sections 4.10, 12.3).
1779    pub fn load_metadata(&self) -> Result<TabletDescriptor, TabletError> {
1780        let Some(bytes) = crate::node::read_meta_file(&self.metadata_path())? else {
1781            return Err(TabletError::MissingMetadata(self.metadata_path()));
1782        };
1783        let file: TabletMetaFile = crate::node::decode_json(TABLET_META_FILENAME, &bytes)?;
1784        if file.format_version < MIN_SUPPORTED_TABLET_META_FORMAT_VERSION
1785            || file.format_version > TABLET_META_FORMAT_VERSION
1786        {
1787            return Err(ClusterError::UnsupportedFormatVersion {
1788                file: TABLET_META_FILENAME,
1789                found: file.format_version,
1790                min: MIN_SUPPORTED_TABLET_META_FORMAT_VERSION,
1791                max: TABLET_META_FORMAT_VERSION,
1792            }
1793            .into());
1794        }
1795        if file.checksum != tablet_checksum(&file.tablet)? {
1796            return Err(ClusterError::CorruptMetadata {
1797                file: TABLET_META_FILENAME,
1798                detail: "checksum mismatch".to_owned(),
1799            }
1800            .into());
1801        }
1802        self.check_descriptor_identity(&file.tablet)?;
1803        file.tablet.validate()?;
1804        Ok(file.tablet)
1805    }
1806
1807    /// Opens the layout for use: verifies the persisted metadata and the
1808    /// presence of every required directory, failing closed on any missing
1809    /// or corrupt piece (spec section 12.3).
1810    pub fn validate(&self) -> Result<TabletDescriptor, TabletError> {
1811        let descriptor = self.load_metadata()?;
1812        for dir in self.required_dirs() {
1813            if !dir.is_dir() {
1814                return Err(ClusterError::CorruptMetadata {
1815                    file: TABLET_META_FILENAME,
1816                    detail: format!("required directory {} is missing", dir.display()),
1817                }
1818                .into());
1819            }
1820        }
1821        Ok(descriptor)
1822    }
1823
1824    /// The descriptor must name exactly this layout's tablet and group.
1825    fn check_descriptor_identity(&self, descriptor: &TabletDescriptor) -> Result<(), TabletError> {
1826        if descriptor.tablet_id != self.tablet_id || descriptor.raft_group_id != self.raft_group_id
1827        {
1828            return Err(TabletError::TabletMismatch {
1829                path: self.tablet_dir(),
1830                expected: self.tablet_id,
1831                found: descriptor.tablet_id,
1832                expected_group: self.raft_group_id,
1833                found_group: descriptor.raft_group_id,
1834            });
1835        }
1836        Ok(())
1837    }
1838
1839    /// Removes the local replica: the tablet directory and the Raft-group
1840    /// directory (spec section 12.5 step 11; the consensus membership removal
1841    /// is the runtime's half). Idempotent — a partially torn-down or absent
1842    /// directory is fine — but never destructive across identity: a
1843    /// `tablet.json` that names a different tablet or group fails closed with
1844    /// [`TabletError::TabletMismatch`] instead of deleting foreign state.
1845    pub fn teardown(&self) -> Result<(), TabletError> {
1846        if self.tablet_dir().is_dir() {
1847            match self.load_metadata() {
1848                Ok(_) | Err(TabletError::MissingMetadata(_)) => {}
1849                // Corrupt or foreign metadata blocks automatic teardown.
1850                Err(error) => return Err(error),
1851            }
1852            fs::remove_dir_all(self.tablet_dir()).map_err(ClusterError::Io)?;
1853        }
1854        if self.group_dir().is_dir() {
1855            fs::remove_dir_all(self.group_dir()).map_err(ClusterError::Io)?;
1856        }
1857        Ok(())
1858    }
1859}
1860
1861/// The durable tablet metadata envelope: versioned and checksummed, so a
1862/// torn or tampered file fails closed instead of opening a wrong tablet.
1863#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1864#[serde(deny_unknown_fields)]
1865struct TabletMetaFile {
1866    /// Durable format version; see [`TABLET_META_FORMAT_VERSION`].
1867    format_version: u32,
1868    /// Lowercase-hex SHA-256 of the canonical JSON encoding of `tablet`.
1869    checksum: String,
1870    /// The persisted descriptor.
1871    tablet: TabletDescriptor,
1872}
1873
1874impl TabletMetaFile {
1875    fn envelope(tablet: &TabletDescriptor) -> Result<Self, TabletError> {
1876        Ok(Self {
1877            format_version: TABLET_META_FORMAT_VERSION,
1878            checksum: tablet_checksum(tablet)?,
1879            tablet: tablet.clone(),
1880        })
1881    }
1882}
1883
1884/// SHA-256 of the canonical (compact JSON) encoding of the descriptor.
1885fn tablet_checksum(tablet: &TabletDescriptor) -> Result<String, ClusterError> {
1886    let bytes = serde_json::to_vec(tablet).map_err(|error| ClusterError::CorruptMetadata {
1887        file: TABLET_META_FILENAME,
1888        detail: format!("encode: {error}"),
1889    })?;
1890    Ok(hex_encode(&Sha256::digest(&bytes)))
1891}
1892
1893/// Scan `<node_data>/tablets/*/tablet.json` and load every valid descriptor.
1894///
1895/// Used by the §15 `SHOW TABLETS` / `SHOW REPLICAS` admin surface when a
1896/// tablet runtime has persisted local metadata. Missing tablets dir is an
1897/// empty list (standalone or not yet hosting tablets); corrupt individual
1898/// files are skipped and counted in the returned issues.
1899pub fn list_tablets_on_disk(
1900    node_data: impl AsRef<Path>,
1901) -> Result<(Vec<TabletDescriptor>, Vec<String>), ClusterError> {
1902    let root = node_data.as_ref().join(TABLETS_DIR);
1903    if !root.is_dir() {
1904        return Ok((Vec::new(), Vec::new()));
1905    }
1906    let mut tablets = Vec::new();
1907    let mut issues = Vec::new();
1908    let entries = fs::read_dir(&root).map_err(ClusterError::Io)?;
1909    for entry in entries {
1910        let entry = entry.map_err(ClusterError::Io)?;
1911        let path = entry.path();
1912        if !path.is_dir() {
1913            continue;
1914        }
1915        let meta = path.join(TABLET_META_FILENAME);
1916        if !meta.is_file() {
1917            continue;
1918        }
1919        // Load without TabletLayout identity check (group id unknown until
1920        // decode). Verify format/checksum via the same envelope path.
1921        match load_tablet_meta_file(&meta) {
1922            Ok(descriptor) => tablets.push(descriptor),
1923            Err(error) => issues.push(format!("{}: {error}", meta.display())),
1924        }
1925    }
1926    tablets.sort_by_key(|t| t.tablet_id);
1927    Ok((tablets, issues))
1928}
1929
1930fn load_tablet_meta_file(path: &Path) -> Result<TabletDescriptor, ClusterError> {
1931    let Some(bytes) = crate::node::read_meta_file(path)? else {
1932        return Err(ClusterError::CorruptMetadata {
1933            file: TABLET_META_FILENAME,
1934            detail: format!("missing {}", path.display()),
1935        });
1936    };
1937    let file: TabletMetaFile = crate::node::decode_json(TABLET_META_FILENAME, &bytes)?;
1938    if file.format_version < MIN_SUPPORTED_TABLET_META_FORMAT_VERSION
1939        || file.format_version > TABLET_META_FORMAT_VERSION
1940    {
1941        return Err(ClusterError::UnsupportedFormatVersion {
1942            file: TABLET_META_FILENAME,
1943            found: file.format_version,
1944            min: MIN_SUPPORTED_TABLET_META_FORMAT_VERSION,
1945            max: TABLET_META_FORMAT_VERSION,
1946        });
1947    }
1948    if file.checksum != tablet_checksum(&file.tablet)? {
1949        return Err(ClusterError::CorruptMetadata {
1950            file: TABLET_META_FILENAME,
1951            detail: "checksum mismatch".to_owned(),
1952        });
1953    }
1954    file.tablet
1955        .validate()
1956        .map_err(|e| ClusterError::CorruptMetadata {
1957            file: TABLET_META_FILENAME,
1958            detail: e.to_string(),
1959        })?;
1960    Ok(file.tablet)
1961}
1962
1963/// Process-local registry of owned tablet storage cores (spec section 12.3:
1964/// one tablet storage core is owned by one node process).
1965///
1966/// This is the in-process half of the rule, mirroring the open-reservation
1967/// concept of the Stage 1 shared-core registry (S1A-002): while a
1968/// [`TabletOwnershipGuard`] is alive, a second reservation of the same
1969/// canonical tablet directory fails closed with
1970/// [`TabletError::AlreadyOwned`]. The cross-process half stays with the
1971/// storage core's file lease (`_meta/.lock`), exactly as in single-node
1972/// mode; this registry never touches the file system beyond canonicalizing
1973/// the tablet directory.
1974#[derive(Debug, Default)]
1975pub struct TabletOwnershipRegistry {
1976    reservations: Mutex<HashMap<PathBuf, TabletId>>,
1977}
1978
1979impl TabletOwnershipRegistry {
1980    /// An empty registry.
1981    pub fn new() -> Self {
1982        Self::default()
1983    }
1984
1985    /// The process-global registry.
1986    pub fn global() -> &'static Self {
1987        static REGISTRY: OnceLock<TabletOwnershipRegistry> = OnceLock::new();
1988        REGISTRY.get_or_init(Self::new)
1989    }
1990
1991    /// Reserves `layout`'s tablet directory for this process. The directory
1992    /// must already exist (create the layout first); the reservation keys on
1993    /// the canonical path so aliased paths cannot double-open a tablet.
1994    pub fn try_reserve(
1995        &self,
1996        layout: &TabletLayout,
1997    ) -> Result<TabletOwnershipGuard<'_>, TabletError> {
1998        let path = layout
1999            .tablet_dir()
2000            .canonicalize()
2001            .map_err(ClusterError::Io)?;
2002        let mut reservations = self
2003            .reservations
2004            .lock()
2005            .expect("tablet ownership registry lock poisoned");
2006        if let Some(holder) = reservations.get(&path) {
2007            return Err(TabletError::AlreadyOwned {
2008                tablet: *holder,
2009                path,
2010            });
2011        }
2012        reservations.insert(path.clone(), layout.tablet_id());
2013        Ok(TabletOwnershipGuard {
2014            registry: self,
2015            path,
2016            tablet_id: layout.tablet_id(),
2017        })
2018    }
2019
2020    /// Number of live reservations (diagnostics).
2021    pub fn len(&self) -> usize {
2022        self.reservations
2023            .lock()
2024            .expect("tablet ownership registry lock poisoned")
2025            .len()
2026    }
2027
2028    /// Whether no tablet storage core is reserved.
2029    pub fn is_empty(&self) -> bool {
2030        self.len() == 0
2031    }
2032}
2033
2034/// RAII reservation of one tablet directory; dropping releases the
2035/// reservation so the tablet may be reopened.
2036#[derive(Debug)]
2037pub struct TabletOwnershipGuard<'a> {
2038    registry: &'a TabletOwnershipRegistry,
2039    path: PathBuf,
2040    tablet_id: TabletId,
2041}
2042
2043impl TabletOwnershipGuard<'_> {
2044    /// The tablet whose storage core this guard owns.
2045    pub fn tablet_id(&self) -> TabletId {
2046        self.tablet_id
2047    }
2048
2049    /// The canonical reserved path.
2050    pub fn path(&self) -> &Path {
2051        &self.path
2052    }
2053}
2054
2055impl Drop for TabletOwnershipGuard<'_> {
2056    fn drop(&mut self) {
2057        let mut reservations = self
2058            .registry
2059            .reservations
2060            .lock()
2061            .expect("tablet ownership registry lock poisoned");
2062        if reservations.get(&self.path) == Some(&self.tablet_id) {
2063            reservations.remove(&self.path);
2064        }
2065    }
2066}
2067
2068// ---------------------------------------------------------------------------
2069// Tests
2070// ---------------------------------------------------------------------------
2071
2072#[cfg(test)]
2073mod tests {
2074    use super::*;
2075
2076    fn node(byte: u8) -> NodeId {
2077        NodeId::from_bytes([byte; 16])
2078    }
2079
2080    fn tablet_id(byte: u8) -> TabletId {
2081        TabletId::from_bytes([byte; 16])
2082    }
2083
2084    fn group_id(byte: u8) -> RaftGroupId {
2085        RaftGroupId::from_bytes([byte; 16])
2086    }
2087
2088    fn key(bytes: &[u8]) -> Key {
2089        Key::from_bytes(bytes.to_vec())
2090    }
2091
2092    fn text_key(text: &str) -> Key {
2093        RowKeyEncoder::encode_key(&[KeyValue::Text(text.to_owned())])
2094    }
2095
2096    fn int_key(value: i64) -> Key {
2097        RowKeyEncoder::encode_key(&[KeyValue::Int(value)])
2098    }
2099
2100    fn bounds(low: Bound<Key>, high: Bound<Key>) -> PartitionBounds {
2101        PartitionBounds::new(low, high).unwrap()
2102    }
2103
2104    fn descriptor(state: TabletState) -> TabletDescriptor {
2105        TabletDescriptor {
2106            tablet_id: tablet_id(9),
2107            table_id: TableId::new(3),
2108            database_id: mongreldb_types::ids::DatabaseId::ZERO,
2109            raft_group_id: group_id(7),
2110            partition: bounds(
2111                Bound::Included(text_key("a")),
2112                Bound::Excluded(text_key("m")),
2113            ),
2114            replicas: vec![
2115                ReplicaDescriptor {
2116                    node_id: node(1),
2117                    role: ReplicaRole::Voter,
2118                    raft_node_id: 11,
2119                },
2120                ReplicaDescriptor {
2121                    node_id: node(2),
2122                    role: ReplicaRole::Voter,
2123                    raft_node_id: 12,
2124                },
2125                ReplicaDescriptor {
2126                    node_id: node(3),
2127                    role: ReplicaRole::Learner,
2128                    raft_node_id: 13,
2129                },
2130            ],
2131            leader_hint: Some(node(1)),
2132            generation: 7,
2133            state,
2134        }
2135    }
2136
2137    // -- descriptor serde --------------------------------------------------
2138
2139    #[test]
2140    fn descriptor_round_trips_serde_in_every_state() {
2141        for state in [
2142            TabletState::Creating,
2143            TabletState::Active,
2144            TabletState::Splitting,
2145            TabletState::Merging,
2146            TabletState::Retiring,
2147            TabletState::Retired,
2148        ] {
2149            let descriptor = descriptor(state);
2150            let json = serde_json::to_vec(&descriptor).unwrap();
2151            let back: TabletDescriptor = serde_json::from_slice(&json).unwrap();
2152            assert_eq!(back, descriptor);
2153        }
2154    }
2155
2156    #[test]
2157    fn descriptor_rejects_unknown_fields() {
2158        let descriptor = descriptor(TabletState::Active);
2159        let mut value: serde_json::Value =
2160            serde_json::from_slice(&serde_json::to_vec(&descriptor).unwrap()).unwrap();
2161        value["unexpected"] = serde_json::json!(1);
2162        assert!(serde_json::from_value::<TabletDescriptor>(value).is_err());
2163    }
2164
2165    // -- descriptor validation ---------------------------------------------
2166
2167    #[test]
2168    fn descriptor_validation_catches_structural_violations() {
2169        let mut zero_tablet = descriptor(TabletState::Active);
2170        zero_tablet.tablet_id = TabletId::ZERO;
2171        assert!(matches!(
2172            zero_tablet.validate(),
2173            Err(TabletError::InvalidDescriptor(_))
2174        ));
2175
2176        let mut duplicate_nodes = descriptor(TabletState::Active);
2177        duplicate_nodes.replicas[1].node_id = node(1);
2178        assert!(matches!(
2179            duplicate_nodes.validate(),
2180            Err(TabletError::InvalidDescriptor(_))
2181        ));
2182
2183        let mut duplicate_raft_ids = descriptor(TabletState::Active);
2184        duplicate_raft_ids.replicas[1].raft_node_id = 11;
2185        assert!(matches!(
2186            duplicate_raft_ids.validate(),
2187            Err(TabletError::InvalidDescriptor(_))
2188        ));
2189
2190        let mut foreign_leader = descriptor(TabletState::Active);
2191        foreign_leader.leader_hint = Some(node(8));
2192        assert!(matches!(
2193            foreign_leader.validate(),
2194            Err(TabletError::InvalidDescriptor(_))
2195        ));
2196
2197        // A non-creating tablet needs at least one voter.
2198        let mut learner_only = descriptor(TabletState::Active);
2199        learner_only.replicas.iter_mut().for_each(|replica| {
2200            replica.role = ReplicaRole::Learner;
2201        });
2202        assert!(matches!(
2203            learner_only.validate(),
2204            Err(TabletError::InvalidDescriptor(_))
2205        ));
2206        // In Creating, an all-learner replica set is the norm (spec 12.5).
2207        let mut creating = learner_only.clone();
2208        creating.state = TabletState::Creating;
2209        creating.validate().unwrap();
2210
2211        descriptor(TabletState::Active).validate().unwrap();
2212    }
2213
2214    // -- tablet state transition graph --------------------------------------
2215
2216    #[test]
2217    fn transition_graph_allows_exactly_the_documented_edges() {
2218        use TabletState::{Active, Creating, Merging, Retired, Retiring, Splitting};
2219        let allowed = [
2220            (Creating, Active),
2221            (Creating, Retired),
2222            (Active, Splitting),
2223            (Active, Merging),
2224            (Active, Retiring),
2225            (Splitting, Active),
2226            (Splitting, Retiring),
2227            (Merging, Active),
2228            (Merging, Retiring),
2229            (Retiring, Retired),
2230        ];
2231        let states = [Creating, Active, Splitting, Merging, Retiring, Retired];
2232        for from in states {
2233            for to in states {
2234                assert_eq!(
2235                    from.can_transition_to(to),
2236                    allowed.contains(&(from, to)),
2237                    "unexpected graph edge {from} -> {to}"
2238                );
2239            }
2240        }
2241    }
2242
2243    #[test]
2244    fn try_transition_enforces_the_graph() {
2245        let mut tablet = descriptor(TabletState::Active);
2246        let error = tablet.try_transition(TabletState::Creating).unwrap_err();
2247        assert!(matches!(
2248            error,
2249            TabletError::InvalidStateTransition { tablet, from, to }
2250                if tablet == tablet_id(9)
2251                    && from == TabletState::Active
2252                    && to == TabletState::Creating
2253        ));
2254        tablet.try_transition(TabletState::Splitting).unwrap();
2255        tablet.try_transition(TabletState::Retiring).unwrap();
2256        tablet.try_transition(TabletState::Retired).unwrap();
2257        // Retired is terminal.
2258        assert!(tablet.try_transition(TabletState::Active).is_err());
2259    }
2260
2261    #[test]
2262    fn published_transition_stages_an_atomic_publication() {
2263        let tablet = descriptor(TabletState::Active);
2264        let splitting = tablet.published_transition(TabletState::Splitting).unwrap();
2265        // The staged copy advanced; the original is untouched.
2266        assert_eq!(splitting.state, TabletState::Splitting);
2267        assert_eq!(splitting.generation, 8);
2268        assert_eq!(tablet.state, TabletState::Active);
2269        assert_eq!(tablet.generation, 7);
2270        splitting.validate().unwrap();
2271
2272        // The graph still applies: Active -> Creating is not an edge.
2273        assert!(matches!(
2274            tablet.published_transition(TabletState::Creating),
2275            Err(TabletError::InvalidStateTransition { .. })
2276        ));
2277    }
2278
2279    #[test]
2280    fn routability_matches_the_split_merge_protocol() {
2281        assert!(TabletState::Active.is_routable());
2282        assert!(TabletState::Splitting.is_routable());
2283        assert!(TabletState::Merging.is_routable());
2284        assert!(!TabletState::Creating.is_routable());
2285        assert!(!TabletState::Retiring.is_routable());
2286        assert!(!TabletState::Retired.is_routable());
2287    }
2288
2289    // -- partition bounds ---------------------------------------------------
2290
2291    #[test]
2292    fn bounds_containment_respects_inclusion() {
2293        let range = bounds(Bound::Included(key(b"b")), Bound::Excluded(key(b"f")));
2294        assert!(!range.contains(&key(b"a")));
2295        assert!(range.contains(&key(b"b")));
2296        assert!(range.contains(&key(b"e")));
2297        assert!(!range.contains(&key(b"f")));
2298        assert!(!range.contains(&key(b"z")));
2299
2300        let open = bounds(Bound::Excluded(key(b"b")), Bound::Included(key(b"f")));
2301        assert!(!open.contains(&key(b"b")));
2302        assert!(open.contains(&key(b"f")));
2303
2304        let everything = PartitionBounds::unbounded();
2305        assert!(everything.contains(&key(b"")));
2306        assert!(everything.contains(&key(b"zzzz")));
2307    }
2308
2309    #[test]
2310    fn bounds_overlap_and_adjacency_matrix() {
2311        let lower = bounds(Bound::Included(key(b"a")), Bound::Excluded(key(b"c")));
2312        let upper = bounds(Bound::Included(key(b"c")), Bound::Excluded(key(b"e")));
2313        // [a, c) and [c, e): adjacent, no overlap.
2314        assert!(!lower.overlaps(&upper));
2315        assert!(lower.is_adjacent_to(&upper));
2316        assert!(upper.is_adjacent_to(&lower));
2317
2318        // [a, c] and [c, e): overlap at c, not adjacent.
2319        let lower_closed = bounds(Bound::Included(key(b"a")), Bound::Included(key(b"c")));
2320        assert!(lower_closed.overlaps(&upper));
2321        assert!(!lower_closed.is_adjacent_to(&upper));
2322
2323        // [a, c) and (c, e]: c uncovered: neither overlap nor adjacent.
2324        let upper_open = bounds(Bound::Excluded(key(b"c")), Bound::Included(key(b"e")));
2325        assert!(!lower.overlaps(&upper_open));
2326        assert!(!lower.is_adjacent_to(&upper_open));
2327
2328        // Nested ranges overlap; unbounded overlaps everything.
2329        let nested = bounds(Bound::Included(key(b"a0")), Bound::Excluded(key(b"b")));
2330        assert!(lower.overlaps(&nested));
2331        assert!(PartitionBounds::unbounded().overlaps(&lower));
2332        assert!(!PartitionBounds::unbounded().is_adjacent_to(&lower));
2333
2334        // Disjoint ranges with a gap.
2335        let far = bounds(Bound::Included(key(b"x")), Bound::Unbounded);
2336        assert!(!lower.overlaps(&far));
2337        assert!(!lower.is_adjacent_to(&far));
2338    }
2339
2340    #[test]
2341    fn bounds_validation_rejects_empty_and_inverted_ranges() {
2342        assert!(matches!(
2343            PartitionBounds::new(Bound::Included(key(b"m")), Bound::Excluded(key(b"a"))),
2344            Err(TabletError::InvalidBounds(_))
2345        ));
2346        // [k, k) is empty.
2347        assert!(matches!(
2348            PartitionBounds::new(Bound::Included(key(b"k")), Bound::Excluded(key(b"k"))),
2349            Err(TabletError::InvalidBounds(_))
2350        ));
2351        // (k, k] and (k, k) are empty.
2352        assert!(
2353            PartitionBounds::new(Bound::Excluded(key(b"k")), Bound::Included(key(b"k"))).is_err()
2354        );
2355        // [k, k] is the single point k.
2356        let point = bounds(Bound::Included(key(b"k")), Bound::Included(key(b"k")));
2357        assert!(point.contains(&key(b"k")));
2358        assert!(!point.contains(&key(b"j")));
2359    }
2360
2361    #[test]
2362    fn split_at_partitions_the_range_with_no_gap_or_overlap() {
2363        let range = bounds(Bound::Included(key(b"b")), Bound::Excluded(key(b"f")));
2364        let (lower, upper) = range.split_at(&key(b"d")).unwrap();
2365        assert_eq!(
2366            lower,
2367            bounds(Bound::Included(key(b"b")), Bound::Excluded(key(b"d")))
2368        );
2369        assert_eq!(
2370            upper,
2371            bounds(Bound::Included(key(b"d")), Bound::Excluded(key(b"f")))
2372        );
2373        // The halves meet at the split key: adjacent, never overlapping.
2374        assert!(lower.meets_start_of(&upper));
2375        assert!(lower.is_adjacent_to(&upper));
2376        assert!(!lower.overlaps(&upper));
2377        for candidate in [b"a", b"b", b"c", b"d", b"e", b"f", b"g"] {
2378            let candidate = key(candidate);
2379            assert_eq!(
2380                range.contains(&candidate),
2381                lower.contains(&candidate) || upper.contains(&candidate),
2382                "coverage mismatch at {candidate}"
2383            );
2384            assert!(
2385                !(lower.contains(&candidate) && upper.contains(&candidate)),
2386                "double coverage at {candidate}"
2387            );
2388        }
2389
2390        // Unbounded sides split fine; the split key itself lands in the upper half.
2391        let whole = PartitionBounds::unbounded();
2392        let (low_half, high_half) = whole.split_at(&key(b"k")).unwrap();
2393        assert!(matches!(low_half.low, Bound::Unbounded));
2394        assert!(matches!(high_half.high, Bound::Unbounded));
2395        assert!(!low_half.contains(&key(b"k")));
2396        assert!(high_half.contains(&key(b"k")));
2397
2398        // Keys outside the range, or at the very edge, never split.
2399        assert!(range.split_at(&key(b"a")).is_none());
2400        assert!(range.split_at(&key(b"f")).is_none());
2401        assert!(range.split_at(&key(b"b")).is_none()); // would empty the lower half
2402                                                       // A single-point range cannot split.
2403        let point = bounds(Bound::Included(key(b"b")), Bound::Included(key(b"b")));
2404        assert!(point.split_at(&key(b"b")).is_none());
2405        // An included high endpoint may split: the upper half is the point.
2406        let closed = bounds(Bound::Included(key(b"b")), Bound::Included(key(b"f")));
2407        let (_, upper) = closed.split_at(&key(b"f")).unwrap();
2408        assert_eq!(
2409            upper,
2410            bounds(Bound::Included(key(b"f")), Bound::Included(key(b"f")))
2411        );
2412    }
2413
2414    #[test]
2415    fn union_adjacent_combines_only_adjacent_ranges() {
2416        let lower = bounds(Bound::Included(key(b"a")), Bound::Excluded(key(b"c")));
2417        let upper = bounds(Bound::Included(key(b"c")), Bound::Excluded(key(b"e")));
2418        let combined = bounds(Bound::Included(key(b"a")), Bound::Excluded(key(b"e")));
2419        // Order-independent.
2420        assert_eq!(lower.union_adjacent(&upper).unwrap(), combined);
2421        assert_eq!(upper.union_adjacent(&lower).unwrap(), combined);
2422
2423        // Overlap, gaps, and unbounded meets never combine.
2424        let overlapping = bounds(Bound::Included(key(b"a")), Bound::Included(key(b"c")));
2425        assert!(lower.union_adjacent(&overlapping).is_none());
2426        let gapped = bounds(Bound::Excluded(key(b"c")), Bound::Excluded(key(b"e")));
2427        assert!(lower.union_adjacent(&gapped).is_none());
2428        assert!(PartitionBounds::unbounded()
2429            .union_adjacent(&lower)
2430            .is_none());
2431
2432        // The merged chain covers exactly the union.
2433        let left = bounds(Bound::Unbounded, Bound::Excluded(key(b"c")));
2434        assert_eq!(
2435            left.union_adjacent(&upper).unwrap(),
2436            bounds(Bound::Unbounded, Bound::Excluded(key(b"e")))
2437        );
2438    }
2439
2440    // -- row key encoder -----------------------------------------------------
2441
2442    #[test]
2443    fn encoded_keys_preserve_typed_order() {
2444        let ints: Vec<Key> = [i64::MIN, -7, -1, 0, 1, 42, i64::MAX]
2445            .into_iter()
2446            .map(int_key)
2447            .collect();
2448        let mut shuffled = ints.clone();
2449        shuffled.reverse();
2450        shuffled.sort();
2451        assert_eq!(shuffled, ints);
2452
2453        // Text is bytewise, prefix-free: "a" < "a\0" < "aa" < "b".
2454        let mut texts = [
2455            text_key("aa"),
2456            text_key("a"),
2457            RowKeyEncoder::encode_key(&[KeyValue::Text("a\0".to_owned())]),
2458            text_key("b"),
2459        ];
2460        texts.sort();
2461        assert_eq!(texts[0], text_key("a"));
2462        assert_eq!(texts[2], text_key("aa"));
2463
2464        // Cross-type order is the tag order.
2465        let values = [
2466            KeyValue::Null,
2467            KeyValue::Bool(true),
2468            KeyValue::Int(1),
2469            KeyValue::TimestampMicros(1),
2470            KeyValue::Text("x".to_owned()),
2471            KeyValue::Bytes(vec![0xFF]),
2472        ];
2473        let mut cross: Vec<Key> = values
2474            .iter()
2475            .map(|value| RowKeyEncoder::encode_key(std::slice::from_ref(value)))
2476            .collect();
2477        let sorted = cross.clone();
2478        cross.reverse();
2479        cross.sort();
2480        assert_eq!(cross, sorted);
2481    }
2482
2483    #[test]
2484    fn encoded_keys_decode_back_to_their_components() {
2485        let values = vec![
2486            KeyValue::Null,
2487            KeyValue::Bool(true),
2488            KeyValue::Int(-42),
2489            KeyValue::TimestampMicros(1_700_000_000_000_000),
2490            KeyValue::Text("tenant\0-42".to_owned()),
2491            KeyValue::Bytes(vec![0x00, 0x01, 0xFF]),
2492        ];
2493        let encoded = RowKeyEncoder::encode_key(&values);
2494        assert_eq!(RowKeyEncoder::decode_components(&encoded).unwrap(), values);
2495
2496        // Malformed input fails closed.
2497        assert!(matches!(
2498            RowKeyEncoder::decode_components(&key(&[0x7F])),
2499            Err(PartitionError::MalformedKey(_))
2500        ));
2501        assert!(matches!(
2502            RowKeyEncoder::decode_components(&key(&[TAG_TEXT, b'a'])),
2503            Err(PartitionError::MalformedKey(_))
2504        ));
2505    }
2506
2507    // -- partitioning: validation and extraction ------------------------------
2508
2509    #[test]
2510    fn partitioning_validation_fails_closed_on_bad_declarations() {
2511        let no_columns = Partitioning::Hash {
2512            columns: vec![],
2513            buckets: 16,
2514        };
2515        assert!(matches!(
2516            no_columns.validate(),
2517            Err(PartitionError::InvalidPartitioning(_))
2518        ));
2519        let no_buckets = Partitioning::Hash {
2520            columns: vec![ColumnId(1)],
2521            buckets: 0,
2522        };
2523        assert!(no_buckets.validate().is_err());
2524        let unsorted = Partitioning::Range {
2525            columns: vec![ColumnId(1)],
2526            splits: vec![int_key(20), int_key(10)],
2527        };
2528        assert!(unsorted.validate().is_err());
2529        let no_tenant_buckets = Partitioning::Tenant {
2530            tenant_column: ColumnId(1),
2531            buckets_per_tenant: 0,
2532        };
2533        assert!(no_tenant_buckets.validate().is_err());
2534        assert!(TimeInterval::micros(0).is_err());
2535        assert!(TimeInterval::days(1).is_ok());
2536    }
2537
2538    #[test]
2539    fn partition_key_extracts_declared_columns_in_declared_order() {
2540        let partitioning = Partitioning::Hash {
2541            columns: vec![ColumnId(5), ColumnId(2)],
2542            buckets: 16,
2543        };
2544        let mut values = BTreeMap::new();
2545        values.insert(ColumnId(2), KeyValue::Int(2));
2546        values.insert(ColumnId(5), KeyValue::Text("tenant".to_owned()));
2547        values.insert(ColumnId(9), KeyValue::Bool(true)); // not a partition column
2548        let key = partitioning.partition_key(&values).unwrap();
2549        // Declared order (5, 2), not column-id order (2, 5).
2550        assert_eq!(
2551            RowKeyEncoder::decode_components(&key).unwrap(),
2552            vec![KeyValue::Text("tenant".to_owned()), KeyValue::Int(2)]
2553        );
2554
2555        // A missing declared column fails closed.
2556        let mut incomplete = BTreeMap::new();
2557        incomplete.insert(ColumnId(2), KeyValue::Int(2));
2558        assert_eq!(
2559            partitioning.partition_key(&incomplete).unwrap_err(),
2560            PartitionError::MissingPartitionColumn {
2561                column: ColumnId(5)
2562            }
2563        );
2564    }
2565
2566    // -- partitioning: route() --------------------------------------------------
2567
2568    #[test]
2569    fn hash_route_is_deterministic_and_bounded() {
2570        let partitioning = Partitioning::Hash {
2571            columns: vec![ColumnId(1)],
2572            buckets: 16,
2573        };
2574        let mut values = BTreeMap::new();
2575        values.insert(ColumnId(1), KeyValue::Int(42));
2576        let key = partitioning.partition_key(&values).unwrap();
2577        let slot = partitioning.route(&key).unwrap();
2578        assert_eq!(slot, partitioning.route(&key).unwrap());
2579        assert!(slot < 16);
2580
2581        // Distinct keys spread over distinct buckets.
2582        let buckets: std::collections::BTreeSet<u64> = (0..100)
2583            .map(|i| {
2584                let key = RowKeyEncoder::encode_key(&[KeyValue::Int(i)]);
2585                partitioning.route(&key).unwrap()
2586            })
2587            .collect();
2588        assert!(buckets.len() > 8, "poor spread: {buckets:?}");
2589    }
2590
2591    #[test]
2592    fn range_route_indexes_the_split_points() {
2593        let partitioning = Partitioning::Range {
2594            columns: vec![ColumnId(1)],
2595            splits: vec![int_key(10), int_key(20)],
2596        };
2597        let route = |value: i64| partitioning.route(&int_key(value)).unwrap();
2598        assert_eq!(route(i64::MIN), 0);
2599        assert_eq!(route(9), 0);
2600        assert_eq!(route(10), 1); // a split starts its own partition
2601        assert_eq!(route(19), 1);
2602        assert_eq!(route(20), 2);
2603        assert_eq!(route(i64::MAX), 2);
2604
2605        // range_bounds reconstructs the covering partition chain.
2606        let zero = partitioning.range_bounds(0).unwrap();
2607        let one = partitioning.range_bounds(1).unwrap();
2608        let two = partitioning.range_bounds(2).unwrap();
2609        assert!(partitioning.range_bounds(3).is_none());
2610        assert!(zero.is_adjacent_to(&one));
2611        assert!(one.is_adjacent_to(&two));
2612        assert!(!zero.is_adjacent_to(&two));
2613        assert!(zero.contains(&int_key(9)));
2614        assert!(!zero.contains(&int_key(10)));
2615        assert!(two.contains(&int_key(20)));
2616        assert!(matches!(zero.low, Bound::Unbounded));
2617        assert!(matches!(two.high, Bound::Unbounded));
2618    }
2619
2620    #[test]
2621    fn tenant_route_is_per_tenant_deterministic_and_bounded() {
2622        let partitioning = Partitioning::Tenant {
2623            tenant_column: ColumnId(1),
2624            buckets_per_tenant: 8,
2625        };
2626        let key_for = |tenant: &str| {
2627            let mut values = BTreeMap::new();
2628            values.insert(ColumnId(1), KeyValue::Text(tenant.to_owned()));
2629            partitioning.partition_key(&values).unwrap()
2630        };
2631        let acme = key_for("acme");
2632        let slot = partitioning.route(&acme).unwrap();
2633        assert_eq!(slot, partitioning.route(&acme).unwrap());
2634        assert!(slot < 8);
2635        // The slot is per tenant; the routed key disambiguates (spec 12.2).
2636        let routed = partitioning.routed_key(&acme).unwrap();
2637        let mut expected = acme.as_bytes().to_vec();
2638        expected.extend_from_slice(&slot.to_be_bytes());
2639        assert_eq!(routed, key(&expected));
2640
2641        let initech = key_for("initech");
2642        assert_eq!(
2643            partitioning.route(&initech).unwrap(),
2644            partitioning.route(&key_for("initech")).unwrap()
2645        );
2646    }
2647
2648    #[test]
2649    fn time_range_route_buckets_by_interval() {
2650        let interval = TimeInterval::hours(1).unwrap();
2651        let partitioning = Partitioning::TimeRange {
2652            timestamp_column: ColumnId(3),
2653            interval,
2654        };
2655        let key_for = |micros: i64| {
2656            let mut values = BTreeMap::new();
2657            values.insert(ColumnId(3), KeyValue::TimestampMicros(micros));
2658            partitioning.partition_key(&values).unwrap()
2659        };
2660        let width = i64::try_from(interval.as_micros()).unwrap();
2661        assert_eq!(partitioning.route(&key_for(0)).unwrap(), 0);
2662        assert_eq!(partitioning.route(&key_for(width - 1)).unwrap(), 0);
2663        assert_eq!(partitioning.route(&key_for(width)).unwrap(), 1);
2664        assert_eq!(partitioning.route(&key_for(5 * width / 2)).unwrap(), 2);
2665
2666        // Pre-epoch timestamps fail closed.
2667        assert!(matches!(
2668            partitioning.route(&key_for(-1)),
2669            Err(PartitionError::NegativeSlot { micros: -1 })
2670        ));
2671
2672        // The timestamp column must carry a timestamp.
2673        let mut wrong_type = BTreeMap::new();
2674        wrong_type.insert(ColumnId(3), KeyValue::Int(0));
2675        assert_eq!(
2676            partitioning.partition_key(&wrong_type).unwrap_err(),
2677            PartitionError::PartitionColumnType {
2678                column: ColumnId(3),
2679                expected: "timestamp-micros",
2680                found: "int",
2681            }
2682        );
2683    }
2684
2685    // -- table partitioning records and colocation -----------------------------
2686
2687    #[test]
2688    fn table_partitioning_record_round_trips_with_colocation() {
2689        let record = TablePartitioningRecord {
2690            table_id: TableId::new(4),
2691            partitioning: Partitioning::Range {
2692                columns: vec![ColumnId(1)],
2693                splits: vec![int_key(100)],
2694            },
2695            origin: PartitioningOrigin::Declared,
2696            colocated_with: Some(ColocatedWith(TableId::new(2))),
2697        };
2698        record.validate().unwrap();
2699        let json = serde_json::to_vec(&record).unwrap();
2700        let back: TablePartitioningRecord = serde_json::from_slice(&json).unwrap();
2701        assert_eq!(back, record);
2702    }
2703
2704    #[test]
2705    fn automatic_defaults_are_visible_in_schema_metadata() {
2706        let record =
2707            TablePartitioningRecord::automatic_default(TableId::new(4), vec![ColumnId(1)], 64);
2708        assert_eq!(record.origin, PartitioningOrigin::AutomaticDefault);
2709        assert_eq!(record.partitioning.partition_columns(), vec![ColumnId(1)]);
2710        assert_eq!(record.colocated_with, None);
2711        record.validate().unwrap();
2712
2713        // The default is serialized, not implicit.
2714        let json = serde_json::to_value(&record).unwrap();
2715        assert_eq!(json["origin"], serde_json::json!("AutomaticDefault"));
2716    }
2717
2718    #[test]
2719    fn record_validation_rejects_zero_table_and_self_colocation() {
2720        let mut zero =
2721            TablePartitioningRecord::automatic_default(TableId::ZERO, vec![ColumnId(1)], 4);
2722        assert!(matches!(
2723            zero.validate(),
2724            Err(TabletError::InvalidDescriptor(_))
2725        ));
2726        zero.table_id = TableId::new(4);
2727        zero.colocated_with = Some(ColocatedWith(TableId::new(4)));
2728        assert!(matches!(
2729            zero.validate(),
2730            Err(TabletError::InvalidDescriptor(_))
2731        ));
2732    }
2733
2734    // -- tablet routing (spec 12.4) ---------------------------------------------
2735
2736    fn routed_hash_tablet(byte: u8, state: TabletState, slots: (u64, u64)) -> TabletDescriptor {
2737        let mut tablet = descriptor(state);
2738        tablet.tablet_id = tablet_id(byte);
2739        tablet.table_id = TableId::new(5);
2740        tablet.partition = hash_slot_bounds(slots.0, slots.1);
2741        tablet
2742    }
2743
2744    #[test]
2745    fn point_reads_and_writes_route_directly() {
2746        let partitioning = Partitioning::Hash {
2747            columns: vec![ColumnId(1)],
2748            buckets: 4,
2749        };
2750        let tablets = vec![
2751            routed_hash_tablet(1, TabletState::Active, (0, 2)),
2752            routed_hash_tablet(2, TabletState::Active, (2, 4)),
2753        ];
2754        for value in 0..50 {
2755            let key = RowKeyEncoder::encode_key(&[KeyValue::Int(value)]);
2756            let routed = partitioning.routed_key(&key).unwrap();
2757            let slot = partitioning.route(&key).unwrap();
2758            let expected = tablet_id(if slot < 2 { 1 } else { 2 });
2759            assert_eq!(
2760                find_tablet_for_key(&tablets, TableId::new(5), &routed).map(|t| t.tablet_id),
2761                Some(expected),
2762                "value {value} routed wrong"
2763            );
2764            // Other tables and unroutable states never match.
2765            assert!(find_tablet_for_key(&tablets, TableId::new(6), &routed).is_none());
2766        }
2767        // A Creating child tablet is never exposed before catch-up.
2768        let children = vec![
2769            routed_hash_tablet(1, TabletState::Active, (0, 2)),
2770            routed_hash_tablet(3, TabletState::Creating, (0, 1)),
2771        ];
2772        let routed = partitioning
2773            .routed_key(&RowKeyEncoder::encode_key(&[KeyValue::Int(1)]))
2774            .unwrap();
2775        let slot = partitioning
2776            .route(&RowKeyEncoder::encode_key(&[KeyValue::Int(1)]))
2777            .unwrap();
2778        if slot < 2 {
2779            assert_eq!(
2780                find_tablet_for_key(&children, TableId::new(5), &routed).map(|t| t.tablet_id),
2781                Some(tablet_id(1))
2782            );
2783        }
2784    }
2785
2786    #[test]
2787    fn range_queries_route_to_all_overlapping_tablets_in_order() {
2788        let t = |byte: u8, low: Bound<Key>, high: Bound<Key>| {
2789            let mut tablet = descriptor(TabletState::Active);
2790            tablet.tablet_id = tablet_id(byte);
2791            tablet.table_id = TableId::new(5);
2792            tablet.partition = bounds(low, high);
2793            tablet
2794        };
2795        let tablets = vec![
2796            t(3, Bound::Included(text_key("m")), Bound::Unbounded),
2797            t(1, Bound::Unbounded, Bound::Excluded(text_key("e"))),
2798            t(
2799                2,
2800                Bound::Included(text_key("e")),
2801                Bound::Excluded(text_key("m")),
2802            ),
2803            {
2804                let mut retired = t(4, Bound::Unbounded, Bound::Excluded(text_key("e")));
2805                retired.state = TabletState::Retired;
2806                retired
2807            },
2808        ];
2809        // Point-narrow range inside the middle tablet.
2810        let narrow = bounds(
2811            Bound::Included(text_key("f")),
2812            Bound::Excluded(text_key("g")),
2813        );
2814        assert_eq!(
2815            tablets_overlapping(&tablets, TableId::new(5), &narrow)
2816                .iter()
2817                .map(|tablet| tablet.tablet_id)
2818                .collect::<Vec<_>>(),
2819            vec![tablet_id(2)]
2820        );
2821        // Full scan: all routable tablets, sorted by low endpoint, retired excluded.
2822        assert_eq!(
2823            tablets_overlapping(&tablets, TableId::new(5), &PartitionBounds::unbounded())
2824                .iter()
2825                .map(|tablet| tablet.tablet_id)
2826                .collect::<Vec<_>>(),
2827            vec![tablet_id(1), tablet_id(2), tablet_id(3)]
2828        );
2829    }
2830
2831    #[test]
2832    fn stale_generations_classify_per_tablet_state() {
2833        let tablet = descriptor(TabletState::Active);
2834        assert!(check_generation(&tablet, 7).is_ok());
2835
2836        let mut splitting = tablet.clone();
2837        splitting.state = TabletState::Splitting;
2838        let error = check_generation(&splitting, 6).unwrap_err();
2839        assert_eq!(
2840            error,
2841            RoutingError::TabletSplit {
2842                tablet_id: tablet_id(9),
2843                used_generation: 6,
2844                current_generation: 7,
2845            }
2846        );
2847        assert_eq!(error.category(), ErrorCategory::TabletSplitting);
2848
2849        let mut retiring = tablet.clone();
2850        retiring.state = TabletState::Retiring;
2851        let error = check_generation(&retiring, 6).unwrap_err();
2852        assert!(matches!(error, RoutingError::TabletMoved { .. }));
2853        assert_eq!(error.category(), ErrorCategory::TabletMoved);
2854
2855        // Any other mismatch — including a request newer than the replica —
2856        // is plain stale metadata.
2857        let error = check_generation(&tablet, 6).unwrap_err();
2858        assert!(matches!(error, RoutingError::StaleMetadata { .. }));
2859        assert_eq!(error.category(), ErrorCategory::StaleMetadata);
2860        let error = check_generation(&tablet, 8).unwrap_err();
2861        assert!(matches!(error, RoutingError::StaleMetadata { .. }));
2862
2863        // Review N2: newer generation while Splitting is StaleMetadata
2864        // (replica behind), not TabletSplit.
2865        let error = check_generation(&splitting, 9).unwrap_err();
2866        assert!(
2867            matches!(error, RoutingError::StaleMetadata { .. }),
2868            "got {error:?}"
2869        );
2870    }
2871
2872    #[test]
2873    fn list_tablets_on_disk_reads_real_metadata() {
2874        let dir = tempfile::tempdir().unwrap();
2875        let desc = descriptor(TabletState::Active);
2876        let layout = TabletLayout::new(dir.path(), desc.tablet_id, desc.raft_group_id);
2877        layout.create(&desc).unwrap();
2878        let (listed, issues) = list_tablets_on_disk(dir.path()).unwrap();
2879        assert!(issues.is_empty());
2880        assert_eq!(listed.len(), 1);
2881        assert_eq!(listed[0].tablet_id, desc.tablet_id);
2882        assert_eq!(listed[0].replicas.len(), desc.replicas.len());
2883    }
2884
2885    // -- tablet layout (spec 12.3) ---------------------------------------------
2886
2887    fn layout_fixture(root: &Path) -> (TabletLayout, TabletDescriptor) {
2888        let descriptor = descriptor(TabletState::Active);
2889        let layout = TabletLayout::new(root, descriptor.tablet_id, descriptor.raft_group_id);
2890        (layout, descriptor)
2891    }
2892
2893    #[test]
2894    fn layout_create_makes_the_spec_directory_tree_and_metadata() {
2895        let dir = tempfile::tempdir().unwrap();
2896        let (layout, descriptor) = layout_fixture(dir.path());
2897        layout.create(&descriptor).unwrap();
2898
2899        for path in [
2900            layout.state_dir(),
2901            layout.runs_dir(),
2902            layout.indexes_dir(),
2903            layout.temp_dir(),
2904            layout.raft_dir(),
2905            layout.snapshots_dir(),
2906        ] {
2907            assert!(path.is_dir(), "missing {}", path.display());
2908        }
2909        assert!(layout.metadata_path().is_file());
2910        // The spec tree: node-data/tablets/<tablet-id>, groups/<group-id>.
2911        assert_eq!(
2912            layout.tablet_dir(),
2913            dir.path()
2914                .join(TABLETS_DIR)
2915                .join(descriptor.tablet_id.to_hex())
2916        );
2917        assert_eq!(
2918            layout.group_dir(),
2919            dir.path()
2920                .join(GROUPS_DIR)
2921                .join(descriptor.raft_group_id.to_hex())
2922        );
2923
2924        // The persisted metadata verifies and round-trips.
2925        assert_eq!(layout.validate().unwrap(), descriptor);
2926        assert_eq!(layout.load_metadata().unwrap(), descriptor);
2927    }
2928
2929    #[test]
2930    fn layout_create_is_idempotent_but_never_repurposes_a_directory() {
2931        let dir = tempfile::tempdir().unwrap();
2932        let (layout, descriptor) = layout_fixture(dir.path());
2933        layout.create(&descriptor).unwrap();
2934        // Identical re-create succeeds (crash-recovered create).
2935        layout.create(&descriptor).unwrap();
2936        // Different metadata fails closed.
2937        let mut other = descriptor.clone();
2938        other.generation = 8;
2939        assert!(matches!(
2940            layout.create(&other),
2941            Err(TabletError::MetadataConflict(_))
2942        ));
2943    }
2944
2945    #[test]
2946    fn layout_store_metadata_advances_the_descriptor_atomically() {
2947        let dir = tempfile::tempdir().unwrap();
2948        let (layout, descriptor) = layout_fixture(dir.path());
2949        layout.create(&descriptor).unwrap();
2950        let mut advanced = descriptor.clone();
2951        advanced.generation = 8;
2952        advanced.state = TabletState::Splitting;
2953        layout.store_metadata(&advanced).unwrap();
2954        assert_eq!(layout.load_metadata().unwrap(), advanced);
2955
2956        // A descriptor for another tablet/group is refused.
2957        let mut foreign = advanced.clone();
2958        foreign.tablet_id = tablet_id(8);
2959        assert!(matches!(
2960            layout.store_metadata(&foreign),
2961            Err(TabletError::TabletMismatch { .. })
2962        ));
2963    }
2964
2965    #[test]
2966    fn layout_open_fails_closed_on_missing_or_corrupt_metadata() {
2967        let dir = tempfile::tempdir().unwrap();
2968        let (layout, descriptor) = layout_fixture(dir.path());
2969
2970        // Nothing created: metadata is missing.
2971        assert!(matches!(
2972            layout.load_metadata(),
2973            Err(TabletError::MissingMetadata(_))
2974        ));
2975        layout.create(&descriptor).unwrap();
2976
2977        // Garbage bytes.
2978        std::fs::write(layout.metadata_path(), b"{ not json").unwrap();
2979        assert!(matches!(
2980            layout.load_metadata(),
2981            Err(TabletError::Metadata(ClusterError::CorruptMetadata { .. }))
2982        ));
2983
2984        // Unknown format version.
2985        layout.store_metadata(&descriptor).unwrap();
2986        let mut value: serde_json::Value =
2987            serde_json::from_slice(&std::fs::read(layout.metadata_path()).unwrap()).unwrap();
2988        value["format_version"] = serde_json::json!(99);
2989        std::fs::write(layout.metadata_path(), serde_json::to_vec(&value).unwrap()).unwrap();
2990        assert!(matches!(
2991            layout.load_metadata(),
2992            Err(TabletError::Metadata(
2993                ClusterError::UnsupportedFormatVersion { found: 99, .. }
2994            ))
2995        ));
2996    }
2997
2998    #[test]
2999    fn layout_open_fails_closed_on_checksum_or_identity_drift() {
3000        let dir = tempfile::tempdir().unwrap();
3001        let (layout, descriptor) = layout_fixture(dir.path());
3002        layout.create(&descriptor).unwrap();
3003
3004        // Tamper with the payload: the checksum no longer matches.
3005        let path = layout.metadata_path();
3006        let mut value: serde_json::Value =
3007            serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
3008        value["tablet"]["generation"] = serde_json::json!(8);
3009        std::fs::write(&path, serde_json::to_vec(&value).unwrap()).unwrap();
3010        assert!(matches!(
3011            layout.load_metadata(),
3012            Err(TabletError::Metadata(ClusterError::CorruptMetadata { .. }))
3013        ));
3014
3015        // A foreign tablet's (self-consistent) metadata in this directory is
3016        // an identity error, never a silent open.
3017        let other_dir = tempfile::tempdir().unwrap();
3018        let mut foreign = descriptor.clone();
3019        foreign.tablet_id = tablet_id(8);
3020        let foreign_layout =
3021            TabletLayout::new(other_dir.path(), foreign.tablet_id, foreign.raft_group_id);
3022        foreign_layout.create(&foreign).unwrap();
3023        std::fs::write(
3024            &path,
3025            std::fs::read(foreign_layout.metadata_path()).unwrap(),
3026        )
3027        .unwrap();
3028        assert!(matches!(
3029            layout.load_metadata(),
3030            Err(TabletError::TabletMismatch { .. })
3031        ));
3032    }
3033
3034    #[test]
3035    fn layout_validate_requires_the_full_directory_tree() {
3036        let dir = tempfile::tempdir().unwrap();
3037        let (layout, descriptor) = layout_fixture(dir.path());
3038        layout.create(&descriptor).unwrap();
3039        layout.validate().unwrap();
3040
3041        std::fs::remove_dir(layout.runs_dir()).unwrap();
3042        assert!(matches!(
3043            layout.validate(),
3044            Err(TabletError::Metadata(ClusterError::CorruptMetadata { .. }))
3045        ));
3046    }
3047
3048    #[test]
3049    fn teardown_removes_the_replica_idempotently_but_never_foreign_state() {
3050        let dir = tempfile::tempdir().unwrap();
3051        let (layout, descriptor) = layout_fixture(dir.path());
3052        layout.create(&descriptor).unwrap();
3053        assert!(layout.tablet_dir().is_dir() && layout.group_dir().is_dir());
3054
3055        layout.teardown().unwrap();
3056        assert!(!layout.tablet_dir().exists());
3057        assert!(!layout.group_dir().exists());
3058        // Tearing down an absent replica is fine (crash-resumed removal).
3059        layout.teardown().unwrap();
3060
3061        // A directory holding foreign metadata is never deleted.
3062        let other_dir = tempfile::tempdir().unwrap();
3063        let mut foreign = descriptor.clone();
3064        foreign.tablet_id = tablet_id(8);
3065        let foreign_layout =
3066            TabletLayout::new(other_dir.path(), foreign.tablet_id, foreign.raft_group_id);
3067        foreign_layout.create(&foreign).unwrap();
3068        layout.create(&descriptor).unwrap();
3069        std::fs::write(
3070            layout.metadata_path(),
3071            std::fs::read(foreign_layout.metadata_path()).unwrap(),
3072        )
3073        .unwrap();
3074        assert!(matches!(
3075            layout.teardown(),
3076            Err(TabletError::TabletMismatch { .. })
3077        ));
3078        assert!(layout.tablet_dir().is_dir(), "foreign state was deleted");
3079    }
3080
3081    // -- process-local ownership (spec 4.1 / 12.3) -------------------------------
3082
3083    #[test]
3084    fn one_tablet_storage_core_is_owned_by_one_process() {
3085        let dir = tempfile::tempdir().unwrap();
3086        let (layout, descriptor) = layout_fixture(dir.path());
3087        layout.create(&descriptor).unwrap();
3088
3089        let registry = TabletOwnershipRegistry::new();
3090        let guard = registry.try_reserve(&layout).unwrap();
3091        assert_eq!(guard.tablet_id(), descriptor.tablet_id);
3092        assert_eq!(registry.len(), 1);
3093
3094        // A second open of the same tablet fails closed — even by the owner.
3095        let error = registry.try_reserve(&layout).unwrap_err();
3096        assert!(matches!(
3097            error,
3098            TabletError::AlreadyOwned { tablet, .. } if tablet == descriptor.tablet_id
3099        ));
3100
3101        // A different tablet directory reserves independently.
3102        let mut distinct = descriptor.clone();
3103        distinct.tablet_id = tablet_id(10);
3104        let distinct_layout =
3105            TabletLayout::new(dir.path(), distinct.tablet_id, distinct.raft_group_id);
3106        distinct_layout.create(&distinct).unwrap();
3107        let _distinct_guard = registry.try_reserve(&distinct_layout).unwrap();
3108        assert_eq!(registry.len(), 2);
3109
3110        // Dropping the guard releases the reservation.
3111        drop(guard);
3112        assert_eq!(registry.len(), 1);
3113        let _reacquired = registry.try_reserve(&layout).unwrap();
3114        assert_eq!(registry.len(), 2);
3115    }
3116
3117    #[test]
3118    fn ownership_reservation_keys_on_the_canonical_path() {
3119        let dir = tempfile::tempdir().unwrap();
3120        let (layout, descriptor) = layout_fixture(dir.path());
3121        layout.create(&descriptor).unwrap();
3122
3123        let registry = TabletOwnershipRegistry::new();
3124        let _guard = registry.try_reserve(&layout).unwrap();
3125        // The same tablet reached through an aliased path (`/./`) is the
3126        // same reservation.
3127        let aliased_root = dir.path().join(".");
3128        let aliased =
3129            TabletLayout::new(aliased_root, descriptor.tablet_id, descriptor.raft_group_id);
3130        assert!(matches!(
3131            registry.try_reserve(&aliased),
3132            Err(TabletError::AlreadyOwned { .. })
3133        ));
3134    }
3135}