CCIP-1716 Adding retention to filters used by LogPoller

core/chains/evm/config/toml/defaults/fallback.toml

@@ -6,7 +6,8 @@ FinalityTagEnabled = false
 LogBackfillBatchSize = 1000
 LogPollInterval = '15s'
 LogKeepBlocksDepth = 100000
-LogPrunePageSize = 0
+# CCIP uses paging when removing logs to avoid pushing too much pressure on the database
+LogPrunePageSize = 10000
 BackupLogPollerBlockDelay = 100
 MinContractPayment = '.00001 link'
 MinIncomingConfirmations = 3

core/config/docs/chains-evm.toml

@@ -56,7 +56,8 @@ LogPollInterval = '15s' # Default
 LogKeepBlocksDepth = 100000 # Default
 # **ADVANCED**
 # LogPrunePageSize defines size of the page for pruning logs. Controls how many logs/blocks (at most) are deleted in a single prune tick. Default value 0 means no paging, delete everything at once.
-LogPrunePageSize = 0 # Default
+# CCIP uses paging when removing logs to avoid pushing too much presssure on the database so this value should not be set to 0.
+LogPrunePageSize = 10000 # Default
 # **ADVANCED**
 # BackupLogPollerBlockDelay works in conjunction with Feature.LogPoller. Controls the block delay of Backup LogPoller, affecting how far behind the latest finalized block it starts and how often it runs.
 # BackupLogPollerDelay=0 will disable Backup LogPoller (_not recommended for production environment_).

core/services/chainlink/config_test.go

@@ -530,7 +530,7 @@ func TestConfig_Marshal(t *testing.T) {
 				LogBackfillBatchSize:      ptr[uint32](17),
 				LogPollInterval:           &minute,
 				LogKeepBlocksDepth:        ptr[uint32](100000),
-				LogPrunePageSize:          ptr[uint32](0),
+				LogPrunePageSize:          ptr[uint32](10000),
 				BackupLogPollerBlockDelay: ptr[uint64](532),
 				MinContractPayment:        commonassets.NewLinkFromJuels(math.MaxInt64),
 				MinIncomingConfirmations:  ptr[uint32](13),
@@ -928,7 +928,7 @@ LinkContractAddress = '0x538aAaB4ea120b2bC2fe5D296852D948F07D849e'
 LogBackfillBatchSize = 17
 LogPollInterval = '1m0s'
 LogKeepBlocksDepth = 100000
-LogPrunePageSize = 0
+LogPrunePageSize = 10000
 BackupLogPollerBlockDelay = 532
 MinIncomingConfirmations = 13
 MinContractPayment = '9.223372036854775807 link'

core/services/chainlink/testdata/config-full.toml

@@ -254,7 +254,7 @@ LinkContractAddress = '0x538aAaB4ea120b2bC2fe5D296852D948F07D849e'
 LogBackfillBatchSize = 17
 LogPollInterval = '1m0s'
 LogKeepBlocksDepth = 100000
-LogPrunePageSize = 0
+LogPrunePageSize = 10000
 BackupLogPollerBlockDelay = 532
 MinIncomingConfirmations = 13
 MinContractPayment = '9.223372036854775807 link'

core/services/chainlink/testdata/config-multi-chain-effective.toml

@@ -241,7 +241,7 @@ LinkContractAddress = '0x514910771AF9Ca656af840dff83E8264EcF986CA'
 LogBackfillBatchSize = 1000
 LogPollInterval = '15s'
 LogKeepBlocksDepth = 100000
-LogPrunePageSize = 0
+LogPrunePageSize = 10000
 BackupLogPollerBlockDelay = 100
 MinIncomingConfirmations = 3
 MinContractPayment = '0.1 link'
@@ -331,7 +331,7 @@ LinkContractAddress = '0xa36085F69e2889c224210F603D836748e7dC0088'
 LogBackfillBatchSize = 1000
 LogPollInterval = '15s'
 LogKeepBlocksDepth = 100000
-LogPrunePageSize = 0
+LogPrunePageSize = 10000
 BackupLogPollerBlockDelay = 100
 MinIncomingConfirmations = 3
 MinContractPayment = '0.1 link'
@@ -416,7 +416,7 @@ LinkContractAddress = '0xb0897686c545045aFc77CF20eC7A532E3120E0F1'
 LogBackfillBatchSize = 1000
 LogPollInterval = '1s'
 LogKeepBlocksDepth = 100000
-LogPrunePageSize = 0
+LogPrunePageSize = 10000
 BackupLogPollerBlockDelay = 100
 MinIncomingConfirmations = 5
 MinContractPayment = '0.00001 link'

core/services/ocr2/plugins/ccip/internal/ccipdata/reader.go

@@ -2,6 +2,7 @@ package ccipdata
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/ethereum/go-ethereum/core/types"
 
@@ -18,6 +19,24 @@ const (
 	V1_5_0 = "1.5.0-dev"
 )
 
+const (
+	// CommitExecLogsRetention defines the duration for which logs critical for Commit/Exec plugins processing are retained.
+	// Although Exec relies on permissionlessExecThreshold which is lower than 24hours for picking eligible CommitRoots,
+	// Commit still can reach to older logs because it filters them by sequence numbers. For instance, in case of RMN curse on chain,
+	// we might have logs waiting in OnRamp to be committed first. When outage takes days we still would
+	// be able to bring back processing without replaying any logs from chain. You can read that param as
+	// "how long CCIP can be down and still be able to process all the messages after getting back to life".
+	// Breaching this threshold would require replaying chain using LogPoller from the beginning of the outage.
+	CommitExecLogsRetention = 30 * 24 * time.Hour // 30 days
+	// CacheEvictionLogsRetention defines the duration for which logs used for caching on-chain data are kept.
+	// Restarting node clears the cache entirely and rebuilds it from scratch by fetching data from chain,
+	// so we don't need to keep these logs for very long. All events relying on cache.NewLogpollerEventsBased should use this retention.
+	CacheEvictionLogsRetention = 7 * 24 * time.Hour // 7 days
+	// PriceUpdatesLogsRetention defines the duration for which logs with price updates are kept.
+	// These logs are emitted whenever the token price or gas price is updated and Commit scans very small time windows (e.g. 2 hours)
+	PriceUpdatesLogsRetention = 1 * 24 * time.Hour // 1 day
+)
+
 type Event[T any] struct {
 	Data T
 	cciptypes.TxMeta

core/services/ocr2/plugins/ccip/internal/ccipdata/usdc_reader.go

@@ -96,6 +96,7 @@ func NewUSDCReader(lggr logger.Logger, jobID string, transmitter common.Address,
 			Name:      logpoller.FilterName(MESSAGE_SENT_FILTER_NAME, jobID, transmitter.Hex()),
 			EventSigs: []common.Hash{eventSig},
 			Addresses: []common.Address{transmitter},
+			Retention: CommitExecLogsRetention,
 		},
 		transmitterAddress: transmitter,
 	}

core/services/ocr2/plugins/ccip/internal/ccipdata/v1_0_0/commit_store.go

@@ -391,6 +391,7 @@ func NewCommitStore(lggr logger.Logger, addr common.Address, ec client.Client, l
 			Name:      logpoller.FilterName(EXEC_REPORT_ACCEPTS, addr.String()),
 			EventSigs: []common.Hash{eventSig},
 			Addresses: []common.Address{addr},
+			Retention: ccipdata.CommitExecLogsRetention,
 		},
 	}
 	return &CommitStore{

core/services/ocr2/plugins/ccip/internal/ccipdata/v1_0_0/offramp.go

@@ -603,10 +603,6 @@ func (o *OffRamp) DecodeExecutionReport(report []byte) (cciptypes.ExecReport, er
 	return DecodeExecReport(o.ExecutionReportArgs, report)
 }
 
-func (o *OffRamp) TokenEvents() []common.Hash {
-	return offRamp_poolAddedPoolRemovedEvents
-}
-
 func (o *OffRamp) RegisterFilters(qopts ...pg.QOpt) error {
 	return logpollerutil.RegisterLpFilters(o.lp, o.filters, qopts...)
 }
@@ -624,16 +620,19 @@ func NewOffRamp(lggr logger.Logger, addr common.Address, ec client.Client, lp lo
 			Name:      logpoller.FilterName(EXEC_EXECUTION_STATE_CHANGES, addr.String()),
 			EventSigs: []common.Hash{ExecutionStateChangedEvent},
 			Addresses: []common.Address{addr},
+			Retention: ccipdata.CommitExecLogsRetention,
 		},
 		{
 			Name:      logpoller.FilterName(EXEC_TOKEN_POOL_ADDED, addr.String()),
 			EventSigs: []common.Hash{PoolAddedEvent},
 			Addresses: []common.Address{addr},
+			Retention: ccipdata.CacheEvictionLogsRetention,
 		},
 		{
 			Name:      logpoller.FilterName(EXEC_TOKEN_POOL_REMOVED, addr.String()),
 			EventSigs: []common.Hash{PoolRemovedEvent},
 			Addresses: []common.Address{addr},
+			Retention: ccipdata.CacheEvictionLogsRetention,
 		},
 	}
 

core/services/ocr2/plugins/ccip/internal/ccipdata/v1_0_0/onramp.go

@@ -59,11 +59,13 @@ func NewOnRamp(lggr logger.Logger, sourceSelector, destSelector uint64, onRampAd
 			Name:      logpoller.FilterName(ccipdata.COMMIT_CCIP_SENDS, onRampAddress),
 			EventSigs: []common.Hash{eventSig},
 			Addresses: []common.Address{onRampAddress},
+			Retention: ccipdata.CommitExecLogsRetention,
 		},
 		{
 			Name:      logpoller.FilterName(ccipdata.CONFIG_CHANGED, onRampAddress),
 			EventSigs: []common.Hash{configSetEventSig},
 			Addresses: []common.Address{onRampAddress},
+			Retention: ccipdata.CacheEvictionLogsRetention,
 		},
 	}
 	cachedStaticConfig := cache.OnceCtxFunction[evm_2_evm_onramp_1_0_0.EVM2EVMOnRampStaticConfig](func(ctx context.Context) (evm_2_evm_onramp_1_0_0.EVM2EVMOnRampStaticConfig, error) {

core/services/ocr2/plugins/ccip/internal/ccipdata/v1_0_0/price_registry.go

@@ -63,16 +63,19 @@ func NewPriceRegistry(lggr logger.Logger, priceRegistryAddr common.Address, lp l
 			Name:      logpoller.FilterName(ccipdata.COMMIT_PRICE_UPDATES, priceRegistryAddr.String()),
 			EventSigs: []common.Hash{UsdPerUnitGasUpdated, usdPerTokenUpdated},
 			Addresses: []common.Address{priceRegistryAddr},
+			Retention: ccipdata.PriceUpdatesLogsRetention,
 		},
 		{
 			Name:      logpoller.FilterName(ccipdata.FEE_TOKEN_ADDED, priceRegistryAddr.String()),
 			EventSigs: []common.Hash{feeTokenAdded},
 			Addresses: []common.Address{priceRegistryAddr},
+			Retention: ccipdata.CacheEvictionLogsRetention,
 		},
 		{
 			Name:      logpoller.FilterName(ccipdata.FEE_TOKEN_REMOVED, priceRegistryAddr.String()),
 			EventSigs: []common.Hash{feeTokenRemoved},
 			Addresses: []common.Address{priceRegistryAddr},
+			Retention: ccipdata.CacheEvictionLogsRetention,
 		}}
 	if registerFilters {
 		err = logpollerutil.RegisterLpFilters(lp, filters)

core/services/ocr2/plugins/ccip/internal/ccipdata/v1_2_0/commit_store.go

@@ -405,6 +405,7 @@ func NewCommitStore(lggr logger.Logger, addr common.Address, ec client.Client, l
 			Name:      logpoller.FilterName(v1_0_0.EXEC_REPORT_ACCEPTS, addr.String()),
 			EventSigs: []common.Hash{eventSig},
 			Addresses: []common.Address{addr},
+			Retention: ccipdata.CommitExecLogsRetention,
 		},
 	}
 

core/services/ocr2/plugins/ccip/internal/ccipdata/v1_2_0/onramp.go

@@ -77,11 +77,13 @@ func NewOnRamp(lggr logger.Logger, sourceSelector, destSelector uint64, onRampAd
 			Name:      logpoller.FilterName(ccipdata.COMMIT_CCIP_SENDS, onRampAddress),
 			EventSigs: []common.Hash{CCIPSendRequestEventSig},
 			Addresses: []common.Address{onRampAddress},
+			Retention: ccipdata.CommitExecLogsRetention,
 		},
 		{
 			Name:      logpoller.FilterName(ccipdata.CONFIG_CHANGED, onRampAddress),
 			EventSigs: []common.Hash{ConfigSetEventSig},
 			Addresses: []common.Address{onRampAddress},
+			Retention: ccipdata.CacheEvictionLogsRetention,
 		},
 	}
 	cachedStaticConfig := cache.OnceCtxFunction[evm_2_evm_onramp_1_2_0.EVM2EVMOnRampStaticConfig](func(ctx context.Context) (evm_2_evm_onramp_1_2_0.EVM2EVMOnRampStaticConfig, error) {

core/services/ocr2/plugins/ccip/internal/ccipdata/v1_5_0/onramp.go

@@ -78,11 +78,13 @@ func NewOnRamp(lggr logger.Logger, sourceSelector, destSelector uint64, onRampAd
 			Name:      logpoller.FilterName(ccipdata.COMMIT_CCIP_SENDS, onRampAddress),
 			EventSigs: []common.Hash{CCIPSendRequestEventSig},
 			Addresses: []common.Address{onRampAddress},
+			Retention: ccipdata.CommitExecLogsRetention,
 		},
 		{
 			Name:      logpoller.FilterName(ccipdata.CONFIG_CHANGED, onRampAddress),
 			EventSigs: []common.Hash{ConfigSetEventSig},
 			Addresses: []common.Address{onRampAddress},
+			Retention: ccipdata.CacheEvictionLogsRetention,
 		},
 	}
 	cachedStaticConfig := cache.OnceCtxFunction[evm_2_evm_onramp.EVM2EVMOnRampStaticConfig](func(ctx context.Context) (evm_2_evm_onramp.EVM2EVMOnRampStaticConfig, error) {

core/web/resolver/testdata/config-full.toml

@@ -254,7 +254,7 @@ LinkContractAddress = '0x538aAaB4ea120b2bC2fe5D296852D948F07D849e'
 LogBackfillBatchSize = 17
 LogPollInterval = '1m0s'
 LogKeepBlocksDepth = 100000
-LogPrunePageSize = 0
+LogPrunePageSize = 10000
 BackupLogPollerBlockDelay = 100
 MinIncomingConfirmations = 13
 MinContractPayment = '9.223372036854775807 link'

core/web/resolver/testdata/config-multi-chain-effective.toml

@@ -241,7 +241,7 @@ LinkContractAddress = '0x514910771AF9Ca656af840dff83E8264EcF986CA'
 LogBackfillBatchSize = 1000
 LogPollInterval = '15s'
 LogKeepBlocksDepth = 100000
-LogPrunePageSize = 0
+LogPrunePageSize = 10000
 BackupLogPollerBlockDelay = 100
 MinIncomingConfirmations = 3
 MinContractPayment = '0.1 link'
@@ -331,7 +331,7 @@ LinkContractAddress = '0xa36085F69e2889c224210F603D836748e7dC0088'
 LogBackfillBatchSize = 1000
 LogPollInterval = '15s'
 LogKeepBlocksDepth = 100000
-LogPrunePageSize = 0
+LogPrunePageSize = 10000
 BackupLogPollerBlockDelay = 100
 MinIncomingConfirmations = 3
 MinContractPayment = '0.1 link'
@@ -416,7 +416,7 @@ LinkContractAddress = '0xb0897686c545045aFc77CF20eC7A532E3120E0F1'
 LogBackfillBatchSize = 1000
 LogPollInterval = '1s'
 LogKeepBlocksDepth = 100000
-LogPrunePageSize = 0
+LogPrunePageSize = 10000
 BackupLogPollerBlockDelay = 100
 MinIncomingConfirmations = 5
 MinContractPayment = '0.00001 link'

docs/CHANGELOG_CCIP.md

@@ -13,7 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - Added optional `PriceReportingDisabled` to signal lane should not report gas and token prices.
 - `OffRamp` offchain config changed:
   - `MaxGasPrice` and `DestMaxGasPrice` will be ignored if specified. DestMaxGasPrice will be read from PriceMax in chain TOML.
-
+- All LogPoller's filters have now retention defined. This means that the LogPoller will remove logs older than the retention defined per filter. 
+  Additionally, logs are now removed from the database in batches, defined by `LogPrunePageSize` in the toml config. Default value for all chains is 10_000.
 
 ## 1.4.0 - 2024-02-16