Fix V2 statement wire format and remove hidden FPR defaults

Author: AndreiEres

lib/src/network/codec/statement.rs

@@ -48,8 +48,8 @@ const PROOF_ON_CHAIN: u8 = 3;
 pub use super::affinity::AffinityFilter;
 
 #[derive(Debug, Clone)]
-pub enum StatementMessage<'a> {
-    Statements(Vec<&'a [u8]>),
+pub enum StatementMessage {
+    Statements(Vec<([u8; 32], Statement)>),
     ExplicitTopicAffinity(AffinityFilter),
 }
 
@@ -210,14 +210,15 @@ const V2_TAG_AFFINITY: u8 = 0x01;
 
 pub fn decode_statement_message(
     bytes: &[u8],
-) -> Result<StatementMessage<'_>, DecodeStatementMessageError> {
+) -> Result<StatementMessage, DecodeStatementMessageError> {
     if bytes.is_empty() {
         return Err(DecodeStatementMessageError::Empty);
     }
 
     match bytes[0] {
         V2_TAG_STATEMENTS => {
-            let stmts = extract_statement_bytes(&bytes[1..])?;
+            let stmts = decode_statement_notification(&bytes[1..])
+                .map_err(DecodeStatementMessageError::InvalidStatements)?;
             Ok(StatementMessage::Statements(stmts))
         }
         V2_TAG_AFFINITY => {
@@ -230,67 +231,27 @@ pub fn decode_statement_message(
 }
 
 pub fn encode_statements_message(statements: &[&[u8]]) -> Vec<u8> {
+    let tag_len = 1;
+    let max_compact_len = 5;
     let total_len: usize = statements.iter().map(|s| s.len()).sum();
-    let mut out = Vec::with_capacity(1 + 5 + total_len);
+    let mut out = Vec::with_capacity(tag_len + max_compact_len + total_len);
     out.push(V2_TAG_STATEMENTS);
     out.extend_from_slice(crate::util::encode_scale_compact_usize(statements.len()).as_ref());
     for stmt in statements {
-        out.extend_from_slice(crate::util::encode_scale_compact_usize(stmt.len()).as_ref());
         out.extend_from_slice(stmt);
     }
     out
 }
 
 pub fn encode_topic_affinity_message(filter: &AffinityFilter) -> Vec<u8> {
+    let tag_len = 1;
     let encoded = filter.encode_to_vec();
-    let mut out = Vec::with_capacity(1 + encoded.len());
+    let mut out = Vec::with_capacity(tag_len + encoded.len());
     out.push(V2_TAG_AFFINITY);
     out.extend_from_slice(&encoded);
     out
 }
 
-fn extract_statement_bytes(data: &[u8]) -> Result<Vec<&[u8]>, DecodeStatementMessageError> {
-    let (mut remaining, count) =
-        crate::util::nom_scale_compact_usize::<nom::error::Error<&[u8]>>(data).map_err(|_| {
-            DecodeStatementMessageError::InvalidStatements(DecodeStatementNotificationError(
-                nom::error::ErrorKind::Fail,
-            ))
-        })?;
-
-    if count > MAX_STATEMENTS_PER_NOTIFICATION {
-        return Err(DecodeStatementMessageError::InvalidStatements(
-            DecodeStatementNotificationError(nom::error::ErrorKind::TooLarge),
-        ));
-    }
-
-    let mut statements = Vec::with_capacity(count);
-    for _ in 0..count {
-        let (rest, len) = crate::util::nom_scale_compact_usize::<nom::error::Error<&[u8]>>(
-            remaining,
-        )
-        .map_err(|_| {
-            DecodeStatementMessageError::InvalidStatements(DecodeStatementNotificationError(
-                nom::error::ErrorKind::Fail,
-            ))
-        })?;
-        if rest.len() < len {
-            return Err(DecodeStatementMessageError::InvalidStatements(
-                DecodeStatementNotificationError(nom::error::ErrorKind::Eof),
-            ));
-        }
-        statements.push(&rest[..len]);
-        remaining = &rest[len..];
-    }
-
-    if !remaining.is_empty() {
-        return Err(DecodeStatementMessageError::InvalidStatements(
-            DecodeStatementNotificationError(nom::error::ErrorKind::NonEmpty),
-        ));
-    }
-
-    Ok(statements)
-}
-
 #[derive(Debug, derive_more::Display, derive_more::Error, Clone)]
 pub enum DecodeStatementMessageError {
     #[display("Empty V2 statement message")]
@@ -751,15 +712,56 @@ mod tests {
         match decoded {
             StatementMessage::Statements(stmts) => {
                 assert_eq!(stmts.len(), 2);
-                assert_eq!(stmts[0], encoded1.as_slice());
-                assert_eq!(stmts[1], encoded2.as_slice());
+                assert_eq!(stmts[0].1, statement1);
+                assert_eq!(stmts[1].1, statement2);
             }
             _ => panic!("Expected Statements variant"),
         }
     }
 
     #[test]
-    fn v2_affinity_roundtrip() {
+    fn v2_statements_encoding_snapshot() {
+        let statement = Statement {
+            proof: Some(Proof::OnChain {
+                who: [42u8; 32],
+                block_hash: [24u8; 32],
+                event_index: 66,
+            }),
+            decryption_key: Some([0xde; 32]),
+            expiry: 999,
+            channel: Some([0xcc; 32]),
+            topics: vec![[0x01; 32], [0x02; 32]],
+            data: Some(vec![55, 99]),
+        };
+
+        let stmt_bytes = encode_statement(&statement).unwrap();
+        let v2_encoded = encode_statements_message(&[&stmt_bytes]);
+
+        let digest: [u8; 32] = blake2_rfc::blake2b::blake2b(32, &[], &v2_encoded)
+            .as_bytes()
+            .try_into()
+            .unwrap();
+        assert_eq!(
+            digest,
+            [
+                44, 71, 235, 73, 238, 115, 6, 15, 128, 174, 159, 216, 166, 76, 26, 101, 28, 143,
+                88, 21, 22, 128, 169, 62, 180, 19, 164, 234, 174, 210, 81, 105
+            ],
+            "blake2_256 digest must match polkadot-sdk snapshot"
+        );
+
+        let decoded = decode_statement_message(&v2_encoded).unwrap();
+        match decoded {
+            StatementMessage::Statements(stmts) => {
+                assert_eq!(stmts.len(), 1);
+                assert_eq!(stmts[0].1, statement);
+            }
+            _ => panic!("Expected Statements variant"),
+        }
+    }
+
+    #[test]
+    fn v2_affinity_encoding_snapshot() {
         let topic1 = [0x01u8; 32];
         let topic2 = [0x02u8; 32];
         let topic3 = [0x03u8; 32];
@@ -769,8 +771,21 @@ mod tests {
         filter.insert(&topic2);
 
         let encoded = encode_topic_affinity_message(&filter);
-        let decoded = decode_statement_message(&encoded).unwrap();
 
+        let digest: [u8; 32] = blake2_rfc::blake2b::blake2b(32, &[], &encoded)
+            .as_bytes()
+            .try_into()
+            .unwrap();
+        assert_eq!(
+            digest,
+            [
+                82, 59, 251, 163, 43, 156, 130, 249, 35, 214, 187, 99, 4, 105, 179, 131, 42, 117,
+                191, 57, 160, 243, 233, 20, 204, 239, 62, 120, 55, 5, 234, 62
+            ],
+            "blake2_256 digest must match polkadot-sdk snapshot"
+        );
+
+        let decoded = decode_statement_message(&encoded).unwrap();
         let StatementMessage::ExplicitTopicAffinity(af) = decoded else {
             panic!("Expected ExplicitTopicAffinity variant");
         };

lib/src/network/service.rs

@@ -2877,23 +2877,7 @@ where
                             }
                             codec::StatementProtocolVersion::V2 => {
                                 match codec::decode_statement_message(&notification) {
-                                    Ok(codec::StatementMessage::Statements(stmts)) => {
-                                        let mut statements = Vec::with_capacity(stmts.len());
-                                        for raw in stmts {
-                                            let hash = codec::statement_hash(raw);
-                                            match codec::decode_statement(raw) {
-                                                Ok(s) => statements.push((hash, s)),
-                                                Err(err) => {
-                                                    return Some(Event::ProtocolError {
-                                                        error:
-                                                            ProtocolError::BadStatementNotification(
-                                                                err,
-                                                            ),
-                                                        peer_id: self.peers[peer_index.0].clone(),
-                                                    });
-                                                }
-                                            }
-                                        }
+                                    Ok(codec::StatementMessage::Statements(statements)) => {
                                         if statements.is_empty() {
                                             continue;
                                         }
@@ -4119,7 +4103,6 @@ where
         let Some(&peer_index) = self.peers_by_peer_id.get(target) else {
             return Err(QueueNotificationError::NoConnection);
         };
-
         let chain_index = chain_id.0;
 
         let (protocol, notification) = self
@@ -4159,7 +4142,6 @@ where
         let Some(&peer_index) = self.peers_by_peer_id.get(target) else {
             return Err(SendTopicAffinityError::NoConnection);
         };
-
         let chain_index = chain_id.0;
 
         match self.find_statement_protocol_for_peer(peer_index, chain_index) {
@@ -4739,15 +4721,28 @@ pub enum Event<TConn> {
         statements: Vec<([u8; 32], codec::Statement)>,
     },
 
+    /// A statement protocol substream has been successfully negotiated with a peer.
+    ///
+    /// Indicates which protocol version (V1 or V2) was agreed upon.
     StatementProtocolConnected {
+        /// Identity of the remote peer.
         peer_id: PeerId,
+        /// Index of the chain the substream relates to.
         chain_id: ChainId,
+        /// Negotiated statement protocol version.
         version: codec::StatementProtocolVersion,
     },
 
+    /// Received a topic affinity bloom filter from a V2 peer.
+    ///
+    /// The filter indicates which statement topics the peer is interested in.
+    /// Only statements matching the filter need to be sent to this peer.
     StatementTopicAffinityReceived {
+        /// Identity of the remote peer.
         peer_id: PeerId,
+        /// Index of the chain the affinity relates to.
         chain_id: ChainId,
+        /// Bloom filter representing the peer's topic interests.
         filter: codec::AffinityFilter,
     },
 

light-base/src/json_rpc_service.rs

@@ -116,9 +116,8 @@ pub struct Config<TPlat: PlatformRef> {
     /// Hash of the genesis block of the chain.
     pub genesis_block_hash: [u8; 32],
 
-    /// False positive rate for bloom filters used in statement topic affinity.
-    /// `None` if the statement protocol is disabled.
-    pub bloom_false_positive_rate: Option<f64>,
+    /// Statement protocol configuration. `None` if the statement protocol is disabled.
+    pub statement_protocol_config: Option<network_service::StatementProtocolConfig>,
 }
 
 /// Creates a new JSON-RPC service with the given configuration.
@@ -157,7 +156,7 @@ pub fn service<TPlat: PlatformRef>(config: Config<TPlat>) -> Frontend<TPlat> {
                 system_name: config.system_name,
                 system_version: config.system_version,
                 genesis_block_hash: config.genesis_block_hash,
-                bloom_false_positive_rate: config.bloom_false_positive_rate,
+                statement_protocol_config: config.statement_protocol_config,
             },
             requests_rx,
             responses_tx,

light-base/src/json_rpc_service/background.rs

@@ -87,7 +87,8 @@ pub(super) struct Config<TPlat: PlatformRef> {
     /// Hash of the genesis block of the chain.
     pub genesis_block_hash: [u8; 32],
 
-    pub bloom_false_positive_rate: Option<f64>,
+    /// Statement protocol configuration. `None` if the statement protocol is disabled.
+    pub statement_protocol_config: Option<network_service::StatementProtocolConfig>,
 }
 
 /// Fields used to process JSON-RPC requests in the background.
@@ -118,7 +119,7 @@ struct Background<TPlat: PlatformRef> {
     /// Randomness used for various purposes, such as generating subscription IDs.
     randomness: ChaCha20Rng,
 
-    bloom_false_positive_rate: Option<f64>,
+    statement_protocol_config: Option<network_service::StatementProtocolConfig>,
 
     bloom_seed: u128,
 
@@ -517,7 +518,7 @@ pub(super) async fn run<TPlat: PlatformRef>(
             config.platform.fill_random_bytes(&mut seed);
             seed
         }),
-        bloom_false_positive_rate: config.bloom_false_positive_rate,
+        statement_protocol_config: config.statement_protocol_config,
         bloom_seed: {
             let mut seed_bytes = [0u8; 16];
             config.platform.fill_random_bytes(&mut seed_bytes);
@@ -779,7 +780,7 @@ pub(super) async fn run<TPlat: PlatformRef>(
                 if matches!(version, network_service::StatementProtocolVersion::V2) {
                     me.v2_statement_peers.insert(peer_id.clone());
                     if !me.statement_subscriptions.is_empty() {
-                        let fpr = me.bloom_false_positive_rate.unwrap_or(0.01);
+                        let fpr = me.statement_protocol_config.as_ref().expect("V2 peers require statement protocol; qed").false_positive_rate();
                         let combined_filter = build_combined_affinity_filter(
                             &me.statement_subscriptions,
                             me.bloom_seed,
@@ -2926,7 +2927,7 @@ pub(super) async fn run<TPlat: PlatformRef>(
                             .insert(subscription_id.clone(), filter);
 
                         if !me.v2_statement_peers.is_empty() {
-                            let fpr = me.bloom_false_positive_rate.unwrap_or(0.01);
+                            let fpr = me.statement_protocol_config.as_ref().expect("V2 peers require statement protocol; qed").false_positive_rate();
                             let combined_filter = build_combined_affinity_filter(
                                 &me.statement_subscriptions,
                                 me.bloom_seed,
@@ -2955,7 +2956,7 @@ pub(super) async fn run<TPlat: PlatformRef>(
                         let existed = me.statement_subscriptions.remove(&subscription).is_some();
 
                         if existed && !me.v2_statement_peers.is_empty() {
-                            let fpr = me.bloom_false_positive_rate.unwrap_or(0.01);
+                            let fpr = me.statement_protocol_config.as_ref().expect("V2 peers require statement protocol; qed").false_positive_rate();
                             let combined_filter = build_combined_affinity_filter(
                                 &me.statement_subscriptions,
                                 me.bloom_seed,

light-base/src/lib.rs

@@ -672,11 +672,6 @@ impl<TPlat: platform::PlatformRef, TChain> Client<TPlat, TChain> {
             }
         };
 
-        let bloom_false_positive_rate = config
-            .statement_protocol_config
-            .as_ref()
-            .map(|c| c.false_positive_rate());
-
         // Start the services of the chain to add, or grab the services if they already exist.
         let (services, log_name) = match chains_by_key.entry(new_chain_key.clone()) {
             Entry::Occupied(mut entry) => {
@@ -702,6 +697,7 @@ impl<TPlat: platform::PlatformRef, TChain> Client<TPlat, TChain> {
                     );
 
                     let statement_protocol_config = config.statement_protocol_config;
+                    let statement_protocol_config_for_rpc = statement_protocol_config.clone();
 
                     let config = match (&relay_chain, &chain_information) {
                         (Some((relay_chain, para_id, _)), Some(chain_information)) => {
@@ -956,7 +952,7 @@ impl<TPlat: platform::PlatformRef, TChain> Client<TPlat, TChain> {
                 system_name: self.platform.client_name().into_owned(),
                 system_version: self.platform.client_version().into_owned(),
                 genesis_block_hash,
-                bloom_false_positive_rate,
+                statement_protocol_config: statement_protocol_config_for_rpc,
             });
 
             Some(frontend)

light-base/src/sync_service/standalone.rs

@@ -811,10 +811,13 @@ pub(super) async fn start_standalone_chain<TPlat: PlatformRef>(
 
             WakeUpReason::NetworkEvent(network_service::Event::StatementsNotification {
                 ..
-            }) => {}
-            WakeUpReason::NetworkEvent(network_service::Event::StatementProtocolConnected {
+            })
+            | WakeUpReason::NetworkEvent(network_service::Event::StatementProtocolConnected {
                 ..
-            }) => {}
+            }) => {
+                // Statement store protocol events are handled by the JSON-RPC service.
+                // The standalone sync service doesn't need to react to them.
+            }
 
             WakeUpReason::MustSubscribeNetworkEvents => {
                 debug_assert!(task.from_network_service.is_none());

wasm-node/javascript/src/internals/client.ts

@@ -521,7 +521,7 @@ export function start(options: ClientOptions, wasmModule: SmoldotBytecode | Prom
                 if (statementStoreMaxSeenStatements > 0xffffffff) {
                     statementStoreMaxSeenStatements = 0xffffffff;
                 }
-                statementStoreFalsePositiveRate = options.statementStore.falsePositiveRate === undefined ? 0.01 : options.statementStore.falsePositiveRate;
+                statementStoreFalsePositiveRate = options.statementStore.falsePositiveRate;
                 if (statementStoreFalsePositiveRate <= 0.0 || statementStoreFalsePositiveRate >= 1.0 || isNaN(statementStoreFalsePositiveRate)) {
                     throw new AddChainError("Invalid value for `statementStore.falsePositiveRate`");
                 }

wasm-node/javascript/src/public-types.ts

@@ -464,6 +464,6 @@ export interface AddChainOptions {
      */
     statementStore?: {
         maxSeenStatements?: number,
-        falsePositiveRate?: number,
+        falsePositiveRate: number,
     }
 }