test cache invalidation and add comments

Author: dimkouv

commit/plugin.go

@@ -487,6 +487,11 @@ func (p *Plugin) Outcome(
 	if err != nil {
 		lggr.Errorw("failed to get main outcome and cache invalidation", "err", err)
 	}
+
+	// We invalidate the cache when we detect that inflight price updates appeared on-chain.
+	// This is because at this moment we know that prices are updated
+	// and want to instantly sync a potentially outdated cache.
+	// Otherwise, oracles might re-observe the old prices in the next round.
 	ctx = context.WithValue(ctx, consts.InvalidateCacheKey, invalidatePriceCache)
 
 	tokenPriceOutcome, err := p.tokenPriceProcessor.Outcome(

commit/tokenprice/processor_test.go

@@ -0,0 +1,64 @@
+package tokenprice
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink-ccip/pkg/consts"
+	cciptypes "github.com/smartcontractkit/chainlink-ccip/pkg/types/ccipocr3"
+)
+
+func TestProcessor_Outcome_cacheInvalidation(t *testing.T) {
+	lggr := logger.Test(t)
+	ctx := tests.Context(t)
+
+	aobs := &asyncObserver{
+		lggr:            lggr,
+		cancelFunc:      nil,
+		mu:              sync.RWMutex{},
+		triggerSyncChan: make(chan time.Time),
+	}
+
+	p := &processor{
+		lggr: lggr,
+		obs:  aobs,
+	}
+
+	aobs.mu.Lock()
+	aobs.tokenPriceMap = cciptypes.TokenPriceMap{"A": cciptypes.NewBigIntFromInt64(123)}
+	aobs.mu.Unlock()
+
+	// cache is not invalidated with a normal context
+	_, _ = p.Outcome(ctx, Outcome{}, Query{}, nil)
+	aobs.mu.Lock()
+	updates := aobs.tokenPriceMap
+	aobs.mu.Unlock()
+	require.Len(t, updates, 1)
+
+	// cache is not invalidated with invalidation context set to false
+	ctx = context.WithValue(ctx, consts.InvalidateCacheKey, false)
+	_, _ = p.Outcome(ctx, Outcome{}, Query{}, nil)
+	aobs.mu.Lock()
+	updates = aobs.tokenPriceMap
+	aobs.mu.Unlock()
+	require.Len(t, updates, 1)
+
+	// cache is invalidated with invalidation context set to true and a sync op is triggered
+	wg := sync.WaitGroup{}
+	go func() {
+		<-aobs.triggerSyncChan
+	}()
+	ctx = context.WithValue(ctx, consts.InvalidateCacheKey, true)
+	_, _ = p.Outcome(ctx, Outcome{}, Query{}, nil)
+	aobs.mu.Lock()
+	updates = aobs.tokenPriceMap
+	aobs.mu.Unlock()
+	require.Len(t, updates, 0)
+	wg.Wait() // wait until receiving the sync operation signal
+}