Commit 9d2e72cb authored by mergify[bot]'s avatar mergify[bot] Committed by GitHub

Merge branch 'develop' into jg/ci

parents 71611d2c 6e524df2
......@@ -64,6 +64,8 @@ type Metrics struct {
ChannelClosedReason prometheus.Gauge
ChannelNumFrames prometheus.Gauge
ChannelComprRatio prometheus.Histogram
ChannelInputBytesTotal prometheus.Counter
ChannelOutputBytesTotal prometheus.Counter
BatcherTxEvs opmetrics.EventVec
}
......@@ -144,6 +146,16 @@ func NewMetrics(procName string) *Metrics {
Help: "Compression ratios of closed channel.",
Buckets: append([]float64{0.1, 0.2}, prometheus.LinearBuckets(0.3, 0.05, 14)...),
}),
ChannelInputBytesTotal: factory.NewCounter(prometheus.CounterOpts{
Namespace: ns,
Name: "input_bytes_total",
Help: "Total number of bytes to a channel.",
}),
ChannelOutputBytesTotal: factory.NewCounter(prometheus.CounterOpts{
Namespace: ns,
Name: "output_bytes_total",
Help: "Total number of compressed output bytes from a channel.",
}),
BatcherTxEvs: opmetrics.NewEventVec(factory, ns, "", "batcher_tx", "BatcherTx", []string{"stage"}),
}
......@@ -219,6 +231,8 @@ func (m *Metrics) RecordChannelClosed(id derive.ChannelID, numPendingBlocks int,
m.ChannelNumFrames.Set(float64(numFrames))
m.ChannelInputBytes.WithLabelValues(StageClosed).Set(float64(inputBytes))
m.ChannelOutputBytes.Set(float64(outputComprBytes))
m.ChannelInputBytesTotal.Add(float64(inputBytes))
m.ChannelOutputBytesTotal.Add(float64(outputComprBytes))
var comprRatio float64
if inputBytes > 0 {
......
......@@ -191,7 +191,7 @@ func TestL2EngineAPIFail(gt *testing.T) {
}
func TestEngineAPITests(t *testing.T) {
test.RunEngineAPITests(t, func() engineapi.EngineBackend {
test.RunEngineAPITests(t, func(t *testing.T) engineapi.EngineBackend {
jwtPath := e2eutils.WriteDefaultJWT(t)
dp := e2eutils.MakeDeployParams(t, defaultRollupTestParams)
sd := e2eutils.Setup(t, dp, defaultAlloc)
......
......@@ -24,6 +24,7 @@ type IterativeBatchCall[K any, V any] struct {
makeRequest func(K) (V, rpc.BatchElem)
getBatch BatchCallContextFn
getSingle CallContextFn
requestsValues []V
scheduled chan rpc.BatchElem
......@@ -35,6 +36,7 @@ func NewIterativeBatchCall[K any, V any](
requestsKeys []K,
makeRequest func(K) (V, rpc.BatchElem),
getBatch BatchCallContextFn,
getSingle CallContextFn,
batchSize int) *IterativeBatchCall[K, V] {
if len(requestsKeys) < batchSize {
......@@ -47,6 +49,7 @@ func NewIterativeBatchCall[K any, V any](
out := &IterativeBatchCall[K, V]{
completed: 0,
getBatch: getBatch,
getSingle: getSingle,
requestsKeys: requestsKeys,
batchSize: batchSize,
makeRequest: makeRequest,
......@@ -119,12 +122,24 @@ func (ibc *IterativeBatchCall[K, V]) Fetch(ctx context.Context) error {
break
}
if len(batch) == 0 {
return nil
}
if ibc.batchSize == 1 {
first := batch[0]
if err := ibc.getSingle(ctx, &first.Result, first.Method, first.Args...); err != nil {
ibc.scheduled <- first
return err
}
} else {
if err := ibc.getBatch(ctx, batch); err != nil {
for _, r := range batch {
ibc.scheduled <- r
}
return fmt.Errorf("failed batch-retrieval: %w", err)
}
}
var result error
for _, elem := range batch {
if elem.Error != nil {
......
......@@ -35,6 +35,7 @@ type batchTestCase struct {
batchSize int
batchCalls []batchCall
singleCalls []elemCall
mock.Mock
}
......@@ -53,7 +54,14 @@ func (tc *batchTestCase) GetBatch(ctx context.Context, b []rpc.BatchElem) error
if ctx.Err() != nil {
return ctx.Err()
}
return tc.Mock.MethodCalled("get", b).Get(0).([]error)[0]
return tc.Mock.MethodCalled("getBatch", b).Get(0).([]error)[0]
}
func (tc *batchTestCase) GetSingle(ctx context.Context, result any, method string, args ...any) error {
if ctx.Err() != nil {
return ctx.Err()
}
return tc.Mock.MethodCalled("getSingle", (*(result.(*interface{}))).(*string), method, args[0]).Get(0).([]error)[0]
}
var mockErr = errors.New("mockErr")
......@@ -64,7 +72,7 @@ func (tc *batchTestCase) Run(t *testing.T) {
keys[i] = i
}
makeMock := func(bci int, bc batchCall) func(args mock.Arguments) {
makeBatchMock := func(bci int, bc batchCall) func(args mock.Arguments) {
return func(args mock.Arguments) {
batch := args[0].([]rpc.BatchElem)
for i, elem := range batch {
......@@ -94,10 +102,30 @@ func (tc *batchTestCase) Run(t *testing.T) {
})
}
if len(bc.elems) > 0 {
tc.On("get", batch).Once().Run(makeMock(bci, bc)).Return([]error{bc.rpcErr}) // wrap to preserve nil as type of error
tc.On("getBatch", batch).Once().Run(makeBatchMock(bci, bc)).Return([]error{bc.rpcErr}) // wrap to preserve nil as type of error
}
}
makeSingleMock := func(eci int, ec elemCall) func(args mock.Arguments) {
return func(args mock.Arguments) {
result := args[0].(*string)
id := args[2].(int)
require.Equal(t, ec.id, id, "element should match expected element")
if ec.err {
*result = ""
} else {
*result = fmt.Sprintf("mock result id %d", id)
}
}
}
// mock the results of unbatched calls
for eci, ec := range tc.singleCalls {
var ret error
if ec.err {
ret = mockErr
}
tc.On("getSingle", new(string), "testing_foobar", ec.id).Once().Run(makeSingleMock(eci, ec)).Return([]error{ret})
}
iter := NewIterativeBatchCall[int, *string](keys, makeTestRequest, tc.GetBatch, tc.batchSize)
iter := NewIterativeBatchCall[int, *string](keys, makeTestRequest, tc.GetBatch, tc.GetSingle, tc.batchSize)
for i, bc := range tc.batchCalls {
ctx := context.Background()
if bc.makeCtx != nil {
......@@ -116,6 +144,20 @@ func (tc *batchTestCase) Run(t *testing.T) {
}
}
}
for i, ec := range tc.singleCalls {
ctx := context.Background()
err := iter.Fetch(ctx)
if err == io.EOF {
require.Equal(t, i, len(tc.singleCalls)-1, "EOF only on last call")
} else {
require.False(t, iter.Complete())
if ec.err {
require.Error(t, err)
} else {
require.NoError(t, err)
}
}
}
require.True(t, iter.Complete(), "batch iter should be complete after the expected calls")
out, err := iter.Result()
require.NoError(t, err)
......@@ -154,6 +196,37 @@ func TestFetchBatched(t *testing.T) {
},
},
},
{
name: "single element",
items: 1,
batchSize: 4,
singleCalls: []elemCall{
{id: 0, err: false},
},
},
{
name: "unbatched",
items: 4,
batchSize: 1,
singleCalls: []elemCall{
{id: 0, err: false},
{id: 1, err: false},
{id: 2, err: false},
{id: 3, err: false},
},
},
{
name: "unbatched with retry",
items: 4,
batchSize: 1,
singleCalls: []elemCall{
{id: 0, err: false},
{id: 1, err: true},
{id: 2, err: false},
{id: 3, err: false},
{id: 1, err: false},
},
},
{
name: "split",
items: 5,
......@@ -240,7 +313,7 @@ func TestFetchBatched(t *testing.T) {
},
{
name: "context timeout",
items: 1,
items: 2,
batchSize: 3,
batchCalls: []batchCall{
{
......@@ -255,6 +328,7 @@ func TestFetchBatched(t *testing.T) {
{
elems: []elemCall{
{id: 0, err: false},
{id: 1, err: false},
},
err: "",
},
......
......@@ -373,6 +373,7 @@ func (job *receiptsFetchingJob) runFetcher(ctx context.Context) error {
job.txHashes,
makeReceiptRequest,
job.client.BatchCallContext,
job.client.CallContext,
job.maxBatchSize,
)
}
......
......@@ -22,6 +22,7 @@ type OracleBackedL2Chain struct {
oracle Oracle
chainCfg *params.ChainConfig
engine consensus.Engine
oracleHead *types.Header
head *types.Header
safe *types.Header
finalized *types.Header
......@@ -47,6 +48,7 @@ func NewOracleBackedL2Chain(logger log.Logger, oracle Oracle, chainCfg *params.C
head: head.Header(),
safe: head.Header(),
finalized: head.Header(),
oracleHead: head.Header(),
blocks: make(map[common.Hash]*types.Block),
db: NewOracleBackedDB(oracle),
}, nil
......@@ -82,11 +84,7 @@ func (o *OracleBackedL2Chain) CurrentFinalBlock() *types.Header {
}
func (o *OracleBackedL2Chain) GetHeaderByHash(hash common.Hash) *types.Header {
block := o.GetBlockByHash(hash)
if block == nil {
return nil
}
return block.Header()
return o.GetBlockByHash(hash).Header()
}
func (o *OracleBackedL2Chain) GetBlockByHash(hash common.Hash) *types.Block {
......@@ -96,15 +94,18 @@ func (o *OracleBackedL2Chain) GetBlockByHash(hash common.Hash) *types.Block {
return block
}
// Retrieve from the oracle
block = o.oracle.BlockByHash(hash)
if block == nil {
return nil
}
return block
return o.oracle.BlockByHash(hash)
}
func (o *OracleBackedL2Chain) GetBlock(hash common.Hash, number uint64) *types.Block {
block := o.GetBlockByHash(hash)
var block *types.Block
if o.oracleHead.Number.Uint64() < number {
// For blocks above the chain head, only consider newly built blocks
// Avoids requesting an unknown block from the oracle which would panic.
block = o.blocks[hash]
} else {
block = o.GetBlockByHash(hash)
}
if block == nil {
return nil
}
......@@ -116,9 +117,6 @@ func (o *OracleBackedL2Chain) GetBlock(hash common.Hash, number uint64) *types.B
func (o *OracleBackedL2Chain) GetHeader(hash common.Hash, u uint64) *types.Header {
block := o.GetBlock(hash, u)
if block == nil {
return nil
}
return block.Header()
}
......
......@@ -42,17 +42,6 @@ func TestGetBlocks(t *testing.T) {
}
}
func TestUnknownBlock(t *testing.T) {
_, chain := setupOracleBackedChain(t, 1)
hash := common.HexToHash("0x556677881122")
blockNumber := uint64(1)
require.Nil(t, chain.GetBlockByHash(hash))
require.Nil(t, chain.GetHeaderByHash(hash))
require.Nil(t, chain.GetBlock(hash, blockNumber))
require.Nil(t, chain.GetHeader(hash, blockNumber))
require.False(t, chain.HasBlockAndState(hash, blockNumber))
}
func TestCanonicalHashNotFoundPastChainHead(t *testing.T) {
blocks, chain := setupOracleBackedChainWithLowerHead(t, 5, 3)
......@@ -69,7 +58,7 @@ func TestCanonicalHashNotFoundPastChainHead(t *testing.T) {
func TestAppendToChain(t *testing.T) {
blocks, chain := setupOracleBackedChainWithLowerHead(t, 4, 3)
newBlock := blocks[4]
require.Nil(t, chain.GetBlockByHash(newBlock.Hash()), "block unknown before being added")
require.Nil(t, chain.GetBlock(newBlock.Hash(), newBlock.NumberU64()), "block unknown before being added")
require.NoError(t, chain.InsertBlockWithoutSetHead(newBlock))
require.Equal(t, blocks[3].Header(), chain.CurrentHeader(), "should not update chain head yet")
......@@ -113,9 +102,7 @@ func TestUpdateStateDatabaseWhenImportingBlock(t *testing.T) {
require.NotEqual(t, blocks[1].Root(), newBlock.Root(), "block should have modified world state")
require.Panics(t, func() {
_, _ = chain.StateAt(newBlock.Root())
}, "state from non-imported block should not be available")
require.False(t, chain.HasBlockAndState(newBlock.Root(), newBlock.NumberU64()), "state from non-imported block should not be available")
err = chain.InsertBlockWithoutSetHead(newBlock)
require.NoError(t, err)
......@@ -181,7 +168,7 @@ func setupOracle(t *testing.T, blockCount int, headBlockNumber int) (*params.Cha
genesisBlock := l2Genesis.MustCommit(db)
blocks, _ := core.GenerateChain(chainCfg, genesisBlock, consensus, db, blockCount, func(i int, gen *core.BlockGen) {})
blocks = append([]*types.Block{genesisBlock}, blocks...)
oracle := newStubBlockOracle(blocks[:headBlockNumber+1], db)
oracle := newStubBlockOracle(t, blocks[:headBlockNumber+1], db)
return chainCfg, blocks, oracle
}
......@@ -213,23 +200,27 @@ type stubBlockOracle struct {
kvStateOracle
}
func newStubBlockOracle(chain []*types.Block, db ethdb.Database) *stubBlockOracle {
func newStubBlockOracle(t *testing.T, chain []*types.Block, db ethdb.Database) *stubBlockOracle {
blocks := make(map[common.Hash]*types.Block, len(chain))
for _, block := range chain {
blocks[block.Hash()] = block
}
return &stubBlockOracle{
blocks: blocks,
kvStateOracle: kvStateOracle{source: db},
kvStateOracle: kvStateOracle{t: t, source: db},
}
}
func (o stubBlockOracle) BlockByHash(blockHash common.Hash) *types.Block {
return o.blocks[blockHash]
block, ok := o.blocks[blockHash]
if !ok {
o.t.Fatalf("requested unknown block %s", blockHash)
}
return block
}
func TestEngineAPITests(t *testing.T) {
test.RunEngineAPITests(t, func() engineapi.EngineBackend {
test.RunEngineAPITests(t, func(t *testing.T) engineapi.EngineBackend {
_, chain := setupOracleBackedChain(t, 0)
return chain
})
......
......@@ -301,7 +301,7 @@ func (ea *L2EngineAPI) NewPayloadV1(ctx context.Context, payload *eth.ExecutionP
}
// If we already have the block locally, ignore the entire execution and just
// return a fake success.
if block := ea.backend.GetBlockByHash(payload.BlockHash); block != nil {
if block := ea.backend.GetBlock(payload.BlockHash, uint64(payload.BlockNumber)); block != nil {
ea.log.Warn("Ignoring already known beacon payload", "number", payload.BlockNumber, "hash", payload.BlockHash, "age", common.PrettyAge(time.Unix(int64(block.Time()), 0)))
hash := block.Hash()
return &eth.PayloadStatusV1{Status: eth.ExecutionValid, LatestValidHash: &hash}, nil
......
......@@ -18,7 +18,7 @@ import (
var gasLimit = eth.Uint64Quantity(30_000_000)
var feeRecipient = common.Address{}
func RunEngineAPITests(t *testing.T, createBackend func() engineapi.EngineBackend) {
func RunEngineAPITests(t *testing.T, createBackend func(t *testing.T) engineapi.EngineBackend) {
t.Run("CreateBlock", func(t *testing.T) {
api := newTestHelper(t, createBackend)
......@@ -292,10 +292,10 @@ type testHelper struct {
assert *require.Assertions
}
func newTestHelper(t *testing.T, createBackend func() engineapi.EngineBackend) *testHelper {
func newTestHelper(t *testing.T, createBackend func(t *testing.T) engineapi.EngineBackend) *testHelper {
logger := testlog.Logger(t, log.LvlDebug)
ctx := context.Background()
backend := createBackend()
backend := createBackend(t)
api := engineapi.NewL2EngineAPI(logger, backend)
test := &testHelper{
t: t,
......
......@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"github.com/ethereum-optimism/optimism/op-node/eth"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/core/rawdb"
......@@ -24,19 +25,25 @@ type CallContext interface {
type FetchingL2Oracle struct {
ctx context.Context
logger log.Logger
head eth.BlockInfo
blockSource BlockSource
callContext CallContext
}
func NewFetchingL2Oracle(ctx context.Context, logger log.Logger, l2Url string) (*FetchingL2Oracle, error) {
func NewFetchingL2Oracle(ctx context.Context, logger log.Logger, l2Url string, l2Head common.Hash) (*FetchingL2Oracle, error) {
rpcClient, err := rpc.Dial(l2Url)
if err != nil {
return nil, err
}
ethClient := ethclient.NewClient(rpcClient)
head, err := ethClient.HeaderByHash(ctx, l2Head)
if err != nil {
return nil, fmt.Errorf("retrieve l2 head %v: %w", l2Head, err)
}
return &FetchingL2Oracle{
ctx: ctx,
logger: logger,
head: eth.HeaderBlockInfo(head),
blockSource: ethClient,
callContext: rpcClient,
}, nil
......@@ -78,5 +85,8 @@ func (o *FetchingL2Oracle) BlockByHash(blockHash common.Hash) *types.Block {
if err != nil {
panic(fmt.Errorf("fetch block %s: %w", blockHash.Hex(), err))
}
if block.NumberU64() > o.head.NumberU64() {
panic(fmt.Errorf("fetched block %v number %d above head block number %d", blockHash, block.NumberU64(), o.head.NumberU64()))
}
return block
}
......@@ -5,12 +5,14 @@ import (
"encoding/json"
"errors"
"fmt"
"math/big"
"math/rand"
"reflect"
"testing"
"github.com/ethereum-optimism/optimism/op-node/testutils"
cll2 "github.com/ethereum-optimism/optimism/op-program/client/l2"
"github.com/ethereum/go-ethereum/trie"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
......@@ -23,35 +25,7 @@ import (
// Require the fetching oracle to implement StateOracle
var _ cll2.StateOracle = (*FetchingL2Oracle)(nil)
type callContextRequest struct {
ctx context.Context
method string
args []interface{}
}
type stubCallContext struct {
nextResult any
nextErr error
requests []callContextRequest
}
func (c *stubCallContext) CallContext(ctx context.Context, result any, method string, args ...interface{}) error {
if result != nil && reflect.TypeOf(result).Kind() != reflect.Ptr {
return fmt.Errorf("call result parameter must be pointer or nil interface: %v", result)
}
c.requests = append(c.requests, callContextRequest{ctx: ctx, method: method, args: args})
if c.nextErr != nil {
return c.nextErr
}
res, err := json.Marshal(c.nextResult)
if err != nil {
return fmt.Errorf("json marshal: %w", err)
}
err = json.Unmarshal(res, result)
if err != nil {
return fmt.Errorf("json unmarshal: %w", err)
}
return nil
}
const headBlockNumber = 1000
func TestNodeByHash(t *testing.T) {
rng := rand.New(rand.NewSource(1234))
......@@ -152,31 +126,12 @@ func TestCodeByHash(t *testing.T) {
})
}
type blockRequest struct {
ctx context.Context
blockHash common.Hash
}
type stubBlockSource struct {
requests []blockRequest
nextErr error
nextResult *types.Block
}
func (s *stubBlockSource) BlockByHash(ctx context.Context, blockHash common.Hash) (*types.Block, error) {
s.requests = append(s.requests, blockRequest{
ctx: ctx,
blockHash: blockHash,
})
return s.nextResult, s.nextErr
}
func TestBlockByHash(t *testing.T) {
rng := rand.New(rand.NewSource(1234))
hash := testutils.RandomHash(rng)
t.Run("Success", func(t *testing.T) {
block, _ := testutils.RandomBlock(rng, 1)
block := blockWithNumber(rng, headBlockNumber-1)
stub := &stubBlockSource{nextResult: block}
fetcher := newFetcher(stub, nil)
......@@ -194,7 +149,7 @@ func TestBlockByHash(t *testing.T) {
})
t.Run("RequestArgs", func(t *testing.T) {
stub := &stubBlockSource{}
stub := &stubBlockSource{nextResult: blockWithNumber(rng, 1)}
fetcher := newFetcher(stub, nil)
fetcher.BlockByHash(hash)
......@@ -203,11 +158,86 @@ func TestBlockByHash(t *testing.T) {
req := stub.requests[0]
require.Equal(t, hash, req.blockHash)
})
t.Run("PanicWhenBlockAboveHeadRequested", func(t *testing.T) {
// Block that the source can provide but is above the head block number
block := blockWithNumber(rng, headBlockNumber+1)
stub := &stubBlockSource{nextResult: block}
fetcher := newFetcher(stub, nil)
require.Panics(t, func() {
fetcher.BlockByHash(block.Hash())
})
})
}
func blockWithNumber(rng *rand.Rand, num int64) *types.Block {
header := testutils.RandomHeader(rng)
header.Number = big.NewInt(num)
return types.NewBlock(header, nil, nil, nil, trie.NewStackTrie(nil))
}
type blockRequest struct {
ctx context.Context
blockHash common.Hash
}
type stubBlockSource struct {
requests []blockRequest
nextErr error
nextResult *types.Block
}
func (s *stubBlockSource) BlockByHash(ctx context.Context, blockHash common.Hash) (*types.Block, error) {
s.requests = append(s.requests, blockRequest{
ctx: ctx,
blockHash: blockHash,
})
return s.nextResult, s.nextErr
}
type callContextRequest struct {
ctx context.Context
method string
args []interface{}
}
type stubCallContext struct {
nextResult any
nextErr error
requests []callContextRequest
}
func (c *stubCallContext) CallContext(ctx context.Context, result any, method string, args ...interface{}) error {
if result != nil && reflect.TypeOf(result).Kind() != reflect.Ptr {
return fmt.Errorf("call result parameter must be pointer or nil interface: %v", result)
}
c.requests = append(c.requests, callContextRequest{ctx: ctx, method: method, args: args})
if c.nextErr != nil {
return c.nextErr
}
res, err := json.Marshal(c.nextResult)
if err != nil {
return fmt.Errorf("json marshal: %w", err)
}
err = json.Unmarshal(res, result)
if err != nil {
return fmt.Errorf("json unmarshal: %w", err)
}
return nil
}
func newFetcher(blockSource BlockSource, callContext CallContext) *FetchingL2Oracle {
rng := rand.New(rand.NewSource(int64(1)))
head := testutils.MakeBlockInfo(func(i *testutils.MockBlockInfo) {
i.InfoNum = headBlockNumber
})(rng)
return &FetchingL2Oracle{
ctx: context.Background(),
logger: log.New(),
head: head,
blockSource: blockSource,
callContext: callContext,
}
......
......@@ -19,7 +19,7 @@ func NewFetchingEngine(ctx context.Context, logger log.Logger, cfg *config.Confi
if err != nil {
return nil, err
}
oracle, err := NewFetchingL2Oracle(ctx, logger, cfg.L2URL)
oracle, err := NewFetchingL2Oracle(ctx, logger, cfg.L2URL, cfg.L2Head)
if err != nil {
return nil, fmt.Errorf("connect l2 oracle: %w", err)
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment