liminal-server 0.25.4

//! SRV-037: a marker that refuses an attach or a detach settles at that
//! refusal.
//!
//! A `MarkerSettlementBackpressure` refusal installs a waiter
//! ([`InstalledParticipantService::register_settlement_waiter`]), and a waiter
//! fires only where a marker candidate is drained. The record-admission
//! `DrainFirst` arm drains on the NEXT record admitted to the conversation, so
//! in a quiet conversation (a seat re-enrolling after its own Leave is the
//! only party, and the one waiting) nothing ever would. The refusal therefore
//! drains the lane itself, right after its waiter is installed, through the
//! same drain-and-tell a clean stop uses: the drain commits under the live
//! conversation lock, records the settled epoch, and the tell fires this
//! connection's waiter, so its `MarkerSettled` is in its inbox before the
//! dispatch returns. The response is written before the inbox is read, so the
//! refusal still precedes its wake on the wire.
//!
//! A2: the refusal is computed under the conversation lock and the waiter is
//! installed after it is released, so another connection's record can drain
//! the marker in between, with no waiter yet to fire. Once this drain has
//! emptied the lane, the refused epoch is settled whoever drained it, and its
//! waiter is fired by that epoch.

use liminal_protocol::wire::{MarkerSettlementBackpressure, ServerValue};

use super::{InstalledParticipantService, ParticipantSemanticError};

/// Every settlement refusal is followed by its `MarkerSettled` wake.
///
/// The guarantee a client may build against (SRV-037 with A2, liminal-server
/// 0.24.4): a credential attach or detach refused
/// `MarkerSettlementBackpressure` receives its `0x0202 MarkerSettled`, for the
/// epoch its refusal named, on its own connection and once, with no other
/// traffic on the conversation, including when another connection's record
/// drained the marker between the refusal and its waiter's install. A client
/// that waits on that wake alone, with no window, names this constant, so a
/// pin to a release without the guarantee fails to compile rather than hang.
pub const MARKER_SETTLED_FOLLOWS_EVERY_SETTLEMENT_REFUSAL: bool = true;

impl InstalledParticipantService {
    /// Drains the lane a marker-settlement refusal waits on, and tells it.
    ///
    /// Every other value is left alone. A drain the lane refuses (a binding
    /// terminal under an armed fenced-attach recovery block, or a drain that
    /// does not shorten the lane) leaves the refusal as answered and the
    /// waiter installed, with one warning: the request was answered truthfully
    /// and closing the connection for the lane's state would answer nothing.
    /// Every drain committed before that refusal has already been told.
    pub(super) fn settle_refused_marker(
        &self,
        value: &ServerValue,
    ) -> Result<(), ParticipantSemanticError> {
        let ServerValue::MarkerSettlementBackpressure(refusal) = value else {
            return Ok(());
        };
        let (conversation_id, refused_epoch) = match *refusal {
            MarkerSettlementBackpressure::CredentialAttach {
                conversation_id,
                refused_epoch,
            }
            | MarkerSettlementBackpressure::Detach {
                conversation_id,
                refused_epoch,
            } => (conversation_id, refused_epoch),
        };
        let (result, impact) = self
            .handler
            .drain_candidate_lane_with_impact(conversation_id)
            .into_parts();
        // A failed tell is the registry's own fault and closes, as every
        // other dispatch's does; only the lane's refusal is answered by a warn.
        self.notify_impact(&impact)?;
        if result.is_ok() {
            // SRV-037 A2: the refused marker was the lane head when the refusal
            // was computed, and the lane drains in order, so an emptied lane
            // has settled it whoever drained it -- this drain, or a record
            // another connection admitted after the refusal and before this
            // waiter existed, whose drain had no waiter to fire. Fired here
            // by its own epoch; a waiter is one-shot, so a wake this drain
            // already fired is never sent twice.
            self.fire_settlements(conversation_id, &[refused_epoch])?;
        }
        if let Err(error) = result {
            tracing::warn!(
                conversation_id,
                %error,
                "a marker-settlement refusal could not drain its conversation's lane; the \
                 refusal stands and its wake waits for the next drain"
            );
        }
        Ok(())
    }

    /// Test hook onto [`Self::settle_refused_marker`], beside the registry's
    /// own hooks in dispatch.rs.
    #[cfg(test)]
    pub(in crate::server::participant) fn settle_refused_marker_for_test(
        &self,
        value: &ServerValue,
    ) -> Result<(), ParticipantSemanticError> {
        self.settle_refused_marker(value)
    }
}