orbit_core/ring/

mod.rs

//! Ring buffers — orbit-core's runtime substrate.
//!
//! > *"orbit runtime yani ring"* — the place where the fleet's
//! > shared state actually lives at the lowest level. Higher-level
//! > shapes (cache mutations, metrics snapshots, event streams, etc.)
//! > reduce to *one or more rings*.
//!
//! ## Shape
//!
//! One [`Ring`] per [`OrbitTyped`] kind. A ring is one or more fixed-size
//! circular lanes of [`Frame`]s. Shared rings use one lock-free multi-writer
//! claim sequence. Shared-ordered rings serialize writers and expose only
//! committed counters. Per-node rings give every fleet member a disjoint lane
//! whose head advances only after a frame is committed. When a lane head
//! exceeds its capacity, the oldest slot in that lane is overwritten.
//!
//! The frame layout mirrors the `nwd1` seed (see VISION §13):
//!
//! ```text
//! ┌──────────┬──────┬──────┬─────────────┐
//! │ id (8)   │ kind │ ver  │ payload (N) │
//! │ NetId64  │  u8  │ u64  │   bytes     │
//! └──────────┴──────┴──────┴─────────────┘
//! ```
//!
//! Two `kind` bytes coexist on the wire and they mean different
//! things (intentional, two-axis encoding):
//!
//! - `frame.id.kind()` — *which Rust type* (the data shape).
//! - `frame.kind`      — *which message class* (state / event /
//!   command / ack / invalidate / …). V0 leaves this open at `0`;
//!   concrete classes appear when subscriber semantics arrive.
//!
//! ## V0 backing
//!
//! `RwLock<Option<Frame>>` per slot — simple, correct, slow. The SHM
//! backing uses per-slot sequence numbers to prevent torn reads. Per-node
//! lanes serialize concurrent publishers inside one process, commit the
//! slot, then release-store their lane head.
//!
//! ## Who mints
//!
//! Writes go through [`Ring::write`], which mints the [`NetId64`]. The
//! COUNTER part is the position inside either the shared sequence or the
//! writer node's lane. NetId64s are therefore minted
//! **server-side, by the writer process**. Browsers / external
//! clients receive ids; they do not generate them.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock, RwLock};

use bytes::Bytes;

use crate::id::NetId64;
use crate::{NodeId, OrbitTyped};

pub mod cursor;
#[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
mod readiness;
#[cfg(unix)]
pub mod shm;

#[cfg(any(
    target_os = "linux",
    target_os = "freebsd",
    target_os = "macos"
))]
pub use readiness::RingEventFd;

/// Writer ownership for one [`OrbitTyped`] ring.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[repr(u8)]
pub enum RingTopology {
    /// Every fleet member publishes into one shared multi-writer sequence.
    Shared = 0,
    /// Every fleet member owns an independent single-process writer lane.
    /// The embedder must not run two active processes with the same node id;
    /// local concurrent tasks are serialized by the ring handle.
    PerNode = 1,
    /// Every fleet member publishes into one globally ordered sequence.
    /// Writers are serialized by a process-recoverable OS lock associated
    /// with the SHM name, and the head advances only after the slot commits.
    SharedOrdered = 2,
}

/// Physical policy for one [`OrbitTyped`] ring.
///
/// The policy is part of the wire contract: every process using the
/// same `OrbitTyped::KIND` must declare the same values.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RingSpec {
    /// Number of slots retained by the ring, per lane.
    pub capacity: usize,
    /// Maximum payload bytes stored inline in each slot.
    pub payload_capacity: usize,
    /// How writers own and publish physical lanes.
    pub topology: RingTopology,
}

impl RingSpec {
    pub const fn new(
        capacity: usize,
        payload_capacity: usize,
    ) -> Self {
        Self { capacity, payload_capacity, topology: RingTopology::Shared }
    }

    /// Declare one independent writer lane per fleet node.
    pub const fn per_node(
        capacity: usize,
        payload_capacity: usize,
    ) -> Self {
        Self { capacity, payload_capacity, topology: RingTopology::PerNode }
    }

    /// Declare one crash-recoverable, globally ordered writer lane.
    pub const fn shared_ordered(
        capacity: usize,
        payload_capacity: usize,
    ) -> Self {
        Self {
            capacity,
            payload_capacity,
            topology: RingTopology::SharedOrdered,
        }
    }

    pub(crate) fn assert_valid(self) {
        assert!(self.capacity > 0, "ring capacity must be > 0");
        assert!(
            self.capacity.is_power_of_two(),
            "ring capacity must be a power of two"
        );
        assert!(
            self.payload_capacity <= u32::MAX as usize,
            "ring payload capacity must fit in u32"
        );
    }
}

struct RingLane {
    write_pos: AtomicU64,
    write_lock: Mutex<()>,
    slots: Vec<RwLock<Option<Frame>>>,
}

impl RingLane {
    fn new(capacity: usize) -> Self {
        let mut slots = Vec::with_capacity(capacity);
        for _ in 0..capacity {
            slots.push(RwLock::new(None));
        }
        Self { write_pos: AtomicU64::new(0), write_lock: Mutex::new(()), slots }
    }
}

/// One record in a ring — the on-wire shape (mirrors `nwd1::Frame`).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Frame {
    pub id: NetId64,
    pub kind: u8,
    pub ver: u64,
    pub payload: Bytes,
}

/// A fixed-capacity, fleet-wide append-only log keyed on KIND byte.
///
/// V0 is single-process; V1 is SHM-backed. The API is the same.
pub struct Ring {
    /// The KIND this ring carries — equals `T::KIND` for the
    /// `OrbitTyped` value-shape it's storing.
    kind: u8,
    /// Number of slots; constant for the ring's lifetime.
    capacity: usize,
    /// Maximum inline payload bytes for this ring lane.
    payload_capacity: usize,
    topology: RingTopology,
    /// Ring-wide semantic version allocator shared by every writer lane.
    version_counter: AtomicU64,
    pub(crate) publication: RwLock<()>,
    lanes: Vec<RingLane>,
}

impl Ring {
    /// Create the process-local ring declared by `T::RING_SPEC` for a
    /// single-member fleet.
    pub fn new<T: OrbitTyped>() -> Self {
        Self::new_for_fleet::<T>(1)
    }

    /// Create the process-local ring declared by `T::RING_SPEC` with the
    /// physical lane count required by `fleet_capacity`.
    pub fn new_for_fleet<T: OrbitTyped>(fleet_capacity: u16) -> Self {
        assert!(fleet_capacity > 0, "ring fleet capacity must be > 0");
        let spec = T::RING_SPEC;
        spec.assert_valid();
        let capacity = spec.capacity;
        let lane_count = match spec.topology {
            RingTopology::Shared | RingTopology::SharedOrdered => 1,
            RingTopology::PerNode => usize::from(fleet_capacity),
        };
        let mut lanes = Vec::with_capacity(lane_count);
        for _ in 0..lane_count {
            lanes.push(RingLane::new(capacity));
        }
        Self {
            kind: T::KIND,
            capacity,
            payload_capacity: spec.payload_capacity,
            topology: spec.topology,
            version_counter: AtomicU64::new(0),
            publication: RwLock::new(()),
            lanes,
        }
    }

    /// The KIND byte this ring carries (equals `T::KIND`).
    pub fn kind(&self) -> u8 {
        self.kind
    }

    /// Total slot count — fixed at construction.
    pub fn capacity(&self) -> usize {
        self.capacity
    }

    /// Maximum inline payload bytes for this ring lane.
    pub fn payload_capacity(&self) -> usize {
        self.payload_capacity
    }

    pub fn spec(&self) -> RingSpec {
        RingSpec {
            capacity: self.capacity,
            payload_capacity: self.payload_capacity,
            topology: self.topology,
        }
    }

    /// Head of the sole shared lane, or lane zero for a per-node ring.
    pub fn head(&self) -> u64 {
        self.lanes[0].write_pos.load(Ordering::Acquire)
    }

    /// Number of physical writer lanes in this ring.
    pub fn lane_count(&self) -> usize {
        self.lanes.len()
    }

    /// Current head for `node_id`'s logical lane.
    pub fn lane_head(
        &self,
        node_id: NodeId,
    ) -> u64 {
        self.lane(node_id).write_pos.load(Ordering::Acquire)
    }

    /// Allocate one non-zero semantic version shared by every writer lane.
    ///
    /// This counter is independent of physical ring positions. Semantic
    /// layers can use it when per-node lanes need one deterministic
    /// last-write-wins order.
    pub fn next_version(&self) -> u64 {
        self.version_counter
            .fetch_add(1, Ordering::AcqRel)
            .checked_add(1)
            .expect("ring semantic version exhausted")
    }

    /// Last semantic version allocated for this ring.
    pub fn current_version(&self) -> u64 {
        self.version_counter.load(Ordering::Acquire)
    }

    /// Append a frame. Atomically reserves the next counter, mints
    /// the [`NetId64`], and writes the frame into the corresponding
    /// slot. Returns the minted id.
    ///
    /// `frame_kind` is the message class byte (V0: pass `0`).
    /// `ver` is the version / tick at write time (V0: caller's
    /// choice).
    pub fn write(
        &self,
        node_id: NodeId,
        frame_kind: u8,
        ver: u64,
        payload: Bytes,
    ) -> NetId64 {
        assert!(
            payload.len() <= self.payload_capacity,
            "payload {} > ring payload capacity {}",
            payload.len(),
            self.payload_capacity
        );
        let lane = self.lane(node_id);
        match self.topology {
            RingTopology::Shared => {
                let counter = lane.write_pos.fetch_add(1, Ordering::AcqRel);
                self.write_frame(
                    lane, node_id, counter, frame_kind, ver, payload,
                )
            }
            RingTopology::PerNode | RingTopology::SharedOrdered => {
                let _write = lane
                    .write_lock
                    .lock()
                    .unwrap_or_else(|error| error.into_inner());
                let counter = lane.write_pos.load(Ordering::Relaxed);
                let id = self.write_frame(
                    lane, node_id, counter, frame_kind, ver, payload,
                );
                lane.write_pos
                    .store(counter.wrapping_add(1), Ordering::Release);
                id
            }
        }
    }

    /// Append a contiguous batch to one lane and return its consecutive ids.
    ///
    /// Per-node and shared-ordered lanes expose the new head only after the
    /// whole batch commits. An empty batch is a no-op. A batch larger than the
    /// ring is rejected because its first frames could not remain addressable
    /// when the method returns.
    pub fn write_batch(
        &self,
        node_id: NodeId,
        frame_kind: u8,
        ver: u64,
        payloads: Vec<Bytes>,
    ) -> Vec<NetId64> {
        assert!(
            payloads.len() <= self.capacity,
            "batch {} > ring capacity {}",
            payloads.len(),
            self.capacity
        );
        for payload in &payloads {
            assert!(
                payload.len() <= self.payload_capacity,
                "payload {} > ring payload capacity {}",
                payload.len(),
                self.payload_capacity
            );
        }
        if payloads.is_empty() {
            return Vec::new();
        }

        let lane = self.lane(node_id);
        match self.topology {
            RingTopology::Shared => {
                let start = lane
                    .write_pos
                    .fetch_add(payloads.len() as u64, Ordering::AcqRel);
                payloads
                    .into_iter()
                    .enumerate()
                    .map(|(offset, payload)| {
                        self.write_frame(
                            lane,
                            node_id,
                            start.wrapping_add(offset as u64),
                            frame_kind,
                            ver,
                            payload,
                        )
                    })
                    .collect()
            }
            RingTopology::PerNode | RingTopology::SharedOrdered => {
                let _write = lane
                    .write_lock
                    .lock()
                    .unwrap_or_else(|error| error.into_inner());
                let start = lane.write_pos.load(Ordering::Relaxed);
                let ids = payloads
                    .into_iter()
                    .enumerate()
                    .map(|(offset, payload)| {
                        self.write_frame(
                            lane,
                            node_id,
                            start.wrapping_add(offset as u64),
                            frame_kind,
                            ver,
                            payload,
                        )
                    })
                    .collect::<Vec<_>>();
                lane.write_pos.store(
                    start.wrapping_add(ids.len() as u64),
                    Ordering::Release,
                );
                ids
            }
        }
    }

    /// Read the slot that the given [`NetId64`]'s counter points at.
    ///
    /// Returns:
    /// - `Some(frame)` if the slot's stored id matches the queried id
    ///   exactly (the slot has not been overwritten by a later writer).
    /// - `None` if the slot is empty, has wrapped past, or holds a
    ///   different id than the one asked for.
    pub fn read(
        &self,
        id: NetId64,
    ) -> Option<Frame> {
        if id.kind() != self.kind {
            return None;
        }
        let lane = self.lane_for_frame(id)?;
        let slot_idx = (id.counter() as usize) % self.capacity;
        let guard = lane.slots[slot_idx].read().expect("ring slot poisoned");
        match &*guard {
            Some(f) if f.id == id => Some(f.clone()),
            _ => None,
        }
    }

    /// Read the most recent frame, regardless of who wrote it.
    /// Useful for "what's the current state?" — ignores
    /// counter-by-counter walking.
    pub fn read_head(&self) -> Option<Frame> {
        let head = self.head();
        if head == 0 {
            return None;
        }
        let slot_idx = ((head - 1) as usize) % self.capacity;
        self.lanes[0].slots[slot_idx]
            .read()
            .expect("ring slot poisoned")
            .clone()
    }

    /// Read whatever frame currently occupies the slot at
    /// `counter % capacity`, regardless of which counter is
    /// stored in it. Used by walking readers that need slot-by-slot
    /// access without knowing the writer's `NetId64` ahead of time.
    ///
    /// Returns `None` if the slot is empty.
    pub fn read_at(
        &self,
        counter: u64,
    ) -> Option<Frame> {
        let slot_idx = (counter as usize) % self.capacity;
        self.lanes[0].slots[slot_idx]
            .read()
            .expect("ring slot poisoned")
            .clone()
    }

    pub(crate) fn read_state_at(
        &self,
        counter: u64,
    ) -> cursor::RingRead {
        match self.read_at(counter) {
            Some(frame) if frame.id.counter() == counter => {
                cursor::RingRead::Ready(frame)
            }
            Some(frame) if frame.id.counter() > counter => {
                cursor::RingRead::Unavailable
            }
            Some(_) | None => cursor::RingRead::Pending,
        }
    }

    pub(crate) fn read_lane_at(
        &self,
        node_id: NodeId,
        counter: u64,
    ) -> Option<Frame> {
        let lane = self.lane(node_id);
        let slot_idx = (counter as usize) % self.capacity;
        lane.slots[slot_idx].read().expect("ring slot poisoned").clone()
    }

    pub(crate) fn read_lane_state_at(
        &self,
        node_id: NodeId,
        counter: u64,
    ) -> cursor::RingRead {
        match self.read_lane_at(node_id, counter) {
            Some(frame) if frame.id.counter() == counter => {
                cursor::RingRead::Ready(frame)
            }
            Some(frame) if frame.id.counter() > counter => {
                cursor::RingRead::Unavailable
            }
            Some(_) | None if self.topology != RingTopology::Shared => {
                cursor::RingRead::Unavailable
            }
            Some(_) | None => cursor::RingRead::Pending,
        }
    }

    /// Clear all slots and reset every lane head to zero.
    ///
    /// Intended for owner-controlled boot-time cleanup. Do not call
    /// while other threads are publishing to this ring.
    pub fn reset(&self) {
        for lane in &self.lanes {
            for slot in &lane.slots {
                *slot.write().expect("ring slot poisoned") = None;
            }
            lane.write_pos.store(0, Ordering::Release);
        }
        self.version_counter.store(0, Ordering::Release);
    }

    fn lane(
        &self,
        node_id: NodeId,
    ) -> &RingLane {
        let index = match self.topology {
            RingTopology::Shared | RingTopology::SharedOrdered => 0,
            RingTopology::PerNode => usize::from(node_id.get()),
        };
        self.lanes.get(index).unwrap_or_else(|| {
            panic!(
                "node {} is outside ring lane count {}",
                node_id.get(),
                self.lanes.len()
            )
        })
    }

    fn lane_for_frame(
        &self,
        id: NetId64,
    ) -> Option<&RingLane> {
        let index = match self.topology {
            RingTopology::Shared | RingTopology::SharedOrdered => 0,
            RingTopology::PerNode => usize::from(id.node()),
        };
        self.lanes.get(index)
    }

    fn write_frame(
        &self,
        lane: &RingLane,
        node_id: NodeId,
        counter: u64,
        frame_kind: u8,
        ver: u64,
        payload: Bytes,
    ) -> NetId64 {
        let id = NetId64::make(self.kind, node_id.get(), counter);
        let slot_idx = (counter as usize) % self.capacity;
        let frame = Frame { id, kind: frame_kind, ver, payload };
        let mut guard =
            lane.slots[slot_idx].write().expect("ring slot poisoned");
        *guard = Some(frame);
        id
    }
}

impl cursor::RingFrameSource for Ring {
    fn kind(&self) -> u8 {
        Ring::kind(self)
    }

    fn head(&self) -> u64 {
        Ring::head(self)
    }

    fn capacity(&self) -> usize {
        Ring::capacity(self)
    }

    fn read_at(
        &self,
        counter: u64,
    ) -> Option<Frame> {
        Ring::read_at(self, counter)
    }

    fn read_state_at(
        &self,
        counter: u64,
    ) -> cursor::RingRead {
        Ring::read_state_at(self, counter)
    }
}

#[cfg(unix)]
impl cursor::RingFrameSource for shm::ShmRing {
    fn kind(&self) -> u8 {
        shm::ShmRing::kind(self)
    }

    fn head(&self) -> u64 {
        shm::ShmRing::head(self)
    }

    fn capacity(&self) -> usize {
        shm::ShmRing::capacity(self)
    }

    fn read_at(
        &self,
        counter: u64,
    ) -> Option<Frame> {
        shm::ShmRing::read_at(self, counter)
    }

    fn read_state_at(
        &self,
        counter: u64,
    ) -> cursor::RingRead {
        shm::ShmRing::read_state_at(self, counter)
    }
}

impl std::fmt::Debug for Ring {
    fn fmt(
        &self,
        f: &mut std::fmt::Formatter<'_>,
    ) -> std::fmt::Result {
        f.debug_struct("Ring")
            .field("kind", &self.kind)
            .field("capacity", &self.capacity)
            .field("payload_capacity", &self.payload_capacity)
            .field("topology", &self.topology)
            .field("lane_count", &self.lanes.len())
            .field("head", &self.head())
            .finish()
    }
}

/// One slot per `OrbitTyped::KIND` byte.
///
/// What a kind resolves to is fixed once it is installed, so a hot-path
/// lookup is an index and one atomic load: no hashing, no shard lock and no
/// reference-count traffic.
pub(crate) struct ByKind<V> {
    slots: Box<[OnceLock<V>; 256]>,
}

impl<V> ByKind<V> {
    pub(crate) fn new() -> Self {
        Self { slots: Box::new(std::array::from_fn(|_| OnceLock::new())) }
    }

    pub(crate) fn get(
        &self,
        kind: u8,
    ) -> Option<&V> {
        self.slots[usize::from(kind)].get()
    }

    pub(crate) fn get_or_init(
        &self,
        kind: u8,
        init: impl FnOnce() -> V,
    ) -> &V {
        self.slots[usize::from(kind)].get_or_init(init)
    }

    /// Install `value` unless another caller installed one first, and return
    /// whichever is installed.
    pub(crate) fn get_or_insert(
        &self,
        kind: u8,
        value: V,
    ) -> &V {
        let slot = &self.slots[usize::from(kind)];
        let _ = slot.set(value);
        slot.get().expect("a kind slot holds a value once set")
    }

    pub(crate) fn len(&self) -> usize {
        self.slots.iter().filter(|slot| slot.get().is_some()).count()
    }
}

/// Type-keyed registry of rings. A `Fleet` holds one of these and
/// hands out the ring per `OrbitTyped` kind on demand.
pub(crate) struct RingRegistry {
    fleet_capacity: u16,
    rings: ByKind<Arc<Ring>>,
}

impl RingRegistry {
    pub fn new(fleet_capacity: u16) -> Self {
        Self { fleet_capacity, rings: ByKind::new() }
    }

    /// Get-or-create the ring declared by `T`, borrowed for the registry's
    /// lifetime.
    pub fn ring_for<T: OrbitTyped>(&self) -> &Arc<Ring> {
        let ring = self.rings.get_or_init(T::KIND, || {
            Arc::new(Ring::new_for_fleet::<T>(self.fleet_capacity))
        });
        assert_eq!(
            ring.spec(),
            T::RING_SPEC,
            "OrbitTyped KIND {} was reused with a different ring spec",
            T::KIND
        );
        ring
    }

    /// Get-or-create the ring declared by `T` as an owned handle.
    pub fn get_or_create<T: OrbitTyped>(&self) -> Arc<Ring> {
        self.ring_for::<T>().clone()
    }

    /// Look up a ring by KIND byte (e.g. when only the id is known).
    pub fn lookup(
        &self,
        kind: u8,
    ) -> Option<Arc<Ring>> {
        self.rings.get(kind).cloned()
    }
}