CCIP-5449 Backgroung caching of config data (#713)

Author: 0xnogo

commit/plugin.go

@@ -548,6 +548,7 @@ func (p *Plugin) Close() error {
 		p.tokenPriceProcessor,
 		p.chainFeeProcessor,
 		p.discoveryProcessor,
+		p.ccipReader,
 	}
 
 	// Chains without RMN don't initialize the RMNHomeReader

execute/plugin.go

@@ -7,6 +7,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"io"
 	"slices"
 	"sort"
 	"time"
@@ -20,6 +21,8 @@ import (
 	"github.com/smartcontractkit/libocr/quorumhelper"
 	libocrtypes "github.com/smartcontractkit/libocr/ragep2p/types"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
+
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives"
@@ -744,7 +747,14 @@ func (p *Plugin) ShouldTransmitAcceptedReport(
 }
 
 func (p *Plugin) Close() error {
-	return p.tokenDataObserver.Close()
+	p.lggr.Infow("closing exec plugin")
+
+	closeable := []io.Closer{
+		p.tokenDataObserver,
+		p.ccipReader,
+	}
+
+	return services.CloseAll(closeable...)
 }
 
 func (p *Plugin) supportedChains(id commontypes.OracleID) (mapset.Set[cciptypes.ChainSelector], error) {

execute/plugin_test.go

@@ -632,8 +632,18 @@ func Test_getPendingReportsForExecution(t *testing.T) {
 }
 
 func TestPlugin_Close(t *testing.T) {
-	p := &Plugin{tokenDataObserver: &tokendata.NoopTokenDataObserver{}}
+	lggr := logger.Test(t)
+	mockReader := readerpkg_mock.NewMockCCIPReader(t)
+
+	// Set up expectation for Close() method
+	mockReader.On("Close").Return(nil)
+	p := &Plugin{
+		lggr:              lggr,
+		tokenDataObserver: &tokendata.NoopTokenDataObserver{},
+		ccipReader:        mockReader}
+
 	require.NoError(t, p.Close())
+	mockReader.AssertExpectations(t)
 }
 
 func TestPlugin_Query(t *testing.T) {

internal/mocks/inmem/ccipreader_inmem.go

@@ -210,5 +210,12 @@ func (r InMemoryCCIPReader) GetOffRampSourceChainsConfig(ctx context.Context, ch
 	return nil, nil
 }
 
+// Close implements the reader.CCIPReader interface
+func (r InMemoryCCIPReader) Close() error {
+	// Since this is an in-memory implementation with no persistent connections
+	// or resources to clean up, we can simply return nil
+	return nil
+}
+
 // Interface compatibility check.
 var _ reader.CCIPReader = InMemoryCCIPReader{}

mocks/pkg/reader/ccip_reader.go

@@ -86,6 +86,14 @@ func newCCIPChainReaderInternal(
 		lggr.Errorw("failed to sync contracts", "err", err)
 	}
 
+	// After contracts are synced, start the background polling
+	lggr.Info("Starting config background polling")
+	if err := reader.configPoller.Start(ctx); err != nil {
+		// Log the error but don't fail - we can still function without background polling
+		// by fetching configs on demand
+		lggr.Errorw("failed to start config background polling", "err", err)
+	}
+
 	return reader
 }
 
@@ -96,6 +104,15 @@ func (r *ccipChainReader) WithExtendedContractReader(
 	return r
 }
 
+func (r *ccipChainReader) Close() error {
+	if err := r.configPoller.Close(); err != nil {
+		r.lggr.Warnw("Error closing config poller", "err", err)
+		// Continue with shutdown even if there's an error
+	}
+	r.lggr.Info("Stopped CCIP chain reader")
+	return nil
+}
+
 // ---------------------------------------------------
 // The following types are used to decode the events
 // but should be replaced by chain-reader modifiers and use the base cciptypes.CommitReport type.
@@ -2041,5 +2058,39 @@ func validateSendRequestedEvent(
 	return nil
 }
 
+// ccipReaderInternal defines the interface that ConfigPoller needs from the ccipChainReader
+// This allows for better encapsulation and easier testing through mocking
+type ccipReaderInternal interface {
+	// getDestChain returns the destination chain selector
+	getDestChain() cciptypes.ChainSelector
+
+	// getContractReader returns the contract reader for the specified chain
+	getContractReader(chain cciptypes.ChainSelector) (contractreader.Extended, bool)
+
+	// prepareBatchConfigRequests prepares the batch requests for fetching chain configuration
+	prepareBatchConfigRequests(chainSel cciptypes.ChainSelector) contractreader.ExtendedBatchGetLatestValuesRequest
+
+	// processConfigResults processes the batch results into a ChainConfigSnapshot
+	processConfigResults(
+		chainSel cciptypes.ChainSelector,
+		batchResult types.BatchGetLatestValuesResult) (ChainConfigSnapshot, error)
+
+	// fetchFreshSourceChainConfigs fetches source chain configurations from the specified destination chain
+	fetchFreshSourceChainConfigs(
+		ctx context.Context, destChain cciptypes.ChainSelector,
+		sourceChains []cciptypes.ChainSelector) (map[cciptypes.ChainSelector]SourceChainConfig, error)
+}
+
+// getDestChain returns the destination chain selector
+func (r *ccipChainReader) getDestChain() cciptypes.ChainSelector {
+	return r.destChain
+}
+
+// getContractReader returns the contract reader for the specified chain
+func (r *ccipChainReader) getContractReader(chain cciptypes.ChainSelector) (contractreader.Extended, bool) {
+	reader, exists := r.contractReaders[chain]
+	return reader, exists
+}
+
 // Interface compliance check
 var _ CCIPReader = (*ccipChainReader)(nil)

pkg/reader/ccip.go

@@ -271,4 +271,6 @@ type CCIPReader interface {
 	// If a config was not found it will be missing from the returned map.
 	GetOffRampSourceChainsConfig(ctx context.Context, sourceChains []cciptypes.ChainSelector,
 	) (map[cciptypes.ChainSelector]StaticSourceChainConfig, error)
+
+	Close() error
 }

pkg/reader/ccip_interface.go

@@ -90,6 +90,14 @@ func TestCCIPChainReader_getSourceChainsConfig(t *testing.T) {
 		}, nil, chainC, offrampAddress, mockAddrCodec,
 	)
 
+	// Add cleanup to ensure resources are released
+	t.Cleanup(func() {
+		err := ccipReader.Close()
+		if err != nil {
+			t.Logf("Error closing ccipReader: %v", err)
+		}
+	})
+
 	addrStr, err := mockAddrCodec.AddressBytesToString(offrampAddress, 111_111)
 	require.NoError(t, err)
 
@@ -814,6 +822,14 @@ func TestCCIPChainReader_getFeeQuoterTokenPriceUSD(t *testing.T) {
 		}, nil, chainC, offrampAddress, mockAddrCodec,
 	)
 
+	// Add cleanup to properly shut down the background polling
+	t.Cleanup(func() {
+		err := ccipReader.Close()
+		if err != nil {
+			t.Logf("Error closing ccipReader: %v", err)
+		}
+	})
+
 	feeQuoterAddressStr, err := mockAddrCodec.AddressBytesToString(feeQuoterAddress, 111_111)
 	require.NoError(t, err)
 	require.NoError(t, ccipReader.contractReaders[chainC].Bind(
@@ -850,6 +866,14 @@ func TestCCIPFeeComponents_HappyPath(t *testing.T) {
 		internal.NewMockAddressCodecHex(t),
 	)
 
+	// Add cleanup to ensure resources are released
+	t.Cleanup(func() {
+		err := ccipReader.Close()
+		if err != nil {
+			t.Logf("Error closing ccipReader: %v", err)
+		}
+	})
+
 	ctx := context.Background()
 	feeComponents := ccipReader.GetChainsFeeComponents(ctx, []cciptypes.ChainSelector{chainA, chainB, chainC})
 	assert.Len(t, feeComponents, 2)
@@ -879,6 +903,14 @@ func TestCCIPFeeComponents_NotFoundErrors(t *testing.T) {
 		internal.NewMockAddressCodecHex(t),
 	)
 
+	// Add cleanup to ensure resources are released
+	t.Cleanup(func() {
+		err := ccipReader.Close()
+		if err != nil {
+			t.Logf("Error closing ccipReader: %v", err)
+		}
+	})
+
 	ctx := context.Background()
 	_, err := ccipReader.GetDestChainFeeComponents(ctx)
 	require.Error(t, err)
@@ -1532,13 +1564,6 @@ func (m *mockConfigCache) GetChainConfig(
 	return args.Get(0).(ChainConfigSnapshot), args.Error(1)
 }
 
-func (m *mockConfigCache) RefreshChainConfig(
-	ctx context.Context,
-	chainSel cciptypes.ChainSelector) (ChainConfigSnapshot, error) {
-	args := m.Called(ctx, chainSel)
-	return args.Get(0).(ChainConfigSnapshot), args.Error(1)
-}
-
 func (m *mockConfigCache) GetOfframpSourceChainConfigs(
 	ctx context.Context,
 	destChain cciptypes.ChainSelector,
@@ -1547,10 +1572,26 @@ func (m *mockConfigCache) GetOfframpSourceChainConfigs(
 	return args.Get(0).(map[cciptypes.ChainSelector]StaticSourceChainConfig), args.Error(1)
 }
 
-func (m *mockConfigCache) RefreshSourceChainConfigs(
-	ctx context.Context,
-	destChain cciptypes.ChainSelector,
-	sourceChains []cciptypes.ChainSelector) (map[cciptypes.ChainSelector]StaticSourceChainConfig, error) {
-	args := m.Called(ctx, destChain, sourceChains)
-	return args.Get(0).(map[cciptypes.ChainSelector]StaticSourceChainConfig), args.Error(1)
+// Update Start method to accept context parameter
+func (m *mockConfigCache) Start(ctx context.Context) error {
+	return m.Called(ctx).Error(0)
+}
+
+func (m *mockConfigCache) Close() error {
+	return m.Called().Error(0)
+}
+
+// Implement HealthReport method for services.Service interface
+func (m *mockConfigCache) HealthReport() map[string]error {
+	args := m.Called()
+	return args.Get(0).(map[string]error)
+}
+
+// Implement Name method for the Service interface
+func (m *mockConfigCache) Name() string {
+	return m.Called().String(0)
+}
+
+func (m *mockConfigCache) Ready() error {
+	return m.Called().Error(0)
 }