liminal-server 0.24.5

//! DUR-020: the server listens before it restores any conversation.
//!
//! Construction reads the registry and records every registered conversation
//! as pending. Nothing else is read before the listener binds, so the time to
//! listening does not grow with the number of conversations or with their
//! history. A pending conversation is restored by its first touch, whatever
//! path touches it, through the one restore unit boot used to run for it
//! ([`ProductionParticipantHandler::restore_one_conversation`]: checkpoint or
//! rebuild, fate repair, observer reconciliation, capacity fold and the
//! R-BOOT-DRAIN), and then by the unclean-restart repair of this incarnation.
//! The unit runs at most once per conversation per process: a gate per
//! pending conversation serialises the touches that race on it, and the
//! conversation stops being pending when its unit has run. The background
//! warm ([`ProductionParticipantHandler::restore_remaining`]) touches every
//! conversation still pending, so a conversation nobody asks for is restored
//! and delivers what it owes.

use std::cell::Cell;
use std::sync::atomic::Ordering;
use std::time::Instant;

use liminal_protocol::wire::ConversationId;

use crate::server::participant::{ConnectionFateWorkItem, ParticipantSemanticError};

use super::handler::{ProductionParticipantHandler, elapsed_ms};
#[path = "pending_restores.rs"]
mod pending;
use super::state::StateError;
pub(super) use pending::PendingRestores;
use pending::{Gate, GateState, poisoned_refusal};

thread_local! {
    /// The conversation whose restore unit this thread is running. The unit
    /// reaches its own conversation's cell through the ordinary accessors,
    /// which must not wait on the gate the unit itself holds.
    static RESTORING: Cell<Option<ConversationId>> = const { Cell::new(None) };
}

/// Marks this thread as running one conversation's restore unit, restoring
/// the previous mark on drop, a refusal included.
struct RestoringMark {
    previous: Option<ConversationId>,
}

impl RestoringMark {
    fn enter(conversation_id: ConversationId) -> Self {
        Self {
            previous: RESTORING.with(|current| current.replace(Some(conversation_id))),
        }
    }
}

impl Drop for RestoringMark {
    fn drop(&mut self) {
        RESTORING.with(|current| current.set(self.previous));
    }
}

impl ProductionParticipantHandler {
    /// Runs a pending conversation's restore unit, once, before its caller
    /// reads or locks the conversation's cell.
    ///
    /// A conversation that is not pending (restored already, never
    /// registered, or the one this thread is restoring) returns at once. A
    /// unit that refuses records the conversation unloadable, as boot
    /// recorded it, and answers the toucher with that named refusal; the
    /// conversation is no longer pending, so a later touch loads it through
    /// the ordinary request path, exactly as a touch after a refused boot
    /// restore did.
    pub(super) fn ensure_restored(
        &self,
        conversation_id: ConversationId,
    ) -> Result<(), ParticipantSemanticError> {
        if RESTORING.with(Cell::get) == Some(conversation_id) {
            return Ok(());
        }
        let Some(gate) = self.pending.gate(conversation_id)? else {
            // DUR-024 R3: a conversation contained as unloadable in this
            // process stays contained for the life of the process. Its one
            // restore attempt ran (and was logged, with its elapsed time);
            // every later touch, whatever touches (a connection-fate open, a
            // seat warm, a post), is answered by the containment at once and
            // never starts another authoritative replay. Only a restart tries
            // again: the door's room of 10 October was re-opened by recovery
            // after its refusal and replayed a second 31 minutes with the
            // ports unbound.
            match self.unloadable_record().retained(conversation_id) {
                Ok(Some(contained)) => {
                    return Err(ParticipantSemanticError::ConversationUnloadable {
                        conversation_id,
                        reason: contained.reason,
                    });
                }
                Ok(None) => {}
                // Fail closed: a containment that cannot be read must not
                // start the replay it exists to prevent.
                Err(poisoned) => {
                    return Err(ParticipantSemanticError::ConversationUnloadable {
                        conversation_id,
                        reason: poisoned.to_string(),
                    });
                }
            }
            return self.repair_refused(conversation_id);
        };
        if !self.claim_gate(&gate, conversation_id)? {
            return Ok(());
        }
        let mark = RestoringMark::enter(conversation_id);
        let started = Instant::now();
        // ONE FALLIBLE UNIT, not a guard per `?`: whatever refuses inside
        // it, the refusal is this conversation's and nobody else's.
        #[cfg(test)]
        let restored = crate::boot_history_harness::during_stage("authority", || {
            self.restore_one_conversation(conversation_id)
        });
        #[cfg(not(test))]
        let restored = self.restore_one_conversation(conversation_id);
        let outcome = restored.and_then(|(rows, still_owed)| {
            // The owed checkpoint is published from the restored owner first,
            // before a held fate opens a pending finalization that leaves the
            // owner uncapturable until it settles (the order every boot had
            // while recovery applied the fate after the unit).
            let still_owed = still_owed && !self.publish_owed_checkpoint(conversation_id)?;
            for work_item in self.pending.take_deferred(conversation_id)? {
                self.apply_deferred_fate(conversation_id, &work_item)?;
            }
            let repaired = self.pending.owed_repair()?.map_or(Ok(0), |incarnation| {
                self.repair_one_unclean_restart(conversation_id, incarnation)
            })?;
            // R-BOOT-DRAIN: a terminal the repair pended is drained by this
            // unit, never left on the lane for the next boot.
            if repaired > 0 {
                self.drain_after_repair(conversation_id)?;
            }
            // DUR-020 R6: an owner the restore could not capture publishes
            // once its repair has finished what refused the capture.
            if still_owed {
                self.publish_owed_checkpoint(conversation_id)?;
            }
            Ok((rows, repaired))
        });
        // The mark is dropped only now: the repair above reaches this
        // conversation's cell through `cell()`, which must not wait on the
        // gate this thread holds.
        drop(mark);
        let answer = match outcome {
            Ok((rows, repaired)) => {
                tracing::info!(
                    step = "participant restore",
                    conversation_id,
                    log_rows = rows,
                    repaired_rows = repaired,
                    conversation_elapsed_ms = elapsed_ms(started),
                    "participant conversation restored"
                );
                Ok(())
            }
            Err(error) => {
                let refusal = self.record_unloadable(conversation_id, &error);
                tracing::warn!(
                    step = "participant restore",
                    conversation_id,
                    refusal = %refusal,
                    conversation_elapsed_ms = elapsed_ms(started),
                    "participant conversation refused at restore; it stays contained until the process restarts"
                );
                if let Err(owed) = self.pending.refuse(conversation_id) {
                    tracing::error!(
                        conversation_id,
                        error = %owed,
                        "a refused conversation's owed repair could not be recorded"
                    );
                }
                Err(refusal)
            }
        };
        // The answer is recorded before the gate is released, so a poisoned
        // pending lock can never drop it.
        if let Err(error) = self.pending.finish(conversation_id) {
            tracing::error!(
                conversation_id,
                error = %error,
                "a restored conversation could not leave the pending set"
            );
        }
        // Done only now, so a waiter on this gate finds the outcome recorded
        // and never runs the unit a second time.
        gate.done()?;
        answer
    }

    /// Takes the gate to run the unit: `true` when this thread runs it,
    /// `false` when it has run. A unit another thread is running is waited
    /// on, except during a server stop, which does not wait for another
    /// thread's unit: that unit runs on, and this touch is answered at once.
    fn claim_gate(
        &self,
        gate: &Gate,
        conversation_id: ConversationId,
    ) -> Result<bool, ParticipantSemanticError> {
        let mut state = gate.state()?;
        loop {
            match *state {
                GateState::Done => return Ok(false),
                GateState::Pending => break,
                GateState::Running if self.warm_stop.load(Ordering::SeqCst) => {
                    return Err(ParticipantSemanticError::Unavailable);
                }
                GateState::Running => {
                    state = gate.changed.wait(state).map_err(|_| {
                        ParticipantSemanticError::Internal {
                            message: format!(
                                "participant conversation {conversation_id} restore gate is poisoned"
                            ),
                        }
                    })?;
                }
            }
        }
        *state = GateState::Running;
        drop(state);
        Ok(true)
    }

    /// The first touch of a conversation whose restore unit refused runs the
    /// unclean-restart repair it still owes, once it is owed, before the
    /// touch loads it through the ordinary request path.
    fn repair_refused(
        &self,
        conversation_id: ConversationId,
    ) -> Result<(), ParticipantSemanticError> {
        let Some(owed) = self.pending.refused_gate(conversation_id)? else {
            return Ok(());
        };
        let mut owed = owed.lock().map_err(|_| poisoned_refusal())?;
        let Some(incarnation) = *owed else {
            return Ok(());
        };
        let mark = RestoringMark::enter(conversation_id);
        let repaired = self.repair_one_unclean_restart(conversation_id, incarnation);
        drop(mark);
        if repaired.is_ok() {
            *owed = None;
        }
        drop(owed);
        match repaired {
            Ok(_) => self.pending.clear_refused(conversation_id),
            Err(error) => Err(self.record_unloadable(conversation_id, &error)),
        }
    }

    /// Restores every conversation still pending, in id order, and answers
    /// how many refused. This is the background warm, and the one place a
    /// server-scope capacity decision waits for, because the capacity ledger
    /// is exact only once every conversation has folded its contribution.
    ///
    /// # Errors
    ///
    /// Only a poisoned pending-restore lock; a conversation that refuses is
    /// recorded unloadable and counted, never fatal to the others.
    pub(super) fn restore_remaining(
        &self,
        reason: &'static str,
    ) -> Result<usize, ParticipantSemanticError> {
        let remaining = self.pending.remaining()?;
        let total = remaining.len();
        let started = Instant::now();
        if total > 0 {
            tracing::info!(
                step = "participant warm",
                conversations = total,
                reason,
                "participant warm finishing in the caller: it waits for every pending conversation"
            );
        }
        let mut refused = Vec::new();
        for (index, conversation_id) in remaining.into_iter().enumerate() {
            if let Err(refusal) = self.ensure_restored(conversation_id) {
                refused.push((conversation_id, refusal.to_string()));
            }
            tracing::debug!(
                step = "participant warm",
                conversation_id,
                done = index.saturating_add(1),
                total,
                elapsed_ms = elapsed_ms(started),
                "participant warm progress"
            );
        }
        if total > 0 {
            tracing::info!(
                step = "participant warm",
                conversations = total,
                unloadable = refused.len(),
                refused = ?refused,
                elapsed_ms = elapsed_ms(started),
                "participant warm finished"
            );
        }
        Ok(refused.len())
    }

    /// Runs one conversation's unclean-restart repair for `incarnation`, as
    /// a fate-source operation, and answers the rows it appended.
    /// Applies one connection fate held while the conversation was pending
    /// (DUR-024), inside its restore unit, exactly as boot recovery would
    /// have applied it. The impacts are not dispatched: no connection of
    /// this process is bound to a conversation that was not yet restored.
    fn apply_deferred_fate(
        &self,
        conversation_id: ConversationId,
        work_item: &ConnectionFateWorkItem,
    ) -> Result<(), ParticipantSemanticError> {
        let mut impact =
            crate::server::participant::dispatch_impact::DispatchImpactAccumulator::new();
        self.with_conversation_fate_source(
            conversation_id,
            Some(&mut impact),
            |authority, appender, impact| {
                let impact = impact.ok_or_else(|| {
                    StateError::invariant("deferred connection fate lost its impact owner")
                })?;
                let transaction = authority.prepare_connection_fate_transaction(work_item);
                transaction.complete_with_impact(authority, appender, impact)
            },
        )?;
        tracing::info!(
            step = "participant restore",
            conversation_id,
            open_sequence = work_item.open_sequence,
            "a connection fate held at boot was applied in the conversation's restore"
        );
        Ok(())
    }

    pub(super) fn repair_one_unclean_restart(
        &self,
        conversation_id: ConversationId,
        incarnation: u64,
    ) -> Result<u64, ParticipantSemanticError> {
        self.with_conversation_fate_source(conversation_id, None, |authority, appender, impact| {
            if impact.is_some() {
                return Err(StateError::invariant(
                    "startup connection-fate repair gained a dispatch impact owner",
                ));
            }
            let head = authority.next_log_sequence;
            let transaction = authority.prepare_unclean_server_restart_transaction(incarnation)?;
            transaction.complete(authority, appender)?;
            Ok(authority.next_log_sequence.saturating_sub(head))
        })
    }

    /// The conversations this process holds a restored owner for, in id order.
    pub(super) fn restored_conversation_ids(
        &self,
    ) -> Result<Vec<ConversationId>, ParticipantSemanticError> {
        let mut restored = Vec::new();
        for (conversation_id, cell) in self.conversation_cells()? {
            let owner = cell
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: format!(
                        "participant conversation {conversation_id} owner lock is poisoned"
                    ),
                })?;
            if owner.is_some() {
                restored.push(conversation_id);
            }
        }
        restored.sort_unstable();
        Ok(restored)
    }
}

/// What one conversation's history check covered: every identity its
/// checkpoint held, and how many of them were read from the index to be
/// compared (the rest were verified at an earlier root and are unchanged).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) struct HistoryCheckRead {
    pub(super) checked: usize,
    pub(super) identities: usize,
    /// The value bytes written to the verified record.
    pub(super) recorded_bytes: usize,
}

impl ProductionParticipantHandler {
    /// The identities a restored conversation's checkpoint body held,
    /// captured at its restore, against the history index at that
    /// checkpoint's root. They are taken from the owner under its lock; the
    /// index reads run after it is released, so requests are not held.
    pub(super) fn verify_identity_history(
        &self,
        conversation_id: ConversationId,
    ) -> Result<HistoryCheckRead, ParticipantSemanticError> {
        let Some(cell) = self.existing_cell(conversation_id)? else {
            return Ok(HistoryCheckRead::default());
        };
        let mut owner = cell
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: format!(
                    "participant conversation {conversation_id} owner lock is poisoned"
                ),
            })?;
        let check = owner
            .as_mut()
            .and_then(|authority| authority.checkpoint_history_check.take());
        drop(owner);
        let Some(check) = check else {
            return Ok(HistoryCheckRead::default());
        };
        let check = super::restore_checkpoint_history_capture::validate_identity_history(
            self.store.as_ref(),
            check,
        );
        let verified = liminal::durability::bridge::block_on(check)
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("participant history check could not run: {error}"),
            })?
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("participant history check refused: {error}"),
            })?;
        // Every identity matched the index at this root, so the owner may
        // adopt the digests, unless a crossing has moved its root since.
        let mut owner = cell
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: format!(
                    "participant conversation {conversation_id} owner lock is poisoned"
                ),
            })?;
        let read = HistoryCheckRead {
            checked: verified.checked,
            identities: verified.identities,
            recorded_bytes: verified.recorded_bytes,
        };
        if let Some(authority) = owner.as_mut()
            && authority.checkpoint_history_root == Some(verified.digests.root())
            && authority.history_digests.is_none()
        {
            authority.history_digests = Some(verified.digests);
        }
        drop(owner);
        Ok(read)
    }
}