ozzy_replication/wire/
compact.rs

1//! Same-channel receipt counters. Opening still verifies full history.
2
3use ozzy_proto::Opcode;
4
5use crate::{
6    OpNumber, Prefix,
7    flow::{FlowError, Report},
8};
9
10use super::WireError;
11
12/// One single-part message, small enough for OMQ's routed inline storage.
13pub const COMPACT_STATE_BYTES: usize = 29;
14
15/// An authenticated channel handle plus cumulative receive counters.
16/// Handles bind peer, link session, configuration, view, receive epoch and base
17/// at full channel opening. They must never be recycled within a link session.
18#[derive(Clone, Copy, Debug, PartialEq, Eq)]
19pub struct CompactState {
20    /// Nonzero alias issued at full channel opening.
21    pub handle: u32,
22    /// Monotonic receiver report revision.
23    pub revision: u64,
24    /// Last retained operation, never a persistence vote.
25    pub received: OpNumber,
26    /// Cumulative canonical bytes retained within this channel.
27    pub received_bytes: u64,
28}
29
30impl CompactState {
31    /// Copy counters without changing their confirmation or durability meaning.
32    pub fn from_report(handle: u32, report: Report) -> Result<Self, WireError> {
33        report.validate_shape()?;
34        let state = Self {
35            handle,
36            revision: report.revision,
37            received: report.received.op,
38            received_bytes: report.received_bytes,
39        };
40        state.validate()?;
41        Ok(state)
42    }
43
44    fn validate(self) -> Result<(), WireError> {
45        if self.handle == 0 || self.revision == 0 || self.received.0 == u64::MAX {
46            return Err(FlowError::Report.into());
47        }
48        Ok(())
49    }
50
51    /// Exact network order: opcode, u32 handle, five u64 counters.
52    pub fn encode(self) -> Result<[u8; COMPACT_STATE_BYTES], WireError> {
53        self.validate()?;
54        let mut bytes = [0; COMPACT_STATE_BYTES];
55        bytes[0] = Opcode::ReplicaReceipt as u8;
56        bytes[1..5].copy_from_slice(&self.handle.to_be_bytes());
57        for (chunk, value) in bytes[5..].as_chunks_mut::<8>().0.iter_mut().zip([
58            self.revision,
59            self.received.0,
60            self.received_bytes,
61        ]) {
62            chunk.copy_from_slice(&value.to_be_bytes());
63        }
64        Ok(bytes)
65    }
66
67    /// Reject extensions and truncations before interpreting any counters.
68    pub fn decode(bytes: &[u8]) -> Result<Self, WireError> {
69        if bytes.len() != COMPACT_STATE_BYTES {
70            return Err(WireError::Length);
71        }
72        if bytes[0] != Opcode::ReplicaReceipt as u8 {
73            return Err(WireError::UnsupportedCommand);
74        }
75        let read = |offset| u64::from_be_bytes(bytes[offset..offset + 8].try_into().unwrap());
76        let state = Self {
77            handle: u32::from_be_bytes(bytes[1..5].try_into().unwrap()),
78            revision: read(5),
79            received: OpNumber(read(13)),
80            received_bytes: read(21),
81        };
82        state.validate()?;
83        Ok(state)
84    }
85
86    /// Restore the fixed opening fields and an independently known prefix.
87    /// The caller validates handle/session, then feeds this into the ordinary
88    /// monotonic sender ledger. Missing history requires a full report.
89    pub fn report(self, bound: Report, received: Prefix) -> Result<Report, WireError> {
90        self.validate()?;
91        if received.op != self.received {
92            return Err(WireError::History);
93        }
94        let report = Report {
95            revision: self.revision,
96            received,
97            received_bytes: self.received_bytes,
98
99            ..bound
100        };
101        report.validate_shape()?;
102        Ok(report)
103    }
104}
ozzy_replication/wire/compact.rs

ozzy_replication/wire/
compact.rs
