Implement new API for sign_and_submit_then_watch (#354)

* WIP Implementing new event subscription API

* back to lifetimes, fix example

* no more need for accept_weak_inclusion

* thread lifetime through to prevent 'temporary dropped' issue

* make working with events a little nicer

* Get tests compiling

* fmt and clippy

* _name back to name

* dont take ownership, just have stronger note

* Attempt to fix test

* remove commented-out code

* Add a couple more helper methods and a test

* Remove custom ExtrinsicFailed handling; treat them like other events

* Handle runtime errors in TransactionProgress related bits

* cargo fmt + clippy

* Fix some of the failing tests

* remove unused import

* fix transfer_error test

* Fix compile errors against new substrate latest

* Comment tweaks, and force test-runtime rebuild

* Drop the TransactionProgress subscription when we hit 'end' statuses

* cargo fmt

* find_event to find_first_event and helper to return all matching events

* TransactionProgressStatus to TransactionStatus

* Copy and improve docs on TransactionStatus from substrate

* debug impl for Client to avoid manual debug impls elsewhere

* Add and tweak comments, specifically a note about block inclusion on errors

* clippy + fmt

* Fix docs

* Ignore 'error' statuses and adhere to the substrate docs

* tweak and improve some comments per @dvdplm's suggestions

* Break transaction* structs into separate file

* fmt and fix doc link

codegen/src/api/calls.rs

@@ -68,7 +68,7 @@ pub fn generate_calls(
                 pub fn #fn_name(
                     &self,
                     #( #call_fn_args, )*
-                ) -> ::subxt::SubmittableExtrinsic<T, #call_struct_name> {
+                ) -> ::subxt::SubmittableExtrinsic<'a, T, #call_struct_name> {
                     let call = #call_struct_name { #( #call_args, )* };
                     ::subxt::SubmittableExtrinsic::new(self.client, call)
                 }

codegen/src/ir.rs

@@ -72,6 +72,7 @@ impl ItemMod {
     }
 }
 
+#[allow(clippy::large_enum_variant)]
 #[derive(Debug, PartialEq, Eq)]
 #[allow(clippy::large_enum_variant)]
 pub enum Item {

examples/submit_and_watch.rs

@@ -42,14 +42,20 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .build()
         .await?
         .to_runtime_api::<polkadot::RuntimeApi<polkadot::DefaultConfig>>();
-    let result = api
+
+    let balance_transfer = api
         .tx()
         .balances()
         .transfer(dest, 10_000)
         .sign_and_submit_then_watch(&signer)
+        .await?
+        .wait_for_finalized_success()
         .await?;
 
-    if let Some(event) = result.find_event::<polkadot::balances::events::Transfer>()? {
+    let transfer_event =
+        balance_transfer.find_first_event::<polkadot::balances::events::Transfer>()?;
+
+    if let Some(event) = transfer_event {
         println!("Balance transfer success: value: {:?}", event.2);
     } else {
         println!("Failed to find Balances::Transfer Event");

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,27 +29,26 @@ 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)]
 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");
             RpcClient::try_from_url(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),
@@ -158,3 +161,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