rpc: Implement transaction RPC API (#12328)

* rpc/tx: Add transaction structures for serialization

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* rpc/tx: Add public facing `TransactionEvent`

To circumvent the fact that serde does not allow mixing
`#[serde(tag = "event")]` with
`#[serde(tag = "event", content = "block")]`
the public facing subscription structure is serialized
and deserialized to an intermmediate representation.

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* rpc/tx: Add trait for the `transaction` API

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* rpc/tx: Convert RPC errors to transaction events

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* rpc/tx: Implement `transaction` RPC methods

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* tx-pool: Propagate tx index to events

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* tx-pool: Adjust testing to reflect tx index in events

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* rpc/tx: Convert tx-pool events for the new RPC spec

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* rpc/tx: Convert tx-pool `FinalityTimeout` event to `Dropped`

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* service: Enable the `transaction` API

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* rpc/tx: Add tests for tx event encoding and decoding

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* tx: Add indentation for subscriptions

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* rpc/tx: Fix documentation

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* rpc/tx: Serialize usize to hex

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* tx-pool: Rename closure parameters

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* service: Separate RPC spec versions

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* rpc/tx: Use `H256` for testing block's hash

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* rpc/tx: Serialize numbers as string

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* tx-pool: Backward compatibility with RPC v1

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* Update client/rpc-spec-v2/src/transaction/transaction.rs

Co-authored-by: Niklas Adolfsson <niklasadolfsson1@gmail.com>

* rpc/tx: Remove comment about serde clone

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* rpc/tx: Use RPC custom error code for invalid tx format

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* Update client/rpc-spec-v2/src/transaction/event.rs

Co-authored-by: James Wilson <james@jsdw.me>

* rpc/tx: Adjust internal structures for serialization/deserialization

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>
Co-authored-by: Niklas Adolfsson <niklasadolfsson1@gmail.com>
Co-authored-by: James Wilson <james@jsdw.me>

substrate/client/transaction-pool/api/Cargo.toml

@@ -15,3 +15,6 @@ serde = { version = "1.0.136", features = ["derive"] }
 thiserror = "1.0.30"
 sp-blockchain = { version = "4.0.0-dev", path = "../../../primitives/blockchain" }
 sp-runtime = { version = "6.0.0", default-features = false, path = "../../../primitives/runtime" }
+
+[dev-dependencies]
+serde_json = "1.0"

substrate/client/transaction-pool/api/src/lib.rs

@@ -108,15 +108,18 @@ pub enum TransactionStatus<Hash, BlockHash> {
 	Ready,
 	/// The transaction has been broadcast to the given peers.
 	Broadcast(Vec<String>),
-	/// Transaction has been included in block with given hash.
-	InBlock(BlockHash),
+	/// Transaction has been included in block with given hash
+	/// at the given position.
+	#[serde(with = "v1_compatible")]
+	InBlock((BlockHash, TxIndex)),
 	/// The block this transaction was included in has been retracted.
 	Retracted(BlockHash),
 	/// Maximum number of finality watchers has been reached,
 	/// old watchers are being removed.
 	FinalityTimeout(BlockHash),
-	/// Transaction has been finalized by a finality-gadget, e.g GRANDPA
-	Finalized(BlockHash),
+	/// Transaction has been finalized by a finality-gadget, e.g GRANDPA.
+	#[serde(with = "v1_compatible")]
+	Finalized((BlockHash, TxIndex)),
 	/// Transaction has been replaced in the pool, by another transaction
 	/// that provides the same tags. (e.g. same (sender, nonce)).
 	Usurped(Hash),
@@ -143,6 +146,8 @@ pub type TransactionFor<P> = <<P as TransactionPool>::Block as BlockT>::Extrinsi
 pub type TransactionStatusStreamFor<P> = TransactionStatusStream<TxHash<P>, BlockHash<P>>;
 /// Transaction type for a local pool.
 pub type LocalTransactionFor<P> = <<P as LocalTransactionPool>::Block as BlockT>::Extrinsic;
+/// Transaction's index within the block in which it was included.
+pub type TxIndex = usize;
 
 /// Typical future type used in transaction pool api.
 pub type PoolFuture<T, E> = std::pin::Pin<Box<dyn Future<Output = Result<T, E>> + Send>>;
@@ -362,3 +367,52 @@ impl<TPool: LocalTransactionPool> OffchainSubmitTransaction<TPool::Block> for TP
 		})
 	}
 }
+
+/// Wrapper functions to keep the API backwards compatible over the wire for the old RPC spec.
+mod v1_compatible {
+	use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+	pub fn serialize<S, H>(data: &(H, usize), serializer: S) -> Result<S::Ok, S::Error>
+	where
+		S: Serializer,
+		H: Serialize,
+	{
+		let (hash, _) = data;
+		serde::Serialize::serialize(&hash, serializer)
+	}
+
+	pub fn deserialize<'de, D, H>(deserializer: D) -> Result<(H, usize), D::Error>
+	where
+		D: Deserializer<'de>,
+		H: Deserialize<'de>,
+	{
+		let hash: H = serde::Deserialize::deserialize(deserializer)?;
+		Ok((hash, 0))
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn tx_status_compatibility() {
+		let event: TransactionStatus<u8, u8> = TransactionStatus::InBlock((1, 2));
+		let ser = serde_json::to_string(&event).unwrap();
+
+		let exp = r#"{"inBlock":1}"#;
+		assert_eq!(ser, exp);
+
+		let event_dec: TransactionStatus<u8, u8> = serde_json::from_str(exp).unwrap();
+		assert_eq!(event_dec, TransactionStatus::InBlock((1, 0)));
+
+		let event: TransactionStatus<u8, u8> = TransactionStatus::Finalized((1, 2));
+		let ser = serde_json::to_string(&event).unwrap();
+
+		let exp = r#"{"finalized":1}"#;
+		assert_eq!(ser, exp);
+
+		let event_dec: TransactionStatus<u8, u8> = serde_json::from_str(exp).unwrap();
+		assert_eq!(event_dec, TransactionStatus::Finalized((1, 0)));
+	}
+}

substrate/client/transaction-pool/src/graph/listener.rs

@@ -104,13 +104,18 @@ impl<H: hash::Hash + traits::Member + Serialize, C: ChainApi> Listener<H, C> {
 	/// Transaction was pruned from the pool.
 	pub fn pruned(&mut self, block_hash: BlockHash<C>, tx: &H) {
 		debug!(target: "txpool", "[{:?}] Pruned at {:?}", tx, block_hash);
-		self.fire(tx, |s| s.in_block(block_hash));
-		self.finality_watchers.entry(block_hash).or_insert(vec![]).push(tx.clone());
+		// Get the transactions included in the given block hash.
+		let txs = self.finality_watchers.entry(block_hash).or_insert(vec![]);
+		txs.push(tx.clone());
+		// Current transaction is the last one included.
+		let tx_index = txs.len() - 1;
+
+		self.fire(tx, |watcher| watcher.in_block(block_hash, tx_index));
 
 		while self.finality_watchers.len() > MAX_FINALITY_WATCHERS {
 			if let Some((hash, txs)) = self.finality_watchers.pop_front() {
 				for tx in txs {
-					self.fire(&tx, |s| s.finality_timeout(hash));
+					self.fire(&tx, |watcher| watcher.finality_timeout(hash));
 				}
 			}
 		}
@@ -120,7 +125,7 @@ impl<H: hash::Hash + traits::Member + Serialize, C: ChainApi> Listener<H, C> {
 	pub fn retracted(&mut self, block_hash: BlockHash<C>) {
 		if let Some(hashes) = self.finality_watchers.remove(&block_hash) {
 			for hash in hashes {
-				self.fire(&hash, |s| s.retracted(block_hash))
+				self.fire(&hash, |watcher| watcher.retracted(block_hash))
 			}
 		}
 	}
@@ -128,9 +133,9 @@ impl<H: hash::Hash + traits::Member + Serialize, C: ChainApi> Listener<H, C> {
 	/// Notify all watchers that transactions have been finalized
 	pub fn finalized(&mut self, block_hash: BlockHash<C>) {
 		if let Some(hashes) = self.finality_watchers.remove(&block_hash) {
-			for hash in hashes {
+			for (tx_index, hash) in hashes.into_iter().enumerate() {
 				log::debug!(target: "txpool", "[{:?}] Sent finalization event (block {:?})", hash, block_hash);
-				self.fire(&hash, |s| s.finalized(block_hash))
+				self.fire(&hash, |watcher| watcher.finalized(block_hash, tx_index))
 			}
 		}
 	}

substrate/client/transaction-pool/src/graph/pool.rs

@@ -770,7 +770,7 @@ mod tests {
 			assert_eq!(stream.next(), Some(TransactionStatus::Ready));
 			assert_eq!(
 				stream.next(),
-				Some(TransactionStatus::InBlock(H256::from_low_u64_be(2).into())),
+				Some(TransactionStatus::InBlock((H256::from_low_u64_be(2).into(), 0))),
 			);
 		}
 
@@ -803,7 +803,7 @@ mod tests {
 			assert_eq!(stream.next(), Some(TransactionStatus::Ready));
 			assert_eq!(
 				stream.next(),
-				Some(TransactionStatus::InBlock(H256::from_low_u64_be(2).into())),
+				Some(TransactionStatus::InBlock((H256::from_low_u64_be(2).into(), 0))),
 			);
 		}
 

substrate/client/transaction-pool/src/graph/watcher.rs

@@ -84,13 +84,13 @@ impl<H: Clone, BH: Clone> Sender<H, BH> {
 	}
 
 	/// Extrinsic has been included in block with given hash.
-	pub fn in_block(&mut self, hash: BH) {
-		self.send(TransactionStatus::InBlock(hash));
+	pub fn in_block(&mut self, hash: BH, index: usize) {
+		self.send(TransactionStatus::InBlock((hash, index)));
 	}
 
 	/// Extrinsic has been finalized by a finality gadget.
-	pub fn finalized(&mut self, hash: BH) {
-		self.send(TransactionStatus::Finalized(hash));
+	pub fn finalized(&mut self, hash: BH, index: usize) {
+		self.send(TransactionStatus::Finalized((hash, index)));
 		self.is_finalized = true;
 	}
 

substrate/client/transaction-pool/tests/pool.rs

@@ -328,7 +328,7 @@ fn should_revalidate_across_many_blocks() {
 
 	block_on(
 		watcher1
-			.take_while(|s| future::ready(*s != TransactionStatus::InBlock(block_hash)))
+			.take_while(|s| future::ready(*s != TransactionStatus::InBlock((block_hash, 0))))
 			.collect::<Vec<_>>(),
 	);
 
@@ -398,24 +398,24 @@ fn should_push_watchers_during_maintenance() {
 		futures::executor::block_on_stream(watcher0).collect::<Vec<_>>(),
 		vec![
 			TransactionStatus::Ready,
-			TransactionStatus::InBlock(header_hash),
-			TransactionStatus::Finalized(header_hash)
+			TransactionStatus::InBlock((header_hash, 0)),
+			TransactionStatus::Finalized((header_hash, 0))
 		],
 	);
 	assert_eq!(
 		futures::executor::block_on_stream(watcher1).collect::<Vec<_>>(),
 		vec![
 			TransactionStatus::Ready,
-			TransactionStatus::InBlock(header_hash),
-			TransactionStatus::Finalized(header_hash)
+			TransactionStatus::InBlock((header_hash, 1)),
+			TransactionStatus::Finalized((header_hash, 1))
 		],
 	);
 	assert_eq!(
 		futures::executor::block_on_stream(watcher2).collect::<Vec<_>>(),
 		vec![
 			TransactionStatus::Ready,
-			TransactionStatus::InBlock(header_hash),
-			TransactionStatus::Finalized(header_hash)
+			TransactionStatus::InBlock((header_hash, 2)),
+			TransactionStatus::Finalized((header_hash, 2))
 		],
 	);
 }
@@ -450,8 +450,8 @@ fn finalization() {
 
 	let mut stream = futures::executor::block_on_stream(watcher);
 	assert_eq!(stream.next(), Some(TransactionStatus::Ready));
-	assert_eq!(stream.next(), Some(TransactionStatus::InBlock(header.hash())));
-	assert_eq!(stream.next(), Some(TransactionStatus::Finalized(header.hash())));
+	assert_eq!(stream.next(), Some(TransactionStatus::InBlock((header.hash(), 0))));
+	assert_eq!(stream.next(), Some(TransactionStatus::Finalized((header.hash(), 0))));
 	assert_eq!(stream.next(), None);
 }
 
@@ -573,30 +573,31 @@ fn fork_aware_finalization() {
 	for (canon_watcher, h) in canon_watchers {
 		let mut stream = futures::executor::block_on_stream(canon_watcher);
 		assert_eq!(stream.next(), Some(TransactionStatus::Ready));
-		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(h)));
-		assert_eq!(stream.next(), Some(TransactionStatus::Finalized(h)));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock((h, 0))));
+		assert_eq!(stream.next(), Some(TransactionStatus::Finalized((h, 0))));
 		assert_eq!(stream.next(), None);
 	}
 
 	{
 		let mut stream = futures::executor::block_on_stream(from_dave_watcher);
 		assert_eq!(stream.next(), Some(TransactionStatus::Ready));
-		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(c2)));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock((c2, 0))));
 		assert_eq!(stream.next(), Some(TransactionStatus::Retracted(c2)));
 		assert_eq!(stream.next(), Some(TransactionStatus::Ready));
-		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(e1)));
-		assert_eq!(stream.next(), Some(TransactionStatus::Finalized(e1)));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock((e1, 0))));
+		assert_eq!(stream.next(), Some(TransactionStatus::Finalized((e1, 0))));
 		assert_eq!(stream.next(), None);
 	}
 
 	{
 		let mut stream = futures::executor::block_on_stream(from_bob_watcher);
 		assert_eq!(stream.next(), Some(TransactionStatus::Ready));
-		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(d2)));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock((d2, 0))));
 		assert_eq!(stream.next(), Some(TransactionStatus::Retracted(d2)));
 		assert_eq!(stream.next(), Some(TransactionStatus::Ready));
-		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(e1)));
-		assert_eq!(stream.next(), Some(TransactionStatus::Finalized(e1)));
+		// In block e1 we submitted: [dave, bob] xts in this order.
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock((e1, 1))));
+		assert_eq!(stream.next(), Some(TransactionStatus::Finalized((e1, 1))));
 		assert_eq!(stream.next(), None);
 	}
 }
@@ -646,10 +647,10 @@ fn prune_and_retract_tx_at_same_time() {
 	{
 		let mut stream = futures::executor::block_on_stream(watcher);
 		assert_eq!(stream.next(), Some(TransactionStatus::Ready));
-		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(b1)));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock((b1, 0))));
 		assert_eq!(stream.next(), Some(TransactionStatus::Retracted(b1)));
-		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(b2)));
-		assert_eq!(stream.next(), Some(TransactionStatus::Finalized(b2)));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock((b2, 0))));
+		assert_eq!(stream.next(), Some(TransactionStatus::Finalized((b2, 0))));
 		assert_eq!(stream.next(), None);
 	}
 }