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
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
use super::{
Channel, FlowError, Operation, PipelineLimits, Prefix, Report, VecDeque, batch, ledger,
};
#[cfg(test)]
mod tests;
/// One peer's absolute send reservations and bounded sent-but-unreceipted ledger.
///
/// No payload ownership or quorum state lives here. Keep one instance per peer;
/// a slow peer cannot consume another peer's local capacity. Updates allocate no memory.
#[derive(Debug)]
pub struct Sender {
report: Report,
sent: Prefix,
sent_bytes: u64,
pending: VecDeque<Operation>,
limits: PipelineLimits,
}
impl Sender {
/// Open after a fresh correlated exchange with the bound peer and epoch.
///
/// `verified_received` must come from exact local-history lookup, not a copied
/// untrusted report field. This comparison grants no election or commit authority.
/// Opening never implicitly replaces another instance; the adapter must fence
/// its old packets and status correlation before adopting a new epoch.
/// `limits` bounds local sent-but-unreceipted metadata/body ownership, not
/// the remote receive window. A larger remote window never enlarges it.
pub fn open(
report: Report,
limits: PipelineLimits,
verified_received: Prefix,
) -> Result<Self, FlowError> {
report.validate_shape()?;
if report.received != verified_received {
return Err(FlowError::History);
}
Ok(Self::from_checked_report(report, limits, ledger(limits)?))
}
pub(super) fn from_checked_report(
report: Report,
limits: PipelineLimits,
pending: VecDeque<Operation>,
) -> Self {
Self {
report,
sent: report.received,
sent_bytes: report.received_bytes,
pending,
limits,
}
}
/// Replace an explicitly retired channel after a fresh, history-verified open.
///
/// Preserves startup metadata capacity instead of allocating another ledger.
/// The adapter must fence old payload ownership, validate probe/session/scope
/// correlation, and independently verify both reported base and receipt history.
/// `verified_received` has the same requirements as in `open`.
/// Same-channel status belongs in `observe`; it must never reset send charges.
pub fn reopen(&mut self, report: Report, verified_received: Prefix) -> Result<(), FlowError> {
if report.channel == self.report.channel {
return Err(FlowError::Channel);
}
report.validate_shape()?;
if report.received != verified_received {
return Err(FlowError::History);
}
self.pending.clear();
self.report = report;
self.sent = report.received;
self.sent_bytes = report.received_bytes;
Ok(())
}
/// Advance beyond PEER reservations after independently checking a PUB receipt
/// against local history. Counters are receiver-owned, just as at channel open.
/// Existing reservations are retired only when the verified prefix covers them.
pub(super) fn observe_publication(
&mut self,
report: Report,
verified: Prefix,
) -> Result<bool, FlowError> {
if report.received != verified {
return Err(FlowError::History);
}
if report.received.op <= self.sent.op {
return self.observe(report);
}
report.validate_shape()?;
if report.channel != self.report.channel {
return Err(FlowError::Channel);
}
if report.base != self.report.base {
return Err(FlowError::History);
}
if report.revision < self.report.revision {
return Ok(false);
}
if report.revision == self.report.revision || report.received_bytes <= self.sent_bytes {
return Err(FlowError::Report);
}
self.pending.clear();
self.report = report;
self.sent = report.received;
self.sent_bytes = report.received_bytes;
Ok(true)
}
/// Latest verified receiver accounting, including the channel's fixed base.
pub const fn report(&self) -> Report {
self.report
}
/// Exact channel to attach to payloads and check on incoming reports.
pub const fn channel(&self) -> Channel {
self.report.channel
}
/// Highest PEER reservation or independently verified PUB receipt.
/// Not evidence of transport submission, persistence, or a quorum.
pub const fn sent(&self) -> Prefix {
self.sent
}
/// Latest verified volatile receipt. Never use as a durable quorum ACK.
pub const fn received(&self) -> Prefix {
self.report.received
}
/// Remaining local outstanding capacity.
/// May be zero; not a reservation. Reports cannot grow the local ledger.
pub fn available(&self) -> PipelineLimits {
PipelineLimits {
max_operations: self.limits.max_operations - self.pending.len(),
max_body_bytes: self.limits.max_body_bytes
- (self.sent_bytes - self.report.received_bytes) as usize,
}
}
/// Bounded outstanding metadata, in canonical order, for probe/repair planning.
/// Retransmission reuses these existing reservations rather than calling `record_send`.
pub fn outstanding(&self) -> impl ExactSizeIterator<Item = &Operation> {
self.pending.iter()
}
/// Charge local metadata once for a new contiguous suffix in the send path.
///
/// Reserve local payload/outbox capacity first. After success the adapter must
/// retain the transmission across local backpressure or fence the whole epoch.
/// Do not call again for a retry, duplicate, or packet still queued locally.
pub fn record_send(
&mut self,
channel: Channel,
operations: &[Operation],
) -> Result<(), FlowError> {
if channel != self.report.channel {
return Err(FlowError::Channel);
}
if operations.len() > self.limits.max_operations - self.pending.len() {
return Err(FlowError::Capacity);
}
let (end, bytes) = batch(operations, self.sent)?;
let total = self
.sent_bytes
.checked_add(bytes)
.ok_or(FlowError::Exhausted)?;
if total - self.report.received_bytes > self.limits.max_body_bytes as u64 {
return Err(FlowError::Capacity);
}
self.pending.extend(operations.iter().copied());
self.sent = end;
self.sent_bytes = total;
Ok(())
}
/// Incorporate same-epoch receipt progress. Return false for stale or exact repeats.
///
/// Check the exact receipt boundary and cumulative bytes against reserved sends
/// before freeing metadata. Conflicting same-revision reports and all retractions
/// fail without changing state. Epoch changes require a separate correlated open.
pub fn observe(&mut self, report: Report) -> Result<bool, FlowError> {
if report.channel != self.report.channel {
return Err(FlowError::Channel);
}
report.validate_shape()?;
if report.base != self.report.base {
return Err(FlowError::History);
}
if report.revision < self.report.revision {
return Ok(false);
}
if report.revision == self.report.revision {
return if report == self.report {
Ok(false)
} else {
Err(FlowError::Report)
};
}
if report.received.op < self.report.received.op
|| report.received_bytes < self.report.received_bytes
{
return Err(FlowError::Report);
}
let mut end = self.report.received;
let mut bytes = self.report.received_bytes;
let mut count = 0;
for operation in &self.pending {
if operation.prefix.op > report.received.op {
break;
}
end = operation.prefix;
bytes = bytes
.checked_add(operation.body_bytes)
.ok_or(FlowError::Exhausted)?;
count += 1;
}
if end != report.received || bytes != report.received_bytes {
return Err(FlowError::History);
}
self.pending.drain(..count);
self.report = report;
Ok(true)
}
}