liminal_server/server/participant/

dispatch_impact.rs

//! Typed post-commit tells for participant obligation dispatch.
//!
//! One accumulator belongs to one semantic request while its conversation owner
//! is locked. Producers record effects immediately after each durable subcommit
//! is installed. Finishing the accumulator preserves every committed prefix,
//! including when a later operation refuses or fails.

use std::cmp::Ordering;
use std::collections::{BTreeMap, BTreeSet};

use liminal_protocol::wire::{BindingEpoch, ConversationId, ParticipantId};

/// Exhaustive reasons a committed participant operation can change dispatch
/// permission.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub enum DispatchEffect {
    /// A Produced batch installed at least one live recipient obligation.
    Published,
    /// A normal or marker acknowledgement changed a dispatch cursor or verdict.
    Acknowledged,
    /// A current binding was installed, replaced, detached, or retired.
    BindingChanged,
    /// Coupled closure-debt episode state changed.
    EpisodeChanged,
    /// A permanent `Left` commit discharged one participant.
    Retired,
}

/// Exact poststate binding eligible to be told about changed permission.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DispatchTarget {
    participant_id: ParticipantId,
    binding_epoch: BindingEpoch,
}

/// The order of the wake-ups only: each target is woken once, whatever the
/// order. It does not mean a later binding.
impl Ord for DispatchTarget {
    fn cmp(&self, other: &Self) -> Ordering {
        (self.participant_id, self.binding_epoch.canonical_sort_key()).cmp(&(
            other.participant_id,
            other.binding_epoch.canonical_sort_key(),
        ))
    }
}

impl PartialOrd for DispatchTarget {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl DispatchTarget {
    /// Captures one exact current participant binding.
    #[must_use]
    pub const fn new(participant_id: ParticipantId, binding_epoch: BindingEpoch) -> Self {
        Self {
            participant_id,
            binding_epoch,
        }
    }

    /// Returns the permanent participant identity.
    #[must_use]
    pub const fn participant_id(self) -> ParticipantId {
        self.participant_id
    }

    /// Returns the exact current binding, including connection incarnation.
    #[must_use]
    pub const fn binding_epoch(self) -> BindingEpoch {
        self.binding_epoch
    }
}

/// Complete request-level post-commit dispatch impact.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum DispatchImpact {
    /// No committed subcommit changed dispatch permission.
    Unchanged,
    /// At least one truthful effect occurred for one conversation.
    Changed {
        /// Conversation whose locked authority produced the effects.
        conversation_id: ConversationId,
        /// Effect map. Individual truthful effects may have no current binding
        /// target. May be empty when only a settlement cleared.
        effects: BTreeMap<DispatchEffect, BTreeSet<DispatchTarget>>,
        /// Marker settlement epochs this request's durable drains CLEARED
        /// (participant contract §0.16 condition 2).
        ///
        /// A separate lane from `effects` on purpose: an effect names a
        /// participant BINDING eligible to be told about changed dispatch
        /// permission, and a connection refused `MarkerSettlementBackpressure`
        /// holds no binding — that is precisely why the amendment needs a
        /// pushed wake instead of a `ParticipantDelivery`. Routing settlements
        /// through `DispatchTarget` would therefore be routing them to the one
        /// population that cannot receive them.
        settled_epochs: Vec<u64>,
        /// Whether this request's durable drains removed at least one head of
        /// the conversation's immutable-candidate lane, marker or binding
        /// terminal alike.
        ///
        /// A connection fate refused by candidate-lane `Precedence` is parked
        /// until exactly this happens, so this is the tell that wakes it. It is
        /// a lane of its own for the same reason `settled_epochs` is: the
        /// parked fate belongs to a connection that has already gone, so no
        /// `DispatchTarget` can name it.
        candidate_lane_drained: bool,
    },
}

impl DispatchImpact {
    /// Returns the changed conversation, if any.
    #[must_use]
    pub const fn conversation_id(&self) -> Option<ConversationId> {
        match self {
            Self::Unchanged => None,
            Self::Changed {
                conversation_id, ..
            } => Some(*conversation_id),
        }
    }

    /// Returns the marker settlement epochs cleared by this request.
    #[must_use]
    pub fn settled_epochs(&self) -> &[u64] {
        match self {
            Self::Unchanged => &[],
            Self::Changed { settled_epochs, .. } => settled_epochs,
        }
    }

    /// Returns whether a durable drain removed a candidate-lane head.
    #[must_use]
    pub const fn candidate_lane_drained(&self) -> bool {
        match self {
            Self::Unchanged => false,
            Self::Changed {
                candidate_lane_drained,
                ..
            } => *candidate_lane_drained,
        }
    }

    /// Returns the effect map for a changed impact.
    #[must_use]
    pub const fn effects(&self) -> Option<&BTreeMap<DispatchEffect, BTreeSet<DispatchTarget>>> {
        match self {
            Self::Unchanged => None,
            Self::Changed { effects, .. } => Some(effects),
        }
    }

    /// Unions and deduplicates all exact targets without applying effect
    /// precedence.
    #[must_use]
    pub fn target_union(&self) -> BTreeSet<DispatchTarget> {
        match self {
            Self::Unchanged => BTreeSet::new(),
            Self::Changed { effects, .. } => effects
                .values()
                .flat_map(|targets| targets.iter().copied())
                .collect(),
        }
    }
}

/// Lossless request-scoped accumulator for installed subcommit effects.
#[derive(Debug, Default)]
pub struct DispatchImpactAccumulator {
    effects: BTreeMap<DispatchEffect, BTreeSet<DispatchTarget>>,
    staged_effects: BTreeMap<DispatchEffect, BTreeSet<DispatchTarget>>,
    settled_epochs: Vec<u64>,
    candidate_lane_drained: bool,
}

impl DispatchImpactAccumulator {
    /// Starts an empty request accumulator.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            effects: BTreeMap::new(),
            staged_effects: BTreeMap::new(),
            settled_epochs: Vec::new(),
            candidate_lane_drained: false,
        }
    }

    /// Records one truthful effect after its durable subcommit is installed.
    ///
    /// Repeated effects union exact target sets. An empty target iterator still
    /// records the effect because lack of a current binding does not erase a
    /// committed permission change.
    pub fn record(
        &mut self,
        effect: DispatchEffect,
        targets: impl IntoIterator<Item = DispatchTarget>,
    ) {
        self.effects.entry(effect).or_default().extend(targets);
    }

    /// Stages an effect whose enclosing durability/reconciliation barrier has
    /// not installed every coupled owner yet.
    pub(crate) fn stage(
        &mut self,
        effect: DispatchEffect,
        targets: impl IntoIterator<Item = DispatchTarget>,
    ) {
        self.staged_effects
            .entry(effect)
            .or_default()
            .extend(targets);
    }

    /// Commits every staged effect after the enclosing barrier is installed.
    pub(crate) fn install_staged(&mut self) {
        let staged = core::mem::take(&mut self.staged_effects);
        for (effect, targets) in staged {
            self.record(effect, targets);
        }
    }

    /// Discards effects whose enclosing operation was never confirmed.
    pub(crate) fn discard_staged(&mut self) {
        self.staged_effects.clear();
    }

    /// Merges another installed-prefix accumulator without replacing any
    /// earlier effect or target.
    pub fn merge(&mut self, prefix: Self) {
        for (effect, targets) in prefix.effects {
            self.record(effect, targets);
        }
        for (effect, targets) in prefix.staged_effects {
            self.stage(effect, targets);
        }
    }

    /// Returns whether no committed subcommit recorded an effect.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.effects.is_empty()
    }

    /// Converts the request accumulator into its externally carried impact.
    #[must_use]
    pub fn finish(self, conversation_id: ConversationId) -> DispatchImpact {
        if self.effects.is_empty() && self.settled_epochs.is_empty() && !self.candidate_lane_drained
        {
            DispatchImpact::Unchanged
        } else {
            DispatchImpact::Changed {
                conversation_id,
                effects: self.effects,
                settled_epochs: self.settled_epochs,
                candidate_lane_drained: self.candidate_lane_drained,
            }
        }
    }

    /// Records one marker settlement epoch AFTER its drain row is durable.
    ///
    /// Ordering is the whole point: the wake promises the refused connection
    /// that retrying now will not meet the same candidate, so it may only be
    /// recorded once the drain that removed the candidate has appended and the
    /// resulting owner is installed.
    pub(crate) fn record_marker_settled(&mut self, refused_epoch: u64) {
        self.settled_epochs.push(refused_epoch);
    }

    /// Records that a candidate-lane head left the lane AFTER its drain row is
    /// durable and the resulting owner is installed.
    ///
    /// Same ordering rule as [`Self::record_marker_settled`]: the tell wakes a
    /// parked connection fate to retry its terminal admission, which must not
    /// meet the candidate this drain removed.
    pub(crate) const fn record_candidate_lane_drained(&mut self) {
        self.candidate_lane_drained = true;
    }
}