feat: add support for preconfirmations (#1644)

Closes #1642 

# Release notes

Add support of pre confirmations.

In this release, we:

- Added two need functions in the `Provider` to access preconfirmations
`send_transaction_and_await_status` and `subscribe_transaction_status`

# Checklist

- [x] All **changes** are **covered** by **tests** (or not applicable)
- [x] All **changes** are **documented** (or not applicable)
- [x] I **reviewed** the **entire PR** myself (preferably, on GH UI)
- [x] I **described** all **Breaking Changes** (or there's none)

---------

Co-authored-by: hal3e <[email protected]>

Author: AurelienFT

e2e/Cargo.toml

@@ -37,6 +37,7 @@ fuels = { workspace = true, features = [
   "accounts-signer-aws-kms",
   "test-helpers",
 ] }
+futures = { workspace = true }
 testcontainers = { workspace = true }
 tokio = { workspace = true, features = ["test-util"] }
 

e2e/tests/providers.rs

@@ -19,6 +19,7 @@ use fuels::{
         tx_status::{Success, TxStatus},
     },
 };
+use futures::StreamExt;
 use rand::thread_rng;
 
 #[tokio::test]
@@ -1011,6 +1012,94 @@ async fn test_build_with_provider() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn send_transaction_and_await_status() -> Result<()> {
+    let wallet = launch_provider_and_get_wallet().await?;
+    let provider = wallet.provider();
+
+    let consensus_parameters = provider.consensus_parameters().await?;
+    let inputs = wallet
+        .get_asset_inputs_for_amount(*consensus_parameters.base_asset_id(), 100, None)
+        .await?;
+    let outputs = wallet.get_asset_outputs_for_amount(
+        &Bech32Address::default(),
+        *consensus_parameters.base_asset_id(),
+        100,
+    );
+
+    // Given
+    let mut tb = ScriptTransactionBuilder::prepare_transfer(inputs, outputs, TxPolicies::default());
+    wallet.add_witnesses(&mut tb)?;
+
+    let tx = tb.build(provider).await?;
+
+    // When
+    let status = provider.send_transaction_and_await_status(tx, true).await?;
+
+    // Then
+    assert_eq!(status.len(), 3);
+    assert!(status.iter().enumerate().all(|(i, tx_status)| {
+        matches!(
+            (i, tx_status.clone().unwrap()),
+            (0, TxStatus::Submitted { .. })
+                | (1, TxStatus::PreconfirmationSuccess { .. })
+                | (2, TxStatus::Success { .. })
+        )
+    }));
+    Ok(())
+}
+
+#[tokio::test]
+async fn send_transaction_and_subscribe_status() -> Result<()> {
+    let config = NodeConfig {
+        block_production: Trigger::Never,
+        ..NodeConfig::default()
+    };
+    let wallet =
+        launch_custom_provider_and_get_wallets(WalletsConfig::default(), Some(config), None)
+            .await?[0]
+            .clone();
+    let provider = wallet.provider();
+
+    let consensus_parameters = provider.consensus_parameters().await?;
+    let inputs = wallet
+        .get_asset_inputs_for_amount(*consensus_parameters.base_asset_id(), 100, None)
+        .await?;
+    let outputs = wallet.get_asset_outputs_for_amount(
+        &Bech32Address::default(),
+        *consensus_parameters.base_asset_id(),
+        100,
+    );
+
+    // Given
+    let mut tb = ScriptTransactionBuilder::prepare_transfer(inputs, outputs, TxPolicies::default());
+    wallet.add_witnesses(&mut tb)?;
+
+    let tx = tb.build(provider).await?;
+    let tx_id = tx.id(consensus_parameters.chain_id());
+
+    // When
+    let mut statuses = provider.subscribe_transaction_status(&tx_id, true).await?;
+    let _ = provider.send_transaction(tx).await?;
+
+    // Then
+    assert!(matches!(
+        statuses.next().await.unwrap()?,
+        TxStatus::Submitted { .. }
+    ));
+    provider.produce_blocks(1, None).await?;
+    assert!(matches!(
+        statuses.next().await.unwrap()?,
+        TxStatus::PreconfirmationSuccess { .. }
+    ));
+    assert!(matches!(
+        statuses.next().await.unwrap()?,
+        TxStatus::Success { .. }
+    ));
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn can_produce_blocks_with_trig_never() -> Result<()> {
     let config = NodeConfig {

packages/fuels-accounts/Cargo.toml

@@ -27,6 +27,7 @@ fuel-crypto = { workspace = true, features = ["random"] }
 fuel-tx = { workspace = true }
 fuel-types = { workspace = true, features = ["random"] }
 fuels-core = { workspace = true, default-features = false }
+futures = { workspace = true}
 google-cloud-kms = { workspace = true, features = ["auth"], optional = true }
 itertools = { workspace = true }
 k256 = { workspace = true, features = ["ecdsa-core", "pem"] }

packages/fuels-accounts/src/provider.rs

@@ -45,6 +45,7 @@ use fuels_core::{
         tx_status::TxStatus,
     },
 };
+use futures::StreamExt;
 pub use retry_util::{Backoff, RetryConfig};
 pub use supported_fuel_core_version::SUPPORTED_FUEL_CORE_VERSION;
 use tai64::Tai64;
@@ -207,6 +208,56 @@ impl Provider {
         Ok(tx_status)
     }
 
+    /// Similar to `send_transaction_and_await_commit`,
+    /// but collect all the status received until a final one and return them.
+    pub async fn send_transaction_and_await_status<T: Transaction>(
+        &self,
+        tx: T,
+        include_preconfirmation: bool,
+    ) -> Result<Vec<Result<TxStatus>>> {
+        #[cfg(feature = "coin-cache")]
+        let base_asset_id = *self.consensus_parameters().await?.base_asset_id();
+        #[cfg(feature = "coin-cache")]
+        let used_base_coins = tx.used_coins(&base_asset_id);
+
+        #[cfg(feature = "coin-cache")]
+        self.check_inputs_already_in_cache(&used_base_coins).await?;
+
+        let tx = self.prepare_transaction_for_sending(tx).await?.into();
+        let mut stream = self
+            .uncached_client()
+            .submit_and_await_status(&tx, include_preconfirmation)
+            .await?;
+
+        let mut statuses = Vec::new();
+
+        // Process stream items until we get a final status
+        while let Some(status) = stream.next().await {
+            let tx_status = status.map(TxStatus::from).map_err(Into::into);
+
+            let is_final = tx_status.as_ref().ok().is_some_and(|s| s.is_final());
+
+            statuses.push(tx_status);
+
+            if is_final {
+                break;
+            }
+        }
+
+        // Handle cache updates for failures
+        #[cfg(feature = "coin-cache")]
+        if statuses.iter().any(|status| {
+            matches!(
+                &status,
+                Ok(TxStatus::SqueezedOut { .. }) | Ok(TxStatus::Failure { .. }),
+            )
+        }) {
+            self.coins_cache.lock().await.remove_items(used_base_coins);
+        }
+
+        Ok(statuses)
+    }
+
     async fn prepare_transaction_for_sending<T: Transaction>(&self, mut tx: T) -> Result<T> {
         let consensus_parameters = self.consensus_parameters().await?;
         tx.precompute(&consensus_parameters.chain_id())?;
@@ -314,6 +365,19 @@ impl Provider {
             .into())
     }
 
+    pub async fn subscribe_transaction_status<'a>(
+        &'a self,
+        tx_id: &'a TxId,
+        include_preconfirmation: bool,
+    ) -> Result<impl futures::Stream<Item = Result<TxStatus>> + 'a> {
+        let stream = self
+            .uncached_client()
+            .subscribe_transaction_status(tx_id, include_preconfirmation)
+            .await?;
+
+        Ok(stream.map(|status| status.map(Into::into).map_err(Into::into)))
+    }
+
     pub async fn chain_info(&self) -> Result<ChainInfo> {
         Ok(self.uncached_client().chain_info().await?.into())
     }

packages/fuels-accounts/src/provider/retryable_client.rs

@@ -18,6 +18,7 @@ use fuel_core_types::services::executor::TransactionExecutionStatus;
 use fuel_tx::{BlobId, ConsensusParameters, Transaction, TxId, UtxoId};
 use fuel_types::{Address, AssetId, BlockHeight, ContractId, Nonce};
 use fuels_core::types::errors::{Error, Result, error};
+use futures::Stream;
 
 use super::{
     cache::CacheableRpcs,
@@ -146,6 +147,30 @@ impl RetryableClient {
         self.wrap(|| self.client.submit_and_await_commit(tx)).await
     }
 
+    pub async fn submit_and_await_status<'a>(
+        &'a self,
+        tx: &'a Transaction,
+        include_preconfirmation: bool,
+    ) -> RequestResult<impl Stream<Item = io::Result<TransactionStatus>> + 'a> {
+        self.wrap(|| {
+            self.client
+                .submit_and_await_status_opt(tx, None, Some(include_preconfirmation))
+        })
+        .await
+    }
+
+    pub async fn subscribe_transaction_status<'a>(
+        &'a self,
+        id: &'a TxId,
+        include_preconfirmation: bool,
+    ) -> RequestResult<impl Stream<Item = io::Result<TransactionStatus>> + 'a> {
+        self.wrap(|| {
+            self.client
+                .subscribe_transaction_status_opt(id, Some(include_preconfirmation))
+        })
+        .await
+    }
+
     pub async fn submit(&self, tx: &Transaction) -> RequestResult<TransactionId> {
         self.wrap(|| self.client.submit(tx)).await
     }

packages/fuels-core/src/types/tx_status.rs

@@ -170,6 +170,13 @@ impl TxStatus {
             _ => vec![],
         }
     }
+
+    pub fn is_final(&self) -> bool {
+        matches!(
+            self,
+            TxStatus::Success(_) | TxStatus::Failure(_) | TxStatus::SqueezedOut(_)
+        )
+    }
 }
 
 #[cfg(feature = "std")]

packages/fuels-test-helpers/src/fuel_bin_service.rs

@@ -72,6 +72,9 @@ impl ExtendedConfig {
                     block_time.as_millis()
                 ));
             }
+            Trigger::Open { period } => {
+                args.push(format!("--poa-open-period={}ms", period.as_millis()));
+            }
         };
 
         let body_limit = self.node_config.graphql_request_body_bytes_limit;

packages/fuels-test-helpers/src/node_types.rs

@@ -11,6 +11,7 @@ pub enum Trigger {
     Instant,
     Never,
     Interval { block_time: Duration },
+    Open { period: Duration },
 }
 
 #[cfg(feature = "fuel-core-lib")]
@@ -20,6 +21,7 @@ impl From<Trigger> for fuel_core_poa::Trigger {
             Trigger::Instant => fuel_core_poa::Trigger::Instant,
             Trigger::Never => fuel_core_poa::Trigger::Never,
             Trigger::Interval { block_time } => fuel_core_poa::Trigger::Interval { block_time },
+            Trigger::Open { period } => fuel_core_poa::Trigger::Open { period },
         }
     }
 }