rpc-v2/tx: Implement transaction_unstable_broadcast and `transactio…

…n_unstable_stop` (#3079)

This PR implements the
[transaction_unstable_broadcast](https://github.com/paritytech/json-rpc-interface-spec/blob/main/src/api/transaction_unstable_broadcast.md)
and
[transaction_unstable_stop](https://github.com/paritytech/json-rpc-interface-spec/blob/main/src/api/transaction_unstable_stop.md).


The
[transaction_unstable_broadcast](https://github.com/paritytech/json-rpc-interface-spec/blob/main/src/api/transaction_unstable_broadcast.md)
submits the provided transaction at the best block of the chain.
If the transaction is dropped or declared invalid, the API tries to
resubmit the transaction at the next available best block.

### Broadcasting 
The broadcasting operation continues until either:

- the user called `transaction_unstable_stop` with the operation ID that
identifies the broadcasting operation
- the transaction state is one of the following: 
  - Finalized: the transaction is part of the chain
- FinalizedTimeout: we have waited for 256 finalized blocks and timedout
  - Usurped the transaction has been replaced in the tx pool
  
The broadcasting retires to submit the transaction when the transaction
state is:
- Invalid: the transaction might become valid at a later time
- Dropped: the transaction pool's capacity is full at the moment, but
might clear when other transactions are finalized/dropped

### Stopping

The `transaction_unstable_broadcast` spawns an abortable future and
tracks the abort handler.
When the
[transaction_unstable_stop](https://github.com/paritytech/json-rpc-interface-spec/blob/main/src/api/transaction_unstable_stop.md)
is called with a valid operation ID; the abort handler of the
corresponding `transaction_unstable_broadcast` future is called. This
behavior ensures the broadcast future is finishes on the next polling.
When the `transaction_unstable_stop` is called with an invalid operation
ID, an invalid jsonrpc specific error object is returned.


### Testing

This PR adds the testing harness of the transaction API and validates
two basic scenarios:
- transaction enters and exits the transaction pool
- transaction stop returns appropriate values when called with valid and
invalid operation IDs


Closes: #3039

Note that the API should be enabled after:
#3084.

cc @paritytech/subxt-team

---------

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>
Co-authored-by: Sebastian Kunert <skunert49@gmail.com>

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 = { workspace = true, default-features = true }
 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 {
+	/// 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;