Merge pull request #7252 from ethereum-optimism/felipe/proxyd-ha

feat(proxyd): high availability

proxyd/cache.go

@@ -7,8 +7,8 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/redis/go-redis/v9"
 
-	"github.com/go-redis/redis/v8"
 	"github.com/golang/snappy"
 	lru "github.com/hashicorp/golang-lru"
 )
@@ -78,7 +78,7 @@ func (c *redisCache) Get(ctx context.Context, key string) (string, error) {
 
 func (c *redisCache) Put(ctx context.Context, key string, value string) error {
 	start := time.Now()
-	err := c.rdb.SetEX(ctx, c.namespaced(key), value, redisTTL).Err()
+	err := c.rdb.SetEx(ctx, c.namespaced(key), value, redisTTL).Err()
 	redisCacheDurationSumm.WithLabelValues("SETEX").Observe(float64(time.Since(start).Milliseconds()))
 
 	if err != nil {

proxyd/config.go

@@ -110,6 +110,10 @@ type BackendGroupConfig struct {
 	ConsensusMaxBlockLag        uint64       `toml:"consensus_max_block_lag"`
 	ConsensusMaxBlockRange      uint64       `toml:"consensus_max_block_range"`
 	ConsensusMinPeerCount       int          `toml:"consensus_min_peer_count"`
+
+	ConsensusHA                  bool         `toml:"consensus_ha"`
+	ConsensusHAHeartbeatInterval TOMLDuration `toml:"consensus_ha_heartbeat_interval"`
+	ConsensusHALockPeriod        TOMLDuration `toml:"consensus_ha_lock_period"`
 }
 
 type BackendGroupsConfig map[string]*BackendGroupConfig

proxyd/consensus_poller.go

@@ -23,6 +23,7 @@ type OnConsensusBroken func()
 // resolves the highest common block for multiple nodes, and reconciles the consensus
 // in case of block hash divergence to minimize re-orgs
 type ConsensusPoller struct {
+	ctx        context.Context
 	cancelFunc context.CancelFunc
 	listeners  []OnConsensusBroken
 
@@ -220,6 +221,7 @@ func NewConsensusPoller(bg *BackendGroup, opts ...ConsensusOpt) *ConsensusPoller
 	state := make(map[*Backend]*backendState, len(bg.Backends))
 
 	cp := &ConsensusPoller{
+		ctx:          ctx,
 		cancelFunc:   cancelFunc,
 		backendGroup: bg,
 		backendState: state,

proxyd/frontend_rate_limiter.go

@@ -6,7 +6,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/go-redis/redis/v8"
+	"github.com/redis/go-redis/v9"
 )
 
 type FrontendRateLimiter interface {

proxyd/frontend_rate_limiter_test.go

@@ -7,7 +7,7 @@ import (
 	"time"
 
 	"github.com/alicebob/miniredis"
-	"github.com/go-redis/redis/v8"
+	"github.com/redis/go-redis/v9"
 	"github.com/stretchr/testify/require"
 )
 

proxyd/go.mod

@@ -7,7 +7,6 @@ require (
 	github.com/alicebob/miniredis v2.5.0+incompatible
 	github.com/emirpasic/gods v1.18.1
 	github.com/ethereum/go-ethereum v1.12.1
-	github.com/go-redis/redis/v8 v8.11.4
 	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb
 	github.com/gorilla/mux v1.8.0
 	github.com/gorilla/websocket v1.5.0
@@ -44,11 +43,14 @@ require (
 	github.com/ethereum/c-kzg-4844 v0.3.0 // indirect
 	github.com/getsentry/sentry-go v0.18.0 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
+	github.com/go-redsync/redsync/v4 v4.9.4 // indirect
 	github.com/go-stack/stack v1.8.1 // indirect
 	github.com/gofrs/flock v0.8.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/gomodule/redigo v1.8.8 // indirect
+	github.com/gomodule/redigo v1.8.9 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/holiman/bloomfilter/v2 v2.0.3 // indirect
 	github.com/holiman/uint256 v1.2.3 // indirect
 	github.com/klauspost/compress v1.15.15 // indirect
@@ -62,6 +64,7 @@ require (
 	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/common v0.39.0 // indirect
 	github.com/prometheus/procfs v0.9.0 // indirect
+	github.com/redis/go-redis/v9 v9.1.0 // indirect
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/rogpeppe/go-internal v1.9.0 // indirect
 	github.com/shirou/gopsutil v3.21.11+incompatible // indirect

proxyd/metrics.go

@@ -262,6 +262,33 @@ var (
 		"backend_group_name",
 	})
 
+	consensusHALatestBlock = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: MetricsNamespace,
+		Name:      "group_consensus_ha_latest_block",
+		Help:      "Consensus HA latest block",
+	}, []string{
+		"backend_group_name",
+		"leader",
+	})
+
+	consensusHASafeBlock = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: MetricsNamespace,
+		Name:      "group_consensus_ha_safe_block",
+		Help:      "Consensus HA safe block",
+	}, []string{
+		"backend_group_name",
+		"leader",
+	})
+
+	consensusHAFinalizedBlock = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: MetricsNamespace,
+		Name:      "group_consensus_ha_finalized_block",
+		Help:      "Consensus HA finalized block",
+	}, []string{
+		"backend_group_name",
+		"leader",
+	})
+
 	backendLatestBlockBackend = promauto.NewGaugeVec(prometheus.GaugeOpts{
 		Namespace: MetricsNamespace,
 		Name:      "backend_latest_block",
@@ -438,6 +465,18 @@ func RecordBatchSize(size int) {
 	batchSizeHistogram.Observe(float64(size))
 }
 
+func RecordGroupConsensusHALatestBlock(group *BackendGroup, leader string, blockNumber hexutil.Uint64) {
+	consensusHALatestBlock.WithLabelValues(group.Name, leader).Set(float64(blockNumber))
+}
+
+func RecordGroupConsensusHASafeBlock(group *BackendGroup, leader string, blockNumber hexutil.Uint64) {
+	consensusHASafeBlock.WithLabelValues(group.Name, leader).Set(float64(blockNumber))
+}
+
+func RecordGroupConsensusHAFinalizedBlock(group *BackendGroup, leader string, blockNumber hexutil.Uint64) {
+	consensusHAFinalizedBlock.WithLabelValues(group.Name, leader).Set(float64(blockNumber))
+}
+
 func RecordGroupConsensusLatestBlock(group *BackendGroup, blockNumber hexutil.Uint64) {
 	consensusLatestBlock.WithLabelValues(group.Name).Set(float64(blockNumber))
 }

proxyd/proxyd.go

 package proxyd
 
 import (
+	"context"
 	"crypto/tls"
 	"errors"
 	"fmt"
@@ -10,8 +11,8 @@ import (
 
 	"github.com/ethereum/go-ethereum/common/math"
 	"github.com/ethereum/go-ethereum/log"
-	"github.com/go-redis/redis/v8"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/redis/go-redis/v9"
 	"golang.org/x/sync/semaphore"
 )
 
@@ -313,8 +314,28 @@ func Start(config *Config) (*Server, func(), error) {
 				copts = append(copts, WithMaxBlockRange(bgcfg.ConsensusMaxBlockRange))
 			}
 
+			var tracker ConsensusTracker
+			if bgcfg.ConsensusHA {
+				if redisClient == nil {
+					log.Crit("cant start - consensus high availability requires redis")
+				}
+				topts := make([]RedisConsensusTrackerOpt, 0)
+				if bgcfg.ConsensusHALockPeriod > 0 {
+					topts = append(topts, WithLockPeriod(time.Duration(bgcfg.ConsensusHALockPeriod)))
+				}
+				if bgcfg.ConsensusHAHeartbeatInterval > 0 {
+					topts = append(topts, WithLockPeriod(time.Duration(bgcfg.ConsensusHAHeartbeatInterval)))
+				}
+				tracker = NewRedisConsensusTracker(context.Background(), redisClient, bg, bg.Name, topts...)
+				copts = append(copts, WithTracker(tracker))
+			}
+
 			cp := NewConsensusPoller(bg, copts...)
 			bg.Consensus = cp
+
+			if bgcfg.ConsensusHA {
+				tracker.(*RedisConsensusTracker).Init()
+			}
 		}
 	}
 

proxyd/redis.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"time"
 
-	"github.com/go-redis/redis/v8"
+	"github.com/redis/go-redis/v9"
 )
 
 func NewRedisClient(url string) (*redis.Client, error) {

proxyd/server.go

@@ -22,10 +22,10 @@ import (
 	"github.com/ethereum/go-ethereum/core/txpool"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
-	"github.com/go-redis/redis/v8"
 	"github.com/gorilla/mux"
 	"github.com/gorilla/websocket"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/redis/go-redis/v9"
 	"github.com/rs/cors"
 	"github.com/syndtr/goleveldb/leveldb/opt"
 )