Merge pull request #324 from RGB-WG/async

Async support for wallet and witness syncrhonization

Author: dr-orlovsky

.github/workflows/build.yml

@@ -33,7 +33,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        feature: [ bitcoin, liquid, prime, uri, binfile, serde, stl ]
+        feature: [ async, bitcoin, liquid, prime, uri, binfile, serde, stl ]
     steps:
       - uses: actions/checkout@v4
       - uses: dtolnay/rust-toolchain@stable

Cargo.toml

@@ -83,8 +83,9 @@ rgb-persist-fs.workspace = true
 
 [features]
 default = ["std", "bitcoin"]
-all = ["std", "bitcoin", "liquid", "prime", "binfile", "uri", "stl", "serde"]
+all = ["std", "bitcoin", "liquid", "prime", "binfile", "uri", "stl", "serde", "async"]
 std = ["rgb-invoice/std", "indexmap/std"]
+async = []
 
 bitcoin = ["bp-core", "bp-invoice", "rgb-core/bitcoin", "rgb-invoice/bitcoin"]
 liquid = ["rgb-core/liquid", "rgb-invoice/liquid"]

src/contracts.rs

@@ -126,6 +126,27 @@ where
             panic!("Contract {id} not found")
         }
     }
+
+    #[cfg(feature = "async")]
+    #[allow(clippy::await_holding_refcell_ref)]
+    async fn with_contract_mut_async<R>(
+        &mut self,
+        id: ContractId,
+        f: impl AsyncFnOnce(&mut Contract<Sp::Stock, Sp::Pile>) -> R,
+    ) -> R {
+        // We need this bullshit due to a failed rust `RefCell` implementation which panics if we do
+        // this block any other way.
+        if self.contracts.borrow().contains_key(&id) {
+            return f(self.contracts.borrow_mut().get_mut(&id).unwrap()).await;
+        }
+        if let Some(mut contract) = self.persistence.contract(id) {
+            let res = f(&mut contract).await;
+            self.contracts.borrow_mut().insert(id, contract);
+            res
+        } else {
+            panic!("Contract {id} not found")
+        }
+    }
 }
 
 impl<Sp, S, C> Contracts<Sp, S, C>
@@ -270,13 +291,16 @@ where
         self.with_contract_mut(contract_id, |contract| contract.call(call, seals))
     }
 
-    /// Synchronize the status of all witnesses and single-use seal definitions.
+    #[cfg(not(feature = "async"))]
+    /// Update the status of all witnesses and single-use seal definitions.
     ///
     /// Applies rollbacks or forwards if required and recomputes the state of the affected
     /// contracts.
-    pub fn sync_witnesses<E: core::error::Error>(
+    pub fn update_witnesses<E: core::error::Error>(
         &mut self,
         resolver: impl Fn(<<Sp::Pile as Pile>::Seal as RgbSeal>::WitnessId) -> Result<WitnessStatus, E>,
+        last_block_height: u64,
+        min_conformations: u32,
     ) -> Result<(), MultiError<SyncError<E>, <Sp::Stock as Stock>::Error>> {
         let mut changed_statuses = IndexMap::<_, WitnessStatus>::new();
         let contract_ids = self.persistence.contract_ids().collect::<IndexSet<_>>();
@@ -285,13 +309,16 @@ where
                 contract_id,
                 |contract| -> Result<(), MultiError<SyncError<E>, <Sp::Stock as Stock>::Error>> {
                     for witness_id in contract.witness_ids() {
+                        let old_status = contract.witness_status(witness_id);
+                        if matches!(old_status, WitnessStatus::Mined(height) if last_block_height - height.get() > min_conformations as u64) {
+                            continue
+                        }
                         let new_status = match changed_statuses.get(&witness_id) {
                             None => resolver(witness_id)
                                 .map_err(SyncError::Status)
                                 .map_err(MultiError::A),
                             Some(witness_id) => Ok(*witness_id),
                         }?;
-                        let old_status = contract.witness_status(witness_id);
                         if new_status != old_status {
                             changed_statuses.insert(witness_id, new_status);
                         }
@@ -306,6 +333,51 @@ where
         Ok(())
     }
 
+    #[cfg(feature = "async")]
+    /// Update the status of all witnesses and single-use seal definitions.
+    ///
+    /// Applies rollbacks or forwards if required and recomputes the state of the affected
+    /// contracts.
+    pub async fn update_witnesses_async<E: core::error::Error>(
+        &mut self,
+        resolver: impl AsyncFn(
+            <<Sp::Pile as Pile>::Seal as RgbSeal>::WitnessId,
+        ) -> Result<WitnessStatus, E>,
+        last_block_height: u64,
+        min_conformations: u32,
+    ) -> Result<(), MultiError<SyncError<E>, <Sp::Stock as Stock>::Error>> {
+        let mut changed_statuses = IndexMap::<_, WitnessStatus>::new();
+        let contract_ids = self.persistence.contract_ids().collect::<IndexSet<_>>();
+        for contract_id in contract_ids {
+            self.with_contract_mut_async(
+                contract_id,
+                async |contract| -> Result<(), MultiError<SyncError<E>, <Sp::Stock as Stock>::Error>> {
+                    for witness_id in contract.witness_ids() {
+                        let old_status = contract.witness_status(witness_id);
+                        if matches!(old_status, WitnessStatus::Mined(height) if last_block_height - height.get() > min_conformations as u64) {
+                            continue
+                        }
+                        let new_status = match changed_statuses.get(&witness_id) {
+                            None => resolver(witness_id)
+                                .await
+                                .map_err(SyncError::Status)
+                                .map_err(MultiError::A),
+                            Some(witness_id) => Ok(*witness_id),
+                        }?;
+                        if new_status != old_status {
+                            changed_statuses.insert(witness_id, new_status);
+                        }
+                    }
+                    contract
+                        .sync(changed_statuses.iter().map(|(id, status)| (*id, *status)))
+                        .map_err(MultiError::from_other_a)?;
+                    Ok(())
+                },
+            ).await?;
+        }
+        Ok(())
+    }
+
     /// Include an operation and its witness to the history of known operations and the contract
     /// state.
     ///

src/lib.rs

@@ -40,6 +40,7 @@
 #![cfg_attr(coverage_nightly, feature(coverage_attribute))]
 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
 #![allow(clippy::type_complexity)]
+#![cfg_attr(feature = "async", allow(async_fn_in_trait))]
 
 extern crate alloc;
 

src/popls/bp.rs

@@ -64,7 +64,11 @@ pub trait WalletProvider {
 
     fn has_utxo(&self, outpoint: Outpoint) -> bool;
     fn utxos(&self) -> impl Iterator<Item = Outpoint>;
-    fn sync_utxos(&mut self) -> Result<(), Self::Error>;
+
+    #[cfg(not(feature = "async"))]
+    fn update_utxos(&mut self) -> Result<(), Self::Error>;
+    #[cfg(feature = "async")]
+    async fn update_utxos_async(&mut self) -> Result<(), Self::Error>;
 
     fn register_seal(&mut self, seal: WTxoSeal);
     fn resolve_seals(
@@ -76,12 +80,32 @@ pub trait WalletProvider {
     fn next_address(&mut self) -> Address;
     fn next_nonce(&mut self) -> u64;
 
+    #[cfg(not(feature = "async"))]
     /// Returns a closure which can retrieve a witness status of an arbitrary transaction id
     /// (including the ones that are not related to the wallet).
     fn txid_resolver(&self) -> impl Fn(Txid) -> Result<WitnessStatus, Self::Error>;
+    #[cfg(feature = "async")]
+    /// Returns a closure which can retrieve a witness status of an arbitrary transaction id
+    /// (including the ones that are not related to the wallet).
+    fn txid_resolver_async(&self) -> impl AsyncFn(Txid) -> Result<WitnessStatus, Self::Error>;
+
+    #[cfg(not(feature = "async"))]
+    /// Returns the height of the last known block.
+    fn last_block_height(&self) -> Result<u64, Self::Error>;
+    #[cfg(feature = "async")]
+    /// Returns the height of the last known block.
+    async fn last_block_height_async(&self) -> Result<u64, Self::Error>;
 
+    #[cfg(not(feature = "async"))]
     /// Broadcasts the transaction, also updating UTXO set accordingly.
     fn broadcast(&mut self, tx: &Tx, change: Option<(Vout, u32, u32)>) -> Result<(), Self::Error>;
+    #[cfg(feature = "async")]
+    /// Broadcasts the transaction, also updating UTXO set accordingly.
+    async fn broadcast_async(
+        &mut self,
+        tx: &Tx,
+        change: Option<(Vout, u32, u32)>,
+    ) -> Result<(), Self::Error>;
 }
 
 pub trait Coinselect {
@@ -847,20 +871,62 @@ where
             .consume(allow_unknown, reader, seal_resolver, sig_validator)
     }
 
-    /// Synchronize a wallet UTXO set and the status of all witnesses and single-use seal
+    #[cfg(not(feature = "async"))]
+    /// Update a wallet UTXO set and the status of all witnesses and single-use seal
     /// definitions.
     ///
     /// Applies rollbacks or forwards if required and recomputes the state of the affected
     /// contracts.
-    pub fn sync(
+    pub fn update(
         &mut self,
+        min_conformations: u32,
     ) -> Result<(), MultiError<SyncError<W::Error>, <Sp::Stock as Stock>::Error>> {
         self.wallet
-            .sync_utxos()
+            .update_utxos()
+            .map_err(SyncError::Wallet)
+            .map_err(MultiError::from_a)?;
+        let last_height = self
+            .wallet
+            .last_block_height()
             .map_err(SyncError::Wallet)
             .map_err(MultiError::from_a)?;
         self.contracts
-            .sync_witnesses(self.wallet.txid_resolver())
+            .update_witnesses(self.wallet.txid_resolver(), last_height, min_conformations)
+            .map_err(MultiError::from_other_a)
+    }
+
+    #[cfg(feature = "async")]
+    /// Update a wallet UTXO set and the status of all witnesses and single-use seal
+    /// definitions.
+    ///
+    /// Applies rollbacks or forwards if required and recomputes the state of the affected
+    /// contracts.
+    pub async fn update_async(
+        &mut self,
+        min_conformations: u32,
+    ) -> Result<(), MultiError<SyncError<W::Error>, <Sp::Stock as Stock>::Error>>
+    where
+        Sp::Stock: 'static,
+        Sp::Pile: 'static,
+    {
+        self.wallet
+            .update_utxos_async()
+            .await
+            .map_err(SyncError::Wallet)
+            .map_err(MultiError::from_a)?;
+        let last_height = self
+            .wallet
+            .last_block_height_async()
+            .await
+            .map_err(SyncError::Wallet)
+            .map_err(MultiError::from_a)?;
+        self.contracts
+            .update_witnesses_async(
+                self.wallet.txid_resolver_async(),
+                last_height,
+                min_conformations,
+            )
+            .await
             .map_err(MultiError::from_other_a)
     }
 }