Fix/fallback gas for substrate (#177)

* Update compiler semaphore

* Fix the fallback gas for substrate chains

* Fix the concurrency & the substrate gas limit fallback value

crates/node/src/node_implementations/lighthouse_geth.rs

@@ -17,7 +17,7 @@ use std::{
     pin::Pin,
     process::{Command, Stdio},
     sync::{
-        Arc, LazyLock,
+        Arc,
         atomic::{AtomicU32, Ordering},
     },
     time::{Duration, SystemTime, UNIX_EPOCH},
@@ -47,7 +47,7 @@ use futures::{Stream, StreamExt};
 use revive_common::EVMVersion;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use serde_with::serde_as;
-use tokio::sync::{OnceCell, Semaphore};
+use tokio::sync::OnceCell;
 use tracing::{Instrument, info, instrument};
 
 use revive_dt_common::{
@@ -105,7 +105,6 @@ pub struct LighthouseGethNode {
 
     persistent_http_provider: OnceCell<ConcreteProvider<Ethereum, Arc<EthereumWallet>>>,
     persistent_ws_provider: OnceCell<ConcreteProvider<Ethereum, Arc<EthereumWallet>>>,
-    http_provider_requests_semaphore: LazyLock<Semaphore>,
 }
 
 impl LighthouseGethNode {
@@ -176,7 +175,6 @@ impl LighthouseGethNode {
             nonce_manager: Default::default(),
             persistent_http_provider: OnceCell::const_new(),
             persistent_ws_provider: OnceCell::const_new(),
-            http_provider_requests_semaphore: LazyLock::new(|| Semaphore::const_new(500)),
         }
     }
 
@@ -566,8 +564,6 @@ impl EthereumNode for LighthouseGethNode {
         transaction: TransactionRequest,
     ) -> Pin<Box<dyn Future<Output = anyhow::Result<TxHash>> + '_>> {
         Box::pin(async move {
-            let _permit = self.http_provider_requests_semaphore.acquire().await;
-
             let provider = self
                 .http_provider()
                 .await

crates/node/src/node_implementations/substrate.rs

@@ -54,7 +54,10 @@ use crate::{
     Node,
     constants::{CHAIN_ID, INITIAL_BALANCE},
     helpers::{Process, ProcessReadinessWaitBehavior},
-    provider_utils::{ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider},
+    provider_utils::{
+        ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider,
+        execute_transaction,
+    },
 };
 
 static NODE_COUNT: AtomicU32 = AtomicU32::new(0);
@@ -80,7 +83,7 @@ pub struct SubstrateNode {
 }
 
 impl SubstrateNode {
-    const BASE_DIRECTORY: &str = "Substrate";
+    const BASE_DIRECTORY: &str = "substrate";
     const LOGS_DIRECTORY: &str = "logs";
     const DATA_DIRECTORY: &str = "chains";
 
@@ -346,7 +349,7 @@ impl SubstrateNode {
             .get_or_try_init(|| async move {
                 construct_concurrency_limited_provider::<ReviveNetwork, _>(
                     self.rpc_url.as_str(),
-                    FallbackGasFiller::new(250_000_000, 5_000_000_000, 1_000_000_000),
+                    FallbackGasFiller::new(u64::MAX, 5_000_000_000, 1_000_000_000),
                     ChainIdFiller::new(Some(CHAIN_ID)),
                     NonceFiller::new(self.nonce_manager.clone()),
                     self.wallet.clone(),
@@ -408,23 +411,12 @@ impl EthereumNode for SubstrateNode {
         &self,
         transaction: TransactionRequest,
     ) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
-        static SEMAPHORE: std::sync::LazyLock<tokio::sync::Semaphore> =
-            std::sync::LazyLock::new(|| tokio::sync::Semaphore::new(500));
-
         Box::pin(async move {
-            let _permit = SEMAPHORE.acquire().await?;
-
-            let receipt = self
+            let provider = self
                 .provider()
                 .await
-                .context("Failed to create provider for transaction submission")?
-                .send_transaction(transaction)
-                .await
-                .context("Failed to submit transaction to substrate proxy")?
-                .get_receipt()
-                .await
-                .context("Failed to fetch transaction receipt from substrate proxy")?;
-            Ok(receipt)
+                .context("Failed to create the provider")?;
+            execute_transaction(provider, transaction).await
         })
     }
 

crates/node/src/provider_utils/provider.rs

@@ -1,14 +1,16 @@
-use std::sync::LazyLock;
+use std::{ops::ControlFlow, sync::LazyLock, time::Duration};
 
 use alloy::{
-    network::{Network, NetworkWallet, TransactionBuilder4844},
+    network::{Ethereum, Network, NetworkWallet, TransactionBuilder4844},
     providers::{
-        Identity, ProviderBuilder, RootProvider,
+        Identity, PendingTransactionBuilder, Provider, ProviderBuilder, RootProvider,
         fillers::{ChainIdFiller, FillProvider, JoinFill, NonceFiller, TxFiller, WalletFiller},
     },
     rpc::client::ClientBuilder,
 };
 use anyhow::{Context, Result};
+use revive_dt_common::futures::{PollingWaitBehavior, poll};
+use tracing::debug;
 
 use crate::provider_utils::{ConcurrencyLimiterLayer, FallbackGasFiller};
 
@@ -61,3 +63,66 @@ where
 
     Ok(provider)
 }
+
+pub async fn execute_transaction<N, W>(
+    provider: ConcreteProvider<N, W>,
+    transaction: N::TransactionRequest,
+) -> Result<N::ReceiptResponse>
+where
+    N: Network<
+            TransactionRequest: TransactionBuilder4844,
+            TxEnvelope = <Ethereum as Network>::TxEnvelope,
+        >,
+    W: NetworkWallet<N>,
+    Identity: TxFiller<N>,
+    FallbackGasFiller: TxFiller<N>,
+    ChainIdFiller: TxFiller<N>,
+    NonceFiller: TxFiller<N>,
+    WalletFiller<W>: TxFiller<N>,
+{
+    let sendable_transaction = provider
+        .fill(transaction)
+        .await
+        .context("Failed to fill transaction")?;
+
+    let transaction_envelope = sendable_transaction
+        .try_into_envelope()
+        .context("Failed to convert transaction into an envelope")?;
+    let tx_hash = *transaction_envelope.tx_hash();
+
+    let mut pending_transaction = match provider.send_tx_envelope(transaction_envelope).await {
+        Ok(pending_transaction) => pending_transaction,
+        Err(error) => {
+            let error_string = error.to_string();
+
+            if error_string.contains("Transaction Already Imported") {
+                PendingTransactionBuilder::<N>::new(provider.root().clone(), tx_hash)
+            } else {
+                return Err(error).context(format!("Failed to submit transaction {tx_hash}"));
+            }
+        }
+    };
+    debug!(%tx_hash, "Submitted Transaction");
+
+    pending_transaction.set_timeout(Some(Duration::from_secs(120)));
+    let tx_hash = pending_transaction.watch().await.context(format!(
+        "Transaction inclusion watching timeout for {tx_hash}"
+    ))?;
+
+    poll(
+        Duration::from_secs(60),
+        PollingWaitBehavior::Constant(Duration::from_secs(3)),
+        || {
+            let provider = provider.clone();
+
+            async move {
+                match provider.get_transaction_receipt(tx_hash).await {
+                    Ok(Some(receipt)) => Ok(ControlFlow::Break(receipt)),
+                    _ => Ok(ControlFlow::Continue(())),
+                }
+            }
+        },
+    )
+    .await
+    .context(format!("Polling for receipt failed for {tx_hash}"))
+}