statement-store: implement new rpc api #10690

alexggh · 2025-12-18T13:16:54Z

Work in progress

Description

Implements the proposed changes from here #10452, where the API surface has been limited to two functions submit and a subscribe_statement.

Submit changes

For submitting the only changes are implementing the logic of the newly expiry field where statements with an expiration timestamp lower than current time get rejected.

Subscribe changes

For subscription the approach is to have a configurable number of workers that keep track of all subscriptions, each time a new subscription arrives it gets assigned by a round-robin protocol to one of the workers.

When a new statements gets accepted by the statement-store it notifies all workers about the statement, the workers then go and evaluate all subscription filters they are assigned to and notifies each of them accordingly.

Remaining work

Add more tests.
Evaluate performance under load and port statement_store_bench.rs to using subscribe API, ported and performance stay the same on the benchmark, because bottlenecks are elsewhere and being addressed.

[2026-01-07T17:18:51Z INFO  tests::zombie_ci::statement_store_bench] Participants: 49998, each sent: 6, received: 0
[2026-01-07T17:18:51Z INFO  tests::zombie_ci::statement_store_bench] Summary        min       avg       max
[2026-01-07T17:18:51Z INFO  tests::zombie_ci::statement_store_bench]  time, s         8        19        24
[2026-01-07T17:18:51Z INFO  tests::zombie_ci::statement_store_bench]  retries         0         0         0

Evaluate and set proper RPC limits. The normal RPC limits apply here we just need to make sure RPC nodes run with limits high enough, so each user can connect to the node all at the same time.
Periodically scan existing statements and removed expired ones.
Plan for deploying it with minimising impact for people prototype with the old APIs
- This modifies statement format(expiry field), Runtime and Node need to run latest version.

Local testing

To obtain a local node with the new APIs available run:

cargo build -p polkadot -p polkadot-parachain-bin  --bin polkadot --bin polkadot-prepare-worker --bin polkadot-execute-worker --bin polkadot-parachain  --features fast-runtime;
 
ZOMBIE_PROVIDER=native cargo test -p cumulus-zombienet-sdk-tests --features zombie-ci zombie_ci::statement_store::statement_store

The rpc addresses are visible in the output:

charlie: direct link (pjs) https://polkadot.js.org/apps/?rpc=ws://127.0.0.1:26997#/explorer
dave: direct link (pjs) https://polkadot.js.org/apps/?rpc=ws://127.0.0.1:12623#/explorer

…e_dev --bump major'

Co-authored-by: Alexandru Gheorghe <[email protected]>

Signed-off-by: Alexandru Gheorghe <[email protected]>

…ne_shot_the_api

…dation

Signed-off-by: Alexandru Gheorghe <[email protected]>

…dation

Signed-off-by: Alexandru Gheorghe <[email protected]>

AndreiEres

Great work, Alex!
Looks good, left a few questions.

cumulus/parachains/runtimes/people/people-westend/src/lib.rs

substrate/primitives/statement-store/src/lib.rs

substrate/primitives/statement-store/src/store_api.rs

substrate/bin/node/rpc/src/lib.rs

substrate/client/statement-store/src/lib.rs

substrate/client/statement-store/src/subscription.rs

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

s0me0ne-unkn0wn

I'm just skimming it for now. The API itself looks good and quite intuitive. I'm yet to understand the mechanics happening inside.

cumulus/parachains/runtimes/people/people-westend/src/lib.rs

P1sar · 2026-01-14T12:51:24Z

Closes: #10568, #10457

Signed-off-by: Alexandru Gheorghe <[email protected]>

AndreiEres · 2026-01-19T15:09:51Z

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

+	log::info!("Statement store test passed");
+	log::info!("Keeping network alive");

+	tokio::time::sleep(Duration::from_hours(24)).await;


why we need hang the test for 24h?

It is there for App guys to have a node setup they can run locally, will remove it before merge.

substrate/primitives/statement-store/src/store_api.rs

substrate/client/statement-store/src/lib.rs

…tore-not-api3

Signed-off-by: Alexandru Gheorghe <[email protected]>

P1sar · 2026-01-22T12:01:32Z

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

+	fn subscribe_statement(
+		&self,
+		pending: PendingSubscriptionSink,
+		_ext: &Extensions,


Would it make sense to reuse existing subscription routine if sessionID or/and address with topic filter is the same?
Otherwise malicious actor can create unlimited amount of the same subscriptions

Both the number of connections and the number of subscriptions per connection accepted by a node are already limited and configurable with the following cli arguments--rpc-max-subscriptions-per-connection and --rpc-max-connection .

I think that covers your attack scenario or am I missing something ?

rpc-max-subscriptions-per-connection caps the number of subscriptions per connection, which helps protect the network stack. But I’m still wondering whether this can be used as a DDoS vector against a node at the application layer.

The default/typical limit is quite high (e.g., ~100k), and in our case the bottleneck isn’t the number of WebSocket connections, it’s the subscription handling itself. A single client/browser can still legitimately create ~100k statement subscriptions using the same topic, which can overwhelm our logical workers / subscription-processing pipeline.

So the concern here isn’t connection count, but application logic load: one connection can create an excessive number of subscriptions and effectively exhaust CPU/memory/worker capacity

The default/typical limit is quite high (e.g., ~100k),

Not sure I understand where the 100k is coming, the current polkadot-parachain configuration is 1024:

--rpc-max-subscriptions-per-connection <RPC_MAX_SUBSCRIPTIONS_PER_CONNECTION> Set the maximum concurrent subscriptions per connection [default: 1024]

Which is not the high, if server wants to protect itself it can set that to a lower numbers to reduce the risk, my thinking is that this is a generic mechanism that already exists, so I don't think we should implement one just for our subscription API.

But I am not talking about rate limit. My assumption is to reuse subscription routines if it is subscribed for the same topics. eg if 100k clients subscribed for topicA we can have only one routine that serves all all 100k subscriptions.

But I am not talking about rate limit. My assumption is to reuse subscription routines if it is subscribed for the same topics. eg if 100k clients subscribed for topicA we can have only one routine that serves all all 100k subscriptions.

Ok, this task here is just for pushing the data towards the subscription stream, it uses the common pattern I saw in polkadot-sdk, as far as I can tell they are really cheap. The main grunt of work to determine if a statement matches filters for subscriptions is done in the subscription workers here https://github.com/paritytech/polkadot-sdk/pull/10690/files#diff-c3deb7e86e76f267c5ac9191aa895ddcd46f7807ae858e84592ad3d3c6c169b3R92 and those are limited in size.

The problem is that most of our other RPCs are already easy dosable. At some point we had some working on this, but I think this stopped.

bkchr · 2026-01-23T14:11:57Z

substrate/primitives/statement-store/src/store_api.rs

+/// Topic filter for statement subscriptions.
+#[derive(Clone, Debug)]
+pub enum CheckedTopicFilter {
+	/// Matches all topics.
+	Any,
+	/// Matches only statements including all of the given topics.
+	/// Bytes are expected to be a 32-byte topic. Up to `4` topics can be provided.
+	MatchAll(HashSet<Topic>),
+	/// Matches statements including any of the given topics.
+	/// Bytes are expected to be a 32-byte topic. Up to `128` topics can be provided.
+	MatchAny(HashSet<Topic>),
+}


Why don't we directly use BoundedVec<Topic> above? Then we don't need to duplicate this.

The pre-processing into a HashSet is still needed for the match function, so even with using BoundedVec, we would still need a different structure, I would just save the length check.

But, I'm a bit reluctant to change the RPC API parameter at this point, since I know @ERussel is already working on using the new API.

- MatchAll(Vec<Bytes>), + MatchAll(BoundedVec<Bytes, ConstU32<{ MAX_TOPICS as u32 }>>), - MatchAny(Vec<Bytes>), + MatchAny(BoundedVec<Bytes, ConstU32<{ MAX_ANY_TOPICS as u32 }>>)

@ERussel how much friction changes like this could create on your end ?

I don't see any problems on the client side with this improvement

Done, updated to use BoundedVec, I couldn't remove the CheckedTopicFilter, because it does some other validations as well.

@alexggh what other valdations does it do? I just see the check if topic is 32 bytes. But Topic being an array of 32 bytes, it should only accept 32 bytes at decoding and not less or more?

substrate/primitives/statement-store/src/lib.rs

bkchr · 2026-01-23T14:21:34Z

substrate/primitives/statement-store/src/lib.rs

 	}

 	/// Returns decryption key if any.
+	#[allow(deprecated)]


I don't think anyone is using these methods and it can be directly removed.

We discussed this internally, since we are not sure about the future of this field, we aggreed to keep the field and implementation around.

substrate/primitives/statement-store/src/lib.rs

substrate/client/statement-store/src/subscription.rs

substrate/client/statement-store/src/lib.rs

bkchr · 2026-01-23T15:39:56Z

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

+	fn subscribe_statement(
+		&self,
+		pending: PendingSubscriptionSink,
+		_ext: &Extensions,


The problem is that most of our other RPCs are already easy dosable. At some point we had some working on this, but I think this stopped.

Signed-off-by: Alexandru Gheorghe <[email protected]>

…tore-not-api4

Signed-off-by: Alexandru Gheorghe <[email protected]>

paritytech-workflow-stopper · 2026-01-29T16:54:20Z

All GitHub workflows were cancelled due to failure one of the required jobs.
Failed workflow url: https://github.com/paritytech/polkadot-sdk/actions/runs/21486888278
Failed job name: fmt

Signed-off-by: Alexandru Gheorghe <[email protected]>

bkchr · 2026-01-30T14:25:05Z

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

@@ -0,0 +1,166 @@
+use super::*;


missing license header.

bkchr · 2026-01-30T14:45:09Z

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

+		spawn_subscription_task(&self.executor, async {
+			PendingSubscription::from(pending)
+				.pipe_from_stream(subscription_stream, BoundedVecDeque::new(128))
+				.await;
+		});


Suggested change

spawn_subscription_task(&self.executor, async {

PendingSubscription::from(pending)

.pipe_from_stream(subscription_stream, BoundedVecDeque::new(128))

.await;

});

spawn_subscription_task(&self.executor,

PendingSubscription::from(pending)

.pipe_from_stream(subscription_stream, BoundedVecDeque::new(128))

);

bkchr · 2026-01-30T14:52:25Z

substrate/primitives/statement-store/src/store_api.rs

+/// Topic filter for statement subscriptions.
+#[derive(Clone, Debug)]
+pub enum CheckedTopicFilter {
+	/// Matches all topics.
+	Any,
+	/// Matches only statements including all of the given topics.
+	/// Bytes are expected to be a 32-byte topic. Up to `4` topics can be provided.
+	MatchAll(HashSet<Topic>),
+	/// Matches statements including any of the given topics.
+	/// Bytes are expected to be a 32-byte topic. Up to `128` topics can be provided.
+	MatchAny(HashSet<Topic>),
+}


@alexggh what other valdations does it do? I just see the check if topic is 32 bytes. But Topic being an array of 32 bytes, it should only accept 32 bytes at decoding and not less or more?

bkchr · 2026-01-30T15:31:05Z

substrate/client/statement-store/src/lib.rs

+		mut f: impl FnMut(&Hash) -> Result<()>,
+	) -> Result<()> {
+		let key_set = self.by_dec_key.get(&key);
+		if key_set.map_or(0, |s| s.len()) == 0 {


Suggested change

if key_set.map_or(0, |s| s.len()) == 0 {

if key_set.map_or(true, |s| s.is_empty()) {

bkchr · 2026-01-30T15:36:57Z

substrate/client/statement-store/src/lib.rs

+		let key_set = self.by_dec_key.get(&key);
+		if key_set.map_or(0, |s| s.len()) == 0 {
+			// Key does not exist in the index.
+			return Ok(())
+		}


Suggested change

let key_set = self.by_dec_key.get(&key);

if key_set.map_or(0, |s| s.len()) == 0 {

// Key does not exist in the index.

return Ok(())

}

let Some(key_set) = self.by_dec_key.get(&key).filter(|k| !k.is_empty()) else { return; };

bkchr · 2026-01-30T16:38:37Z

substrate/client/statement-store/src/subscription.rs

+		for _ in 0..num_matcher_workers {
+			let (subscription_matcher_sender, subscription_matcher_receiver) =
+				async_channel::bounded(MATCHERS_TASK_CHANNEL_BUFFER_SIZE);
+			subscriptions_matchers_senders.push(subscription_matcher_sender);
+			task_spawner.spawn_blocking(
+				"statement-store-subscription-filters",
+				Some("statement-store"),
+				Box::pin(async move {
+					let mut subscriptions = SubscriptionsInfo::new();
+					log::debug!(
+						target: LOG_TARGET,
+						"Started statement subscription matcher task"


Suggested change

for _ in 0..num_matcher_workers {

let (subscription_matcher_sender, subscription_matcher_receiver) =

async_channel::bounded(MATCHERS_TASK_CHANNEL_BUFFER_SIZE);

subscriptions_matchers_senders.push(subscription_matcher_sender);

task_spawner.spawn_blocking(

"statement-store-subscription-filters",

Some("statement-store"),

Box::pin(async move {

let mut subscriptions = SubscriptionsInfo::new();

log::debug!(

target: LOG_TARGET,

"Started statement subscription matcher task"

for task in 0..num_matcher_workers {

let (subscription_matcher_sender, subscription_matcher_receiver) =

async_channel::bounded(MATCHERS_TASK_CHANNEL_BUFFER_SIZE);

subscriptions_matchers_senders.push(subscription_matcher_sender);

task_spawner.spawn_blocking(

"statement-store-subscription-filters",

Some("statement-store"),

Box::pin(async move {

let mut subscriptions = SubscriptionsInfo::new();

log::debug!(

target: LOG_TARGET,

"Started statement subscription matcher task: {task}"

bkchr · 2026-01-30T16:38:45Z

substrate/client/statement-store/src/subscription.rs

+								// Expected when the subscription manager is dropped at shutdown.
+								log::error!(
+									target: LOG_TARGET,
+									"Statement subscription matcher channel closed"


Suggested change

"Statement subscription matcher channel closed"

"Statement subscription matcher channel closed: {task}"

bkchr · 2026-01-30T16:41:49Z

substrate/client/statement-store/src/subscription.rs

+					.insert(subscription_info.seq_id, subscription_info.clone());
+			},
+			CheckedTopicFilter::MatchAll(topics) =>
+				for topic in topics {
+					self.subscriptions_match_all_by_topic.entry(*topic).or_default()
+						[topics.len() - 1]
+						.insert(subscription_info.seq_id, subscription_info.clone());
+				},
+			CheckedTopicFilter::MatchAny(topics) =>
+				for topic in topics {
+					self.subscriptions_match_any_by_topic
+						.entry(*topic)
+						.or_default()
+						.insert(subscription_info.seq_id, subscription_info.clone());


Suggested change

.insert(subscription_info.seq_id, subscription_info.clone());

},

CheckedTopicFilter::MatchAll(topics) =>

for topic in topics {

self.subscriptions_match_all_by_topic.entry(*topic).or_default()

[topics.len() - 1]

.insert(subscription_info.seq_id, subscription_info.clone());

},

CheckedTopicFilter::MatchAny(topics) =>

for topic in topics {

self.subscriptions_match_any_by_topic

.entry(*topic)

.or_default()

.insert(subscription_info.seq_id, subscription_info.clone());

.insert(subscription_info.seq_id, subscription_info);

},

CheckedTopicFilter::MatchAll(topics) =>

for topic in topics {

self.subscriptions_match_all_by_topic.entry(*topic).or_default()

[topics.len() - 1]

.insert(subscription_info.seq_id, subscription_info);

},

CheckedTopicFilter::MatchAny(topics) =>

for topic in topics {

self.subscriptions_match_any_by_topic

.entry(*topic)

.or_default()

.insert(subscription_info.seq_id, subscription_info);

bkchr · 2026-01-30T16:50:02Z

substrate/client/statement-store/src/subscription.rs

+		// Check all combinations of topics in the statement to find matching subscriptions.
+		// This works well because the maximum allowed topics is small (MAX_TOPICS = 4).
+		for num_topics_to_check in 1..=num_topics {
+			for topics_combination in statement.topics().iter().combinations(num_topics_to_check) {


Why do we need the combinations? We are any way iterating all the topics? Could we not just use statement.topics() below to fin the topic with the fewest?

bkchr · 2026-01-30T16:54:29Z

substrate/client/statement-store/src/subscription.rs

+	) -> std::task::Poll<Option<Self::Item>> {
+		self.rx.poll_next_unpin(cx)
+	}
+}


Suggested change

}

}

AndreiEres and others added 28 commits November 25, 2025 16:04

Extend submission result types

853a9df

Remove NetworkPriority (which was never used)

c6d8e87

Add Invalid result

1cc5ae9

Handle encoding_too_large

550d9fa

Add rejected variant

b153d0f

Update from github-actions[bot] running command 'prdoc --audience nod…

32b198b

…e_dev --bump major'

Update

31a606b

Update

bc77940

Update PR doc

67d915f

Fix PR doc

068364a

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

4b6cc03

Co-authored-by: Alexandru Gheorghe <[email protected]>

Fix assertion

89c3c37

Don't copy types

a3a0a90

statement-store: api changes

534ec44

Signed-off-by: Alexandru Gheorghe <[email protected]>

Merge remote-tracking branch 'origin/AndreiEres/sss-rpc-types' into o…

b3d1b36

…ne_shot_the_api

Merge remote-tracking branch 'origin/master' into alexggh/api-consoli…

6ef12d3

…dation

remove replacement_preference_mask

af1be1d

Signed-off-by: Alexandru Gheorghe <[email protected]>

remove unused

c394914

Signed-off-by: Alexandru Gheorghe <[email protected]>

remove unused

3ad1478

Signed-off-by: Alexandru Gheorghe <[email protected]>

remove unused

31d3f53

Signed-off-by: Alexandru Gheorghe <[email protected]>

Merge remote-tracking branch 'origin/master' into alexggh/api-consoli…

03fcc46

…dation

address feedback

acf0103

Signed-off-by: Alexandru Gheorghe <[email protected]>

fixup documentation

65cd2ec

Signed-off-by: Alexandru Gheorghe <[email protected]>

fix documentation

e97cb8f

Signed-off-by: Alexandru Gheorghe <[email protected]>

remove non-api changes for now

eca03c7

Signed-off-by: Alexandru Gheorghe <[email protected]>

fix typos

f43fc81

Signed-off-by: Alexandru Gheorghe <[email protected]>

remove original api

849c17f

Signed-off-by: Alexandru Gheorghe <[email protected]>

statement-store: implement new api

dad2bd3

Signed-off-by: Alexandru Gheorghe <[email protected]>

alexggh mentioned this pull request Dec 19, 2025

[DRAFT] statement-store: re-defining RPC API surface #10452

Draft

update and use statement_store_bench

ec178a0

Signed-off-by: Alexandru Gheorghe <[email protected]>

alexggh added 2 commits January 9, 2026 18:20

Merge remote-tracking branch 'origin/master' into pr-10690

cf81c8a

make clippy happy

a7e65cd

Signed-off-by: Alexandru Gheorghe <[email protected]>

alexggh added the T0-node This PR/Issue is related to the topic “node”. label Jan 12, 2026

remove unused function

004f80b

Signed-off-by: Alexandru Gheorghe <[email protected]>

alexggh requested review from AndreiEres, P1sar and s0me0ne-unkn0wn January 12, 2026 15:28

AndreiEres reviewed Jan 12, 2026

View reviewed changes

s0me0ne-unkn0wn reviewed Jan 12, 2026

View reviewed changes

cumulus/parachains/runtimes/people/people-westend/src/lib.rs Outdated Show resolved Hide resolved

review feedback

4ec4df4

Signed-off-by: Alexandru Gheorghe <[email protected]>

AndreiEres approved these changes Jan 19, 2026

View reviewed changes

alexggh added 5 commits January 21, 2026 11:12

Merge remote-tracking branch 'origin/master' into alexggh/statement-s…

1f0bc62

…tore-not-api3

fix benchmark

9d9bad5

Signed-off-by: Alexandru Gheorghe <[email protected]>

fix CheckedTopicFilter

a1abb39

Signed-off-by: Alexandru Gheorghe <[email protected]>

review feedback

e0c9bf5

Signed-off-by: Alexandru Gheorghe <[email protected]>

fix test

cf4184a

Signed-off-by: Alexandru Gheorghe <[email protected]>

P1sar reviewed Jan 22, 2026

View reviewed changes

P1sar approved these changes Jan 23, 2026

View reviewed changes

bkchr reviewed Jan 23, 2026

View reviewed changes

alexggh and others added 5 commits January 27, 2026 10:55

review feedback

24f974d

Signed-off-by: Alexandru Gheorghe <[email protected]>

Merge remote-tracking branch 'origin/master' into alexggh/statement-s…

20e2bcc

…tore-not-api4

add prdoc

65680e9

Signed-off-by: Alexandru Gheorghe <[email protected]>

Change bump level for sc-statement-store and sc-utils

846a5ac

use bounded vec

e94e005

Signed-off-by: Alexandru Gheorghe <[email protected]>

alexggh added 3 commits January 29, 2026 18:58

make cargo fmt happy

5ba38bc

Signed-off-by: Alexandru Gheorghe <[email protected]>

fix statement_store_memory_stress_bench test

b15ba0b

Signed-off-by: Alexandru Gheorghe <[email protected]>

fix expiration

c40ce28

Signed-off-by: Alexandru Gheorghe <[email protected]>

bkchr approved these changes Jan 30, 2026

View reviewed changes

	if key_set.map_or(0, \|s\| s.len()) == 0 {
	if key_set.map_or(true, \|s\| s.is_empty()) {

	"Statement subscription matcher channel closed"
	"Statement subscription matcher channel closed: {task}"

statement-store: implement new rpc api #10690

Are you sure you want to change the base?

statement-store: implement new rpc api #10690

Uh oh!

Conversation

alexggh commented Dec 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Work in progress

Description

Submit changes

Subscribe changes

Remaining work

Local testing

Uh oh!

AndreiEres left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

s0me0ne-unkn0wn left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

P1sar commented Jan 14, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alexggh Jan 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

P1sar Jan 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alexggh Jan 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alexggh Jan 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

alexggh commented Dec 18, 2025 •

edited

Loading

alexggh Jan 22, 2026 •

edited

Loading

P1sar Jan 22, 2026 •

edited

Loading

alexggh Jan 23, 2026 •

edited

Loading

alexggh Jan 29, 2026 •

edited

Loading