liminal_protocol/lifecycle/
observer_progress_projection.rs

1//! The exact observer-progress projection a committed source emits, moved out of
2//! `observer_recovery.rs` unchanged so that file stays within the 500-line bound.
3
4use crate::wire::{ConversationId, DeliverySeq};
5
6/// Move-only exact observer-progress projection emitted by a committed source.
7///
8/// Participant acknowledgement and binding-ending transitions own the progress
9/// value. Consuming code can persist and publish this projection but cannot
10/// construct one from a guessed maximum or record-delivery observation.
11///
12/// ```compile_fail
13/// use liminal_protocol::lifecycle::ObserverProgressProjection;
14///
15/// fn require_clone<T: Clone>() {}
16/// require_clone::<ObserverProgressProjection>();
17/// ```
18#[derive(Debug, Eq)]
19pub struct ObserverProgressProjection {
20    conversation_id: ConversationId,
21    new_observer_progress: DeliverySeq,
22    source: Option<crate::lifecycle::checkpoint_codec::ObserverProjectionSource>,
23}
24
25impl PartialEq for ObserverProgressProjection {
26    fn eq(&self, other: &Self) -> bool {
27        self.conversation_id == other.conversation_id
28            && self.new_observer_progress == other.new_observer_progress
29    }
30}
31
32impl ObserverProgressProjection {
33    #[cfg(any(test, feature = "test-support"))]
34    pub(in crate::lifecycle) const fn new(
35        conversation_id: ConversationId,
36        new_observer_progress: DeliverySeq,
37    ) -> Self {
38        Self {
39            conversation_id,
40            new_observer_progress,
41            source: None,
42        }
43    }
44
45    pub(in crate::lifecycle) const fn sourced(
46        conversation_id: ConversationId,
47        new_observer_progress: DeliverySeq,
48        source: crate::lifecycle::checkpoint_codec::ObserverProjectionSource,
49    ) -> Self {
50        Self::with_checkpoint_source(conversation_id, new_observer_progress, Some(source))
51    }
52
53    pub(in crate::lifecycle) const fn with_checkpoint_source(
54        conversation_id: ConversationId,
55        new_observer_progress: DeliverySeq,
56        source: Option<crate::lifecycle::checkpoint_codec::ObserverProjectionSource>,
57    ) -> Self {
58        Self {
59            conversation_id,
60            new_observer_progress,
61            source,
62        }
63    }
64
65    /// Serializes the original native producer retained by this projection.
66    ///
67    /// # Errors
68    /// Refuses a projection without native producer provenance.
69    #[doc(hidden)]
70    pub fn encode_checkpoint_source(
71        &self,
72    ) -> Result<alloc::vec::Vec<u8>, crate::lifecycle::OwnerCheckpointRestoreError> {
73        crate::lifecycle::checkpoint_codec::encode_observer_source(
74            self.conversation_id,
75            self.new_observer_progress,
76            self.source
77                .ok_or(crate::lifecycle::OwnerCheckpointRestoreError::Field(
78                    "observer_projection.native_source",
79                ))?,
80        )
81    }
82
83    /// Returns the participant named by the original native projection source.
84    #[doc(hidden)]
85    #[must_use]
86    pub const fn checkpoint_source_participant_id(&self) -> Option<crate::wire::ParticipantId> {
87        match self.source {
88            Some(source) => Some(source.participant_id()),
89            None => None,
90        }
91    }
92
93    /// Returns the binding epoch retained by the native source, when it ended one.
94    #[doc(hidden)]
95    #[must_use]
96    pub const fn checkpoint_source_binding_epoch(&self) -> Option<crate::wire::BindingEpoch> {
97        match self.source {
98            Some(source) => source.binding_epoch(),
99            None => None,
100        }
101    }
102
103    /// Returns the native producer class used to associate saved witness metadata.
104    #[doc(hidden)]
105    #[must_use]
106    pub const fn checkpoint_source_kind(
107        &self,
108    ) -> Option<crate::lifecycle::checkpoint_codec::ObserverProjectionCheckpointKind> {
109        match self.source {
110            Some(source) => Some(source.kind()),
111            None => None,
112        }
113    }
114
115    /// Builds a projection no transition produced, for a test that must
116    /// present a source a real transition never would.
117    ///
118    /// Absent from production builds (`test-support` / `cfg(test)` only).
119    #[cfg(any(test, feature = "test-support"))]
120    #[must_use]
121    pub const fn for_test(
122        conversation_id: ConversationId,
123        new_observer_progress: DeliverySeq,
124    ) -> Self {
125        Self::new(conversation_id, new_observer_progress)
126    }
127
128    /// Returns the conversation whose hard observer progress may advance.
129    #[must_use]
130    pub const fn conversation_id(&self) -> ConversationId {
131        self.conversation_id
132    }
133
134    /// Returns the exact protocol-produced hard observer progress.
135    #[must_use]
136    pub const fn new_observer_progress(&self) -> DeliverySeq {
137        self.new_observer_progress
138    }
139}
140
141impl ObserverProgressProjection {
142    /// Borrows the source conversation and exact hard progress into canonical bytes.
143    #[must_use]
144    pub fn encode_checkpoint(&self) -> [u8; 16] {
145        let mut body = [0; 16];
146        body[..8].copy_from_slice(&self.conversation_id.to_be_bytes());
147        body[8..].copy_from_slice(&self.new_observer_progress.to_be_bytes());
148        body
149    }
150
151    /// Reassociates saved bytes with their protocol-produced committed source.
152    ///
153    /// The validated source moves into the result; this does not mint a projection.
154    ///
155    /// # Errors
156    ///
157    /// Refuses any framing, conversation, or exact progress mismatch.
158    pub fn restore_checkpoint(
159        body: &[u8],
160        source: Self,
161    ) -> Result<Self, crate::lifecycle::OwnerCheckpointRestoreError> {
162        if body != source.encode_checkpoint() {
163            return Err(crate::lifecycle::OwnerCheckpointRestoreError::Field(
164                "observer_projection.source",
165            ));
166        }
167        Ok(source)
168    }
169}
liminal_protocol/lifecycle/observer_progress_projection.rs

liminal_protocol/lifecycle/
observer_progress_projection.rs
