Merge pull request #7754 from ethereum-optimism/felipe/avoid-caching-empty-get-raw-receipts

feat(proxyd): avoid caching debug_getRawReceipts responses with 0 receipts

proxyd/backend.go

@@ -665,9 +665,9 @@ type BackendGroup struct {
 	Consensus *ConsensusPoller
 }
 
-func (bg *BackendGroup) Forward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool) ([]*RPCRes, error) {
+func (bg *BackendGroup) Forward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool) ([]*RPCRes, string, error) {
 	if len(rpcReqs) == 0 {
-		return nil, nil
+		return nil, "", nil
 	}
 
 	backends := bg.Backends
@@ -731,7 +731,7 @@ func (bg *BackendGroup) Forward(ctx context.Context, rpcReqs []*RPCReq, isBatch
 			if errors.Is(err, ErrConsensusGetReceiptsCantBeBatched) ||
 				errors.Is(err, ErrConsensusGetReceiptsInvalidTarget) ||
 				errors.Is(err, ErrMethodNotWhitelisted) {
-				return nil, err
+				return nil, "", err
 			}
 			if errors.Is(err, ErrBackendOffline) {
 				log.Warn(
@@ -773,11 +773,12 @@ func (bg *BackendGroup) Forward(ctx context.Context, rpcReqs []*RPCReq, isBatch
 			}
 		}
 
-		return res, nil
+		servedBy := fmt.Sprintf("%s/%s", bg.Name, back.Name)
+		return res, servedBy, nil
 	}
 
 	RecordUnserviceableRequest(ctx, RPCRequestSourceHTTP)
-	return nil, ErrNoBackends
+	return nil, "", ErrNoBackends
 }
 
 func (bg *BackendGroup) ProxyWS(ctx context.Context, clientConn *websocket.Conn, methodWhitelist *StringSet) (*WSProxier, error) {

proxyd/cache.go

@@ -128,7 +128,7 @@ type rpcCache struct {
 func newRPCCache(cache Cache) RPCCache {
 	staticHandler := &StaticMethodHandler{cache: cache}
 	debugGetRawReceiptsHandler := &StaticMethodHandler{cache: cache,
-		filter: func(req *RPCReq) bool {
+		filterGet: func(req *RPCReq) bool {
 			// cache only if the request is for a block hash
 
 			var p []rpc.BlockNumberOrHash
@@ -141,6 +141,14 @@ func newRPCCache(cache Cache) RPCCache {
 			}
 			return p[0].BlockHash != nil
 		},
+		filterPut: func(req *RPCReq, res *RPCRes) bool {
+			// don't cache if response contains 0 receipts
+			rawReceipts, ok := res.Result.([]interface{})
+			if !ok {
+				return false
+			}
+			return len(rawReceipts) > 0
+		},
 	}
 	handlers := map[string]RPCMethodHandler{
 		"eth_chainId":                           staticHandler,

proxyd/cache_test.go

@@ -110,7 +110,7 @@ func TestRPCCacheImmutableRPCs(t *testing.T) {
 			},
 			res: &RPCRes{
 				JSONRPC: "2.0",
-				Result:  `{"debug_getRawReceipts":"!"}`,
+				Result:  []interface{}{"a"},
 				ID:      ID,
 			},
 			name: "debug_getRawReceipts",

proxyd/config.go

@@ -24,6 +24,8 @@ type ServerConfig struct {
 
 	EnableRequestLog     bool `toml:"enable_request_log"`
 	MaxRequestBodyLogLen int  `toml:"max_request_body_log_len"`
+
+	EnableXServedByHeader bool `toml:"enable_served_by_header"`
 }
 
 type CacheConfig struct {

proxyd/integration_tests/caching_test.go

@@ -27,6 +27,7 @@ func TestCaching(t *testing.T) {
 	hdlr.SetRoute("eth_getTransactionByBlockHashAndIndex", "999", "eth_getTransactionByBlockHashAndIndex")
 	hdlr.SetRoute("eth_getUncleByBlockHashAndIndex", "999", "eth_getUncleByBlockHashAndIndex")
 	hdlr.SetRoute("eth_getTransactionReceipt", "999", "eth_getTransactionReceipt")
+	hdlr.SetRoute("debug_getRawReceipts", "999", "debug_getRawReceipts")
 	/* not cacheable */
 	hdlr.SetRoute("eth_getBlockByNumber", "999", "eth_getBlockByNumber")
 	hdlr.SetRoute("eth_blockNumber", "999", "eth_blockNumber")
@@ -180,6 +181,30 @@ func TestCaching(t *testing.T) {
 		RequireEqualJSON(t, resRaw, resCache)
 		require.Equal(t, 2, countRequests(backend, "eth_getBlockByHash"))
 	})
+
+	t.Run("debug_getRawReceipts with 0 receipts should not be cached", func(t *testing.T) {
+		backend.Reset()
+		hdlr.SetRoute("debug_getRawReceipts", "999", []string{})
+		resRaw, _, err := client.SendRPC("debug_getRawReceipts", []interface{}{"0x88420081ab9c6d50dc57af36b541c6b8a7b3e9c0d837b0414512c4c5883560ff"})
+		require.NoError(t, err)
+		resCache, _, err := client.SendRPC("debug_getRawReceipts", []interface{}{"0x88420081ab9c6d50dc57af36b541c6b8a7b3e9c0d837b0414512c4c5883560ff"})
+		require.NoError(t, err)
+		RequireEqualJSON(t, []byte("{\"id\":999,\"jsonrpc\":\"2.0\",\"result\":[]}"), resRaw)
+		RequireEqualJSON(t, resRaw, resCache)
+		require.Equal(t, 2, countRequests(backend, "debug_getRawReceipts"))
+	})
+
+	t.Run("debug_getRawReceipts with more than 0 receipts should be cached", func(t *testing.T) {
+		backend.Reset()
+		hdlr.SetRoute("debug_getRawReceipts", "999", []string{"a"})
+		resRaw, _, err := client.SendRPC("debug_getRawReceipts", []interface{}{"0x88420081ab9c6d50dc57af36b541c6b8a7b3e9c0d837b0414512c4c5883560bb"})
+		require.NoError(t, err)
+		resCache, _, err := client.SendRPC("debug_getRawReceipts", []interface{}{"0x88420081ab9c6d50dc57af36b541c6b8a7b3e9c0d837b0414512c4c5883560bb"})
+		require.NoError(t, err)
+		RequireEqualJSON(t, []byte("{\"id\":999,\"jsonrpc\":\"2.0\",\"result\":[\"a\"]}"), resRaw)
+		RequireEqualJSON(t, resRaw, resCache)
+		require.Equal(t, 1, countRequests(backend, "debug_getRawReceipts"))
+	})
 }
 
 func TestBatchCaching(t *testing.T) {

proxyd/integration_tests/mock_backend_test.go

@@ -77,6 +77,7 @@ func (h *BatchRPCResponseRouter) SetRoute(method string, id string, result inter
 
 	switch result.(type) {
 	case string:
+	case []string:
 	case nil:
 		break
 	default:

proxyd/integration_tests/testdata/caching.toml

@@ -33,3 +33,4 @@ eth_getTransactionByHash = "main"
 eth_getTransactionByBlockHashAndIndex = "main"
 eth_getUncleByBlockHashAndIndex = "main"
 eth_getTransactionReceipt = "main"
+debug_getRawReceipts = "main"

proxyd/integration_tests/testdata/consensus_responses.yml

@@ -185,6 +185,27 @@
       }
     }
 - method: debug_getRawReceipts
+  block: 0x55
+  response: >
+    {
+      "jsonrpc": "2.0",
+      "id": 67,
+      "result": {
+        "_": "debug_getRawReceipts"
+      }
+    }
+- method: debug_getRawReceipts
+  block: 0xc6ef2fc5426d6ad6fd9e2a26abeab0aa2411b7ab17f30a99d3cb96aed1d1055b
+  response: >
+    {
+      "jsonrpc": "2.0",
+      "id": 67,
+      "result": {
+        "_": "debug_getRawReceipts"
+      }
+    }
+- method: debug_getRawReceipts
+  block: 0x101
   response: >
     {
       "jsonrpc": "2.0",

proxyd/methods.go

@@ -17,9 +17,10 @@ type RPCMethodHandler interface {
 }
 
 type StaticMethodHandler struct {
-	cache  Cache
-	m      sync.RWMutex
-	filter func(*RPCReq) bool
+	cache     Cache
+	m         sync.RWMutex
+	filterGet func(*RPCReq) bool
+	filterPut func(*RPCReq, *RPCRes) bool
 }
 
 func (e *StaticMethodHandler) key(req *RPCReq) string {
@@ -34,7 +35,7 @@ func (e *StaticMethodHandler) GetRPCMethod(ctx context.Context, req *RPCReq) (*R
 	if e.cache == nil {
 		return nil, nil
 	}
-	if e.filter != nil && !e.filter(req) {
+	if e.filterGet != nil && !e.filterGet(req) {
 		return nil, nil
 	}
 
@@ -67,7 +68,12 @@ func (e *StaticMethodHandler) PutRPCMethod(ctx context.Context, req *RPCReq, res
 	if e.cache == nil {
 		return nil
 	}
-	if e.filter != nil && !e.filter(req) {
+	// if there is a filter on get, we don't want to cache it because its irretrievable
+	if e.filterGet != nil && !e.filterGet(req) {
+		return nil
+	}
+	// response filter
+	if e.filterPut != nil && !e.filterPut(req, res) {
 		return nil
 	}
 

proxyd/proxyd.go

@@ -235,6 +235,7 @@ func Start(config *Config) (*Server, func(), error) {
 		resolvedAuth,
 		secondsToDuration(config.Server.TimeoutSeconds),
 		config.Server.MaxUpstreamBatchSize,
+		config.Server.EnableXServedByHeader,
 		rpcCache,
 		config.RateLimit,
 		config.SenderRateLimit,

proxyd/server.go

@@ -60,6 +60,7 @@ type Server struct {
 	timeout                time.Duration
 	maxUpstreamBatchSize   int
 	maxBatchSize           int
+	enableServedByHeader   bool
 	upgrader               *websocket.Upgrader
 	mainLim                FrontendRateLimiter
 	overrideLims           map[string]FrontendRateLimiter
@@ -85,6 +86,7 @@ func NewServer(
 	authenticatedPaths map[string]string,
 	timeout time.Duration,
 	maxUpstreamBatchSize int,
+	enableServedByHeader bool,
 	cache RPCCache,
 	rateLimitConfig RateLimitConfig,
 	senderRateLimitConfig SenderRateLimitConfig,
@@ -175,6 +177,7 @@ func NewServer(
 		authenticatedPaths:   authenticatedPaths,
 		timeout:              timeout,
 		maxUpstreamBatchSize: maxUpstreamBatchSize,
+		enableServedByHeader: enableServedByHeader,
 		cache:                cache,
 		enableRequestLog:     enableRequestLog,
 		maxRequestBodyLogLen: maxRequestBodyLogLen,
@@ -354,7 +357,7 @@ func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 
-		batchRes, batchContainsCached, err := s.handleBatchRPC(ctx, reqs, isLimited, true)
+		batchRes, batchContainsCached, servedBy, err := s.handleBatchRPC(ctx, reqs, isLimited, true)
 		if err == context.DeadlineExceeded {
 			writeRPCError(ctx, w, nil, ErrGatewayTimeout)
 			return
@@ -368,14 +371,16 @@ func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) {
 			writeRPCError(ctx, w, nil, ErrInternal)
 			return
 		}
-
+		if s.enableServedByHeader {
+			w.Header().Set("x-served-by", servedBy)
+		}
 		setCacheHeader(w, batchContainsCached)
 		writeBatchRPCRes(ctx, w, batchRes)
 		return
 	}
 
 	rawBody := json.RawMessage(body)
-	backendRes, cached, err := s.handleBatchRPC(ctx, []json.RawMessage{rawBody}, isLimited, false)
+	backendRes, cached, servedBy, err := s.handleBatchRPC(ctx, []json.RawMessage{rawBody}, isLimited, false)
 	if err != nil {
 		if errors.Is(err, ErrConsensusGetReceiptsCantBeBatched) ||
 			errors.Is(err, ErrConsensusGetReceiptsInvalidTarget) {
@@ -385,11 +390,14 @@ func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) {
 		writeRPCError(ctx, w, nil, ErrInternal)
 		return
 	}
+	if s.enableServedByHeader {
+		w.Header().Set("x-served-by", servedBy)
+	}
 	setCacheHeader(w, cached)
 	writeRPCRes(ctx, w, backendRes[0])
 }
 
-func (s *Server) handleBatchRPC(ctx context.Context, reqs []json.RawMessage, isLimited limiterFunc, isBatch bool) ([]*RPCRes, bool, error) {
+func (s *Server) handleBatchRPC(ctx context.Context, reqs []json.RawMessage, isLimited limiterFunc, isBatch bool) ([]*RPCRes, bool, string, error) {
 	// A request set is transformed into groups of batches.
 	// Each batch group maps to a forwarded JSON-RPC batch request (subject to maxUpstreamBatchSize constraints)
 	// A groupID is used to decouple Requests that have duplicate ID so they're not part of the same batch that's
@@ -475,6 +483,7 @@ func (s *Server) handleBatchRPC(ctx context.Context, reqs []json.RawMessage, isL
 		batches[batchGroup] = append(batches[batchGroup], batchElem{parsedReq, i})
 	}
 
+	servedBy := make(map[string]bool, 0)
 	var cached bool
 	for group, batch := range batches {
 		var cacheMisses []batchElem
@@ -499,17 +508,18 @@ func (s *Server) handleBatchRPC(ctx context.Context, reqs []json.RawMessage, isL
 					"batch_index", i,
 				)
 				batchRPCShortCircuitsTotal.Inc()
-				return nil, false, context.DeadlineExceeded
+				return nil, false, "", context.DeadlineExceeded
 			}
 
 			start := i * s.maxUpstreamBatchSize
 			end := int(math.Min(float64(start+s.maxUpstreamBatchSize), float64(len(cacheMisses))))
 			elems := cacheMisses[start:end]
-			res, err := s.BackendGroups[group.backendGroup].Forward(ctx, createBatchRequest(elems), isBatch)
+			res, sb, err := s.BackendGroups[group.backendGroup].Forward(ctx, createBatchRequest(elems), isBatch)
+			servedBy[sb] = true
 			if err != nil {
 				if errors.Is(err, ErrConsensusGetReceiptsCantBeBatched) ||
 					errors.Is(err, ErrConsensusGetReceiptsInvalidTarget) {
-					return nil, false, err
+					return nil, false, "", err
 				}
 				log.Error(
 					"error forwarding RPC batch",
@@ -541,7 +551,15 @@ func (s *Server) handleBatchRPC(ctx context.Context, reqs []json.RawMessage, isL
 		}
 	}
 
-	return responses, cached, nil
+	servedByString := ""
+	for sb, _ := range servedBy {
+		if servedByString != "" {
+			servedByString += ", "
+		}
+		servedByString += sb
+	}
+
+	return responses, cached, servedByString, nil
 }
 
 func (s *Server) HandleWS(w http.ResponseWriter, r *http.Request) {

proxyd/tools/mockserver/handler/handler.go

@@ -77,7 +77,7 @@ func (mh *MockedHandler) Handler(w http.ResponseWriter, req *http.Request) {
 	for _, r := range requests {
 		method := r["method"]
 		block := ""
-		if method == "eth_getBlockByNumber" {
+		if method == "eth_getBlockByNumber" || method == "debug_getRawReceipts" {
 			block = (r["params"].([]interface{})[0]).(string)
 		}