[stable2603] Backport #11139 (#11166)

Backport #11139 into `stable2603` from alexggh.

See the
[documentation](https://github.com/paritytech/polkadot-sdk/blob/master/docs/BACKPORT.md)
on how to use this bot.

<!--
  # To be used by other automation, do not modify:
  original-pr-number: #${pull_number}
-->

Signed-off-by: Alexandru Gheorghe <alexandru.gheorghe@parity.io>
Co-authored-by: Alexandru Gheorghe <49718502+alexggh@users.noreply.github.com>

Author: paritytech-release-backport-bot[bot]

cumulus/zombienet/zombienet-sdk/tests/zombie_ci/statement_store.rs

@@ -7,7 +7,7 @@
 use std::time::Duration;
 
 use sp_core::{Bytes, Encode};
-use sp_statement_store::{SubmitResult, Topic, TopicFilter};
+use sp_statement_store::{StatementEvent, SubmitResult, Topic, TopicFilter};
 use zombienet_sdk::subxt::ext::subxt_rpcs::rpc_params;
 
 use crate::zombie_ci::statement_store_bench::{get_keypair, spawn_network};
@@ -39,7 +39,7 @@ async fn statement_store() -> Result<(), anyhow::Error> {
 	// Subscribe to statements with topic "topic" to dave.
 	let stop_after_secs = 20;
 	let mut subscription = dave_rpc
-		.subscribe::<Bytes>(
+		.subscribe::<StatementEvent>(
 			"statement_subscribeStatement",
 			rpc_params![TopicFilter::MatchAll(vec![topic].try_into().expect("Single topic"))],
 			"statement_unsubscribeStatement",
@@ -50,14 +50,27 @@ async fn statement_store() -> Result<(), anyhow::Error> {
 	let _: SubmitResult =
 		charlie_rpc.request("statement_submit", rpc_params![statement.clone()]).await?;
 
-	let statement_bytes =
-		tokio::time::timeout(Duration::from_secs(stop_after_secs), subscription.next())
-			.await
-			.expect("Should not timeout")
-			.expect("Should receive")
-			.expect("Should not error");
+	loop {
+		let subscribe_item =
+			tokio::time::timeout(Duration::from_secs(stop_after_secs), subscription.next())
+				.await
+				.expect("Should not timeout")
+				.expect("Should receive")
+				.expect("Should not error");
 
-	assert_eq!(statement_bytes, statement);
+		let statement_bytes = match subscribe_item {
+			StatementEvent::NewStatements { statements: mut batch, .. } => {
+				if batch.is_empty() {
+					continue;
+				}
+				assert_eq!(batch.len(), 1, "Expected exactly one statement in batch");
+				batch.remove(0)
+			},
+		};
+		assert_eq!(statement_bytes, statement);
+
+		break;
+	}
 	// Now make sure no more statements are received.
 	assert!(tokio::time::timeout(Duration::from_secs(stop_after_secs), subscription.next())
 		.await

cumulus/zombienet/zombienet-sdk/tests/zombie_ci/statement_store_bench.rs

@@ -10,8 +10,8 @@ use log::{debug, info, trace};
 use sc_statement_store::{DEFAULT_MAX_TOTAL_SIZE, DEFAULT_MAX_TOTAL_STATEMENTS};
 use sp_core::{blake2_256, hexdisplay::HexDisplay, sr25519, Bytes, Pair};
 use sp_statement_store::{
-	statement_allowance_key, Channel, Statement, StatementAllowance, SubmitResult, Topic,
-	TopicFilter,
+	statement_allowance_key, Channel, Statement, StatementAllowance, StatementEvent, SubmitResult,
+	Topic, TopicFilter,
 };
 use std::{
 	cell::Cell,
@@ -614,7 +614,7 @@ impl Participant {
 		for idx in &pending {
 			let subscription = self
 				.rpc_client
-				.subscribe::<Bytes>(
+				.subscribe::<StatementEvent>(
 					"statement_subscribeStatement",
 					rpc_params![TopicFilter::MatchAll(
 						vec![topic_public_key(), topic_idx(*idx)].try_into().expect("Two topics")
@@ -628,13 +628,28 @@ impl Participant {
 		let mut futures: FuturesUnordered<_> = subscriptions
 			.into_iter()
 			.map(|(idx, mut subscription)| async move {
-				let statement_bytes =
-					timeout(Duration::from_secs(SUBSCRIBE_TIMEOUT_SECS), subscription.next())
-						.await
-						.map_err(|_| anyhow!("Timeout waiting for session key"))?
-						.ok_or_else(|| anyhow!("Subscription ended unexpectedly"))?
-						.map_err(|e| anyhow!("Subscription error: {}", e))?;
-				let statement = Statement::decode(&mut &statement_bytes[..])
+				let mut batch;
+				loop {
+					let item =
+						timeout(Duration::from_secs(SUBSCRIBE_TIMEOUT_SECS), subscription.next())
+							.await
+							.map_err(|_| anyhow!("Timeout waiting for session key"))?
+							.ok_or_else(|| anyhow!("Subscription ended unexpectedly"))?
+							.map_err(|e| anyhow!("Subscription error: {}", e))?;
+					let StatementEvent::NewStatements { statements, .. } = item;
+					if statements.is_empty() {
+						continue; // Ignore empty batches
+					} else {
+						batch = statements;
+						break;
+					}
+				}
+
+				if batch.len() != 1 {
+					return Err(anyhow!("Expected exactly one statement, got: {}", batch.len()));
+				}
+
+				let statement = Statement::decode(&mut &batch.remove(0)[..])
 					.map_err(|e| anyhow!("Failed to decode statement: {}", e))?;
 				let data = statement.data().ok_or_else(|| anyhow!("Statement missing data"))?;
 				let session_key = sr25519::Public::from_raw(
@@ -686,7 +701,7 @@ impl Participant {
 		for &(sender_idx, sender_session_key) in &pending {
 			let subscription = self
 				.rpc_client
-				.subscribe::<Bytes>(
+				.subscribe::<StatementEvent>(
 					"statement_subscribeStatement",
 					rpc_params![TopicFilter::MatchAll(
 						vec![topic_message(), topic_pair(&sender_session_key, &own_session_key)]
@@ -702,13 +717,27 @@ impl Participant {
 		let mut futures: FuturesUnordered<_> = subscriptions
 			.into_iter()
 			.map(|(sender_idx, mut subscription)| async move {
-				let statement_bytes =
-					timeout(Duration::from_secs(SUBSCRIBE_TIMEOUT_SECS), subscription.next())
-						.await
-						.map_err(|_| anyhow!("Timeout waiting for message"))?
-						.ok_or_else(|| anyhow!("Subscription ended unexpectedly"))?
-						.map_err(|e| anyhow!("Subscription error: {}", e))?;
-				let statement = Statement::decode(&mut &statement_bytes[..])
+				let mut batch;
+				loop {
+					let item =
+						timeout(Duration::from_secs(SUBSCRIBE_TIMEOUT_SECS), subscription.next())
+							.await
+							.map_err(|_| anyhow!("Timeout waiting for message"))?
+							.ok_or_else(|| anyhow!("Subscription ended unexpectedly"))?
+							.map_err(|e| anyhow!("Subscription error: {}", e))?;
+					batch = match item {
+						StatementEvent::NewStatements { statements: batch, .. } => batch,
+					};
+					if batch.is_empty() {
+						continue; // Ignore empty batches
+					} else {
+						break;
+					}
+				}
+				if batch.len() != 1 {
+					return Err(anyhow!("Expected exactly one statement, got: {}", batch.len()));
+				}
+				let statement = Statement::decode(&mut &batch.remove(0)[..])
 					.map_err(|e| anyhow!("Failed to decode statement: {}", e))?;
 				let data = statement.data().ok_or_else(|| anyhow!("Statement missing data"))?;
 				let req = StatementMessage::decode(&mut &data[..])
@@ -929,7 +958,7 @@ async fn statement_store_latency_bench() -> Result<(), anyhow::Error> {
 						let topic: Topic = blake2_256(topic_str.as_bytes()).into();
 
 						let subscription = rpc_client
-							.subscribe::<Bytes>(
+							.subscribe::<StatementEvent>(
 								"statement_subscribeStatement",
 								rpc_params![TopicFilter::MatchAll(
 									vec![topic].try_into().expect("Single topic")
@@ -1008,7 +1037,9 @@ async fn statement_store_latency_bench() -> Result<(), anyhow::Error> {
 						.into_iter()
 						.map(|(msg_idx, topic_str, mut subscription)| async move {
 							match timeout(total_timeout, subscription.next()).await {
-								Ok(Some(Ok(_statement_bytes))) => Ok((msg_idx, topic_str)),
+								Ok(Some(Ok(StatementEvent::NewStatements { .. }))) => {
+									Ok((msg_idx, topic_str))
+								},
 								Ok(Some(Err(e))) => Err(anyhow!(
 									"Subscription error for message {}: {}",
 									msg_idx,

prdoc/pr_11139.prdoc

@@ -0,0 +1,18 @@
+title: 'make subscription return statement event instead of bytes'
+doc:
+- audience: Node Dev
+  description: |
+    Changes the statement subscription RPC to return `StatementEvent` instead of raw `Bytes`.
+    When a subscription is initiated, the endpoint now sends NewStatements event batches
+    with matching statements already in the store. If there are not statements in the store
+    an empty batch is sent.
+
+crates:
+- name: sp-statement-store
+  bump: minor
+- name: sc-statement-store
+  bump: minor
+- name: sc-rpc
+  bump: minor
+- name: sc-rpc-api
+  bump: minor

substrate/client/rpc-api/src/statement/mod.rs

@@ -20,7 +20,7 @@
 
 use jsonrpsee::{core::RpcResult, proc_macros::rpc};
 use sp_core::Bytes;
-use sp_statement_store::{SubmitResult, TopicFilter};
+use sp_statement_store::{StatementEvent, SubmitResult, TopicFilter};
 
 pub mod error;
 
@@ -37,14 +37,22 @@ pub trait StatementApi {
 	///
 	/// # Returns
 	///
-	/// Returns a stream of SCALE-encoded statements as `Bytes`.
-	/// When a subscription is initiated the endpoint will immediately return the matching
-	/// statements already in the store. Subsequent matching statements will be pushed to the client
-	/// as they are added to the store.
+	/// Returns a stream of `StatementEvent` values.
+	/// When a subscription is initiated the endpoint will first return all matching statements
+	/// already in the store in batches as `StatementEvent::NewStatements`.
+	///
+	/// NewStatements includes an Optional field `remaining` which indicates how many more
+	/// statements are left to be sent in the initial batch of existing statements. The field
+	/// guarantees to the client that it will receive at least this many more statements in the
+	/// subscription stream, but it may receive more if new statements are added to the store that
+	/// match the filter.
+	///
+	///  If there are no statements in the store matching the filter, an empty batch of statements
+	/// is sent.
 	#[subscription(
 		name = "statement_subscribeStatement" => "statement_statement",
 		unsubscribe = "statement_unsubscribeStatement",
-		item = Bytes,
+		item = StatementEvent,
 		with_extensions,
 	)]
 	fn subscribe_statement(&self, topic_filter: TopicFilter);

substrate/client/rpc/Cargo.toml

@@ -16,6 +16,7 @@ workspace = true
 targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
+async-channel = { workspace = true }
 codec = { workspace = true, default-features = true }
 futures = { workspace = true }
 jsonrpsee = { features = ["server"], workspace = true }
@@ -45,7 +46,6 @@ tokio = { workspace = true, default-features = true }
 
 [dev-dependencies]
 assert_matches = { workspace = true }
-async-channel = { workspace = true }
 futures-timer = { workspace = true }
 pretty_assertions = { workspace = true }
 sc-keystore = { workspace = true }

substrate/client/rpc/src/statement/mod.rs

@@ -27,9 +27,17 @@ use jsonrpsee::{
 /// Re-export the API for backward compatibility.
 pub use sc_rpc_api::statement::{error::Error, StatementApiServer};
 use sp_core::Bytes;
-use sp_statement_store::{OptimizedTopicFilter, StatementSource, SubmitResult, TopicFilter};
+use sp_statement_store::{
+	OptimizedTopicFilter, StatementEvent, StatementSource, SubmitResult, TopicFilter,
+};
 use std::sync::Arc;
 const LOG_TARGET: &str = "statement-store-rpc";
+// The maximum size of a chunk of statements to send in a single JSON response. This is needed to
+// avoid hitting the maximum JSON size limit in the RPC response. Each statement is SCALE-encoded
+// and then hex-encoded in the JSON response, so the size of the JSON response is approximately 2x.
+// This value is chosen to be large enough to send a reasonable number of statements in a single
+// chunk, but small enough to avoid hitting the JSON size limit.
+const MAX_CHUNK_BYTES_LIMIT: usize = 4 * 1024 * 1024;
 
 use crate::{
 	utils::{spawn_subscription_task, BoundedVecDeque, PendingSubscription},
@@ -39,6 +47,56 @@ use crate::{
 #[cfg(test)]
 mod tests;
 
+/// Send existing statements in chunks over the subscription channel.
+///
+/// Splits the statements into chunks that fit within [`MAX_CHUNK_BYTES_LIMIT`] to avoid
+/// exceeding the RPC max response size, then sends each chunk as a
+/// [`StatementEvent::NewStatements`].
+async fn send_in_chunks(
+	existing_statements: Vec<Vec<u8>>,
+	subscription_sender: async_channel::Sender<StatementEvent>,
+) {
+	let mut iter = existing_statements.into_iter().peekable();
+	loop {
+		let mut chunk = Vec::<Bytes>::new();
+		let mut chunk_json_size = 0usize;
+		while let Some(statement) = iter.peek() {
+			// Each SCALE-encoded byte becomes 2 hex chars in the JSON response
+			let json_size_estimate = statement.len() * 2;
+			// If a single statement exceeds the max chunk size, skip it but continue sending the
+			// rest of the statements. This would never happen in practice because the statement
+			// store should reject statements that are too large, but we add this check to be safe.
+			if json_size_estimate > MAX_CHUNK_BYTES_LIMIT {
+				iter.next();
+				continue;
+			}
+			if chunk_json_size + json_size_estimate > MAX_CHUNK_BYTES_LIMIT {
+				break;
+			}
+			let Some(statement) = iter.next() else { break };
+			chunk_json_size += json_size_estimate;
+			chunk.push(statement.into());
+		}
+		if chunk.is_empty() {
+			break;
+		}
+		let remaining = iter.len();
+		if let Err(e) = subscription_sender
+			.send(StatementEvent::NewStatements {
+				statements: chunk,
+				remaining: Some(remaining as u32),
+			})
+			.await
+		{
+			log::warn!(
+				target: LOG_TARGET,
+				"Failed to send existing statement in subscription: {:?}", e
+			);
+			break;
+		}
+	}
+}
+
 /// Trait alias for statement store API required by the RPC.
 pub trait StatementStoreApi:
 	sp_statement_store::StatementStore + sc_statement_store::StatementStoreSubscriptionApi
@@ -103,22 +161,10 @@ impl StatementApiServer for StatementStore {
 				.pipe_from_stream(subscription_stream, BoundedVecDeque::new(128)),
 		);
 
-		// Send existing statements before returning, to make sure we did not miss any statements.
 		self.executor.spawn(
 			"statement-store-rpc-send",
 			Some("rpc"),
-			async move {
-				for statement in existing_statements {
-					if let Err(e) = subscription_sender.send(statement.into()).await {
-						log::warn!(
-							target: LOG_TARGET,
-							"Failed to send existing statement in subscription: {:?}", e
-						);
-						break;
-					}
-				}
-			}
-			.boxed(),
+			send_in_chunks(existing_statements, subscription_sender).boxed(),
 		)
 	}
 }