fix: 🚑 nonce management in the blockchain service (#549)

* feat: ✨ initial transaction pool and nonce manager logic

* feat: ♻️ refactor watches to sync with new transaction pool

* fix: 🐛 issue with status subscriptions

* fix: 🩹 also process tx status updates on finality

* feat: 🔊 improve log clarity by specifying which tx is being dropped

* test: ✅ add Retracted and Usurped test cases for new logic

* fix: 🎨 run cargo fmt

* feat: ✨ add event extraction from successful transactions

* style: 🎨 run cargo fmt

* test: ✅ add Invalid transaction and nonce gap fulfilling test

* feat: 🔊 Add error logs in case of failing to send status update of transactions

* feat: ✅ add a way to select log level for actors in integration tests

* feat: 🔊 Add logging of which call is related to which hash

* fix: 🩹 Remove leftover `only: true` flag

* perf: ⚡ Decrease retry timout in test

* fix: 🩹 amend review (pool -> manager)

* feat: ✨ add tip tracking in transaction manager

* docs: 💡 add permalink to Polkadot SDK's `TransactionStatus`

* test: ✅ rename and fix transaction manager integration test

* fix: ✅ single volunteer test

* test: ✅ improve BSP volunteer for multiple files test

* feat: ✨ Allow arbitrary strings for log level in tests

* feat: 🔊 Improve logs of transactions adding nonce

* fix: 🐛 add initial check when watching for tx success

* fix: 🐛 check for the `ExtrinsicSuccess` event to ensure the new ext didn't fail

* test: ✅ minor test fixes

* fix: 🩹 Fix typos and names of tests

* fix: 🐛 Update transaction status in `run()` cycle of blockchain service

* fix: 🩹 Remove processing of transaction status in block import and block finalised

* fix: 🐛 skip to correct block when testing MSP charge

* fix: 🩹 Remove `only: true` flag from test

* fix: 🩹 Move `tx_status_receiver` out of `actor` struct

* fix: 🐛 Avoid cross-talk between updates from replaced transactions

* test: ✅ fix invalid transaction test by dropping retry tx

---------

Co-authored-by: Facundo Farall <[email protected]>

Author: TDemeco

Cargo.lock

@@ -36,6 +36,7 @@ sc-client-api = { workspace = true }
 sc-service = { workspace = true }
 sc-network = { workspace = true }
 sc-network-types = { workspace = true }
+sc-transaction-pool-api = { workspace = true }
 sc-utils = { workspace = true }
 sc-tracing = { workspace = true }
 sp-api = { workspace = true }

client/blockchain-service/Cargo.toml

@@ -16,7 +16,7 @@ use shc_common::types::{BlockNumber, ProviderId, StorageDataUnit};
 use shc_forest_manager::traits::ForestStorageHandler;
 
 use crate::{
-    transaction::SubmittedTransaction, types::ManagedProvider, types::SendExtrinsicOptions,
+    types::{ManagedProvider, SendExtrinsicOptions, SubmittedExtrinsicInfo},
     BlockchainService,
 };
 
@@ -38,7 +38,7 @@ pub struct CapacityRequestQueue<Runtime: StorageEnableRuntime> {
     /// This is reset when the `pending_requests` is moved to `requests_waiting_for_inclusion` when they have been batched in a single transaction.
     total_required: StorageDataUnit<Runtime>,
     /// The last submitted transaction which `requests_waiting_for_inclusion` is waiting for.
-    last_submitted_transaction: Option<SubmittedTransaction<Runtime>>,
+    last_submitted_transaction: Option<SubmittedExtrinsicInfo<Runtime>>,
 }
 
 impl<Runtime: StorageEnableRuntime> CapacityRequestQueue<Runtime> {
@@ -53,7 +53,7 @@ impl<Runtime: StorageEnableRuntime> CapacityRequestQueue<Runtime> {
     }
 
     /// Get the last submitted transaction.
-    pub fn last_submitted_transaction(&self) -> Option<&SubmittedTransaction<Runtime>> {
+    pub fn last_submitted_transaction(&self) -> Option<&SubmittedExtrinsicInfo<Runtime>> {
         self.last_submitted_transaction.as_ref()
     }
 
@@ -131,10 +131,10 @@ impl<Runtime: StorageEnableRuntime> CapacityRequestQueue<Runtime> {
         !self.requests_waiting_for_inclusion.is_empty()
     }
 
-    /// Add all pending requests to the list of requests waiting for inclusion of the [`SubmittedTransaction`].
+    /// Add all pending requests to the list of requests waiting for inclusion of the [`SubmittedExtrinsicInfo`].
     pub fn add_pending_requests_to_waiting_for_inclusion(
         &mut self,
-        submitted_transaction: SubmittedTransaction<Runtime>,
+        submitted_transaction: SubmittedExtrinsicInfo<Runtime>,
     ) {
         self.requests_waiting_for_inclusion
             .extend(self.pending_requests.drain(..));
@@ -335,20 +335,20 @@ where
 
         // Send extrinsic to increase capacity
         match self
-            .send_extrinsic(call, &SendExtrinsicOptions::new(extrinsic_retry_timeout))
+            .send_extrinsic(
+                call,
+                &SendExtrinsicOptions::new(
+                    extrinsic_retry_timeout,
+                    Some("storageProviders".to_string()),
+                    Some("changeCapacity".to_string()),
+                ),
+            )
             .await
         {
             Ok(output) => {
                 // Add all pending requests to the list of requests waiting for inclusion.
                 if let Some(capacity_manager) = self.capacity_manager.as_mut() {
-                    capacity_manager.add_pending_requests_to_waiting_for_inclusion(
-                        SubmittedTransaction::new(
-                            output.receiver,
-                            output.hash,
-                            output.nonce,
-                            extrinsic_retry_timeout,
-                        ),
-                    );
+                    capacity_manager.add_pending_requests_to_waiting_for_inclusion(output);
                 } else {
                     error!(target: LOG_TARGET, "Capacity manager not initialized");
                 }

client/blockchain-service/src/capacity_manager.rs

@@ -2,7 +2,6 @@ use anyhow::Result;
 use async_trait::async_trait;
 use log::{debug, warn};
 use sc_network::Multiaddr;
-use serde_json::Number;
 use shc_common::{
     traits::{KeyTypeOperations, StorageEnableRuntime},
     types::StorageEnableEvents,
@@ -34,11 +33,12 @@ use shc_forest_manager::traits::ForestStorageHandler;
 use crate::{
     capacity_manager::CapacityRequestData,
     handler::BlockchainService,
-    transaction::SubmittedTransaction,
+    transaction_manager::wait_for_transaction_status,
     types::{
         ConfirmStoringRequest, Extrinsic, ExtrinsicResult, FileDeletionRequest, MinimalBlockInfo,
-        RespondStorageRequest, RetryStrategy, SendExtrinsicOptions,
-        StopStoringForInsolventUserRequest, SubmitProofRequest, WatchTransactionError,
+        RespondStorageRequest, RetryStrategy, SendExtrinsicOptions, StatusToWait,
+        StopStoringForInsolventUserRequest, SubmitProofRequest, SubmittedExtrinsicInfo,
+        WatchTransactionError,
     },
 };
 
@@ -51,7 +51,7 @@ const LOG_TARGET: &str = "blockchain-service-interface";
     default_inner_channel_type = tokio::sync::oneshot::Receiver,
 )]
 pub enum BlockchainServiceCommand<Runtime: StorageEnableRuntime> {
-    #[command(success_type = SubmittedTransaction<Runtime>)]
+    #[command(success_type = SubmittedExtrinsicInfo<Runtime>)]
     SendExtrinsic {
         call: Runtime::Call,
         options: SendExtrinsicOptions,
@@ -61,36 +61,25 @@ pub enum BlockchainServiceCommand<Runtime: StorageEnableRuntime> {
         block_hash: Runtime::Hash,
         extrinsic_hash: Runtime::Hash,
     },
-    UnwatchExtrinsic {
-        subscription_id: Number,
-    },
     #[command(success_type = MinimalBlockInfo<Runtime>)]
     GetBestBlockInfo,
     #[command(mode = "AsyncResponse")]
-    WaitForBlock {
-        block_number: BlockNumber<Runtime>,
-    },
+    WaitForBlock { block_number: BlockNumber<Runtime> },
     #[command(mode = "AsyncResponse")]
     WaitForNumBlocks {
         number_of_blocks: BlockNumber<Runtime>,
     },
     #[command(mode = "AsyncResponse", error_type = ApiError)]
-    WaitForTick {
-        tick_number: TickNumber<Runtime>,
-    },
+    WaitForTick { tick_number: TickNumber<Runtime> },
     #[command(success_type = bool, error_type = IsStorageRequestOpenToVolunteersError)]
-    IsStorageRequestOpenToVolunteers {
-        file_key: Runtime::Hash,
-    },
+    IsStorageRequestOpenToVolunteers { file_key: Runtime::Hash },
     #[command(success_type = BlockNumber<Runtime>, error_type = QueryFileEarliestVolunteerTickError)]
     QueryFileEarliestVolunteerTick {
         bsp_id: ProviderId<Runtime>,
         file_key: Runtime::Hash,
     },
     #[command(success_type = BlockNumber<Runtime>, error_type = QueryEarliestChangeCapacityBlockError)]
-    QueryEarliestChangeCapacityBlock {
-        bsp_id: ProviderId<Runtime>,
-    },
+    QueryEarliestChangeCapacityBlock { bsp_id: ProviderId<Runtime> },
     #[command(success_type = <<Runtime as StorageEnableRuntime>::Signature as KeyTypeOperations>::Public)]
     GetNodePublicKey,
     #[command(success_type = Vec<ChunkId>, error_type = QueryBspConfirmChunksToProveForFileError)]
@@ -104,9 +93,7 @@ pub enum BlockchainServiceCommand<Runtime: StorageEnableRuntime> {
         file_key: Runtime::Hash,
     },
     #[command(success_type = Vec<Multiaddr>, error_type = QueryProviderMultiaddressesError)]
-    QueryProviderMultiaddresses {
-        provider_id: ProviderId<Runtime>,
-    },
+    QueryProviderMultiaddresses { provider_id: ProviderId<Runtime> },
     QueueSubmitProofRequest {
         request: SubmitProofRequest<Runtime>,
     },
@@ -145,21 +132,13 @@ pub enum BlockchainServiceCommand<Runtime: StorageEnableRuntime> {
     #[command(success_type = BlockNumber<Runtime>, error_type = ApiError)]
     QueryLastCheckpointChallengeTick,
     #[command(success_type = Vec<CustomChallenge<Runtime>>, error_type = GetCheckpointChallengesError)]
-    QueryLastCheckpointChallenges {
-        tick: BlockNumber<Runtime>,
-    },
+    QueryLastCheckpointChallenges { tick: BlockNumber<Runtime> },
     #[command(success_type = Runtime::Hash, error_type = GetBspInfoError)]
-    QueryProviderForestRoot {
-        provider_id: ProviderId<Runtime>,
-    },
+    QueryProviderForestRoot { provider_id: ProviderId<Runtime> },
     #[command(success_type = StorageDataUnit<Runtime>, error_type = QueryStorageProviderCapacityError)]
-    QueryStorageProviderCapacity {
-        provider_id: ProviderId<Runtime>,
-    },
+    QueryStorageProviderCapacity { provider_id: ProviderId<Runtime> },
     #[command(success_type = StorageDataUnit<Runtime>, error_type = QueryAvailableStorageCapacityError)]
-    QueryAvailableStorageCapacity {
-        provider_id: ProviderId<Runtime>,
-    },
+    QueryAvailableStorageCapacity { provider_id: ProviderId<Runtime> },
     #[command(success_type = Option<StorageProviderId<Runtime>>)]
     QueryStorageProviderId {
         maybe_node_pub_key:
@@ -171,15 +150,11 @@ pub enum BlockchainServiceCommand<Runtime: StorageEnableRuntime> {
         min_debt: Balance<Runtime>,
     },
     #[command(success_type = Option<Balance<Runtime>>)]
-    QueryWorstCaseScenarioSlashableAmount {
-        provider_id: ProviderId<Runtime>,
-    },
+    QueryWorstCaseScenarioSlashableAmount { provider_id: ProviderId<Runtime> },
     #[command(success_type = Balance<Runtime>)]
     QuerySlashAmountPerMaxFileSize,
     #[command(success_type = Option<MainStorageProviderId<Runtime>>, error_type = QueryMspIdOfBucketIdError)]
-    QueryMspIdOfBucketId {
-        bucket_id: BucketId<Runtime>,
-    },
+    QueryMspIdOfBucketId { bucket_id: BucketId<Runtime> },
     ReleaseForestRootWriteLock {
         forest_root_write_tx: tokio::sync::oneshot::Sender<()>,
     },
@@ -206,9 +181,7 @@ pub enum BlockchainServiceCommand<Runtime: StorageEnableRuntime> {
         bsp_id: BackupStorageProviderId<Runtime>,
     },
     #[command(success_type = Vec<BucketId<Runtime>>)]
-    QueryBucketsForMsp {
-        msp_id: ProviderId<Runtime>,
-    },
+    QueryBucketsForMsp { msp_id: ProviderId<Runtime> },
 }
 
 /// Interface for interacting with the BlockchainService actor.
@@ -280,25 +253,34 @@ where
         for retry_count in 0..=retry_strategy.max_retries {
             debug!(target: LOG_TARGET, "Submitting transaction {:?} with tip {}", call, tip);
 
-            let extrinsic_options = SendExtrinsicOptions::new(options.timeout())
-                .with_tip(tip as u128)
-                .with_nonce(nonce);
+            let extrinsic_options =
+                SendExtrinsicOptions::new(options.timeout(), options.module(), options.method())
+                    .with_tip(tip as u128)
+                    .with_nonce(nonce);
 
-            let mut transaction = self.send_extrinsic(call.clone(), extrinsic_options).await?;
+            let submitted_ext_info = self.send_extrinsic(call.clone(), extrinsic_options).await?;
 
-            let result: Result<Option<StorageHubEventsVec<Runtime>>, _> = if with_events {
-                transaction
-                    .watch_for_success_with_events(&self)
-                    .await
-                    .map(Some)
-            } else {
-                transaction.watch_for_success(&self).await.map(|_| None)
-            };
+            // Wait for transaction to be included in a block
+            let result = wait_for_transaction_status(
+                submitted_ext_info.nonce,
+                submitted_ext_info.status_subscription.clone(),
+                StatusToWait::InBlock,
+                options.timeout(),
+            )
+            .await;
 
             match result {
-                Ok(maybe_events) => {
-                    debug!(target: LOG_TARGET, "Transaction with hash {:?} succeeded", transaction.hash());
-                    return Ok(maybe_events);
+                Ok(block_hash) => {
+                    debug!(target: LOG_TARGET, "Transaction with hash {:?} succeeded", submitted_ext_info.hash);
+
+                    if with_events {
+                        let extrinsic = self
+                            .get_extrinsic_from_block(block_hash, submitted_ext_info.hash)
+                            .await?;
+                        return Ok(Some(extrinsic.events));
+                    } else {
+                        return Ok(None);
+                    }
                 }
                 Err(err) => {
                     warn!(target: LOG_TARGET, "Transaction failed: {:?}", err);
@@ -309,18 +291,17 @@ where
                         }
                     }
 
-                    warn!(target: LOG_TARGET, "Failed to submit transaction with hash {:?}, attempt #{}", transaction.hash(), retry_count + 1);
+                    warn!(target: LOG_TARGET, "Failed to submit transaction with hash {:?}, attempt #{}", submitted_ext_info.hash, retry_count + 1);
 
-                    // TODO: Add pending transaction pool implementation to be able to resubmit transactions with nonces lower than the current one to avoid this transaction from being stuck.
                     if let WatchTransactionError::Timeout = err {
                         // Increase the tip to incentivise the collators to include the transaction in a block with priority
                         tip = retry_strategy.compute_tip(retry_count + 1);
                         // Reuse the same nonce since the transaction was not included in a block.
-                        nonce = Some(transaction.nonce());
+                        nonce = Some(submitted_ext_info.nonce);
 
                         // Log warning if this is not the last retry.
                         if retry_count < retry_strategy.max_retries {
-                            warn!(target: LOG_TARGET, "Retrying with increased tip {} and nonce {}", tip, transaction.nonce());
+                            warn!(target: LOG_TARGET, "Retrying with increased tip {} and nonce {}", tip, submitted_ext_info.nonce);
                         }
                     }
                 }