Use transaction queue for DKG/Substrate (fixes #450) (#461)

* Use transaction queue for DKG/Substrate (fixes #450)

* dont need chain id param

* address review comments

crates/proposal-signing-backends/src/dkg.rs

@@ -1,35 +1,45 @@
+use std::borrow::Cow;
 use std::sync::Arc;
-use futures::StreamExt;
 use tokio::sync::Mutex;
 use webb::substrate::dkg_runtime::api::runtime_types::webb_proposals::header::{TypedChainId, ResourceId};
-use webb::substrate::dkg_runtime::api::runtime_types::sp_core::bounded::bounded_vec::BoundedVec;
 use webb::substrate::dkg_runtime::api::runtime_types::webb_proposals::nonce::Nonce;
 use webb::substrate::subxt::{OnlineClient, PolkadotConfig};
 use sp_core::sr25519::Pair as Sr25519Pair;
+use webb::evm::ethers::utils;
 use webb_proposals::ProposalTrait;
 use webb::substrate::scale::{Encode, Decode};
 use webb_relayer_utils::metric;
-use webb::substrate::subxt::tx::{PairSigner, TxStatus as TransactionStatus};
 use webb::substrate::dkg_runtime::api as RuntimeApi;
+use webb::substrate::subxt::dynamic::Value;
+use webb_relayer_store::{QueueStore, SledStore};
+use webb_relayer_store::sled::SledQueueKey;
+use webb_relayer_types::dynamic_payload::WebbDynamicTxPayload;
+use webb::substrate::subxt::tx::{PairSigner};
+use webb::substrate::dkg_runtime::api::runtime_types::sp_core::bounded::bounded_vec::BoundedVec;
+
 type DkgConfig = PolkadotConfig;
 type DkgClient = OnlineClient<DkgConfig>;
 /// A ProposalSigningBackend that uses the DKG System for Signing Proposals.
 pub struct DkgProposalSigningBackend {
     pub client: DkgClient,
     pub pair: PairSigner<PolkadotConfig, Sr25519Pair>,
     pub typed_chain_id: webb_proposals::TypedChainId,
+    /// Something that implements the QueueStore trait.
+    store: Arc<SledStore>,
 }
 
 impl DkgProposalSigningBackend {
     pub fn new(
         client: OnlineClient<PolkadotConfig>,
         pair: PairSigner<PolkadotConfig, Sr25519Pair>,
         typed_chain_id: webb_proposals::TypedChainId,
+        store: Arc<SledStore>,
     ) -> Self {
         Self {
             client,
             pair,
             typed_chain_id,
+            store,
         }
     }
 }
@@ -85,7 +95,7 @@ impl super::ProposalSigningBackend for DkgProposalSigningBackend {
     async fn handle_proposal(
         &self,
         proposal: &(impl ProposalTrait + Sync + Send + 'static),
-        metrics: Arc<Mutex<metric::Metrics>>,
+        _metrics: Arc<Mutex<metric::Metrics>>,
     ) -> webb_relayer_utils::Result<()> {
         let tx_api = RuntimeApi::tx().dkg_proposals();
         let resource_id = proposal.header().resource_id();
@@ -96,73 +106,42 @@ impl super::ProposalSigningBackend for DkgProposalSigningBackend {
         tracing::debug!(
             nonce = %hex::encode(nonce.encode()),
             resource_id = %hex::encode(resource_id.into_bytes()),
-            src_chain_id = ?src_chain_id,
+            src_chain_id = ?self.typed_chain_id,
             proposal = %hex::encode(proposal.to_vec()),
             "sending proposal to DKG runtime"
         );
+
         let xt = tx_api.acknowledge_proposal(
-            nonce,
+            nonce.clone(),
             src_chain_id,
             ResourceId(resource_id.into_bytes()),
             BoundedVec(proposal.to_vec()),
         );
+        // webb dynamic payload
+        let execute_proposal_tx = WebbDynamicTxPayload {
+            pallet_name: Cow::Borrowed("DKGProposals"),
+            call_name: Cow::Borrowed("acknowledge_proposal"),
+            fields: vec![
+                Value::u128(u128::from(nonce.0)),
+                Value::u128(u128::from(self.typed_chain_id.chain_id())),
+                Value::from_bytes(
+                    ResourceId(resource_id.into_bytes()).encode(),
+                ),
+                Value::from_bytes(BoundedVec(proposal.to_vec()).encode()),
+            ],
+        };
+        let data_hash = utils::keccak256(xt.call_data().encode());
+        let tx_key = SledQueueKey::from_substrate_with_custom_key(
+            self.typed_chain_id.underlying_chain_id(),
+            make_acknowledge_proposal_key(data_hash),
+        );
+        // Enqueue WebbDynamicTxPayload in protocol-substrate transaction queue
+        QueueStore::<WebbDynamicTxPayload>::enqueue_item(
+            &self.store,
+            tx_key,
+            execute_proposal_tx,
+        )?;
 
-        // TODO: here we should have a substrate based tx queue in the background
-        // where just send the raw xt bytes and let it handle the work for us.
-        // but this here for now.
-        let signer = &self.pair;
-        let mut progress = self
-            .client
-            .tx()
-            .sign_and_submit_then_watch_default(&xt, signer)
-            .await?;
-
-        while let Some(event) = progress.next().await {
-            let e = match event {
-                Ok(e) => e,
-                Err(err) => {
-                    tracing::error!(error = %err, "failed to watch for tx events");
-                    return Err(err.into());
-                }
-            };
-
-            match e {
-                TransactionStatus::Future => {}
-                TransactionStatus::Ready => {
-                    tracing::trace!("tx ready");
-                }
-                TransactionStatus::Broadcast(_) => {}
-                TransactionStatus::InBlock(_) => {
-                    tracing::trace!("tx in block");
-                }
-                TransactionStatus::Retracted(_) => {
-                    tracing::warn!("tx retracted");
-                }
-                TransactionStatus::FinalityTimeout(_) => {
-                    tracing::warn!("tx timeout");
-                }
-                TransactionStatus::Finalized(v) => {
-                    let maybe_success = v.wait_for_success().await;
-                    match maybe_success {
-                        Ok(_events) => {
-                            metrics.lock().await.proposals_signed.inc();
-                            tracing::debug!("tx finalized");
-                        }
-                        Err(err) => {
-                            tracing::error!(error = %err, "tx failed");
-                            return Err(err.into());
-                        }
-                    }
-                }
-                TransactionStatus::Usurped(_) => {}
-                TransactionStatus::Dropped => {
-                    tracing::warn!("tx dropped");
-                }
-                TransactionStatus::Invalid => {
-                    tracing::warn!("tx invalid");
-                }
-            }
-        }
         Ok(())
     }
 }
@@ -190,3 +169,12 @@ fn webb_proposals_typed_chain_converter(
         webb_proposals::TypedChainId::Ink(id) => TypedChainId::Ink(id),
     }
 }
+
+pub fn make_acknowledge_proposal_key(data_hash: [u8; 32]) -> [u8; 64] {
+    let mut result = [0u8; 64];
+    let prefix = b"acknowledge_proposal_fixed_key__";
+    debug_assert!(prefix.len() == 32);
+    result[0..32].copy_from_slice(prefix);
+    result[32..64].copy_from_slice(&data_hash);
+    result
+}

event-watchers/substrate/src/signature_bridge_watcher.rs

@@ -393,6 +393,7 @@ fn secp256k1_ecdsa_recover(
 fn make_execute_proposal_key(data_hash: [u8; 32]) -> [u8; 64] {
     let mut result = [0u8; 64];
     let prefix = b"execute_proposal_with_signature_";
+    debug_assert!(prefix.len() == 32);
     result[0..32].copy_from_slice(prefix);
     result[32..64].copy_from_slice(&data_hash);
     result

services/webb-relayer/src/service/mod.rs

@@ -137,6 +137,7 @@ pub async fn make_proposal_signing_backend(
                 dkg_client,
                 subxt::tx::PairSigner::new(pair),
                 typed_chain_id,
+                store.clone(),
             );
             Ok(ProposalSigningBackendSelector::Dkg(backend))
         }