rebase

go/proxyd/cache.go

@@ -81,6 +81,9 @@ func (c *redisCache) Put(ctx context.Context, key string, value string) error {
 
 func (c *redisCache) Remove(ctx context.Context, key string) error {
 	err := c.rdb.Del(ctx, key).Err()
+	if err != nil {
+		RecordRedisError("CacheDel")
+	}
 	return err
 }
 
@@ -122,7 +125,15 @@ func (c *rpcCache) GetRPC(ctx context.Context, req *RPCReq) (*RPCRes, error) {
 	if handler == nil {
 		return nil, nil
 	}
-	return handler.GetRPCMethod(ctx, req)
+	res, err := handler.GetRPCMethod(ctx, req)
+	if res != nil {
+		if res == nil {
+			RecordCacheMiss(req.Method)
+		} else {
+			RecordCacheHit(req.Method)
+		}
+	}
+	return res, err
 }
 
 func (c *rpcCache) PutRPC(ctx context.Context, req *RPCReq, res *RPCRes, blockNumberSync uint64) error {

go/proxyd/integration_tests/failover_test.go

@@ -12,7 +12,7 @@ import (
 )
 
 const (
-	goodResponse = `{"jsonrpc": "2.0", "result": "hello", "id": 999}`
+	goodResponse       = `{"jsonrpc": "2.0", "result": "hello", "id": 999}`
 	noBackendsResponse = `{"error":{"code":-32011,"message":"no backends available for method"},"id":999,"jsonrpc":"2.0"}`
 )
 

go/proxyd/proxyd.go

@@ -113,7 +113,7 @@ func Start(config *Config) (func(), error) {
 		opts = append(opts, WithProxydIP(os.Getenv("PROXYD_IP")))
 		back, err := NewBackend(name, rpcURL, wsURL, lim, opts...)
 		if err != nil {
-			return err
+			return nil, err
 		}
 		back.Start()
 		defer back.Stop()
@@ -170,8 +170,8 @@ func Start(config *Config) (func(), error) {
 	}
 
 	var rpcCache RPCCache
+	stopLVCs := make(chan struct{})
 	if config.Cache.Enabled {
-		var getLatestBlockNumFn GetLatestBlockNumFn
 		if config.Cache.BlockSyncRPCURL == "" {
 			return nil, fmt.Errorf("block sync node required for caching")
 		}
@@ -179,6 +179,9 @@ func Start(config *Config) (func(), error) {
 		if err != nil {
 			return nil, err
 		}
+		if blockSyncRPCURL == "" {
+			return nil, fmt.Errorf("block sync node config is required for caching")
+		}
 
 		var cache Cache
 		if redisURL != "" {
@@ -189,20 +192,14 @@ func Start(config *Config) (func(), error) {
 			log.Warn("redis is not configured, using in-memory cache")
 			cache = newMemoryCache()
 		}
-
-		if config.Cache.BlockSyncRPCURL == "" {
-			return fmt.Errorf("block sync node config is required for caching")
-		}
 		// Ideally, the BlocKSyncRPCURL should be the sequencer or a HA replica that's not far behind
 		ethClient, err := ethclient.Dial(config.Cache.BlockSyncRPCURL)
 		if err != nil {
 			return nil, err
 		}
 		defer ethClient.Close()
-		lvcCtx, lvcCancel := context.WithCancel(context.Background())
-		defer lvcCancel()
-		blockNumFn := makeGetLatestBlockNumFn(ethClient, lvcCtx.Done())
-		gasPriceFn := makeGetLatestGasPriceFn(ethClient, lvcCtx.Done())
+		blockNumFn := makeGetLatestBlockNumFn(ethClient, stopLVCs)
+		gasPriceFn := makeGetLatestGasPriceFn(ethClient, stopLVCs)
 		rpcCache = newRPCCache(cache, blockNumFn, gasPriceFn)
 	}
 
@@ -256,7 +253,8 @@ func Start(config *Config) (func(), error) {
 
 	return func() {
 		log.Info("shutting down proxyd")
-        // TODO(inphi): Stop LVCs here
+		// TODO(inphi): Stop LVCs here
+		close(stopLVCs)
 		srv.Shutdown()
 		if err := lim.FlushBackendWSConns(backendNames); err != nil {
 			log.Error("error flushing backend ws conns", "err", err)

go/proxyd/rpc.go

@@ -121,4 +121,4 @@ func IsBatch(raw []byte) bool {
 		return c == '['
 	}
 	return false
-}
\ No newline at end of file
+}