nodedb_types/

calvin.rs

// SPDX-License-Identifier: Apache-2.0

//! Primitive Calvin scheduling types shared between `nodedb-physical`
//! (the physical-plan IR layer) and `nodedb-cluster` (the distributed
//! Calvin sequencer / scheduler).
//!
//! Provides [`SortedVec`], [`EngineKeySet`], and [`PassiveReadKey`] —
//! the building blocks of Calvin read/write sets. `DependentReadSpec`
//! and other scheduler-internal aggregates stay in `nodedb-cluster`.

use serde::{Deserialize, Serialize};

use crate::{KeyRepr, Lsn};

/// A newtype over `Vec<T>` that guarantees sorted, deduplicated contents.
///
/// Constructed via [`SortedVec::new`], which sorts and deduplicates at
/// construction time. This property is load-bearing for byte-determinism:
/// two `SortedVec`s built from the same logical set (in any insertion order)
/// produce identical serialized bytes.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SortedVec<T>(Vec<T>);

impl<T: zerompk::ToMessagePack> zerompk::ToMessagePack for SortedVec<T> {
    fn write<W: zerompk::Write>(&self, writer: &mut W) -> zerompk::Result<()> {
        self.0.write(writer)
    }
}

impl<'de, T> zerompk::FromMessagePack<'de> for SortedVec<T>
where
    T: zerompk::FromMessagePack<'de> + Ord + Clone,
{
    fn read<R: zerompk::Read<'de>>(reader: &mut R) -> zerompk::Result<Self> {
        let v = Vec::<T>::read(reader)?;
        Ok(Self::new(v))
    }
}

impl<T: Ord + Clone> SortedVec<T> {
    /// Build from any slice. Sorts and deduplicates in place.
    pub fn new(mut items: Vec<T>) -> Self {
        items.sort();
        items.dedup();
        Self(items)
    }

    pub fn as_slice(&self) -> &[T] {
        &self.0
    }

    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    pub fn len(&self) -> usize {
        self.0.len()
    }

    pub fn iter(&self) -> std::slice::Iter<'_, T> {
        self.0.iter()
    }
}

impl<T: Ord + Clone> From<Vec<T>> for SortedVec<T> {
    fn from(v: Vec<T>) -> Self {
        Self::new(v)
    }
}

/// A typed key set for one engine within a read or write set.
///
/// Keys are normalized to surrogates (or byte keys for KV) at admission, so
/// all engine-specific naming is resolved upstream of the sequencer.
#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    zerompk::ToMessagePack,
    zerompk::FromMessagePack,
)]
pub enum EngineKeySet {
    /// Document engine (schemaless or strict): identified by surrogate.
    Document {
        collection: String,
        surrogates: SortedVec<u32>,
    },
    /// Vector engine: identified by surrogate.
    Vector {
        collection: String,
        surrogates: SortedVec<u32>,
    },
    /// Key-Value engine: identified by raw byte keys.
    Kv {
        collection: String,
        keys: SortedVec<Vec<u8>>,
    },
    /// Graph edge engine: identified by (src_surrogate, dst_surrogate) pairs.
    ///
    /// `edges` carries the surrogate-pair IDENTITY used for Calvin locking and
    /// conflict detection. `home_vshards` carries the from_key ROUTING homes —
    /// the set of `VShardId::from_key(src_str)` / `from_key(dst_str)` u32 ids
    /// for every edge in this set. Routing (which vShards participate) is driven
    /// by `home_vshards`, NOT by the collection name, because a graph edge is
    /// dual-homed across its two endpoint key-hashed vShards.
    Edge {
        collection: String,
        edges: SortedVec<(u32, u32)>,
        home_vshards: SortedVec<u32>,
    },
}

impl EngineKeySet {
    /// O(1) estimate of the serialized byte size of this key set.
    ///
    /// Used by the dependent-read cap check at sequencer admission to bound
    /// the total bytes that would be Raft-replicated in a `CalvinReadResult`
    /// entry.  This is an estimate, not an exact count; do NOT use it as a
    /// correctness check — only as a pre-flight guard.
    pub fn serialized_size_hint(&self) -> usize {
        match self {
            // u32 surrogates: 4 bytes each.
            Self::Document { surrogates, .. } | Self::Vector { surrogates, .. } => {
                surrogates.len() * 4
            }
            // KV keys: sum of key byte lengths.
            Self::Kv { keys, .. } => keys.iter().map(|k| k.len()).sum(),
            // Edge: two u32 per edge = 8 bytes each.
            Self::Edge { edges, .. } => edges.len() * 8,
        }
    }

    /// The collection this key set belongs to.
    pub fn collection(&self) -> &str {
        match self {
            Self::Document { collection, .. }
            | Self::Vector { collection, .. }
            | Self::Kv { collection, .. }
            | Self::Edge { collection, .. } => collection,
        }
    }

    /// Returns `true` if this key set contains no keys.
    pub fn is_empty(&self) -> bool {
        match self {
            Self::Document { surrogates, .. } => surrogates.is_empty(),
            Self::Vector { surrogates, .. } => surrogates.is_empty(),
            Self::Kv { keys, .. } => keys.is_empty(),
            Self::Edge { edges, .. } => edges.is_empty(),
        }
    }
}

/// A single key that a passive participant must read and broadcast.
///
/// Wraps an [`EngineKeySet`]; per the dependent-read protocol each
/// `PassiveReadKey` contains a single-element (or small) key set.  The
/// sequencer does not enforce single-element sets; the scheduler enforces the
/// total byte budget via `DependentReadSpec::total_bytes()` (which lives in
/// `nodedb-cluster`).
#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    zerompk::ToMessagePack,
    zerompk::FromMessagePack,
)]
pub struct PassiveReadKey {
    /// The engine key set to read on the passive vshard.
    pub engine_key: EngineKeySet,
}

/// Which peer engine served a read. Mirrors the top-level physical-plan
/// engine variants one-to-one so the classifier is total and a new engine
/// forces a decision at compile time.
///
/// Encoded as a plain integer discriminant (`#[msgpack(c_enum)]`): stable,
/// compact, and comparable across the wire. The discriminant assignment is
/// load-bearing for on-wire stability — do NOT reorder variants.
#[derive(
    Debug,
    Clone,
    Copy,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    zerompk::ToMessagePack,
    zerompk::FromMessagePack,
)]
#[msgpack(c_enum)]
pub enum EngineTag {
    Vector,
    Graph,
    Document,
    Kv,
    Text,
    Columnar,
    Timeseries,
    Spatial,
    Crdt,
    Query,
    Meta,
    Array,
    ClusterArray,
}

/// The identity a read observed within a collection, carried on the
/// replicated Calvin `TxClass`.
///
/// `Point` carries the exact row identity ([`KeyRepr`]) for a keyed lookup
/// (per-key optimistic-concurrency validation). `Predicate` is the coarse,
/// collection-scoped observation for scans / searches / aggregates — safe
/// against phantoms, never under-approximating. `IndexEq` / `IndexRange` carry
/// the indexed dimension of a secondary-index equality / range read (canonical
/// stringified index value, identical to the index-key segment) for narrower
/// per-value validation.
///
/// New variants are APPENDED only — the on-wire encoding is positional, so
/// reordering would break cross-version decode. Do NOT reorder variants.
#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    zerompk::ToMessagePack,
    zerompk::FromMessagePack,
)]
pub enum ReadKeyIdent {
    /// A single-row keyed observation.
    Point(KeyRepr),
    /// A collection-scoped predicate observation.
    Predicate,
    /// A secondary-index equality observation on one indexed field.
    IndexEq { field: String, value: String },
    /// A secondary-index range observation on one indexed field. `lo`/`hi` are
    /// optional so a one-sided native range is representable; both `None` is
    /// never emitted.
    IndexRange {
        field: String,
        lo: Option<String>,
        hi: Option<String>,
    },
}

/// One LSN-versioned, predicate-aware read observed by a transaction, carried
/// on the replicated Calvin `TxClass` so participants can validate it at the
/// commit serialization point.
///
/// `read_lsn` is the responding shard's write-LSN watermark at read time. The
/// enclosing `TxClass` scopes the tenant; per-database scoping is carried by
/// the transaction as a whole.
#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    zerompk::ToMessagePack,
    zerompk::FromMessagePack,
)]
pub struct VersionedReadEntry {
    /// Which engine served the read.
    pub engine: EngineTag,
    /// The collection the read observed.
    pub collection: String,
    /// Point-key or collection-scoped-predicate identity of the observation.
    pub key: ReadKeyIdent,
    /// The responding shard's write-LSN watermark at read time.
    pub read_lsn: Lsn,
}

/// The LSN-versioned read-set of a Calvin transaction.
///
/// Empty for pure-write transactions and for autocommit statements (which
/// accumulate no session read-set). Populated at commit time from the neutral
/// session read-set, and validated per-participant against the local
/// write-version index at Calvin stage time (the commit vote on
/// `read_set_valid`).
#[derive(
    Debug,
    Clone,
    Default,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    zerompk::ToMessagePack,
    zerompk::FromMessagePack,
)]
pub struct VersionedReadSet(pub Vec<VersionedReadEntry>);

impl VersionedReadSet {
    /// Build from a vector of entries.
    pub fn new(entries: Vec<VersionedReadEntry>) -> Self {
        Self(entries)
    }

    /// Returns `true` if no reads were recorded.
    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    /// Number of recorded read entries.
    pub fn len(&self) -> usize {
        self.0.len()
    }

    /// Iterate the recorded read entries.
    pub fn iter(&self) -> std::slice::Iter<'_, VersionedReadEntry> {
        self.0.iter()
    }

    /// Borrow the entries as a slice.
    pub fn as_slice(&self) -> &[VersionedReadEntry] {
        &self.0
    }
}