diff --git a/.github/workflows/zombienet.yml b/.github/workflows/zombienet.yml index dac321ec59..c214c9eb2f 100644 --- a/.github/workflows/zombienet.yml +++ b/.github/workflows/zombienet.yml @@ -77,6 +77,9 @@ jobs: - job-name: "zombienet-smoldot-0013-statement_store_browser" test: "statement_store_browser" runner-type: "default" + - job-name: "zombienet-smoldot-0015-statement_store_spec_api" + test: "statement_store_spec_api" + runner-type: "default" - job-name: "zombienet-smoldot-0007-bulletin_fetch" test: "bulletin_fetch" runner-type: "default" diff --git a/e2e-tests/shared/statement_store_spec_api.js b/e2e-tests/shared/statement_store_spec_api.js new file mode 100644 index 0000000000..7cc65b9a71 --- /dev/null +++ b/e2e-tests/shared/statement_store_spec_api.js @@ -0,0 +1,247 @@ +// Smoldot +// Copyright (C) 2019-2026 Parity Technologies (UK) Ltd. +// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 + +// This program 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. + +// This program 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 this program. If not, see . + +// Statement-store `statement_unstable_*` test body — runs on either host via the +// ctx abstraction. Walks the whole subscription lifecycle: subscribe, turn down +// the filters the specification doesn't define, attach a `matchAll` filter, +// collect its `replayDone`, assert that only matching statements arrive and that +// they carry the filter id, submit a statement of our own, then detach the filter +// and close the subscription. + +import { createRpc } from "./rpc.js"; + +export const fileInputs = ["RELAY_CHAIN_SPEC", "PARA_CHAIN_SPEC"]; +export const envInputs = [ + "TOPIC_A", + "STATEMENT_A_HEX", + "STATEMENT_B_HEX", + "STATEMENT_C_HEX", + "LISTEN_MS", +]; + +/// Sends `method` and resolves with the whole response object, error included. +async function sendAndTakeResponse(rpc, chain, method, params) { + const id = rpc.sendRpc(chain, method, params).toString(); + const response = await rpc.readJsonRpcUntil( + chain, + (msg) => (msg.id === id ? msg : undefined), + Date.now() + 20_000, + ); + if (response === undefined) throw new Error(`Timed out waiting for ${method}`); + return response; +} + +export default async function statementStoreSpecApi(ctx) { + const { report, env, files } = ctx; + const rpc = createRpc(ctx.client); + + const topicAHex = env.TOPIC_A; + const stmtAHex = env.STATEMENT_A_HEX; + const stmtBHex = env.STATEMENT_B_HEX; + const stmtCHex = env.STATEMENT_C_HEX; + const listenMs = Number.parseInt(env.LISTEN_MS || "10000", 10); + + if ( + !files.RELAY_CHAIN_SPEC || + !files.PARA_CHAIN_SPEC || + !topicAHex || + !stmtAHex || + !stmtBHex || + !stmtCHex + ) { + throw new Error( + "Required env vars: RELAY_CHAIN_SPEC, PARA_CHAIN_SPEC, TOPIC_A, STATEMENT_A_HEX, " + + "STATEMENT_B_HEX, STATEMENT_C_HEX", + ); + } + + const relay = await rpc.addChain({ chainSpec: files.RELAY_CHAIN_SPEC }); + report("addChain relay", true); + + const para = await rpc.addChain({ + chainSpec: files.PARA_CHAIN_SPEC, + statementStore: {}, + potentialRelayChains: [relay], + }); + report("addChain parachain with statementStore", true); + + const subscribeResponse = await sendAndTakeResponse(rpc, para, "statement_unstable_subscribe", []); + if (subscribeResponse.error) { + throw new Error(`statement_unstable_subscribe failed: ${JSON.stringify(subscribeResponse.error)}`); + } + const subId = subscribeResponse.result; + if (typeof subId !== "string" || subId.length === 0) { + throw new Error(`Unexpected subscription id: ${JSON.stringify(subId)}`); + } + report("statement_unstable_subscribe accepted", true, `subId=${subId}`); + + // Rejections first: a filter is attached afterwards, and every read discards + // the messages it doesn't match, so this must happen before statements flow. + const badEncoding = await sendAndTakeResponse(rpc, para, "statement_unstable_submit", ["0xffff"]); + const badEncodingOk = badEncoding.error?.code === -32602; + report( + "submit rejects bytes that don't decode into a statement", + badEncodingOk, + JSON.stringify(badEncoding.error ?? badEncoding.result), + ); + if (!badEncodingOk) throw new Error("expected error -32602 for an undecodable statement"); + + const unknownSub = await sendAndTakeResponse(rpc, para, "statement_unstable_add_filter", [ + "0000000000000000000000000000000000000000000000000000000000000000", + "any", + ]); + const unknownSubOk = unknownSub.error?.code === -32801; + report( + "add_filter on an unknown subscription is refused", + unknownSubOk, + JSON.stringify(unknownSub.error ?? unknownSub.result), + ); + if (!unknownSubOk) throw new Error("expected error -32801 for an unknown subscription"); + + const matchAny = await sendAndTakeResponse(rpc, para, "statement_unstable_add_filter", [ + subId, + { matchAny: [topicAHex] }, + ]); + const matchAnyOk = matchAny.error?.code === -32602; + report( + "add_filter rejects a matchAny filter", + matchAnyOk, + JSON.stringify(matchAny.error ?? matchAny.result), + ); + if (!matchAnyOk) throw new Error("expected error -32602 for a matchAny filter"); + + const emptyMatchAll = await sendAndTakeResponse(rpc, para, "statement_unstable_add_filter", [ + subId, + { matchAll: [] }, + ]); + const emptyMatchAllOk = emptyMatchAll.error?.code === -32602; + report( + "add_filter rejects an empty matchAll filter", + emptyMatchAllOk, + JSON.stringify(emptyMatchAll.error ?? emptyMatchAll.result), + ); + if (!emptyMatchAllOk) throw new Error("expected error -32602 for an empty matchAll filter"); + + const addFilter = await sendAndTakeResponse(rpc, para, "statement_unstable_add_filter", [ + subId, + { matchAll: [topicAHex] }, + ]); + if (addFilter.error) { + throw new Error(`statement_unstable_add_filter failed: ${JSON.stringify(addFilter.error)}`); + } + const filterId = addFilter.result; + if (typeof filterId !== "string" || filterId.length === 0) { + throw new Error(`Unexpected filter id: ${JSON.stringify(filterId)}`); + } + report("statement_unstable_add_filter accepted", true, `filterId=${filterId}`); + + // A light client holds no statement store, so the replay covers nothing and + // completes immediately. Seeing `replayStatements` would mean smoldot claims a + // store it doesn't have. + const replayDone = await rpc.readJsonRpcUntil( + para, + (msg) => { + if (msg.method !== "statement_unstable_subscribeEvent") return undefined; + if (msg.params?.subscription !== subId) return undefined; + const result = msg.params.result; + if (result?.event === "replayStatements") { + throw new Error("a light client must not report replayStatements"); + } + if (result?.event === "replayDone" && result.filterId === filterId) return true; + return undefined; + }, + Date.now() + 20_000, + ); + report("replayDone received for the new filter", replayDone === true); + if (replayDone !== true) throw new Error("replayDone was not received"); + + // Block until Rust signals that smoldot is peered with both collators at the + // statement-store level. Both already hold stmt_A and push it once they learn + // about our topic affinity. + await ctx.waitSync("READY"); + report("Rust signalled READY", true); + + let countA = 0; + let countB = 0; + let countOther = 0; + let badFilterIds = 0; + let stopped = false; + const listenDeadline = Date.now() + listenMs; + + await rpc.readJsonRpcUntil( + para, + (msg) => { + if (msg.method !== "statement_unstable_subscribeEvent") return undefined; + if (msg.params?.subscription !== subId) return undefined; + const result = msg.params.result; + if (result?.event === "stop") { + stopped = true; + return true; + } + if (result?.event !== "newStatements") return undefined; + for (const item of result.statements ?? []) { + if (item.statement === stmtAHex) countA += 1; + else if (item.statement === stmtBHex) countB += 1; + else countOther += 1; + // Every reported statement matches the only attached filter. + if (!Array.isArray(item.filterIds) || !item.filterIds.includes(filterId)) { + badFilterIds += 1; + } + } + return undefined; + }, + listenDeadline, + ); + + const ok = countA === 1 && countB === 0 && countOther === 0 && badFilterIds === 0 && !stopped; + const detail = + `stmt_A count=${countA} | stmt_B count=${countB} | other count=${countOther} | ` + + `items missing the filter id=${badFilterIds} | stopped=${stopped}`; + report("newStatements: stmt_A once with its filter id, stmt_B never", ok, detail); + if (!ok) throw new Error(`newStatements assertion failed: ${detail}`); + + // Submitted only now that the counting window is closed, and on a topic the + // attached filter doesn't cover, so a statement gossiped back to us can't be + // mistaken for one of the statements counted above. Rust then checks that bob + // received it, which is what proves the broadcast left smoldot. + const submit = await sendAndTakeResponse(rpc, para, "statement_unstable_submit", [stmtCHex]); + const submitOk = !submit.error && submit.result?.status === "new"; + report("submit broadcasts a valid statement", submitOk, JSON.stringify(submit.error ?? submit.result)); + if (!submitOk) throw new Error(`unexpected submit response: ${JSON.stringify(submit)}`); + + const removeFilter = await sendAndTakeResponse(rpc, para, "statement_unstable_remove_filter", [ + subId, + filterId, + ]); + const removeOk = !removeFilter.error && removeFilter.result === null; + report("statement_unstable_remove_filter answers null", removeOk); + if (!removeOk) throw new Error(`unexpected remove_filter response: ${JSON.stringify(removeFilter)}`); + + // Removing a filter that is no longer attached is a no-op, not an error. + const removeAgain = await sendAndTakeResponse(rpc, para, "statement_unstable_remove_filter", [ + subId, + filterId, + ]); + const removeAgainOk = !removeAgain.error && removeAgain.result === null; + report("statement_unstable_remove_filter is idempotent", removeAgainOk); + if (!removeAgainOk) throw new Error("removing a filter twice must not fail"); + + const unsubscribe = await sendAndTakeResponse(rpc, para, "statement_unstable_unsubscribe", [subId]); + const unsubscribeOk = !unsubscribe.error && unsubscribe.result === null; + report("statement_unstable_unsubscribe answers null", unsubscribeOk); + if (!unsubscribeOk) throw new Error(`unexpected unsubscribe response: ${JSON.stringify(unsubscribe)}`); +} diff --git a/e2e-tests/tests/statement_store_spec_api.rs b/e2e-tests/tests/statement_store_spec_api.rs new file mode 100644 index 0000000000..e0eed6b462 --- /dev/null +++ b/e2e-tests/tests/statement_store_spec_api.rs @@ -0,0 +1,159 @@ +// Smoldot +// Copyright (C) 2019-2026 Parity Technologies (UK) Ltd. +// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 + +// This program 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. + +// This program 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 this program. If not, see . + +use log::info; +use smoldot_e2e_tests::statement::*; +use smoldot_e2e_tests::*; + +/// Smoldot serves the `statement_unstable_*` JSON-RPC API: filters are attached +/// to a subscription and detached from it, only statements matching an attached +/// filter are reported, and each one carries the ids of the filters it matched. +/// +/// Flow: +/// 1. Spawn alice + bob. +/// 2. Submit stmt_A and stmt_B to alice; wait for both to reach bob via +/// gossip, so that both collators hold them. +/// 3. Start smoldot; it subscribes, turns down the filters the specification +/// doesn't define, then attaches a `matchAll` filter on stmt_A's topic. +/// 4. Smoldot must report `replayDone` for that filter, then deliver stmt_A +/// exactly once carrying the filter id, and never stmt_B. +/// 5. The filter is detached and the subscription closed, both answering null. +#[tokio::test(flavor = "multi_thread")] +async fn serves_the_unstable_statement_api() -> Result<(), anyhow::Error> { + let _ = env_logger::try_init_from_env( + env_logger::Env::default().filter_or(env_logger::DEFAULT_FILTER_ENV, "info"), + ); + + let (seed, pubkey) = test_keypair(); + + let base_dir = resolve_base_dir()?; + let para_spec_path = create_para_chain_spec_with_allowances(&[pubkey], &base_dir)?; + info!( + "Parachain chain spec created at {}", + para_spec_path.display() + ); + + let network = spawn_network(&base_dir, ¶_spec_path).await?; + info!("Network spawned"); + + let (relay_base, para_base) = spawned_chain_spec_paths(&network)?; + + let base_dir_str = base_dir.to_str().expect("UTF-8 path").to_owned(); + let (relay_spec_path, para_spec_path) = + prepare_runtime_specs(&network, &relay_base, ¶_base, &base_dir_str).await?; + + // Subscribe on bob first so we don't miss gossip from alice. Bind the RPC + // client — dropping it closes the websocket and terminates the subscription. + let alice = network.get_node("alice")?; + let bob = network.get_node("bob")?; + let bob_rpc = bob.rpc().await?; + let mut bob_sub = subscribe_any(&bob_rpc).await?; + + info!("Ensuring smoldot JS bundle is built"); + ensure_smoldot_built(); + info!("Ensuring JS test dependencies are installed"); + ensure_js_deps_installed(); + + // Statements *and topics* need to be distinct per host, otherwise they + // persist in the collators' store and get pushed to the next host during + // the initial sync. + for (idx, host) in [Host::Node /* Host::Browser */].into_iter().enumerate() { + let mut topic_a = [0xaau8; 32]; + let mut topic_b = [0xbbu8; 32]; + let mut topic_c = [0xccu8; 32]; + topic_a[31] = idx as u8; + topic_b[31] = idx as u8; + topic_c[31] = idx as u8; + let stmt_a_hex = create_test_statement( + &seed, + &topic_a, + format!("spec-api-test-A-{host:?}").as_bytes(), + ); + let stmt_b_hex = create_test_statement( + &seed, + &topic_b, + format!("spec-api-test-B-{host:?}").as_bytes(), + ); + // Submitted by smoldot rather than by a collator, on a topic the subscription's filter + // doesn't cover, so it can't be confused with the statements counted on the JS side. + let stmt_c_hex = create_test_statement( + &seed, + &topic_c, + format!("spec-api-test-C-{host:?}").as_bytes(), + ); + + submit_statement(alice, &stmt_a_hex, "stmt_A").await?; + submit_statement(alice, &stmt_b_hex, "stmt_B").await?; + + let received = receive_statements(2, &mut bob_sub, 120).await?; + assert!(received.contains(&stmt_a_hex) && received.contains(&stmt_b_hex)); + info!("Both statements confirmed on bob via gossip"); + + let sync = SyncFile::new()?; + let sync_path_str = sync.path().to_str().unwrap().to_string(); + + let topic_a_hex = format!("0x{}", hex::encode(topic_a)); + let relay_spec_str = relay_spec_path.to_str().unwrap().to_string(); + let para_spec_str = para_spec_path.to_str().unwrap().to_string(); + let stmt_a_hex_js = stmt_a_hex.clone(); + let stmt_b_hex_js = stmt_b_hex.clone(); + let stmt_c_hex_js = stmt_c_hex.clone(); + let topic_a_hex_js = topic_a_hex.clone(); + + info!("Spawning test statement_store_spec_api within host {host:?} (topicA={topic_a_hex})"); + let js_handle = tokio::spawn(async move { + run_shared_test( + host, + "statement_store_spec_api", + &[ + ("RELAY_CHAIN_SPEC", relay_spec_str.as_str()), + ("PARA_CHAIN_SPEC", para_spec_str.as_str()), + ("TOPIC_A", topic_a_hex_js.as_str()), + ("STATEMENT_A_HEX", stmt_a_hex_js.as_str()), + ("STATEMENT_B_HEX", stmt_b_hex_js.as_str()), + ("STATEMENT_C_HEX", stmt_c_hex_js.as_str()), + ("SYNC_PATH", sync_path_str.as_str()), + ], + ) + .await + }); + + // Wait for smoldot to peer with both collators at the statement-store + // level. Each collator already holds stmt_A, and pushes it once it + // learns about smoldot's topic affinity. + wait_until_peered(alice, 1, 120).await?; + wait_until_peered(bob, 1, 120).await?; + + sync.send("READY")?; + info!("Signalled JS READY"); + + let js_result = js_handle.await.expect("JS task panicked"); + js_result.map_err(|e| anyhow::anyhow!("JS test failed: {e}"))?; + + // `statement_unstable_submit` answered `new`, which only says smoldot handed the statement + // to at least one peer. Seeing it arrive on bob is what proves it reached the network. + let received = receive_statements(1, &mut bob_sub, 120).await?; + assert!( + received.contains(&stmt_c_hex), + "stmt_C submitted through smoldot never reached bob" + ); + info!("stmt_C submitted by smoldot confirmed on bob"); + + info!("Unstable statement API test passed on host {host:?}"); + } + Ok(()) +} diff --git a/lib/src/json_rpc/methods.rs b/lib/src/json_rpc/methods.rs index 17a039d1f4..4b0b898640 100644 --- a/lib/src/json_rpc/methods.rs +++ b/lib/src/json_rpc/methods.rs @@ -464,6 +464,8 @@ define_methods! { /// Returns, as an opaque string, the version of the client serving these JSON-RPC requests. system_version() -> Cow<'a, str>, + // Legacy statement-store JSON RPC API + /// Broadcast a new statement to peers (light node has no local statement-store). statement_submit(encoded: HexString) -> StatementSubmitResult, /// Subscribe to statements matching the given filter. Returns subscription ID. @@ -471,6 +473,44 @@ define_methods! { /// Unsubscribe from statement notifications. statement_unsubscribeStatement(subscription: String) -> bool, + // Unstable statement-store JSON RPC API, defined in + // https://github.com/paritytech/json-rpc-interface-spec/ and implemented in polkadot-sdk by + // https://github.com/paritytech/polkadot-sdk/pull/11989. Where a storeless light client cannot + // reproduce the reference behavior, the divergence is documented on the item concerned. + + /// Validate a SCALE-encoded statement and broadcast it to peers. + statement_unstable_submit(encoded: HexString) -> StatementSubmitOutcome, + /// Open a statement subscription. It starts with no filter attached, and therefore doesn't + /// generate any notification until one is added. + /// + /// Never fails with `-32800`: the number of statement subscriptions per client isn't capped. The + /// specification requires accepting at least two and only permits erroring beyond that. + statement_unstable_subscribe() -> Cow<'a, str>, + /// Attach a filter to a `statement_unstable_subscribe` subscription. A light client keeps no + /// statement store, so it replays no statement and emits `replayDone` right away. + /// + /// Two consequences of that emptiness, both diverging from what the specification assumes of a + /// full node: + /// + /// - `replayDone` doesn't mean the client holds what the server held. The statements matching the + /// new filter arrive afterwards as `newStatements`, once peers learn about the updated topic + /// affinity. + /// - Statements already delivered to this subscription are never re-announced under the new + /// filter id, because re-reporting them is what the at-most-once guarantee forbids. Obtaining a + /// full backlog for a filter requires a fresh subscription. + statement_unstable_add_filter( + subscription: Cow<'a, str>, + #[rename = "topicFilter"] topic_filter: StatementTopicFilter + ) -> StatementAddFilterResult<'a>, + /// Detach a filter from a `statement_unstable_subscribe` subscription. + statement_unstable_remove_filter( + subscription: Cow<'a, str>, + #[rename = "filterId"] filter_id: Cow<'a, str> + ) -> (), + /// Stop a subscription started with `statement_unstable_subscribe`, together with all the + /// filters attached to it. + statement_unstable_unsubscribe(subscription: Cow<'a, str>) -> (), + // The functions below are experimental and are defined in the document https://github.com/paritytech/json-rpc-interface-spec/ chainHead_v1_body( #[rename = "followSubscription"] follow_subscription: Cow<'a, str>, @@ -566,6 +606,7 @@ define_methods! { // Statement notification sent when statements matching subscribed topics are received. statement_statement(subscription: Cow<'a, str>, result: StatementEvent) -> (), + statement_unstable_subscribeEvent(subscription: Cow<'a, str>, result: StatementSubscribeEvent<'a>) -> (), } #[derive(Clone, PartialEq, Eq, Hash)] @@ -1129,6 +1170,145 @@ pub enum InternalError { NoConnectedPeers, } +/// Outcome of a [`MethodCall::statement_unstable_submit`]. +/// +/// A light client keeps no statement store, so the specification's store-dependent statuses are +/// omitted: `known` and every `rejected` reason. `known` is in any case unreachable through local RPC +/// submission in polkadot-sdk too, a locally-sourced statement always being resubmittable. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(tag = "status", rename_all = "camelCase")] +pub enum StatementSubmitOutcome { + /// The statement passed validation and was broadcast to at least one peer. + New, + /// The statement failed validation. + Invalid(StatementSubmitInvalidReason), +} + +/// Reason of a [`StatementSubmitOutcome::Invalid`]. +/// +/// The specification's `badProof` is omitted: telling a bad proof from a good one means verifying a +/// signature, more CPU than a light client should spend on a submission, so only the presence of a +/// proof is checked. The trade-off is that a statement carrying an invalid signature is relayed, and +/// the peers that do verify it answer with a reputation penalty. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(tag = "reason", rename_all = "camelCase")] +pub enum StatementSubmitInvalidReason { + /// The statement has no authenticity proof. + NoProof, + /// The encoded statement exceeds the maximum allowed size. + EncodingTooLarge { + /// Size in bytes of the submitted encoding. + #[serde(rename = "submittedSize")] + submitted_size: usize, + /// Maximum allowed size in bytes. + #[serde(rename = "maxSize")] + max_size: usize, + }, + /// The statement's expiry is in the past. + AlreadyExpired, +} + +/// Topic filter accepted by [`MethodCall::statement_unstable_add_filter`]. +/// +/// Only `"any"` and `{"matchAll": [...]}` are accepted, the specification defining no other +/// variant. A `matchAll` must carry between 1 and 4 topics. +#[derive(Debug, Clone)] +pub struct StatementTopicFilter(pub TopicFilter); + +impl serde::Serialize for StatementTopicFilter { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + self.0.serialize(serializer) + } +} + +impl<'de> serde::Deserialize<'de> for StatementTopicFilter { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + use serde::de::Error; + + // `TopicFilter` already rejects unknown variants, non-32-byte topics, and a `matchAll` + // outside of 1 to 4 topics. Only `matchAny`, which the specification doesn't define, is + // left to turn down. + match TopicFilter::deserialize(deserializer)? { + TopicFilter::MatchAny(_) => Err(D::Error::custom( + r#"unknown filter key: matchAny, expected "matchAll""#, + )), + filter => Ok(StatementTopicFilter(filter)), + } + } +} + +/// Return value of [`MethodCall::statement_unstable_add_filter`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum StatementAddFilterResult<'a> { + /// The filter was attached. Contains the opaque string identifying it within the subscription. + FilterId(Cow<'a, str>), + /// The subscription can't accept another filter. No filter was attached. + LimitReached, +} + +impl<'a> serde::Serialize for StatementAddFilterResult<'a> { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + use serde::ser::SerializeMap; + + match self { + StatementAddFilterResult::FilterId(filter_id) => serializer.serialize_str(filter_id), + StatementAddFilterResult::LimitReached => { + let mut map = serializer.serialize_map(Some(1))?; + map.serialize_entry("result", "limitReached")?; + map.end() + } + } + } +} + +/// Notification event of a [`MethodCall::statement_unstable_subscribe`] subscription. +/// +/// A light client keeps no statement store, so `replayStatements` is never emitted: every statement +/// it learns about is reported through `newStatements`. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(tag = "event", rename_all = "camelCase")] +pub enum StatementSubscribeEvent<'a> { + /// Statements that were already in the store when a filter was attached. + ReplayStatements { + /// Filter that produced this batch. + #[serde(rename = "filterId")] + filter_id: Cow<'a, str>, + /// Never empty. + statements: Vec, + }, + /// The replay caused by attaching a filter is complete. + ReplayDone { + /// Filter whose replay completed. + #[serde(rename = "filterId")] + filter_id: Cow<'a, str>, + }, + /// Statements that newly entered the store. + NewStatements { + /// Never empty. + statements: Vec>, + }, + /// The server can no longer maintain the guarantees of the subscription, which is now dead. + Stop, +} + +/// Entry of a [`StatementSubscribeEvent::NewStatements`]. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct StatementSubscribeEventItem<'a> { + pub statement: HexString, + /// Filters of the subscription matching this statement. Never empty, free of duplicates. + #[serde(rename = "filterIds")] + pub filter_ids: Vec>, +} + /// Notification event for statement subscriptions. /// /// JSON format is compatible with polkadot-sdk's `StatementEvent`. @@ -1245,6 +1425,12 @@ impl TopicFilter { pub fn match_all( topics: Vec, ) -> Result { + // The RPC API specification restricts `matchAll` to between 1 and 4 topics. + if topics.is_empty() { + return Err(alloc::string::String::from( + "Too few topics for MatchAll: got 0, min 1", + )); + } if topics.len() > crate::network::codec::MAX_TOPICS { return Err(alloc::format!( "Too many topics for MatchAll: got {}, max {}", @@ -1278,6 +1464,9 @@ impl TopicFilter { statement_topics.iter().any(|t| filter_topics.contains(t)) } TopicFilter::MatchAll(filter_topics) => { + if filter_topics.is_empty() { + return false; + } filter_topics.iter().all(|t| statement_topics.contains(t)) } } @@ -1663,6 +1852,213 @@ mod tests { )); } + #[test] + fn statement_unstable_submit_parse_valid() { + let (id, call) = super::parse_jsonrpc_client_to_server( + r#"{"jsonrpc":"2.0","id":5,"method":"statement_unstable_submit","params":["0x1234"]}"#, + ) + .unwrap(); + + assert_eq!(id, "5"); + assert!(matches!( + call, + super::MethodCall::statement_unstable_submit { .. } + )); + } + + #[test] + fn statement_unstable_subscribe_parse_valid() { + let (id, call) = super::parse_jsonrpc_client_to_server( + r#"{"jsonrpc":"2.0","id":6,"method":"statement_unstable_subscribe","params":[]}"#, + ) + .unwrap(); + + assert_eq!(id, "6"); + assert!(matches!( + call, + super::MethodCall::statement_unstable_subscribe {} + )); + } + + #[test] + fn statement_unstable_unsubscribe_parse_valid() { + let (id, call) = super::parse_jsonrpc_client_to_server( + r#"{"jsonrpc":"2.0","id":7,"method":"statement_unstable_unsubscribe","params":["sub1"]}"#, + ) + .unwrap(); + + assert_eq!(id, "7"); + assert!(matches!( + call, + super::MethodCall::statement_unstable_unsubscribe { .. } + )); + } + + #[test] + fn statement_unstable_add_filter_parse_any() { + let (_, call) = super::parse_jsonrpc_client_to_server( + r#"{"jsonrpc":"2.0","id":8,"method":"statement_unstable_add_filter","params":["sub1","any"]}"#, + ) + .unwrap(); + + assert!(matches!( + call, + super::MethodCall::statement_unstable_add_filter { + topic_filter: super::StatementTopicFilter(super::TopicFilter::Any), + .. + } + )); + } + + #[test] + fn statement_unstable_add_filter_parse_match_all() { + let (_, call) = super::parse_jsonrpc_client_to_server( + r#"{"jsonrpc":"2.0","id":8,"method":"statement_unstable_add_filter","params":["sub1",{"matchAll":["0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"]}]}"#, + ) + .unwrap(); + + assert!(matches!( + call, + super::MethodCall::statement_unstable_add_filter { + topic_filter: super::StatementTopicFilter(super::TopicFilter::MatchAll(_)), + .. + } + )); + } + + #[test] + fn statement_unstable_add_filter_rejects_unspecified_filters() { + // `matchAny` isn't part of the specification, and `matchAll` must carry 1 to 4 topics. + for params in [ + r#"["sub1",{"matchAny":["0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"]}]"#, + r#"["sub1",{"matchAll":[]}]"#, + r#"["sub1","all"]"#, + r#"["sub1",{"matchAll":["0x12"]}]"#, + ] { + let request = alloc::format!( + r#"{{"jsonrpc":"2.0","id":8,"method":"statement_unstable_add_filter","params":{params}}}"# + ); + assert!( + matches!( + super::parse_jsonrpc_client_to_server(&request), + Err(super::ParseClientToServerError::Method { .. }) + ), + "expected {params} to be rejected" + ); + } + } + + #[test] + fn statement_add_filter_result_serialization() { + assert_eq!( + serde_json::to_string(&super::StatementAddFilterResult::FilterId( + alloc::borrow::Cow::Borrowed("7") + )) + .unwrap(), + r#""7""# + ); + assert_eq!( + serde_json::to_string(&super::StatementAddFilterResult::LimitReached).unwrap(), + r#"{"result":"limitReached"}"# + ); + } + + #[test] + fn statement_subscribe_event_serialization() { + use alloc::borrow::Cow; + + assert_eq!( + serde_json::to_string(&super::StatementSubscribeEvent::ReplayStatements { + filter_id: Cow::Borrowed("3"), + statements: vec![super::HexString(vec![0x12, 0x34])], + }) + .unwrap(), + r#"{"event":"replayStatements","filterId":"3","statements":["0x1234"]}"# + ); + assert_eq!( + serde_json::to_string(&super::StatementSubscribeEvent::ReplayDone { + filter_id: Cow::Borrowed("3"), + }) + .unwrap(), + r#"{"event":"replayDone","filterId":"3"}"# + ); + assert_eq!( + serde_json::to_string(&super::StatementSubscribeEvent::NewStatements { + statements: vec![super::StatementSubscribeEventItem { + statement: super::HexString(vec![0xab]), + filter_ids: vec![Cow::Borrowed("1"), Cow::Borrowed("2")], + }], + }) + .unwrap(), + r#"{"event":"newStatements","statements":[{"statement":"0xab","filterIds":["1","2"]}]}"# + ); + assert_eq!( + serde_json::to_string(&super::StatementSubscribeEvent::Stop).unwrap(), + r#"{"event":"stop"}"# + ); + } + + #[test] + fn statement_unstable_unsubscribe_response_is_null() { + assert_eq!( + super::Response::statement_unstable_unsubscribe(()).to_json_response("1"), + r#"{"jsonrpc":"2.0","id":1,"result":null}"# + ); + } + + #[test] + fn statement_submit_outcome_serialization() { + assert_eq!( + serde_json::to_string(&super::StatementSubmitOutcome::New).unwrap(), + r#"{"status":"new"}"# + ); + assert_eq!( + serde_json::to_string(&super::StatementSubmitOutcome::Invalid( + super::StatementSubmitInvalidReason::NoProof + )) + .unwrap(), + r#"{"status":"invalid","reason":"noProof"}"# + ); + assert_eq!( + serde_json::to_string(&super::StatementSubmitOutcome::Invalid( + super::StatementSubmitInvalidReason::AlreadyExpired + )) + .unwrap(), + r#"{"status":"invalid","reason":"alreadyExpired"}"# + ); + assert_eq!( + serde_json::to_string(&super::StatementSubmitOutcome::Invalid( + super::StatementSubmitInvalidReason::EncodingTooLarge { + submitted_size: 2_000_000, + max_size: 1_048_575, + } + )) + .unwrap(), + r#"{"status":"invalid","reason":"encodingTooLarge","submittedSize":2000000,"maxSize":1048575}"# + ); + } + + #[test] + fn statement_submit_outcome_deserialization_round_trip() { + for outcome in [ + super::StatementSubmitOutcome::New, + super::StatementSubmitOutcome::Invalid(super::StatementSubmitInvalidReason::NoProof), + super::StatementSubmitOutcome::Invalid( + super::StatementSubmitInvalidReason::AlreadyExpired, + ), + super::StatementSubmitOutcome::Invalid( + super::StatementSubmitInvalidReason::EncodingTooLarge { + submitted_size: 2_000_000, + max_size: 1_048_575, + }, + ), + ] { + let json = serde_json::to_string(&outcome).unwrap(); + let decoded: super::StatementSubmitOutcome = serde_json::from_str(&json).unwrap(); + assert_eq!(decoded, outcome); + } + } + #[test] fn topic_filter_any_matches_everything() { let filter = super::TopicFilter::Any; @@ -1707,10 +2103,26 @@ mod tests { } #[test] - fn topic_filter_match_all_empty_matches_everything() { - let filter = super::TopicFilter::match_all(Vec::new()).unwrap(); - assert!(filter.matches(&[])); - assert!(filter.matches(&[[1u8; 32]])); + fn topic_filter_match_all_rejects_no_topic() { + assert!(super::TopicFilter::match_all(Vec::new()).is_err()); + } + + #[test] + fn topic_filter_match_all_empty_never_matches() { + // Same as an empty `MatchAny`: a filter requesting no topic matches no statement. + let filter = super::TopicFilter::MatchAll(Vec::new()); + assert!(!filter.matches(&[])); + assert!(!filter.matches(&[[1u8; 32]])); + } + + #[test] + fn statement_subscribe_parse_rejects_empty_match_all() { + assert!(matches!( + super::parse_jsonrpc_client_to_server( + r#"{"jsonrpc":"2.0","id":2,"method":"statement_subscribeStatement","params":[{"matchAll":[]}]}"#, + ), + Err(super::ParseClientToServerError::Method { .. }) + )); } #[test] diff --git a/lib/src/json_rpc/service/client_main_task.rs b/lib/src/json_rpc/service/client_main_task.rs index 17eb75d968..e884ceadf3 100644 --- a/lib/src/json_rpc/service/client_main_task.rs +++ b/lib/src/json_rpc/service/client_main_task.rs @@ -435,6 +435,9 @@ impl ClientMainTask { | methods::MethodCall::system_removeReservedPeer { .. } | methods::MethodCall::system_version { .. } | methods::MethodCall::statement_submit { .. } + | methods::MethodCall::statement_unstable_submit { .. } + | methods::MethodCall::statement_unstable_add_filter { .. } + | methods::MethodCall::statement_unstable_remove_filter { .. } | methods::MethodCall::chainSpec_v1_chainName { .. } | methods::MethodCall::chainSpec_v1_genesisHash { .. } | methods::MethodCall::chainSpec_v1_properties { .. } @@ -470,6 +473,7 @@ impl ClientMainTask { | methods::MethodCall::state_subscribeRuntimeVersion { .. } | methods::MethodCall::state_subscribeStorage { .. } | methods::MethodCall::statement_subscribeStatement { .. } + | methods::MethodCall::statement_unstable_subscribe { .. } | methods::MethodCall::transaction_v1_broadcast { .. } | methods::MethodCall::transactionWatch_v1_submitAndWatch { .. } | methods::MethodCall::sudo_network_unstable_watch { .. } @@ -550,6 +554,7 @@ impl ClientMainTask { | methods::MethodCall::transactionWatch_v1_unwatch { subscription, .. } | methods::MethodCall::sudo_network_unstable_unwatch { subscription, .. } | methods::MethodCall::bitswap_unstable_unstream { subscription, .. } + | methods::MethodCall::statement_unstable_unsubscribe { subscription, .. } | methods::MethodCall::chainHead_v1_unfollow { follow_subscription: subscription, .. @@ -583,6 +588,9 @@ impl ClientMainTask { methods::MethodCall::bitswap_unstable_unstream { .. } => { methods::Response::bitswap_unstable_unstream(()) } + methods::MethodCall::statement_unstable_unsubscribe { + .. + } => methods::Response::statement_unstable_unsubscribe(()), methods::MethodCall::chainHead_v1_unfollow { .. } => { methods::Response::chainHead_v1_unfollow(()) } @@ -613,6 +621,11 @@ impl ClientMainTask { methods::Response::bitswap_unstable_unstream(()) .to_json_response(request_id) } + methods::MethodCall::statement_unstable_unsubscribe { .. } => { + // Per spec: no error if subscription is unknown or already-completed. + methods::Response::statement_unstable_unsubscribe(()) + .to_json_response(request_id) + } _ => parse::build_error_response( request_id, ErrorResponse::InvalidParams(None), @@ -1191,6 +1204,11 @@ impl SubscriptionStartProcess { &self.subscription_id, )) } + methods::MethodCall::statement_unstable_subscribe { .. } => { + methods::Response::statement_unstable_subscribe(Cow::Borrowed( + &self.subscription_id, + )) + } methods::MethodCall::transactionWatch_v1_submitAndWatch { .. } => { methods::Response::transactionWatch_v1_submitAndWatch(Cow::Borrowed( &self.subscription_id, diff --git a/lib/src/network/codec/statement.rs b/lib/src/network/codec/statement.rs index 023af5c7f6..eaa1b0dbcd 100644 --- a/lib/src/network/codec/statement.rs +++ b/lib/src/network/codec/statement.rs @@ -32,6 +32,17 @@ pub const MAX_ANY_TOPICS: usize = 128; /// Maximum number of statements allowed in a single notification. const MAX_STATEMENTS_PER_NOTIFICATION: usize = 10_000; +/// Maximum size in bytes of a single SCALE-encoded statement. +/// +/// Matches the limit polkadot-sdk enforces, and is reported to JSON-RPC clients as the `maxSize` of +/// an `encodingTooLarge` submission, so the two must agree for a client to predict what a full node +/// will accept. +/// +/// Note that a statement of exactly this size does not fit in a V2 notification: that framing adds +/// a message tag on top of the vector length prefix, putting a single-element batch one byte over +/// the 1 MiB notification limit. +pub const MAX_STATEMENT_SIZE: usize = 1024 * 1024 - 1; + const FIELD_PROOF: u8 = 0; const FIELD_DECRYPTION_KEY: u8 = 1; const FIELD_EXPIRY: u8 = 2; diff --git a/light-base/src/json_rpc_service/background.rs b/light-base/src/json_rpc_service/background.rs index 9b646b012e..c28cd5bef6 100644 --- a/light-base/src/json_rpc_service/background.rs +++ b/light-base/src/json_rpc_service/background.rs @@ -254,16 +254,18 @@ impl Background { /// If no update was ever sent, or the last update was more than the configured /// affinity update interval ago, the update fires immediately. /// Otherwise, it fires after the remaining interval. + /// + /// Does nothing when the chain runs without the statement protocol. Subscriptions can still be + /// created in that case, so this is reachable, and there is no affinity to advertise. fn schedule_statement_affinity_update(&mut self) { if self.statement_affinity_stale { return; } + let Some(config) = self.statement_protocol_config.as_ref() else { + return; + }; self.statement_affinity_stale = true; - let interval = self - .statement_protocol_config - .as_ref() - .expect("affinity updates require statement protocol; qed") - .affinity_update_interval(); + let interval = config.affinity_update_interval(); let delay = match &self.last_statement_affinity_update { Some(last) => { let elapsed = self.platform.now() - last.clone(); @@ -833,6 +835,8 @@ pub(super) async fn run( me.statement_affinity_stale = false; me.last_statement_affinity_update = Some(me.platform.now()); + // This wake-up only ever fires from `schedule_statement_affinity_update`, which + // returns without scheduling anything when the config is absent. let combined_filter = me.statement_subscriptions.build_combined_affinity_filter( me.statement_protocol_config .as_ref() @@ -856,24 +860,91 @@ pub(super) async fn run( // The reverse `topic` -> `subscription` index inside `statement_subscriptions` // keeps this proportional to the number of subscriptions sharing a topic with the // incoming statements, rather than the total number of subscriptions. - for (sub_id, matching) in me.statement_subscriptions.matching(&statements) { - let notification = methods::ServerToClient::statement_statement { - subscription: Cow::Owned(sub_id), - result: methods::StatementEvent::NewStatements { - statements: matching, - remaining: None, - }, + let mut stopped_subscriptions = Vec::new(); + + for (sub_id, matched) in me.statement_subscriptions.matching(&statements) { + if !matched.statements.is_empty() { + let notification = match matched.kind { + super::statement::SubscriptionKind::Legacy => { + methods::ServerToClient::statement_statement { + subscription: Cow::Borrowed(&sub_id), + result: methods::StatementEvent::NewStatements { + statements: matched + .statements + .into_iter() + .map(|statement| statement.encoded) + .collect(), + remaining: None, + }, + } + } + super::statement::SubscriptionKind::Unstable => { + methods::ServerToClient::statement_unstable_subscribeEvent { + subscription: Cow::Borrowed(&sub_id), + result: methods::StatementSubscribeEvent::NewStatements { + statements: matched + .statements + .into_iter() + .map(|statement| methods::StatementSubscribeEventItem { + statement: statement.encoded, + filter_ids: statement + .filter_ids + .into_iter() + .map(|id| Cow::Owned(id.to_string())) + .collect(), + }) + .collect(), + }, + } + } + } + .to_json_request_object_parameters(None); + if me.responses_tx.send(notification).await.is_err() { + log!( + &me.platform, + Debug, + &me.log_target, + "Failed to send statement notification: response channel closed" + ); + } } - .to_json_request_object_parameters(None); - if me.responses_tx.send(notification).await.is_err() { + + // The subscription can no longer deduplicate what it has already reported. + // Rather than risk reporting a statement twice, it is killed. + if matched.must_stop { log!( &me.platform, Debug, &me.log_target, - "Failed to send statement notification: response channel closed" + format!( + "Stopping statement subscription {sub_id}: more than {} statements \ + delivered, deduplication can no longer be guaranteed", + me.statement_protocol_config + .as_ref() + .map(|c| c.max_seen_statements().get()) + .unwrap_or(0) + ) ); + + let notification = + methods::ServerToClient::statement_unstable_subscribeEvent { + subscription: Cow::Borrowed(&sub_id), + result: methods::StatementSubscribeEvent::Stop, + } + .to_json_request_object_parameters(None); + let _ = me.responses_tx.send(notification).await; + + stopped_subscriptions.push(sub_id); } } + + for sub_id in stopped_subscriptions { + // Only an unstable subscription ever reports `must_stop`; a legacy one has no + // way of reporting that it can no longer deduplicate and keeps evicting. + me.statement_subscriptions + .remove(&sub_id, super::statement::SubscriptionKind::Unstable); + me.schedule_statement_affinity_update(); + } } WakeUpReason::IncomingJsonRpcRequest(request_json) => { @@ -1001,6 +1072,11 @@ pub(super) async fn run( | methods::MethodCall::chainSpec_v1_genesisHash { .. } | methods::MethodCall::chainSpec_v1_properties { .. } | methods::MethodCall::rpc_methods { .. } + | methods::MethodCall::statement_unstable_submit { .. } + | methods::MethodCall::statement_unstable_subscribe { .. } + | methods::MethodCall::statement_unstable_add_filter { .. } + | methods::MethodCall::statement_unstable_remove_filter { .. } + | methods::MethodCall::statement_unstable_unsubscribe { .. } | methods::MethodCall::sudo_unstable_p2pDiscover { .. } | methods::MethodCall::sudo_unstable_version { .. } | methods::MethodCall::transaction_v1_broadcast { .. } @@ -3050,7 +3126,9 @@ pub(super) async fn run( } methods::MethodCall::statement_unsubscribeStatement { subscription } => { - let existed = me.statement_subscriptions.remove(&subscription); + let existed = me + .statement_subscriptions + .remove(&subscription, super::statement::SubscriptionKind::Legacy); if existed { me.schedule_statement_affinity_update(); @@ -3065,6 +3143,165 @@ pub(super) async fn run( .await; } + methods::MethodCall::statement_unstable_submit { encoded } => { + let network = me.network_service.clone(); + let result = super::statement::validate_and_broadcast_statement_unstable( + &encoded.0, + me.platform.now_from_unix_epoch(), + |bytes| async move { network.broadcast_statement(bytes).await }, + ) + .await; + + let response = match result { + Ok(outcome) => methods::Response::statement_unstable_submit(outcome) + .to_json_response(request_id_json), + Err(super::statement::StatementSubmitError::InvalidEncoding) => { + parse::build_error_response( + request_id_json, + parse::ErrorResponse::InvalidParams(Some( + "The `encoded` parameter doesn't decode into a statement", + )), + None, + ) + } + Err(super::statement::StatementSubmitError::NoConnectedPeers) => { + parse::build_error_response( + request_id_json, + parse::ErrorResponse::InternalError, + Some(r#""No connected peers to broadcast the statement to""#), + ) + } + }; + + let _ = me.responses_tx.send(response).await; + } + + methods::MethodCall::statement_unstable_subscribe {} => { + let subscription_id: String = { + let mut id = [0u8; 32]; + me.randomness.fill_bytes(&mut id); + hex::encode(id) + }; + + // The subscription starts with no filter attached. It therefore matches no + // statement, and the topic affinity advertised to peers is unchanged. + me.statement_subscriptions.insert_empty( + subscription_id.clone(), + me.statement_protocol_config + .as_ref() + .map(|c| c.max_seen_statements()), + ); + + let _ = me + .responses_tx + .send( + methods::Response::statement_unstable_subscribe(Cow::Owned( + subscription_id, + )) + .to_json_response(request_id_json), + ) + .await; + } + + methods::MethodCall::statement_unstable_add_filter { + subscription, + topic_filter, + } => { + let mut added = None; + + let response = match me + .statement_subscriptions + .add_filter(&subscription, topic_filter.0) + { + Ok(filter_id) => { + me.schedule_statement_affinity_update(); + added = Some(filter_id); + methods::Response::statement_unstable_add_filter( + methods::StatementAddFilterResult::FilterId(Cow::Owned( + filter_id.to_string(), + )), + ) + .to_json_response(request_id_json) + } + Err(super::statement::AddFilterError::LimitReached) => { + methods::Response::statement_unstable_add_filter( + methods::StatementAddFilterResult::LimitReached, + ) + .to_json_response(request_id_json) + } + Err(super::statement::AddFilterError::UnknownSubscription) => { + parse::build_error_response( + request_id_json, + parse::ErrorResponse::ApplicationDefined( + -32801, + "unknown subscription", + ), + None, + ) + } + }; + + let _ = me.responses_tx.send(response).await; + + // A light client keeps no statement store, so the replay of a newly-added + // filter runs over an empty snapshot and completes right away. The + // statements that peers send once they learn about the updated topic + // affinity are reported through `newStatements`. + // + // `replayDone` therefore precedes the statements matching the filter rather + // than following them, by up to the affinity update interval. A client + // reading it as "I now hold what the server holds" concludes too early. + if let Some(filter_id) = added { + let notification = + methods::ServerToClient::statement_unstable_subscribeEvent { + subscription, + result: methods::StatementSubscribeEvent::ReplayDone { + filter_id: Cow::Owned(filter_id.to_string()), + }, + } + .to_json_request_object_parameters(None); + let _ = me.responses_tx.send(notification).await; + } + } + + methods::MethodCall::statement_unstable_remove_filter { + subscription, + filter_id, + } => { + if let Some(filter_id) = super::statement::FilterId::from_string(&filter_id) + && me + .statement_subscriptions + .remove_filter(&subscription, filter_id) + { + me.schedule_statement_affinity_update(); + } + + let _ = me + .responses_tx + .send( + methods::Response::statement_unstable_remove_filter(()) + .to_json_response(request_id_json), + ) + .await; + } + + methods::MethodCall::statement_unstable_unsubscribe { subscription } => { + if me + .statement_subscriptions + .remove(&subscription, super::statement::SubscriptionKind::Unstable) + { + me.schedule_statement_affinity_update(); + } + + let _ = me + .responses_tx + .send( + methods::Response::statement_unstable_unsubscribe(()) + .to_json_response(request_id_json), + ) + .await; + } + _method @ (methods::MethodCall::account_nextIndex { .. } | methods::MethodCall::author_hasKey { .. } | methods::MethodCall::author_hasSessionKeys { .. } diff --git a/light-base/src/json_rpc_service/statement.rs b/light-base/src/json_rpc_service/statement.rs index f6d6fd7e4f..acaddd585b 100644 --- a/light-base/src/json_rpc_service/statement.rs +++ b/light-base/src/json_rpc_service/statement.rs @@ -17,9 +17,10 @@ use crate::network_service::{self, BroadcastStatementResult}; use alloc::{string::String, vec::Vec}; -use core::{num::NonZero, time::Duration}; +use core::{fmt, num::NonZero, time::Duration}; use smoldot::json_rpc::methods::{ - HexString, InternalError, InvalidReason, StatementSubmitResult, TopicFilter, + HexString, InternalError, InvalidReason, StatementSubmitInvalidReason, StatementSubmitOutcome, + StatementSubmitResult, TopicFilter, }; use smoldot::network::codec; @@ -99,30 +100,221 @@ where } } +/// Failure of a `statement_unstable_submit` request, reported as a JSON-RPC error rather than a +/// [`StatementSubmitOutcome`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum StatementSubmitError { + /// The submitted bytes don't decode into a statement. + InvalidEncoding, + /// The statement is valid but reached no peer. + NoConnectedPeers, +} + +/// Validates a SCALE-encoded statement and broadcasts it, following the +/// `statement_unstable_submit` semantics. +/// +/// The checks run in the order polkadot-sdk's `Store::submit` applies them — expiry, then size, then +/// proof — so that a client submitting a statement failing several of them is told the same reason a +/// full node would give. Checks needing a local store or chain state are skipped. +pub async fn validate_and_broadcast_statement_unstable( + encoded: &[u8], + now_from_unix_epoch: Duration, + broadcast: F, +) -> Result +where + F: FnOnce(Vec) -> Fut, + Fut: core::future::Future, +{ + let Ok(statement) = codec::decode_statement(encoded) else { + return Err(StatementSubmitError::InvalidEncoding); + }; + + // The most significant 32 bits of `expiry` are the expiration timestamp in seconds since + // the UNIX epoch. A statement expiring exactly now is already expired, the deadline being the + // first instant at which the statement no longer holds. + if now_from_unix_epoch.as_secs() >= statement.expiry >> 32 { + return Ok(StatementSubmitOutcome::Invalid( + StatementSubmitInvalidReason::AlreadyExpired, + )); + } + + if encoded.len() > codec::MAX_STATEMENT_SIZE { + return Ok(StatementSubmitOutcome::Invalid( + StatementSubmitInvalidReason::EncodingTooLarge { + submitted_size: encoded.len(), + max_size: codec::MAX_STATEMENT_SIZE, + }, + )); + } + + if statement.proof.is_none() { + return Ok(StatementSubmitOutcome::Invalid( + StatementSubmitInvalidReason::NoProof, + )); + } + + // Counted on `sent` rather than `total`: a peer can be gossip-connected while its statement + // substream is absent or its notification queue full, in which case the statement reached + // nobody and reporting `new` would tell the client it was published when it wasn't. + let broadcasted = broadcast(encoded.to_vec()).await; + if broadcasted.sent == 0 { + return Err(StatementSubmitError::NoConnectedPeers); + } + + Ok(StatementSubmitOutcome::New) +} + +/// Maximum number of filters attached to a single subscription. +/// +/// Keeps one subscription useful for multiplexing while bounding the per-statement list of matching +/// filter ids. Matches the limit polkadot-sdk enforces, so that a client which stays within it is +/// accepted by both. +pub(super) const MAX_FILTERS_PER_SUBSCRIPTION: usize = 128; + +/// Identifies a filter within the subscription it is attached to. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub(super) struct FilterId(u64); + +impl FilterId { + /// Parses the string representation returned by [`fmt::Display`]. + pub(super) fn from_string(value: &str) -> Option { + value.parse().ok().map(FilterId) + } +} + +impl fmt::Display for FilterId { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + fmt::Display::fmt(&self.0, f) + } +} + +/// Why a filter couldn't be attached to a subscription. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) enum AddFilterError { + /// No subscription with the given id. + UnknownSubscription, + /// The subscription already holds [`MAX_FILTERS_PER_SUBSCRIPTION`] filters. + LimitReached, +} + +/// A statement accepted by a subscription, together with the filters that matched it. +pub(super) struct MatchedStatement { + /// Re-encoded statement. + pub(super) encoded: HexString, + /// Filters of the subscription matching this statement. Never empty, free of duplicates. + pub(super) filter_ids: Vec, +} + +/// Which JSON-RPC API a subscription was created through, and therefore in which format its +/// notifications are sent. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum SubscriptionKind { + /// `statement_subscribeStatement`, holding the single filter given at creation. + Legacy, + /// `statement_unstable_subscribe`, whose filters are attached and detached dynamically. + Unstable, +} + +/// Statements one subscription accepted out of a batch. +pub(super) struct MatchedSubscription { + pub(super) kind: SubscriptionKind, + pub(super) statements: Vec, + + /// `true` if the subscription can no longer guarantee that a statement is reported at most + /// once, and must therefore be stopped after the statements above have been reported. + pub(super) must_stop: bool, +} + +/// Outcome of offering a statement to a subscription. +enum Acceptance { + /// The subscription hasn't been given this statement yet. + New, + /// The subscription has already been given this statement. + Duplicate, + /// The deduplication cache is full. Remembering one more statement would evict the oldest + /// entry, after which that statement could be reported a second time. + CacheFull, +} + +/// One statement subscription: a set of topic filters sharing one deduplication cache. pub(super) struct StatementSubscription { - topic_filter: TopicFilter, + kind: SubscriptionKind, + + filters: hashbrown::HashMap, + + /// Id given to the next filter attached to this subscription. Ids are never reused, so that a + /// removed filter can't be confused with a later one. + next_filter_id: u64, + seen: Option>, } impl StatementSubscription { - pub(super) fn new(topic_filter: TopicFilter, max_seen: Option>) -> Self { + fn new(kind: SubscriptionKind, max_seen: Option>) -> Self { Self { - topic_filter, + kind, + filters: hashbrown::HashMap::with_hasher(Default::default()), + next_filter_id: 0, seen: max_seen .map(|cap| lru::LruCache::with_hasher(cap, fnv::FnvBuildHasher::default())), } } - pub(super) fn accept(&mut self, hash: &[u8; 32], statement: &codec::Statement) -> bool { - if !self.topic_filter.matches(&statement.topics) { - return false; - } - if let Some(seen) = &mut self.seen { - if seen.put(*hash, ()).is_some() { - return false; + /// Attaches `topic_filter` to this subscription and returns the id identifying it. + fn add_filter(&mut self, topic_filter: TopicFilter) -> FilterId { + let filter_id = FilterId(self.next_filter_id); + self.next_filter_id += 1; + self.filters.insert(filter_id, topic_filter); + filter_id + } + + /// Returns the ids of all the filters of this subscription that match `topics`. + fn matching_filters(&self, topics: &[codec::Topic]) -> Vec { + self.filters + .iter() + .filter(|(_, topic_filter)| topic_filter.matches(topics)) + .map(|(filter_id, _)| *filter_id) + .collect() + } + + /// Returns the topics that at least one filter of this subscription references, and whether one + /// of them matches every statement irrespective of its topics. + fn indexed_topics(&self) -> (Vec, bool) { + let mut topics = Vec::new(); + let mut wildcard = false; + + for topic_filter in self.filters.values() { + match indexed_topics(topic_filter) { + None => wildcard = true, + Some(filter_topics) => topics.extend(filter_topics), } } - true + + (topics, wildcard) + } + + /// Records the statement of the given hash as delivered to this subscription. + /// + /// A legacy subscription has no way of reporting that it can no longer deduplicate, and + /// therefore keeps evicting its oldest entry rather than ever returning + /// [`Acceptance::CacheFull`]. + fn accept(&mut self, hash: &[u8; 32]) -> Acceptance { + let Some(seen) = &mut self.seen else { + return Acceptance::New; + }; + + if seen.contains(hash) { + // Refreshes the entry, keeping the statements seen most recently in the cache. + let _ = seen.get(hash); + return Acceptance::Duplicate; + } + + if matches!(self.kind, SubscriptionKind::Unstable) && seen.len() >= seen.cap().get() { + return Acceptance::CacheFull; + } + + seen.put(*hash, ()); + Acceptance::New } } @@ -136,15 +328,15 @@ pub(super) struct StatementSubscriptions { subscriptions: hashbrown::HashMap, /// Reverse index: maps a topic to the IDs of all subscriptions whose filter references it. - /// Only populated for `MatchAny`/`MatchAll` filters with a non-empty topic list. + /// Only `MatchAny`/`MatchAll` filters contribute entries, one per topic they name. by_topic: hashbrown::HashMap< [u8; 32], hashbrown::HashSet, fnv::FnvBuildHasher, >, - /// IDs of subscriptions that match every statement irrespective of its topics: either - /// `TopicFilter::Any`, or a `TopicFilter::MatchAll` whose topic list is empty. + /// IDs of subscriptions having a filter that matches every statement irrespective of its + /// topics, i.e. a `TopicFilter::Any`. wildcard: hashbrown::HashSet, } @@ -164,54 +356,123 @@ impl StatementSubscriptions { self.subscriptions.is_empty() } - /// Inserts a new subscription and updates the reverse index. + /// Inserts a new `statement_unstable_subscribe` subscription, with no filter attached. Until a + /// filter is added, it matches no statement. + pub(super) fn insert_empty(&mut self, id: String, max_seen: Option>) { + self.subscriptions.insert( + id, + StatementSubscription::new(SubscriptionKind::Unstable, max_seen), + ); + } + + /// Inserts a new `statement_subscribeStatement` subscription, holding `topic_filter` as its + /// only filter. pub(super) fn insert( &mut self, id: String, topic_filter: TopicFilter, max_seen: Option>, ) { - match &topic_filter { - TopicFilter::Any => { - self.wildcard.insert(id.clone()); + self.subscriptions.insert( + id.clone(), + StatementSubscription::new(SubscriptionKind::Legacy, max_seen), + ); + self.add_filter_of_any_kind(&id, topic_filter) + .expect("subscription was just inserted and holds no filter yet; qed"); + } + + /// Attaches a filter to an existing `statement_unstable_subscribe` subscription and updates the + /// reverse index. Returns the id identifying the filter within that subscription. + /// + /// A legacy subscription is reported as unknown. Both APIs share this registry and a single + /// subscription-id namespace, so without this check a `statement_subscribeStatement` id would be + /// accepted here, and its holder would start receiving events of an API it never called. + pub(super) fn add_filter( + &mut self, + sub_id: &str, + topic_filter: TopicFilter, + ) -> Result { + if self.subscriptions.get(sub_id).map(|sub| sub.kind) != Some(SubscriptionKind::Unstable) { + return Err(AddFilterError::UnknownSubscription); + } + self.add_filter_of_any_kind(sub_id, topic_filter) + } + + /// [`StatementSubscriptions::add_filter`] without the subscription-kind check, so that a legacy + /// subscription can be given the single filter it is created with. + fn add_filter_of_any_kind( + &mut self, + sub_id: &str, + topic_filter: TopicFilter, + ) -> Result { + // Determined before handing the filter over to the subscription, which then owns it. + let indexed_topics = indexed_topics(&topic_filter); + + let filter_id = { + let sub = self + .subscriptions + .get_mut(sub_id) + .ok_or(AddFilterError::UnknownSubscription)?; + if sub.filters.len() >= MAX_FILTERS_PER_SUBSCRIPTION { + return Err(AddFilterError::LimitReached); } - // An empty `MatchAll` filter matches every statement. - TopicFilter::MatchAll(topics) if topics.is_empty() => { - self.wildcard.insert(id.clone()); + sub.add_filter(topic_filter) + }; + + match indexed_topics { + None => { + self.wildcard.insert(String::from(sub_id)); } - TopicFilter::MatchAll(topics) | TopicFilter::MatchAny(topics) => { + Some(topics) => { for topic in topics { self.by_topic - .entry(*topic) + .entry(topic) .or_insert_with(|| hashbrown::HashSet::with_hasher(Default::default())) - .insert(id.clone()); + .insert(String::from(sub_id)); } } } - self.subscriptions - .insert(id, StatementSubscription::new(topic_filter, max_seen)); + Ok(filter_id) } - /// Removes a subscription and cleans up the reverse index. Returns whether it existed. - pub(super) fn remove(&mut self, id: &str) -> bool { - let Some(sub) = self.subscriptions.remove(id) else { + /// Detaches a filter from a `statement_unstable_subscribe` subscription and updates the reverse + /// index. Returns whether that filter was attached to that subscription. + /// + /// A legacy subscription answers `false`, for the reason given on + /// [`StatementSubscriptions::add_filter`]. Detaching its only filter would otherwise leave it + /// alive and matching nothing, with no way for its holder to tell. + pub(super) fn remove_filter(&mut self, sub_id: &str, filter_id: FilterId) -> bool { + let Some(sub) = self.subscriptions.get_mut(sub_id) else { + return false; + }; + if !matches!(sub.kind, SubscriptionKind::Unstable) { + return false; + } + let Some(removed) = sub.filters.remove(&filter_id) else { return false; }; - match &sub.topic_filter { - TopicFilter::Any => { - self.wildcard.remove(id); - } - TopicFilter::MatchAll(topics) if topics.is_empty() => { - self.wildcard.remove(id); + // The index entries of the removed filter are dropped only where no remaining filter of + // the subscription still references them. + let removed_topics = indexed_topics(&removed); + let (remaining_topics, remaining_wildcard) = sub.indexed_topics(); + + match removed_topics { + None => { + if !remaining_wildcard { + self.wildcard.remove(sub_id); + } } - TopicFilter::MatchAll(topics) | TopicFilter::MatchAny(topics) => { + Some(topics) => { for topic in topics { - if let Some(ids) = self.by_topic.get_mut(topic) { - ids.remove(id); + if remaining_topics.contains(&topic) { + continue; + } + if let Some(ids) = self.by_topic.get_mut(&topic) { + ids.remove(sub_id); if ids.is_empty() { - self.by_topic.remove(topic); + self.by_topic.remove(&topic); } } } @@ -221,10 +482,45 @@ impl StatementSubscriptions { true } + /// Removes a subscription together with all its filters, and cleans up the reverse index. + /// Returns whether a subscription of that kind existed under this id. + /// + /// `kind` guards against one API tearing down the other's subscription, for the reason given on + /// [`StatementSubscriptions::add_filter`]. Cancelling an unstable subscription through the legacy + /// method would otherwise drop it without ever sending the `stop` event its holder waits for. + pub(super) fn remove(&mut self, id: &str, kind: SubscriptionKind) -> bool { + if self.subscriptions.get(id).map(|sub| sub.kind) != Some(kind) { + return false; + } + let Some(sub) = self.subscriptions.remove(id) else { + return false; + }; + + let (topics, wildcard) = sub.indexed_topics(); + + if wildcard { + self.wildcard.remove(id); + } + + for topic in topics { + if let Some(ids) = self.by_topic.get_mut(&topic) { + ids.remove(id); + if ids.is_empty() { + self.by_topic.remove(&topic); + } + } + } + + true + } + pub(super) fn shrink_to_fit(&mut self) { self.subscriptions.shrink_to_fit(); - for ids in self.by_topic.values_mut() { - ids.shrink_to_fit(); + for sub in self.subscriptions.values_mut() { + sub.filters.shrink_to_fit(); + } + for entries in self.by_topic.values_mut() { + entries.shrink_to_fit(); } self.by_topic.shrink_to_fit(); self.wildcard.shrink_to_fit(); @@ -232,14 +528,15 @@ impl StatementSubscriptions { /// Matches a batch of statements against the subscriptions. /// - /// Returns, for every subscription that accepts at least one statement, the list of re-encoded - /// matching statements. Uses the reverse index to only consider subscriptions that either match - /// everything or share a topic with the statement; the precise per-subscription filter and - /// deduplication is then applied via [`StatementSubscription::accept`]. + /// Returns, for every subscription that accepts at least one statement, those statements + /// together with the filters that matched each of them. Uses the reverse index to only consider + /// subscriptions that either match everything or share a topic with the statement; the precise + /// per-filter check and the deduplication are then applied to each candidate. A statement + /// matched by several filters of the same subscription is reported once, carrying all of them. pub(super) fn matching( &mut self, statements: &[([u8; 32], codec::Statement)], - ) -> Vec<(String, Vec)> { + ) -> Vec<(String, MatchedSubscription)> { // Disjoint borrows: `subscriptions` is mutated while `by_topic`/`wildcard` are only read. let Self { subscriptions, @@ -247,8 +544,8 @@ impl StatementSubscriptions { wildcard, } = self; - // Subscription ID -> its matching re-encoded statements. - let mut out: hashbrown::HashMap<&str, Vec, fnv::FnvBuildHasher> = + // Subscription ID -> the statements it accepted. + let mut out: hashbrown::HashMap<&str, MatchedSubscription, fnv::FnvBuildHasher> = hashbrown::HashMap::with_hasher(Default::default()); // Reused across statements to avoid reallocating. let mut candidates: hashbrown::HashSet<&str, fnv::FnvBuildHasher> = @@ -269,20 +566,47 @@ impl StatementSubscriptions { let sub = subscriptions .get_mut(*id) .expect("`candidates` is a subset of `subscriptions`; qed"); - if sub.accept(hash, statement) { - let encoded = encoded.get_or_insert_with(|| { - HexString( - codec::encode_statement(statement) - .expect("re-encoding a decoded statement always succeeds; qed"), - ) - }); - out.entry(*id).or_default().push(encoded.clone()); + let filter_ids = sub.matching_filters(&statement.topics); + if filter_ids.is_empty() { + continue; + } + + match sub.accept(hash) { + Acceptance::Duplicate => {} + Acceptance::New => { + let encoded = encoded.get_or_insert_with(|| { + HexString( + codec::encode_statement(statement) + .expect("re-encoding a decoded statement always succeeds; qed"), + ) + }); + out.entry(*id) + .or_insert_with(|| MatchedSubscription { + kind: sub.kind, + statements: Vec::new(), + must_stop: false, + }) + .statements + .push(MatchedStatement { + encoded: encoded.clone(), + filter_ids, + }); + } + Acceptance::CacheFull => { + out.entry(*id) + .or_insert_with(|| MatchedSubscription { + kind: sub.kind, + statements: Vec::new(), + must_stop: false, + }) + .must_stop = true; + } } } } out.into_iter() - .map(|(id, matching)| (String::from(id), matching)) + .map(|(id, matched)| (String::from(id), matched)) .collect() } @@ -293,12 +617,14 @@ impl StatementSubscriptions { let mut all_topics: Vec<&[u8; 32]> = Vec::new(); for sub in self.subscriptions.values() { - match &sub.topic_filter { - TopicFilter::Any => { - return network_service::AffinityFilter::match_all(config.bloom_seed()); - } - TopicFilter::MatchAll(topics) | TopicFilter::MatchAny(topics) => { - all_topics.extend(topics.iter()); + for topic_filter in sub.filters.values() { + match topic_filter { + TopicFilter::Any => { + return network_service::AffinityFilter::match_all(config.bloom_seed()); + } + TopicFilter::MatchAll(topics) | TopicFilter::MatchAny(topics) => { + all_topics.extend(topics.iter()); + } } } } @@ -311,6 +637,15 @@ impl StatementSubscriptions { } } +/// Returns the topics under which a filter is indexed, or `None` if it matches every statement +/// irrespective of its topics. +fn indexed_topics(topic_filter: &TopicFilter) -> Option> { + match topic_filter { + TopicFilter::Any => None, + TopicFilter::MatchAll(topics) | TopicFilter::MatchAny(topics) => Some(topics.clone()), + } +} + #[cfg(test)] mod tests { use super::*; @@ -340,6 +675,21 @@ mod tests { subs } + /// Builds one `statement_unstable_subscribe` subscription holding `filters`. Unstable + /// subscriptions are the only ones that can hold more than one filter. + fn make_unstable_subscription( + id: &str, + filters: Vec, + max_seen: Option>, + ) -> StatementSubscriptions { + let mut subs = StatementSubscriptions::with_capacity(1); + subs.insert_empty(id.to_string(), max_seen); + for filter in filters { + subs.add_filter(id, filter).unwrap(); + } + subs + } + fn statement_with_topics(topics: Vec<[u8; 32]>) -> codec::Statement { codec::Statement { proof: None, @@ -363,6 +713,145 @@ mod tests { .unwrap() } + const NOW: Duration = Duration::from_secs(1_000); + + /// Expiration timestamp, in the most significant 32 bits, later than [`NOW`]. + const FUTURE_EXPIRY: u64 = 2_000 << 32; + + fn encoded_statement(with_proof: bool, expiry: u64, data: Option>) -> Vec { + codec::encode_statement(&codec::Statement { + proof: with_proof.then_some(codec::Proof::Sr25519 { + signature: [0; 64], + signer: [0; 32], + }), + decryption_key: None, + expiry, + channel: None, + topics: Vec::new(), + data, + }) + .unwrap() + } + + #[test] + fn unstable_submit_invalid_encoding() { + let result = block_on(validate_and_broadcast_statement_unstable( + &[0xff, 0xff], + NOW, + |_| async { unreachable!() }, + )); + assert_eq!(result, Err(StatementSubmitError::InvalidEncoding)); + } + + #[test] + fn unstable_submit_already_expired() { + // The statement also has no proof: the expiry check runs first. + let encoded = encoded_statement(false, 500 << 32, None); + let result = block_on(validate_and_broadcast_statement_unstable( + &encoded, + NOW, + |_| async { unreachable!() }, + )); + assert_eq!( + result, + Ok(StatementSubmitOutcome::Invalid( + StatementSubmitInvalidReason::AlreadyExpired + )) + ); + } + + #[test] + fn unstable_submit_expiry_equal_to_now_is_expired() { + let encoded = encoded_statement(true, NOW.as_secs() << 32, None); + let result = block_on(validate_and_broadcast_statement_unstable( + &encoded, + NOW, + |_| async { unreachable!() }, + )); + assert_eq!( + result, + Ok(StatementSubmitOutcome::Invalid( + StatementSubmitInvalidReason::AlreadyExpired + )) + ); + } + + #[test] + fn unstable_submit_encoding_too_large() { + // The statement also has no proof: the size check runs before the proof check. + let encoded = encoded_statement(false, FUTURE_EXPIRY, Some(vec![0; 1024 * 1024])); + assert!(encoded.len() > codec::MAX_STATEMENT_SIZE); + let result = block_on(validate_and_broadcast_statement_unstable( + &encoded, + NOW, + |_| async { unreachable!() }, + )); + assert_eq!( + result, + Ok(StatementSubmitOutcome::Invalid( + StatementSubmitInvalidReason::EncodingTooLarge { + submitted_size: encoded.len(), + max_size: codec::MAX_STATEMENT_SIZE, + } + )) + ); + } + + #[test] + fn unstable_submit_no_proof() { + let encoded = encoded_statement(false, FUTURE_EXPIRY, None); + let result = block_on(validate_and_broadcast_statement_unstable( + &encoded, + NOW, + |_| async { unreachable!() }, + )); + assert_eq!( + result, + Ok(StatementSubmitOutcome::Invalid( + StatementSubmitInvalidReason::NoProof + )) + ); + } + + #[test] + fn unstable_submit_no_peers() { + let encoded = encoded_statement(true, FUTURE_EXPIRY, None); + let result = block_on(validate_and_broadcast_statement_unstable( + &encoded, + NOW, + |_| async { BroadcastStatementResult { sent: 0, total: 0 } }, + )); + assert_eq!(result, Err(StatementSubmitError::NoConnectedPeers)); + } + + #[test] + fn unstable_submit_reaching_no_peer_is_not_new() { + // Gossip-connected peers whose statement substream is missing or whose queue is full leave + // the statement unsent. Answering `new` would tell the client it was published. + let encoded = encoded_statement(true, FUTURE_EXPIRY, None); + let result = block_on(validate_and_broadcast_statement_unstable( + &encoded, + NOW, + |_| async { BroadcastStatementResult { sent: 0, total: 5 } }, + )); + assert_eq!(result, Err(StatementSubmitError::NoConnectedPeers)); + } + + #[test] + fn unstable_submit_new() { + let encoded = encoded_statement(true, FUTURE_EXPIRY, None); + let expected_bytes = encoded.clone(); + let result = block_on(validate_and_broadcast_statement_unstable( + &encoded, + NOW, + |bytes| async move { + assert_eq!(bytes, expected_bytes); + BroadcastStatementResult { sent: 3, total: 5 } + }, + )); + assert_eq!(result, Ok(StatementSubmitOutcome::New)); + } + #[test] fn validate_and_broadcast_invalid_encoding() { let result = block_on(validate_and_broadcast_statement(&[0xff, 0xff], |_| async { @@ -442,49 +931,116 @@ mod tests { #[test] fn accept_fresh_statement_passes() { - let t1 = [1u8; 32]; - let mut sub = - StatementSubscription::new(TopicFilter::match_any(vec![t1]).unwrap(), NonZero::new(8)); - let stmt = statement_with_topics(vec![t1]); - assert!(sub.accept(&[0xbb; 32], &stmt)); + let mut sub = StatementSubscription::new(SubscriptionKind::Unstable, NonZero::new(8)); + assert!(matches!(sub.accept(&[0xbb; 32]), Acceptance::New)); } #[test] - fn accept_duplicate_returns_false() { - let mut sub = StatementSubscription::new(TopicFilter::Any, NonZero::new(8)); - let stmt = statement_with_topics(vec![]); + fn accept_reports_duplicates() { + let mut sub = StatementSubscription::new(SubscriptionKind::Unstable, NonZero::new(8)); let hash = [0xcc; 32]; - assert!(sub.accept(&hash, &stmt)); - assert!(!sub.accept(&hash, &stmt)); + assert!(matches!(sub.accept(&hash), Acceptance::New)); + assert!(matches!(sub.accept(&hash), Acceptance::Duplicate)); + } + + #[test] + fn accept_reports_a_full_cache_on_an_unstable_subscription() { + let mut sub = StatementSubscription::new(SubscriptionKind::Unstable, NonZero::new(2)); + let h_a = [0xa; 32]; + let h_b = [0xb; 32]; + let h_c = [0xc; 32]; + + assert!(matches!(sub.accept(&h_a), Acceptance::New)); + assert!(matches!(sub.accept(&h_b), Acceptance::New)); + // Remembering a third statement would evict `h_a`, after which `h_a` could be reported a + // second time. The subscription reports that it can't deduplicate any further instead. + assert!(matches!(sub.accept(&h_c), Acceptance::CacheFull)); + // The statements already remembered are still recognised as duplicates. + assert!(matches!(sub.accept(&h_a), Acceptance::Duplicate)); } #[test] - fn accept_lru_eviction_allows_resubmit() { - let mut sub = StatementSubscription::new(TopicFilter::Any, NonZero::new(2)); - let stmt = statement_with_topics(vec![]); + fn accept_keeps_evicting_on_a_legacy_subscription() { + // The legacy API has no way of reporting that deduplication stopped, so its subscriptions + // keep evicting their oldest entry. + let mut sub = StatementSubscription::new(SubscriptionKind::Legacy, NonZero::new(2)); let h_a = [0xa; 32]; let h_b = [0xb; 32]; let h_c = [0xc; 32]; - assert!(sub.accept(&h_a, &stmt)); - assert!(sub.accept(&h_b, &stmt)); - // Inserting a third eviction-capacity 2 item evicts h_a (oldest). - assert!(sub.accept(&h_c, &stmt)); - // h_a was evicted: it is accepted again as if fresh. - assert!(sub.accept(&h_a, &stmt)); + assert!(matches!(sub.accept(&h_a), Acceptance::New)); + assert!(matches!(sub.accept(&h_b), Acceptance::New)); + assert!(matches!(sub.accept(&h_c), Acceptance::New)); + // `h_a` was evicted: it is accepted again as if fresh. + assert!(matches!(sub.accept(&h_a), Acceptance::New)); + } + + #[test] + fn accept_without_a_cache_never_deduplicates() { + let mut sub = StatementSubscription::new(SubscriptionKind::Unstable, None); + let hash = [0xdd; 32]; + assert!(matches!(sub.accept(&hash), Acceptance::New)); + assert!(matches!(sub.accept(&hash), Acceptance::New)); } #[test] fn dedup_is_per_subscription() { - let mut sub_a = StatementSubscription::new(TopicFilter::Any, NonZero::new(8)); - let mut sub_b = StatementSubscription::new(TopicFilter::Any, NonZero::new(8)); - let stmt = statement_with_topics(vec![]); + let mut sub_a = StatementSubscription::new(SubscriptionKind::Unstable, NonZero::new(8)); + let mut sub_b = StatementSubscription::new(SubscriptionKind::Unstable, NonZero::new(8)); let hash = [0xee; 32]; - assert!(sub_a.accept(&hash, &stmt)); - assert!(!sub_a.accept(&hash, &stmt)); + assert!(matches!(sub_a.accept(&hash), Acceptance::New)); + assert!(matches!(sub_a.accept(&hash), Acceptance::Duplicate)); // Same hash on a different subscription is still fresh: caches are independent. - assert!(sub_b.accept(&hash, &stmt)); + assert!(matches!(sub_b.accept(&hash), Acceptance::New)); + } + + #[test] + fn matching_filters_returns_every_match() { + let t1 = [1u8; 32]; + let t2 = [2u8; 32]; + let mut sub = StatementSubscription::new(SubscriptionKind::Unstable, None); + let f1 = sub.add_filter(TopicFilter::match_any(vec![t1]).unwrap()); + let f2 = sub.add_filter(TopicFilter::match_any(vec![t2]).unwrap()); + let any = sub.add_filter(TopicFilter::Any); + + let sorted = |mut ids: Vec| { + ids.sort_unstable(); + ids + }; + + // Each topic matches its own filter plus the wildcard one. + assert_eq!(sorted(sub.matching_filters(&[t1])), sorted(vec![f1, any])); + assert_eq!(sorted(sub.matching_filters(&[t2])), sorted(vec![f2, any])); + // An unrelated topic matches the wildcard filter only. + assert_eq!(sub.matching_filters(&[[9u8; 32]]), vec![any]); + } + + #[test] + fn indexed_topics_gathers_every_filter() { + let t1 = [1u8; 32]; + let t2 = [2u8; 32]; + let mut sub = StatementSubscription::new(SubscriptionKind::Unstable, None); + sub.add_filter(TopicFilter::match_any(vec![t1]).unwrap()); + sub.add_filter(TopicFilter::match_all(vec![t2]).unwrap()); + + let (mut topics, wildcard) = sub.indexed_topics(); + topics.sort_unstable(); + assert_eq!(topics, vec![t1, t2]); + assert!(!wildcard); + + // A single wildcard filter is enough to flag the whole subscription. + sub.add_filter(TopicFilter::Any); + assert!(sub.indexed_topics().1); + } + + #[test] + fn filter_ids_are_never_reused() { + let mut sub = StatementSubscription::new(SubscriptionKind::Unstable, None); + let first = sub.add_filter(TopicFilter::Any); + sub.filters.remove(&first); + let second = sub.add_filter(TopicFilter::Any); + assert_ne!(first, second); } /// Builds a `(hash, statement)` batch entry from a list of topics. @@ -493,7 +1049,7 @@ mod tests { } /// Collects the IDs of all subscriptions that matched at least once. - fn matched_ids(matches: &[(String, Vec)]) -> Vec { + fn matched_ids(matches: &[(String, MatchedSubscription)]) -> Vec { let mut ids: Vec = matches.iter().map(|(id, _)| id.clone()).collect(); ids.sort(); ids @@ -519,23 +1075,14 @@ mod tests { #[test] fn matching_wildcard_filters_match_every_statement() { - // `Any` and an empty `MatchAll` both match every statement, with or without topics. - let mut subs = make_subscriptions(vec![ - ("any", TopicFilter::Any, None), - ("all", TopicFilter::match_all(vec![]).unwrap(), None), - ]); + // `Any` matches every statement, with or without topics. + let mut subs = make_subscriptions(vec![("any", TopicFilter::Any, None)]); let matches = subs.matching(&[batch_entry(0x01, vec![[7u8; 32]])]); - assert_eq!( - matched_ids(&matches), - vec!["all".to_string(), "any".to_string()] - ); + assert_eq!(matched_ids(&matches), vec!["any".to_string()]); let matches = subs.matching(&[batch_entry(0x02, vec![])]); - assert_eq!( - matched_ids(&matches), - vec!["all".to_string(), "any".to_string()] - ); + assert_eq!(matched_ids(&matches), vec!["any".to_string()]); } #[test] @@ -582,9 +1129,9 @@ mod tests { )]); let entry = batch_entry(0xaa, vec![t1]); - let matches = subs.matching(&[entry.clone()]); + let matches = subs.matching(core::slice::from_ref(&entry)); assert_eq!(matches.len(), 1); - assert_eq!(matches[0].1.len(), 1); + assert_eq!(matches[0].1.statements.len(), 1); // The same statement hash is deduplicated and produces no further notification. let matches = subs.matching(&[entry]); @@ -600,7 +1147,7 @@ mod tests { let matches = subs.matching(&[batch_entry(0x01, vec![t1]), batch_entry(0x02, vec![t1])]); assert_eq!(matches.len(), 1); assert_eq!(matches[0].0, "a"); - assert_eq!(matches[0].1.len(), 2); + assert_eq!(matches[0].1.statements.len(), 2); } #[test] @@ -609,11 +1156,347 @@ mod tests { let mut subs = make_subscriptions(vec![("a", TopicFilter::match_any(vec![t1]).unwrap(), None)]); - assert!(subs.remove("a")); - assert!(!subs.remove("a")); + assert!(subs.remove("a", SubscriptionKind::Legacy)); + assert!(!subs.remove("a", SubscriptionKind::Legacy)); assert!(subs.is_empty()); // The topic entry must have been cleaned up, so a matching statement finds nothing. let matches = subs.matching(&[batch_entry(0xaa, vec![t1])]); assert!(matches.is_empty()); } + + #[test] + fn insert_empty_matches_nothing_until_a_filter_is_added() { + let t1 = [1u8; 32]; + let mut subs = StatementSubscriptions::with_capacity(1); + subs.insert_empty("a".to_string(), None); + + assert!(!subs.is_empty()); + // No filter attached yet, so no statement can match. + let matches = subs.matching(&[batch_entry(0x01, vec![t1]), batch_entry(0x02, vec![])]); + assert!(matches.is_empty()); + + subs.add_filter("a", TopicFilter::match_any(vec![t1]).unwrap()) + .unwrap(); + let matches = subs.matching(&[batch_entry(0x03, vec![t1])]); + assert_eq!(matched_ids(&matches), vec!["a".to_string()]); + } + + #[test] + fn insert_empty_contributes_no_topic_affinity() { + let config = test_config(); + let mut subs = StatementSubscriptions::with_capacity(1); + subs.insert_empty("a".to_string(), None); + + let filter = subs.build_combined_affinity_filter(&config); + assert!(!filter.contains(&[1u8; 32])); + } + + #[test] + fn add_filter_on_unknown_subscription_is_reported() { + let mut subs = make_subscriptions(vec![]); + assert_eq!( + subs.add_filter("nope", TopicFilter::Any), + Err(AddFilterError::UnknownSubscription) + ); + } + + #[test] + fn add_filter_stops_at_the_limit() { + let mut subs = StatementSubscriptions::with_capacity(1); + subs.insert_empty("a".to_string(), None); + + for _ in 0..MAX_FILTERS_PER_SUBSCRIPTION { + subs.add_filter("a", TopicFilter::Any).unwrap(); + } + assert_eq!( + subs.add_filter("a", TopicFilter::Any), + Err(AddFilterError::LimitReached) + ); + } + + #[test] + fn the_unstable_api_cannot_reach_a_legacy_subscription() { + let t1 = [1u8; 32]; + let mut subs = + make_subscriptions(vec![("a", TopicFilter::match_any(vec![t1]).unwrap(), None)]); + + // Both APIs share one id namespace, so a legacy id must read as unknown here rather than + // let its holder be widened, stripped, or torn down through the other API. + assert_eq!( + subs.add_filter("a", TopicFilter::Any), + Err(AddFilterError::UnknownSubscription) + ); + assert!(!subs.remove_filter("a", FilterId(0))); + assert!(!subs.remove("a", SubscriptionKind::Unstable)); + + // The subscription is untouched: still alive, still matching only its own topic. + let matches = subs.matching(&[batch_entry(0x01, vec![t1])]); + assert_eq!(matched_ids(&matches), vec!["a".to_string()]); + let matches = subs.matching(&[batch_entry(0x02, vec![[9u8; 32]])]); + assert!(matches.is_empty()); + } + + #[test] + fn the_legacy_api_cannot_reach_an_unstable_subscription() { + let mut subs = make_unstable_subscription("a", vec![TopicFilter::Any], None); + + // Cancelling it here would drop it without ever sending the `stop` event its holder waits + // for. + assert!(!subs.remove("a", SubscriptionKind::Legacy)); + assert!(!subs.is_empty()); + assert!(subs.remove("a", SubscriptionKind::Unstable)); + assert!(subs.is_empty()); + } + + #[test] + fn remove_filter_keeps_the_topics_of_the_other_filters() { + let t1 = [1u8; 32]; + let t2 = [2u8; 32]; + let mut subs = StatementSubscriptions::with_capacity(1); + subs.insert_empty("a".to_string(), None); + let f1 = subs + .add_filter("a", TopicFilter::match_any(vec![t1, t2]).unwrap()) + .unwrap(); + subs.add_filter("a", TopicFilter::match_any(vec![t2]).unwrap()) + .unwrap(); + + assert!(subs.remove_filter("a", f1)); + // `t2` is still referenced by the remaining filter, `t1` is not. + let matches = subs.matching(&[batch_entry(0x01, vec![t2])]); + assert_eq!(matched_ids(&matches), vec!["a".to_string()]); + let matches = subs.matching(&[batch_entry(0x02, vec![t1])]); + assert!(matches.is_empty()); + } + + #[test] + fn remove_filter_reports_unknown_ids() { + let mut subs = StatementSubscriptions::with_capacity(1); + subs.insert_empty("a".to_string(), None); + let f1 = subs.add_filter("a", TopicFilter::Any).unwrap(); + + assert!(!subs.remove_filter("nope", f1)); + assert!(subs.remove_filter("a", f1)); + // Removing twice reports that nothing was attached anymore. + assert!(!subs.remove_filter("a", f1)); + } + + #[test] + fn matching_reports_the_filters_that_matched() { + let t1 = [1u8; 32]; + let t2 = [2u8; 32]; + let mut subs = StatementSubscriptions::with_capacity(1); + subs.insert_empty("a".to_string(), None); + let f1 = subs + .add_filter("a", TopicFilter::match_any(vec![t1]).unwrap()) + .unwrap(); + let f2 = subs + .add_filter("a", TopicFilter::match_any(vec![t2]).unwrap()) + .unwrap(); + + // Both filters match, so both ids are reported for the single accepted statement. + let matches = subs.matching(&[batch_entry(0x01, vec![t1, t2])]); + assert_eq!(matches.len(), 1); + assert_eq!(matches[0].1.statements.len(), 1); + let mut filter_ids = matches[0].1.statements[0].filter_ids.clone(); + filter_ids.sort_unstable(); + let mut expected = vec![f1, f2]; + expected.sort_unstable(); + assert_eq!(filter_ids, expected); + } + + #[test] + fn matching_reports_the_subscription_kind() { + let t1 = [1u8; 32]; + let mut subs = make_subscriptions(vec![( + "legacy", + TopicFilter::match_any(vec![t1]).unwrap(), + None, + )]); + subs.insert_empty("unstable".to_string(), None); + subs.add_filter("unstable", TopicFilter::match_any(vec![t1]).unwrap()) + .unwrap(); + + let matches = subs.matching(&[batch_entry(0x01, vec![t1])]); + for (id, matched) in matches { + match id.as_str() { + "legacy" => assert_eq!(matched.kind, SubscriptionKind::Legacy), + "unstable" => assert_eq!(matched.kind, SubscriptionKind::Unstable), + other => panic!("unexpected subscription {other}"), + } + } + } + + #[test] + fn matching_reports_a_subscription_that_must_stop() { + let mut subs = StatementSubscriptions::with_capacity(1); + subs.insert_empty("a".to_string(), NonZero::new(2)); + subs.add_filter("a", TopicFilter::Any).unwrap(); + + // Fills the deduplication cache exactly. + let matches = subs.matching(&[batch_entry(0x01, vec![]), batch_entry(0x02, vec![])]); + assert_eq!(matches[0].1.statements.len(), 2); + assert!(!matches[0].1.must_stop); + + // A third statement can't be remembered, so the subscription must be stopped instead of + // risking a second report of one of the first two. + let matches = subs.matching(&[batch_entry(0x03, vec![])]); + assert_eq!(matches.len(), 1); + assert!(matches[0].1.statements.is_empty()); + assert!(matches[0].1.must_stop); + } + + #[test] + fn matching_delivers_what_precedes_a_full_cache() { + let mut subs = StatementSubscriptions::with_capacity(1); + subs.insert_empty("a".to_string(), NonZero::new(2)); + subs.add_filter("a", TopicFilter::Any).unwrap(); + + // The batch fills the cache and then overflows within the same call: the statements that + // fit are reported, and the subscription is flagged for stopping. + let matches = subs.matching(&[ + batch_entry(0x01, vec![]), + batch_entry(0x02, vec![]), + batch_entry(0x03, vec![]), + ]); + assert_eq!(matches[0].1.statements.len(), 2); + assert!(matches[0].1.must_stop); + } + + #[test] + fn matching_never_stops_a_legacy_subscription() { + let mut subs = make_subscriptions(vec![("a", TopicFilter::Any, NonZero::new(2))]); + + let matches = subs.matching(&[ + batch_entry(0x01, vec![]), + batch_entry(0x02, vec![]), + batch_entry(0x03, vec![]), + ]); + assert_eq!(matches[0].1.statements.len(), 3); + assert!(!matches[0].1.must_stop); + } + + #[test] + fn filter_id_string_round_trip() { + let mut sub = StatementSubscription::new(SubscriptionKind::Unstable, None); + let filter_id = sub.add_filter(TopicFilter::Any); + assert_eq!( + FilterId::from_string(&filter_id.to_string()), + Some(filter_id) + ); + assert_eq!(FilterId::from_string("not-a-number"), None); + } + + #[test] + fn matching_accepts_a_statement_matched_by_any_filter() { + let t1 = [1u8; 32]; + let t2 = [2u8; 32]; + let mut subs = make_unstable_subscription( + "a", + vec![ + TopicFilter::match_any(vec![t1]).unwrap(), + TopicFilter::match_any(vec![t2]).unwrap(), + ], + None, + ); + + // Each filter pulls in the statements of its own topic. + let matches = subs.matching(&[batch_entry(0x01, vec![t1]), batch_entry(0x02, vec![t2])]); + assert_eq!(matched_ids(&matches), vec!["a".to_string()]); + assert_eq!(matches[0].1.statements.len(), 2); + + // A statement matching none of the filters is not accepted. + let matches = subs.matching(&[batch_entry(0x03, vec![[9u8; 32]])]); + assert!(matches.is_empty()); + } + + #[test] + fn matching_reports_a_statement_once_per_subscription() { + let t1 = [1u8; 32]; + let t2 = [2u8; 32]; + let mut subs = make_unstable_subscription( + "a", + vec![ + TopicFilter::match_any(vec![t1]).unwrap(), + TopicFilter::match_any(vec![t2]).unwrap(), + TopicFilter::Any, + ], + NonZero::new(8), + ); + + // All three filters match, yet the statement is reported a single time. + let matches = subs.matching(&[batch_entry(0xaa, vec![t1, t2])]); + assert_eq!(matches.len(), 1); + assert_eq!(matches[0].1.statements.len(), 1); + } + + #[test] + fn matching_dedups_a_filter_indexed_under_several_topics() { + let t1 = [1u8; 32]; + let t2 = [2u8; 32]; + // Without a deduplication cache, a filter indexed under both topics would otherwise report + // the statement once per topic it shares with it. + let mut subs = make_subscriptions(vec![( + "a", + TopicFilter::match_any(vec![t1, t2]).unwrap(), + None, + )]); + + let matches = subs.matching(&[batch_entry(0xaa, vec![t1, t2])]); + assert_eq!(matches.len(), 1); + assert_eq!(matches[0].1.statements.len(), 1); + } + + #[test] + fn remove_cleans_up_every_filter() { + let t1 = [1u8; 32]; + let t2 = [2u8; 32]; + let mut subs = make_unstable_subscription( + "a", + vec![ + TopicFilter::match_any(vec![t1]).unwrap(), + TopicFilter::match_any(vec![t2]).unwrap(), + TopicFilter::Any, + ], + None, + ); + + assert!(subs.remove("a", SubscriptionKind::Unstable)); + assert!(subs.is_empty()); + // Neither the topic entries nor the wildcard entry may survive. + let matches = subs.matching(&[batch_entry(0xaa, vec![t1]), batch_entry(0xbb, vec![t2])]); + assert!(matches.is_empty()); + } + + #[test] + fn affinity_covers_every_filter_of_a_subscription() { + let config = test_config(); + let t1 = [1u8; 32]; + let t2 = [2u8; 32]; + let subs = make_unstable_subscription( + "a", + vec![ + TopicFilter::match_any(vec![t1]).unwrap(), + TopicFilter::match_any(vec![t2]).unwrap(), + ], + None, + ); + + let filter = subs.build_combined_affinity_filter(&config); + assert!(filter.contains(&t1)); + assert!(filter.contains(&t2)); + } + + #[test] + fn affinity_matches_all_when_one_filter_is_any() { + let config = test_config(); + let t1 = [1u8; 32]; + let subs = make_unstable_subscription( + "a", + vec![TopicFilter::match_any(vec![t1]).unwrap(), TopicFilter::Any], + None, + ); + + let filter = subs.build_combined_affinity_filter(&config); + assert!(filter.contains(&[99u8; 32])); + } }