rpc-v2/tx: Implement transaction_unstable_broadcast and transaction_unstable_stop

prdoc/pr_3079.prdoc

@@ -0,0 +1,15 @@
+# Schema: Polkadot SDK PRDoc Schema (prdoc) v1.0.0
+# See doc at https://raw.githubusercontent.com/paritytech/polkadot-sdk/master/prdoc/schema_user.json
+
+title: Implement transaction_unstable_broadcast and transaction_unstable_stop
+
+doc:
+  - audience: Node Dev
+    description: |
+      A new RPC class is added to handle transactions. The `transaction_unstable_broadcast` broadcasts
+      the provided transaction to the peers of the node, until the `transaction_unstable_stop` is called.
+      The APIs are marked as unstable and subject to change in the future.
+      To know if the transaction was added to the chain, users can decode the bodies of announced finalized blocks.
+      This is a low-level approach for `transactionWatch_unstable_submitAndWatch`.
+
+crates: [ ]

substrate/client/rpc-spec-v2/Cargo.toml

@@ -41,16 +41,19 @@ tokio = { version = "1.22.0", features = ["sync"] }
 array-bytes = "6.1"
 log = "0.4.17"
 futures-util = { version = "0.3.30", default-features = false }
+rand = "0.8.5"
 
 [dev-dependencies]
 serde_json = "1.0.111"
 tokio = { version = "1.22.0", features = ["macros"] }
 substrate-test-runtime-client = { path = "../../test-utils/runtime/client" }
 substrate-test-runtime = { path = "../../test-utils/runtime" }
+substrate-test-runtime-transaction-pool = { path = "../../test-utils/runtime/transaction-pool" }
 sp-consensus = { path = "../../primitives/consensus/common" }
 sp-externalities = { path = "../../primitives/externalities" }
 sp-maybe-compressed-blob = { path = "../../primitives/maybe-compressed-blob" }
 sc-block-builder = { path = "../block-builder" }
 sc-service = { path = "../service", features = ["test-helpers"] }
 assert_matches = "1.3.0"
 pretty_assertions = "1.2.1"
+sc-transaction-pool = { path = "../transaction-pool" }

substrate/client/rpc-spec-v2/src/chain_head/mod.rs

@@ -23,7 +23,7 @@
 //! Methods are prefixed by `chainHead`.
 
 #[cfg(test)]
-mod test_utils;
+pub mod test_utils;
 #[cfg(test)]
 mod tests;
 

substrate/client/rpc-spec-v2/src/transaction/api.rs

@@ -18,8 +18,8 @@
 
 //! API trait for transactions.
 
-use crate::transaction::event::TransactionEvent;
-use jsonrpsee::proc_macros::rpc;
+use crate::transaction::{error::ErrorBroadcast, event::TransactionEvent};
+use jsonrpsee::{core::RpcResult, proc_macros::rpc};
 use sp_core::Bytes;
 
 #[rpc(client, server)]
@@ -28,10 +28,33 @@ pub trait TransactionApi<Hash: Clone> {
 	///
 	/// See [`TransactionEvent`](crate::transaction::event::TransactionEvent) for details on
 	/// transaction life cycle.
+	///
+	/// # Unstable
+	///
+	/// This method is unstable and subject to change in the future.
 	#[subscription(
 		name = "transactionWatch_unstable_submitAndWatch" => "transactionWatch_unstable_watchEvent",
 		unsubscribe = "transactionWatch_unstable_unwatch",
 		item = TransactionEvent<Hash>,
 	)]
 	fn submit_and_watch(&self, bytes: Bytes);
 }
+
+#[rpc(client, server)]
+pub trait TransactionBroadcastApi<Hash: Clone> {
+	/// Broadcast an extrinsic to the chain.
+	///
+	/// # Unstable
+	///
+	/// This method is unstable and subject to change in the future.
+	#[method(name = "transaction_unstable_broadcast")]
+	fn broadcast(&self, bytes: Bytes) -> RpcResult<Option<String>>;
+
+	/// Broadcast an extrinsic to the chain.
+	///
+	/// # Unstable
+	///
+	/// This method is unstable and subject to change in the future.
+	#[method(name = "transaction_unstable_stop")]
+	fn stop_broadcast(&self, operation_id: String) -> Result<(), ErrorBroadcast>;
+}

substrate/client/rpc-spec-v2/src/transaction/error.rs

@@ -21,6 +21,7 @@
 //! Errors are interpreted as transaction events for subscriptions.
 
 use crate::transaction::event::{TransactionError, TransactionEvent};
+use jsonrpsee::types::error::ErrorObject;
 use sc_transaction_pool_api::error::Error as PoolError;
 use sp_runtime::transaction_validity::InvalidTransaction;
 
@@ -98,3 +99,29 @@ impl<Hash> From<Error> for TransactionEvent<Hash> {
 		}
 	}
 }
+
+/// TransactionBroadcast error.
+#[derive(Debug, thiserror::Error)]
+pub enum ErrorBroadcast {
+	/// The provided operation ID is invalid.
+	#[error("Invalid operation id")]
+	InvalidOperationID,
+}
+
+/// General purpose errors, as defined in
+/// <https://www.jsonrpc.org/specification#error_object>.
+pub mod json_rpc_spec {
+	/// Invalid parameter error.
+	pub const INVALID_PARAM_ERROR: i32 = -32602;
+}
+
+impl From<ErrorBroadcast> for ErrorObject<'static> {
+	fn from(e: ErrorBroadcast) -> Self {
+		let msg = e.to_string();
+
+		match e {
+			ErrorBroadcast::InvalidOperationID =>
+				ErrorObject::owned(json_rpc_spec::INVALID_PARAM_ERROR, msg, None::<()>),
+		}
+	}
+}

substrate/client/rpc-spec-v2/src/transaction/mod.rs

@@ -25,14 +25,19 @@
 //!
 //! Methods are prefixed by `transaction`.
 
+#[cfg(test)]
+mod tests;
+
 pub mod api;
 pub mod error;
 pub mod event;
 pub mod transaction;
+pub mod transaction_broadcast;
 
-pub use api::TransactionApiServer;
+pub use api::{TransactionApiServer, TransactionBroadcastApiServer};
 pub use event::{
 	TransactionBlock, TransactionBroadcasted, TransactionDropped, TransactionError,
 	TransactionEvent,
 };
 pub use transaction::Transaction;
+pub use transaction_broadcast::TransactionBroadcast;

substrate/client/rpc-spec-v2/src/transaction/tests.rs

@@ -0,0 +1,175 @@
+// This file is part of Substrate.
+
+// Copyright (C) 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 <https://www.gnu.org/licenses/>.
+
+use super::*;
+use crate::{
+	chain_head::test_utils::ChainHeadMockClient, hex_string,
+	transaction::TransactionBroadcast as RpcTransactionBroadcast,
+};
+use assert_matches::assert_matches;
+use codec::Encode;
+use futures::Future;
+use jsonrpsee::{core::error::Error, rpc_params, RpcModule};
+use sc_transaction_pool::*;
+use sc_transaction_pool_api::{ChainEvent, MaintainedTransactionPool, TransactionPool};
+use sp_core::testing::TaskExecutor;
+use std::{pin::Pin, sync::Arc, time::Duration};
+use substrate_test_runtime_client::{prelude::*, AccountKeyring::*, Client};
+use substrate_test_runtime_transaction_pool::{uxt, TestApi};
+
+type Block = substrate_test_runtime_client::runtime::Block;
+
+/// Initial Alice account nonce.
+const ALICE_NONCE: u64 = 209;
+
+fn create_basic_pool_with_genesis(
+	test_api: Arc<TestApi>,
+) -> (BasicPool<TestApi, Block>, Pin<Box<dyn Future<Output = ()> + Send>>) {
+	let genesis_hash = {
+		test_api
+			.chain()
+			.read()
+			.block_by_number
+			.get(&0)
+			.map(|blocks| blocks[0].0.header.hash())
+			.expect("there is block 0. qed")
+	};
+	BasicPool::new_test(test_api, genesis_hash, genesis_hash)
+}
+
+fn maintained_pool() -> (BasicPool<TestApi, Block>, Arc<TestApi>, futures::executor::ThreadPool) {
+	let api = Arc::new(TestApi::with_alice_nonce(ALICE_NONCE));
+	let (pool, background_task) = create_basic_pool_with_genesis(api.clone());
+
+	let thread_pool = futures::executor::ThreadPool::new().unwrap();
+	thread_pool.spawn_ok(background_task);
+	(pool, api, thread_pool)
+}
+
+fn setup_api() -> (
+	Arc<TestApi>,
+	Arc<BasicPool<TestApi, Block>>,
+	Arc<ChainHeadMockClient<Client<Backend>>>,
+	RpcModule<
+		TransactionBroadcast<BasicPool<TestApi, Block>, ChainHeadMockClient<Client<Backend>>>,
+	>,
+) {
+	let (pool, api, _) = maintained_pool();
+	let pool = Arc::new(pool);
+
+	let builder = TestClientBuilder::new();
+	let client = Arc::new(builder.build());
+	let client_mock = Arc::new(ChainHeadMockClient::new(client.clone()));
+
+	let tx_api = RpcTransactionBroadcast::new(
+		client_mock.clone(),
+		pool.clone(),
+		Arc::new(TaskExecutor::default()),
+	)
+	.into_rpc();
+
+	(api, pool, client_mock, tx_api)
+}
+
+#[tokio::test]
+async fn tx_broadcast_enters_pool() {
+	let (api, pool, client_mock, tx_api) = setup_api();
+
+	// Start at block 1.
+	let block_1_header = api.push_block(1, vec![], true);
+
+	let uxt = uxt(Alice, ALICE_NONCE);
+	let xt = hex_string(&uxt.encode());
+
+	let operation_id: String =
+		tx_api.call("transaction_unstable_broadcast", rpc_params![&xt]).await.unwrap();
+
+	// Announce block 1 to `transaction_unstable_broadcast`.
+	client_mock.trigger_import_stream(block_1_header).await;
+
+	// Ensure the tx propagated from `transaction_unstable_broadcast` to the transaction pool.
+
+	// TODO: Improve testability by extending the `transaction_unstable_broadcast` with
+	// a middleware trait that intercepts the transaction status for testing.
+	let mut num_retries = 12;
+	while num_retries > 0 && pool.status().ready != 1 {
+		tokio::time::sleep(Duration::from_secs(5)).await;
+		num_retries -= 1;
+	}
+	assert_eq!(1, pool.status().ready);
+	assert_eq!(uxt.encode().len(), pool.status().ready_bytes);
+
+	// Import block 2 with the transaction included.
+	let block_2_header = api.push_block(2, vec![uxt.clone()], true);
+	let block_2 = block_2_header.hash();
+
+	// Announce block 2 to the pool.
+	let event = ChainEvent::NewBestBlock { hash: block_2, tree_route: None };
+	pool.maintain(event).await;
+
+	assert_eq!(0, pool.status().ready);
+
+	// Stop call can still be made.
+	let _: () = tx_api
+		.call("transaction_unstable_stop", rpc_params![&operation_id])
+		.await
+		.unwrap();
+}
+
+#[tokio::test]
+async fn tx_broadcast_invalid_tx() {
+	let (_, pool, _, tx_api) = setup_api();
+
+	// Invalid parameters.
+	let err = tx_api
+		.call::<_, serde_json::Value>("transaction_unstable_broadcast", [1u8])
+		.await
+		.unwrap_err();
+	assert_matches!(err,
+		Error::Call(err) if err.code() == super::error::json_rpc_spec::INVALID_PARAM_ERROR && err.message() == "Invalid params"
+	);
+
+	assert_eq!(0, pool.status().ready);
+
+	// Invalid transaction that cannot be decoded. The broadcast silently exits.
+	let xt = "0xdeadbeef";
+	let operation_id: String =
+		tx_api.call("transaction_unstable_broadcast", rpc_params![&xt]).await.unwrap();
+
+	assert_eq!(0, pool.status().ready);
+
+	// Ensure stop can be called, the tx was decoded and the broadcast future terminated.
+	let _: () = tx_api
+		.call("transaction_unstable_stop", rpc_params![&operation_id])
+		.await
+		.unwrap();
+}
+
+#[tokio::test]
+async fn tx_invalid_stop() {
+	let (_, _, _, tx_api) = setup_api();
+
+	// Make an invalid stop call.
+	let err = tx_api
+		.call::<_, serde_json::Value>("transaction_unstable_stop", ["invalid_operation_id"])
+		.await
+		.unwrap_err();
+	assert_matches!(err,
+		Error::Call(err) if err.code() == super::error::json_rpc_spec::INVALID_PARAM_ERROR && err.message() == "Invalid operation id"
+	);
+}

substrate/client/rpc-spec-v2/src/transaction/transaction.rs

@@ -29,21 +29,18 @@ use crate::{
 	},
 	SubscriptionTaskExecutor,
 };
+use codec::Decode;
+use futures::{StreamExt, TryFutureExt};
 use jsonrpsee::{core::async_trait, types::error::ErrorObject, PendingSubscriptionSink};
+use sc_rpc::utils::pipe_from_stream;
 use sc_transaction_pool_api::{
 	error::IntoPoolError, BlockHash, TransactionFor, TransactionPool, TransactionSource,
 	TransactionStatus,
 };
-use std::sync::Arc;
-
-use sc_rpc::utils::pipe_from_stream;
-use sp_api::ProvideRuntimeApi;
 use sp_blockchain::HeaderBackend;
 use sp_core::Bytes;
 use sp_runtime::traits::Block as BlockT;
-
-use codec::Decode;
-use futures::{StreamExt, TryFutureExt};
+use std::sync::Arc;
 
 /// An API for transaction RPC calls.
 pub struct Transaction<Pool, Client> {
@@ -82,7 +79,7 @@ where
 	Pool: TransactionPool + Sync + Send + 'static,
 	Pool::Hash: Unpin,
 	<Pool::Block as BlockT>::Hash: Unpin,
-	Client: HeaderBackend<Pool::Block> + ProvideRuntimeApi<Pool::Block> + Send + Sync + 'static,
+	Client: HeaderBackend<Pool::Block> + Send + Sync + 'static,
 {
 	fn submit_and_watch(&self, pending: PendingSubscriptionSink, xt: Bytes) {
 		let client = self.client.clone();