Remove polling from request count tests

Author: Michael de Hoog

proxyd/backend.go

@@ -380,7 +380,7 @@ func (b *Backend) Forward(ctx context.Context, reqs []*RPCReq, isBatch bool) ([]
 			"max_attempts", b.maxRetries+1,
 			"method", metricLabelMethod,
 		)
-		res, err := b.doForward(ctx, reqs, isBatch)
+		res, err := b.doForward(ctx, reqs, isBatch, false)
 		switch err {
 		case nil: // do nothing
 		case ErrBackendResponseTooLarge:
@@ -454,6 +454,10 @@ func (b *Backend) ProxyWS(clientConn *websocket.Conn, methodWhitelist *StringSet
 
 // ForwardRPC makes a call directly to a backend and populate the response into `res`
 func (b *Backend) ForwardRPC(ctx context.Context, res *RPCRes, id string, method string, params ...any) error {
+	return b.forwardRPC(ctx, false, res, id, method, params...)
+}
+
+func (b *Backend) forwardRPC(ctx context.Context, isPoll bool, res *RPCRes, id string, method string, params ...any) error {
 	jsonParams, err := json.Marshal(params)
 	if err != nil {
 		return err
@@ -466,7 +470,7 @@ func (b *Backend) ForwardRPC(ctx context.Context, res *RPCRes, id string, method
 		ID:      []byte(id),
 	}
 
-	slicedRes, err := b.doForward(ctx, []*RPCReq{&rpcReq}, false)
+	slicedRes, err := b.doForward(ctx, []*RPCReq{&rpcReq}, false, isPoll)
 	if err != nil {
 		return err
 	}
@@ -482,9 +486,19 @@ func (b *Backend) ForwardRPC(ctx context.Context, res *RPCRes, id string, method
 	return nil
 }
 
-func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool) ([]*RPCRes, error) {
+func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch, isPoll bool) ([]*RPCRes, error) {
 	// we are concerned about network error rates, so we record 1 request independently of how many are in the batch
-	b.networkRequestsSlidingWindow.Incr()
+	// (we don't count polling towards error rates)
+	if !isPoll {
+		b.networkRequestsSlidingWindow.Incr()
+	}
+
+	incrementError := func() {
+		if !isPoll {
+			b.intermittentErrorsSlidingWindow.Incr()
+		}
+		RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
+	}
 
 	translatedReqs := make(map[string]*RPCReq, len(rpcReqs))
 	// translate consensus_getReceipts to receipts target
@@ -552,8 +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 {
-		b.intermittentErrorsSlidingWindow.Incr()
-		RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
+		incrementError()
 		return nil, wrapErr(err, "error creating backend request")
 	}
 
@@ -583,8 +596,7 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
 	start := time.Now()
 	httpRes, err := b.client.DoLimited(httpReq)
 	if err != nil {
-		b.intermittentErrorsSlidingWindow.Incr()
-		RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
+		incrementError()
 		return nil, wrapErr(err, "error in backend request")
 	}
 
@@ -602,8 +614,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 {
-		b.intermittentErrorsSlidingWindow.Incr()
-		RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
+		incrementError()
 		return nil, fmt.Errorf("response code %d", httpRes.StatusCode)
 	}
 
@@ -613,8 +624,7 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
 		return nil, ErrBackendResponseTooLarge
 	}
 	if err != nil {
-		b.intermittentErrorsSlidingWindow.Incr()
-		RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
+		incrementError()
 		return nil, wrapErr(err, "error reading response body")
 	}
 
@@ -629,21 +639,17 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
 		}
 	} else {
 		if err := json.Unmarshal(resB, &rpcRes); err != nil {
+			incrementError()
 			// 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()
-				RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
 				return nil, ErrBackendUnexpectedJSONRPC
 			}
-			b.intermittentErrorsSlidingWindow.Incr()
-			RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
 			return nil, ErrBackendBadResponse
 		}
 	}
 
 	if len(rpcReqs) != len(rpcRes) {
-		b.intermittentErrorsSlidingWindow.Incr()
-		RecordBackendNetworkErrorRateSlidingWindow(b, b.ErrorRate())
+		incrementError()
 		return nil, ErrBackendUnexpectedJSONRPC
 	}
 

proxyd/consensus_poller.go

@@ -610,7 +610,7 @@ func (cp *ConsensusPoller) Reset() {
 // 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
-	err = be.ForwardRPC(ctx, &rpcRes, "67", "eth_getBlockByNumber", block, false)
+	err = be.forwardRPC(ctx, true, &rpcRes, "67", "eth_getBlockByNumber", block, false)
 	if err != nil {
 		return 0, "", err
 	}

proxyd/example.config.toml

@@ -94,6 +94,8 @@ backends = ["infura"]
 # max_block_range = 10000
 # Whether to make rate limits block-range aware (e.g. eth_getLogs request for 10k blocks = 10k tokens)
 # rate_limit_range = true
+# Polling interval for the nonconsensus poller (set to negative to disable polling, e.g "-1s")
+# nonconsensus_poller_interval = "1s"
 # Enable consensus awareness for backend group, making it act as a load balancer, default false
 # consensus_aware = true
 # Period in which the backend wont serve requests if banned, default 5m

proxyd/integration_tests/mock_backend_test.go

@@ -25,6 +25,7 @@ type MockBackend struct {
 	server   *httptest.Server
 	mtx      sync.RWMutex
 	requests []*RecordedRequest
+	polls    []*RecordedRequest
 }
 
 func SingleResponseHandler(code int, response string) http.HandlerFunc {
@@ -222,14 +223,15 @@ func (m *MockBackend) Close() {
 
 func (m *MockBackend) SetHandler(handler http.Handler) {
 	m.mtx.Lock()
+	defer m.mtx.Unlock()
 	m.handler = handler
-	m.mtx.Unlock()
 }
 
 func (m *MockBackend) Reset() {
 	m.mtx.Lock()
+	defer m.mtx.Unlock()
 	m.requests = nil
-	m.mtx.Unlock()
+	m.polls = nil
 }
 
 func (m *MockBackend) Requests() []*RecordedRequest {
@@ -240,21 +242,36 @@ func (m *MockBackend) Requests() []*RecordedRequest {
 	return out
 }
 
+func (m *MockBackend) Polls() []*RecordedRequest {
+	m.mtx.RLock()
+	defer m.mtx.RUnlock()
+	out := make([]*RecordedRequest, len(m.polls))
+	copy(out, m.polls)
+	return out
+}
+
 func (m *MockBackend) wrappedHandler(w http.ResponseWriter, r *http.Request) {
 	m.mtx.Lock()
+	defer m.mtx.Unlock()
 	body, err := io.ReadAll(r.Body)
 	if err != nil {
 		panic(err)
 	}
+	var req proxyd.RPCReq
+	_ = json.Unmarshal(body, &req)
 	clone := r.Clone(context.Background())
 	clone.Body = io.NopCloser(bytes.NewReader(body))
-	m.requests = append(m.requests, &RecordedRequest{
+	rr := &RecordedRequest{
 		Method:  r.Method,
 		Headers: r.Header.Clone(),
 		Body:    body,
-	})
+	}
+	if string(req.ID) == proxyd.PollerRequestId {
+		m.polls = append(m.polls, rr)
+	} else {
+		m.requests = append(m.requests, rr)
+	}
 	m.handler.ServeHTTP(w, clone)
-	m.mtx.Unlock()
 }
 
 type MockWSBackend struct {
@@ -308,8 +325,8 @@ func (m *MockWSBackend) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 	}()
 	m.connsMu.Lock()
+	defer m.connsMu.Unlock()
 	m.conns = append(m.conns, conn)
-	m.connsMu.Unlock()
 }
 
 func (m *MockWSBackend) URL() string {
@@ -320,8 +337,8 @@ func (m *MockWSBackend) Close() {
 	m.server.Close()
 
 	m.connsMu.Lock()
+	defer m.connsMu.Unlock()
 	for _, conn := range m.conns {
 		conn.Close()
 	}
-	m.connsMu.Unlock()
 }

proxyd/integration_tests/testdata/multicall.toml

@@ -21,6 +21,7 @@ rpc_url = "$NODE3_URL"
 [backend_groups.node]
 backends = ["node1", "node2", "node3"]
 routing_strategy = "multicall"
+nonconsensus_poller_interval = "-1s"
 
 [rpc_method_mappings]
 eth_call = "node"

proxyd/integration_tests/testdata/retries.toml

@@ -13,6 +13,7 @@ ws_url = "$GOOD_BACKEND_RPC_URL"
 [backend_groups]
 [backend_groups.main]
 backends = ["good"]
+nonconsensus_poller_interval = "-1s"
 
 [rpc_method_mappings]
-eth_chainId = "main"
+eth_chainId = "main"

proxyd/nonconsensus_poller.go

@@ -11,6 +11,8 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 )
 
+const PollerRequestId = "901134"
+
 func NewNonconsensusPoller(bg *BackendGroup, opts ...NonconsensusOpt) *NonconsensusPoller {
 	ctx, cancel := context.WithCancel(context.Background())
 	np := &NonconsensusPoller{
@@ -24,9 +26,11 @@ func NewNonconsensusPoller(bg *BackendGroup, opts ...NonconsensusOpt) *Nonconsen
 	for _, opt := range opts {
 		opt(np)
 	}
-	for _, backend := range bg.Backends {
-		np.wg.Add(1)
-		go np.poll(backend)
+	if np.interval > 0 {
+		for _, backend := range bg.Backends {
+			np.wg.Add(1)
+			go np.poll(backend)
+		}
 	}
 	return np
 }
@@ -106,7 +110,7 @@ func (p *NonconsensusPoller) update(be *Backend, label string, ptr *uint64) {
 
 func (p *NonconsensusPoller) fetchBlock(ctx context.Context, be *Backend, block string) (uint64, error) {
 	var rpcRes RPCRes
-	err := be.ForwardRPC(ctx, &rpcRes, "68", "eth_getBlockByNumber", block, false)
+	err := be.forwardRPC(ctx, true, &rpcRes, PollerRequestId, "eth_getBlockByNumber", block, false)
 	if err != nil {
 		return 0, err
 	}

proxyd/proxyd.go

@@ -487,7 +487,7 @@ func Start(config *Config) (*Server, func(), error) {
 		} else {
 			opts := make([]NonconsensusOpt, 0)
 
-			if bgcfg.NonconsensusPollerInterval > 0 {
+			if bgcfg.NonconsensusPollerInterval != 0 {
 				opts = append(opts, WithPollingInterval(time.Duration(bgcfg.NonconsensusPollerInterval)))
 			}