ozzy_replication/
configuration_record.rs

1use ozzy_proto::{GroupId, NodeId, layout::Fixed};
2
3use crate::{Configuration, Digest, ReplicationError};
4
5/// Exact encoded length of a fixed-three-voter durable configuration.
6pub const CONFIGURATION_RECORD_BYTES: usize = 256;
7const HASH_CONTEXT: &str = "ozzy fixed durable replica configuration v1";
8const MAGIC: &[u8; 8] = b"OZYVOTER";
9const DIGEST_START: usize = 192;
10// Version 2 (XXH3-128 integrity), length 256, external principal binding,
11// current native wire version, canonical operation schema 1, reserved zero.
12const HEADER: [u8; 8] = [0, 2, 1, 0, 1, ozzy_proto::VERSION, 1, 0];
13
14/// One persistent voter and its independently authenticated principal binding.
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub struct ConfiguredVoter {
17    /// Stable node identity. Its array position participates in primary selection.
18    pub node_id: NodeId,
19    /// Fingerprint of the adapter's canonical authentication-principal binding.
20    /// This is not a socket identity, endpoint hash, or proof of authentication.
21    /// All replicas must use the same fingerprint definition and mapping.
22    pub principal: Digest,
23}
24
25/// Canonical immutable configuration for three authenticated durable voters.
26///
27/// The record fixes the native wire version and canonical operation schema 1. Endpoint addresses and local tuning are deliberately excluded.
28/// Authentication still belongs to the transport adapter: loading this record
29/// does not authenticate a peer or authorize a restarted voter to enter normal.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub struct ConfigurationRecord {
32    configuration: Configuration,
33    voters: [ConfiguredVoter; 3],
34}
35
36impl ConfigurationRecord {
37    /// Construct a complete fixed configuration and derive its scope digest.
38    /// Voter IDs and principal fingerprints must be nonzero and pairwise distinct.
39    pub fn new(
40        group_id: GroupId,
41        configuration_epoch: u64,
42        voters: [ConfiguredVoter; 3],
43    ) -> Result<Self, ConfigurationRecordError> {
44        if configuration_epoch == 0
45            || voters.iter().any(|voter| voter.principal == Digest::ZERO)
46            || voters[0].principal == voters[1].principal
47            || voters[0].principal == voters[2].principal
48            || voters[1].principal == voters[2].principal
49        {
50            return Err(ConfigurationRecordError::InvalidConfiguration);
51        }
52        let bytes = unsigned_record(group_id, configuration_epoch, &voters);
53        let configuration = Configuration::new(
54            group_id,
55            configuration_epoch,
56            record_digest(&bytes),
57            voters.map(|voter| voter.node_id),
58        )?;
59        Ok(Self {
60            configuration,
61            voters,
62        })
63    }
64
65    /// Verified group, ordered voter identities, and complete configuration digest.
66    pub const fn configuration(self) -> Configuration {
67        self.configuration
68    }
69
70    /// Ordered voter-to-principal bindings the transport adapter must enforce.
71    pub const fn voters(&self) -> &[ConfiguredVoter; 3] {
72        &self.voters
73    }
74
75    /// Encode without allocation. All integer fields use network byte order.
76    pub fn encode(self) -> [u8; CONFIGURATION_RECORD_BYTES] {
77        let scope = self.configuration.scope();
78        let mut bytes = unsigned_record(scope.group_id, scope.configuration_epoch, &self.voters);
79        bytes[DIGEST_START..DIGEST_START + 32]
80            .copy_from_slice(scope.configuration_digest.as_bytes());
81        bytes
82    }
83
84    /// Validate exact length, supported versions, canonical fields, and digest.
85    /// Unknown fields and malformed membership are rejected, never normalized.
86    pub fn decode(bytes: &[u8]) -> Result<Self, ConfigurationRecordError> {
87        let bytes: &[u8; CONFIGURATION_RECORD_BYTES] = bytes
88            .try_into()
89            .map_err(|_| ConfigurationRecordError::Length)?;
90        if bytes.array::<0, 8>() != *MAGIC
91            || bytes.array::<8, 8>() != HEADER
92            || bytes.u8_at::<40>() != 3
93            || bytes.slice::<41, 7>().iter().any(|&byte| byte != 0)
94            || bytes.slice::<224, 32>().iter().any(|&byte| byte != 0)
95        {
96            return Err(ConfigurationRecordError::UnsupportedFields);
97        }
98        let stored = Digest::from_bytes(bytes.array::<DIGEST_START, 32>());
99        if record_digest(bytes) != stored {
100            return Err(ConfigurationRecordError::DigestMismatch);
101        }
102        let group = GroupId::from_bytes(bytes.array::<16, 16>());
103        let epoch = bytes.u64_at::<32>();
104        let voters = [
105            bytes.array::<48, 48>(),
106            bytes.array::<96, 48>(),
107            bytes.array::<144, 48>(),
108        ]
109        .map(|entry| ConfiguredVoter {
110            node_id: NodeId::from_bytes(entry.array::<0, 16>()),
111            principal: Digest::from_bytes(entry.array::<16, 32>()),
112        });
113        let record = Self::new(group, epoch, voters)?;
114        debug_assert_eq!(record.configuration.scope().configuration_digest, stored);
115        Ok(record)
116    }
117}
118
119fn unsigned_record(
120    group: GroupId,
121    epoch: u64,
122    voters: &[ConfiguredVoter; 3],
123) -> [u8; CONFIGURATION_RECORD_BYTES] {
124    let mut bytes = [0; CONFIGURATION_RECORD_BYTES];
125    bytes[..8].copy_from_slice(MAGIC);
126    bytes[8..16].copy_from_slice(&HEADER);
127    bytes[16..32].copy_from_slice(group.as_bytes());
128    bytes[32..40].copy_from_slice(&epoch.to_be_bytes());
129    bytes[40] = 3;
130    // Voters fill 48-byte entries 1 through 3.
131    let entries = bytes.as_chunks_mut::<48>().0.iter_mut().skip(1);
132    for (entry, voter) in entries.zip(voters) {
133        entry.put::<0, 16>(*voter.node_id.as_bytes());
134        entry.put::<16, 32>(*voter.principal.as_bytes());
135    }
136    bytes
137}
138
139fn record_digest(bytes: &[u8; CONFIGURATION_RECORD_BYTES]) -> Digest {
140    let mut hasher = ozzy_journal::integrity::IntegrityHasher::new(HASH_CONTEXT);
141    hasher.update(bytes.slice::<0, DIGEST_START>());
142    hasher.update(&[0; 32]);
143    hasher.update(bytes.slice::<{ DIGEST_START + 32 }, 32>());
144    hasher.finish()
145}
146
147/// Rejected durable configuration. No error permits silently replacing membership.
148#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
149pub enum ConfigurationRecordError {
150    /// Record must have the one exact supported length.
151    #[error("invalid replica configuration record length")]
152    Length,
153    /// Magic, versions, policy, voter count, or reserved fields are unsupported.
154    #[error("unsupported replica configuration record fields")]
155    UnsupportedFields,
156    /// Stored configuration bytes no longer match their domain-separated digest.
157    #[error("replica configuration record digest mismatch")]
158    DigestMismatch,
159    /// Zero epoch/principal or repeated principal binding.
160    #[error("invalid replica configuration principals or epoch")]
161    InvalidConfiguration,
162    /// Group or ordered voter identities violate the replication core's rules.
163    #[error(transparent)]
164    Replication(#[from] ReplicationError),
165}
ozzy_replication/configuration_record.rs

ozzy_replication/
configuration_record.rs
