diff --git a/Cargo.lock b/Cargo.lock index 5d83a14bec1e..71cd08294e5f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5233,6 +5233,7 @@ dependencies = [ name = "cumulus-primitives-core" version = "0.7.0" dependencies = [ + "cumulus-primitives-spec-messaging", "parity-scale-codec", "polkadot-core-primitives", "polkadot-parachain-primitives", @@ -5270,6 +5271,21 @@ dependencies = [ "sp-trie 29.0.0", ] +[[package]] +name = "cumulus-primitives-spec-messaging" +version = "0.1.0" +dependencies = [ + "hex-literal", + "parity-scale-codec", + "polkadot-ckb-merkle-mountain-range", + "polkadot-core-primitives", + "polkadot-parachain-primitives", + "polkadot-primitives", + "scale-info", + "sp-core 28.0.0", + "sp-runtime 31.0.1", +] + [[package]] name = "cumulus-primitives-storage-weight-reclaim" version = "1.0.0" diff --git a/Cargo.toml b/Cargo.toml index 10695fa00d3f..b82fabf9ed4c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -143,6 +143,7 @@ members = [ "cumulus/primitives/core", "cumulus/primitives/parachain-inherent", "cumulus/primitives/proof-size-hostfunction", + "cumulus/primitives/spec-messaging", "cumulus/primitives/storage-weight-reclaim", "cumulus/primitives/timestamp", "cumulus/primitives/utility", @@ -791,6 +792,7 @@ cumulus-primitives-aura = { path = "cumulus/primitives/aura", default-features = cumulus-primitives-core = { path = "cumulus/primitives/core", default-features = false } cumulus-primitives-parachain-inherent = { path = "cumulus/primitives/parachain-inherent", default-features = false } cumulus-primitives-proof-size-hostfunction = { path = "cumulus/primitives/proof-size-hostfunction", default-features = false } +cumulus-primitives-spec-messaging = { path = "cumulus/primitives/spec-messaging", default-features = false } cumulus-primitives-storage-weight-reclaim = { path = "cumulus/primitives/storage-weight-reclaim", default-features = false } cumulus-primitives-utility = { path = "cumulus/primitives/utility", default-features = false } cumulus-relay-chain-inprocess-interface = { path = "cumulus/client/relay-chain-inprocess-interface", default-features = false } diff --git a/cumulus/primitives/core/Cargo.toml b/cumulus/primitives/core/Cargo.toml index 592026b82412..bc5d5a7678c0 100644 --- a/cumulus/primitives/core/Cargo.toml +++ b/cumulus/primitives/core/Cargo.toml @@ -27,10 +27,14 @@ polkadot-parachain-primitives = { workspace = true } polkadot-primitives = { workspace = true } xcm = { workspace = true } +# Cumulus +cumulus-primitives-spec-messaging = { workspace = true } + [features] default = ["std"] std = [ "codec/std", + "cumulus-primitives-spec-messaging/std", "polkadot-core-primitives/std", "polkadot-parachain-primitives/std", "polkadot-primitives/std", diff --git a/cumulus/primitives/core/src/lib.rs b/cumulus/primitives/core/src/lib.rs index 8f4c52293a93..b6a54ec76a75 100644 --- a/cumulus/primitives/core/src/lib.rs +++ b/cumulus/primitives/core/src/lib.rs @@ -35,6 +35,15 @@ pub mod parachain_block_data; pub mod scheduling; pub use parachain_block_data::ParachainBlockData; +// Parachain-side speculative-messaging off-chain types, surfaced here as the +// parachain primitives hub (the relay-visible commitments stay in +// `polkadot-primitives::v9`). +pub use cumulus_primitives_spec_messaging::{ + build_requires, leaf_hash, verify_messages_response, ConsumptionRecord, Interval, + MMRExtensionProof, MaxSpeculativeMessageLen, MessagePosition, MessagesRequest, + MessagesResponse, MmrFrontier, MmrRoot, Register, RequiresLift, SpecHasher, SpecMsgKind, + SpecMsgSignal, StreamId, StreamProof, TreeInclusionProof, WindowGrant, +}; pub use polkadot_core_primitives::InboundDownwardMessage; pub use polkadot_parachain_primitives::primitives::{ DmpMessageHandler, Id as ParaId, IsSystem, UpwardMessage, ValidationParams, XcmpMessageFormat, @@ -737,4 +746,28 @@ sp_api::decl_runtime_apis! { /// The collator will include them in the relay chain proof that is passed alongside the parachain inherent into the runtime. fn keys_to_prove() -> RelayProofRequest; } + + /// API for the speculative-messaging outbox (sender side): serves the off-chain fetch protocol. + pub trait SpeculativeOutboxApi { + /// The sender's current `StreamsRoot` — the `Provides` commitment over all active streams + /// (`None` when no stream is active). + fn streams_root() -> Option; + + /// Serve a range of a stream's messages with the proofs binding them to the request's `under` + /// `StreamsRoot` (payloads + `extension` + `tree_proof`), authenticated by + /// [`verify_messages_response`]. A payload-free request (`max_bytes == 0`) yields lift + /// material. `None` when the stream/root cannot be served. + fn messages_response(request: MessagesRequest) -> Option; + } + + /// API for the speculative-messaging inbox (receiver side). + pub trait SpeculativeInboxApi { + /// This block's consumption record: per touched stream, the MMR interval it consumed. The + /// `validate_block` wrapper synthesizes the `Requires` set from it via POV-carried lifts (a + /// block never emits `Requires` directly); the node reads it for a block's dependencies. + /// + /// The receiver's per-stream frontier (where the collator resumes fetching) is ordinary + /// receiver state, read directly off the parent — not exposed here. + fn consumption_record() -> ConsumptionRecord; + } } diff --git a/cumulus/primitives/spec-messaging/Cargo.toml b/cumulus/primitives/spec-messaging/Cargo.toml new file mode 100644 index 000000000000..f6750d9f0366 --- /dev/null +++ b/cumulus/primitives/spec-messaging/Cargo.toml @@ -0,0 +1,39 @@ +[package] +name = "cumulus-primitives-spec-messaging" +version = "0.1.0" +authors.workspace = true +edition.workspace = true +license = "Apache-2.0" +description = "Speculative Messaging primitives" +homepage.workspace = true +repository.workspace = true + +[lints] +workspace = true + +[dependencies] +codec = { features = ["derive"], workspace = true } +mmr-lib = { workspace = true, default-features = false } +scale-info = { features = ["derive"], workspace = true } +sp-core = { workspace = true } +sp-runtime = { workspace = true } + +# Polkadot +polkadot-core-primitives = { workspace = true } +polkadot-parachain-primitives = { workspace = true } +polkadot-primitives = { workspace = true } + +[dev-dependencies] +hex-literal = { workspace = true } + +[features] +default = ["std"] +std = [ + "codec/std", + "mmr-lib/std", + "polkadot-core-primitives/std", + "polkadot-parachain-primitives/std", + "polkadot-primitives/std", + "sp-core/std", + "sp-runtime/std", +] diff --git a/cumulus/primitives/spec-messaging/src/flow_control.rs b/cumulus/primitives/spec-messaging/src/flow_control.rs new file mode 100644 index 000000000000..56e0e84ce685 --- /dev/null +++ b/cumulus/primitives/spec-messaging/src/flow_control.rs @@ -0,0 +1,206 @@ +// Copyright (C) Parity Technologies (UK) Ltd. +// This file is part of Cumulus. +// SPDX-License-Identifier: Apache-2.0 + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Channel flow-control primitives (spec-msg v0.5) — **type primitives only**. +//! +//! The wire/encoding types of the channel protocol: the sender's channel-stream leaf payload +//! ([`SpecMsgKind`] / [`SpecMsgSignal`]) and the receiver's out-of-band confirmation [`Register`] +//! (with its advisory [`WindowGrant`]). The messaging pallet's *logic* — windowing, version +//! enforcement, `OutChannels`/`InChannels` storage (keyed by a `ChannelId`), the accept extrinsic — +//! lives with the pallet at integration, not here. +//! +//! # Consensus surface +//! +//! - [`SpecMsgKind`] is the SCALE-encoded payload of every **channel-stream** MMR leaf: it fills +//! the `payload` in [`crate::message::leaf_hash`]'s `LEAF_TAG ++ LEAF_VERSION ++ payload` +//! framing. The framing and `LEAF_VERSION` are untouched — `SpecMsgKind` is the (otherwise +//! opaque) `payload` content, so no leaf-format version bump is needed to evolve it. +//! - [`Register`] is the SCALE-encoded payload of every **ack-stream** leaf — the receiver's whole +//! channel voice (lossy, latest-wins), committed via the receiver's `StreamsRoot` and read +//! out-of-band by the sender. +//! +//! # Frozen core +//! +//! Per the design a **frozen core** must parse at every protocol version: [`SpecMsgSignal`]'s +//! `OpenChannel` (variant index **0** — it must decode before any version is announced), +//! `CloseChannel`, `Upgrade`, and the [`Register`] format itself (the machinery the receiver's +//! announcements ride on). These encodings are consensus-critical; the frozen vectors in the tests +//! pin them. Protocol versioning is by monotonic per-side announcements (sender in-band via +//! `OpenChannel`/`Upgrade`, receiver via [`Register::version`]); the effective version is the min +//! of the two latest announcements. + +use alloc::vec::Vec; +use codec::{Decode, DecodeWithMemTracking, Encode, MaxEncodedLen}; +use scale_info::TypeInfo; + +use crate::message::MessagePosition; + +/// The payload of every channel-stream MMR leaf: protocol signalling or userspace data. +/// +/// SCALE-encoded into the leaf preimage's `payload` (see the module docs). The transport is +/// deliberately blind to `Data`'s meaning — the only distinction it acts on is `Signal` vs `Data` +/// (window accounting, pallet-internal consumption); demultiplexing among userspace protocols +/// (incl. the XCM envelope) is an upper-layer convention. +#[derive(Clone, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, TypeInfo)] +pub enum SpecMsgKind { + /// Channel lifecycle signalling. Emitted and consumed by the messaging pallet's own lifecycle + /// logic, never by applications; window-counted and ordered like any other leaf. + Signal(SpecMsgSignal), + /// Userspace payload bytes, delivered in order. The size bound is the sender STF's + /// message-size constant ([`crate::message::MAX_SPECULATIVE_MESSAGE_LEN`]), not a type bound + /// — matching the design's `Data(Vec)`. + Data(Vec), +} + +/// Sender-side channel lifecycle signal — part of the design's **frozen core**. +/// +/// `OpenChannel` MUST stay variant index **0**: it has to parse before any version announcement +/// exists. `CloseChannel` and `Upgrade` complete the frozen core. Only variants *beyond* the core +/// would ever be version-gated. +#[derive( + Clone, + Copy, + Encode, + Decode, + DecodeWithMemTracking, + MaxEncodedLen, + PartialEq, + Eq, + Debug, + TypeInfo, +)] +pub enum SpecMsgSignal { + /// Open the channel and announce the sender's initial protocol version. **Variant index 0** + /// (consensus-critical — must decode before any announcement exists). + OpenChannel { + /// The sender's initial protocol-version announcement. + version: u8, + }, + /// Sender-side half-close. + CloseChannel, + /// Raise the sender's version announcement mid-channel (announcements never decrease). + Upgrade { + /// The sender's new (higher) protocol-version announcement. + version: u8, + }, +} + +/// Advisory send-window credit a receiver grants a sender, beyond the confirmed watermark. +/// +/// **Advisory**, not enforced: registers are lossy and read with delay, and on an ordered stream +/// the receiver can't reject without stalling — so the hard bound is the sender STF's message-size +/// constant. A grant may shrink between publishes. +#[derive( + Clone, + Copy, + Encode, + Decode, + DecodeWithMemTracking, + MaxEncodedLen, + PartialEq, + Eq, + Debug, + Default, + TypeInfo, +)] +pub struct WindowGrant { + /// Max messages the sender may have in flight beyond the watermark. + pub max_messages: u32, + /// Max cumulative bytes in flight beyond the watermark. + pub max_bytes: u64, + /// Max size of a single message. + pub max_message_size: u32, +} + +/// The receiver's entire channel voice — part of the design's **frozen core**. +/// +/// Published as the ack-stream's (lossy, latest-wins) leaf payload and read out-of-band by the +/// sender via an inclusion proof. The first publish is the sender-visible channel *acceptance* +/// (produced by the receiver's accept extrinsic). +#[derive( + Clone, + Copy, + Encode, + Decode, + DecodeWithMemTracking, + MaxEncodedLen, + PartialEq, + Eq, + Debug, + Default, + TypeInfo, +)] +pub struct Register { + /// The receiver's monotonic protocol-version announcement. + pub version: u8, + /// Cumulative confirmation watermark: consumed up to this position (monotonic, idempotent — a + /// single `u64`, so losing or delaying confirmations is harmless, the latest supersedes). + pub up_to: MessagePosition, + /// Advisory send-window credit beyond `up_to`. + pub grant: WindowGrant, + /// Receiver-side close / rejection. + pub closed: bool, +} + +#[cfg(test)] +mod tests { + use super::*; + use codec::Decode; + + #[test] + fn signal_frozen_core_variant_indices() { + // FROZEN CORE: OpenChannel MUST be variant index 0 (it must parse before any version + // announcement exists). Any renumbering is a consensus break. + assert_eq!(SpecMsgSignal::OpenChannel { version: 7 }.encode(), vec![0x00, 0x07]); + assert_eq!(SpecMsgSignal::CloseChannel.encode(), vec![0x01]); + assert_eq!(SpecMsgSignal::Upgrade { version: 9 }.encode(), vec![0x02, 0x09]); + } + + #[test] + fn spec_msg_kind_variant_layout() { + // The transport distinguishes only Signal (variant 0) vs Data (variant 1). + assert_eq!( + SpecMsgKind::Signal(SpecMsgSignal::OpenChannel { version: 1 }).encode(), + vec![0x00, 0x00, 0x01], + ); + // Data: variant 1 ++ compact(len) ++ bytes. + assert_eq!(SpecMsgKind::Data(vec![0xAA]).encode(), vec![0x01, 0x04, 0xAA]); + } + + #[test] + fn register_frozen_layout() { + // FROZEN CORE: the Register wire format is read cross-chain out-of-band; pin its exact byte + // layout (little-endian SCALE ints). Any field reorder / width change is a consensus break. + let r = Register { + version: 2, + up_to: MessagePosition(5), + grant: WindowGrant { max_messages: 10, max_bytes: 1000, max_message_size: 100 }, + closed: true, + }; + assert_eq!( + r.encode(), + vec![ + 0x02, // version: u8 + 0x05, 0, 0, 0, 0, 0, 0, 0, // up_to: u64 LE = 5 + 0x0a, 0, 0, 0, // grant.max_messages: u32 LE = 10 + 0xe8, 0x03, 0, 0, 0, 0, 0, 0, // grant.max_bytes: u64 LE = 1000 + 0x64, 0, 0, 0, // grant.max_message_size: u32 LE = 100 + 0x01, // closed: bool = true + ], + ); + assert_eq!(Register::decode(&mut &r.encode()[..]).unwrap(), r); + } +} diff --git a/cumulus/primitives/spec-messaging/src/lib.rs b/cumulus/primitives/spec-messaging/src/lib.rs new file mode 100644 index 000000000000..61a6323025d1 --- /dev/null +++ b/cumulus/primitives/spec-messaging/src/lib.rs @@ -0,0 +1,108 @@ +// Copyright (C) Parity Technologies (UK) Ltd. +// This file is part of Cumulus. +// SPDX-License-Identifier: Apache-2.0 + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Primitives for the Speculative Messaging protocol. +//! +//! Speculative Messaging lets parachains exchange messages without waiting for +//! full relay-chain confirmation. Each sender parachain accumulates its outgoing +//! messages per stream into a Merkle Mountain Range (MMR) and commits every +//! stream's root into a single `StreamsRoot` — a keyed commitment tree +//! (`polkadot_primitives::v9::StreamsRoot` — relay-visible, so it lives in +//! `polkadot-primitives`). The relay chain then matches sender commitments +//! (`Provides`) against receiver expectations (`Requires`/`RequiresSet`), allowing +//! both sides to process messages speculatively and confirm them after the fact. +//! +//! This crate holds the **parachain-side** primitives that build those +//! commitments and the off-chain wire types. +//! +//! # Key types +//! +//! - [`mmr::SpecMerge`] — the domain-tagged `mmr_lib::Merge` that backs each per-stream MMR; +//! [`mmr::root_from_peaks`] derives a stream root from peaks-only state. Inclusion and ancestry +//! proofs come from `mmr_lib` itself. +//! - [`message`] — the off-chain fetch protocol (`MessagesRequest`/`MessagesResponse` + +//! [`message::verify_messages_response`]) and the protocol's concrete instantiation +//! (`SpecHasher`, [`message::leaf_hash`] for a payload → MMR leaf, the payload bound). +//! - [`lift`] — the requires-lift / consumption-record POV types (`RequiresLift`, +//! `ConsumptionRecord`, `MMRExtensionProof`, …) and the `build_requires` synthesizer. + +#![cfg_attr(not(feature = "std"), no_std)] + +extern crate alloc; + +pub mod flow_control; +pub mod lift; +pub mod message; +pub mod mmr; +pub mod stream; +pub mod streams_root; + +pub use flow_control::{Register, SpecMsgKind, SpecMsgSignal, WindowGrant}; +pub use lift::{ + build_requires, build_requires_entry, stitch, ConsumptionRecord, Interval, LiftError, + MMRExtensionProof, MmrFrontier, MmrRoot, RequiresLift, TreeInclusionProof, +}; +pub use message::{ + leaf_hash, verify_messages_response, MaxSpeculativeMessageLen, MessagePosition, + MessagesRequest, MessagesResponse, SpecHasher, MAX_SPECULATIVE_MESSAGE_LEN, +}; +pub use stream::{StreamId, STREAM_ID_LEN}; +pub use streams_root::{StreamProof, StreamsRoot}; + +// Domain Tags to ensure that the same message structure used in different +// contexts (e.g. leaf vs inner node) do not collide on the same hash. Tag values +// are part of the hash preimages, so they are kept stable (the now-removed +// `EMPTY_TAG = 0x1` slot is intentionally left unused rather than renumbering). + +/// Tag for a leaf node. +pub const LEAF_TAG: u8 = 0x2; + +/// Tag for an inner node. +pub const INNER_TAG: u8 = 0x3; + +/// Tag for a peak. +pub const PEAK_TAG: u8 = 0x4; + +/// Tag for a `StreamsRoot` commitment-tree leaf: `H(STREAMS_LEAF_TAG ++ key[8] ++ +/// stream_root[32])`. Distinct from [`LEAF_TAG`] so a trie leaf can never collide with an MMR +/// message leaf. +pub const STREAMS_LEAF_TAG: u8 = 0x5; + +/// Tag for a `StreamsRoot` commitment-tree inner (branch) node: +/// `H(STREAMS_INNER_TAG ++ split_bit ++ left[32] ++ right[32])`. Distinct from [`INNER_TAG`]. +pub const STREAMS_INNER_TAG: u8 = 0x6; + +// Leaf versioning to allow for future changes to the leaf structure without +// breaking compatibility with old messages. + +/// Leaf Version. +pub const LEAF_VERSION: u8 = 0x0; + +/// Consensus-digest engine id carrying the sender's **`StreamsRoot`** (spec-msg v0.5): the root of +/// a keyed commitment tree over `(StreamId, current stream root)` for every active stream, +/// committed each block so the head is self-sufficient. A foreign receiver reads this root from the +/// header (anchored by the relay's `para_heads`/`paras::Heads`) and proves the stream it consumes +/// against it, obtaining the sender's committed stream root without a sender-state proof (see the +/// `streams_root` module). +/// +/// This is the sole spec-msg header digest in v0.5 (it supersedes the pre-v0.5 flat-`provides` +/// digest); the name matches the design's `SPMS_ENGINE_ID`. +/// +/// A foreign node verifies against this digest **directly** — a pure function of header, response +/// and proofs, no chain state — so its format is **protocol-standard, not chain-internal**: the +/// engine-id value is consensus-visible across chains and must be finalized with the design before +/// mainnet. The current `*b"spmf"` is a placeholder. +pub const SPMS_ENGINE_ID: sp_runtime::ConsensusEngineId = *b"spmf"; diff --git a/cumulus/primitives/spec-messaging/src/lift.rs b/cumulus/primitives/spec-messaging/src/lift.rs new file mode 100644 index 000000000000..c8b2fc14e6ad --- /dev/null +++ b/cumulus/primitives/spec-messaging/src/lift.rs @@ -0,0 +1,688 @@ +// Copyright (C) Parity Technologies (UK) Ltd. +// This file is part of Cumulus. +// SPDX-License-Identifier: Apache-2.0 + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Requires-lift & consumption-record primitives (spec-msg v0.5). +//! +//! Blocks never emit `Requires` and never see a `StreamsRoot`. Processing the messaging inherent +//! writes a [`ConsumptionRecord`] — which streams the block touched, and the MMR [`Interval`] it +//! consumed on each. The `validate_block` wrapper then **synthesizes** the candidate's `Requires` +//! set from the records plus POV-carried **lifts** ([`RequiresLift`]): one per recorded stream, +//! binding that stream's consumption to a *current* committed `StreamsRoot`. The relay chain only +//! ever sees entries it can match. This unifies 0.3's in-block catch-up and late-block proofs into +//! one mechanism. +//! +//! The invariant [`build_requires`] enforces, per stream and candidate: +//! +//! > the recorded intervals form a **proven forward chain** — each block continues where the +//! > previous ended, or a POV proof shows the gap is a forward extension ([`stitch`]) — and one +//! > lift binds the chain's endpoint to a committed root, transitively binding everything consumed. +//! +//! Trust split: the **record** is authoritative (STF output — consumption can't be hidden); the +//! **lifts** are untrusted POV data, verified here against the record's key and a committed root. +//! Lifts are matched *positionally* to a source's `StreamId`-sorted record streams; a mispaired +//! lift cannot verify because the `tree_proof` walk binds the record's key (see +//! [`crate::streams_root::streams_root_from_proof`]). + +use alloc::{collections::BTreeMap, vec::Vec}; +use codec::{Decode, DecodeWithMemTracking, Encode}; +use mmr_lib::{helper::get_peaks, leaf_index_to_mmr_size, NodeMerkleProof}; +use polkadot_core_primitives::Hash; +use polkadot_parachain_primitives::primitives::Id as ParaId; +use polkadot_primitives::RequiresSet; +use scale_info::TypeInfo; + +use crate::{ + mmr::{root_from_peaks, SpecMerge}, + stream::StreamId, + streams_root::{streams_root_from_proof, StreamProof, StreamsRoot}, + SpecHasher, +}; + +/// The keyed `StreamsRoot`-tree inclusion proof a lift walks from a stream root up to the +/// `StreamsRoot` (the design's `TreeInclusionProof`; concretely the keyed Patricia +/// [`StreamProof`]). +pub type TreeInclusionProof = StreamProof; + +/// Upper bound on an [`MMRExtensionProof`]'s connecting nodes. A valid append-only ancestry proof +/// over an MMR whose leaf count fits in `u64` needs O(log n) nodes — never more than a small +/// multiple of the 64-bit height. `4 * 64` is a generous ceiling no valid proof reaches; rejecting +/// beyond it is pure defense-in-depth (PoV size already bounds the decoded length loosely). +const MAX_EXTENSION_CONNECTING_NODES: usize = 4 * 64; + +/// A bagged MMR root — a stream's committed root at some point in its history. Newtyped so it can +/// never be confused with a [`StreamsRoot`] (the commitment-tree root): "confusing roots must not +/// typecheck". +#[derive(Clone, Copy, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, TypeInfo)] +pub struct MmrRoot(pub Hash); + +/// A peaks-only MMR frontier — the O(log n) state a stream continues from (`mmr::Mmr::into_parts`). +/// `leaf_count` fixes the placement of an extension proof's connecting nodes relative to `peaks`. +#[derive(Clone, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, TypeInfo, Default)] +pub struct MmrFrontier { + /// MMR peaks, highest to lowest. + pub peaks: Vec, + /// Number of leaves these peaks summarize. + pub leaf_count: u64, +} + +impl MmrFrontier { + /// The bagged root of this frontier, or `None` for an empty frontier (which has no root). + pub fn root(&self) -> Option { + (!self.peaks.is_empty()).then(|| MmrRoot(root_from_peaks::(&self.peaks))) + } + + /// The `mmr_lib` node count (size) of this frontier's MMR. + fn mmr_size(&self) -> u64 { + if self.leaf_count == 0 { + 0 + } else { + leaf_index_to_mmr_size(self.leaf_count - 1) + } + } +} + +/// An append-only MMR extension proof: **O(log n) connecting nodes** that, with a frontier's own +/// peaks, reconstruct a newer root — the concrete `polkadot-ckb-merkle-mountain-range` +/// (`gen_ancestry_proof`) form of the design's `connecting_nodes`. Verification **computes and +/// returns** the new root (never declared alongside the proof), treating the frontier's peaks as +/// opaque fixed subtrees whose placement is fixed by its `leaf_count`. Payloads / appended leaves +/// are never needed. +#[derive(Clone, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, TypeInfo, Default)] +pub struct MMRExtensionProof { + /// `mmr_lib` node count of the extended MMR (with the frontier's `leaf_count`, this places the + /// connecting nodes). `0` for the identity extension. + pub new_mmr_size: u64, + /// The O(log n) connecting nodes `(position, hash)` — the extended MMR's peaks / witnesses not + /// derivable from the frontier's own peaks (from `gen_ancestry_proof`). Empty for the + /// identity. + pub connecting_nodes: Vec<(u64, Hash)>, +} + +impl MMRExtensionProof { + /// The identity extension: no connecting nodes, yielding a frontier's own root unchanged (the + /// common caught-up case where the endpoint already *is* the current stream root). + pub fn identity() -> Self { + Self { new_mmr_size: 0, connecting_nodes: Vec::new() } + } + + /// Whether this is an identity (empty) extension. + pub fn is_identity(&self) -> bool { + self.connecting_nodes.is_empty() + } + + /// Extend `from` and **return** the new root, computed from `from`'s peaks + the connecting + /// nodes; `None` if the proof is not well-formed for that placement. The identity extension + /// yields `from`'s own root. `from` is the verifier's own state — trusted, not re-checked. + pub fn verify(&self, from: &MmrFrontier) -> Option { + let from_root = from.root()?; + if self.is_identity() { + return Some(from_root); + } + // Defense-in-depth: bound an untrusted proof's connecting nodes before doing O(n) work in + // `calculate_root`. A valid ancestry proof is far below this ceiling. + if self.connecting_nodes.len() > MAX_EXTENSION_CONNECTING_NODES { + return None; + } + // Place `from`'s peaks at their canonical positions in the *previous* (pre-extension) MMR. + let prev_positions = get_peaks(from.mmr_size()); + if prev_positions.len() != from.peaks.len() { + return None; + } + let nodes: Vec<(u64, Hash)> = + prev_positions.into_iter().zip(from.peaks.iter().copied()).collect(); + // Reconstruct the extended root from those peaks + the connecting nodes (`calculate_root` + // merges them under `SpecMerge`, matching the sender's MMR). + let proof = NodeMerkleProof::>::new( + self.new_mmr_size, + self.connecting_nodes.clone(), + ); + proof.calculate_root(nodes).ok().map(MmrRoot) + } +} + +/// Per stream and block: consumption entered the block at `start` and left it at `end`. +/// +/// The candidate's lift binds only the LAST state to a committed root; the intervals stretch that +/// guarantee back over the whole bundle — each block must start where the previous ended, or prove +/// the jump moved forward ([`stitch`]). +/// +/// - **Channels**: `start` = the frontier root before the block's incoming messages, `end` = the +/// frontier after. `start == previous end` holds by construction, so the chain check is free. +/// - **Register / event reads**: `end` = the context the reads were verified against, `start` = its +/// root (nothing advances). Contexts *can* jump between blocks (a fresher read mid-bundle is the +/// point), so a fabricated context breaks the chain instead of hiding behind a later genuine one. +/// +/// `end` is a full frontier because the next gap check, or the lift, extends from it. +#[derive(Clone, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, TypeInfo)] +pub struct Interval { + /// Root the block's consumption on this stream started from. + pub start: MmrRoot, + /// Frontier the block's consumption on this stream ended at. + pub end: MmrFrontier, +} + +/// Written per block: the streams this block touched and the interval it consumed on each, grouped +/// by source. Per source the entries are `StreamId`-sorted and unique (the messaging inherent +/// carries at most one item per stream). This is the API view of a flat, host-append-only outbox +/// vec, grouped and sorted at read time. +#[derive(Clone, Encode, Decode, PartialEq, Eq, Debug, TypeInfo, Default)] +pub struct ConsumptionRecord { + /// `source -> [(stream, interval)]`; each source's streams `StreamId`-sorted and unique. + pub entries: BTreeMap>, +} + +/// One lift, carried in the POV (never in the block or commitments). Matched positionally to a +/// source's consumption-record streams (`StreamId`-sorted); a mispaired lift cannot verify because +/// `tree_proof` binds the record's key. +#[derive(Clone, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, TypeInfo)] +pub struct RequiresLift { + /// One extension per gap in the stream's interval chain, in gap order; empty for a caught-up + /// single-context stream. + pub advances: Vec, + /// Extends the chain's endpoint to the stream's current root; verification yields that root + /// (the identity extension when the endpoint already is the current root). + pub extension: MMRExtensionProof, + /// Keyed walk from the current stream root to the `StreamsRoot` the requires entry becomes. + pub tree_proof: TreeInclusionProof, +} + +/// Everything that can invalidate requires synthesis. All are deterministic functions of the +/// records and POV, so every validator reaches the same verdict. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum LiftError { + /// A stream's interval list was empty. + EmptyRecord, + /// A gap needed a forward-extension proof but `advances` was exhausted. + MissingAdvance, + /// `advances` carried more proofs than there were gaps. + StrayAdvance, + /// A gap's proof did not forward-extend the previous endpoint to the next start. + BrokenChain, + /// The endpoint extension did not verify against the stitched endpoint. + BadExtension, + /// The tree proof did not fold to a `StreamsRoot` (malformed / wrong key). + BadTreeProof, + /// A source's stream count and lift count disagreed. + LiftCountMismatch, + /// A source's streams lifted to more than one `StreamsRoot`. + DivergentRoots, + /// The recorded sources and the provided lifts' sources did not match exactly. + LiftSourceMismatch, + /// More sources than the `RequiresSet` bound allows. + TooManySources, +} + +/// Stitch one stream's intervals (bundle order) into its endpoint frontier. `advances` must supply +/// exactly one extension per gap — a `start` that is not the previous `end`'s bagged root — in gap +/// order. Gaps are proven *forward* (extension proofs only exist forward), so verified states can +/// never regress. +pub fn stitch( + intervals: &[Interval], + advances: &[MMRExtensionProof], +) -> Result { + let (first, rest) = intervals.split_first().ok_or(LiftError::EmptyRecord)?; + let mut gaps = advances.iter(); + let mut current = first.end.clone(); + for next in rest { + let current_root = current.root().ok_or(LiftError::BrokenChain)?; + if next.start != current_root { + // A gap must be a proven forward extension of where we ended. + let proof = gaps.next().ok_or(LiftError::MissingAdvance)?; + let extended = proof.verify(¤t).ok_or(LiftError::BrokenChain)?; + if extended != next.start { + return Err(LiftError::BrokenChain); + } + } + current = next.end.clone(); + } + if gaps.next().is_some() { + return Err(LiftError::StrayAdvance); + } + Ok(current) +} + +/// Build the single requires entry for one source: stitch each stream's intervals into an endpoint, +/// extend it to the stream's current root, walk the tree to the `StreamsRoot`, and require *all* of +/// the source's streams to lift to the **same** root (one entry per source). `streams` iterates in +/// `StreamId` order; `lifts` matches it positionally. +pub fn build_requires_entry( + source: ParaId, + streams: &[(StreamId, Vec)], + lifts: &[RequiresLift], +) -> Result<(ParaId, StreamsRoot), LiftError> { + if streams.len() != lifts.len() { + return Err(LiftError::LiftCountMismatch); + } + let mut entry: Option = None; + for ((stream, intervals), lift) in streams.iter().zip(lifts) { + let endpoint = stitch(intervals, &lift.advances)?; + // The endpoint is contained in the stream's current root (computed, not declared)... + let current = lift.extension.verify(&endpoint).ok_or(LiftError::BadExtension)?; + // ...and the keyed tree walk from it yields the StreamsRoot this stream lifts to. The walk + // binds the record's `stream` key, so a mispaired lift cannot verify. + let root = streams_root_from_proof(*stream, current.0, &lift.tree_proof) + .ok_or(LiftError::BadTreeProof)?; + match entry { + None => entry = Some(root), + Some(prev) if prev != root => return Err(LiftError::DivergentRoots), + Some(_) => {}, + } + } + entry.map(|root| (source, root)).ok_or(LiftError::EmptyRecord) +} + +/// Synthesize the candidate's [`RequiresSet`] from the per-block consumption records (bundle order) +/// and the POV lifts (grouped per source). Sources must match exactly — a recorded source without +/// lifts, or lifts for an unrecorded source, invalidates the candidate. +pub fn build_requires( + records: &[ConsumptionRecord], + lifts: &BTreeMap>, +) -> Result { + // Merge per (source, stream) intervals across the bundle, preserving block order. + let mut merged: BTreeMap>> = BTreeMap::new(); + for record in records { + for (source, streams) in &record.entries { + let by_stream = merged.entry(*source).or_default(); + for (stream, interval) in streams { + by_stream.entry(*stream).or_default().push(interval.clone()); + } + } + } + // Recorded sources and lift sources must be exactly equal (both iterate ParaId-sorted). + if !merged.keys().eq(lifts.keys()) { + return Err(LiftError::LiftSourceMismatch); + } + let entries = merged + .iter() + .zip(lifts.values()) + .map(|((source, by_stream), source_lifts)| { + let streams: Vec<(StreamId, Vec)> = + by_stream.iter().map(|(s, i)| (*s, i.clone())).collect(); + build_requires_entry(*source, &streams, source_lifts) + }) + .collect::, _>>()?; + RequiresSet::try_from_iter(entries).map_err(|_| LiftError::TooManySources) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + mmr::{Mmr, MmrAccumulator}, + streams_root::{gen_stream_proof, streams_root}, + }; + use mmr_lib::util::{MemMMR, MemStore}; + use polkadot_primitives::{v9::MAX_SOURCES_PER_BLOCK, MAX_POV_SIZE}; + use sp_core::H256; + + type H = SpecHasher; + + fn leaves(n: u64) -> Vec { + (0..n).map(|i| H256::repeat_byte(i as u8 + 1)).collect() + } + + /// Peaks-only frontier after the first `k` leaves. + fn frontier_at(all: &[Hash], k: usize) -> MmrFrontier { + let mut mmr = Mmr::::new(); + for l in &all[..k] { + mmr.append(*l); + } + let (peaks, leaf_count) = mmr.into_parts(); + MmrFrontier { peaks, leaf_count } + } + + /// Bagged root over the first `k` leaves. + fn root_at(all: &[Hash], k: usize) -> MmrRoot { + frontier_at(all, k).root().unwrap() + } + + /// `mmr_lib` node count for `k` leaves. + fn mmr_size(k: usize) -> u64 { + if k == 0 { + 0 + } else { + leaf_index_to_mmr_size((k - 1) as u64) + } + } + + /// O(log n) extension proof from `k` leaves to `n` leaves (`0 < k < n`), via + /// `gen_ancestry_proof`. + fn extension(all: &[Hash], k: usize, n: usize) -> MMRExtensionProof { + let store = MemStore::::default(); + let mut mmr = MemMMR::>::new(0, &store); + for l in &all[..n] { + mmr.push(*l).unwrap(); + } + let ap = mmr.gen_ancestry_proof(mmr_size(k)).unwrap(); + MMRExtensionProof { + new_mmr_size: ap.prev_peaks_proof.mmr_size(), + connecting_nodes: ap.prev_peaks_proof.proof_items().to_vec(), + } + } + + fn ch(recipient: u32) -> StreamId { + StreamId::Channel { recipient: recipient.into(), domain: 0, num: 0 } + } + + /// A keyed `StreamsRoot`-trie membership proof for one stream out of `n_streams` active + /// streams. Recipients are clustered (`2000..2000+n`), so keys diverge only in the low bits — + /// the realistic worst case for Patricia depth. + fn tree_proof(n_streams: u32) -> StreamProof { + let entries: Vec<(StreamId, Hash)> = + (0..n_streams).map(|i| (ch(2_000 + i), H256::repeat_byte(i as u8))).collect(); + let target = ch(2_000 + n_streams / 2); + gen_stream_proof(entries, target).expect("target stream is present; qed").1 + } + + /// An extension proof of a given connecting-node count, for sizing the design's *day-scale* + /// worst case (≈10⁹ leaves → ~30 nodes) without building a 10⁹-leaf MMR: PoV cost is the + /// *encoded* size, which is a pure function of the node count, so synthetic nodes size + /// identically to real ones. Each node is `(u64 position, Hash)` = 40 B — 8 B more than the + /// design's hash-only (32 B) count. + fn synthetic_extension(nodes: usize) -> MMRExtensionProof { + MMRExtensionProof { + new_mmr_size: u64::MAX / 2, // a day-scale size; placement is irrelevant to encoded size + connecting_nodes: (0..nodes as u64).map(|i| (i, H256::repeat_byte(i as u8))).collect(), + } + } + + #[test] + fn extension_computes_root_and_rejects_tampering() { + let all = leaves(5); + let ext = extension(&all, 2, 5); + // Computes root@5 from frontier@2 + O(log n) connecting nodes (never a declared root). + assert_eq!(ext.verify(&frontier_at(&all, 2)), Some(root_at(&all, 5))); + // Wrong starting frontier → does not reconstruct the extended root. + assert_ne!(ext.verify(&frontier_at(&all, 3)), Some(root_at(&all, 5))); + // Tampered connecting node → a different computed root. + let mut bad = ext.clone(); + bad.connecting_nodes[0].1 = H256::repeat_byte(0xff); + assert_ne!(bad.verify(&frontier_at(&all, 2)), Some(root_at(&all, 5))); + // Identity extension yields the frontier's own root. + let id = MMRExtensionProof::identity(); + assert_eq!(id.verify(&frontier_at(&all, 3)), Some(root_at(&all, 3))); + } + + #[test] + fn over_long_extension_proof_is_rejected() { + let all = leaves(5); + let mut ext = extension(&all, 2, 5); + // Pad the connecting nodes beyond the defense-in-depth ceiling: must reject, not do the + // work. + ext.connecting_nodes = vec![(0u64, H256::zero()); MAX_EXTENSION_CONNECTING_NODES + 1]; + assert_eq!(ext.verify(&frontier_at(&all, 2)), None); + } + + #[test] + fn stitch_caught_up_needs_no_advance() { + let all = leaves(5); + let i1 = Interval { start: MmrRoot(H256::zero()), end: frontier_at(&all, 2) }; + let i2 = Interval { start: root_at(&all, 2), end: frontier_at(&all, 5) }; + assert_eq!(stitch(&[i1, i2], &[]).unwrap(), frontier_at(&all, 5)); + } + + #[test] + fn stitch_bridges_a_gap_and_flags_bad_advance_counts() { + let all = leaves(5); + let i1 = Interval { start: MmrRoot(H256::zero()), end: frontier_at(&all, 2) }; + // Block 2 jumped forward to context @4 before continuing to @5. + let i2 = Interval { start: root_at(&all, 4), end: frontier_at(&all, 5) }; + let advance = extension(&all, 2, 4); + + assert_eq!( + stitch(&[i1.clone(), i2.clone()], &[advance.clone()]).unwrap(), + frontier_at(&all, 5) + ); + // Missing the gap proof. + assert_eq!(stitch(&[i1.clone(), i2.clone()], &[]), Err(LiftError::MissingAdvance)); + // A stray advance where the chain was already continuous. + let cont = Interval { start: root_at(&all, 2), end: frontier_at(&all, 5) }; + assert_eq!(stitch(&[i1, cont], &[advance]), Err(LiftError::StrayAdvance)); + } + + #[test] + fn hot_path_single_block_single_stream() { + let all = leaves(3); + let root3 = root_at(&all, 3); + let stream = ch(2000); + // StreamsRoot over the single active stream at root@3. + let entries = vec![(stream, root3.0)]; + let expected = streams_root(entries.clone()).unwrap(); + let (_r, tree_proof) = gen_stream_proof(entries, stream).unwrap(); + + // One interval, endpoint = frontier@3; identity extension (caught up). + let interval = Interval { start: MmrRoot(H256::zero()), end: frontier_at(&all, 3) }; + let lift = RequiresLift { + advances: Vec::new(), + extension: MMRExtensionProof::identity(), + tree_proof, + }; + let entry = + build_requires_entry(ParaId::from(1000), &[(stream, vec![interval])], &[lift]).unwrap(); + assert_eq!(entry, (ParaId::from(1000), expected)); + } + + #[test] + fn build_requires_end_to_end_and_source_match() { + let all = leaves(3); + let root3 = root_at(&all, 3); + let source = ParaId::from(1000); + let stream = ch(2000); + let entries = vec![(stream, root3.0)]; + let expected = streams_root(entries.clone()).unwrap(); + let (_r, tree_proof) = gen_stream_proof(entries, stream).unwrap(); + + let record = ConsumptionRecord { + entries: BTreeMap::from([( + source, + vec![( + stream, + Interval { start: MmrRoot(H256::zero()), end: frontier_at(&all, 3) }, + )], + )]), + }; + let mut lifts = BTreeMap::new(); + lifts.insert( + source, + vec![RequiresLift { + advances: Vec::new(), + extension: MMRExtensionProof::identity(), + tree_proof, + }], + ); + + let requires = build_requires(&[record.clone()], &lifts).unwrap(); + assert_eq!(requires.get(source), Some(&expected)); + + // A lift for an unrecorded source (or a missing one) is rejected. + let mut extra = lifts.clone(); + extra.insert(ParaId::from(9999), Vec::new()); + assert_eq!(build_requires(&[record], &extra), Err(LiftError::LiftSourceMismatch)); + } + + #[test] + fn divergent_roots_rejected() { + let all = leaves(3); + let root3 = root_at(&all, 3); + let source = ParaId::from(1000); + let (sa, sb) = (ch(2000), ch(3000)); + + // Two single-stream trees → each stream's tree_proof folds to a DIFFERENT StreamsRoot. + let (_ra, proof_a) = gen_stream_proof(vec![(sa, root3.0)], sa).unwrap(); + let (_rb, proof_b) = gen_stream_proof(vec![(sb, root3.0)], sb).unwrap(); + + let mk = |tree_proof| RequiresLift { + advances: Vec::new(), + extension: MMRExtensionProof::identity(), + tree_proof, + }; + let streams = vec![ + (sa, vec![Interval { start: MmrRoot(H256::zero()), end: frontier_at(&all, 3) }]), + (sb, vec![Interval { start: MmrRoot(H256::zero()), end: frontier_at(&all, 3) }]), + ]; + assert_eq!( + build_requires_entry(source, &streams, &[mk(proof_a), mk(proof_b)]), + Err(LiftError::DivergentRoots), + ); + } + + /// Empirical PoV-cost report for the requires-lift (run with `-- --nocapture`). + /// + /// v0.5 replaces the pre-v0.5 point-3 two-level `sp_trie` read (relay `paras::Heads` proof + a + /// sender-state proof) with the POV-carried [`RequiresLift`], committing against the sender's + /// own keyed binary Patricia trie — no `sp_trie` witnesses. This measures the real + /// SCALE-encoded size along its two dimensions — the `tree_proof` ([`StreamProof`], `O(log S)` + /// in active streams) and the `extension` ([`MMRExtensionProof`], `O(log n)` in messages) — + /// and asserts both stay logarithmic and the protocol's bounded worst case sits well under + /// `MAX_POV_SIZE`. + #[test] + fn pov_cost_report() { + println!("\n===== spec-msg v0.5 requires-lift PoV cost (SCALE-encoded, real proofs) ====="); + + // tree_proof: keyed StreamsRoot Patricia trie, O(log S) in a source's active streams. + println!("\n[tree_proof] StreamProof over S active streams (clustered keys)"); + println!(" S | steps | encoded"); + for s in [1u32, 4, 16, 64, 256, 1024] { + let proof = tree_proof(s); + // Patricia-compressed: depth is O(log S), never the 64-bit key width. + assert!( + proof.steps.len() <= 20, + "tree proof depth {} too large for S={s}", + proof.steps.len() + ); + println!(" {s:>4} | {:>5} | {:>5} B", proof.steps.len(), proof.encoded_size()); + } + + // extension: MMR ancestry, O(log n) in the stream's total messages. Build one MMR to N + // leaves; measure the ancestry proof bridging an endpoint `behind` leaves back from the + // tip (the unconsumed tail a lagging receiver must extend over). + const N: usize = 100_000; + let store = MemStore::::default(); + let mut mmr = MemMMR::>::new(0, &store); + for i in 0..N as u64 { + mmr.push(H256::from_low_u64_be(i)).unwrap(); + } + // NB: each connecting node is `(u64 position, Hash)` = 40 B, vs the design's hash-only 32 B + // count — so measured `encoded` runs ~25% above a "N hashes × 32 B" estimate. + println!("\n[extension] MMRExtensionProof to tip N={N}, endpoint `behind` leaves back"); + println!(" behind | nodes | encoded (40 B/node)"); + let mut sample_ext = MMRExtensionProof::identity(); + for behind in [1usize, 16, 256, 4_096, N / 2] { + let k = N - behind; + let ap = mmr.gen_ancestry_proof(mmr_size(k)).unwrap(); + let ext = MMRExtensionProof { + new_mmr_size: ap.prev_peaks_proof.mmr_size(), + connecting_nodes: ap.prev_peaks_proof.proof_items().to_vec(), + }; + // O(log N) connecting nodes — independent of how far the tail stretches. + assert!( + ext.connecting_nodes.len() <= 64, + "extension not O(log N): {}", + ext.connecting_nodes.len() + ); + println!( + " {behind:>7} | {:>5} | {:>5} B", + ext.connecting_nodes.len(), + ext.encoded_size() + ); + sample_ext = ext; + } + + // combined: one RequiresLift = extension + tree_proof (advances empty in the hot path; each + // interval-chain gap would add one more MMRExtensionProof of the same order). + let lift = RequiresLift { + advances: Vec::new(), + extension: sample_ext.clone(), // endpoint N/2 behind on an N-message stream + tree_proof: tree_proof(64), // source with 64 active streams + }; + let per_lift = lift.encoded_size(); + println!( + "\n[lift] one RequiresLift (ext@N/2 + tree_proof@S=64, advances=0): {per_lift} B" + ); + + // Per source: one lift per stream consumed from it (all fold to the SAME StreamsRoot). Per + // candidate: summed across sources. + println!("\n[candidate] requires-lift PoV = Σ sources × streams/source × per_lift"); + for (streams_per_source, sources) in [(1usize, 1usize), (2, 10), (4, 50)] { + let candidate = sources * streams_per_source * per_lift; + println!(" {sources:>3} sources × {streams_per_source} stream(s) = {candidate} B"); + } + let budget = MAX_POV_SIZE as usize; + + // --- Angle 1: empirical PoV share (real proofs, ≤100k messages, no gaps) --- + // A full-source candidate (MAX_SOURCES_PER_BLOCK sources) with a conservative *measured* + // lift (deepest 100k-message extension + a 1024-stream tree_proof): what fraction of the + // budget do real proofs at representative scale consume? Guards the typical footprint at + // <10%. + const STREAMS_PER_SOURCE: usize = 4; + let empirical_lift = RequiresLift { + advances: Vec::new(), + extension: sample_ext, // deepest 100k-message extension measured above + tree_proof: tree_proof(1024), + }; + let empirical_per_lift = empirical_lift.encoded_size(); + let full_candidate = + MAX_SOURCES_PER_BLOCK as usize * STREAMS_PER_SOURCE * empirical_per_lift; + println!( + "\n[empirical share] {} sources × {STREAMS_PER_SOURCE} streams × {empirical_per_lift} B \ + = {full_candidate} B ({:.2}% of MAX_POV_SIZE = {} MiB)", + MAX_SOURCES_PER_BLOCK, + 100.0 * full_candidate as f64 / budget as f64, + budget / (1024 * 1024), + ); + assert!( + full_candidate < budget / 10, + "empirical full-source candidate {full_candidate} B exceeds 10% of MAX_POV_SIZE", + ); + + // --- Angle 2: day-scale worst-case lift (the design's "Proof Size Considerations" sizing) + // --- The measurements above top out at N=100k messages; the design sizes against ~24 h + // of lag (≈10⁹ leaves → ~30 connecting nodes) *plus* one read-context gap (an advance + // proof). We can't build a 10⁹-leaf MMR, but PoV cost is the encoded size, so synthesize + // proofs of the worst-case shape. Ceiling per the design: ~4 KB/stream + ~2 KB/gap. + const DAY_SCALE_EXT_NODES: usize = 30; // log₂(10⁹) ≈ 30 (design's "~30 hashes") + let worst_lift = RequiresLift { + advances: vec![synthetic_extension(DAY_SCALE_EXT_NODES)], // one read-context gap + extension: synthetic_extension(DAY_SCALE_EXT_NODES), // day-scale unconsumed tail + tree_proof: tree_proof(1024), // 1024 active streams + }; + let worst_per_lift = worst_lift.encoded_size(); + println!( + "\n[worst-case lift] day-scale (~10⁹ leaves, {DAY_SCALE_EXT_NODES}-node ext + 1 gap + tree@S=1024): {worst_per_lift} B" + ); + // Regression guard: one worst-case lift stays bounded. The design ceilings a touched stream + // at ~4 KB + ~2 KB/gap ≈ 6 KB; guard at 8 KB so an O(n) extension (non-logarithmic) or an + // uncompressed tree walk trips this instead of silently eating the PoV. + assert!(worst_per_lift < 8 * 1024, "worst-case lift {worst_per_lift} B exceeds ~8 KB"); + + // Authoring-time budget (design: "the block must guarantee at authoring time that the worst + // case still fits"): the bound is *how many touched streams fit*, not a fixed per-candidate + // total. Even at the day-scale worst case the PoV must comfortably hold the source cap. + let max_streams = budget / worst_per_lift; + println!( + "[headroom] MAX_POV_SIZE {} MiB / {worst_per_lift} B ≈ {max_streams} day-scale touched streams \ + (source cap = {MAX_SOURCES_PER_BLOCK})", + budget / (1024 * 1024), + ); + // A candidate names ≤ MAX_SOURCES_PER_BLOCK sources; even at day-scale the PoV must hold at + // least that many touched streams, or authoring couldn't fill a block. (Ample margin here.) + assert!( + max_streams >= MAX_SOURCES_PER_BLOCK as usize, + "day-scale worst case leaves room for only {max_streams} streams (< {MAX_SOURCES_PER_BLOCK})", + ); + println!("===== end =====\n"); + } +} diff --git a/cumulus/primitives/spec-messaging/src/message.rs b/cumulus/primitives/spec-messaging/src/message.rs new file mode 100644 index 000000000000..bba57d5ee288 --- /dev/null +++ b/cumulus/primitives/spec-messaging/src/message.rs @@ -0,0 +1,410 @@ +// Copyright (C) Parity Technologies (UK) Ltd. +// This file is part of Cumulus. +// SPDX-License-Identifier: Apache-2.0 + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Off-chain message types + the protocol's concrete instantiation. +//! +//! Two things live here, both **off-chain** (the relay chain decodes none of them — it only matches +//! the `StreamsRoot` commitments in the `Provides`/`Requires` UMP signals): +//! +//! - **Concrete instantiation** — [`SpecHasher`] (the hasher), [`MaxSpeculativeMessageLen`] (the +//! payload bound), and [`leaf_hash`] (a payload → MMR leaf). A v0.5 message is *just a bounded +//! payload keyed by `StreamId`*; there is no message struct — source / stream / position are all +//! structural (the sender's outbox, the `StreamId` key, the MMR leaf index), so only the payload +//! and its hash are primitives. +//! - **Fetch protocol** — [`MessagesRequest`] / [`MessagesResponse`]: a collator fetches a range of +//! a source stream's messages and authenticates the response ([`verify_messages_response`]) +//! against a `StreamsRoot` it has independently verified, reusing the same `extension` + +//! `tree_proof` machinery as the requires-lift (see [`crate::lift`]). The lossy event/register +//! read path (`EventRequest`/`EventResponse`) lands with the event-stream subsystem. + +use alloc::vec::Vec; +use codec::{Decode, DecodeWithMemTracking, Encode, MaxEncodedLen}; +use polkadot_core_primitives::Hash; +use scale_info::TypeInfo; +use sp_core::ConstU32; +use sp_runtime::traits::Hash as HashT; + +use crate::{ + lift::{MMRExtensionProof, MmrFrontier}, + mmr::{Mmr, MmrAccumulator}, + stream::StreamId, + streams_root::{streams_root_from_proof, StreamProof, StreamsRoot}, + LEAF_TAG, +}; + +/// Maximum size in bytes of a single speculative message payload. +pub const MAX_SPECULATIVE_MESSAGE_LEN: u32 = 102_400; + +/// Bound for a single speculative message payload. +pub type MaxSpeculativeMessageLen = ConstU32; + +/// The hash function used throughout speculative messaging (leaf hashing, MMR merges, stream +/// roots). The crate primitives are generic over the hasher; this alias is the single concrete +/// choice for the protocol, so switching (e.g. to Keccak256) is a one-line change here. +pub type SpecHasher = sp_runtime::traits::BlakeTwo256; + +/// Hash a payload into a stream MMR leaf under `leaf_version`: `H(LEAF_TAG ++ leaf_version ++ +/// payload)`. +/// +/// A pure function of `(leaf_version, payload)` — the only thing v0.5 needs of a "message". Source, +/// stream, and position are structural (the sender's outbox, the `StreamId` key, the MMR leaf +/// index), so none are in the preimage. `leaf_version` domains are hash-disjoint, so only the +/// correct version reproduces a committed root. +pub fn leaf_hash>(leaf_version: u8, payload: &[u8]) -> Hash { + let mut preimage = Vec::new(); + preimage.extend_from_slice(&LEAF_TAG.to_le_bytes()); + preimage.extend_from_slice(&leaf_version.to_le_bytes()); + preimage.extend_from_slice(payload); + ::hash(&preimage) +} + +/// A leaf position within a stream's MMR (its leaf count). Newtype, per the design's +/// `MessagePosition` — a position is not an arbitrary `u64`. +#[derive( + Clone, + Copy, + Encode, + Decode, + DecodeWithMemTracking, + MaxEncodedLen, + PartialEq, + Eq, + PartialOrd, + Ord, + Debug, + Default, + TypeInfo, +)] +pub struct MessagePosition(pub u64); + +/// Off-chain request for a range of a stream's messages, to be verified under a chosen +/// `StreamsRoot`. +#[derive(Clone, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, TypeInfo)] +pub struct MessagesRequest { + /// The requested stream of the serving chain. + pub stream: StreamId, + /// Where to start — typically the receiver's frontier leaf count. + pub start: MessagePosition, + /// The `StreamsRoot` the response must verify under: the requester's chosen dependency, a root + /// it has independently authenticated — never a newer, possibly-unconfirmed one (freshness is + /// the requester's own job: re-request under a newer root once verified). + pub under: StreamsRoot, + /// Response size bound (the server may cap harder); fetching stays chunked and resumable no + /// matter how large the backlog. `0` requests a payload-free proof (lift material). + pub max_bytes: u32, +} + +/// Off-chain response: payloads from `base` on, plus the proofs binding them — and everything +/// before them — to the requested `StreamsRoot`. Independently verifiable via +/// [`verify_messages_response`]; nothing is trusted (fabricated peaks/payloads/proofs cannot +/// reproduce a committed root). +#[derive(Clone, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, TypeInfo)] +pub struct MessagesResponse { + /// Position of the first payload (trust-free hint; a lie only fails the proofs). + pub base: MessagePosition, + /// Leaf-format version for these payloads (trust-free hint: versions are hash-disjoint, so + /// only the correct one reproduces a committed root). A response never spans a version + /// change. + pub leaf_version: u8, + /// The payloads, in MMR order; payload `i` has position `base + i`. + pub payloads: Vec>, + /// The stream's peak set at `base` (≤ 64 hashes). Lets a consumer holding no frontier + /// recompute; fabricated peaks cannot extend to a committed root. + pub start_peaks: Vec, + /// From the frontier recomputed over `payloads` to the stream's entry under `under`; the + /// identity extension when the payloads already reach it. + pub extension: MMRExtensionProof, + /// Walked from the extension's output, yields the `StreamsRoot` the response verifies under. + pub tree_proof: StreamProof, +} + +/// Verify a [`MessagesResponse`] for `stream` under `under`, returning the authenticated payloads +/// on success. +/// +/// Recomputes the stream frontier from `start_peaks` + the hashed `payloads`, extends it to the +/// stream's current root (`extension`), and walks the keyed tree (`tree_proof`) to a `StreamsRoot` +/// — which must equal `under`. This is the same `extension` + `tree_proof` check the requires-lift +/// performs, so a fetched batch and a consumption lift authenticate identically. +pub fn verify_messages_response( + stream: StreamId, + under: StreamsRoot, + resp: &MessagesResponse, +) -> Option>> { + // `start_peaks` and `base` are untrusted. A well-formed frontier has exactly one peak per set + // bit of its leaf count; reject any other shape so a crafted response cannot drive + // `Mmr::append` into a pop-from-empty panic. This equality also bounds `start_peaks` to ≤ 64 + // (a `u64` has ≤ 64 set bits), enforcing the documented peak-set limit. Also reject a `base + + // len` that would overflow the leaf counter (only reachable near `u64::MAX`, but keeps the + // accumulator arithmetic total). A rejected response is simply unverifiable — the correct + // outcome, reached without panicking. + if resp.start_peaks.len() != resp.base.0.count_ones() as usize { + return None; + } + resp.base.0.checked_add(resp.payloads.len() as u64)?; + + // Frontier at `base` (the response's peak set), then append the response's payloads as leaves. + let mut mmr = Mmr::::from_parts(resp.start_peaks.clone(), resp.base.0); + for payload in &resp.payloads { + mmr.append(leaf_hash::(resp.leaf_version, payload)); + } + let (peaks, leaf_count) = mmr.into_parts(); + let frontier = MmrFrontier { peaks, leaf_count }; + + // Extend to the stream's current root, then walk the tree to the committed `StreamsRoot`. + let current = resp.extension.verify(&frontier)?; + let root = streams_root_from_proof(stream, current.0, &resp.tree_proof)?; + (root == under).then(|| resp.payloads.clone()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + lift::{build_requires, ConsumptionRecord, Interval, RequiresLift}, + mmr::{root_from_peaks, SpecMerge}, + streams_root::{gen_stream_proof, streams_root}, + }; + use alloc::collections::BTreeMap; + use mmr_lib::{ + leaf_index_to_mmr_size, + util::{MemMMR, MemStore}, + }; + use polkadot_parachain_primitives::primitives::Id as ParaId; + use sp_core::H256; + + fn ch(recipient: u32) -> StreamId { + StreamId::Channel { recipient: recipient.into(), domain: 0, num: 0 } + } + + fn mmr_size(k: usize) -> u64 { + if k == 0 { + 0 + } else { + leaf_index_to_mmr_size((k - 1) as u64) + } + } + + /// Peaks-only frontier after the first `k` leaves. + fn frontier_at(leaves: &[Hash], k: usize) -> MmrFrontier { + let mut mmr = Mmr::::new(); + for l in &leaves[..k] { + mmr.append(*l); + } + let (peaks, leaf_count) = mmr.into_parts(); + MmrFrontier { peaks, leaf_count } + } + + /// The stream root over the first `n` leaves. + fn current_root(leaves: &[Hash], n: usize) -> Hash { + let store = MemStore::::default(); + let mut mmr = MemMMR::>::new(0, &store); + for l in &leaves[..n] { + mmr.push(*l).unwrap(); + } + mmr.get_root().unwrap() + } + + /// O(log n) ancestry extension from `k` leaves to `n` leaves. + fn ancestry(leaves: &[Hash], k: usize, n: usize) -> MMRExtensionProof { + let store = MemStore::::default(); + let mut mmr = MemMMR::>::new(0, &store); + for l in &leaves[..n] { + mmr.push(*l).unwrap(); + } + let ap = mmr.gen_ancestry_proof(mmr_size(k)).unwrap(); + MMRExtensionProof { + new_mmr_size: ap.prev_peaks_proof.mmr_size(), + connecting_nodes: ap.prev_peaks_proof.proof_items().to_vec(), + } + } + + /// End-to-end composition, exercising the non-trivial branches (`base > 0`, a *real* extension + /// bridging an unconsumed tail): a sender stream of 6 messages commits a `StreamsRoot`; a + /// **partial** fetch (base 2, messages 2..3, extension bridging 4→6) verifies under that root; + /// and the receiver's consumption of those messages lifts (the same 4→6 extension + + /// `tree_proof`) into the **same** `StreamsRoot` via `build_requires`. Fetch and requires run + /// the identical `extension` + `tree_proof` check — so this demonstrates they compose. + #[test] + fn composition_fetch_verify_then_build_requires() { + let source = ParaId::from(1000); + let stream = ch(2000); + + let payloads: Vec> = (0..6u8).map(|i| vec![i, i.wrapping_add(10)]).collect(); + let leaves: Vec = payloads.iter().map(|p| leaf_hash::(0, p)).collect(); + + // Sender's current stream root (6 leaves) and the StreamsRoot committing it. + let r_current = current_root(&leaves, 6); + let under = streams_root(vec![(stream, r_current)]).unwrap(); + let (_r, tree_proof) = gen_stream_proof(vec![(stream, r_current)], stream).unwrap(); + + // Fetch: base 2 (non-empty start_peaks), payloads [2,3] only → a real extension (4 → 6). + let resp = MessagesResponse { + base: MessagePosition(2), + leaf_version: 0, + payloads: vec![payloads[2].clone(), payloads[3].clone()], + start_peaks: frontier_at(&leaves, 2).peaks, + extension: ancestry(&leaves, 4, 6), + tree_proof: tree_proof.clone(), + }; + assert_eq!( + verify_messages_response(stream, under, &resp), + Some(vec![payloads[2].clone(), payloads[3].clone()]), + ); + + // Requires: receiver consumed messages 2,3 (frontier 2 → 4); lift the endpoint (4 → 6). + let record = ConsumptionRecord { + entries: BTreeMap::from([( + source, + vec![( + stream, + Interval { + start: frontier_at(&leaves, 2).root().unwrap(), + end: frontier_at(&leaves, 4), + }, + )], + )]), + }; + let mut lifts = BTreeMap::new(); + lifts.insert( + source, + vec![RequiresLift { + advances: Vec::new(), + extension: ancestry(&leaves, 4, 6), + tree_proof, + }], + ); + + let requires = build_requires(&[record], &lifts).unwrap(); + // The fetch and the requires resolve to the SAME committed StreamsRoot. + assert_eq!(requires.get(source), Some(&under)); + } + + #[test] + fn leaf_hash_is_payload_only_and_version_disjoint() { + let payload = b"hello"; + // Matches the explicit preimage `H(LEAF_TAG ++ version ++ payload)`. + let mut pre = Vec::new(); + pre.push(LEAF_TAG); + pre.push(0u8); + pre.extend_from_slice(payload); + assert_eq!(leaf_hash::(0, payload), ::hash(&pre)); + // Different version → different leaf (hash-disjoint domains). + assert_ne!(leaf_hash::(0, payload), leaf_hash::(1, payload)); + } + + /// End-to-end: a sender builds a stream MMR, commits its root into a `StreamsRoot`, and serves + /// a response covering the whole stream from `base = 0`; the receiver authenticates it under + /// the committed root and recovers the payloads. + #[test] + fn messages_response_round_trips_under_streams_root() { + let payloads: Vec> = (0..4u8).map(|i| vec![i, i + 1, i + 2]).collect(); + + // Sender stream MMR over the payload leaves; its root is the stream root. + let store = MemStore::::default(); + let mut src = MemMMR::>::new(0, &store); + for p in &payloads { + src.push(leaf_hash::(0, p)).unwrap(); + } + let stream_root = src.get_root().unwrap(); + + let stream = ch(2000); + let entries = vec![(stream, stream_root)]; + let under = streams_root(entries.clone()).unwrap(); + let (_r, tree_proof) = gen_stream_proof(entries, stream).unwrap(); + + // Full stream from base 0: empty start_peaks, identity extension (payloads reach the root). + let resp = MessagesResponse { + base: MessagePosition(0), + leaf_version: 0, + payloads: payloads.clone(), + start_peaks: Vec::new(), + extension: MMRExtensionProof::identity(), + tree_proof, + }; + + assert_eq!(verify_messages_response(stream, under, &resp), Some(payloads)); + // Sanity: the recomputed root matches the sender's. + let mut mmr = Mmr::::new(); + for p in &resp.payloads { + mmr.append(leaf_hash::(0, p)); + } + assert_eq!(root_from_peaks::(mmr.peaks()), stream_root); + } + + #[test] + fn messages_response_rejects_wrong_root_or_tampered_payload() { + let payloads: Vec> = (0..3u8).map(|i| vec![i]).collect(); + let store = MemStore::::default(); + let mut src = MemMMR::>::new(0, &store); + for p in &payloads { + src.push(leaf_hash::(0, p)).unwrap(); + } + let stream = ch(2000); + let entries = vec![(stream, src.get_root().unwrap())]; + let under = streams_root(entries.clone()).unwrap(); + let (_r, tree_proof) = gen_stream_proof(entries, stream).unwrap(); + + let resp = MessagesResponse { + base: MessagePosition(0), + leaf_version: 0, + payloads, + start_peaks: Vec::new(), + extension: MMRExtensionProof::identity(), + tree_proof, + }; + + // Wrong `under`. + assert_eq!( + verify_messages_response(stream, StreamsRoot(H256::repeat_byte(0xff)), &resp), + None + ); + // Tampered payload → recomputed root diverges → fails. + let mut bad = resp.clone(); + bad.payloads[0] = vec![0xAA]; + assert_eq!(verify_messages_response(stream, under, &bad), None); + } + + #[test] + fn malformed_frontier_rejects_without_panicking() { + let stream = ch(2000); + let under = StreamsRoot(H256::repeat_byte(0xff)); + + // `start_peaks` shape inconsistent with `base` (empty peaks, base = 1): the pre-fix + // pop-from-empty panic in `Mmr::append`. Must return None, not panic. + let crafted = MessagesResponse { + base: MessagePosition(1), + leaf_version: 0, + payloads: vec![vec![0x01]], + start_peaks: Vec::new(), + extension: MMRExtensionProof::identity(), + tree_proof: StreamProof { steps: Vec::new() }, + }; + assert_eq!(verify_messages_response(stream, under, &crafted), None); + + // `base + payloads` overflowing the leaf counter must also reject cleanly. `base = + // u64::MAX` has 64 set bits, so give 64 peaks to pass the shape check and hit the + // overflow guard. + let overflow = MessagesResponse { + base: MessagePosition(u64::MAX), + leaf_version: 0, + payloads: vec![vec![0x01]], + start_peaks: vec![H256::zero(); 64], + extension: MMRExtensionProof::identity(), + tree_proof: StreamProof { steps: Vec::new() }, + }; + assert_eq!(verify_messages_response(stream, under, &overflow), None); + } +} diff --git a/cumulus/primitives/spec-messaging/src/mmr.rs b/cumulus/primitives/spec-messaging/src/mmr.rs new file mode 100644 index 000000000000..bc1d45b52e64 --- /dev/null +++ b/cumulus/primitives/spec-messaging/src/mmr.rs @@ -0,0 +1,263 @@ +// Copyright (C) Parity Technologies (UK) Ltd. +// This file is part of Cumulus. +// SPDX-License-Identifier: Apache-2.0 + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! MMR primitives for the Speculative Messaging protocol. +//! +//! The hash function is a type parameter `H` (a `sp_runtime::traits::Hash` with a +//! 32-byte output), so the protocol can switch hashers (e.g. blake2 ↔ keccak) +//! without changing this code. Consumers pick a concrete hasher; the POC uses +//! `polkadot_primitives::v9::SpecHasher` (blake2_256). +//! +//! Domain separation: leaves are hashed by [`crate::message::leaf_hash`] (`LEAF_TAG`), inner +//! nodes by [`SpecMerge::merge`] (`INNER_TAG`), and peak-bagging by +//! [`SpecMerge::merge_peaks`] (`PEAK_TAG`). `mmr_lib` calls `merge` for tree nodes +//! and the overridable `merge_peaks` when bagging peaks, so no two roles collide on +//! the same hash. +//! +//! [`SpecMerge`] is the `mmr_lib::Merge` used for inclusion proofs (`gen_proof`/ +//! `MerkleProof::verify`) and append-only ancestry proofs (`gen_ancestry_proof`/ +//! `verify_incremental`). [`Mmr`] is a peaks-only [`MmrAccumulator`] for the +//! sender's on-chain state. + +use alloc::vec::Vec; +use core::marker::PhantomData; +use mmr_lib::{Error as MmrError, Merge}; +use polkadot_core_primitives::Hash; +use sp_runtime::traits::Hash as HashT; + +use crate::{INNER_TAG, PEAK_TAG}; + +/// Domain-tagged merge for the speculative-messaging MMR, generic over the hash +/// function `H`. Used as the `mmr_lib::Merge` implementation for the subtree. +pub struct SpecMerge(PhantomData); + +impl> Merge for SpecMerge { + type Item = Hash; + + /// Inner-node merge: `H(INNER_TAG ++ left ++ right)`. + fn merge(left: &Hash, right: &Hash) -> Result { + Ok(tagged_node::(INNER_TAG, left, right)) + } + + /// Peak-bagging merge: `H(PEAK_TAG ++ left ++ right)`. Domain-separated from + /// `merge` so a bagged value can never be reinterpreted as an inner node. + fn merge_peaks(left: &Hash, right: &Hash) -> Result { + Ok(tagged_node::(PEAK_TAG, left, right)) + } +} + +/// `H(tag ++ left ++ right)`. +fn tagged_node>(tag: u8, left: &Hash, right: &Hash) -> Hash { + let mut preimage = [0u8; 1 + 32 + 32]; + preimage[0] = tag; + preimage[1..33].copy_from_slice(left.as_bytes()); + preimage[33..65].copy_from_slice(right.as_bytes()); + ::hash(&preimage) +} + +/// Compute the MMR root from its non-empty peaks (highest to lowest), matching +/// `mmr_lib`'s bagging (`merge_peaks(right, left)` folded right-to-left). +/// +/// This lets the on-chain outbox keep only the O(log n) peaks and still derive the +/// same stream root that `mmr_lib`'s `MMR::get_root` and `MerkleProof::verify` +/// produce. +/// +/// # Panics +/// +/// Panics on an empty `peaks` slice. An empty MMR has no root (`mmr_lib::MMR:: +/// get_root` itself errors on empty), and the protocol never commits one: a +/// per-stream outbox state only exists after at least one append, and empty +/// streams are omitted from the `StreamsRoot` tree. So callers always pass a +/// non-empty slice. +pub fn root_from_peaks>(peaks: &[Hash]) -> Hash { + let mut iter = peaks.iter().rev(); + let mut acc = *iter + .next() + .expect("root_from_peaks called on empty peaks; an empty MMR has no root; qed"); + for left in iter { + // mmr_lib bags as merge_peaks(right, left); `acc` carries the right side. + acc = + as Merge>::merge_peaks(&acc, left).expect("SpecMerge is infallible; qed"); + } + acc +} + +/// Generic interface for accumulating MMR leaves and producing a root commitment. +/// Abstracted as a trait so the underlying scheme (MMR, or in the future MMB) can +/// change without breaking callers. +pub trait MmrAccumulator { + /// Append a new leaf hash to the accumulator. + fn append(&mut self, leaf: Hash); + + /// The current root commitment over all appended leaves. + fn root(&self) -> Hash; + + /// The number of leaves appended so far. + fn size(&self) -> u64; +} + +/// Peaks-only MMR accumulator (O(log n) state), generic over the hash function. +/// +/// Peak hashes are identical to `mmr_lib`'s, so [`root`](MmrAccumulator::root) +/// (via [`root_from_peaks`]) equals `mmr_lib::MMR::get_root`, and proofs generated +/// by a full `mmr_lib::MMR` over the same leaves verify against this root. The +/// sender keeps `(peaks, size)` in storage and round-trips through +/// [`from_parts`](Mmr::from_parts) / [`into_parts`](Mmr::into_parts). +pub struct Mmr { + peaks: Vec, + size: u64, + _marker: PhantomData, +} + +impl Mmr { + /// An empty accumulator. + pub fn new() -> Self { + Self { peaks: Vec::new(), size: 0, _marker: PhantomData } + } + + /// Reconstruct an accumulator from previously stored `(peaks, size)`. + pub fn from_parts(peaks: Vec, size: u64) -> Self { + Self { peaks, size, _marker: PhantomData } + } + + /// Decompose into `(peaks, size)` for storage. + pub fn into_parts(self) -> (Vec, u64) { + (self.peaks, self.size) + } + + /// The current peaks (highest to lowest). + pub fn peaks(&self) -> &[Hash] { + &self.peaks + } +} + +impl Default for Mmr { + fn default() -> Self { + Self::new() + } +} + +impl> MmrAccumulator for Mmr { + /// Append a leaf, merging equal-height peaks via [`SpecMerge`] (matching + /// `mmr_lib`'s internal node construction). After this call, + /// `peaks.len() == size.count_ones()`. + fn append(&mut self, leaf: Hash) { + let mut node = leaf; + // Going from `size` to `size + 1` leaves merges exactly + // `trailing_zeros(size + 1)` pairs of peaks (binary-counter carry). + let merges = (self.size + 1).trailing_zeros(); + for _ in 0..merges { + let left = self.peaks.pop().expect("a peak exists for each merge; qed"); + node = + as Merge>::merge(&left, &node).expect("SpecMerge is infallible; qed"); + } + self.peaks.push(node); + self.size += 1; + } + + fn root(&self) -> Hash { + root_from_peaks::(&self.peaks) + } + + fn size(&self) -> u64 { + self.size + } +} + +#[cfg(test)] +mod tests { + use super::*; + use mmr_lib::{ + leaf_index_to_pos, + util::{MemMMR, MemStore}, + MerkleProof, + }; + use sp_runtime::traits::BlakeTwo256; + + type H = BlakeTwo256; + + fn h(byte: u8) -> Hash { + Hash::repeat_byte(byte) + } + + #[test] + fn merge_and_merge_peaks_are_domain_separated() { + let a = h(1); + let b = h(2); + // Inner-node merge and peak-bagging of the same inputs must differ (domain tags). + assert_ne!( + as Merge>::merge(&a, &b).unwrap(), + as Merge>::merge_peaks(&a, &b).unwrap() + ); + } + + #[test] + fn merge_is_order_sensitive() { + let a = h(1); + let b = h(2); + assert_ne!( + as Merge>::merge(&a, &b).unwrap(), + as Merge>::merge(&b, &a).unwrap() + ); + } + + #[test] + fn accumulator_root_matches_mmr_lib() { + // Build via the peaks-only accumulator and via a full mmr_lib MMR; the + // roots and peak-count invariant must agree. + let mut acc = Mmr::::new(); + let store = MemStore::::default(); + let mut reference = MemMMR::>::new(0, &store); + for i in 1..=5u8 { + acc.append(h(i)); + reference.push(h(i)).unwrap(); + assert_eq!(acc.peaks().len() as u32, acc.size().count_ones()); + } + assert_eq!(acc.root(), reference.get_root().unwrap()); + } + + #[test] + fn mmr_root_frozen_vector() { + // FROZEN consensus vector: appending these fixed leaves must always bag to this exact root. + // Any change to `SpecMerge`'s inner/peak domain tags or the peak-bagging order is a + // consensus break and breaks this test. (Round-trip tests can't catch a silent + // byte-format change.) + let mut acc = Mmr::::new(); + for i in 1..=5u8 { + acc.append(h(i)); + } + assert_eq!( + acc.root(), + Hash::from(hex_literal::hex!( + "9aadc77e51dcbf70a5a11751163a4478d054674b5976c903e273bb7c3712158b" + )) + ); + } + + #[test] + fn inclusion_proof_round_trips() { + let store = MemStore::::default(); + let mut mmr = MemMMR::>::new(0, &store); + let positions: Vec = (0..6u8).map(|i| mmr.push(h(i)).unwrap()).collect(); + let root = mmr.get_root().unwrap(); + + let proof: MerkleProof> = + mmr.gen_proof(vec![leaf_index_to_pos(1), leaf_index_to_pos(4)]).unwrap(); + + assert!(proof.verify(root, vec![(positions[1], h(1)), (positions[4], h(4))]).unwrap()); + assert!(!proof.verify(root, vec![(positions[1], h(99)), (positions[4], h(4))]).unwrap()); + } +} diff --git a/cumulus/primitives/spec-messaging/src/stream.rs b/cumulus/primitives/spec-messaging/src/stream.rs new file mode 100644 index 000000000000..3754c2bc9557 --- /dev/null +++ b/cumulus/primitives/spec-messaging/src/stream.rs @@ -0,0 +1,241 @@ +//! `StreamId` — the relay-invisible, parachain-structured stream key (spec-msg v0.5). +//! +//! Each stream a chain maintains is one MMR keyed by a `StreamId`. The relay chain never sees an id +//! — only the `StreamsRoot` (the keyed-tree root over stream roots) reaches it; ids live in tree +//! paths, proofs and networking. The full stream key is `(sender ParaId, StreamId)`, with the +//! sender implicit in the sender's own candidate. +//! +//! # Canonical encoding (CONSENSUS-CRITICAL, frozen) +//! +//! Manual SCALE, always **exactly 8 bytes**, big-endian — so lexicographic byte order equals +//! numeric field-tuple order (kinds cluster; sequential ids are neighbours). SCALE-encoding a +//! `StreamId` *is* the commitment-tree key derivation; there is no second format to confuse it +//! with, which is why the impl is manual (default SCALE ints are little-endian). +//! +//! ```text +//! Channel → 0x00 ++ recipient(be32) ++ domain(u8) ++ num(be16) +//! Ack → 0x01 ++ recipient(be32) ++ domain(u8) ++ num(be16) +//! Broadcast → 0x02 ++ domain(be16) ++ subdomain(u8) ++ num(be32) +//! Private → kind(0x80..=0xFF) ++ body[7] +//! ``` +//! +//! `Decode` enforces canonicality (fixed 8-byte length, no redundant encodings) and REJECTS +//! reserved kinds `0x03..=0x7F`: a correct consensus path only ever keys streams it knows, so an +//! unknown kind is a loud boundary error, not a value. `Ord` equals the canonical-encoding order +//! (verified by test). + +use codec::{ + Decode, DecodeWithMemTracking, Encode, EncodeLike, Error as CodecError, Input, MaxEncodedLen, + Output, +}; +use polkadot_parachain_primitives::primitives::Id as ParaId; + +/// Encoded length of every `StreamId`. +pub const STREAM_ID_LEN: usize = 8; + +/// A relay-invisible, parachain-structured stream key. See the module docs for the frozen encoding. +/// +/// The `recipient` / addressee is the party that *reads* the stream: a channel's messages are for +/// the recipient; an ack register is for the chain it grants credit to; broadcasts have no +/// addressee. +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, scale_info::TypeInfo)] +pub enum StreamId { + /// Ordered, flow-controlled, guaranteed, unidirectional channel to `recipient`. + Channel { recipient: ParaId, domain: u8, num: u16 }, + /// The receiver's lossy confirmation register for a channel, addressed to `recipient` (the + /// sender). + Ack { recipient: ParaId, domain: u8, num: u16 }, + /// Sender-wide event stream (pub-sub), no addressee. + Broadcast { domain: u16, subdomain: u8, num: u32 }, + /// Private-use kind (`0x80..=0xFF`); 7 chain-defined body bytes. + Private { kind: u8, body: [u8; 7] }, +} + +impl StreamId { + /// The chain the stream is ADDRESSED TO (its `recipient`), if the kind has one. + pub fn recipient(&self) -> Option { + match self { + StreamId::Channel { recipient, .. } | StreamId::Ack { recipient, .. } => { + Some(*recipient) + }, + StreamId::Broadcast { .. } | StreamId::Private { .. } => None, + } + } +} + +impl Encode for StreamId { + fn size_hint(&self) -> usize { + STREAM_ID_LEN + } + + fn encode_to(&self, dest: &mut T) { + match self { + StreamId::Channel { recipient, domain, num } => { + dest.push_byte(0x00); + dest.write(&u32::from(*recipient).to_be_bytes()); + dest.push_byte(*domain); + dest.write(&num.to_be_bytes()); + }, + StreamId::Ack { recipient, domain, num } => { + dest.push_byte(0x01); + dest.write(&u32::from(*recipient).to_be_bytes()); + dest.push_byte(*domain); + dest.write(&num.to_be_bytes()); + }, + StreamId::Broadcast { domain, subdomain, num } => { + dest.push_byte(0x02); + dest.write(&domain.to_be_bytes()); + dest.push_byte(*subdomain); + dest.write(&num.to_be_bytes()); + }, + StreamId::Private { kind, body } => { + dest.push_byte(*kind); + dest.write(body); + }, + } + } +} + +impl EncodeLike for StreamId {} +impl DecodeWithMemTracking for StreamId {} + +impl MaxEncodedLen for StreamId { + fn max_encoded_len() -> usize { + STREAM_ID_LEN + } +} + +impl Decode for StreamId { + fn decode(input: &mut I) -> Result { + let kind = input.read_byte()?; + match kind { + 0x00 | 0x01 => { + let mut r = [0u8; 4]; + input.read(&mut r)?; + let recipient = ParaId::from(u32::from_be_bytes(r)); + let domain = input.read_byte()?; + let mut n = [0u8; 2]; + input.read(&mut n)?; + let num = u16::from_be_bytes(n); + Ok(if kind == 0x00 { + StreamId::Channel { recipient, domain, num } + } else { + StreamId::Ack { recipient, domain, num } + }) + }, + 0x02 => { + let mut d = [0u8; 2]; + input.read(&mut d)?; + let domain = u16::from_be_bytes(d); + let subdomain = input.read_byte()?; + let mut n = [0u8; 4]; + input.read(&mut n)?; + let num = u32::from_be_bytes(n); + Ok(StreamId::Broadcast { domain, subdomain, num }) + }, + 0x80..=0xFF => { + let mut body = [0u8; 7]; + input.read(&mut body)?; + Ok(StreamId::Private { kind, body }) + }, + // 0x03..=0x7F are reserved for future standard kinds and REJECTED (canonicality). + _ => Err(CodecError::from("StreamId: reserved/unknown kind")), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn sample_ids() -> Vec { + vec![ + StreamId::Channel { recipient: ParaId::from(1000), domain: 0, num: 0 }, + StreamId::Channel { recipient: ParaId::from(1000), domain: 0, num: 1 }, + StreamId::Channel { recipient: ParaId::from(1000), domain: 1, num: 0 }, + StreamId::Channel { recipient: ParaId::from(2000), domain: 0, num: 0 }, + StreamId::Ack { recipient: ParaId::from(1000), domain: 0, num: 0 }, + StreamId::Broadcast { domain: 0, subdomain: 0, num: 0 }, + StreamId::Broadcast { domain: 1, subdomain: 0, num: 0 }, + StreamId::Private { kind: 0x80, body: [0; 7] }, + StreamId::Private { kind: 0xFF, body: [7; 7] }, + ] + } + + #[test] + fn encoding_layout_is_exact() { + // Frozen test vectors — any change here is a consensus break. + assert_eq!( + StreamId::Channel { recipient: ParaId::from(2000), domain: 0, num: 0 }.encode(), + vec![0x00, 0x00, 0x00, 0x07, 0xD0, 0x00, 0x00, 0x00], + ); + assert_eq!( + StreamId::Ack { recipient: ParaId::from(1000), domain: 0, num: 5 }.encode(), + vec![0x01, 0x00, 0x00, 0x03, 0xE8, 0x00, 0x00, 0x05], + ); + assert_eq!( + StreamId::Broadcast { domain: 0, subdomain: 0, num: 1 }.encode(), + vec![0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01], + ); + assert_eq!( + StreamId::Private { kind: 0x80, body: [1, 2, 3, 4, 5, 6, 7] }.encode(), + vec![0x80, 1, 2, 3, 4, 5, 6, 7], + ); + for id in sample_ids() { + assert_eq!( + id.encode().len(), + STREAM_ID_LEN, + "{id:?} must encode to {STREAM_ID_LEN} bytes" + ); + } + } + + #[test] + fn decode_round_trips() { + for id in sample_ids() { + assert_eq!(StreamId::decode(&mut &id.encode()[..]).unwrap(), id); + } + } + + #[test] + fn decode_rejects_reserved_kinds() { + for kind in 0x03u8..=0x7F { + let bytes = [kind, 0, 0, 0, 0, 0, 0, 0]; + assert!( + StreamId::decode(&mut &bytes[..]).is_err(), + "reserved kind {kind:#x} must reject" + ); + } + } + + #[test] + fn decode_rejects_truncated() { + let full = StreamId::Channel { recipient: ParaId::from(7), domain: 0, num: 0 }.encode(); + for len in 0..full.len() { + assert!(StreamId::decode(&mut &full[..len]).is_err(), "len {len} must reject"); + } + } + + #[test] + fn ord_equals_canonical_encoding_order() { + let mut by_ord = sample_ids(); + by_ord.sort(); + let mut by_bytes = sample_ids(); + by_bytes.sort_by_key(|id| id.encode()); + assert_eq!(by_ord, by_bytes, "Ord must match canonical-encoding lexicographic order"); + } + + #[test] + fn recipient_addressing() { + assert_eq!( + StreamId::Channel { recipient: ParaId::from(9), domain: 0, num: 0 }.recipient(), + Some(ParaId::from(9)), + ); + assert_eq!( + StreamId::Ack { recipient: ParaId::from(9), domain: 0, num: 0 }.recipient(), + Some(ParaId::from(9)), + ); + assert_eq!(StreamId::Broadcast { domain: 0, subdomain: 0, num: 0 }.recipient(), None); + assert_eq!(StreamId::Private { kind: 0x80, body: [0; 7] }.recipient(), None); + } +} diff --git a/cumulus/primitives/spec-messaging/src/streams_root.rs b/cumulus/primitives/spec-messaging/src/streams_root.rs new file mode 100644 index 000000000000..fc4bd68347bc --- /dev/null +++ b/cumulus/primitives/spec-messaging/src/streams_root.rs @@ -0,0 +1,388 @@ +//! `StreamsRoot` — the sender's per-block stream commitment (spec-msg v0.5). +//! +//! Each block, a sender commits ONE hash — the `StreamsRoot`, the root of a **keyed binary Patricia +//! trie** over `(StreamId, current stream root)` for every active stream. A receiver reads it (from +//! the sender's header digest, or `paras::Heads`) and proves the stream it consumes against it — +//! obtaining the sender's committed root for that stream without a sender-state proof. +//! +//! # Why a keyed trie (not a positional Merkle tree) +//! +//! The trie is keyed by the `StreamId`'s frozen 8-byte encoding, so a proof is a **walk driven by +//! the key**: verification recomputes the leaf from the caller's `StreamId` and checks each branch +//! direction against that key's bits. This is the property the requires-lift synthesis depends on — +//! lifts are matched *positionally* to the consumption record's streams, and a mispaired lift (a +//! proof built for stream A checked as stream B) cannot verify, because the walk binds the key. A +//! positional `binary_merkle_tree` (index + siblings) does not give this. Patricia compression +//! (branch only where keys diverge) also keeps proofs `O(log S)` in the number of streams, not +//! `O(64)` in the key width, and makes the tree stable under insertion (only the inserted path is +//! perturbed). +//! +//! Roles: +//! - **sender / node** builds the root ([`streams_root`]) and serves membership proofs +//! ([`gen_stream_proof`]); +//! - **receiver** reads the root ([`read_streams_root`]) and verifies membership +//! ([`verify_stream_membership`]). +//! +//! Hashing (domain-tagged, `blake2_256` via [`SpecHasher`]): +//! - leaf `= H(STREAMS_LEAF_TAG ++ key[8] ++ stream_root[32])` — binds the full key; +//! - inner `= H(STREAMS_INNER_TAG ++ split_bit[1] ++ left[32] ++ right[32])` — binds the branch +//! depth. + +use alloc::vec::Vec; +use codec::{Decode, DecodeWithMemTracking, Encode}; +use polkadot_core_primitives::Hash; +use sp_runtime::{traits::Hash as HashT, Digest, DigestItem}; + +use crate::{ + stream::StreamId, SpecHasher, SPMS_ENGINE_ID, STREAMS_INNER_TAG, STREAMS_LEAF_TAG, + STREAM_ID_LEN, +}; + +/// Bit width of a `StreamId` key (the Patricia trie's maximum depth). +const KEY_BITS: usize = STREAM_ID_LEN * 8; + +/// The sender's per-block stream commitment root — a newtype over `Hash`, deliberately +/// distinct from an MMR / stream root ("confusing roots must not typecheck"). Defined +/// relay-side and re-exported here so both sides share one type. +pub use polkadot_primitives::StreamsRoot; + +/// One step of a `StreamsRoot` trie walk, from an inner node toward the proven leaf. +#[derive( + Clone, Copy, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, scale_info::TypeInfo, +)] +pub struct TreeStep { + /// Key-bit index this node branches on (`0..KEY_BITS`, `0` = MSB of the key). Bound into the + /// inner-node hash, and checked against the proven key's bit during verification. + pub split_bit: u8, + /// Which side the proven key descends to at this node: `false` = left (bit 0), `true` = right + /// (bit 1). Must equal the key's bit at `split_bit`, else the proof is rejected. + pub target_right: bool, + /// Hash of the sibling subtree (the side the proven key does NOT descend to). + pub sibling: Hash, +} + +/// A membership proof that `(stream, stream_root)` is committed under a `StreamsRoot` — served by +/// the sender/node, verified by the receiver. The walk is keyed: it carries no leaf index, only the +/// branch steps from the root to the leaf (root-first), and the key drives verification. +#[derive( + Clone, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, scale_info::TypeInfo, +)] +pub struct StreamProof { + /// Branch steps from the root down to the leaf (root-first order). Empty for a single-stream + /// tree. + pub steps: Vec, +} + +/// Bit `i` of an 8-byte key, `0` = MSB of byte 0 (so big-endian byte order = bit order). +fn bit(key: &[u8; STREAM_ID_LEN], i: usize) -> bool { + (key[i >> 3] >> (7 - (i & 7))) & 1 == 1 +} + +/// The `StreamId`'s frozen 8-byte canonical encoding, as the trie key. +fn key_of(stream: &StreamId) -> [u8; STREAM_ID_LEN] { + let mut key = [0u8; STREAM_ID_LEN]; + // `StreamId::encode` is always exactly `STREAM_ID_LEN` bytes (frozen, consensus-critical). + key.copy_from_slice(&stream.encode()); + key +} + +/// First bit index (from the MSB) at which two distinct keys differ. Callers guarantee `a != b`. +fn first_diverging_bit(a: &[u8; STREAM_ID_LEN], b: &[u8; STREAM_ID_LEN]) -> usize { + (0..KEY_BITS).find(|&i| bit(a, i) != bit(b, i)).expect("keys are distinct; qed") +} + +/// `H(STREAMS_LEAF_TAG ++ key ++ stream_root)`. +fn hash_leaf(key: &[u8; STREAM_ID_LEN], stream_root: &Hash) -> Hash { + let mut pre = [0u8; 1 + STREAM_ID_LEN + 32]; + pre[0] = STREAMS_LEAF_TAG; + pre[1..1 + STREAM_ID_LEN].copy_from_slice(key); + pre[1 + STREAM_ID_LEN..].copy_from_slice(stream_root.as_bytes()); + ::hash(&pre) +} + +/// `H(STREAMS_INNER_TAG ++ split_bit ++ left ++ right)`. +fn hash_inner(split_bit: u8, left: &Hash, right: &Hash) -> Hash { + let mut pre = [0u8; 1 + 1 + 32 + 32]; + pre[0] = STREAMS_INNER_TAG; + pre[1] = split_bit; + pre[2..34].copy_from_slice(left.as_bytes()); + pre[34..].copy_from_slice(right.as_bytes()); + ::hash(&pre) +} + +/// Split a key-sorted, non-empty slice at its top-level branch: the bit where its first and last +/// keys diverge, with all `bit == 0` keys before all `bit == 1` keys. Returns `(split_bit, left, +/// right)`, both sides non-empty. Callers guarantee `entries.len() >= 2` and distinct keys. +fn split<'a>( + entries: &'a [([u8; STREAM_ID_LEN], Hash)], +) -> (usize, &'a [([u8; STREAM_ID_LEN], Hash)], &'a [([u8; STREAM_ID_LEN], Hash)]) { + let split_bit = first_diverging_bit(&entries[0].0, &entries[entries.len() - 1].0); + let p = entries.partition_point(|(k, _)| !bit(k, split_bit)); + let (left, right) = entries.split_at(p); + (split_bit, left, right) +} + +/// Trie root over a key-sorted, non-empty slice. +fn node_hash(entries: &[([u8; STREAM_ID_LEN], Hash)]) -> Hash { + if let [(key, root)] = entries { + return hash_leaf(key, root); + } + let (split_bit, left, right) = split(entries); + hash_inner(split_bit as u8, &node_hash(left), &node_hash(right)) +} + +/// Sort `entries` into canonical (key) order for trie construction. +fn sorted(mut entries: Vec<(StreamId, Hash)>) -> Vec<([u8; STREAM_ID_LEN], Hash)> { + entries.sort_by_key(|(stream, _)| *stream); + entries.into_iter().map(|(stream, root)| (key_of(&stream), root)).collect() +} + +/// The `StreamsRoot` over `(stream, stream_root)` entries, `None` when there are no active streams. +/// Entries need not be sorted or deduplicated by the caller for ordering, but keys MUST be unique +/// (the messaging inherent carries at most one item per stream). Sender/node side. +pub fn streams_root(entries: Vec<(StreamId, Hash)>) -> Option { + let entries = sorted(entries); + if entries.is_empty() { + return None; + } + Some(StreamsRoot(node_hash(&entries))) +} + +/// Walk `entries` (key-sorted) toward `key`, pushing each branch step (root-first) and returning +/// the subtree's root. The key is guaranteed present in `entries`. +fn prove( + entries: &[([u8; STREAM_ID_LEN], Hash)], + key: &[u8; STREAM_ID_LEN], + steps: &mut Vec, +) -> Hash { + if let [(k, root)] = entries { + return hash_leaf(k, root); + } + let (split_bit, left, right) = split(entries); + let target_right = bit(key, split_bit); + let (sibling, descend) = + if target_right { (node_hash(left), right) } else { (node_hash(right), left) }; + steps.push(TreeStep { split_bit: split_bit as u8, target_right, sibling }); + let child = prove(descend, key, steps); + if target_right { + hash_inner(split_bit as u8, &sibling, &child) + } else { + hash_inner(split_bit as u8, &child, &sibling) + } +} + +/// Build the streams trie over `entries` and return `(StreamsRoot, proof)` proving `stream`'s +/// membership, or `None` if `stream` is absent. Sender/node side; the receiver only runs +/// [`verify_stream_membership`]. +pub fn gen_stream_proof( + entries: Vec<(StreamId, Hash)>, + stream: StreamId, +) -> Option<(StreamsRoot, StreamProof)> { + let entries = sorted(entries); + let key = key_of(&stream); + if !entries.iter().any(|(k, _)| *k == key) { + return None; + } + let mut steps = Vec::new(); + let root = prove(&entries, &key, &mut steps); + Some((StreamsRoot(root), StreamProof { steps })) +} + +/// Read the sender's `StreamsRoot` from its header digest, if present. +pub fn read_streams_root(digest: &Digest) -> Option { + digest.logs().iter().find_map(|item| match item { + DigestItem::Consensus(id, payload) if *id == SPMS_ENGINE_ID => { + StreamsRoot::decode(&mut &payload[..]).ok() + }, + _ => None, + }) +} + +/// Fold a keyed membership proof for `(stream, stream_root)` up to the `StreamsRoot` it *implies*, +/// or `None` if a step is malformed (out-of-range branch, or a direction that contradicts the +/// proven key). Unlike [`verify_stream_membership`] this **yields** the root rather than comparing +/// it — the shape the requires-lift `tree_proof` needs (`compute_tree_root` in the design): the +/// walk is driven by `stream`'s key, so a proof built for a different stream folds to a different +/// root. +pub fn streams_root_from_proof( + stream: StreamId, + stream_root: Hash, + membership: &StreamProof, +) -> Option { + let key = key_of(&stream); + // A keyed Patricia trie over `KEY_BITS`-bit keys has depth ≤ `KEY_BITS`, so a valid proof + // carries at most `KEY_BITS` steps. Reject a longer (untrusted) proof up front — bounds the + // fold work regardless of how large a malformed step vec is (PoV size only caps it loosely). + if membership.steps.len() > KEY_BITS { + return None; + } + let mut node = hash_leaf(&key, &stream_root); + // Steps are root-first; fold leaf -> root. + for step in membership.steps.iter().rev() { + let split_bit = step.split_bit as usize; + // Reject an out-of-range branch (would otherwise index the key out of bounds) and any step + // whose direction contradicts the proven key — this is the key binding. + if split_bit >= KEY_BITS || bit(&key, split_bit) != step.target_right { + return None; + } + node = if step.target_right { + hash_inner(step.split_bit, &step.sibling, &node) + } else { + hash_inner(step.split_bit, &node, &step.sibling) + }; + } + Some(StreamsRoot(node)) +} + +/// Verify that `stream_root` is the sender's committed root for `stream`, against a `StreamsRoot` +/// (read from the sender's header via [`read_streams_root`]). Receiver side. +/// +/// The walk is keyed: the leaf is recomputed from `stream`, and each step's direction is checked +/// against `stream`'s bit at that branch — so a proof built for a different stream cannot verify. +pub fn verify_stream_membership( + streams_root: StreamsRoot, + stream: StreamId, + stream_root: Hash, + membership: &StreamProof, +) -> bool { + streams_root_from_proof(stream, stream_root, membership) == Some(streams_root) +} + +#[cfg(test)] +mod tests { + use super::*; + use sp_core::H256; + + fn ch(recipient: u32) -> StreamId { + StreamId::Channel { recipient: recipient.into(), domain: 0, num: 0 } + } + fn root(b: u8) -> Hash { + H256::repeat_byte(b) + } + + fn entries() -> Vec<(StreamId, Hash)> { + // deliberately unsorted — the primitives must canonicalize + vec![ + (ch(4000), root(0xB)), + (ch(2000), root(0xA)), + (ch(3000), root(0xC)), + (StreamId::Ack { recipient: 2000.into(), domain: 0, num: 0 }, root(0xD)), + (StreamId::Broadcast { domain: 1, subdomain: 0, num: 7 }, root(0xE)), + ] + } + + #[test] + fn streams_root_frozen_vector() { + // FROZEN consensus vector: this fixed entry set must always hash to this exact + // `StreamsRoot`. Any change to the trie node encoding — leaf/inner tags, `split_bit`, + // entry ordering, or the key binding — is a consensus break and breaks this test. + // (Round-trip tests can't catch a silent byte-format change; this pins the bytes.) + // Companion to `StreamId`'s frozen encoding. + assert_eq!( + streams_root(entries()).unwrap(), + StreamsRoot(H256(hex_literal::hex!( + "12de235fb6d96933d6ba770efe16f891c87fb8a88a75e37a384b88ba11dee0dc" + ))), + ); + } + + #[test] + fn round_trip_membership_verifies() { + for (stream, sr) in entries() { + let (r, proof) = gen_stream_proof(entries(), stream).unwrap(); + assert_eq!(Some(r), streams_root(entries())); + assert!(verify_stream_membership(r, stream, sr, &proof), "{stream:?} must verify"); + } + } + + #[test] + fn wrong_root_or_stream_or_value_fails() { + let (r, proof) = gen_stream_proof(entries(), ch(3000)).unwrap(); + // wrong stream_root for the (correct) stream + assert!(!verify_stream_membership(r, ch(3000), root(0xFF), &proof)); + // right value/proof but claimed for a different stream (key binding must reject) + assert!(!verify_stream_membership(r, ch(2000), root(0xC), &proof)); + // wrong streams root + assert!(!verify_stream_membership(StreamsRoot(root(0x00)), ch(3000), root(0xC), &proof)); + } + + #[test] + fn proof_is_not_transferable_between_streams() { + // A proof minted for one stream must not verify for any other stream, even with that + // other stream's own committed value — the keyed walk binds the StreamId. + let (r, proof) = gen_stream_proof(entries(), ch(4000)).unwrap(); + for (other, other_root) in entries() { + if other == ch(4000) { + continue; + } + assert!( + !verify_stream_membership(r, other, other_root, &proof), + "proof for ch(4000) must not verify as {other:?}" + ); + } + } + + #[test] + fn order_independent_root() { + let mut reversed = entries(); + reversed.reverse(); + assert_eq!(streams_root(entries()), streams_root(reversed)); + } + + #[test] + fn single_stream_tree_has_empty_proof() { + let one = vec![(ch(2000), root(0xA))]; + let (r, proof) = gen_stream_proof(one.clone(), ch(2000)).unwrap(); + assert!(proof.steps.is_empty()); + assert_eq!(Some(r), streams_root(one)); + assert!(verify_stream_membership(r, ch(2000), root(0xA), &proof)); + } + + #[test] + fn proof_is_logarithmic() { + // 64 clustered channels: a Patricia proof is O(log S), never the 64-bit key width. + let many: Vec<_> = (0..64u32).map(|i| (ch(2000 + i), root(i as u8))).collect(); + let (_r, proof) = gen_stream_proof(many, ch(2000 + 33)).unwrap(); + assert!(proof.steps.len() <= 7, "log2(64) = 6-ish, got {}", proof.steps.len()); + } + + #[test] + fn absent_stream_has_no_proof() { + assert!(gen_stream_proof(entries(), ch(9999)).is_none()); + } + + #[test] + fn out_of_range_split_bit_is_rejected() { + let (r, mut proof) = gen_stream_proof(entries(), ch(3000)).unwrap(); + proof + .steps + .push(TreeStep { split_bit: 64, target_right: false, sibling: root(0) }); + assert!(!verify_stream_membership(r, ch(3000), root(0xC), &proof)); + } + + #[test] + fn over_long_proof_is_rejected() { + let (r, _proof) = gen_stream_proof(entries(), ch(3000)).unwrap(); + // A valid proof can never exceed the trie depth (`KEY_BITS`); a padded one must reject. + let proof = StreamProof { + steps: vec![ + TreeStep { split_bit: 0, target_right: false, sibling: root(0) }; + KEY_BITS + 1 + ], + }; + assert!(!verify_stream_membership(r, ch(3000), root(0xC), &proof)); + } + + #[test] + fn read_streams_root_extracts_digest() { + let r = streams_root(entries()).unwrap(); + let digest = Digest { + logs: vec![ + DigestItem::Other(vec![1, 2, 3]), + DigestItem::Consensus(SPMS_ENGINE_ID, r.encode()), + ], + }; + assert_eq!(read_streams_root(&digest), Some(r)); + assert_eq!(read_streams_root(&Digest { logs: vec![] }), None); + } +} diff --git a/polkadot/primitives/src/lib.rs b/polkadot/primitives/src/lib.rs index 8570da3aeb2a..2e37e6197075 100644 --- a/polkadot/primitives/src/lib.rs +++ b/polkadot/primitives/src/lib.rs @@ -58,11 +58,11 @@ pub use v9::{ InherentData, InvalidDisputeStatementKind, Moment, MultiDisputeStatementSet, NodeFeatures, Nonce, OccupiedCore, OccupiedCoreAssumption, OutboundHrmpMessage, ParathreadClaim, ParathreadEntry, PersistedValidationData, PvfCheckStatement, PvfExecKind, PvfPrepKind, - RuntimeMetricLabel, RuntimeMetricLabelValue, RuntimeMetricLabelValues, RuntimeMetricLabels, - RuntimeMetricOp, RuntimeMetricUpdate, ScheduledCore, SchedulerParams, ScrapedOnChainVotes, - SessionIndex, SessionInfo, Signature, Signed, SignedAvailabilityBitfield, - SignedAvailabilityBitfields, SignedStatement, SigningContext, Slot, TransposedClaimQueue, - UMPSignal, UncheckedSigned, UncheckedSignedAvailabilityBitfield, + RequiresSet, RuntimeMetricLabel, RuntimeMetricLabelValue, RuntimeMetricLabelValues, + RuntimeMetricLabels, RuntimeMetricOp, RuntimeMetricUpdate, ScheduledCore, SchedulerParams, + ScrapedOnChainVotes, SessionIndex, SessionInfo, Signature, Signed, SignedAvailabilityBitfield, + SignedAvailabilityBitfields, SignedStatement, SigningContext, Slot, StreamsRoot, + TransposedClaimQueue, UMPSignal, UncheckedSigned, UncheckedSignedAvailabilityBitfield, UncheckedSignedAvailabilityBitfields, UncheckedSignedStatement, UpgradeGoAhead, UpgradeRestriction, UpwardMessage, ValidDisputeStatementKind, ValidationCode, ValidationCodeHash, ValidatorId, ValidatorIndex, ValidatorSignature, ValidityAttestation, diff --git a/polkadot/primitives/src/v9/mod.rs b/polkadot/primitives/src/v9/mod.rs index 7029cb9616cf..ae411ecec41f 100644 --- a/polkadot/primitives/src/v9/mod.rs +++ b/polkadot/primitives/src/v9/mod.rs @@ -74,6 +74,8 @@ pub use signed::{EncodeAs, Signed, UncheckedSigned}; pub mod async_backing; pub mod executor_params; pub mod slashing; +pub mod speculative; +pub use speculative::{CommitmentError, RequiresSet, StreamsRoot, MAX_SOURCES_PER_BLOCK}; pub use async_backing::AsyncBackingParams; pub use executor_params::{ @@ -1761,10 +1763,12 @@ pub mod node_features { CandidateReceiptV2 = 3, /// Enables support for scheduling information in the Candidate Descriptor. CandidateReceiptV3 = 4, + /// Enables speculative messaging + SpeculativeMessaging = 5, /// First unassigned feature bit. /// Every time a new feature flag is assigned it should take this value. /// and this should be incremented. - FirstUnassigned = 5, + FirstUnassigned = 6, } impl FeatureIndex { @@ -2716,6 +2720,13 @@ pub enum UMPSignal { SelectCore(CoreSelector, ClaimQueueOffset), /// A message sent by a parachain to promote the reputation of a given peerid. ApprovedPeer(ApprovedPeerId), + /// A speculative-messaging signal carrying this candidate's outgoing stream commitment root + /// (`Provides(StreamsRoot)`). The relay chain records these into its provides window. + Provides(StreamsRoot), + /// A speculative-messaging signal carrying this candidate's expected incoming per-source + /// stream commitment roots (`Requires(RequiresSet)`). The relay chain matches these against + /// the provides window of the corresponding source parachains. + Requires(RequiresSet), } /// The default claim queue offset to be used if it's not configured/accessible in the parachain @@ -2732,6 +2743,8 @@ pub type ApprovedPeerId = BoundedVec>; pub struct CandidateUMPSignals { pub(super) select_core: Option<(CoreSelector, ClaimQueueOffset)>, pub(super) approved_peer: Option, + pub(super) provides: Option, + pub(super) requires: Option, } impl CandidateUMPSignals { @@ -2745,9 +2758,22 @@ impl CandidateUMPSignals { self.approved_peer.as_ref() } + /// Get a reference to the speculative-messaging `provides` UMP signal. + pub fn provides(&self) -> Option<&StreamsRoot> { + self.provides.as_ref() + } + + /// Get a reference to the speculative-messaging `requires` UMP signal. + pub fn requires(&self) -> Option<&RequiresSet> { + self.requires.as_ref() + } + /// Returns `true` if UMP signals are empty. pub fn is_empty(&self) -> bool { - self.select_core.is_none() && self.approved_peer.is_none() + self.select_core.is_none() && + self.approved_peer.is_none() && + self.provides.is_none() && + self.requires.is_none() } fn try_decode_signal( @@ -2763,6 +2789,12 @@ impl CandidateUMPSignals { UMPSignal::SelectCore(core_selector, cq_offset) if self.select_core.is_none() => { self.select_core = Some((core_selector, cq_offset)); }, + UMPSignal::Provides(provides) if self.provides.is_none() => { + self.provides = Some(provides); + }, + UMPSignal::Requires(requires) if self.requires.is_none() => { + self.requires = Some(requires); + }, _ => { // This means that we got duplicate UMP signals. return Err(CommittedCandidateReceiptError::DuplicateUMPSignal); @@ -2778,13 +2810,17 @@ impl CandidateUMPSignals { select_core: Option<(CoreSelector, ClaimQueueOffset)>, approved_peer: Option, ) -> Self { - Self { select_core, approved_peer } + Self { select_core, approved_peer, provides: None, requires: None } } } /// Separator between `XCM` and `UMPSignal`. pub const UMP_SEPARATOR: Vec = vec![]; +/// The maximum number of `UMPSignal` messages a candidate may carry — one per +/// [`UMPSignal`] variant (`SelectCore`, `ApprovedPeer`, `Provides`, `Requires`). +pub const MAX_UMP_SIGNALS: usize = 4; + /// Utility function for skipping the ump signals. pub fn skip_ump_signals<'a>( upward_messages: impl Iterator>, @@ -2806,15 +2842,15 @@ impl CandidateCommitments { return Ok(res); } - // Process first signal - let Some(first_signal) = signals_iter.next() else { return Ok(res) }; - res.try_decode_signal(&mut first_signal.as_slice())?; - - // Process second signal - let Some(second_signal) = signals_iter.next() else { return Ok(res) }; - res.try_decode_signal(&mut second_signal.as_slice())?; + // At most one signal of each `UMPSignal` variant is allowed; `try_decode_signal` + // rejects duplicates. Bound the loop by the number of variants so a malformed + // candidate cannot force unbounded work. + for _ in 0..MAX_UMP_SIGNALS { + let Some(signal) = signals_iter.next() else { return Ok(res) }; + res.try_decode_signal(&mut signal.as_slice())?; + } - // At most two signals are allowed + // No more signals than there are distinct variants are allowed. if signals_iter.next().is_some() { return Err(CommittedCandidateReceiptError::TooManyUMPSignals); } @@ -3295,6 +3331,52 @@ pub mod tests { assert_eq!(info.last_rotation_at(), 15); } + #[test] + fn ump_signals_extracts_provides_and_requires_roots() { + let provides = StreamsRoot(Hash::repeat_byte(1)); + let requires = + RequiresSet::try_from_iter([(Id::from(2u32), StreamsRoot(Hash::repeat_byte(2)))]) + .unwrap(); + + let upward_messages = alloc::vec![ + UMP_SEPARATOR, + UMPSignal::Provides(provides.clone()).encode(), + UMPSignal::Requires(requires.clone()).encode(), + ]; + + let commitments = CandidateCommitments { + upward_messages: upward_messages.try_into().unwrap(), + ..Default::default() + }; + + let signals = commitments.ump_signals().expect("valid signals"); + assert_eq!(signals.provides(), Some(&provides)); + assert_eq!(signals.requires(), Some(&requires)); + } + + #[test] + fn ump_signals_rejects_more_than_max_signals() { + // One of every variant (== `MAX_UMP_SIGNALS`) plus one extra is rejected. + let upward_messages = alloc::vec![ + UMP_SEPARATOR, + UMPSignal::SelectCore(CoreSelector(0), ClaimQueueOffset(0)).encode(), + UMPSignal::ApprovedPeer(Default::default()).encode(), + UMPSignal::Provides(Default::default()).encode(), + UMPSignal::Requires(RequiresSet::default()).encode(), + UMPSignal::SelectCore(CoreSelector(1), ClaimQueueOffset(1)).encode(), + ]; + + let commitments = CandidateCommitments { + upward_messages: upward_messages.try_into().unwrap(), + ..Default::default() + }; + + assert_eq!( + commitments.ump_signals(), + Err(CommittedCandidateReceiptError::TooManyUMPSignals) + ); + } + #[test] fn group_for_core_is_core_for_group() { for cores in 1..=256 { diff --git a/polkadot/primitives/src/v9/speculative.rs b/polkadot/primitives/src/v9/speculative.rs new file mode 100644 index 000000000000..3702777a30f8 --- /dev/null +++ b/polkadot/primitives/src/v9/speculative.rs @@ -0,0 +1,266 @@ +// Copyright (C) Parity Technologies (UK) Ltd. +// This file is part of Polkadot. + +// Polkadot is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// Polkadot is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with Polkadot. If not, see . + +//! Relay-visible speculative-messaging commitments (Phase 1). +//! +//! The relay chain only sees the **commitments** carried in the `UMPSignal::Provides`/`Requires` +//! signals (inside `CandidateCommitments.upward_messages`): +//! - **`Provides(StreamsRoot)`** — one root per sender block, the root of the sender's keyed stream +//! commitment tree. Recorded into the relay's provides window. +//! - **`Requires(RequiresSet)`** — a receiver's expected `(source, StreamsRoot)` entries: a +//! canonical sorted set (strictly increasing `ParaId`, one entry per source), matched against +//! each source's window. +//! +//! These types live here (not in `cumulus-primitives-spec-messaging`) because the `UMPSignal` enum +//! is defined in `polkadot-primitives::v9` and the relay chain decodes them — putting them in the +//! parachain crate would force a `polkadot -> cumulus` dependency. The parachain off-chain / PoV +//! types and the MMR primitives that *build* these commitments live in that crate. + +use alloc::vec::Vec; +use polkadot_core_primitives::Hash; +use polkadot_parachain_primitives::primitives::Id as ParaId; +use sp_core::ConstU32; +use sp_runtime::BoundedVec; + +/// Maximum number of source parachains a receiver can consume from in one block. +/// Bounds the size of the `requires` commitment (one entry per source). +pub const MAX_SOURCES_PER_BLOCK: u32 = 128; + +/// Root of a sender's stream commitment tree: a binary compact (Patricia) trie keyed by the +/// canonical SCALE encoding of `StreamId` (8 bytes), leaves = the streams' MMR roots. +/// +/// A **newtype** over `Hash` (wire-transparent), deliberately *distinct* from an MMR / stream root +/// so the two kinds of root cannot be confused — they flow through different checks ("confusing +/// roots must not typecheck"; see the design's "Stream Commitment Tree"). The relay chain treats it +/// as opaque; the tree structure and inclusion proofs live in +/// `cumulus-primitives-spec-messaging::streams_root`, which re-exports this type. +#[derive( + Clone, + Copy, + codec::Encode, + codec::Decode, + codec::DecodeWithMemTracking, + codec::MaxEncodedLen, + Debug, + Default, + Eq, + PartialEq, + scale_info::TypeInfo, +)] +#[cfg_attr(feature = "std", derive(Hash))] +pub struct StreamsRoot(pub Hash); + +/// A receiver's `requires` commitment: a canonical, bounded set of `(ParaId, StreamsRoot)` entries +/// — one per **source** parachain (covering all its streams), sorted by strictly-increasing +/// `ParaId`. +/// +/// The order is enforced at decode so the encoding is canonical — collators, PVF, and the relay +/// chain all produce the same bytes for the same set. This is the design's `RequiresSet` (Appendix +/// D); the relay matches each `(source, StreamsRoot)` against that source's provides window. +#[derive( + Clone, codec::Encode, codec::MaxEncodedLen, Debug, Default, Eq, PartialEq, scale_info::TypeInfo, +)] +#[cfg_attr(feature = "std", derive(Hash))] +pub struct RequiresSet(BoundedVec<(ParaId, StreamsRoot), ConstU32>); + +/// Decode is manually implemented to enforce that `ParaId`s are strictly increasing (canonical +/// form). +impl codec::Decode for RequiresSet { + fn decode(input: &mut I) -> Result { + let inner = + BoundedVec::<(ParaId, StreamsRoot), ConstU32>::decode(input)?; + + for pair in inner.windows(2) { + if pair[0].0 >= pair[1].0 { + return Err(codec::Error::from( + "RequiresSet entries must be sorted by increasing ParaId", + )); + } + } + + Ok(Self(inner)) + } +} + +// Marker: the manual `Decode` (a sortedness check over `BoundedVec`) is mem-tracking safe. A derive +// does not compose with a manual `Decode`, so it is impl'd by hand. +impl codec::DecodeWithMemTracking for RequiresSet {} + +impl RequiresSet { + /// The committed `StreamsRoot` for `source`, if present (O(log n) lookup). + pub fn get(&self, source: ParaId) -> Option<&StreamsRoot> { + self.0 + .binary_search_by_key(&source, |(id, _)| *id) + .ok() + .map(|idx| &self.0[idx].1) + } + + /// The number of entries in the set. + pub fn len(&self) -> usize { + self.0.len() + } + + /// Whether the set has no entries. + pub fn is_empty(&self) -> bool { + self.0.is_empty() + } + + /// Iterate the `(source, StreamsRoot)` entries in ascending `ParaId` order. + pub fn iter(&self) -> impl Iterator { + self.0.iter() + } + + /// Build a [`RequiresSet`] from an arbitrary (possibly unordered) iterator, sorting by `ParaId` + /// to produce the canonical encoding. + pub fn try_from_iter( + it: impl IntoIterator, + ) -> Result { + let mut entries: Vec<(ParaId, StreamsRoot)> = it.into_iter().collect(); + entries.sort_by_key(|(source, _)| *source); + + if entries.windows(2).any(|w| w[0].0 == w[1].0) { + return Err(CommitmentError::DuplicateParaId); + } + + let inner = BoundedVec::try_from(entries).map_err(|_| CommitmentError::TooManyEntries)?; + + Ok(Self(inner)) + } +} + +impl<'a> IntoIterator for &'a RequiresSet { + type Item = &'a (ParaId, StreamsRoot); + type IntoIter = core::slice::Iter<'a, (ParaId, StreamsRoot)>; + + fn into_iter(self) -> Self::IntoIter { + self.0.iter() + } +} + +/// Errors that can occur when constructing a [`RequiresSet`]. +#[derive(Debug, PartialEq, Eq)] +pub enum CommitmentError { + /// The same source `ParaId` appears more than once. + DuplicateParaId, + /// More entries were provided than `MAX_SOURCES_PER_BLOCK` allows. + TooManyEntries, +} + +#[cfg(test)] +mod tests { + use super::*; + use codec::{Decode, Encode}; + + fn sr(byte: u8) -> StreamsRoot { + StreamsRoot(Hash::repeat_byte(byte)) + } + + /// `n` entries with distinct, strictly-increasing `ParaId`s. + fn many(n: u32) -> Vec<(ParaId, StreamsRoot)> { + (0..n).map(|i| (ParaId::from(i), sr(i as u8))).collect() + } + + #[test] + fn try_from_iter_sorts_entries() { + let set = RequiresSet::try_from_iter([ + (ParaId::from(3), sr(3)), + (ParaId::from(1), sr(1)), + (ParaId::from(2), sr(2)), + ]) + .unwrap(); + + let ids: Vec = set.iter().map(|(id, _)| *id).collect(); + assert_eq!(ids, vec![ParaId::from(1), ParaId::from(2), ParaId::from(3)]); + } + + #[test] + fn try_from_iter_rejects_duplicate_para_id() { + let result = + RequiresSet::try_from_iter([(ParaId::from(1), sr(1)), (ParaId::from(1), sr(2))]); + assert_eq!(result, Err(CommitmentError::DuplicateParaId)); + } + + #[test] + fn try_from_iter_rejects_too_many_entries() { + let result = RequiresSet::try_from_iter(many(MAX_SOURCES_PER_BLOCK + 1)); + assert_eq!(result, Err(CommitmentError::TooManyEntries)); + } + + #[test] + fn encode_decode_round_trip_works() { + let set = RequiresSet::try_from_iter([(ParaId::from(1), sr(1)), (ParaId::from(2), sr(2))]) + .unwrap(); + + let encoded = set.encode(); + let decoded = RequiresSet::decode(&mut &encoded[..]).unwrap(); + + assert_eq!(set, decoded); + } + + #[test] + fn decode_rejects_out_of_order_para_ids() { + let bad: Vec<(ParaId, StreamsRoot)> = + vec![(ParaId::from(2), sr(2)), (ParaId::from(1), sr(1))]; + let encoded = bad.encode(); + + assert!(RequiresSet::decode(&mut &encoded[..]).is_err()); + } + + #[test] + fn decode_rejects_duplicate_para_ids() { + let bad: Vec<(ParaId, StreamsRoot)> = + vec![(ParaId::from(1), sr(1)), (ParaId::from(1), sr(2))]; + let encoded = bad.encode(); + + assert!(RequiresSet::decode(&mut &encoded[..]).is_err()); + } + + #[test] + fn decode_rejects_too_many_entries() { + let encoded = many(MAX_SOURCES_PER_BLOCK + 1).encode(); + assert!(RequiresSet::decode(&mut &encoded[..]).is_err()); + } + + #[test] + fn get_finds_existing_and_missing_entries() { + let set = RequiresSet::try_from_iter([(ParaId::from(1), sr(1)), (ParaId::from(3), sr(3))]) + .unwrap(); + + assert_eq!(set.get(ParaId::from(1)), Some(&sr(1))); + assert_eq!(set.get(ParaId::from(3)), Some(&sr(3))); + assert_eq!(set.get(ParaId::from(2)), None); + } + + #[test] + fn len_and_is_empty() { + let empty = RequiresSet::default(); + assert!(empty.is_empty()); + assert_eq!(empty.len(), 0); + + let set = RequiresSet::try_from_iter([(ParaId::from(1), sr(1))]).unwrap(); + assert!(!set.is_empty()); + assert_eq!(set.len(), 1); + } + + #[test] + fn into_iter_yields_sorted_entries() { + let set = RequiresSet::try_from_iter([(ParaId::from(2), sr(2)), (ParaId::from(1), sr(1))]) + .unwrap(); + + let collected: Vec<_> = (&set).into_iter().collect(); + assert_eq!(collected, vec![&(ParaId::from(1), sr(1)), &(ParaId::from(2), sr(2))]); + } +}