ozzy_replication/flow/
sender.rs

1use super::{
2    Channel, FlowError, Operation, PipelineLimits, Prefix, Report, VecDeque, batch, ledger,
3};
4
5#[cfg(test)]
6mod tests;
7
8/// One peer's absolute send reservations and bounded sent-but-unreceipted ledger.
9///
10/// No payload ownership or quorum state lives here. Keep one instance per peer;
11/// a slow peer cannot consume another peer's local capacity. Updates allocate no memory.
12#[derive(Debug)]
13pub struct Sender {
14    report: Report,
15    sent: Prefix,
16    sent_bytes: u64,
17    pending: VecDeque<Operation>,
18    limits: PipelineLimits,
19}
20
21impl Sender {
22    /// Open after a fresh correlated exchange with the bound peer and epoch.
23    ///
24    /// `verified_received` must come from exact local-history lookup, not a copied
25    /// untrusted report field. This comparison grants no election or commit authority.
26    /// Opening never implicitly replaces another instance; the adapter must fence
27    /// its old packets and status correlation before adopting a new epoch.
28    /// `limits` bounds local sent-but-unreceipted metadata/body ownership, not
29    /// the remote receive window. A larger remote window never enlarges it.
30    pub fn open(
31        report: Report,
32        limits: PipelineLimits,
33        verified_received: Prefix,
34    ) -> Result<Self, FlowError> {
35        report.validate_shape()?;
36        if report.received != verified_received {
37            return Err(FlowError::History);
38        }
39        Ok(Self::from_checked_report(report, limits, ledger(limits)?))
40    }
41
42    pub(super) fn from_checked_report(
43        report: Report,
44        limits: PipelineLimits,
45        pending: VecDeque<Operation>,
46    ) -> Self {
47        Self {
48            report,
49            sent: report.received,
50            sent_bytes: report.received_bytes,
51            pending,
52            limits,
53        }
54    }
55
56    /// Replace an explicitly retired channel after a fresh, history-verified open.
57    ///
58    /// Preserves startup metadata capacity instead of allocating another ledger.
59    /// The adapter must fence old payload ownership, validate probe/session/scope
60    /// correlation, and independently verify both reported base and receipt history.
61    /// `verified_received` has the same requirements as in `open`.
62    /// Same-channel status belongs in `observe`; it must never reset send charges.
63    pub fn reopen(&mut self, report: Report, verified_received: Prefix) -> Result<(), FlowError> {
64        if report.channel == self.report.channel {
65            return Err(FlowError::Channel);
66        }
67        report.validate_shape()?;
68        if report.received != verified_received {
69            return Err(FlowError::History);
70        }
71        self.pending.clear();
72        self.report = report;
73        self.sent = report.received;
74        self.sent_bytes = report.received_bytes;
75        Ok(())
76    }
77
78    /// Advance beyond PEER reservations after independently checking a PUB receipt
79    /// against local history. Counters are receiver-owned, just as at channel open.
80    /// Existing reservations are retired only when the verified prefix covers them.
81    pub(super) fn observe_publication(
82        &mut self,
83        report: Report,
84        verified: Prefix,
85    ) -> Result<bool, FlowError> {
86        if report.received != verified {
87            return Err(FlowError::History);
88        }
89        if report.received.op <= self.sent.op {
90            return self.observe(report);
91        }
92        report.validate_shape()?;
93        if report.channel != self.report.channel {
94            return Err(FlowError::Channel);
95        }
96        if report.base != self.report.base {
97            return Err(FlowError::History);
98        }
99        if report.revision < self.report.revision {
100            return Ok(false);
101        }
102        if report.revision == self.report.revision || report.received_bytes <= self.sent_bytes {
103            return Err(FlowError::Report);
104        }
105        self.pending.clear();
106        self.report = report;
107        self.sent = report.received;
108        self.sent_bytes = report.received_bytes;
109        Ok(true)
110    }
111
112    /// Latest verified receiver accounting, including the channel's fixed base.
113    pub const fn report(&self) -> Report {
114        self.report
115    }
116
117    /// Exact channel to attach to payloads and check on incoming reports.
118    pub const fn channel(&self) -> Channel {
119        self.report.channel
120    }
121
122    /// Highest PEER reservation or independently verified PUB receipt.
123    /// Not evidence of transport submission, persistence, or a quorum.
124    pub const fn sent(&self) -> Prefix {
125        self.sent
126    }
127
128    /// Latest verified volatile receipt. Never use as a durable quorum ACK.
129    pub const fn received(&self) -> Prefix {
130        self.report.received
131    }
132
133    /// Remaining local outstanding capacity.
134    /// May be zero; not a reservation. Reports cannot grow the local ledger.
135    pub fn available(&self) -> PipelineLimits {
136        PipelineLimits {
137            max_operations: self.limits.max_operations - self.pending.len(),
138            max_body_bytes: self.limits.max_body_bytes
139                - (self.sent_bytes - self.report.received_bytes) as usize,
140        }
141    }
142
143    /// Bounded outstanding metadata, in canonical order, for probe/repair planning.
144    /// Retransmission reuses these existing reservations rather than calling `record_send`.
145    pub fn outstanding(&self) -> impl ExactSizeIterator<Item = &Operation> {
146        self.pending.iter()
147    }
148
149    /// Charge local metadata once for a new contiguous suffix in the send path.
150    ///
151    /// Reserve local payload/outbox capacity first. After success the adapter must
152    /// retain the transmission across local backpressure or fence the whole epoch.
153    /// Do not call again for a retry, duplicate, or packet still queued locally.
154    pub fn record_send(
155        &mut self,
156        channel: Channel,
157        operations: &[Operation],
158    ) -> Result<(), FlowError> {
159        if channel != self.report.channel {
160            return Err(FlowError::Channel);
161        }
162        if operations.len() > self.limits.max_operations - self.pending.len() {
163            return Err(FlowError::Capacity);
164        }
165        let (end, bytes) = batch(operations, self.sent)?;
166        let total = self
167            .sent_bytes
168            .checked_add(bytes)
169            .ok_or(FlowError::Exhausted)?;
170        if total - self.report.received_bytes > self.limits.max_body_bytes as u64 {
171            return Err(FlowError::Capacity);
172        }
173        self.pending.extend(operations.iter().copied());
174        self.sent = end;
175        self.sent_bytes = total;
176        Ok(())
177    }
178
179    /// Incorporate same-epoch receipt progress. Return false for stale or exact repeats.
180    ///
181    /// Check the exact receipt boundary and cumulative bytes against reserved sends
182    /// before freeing metadata. Conflicting same-revision reports and all retractions
183    /// fail without changing state. Epoch changes require a separate correlated open.
184    pub fn observe(&mut self, report: Report) -> Result<bool, FlowError> {
185        if report.channel != self.report.channel {
186            return Err(FlowError::Channel);
187        }
188        report.validate_shape()?;
189        if report.base != self.report.base {
190            return Err(FlowError::History);
191        }
192        if report.revision < self.report.revision {
193            return Ok(false);
194        }
195        if report.revision == self.report.revision {
196            return if report == self.report {
197                Ok(false)
198            } else {
199                Err(FlowError::Report)
200            };
201        }
202        if report.received.op < self.report.received.op
203            || report.received_bytes < self.report.received_bytes
204        {
205            return Err(FlowError::Report);
206        }
207        let mut end = self.report.received;
208        let mut bytes = self.report.received_bytes;
209        let mut count = 0;
210        for operation in &self.pending {
211            if operation.prefix.op > report.received.op {
212                break;
213            }
214            end = operation.prefix;
215            bytes = bytes
216                .checked_add(operation.body_bytes)
217                .ok_or(FlowError::Exhausted)?;
218            count += 1;
219        }
220        if end != report.received || bytes != report.received_bytes {
221            return Err(FlowError::History);
222        }
223        self.pending.drain(..count);
224        self.report = report;
225        Ok(true)
226    }
227}
ozzy_replication/flow/sender.rs

ozzy_replication/flow/
sender.rs
