liminal-server 0.24.5

//! The single production participant semantic handler.
//!
//! One handler per server. Each conversation has exactly one live in-memory
//! authority owner, rebuilt from its durable transition-input log on first
//! touch (and after any failed operation, which discards the owner so the
//! next touch cold-replays durable reality). A short registry lock selects
//! the conversation cell; a per-conversation lock covers replay and each
//! operation. Everything is event-driven: cells are created on request
//! arrival, discarded on error, and evicted entirely when the touched
//! conversation has no durable log — no timer, sweep, or polling loop
//! exists, and refused probes of unknown conversation ids leave neither
//! durable nor in-memory residue.

use crate::config::types::ParticipantConfig;
use crate::health::reissue::{
    OperatorCredentialReissueError, OperatorCredentialReissueOutcome,
    OperatorCredentialReissueRequest, OperatorCredentialReissuer,
};
use crate::health::unloadable::{UnloadableConversation, UnloadableConversationRecord};
use crate::server::participant::{
    ObserverPublicationTarget, ParticipantConnectionContext, ParticipantConnectionConversations,
    ParticipantSemanticError, ParticipantServiceFatal, dispatch_impact::DispatchImpactAccumulator,
};
use crate::server::signals::ServerSignals;

use liminal::durability::bridge::block_on;
use liminal::durability::{DurabilityError, DurableStore};

use liminal_protocol::lifecycle::{CapacityCounter, ObserverRecoveryAggregate};
use liminal_protocol::wire::ConversationId;

use std::cell::{Cell, RefCell};
use std::collections::{BTreeMap, HashMap};
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Instant;

use super::barrier::{OperationFacts, ReceiptCapacityLimits};
use super::capacity::ServerCapacity;
use super::capacity_snapshot::CapacitySeed;
use super::checkpoint_batches::{
    CheckpointPublication, CheckpointPublisher, PublicationBound, PublicationLayoutError,
    PublicationRefusal, publication_bound,
};
#[cfg(test)]
use super::dispatch_work::{ObligationDispatchWorkCounters, ObligationDispatchWorkSnapshot};
use super::facts;
use super::lazy_restore::PendingRestores;
use super::log::{OperationLog, OperationLogError, StoredOperation};
use super::operation_batch::OperationBatch;
use super::outbox::{ConversationOutboxError, ConversationOutboxLimits};
use super::outbox_log::{OutboxLog, OutboxLogError};
use super::outbox_projection::owes_extension_row;
use super::registry::ConversationRegistry;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::routed_stream::RoutedStreamRefusal;
use super::state::{ConversationAuthority, DurableAppend, LifecycleReceipt, StateError};
use super::write_issuance::WriteIssuance;

#[derive(Debug)]
pub(super) struct ObserverArmTarget {
    pub(super) refused_epoch: u64,
    pub(super) connection_incarnation: liminal_protocol::wire::ConnectionIncarnation,
    pub(super) target: ObserverPublicationTarget,
}

/// One exclusively serialized observer owner: durable protocol aggregate/head
/// plus volatile weak live targets for its installed arms. Restoring durable
/// arms intentionally starts with an empty target map because no socket
/// survives a process restart.
#[derive(Debug)]
pub(super) struct ObserverOwner {
    pub(super) aggregate: ObserverRecoveryAggregate,
    pub(super) head: u64,
    pub(super) arm_targets: BTreeMap<ConversationId, ObserverArmTarget>,
}

/// Lifecycle receipts held for a commit whose outcome was not confirmed, with
/// the log head its operation started from.
#[derive(Debug)]
struct OwedReceipts {
    starting_log_sequence: u64,
    receipts: Vec<LifecycleReceipt>,
}

/// One conversation's owner, absent until it restores or enrolls.
pub(super) type ConversationCell = Arc<Mutex<Option<ConversationAuthority>>>;

/// Production semantic handler backed by the shared durable store.
///
/// Constructed exactly once per server by the connection-services layer when
/// the deployment's `[participant]` configuration is present.
#[derive(Debug)]
pub struct ProductionParticipantHandler {
    signals: OnceLock<ServerSignals>,
    pub(super) store: Arc<dyn DurableStore>,
    pub(super) config: ParticipantConfig,
    outbox_limits: ConversationOutboxLimits,
    conversations: Mutex<HashMap<ConversationId, Arc<Mutex<Option<ConversationAuthority>>>>>,
    /// DUR-020: registered conversations not yet restored in this process.
    /// Construction fills it from the registry and restores nothing; each
    /// conversation's first touch runs its restore unit (`lazy_restore`).
    pub(super) pending: PendingRestores,
    /// This handler as its warm thread holds it: weakly, so a server that
    /// drops its handler ends the warm at the next conversation boundary.
    pub(super) self_ref: OnceLock<std::sync::Weak<Self>>,
    /// Set when the server stops; the warm checks it between conversations.
    pub(super) warm_stop: AtomicBool,
    pub(super) warm_worker: Arc<super::warm::WarmWorker>,
    /// First post-Open fatal; once set, every semantic/publication entry refuses.
    service_fatal: Mutex<Option<ParticipantServiceFatal>>,
    /// Server-wide observer-recovery aggregate paired with its durable row
    /// head (`None` until first restored).
    pub(super) observer: Mutex<Option<ObserverOwner>>,
    pub(super) observer_head_uncertain: AtomicBool,
    /// Server-scope stage-8 occupancy ledger (identity slots, live receipts,
    /// provenance fingerprints), folded from each conversation as its restore
    /// unit runs and kept exact by commit reservations and replay folds. It
    /// is exact against durable truth once `pending` is empty, which is why
    /// enrollment, the one request a server-scope count refuses, finishes
    /// the warm first (DUR-020 R3).
    pub(super) capacity: ServerCapacity,
    /// The clean-stop capacity snapshot this process seeded from, if any,
    /// and the witness its next write names (DUR-020 R3).
    pub(super) capacity_seed: CapacitySeed,
    /// Durable registry of created conversations: one row appended before
    /// each conversation's genesis append, read at startup to record every
    /// durable conversation as pending (DUR-020).
    pub(super) registry: ConversationRegistry,
    /// CONTAINMENT: conversations whose durable state could not be loaded,
    /// each against the load failure's own class and text.
    ///
    /// A conversation lands here instead of taking the node down with it. The
    /// record is the node's answer to "which one is broken?", which is the half
    /// of the property that a boot-succeeds assertion cannot see: containment
    /// without attribution is a node that serves nothing on one conversation
    /// forever and never says so.
    ///
    /// The record is SHARED, not private: a clone of it is published onto the
    /// health endpoint's `GET /unloadable-conversations` at server startup
    /// ([`Self::unloadable_record`]), so the answer this handler writes is the
    /// same answer an operator reads. Nothing pushes — the endpoint snapshots
    /// this record only when it is scraped.
    unloadable: UnloadableConversationRecord,
    /// Lifecycle receipts owed by a commit whose outcome this process could
    /// not confirm: its batch was issued, and the operation failed without
    /// proof that its rows landed. A receipt is owed by the commit, not by
    /// the answer that first reported it, so it is held here, with the log
    /// head the operation started from, until the next load of the
    /// conversation shows whether the rows landed. Landed, the receipts are
    /// emitted then, exactly once. Not landed, they are dropped, because a
    /// re-presented request commits live and records its own. Held in memory
    /// only: a process that stops before that load owes them to nobody, as
    /// a crash between any commit and its emission does.
    owed_receipts: Mutex<HashMap<ConversationId, OwedReceipts>>,
    /// Exact W2 work observation points, isolated per handler and test-only.
    #[cfg(test)]
    pub(super) obligation_dispatch_work: ObligationDispatchWorkCounters,
    /// The participant clock's pin, for deterministic time-window tests.
    ///
    /// `0` means the wall clock, and production never writes anything else:
    /// only the test-only [`Self::pin_clock_ms`] stores a nonzero reading, so a
    /// test can step through the receipt and provenance window states instead
    /// of racing the wall clock. Production pays one atomic load per clock
    /// read for it, and in return [`Self::now_ms`] is one method in every build.
    now_override: AtomicU64,
    /// DUR-024: checkpoint publications run off the owner's write path, and
    /// the ledger of where each conversation's publication stands.
    pub(super) publisher: CheckpointPublisher,
    /// Takes of a conversation's owner cell through [`Self::cell`], for the
    /// pin that a publication takes none while it submits (DUR-024 R3).
    #[cfg(test)]
    pub(super) owner_lock_takes: AtomicU64,
}

impl ProductionParticipantHandler {
    pub(in crate::server) fn install_signals(
        &self,
        signals: ServerSignals,
    ) -> Result<(), ParticipantSemanticError> {
        self.signals
            .set(signals)
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant lifecycle signals were already installed".to_owned(),
            })
    }

    /// Creates the handler over the server's shared durable store. It reads
    /// the conversation registry and records every registered conversation
    /// as pending; it restores none of them (DUR-020 R1), so the server
    /// listens in the same time whatever the number of conversations. Each
    /// conversation restores on its first touch, and the warm restores the
    /// rest while the server serves. A restart still forgets no reserved
    /// identity slot: enrollment, the one request a server-scope count
    /// refuses, finishes the warm before it decides.
    ///
    /// # Errors
    ///
    /// Returns [`ParticipantSemanticError`] when the registry cannot be read:
    /// without it no conversation can be named, so the server refuses to
    /// start over state it cannot account for.
    pub fn new(
        store: Arc<dyn DurableStore>,
        config: ParticipantConfig,
    ) -> Result<Self, ParticipantSemanticError> {
        let outbox_limits = ConversationOutboxLimits::try_new(
            config.max_retained_record_rows,
            config.identity_slots,
        )
        .map_err(|error| ParticipantSemanticError::Internal {
            message: format!("participant outbox limit configuration failed: {error}"),
        })?;
        let registry = ConversationRegistry::new(Arc::clone(&store));
        let handler = Self {
            signals: OnceLock::new(),
            store,
            config,
            outbox_limits,
            conversations: Mutex::new(HashMap::new()),
            pending: PendingRestores::default(),
            self_ref: OnceLock::new(),
            warm_stop: AtomicBool::new(false),
            warm_worker: Arc::new(super::warm::WarmWorker::default()),
            service_fatal: Mutex::new(None),
            observer: Mutex::new(None),
            observer_head_uncertain: AtomicBool::new(false),
            capacity: ServerCapacity::default(),
            capacity_seed: CapacitySeed::default(),
            registry,
            unloadable: UnloadableConversationRecord::default(),
            owed_receipts: Mutex::new(HashMap::new()),
            #[cfg(test)]
            obligation_dispatch_work: ObligationDispatchWorkCounters::default(),
            now_override: AtomicU64::new(0),
            publisher: CheckpointPublisher::default(),
            #[cfg(test)]
            owner_lock_takes: AtomicU64::new(0),
        };
        if handler.config.pruning.is_none() {
            // DUR-024 R4: no section, the pinned default, said once by name
            // at start. Pruning is never off: a conversation whose rows are
            // never cut grows without limit, and its next checkpoint with it.
            tracing::info!(
                keep_acknowledged_behind_checkpoint =
                    super::prune::DEFAULT_KEEP_ACKNOWLEDGED_BEHIND_CHECKPOINT,
                "pruning on by default: {} rows behind the checkpoint ([participant.pruning] is absent)",
                super::prune::DEFAULT_KEEP_ACKNOWLEDGED_BEHIND_CHECKPOINT
            );
        }
        handler.note_registered_conversations()?;
        Ok(handler)
    }

    /// Reads the participant clock as Unix milliseconds: the wall clock, or
    /// the reading a test pinned with [`Self::pin_clock_ms`].
    fn now_ms(&self) -> Result<u64, facts::FactsError> {
        match self.now_override.load(Ordering::SeqCst) {
            0 => facts::now_unix_millis(),
            fixed => Ok(fixed),
        }
    }

    /// Pins the participant clock to a fixed Unix-millisecond reading, or
    /// releases it back to the wall clock when passed `0`.
    ///
    /// Harness-owned and test-only: it is the only writer of
    /// [`Self::now_override`]. Time-window tests set an explicit
    /// base and step it across the deterministic window states they assert.
    #[cfg(test)]
    pub(super) fn pin_clock_ms(&self, now_ms: u64) {
        self.now_override.store(now_ms, Ordering::SeqCst);
    }

    #[cfg(test)]
    pub(crate) fn obligation_dispatch_work_snapshot(&self) -> ObligationDispatchWorkSnapshot {
        self.obligation_dispatch_work.snapshot()
    }

    pub(super) fn current_service_fatal(
        &self,
    ) -> Result<Option<ParticipantServiceFatal>, ParticipantSemanticError> {
        let fatal = self
            .service_fatal
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant service fatal latch is poisoned".to_owned(),
            })?
            .clone();
        Ok(fatal)
    }

    pub(super) fn ensure_service_live(&self) -> Result<(), ParticipantSemanticError> {
        if let Some(fatal) = self.current_service_fatal()? {
            return Err(ParticipantSemanticError::ServiceFatal(fatal));
        }
        Ok(())
    }

    pub(super) fn latch_connection_fate_fatal(
        &self,
        open_sequence: u64,
        conversation_id: ConversationId,
    ) -> Result<ParticipantServiceFatal, ParticipantSemanticError> {
        let mut fatal =
            self.service_fatal
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "participant service fatal latch is poisoned".to_owned(),
                })?;
        let selected = fatal
            .get_or_insert_with(|| ParticipantServiceFatal::ConnectionFateIntentIncomplete {
                open_sequence,
                conversation_id,
            })
            .clone();
        drop(fatal);
        Ok(selected)
    }

    /// Builds the handler and finishes the warm before returning: the
    /// pre-DUR-020 boot, which every test that pins what a boot restores,
    /// refuses or drains still means.
    #[cfg(test)]
    pub(super) fn new_restored(
        store: Arc<dyn DurableStore>,
        config: ParticipantConfig,
    ) -> Result<Self, ParticipantSemanticError> {
        let handler = Self::new(store, config)?;
        handler.restore_remaining("a test boot restores every conversation before returning")?;
        Ok(handler)
    }

    /// Startup: reads the registry and records every registered conversation
    /// as pending. Nothing about any conversation is read here (DUR-020 R1).
    fn note_registered_conversations(&self) -> Result<(), ParticipantSemanticError> {
        // THE REGISTRY READ STAYS FATAL, and that is the property stated
        // precisely rather than an exception carved out of it. Containment
        // here is PER-CONVERSATION containment: without the registry read
        // there is no per-conversation subject to attribute a failure to, so a
        // failure with no conversation to name is the one thing that still
        // should stop the node.
        let started = Instant::now();
        #[cfg(test)]
        let conversation_ids = crate::boot_history_harness::during_stage("registry", || {
            self.restore_registry_finishing_sweeps()
        })?;
        #[cfg(not(test))]
        let conversation_ids = self.restore_registry_finishing_sweeps()?;
        // DUR-028 R4: a store written before the namespace is brought under
        // it here, once, before any conversation is restored and before the
        // listener binds; a boot that finds the marker makes one read.
        super::route_key_migration::migrate_route_keys(&self.store, &conversation_ids).map_err(
            |error| ParticipantSemanticError::Internal {
                message: error.to_string(),
            },
        )?;
        let pending = self.pending.register(conversation_ids)?;
        #[cfg(test)]
        crate::boot_history_harness::during_stage("registry", || {
            self.seed_capacity_from_snapshot();
        });
        #[cfg(not(test))]
        self.seed_capacity_from_snapshot();
        tracing::info!(
            step = "participant restore",
            conversations = pending,
            elapsed_ms = elapsed_ms(started),
            "participant restore deferred: every registered conversation restores on its first touch or in the warm"
        );
        Ok(())
    }
}

/// The operator surface reaches the participant through the handler itself
/// (R18 amendment A7, §0.18): possession of the operator surface is the
/// authority, and this impl is the only bridge between it and the serialized
/// participant-state point.
impl OperatorCredentialReissuer for ProductionParticipantHandler {
    fn reissue(
        &self,
        request: OperatorCredentialReissueRequest,
    ) -> Result<OperatorCredentialReissueOutcome, OperatorCredentialReissueError> {
        self.operator_credential_reissue(request)
            .map_err(|error| OperatorCredentialReissueError {
                message: error.to_string(),
            })
    }
}

#[path = "handler_cells.rs"]
mod cells;
#[path = "handler_errors.rs"]
mod errors;
#[path = "handler_log_appender.rs"]
mod log_appender;
#[path = "handler_log_staging.rs"]
mod log_staging;
#[path = "handler_operations.rs"]
mod operations;
#[path = "handler_restore.rs"]
mod restore;
#[cfg(test)]
pub(super) use errors::outbox_log_error;
use errors::state_error_with_issuance;
pub(super) use errors::{bridge_error, elapsed_ms, log_error, restore_error, state_error};
pub(super) use log_appender::LogAppender;
use log_appender::{checkpoint_state_error, install_staged, lock_owner};

#[cfg(test)]
mod batch_tests;