smoldot_light/json_rpc_service/
statement.rs

1// Smoldot
2// Copyright (C) 2019-2022  Parity Technologies (UK) Ltd.
3// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
4
5// This program is free software: you can redistribute it and/or modify
6// it under the terms of the GNU General Public License as published by
7// the Free Software Foundation, either version 3 of the License, or
8// (at your option) any later version.
9
10// This program is distributed in the hope that it will be useful,
11// but WITHOUT ANY WARRANTY; without even the implied warranty of
12// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13// GNU General Public License for more details.
14
15// You should have received a copy of the GNU General Public License
16// along with this program.  If not, see <http://www.gnu.org/licenses/>.
17
18use crate::network_service::{self, BroadcastStatementResult};
19use alloc::{format, string::String, vec::Vec};
20use core::{num::NonZero, time::Duration};
21use smoldot::json_rpc::methods::{HexString, InvalidReason, StatementSubmitResult, TopicFilter};
22use smoldot::json_rpc::parse;
23use smoldot::network::codec;
24
25/// Configuration for the Statement Store protocol.
26#[derive(Debug, Clone)]
27pub struct StatementProtocolConfig {
28    /// Per-subscription LRU cache size used for deduplicating delivered statements.
29    max_seen_statements: NonZero<usize>,
30    false_positive_rate: f64,
31    bloom_seed: u128,
32    affinity_update_interval: Duration,
33}
34
35impl StatementProtocolConfig {
36    pub fn new(
37        max_seen_statements: NonZero<usize>,
38        false_positive_rate: f64,
39        bloom_seed: u128,
40        affinity_update_interval: Duration,
41    ) -> Self {
42        assert!(
43            false_positive_rate.is_finite()
44                && false_positive_rate > 0.0
45                && false_positive_rate < 1.0
46        );
47        assert!(!affinity_update_interval.is_zero());
48        StatementProtocolConfig {
49            max_seen_statements,
50            false_positive_rate,
51            bloom_seed,
52            affinity_update_interval,
53        }
54    }
55
56    pub fn max_seen_statements(&self) -> NonZero<usize> {
57        self.max_seen_statements
58    }
59
60    pub fn false_positive_rate(&self) -> f64 {
61        self.false_positive_rate
62    }
63
64    pub fn bloom_seed(&self) -> u128 {
65        self.bloom_seed
66    }
67
68    pub fn affinity_update_interval(&self) -> Duration {
69        self.affinity_update_interval
70    }
71}
72
73/// JSON-RPC error code answering a submission the statement store couldn't process.
74///
75/// Matches polkadot-sdk.
76pub const STATEMENT_STORE_ERROR_CODE: i64 = 7001;
77
78/// Submission failure reported as a JSON-RPC error
79#[derive(Debug, Clone, PartialEq, Eq)]
80pub enum StatementSubmitError {
81    /// The submitted bytes don't decode into a statement.
82    InvalidEncoding,
83    /// The statement is valid but reached none of the gossip-connected peers. A light
84    /// client keeps no store, so a statement nobody received is permanently lost.
85    NotSent { connected: usize },
86}
87
88impl StatementSubmitError {
89    /// Builds the JSON-RPC error response answering this failure.
90    ///
91    /// # Panic
92    ///
93    /// Panics if `request_id_json` isn't valid JSON.
94    pub fn to_json_rpc_error(&self, request_id_json: &str) -> String {
95        // Both messages carry polkadot-sdk's prefix. It answers every statement-store failure with
96        // the same code, so the message is all that tells these two apart.
97        let message = match self {
98            StatementSubmitError::InvalidEncoding => {
99                String::from("Statement store error: Error decoding statement")
100            }
101            StatementSubmitError::NotSent { connected: 0 } => String::from(
102                "Statement store error: No connected peers to broadcast the statement to",
103            ),
104            // Connected peers can still decline to queue, so the count separates the two.
105            StatementSubmitError::NotSent { connected } => format!(
106                "Statement store error: none of the {connected} connected peers accepted the \
107                 statement"
108            ),
109        };
110
111        parse::build_error_response(
112            request_id_json,
113            parse::ErrorResponse::ApplicationDefined(STATEMENT_STORE_ERROR_CODE, &message),
114            None,
115        )
116    }
117}
118
119/// Validates a SCALE-encoded statement and broadcasts it to the network.
120///
121/// The checks run in the order polkadot-sdk's `Store::submit` applies them — expiry, then size,
122/// then proof — so that a client submitting a statement failing several of them is told the same
123/// reason a full node would give. Checks needing a local store or chain state are skipped.
124///
125/// The `broadcast` closure is only called if the statement is valid.
126pub async fn validate_and_broadcast_statement<F, Fut>(
127    encoded: &[u8],
128    now_from_unix_epoch: Duration,
129    broadcast: F,
130) -> Result<StatementSubmitResult, StatementSubmitError>
131where
132    F: FnOnce(Vec<u8>) -> Fut,
133    Fut: core::future::Future<Output = BroadcastStatementResult>,
134{
135    let Ok(statement) = codec::decode_statement(encoded) else {
136        return Err(StatementSubmitError::InvalidEncoding);
137    };
138
139    if now_from_unix_epoch.as_secs() >= statement.expiry >> 32 {
140        return Ok(StatementSubmitResult::Invalid(
141            InvalidReason::AlreadyExpired,
142        ));
143    }
144
145    if encoded.len() > codec::MAX_STATEMENT_SIZE {
146        return Ok(StatementSubmitResult::Invalid(
147            InvalidReason::EncodingTooLarge {
148                submitted_size: encoded.len(),
149                max_size: codec::MAX_STATEMENT_SIZE,
150            },
151        ));
152    }
153
154    if statement.proof.is_none() {
155        return Ok(StatementSubmitResult::Invalid(InvalidReason::NoProof));
156    }
157
158    let broadcasted = broadcast(encoded.to_vec()).await;
159    if broadcasted.sent == 0 {
160        return Err(StatementSubmitError::NotSent {
161            connected: broadcasted.total,
162        });
163    }
164
165    Ok(StatementSubmitResult::New)
166}
167
168pub(super) struct StatementSubscription {
169    topic_filter: TopicFilter,
170    seen: Option<lru::LruCache<[u8; 32], (), fnv::FnvBuildHasher>>,
171}
172
173impl StatementSubscription {
174    pub(super) fn new(topic_filter: TopicFilter, max_seen: Option<NonZero<usize>>) -> Self {
175        Self {
176            topic_filter,
177            seen: max_seen
178                .map(|cap| lru::LruCache::with_hasher(cap, fnv::FnvBuildHasher::default())),
179        }
180    }
181
182    pub(super) fn accept(&mut self, hash: &[u8; 32], statement: &codec::Statement) -> bool {
183        if !self.topic_filter.matches(&statement.topics) {
184            return false;
185        }
186        if let Some(seen) = &mut self.seen {
187            if seen.put(*hash, ()).is_some() {
188                return false;
189            }
190        }
191        true
192    }
193}
194
195/// Set of active statement subscriptions together with a reverse index mapping each topic to the
196/// subscriptions that reference it.
197///
198/// The reverse index lets statement matching scale with the number of subscriptions that share a
199/// topic with the incoming statement, rather than with the total number of subscriptions.
200pub(super) struct StatementSubscriptions {
201    /// Maps subscription ID to its state.
202    subscriptions: hashbrown::HashMap<String, StatementSubscription, fnv::FnvBuildHasher>,
203
204    /// Reverse index: maps a topic to the IDs of all subscriptions whose filter references it.
205    /// Only populated for `MatchAny`/`MatchAll` filters with a non-empty topic list.
206    by_topic: hashbrown::HashMap<
207        [u8; 32],
208        hashbrown::HashSet<String, fnv::FnvBuildHasher>,
209        fnv::FnvBuildHasher,
210    >,
211
212    /// IDs of subscriptions that match every statement irrespective of its topics: either
213    /// `TopicFilter::Any`, or a `TopicFilter::MatchAll` whose topic list is empty.
214    wildcard: hashbrown::HashSet<String, fnv::FnvBuildHasher>,
215}
216
217impl StatementSubscriptions {
218    pub(super) fn with_capacity(capacity: usize) -> Self {
219        Self {
220            subscriptions: hashbrown::HashMap::with_capacity_and_hasher(
221                capacity,
222                Default::default(),
223            ),
224            by_topic: hashbrown::HashMap::with_hasher(Default::default()),
225            wildcard: hashbrown::HashSet::with_hasher(Default::default()),
226        }
227    }
228
229    pub(super) fn is_empty(&self) -> bool {
230        self.subscriptions.is_empty()
231    }
232
233    /// Inserts a new subscription and updates the reverse index.
234    pub(super) fn insert(
235        &mut self,
236        id: String,
237        topic_filter: TopicFilter,
238        max_seen: Option<NonZero<usize>>,
239    ) {
240        match &topic_filter {
241            TopicFilter::Any => {
242                self.wildcard.insert(id.clone());
243            }
244            // An empty `MatchAll` filter matches every statement.
245            TopicFilter::MatchAll(topics) if topics.is_empty() => {
246                self.wildcard.insert(id.clone());
247            }
248            TopicFilter::MatchAll(topics) | TopicFilter::MatchAny(topics) => {
249                for topic in topics {
250                    self.by_topic
251                        .entry(*topic)
252                        .or_insert_with(|| hashbrown::HashSet::with_hasher(Default::default()))
253                        .insert(id.clone());
254                }
255            }
256        }
257
258        self.subscriptions
259            .insert(id, StatementSubscription::new(topic_filter, max_seen));
260    }
261
262    /// Removes a subscription and cleans up the reverse index. Returns whether it existed.
263    pub(super) fn remove(&mut self, id: &str) -> bool {
264        let Some(sub) = self.subscriptions.remove(id) else {
265            return false;
266        };
267
268        match &sub.topic_filter {
269            TopicFilter::Any => {
270                self.wildcard.remove(id);
271            }
272            TopicFilter::MatchAll(topics) if topics.is_empty() => {
273                self.wildcard.remove(id);
274            }
275            TopicFilter::MatchAll(topics) | TopicFilter::MatchAny(topics) => {
276                for topic in topics {
277                    if let Some(ids) = self.by_topic.get_mut(topic) {
278                        ids.remove(id);
279                        if ids.is_empty() {
280                            self.by_topic.remove(topic);
281                        }
282                    }
283                }
284            }
285        }
286
287        true
288    }
289
290    pub(super) fn shrink_to_fit(&mut self) {
291        self.subscriptions.shrink_to_fit();
292        for ids in self.by_topic.values_mut() {
293            ids.shrink_to_fit();
294        }
295        self.by_topic.shrink_to_fit();
296        self.wildcard.shrink_to_fit();
297    }
298
299    /// Matches a batch of statements against the subscriptions.
300    ///
301    /// Returns, for every subscription that accepts at least one statement, the list of re-encoded
302    /// matching statements. Uses the reverse index to only consider subscriptions that either match
303    /// everything or share a topic with the statement; the precise per-subscription filter and
304    /// deduplication is then applied via [`StatementSubscription::accept`].
305    pub(super) fn matching(
306        &mut self,
307        statements: &[([u8; 32], codec::Statement)],
308    ) -> Vec<(String, Vec<HexString>)> {
309        // Disjoint borrows: `subscriptions` is mutated while `by_topic`/`wildcard` are only read.
310        let Self {
311            subscriptions,
312            by_topic,
313            wildcard,
314        } = self;
315
316        // Subscription ID -> its matching re-encoded statements.
317        let mut out: hashbrown::HashMap<&str, Vec<HexString>, fnv::FnvBuildHasher> =
318            hashbrown::HashMap::with_hasher(Default::default());
319        // Reused across statements to avoid reallocating.
320        let mut candidates: hashbrown::HashSet<&str, fnv::FnvBuildHasher> =
321            hashbrown::HashSet::with_hasher(Default::default());
322
323        for (hash, statement) in statements {
324            candidates.clear();
325            candidates.extend(wildcard.iter().map(String::as_str));
326            for topic in &statement.topics {
327                if let Some(ids) = by_topic.get(topic) {
328                    candidates.extend(ids.iter().map(String::as_str));
329                }
330            }
331
332            // Re-encoded lazily on first match and reused for every matching subscription.
333            let mut encoded: Option<HexString> = None;
334            for id in &candidates {
335                let sub = subscriptions
336                    .get_mut(*id)
337                    .expect("`candidates` is a subset of `subscriptions`; qed");
338                if sub.accept(hash, statement) {
339                    let encoded = encoded.get_or_insert_with(|| {
340                        HexString(
341                            codec::encode_statement(statement)
342                                .expect("re-encoding a decoded statement always succeeds; qed"),
343                        )
344                    });
345                    out.entry(*id).or_default().push(encoded.clone());
346                }
347            }
348        }
349
350        out.into_iter()
351            .map(|(id, matching)| (String::from(id), matching))
352            .collect()
353    }
354
355    pub(super) fn build_combined_affinity_filter(
356        &self,
357        config: &StatementProtocolConfig,
358    ) -> network_service::AffinityFilter {
359        let mut all_topics: Vec<&[u8; 32]> = Vec::new();
360
361        for sub in self.subscriptions.values() {
362            match &sub.topic_filter {
363                TopicFilter::Any => {
364                    return network_service::AffinityFilter::match_all(config.bloom_seed());
365                }
366                TopicFilter::MatchAll(topics) | TopicFilter::MatchAny(topics) => {
367                    all_topics.extend(topics.iter());
368                }
369            }
370        }
371
372        network_service::AffinityFilter::from_topics(
373            all_topics.into_iter(),
374            config.bloom_seed(),
375            config.false_positive_rate(),
376        )
377    }
378}
379
380#[cfg(test)]
381mod tests {
382    use super::*;
383    use alloc::string::ToString as _;
384    use core::time::Duration;
385    use futures_lite::future::block_on;
386
387    const SEED: u128 = 0x5EED_5EED_5EED_5EED_5EED_5EED_5EED_5EED;
388    const FPR: f64 = 0.01;
389
390    fn test_config() -> StatementProtocolConfig {
391        StatementProtocolConfig::new(
392            NonZero::new(128).unwrap(),
393            FPR,
394            SEED,
395            Duration::from_secs(1),
396        )
397    }
398
399    fn make_subscriptions(
400        entries: Vec<(&str, TopicFilter, Option<NonZero<usize>>)>,
401    ) -> StatementSubscriptions {
402        let mut subs = StatementSubscriptions::with_capacity(entries.len());
403        for (id, filter, max_seen) in entries {
404            subs.insert(id.to_string(), filter, max_seen);
405        }
406        subs
407    }
408
409    fn statement_with_topics(topics: Vec<[u8; 32]>) -> codec::Statement {
410        codec::Statement {
411            proof: None,
412            decryption_key: None,
413            expiry: 42,
414            channel: None,
415            topics,
416            data: None,
417        }
418    }
419
420    const NOW: Duration = Duration::from_secs(1_000);
421
422    /// Expiration timestamp, in the most significant 32 bits, later than [`NOW`].
423    const FUTURE_EXPIRY: u64 = 2_000 << 32;
424
425    fn encoded_statement(with_proof: bool, expiry: u64, data: Option<Vec<u8>>) -> Vec<u8> {
426        codec::encode_statement(&codec::Statement {
427            proof: with_proof.then(|| codec::Proof::Ed25519 {
428                signature: [0; 64],
429                signer: [0; 32],
430            }),
431            decryption_key: None,
432            expiry,
433            channel: None,
434            topics: Vec::new(),
435            data,
436        })
437        .unwrap()
438    }
439
440    #[test]
441    fn validate_and_broadcast_invalid_encoding() {
442        let result = block_on(validate_and_broadcast_statement(
443            &[0xff, 0xff],
444            NOW,
445            |_| async { unreachable!() },
446        ));
447        assert_eq!(result, Err(StatementSubmitError::InvalidEncoding));
448    }
449
450    #[test]
451    fn validate_and_broadcast_already_expired() {
452        // The statement also has no proof: the expiry check runs first.
453        let encoded = encoded_statement(false, 500 << 32, None);
454        let result = block_on(validate_and_broadcast_statement(&encoded, NOW, |_| async {
455            unreachable!()
456        }));
457        assert_eq!(
458            result,
459            Ok(StatementSubmitResult::Invalid(
460                InvalidReason::AlreadyExpired
461            ))
462        );
463    }
464
465    #[test]
466    fn validate_and_broadcast_expiry_equal_to_now_is_expired() {
467        let encoded = encoded_statement(true, NOW.as_secs() << 32, None);
468        let result = block_on(validate_and_broadcast_statement(&encoded, NOW, |_| async {
469            unreachable!()
470        }));
471        assert_eq!(
472            result,
473            Ok(StatementSubmitResult::Invalid(
474                InvalidReason::AlreadyExpired
475            ))
476        );
477    }
478
479    #[test]
480    fn validate_and_broadcast_encoding_too_large() {
481        // The statement also has no proof: the size check runs before the proof check.
482        let encoded = encoded_statement(false, FUTURE_EXPIRY, Some(vec![0; 1024 * 1024]));
483        assert!(encoded.len() > codec::MAX_STATEMENT_SIZE);
484        let result = block_on(validate_and_broadcast_statement(&encoded, NOW, |_| async {
485            unreachable!()
486        }));
487        assert_eq!(
488            result,
489            Ok(StatementSubmitResult::Invalid(
490                InvalidReason::EncodingTooLarge {
491                    submitted_size: encoded.len(),
492                    max_size: codec::MAX_STATEMENT_SIZE,
493                }
494            ))
495        );
496    }
497
498    #[test]
499    fn validate_and_broadcast_no_proof() {
500        let encoded = encoded_statement(false, FUTURE_EXPIRY, None);
501        let result = block_on(validate_and_broadcast_statement(&encoded, NOW, |_| async {
502            unreachable!()
503        }));
504        assert_eq!(
505            result,
506            Ok(StatementSubmitResult::Invalid(InvalidReason::NoProof))
507        );
508    }
509
510    #[test]
511    fn validate_and_broadcast_no_peers() {
512        let encoded = encoded_statement(true, FUTURE_EXPIRY, None);
513        let result = block_on(validate_and_broadcast_statement(&encoded, NOW, |_| async {
514            BroadcastStatementResult { sent: 0, total: 0 }
515        }));
516        assert_eq!(result, Err(StatementSubmitError::NotSent { connected: 0 }));
517    }
518
519    #[test]
520    fn submit_errors_carry_the_statement_store_code() {
521        // Both failures answer with polkadot-sdk's single statement-store code, telling themselves
522        // apart by message alone, exactly as it does.
523        assert_eq!(
524            StatementSubmitError::InvalidEncoding.to_json_rpc_error("7"),
525            r#"{"jsonrpc":"2.0","id":7,"error":{"code":7001,"message":"Statement store error: Error decoding statement"}}"#
526        );
527        assert_eq!(
528            StatementSubmitError::NotSent { connected: 0 }.to_json_rpc_error("7"),
529            r#"{"jsonrpc":"2.0","id":7,"error":{"code":7001,"message":"Statement store error: No connected peers to broadcast the statement to"}}"#
530        );
531        // Connected peers that took nothing must not read as no peers.
532        assert_eq!(
533            StatementSubmitError::NotSent { connected: 5 }.to_json_rpc_error("7"),
534            r#"{"jsonrpc":"2.0","id":7,"error":{"code":7001,"message":"Statement store error: none of the 5 connected peers accepted the statement"}}"#
535        );
536    }
537
538    #[test]
539    fn validate_and_broadcast_reaching_no_peer_is_not_new() {
540        // Gossip-connected peers whose statement substream is missing or whose queue is full leave
541        // the statement unsent. Answering `new` would tell the client it was published.
542        let encoded = encoded_statement(true, FUTURE_EXPIRY, None);
543        let result = block_on(validate_and_broadcast_statement(&encoded, NOW, |_| async {
544            BroadcastStatementResult { sent: 0, total: 5 }
545        }));
546        assert_eq!(result, Err(StatementSubmitError::NotSent { connected: 5 }));
547    }
548
549    #[test]
550    fn validate_and_broadcast_new() {
551        let encoded = encoded_statement(true, FUTURE_EXPIRY, None);
552        let result = block_on(validate_and_broadcast_statement(&encoded, NOW, |_| async {
553            BroadcastStatementResult { sent: 3, total: 5 }
554        }));
555        assert_eq!(result, Ok(StatementSubmitResult::New));
556    }
557
558    #[test]
559    fn build_combined_affinity_empty_subscriptions() {
560        let config = test_config();
561        let subs = make_subscriptions(vec![]);
562        let filter = subs.build_combined_affinity_filter(&config);
563
564        // Empty subscription set: no topics are ever in the filter.
565        assert!(!filter.contains(&[1u8; 32]));
566        // A statement with no topics (broadcast) still matches.
567        let broadcast: &[&[u8; 32]] = &[];
568        assert!(filter.matches_statement(broadcast));
569    }
570
571    #[test]
572    fn build_combined_affinity_any_filter_matches_everything() {
573        let config = test_config();
574        let subs = make_subscriptions(vec![("s", TopicFilter::Any, None)]);
575        let filter = subs.build_combined_affinity_filter(&config);
576
577        // TopicFilter::Any returns the broadcast `match_all` filter: every topic matches.
578        assert!(filter.contains(&[1u8; 32]));
579        assert!(filter.contains(&[99u8; 32]));
580        let t = [7u8; 32];
581        assert!(filter.matches_statement(&[&t]));
582    }
583
584    #[test]
585    fn build_combined_affinity_match_any_union() {
586        let config = test_config();
587        let t1 = [1u8; 32];
588        let t2 = [2u8; 32];
589        let subs = make_subscriptions(vec![
590            ("a", TopicFilter::match_any(vec![t1]).unwrap(), None),
591            ("b", TopicFilter::match_any(vec![t2]).unwrap(), None),
592        ]);
593        let filter = subs.build_combined_affinity_filter(&config);
594
595        assert!(filter.contains(&t1));
596        assert!(filter.contains(&t2));
597    }
598
599    #[test]
600    fn accept_fresh_statement_passes() {
601        let t1 = [1u8; 32];
602        let mut sub =
603            StatementSubscription::new(TopicFilter::match_any(vec![t1]).unwrap(), NonZero::new(8));
604        let stmt = statement_with_topics(vec![t1]);
605        assert!(sub.accept(&[0xbb; 32], &stmt));
606    }
607
608    #[test]
609    fn accept_duplicate_returns_false() {
610        let mut sub = StatementSubscription::new(TopicFilter::Any, NonZero::new(8));
611        let stmt = statement_with_topics(vec![]);
612        let hash = [0xcc; 32];
613        assert!(sub.accept(&hash, &stmt));
614        assert!(!sub.accept(&hash, &stmt));
615    }
616
617    #[test]
618    fn accept_lru_eviction_allows_resubmit() {
619        let mut sub = StatementSubscription::new(TopicFilter::Any, NonZero::new(2));
620        let stmt = statement_with_topics(vec![]);
621        let h_a = [0xa; 32];
622        let h_b = [0xb; 32];
623        let h_c = [0xc; 32];
624
625        assert!(sub.accept(&h_a, &stmt));
626        assert!(sub.accept(&h_b, &stmt));
627        // Inserting a third eviction-capacity 2 item evicts h_a (oldest).
628        assert!(sub.accept(&h_c, &stmt));
629        // h_a was evicted: it is accepted again as if fresh.
630        assert!(sub.accept(&h_a, &stmt));
631    }
632
633    #[test]
634    fn dedup_is_per_subscription() {
635        let mut sub_a = StatementSubscription::new(TopicFilter::Any, NonZero::new(8));
636        let mut sub_b = StatementSubscription::new(TopicFilter::Any, NonZero::new(8));
637        let stmt = statement_with_topics(vec![]);
638        let hash = [0xee; 32];
639
640        assert!(sub_a.accept(&hash, &stmt));
641        assert!(!sub_a.accept(&hash, &stmt));
642        // Same hash on a different subscription is still fresh: caches are independent.
643        assert!(sub_b.accept(&hash, &stmt));
644    }
645
646    /// Builds a `(hash, statement)` batch entry from a list of topics.
647    fn batch_entry(hash: u8, topics: Vec<[u8; 32]>) -> ([u8; 32], codec::Statement) {
648        ([hash; 32], statement_with_topics(topics))
649    }
650
651    /// Collects the IDs of all subscriptions that matched at least once.
652    fn matched_ids(matches: &[(String, Vec<HexString>)]) -> Vec<String> {
653        let mut ids: Vec<String> = matches.iter().map(|(id, _)| id.clone()).collect();
654        ids.sort();
655        ids
656    }
657
658    #[test]
659    fn matching_match_any_only_returns_relevant_subscriptions() {
660        let t1 = [1u8; 32];
661        let t2 = [2u8; 32];
662        let mut subs = make_subscriptions(vec![
663            ("a", TopicFilter::match_any(vec![t1]).unwrap(), None),
664            ("b", TopicFilter::match_any(vec![t2]).unwrap(), None),
665        ]);
666
667        // A statement carrying only `t1` must match `a` and not `b`.
668        let matches = subs.matching(&[batch_entry(0xaa, vec![t1])]);
669        assert_eq!(matched_ids(&matches), vec!["a".to_string()]);
670
671        // A statement with an unrelated topic matches nothing.
672        let matches = subs.matching(&[batch_entry(0xbb, vec![[9u8; 32]])]);
673        assert!(matches.is_empty());
674    }
675
676    #[test]
677    fn matching_wildcard_filters_match_every_statement() {
678        // `Any` and an empty `MatchAll` both match every statement, with or without topics.
679        let mut subs = make_subscriptions(vec![
680            ("any", TopicFilter::Any, None),
681            ("all", TopicFilter::match_all(vec![]).unwrap(), None),
682        ]);
683
684        let matches = subs.matching(&[batch_entry(0x01, vec![[7u8; 32]])]);
685        assert_eq!(
686            matched_ids(&matches),
687            vec!["all".to_string(), "any".to_string()]
688        );
689
690        let matches = subs.matching(&[batch_entry(0x02, vec![])]);
691        assert_eq!(
692            matched_ids(&matches),
693            vec!["all".to_string(), "any".to_string()]
694        );
695    }
696
697    #[test]
698    fn matching_match_all_requires_every_topic() {
699        let t1 = [1u8; 32];
700        let t2 = [2u8; 32];
701        let mut subs = make_subscriptions(vec![(
702            "all",
703            TopicFilter::match_all(vec![t1, t2]).unwrap(),
704            None,
705        )]);
706
707        // A statement carrying only one of the required topics is a candidate via the reverse
708        // index but must be rejected by the precise re-check.
709        let matches = subs.matching(&[batch_entry(0xaa, vec![t1])]);
710        assert!(matches.is_empty());
711
712        // A statement carrying both topics matches.
713        let matches = subs.matching(&[batch_entry(0xbb, vec![t1, t2])]);
714        assert_eq!(matched_ids(&matches), vec!["all".to_string()]);
715    }
716
717    #[test]
718    fn matching_empty_match_any_never_matches() {
719        let mut subs = make_subscriptions(vec![(
720            "none",
721            TopicFilter::match_any(vec![]).unwrap(),
722            None,
723        )]);
724
725        let matches = subs.matching(&[batch_entry(0x01, vec![[1u8; 32]])]);
726        assert!(matches.is_empty());
727        let matches = subs.matching(&[batch_entry(0x02, vec![])]);
728        assert!(matches.is_empty());
729    }
730
731    #[test]
732    fn matching_dedup_applies_across_batches() {
733        let t1 = [1u8; 32];
734        let mut subs = make_subscriptions(vec![(
735            "a",
736            TopicFilter::match_any(vec![t1]).unwrap(),
737            NonZero::new(8),
738        )]);
739
740        let entry = batch_entry(0xaa, vec![t1]);
741        let matches = subs.matching(&[entry.clone()]);
742        assert_eq!(matches.len(), 1);
743        assert_eq!(matches[0].1.len(), 1);
744
745        // The same statement hash is deduplicated and produces no further notification.
746        let matches = subs.matching(&[entry]);
747        assert!(matches.is_empty());
748    }
749
750    #[test]
751    fn matching_groups_multiple_statements_per_subscription() {
752        let t1 = [1u8; 32];
753        let mut subs =
754            make_subscriptions(vec![("a", TopicFilter::match_any(vec![t1]).unwrap(), None)]);
755
756        let matches = subs.matching(&[batch_entry(0x01, vec![t1]), batch_entry(0x02, vec![t1])]);
757        assert_eq!(matches.len(), 1);
758        assert_eq!(matches[0].0, "a");
759        assert_eq!(matches[0].1.len(), 2);
760    }
761
762    #[test]
763    fn remove_cleans_reverse_index() {
764        let t1 = [1u8; 32];
765        let mut subs =
766            make_subscriptions(vec![("a", TopicFilter::match_any(vec![t1]).unwrap(), None)]);
767
768        assert!(subs.remove("a"));
769        assert!(!subs.remove("a"));
770        assert!(subs.is_empty());
771        // The topic entry must have been cleaned up, so a matching statement finds nothing.
772        let matches = subs.matching(&[batch_entry(0xaa, vec![t1])]);
773        assert!(matches.is_empty());
774    }
775}
smoldot_light/json_rpc_service/statement.rs

smoldot_light/json_rpc_service/
statement.rs
