referrerpolicy=no-referrer-when-downgrade

sc_rpc_spec_v2/transaction/
transaction_broadcast.rs

1// This file is part of Substrate.
2
3// Copyright (C) Parity Technologies (UK) Ltd.
4// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
5
6// This program is free software: you can redistribute it and/or modify
7// it under the terms of the GNU General Public License as published by
8// the Free Software Foundation, either version 3 of the License, or
9// (at your option) any later version.
10
11// This program is distributed in the hope that it will be useful,
12// but WITHOUT ANY WARRANTY; without even the implied warranty of
13// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14// GNU General Public License for more details.
15
16// You should have received a copy of the GNU General Public License
17// along with this program. If not, see <https://www.gnu.org/licenses/>.
18
19//! API implementation for broadcasting transactions.
20
21use crate::{
22	common::connections::RpcConnections, transaction::api::TransactionBroadcastApiServer,
23	SubscriptionTaskExecutor,
24};
25use codec::Decode;
26use futures::{FutureExt, Stream, StreamExt};
27use futures_util::stream::AbortHandle;
28use jsonrpsee::{
29	core::{async_trait, RpcResult},
30	ConnectionId, Extensions,
31};
32use parking_lot::RwLock;
33use rand::{distributions::Alphanumeric, Rng};
34use sc_client_api::BlockchainEvents;
35use sc_transaction_pool_api::{
36	error::IntoPoolError, TransactionFor, TransactionPool, TransactionSource,
37};
38use sp_blockchain::HeaderBackend;
39use sp_core::Bytes;
40use sp_runtime::traits::Block as BlockT;
41use std::{collections::HashMap, sync::Arc};
42
43use super::error::ErrorBroadcast;
44
45/// An API for transaction RPC calls.
46pub struct TransactionBroadcast<Pool: TransactionPool, Client> {
47	/// Substrate client.
48	client: Arc<Client>,
49	/// Transactions pool.
50	pool: Arc<Pool>,
51	/// Executor to spawn subscriptions.
52	executor: SubscriptionTaskExecutor,
53	/// The broadcast operation IDs.
54	broadcast_ids: Arc<RwLock<HashMap<String, BroadcastState<Pool>>>>,
55	/// Keep track of how many concurrent operations are active for each connection.
56	rpc_connections: RpcConnections,
57}
58
59/// The state of a broadcast operation.
60struct BroadcastState<Pool: TransactionPool> {
61	/// Handle to abort the running future that broadcasts the transaction.
62	handle: AbortHandle,
63	/// Associated tx hash.
64	tx_hash: <Pool as TransactionPool>::Hash,
65}
66
67impl<Pool: TransactionPool, Client> TransactionBroadcast<Pool, Client> {
68	/// Creates a new [`TransactionBroadcast`].
69	pub fn new(
70		client: Arc<Client>,
71		pool: Arc<Pool>,
72		executor: SubscriptionTaskExecutor,
73		max_transactions_per_connection: usize,
74	) -> Self {
75		TransactionBroadcast {
76			client,
77			pool,
78			executor,
79			broadcast_ids: Default::default(),
80			rpc_connections: RpcConnections::new(max_transactions_per_connection),
81		}
82	}
83
84	/// Generate an unique operation ID for the `transaction_broadcast` RPC method.
85	pub fn generate_unique_id(&self) -> String {
86		let generate_operation_id = || {
87			// The length of the operation ID.
88			const OPERATION_ID_LEN: usize = 16;
89
90			rand::thread_rng()
91				.sample_iter(Alphanumeric)
92				.take(OPERATION_ID_LEN)
93				.map(char::from)
94				.collect::<String>()
95		};
96
97		let mut id = generate_operation_id();
98
99		let broadcast_ids = self.broadcast_ids.read();
100
101		while broadcast_ids.contains_key(&id) {
102			id = generate_operation_id();
103		}
104
105		id
106	}
107}
108
109/// Currently we treat all RPC transactions as externals.
110///
111/// Possibly in the future we could allow opt-in for special treatment
112/// of such transactions, so that the block authors can inject
113/// some unique transactions via RPC and have them included in the pool.
114const TX_SOURCE: TransactionSource = TransactionSource::External;
115
116#[async_trait]
117impl<Pool, Client> TransactionBroadcastApiServer for TransactionBroadcast<Pool, Client>
118where
119	Pool: TransactionPool + Sync + Send + 'static,
120	Pool::Error: IntoPoolError,
121	<Pool::Block as BlockT>::Hash: Unpin,
122	Client: HeaderBackend<Pool::Block> + BlockchainEvents<Pool::Block> + Send + Sync + 'static,
123{
124	async fn broadcast(&self, ext: &Extensions, bytes: Bytes) -> RpcResult<Option<String>> {
125		let pool = self.pool.clone();
126		let conn_id = ext
127			.get::<ConnectionId>()
128			.copied()
129			.expect("ConnectionId is always set by jsonrpsee; qed");
130
131		// The unique ID of this operation.
132		let id = self.generate_unique_id();
133
134		// Ensure that the connection has not reached the maximum number of active operations.
135		let Some(reserved_connection) = self.rpc_connections.reserve_space(conn_id) else {
136			return Ok(None)
137		};
138		let Some(reserved_identifier) = reserved_connection.register(id.clone()) else {
139			// This can only happen if the generated operation ID is not unique.
140			return Ok(None)
141		};
142
143		// The JSON-RPC server might check whether the transaction is valid before broadcasting it.
144		// If it does so and if the transaction is invalid, the server should silently do nothing
145		// and the JSON-RPC client is not informed of the problem. Invalid transactions should still
146		// count towards the limit to the number of simultaneously broadcasted transactions.
147		let Ok(decoded_extrinsic) = TransactionFor::<Pool>::decode(&mut &bytes[..]) else {
148			return Ok(Some(id));
149		};
150		// Save the tx hash to remove it later.
151		let tx_hash = pool.hash_of(&decoded_extrinsic);
152
153		// Get a stream of best block hashes that immediately produces the current best block.
154		// This is used for the broadcast method to retry submitting the transaction to a future
155		// block if the error is retriable (for example, the transaction is invalid at the moment,
156		// but will become valid at a later block N + 1).
157		//
158		// Providing the best hash immediately is important for chains that are configured with
159		// `InstantSeal`.
160		let best_hash = self.client.info().best_hash;
161
162		// The compiler can no longer deduce the type of the stream and complains
163		// about `one type is more general than the other`.
164		let mut best_block_import_stream: std::pin::Pin<
165			Box<dyn Stream<Item = <Pool::Block as BlockT>::Hash> + Send>,
166		> = Box::pin(futures::stream::select(
167			futures::stream::iter(std::iter::once(best_hash)),
168			self.client.import_notification_stream().filter_map(|notification| async move {
169				notification.is_new_best.then_some(notification.hash)
170			}),
171		));
172
173		let broadcast_transaction_fut = async move {
174			// Flag to determine if the we should broadcast the transaction again.
175			let mut is_done = false;
176
177			while !is_done {
178				// Wait for the last block to become available.
179				let Some(best_block_hash) =
180					last_stream_element(&mut best_block_import_stream).await
181				else {
182					return;
183				};
184
185				let mut stream = match pool
186					.submit_and_watch(best_block_hash, TX_SOURCE, decoded_extrinsic.clone())
187					.await
188				{
189					Ok(stream) => stream,
190					// The transaction was not included to the pool.
191					Err(e) => {
192						let Ok(pool_err) = e.into_pool_error() else { return };
193
194						if pool_err.is_retriable() {
195							// Try to resubmit the transaction at a later block for
196							// recoverable errors.
197							continue
198						} else {
199							return;
200						}
201					},
202				};
203
204				while let Some(event) = stream.next().await {
205					// Check if the transaction could be submitted again
206					// at a later time.
207					if event.is_retriable() {
208						break;
209					}
210
211					// Stop if this is the final event of the transaction stream
212					// and the event is not retriable.
213					if event.is_final() {
214						is_done = true;
215						break;
216					}
217				}
218			}
219		};
220
221		// Convert the future into an abortable future, for easily terminating it from the
222		// `transaction_stop` method.
223		let (fut, handle) = futures::future::abortable(broadcast_transaction_fut);
224		let broadcast_ids = self.broadcast_ids.clone();
225		let drop_id = id.clone();
226		let pool = self.pool.clone();
227		// The future expected by the executor must be `Future<Output = ()>` instead of
228		// `Future<Output = Result<(), Aborted>>`.
229		let fut = fut.then(move |result| {
230			async move {
231				// Connection space is cleaned when this object is dropped.
232				drop(reserved_identifier);
233
234				// Remove the entry from the broadcast IDs map.
235				let Some(broadcast_state) = broadcast_ids.write().remove(&drop_id) else { return };
236
237				// The broadcast was not stopped.
238				if result.is_ok() {
239					return
240				}
241
242				// Best effort pool removal (tx can already be finalized).
243				pool.report_invalid(None, [(broadcast_state.tx_hash, None)].into()).await;
244			}
245		});
246
247		// Keep track of this entry and the abortable handle.
248		{
249			let mut broadcast_ids = self.broadcast_ids.write();
250			broadcast_ids.insert(id.clone(), BroadcastState { handle, tx_hash });
251		}
252
253		sc_rpc::utils::spawn_subscription_task(&self.executor, fut);
254
255		Ok(Some(id))
256	}
257
258	async fn stop_broadcast(
259		&self,
260		ext: &Extensions,
261		operation_id: String,
262	) -> Result<(), ErrorBroadcast> {
263		let conn_id = ext
264			.get::<ConnectionId>()
265			.copied()
266			.expect("ConnectionId is always set by jsonrpsee; qed");
267
268		// The operation ID must correlate to the same connection ID.
269		if !self.rpc_connections.contains_identifier(conn_id, &operation_id) {
270			return Err(ErrorBroadcast::InvalidOperationID)
271		}
272
273		let mut broadcast_ids = self.broadcast_ids.write();
274
275		let Some(broadcast_state) = broadcast_ids.remove(&operation_id) else {
276			return Err(ErrorBroadcast::InvalidOperationID)
277		};
278
279		broadcast_state.handle.abort();
280
281		Ok(())
282	}
283}
284
285/// Returns the last element of the provided stream, or `None` if the stream is closed.
286async fn last_stream_element<S>(stream: &mut S) -> Option<S::Item>
287where
288	S: Stream + Unpin,
289{
290	let Some(mut element) = stream.next().await else { return None };
291
292	// We are effectively polling the stream for the last available item at this time.
293	// The `now_or_never` returns `None` if the stream is `Pending`.
294	//
295	// If the stream contains `Hash0x1 Hash0x2 Hash0x3 Hash0x4`, we want only `Hash0x4`.
296	while let Some(next) = stream.next().now_or_never() {
297		let Some(next) = next else {
298			// Nothing to do if the stream terminated.
299			return None
300		};
301		element = next;
302	}
303
304	Some(element)
305}
306
307#[cfg(test)]
308mod tests {
309	use super::*;
310	use tokio_stream::wrappers::ReceiverStream;
311
312	#[tokio::test]
313	async fn check_last_stream_element() {
314		let (tx, rx) = tokio::sync::mpsc::channel(16);
315
316		let mut stream = ReceiverStream::new(rx);
317		// Check the stream with one element queued.
318		tx.send(1).await.unwrap();
319		assert_eq!(last_stream_element(&mut stream).await, Some(1));
320
321		// Check the stream with multiple elements.
322		tx.send(1).await.unwrap();
323		tx.send(2).await.unwrap();
324		tx.send(3).await.unwrap();
325		assert_eq!(last_stream_element(&mut stream).await, Some(3));
326
327		// Drop the stream with some elements
328		tx.send(1).await.unwrap();
329		tx.send(2).await.unwrap();
330		drop(tx);
331		assert_eq!(last_stream_element(&mut stream).await, None);
332	}
333}