fix: reload queued senders on USDC storage changes

When USDC predeploy storage slots change in a canonical update, queued
transactions whose senders have affected balance slots must be
re-evaluated. This adds extend_reload_queued_senders to the
ChangedAccountsHook trait and implements it in SeismicBalanceHook to
detect USDC slot mutations and mark the corresponding queued senders
as dirty.

Author: drappi-ai

crates/seismic/txpool/src/maintain.rs

@@ -1,9 +1,11 @@
 //! Seismic-specific pool maintenance hook that augments native balances with
 //! USDC predeploy balances.
 
-use reth_execution_types::ChangedAccount;
+use alloy_primitives::{Address, B256};
+use reth_execution_types::{ChangedAccount, ExecutionOutcome};
 use reth_provider::StateProvider;
 use reth_transaction_pool::maintain::ChangedAccountsHook;
+use std::collections::HashSet;
 use tracing::debug;
 
 /// A [`ChangedAccountsHook`] that reads each sender's USDC predeploy balance
@@ -34,4 +36,40 @@ impl ChangedAccountsHook for SeismicBalanceHook {
             }
         }
     }
+
+    fn extend_reload_queued_senders<R>(
+        &self,
+        queued_senders: &HashSet<Address>,
+        old: Option<&ExecutionOutcome<R>>,
+        new: &ExecutionOutcome<R>,
+        dirty_addresses: &mut HashSet<Address>,
+    ) {
+        dirty_addresses.extend(queued_senders_with_changed_usdc_slots(queued_senders, old, new));
+    }
+}
+
+fn queued_senders_with_changed_usdc_slots<'a, R>(
+    queued_senders: &'a HashSet<Address>,
+    old: Option<&ExecutionOutcome<R>>,
+    new: &ExecutionOutcome<R>,
+) -> impl Iterator<Item = Address> + 'a {
+    let changed_slots = changed_usdc_storage_slots(new)
+        .into_iter()
+        .chain(old.into_iter().flat_map(changed_usdc_storage_slots))
+        .collect::<HashSet<_>>();
+
+    queued_senders.iter().copied().filter(move |address| {
+        changed_slots.contains(&crate::usdc::usdc_balance_storage_key(address))
+    })
+}
+
+fn changed_usdc_storage_slots<R>(state: &ExecutionOutcome<R>) -> impl Iterator<Item = B256> + '_ {
+    state
+        .bundle_accounts_iter()
+        .filter_map(|(address, account)| (address == crate::usdc::USDC_CONTRACT).then_some(account))
+        .flat_map(|account| {
+            account.storage.iter().filter_map(|(slot, value)| {
+                value.is_changed().then(|| B256::from(slot.to_be_bytes::<32>()))
+            })
+        })
 }

crates/seismic/txpool/src/usdc.rs

@@ -22,7 +22,7 @@ const BALANCES_MAPPING_SLOT: U256 = U256::from_limbs([3, 0, 0, 0]);
 /// For a Solidity `mapping(address => uint256)` at slot `s`, the value for key
 /// `k` is stored at `keccak256(abi.encode(k, s))` — i.e. `k` left-padded to 32
 /// bytes concatenated with `s` as a 32-byte big-endian integer.
-fn usdc_balance_storage_key(address: &Address) -> B256 {
+pub(crate) fn usdc_balance_storage_key(address: &Address) -> B256 {
     let mut buf = [0u8; 64];
     // address is 20 bytes, right-aligned in the first 32-byte word
     buf[12..32].copy_from_slice(address.as_slice());

crates/transaction-pool/src/maintain.rs

@@ -17,7 +17,7 @@ use futures_util::{
 };
 use reth_chain_state::CanonStateNotification;
 use reth_chainspec::{ChainSpecProvider, EthChainSpec};
-use reth_execution_types::ChangedAccount;
+use reth_execution_types::{ChangedAccount, ExecutionOutcome};
 use reth_fs_util::FsPathError;
 use reth_primitives_traits::{
     transaction::signed::SignedTransaction, NodePrimitives, SealedHeader,
@@ -106,6 +106,17 @@ pub trait ChangedAccountsHook: Send + Sync + 'static {
     /// load native balance/nonce, so implementations always see a consistent
     /// view of the chain.
     fn transform(&self, state: &dyn StateProvider, accounts: &mut Vec<ChangedAccount>);
+
+    /// Extends the dirty set with queued senders whose executability may have
+    /// changed due to off-account state updates in the canonical update.
+    fn extend_reload_queued_senders<R>(
+        &self,
+        _queued_senders: &HashSet<Address>,
+        _old: Option<&ExecutionOutcome<R>>,
+        _new: &ExecutionOutcome<R>,
+        _dirty_addresses: &mut HashSet<Address>,
+    ) {
+    }
 }
 
 /// No-op implementation for chains that don't need balance augmentation.
@@ -498,6 +509,14 @@ pub async fn maintain_transaction_pool_with_hook<N, Client, P, St, Tasks, H>(
                 metrics.inc_reinserted_transactions(pruned_old_transactions.len());
                 let _ = pool.add_external_transactions(pruned_old_transactions).await;
 
+                let queued_senders = queued_senders(&pool);
+                hook.extend_reload_queued_senders(
+                    &queued_senders,
+                    Some(old_state),
+                    new_state,
+                    &mut dirty_addresses,
+                );
+
                 // keep track of new mined blob transactions
                 blob_store_tracker.add_new_chain_blocks(&new_blocks);
             }
@@ -575,6 +594,14 @@ pub async fn maintain_transaction_pool_with_hook<N, Client, P, St, Tasks, H>(
                 };
                 pool.on_canonical_state_change(update);
 
+                let queued_senders = queued_senders(&pool);
+                hook.extend_reload_queued_senders(
+                    &queued_senders,
+                    None,
+                    state,
+                    &mut dirty_addresses,
+                );
+
                 // keep track of mined blob transactions
                 blob_store_tracker.add_new_chain_blocks(&blocks);
             }
@@ -683,6 +710,10 @@ where
     Ok(res)
 }
 
+fn queued_senders<P: TransactionPool>(pool: &P) -> HashSet<Address> {
+    pool.queued_transactions().into_iter().map(|tx| tx.sender()).collect()
+}
+
 /// Loads transactions from a file, decodes them from the JSON or RLP format, and
 /// inserts them into the transaction pool on node boot up.
 /// The file is removed after the transactions have been successfully processed.