liminal_protocol/lifecycle/

observer_recovery_transactions.rs

//! The ownership barriers between an observer-recovery decision and its commit, moved out of
//! `observer_recovery.rs` unchanged so that file stays within the 500-line bound.

use alloc::vec::Vec;

use super::{ObserverProgressAdvanceError, ObserverProgressTrackError, ObserverRecoveryAggregate};
use crate::wire::{
    ConversationId, DeliverySeq, ObserverEpoch, ObserverRecoveryAccepted, ObserverRecoveryResponse,
};

/// Total transactional observer-recovery decision against the owned aggregate.
#[derive(Debug, PartialEq, Eq)]
pub enum ObserverRecoveryTransactionDecision {
    /// The whole batch was refused; the aggregate is unchanged.
    Respond {
        /// Unchanged aggregate returned to its owner.
        aggregate: ObserverRecoveryAggregate,
        /// Exact refusal response.
        response: ObserverRecoveryResponse,
    },
    /// Every entry validated; the arm plan may commit atomically.
    Commit(ObserverRecoveryTransaction),
}

/// Ownership barrier between arm selection and atomic arm installation.
///
/// The aggregate is unreachable while the transaction is pending, so no
/// progress can advance between the equal-epoch read and the installation.
/// Consuming [`Self::commit`] installs every arm of the plan at once — a
/// subset is unrepresentable — and consuming [`Self::abort`] installs none,
/// so a crash at any point leaves either the complete arm plan or no arm at
/// all, never a partially-armed request.
#[derive(Debug, PartialEq, Eq)]
pub struct ObserverRecoveryTransaction {
    pub(super) aggregate: ObserverRecoveryAggregate,
    pub(super) commit: ObserverRecoveryCommit,
}

impl ObserverRecoveryTransaction {
    /// Borrows the complete equal-epoch arm plan for the durable append.
    #[must_use]
    pub fn arms(&self) -> &[ObserverRecoveryArm] {
        self.commit.arms()
    }

    /// Borrows the exact request-ordered success response.
    #[must_use]
    pub const fn outcome(&self) -> &ObserverRecoveryAccepted {
        self.commit.outcome()
    }

    /// Installs the whole arm plan after a confirmed durable append.
    ///
    /// Installation is idempotent: replaying a durable recovery against the
    /// post-state reinstalls the identical equal-epoch arms, so at most one
    /// arm per conversation ever exists and crash replay converges.
    #[must_use]
    pub fn commit(mut self) -> (ObserverRecoveryAggregate, ObserverRecoveryAccepted) {
        let (arms, outcome) = self.commit.into_parts();
        for arm in arms {
            self.aggregate
                .armed
                .insert(arm.conversation_id(), arm.refused_epoch());
        }
        (self.aggregate, outcome)
    }

    /// Cancels a failed durable append; not one arm is installed.
    #[must_use]
    pub fn abort(self) -> ObserverRecoveryAggregate {
        self.aggregate
    }
}

/// Total transactional progress-advance decision against the owned aggregate.
#[derive(Debug, PartialEq, Eq)]
pub enum ObserverProgressAdvanceDecision {
    /// The advance was refused; the aggregate is unchanged.
    Refuse {
        /// Unchanged aggregate returned to its owner.
        aggregate: ObserverRecoveryAggregate,
        /// Exact refusal.
        error: ObserverProgressAdvanceError,
    },
    /// The advance validated; the progress write and arm fire may commit
    /// atomically.
    Commit(ObserverProgressAdvanceTransaction),
}

/// Ownership barrier between a validated progress advance and its durable
/// append.
///
/// The aggregate is unreachable while the transaction is pending, so no
/// recovery batch can read the not-yet-durable progress and no other advance
/// can race the fired-arm plan. Consuming [`Self::commit`] applies the
/// progress write and removes the fired arm as ONE mutation — after the
/// caller has confirmed the durable append of the progress row, the arm-row
/// deletion, and the wake — and consuming [`Self::abort`] returns the
/// aggregate byte-for-byte unchanged, arm still installed, so a failed
/// append never leaves live progress ahead of durable state or a
/// phantom-fired arm.
#[derive(Debug, PartialEq, Eq)]
pub struct ObserverProgressAdvanceTransaction {
    pub(super) aggregate: ObserverRecoveryAggregate,
    pub(super) conversation_id: ConversationId,
    pub(super) presented_progress: DeliverySeq,
    pub(super) fired: Option<ObserverRecoveryArm>,
}

impl ObserverProgressAdvanceTransaction {
    /// Returns the conversation whose progress row the durable append writes.
    #[must_use]
    pub const fn conversation_id(&self) -> ConversationId {
        self.conversation_id
    }

    /// Returns the new hard observer progress for the durable append.
    #[must_use]
    pub const fn presented_progress(&self) -> DeliverySeq {
        self.presented_progress
    }

    /// Returns the fired-arm plan for the durable append.
    ///
    /// `Some` names the installed arm whose durable row the same append must
    /// delete and whose parked rows the same append must wake; `None` means
    /// the conversation carried no arm.
    #[must_use]
    pub const fn fired_arm(&self) -> Option<ObserverRecoveryArm> {
        self.fired
    }

    /// Applies the advance after a confirmed durable append.
    ///
    /// The progress write and the arm removal are one mutation: the returned
    /// arm is exactly [`Self::fired_arm`], surrendered so the caller wakes
    /// the parked rows it durably committed to waking.
    #[must_use]
    pub fn commit(mut self) -> (ObserverRecoveryAggregate, Option<ObserverRecoveryArm>) {
        self.aggregate
            .progress
            .insert(self.conversation_id, self.presented_progress);
        if self.fired.is_some() {
            self.aggregate.armed.remove(&self.conversation_id);
        }
        (self.aggregate, self.fired)
    }

    /// Cancels a failed durable append; neither progress nor arm changes.
    #[must_use]
    pub fn abort(self) -> ObserverRecoveryAggregate {
        self.aggregate
    }
}

/// Total transactional registration decision against the owned aggregate.
#[derive(Debug, PartialEq, Eq)]
pub enum ObserverProgressTrackDecision {
    /// The registration was refused; the aggregate is unchanged.
    Refuse {
        /// Unchanged aggregate returned to its owner.
        aggregate: ObserverRecoveryAggregate,
        /// Exact refusal.
        error: ObserverProgressTrackError,
    },
    /// The registration validated; the progress row may commit atomically.
    Commit(ObserverProgressTrackTransaction),
}

/// Ownership barrier between a validated registration and its durable append.
///
/// The aggregate is unreachable while the transaction is pending, so no
/// recovery batch can plan an equal-epoch arm against a progress row whose
/// durable append is not yet confirmed. Consuming [`Self::commit`] installs
/// the new progress row — after the caller has confirmed the durable append —
/// and consuming [`Self::abort`] returns the aggregate byte-for-byte
/// unchanged, conversation still untracked, so a failed append never leaves
/// a live progress row that durable state does not hold.
#[derive(Debug, PartialEq, Eq)]
pub struct ObserverProgressTrackTransaction {
    pub(super) aggregate: ObserverRecoveryAggregate,
    pub(super) conversation_id: ConversationId,
    pub(super) observer_progress: DeliverySeq,
}

impl ObserverProgressTrackTransaction {
    /// Returns the conversation whose progress row the durable append writes.
    #[must_use]
    pub const fn conversation_id(&self) -> ConversationId {
        self.conversation_id
    }

    /// Returns the registered hard observer progress for the durable append.
    #[must_use]
    pub const fn observer_progress(&self) -> DeliverySeq {
        self.observer_progress
    }

    /// Installs the registered progress row after a confirmed durable append.
    #[must_use]
    pub fn commit(mut self) -> ObserverRecoveryAggregate {
        self.aggregate
            .progress
            .insert(self.conversation_id, self.observer_progress);
        self.aggregate
    }

    /// Cancels a failed durable append; the conversation stays untracked.
    #[must_use]
    pub fn abort(self) -> ObserverRecoveryAggregate {
        self.aggregate
    }
}

/// One equal observer epoch that must be armed by an accepted recovery batch.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ObserverRecoveryArm {
    pub(super) conversation_id: ConversationId,
    pub(super) refused_epoch: ObserverEpoch,
}

impl ObserverRecoveryArm {
    /// Returns the conversation whose progress event must wake the parked rows.
    #[must_use]
    pub const fn conversation_id(self) -> ConversationId {
        self.conversation_id
    }

    /// Returns the exact refusal epoch being armed.
    #[must_use]
    pub const fn refused_epoch(self) -> ObserverEpoch {
        self.refused_epoch
    }
}

/// Whole-batch observer-recovery commit selected after exhaustive validation.
///
/// The arm list and response are produced together. A consumer must persist all
/// arms atomically before sending [`Self::outcome`], so no validation failure or
/// crash can expose a partially armed request.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ObserverRecoveryCommit {
    pub(super) arms: Vec<ObserverRecoveryArm>,
    pub(super) outcome: ObserverRecoveryAccepted,
}

impl ObserverRecoveryCommit {
    /// Borrows the complete equal-epoch arm plan in request order.
    #[must_use]
    pub fn arms(&self) -> &[ObserverRecoveryArm] {
        &self.arms
    }

    /// Borrows the exact request-ordered success response.
    #[must_use]
    pub const fn outcome(&self) -> &ObserverRecoveryAccepted {
        &self.outcome
    }

    /// Consumes the commit into the atomic arm plan and success response.
    #[must_use]
    pub fn into_parts(self) -> (Vec<ObserverRecoveryArm>, ObserverRecoveryAccepted) {
        (self.arms, self.outcome)
    }
}