hacky

src/client.rs

@@ -15,10 +15,12 @@
 // along with subxt.  If not, see <http://www.gnu.org/licenses/>.
 
 use futures::future;
+use sp_runtime::traits::Hash;
 pub use sp_runtime::traits::SignedExtension;
 pub use sp_version::RuntimeVersion;
 
 use crate::{
+    error::Error,
     events::EventsDecoder,
     extrinsic::{
         self,
@@ -27,19 +29,19 @@ use crate::{
         UncheckedExtrinsic,
     },
     rpc::{
-        ExtrinsicSuccess,
         Rpc,
         RpcClient,
         SystemProperties,
     },
     storage::StorageClient,
+    transaction::TransactionProgress,
     AccountData,
     Call,
     Config,
-    Error,
     ExtrinsicExtraData,
     Metadata,
 };
+use std::sync::Arc;
 
 /// ClientBuilder for constructing a Client.
 #[derive(Default)]
@@ -47,7 +49,6 @@ pub struct ClientBuilder {
     url: Option<String>,
     client: Option<RpcClient>,
     page_size: Option<u32>,
-    accept_weak_inclusion: bool,
 }
 
 impl ClientBuilder {
@@ -57,7 +58,6 @@ impl ClientBuilder {
             url: None,
             client: None,
             page_size: None,
-            accept_weak_inclusion: false,
         }
     }
 
@@ -79,12 +79,6 @@ impl ClientBuilder {
         self
     }
 
-    /// Only check that transactions are InBlock on submit.
-    pub fn accept_weak_inclusion(mut self) -> Self {
-        self.accept_weak_inclusion = true;
-        self
-    }
-
     /// Creates a new Client.
     pub async fn build<T: Config>(self) -> Result<Client<T>, Error> {
         let client = if let Some(client) = self.client {
@@ -93,10 +87,7 @@ impl ClientBuilder {
             let url = self.url.as_deref().unwrap_or("ws://127.0.0.1:9944");
             crate::rpc::build_ws_client(url).await?
         };
-        let mut rpc = Rpc::new(client);
-        if self.accept_weak_inclusion {
-            rpc.accept_weak_inclusion();
-        }
+        let rpc = Rpc::new(client);
         let (metadata, genesis_hash, runtime_version, properties) = future::join4(
             rpc.metadata(),
             rpc.genesis_hash(),
@@ -111,7 +102,7 @@ impl ClientBuilder {
         Ok(Client {
             rpc,
             genesis_hash: genesis_hash?,
-            metadata,
+            metadata: Arc::new(metadata),
             events_decoder,
             properties: properties.unwrap_or_else(|_| Default::default()),
             runtime_version: runtime_version?,
@@ -121,28 +112,28 @@ impl ClientBuilder {
 }
 
 /// Client to interface with a substrate node.
+#[derive(Clone)]
 pub struct Client<T: Config> {
     rpc: Rpc<T>,
     genesis_hash: T::Hash,
-    metadata: Metadata,
+    metadata: Arc<Metadata>,
     events_decoder: EventsDecoder<T>,
     properties: SystemProperties,
     runtime_version: RuntimeVersion,
-    // _marker: PhantomData<(fn() -> T::Signature, T::Extra)>,
     iter_page_size: u32,
 }
 
-impl<T: Config> Clone for Client<T> {
-    fn clone(&self) -> Self {
-        Self {
-            rpc: self.rpc.clone(),
-            genesis_hash: self.genesis_hash,
-            metadata: self.metadata.clone(),
-            events_decoder: self.events_decoder.clone(),
-            properties: self.properties.clone(),
-            runtime_version: self.runtime_version.clone(),
-            iter_page_size: self.iter_page_size,
-        }
+impl<T: Config> std::fmt::Debug for Client<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Client")
+            .field("rpc", &"<Rpc>")
+            .field("genesis_hash", &self.genesis_hash)
+            .field("metadata", &"<Metadata>")
+            .field("events_decoder", &"<EventsDecoder>")
+            .field("properties", &self.properties)
+            .field("runtime_version", &self.runtime_version.to_string())
+            .field("iter_page_size", &self.iter_page_size)
+            .finish()
     }
 }
 
@@ -194,37 +185,40 @@ impl<T: Config> Client<T> {
 }
 
 /// A constructed call ready to be signed and submitted.
-pub struct SubmittableExtrinsic<'a, T: Config, C> {
-    client: &'a Client<T>,
+pub struct SubmittableExtrinsic<'client, T: Config, C> {
+    client: &'client Client<T>,
     call: C,
 }
 
-impl<'a, T, C> SubmittableExtrinsic<'a, T, C>
+impl<'client, T, C> SubmittableExtrinsic<'client, T, C>
 where
     T: Config + ExtrinsicExtraData<T>,
     C: Call + Send + Sync,
 {
     /// Create a new [`SubmittableExtrinsic`].
-    pub fn new(client: &'a Client<T>, call: C) -> Self {
+    pub fn new(client: &'client Client<T>, call: C) -> Self {
         Self { client, call }
     }
 
     /// Creates and signs an extrinsic and submits it to the chain.
     ///
-    /// Returns when the extrinsic has successfully been included in the block, together with any
-    /// events which were triggered by the extrinsic.
+    /// Returns a [`TransactionProgress`], which can be used to track the status of the transaction
+    /// and obtain details about it, once it has made it into a block.
     pub async fn sign_and_submit_then_watch(
         self,
         signer: &(dyn Signer<T> + Send + Sync),
-    ) -> Result<ExtrinsicSuccess<T>, Error>
+    ) -> Result<TransactionProgress<'client, T>, Error>
     where
         <<<T as ExtrinsicExtraData<T>>::Extra as SignedExtra<T>>::Extra as SignedExtension>::AdditionalSigned: Send + Sync + 'static
     {
+        // Sign the call data to create our extrinsic.
         let extrinsic = self.create_signed(signer, Default::default()).await?;
-        self.client
-            .rpc()
-            .submit_and_watch_extrinsic(extrinsic, self.client.events_decoder())
-            .await
+        // Get a hash of the extrinsic (we'll need this later).
+        let ext_hash = T::Hashing::hash_of(&extrinsic);
+        // Submit and watch for transaction progress.
+        let sub = self.client.rpc().watch_extrinsic(extrinsic).await?;
+
+        Ok(TransactionProgress::new(sub, self.client, ext_hash))
     }
 
     /// Creates and signs an extrinsic and submits to the chain for block inclusion.

src/error.rs

@@ -63,6 +63,9 @@ pub enum Error {
     /// Events decoding error.
     #[error("Events decoding error: {0}")]
     EventsDecoding(#[from] EventsDecodingError),
+    /// Transaction progress error.
+    #[error("Transaction error: {0}")]
+    Transaction(#[from] TransactionError),
     /// Other error.
     #[error("Other error: {0}")]
     Other(String),
@@ -104,6 +107,9 @@ pub enum RuntimeError {
     /// There are no providers so the account cannot be created.
     #[error("There are no providers so the account cannot be created.")]
     NoProviders,
+    /// There are too many consumers so the account cannot be created.
+    #[error("There are too many consumers so the account cannot be created.")]
+    TooManyConsumers,
     /// Bad origin.
     #[error("Bad origin: throw by ensure_signed, ensure_root or ensure_none.")]
     BadOrigin,
@@ -138,6 +144,7 @@ impl RuntimeError {
             DispatchError::CannotLookup => Ok(Self::CannotLookup),
             DispatchError::ConsumerRemaining => Ok(Self::ConsumerRemaining),
             DispatchError::NoProviders => Ok(Self::NoProviders),
+            DispatchError::TooManyConsumers => Ok(Self::TooManyConsumers),
             DispatchError::Arithmetic(_math_error) => {
                 Ok(Self::Other("math_error".into()))
             }
@@ -158,3 +165,16 @@ pub struct PalletError {
     /// The error description.
     pub description: Vec<String>,
 }
+
+/// Transaction error.
+#[derive(Clone, Debug, Eq, Error, PartialEq)]
+pub enum TransactionError {
+    /// The finality subscription expired (after ~512 blocks we give up if the
+    /// block hasn't yet been finalized).
+    #[error("The finality subscription expired")]
+    FinalitySubscriptionTimeout,
+    /// The block hash that the tranaction was added to could not be found.
+    /// This is probably because the block was retracted before being finalized.
+    #[error("The block containing the transaction can no longer be found (perhaps it was on a non-finalized fork?)")]
+    BlockHashNotFound,
+}

src/events.rs

@@ -19,6 +19,7 @@ use codec::{
     Compact,
     Decode,
     Encode,
+    Error as CodecError,
     Input,
 };
 use std::marker::PhantomData;
@@ -30,9 +31,9 @@ use crate::{
     },
     Config,
     Error,
+    Event,
     Metadata,
     Phase,
-    RuntimeError,
 };
 use scale_info::{
     TypeDef,
@@ -56,6 +57,17 @@ pub struct RawEvent {
     pub data: Bytes,
 }
 
+impl RawEvent {
+    /// Attempt to decode this [`RawEvent`] into a specific event.
+    pub fn as_event<E: Event>(&self) -> Result<Option<E>, CodecError> {
+        if self.pallet == E::PALLET && self.variant == E::EVENT {
+            Ok(Some(E::decode(&mut &self.data[..])?))
+        } else {
+            Ok(None)
+        }
+    }
+}
+
 /// Events decoder.
 #[derive(Debug, Clone)]
 pub struct EventsDecoder<T> {
@@ -76,7 +88,10 @@ where
     }
 
     /// Decode events.
-    pub fn decode_events(&self, input: &mut &[u8]) -> Result<Vec<(Phase, Raw)>, Error> {
+    pub fn decode_events(
+        &self,
+        input: &mut &[u8],
+    ) -> Result<Vec<(Phase, RawEvent)>, Error> {
         let compact_len = <Compact<u32>>::decode(input)?;
         let len = compact_len.0 as usize;
         log::debug!("decoding {} events", len);
@@ -98,13 +113,7 @@ where
             let event_metadata = self.metadata.event(pallet_index, variant_index)?;
 
             let mut event_data = Vec::<u8>::new();
-            let mut event_errors = Vec::<RuntimeError>::new();
-            let result = self.decode_raw_event(
-                event_metadata,
-                input,
-                &mut event_data,
-                &mut event_errors,
-            );
+            let result = self.decode_raw_event(event_metadata, input, &mut event_data);
             let raw = match result {
                 Ok(()) => {
                     log::debug!("raw bytes: {}", hex::encode(&event_data),);
@@ -121,18 +130,11 @@ where
                     let topics = Vec::<T::Hash>::decode(input)?;
                     log::debug!("topics: {:?}", topics);
 
-                    Raw::Event(event)
+                    event
                 }
                 Err(err) => return Err(err),
             };
-
-            if event_errors.is_empty() {
-                r.push((phase.clone(), raw));
-            }
-
-            for err in event_errors {
-                r.push((phase.clone(), Raw::Error(err)));
-            }
+            r.push((phase.clone(), raw));
         }
         Ok(r)
     }
@@ -142,7 +144,6 @@ where
         event_metadata: &EventMetadata,
         input: &mut &[u8],
         output: &mut Vec<u8>,
-        errors: &mut Vec<RuntimeError>,
     ) -> Result<(), Error> {
         log::debug!(
             "Decoding Event '{}::{}'",
@@ -151,24 +152,6 @@ where
         );
         for arg in event_metadata.variant().fields() {
             let type_id = arg.ty().id();
-            if event_metadata.pallet() == "System"
-                && event_metadata.event() == "ExtrinsicFailed"
-            {
-                let ty = self
-                    .metadata
-                    .resolve_type(type_id)
-                    .ok_or(MetadataError::TypeNotFound(type_id))?;
-
-                if ty.path().ident() == Some("DispatchError".to_string()) {
-                    let dispatch_error = sp_runtime::DispatchError::decode(input)?;
-                    log::info!("Dispatch Error {:?}", dispatch_error);
-                    dispatch_error.encode_to(output);
-                    let runtime_error =
-                        RuntimeError::from_dispatch(&self.metadata, dispatch_error)?;
-                    errors.push(runtime_error);
-                    continue
-                }
-            }
             self.decode_type(type_id, input, output)?
         }
         Ok(())
@@ -344,15 +327,6 @@ where
     }
 }
 
-/// Raw event or error event
-#[derive(Debug)]
-pub enum Raw {
-    /// Event
-    Event(RawEvent),
-    /// Error
-    Error(RuntimeError),
-}
-
 #[derive(Debug, thiserror::Error)]
 pub enum EventsDecodingError {
     /// Unsupported primitive type

src/extrinsic/extra.rs

@@ -25,7 +25,10 @@ use core::{
 use scale_info::TypeInfo;
 use sp_runtime::{
     generic::Era,
-    traits::SignedExtension,
+    traits::{
+        DispatchInfoOf,
+        SignedExtension,
+    },
     transaction_validity::TransactionValidityError,
 };
 
@@ -68,6 +71,15 @@ where
     ) -> Result<Self::AdditionalSigned, TransactionValidityError> {
         Ok(self.1)
     }
+    fn pre_dispatch(
+        self,
+        _who: &Self::AccountId,
+        _call: &Self::Call,
+        _info: &DispatchInfoOf<Self::Call>,
+        _len: usize,
+    ) -> Result<Self::Pre, TransactionValidityError> {
+        Ok(())
+    }
 }
 
 /// Ensure the transaction version registered in the transaction is the same as at present.
@@ -99,6 +111,15 @@ where
     ) -> Result<Self::AdditionalSigned, TransactionValidityError> {
         Ok(self.1)
     }
+    fn pre_dispatch(
+        self,
+        _who: &Self::AccountId,
+        _call: &Self::Call,
+        _info: &DispatchInfoOf<Self::Call>,
+        _len: usize,
+    ) -> Result<Self::Pre, TransactionValidityError> {
+        Ok(())
+    }
 }
 
 /// Check genesis hash
@@ -130,6 +151,15 @@ where
     ) -> Result<Self::AdditionalSigned, TransactionValidityError> {
         Ok(self.1)
     }
+    fn pre_dispatch(
+        self,
+        _who: &Self::AccountId,
+        _call: &Self::Call,
+        _info: &DispatchInfoOf<Self::Call>,
+        _len: usize,
+    ) -> Result<Self::Pre, TransactionValidityError> {
+        Ok(())
+    }
 }
 
 /// Check for transaction mortality.
@@ -163,6 +193,15 @@ where
     ) -> Result<Self::AdditionalSigned, TransactionValidityError> {
         Ok(self.1)
     }
+    fn pre_dispatch(
+        self,
+        _who: &Self::AccountId,
+        _call: &Self::Call,
+        _info: &DispatchInfoOf<Self::Call>,
+        _len: usize,
+    ) -> Result<Self::Pre, TransactionValidityError> {
+        Ok(())
+    }
 }
 
 /// Nonce check and increment to give replay protection for transactions.
@@ -184,6 +223,15 @@ where
     ) -> Result<Self::AdditionalSigned, TransactionValidityError> {
         Ok(())
     }
+    fn pre_dispatch(
+        self,
+        _who: &Self::AccountId,
+        _call: &Self::Call,
+        _info: &DispatchInfoOf<Self::Call>,
+        _len: usize,
+    ) -> Result<Self::Pre, TransactionValidityError> {
+        Ok(())
+    }
 }
 
 /// Resource limit check.
@@ -205,6 +253,15 @@ where
     ) -> Result<Self::AdditionalSigned, TransactionValidityError> {
         Ok(())
     }
+    fn pre_dispatch(
+        self,
+        _who: &Self::AccountId,
+        _call: &Self::Call,
+        _info: &DispatchInfoOf<Self::Call>,
+        _len: usize,
+    ) -> Result<Self::Pre, TransactionValidityError> {
+        Ok(())
+    }
 }
 
 /// Require the transactor pay for themselves and maybe include a tip to gain additional priority
@@ -230,6 +287,15 @@ impl SignedExtension for ChargeAssetTxPayment {
     ) -> Result<Self::AdditionalSigned, TransactionValidityError> {
         Ok(())
     }
+    fn pre_dispatch(
+        self,
+        _who: &Self::AccountId,
+        _call: &Self::Call,
+        _info: &DispatchInfoOf<Self::Call>,
+        _len: usize,
+    ) -> Result<Self::Pre, TransactionValidityError> {
+        Ok(())
+    }
 }
 
 /// Trait for implementing transaction extras for a runtime.
@@ -318,4 +384,13 @@ impl<T: Config + Clone + Debug + Eq + Send + Sync> SignedExtension for DefaultEx
     ) -> Result<Self::AdditionalSigned, TransactionValidityError> {
         self.extra().additional_signed()
     }
+    fn pre_dispatch(
+        self,
+        _who: &Self::AccountId,
+        _call: &Self::Call,
+        _info: &DispatchInfoOf<Self::Call>,
+        _len: usize,
+    ) -> Result<Self::Pre, TransactionValidityError> {
+        Ok(())
+    }
 }

src/lib.rs

@@ -45,7 +45,6 @@ pub use subxt_macro::subxt;
 
 pub use bitvec;
 pub use codec;
-pub use sp_arithmetic;
 pub use sp_core;
 pub use sp_runtime;
 
@@ -68,6 +67,7 @@ mod metadata;
 pub mod rpc;
 pub mod storage;
 mod subscription;
+mod transaction;
 
 pub use crate::{
     client::{
@@ -84,6 +84,7 @@ pub use crate::{
         Error,
         PalletError,
         RuntimeError,
+        TransactionError,
     },
     events::{
         EventsDecoder,
@@ -103,7 +104,6 @@ pub use crate::{
     },
     rpc::{
         BlockNumber,
-        ExtrinsicSuccess,
         ReadProof,
         RpcClient,
         SystemProperties,
@@ -119,6 +119,12 @@ pub use crate::{
         EventSubscription,
         FinalizedEventStorageSubscription,
     },
+    transaction::{
+        TransactionEvents,
+        TransactionInBlock,
+        TransactionProgress,
+        TransactionStatus,
+    },
 };
 
 /// Call trait.

src/rpc.rs

@@ -26,7 +26,6 @@ use std::sync::Arc;
 use codec::{
     Decode,
     Encode,
-    Error as CodecError,
 };
 use core::{
     convert::TryInto,
@@ -69,30 +68,21 @@ use sp_core::{
     Bytes,
     U256,
 };
-use sp_runtime::{
-    generic::{
-        Block,
-        SignedBlock,
-    },
-    traits::Hash,
+use sp_runtime::generic::{
+    Block,
+    SignedBlock,
 };
 use sp_version::RuntimeVersion;
 
 use crate::{
     error::Error,
-    events::{
-        EventsDecoder,
-        RawEvent,
-    },
     storage::StorageKeyPrefix,
     subscription::{
         EventStorageSubscription,
-        EventSubscription,
         FinalizedEventStorageSubscription,
         SystemEvents,
     },
     Config,
-    Event,
     Metadata,
 };
 
@@ -154,7 +144,7 @@ pub type SystemProperties = serde_json::Map<String, serde_json::Value>;
 /// must be kept compatible with that type from the target substrate version.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
-pub enum TransactionStatus<Hash, BlockHash> {
+pub enum SubstrateTransactionStatus<Hash, BlockHash> {
     /// Transaction is part of the future queue.
     Future,
     /// Transaction is part of the ready queue.
@@ -199,7 +189,6 @@ pub struct Rpc<T: Config> {
     /// Rpc client for sending requests.
     pub client: Arc<RpcClient>,
     marker: PhantomData<T>,
-    accept_weak_inclusion: bool,
 }
 
 impl<T: Config> Clone for Rpc<T> {
@@ -207,7 +196,6 @@ impl<T: Config> Clone for Rpc<T> {
         Self {
             client: self.client.clone(),
             marker: PhantomData,
-            accept_weak_inclusion: self.accept_weak_inclusion,
         }
     }
 }
@@ -218,16 +206,9 @@ impl<T: Config> Rpc<T> {
         Self {
             client: Arc::new(client),
             marker: PhantomData,
-            accept_weak_inclusion: false,
         }
     }
 
-    /// Configure the Rpc to accept non-finalized blocks
-    /// in `submit_and_watch_extrinsic`
-    pub fn accept_weak_inclusion(&mut self) {
-        self.accept_weak_inclusion = true;
-    }
-
     /// Fetch a storage key
     pub async fn storage(
         &self,
@@ -456,7 +437,7 @@ impl<T: Config> Rpc<T> {
     pub async fn watch_extrinsic<E: Encode>(
         &self,
         extrinsic: E,
-    ) -> Result<Subscription<TransactionStatus<T::Hash, T::Hash>>, Error> {
+    ) -> Result<Subscription<SubstrateTransactionStatus<T::Hash, T::Hash>>, Error> {
         let bytes: Bytes = extrinsic.encode().into();
         let params = rpc_params![bytes];
         let subscription = self
@@ -470,99 +451,6 @@ impl<T: Config> Rpc<T> {
         Ok(subscription)
     }
 
-    /// Create and submit an extrinsic and return corresponding Event if successful
-    pub async fn submit_and_watch_extrinsic<'a, E: Encode + 'static>(
-        &self,
-        extrinsic: E,
-        decoder: &'a EventsDecoder<T>,
-    ) -> Result<ExtrinsicSuccess<T>, Error> {
-        let ext_hash = T::Hashing::hash_of(&extrinsic);
-        log::info!("Submitting Extrinsic `{:?}`", ext_hash);
-
-        let events_sub = if self.accept_weak_inclusion {
-            self.subscribe_events().await
-        } else {
-            self.subscribe_finalized_events().await
-        }?;
-        let mut xt_sub = self.watch_extrinsic(extrinsic).await?;
-
-        while let Some(status) = xt_sub.next().await {
-            log::info!("Received status {:?}", status);
-            let status = status?;
-            match status {
-                TransactionStatus::Future
-                | TransactionStatus::Ready
-                | TransactionStatus::Broadcast(_)
-                | TransactionStatus::Retracted(_) => continue,
-                TransactionStatus::InBlock(block_hash) => {
-                    if self.accept_weak_inclusion {
-                        return self
-                            .process_block(events_sub, decoder, block_hash, ext_hash)
-                            .await
-                    }
-                    continue
-                }
-                TransactionStatus::Invalid => return Err("Extrinsic Invalid".into()),
-                TransactionStatus::Usurped(_) => return Err("Extrinsic Usurped".into()),
-                TransactionStatus::Dropped => return Err("Extrinsic Dropped".into()),
-                TransactionStatus::Finalized(block_hash) => {
-                    // read finalized blocks by default
-                    return self
-                        .process_block(events_sub, decoder, block_hash, ext_hash)
-                        .await
-                }
-                TransactionStatus::FinalityTimeout(_) => {
-                    return Err("Extrinsic FinalityTimeout".into())
-                }
-            }
-        }
-        Err(RpcError::Custom("RPC subscription dropped".into()).into())
-    }
-
-    async fn process_block(
-        &self,
-        events_sub: EventStorageSubscription<T>,
-        decoder: &EventsDecoder<T>,
-        block_hash: T::Hash,
-        ext_hash: T::Hash,
-    ) -> Result<ExtrinsicSuccess<T>, Error> {
-        log::info!("Fetching block {:?}", block_hash);
-        if let Some(signed_block) = self.block(Some(block_hash)).await? {
-            log::info!(
-                "Found block {:?}, with {} extrinsics",
-                block_hash,
-                signed_block.block.extrinsics.len()
-            );
-            let ext_index = signed_block
-                .block
-                .extrinsics
-                .iter()
-                .position(|ext| {
-                    let hash = T::Hashing::hash_of(ext);
-                    hash == ext_hash
-                })
-                .ok_or_else(|| {
-                    Error::Other(format!(
-                        "Failed to find Extrinsic with hash {:?}",
-                        ext_hash,
-                    ))
-                })?;
-            let mut sub = EventSubscription::new(events_sub, decoder);
-            sub.filter_extrinsic(block_hash, ext_index);
-            let mut events = vec![];
-            while let Some(event) = sub.next().await {
-                events.push(event?);
-            }
-            Ok(ExtrinsicSuccess {
-                block: block_hash,
-                extrinsic: ext_hash,
-                events,
-            })
-        } else {
-            Err(format!("Failed to find block {:?}", block_hash).into())
-        }
-    }
-
     /// Insert a key into the keystore.
     pub async fn insert_key(
         &self,
@@ -606,38 +494,6 @@ impl<T: Config> Rpc<T> {
     }
 }
 
-/// Captures data for when an extrinsic is successfully included in a block
-#[derive(Debug)]
-pub struct ExtrinsicSuccess<T: Config> {
-    /// Block hash.
-    pub block: T::Hash,
-    /// Extrinsic hash.
-    pub extrinsic: T::Hash,
-    /// Raw runtime events, can be decoded by the caller.
-    pub events: Vec<RawEvent>,
-}
-
-impl<T: Config> ExtrinsicSuccess<T> {
-    /// Find the Event for the given module/variant, with raw encoded event data.
-    /// Returns `None` if the Event is not found.
-    pub fn find_event_raw(&self, module: &str, variant: &str) -> Option<&RawEvent> {
-        self.events
-            .iter()
-            .find(|raw| raw.pallet == module && raw.variant == variant)
-    }
-
-    /// Find the Event for the given module/variant, attempting to decode the event data.
-    /// Returns `None` if the Event is not found.
-    /// Returns `Err` if the data fails to decode into the supplied type.
-    pub fn find_event<E: Event>(&self) -> Result<Option<E>, CodecError> {
-        if let Some(event) = self.find_event_raw(E::PALLET, E::EVENT) {
-            Ok(Some(E::decode(&mut &event.data[..])?))
-        } else {
-            Ok(None)
-        }
-    }
-}
-
 /// Build WS RPC client from URL
 pub async fn build_ws_client(url: &str) -> Result<RpcClient, RpcError> {
     let (sender, receiver) = ws_transport(url).await?;

src/subscription.rs

@@ -29,7 +29,6 @@ use crate::{
     error::Error,
     events::{
         EventsDecoder,
-        Raw,
         RawEvent,
     },
     rpc::Rpc,
@@ -45,7 +44,7 @@ pub struct EventSubscription<'a, T: Config> {
     block: Option<T::Hash>,
     extrinsic: Option<usize>,
     event: Option<(&'static str, &'static str)>,
-    events: VecDeque<Raw>,
+    events: VecDeque<RawEvent>,
     finished: bool,
 }
 
@@ -56,11 +55,11 @@ enum BlockReader<'a, T: Config> {
     },
     /// Mock event listener for unit tests
     #[cfg(test)]
-    Mock(Box<dyn Iterator<Item = (T::Hash, Result<Vec<(Phase, Raw)>, Error>)>>),
+    Mock(Box<dyn Iterator<Item = (T::Hash, Result<Vec<(Phase, RawEvent)>, Error>)>>),
 }
 
 impl<'a, T: Config> BlockReader<'a, T> {
-    async fn next(&mut self) -> Option<(T::Hash, Result<Vec<(Phase, Raw)>, Error>)> {
+    async fn next(&mut self) -> Option<(T::Hash, Result<Vec<(Phase, RawEvent)>, Error>)> {
         match self {
             BlockReader::Decoder {
                 subscription,
@@ -123,10 +122,7 @@ impl<'a, T: Config> EventSubscription<'a, T> {
     pub async fn next(&mut self) -> Option<Result<RawEvent, Error>> {
         loop {
             if let Some(raw_event) = self.events.pop_front() {
-                match raw_event {
-                    Raw::Event(event) => return Some(Ok(event)),
-                    Raw::Error(err) => return Some(Err(err.into())),
-                };
+                return Some(Ok(raw_event))
             }
             if self.finished {
                 return None
@@ -152,10 +148,8 @@ impl<'a, T: Config> EventSubscription<'a, T> {
                             }
                         }
                         if let Some((module, variant)) = self.event {
-                            if let Raw::Event(ref event) = raw {
-                                if event.pallet != module || event.variant != variant {
-                                    continue
-                                }
+                            if raw.pallet != module || raw.variant != variant {
+                                continue
                             }
                         }
                         self.events.push_back(raw);
@@ -261,8 +255,6 @@ where
 
 #[cfg(test)]
 mod tests {
-    use crate::RuntimeError;
-
     use super::*;
     use sp_core::H256;
     #[derive(Clone)]
@@ -293,51 +285,6 @@ mod tests {
         }
     }
 
-    fn raw_event(id: u8) -> RawEvent {
-        RawEvent {
-            data: sp_core::Bytes::from(Vec::new()),
-            pallet: "SomePallet".to_string(),
-            variant: "SomeVariant".to_string(),
-            pallet_index: id,
-            variant_index: id,
-        }
-    }
-
-    fn event(id: u8) -> Raw {
-        Raw::Event(raw_event(id))
-    }
-
-    #[async_std::test]
-    async fn test_error_does_not_stop_subscription() {
-        let mut subscription: EventSubscription<MockConfig> = EventSubscription {
-            block_reader: BlockReader::Mock(Box::new(
-                vec![(
-                    H256::from([0; 32]),
-                    Ok(vec![
-                        (
-                            Phase::ApplyExtrinsic(0),
-                            Raw::Error(RuntimeError::BadOrigin),
-                        ),
-                        (Phase::ApplyExtrinsic(0), event(1)),
-                    ]),
-                )]
-                .into_iter(),
-            )),
-            block: None,
-            extrinsic: None,
-            event: None,
-            events: Default::default(),
-            finished: false,
-        };
-
-        assert!(matches!(
-            subscription.next().await.unwrap().unwrap_err(),
-            Error::Runtime(RuntimeError::BadOrigin)
-        ));
-        assert_eq!(subscription.next().await.unwrap().unwrap(), raw_event(1));
-        assert!(subscription.next().await.is_none());
-    }
-
     #[async_std::test]
     /// test that filters work correctly, and are independent of each other
     async fn test_filters() {
@@ -375,7 +322,7 @@ mod tests {
                                             .iter()
                                             .take(half_len)
                                             .map(|(_, phase, event)| {
-                                                (phase.clone(), Raw::Event(event.clone()))
+                                                (phase.clone(), event.clone())
                                             })
                                             .collect()),
                                     ),
@@ -385,7 +332,7 @@ mod tests {
                                             .iter()
                                             .skip(half_len)
                                             .map(|(_, phase, event)| {
-                                                (phase.clone(), Raw::Event(event.clone()))
+                                                (phase.clone(), event.clone())
                                             })
                                             .collect()),
                                     ),

src/transaction.rs

@@ -0,0 +1,456 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+// This file is part of subxt.
+//
+// subxt 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.
+//
+// subxt 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 subxt.  If not, see <http://www.gnu.org/licenses/>.
+
+use sp_core::storage::StorageKey;
+use sp_runtime::traits::Hash;
+pub use sp_runtime::traits::SignedExtension;
+pub use sp_version::RuntimeVersion;
+
+use crate::{
+    client::Client,
+    error::{
+        Error,
+        TransactionError,
+    },
+    rpc::SubstrateTransactionStatus,
+    subscription::SystemEvents,
+    Config,
+    Phase,
+};
+use jsonrpsee::core::client::Subscription as RpcSubscription;
+
+/// This struct represents a subscription to the progress of some transaction, and is
+/// returned from [`crate::SubmittableExtrinsic::sign_and_submit_then_watch()`].
+#[derive(Debug)]
+pub struct TransactionProgress<'client, T: Config> {
+    sub: Option<RpcSubscription<SubstrateTransactionStatus<T::Hash, T::Hash>>>,
+    ext_hash: T::Hash,
+    client: &'client Client<T>,
+}
+
+impl<'client, T: Config> TransactionProgress<'client, T> {
+    pub(crate) fn new(
+        sub: RpcSubscription<SubstrateTransactionStatus<T::Hash, T::Hash>>,
+        client: &'client Client<T>,
+        ext_hash: T::Hash,
+    ) -> Self {
+        Self {
+            sub: Some(sub),
+            client,
+            ext_hash,
+        }
+    }
+
+    /// Return the next transaction status when it's emitted.
+    pub async fn next(&mut self) -> Option<Result<TransactionStatus<'client, T>, Error>> {
+        // Return `None` if the subscription has been dropped:
+        let sub = match &mut self.sub {
+            Some(sub) => sub,
+            None => return None,
+        };
+
+        // Return the next item otherwise:
+        let res = sub.next().await?;
+        Some(res.map(|status| {
+            match status {
+                SubstrateTransactionStatus::Future => TransactionStatus::Future,
+                SubstrateTransactionStatus::Ready => TransactionStatus::Ready,
+                SubstrateTransactionStatus::Broadcast(peers) => {
+                    TransactionStatus::Broadcast(peers)
+                }
+                SubstrateTransactionStatus::InBlock(hash) => {
+                    TransactionStatus::InBlock(TransactionInBlock {
+                        block_hash: hash,
+                        ext_hash: self.ext_hash,
+                        client: self.client,
+                    })
+                }
+                SubstrateTransactionStatus::Retracted(hash) => {
+                    TransactionStatus::Retracted(hash)
+                }
+                SubstrateTransactionStatus::Usurped(hash) => {
+                    TransactionStatus::Usurped(hash)
+                }
+                SubstrateTransactionStatus::Dropped => TransactionStatus::Dropped,
+                SubstrateTransactionStatus::Invalid => TransactionStatus::Invalid,
+                // Only the following statuses are actually considered "final" (see the substrate
+                // docs on `TransactionStatus`). Basically, either the transaction makes it into a
+                // block, or we eventually give up on waiting for it to make it into a block.
+                // Even `Dropped`/`Invalid`/`Usurped` transactions might make it into a block eventually.
+                //
+                // As an example, a transaction that is `Invalid` on one node due to having the wrong
+                // nonce might still be valid on some fork on another node which ends up being finalized.
+                // Equally, a transaction `Dropped` from one node may still be in the transaction pool,
+                // and make it into a block, on another node. Likewise with `Usurped`.
+                SubstrateTransactionStatus::FinalityTimeout(hash) => {
+                    self.sub = None;
+                    TransactionStatus::FinalityTimeout(hash)
+                }
+                SubstrateTransactionStatus::Finalized(hash) => {
+                    self.sub = None;
+                    TransactionStatus::Finalized(TransactionInBlock {
+                        block_hash: hash,
+                        ext_hash: self.ext_hash,
+                        client: self.client,
+                    })
+                }
+            }
+        }).map_err(Into::into))
+    }
+
+    /// Wait for the transaction to be in a block (but not necessarily finalized), and return
+    /// an [`TransactionInBlock`] instance when this happens, or an error if there was a problem
+    /// waiting for this to happen.
+    ///
+    /// **Note:** consumes `self`. If you'd like to perform multiple actions as the state of the
+    /// transaction progresses, use [`TransactionProgress::next()`] instead.
+    ///
+    /// **Note:** transaction statuses like `Invalid` and `Usurped` are ignored, because while they
+    /// may well indicate with some probability that the transaction will not make it into a block,
+    /// there is no guarantee that this is true. Thus, we prefer to "play it safe" here. Use the lower
+    /// level [`TransactionProgress::next()`] API if you'd like to handle these statuses yourself.
+    pub async fn wait_for_in_block(
+        mut self,
+    ) -> Option<Result<TransactionInBlock<'client, T>, Error>> {
+        loop {
+            match self.next().await? {
+                Ok(status) => match status {
+                    // Finalized or otherwise in a block! Return.
+                    TransactionStatus::InBlock(s) | TransactionStatus::Finalized(s) => {
+                        return Some(Ok(s))
+                    }
+                    // Error scenarios; return the error.
+                    TransactionStatus::FinalityTimeout(_) => {
+                        return Some(Err(TransactionError::FinalitySubscriptionTimeout.into()))
+                    }
+                    // Ignore anything else and wait for next status event:
+                    _ => continue,
+                }
+                Err(err) => return Some(Err(err)),
+            }
+        }
+    }
+
+    /// Wait for the transaction to be finalized, and return a [`TransactionInBlock`]
+    /// instance when it is, or an error if there was a problem waiting for finalization.
+    ///
+    /// **Note:** consumes `self`. If you'd like to perform multiple actions as the state of the
+    /// transaction progresses, use [`TransactionProgress::next()`] instead.
+    ///
+    /// **Note:** transaction statuses like `Invalid` and `Usurped` are ignored, because while they
+    /// may well indicate with some probability that the transaction will not make it into a block,
+    /// there is no guarantee that this is true. Thus, we prefer to "play it safe" here. Use the lower
+    /// level [`TransactionProgress::next()`] API if you'd like to handle these statuses yourself.
+    pub async fn wait_for_finalized(
+        mut self,
+    ) -> Option<Result<TransactionInBlock<'client, T>, Error>> {
+        loop {
+            match self.next().await? {
+                Ok(status) => match status {
+                    // finalized! return.
+                    TransactionStatus::Finalized(s) => return Some(Ok(s)),
+                    // error scenarios; return the error.
+                    TransactionStatus::FinalityTimeout(_) => {
+                        return Some(Err(TransactionError::FinalitySubscriptionTimeout.into()))
+                    }
+                    // ignore and wait for next status event:
+                    _ => continue,
+                }
+                Err(err) => return Some(Err(err)),
+            }
+        }
+    }
+
+    /// Wait for the transaction to be finalized, and for the transaction events to indicate
+    /// that the transaction was successful. Returns the events associated with the transaction,
+    /// as well as a couple of other details (block hash and extrinsic hash).
+    ///
+    /// **Note:** consumes self. If you'd like to perform multiple actions as progress is made,
+    /// use [`TransactionProgress::next()`] instead.
+    ///
+    /// **Note:** transaction statuses like `Invalid` and `Usurped` are ignored, because while they
+    /// may well indicate with some probability that the transaction will not make it into a block,
+    /// there is no guarantee that this is true. Thus, we prefer to "play it safe" here. Use the lower
+    /// level [`TransactionProgress::next()`] API if you'd like to handle these statuses yourself.
+    pub async fn wait_for_finalized_success(self) -> Option<Result<TransactionEvents<T>, Error>> {
+        let finalized = match self.wait_for_finalized().await? {
+            Ok(f) => f,
+            Err(err) => return Some(Err(err)),
+        };
+
+        Some(finalized.wait_for_success().await)
+    }
+}
+
+//* Dev note: The below is adapted from the substrate docs on `TransactionStatus`, which this
+//* enum was adapted from (and which is an exact copy of `SubstrateTransactionStatus` in this crate).
+//* Note that the number of finality watchers is, at the time of writing, found in the constant
+//* `MAX_FINALITY_WATCHERS` in the `sc_transaction_pool` crate.
+//*
+/// Possible transaction statuses returned from our [`TransactionProgress::next()`] call.
+///
+/// These status events can be grouped based on their kinds as:
+///
+/// 1. Entering/Moving within the pool:
+///    - `Future`
+///    - `Ready`
+/// 2. Inside `Ready` queue:
+///    - `Broadcast`
+/// 3. Leaving the pool:
+///    - `InBlock`
+///    - `Invalid`
+///    - `Usurped`
+///    - `Dropped`
+/// 4. Re-entering the pool:
+///    - `Retracted`
+/// 5. Block finalized:
+///    - `Finalized`
+///    - `FinalityTimeout`
+///
+/// The events will always be received in the order described above, however
+/// there might be cases where transactions alternate between `Future` and `Ready`
+/// pool, and are `Broadcast` in the meantime.
+///
+/// Note that there are conditions that may cause transactions to reappear in the pool:
+///
+/// 1. Due to possible forks, the transaction that ends up being included
+///    in one block may later re-enter the pool or be marked as invalid.
+/// 2. A transaction that is `Dropped` at one point may later re-enter the pool if
+///    some other transactions are removed.
+/// 3. `Invalid` transactions may become valid at some point in the future.
+///    (Note that runtimes are encouraged to use `UnknownValidity` to inform the
+///    pool about such cases).
+/// 4. `Retracted` transactions might be included in a future block.
+///
+/// The stream is considered finished only when either the `Finalized` or `FinalityTimeout`
+/// event is triggered. You are however free to unsubscribe from notifications at any point.
+/// The first one will be emitted when the block in which the transaction was included gets
+/// finalized. The `FinalityTimeout` event will be emitted when the block did not reach finality
+/// within 512 blocks. This either indicates that finality is not available for your chain,
+/// or that finality gadget is lagging behind.
+#[derive(Debug)]
+pub enum TransactionStatus<'client, T: Config> {
+    /// The transaction is part of the "future" queue.
+    Future,
+    /// The transaction is part of the "ready" queue.
+    Ready,
+    /// The transaction has been broadcast to the given peers.
+    Broadcast(Vec<String>),
+    /// The transaction has been included in a block with given hash.
+    InBlock(TransactionInBlock<'client, T>),
+    /// The block this transaction was included in has been retracted,
+    /// probably because it did not make it onto the blocks which were
+    /// finalized.
+    Retracted(T::Hash),
+    /// A block containing the transaction did not reach finality within 512
+    /// blocks, and so the subscription has ended.
+    FinalityTimeout(T::Hash),
+    /// The transaction has been finalized by a finality-gadget, e.g GRANDPA.
+    Finalized(TransactionInBlock<'client, T>),
+    /// The transaction has been replaced in the pool by another transaction
+    /// that provides the same tags. (e.g. same (sender, nonce)).
+    Usurped(T::Hash),
+    /// The transaction has been dropped from the pool because of the limit.
+    Dropped,
+    /// The transaction is no longer valid in the current state.
+    Invalid,
+}
+
+impl<'client, T: Config> TransactionStatus<'client, T> {
+    /// A convenience method to return the `Finalized` details. Returns
+    /// [`None`] if the enum variant is not [`TransactionStatus::Finalized`].
+    pub fn as_finalized(&self) -> Option<&TransactionInBlock<'client, T>> {
+        match self {
+            Self::Finalized(val) => Some(val),
+            _ => None,
+        }
+    }
+
+    /// A convenience method to return the `InBlock` details. Returns
+    /// [`None`] if the enum variant is not [`TransactionStatus::InBlock`].
+    pub fn as_in_block(&self) -> Option<&TransactionInBlock<'client, T>> {
+        match self {
+            Self::InBlock(val) => Some(val),
+            _ => None,
+        }
+    }
+}
+
+/// This struct represents a transaction that has made it into a block.
+#[derive(Debug)]
+pub struct TransactionInBlock<'client, T: Config> {
+    block_hash: T::Hash,
+    ext_hash: T::Hash,
+    client: &'client Client<T>,
+}
+
+impl<'client, T: Config> TransactionInBlock<'client, T> {
+    /// Return the hash of the block that the transaction has made it into.
+    pub fn block_hash(&self) -> T::Hash {
+        self.block_hash
+    }
+
+    /// Return the hash of the extrinsic that was submitted.
+    pub fn extrinsic_hash(&self) -> T::Hash {
+        self.ext_hash
+    }
+
+    /// Fetch the events associated with this transaction. If the transaction
+    /// was successful (ie no `ExtrinsicFailed`) events were found, then we return
+    /// the events associated with it. If the transaction was not successful, or
+    /// something else went wrong, we return an error.
+    ///
+    /// **Note:** If multiple `ExtrinsicFailed` errors are returned (for instance
+    /// because a pallet chooses to emit one as an event, which is considered
+    /// abnormal behaviour), it is not specified which of the errors is returned here.
+    /// You can use [`TransactionInBlock::fetch_events`] instead if you'd like to
+    /// work with multiple "error" events.
+    ///
+    /// **Note:** This has to download block details from the node and decode events
+    /// from them.
+    pub async fn wait_for_success(&self) -> Result<TransactionEvents<T>, Error> {
+        let events = self.fetch_events().await?;
+
+        // Try to find any errors; return the first one we encounter.
+        for ev in events.as_slice() {
+            if &ev.pallet == "System" && &ev.variant == "ExtrinsicFailed" {
+                use codec::Decode;
+                let dispatch_error = sp_runtime::DispatchError::decode(&mut &*ev.data)?;
+                let runtime_error = crate::RuntimeError::from_dispatch(
+                    self.client.metadata(),
+                    dispatch_error,
+                )?;
+                return Err(runtime_error.into())
+            }
+        }
+
+        Ok(events)
+    }
+
+    /// Fetch all of the events associated with this transaction. This succeeds whether
+    /// the transaction was a success or not; it's up to you to handle the error and
+    /// success events however you prefer.
+    ///
+    /// **Note:** This has to download block details from the node and decode events
+    /// from them.
+    pub async fn fetch_events(&self) -> Result<TransactionEvents<T>, Error> {
+        let block = self
+            .client
+            .rpc()
+            .block(Some(self.block_hash))
+            .await?
+            .ok_or(Error::Transaction(TransactionError::BlockHashNotFound))?;
+
+        let extrinsic_idx = block.block.extrinsics
+            .iter()
+            .position(|ext| {
+                let hash = T::Hashing::hash_of(ext);
+                hash == self.ext_hash
+            })
+            // If we successfully obtain the block hash we think contains our
+            // extrinsic, the extrinsic should be in there somewhere..
+            .ok_or(Error::Transaction(TransactionError::BlockHashNotFound))?;
+
+        let raw_events = self
+            .client
+            .rpc()
+            .storage(
+                &StorageKey::from(SystemEvents::new()),
+                Some(self.block_hash),
+            )
+            .await?
+            .map(|s| s.0)
+            .unwrap_or_else(Vec::new);
+
+        let events = self
+            .client
+            .events_decoder()
+            .decode_events(&mut &*raw_events)?
+            .into_iter()
+            .filter(move |(phase, _raw)| {
+                phase == &Phase::ApplyExtrinsic(extrinsic_idx as u32)
+            })
+            .map(|(_phase, event)| event)
+            .collect();
+
+        Ok(TransactionEvents {
+            block_hash: self.block_hash,
+            ext_hash: self.ext_hash,
+            events,
+        })
+    }
+}
+
+/// This represents the events related to our transaction.
+/// We can iterate over the events, or look for a specific one.
+#[derive(Debug)]
+pub struct TransactionEvents<T: Config> {
+    block_hash: T::Hash,
+    ext_hash: T::Hash,
+    events: Vec<crate::RawEvent>,
+}
+
+impl<T: Config> TransactionEvents<T> {
+    /// Return the hash of the block that the transaction has made it into.
+    pub fn block_hash(&self) -> T::Hash {
+        self.block_hash
+    }
+
+    /// Return the hash of the extrinsic.
+    pub fn extrinsic_hash(&self) -> T::Hash {
+        self.ext_hash
+    }
+
+    /// Return a slice of the returned events.
+    pub fn as_slice(&self) -> &[crate::RawEvent] {
+        &self.events
+    }
+
+    /// Find all of the events matching the event type provided as a generic parameter. This
+    /// will return an error if a matching event is found but cannot be properly decoded.
+    pub fn find_events<E: crate::Event>(&self) -> Result<Vec<E>, Error> {
+        self.events
+            .iter()
+            .filter_map(|e| e.as_event::<E>().map_err(Into::into).transpose())
+            .collect()
+    }
+
+    /// Find the first event that matches the event type provided as a generic parameter. This
+    /// will return an error if a matching event is found but cannot be properly decoded.
+    ///
+    /// Use [`TransactionEvents::find_events`], or iterate over [`TransactionEvents`] yourself
+    /// if you'd like to handle multiple events of the same type.
+    pub fn find_first_event<E: crate::Event>(&self) -> Result<Option<E>, Error> {
+        self.events
+            .iter()
+            .filter_map(|e| e.as_event::<E>().transpose())
+            .next()
+            .transpose()
+            .map_err(Into::into)
+    }
+
+    /// Find an event. Returns true if it was found.
+    pub fn has_event<E: crate::Event>(&self) -> Result<bool, Error> {
+        Ok(self.find_first_event::<E>()?.is_some())
+    }
+}
+
+impl<T: Config> std::ops::Deref for TransactionEvents<T> {
+    type Target = [crate::RawEvent];
+    fn deref(&self) -> &Self::Target {
+        &self.events
+    }
+}