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 } 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)