Add tests to jsonrpc, to assure our jsonrpc clients behave as expected.

Makefile

@@ -34,6 +34,7 @@ test:
 	CGO_ENABLED=1 go test -timeout=$(TEST_TIMEOUT) -race -bench=. -benchmem -cover ./...
 
 gen-mocks: bin/moq ./client/jsonrpc/ ./client/duneapi/
+	./bin/moq -pkg jsonrpc_mock -out ./mocks/jsonrpc/httpclient.go ./client/jsonrpc HTTPClient
 	./bin/moq -pkg jsonrpc_mock -out ./mocks/jsonrpc/rpcnode.go ./client/jsonrpc BlockchainClient
 	./bin/moq -pkg duneapi_mock -out ./mocks/duneapi/client.go ./client/duneapi BlockchainIngester
 

client/jsonrpc/arbitrum_nitro.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"log/slog"
 	"time"
 
 	"github.com/duneanalytics/blockchain-ingester/models"
@@ -18,14 +17,6 @@ type ArbitrumNitroClient struct {
 
 var _ BlockchainClient = &ArbitrumNitroClient{}
 
-func NewArbitrumNitroClient(log *slog.Logger, cfg Config) (*ArbitrumNitroClient, error) {
-	rpcClient, err := newClient(log.With("module", "jsonrpc"), cfg)
-	if err != nil {
-		return nil, err
-	}
-	return &ArbitrumNitroClient{*rpcClient}, nil
-}
-
 // BlockByNumber returns the block with the given blockNumber.
 // it uses 3 different methods to get the block:
 //  1. eth_getBlockByNumber

client/jsonrpc/arbitrum_nitro_test.go

@@ -0,0 +1,82 @@
+package jsonrpc_test
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/duneanalytics/blockchain-ingester/models"
+	"github.com/stretchr/testify/require"
+)
+
+func TestArbitrumNitroBasic(t *testing.T) {
+	getBlockByNumberResponse := readFileForTest(
+		"testdata/arbitrumnitro-DEGEN-block-0x16870e9-eth_getBlockByNumber.json")
+	debugtraceBlockByNumberResponse := readFileForTest(
+		"testdata/arbitrumnitro-DEGEN-block-0x16870e9-debug_traceBlockByNumber.json")
+	tx0ReceiptResponse := readFileForTest(
+		"testdata/arbitrumnitro-DEGEN-block-0x16870e9-eth_getTransactionReceipt-0x0.json")
+	tx1ReceiptResponse := readFileForTest(
+		"testdata/arbitrumnitro-DEGEN-block-0x16870e9-eth_getTransactionReceipt-0x1.json")
+
+	var expectedPayload bytes.Buffer
+	expectedPayload.Write(getBlockByNumberResponse.Bytes())
+	expectedPayload.Write(debugtraceBlockByNumberResponse.Bytes())
+	expectedPayload.Write(tx0ReceiptResponse.Bytes())
+	expectedPayload.Write(tx1ReceiptResponse.Bytes())
+	expectedPayloadBytes := expectedPayload.Bytes()
+
+	tx0Hash := "0x19ee83020d4dad7e96dbb2c01ce2441e75717ee038a022fc6a3b61300b1b801c"
+	tx1Hash := "0x4e805891b568698f8419f8e162d70ed9675e42a32e4972cbeb7f78d7fd51de76"
+	blockNumberHex := "0x16870e9"
+	blockNumber := int64(23621865)
+	httpClientMock := MockHTTPRequests(
+		[]MockedRequest{
+			{
+				Req: jsonRPCRequest{
+					Method: "eth_getBlockByNumber",
+					Params: []interface{}{blockNumberHex, true},
+				},
+				Resp: jsonRPCResponse{
+					Body: getBlockByNumberResponse,
+				},
+			},
+			{
+				Req: jsonRPCRequest{
+					Method: "debug_traceBlockByNumber",
+					Params: []interface{}{blockNumberHex, map[string]string{"tracer": "callTracer"}},
+				},
+				Resp: jsonRPCResponse{
+					Body: debugtraceBlockByNumberResponse,
+				},
+			},
+			{
+				Req: jsonRPCRequest{
+					Method: "eth_getTransactionReceipt",
+					Params: []interface{}{tx0Hash},
+				},
+				Resp: jsonRPCResponse{
+					Body: tx0ReceiptResponse,
+				},
+			},
+			{
+				Req: jsonRPCRequest{
+					Method: "eth_getTransactionReceipt",
+					Params: []interface{}{tx1Hash},
+				},
+				Resp: jsonRPCResponse{
+					Body: tx1ReceiptResponse,
+				},
+			},
+		})
+
+	opstack, err := NewTestRPCClient(httpClientMock, models.ArbitrumNitro)
+	require.NoError(t, err)
+
+	block, err := opstack.BlockByNumber(context.Background(), blockNumber)
+	require.NoError(t, err)
+	require.NotNil(t, block)
+	require.Equal(t, blockNumber, block.BlockNumber)
+	require.False(t, block.Errored())
+	require.Equal(t, expectedPayloadBytes, block.Payload)
+}

client/jsonrpc/client.go

@@ -29,44 +29,36 @@ const (
 )
 
 type rpcClient struct {
-	client      *retryablehttp.Client
-	cfg         Config
-	log         *slog.Logger
 	bufPool     *sync.Pool
+	cfg         Config
+	client      HTTPClient
 	httpHeaders map[string]string
+	log         *slog.Logger
 	wrkPool     *ants.Pool
 }
 
-func NewClient(logger *slog.Logger, cfg Config) (BlockchainClient, error) {
+func NewClient(log *slog.Logger, cfg Config) (BlockchainClient, error) {
+	// use the production http client w/ retries
+	return NewRPCClient(log, NewHTTPClient(log), cfg)
+}
+
+func NewRPCClient(log *slog.Logger, client HTTPClient, cfg Config) (BlockchainClient, error) {
+	rpcClient, err := newClient(log.With("module", "jsonrpc"), client, cfg)
+	if err != nil {
+		return nil, err
+	}
 	switch cfg.EVMStack {
 	case models.OpStack:
-		return NewOpStackClient(logger, cfg)
+		return &OpStackClient{*rpcClient}, nil
 	case models.ArbitrumNitro:
-		return NewArbitrumNitroClient(logger, cfg)
+		return &ArbitrumNitroClient{*rpcClient}, nil
 	default:
 		return nil, fmt.Errorf("unsupported EVM stack: %s", cfg.EVMStack)
 	}
 }
 
-func newClient(log *slog.Logger, cfg Config) (*rpcClient, error) { // revive:disable-line:unexported-return
-	client := retryablehttp.NewClient()
-	client.RetryMax = MaxRetries
-	client.Logger = log
-	checkRetry := func(ctx context.Context, resp *http.Response, err error) (bool, error) {
-		yes, err2 := retryablehttp.DefaultRetryPolicy(ctx, resp, err)
-		if yes {
-			if resp == nil {
-				log.Warn("Retrying request to RPC client", "error", err2)
-			} else {
-				log.Warn("Retrying request to RPC client", "statusCode", resp.Status, "error", err2)
-			}
-		}
-		return yes, err2
-	}
-	client.CheckRetry = checkRetry
-	client.Backoff = retryablehttp.LinearJitterBackoff
-	client.HTTPClient.Timeout = DefaultRequestTimeout
-
+func newClient(log *slog.Logger, client HTTPClient, cfg Config,
+) (*rpcClient, error) { // revive:disable-line:unexported-return
 	if cfg.TotalRPCConcurrency == 0 {
 		cfg.TotalRPCConcurrency = DefaultMaxRPCConcurrency
 	}
@@ -98,8 +90,7 @@ func newClient(log *slog.Logger, cfg Config) (*rpcClient, error) { // revive:dis
 
 func (c *rpcClient) LatestBlockNumber() (int64, error) {
 	buf := c.bufPool.Get().(*bytes.Buffer)
-	defer c.bufPool.Put(buf)
-	buf.Reset()
+	defer c.putBuffer(buf)
 
 	err := c.getResponseBody(context.Background(), "eth_blockNumber", []any{}, buf)
 	if err != nil {

client/jsonrpc/httpclient.go

@@ -0,0 +1,34 @@
+package jsonrpc
+
+import (
+	"context"
+	"log/slog"
+	"net/http"
+
+	"github.com/hashicorp/go-retryablehttp"
+)
+
+type HTTPClient interface {
+	Do(req *retryablehttp.Request) (*http.Response, error)
+}
+
+func NewHTTPClient(log *slog.Logger) *retryablehttp.Client {
+	client := retryablehttp.NewClient()
+	client.RetryMax = MaxRetries
+	client.Logger = log
+	checkRetry := func(ctx context.Context, resp *http.Response, err error) (bool, error) {
+		yes, err2 := retryablehttp.DefaultRetryPolicy(ctx, resp, err)
+		if yes {
+			if resp == nil {
+				log.Warn("Retrying request to RPC client", "error", err2)
+			} else {
+				log.Warn("Retrying request to RPC client", "statusCode", resp.Status, "error", err2)
+			}
+		}
+		return yes, err2
+	}
+	client.CheckRetry = checkRetry
+	client.Backoff = retryablehttp.LinearJitterBackoff
+	client.HTTPClient.Timeout = DefaultRequestTimeout
+	return client
+}

client/jsonrpc/httpclient_test.go

@@ -0,0 +1,112 @@
+package jsonrpc_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"log/slog"
+	"net/http"
+	"os"
+
+	"github.com/duneanalytics/blockchain-ingester/client/jsonrpc"
+	jsonrpc_mock "github.com/duneanalytics/blockchain-ingester/mocks/jsonrpc"
+	"github.com/duneanalytics/blockchain-ingester/models"
+	"github.com/hashicorp/go-retryablehttp"
+)
+
+type jsonRPCRequest struct {
+	Method      string        `json:"method"`
+	Params      []interface{} `json:"params"`
+	HTTPHeaders http.Header
+}
+
+type jsonRPCResponse struct {
+	Body        io.Reader
+	StatusCode  int    // optional, default to 200
+	ContentType string // optional, default to "application/json"
+}
+
+type MockedRequest struct {
+	Req  jsonRPCRequest
+	Resp jsonRPCResponse
+}
+
+func MockHTTPRequests(requests []MockedRequest) *jsonrpc_mock.HTTPClientMock {
+	// helper function to setup a mock http client with recorded request responses
+	// non-registered requests will return an error
+	return &jsonrpc_mock.HTTPClientMock{
+		DoFunc: func(req *retryablehttp.Request) (*http.Response, error) {
+			if req.Method != http.MethodPost {
+				return nil, fmt.Errorf("expected POST method, got %s", req.Method)
+			}
+			// we use httpretryable.Client, so we can't use req.Body directly
+			// we need to read the body and then reset it
+
+			body, err := req.BodyBytes()
+			if err != nil {
+				return nil, err
+			}
+			var jsonReq jsonRPCRequest
+			if err := json.Unmarshal(body, &jsonReq); err != nil {
+				return nil, err
+			}
+			jsonReqParams := fmt.Sprintf("%+v", jsonReq.Params)
+			// looking for a matching request
+			for _, r := range requests {
+				if r.Req.Method == jsonReq.Method {
+					// we do this because reflect.DeepEquals() Comparison fails on map[string]any != map[string]string
+					if jsonReqParams != fmt.Sprintf("%+v", r.Req.Params) {
+						continue
+					}
+					// this is a match, validate registered headers
+					for k, v := range r.Req.HTTPHeaders {
+						if req.Header.Get(k) != v[0] {
+							return nil, fmt.Errorf("expected header %s to be %s, got %s", k, v[0], req.Header.Get(k))
+						}
+					}
+					// all headers match, return the response
+					resp := &http.Response{
+						StatusCode: 200,
+						Body:       io.NopCloser(r.Resp.Body),
+						Header:     make(http.Header),
+					}
+					if r.Resp.StatusCode != 0 {
+						resp.StatusCode = r.Resp.StatusCode
+					}
+					resp.Header.Set("Content-Type", "application/json")
+					if r.Resp.ContentType != "" {
+						resp.Header.Set("Content-Type", r.Resp.ContentType)
+					}
+					return resp, nil
+				}
+			}
+			// for simplificy, we include a default response for eth_blockNumber with a valid response
+			if jsonReq.Method == "eth_blockNumber" {
+				resp := &http.Response{
+					StatusCode: 200,
+					Body:       io.NopCloser(bytes.NewReader([]byte(`{"jsonrpc":"2.0","id":1,"result":"0x7a549b"}`))),
+				}
+				return resp, nil
+			}
+			return nil, fmt.Errorf("no matching request found, req: %+v", jsonReq)
+		},
+	}
+}
+
+func NewTestLogger() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+func NewTestRPCClient(httpClient jsonrpc.HTTPClient, stack models.EVMStack) (jsonrpc.BlockchainClient, error) {
+	return jsonrpc.NewRPCClient(NewTestLogger(), httpClient, jsonrpc.Config{EVMStack: stack})
+}
+
+func readFileForTest(filename string) *bytes.Buffer {
+	data, err := os.ReadFile(filename)
+	if err != nil {
+		log.Panicf("Failed to read file: %v", err)
+	}
+	return bytes.NewBuffer(data)
+}

client/jsonrpc/opstack.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"log/slog"
 	"time"
 
 	"github.com/duneanalytics/blockchain-ingester/models"
@@ -17,14 +16,6 @@ type OpStackClient struct {
 
 var _ BlockchainClient = &OpStackClient{}
 
-func NewOpStackClient(log *slog.Logger, cfg Config) (*OpStackClient, error) {
-	rpcClient, err := newClient(log.With("module", "jsonrpc"), cfg)
-	if err != nil {
-		return nil, err
-	}
-	return &OpStackClient{*rpcClient}, nil
-}
-
 // BlockByNumber returns the block with the given blockNumber.
 // it uses 3 different methods to get the block:
 // 1. eth_getBlockByNumber