cumulus_primitives_spec_messaging/
stream.rs

1// Copyright (C) Parity Technologies (UK) Ltd.
2// This file is part of Cumulus.
3// SPDX-License-Identifier: Apache-2.0
4
5// Licensed under the Apache License, Version 2.0 (the "License");
6// you may not use this file except in compliance with the License.
7// You may obtain a copy of the License at
8//
9// 	http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing, software
12// distributed under the License is distributed on an "AS IS" BASIS,
13// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14// See the License for the specific language governing permissions and
15// limitations under the License.
16
17//! `StreamId`: the parachain-structured stream key.
18//!
19//! Each stream a chain maintains is one MMR keyed by a `StreamId`. The relay chain never sees an
20//! id; only the `StreamsRoot` reaches it. Ids live in tree paths, proofs and networking. The full
21//! stream key is `(sender ParaId, StreamId)`, with the sender implicit in its own candidate.
22//!
23//! # Tree key (consensus-critical, frozen)
24//!
25//! [`StreamId::to_bytes`] is the commitment-tree key. It is big-endian, so `Ord` equals the key's
26//! byte order: kinds cluster and sequential ids are neighbours.
27
28use codec::{Decode, DecodeWithMemTracking, Encode, Error as CodecError, Input, MaxEncodedLen};
29use polkadot_parachain_primitives::primitives::Id as ParaId;
30use scale_info::TypeInfo;
31
32pub const STREAM_ID_LEN: usize = 8;
33
34/// A private-use kind byte.
35///
36/// Byte 0 of the tree key is the discriminant for standard kinds and the payload for `Private`, so
37/// a kind below [`Self::MIN`] would collide with a standard kind. [`Self::new`] is the only check.
38#[derive(
39	Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Encode, MaxEncodedLen, TypeInfo,
40)]
41pub struct PrivateKind(u8);
42
43impl PrivateKind {
44	/// Lowest private-use kind byte; everything below is standard or reserved.
45	pub const MIN: u8 = 0x80;
46
47	pub const fn new(kind: u8) -> Option<Self> {
48		if kind >= Self::MIN {
49			Some(Self(kind))
50		} else {
51			None
52		}
53	}
54
55	pub const fn get(self) -> u8 {
56		self.0
57	}
58}
59
60impl Decode for PrivateKind {
61	fn decode<I: Input>(input: &mut I) -> Result<Self, CodecError> {
62		Self::new(u8::decode(input)?).ok_or_else(|| "PrivateKind: below MIN".into())
63	}
64}
65
66impl DecodeWithMemTracking for PrivateKind {}
67
68/// A parachain-structured stream key.
69///
70/// `recipient` is the party that reads the stream: a channel's messages are for the recipient, an
71/// ack register is for the chain it grants credit to, broadcasts have no addressee.
72///
73/// SCALE variant indices follow the tree key's kind bytes.
74#[derive(
75	Clone,
76	Copy,
77	PartialEq,
78	Eq,
79	PartialOrd,
80	Ord,
81	Hash,
82	Debug,
83	Encode,
84	Decode,
85	DecodeWithMemTracking,
86	MaxEncodedLen,
87	TypeInfo,
88)]
89pub enum StreamId {
90	/// Ordered, flow-controlled, guaranteed, unidirectional channel to `recipient`.
91	#[codec(index = 0)]
92	Channel { recipient: ParaId, domain: u8, num: u16 },
93	/// The receiver's lossy confirmation register for a channel, addressed to `recipient` (the
94	/// sender).
95	#[codec(index = 1)]
96	Ack { recipient: ParaId, domain: u8, num: u16 },
97	/// Sender-wide event stream (pub-sub), no addressee.
98	#[codec(index = 2)]
99	Broadcast { domain: u16, subdomain: u8, num: u32 },
100	/// Private-use kind; 7 chain-defined body bytes.
101	#[codec(index = 0x80)]
102	Private { kind: PrivateKind, body: [u8; 7] },
103}
104
105impl StreamId {
106	pub fn recipient(&self) -> Option<ParaId> {
107		match self {
108			StreamId::Channel { recipient, .. } | StreamId::Ack { recipient, .. } => {
109				Some(*recipient)
110			},
111			StreamId::Broadcast { .. } | StreamId::Private { .. } => None,
112		}
113	}
114
115	/// The commitment-tree key.
116	pub fn to_bytes(&self) -> [u8; STREAM_ID_LEN] {
117		let mut out = [0u8; STREAM_ID_LEN];
118		match self {
119			StreamId::Channel { recipient, domain, num } => {
120				out[0] = 0x00;
121				out[1..5].copy_from_slice(&u32::from(*recipient).to_be_bytes());
122				out[5] = *domain;
123				out[6..8].copy_from_slice(&num.to_be_bytes());
124			},
125			StreamId::Ack { recipient, domain, num } => {
126				out[0] = 0x01;
127				out[1..5].copy_from_slice(&u32::from(*recipient).to_be_bytes());
128				out[5] = *domain;
129				out[6..8].copy_from_slice(&num.to_be_bytes());
130			},
131			StreamId::Broadcast { domain, subdomain, num } => {
132				out[0] = 0x02;
133				out[1..3].copy_from_slice(&domain.to_be_bytes());
134				out[3] = *subdomain;
135				out[4..8].copy_from_slice(&num.to_be_bytes());
136			},
137			StreamId::Private { kind, body } => {
138				out[0] = kind.get();
139				out[1..8].copy_from_slice(body);
140			},
141		}
142		out
143	}
144
145	/// The inverse of [`Self::to_bytes`]. Returns `Err` if the kind is reserved.
146	pub fn from_bytes(bytes: [u8; STREAM_ID_LEN]) -> Result<Self, CodecError> {
147		let [kind, rest @ ..] = bytes;
148		let recipient = || ParaId::from(u32::from_be_bytes([rest[0], rest[1], rest[2], rest[3]]));
149		match kind {
150			0x00 => Ok(StreamId::Channel {
151				recipient: recipient(),
152				domain: rest[4],
153				num: u16::from_be_bytes([rest[5], rest[6]]),
154			}),
155			0x01 => Ok(StreamId::Ack {
156				recipient: recipient(),
157				domain: rest[4],
158				num: u16::from_be_bytes([rest[5], rest[6]]),
159			}),
160			0x02 => Ok(StreamId::Broadcast {
161				domain: u16::from_be_bytes([rest[0], rest[1]]),
162				subdomain: rest[2],
163				num: u32::from_be_bytes([rest[3], rest[4], rest[5], rest[6]]),
164			}),
165			0x80..=0xFF => {
166				let kind = PrivateKind::new(kind).ok_or("StreamId: private kind out of range")?;
167				Ok(StreamId::Private { kind, body: rest })
168			},
169			_ => Err(CodecError::from("StreamId: reserved/unknown kind")),
170		}
171	}
172}
173
174#[cfg(test)]
175mod tests {
176	use super::*;
177
178	fn sample_ids() -> Vec<StreamId> {
179		vec![
180			StreamId::Channel { recipient: ParaId::from(1000), domain: 0, num: 0 },
181			StreamId::Channel { recipient: ParaId::from(1000), domain: 0, num: 1 },
182			StreamId::Channel { recipient: ParaId::from(1000), domain: 1, num: 0 },
183			StreamId::Channel { recipient: ParaId::from(2000), domain: 0, num: 0 },
184			StreamId::Ack { recipient: ParaId::from(1000), domain: 0, num: 0 },
185			StreamId::Broadcast { domain: 0, subdomain: 0, num: 0 },
186			StreamId::Broadcast { domain: 1, subdomain: 0, num: 0 },
187			StreamId::Private { kind: PrivateKind::new(0x80).unwrap(), body: [0; 7] },
188			StreamId::Private { kind: PrivateKind::new(0xFF).unwrap(), body: [7; 7] },
189		]
190	}
191
192	#[test]
193	fn private_kind_rejects_standard_and_reserved_bytes() {
194		for byte in 0x00..PrivateKind::MIN {
195			assert!(PrivateKind::new(byte).is_none(), "0x{byte:02X} must not be a private kind");
196		}
197		for byte in PrivateKind::MIN..=0xFF {
198			assert_eq!(PrivateKind::new(byte).map(PrivateKind::get), Some(byte));
199		}
200	}
201
202	#[test]
203	fn tree_key_layout_is_exact() {
204		assert_eq!(
205			StreamId::Channel { recipient: ParaId::from(2000), domain: 0, num: 0 }.to_bytes(),
206			[0x00, 0x00, 0x00, 0x07, 0xD0, 0x00, 0x00, 0x00],
207		);
208		assert_eq!(
209			StreamId::Ack { recipient: ParaId::from(1000), domain: 0, num: 5 }.to_bytes(),
210			[0x01, 0x00, 0x00, 0x03, 0xE8, 0x00, 0x00, 0x05],
211		);
212		assert_eq!(
213			StreamId::Broadcast { domain: 0, subdomain: 0, num: 1 }.to_bytes(),
214			[0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01],
215		);
216		assert_eq!(
217			StreamId::Private {
218				kind: PrivateKind::new(0x80).unwrap(),
219				body: [1, 2, 3, 4, 5, 6, 7]
220			}
221			.to_bytes(),
222			[0x80, 1, 2, 3, 4, 5, 6, 7],
223		);
224	}
225
226	#[test]
227	fn from_bytes_inverts_to_bytes() {
228		for id in sample_ids() {
229			assert_eq!(StreamId::from_bytes(id.to_bytes()).unwrap(), id);
230		}
231	}
232
233	#[test]
234	fn from_bytes_rejects_reserved_kinds() {
235		for kind in 0x03u8..=0x7F {
236			assert!(
237				StreamId::from_bytes([kind, 0, 0, 0, 0, 0, 0, 0]).is_err(),
238				"reserved kind {kind:#x} must reject"
239			);
240		}
241	}
242
243	#[test]
244	fn decode_round_trips() {
245		for id in sample_ids() {
246			let encoded = id.encode();
247			assert!(encoded.len() <= StreamId::max_encoded_len());
248			assert_eq!(StreamId::decode(&mut &encoded[..]).unwrap(), id);
249		}
250	}
251
252	#[test]
253	fn decode_rejects_private_kind_below_min() {
254		let mut encoded =
255			StreamId::Private { kind: PrivateKind::new(0x80).unwrap(), body: [0; 7] }.encode();
256		encoded[1] = PrivateKind::MIN - 1;
257		assert!(StreamId::decode(&mut &encoded[..]).is_err());
258	}
259
260	#[test]
261	fn decode_rejects_truncated() {
262		let full = StreamId::Channel { recipient: ParaId::from(7), domain: 0, num: 0 }.encode();
263		for len in 0..full.len() {
264			assert!(StreamId::decode(&mut &full[..len]).is_err(), "len {len} must reject");
265		}
266	}
267
268	#[test]
269	fn ord_equals_tree_key_order() {
270		let mut by_ord = sample_ids();
271		by_ord.sort();
272		let mut by_key = sample_ids();
273		by_key.sort_by_key(StreamId::to_bytes);
274		assert_eq!(by_ord, by_key, "Ord must match the tree key's lexicographic order");
275	}
276
277	#[test]
278	fn recipient_addressing() {
279		assert_eq!(
280			StreamId::Channel { recipient: ParaId::from(9), domain: 0, num: 0 }.recipient(),
281			Some(ParaId::from(9)),
282		);
283		assert_eq!(
284			StreamId::Ack { recipient: ParaId::from(9), domain: 0, num: 0 }.recipient(),
285			Some(ParaId::from(9)),
286		);
287		assert_eq!(StreamId::Broadcast { domain: 0, subdomain: 0, num: 0 }.recipient(), None);
288		assert_eq!(
289			StreamId::Private { kind: PrivateKind::new(0x80).unwrap(), body: [0; 7] }.recipient(),
290			None
291		);
292	}
293}
cumulus_primitives_spec_messaging/stream.rs

cumulus_primitives_spec_messaging/
stream.rs
