ethrex_blockchain/
payload.rs

1use std::{
2    cmp::{Ordering, max},
3    ops::Div,
4    sync::Arc,
5    time::{Duration, Instant},
6};
7
8use rustc_hash::FxHashMap;
9
10use ethrex_common::{
11    Address, Bloom, Bytes, H256, U256,
12    constants::{
13        DEFAULT_OMMERS_HASH, DEFAULT_REQUESTS_HASH, GAS_PER_BLOB, MAX_RLP_BLOCK_SIZE,
14        TX_MAX_GAS_LIMIT_AMSTERDAM,
15    },
16    types::{
17        AccountUpdate, BlobsBundle, Block, BlockBody, BlockHash, BlockHeader, BlockNumber,
18        ChainConfig, MempoolTransaction, Receipt, Transaction, TxKind, TxType, Withdrawal,
19        block_access_list::BlockAccessList,
20        bloom_from_logs, calc_excess_blob_gas, calculate_base_fee_per_blob_gas,
21        calculate_base_fee_per_gas, compute_receipts_root, compute_transactions_root,
22        compute_withdrawals_root,
23        requests::{EncodedRequests, compute_requests_hash},
24    },
25};
26
27use ethrex_crypto::NativeCrypto;
28use ethrex_crypto::keccak::Keccak256;
29use ethrex_vm::{Evm, EvmError, check_2d_gas_allowance};
30
31use ethrex_rlp::encode::RLPEncode;
32use ethrex_storage::{Store, error::StoreError};
33
34use ethrex_metrics::metrics;
35
36#[cfg(feature = "metrics")]
37use ethrex_metrics::blocks::METRICS_BLOCKS;
38#[cfg(feature = "metrics")]
39use ethrex_metrics::transactions::{METRICS_TX, MetricsTxType};
40use tokio_util::sync::CancellationToken;
41
42use crate::{
43    Blockchain, BlockchainType, MAX_PAYLOADS,
44    constants::{GAS_LIMIT_BOUND_DIVISOR, MIN_GAS_LIMIT, TX_GAS_COST},
45    error::{ChainError, InvalidBlockError},
46    mempool::PendingTxFilter,
47    new_evm,
48    vm::StoreVmDatabase,
49};
50
51use thiserror::Error;
52use tracing::{debug, warn};
53
54#[derive(Debug)]
55pub struct PayloadBuildTask {
56    task: tokio::task::JoinHandle<Result<PayloadBuildResult, ChainError>>,
57    cancel: CancellationToken,
58}
59
60#[derive(Debug)]
61pub enum PayloadOrTask {
62    Payload(Box<PayloadBuildResult>),
63    Task(PayloadBuildTask),
64}
65
66impl PayloadBuildTask {
67    /// Finishes the current payload build process and returns its result
68    pub async fn finish(self) -> Result<PayloadBuildResult, ChainError> {
69        self.cancel.cancel();
70        self.task
71            .await
72            .map_err(|_| ChainError::Custom("Failed to join task".to_string()))?
73    }
74}
75
76impl PayloadOrTask {
77    /// Converts self into a `PayloadOrTask::Payload` by finishing the current build task
78    /// If self is already a `PayloadOrTask::Payload` this is a NoOp
79    pub async fn to_payload(self) -> Result<Self, ChainError> {
80        Ok(match self {
81            PayloadOrTask::Payload(_) => self,
82            PayloadOrTask::Task(task) => PayloadOrTask::Payload(Box::new(task.finish().await?)),
83        })
84    }
85}
86
87pub struct BuildPayloadArgs {
88    pub parent: BlockHash,
89    pub timestamp: u64,
90    pub fee_recipient: Address,
91    pub random: H256,
92    pub withdrawals: Option<Vec<Withdrawal>>,
93    pub beacon_root: Option<H256>,
94    pub slot_number: Option<u64>,
95    pub version: u8,
96    pub elasticity_multiplier: u64,
97    pub gas_ceil: u64,
98}
99
100#[derive(Debug, Error)]
101pub enum BuildPayloadArgsError {
102    #[error("Payload hashed has wrong size")]
103    FailedToConvertPayload,
104}
105
106impl BuildPayloadArgs {
107    /// Computes an 8-byte identifier by hashing the components of the payload arguments.
108    pub fn id(&self) -> Result<u64, BuildPayloadArgsError> {
109        let mut hasher = Keccak256::new();
110        hasher.update(self.parent);
111        hasher.update(self.timestamp.to_be_bytes());
112        hasher.update(self.random);
113        hasher.update(self.fee_recipient);
114        if let Some(withdrawals) = &self.withdrawals {
115            hasher.update(withdrawals.encode_to_vec());
116        }
117        if let Some(beacon_root) = self.beacon_root {
118            hasher.update(beacon_root);
119        }
120        // execution-apis#796 / glamsterdam-devnet-4: hash slot_number and
121        // gas_ceil so two FCUv4 calls that differ only in CL-supplied
122        // targetGasLimit (or in slot_number) do not collide on payload_id.
123        // Scoped to V4: for V1/V2/V3 gas_ceil is always the static
124        // --builder.gas-limit (no collision to disambiguate), so hashing it
125        // there would change those IDs without fixing anything.
126        if self.version >= 4 {
127            if let Some(slot_number) = self.slot_number {
128                hasher.update(slot_number.to_be_bytes());
129            }
130            hasher.update(self.gas_ceil.to_be_bytes());
131        }
132        let res = &mut hasher.finalize()[..8];
133        res[0] = self.version;
134        Ok(u64::from_be_bytes(res.try_into().map_err(|_| {
135            BuildPayloadArgsError::FailedToConvertPayload
136        })?))
137    }
138}
139
140/// Creates a new payload based on the payload arguments
141// Basic payload block building, can and should be improved
142pub fn create_payload(
143    args: &BuildPayloadArgs,
144    storage: &Store,
145    extra_data: Bytes,
146) -> Result<Block, ChainError> {
147    let parent_block = storage
148        .get_block_header_by_hash(args.parent)?
149        .ok_or_else(|| ChainError::ParentNotFound)?;
150    let chain_config = storage.get_chain_config();
151    let fork = chain_config.fork(args.timestamp);
152    let gas_limit = calc_gas_limit(parent_block.gas_limit, args.gas_ceil);
153    let excess_blob_gas = chain_config
154        .get_fork_blob_schedule(args.timestamp)
155        .map(|schedule| calc_excess_blob_gas(&parent_block, schedule, fork));
156
157    let header = BlockHeader {
158        parent_hash: args.parent,
159        ommers_hash: *DEFAULT_OMMERS_HASH,
160        coinbase: args.fee_recipient,
161        state_root: parent_block.state_root,
162        transactions_root: compute_transactions_root(&[], &NativeCrypto),
163        receipts_root: compute_receipts_root(&[], &NativeCrypto),
164        logs_bloom: Bloom::default(),
165        difficulty: U256::zero(),
166        number: parent_block.number.saturating_add(1),
167        gas_limit,
168        gas_used: 0,
169        timestamp: args.timestamp,
170        extra_data,
171        prev_randao: args.random,
172        nonce: 0,
173        base_fee_per_gas: calculate_base_fee_per_gas(
174            gas_limit,
175            parent_block.gas_limit,
176            parent_block.gas_used,
177            parent_block.base_fee_per_gas.unwrap_or_default(),
178            args.elasticity_multiplier,
179        ),
180        withdrawals_root: chain_config
181            .is_shanghai_activated(args.timestamp)
182            .then_some(compute_withdrawals_root(
183                args.withdrawals.as_ref().unwrap_or(&Vec::new()),
184                &NativeCrypto,
185            )),
186        blob_gas_used: chain_config
187            .is_cancun_activated(args.timestamp)
188            .then_some(0),
189        excess_blob_gas,
190        parent_beacon_block_root: args.beacon_root,
191        requests_hash: chain_config
192            .is_prague_activated(args.timestamp)
193            .then_some(*DEFAULT_REQUESTS_HASH),
194        slot_number: args.slot_number,
195        ..Default::default()
196    };
197
198    let body = BlockBody {
199        transactions: Vec::new(),
200        ommers: Vec::new(),
201        withdrawals: args.withdrawals.clone(),
202    };
203
204    // Delay applying withdrawals until the payload is requested and built
205    Ok(Block::new(header, body))
206}
207
208pub fn calc_gas_limit(parent_gas_limit: u64, builder_gas_ceil: u64) -> u64 {
209    let delta = parent_gas_limit / GAS_LIMIT_BOUND_DIVISOR - 1;
210    let mut limit = parent_gas_limit;
211    let desired_limit = max(builder_gas_ceil, MIN_GAS_LIMIT);
212    if limit < desired_limit {
213        limit = parent_gas_limit + delta;
214        if limit > desired_limit {
215            limit = desired_limit
216        }
217        return limit;
218    }
219    if limit > desired_limit {
220        limit = parent_gas_limit - delta;
221        if limit < desired_limit {
222            limit = desired_limit
223        }
224    }
225    limit
226}
227
228#[derive(Clone)]
229pub struct PayloadBuildContext {
230    pub payload: Block,
231    pub remaining_gas: u64,
232    /// Cumulative gas spent (post-refund) for receipt tracking.
233    /// Per EIP-7778 this differs from `remaining_gas` which tracks pre-refund gas.
234    pub cumulative_gas_spent: u64,
235    /// EIP-8037 (Amsterdam+): cumulative regular (non-state) gas used.
236    pub block_regular_gas_used: u64,
237    /// EIP-8037 (Amsterdam+): cumulative state gas used.
238    pub block_state_gas_used: u64,
239    /// Whether Amsterdam fork is active for this block.
240    pub is_amsterdam: bool,
241    pub receipts: Vec<Receipt>,
242    pub requests: Option<Vec<EncodedRequests>>,
243    pub block_value: U256,
244    base_fee_per_blob_gas: U256,
245    pub blobs_bundle: BlobsBundle,
246    pub store: Store,
247    pub vm: Evm,
248    pub account_updates: Vec<AccountUpdate>,
249    pub payload_size: u64,
250    /// Block Access List for EIP-7928
251    pub block_access_list: Option<BlockAccessList>,
252}
253
254impl PayloadBuildContext {
255    pub fn new(
256        payload: Block,
257        storage: &Store,
258        blockchain_type: &BlockchainType,
259    ) -> Result<Self, EvmError> {
260        let config = storage.get_chain_config();
261        let base_fee_per_blob_gas = calculate_base_fee_per_blob_gas(
262            payload.header.excess_blob_gas.unwrap_or_default(),
263            config
264                .get_fork_blob_schedule(payload.header.timestamp)
265                .map(|schedule| schedule.base_fee_update_fraction)
266                .unwrap_or_default(),
267        );
268
269        let parent_header = storage
270            .get_block_header_by_hash(payload.header.parent_hash)
271            .map_err(|e| EvmError::DB(e.to_string()))?
272            .ok_or_else(|| EvmError::DB("parent header not found".to_string()))?;
273        let vm_db = StoreVmDatabase::new(storage.clone(), parent_header)?;
274        let mut vm = new_evm(blockchain_type, vm_db)?;
275
276        // Enable BAL recording for Amsterdam and later forks (EIP-7928)
277        if config.is_amsterdam_activated(payload.header.timestamp) {
278            vm.enable_bal_recording();
279            // Set index 0 for pre-execution phase (system contracts)
280            vm.set_bal_index(0);
281        }
282
283        let is_amsterdam = config.is_amsterdam_activated(payload.header.timestamp);
284        let payload_size = payload.length() as u64;
285        Ok(PayloadBuildContext {
286            remaining_gas: payload.header.gas_limit,
287            cumulative_gas_spent: 0,
288            block_regular_gas_used: 0,
289            block_state_gas_used: 0,
290            is_amsterdam,
291            receipts: vec![],
292            requests: config
293                .is_prague_activated(payload.header.timestamp)
294                .then_some(Vec::new()),
295            block_value: U256::zero(),
296            base_fee_per_blob_gas,
297            payload,
298            blobs_bundle: BlobsBundle::default(),
299            store: storage.clone(),
300            vm,
301            account_updates: Vec::new(),
302            payload_size,
303            block_access_list: None,
304        })
305    }
306
307    pub fn gas_used(&self) -> u64 {
308        if self.is_amsterdam {
309            // EIP-8037: block gas = max(sum_regular, sum_state)
310            self.block_regular_gas_used.max(self.block_state_gas_used)
311        } else {
312            self.payload.header.gas_limit - self.remaining_gas
313        }
314    }
315}
316
317impl PayloadBuildContext {
318    fn parent_hash(&self) -> BlockHash {
319        self.payload.header.parent_hash
320    }
321
322    pub fn block_number(&self) -> BlockNumber {
323        self.payload.header.number
324    }
325
326    fn chain_config(&self) -> ChainConfig {
327        self.store.get_chain_config()
328    }
329
330    fn base_fee_per_gas(&self) -> Option<u64> {
331        self.payload.header.base_fee_per_gas
332    }
333}
334
335#[derive(Debug, Clone)]
336pub struct PayloadBuildResult {
337    pub blobs_bundle: BlobsBundle,
338    pub block_value: U256,
339    pub receipts: Vec<Receipt>,
340    pub requests: Vec<EncodedRequests>,
341    pub account_updates: Vec<AccountUpdate>,
342    pub payload: Block,
343    /// Block Access List for EIP-7928
344    pub block_access_list: Option<BlockAccessList>,
345}
346
347impl From<PayloadBuildContext> for PayloadBuildResult {
348    fn from(value: PayloadBuildContext) -> Self {
349        let PayloadBuildContext {
350            blobs_bundle,
351            block_value,
352            requests,
353            receipts,
354            account_updates,
355            payload,
356            block_access_list,
357            ..
358        } = value;
359
360        Self {
361            blobs_bundle,
362            block_value,
363            requests: requests.unwrap_or_default(),
364            receipts,
365            account_updates,
366            payload,
367            block_access_list,
368        }
369    }
370}
371
372impl Blockchain {
373    /// Attempts to fetch a payload given it's id. If the payload is still being built, it will be finished.
374    /// Fails if there is no payload or active payload build task for the given id.
375    pub async fn get_payload(&self, payload_id: u64) -> Result<PayloadBuildResult, ChainError> {
376        let mut payloads = self.payloads.lock().await;
377        // Find the given payload and finish the active build process if needed
378        let idx = payloads
379            .iter()
380            .position(|(id, _)| id == &payload_id)
381            .ok_or(ChainError::UnknownPayload)?;
382        let finished_payload = (payload_id, payloads.remove(idx).1.to_payload().await?);
383        payloads.insert(idx, finished_payload);
384        // Return the held payload
385        match &payloads[idx].1 {
386            PayloadOrTask::Payload(payload) => Ok(*payload.clone()),
387            _ => unreachable!("we already converted the payload into a finished version"),
388        }
389    }
390
391    /// Starts a payload build process. The built payload can be retrieved by calling `get_payload`.
392    /// The build process will run for the full block building timeslot or until `get_payload` is called
393    pub async fn initiate_payload_build(self: Arc<Blockchain>, payload: Block, payload_id: u64) {
394        let self_clone = self.clone();
395        let cancel_token = CancellationToken::new();
396        let cancel_token_clone = cancel_token.clone();
397        let payload_build_task = tokio::task::spawn(async move {
398            self_clone
399                .build_payload_loop(payload, cancel_token_clone)
400                .await
401        });
402        let mut payloads = self.payloads.lock().await;
403        if payloads.len() >= MAX_PAYLOADS {
404            // Remove oldest unclaimed payload
405            payloads.remove(0);
406        }
407        payloads.push((
408            payload_id,
409            PayloadOrTask::Task(PayloadBuildTask {
410                task: payload_build_task,
411                cancel: cancel_token,
412            }),
413        ));
414    }
415
416    /// Build the given payload and keep on rebuilding it until either the time slot
417    /// given by `SECONDS_PER_SLOT` is up or the `cancel_token` is cancelled
418    pub async fn build_payload_loop(
419        self: Arc<Blockchain>,
420        payload: Block,
421        cancel_token: CancellationToken,
422    ) -> Result<PayloadBuildResult, ChainError> {
423        let start = Instant::now();
424        const SECONDS_PER_SLOT: Duration = Duration::from_secs(12);
425        // Attempt to rebuild the payload as many times within the given timeframe to maximize fee revenue
426        // TODO(#4997): start with an empty block
427        // Snapshot the mempool sequence *before* the build so any tx that lands
428        // during the build is seen as newer than the current `res`.
429        let mut last_built_seq = self.mempool.tx_seq();
430        let mut res = self.build_payload(payload.clone())?;
431        while start.elapsed() < SECONDS_PER_SLOT && !cancel_token.is_cancelled() {
432            // Wait for new transactions, cancellation, or slot deadline before rebuilding
433            let remaining = SECONDS_PER_SLOT.saturating_sub(start.elapsed());
434            let notified = self.mempool.tx_added().notified();
435            tokio::select! {
436                _ = notified => {}
437                _ = cancel_token.cancelled() => break,
438                _ = tokio::time::sleep(remaining) => break,
439            }
440            let payload = payload.clone();
441            let self_clone = self.clone();
442            let seq_before = self.mempool.tx_seq();
443            let building_task =
444                tokio::task::spawn_blocking(move || self_clone.build_payload(payload));
445            // Cancel the current build process and return the previous payload if it is requested earlier
446            // TODO(#5011): this doesn't stop the building task, but only keeps it running in the background,
447            //   which wastes CPU resources.
448            match cancel_token.run_until_cancelled(building_task).await {
449                Some(Ok(current_res)) => {
450                    res = current_res?;
451                    last_built_seq = seq_before;
452                }
453                Some(Err(err)) => {
454                    warn!(%err, "Payload-building task panicked");
455                }
456                None => {}
457            }
458        }
459
460        // If a tx landed after the snapshot that produced `res`, do one final
461        // build before returning. Covers both races: (a) cancellation dropping
462        // an in-progress rebuild via `run_until_cancelled`, and (b) the slot-
463        // timeout `select!` arm winning over a simultaneous `tx_added`
464        // notification near the slot boundary.
465        if self.mempool.tx_seq() > last_built_seq {
466            let blockchain = self.clone();
467            match tokio::task::spawn_blocking(move || blockchain.build_payload(payload)).await {
468                Ok(Ok(final_res)) => res = final_res,
469                Ok(Err(err)) => {
470                    warn!(%err, "Final payload rebuild failed; returning previous result")
471                }
472                Err(err) => warn!(%err, "Final payload rebuild task panicked"),
473            }
474        }
475
476        Ok(res)
477    }
478
479    /// Completes the payload building process, return the block value
480    pub fn build_payload(&self, payload: Block) -> Result<PayloadBuildResult, ChainError> {
481        let since = Instant::now();
482
483        debug!("Building payload");
484        let base_fee = payload.header.base_fee_per_gas.unwrap_or_default();
485        let mut context = PayloadBuildContext::new(payload, &self.storage, &self.options.r#type)?;
486
487        if let BlockchainType::L1 = self.options.r#type {
488            self.apply_system_operations(&mut context)?;
489        }
490        self.fill_transactions(&mut context)?;
491        // EIP-7928: Post-tx phase uses index n+1 for both requests and withdrawals.
492        // Order must match geth: requests (system calls) BEFORE withdrawals.
493        if context
494            .chain_config()
495            .is_amsterdam_activated(context.payload.header.timestamp)
496        {
497            let post_tx_index =
498                u32::try_from(context.payload.body.transactions.len() + 1).unwrap_or(u32::MAX);
499            context.vm.set_bal_index(post_tx_index);
500            // Record withdrawal recipients as touched addresses per EIP-7928
501            if let Some(recorder) = context.vm.db.bal_recorder_mut()
502                && let Some(withdrawals) = &context.payload.body.withdrawals
503            {
504                recorder.extend_touched_addresses(withdrawals.iter().map(|w| w.address));
505            }
506        }
507        self.extract_requests(&mut context)?;
508        self.apply_withdrawals(&mut context)?;
509        self.finalize_payload(&mut context)?;
510
511        let interval = Instant::now().duration_since(since).as_millis();
512
513        tracing::debug!(
514            "[METRIC] BUILDING PAYLOAD TOOK: {interval} ms, base fee {}",
515            base_fee
516        );
517        metrics!(METRICS_BLOCKS.set_block_building_ms(interval as i64));
518        metrics!(METRICS_BLOCKS.set_block_building_base_fee(base_fee as i64));
519        let gas_used = context.gas_used();
520        if gas_used > 0 {
521            let as_gigas = (gas_used as f64).div(10_f64.powf(9_f64));
522
523            if interval != 0 {
524                let throughput = (as_gigas) / (interval as f64) * 1000_f64;
525                metrics!(METRICS_BLOCKS.set_latest_gigagas_block_building(throughput));
526
527                tracing::debug!(
528                    "[METRIC] BLOCK BUILDING THROUGHPUT: {throughput} Gigagas/s TIME SPENT: {interval} msecs"
529                );
530            }
531        }
532
533        Ok(context.into())
534    }
535
536    pub fn apply_withdrawals(&self, context: &mut PayloadBuildContext) -> Result<(), EvmError> {
537        let binding = Vec::new();
538        let withdrawals = context
539            .payload
540            .body
541            .withdrawals
542            .as_ref()
543            .unwrap_or(&binding);
544        context.vm.process_withdrawals(withdrawals)
545    }
546
547    // This function applies system level operations:
548    // - Call beacon root contract, and obtain the new state root
549    // - Call block hash process contract, and store parent block hash
550    pub fn apply_system_operations(
551        &self,
552        context: &mut PayloadBuildContext,
553    ) -> Result<(), EvmError> {
554        context.vm.apply_system_calls(&context.payload.header)
555    }
556
557    /// Fetches suitable transactions from the mempool
558    /// Returns two transaction queues, one for plain and one for blob txs
559    pub fn fetch_mempool_transactions(
560        &self,
561        context: &mut PayloadBuildContext,
562    ) -> Result<(TransactionQueue, TransactionQueue), ChainError> {
563        let blob_fee: u64 = context.base_fee_per_blob_gas.try_into().map_err(|_| {
564            ChainError::Custom("base_fee_per_blob_gas does not fit in u64".to_owned())
565        })?;
566        let tx_filter = PendingTxFilter {
567            /*TODO(https://github.com/lambdaclass/ethrex/issues/680): add tip filter */
568            base_fee: context.base_fee_per_gas(),
569            blob_fee: Some(blob_fee),
570            ..Default::default()
571        };
572        let plain_tx_filter = PendingTxFilter {
573            only_plain_txs: true,
574            ..tx_filter
575        };
576        let blob_tx_filter = PendingTxFilter {
577            only_blob_txs: true,
578            ..tx_filter
579        };
580        Ok((
581            // Plain txs
582            TransactionQueue::new(
583                self.mempool.filter_transactions(&plain_tx_filter)?,
584                context.base_fee_per_gas(),
585            )?,
586            // Blob txs
587            TransactionQueue::new(
588                self.mempool.filter_transactions(&blob_tx_filter)?,
589                context.base_fee_per_gas(),
590            )?,
591        ))
592    }
593
594    /// EIP-7872: Computes effective max blobs per block.
595    /// Returns min(protocol_max, user_configured_max).
596    fn effective_max_blobs(&self, context: &PayloadBuildContext) -> usize {
597        let protocol_max = context
598            .chain_config()
599            .get_fork_blob_schedule(context.payload.header.timestamp)
600            .map(|schedule| schedule.max)
601            .unwrap_or_default();
602        match self.options.max_blobs_per_block {
603            Some(user_max) => protocol_max.min(user_max) as usize,
604            None => protocol_max as usize,
605        }
606    }
607
608    /// Fills the payload with transactions taken from the mempool
609    /// Returns the block value
610    pub fn fill_transactions(&self, context: &mut PayloadBuildContext) -> Result<(), ChainError> {
611        let chain_config = context.chain_config();
612        let max_blob_number_per_block = self.effective_max_blobs(context);
613
614        debug!("Fetching transactions from mempool");
615        // Fetch mempool transactions
616        let (mut plain_txs, mut blob_txs) = self.fetch_mempool_transactions(context)?;
617        // Execute and add transactions to payload (if suitable)
618        loop {
619            // Check if we have enough gas to run more transactions
620            if context.remaining_gas < TX_GAS_COST {
621                debug!("No more gas to run transactions");
622                break;
623            };
624            if !blob_txs.is_empty() && context.blobs_bundle.blobs.len() >= max_blob_number_per_block
625            {
626                debug!("No more blob gas to run blob transactions");
627                blob_txs.clear();
628            }
629            // Fetch the next transactions
630            let (head_tx, is_blob) = match (plain_txs.peek(), blob_txs.peek()) {
631                (None, None) => break,
632                (None, Some(tx)) => (tx, true),
633                (Some(tx), None) => (tx, false),
634                (Some(a), Some(b)) if b < a => (b, true),
635                (Some(tx), _) => (tx, false),
636            };
637
638            let txs = if is_blob {
639                &mut blob_txs
640            } else {
641                &mut plain_txs
642            };
643
644            // Check if we have enough gas to run the transaction.
645            // EIP-7825/EIP-8037: for Amsterdam, cap at TX_MAX_GAS_LIMIT since
646            // remaining_gas tracks regular gas only.
647            let tx_gas_reservation = if context.is_amsterdam {
648                head_tx.tx.gas_limit().min(TX_MAX_GAS_LIMIT_AMSTERDAM)
649            } else {
650                head_tx.tx.gas_limit()
651            };
652            if context.remaining_gas < tx_gas_reservation {
653                debug!(
654                    "Skipping transaction: {}, no gas left",
655                    head_tx.tx.hash(&NativeCrypto)
656                );
657                // We don't have enough gas left for the transaction, so we skip all txs from this account
658                txs.pop();
659                continue;
660            }
661
662            // Check adding a transaction wouldn't exceed the Osaka block size limit of 10 MiB
663            // if inclusion of the transaction puts the block size over the size limit
664            // we don't add any more txs to the payload.
665            let potential_rlp_block_size =
666                context.payload_size + head_tx.encode_canonical_to_vec().len() as u64;
667            if context
668                .chain_config()
669                .is_osaka_activated(context.payload.header.timestamp)
670                && potential_rlp_block_size > MAX_RLP_BLOCK_SIZE
671            {
672                break;
673            }
674            context.payload_size = potential_rlp_block_size;
675
676            // TODO: maybe fetch hash too when filtering mempool so we don't have to compute it here (we can do this in the same refactor as adding timestamp)
677            let tx_hash = head_tx.tx.hash(&NativeCrypto);
678
679            // Check whether the tx is replay-protected
680            if head_tx.tx.protected() && !chain_config.is_eip155_activated(context.block_number()) {
681                // Ignore replay protected tx & all txs from the sender
682                // Pull transaction from the mempool
683                debug!("Ignoring replay-protected transaction: {}", tx_hash);
684                txs.pop();
685                self.remove_transaction_from_pool(&tx_hash)?;
686                continue;
687            }
688
689            match self.apply_tx_to_payload(head_tx, context) {
690                Ok(()) => txs.shift()?,
691                Err(_) => txs.pop(),
692            }
693        }
694        Ok(())
695    }
696
697    /// Apply a single transaction to the in-progress payload.
698    ///
699    /// Runs the full per-tx pipeline: EIP-8037 2D inclusion check, EIP-7928
700    /// BAL index/checkpoint setup, sender/recipient recording, dispatch to
701    /// blob/plain execution, and on failure rolls the BAL recorder back so
702    /// rejected txs leave no trace. On success the tx is appended to the
703    /// payload body and the receipt to `context.receipts`.
704    ///
705    /// Caller is responsible for mempool bookkeeping (advancing or dropping
706    /// the sender's queue) — this function only mutates the payload context.
707    pub fn apply_tx_to_payload(
708        &self,
709        head: HeadTransaction,
710        context: &mut PayloadBuildContext,
711    ) -> Result<(), ChainError> {
712        let tx_hash = head.tx.hash(&NativeCrypto);
713
714        // EIP-8037 (Amsterdam+, PR #2703): per-tx 2D inclusion check against
715        // running block totals. Run BEFORE we touch the BAL recorder so a
716        // rejected tx doesn't even produce a sender/recipient touch.
717        if context.is_amsterdam
718            && let Err(e) = check_2d_gas_allowance(
719                &head.tx,
720                context.block_regular_gas_used,
721                context.block_state_gas_used,
722                context.payload.header.gas_limit,
723            )
724        {
725            debug!("Skipping tx {tx_hash:x}: fails 2D inclusion check: {e}");
726            return Err(e.into());
727        }
728
729        // Set BAL index for this transaction (1-indexed per EIP-7928).
730        // Must happen BEFORE tx_checkpoint: set_bal_index flushes net-zero
731        // filters for the previous (committed) tx, which may insert reads.
732        let tx_index =
733            u32::try_from(context.payload.body.transactions.len() + 1).unwrap_or(u32::MAX);
734        context.vm.set_bal_index(tx_index);
735
736        // EIP-7928: lightweight tx-level checkpoint before trying the tx.
737        // If the tx is rejected, restore so only included txs affect the BAL.
738        // Taken after set_bal_index (which flushes previous tx) but before
739        // this tx's touches, so rejected txs leave no trace.
740        let bal_checkpoint = context
741            .vm
742            .db
743            .bal_recorder
744            .as_ref()
745            .map(|r| r.tx_checkpoint());
746
747        if let Some(recorder) = context.vm.db.bal_recorder_mut() {
748            recorder.record_touched_address(head.tx.sender());
749            if let TxKind::Call(to) = head.to() {
750                recorder.record_touched_address(to);
751            }
752        }
753
754        let receipt = match self.apply_transaction(&head, context) {
755            Ok(receipt) => {
756                metrics!(METRICS_TX.inc_tx_with_type(MetricsTxType(head.tx_type())));
757                receipt
758            }
759            Err(e) => {
760                debug!("Failed to execute transaction: {tx_hash:x}, {e}");
761                metrics!(METRICS_TX.inc_tx_errors(e.to_metric()));
762                if let (Some(recorder), Some(checkpoint)) =
763                    (context.vm.db.bal_recorder_mut(), bal_checkpoint)
764                {
765                    recorder.tx_restore(checkpoint);
766                }
767                return Err(e);
768            }
769        };
770
771        debug!("Adding transaction: {} to payload", tx_hash);
772        context.payload.body.transactions.push(head.into());
773        context.receipts.push(receipt);
774        Ok(())
775    }
776
777    /// Executes the transaction, updates gas-related context values & return the receipt
778    /// The payload build context should have enough remaining gas to cover the transaction's gas_limit
779    fn apply_transaction(
780        &self,
781        head: &HeadTransaction,
782        context: &mut PayloadBuildContext,
783    ) -> Result<Receipt, ChainError> {
784        match **head {
785            Transaction::EIP4844Transaction(_) => self.apply_blob_transaction(head, context),
786            _ => apply_plain_transaction(head, context),
787        }
788    }
789
790    /// Runs a blob transaction, updates the gas count & blob data and returns the receipt
791    fn apply_blob_transaction(
792        &self,
793        head: &HeadTransaction,
794        context: &mut PayloadBuildContext,
795    ) -> Result<Receipt, ChainError> {
796        // Fetch blobs bundle
797        let tx_hash = head.tx.hash(&NativeCrypto);
798        let max_blob_number_per_block = self.effective_max_blobs(context);
799        let Some(blobs_bundle) = self.mempool.get_blobs_bundle(tx_hash)? else {
800            // No blob tx should enter the mempool without its blobs bundle so this is an internal error
801            return Err(
802                StoreError::Custom(format!("No blobs bundle found for blob tx {tx_hash}")).into(),
803            );
804        };
805        if context.blobs_bundle.blobs.len() + blobs_bundle.blobs.len() > max_blob_number_per_block {
806            // This error will only be used for debug tracing
807            return Err(EvmError::Custom("max data blobs reached".to_string()).into());
808        };
809        // Apply transaction
810        let receipt = apply_plain_transaction(head, context)?;
811        // Update context with blob data
812        let prev_blob_gas = context.payload.header.blob_gas_used.unwrap_or_default();
813        context.payload.header.blob_gas_used =
814            Some(prev_blob_gas + (blobs_bundle.blobs.len() * GAS_PER_BLOB as usize) as u64);
815        context.blobs_bundle += blobs_bundle;
816        Ok(receipt)
817    }
818
819    pub fn extract_requests(&self, context: &mut PayloadBuildContext) -> Result<(), EvmError> {
820        if !context
821            .chain_config()
822            .is_prague_activated(context.payload.header.timestamp)
823        {
824            return Ok(());
825        };
826
827        let requests = context
828            .vm
829            .extract_requests(&context.receipts, &context.payload.header)?;
830
831        context.requests = Some(requests.iter().map(|r| r.encode()).collect());
832
833        Ok(())
834    }
835
836    pub fn finalize_payload(&self, context: &mut PayloadBuildContext) -> Result<(), ChainError> {
837        // Take BAL from VM before getting state transitions (which clears state)
838        let block_access_list = context.vm.take_bal();
839
840        let account_updates = context.vm.get_state_transitions()?;
841
842        let ret_acount_updates_list = self
843            .storage
844            .apply_account_updates_batch(context.parent_hash(), &account_updates)?
845            .ok_or(ChainError::ParentStateNotFound)?;
846
847        let state_root = ret_acount_updates_list.state_trie_hash;
848
849        context.payload.header.state_root = state_root;
850        context.payload.header.transactions_root =
851            compute_transactions_root(&context.payload.body.transactions, &NativeCrypto);
852        context.payload.header.receipts_root =
853            compute_receipts_root(&context.receipts, &NativeCrypto);
854        context.payload.header.requests_hash = context
855            .requests
856            .as_ref()
857            .map(|requests| compute_requests_hash(requests));
858        let gas_used = context.gas_used();
859        if context.is_amsterdam {
860            debug!(
861                "EIP-8037 block finalize: gas_used={gas_used} regular={} state={} txs={}",
862                context.block_regular_gas_used,
863                context.block_state_gas_used,
864                context.payload.body.transactions.len(),
865            );
866        }
867        context.payload.header.gas_used = gas_used;
868        context.account_updates = account_updates;
869
870        // Set BAL hash in block header (EIP-7928)
871        context.payload.header.block_access_list_hash = block_access_list
872            .as_ref()
873            .map(|bal| bal.compute_hash(&NativeCrypto));
874        context.block_access_list = block_access_list;
875
876        let mut logs = vec![];
877        for receipt in context.receipts.iter().cloned() {
878            for log in receipt.logs {
879                logs.push(log);
880            }
881        }
882
883        context.payload.header.logs_bloom = bloom_from_logs(&logs, &NativeCrypto);
884        Ok(())
885    }
886}
887
888/// Runs a plain (non blob) transaction, updates the gas count and returns the receipt
889pub fn apply_plain_transaction(
890    head: &HeadTransaction,
891    context: &mut PayloadBuildContext,
892) -> Result<Receipt, ChainError> {
893    let (receipt, report) = context.vm.execute_tx(
894        &head.tx,
895        &context.payload.header,
896        &mut context.cumulative_gas_spent,
897        head.tx.sender(),
898    )?;
899
900    // EIP-8037 (Amsterdam+): track regular and state gas separately
901    let tx_state_gas = report.state_gas_used;
902    let tx_regular_gas = report.gas_used.saturating_sub(tx_state_gas);
903
904    // Compute new totals before committing them
905    let new_regular = context
906        .block_regular_gas_used
907        .saturating_add(tx_regular_gas);
908    let new_state = context.block_state_gas_used.saturating_add(tx_state_gas);
909
910    // EIP-8037 (Amsterdam+): post-execution block gas overflow check
911    // Reject the transaction if adding it would cause max(regular, state) to exceed the gas limit
912    if context.is_amsterdam && new_regular.max(new_state) > context.payload.header.gas_limit {
913        // Rollback transaction state before returning error:
914        // 1. Undo DB mutations (nonce, balance, storage, etc.)
915        // 2. Revert cumulative gas counter inflation
916        // This ensures the next transaction executes against clean state.
917        context.vm.undo_last_tx()?;
918        // `cumulative_gas_spent` was bumped inside `execute_tx` above; revert it
919        // now that the tx is being rejected. Use `saturating_sub` as a defensive
920        // guard — cumulative must always dominate this tx's contribution unless
921        // some upstream bug leaks a stale value, in which case we'd rather clamp
922        // to 0 than underflow the counter.
923        debug_assert!(
924            context.cumulative_gas_spent >= report.gas_spent,
925            "cumulative_gas_spent underflow on tx rollback"
926        );
927        context.cumulative_gas_spent = context
928            .cumulative_gas_spent
929            .saturating_sub(report.gas_spent);
930
931        return Err(EvmError::Custom(format!(
932            "block gas limit exceeded (state gas overflow): \
933             max({new_regular}, {new_state}) = {} > gas_limit {}",
934            new_regular.max(new_state),
935            context.payload.header.gas_limit
936        ))
937        .into());
938    }
939
940    // Commit the new totals
941    context.block_regular_gas_used = new_regular;
942    context.block_state_gas_used = new_state;
943
944    if context.is_amsterdam {
945        debug!(
946            "EIP-8037 tx gas: regular={tx_regular_gas} state={tx_state_gas} gas_used={} gas_spent={} block_regular={} block_state={} block_max={}",
947            report.gas_used,
948            report.gas_spent,
949            context.block_regular_gas_used,
950            context.block_state_gas_used,
951            context
952                .block_regular_gas_used
953                .max(context.block_state_gas_used),
954        );
955    }
956
957    // Update remaining_gas for block gas limit checks.
958    // EIP-8037 (Amsterdam+): remaining_gas reflects both regular and state gas dimensions.
959    // For pre-tx heuristic checks, this ensures we reject txs when either dimension is full.
960    if context.is_amsterdam {
961        context.remaining_gas = context
962            .payload
963            .header
964            .gas_limit
965            .saturating_sub(new_regular.max(new_state));
966    } else {
967        context.remaining_gas = context.remaining_gas.saturating_sub(report.gas_used);
968    }
969
970    // Block value uses gas_spent (what the user actually pays) for tip calculation
971    context.block_value += U256::from(report.gas_spent) * head.tip;
972    Ok(receipt)
973}
974
975/// A struct representing suitable mempool transactions waiting to be included in a block
976// TODO: Consider using VecDequeue instead of Vec
977pub struct TransactionQueue {
978    // The first transaction for each account along with its tip, sorted by highest tip
979    heads: Vec<HeadTransaction>,
980    // The remaining txs grouped by account and sorted by nonce
981    txs: FxHashMap<Address, Vec<MempoolTransaction>>,
982    // Base Fee stored for tip calculations
983    base_fee: Option<u64>,
984}
985
986#[derive(Clone, Debug, Eq, PartialEq)]
987pub struct HeadTransaction {
988    pub tx: MempoolTransaction,
989    pub tip: U256,
990}
991
992impl std::ops::Deref for HeadTransaction {
993    type Target = Transaction;
994
995    fn deref(&self) -> &Self::Target {
996        &self.tx
997    }
998}
999
1000impl From<HeadTransaction> for Transaction {
1001    fn from(val: HeadTransaction) -> Self {
1002        val.tx.transaction().clone()
1003    }
1004}
1005
1006impl TransactionQueue {
1007    /// Creates a new TransactionQueue from a set of transactions grouped by sender and sorted by nonce
1008    fn new(
1009        mut txs: FxHashMap<Address, Vec<MempoolTransaction>>,
1010        base_fee: Option<u64>,
1011    ) -> Result<Self, ChainError> {
1012        let mut heads = Vec::with_capacity(100);
1013        for (_, txs) in txs.iter_mut() {
1014            // Pull the first tx from each list and add it to the heads list
1015            // This should be a newly filtered tx list so we are guaranteed to have a first element
1016            let head_tx = txs.remove(0);
1017            heads.push(HeadTransaction {
1018                // We already ran this method when filtering the transactions from the mempool so it shouldn't fail
1019                tip: head_tx
1020                    .effective_gas_tip(base_fee)
1021                    .ok_or(ChainError::InvalidBlock(
1022                        InvalidBlockError::InvalidTransaction("Attempted to add an invalid transaction to the block. The transaction filter must have failed.".to_owned()),
1023                    ))?,
1024                tx: head_tx,
1025            });
1026        }
1027        // Sort heads by higest tip (and lowest timestamp if tip is equal)
1028        heads.sort();
1029        Ok(TransactionQueue {
1030            heads,
1031            txs,
1032            base_fee,
1033        })
1034    }
1035
1036    /// Remove all transactions from the queue
1037    pub fn clear(&mut self) {
1038        self.heads.clear();
1039        self.txs.clear();
1040    }
1041
1042    /// Returns true if there are no more transactions in the queue
1043    pub fn is_empty(&self) -> bool {
1044        self.heads.is_empty()
1045    }
1046
1047    /// Returns the head transaction with the highest tip
1048    /// If there is more than one transaction with the highest tip, return the one with the lowest timestamp
1049    pub fn peek(&self) -> Option<HeadTransaction> {
1050        self.heads.first().cloned()
1051    }
1052
1053    /// Removes current head transaction and all transactions from the given sender
1054    pub fn pop(&mut self) {
1055        if !self.is_empty() {
1056            let sender = self.heads.remove(0).tx.sender();
1057            self.txs.remove(&sender);
1058        }
1059    }
1060
1061    /// Remove the top transaction
1062    /// Add a tx from the same sender to the head transactions
1063    pub fn shift(&mut self) -> Result<(), ChainError> {
1064        let tx = self.heads.remove(0);
1065        if let Some(txs) = self.txs.get_mut(&tx.tx.sender()) {
1066            // Fetch next head
1067            if !txs.is_empty() {
1068                let head_tx = txs.remove(0);
1069                let head = HeadTransaction {
1070                    // We already ran this method when filtering the transactions from the mempool so it shouldn't fail
1071                    tip: head_tx.effective_gas_tip(self.base_fee).ok_or(
1072                        ChainError::InvalidBlock(
1073                            InvalidBlockError::InvalidTransaction("Attempted to add an invalid transaction to the block. The transaction filter must have failed.".to_owned()),
1074                        ),
1075                    )?,
1076                    tx: head_tx,
1077                };
1078                // Insert head into heads list while maintaing order
1079                let index = match self.heads.binary_search(&head) {
1080                    Ok(index) => index, // Same ordering shouldn't be possible when adding timestamps
1081                    Err(index) => index,
1082                };
1083                self.heads.insert(index, head);
1084            }
1085        }
1086        Ok(())
1087    }
1088}
1089
1090// Orders transactions by highest tip, if tip is equal, orders by lowest timestamp
1091impl Ord for HeadTransaction {
1092    fn cmp(&self, other: &Self) -> Ordering {
1093        match (self.tx_type(), other.tx_type()) {
1094            (TxType::Privileged, TxType::Privileged) => return self.nonce().cmp(&other.nonce()),
1095            (TxType::Privileged, _) => return Ordering::Less,
1096            (_, TxType::Privileged) => return Ordering::Greater,
1097            _ => (),
1098        };
1099        match other.tip.cmp(&self.tip) {
1100            Ordering::Equal => self.tx.time().cmp(&other.tx.time()),
1101            ordering => ordering,
1102        }
1103    }
1104}
1105
1106impl PartialOrd for HeadTransaction {
1107    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
1108        Some(self.cmp(other))
1109    }
1110}
ethrex_blockchain/payload.rs

ethrex_blockchain/
payload.rs
