Merge branch 'ccip-develop' into CCIP-2836

.github/workflows/ccip-ocr3-build-lint-test.yml

@@ -24,14 +24,14 @@ jobs:
       - name: Display Go version
         run: go version
       - name: Build
-        run: go build -v ./...
+        run: make
       - name: Install linter
         run: |
           curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v1.59.0
       - name: Run linter
-        run: golangci-lint run -c .golangci.yml
+        run: make lint
       - name: Run tests
-        run: go test -race -fullpath -shuffle on -count 20 -coverprofile=coverage.out ./...
+        run: TEST_COUNT=20 COVERAGE_FILE=coverage.out make test
       - name: Generate coverage report
         if: github.event_name == 'pull_request'
         run: |

.github/workflows/codegen.yml

@@ -0,0 +1,42 @@
+# All code generation should be run prior to pull request. Running it again should not produce a diff.
+name: "Codegen Verifier"
+
+on:
+  pull_request:
+  push:
+    branches:
+      - 'ccip-develop'
+
+jobs:
+  codegen-verifier:
+    runs-on: ubuntu-20.04
+    strategy:
+      matrix:
+        go-version: ['1.21']
+    defaults:
+      run:
+        working-directory: .
+    steps:
+      - uses: actions/checkout@9bb56186c3b09b4f86b1c65136769dd318469633 # v4.1.2
+      - name: Setup Go ${{ matrix.go-version }}
+        uses: actions/setup-go@0c52d547c9bc32b1aa3301fd7a9cb496313a4491 # v5.0.0
+        with:
+          go-version: ${{ matrix.go-version }}
+      - name: Display Go version
+        run: go version
+      - name: Re-Generate mocks
+        run: |
+          rm -rf mocks # delete all mocks to ensure any deprecated files have been removed.
+          make generate
+      - name: Tidy
+        run: go mod tidy
+      - name: ensure no changes
+        run: |
+          set -e
+          git_status=$(git status --porcelain=v1)
+          if [ ! -z "$git_status" ]; then
+            git status
+            git diff
+            echo "Error: modified files detected, run 'make generate' / 'go mod tidy'."
+            exit 1
+          fi

.golangci.yml

@@ -1,6 +1,7 @@
 run:
   timeout: 60s
 linters:
+  verbose: true
   enable:
     - exhaustive
     - exportloopref
@@ -86,6 +87,8 @@ linters-settings:
     # Default: 120.
     line-length: 120
 issues:
+  max-issues-per-linter: 0
+  max-same-issues: 0
   exclude-rules:
     - path: test
       text: "^G404:"

.mockery.yaml

@@ -0,0 +1,13 @@
+with-expecter: true
+filename: "{{.InterfaceName | snakecase}}.go"
+dir: mocks/{{ replaceAll .InterfaceDirRelative "internal" "internal_" }}
+packages:
+    github.com/smartcontractkit/chainlink-ccip/execute/types:
+        interfaces:
+            TokenDataReader:
+    github.com/smartcontractkit/chainlink-ccip/execute/internal/gen:
+        interfaces:
+            ExecutePluginCodec:
+    github.com/smartcontractkit/chainlink-ccip/internal/reader:
+        interfaces:
+            HomeChain:

Makefile

@@ -1,12 +1,22 @@
+TEST_COUNT ?= 10
+COVERAGE_FILE ?= coverage.out
 
-ensure_go_1_21:
+build: ensure_go_version
+	go build -v ./...
+
+generate: ensure_go_version
+	go install github.com/vektra/mockery/v2@v2.43.0
+	mockery
+
+test: ensure_go_version
+	go test -race -fullpath -shuffle on -count $(TEST_COUNT) -coverprofile=$(COVERAGE_FILE) ./...
+
+lint: ensure_go_version
+	golangci-lint run -c .golangci.yml
+
+ensure_go_version:
 	@go version | grep -q 'go1.21' || (echo "Please use go1.21" && exit 1)
 
 ensure_golangcilint_1_59:
 	@golangci-lint --version | grep -q '1.59' || (echo "Please use golangci-lint 1.59" && exit 1)
 
-test: ensure_go_1_21
-	go test -race -fullpath -shuffle on -count 10 ./...
-
-lint: ensure_go_1_21
-	golangci-lint run -c .golangci.yml

commit/plugin.go

@@ -5,11 +5,11 @@ import (
 	"errors"
 	"fmt"
 	"sort"
-	"sync"
 	"time"
 
 	mapset "github.com/deckarep/golang-set/v2"
 
+	"github.com/smartcontractkit/chainlink-ccip/internal/plugincommon"
 	"github.com/smartcontractkit/chainlink-ccip/internal/reader"
 	"github.com/smartcontractkit/chainlink-ccip/pluginconfig"
 	"github.com/smartcontractkit/chainlink-ccip/plugintypes"
@@ -34,14 +34,13 @@ type Plugin struct {
 	oracleIDToP2pID   map[commontypes.OracleID]libocrtypes.PeerID
 	cfg               pluginconfig.CommitPluginConfig
 	ccipReader        reader.CCIP
+	readerSyncer      *plugincommon.BackgroundReaderSyncer
 	tokenPricesReader reader.TokenPrices
 	reportCodec       cciptypes.CommitPluginCodec
 	msgHasher         cciptypes.MessageHasher
 	lggr              logger.Logger
 
-	homeChain        reader.HomeChain
-	bgSyncCancelFunc context.CancelFunc
-	bgSyncWG         *sync.WaitGroup
+	homeChain reader.HomeChain
 }
 
 func NewPlugin(
@@ -56,32 +55,28 @@ func NewPlugin(
 	lggr logger.Logger,
 	homeChain reader.HomeChain,
 ) *Plugin {
-	p := &Plugin{
+	readerSyncer := plugincommon.NewBackgroundReaderSyncer(
+		lggr,
+		ccipReader,
+		syncTimeout(cfg.SyncTimeout),
+		syncFrequency(cfg.SyncFrequency),
+	)
+	if err := readerSyncer.Start(context.Background()); err != nil {
+		lggr.Errorw("error starting background reader syncer", "err", err)
+	}
+
+	return &Plugin{
 		nodeID:            nodeID,
 		oracleIDToP2pID:   oracleIDToP2pID,
 		cfg:               cfg,
 		ccipReader:        ccipReader,
+		readerSyncer:      readerSyncer,
 		tokenPricesReader: tokenPricesReader,
 		reportCodec:       reportCodec,
 		msgHasher:         msgHasher,
 		lggr:              lggr,
 		homeChain:         homeChain,
 	}
-
-	bgSyncCtx, bgSyncCf := context.WithCancel(context.Background())
-	p.bgSyncCancelFunc = bgSyncCf
-	p.bgSyncWG = &sync.WaitGroup{}
-	p.bgSyncWG.Add(1)
-	backgroundReaderSync(
-		bgSyncCtx,
-		p.bgSyncWG,
-		lggr,
-		ccipReader,
-		syncTimeout(cfg.SyncTimeout),
-		time.NewTicker(syncFrequency(p.cfg.SyncFrequency)).C,
-	)
-
-	return p
 }
 
 // Query phase is not used.
@@ -200,14 +195,22 @@ func (p *Plugin) Observation(
 
 }
 
-func (p *Plugin) ValidateObservation(_ ocr3types.OutcomeContext, _ types.Query, ao types.AttributedObservation) error {
+func (p *Plugin) ValidateObservation(
+	outCtx ocr3types.OutcomeContext, _ types.Query, ao types.AttributedObservation) error {
 	obs, err := plugintypes.DecodeCommitPluginObservation(ao.Observation)
 	if err != nil {
 		return fmt.Errorf("decode commit plugin observation: %w", err)
 	}
 
-	if err := validateObservedSequenceNumbers(obs.NewMsgs, obs.MaxSeqNums); err != nil {
-		return fmt.Errorf("validate sequence numbers: %w", err)
+	if outCtx.PreviousOutcome != nil {
+		prevOutcome, err := plugintypes.DecodeCommitPluginOutcome(outCtx.PreviousOutcome)
+		if err != nil {
+			return fmt.Errorf("decode commit plugin previous outcome: %w", err)
+		}
+
+		if err := validateObservedSequenceNumbers(obs.NewMsgs, prevOutcome.MaxSeqNums); err != nil {
+			return fmt.Errorf("validate sequence numbers: %w", err)
+		}
 	}
 
 	observerSupportedChains, err := p.supportedChains(ao.Observer)
@@ -358,16 +361,18 @@ func (p *Plugin) ShouldTransmitAcceptedReport(
 }
 
 func (p *Plugin) Close() error {
-	timeout := 10 * time.Second
+	timeout := 10 * time.Second // todo: cfg
 	ctx, cf := context.WithTimeout(context.Background(), timeout)
 	defer cf()
 
+	if err := p.readerSyncer.Close(); err != nil {
+		p.lggr.Errorw("error closing reader syncer", "err", err)
+	}
+
 	if err := p.ccipReader.Close(ctx); err != nil {
 		return fmt.Errorf("close ccip reader: %w", err)
 	}
 
-	p.bgSyncCancelFunc()
-	p.bgSyncWG.Wait()
 	return nil
 }
 

commit/plugin_functions.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"fmt"
 	"sort"
-	"sync"
-	"time"
 
 	mapset "github.com/deckarep/golang-set/v2"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/types"
@@ -636,53 +634,6 @@ func validateMerkleRootsState(
 	return true, nil
 }
 
-func backgroundReaderSync(
-	ctx context.Context,
-	wg *sync.WaitGroup,
-	lggr logger.Logger,
-	reader reader.CCIP,
-	syncTimeout time.Duration,
-	ticker <-chan time.Time,
-) {
-	go func() {
-		defer wg.Done()
-
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-ticker:
-				if err := syncReader(ctx, lggr, reader, syncTimeout); err != nil {
-					lggr.Errorw("runBackgroundReaderSync failed", "err", err)
-				}
-			}
-		}
-	}()
-}
-
-func syncReader(
-	ctx context.Context,
-	lggr logger.Logger,
-	reader reader.CCIP,
-	syncTimeout time.Duration,
-) error {
-	timeoutCtx, cf := context.WithTimeout(ctx, syncTimeout)
-	defer cf()
-
-	updated, err := reader.Sync(timeoutCtx)
-	if err != nil {
-		return err
-	}
-
-	if !updated {
-		lggr.Debug("no updates found after trying to sync")
-	} else {
-		lggr.Info("ccip reader sync success")
-	}
-
-	return nil
-}
-
 type observedMsgsConsensus struct {
 	seqNumRange cciptypes.SeqNumRange
 	merkleRoot  [32]byte

commit/plugin_functions_test.go

@@ -3,10 +3,8 @@ package commit
 import (
 	"context"
 	"encoding/binary"
-	"fmt"
 	"math/big"
 	"slices"
-	"sync"
 	"testing"
 	"time"
 
@@ -1482,50 +1480,6 @@ func Test_gasPricesConsensus(t *testing.T) {
 	}
 }
 
-func Test_backgroundReaderSync(t *testing.T) {
-	ctx, cf := context.WithCancel(context.Background())
-	lggr := logger.Test(t)
-	reader := mocks.NewCCIPReader()
-	syncTimeout := 50 * time.Millisecond
-	ticker := make(chan time.Time)
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	// start background syncing
-	backgroundReaderSync(ctx, wg, lggr, reader, syncTimeout, ticker)
-
-	// send a tick to trigger the first sync that errors
-	reader.On("Sync", mock.Anything).Return(false, fmt.Errorf("some err")).Once()
-	ticker <- time.Now()
-
-	// send a tick to trigger the second sync that succeeds without changes
-	reader.On("Sync", mock.Anything).Return(false, nil).Once()
-	ticker <- time.Now()
-
-	// make sync hang to see the context timeout
-	reader.On("Sync", mock.Anything).Run(func(args mock.Arguments) {
-		ctx := args.Get(0).(context.Context)
-		for { // simulate endless work until context times out
-			select {
-			case <-ctx.Done():
-				t.Log("context cancelled as expected")
-				return
-			default:
-				time.Sleep(time.Millisecond) // sleep to not block the CPU
-			}
-		}
-	}).Return(false, nil).Once()
-	ticker <- time.Now()
-
-	// send a tick to trigger the fourth sync that succeeds with changes
-	reader.On("Sync", mock.Anything).Return(true, nil).Once()
-	ticker <- time.Now()
-
-	cf()      // trigger bg sync to stop
-	wg.Wait() // wait for it to stop
-	reader.AssertExpectations(t)
-}
-
 func Test_validateMerkleRootsState(t *testing.T) {
 	testCases := []struct {
 		name               string

execute/internal/gen/codec.go

@@ -0,0 +1,9 @@
+// Package gen wraps an external type to generate a mock object.
+package gen
+
+import cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
+
+// ExecutePluginCodec is defined in chainlink-common.
+type ExecutePluginCodec interface {
+	cciptypes.ExecutePluginCodec
+}