Merge pull request #224 from hyperledger-firefly/ws-only-listener

Move all RPC calls to single client that chooses WS/HTTP

Author: Chengxuan

config.md

@@ -82,6 +82,7 @@
 |maxIdleConnsPerHost|The max number of idle connections, per unique hostname. Zero means net/http uses the default of only 2.|`int`|`100`
 |passthroughHeadersEnabled|Enable passing through the set of allowed HTTP request headers|`boolean`|`false`
 |requestTimeout|The maximum amount of time that a request is allowed to remain open|[`time.Duration`](https://pkg.go.dev/time#Duration)|`30s`
+|rpcRoutingMode|Which connection JSON/RPC calls are made on when a WebSocket is enabled. 'auto' routes node-sticky chain queries (blocks, logs, filters, receipts) to the WebSocket and stateless calls (submission, gas, balance) to the HTTP connection pool. 'ws' routes everything to the WebSocket. 'http' routes everything to the HTTP connection pool, leaving the WebSocket to carry only the newHeads subscription. 'legacy' reproduces the routing used before this setting existed, where only the block listener used the WebSocket|http,ws,auto,legacy|`auto`
 |tlsHandshakeTimeout|The maximum amount of time to wait for a successful TLS handshake|[`time.Duration`](https://pkg.go.dev/time#Duration)|`10s`
 |traceTXForRevertReason|Enable the use of transaction trace functions (e.g. debug_traceTransaction) to obtain transaction revert reasons. This can place a high load on the EVM client.|`boolean`|`false`
 |txCacheSize|Maximum of transactions to hold in the transaction info cache|`int`|`250`

internal/ethereum/config.go

@@ -19,6 +19,7 @@ package ethereum
 import (
 	"github.com/hyperledger-firefly/common/pkg/config"
 	"github.com/hyperledger-firefly/common/pkg/wsclient"
+	"github.com/hyperledger-firefly/evmconnect/pkg/ethrpc"
 	"github.com/hyperledger-firefly/transaction-manager/pkg/ffcapi"
 )
 
@@ -51,6 +52,7 @@ const (
 	HederaCompatibilityMode       = "hederaCompatibilityMode"
 	TraceTXForRevertReason        = "traceTXForRevertReason"
 	WebSocketsEnabled             = "ws.enabled"
+	RPCRoutingMode                = "rpcRoutingMode"
 	MaxAsyncBlockFetchConcurrency = "maxAsyncBlockFetchConcurrency"
 	UseGetBlockReceipts           = "useGetBlockReceipts"
 )
@@ -72,6 +74,7 @@ const (
 func InitConfig(conf config.Section) {
 	wsclient.InitConfig(conf)
 	conf.AddKnownKey(WebSocketsEnabled, false)
+	conf.AddKnownKey(RPCRoutingMode, string(ethrpc.RoutingModeAuto))
 	conf.AddKnownKey(BlockCacheSize, 250)
 	conf.AddKnownKey(ReceiptCacheEnabled, false)
 	conf.AddKnownKey(ReceiptCacheSize, 5000)

internal/ethereum/estimate_gas.go

@@ -70,7 +70,7 @@ func (c *ethConnector) gasEstimate(ctx context.Context, tx *ethsigner.Transactio
 
 	// Do the gas estimation
 	var gasEstimate ethtypes.HexInteger
-	rpcErr := c.backend.CallRPC(ctx, &gasEstimate, "eth_estimateGas", tx)
+	rpcErr := c.rpc.CallRPC(ctx, &gasEstimate, "eth_estimateGas", tx)
 	if rpcErr != nil {
 		if reason, revertErr := c.attemptProcessingRevertData(ctx, errors, rpcErr); revertErr != nil {
 			return nil, reason, revertErr

internal/ethereum/ethereum.go

@@ -38,13 +38,11 @@ import (
 	"github.com/hyperledger-firefly/evmconnect/pkg/ethrpc"
 	"github.com/hyperledger-firefly/signer/pkg/abi"
 	"github.com/hyperledger-firefly/signer/pkg/ethtypes"
-	"github.com/hyperledger-firefly/signer/pkg/rpcbackend"
 	"github.com/hyperledger-firefly/transaction-manager/pkg/ffcapi"
 )
 
 type ethConnector struct {
-	backend           rpcbackend.Backend
-	wsBackend         rpcbackend.WebSocketRPCClient
+	rpc               ethrpc.Client
 	chainTrackingMode ffcapi.ChainTrackingMode
 
 	serializer                 *abi.Serializer
@@ -68,18 +66,48 @@ type ethConnector struct {
 type Connector interface {
 	ffcapi.API
 
-	// RPC returns the http JSON/RPC client
-	RPC() rpcbackend.RPC
-
-	// WSRPC returns the websocket JSON/RPC client
-	// NOTE: websocket client will be nil if websockets are not enabled
-	WSRPC() rpcbackend.WebSocketRPCClient
+	// RPC returns the JSON/RPC client used for everything the connector does. It owns both
+	// the HTTP connection pool and the WebSocket (when enabled), and routes each call to one
+	// of them based on the method name and the configured routing mode.
+	RPC() ethrpc.Client
 
 	// Get the high level block listener functionality, which provides a view of the head of the chain
 	BlockListener() ethblocklistener.BlockListener
 }
 
+// NewEthereumConnector builds the JSON/RPC client from configuration, and the connector over it.
 func NewEthereumConnector(ctx context.Context, conf config.Section) (cc Connector, err error) {
+	if conf.GetString(ffresty.HTTPConfigURL) == "" {
+		return nil, i18n.NewError(ctx, msgs.MsgMissingBackendURL)
+	}
+
+	var wsConf *wsclient.WSConfig
+	if conf.GetBool(WebSocketsEnabled) {
+		if wsConf, err = wsclient.GenerateConfig(ctx, conf); err != nil {
+			return nil, err
+		}
+	}
+	httpConf, err := ffresty.GenerateConfig(ctx, conf)
+	if err != nil {
+		return nil, err
+	}
+
+	rpc, err := ethrpc.NewClient(ctx, &ethrpc.Config{
+		RoutingMode:           ethrpc.RoutingMode(conf.GetString(RPCRoutingMode)),
+		HTTP:                  httpConf,
+		WS:                    wsConf,
+		MaxConcurrentRequests: conf.GetInt64(MaxConcurrentRequests),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return NewEthereumConnectorWithRPC(ctx, conf, rpc)
+}
+
+// NewEthereumConnectorWithRPC builds the connector over a pre-constructed JSON/RPC client.
+// The caller owns closing the client - see ethConnector.WaitClosed.
+func NewEthereumConnectorWithRPC(ctx context.Context, conf config.Section, rpc ethrpc.Client) (cc Connector, err error) {
 
 	chainTrackingMode := ffcapi.ChainTrackingMode(conf.GetString(ChainTrackingMode))
 	if chainTrackingMode == "" {
@@ -90,6 +118,7 @@ func NewEthereumConnector(ctx context.Context, conf config.Section) (cc Connecto
 	}
 
 	c := &ethConnector{
+		rpc:                        rpc,
 		eventStreams:               make(map[fftypes.UUID]*eventStream),
 		catchupPageSize:            conf.GetInt64(EventsCatchupPageSize),
 		catchupThreshold:           conf.GetInt64(EventsCatchupThreshold),
@@ -115,41 +144,13 @@ func NewEthereumConnector(ctx context.Context, conf config.Section) (cc Connecto
 		return nil, i18n.WrapError(ctx, err, msgs.MsgCacheInitFail, "transaction")
 	}
 
-	if conf.GetString(ffresty.HTTPConfigURL) == "" {
-		return nil, i18n.NewError(ctx, msgs.MsgMissingBackendURL)
-	}
 	c.gasEstimationFactor = big.NewFloat(conf.GetFloat64(ConfigGasEstimationFactor))
 
 	c.catchupDownscaleRegex, err = regexp.Compile(conf.GetString(EventsCatchupDownscaleRegex))
 	if err != nil {
 		return nil, i18n.WrapError(ctx, err, msgs.MsgInvalidRegex, c.catchupDownscaleRegex)
 	}
 
-	var wsConf *wsclient.WSConfig
-	var httpConf *ffresty.Config
-	if conf.GetBool(WebSocketsEnabled) {
-		// If websockets are enabled, then they are used selectively (block listening/query)
-		// not as a full replacement for HTTP.
-		wsConf, err = wsclient.GenerateConfig(ctx, conf)
-	}
-
-	if err == nil {
-		httpConf, err = ffresty.GenerateConfig(ctx, conf)
-	}
-
-	if err != nil {
-		return nil, err
-	}
-
-	httpClient := ffresty.NewWithConfig(ctx, *httpConf)
-	c.backend = rpcbackend.NewRPCClientWithOption(httpClient, rpcbackend.RPCClientOptions{
-		MaxConcurrentRequest: conf.GetInt64(MaxConcurrentRequests),
-	})
-
-	if wsConf != nil {
-		c.wsBackend = rpcbackend.NewWSRPCClient(wsConf)
-	}
-
 	c.serializer = abi.NewSerializer().SetByteSerializer(abi.HexByteSerializer0xPrefix)
 	switch conf.Get(ConfigDataFormat) {
 	case "map":
@@ -169,7 +170,7 @@ func NewEthereumConnector(ctx context.Context, conf config.Section) (cc Connecto
 		return name
 	})
 
-	if c.blockListener, err = ethblocklistener.NewBlockListenerSupplyBackend(ctx, c.retry.Retry, &ethblocklistener.BlockListenerConfig{
+	if c.blockListener, err = ethblocklistener.NewBlockListener(ctx, c.retry.Retry, &ethblocklistener.BlockListenerConfig{
 		BlockPollingInterval:          conf.GetDuration(BlockPollingInterval),
 		MonitoredHeadLength:           int(c.checkpointBlockGap),
 		HederaCompatibilityMode:       conf.GetBool(HederaCompatibilityMode),
@@ -179,19 +180,15 @@ func NewEthereumConnector(ctx context.Context, conf config.Section) (cc Connecto
 		MaxAsyncBlockFetchConcurrency: conf.GetInt(MaxAsyncBlockFetchConcurrency),
 		UseGetBlockReceipts:           conf.GetBool(UseGetBlockReceipts),
 		ChainTrackingMode:             c.chainTrackingMode,
-	}, c.backend, c.wsBackend); err != nil {
+	}, c.rpc); err != nil {
 		return nil, err
 	}
 
 	return c, nil
 }
 
-func (c *ethConnector) RPC() rpcbackend.RPC {
-	return c.backend
-}
-
-func (c *ethConnector) WSRPC() rpcbackend.WebSocketRPCClient {
-	return c.wsBackend
+func (c *ethConnector) RPC() ethrpc.Client {
+	return c.rpc
 }
 
 func (c *ethConnector) BlockListener() ethblocklistener.BlockListener {
@@ -206,6 +203,9 @@ func (c *ethConnector) WaitClosed() {
 	for _, s := range c.eventStreams {
 		<-s.streamLoopDone
 	}
+	if c.rpc != nil {
+		c.rpc.Close()
+	}
 }
 
 func withDeprecatedConfFallback[T any](conf config.Section, getter func(string) T, deprecatedKey, newKey string) T {

internal/ethereum/ethereum_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/hyperledger-firefly/evmconnect/pkg/ethblocklistener"
 	"github.com/hyperledger-firefly/evmconnect/pkg/ethrpc"
 	"github.com/hyperledger-firefly/signer/pkg/abi"
+	"github.com/hyperledger-firefly/signer/pkg/rpcbackend"
 	"github.com/hyperledger-firefly/transaction-manager/pkg/ffcapi"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
@@ -40,6 +41,12 @@ import (
 
 func strPtr(s string) *string { return &s }
 
+func utRPC(t *testing.T, mRPC rpcbackend.RPC) ethrpc.Client {
+	rpc, err := ethrpc.NewClientWithBackends(t.Context(), ethrpc.RoutingModeHTTP, mRPC, nil)
+	require.NoError(t, err)
+	return rpc
+}
+
 func newTestConnector(t *testing.T, confSetup ...func(conf config.Section)) (context.Context, *ethConnector, *rpcbackendmocks.Backend, func()) {
 	ctx, c, mRPC, done := newTestConnectorWithNoBlockerFilterDefaultMocks(t, confSetup...)
 
@@ -64,13 +71,13 @@ func newTestConnectorWithNoBlockerFilterDefaultMocks(t *testing.T, confSetup ...
 		fn(conf)
 	}
 	ctx, done := context.WithCancel(context.Background())
-	cc, err := NewEthereumConnector(ctx, conf)
+	rpc, err := ethrpc.NewClientWithBackends(ctx, ethrpc.RoutingModeHTTP, mRPC, nil)
+	require.NoError(t, err)
+	cc, err := NewEthereumConnectorWithRPC(ctx, conf, rpc)
 	assert.NoError(t, err)
 	assert.NotNil(t, cc.RPC())
 
 	c := cc.(*ethConnector)
-	c.backend = mRPC
-	cc.BlockListener().UTSetBackend(mRPC)
 	return ctx, c, mRPC, func() {
 		done()
 		mRPC.AssertExpectations(t)
@@ -87,12 +94,17 @@ func TestConnectorInit(t *testing.T) {
 	cc, err := NewEthereumConnector(context.Background(), conf)
 	assert.Regexp(t, "FF23025", err)
 
+	conf.Set(ffresty.HTTPConfigURL, "http://localhost:8545")
 	conf.Set(ChainTrackingMode, "wrong")
 	_, err = NewEthereumConnector(context.Background(), conf)
 	assert.Regexp(t, "FF23069.*wrong", err)
 
+	conf.Set(RPCRoutingMode, "wrong")
+	_, err = NewEthereumConnector(context.Background(), conf)
+	assert.Regexp(t, "FF23075.*wrong", err)
+
+	conf.Set(RPCRoutingMode, ethrpc.RoutingModeAuto)
 	conf.Set(ChainTrackingMode, "")
-	conf.Set(ffresty.HTTPConfigURL, "http://localhost:8545")
 	conf.Set(WebSocketsEnabled, true)
 	conf.Set(EventsCatchupThreshold, 1)
 	conf.Set(EventsCatchupPageSize, 500)
@@ -237,7 +249,7 @@ func TestRetryDefaultsFor429(t *testing.T) {
 	cc, err := NewEthereumConnector(ctx, conf)
 	assert.NoError(t, err)
 	assert.NotNil(t, cc.RPC())
-	assert.Nil(t, cc.WSRPC())
+	assert.False(t, cc.RPC().HasWebSocket())
 	defer done()
 
 	// Start a simple HTTP server that always replies with 429 Too Many Requests

internal/ethereum/event_listener.go

@@ -240,8 +240,8 @@ func (l *listener) listenerCatchupLoop() {
 				}
 			} else {
 				log.L(ctx).Errorf("Failed to query block range fromBlock=%d toBlock=%d: %s", fromBlock, toBlock, err)
-				failCount++
 			}
+			failCount++ // for exponential backoff calculation
 			continue
 		}
 		log.L(ctx).Infof("Listener catchup fromBlock=%d toBlock=%d events=%d", fromBlock, toBlock, len(events))

internal/ethereum/event_stream.go

@@ -304,7 +304,7 @@ func (es *eventStream) leadGroupCatchup() bool {
 func (es *eventStream) uninstallFilter(filter *string) {
 	if *filter != "" {
 		var res bool
-		if err := es.c.backend.CallRPC(es.ctx, &res, "eth_uninstallFilter", filter); err != nil {
+		if err := es.c.rpc.CallRPC(es.ctx, &res, "eth_uninstallFilter", filter); err != nil {
 			log.L(es.ctx).Warnf("Error uninstalling filter '%v': %s", filter, err.Message)
 		} else {
 			log.L(es.ctx).Debugf("Uninstalled filter '%v': %t", filter, res)
@@ -368,7 +368,7 @@ func (es *eventStream) leadGroupSteadyState() bool {
 				}
 
 				// Create the new filter
-				err := es.c.backend.CallRPC(es.ctx, &filter, "eth_newFilter", &ethrpc.LogFilterJSONRPC{
+				err := es.c.rpc.CallRPC(es.ctx, &filter, "eth_newFilter", &ethrpc.LogFilterJSONRPC{
 					FromBlock: ethtypes.NewHexInteger64(fromBlock),
 					Topics: [][]ethtypes.HexBytes0xPrefix{
 						ag.signatureSet,
@@ -384,7 +384,7 @@ func (es *eventStream) leadGroupSteadyState() bool {
 			}
 			// Get the next batch of logs
 			var ethLogs []*ethrpc.LogJSONRPC
-			rpcErr := es.c.backend.CallRPC(es.ctx, &ethLogs, filterRPCMethodToUse, filter)
+			rpcErr := es.c.rpc.CallRPC(es.ctx, &ethLogs, filterRPCMethodToUse, filter)
 			// If we fail to query we just retry - setting filter to nil if not found
 			if rpcErr != nil {
 				if etherrors.MapError(etherrors.FilterRPCMethods, rpcErr.Error()) == ffcapi.ErrorReasonNotFound {
@@ -596,7 +596,7 @@ func (es *eventStream) getBlockRangeEvents(ctx context.Context, ag *aggregatedLi
 		logFilterJSONRPCReq.Address = []*ethtypes.Address0xHex{ag.listeners[0].config.filters[0].Address}
 	}
 
-	rpcErr := es.c.backend.CallRPC(ctx, &ethLogs, "eth_getLogs", logFilterJSONRPCReq)
+	rpcErr := es.c.rpc.CallRPC(ctx, &ethLogs, "eth_getLogs", logFilterJSONRPCReq)
 	if rpcErr != nil {
 		return nil, rpcErr.Error()
 	}

internal/ethereum/event_stream_test.go

@@ -996,7 +996,7 @@ func TestStreamCleanupFilterOK(t *testing.T) {
 	es := &eventStream{
 		ctx: context.Background(),
 		c: &ethConnector{
-			backend: mRPC,
+			rpc: utRPC(t, mRPC),
 		},
 	}
 
@@ -1015,7 +1015,7 @@ func TestStreamCleanupFilterFailLog(t *testing.T) {
 	es := &eventStream{
 		ctx: context.Background(),
 		c: &ethConnector{
-			backend: mRPC,
+			rpc: utRPC(t, mRPC),
 		},
 	}
 

internal/ethereum/exec_query.go

@@ -91,7 +91,7 @@ func (c *ethConnector) callTransaction(ctx context.Context, tx *ethsigner.Transa
 	if blockNumber != nil {
 		blockNumberStr = *blockNumber
 	}
-	rpcErr := c.backend.CallRPC(ctx, &outputData, "eth_call", tx, blockNumberStr)
+	rpcErr := c.rpc.CallRPC(ctx, &outputData, "eth_call", tx, blockNumberStr)
 	if rpcErr != nil {
 		if reason, revertErr := c.attemptProcessingRevertData(ctx, errors, rpcErr); revertErr != nil {
 			return nil, reason, revertErr

internal/ethereum/get_address_balance.go

@@ -31,7 +31,7 @@ func (c *ethConnector) AddressBalance(ctx context.Context, req *ffcapi.AddressBa
 	if blockTag == "" {
 		blockTag = "latest"
 	}
-	rpcErr := c.backend.CallRPC(ctx, &addressBalance, "eth_getBalance", req.Address, blockTag)
+	rpcErr := c.rpc.CallRPC(ctx, &addressBalance, "eth_getBalance", req.Address, blockTag)
 	if rpcErr != nil {
 		return nil, "", rpcErr.Error()
 	}