Commit 09ffddf0 authored by mergify[bot]'s avatar mergify[bot] Committed by GitHub

Merge branch 'develop' into zhwrd/circleci-oidc

parents f2621a26 cc738b3d
---
'@eth-optimism/proxyd': minor
---
Allow disabling backend rate limiter
---
'@eth-optimism/proxyd': minor
---
Support pattern matching in exempt origins/user agents
---
'@eth-optimism/proxyd': minor
---
adds server.log_level config
......@@ -10,6 +10,7 @@ const (
DevOptimismMintableERC20Factory = "0x6900000000000000000000000000000000000004"
DevAddressManager = "0x6900000000000000000000000000000000000005"
DevProxyAdmin = "0x6900000000000000000000000000000000000006"
DevWETH9 = "0x6900000000000000000000000000000000000007"
)
var (
......@@ -20,6 +21,7 @@ var (
DevOptimismMintableERC20FactoryAddr = common.HexToAddress(DevOptimismMintableERC20Factory)
DevAddressManagerAddr = common.HexToAddress(DevAddressManager)
DevProxyAdminAddr = common.HexToAddress(DevProxyAdmin)
DevWETH9Addr = common.HexToAddress(DevWETH9)
DevPredeploys = make(map[string]*common.Address)
)
......@@ -32,4 +34,5 @@ func init() {
DevPredeploys["OptimismMintableERC20Factory"] = &DevOptimismMintableERC20FactoryAddr
DevPredeploys["AddressManager"] = &DevAddressManagerAddr
DevPredeploys["Admin"] = &DevProxyAdminAddr
DevPredeploys["WETH9"] = &DevWETH9Addr
}
......@@ -165,6 +165,16 @@ func BuildL1DeveloperGenesis(config *DeployConfig) (*core.Genesis, error) {
for name, proxyAddr := range predeploys.DevPredeploys {
memDB.SetState(*proxyAddr, ImplementationSlot, depsByName[name].Address.Hash())
// Special case for WETH since it was not designed to be behind a proxy
if name == "WETH9" {
name, _ := state.EncodeStringValue("Wrapped Ether", 0)
symbol, _ := state.EncodeStringValue("WETH", 0)
decimals, _ := state.EncodeUintValue(18, 0)
memDB.SetState(*proxyAddr, common.Hash{}, name)
memDB.SetState(*proxyAddr, common.Hash{31: 0x01}, symbol)
memDB.SetState(*proxyAddr, common.Hash{31: 0x02}, decimals)
}
}
stateDB, err := backend.Blockchain().State()
......@@ -183,6 +193,7 @@ func BuildL1DeveloperGenesis(config *DeployConfig) (*core.Genesis, error) {
memDB.CreateAccount(depAddr)
memDB.SetCode(depAddr, dep.Bytecode)
for iter.Next() {
_, data, _, err := rlp.Split(iter.Value)
if err != nil {
......@@ -250,6 +261,9 @@ func deployL1Contracts(config *DeployConfig, backend *backends.SimulatedBackend)
common.Address{19: 0x01},
},
},
{
Name: "WETH9",
},
}...)
return deployer.Deploy(backend, constructors, l1Deployer)
}
......@@ -308,6 +322,11 @@ func l1Deployer(backend *backends.SimulatedBackend, opts *bind.TransactOpts, dep
backend,
common.Address{},
)
case "WETH9":
_, tx, _, err = bindings.DeployWETH9(
opts,
backend,
)
default:
if strings.HasSuffix(deployment.Name, "Proxy") {
_, tx, _, err = bindings.DeployProxy(opts, backend, deployer.TestAddress)
......
......@@ -92,6 +92,18 @@ func TestBuildL1DeveloperGenesis(t *testing.T) {
require.NoError(t, err)
require.Equal(t, predeploys.DevL1StandardBridgeAddr, bridgeAddr)
weth9, err := bindings.NewWETH9(predeploys.DevWETH9Addr, sim)
require.NoError(t, err)
decimals, err := weth9.Decimals(callOpts)
require.NoError(t, err)
require.Equal(t, uint8(18), decimals)
symbol, err := weth9.Symbol(callOpts)
require.NoError(t, err)
require.Equal(t, "WETH", symbol)
name, err := weth9.Name(callOpts)
require.NoError(t, err)
require.Equal(t, "Wrapped Ether", name)
// test that we can do deposits, etc.
priv, err := crypto.HexToECDSA("ac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80")
require.NoError(t, err)
......
......@@ -149,6 +149,13 @@ func (s *BasicUser[B]) ActRandomTxToAddr(t Testing) {
s.txToAddr = to
}
func (s *BasicUser[B]) ActSetTxCalldata(calldata []byte) Action {
return func(t Testing) {
require.NotNil(t, calldata)
s.txCallData = calldata
}
}
func (s *BasicUser[B]) ActSetTxToAddr(to *common.Address) Action {
return func(t Testing) {
s.txToAddr = to
......@@ -165,6 +172,12 @@ func (s *BasicUser[B]) ActRandomTxValue(t Testing) {
s.txOpts.Value = big.NewInt(s.rng.Int63())
}
func (s *BasicUser[B]) ActSetTxValue(value *big.Int) Action {
return func(t Testing) {
s.txOpts.Value = value
}
}
func (s *BasicUser[B]) ActRandomTxData(t Testing) {
dataLen := s.rng.Intn(128_000)
out := make([]byte, dataLen)
......@@ -190,6 +203,13 @@ func (s *BasicUser[B]) TxValue() *big.Int {
return big.NewInt(0)
}
func (s *BasicUser[B]) LastTxReceipt(t Testing) *types.Receipt {
require.NotEqual(t, s.lastTxHash, common.Hash{}, "must send tx before getting last receipt")
receipt, err := s.env.EthCl.TransactionReceipt(t.Ctx(), s.lastTxHash)
require.NoError(t, err)
return receipt
}
// ActMakeTx makes a tx with the predetermined contents (see randomization and other actions)
// and sends it to the tx pool
func (s *BasicUser[B]) ActMakeTx(t Testing) {
......@@ -210,10 +230,13 @@ func (s *BasicUser[B]) ActMakeTx(t Testing) {
ChainID: s.env.Signer.ChainID(),
Nonce: s.PendingNonce(t),
Gas: gas,
Data: s.txCallData,
})
err = s.env.EthCl.SendTransaction(t.Ctx(), tx)
require.NoError(t, err, "must send tx")
s.lastTxHash = tx.Hash()
// reset the calldata
s.txCallData = []byte{}
}
func (s *BasicUser[B]) ActCheckReceiptStatusOfLastTx(success bool) func(t Testing) {
......
......@@ -10,6 +10,8 @@ import (
"strconv"
"time"
libp2pmetrics "github.com/libp2p/go-libp2p-core/metrics"
pb "github.com/libp2p/go-libp2p-pubsub/pb"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/collectors"
"github.com/prometheus/client_golang/prometheus/promauto"
......@@ -30,6 +32,32 @@ const (
BatchMethod = "<batch>"
)
type Metricer interface {
RecordInfo(version string)
RecordUp()
RecordRPCServerRequest(method string) func()
RecordRPCClientRequest(method string) func(err error)
RecordRPCClientResponse(method string, err error)
SetDerivationIdle(status bool)
RecordPipelineReset()
RecordSequencingError()
RecordPublishingError()
RecordDerivationError()
RecordReceivedUnsafePayload(payload *eth.ExecutionPayload)
recordRef(layer string, name string, num uint64, timestamp uint64, h common.Hash)
RecordL1Ref(name string, ref eth.L1BlockRef)
RecordL2Ref(name string, ref eth.L2BlockRef)
RecordUnsafePayloadsBuffer(length uint64, memSize uint64, next eth.BlockID)
CountSequencedTxs(count int)
RecordL1ReorgDepth(d uint64)
RecordGossipEvent(evType int32)
IncPeerCount()
DecPeerCount()
IncStreamCount()
DecStreamCount()
RecordBandwidth(ctx context.Context, bwc *libp2pmetrics.BandwidthCounter)
}
type Metrics struct {
Info *prometheus.GaugeVec
Up prometheus.Gauge
......@@ -67,6 +95,12 @@ type Metrics struct {
TransactionsSequencedTotal prometheus.Counter
// P2P Metrics
PeerCount prometheus.Gauge
StreamCount prometheus.Gauge
GossipEventsTotal *prometheus.CounterVec
BandwidthTotal *prometheus.GaugeVec
registry *prometheus.Registry
}
......@@ -217,6 +251,35 @@ func NewMetrics(procName string) *Metrics {
Help: "Count of total transactions sequenced",
}),
PeerCount: promauto.With(registry).NewGauge(prometheus.GaugeOpts{
Namespace: ns,
Subsystem: "p2p",
Name: "peer_count",
Help: "Count of currently connected p2p peers",
}),
StreamCount: promauto.With(registry).NewGauge(prometheus.GaugeOpts{
Namespace: ns,
Subsystem: "p2p",
Name: "stream_count",
Help: "Count of currently connected p2p streams",
}),
GossipEventsTotal: promauto.With(registry).NewCounterVec(prometheus.CounterOpts{
Namespace: ns,
Subsystem: "p2p",
Name: "gossip_events_total",
Help: "Count of gossip events by type",
}, []string{
"type",
}),
BandwidthTotal: promauto.With(registry).NewGaugeVec(prometheus.GaugeOpts{
Namespace: ns,
Subsystem: "p2p",
Name: "bandwidth_bytes_total",
Help: "P2P bandwidth by direction",
}, []string{
"direction",
}),
registry: registry,
}
}
......@@ -348,6 +411,42 @@ func (m *Metrics) RecordL1ReorgDepth(d uint64) {
m.L1ReorgDepth.Observe(float64(d))
}
func (m *Metrics) RecordGossipEvent(evType int32) {
m.GossipEventsTotal.WithLabelValues(pb.TraceEvent_Type_name[evType]).Inc()
}
func (m *Metrics) IncPeerCount() {
m.PeerCount.Inc()
}
func (m *Metrics) DecPeerCount() {
m.PeerCount.Dec()
}
func (m *Metrics) IncStreamCount() {
m.StreamCount.Inc()
}
func (m *Metrics) DecStreamCount() {
m.StreamCount.Dec()
}
func (m *Metrics) RecordBandwidth(ctx context.Context, bwc *libp2pmetrics.BandwidthCounter) {
tick := time.NewTicker(10 * time.Second)
defer tick.Stop()
for {
select {
case <-tick.C:
bwTotals := bwc.GetBandwidthTotals()
m.BandwidthTotal.WithLabelValues("in").Set(float64(bwTotals.TotalIn))
m.BandwidthTotal.WithLabelValues("out").Set(float64(bwTotals.TotalOut))
case <-ctx.Done():
return
}
}
}
// Serve starts the metrics server on the given hostname and port.
// The server will be closed when the passed-in context is cancelled.
func (m *Metrics) Serve(ctx context.Context, hostname string, port int) error {
......@@ -364,3 +463,78 @@ func (m *Metrics) Serve(ctx context.Context, hostname string, port int) error {
}()
return server.ListenAndServe()
}
type noopMetricer struct{}
var NoopMetrics = new(noopMetricer)
func (n *noopMetricer) RecordInfo(version string) {
}
func (n *noopMetricer) RecordUp() {
}
func (n *noopMetricer) RecordRPCServerRequest(method string) func() {
return func() {}
}
func (n *noopMetricer) RecordRPCClientRequest(method string) func(err error) {
return func(err error) {}
}
func (n *noopMetricer) RecordRPCClientResponse(method string, err error) {
}
func (n *noopMetricer) SetDerivationIdle(status bool) {
}
func (n *noopMetricer) RecordPipelineReset() {
}
func (n *noopMetricer) RecordSequencingError() {
}
func (n *noopMetricer) RecordPublishingError() {
}
func (n *noopMetricer) RecordDerivationError() {
}
func (n *noopMetricer) RecordReceivedUnsafePayload(payload *eth.ExecutionPayload) {
}
func (n *noopMetricer) recordRef(layer string, name string, num uint64, timestamp uint64, h common.Hash) {
}
func (n *noopMetricer) RecordL1Ref(name string, ref eth.L1BlockRef) {
}
func (n *noopMetricer) RecordL2Ref(name string, ref eth.L2BlockRef) {
}
func (n *noopMetricer) RecordUnsafePayloadsBuffer(length uint64, memSize uint64, next eth.BlockID) {
}
func (n *noopMetricer) CountSequencedTxs(count int) {
}
func (n *noopMetricer) RecordL1ReorgDepth(d uint64) {
}
func (n *noopMetricer) RecordGossipEvent(evType int32) {
}
func (n *noopMetricer) IncPeerCount() {
}
func (n *noopMetricer) DecPeerCount() {
}
func (n *noopMetricer) IncStreamCount() {
}
func (n *noopMetricer) DecStreamCount() {
}
func (n *noopMetricer) RecordBandwidth(ctx context.Context, bwc *libp2pmetrics.BandwidthCounter) {
}
......@@ -196,7 +196,7 @@ func (n *OpNode) initMetricsServer(ctx context.Context, cfg *Config) error {
func (n *OpNode) initP2P(ctx context.Context, cfg *Config) error {
if cfg.P2P != nil {
p2pNode, err := p2p.NewNodeP2P(n.resourcesCtx, &cfg.Rollup, n.log, cfg.P2P, n)
p2pNode, err := p2p.NewNodeP2P(n.resourcesCtx, &cfg.Rollup, n.log, cfg.P2P, n, n.metrics)
if err != nil || p2pNode == nil {
return err
}
......
......@@ -41,7 +41,7 @@ import (
type SetupP2P interface {
Check() error
// Host creates a libp2p host service. Returns nil, nil if p2p is disabled.
Host(log log.Logger) (host.Host, error)
Host(log log.Logger, reporter metrics.Reporter) (host.Host, error)
// Discovery creates a disc-v5 service. Returns nil, nil, nil if discovery is disabled.
Discovery(log log.Logger, rollupCfg *rollup.Config, tcpPort uint16) (*enode.LocalNode, *discover.UDPv5, error)
TargetPeers() uint
......@@ -91,8 +91,6 @@ type Config struct {
ConnGater func(conf *Config) (connmgr.ConnectionGater, error)
ConnMngr func(conf *Config) (connmgr.ConnManager, error)
// nil to disable bandwidth metrics
BandwidthMetrics metrics.Reporter
}
type ConnectionGater interface {
......
......@@ -46,6 +46,10 @@ var MessageDomainValidSnappy = [4]byte{1, 0, 0, 0}
const MaxGossipSize = 1 << 20
type GossipMetricer interface {
RecordGossipEvent(evType int32)
}
func blocksTopicV1(cfg *rollup.Config) string {
return fmt.Sprintf("/optimism/%s/0/blocks", cfg.L2ChainID.String())
}
......@@ -115,7 +119,7 @@ func BuildGlobalGossipParams(cfg *rollup.Config) pubsub.GossipSubParams {
return params
}
func NewGossipSub(p2pCtx context.Context, h host.Host, cfg *rollup.Config) (*pubsub.PubSub, error) {
func NewGossipSub(p2pCtx context.Context, h host.Host, cfg *rollup.Config, m GossipMetricer) (*pubsub.PubSub, error) {
denyList, err := pubsub.NewTimeCachedBlacklist(30 * time.Second)
if err != nil {
return nil, err
......@@ -132,6 +136,7 @@ func NewGossipSub(p2pCtx context.Context, h host.Host, cfg *rollup.Config) (*pub
pubsub.WithPeerExchange(false),
pubsub.WithBlacklist(denyList),
pubsub.WithGossipSubParams(BuildGlobalGossipParams(cfg)),
pubsub.WithEventTracer(&gossipTracer{m: m}),
)
// TODO: pubsub.WithPeerScoreInspect(inspect, InspectInterval) to update peerstore scores with gossip scores
}
......@@ -441,3 +446,13 @@ func LogTopicEvents(ctx context.Context, log log.Logger, evHandler *pubsub.Topic
}
}
}
type gossipTracer struct {
m GossipMetricer
}
func (g *gossipTracer) Trace(evt *pb.TraceEvent) {
if g.m != nil {
g.m.RecordGossipEvent(int32(*evt.Type))
}
}
......@@ -8,6 +8,7 @@ import (
"github.com/libp2p/go-libp2p-core/connmgr"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/metrics"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/libp2p/go-libp2p-peerstore/pstoreds"
lconf "github.com/libp2p/go-libp2p/config"
......@@ -41,7 +42,7 @@ func (e *extraHost) ConnectionManager() connmgr.ConnManager {
var _ ExtraHostFeatures = (*extraHost)(nil)
func (conf *Config) Host(log log.Logger) (host.Host, error) {
func (conf *Config) Host(log log.Logger, reporter metrics.Reporter) (host.Host, error) {
if conf.DisableP2P {
return nil, nil
}
......@@ -115,7 +116,7 @@ func (conf *Config) Host(log log.Logger) (host.Host, error) {
ResourceManager: nil, // TODO use resource manager interface to manage resources per peer better.
NATManager: nat,
Peerstore: ps,
Reporter: conf.BandwidthMetrics, // may be nil if disabled
Reporter: reporter, // may be nil if disabled
MultiaddrResolver: madns.DefaultResolver,
// Ping is a small built-in libp2p protocol that helps us check/debug latency between peers.
DisablePing: false,
......
......@@ -21,7 +21,6 @@ import (
"github.com/stretchr/testify/require"
"github.com/ethereum-optimism/optimism/op-node/eth"
"github.com/ethereum-optimism/optimism/op-node/metrics"
"github.com/ethereum-optimism/optimism/op-node/rollup"
"github.com/ethereum-optimism/optimism/op-node/testlog"
"github.com/ethereum/go-ethereum/log"
......@@ -65,10 +64,10 @@ func TestingConfig(t *testing.T) *Config {
func TestP2PSimple(t *testing.T) {
confA := TestingConfig(t)
confB := TestingConfig(t)
hostA, err := confA.Host(testlog.Logger(t, log.LvlError).New("host", "A"))
hostA, err := confA.Host(testlog.Logger(t, log.LvlError).New("host", "A"), nil)
require.NoError(t, err, "failed to launch host A")
defer hostA.Close()
hostB, err := confB.Host(testlog.Logger(t, log.LvlError).New("host", "B"))
hostB, err := confB.Host(testlog.Logger(t, log.LvlError).New("host", "B"), nil)
require.NoError(t, err, "failed to launch host B")
defer hostB.Close()
err = hostA.Connect(context.Background(), peer.AddrInfo{ID: hostB.ID(), Addrs: hostB.Addrs()})
......@@ -132,7 +131,7 @@ func TestP2PFull(t *testing.T) {
// TODO: maybe swap the order of sec/mux preferences, to test that negotiation works
logA := testlog.Logger(t, log.LvlError).New("host", "A")
nodeA, err := NewNodeP2P(context.Background(), &rollup.Config{}, logA, &confA, &mockGossipIn{})
nodeA, err := NewNodeP2P(context.Background(), &rollup.Config{}, logA, &confA, &mockGossipIn{}, nil)
require.NoError(t, err)
defer nodeA.Close()
......@@ -143,7 +142,7 @@ func TestP2PFull(t *testing.T) {
conns <- conn
}})
backend := NewP2PAPIBackend(nodeA, logA, metrics.NewMetrics(""))
backend := NewP2PAPIBackend(nodeA, logA, nil)
srv := rpc.NewServer()
require.NoError(t, srv.RegisterName("opp2p", backend))
client := rpc.DialInProc(srv)
......@@ -155,7 +154,7 @@ func TestP2PFull(t *testing.T) {
logB := testlog.Logger(t, log.LvlError).New("host", "B")
nodeB, err := NewNodeP2P(context.Background(), &rollup.Config{}, logB, &confB, &mockGossipIn{})
nodeB, err := NewNodeP2P(context.Background(), &rollup.Config{}, logB, &confB, &mockGossipIn{}, nil)
require.NoError(t, err)
defer nodeB.Close()
hostB := nodeB.Host()
......@@ -289,7 +288,7 @@ func TestDiscovery(t *testing.T) {
resourcesCtx, resourcesCancel := context.WithCancel(context.Background())
defer resourcesCancel()
nodeA, err := NewNodeP2P(context.Background(), rollupCfg, logA, &confA, &mockGossipIn{})
nodeA, err := NewNodeP2P(context.Background(), rollupCfg, logA, &confA, &mockGossipIn{}, nil)
require.NoError(t, err)
defer nodeA.Close()
hostA := nodeA.Host()
......@@ -304,7 +303,7 @@ func TestDiscovery(t *testing.T) {
confB.DiscoveryDB = discDBC
// Start B
nodeB, err := NewNodeP2P(context.Background(), rollupCfg, logB, &confB, &mockGossipIn{})
nodeB, err := NewNodeP2P(context.Background(), rollupCfg, logB, &confB, &mockGossipIn{}, nil)
require.NoError(t, err)
defer nodeB.Close()
hostB := nodeB.Host()
......@@ -319,7 +318,7 @@ func TestDiscovery(t *testing.T) {
}})
// Start C
nodeC, err := NewNodeP2P(context.Background(), rollupCfg, logC, &confC, &mockGossipIn{})
nodeC, err := NewNodeP2P(context.Background(), rollupCfg, logC, &confC, &mockGossipIn{}, nil)
require.NoError(t, err)
defer nodeC.Close()
hostC := nodeC.Host()
......
......@@ -6,9 +6,11 @@ import (
"fmt"
"strconv"
"github.com/ethereum-optimism/optimism/op-node/metrics"
"github.com/hashicorp/go-multierror"
"github.com/libp2p/go-libp2p-core/connmgr"
"github.com/libp2p/go-libp2p-core/host"
p2pmetrics "github.com/libp2p/go-libp2p-core/metrics"
pubsub "github.com/libp2p/go-libp2p-pubsub"
"github.com/libp2p/go-libp2p/p2p/protocol/identify"
ma "github.com/multiformats/go-multiaddr"
......@@ -30,12 +32,12 @@ type NodeP2P struct {
gsOut GossipOut // p2p gossip application interface for publishing
}
func NewNodeP2P(resourcesCtx context.Context, rollupCfg *rollup.Config, log log.Logger, setup SetupP2P, gossipIn GossipIn) (*NodeP2P, error) {
func NewNodeP2P(resourcesCtx context.Context, rollupCfg *rollup.Config, log log.Logger, setup SetupP2P, gossipIn GossipIn, metrics metrics.Metricer) (*NodeP2P, error) {
if setup == nil {
return nil, errors.New("p2p node cannot be created without setup")
}
var n NodeP2P
if err := n.init(resourcesCtx, rollupCfg, log, setup, gossipIn); err != nil {
if err := n.init(resourcesCtx, rollupCfg, log, setup, gossipIn, metrics); err != nil {
closeErr := n.Close()
if closeErr != nil {
log.Error("failed to close p2p after starting with err", "closeErr", closeErr, "err", err)
......@@ -48,10 +50,12 @@ func NewNodeP2P(resourcesCtx context.Context, rollupCfg *rollup.Config, log log.
return &n, nil
}
func (n *NodeP2P) init(resourcesCtx context.Context, rollupCfg *rollup.Config, log log.Logger, setup SetupP2P, gossipIn GossipIn) error {
func (n *NodeP2P) init(resourcesCtx context.Context, rollupCfg *rollup.Config, log log.Logger, setup SetupP2P, gossipIn GossipIn, metrics metrics.Metricer) error {
bwc := p2pmetrics.NewBandwidthCounter()
var err error
// nil if disabled.
n.host, err = setup.Host(log)
n.host, err = setup.Host(log, bwc)
if err != nil {
if n.dv5Udp != nil {
n.dv5Udp.Close()
......@@ -66,10 +70,10 @@ func (n *NodeP2P) init(resourcesCtx context.Context, rollupCfg *rollup.Config, l
n.connMgr = extra.ConnectionManager()
}
// notify of any new connections/streams/etc.
n.host.Network().Notify(NewNetworkNotifier(log))
n.host.Network().Notify(NewNetworkNotifier(log, metrics))
// unregister identify-push handler. Only identifying on dial is fine, and more robust against spam
n.host.RemoveStreamHandler(identify.IDDelta)
n.gs, err = NewGossipSub(resourcesCtx, n.host, rollupCfg)
n.gs, err = NewGossipSub(resourcesCtx, n.host, rollupCfg, metrics)
if err != nil {
return fmt.Errorf("failed to start gossipsub router: %w", err)
}
......@@ -90,6 +94,10 @@ func (n *NodeP2P) init(resourcesCtx context.Context, rollupCfg *rollup.Config, l
if err != nil {
return fmt.Errorf("failed to start discv5: %w", err)
}
if metrics != nil {
go metrics.RecordBandwidth(resourcesCtx, bwc)
}
}
return nil
}
......
package p2p
import (
"github.com/ethereum-optimism/optimism/op-node/metrics"
"github.com/libp2p/go-libp2p-core/network"
ma "github.com/multiformats/go-multiaddr"
"github.com/ethereum/go-ethereum/log"
)
// TODO: add metrics here as well
type NotificationsMetricer interface {
IncPeerCount()
DecPeerCount()
IncStreamCount()
DecStreamCount()
}
type notifications struct {
log log.Logger
m NotificationsMetricer
}
func (notif *notifications) Listen(n network.Network, a ma.Multiaddr) {
......@@ -20,20 +27,27 @@ func (notif *notifications) ListenClose(n network.Network, a ma.Multiaddr) {
notif.log.Info("stopped listening network address", "addr", a)
}
func (notif *notifications) Connected(n network.Network, v network.Conn) {
notif.m.IncPeerCount()
notif.log.Info("connected to peer", "peer", v.RemotePeer(), "addr", v.RemoteMultiaddr())
}
func (notif *notifications) Disconnected(n network.Network, v network.Conn) {
notif.m.DecPeerCount()
notif.log.Info("disconnected from peer", "peer", v.RemotePeer(), "addr", v.RemoteMultiaddr())
}
func (notif *notifications) OpenedStream(n network.Network, v network.Stream) {
notif.m.IncStreamCount()
c := v.Conn()
notif.log.Trace("opened stream", "protocol", v.Protocol(), "peer", c.RemotePeer(), "addr", c.RemoteMultiaddr())
}
func (notif *notifications) ClosedStream(n network.Network, v network.Stream) {
notif.m.DecStreamCount()
c := v.Conn()
notif.log.Trace("opened stream", "protocol", v.Protocol(), "peer", c.RemotePeer(), "addr", c.RemoteMultiaddr())
}
func NewNetworkNotifier(log log.Logger) network.Notifiee {
return &notifications{log: log}
func NewNetworkNotifier(log log.Logger, m metrics.Metricer) network.Notifiee {
if m == nil {
m = metrics.NoopMetrics
}
return &notifications{log: log, m: m}
}
......@@ -5,6 +5,7 @@ import (
"fmt"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/metrics"
"github.com/ethereum-optimism/optimism/op-node/rollup"
"github.com/ethereum/go-ethereum/log"
......@@ -38,7 +39,7 @@ func (p *Prepared) Check() error {
}
// Host creates a libp2p host service. Returns nil, nil if p2p is disabled.
func (p *Prepared) Host(log log.Logger) (host.Host, error) {
func (p *Prepared) Host(log log.Logger, reporter metrics.Reporter) (host.Host, error) {
return p.HostP2P, nil
}
......
......@@ -54,12 +54,16 @@ type Node interface {
type APIBackend struct {
node Node
log log.Logger
m *metrics.Metrics
m metrics.Metricer
}
var _ API = (*APIBackend)(nil)
func NewP2PAPIBackend(node Node, log log.Logger, m *metrics.Metrics) *APIBackend {
func NewP2PAPIBackend(node Node, log log.Logger, m metrics.Metricer) *APIBackend {
if m == nil {
m = metrics.NoopMetrics
}
return &APIBackend{
node: node,
log: log,
......
......@@ -59,7 +59,11 @@ func (bq *BatchQueue) Origin() eth.L1BlockRef {
}
func (bq *BatchQueue) NextBatch(ctx context.Context, safeL2Head eth.L2BlockRef) (*BatchData, error) {
originBehind := bq.origin.Number < safeL2Head.L1Origin.Number
// Note: We use the origin that we will have to determine if it's behind. This is important
// because it's the future origin that gets saved into the l1Blocks array.
// We always update the origin of this stage if it is not the same so after the update code
// runs, this is consistent.
originBehind := bq.prev.Origin().Number < safeL2Head.L1Origin.Number
// Advance origin if needed
// Note: The entire pipeline has the same origin
......
......@@ -405,7 +405,7 @@ func (eq *EngineQueue) forceNextSafeAttributes(ctx context.Context) error {
// ResetStep Walks the L2 chain backwards until it finds an L2 block whose L1 origin is canonical.
// The unsafe head is set to the head of the L2 chain, unless the existing safe head is not canonical.
func (eq *EngineQueue) Reset(ctx context.Context, _ eth.L1BlockRef) error {
result, err := sync.FindL2Heads(ctx, eq.cfg, eq.l1Fetcher, eq.engine)
result, err := sync.FindL2Heads(ctx, eq.cfg, eq.l1Fetcher, eq.engine, eq.log)
if err != nil {
return NewTemporaryError(fmt.Errorf("failed to find the L2 Heads to start from: %w", err))
}
......
......@@ -266,7 +266,7 @@ func (s *Driver) eventLoop() {
s.log.Warn("not creating block, node is deriving new l2 data", "head_l1", l1Head)
break
}
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
ctx, cancel := context.WithTimeout(ctx, 20*time.Minute)
err := s.createNewL2Block(ctx)
cancel()
if err != nil {
......@@ -309,9 +309,7 @@ func (s *Driver) eventLoop() {
s.metrics.SetDerivationIdle(false)
s.idleDerivation = false
s.log.Debug("Derivation process step", "onto_origin", s.derivation.Origin(), "attempts", stepAttempts)
stepCtx, cancel := context.WithTimeout(ctx, time.Second*10) // TODO pick a timeout for executing a single step
err := s.derivation.Step(stepCtx)
cancel()
err := s.derivation.Step(context.Background())
stepAttempts += 1 // count as attempt by default. We reset to 0 if we are making healthy progress.
if err == io.EOF {
s.log.Debug("Derivation process went idle", "progress", s.derivation.Origin())
......
......@@ -32,6 +32,7 @@ import (
"github.com/ethereum-optimism/optimism/op-node/rollup"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
)
type L1Chain interface {
......@@ -101,7 +102,7 @@ func currentHeads(ctx context.Context, cfg *rollup.Config, l2 L2Chain) (*FindHea
// Plausible: meaning that the blockhash of the L2 block's L1 origin
// (as reported in the L1 Attributes deposit within the L2 block) is not canonical at another height in the L1 chain,
// and the same holds for all its ancestors.
func FindL2Heads(ctx context.Context, cfg *rollup.Config, l1 L1Chain, l2 L2Chain) (result *FindHeadsResult, err error) {
func FindL2Heads(ctx context.Context, cfg *rollup.Config, l1 L1Chain, l2 L2Chain, lgr log.Logger) (result *FindHeadsResult, err error) {
// Fetch current L2 forkchoice state
result, err = currentHeads(ctx, cfg, l2)
if err != nil {
......@@ -137,6 +138,8 @@ func FindL2Heads(ctx context.Context, cfg *rollup.Config, l1 L1Chain, l2 L2Chain
ahead = notFound
}
lgr.Trace("walking sync start", "number", n.Number)
// Don't walk past genesis. If we were at the L2 genesis, but could not find its L1 origin,
// the L2 chain is building on the wrong L1 branch.
if n.Number == cfg.Genesis.L2.Number {
......
......@@ -73,7 +73,9 @@ func (c *syncStartTestCase) Run(t *testing.T) {
Genesis: genesis,
SeqWindowSize: c.SeqWindowSize,
}
result, err := FindL2Heads(context.Background(), cfg, chain, chain)
lgr := log.New()
lgr.SetHandler(log.DiscardHandler())
result, err := FindL2Heads(context.Background(), cfg, chain, chain, lgr)
if c.ExpectedErr != nil {
require.ErrorIs(t, err, c.ExpectedErr, "expected error")
return
......
......@@ -7,7 +7,7 @@ import '@eth-optimism/hardhat-deploy-config'
const upgradeABIs = {
L2OutputOracleProxy: async (deployConfig) => [
'initialize(bytes32,uint256,address,address)',
'initialize(bytes32,uint256,address)',
[
deployConfig.l2OutputOracleGenesisL2Output,
deployConfig.l2OutputOracleProposer,
......
# @eth-optimism/proxyd
## 3.12.0
### Minor Changes
- e9f2c701: Allow disabling backend rate limiter
- ca45a85e: Support pattern matching in exempt origins/user agents
- f4faa44c: adds server.log_level config
## 3.11.0
### Minor Changes
......
{
"name": "@eth-optimism/proxyd",
"version": "3.11.0",
"version": "3.12.0",
"private": true,
"dependencies": {}
}
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