Merge pull request #1 from wille/batch

- Support request batching
- endpoint.headers setting

Author: wille

internal/http.go

@@ -25,7 +25,7 @@ var defaultClient = &http.Client{
 	},
 }
 
-func ProxyHTTP(ctx context.Context, endpoint *Endpoint, req *rpc.Request, timing *servertiming.Header) (*rpc.Response, *Provider, error) {
+func ProxyHTTP(ctx context.Context, endpoint *Endpoint, req *rpc.BatchRequest, timing *servertiming.Header) (*rpc.BatchResponse, *Provider, error) {
 	providers := endpoint.GetActiveProviders()
 
 	for _, provider := range providers {
@@ -66,6 +66,13 @@ func SendHTTPRequest(ctx context.Context, provider *Provider, body []byte) ([]by
 	}
 
 	req.Header = make(http.Header)
+
+	if provider.Headers != nil {
+		for k, v := range provider.Headers {
+			req.Header.Set(k, v)
+		}
+	}
+
 	req.Header.Set("User-Agent", UserAgent)
 	req.Header.Set("Content-Type", "application/json; charset=utf-8")
 
@@ -93,15 +100,15 @@ func SendHTTPRequest(ctx context.Context, provider *Provider, body []byte) ([]by
 	return b, nil
 }
 
-func SendHTTPRPCRequest(ctx context.Context, p *Provider, rpcreq *rpc.Request) (*rpc.Response, error) {
-	req := rpc.SerializeRequest(rpcreq)
+func SendHTTPRPCRequest(ctx context.Context, p *Provider, req *rpc.BatchRequest) (*rpc.BatchResponse, error) {
+	body := rpc.SerializeBatchRequest(req)
 
-	b, err := SendHTTPRequest(ctx, p, req)
+	b, err := SendHTTPRequest(ctx, p, body)
 	if err != nil {
 		return nil, err
 	}
 
-	response, err := rpc.DecodeResponse(b)
+	response, err := rpc.DecodeBatchResponse(b)
 	if err != nil {
 		return nil, fmt.Errorf("bad response: %w, raw: %s", err, string(b))
 	}
@@ -127,19 +134,26 @@ func IncomingHttpHandler(ctx context.Context, endpoint *Endpoint, w http.Respons
 		return
 	}
 
-	rpcReq, err := rpc.DecodeRequest(body)
+	req, err := rpc.DecodeBatchRequest(body)
 	if err != nil {
-		log.Error("http: bad request", "error", err, "msg", rpc.FormatRawBody(string(body)))
+		log.Error("http: bad request", "error", err, "body", rpc.FormatRawBody(string(body)))
 		http.Error(w, "bad request", http.StatusBadRequest)
 		return
 	}
 
-	log = log.With("rpc_id", rpc.GetRequestIDString(rpcReq.ID), "method", rpcReq.Method)
+	if req.IsBatch {
+		rpc.BatchIDCounter++
+		log = log.With("batch_id", rpc.BatchIDCounter, "batch_size", len(req.Requests))
+	} else {
+		log = log.With("rpc_id", req.Requests[0].GetID(), "method", req.Requests[0].Method)
+	}
 
-	res, provider, err := ProxyHTTP(ctx, endpoint, rpcReq, timing)
+	res, provider, err := ProxyHTTP(ctx, endpoint, req, timing)
 
 	if err != nil {
-		metrics.RecordRequest(endpoint.Name, provider.Name, "http", rpcReq.Method, time.Since(start).Seconds(), true)
+		for _, req := range req.Requests {
+			metrics.RecordFailedRequest(endpoint.Name, provider.Name, "http", req.Method)
+		}
 
 		if err == ErrNoProvidersAvailable {
 			log.Error("no providers available")
@@ -154,9 +168,33 @@ func IncomingHttpHandler(ctx context.Context, endpoint *Endpoint, w http.Respons
 
 	log = log.With("provider", provider.Name, "request_time", time.Since(start))
 
-	log.Debug("request")
+	for i, res := range req.Requests {
+		if req.IsBatch {
+			log.Debug("request", "batch_index", i, "rpc_id", res.GetID(), "method", res.Method)
+		} else {
+			// id and method is already set for this single request
+			log.Debug("request")
+		}
+	}
 
-	metrics.RecordRequest(endpoint.Name, provider.Name, "http", rpcReq.Method, time.Since(start).Seconds(), res.IsError())
+	for i, res := range res.Responses {
+		method := req.Requests[i].Method
+
+		if res.IsError() {
+			log.Error("error", "error", res.Error)
+			metrics.RecordFailedRequest(endpoint.Name, provider.Name, "http", method)
+		} else {
+			metrics.RecordRequest(endpoint.Name, provider.Name, "http", method, time.Since(start).Seconds())
+		}
+	}
+
+	if req.IsBatch {
+		if len(res.Responses) != len(req.Requests) {
+			log.Error("batch response size mismatch", "request_size", len(req.Requests), "response_size", len(res.Responses))
+			http.Error(w, "batch response size mismatch", http.StatusInternalServerError)
+			return
+		}
+	}
 
 	if !endpoint.Public {
 		w.Header().Set("X-Provider", provider.Name)
@@ -168,7 +206,7 @@ func IncomingHttpHandler(ctx context.Context, endpoint *Endpoint, w http.Respons
 
 	w.Header().Set("Content-Type", "application/json; charset=utf-8")
 
-	_, err = w.Write(rpc.SerializeResponse(res))
+	_, err = w.Write(rpc.SerializeBatchResponse(res))
 	if err != nil {
 		log.Error("error writing body", "error", err)
 		return

internal/metrics/metrics.go

@@ -74,14 +74,15 @@ func MetricsHandler() http.Handler {
 }
 
 // RecordRequest records metrics for a request
-func RecordRequest(endpoint, provider, transport, method string, duration float64, failed bool) {
+func RecordRequest(endpoint, provider, transport, method string, duration float64) {
 	requestsTotal.WithLabelValues(endpoint, provider, transport, method).Inc()
-	if failed {
-		failedRequestsTotal.WithLabelValues(endpoint, provider, transport, method).Inc()
-	}
 	requestDuration.WithLabelValues(endpoint, provider, transport, method).Observe(duration)
 }
 
+func RecordFailedRequest(endpoint, provider, transport, method string) {
+	failedRequestsTotal.WithLabelValues(endpoint, provider, transport, method).Inc()
+}
+
 func RecordOpenConnection(endpoint, provider string) {
 	openConnections.WithLabelValues(endpoint, provider, "ws").Inc()
 	totalConnections.WithLabelValues(endpoint, provider, "ws").Inc()

internal/provider.go

@@ -23,6 +23,8 @@ type Provider struct {
 	Http         string `yaml:"http"`
 	Ws           string `yaml:"ws"`
 
+	Headers map[string]string `yaml:"headers,omitempty"`
+
 	// The timeout for the provider. Use Endpoint.GetTimeout() to get the actual timeout
 	Timeout time.Duration `yaml:"timeout,omitempty"`
 
@@ -130,17 +132,17 @@ func (e *Provider) Healthcheck(p *Endpoint) error {
 	ctx := context.Background()
 
 	fn := func(ctx context.Context, req *rpc.Request, errRpcError bool) (*rpc.Response, error) {
-		res, err := SendHTTPRPCRequest(ctx, e, req)
+		res, err := SendHTTPRPCRequest(ctx, e, rpc.NewBatchRequest(req))
 		if err != nil {
 			return nil, err
 		}
 
-		if errRpcError && res.IsError() {
-			_, err := res.GetError()
+		if errRpcError && res.Responses[0].IsError() {
+			_, err := res.Responses[0].GetError()
 			return nil, err
 		}
 
-		return res, nil
+		return res.Responses[0], nil
 	}
 
 	switch p.Kind {

internal/provider_test.go

@@ -70,7 +70,10 @@ func TestRateLimitWithRetryAfter(t *testing.T) {
 	}
 
 	// Send a request that will trigger rate limiting
-	_, _, err := ProxyHTTP(context.Background(), provider, rpc.NewRequest("1", "eth_blockNumber", []interface{}{}), &servertiming.Header{})
+	req := rpc.NewBatchRequest(
+		rpc.NewRequest("1", "eth_blockNumber", []interface{}{}),
+	)
+	_, _, err := ProxyHTTP(context.Background(), provider, req, &servertiming.Header{})
 
 	// Verify the error and retry time was set correctly
 	assert.Error(t, err)
@@ -126,7 +129,10 @@ func TestSlowProvider(t *testing.T) {
 	timing := &servertiming.Header{}
 
 	// Attempt to proxy a request - it should use the fast provider because the first one is too slow
-	resp, endpoint, err := ProxyHTTP(context.Background(), provider, rpc.NewRequest("1", "eth_blockNumber", []interface{}{}), timing)
+	req := rpc.NewBatchRequest(
+		rpc.NewRequest("1", "eth_blockNumber", []interface{}{}),
+	)
+	resp, endpoint, err := ProxyHTTP(context.Background(), provider, req, timing)
 
 	// Verify we got a response
 	assert.NoError(t, err)
@@ -166,7 +172,10 @@ func TestNonRespondingProvider(t *testing.T) {
 
 	// Attempt to use the non-responding provider first
 	timing := &servertiming.Header{}
-	resp, endpoint, err := ProxyHTTP(context.Background(), provider, rpc.NewRequest("1", "eth_blockNumber", []interface{}{}), timing)
+	req := rpc.NewBatchRequest(
+		rpc.NewRequest("1", "eth_blockNumber", []interface{}{}),
+	)
+	resp, endpoint, err := ProxyHTTP(context.Background(), provider, req, timing)
 
 	// Verify we got a response from the working provider
 	assert.NoError(t, err)

internal/rpc/batch.go

@@ -0,0 +1,150 @@
+package rpc
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+type BatchResponse struct {
+	Responses []*Response
+	IsBatch   bool
+}
+
+var _ json.Unmarshaler = &BatchResponse{}
+var _ json.Marshaler = &BatchResponse{}
+
+func NewBatchResponse(res ...*Response) *BatchResponse {
+	return &BatchResponse{
+		Responses: res,
+		IsBatch:   len(res) > 1,
+	}
+}
+
+// UnmarshalJSON for a batch request supports decoding a single request as well
+func (r *BatchResponse) UnmarshalJSON(b []byte) error {
+	switch b[0] {
+	case '[':
+		var res []*Response
+		err := json.Unmarshal(b, &res)
+		if err != nil {
+			return err
+		}
+		r.Responses = res
+		r.IsBatch = true
+		return nil
+	case '{':
+		var res Response
+		err := json.Unmarshal(b, &res)
+		if err != nil {
+			return err
+		}
+		r.Responses = []*Response{&res}
+		r.IsBatch = false
+		return nil
+	}
+
+	return fmt.Errorf("invalid request: %s", FormatRawBody(string(b)))
+}
+
+// MarshalJSON will encode a single (non batched) request to a single object or multiple requests into an array
+func (r BatchResponse) MarshalJSON() ([]byte, error) {
+	if len(r.Responses) == 0 {
+		return nil, fmt.Errorf("empty batch response")
+	}
+
+	// If the batch is just one single request then unbatch it
+	if !r.IsBatch {
+		return json.Marshal(r.Responses[0])
+	}
+
+	return json.Marshal(r.Responses)
+}
+
+type BatchRequest struct {
+	Requests []*Request
+	IsBatch  bool
+}
+
+var _ json.Unmarshaler = &BatchRequest{}
+var _ json.Marshaler = &BatchRequest{}
+
+func NewBatchRequest(req ...*Request) *BatchRequest {
+	return &BatchRequest{
+		Requests: req,
+		IsBatch:  len(req) > 1,
+	}
+}
+
+// UnmarshalJSON for a batch request supports decoding a single request as well
+func (r *BatchRequest) UnmarshalJSON(b []byte) error {
+	switch b[0] {
+	case '[':
+		var req []*Request
+		err := json.Unmarshal(b, &req)
+		if err != nil {
+			return err
+		}
+		r.Requests = req
+		r.IsBatch = true
+		return nil
+	case '{':
+		var req Request
+		err := json.Unmarshal(b, &req)
+		if err != nil {
+			return err
+		}
+		r.Requests = []*Request{&req}
+		r.IsBatch = false
+		return nil
+	}
+
+	return fmt.Errorf("invalid request: %s", FormatRawBody(string(b)))
+}
+
+// MarshalJSON will encode a single (non batched) request to a single object or multiple requests into an array
+func (r BatchRequest) MarshalJSON() ([]byte, error) {
+	if len(r.Requests) == 0 {
+		return nil, fmt.Errorf("empty batch request")
+	}
+
+	// If the batch is just one single request then unbatch it
+	if !r.IsBatch {
+		return json.Marshal(r.Requests[0])
+	}
+
+	return json.Marshal(r.Requests)
+}
+
+func DecodeBatchRequest(b []byte) (*BatchRequest, error) {
+	var batch BatchRequest
+	err := json.Unmarshal(b, &batch)
+	if err != nil {
+		return nil, err
+	}
+	return &batch, nil
+}
+
+func DecodeBatchResponse(b []byte) (*BatchResponse, error) {
+	var batch BatchResponse
+	err := json.Unmarshal(b, &batch)
+	if err != nil {
+		return nil, err
+	}
+	return &batch, nil
+}
+
+func SerializeBatchRequest(req *BatchRequest) []byte {
+	b, err := json.Marshal(req)
+	if err != nil {
+		panic(err)
+	}
+	return b
+}
+
+func SerializeBatchResponse(res *BatchResponse) []byte {
+	b, err := json.Marshal(res)
+	if err != nil {
+		panic(err)
+	}
+	return b
+}