feat(l1): switch EIP-8070 blob sampling on at Amsterdam

cmd/ethrex/cli.rs

@@ -524,7 +524,7 @@ pub struct Options {
         long = "blob-sampling",
         action = ArgAction::SetTrue,
         default_value = "false",
-        help = "Enable EIP-8070 PeerDAS blob sampling (sampler/provider state machine). Disabled by default; when off the node always acts as provider (p=1.0).",
+        help = "Enable EIP-8070 PeerDAS blob sampling (sampler/provider state machine) before Amsterdam. From Amsterdam onwards it is on regardless; until then the node always acts as provider (p=1.0).",
         help_heading = "P2P options",
         env = "ETHREX_BLOB_SAMPLING"
     )]

cmd/ethrex/initializers.rs

@@ -922,7 +922,8 @@ pub async fn init_l1(
             bal_parallel_exec_enabled: !opts.no_bal_parallel_exec,
             bal_prefetch_enabled: !opts.no_bal_prefetch,
             bal_parallel_trie_enabled: !opts.no_bal_parallel_trie,
-            blob_sampling_enabled: opts.blob_sampling || opts.blob_eager_provider,
+            blob_txs_supported: true,
+        force_blob_sampling: opts.blob_sampling || opts.blob_eager_provider,
             blob_eager_provider: opts.blob_eager_provider,
             max_reorg_depth: opts.max_reorg_depth,
             gap_admit_occupancy_threshold: opts.mempool_gap_admit_occupancy_threshold,

cmd/ethrex/l2/initializers.rs

@@ -256,8 +256,9 @@ pub async fn init_l2(
         max_mempool_size: opts.node_opts.mempool_max_size,
         r#type: BlockchainType::L2(l2_config),
         perf_logs_enabled: true,
-        max_blobs_per_block: None, // L2 doesn't support blob transactions
-        blob_sampling_enabled: false, // L2 rejects blob txs; no eth/72 sampling
+        max_blobs_per_block: None,  // L2 doesn't support blob transactions
+        blob_txs_supported: false,  // L2 rejects blob txs
+        force_blob_sampling: false, // ... so no eth/72 sampling, at any fork
         blob_eager_provider: false,
         precompute_witnesses: opts.node_opts.precompute_witnesses,
         private_mempool: opts.node_opts.mempool_private,
@@ -510,7 +511,8 @@ pub async fn init_native_rollup_l2(
         r#type: BlockchainType::L1,
         perf_logs_enabled: true,
         max_blobs_per_block: None,
-        blob_sampling_enabled: false, // L2 rejects blob txs; no eth/72 sampling
+        blob_txs_supported: false,  // L2 rejects blob txs
+        force_blob_sampling: false, // ... so no eth/72 sampling, at any fork
         blob_eager_provider: false,
         precompute_witnesses: opts.node_opts.precompute_witnesses,
         precompile_cache_enabled: true,

crates/blockchain/blockchain.rs

@@ -396,14 +396,28 @@ pub struct BlockchainOptions {
     /// `--no-bal-parallel-trie`) to fall back to streaming `AccountUpdate`s from
     /// the executor and merkleizing post-execution.
     pub bal_parallel_trie_enabled: bool,
-    /// EIP-8070: when true, activate the sampler/provider state machine.
-    /// When false (default), the node always acts as provider (p=1.0).
-    pub blob_sampling_enabled: bool,
+    /// Whether this chain accepts blob transactions at all.
+    ///
+    /// False for rollups, which reject them outright, so the EIP-8070 blobpool
+    /// state machine must not switch itself on when their fork schedule reaches
+    /// Amsterdam. `BlockchainType` alone cannot carry this: native rollups run as
+    /// [`BlockchainType::L1`] on purpose, because their blocks must re-execute
+    /// under an unmodified L1 environment.
+    pub blob_txs_supported: bool,
+    /// EIP-8070: when true, activate the sampler/provider state machine at
+    /// startup regardless of the chain's fork schedule (`--blob-sampling`).
+    ///
+    /// When false (default), sampling still switches itself on once the chain
+    /// head reaches Amsterdam; see [`Blockchain::blob_sampling_enabled`]. This
+    /// flag only brings that forward, for devnets and for chains that schedule
+    /// the fork later than they want the state machine running.
+    pub force_blob_sampling: bool,
     /// EIP-8070: when true, always act as provider (p=1.0) regardless of role
     /// randomization, as block builders SHOULD (EIP-8070, "Execution clients ::
     /// Local block builders"). Enabled via `--blob-eager-provider`; a node that
-    /// builds payloads latches it at runtime regardless. Only meaningful when
-    /// `blob_sampling_enabled` is also true.
+    /// builds payloads latches it at runtime regardless. Implies
+    /// `force_blob_sampling`, since eager provider is a role *within* the
+    /// sampling state machine.
     pub blob_eager_provider: bool,
     /// Optional operator override for the maximum reorg depth. `None` ; cap is purely
     /// physical (layer-cache retention plus journal reach; bounded indirectly by finality
@@ -446,7 +460,8 @@ impl Default for BlockchainOptions {
             bal_parallel_exec_enabled: true,
             bal_prefetch_enabled: true,
             bal_parallel_trie_enabled: true,
-            blob_sampling_enabled: false,
+            blob_txs_supported: true,
+            force_blob_sampling: false,
             blob_eager_provider: false,
             max_reorg_depth: None,
             gap_admit_occupancy_threshold: DEFAULT_GAP_ADMIT_OCCUPANCY_THRESHOLD,
@@ -588,7 +603,7 @@ impl Blockchain {
     pub fn new(store: Store, blockchain_opts: BlockchainOptions) -> Self {
         let mempool = if blockchain_opts.blob_eager_provider {
             Mempool::new_with_eager_provider(blockchain_opts.max_mempool_size)
-        } else if blockchain_opts.blob_sampling_enabled {
+        } else if blockchain_opts.force_blob_sampling {
             Mempool::new_with_sampling(blockchain_opts.max_mempool_size)
         } else {
             Mempool::new(blockchain_opts.max_mempool_size)
@@ -4177,6 +4192,50 @@ impl Blockchain {
         self.snap_syncing.load(Ordering::Relaxed)
     }
 
+    /// Whether the EIP-8070 sampler/provider state machine is active.
+    ///
+    /// This is what decides whether eth/72 may be offered to a peer, and which
+    /// role this node takes for each blob transaction. It turns on when the
+    /// chain head reaches Amsterdam, the fork EIP-8070 rides on, or earlier if
+    /// the operator passed `--blob-sampling` / `--blob-eager-provider`.
+    ///
+    /// Gating on the fork rather than on a flag alone keeps the two halves of
+    /// eth/72 in step. eth/72 always elides blob payloads from
+    /// `PooledTransactions`, so blobs arrive only through `GetCells`, which only
+    /// this state machine issues. A node that advertised the capability without
+    /// it would accept blob transactions it could never reconstruct.
+    ///
+    /// Latching happens here rather than on block import so that the fork check
+    /// costs nothing on the import path; the p2p paths that ask this question
+    /// run per connection or per announcement, and pay one atomic load each once
+    /// the fork is behind us.
+    pub fn blob_sampling_enabled(&self) -> bool {
+        if self.mempool.blob_sampling_enabled() {
+            return true;
+        }
+        // Rollups reject blob txs outright, so their fork schedule must not drag
+        // the blobpool state machine in. The explicit flag carries native rollups,
+        // which run as `BlockchainType::L1`; the type check stays as a backstop for
+        // any L2 that forgets to set it.
+        if !self.options.blob_txs_supported || !matches!(self.options.r#type, BlockchainType::L1) {
+            return false;
+        }
+        let head = self.storage.latest_block_timestamp();
+        if !self.storage.get_chain_config().is_amsterdam_activated(head) {
+            return false;
+        }
+        self.mempool.enable_blob_sampling();
+        true
+    }
+
+    /// Latch eager-provider mode on, resolving the fork-driven sampling latch
+    /// first: [`Mempool::latch_eager_provider`] is inert while sampling is off.
+    pub fn latch_eager_provider(&self) {
+        if self.blob_sampling_enabled() {
+            self.mempool.latch_eager_provider();
+        }
+    }
+
     pub fn get_p2p_transaction_by_hash(&self, hash: &H256) -> Result<P2PTransaction, StoreError> {
         // --mempool.private: never serve private txs over P2P, even if a peer
         // somehow learned the hash. The spec for `GetPooledTransactions`

crates/blockchain/mempool.rs

@@ -661,8 +661,13 @@ pub struct Mempool {
     /// to return.
     tx_seq: AtomicU64,
     /// When true, the EIP-8070 sampler/provider state machine is active.
-    /// When false (default), the node always acts as provider (p=1.0).
-    pub blob_sampling_enabled: bool,
+    /// When false, the node always acts as provider (p=1.0).
+    ///
+    /// Seeded from the operator's `--blob-sampling` / `--blob-eager-provider`
+    /// choice, then latched on by `Blockchain::blob_sampling_enabled` once the
+    /// chain head reaches Amsterdam, the fork EIP-8070 rides on. One-way: see
+    /// [`Self::enable_blob_sampling`].
+    blob_sampling: AtomicBool,
     /// When true, this node always acts as provider (p=1.0) for every blob tx
     /// regardless of the pseudo-random role decision. Block builders SHOULD
     /// permanently act in eager mode (EIP-8070) to ensure they hold complete blob
@@ -688,7 +693,7 @@ impl Mempool {
             inner: RwLock::new(MempoolInner::new(max_mempool_size)),
             tx_added: tokio::sync::Notify::new(),
             tx_seq: AtomicU64::new(0),
-            blob_sampling_enabled: false,
+            blob_sampling: AtomicBool::new(false),
             eager_provider: AtomicBool::new(false),
             custody_generation: AtomicU64::new(0),
         }
@@ -697,7 +702,7 @@ impl Mempool {
     /// Create a mempool with blob sampling enabled.
     pub fn new_with_sampling(max_mempool_size: usize) -> Self {
         Mempool {
-            blob_sampling_enabled: true,
+            blob_sampling: AtomicBool::new(true),
             ..Self::new(max_mempool_size)
         }
     }
@@ -716,12 +721,32 @@ impl Mempool {
     /// "Execution clients :: Local block builders").
     pub fn new_with_eager_provider(max_mempool_size: usize) -> Self {
         Mempool {
-            blob_sampling_enabled: true,
+            blob_sampling: AtomicBool::new(true),
             eager_provider: AtomicBool::new(true),
             ..Self::new(max_mempool_size)
         }
     }
 
+    /// Whether the EIP-8070 sampler/provider state machine is active.
+    pub fn blob_sampling_enabled(&self) -> bool {
+        self.blob_sampling.load(Ordering::Acquire)
+    }
+
+    /// Turn the EIP-8070 state machine on permanently.
+    ///
+    /// Called by `Blockchain::blob_sampling_enabled` when the chain head first
+    /// reaches Amsterdam, and one-way for the same reason [`Self::latch_eager_provider`]
+    /// is: a chain that reorgs back below the fork boundary would otherwise flap
+    /// the advertised eth capability set, dropping and re-forming peer
+    /// connections over a boundary the chain is about to cross again anyway.
+    pub(crate) fn enable_blob_sampling(&self) {
+        if !self.blob_sampling.swap(true, Ordering::AcqRel) {
+            info!(
+                "Amsterdam reached: enabling EIP-8070 blob sampling (node now advertises eth/72)"
+            );
+        }
+    }
+
     /// Whether this node acts as an eager provider (p=1.0 for every blob tx).
     pub fn is_eager_provider(&self) -> bool {
         self.eager_provider.load(Ordering::Acquire)
@@ -733,10 +758,10 @@ impl Mempool {
     /// The latch is one-way on purpose — dropping back to sampling between slots
     /// would leak the node's proposer schedule through its fetch pattern.
     ///
-    /// Inert unless sampling is enabled: without `--blob-sampling` the node is
-    /// already a full-replication provider.
+    /// Inert unless sampling is enabled: before Amsterdam (and without
+    /// `--blob-sampling`) the node is already a full-replication provider.
     pub fn latch_eager_provider(&self) {
-        if !self.blob_sampling_enabled {
+        if !self.blob_sampling_enabled() {
             return;
         }
         if !self.eager_provider.swap(true, Ordering::AcqRel) {
@@ -1777,6 +1802,20 @@ impl Mempool {
             .unwrap_or(0))
     }
 
+    /// Whether `peer_id` announced availability for `tx_hash` via
+    /// `NewPooledTransactionHashes72`.
+    ///
+    /// devp2p `caps/eth.md` only permits fetching cells "from peers that announced
+    /// overlapping availability", so any path that picks a peer for a transaction
+    /// it did not learn from that peer's own announcement must consult this first.
+    pub fn peer_announced_tx(&self, tx_hash: H256, peer_id: H256) -> Result<bool, StoreError> {
+        Ok(self
+            .read()?
+            .provider_announcers
+            .get(&tx_hash)
+            .is_some_and(|peers| peers.contains(&peer_id)))
+    }
+
     /// Forget a peer's last-advertised cell availability (called on disconnect).
     pub fn clear_peer_cell_availability(&self, peer_id: H256) -> Result<(), StoreError> {
         self.write()?.peer_cell_availability.remove(&peer_id);

crates/blockchain/payload.rs

@@ -441,7 +441,7 @@ impl Blockchain {
         // EIP-8070: a node that builds payloads SHOULD permanently act as an eager
         // provider, so it holds complete blob data for every blob tx it may include.
         // Inert unless blob sampling is enabled.
-        self.mempool.latch_eager_provider();
+        self.latch_eager_provider();
         let self_clone = self.clone();
         let cancel_token = CancellationToken::new();
         let cancel_token_clone = cancel_token.clone();

crates/networking/p2p/rlpx/connection/server.rs

@@ -751,7 +751,7 @@ impl PeerConnectionServer {
             // mode latched on, fetch the now-wanted columns this peer can serve for
             // pending blob txs. Inert unless sampling is enabled, and only ever
             // asked of an eth/72 peer.
-            if state.blockchain.mempool.blob_sampling_enabled && supports_eth72(state) {
+            if state.blockchain.blob_sampling_enabled() && supports_eth72(state) {
                 let generation = state.blockchain.mempool.custody_generation();
                 if generation != state.last_custody_generation {
                     state.last_custody_generation = generation;
@@ -763,9 +763,24 @@ impl PeerConnectionServer {
                         .peer_cell_mask(state.node.node_id())
                         .unwrap_or(None)
                         .unwrap_or(u128::MAX);
+                    let peer_id = state.node.node_id();
                     match state.blockchain.mempool.blob_txs_missing_cells() {
                         Ok(missing_list) => {
                             for (tx_hash, missing) in missing_list {
+                                // `blob_txs_missing_cells` spans the whole pool, but a
+                                // peer may only be asked for cells it advertised: devp2p
+                                // `caps/eth.md` fetches "from peers that announced
+                                // overlapping availability". Without this the sweep asks
+                                // every eth/72 peer for every pending blob tx, including
+                                // ones it never announced.
+                                if !state
+                                    .blockchain
+                                    .mempool
+                                    .peer_announced_tx(tx_hash, peer_id)
+                                    .unwrap_or(false)
+                                {
+                                    continue;
+                                }
                                 let fetch_mask = missing & peer_available;
                                 if fetch_mask != 0 {
                                     state
@@ -1272,10 +1287,11 @@ where
     // eth/72 (EIP-8070) is only safe to negotiate when blob sampling is enabled:
     // it always elides blob payloads in PooledTransactions, and a node that does
     // not run the sampler/provider cell-fetch loop would receive blob txs it can
-    // never reconstruct. With sampling off we cap at eth/71 so default nodes keep
-    // full-blob propagation unchanged. The EIP's Backwards Compatibility section
-    // explicitly supports this gradual, version-gated rollout.
-    let offer_eth72 = state.blockchain.mempool.blob_sampling_enabled;
+    // never reconstruct. Sampling switches on at Amsterdam, so pre-fork peers cap
+    // at eth/71 and keep full-blob propagation unchanged. The EIP's Backwards
+    // Compatibility section explicitly supports this gradual, version-gated
+    // rollout.
+    let offer_eth72 = state.blockchain.blob_sampling_enabled();
     // This allow is because in l2 we mut the capabilities
     // to include the l2 cap
     let snap_capabilities =
@@ -1810,7 +1826,7 @@ async fn handle_incoming_message(
                     announcement.get_transactions_to_request(&state.blockchain, peer_id)?;
 
                 if !hashes.is_empty() {
-                    if !state.blockchain.mempool.blob_sampling_enabled {
+                    if !state.blockchain.blob_sampling_enabled() {
                         // Sampling disabled: always provider — request everything.
                         // Trim to the truly-requested subset so the flush does not
                         // re-request hashes already in-flight from another peer.
@@ -2076,7 +2092,7 @@ async fn handle_incoming_message(
         }
         // eth/72 (EIP-8070): PooledTransactions72 handler.
         // Blob txs arrive with elided blobs — do NOT trigger the missing-blob disconnect.
-        Message::PooledTransactions72(msg) if peer_supports_eth => {
+        Message::PooledTransactions72(mut msg) if peer_supports_eth => {
             if !msg.pooled_transactions.is_empty() {
                 state.received_txs_from_peer = true;
             }
@@ -2105,6 +2121,11 @@ async fn handle_incoming_message(
                         ));
                     }
                 }
+                if let Some((announced, _, _, _)) = &removed_request {
+                    // Tolerated by `validate_requested`, but not admitted: only the
+                    // transactions this request asked for reach the pool.
+                    msg.retain_requested(announced);
+                }
                 #[cfg(feature = "l2")]
                 let is_l2_mode = state.l2_state.is_supported();
                 #[cfg(not(feature = "l2"))]
@@ -2137,7 +2158,7 @@ async fn handle_incoming_message(
                 }
                 // EIP-8070 sampler: after tx validation, check if we have enough provider
                 // announcements to start fetching cells.
-                if state.blockchain.mempool.blob_sampling_enabled {
+                if state.blockchain.blob_sampling_enabled() {
                     let peer_id = state.node.node_id();
                     let mempool = &state.blockchain.mempool;
                     let local_pubkey = public_key_from_signing_key(&state.signer);