liminal-server 0.21.1

//! Durable row log for the server-wide observer-recovery aggregate.
//!
//! Rows persist observer-progress registrations, progress advances, and
//! installed arm plans. Restore folds rows back through the aggregate's own
//! A4 transactions, so validation lives in the protocol crate — a row the
//! aggregate would refuse fails restore loudly.
//!
//! Restore never reads the whole stream twice (DUR-008 A3). Every
//! [`OBSERVER_SNAPSHOT_EVERY`] rows the appender writes a snapshot row of the
//! whole aggregate, in the same `append_all` and flush as the row after it,
//! and records the snapshot's position under a head key, as a hint. Restore
//! reads that key, loads the snapshot with one point read, verifies it, and
//! folds only the rows after it. A key that names a row which is not a
//! snapshot is refused by name and logged; it and an absent key are followed
//! by a fold of the whole stream, after which a long stream takes a snapshot.
//! A snapshot met in the tail must equal the fold so far.

use liminal::durability::settle::{append_all_settled, cas_settled, flush_settled};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use liminal::durability::{DurabilityError, DurableStore};
use liminal_protocol::lifecycle::{
    ObserverProgressAdvanceDecision, ObserverProgressTrackDecision, ObserverRecoveryAggregate,
};
use liminal_protocol::wire::{ConversationId, DeliverySeq, ObserverEpoch};
use serde::{Deserialize, Serialize};

use super::log::OperationLogError;

/// Stream key of the server-wide observer-recovery row log.
pub(super) const OBSERVER_STREAM_KEY: &str = "liminal:participant-observer-recovery";
/// Durable page size used during restore reads.
const READ_BATCH_SIZE: usize = 64;
/// Stored-row schema version of the operation rows.
const SCHEMA_VERSION: u8 = 1;
/// Stored-row schema version of a snapshot row.
///
/// Downgrade across a snapshot row is unsupported. A binary from before
/// snapshots parses the whole row before it checks the version, so it fails
/// on the unknown `snapshot` variant (a JSON error), not by version name. The
/// supported path is upgrade: a stream with no snapshot row is folded whole
/// once, and a snapshot is then taken (see [`ObserverLog::restore`]).
const SNAPSHOT_SCHEMA_VERSION: u8 = 2;
/// Snapshot records that failed to write since the process started; each is a
/// restore that folds from an older snapshot, never a lost row.
static SNAPSHOT_RECORD_FAILURES: AtomicU64 = AtomicU64::new(0);

/// How many snapshot records have failed to write since the process started.
#[cfg(test)]
pub(super) fn snapshot_record_failures() -> u64 {
    SNAPSHOT_RECORD_FAILURES.load(Ordering::Relaxed)
}

/// Value key holding one past the sequence of the latest durable snapshot row.
pub(super) const OBSERVER_SNAPSHOT_KEY: &str = "liminal:participant-observer-recovery:snapshot";
/// How many rows the stream may hold after its latest snapshot: a restore
/// folds at most this many rows, whatever the stream's history.
pub(super) const OBSERVER_SNAPSHOT_EVERY: u64 = 256;

/// One durable observer-recovery row.
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "snake_case", tag = "row")]
pub(super) enum ObserverRow {
    /// A conversation's observer progress became tracked.
    Track {
        /// Newly tracked conversation.
        conversation_id: ConversationId,
        /// Initial durable observer progress.
        observer_progress: DeliverySeq,
    },
    /// A tracked conversation's observer progress advanced.
    Advance {
        /// Advancing conversation.
        conversation_id: ConversationId,
        /// New strictly greater progress.
        observer_progress: DeliverySeq,
    },
    /// One complete equal-epoch arm plan was installed atomically.
    Arms {
        /// Every arm of the plan, in conversation order.
        arms: Vec<(ConversationId, ObserverEpoch)>,
    },
    /// The whole aggregate as folded through the row before this one.
    Snapshot {
        /// Every progress row, in conversation order.
        progress: Vec<(ConversationId, DeliverySeq)>,
        /// Every installed arm, in conversation order.
        armed: Vec<(ConversationId, ObserverEpoch)>,
    },
}

impl ObserverRow {
    /// The snapshot row of `aggregate`.
    pub(super) fn snapshot_of(aggregate: &ObserverRecoveryAggregate) -> Self {
        Self::Snapshot {
            progress: aggregate.progress_rows(),
            armed: aggregate.armed_rows(),
        }
    }
}

/// Whether a snapshot row is owed at `head`: the snapshot then takes `head`
/// and the row being appended takes the sequence after it.
pub(super) const fn snapshot_due(head: u64) -> bool {
    head % (OBSERVER_SNAPSHOT_EVERY + 1) == OBSERVER_SNAPSHOT_EVERY
}

/// The snapshot row owed before a row appended at `head`, taken from the
/// aggregate the row applies to; `None` when none is due. Callers take it
/// before the aggregate moves into its transaction.
pub(super) fn snapshot_before(
    head: u64,
    aggregate: &ObserverRecoveryAggregate,
) -> Option<ObserverRow> {
    snapshot_due(head).then(|| ObserverRow::snapshot_of(aggregate))
}

/// The head after appending `rows` operation rows at `head`, counting the
/// snapshot rows the appender writes on the way.
pub(super) fn head_after(head: u64, rows: u64) -> Option<u64> {
    let mut next = head;
    for _ in 0..rows {
        next = next.checked_add(if snapshot_due(next) { 2 } else { 1 })?;
    }
    Some(next)
}

/// Why a recorded snapshot position was not trusted.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum SnapshotRefusal {
    /// The stream holds rows and no snapshot position is recorded.
    Missing,
    /// The recorded position names a row the stream does not hold.
    BeyondLog,
    /// The recorded position names a row that is not a snapshot.
    NotASnapshot,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
struct StoredRow {
    schema_version: u8,
    row: ObserverRow,
}

/// Restored aggregate together with the next durable row sequence.
#[derive(Debug)]
pub(super) struct RestoredObserver {
    /// Aggregate rebuilt through its own transactions.
    pub(super) aggregate: ObserverRecoveryAggregate,
    /// Optimistic head for the next append.
    pub(super) next_sequence: u64,
}

/// Append-only handle over the observer row stream.
#[derive(Debug)]
pub(super) struct ObserverLog {
    store: Arc<dyn DurableStore>,
}

impl ObserverLog {
    /// Creates a stateless handle over the shared store.
    pub(super) fn new(store: Arc<dyn DurableStore>) -> Self {
        Self { store }
    }

    /// Appends one row at the optimistic head and flushes once, answering the
    /// next head.
    ///
    /// `snapshot` is [`snapshot_before`] at `expected_sequence`: when one is
    /// due it takes `expected_sequence`, the row takes the sequence after it,
    /// both go in one `append_all` and one flush, and the snapshot's position
    /// is then recorded. A snapshot given when none is due, or missing when
    /// one is, is refused before anything is written.
    pub(super) async fn append(
        &self,
        row: &ObserverRow,
        expected_sequence: u64,
        snapshot: Option<ObserverRow>,
    ) -> Result<u64, OperationLogError> {
        self.append_rows(row, expected_sequence, snapshot, true)
            .await
    }

    /// [`Self::append`] without its flush, for fixtures that write a long
    /// history through the same placement and then flush once.
    #[cfg(test)]
    pub(super) async fn append_unflushed_for_tests(
        &self,
        row: &ObserverRow,
        expected_sequence: u64,
        snapshot: Option<ObserverRow>,
    ) -> Result<u64, OperationLogError> {
        self.append_rows(row, expected_sequence, snapshot, false)
            .await
    }

    /// Flushes the shared store, for fixtures that appended without flushing.
    #[cfg(test)]
    pub(super) async fn flush_for_tests(&self) -> Result<(), OperationLogError> {
        Ok(flush_settled(self.store.as_ref()).await?)
    }

    async fn append_rows(
        &self,
        row: &ObserverRow,
        expected_sequence: u64,
        snapshot: Option<ObserverRow>,
        flush: bool,
    ) -> Result<u64, OperationLogError> {
        if snapshot.is_some() != snapshot_due(expected_sequence)
            || matches!(row, ObserverRow::Snapshot { .. })
            || snapshot
                .as_ref()
                .is_some_and(|snapshot| !matches!(snapshot, ObserverRow::Snapshot { .. }))
        {
            return Err(OperationLogError::CorruptRow {
                sequence: expected_sequence,
            });
        }
        let mut payloads = Vec::with_capacity(2);
        if let Some(snapshot) = &snapshot {
            payloads.push(encode(snapshot)?);
        }
        payloads.push(encode(row)?);
        let rows = if snapshot.is_some() { 2 } else { 1 };
        let next = append_all_settled(
            self.store.as_ref(),
            OBSERVER_STREAM_KEY,
            payloads,
            expected_sequence,
        )
        .await?;
        let wanted = expected_sequence
            .checked_add(rows)
            .ok_or(OperationLogError::Sequence {
                expected: u64::MAX,
                actual: expected_sequence,
            })?;
        if next != wanted {
            return Err(OperationLogError::AssignedSequence {
                expected: wanted,
                actual: next,
            });
        }
        if flush {
            flush_settled(self.store.as_ref()).await?;
        }
        if snapshot.is_some() {
            self.record_snapshot_hint(expected_sequence).await;
        }
        Ok(next)
    }

    /// Records the snapshot row at `sequence` as the latest, as a hint.
    ///
    /// The rows are durable before this runs, and restore already tolerates
    /// a record that is stale or absent (it folds from the older snapshot, or
    /// the whole stream). So a failure here is logged and counted
    /// ([`snapshot_record_failures`]) and never answered as a failed append:
    /// an append whose rows landed must move its caller's head past them.
    async fn record_snapshot_hint(&self, sequence: u64) {
        if let Err(error) = self.record_snapshot(sequence).await {
            SNAPSHOT_RECORD_FAILURES.fetch_add(1, Ordering::Relaxed);
            tracing::warn!(
                stream = OBSERVER_STREAM_KEY,
                sequence,
                %error,
                "observer snapshot record not written; restore folds from the previous one"
            );
        }
    }

    /// Records the snapshot row at `sequence` as the latest, unless a later
    /// one is already recorded.
    async fn record_snapshot(&self, sequence: u64) -> Result<(), DurabilityError> {
        let Some(recorded) = sequence.checked_add(1) else {
            return Err(DurabilityError::ConfigError(
                "observer snapshot position overflowed".to_owned(),
            ));
        };
        let mut stored = self
            .store
            .read_value(OBSERVER_SNAPSHOT_KEY)
            .await?
            .unwrap_or(0);
        while stored < recorded {
            match cas_settled(self.store.as_ref(), OBSERVER_SNAPSHOT_KEY, stored, recorded).await {
                Ok(()) => return Ok(()),
                Err(DurabilityError::CursorRegression {
                    stored: observed, ..
                }) if observed != stored => stored = observed,
                Err(error) => return Err(error),
            }
        }
        Ok(())
    }

    /// Rebuilds the aggregate from the latest recorded snapshot and the rows
    /// after it, in one fold over that tail. A refused or absent record falls
    /// back to folding the whole stream.
    pub(super) async fn restore(&self) -> Result<RestoredObserver, OperationLogError> {
        let (aggregate, from) = match self.load_snapshot().await? {
            Ok(loaded) => loaded,
            Err(refusal) => {
                // An absent record is a stream that has not reached its first
                // snapshot, or one written before snapshots: folded whole,
                // without a read spent deciding which.
                if refusal != SnapshotRefusal::Missing {
                    tracing::warn!(
                        stream = OBSERVER_STREAM_KEY,
                        ?refusal,
                        "observer snapshot record refused; folding the whole stream"
                    );
                }
                (ObserverRecoveryAggregate::new(), 0)
            }
        };
        let (aggregate, mut next_sequence, latest) = self.fold_from(aggregate, from).await?;
        if let Some(latest) = latest {
            self.record_snapshot_hint(latest).await;
        } else if from == 0 && next_sequence > OBSERVER_SNAPSHOT_EVERY {
            // A stream written before snapshots (or one whose every snapshot
            // record was lost) would otherwise be folded whole on every boot
            // until an append happened to land on a due position.
            next_sequence = self
                .write_restore_snapshot(&aggregate, next_sequence)
                .await?;
        }
        Ok(RestoredObserver {
            aggregate,
            next_sequence,
        })
    }

    /// Appends a snapshot of `aggregate` at `head` after a whole-stream fold
    /// found none, under the same rule as an appended one (one `append_all`,
    /// one flush, then the record as a hint), answering the head after it.
    async fn write_restore_snapshot(
        &self,
        aggregate: &ObserverRecoveryAggregate,
        head: u64,
    ) -> Result<u64, OperationLogError> {
        let payload = encode(&ObserverRow::snapshot_of(aggregate))?;
        let next = append_all_settled(
            self.store.as_ref(),
            OBSERVER_STREAM_KEY,
            vec![payload],
            head,
        )
        .await?;
        let wanted = head.checked_add(1).ok_or(OperationLogError::Sequence {
            expected: u64::MAX,
            actual: head,
        })?;
        if next != wanted {
            return Err(OperationLogError::AssignedSequence {
                expected: wanted,
                actual: next,
            });
        }
        flush_settled(self.store.as_ref()).await?;
        self.record_snapshot_hint(head).await;
        Ok(next)
    }

    /// The reference restore: a fold of the whole stream from its first row.
    #[cfg(test)]
    pub(super) async fn restore_whole_for_tests(
        &self,
    ) -> Result<RestoredObserver, OperationLogError> {
        let (aggregate, next_sequence, _) =
            self.fold_from(ObserverRecoveryAggregate::new(), 0).await?;
        Ok(RestoredObserver {
            aggregate,
            next_sequence,
        })
    }

    /// Reconciles the one uncertain progress append without copying its owner.
    pub(super) async fn recover_progress_tail(
        &self,
        aggregate: &mut ObserverRecoveryAggregate,
        head: &mut u64,
    ) -> Result<Option<(liminal_protocol::lifecycle::ObserverRecoveryArm, u64)>, OperationLogError>
    {
        let entries = self.store.read_from(OBSERVER_STREAM_KEY, *head, 3).await?;
        if entries.is_empty() {
            // The issued append never landed when the stream's head still
            // stands at the issued sequence: nothing to reconcile, and the
            // head is the original one. A stream past the issued row whose
            // rows cannot be read stays unresolved.
            let next = self
                .store
                .read_stream_next_seq(OBSERVER_STREAM_KEY)
                .await?
                .unwrap_or(0);
            if next == *head {
                return Ok(None);
            }
            return Err(DurabilityError::AppendUnresolved {
                stream_key: OBSERVER_STREAM_KEY.to_owned(),
                sequence: *head,
                cause: format!(
                    "observer recovery found the stream head at {next}, past the issued progress append it could not read"
                ),
            }
            .into());
        }
        if entries.len() > 2 {
            return Err(DurabilityError::AppendUnresolved {
                stream_key: OBSERVER_STREAM_KEY.to_owned(),
                sequence: *head,
                cause: "observer recovery did not establish the issued progress append".to_owned(),
            }
            .into());
        }
        let mut publication = None;
        for entry in entries {
            if entry.sequence != *head {
                return Err(OperationLogError::Sequence {
                    expected: *head,
                    actual: entry.sequence,
                });
            }
            let next = head.checked_add(1).ok_or(OperationLogError::Sequence {
                expected: u64::MAX,
                actual: entry.sequence,
            })?;
            match decode(&entry.payload)? {
                ObserverRow::Track {
                    conversation_id,
                    observer_progress,
                } => {
                    match std::mem::take(aggregate).decide_track(conversation_id, observer_progress)
                    {
                        ObserverProgressTrackDecision::Commit(transaction) => {
                            *aggregate = transaction.commit();
                        }
                        ObserverProgressTrackDecision::Refuse {
                            aggregate: retained,
                            ..
                        } => {
                            *aggregate = retained;
                            return Err(refused(entry.sequence));
                        }
                    }
                }
                ObserverRow::Advance {
                    conversation_id,
                    observer_progress,
                } => {
                    match std::mem::take(aggregate)
                        .decide_progress_advance(conversation_id, observer_progress)
                    {
                        ObserverProgressAdvanceDecision::Commit(transaction) => {
                            let (updated, fired) = transaction.commit();
                            *aggregate = updated;
                            publication = fired.map(|arm| (arm, observer_progress));
                        }
                        ObserverProgressAdvanceDecision::Refuse {
                            aggregate: retained,
                            ..
                        } => {
                            *aggregate = retained;
                            return Err(refused(entry.sequence));
                        }
                    }
                }
                ObserverRow::Snapshot { progress, armed } => {
                    if aggregate.progress_rows() != progress || aggregate.armed_rows() != armed {
                        return Err(refused(entry.sequence));
                    }
                }
                ObserverRow::Arms { .. } => return Err(refused(entry.sequence)),
            }
            *head = next;
        }
        Ok(publication)
    }

    /// Reads the recorded snapshot position and the snapshot row it names,
    /// answering the aggregate it restores and the sequence after it.
    async fn load_snapshot(
        &self,
    ) -> Result<Result<(ObserverRecoveryAggregate, u64), SnapshotRefusal>, OperationLogError> {
        let Some(recorded) = self
            .store
            .read_value(OBSERVER_SNAPSHOT_KEY)
            .await?
            .filter(|recorded| *recorded > 0)
        else {
            return Ok(Err(SnapshotRefusal::Missing));
        };
        let at = recorded - 1;
        let Some(entry) = self.store.read_at(OBSERVER_STREAM_KEY, at).await? else {
            return Ok(Err(SnapshotRefusal::BeyondLog));
        };
        if entry.sequence != at {
            return Ok(Err(SnapshotRefusal::BeyondLog));
        }
        let ObserverRow::Snapshot { progress, armed } = decode(&entry.payload)? else {
            return Ok(Err(SnapshotRefusal::NotASnapshot));
        };
        let aggregate =
            ObserverRecoveryAggregate::restore(&progress, &armed).map_err(|_| refused(at))?;
        Ok(Ok((aggregate, recorded)))
    }

    /// Folds every row from `from` to the stream's end onto `aggregate`,
    /// answering the aggregate, the next sequence and the latest snapshot row
    /// met on the way.
    async fn fold_from(
        &self,
        mut aggregate: ObserverRecoveryAggregate,
        from: u64,
    ) -> Result<(ObserverRecoveryAggregate, u64, Option<u64>), OperationLogError> {
        let mut sequence = from;
        let mut latest = None;
        loop {
            let entries = self
                .store
                .read_from(OBSERVER_STREAM_KEY, sequence, READ_BATCH_SIZE)
                .await?;
            if entries.is_empty() {
                break;
            }
            let count = entries.len();
            for entry in entries {
                if entry.sequence != sequence {
                    return Err(OperationLogError::Sequence {
                        expected: sequence,
                        actual: entry.sequence,
                    });
                }
                let row = decode(&entry.payload)?;
                if matches!(row, ObserverRow::Snapshot { .. }) {
                    latest = Some(entry.sequence);
                }
                aggregate = fold_row(aggregate, row, entry.sequence)?;
                sequence = sequence.checked_add(1).ok_or(OperationLogError::Sequence {
                    expected: u64::MAX,
                    actual: entry.sequence,
                })?;
            }
            if count < READ_BATCH_SIZE {
                break;
            }
        }
        Ok((aggregate, sequence, latest))
    }
}

/// Encodes one row at its schema version.
fn encode(row: &ObserverRow) -> Result<Vec<u8>, OperationLogError> {
    let schema_version = if matches!(row, ObserverRow::Snapshot { .. }) {
        SNAPSHOT_SCHEMA_VERSION
    } else {
        SCHEMA_VERSION
    };
    Ok(serde_json::to_vec(&StoredRow {
        schema_version,
        row: row.clone(),
    })?)
}

/// Encodes one row exactly as the appender stores it, for fixtures that write
/// a stream the way an older binary did (operation rows only, schema 1).
#[cfg(test)]
pub(super) fn encode_for_tests(row: &ObserverRow) -> Result<Vec<u8>, OperationLogError> {
    encode(row)
}

/// Decodes one row, refusing a schema version its kind does not carry.
fn decode(payload: &[u8]) -> Result<ObserverRow, OperationLogError> {
    let stored: StoredRow = serde_json::from_slice(payload)?;
    let expected = if matches!(stored.row, ObserverRow::Snapshot { .. }) {
        SNAPSHOT_SCHEMA_VERSION
    } else {
        SCHEMA_VERSION
    };
    if stored.schema_version != expected {
        return Err(OperationLogError::SchemaVersion(stored.schema_version));
    }
    Ok(stored.row)
}

/// Applies one durable row through the aggregate's own barriered mutations.
fn fold_row(
    aggregate: ObserverRecoveryAggregate,
    row: ObserverRow,
    sequence: u64,
) -> Result<ObserverRecoveryAggregate, OperationLogError> {
    match row {
        ObserverRow::Track {
            conversation_id,
            observer_progress,
        } => match aggregate.decide_track(conversation_id, observer_progress) {
            ObserverProgressTrackDecision::Commit(transaction) => Ok(transaction.commit()),
            ObserverProgressTrackDecision::Refuse { .. } => Err(refused(sequence)),
        },
        ObserverRow::Advance {
            conversation_id,
            observer_progress,
        } => match aggregate.decide_progress_advance(conversation_id, observer_progress) {
            ObserverProgressAdvanceDecision::Commit(transaction) => {
                let (aggregate, _fired) = transaction.commit();
                Ok(aggregate)
            }
            ObserverProgressAdvanceDecision::Refuse { .. } => Err(refused(sequence)),
        },
        ObserverRow::Arms { arms } => aggregate
            .install_restored_arms(&arms)
            .map_err(|_| refused(sequence)),
        ObserverRow::Snapshot { progress, armed } => {
            // A snapshot restates the fold so far; one that disagrees is a
            // corrupt stream.
            if aggregate.progress_rows() == progress && aggregate.armed_rows() == armed {
                Ok(aggregate)
            } else {
                Err(refused(sequence))
            }
        }
    }
}

/// A durable row the aggregate itself refuses is a corrupt stream.
const fn refused(sequence: u64) -> OperationLogError {
    OperationLogError::CorruptRow { sequence }
}