ozzy_replication/wire/
prepare.rs

1//! Bounded canonical operation batches, borrowing metadata and payload frames.
2
3use std::slice::Iter;
4
5use ozzy_journal::operation::{
6    CanonicalOperation, OperationKind, canonical_body_digest,
7    logical_operation_digest_with_body_digest,
8};
9use ozzy_proto::NodeId;
10use ozzy_proto::{Envelope, Opcode};
11
12use super::{
13    COMMON_BYTES, EncodeContext, Reader, WireError, WireLimits, Writer, validate_prefix,
14    validate_scope,
15};
16use super::{ReadFields, WriteFields};
17use crate::flow::ReceiveEpoch;
18use crate::{Digest, OpNumber, Prefix, Scope};
19
20mod plan;
21use plan::Planned;
22
23pub(super) const DESCRIPTOR_BYTES: usize = 86;
24pub(super) const BATCH_BYTES: usize = COMMON_BYTES + 8 + 32 + 40 + 4;
25
26/// Immutable canonical bytes and their already computed body/envelope digests.
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub struct Operation<'a> {
29    canonical: CanonicalOperation<'a>,
30    body_digest: Digest,
31    digest: Digest,
32}
33
34impl<'a> Operation<'a> {
35    /// Reuse an independently verified body digest, hashing only its envelope.
36    ///
37    /// The caller asserts body bytes/digest and canonical application transition
38    /// have been validated. This constructor deliberately does not rehash payload.
39    pub fn from_verified(canonical: CanonicalOperation<'a>, body_digest: Digest) -> Self {
40        Self {
41            digest: logical_operation_digest_with_body_digest(&canonical, body_digest),
42            canonical,
43            body_digest,
44        }
45    }
46    /// Borrow the exact canonical body and envelope fields.
47    pub const fn canonical(self) -> CanonicalOperation<'a> {
48        self.canonical
49    }
50    /// Verified body digest, reusable by journal encoding after application validation.
51    pub const fn body_digest(self) -> Digest {
52        self.body_digest
53    }
54    /// Complete chain position at this operation.
55    pub const fn prefix(self) -> Prefix {
56        Prefix {
57            op: OpNumber(self.canonical.op_number),
58            digest: self.digest,
59        }
60    }
61}
62
63/// Decoder-verified body and envelope, borrowing the exact immutable packet bytes.
64///
65/// Only a successfully decoded batch can produce this proof. Unlike
66/// [`Operation::from_verified`], there is no caller-asserted constructor. It
67/// allows admission to reuse hashing, not to skip schema, state, role, scope,
68/// history, or capacity validation. It is neither authentication nor a vote.
69///
70/// A caller-asserted outgoing operation cannot be promoted into a receive proof:
71///
72/// ```compile_fail
73/// use ozzy_replication::wire::{Operation, VerifiedOperation};
74/// fn trust(operation: Operation<'_>) -> VerifiedOperation<'_> {
75///     operation.into()
76/// }
77/// ```
78#[derive(Debug, Clone, Copy, PartialEq, Eq)]
79pub struct VerifiedOperation<'a>(Operation<'a>);
80
81impl<'a> VerifiedOperation<'a> {
82    /// Borrow the body and envelope covered by the decoder's integrity checks.
83    pub const fn canonical(self) -> CanonicalOperation<'a> {
84        self.0.canonical()
85    }
86
87    /// Body digest recomputed and matched by the decoder, not merely supplied.
88    pub const fn body_digest(self) -> Digest {
89        self.0.body_digest()
90    }
91
92    /// Hash-validated chain position, not a retained or committed boundary.
93    pub const fn prefix(self) -> Prefix {
94        self.0.prefix()
95    }
96}
97
98/// One bounded outbound prepare, borrowing prevalidated canonical operations.
99#[derive(Debug, Clone, Copy)]
100pub struct Prepare<'a> {
101    /// Active view; each operation preserves its own original view.
102    pub scope: Scope,
103    /// Primary's piggyback commit knowledge, independently checked by the core.
104    pub committed: Prefix,
105    /// Consecutive whole operations; never split a canonical operation on the wire.
106    pub operations: &'a [Operation<'a>],
107}
108
109/// Exact prefixes of caller-owned metadata/payload buffers for PREPARE or OPS.
110#[derive(Debug, Clone, Copy, PartialEq, Eq)]
111pub struct PrepareEncoding {
112    /// First Ozzy frame.
113    pub header: [u8; 64],
114    /// Valid second-frame bytes.
115    pub metadata_bytes: usize,
116    /// Valid third-frame bytes.
117    pub payload_bytes: usize,
118}
119
120/// Canonical metadata encoded before a transport session or receive epoch is
121/// available. This is not a sendable packet and contains no envelope header.
122#[derive(Debug, Clone, Copy, PartialEq, Eq)]
123pub struct PrepareMetadata {
124    /// Exact metadata prefix in the supplied buffer, without a receive epoch.
125    pub metadata_bytes: usize,
126    /// Exact length of the shared canonical body backing to pair with it.
127    pub payload_bytes: usize,
128}
129
130/// Encode bounded canonical descriptors without inventing a link session.
131/// Bind a real session, receive epoch, and current commit before transmission.
132/// Invalid fields or capacity leave the output unchanged, as for wire encoding.
133pub fn encode_prepare_unbound(
134    sender: NodeId,
135    prepare: Prepare<'_>,
136    metadata: &mut [u8],
137    limits: WireLimits,
138) -> Result<PrepareMetadata, WireError> {
139    Planned::new(sender, None, prepare, limits)?.write(metadata, None)
140}
141
142/// Encode into reusable caller-owned buffers, preserving them on rejection.
143///
144/// Reuses the supplied body/envelope digests; it does not recompress or rehash
145/// payloads. The caller must retain the exact output until transport releases it.
146pub fn encode_prepare(
147    context: EncodeContext,
148    prepare: Prepare<'_>,
149    metadata: &mut [u8],
150    payload: &mut [u8],
151) -> Result<PrepareEncoding, WireError> {
152    Planned::new(context.sender, None, prepare, context.limits)?.bind(
153        context,
154        metadata,
155        Some(payload),
156    )
157}
158
159/// Encode only PREPARE framing. Pair with the identical canonical body backing.
160pub fn encode_prepare_metadata(
161    context: EncodeContext,
162    prepare: Prepare<'_>,
163    metadata: &mut [u8],
164) -> Result<PrepareEncoding, WireError> {
165    Planned::new(context.sender, None, prepare, context.limits)?.bind(context, metadata, None)
166}
167
168/// Encode epoch-bound normal data with a mandatory receiver-issued epoch.
169///
170/// Distinct `PREPARE_FLOW` bytes prevent silent mixing with legacy PREPARE schemas.
171/// Adds exactly 16 metadata bytes per batch. Canonical identities/body digests
172/// are unchanged, and no additional body hashing or allocation is performed.
173pub fn encode_flow_prepare(
174    context: EncodeContext,
175    epoch: ReceiveEpoch,
176    prepare: Prepare<'_>,
177    metadata: &mut [u8],
178    payload: &mut [u8],
179) -> Result<PrepareEncoding, WireError> {
180    Planned::new(context.sender, Some(epoch), prepare, context.limits)?.bind(
181        context,
182        metadata,
183        Some(payload),
184    )
185}
186
187/// Encode only the header and metadata of a credited `PREPARE_FLOW`.
188///
189/// The payload frame of a prepare is the concatenation of its canonical bodies
190/// and does not depend on the receiver or its epoch. A sender that already
191/// encoded that exact payload for another peer can pair it with this metadata
192/// instead of copying the bodies again. `payload_bytes` reports the length the
193/// paired payload must have.
194pub fn encode_flow_prepare_metadata(
195    context: EncodeContext,
196    epoch: ReceiveEpoch,
197    prepare: Prepare<'_>,
198    metadata: &mut [u8],
199) -> Result<PrepareEncoding, WireError> {
200    Planned::new(context.sender, Some(epoch), prepare, context.limits)?
201        .bind(context, metadata, None)
202}
203
204/// Hash-validated prepare borrowing the packet's backing buffers.
205///
206/// No per-operation vector is allocated. Canonical body schemas and application
207/// transitions still need validation before constructing core prepare metadata.
208#[derive(Debug, Clone, Copy, PartialEq, Eq)]
209pub struct PrepareBatch<'a> {
210    committed: Prefix,
211    batch: DecodedBatch<'a>,
212}
213
214impl<'a> PrepareBatch<'a> {
215    /// Active primary view and immutable group configuration.
216    pub const fn scope(self) -> Scope {
217        self.batch.scope
218    }
219    /// Piggyback commit claim, not local commit authority.
220    pub const fn committed(self) -> Prefix {
221        self.committed
222    }
223    /// Exact prefix immediately before the first operation.
224    pub const fn predecessor(self) -> Prefix {
225        self.batch.predecessor
226    }
227    /// Exact hash-validated end of this batch, not a durable or committed boundary.
228    pub const fn end(self) -> Prefix {
229        self.batch.accepted
230    }
231    /// Iterate verified descriptors and borrowed bodies without hashing again.
232    pub fn operations(self) -> Operations<'a> {
233        self.batch.operations()
234    }
235
236    /// Iterate immutable decoder proofs without allocating or rehashing bodies.
237    pub fn verified_operations(
238        self,
239    ) -> impl ExactSizeIterator<Item = VerifiedOperation<'a>> + std::iter::FusedIterator {
240        self.operations().map(VerifiedOperation)
241    }
242}
243
244#[derive(Debug, Clone, Copy, PartialEq, Eq)]
245pub(super) struct DecodedBatch<'a> {
246    pub scope: Scope,
247    pub predecessor: Prefix,
248    pub accepted: Prefix,
249    descriptors: &'a [u8],
250    pub payload: &'a [u8],
251}
252
253impl<'a> DecodedBatch<'a> {
254    pub(super) fn operations(self) -> Operations<'a> {
255        Operations {
256            descriptors: self.descriptors.as_chunks::<DESCRIPTOR_BYTES>().0.iter(),
257            payload: self.payload,
258            scope: self.scope,
259            previous: self.predecessor,
260        }
261    }
262}
263
264/// Allocation-free iteration over an already checked operation batch.
265#[derive(Debug, Clone)]
266pub struct Operations<'a> {
267    descriptors: Iter<'a, [u8; DESCRIPTOR_BYTES]>,
268    payload: &'a [u8],
269    scope: Scope,
270    previous: Prefix,
271}
272
273impl<'a> Iterator for Operations<'a> {
274    type Item = Operation<'a>;
275    #[expect(
276        clippy::expect_used,
277        reason = "decode_batch checked every descriptor and body length"
278    )]
279    fn next(&mut self) -> Option<Self::Item> {
280        let descriptor = self.descriptors.next()?;
281        let mut reader = Reader::new(descriptor);
282        let operation = read_operation(&mut reader, &mut self.payload, self.scope, self.previous)
283            .expect("validated batch");
284        self.previous = operation.prefix();
285        Some(operation)
286    }
287    fn size_hint(&self) -> (usize, Option<usize>) {
288        self.descriptors.size_hint()
289    }
290}
291
292impl ExactSizeIterator for Operations<'_> {}
293impl std::iter::FusedIterator for Operations<'_> {}
294
295pub(super) fn decode_prepare<'a>(
296    scope: Scope,
297    mut reader: Reader<'a>,
298    payload: &'a [u8],
299    limits: WireLimits,
300) -> Result<PrepareBatch<'a>, WireError> {
301    let first_op = reader.u64()?;
302    let predecessor = Prefix {
303        op: OpNumber(first_op.checked_sub(1).ok_or(WireError::Chain)?),
304        digest: reader.digest()?,
305    };
306    validate_prefix(predecessor)?;
307    let committed = reader.prefix()?;
308    let chain = Chain {
309        scope,
310        predecessor,
311        committed: Some(committed),
312    };
313    let batch = decode_batch(chain, reader, payload, limits)?;
314    Ok(PrepareBatch { committed, batch })
315}
316
317/// Chain position a batch continues, and the commit claim it must not contradict.
318#[derive(Debug, Clone, Copy)]
319pub(super) struct Chain {
320    pub scope: Scope,
321    pub predecessor: Prefix,
322    pub committed: Option<Prefix>,
323}
324
325impl Chain {
326    fn check_start(self) -> Result<(), WireError> {
327        validate_prefix(self.predecessor)?;
328        self.check_commit(self.predecessor)
329    }
330
331    fn check_commit(self, position: Prefix) -> Result<(), WireError> {
332        match self.committed {
333            Some(committed) if committed.op == position.op && committed != position => {
334                Err(WireError::Chain)
335            }
336            _ => Ok(()),
337        }
338    }
339}
340
341pub(super) fn decode_batch<'a>(
342    chain: Chain,
343    mut reader: Reader<'a>,
344    mut payload: &'a [u8],
345    limits: WireLimits,
346) -> Result<DecodedBatch<'a>, WireError> {
347    chain.check_start()?;
348    let Chain {
349        scope, predecessor, ..
350    } = chain;
351    let count = usize::try_from(reader.u32()?).map_err(|_| WireError::Limit)?;
352    let descriptor_bytes = metadata_length(count, 0, limits)?;
353    if reader.rest().len() != descriptor_bytes {
354        return Err(WireError::Length);
355    }
356    let descriptors = reader.rest();
357    let original_payload = payload;
358    let mut previous = predecessor;
359    for _ in 0..count {
360        let operation = read_operation(&mut reader, &mut payload, scope, previous)?;
361        validate_step(scope, previous, operation)?;
362        if canonical_body_digest(operation.canonical.body) != operation.body_digest
363            || logical_operation_digest_with_body_digest(
364                &operation.canonical,
365                operation.body_digest,
366            ) != operation.digest
367        {
368            return Err(WireError::Digest);
369        }
370        chain.check_commit(operation.prefix())?;
371        previous = operation.prefix();
372    }
373    if !payload.is_empty() {
374        return Err(WireError::Payload);
375    }
376    reader.finish()?;
377    Ok(DecodedBatch {
378        scope,
379        predecessor,
380        accepted: previous,
381        descriptors,
382        payload: original_payload,
383    })
384}
385
386fn read_operation<'a>(
387    reader: &mut Reader<'_>,
388    payload: &mut &'a [u8],
389    scope: Scope,
390    previous: Prefix,
391) -> Result<Operation<'a>, WireError> {
392    let op_number = reader.u64()?;
393    let original_view = reader.u64()?;
394    let kind = u16::from_be_bytes(reader.array()?);
395    let kind = OperationKind::try_from(kind).map_err(|_| WireError::OperationKind(kind))?;
396    let bytes = usize::try_from(reader.u32()?).map_err(|_| WireError::Limit)?;
397    if bytes == 0 {
398        return Err(WireError::Limit);
399    }
400    let body_digest = reader.digest()?;
401    let digest = reader.digest()?;
402    let (body, rest) = payload.split_at_checked(bytes).ok_or(WireError::Length)?;
403    *payload = rest;
404    Ok(Operation {
405        canonical: CanonicalOperation {
406            group_id: scope.group_id,
407            configuration_epoch: scope.configuration_epoch,
408            original_view,
409            op_number,
410            previous_digest: previous.digest,
411            kind,
412            body,
413        },
414        body_digest,
415        digest,
416    })
417}
418
419fn validate_step(
420    scope: Scope,
421    previous: Prefix,
422    operation: Operation<'_>,
423) -> Result<(), WireError> {
424    let canonical = operation.canonical;
425    validate_prefix(operation.prefix())?;
426    if canonical.group_id != scope.group_id
427        || canonical.configuration_epoch != scope.configuration_epoch
428        || canonical.original_view > scope.view
429        || previous.op.0.checked_add(1) != Some(canonical.op_number)
430        || canonical.previous_digest != previous.digest
431    {
432        return Err(WireError::Chain);
433    }
434    Ok(())
435}
436
437pub(super) fn metadata_length(
438    count: usize,
439    fixed_bytes: usize,
440    limits: WireLimits,
441) -> Result<usize, WireError> {
442    if count == 0 || count > limits.max_operations || u32::try_from(count).is_err() {
443        return Err(WireError::Limit);
444    }
445    let bytes = count
446        .checked_mul(DESCRIPTOR_BYTES)
447        .and_then(|bytes| bytes.checked_add(fixed_bytes))
448        .ok_or(WireError::Limit)?;
449    if bytes > limits.envelope.max_metadata_bytes {
450        return Err(WireError::Limit);
451    }
452    Ok(bytes)
453}
454
455pub(super) fn validate_batch(
456    chain: Chain,
457    operations: &[Operation<'_>],
458    limits: WireLimits,
459) -> Result<usize, WireError> {
460    metadata_length(operations.len(), 0, limits)?;
461    chain.check_start()?;
462    let mut previous = chain.predecessor;
463    let mut bytes = 0usize;
464    for &operation in operations {
465        validate_step(chain.scope, previous, operation)?;
466        chain.check_commit(operation.prefix())?;
467        bytes = bytes
468            .checked_add(operation.canonical.body.len())
469            .ok_or(WireError::Limit)?;
470        if bytes > limits.envelope.max_payload_bytes || operation.canonical.body.is_empty() {
471            return Err(WireError::Limit);
472        }
473        previous = operation.prefix();
474    }
475    Ok(bytes)
476}
477
478// Call only after complete count/length validation and output capacity checks.
479// `None` writes the per-operation metadata without copying bodies, for a
480// caller that pairs the metadata with an already encoded identical payload.
481pub(super) fn write_batch(
482    writer: &mut Writer<'_>,
483    mut payload: Option<&mut [u8]>,
484    operations: &[Operation<'_>],
485) -> Result<(), WireError> {
486    writer.bytes(&(operations.len() as u32).to_be_bytes())?;
487    let mut offset = 0_usize;
488    for &operation in operations {
489        let canonical = operation.canonical;
490        writer.u64(canonical.op_number)?;
491        writer.u64(canonical.original_view)?;
492        writer.bytes(&(canonical.kind as u16).to_be_bytes())?;
493        writer.bytes(&(canonical.body.len() as u32).to_be_bytes())?;
494        writer.bytes(operation.body_digest.as_bytes())?;
495        writer.bytes(operation.digest.as_bytes())?;
496        let end = offset
497            .checked_add(canonical.body.len())
498            .ok_or(WireError::Length)?;
499        if let Some(payload) = payload.as_deref_mut() {
500            let body = payload.get_mut(offset..end).ok_or(WireError::Capacity)?;
501            body.copy_from_slice(canonical.body);
502        }
503        offset = end;
504    }
505    Ok(())
506}
ozzy_replication/wire/prepare.rs

ozzy_replication/wire/
prepare.rs
