diff --git a/substrate/Cargo.lock b/substrate/Cargo.lock index 1adf135f0e508152ce45f7352e276934868dde31..3a14f98546690160355369cbe910049359f6764b 100644 --- a/substrate/Cargo.lock +++ b/substrate/Cargo.lock @@ -2787,7 +2787,7 @@ dependencies = [ "parking_lot 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", "serde 1.0.70 (registry+https://github.com/rust-lang/crates.io-index)", "serde_derive 1.0.70 (registry+https://github.com/rust-lang/crates.io-index)", - "transaction-pool 1.12.1 (registry+https://github.com/rust-lang/crates.io-index)", + "transaction-pool 1.12.3 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -2930,6 +2930,7 @@ dependencies = [ "jsonrpc-pubsub 8.0.1 (git+https://github.com/paritytech/jsonrpc.git)", "jsonrpc-ws-server 8.0.0 (git+https://github.com/paritytech/jsonrpc.git)", "log 0.3.9 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.70 (registry+https://github.com/rust-lang/crates.io-index)", "substrate-rpc 0.1.0", "substrate-runtime-primitives 0.1.0", ] @@ -3632,11 +3633,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" [[package]] name = "transaction-pool" -version = "1.12.1" +version = "1.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "error-chain 0.12.0 (registry+https://github.com/rust-lang/crates.io-index)", - "log 0.3.9 (registry+https://github.com/rust-lang/crates.io-index)", + "log 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)", "smallvec 0.4.5 (registry+https://github.com/rust-lang/crates.io-index)", "trace-time 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -4251,7 +4252,7 @@ dependencies = [ "checksum tokio-udp 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "137bda266504893ac4774e0ec4c2108f7ccdbcb7ac8dced6305fe9e4e0b5041a" "checksum trace-time 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "5aea07da6582e957c6e737eeb63a5af79e648eeeaaaba8fd9a417f1124bafa41" "checksum traitobject 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "efd1f82c56340fdf16f2a953d7bda4f8fdffba13d93b00844c25572110b26079" -"checksum transaction-pool 1.12.1 (registry+https://github.com/rust-lang/crates.io-index)" = "be1efb673ddf49ab4a99893eb3af02f6563636033fb832c2b7f937641ad62b17" +"checksum transaction-pool 1.12.3 (registry+https://github.com/rust-lang/crates.io-index)" = "0f1fc049d9f039d3e9264f97f978309b94b5a5c56a5c18d28f91f469cef2b367" "checksum triehash 0.1.0 (git+https://github.com/paritytech/parity.git)" = "<none>" "checksum triehash 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "2033893a813c70e7d8a739ca6c36dc0a7a2c913ec718d7cbf84a3837bbe3c7ce" "checksum try-lock 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "ee2aa4715743892880f70885373966c83d73ef1b0838a664ef0c76fffd35e7c2" diff --git a/substrate/demo/cli/src/lib.rs b/substrate/demo/cli/src/lib.rs index 7a8869db8410a6b4adc9b6c77dfc929a81197f35..27a317e45fb8e867ea9fee070170ae405a66fb10 100644 --- a/substrate/demo/cli/src/lib.rs +++ b/substrate/demo/cli/src/lib.rs @@ -59,6 +59,7 @@ use demo_executor::NativeExecutor; struct DummyPool; impl extrinsic_pool::api::ExtrinsicPool<UncheckedExtrinsic, BlockId, Hash> for DummyPool { type Error = extrinsic_pool::txpool::Error; + type InPool = (); fn submit(&self, _block: BlockId, _: Vec<UncheckedExtrinsic>) -> Result<Vec<Hash>, Self::Error> @@ -79,6 +80,10 @@ impl extrinsic_pool::api::ExtrinsicPool<UncheckedExtrinsic, BlockId, Hash> for D fn import_notification_stream(&self) -> extrinsic_pool::api::EventStream { unreachable!() } + + fn all(&self) -> Self::InPool { + unreachable!() + } } struct DummySystem; @@ -176,7 +181,7 @@ pub fn run<I, T>(args: I) -> error::Result<()> where let state = rpc::apis::state::State::new(client.clone(), runtime.executor()); let chain = rpc::apis::chain::Chain::new(client.clone(), runtime.executor()); let author = rpc::apis::author::Author::new(client.clone(), Arc::new(DummyPool), runtime.executor()); - rpc::rpc_handler::<Block, _, _, _, _>(state, chain, author, DummySystem) + rpc::rpc_handler::<Block, _, _, _, _, _>(state, chain, author, DummySystem) }; let http_address = "127.0.0.1:9933".parse().unwrap(); let ws_address = "127.0.0.1:9944".parse().unwrap(); diff --git a/substrate/polkadot/transaction-pool/src/lib.rs b/substrate/polkadot/transaction-pool/src/lib.rs index 676b0d9773b42bd3661d74d6db2c5db2d13923ba..96ad1050d95d09dd9164dfffeac4800fd26daa77 100644 --- a/substrate/polkadot/transaction-pool/src/lib.rs +++ b/substrate/polkadot/transaction-pool/src/lib.rs @@ -38,7 +38,7 @@ mod error; use std::{ cmp::Ordering, - collections::HashMap, + collections::{BTreeMap, HashMap}, ops::Deref, sync::Arc, }; @@ -54,6 +54,7 @@ use extrinsic_pool::{ use polkadot_api::PolkadotApi; use primitives::{AccountId, BlockId, Hash, Index, UncheckedExtrinsic as FutureProofUncheckedExtrinsic}; use runtime::{Address, UncheckedExtrinsic}; +use substrate_primitives::Bytes; use substrate_runtime_primitives::traits::{Bounded, Checkable, Hash as HashT, BlakeTwo256}; pub use extrinsic_pool::txpool::{Options, Status, LightStatus, VerifiedTransaction as VerifiedTransactionOps}; @@ -182,9 +183,14 @@ impl txpool::Scoring<VerifiedTransaction> for Scoring { } } - fn should_replace(&self, old: &VerifiedTransaction, _new: &VerifiedTransaction) -> bool { - // Always replace not fully verified transactions. - !old.is_fully_verified() + fn should_replace(&self, old: &VerifiedTransaction, _new: &VerifiedTransaction) -> Choice { + if old.is_fully_verified() { + // Don't allow new transactions if we are reaching the limit. + Choice::RejectNew + } else { + // Always replace not fully verified transactions. + Choice::ReplaceOld + } } } @@ -415,6 +421,7 @@ impl<A> ExtrinsicPool<FutureProofUncheckedExtrinsic, BlockId, Hash> for Transact A: PolkadotApi, { type Error = Error; + type InPool = BTreeMap<AccountId, Vec<Bytes>>; fn submit(&self, block: BlockId, xts: Vec<FutureProofUncheckedExtrinsic>) -> Result<Vec<Hash>> { xts.into_iter() @@ -446,6 +453,17 @@ impl<A> ExtrinsicPool<FutureProofUncheckedExtrinsic, BlockId, Hash> for Transact fn import_notification_stream(&self) -> EventStream { self.inner.import_notification_stream() } + + fn all(&self) -> Self::InPool { + self.inner.all(|it| it.fold(Default::default(), |mut map: Self::InPool, tx| { + // Map with `null` key is not serializable, so we fallback to default accountId. + map.entry(tx.sender().unwrap_or_default()) + .or_insert_with(Vec::new) + // use bytes type to make it serialize nicer. + .push(Bytes(tx.primitive_extrinsic())); + map + })) + } } #[cfg(test)] diff --git a/substrate/substrate/extrinsic-pool/src/api.rs b/substrate/substrate/extrinsic-pool/src/api.rs index 0d8bcfbf223d0e6ede934059a197e038c5ccf799..148ab49e4f4d931fe172b99330934f0216b873e4 100644 --- a/substrate/substrate/extrinsic-pool/src/api.rs +++ b/substrate/substrate/extrinsic-pool/src/api.rs @@ -16,6 +16,9 @@ //! External API for extrinsic pool. +use std::fmt::Debug; + +use serde::{Serialize, de::DeserializeOwned}; use txpool; use futures::sync::mpsc; @@ -43,6 +46,9 @@ pub trait ExtrinsicPool<Ex, BlockId, Hash>: Send + Sync + 'static { /// Error type type Error: Error; + /// Pooled extrinsics + type InPool: Debug + Serialize + DeserializeOwned + Send + Sync + 'static; + /// Submit a collection of extrinsics to the pool. fn submit(&self, block: BlockId, xt: Vec<Ex>) -> Result<Vec<Hash>, Self::Error>; @@ -54,4 +60,7 @@ pub trait ExtrinsicPool<Ex, BlockId, Hash>: Send + Sync + 'static { /// Return an event stream of transactions imported to the pool. fn import_notification_stream(&self) -> EventStream; + + /// Return all extrinsics in the pool aggregated by the sender. + fn all(&self) -> Self::InPool; } diff --git a/substrate/substrate/extrinsic-pool/src/pool.rs b/substrate/substrate/extrinsic-pool/src/pool.rs index 2fffc1442e42875868ab852647b60c8a944340f7..e874086f8d4659583b5f6190e33e157cc11db3da 100644 --- a/substrate/substrate/extrinsic-pool/src/pool.rs +++ b/substrate/substrate/extrinsic-pool/src/pool.rs @@ -149,4 +149,19 @@ impl<Hash, VEx, S, E> Pool<Hash, VEx, S, E> where { f(self.pool.read().pending(ready)) } + + /// Retrieve all transactions in the pool. The transactions might be unordered. + pub fn all<F, T>(&self, f: F) -> T where + F: FnOnce(txpool::UnorderedIterator<VEx, AlwaysReady, S>) -> T, + { + f(self.pool.read().unordered_pending(AlwaysReady)) + } +} + +/// A Readiness implementation that returns `Ready` for all transactions. +pub struct AlwaysReady; +impl<VEx> txpool::Ready<VEx> for AlwaysReady { + fn is_ready(&mut self, _tx: &VEx) -> txpool::Readiness { + txpool::Readiness::Ready + } } diff --git a/substrate/substrate/rpc-servers/Cargo.toml b/substrate/substrate/rpc-servers/Cargo.toml index dec067b409fb4ce2847ce4ef45ee51848a876dcd..6b7038d44a6286f07b65ecb1e85f5239d652ad79 100644 --- a/substrate/substrate/rpc-servers/Cargo.toml +++ b/substrate/substrate/rpc-servers/Cargo.toml @@ -9,5 +9,6 @@ jsonrpc-http-server = { git = "https://github.com/paritytech/jsonrpc.git" } jsonrpc-pubsub = { git = "https://github.com/paritytech/jsonrpc.git" } jsonrpc-ws-server = { git = "https://github.com/paritytech/jsonrpc.git" } log = "0.3" +serde = "1.0" substrate-rpc = { path = "../rpc", version = "0.1" } substrate-runtime-primitives = { path = "../runtime/primitives" } diff --git a/substrate/substrate/rpc-servers/src/lib.rs b/substrate/substrate/rpc-servers/src/lib.rs index 11fa9aa9b60af738754b089561599189f0c91cf4..7344a0453c1866f71ecddac73b2e7eb3a69310cb 100644 --- a/substrate/substrate/rpc-servers/src/lib.rs +++ b/substrate/substrate/rpc-servers/src/lib.rs @@ -24,6 +24,7 @@ extern crate jsonrpc_core as rpc; extern crate jsonrpc_http_server as http; extern crate jsonrpc_pubsub as pubsub; extern crate jsonrpc_ws_server as ws; +extern crate serde; extern crate substrate_runtime_primitives; #[macro_use] @@ -38,16 +39,17 @@ pub type HttpServer = http::Server; pub type WsServer = ws::Server; /// Construct rpc `IoHandler` -pub fn rpc_handler<Block: BlockT, S, C, A, Y>( +pub fn rpc_handler<Block, PendingExtrinsics, S, C, A, Y>( state: S, chain: C, author: A, system: Y, ) -> RpcHandler where - Block: 'static, + Block: BlockT + 'static, + PendingExtrinsics: serde::Serialize + serde::de::DeserializeOwned + Send + Sync + 'static, S: apis::state::StateApi<Block::Hash, Metadata=Metadata>, C: apis::chain::ChainApi<Block::Hash, Block::Header, Metadata=Metadata>, - A: apis::author::AuthorApi<Block::Hash, Block::Extrinsic, Metadata=Metadata>, + A: apis::author::AuthorApi<Block::Hash, Block::Extrinsic, PendingExtrinsics, Metadata=Metadata>, Y: apis::system::SystemApi, { let mut io = pubsub::PubSubHandler::default(); diff --git a/substrate/substrate/rpc/src/author/mod.rs b/substrate/substrate/rpc/src/author/mod.rs index 9077abdc3ffd86034748377f2cf9a72f2b5240cf..1d8385c0e9e2872e55471864fb1803b5731d8204 100644 --- a/substrate/substrate/rpc/src/author/mod.rs +++ b/substrate/substrate/rpc/src/author/mod.rs @@ -41,7 +41,7 @@ use self::error::Result; build_rpc_trait! { /// Substrate authoring RPC API - pub trait AuthorApi<Hash, Extrinsic> { + pub trait AuthorApi<Hash, Extrinsic, PendingExtrinsics> { type Metadata; /// Submit extrinsic for inclusion in block. @@ -51,6 +51,10 @@ build_rpc_trait! { #[rpc(name = "author_submitExtrinsic")] fn submit_extrinsic(&self, Bytes) -> Result<Hash>; + /// Returns all pending extrinsics, potentially grouped by sender. + #[rpc(name = "author_pendingExtrinsics")] + fn pending_extrinsics(&self) -> Result<PendingExtrinsics>; + #[pubsub(name = "author_extrinsicUpdate")] { /// Submit an extrinsic to watch. #[rpc(name = "author_submitAndWatchExtrinsic")] @@ -60,7 +64,6 @@ build_rpc_trait! { #[rpc(name = "author_unwatchExtrinsic")] fn unwatch_extrinsic(&self, SubscriptionId) -> Result<bool>; } - } } @@ -85,12 +88,13 @@ impl<B, E, Block: traits::Block, P> Author<B, E, Block, P> { } } -impl<B, E, Block, P, Ex, Hash> AuthorApi<Hash, Ex> for Author<B, E, Block, P> where +impl<B, E, Block, P, Ex, Hash, InPool> AuthorApi<Hash, Ex, InPool> for Author<B, E, Block, P> where B: client::backend::Backend<Block> + Send + Sync + 'static, E: client::CallExecutor<Block> + Send + Sync + 'static, Block: traits::Block + 'static, - Hash: traits::MaybeSerializeDebug + Sync + Send + 'static, - P: ExtrinsicPool<Ex, generic::BlockId<Block>, Hash>, + Hash: traits::MaybeSerializeDebug + Send + Sync + 'static, + InPool: traits::MaybeSerializeDebug + Send + Sync + 'static, + P: ExtrinsicPool<Ex, generic::BlockId<Block>, Hash, InPool=InPool>, P::Error: 'static, Ex: Codec, { @@ -112,6 +116,10 @@ impl<B, E, Block, P, Ex, Hash> AuthorApi<Hash, Ex> for Author<B, E, Block, P> wh ) } + fn pending_extrinsics(&self) -> Result<InPool> { + Ok(self.pool.all()) + } + fn watch_extrinsic(&self, _metadata: Self::Metadata, subscriber: pubsub::Subscriber<Status<Hash>>, xt: Bytes) { let submit = || -> Result<_> { @@ -146,3 +154,4 @@ impl<B, E, Block, P, Ex, Hash> AuthorApi<Hash, Ex> for Author<B, E, Block, P> wh Ok(self.subscriptions.cancel(id)) } } + diff --git a/substrate/substrate/rpc/src/author/tests.rs b/substrate/substrate/rpc/src/author/tests.rs index 116ca7ea2c7bc4620d4a536e528f4bc2d8db8358..ebca4596adb7cd483d298df6b113b9deadd8f6a7 100644 --- a/substrate/substrate/rpc/src/author/tests.rs +++ b/substrate/substrate/rpc/src/author/tests.rs @@ -46,6 +46,7 @@ impl fmt::Display for Error { impl<BlockHash> api::ExtrinsicPool<Extrinsic, BlockHash, u64> for DummyTxPool { type Error = Error; + type InPool = Vec<u8>; /// Submit extrinsic for inclusion in block. fn submit(&self, _block: BlockHash, xt: Vec<Extrinsic>) -> Result<Vec<Hash>, Self::Error> { @@ -79,6 +80,10 @@ impl<BlockHash> api::ExtrinsicPool<Extrinsic, BlockHash, u64> for DummyTxPool { fn import_notification_stream(&self) -> api::EventStream { unreachable!() } + + fn all(&self) -> Self::InPool { + vec![1, 2, 3, 4, 5] + } } #[test] @@ -143,3 +148,18 @@ fn should_watch_extrinsic() { Some(r#"{"jsonrpc":"2.0","method":"test","params":{"result":{"usurped":5},"subscription":0}}"#.into()) ); } + +#[test] +fn should_return_pending_extrinsics() { + let runtime = runtime::Runtime::new().unwrap(); + let p = Author { + client: Arc::new(test_client::new()), + pool: Arc::new(DummyTxPool::default()), + subscriptions: Subscriptions::new(runtime.executor()), + }; + + assert_matches!( + p.pending_extrinsics(), + Ok(ref expected) if expected == &[1u8, 2, 3, 4, 5] + ); +} diff --git a/substrate/substrate/service/src/lib.rs b/substrate/substrate/service/src/lib.rs index 9fcd157f733c8dc8286842cbc41373dfd71437b4..a265ec41e16a363ac83dc3d174ec44e22b5f1bab 100644 --- a/substrate/substrate/service/src/lib.rs +++ b/substrate/substrate/service/src/lib.rs @@ -210,7 +210,8 @@ impl<Components> Service<Components> let chain = rpc::apis::chain::Chain::new(client.clone(), task_executor.clone()); let state = rpc::apis::state::State::new(client.clone(), task_executor.clone()); let author = rpc::apis::author::Author::new(client.clone(), extrinsic_pool.api(), task_executor.clone()); - rpc::rpc_handler::<ComponentBlock<Components>, _, _, _, _>( + + rpc::rpc_handler::<ComponentBlock<Components>, _, _, _, _, _>( state, chain, author,