cumulus_primitives_spec_messaging/
inherent.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//! The node/runtime consumption interface: what the runtime asks the node to fetch
18//! ([`ConsumedStream`], via the `consumed_streams()` runtime API) and what the node delivers in the
19//! messaging inherent ([`MessagingInherentData`] / [`ConsumeItem`]).
20//!
21//! The inherent carries payloads and placement hints. No roots, no proofs. The runtime verifies by
22//! recomputation: hash payloads into leaves, append to a frontier, let the candidate's lift bind
23//! the endpoint. A wrong payload, `base` or peak set yields an endpoint no lift can bind.
24
25use alloc::vec::Vec;
26use codec::{Decode, DecodeWithMemTracking, Encode, MaxEncodedLen};
27use polkadot_core_primitives::Hash;
28use polkadot_parachain_primitives::primitives::Id as ParaId;
29use scale_info::TypeInfo;
30use sp_core::ConstU32;
31use sp_runtime::BoundedVec;
32
33use crate::{
34	mmr::{MessagePosition, MAX_MMR_HASHES},
35	stream::StreamId,
36};
37
38/// Key of the messaging inherent in `InherentData`.
39pub const INHERENT_IDENTIFIER: sp_inherents::InherentIdentifier = *b"specmsg0";
40
41pub type Payload = Vec<u8>;
42
43/// One stream the runtime wants fetched, per source, as returned by `consumed_streams()`. `from` is
44/// the fetch cursor: positions `>= from` are wanted. Suspended channels are omitted, which is how
45/// collators learn to stop fetching.
46#[derive(
47	Clone,
48	Copy,
49	Encode,
50	Decode,
51	DecodeWithMemTracking,
52	MaxEncodedLen,
53	PartialEq,
54	Eq,
55	Debug,
56	TypeInfo,
57)]
58pub enum ConsumedStream {
59	/// The source's `Channel`. Ordered prefix consumption; `from` is the tracked frontier's leaf
60	/// count.
61	#[codec(index = 0)]
62	Channel { domain: u8, num: u16, from: MessagePosition },
63	/// The source's `Broadcast`. Lossy, latest-wins; `from` is the highwater plus one.
64	#[codec(index = 1)]
65	Broadcast { domain: u16, subdomain: u8, num: u32, from: MessagePosition },
66}
67
68impl ConsumedStream {
69	/// The fetch-cursor view of `stream`. `None` for kinds the runtime never asks to fetch: `Ack`
70	/// registers are read out-of-band by the sender, `Private` is chain-defined.
71	pub fn project(stream: &StreamId, from: MessagePosition) -> Option<Self> {
72		match *stream {
73			StreamId::Channel { domain, num, .. } => Some(Self::Channel { domain, num, from }),
74			StreamId::Broadcast { domain, subdomain, num } => {
75				Some(Self::Broadcast { domain, subdomain, num, from })
76			},
77			StreamId::Ack { .. } | StreamId::Private { .. } => None,
78		}
79	}
80
81	/// The `StreamId` this view names on the source's side. A channel's `recipient` is us.
82	pub fn stream_id(&self, recipient: ParaId) -> StreamId {
83		match *self {
84			Self::Channel { domain, num, .. } => StreamId::Channel { recipient, domain, num },
85			Self::Broadcast { domain, subdomain, num, .. } => {
86				StreamId::Broadcast { domain, subdomain, num }
87			},
88		}
89	}
90
91	pub fn from(&self) -> MessagePosition {
92		match *self {
93			Self::Channel { from, .. } | Self::Broadcast { from, .. } => from,
94		}
95	}
96}
97
98/// The messaging inherent: this block's consumption, per `(source, stream)`. A block that consumed
99/// nothing carries none.
100#[derive(Clone, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, TypeInfo, Default)]
101pub struct MessagingInherentData {
102	/// One item per stream consumed this block.
103	pub items: Vec<(ParaId, StreamId, ConsumeItem)>,
104}
105
106impl MessagingInherentData {
107	/// Whether this block consumed nothing. Then no inherent is placed.
108	pub fn is_empty(&self) -> bool {
109		self.items.is_empty()
110	}
111}
112
113/// Node side. Places the data under [`INHERENT_IDENTIFIER`], or nothing when empty, so a block that
114/// fetched nothing carries no inherent call.
115#[cfg(feature = "std")]
116#[async_trait::async_trait]
117impl sp_inherents::InherentDataProvider for MessagingInherentData {
118	async fn provide_inherent_data(
119		&self,
120		inherent_data: &mut sp_inherents::InherentData,
121	) -> Result<(), sp_inherents::Error> {
122		if self.is_empty() {
123			return Ok(());
124		}
125		inherent_data.put_data(INHERENT_IDENTIFIER, self)
126	}
127
128	async fn try_handle_error(
129		&self,
130		_: &sp_inherents::InherentIdentifier,
131		_: &[u8],
132	) -> Option<Result<(), sp_inherents::Error>> {
133		// An import-time error on this inherent invalidates the block.
134		None
135	}
136}
137
138/// One stream's consumption this block. Both arms verify the same way (hash payloads into leaves,
139/// append to a frontier, let the lift bind the endpoint) and differ only in where the frontier
140/// comes from.
141#[derive(Clone, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, TypeInfo)]
142pub enum ConsumeItem {
143	/// Prefix discipline: the frontier is the stored inbound frontier; `payloads` are the stream's
144	/// next messages, in order. Any deviation in order or count yields an endpoint no lift can
145	/// bind.
146	#[codec(index = 0)]
147	Channel { payloads: Vec<Payload> },
148	/// Inclusion discipline (registers, event streams): the item carries the frontier, since a
149	/// lossy consumer keeps none. `start_peaks` are the peaks at `base`. At least one payload;
150	/// a register or head read has exactly one.
151	///
152	/// `base` and `start_peaks` are unchecked hints: a lie in either yields a root no lift can
153	/// bind, because the stream root commits the leaf count.
154	///
155	/// Replay: a register read below the stored `RegisterRead::at` is ignored; an event read must
156	/// be above the stream's highwater, which then moves to `base + len - 1`.
157	#[codec(index = 1)]
158	Events {
159		base: MessagePosition,
160		start_peaks: BoundedVec<Hash, ConstU32<MAX_MMR_HASHES>>,
161		payloads: Vec<Payload>,
162	},
163}
164
165#[cfg(test)]
166mod tests {
167	use super::*;
168
169	/// The inherent is in the block, so its variant indices are consensus-relevant.
170	#[test]
171	fn variant_indices_are_frozen() {
172		let ch = ConsumedStream::Channel { domain: 0, num: 0, from: MessagePosition(0) };
173		let bc =
174			ConsumedStream::Broadcast { domain: 0, subdomain: 0, num: 0, from: MessagePosition(0) };
175		assert_eq!(ch.encode()[0], 0);
176		assert_eq!(bc.encode()[0], 1);
177
178		let item_ch = ConsumeItem::Channel { payloads: Vec::new() };
179		let item_ev = ConsumeItem::Events {
180			base: MessagePosition(0),
181			start_peaks: BoundedVec::new(),
182			payloads: Vec::new(),
183		};
184		assert_eq!(item_ch.encode()[0], 0);
185		assert_eq!(item_ev.encode()[0], 1);
186	}
187
188	#[test]
189	fn consumed_stream_projects_and_names_its_stream() {
190		let us = ParaId::from(2000);
191		let from = MessagePosition(7);
192		let channel = StreamId::Channel { recipient: us, domain: 1, num: 2 };
193		let broadcast = StreamId::Broadcast { domain: 3, subdomain: 4, num: 5 };
194		for stream in [channel, broadcast] {
195			let view = ConsumedStream::project(&stream, from).expect("fetchable kinds project");
196			assert_eq!(view.stream_id(us), stream, "round-trips through the view");
197			assert_eq!(view.from(), from);
198		}
199		// Never fetched: registers are read out-of-band, private kinds are chain-defined.
200		let ack = StreamId::Ack { recipient: us, domain: 0, num: 0 };
201		assert_eq!(ConsumedStream::project(&ack, from), None);
202		let private =
203			StreamId::Private { kind: crate::PrivateKind::new(0x80).unwrap(), body: [0; 7] };
204		assert_eq!(ConsumedStream::project(&private, from), None);
205	}
206
207	#[test]
208	fn empty_data_places_no_inherent_and_non_empty_round_trips() {
209		use sp_inherents::InherentDataProvider;
210		let mut inherent_data = sp_inherents::InherentData::new();
211		futures::executor::block_on(
212			MessagingInherentData::default().provide_inherent_data(&mut inherent_data),
213		)
214		.expect("providing empty data succeeds");
215		assert_eq!(
216			inherent_data.get_data::<MessagingInherentData>(&INHERENT_IDENTIFIER).unwrap(),
217			None,
218			"absent = consumed nothing"
219		);
220
221		let data = MessagingInherentData {
222			items: alloc::vec![(
223				ParaId::from(1000),
224				StreamId::Channel { recipient: ParaId::from(2000), domain: 0, num: 1 },
225				ConsumeItem::Channel { payloads: alloc::vec![alloc::vec![1, 2, 3]] },
226			)],
227		};
228		let mut inherent_data = sp_inherents::InherentData::new();
229		futures::executor::block_on(data.provide_inherent_data(&mut inherent_data))
230			.expect("providing data succeeds");
231		assert_eq!(
232			inherent_data.get_data::<MessagingInherentData>(&INHERENT_IDENTIFIER).unwrap(),
233			Some(data)
234		);
235	}
236
237	#[test]
238	fn inherent_round_trips() {
239		let data = MessagingInherentData {
240			items: alloc::vec![(
241				ParaId::from(1000),
242				StreamId::Channel { recipient: ParaId::from(2000), domain: 0, num: 1 },
243				ConsumeItem::Channel { payloads: alloc::vec![alloc::vec![1, 2, 3]] },
244			)],
245		};
246		assert_eq!(MessagingInherentData::decode(&mut &data.encode()[..]).unwrap(), data);
247	}
248}
cumulus_primitives_spec_messaging/inherent.rs

cumulus_primitives_spec_messaging/
inherent.rs
