ozzy_replication/flow/transmit/
open.rs

1//! Adopting a verified report: publication progress, a reopen, or the first ledger.
2
3use std::time::Duration;
4
5use super::{FlowError, OpenRequest, Prefix, Report, Sender, TransmitError, Transmitter};
6
7impl Transmitter {
8    /// Adopt only the still-current candidate after exact local-history verification.
9    ///
10    /// Prefix arguments must come from current-generation local history, not copied
11    /// report fields. Stale callbacks return false. Old pending payloads must be
12    /// fenced by their epochs; they cannot become reservations in the new channel.
13    pub fn open_verified(
14        &mut self,
15        request: OpenRequest,
16        base: Prefix,
17        received: Prefix,
18        now: Duration,
19    ) -> Result<bool, TransmitError> {
20        if self.candidate != Some(request) {
21            return Ok(false);
22        }
23        let probe = if let Some(id) = request.request_id {
24            let Some(probe) = self.probes.pending().filter(|probe| {
25                probe.scope == request.report.channel.scope && probe.request_id == id
26            }) else {
27                return Ok(false);
28            };
29            self.probes.completion_deadline(probe.scope, id, now)?;
30            Some(probe)
31        } else {
32            if !self.broadcast
33                || self
34                    .sender()
35                    .is_none_or(|sender| sender.channel() != request.report.channel)
36            {
37                return Ok(false);
38            }
39            None
40        };
41        let report = request.report;
42        if report.base != base || report.received != received {
43            return Err(FlowError::History.into());
44        }
45        report.validate_shape()?;
46        let same_channel = self
47            .sender()
48            .is_some_and(|sender| sender.channel() == report.channel);
49        let previous = self.sender().map(|sender| sender.received().op);
50        if !self.adopt(report, received, same_channel)? {
51            self.clear_candidate();
52            return Ok(false);
53        }
54        if let Some(probe) = probe {
55            self.probes.complete(probe.scope, probe.request_id, now)?;
56            if self.broadcast {
57                let target = self
58                    .candidate_limit
59                    .unwrap_or(probe.tail.op)
60                    .min(probe.tail.op);
61                self.catch_up = (report.received.op < target).then_some(target);
62            }
63        }
64        self.active = true;
65        self.clear_candidate();
66        if !same_channel {
67            self.repair = None;
68            self.receipt_progress.reset();
69        }
70        if !same_channel || previous.is_none_or(|op| received.op > op) {
71            self.receipt_progress.advanced(received.op.0, now);
72        }
73        self.trim_repair();
74        Ok(true)
75    }
76
77    /// Move the ledger to a verified report: accept PUB progress on the same
78    /// channel, reopen on a new one, or install the reserved ledger. Returns
79    /// false when same-channel progress no longer applies.
80    fn adopt(
81        &mut self,
82        report: Report,
83        received: Prefix,
84        same_channel: bool,
85    ) -> Result<bool, FlowError> {
86        match &mut self.sender {
87            Some(sender) if self.broadcast && same_channel => {
88                sender.observe_publication(report, received)
89            }
90            Some(sender) => sender.reopen(report, received).map(|()| true),
91            None => {
92                self.sender = Some(Sender::from_checked_report(
93                    report,
94                    self.limits,
95                    std::mem::take(&mut self.spare),
96                ));
97                Ok(true)
98            }
99        }
100    }
101}
ozzy_replication/flow/transmit/open.rs

ozzy_replication/flow/transmit/
open.rs
