ozzy_replication/wire/
compact.rs

1//! Same-channel receipt counters. Opening still verifies full history.
2
3use ozzy_proto::{Opcode, layout::Fixed};
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        let bytes: &[u8; COMPACT_STATE_BYTES] = bytes.try_into().map_err(|_| WireError::Length)?;
70        if bytes.u8_at::<0>() != Opcode::ReplicaReceipt as u8 {
71            return Err(WireError::UnsupportedCommand);
72        }
73        let state = Self {
74            handle: bytes.u32_at::<1>(),
75            revision: bytes.u64_at::<5>(),
76            received: OpNumber(bytes.u64_at::<13>()),
77            received_bytes: bytes.u64_at::<21>(),
78        };
79        state.validate()?;
80        Ok(state)
81    }
82
83    /// Restore the fixed opening fields and an independently known prefix.
84    /// The caller validates handle/session, then feeds this into the ordinary
85    /// monotonic sender ledger. Missing history requires a full report.
86    pub fn report(self, bound: Report, received: Prefix) -> Result<Report, WireError> {
87        self.validate()?;
88        if received.op != self.received {
89            return Err(WireError::History);
90        }
91        let report = Report {
92            revision: self.revision,
93            received,
94            received_bytes: self.received_bytes,
95
96            ..bound
97        };
98        report.validate_shape()?;
99        Ok(report)
100    }
101}
ozzy_replication/wire/compact.rs

ozzy_replication/wire/
compact.rs
