liminal_protocol/lifecycle/

observer_recovery.rs

use alloc::collections::{BTreeMap, BTreeSet};
use alloc::vec::Vec;

use crate::wire::{
    ConversationId, DeliverySeq, InvalidObserverEpoch, InvalidObserverEpochList, ObserverEpoch,
    ObserverProgressStatus, ObserverRecoveryAccepted, ObserverRecoveryHandshake,
    ObserverRecoveryResponse,
};

#[path = "observer_recovery_errors.rs"]
mod errors;
pub use errors::{
    ObserverProgressAdvanceError, ObserverProgressTrackError, ObserverRecoveryAggregateRestoreError,
};

#[path = "observer_progress_projection.rs"]
mod projection;
pub use projection::ObserverProgressProjection;

#[path = "observer_recovery_retire.rs"]
mod retire;
pub use retire::{
    ObserverProgressRetireDecision, ObserverProgressRetireError, ObserverProgressRetireTransaction,
};

#[path = "observer_recovery_transactions.rs"]
mod transactions;
pub use transactions::{
    ObserverProgressAdvanceDecision, ObserverProgressAdvanceTransaction,
    ObserverProgressTrackDecision, ObserverProgressTrackTransaction, ObserverRecoveryArm,
    ObserverRecoveryCommit, ObserverRecoveryTransaction, ObserverRecoveryTransactionDecision,
};

/// Exclusively owned observer-recovery aggregate: per-conversation hard
/// observer progress plus every installed equal-epoch arm, as ONE owned unit.
///
/// This is the A4 transactional surface
/// (`docs/design/LP-GAP-CLOSURE-GOAL.md`): the equal-epoch progress read and
/// the arm installation share one serialization boundary *by construction*,
/// because [`Self::decide_recovery`] consumes the aggregate and only
/// [`ObserverRecoveryTransaction::commit`] or
/// [`ObserverRecoveryTransaction::abort`] returns it. No acknowledgement or
/// binding fate can advance progress between the read and the installation,
/// and a crash while the transaction is pending installs nothing.
///
/// The acknowledgement/binding-fate feed carries the same barrier:
/// [`Self::decide_progress_advance`] consumes the aggregate and only
/// [`ObserverProgressAdvanceTransaction::commit`] applies the progress write
/// and surrenders the fired arm — after the caller's durable append is
/// confirmed — while [`ObserverProgressAdvanceTransaction::abort`] returns
/// the aggregate byte-for-byte unchanged, arm still installed. So does the
/// registration feed: [`Self::decide_track`] consumes the aggregate and only
/// [`ObserverProgressTrackTransaction::commit`] installs the new progress
/// row, after the caller's durable append is confirmed, so a recovery batch
/// can never plan an arm against a progress row that is not yet durable. No
/// mutation in this module is exempt from the decide/commit/abort
/// discipline, so live and durable state can never disagree about progress
/// or an installed arm.
///
/// The aggregate is deliberately not `Clone`: at most one owner may read
/// progress for arm selection.
///
/// ```compile_fail
/// use liminal_protocol::lifecycle::ObserverRecoveryAggregate;
///
/// fn require_clone<T: Clone>() {}
/// require_clone::<ObserverRecoveryAggregate>();
/// ```
#[derive(Debug, Default, PartialEq, Eq)]
pub struct ObserverRecoveryAggregate {
    progress: BTreeMap<ConversationId, DeliverySeq>,
    armed: BTreeMap<ConversationId, ObserverEpoch>,
}

impl ObserverRecoveryAggregate {
    /// Creates an empty aggregate tracking no conversations.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            progress: BTreeMap::new(),
            armed: BTreeMap::new(),
        }
    }

    /// Rebuilds the aggregate from durable progress and arm rows.
    ///
    /// # Errors
    ///
    /// Returns [`ObserverRecoveryAggregateRestoreError`] for duplicate rows,
    /// an arm without its progress row, or an arm whose epoch is not the
    /// exact current progress of its conversation.
    pub fn restore(
        progress_rows: &[(ConversationId, DeliverySeq)],
        armed_rows: &[(ConversationId, ObserverEpoch)],
    ) -> Result<Self, ObserverRecoveryAggregateRestoreError> {
        let mut progress = BTreeMap::new();
        for (conversation_id, observer_progress) in progress_rows {
            if progress
                .insert(*conversation_id, *observer_progress)
                .is_some()
            {
                return Err(ObserverRecoveryAggregateRestoreError::DuplicateProgress {
                    conversation_id: *conversation_id,
                });
            }
        }
        let mut armed = BTreeMap::new();
        for (conversation_id, armed_epoch) in armed_rows {
            let Some(current_observer_progress) = progress.get(conversation_id).copied() else {
                return Err(ObserverRecoveryAggregateRestoreError::ArmWithoutProgress {
                    conversation_id: *conversation_id,
                });
            };
            if current_observer_progress != *armed_epoch {
                return Err(ObserverRecoveryAggregateRestoreError::ArmEpochMismatch {
                    conversation_id: *conversation_id,
                    armed_epoch: *armed_epoch,
                    current_observer_progress,
                });
            }
            if armed.insert(*conversation_id, *armed_epoch).is_some() {
                return Err(ObserverRecoveryAggregateRestoreError::DuplicateArm {
                    conversation_id: *conversation_id,
                });
            }
        }
        Ok(Self { progress, armed })
    }

    /// Installs durable arm rows onto a restored aggregate, one arm at a time,
    /// with the checks of [`Self::restore`]: each arm's conversation must be
    /// tracked at exactly the arm's epoch, and an arm already installed at the
    /// same epoch is the same arm. The cost is per arm, never a rebuild of the
    /// whole aggregate, so a restore folds its tail in one pass.
    ///
    /// # Errors
    ///
    /// Returns [`ObserverRecoveryAggregateRestoreError`] for an arm without its
    /// progress row, an arm whose epoch is not its conversation's progress,
    /// or a second arm at a different epoch for an armed conversation.
    pub fn install_restored_arms(
        mut self,
        arms: &[(ConversationId, ObserverEpoch)],
    ) -> Result<Self, ObserverRecoveryAggregateRestoreError> {
        for (conversation_id, armed_epoch) in arms {
            let Some(current_observer_progress) = self.progress.get(conversation_id).copied()
            else {
                return Err(ObserverRecoveryAggregateRestoreError::ArmWithoutProgress {
                    conversation_id: *conversation_id,
                });
            };
            if current_observer_progress != *armed_epoch {
                return Err(ObserverRecoveryAggregateRestoreError::ArmEpochMismatch {
                    conversation_id: *conversation_id,
                    armed_epoch: *armed_epoch,
                    current_observer_progress,
                });
            }
            match self.armed.get(conversation_id) {
                Some(installed) if installed == armed_epoch => {}
                Some(_) => {
                    return Err(ObserverRecoveryAggregateRestoreError::DuplicateArm {
                        conversation_id: *conversation_id,
                    });
                }
                None => {
                    self.armed.insert(*conversation_id, *armed_epoch);
                }
            }
        }
        Ok(self)
    }

    /// Returns the current hard observer progress for one conversation.
    #[must_use]
    pub fn observer_progress(&self, conversation_id: ConversationId) -> Option<DeliverySeq> {
        self.progress.get(&conversation_id).copied()
    }

    /// Returns the installed equal-epoch arm for one conversation, if any.
    #[must_use]
    pub fn armed_epoch(&self, conversation_id: ConversationId) -> Option<ObserverEpoch> {
        self.armed.get(&conversation_id).copied()
    }

    /// Returns every durable progress row in conversation order.
    #[must_use]
    pub fn progress_rows(&self) -> Vec<(ConversationId, DeliverySeq)> {
        self.progress
            .iter()
            .map(|(conversation_id, observer_progress)| (*conversation_id, *observer_progress))
            .collect()
    }

    /// Returns every installed arm row in conversation order.
    #[must_use]
    pub fn armed_rows(&self) -> Vec<(ConversationId, ObserverEpoch)> {
        self.armed
            .iter()
            .map(|(conversation_id, armed_epoch)| (*conversation_id, *armed_epoch))
            .collect()
    }

    /// Consumes the aggregate into one registration-track transaction.
    ///
    /// This is the registration feed for newly tracked conversations, and it
    /// carries the same barrier as the other two mutations. Validation
    /// happens here — an already tracked conversation refuses with the
    /// aggregate unchanged — and a validated registration returns a pending
    /// [`ObserverProgressTrackTransaction`] carrying the new progress row for
    /// the caller's durable append. Nothing mutates until
    /// [`ObserverProgressTrackTransaction::commit`] confirms the append;
    /// [`ObserverProgressTrackTransaction::abort`] returns the aggregate
    /// byte-for-byte unchanged, conversation still untracked. A recovery
    /// batch therefore can never read a progress row whose durable append
    /// has not been confirmed, so a crash after an arm install can never
    /// strand a durable arm row without its progress row
    /// ([`ObserverRecoveryAggregateRestoreError::ArmWithoutProgress`]).
    #[must_use]
    pub fn decide_track(
        self,
        conversation_id: ConversationId,
        observer_progress: DeliverySeq,
    ) -> ObserverProgressTrackDecision {
        if self.progress.contains_key(&conversation_id) {
            return ObserverProgressTrackDecision::Refuse {
                aggregate: self,
                error: ObserverProgressTrackError::AlreadyTracked { conversation_id },
            };
        }
        ObserverProgressTrackDecision::Commit(ObserverProgressTrackTransaction {
            aggregate: self,
            conversation_id,
            observer_progress,
        })
    }

    /// Consumes the aggregate into one progress-advance transaction.
    ///
    /// This is the acknowledgement/binding-fate feed. Validation happens
    /// here — unknown conversation and non-advancing progress refuse with
    /// the aggregate unchanged — and a validated advance returns a pending
    /// [`ObserverProgressAdvanceTransaction`] carrying the new progress row
    /// and the fired-arm plan for the caller's durable append. Nothing
    /// mutates until [`ObserverProgressAdvanceTransaction::commit`] confirms
    /// the append; [`ObserverProgressAdvanceTransaction::abort`] returns the
    /// aggregate byte-for-byte unchanged. Every installed arm is equal-epoch
    /// with its conversation's progress, so a strictly advancing write always
    /// fires the installed arm: the fired arm is surrendered by `commit` so
    /// the caller wakes the parked rows in the same durable transaction
    /// (LAW-1: wake on the event, never poll).
    #[must_use]
    pub fn decide_progress_advance(
        self,
        conversation_id: ConversationId,
        presented_progress: DeliverySeq,
    ) -> ObserverProgressAdvanceDecision {
        let Some(current) = self.progress.get(&conversation_id).copied() else {
            return ObserverProgressAdvanceDecision::Refuse {
                aggregate: self,
                error: ObserverProgressAdvanceError::ConversationUnknown { conversation_id },
            };
        };
        if presented_progress <= current {
            return ObserverProgressAdvanceDecision::Refuse {
                aggregate: self,
                error: ObserverProgressAdvanceError::NotAdvancing {
                    conversation_id,
                    current_observer_progress: current,
                    presented_progress,
                },
            };
        }
        let fired = self
            .armed
            .get(&conversation_id)
            .copied()
            .map(|refused_epoch| ObserverRecoveryArm {
                conversation_id,
                refused_epoch,
            });
        ObserverProgressAdvanceDecision::Commit(ObserverProgressAdvanceTransaction {
            aggregate: self,
            conversation_id,
            presented_progress,
            fired,
        })
    }

    /// Consumes the aggregate into one observer-recovery transaction.
    ///
    /// This is the only public door to arm selection: it binds the
    /// crate-internal selection's progress reads to the owned aggregate so
    /// the read and the arm installation are one owned unit by construction.
    /// A refused batch returns the aggregate unchanged alongside the exact
    /// refusal response.
    #[must_use]
    pub fn decide_recovery(
        self,
        request: &ObserverRecoveryHandshake,
        max_entries: u64,
        connection_conversation_limit: u64,
        tracked_conversations: &[ConversationId],
    ) -> ObserverRecoveryTransactionDecision {
        let decision = apply_observer_recovery(
            request,
            max_entries,
            connection_conversation_limit,
            tracked_conversations,
            |conversation_id| self.progress.get(&conversation_id).copied(),
        );
        match decision {
            ObserverRecoveryDecision::Respond(response) => {
                ObserverRecoveryTransactionDecision::Respond {
                    aggregate: self,
                    response,
                }
            }
            ObserverRecoveryDecision::Commit(commit) => {
                ObserverRecoveryTransactionDecision::Commit(ObserverRecoveryTransaction {
                    aggregate: self,
                    commit,
                })
            }
        }
    }
}

/// Total observer-recovery decision.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ObserverRecoveryDecision {
    /// Validation or connection-capacity preflight refused the whole batch.
    Respond(ObserverRecoveryResponse),
    /// Every entry validated and the complete arm plan may commit atomically.
    Commit(ObserverRecoveryCommit),
}

fn wire_count(value: usize) -> u64 {
    u64::try_from(value).unwrap_or(u64::MAX)
}

/// Applies the observer-recovery list, capacity, and epoch precedence.
///
/// Validation order is the frozen R-D1 order: list length, duplicate
/// conversation, request-index connection capacity, then request-index unknown
/// or ahead epoch. Only after every entry passes is an arm plan produced.
/// `observer_progress` must return the current hard observer progress for a
/// known conversation and `None` for an unknown conversation.
///
/// This raw selection is deliberately crate-internal: its progress reads and
/// the installation of every returned arm must share one serialization
/// boundary (releasing it between evaluation and installation would violate
/// the equal-epoch subscribe-then-snapshot rule), and that requirement is
/// enforceable only by ownership.
/// [`ObserverRecoveryAggregate::decide_recovery`] is the one public door to
/// arm selection: it binds these progress reads to the owned aggregate, so
/// no caller can interleave an acknowledgement or fate advance between
/// evaluation and installation.
#[must_use]
pub(super) fn apply_observer_recovery<F>(
    request: &ObserverRecoveryHandshake,
    max_entries: u64,
    connection_conversation_limit: u64,
    tracked_conversations: &[ConversationId],
    mut observer_progress: F,
) -> ObserverRecoveryDecision
where
    F: FnMut(ConversationId) -> Option<DeliverySeq>,
{
    let presented_entries = wire_count(request.observer_refusals.len());
    if presented_entries > max_entries {
        return ObserverRecoveryDecision::Respond(
            ObserverRecoveryResponse::invalid_observer_epoch_list(
                InvalidObserverEpochList::TooManyEntries {
                    presented_entries,
                    max_entries,
                },
            ),
        );
    }

    let mut first_indices = BTreeMap::new();
    for (index, refusal) in request.observer_refusals.iter().enumerate() {
        let request_index = wire_count(index);
        if let Some(first_index) = first_indices.insert(refusal.conversation_id, request_index) {
            return ObserverRecoveryDecision::Respond(
                ObserverRecoveryResponse::invalid_observer_epoch_list(
                    InvalidObserverEpochList::DuplicateConversation {
                        conversation_id: refusal.conversation_id,
                        first_index,
                        duplicate_index: request_index,
                    },
                ),
            );
        }
    }

    let mut tracked: BTreeSet<_> = tracked_conversations.iter().copied().collect();
    for refusal in &request.observer_refusals {
        if tracked.contains(&refusal.conversation_id) {
            continue;
        }
        let occupied = wire_count(tracked.len());
        if occupied >= connection_conversation_limit {
            return ObserverRecoveryDecision::Respond(
                ObserverRecoveryResponse::connection_capacity_exceeded(
                    refusal.conversation_id,
                    connection_conversation_limit,
                ),
            );
        }
        tracked.insert(refusal.conversation_id);
    }

    let mut validated = Vec::with_capacity(request.observer_refusals.len());
    for refusal in &request.observer_refusals {
        let Some(current_observer_progress) = observer_progress(refusal.conversation_id) else {
            return ObserverRecoveryDecision::Respond(
                ObserverRecoveryResponse::invalid_observer_epoch(
                    InvalidObserverEpoch::ConversationUnknown {
                        conversation_id: refusal.conversation_id,
                        presented_epoch: refusal.refused_epoch,
                    },
                ),
            );
        };
        if refusal.refused_epoch > current_observer_progress {
            return ObserverRecoveryDecision::Respond(
                ObserverRecoveryResponse::invalid_observer_epoch(
                    InvalidObserverEpoch::EpochAhead {
                        conversation_id: refusal.conversation_id,
                        presented_epoch: refusal.refused_epoch,
                        current_observer_progress,
                    },
                ),
            );
        }
        validated.push((refusal, current_observer_progress));
    }

    let mut arms = Vec::new();
    let mut statuses = Vec::with_capacity(validated.len());
    for (refusal, current_observer_progress) in validated {
        let armed = refusal.refused_epoch == current_observer_progress;
        if armed {
            arms.push(ObserverRecoveryArm {
                conversation_id: refusal.conversation_id,
                refused_epoch: refusal.refused_epoch,
            });
        }
        statuses.push(ObserverProgressStatus {
            conversation_id: refusal.conversation_id,
            refused_epoch: refusal.refused_epoch,
            current_observer_progress,
            armed,
            progressed: !armed,
        });
    }

    ObserverRecoveryDecision::Commit(ObserverRecoveryCommit {
        arms,
        outcome: ObserverRecoveryAccepted { statuses },
    })
}