liminal-server 0.21.0

//! 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::collections::HashMap;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Condvar, Mutex, MutexGuard, Weak};
use std::time::Instant;

use liminal_protocol::wire::ConversationId;

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

use super::handler::{ProductionParticipantHandler, elapsed_ms};
use super::state::StateError;

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));
    }
}

/// The registered conversations this process has not restored yet, each with
/// the gate its first touch takes, and the server incarnation whose
/// unclean-restart repair a restore owes.
#[derive(Debug, Default)]
pub(super) struct PendingRestores {
    gates: Mutex<GateMap>,
    repair_incarnation: Mutex<Option<u64>>,
    /// Conversations whose restore unit refused, each with the
    /// unclean-restart repair it still owes (`None` until startup owes one).
    /// A refused conversation loads later through the ordinary request path,
    /// which runs no repair, so its next touch runs the owed repair first.
    refused: Mutex<RefusedMap>,
    /// Connection fates met while their conversation was still pending
    /// (DUR-024, 11 Oct 2026): boot's connection-fate recovery no longer
    /// restores a conversation to apply a fate, which made the door's room
    /// replay for 30 minutes before the ports bound. The fate is applied
    /// inside the conversation's own restore unit, first, before the
    /// unclean-restart repair, with its class preserved.
    deferred_fates: Mutex<HashMap<ConversationId, Vec<ConnectionFateWorkItem>>>,
}

/// Each pending conversation's gate.
type GateMap = HashMap<ConversationId, Arc<Gate>>;

/// Where one pending conversation's restore unit stands. The lock is held
/// only to read or move the state, never across the unit, so a server stop
/// can wake every touch waiting on a unit that is still running (Gaia, 11
/// October 2026: the door's stop waited 60 s on the room's rebuild).
#[derive(Debug, Default)]
struct Gate {
    state: Mutex<GateState>,
    changed: Condvar,
}

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
enum GateState {
    /// Nobody has started the unit.
    #[default]
    Pending,
    /// One thread is running the unit.
    Running,
    /// The unit has run, restored or refused.
    Done,
}

impl Gate {
    fn state(&self) -> Result<MutexGuard<'_, GateState>, ParticipantSemanticError> {
        self.state
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant restore gate is poisoned".to_owned(),
            })
    }

    /// Records the unit as run and wakes every touch waiting on it.
    fn done(&self) -> Result<(), ParticipantSemanticError> {
        *self.state()? = GateState::Done;
        self.changed.notify_all();
        Ok(())
    }
}

/// Each refused conversation's owed repair incarnation, behind its own lock.
type RefusedMap = HashMap<ConversationId, Arc<Mutex<Option<u64>>>>;

impl PendingRestores {
    fn gates(&self) -> Result<MutexGuard<'_, GateMap>, ParticipantSemanticError> {
        self.gates
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant pending-restore lock is poisoned".to_owned(),
            })
    }

    /// Records the registered conversations as pending; answers how many.
    pub(super) fn register(
        &self,
        conversation_ids: impl IntoIterator<Item = ConversationId>,
    ) -> Result<usize, ParticipantSemanticError> {
        let mut gates = self.gates()?;
        for conversation_id in conversation_ids {
            gates.entry(conversation_id).or_default();
        }
        Ok(gates.len())
    }

    fn gate(
        &self,
        conversation_id: ConversationId,
    ) -> Result<Option<Arc<Gate>>, ParticipantSemanticError> {
        Ok(self.gates()?.get(&conversation_id).map(Arc::clone))
    }

    /// Wakes every touch waiting on a running unit, for a server stop: each
    /// rechecks the stop and leaves without waiting for the unit.
    pub(super) fn wake_waiters(&self) -> Result<(), ParticipantSemanticError> {
        let gates: Vec<Arc<Gate>> = self.gates()?.values().map(Arc::clone).collect();
        for gate in gates {
            // Taken so a waiter between its stop check and its wait cannot
            // miss the wake.
            let state = gate.state()?;
            gate.changed.notify_all();
            drop(state);
        }
        Ok(())
    }

    fn deferred(
        &self,
    ) -> Result<
        MutexGuard<'_, HashMap<ConversationId, Vec<ConnectionFateWorkItem>>>,
        ParticipantSemanticError,
    > {
        self.deferred_fates
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant deferred-fate lock is poisoned".to_owned(),
            })
    }

    /// Holds a connection fate for a conversation that is still pending and
    /// whose restore is not running: `true` when held, so the caller does
    /// not restore it now. The gate's state is held while the fate is
    /// recorded, so a restore that starts after this finds it, and a restore
    /// already running is not waited on here: the caller applies the fate
    /// through the ordinary path instead.
    pub(super) fn defer_fate_if_pending(
        &self,
        conversation_id: ConversationId,
        work_item: &ConnectionFateWorkItem,
    ) -> Result<bool, ParticipantSemanticError> {
        let Some(gate) = self.gate(conversation_id)? else {
            return Ok(false);
        };
        let state = gate.state()?;
        if *state != GateState::Pending {
            return Ok(false);
        }
        self.deferred()?
            .entry(conversation_id)
            .or_default()
            .push(work_item.clone());
        drop(state);
        Ok(true)
    }

    /// The fates held for a conversation, taken by its restore unit.
    fn take_deferred(
        &self,
        conversation_id: ConversationId,
    ) -> Result<Vec<ConnectionFateWorkItem>, ParticipantSemanticError> {
        Ok(self
            .deferred()?
            .remove(&conversation_id)
            .unwrap_or_default())
    }

    fn finish(&self, conversation_id: ConversationId) -> Result<(), ParticipantSemanticError> {
        self.gates()?.remove(&conversation_id);
        Ok(())
    }

    /// The conversations still pending, in ascending id order.
    pub(super) fn remaining(&self) -> Result<Vec<ConversationId>, ParticipantSemanticError> {
        let mut remaining: Vec<ConversationId> = self.gates()?.keys().copied().collect();
        remaining.sort_unstable();
        Ok(remaining)
    }

    /// Records the incarnation whose unclean-restart repair every restore
    /// from now on owes, and every conversation already refused owes too.
    pub(super) fn owe_repair(&self, incarnation: u64) -> Result<(), ParticipantSemanticError> {
        *self.repair_incarnation()? = Some(incarnation);
        for owed in self.refused()?.values() {
            let mut owed = owed.lock().map_err(|_| poisoned_refusal())?;
            if owed.is_none() {
                *owed = Some(incarnation);
            }
        }
        Ok(())
    }

    fn refused(&self) -> Result<MutexGuard<'_, RefusedMap>, ParticipantSemanticError> {
        self.refused.lock().map_err(|_| poisoned_refusal())
    }

    /// Records a refused restore with the repair it owes now, if any.
    fn refuse(&self, conversation_id: ConversationId) -> Result<(), ParticipantSemanticError> {
        let owed = self.owed_repair()?;
        self.refused()?
            .entry(conversation_id)
            .or_insert_with(|| Arc::new(Mutex::new(owed)));
        Ok(())
    }

    fn refused_gate(
        &self,
        conversation_id: ConversationId,
    ) -> Result<Option<Arc<Mutex<Option<u64>>>>, ParticipantSemanticError> {
        Ok(self.refused()?.get(&conversation_id).map(Arc::clone))
    }

    fn clear_refused(
        &self,
        conversation_id: ConversationId,
    ) -> Result<(), ParticipantSemanticError> {
        self.refused()?.remove(&conversation_id);
        Ok(())
    }

    fn repair_incarnation(&self) -> Result<MutexGuard<'_, Option<u64>>, ParticipantSemanticError> {
        self.repair_incarnation
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant repair-incarnation lock is poisoned".to_owned(),
            })
    }

    fn owed_repair(&self) -> Result<Option<u64>, ParticipantSemanticError> {
        Ok(*self.repair_incarnation()?)
    }
}

fn poisoned_refusal() -> ParticipantSemanticError {
    ParticipantSemanticError::Internal {
        message: "participant refused-restore lock is poisoned".to_owned(),
    }
}

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

/// The server's hold on the participant warm: stopping it ends the warm at
/// its next conversation boundary, without waiting (DUR-020 R5).
#[derive(Clone, Debug)]
pub(in crate::server) struct ParticipantWarm {
    handler: Weak<ProductionParticipantHandler>,
}

impl ParticipantWarm {
    /// Stops the warm. A conversation already restoring finishes its unit,
    /// and every touch waiting on that unit is woken and answered at once.
    pub(in crate::server) fn stop(&self) {
        if let Some(handler) = self.handler.upgrade() {
            handler.warm_stop.store(true, Ordering::SeqCst);
            if let Err(error) = handler.pending.wake_waiters() {
                tracing::error!(
                    error = %error,
                    "a server stop could not wake the touches waiting on a restore"
                );
            }
        }
    }

    /// Finishes the warm on the caller's thread and answers how many
    /// conversations refused. Each conversation restores once behind its own
    /// gate, so the store work is the same however the warm thread and this
    /// call split it; a measurement that must not race the warm calls this
    /// before it reads.
    ///
    /// # Errors
    ///
    /// Only a poisoned pending-restore lock.
    pub(in crate::server) fn finish(&self) -> Result<usize, ParticipantSemanticError> {
        self.handler.upgrade().map_or(Ok(0), |handler| {
            handler.restore_remaining("a measurement waiting for the warm")
        })
    }

    /// The clean stop's capacity snapshot (DUR-020 R3), written after the
    /// connections have closed and before the final flush.
    pub(in crate::server) fn record_clean_stop(&self) {
        if let Some(handler) = self.handler.upgrade() {
            handler.record_clean_stop_capacity();
        }
    }
}

impl ProductionParticipantHandler {
    /// Gives the handler its own weak reference, which the warm thread
    /// holds, and answers the server's stop handle for that warm.
    ///
    /// # Errors
    ///
    /// Refuses a second installation by name.
    pub(in crate::server) fn install_warm(
        self: &Arc<Self>,
    ) -> Result<ParticipantWarm, ParticipantSemanticError> {
        self.self_ref.set(Arc::downgrade(self)).map_err(|_| {
            ParticipantSemanticError::Internal {
                message: "participant warm was already installed".to_owned(),
            }
        })?;
        Ok(ParticipantWarm {
            handler: Arc::downgrade(self),
        })
    }

    /// Starts the warm on its own thread, once the incarnation's repair is
    /// owed, so every conversation it restores is repaired by its own unit.
    /// The listener never waits for it. A handler with no installed warm (a
    /// test that builds the handler alone) starts none: its conversations
    /// restore on their first touch.
    pub(super) fn start_warm(&self) {
        let Some(handler) = self.self_ref.get().cloned() else {
            return;
        };
        let spawned = std::thread::Builder::new()
            .name("liminal-participant-warm".to_owned())
            .spawn(move || warm_in_background(&handler));
        if let Err(error) = spawned {
            // Never silent, and never fatal: without the warm every
            // conversation still restores on its first touch, and enrollment
            // still finishes the warm before its server-scope decision.
            tracing::error!(
                error = %error,
                "participant warm thread could not start; conversations restore on first touch"
            );
        }
    }
}

/// The warm loop: restores each pending conversation in id order, holding the
/// handler only for one conversation at a time, until none is pending, the
/// server stops it, or the server drops the handler.
fn warm_in_background(handler: &Weak<ProductionParticipantHandler>) {
    let Some(owner) = handler.upgrade() else {
        return;
    };
    let remaining = match owner.pending.remaining() {
        Ok(remaining) => remaining,
        Err(error) => {
            tracing::error!(error = %error, "participant warm could not list pending conversations");
            return;
        }
    };
    drop(owner);
    let total = remaining.len();
    let started = Instant::now();
    tracing::info!(
        step = "participant warm",
        conversations = total,
        "participant warm started"
    );
    let mut refused = Vec::new();
    for (index, conversation_id) in remaining.into_iter().enumerate() {
        let Some(owner) = handler.upgrade() else {
            return;
        };
        if owner.warm_stop.load(Ordering::SeqCst) {
            tracing::info!(
                step = "participant warm",
                done = index,
                total,
                elapsed_ms = elapsed_ms(started),
                "participant warm stopped with the server; the rest restore on the next start"
            );
            return;
        }
        if let Err(refusal) = owner.ensure_restored(conversation_id) {
            refused.push((conversation_id, refusal.to_string()));
        }
    }
    tracing::info!(
        step = "participant warm",
        conversations = total,
        unloadable = refused.len(),
        refused = ?refused,
        elapsed_ms = elapsed_ms(started),
        "participant warm finished"
    );
    verify_restored_histories(handler);
}

/// After the warm: cross-checks every restored conversation's identities
/// against its checkpoint's history index, the check the restore leaves off
/// its path so that it reads the same at any history length (Waffles, 11
/// October 09:14: nothing is skipped, only moved). A mismatch is logged by
/// name and counted, and the conversation keeps being served. A root
/// verified on an earlier boot is recorded, and only what changed since it
/// is read (Waffles, 12 October 07:46: the full check re-read 50,011
/// identities in 164 s on every boot, growing with the history). The
/// finished line names how many identities were covered, how many were read
/// from the index, and which conversations disagreed.
fn verify_restored_histories(handler: &Weak<ProductionParticipantHandler>) {
    let Some(owner) = handler.upgrade() else {
        return;
    };
    let restored = match owner.restored_conversation_ids() {
        Ok(restored) => restored,
        Err(error) => {
            tracing::error!(error = %error, "participant history check could not list restored conversations");
            return;
        }
    };
    drop(owner);
    let started = Instant::now();
    let mut identities = 0_usize;
    let mut read = 0_usize;
    let mut mismatched = Vec::new();
    for conversation_id in restored {
        let Some(owner) = handler.upgrade() else {
            return;
        };
        if owner.warm_stop.load(Ordering::SeqCst) {
            return;
        }
        let checked = match owner.verify_identity_history(conversation_id) {
            Ok(checked) => checked,
            Err(error) => {
                // Logged by name and counted, never contained (Waffles' ruling
                // of 11 October 09:14): on 11 October a705852d contained the
                // live room on this check while the restore itself was sound.
                tracing::error!(
                    step = "participant history check",
                    conversation_id = ?conversation_id,
                    error = %error,
                    "a restored conversation's identities disagree with its checkpoint's history index"
                );
                mismatched.push(conversation_id);
                continue;
            }
        };
        identities = identities.saturating_add(checked.identities);
        read = read.saturating_add(checked.checked);
    }
    tracing::info!(
        step = "participant history check",
        identities,
        read,
        mismatches = mismatched.len(),
        mismatched = ?mismatched,
        elapsed_ms = elapsed_ms(started),
        "history cross-check finished"
    );
}

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