ozzy_replication/view_change/
selection.rs

1//! History selection from a quorum of reports, verified against the chosen
2//! source's pinned history.
3
4use super::{LogSource, SelectedView, ViewChange, ViewChangeError};
5use crate::{Digest, FrozenLog, OpNumber, Prefix, VoterSet};
6
7impl ViewChange {
8    /// Select the highest last-normal-view, then longest accepted lineage.
9    ///
10    /// `lookup` must be a nonblocking lookup into already verified, pinned
11    /// canonical history anchored by the supplied source's exact tail digest.
12    /// An isolated matching hash or a chunk from another generation is not such
13    /// evidence. It returns the source's digest at the prefix's op; the prefix
14    /// is the expected value, a fetch hint and never evidence. Missing bytes
15    /// return `None`; fetch/verify them outside the core.
16    /// At most six protected-prefix lookups are needed for three voters.
17    ///
18    /// Selection includes this primary's report and another distinct voter. It
19    /// protects every reported commit and same-normal-view accepted prefix.
20    /// Higher-normal-view histories may discard only older uncertain suffixes.
21    pub fn select(
22        &mut self,
23        mut lookup: impl FnMut(LogSource, Prefix) -> Option<Digest>,
24    ) -> Result<SelectedView, ViewChangeError> {
25        self.require_promised()?;
26        self.require_primary()?;
27        if let Some(selected) = self.selected {
28            return Ok(selected);
29        }
30        let reports = self.reports;
31        let Some(mut report) = reports[self.local] else {
32            return Err(ViewChangeError::ReportQuorumMissing);
33        };
34        let local = report.log;
35        let mut chosen = self.configuration.voter(self.local);
36        let voters: VoterSet = reports
37            .iter()
38            .filter_map(|(voter, candidate)| candidate.map(|_| voter))
39            .collect();
40        if voters.len() < 2 {
41            return Err(ViewChangeError::ReportQuorumMissing);
42        }
43        for (voter, candidate) in reports.iter() {
44            let Some(candidate) = candidate else {
45                continue;
46            };
47            let rank = (candidate.log.last_normal_view, candidate.log.accepted.op);
48            let best = (report.log.last_normal_view, report.log.accepted.op);
49            if rank == best && candidate.log.accepted != report.log.accepted {
50                return Err(self.conflict());
51            }
52            if rank > best {
53                report = *candidate;
54                chosen = self.configuration.voter(voter);
55            }
56        }
57        let source = LogSource {
58            voter: chosen,
59            generation: report.generation,
60            accepted: report.log.accepted,
61        };
62        let mut committed = Prefix::GENESIS;
63        for candidate in reports.values().flatten() {
64            if candidate.log.committed.op > committed.op {
65                committed = candidate.log.committed;
66            }
67            self.verify_prefix(source, candidate.log.committed, &mut lookup)?;
68            if candidate.log.last_normal_view == report.log.last_normal_view {
69                self.verify_prefix(source, candidate.log.accepted, &mut lookup)?;
70            }
71        }
72        let selected = SelectedView {
73            scope: self.scope(),
74            source,
75            committed,
76            retained: retained(local, report.log, committed),
77            voters,
78        };
79        self.selected = Some(selected);
80        Ok(selected)
81    }
82
83    pub(crate) fn verify_prefix(
84        &mut self,
85        source: LogSource,
86        prefix: Prefix,
87        lookup: &mut impl FnMut(LogSource, Prefix) -> Option<Digest>,
88    ) -> Result<(), ViewChangeError> {
89        if prefix.op > source.accepted.op {
90            return Err(self.conflict());
91        }
92        let digest = if prefix.op == OpNumber(0) {
93            Digest::ZERO
94        } else if prefix.op == source.accepted.op {
95            source.accepted.digest
96        } else {
97            lookup(source, prefix).ok_or(ViewChangeError::HistoryMissing)?
98        };
99        if digest != prefix.digest {
100            return Err(self.conflict());
101        }
102        Ok(())
103    }
104
105    /// The checks of `prefix` against `source` that need no history. A
106    /// staging stream through it checks the rest.
107    pub(crate) fn require_streamed(
108        &mut self,
109        source: LogSource,
110        prefix: Prefix,
111    ) -> Result<(), ViewChangeError> {
112        if prefix.op > source.accepted.op
113            || (prefix.op == source.accepted.op && prefix != source.accepted)
114            || (prefix.op == OpNumber(0) && prefix.digest != Digest::ZERO)
115        {
116            return Err(self.conflict());
117        }
118        Ok(())
119    }
120}
121
122/// This voter's longest prefix in the selected lineage, at most `committed`.
123/// Selection verifies its accepted prefix when it shares the `chosen` normal
124/// view; otherwise only its commit.
125fn retained(local: FrozenLog, chosen: FrozenLog, committed: Prefix) -> Prefix {
126    let verified = if local.last_normal_view == chosen.last_normal_view {
127        local.accepted
128    } else {
129        local.committed
130    };
131    if verified.op > committed.op {
132        committed
133    } else {
134        verified
135    }
136}
ozzy_replication/view_change/selection.rs

ozzy_replication/view_change/
selection.rs
