liminal-server 0.23.1

//! Boot-recovery candidate-lane drain (F8B `R-BOOT-DRAIN`, `R-BOOT-VERDICT`).
//!
//! `docs/design/F8B-INTENT-DEADLOCK.md` §6.2. Boot recovery empties each
//! restored conversation's immutable-candidate lane BEFORE any retained
//! connection-fate `Open` replays, through the same `persist_drain_first`
//! machinery a live publish uses. Without it a store whose lane rests on a
//! crash-restored pending binding terminal deadlocks: the replayed `Open`
//! must admit a terminal of its own, the occupied lane refuses it
//! (`Precedence`), and nothing in the boot path can clear the occupant.
//!
//! The drain is BOOT-ONLY by construction. It is called from
//! `restore_all_conversations`, never from `replay_and_repair`, which is also
//! the load a request runs for a conversation with no live owner.
//!
//! Each drained row writes its own Unit 2 extension row in place, exactly as a
//! live commit does, so the restored owner stays its log's current state and
//! boot never reads the log back after a drain.
//!
//! Every attempt ends in a NAMED verdict, never a bare `?` — §6.2
//! R-BOOT-VERDICT — which is why the drain returns [`BootDrainVerdict`]
//! rather than a `Result`. `Drained` and `AlreadyEmpty` proceed to replay;
//! the two refusals refuse the boot loudly, naming the conversation, the
//! candidate shape and this design document, so an operator can tell them
//! from the deadlock they replace.

use std::cell::RefCell;
use std::sync::Arc;

use liminal_protocol::lifecycle::ImmutableSequenceCandidate;
use liminal_protocol::wire::ConversationId;

use crate::server::participant::{
    BootDrainRefusal, ParticipantSemanticError, dispatch_impact::DispatchImpactAccumulator,
};

use super::handler::{LogAppender, ProductionParticipantHandler};
use super::log::OperationLog;
use super::operation_batch::OperationBatch;
use super::outbox_log::OutboxLog;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::state::ConversationAuthority;

/// Named outcome of one restored conversation's boot drain attempt (§6.2
/// R-BOOT-VERDICT). The two refusals carry the exact lane head that refused
/// them, so the boot's own error can name the shape it could not repair.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum BootDrainVerdict {
    /// The lane held candidates and every one of them drained. N markers need
    /// N drains, so `drains` counts heads removed, not conversations.
    Drained {
        /// Heads removed from this conversation's lane by this boot.
        drains: usize,
        /// F8B R-SEAL (§6.6): whether one of those drains erased the
        /// conversation's final enrollment token and closed it. Sealing is a
        /// PROPERTY of a successful drain, never a fifth verdict — a verdict
        /// enum that split them would let a log reader believe a seal happened
        /// without a drain.
        sealed: bool,
    },
    /// The lane was already empty when boot reached it.
    AlreadyEmpty,
    /// §5.3(ii): the lane head is a pending binding terminal under an armed
    /// fenced-attach recovery block. `validate_first_terminal_candidate`
    /// refuses the drain outright while `recovery.is_some()`, and the only
    /// consumer of a recovery block is a live fenced attach that boot cannot
    /// perform — so this store is NOT repairable by the boot drain. The
    /// verdict is the honest answer, not a repair.
    RefusedRecoveryArmed {
        /// Conversation whose lane refused.
        conversation_id: ConversationId,
        /// Exact lane head that refused.
        candidate: ImmutableSequenceCandidate,
    },
    /// Any other drain refusal.
    RefusedShape {
        /// Conversation whose lane refused.
        conversation_id: ConversationId,
        /// Exact lane head that refused.
        candidate: ImmutableSequenceCandidate,
        /// The drain's own refusal, rendered for the operator.
        reason: String,
    },
}

impl BootDrainVerdict {
    /// Whether this boot appended drain rows, so the caller must run the
    /// commit tail (observer reconcile, capacity fold) over the restored owner.
    pub(super) const fn drained_any(&self) -> bool {
        matches!(*self, Self::Drained { .. })
    }

    /// Emits the verdict and converts the two refusals into the boot's own
    /// refusal.
    ///
    /// The refusal travels TYPED — [`BootDrainRefusal`] — so a consumer
    /// discriminates recovery-armed from every other shape without reading a
    /// formatted message, exactly as the foundation leg's
    /// `BindingTerminalAdmissionRefused` carrier does for lane occupancy.
    pub(super) fn observe(self) -> Result<(), ParticipantSemanticError> {
        match self {
            Self::Drained { drains, sealed } => {
                // ONE event: the operator sees "drained AND sealed" together,
                // per R-BOOT-VERDICT's no-silent-skip rule (§6.6).
                tracing::info!(
                    drains,
                    sealed,
                    "F8B boot drain emptied a restored immutable-candidate lane"
                );
                Ok(())
            }
            Self::AlreadyEmpty => Ok(()),
            Self::RefusedRecoveryArmed {
                conversation_id,
                candidate,
            } => Err(refused(
                conversation_id,
                BootDrainRefusal::RecoveryArmed,
                candidate,
                "the lane rests under an armed fenced-attach recovery block, which only a live \
                 fenced attach can consume"
                    .to_owned(),
            )),
            Self::RefusedShape {
                conversation_id,
                candidate,
                reason,
            } => Err(refused(
                conversation_id,
                BootDrainRefusal::Shape,
                candidate,
                reason,
            )),
        }
    }
}

/// Builds the loud boot refusal and emits it before it travels, so the reason
/// reaches the operator's log even where an outer startup mapping reduces the
/// error to its own phase text.
fn refused(
    conversation_id: ConversationId,
    refusal: BootDrainRefusal,
    candidate: ImmutableSequenceCandidate,
    reason: String,
) -> ParticipantSemanticError {
    let error = ParticipantSemanticError::BootDrainRefused {
        conversation_id,
        refusal,
        candidate: format!("{candidate:?}"),
        reason,
    };
    tracing::error!(%error, "F8B boot drain refused the boot");
    error
}

impl ProductionParticipantHandler {
    /// R-BOOT-DRAIN after the restore unit's unclean-restart repair.
    ///
    /// The restore unit drains the restored lane before it installs the
    /// owner, and only then runs the owed repair. A repair that ends a dead
    /// incarnation's binding may pend its terminal (the retained suffix is at
    /// its cap), and that terminal occupies the lane again. Left there, the
    /// conversation listened on an occupied lane and the next boot drained it
    /// instead, publishing a new checkpoint for work this boot owed. The
    /// unit drains it here, through the same drain and commit tail, so a
    /// restored conversation always reaches listening with an empty lane and
    /// the next boot restores this boot's checkpoint and writes nothing
    /// (decision of 11 October 2026).
    ///
    /// # Errors
    ///
    /// A refused drain verdict or a commit-tail failure, named as the boot
    /// drain names them.
    pub(super) fn drain_after_repair(
        &self,
        conversation_id: ConversationId,
    ) -> Result<(), ParticipantSemanticError> {
        let cell = self.cell(conversation_id)?;
        let mut owner = cell
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: format!(
                    "participant conversation {conversation_id} owner lock is poisoned"
                ),
            })?;
        let Some(authority) = owner.as_mut() else {
            return Ok(());
        };
        let log = OperationLog::new(Arc::clone(&self.store), conversation_id);
        let before_drain = authority.next_log_sequence;
        let verdict = self.drain_restored_candidate_lane(conversation_id, authority, &log);
        let drained = verdict.drained_any();
        verdict.observe()?;
        if drained {
            self.commit_boot_drain(conversation_id, authority, &log, before_drain, false)?;
        }
        drop(owner);
        Ok(())
    }

    /// Drains one restored conversation's immutable-candidate lane to empty
    /// (§6.2 R-BOOT-DRAIN) and returns the attempt's named verdict.
    ///
    /// The head is drained repeatedly until the lane reports empty — N markers
    /// need N drains, because the marker drain removes exactly the head. A
    /// head that survived its own successful drain would spin boot forever, so
    /// a lane that fails to shrink is refused as a shape defect instead of
    /// looped on: nothing in the protocol permits it, and boot must terminate.
    ///
    /// The impact accumulator is fresh and DISCARDED — boot has no connection
    /// to receive an impact; the listener is not bound until after
    /// `SupervisorInner::new` returns.
    ///
    /// That discard is ALSO the boot-drain answer to participant contract
    /// §0.16's `0x0202 MarkerSettled` wake, and it is correct rather than a
    /// gap. The amendment scopes the wake to "connections that received the
    /// refusal in THIS PROCESS LIFETIME"; at boot no connection has received
    /// anything, the settlement-waiter registry is empty by construction, and
    /// a wake fired here could only ever be a broadcast to parties that were
    /// never refused — which obligation 3 outlaws. `persist_next_marker` still
    /// records the settled epoch on the accumulator it is handed, so the same
    /// clearing write serves both callers; only this one has nobody to tell.
    pub(super) fn drain_restored_candidate_lane(
        &self,
        conversation_id: ConversationId,
        replayed: &mut ConversationAuthority,
        log: &OperationLog,
    ) -> BootDrainVerdict {
        // Board #60 §3c: the boot drain's marker sources complete their own
        // Unit 2 extension rows here, exactly as a live commit does. Boot is
        // the one place that appends base rows with no reconcile behind it, so
        // an appender without an extension log would leave the drain's
        // projection for the NEXT touch's replay to repair.
        let outbox_log = OutboxLog::new(Arc::clone(&self.store), conversation_id);
        // The drain's rows commit where a live commit puts them: one routed
        // batch per drained head under the conversation's route, base rows and
        // their extension rows together. A flat append here met the flat
        // stream's own head, which a routed conversation left at zero, and
        // the sequence conflict made the conversation unloadable at boot. The
        // flat arm remains only under the legacy-rows test arming.
        let route = checkpoint_conversation_route(conversation_id);
        #[cfg(test)]
        let flat_legacy = super::log::legacy_rows::armed();
        #[cfg(not(test))]
        let flat_legacy = false;
        let mut impact = DispatchImpactAccumulator::new();
        let mut drains: usize = 0;
        while let Some(head) = lane_head(replayed) {
            let batch = RefCell::new(Some(OperationBatch::new(&route)));
            let appender = LogAppender::new(
                log,
                &self.registry,
                &outbox_log,
                None,
                (!flat_legacy).then_some(&batch),
                0,
                &self.publisher,
            );
            if head.recovery_armed
                && matches!(
                    head.candidate,
                    ImmutableSequenceCandidate::BindingTerminal { .. }
                )
            {
                // §5.3(ii), measured: the terminal drain refuses while any
                // recovery block is armed. A marker head is unaffected — the
                // marker drain does not read `recovery` — so the guard is
                // asked only of a binding-terminal head.
                return BootDrainVerdict::RefusedRecoveryArmed {
                    conversation_id,
                    candidate: head.candidate,
                };
            }
            let owner = match replayed.take_frontier() {
                Ok(owner) => owner,
                Err(error) => {
                    return BootDrainVerdict::RefusedShape {
                        conversation_id,
                        candidate: head.candidate,
                        reason: error.to_string(),
                    };
                }
            };
            if let Err(error) =
                replayed.persist_drain_first(head.candidate, owner, &appender, &mut impact)
            {
                return BootDrainVerdict::RefusedShape {
                    conversation_id,
                    candidate: head.candidate,
                    reason: error.to_string(),
                };
            }
            let owed = appender.outstanding_extension_rows();
            if owed != 0 {
                return BootDrainVerdict::RefusedShape {
                    conversation_id,
                    candidate: head.candidate,
                    reason: format!(
                        "the drain left {owed} Unit 2 extension rows unwritten; the next load of \
                         this conversation writes them"
                    ),
                };
            }
            if let Err(reason) = commit_drained_rows(&appender) {
                return BootDrainVerdict::RefusedShape {
                    conversation_id,
                    candidate: head.candidate,
                    reason,
                };
            }
            if lane_head(replayed).is_some_and(|next| next.remaining >= head.remaining) {
                return BootDrainVerdict::RefusedShape {
                    conversation_id,
                    candidate: head.candidate,
                    reason: "the drained head did not leave the immutable-candidate lane"
                        .to_owned(),
                };
            }
            drains = match drains.checked_add(1) {
                Some(drains) => drains,
                None => {
                    return BootDrainVerdict::RefusedShape {
                        conversation_id,
                        candidate: head.candidate,
                        reason: "the boot drain count overflowed".to_owned(),
                    };
                }
            };
        }
        if drains == 0 {
            BootDrainVerdict::AlreadyEmpty
        } else {
            // R-SEAL (§6.6): the marker the drain apply derives, read back
            // rather than inferred here — the verdict reports what the apply
            // decided, so live, replay and boot can never disagree.
            BootDrainVerdict::Drained {
                drains,
                sealed: replayed.is_closed(),
            }
        }
    }
}

/// Commits the rows one drained head staged, as one durable operation on the
/// conversation's route; a flat appender (the legacy-rows arming) stages
/// nothing and commits nothing here.
fn commit_drained_rows(appender: &LogAppender<'_>) -> Result<(), String> {
    let staged = appender
        .has_staged()
        .map_err(|error| format!("the drain's staged rows could not be read: {error}"))?;
    if staged {
        appender
            .commit_staged()
            .map_err(|error| format!("the drain's rows did not commit: {error}"))?;
    }
    Ok(())
}

/// The lane's head plus the two facts the drain decision needs from it.
struct LaneHead {
    candidate: ImmutableSequenceCandidate,
    recovery_armed: bool,
    remaining: usize,
}

/// Reads the restored owner's lane head. A conversation with no coupled
/// frontier owner has no lane, and boot leaves it exactly as replay left it.
fn lane_head(replayed: &ConversationAuthority) -> Option<LaneHead> {
    let frontiers = replayed.frontier()?.frontiers();
    let candidates = frontiers.sequence().immutable_candidates();
    Some(LaneHead {
        candidate: *candidates.first()?,
        recovery_armed: frontiers.sequence().recovery().is_some()
            || frontiers.order().recovery().is_some(),
        remaining: candidates.len(),
    })
}