ozzy_replication/wire/
publication.rs

1//! One shared PREPARE on the group topic; receiver admission remains local.
2
3use super::{PeerBinding, PrepareBatch, Reader, WireError, WireLimits, prepare};
4use crate::{Prefix, Scope};
5use ozzy_proto::{GroupId, NodeId, Opcode};
6
7/// One already encoded, consecutive PREPARE retained by its actor. An empty
8/// header denotes actor-owned unbound metadata, never an accepted wire packet.
9#[derive(Debug, Clone, Copy)]
10pub struct PublicationPart<'a> {
11    /// Encoded header, metadata, and payload frames, in that order.
12    pub frames: [&'a [u8]; 3],
13    /// Exact canonical prefix immediately before these operations.
14    pub predecessor: Prefix,
15    /// Exact canonical prefix after these consecutive operations.
16    pub end: Prefix,
17}
18
19struct ValidatedPublicationPart<'a> {
20    metadata: &'a [u8],
21    skip: usize,
22    payload: &'a [u8],
23    count: usize,
24    scope: &'a [u8],
25}
26
27/// One group ID is the complete subscription prefix.
28pub const PUBLICATION_TOPIC_BYTES: usize = 16;
29/// Publication prefix size; no recipient masks or per-follower credit fields.
30pub const PUBLICATION_BYTES: usize = PUBLICATION_TOPIC_BYTES;
31
32/// Common topic for every follower of this group.
33pub fn publication_topic(group: GroupId) -> [u8; PUBLICATION_TOPIC_BYTES] {
34    *group.as_bytes()
35}
36
37/// Reframe an actor-owned PREPARE template without copying or hashing its body.
38/// The caller retains the original, already validated payload for publication.
39pub fn encode_publication(
40    sender: NodeId,
41    template: &[&[u8]],
42    committed: Prefix,
43    metadata: &mut [u8],
44    limits: WireLimits,
45) -> Result<super::PrepareEncoding, WireError> {
46    let (packet, skip) = publication_template(sender, template, limits)?;
47    super::validate_prefix(committed)?;
48    let length = packet.metadata.len() - skip;
49    let header = ozzy_proto::Envelope {
50        opcode: Opcode::PreparePub,
51        response: false,
52        request_id: None,
53        sender,
54        session: None,
55    }
56    .encode_header(length, packet.payload.len(), limits.envelope)?;
57    if metadata.len() < length {
58        return Err(WireError::Capacity);
59    }
60    metadata[..80].copy_from_slice(&packet.metadata[..80]);
61    metadata[80..length].copy_from_slice(&packet.metadata[80 + skip..]);
62    metadata[120..128].copy_from_slice(&committed.op.0.to_be_bytes());
63    metadata[128..160].copy_from_slice(committed.digest.as_bytes());
64    Ok(super::PrepareEncoding {
65        header,
66        metadata_bytes: length,
67        payload_bytes: packet.payload.len(),
68    })
69}
70
71/// Combine consecutive actor-owned PREPARE templates into one publication.
72///
73/// Bodies are copied once into `payload`; canonical bodies and digests are not
74/// decoded, recompressed, or rehashed. Input templates were produced only after
75/// journal validation. This function rechecks framing, bounds, and chain joins.
76pub fn encode_publication_group(
77    sender: NodeId,
78    parts: &[PublicationPart<'_>],
79    committed: Prefix,
80    metadata: &mut [u8],
81    payload: &mut Vec<u8>,
82    limits: WireLimits,
83) -> Result<super::PrepareEncoding, WireError> {
84    if parts.is_empty() {
85        return Err(WireError::Limit);
86    }
87    super::validate_prefix(committed)?;
88    let mut operation_count = 0usize;
89    let mut payload_bytes = 0usize;
90    let mut normalized = Vec::with_capacity(parts.len());
91    let mut previous = parts[0].predecessor;
92    let mut scope = None;
93    for part in parts {
94        let validated = validate_publication_part(sender, part, previous, scope, limits)?;
95        scope.get_or_insert(validated.scope);
96        operation_count = operation_count
97            .checked_add(validated.count)
98            .ok_or(WireError::Limit)?;
99        payload_bytes = payload_bytes
100            .checked_add(validated.payload.len())
101            .ok_or(WireError::Limit)?;
102        if payload_bytes > limits.envelope.max_payload_bytes {
103            return Err(WireError::Limit);
104        }
105        normalized.push((validated.metadata, validated.skip, validated.payload));
106        previous = part.end;
107    }
108    let metadata_bytes = prepare::metadata_length(operation_count, 164, limits)?;
109    if metadata.len() < metadata_bytes || u32::try_from(operation_count).is_err() {
110        return Err(WireError::Capacity);
111    }
112    let (first, first_skip, _) = normalized[0];
113    metadata[..80].copy_from_slice(&first[..80]);
114    metadata[80..160].copy_from_slice(&first[80 + first_skip..160 + first_skip]);
115    metadata[120..128].copy_from_slice(&committed.op.0.to_be_bytes());
116    metadata[128..160].copy_from_slice(committed.digest.as_bytes());
117    metadata[160..164].copy_from_slice(&(operation_count as u32).to_be_bytes());
118    let mut cursor = 164;
119    payload.clear();
120    payload.reserve(payload_bytes);
121    for (batch, skip, body) in normalized {
122        let descriptors = &batch[164 + skip..];
123        metadata[cursor..cursor + descriptors.len()].copy_from_slice(descriptors);
124        cursor += descriptors.len();
125        payload.extend_from_slice(body);
126    }
127    let header = ozzy_proto::Envelope {
128        opcode: Opcode::PreparePub,
129        response: false,
130        request_id: None,
131        sender,
132        session: None,
133    }
134    .encode_header(metadata_bytes, payload_bytes, limits.envelope)?;
135    Ok(super::PrepareEncoding {
136        header,
137        metadata_bytes,
138        payload_bytes,
139    })
140}
141
142fn validate_publication_part<'a>(
143    sender: NodeId,
144    part: &PublicationPart<'a>,
145    previous: Prefix,
146    scope: Option<&[u8]>,
147    limits: WireLimits,
148) -> Result<ValidatedPublicationPart<'a>, WireError> {
149    if part.predecessor != previous {
150        return Err(WireError::Chain);
151    }
152    let count = part
153        .end
154        .op
155        .0
156        .checked_sub(part.predecessor.op.0)
157        .and_then(|count| usize::try_from(count).ok())
158        .filter(|count| *count != 0)
159        .ok_or(WireError::Chain)?;
160    let (packet, skip) = publication_template(sender, &part.frames, limits)?;
161    let common = &packet.metadata[..80];
162    if scope.is_some_and(|expected| expected != common) {
163        return Err(WireError::Scope);
164    }
165    let tail = &packet.metadata[80 + skip..];
166    let first = u64::from_be_bytes(tail[..8].try_into().unwrap());
167    let predecessor = Prefix {
168        op: crate::OpNumber(first.checked_sub(1).ok_or(WireError::Chain)?),
169        digest: crate::Digest::from_bytes(tail[8..40].try_into().unwrap()),
170    };
171    let encoded_count = u32::from_be_bytes(tail[80..84].try_into().unwrap()) as usize;
172    let expected = encoded_count
173        .checked_mul(86)
174        .and_then(|bytes| bytes.checked_add(84))
175        .ok_or(WireError::Limit)?;
176    if predecessor != part.predecessor {
177        return Err(WireError::Chain);
178    }
179    if encoded_count != count || tail.len() != expected {
180        return Err(WireError::Length);
181    }
182    let descriptors = &tail[84..];
183    let body_bytes = descriptors
184        .as_chunks::<86>()
185        .0
186        .iter()
187        .try_fold(0usize, |sum, descriptor| {
188            sum.checked_add(u32::from_be_bytes(descriptor[18..22].try_into().unwrap()) as usize)
189        })
190        .ok_or(WireError::Limit)?;
191    let last = descriptors
192        .rchunks_exact(86)
193        .next()
194        .expect("nonempty batch");
195    if body_bytes != packet.payload.len() || last[54..86] != *part.end.digest.as_bytes() {
196        return Err(WireError::Chain);
197    }
198    Ok(ValidatedPublicationPart {
199        metadata: packet.metadata,
200        skip,
201        payload: packet.payload,
202        count,
203        scope: common,
204    })
205}
206
207// Outgoing actor-owned templates may precede any PEER session. Their metadata
208// was encoded after journal validation. Wire decoding still requires a complete
209// envelope, established binding, and canonical integrity validation.
210fn publication_template<'a>(
211    sender: NodeId,
212    frames: &[&'a [u8]],
213    limits: WireLimits,
214) -> Result<(ozzy_proto::Packet<'a>, usize), WireError> {
215    let packet = if let [header, metadata, payload] = frames
216        && header.is_empty()
217    {
218        let envelope = ozzy_proto::Envelope {
219            opcode: Opcode::PreparePub,
220            response: false,
221            request_id: None,
222            sender,
223            session: None,
224        };
225        envelope.validate_frames(metadata.len(), payload.len(), limits.envelope)?;
226        ozzy_proto::Packet {
227            envelope,
228            metadata,
229            payload,
230        }
231    } else {
232        ozzy_proto::decode_packet(frames, limits.envelope)?
233    };
234    let skip = match packet.envelope.opcode {
235        Opcode::Prepare => 0,
236        Opcode::PreparePub if frames[0].is_empty() => 0,
237        Opcode::PrepareFlow => 16,
238        _ => return Err(WireError::Peer),
239    };
240    if packet.envelope.sender != sender
241        || packet.metadata.len() < 164 + skip
242        || packet.metadata[32..48] != *sender.as_bytes()
243    {
244        return Err(WireError::Peer);
245    }
246    Ok((packet, skip))
247}
248
249/// Decode only from a separately bound publisher socket. Reject wrong leader,
250/// scope or group before hashing any payload. PUB receipt is
251/// neither admission nor a vote; ordinary receive and core checks still apply.
252pub fn decode_publication<'a>(
253    frames: &[&'a [u8]],
254    binding: PeerBinding,
255    local: NodeId,
256    scope: Scope,
257    limits: WireLimits,
258) -> Result<PrepareBatch<'a>, WireError> {
259    if frames.len() != 4
260        || frames[0].len() != PUBLICATION_BYTES
261        || binding.peer != binding.configuration.primary(scope.view)
262        || binding.peer == local
263        || frames[0][..16] != *scope.group_id.as_bytes()
264    {
265        return Err(WireError::Peer);
266    }
267    let packet = ozzy_proto::decode_packet(&frames[1..], limits.envelope)?;
268    if packet.envelope.sender != binding.peer
269        || packet.envelope.session.is_some()
270        || packet.envelope.opcode != Opcode::PreparePub
271        || packet.envelope.response
272        || packet.envelope.request_id.is_some()
273    {
274        return Err(WireError::Peer);
275    }
276    let mut reader = Reader::new(packet.metadata);
277    if reader.scope(binding)? != scope {
278        return Err(WireError::Peer);
279    }
280    prepare::decode_prepare(scope, reader, packet.payload, limits)
281}
ozzy_replication/wire/publication.rs

ozzy_replication/wire/
publication.rs
