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

ozzy_replication/wire/
prepare.rs
