From 66a405721e3d486f11160bb5482d268c6b2e34d3 Mon Sep 17 00:00:00 2001 From: rekyyang <22545420+rekyyang@users.noreply.github.com> Date: Mon, 1 Jun 2026 11:01:56 +0800 Subject: [PATCH 1/4] batch fetch --- consensus_poller.go | 99 +++++++++++++++++++++++++++++++++++++-------- 1 file changed, 83 insertions(+), 16 deletions(-) diff --git a/consensus_poller.go b/consensus_poller.go index c476d61..e954286 100644 --- a/consensus_poller.go +++ b/consensus_poller.go @@ -2,6 +2,7 @@ package proxyd import ( "context" + "encoding/json" "fmt" "strconv" "strings" @@ -341,9 +342,9 @@ 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 { @@ -351,25 +352,11 @@ func (cp *ConsensusPoller) UpdateBackend(ctx context.Context, be *Backend) { 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 From 3ea0b55f076353cf2be2e8ae2cc94acd1062c72e Mon Sep 17 00:00:00 2001 From: rekyyang <22545420+rekyyang@users.noreply.github.com> Date: Wed, 3 Jun 2026 16:23:39 +0800 Subject: [PATCH 2/4] add debug logs --- backend.go | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/backend.go b/backend.go index 0c61424..5e84061 100644 --- a/backend.go +++ b/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 From f8a2cfff9241f28e64e5a4bab169191ef6cbb055 Mon Sep 17 00:00:00 2001 From: rekyyang <22545420+rekyyang@users.noreply.github.com> Date: Wed, 3 Jun 2026 17:54:34 +0800 Subject: [PATCH 3/4] add debug logs --- consensus_poller.go | 8 ++++---- .../block_height_zero_and_network_errors_test.go | 8 +++++--- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/consensus_poller.go b/consensus_poller.go index e954286..139967f 100644 --- a/consensus_poller.go +++ b/consensus_poller.go @@ -606,7 +606,7 @@ func (cp *ConsensusPoller) fetchBlocksBatch(ctx context.Context, be *Backend) ( JSONRPC: JSONRPCVersion, Method: "eth_getBlockByNumber", Params: params, - ID: []byte(id), + ID: []byte(`"` + id + `"`), } } @@ -644,7 +644,7 @@ func (cp *ConsensusPoller) fetchBlocksBatch(ctx context.Context, be *Backend) ( byID[string(r.ID)] = r } - latestRes, ok := byID["latest"] + latestRes, ok := byID[`"latest"`] if !ok { return 0, "", 0, 0, fmt.Errorf("missing latest block response from backend %s", be.Name) } @@ -653,7 +653,7 @@ func (cp *ConsensusPoller) fetchBlocksBatch(ctx context.Context, be *Backend) ( return 0, "", 0, 0, err } - safeRes, ok := byID["safe"] + safeRes, ok := byID[`"safe"`] if !ok { return 0, "", 0, 0, fmt.Errorf("missing safe block response from backend %s", be.Name) } @@ -662,7 +662,7 @@ func (cp *ConsensusPoller) fetchBlocksBatch(ctx context.Context, be *Backend) ( return 0, "", 0, 0, err } - finalizedRes, ok := byID["finalized"] + finalizedRes, ok := byID[`"finalized"`] if !ok { return 0, "", 0, 0, fmt.Errorf("missing finalized block response from backend %s", be.Name) } diff --git a/integration_tests/block_height_zero_and_network_errors_test.go b/integration_tests/block_height_zero_and_network_errors_test.go index 2f241fd..60ce0bb 100644 --- a/integration_tests/block_height_zero_and_network_errors_test.go +++ b/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)) From d2d688b27331a28e497473f647b7e014d3b201f0 Mon Sep 17 00:00:00 2001 From: rekyyang <22545420+rekyyang@users.noreply.github.com> Date: Wed, 3 Jun 2026 21:27:50 +0800 Subject: [PATCH 4/4] fix ut --- integration_tests/ws_test.go | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/integration_tests/ws_test.go b/integration_tests/ws_test.go index ac47278..e846e98 100644 --- a/integration_tests/ws_test.go +++ b/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") + } }