feat: batch fetch backends status

backend.go

@@ -489,6 +489,14 @@ func (b *Backend) ForwardRPC(ctx context.Context, res *RPCRes, id string, method
 }
 
 func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool, sem *semaphore.Weighted) ([]*RPCRes, error) {
+	var netErr error
+
+	defer func() {
+		if netErr != nil {
+			log.Info("doForward net error", "err", netErr.Error())
+		}
+	}()
+
 	// we are concerned about network error rates, so we record 1 request independently of how many are in the batch
 	b.networkRequestsSlidingWindow.Incr()
 
@@ -558,6 +566,7 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
 
 	httpReq, err := http.NewRequestWithContext(ctx, "POST", b.rpcURL, bytes.NewReader(body))
 	if err != nil {
+		netErr = err
 		b.intermittentErrorsSlidingWindow.Incr()
 		RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
 		return nil, wrapErr(err, "error creating backend request")
@@ -594,6 +603,7 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
 	start := time.Now()
 	httpRes, err := b.client.DoWithSemaphore(httpReq, sem)
 	if err != nil {
+		netErr = err
 		b.intermittentErrorsSlidingWindow.Incr()
 		RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
 		return nil, wrapErr(err, "error in backend request")
@@ -613,6 +623,7 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
 
 	// Alchemy returns a 400 on bad JSONs, so handle that case
 	if httpRes.StatusCode != 200 && httpRes.StatusCode != 400 {
+		netErr = errors.New("Not 200")
 		b.intermittentErrorsSlidingWindow.Incr()
 		RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
 		return nil, fmt.Errorf("response code %d", httpRes.StatusCode)
@@ -624,6 +635,7 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
 		return nil, ErrBackendResponseTooLarge
 	}
 	if err != nil {
+		netErr = err
 		b.intermittentErrorsSlidingWindow.Incr()
 		RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
 		return nil, wrapErr(err, "error reading response body")
@@ -640,6 +652,7 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
 		}
 	} else {
 		if err := json.Unmarshal(resB, &rpcRes); err != nil {
+			netErr = err
 			// Infura may return a single JSON-RPC response if, for example, the batch contains a request for an unsupported method
 			if responseIsNotBatched(resB) {
 				b.intermittentErrorsSlidingWindow.Incr()
@@ -653,6 +666,7 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
 	}
 
 	if len(rpcReqs) != len(rpcRes) {
+		netErr = errors.New("number of RPC requests does not match number of responses")
 		b.intermittentErrorsSlidingWindow.Incr()
 		RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
 		return nil, ErrBackendUnexpectedJSONRPC

consensus_poller.go

@@ -2,6 +2,7 @@ package proxyd
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strconv"
 	"strings"
@@ -341,35 +342,21 @@ func (cp *ConsensusPoller) UpdateBackend(ctx context.Context, be *Backend) {
 		RecordConsensusBackendPeerCount(be, peerCount)
 	}
 
-	latestBlockNumber, latestBlockHash, err := cp.fetchBlock(ctx, be, "latest")
+	latestBlockNumber, latestBlockHash, safeBlockNumber, finalizedBlockNumber, err := cp.fetchBlocksBatch(ctx, be)
 	if err != nil {
-		log.Warn("error updating backend - latest block will not be updated", "name", be.Name, "err", err)
+		log.Warn("error updating backend - blocks will not be updated", "name", be.Name, "err", err)
 		return
 	}
 	if latestBlockNumber == 0 {
 		log.Warn("error backend responded a 200 with blockheight 0 for latest block", "name", be.Name)
 		be.intermittentErrorsSlidingWindow.Incr()
 		return
 	}
-
-	safeBlockNumber, _, err := cp.fetchBlock(ctx, be, "safe")
-	if err != nil {
-		log.Warn("error updating backend - safe block will not be updated", "name", be.Name, "err", err)
-		return
-	}
-
 	if safeBlockNumber == 0 {
 		log.Warn("error backend responded a 200 with blockheight 0 for safe block", "name", be.Name)
 		be.intermittentErrorsSlidingWindow.Incr()
 		return
 	}
-
-	finalizedBlockNumber, _, err := cp.fetchBlock(ctx, be, "finalized")
-	if err != nil {
-		log.Warn("error updating backend - finalized block will not be updated", "name", be.Name, "err", err)
-		return
-	}
-
 	if finalizedBlockNumber == 0 {
 		log.Warn("error backend responded a 200 with blockheight 0 for finalized block", "name", be.Name)
 		be.intermittentErrorsSlidingWindow.Incr()
@@ -607,6 +594,86 @@ func (cp *ConsensusPoller) Reset() {
 	}
 }
 
+// fetchBlocksBatch fetches latest, safe and finalized blocks in a single batch RPC call
+// to ensure all three heights reflect the same point-in-time snapshot on the backend.
+func (cp *ConsensusPoller) fetchBlocksBatch(ctx context.Context, be *Backend) (
+	latestBlockNumber hexutil.Uint64, latestBlockHash string,
+	safeBlockNumber hexutil.Uint64, finalizedBlockNumber hexutil.Uint64, err error,
+) {
+	buildReq := func(id string, tag string) *RPCReq {
+		params, _ := json.Marshal([]interface{}{tag, false})
+		return &RPCReq{
+			JSONRPC: JSONRPCVersion,
+			Method:  "eth_getBlockByNumber",
+			Params:  params,
+			ID:      []byte(`"` + id + `"`),
+		}
+	}
+
+	reqs := []*RPCReq{
+		buildReq("latest", "latest"),
+		buildReq("safe", "safe"),
+		buildReq("finalized", "finalized"),
+	}
+
+	results, err := be.doForward(ctx, reqs, true, be.consensusSemaphore)
+	if err != nil {
+		return 0, "", 0, 0, err
+	}
+	if len(results) != 3 {
+		return 0, "", 0, 0, fmt.Errorf("unexpected batch response length %d (expected 3) from backend %s", len(results), be.Name)
+	}
+
+	parseResult := func(res *RPCRes, tag string) (hexutil.Uint64, string, error) {
+		if res.IsError() {
+			return 0, "", fmt.Errorf("eth_getBlockByNumber(%s) error from backend %s: %s", tag, be.Name, res.Error.Error())
+		}
+		jsonMap, ok := res.Result.(map[string]interface{})
+		if !ok {
+			return 0, "", fmt.Errorf("unexpected response to eth_getBlockByNumber(%s) on backend %s", tag, be.Name)
+		}
+		num := hexutil.Uint64(hexutil.MustDecodeUint64(jsonMap["number"].(string)))
+		hash, _ := jsonMap["hash"].(string)
+		return num, hash, nil
+	}
+
+	// sortBatchRPCResponse orders by request ID, but our IDs are strings "latest"/"safe"/"finalized"
+	// so we match by ID manually.
+	byID := make(map[string]*RPCRes, 3)
+	for _, r := range results {
+		byID[string(r.ID)] = r
+	}
+
+	latestRes, ok := byID[`"latest"`]
+	if !ok {
+		return 0, "", 0, 0, fmt.Errorf("missing latest block response from backend %s", be.Name)
+	}
+	latestBlockNumber, latestBlockHash, err = parseResult(latestRes, "latest")
+	if err != nil {
+		return 0, "", 0, 0, err
+	}
+
+	safeRes, ok := byID[`"safe"`]
+	if !ok {
+		return 0, "", 0, 0, fmt.Errorf("missing safe block response from backend %s", be.Name)
+	}
+	safeBlockNumber, _, err = parseResult(safeRes, "safe")
+	if err != nil {
+		return 0, "", 0, 0, err
+	}
+
+	finalizedRes, ok := byID[`"finalized"`]
+	if !ok {
+		return 0, "", 0, 0, fmt.Errorf("missing finalized block response from backend %s", be.Name)
+	}
+	finalizedBlockNumber, _, err = parseResult(finalizedRes, "finalized")
+	if err != nil {
+		return 0, "", 0, 0, err
+	}
+
+	return latestBlockNumber, latestBlockHash, safeBlockNumber, finalizedBlockNumber, nil
+}
+
 // fetchBlock is a convenient wrapper to make a request to get a block directly from the backend
 func (cp *ConsensusPoller) fetchBlock(ctx context.Context, be *Backend, block string) (blockNumber hexutil.Uint64, blockHash string, err error) {
 	var rpcRes RPCRes

integration_tests/block_height_zero_and_network_errors_test.go

@@ -336,12 +336,14 @@ func TestBlockHeightZero(t *testing.T) {
 		overrideBlock("node1", "finalized", "0x0", 403)
 		overridePeerCount("node1", 0, 500)
 
-		for i := 1; i < 7; i++ {
+		for i := 1; i < 5; i++ {
 			require.False(t, bg.Consensus.IsBanned(nodes["node1"].backend), "Execpted node 1 to be not banned on iteration ", i)
 			require.False(t, bg.Consensus.IsBanned(nodes["node2"].backend), "Execpted node 2 to be not banned on iteration ", i)
 			update()
-			// On the 5th update (i=6), node 1 will be banned due to error rate and not increment window
-			if i < 6 {
+			// On the 4th iteration (i=4), node 1 will be banned due to error rate and not increment window.
+			// With batch fetch, each consensus poll counts as 1 network request instead of 3 separate
+			// requests, so the error rate threshold is reached sooner than with individual calls.
+			if i < 4 {
 				require.Equal(t, nodes["node1"].intermittentNetErrorWindow.Count(), uint(i))
 			}
 			require.Equal(t, nodes["node2"].intermittentNetErrorWindow.Count(), uint(0))

integration_tests/ws_test.go

@@ -218,10 +218,13 @@ func TestWSClientExceedReadLimit(t *testing.T) {
 	}, nil)
 	require.NoError(t, err)
 
-	closed := false
+	closedCh := make(chan struct{}, 1)
 	originalHandler := client.conn.CloseHandler()
 	client.conn.SetCloseHandler(func(code int, text string) error {
-		closed = true
+		select {
+		case closedCh <- struct{}{}:
+		default:
+		}
 		return originalHandler(code, text)
 	})
 
@@ -236,6 +239,12 @@ func TestWSClientExceedReadLimit(t *testing.T) {
 		[]byte(clientReq),
 	)
 	require.Error(t, err)
-	require.True(t, closed)
+
+	// Wait for the server's close frame to be processed by the readPump goroutine.
+	select {
+	case <-closedCh:
+	case <-time.After(5 * time.Second):
+		t.Fatal("timed out waiting for connection close")
+	}
 
 }