diff --git a/cmd/geth/main.go b/cmd/geth/main.go index 180d7bc0ea..3aca4af089 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -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) diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index c362c95d11..ec6b4f8198 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -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) diff --git a/core/block_validator.go b/core/block_validator.go index ec8e039067..45a2957900 100644 --- a/core/block_validator.go +++ b/core/block_validator.go @@ -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 diff --git a/core/rawdb/accessors_row_consumption.go b/core/rawdb/accessors_row_consumption.go new file mode 100644 index 0000000000..8e584a4366 --- /dev/null +++ b/core/rawdb/accessors_row_consumption.go @@ -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 +} diff --git a/core/rawdb/schema.go b/core/rawdb/schema.go index 5dc9f7710e..8a3905dd76 100644 --- a/core/rawdb/schema.go +++ b/core/rawdb/schema.go @@ -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()...) diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index 32c6c0e8fe..83f3731797 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/core/txpool/blobpool/blobpool.go @@ -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 +} diff --git a/core/txpool/legacypool/legacypool.go b/core/txpool/legacypool/legacypool.go index 0e33923274..3dcefd683d 100644 --- a/core/txpool/legacypool/legacypool.go +++ b/core/txpool/legacypool/legacypool.go @@ -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. // diff --git a/core/txpool/subpool.go b/core/txpool/subpool.go index de05b38d43..4201cbe773 100644 --- a/core/txpool/subpool.go +++ b/core/txpool/subpool.go @@ -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 } diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go index 0d4e05da4c..bff38df397 100644 --- a/core/txpool/txpool.go +++ b/core/txpool/txpool.go @@ -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 +} diff --git a/core/types.go b/core/types.go index 36eb0d1ded..1533219cc3 100644 --- a/core/types.go +++ b/core/types.go @@ -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. diff --git a/core/types/block.go b/core/types/block.go index 55f904b3e2..9d4528f759 100644 --- a/core/types/block.go +++ b/core/types/block.go @@ -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 { diff --git a/core/types/transaction.go b/core/types/transaction.go index 621ae67c01..845b6378ea 100644 --- a/core/types/transaction.go +++ b/core/types/transaction.go @@ -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() -} diff --git a/eth/backend.go b/eth/backend.go index 85139c0237..a1bb49858a 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -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 != "" { diff --git a/eth/ethconfig/config.go b/eth/ethconfig/config.go index 3ec771f218..cbf5f6f8cc 100644 --- a/eth/ethconfig/config.go +++ b/eth/ethconfig/config.go @@ -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 diff --git a/eth/ethconfig/gen_config.go b/eth/ethconfig/gen_config.go index aed74f2db9..26ce2e3f6c 100644 --- a/eth/ethconfig/gen_config.go +++ b/eth/ethconfig/gen_config.go @@ -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 } diff --git a/miner/miner.go b/miner/miner.go index b7273948f5..e210c0e803 100644 --- a/miner/miner.go +++ b/miner/miner.go @@ -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. diff --git a/miner/miner_test.go b/miner/miner_test.go index 36d5166c6d..b3206a5d99 100644 --- a/miner/miner_test.go +++ b/miner/miner_test.go @@ -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 + 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, + 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 diff --git a/miner/ordering.go b/miner/ordering.go index 4c3055f0d3..87a4865ba3 100644 --- a/miner/ordering.go +++ b/miner/ordering.go @@ -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() +} diff --git a/miner/worker.go b/miner/worker.go index 96eccd2d3b..9e57560e23 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -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 @@ -216,9 +273,10 @@ type worker struct { snapshotState *state.StateDB // atomic status counters - 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. + 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() + } +} diff --git a/miner/worker_test.go b/miner/worker_test.go index 9c4694c0e2..59ea89f415 100644 --- a/miner/worker_test.go +++ b/miner/worker_test.go @@ -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) diff --git a/rollup/tracing/tracing.go b/rollup/tracing/tracing.go index 4e9760fa9d..45031d3c0a 100644 --- a/rollup/tracing/tracing.go +++ b/rollup/tracing/tracing.go @@ -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()