liminal_protocol/wire/
push.rs

1use alloc::vec::Vec;
2
3use super::{
4    BindingEpoch, CloseCause, ConversationId, DeliverySeq, ObserverEpoch, ParticipantId,
5    PushDiscriminant, RecordKind, SettlementEpoch,
6};
7
8/// Causes valid only for a `Detached` lifecycle record.
9#[derive(Clone, Copy, Debug, PartialEq, Eq)]
10pub enum DetachedCause {
11    /// Clean deregistration.
12    CleanDeregister,
13    /// Binding supersession.
14    Superseded,
15    /// Orderly server shutdown.
16    ServerShutdown,
17}
18
19impl DetachedCause {
20    /// Converts to the shared close-cause registry without permitting Died-only causes.
21    #[must_use]
22    pub const fn close_cause(self) -> CloseCause {
23        match self {
24            Self::CleanDeregister => CloseCause::CleanDeregister,
25            Self::Superseded => CloseCause::Superseded,
26            Self::ServerShutdown => CloseCause::ServerShutdown,
27        }
28    }
29}
30
31/// Causes valid only for a `Died` lifecycle record.
32#[derive(Clone, Copy, Debug, PartialEq, Eq)]
33pub enum DiedCause {
34    /// Transport connection was lost.
35    ConnectionLost,
36    /// Participant process was killed.
37    ProcessKilled,
38    /// Participant protocol error.
39    ProtocolError,
40    /// Binding was recovered after an unclean restart.
41    UncleanServerRestart {
42        /// Server incarnation that previously owned the binding.
43        prior_server_incarnation: u64,
44    },
45}
46
47impl DiedCause {
48    /// Converts to the shared close-cause registry without permitting Detached-only causes.
49    #[must_use]
50    pub const fn close_cause(self) -> CloseCause {
51        match self {
52            Self::ConnectionLost => CloseCause::ConnectionLost,
53            Self::ProcessKilled => CloseCause::ProcessKilled,
54            Self::ProtocolError => CloseCause::ProtocolError,
55            Self::UncleanServerRestart {
56                prior_server_incarnation,
57            } => CloseCause::UncleanServerRestart {
58                prior_server_incarnation,
59            },
60        }
61    }
62}
63
64/// Exact record-kind body carried by `ParticipantDelivery`.
65#[derive(Clone, Debug, PartialEq, Eq)]
66pub enum ParticipantRecord {
67    /// Ordinary application record.
68    OrdinaryRecord {
69        /// Verified sender participant.
70        sender_participant_id: ParticipantId,
71        /// Opaque application payload.
72        payload: Vec<u8>,
73    },
74    /// Participant binding was attached.
75    Attached {
76        /// Affected participant.
77        affected_participant_id: ParticipantId,
78        /// New binding epoch.
79        binding_epoch: BindingEpoch,
80    },
81    /// Binding ended with a Detached-class cause.
82    Detached {
83        /// Affected participant.
84        affected_participant_id: ParticipantId,
85        /// Ended binding epoch.
86        binding_epoch: BindingEpoch,
87        /// Type-restricted Detached cause.
88        cause: DetachedCause,
89    },
90    /// Binding ended with a Died-class cause.
91    Died {
92        /// Affected participant.
93        affected_participant_id: ParticipantId,
94        /// Ended binding epoch.
95        binding_epoch: BindingEpoch,
96        /// Type-restricted Died cause.
97        cause: DiedCause,
98    },
99    /// Participant permanently left.
100    Left {
101        /// Affected participant.
102        affected_participant_id: ParticipantId,
103        /// Binding ended by the same Leave commit, if any.
104        ended_binding_epoch: Option<BindingEpoch>,
105    },
106    /// Retained history was explicitly abandoned and compacted.
107    HistoryCompacted {
108        /// Affected participant.
109        affected_participant_id: ParticipantId,
110        /// Last sequence known delivered before abandonment.
111        abandoned_after: DeliverySeq,
112        /// Last abandoned sequence.
113        abandoned_through: DeliverySeq,
114        /// Physical floor selected by the compaction decision.
115        physical_floor_at_decision: DeliverySeq,
116    },
117}
118
119impl ParticipantRecord {
120    /// Returns the explicit record-kind selector.
121    #[must_use]
122    pub const fn record_kind(&self) -> RecordKind {
123        match self {
124            Self::OrdinaryRecord { .. } => RecordKind::OrdinaryRecord,
125            Self::Attached { .. } => RecordKind::Attached,
126            Self::Detached { .. } => RecordKind::Detached,
127            Self::Died { .. } => RecordKind::Died,
128            Self::Left { .. } => RecordKind::Left,
129            Self::HistoryCompacted { .. } => RecordKind::HistoryCompacted,
130        }
131    }
132}
133
134/// Complete participant delivery push body (`0x0201`).
135///
136/// Delivery is at-least-once: the server may push the same
137/// `(conversation_id, delivery_seq)` more than once on one healthy connection
138/// with no transport loss — the publication scan decides offer count and
139/// interleave position, and neither carries application meaning. Every repeat
140/// of one pair
141/// is byte-identical to its first offer, so consumers deduplicate on the pair
142/// (participant contract R-C3, amendment A3, `«PUSH-REOFFER-SCHEDULE»`).
143#[derive(Clone, Debug, PartialEq, Eq)]
144pub struct ParticipantDelivery {
145    /// Conversation multiplexing key.
146    pub conversation_id: ConversationId,
147    /// Delivered record sequence.
148    pub delivery_seq: DeliverySeq,
149    /// Exact tagged record body.
150    pub record: ParticipantRecord,
151}
152
153/// Exhaustive pushed participant control/value.
154#[derive(Clone, Debug, PartialEq, Eq)]
155pub enum ServerPush {
156    /// Observer progress wake (`0x0200`).
157    ObserverProgressed {
158        /// Conversation whose observer advanced.
159        conversation_id: ConversationId,
160        /// Refusal epoch the progress may wake.
161        refused_epoch: ObserverEpoch,
162        /// Current observer progress.
163        observer_progress: DeliverySeq,
164    },
165    /// Participant record delivery (`0x0201`).
166    ParticipantDelivery(ParticipantDelivery),
167    /// Marker-settlement clearing wake (`0x0202`).
168    ///
169    /// Paired with the `MarkerSettlementBackpressure` refusal and delivered
170    /// CONNECTION-SCOPED: exactly to the connections that received that refusal
171    /// in this process lifetime, mirroring `ObserverProgressed`'s connection-level
172    /// delivery. A refused attach client holds no binding and therefore receives
173    /// no `ParticipantDelivery`, so without this wake the only honest client
174    /// behavior would be polling.
175    ///
176    /// It is NEVER sent to a connection refused at the enrollment wrapper: that
177    /// wrapper carries no membership predicate, so its refusal
178    /// (`EnrollmentSettlementBackpressure`) carries neither epoch nor wake
179    /// (participant contract §0.16 condition 2, enrollment wrapper).
180    MarkerSettled {
181        /// Conversation whose marker settlement cleared.
182        conversation_id: ConversationId,
183        /// Refusal epoch this clearing wakes; matched by the stage-11 retry
184        /// against the refusal's own `refused_epoch`.
185        refused_epoch: SettlementEpoch,
186    },
187}
188
189impl ServerPush {
190    /// Returns the stable push discriminant.
191    #[must_use]
192    pub const fn discriminant(&self) -> PushDiscriminant {
193        match self {
194            Self::ObserverProgressed { .. } => PushDiscriminant::ObserverProgressed,
195            Self::ParticipantDelivery(_) => PushDiscriminant::ParticipantDelivery,
196            Self::MarkerSettled { .. } => PushDiscriminant::MarkerSettled,
197        }
198    }
199}
liminal_protocol/wire/push.rs

liminal_protocol/wire/
push.rs
