liminal-server 0.24.1

//! Durable registry of every production participant conversation.
//!
//! One server-wide stream holds one row per created conversation, published
//! IMMEDIATELY BEFORE the conversation's own genesis append (the one
//! conversation-creating write). Startup reads the registry to enumerate
//! every durable conversation for the server-scope capacity restore.
//!
//! Ordering makes the registry complete by construction: a conversation
//! stream can exist only if its registry row was durable first. The reverse
//! window (registry row durable, genesis append failed) leaves a row whose
//! conversation replays empty — startup skips it exactly like a refused
//! probe, and a later retry of the creating enrollment re-registers, so
//! duplicate rows are legal and deduplicated on read.
//!
//! Layout (DUR-019). Rows an install wrote before routing stay flat below
//! the native head and are never rewritten; every row since is a routed row
//! of stream [`REGISTRY_STREAM_KEY`] on route [`REGISTRY_STREAM_KEY`],
//! published with its routed head as one [`OperationBatch`]. The
//! registration that crosses `CHECKPOINT_EVERY` carries the registry
//! snapshot ([`super::registry_snapshot`]) in that same batch, so restore
//! reads the snapshot and a tail of fewer than 256 rows, never the prefix.
//!
//! Retirement (DUR-027 R4). Rows are written at schema 2 with a kind:
//! `register` adds an id to the live set, `retire` moves it from live to
//! sweeping (the retirement's commit point), and `swept` removes it from
//! sweeping once its streams are cut. A schema 1 row is a registration. The
//! snapshot carries both sets, so neither restore nor the snapshot grows with
//! conversations that finished.

use std::sync::{Arc, Mutex, MutexGuard, PoisonError};

use haematite::tree::Hash;
use liminal::durability::bridge::block_on;
use liminal::durability::{DurabilityError, DurableStore, StoredEntry};
use serde::{Deserialize, Serialize};

use super::log::OperationLogError;
use super::log_error::RegistryRowRefusal;
use super::operation_batch::{CHECKPOINT_EVERY, OperationBatch};
use super::registry_snapshot::{
    RegistrySets, RegistrySnapshot, RegistrySnapshotError, SNAPSHOT_KEY, snapshot_item,
};
use super::routed_stream::{RoutedStreamHead, read_head, read_page};
use super::write_issuance::WriteIssuance;

#[path = "registry_read.rs"]
mod read;

/// Stream key, and route, of the server-wide conversation registry.
pub(super) const REGISTRY_STREAM_KEY: &str = "liminal:participant-conversation-registry";
/// Durable page size used during restore reads.
const READ_BATCH_SIZE: usize = 64;
/// Stored-row schema version written: a row with its kind.
pub(super) const SCHEMA_VERSION: u8 = 2;
/// Schema version of a row written before retirement: a registration.
pub(super) const SCHEMA_VERSION_V1: u8 = 1;

/// What one registry row does to the two sets.
#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub(super) enum RowKind {
    /// The id becomes live.
    Register,
    /// The id leaves live for sweeping: its retirement committed.
    Retire,
    /// The id leaves sweeping: its streams are cut.
    Swept,
}

#[derive(Clone, Copy, Debug, Deserialize, Serialize)]
struct RegistryRow {
    schema_version: u8,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    kind: Option<RowKind>,
    conversation_id: u64,
}

impl RegistrySets {
    /// Applies one row, refusing a retire of an id that is not live, a swept
    /// of an id that is not sweeping, and a registration of a sweeping id.
    /// A registration of a live id is a legal duplicate (see the module
    /// documentation) and changes nothing.
    pub(super) fn apply(
        &mut self,
        kind: RowKind,
        conversation_id: u64,
    ) -> Result<(), RegistryRowRefusal> {
        match kind {
            RowKind::Register if self.sweeping.contains(&conversation_id) => {
                Err(RegistryRowRefusal::RegisterSweeping)
            }
            RowKind::Register => {
                self.live.insert(conversation_id);
                Ok(())
            }
            RowKind::Retire if self.live.remove(&conversation_id) => {
                self.sweeping.insert(conversation_id);
                Ok(())
            }
            RowKind::Retire => Err(RegistryRowRefusal::RetireNotLive),
            RowKind::Swept if self.sweeping.remove(&conversation_id) => Ok(()),
            RowKind::Swept => Err(RegistryRowRefusal::SweptNotSweeping),
        }
    }
}

/// Which of the registry's sets holds one id.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum Membership {
    /// Registered and not retired.
    Live,
    /// Retired and not yet swept.
    Sweeping,
    /// Never registered, or swept.
    Absent,
}

/// Durable conversation-registry handle with its serialized append head.
///
/// Appends are rare (once per created conversation) and serialized under one
/// lock so concurrent creations cannot race the optimistic sequence.
#[derive(Debug)]
pub(super) struct ConversationRegistry {
    store: Arc<dyn DurableStore>,
    head: Mutex<RegistryHead>,
}

/// What this process knows of the stored snapshot's bytes: the witness its
/// next compare-and-swap is staged with.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
enum SnapshotWitness {
    /// A read proved no snapshot stands.
    Absent,
    /// The hash of the snapshot bytes that stand.
    Held(Hash),
    /// Not known: the next crossing reads the key once first.
    #[default]
    Unknown,
}

/// The append head, the sets rows `0..sequence` leave and the snapshot
/// witness.
#[derive(Debug, Default)]
struct RegistryHead {
    sequence: u64,
    uncertain: bool,
    sets: RegistrySets,
    witness: SnapshotWitness,
}

/// The registry's two heads as one read found them.
struct Heads {
    /// The routed head, when a routed registration has landed.
    routed: Option<RoutedStreamHead>,
    /// The flat native head; rows below it are flat.
    native: u64,
    /// The routed head when one stands, the native head otherwise.
    registry: u64,
}

impl ConversationRegistry {
    /// Creates the handle; the head is set by [`Self::restore`].
    pub(super) fn new(store: Arc<dyn DurableStore>) -> Self {
        Self {
            store,
            head: Mutex::new(RegistryHead::default()),
        }
    }

    /// Locks the append head, recovering from poison: the critical sections
    /// below contain no panicking operation (panics are denied crate-wide),
    /// so a poisoned guard can only come from a foreign unwind and never
    /// observes a torn head.
    fn head(&self) -> MutexGuard<'_, RegistryHead> {
        self.head.lock().unwrap_or_else(PoisonError::into_inner)
    }

    /// Durably registers one conversation id, flushed before returning — the
    /// caller's genesis append may only follow a durable registry row.
    ///
    /// # Errors
    ///
    /// Returns [`OperationLogError`] when the append or flush fails; the
    /// caller aborts the conversation-creating commit and publishes nothing.
    #[cfg(test)]
    pub(super) fn register(&self, conversation_id: u64) -> Result<(), OperationLogError> {
        self.register_with_issuance(conversation_id, &WriteIssuance::default())
    }

    /// Durably retires one live conversation id: the retirement's commit
    /// point (DUR-027 R4, R5 step 1).
    pub(super) fn retire(&self, conversation_id: u64) -> Result<(), OperationLogError> {
        self.publish(RowKind::Retire, conversation_id, &WriteIssuance::default())
    }

    /// Durably records one sweeping id as swept (DUR-027 R5 step 5).
    pub(super) fn swept(&self, conversation_id: u64) -> Result<(), OperationLogError> {
        self.publish(RowKind::Swept, conversation_id, &WriteIssuance::default())
    }

    /// Which set holds `conversation_id`, as the rows this process restored
    /// and published leave it.
    pub(super) fn membership(&self, conversation_id: u64) -> Membership {
        let head = self.head();
        if head.sets.live.contains(&conversation_id) {
            Membership::Live
        } else if head.sets.sweeping.contains(&conversation_id) {
            Membership::Sweeping
        } else {
            Membership::Absent
        }
    }

    /// Publishes one registration; see [`Self::publish`].
    ///
    /// # Errors
    ///
    /// As [`Self::publish`].
    pub(super) fn register_with_issuance(
        &self,
        conversation_id: u64,
        issuance: &WriteIssuance,
    ) -> Result<(), OperationLogError> {
        self.publish(RowKind::Register, conversation_id, issuance)
    }

    /// Publishes one registry row of `kind` with its routed head as one
    /// batch, and at a `CHECKPOINT_EVERY` crossing the snapshot with them. A
    /// row the sets refuse (a retire of an id not live, a swept of one not
    /// sweeping, a registration of a sweeping id) is refused before anything
    /// is staged.
    ///
    /// A staging refusal or a store `NotStarted` submits nothing and leaves
    /// the head certain; every other commit failure leaves it uncertain, and
    /// the next registration reads it back before it writes.
    ///
    /// # Errors
    ///
    /// Returns [`OperationLogError`] for a refused row, an unresolved
    /// earlier issuance, a staging refusal or the batch's own failure.
    pub(super) fn publish(
        &self,
        kind: RowKind,
        conversation_id: u64,
        issuance: &WriteIssuance,
    ) -> Result<(), OperationLogError> {
        let payload = serde_json::to_vec(&RegistryRow {
            schema_version: SCHEMA_VERSION,
            kind: Some(kind),
            conversation_id,
        })?;
        let mut head = self.head();
        if head.uncertain {
            self.resolve_uncertain(&mut head)?;
        }
        let mut sets = head.sets.clone();
        sets.apply(kind, conversation_id)
            .map_err(|refusal| OperationLogError::RegistryRow {
                sequence: head.sequence,
                conversation_id,
                refusal,
            })?;
        let after = following(head.sequence)?;
        let crossing = head.sequence / CHECKPOINT_EVERY != after / CHECKPOINT_EVERY;
        let mut batch = OperationBatch::new(REGISTRY_STREAM_KEY);
        let snapshot = self.stage_row(&mut batch, &mut head, payload, &sets, crossing)?;
        let committed = block_on(batch.commit(self.store.as_ref(), || issuance.mark_submitted()));
        let failure: OperationLogError = match committed {
            Ok(Ok(())) => {
                head.sequence = after;
                head.sets = sets;
                if let Some(hash) = snapshot {
                    head.witness = SnapshotWitness::Held(hash);
                }
                drop(head);
                return Ok(());
            }
            Ok(Err(error @ DurabilityError::NotStarted { .. })) => return Err(error.into()),
            Ok(Err(error)) => error.into(),
            Err(error) => error.into(),
        };
        head.uncertain = true;
        if crossing {
            head.witness = SnapshotWitness::Unknown;
        }
        drop(head);
        Err(failure)
    }

    /// Stages the row at the head and, at a crossing, the snapshot over the
    /// sets the row leaves; answers the staged snapshot's hash. The
    /// registry's first row (DUR-028 R4) carries the route-key migration
    /// marker with it: a store this code writes is under the namespace from
    /// its first conversation, so no boot of it ever migrates.
    fn stage_row(
        &self,
        batch: &mut OperationBatch,
        head: &mut RegistryHead,
        payload: Vec<u8>,
        sets: &RegistrySets,
        crossing: bool,
    ) -> Result<Option<Hash>, OperationLogError> {
        let timestamp = super::facts::now_unix_nanos()?;
        let after = block_on(batch.stage_stream(
            self.store.as_ref(),
            REGISTRY_STREAM_KEY,
            vec![payload],
            head.sequence,
            timestamp,
        ))??;
        if head.sequence == 0 {
            batch.stage(super::route_key_migration::fresh_marker())?;
        }
        if !crossing {
            return Ok(None);
        }
        let expected = match head.witness {
            SnapshotWitness::Absent => None,
            SnapshotWitness::Held(hash) => Some(hash),
            SnapshotWitness::Unknown => {
                let stored = block_on(self.store.get_routed(REGISTRY_STREAM_KEY, SNAPSHOT_KEY))??;
                let expected = stored.as_deref().map(Hash::of);
                head.witness = expected.map_or(SnapshotWitness::Absent, SnapshotWitness::Held);
                expected
            }
        };
        let (item, hash) = snapshot_item(after, sets, expected)?;
        batch.stage(item)?;
        Ok(Some(hash))
    }

    /// Reads back an issuance whose answer was lost: at the issued sequence
    /// nothing landed; one past it, the batch landed whole; anywhere else it
    /// stays unresolved.
    fn resolve_uncertain(&self, head: &mut RegistryHead) -> Result<(), OperationLogError> {
        let issued = head.sequence;
        let routed = block_on(read_head(
            self.store.as_ref(),
            REGISTRY_STREAM_KEY,
            REGISTRY_STREAM_KEY,
        ))??;
        let standing = match &routed {
            Some(routed) => routed.next_sequence,
            None => block_on(self.store.read_stream_next_seq(REGISTRY_STREAM_KEY))??.unwrap_or(0),
        };
        if standing == issued {
            head.uncertain = false;
            return Ok(());
        }
        let landed = following(issued)?;
        let (Some(routed), true) = (routed, standing == landed) else {
            return Err(unresolved(
                issued,
                format!(
                    "registry recovery found the stream head at {standing}, neither at the issued row nor one routed row past it"
                ),
            ));
        };
        let page = block_on(read_page(
            self.store.as_ref(),
            REGISTRY_STREAM_KEY,
            REGISTRY_STREAM_KEY,
            &routed,
            issued,
            1,
        ))??;
        let [entry] = page.as_slice() else {
            return Err(unresolved(
                issued,
                "registry recovery did not read the issued row".to_owned(),
            ));
        };
        let (kind, conversation_id) = row_of(entry, issued)?;
        if issued / CHECKPOINT_EVERY != landed / CHECKPOINT_EVERY {
            head.witness = SnapshotWitness::Held(self.landed_snapshot(landed)?);
        }
        head.sets.apply(kind, conversation_id).map_err(|refusal| {
            OperationLogError::RegistryRow {
                sequence: issued,
                conversation_id,
                refusal,
            }
        })?;
        head.sequence = landed;
        head.uncertain = false;
        Ok(())
    }

    /// The hash of the snapshot a landed crossing batch carried, refused by
    /// name unless it covers exactly the head the batch advanced to.
    fn landed_snapshot(&self, landed: u64) -> Result<Hash, OperationLogError> {
        let stored = block_on(self.store.get_routed(REGISTRY_STREAM_KEY, SNAPSHOT_KEY))??;
        let covered_next_seq = stored
            .as_deref()
            .map(RegistrySnapshot::decode)
            .transpose()
            .map_err(OperationLogError::RegistrySnapshot)?
            .map(|snapshot| snapshot.covered_next_seq);
        match stored {
            Some(bytes) if covered_next_seq == Some(landed) => Ok(Hash::of(&bytes)),
            _ => Err(OperationLogError::RegistrySnapshot(
                RegistrySnapshotError::CrossingUnmatched {
                    head: landed,
                    covered_next_seq,
                },
            )),
        }
    }
}

/// Decodes one registry row read at `expected`, checking its sequence and
/// schema version, and answers its kind and conversation id: a schema 1 row,
/// which carries no kind, is a registration; a schema 2 row carries one.
fn row_of(entry: &StoredEntry, expected: u64) -> Result<(RowKind, u64), OperationLogError> {
    if entry.sequence != expected {
        return Err(OperationLogError::Sequence {
            expected,
            actual: entry.sequence,
        });
    }
    let row: RegistryRow = serde_json::from_slice(&entry.payload)?;
    let kind = match (row.schema_version, row.kind) {
        (SCHEMA_VERSION_V1, None) => RowKind::Register,
        (SCHEMA_VERSION, Some(kind)) => kind,
        (version, _) => return Err(OperationLogError::SchemaVersion(version)),
    };
    Ok((kind, row.conversation_id))
}

/// Applies the row read at `sequence` to `sets`, naming a refusal.
fn apply_row(
    sets: &mut RegistrySets,
    entry: &StoredEntry,
    sequence: u64,
) -> Result<(), OperationLogError> {
    let (kind, conversation_id) = row_of(entry, sequence)?;
    sets.apply(kind, conversation_id)
        .map_err(|refusal| OperationLogError::RegistryRow {
            sequence,
            conversation_id,
            refusal,
        })
}

/// The sequence after `sequence`, refused by name at the end of its range.
fn following(sequence: u64) -> Result<u64, OperationLogError> {
    sequence.checked_add(1).ok_or(OperationLogError::Sequence {
        expected: u64::MAX,
        actual: sequence,
    })
}

/// The issuance whose landing the head read back cannot establish.
fn unresolved(sequence: u64, cause: String) -> OperationLogError {
    DurabilityError::AppendUnresolved {
        stream_key: REGISTRY_STREAM_KEY.to_owned(),
        sequence,
        cause,
    }
    .into()
}