liminal-server 0.24.1

//! The registry's read path: restore from the snapshot and its tail, the
//! flat and routed row walks, and the one-time migration to a snapshot (split
//! from [`super`] by a pure move under the 500-line lens, DUR-027 R1).

use haematite::tree::Hash;
use liminal::durability::bridge::block_on;

use super::super::log::OperationLogError;
use super::super::operation_batch::CHECKPOINT_EVERY;
use super::super::registry_snapshot::{
    RegistrySets, RegistrySnapshot, RegistrySnapshotError, SNAPSHOT_KEY, write_registry_snapshot,
};
use super::super::routed_stream::{RoutedStreamHead, read_head, read_page};
use super::{
    ConversationRegistry, Heads, READ_BATCH_SIZE, REGISTRY_STREAM_KEY, RegistryHead,
    SnapshotWitness, apply_row, following,
};

impl ConversationRegistry {
    /// Reads every live conversation id (ascending) from the snapshot and its
    /// tail, and installs the durable append head; see [`Self::restore_sets`].
    ///
    /// 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(in super::super) fn restore(&self) -> Result<Vec<u64>, OperationLogError> {
        Ok(self.restore_sets()?.live.into_iter().collect())
    }

    /// Reads the live and the sweeping ids from the snapshot and its tail,
    /// and installs the durable append head. The sweeping ids are the
    /// retirements a stop interrupted, which boot finishes (DUR-027 R6).
    ///
    /// # Errors
    ///
    /// As [`Self::restore`], and a row the sets refuse by name.
    pub(in super::super) fn restore_sets(&self) -> Result<RegistrySets, 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)
                    .map(|snapshot| (snapshot, Hash::of(bytes)))
                    .map_err(OperationLogError::RegistrySnapshot)
            })
            .transpose()?;
        let heads = self.read_heads()?;
        let (sets, witness) = match snapshot {
            Some((snapshot, hash)) => (
                self.read_tail(snapshot, &heads)?,
                SnapshotWitness::Held(hash),
            ),
            None => self.migrate(&heads)?,
        };
        let restored = sets.clone();
        *head = RegistryHead {
            sequence: heads.registry,
            uncertain: false,
            sets,
            witness,
        };
        drop(head);
        Ok(restored)
    }

    /// The live and the sweeping ids as the store holds them, read without
    /// writing: the snapshot and its tail when a snapshot stands, the whole
    /// walk otherwise; no snapshot is written and no head is installed
    /// (DUR-028 R6, the store census).
    ///
    /// # Errors
    ///
    /// As [`Self::restore_sets`].
    pub(in super::super) fn read_sets(&self) -> Result<RegistrySets, OperationLogError> {
        let stored = block_on(self.store.get_routed(REGISTRY_STREAM_KEY, SNAPSHOT_KEY))??;
        let heads = self.read_heads()?;
        if let Some(bytes) = stored.as_deref() {
            let snapshot =
                RegistrySnapshot::decode(bytes).map_err(OperationLogError::RegistrySnapshot)?;
            return self.read_tail(snapshot, &heads);
        }
        let mut sets = RegistrySets::default();
        let flat_end = self.read_flat(0, None, &mut sets)?;
        tail_ends_at(heads.native, flat_end)?;
        let end = self.read_routed(heads.routed.as_ref(), heads.native, &mut sets)?;
        tail_ends_at(heads.registry, end)?;
        Ok(sets)
    }

    /// 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<RegistrySets, 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 sets = snapshot.sets;
        if covered_next_seq < heads.native {
            let end = self.read_flat(covered_next_seq, Some(heads.native), &mut sets)?;
            tail_ends_at(heads.native, end)?;
        }
        let end = self.read_routed(
            heads.routed.as_ref(),
            covered_next_seq.max(heads.native),
            &mut sets,
        )?;
        tail_ends_at(head, end)?;
        Ok(sets)
    }

    /// 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<(RegistrySets, SnapshotWitness), OperationLogError> {
        let mut sets = RegistrySets::default();
        let flat_end = self.read_flat(0, None, &mut sets)?;
        tail_ends_at(heads.native, flat_end)?;
        let end = self.read_routed(heads.routed.as_ref(), heads.native, &mut sets)?;
        tail_ends_at(heads.registry, end)?;
        if heads.registry == 0 {
            return Ok((sets, 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, &sets) {
            Ok(hash) => Ok((sets, 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((sets, 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>,
        sets: &mut RegistrySets,
    ) -> 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 {
                apply_row(sets, 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,
        sets: &mut RegistrySets,
    ) -> 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 {
                apply_row(sets, entry, sequence)?;
                sequence = following(sequence)?;
            }
        }
        Ok(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,
            },
        ))
    }
}