liminal-server 0.20.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.

use std::collections::BTreeSet;
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::operation_batch::{CHECKPOINT_EVERY, OperationBatch};
use super::registry_snapshot::{
    RegistrySnapshot, RegistrySnapshotError, SNAPSHOT_KEY, snapshot_item, write_registry_snapshot,
};
use super::routed_stream::{RoutedStreamHead, read_head, read_page};
use super::write_issuance::WriteIssuance;

/// 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.
pub(super) const SCHEMA_VERSION: u8 = 1;

#[derive(Clone, Copy, Debug, Deserialize, Serialize)]
struct RegistryRow {
    schema_version: u8,
    conversation_id: u64,
}

/// 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 ids of rows `0..sequence` and the snapshot witness.
#[derive(Debug, Default)]
struct RegistryHead {
    sequence: u64,
    uncertain: bool,
    ids: BTreeSet<u64>,
    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)
    }

    /// Reads every registered conversation id (deduplicated, ascending) from
    /// the snapshot and its tail, and installs the durable append head.
    ///
    /// With no snapshot (an install from before DUR-019) it walks the rows
    /// from 0 as before, then writes the snapshot once; a failed write is
    /// logged and repeated on the next boot, because it changes no row.
    ///
    /// # Errors
    ///
    /// Returns [`OperationLogError`] for an unreadable stream, undecodable
    /// row, unsupported schema version or refused snapshot — a corrupt
    /// registry fails startup loudly, and a refused snapshot is never walked
    /// around.
    pub(super) fn restore(&self) -> Result<Vec<u64>, OperationLogError> {
        let mut head = self.head();
        let stored = block_on(self.store.get_routed(REGISTRY_STREAM_KEY, SNAPSHOT_KEY))??;
        let snapshot = stored
            .as_deref()
            .map(|bytes| {
                RegistrySnapshot::decode(bytes, SCHEMA_VERSION)
                    .map(|snapshot| (snapshot, Hash::of(bytes)))
                    .map_err(OperationLogError::RegistrySnapshot)
            })
            .transpose()?;
        let heads = self.read_heads()?;
        let (ids, witness) = match snapshot {
            Some((snapshot, hash)) => (
                self.read_tail(snapshot, &heads)?,
                SnapshotWitness::Held(hash),
            ),
            None => self.migrate(&heads)?,
        };
        let restored = ids.iter().copied().collect();
        *head = RegistryHead {
            sequence: heads.registry,
            uncertain: false,
            ids,
            witness,
        };
        drop(head);
        Ok(restored)
    }

    /// Reads the routed head, then the native head, and refuses a routed
    /// head that stands below the native one.
    fn read_heads(&self) -> Result<Heads, OperationLogError> {
        let routed = block_on(read_head(
            self.store.as_ref(),
            REGISTRY_STREAM_KEY,
            REGISTRY_STREAM_KEY,
        ))??;
        let native = block_on(self.store.read_stream_next_seq(REGISTRY_STREAM_KEY))??.unwrap_or(0);
        let registry = match &routed {
            Some(routed) if routed.next_sequence < native => {
                return Err(OperationLogError::RegistrySnapshot(
                    RegistrySnapshotError::RoutedHeadBehindNative {
                        routed: routed.next_sequence,
                        native,
                    },
                ));
            }
            Some(routed) => routed.next_sequence,
            None => native,
        };
        Ok(Heads {
            routed,
            native,
            registry,
        })
    }

    /// The tail behind a decoded snapshot: flat rows up to the native head,
    /// then routed rows up to the registry head, with no read when empty.
    fn read_tail(
        &self,
        snapshot: RegistrySnapshot,
        heads: &Heads,
    ) -> Result<BTreeSet<u64>, OperationLogError> {
        let covered_next_seq = snapshot.covered_next_seq;
        let head = heads.registry;
        if covered_next_seq > head {
            return Err(OperationLogError::RegistrySnapshot(
                RegistrySnapshotError::AheadOfStream {
                    covered_next_seq,
                    head,
                },
            ));
        }
        if head.saturating_sub(covered_next_seq) >= CHECKPOINT_EVERY {
            return Err(OperationLogError::RegistrySnapshot(
                RegistrySnapshotError::TailBeyondBound {
                    covered_next_seq,
                    head,
                },
            ));
        }
        let mut ids = snapshot.ids;
        if covered_next_seq < heads.native {
            let end = self.read_flat(covered_next_seq, Some(heads.native), &mut ids)?;
            tail_ends_at(heads.native, end)?;
        }
        let end = self.read_routed(
            heads.routed.as_ref(),
            covered_next_seq.max(heads.native),
            &mut ids,
        )?;
        tail_ends_at(head, end)?;
        Ok(ids)
    }

    /// Walks an install with no snapshot from 0, as restore always did, then
    /// writes the snapshot at the registry head when any row was read.
    fn migrate(
        &self,
        heads: &Heads,
    ) -> Result<(BTreeSet<u64>, SnapshotWitness), OperationLogError> {
        let mut ids = BTreeSet::new();
        let flat_end = self.read_flat(0, None, &mut ids)?;
        tail_ends_at(heads.native, flat_end)?;
        let end = self.read_routed(heads.routed.as_ref(), heads.native, &mut ids)?;
        tail_ends_at(heads.registry, end)?;
        if heads.registry == 0 {
            return Ok((ids, SnapshotWitness::Absent));
        }
        // The snapshot read that found nothing is the absence witness. This
        // write changes no row, so a failure leaves the rows complete: the
        // ids stand, the next crossing in this process reads the key once,
        // and the next boot walks again.
        match write_registry_snapshot(self.store.as_ref(), None, heads.registry, &ids) {
            Ok(hash) => Ok((ids, SnapshotWitness::Held(hash))),
            Err(error) => {
                tracing::warn!(
                    route = REGISTRY_STREAM_KEY,
                    key = "registry/snapshot/v1",
                    covered_next_seq = heads.registry,
                    error = %error,
                    "registry snapshot migration write failed; the walked rows stand and the next boot writes it again"
                );
                Ok((ids, SnapshotWitness::Unknown))
            }
        }
    }

    /// Reads flat rows from `from`: unbounded (the walk, ending on a short
    /// page as it always did) or bounded by `until` (a tail, issuing no read
    /// when empty). Answers the sequence after the last row read.
    fn read_flat(
        &self,
        from: u64,
        until: Option<u64>,
        ids: &mut BTreeSet<u64>,
    ) -> Result<u64, OperationLogError> {
        let mut sequence = from;
        loop {
            let limit = match until {
                Some(end) if sequence >= end => break,
                Some(end) => usize::try_from(end.saturating_sub(sequence))
                    .map_or(READ_BATCH_SIZE, |remaining| remaining.min(READ_BATCH_SIZE)),
                None => READ_BATCH_SIZE,
            };
            let entries = block_on(self.store.read_from(REGISTRY_STREAM_KEY, sequence, limit))??;
            let count = entries.len();
            for entry in &entries {
                ids.insert(row_id(entry, sequence)?);
                sequence = following(sequence)?;
            }
            if count < limit {
                break;
            }
        }
        Ok(sequence)
    }

    /// Reads routed rows from `from` up to the validated routed head, in
    /// pages that never read past it.
    fn read_routed(
        &self,
        routed: Option<&RoutedStreamHead>,
        from: u64,
        ids: &mut BTreeSet<u64>,
    ) -> Result<u64, OperationLogError> {
        let mut sequence = from;
        let Some(routed) = routed else {
            return Ok(sequence);
        };
        while sequence < routed.next_sequence {
            let page = block_on(read_page(
                self.store.as_ref(),
                REGISTRY_STREAM_KEY,
                REGISTRY_STREAM_KEY,
                routed,
                sequence,
                READ_BATCH_SIZE,
            ))??;
            if page.is_empty() {
                break;
            }
            for entry in &page {
                ids.insert(row_id(entry, sequence)?);
                sequence = following(sequence)?;
            }
        }
        Ok(sequence)
    }

    /// 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())
    }

    /// Publishes one registry row with its routed head as one batch, and at a
    /// `CHECKPOINT_EVERY` crossing the snapshot with them.
    ///
    /// 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 an unresolved earlier issuance, a
    /// staging refusal or the batch's own failure.
    pub(super) fn register_with_issuance(
        &self,
        conversation_id: u64,
        issuance: &WriteIssuance,
    ) -> Result<(), OperationLogError> {
        let payload = serde_json::to_vec(&RegistryRow {
            schema_version: SCHEMA_VERSION,
            conversation_id,
        })?;
        let mut head = self.head();
        if head.uncertain {
            self.resolve_uncertain(&mut head)?;
        }
        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_registration(&mut batch, &mut head, payload, conversation_id, 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.ids.insert(conversation_id);
                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
    /// ids plus this one; answers the staged snapshot's hash.
    fn stage_registration(
        &self,
        batch: &mut OperationBatch,
        head: &mut RegistryHead,
        payload: Vec<u8>,
        conversation_id: u64,
        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 !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 mut ids = head.ids.clone();
        ids.insert(conversation_id);
        let (item, hash) = snapshot_item(after, &ids, 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 conversation_id = row_id(entry, issued)?;
        if issued / CHECKPOINT_EVERY != landed / CHECKPOINT_EVERY {
            head.witness = SnapshotWitness::Held(self.landed_snapshot(landed)?);
        }
        head.ids.insert(conversation_id);
        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(|bytes| RegistrySnapshot::decode(bytes, SCHEMA_VERSION))
            .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 conversation id.
fn row_id(entry: &StoredEntry, expected: u64) -> Result<u64, OperationLogError> {
    if entry.sequence != expected {
        return Err(OperationLogError::Sequence {
            expected,
            actual: entry.sequence,
        });
    }
    let row: RegistryRow = serde_json::from_slice(&entry.payload)?;
    if row.schema_version != SCHEMA_VERSION {
        return Err(OperationLogError::SchemaVersion(row.schema_version));
    }
    Ok(row.conversation_id)
}

/// 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,
    })
}

/// Refuses rows that did not end at the head bounding them.
const fn tail_ends_at(expected_head: u64, tail_end: u64) -> Result<(), OperationLogError> {
    if tail_end == expected_head {
        Ok(())
    } else {
        Err(OperationLogError::RegistrySnapshot(
            RegistrySnapshotError::TailMismatch {
                expected_head,
                tail_end,
            },
        ))
    }
}

/// 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()
}