Struct ObserverRecoveryAggregate 

pub struct ObserverRecoveryAggregate { /* private fields */ }

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.

use liminal_protocol::lifecycle::ObserverRecoveryAggregate;

fn require_clone<T: Clone>() {}
require_clone::<ObserverRecoveryAggregate>();

Implementations

impl ObserverRecoveryAggregate

pub const fn new() -> Self

Creates an empty aggregate tracking no conversations.

pub fn restore( progress_rows: &[(ConversationId, DeliverySeq)], armed_rows: &[(ConversationId, ObserverEpoch)], ) -> Result<Self, ObserverRecoveryAggregateRestoreError>

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 observer_progress( &self, conversation_id: ConversationId, ) -> Option<DeliverySeq>

Returns the current hard observer progress for one conversation.

pub fn armed_epoch( &self, conversation_id: ConversationId, ) -> Option<ObserverEpoch>

Returns the installed equal-epoch arm for one conversation, if any.

pub fn progress_rows(&self) -> Vec<(ConversationId, DeliverySeq)>

Returns every durable progress row in conversation order.

pub fn armed_rows(&self) -> Vec<(ConversationId, ObserverEpoch)>

Returns every installed arm row in conversation order.

pub fn decide_track( self, conversation_id: ConversationId, observer_progress: DeliverySeq, ) -> ObserverProgressTrackDecision

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).

pub fn decide_progress_advance( self, conversation_id: ConversationId, presented_progress: DeliverySeq, ) -> ObserverProgressAdvanceDecision

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).

pub fn decide_recovery( self, request: &ObserverRecoveryHandshake, max_entries: u64, connection_conversation_limit: u64, tracked_conversations: &[ConversationId], ) -> ObserverRecoveryTransactionDecision

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.

Trait Implementations

impl Debug for ObserverRecoveryAggregate

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for ObserverRecoveryAggregate

fn default() -> ObserverRecoveryAggregate

Returns the “default value” for a type.

impl Eq for ObserverRecoveryAggregate

impl PartialEq for ObserverRecoveryAggregate

fn eq(&self, other: &ObserverRecoveryAggregate) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl StructuralPartialEq for ObserverRecoveryAggregate