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

ethrex_blockchain/
payload.rs
