package miner import ( "context" "errors" "fmt" "math/big" "os" "sync" "sync/atomic" "time" // nolint:typecheck "github.com/ethereum/go-ethereum/common" cmath "github.com/ethereum/go-ethereum/common/math" "github.com/ethereum/go-ethereum/common/tracing" "github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/blockstm" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/txpool" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rlp" "github.com/holiman/uint256" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" lru "github.com/hashicorp/golang-lru" ) // newWorkerWithDelay is newWorker() with extra params to induce artficial delays for tests such as commit-interrupt. // nolint:staticcheck func newWorkerWithDelay(config *Config, chainConfig *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux, isLocalBlock func(header *types.Header) bool, init bool, delay uint, opcodeDelay uint) *worker { worker := &worker{ config: config, chainConfig: chainConfig, engine: engine, eth: eth, chain: eth.BlockChain(), mux: mux, isLocalBlock: isLocalBlock, coinbase: config.Etherbase, extra: config.ExtraData, pendingTasks: make(map[common.Hash]*task), txsCh: make(chan core.NewTxsEvent, txChanSize), chainHeadCh: make(chan core.ChainHeadEvent, chainHeadChanSize), newWorkCh: make(chan *newWorkReq), getWorkCh: make(chan *getWorkReq), taskCh: make(chan *task), resultCh: make(chan *types.Block, resultQueueSize), startCh: make(chan struct{}, 1), exitCh: make(chan struct{}), resubmitIntervalCh: make(chan time.Duration), resubmitAdjustCh: make(chan *intervalAdjust, resubmitAdjustChanSize), interruptCommitFlag: config.CommitInterruptFlag, } worker.noempty.Store(true) worker.profileCount = new(int32) // Subscribe for transaction insertion events (whether from network or resurrects) worker.txsSub = eth.TxPool().SubscribeTransactions(worker.txsCh, true) // Subscribe events for blockchain worker.chainHeadSub = eth.BlockChain().SubscribeChainHeadEvent(worker.chainHeadCh) interruptedTxCache, err := lru.New(vm.InterruptedTxCacheSize) if err != nil { log.Warn("Failed to create interrupted tx cache", "err", err) } worker.interruptedTxCache = &vm.TxCache{ Cache: interruptedTxCache, } if !worker.interruptCommitFlag { worker.noempty.Store(false) } // Sanitize recommit interval if the user-specified one is too short. recommit := worker.config.Recommit if recommit < minRecommitInterval { log.Warn("Sanitizing miner recommit interval", "provided", recommit, "updated", minRecommitInterval) recommit = minRecommitInterval } worker.recommit = recommit // Sanitize the timeout config for creating payload. newpayloadTimeout := worker.config.NewPayloadTimeout if newpayloadTimeout == 0 { log.Warn("Sanitizing new payload timeout to default", "provided", newpayloadTimeout, "updated", DefaultConfig.NewPayloadTimeout) newpayloadTimeout = DefaultConfig.NewPayloadTimeout } if newpayloadTimeout < time.Millisecond*100 { log.Warn("Low payload timeout may cause high amount of non-full blocks", "provided", newpayloadTimeout, "default", DefaultConfig.NewPayloadTimeout) } worker.newpayloadTimeout = newpayloadTimeout ctx := tracing.WithTracer(context.Background(), otel.GetTracerProvider().Tracer("MinerWorker")) worker.wg.Add(4) go worker.mainLoopWithDelay(ctx, delay, opcodeDelay) go worker.newWorkLoop(ctx, recommit) go worker.resultLoop() go worker.taskLoop() // Submit first work to initialize pending state. if init { worker.startCh <- struct{}{} } return worker } // mainLoopWithDelay is mainLoop() with extra params to induce artficial delays for tests such as commit-interrupt. // nolint:gocognit func (w *worker) mainLoopWithDelay(ctx context.Context, delay uint, opcodeDelay uint) { defer w.wg.Done() defer w.txsSub.Unsubscribe() defer w.chainHeadSub.Unsubscribe() defer func() { if w.current != nil { w.current.discard() } }() for { select { case req := <-w.newWorkCh: if w.chainConfig.ChainID.Cmp(params.BorMainnetChainConfig.ChainID) == 0 || w.chainConfig.ChainID.Cmp(params.MumbaiChainConfig.ChainID) == 0 { if w.eth.PeerCount() > 0 { //nolint:contextcheck w.commitWorkWithDelay(req.ctx, req.interrupt, req.noempty, req.timestamp, delay, opcodeDelay) } } else { //nolint:contextcheck w.commitWorkWithDelay(req.ctx, req.interrupt, req.noempty, req.timestamp, delay, opcodeDelay) } case req := <-w.getWorkCh: req.result <- w.generateWork(ctx, req.params) case ev := <-w.txsCh: // Apply transactions to the pending state if we're not sealing // // Note all transactions received may not be continuous with transactions // already included in the current sealing block. These transactions will // be automatically eliminated. // nolint : nestif if !w.IsRunning() && w.current != nil { // If block is already full, abort if gp := w.current.gasPool; gp != nil && gp.Gas() < params.TxGas { continue } txs := make(map[common.Address][]*txpool.LazyTransaction, len(ev.Txs)) for _, tx := range ev.Txs { acc, _ := types.Sender(w.current.signer, tx) txs[acc] = append(txs[acc], &txpool.LazyTransaction{ Pool: w.eth.TxPool(), // We don't know where this came from, yolo resolve from everywhere Hash: tx.Hash(), Tx: nil, // Do *not* set this! We need to resolve it later to pull blobs in Time: tx.Time(), GasFeeCap: tx.GasFeeCap(), GasTipCap: tx.GasTipCap(), Gas: tx.Gas(), BlobGas: tx.BlobGas(), }) } txset := newTransactionsByPriceAndNonce(w.current.signer, txs, w.current.header.BaseFee) tcount := w.current.tcount w.commitTransactions(w.current, txset, nil, new(big.Int), context.Background()) // Only update the snapshot if any new transactons were added // to the pending block if tcount != w.current.tcount { w.updateSnapshot(w.current) } } else { // Special case, if the consensus engine is 0 period clique(dev mode), // submit sealing work here since all empty submission will be rejected // by clique. Of course the advance sealing(empty submission) is disabled. if w.chainConfig.Clique != nil && w.chainConfig.Clique.Period == 0 { w.commitWork(ctx, nil, true, time.Now().Unix()) } } w.newTxs.Add(int32(len(ev.Txs))) // System stopped case <-w.exitCh: return case <-w.txsSub.Err(): return case <-w.chainHeadSub.Err(): return } } } // commitWorkWithDelay is commitWork() with extra params to induce artficial delays for tests such as commit-interrupt. func (w *worker) commitWorkWithDelay(ctx context.Context, interrupt *atomic.Int32, noempty bool, timestamp int64, delay uint, opcodeDelay uint) { // Abort committing if node is still syncing if w.syncing.Load() { return } start := time.Now() var ( work *environment err error ) tracing.Exec(ctx, "", "worker.prepareWork", func(ctx context.Context, span trace.Span) { // Set the coinbase if the worker is running or it's required var coinbase common.Address if w.IsRunning() { coinbase = w.etherbase() if coinbase == (common.Address{}) { log.Error("Refusing to mine without etherbase") return } } work, err = w.prepareWork(&generateParams{ timestamp: uint64(timestamp), coinbase: coinbase, }) }) if err != nil { return } // nolint:contextcheck var interruptCtx = context.Background() stopFn := func() {} defer func() { stopFn() }() if !noempty && w.interruptCommitFlag { block := w.chain.GetBlockByHash(w.chain.CurrentBlock().Hash()) interruptCtx, stopFn = getInterruptTimer(ctx, work, block) // nolint : staticcheck interruptCtx = vm.PutCache(interruptCtx, w.interruptedTxCache) // nolint : staticcheck interruptCtx = context.WithValue(interruptCtx, vm.InterruptCtxDelayKey, delay) // nolint : staticcheck interruptCtx = context.WithValue(interruptCtx, vm.InterruptCtxOpcodeDelayKey, opcodeDelay) } ctx, span := tracing.StartSpan(ctx, "commitWork") defer tracing.EndSpan(span) tracing.SetAttributes( span, attribute.Int("number", int(work.header.Number.Uint64())), ) // Create an empty block based on temporary copied state for // sealing in advance without waiting block execution finished. if !noempty && !w.noempty.Load() { _ = w.commit(ctx, work.copy(), nil, false, start) } // Fill pending transactions from the txpool into the block. err = w.fillTransactionsWithDelay(ctx, interrupt, work, interruptCtx) switch { case err == nil: // The entire block is filled, decrease resubmit interval in case // of current interval is larger than the user-specified one. w.resubmitAdjustCh <- &intervalAdjust{inc: false} case errors.Is(err, errBlockInterruptedByRecommit): // Notify resubmit loop to increase resubmitting interval if the // interruption is due to frequent commits. gaslimit := work.header.GasLimit ratio := float64(gaslimit-work.gasPool.Gas()) / float64(gaslimit) if ratio < 0.1 { ratio = 0.1 } w.resubmitAdjustCh <- &intervalAdjust{ ratio: ratio, inc: true, } case errors.Is(err, errBlockInterruptedByNewHead): // If the block building is interrupted by newhead event, discard it // totally. Committing the interrupted block introduces unnecessary // delay, and possibly causes miner to mine on the previous head, // which could result in higher uncle rate. work.discard() return } // Submit the generated block for consensus sealing. _ = w.commit(ctx, work.copy(), w.fullTaskHook, true, start) // Swap out the old work with the new one, terminating any leftover // prefetcher processes in the mean time and starting a new one. if w.current != nil { w.current.discard() } w.current = work } // fillTransactionsWithDelay is fillTransactions() with extra params to induce artficial delays for tests such as commit-interrupt. // nolint:gocognit func (w *worker) fillTransactionsWithDelay(ctx context.Context, interrupt *atomic.Int32, env *environment, interruptCtx context.Context) error { ctx, span := tracing.StartSpan(ctx, "fillTransactions") defer tracing.EndSpan(span) // Split the pending transactions into locals and remotes // Fill the block with all available pending transactions. pending := w.eth.TxPool().Pending(true) localTxs, remoteTxs := make(map[common.Address][]*txpool.LazyTransaction), pending var ( localTxsCount int remoteTxsCount int ) // TODO: move to config or RPC const profiling = false if profiling { doneCh := make(chan struct{}) defer func() { close(doneCh) }() go func(number uint64) { closeFn := func() error { return nil } for { select { case <-time.After(150 * time.Millisecond): // Check if we've not crossed limit if attempt := atomic.AddInt32(w.profileCount, 1); attempt >= 10 { log.Info("Completed profiling", "attempt", attempt) return } log.Info("Starting profiling in fill transactions", "number", number) dir, err := os.MkdirTemp("", fmt.Sprintf("bor-traces-%s-", time.Now().UTC().Format("2006-01-02-150405Z"))) if err != nil { log.Error("Error in profiling", "path", dir, "number", number, "err", err) return } // grab the cpu profile closeFnInternal, err := startProfiler("cpu", dir, number) if err != nil { log.Error("Error in profiling", "path", dir, "number", number, "err", err) return } closeFn = func() error { err := closeFnInternal() log.Info("Completed profiling", "path", dir, "number", number, "error", err) return nil } case <-doneCh: err := closeFn() if err != nil { log.Info("closing fillTransactions", "number", number, "error", err) } return } } }(env.header.Number.Uint64()) } tracing.Exec(ctx, "", "worker.SplittingTransactions", func(ctx context.Context, span trace.Span) { prePendingTime := time.Now() pending := w.eth.TxPool().Pending(true) remoteTxs = pending postPendingTime := time.Now() for _, account := range w.eth.TxPool().Locals() { if txs := remoteTxs[account]; len(txs) > 0 { delete(remoteTxs, account) localTxs[account] = txs } } postLocalsTime := time.Now() tracing.SetAttributes( span, attribute.Int("len of local txs", localTxsCount), attribute.Int("len of remote txs", remoteTxsCount), attribute.String("time taken by Pending()", fmt.Sprintf("%v", postPendingTime.Sub(prePendingTime))), attribute.String("time taken by Locals()", fmt.Sprintf("%v", postLocalsTime.Sub(postPendingTime))), ) }) var ( localEnvTCount int remoteEnvTCount int err error ) if len(localTxs) > 0 { var txs *transactionsByPriceAndNonce tracing.Exec(ctx, "", "worker.LocalTransactionsByPriceAndNonce", func(ctx context.Context, span trace.Span) { var baseFee *uint256.Int if env.header.BaseFee != nil { baseFee = cmath.FromBig(env.header.BaseFee) } txs = newTransactionsByPriceAndNonce(env.signer, localTxs, baseFee.ToBig()) tracing.SetAttributes( span, attribute.Int("len of tx local Heads", txs.GetTxs()), ) }) tracing.Exec(ctx, "", "worker.LocalCommitTransactions", func(ctx context.Context, span trace.Span) { err = w.commitTransactionsWithDelay(env, txs, interrupt, interruptCtx) }) if err != nil { return err } localEnvTCount = env.tcount } if len(remoteTxs) > 0 { var txs *transactionsByPriceAndNonce tracing.Exec(ctx, "", "worker.RemoteTransactionsByPriceAndNonce", func(ctx context.Context, span trace.Span) { var baseFee *uint256.Int if env.header.BaseFee != nil { baseFee = cmath.FromBig(env.header.BaseFee) } txs = newTransactionsByPriceAndNonce(env.signer, remoteTxs, baseFee.ToBig()) tracing.SetAttributes( span, attribute.Int("len of tx remote Heads", txs.GetTxs()), ) }) tracing.Exec(ctx, "", "worker.RemoteCommitTransactions", func(ctx context.Context, span trace.Span) { err = w.commitTransactionsWithDelay(env, txs, interrupt, interruptCtx) }) if err != nil { return err } remoteEnvTCount = env.tcount } tracing.SetAttributes( span, attribute.Int("len of final local txs ", localEnvTCount), attribute.Int("len of final remote txs", remoteEnvTCount), ) return nil } // commitTransactionsWithDelay is commitTransactions() with extra params to induce artficial delays for tests such as commit-interrupt. // nolint:gocognit, unparam func (w *worker) commitTransactionsWithDelay(env *environment, txs *transactionsByPriceAndNonce, interrupt *atomic.Int32, interruptCtx context.Context) error { gasLimit := env.header.GasLimit if env.gasPool == nil { env.gasPool = new(core.GasPool).AddGas(gasLimit) } var coalescedLogs []*types.Log var depsMVReadList [][]blockstm.ReadDescriptor var depsMVFullWriteList [][]blockstm.WriteDescriptor var mvReadMapList []map[blockstm.Key]blockstm.ReadDescriptor var deps map[int]map[int]bool chDeps := make(chan blockstm.TxDep) var count int var depsWg sync.WaitGroup EnableMVHashMap := w.chainConfig.IsCancun(env.header.Number) // create and add empty mvHashMap in statedb if EnableMVHashMap { depsMVReadList = [][]blockstm.ReadDescriptor{} depsMVFullWriteList = [][]blockstm.WriteDescriptor{} mvReadMapList = []map[blockstm.Key]blockstm.ReadDescriptor{} deps = map[int]map[int]bool{} chDeps = make(chan blockstm.TxDep) count = 0 depsWg.Add(1) go func(chDeps chan blockstm.TxDep) { for t := range chDeps { deps = blockstm.UpdateDeps(deps, t) } depsWg.Done() }(chDeps) } var lastTxHash common.Hash mainloop: for { if interruptCtx != nil { if EnableMVHashMap { env.state.AddEmptyMVHashMap() } // case of interrupting by timeout select { case <-interruptCtx.Done(): txCommitInterruptCounter.Inc(1) log.Warn("Tx Level Interrupt", "hash", lastTxHash) break mainloop default: } } // Check interruption signal and abort building if it's fired. if interrupt != nil { if signal := interrupt.Load(); signal != commitInterruptNone { return signalToErr(signal) } } // If we don't have enough gas for any further transactions then we're done. if env.gasPool.Gas() < params.TxGas { log.Trace("Not enough gas for further transactions", "have", env.gasPool, "want", params.TxGas) break } // Retrieve the next transaction and abort if all done. ltx, _ := txs.Peek() if ltx == nil { break } lastTxHash = ltx.Hash // If we don't have enough space for the next transaction, skip the account. if env.gasPool.Gas() < ltx.Gas { log.Trace("Not enough gas left for transaction", "hash", ltx.Hash, "left", env.gasPool.Gas(), "needed", ltx.Gas) txs.Pop() continue } if left := uint64(params.MaxBlobGasPerBlock - env.blobs*params.BlobTxBlobGasPerBlob); left < ltx.BlobGas { log.Trace("Not enough blob gas left for transaction", "hash", ltx.Hash, "left", left, "needed", ltx.BlobGas) txs.Pop() continue } // Transaction seems to fit, pull it up from the pool tx := ltx.Resolve() if tx == nil { log.Trace("Ignoring evicted transaction", "hash", ltx.Hash) txs.Pop() continue } // Error may be ignored here. The error has already been checked // during transaction acceptance is the transaction pool. from, _ := types.Sender(env.signer, tx) // not prioritising conditional transaction, yet. //nolint:nestif if options := tx.GetOptions(); options != nil { if err := env.header.ValidateBlockNumberOptionsPIP15(options.BlockNumberMin, options.BlockNumberMax); err != nil { log.Trace("Dropping conditional transaction", "from", from, "hash", tx.Hash(), "reason", err) txs.Pop() continue } if err := env.header.ValidateTimestampOptionsPIP15(options.TimestampMin, options.TimestampMax); err != nil { log.Trace("Dropping conditional transaction", "from", from, "hash", tx.Hash(), "reason", err) txs.Pop() continue } if err := env.state.ValidateKnownAccounts(options.KnownAccounts); err != nil { log.Trace("Dropping conditional transaction", "from", from, "hash", tx.Hash(), "reason", err) txs.Pop() continue } } // Check whether the tx is replay protected. If we're not in the EIP155 hf // phase, start ignoring the sender until we do. if tx.Protected() && !w.chainConfig.IsEIP155(env.header.Number) { log.Trace("Ignoring replay protected transaction", "hash", ltx.Hash, "eip155", w.chainConfig.EIP155Block) txs.Pop() continue } // Start executing the transaction env.state.SetTxContext(tx.Hash(), env.tcount) logs, err := w.commitTransaction(env, tx, interruptCtx) if interruptCtx != nil { if delay := interruptCtx.Value(vm.InterruptCtxDelayKey); delay != nil { // nolint : durationcheck time.Sleep(time.Duration(delay.(uint)) * time.Millisecond) } } switch { case errors.Is(err, core.ErrNonceTooLow): // New head notification data race between the transaction pool and miner, shift log.Trace("Skipping transaction with low nonce", "hash", ltx.Hash, "sender", from, "nonce", tx.Nonce()) txs.Shift() case errors.Is(err, nil): // Everything ok, collect the logs and shift in the next transaction from the same account coalescedLogs = append(coalescedLogs, logs...) env.tcount++ if EnableMVHashMap { depsMVReadList = append(depsMVReadList, env.state.MVReadList()) depsMVFullWriteList = append(depsMVFullWriteList, env.state.MVFullWriteList()) mvReadMapList = append(mvReadMapList, env.state.MVReadMap()) temp := blockstm.TxDep{ Index: env.tcount - 1, ReadList: depsMVReadList[count], FullWriteList: depsMVFullWriteList, } chDeps <- temp count++ } txs.Shift() default: // Transaction is regarded as invalid, drop all consecutive transactions from // the same sender because of `nonce-too-high` clause. log.Debug("Transaction failed, account skipped", "hash", ltx.Hash, "err", err) txs.Pop() } if EnableMVHashMap { env.state.ClearReadMap() env.state.ClearWriteMap() } } // nolint:nestif if EnableMVHashMap && w.IsRunning() { close(chDeps) depsWg.Wait() var blockExtraData types.BlockExtraData tempVanity := env.header.Extra[:types.ExtraVanityLength] tempSeal := env.header.Extra[len(env.header.Extra)-types.ExtraSealLength:] if len(mvReadMapList) > 0 { tempDeps := make([][]uint64, len(mvReadMapList)) for j := range deps[0] { tempDeps[0] = append(tempDeps[0], uint64(j)) } delayFlag := true for i := 1; i <= len(mvReadMapList)-1; i++ { reads := mvReadMapList[i-1] _, ok1 := reads[blockstm.NewSubpathKey(env.coinbase, state.BalancePath)] _, ok2 := reads[blockstm.NewSubpathKey(common.HexToAddress(w.chainConfig.Bor.CalculateBurntContract(env.header.Number.Uint64())), state.BalancePath)] if ok1 || ok2 { delayFlag = false } for j := range deps[i] { tempDeps[i] = append(tempDeps[i], uint64(j)) } } if err := rlp.DecodeBytes(env.header.Extra[types.ExtraVanityLength:len(env.header.Extra)-types.ExtraSealLength], &blockExtraData); err != nil { log.Error("error while decoding block extra data", "err", err) return err } if delayFlag { blockExtraData.TxDependency = tempDeps } else { blockExtraData.TxDependency = nil } } else { blockExtraData.TxDependency = nil } blockExtraDataBytes, err := rlp.EncodeToBytes(blockExtraData) if err != nil { log.Error("error while encoding block extra data: %v", err) return err } env.header.Extra = []byte{} env.header.Extra = append(tempVanity, blockExtraDataBytes...) env.header.Extra = append(env.header.Extra, tempSeal...) } if !w.IsRunning() && len(coalescedLogs) > 0 { // We don't push the pendingLogsEvent while we are sealing. The reason is that // when we are sealing, the worker will regenerate a sealing block every 3 seconds. // In order to avoid pushing the repeated pendingLog, we disable the pending log pushing. // make a copy, the state caches the logs and these logs get "upgraded" from pending to mined // logs by filling in the block hash when the block was mined by the local miner. This can // cause a race condition if a log was "upgraded" before the PendingLogsEvent is processed. cpy := make([]*types.Log, len(coalescedLogs)) for i, l := range coalescedLogs { cpy[i] = new(types.Log) *cpy[i] = *l } w.pendingLogsFeed.Send(cpy) } return nil }