ozzy_replication/wire/
publication.rs

1//! One shared PREPARE on the group topic; receiver admission remains local.
2
3use super::{
4    PeerBinding, PrepareBatch, ReadFields, Reader, WireError, WireLimits, WriteFields, Writer,
5    prepare,
6};
7use crate::{Prefix, Scope};
8use ozzy_proto::{GroupId, NodeId, Opcode};
9
10/// One group ID is the complete subscription prefix.
11pub const PUBLICATION_TOPIC_BYTES: usize = 16;
12/// Publication prefix size; no recipient masks or per-follower credit fields.
13pub const PUBLICATION_BYTES: usize = PUBLICATION_TOPIC_BYTES;
14
15/// Common topic for every follower of this group.
16pub fn publication_topic(group: GroupId) -> [u8; PUBLICATION_TOPIC_BYTES] {
17    *group.as_bytes()
18}
19
20/// Reframe an actor-owned PREPARE template without copying or hashing its body.
21/// The caller retains the original, already validated payload for publication.
22pub fn encode_publication(
23    sender: NodeId,
24    template: &[&[u8]],
25    committed: Prefix,
26    metadata: &mut [u8],
27    limits: WireLimits,
28) -> Result<super::PrepareEncoding, WireError> {
29    let (packet, skip) = publication_template(sender, template, limits)?;
30    super::validate_prefix(committed)?;
31    let length = packet
32        .metadata
33        .len()
34        .checked_sub(skip)
35        .ok_or(WireError::Peer)?;
36    let header = ozzy_proto::Envelope {
37        opcode: Opcode::PreparePub,
38        response: false,
39        request_id: None,
40        sender,
41        session: None,
42    }
43    .encode_header(length, packet.payload.len(), limits.envelope)?;
44    let (common, tail) = template_fields(packet.metadata, skip)?;
45    let mut writer = Writer::with_len(metadata, length)?;
46    write_head(&mut writer, common, tail, committed)?;
47    writer.bytes(tail.get(80..).ok_or(WireError::Length)?)?;
48    Ok(super::PrepareEncoding {
49        header,
50        metadata_bytes: length,
51        payload_bytes: packet.payload.len(),
52    })
53}
54
55/// Split template metadata into its 80-byte scope and the fields after the
56/// `skip` bytes of a flow epoch: predecessor (40), commit (40), count (4), and
57/// operation descriptors.
58fn template_fields(metadata: &[u8], skip: usize) -> Result<(&[u8], &[u8]), WireError> {
59    let (common, rest) = metadata.split_at_checked(80).ok_or(WireError::Length)?;
60    Ok((common, rest.get(skip..).ok_or(WireError::Length)?))
61}
62
63/// Write the scope and predecessor of `tail` with a new commit prefix.
64fn write_head(
65    writer: &mut Writer<'_>,
66    common: &[u8],
67    tail: &[u8],
68    committed: Prefix,
69) -> Result<(), WireError> {
70    writer.bytes(common)?;
71    writer.bytes(tail.get(..40).ok_or(WireError::Length)?)?;
72    writer.prefix(committed)
73}
74
75// Outgoing actor-owned templates may precede any PEER session. Their metadata
76// was encoded after journal validation. Wire decoding still requires a complete
77// envelope, established binding, and canonical integrity validation.
78fn publication_template<'a>(
79    sender: NodeId,
80    frames: &[&'a [u8]],
81    limits: WireLimits,
82) -> Result<(ozzy_proto::Packet<'a>, usize), WireError> {
83    let packet = if let [header, metadata, payload] = frames
84        && header.is_empty()
85    {
86        let envelope = ozzy_proto::Envelope {
87            opcode: Opcode::PreparePub,
88            response: false,
89            request_id: None,
90            sender,
91            session: None,
92        };
93        envelope.validate_frames(metadata.len(), payload.len(), limits.envelope)?;
94        ozzy_proto::Packet {
95            envelope,
96            metadata,
97            payload,
98        }
99    } else {
100        ozzy_proto::decode_packet(frames, limits.envelope)?
101    };
102    let skip = match packet.envelope.opcode {
103        Opcode::Prepare => 0,
104        Opcode::PreparePub if frames.first().is_some_and(|header| header.is_empty()) => 0,
105        Opcode::PrepareFlow => 16,
106        _ => return Err(WireError::Peer),
107    };
108    if packet.envelope.sender != sender
109        || packet
110            .metadata
111            .get(skip..)
112            .is_none_or(|body| body.len() < 164)
113        || packet.metadata.get(32..48) != Some(sender.as_bytes())
114    {
115        return Err(WireError::Peer);
116    }
117    Ok((packet, skip))
118}
119
120/// Decode only from a separately bound publisher socket. Reject wrong leader,
121/// scope or group before hashing any payload. PUB receipt is
122/// neither admission nor a vote; ordinary receive and core checks still apply.
123pub fn decode_publication<'a>(
124    frames: &[&'a [u8]],
125    binding: PeerBinding,
126    local: NodeId,
127    scope: Scope,
128    limits: WireLimits,
129) -> Result<PrepareBatch<'a>, WireError> {
130    let [topic, frames @ ..] = frames else {
131        return Err(WireError::Peer);
132    };
133    if frames.len() != 3
134        || topic.len() != PUBLICATION_BYTES
135        || binding.peer != binding.configuration.primary(scope.view)
136        || binding.peer == local
137        || *topic != scope.group_id.as_bytes()
138    {
139        return Err(WireError::Peer);
140    }
141    let packet = ozzy_proto::decode_packet(frames, limits.envelope)?;
142    if packet.envelope.sender != binding.peer
143        || packet.envelope.session.is_some()
144        || packet.envelope.opcode != Opcode::PreparePub
145        || packet.envelope.response
146        || packet.envelope.request_id.is_some()
147    {
148        return Err(WireError::Peer);
149    }
150    let mut reader = Reader::new(packet.metadata);
151    if reader.scope(binding)? != scope {
152        return Err(WireError::Peer);
153    }
154    prepare::decode_prepare(scope, reader, packet.payload, limits)
155}
ozzy_replication/wire/publication.rs

ozzy_replication/wire/
publication.rs
