mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-26 06:36:43 +00:00
feat: async ccc checker (#996)
This commit is contained in:
Ömer Faruk Irmak
GitHub
parent
972fe8afc4
commit
233a6adc15
15 changed files with 371 additions and 121 deletions
|
|
@ -168,6 +168,7 @@ var (
|
|||
utils.L1ConfirmationsFlag,
|
||||
utils.L1DeploymentBlockFlag,
|
||||
utils.CircuitCapacityCheckEnabledFlag,
|
||||
utils.CircuitCapacityCheckWorkersFlag,
|
||||
utils.RollupVerifyEnabledFlag,
|
||||
utils.ShadowforkPeersFlag,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -27,6 +27,7 @@ import (
|
|||
"math/big"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
godebug "runtime/debug"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
|
@ -847,6 +848,12 @@ var (
|
|||
Usage: "Enable circuit capacity check during block validation",
|
||||
}
|
||||
|
||||
CircuitCapacityCheckWorkersFlag = cli.UintFlag{
|
||||
Name: "ccc.numworkers",
|
||||
Usage: "Set the number of workers that will be used for background CCC tasks",
|
||||
Value: uint(runtime.GOMAXPROCS(0)),
|
||||
}
|
||||
|
||||
// Rollup verify service settings
|
||||
RollupVerifyEnabledFlag = cli.BoolFlag{
|
||||
Name: "rollup.verify",
|
||||
|
|
@ -1572,6 +1579,10 @@ func setWhitelist(ctx *cli.Context, cfg *ethconfig.Config) {
|
|||
func setCircuitCapacityCheck(ctx *cli.Context, cfg *ethconfig.Config) {
|
||||
if ctx.GlobalIsSet(CircuitCapacityCheckEnabledFlag.Name) {
|
||||
cfg.CheckCircuitCapacity = ctx.GlobalBool(CircuitCapacityCheckEnabledFlag.Name)
|
||||
cfg.CCCMaxWorkers = runtime.GOMAXPROCS(0)
|
||||
if ctx.GlobalIsSet(CircuitCapacityCheckWorkersFlag.Name) {
|
||||
cfg.CCCMaxWorkers = int(ctx.GlobalUint(CircuitCapacityCheckWorkersFlag.Name))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -17,29 +17,22 @@
|
|||
package core
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/scroll-tech/go-ethereum/consensus"
|
||||
"github.com/scroll-tech/go-ethereum/core/rawdb"
|
||||
"github.com/scroll-tech/go-ethereum/core/state"
|
||||
"github.com/scroll-tech/go-ethereum/core/types"
|
||||
"github.com/scroll-tech/go-ethereum/ethdb"
|
||||
"github.com/scroll-tech/go-ethereum/log"
|
||||
"github.com/scroll-tech/go-ethereum/metrics"
|
||||
"github.com/scroll-tech/go-ethereum/params"
|
||||
"github.com/scroll-tech/go-ethereum/rollup/ccc"
|
||||
"github.com/scroll-tech/go-ethereum/trie"
|
||||
)
|
||||
|
||||
var (
|
||||
validateL1MessagesTimer = metrics.NewRegisteredTimer("validator/l1msg", nil)
|
||||
validateRowConsumptionTimer = metrics.NewRegisteredTimer("validator/rowconsumption", nil)
|
||||
validateTraceTimer = metrics.NewRegisteredTimer("validator/trace", nil)
|
||||
validateLockTimer = metrics.NewRegisteredTimer("validator/lock", nil)
|
||||
validateCccTimer = metrics.NewRegisteredTimer("validator/ccc", nil)
|
||||
asyncValidatorTimer = metrics.NewRegisteredTimer("validator/async", nil)
|
||||
)
|
||||
|
||||
// BlockValidator is responsible for validating block headers, uncles and
|
||||
|
|
@ -50,12 +43,7 @@ type BlockValidator struct {
|
|||
config *params.ChainConfig // Chain configuration options
|
||||
bc *BlockChain // Canonical block chain
|
||||
engine consensus.Engine // Consensus engine used for validating
|
||||
|
||||
// circuit capacity checker related fields
|
||||
checkCircuitCapacity bool // whether enable circuit capacity check
|
||||
cMu sync.Mutex // mutex for circuit capacity checker
|
||||
tracer tracerWrapper // scroll tracer wrapper
|
||||
circuitCapacityChecker *ccc.Checker // circuit capacity checker instance
|
||||
asyncValidator func(*types.Block) error // Asynchronously run a validation task
|
||||
}
|
||||
|
||||
// NewBlockValidator returns a new block validator which is safe for re-use
|
||||
|
|
@ -68,15 +56,10 @@ func NewBlockValidator(config *params.ChainConfig, blockchain *BlockChain, engin
|
|||
return validator
|
||||
}
|
||||
|
||||
type tracerWrapper interface {
|
||||
CreateTraceEnvAndGetBlockTrace(*params.ChainConfig, ChainContext, consensus.Engine, ethdb.Database, *state.StateDB, *types.Block, *types.Block, bool) (*types.BlockTrace, error)
|
||||
}
|
||||
|
||||
func (v *BlockValidator) SetupTracerAndCircuitCapacityChecker(tracer tracerWrapper) {
|
||||
v.checkCircuitCapacity = true
|
||||
v.tracer = tracer
|
||||
v.circuitCapacityChecker = ccc.NewChecker(true)
|
||||
log.Info("new CircuitCapacityChecker in BlockValidator", "ID", v.circuitCapacityChecker.ID)
|
||||
// WithAsyncValidator sets up an async validator to be triggered on each new block
|
||||
func (v *BlockValidator) WithAsyncValidator(asyncValidator func(*types.Block) error) Validator {
|
||||
v.asyncValidator = asyncValidator
|
||||
return v
|
||||
}
|
||||
|
||||
// ValidateBody validates the given block's uncles and verifies the block
|
||||
|
|
@ -114,25 +97,13 @@ func (v *BlockValidator) ValidateBody(block *types.Block) error {
|
|||
if err := v.ValidateL1Messages(block); err != nil {
|
||||
return err
|
||||
}
|
||||
if v.checkCircuitCapacity {
|
||||
// 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
|
||||
}
|
||||
rowConsumption, err := v.validateCircuitRowConsumption(block)
|
||||
if err != nil {
|
||||
|
||||
if v.asyncValidator != nil {
|
||||
asyncStart := time.Now()
|
||||
if err := v.asyncValidator(block); err != nil {
|
||||
return err
|
||||
}
|
||||
log.Trace(
|
||||
"Validator write block row consumption",
|
||||
"id", v.circuitCapacityChecker.ID,
|
||||
"number", block.NumberU64(),
|
||||
"hash", block.Hash().String(),
|
||||
"rowConsumption", rowConsumption,
|
||||
)
|
||||
rawdb.WriteBlockRowConsumption(v.bc.db, block.Hash(), rowConsumption)
|
||||
asyncValidatorTimer.UpdateSince(asyncStart)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
@ -286,61 +257,3 @@ func CalcGasLimit(parentGasLimit, desiredLimit uint64) uint64 {
|
|||
}
|
||||
return limit
|
||||
}
|
||||
|
||||
func (v *BlockValidator) createTraceEnvAndGetBlockTrace(block *types.Block) (*types.BlockTrace, error) {
|
||||
parent := v.bc.GetBlock(block.ParentHash(), block.NumberU64()-1)
|
||||
if parent == nil {
|
||||
return nil, errors.New("validateCircuitRowConsumption: no parent block found")
|
||||
}
|
||||
|
||||
statedb, err := v.bc.StateAt(parent.Root())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return v.tracer.CreateTraceEnvAndGetBlockTrace(v.config, v.bc, v.engine, v.bc.db, statedb, parent, block, true)
|
||||
}
|
||||
|
||||
func (v *BlockValidator) validateCircuitRowConsumption(block *types.Block) (*types.RowConsumption, error) {
|
||||
defer func(t0 time.Time) {
|
||||
validateRowConsumptionTimer.Update(time.Since(t0))
|
||||
}(time.Now())
|
||||
|
||||
log.Trace(
|
||||
"Validator apply ccc for block",
|
||||
"id", v.circuitCapacityChecker.ID,
|
||||
"number", block.NumberU64(),
|
||||
"hash", block.Hash().String(),
|
||||
"len(txs)", block.Transactions().Len(),
|
||||
)
|
||||
|
||||
traceStartTime := time.Now()
|
||||
traces, err := v.createTraceEnvAndGetBlockTrace(block)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
validateTraceTimer.Update(time.Since(traceStartTime))
|
||||
|
||||
lockStartTime := time.Now()
|
||||
v.cMu.Lock()
|
||||
defer v.cMu.Unlock()
|
||||
validateLockTimer.Update(time.Since(lockStartTime))
|
||||
|
||||
cccStartTime := time.Now()
|
||||
v.circuitCapacityChecker.Reset()
|
||||
log.Trace("Validator reset ccc", "id", v.circuitCapacityChecker.ID)
|
||||
rc, err := v.circuitCapacityChecker.ApplyBlock(traces)
|
||||
validateCccTimer.Update(time.Since(cccStartTime))
|
||||
|
||||
log.Trace(
|
||||
"Validator apply ccc for block result",
|
||||
"id", v.circuitCapacityChecker.ID,
|
||||
"number", block.NumberU64(),
|
||||
"hash", block.Hash().String(),
|
||||
"len(txs)", block.Transactions().Len(),
|
||||
"rc", rc,
|
||||
"err", err,
|
||||
)
|
||||
|
||||
return rc, err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -33,9 +33,8 @@ type Validator interface {
|
|||
// 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)
|
||||
// WithAsyncValidator sets up an async validator to be triggered on each new block
|
||||
WithAsyncValidator(asyncValidator func(*types.Block) error) Validator
|
||||
}
|
||||
|
||||
// Prefetcher is an interface for pre-caching transaction signatures and state.
|
||||
|
|
|
|||
|
|
@ -2,12 +2,6 @@ package types
|
|||
|
||||
import "slices"
|
||||
|
||||
type RowUsage struct {
|
||||
IsOk bool `json:"is_ok"`
|
||||
RowNumber uint64 `json:"row_number"`
|
||||
RowUsageDetails []SubCircuitRowUsage `json:"row_usage_details"`
|
||||
}
|
||||
|
||||
//go:generate gencodec -type SubCircuitRowUsage -out gen_row_consumption_json.go
|
||||
type SubCircuitRowUsage struct {
|
||||
Name string `json:"name" gencodec:"required"`
|
||||
|
|
@ -15,7 +9,8 @@ type SubCircuitRowUsage struct {
|
|||
}
|
||||
|
||||
// RowConsumptionLimit is the max number of row we support per subcircuit
|
||||
const RowConsumptionLimit = 1_000_000
|
||||
// the actual limit is 1M but for safety we go with 950k
|
||||
const RowConsumptionLimit = 950_000
|
||||
|
||||
type RowConsumption []SubCircuitRowUsage
|
||||
|
||||
|
|
|
|||
|
|
@ -55,9 +55,9 @@ import (
|
|||
"github.com/scroll-tech/go-ethereum/p2p/enode"
|
||||
"github.com/scroll-tech/go-ethereum/params"
|
||||
"github.com/scroll-tech/go-ethereum/rlp"
|
||||
"github.com/scroll-tech/go-ethereum/rollup/ccc"
|
||||
"github.com/scroll-tech/go-ethereum/rollup/rollup_sync_service"
|
||||
"github.com/scroll-tech/go-ethereum/rollup/sync_service"
|
||||
"github.com/scroll-tech/go-ethereum/rollup/tracing"
|
||||
"github.com/scroll-tech/go-ethereum/rpc"
|
||||
)
|
||||
|
||||
|
|
@ -73,6 +73,7 @@ type Ethereum struct {
|
|||
txPool *core.TxPool
|
||||
syncService *sync_service.SyncService
|
||||
rollupSyncService *rollup_sync_service.RollupSyncService
|
||||
asyncChecker *ccc.AsyncChecker
|
||||
blockchain *core.BlockChain
|
||||
handler *handler
|
||||
ethDialCandidates enode.Iterator
|
||||
|
|
@ -199,8 +200,11 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client sync_service.EthCl
|
|||
return nil, err
|
||||
}
|
||||
if config.CheckCircuitCapacity {
|
||||
tracer := tracing.NewTracerWrapper()
|
||||
eth.blockchain.Validator().SetupTracerAndCircuitCapacityChecker(tracer)
|
||||
eth.asyncChecker = ccc.NewAsyncChecker(eth.blockchain, config.CCCMaxWorkers, true)
|
||||
eth.asyncChecker.WithOnFailingBlock(func(b *types.Block, err error) {
|
||||
log.Warn("block failed CCC check, it will be reorged by the sequencer", "hash", b.Hash(), "err", err)
|
||||
})
|
||||
eth.blockchain.Validator().WithAsyncValidator(eth.asyncChecker.Check)
|
||||
}
|
||||
|
||||
// Rewind the chain in case of an incompatible config upgrade.
|
||||
|
|
@ -594,6 +598,9 @@ func (s *Ethereum) Stop() error {
|
|||
s.rollupSyncService.Stop()
|
||||
}
|
||||
s.miner.Close()
|
||||
if s.config.CheckCircuitCapacity {
|
||||
s.asyncChecker.Wait()
|
||||
}
|
||||
s.blockchain.Stop()
|
||||
s.engine.Close()
|
||||
rawdb.PopUncleanShutdownMarker(s.chainDb)
|
||||
|
|
|
|||
|
|
@ -208,6 +208,7 @@ type Config struct {
|
|||
|
||||
// Check circuit capacity in block validator
|
||||
CheckCircuitCapacity bool
|
||||
CCCMaxWorkers int
|
||||
|
||||
// Enable verification of batch consistency between L1 and L2 in rollup
|
||||
EnableRollupVerify bool
|
||||
|
|
|
|||
3
go.mod
3
go.mod
|
|
@ -53,6 +53,7 @@ require (
|
|||
github.com/scroll-tech/da-codec v0.1.1-0.20240822151711-9e32313056ac
|
||||
github.com/scroll-tech/zktrie v0.8.4
|
||||
github.com/shirou/gopsutil v3.21.11+incompatible
|
||||
github.com/sourcegraph/conc v0.3.0
|
||||
github.com/status-im/keycard-go v0.0.0-20190316090335-8537d3370df4
|
||||
github.com/stretchr/testify v1.9.0
|
||||
github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7
|
||||
|
|
@ -99,6 +100,8 @@ require (
|
|||
github.com/tklauser/go-sysconf v0.3.12 // indirect
|
||||
github.com/tklauser/numcpus v0.6.1 // indirect
|
||||
github.com/yusufpapurcu/wmi v1.2.3 // indirect
|
||||
go.uber.org/atomic v1.7.0 // indirect
|
||||
go.uber.org/multierr v1.9.0 // indirect
|
||||
golang.org/x/net v0.16.0 // indirect
|
||||
golang.org/x/term v0.15.0 // indirect
|
||||
google.golang.org/protobuf v1.23.0 // indirect
|
||||
|
|
|
|||
6
go.sum
6
go.sum
|
|
@ -405,6 +405,8 @@ github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeV
|
|||
github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
|
||||
github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
|
||||
github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
|
||||
github.com/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9ySo=
|
||||
github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIKqDmF7Mt0=
|
||||
github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
|
||||
github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
|
||||
github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ=
|
||||
|
|
@ -450,7 +452,11 @@ go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
|
|||
go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
|
||||
go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
|
||||
go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
|
||||
go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
|
||||
go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
|
||||
go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
|
||||
go.uber.org/multierr v1.9.0 h1:7fIwc/ZtS0q++VgcfqFDxSBZVv/Xo49/SYnDFupUwlI=
|
||||
go.uber.org/multierr v1.9.0/go.mod h1:X2jQV1h+kxSjClGpnseKVIxpmcjrj7MNnI0bnlfKTVQ=
|
||||
go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
|
||||
golang.org/x/crypto v0.0.0-20170930174604-9419663f5a44/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
|
||||
golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ import (
|
|||
const (
|
||||
VersionMajor = 5 // Major version component of the current release
|
||||
VersionMinor = 6 // Minor version component of the current release
|
||||
VersionPatch = 3 // Patch version component of the current release
|
||||
VersionPatch = 4 // Patch version component of the current release
|
||||
VersionMeta = "mainnet" // Version metadata to append to the version string
|
||||
)
|
||||
|
||||
|
|
|
|||
224
rollup/ccc/async_checker.go
Normal file
224
rollup/ccc/async_checker.go
Normal file
|
|
@ -0,0 +1,224 @@
|
|||
package ccc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/sourcegraph/conc/stream"
|
||||
|
||||
"github.com/scroll-tech/go-ethereum/common"
|
||||
"github.com/scroll-tech/go-ethereum/core"
|
||||
"github.com/scroll-tech/go-ethereum/core/rawdb"
|
||||
"github.com/scroll-tech/go-ethereum/core/state"
|
||||
"github.com/scroll-tech/go-ethereum/core/types"
|
||||
"github.com/scroll-tech/go-ethereum/core/vm"
|
||||
"github.com/scroll-tech/go-ethereum/ethdb"
|
||||
"github.com/scroll-tech/go-ethereum/log"
|
||||
"github.com/scroll-tech/go-ethereum/metrics"
|
||||
"github.com/scroll-tech/go-ethereum/params"
|
||||
"github.com/scroll-tech/go-ethereum/rollup/tracing"
|
||||
)
|
||||
|
||||
var (
|
||||
failCounter = metrics.NewRegisteredCounter("ccc/async/fail", nil)
|
||||
checkTimer = metrics.NewRegisteredTimer("ccc/async/check", nil)
|
||||
activeWorkersGauge = metrics.NewRegisteredGauge("ccc/async/active_workers", nil)
|
||||
)
|
||||
|
||||
type Blockchain interface {
|
||||
Database() ethdb.Database
|
||||
GetBlock(hash common.Hash, number uint64) *types.Block
|
||||
StateAt(root common.Hash) (*state.StateDB, error)
|
||||
Config() *params.ChainConfig
|
||||
GetVMConfig() *vm.Config
|
||||
CurrentHeader() *types.Header
|
||||
core.ChainContext
|
||||
}
|
||||
|
||||
// AsyncChecker allows a caller to spawn CCC verification tasks
|
||||
type AsyncChecker struct {
|
||||
bc Blockchain
|
||||
onFailingBlock func(*types.Block, error)
|
||||
|
||||
workers *stream.Stream
|
||||
freeCheckers chan *Checker
|
||||
|
||||
// local state to keep track of the chain progressing and terminate tasks early if needed
|
||||
currentHead *types.Header
|
||||
forkCtx context.Context
|
||||
forkCtxCancelFunc context.CancelFunc
|
||||
}
|
||||
|
||||
type ErrorWithTxnIdx struct {
|
||||
txIdx uint
|
||||
err error
|
||||
shouldSkip bool
|
||||
}
|
||||
|
||||
func (e *ErrorWithTxnIdx) Error() string {
|
||||
return fmt.Sprintf("txn at index %d failed with %s", e.txIdx, e.err)
|
||||
}
|
||||
|
||||
func (e *ErrorWithTxnIdx) Unwrap() error {
|
||||
return e.err
|
||||
}
|
||||
|
||||
func NewAsyncChecker(bc Blockchain, numWorkers int, lightMode bool) *AsyncChecker {
|
||||
forkCtx, forkCtxCancelFunc := context.WithCancel(context.Background())
|
||||
return &AsyncChecker{
|
||||
bc: bc,
|
||||
freeCheckers: func(count int) chan *Checker {
|
||||
checkers := make(chan *Checker, count)
|
||||
for i := 0; i < count; i++ {
|
||||
checkers <- NewChecker(lightMode)
|
||||
}
|
||||
return checkers
|
||||
}(numWorkers),
|
||||
workers: stream.New().WithMaxGoroutines(numWorkers),
|
||||
currentHead: bc.CurrentHeader(),
|
||||
forkCtx: forkCtx,
|
||||
forkCtxCancelFunc: forkCtxCancelFunc,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *AsyncChecker) WithOnFailingBlock(onFailingBlock func(*types.Block, error)) *AsyncChecker {
|
||||
c.onFailingBlock = onFailingBlock
|
||||
return c
|
||||
}
|
||||
|
||||
func (c *AsyncChecker) Wait() {
|
||||
c.workers.Wait()
|
||||
}
|
||||
|
||||
// Check spawns an async CCC verification task.
|
||||
func (c *AsyncChecker) Check(block *types.Block) error {
|
||||
if block.NumberU64() > c.currentHead.Number.Uint64()+1 {
|
||||
log.Error("non continuous chain observed in AsyncChecker", "prev", c.currentHead, "got", block.Header())
|
||||
} else if block.ParentHash() != c.currentHead.Hash() {
|
||||
// seems like there is a fork happening, a block from the canonical chain must have failed CCC check
|
||||
// assume the incoming block is the new tip in the fork
|
||||
c.forkCtx, c.forkCtxCancelFunc = context.WithCancel(context.Background())
|
||||
}
|
||||
|
||||
c.currentHead = block.Header()
|
||||
checker := <-c.freeCheckers
|
||||
// all blocks in the same fork share the same context to allow terminating them all at once if needed
|
||||
ctx, ctxCancelFunc := c.forkCtx, c.forkCtxCancelFunc
|
||||
c.workers.Go(func() stream.Callback {
|
||||
return c.checkerTask(block, checker, ctx, ctxCancelFunc)
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func isForkStillActive(forkCtx context.Context) bool {
|
||||
select {
|
||||
case <-forkCtx.Done():
|
||||
// an ancestor block of this block failed CCC check, this fork is not active anymore
|
||||
return false
|
||||
default:
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (c *AsyncChecker) checkerTask(block *types.Block, ccc *Checker, forkCtx context.Context, forkCtxCancelFunc context.CancelFunc) stream.Callback {
|
||||
activeWorkersGauge.Inc(1)
|
||||
checkStart := time.Now()
|
||||
defer func() {
|
||||
checkTimer.UpdateSince(checkStart)
|
||||
c.freeCheckers <- ccc
|
||||
activeWorkersGauge.Dec(1)
|
||||
}()
|
||||
|
||||
noopCb := func() {}
|
||||
parent := c.bc.GetBlock(block.ParentHash(), block.NumberU64()-1)
|
||||
if parent == nil {
|
||||
return noopCb // not part of a chain
|
||||
}
|
||||
|
||||
var err error
|
||||
failingCallback := func() {
|
||||
failCounter.Inc(1)
|
||||
if isForkStillActive(forkCtx) {
|
||||
// we failed the CCC check, cancel the context to signal all tasks preceding this one to terminate early
|
||||
forkCtxCancelFunc()
|
||||
if c.onFailingBlock != nil {
|
||||
c.onFailingBlock(block, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
statedb, err := c.bc.StateAt(parent.Root())
|
||||
if err != nil {
|
||||
return failingCallback
|
||||
}
|
||||
|
||||
header := block.Header()
|
||||
header.GasUsed = 0
|
||||
gasPool := new(core.GasPool).AddGas(header.GasLimit)
|
||||
ccc.Reset()
|
||||
|
||||
var accRc *types.RowConsumption
|
||||
for txIdx, tx := range block.Transactions() {
|
||||
if !isForkStillActive(forkCtx) {
|
||||
return noopCb
|
||||
}
|
||||
|
||||
var curRc *types.RowConsumption
|
||||
curRc, err = c.checkTxAndApply(parent, header, statedb, gasPool, tx, ccc)
|
||||
if err != nil {
|
||||
err = &ErrorWithTxnIdx{
|
||||
txIdx: uint(txIdx),
|
||||
err: err,
|
||||
// if the txn is the first in block or the additional resource utilization caused
|
||||
// by this txn alone is enough to overflow the circuit, skip
|
||||
shouldSkip: txIdx == 0 || curRc.Difference(*accRc).IsOverflown(),
|
||||
}
|
||||
return failingCallback
|
||||
}
|
||||
accRc = curRc
|
||||
}
|
||||
|
||||
return func() {
|
||||
if isForkStillActive(forkCtx) {
|
||||
// all good, write the row consumption
|
||||
log.Debug("CCC passed", "blockhash", block.Hash(), "height", block.NumberU64())
|
||||
rawdb.WriteBlockRowConsumption(c.bc.Database(), block.Hash(), accRc)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *AsyncChecker) checkTxAndApply(parent *types.Block, header *types.Header, state *state.StateDB, gasPool *core.GasPool, tx *types.Transaction, ccc *Checker) (*types.RowConsumption, error) {
|
||||
// don't commit the state during tracing for circuit capacity checker, otherwise we cannot revert.
|
||||
// and even if we don't commit the state, the `refund` value will still be correct, as explained in `CommitTransaction`
|
||||
commitStateAfterApply := false
|
||||
snap := state.Snapshot()
|
||||
|
||||
// 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
|
||||
trace, err := tracing.NewTracerWrapper().CreateTraceEnvAndGetBlockTrace(c.bc.Config(), c.bc, c.bc.Engine(), c.bc.Database(),
|
||||
state, parent, types.NewBlockWithHeader(header).WithBody([]*types.Transaction{tx}, nil), commitStateAfterApply)
|
||||
// `w.current.traceEnv.State` & `w.current.state` share a same pointer to the state, so only need to revert `w.current.state`
|
||||
// revert to snapshot for calling `core.ApplyMessage` again, (both `traceEnv.GetBlockTrace` & `core.ApplyTransaction` will call `core.ApplyMessage`)
|
||||
state.RevertToSnapshot(snap)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
rc, err := ccc.ApplyTransaction(trace)
|
||||
if err != nil {
|
||||
return rc, err
|
||||
}
|
||||
|
||||
_, err = core.ApplyTransaction(c.bc.Config(), c.bc, nil /* coinbase will default to chainConfig.Scroll.FeeVaultAddress */, gasPool,
|
||||
state, header, tx, &header.GasUsed, *c.bc.GetVMConfig())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return rc, nil
|
||||
}
|
||||
81
rollup/ccc/async_checker_test.go
Normal file
81
rollup/ccc/async_checker_test.go
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
package ccc
|
||||
|
||||
import (
|
||||
"math/big"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/scroll-tech/go-ethereum/common"
|
||||
"github.com/scroll-tech/go-ethereum/consensus/ethash"
|
||||
"github.com/scroll-tech/go-ethereum/core"
|
||||
"github.com/scroll-tech/go-ethereum/core/rawdb"
|
||||
"github.com/scroll-tech/go-ethereum/core/types"
|
||||
"github.com/scroll-tech/go-ethereum/core/vm"
|
||||
"github.com/scroll-tech/go-ethereum/crypto"
|
||||
"github.com/scroll-tech/go-ethereum/params"
|
||||
)
|
||||
|
||||
func TestAsyncChecker(t *testing.T) {
|
||||
// testKey is a private key to use for funding a tester account.
|
||||
testKey, _ := crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
|
||||
// testAddr is the Ethereum address of the tester account.
|
||||
testAddr := crypto.PubkeyToAddress(testKey.PublicKey)
|
||||
|
||||
// Create a database pre-initialize with a genesis block
|
||||
db := rawdb.NewMemoryDatabase()
|
||||
(&core.Genesis{
|
||||
Config: params.TestChainConfig,
|
||||
Alloc: core.GenesisAlloc{testAddr: {Balance: new(big.Int).Mul(big.NewInt(1000), big.NewInt(params.Ether))}},
|
||||
}).MustCommit(db)
|
||||
|
||||
chain, _ := core.NewBlockChain(db, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, nil)
|
||||
asyncChecker := NewAsyncChecker(chain, 1, false)
|
||||
chain.Validator().WithAsyncValidator(asyncChecker.Check)
|
||||
|
||||
bs, _ := core.GenerateChain(params.TestChainConfig, chain.Genesis(), ethash.NewFaker(), db, 100, func(i int, block *core.BlockGen) {
|
||||
for i := 0; i < 10; i++ {
|
||||
signer := types.MakeSigner(params.TestChainConfig, block.Number())
|
||||
tx, err := types.SignTx(types.NewTransaction(block.TxNonce(testAddr), testAddr, big.NewInt(1000), params.TxGas, block.BaseFee(), nil), signer, testKey)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
block.AddTx(tx)
|
||||
}
|
||||
})
|
||||
|
||||
noReorgBlocks := bs[:len(bs)/2]
|
||||
if _, err := chain.InsertChain(noReorgBlocks); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
for _, block := range noReorgBlocks {
|
||||
require.NotNil(t, rawdb.ReadBlockRowConsumption(db, block.Hash()))
|
||||
}
|
||||
|
||||
reorgBlocks := bs[len(bs)/2:]
|
||||
skippedTxn := reorgBlocks[3].Transactions()[3]
|
||||
checker := <-asyncChecker.freeCheckers
|
||||
checker.Skip(skippedTxn.Hash(), ErrBlockRowConsumptionOverflow)
|
||||
// trigger an error on some later height, we shouldn't get a notification for this
|
||||
checker.ScheduleError(50, ErrBlockRowConsumptionOverflow)
|
||||
|
||||
asyncChecker.freeCheckers <- checker
|
||||
|
||||
var failingBlockHash common.Hash
|
||||
var errWithIdx *ErrorWithTxnIdx
|
||||
asyncChecker.WithOnFailingBlock(func(b *types.Block, err error) {
|
||||
failingBlockHash = b.Hash()
|
||||
require.ErrorAs(t, err, &errWithIdx)
|
||||
})
|
||||
|
||||
if _, err := chain.InsertChain(reorgBlocks); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
time.Sleep(3 * time.Second)
|
||||
require.Equal(t, reorgBlocks[3].Hash(), failingBlockHash)
|
||||
require.Equal(t, uint(3), errWithIdx.txIdx)
|
||||
require.True(t, errWithIdx.shouldSkip)
|
||||
}
|
||||
|
|
@ -113,7 +113,7 @@ func (ccc *Checker) applyTransactionRustTrace(rustTrace unsafe.Pointer) (*types.
|
|||
return nil, ErrUnknown
|
||||
}
|
||||
if !result.AccRowUsage.IsOk {
|
||||
return nil, ErrBlockRowConsumptionOverflow
|
||||
return (*types.RowConsumption)(&result.AccRowUsage.RowUsageDetails), ErrBlockRowConsumptionOverflow
|
||||
}
|
||||
return (*types.RowConsumption)(&result.AccRowUsage.RowUsageDetails), nil
|
||||
}
|
||||
|
|
@ -153,7 +153,7 @@ func (ccc *Checker) ApplyBlock(traces *types.BlockTrace) (*types.RowConsumption,
|
|||
return nil, ErrUnknown
|
||||
}
|
||||
if !result.AccRowUsage.IsOk {
|
||||
return nil, ErrBlockRowConsumptionOverflow
|
||||
return (*types.RowConsumption)(&result.AccRowUsage.RowUsageDetails), ErrBlockRowConsumptionOverflow
|
||||
}
|
||||
return (*types.RowConsumption)(&result.AccRowUsage.RowUsageDetails), nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -44,7 +44,10 @@ func (ccc *Checker) ApplyTransaction(traces *types.BlockTrace) (*types.RowConsum
|
|||
}
|
||||
if ccc.skipError != nil {
|
||||
if traces.Transactions[0].TxHash == ccc.skipHash {
|
||||
return nil, ccc.skipError
|
||||
return &types.RowConsumption{types.SubCircuitRowUsage{
|
||||
Name: "mock",
|
||||
RowNumber: 1_000_001,
|
||||
}}, ccc.skipError
|
||||
}
|
||||
}
|
||||
return &types.RowConsumption{types.SubCircuitRowUsage{
|
||||
|
|
|
|||
|
|
@ -15,8 +15,14 @@ type WrappedCommonResult struct {
|
|||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
type RowUsage struct {
|
||||
IsOk bool `json:"is_ok"`
|
||||
RowNumber uint64 `json:"row_number"`
|
||||
RowUsageDetails []types.SubCircuitRowUsage `json:"row_usage_details"`
|
||||
}
|
||||
|
||||
type WrappedRowUsage struct {
|
||||
AccRowUsage *types.RowUsage `json:"acc_row_usage,omitempty"`
|
||||
AccRowUsage *RowUsage `json:"acc_row_usage,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue