ozzy-replication 0.5.0

Deterministic replication state for Ozzy.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

use super::Error;
use crate::{Digest, Scope};
use ozzy_proto::{GroupId, NodeId, layout::Fixed};

/// Exact encoded single-broker configuration length.
pub const CONFIGURATION_BYTES: usize = 128;
const DIGEST_START: usize = 88;

/// Persistent single-broker identity, principal binding and local durability.
/// Encoding is distinct from fixed-three membership. No implicit conversion exists.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Configuration {
    scope: Scope,
    broker: NodeId,
    principal: Digest,
}

impl Configuration {
    /// Validate nonzero identity, epoch, and principal; bind them in a checksummed scope.
    pub fn new(
        group: GroupId,
        epoch: u64,
        broker: NodeId,
        principal: Digest,
    ) -> Result<Self, Error> {
        if group.as_bytes() == &[0; 16]
            || broker.as_bytes() == &[0; 16]
            || epoch == 0
            || principal == Digest::ZERO
        {
            return Err(Error::Configuration);
        }
        let mut configuration = Self {
            scope: Scope {
                group_id: group,
                configuration_epoch: epoch,
                configuration_digest: Digest::ZERO,
                view: 0,
            },
            broker,
            principal,
        };
        configuration.scope.configuration_digest = checksum(&configuration.encode());
        Ok(configuration)
    }

    /// Persistent local scope; its view is always zero.
    pub const fn scope(self) -> Scope {
        self.scope
    }
    /// The sole configured broker identity.
    pub const fn broker(self) -> NodeId {
        self.broker
    }
    /// Persistent principal binding; it does not authenticate a connection.
    pub const fn principal(self) -> Digest {
        self.principal
    }

    /// Fixed fields and integers in network byte order. Reserved bytes stay zero.
    pub fn encode(self) -> [u8; CONFIGURATION_BYTES] {
        let mut bytes = [0; CONFIGURATION_BYTES];
        bytes[..8].copy_from_slice(b"OZYLOCAL");
        bytes[8..12].copy_from_slice(&[
            ozzy_journal::integrity::PROFILE,
            ozzy_proto::VERSION,
            1,
            1,
        ]);
        bytes[16..32].copy_from_slice(self.scope.group_id.as_bytes());
        bytes[32..40].copy_from_slice(&self.scope.configuration_epoch.to_be_bytes());
        bytes[40..56].copy_from_slice(self.broker.as_bytes());
        bytes[56..88].copy_from_slice(self.principal.as_bytes());
        bytes[DIGEST_START..120].copy_from_slice(self.scope.configuration_digest.as_bytes());
        bytes
    }

    /// Reject corruption, unsupported fields, or replicated membership records.
    pub fn decode(bytes: &[u8]) -> Result<Self, Error> {
        let bytes: &[u8; CONFIGURATION_BYTES] = bytes.try_into().map_err(|_| Error::Encoding)?;
        let configuration = Self::new(
            GroupId::from_bytes(bytes.array::<16, 16>()),
            bytes.u64_at::<32>(),
            NodeId::from_bytes(bytes.array::<40, 16>()),
            Digest::from_bytes(bytes.array::<56, 32>()),
        )?;
        if *bytes != configuration.encode() {
            return Err(Error::Encoding);
        }
        Ok(configuration)
    }
}

fn checksum(bytes: &[u8; CONFIGURATION_BYTES]) -> Digest {
    let mut hasher =
        ozzy_journal::integrity::IntegrityHasher::new("ozzy single broker configuration");
    hasher.update(&bytes[..DIGEST_START]);
    hasher.update(&[0; 32]);
    hasher.update(&bytes[120..]);
    hasher.finish()
}