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

ozzy_replication/
wire.rs
