Exec: commit root caching

execute/internal/cache/commit_root_cache.go

@@ -0,0 +1,122 @@
+package cache
+
+import (
+	"sync"
+	"time"
+
+	"github.com/patrickmn/go-cache"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+
+	"github.com/smartcontractkit/chainlink-ccip/pkg/types/ccipocr3"
+)
+
+const (
+	// EvictionGracePeriod defines how long after the messageVisibilityInterval a root is still kept in the cache
+	EvictionGracePeriod = 1 * time.Hour
+	// CleanupInterval defines how often roots cache is scanned to evict stale roots
+	CleanupInterval = 30 * time.Minute
+)
+
+// CommitsRootsCache keeps track of commit roots (and messages?) that are eligible for execution.
+//
+// This cache is used when:
+//   - a commit root is fully executed and we want to skip it in the future. This would be called
+//     in the GetCommitReports phase after checking if each SeqNr is executed (or if the Txm state is finalized).
+//   - remember the oldest pending commit root to limit the database scan to only the unfinalized part of the chain.
+type CommitsRootsCache interface {
+	CanExecute(source ccipocr3.ChainSelector, merkleRoot ccipocr3.Bytes32) bool
+	MarkAsExecuted(source ccipocr3.ChainSelector, merkleRoot ccipocr3.Bytes32)
+	Snooze(source ccipocr3.ChainSelector, merkleRoot ccipocr3.Bytes32)
+}
+
+func NewCommitRootsCache(
+	lggr logger.Logger,
+	messageVisibilityInterval time.Duration,
+	rootSnoozeTime time.Duration,
+) CommitsRootsCache {
+	return internalNewCommitRootsCache(
+		lggr,
+		messageVisibilityInterval,
+		rootSnoozeTime,
+		CleanupInterval,
+		EvictionGracePeriod,
+	)
+}
+
+func internalNewCommitRootsCache(
+	lggr logger.Logger,
+	messageVisibilityInterval time.Duration,
+	rootSnoozeTime time.Duration,
+	cleanupInterval time.Duration,
+	evictionGracePeriod time.Duration,
+) *commitRootsCache {
+	snoozedRoots := cache.New(rootSnoozeTime, cleanupInterval)
+	executedRoots := cache.New(messageVisibilityInterval+evictionGracePeriod, cleanupInterval)
+
+	return &commitRootsCache{
+		lggr:                      lggr,
+		rootSnoozeTime:            rootSnoozeTime,
+		executedRoots:             executedRoots,
+		snoozedRoots:              snoozedRoots,
+		messageVisibilityInterval: messageVisibilityInterval,
+		cacheMu:                   sync.RWMutex{},
+	}
+}
+
+type commitRootsCache struct {
+	lggr                      logger.Logger
+	messageVisibilityInterval time.Duration
+	rootSnoozeTime            time.Duration
+
+	cacheMu sync.RWMutex
+
+	// snoozedRoots used only for temporary snoozing roots. It's a cache with TTL (usually around 5 minutes,
+	// but this configuration is set up on chain using rootSnoozeTime)
+	snoozedRoots *cache.Cache
+	// executedRoots is a cache with TTL (usually around 8 hours, but this configuration is set up on chain using
+	// messageVisibilityInterval). We keep executed roots there to make sure we don't accidentally try to reprocess
+	// already executed CommitReport
+	executedRoots *cache.Cache
+}
+
+func getKey(source ccipocr3.ChainSelector, merkleRoot ccipocr3.Bytes32) string {
+	return source.String() + "_" + merkleRoot.String()
+}
+
+// MarkAsExecuted marks the root as executed. It means that all the messages from the root were executed and the
+// ExecutionStateChange event was finalized.
+func (r *commitRootsCache) MarkAsExecuted(sel ccipocr3.ChainSelector, merkleRoot ccipocr3.Bytes32) {
+	prettyMerkleRoot := getKey(sel, merkleRoot)
+	r.lggr.Infow("Marking root as executed and removing entirely from cache", "merkleRoot", prettyMerkleRoot)
+
+	r.cacheMu.Lock()
+	defer r.cacheMu.Unlock()
+	r.executedRoots.SetDefault(prettyMerkleRoot, struct{}{})
+}
+
+// Snooze temporarily snoozes the root. It means that the root is not eligible for execution for a certain period of
+// time.
+func (r *commitRootsCache) Snooze(sel ccipocr3.ChainSelector, merkleRoot ccipocr3.Bytes32) {
+	prettyMerkleRoot := getKey(sel, merkleRoot)
+	r.lggr.Infow("Snoozing root temporarily", "merkleRoot", prettyMerkleRoot, "rootSnoozeTime", r.rootSnoozeTime)
+	r.snoozedRoots.SetDefault(prettyMerkleRoot, struct{}{})
+}
+
+// CanExecute returns true if the root is not snoozed and not executed.
+func (r *commitRootsCache) CanExecute(sel ccipocr3.ChainSelector, merkleRoot ccipocr3.Bytes32) bool {
+	prettyMerkleRoot := getKey(sel, merkleRoot)
+	return !r.isSnoozed(prettyMerkleRoot) && !r.isExecuted(prettyMerkleRoot)
+}
+
+// IsSnoozed returns true if the root is snoozed.
+func (r *commitRootsCache) isSnoozed(key string) bool {
+	_, snoozed := r.snoozedRoots.Get(key)
+	return snoozed
+}
+
+// isExecuted returns true if the root is executed.
+func (r *commitRootsCache) isExecuted(key string) bool {
+	_, executed := r.executedRoots.Get(key)
+	return executed
+}

execute/observation.go

@@ -5,14 +5,15 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/smartcontractkit/chainlink-common/pkg/logger"
-
 	"golang.org/x/exp/maps"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/types"
 
 	"github.com/smartcontractkit/chainlink-ccip/execute/exectypes"
+	"github.com/smartcontractkit/chainlink-ccip/execute/optimizers"
 	typeconv "github.com/smartcontractkit/chainlink-ccip/internal/libs/typeconv"
 	dt "github.com/smartcontractkit/chainlink-ccip/internal/plugincommon/discovery/discoverytypes"
 	"github.com/smartcontractkit/chainlink-ccip/pkg/logutil"
@@ -131,6 +132,8 @@ func (p *Plugin) getCommitReportsObservation(
 	lggr logger.Logger,
 	observation exectypes.Observation,
 ) (exectypes.Observation, error) {
+	// TODO: set fetchFrom to the oldest pending commit report.
+	// TODO: or, cache commit reports so that we don't need to fetch them again.
 	fetchFrom := time.Now().Add(-p.offchainCfg.MessageVisibilityInterval.Duration()).UTC()
 
 	// Phase 1: Gather commit reports from the destination chain and determine which messages are required to build
@@ -159,14 +162,30 @@ func (p *Plugin) getCommitReportsObservation(
 	}
 
 	// Get pending exec reports.
-	groupedCommits, err := getPendingExecutedReports(ctx, p.ccipReader, fetchFrom, lggr)
+	groupedCommits, fullyExecutedCommits, err := getPendingExecutedReports(
+		ctx,
+		p.ccipReader,
+		p.commitRootsCache.CanExecute,
+		fetchFrom,
+		lggr,
+	)
 	if err != nil {
 		return exectypes.Observation{}, err
 	}
 
-	// Remove cursed observations.
+	// If fully executed reports are detected, mark them in the cache.
+	// This cache will be re-initialized on each plugin restart.
+	for _, fullyExecutedCommit := range fullyExecutedCommits {
+		p.commitRootsCache.MarkAsExecuted(fullyExecutedCommit.SourceChain, fullyExecutedCommit.MerkleRoot)
+	}
+
+	// Remove and snooze commit reports from cursed chains.
 	for chainSelector, isCursed := range ci.CursedSourceChains {
 		if isCursed {
+			// Snooze everything on a cursed chain.
+			for _, commit := range groupedCommits[chainSelector] {
+				p.commitRootsCache.Snooze(chainSelector, commit.MerkleRoot)
+			}
 			delete(groupedCommits, chainSelector)
 		}
 	}
@@ -292,8 +311,10 @@ func (p *Plugin) getMessagesObservation(
 	observation.TokenData = tkData
 
 	// Make sure encoded observation fits within the maximum observation size.
-	//observation, err = truncateObservation(observation, maxObservationLength, p.emptyEncodedSizes)
-	observation, err = p.observationOptimizer.TruncateObservation(observation)
+	observationOptimizer := optimizers.NewObservationOptimizer(
+		lggr,
+		maxObservationLength)
+	observation, err = observationOptimizer.TruncateObservation(observation)
 	if err != nil {
 		return exectypes.Observation{}, fmt.Errorf("unable to truncate observation: %w", err)
 	}

execute/observation_test.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/smartcontractkit/chainlink-ccip/execute/costlymessages"
 	"github.com/smartcontractkit/chainlink-ccip/execute/exectypes"
-	"github.com/smartcontractkit/chainlink-ccip/execute/optimizers"
 	"github.com/smartcontractkit/chainlink-ccip/execute/tokendata"
 	"github.com/smartcontractkit/chainlink-ccip/internal/mocks"
 	readerpkg_mock "github.com/smartcontractkit/chainlink-ccip/mocks/pkg/reader"
@@ -25,7 +24,6 @@ func Test_getMessagesObservation(t *testing.T) {
 	msgHasher := mocks.NewMessageHasher()
 	tokenDataObserver := tokendata.NoopTokenDataObserver{}
 	costlyMessageObserver := costlymessages.NoopObserver{}
-	observationOptimizer := optimizers.NewObservationOptimizer(maxObservationLength)
 
 	//emptyMsgHash, err := msgHasher.Hash(ctx, cciptypes.Message{})
 	//require.NoError(t, err)
@@ -36,7 +34,6 @@ func Test_getMessagesObservation(t *testing.T) {
 		msgHasher:             msgHasher,
 		tokenDataObserver:     &tokenDataObserver,
 		costlyMessageObserver: &costlyMessageObserver,
-		observationOptimizer:  observationOptimizer,
 	}
 
 	tests := []struct {

execute/optimizers/type_optimizer.go

@@ -3,13 +3,13 @@ package optimizers
 import (
 	"sort"
 
+	"golang.org/x/exp/maps"
+
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 
 	"github.com/smartcontractkit/chainlink-ccip/execute/exectypes"
 	"github.com/smartcontractkit/chainlink-ccip/execute/internal"
-
-	"golang.org/x/exp/maps"
-
+	"github.com/smartcontractkit/chainlink-ccip/pkg/logutil"
 	cciptypes "github.com/smartcontractkit/chainlink-ccip/pkg/types/ccipocr3"
 )
 
@@ -19,8 +19,9 @@ type ObservationOptimizer struct {
 	lggr              logger.Logger
 }
 
-func NewObservationOptimizer(maxEncodedSize int) ObservationOptimizer {
+func NewObservationOptimizer(lggr logger.Logger, maxEncodedSize int) ObservationOptimizer {
 	return ObservationOptimizer{
+		lggr:              logutil.WithComponent(lggr, "ObservationOptimizer"),
 		maxEncodedSize:    maxEncodedSize,
 		emptyEncodedSizes: NewEmptyEncodeSizes(),
 	}

execute/optimizers/type_optimizer_test.go

@@ -3,10 +3,11 @@ package optimizers
 import (
 	"testing"
 
-	"github.com/smartcontractkit/chainlink-ccip/execute/exectypes"
-
 	"github.com/stretchr/testify/require"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+
+	"github.com/smartcontractkit/chainlink-ccip/execute/exectypes"
 	"github.com/smartcontractkit/chainlink-ccip/internal/libs/testhelpers/rand"
 	cciptypes "github.com/smartcontractkit/chainlink-ccip/pkg/types/ccipocr3"
 )
@@ -79,7 +80,8 @@ func Test_truncateLastCommit(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			op := NewObservationOptimizer(10000)
+			lggr := logger.Test(t)
+			op := NewObservationOptimizer(lggr, 10000)
 			truncated, _ := op.truncateLastCommit(tt.observation, tt.chain)
 			require.Equal(t, tt.expected, truncated)
 		})
@@ -166,7 +168,8 @@ func Test_truncateChain(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			op := NewObservationOptimizer(10000) // size not important here
+			lggr := logger.Test(t)
+			op := NewObservationOptimizer(lggr, 10000) // size not important here
 			truncated := op.truncateChain(tt.observation, tt.chain)
 			require.Equal(t, tt.expected, truncated)
 		})
@@ -325,7 +328,8 @@ func Test_truncateObservation(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			tt.observation.TokenData = makeNoopTokenDataObservations(tt.observation.Messages)
 			tt.expected.TokenData = tt.observation.TokenData
-			op := NewObservationOptimizer(tt.maxSize)
+			lggr := logger.Test(t)
+			op := NewObservationOptimizer(lggr, tt.maxSize)
 			obs, err := op.TruncateObservation(tt.observation)
 			if tt.wantErr {
 				require.Error(t, err)