feat(sequencer): add ttl and invalid cache to app mempool

crates/astria-core/src/protocol/abci.rs

@@ -18,6 +18,8 @@ impl AbciErrorCode {
     pub const INSUFFICIENT_FUNDS: Self = Self(6);
     pub const INVALID_CHAIN_ID: Self = Self(7);
     pub const VALUE_NOT_FOUND: Self = Self(8);
+    pub const TRANSACTION_EXPIRED: Self = Self(9);
+    pub const TRANSACTION_FAILED: Self = Self(10);
 }
 
 impl AbciErrorCode {
@@ -33,6 +35,8 @@ impl AbciErrorCode {
             6 => "insufficient funds".into(),
             7 => "the provided chain id was invalid".into(),
             8 => "the requested value was not found".into(),
+            9 => "the transaction expired in the app's mempool".into(),
+            10 => "the transaction failed to execute in prepare_proposal()".into(),
             other => format!("unknown non-zero abci error code: {other}").into(),
         }
     }
@@ -61,6 +65,8 @@ impl From<NonZeroU32> for AbciErrorCode {
             6 => Self::INSUFFICIENT_FUNDS,
             7 => Self::INVALID_CHAIN_ID,
             8 => Self::VALUE_NOT_FOUND,
+            9 => Self::TRANSACTION_EXPIRED,
+            10 => Self::TRANSACTION_FAILED,
             other => Self(other),
         }
     }

crates/astria-sequencer/Cargo.toml

@@ -67,6 +67,7 @@ config = { package = "astria-config", path = "../astria-config", features = [
   "tests",
 ] }
 insta = { workspace = true, features = ["json"] }
+tokio = { workspace = true, features = ["test-util"] }
 
 [build-dependencies]
 astria-build-info = { path = "../astria-build-info", features = ["build"] }

crates/astria-sequencer/src/app/mod.rs

@@ -88,7 +88,10 @@ use crate::{
     component::Component as _,
     genesis::GenesisState,
     ibc::component::IbcComponent,
-    mempool::Mempool,
+    mempool::{
+        Mempool,
+        RemovalReason,
+    },
     metrics::Metrics,
     proposal::{
         block_size_constraints::BlockSizeConstraints,
@@ -547,12 +550,20 @@ impl App {
                     );
                     failed_tx_count = failed_tx_count.saturating_add(1);
 
-                    // we re-insert the tx into the mempool if it failed to execute
-                    // due to an invalid nonce, as it may be valid in the future.
-                    // if it's invalid due to the nonce being too low, it'll be
-                    // removed from the mempool in `update_mempool_after_finalization`.
                     if e.downcast_ref::<InvalidNonce>().is_some() {
+                        // we re-insert the tx into the mempool if it failed to execute
+                        // due to an invalid nonce, as it may be valid in the future.
+                        // if it's invalid due to the nonce being too low, it'll be
+                        // removed from the mempool in `update_mempool_after_finalization`.
                         txs_to_readd_to_mempool.push((enqueued_tx, priority));
+                    } else {
+                        // the transaction should be removed from the cometbft mempool
+                        self.mempool
+                            .track_removal_comet_bft(
+                                enqueued_tx.tx_hash(),
+                                RemovalReason::FailedPrepareProposal(e.to_string()),
+                            )
+                            .await;
                     }
                 }
             }
@@ -1119,9 +1130,7 @@ async fn update_mempool_after_finalization<S: StateReadExt>(
     state: S,
 ) -> anyhow::Result<()> {
     let current_account_nonce_getter = |address: Address| state.get_account_nonce(address);
-    mempool
-        .update_priorities(current_account_nonce_getter)
-        .await
+    mempool.run_maintenance(current_account_nonce_getter).await
 }
 
 /// relevant data of a block being executed.