Skip to content

execution/stagedsync, execution/exec: dedup exec3 block-task construction and AA execution #22174

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
yperbasis wants to merge 2 commits into
base: main
Choose a base branch
from
+97 −186
Draft

execution/stagedsync, execution/exec: dedup exec3 block-task construction and AA execution #22174

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6dc80e4
execution/exec: dedup execAATxn into TxTask.executeAA
yperbasis Jul 2, 2026
4551188
execution/stagedsync: extract shared buildBlockTasks helper
yperbasis Jul 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 3 additions & 66 deletions execution/exec/historical_trace_worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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 {
Comment on lines +257 to +259
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
}

Expand Down
145 changes: 81 additions & 64 deletions execution/stagedsync/exec3.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down
69 changes: 13 additions & 56 deletions execution/stagedsync/exec3_serial.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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())
Expand All @@ -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)

Expand Down
Loading