ozzy_replication/flow/
receiver.rs

1#[cfg(test)]
2mod tests;
3
4use super::{
5    Channel, FlowError, Operation, PipelineLimits, Prefix, ReceiveEpoch, Report, VecDeque, batch,
6    capacities, ledger,
7};
8
9/// Local retained-body accounting. Capacity is never advertised to a sender.
10/// The adapter reserves actual payload backing before retaining a suffix.
11#[derive(Debug)]
12pub struct Receiver {
13    report: Report,
14    released: Prefix,
15    released_bytes: u64,
16    retained: VecDeque<Operation>,
17    limits: PipelineLimits,
18}
19
20impl Receiver {
21    /// Start at an exact applied prefix with a bounded local metadata window.
22    pub fn new(channel: Channel, base: Prefix, limits: PipelineLimits) -> Result<Self, FlowError> {
23        capacities(limits)?;
24        let report = Report {
25            channel,
26            revision: 1,
27            base,
28            received: base,
29            received_bytes: 0,
30        };
31        report.validate_shape()?;
32        Ok(Self {
33            report,
34            released: base,
35            released_bytes: 0,
36            retained: ledger(limits)?,
37            limits,
38        })
39    }
40
41    /// Exact contiguous receipt. This supplies no persistence or vote evidence.
42    pub const fn report(&self) -> Report {
43        self.report
44    }
45
46    /// Local room after every still-retained body, independently of queue slots.
47    pub fn available(&self) -> PipelineLimits {
48        PipelineLimits {
49            max_operations: self.limits.max_operations - self.retained.len(),
50            max_body_bytes: self.limits.max_body_bytes
51                - (self.report.received_bytes - self.released_bytes) as usize,
52        }
53    }
54
55    /// Install a fresh incarnation after all old payload ownership is fenced.
56    pub fn reinitialize(&mut self, channel: Channel, applied: Prefix) -> Result<(), FlowError> {
57        if channel.epoch == self.report.channel.epoch {
58            return Err(FlowError::Channel);
59        }
60        let report = Report {
61            channel,
62            revision: 1,
63            base: applied,
64            received: applied,
65            received_bytes: 0,
66        };
67        report.validate_shape()?;
68        self.retained.clear();
69        self.released = applied;
70        self.released_bytes = 0;
71        self.report = report;
72        Ok(())
73    }
74
75    /// Retract only unadmitted history. Accepted bodies remain charged.
76    /// Caller fences old packets and supplies a never-reused receive epoch.
77    pub fn retract(&mut self, epoch: ReceiveEpoch, through: Prefix) -> Result<(), FlowError> {
78        if epoch == self.report.channel.epoch {
79            return Err(FlowError::Channel);
80        }
81        let count = if through == self.released {
82            0
83        } else {
84            self.retained
85                .iter()
86                .position(|op| op.prefix == through)
87                .ok_or(FlowError::History)?
88                + 1
89        };
90        let received_bytes = self
91            .retained
92            .iter()
93            .take(count)
94            .try_fold(0u64, |sum, op| sum.checked_add(op.body_bytes))
95            .ok_or(FlowError::Exhausted)?;
96        let report = Report {
97            channel: Channel {
98                epoch,
99                ..self.report.channel
100            },
101            revision: 1,
102            base: self.released,
103            received: through,
104            received_bytes,
105        };
106        report.validate_shape()?;
107        self.retained.truncate(count);
108        self.released_bytes = 0;
109        self.report = report;
110        Ok(())
111    }
112
113    /// Charge a complete contiguous suffix after physical backing is reserved.
114    /// Duplicates are filtered by the adapter before this transition.
115    pub fn retain(&mut self, channel: Channel, operations: &[Operation]) -> Result<(), FlowError> {
116        if channel != self.report.channel {
117            return Err(FlowError::Channel);
118        }
119        let (end, bytes) = batch(operations, self.report.received)?;
120        let available = self.available();
121        if operations.len() > available.max_operations || bytes > available.max_body_bytes as u64 {
122            return Err(FlowError::Capacity);
123        }
124        let total = self
125            .report
126            .received_bytes
127            .checked_add(bytes)
128            .ok_or(FlowError::Exhausted)?;
129        let revision = self
130            .report
131            .revision
132            .checked_add(1)
133            .ok_or(FlowError::Exhausted)?;
134        self.retained.extend(operations.iter().copied());
135        self.report.received = end;
136        self.report.received_bytes = total;
137        self.report.revision = revision;
138        Ok(())
139    }
140
141    /// Release after application and durable persistence.
142    /// This changes local room only. It emits no receipt or admission grant.
143    pub fn release(&mut self, through: Prefix) -> Result<(), FlowError> {
144        if through == self.released {
145            return Ok(());
146        }
147        let count = self
148            .retained
149            .iter()
150            .position(|op| op.prefix == through)
151            .ok_or(FlowError::History)?
152            + 1;
153        let bytes: u64 = self
154            .retained
155            .iter()
156            .take(count)
157            .map(|op| op.body_bytes)
158            .sum();
159        let released_bytes = self
160            .released_bytes
161            .checked_add(bytes)
162            .ok_or(FlowError::Exhausted)?;
163        self.retained.drain(..count);
164        self.released = through;
165        self.released_bytes = released_bytes;
166        Ok(())
167    }
168}
ozzy_replication/flow/receiver.rs

ozzy_replication/flow/
receiver.rs
