cumulus_primitives_spec_messaging/
channel.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//! Channel state primitives: the key and the two per-direction states the messaging
18//! pallet keeps in `OutChannels` / `InChannels` and returns from the runtime API's `out_channels()`
19//! / `in_channels()`.
20//!
21//! State shapes, not wire types. The channel protocol (the in-band [`crate::SpecMsgSignal`]s and
22//! the out-of-band [`Register`]) is in [`crate::flow_control`]; the logic is the pallet's.
23
24use codec::{Decode, DecodeWithMemTracking, Encode, MaxEncodedLen};
25use polkadot_parachain_primitives::primitives::Id as ParaId;
26use scale_info::TypeInfo;
27
28use crate::{flow_control::Register, mmr::MessagePosition};
29
30/// Channel discriminator.
31#[derive(
32	Clone,
33	Copy,
34	Encode,
35	Decode,
36	DecodeWithMemTracking,
37	MaxEncodedLen,
38	PartialEq,
39	Eq,
40	PartialOrd,
41	Ord,
42	Debug,
43	TypeInfo,
44)]
45pub struct ChannelId {
46	/// The other end: the recipient of an outbound channel, the sender of an inbound one.
47	pub peer: ParaId,
48	/// Allocation-convention field, `0` by default (see `StreamId`).
49	pub domain: u8,
50	pub num: u16,
51}
52
53/// A channel's phase, derived from [`OutChannelState`]. Not stored.
54#[derive(Clone, Copy, PartialEq, Eq, Debug)]
55pub enum ChannelPhase {
56	/// Sent `OpenChannel`, no register read yet.
57	Opening,
58	/// A register has been read and neither side has closed.
59	Open,
60	/// We sent `CloseChannel`, or the peer's register says `closed`.
61	Closed,
62}
63
64/// A register as read: the peer's latest `Register` and its position on the peer's `Ack` stream.
65/// Kept together so no register can be stored without its position.
66#[derive(
67	Clone,
68	Copy,
69	Encode,
70	Decode,
71	DecodeWithMemTracking,
72	MaxEncodedLen,
73	PartialEq,
74	Eq,
75	Debug,
76	TypeInfo,
77)]
78pub struct RegisterRead {
79	/// Position of `register` on the peer's `Ack` stream. A read below it is ignored, so an older
80	/// grant or `closed` flag cannot replace a newer one: `up_to` and `version` stay unchanged
81	/// across a suspend, a resume or a close, so they cannot order registers. Sound only because
82	/// the stream root binds the leaf count.
83	pub at: MessagePosition,
84	/// The peer's watermark, credit, version announcement and closed flag.
85	pub register: Register,
86}
87
88/// Sender side, per outbound channel.
89#[derive(
90	Clone,
91	Copy,
92	Encode,
93	Decode,
94	DecodeWithMemTracking,
95	MaxEncodedLen,
96	PartialEq,
97	Eq,
98	Debug,
99	TypeInfo,
100)]
101pub struct OutChannelState {
102	/// Whether we sent `CloseChannel`. The peer's close arrives in the register; this is the one
103	/// phase bit the register cannot carry.
104	pub closed_by_us: bool,
105	/// Our latest in-band version announcement.
106	pub announced_version: u8,
107	/// Latest register read, with its position. `None` until the first read.
108	pub register: Option<RegisterRead>,
109}
110
111impl OutChannelState {
112	/// The lower of our announced version and the peer's. `None` while `Opening`: no register has
113	/// been read, so the peer's is unknown.
114	pub fn effective_version(&self) -> Option<u8> {
115		self.register.map(|read| self.announced_version.min(read.register.version))
116	}
117
118	pub fn phase(&self) -> ChannelPhase {
119		if self.closed_by_us || self.register.is_some_and(|read| read.register.closed) {
120			ChannelPhase::Closed
121		} else if self.register.is_none() {
122			ChannelPhase::Opening
123		} else {
124			ChannelPhase::Open
125		}
126	}
127}
128
129/// Receiver side, per inbound channel.
130#[derive(
131	Clone,
132	Copy,
133	Encode,
134	Decode,
135	DecodeWithMemTracking,
136	MaxEncodedLen,
137	PartialEq,
138	Eq,
139	Debug,
140	TypeInfo,
141)]
142pub struct InChannelState {
143	/// The register we last published on our `Ack` stream. This is our channel state as the peer
144	/// sees it, and it decides when the next publish is due.
145	pub published: Register,
146	/// The sender's latest in-band version announcement (from consumed `OpenChannel` /
147	/// `Upgrade` signals).
148	pub peer_version: u8,
149	/// Upper-layer consumption switch. While set, the STF refuses this channel's messages,
150	/// `consumed_streams()` omits the stream, and published registers grant zero. A pause, not a
151	/// close: all state persists.
152	pub suspended: bool,
153}
154
155impl InChannelState {
156	pub fn effective_version(&self) -> u8 {
157		self.published.version.min(self.peer_version)
158	}
159}
160
161#[cfg(test)]
162mod tests {
163	use super::*;
164	use crate::flow_control::WindowGrant;
165
166	fn register(version: u8, closed: bool) -> Register {
167		Register { version, up_to: MessagePosition(0), grant: WindowGrant::default(), closed }
168	}
169
170	fn read(version: u8, closed: bool) -> RegisterRead {
171		RegisterRead { at: MessagePosition(0), register: register(version, closed) }
172	}
173
174	#[test]
175	fn out_channel_phase_is_a_view() {
176		let mut out = OutChannelState { closed_by_us: false, announced_version: 1, register: None };
177		assert_eq!(out.phase(), ChannelPhase::Opening);
178		out.register = Some(read(1, false));
179		assert_eq!(out.phase(), ChannelPhase::Open);
180		// Either side's close closes the channel. Ours counts before any register read.
181		out.register = Some(read(1, true));
182		assert_eq!(out.phase(), ChannelPhase::Closed);
183		let ours = OutChannelState { closed_by_us: true, announced_version: 1, register: None };
184		assert_eq!(ours.phase(), ChannelPhase::Closed);
185	}
186
187	#[test]
188	fn effective_version_is_the_min() {
189		let inbound =
190			InChannelState { published: register(3, false), peer_version: 2, suspended: false };
191		assert_eq!(inbound.effective_version(), 2);
192
193		let mut out = OutChannelState { closed_by_us: false, announced_version: 3, register: None };
194		assert_eq!(out.effective_version(), None, "unknown until the first register read");
195		out.register = Some(read(2, false));
196		assert_eq!(out.effective_version(), Some(2));
197	}
198
199	#[test]
200	fn channel_id_orders_by_peer_then_domain_then_num() {
201		let a = ChannelId { peer: ParaId::from(1), domain: 1, num: 9 };
202		let b = ChannelId { peer: ParaId::from(2), domain: 0, num: 0 };
203		let c = ChannelId { peer: ParaId::from(2), domain: 0, num: 1 };
204		assert!(a < b && b < c);
205	}
206}
cumulus_primitives_spec_messaging/channel.rs

cumulus_primitives_spec_messaging/
channel.rs
