liminal_server/server/participant/

dispatch.rs

//! Participant transport-to-semantics dispatch boundary.
//!
//! This module contains no lifecycle rules. The shared protocol crate gates and
//! decodes inbound frames, while an injected semantic handler returns one typed
//! protocol value. The server then performs only generic-frame encoding.

use std::collections::BTreeSet;
use std::sync::Arc;

use liminal::durability::DurableStore;
use liminal::protocol::Frame;
use liminal_protocol::lifecycle::ConnectionConversationTracking;
use liminal_protocol::wire::{
    BindingEpoch, ClientRequest, CodecError, ConnectionIncarnation, ConversationId,
    ObserverRecoveryHandshake, ParticipantId, ServerValue, ValidatedFrameLimit,
};

use super::transport::{
    ParticipantIngress, ParticipantSession, encode_server_value, gate_generic_frame,
    normalize_configured_frame_limit,
};
use super::{
    ObserverPublicationTarget, ParticipantOfferedProgress, ParticipantPublication,
    ParticipantPublicationInbox, ParticipantPublicationRegistry,
};

/// Connection-local semantic-conversation dispatch map (contract R-D1: the
/// connection's binding/interest/dispatch maps are bounded by the signed
/// `max_semantic_conversations_per_connection`).
///
/// One value lives in each connection process's state for the connection's
/// lifetime and is dropped with it. A conversation enters the map exactly
/// when a semantic operation for it COMMITS on this connection (the crate's
/// `ConnectionConversationCapacityCommit::newly_tracked` verdict) or when an
/// observer-recovery batch arms its refusal-only recipient; refusals and
/// replays leave the map untouched, exactly as the crate's stage-6 selector
/// leaves its counter unchanged. Growth is therefore bounded by the signed
/// limit the stage-6 selector enforces.
#[derive(Debug, Default)]
pub struct ParticipantConnectionConversations {
    tracked: BTreeSet<ConversationId>,
}

impl ParticipantConnectionConversations {
    /// Stage-6 tracking fact for one conversation on this connection.
    #[must_use]
    pub fn tracking(&self, conversation_id: ConversationId) -> ConnectionConversationTracking {
        if self.tracked.contains(&conversation_id) {
            ConnectionConversationTracking::AlreadyTracked
        } else {
            ConnectionConversationTracking::Untracked
        }
    }

    /// Current connection-conversation occupancy.
    #[must_use]
    pub fn occupied(&self) -> u64 {
        // `usize` fits `u64` on every supported target; if that ever stopped
        // holding, saturating at MAX fails CLOSED (capacity reads as full)
        // rather than silently under-counting occupancy.
        u64::try_from(self.tracked.len()).unwrap_or(u64::MAX)
    }

    /// Installs one conversation slot after a capacity-committing operation.
    pub fn track(&mut self, conversation_id: ConversationId) {
        self.tracked.insert(conversation_id);
    }

    /// Sorted tracked conversations (the observer-recovery preflight's
    /// current-occupancy input).
    #[must_use]
    pub fn tracked_conversations(&self) -> Vec<ConversationId> {
        self.tracked.iter().copied().collect()
    }
}

/// Connection-scoped authority facts supplied to participant semantics.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ParticipantConnectionContext {
    connection_incarnation: ConnectionIncarnation,
}

impl ParticipantConnectionContext {
    /// Captures the durably allocated incarnation of the receiving connection.
    #[must_use]
    pub const fn new(connection_incarnation: ConnectionIncarnation) -> Self {
        Self {
            connection_incarnation,
        }
    }

    /// Returns the durably allocated receiving-connection incarnation.
    #[must_use]
    pub const fn connection_incarnation(self) -> ConnectionIncarnation {
        self.connection_incarnation
    }
}

/// Non-wire semantic service failure.
///
/// A failure is terminal to the connection attempt. It is deliberately not
/// convertible to [`ServerValue`], preventing the server from inventing a
/// lifecycle response when the protocol-owned transition did not produce one.
#[derive(Clone, Debug, thiserror::Error, PartialEq, Eq)]
pub enum ParticipantSemanticError {
    /// The complete semantic service is not installed.
    #[error("participant semantic service is unavailable")]
    Unavailable,
    /// Durable state or a protocol invariant prevented semantic completion.
    #[error("participant semantic service failed: {message}")]
    Internal {
        /// Diagnostic text for server logs; never placed on the participant wire.
        message: String,
    },
}

/// Server-owned adapter from a decoded request to a protocol-owned value.
pub trait ParticipantSemanticHandler: core::fmt::Debug + Send + Sync {
    /// Applies one already authenticated and capability-gated request.
    ///
    /// `conversations` is the receiving connection's semantic-conversation
    /// dispatch map: the handler reads it for the crate's stage-6
    /// connection-conversation capacity facts and installs a slot exactly
    /// when an operation's capacity commit reports `newly_tracked`.
    ///
    /// # Errors
    ///
    /// Returns [`ParticipantSemanticError`] when no protocol value can be
    /// produced. The caller closes rather than fabricating a response.
    ///
    /// Production handlers override this with the signed
    /// `max_semantic_conversations_per_connection`; semantic-only fixtures own
    /// no publication conversations.
    fn publication_conversation_limit(&self) -> u64 {
        0
    }

    /// Resolves all live current bindings with pending durable obligations for
    /// one conversation. Production overrides this; semantic-only fixtures have
    /// no publication source.
    ///
    /// # Errors
    ///
    /// Returns a semantic fault when durable readiness cannot be resolved.
    fn ready_connection_incarnations(
        &self,
        _conversation_id: ConversationId,
    ) -> Result<Vec<ConnectionIncarnation>, ParticipantSemanticError> {
        Ok(Vec::new())
    }

    /// Selects the least durable recipient obligation for this incarnation,
    /// restarting from durable ack when `offered` names an older binding.
    ///
    /// # Errors
    ///
    /// Returns a semantic fault when the durable obligation owner is unavailable.
    fn next_publication(
        &self,
        _connection_incarnation: ConnectionIncarnation,
        _conversation_id: ConversationId,
        _offered: Option<ParticipantOfferedProgress>,
    ) -> Result<Option<ParticipantPublication>, ParticipantSemanticError> {
        Ok(None)
    }

    /// Checks that a held head still belongs to the exact current binding before
    /// it is offered after writable readiness.
    ///
    /// # Errors
    ///
    /// Returns a semantic fault when current binding authority cannot be read.
    fn publication_binding_is_current(
        &self,
        _conversation_id: ConversationId,
        _participant_id: ParticipantId,
        _binding_epoch: BindingEpoch,
    ) -> Result<bool, ParticipantSemanticError> {
        Ok(false)
    }

    /// Records exact successful marker enqueue testimony. Non-marker offers are
    /// ignored by production after validating their current binding.
    ///
    /// # Errors
    ///
    /// Returns a semantic fault when exact offer testimony cannot be recorded.
    fn record_publication_offer(
        &self,
        _publication: &ParticipantPublication,
    ) -> Result<(), ParticipantSemanticError> {
        Ok(())
    }

    /// Applies observer recovery with the weak exact-live-connection target
    /// captured by the installed service. Semantic-only handlers delegate to
    /// their ordinary request path and do not own observer publication.
    ///
    /// # Errors
    ///
    /// Returns [`ParticipantSemanticError`] under the same contract as
    /// [`Self::handle`].
    fn handle_observer_recovery(
        &self,
        context: ParticipantConnectionContext,
        conversations: &mut ParticipantConnectionConversations,
        request: ObserverRecoveryHandshake,
        target: Option<ObserverPublicationTarget>,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        drop(target);
        self.handle(
            context,
            conversations,
            ClientRequest::ObserverRecovery(request),
        )
    }

    /// Applies one decoded participant request to protocol-owned authority.
    ///
    /// # Errors
    ///
    /// Returns [`ParticipantSemanticError`] when durable or protocol authority
    /// cannot produce a truthful terminal value. The connection fails rather
    /// than fabricating a response.
    fn handle(
        &self,
        context: ParticipantConnectionContext,
        conversations: &mut ParticipantConnectionConversations,
        request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError>;
}

/// Server-sealed participant activation token installed on a connection
/// supervisor.
///
/// The semantic handler and its durable store form one value so participant
/// capability activation cannot observe one without the other. The supervisor
/// uses the store to durably allocate connection incarnations before spawning a
/// connection process, and the process uses the handler only after that exact
/// incarnation has been carried into its state. The token atomically carries the
/// pair declared by server composition; it does not independently prove storage
/// namespace identity.
///
/// Construction and access are server-private. Until a complete production
/// lifecycle handler exists, external [`ConnectionServices`](crate::server::connection::ConnectionServices)
/// implementations cannot manufacture an activation token or advertise the
/// participant capability.
#[derive(Clone, Debug)]
pub struct InstalledParticipantService {
    handler: Arc<dyn ParticipantSemanticHandler>,
    durable_store: Arc<dyn DurableStore>,
    frame_limit: ValidatedFrameLimit,
    publication_registry: Arc<ParticipantPublicationRegistry>,
}

impl InstalledParticipantService {
    /// Pairs a semantic handler, its declared durable store, and the raw
    /// configured participant wire-frame limit.
    ///
    /// Production construction happens exactly once, in the server's
    /// connection-services layer, from the deployment's `[participant]`
    /// configuration; tests construct it directly with fixture handlers.
    ///
    /// # Errors
    ///
    /// Returns the shared codec error when the configured limit is smaller than
    /// the protocol's minimum complete frame.
    pub(crate) fn new(
        handler: Arc<dyn ParticipantSemanticHandler>,
        durable_store: Arc<dyn DurableStore>,
        configured_wf: u64,
    ) -> Result<Self, CodecError> {
        Ok(Self {
            handler,
            durable_store,
            frame_limit: normalize_configured_frame_limit(configured_wf)?,
            publication_registry: Arc::new(ParticipantPublicationRegistry::default()),
        })
    }

    /// Clones the durable store used by the installed participant service.
    #[must_use]
    pub(crate) fn durable_store(&self) -> Arc<dyn DurableStore> {
        Arc::clone(&self.durable_store)
    }

    /// Returns the normalized configured complete-frame limit advertised by
    /// this installed participant service.
    #[must_use]
    pub(crate) const fn frame_limit(&self) -> ValidatedFrameLimit {
        self.frame_limit
    }

    /// Returns the signed semantic-conversation allowance shared by publication
    /// readiness and connection-held encoded heads.
    #[must_use]
    pub(crate) fn publication_conversation_limit(&self) -> u64 {
        self.handler.publication_conversation_limit()
    }

    /// Creates the strongly connection-owned ready inbox at process spawn.
    #[must_use]
    pub(crate) fn new_publication_inbox(&self) -> ParticipantPublicationInbox {
        ParticipantPublicationInbox::new(self.handler.publication_conversation_limit())
    }

    /// Returns the shared weak publication registry.
    #[must_use]
    pub(crate) fn publication_registry(&self) -> &ParticipantPublicationRegistry {
        &self.publication_registry
    }

    /// Selects one exact durable publication through the installed production
    /// source.
    pub(crate) fn next_publication(
        &self,
        connection_incarnation: ConnectionIncarnation,
        conversation_id: ConversationId,
        offered: Option<ParticipantOfferedProgress>,
    ) -> Result<Option<ParticipantPublication>, ParticipantSemanticError> {
        self.handler
            .next_publication(connection_incarnation, conversation_id, offered)
    }

    pub(crate) fn publication_binding_is_current(
        &self,
        conversation_id: ConversationId,
        participant_id: ParticipantId,
        binding_epoch: BindingEpoch,
    ) -> Result<bool, ParticipantSemanticError> {
        self.handler
            .publication_binding_is_current(conversation_id, participant_id, binding_epoch)
    }

    pub(crate) fn record_publication_offer(
        &self,
        publication: &ParticipantPublication,
    ) -> Result<(), ParticipantSemanticError> {
        self.handler.record_publication_offer(publication)
    }

    fn notify_ready(
        &self,
        conversation_id: ConversationId,
    ) -> Result<(), ParticipantSemanticError> {
        for incarnation in self
            .handler
            .ready_connection_incarnations(conversation_id)?
        {
            self.publication_registry
                .notify(incarnation, conversation_id)
                .map_err(|error| ParticipantSemanticError::Internal {
                    message: format!("participant publication wake failed: {error}"),
                })?;
        }
        Ok(())
    }
}

impl ParticipantSemanticHandler for InstalledParticipantService {
    fn publication_conversation_limit(&self) -> u64 {
        self.handler.publication_conversation_limit()
    }

    fn handle(
        &self,
        context: ParticipantConnectionContext,
        conversations: &mut ParticipantConnectionConversations,
        request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        if let ClientRequest::ObserverRecovery(request) = request {
            let target = self
                .publication_registry
                .observer_target(context.connection_incarnation())
                .map_err(|error| ParticipantSemanticError::Internal {
                    message: format!("observer publication target failed: {error}"),
                })?;
            return self
                .handler
                .handle_observer_recovery(context, conversations, request, target);
        }
        let conversation_id = request_conversation_id(&request);
        let value = self.handler.handle(context, conversations, request)?;
        if let Some(conversation_id) = conversation_id {
            self.notify_ready(conversation_id)?;
        }
        Ok(value)
    }
}

const fn request_conversation_id(request: &ClientRequest) -> Option<ConversationId> {
    match request {
        ClientRequest::Enrollment(request) => Some(request.conversation_id),
        ClientRequest::CredentialAttach(request) => Some(request.conversation_id),
        ClientRequest::Detach(request) => Some(request.conversation_id),
        ClientRequest::ParticipantAck(request) => Some(request.conversation_id),
        ClientRequest::Leave(request) => Some(request.conversation_id),
        ClientRequest::MarkerAck(request) => Some(request.conversation_id),
        ClientRequest::RecordAdmission(request) => Some(request.conversation_id),
        ClientRequest::ObserverRecovery(_) => None,
    }
}

/// Result of dispatching one generic frame through participant transport.
#[derive(Debug)]
pub enum ParticipantDispatch {
    /// The generic frame belongs to another protocol.
    NotParticipant,
    /// Exact encoded response selected by the shared gate or semantic handler.
    Respond(Frame),
    /// Exact crate-owned pre-semantic rejection, followed by connection close.
    RespondThenClose(Frame),
    /// No truthful participant response exists; the connection must fail closed.
    Fatal(ParticipantDispatchError),
}

/// Failure after a generic frame has entered participant dispatch.
#[derive(Debug, thiserror::Error)]
pub enum ParticipantDispatchError {
    /// The preserved generic frame could not represent a canonical participant frame.
    #[error("invalid generic participant frame")]
    InvalidGenericFrame,
    /// The semantic handler could not produce a protocol value.
    #[error(transparent)]
    Semantic(#[from] ParticipantSemanticError),
    /// The crate-produced value could not be encoded into the generic transport.
    #[error("failed to encode participant response: {0:?}")]
    Encode(CodecError),
}

/// Gates, decodes, semantically applies, and encodes one participant frame.
///
/// Transport rejection values originate in `liminal-protocol`; semantic values
/// originate only in `handler`. No lifecycle outcome is constructed here.
#[must_use]
pub fn dispatch_generic_frame(
    frame: &Frame,
    authenticated: bool,
    session: ParticipantSession,
    context: ParticipantConnectionContext,
    conversations: &mut ParticipantConnectionConversations,
    handler: &dyn ParticipantSemanticHandler,
) -> ParticipantDispatch {
    let (value, close_after_response) = match gate_generic_frame(frame, authenticated, session) {
        ParticipantIngress::NotParticipant => return ParticipantDispatch::NotParticipant,
        ParticipantIngress::Rejected(rejection) => {
            (ServerValue::ParticipantTransportRejected(rejection), true)
        }
        ParticipantIngress::InvalidGenericFrame => {
            return ParticipantDispatch::Fatal(ParticipantDispatchError::InvalidGenericFrame);
        }
        ParticipantIngress::Request(request) => {
            match handler.handle(context, conversations, request) {
                Ok(value) => (value, false),
                Err(error) => {
                    return ParticipantDispatch::Fatal(ParticipantDispatchError::Semantic(error));
                }
            }
        }
    };
    match encode_server_value(value) {
        Ok(frame) if close_after_response => ParticipantDispatch::RespondThenClose(frame),
        Ok(frame) => ParticipantDispatch::Respond(frame),
        Err(error) => ParticipantDispatch::Fatal(ParticipantDispatchError::Encode(error)),
    }
}