zakura_network/zakura/block_sync/
wire.rs

1use super::{config::*, error::*, *};
2
3/// Zakura stream kind reserved for native block sync.
4pub const ZAKURA_STREAM_BLOCK_SYNC: u16 = 6;
5/// Capability bit for the native block-sync service.
6pub const ZAKURA_CAP_BLOCK_SYNC: u64 = 1 << 3;
7/// Version of the native block-sync stream.
8pub const ZAKURA_BLOCK_SYNC_STREAM_VERSION: u16 = 2;
9
10/// Peer status advertisement.
11pub const MSG_BS_STATUS: u8 = 1;
12/// Request a contiguous range of block bodies by height.
13pub const MSG_BS_GET_BLOCKS: u8 = 2;
14/// Respond with one full block body.
15pub const MSG_BS_BLOCK: u8 = 3;
16/// Terminate a `GetBlocks` response.
17pub const MSG_BS_BLOCKS_DONE: u8 = 4;
18/// Report that a requested range is not servable.
19pub const MSG_BS_RANGE_UNAVAILABLE: u8 = 5;
20
21/// Maximum block bodies ever requested or reported by stream 6.
22pub const MAX_BS_BLOCKS_PER_REQUEST: u32 = 128;
23/// Maximum encoded stream-6 message bytes.
24///
25/// This cap is intentionally larger than Zebra's consensus block-size limit so
26/// stream-6 can read and classify slightly oversized or future-expanded frames
27/// in the block-sync codec instead of dropping them at the raw transport gate.
28/// Decoded `Block` messages are still bounded by [`block::MAX_BLOCK_BYTES`].
29pub const MAX_BS_MESSAGE_BYTES: usize = 3 * 1024 * 1024;
30
31pub(super) const BLOCK_SYNC_MESSAGE_TYPE_BYTES: usize = 1;
32
33const _: () = assert!(MAX_BS_MESSAGE_BYTES < 4 * 1024 * 1024);
34const _: () = assert!(MAX_BS_MESSAGE_BYTES > block::MAX_BLOCK_BYTES as usize);
35
36/// Native stream-6 block-sync message.
37#[derive(Clone, Debug, Eq, PartialEq)]
38pub enum BlockSyncMessage {
39    /// Servable range and serving capacity advertisement.
40    Status(BlockSyncStatus),
41    /// Request `count` block bodies starting at `start_height`.
42    GetBlocks {
43        /// First requested height.
44        start_height: block::Height,
45        /// Requested block count.
46        count: u32,
47    },
48    /// One full block body.
49    Block(Arc<block::Block>),
50    /// End of a `GetBlocks` response.
51    BlocksDone {
52        /// First requested height.
53        start_height: block::Height,
54        /// Number of blocks returned.
55        returned: u32,
56    },
57    /// The peer cannot serve this range.
58    RangeUnavailable {
59        /// First unavailable height.
60        start_height: block::Height,
61        /// Unavailable block count.
62        count: u32,
63    },
64}
65
66impl BlockSyncMessage {
67    /// Returns this message's stream-6 discriminator.
68    pub fn message_type(&self) -> u8 {
69        match self {
70            Self::Status(_) => MSG_BS_STATUS,
71            Self::GetBlocks { .. } => MSG_BS_GET_BLOCKS,
72            Self::Block(_) => MSG_BS_BLOCK,
73            Self::BlocksDone { .. } => MSG_BS_BLOCKS_DONE,
74            Self::RangeUnavailable { .. } => MSG_BS_RANGE_UNAVAILABLE,
75        }
76    }
77
78    /// Encode this message as `[u8 message_type][bounded fields...]`.
79    pub fn encode(&self) -> Result<Vec<u8>, BlockSyncWireError> {
80        let mut bytes = Vec::new();
81        bytes.write_u8(self.message_type())?;
82        match self {
83            Self::Status(status) => status.encode_to(&mut bytes)?,
84            Self::GetBlocks {
85                start_height,
86                count,
87            }
88            | Self::RangeUnavailable {
89                start_height,
90                count,
91            } => {
92                validate_block_count(*count)?;
93                write_height(&mut bytes, *start_height)?;
94                bytes.write_u32::<LittleEndian>(*count)?;
95            }
96            Self::Block(block) => {
97                block.zcash_serialize(&mut bytes)?;
98                validate_encoded_block_len(
99                    bytes.len().saturating_sub(BLOCK_SYNC_MESSAGE_TYPE_BYTES),
100                )?;
101            }
102            Self::BlocksDone {
103                start_height,
104                returned,
105            } => {
106                validate_block_count(*returned)?;
107                write_height(&mut bytes, *start_height)?;
108                bytes.write_u32::<LittleEndian>(*returned)?;
109            }
110        }
111        validate_payload_len(bytes.len())?;
112        Ok(bytes)
113    }
114
115    /// Decode a stream-6 message.
116    pub fn decode(bytes: &[u8]) -> Result<Self, BlockSyncWireError> {
117        validate_payload_len(bytes.len())?;
118        let mut reader = Cursor::new(bytes);
119        let message_type = reader.read_u8()?;
120        let message = match message_type {
121            MSG_BS_STATUS => Self::Status(BlockSyncStatus::decode_from(&mut reader)?),
122            MSG_BS_GET_BLOCKS => {
123                let start_height = read_height(&mut reader)?;
124                let count = reader.read_u32::<LittleEndian>()?;
125                validate_block_count(count)?;
126                Self::GetBlocks {
127                    start_height,
128                    count,
129                }
130            }
131            MSG_BS_BLOCK => {
132                let block_start = usize::try_from(reader.position())
133                    .map_err(|_| BlockSyncWireError::NumericOverflow("block payload offset"))?;
134                let block = Arc::new(block::Block::zcash_deserialize(&mut reader)?);
135                let block_end = usize::try_from(reader.position())
136                    .map_err(|_| BlockSyncWireError::NumericOverflow("block payload end"))?;
137                validate_encoded_block_len(block_end.saturating_sub(block_start))?;
138                Self::Block(block)
139            }
140            MSG_BS_BLOCKS_DONE => {
141                let start_height = read_height(&mut reader)?;
142                let returned = reader.read_u32::<LittleEndian>()?;
143                validate_block_count(returned)?;
144                Self::BlocksDone {
145                    start_height,
146                    returned,
147                }
148            }
149            MSG_BS_RANGE_UNAVAILABLE => {
150                let start_height = read_height(&mut reader)?;
151                let count = reader.read_u32::<LittleEndian>()?;
152                validate_block_count(count)?;
153                Self::RangeUnavailable {
154                    start_height,
155                    count,
156                }
157            }
158            value => return Err(BlockSyncWireError::UnknownMessageType(value)),
159        };
160        reject_trailing(bytes, &reader)?;
161        Ok(message)
162    }
163
164    /// Convert this message into a bounded Zakura frame.
165    pub fn encode_frame(&self) -> Result<Frame, BlockSyncWireError> {
166        Ok(Frame {
167            message_type: u16::from(self.message_type()),
168            flags: 0,
169            payload: self.encode()?,
170        })
171    }
172
173    /// Decode this message from a Zakura frame after checking flags and type agreement.
174    pub fn decode_frame(frame: Frame) -> Result<Self, BlockSyncWireError> {
175        Self::decode_frame_with_raw_block_payload(frame).map(|(message, _)| message)
176    }
177
178    /// Decode this message and return the raw frame payload for block bodies.
179    ///
180    /// The returned raw payload includes the stream-6 message type byte. Keeping
181    /// the whole payload lets the decoder move the frame allocation into the
182    /// reorder buffer without copying; consumers skip
183    /// [`BLOCK_SYNC_MESSAGE_TYPE_BYTES`] before deserializing the block body.
184    pub(super) fn decode_frame_with_raw_block_payload(
185        frame: Frame,
186    ) -> Result<(Self, Option<Arc<[u8]>>), BlockSyncWireError> {
187        if frame.flags != 0 {
188            return Err(BlockSyncWireError::UnsupportedFlags(frame.flags));
189        }
190        let frame_message_type = u8::try_from(frame.message_type)
191            .map_err(|_| BlockSyncWireError::UnknownFrameMessageType(frame.message_type))?;
192
193        // If this is a block message, keep the original raw block payload as well;
194        // it can be stored in compact form in the reorder backlog.
195        let (message, raw_block_payload) = if frame_message_type == MSG_BS_BLOCK {
196            let raw_block_payload = Arc::<[u8]>::from(frame.payload.into_boxed_slice());
197            let message = Self::decode(&raw_block_payload)?;
198            (message, Some(raw_block_payload))
199        } else {
200            (Self::decode(&frame.payload)?, None)
201        };
202
203        if frame_message_type != message.message_type() {
204            return Err(BlockSyncWireError::MismatchedFrameMessageType {
205                frame: frame.message_type,
206                payload: message.message_type(),
207            });
208        }
209        Ok((message, raw_block_payload))
210    }
211
212    /// Exact serialized length of a `Block` body, derived from the frame payload
213    /// length the per-peer decode task already has in hand.
214    ///
215    /// A block payload is `[message_type][block serialization]` with no trailing
216    /// bytes — `encode` writes exactly those two parts and `decode` calls
217    /// `reject_trailing`, so the block occupies exactly
218    /// `payload_len - BLOCK_SYNC_MESSAGE_TYPE_BYTES`. Returning this lets the
219    /// reactor account body bytes without re-serializing the whole block on its
220    /// single thread (the per-peer task already paid the deserialization cost).
221    /// Returns `None` for non-block messages.
222    pub(super) fn block_body_wire_bytes(&self, payload_len: usize) -> Option<u64> {
223        matches!(self, Self::Block(_)).then(|| {
224            u64::try_from(payload_len.saturating_sub(BLOCK_SYNC_MESSAGE_TYPE_BYTES))
225                .unwrap_or(u64::MAX)
226        })
227    }
228}
229
230pub(super) fn validate_block_count(count: u32) -> Result<(), BlockSyncWireError> {
231    if count == 0 {
232        return Err(BlockSyncWireError::ZeroBlockCount);
233    }
234    if count > MAX_BS_BLOCKS_PER_REQUEST {
235        return Err(BlockSyncWireError::BlockCountLimit {
236            actual: count,
237            max: MAX_BS_BLOCKS_PER_REQUEST,
238        });
239    }
240    Ok(())
241}
242
243pub(super) fn validate_payload_len(len: usize) -> Result<(), BlockSyncWireError> {
244    if len > MAX_BS_MESSAGE_BYTES {
245        return Err(BlockSyncWireError::OversizedPayload {
246            actual: len,
247            max: MAX_BS_MESSAGE_BYTES,
248        });
249    }
250    Ok(())
251}
252
253fn validate_encoded_block_len(len: usize) -> Result<(), BlockSyncWireError> {
254    let max = usize::try_from(block::MAX_BLOCK_BYTES)
255        .map_err(|_| BlockSyncWireError::NumericOverflow("max block bytes"))?;
256    if len > max {
257        return Err(BlockSyncWireError::OversizedBlock { actual: len, max });
258    }
259    Ok(())
260}
261
262pub(super) fn write_height<W: Write>(
263    writer: &mut W,
264    height: block::Height,
265) -> Result<(), BlockSyncWireError> {
266    writer.write_u32::<LittleEndian>(height.0)?;
267    Ok(())
268}
269
270pub(super) fn read_height<R: Read>(reader: &mut R) -> Result<block::Height, BlockSyncWireError> {
271    let raw = reader.read_u32::<LittleEndian>()?;
272    block::Height::try_from(raw).map_err(|_| BlockSyncWireError::HeightOutOfRange(raw))
273}
274
275pub(super) fn reject_trailing(
276    bytes: &[u8],
277    reader: &Cursor<&[u8]>,
278) -> Result<(), BlockSyncWireError> {
279    let consumed = usize::try_from(reader.position())
280        .map_err(|_| BlockSyncWireError::NumericOverflow("payload cursor"))?;
281    if consumed != bytes.len() {
282        return Err(BlockSyncWireError::TrailingBytes);
283    }
284    Ok(())
285}
zakura_network/zakura/block_sync/wire.rs

zakura_network/zakura/block_sync/
wire.rs
