Make preconfirmation optional on API endpoints

.changes/changed/2925.md

@@ -0,0 +1 @@
+Make preconfirmation optional on API endpoints.

bin/e2e-test-client/src/test_context.rs

@@ -186,6 +186,7 @@ impl Wallet {
         let tx_id = tx.id(&self.consensus_params.chain_id());
         println!("submitting tx... {:?}", tx_id);
         let status = self.client.submit_and_await_commit(&tx).await?;
+        println!("Status for {:?} is {:?}", tx_id, status);
 
         // we know the transferred coin should be output 0 from above
         let transferred_utxo = UtxoId::new(tx_id, 0);

crates/client/assets/schema.sdl

@@ -1338,7 +1338,11 @@ type Subscription {
 		"""
 		The ID of the transaction
 		"""
-		id: TransactionId!
+		id: TransactionId!,
+		"""
+		If true, accept to receive the preconfirmation status
+		"""
+		allowPreconfirmation: Boolean
 	): TransactionStatus!
 	"""
 	Submits transaction to the `TxPool` and await either success or failure.
@@ -1347,9 +1351,9 @@ type Subscription {
 	"""
 	Submits the transaction to the `TxPool` and returns a stream of events.
 	Compared to the `submitAndAwait`, the stream also contains
-	`SubmittedStatus` as an intermediate state.
+	`SubmittedStatus` and potentially preconfirmation as an intermediate state.
 	"""
-	submitAndAwaitStatus(tx: HexString!, estimatePredicates: Boolean): TransactionStatus!
+	submitAndAwaitStatus(tx: HexString!, estimatePredicates: Boolean, allowPreconfirmation: Boolean): TransactionStatus!
 	contractStorageSlots(contractId: ContractId!): StorageSlot!
 	contractStorageBalances(contractId: ContractId!): ContractBalance!
 }
@@ -1391,7 +1395,7 @@ type Transaction {
 	outputContract: ContractOutput
 	witnesses: [HexString!]
 	receiptsRoot: Bytes32
-	status: TransactionStatus
+	status(allowPreconfirmation: Boolean): TransactionStatus
 	script: HexString
 	scriptData: HexString
 	bytecodeWitnessIndex: U16

crates/client/src/client.rs

@@ -939,7 +939,7 @@ impl FuelClient {
         &'a self,
         tx: &'a Transaction,
     ) -> io::Result<impl Stream<Item = io::Result<TransactionStatus>> + 'a> {
-        self.submit_and_await_status_opt(tx, None).await
+        self.submit_and_await_status_opt(tx, None, None).await
     }
 
     /// Similar to [`Self::submit_and_await_commit_opt`], but includes all intermediate states.
@@ -948,13 +948,16 @@ impl FuelClient {
         &'a self,
         tx: &'a Transaction,
         estimate_predicates: Option<bool>,
+        allow_preconfirmation: Option<bool>,
     ) -> io::Result<impl Stream<Item = io::Result<TransactionStatus>> + 'a> {
         use cynic::SubscriptionBuilder;
+        use schema::tx::SubmitAndAwaitStatusArg;
         let tx = tx.clone().to_bytes();
         let s = schema::tx::SubmitAndAwaitStatusSubscription::build(
-            TxWithEstimatedPredicatesArg {
+            SubmitAndAwaitStatusArg {
                 tx: HexString(Bytes(tx)),
                 estimate_predicates,
+                allow_preconfirmation,
             },
         );
 
@@ -1206,14 +1209,29 @@ impl FuelClient {
 
     #[tracing::instrument(skip(self), level = "debug")]
     #[cfg(feature = "subscriptions")]
-    /// Subscribe to the status of a transaction
+    /// Similar to [`Self::subscribe_transaction_status_opt`], but with default options.
     pub async fn subscribe_transaction_status<'a>(
         &'a self,
         id: &'a TxId,
     ) -> io::Result<impl futures::Stream<Item = io::Result<TransactionStatus>> + 'a> {
+        self.subscribe_transaction_status_opt(id, None).await
+    }
+
+    #[cfg(feature = "subscriptions")]
+    /// Subscribe to the status of a transaction
+    pub async fn subscribe_transaction_status_opt<'a>(
+        &'a self,
+        id: &'a TxId,
+        allow_preconfirmation: Option<bool>,
+    ) -> io::Result<impl Stream<Item = io::Result<TransactionStatus>> + 'a> {
         use cynic::SubscriptionBuilder;
+        use schema::tx::StatusChangeSubscriptionArgs;
         let tx_id: TransactionId = (*id).into();
-        let s = schema::tx::StatusChangeSubscription::build(TxIdArgs { id: tx_id });
+        let s =
+            schema::tx::StatusChangeSubscription::build(StatusChangeSubscriptionArgs {
+                id: tx_id,
+                allow_preconfirmation,
+            });
 
         tracing::debug!("subscribing");
         let stream = self.subscribe(s).await?.map(|tx| {

crates/client/src/client/schema/tx.rs

@@ -472,14 +472,21 @@ pub struct TransactionsByOwnerQuery {
     pub transactions_by_owner: TransactionConnection,
 }
 
+#[derive(cynic::QueryVariables, Debug)]
+pub struct StatusChangeSubscriptionArgs {
+    pub id: TransactionId,
+    #[cynic(skip_serializing_if = "Option::is_none")]
+    pub allow_preconfirmation: Option<bool>,
+}
+
 #[derive(cynic::QueryFragment, Clone, Debug)]
 #[cynic(
     schema_path = "./assets/schema.sdl",
     graphql_type = "Subscription",
-    variables = "TxIdArgs"
+    variables = "StatusChangeSubscriptionArgs"
 )]
 pub struct StatusChangeSubscription {
-    #[arguments(id: $id)]
+    #[arguments(id: $id, allowPreconfirmation: $allow_preconfirmation)]
     pub status_change: TransactionStatus,
 }
 
@@ -497,6 +504,15 @@ pub struct TxWithEstimatedPredicatesArg {
     pub estimate_predicates: Option<bool>,
 }
 
+#[derive(cynic::QueryVariables)]
+pub struct SubmitAndAwaitStatusArg {
+    pub tx: HexString,
+    #[cynic(skip_serializing_if = "Option::is_none")]
+    pub estimate_predicates: Option<bool>,
+    #[cynic(skip_serializing_if = "Option::is_none")]
+    pub allow_preconfirmation: Option<bool>,
+}
+
 #[derive(cynic::QueryFragment, Clone, Debug)]
 #[cynic(
     schema_path = "./assets/schema.sdl",
@@ -643,10 +659,10 @@ pub struct SubmitAndAwaitSubscriptionWithTransaction {
 #[cynic(
     schema_path = "./assets/schema.sdl",
     graphql_type = "Subscription",
-    variables = "TxWithEstimatedPredicatesArg"
+    variables = "SubmitAndAwaitStatusArg"
 )]
 pub struct SubmitAndAwaitStatusSubscription {
-    #[arguments(tx: $tx, estimatePredicates: $estimate_predicates)]
+    #[arguments(tx: $tx, estimatePredicates: $estimate_predicates, allowPreconfirmation: $allow_preconfirmation)]
     pub submit_and_await_status: TransactionStatus,
 }
 

crates/fuel-core/src/combined_database.rs

@@ -523,6 +523,14 @@ impl CombinedDatabase {
 
         Ok(())
     }
+
+    pub fn shutdown(self) {
+        self.on_chain.shutdown();
+        self.off_chain.shutdown();
+        self.relayer.shutdown();
+        self.gas_price.shutdown();
+        self.compression.shutdown();
+    }
 }
 
 /// A trait for listening to shutdown signals.

crates/fuel-core/src/database.rs

@@ -181,6 +181,12 @@ where
         self.iter_all_filtered::<T, _>(prefix, None, Some(direction))
             .map_ok(|(key, value)| TableEntry { key, value })
     }
+
+    pub fn shutdown(self) {
+        let (storage, _) = self.into_inner();
+
+        storage.data.shutdown()
+    }
 }
 
 impl<Description> GenesisDatabase<Description>

crates/fuel-core/src/query/subscriptions.rs

@@ -20,6 +20,7 @@ pub(crate) trait TxnStatusChangeState {
     async fn get_tx_status(
         &self,
         id: Bytes32,
+        allow_preconfirmation: bool,
     ) -> StorageResult<Option<TransactionStatus>>;
 }
 
@@ -28,14 +29,15 @@ pub(crate) async fn transaction_status_change<'a, State>(
     state: State,
     stream: BoxStream<'a, TxStatusMessage>,
     transaction_id: Bytes32,
+    allow_preconfirmation: bool,
 ) -> impl Stream<Item = anyhow::Result<ApiTxStatus>> + 'a
 where
     State: TxnStatusChangeState + Send + Sync + 'a,
 {
     // Check the database first to see if the transaction already
     // has a status.
     let maybe_db_status = state
-        .get_tx_status(transaction_id)
+        .get_tx_status(transaction_id, allow_preconfirmation)
         .await
         .transpose()
         .map(TxStatusMessage::from);
@@ -50,7 +52,7 @@ where
         .chain(stream)
         // Keep taking the stream until the oneshot channel is closed.
         .take_until(closed)
-        .map(move |status| {
+        .filter_map(move |status | {
             if status.is_final() {
                 if let Some(close) = close.take() {
                     let _ = close.send(());
@@ -60,11 +62,15 @@ where
             match status {
                 TxStatusMessage::Status(status) => {
                     let status = ApiTxStatus::new(transaction_id, status);
-                    Ok(status)
+                    if !allow_preconfirmation && status.is_preconfirmation() {
+                        futures::future::ready(None)
+                    } else {
+                        futures::future::ready(Some(Ok(status)))
+                    }
                 },
                 // Map a failed status to an error for the api.
                 TxStatusMessage::FailedStatus => {
-                    Err(anyhow::anyhow!("Failed to get transaction status"))
+                    futures::future::ready(Some(Err(anyhow::anyhow!("Failed to get transaction status"))))
                 }
             }
         })

crates/fuel-core/src/query/subscriptions/test.rs

@@ -265,16 +265,16 @@ fn test_tsc_inner(
     let out = RT.with(|rt| {
         rt.block_on(async {
             let mut mock_state = super::MockTxnStatusChangeState::new();
-            mock_state
-                .expect_get_tx_status()
-                .returning(move |_| match state.clone() {
+            mock_state.expect_get_tx_status().returning(move |_, _| {
+                match state.clone() {
                     Ok(Some(t)) => Ok(Some(t)),
                     Ok(None) => Ok(None),
                     Err(_) => Err(StorageError::NotFound("", "")),
-                });
+                }
+            });
 
             let stream = futures::stream::iter(stream).boxed();
-            super::transaction_status_change(mock_state, stream, txn_id(0))
+            super::transaction_status_change(mock_state, stream, txn_id(0), true)
                 .await
                 .collect::<Vec<_>>()
                 .await

crates/fuel-core/src/schema/tx.rs

@@ -42,6 +42,7 @@ use crate::{
                 AssembleTx,
             },
             types::{
+                get_tx_status,
                 AssembleTransactionResult,
                 TransactionStatus,
             },
@@ -624,6 +625,8 @@ impl TxStatusSubscription {
         &self,
         ctx: &'a Context<'a>,
         #[graphql(desc = "The ID of the transaction")] id: TransactionId,
+        #[graphql(desc = "If true, accept to receive the preconfirmation status")]
+        allow_preconfirmation: Option<bool>,
     ) -> anyhow::Result<impl Stream<Item = async_graphql::Result<TransactionStatus>> + 'a>
     {
         let tx_status_manager = ctx.data_unchecked::<DynTxStatusManager>();
@@ -634,11 +637,14 @@ impl TxStatusSubscription {
             tx_status_manager,
             query,
         };
-        Ok(
-            transaction_status_change(status_change_state, rx, id.into())
-                .await
-                .map_err(async_graphql::Error::from),
+        Ok(transaction_status_change(
+            status_change_state,
+            rx,
+            id.into(),
+            allow_preconfirmation.unwrap_or(false),
         )
+        .await
+        .map_err(async_graphql::Error::from))
     }
 
     /// Submits transaction to the `TxPool` and await either success or failure.
@@ -653,7 +659,7 @@ impl TxStatusSubscription {
     > {
         use tokio_stream::StreamExt;
         let subscription =
-            submit_and_await_status(ctx, tx, estimate_predicates.unwrap_or(false))
+            submit_and_await_status(ctx, tx, estimate_predicates.unwrap_or(false), false)
                 .await?;
 
         Ok(subscription
@@ -663,24 +669,32 @@ impl TxStatusSubscription {
 
     /// Submits the transaction to the `TxPool` and returns a stream of events.
     /// Compared to the `submitAndAwait`, the stream also contains
-    /// `SubmittedStatus` as an intermediate state.
+    /// `SubmittedStatus` and potentially preconfirmation as an intermediate state.
     #[graphql(complexity = "query_costs().submit_and_await + child_complexity")]
     async fn submit_and_await_status<'a>(
         &self,
         ctx: &'a Context<'a>,
         tx: HexString,
         estimate_predicates: Option<bool>,
+        allow_preconfirmation: Option<bool>,
     ) -> async_graphql::Result<
         impl Stream<Item = async_graphql::Result<TransactionStatus>> + 'a,
     > {
-        submit_and_await_status(ctx, tx, estimate_predicates.unwrap_or(false)).await
+        submit_and_await_status(
+            ctx,
+            tx,
+            estimate_predicates.unwrap_or(false),
+            allow_preconfirmation.unwrap_or(false),
+        )
+        .await
     }
 }
 
 async fn submit_and_await_status<'a>(
     ctx: &'a Context<'a>,
     tx: HexString,
     estimate_predicates: bool,
+    allow_preconfirmation: bool,
 ) -> async_graphql::Result<
     impl Stream<Item = async_graphql::Result<TransactionStatus>> + 'a,
 > {
@@ -703,13 +717,21 @@ async fn submit_and_await_status<'a>(
     txpool.insert(tx).await?;
 
     Ok(subscription
-        .map(move |event| match event {
-            TxStatusMessage::Status(status) => {
-                let status = TransactionStatus::new(tx_id, status);
-                Ok(status)
-            }
-            TxStatusMessage::FailedStatus => {
-                Err(anyhow::anyhow!("Failed to get transaction status").into())
+        .filter_map(move |status| {
+            match status {
+                TxStatusMessage::Status(status) => {
+                    let status = TransactionStatus::new(tx_id, status);
+                    if !allow_preconfirmation && status.is_preconfirmation() {
+                        None
+                    } else {
+                        Some(Ok(status))
+                    }
+                }
+                // Map a failed status to an error for the api.
+                TxStatusMessage::FailedStatus => Some(Err(anyhow::anyhow!(
+                    "Failed to get transaction status"
+                )
+                .into())),
             }
         })
         .take(3))
@@ -724,14 +746,15 @@ impl<'a> TxnStatusChangeState for StatusChangeState<'a> {
     async fn get_tx_status(
         &self,
         id: Bytes32,
+        allow_preconfirmation: bool,
     ) -> StorageResult<Option<transaction_status::TransactionStatus>> {
-        match self.query.tx_status(&id) {
-            Ok(status) => Ok(Some(status.into())),
-            Err(StorageError::NotFound(_, _)) => {
-                Ok(self.tx_status_manager.status(id).await?)
-            }
-            Err(err) => Err(err),
-        }
+        get_tx_status(
+            &id,
+            self.query.as_ref(),
+            self.tx_status_manager,
+            allow_preconfirmation,
+        )
+        .await
     }
 }