forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 10:22:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat: add ccc into miner (#674)

Browse source 
* add core/rawdb/accessors_row_consumption.go

* update core/rawdb/schema.go

* update eth/ethconfig/config.go

* update eth/ethconfig/gen_config.go

* update miner/miner.go

* update eth/backend.go

* update core/types.go

* update cmd/geth/main.go

* update cmd/utils/flags.go

* update core/types/block.go

* update core/block_validator.go

* WIP: update miner/worker.go

* WIP: update miner/worker.go

* WIP: update miner/worker.go

* WIP: update miner/worker.go

* WIP: update miner/worker.go

* WIP: update miner/worker.go add withTimer

* WIP: update miner/worker.go

* WIP: update miner/worker.go add some func

* WIP: update miner/worker.go `makeEnv`

* fix

* fix

* minor

* init traces return

* init accRows

* update `applyTransaction`

* fix `TestPriorityClient`

* miner: update metrics

* minor

* add `l2CommitTxTraceStateRevertTimer`

* update `calcAndSetAccRowsForEnv`

* start to work on `commitTransactions`

* seal block early if we're over time

* If we have collected enough transactions then we're done

* check L1 message queue index

* A single L1 message leads to out-of-gas

* more on nil err

* more error

* skip RemoveTx for now

* minor

* l1TxStrangeErr

* fix interface

* start to work on `commitWork`

* minor

* draft `fillTransactions`

* fix some types

* refactor

* fix

* add `circuitCapacityReached`

* minor

* fix `txToLazyTx`

* fix `(m *mockBackend) ChainDb()`

* fix `removeTx`

* update comments
This commit is contained in:
HAOYUatHZ 2024-04-28 18:16:10 +08:00 committed by GitHub
parent bdcede5221
commit ca1ba1da87
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 812 additions and 87 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cmd/geth/main.go
View file

@ -123,6 +123,7 @@ var (
utils.MinerExtraDataFlag,
utils.MinerRecommitIntervalFlag,
utils.MinerNewPayloadTimeout,
utils.MinerStoreSkippedTxTracesFlag,
utils.NATFlag,
utils.NoDiscoverFlag,
utils.DiscoveryV4Flag,
@ -147,6 +148,7 @@ var (
utils.L1EndpointFlag,
utils.L1ConfirmationsFlag,
utils.L1DeploymentBlockFlag,
utils.CircuitCapacityCheckEnabledFlag,
utils.RollupVerifyEnabledFlag,
}, utils.NetworkFlags, utils.DatabaseFlags)

20
cmd/utils/flags.go
View file

@ -503,6 +503,10 @@ var (
Value: ethconfig.Defaults.Miner.NewPayloadTimeout,
Category: flags.MinerCategory,
}
MinerStoreSkippedTxTracesFlag = &cli.BoolFlag{
Name: "miner.storeskippedtxtraces",
Usage: "Store the wrapped traces when storing a skipped tx",
}
// Account settings
UnlockedAccountFlag = &cli.StringFlag{
@ -962,6 +966,12 @@ Please note that --` + MetricsHTTPFlag.Name + ` must be set to start the server.
Usage: "L1 block height to start syncing from. Should be set to the L1 message queue deployment block number.",
}
// Circuit capacity check settings
CircuitCapacityCheckEnabledFlag = &cli.BoolFlag{
Name: "ccc",
Usage: "Enable circuit capacity check during block validation",
}
// Rollup verify service settings
RollupVerifyEnabledFlag = &cli.BoolFlag{
Name: "rollup.verify",
@ -1641,6 +1651,9 @@ func setMiner(ctx *cli.Context, cfg *miner.Config) {
if ctx.IsSet(MinerNewPayloadTimeout.Name) {
cfg.NewPayloadTimeout = ctx.Duration(MinerNewPayloadTimeout.Name)
}
if ctx.IsSet(MinerStoreSkippedTxTracesFlag.Name) {
cfg.StoreSkippedTxTraces = ctx.Bool(MinerStoreSkippedTxTracesFlag.Name)
}
}
func setRequiredBlocks(ctx *cli.Context, cfg *ethconfig.Config) {
@ -1671,6 +1684,12 @@ func setRequiredBlocks(ctx *cli.Context, cfg *ethconfig.Config) {
}
}
func setCircuitCapacityCheck(ctx *cli.Context, cfg *ethconfig.Config) {
if ctx.IsSet(CircuitCapacityCheckEnabledFlag.Name) {
cfg.CheckCircuitCapacity = ctx.Bool(CircuitCapacityCheckEnabledFlag.Name)
}
}
func setEnableRollupVerify(ctx *cli.Context, cfg *ethconfig.Config) {
if ctx.IsSet(RollupVerifyEnabledFlag.Name) {
cfg.EnableRollupVerify = ctx.Bool(RollupVerifyEnabledFlag.Name)
@ -1740,6 +1759,7 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
setMiner(ctx, &cfg.Miner)
setRequiredBlocks(ctx, cfg)
setLes(ctx, cfg)
setCircuitCapacityCheck(ctx, cfg)
setEnableRollupVerify(ctx, cfg)
setMaxBlockRange(ctx, cfg)

8
core/block_validator.go
View file

@ -155,9 +155,9 @@ func (v *BlockValidator) ValidateBody(block *types.Block) error {
// if a block's RowConsumption has been stored, which means it has been processed before,
// (e.g., in miner/worker.go or in insertChain),
// we simply skip its calculation and validation
// if rawdb.ReadBlockRowConsumption(v.bc.db, block.Hash()) != nil {
// return nil
// }
if rawdb.ReadBlockRowConsumption(v.bc.db, block.Hash()) != nil {
return nil
}
rowConsumption, err := v.validateCircuitRowConsumption(block)
if err != nil {
return err
@ -169,7 +169,7 @@ func (v *BlockValidator) ValidateBody(block *types.Block) error {
"hash", block.Hash().String(),
"rowConsumption", rowConsumption,
)
// rawdb.WriteBlockRowConsumption(v.bc.db, block.Hash(), rowConsumption)
rawdb.WriteBlockRowConsumption(v.bc.db, block.Hash(), rowConsumption)
}
return nil

51
core/rawdb/accessors_row_consumption.go Normal file
View file

@ -0,0 +1,51 @@
package rawdb
import (
"bytes"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/rlp"
)
// WriteBlockRowConsumption writes a RowConsumption of the block to the database.
func WriteBlockRowConsumption(db ethdb.KeyValueWriter, l2BlockHash common.Hash, rc *types.RowConsumption) {
if rc == nil {
return
}
bytes, err := rlp.EncodeToBytes(&rc)
if err != nil {
log.Crit("Failed to RLP encode RowConsumption ", "err", err)
}
if err := db.Put(rowConsumptionKey(l2BlockHash), bytes); err != nil {
log.Crit("Failed to store RowConsumption ", "err", err)
}
}
// ReadBlockRowConsumption retrieves the RowConsumption corresponding to the block hash.
func ReadBlockRowConsumption(db ethdb.Reader, l2BlockHash common.Hash) *types.RowConsumption {
data := ReadBlockRowConsumptionRLP(db, l2BlockHash)
if len(data) == 0 {
return nil
}
rc := new(types.RowConsumption)
if err := rlp.Decode(bytes.NewReader(data), rc); err != nil {
log.Crit("Invalid RowConsumption message RLP", "l2BlockHash", l2BlockHash.String(), "data", data, "err", err)
}
return rc
}
// ReadBlockRowConsumption retrieves the RowConsumption in its raw RLP database encoding.
func ReadBlockRowConsumptionRLP(db ethdb.Reader, l2BlockHash common.Hash) rlp.RawValue {
data, err := db.Get(rowConsumptionKey(l2BlockHash))
if err != nil && isNotFoundErr(err) {
return nil
}
if err != nil {
log.Crit("Failed to load RowConsumption", "l2BlockHash", l2BlockHash.String(), "err", err)
}
return data
}

8
core/rawdb/schema.go
View file

@ -153,6 +153,9 @@ var (
batchMetaPrefix = []byte("R-bm")
finalizedL2BlockNumberKey = []byte("R-finalized")
// Row consumption
rowConsumptionPrefix = []byte("rc") // rowConsumptionPrefix + hash -> row consumption by block
// Skipped transactions
numSkippedTransactionsKey = []byte("NumberOfSkippedTransactions")
skippedTransactionPrefix = []byte("skip") // skippedTransactionPrefix + tx hash -> skipped transaction
@ -387,6 +390,11 @@ func FirstQueueIndexNotInL2BlockKey(l2BlockHash common.Hash) []byte {
return append(firstQueueIndexNotInL2BlockPrefix, l2BlockHash.Bytes()...)
}
// rowConsumptionKey = rowConsumptionPrefix + hash
func rowConsumptionKey(hash common.Hash) []byte {
return append(rowConsumptionPrefix, hash.Bytes()...)
}
// SkippedTransactionKey = skippedTransactionPrefix + tx hash
func SkippedTransactionKey(txHash common.Hash) []byte {
return append(skippedTransactionPrefix, txHash.Bytes()...)

5
core/txpool/blobpool/blobpool.go
View file

@ -1557,3 +1557,8 @@ func (p *BlobPool) Status(hash common.Hash) txpool.TxStatus {
}
return txpool.TxStatusUnknown
}
// added for interface compatibility, do nothing
func (p *BlobPool) RemoveTx(hash common.Hash, outofbound bool, unreserve bool) int {
return 0
}

9
core/txpool/legacypool/legacypool.go
View file

@ -1063,6 +1063,15 @@ func (pool *LegacyPool) Has(hash common.Hash) bool {
return pool.all.Get(hash) != nil
}
// RemoveTx is similar to removeTx, but with locking to prevent concurrency.
// Note: currently should only be called by miner/worker.go.
func (pool *LegacyPool) RemoveTx(hash common.Hash, outofbound bool, unreserve bool) int {
pool.mu.Lock()
defer pool.mu.Unlock()
return pool.removeTx(hash, outofbound, unreserve)
}
// removeTx removes a single transaction from the queue, moving all subsequent
// transactions back to the future queue.
//

3
core/txpool/subpool.go
View file

@ -137,4 +137,7 @@ type SubPool interface {
// Status returns the known status (unknown/pending/queued) of a transaction
// identified by their hashes.
Status(hash common.Hash) TxStatus
// RemoveTx removes a transaction from the pool, returning the number of transactions removed.
RemoveTx(hash common.Hash, outofbound bool, unreserve bool) int
}

9
core/txpool/txpool.go
View file

@ -415,3 +415,12 @@ func (p *TxPool) Status(hash common.Hash) TxStatus {
}
return TxStatusUnknown
}
// RemoveTx removes a transaction from the pool, returning the number of transactions removed.
func (p *TxPool) RemoveTx(hash common.Hash, outofbound bool, unreserve bool) int {
var ret int
for _, subpool := range p.subpools {
ret += subpool.RemoveTx(hash, outofbound, unreserve)
}
return ret
}

4
core/types.go
View file

@ -34,6 +34,10 @@ type Validator interface {
// ValidateState validates the given statedb and optionally the receipts and
// gas used.
ValidateState(block *types.Block, state *state.StateDB, receipts types.Receipts, usedGas uint64) error
// SetupTracerAndCircuitCapacityChecker sets up ScrollTracerWrapper and CircuitCapacityChecker for validator,
// to get scroll-related traces and to validate the circuit row consumption
SetupTracerAndCircuitCapacityChecker(tracer tracerWrapper)
}
// Prefetcher is an interface for pre-caching transaction signatures and state.

5
core/types/block.go
View file

@ -557,6 +557,11 @@ func (b *Block) CountL2Tx() int {
type Blocks []*Block
type BlockWithRowConsumption struct {
*Block
*RowConsumption
}
// HeaderParentHashFromRLP returns the parentHash of an RLP-encoded
// header. If 'header' is invalid, the zero hash is returned.
func HeaderParentHashFromRLP(header []byte) common.Hash {

45
core/types/transaction.go
View file

@ -19,17 +19,14 @@ package types
import (
"bytes"
"errors"
"fmt"
"io"
"math/big"
"sort"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/math"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/rlp"
)
@ -613,45 +610,3 @@ func copyAddressPtr(a *common.Address) *common.Address {
cpy := *a
return &cpy
}
// L1MessagesByQueueIndex represents a set of L1 messages ordered by their queue indices.
type L1MessagesByQueueIndex struct {
msgs []L1MessageTx
}
func NewL1MessagesByQueueIndex(msgs []L1MessageTx) (*L1MessagesByQueueIndex, error) {
// sort by queue index
sort.Slice(msgs, func(i, j int) bool {
return msgs[i].QueueIndex < msgs[j].QueueIndex
})
// check for duplicates/gaps
for ii := 0; ii < len(msgs)-1; ii++ {
current := msgs[ii].QueueIndex
next := msgs[ii+1].QueueIndex
if next != current+1 {
return nil, fmt.Errorf("invalid L1 message set, current index: %d, next index: %d", current, next)
}
}
return &L1MessagesByQueueIndex{msgs: msgs}, nil
}
func (t *L1MessagesByQueueIndex) Peek() *Transaction {
if len(t.msgs) == 0 {
return nil
}
return NewTx(&t.msgs[0])
}
func (t *L1MessagesByQueueIndex) Shift() {
t.msgs = t.msgs[1:]
}
func (t *L1MessagesByQueueIndex) Pop() {
log.Error("Pop() is called on L1MessagesByQueueIndex")
// this is a logic error, the intention should be "Shift()",
// so we will follow the same behavior in Pop
t.Shift()
}

5
eth/backend.go
View file

@ -59,6 +59,7 @@ import (
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rollup/rollup_sync_service"
"github.com/ethereum/go-ethereum/rollup/sync_service"
"github.com/ethereum/go-ethereum/rollup/tracing"
"github.com/ethereum/go-ethereum/rpc"
)
@ -218,6 +219,10 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client sync_service.EthCl
if err != nil {
return nil, err
}
if config.CheckCircuitCapacity {
tracer := tracing.NewTracerWrapper()
eth.blockchain.Validator().SetupTracerAndCircuitCapacityChecker(tracer)
}
eth.bloomIndexer.Start(eth.blockchain)
if config.BlobPool.Datadir != "" {

3
eth/ethconfig/config.go
View file

@ -170,6 +170,9 @@ type Config struct {
// OverrideVerkle (TODO: remove after the fork)
OverrideVerkle *uint64 `toml:",omitempty"`
// Check circuit capacity in block validator
CheckCircuitCapacity bool
// Enable verification of batch consistency between L1 and L2 in rollup
EnableRollupVerify bool

6
eth/ethconfig/gen_config.go
View file

@ -56,6 +56,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
RPCTxFeeCap float64
OverrideCancun *uint64 `toml:",omitempty"`
OverrideVerkle *uint64 `toml:",omitempty"`
CheckCircuitCapacity bool
EnableRollupVerify bool
MaxBlockRange int64
}
@ -99,6 +100,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
enc.RPCTxFeeCap = c.RPCTxFeeCap
enc.OverrideCancun = c.OverrideCancun
enc.OverrideVerkle = c.OverrideVerkle
enc.CheckCircuitCapacity = c.CheckCircuitCapacity
enc.EnableRollupVerify = c.EnableRollupVerify
enc.MaxBlockRange = c.MaxBlockRange
return &enc, nil
@ -146,6 +148,7 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
RPCTxFeeCap *float64
OverrideCancun *uint64 `toml:",omitempty"`
OverrideVerkle *uint64 `toml:",omitempty"`
CheckCircuitCapacity *bool
EnableRollupVerify *bool
MaxBlockRange *int64
}
@ -270,6 +273,9 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
if dec.OverrideVerkle != nil {
c.OverrideVerkle = dec.OverrideVerkle
}
if dec.CheckCircuitCapacity != nil {
c.CheckCircuitCapacity = *dec.CheckCircuitCapacity
}
if dec.EnableRollupVerify != nil {
c.EnableRollupVerify = *dec.EnableRollupVerify
}

6
miner/miner.go
View file

@ -31,9 +31,11 @@ import (
"github.com/ethereum/go-ethereum/core/txpool"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rollup/sync_service"
)
// Backend wraps all methods required for mining. Only full node is capable
@ -41,6 +43,8 @@ import (
type Backend interface {
BlockChain() *core.BlockChain
TxPool() *txpool.TxPool
ChainDb() ethdb.Database
SyncService() *sync_service.SyncService
}
// Config is the configuration parameters of mining.
@ -53,6 +57,8 @@ type Config struct {
Recommit time.Duration // The time interval for miner to re-create mining work.
NewPayloadTimeout time.Duration // The maximum time allowance for creating a new payload
StoreSkippedTxTraces bool // Whether store the wrapped traces when storing a skipped tx
}
// DefaultConfig contains default settings for miner.

16
miner/miner_test.go
View file

@ -34,20 +34,24 @@ import (
"github.com/ethereum/go-ethereum/core/vm"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rollup/sync_service"
"github.com/ethereum/go-ethereum/trie"
)
type mockBackend struct {
bc *core.BlockChain
txPool *txpool.TxPool
chainDb ethdb.Database
}
func NewMockBackend(bc *core.BlockChain, txPool *txpool.TxPool) *mockBackend {
func NewMockBackend(bc *core.BlockChain, txPool *txpool.TxPool, chainDb ethdb.Database) *mockBackend {
return &mockBackend{
bc: bc,
txPool: txPool,
chainDb: chainDb,
}
}
@ -59,6 +63,14 @@ func (m *mockBackend) TxPool() *txpool.TxPool {
return m.txPool
}
func (m *mockBackend) ChainDb() ethdb.Database {
return m.chainDb
}
func (m *mockBackend) SyncService() *sync_service.SyncService {
return nil
}
func (m *mockBackend) StateAtBlock(block *types.Block, reexec uint64, base *state.StateDB, checkLive bool, preferDisk bool) (statedb *state.StateDB, err error) {
return nil, errors.New("not supported")
}
@ -312,7 +324,7 @@ func createMiner(t *testing.T) (*Miner, *event.TypeMux, func(skipMiner bool)) {
pool := legacypool.New(testTxPoolConfig, blockchain)
txpool, _ := txpool.New(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain, []txpool.SubPool{pool})
backend := NewMockBackend(bc, txpool)
backend := NewMockBackend(bc, txpool, chainDB)
// Create event Mux
mux := new(event.TypeMux)
// Create Miner

58
miner/ordering.go
View file

@ -18,12 +18,15 @@ package miner
import (
"container/heap"
"fmt"
"math/big"
"sort"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/math"
"github.com/ethereum/go-ethereum/core/txpool"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/log"
)
// txWithMinerFee wraps a transaction with its gas price or effective miner gasTipCap
@ -145,3 +148,58 @@ func (t *transactionsByPriceAndNonce) Shift() {
func (t *transactionsByPriceAndNonce) Pop() {
heap.Pop(&t.heads)
}
// l1MessagesByQueueIndex represents a set of L1 messages ordered by their queue indices.
type l1MessagesByQueueIndex struct {
txPool *txpool.TxPool
msgs []types.L1MessageTx
}
func newL1MessagesByQueueIndex(txPool *txpool.TxPool, msgs []types.L1MessageTx) (*l1MessagesByQueueIndex, error) {
// sort by queue index
sort.Slice(msgs, func(i, j int) bool {
return msgs[i].QueueIndex < msgs[j].QueueIndex
})
// check for duplicates/gaps
for ii := 0; ii < len(msgs)-1; ii++ {
current := msgs[ii].QueueIndex
next := msgs[ii+1].QueueIndex
if next != current+1 {
return nil, fmt.Errorf("invalid L1 message set, current index: %d, next index: %d", current, next)
}
}
return &l1MessagesByQueueIndex{txPool: txPool, msgs: msgs}, nil
}
func (t *l1MessagesByQueueIndex) Peek() *txpool.LazyTransaction {
if len(t.msgs) == 0 {
return nil
}
return txToLazyTx(t.txPool, types.NewTx(&t.msgs[0]))
}
func (t *l1MessagesByQueueIndex) Shift() {
t.msgs = t.msgs[1:]
}
func (t *l1MessagesByQueueIndex) Pop() {
log.Error("Pop() is called on l1MessagesByQueueIndex")
// this is a logic error, the intention should be "Shift()",
// so we will follow the same behavior in Pop
t.Shift()
}
// orderedTransactionSet represents a set of transactions and some ordering on top of this set.
type orderedTransactionSet interface {
// Peek returns the next transaction.
Peek() *txpool.LazyTransaction
// Shift removes the next transaction.
Shift()
// Pop removes all transactions from the current account.
Pop()
}

613
miner/worker.go
View file

@ -29,13 +29,17 @@ import (
"github.com/ethereum/go-ethereum/consensus/misc/eip1559"
"github.com/ethereum/go-ethereum/consensus/misc/eip4844"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/rawdb"
"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/metrics"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rollup/circuitcapacitychecker"
"github.com/ethereum/go-ethereum/rollup/tracing"
"github.com/ethereum/go-ethereum/trie"
)
@ -79,6 +83,40 @@ var (
errBlockInterruptedByTimeout = errors.New("timeout while building block")
)
var (
// Metrics for the skipped txs
l1TxGasLimitExceededCounter = metrics.NewRegisteredCounter("miner/skipped_txs/l1/gas_limit_exceeded", nil)
l1TxRowConsumptionOverflowCounter = metrics.NewRegisteredCounter("miner/skipped_txs/l1/row_consumption_overflow", nil)
l2TxRowConsumptionOverflowCounter = metrics.NewRegisteredCounter("miner/skipped_txs/l2/row_consumption_overflow", nil)
l1TxCccUnknownErrCounter = metrics.NewRegisteredCounter("miner/skipped_txs/l1/ccc_unknown_err", nil)
l2TxCccUnknownErrCounter = metrics.NewRegisteredCounter("miner/skipped_txs/l2/ccc_unknown_err", nil)
l1TxStrangeErrCounter = metrics.NewRegisteredCounter("miner/skipped_txs/l1/strange_err", nil)
l2CommitTxsTimer = metrics.NewRegisteredTimer("miner/commit/txs_all", nil)
l2CommitTxTimer = metrics.NewRegisteredTimer("miner/commit/tx_all", nil)
l2CommitTxFailedTimer = metrics.NewRegisteredTimer("miner/commit/tx_all_failed", nil)
l2CommitTxTraceTimer = metrics.NewRegisteredTimer("miner/commit/tx_trace", nil)
l2CommitTxTraceStateRevertTimer = metrics.NewRegisteredTimer("miner/commit/tx_trace_state_revert", nil)
l2CommitTxCCCTimer = metrics.NewRegisteredTimer("miner/commit/tx_ccc", nil)
l2CommitTxApplyTimer = metrics.NewRegisteredTimer("miner/commit/tx_apply", nil)
l2CommitNewWorkTimer = metrics.NewRegisteredTimer("miner/commit/new_work_all", nil)
l2CommitNewWorkL1CollectTimer = metrics.NewRegisteredTimer("miner/commit/new_work_collect_l1", nil)
l2CommitNewWorkPrepareTimer = metrics.NewRegisteredTimer("miner/commit/new_work_prepare", nil)
l2CommitNewWorkCommitUncleTimer = metrics.NewRegisteredTimer("miner/commit/new_work_uncle", nil)
l2CommitNewWorkTidyPendingTxTimer = metrics.NewRegisteredTimer("miner/commit/new_work_tidy_pending", nil)
l2CommitNewWorkCommitL1MsgTimer = metrics.NewRegisteredTimer("miner/commit/new_work_commit_l1_msg", nil)
l2CommitNewWorkPrioritizedTxCommitTimer = metrics.NewRegisteredTimer("miner/commit/new_work_prioritized", nil)
l2CommitNewWorkRemoteLocalCommitTimer = metrics.NewRegisteredTimer("miner/commit/new_work_remote_local", nil)
l2CommitNewWorkLocalPriceAndNonceTimer = metrics.NewRegisteredTimer("miner/commit/new_work_local_price_and_nonce", nil)
l2CommitNewWorkRemotePriceAndNonceTimer = metrics.NewRegisteredTimer("miner/commit/new_work_remote_price_and_nonce", nil)
l2CommitTimer = metrics.NewRegisteredTimer("miner/commit/all", nil)
l2CommitTraceTimer = metrics.NewRegisteredTimer("miner/commit/trace", nil)
l2CommitCCCTimer = metrics.NewRegisteredTimer("miner/commit/ccc", nil)
l2ResultTimer = metrics.NewRegisteredTimer("miner/result/all", nil)
)
// environment is the worker's current environment and holds all
// information of the sealing block generation.
type environment struct {
@ -94,6 +132,13 @@ type environment struct {
receipts []*types.Receipt
sidecars []*types.BlobTxSidecar
blobs int
l1TxCount int // l1 msg count in cycle
// circuit capacity check related fields
traceEnv *tracing.TraceEnv // env for tracing
accRows *types.RowConsumption // accumulated row consumption for a block
nextL1MsgIndex uint64 // next L1 queue index to be processed
}
// copy creates a deep copy of environment.
@ -135,6 +180,9 @@ type task struct {
state *state.StateDB
block *types.Block
createdAt time.Time
accRows *types.RowConsumption // accumulated row consumption in the circuit side
nextL1MsgIndex uint64 // next L1 queue index to be processed
}
const (
@ -170,6 +218,13 @@ type intervalAdjust struct {
inc bool
}
// prioritizedTransaction represents a single transaction that
// should be processed as the first transaction in the next block.
type prioritizedTransaction struct {
blockNumber uint64
tx *types.Transaction
}
// worker is the main object which takes care of submitting new work to consensus engine
// and gathering the sealing result.
type worker struct {
@ -188,6 +243,8 @@ type worker struct {
txsSub event.Subscription
chainHeadCh chan core.ChainHeadEvent
chainHeadSub event.Subscription
l1MsgsCh chan core.NewL1MsgsEvent
l1MsgsSub event.Subscription
// Channels
newWorkCh chan *newWorkReq
@ -219,6 +276,7 @@ type worker struct {
running atomic.Bool // The indicator whether the consensus engine is running or not.
newTxs atomic.Int32 // New arrival transaction count since last sealing work submitting.
syncing atomic.Bool // The indicator whether the node is still syncing.
newL1Msgs atomic.Int32 // New arrival L1 message count since last sealing work submitting.
// newpayloadTimeout is the maximum timeout allowance for creating payload.
// The default value is 2 seconds but node operator can set it to arbitrary
@ -234,11 +292,15 @@ type worker struct {
// External functions
isLocalBlock func(header *types.Header) bool // Function used to determine whether the specified block is mined by local miner.
circuitCapacityChecker *circuitcapacitychecker.CircuitCapacityChecker
prioritizedTx *prioritizedTransaction
// Test hooks
newTaskHook func(*task) // Method to call upon receiving a new sealing task.
skipSealHook func(*task) bool // Method to decide whether skipping the sealing.
fullTaskHook func() // Method to call before pushing the full sealing task.
resubmitHook func(time.Duration, time.Duration) // Method to call upon updating resubmitting interval.
beforeTxHook func() // Method to call before processing a transaction.
}
func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux, isLocalBlock func(header *types.Header) bool, init bool) *worker {
@ -263,12 +325,28 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus
exitCh: make(chan struct{}),
resubmitIntervalCh: make(chan time.Duration),
resubmitAdjustCh: make(chan *intervalAdjust, resubmitAdjustChanSize),
l1MsgsCh: make(chan core.NewL1MsgsEvent, txChanSize),
circuitCapacityChecker: circuitcapacitychecker.NewCircuitCapacityChecker(true),
}
log.Info("created new worker", "CircuitCapacityChecker ID", worker.circuitCapacityChecker.ID)
// 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)
// Subscribe NewL1MsgsEvent for sync service
if s := eth.SyncService(); s != nil {
worker.l1MsgsSub = s.SubscribeNewL1MsgsEvent(worker.l1MsgsCh)
} else {
// create an empty subscription so that the tests won't fail
worker.l1MsgsSub = event.NewSubscription(func(quit <-chan struct{}) error {
<-quit
return nil
})
}
// Sanitize recommit interval if the user-specified one is too short.
recommit := worker.config.Recommit
if recommit < minRecommitInterval {
@ -301,6 +379,12 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus
return worker
}
// getCCC returns a pointer to this worker's CCC instance.
// Only used in tests.
func (w *worker) getCCC() *circuitcapacitychecker.CircuitCapacityChecker {
return w.circuitCapacityChecker
}
// setEtherbase sets the etherbase used to initialize the block coinbase field.
func (w *worker) setEtherbase(addr common.Address) {
w.mu.Lock()
@ -435,6 +519,7 @@ func (w *worker) newWorkLoop(recommit time.Duration) {
}
timer.Reset(recommit)
w.newTxs.Store(0)
w.newL1Msgs.Store(0)
}
// clearPending cleans the stale pending tasks.
clearPending := func(number uint64) {
@ -464,7 +549,7 @@ func (w *worker) newWorkLoop(recommit time.Duration) {
// higher priced transactions. Disable this overhead for pending blocks.
if w.isRunning() && (w.chainConfig.Clique == nil || w.chainConfig.Clique.Period > 0) {
// Short circuit if no new transaction arrives.
if w.newTxs.Load() == 0 {
if w.newTxs.Load() == 0 && w.newL1Msgs.Load() == 0 {
timer.Reset(recommit)
continue
}
@ -513,6 +598,7 @@ func (w *worker) newWorkLoop(recommit time.Duration) {
func (w *worker) mainLoop() {
defer w.wg.Done()
defer w.txsSub.Unsubscribe()
defer w.l1MsgsSub.Unsubscribe()
defer w.chainHeadSub.Unsubscribe()
defer func() {
if w.current != nil {
@ -524,6 +610,7 @@ func (w *worker) mainLoop() {
select {
case req := <-w.newWorkCh:
w.commitWork(req.interrupt, req.timestamp)
// new block created.
case req := <-w.getWorkCh:
req.result <- w.generateWork(req.params)
@ -572,11 +659,16 @@ func (w *worker) mainLoop() {
}
w.newTxs.Add(int32(len(ev.Txs)))
case ev := <-w.l1MsgsCh:
w.newL1Msgs.Add(int32(ev.Count))
// System stopped
case <-w.exitCh:
return
case <-w.txsSub.Err():
return
case <-w.l1MsgsSub.Err():
return
case <-w.chainHeadSub.Err():
return
}
@ -660,6 +752,7 @@ func (w *worker) resultLoop() {
log.Error("Block found but no relative pending task", "number", block.Number(), "sealhash", sealhash, "hash", hash)
continue
}
startTime := time.Now()
// Different block could share same sealhash, deep copy here to prevent write-write conflict.
var (
receipts = make([]*types.Receipt, len(task.receipts))
@ -686,9 +779,42 @@ func (w *worker) resultLoop() {
}
logs = append(logs, receipt.Logs...)
}
// It's possible that we've stored L1 queue index for this block previously,
// in this case do not overwrite it.
if index := rawdb.ReadFirstQueueIndexNotInL2Block(w.eth.ChainDb(), hash); index == nil {
// Store first L1 queue index not processed by this block.
// Note: This accounts for both included and skipped messages. This
// way, if a block only skips messages, we won't reprocess the same
// messages from the next block.
log.Trace(
"Worker WriteFirstQueueIndexNotInL2Block",
"number", block.Number(),
"hash", hash.String(),
"task.nextL1MsgIndex", task.nextL1MsgIndex,
)
rawdb.WriteFirstQueueIndexNotInL2Block(w.eth.ChainDb(), hash, task.nextL1MsgIndex)
} else {
log.Trace(
"Worker WriteFirstQueueIndexNotInL2Block: not overwriting existing index",
"number", block.Number(),
"hash", hash.String(),
"index", *index,
"task.nextL1MsgIndex", task.nextL1MsgIndex,
)
}
// Store circuit row consumption.
log.Trace(
"Worker write block row consumption",
"id", w.circuitCapacityChecker.ID,
"number", block.Number(),
"hash", hash.String(),
"accRows", task.accRows,
)
rawdb.WriteBlockRowConsumption(w.eth.ChainDb(), hash, task.accRows)
// Commit block and state to database.
_, err := w.chain.WriteBlockAndSetHead(block, receipts, logs, task.state, true)
if err != nil {
l2ResultTimer.Update(time.Since(startTime))
log.Error("Failed writing block to chain", "err", err)
continue
}
@ -698,6 +824,8 @@ func (w *worker) resultLoop() {
// Broadcast the block and announce chain insertion event
w.mux.Post(core.NewMinedBlockEvent{Block: block})
l2ResultTimer.Update(time.Since(startTime))
case <-w.exitCh:
return
}
@ -712,6 +840,18 @@ func (w *worker) makeEnv(parent *types.Header, header *types.Header, coinbase co
if err != nil {
return nil, err
}
// don't finalize the state during tracing for circuit capacity checker, otherwise we cannot revert.
// and even if we don't finalize the state, the `refund` value will still be correct, as explained in `CommitTransaction`
finaliseStateAfterApply := false
traceEnv, err := tracing.CreateTraceEnv(w.chainConfig, w.chain, w.engine, w.eth.ChainDb(), state, parent,
// new block with a placeholder tx, for traceEnv's ExecutionResults length & TxStorageTraces length
types.NewBlockWithHeader(header).WithBody([]*types.Transaction{types.NewTx(&types.LegacyTx{})}, nil),
finaliseStateAfterApply)
if err != nil {
return nil, err
}
state.StartPrefetcher("miner")
// Note the passed coinbase may be different with header.Coinbase.
@ -720,10 +860,15 @@ func (w *worker) makeEnv(parent *types.Header, header *types.Header, coinbase co
state: state,
coinbase: coinbase,
header: header,
traceEnv: traceEnv,
accRows: nil,
}
// Keep track of transactions which return errors so they can be removed
env.tcount = 0
env.blockSize = 0
env.blockSize = 0
env.l1TxCount = 0
env.nextL1MsgIndex = traceEnv.StartL1QueueIndex
return env, nil
}
@ -743,20 +888,21 @@ func (w *worker) updateSnapshot(env *environment) {
w.snapshotState = env.state.Copy()
}
func (w *worker) commitTransaction(env *environment, tx *types.Transaction) ([]*types.Log, error) {
func (w *worker) commitTransaction(env *environment, tx *types.Transaction) ([]*types.Log, *types.BlockTrace, error) {
if tx.Type() == types.BlobTxType {
return w.commitBlobTransaction(env, tx)
}
receipt, err := w.applyTransaction(env, tx)
receipt, traces, accRows, err := w.applyTransaction(env, tx)
if err != nil {
return nil, err
return nil, nil, err
}
env.txs = append(env.txs, tx)
env.receipts = append(env.receipts, receipt)
return receipt.Logs, nil
env.accRows = accRows
return receipt.Logs, traces, nil
}
func (w *worker) commitBlobTransaction(env *environment, tx *types.Transaction) ([]*types.Log, error) {
func (w *worker) commitBlobTransaction(env *environment, tx *types.Transaction) ([]*types.Log, *types.BlockTrace, error) {
sc := tx.BlobTxSidecar()
if sc == nil {
panic("blob transaction without blobs in miner")
@ -766,48 +912,146 @@ func (w *worker) commitBlobTransaction(env *environment, tx *types.Transaction)
// and not during execution. This means core.ApplyTransaction will not return an error if the
// tx has too many blobs. So we have to explicitly check it here.
if (env.blobs+len(sc.Blobs))*params.BlobTxBlobGasPerBlob > params.MaxBlobGasPerBlock {
return nil, errors.New("max data blobs reached")
return nil, nil, errors.New("max data blobs reached")
}
receipt, err := w.applyTransaction(env, tx)
receipt, traces, accRows, err := w.applyTransaction(env, tx)
if err != nil {
return nil, err
return nil, nil, err
}
env.txs = append(env.txs, tx.WithoutBlobTxSidecar())
env.receipts = append(env.receipts, receipt)
env.accRows = accRows
env.sidecars = append(env.sidecars, sc)
env.blobs += len(sc.Blobs)
*env.header.BlobGasUsed += receipt.BlobGasUsed
return receipt.Logs, nil
return receipt.Logs, traces, nil
}
// applyTransaction runs the transaction. If execution fails, state and gas pool are reverted.
func (w *worker) applyTransaction(env *environment, tx *types.Transaction) (*types.Receipt, error) {
func (w *worker) applyTransaction(env *environment, tx *types.Transaction) (*types.Receipt, *types.BlockTrace, *types.RowConsumption, error) {
var (
snap = env.state.Snapshot()
traces *types.BlockTrace
accRows *types.RowConsumption
receipt *types.Receipt
err error
)
// do not do CCC checks on follower nodes
if w.isRunning() {
defer func(t0 time.Time) {
l2CommitTxTimer.Update(time.Since(t0))
if err != nil {
l2CommitTxFailedTimer.Update(time.Since(t0))
}
}(time.Now())
// do gas limit check up-front and do not run CCC if it fails
if env.gasPool.Gas() < tx.Gas() {
return nil, nil, nil, core.ErrGasLimitReached
}
snap := env.state.Snapshot()
log.Trace(
"Worker apply ccc for tx",
"id", w.circuitCapacityChecker.ID,
"txHash", tx.Hash().Hex(),
)
// 1. we have to check circuit capacity before `core.ApplyTransaction`,
// because if the tx can be successfully executed but circuit capacity overflows, it will be inconvenient to revert.
// 2. even if we don't commit to the state during the tracing (which means `clearJournalAndRefund` is not called during the tracing),
// the `refund` value will still be correct, because:
// 2.1 when starting handling the first tx, `state.refund` is 0 by default,
// 2.2 after tracing, the state is either committed in `core.ApplyTransaction`, or reverted, so the `state.refund` can be cleared,
// 2.3 when starting handling the following txs, `state.refund` comes as 0
withTimer(l2CommitTxTraceTimer, func() {
traces, err = env.traceEnv.GetBlockTrace(
types.NewBlockWithHeader(env.header).WithBody([]*types.Transaction{tx}, nil),
)
})
withTimer(l2CommitTxTraceStateRevertTimer, func() {
// `env.traceEnv.State` & `env.state` share a same pointer to the state, so only need to revert `env.state`
// revert to snapshot for calling `core.ApplyMessage` again, (both `traceEnv.GetBlockTrace` & `core.ApplyTransaction` will call `core.ApplyMessage`)
env.state.RevertToSnapshot(snap)
})
if err != nil {
return nil, nil, nil, err
}
withTimer(l2CommitTxCCCTimer, func() {
accRows, err = w.circuitCapacityChecker.ApplyTransaction(traces)
})
if err != nil {
return nil, traces, accRows, err
}
log.Trace(
"Worker apply ccc for tx result",
"id", w.circuitCapacityChecker.ID,
"txHash", tx.Hash().Hex(),
"accRows", accRows,
)
}
var (
snap = env.state.Snapshot() // create new snapshot for `core.ApplyTransaction`
gp = env.gasPool.Gas()
)
receipt, err := core.ApplyTransaction(w.chainConfig, w.chain, &env.coinbase, env.gasPool, env.state, env.header, tx, &env.header.GasUsed, *w.chain.GetVMConfig())
withTimer(l2CommitTxApplyTimer, func() {
receipt, err = core.ApplyTransaction(w.chainConfig, w.chain, &env.coinbase, env.gasPool, env.state, env.header, tx, &env.header.GasUsed, *w.chain.GetVMConfig())
})
if err != nil {
env.state.RevertToSnapshot(snap)
env.gasPool.SetGas(gp)
if accRows != nil {
// At this point, we have called CCC but the transaction failed in `ApplyTransaction`.
// If we skip this tx and continue to pack more, the next tx will likely fail with
// `circuitcapacitychecker.ErrUnknown`. However, at this point we cannot decide whether
// we should seal the block or skip the tx and continue, so we simply return the error.
log.Error(
"GetBlockTrace passed but ApplyTransaction failed, ccc is left in inconsistent state",
"blockNumber", env.header.Number,
"txHash", tx.Hash().Hex(),
"err", err,
)
}
return receipt, err
}
return receipt, traces, accRows, err
}
func (w *worker) commitTransactions(env *environment, txs *transactionsByPriceAndNonce, interrupt *atomic.Int32) error {
func (w *worker) commitTransactions(env *environment, txs orderedTransactionSet, interrupt *atomic.Int32) (bool, error) {
defer func(t0 time.Time) {
l2CommitTxsTimer.Update(time.Since(t0))
}(time.Now())
var circuitCapacityReached bool
gasLimit := env.header.GasLimit
if env.gasPool == nil {
env.gasPool = new(core.GasPool).AddGas(gasLimit)
}
var coalescedLogs []*types.Log
var loops int64
loop:
for {
if w.beforeTxHook != nil {
w.beforeTxHook()
}
loops++
// Check interruption signal and abort building if it's fired.
if interrupt != nil {
if signal := interrupt.Load(); signal != commitInterruptNone {
return signalToErr(signal)
return circuitCapacityReached, signalToErr(signal)
}
}
// seal block early if we're over time
// note: current.header.Time = max(parent.Time + cliquePeriod, now())
if env.tcount > 0 && w.chainConfig.Clique != nil && uint64(time.Now().Unix()) > env.header.Time {
circuitCapacityReached = true // skip subsequent invocations of commitTransactions
break
}
// 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)
@ -818,6 +1062,12 @@ func (w *worker) commitTransactions(env *environment, txs *transactionsByPriceAn
if ltx == nil {
break
}
// If we have collected enough transactions then we're done
// Originally we only limit l2txs count, but now strictly limit total txs number.
if !w.chainConfig.Scroll.IsValidTxCount(env.tcount + 1) {
log.Trace("Transaction count limit reached", "have", env.tcount, "want", w.chainConfig.Scroll.MaxTxPerBlock)
break
}
// 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)
@ -836,11 +1086,12 @@ func (w *worker) commitTransactions(env *environment, txs *transactionsByPriceAn
txs.Pop()
continue
}
// If we have collected enough transactions then we're done
// Originally we only limit l2txs count, but now strictly limit total txs number.
// log.Info("w.chainConfig", "w.chainConfig.Scroll", w.chainConfig.Scroll)
if !w.chainConfig.Scroll.IsValidTxCount(env.tcount + 1) {
log.Trace("Transaction count limit reached", "have", env.tcount, "want", w.chainConfig.Scroll.MaxTxPerBlock)
if tx.IsL1MessageTx() && tx.AsL1MessageTx().QueueIndex != env.nextL1MsgIndex {
log.Error(
"Unexpected L1 message queue index in worker",
"expected", env.nextL1MsgIndex,
"got", tx.AsL1MessageTx().QueueIndex,
)
break
}
if !tx.IsL1MessageTx() && !w.chainConfig.Scroll.IsValidBlockSize(env.blockSize+tx.Size()) {
@ -862,7 +1113,7 @@ func (w *worker) commitTransactions(env *environment, txs *transactionsByPriceAn
// Start executing the transaction
env.state.SetTxContext(tx.Hash(), env.tcount)
logs, err := w.commitTransaction(env, tx)
logs, traces, err := w.commitTransaction(env, tx)
switch {
case errors.Is(err, core.ErrNonceTooLow):
// New head notification data race between the transaction pool and miner, shift
@ -876,15 +1127,151 @@ func (w *worker) commitTransactions(env *environment, txs *transactionsByPriceAn
txs.Shift()
if tx.IsL1MessageTx() {
queueIndex := tx.AsL1MessageTx().QueueIndex
log.Debug("Including L1 message", "queueIndex", queueIndex, "tx", tx.Hash().String())
env.l1TxCount++
env.nextL1MsgIndex = queueIndex + 1
} else {
// only consider block size limit for L2 transactions
env.blockSize += tx.Size()
}
case errors.Is(err, core.ErrGasLimitReached) && tx.IsL1MessageTx():
// If this block already contains some L1 messages,
// terminate here and try again in the next block.
if env.l1TxCount > 0 {
break loop
}
// A single L1 message leads to out-of-gas. Skip it.
queueIndex := tx.AsL1MessageTx().QueueIndex
log.Info("Skipping L1 message", "queueIndex", queueIndex, "tx", tx.Hash().String(), "block", env.header.Number, "reason", "gas limit exceeded")
env.nextL1MsgIndex = queueIndex + 1
txs.Shift()
if w.config.StoreSkippedTxTraces {
rawdb.WriteSkippedTransaction(w.eth.ChainDb(), tx, traces, "gas limit exceeded", env.header.Number.Uint64(), nil)
} else {
rawdb.WriteSkippedTransaction(w.eth.ChainDb(), tx, nil, "gas limit exceeded", env.header.Number.Uint64(), nil)
}
l1TxGasLimitExceededCounter.Inc(1)
// Circuit capacity check
case errors.Is(err, circuitcapacitychecker.ErrBlockRowConsumptionOverflow):
if env.tcount >= 1 {
// 1. Circuit capacity limit reached in a block, and it's not the first tx:
// don't pop or shift, just quit the loop immediately;
// though it might still be possible to add some "smaller" txs,
// but it's a trade-off between tracing overhead & block usage rate
log.Trace("Circuit capacity limit reached in a block", "acc_rows", env.accRows, "tx", tx.Hash().String())
log.Info("Skipping message", "tx", tx.Hash().String(), "block", env.header.Number, "reason", "accumulated row consumption overflow")
// Prioritize transaction for the next block.
// If there are no new L1 messages, this transaction will be the 1st transaction in the next block,
// at which point we can definitively decide if we should skip it or not.
log.Debug("Prioritizing transaction for next block", "blockNumber", env.header.Number.Uint64()+1, "tx", tx.Hash().String())
w.prioritizedTx = &prioritizedTransaction{
blockNumber: env.header.Number.Uint64() + 1,
tx: tx,
}
w.newTxs.Add(int32(1))
circuitCapacityReached = true
break loop
} else {
// 2. Circuit capacity limit reached in a block, and it's the first tx: skip the tx
log.Trace("Circuit capacity limit reached for a single tx", "tx", tx.Hash().String())
if tx.IsL1MessageTx() {
// Skip L1 message transaction,
// shift to the next from the account because we shouldn't skip the entire txs from the same account
txs.Shift()
queueIndex := tx.AsL1MessageTx().QueueIndex
log.Info("Skipping L1 message", "queueIndex", queueIndex, "tx", tx.Hash().String(), "block", env.header.Number, "reason", "first tx row consumption overflow")
env.nextL1MsgIndex = queueIndex + 1
l1TxRowConsumptionOverflowCounter.Inc(1)
} else {
// Skip L2 transaction and all other transactions from the same sender account
log.Info("Skipping L2 message", "tx", tx.Hash().String(), "block", env.header.Number, "reason", "first tx row consumption overflow")
txs.Pop()
w.eth.TxPool().RemoveTx(tx.Hash(), true, true)
l2TxRowConsumptionOverflowCounter.Inc(1)
}
// Reset ccc so that we can process other transactions for this block
w.circuitCapacityChecker.Reset()
log.Trace("Worker reset ccc", "id", w.circuitCapacityChecker.ID)
circuitCapacityReached = false
// Store skipped transaction in local db
if w.config.StoreSkippedTxTraces {
rawdb.WriteSkippedTransaction(w.eth.ChainDb(), tx, traces, "row consumption overflow", env.header.Number.Uint64(), nil)
} else {
rawdb.WriteSkippedTransaction(w.eth.ChainDb(), tx, nil, "row consumption overflow", env.header.Number.Uint64(), nil)
}
}
case (errors.Is(err, circuitcapacitychecker.ErrUnknown) && tx.IsL1MessageTx()):
// Circuit capacity check: unknown circuit capacity checker error for L1MessageTx,
// shift to the next from the account because we shouldn't skip the entire txs from the same account
queueIndex := tx.AsL1MessageTx().QueueIndex
log.Trace("Unknown circuit capacity checker error for L1MessageTx", "tx", tx.Hash().String(), "queueIndex", queueIndex)
log.Info("Skipping L1 message", "queueIndex", queueIndex, "tx", tx.Hash().String(), "block", env.header.Number, "reason", "unknown row consumption error")
env.nextL1MsgIndex = queueIndex + 1
// TODO: propagate more info about the error from CCC
if w.config.StoreSkippedTxTraces {
rawdb.WriteSkippedTransaction(w.eth.ChainDb(), tx, traces, "unknown circuit capacity checker error", env.header.Number.Uint64(), nil)
} else {
rawdb.WriteSkippedTransaction(w.eth.ChainDb(), tx, nil, "unknown circuit capacity checker error", env.header.Number.Uint64(), nil)
}
l1TxCccUnknownErrCounter.Inc(1)
// Normally we would do `txs.Shift()` here.
// However, after `ErrUnknown`, ccc might remain in an
// inconsistent state, so we cannot pack more transactions.
circuitCapacityReached = true
w.checkCurrentTxNumWithCCC(env.tcount)
break loop
case (errors.Is(err, circuitcapacitychecker.ErrUnknown) && !tx.IsL1MessageTx()):
// Circuit capacity check: unknown circuit capacity checker error for L2MessageTx, skip the account
log.Trace("Unknown circuit capacity checker error for L2MessageTx", "tx", tx.Hash().String())
log.Info("Skipping L2 message", "tx", tx.Hash().String(), "block", env.header.Number, "reason", "unknown row consumption error")
// TODO: propagate more info about the error from CCC
if w.config.StoreSkippedTxTraces {
rawdb.WriteSkippedTransaction(w.eth.ChainDb(), tx, traces, "unknown circuit capacity checker error", env.header.Number.Uint64(), nil)
} else {
rawdb.WriteSkippedTransaction(w.eth.ChainDb(), tx, nil, "unknown circuit capacity checker error", env.header.Number.Uint64(), nil)
}
l2TxCccUnknownErrCounter.Inc(1)
// Normally we would do `txs.Pop()` here.
// However, after `ErrUnknown`, ccc might remain in an
// inconsistent state, so we cannot pack more transactions.
w.eth.TxPool().RemoveTx(tx.Hash(), true, true)
circuitCapacityReached = true
w.checkCurrentTxNumWithCCC(env.tcount)
break loop
case (errors.Is(err, core.ErrInsufficientFunds) || errors.Is(errors.Unwrap(err), core.ErrInsufficientFunds)):
log.Trace("Skipping tx with insufficient funds", "sender", from, "tx", tx.Hash().String())
txs.Pop()
w.eth.TxPool().RemoveTx(tx.Hash(), true, true)
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)
log.Debug("Transaction failed, account skipped", "hash", ltx.Hash.String(), "err", err)
if tx.IsL1MessageTx() {
queueIndex := tx.AsL1MessageTx().QueueIndex
log.Info("Skipping L1 message", "queueIndex", queueIndex, "tx", tx.Hash().String(), "block", env.header.Number, "reason", "strange error", "err", err)
env.nextL1MsgIndex = queueIndex + 1
if w.config.StoreSkippedTxTraces {
rawdb.WriteSkippedTransaction(w.eth.ChainDb(), tx, traces, fmt.Sprintf("strange error: %v", err), env.header.Number.Uint64(), nil)
} else {
rawdb.WriteSkippedTransaction(w.eth.ChainDb(), tx, nil, fmt.Sprintf("strange error: %v", err), env.header.Number.Uint64(), nil)
}
l1TxStrangeErrCounter.Inc(1)
}
txs.Pop()
}
}
@ -903,7 +1290,7 @@ func (w *worker) commitTransactions(env *environment, txs *transactionsByPriceAn
}
w.pendingLogsFeed.Send(cpy)
}
return nil
return circuitCapacityReached, nil
}
// generateParams wraps various of settings for generating sealing task.
@ -1001,10 +1388,45 @@ func (w *worker) prepareWork(genParams *generateParams) (*environment, error) {
return env, nil
}
func txToLazyTx(txPool *txpool.TxPool, tx *types.Transaction) *txpool.LazyTransaction {
if tx.IsL1MessageTx() {
return &txpool.LazyTransaction{
Pool: nil, // we should never resolve a L1MessageTx from the txpool and we never need to
Hash: tx.Hash(),
Tx: tx, // set the tx directly, we don't need to resolve it
Time: tx.Time(),
GasFeeCap: tx.GasFeeCap(),
GasTipCap: tx.GasTipCap(),
Gas: tx.Gas(),
BlobGas: tx.BlobGas(),
}
}
return &txpool.LazyTransaction{
Pool: txPool,
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(),
}
}
// fillTransactions retrieves the pending transactions from the txpool and fills them
// into the given sealing block. The transaction selection and ordering strategy can
// be customized with the plugin in the future.
func (w *worker) fillTransactions(interrupt *atomic.Int32, env *environment) error {
// fetch l1Txs
var l1Messages []types.L1MessageTx
if w.chainConfig.Scroll.ShouldIncludeL1Messages() {
withTimer(l2CommitNewWorkL1CollectTimer, func() {
l1Messages = w.collectPendingL1Messages(env.nextL1MsgIndex)
})
}
tidyPendingStart := time.Now()
pending := w.eth.TxPool().Pending(true)
// Split the pending transactions into locals and remotes.
@ -1015,20 +1437,67 @@ func (w *worker) fillTransactions(interrupt *atomic.Int32, env *environment) err
localTxs[account] = txs
}
}
l2CommitNewWorkTidyPendingTxTimer.UpdateSince(tidyPendingStart)
// Fill the block with all available pending transactions.
if len(localTxs) > 0 {
var circuitCapacityReached bool
var err error
commitL1MsgStart := time.Now()
if w.chainConfig.Scroll.ShouldIncludeL1Messages() && len(l1Messages) > 0 {
log.Trace("Processing L1 messages for inclusion", "count", len(l1Messages))
txs, err := newL1MessagesByQueueIndex(w.eth.TxPool(), l1Messages)
if err != nil {
log.Error("Failed to create L1 message set", "l1Messages", l1Messages, "err", err)
return err
}
circuitCapacityReached, err = w.commitTransactions(env, txs, interrupt)
if err != nil {
l2CommitNewWorkCommitL1MsgTimer.UpdateSince(commitL1MsgStart)
return err
}
}
l2CommitNewWorkCommitL1MsgTimer.UpdateSince(commitL1MsgStart)
prioritizedTxStart := time.Now()
if w.prioritizedTx != nil && w.current.header.Number.Uint64() > w.prioritizedTx.blockNumber {
w.prioritizedTx = nil
}
if !circuitCapacityReached && w.prioritizedTx != nil && w.current.header.Number.Uint64() == w.prioritizedTx.blockNumber {
tx := w.prioritizedTx.tx
from, _ := types.Sender(w.current.signer, tx) // error already checked before
// we don't know where this came from, yolo resolve from everywhere (w.eth.TxPool())
txList := map[common.Address][]*txpool.LazyTransaction{from: {txToLazyTx(w.eth.TxPool(), tx)}}
// usually we should distinguish l1txs and l2txs:
// use `newL1MessagesByQueueIndex` for l1txs and `newTransactionsByPriceAndNonce` is for l2txs;
// but here there's only 1 tx, and hence no need for sorting, we could just simply use `newTransactionsByPriceAndNonce`
// (but we fill the LazyTransaction's tx first, in case it's a l1tx and cannot be resolved from the mempool).
txs := newTransactionsByPriceAndNonce(w.current.signer, txList, env.header.BaseFee)
circuitCapacityReached, err = w.commitTransactions(env, txs, interrupt)
if err != nil {
l2CommitNewWorkPrioritizedTxCommitTimer.UpdateSince(prioritizedTxStart)
return err
}
}
l2CommitNewWorkPrioritizedTxCommitTimer.UpdateSince(prioritizedTxStart)
remoteLocalStart := time.Now()
if !circuitCapacityReached && len(localTxs) > 0 {
localTxPriceAndNonceStart := time.Now()
txs := newTransactionsByPriceAndNonce(env.signer, localTxs, env.header.BaseFee)
if err := w.commitTransactions(env, txs, interrupt); err != nil {
l2CommitNewWorkLocalPriceAndNonceTimer.UpdateSince(localTxPriceAndNonceStart)
if circuitCapacityReached, err = w.commitTransactions(env, txs, interrupt); err != nil {
l2CommitNewWorkRemoteLocalCommitTimer.UpdateSince(remoteLocalStart)
return err
}
}
if len(remoteTxs) > 0 {
if !circuitCapacityReached && len(remoteTxs) > 0 {
remoteTxPriceAndNonceStart := time.Now()
txs := newTransactionsByPriceAndNonce(env.signer, remoteTxs, env.header.BaseFee)
if err := w.commitTransactions(env, txs, interrupt); err != nil {
l2CommitNewWorkRemotePriceAndNonceTimer.UpdateSince(remoteTxPriceAndNonceStart)
if _, err = w.commitTransactions(env, txs, interrupt); err != nil {
l2CommitNewWorkRemoteLocalCommitTimer.UpdateSince(remoteLocalStart)
return err
}
}
l2CommitNewWorkRemoteLocalCommitTimer.UpdateSince(remoteLocalStart)
return nil
}
@ -1070,7 +1539,14 @@ func (w *worker) commitWork(interrupt *atomic.Int32, timestamp int64) {
if w.syncing.Load() {
return
}
defer func(t0 time.Time) {
l2CommitNewWorkTimer.Update(time.Since(t0))
}(time.Now())
start := time.Now()
w.circuitCapacityChecker.Reset()
log.Trace("Worker reset ccc", "id", w.circuitCapacityChecker.ID)
// Set the coinbase if the worker is running or it's required
var coinbase common.Address
@ -1081,6 +1557,9 @@ func (w *worker) commitWork(interrupt *atomic.Int32, timestamp int64) {
return
}
}
// TODO:
// 1. l2CommitNewWorkPrepareTimer
// 2. no need for l2CommitNewWorkCommitUncleTimer any more?
work, err := w.prepareWork(&generateParams{
timestamp: uint64(timestamp),
coinbase: coinbase,
@ -1128,11 +1607,57 @@ func (w *worker) commitWork(interrupt *atomic.Int32, timestamp int64) {
w.current = work
}
func (w *worker) calcAndSetAccRowsForEnv(env *environment) error {
log.Trace(
"Worker apply ccc for empty block",
"id", w.circuitCapacityChecker.ID,
"number", env.header.Number,
"hash", env.header.Hash().String(),
)
var traces *types.BlockTrace
var err error
withTimer(l2CommitTraceTimer, func() {
traces, err = env.traceEnv.GetBlockTrace(types.NewBlockWithHeader(env.header))
})
if err != nil {
return err
}
if traces == nil {
log.Warn("running in light mode and traces is nil, don't update `env.accRows`")
return nil
}
// truncate ExecutionResults&TxStorageTraces, because we declare their lengths with a dummy tx before;
// however, we need to clean it up for an empty block
traces.ExecutionResults = traces.ExecutionResults[:0]
traces.TxStorageTraces = traces.TxStorageTraces[:0]
var accRows *types.RowConsumption
withTimer(l2CommitCCCTimer, func() {
accRows, err = w.circuitCapacityChecker.ApplyBlock(traces)
})
if err != nil {
return err
}
log.Trace(
"Worker apply ccc for empty block result",
"id", w.circuitCapacityChecker.ID,
"number", env.header.Number,
"hash", env.header.Hash().String(),
"accRows", accRows,
)
env.accRows = accRows
return nil
}
// commit runs any post-transaction state modifications, assembles the final block
// and commits new work if consensus engine is running.
// Note the assumption is held that the mutation is allowed to the passed env, do
// the deep copy first.
func (w *worker) commit(env *environment, interval func(), update bool, start time.Time) error {
defer func(t0 time.Time) {
l2CommitTimer.Update(time.Since(t0))
}(time.Now())
if w.isRunning() {
if interval != nil {
interval()
@ -1140,6 +1665,12 @@ func (w *worker) commit(env *environment, interval func(), update bool, start ti
// Create a local environment copy, avoid the data race with snapshot state.
// https://github.com/ethereum/go-ethereum/issues/24299
env := env.copy()
// set env.accRows for empty-but-not-genesis block
if (env.header.Number.Uint64() != 0) && (env.accRows == nil || len(*env.accRows) == 0) {
if err := w.calcAndSetAccRowsForEnv(env); err != nil {
return err
}
}
// Withdrawals are set to nil here, because this is only called in PoW.
block, err := w.engine.FinalizeAndAssemble(w.chain, env.header, env.state, env.txs, nil, env.receipts, nil)
if err != nil {
@ -1148,7 +1679,7 @@ func (w *worker) commit(env *environment, interval func(), update bool, start ti
// If we're post merge, just ignore
if !w.isTTDReached(block.Header()) {
select {
case w.taskCh <- &task{receipts: env.receipts, state: env.state, block: block, createdAt: time.Now()}:
case w.taskCh <- &task{receipts: env.receipts, state: env.state, block: block, createdAt: time.Now(), accRows: env.accRows, nextL1MsgIndex: env.nextL1MsgIndex}:
fees := totalFees(block, env.receipts)
feesInEther := new(big.Float).Quo(new(big.Float).SetInt(fees), big.NewFloat(params.Ether))
log.Info("Commit new sealing work", "number", block.Number(), "sealhash", w.engine.SealHash(block.Header()),
@ -1189,6 +1720,22 @@ func (w *worker) isTTDReached(header *types.Header) bool {
return td != nil && ttd != nil && td.Cmp(ttd) >= 0
}
func (w *worker) checkCurrentTxNumWithCCC(expected int) {
match, got, err := w.circuitCapacityChecker.CheckTxNum(expected)
if err != nil {
log.Error("failed to CheckTxNum in ccc", "err", err)
return
}
if !match {
log.Error("tx count in miner is different with CCC", "w.current.tcount", w.current.tcount, "got", got)
}
}
func (w *worker) collectPendingL1Messages(startIndex uint64) []types.L1MessageTx {
maxCount := w.chainConfig.Scroll.L1Config.NumL1MessagesPerBlock
return rawdb.ReadL1MessagesFrom(w.eth.ChainDb(), startIndex, maxCount)
}
// copyReceipts makes a deep copy of the given receipts.
func copyReceipts(receipts []*types.Receipt) []*types.Receipt {
result := make([]*types.Receipt, len(receipts))
@ -1223,3 +1770,11 @@ func signalToErr(signal int32) error {
panic(fmt.Errorf("undefined signal %d", signal))
}
}
func withTimer(timer metrics.Timer, f func()) {
if metrics.Enabled {
timer.Time(f)
} else {
f()
}
}

4
miner/worker_test.go
View file

@ -37,6 +37,7 @@ import (
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rollup/sync_service"
)
const (
@ -147,6 +148,9 @@ func newTestWorkerBackend(t *testing.T, chainConfig *params.ChainConfig, engine
func (b *testWorkerBackend) BlockChain() *core.BlockChain { return b.chain }
func (b *testWorkerBackend) TxPool() *txpool.TxPool { return b.txPool }
func (b *testWorkerBackend) ChainDb() ethdb.Database { return b.db }
func (b *testWorkerBackend) SyncService() *sync_service.SyncService { return nil }
func (b *testWorkerBackend) newRandomTx(creation bool) *types.Transaction {
var tx *types.Transaction
gasPrice := big.NewInt(10 * params.InitialBaseFee)

5
rollup/tracing/tracing.go
View file

@ -169,6 +169,11 @@ func CreateTraceEnv(chainConfig *params.ChainConfig, chainContext core.ChainCont
}
func (env *TraceEnv) GetBlockTrace(block *types.Block) (*types.BlockTrace, error) {
if env == nil {
log.Warn("running in light mode? trace env is nil and do not support `GetBlockTrace`")
return nil, nil
}
// Execute all the transaction contained within the block concurrently
var (
txs = block.Transactions()