Merge pull request #73 from kaleido-io/tx-cache

add LRU cache for transaction info

Makefile

@@ -36,7 +36,7 @@ $(eval $(call makemock, $$(FF_SIGNER_PATH),   Backend,   rpcbackendmocks))
 $(eval $(call makemock, $$(FFTM_PATH),        Manager,   fftmmocks))
 
 firefly-evmconnect: ${GOFILES}
-		$(VGO) build -o ./firefly-evmconnect -ldflags "-X main.buildDate=`date -u +\"%Y-%m-%dT%H:%M:%SZ\"` -X main.buildVersion=$(BUILD_VERSION)" -tags=prod -tags=prod -v ./evmconnect 
+		$(VGO) build -o ./firefly-evmconnect -ldflags "-X main.buildDate=`date -u +\"%Y-%m-%dT%H:%M:%SZ\"` -X main.buildVersion=$(BUILD_VERSION)" -tags=prod -tags=prod -v ./evmconnect
 go-mod-tidy: .ALWAYS
 		$(VGO) mod tidy
 build: firefly-evmconnect

config.md

@@ -50,7 +50,6 @@
 |Key|Description|Type|Default Value|
 |---|-----------|----|-------------|
 |blockCacheSize|Maximum of blocks to hold in the block info cache|`int`|`250`
-|blockCacheTTL|Time to live for the block info cache|[`time.Duration`](https://pkg.go.dev/time#Duration)|`5m`
 |blockPollingInterval|Interval for polling to check for new blocks|[`time.Duration`](https://pkg.go.dev/time#Duration)|`1s`
 |connectionTimeout|The maximum amount of time that a connection is allowed to remain with no data transmitted|[`time.Duration`](https://pkg.go.dev/time#Duration)|`30s`
 |dataFormat|Configure the JSON data format for query output and events|map,flat_array,self_describing|`map`
@@ -62,6 +61,7 @@
 |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`
 |tlsHandshakeTimeout|The maximum amount of time to wait for a successful TLS handshake|[`time.Duration`](https://pkg.go.dev/time#Duration)|`10s`
+|txCacheSize|Maximum of transactions to hold in the transaction info cache|`int`|`250`
 |url|URL of JSON/RPC endpoint for the Ethereum node/gateway|string|`<nil>`
 
 ## connector.auth

internal/ethereum/config.go

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -26,7 +26,6 @@ const (
 	ConfigDataFormat            = "dataFormat"
 	BlockPollingInterval        = "blockPollingInterval"
 	BlockCacheSize              = "blockCacheSize"
-	BlockCacheTTL               = "blockCacheTTL"
 	EventsCatchupPageSize       = "events.catchupPageSize"
 	EventsCatchupThreshold      = "events.catchupThreshold"
 	EventsCheckpointBlockGap    = "events.checkpointBlockGap"
@@ -35,6 +34,7 @@ const (
 	RetryInitDelay              = "retry.initialDelay"
 	RetryMaxDelay               = "retry.maxDelay"
 	RetryFactor                 = "retry.factor"
+	TxCacheSize                 = "txCacheSize"
 )
 
 const (
@@ -53,7 +53,6 @@ const (
 func InitConfig(conf config.Section) {
 	ffresty.InitConfig(conf)
 	conf.AddKnownKey(BlockCacheSize, 250)
-	conf.AddKnownKey(BlockCacheTTL, "5m")
 	conf.AddKnownKey(BlockPollingInterval, "1s")
 	conf.AddKnownKey(ConfigDataFormat, "map")
 	conf.AddKnownKey(ConfigGasEstimationFactor, DefaultGasEstimationFactor)
@@ -65,4 +64,5 @@ func InitConfig(conf config.Section) {
 	conf.AddKnownKey(RetryFactor, DefaultRetryDelayFactor)
 	conf.AddKnownKey(RetryInitDelay, DefaultRetryInitDelay)
 	conf.AddKnownKey(RetryMaxDelay, DefaultRetryMaxDelay)
+	conf.AddKnownKey(TxCacheSize, 250)
 }

internal/ethereum/ethereum.go

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -51,6 +51,7 @@ type ethConnector struct {
 	mux          sync.Mutex
 	eventStreams map[fftypes.UUID]*eventStream
 	blockCache   *lru.Cache
+	txCache      *lru.Cache
 }
 
 func NewEthereumConnector(ctx context.Context, conf config.Section) (cc ffcapi.API, err error) {
@@ -73,7 +74,12 @@ func NewEthereumConnector(ctx context.Context, conf config.Section) (cc ffcapi.A
 	}
 	c.blockCache, err = lru.New(conf.GetInt(BlockCacheSize))
 	if err != nil {
-		return nil, i18n.WrapError(ctx, err, msgs.MsgCacheInitFail)
+		return nil, i18n.WrapError(ctx, err, msgs.MsgCacheInitFail, "block")
+	}
+
+	c.txCache, err = lru.New(conf.GetInt(TxCacheSize))
+	if err != nil {
+		return nil, i18n.WrapError(ctx, err, msgs.MsgCacheInitFail, "transaction")
 	}
 
 	if conf.GetString(ffresty.HTTPConfigURL) == "" {

internal/ethereum/ethereum_test.go

@@ -106,4 +106,8 @@ func TestConnectorInit(t *testing.T) {
 	cc, err = NewEthereumConnector(context.Background(), conf)
 	assert.Regexp(t, "FF23040", err)
 
+	conf.Set(BlockCacheSize, "1")
+	conf.Set(TxCacheSize, "-1")
+	cc, err = NewEthereumConnector(context.Background(), conf)
+	assert.Regexp(t, "FF23040", err)
 }

internal/ethereum/event_listener_test.go

@@ -294,9 +294,13 @@ func TestFilterEnrichEthLogMethodInputsOk(t *testing.T) {
 			From:  ethtypes.MustNewAddress("0x3968ef051b422d3d1cdc182a88bba8dd922e6fa4"),
 			Input: ethtypes.MustNewHexBytes0xPrefix("0xa9059cbb000000000000000000000000d0f2f5103fd050739a9fb567251bc460cc24d09100000000000000000000000000000000000000000000000000000000000003e8"),
 		}
-	})
+	}).Once() // 1 cache miss and hit
 
-	ev, ok, err := l.filterEnrichEthLog(context.Background(), l.config.filters[0], sampleTransferLog())
+	ev, ok, err := l.filterEnrichEthLog(context.Background(), l.config.filters[0], sampleTransferLog()) // cache miss
+	assert.True(t, ok)
+	assert.NoError(t, err)
+
+	ev, ok, err = l.filterEnrichEthLog(context.Background(), l.config.filters[0], sampleTransferLog()) // cache hit
 	assert.True(t, ok)
 	assert.NoError(t, err)
 	ei := ev.Event.Info.(*eventInfo)

internal/ethereum/get_receipt.go

@@ -75,10 +75,17 @@ type txInfoJSONRPC struct {
 
 func (c *ethConnector) getTransactionInfo(ctx context.Context, hash ethtypes.HexBytes0xPrefix) (*txInfoJSONRPC, error) {
 	var txInfo *txInfoJSONRPC
+	cached, ok := c.txCache.Get(hash.String())
+	if ok {
+		return cached.(*txInfoJSONRPC), nil
+	}
+
 	rpcErr := c.backend.CallRPC(ctx, &txInfo, "eth_getTransactionByHash", hash)
 	var err error
 	if rpcErr != nil {
 		err = rpcErr.Error()
+	} else {
+		c.txCache.Add(hash.String(), txInfo)
 	}
 	return txInfo, err
 }

internal/msgs/en_config_descriptions.go

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -31,11 +31,11 @@ var (
 	ConfigEthereumDataFormat          = ffc("config.connector.dataFormat", "Configure the JSON data format for query output and events", "map,flat_array,self_describing")
 	ConfigEthereumGasEstimationFactor = ffc("config.connector.gasEstimationFactor", "The factor to apply to the gas estimation to determine the gas limit", "float")
 	ConfigBlockCacheSize              = ffc("config.connector.blockCacheSize", "Maximum of blocks to hold in the block info cache", i18n.IntType)
-	ConfigBlockCacheTTL               = ffc("config.connector.blockCacheTTL", "Time to live for the block info cache", i18n.TimeDurationType)
 	ConfigBlockPollingInterval        = ffc("config.connector.blockPollingInterval", "Interval for polling to check for new blocks", i18n.TimeDurationType)
 	ConfigEventsBlockTimestamps       = ffc("config.connector.events.blockTimestamps", "Whether to include the block timestamps in the event information", i18n.BooleanType)
 	ConfigEventsCatchupPageSize       = ffc("config.connector.events.catchupPageSize", "Number of blocks to query per poll when catching up to the head of the blockchain", i18n.IntType)
 	ConfigEventsCatchupThreshold      = ffc("config.connector.events.catchupThreshold", "How many blocks behind the chain head an event stream or listener must be on startup, to enter catchup mode", i18n.IntType)
 	ConfigEventsCheckpointBlockGap    = ffc("config.connector.events.checkpointBlockGap", "The number of blocks at the head of the chain that should be considered unstable (could be dropped from the canonical chain after a re-org). Unless events with a full set of confirmations are detected, the restart checkpoint will this many blocks behind the chain head.", i18n.IntType)
 	ConfigEventsFilterPollingInterval = ffc("config.connector.events.filterPollingInterval", "The interval between polling calls to a filter, when checking for newly arrived events", i18n.TimeDurationType)
+	ConfigTxCacheSize                 = ffc("config.connector.txCacheSize", "Maximum of transactions to hold in the transaction info cache", i18n.IntType)
 )

internal/msgs/en_error_messages.go

@@ -55,7 +55,7 @@ var (
 	MsgMissingEventInFilter      = ffe("FF23037", "Each filter must have an 'event' child containing the ABI definition of the event")
 	MsgListenerAlreadyStarted    = ffe("FF23038", "Listener already started: %s")
 	MsgInvalidCheckpoint         = ffe("FF23039", "Invalid checkpoint: %s")
-	MsgCacheInitFail             = ffe("FF23040", "Failed to initialize cache")
+	MsgCacheInitFail             = ffe("FF23040", "Failed to initialize %s cache")
 	MsgStreamNotStarted          = ffe("FF23041", "Event stream %s not started")
 	MsgStreamAlreadyStarted      = ffe("FF23042", "Event stream %s already started")
 	MsgListenerNotStarted        = ffe("FF23043", "Event listener %s not started in event stream %s")