ozzy_replication/normal/
admission.rs

1//! Prepare admission: chain checks against the accepted history and
2//! retention capacity for the new operations.
3
4use super::{Admission, NormalReplica};
5use crate::{OpNumber, Prefix, PreparedOperation, ReplicationError, Scope};
6use ozzy_proto::NodeId;
7
8/// Checked prepare slice: its chain end and the operations past the accepted prefix.
9#[derive(Debug, Clone, Copy)]
10struct Slice {
11    end: Prefix,
12    first_new: usize,
13    new_body_bytes: usize,
14}
15
16impl NormalReplica {
17    /// Admit a validated local-primary proposal or authenticated backup prepare.
18    ///
19    /// The full slice is checked before any admission. Retransmissions may
20    /// overlap retained operations; identical entries do not consume capacity.
21    /// Missing earlier operations request catch-up; conflicting current history
22    /// faults this core. One call is bounded by the configured pipeline size.
23    pub fn prepare(
24        &mut self,
25        from: NodeId,
26        scope: Scope,
27        operations: &[PreparedOperation],
28    ) -> Result<Admission, ReplicationError> {
29        self.require_scope(scope)?;
30        self.require_primary(from)?;
31        if operations.len() > self.limits.max_operations {
32            return Err(ReplicationError::Capacity);
33        }
34        let slice = self.check_slice(scope, operations)?;
35        let new = operations.get(slice.first_new..).unwrap_or_default();
36        if new.is_empty() {
37            return Ok(Admission::Duplicate);
38        }
39        if !self.ready_for_appends() {
40            return Err(ReplicationError::ActivationPending);
41        }
42        let body_bytes = self.check_capacity(new.len(), slice)?;
43        let ticket = self.journal.admit(new.len() as u64)?;
44        self.operations.extend(new);
45        self.body_bytes = body_bytes;
46        self.accepted = slice.end;
47        Ok(Admission::Write {
48            ticket,
49            first_new: slice.first_new,
50        })
51    }
52
53    /// Check that a prepared slice continues this view's chain and matches the
54    /// retained history it overlaps. A mismatch faults this core.
55    fn check_slice(
56        &mut self,
57        scope: Scope,
58        operations: &[PreparedOperation],
59    ) -> Result<Slice, ReplicationError> {
60        let Some(first) = operations.first() else {
61            return Err(ReplicationError::EmptyPrepare);
62        };
63        let mut predecessor = first
64            .prefix
65            .op
66            .0
67            .checked_sub(1)
68            .ok_or(ReplicationError::HistoryGap)
69            .and_then(|op| self.prefix_at(OpNumber(op)))?;
70        let mut first_new = operations.len();
71        let mut new_body_bytes = 0usize;
72        for (index, operation) in operations.iter().enumerate() {
73            if operation.group_id != scope.group_id
74                || operation.configuration_epoch != scope.configuration_epoch
75                || operation.original_view > scope.view
76                || (operation.prefix.op > self.accepted.op && operation.original_view != scope.view)
77            {
78                return Err(ReplicationError::ScopeMismatch);
79            }
80            if predecessor.op.0.checked_add(1) != Some(operation.prefix.op.0) {
81                return Err(ReplicationError::HistoryGap);
82            }
83            if operation.previous_digest != predecessor.digest {
84                return Err(self.conflict());
85            }
86            if operation.prefix.op <= self.accepted.op {
87                if self.prefix_at(operation.prefix.op)? != operation.prefix {
88                    return Err(self.conflict());
89                }
90            } else {
91                first_new = first_new.min(index);
92                new_body_bytes = new_body_bytes
93                    .checked_add(operation.body_bytes)
94                    .ok_or(ReplicationError::Capacity)?;
95            }
96            predecessor = operation.prefix;
97        }
98        Ok(Slice {
99            end: predecessor,
100            first_new,
101            new_body_bytes,
102        })
103    }
104
105    /// Bound new operations by the retained window and return the retained
106    /// body bytes after admitting them.
107    fn check_capacity(&self, new_count: usize, slice: Slice) -> Result<usize, ReplicationError> {
108        let retained = self.retained_limits();
109        let body_bytes = self
110            .body_bytes
111            .checked_add(slice.new_body_bytes)
112            .ok_or(ReplicationError::Capacity)?;
113        if new_count
114            > retained
115                .max_operations
116                .saturating_sub(self.operations.len())
117            || body_bytes > retained.max_body_bytes
118        {
119            return Err(ReplicationError::Capacity);
120        }
121        // A backup admits only what the primary can keep in flight beyond the
122        // commit it announced; the retained bound above covers its own release
123        // trailing that commit.
124        if !self.is_primary()
125            && slice.end.op.0.saturating_sub(self.live_base().0) > self.limits.max_operations as u64
126        {
127            return Err(ReplicationError::Capacity);
128        }
129        Ok(body_bytes)
130    }
131}
ozzy_replication/normal/admission.rs

ozzy_replication/normal/
admission.rs
