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