zakura_network/zakura/header_sync/
wire.rs

1use super::{config::*, error::*, events::HeaderSyncRequestId, validation::*, *};
2
3/// Zakura stream kind reserved for native header sync.
4pub const ZAKURA_STREAM_HEADER_SYNC: u16 = 5;
5/// Version of the native header-sync stream.
6///
7/// Version 4 carries one tree-aux root for each non-empty range header.
8/// Version 6 extends each tree-aux root with the full set of per-block ZIP-221 history-leaf
9/// inputs a recipient needs to rebuild the leaf and verify the roots against its own header
10/// commitments during header sync, without the block body: the Ironwood note-commitment
11/// root, the three per-pool shielded transaction counts (Sapling/Orchard/Ironwood), and the
12/// block's ZIP-244 `auth_data_root` (the co-input to its NU5+ header commitment).
13/// Version 7 prefixes `GetHeaders`/`Headers` with a non-zero request ID so a response is
14/// correlated with the exact request that solicited it, instead of by FIFO arrival order.
15/// Each of these is a breaking wire change: a peer that speaks an earlier version cannot
16/// exchange header-sync ranges with version 7 and does not negotiate header sync at all.
17pub const ZAKURA_HEADER_SYNC_STREAM_VERSION: u16 = 7;
18
19/// Peer status advertisement.
20pub const MSG_HS_STATUS: u8 = 1;
21/// Request a contiguous range of headers by height.
22pub const MSG_HS_GET_HEADERS: u8 = 2;
23/// Respond with a contiguous run of headers.
24pub const MSG_HS_HEADERS: u8 = 3;
25/// Flood a newly seen tip block, including its full body.
26pub const MSG_HS_NEW_BLOCK: u8 = 4;
27
28/// Maximum encoded header-sync message bytes.
29pub const MAX_HS_MESSAGE_BYTES: usize = 2 * 1024 * 1024;
30/// Default number of headers advertised per response.
31pub const DEFAULT_HS_RANGE: u32 = 1000;
32/// Maximum number of headers ever honored by header sync.
33pub const MAX_HS_RANGE: u32 = 4000;
34/// Default number of in-flight header requests advertised per peer.
35pub const DEFAULT_HS_MAX_INFLIGHT: u16 = 10;
36
37pub(super) const HEADER_SYNC_MESSAGE_TYPE_BYTES: usize = 1;
38pub(super) const HEADER_SYNC_REQUEST_ID_BYTES: usize = 8;
39pub(super) const HEADER_SYNC_COUNT_BYTES: usize = 4;
40pub(super) const HEADER_SYNC_HAS_ROOTS_BYTES: usize = 1;
41pub(super) const HEADER_SYNC_BODY_SIZE_BYTES: usize = 4;
42/// Encoded [`BlockCommitmentRoots`]: height + Sapling root + Orchard root + Ironwood root
43/// + three `u64` shielded tx-counts (Sapling/Orchard/Ironwood) + auth-data root.
44pub(super) const HEADER_SYNC_BLOCK_COMMITMENT_ROOTS_BYTES: usize =
45    4 + 32 + 32 + 32 + 8 + 8 + 8 + 32;
46pub(super) const COMMON_HEADER_BYTES: usize = 1_487;
47pub(super) const REGTEST_HEADER_BYTES: usize = 177;
48pub(super) const LOCAL_MAX_HS_INFLIGHT_PER_PEER: u16 = 16;
49pub(super) const HEADER_SYNC_MAX_RESIDENT_BATCHES: u32 = 16;
50pub(super) const HEADER_SYNC_RETRY_AVOIDANCE: Duration = Duration::from_millis(50);
51pub(super) const STATUS_PUBLICATION_RETRY_DELAY: Duration = Duration::from_millis(50);
52pub(super) const HEADER_SYNC_SEEN_HASH_CAPACITY: usize = 4096;
53pub(super) const DEFAULT_HS_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
54pub(super) const EMPTY_HEADERS_RETRY_DELAY: Duration = Duration::from_secs(1);
55pub(super) const DEFAULT_HS_STATUS_REFRESH_INTERVAL: Duration = Duration::from_secs(30);
56// v1 semantic meters intentionally use strict spacing for unsolicited status refreshes and
57// distinct unseen full-block floods. Cheap duplicate `NewBlock` echoes are deduped before this
58// meter, so they do not consume tokens or cause honest peers to be scored.
59pub(super) const DEFAULT_HS_INBOUND_STATUS_MIN_INTERVAL: Duration = Duration::from_secs(5);
60pub(super) const DEFAULT_HS_INBOUND_NEW_BLOCK_MIN_INTERVAL: Duration = Duration::from_secs(5);
61
62const _: () = assert!(MAX_HS_MESSAGE_BYTES < LOCAL_MAX_MESSAGE_BYTES as usize);
63const _: () = assert!(
64    HEADER_SYNC_MESSAGE_TYPE_BYTES
65        + HEADER_SYNC_REQUEST_ID_BYTES
66        + HEADER_SYNC_COUNT_BYTES
67        + (COMMON_HEADER_BYTES
68            + HEADER_SYNC_BODY_SIZE_BYTES
69            + HEADER_SYNC_BLOCK_COMMITMENT_ROOTS_BYTES)
70            * (DEFAULT_HS_RANGE as usize)
71        < MAX_HS_MESSAGE_BYTES
72);
73
74/// Native versioned header-sync message.
75#[derive(Clone, Debug, Eq, PartialEq)]
76pub enum HeaderSyncMessage {
77    /// Peer tip, anchor, and served-range advertisement.
78    Status(HeaderSyncStatus),
79    /// Request `count` headers starting at `start_height`.
80    GetHeaders {
81        /// First requested height.
82        start_height: block::Height,
83        /// Requested header count.
84        count: u32,
85        /// Whether the requester wants all-or-nothing tree-aux roots.
86        /// A sender who is syncing in vct mode will always request these.
87        /// A sender who is syncing in non-checkpoint mode does not need these but still requests them.
88        /// A sender who is syncing above the last checkpoint height does not request these.
89        want_tree_aux_roots: bool,
90    },
91    /// A bounded contiguous header run with one advisory body-size hint per header.
92    ///
93    /// A `0` size means "unknown"; the hint is not consensus data. Tree-aux roots
94    /// are peer-sourced execution hints and are verified by state before use.
95    Headers {
96        /// Headers in ascending height order.
97        headers: Vec<Arc<block::Header>>,
98        /// Advisory serialized body sizes, parallel to `headers`.
99        body_sizes: Vec<u32>,
100        /// Per-height commitment roots, parallel to `headers`.
101        tree_aux_roots: Vec<BlockCommitmentRoots>,
102    },
103    /// Full block tip-flood payload.
104    NewBlock(Arc<block::Block>),
105}
106
107impl HeaderSyncMessage {
108    /// Returns this message's header-sync discriminator.
109    pub fn message_type(&self) -> u8 {
110        match self {
111            Self::Status(_) => MSG_HS_STATUS,
112            Self::GetHeaders { .. } => MSG_HS_GET_HEADERS,
113            Self::Headers { .. } => MSG_HS_HEADERS,
114            Self::NewBlock(_) => MSG_HS_NEW_BLOCK,
115        }
116    }
117
118    /// Encode this message as `[u8 message_type][request id][bounded fields...]`.
119    ///
120    /// `request_id` is required by `GetHeaders`/`Headers` and unused by the other
121    /// message types, which are not correlated.
122    pub fn encode(
123        &self,
124        request_id: Option<HeaderSyncRequestId>,
125    ) -> Result<Vec<u8>, HeaderSyncWireError> {
126        let mut bytes = Vec::new();
127        bytes.write_u8(self.message_type())?;
128        match self {
129            Self::Status(status) => status.encode_to(&mut bytes)?,
130            Self::GetHeaders {
131                start_height,
132                count,
133                want_tree_aux_roots,
134            } => {
135                validate_get_headers_count(*count)?;
136                bytes.write_u64::<LittleEndian>(
137                    request_id
138                        .ok_or(HeaderSyncWireError::MissingRequestId {
139                            message: "GetHeaders",
140                        })?
141                        .get(),
142                )?;
143                write_height(&mut bytes, *start_height)?;
144                bytes.write_u32::<LittleEndian>(*count)?;
145                bytes.write_u8(u8::from(*want_tree_aux_roots))?;
146            }
147            Self::Headers {
148                headers,
149                body_sizes,
150                tree_aux_roots,
151            } => {
152                validate_headers_len(headers.len(), usize_from_u32(MAX_HS_RANGE, "headers cap")?)?;
153                validate_body_sizes_len(headers.len(), body_sizes.len())?;
154                validate_tree_aux_roots_len(headers.len(), tree_aux_roots.len())?;
155                bytes.write_u64::<LittleEndian>(
156                    request_id
157                        .ok_or(HeaderSyncWireError::MissingRequestId { message: "Headers" })?
158                        .get(),
159                )?;
160                bytes.write_u32::<LittleEndian>(u32_from_usize(headers.len(), "headers count")?)?;
161                bytes.write_u8(u8::from(!tree_aux_roots.is_empty()))?;
162                for (header, body_size) in headers.iter().zip(body_sizes) {
163                    header.zcash_serialize(&mut bytes)?;
164                    bytes.write_u32::<LittleEndian>(*body_size)?;
165                }
166                for roots in tree_aux_roots {
167                    roots.zcash_serialize(&mut bytes)?;
168                }
169            }
170            Self::NewBlock(block) => {
171                block.zcash_serialize(&mut bytes)?;
172            }
173        }
174        if bytes.len() > MAX_HS_MESSAGE_BYTES {
175            return Err(HeaderSyncWireError::OversizedPayload {
176                actual: bytes.len(),
177                max: MAX_HS_MESSAGE_BYTES,
178            });
179        }
180        Ok(bytes)
181    }
182
183    /// Decode a header-sync message and its request ID using the request bounds in `context`.
184    ///
185    /// The request ID is `None` for the message types that do not carry one.
186    pub fn decode(
187        bytes: &[u8],
188        context: HeaderSyncDecodeContext,
189    ) -> Result<(Self, Option<HeaderSyncRequestId>), HeaderSyncWireError> {
190        if bytes.len() > MAX_HS_MESSAGE_BYTES {
191            return Err(HeaderSyncWireError::OversizedPayload {
192                actual: bytes.len(),
193                max: MAX_HS_MESSAGE_BYTES,
194            });
195        }
196        let mut reader = Cursor::new(bytes);
197        let message_type = reader.read_u8()?;
198        let mut request_id = None;
199        let message = match message_type {
200            MSG_HS_STATUS => Self::Status(HeaderSyncStatus::decode_from(&mut reader)?),
201            MSG_HS_GET_HEADERS => {
202                request_id = HeaderSyncRequestId::new(reader.read_u64::<LittleEndian>()?);
203                if request_id.is_none() {
204                    return Err(HeaderSyncWireError::MissingRequestId {
205                        message: "GetHeaders",
206                    });
207                }
208                let start_height = read_height(&mut reader)?;
209                let count = reader.read_u32::<LittleEndian>()?;
210                let want_tree_aux_roots = read_bool_marker(&mut reader, "want_tree_aux_roots")?;
211                validate_get_headers_count(count)?;
212                Self::GetHeaders {
213                    start_height,
214                    count,
215                    want_tree_aux_roots,
216                }
217            }
218            MSG_HS_HEADERS => {
219                request_id = HeaderSyncRequestId::new(reader.read_u64::<LittleEndian>()?);
220                if request_id.is_none() {
221                    return Err(HeaderSyncWireError::MissingRequestId { message: "Headers" });
222                }
223                let count = usize_from_u32(reader.read_u32::<LittleEndian>()?, "headers count")?;
224                let has_roots = read_bool_marker(&mut reader, "has_roots")?;
225                let Some(max_headers) = context.headers_response_limit()? else {
226                    return Err(HeaderSyncWireError::UnsolicitedHeaders);
227                };
228                if has_roots && !context.wants_tree_aux_roots() {
229                    return Err(HeaderSyncWireError::UnrequestedTreeAuxRoots);
230                }
231                validate_headers_len(count, max_headers)?;
232                let mut headers = Vec::with_capacity(count);
233                let mut body_sizes = Vec::with_capacity(count);
234                let mut tree_aux_roots = if has_roots {
235                    Vec::with_capacity(count)
236                } else {
237                    Vec::new()
238                };
239                for _ in 0..count {
240                    headers.push(Arc::new(block::Header::zcash_deserialize(&mut reader)?));
241                    body_sizes.push(reader.read_u32::<LittleEndian>()?);
242                }
243                if has_roots {
244                    for _ in 0..count {
245                        tree_aux_roots.push(BlockCommitmentRoots::zcash_deserialize(&mut reader)?);
246                    }
247                }
248                validate_tree_aux_roots_len(count, tree_aux_roots.len())?;
249                if let Some(requested) = context.requested {
250                    validate_tree_aux_root_heights(requested.start_height, &tree_aux_roots)?;
251                }
252                Self::Headers {
253                    headers,
254                    body_sizes,
255                    tree_aux_roots,
256                }
257            }
258            MSG_HS_NEW_BLOCK => {
259                Self::NewBlock(Arc::new(block::Block::zcash_deserialize(&mut reader)?))
260            }
261            value => return Err(HeaderSyncWireError::UnknownMessageType(value)),
262        };
263        reject_trailing(bytes, &reader)?;
264        Ok((message, request_id))
265    }
266
267    /// Return a `Headers` request ID without decoding the full header payload.
268    pub fn peek_headers_request_id(
269        bytes: &[u8],
270    ) -> Result<HeaderSyncRequestId, HeaderSyncWireError> {
271        let mut reader = Cursor::new(bytes);
272        let message_type = reader.read_u8()?;
273        if message_type != MSG_HS_HEADERS {
274            return Err(HeaderSyncWireError::UnknownMessageType(message_type));
275        }
276        HeaderSyncRequestId::new(reader.read_u64::<LittleEndian>()?)
277            .ok_or(HeaderSyncWireError::MissingRequestId { message: "Headers" })
278    }
279
280    /// Convert this message into a bounded Zakura frame.
281    pub fn encode_frame(
282        &self,
283        request_id: Option<HeaderSyncRequestId>,
284    ) -> Result<Frame, HeaderSyncWireError> {
285        Ok(Frame {
286            message_type: u16::from(self.message_type()),
287            flags: 0,
288            payload: self.encode(request_id)?,
289        })
290    }
291
292    /// Decode this message and its request ID from a Zakura frame, after checking flags
293    /// and type agreement.
294    pub fn decode_frame(
295        frame: Frame,
296        context: HeaderSyncDecodeContext,
297    ) -> Result<(Self, Option<HeaderSyncRequestId>), HeaderSyncWireError> {
298        if frame.flags != 0 {
299            return Err(HeaderSyncWireError::UnsupportedFlags(frame.flags));
300        }
301        let (message, request_id) = Self::decode(&frame.payload, context)?;
302        let frame_message_type = u8::try_from(frame.message_type)
303            .map_err(|_| HeaderSyncWireError::UnknownFrameMessageType(frame.message_type))?;
304        if frame_message_type != message.message_type() {
305            return Err(HeaderSyncWireError::MismatchedFrameMessageType {
306                frame: frame.message_type,
307                payload: message.message_type(),
308            });
309        }
310        Ok((message, request_id))
311    }
312}
zakura_network/zakura/header_sync/wire.rs

zakura_network/zakura/header_sync/
wire.rs
