liminal-server 0.19.1

//! Ordered merge of literal v2 base rows and schema-v1 Unit 2 extension rows.

use liminal_protocol::lifecycle::RecipientAckObligations;
#[cfg(test)]
use std::cell::Cell;
#[cfg(test)]
use std::collections::VecDeque;

use super::outbox::{ConversationOutbox, ConversationOutboxError, ConversationOutboxLimits};
use super::outbox_log::{
    OutboxLog, OutboxLogError, OutboxRestoreCursor, OutboxRow, UNIT2_OUTBOX_RESTORE_BATCH_ROWS,
};
use super::state::{ConversationAuthority, StateError};

/// Restore failure with cursor faults kept separate from semantic replay.
#[derive(Debug, thiserror::Error)]
pub(super) enum RestoreError {
    /// Extension cursor durability, stream, schema, or codec failure.
    #[error(transparent)]
    Extension(#[from] OutboxLogError),
    /// Base/extension merge, authority replay, repair, or finish failure.
    #[error(transparent)]
    Semantic(#[from] StateError),
}

/// How one replay pass holds the extension repair rows it owes.
///
/// Either way the merge never holds more than one page of
/// [`UNIT2_OUTBOX_RESTORE_BATCH_ROWS`] repair rows (DUR-008 D03, Hermes
/// 20:18): what a pass owes beyond one page is written, or not held at all.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum RepairPaging {
    /// The first pass, which has not yet audited the log: up to one page of
    /// repairs is staged and nothing is written until the whole log audits
    /// clean. Owing more than one page drops the staged rows and stops
    /// staging; the pass still applies every repair to the owner, finishes
    /// its audit, and then answers [`StagedRepairs::Overflowed`].
    StageOnePage,
    /// The second pass, run only after a first pass audited the same log
    /// clean: each page of repairs is written the moment it fills, and the
    /// last, partial page when the replay persists.
    WriteEachPage,
}

/// What [`ExtensionMerge::persist_staged_repairs`] did with the repairs the
/// pass owed.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum StagedRepairs {
    /// Every repair the pass owed is durable.
    Persisted,
    /// The pass owed more than one page under [`RepairPaging::StageOnePage`]
    /// and wrote none of it: the replay must run again under
    /// [`RepairPaging::WriteEachPage`].
    Overflowed,
}

#[cfg(test)]
std::thread_local! {
    /// The most repair rows any merge on this thread held at once.
    static STAGED_REPAIR_HIGH_WATER: Cell<usize> = const { Cell::new(0) };
}

/// The most repair rows a merge on this thread has held at once since the
/// last call, which resets it.
#[cfg(test)]
pub(super) fn take_staged_repair_high_water() -> usize {
    STAGED_REPAIR_HIGH_WATER.with(|high_water| high_water.replace(0))
}

#[cfg(test)]
fn record_staged_repairs(rows: usize) {
    STAGED_REPAIR_HIGH_WATER.with(|high_water| high_water.set(high_water.get().max(rows)));
}

/// Move-only merge/reconciliation transaction for one cold first touch.
pub(super) struct ExtensionMerge<'a> {
    log: &'a OutboxLog,
    cursor: OutboxRestoreCursor<'a>,
    outbox: ConversationOutbox,
    repair_head: Option<u64>,
    /// Repair rows already applied to `outbox`, in extension order, and not
    /// yet durable: never more than one page. [`Self::persist_staged_repairs`]
    /// writes them, and under [`RepairPaging::WriteEachPage`] so does each
    /// page as it fills; nothing else does, so the replay decides WHEN a
    /// repair reaches the store.
    staged_repairs: Vec<(u64, OutboxRow)>,
    paging: RepairPaging,
    /// Set once [`RepairPaging::StageOnePage`] owed more than one page and
    /// dropped what it had staged.
    repairs_overflowed: bool,
    /// The first row [`Self::validate_rest`] found unconsumed, kept as its
    /// physical sequence and base boundary so [`Self::finish`] refuses it
    /// after the rest of the stream has been decoded past it.
    leftover: Option<(u64, Option<u64>)>,
}

impl<'a> ExtensionMerge<'a> {
    pub(super) fn new(
        log: &'a OutboxLog,
        conversation_id: u64,
        limits: ConversationOutboxLimits,
        paging: RepairPaging,
    ) -> Result<Self, StateError> {
        Ok(Self {
            log,
            cursor: log.restore_cursor(),
            outbox: ConversationOutbox::restore(conversation_id, Vec::new(), limits)?,
            repair_head: None,
            staged_repairs: Vec::new(),
            paging,
            repairs_overflowed: false,
            leftover: None,
        })
    }

    /// Decodes every extension row the merge has not consumed, through
    /// confirmed EOF, on the merge's own cursor.
    ///
    /// The stream is read once: the merge consumes it in order and this
    /// finishes it. Called before any staged repair is persisted and before any
    /// refusal is answered, it keeps the order a validate-first pass gave: an
    /// undecodable extension row anywhere is the refusal, ahead of every base
    /// or merge failure, and nothing is written over a stream that does not
    /// decode.
    pub(super) async fn validate_rest(&mut self) -> Result<(), OutboxLogError> {
        if self.leftover.is_none() {
            if let Some((physical_sequence, row)) = self.cursor.front().await? {
                self.leftover = Some((*physical_sequence, row.base_log_head()));
            }
        }
        self.cursor.drain().await
    }

    pub(super) fn recipient_ack_obligations(
        &self,
        participant_id: u64,
        acknowledged_through: u64,
    ) -> Result<(RecipientAckObligations, u64), StateError> {
        Ok(self
            .outbox
            .recipient_ack_obligations(participant_id, acknowledged_through)?)
    }

    /// Applies every physical extension row tied at `base_log_head`, then
    /// repairs the current base source only when the durable extension is an
    /// exact prefix with no later row already present.
    ///
    /// The repair row is applied to the owner here and STAGED, not written:
    /// the single replay pass reaches this point before it has audited the
    /// rest of the log, and a log refused later in that audit must leave the
    /// store exactly as it found it. Staging holds one page at most
    /// ([`Self::stage_repair`]).
    pub(super) async fn apply_boundary(
        &mut self,
        authority: &mut ConversationAuthority,
        base_log_head: u64,
        expected_projection: Option<&OutboxRow>,
    ) -> Result<(), RestoreError> {
        let mut projection_seen = false;
        loop {
            let Some((physical_sequence, row)) = self.cursor.front().await? else {
                break;
            };
            let boundary = row.base_log_head().ok_or_else(|| {
                RestoreError::Semantic(StateError::invariant(
                    "Unit 2 extension base boundary overflowed",
                ))
            })?;
            if boundary < base_log_head {
                return Err(RestoreError::Semantic(StateError::invariant(format!(
                    "Unit 2 extension boundary {boundary} is nonmonotone below {base_log_head}"
                ))));
            }
            if boundary > base_log_head {
                break;
            }
            let physical_sequence = *physical_sequence;
            let (_, row) = self.cursor.pop_front().ok_or_else(|| {
                RestoreError::Semantic(StateError::invariant(
                    "Unit 2 extension merge lost its physical head",
                ))
            })?;
            authority
                .begin_observer_progress_source()
                .map_err(RestoreError::Semantic)?;
            match &row {
                OutboxRow::Produced(_) | OutboxRow::AckAdvanced { .. } => {
                    if expected_projection != Some(&row) {
                        return Err(RestoreError::Semantic(StateError::invariant(format!(
                            "Unit 2 extension projection at physical sequence {physical_sequence} conflicts with v2 source"
                        ))));
                    }
                    projection_seen = true;
                }
                OutboxRow::MarkerAckCommitted(marker) => {
                    if marker.extension_sequence != physical_sequence {
                        return Err(RestoreError::Semantic(StateError::invariant(format!(
                            "MarkerAck extension sequence {} differs from physical {physical_sequence}",
                            marker.extension_sequence
                        ))));
                    }
                    authority
                        .replay_marker_ack_extension(marker)
                        .map_err(RestoreError::Semantic)?;
                }
            }
            self.outbox
                .apply_row(physical_sequence, row)
                .map_err(StateError::from)
                .map_err(RestoreError::Semantic)?;
            authority
                .end_observer_progress_source()
                .map_err(RestoreError::Semantic)?;
        }

        if let Some(expected) = expected_projection {
            if projection_seen {
                return Ok(());
            }
            if self.cursor.confirmed_head().is_none() {
                return Err(RestoreError::Semantic(StateError::invariant(
                    "Unit 2 extension is missing a projection before a later physical row",
                )));
            }
            let next_extension_sequence =
                *self
                    .repair_head
                    .get_or_insert(self.cursor.confirmed_head().ok_or_else(|| {
                        RestoreError::Semantic(StateError::invariant(
                            "Unit 2 extension repair attempted before confirmed EOF",
                        ))
                    })?);
            self.stage_repair(next_extension_sequence, expected.clone())
                .await?;
            authority.begin_observer_progress_source()?;
            self.outbox
                .apply_row(next_extension_sequence, expected.clone())
                .map_err(StateError::from)
                .map_err(RestoreError::Semantic)?;
            authority.end_observer_progress_source()?;
            self.repair_head = Some(next_extension_sequence.checked_add(1).ok_or_else(|| {
                RestoreError::Semantic(StateError::invariant("Unit 2 extension head overflowed"))
            })?);
        }
        Ok(())
    }

    /// Holds one owed repair row, never more than one page of them.
    ///
    /// Under [`RepairPaging::StageOnePage`] a row past the first page drops
    /// the page already staged and stops staging: a pass that owes more than
    /// one page writes nothing and is run again. Under
    /// [`RepairPaging::WriteEachPage`] a page is written the moment it fills.
    async fn stage_repair(
        &mut self,
        extension_sequence: u64,
        row: OutboxRow,
    ) -> Result<(), RestoreError> {
        match self.paging {
            RepairPaging::StageOnePage => {
                if self.repairs_overflowed {
                    return Ok(());
                }
                if self.staged_repairs.len() >= UNIT2_OUTBOX_RESTORE_BATCH_ROWS {
                    self.staged_repairs = Vec::new();
                    self.repairs_overflowed = true;
                    return Ok(());
                }
                self.staged_repairs.push((extension_sequence, row));
                #[cfg(test)]
                record_staged_repairs(self.staged_repairs.len());
            }
            RepairPaging::WriteEachPage => {
                self.staged_repairs.push((extension_sequence, row));
                #[cfg(test)]
                record_staged_repairs(self.staged_repairs.len());
                if self.staged_repairs.len() >= UNIT2_OUTBOX_RESTORE_BATCH_ROWS {
                    self.write_staged_page().await?;
                }
            }
        }
        Ok(())
    }

    /// Makes every staged repair row durable, in extension order, at the
    /// sequence it was applied under, as one page.
    ///
    /// The replay calls this once it knows the former validate-first pass
    /// would have reached the same repairs: after the whole log has audited
    /// clean. The rows, their order and their sequences are exactly the ones
    /// the in-line repair appended, so the durable result is byte-identical;
    /// only the moment moves, to after the audit. Repairs are contiguous from
    /// the extension's confirmed head, so the page of at most
    /// [`UNIT2_OUTBOX_RESTORE_BATCH_ROWS`] rows is one `append_all` and one
    /// flush (DUR-008 A3, D03), never one row and one flush at a time.
    ///
    /// A first pass that owed more than one page persists nothing and
    /// answers [`StagedRepairs::Overflowed`]; the replay then runs its second
    /// pass, which writes every page.
    pub(super) async fn persist_staged_repairs(&mut self) -> Result<StagedRepairs, RestoreError> {
        if self.repairs_overflowed {
            return Ok(StagedRepairs::Overflowed);
        }
        self.write_staged_page().await?;
        Ok(StagedRepairs::Persisted)
    }

    /// Writes the staged rows, at most one page, as one `append_all` and one
    /// flush, and empties the stage.
    async fn write_staged_page(&mut self) -> Result<(), RestoreError> {
        let staged = std::mem::take(&mut self.staged_repairs);
        let Some((first_sequence, _)) = staged.first() else {
            return Ok(());
        };
        let first_sequence = *first_sequence;
        let mut expected = first_sequence;
        let mut rows = Vec::with_capacity(staged.len());
        for (extension_sequence, row) in staged {
            if extension_sequence != expected {
                return Err(RestoreError::Semantic(StateError::invariant(format!(
                    "Unit 2 extension repair at sequence {extension_sequence} does not follow \
                     the repair at {}",
                    expected.saturating_sub(1)
                ))));
            }
            rows.push(row);
            expected = expected.checked_add(1).ok_or_else(|| {
                RestoreError::Semantic(StateError::invariant("Unit 2 extension head overflowed"))
            })?;
        }
        self.log
            .append_page(&rows, first_sequence)
            .await
            .map_err(ConversationOutboxError::from)
            .map_err(StateError::from)
            .map_err(RestoreError::Semantic)?;
        Ok(())
    }

    /// Refuses impossible future boundaries and installs the fully validated
    /// move-only owner only after the complete merge has succeeded.
    pub(super) fn finish(
        mut self,
        authority: &mut ConversationAuthority,
        final_base_log_head: u64,
    ) -> Result<(), StateError> {
        // An owner whose repairs never reached the store would answer from
        // rows a restart cannot see. Refused, by name, rather than installed.
        if let Some((extension_sequence, _)) = self.staged_repairs.first() {
            return Err(StateError::invariant(format!(
                "Unit 2 extension repair at sequence {extension_sequence} was staged but never \
                 persisted"
            )));
        }
        if self.repairs_overflowed {
            return Err(StateError::invariant(
                "Unit 2 extension repairs overflowed one staged page and were never persisted",
            ));
        }
        let leftover = self.leftover.or_else(|| {
            self.cursor
                .pop_front()
                .map(|(physical_sequence, row)| (physical_sequence, row.base_log_head()))
        });
        if let Some((physical_sequence, boundary)) = leftover {
            let boundary = boundary.ok_or_else(|| {
                StateError::invariant("Unit 2 extension base boundary overflowed")
            })?;
            return Err(StateError::invariant(format!(
                "Unit 2 extension physical sequence {physical_sequence} has impossible future boundary {boundary} above base head {final_base_log_head}"
            )));
        }
        if self.cursor.confirmed_head().is_none() {
            return Err(StateError::invariant(
                "Unit 2 extension merge finished before confirmed EOF",
            ));
        }
        authority.outbox = Some(self.outbox);
        Ok(())
    }
}

/// Frozen pre-W3 aggregate merge used only for byte/error equivalence oracles.
///
/// Its pending-history ownership and repair placement are deliberately
/// independent of [`ExtensionMerge`]; production cannot select this path.
#[cfg(test)]
pub(super) struct AggregateExtensionMerge<'a> {
    log: &'a OutboxLog,
    pending: VecDeque<(u64, OutboxRow)>,
    outbox: ConversationOutbox,
    next_extension_sequence: u64,
}

#[cfg(test)]
impl<'a> AggregateExtensionMerge<'a> {
    pub(super) fn new(
        log: &'a OutboxLog,
        rows: Vec<(u64, OutboxRow)>,
        conversation_id: u64,
        limits: ConversationOutboxLimits,
    ) -> Result<Self, StateError> {
        let next_extension_sequence = u64::try_from(rows.len())
            .map_err(|_| StateError::invariant("Unit 2 extension stream length exceeds u64"))?;
        Ok(Self {
            log,
            pending: rows.into(),
            outbox: ConversationOutbox::restore(conversation_id, Vec::new(), limits)?,
            next_extension_sequence,
        })
    }

    pub(super) fn recipient_ack_obligations(
        &self,
        participant_id: u64,
        acknowledged_through: u64,
    ) -> Result<(RecipientAckObligations, u64), StateError> {
        Ok(self
            .outbox
            .recipient_ack_obligations(participant_id, acknowledged_through)?)
    }

    pub(super) async fn apply_boundary(
        &mut self,
        authority: &mut ConversationAuthority,
        base_log_head: u64,
        expected_projection: Option<&OutboxRow>,
    ) -> Result<(), StateError> {
        let mut projection_seen = false;
        while let Some((physical_sequence, row)) = self.pending.front() {
            let boundary = row.base_log_head().ok_or_else(|| {
                StateError::invariant("Unit 2 extension base boundary overflowed")
            })?;
            if boundary < base_log_head {
                return Err(StateError::invariant(format!(
                    "Unit 2 extension boundary {boundary} is nonmonotone below {base_log_head}"
                )));
            }
            if boundary > base_log_head {
                break;
            }
            let physical_sequence = *physical_sequence;
            let (_, row) = self.pending.pop_front().ok_or_else(|| {
                StateError::invariant("Unit 2 extension merge lost its physical head")
            })?;
            authority.begin_observer_progress_source()?;
            match &row {
                OutboxRow::Produced(_) | OutboxRow::AckAdvanced { .. } => {
                    if expected_projection != Some(&row) {
                        return Err(StateError::invariant(format!(
                            "Unit 2 extension projection at physical sequence {physical_sequence} conflicts with v2 source"
                        )));
                    }
                    projection_seen = true;
                }
                OutboxRow::MarkerAckCommitted(marker) => {
                    if marker.extension_sequence != physical_sequence {
                        return Err(StateError::invariant(format!(
                            "MarkerAck extension sequence {} differs from physical {physical_sequence}",
                            marker.extension_sequence
                        )));
                    }
                    authority.replay_marker_ack_extension(marker)?;
                }
            }
            self.outbox.apply_row(physical_sequence, row)?;
            authority.end_observer_progress_source()?;
        }

        if let Some(expected) = expected_projection {
            if projection_seen {
                return Ok(());
            }
            if !self.pending.is_empty() {
                return Err(StateError::invariant(
                    "Unit 2 extension is missing a projection before a later physical row",
                ));
            }
            self.log
                .append(expected, self.next_extension_sequence)
                .await
                .map_err(ConversationOutboxError::from)?;
            authority.begin_observer_progress_source()?;
            self.outbox
                .apply_row(self.next_extension_sequence, expected.clone())?;
            authority.end_observer_progress_source()?;
            self.next_extension_sequence = self
                .next_extension_sequence
                .checked_add(1)
                .ok_or_else(|| StateError::invariant("Unit 2 extension head overflowed"))?;
        }
        Ok(())
    }

    pub(super) fn finish(
        self,
        authority: &mut ConversationAuthority,
        final_base_log_head: u64,
    ) -> Result<(), StateError> {
        if let Some((physical_sequence, row)) = self.pending.front() {
            let boundary = row.base_log_head().ok_or_else(|| {
                StateError::invariant("Unit 2 extension base boundary overflowed")
            })?;
            return Err(StateError::invariant(format!(
                "Unit 2 extension physical sequence {physical_sequence} has impossible future boundary {boundary} above base head {final_base_log_head}"
            )));
        }
        authority.outbox = Some(self.outbox);
        Ok(())
    }
}