ozzy_replication/
flow.rs

1//! Bounded local retention and receipt accounting, independent of consensus and durability.
2//!
3//! Adapters authenticate peers, fence epochs on every packet, retain payloads,
4//! verify canonical history, and supply release events only after application.
5//! This module performs no I/O, clock reads, hashing, voting, or implicit recovery.
6//! Probe deadlines advance only from explicit adapter-supplied monotonic time.
7//! See `doc/REPLICATION.md` for the surrounding probe/repair integration contract.
8
9mod probe;
10mod receiver;
11mod sender;
12mod transmit;
13
14pub use probe::{Probe, ProbeError, ProbeScheduler, ProbeTiming};
15pub use receiver::Receiver;
16pub use sender::Sender;
17pub use transmit::{OpenRequest, Repair, StatusOutcome, TransmitError, Transmitter};
18
19use std::collections::VecDeque;
20
21use crate::{Digest, PipelineLimits, Prefix, Scope};
22
23/// Receiver-issued incarnation of volatile retention state.
24///
25/// Supply a fresh unpredictable value on restart, scope change, or retraction.
26/// Never reuse an epoch for reset state. It is not a durable journal generation.
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub struct ReceiveEpoch(u128);
29
30impl ReceiveEpoch {
31    /// Validate a nonzero adapter-supplied identity; does not generate randomness.
32    pub const fn new(value: u128) -> Result<Self, FlowError> {
33        if value == 0 {
34            Err(FlowError::Invalid)
35        } else {
36            Ok(Self(value))
37        }
38    }
39
40    /// Exact identity for wire encoding and equality checks, never chronological order.
41    pub const fn get(self) -> u128 {
42        self.0
43    }
44}
45
46/// Complete scope and receive incarnation for one independently bound peer.
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub struct Channel {
49    /// Group, immutable configuration, and current normal view.
50    pub scope: Scope,
51    /// Receiver-owned volatile incarnation.
52    pub epoch: ReceiveEpoch,
53}
54
55/// Compact metadata for one hash-validated canonical body retained by the adapter.
56///
57/// This is neither payload ownership nor application-validation evidence.
58#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59pub struct Operation {
60    /// Exact contiguous prefix ending at this operation.
61    pub prefix: Prefix,
62    /// Digest of the immediately preceding canonical operation.
63    pub previous_digest: Digest,
64    /// Canonical body bytes; not compressed, framed, or per-record bytes.
65    pub body_bytes: u64,
66}
67
68/// Exact cumulative receipt in one receive epoch.
69///
70/// No field is quorum, durable, or application evidence. The adapter authenticates
71/// and correlates reports before opening an epoch; same-epoch updates are monotonic.
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub struct Report {
74    /// Scope/incarnation fence, in addition to the external peer/session binding.
75    pub channel: Channel,
76    /// Strictly increasing on retention changes, unchanged on retransmission.
77    pub revision: u64,
78    /// Fixed applied prefix at epoch initialization; cumulative counters start here.
79    pub base: Prefix,
80    /// Contiguous prefix whose exact bodies are retained in this epoch.
81    pub received: Prefix,
82    /// Cumulative canonical body bytes after `base` through `received`.
83    pub received_bytes: u64,
84}
85
86/// Rejected accounting transition. Every failure leaves existing state unchanged.
87#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
88pub enum FlowError {
89    /// Zero/invalid identity, prefix, limits, revision, or operation size.
90    #[error("invalid replica flow metadata")]
91    Invalid,
92    /// Scope or receiver incarnation differs; a fresh correlated open is required.
93    #[error("replica flow channel mismatch")]
94    Channel,
95    /// Local count or body capacity is exhausted; retain/backpressure, never over-admit.
96    #[error("replica flow capacity exhausted")]
97    Capacity,
98    /// Prefix/digest/byte accounting does not match the retained contiguous history.
99    #[error("replica flow history mismatch")]
100    History,
101    /// Same-epoch report equivocates or retracts previously advertised state.
102    #[error("replica flow report is inconsistent")]
103    Report,
104    /// A cumulative counter cannot advance; fence and establish a new epoch.
105    #[error("replica flow counter exhausted")]
106    Exhausted,
107    /// Startup could not reserve the configured metadata capacity.
108    #[error("replica flow metadata allocation failed")]
109    Allocation,
110}
111
112fn capacities(limits: PipelineLimits) -> Result<(u64, u64), FlowError> {
113    if limits.max_operations == 0 || limits.max_body_bytes == 0 {
114        return Err(FlowError::Invalid);
115    }
116    Ok((
117        u64::try_from(limits.max_operations).map_err(|_| FlowError::Invalid)?,
118        u64::try_from(limits.max_body_bytes).map_err(|_| FlowError::Invalid)?,
119    ))
120}
121
122fn ledger(limits: PipelineLimits) -> Result<VecDeque<Operation>, FlowError> {
123    capacities(limits)?;
124    let mut entries = VecDeque::new();
125    entries
126        .try_reserve_exact(limits.max_operations)
127        .map_err(|_| FlowError::Allocation)?;
128    Ok(entries)
129}
130
131fn prefix_valid(prefix: Prefix) -> bool {
132    (prefix.op.0 == 0) == (prefix.digest == Digest::ZERO)
133}
134
135fn batch(operations: &[Operation], previous: Prefix) -> Result<(Prefix, u64), FlowError> {
136    if operations.is_empty() {
137        return Err(FlowError::Invalid);
138    }
139    let mut end = previous;
140    let mut bytes = 0u64;
141    for operation in operations {
142        if operation.body_bytes == 0 || !prefix_valid(operation.prefix) {
143            return Err(FlowError::Invalid);
144        }
145        if operation.prefix.op.0 != end.op.0.checked_add(1).ok_or(FlowError::Exhausted)?
146            || operation.previous_digest != end.digest
147        {
148            return Err(FlowError::History);
149        }
150        bytes = bytes
151            .checked_add(operation.body_bytes)
152            .ok_or(FlowError::Exhausted)?;
153        end = operation.prefix;
154    }
155    Ok((end, bytes))
156}
157
158impl Report {
159    /// Validate receipt shape and the caller's local retention configuration.
160    /// Does not authenticate or verify history; heterogeneous peers may differ.
161    pub fn validate(self, limits: PipelineLimits) -> Result<(), FlowError> {
162        capacities(limits)?;
163        self.validate_shape()
164    }
165
166    /// Validate fixed metadata without applying a negotiated receive-window bound.
167    ///
168    /// A local retention window can exceed one wire frame. Codecs use this allocation-free
169    /// check. Senders separately enforce their own outstanding metadata/body
170    /// bounds on every reservation.
171    pub fn validate_shape(self) -> Result<(), FlowError> {
172        if self.channel.scope.group_id.as_bytes() == &[0; 16]
173            || self.channel.scope.configuration_digest == Digest::ZERO
174            || self.revision == 0
175            || !prefix_valid(self.base)
176            || !prefix_valid(self.received)
177        {
178            return Err(FlowError::Invalid);
179        }
180        let received = self
181            .received
182            .op
183            .0
184            .checked_sub(self.base.op.0)
185            .ok_or(FlowError::History)?;
186        if (received == 0 && self.received != self.base)
187            || (received == 0) != (self.received_bytes == 0)
188            || self.received_bytes < received
189        {
190            return Err(FlowError::Report);
191        }
192        Ok(())
193    }
194}
ozzy_replication/flow.rs

ozzy_replication/
flow.rs
