diff --git a/Cargo.lock b/Cargo.lock
index 2d4012b328c..8fd222d46c7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -8340,6 +8340,7 @@ dependencies = [
  "metrics",
  "metrics-util",
  "moka",
+ "parking_lot",
  "proptest",
  "rand 0.8.6",
  "rand 0.9.4",
diff --git a/crates/chain-state/src/execution_stats.rs b/crates/chain-state/src/execution_stats.rs
index 4b87df1ac62..ee5255dcbc6 100644
--- a/crates/chain-state/src/execution_stats.rs
+++ b/crates/chain-state/src/execution_stats.rs
@@ -54,16 +54,28 @@ pub struct ExecutionTimingStats {
     pub eip7702_delegations_set: usize,
     /// Number of EIP-7702 delegations cleared
     pub eip7702_delegations_cleared: usize,
-    /// Account cache hits
+    /// Execution-cache account hits
     pub account_cache_hits: usize,
-    /// Account cache misses
+    /// Execution-cache account misses
     pub account_cache_misses: usize,
-    /// Storage cache hits
+    /// Execution-cache storage hits
     pub storage_cache_hits: usize,
-    /// Storage cache misses
+    /// Execution-cache storage misses
     pub storage_cache_misses: usize,
-    /// Code cache hits
+    /// Execution-cache code hits
     pub code_cache_hits: usize,
-    /// Code cache misses
+    /// Execution-cache code misses
     pub code_cache_misses: usize,
+    /// Txpool-prewarm snapshot account hits
+    pub txpool_snapshot_account_hits: usize,
+    /// Txpool-prewarm snapshot account misses
+    pub txpool_snapshot_account_misses: usize,
+    /// Txpool-prewarm snapshot storage hits
+    pub txpool_snapshot_storage_hits: usize,
+    /// Txpool-prewarm snapshot storage misses
+    pub txpool_snapshot_storage_misses: usize,
+    /// Txpool-prewarm snapshot code hits
+    pub txpool_snapshot_code_hits: usize,
+    /// Txpool-prewarm snapshot code misses
+    pub txpool_snapshot_code_misses: usize,
 }
diff --git a/crates/engine/execution-cache/src/cached_state.rs b/crates/engine/execution-cache/src/cached_state.rs
index e6963da71f7..93bf8e943bc 100644
--- a/crates/engine/execution-cache/src/cached_state.rs
+++ b/crates/engine/execution-cache/src/cached_state.rs
@@ -1,4 +1,5 @@
 //! Execution cache implementation for block processing.
+use crate::TxPoolPrewarmCacheSnapshot;
 use alloy_primitives::{
     map::{DefaultHashBuilder, FbBuildHasher},
     Address, StorageKey, StorageValue, B256,
@@ -92,8 +93,9 @@ type FixedCache<K, V, H = DefaultHashBuilder> = fixed_cache::Cache<K, V, H, Epoc
 /// already caches reads during the block, and the shared cache is updated after the block from the
 /// final [`BundleState`]. See also [`ExecutionCache::insert_state`].
 ///
-/// Normal cache hit/miss metrics are recorded when [`CachedStateMetrics`] is provided. Slow-block
-/// [`CacheStats`] are controlled separately by [`Self::new_with_mode`].
+/// Execution-cache and txpool-snapshot hit/miss metrics are recorded separately when
+/// [`CachedStateMetrics`] is provided. Slow-block [`CacheStats`] are controlled separately by
+/// [`Self::new_with_mode`].
 #[derive(Debug)]
 pub struct CachedStateProvider<S> {
     /// The state provider
@@ -102,11 +104,17 @@ pub struct CachedStateProvider<S> {
     /// The caches used for the provider
     caches: ExecutionCache,
 
+    /// Optional immutable txpool-prewarm snapshot consulted before the regular execution cache.
+    txpool_snapshot: Option<TxPoolPrewarmCacheSnapshot>,
+
     /// Metrics for the cached state provider.
     metrics: Option<CachedStateMetrics>,
 
-    /// Provider-local hit/miss counters flushed when the provider is dropped.
-    metric_counts: CacheMetricCounts,
+    /// Provider-local execution-cache hit/miss counters flushed when the provider is dropped.
+    execution_metric_counts: CacheMetricCounts,
+
+    /// Provider-local txpool-cache hit/miss counters flushed when the provider is dropped.
+    txpool_metric_counts: CacheMetricCounts,
 
     /// Whether cache misses should populate the shared execution cache.
     fill_mode: CacheFillMode,
@@ -146,13 +154,21 @@ impl<S> CachedStateProvider<S> {
         Self {
             state_provider,
             caches,
+            txpool_snapshot: None,
             metrics,
-            metric_counts: CacheMetricCounts::new(),
+            execution_metric_counts: CacheMetricCounts::new(),
+            txpool_metric_counts: CacheMetricCounts::new(),
             fill_mode,
             cache_stats,
         }
     }
 
+    /// Adds an immutable txpool-prewarm snapshot as the first cache lookup tier.
+    pub fn with_txpool_snapshot(mut self, snapshot: Option<TxPoolPrewarmCacheSnapshot>) -> Self {
+        self.txpool_snapshot = snapshot;
+        self
+    }
+
     fn record_account_hit(&self) {
         self.record_metric(CacheMetricKind::AccountHit);
         if let Some(stats) = &self.cache_stats {
@@ -195,21 +211,72 @@ impl<S> CachedStateProvider<S> {
         }
     }
 
+    fn record_txpool_account_hit(&self) {
+        self.record_txpool_metric(CacheMetricKind::AccountHit);
+        if let Some(stats) = &self.cache_stats {
+            stats.record_txpool_snapshot_account_hit();
+        }
+    }
+
+    fn record_txpool_account_miss(&self) {
+        self.record_txpool_metric(CacheMetricKind::AccountMiss);
+        if let Some(stats) = &self.cache_stats {
+            stats.record_txpool_snapshot_account_miss();
+        }
+    }
+
+    fn record_txpool_storage_hit(&self) {
+        self.record_txpool_metric(CacheMetricKind::StorageHit);
+        if let Some(stats) = &self.cache_stats {
+            stats.record_txpool_snapshot_storage_hit();
+        }
+    }
+
+    fn record_txpool_storage_miss(&self) {
+        self.record_txpool_metric(CacheMetricKind::StorageMiss);
+        if let Some(stats) = &self.cache_stats {
+            stats.record_txpool_snapshot_storage_miss();
+        }
+    }
+
+    fn record_txpool_code_hit(&self) {
+        self.record_txpool_metric(CacheMetricKind::CodeHit);
+        if let Some(stats) = &self.cache_stats {
+            stats.record_txpool_snapshot_code_hit();
+        }
+    }
+
+    fn record_txpool_code_miss(&self) {
+        self.record_txpool_metric(CacheMetricKind::CodeMiss);
+        if let Some(stats) = &self.cache_stats {
+            stats.record_txpool_snapshot_code_miss();
+        }
+    }
+
     #[inline]
     fn record_metric(&self, kind: CacheMetricKind) {
         if self.metrics.is_some() {
-            self.metric_counts.record(kind);
+            self.execution_metric_counts.record(kind);
+        }
+    }
+
+    #[inline]
+    fn record_txpool_metric(&self, kind: CacheMetricKind) {
+        if self.metrics.is_some() {
+            self.txpool_metric_counts.record(kind);
         }
     }
 
     fn flush_buffered_metrics(&self) {
-        let counts = self.metric_counts.take();
-        if counts.is_empty() {
+        let execution_counts = self.execution_metric_counts.take();
+        let txpool_counts = self.txpool_metric_counts.take();
+        if execution_counts.is_empty() && txpool_counts.is_empty() {
             return;
         }
 
         if let Some(metrics) = &self.metrics {
-            metrics.record_access_counts(counts);
+            metrics.record_access_counts(execution_counts);
+            metrics.record_txpool_access_counts(txpool_counts);
         }
     }
 
@@ -343,7 +410,7 @@ impl fmt::Display for CachedStateMetricsSource {
     }
 }
 
-/// Metrics for the cached state provider, showing hits / misses for each cache
+/// Metrics for the cached state provider, showing hits and misses for each cache tier.
 #[derive(Metrics, Clone)]
 #[metrics(scope = "sync.caching")]
 pub struct CachedStateMetrics {
@@ -353,23 +420,41 @@ pub struct CachedStateMetrics {
     /// Duration of execution cache creation in seconds
     execution_cache_creation_duration_seconds: Histogram,
 
-    /// Code cache hits
+    /// Execution-cache code hits
     code_cache_hits: Gauge,
 
-    /// Code cache misses
+    /// Execution-cache code misses
     code_cache_misses: Gauge,
 
-    /// Storage cache hits
+    /// Execution-cache storage hits
     storage_cache_hits: Gauge,
 
-    /// Storage cache misses
+    /// Execution-cache storage misses
     storage_cache_misses: Gauge,
 
-    /// Account cache hits
+    /// Execution-cache account hits
     account_cache_hits: Gauge,
 
-    /// Account cache misses
+    /// Execution-cache account misses
     account_cache_misses: Gauge,
+
+    /// Txpool-prewarm snapshot code hits
+    txpool_snapshot_code_hits: Gauge,
+
+    /// Txpool-prewarm snapshot code misses
+    txpool_snapshot_code_misses: Gauge,
+
+    /// Txpool-prewarm snapshot storage hits
+    txpool_snapshot_storage_hits: Gauge,
+
+    /// Txpool-prewarm snapshot storage misses
+    txpool_snapshot_storage_misses: Gauge,
+
+    /// Txpool-prewarm snapshot account hits
+    txpool_snapshot_account_hits: Gauge,
+
+    /// Txpool-prewarm snapshot account misses
+    txpool_snapshot_account_misses: Gauge,
 }
 
 /// Metrics for shared execution cache state.
@@ -418,6 +503,18 @@ impl CachedStateMetrics {
         // account cache
         self.account_cache_hits.set(0);
         self.account_cache_misses.set(0);
+
+        // txpool-prewarm code cache
+        self.txpool_snapshot_code_hits.set(0);
+        self.txpool_snapshot_code_misses.set(0);
+
+        // txpool-prewarm storage cache
+        self.txpool_snapshot_storage_hits.set(0);
+        self.txpool_snapshot_storage_misses.set(0);
+
+        // txpool-prewarm account cache
+        self.txpool_snapshot_account_hits.set(0);
+        self.txpool_snapshot_account_misses.set(0);
     }
 
     /// Returns a new zeroed-out instance of [`CachedStateMetrics`] with a `source` label
@@ -460,6 +557,46 @@ impl CachedStateMetrics {
         }
     }
 
+    fn record_txpool_access(&self, kind: CacheMetricKind, count: u64) {
+        match kind {
+            CacheMetricKind::AccountHit => {
+                self.txpool_snapshot_account_hits.increment(count as f64)
+            }
+            CacheMetricKind::AccountMiss => {
+                self.txpool_snapshot_account_misses.increment(count as f64)
+            }
+            CacheMetricKind::StorageHit => {
+                self.txpool_snapshot_storage_hits.increment(count as f64)
+            }
+            CacheMetricKind::StorageMiss => {
+                self.txpool_snapshot_storage_misses.increment(count as f64)
+            }
+            CacheMetricKind::CodeHit => self.txpool_snapshot_code_hits.increment(count as f64),
+            CacheMetricKind::CodeMiss => self.txpool_snapshot_code_misses.increment(count as f64),
+        }
+    }
+
+    fn record_txpool_access_counts(&self, counts: CacheMetricSnapshot) {
+        if counts.account_hits != 0 {
+            self.record_txpool_access(CacheMetricKind::AccountHit, counts.account_hits);
+        }
+        if counts.account_misses != 0 {
+            self.record_txpool_access(CacheMetricKind::AccountMiss, counts.account_misses);
+        }
+        if counts.storage_hits != 0 {
+            self.record_txpool_access(CacheMetricKind::StorageHit, counts.storage_hits);
+        }
+        if counts.storage_misses != 0 {
+            self.record_txpool_access(CacheMetricKind::StorageMiss, counts.storage_misses);
+        }
+        if counts.code_hits != 0 {
+            self.record_txpool_access(CacheMetricKind::CodeHit, counts.code_hits);
+        }
+        if counts.code_misses != 0 {
+            self.record_txpool_access(CacheMetricKind::CodeMiss, counts.code_misses);
+        }
+    }
+
     /// Records a new execution cache creation with its duration.
     pub fn record_cache_creation(&self, duration: Duration) {
         self.execution_cache_created_total.increment(1);
@@ -470,18 +607,30 @@ impl CachedStateMetrics {
 /// Cache hit/miss statistics for detailed block logging.
 #[derive(Debug, Default)]
 pub struct CacheStats {
-    /// Account cache hits
+    /// Execution-cache account hits
     account_hits: AtomicUsize,
-    /// Account cache misses
+    /// Execution-cache account misses
     account_misses: AtomicUsize,
-    /// Storage cache hits
+    /// Execution-cache storage hits
     storage_hits: AtomicUsize,
-    /// Storage cache misses
+    /// Execution-cache storage misses
     storage_misses: AtomicUsize,
-    /// Code cache hits
+    /// Execution-cache code hits
     code_hits: AtomicUsize,
-    /// Code cache misses
+    /// Execution-cache code misses
     code_misses: AtomicUsize,
+    /// Txpool-prewarm snapshot account hits
+    txpool_snapshot_account_hits: AtomicUsize,
+    /// Txpool-prewarm snapshot account misses
+    txpool_snapshot_account_misses: AtomicUsize,
+    /// Txpool-prewarm snapshot storage hits
+    txpool_snapshot_storage_hits: AtomicUsize,
+    /// Txpool-prewarm snapshot storage misses
+    txpool_snapshot_storage_misses: AtomicUsize,
+    /// Txpool-prewarm snapshot code hits
+    txpool_snapshot_code_hits: AtomicUsize,
+    /// Txpool-prewarm snapshot code misses
+    txpool_snapshot_code_misses: AtomicUsize,
 }
 
 impl CacheStats {
@@ -544,6 +693,66 @@ impl CacheStats {
     pub fn code_misses(&self) -> usize {
         self.code_misses.load(Ordering::Relaxed)
     }
+
+    /// Records a txpool-prewarm snapshot account hit.
+    pub fn record_txpool_snapshot_account_hit(&self) {
+        self.txpool_snapshot_account_hits.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Records a txpool-prewarm snapshot account miss.
+    pub fn record_txpool_snapshot_account_miss(&self) {
+        self.txpool_snapshot_account_misses.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Returns the number of txpool-prewarm snapshot account hits.
+    pub fn txpool_snapshot_account_hits(&self) -> usize {
+        self.txpool_snapshot_account_hits.load(Ordering::Relaxed)
+    }
+
+    /// Returns the number of txpool-prewarm snapshot account misses.
+    pub fn txpool_snapshot_account_misses(&self) -> usize {
+        self.txpool_snapshot_account_misses.load(Ordering::Relaxed)
+    }
+
+    /// Records a txpool-prewarm snapshot storage hit.
+    pub fn record_txpool_snapshot_storage_hit(&self) {
+        self.txpool_snapshot_storage_hits.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Records a txpool-prewarm snapshot storage miss.
+    pub fn record_txpool_snapshot_storage_miss(&self) {
+        self.txpool_snapshot_storage_misses.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Returns the number of txpool-prewarm snapshot storage hits.
+    pub fn txpool_snapshot_storage_hits(&self) -> usize {
+        self.txpool_snapshot_storage_hits.load(Ordering::Relaxed)
+    }
+
+    /// Returns the number of txpool-prewarm snapshot storage misses.
+    pub fn txpool_snapshot_storage_misses(&self) -> usize {
+        self.txpool_snapshot_storage_misses.load(Ordering::Relaxed)
+    }
+
+    /// Records a txpool-prewarm snapshot code hit.
+    pub fn record_txpool_snapshot_code_hit(&self) {
+        self.txpool_snapshot_code_hits.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Records a txpool-prewarm snapshot code miss.
+    pub fn record_txpool_snapshot_code_miss(&self) {
+        self.txpool_snapshot_code_misses.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Returns the number of txpool-prewarm snapshot code hits.
+    pub fn txpool_snapshot_code_hits(&self) -> usize {
+        self.txpool_snapshot_code_hits.load(Ordering::Relaxed)
+    }
+
+    /// Returns the number of txpool-prewarm snapshot code misses.
+    pub fn txpool_snapshot_code_misses(&self) -> usize {
+        self.txpool_snapshot_code_misses.load(Ordering::Relaxed)
+    }
 }
 
 /// A stats handler for fixed-cache that tracks collisions and size.
@@ -637,6 +846,14 @@ impl<K: PartialEq, V> StatsHandler<K, V> for CacheStatsHandler {
 
 impl<S: AccountReader> AccountReader for CachedStateProvider<S> {
     fn basic_account(&self, address: &Address) -> ProviderResult<Option<Account>> {
+        if let Some(snapshot) = &self.txpool_snapshot {
+            if let Some(account) = snapshot.account(address) {
+                self.record_txpool_account_hit();
+                return Ok(account)
+            }
+            self.record_txpool_account_miss();
+        }
+
         if self.should_fill_on_miss() {
             match self.caches.get_or_try_insert_account_with(*address, || {
                 self.state_provider.basic_account(address)
@@ -675,6 +892,14 @@ impl<S: StateProvider> StateProvider for CachedStateProvider<S> {
         account: Address,
         storage_key: StorageKey,
     ) -> ProviderResult<Option<StorageValue>> {
+        if let Some(snapshot) = &self.txpool_snapshot {
+            if let Some(value) = snapshot.storage(account, storage_key) {
+                self.record_txpool_storage_hit();
+                return Ok(nonzero_storage_value(value))
+            }
+            self.record_txpool_storage_miss();
+        }
+
         if self.should_fill_on_miss() {
             match self.caches.get_or_try_insert_storage_with(account, storage_key, || {
                 self.state_provider.storage(account, storage_key).map(Option::unwrap_or_default)
@@ -700,6 +925,14 @@ impl<S: StateProvider> StateProvider for CachedStateProvider<S> {
 
 impl<S: BytecodeReader> BytecodeReader for CachedStateProvider<S> {
     fn bytecode_by_hash(&self, code_hash: &B256) -> ProviderResult<Option<Bytecode>> {
+        if let Some(snapshot) = &self.txpool_snapshot {
+            if let Some(code) = snapshot.bytecode(code_hash) {
+                self.record_txpool_code_hit();
+                return Ok(code)
+            }
+            self.record_txpool_code_miss();
+        }
+
         if self.should_fill_on_miss() {
             match self.caches.get_or_try_insert_code_with(*code_hash, || {
                 self.state_provider.bytecode_by_hash(code_hash)
diff --git a/crates/engine/execution-cache/src/lib.rs b/crates/engine/execution-cache/src/lib.rs
index b276d19b66e..f9cd0cf5f1b 100644
--- a/crates/engine/execution-cache/src/lib.rs
+++ b/crates/engine/execution-cache/src/lib.rs
@@ -17,6 +17,9 @@
 mod cached_state;
 pub use cached_state::*;
 
+mod txpool;
+pub use txpool::*;
+
 use alloy_primitives::B256;
 use metrics::{Counter, Histogram};
 use parking_lot::Mutex;
diff --git a/crates/engine/execution-cache/src/txpool.rs b/crates/engine/execution-cache/src/txpool.rs
new file mode 100644
index 00000000000..66eed02eeb2
--- /dev/null
+++ b/crates/engine/execution-cache/src/txpool.rs
@@ -0,0 +1,101 @@
+//! Immutable snapshots produced by txpool-driven state prewarming.
+
+use alloy_primitives::{Address, StorageKey, StorageValue, B256, U256};
+use reth_primitives_traits::{Account, Bytecode};
+use reth_revm::cached::CachedReads;
+use std::sync::Arc;
+
+/// A deep, immutable txpool-prewarm cache snapshot for one parent state.
+///
+/// Wraps the [`CachedReads`] collected by speculative execution so cache tiers can serve lookups
+/// from it directly.
+#[derive(Clone, Debug)]
+pub struct TxPoolPrewarmCacheSnapshot {
+    parent_hash: B256,
+    reads: Arc<CachedReads>,
+}
+
+impl TxPoolPrewarmCacheSnapshot {
+    /// Creates a snapshot of `reads` collected against `parent_hash`'s state.
+    pub const fn new(parent_hash: B256, reads: Arc<CachedReads>) -> Self {
+        Self { parent_hash, reads }
+    }
+
+    /// Returns the hash of the state this snapshot was warmed against.
+    pub const fn parent_hash(&self) -> B256 {
+        self.parent_hash
+    }
+
+    /// Returns a cached account, preserving cached non-existence.
+    pub fn account(&self, address: &Address) -> Option<Option<Account>> {
+        self.reads.accounts.get(address).map(|account| account.info.as_ref().map(Account::from))
+    }
+
+    /// Returns a cached storage value, preserving cached zero values.
+    pub fn storage(&self, address: Address, key: StorageKey) -> Option<StorageValue> {
+        self.reads.accounts.get(&address)?.storage.get(&U256::from_be_bytes(key.0)).copied()
+    }
+
+    /// Returns cached bytecode.
+    ///
+    /// [`CachedReads`] stores empty bytecode for code it could not find, so an empty entry reads
+    /// as a miss and the caller's fallback tier decides.
+    pub fn bytecode(&self, code_hash: &B256) -> Option<Option<Bytecode>> {
+        let code = self.reads.contracts.get(code_hash)?;
+        (!code.is_empty()).then(|| Some(Bytecode(code.clone())))
+    }
+
+    /// Returns `(accounts, storage slots, bytecodes)` in the snapshot.
+    pub fn entry_counts(&self) -> (usize, usize, usize) {
+        (
+            self.reads.accounts.len(),
+            self.reads.accounts.values().map(|account| account.storage.len()).sum(),
+            self.reads.contracts.len(),
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloy_primitives::map::U256Map;
+    use reth_revm::{
+        cached::CachedAccount,
+        revm::{bytecode::Bytecode as RevmBytecode, state::AccountInfo},
+    };
+
+    #[test]
+    fn lookups_preserve_cache_semantics() {
+        let owner = Address::repeat_byte(0x01);
+        let missing = Address::repeat_byte(0x02);
+        let code_hash = B256::repeat_byte(0x0C);
+        let empty_code_hash = B256::repeat_byte(0x0D);
+
+        let mut reads = CachedReads::default();
+        let mut storage = U256Map::default();
+        storage.insert(U256::from(1), U256::from(7));
+        storage.insert(U256::from(2), U256::ZERO);
+        reads.insert_account(owner, AccountInfo { nonce: 3, ..Default::default() }, storage);
+        reads.accounts.insert(missing, CachedAccount { info: None, storage: Default::default() });
+        reads.contracts.insert(code_hash, RevmBytecode::new_raw([0x60, 0x01].into()));
+        reads.contracts.insert(empty_code_hash, RevmBytecode::default());
+
+        let snapshot = TxPoolPrewarmCacheSnapshot::new(B256::ZERO, Arc::new(reads));
+
+        assert_eq!(snapshot.account(&owner).unwrap().unwrap().nonce, 3);
+        assert_eq!(snapshot.account(&missing), Some(None), "non-existence is a cacheable fact");
+        assert_eq!(snapshot.account(&Address::repeat_byte(0x03)), None);
+
+        let slot = |n: u64| B256::from(U256::from(n));
+        assert_eq!(snapshot.storage(owner, slot(1)), Some(U256::from(7)));
+        assert_eq!(snapshot.storage(owner, slot(2)), Some(U256::ZERO), "zero values are hits");
+        assert_eq!(snapshot.storage(owner, slot(9)), None);
+        assert_eq!(snapshot.storage(missing, slot(1)), None);
+
+        assert!(snapshot.bytecode(&code_hash).unwrap().is_some());
+        assert_eq!(snapshot.bytecode(&empty_code_hash), None, "empty code reads as a miss");
+        assert_eq!(snapshot.bytecode(&B256::repeat_byte(0x0E)), None);
+
+        assert_eq!(snapshot.entry_counts(), (2, 2, 2));
+    }
+}
diff --git a/crates/engine/primitives/src/config.rs b/crates/engine/primitives/src/config.rs
index 25654279e38..108db6ee47e 100644
--- a/crates/engine/primitives/src/config.rs
+++ b/crates/engine/primitives/src/config.rs
@@ -114,6 +114,8 @@ pub struct TreeConfig {
     disable_state_cache: bool,
     /// Whether to disable parallel prewarming.
     disable_prewarming: bool,
+    /// Whether txpool-driven prewarming between payloads is enabled.
+    txpool_prewarming: bool,
     /// Whether to enable state provider metrics.
     state_provider_metrics: bool,
     /// Cross-block cache size in bytes.
@@ -216,6 +218,7 @@ impl Default for TreeConfig {
             always_compare_trie_updates: false,
             disable_state_cache: false,
             disable_prewarming: false,
+            txpool_prewarming: false,
             state_provider_metrics: false,
             cross_block_cache_size: DEFAULT_CROSS_BLOCK_CACHE_SIZE,
             has_enough_parallelism: has_enough_parallelism(),
@@ -288,6 +291,7 @@ impl TreeConfig {
             always_compare_trie_updates,
             disable_state_cache,
             disable_prewarming,
+            txpool_prewarming: false,
             state_provider_metrics,
             cross_block_cache_size,
             has_enough_parallelism,
@@ -382,6 +386,11 @@ impl TreeConfig {
         self.disable_prewarming
     }
 
+    /// Returns whether txpool prewarming is enabled.
+    pub const fn txpool_prewarming(&self) -> bool {
+        self.txpool_prewarming
+    }
+
     /// Returns whether to always compare trie updates from the state root task to the trie updates
     /// from the regular state root calculation.
     pub const fn always_compare_trie_updates(&self) -> bool {
@@ -501,6 +510,12 @@ impl TreeConfig {
         self
     }
 
+    /// Enables or disables txpool transaction prewarming.
+    pub const fn with_txpool_prewarming(mut self, enabled: bool) -> Self {
+        self.txpool_prewarming = enabled;
+        self
+    }
+
     /// Setter for whether to always compare trie updates from the state root task to the trie
     /// updates from the regular state root calculation.
     pub const fn with_always_compare_trie_updates(
@@ -742,6 +757,12 @@ impl TreeConfig {
 mod tests {
     use super::TreeConfig;
 
+    #[test]
+    fn txpool_prewarming_is_disabled_by_default_and_can_be_enabled() {
+        assert!(!TreeConfig::default().txpool_prewarming());
+        assert!(TreeConfig::default().with_txpool_prewarming(true).txpool_prewarming());
+    }
+
     #[test]
     fn state_root_task_requires_parallelism_without_overrides() {
         assert!(TreeConfig::default().with_has_enough_parallelism(true).use_state_root_task());
diff --git a/crates/engine/tree/Cargo.toml b/crates/engine/tree/Cargo.toml
index 2a9281d5a28..5dceb9476cc 100644
--- a/crates/engine/tree/Cargo.toml
+++ b/crates/engine/tree/Cargo.toml
@@ -28,7 +28,7 @@ reth-primitives-traits = { workspace = true, features = ["rayon", "dashmap"] }
 reth-ethereum-primitives.workspace = true
 reth-provider.workspace = true
 reth-prune.workspace = true
-reth-revm = { workspace = true, features = ["optional-balance-check"] }
+reth-revm = { workspace = true, features = ["optional-balance-check", "optional-no-base-fee"] }
 reth-stages-api.workspace = true
 reth-tasks = { workspace = true, features = ["rayon"] }
 reth-trie-parallel.workspace = true
@@ -50,6 +50,7 @@ revm.workspace = true
 
 # common
 futures.workspace = true
+parking_lot.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["rt", "rt-multi-thread", "sync", "macros"] }
 moka = { workspace = true, features = ["sync"] }
diff --git a/crates/engine/tree/src/tree/mod.rs b/crates/engine/tree/src/tree/mod.rs
index b876693b8d2..f67a8732897 100644
--- a/crates/engine/tree/src/tree/mod.rs
+++ b/crates/engine/tree/src/tree/mod.rs
@@ -63,6 +63,7 @@ pub mod state_root_strategy;
 #[cfg(test)]
 mod tests;
 mod trie_updates;
+mod txpool_prewarm;
 pub mod types;
 
 use crate::{persistence::PersistenceResult, tree::error::AdvancePersistenceError};
@@ -75,7 +76,11 @@ pub use persistence_state::PersistenceState;
 pub use reth_engine_primitives::TreeConfig;
 pub use reth_execution_cache::{
     CachedStateCacheMetrics, CachedStateMetrics, CachedStateMetricsSource, CachedStateProvider,
-    ExecutionCache, PayloadExecutionCache, SavedCache,
+    ExecutionCache, PayloadExecutionCache, SavedCache, TxPoolPrewarmCacheSnapshot,
+};
+pub use txpool_prewarm::{
+    Source as TxPoolPrewarmSource, Transaction as TxPoolPrewarmTransaction,
+    Transactions as TxPoolPrewarmTransactions,
 };
 pub use types::{ExecutionEnv, ValidationOutcome, ValidationOutput};
 
@@ -1269,6 +1274,8 @@ where
             return Ok(Some(TreeOutcome::new(outcome)));
         }
 
+        self.payload_validator.on_canonical_head_changed(state.head_block_hash, &self.state);
+
         // Process payload attributes if the head is already canonical
         if let Some(attr) = attrs {
             let tip = self
@@ -1782,7 +1789,12 @@ where
                                         BeaconOnNewPayloadError::Internal(Box::new(e))
                                     }))
                                 {
-                                    error!(target: "engine::tree", payload=?num_hash, elapsed=?start.elapsed(), "Failed to send event: {err:?}");
+                                    error!(
+                                        target: "engine::tree",
+                                        payload=?num_hash,
+                                        elapsed=?latency,
+                                        "Failed to send event: {err:?}"
+                                    );
                                     self.metrics
                                         .engine
                                         .failed_new_payload_response_deliveries
@@ -2728,6 +2740,7 @@ where
         // update the tracked in-memory state with the new chain
         self.canonical_in_memory_state.update_chain(chain_update);
         self.canonical_in_memory_state.set_canonical_head(tip.clone());
+        self.payload_validator.on_canonical_head_changed(tip.hash(), &self.state);
 
         // Update metrics based on new tip
         self.metrics.tree.canonical_chain_height.set(tip.number() as f64);
diff --git a/crates/engine/tree/src/tree/payload_processor/bal_prewarm_pool.rs b/crates/engine/tree/src/tree/payload_processor/bal_prewarm_pool.rs
index 4c39b38b49b..734fa57c229 100644
--- a/crates/engine/tree/src/tree/payload_processor/bal_prewarm_pool.rs
+++ b/crates/engine/tree/src/tree/payload_processor/bal_prewarm_pool.rs
@@ -1,5 +1,5 @@
 use alloy_primitives::{Address, StorageKey};
-use reth_execution_cache::{CachedStateProvider, ExecutionCache};
+use reth_execution_cache::{CachedStateProvider, ExecutionCache, TxPoolPrewarmCacheSnapshot};
 use reth_provider::{
     AccountReader, BytecodeReader, ProviderResult, StateProvider, StateProviderBox,
 };
@@ -27,7 +27,11 @@ enum PrewarmTarget {
 /// FIFO): one `BeginBlock`, then the worker's share of `Warm`s, then one `EndBlock`.
 enum PrewarmMsg {
     /// Open a read txn for the new block: build a provider over the parent state and hold it.
-    BeginBlock { build: Arc<BuildProviderFn>, caches: ExecutionCache },
+    BeginBlock {
+        build: Arc<BuildProviderFn>,
+        caches: ExecutionCache,
+        txpool_snapshot: Option<TxPoolPrewarmCacheSnapshot>,
+    },
     /// Warm one target into the held provider's cache. Ignored if no provider is held.
     Warm(PrewarmTarget),
     /// Drop the held provider (and its read txn).
@@ -66,10 +70,18 @@ impl BalPrewarmPool {
 
     /// Begins a block: hands every worker the provider builder and shared cache so each opens its
     /// own read txn over the parent state. Pair with [`end_block`](Self::end_block).
-    pub(crate) fn begin_block(&self, build: Arc<BuildProviderFn>, caches: ExecutionCache) {
+    pub(crate) fn begin_block(
+        &self,
+        build: Arc<BuildProviderFn>,
+        caches: ExecutionCache,
+        txpool_snapshot: Option<TxPoolPrewarmCacheSnapshot>,
+    ) {
         for worker in &self.workers {
-            let _ = worker
-                .send(PrewarmMsg::BeginBlock { build: build.clone(), caches: caches.clone() });
+            let _ = worker.send(PrewarmMsg::BeginBlock {
+                build: build.clone(),
+                caches: caches.clone(),
+                txpool_snapshot: txpool_snapshot.clone(),
+            });
         }
     }
 
@@ -135,9 +147,12 @@ fn prewarm_loop(rx: crossbeam_channel::Receiver<PrewarmMsg>) {
     // Blocks when idle; the channel disconnects (and the loop ends) when the pool is dropped.
     while let Ok(msg) = rx.recv() {
         match msg {
-            PrewarmMsg::BeginBlock { build, caches } => {
+            PrewarmMsg::BeginBlock { build, caches, txpool_snapshot } => {
                 provider = match (build)() {
-                    Ok(inner) => Some(CachedStateProvider::new_prewarm(inner, caches)),
+                    Ok(inner) => Some(
+                        CachedStateProvider::new_prewarm(inner, caches)
+                            .with_txpool_snapshot(txpool_snapshot),
+                    ),
                     Err(err) => {
                         trace!(target: "engine::tree::bal_prewarm_pool", %err, "failed to build provider");
                         None
diff --git a/crates/engine/tree/src/tree/payload_processor/prewarm.rs b/crates/engine/tree/src/tree/payload_processor/prewarm.rs
index 8feda8e1bec..502dd0e0184 100644
--- a/crates/engine/tree/src/tree/payload_processor/prewarm.rs
+++ b/crates/engine/tree/src/tree/payload_processor/prewarm.rs
@@ -413,7 +413,7 @@ where
             let provider_builder = ctx.provider.clone();
             let build = Arc::new(move || provider_builder.build());
 
-            pool.begin_block(build, caches);
+            pool.begin_block(build, caches, ctx.env.txpool_snapshot.clone());
             for account in prefetch_bal.as_bal() {
                 pool.warm_account(account.address);
                 for change in &account.storage_changes {
@@ -582,7 +582,10 @@ where
         // Use the caches to create a new provider with caching
         if let Some(saved_cache) = &self.saved_cache {
             let caches = saved_cache.cache().clone();
-            state_provider = Box::new(CachedStateProvider::new_prewarm(state_provider, caches));
+            state_provider = Box::new(
+                CachedStateProvider::new_prewarm(state_provider, caches)
+                    .with_txpool_snapshot(self.env.txpool_snapshot.clone()),
+            );
         }
 
         let state_provider = StateProviderDatabase::new(state_provider);
@@ -699,7 +702,10 @@ where
                     match (self.disable_bal_batch_io, &self.saved_cache) {
                         (false, Some(saved)) => {
                             let caches = saved.cache().clone();
-                            Box::new(CachedStateProvider::new_prewarm(inner, caches))
+                            Box::new(
+                                CachedStateProvider::new_prewarm(inner, caches)
+                                    .with_txpool_snapshot(self.env.txpool_snapshot.clone()),
+                            )
                         }
                         _ => Box::new(inner),
                     };
diff --git a/crates/engine/tree/src/tree/payload_validator.rs b/crates/engine/tree/src/tree/payload_validator.rs
index 01c5d4c4282..4b524c60fa5 100644
--- a/crates/engine/tree/src/tree/payload_validator.rs
+++ b/crates/engine/tree/src/tree/payload_validator.rs
@@ -100,6 +100,7 @@ use crate::tree::{
     instrumented_state::{InstrumentedStateProvider, StateProviderMetrics, StateProviderStats},
     payload_processor::PayloadProcessor,
     precompile_cache::{CachedPrecompile, CachedPrecompileMetrics, PrecompileCacheMap},
+    txpool_prewarm,
     types::{InsertPayloadResult, ValidationOutput},
     CacheWaitDurations, CachedStateProvider, EngineApiMetrics, EngineApiTreeState, ExecutionEnv,
     PayloadHandle, StateProviderBuilder, StateProviderDatabase, TreeConfig, WaitForCaches,
@@ -295,6 +296,11 @@ where
     /// State-root strategy used to prepare per-block commitment tasks.
     #[debug(skip)]
     state_root_strategy: Arc<dyn StateRootStrategy<Evm::Primitives, P, Evm>>,
+    /// Persistent txpool prewarming worker and its latest immutable snapshot.
+    ///
+    /// None if txpool prewarming is disabled.
+    #[debug(skip)]
+    txpool_prewarm: Option<txpool_prewarm::Handle<Evm::Primitives, P, Evm>>,
 }
 
 impl<N, P, Evm, V> BasicEngineValidator<P, Evm, V>
@@ -315,6 +321,8 @@ where
         + StateReader
         + HashedPostStateProvider
         + Clone
+        + Send
+        + Sync
         + 'static,
     OverlayStateProviderFactory<P, N>: DatabaseProviderROFactory<Provider: TrieCursorFactory + HashedCursorFactory>
         + Clone
@@ -358,6 +366,7 @@ where
             runtime,
             state_trie_overlays,
             state_root_strategy: Arc::new(DefaultStateRootStrategy::default()),
+            txpool_prewarm: None,
         }
     }
 
@@ -370,6 +379,19 @@ where
         self
     }
 
+    /// Installs the txpool source and starts the persistent cache-prewarming worker.
+    pub fn with_txpool_prewarming(
+        mut self,
+        source: impl crate::tree::TxPoolPrewarmSource<N> + 'static,
+    ) -> Self {
+        self.txpool_prewarm = Some(txpool_prewarm::Handle::spawn(
+            &self.runtime,
+            Arc::new(source),
+            self.evm_config.clone(),
+        ));
+        self
+    }
+
     /// Converts a [`BlockOrPayload`] to a recovered block.
     #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
     pub fn convert_to_block<T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>>(
@@ -466,6 +488,9 @@ where
         Evm: ConfigureEngineEvm<T::ExecutionData, Primitives = N>,
     {
         let parent_hash = input.parent_hash();
+        let _txpool_pause = self.txpool_prewarm.as_ref().map(txpool_prewarm::Handle::pause);
+        let txpool_snapshot =
+            self.txpool_prewarm.as_ref().and_then(|prewarmer| prewarmer.snapshot(parent_hash));
         let _jit_pause = JitPauseGuard::new(&self.evm_config);
 
         // Fetch parent block. This goes to memory most of the time unless the parent block is
@@ -571,6 +596,7 @@ where
             gas_used: input.gas_used(),
             withdrawals: input.withdrawals().map(|w| w.to_vec()),
             decoded_bal: decoded_bal.as_ref().map(Arc::clone),
+            txpool_snapshot: txpool_snapshot.clone(),
         };
 
         // Get an iterator over the transactions in the payload
@@ -662,13 +688,16 @@ where
                 } else {
                     CacheFillMode::LookupOnly
                 };
-                Box::new(CachedStateProvider::new_with_mode(
-                    provider,
-                    caches.clone(),
-                    fill_mode,
-                    cache_metrics.clone(),
-                    cache_stats.clone(),
-                )) as StateProviderBox
+                Box::new(
+                    CachedStateProvider::new_with_mode(
+                        provider,
+                        caches.clone(),
+                        fill_mode,
+                        cache_metrics.clone(),
+                        cache_stats.clone(),
+                    )
+                    .with_txpool_snapshot(txpool_snapshot.clone()),
+                ) as StateProviderBox
             } else {
                 provider
             };
@@ -1650,6 +1679,18 @@ where
             .unwrap_or_default();
         let (code_cache_hits, code_cache_misses) =
             cache_stats.as_ref().map(|s| (s.code_hits(), s.code_misses())).unwrap_or_default();
+        let (txpool_snapshot_account_hits, txpool_snapshot_account_misses) = cache_stats
+            .as_ref()
+            .map(|s| (s.txpool_snapshot_account_hits(), s.txpool_snapshot_account_misses()))
+            .unwrap_or_default();
+        let (txpool_snapshot_storage_hits, txpool_snapshot_storage_misses) = cache_stats
+            .as_ref()
+            .map(|s| (s.txpool_snapshot_storage_hits(), s.txpool_snapshot_storage_misses()))
+            .unwrap_or_default();
+        let (txpool_snapshot_code_hits, txpool_snapshot_code_misses) = cache_stats
+            .as_ref()
+            .map(|s| (s.txpool_snapshot_code_hits(), s.txpool_snapshot_code_misses()))
+            .unwrap_or_default();
 
         // Build execution timing stats for detailed block logging
         Box::new(ExecutionTimingStats {
@@ -1678,6 +1719,12 @@ where
             storage_cache_misses,
             code_cache_hits,
             code_cache_misses,
+            txpool_snapshot_account_hits,
+            txpool_snapshot_account_misses,
+            txpool_snapshot_storage_hits,
+            txpool_snapshot_storage_misses,
+            txpool_snapshot_code_hits,
+            txpool_snapshot_code_misses,
         })
     }
 }
@@ -1741,6 +1788,11 @@ pub trait EngineValidator<
         block: BuiltPayloadExecutedBlock<N>,
     ) -> ProviderResult<ExecutedBlock<N>>;
 
+    /// Notifies the validator that `hash` is the current canonical head.
+    ///
+    /// This may also be called when a forkchoice update reaffirms the existing head.
+    fn on_canonical_head_changed(&self, _hash: B256, _state: &EngineApiTreeState<N>) {}
+
     /// Prepares the resources loaned to a payload builder job.
     ///
     /// `timestamp` is taken from the payload attributes.
@@ -1830,6 +1882,56 @@ where
         ))
     }
 
+    fn on_canonical_head_changed(&self, hash: B256, state: &EngineApiTreeState<N>) {
+        let Some(txpool_prewarm) = self.txpool_prewarm.as_ref() else { return };
+
+        // Obtain the header of the new canonical head; pool transactions are warmed on top of
+        // its state.
+        let parent = match self.sealed_header_by_hash(hash, state) {
+            Ok(Some(header)) => header,
+            Ok(None) => return,
+            Err(err) => {
+                trace!(
+                    target: "engine::tree::txpool_prewarm",
+                    %err,
+                    block_hash = ?hash,
+                    "failed to fetch canonical header for txpool prewarming"
+                );
+                return
+            }
+        };
+        // Warming reuses the head block's own environment rather than predicting the next
+        // block's: attribute-level accuracy (timestamp, basefee) doesn't matter for collecting
+        // state reads, and the worker disables the fee checks a stale basefee would trip.
+        let evm_env = match self.evm_config.evm_env(parent.header()) {
+            Ok(evm_env) => evm_env,
+            Err(err) => {
+                trace!(
+                    target: "engine::tree::txpool_prewarm",
+                    %err,
+                    block_hash = ?parent.hash(),
+                    "failed to derive canonical txpool prewarming environment"
+                );
+                return
+            }
+        };
+
+        let provider_builder = match self.state_provider_builder(parent.hash(), state) {
+            Ok(Some(provider_builder)) => provider_builder,
+            Ok(None) => return,
+            Err(err) => {
+                trace!(
+                    target: "engine::tree::txpool_prewarm",
+                    %err,
+                    block_hash = ?parent.hash(),
+                    "failed to derive canonical txpool prewarming provider"
+                );
+                return
+            }
+        };
+        txpool_prewarm.start(parent.hash(), evm_env, provider_builder)
+    }
+
     fn payload_builder_resources(
         &self,
         parent_hash: B256,
@@ -1843,9 +1945,16 @@ where
             .then(|| self.payload_processor.cache_for(parent_hash));
         let state_root_handle =
             self.payload_state_root_handle_for(parent_hash, parent_header, timestamp, state);
-
-        PayloadBuilderResources::new(execution_cache, state_root_handle)
-            .with_lease(PayloadBuilderLease::new(JitPauseGuard::new(&self.evm_config)))
+        let mut resources = PayloadBuilderResources::new(execution_cache, state_root_handle)
+            .with_lease(PayloadBuilderLease::new(JitPauseGuard::new(&self.evm_config)));
+        // If the txpool prewarming is enabled then we should disable it for the duration
+        // of the payload builder job. This is done by obtaining a lease that will release
+        // the txpool prewarm when dropped.
+        if let Some(txpool_prewarm) = self.txpool_prewarm.as_ref() {
+            let txpool_lease = PayloadBuilderLease::new(txpool_prewarm.pause());
+            resources = resources.with_lease(txpool_lease);
+        }
+        resources
     }
 }
 
diff --git a/crates/engine/tree/src/tree/txpool_prewarm/control.rs b/crates/engine/tree/src/tree/txpool_prewarm/control.rs
new file mode 100644
index 00000000000..da71c87c97c
--- /dev/null
+++ b/crates/engine/tree/src/tree/txpool_prewarm/control.rs
@@ -0,0 +1,170 @@
+use crate::tree::TxPoolPrewarmCacheSnapshot as Snapshot;
+use alloy_primitives::B256;
+use crossbeam_channel::{unbounded, Receiver, Sender};
+use parking_lot::RwLock;
+use std::{
+    fmt::Debug,
+    sync::{Arc, Weak},
+};
+
+/// Sending side of the txpool prewarming worker's control channel.
+pub(super) struct Control<J> {
+    commands: Sender<Command<J>>,
+    publication: Publication,
+}
+
+impl<J> Debug for Control<J> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Control")
+            .field(
+                "published",
+                &self.publication.read().as_ref().map(|snapshot| snapshot.parent_hash()),
+            )
+            .finish_non_exhaustive()
+    }
+}
+
+impl<J> Control<J> {
+    pub(super) fn new() -> (Arc<Self>, Receiver<Command<J>>) {
+        let (commands, receiver) = unbounded();
+        let publication = Arc::new(RwLock::new(None));
+        (Arc::new(Self { commands, publication }), receiver)
+    }
+
+    pub(super) fn publication(&self) -> Publication {
+        Arc::clone(&self.publication)
+    }
+
+    pub(super) fn start(&self, parent_hash: B256, job: J) {
+        let _ = self.commands.send(Command::Start { parent_hash, job });
+    }
+
+    pub(super) fn pause(self: &Arc<Self>) -> PauseGuard<J> {
+        // Fire-and-forget: the worker never interrupts a transaction mid-execution, so waiting
+        // for it to observe the pause would only add that execution time to the caller's
+        // latency without reducing the overlap.
+        let _ = self.commands.send(Command::Pause);
+        PauseGuard { control: Arc::downgrade(self) }
+    }
+
+    pub(super) fn snapshot(&self, parent_hash: B256) -> Option<Snapshot> {
+        self.publication
+            .read()
+            .as_ref()
+            .filter(|snapshot| snapshot.parent_hash() == parent_hash)
+            .cloned()
+    }
+}
+
+/// Immutable snapshot publication shared by the handle and worker.
+pub(super) type Publication = Arc<RwLock<Option<Snapshot>>>;
+
+/// Commands sent to the txpool prewarming worker.
+pub(super) enum Command<J> {
+    /// Starts prewarming for a canonical head, replacing any previous job.
+    Start { parent_hash: B256, job: J },
+    /// Pauses prewarming until a matching [`Resume`](Self::Resume).
+    Pause,
+    /// Releases one active pause.
+    Resume,
+}
+
+/// One outstanding pause, released when dropped.
+pub(super) struct PauseGuard<J> {
+    control: Weak<Control<J>>,
+}
+
+impl<J> Drop for PauseGuard<J> {
+    fn drop(&mut self) {
+        // If the pause was never delivered the channel is disconnected, and this send fails
+        // just the same, so the worker can never see an unmatched resume.
+        if let Some(control) = self.control.upgrade() {
+            let _ = control.commands.send(Command::Resume);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    type TestControl = Control<u64>;
+
+    fn control() -> (Arc<TestControl>, Receiver<Command<u64>>) {
+        TestControl::new()
+    }
+
+    fn snapshot(parent_hash: B256) -> Snapshot {
+        Snapshot::new(parent_hash, Default::default())
+    }
+
+    #[test]
+    fn snapshot_is_published_for_matching_parent() {
+        let (control, _receiver) = control();
+        let parent_hash = B256::repeat_byte(0x01);
+        *control.publication.write() = Some(snapshot(parent_hash));
+
+        assert_eq!(
+            control.snapshot(parent_hash).map(|snapshot| snapshot.parent_hash()),
+            Some(parent_hash)
+        );
+        assert!(control.snapshot(B256::ZERO).is_none());
+    }
+
+    #[test]
+    fn start_sends_job() {
+        let (control, receiver) = control();
+        let parent_hash = B256::repeat_byte(0x01);
+        control.start(parent_hash, 1);
+
+        assert!(matches!(
+            receiver.try_recv(),
+            Ok(Command::Start { parent_hash: received_parent, job: 1 })
+                if received_parent == parent_hash
+        ));
+    }
+
+    #[test]
+    fn pause_queues_without_waiting_and_resumes_on_drop() {
+        let (control, receiver) = control();
+        // Nothing reads the channel, so this returning at all proves pause does not block.
+        let guard = control.pause();
+        assert!(matches!(receiver.try_recv(), Ok(Command::Pause)));
+
+        drop(guard);
+        assert!(matches!(receiver.try_recv(), Ok(Command::Resume)));
+    }
+
+    #[test]
+    fn pause_guard_does_not_retain_control() {
+        let (control, _receiver) = control();
+        let weak_control = Arc::downgrade(&control);
+        let guard = control.pause();
+
+        drop(control);
+        assert!(weak_control.upgrade().is_none());
+        drop(guard);
+    }
+
+    #[test]
+    fn overlapping_pause_guards_send_matching_resumes() {
+        let (control, receiver) = control();
+        let first = control.pause();
+        let second = control.pause();
+        assert!(matches!(receiver.try_recv(), Ok(Command::Pause)));
+        assert!(matches!(receiver.try_recv(), Ok(Command::Pause)));
+
+        drop(first);
+        assert!(matches!(receiver.try_recv(), Ok(Command::Resume)));
+        drop(second);
+        assert!(matches!(receiver.try_recv(), Ok(Command::Resume)));
+    }
+
+    #[test]
+    fn dropping_control_disconnects_worker() {
+        let (control, receiver) = control();
+        drop(control);
+
+        assert!(receiver.recv().is_err());
+    }
+}
diff --git a/crates/engine/tree/src/tree/txpool_prewarm/mod.rs b/crates/engine/tree/src/tree/txpool_prewarm/mod.rs
new file mode 100644
index 00000000000..3070148fd3d
--- /dev/null
+++ b/crates/engine/tree/src/tree/txpool_prewarm/mod.rs
@@ -0,0 +1,115 @@
+//! Txpool-driven state prewarming and immutable snapshot publication.
+
+mod control;
+mod worker;
+
+use self::control::Control;
+use crate::tree::{StateProviderBuilder, TxPoolPrewarmCacheSnapshot};
+use alloy_consensus::transaction::Recovered;
+use alloy_primitives::{Address, B256};
+use reth_evm::{ConfigureEvm, EvmEnvFor};
+use reth_primitives_traits::{NodePrimitives, TxTy};
+use reth_provider::{BlockReader, StateProviderFactory, StateReader};
+use std::{fmt::Debug, sync::Arc};
+
+/// Coordinates a long-lived worker and the latest completed immutable snapshot.
+pub(crate) struct Handle<N, P, Evm>
+where
+    N: NodePrimitives,
+    Evm: ConfigureEvm<Primitives = N>,
+{
+    control: Arc<Control<Job<N, P, Evm>>>,
+}
+
+impl<N, P, Evm> Debug for Handle<N, P, Evm>
+where
+    N: NodePrimitives,
+    Evm: ConfigureEvm<Primitives = N>,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Handle").field("control", &self.control).finish()
+    }
+}
+
+impl<N, P, Evm> Handle<N, P, Evm>
+where
+    N: NodePrimitives,
+    P: BlockReader + StateProviderFactory + StateReader + Clone + Send + Sync + 'static,
+    Evm: ConfigureEvm<Primitives = N> + 'static,
+{
+    /// Spawns the long-lived worker, which owns its mutable read cache and starts a fresh one for
+    /// each new head.
+    pub(crate) fn spawn(
+        runtime: &reth_tasks::Runtime,
+        source: Arc<dyn Source<N>>,
+        evm_config: Evm,
+    ) -> Self {
+        let (control, commands) = Control::new();
+        let publication = control.publication();
+        runtime.spawn_critical_os_thread("txpool-prewarm", "txpool prewarm worker", async move {
+            worker::Worker::new(commands, publication, source, evm_config).run()
+        });
+        Self { control }
+    }
+
+    /// Pauses speculative work.
+    ///
+    /// Returns a guard that will resume the worker when dropped. There could be multiple
+    /// outstanding guards, in which case the worker will not resume until all guards are dropped.
+    ///
+    /// Pausing is asynchronous and never blocks the caller: the worker observes it between
+    /// transactions, so speculative work may overlap the guard's scope by at most one
+    /// transaction.
+    pub(crate) fn pause(&self) -> impl Drop + Send + 'static {
+        self.control.pause()
+    }
+
+    /// Returns the latest fully published snapshot for `parent_hash`, or `None` if no snapshot is
+    /// available for that hash.
+    pub(crate) fn snapshot(&self, parent_hash: B256) -> Option<TxPoolPrewarmCacheSnapshot> {
+        self.control.snapshot(parent_hash)
+    }
+
+    /// Starts continuous warming for the latest canonical head.
+    pub(crate) fn start(
+        &self,
+        parent_hash: B256,
+        evm_env: EvmEnvFor<Evm>,
+        provider_builder: StateProviderBuilder<N, P>,
+    ) {
+        self.control.start(parent_hash, Job { evm_env, provider_builder });
+    }
+}
+
+/// A live, forward-only view of the pool's best transactions for one canonical parent.
+///
+/// Returning [`None`](Iterator::next) only means no transaction is currently ready. The same
+/// iterator can yield transactions that become pending later.
+pub type Transactions<N> = Box<dyn Iterator<Item = Transaction<N>> + Send>;
+
+/// A transaction selected from the txpool for cache-only prewarming.
+#[derive(Debug, Clone)]
+pub struct Transaction<N: NodePrimitives> {
+    /// Transaction hash.
+    pub hash: B256,
+    /// Recovered sender.
+    pub sender: Address,
+    /// Recovered consensus transaction.
+    pub transaction: Recovered<TxTy<N>>,
+}
+
+/// Source of txpool transactions for best-effort cache prewarming.
+pub trait Source<N: NodePrimitives>: Send + Sync + Debug {
+    /// Opens a live best-transactions iterator for `parent_hash`.
+    ///
+    /// The worker opens this once per canonical parent and retains it across empty polls, snapshot
+    /// publications, and validation pauses. Sources should return [`None`] if they are not yet
+    /// tracking `parent_hash`.
+    fn best_transactions(&self, parent_hash: B256) -> Option<Transactions<N>>;
+}
+
+/// A request to warm txpool transactions against one fully validated parent state.
+struct Job<N: NodePrimitives, P, Evm: ConfigureEvm<Primitives = N>> {
+    evm_env: EvmEnvFor<Evm>,
+    provider_builder: StateProviderBuilder<N, P>,
+}
diff --git a/crates/engine/tree/src/tree/txpool_prewarm/worker.rs b/crates/engine/tree/src/tree/txpool_prewarm/worker.rs
new file mode 100644
index 00000000000..54ecd1ffb74
--- /dev/null
+++ b/crates/engine/tree/src/tree/txpool_prewarm/worker.rs
@@ -0,0 +1,582 @@
+use super::{
+    control::{Command, Publication},
+    Job, Source, Transactions,
+};
+use crate::tree::{StateProviderDatabase, TxPoolPrewarmCacheSnapshot as Snapshot};
+use alloy_evm::Evm;
+use alloy_primitives::B256;
+use crossbeam_channel::{Receiver, RecvTimeoutError, TryRecvError};
+use reth_evm::ConfigureEvm;
+use reth_primitives_traits::NodePrimitives;
+use reth_provider::{BlockReader, StateProviderFactory, StateReader};
+use reth_revm::{cached::CachedReads, db::State};
+use std::{
+    sync::Arc,
+    time::{Duration, Instant},
+};
+use tracing::{debug, trace};
+
+/// Maximum interval between snapshot publications and delay when no transaction is ready.
+const REFRESH_INTERVAL: Duration = Duration::from_millis(100);
+
+/// Delay while waiting for pool maintenance to advance to the state being warmed.
+const HEAD_POLL_INTERVAL: Duration = Duration::from_millis(10);
+
+/// The txpool prewarming worker.
+///
+/// A long-lived loop that speculatively executes the pool's best transactions on top of the
+/// current canonical state, recording every state read in a [`CachedReads`]. Roughly every
+/// [`REFRESH_INTERVAL`] it publishes an immutable snapshot of that cache, which block validation
+/// and payload building use to seed their own caches.
+///
+/// The worker is driven by [`Command`]s: `Start` points it at a new parent state, and
+/// `Pause`/`Resume` bracket cache-sensitive work elsewhere. Commands are only applied between
+/// batches, never while an EVM or state provider is alive.
+pub(super) struct Worker<N, P, Evm>
+where
+    N: NodePrimitives,
+    Evm: ConfigureEvm<Primitives = N>,
+{
+    /// Control commands from the [`Handle`](super::Handle).
+    commands: Receiver<Command<Job<N, P, Evm>>>,
+    /// Shared slot the latest snapshot is published into.
+    publication: Publication,
+    /// The txpool view transactions are drawn from.
+    source: Arc<dyn Source<N>>,
+    /// Configures the EVM used for speculative execution.
+    evm_config: Evm,
+    /// The parent state to warm, from the most recent `Start` command.
+    job: Option<(B256, Job<N, P, Evm>)>,
+    /// Outstanding pauses; the worker only warms while this is zero.
+    pauses: u64,
+    /// Read-through cache filled by execution; replaced whenever the warmed parent changes.
+    cache: CachedReads,
+    /// Parent whose state the `cache` reads were collected against.
+    cache_parent: Option<B256>,
+    /// Cache entry counts as of the last publication. The cache only ever grows, so a change
+    /// means it holds unpublished reads.
+    published_entries: (usize, usize, usize),
+    /// Live best-transactions iterator, tagged with the parent it was opened for.
+    transactions: Option<(B256, Transactions<N>)>,
+}
+
+impl<N, P, Evm> Worker<N, P, Evm>
+where
+    N: NodePrimitives,
+    P: BlockReader + StateProviderFactory + StateReader + Clone,
+    Evm: ConfigureEvm<Primitives = N>,
+{
+    pub(super) fn new(
+        commands: Receiver<Command<Job<N, P, Evm>>>,
+        publication: Publication,
+        source: Arc<dyn Source<N>>,
+        evm_config: Evm,
+    ) -> Self {
+        Self {
+            commands,
+            publication,
+            source,
+            evm_config,
+            job: None,
+            pauses: 0,
+            cache: CachedReads::default(),
+            cache_parent: None,
+            published_entries: (0, 0, 0),
+            transactions: None,
+        }
+    }
+
+    /// Runs until the control side is dropped, which is the worker's shutdown signal.
+    pub(super) fn run(mut self) {
+        let _ = self.run_until_disconnected();
+    }
+
+    fn run_until_disconnected(&mut self) -> Result<(), ChannelDisconnected> {
+        loop {
+            let parent_hash = self.wait_until_runnable()?;
+
+            // The pool tracks canonical heads on its own schedule; check back shortly if it is
+            // not tracking this parent yet.
+            if !self.open_transactions(parent_hash) {
+                self.idle(HEAD_POLL_INTERVAL)?;
+                continue
+            }
+
+            if self.cache_parent != Some(parent_hash) {
+                self.cache = CachedReads::default();
+                self.cache_parent = Some(parent_hash);
+                self.published_entries = (0, 0, 0);
+                debug!(
+                    target: "engine::tree::txpool_prewarm",
+                    ?parent_hash,
+                    "started txpool prewarming"
+                );
+            }
+
+            let batch = self.warm_one_batch();
+
+            // A pending command may pause the worker or point it at a new parent: apply it (at
+            // the top of the loop) before spending time on publication.
+            if !self.commands.is_empty() {
+                continue
+            }
+            self.publish_snapshot_if_dirty();
+            if batch == BatchEnd::Rest {
+                self.idle(REFRESH_INTERVAL)?;
+            }
+        }
+    }
+
+    /// Blocks until the worker holds a job and no pauses are outstanding, applying every command
+    /// that arrives in the meantime. Returns the parent hash to warm.
+    fn wait_until_runnable(&mut self) -> Result<B256, ChannelDisconnected> {
+        loop {
+            self.apply_pending_commands()?;
+
+            if self.pauses == 0 &&
+                let Some((parent_hash, _)) = self.job.as_ref()
+            {
+                return Ok(*parent_hash)
+            }
+
+            let command = self.commands.recv().map_err(|_| ChannelDisconnected)?;
+            self.apply(command);
+        }
+    }
+
+    /// Ensures the transaction iterator matches `parent_hash`, opening a fresh one when the head
+    /// changed. Returns `false` while the pool is not yet tracking that parent.
+    fn open_transactions(&mut self, parent_hash: B256) -> bool {
+        if self.transactions.as_ref().is_none_or(|(parent, _)| *parent != parent_hash) {
+            // Release the stale iterator before asking the pool for a new one.
+            self.transactions = None;
+            self.transactions = self
+                .source
+                .best_transactions(parent_hash)
+                .map(|transactions| (parent_hash, transactions));
+        }
+        self.transactions.is_some()
+    }
+
+    /// Speculatively executes pool transactions against the parent state for at most
+    /// [`REFRESH_INTERVAL`], filling the cache with every state read.
+    ///
+    /// Stops early once the pool has no transaction ready or a command arrives. Commands are
+    /// never consumed here: a pending command merely ends the batch and is applied by the main
+    /// loop after the EVM and state provider built here are dropped.
+    fn warm_one_batch(&mut self) -> BatchEnd {
+        let (_, job) = self.job.as_ref().expect("wait_until_runnable installed a job");
+        let (parent_hash, transactions) =
+            self.transactions.as_mut().expect("open_transactions installed an iterator");
+
+        // Building a state provider opens a database transaction; don't bother under a pending
+        // command.
+        if !self.commands.is_empty() {
+            return BatchEnd::GoAgain
+        }
+
+        let state_provider = match job.provider_builder.build() {
+            Ok(provider) => provider,
+            Err(err) => {
+                trace!(
+                    target: "engine::tree::txpool_prewarm",
+                    %err,
+                    ?parent_hash,
+                    "failed to build txpool prewarming state provider"
+                );
+                return BatchEnd::Rest
+            }
+        };
+        let mut state = State::builder()
+            .with_database(self.cache.as_db_mut(StateProviderDatabase::new(state_provider)))
+            .build();
+        // The environment is the head block's own, not a predicted next-block one, and execution
+        // is out of context by design: transaction viability is the pool's business, so nonce,
+        // balance and (one-block-stale) basefee checks must not gate which state gets warmed.
+        let mut evm_env = job.evm_env.clone();
+        evm_env.cfg_env.disable_nonce_check = true;
+        evm_env.cfg_env.disable_balance_check = true;
+        evm_env.cfg_env.disable_base_fee = true;
+        let mut evm = self.evm_config.evm_with_env(&mut state, evm_env);
+
+        let deadline = Instant::now() + REFRESH_INTERVAL;
+        while self.commands.is_empty() && Instant::now() < deadline {
+            let Some(transaction) = transactions.next() else { return BatchEnd::Rest };
+            if let Err(err) = evm.transact(transaction.transaction) {
+                trace!(
+                    target: "engine::tree::txpool_prewarm",
+                    %err,
+                    tx_hash = ?transaction.hash,
+                    sender = %transaction.sender,
+                    "speculative txpool transaction execution failed"
+                );
+            }
+        }
+        BatchEnd::GoAgain
+    }
+
+    /// Publishes a fresh snapshot if the cache gained reads since the last publication.
+    fn publish_snapshot_if_dirty(&mut self) {
+        let entries = entry_counts(&self.cache);
+        if entries == self.published_entries {
+            return
+        }
+
+        let parent_hash = self.cache_parent.expect("reads only accumulate after a cache reset");
+        *self.publication.write() = Some(Snapshot::new(parent_hash, Arc::new(self.cache.clone())));
+        self.published_entries = entries;
+        let (accounts, storage, bytecodes) = entries;
+        debug!(
+            target: "engine::tree::txpool_prewarm",
+            ?parent_hash,
+            accounts,
+            storage,
+            bytecodes,
+            "published txpool prewarming snapshot"
+        );
+    }
+
+    /// Rests until a command arrives (applying it) or `timeout` elapses, whichever comes first.
+    fn idle(&mut self, timeout: Duration) -> Result<(), ChannelDisconnected> {
+        match self.commands.recv_timeout(timeout) {
+            Ok(command) => {
+                self.apply(command);
+                Ok(())
+            }
+            Err(RecvTimeoutError::Timeout) => Ok(()),
+            Err(RecvTimeoutError::Disconnected) => Err(ChannelDisconnected),
+        }
+    }
+
+    /// Applies every command already sitting in the channel, without blocking.
+    fn apply_pending_commands(&mut self) -> Result<(), ChannelDisconnected> {
+        loop {
+            match self.commands.try_recv() {
+                Ok(command) => self.apply(command),
+                Err(TryRecvError::Empty) => return Ok(()),
+                Err(TryRecvError::Disconnected) => return Err(ChannelDisconnected),
+            }
+        }
+    }
+
+    /// Applies a single command.
+    ///
+    /// Only called while no EVM or state provider is alive, so a paused worker holds no
+    /// execution resources.
+    fn apply(&mut self, command: Command<Job<N, P, Evm>>) {
+        match command {
+            Command::Start { parent_hash, job } => self.job = Some((parent_hash, job)),
+            Command::Pause => {
+                self.pauses =
+                    self.pauses.checked_add(1).expect("txpool prewarm pause count overflow");
+            }
+            Command::Resume => {
+                self.pauses = self
+                    .pauses
+                    .checked_sub(1)
+                    .expect("txpool prewarm resumed without a matching pause");
+            }
+        }
+    }
+}
+
+/// What to do after a warming batch.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum BatchEnd {
+    /// Batch again right away: the refresh deadline passed or a command interrupted the batch
+    /// while transactions may still be ready.
+    GoAgain,
+    /// Idle until something changes: the pool had no transaction ready, or no state provider
+    /// could be built.
+    Rest,
+}
+
+/// The control channel closed: every sender is dropped and the worker shuts down.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct ChannelDisconnected;
+
+/// Returns `(accounts, storage slots, bytecodes)` cached in `reads`.
+fn entry_counts(reads: &CachedReads) -> (usize, usize, usize) {
+    (
+        reads.accounts.len(),
+        reads.accounts.values().map(|account| account.storage.len()).sum(),
+        reads.contracts.len(),
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{super::Transaction as PoolTransaction, *};
+    use crate::tree::StateProviderBuilder;
+    use alloy_consensus::{transaction::Recovered, Signed, TxLegacy};
+    use alloy_primitives::{Address, Signature, TxKind, U256};
+    use crossbeam_channel::{unbounded, Sender};
+    use parking_lot::{Mutex, RwLock};
+    use reth_ethereum_primitives::{EthPrimitives, TransactionSigned};
+    use reth_evm_ethereum::EthEvmConfig;
+    use reth_provider::test_utils::MockEthProvider;
+    use std::{
+        collections::{HashMap, VecDeque},
+        sync::atomic::{AtomicUsize, Ordering},
+        thread::{self, JoinHandle},
+    };
+
+    /// Upper bound on any single wait; failures surface as panics well before CI timeouts.
+    const WAIT_LIMIT: Duration = Duration::from_secs(5);
+    const POLL_INTERVAL: Duration = Duration::from_millis(2);
+
+    type TestJob = Job<EthPrimitives, MockEthProvider, EthEvmConfig>;
+
+    /// Drives a live worker thread through its public seams only: commands in, the publication
+    /// slot and the scripted pool out.
+    struct Harness {
+        commands: Sender<Command<TestJob>>,
+        publication: Publication,
+        pool: Arc<ScriptedPool>,
+        worker: Option<JoinHandle<()>>,
+    }
+
+    impl Harness {
+        fn spawn() -> Self {
+            let (commands, receiver) = unbounded();
+            let publication: Publication = Arc::new(RwLock::new(None));
+            let pool = Arc::new(ScriptedPool::default());
+            let worker = thread::spawn({
+                let publication = Arc::clone(&publication);
+                let source: Arc<dyn Source<EthPrimitives>> = pool.clone();
+                move || Worker::new(receiver, publication, source, EthEvmConfig::mainnet()).run()
+            });
+            Self { commands, publication, pool, worker: Some(worker) }
+        }
+
+        /// Drops the control channel and waits for the worker thread to exit.
+        fn shutdown(mut self) {
+            let (disconnected, _) = unbounded();
+            self.commands = disconnected;
+            let worker = self.worker.take().expect("worker already joined");
+            wait_until("the worker thread exits", || worker.is_finished());
+            worker.join().unwrap();
+        }
+
+        /// Points the worker at `parent_hash`, as [`Handle::start`](super::super::Handle) does.
+        fn start(&self, parent_hash: B256) {
+            let job = Job {
+                evm_env: Default::default(),
+                provider_builder: StateProviderBuilder::new(
+                    MockEthProvider::default(),
+                    parent_hash,
+                    None,
+                ),
+            };
+            self.commands.send(Command::Start { parent_hash, job }).unwrap();
+        }
+
+        /// Pauses the worker, as [`Handle::pause`](super::super::Handle) does. Fire-and-forget
+        /// is still deterministic: the worker never publishes work it performs after the pause
+        /// is queued until [`Self::resume`], because publication is skipped while a command is
+        /// pending and a consumed pause blocks the loop.
+        fn pause(&self) {
+            self.commands.send(Command::Pause).unwrap();
+        }
+
+        fn resume(&self) {
+            self.commands.send(Command::Resume).unwrap();
+        }
+
+        /// Blocks until a published snapshot satisfies `accept`.
+        fn published(&self, accept: impl Fn(&Snapshot) -> bool) -> Snapshot {
+            let deadline = Instant::now() + WAIT_LIMIT;
+            loop {
+                let snapshot = self.publication.read().as_ref().cloned();
+                if let Some(snapshot) = snapshot &&
+                    accept(&snapshot)
+                {
+                    return snapshot
+                }
+                assert!(Instant::now() < deadline, "timed out waiting for a matching snapshot");
+                thread::sleep(POLL_INTERVAL);
+            }
+        }
+
+        fn published_for(&self, parent_hash: B256) -> Snapshot {
+            self.published(|snapshot| snapshot.parent_hash() == parent_hash)
+        }
+
+        fn published_entry_counts(&self) -> Option<(usize, usize, usize)> {
+            self.publication.read().as_ref().map(|snapshot| snapshot.entry_counts())
+        }
+    }
+
+    impl Drop for Harness {
+        fn drop(&mut self) {
+            // Disconnect and reap the worker thread so it cannot outlive the test.
+            let (disconnected, _) = unbounded();
+            self.commands = disconnected;
+            if let Some(worker) = self.worker.take() {
+                let _ = worker.join();
+            }
+        }
+    }
+
+    /// A pool the tests script per parent: [`Self::push`] hands a transaction to the iterator
+    /// opened for that parent, and unknown parents read as "not tracking this head yet".
+    #[derive(Debug, Default)]
+    struct ScriptedPool {
+        queues: Arc<Mutex<HashMap<B256, VecDeque<PoolTransaction<EthPrimitives>>>>>,
+        /// Iterators handed out; the worker is expected to open exactly one per parent.
+        opened: AtomicUsize,
+        /// Polls answered with "not tracking this head yet".
+        not_ready: AtomicUsize,
+    }
+
+    impl ScriptedPool {
+        fn push(&self, parent_hash: B256, transaction: PoolTransaction<EthPrimitives>) {
+            self.queues.lock().entry(parent_hash).or_default().push_back(transaction);
+        }
+    }
+
+    impl Source<EthPrimitives> for ScriptedPool {
+        fn best_transactions(&self, parent_hash: B256) -> Option<Transactions<EthPrimitives>> {
+            if !self.queues.lock().contains_key(&parent_hash) {
+                self.not_ready.fetch_add(1, Ordering::Relaxed);
+                return None
+            }
+            self.opened.fetch_add(1, Ordering::Relaxed);
+            let queues = Arc::clone(&self.queues);
+            Some(Box::new(std::iter::from_fn(move || {
+                queues.lock().get_mut(&parent_hash)?.pop_front()
+            })))
+        }
+    }
+
+    /// A signed transfer to `recipient`. The signature is a dummy: the worker executes with the
+    /// attached sender and disabled nonce/balance checks, so it is never recovered or validated.
+    fn transfer(recipient: u8) -> PoolTransaction<EthPrimitives> {
+        let transaction = TxLegacy {
+            gas_limit: 21_000,
+            to: TxKind::Call(Address::repeat_byte(recipient)),
+            value: U256::from(1),
+            ..Default::default()
+        };
+        let hash = B256::repeat_byte(recipient);
+        let signed = TransactionSigned::Legacy(Signed::new_unchecked(
+            transaction,
+            Signature::test_signature(),
+            hash,
+        ));
+        let sender = Address::repeat_byte(0xAA);
+        PoolTransaction { hash, sender, transaction: Recovered::new_unchecked(signed, sender) }
+    }
+
+    fn wait_until(what: &str, condition: impl Fn() -> bool) {
+        let deadline = Instant::now() + WAIT_LIMIT;
+        while !condition() {
+            assert!(Instant::now() < deadline, "timed out waiting until {what}");
+            thread::sleep(POLL_INTERVAL);
+        }
+    }
+
+    #[test]
+    fn warms_pool_transactions_into_a_published_snapshot() {
+        let harness = Harness::spawn();
+        let parent_hash = B256::repeat_byte(0x01);
+
+        // Start before the pool tracks the head, covering the poll-and-retry path.
+        harness.start(parent_hash);
+        wait_until("the untracked head is polled", || {
+            harness.pool.not_ready.load(Ordering::Relaxed) >= 1
+        });
+        harness.pool.push(parent_hash, transfer(0xB0));
+
+        let snapshot = harness.published_for(parent_hash);
+        let (accounts, _, _) = snapshot.entry_counts();
+        assert!(accounts >= 1, "speculative execution should cache account reads");
+    }
+
+    #[test]
+    fn pause_quiesces_the_worker_until_resume() {
+        let harness = Harness::spawn();
+        let parent_hash = B256::repeat_byte(0x01);
+        harness.start(parent_hash);
+        harness.pool.push(parent_hash, transfer(0xB0));
+        let before = harness.published_for(parent_hash).entry_counts();
+
+        harness.pause();
+        harness.pool.push(parent_hash, transfer(0xB1));
+        thread::sleep(REFRESH_INTERVAL * 2);
+        assert_eq!(
+            harness.published_entry_counts(),
+            Some(before),
+            "a paused worker must not publish"
+        );
+
+        harness.resume();
+        harness.published(|snapshot| snapshot.entry_counts() != before);
+    }
+
+    #[test]
+    fn overlapping_pauses_require_matching_resumes() {
+        let harness = Harness::spawn();
+        let parent_hash = B256::repeat_byte(0x01);
+        harness.start(parent_hash);
+        harness.pool.push(parent_hash, transfer(0xB0));
+        let before = harness.published_for(parent_hash).entry_counts();
+
+        harness.pause();
+        harness.pause();
+        harness.pool.push(parent_hash, transfer(0xB1));
+        harness.resume();
+        thread::sleep(REFRESH_INTERVAL * 2);
+        assert_eq!(
+            harness.published_entry_counts(),
+            Some(before),
+            "one resume must not release two pauses"
+        );
+
+        harness.resume();
+        harness.published(|snapshot| snapshot.entry_counts() != before);
+    }
+
+    #[test]
+    fn reuses_the_iterator_per_head_and_reopens_on_switch() {
+        let harness = Harness::spawn();
+        let first = B256::repeat_byte(0x01);
+        let second = B256::repeat_byte(0x02);
+
+        harness.start(first);
+        harness.pool.push(first, transfer(0xB0));
+        let before = harness.published_for(first).entry_counts();
+        harness.pool.push(first, transfer(0xB1));
+        harness.published(|snapshot| snapshot.entry_counts() != before);
+        assert_eq!(harness.pool.opened.load(Ordering::Relaxed), 1, "one iterator per head");
+
+        harness.start(second);
+        harness.pool.push(second, transfer(0xB2));
+        harness.published_for(second);
+        assert_eq!(harness.pool.opened.load(Ordering::Relaxed), 2);
+    }
+
+    #[test]
+    fn newest_start_wins() {
+        let harness = Harness::spawn();
+        let stale = B256::repeat_byte(0x01);
+        let newest = B256::repeat_byte(0x02);
+
+        harness.start(stale);
+        harness.start(newest);
+        harness.pool.push(newest, transfer(0xB0));
+
+        harness.published_for(newest);
+    }
+
+    #[test]
+    fn shuts_down_when_control_is_dropped() {
+        let harness = Harness::spawn();
+        let parent_hash = B256::repeat_byte(0x01);
+        harness.start(parent_hash);
+        harness.pool.push(parent_hash, transfer(0xB0));
+        harness.published_for(parent_hash);
+
+        harness.shutdown();
+    }
+}
diff --git a/crates/engine/tree/src/tree/types.rs b/crates/engine/tree/src/tree/types.rs
index 2030c13c4d6..effab8bd947 100644
--- a/crates/engine/tree/src/tree/types.rs
+++ b/crates/engine/tree/src/tree/types.rs
@@ -6,6 +6,7 @@ use alloy_eips::eip4895::Withdrawal;
 use alloy_primitives::B256;
 use reth_chain_state::{ExecutedBlock, ExecutionTimingStats};
 use reth_evm::{ConfigureEvm, EvmEnvFor};
+use reth_execution_cache::TxPoolPrewarmCacheSnapshot;
 use reth_primitives_traits::{BlockTy, NodePrimitives};
 use std::sync::Arc;
 
@@ -34,6 +35,10 @@ pub struct ExecutionEnv<Evm: ConfigureEvm> {
     /// Optional decoded BAL for the block.
     /// Used to validate and optimize execution.
     pub decoded_bal: Option<Arc<DecodedBal>>,
+    /// Latest completed txpool-prewarm snapshot for this block's parent state.
+    ///
+    /// Can be None if txpool prewarming is disabled or the snapshot is not ready for some reason.
+    pub txpool_snapshot: Option<TxPoolPrewarmCacheSnapshot>,
 }
 
 impl<Evm: ConfigureEvm> ExecutionEnv<Evm>
@@ -52,6 +57,7 @@ where
             gas_used: 0,
             withdrawals: None,
             decoded_bal: None,
+            txpool_snapshot: None,
         }
     }
 }
diff --git a/crates/node/builder/src/lib.rs b/crates/node/builder/src/lib.rs
index 7f1e71d5c17..a8722004437 100644
--- a/crates/node/builder/src/lib.rs
+++ b/crates/node/builder/src/lib.rs
@@ -36,6 +36,8 @@ pub use launch::{
     *,
 };
 
+mod txpool_prewarm;
+
 mod handle;
 pub use handle::NodeHandle;
 
diff --git a/crates/node/builder/src/rpc.rs b/crates/node/builder/src/rpc.rs
index 491db06fc6b..300b9b721c4 100644
--- a/crates/node/builder/src/rpc.rs
+++ b/crates/node/builder/src/rpc.rs
@@ -13,8 +13,8 @@ pub use reth_rpc_builder::{
 pub use reth_trie_db::ChangesetCache;
 
 use crate::{
-    invalid_block_hook::InvalidBlockHookExt, ConfigureEngineEvm, ConsensusEngineEvent,
-    ConsensusEngineHandle,
+    invalid_block_hook::InvalidBlockHookExt, txpool_prewarm, ConfigureEngineEvm,
+    ConsensusEngineEvent, ConsensusEngineHandle,
 };
 use alloy_rpc_types::engine::ClientVersionV1;
 use alloy_rpc_types_engine::ExecutionData;
@@ -1468,7 +1468,8 @@ where
         let data_dir = ctx.config.datadir.clone().resolve_datadir(ctx.config.chain.chain());
         let invalid_block_hook = ctx.create_invalid_block_hook(&data_dir).await?;
 
-        Ok(BasicEngineValidator::new(
+        let txpool_prewarming = tree_config.txpool_prewarming();
+        let mut validator = BasicEngineValidator::new(
             ctx.node.provider().clone(),
             std::sync::Arc::new(ctx.node.consensus().clone()),
             ctx.node.evm_config().clone(),
@@ -1478,7 +1479,14 @@ where
             changeset_cache,
             state_trie_overlays,
             ctx.node.task_executor().clone(),
-        ))
+        );
+
+        if txpool_prewarming {
+            validator = validator
+                .with_txpool_prewarming(txpool_prewarm::Source::new(ctx.node.pool().clone()));
+        }
+
+        Ok(validator)
     }
 }
 
diff --git a/crates/node/builder/src/txpool_prewarm.rs b/crates/node/builder/src/txpool_prewarm.rs
new file mode 100644
index 00000000000..c895ffa1578
--- /dev/null
+++ b/crates/node/builder/src/txpool_prewarm.rs
@@ -0,0 +1,54 @@
+//! Transaction-pool backed candidate source for engine cache prewarming.
+
+use alloy_primitives::B256;
+use reth_engine_tree::tree::{
+    TxPoolPrewarmSource as PrewarmSource, TxPoolPrewarmTransaction as Transaction,
+    TxPoolPrewarmTransactions as Transactions,
+};
+use reth_primitives_traits::{NodePrimitives, TxTy};
+use reth_transaction_pool::{
+    BestTransactions, BestTransactionsAttributes, PoolTransaction, TransactionPool,
+};
+use std::fmt::Debug;
+
+/// [`TransactionPool`]-backed [`PrewarmSource`].
+#[derive(Debug)]
+pub(crate) struct Source<P>(P);
+
+impl<P> Source<P> {
+    /// Creates a new txpool prewarm source.
+    pub(crate) const fn new(pool: P) -> Self {
+        Self(pool)
+    }
+}
+
+impl<N, P> PrewarmSource<N> for Source<P>
+where
+    N: NodePrimitives,
+    P: TransactionPool<Transaction: PoolTransaction<Consensus = TxTy<N>>>
+        + Clone
+        + Send
+        + Sync
+        + Debug
+        + 'static,
+{
+    fn best_transactions(&self, parent_hash: B256) -> Option<Transactions<N>> {
+        let block_info = self.0.block_info();
+        if block_info.last_seen_block_hash != parent_hash {
+            return None
+        }
+
+        let mut best = self.0.best_transactions_with_attributes(BestTransactionsAttributes::new(
+            block_info.pending_basefee,
+            block_info.pending_blob_fee.map(|fee| u64::try_from(fee).unwrap_or(u64::MAX)),
+        ));
+        best.allow_updates_out_of_order();
+        best.skip_blobs();
+
+        Some(Box::new(best.map(|transaction| Transaction {
+            hash: *transaction.hash(),
+            sender: transaction.sender(),
+            transaction: transaction.transaction.clone_into_consensus(),
+        })))
+    }
+}
diff --git a/crates/node/core/src/args/engine.rs b/crates/node/core/src/args/engine.rs
index 3d118d04836..b6e0a60c2cb 100644
--- a/crates/node/core/src/args/engine.rs
+++ b/crates/node/core/src/args/engine.rs
@@ -31,6 +31,7 @@ pub struct DefaultEngineValues {
     invalid_header_hit_eviction_threshold: u8,
     state_cache_disabled: bool,
     prewarming_disabled: bool,
+    txpool_prewarming_enabled: bool,
     state_provider_metrics: bool,
     cross_block_cache_size: usize,
     state_root_task_compare_updates: bool,
@@ -102,6 +103,12 @@ impl DefaultEngineValues {
         self
     }
 
+    /// Set whether to enable txpool prewarming by default
+    pub const fn with_txpool_prewarming_enabled(mut self, v: bool) -> Self {
+        self.txpool_prewarming_enabled = v;
+        self
+    }
+
     /// Set whether to enable state provider metrics by default
     pub const fn with_state_provider_metrics(mut self, v: bool) -> Self {
         self.state_provider_metrics = v;
@@ -247,6 +254,7 @@ impl Default for DefaultEngineValues {
             invalid_header_hit_eviction_threshold: DEFAULT_INVALID_HEADER_HIT_EVICTION_THRESHOLD,
             state_cache_disabled: false,
             prewarming_disabled: false,
+            txpool_prewarming_enabled: false,
             state_provider_metrics: false,
             cross_block_cache_size: DEFAULT_CROSS_BLOCK_CACHE_SIZE_MB,
             state_root_task_compare_updates: false,
@@ -337,6 +345,14 @@ pub struct EngineArgs {
     #[arg(long = "engine.disable-prewarming", alias = "engine.disable-caching-and-prewarming", default_value_t = DefaultEngineValues::get_global().prewarming_disabled)]
     pub prewarming_disabled: bool,
 
+    /// Enable best-effort txpool transaction prewarming between payloads.
+    #[arg(
+        long = "engine.txpool-prewarming",
+        default_value_t = DefaultEngineValues::get_global().txpool_prewarming_enabled,
+        conflicts_with = "state_cache_disabled"
+    )]
+    pub txpool_prewarming_enabled: bool,
+
     /// CAUTION: This CLI flag has no effect anymore. The parallel sparse trie is always enabled.
     #[deprecated]
     #[arg(long = "engine.parallel-sparse-trie", default_value = "true", hide = true)]
@@ -534,6 +550,7 @@ impl Default for EngineArgs {
             invalid_header_hit_eviction_threshold,
             state_cache_disabled,
             prewarming_disabled,
+            txpool_prewarming_enabled,
             state_provider_metrics,
             cross_block_cache_size,
             state_root_task_compare_updates,
@@ -567,6 +584,7 @@ impl Default for EngineArgs {
             caching_and_prewarming_enabled: true,
             state_cache_disabled,
             prewarming_disabled,
+            txpool_prewarming_enabled,
             parallel_sparse_trie_enabled: true,
             parallel_sparse_trie_disabled: false,
             state_provider_metrics,
@@ -652,6 +670,7 @@ impl EngineArgs {
             .with_invalid_header_hit_eviction_threshold(self.invalid_header_hit_eviction_threshold)
             .without_state_cache(self.state_cache_disabled)
             .without_prewarming(self.prewarming_disabled)
+            .with_txpool_prewarming(self.txpool_prewarming_enabled)
             .with_state_provider_metrics(self.state_provider_metrics)
             .with_always_compare_trie_updates(self.state_root_task_compare_updates)
             .with_cross_block_cache_size(self.cross_block_cache_size * 1024 * 1024)
@@ -709,6 +728,30 @@ mod tests {
         );
     }
 
+    #[test]
+    fn txpool_prewarming_is_disabled_by_default_and_can_be_enabled() {
+        let args = CommandParser::<EngineArgs>::parse_from(["reth"]).args;
+        assert!(!args.txpool_prewarming_enabled);
+        assert!(!args.tree_config().txpool_prewarming());
+
+        let args =
+            CommandParser::<EngineArgs>::parse_from(["reth", "--engine.txpool-prewarming"]).args;
+        assert!(args.txpool_prewarming_enabled);
+        assert!(args.tree_config().txpool_prewarming());
+    }
+
+    #[test]
+    fn txpool_prewarming_conflicts_with_disabled_state_cache() {
+        let Err(error) = CommandParser::<EngineArgs>::try_parse_from([
+            "reth",
+            "--engine.txpool-prewarming",
+            "--engine.disable-state-cache",
+        ]) else {
+            panic!("conflicting flags must be rejected")
+        };
+        assert_eq!(error.kind(), clap::error::ErrorKind::ArgumentConflict);
+    }
+
     #[test]
     fn default_backpressure_threshold_uses_parsed_persistence_args() {
         let args = CommandParser::<EngineArgs>::parse_from([
@@ -771,6 +814,8 @@ mod tests {
             caching_and_prewarming_enabled: true,
             state_cache_disabled: true,
             prewarming_disabled: true,
+            // conflicts with --engine.disable-state-cache, covered by its own test below
+            txpool_prewarming_enabled: false,
             parallel_sparse_trie_enabled: true,
             parallel_sparse_trie_disabled: false,
             state_provider_metrics: true,
diff --git a/crates/node/events/src/node.rs b/crates/node/events/src/node.rs
index 7223026273f..7ee9ef5660f 100644
--- a/crates/node/events/src/node.rs
+++ b/crates/node/events/src/node.rs
@@ -337,6 +337,15 @@ impl NodeState {
                     cache.code.hits = stats.code_cache_hits,
                     cache.code.misses = stats.code_cache_misses,
                     cache.code.hit_rate = format!("{:.2}", hit_rate(stats.code_cache_hits, stats.code_cache_misses)),
+                    cache.txpool_snapshot.account.hits = stats.txpool_snapshot_account_hits,
+                    cache.txpool_snapshot.account.misses = stats.txpool_snapshot_account_misses,
+                    cache.txpool_snapshot.account.hit_rate = format!("{:.2}", hit_rate(stats.txpool_snapshot_account_hits, stats.txpool_snapshot_account_misses)),
+                    cache.txpool_snapshot.storage.hits = stats.txpool_snapshot_storage_hits,
+                    cache.txpool_snapshot.storage.misses = stats.txpool_snapshot_storage_misses,
+                    cache.txpool_snapshot.storage.hit_rate = format!("{:.2}", hit_rate(stats.txpool_snapshot_storage_hits, stats.txpool_snapshot_storage_misses)),
+                    cache.txpool_snapshot.code.hits = stats.txpool_snapshot_code_hits,
+                    cache.txpool_snapshot.code.misses = stats.txpool_snapshot_code_misses,
+                    cache.txpool_snapshot.code.hit_rate = format!("{:.2}", hit_rate(stats.txpool_snapshot_code_hits, stats.txpool_snapshot_code_misses)),
                 );
             }
         }
diff --git a/crates/transaction-pool/src/pool/best.rs b/crates/transaction-pool/src/pool/best.rs
index 6d6b46cdf8e..e3e214dd9d8 100644
--- a/crates/transaction-pool/src/pool/best.rs
+++ b/crates/transaction-pool/src/pool/best.rs
@@ -41,6 +41,10 @@ impl<T: TransactionOrdering> crate::traits::BestTransactions for BestTransaction
         self.best.no_updates()
     }
 
+    fn allow_updates_out_of_order(&mut self) {
+        self.best.allow_updates_out_of_order()
+    }
+
     fn skip_blobs(&mut self) {
         self.set_skip_blobs(true)
     }
@@ -111,6 +115,8 @@ pub struct BestTransactions<T: TransactionOrdering> {
     pub(crate) last_priority: Option<Priority<T::PriorityValue>>,
     /// Flag to control whether to skip blob transactions (EIP4844).
     pub(crate) skip_blobs: bool,
+    /// Whether live updates can be yielded after a lower-priority transaction.
+    pub(crate) allow_updates_out_of_order: bool,
 }
 
 impl<T: TransactionOrdering> BestTransactions<T> {
@@ -136,7 +142,8 @@ impl<T: TransactionOrdering> BestTransactions<T> {
         loop {
             match self.new_transaction_receiver.as_mut()?.try_recv() {
                 Ok(tx) => {
-                    if let Some(last_priority) = &self.last_priority &&
+                    if !self.allow_updates_out_of_order &&
+                        let Some(last_priority) = &self.last_priority &&
                         &tx.priority > last_priority
                     {
                         // we skip transactions if we already yielded a transaction with lower
@@ -278,6 +285,10 @@ impl<T: TransactionOrdering> crate::traits::BestTransactions for BestTransaction
         self.last_priority.take();
     }
 
+    fn allow_updates_out_of_order(&mut self) {
+        self.allow_updates_out_of_order = true;
+    }
+
     fn skip_blobs(&mut self) {
         self.set_skip_blobs(true);
     }
@@ -355,6 +366,10 @@ where
         self.best.no_updates()
     }
 
+    fn allow_updates_out_of_order(&mut self) {
+        self.best.allow_updates_out_of_order()
+    }
+
     fn skip_blobs(&mut self) {
         self.set_skip_blobs(true)
     }
@@ -454,6 +469,10 @@ where
         self.inner.no_updates()
     }
 
+    fn allow_updates_out_of_order(&mut self) {
+        self.inner.allow_updates_out_of_order()
+    }
+
     fn set_skip_blobs(&mut self, skip_blobs: bool) {
         if skip_blobs {
             self.buffer.retain(|tx| !tx.transaction.is_eip4844())
@@ -1107,6 +1126,23 @@ mod tests {
         assert!(best.new_transaction_receiver.is_none());
     }
 
+    #[test]
+    fn test_best_transactions_yields_updates_after_empty() {
+        let mut pool = PendingPool::new(MockOrdering::default());
+        let mut best = pool.best();
+        best.allow_updates_out_of_order();
+
+        assert!(best.next().is_none());
+
+        let mut f = MockTransactionFactory::default();
+        let tx = MockTransaction::eip1559().rng_hash();
+        let valid_tx = Arc::new(f.validated(tx));
+        let expected_hash = *valid_tx.hash();
+        pool.add_transaction(valid_tx, 0);
+
+        assert_eq!(*best.next().expect("new transaction should be yielded").hash(), expected_hash);
+    }
+
     #[test]
     fn test_best_update_transaction_priority() {
         let mut pool = PendingPool::new(MockOrdering::default());
diff --git a/crates/transaction-pool/src/pool/pending.rs b/crates/transaction-pool/src/pool/pending.rs
index 57a70c90c91..7ca27ad9b66 100644
--- a/crates/transaction-pool/src/pool/pending.rs
+++ b/crates/transaction-pool/src/pool/pending.rs
@@ -109,6 +109,7 @@ impl<T: TransactionOrdering> PendingPool<T> {
             new_transaction_receiver: Some(self.new_transaction_notifier.subscribe()),
             last_priority: None,
             skip_blobs: false,
+            allow_updates_out_of_order: false,
         }
     }
 
diff --git a/crates/transaction-pool/src/traits.rs b/crates/transaction-pool/src/traits.rs
index 654ef7e5a87..877e1e6e6f1 100644
--- a/crates/transaction-pool/src/traits.rs
+++ b/crates/transaction-pool/src/traits.rs
@@ -1119,6 +1119,14 @@ pub trait BestTransactions: Iterator + Send {
     /// listen to pool updates.
     fn no_updates(&mut self);
 
+    /// Allows newly received transactions to be yielded even if their priority is higher than a
+    /// transaction that was already yielded.
+    ///
+    /// This is useful for long-lived consumers that prefer seeing every update over preserving
+    /// decreasing priority order. The default implementation leaves the iterator's ordering
+    /// behavior unchanged. Implementations must still preserve transaction dependency ordering.
+    fn allow_updates_out_of_order(&mut self) {}
+
     /// Convenience function for [`Self::no_updates`] that returns the iterator again.
     fn without_updates(mut self) -> Self
     where
@@ -1180,6 +1188,10 @@ where
         (**self).no_updates();
     }
 
+    fn allow_updates_out_of_order(&mut self) {
+        (**self).allow_updates_out_of_order();
+    }
+
     fn skip_blobs(&mut self) {
         (**self).skip_blobs();
     }
diff --git a/docs/vocs/docs/pages/cli/reth/node.mdx b/docs/vocs/docs/pages/cli/reth/node.mdx
index 3f6a0593a32..eaec84883dd 100644
--- a/docs/vocs/docs/pages/cli/reth/node.mdx
+++ b/docs/vocs/docs/pages/cli/reth/node.mdx
@@ -1011,6 +1011,9 @@ Engine:
       --engine.disable-prewarming
           Disable parallel prewarming
 
+      --engine.txpool-prewarming
+          Enable best-effort txpool transaction prewarming between payloads
+
       --engine.state-provider-metrics
           Enable state provider latency metrics. This allows the engine to collect and report stats about how long state provider calls took during execution, but this does introduce slight overhead to state provider calls