From 6dc80e45655ee9cf3c066d5ed6a0a0eb439487b2 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 2 Jul 2026 21:59:24 +0200 Subject: [PATCH 1/2] execution/exec: dedup execAATxn into TxTask.executeAA HistoricalTraceWorker.execAATxn duplicated the AA-bundle validation and execution logic of TxTask.executeAA line for line. Replace the copy with a call to executeAA, keeping the TraceFroms/TraceTos filling (success path only, as before) at the historical call site. Behavior-preserving except one log-only delta: the historical copy's 'validated AA bundle' log printed endIdx-startIdx+1 while executeAA still prints startIdx-endIdx; the shared site now wins. That expression is fixed separately (yperbasis/dedup-audit-fixes, b3b25f8f99), which after this dedup fixes both callers at once. --- execution/exec/historical_trace_worker.go | 69 +---------------------- 1 file changed, 3 insertions(+), 66 deletions(-) diff --git a/execution/exec/historical_trace_worker.go b/execution/exec/historical_trace_worker.go index 31f5f015e50..3af12782710 100644 --- a/execution/exec/historical_trace_worker.go +++ b/execution/exec/historical_trace_worker.go @@ -37,7 +37,6 @@ import ( "github.com/erigontech/erigon/db/services" "github.com/erigontech/erigon/execution/chain" "github.com/erigontech/erigon/execution/protocol" - "github.com/erigontech/erigon/execution/protocol/aa" "github.com/erigontech/erigon/execution/protocol/rules" "github.com/erigontech/erigon/execution/state" "github.com/erigontech/erigon/execution/state/genesiswrite" @@ -255,75 +254,13 @@ func (rw *HistoricalTraceWorker) RunTxTask(txTask *TxTask) *TxResult { } func (rw *HistoricalTraceWorker) execAATxn(txTask *TxTask, tracer *calltracer.CallTracer) *TxResult { - result := &TxResult{} - - if !txTask.InBatch { - // this is the first transaction in an AA transaction batch, run all validation frames, then execute execution frames in its own txtask - startIdx := uint64(txTask.TxIndex) - endIdx := startIdx + txTask.AAValidationBatchSize - - validationResults := make([]AAValidationResult, txTask.AAValidationBatchSize+1) - log.Info("🕵️‍♂️[aa] found AA bundle", "startIdx", startIdx, "endIdx", endIdx) - - var outerErr error - for i := startIdx; i <= endIdx; i++ { - // check if next n transactions are AA transactions and run validation - if txTask.Txs[i].Type() == types.AccountAbstractionTxType { - aaTxn, ok := txTask.Txs[i].(*types.AccountAbstractionTransaction) - if !ok { - outerErr = fmt.Errorf("invalid transaction type, expected AccountAbstractionTx, got %T", txTask.Tx) - break - } - - paymasterContext, validationGasUsed, err := aa.ValidateAATransaction(aaTxn, rw.ibs, rw.taskGasPool, txTask.Header, rw.evm, rw.execArgs.ChainConfig) - if err != nil { - outerErr = err - break - } - - validationResults[i-startIdx] = AAValidationResult{ - PaymasterContext: paymasterContext, - GasUsed: validationGasUsed, - } - } else { - outerErr = fmt.Errorf("invalid txcount, expected txn %d to be type %d", i, types.AccountAbstractionTxType) - break - } - } - - if outerErr != nil { - result.Err = outerErr - return result - } - log.Info("✅[aa] validated AA bundle", "len", endIdx-startIdx+1) - - result.ValidationResults = validationResults - } - - if len(result.ValidationResults) == 0 { - result.Err = fmt.Errorf("found RIP-7560 but no remaining validation results, txIndex %d", txTask.TxIndex) + aaTxn := txTask.Tx().(*types.AccountAbstractionTransaction) // type cast checked by the caller + result := txTask.executeAA(aaTxn, rw.evm, rw.taskGasPool, rw.ibs, rw.execArgs.ChainConfig) + if result.Err != nil { return result } - - aaTxn := txTask.Tx().(*types.AccountAbstractionTransaction) // type cast checked earlier - validationRes := result.ValidationResults[0] - result.ValidationResults = result.ValidationResults[1:] - - status, gasUsed, err := aa.ExecuteAATransaction(aaTxn, validationRes.PaymasterContext, validationRes.GasUsed, rw.taskGasPool, rw.evm, txTask.Header, rw.ibs) - if err != nil { - result.Err = err - return result - } - - result.ExecutionResult.ReceiptGasUsed = gasUsed - result.ExecutionResult.BlockRegularGasUsed = gasUsed - // Update the state with pending changes - rw.ibs.SoftFinalise() - result.Logs = rw.ibs.GetLogs(txTask.TxIndex, txTask.TxHash(), txTask.BlockNumber(), txTask.BlockHash()) result.TraceFroms = tracer.Froms() result.TraceTos = tracer.Tos() - - log.Info("🚀[aa] executed AA bundle transaction", "txIndex", txTask.TxIndex, "status", status) return result } From 4551188a75412480afd45aaf9f3e87fc78769e54 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 2 Jul 2026 22:04:45 +0200 Subject: [PATCH 2/2] execution/stagedsync: extract shared buildBlockTasks helper serialExecutor.exec and txExecutor.executeBlocks duplicated the per-block ingest sequence: read-ahead ping, canonical-hash lookup, block fetch with readAheader fallback, EVM block context construction, and the TxTask build loop. Extract txExecutor.buildBlockTasks and call it from both. Deliberate deltas parameterized or kept at the call sites: - getHash fn: serial passes its mutex-wrapped se.getHeader; parallel passes the per-worker-overridden placeholder reading via blockTx. - BlockStateCache: parallel passes a fresh per-block cache; serial nil. - BAL read/decode/validate stays parallel-only, now after the helper. - Amsterdam guard and accumulator.StartChange stay serial-only. - Snapshot step-misalignment check stays parallel-only, untouched. The helper skips already-executed txns before constructing the task (parallel's form; serial previously constructed then discarded, with no side effects from the discarded task) and reports partialBlock, which serial folds into havePartialBlock. Serial tasks now carry Config and Engine from construction; executeBlock reassigns the same values before use. Behavior-preserving. The similar loop in exec.CustomTraceMapReduce is left alone: package exec cannot import stagedsync, and it differs structurally (streams tasks straight to a queue, no readAheader cache or skip logic, inline hash warming, HistoryExecution always true). --- execution/stagedsync/exec3.go | 145 +++++++++++++++------------ execution/stagedsync/exec3_serial.go | 69 +++---------- 2 files changed, 94 insertions(+), 120 deletions(-) diff --git a/execution/stagedsync/exec3.go b/execution/stagedsync/exec3.go index 39c0c1c480f..3cd73f9e72c 100644 --- a/execution/stagedsync/exec3.go +++ b/execution/stagedsync/exec3.go @@ -504,6 +504,71 @@ func (te *txExecutor) onBlockStart(ctx context.Context, blockNum uint64, blockHa } } +func (te *txExecutor) buildBlockTasks(ctx context.Context, tx kv.Tx, blockNum uint64, + initialTxNum uint64, inputTxNum uint64, lastFrozenTxNum uint64, readAhead chan uint64, + getHeader func(hash common.Hash, number uint64) (*types.Header, error), + blockStateCache *state.BlockStateCache) (b *types.Block, txTasks []exec.Task, nextTxNum uint64, partialBlock bool, err error) { + + select { + case readAhead <- blockNum: + default: + } + + canonicalHash, err := rawdb.ReadCanonicalHash(tx, blockNum) + if err != nil { + return nil, nil, inputTxNum, false, err + } + b, ok := te.cfg.readAheader.ReadBlockWithSenders(canonicalHash) + if b == nil || !ok { + b, err = exec.BlockWithSenders(ctx, te.cfg.db, tx, te.cfg.blockReader, blockNum) + if err != nil { + return nil, nil, inputTxNum, false, err + } + } + if b == nil { + return nil, nil, inputTxNum, false, fmt.Errorf("nil block %d", blockNum) + } + go warmTxsHashes(b) + + txs := b.Transactions() + header := b.HeaderNoCopy() + + blockContext := protocol.NewEVMBlockContext(header, protocol.GetHashFn(header, getHeader), + te.cfg.engine, te.cfg.author, te.cfg.chainConfig) + + for txIndex := -1; txIndex <= len(txs); txIndex++ { + if inputTxNum > 0 && inputTxNum <= initialTxNum { + partialBlock = true + inputTxNum++ + continue + } + + // Do not oversend, wait for the result heap to go under certain size + txTask := &exec.TxTask{ + TxNum: inputTxNum, + TxIndex: txIndex, + Header: header, + Uncles: b.Uncles(), + Txs: txs, + EvmBlockContext: blockContext, + Withdrawals: b.Withdrawals(), + // use history reader instead of state reader to catch up to the tx where we left off + HistoryExecution: lastFrozenTxNum > 0 && inputTxNum <= lastFrozenTxNum, + Config: te.cfg.chainConfig, + Engine: te.cfg.engine, + Trace: dbg.TraceTx(blockNum, txIndex), + Hooks: te.hooks, + Logger: te.logger, + BlockStateCache: blockStateCache, + } + + txTasks = append(txTasks, txTask) + inputTxNum++ + } + + return b, txTasks, inputTxNum, partialBlock, nil +} + func (te *txExecutor) executeBlocks(ctx context.Context, startBlockNum uint64, maxBlockNum uint64, blockLimit uint64, initialTxNum uint64, inputTxNum uint64, readAhead chan uint64, initialCycle bool, applyResults chan applyResult, commitResults ...chan applyResult) error { if te.execLoopGroup == nil { return errors.New("no exec group") @@ -561,27 +626,26 @@ func (te *txExecutor) executeBlocks(ctx context.Context, startBlockNum uint64, m } for blockNum := startBlockNum; blockNum <= maxBlockNum; blockNum++ { - select { - case readAhead <- blockNum: - default: - } + // Per-block committed state cache for parallel workers' GetCommittedState. + blockStateCache := state.NewBlockStateCache() - var canonicalHash common.Hash - canonicalHash, err = rawdb.ReadCanonicalHash(blockTx, blockNum) - if err != nil { - return err - } - b, ok := te.cfg.readAheader.ReadBlockWithSenders(canonicalHash) - if b == nil || !ok { - b, err = exec.BlockWithSenders(ctx, te.cfg.db, blockTx, te.cfg.blockReader, blockNum) - } + var b *types.Block + var txTasks []exec.Task + // BlockContext: workers override GetHash with their own per-worker + // function (installWorkerGetHash) using their own roTx. The + // placeholder here uses execRoTx for the serial path fallback. + b, txTasks, inputTxNum, _, err = te.buildBlockTasks(ctx, blockTx, blockNum, + initialTxNum, inputTxNum, lastFrozenTxNum, readAhead, + func(hash common.Hash, number uint64) (h *types.Header, err error) { + h, err = te.cfg.blockReader.Header(ctx, blockTx, hash, number) + if h == nil && err == nil { + h = &types.Header{} + } + return h, err + }, blockStateCache) if err != nil { return err } - if b == nil { - return fmt.Errorf("nil block %d", blockNum) - } - go warmTxsHashes(b) var dbBAL types.BlockAccessList // Read BAL through blockTx (overlay or execRoTx) — do NOT open @@ -601,53 +665,6 @@ func (te *txExecutor) executeBlocks(ctx context.Context, startBlockNum uint64, m } } - txs := b.Transactions() - header := b.HeaderNoCopy() - - // BlockContext: workers override GetHash with their own per-worker - // function (installWorkerGetHash) using their own roTx. The - // placeholder here uses execRoTx for the serial path fallback. - blockContext := protocol.NewEVMBlockContext(header, protocol.GetHashFn(header, func(hash common.Hash, number uint64) (h *types.Header, err error) { - h, err = te.cfg.blockReader.Header(ctx, blockTx, hash, number) - if h == nil && err == nil { - h = &types.Header{} - } - return h, err - }), te.cfg.engine, te.cfg.author, te.cfg.chainConfig) - - var txTasks []exec.Task - // Per-block committed state cache for parallel workers' GetCommittedState. - blockStateCache := state.NewBlockStateCache() - - for txIndex := -1; txIndex <= len(txs); txIndex++ { - if inputTxNum > 0 && inputTxNum <= initialTxNum { - inputTxNum++ - continue - } - - // Do not oversend, wait for the result heap to go under certain size - txTask := &exec.TxTask{ - TxNum: inputTxNum, - TxIndex: txIndex, - Header: header, - Uncles: b.Uncles(), - Txs: txs, - EvmBlockContext: blockContext, - Withdrawals: b.Withdrawals(), - // use history reader instead of state reader to catch up to the tx where we left off - HistoryExecution: lastFrozenTxNum > 0 && inputTxNum <= lastFrozenTxNum, - Config: te.cfg.chainConfig, - Engine: te.cfg.engine, - Trace: dbg.TraceTx(blockNum, txIndex), - Hooks: te.hooks, - Logger: te.logger, - BlockStateCache: blockStateCache, - } - - txTasks = append(txTasks, txTask) - inputTxNum++ - } - lastExecutedStep := kv.Step(inputTxNum / te.doms.StepSize()) // if we're in the initialCycle before we consider the blockLimit we need to make sure we keep executing diff --git a/execution/stagedsync/exec3_serial.go b/execution/stagedsync/exec3_serial.go index 9e454c001fb..fe409b0b564 100644 --- a/execution/stagedsync/exec3_serial.go +++ b/execution/stagedsync/exec3_serial.go @@ -14,7 +14,6 @@ import ( "github.com/erigontech/erigon/common/log/v3" "github.com/erigontech/erigon/db/consensuschain" "github.com/erigontech/erigon/db/kv" - "github.com/erigontech/erigon/db/rawdb" "github.com/erigontech/erigon/db/rawdb/rawtemporaldb" "github.com/erigontech/erigon/db/state/changeset" "github.com/erigontech/erigon/execution/commitment" @@ -89,44 +88,31 @@ func (se *serialExecutor) exec(ctx context.Context, execStage *StageState, u Unw se.doms.SetChangesetAccumulator(changeSet) } - select { - case readAhead <- blockNum: - default: - } - - canonicalHash, err := rawdb.ReadCanonicalHash(se.applyTx, blockNum) + getHashFnMutex := sync.Mutex{} + var txTasks []exec.Task + var partialBlock bool + var err error + b, txTasks, inputTxNum, partialBlock, err = se.buildBlockTasks(ctx, se.applyTx, blockNum, + initialTxNum, inputTxNum, lastFrozenTxNum, readAhead, + func(hash common.Hash, number uint64) (*types.Header, error) { + getHashFnMutex.Lock() + defer getHashFnMutex.Unlock() + return se.getHeader(ctx, hash, number) + }, nil) if err != nil { return nil, rwTx, err } - var ok bool - b, ok = se.cfg.readAheader.ReadBlockWithSenders(canonicalHash) - if b == nil || !ok { - b, err = exec.BlockWithSenders(ctx, se.cfg.db, se.applyTx, se.cfg.blockReader, blockNum) - if err != nil { - return nil, rwTx, err - } - } - if b == nil { - // TODO: panic here and see that overall process deadlock - return nil, rwTx, fmt.Errorf("nil block %d", blockNum) + if partialBlock { + havePartialBlock = true } - go warmTxsHashes(b) - txs := b.Transactions() header := b.HeaderNoCopy() - getHashFnMutex := sync.Mutex{} if se.cfg.chainConfig.AmsterdamTime != nil && *se.cfg.chainConfig.AmsterdamTime > 0 && se.cfg.chainConfig.IsAmsterdam(header.Time) { se.logger.Error(fmt.Sprintf("[%s] BLOCK PROCESSING FAILED: Amsterdam processing is not supported by serial exec", se.logPrefix), "fork-block", blockNum) return nil, rwTx, fmt.Errorf("amsterdam processing is not supported by serial exec from block: %d", blockNum) } - blockContext := protocol.NewEVMBlockContext(header, protocol.GetHashFn(header, func(hash common.Hash, number uint64) (*types.Header, error) { - getHashFnMutex.Lock() - defer getHashFnMutex.Unlock() - return se.getHeader(ctx, hash, number) - }), se.cfg.engine, se.cfg.author, se.cfg.chainConfig) - se.accumulator = accumulator // keep in sync for executeBlock's stateWriter if accumulator != nil { txs, err := se.cfg.blockReader.RawTransactions(context.Background(), se.applyTx, b.NumberU64(), b.NumberU64()) @@ -136,35 +122,6 @@ func (se *serialExecutor) exec(ctx context.Context, execStage *StageState, u Unw accumulator.StartChange(header, txs, false) } - var txTasks []exec.Task - - for txIndex := -1; txIndex <= len(txs); txIndex++ { - // Do not oversend, wait for the result heap to go under certain size - txTask := &exec.TxTask{ - TxNum: inputTxNum, - TxIndex: txIndex, - Header: header, - Uncles: b.Uncles(), - Txs: txs, - EvmBlockContext: blockContext, - Withdrawals: b.Withdrawals(), - // use history reader instead of state reader to catch up to the tx where we left off - HistoryExecution: lastFrozenTxNum > 0 && inputTxNum <= lastFrozenTxNum, - Trace: dbg.TraceTx(blockNum, txIndex), - Hooks: se.hooks, - Logger: se.logger, - } - - if txTask.TxNum > 0 && txTask.TxNum <= initialTxNum { - havePartialBlock = true - inputTxNum++ - continue - } - - txTasks = append(txTasks, txTask) - inputTxNum++ - } - start := time.Now() continueLoop, err := se.executeBlock(ctx, txTasks, execStage.CurrentSyncCycle.IsInitialCycle, false)