ozzy_replication/
wire.rs

1//! Bounded native replica messages. No I/O or implicit consensus authority.
2//!
3//! The transport supplies an independently authenticated voter/session binding.
4//! Scope decoding checks the immutable configuration; the production cores still
5//! decide whether the message's view, history, role, and evidence are admissible.
6
7#![cfg_attr(not(test), warn(clippy::arithmetic_side_effects))]
8
9mod admission;
10mod election;
11pub use admission::ReceiveFence;
12mod flow;
13pub use flow::flow_state_route;
14mod compact;
15pub use compact::{COMPACT_STATE_BYTES, CompactState};
16mod prepare;
17mod publication;
18pub use publication::{
19    PUBLICATION_BYTES, PUBLICATION_TOPIC_BYTES, decode_publication, encode_publication,
20    publication_topic,
21};
22mod checkpoint;
23mod error;
24mod recovery;
25pub use checkpoint::{
26    CheckpointMessage, CheckpointRequest, HistoryFence, HistoryRetired, encode_checkpoint,
27    encode_history_retired,
28};
29mod fields;
30mod transfer;
31
32pub use error::WireError;
33pub use flow::{FlowMessage, FlowProbe, FlowState, encode_flow_probe, encode_flow_state};
34pub use prepare::{
35    Operation, Operations, Prepare, PrepareBatch, PrepareEncoding, PrepareMetadata,
36    VerifiedOperation, encode_flow_prepare, encode_flow_prepare_metadata, encode_prepare,
37    encode_prepare_metadata, encode_prepare_unbound,
38};
39pub use recovery::{
40    RecoveryMessage, RecoveryRequest, RecoveryState, encode_recovery, encode_recovery_state,
41};
42pub use transfer::{FetchOps, OpsBatch, encode_fetch, encode_ops};
43
44use fields::{ReadFields, WriteFields, validate_election_view, validate_prefix, validate_scope};
45use ozzy_proto::layout::{Reader, Writer};
46use ozzy_proto::{Envelope, EnvelopeLimits, Opcode};
47use ozzy_proto::{LinkSessionId, NodeId};
48
49use crate::flow::{FlowError, ReceiveEpoch};
50use crate::{
51    Commit, Configuration, DoViewChange, Prefix, PrepareOk, Scope, StartView, StartViewChange,
52};
53
54const COMMON_BYTES: usize = 80;
55
56/// Authenticated peer and established session supplied by the transport adapter.
57///
58/// Construction checks membership and nonzero IDs, not cryptographic identity.
59/// Never construct this binding from untrusted packet fields or routing IDs alone.
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61pub struct PeerBinding {
62    configuration: Configuration,
63    peer: NodeId,
64    session: LinkSessionId,
65    receive_epoch: Option<ReceiveEpoch>,
66}
67
68impl PeerBinding {
69    /// Bind one independently authenticated configured voter to its live session.
70    pub fn new(
71        configuration: Configuration,
72        peer: NodeId,
73        session: LinkSessionId,
74    ) -> Result<Self, WireError> {
75        configuration.voter_index(peer)?;
76        if session.as_bytes() == &[0; 16] {
77            return Err(WireError::Peer);
78        }
79        Ok(Self {
80            configuration,
81            peer,
82            session,
83            receive_epoch: None,
84        })
85    }
86
87    /// Require epoch-bound normal payloads bound to this locally issued receive epoch.
88    ///
89    /// Set only from the receiver's own current flow state, never incoming fields.
90    /// Rejects legacy PREPARE and stale `PREPARE_FLOW` before decoding/hashing bodies.
91    /// Control and installation-history messages retain their independent rules.
92    #[must_use]
93    pub const fn with_receive_epoch(mut self, epoch: ReceiveEpoch) -> Self {
94        self.receive_epoch = Some(epoch);
95        self
96    }
97}
98
99/// Local sender, established link session, and bounds of one encoded packet.
100#[derive(Debug, Clone, Copy, PartialEq, Eq)]
101pub struct EncodeContext {
102    /// Configured local voter, written as envelope sender and embedded voter.
103    pub sender: NodeId,
104    /// Established session of the destination link.
105    pub session: LinkSessionId,
106    /// Frame and operation bounds of the encoded packet.
107    pub limits: WireLimits,
108}
109
110/// Directional receive and command work bounds.
111#[derive(Debug, Clone, Copy, PartialEq, Eq)]
112pub struct WireLimits {
113    /// Maximum actual metadata and payload frame sizes.
114    pub envelope: EnvelopeLimits,
115    /// Maximum canonical operations in a prepare/history response, independent of records.
116    pub max_operations: usize,
117}
118
119impl WireLimits {
120    /// Conservative shared prepare/history profile. Round the largest fixed
121    /// metadata span up to 16 bytes, preserving room for control exchanges.
122    pub fn for_transfer(max_operations: usize, max_body_bytes: usize) -> Option<Self> {
123        let fixed = transfer::FETCH_BYTES
124            .max(transfer::OPS_FIXED_BYTES)
125            .max(prepare::BATCH_BYTES + 16)
126            .next_multiple_of(16);
127        Some(Self {
128            envelope: EnvelopeLimits {
129                max_metadata_bytes: max_operations
130                    .checked_mul(prepare::DESCRIPTOR_BYTES)?
131                    .checked_add(fixed)?,
132                max_payload_bytes: max_body_bytes,
133            },
134            max_operations,
135        })
136    }
137
138    /// Bound a routed native message, including its 16-byte destination identity.
139    pub fn message_bytes(self) -> Option<usize> {
140        self.envelope
141            .max_metadata_bytes
142            .checked_add(self.envelope.max_payload_bytes)?
143            .checked_add(ozzy_proto::ENVELOPE_BYTES + 16)
144    }
145}
146
147impl Default for WireLimits {
148    fn default() -> Self {
149        Self {
150            envelope: EnvelopeLimits::default(),
151            max_operations: 64,
152        }
153    }
154}
155
156/// Fixed-size replica control metadata. Receipt alone has no authority.
157#[derive(Debug, Clone, Copy, PartialEq, Eq)]
158pub enum Control {
159    /// Cumulative durable backup evidence plus its current session fence.
160    PrepareOk {
161        /// Matching contiguous prefix after successful disk synchronization.
162        ack: PrepareOk,
163    },
164    /// Primary's cumulative commit announcement or idle heartbeat.
165    Commit(Commit),
166    /// Volatile suspicion about this current view, not a durable promise or vote.
167    ExitView(Scope),
168    /// First election phase, emitted after the sender's durable view promise.
169    StartViewChange(StartViewChange),
170    /// Frozen full-WAL descriptor for the candidate's second-phase quorum.
171    DoViewChange(DoViewChange),
172    /// Primary's durably installed selected lineage, not a backup installation.
173    StartView(StartView),
174}
175
176impl Control {
177    /// Claimed prefix for routing/replay hints. This does not validate its evidence.
178    pub const fn acknowledged(self) -> Option<Prefix> {
179        match self {
180            Self::PrepareOk { ack, .. } => Some(ack.durable),
181            _ => None,
182        }
183    }
184
185    /// Exact group, immutable configuration digest, and active/proposed view.
186    pub const fn scope(self) -> Scope {
187        match self {
188            Self::PrepareOk { ack, .. } => ack.scope,
189            Self::Commit(message) => message.scope,
190            Self::ExitView(scope) => scope,
191            Self::StartViewChange(message) => message.scope,
192            Self::DoViewChange(message) => message.scope,
193            Self::StartView(message) => message.scope,
194        }
195    }
196
197    fn opcode(self) -> Opcode {
198        match self {
199            Self::PrepareOk { .. } => Opcode::PrepareOk,
200            Self::Commit(_) => Opcode::Commit,
201            Self::ExitView(_) => Opcode::ExitView,
202            Self::StartViewChange(_) => Opcode::StartViewChange,
203            Self::DoViewChange(_) => Opcode::DoViewChange,
204            Self::StartView(_) => Opcode::StartView,
205        }
206    }
207}
208
209/// Decoded replica command, still subject to role/history checks in the core.
210#[derive(Debug, Clone, Copy, PartialEq, Eq)]
211pub enum ReplicaMessage<'a> {
212    /// Receiver-fenced hint to enter nonvoting recovery, never new authority.
213    HistoryRetired(HistoryRetired),
214    /// Checkpoint state bytes for a separately authorized nonvoting recovery.
215    Checkpoint(CheckpointMessage<'a>),
216    /// Fresh lost-state evidence, neither receipt credit nor a normal/election vote.
217    Recovery(RecoveryMessage),
218    /// Receipt/credit exchange or epoch-bound data; never durable quorum evidence.
219    Flow(FlowMessage<'a>),
220    /// Fixed-size quorum/control evidence from the bound sender.
221    Control(Control),
222    /// Hash-validated canonical bytes, still requiring application validation.
223    Prepare(PrepareBatch<'a>),
224    /// Correlated bounded request for an independently pinned source lineage.
225    FetchOps(FetchOps),
226    /// Hash-validated response bytes, not a normal prepare or commit announcement.
227    Ops(OpsBatch<'a>),
228}
229
230/// Envelope and valid prefix length in the caller's reusable metadata buffer.
231#[derive(Debug, Clone, Copy, PartialEq, Eq)]
232pub struct ControlEncoding {
233    /// First Ozzy frame, with exact lengths for this encoding.
234    pub header: [u8; 64],
235    /// Second frame length; the third frame must be present and empty.
236    pub metadata_bytes: usize,
237}
238
239/// Encode control metadata into caller-owned storage without allocating.
240///
241/// Invalid fields or insufficient capacity leave the output unchanged. Callers
242/// may send only messages actually authorized by the current consensus role.
243pub fn encode_control(
244    sender: NodeId,
245    session: LinkSessionId,
246    message: Control,
247    output: &mut [u8],
248) -> Result<ControlEncoding, WireError> {
249    let scope = message.scope();
250    validate_scope(scope)?;
251    let metadata_bytes = match message {
252        Control::PrepareOk { ack, .. } => {
253            validate_prefix(ack.durable)?;
254            COMMON_BYTES + 40
255        }
256        Control::Commit(message) => {
257            validate_prefix(message.committed)?;
258            COMMON_BYTES + 40
259        }
260        Control::StartViewChange(message) => {
261            validate_election_view(message.scope)?;
262            COMMON_BYTES
263        }
264        Control::ExitView(_) => COMMON_BYTES,
265        Control::DoViewChange(message) => {
266            election::validate_report(message)?;
267            COMMON_BYTES + 16 + 8 + 40 + 40
268        }
269        Control::StartView(message) => {
270            election::validate_start(message)?;
271            COMMON_BYTES + 16 + 40 + 40
272        }
273    };
274    if output.len() < metadata_bytes {
275        return Err(WireError::Capacity);
276    }
277    let header = Envelope {
278        opcode: message.opcode(),
279        response: false,
280        request_id: None,
281        sender,
282        session: Some(session),
283    }
284    .encode_header(metadata_bytes, 0, EnvelopeLimits::default())?;
285    let mut writer = Writer::with_len(output, metadata_bytes)?;
286    writer.scope(scope, sender)?;
287    match message {
288        Control::PrepareOk { ack } => writer.prefix(ack.durable)?,
289        Control::Commit(message) => writer.prefix(message.committed)?,
290        Control::StartViewChange(_) | Control::ExitView(_) => {}
291        Control::DoViewChange(message) => election::write_report(&mut writer, message)?,
292        Control::StartView(message) => election::write_start(&mut writer, message)?,
293    }
294    Ok(ControlEncoding {
295        header,
296        metadata_bytes,
297    })
298}
299
300/// Read only the fixed group/configuration prefix of an implemented PEER
301/// command. This does not authenticate the sender or validate membership, view,
302/// operation hashes, or payloads. Dispatch uses it to select an actor, which must
303/// still call `decode` with an independently established `PeerBinding`.
304pub fn route(packet: ozzy_proto::Packet<'_>, limits: EnvelopeLimits) -> Result<Scope, WireError> {
305    packet
306        .envelope
307        .validate_frames(packet.metadata.len(), packet.payload.len(), limits)?;
308    if !matches!(
309        packet.envelope.opcode,
310        Opcode::ReplicaOpen
311            | Opcode::ReplicaState
312            | Opcode::Prepare
313            | Opcode::PrepareFlow
314            | Opcode::PreparePub
315            | Opcode::PrepareOk
316            | Opcode::Commit
317            | Opcode::ExitView
318            | Opcode::StartViewChange
319            | Opcode::DoViewChange
320            | Opcode::StartView
321            | Opcode::Recovery
322            | Opcode::RecoveryState
323            | Opcode::FetchOps
324            | Opcode::Ops
325            | Opcode::HistoryRetired
326            | Opcode::SnapshotBegin
327            | Opcode::SnapshotChunk
328    ) {
329        return Err(WireError::UnsupportedCommand);
330    }
331    if !matches!(
332        packet.envelope.opcode,
333        Opcode::Prepare
334            | Opcode::PrepareFlow
335            | Opcode::PreparePub
336            | Opcode::Ops
337            | Opcode::SnapshotChunk
338    ) && !packet.payload.is_empty()
339    {
340        return Err(WireError::Payload);
341    }
342    Reader::new(packet.metadata).routing_scope(packet.envelope.sender)
343}
344
345/// Decode a bounded packet and reject mismatched voter/session/configuration.
346///
347/// This does not validate a claim against local disk state or count a quorum.
348/// Old views remain recognizable so the consensus core can apply its fencing rules.
349pub fn decode<'a>(
350    frames: &[&'a [u8]],
351    binding: PeerBinding,
352    limits: WireLimits,
353) -> Result<ReplicaMessage<'a>, WireError> {
354    let packet = ozzy_proto::decode_packet(frames, limits.envelope)?;
355    if packet.envelope.sender != binding.peer || packet.envelope.session != Some(binding.session) {
356        return Err(WireError::Peer);
357    }
358    let mut reader = Reader::new(packet.metadata);
359    let incoming = Incoming {
360        scope: reader.scope(binding)?,
361        envelope: packet.envelope,
362        reader,
363        payload: packet.payload,
364    };
365    match packet.envelope.opcode {
366        Opcode::HistoryRetired => {
367            checkpoint::decode_retired(incoming).map(ReplicaMessage::HistoryRetired)
368        }
369        Opcode::SnapshotBegin | Opcode::SnapshotChunk => {
370            checkpoint::decode(incoming, binding, limits).map(ReplicaMessage::Checkpoint)
371        }
372        Opcode::Recovery => recovery::decode_request(incoming)
373            .map(|request| ReplicaMessage::Recovery(RecoveryMessage::Request(request))),
374        Opcode::RecoveryState => recovery::decode_state(incoming, binding)
375            .map(|state| ReplicaMessage::Recovery(RecoveryMessage::State(state))),
376        Opcode::ReplicaOpen => flow::decode_probe(incoming)
377            .map(|probe| ReplicaMessage::Flow(FlowMessage::Probe(probe))),
378        Opcode::ReplicaState => flow::decode_state(incoming)
379            .map(|state| ReplicaMessage::Flow(FlowMessage::State(state))),
380        Opcode::FetchOps => transfer::decode_fetch(incoming, binding).map(ReplicaMessage::FetchOps),
381        Opcode::Ops => transfer::decode_ops(incoming, limits).map(ReplicaMessage::Ops),
382        _ => decode_uncorrelated(incoming, binding, limits),
383    }
384}
385
386/// Fields after the authenticated scope, borrowing the packet frames.
387struct Incoming<'a> {
388    scope: Scope,
389    envelope: Envelope,
390    reader: Reader<'a>,
391    payload: &'a [u8],
392}
393
394/// Decode normal data and control, which carry no request correlation.
395fn decode_uncorrelated(
396    Incoming {
397        scope,
398        envelope,
399        mut reader,
400        payload,
401    }: Incoming<'_>,
402    binding: PeerBinding,
403    limits: WireLimits,
404) -> Result<ReplicaMessage<'_>, WireError> {
405    if envelope.response || envelope.request_id.is_some() {
406        return Err(WireError::Correlation);
407    }
408    match envelope.opcode {
409        Opcode::PrepareFlow => {
410            let epoch = ReceiveEpoch::new(u128::from_be_bytes(reader.array()?))?;
411            if binding.receive_epoch != Some(epoch) {
412                return Err(FlowError::Channel.into());
413            }
414            prepare::decode_prepare(scope, reader, payload, limits)
415                .map(|batch| ReplicaMessage::Flow(FlowMessage::Prepare { epoch, batch }))
416        }
417        Opcode::Prepare => {
418            if binding.receive_epoch.is_some() {
419                return Err(FlowError::Channel.into());
420            }
421            prepare::decode_prepare(scope, reader, payload, limits).map(ReplicaMessage::Prepare)
422        }
423        _ if !payload.is_empty() => Err(WireError::Payload),
424        opcode => decode_control(scope, opcode, reader).map(ReplicaMessage::Control),
425    }
426}
427
428fn decode_control(
429    scope: Scope,
430    opcode: Opcode,
431    mut reader: Reader<'_>,
432) -> Result<Control, WireError> {
433    let message = match opcode {
434        Opcode::PrepareOk => Control::PrepareOk {
435            ack: PrepareOk {
436                scope,
437                durable: reader.prefix()?,
438            },
439        },
440        Opcode::Commit => Control::Commit(Commit {
441            scope,
442            committed: reader.prefix()?,
443        }),
444        Opcode::StartViewChange => {
445            validate_election_view(scope)?;
446            Control::StartViewChange(StartViewChange { scope })
447        }
448        Opcode::ExitView => Control::ExitView(scope),
449        Opcode::DoViewChange => Control::DoViewChange(election::read_report(scope, &mut reader)?),
450        Opcode::StartView => Control::StartView(election::read_start(scope, &mut reader)?),
451        _ => return Err(WireError::UnsupportedCommand),
452    };
453    reader.finish()?;
454    Ok(message)
455}
ozzy_replication/wire.rs

ozzy_replication/
wire.rs
