liminal_protocol/lifecycle/
observer_recovery_retire.rs

1//! Retirement of one closed conversation from the observer-recovery aggregate
2//! (DUR-027 R2).
3//!
4//! A retired conversation's progress row is removed, so the aggregate holds
5//! live rows only and its snapshot does not grow with finished conversations.
6//! The aggregate keeps no retired set: what keeps a retired id from coming back
7//! is the server's tombstone, and an advance or arm presenting it afterwards is
8//! answered `ConversationUnknown`, as for any untracked conversation.
9
10use super::ObserverRecoveryAggregate;
11use crate::wire::{ConversationId, ObserverEpoch};
12
13/// Failure while retiring a conversation from the aggregate.
14#[derive(Clone, Copy, Debug, PartialEq, Eq)]
15pub enum ObserverProgressRetireError {
16    /// The conversation has no authoritative progress row.
17    ConversationUnknown {
18        /// Unknown conversation.
19        conversation_id: ConversationId,
20    },
21    /// An observer arm is installed on the conversation: an observer is still
22    /// waiting on it, and retiring would answer it by deleting what it waits on.
23    Armed {
24        /// Armed conversation.
25        conversation_id: ConversationId,
26        /// Epoch of the installed arm.
27        epoch: ObserverEpoch,
28    },
29}
30
31/// Total transactional retirement decision against the owned aggregate.
32#[derive(Debug, PartialEq, Eq)]
33pub enum ObserverProgressRetireDecision {
34    /// The retirement was refused; the aggregate is unchanged.
35    Refuse {
36        /// Unchanged aggregate returned to its owner.
37        aggregate: ObserverRecoveryAggregate,
38        /// Exact refusal.
39        error: ObserverProgressRetireError,
40    },
41    /// The retirement validated; the progress row goes when it commits.
42    Commit(ObserverProgressRetireTransaction),
43}
44
45/// Ownership barrier between a validated retirement and its durable append.
46///
47/// The aggregate is unreachable while the transaction is pending, so no arm
48/// can be installed on the conversation between the check and the removal.
49/// Consuming [`Self::commit`] removes the progress row after the caller has
50/// confirmed the durable retire row; consuming [`Self::abort`] returns the
51/// aggregate unchanged.
52#[derive(Debug, PartialEq, Eq)]
53pub struct ObserverProgressRetireTransaction {
54    aggregate: ObserverRecoveryAggregate,
55    conversation_id: ConversationId,
56}
57
58impl ObserverProgressRetireTransaction {
59    /// Returns the conversation whose retire row the durable append writes.
60    #[must_use]
61    pub const fn conversation_id(&self) -> ConversationId {
62        self.conversation_id
63    }
64
65    /// Removes the conversation's progress row after a confirmed durable append.
66    #[must_use]
67    pub fn commit(mut self) -> ObserverRecoveryAggregate {
68        self.aggregate.progress.remove(&self.conversation_id);
69        self.aggregate
70    }
71
72    /// Cancels a failed durable append; the row stays.
73    #[must_use]
74    pub fn abort(self) -> ObserverRecoveryAggregate {
75        self.aggregate
76    }
77}
78
79impl ObserverRecoveryAggregate {
80    /// Consumes the aggregate into one retirement transaction.
81    ///
82    /// An untracked conversation refuses `ConversationUnknown` and an armed one
83    /// refuses `Armed`, each with the aggregate unchanged. Nothing mutates until
84    /// [`ObserverProgressRetireTransaction::commit`].
85    #[must_use]
86    pub fn decide_retire(self, conversation_id: ConversationId) -> ObserverProgressRetireDecision {
87        if !self.progress.contains_key(&conversation_id) {
88            return ObserverProgressRetireDecision::Refuse {
89                aggregate: self,
90                error: ObserverProgressRetireError::ConversationUnknown { conversation_id },
91            };
92        }
93        if let Some(&epoch) = self.armed.get(&conversation_id) {
94            return ObserverProgressRetireDecision::Refuse {
95                aggregate: self,
96                error: ObserverProgressRetireError::Armed {
97                    conversation_id,
98                    epoch,
99                },
100            };
101        }
102        ObserverProgressRetireDecision::Commit(ObserverProgressRetireTransaction {
103            aggregate: self,
104            conversation_id,
105        })
106    }
107}
liminal_protocol/lifecycle/observer_recovery_retire.rs

liminal_protocol/lifecycle/
observer_recovery_retire.rs
