commit plugin - support batch sizes greater than one

commit/merkleroot/observation.go

@@ -8,10 +8,12 @@ import (
 	"sync"
 	"time"
 
+	mapset "github.com/deckarep/golang-set/v2"
 	chainsel "github.com/smartcontractkit/chain-selectors"
 	"github.com/smartcontractkit/libocr/commontypes"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/types"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/hashutil"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
@@ -114,12 +116,10 @@ func (w *Processor) getObservation(ctx context.Context, q Query, previousOutcome
 	switch nextState {
 	case SelectingRangesForReport:
 		offRampNextSeqNums := w.observer.ObserveOffRampNextSeqNums(ctx)
+		onRampLatestSeqNums := w.observer.ObserveLatestOnRampSeqNums(ctx, w.cfg.DestChain)
+
 		return Observation{
-			// TODO: observe OnRamp max seq nums. The use of offRampNextSeqNums here effectively disables batching,
-			// e.g. the ranges selected for each chain will be [x, x] (e.g. [46, 46]), which means reports will only
-			// contain one message per chain. Querying the OnRamp contract requires changes to reader.CCIPReader,
-			// which will need to be done in a future change.
-			OnRampMaxSeqNums:   offRampNextSeqNums,
+			OnRampMaxSeqNums:   onRampLatestSeqNums,
 			OffRampNextSeqNums: offRampNextSeqNums,
 			FChain:             w.observer.ObserveFChain(),
 		}, nextState
@@ -145,9 +145,12 @@ func (w *Processor) getObservation(ctx context.Context, q Query, previousOutcome
 }
 
 type Observer interface {
-	// ObserveOffRampNextSeqNums observes the next sequence numbers for each source chain from the OffRamp
+	// ObserveOffRampNextSeqNums observes the next OffRamp sequence numbers for each source chain
 	ObserveOffRampNextSeqNums(ctx context.Context) []plugintypes.SeqNumChain
 
+	// ObserveLatestOnRampSeqNums observes the latest OnRamp sequence numbers for each configured source chain.
+	ObserveLatestOnRampSeqNums(ctx context.Context, destChain cciptypes.ChainSelector) []plugintypes.SeqNumChain
+
 	// ObserveMerkleRoots computes the merkle roots for the given sequence number ranges
 	ObserveMerkleRoots(ctx context.Context, ranges []plugintypes.ChainRange) []cciptypes.MerkleRootChain
 
@@ -202,6 +205,55 @@ func (o ObserverImpl) ObserveOffRampNextSeqNums(ctx context.Context) []plugintyp
 	return result
 }
 
+// ObserveLatestOnRampSeqNums observes the latest onRamp sequence numbers for each configured source chain.
+func (o ObserverImpl) ObserveLatestOnRampSeqNums(
+	ctx context.Context, destChain cciptypes.ChainSelector) []plugintypes.SeqNumChain {
+
+	allSourceChains, err := o.chainSupport.KnownSourceChainsSlice()
+	if err != nil {
+		o.lggr.Warnw("call to KnownSourceChainsSlice failed", "err", err)
+		return nil
+	}
+
+	supportedChains, err := o.chainSupport.SupportedChains(o.nodeID)
+	if err != nil {
+		o.lggr.Warnw("call to KnownSourceChainsSlice failed", "err", err)
+		return nil
+	}
+
+	sourceChains := mapset.NewSet(allSourceChains...).Intersect(supportedChains).ToSlice()
+	sort.Slice(sourceChains, func(i, j int) bool { return sourceChains[i] < sourceChains[j] })
+
+	latestOnRampSeqNums := make([]plugintypes.SeqNumChain, len(sourceChains))
+	eg := &errgroup.Group{}
+
+	for i, sourceChain := range sourceChains {
+		i, sourceChain := i, sourceChain
+		eg.Go(func() error {
+			nextOnRampSeqNum, err := o.ccipReader.GetExpectedNextSequenceNumber(ctx, sourceChain, destChain)
+			if err != nil {
+				return fmt.Errorf("failed to get expected next sequence number for source chain %d: %w", sourceChain, err)
+			}
+			if nextOnRampSeqNum == 0 {
+				return fmt.Errorf("expected next sequence number for source chain %d is 0", sourceChain)
+			}
+
+			latestOnRampSeqNums[i] = plugintypes.SeqNumChain{
+				ChainSel: sourceChain,
+				SeqNum:   nextOnRampSeqNum - 1, // Latest is the next one minus one.
+			}
+			return nil
+		})
+	}
+
+	if err := eg.Wait(); err != nil {
+		o.lggr.Warnw("call to GetExpectedNextSequenceNumber failed", "err", err)
+		return nil
+	}
+
+	return latestOnRampSeqNums
+}
+
 // ObserveMerkleRoots computes the merkle roots for the given sequence number ranges
 func (o ObserverImpl) ObserveMerkleRoots(
 	ctx context.Context,

commit/merkleroot/observation_test.go

@@ -35,10 +35,10 @@ func Test_Observation(t *testing.T) {
 		},
 	}
 	offRampNextSeqNums := []plugintypes.SeqNumChain{
-		{
-			ChainSel: 456,
-			SeqNum:   9987,
-		},
+		{ChainSel: 456, SeqNum: 9987},
+	}
+	onRampLatestSeqNums := []plugintypes.SeqNumChain{
+		{ChainSel: 456, SeqNum: 9990},
 	}
 	fChain := map[cciptypes.ChainSelector]int{
 		872: 3,
@@ -58,11 +58,12 @@ func Test_Observation(t *testing.T) {
 			getObserver: func(t *testing.T) *merkleroot.MockObserver {
 				observer := merkleroot.NewMockObserver(t)
 				observer.EXPECT().ObserveOffRampNextSeqNums(mock.Anything).Once().Return(offRampNextSeqNums)
+				observer.EXPECT().ObserveLatestOnRampSeqNums(mock.Anything, mock.Anything).Return(onRampLatestSeqNums)
 				observer.EXPECT().ObserveFChain().Once().Return(fChain)
 				return observer
 			},
 			expObs: Observation{
-				OnRampMaxSeqNums:   offRampNextSeqNums,
+				OnRampMaxSeqNums:   onRampLatestSeqNums,
 				OffRampNextSeqNums: offRampNextSeqNums,
 				FChain:             fChain,
 			},

commit/merkleroot/outcome.go

@@ -10,6 +10,8 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
 
+	"github.com/smartcontractkit/chainlink-ccip/internal/libs/cciptypeutil"
+
 	"github.com/smartcontractkit/chainlink-ccip/commit/merkleroot/rmn"
 	"github.com/smartcontractkit/chainlink-ccip/plugintypes"
 	"github.com/smartcontractkit/chainlink-ccip/shared"
@@ -46,7 +48,7 @@ func (w *Processor) getOutcome(
 
 	switch nextState {
 	case SelectingRangesForReport:
-		return reportRangesOutcome(q, consensusObservation), nextState
+		return reportRangesOutcome(q, w.lggr, consensusObservation, w.cfg.BatchLimits), nextState
 	case BuildingReport:
 		if q.RetryRMNSignatures {
 			// We want to retry getting the RMN signatures on the exact same outcome we had before.
@@ -64,10 +66,11 @@ func (w *Processor) getOutcome(
 }
 
 // reportRangesOutcome determines the sequence number ranges for each chain to build a report from in the next round
-// TODO: ensure each range is below a limit
 func reportRangesOutcome(
 	_ Query,
+	lggr logger.Logger,
 	consensusObservation ConsensusObservation,
+	rangeLimitsPerSourceChain map[cciptypes.ChainSelector]uint64,
 ) Outcome {
 	rangesToReport := make([]plugintypes.ChainRange, 0)
 
@@ -82,11 +85,24 @@ func reportRangesOutcome(
 		}
 
 		if offRampNextSeqNum <= onRampMaxSeqNum {
+			rngLimit, ok := rangeLimitsPerSourceChain[chainSel]
+			if !ok {
+				rngLimit = DefaultSeqNumsBatchLimit
+			}
+
+			rng := cciptypes.NewSeqNumRange(offRampNextSeqNum, onRampMaxSeqNum)
+
 			chainRange := plugintypes.ChainRange{
 				ChainSel:    chainSel,
-				SeqNumRange: [2]cciptypes.SeqNum{offRampNextSeqNum, onRampMaxSeqNum},
+				SeqNumRange: cciptypeutil.SeqNumRangeLimit(rng, rngLimit),
 			}
 			rangesToReport = append(rangesToReport, chainRange)
+
+			if wasTruncated := rng.End() != chainRange.SeqNumRange.End(); wasTruncated {
+				lggr.Infof("Range for chain %d: %s (before truncate: %v)", chainSel, chainRange.SeqNumRange, rng)
+			} else {
+				lggr.Infof("Range for chain %d: %s", chainSel, chainRange.SeqNumRange)
+			}
 		}
 
 		offRampNextSeqNums = append(offRampNextSeqNums, plugintypes.SeqNumChain{

commit/merkleroot/outcome_test.go

@@ -7,6 +7,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
+
+	"github.com/smartcontractkit/chainlink-ccip/plugintypes"
 )
 
 func Test_buildReport(t *testing.T) {
@@ -37,3 +39,82 @@ func Test_buildReport(t *testing.T) {
 		}
 	})
 }
+
+func Test_reportRangesOutcome(t *testing.T) {
+	lggr := logger.Test(t)
+
+	testCases := []struct {
+		name                      string
+		consensusObservation      ConsensusObservation
+		rangeLimitsPerSourceChain map[cciptypes.ChainSelector]uint64
+		expectedOutcome           Outcome
+	}{
+		{
+			name: "base empty outcome",
+			expectedOutcome: Outcome{
+				OutcomeType:             ReportIntervalsSelected,
+				RangesSelectedForReport: []plugintypes.ChainRange{},
+				OffRampNextSeqNums:      []plugintypes.SeqNumChain{},
+			},
+		},
+		{
+			name: "simple scenario with one chain",
+			consensusObservation: ConsensusObservation{
+				OnRampMaxSeqNums: map[cciptypes.ChainSelector]cciptypes.SeqNum{
+					1: 20,
+				},
+				OffRampNextSeqNums: map[cciptypes.ChainSelector]cciptypes.SeqNum{
+					1: 18, // off ramp next is 18, on ramp max is 20 so new msgs are: [18, 19, 20]
+				},
+			},
+			rangeLimitsPerSourceChain: map[cciptypes.ChainSelector]uint64{},
+			expectedOutcome: Outcome{
+				OutcomeType: ReportIntervalsSelected,
+				RangesSelectedForReport: []plugintypes.ChainRange{
+					{ChainSel: 1, SeqNumRange: cciptypes.NewSeqNumRange(18, 20)},
+				},
+				OffRampNextSeqNums: []plugintypes.SeqNumChain{
+					{ChainSel: 1, SeqNum: 18},
+				},
+			},
+		},
+		{
+			name: "simple scenario with one chain",
+			consensusObservation: ConsensusObservation{
+				OnRampMaxSeqNums: map[cciptypes.ChainSelector]cciptypes.SeqNum{
+					1: 20,
+					2: 1000,
+					3: 10000,
+				},
+				OffRampNextSeqNums: map[cciptypes.ChainSelector]cciptypes.SeqNum{
+					1: 18,  // off ramp next is 18, on ramp max is 20 so new msgs are: [18, 19, 20]
+					2: 995, // off ramp next is 995, on ramp max is 1000 so new msgs are: [995, 996, 997, 998, 999, 1000]
+					3: 500, // off ramp next is 500, we have new messages up to 10000 (default limit applied)
+				},
+			},
+			rangeLimitsPerSourceChain: map[cciptypes.ChainSelector]uint64{
+				2: 5, // limit to 5 messages, [1000] should be excluded
+			},
+			expectedOutcome: Outcome{
+				OutcomeType: ReportIntervalsSelected,
+				RangesSelectedForReport: []plugintypes.ChainRange{
+					{ChainSel: 1, SeqNumRange: cciptypes.NewSeqNumRange(18, 20)},
+					{ChainSel: 2, SeqNumRange: cciptypes.NewSeqNumRange(995, 999)},
+					{ChainSel: 3, SeqNumRange: cciptypes.NewSeqNumRange(500, 755)},
+				},
+				OffRampNextSeqNums: []plugintypes.SeqNumChain{
+					{ChainSel: 1, SeqNum: 18},
+					{ChainSel: 2, SeqNum: 995},
+					{ChainSel: 3, SeqNum: 500},
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			outc := reportRangesOutcome(Query{}, lggr, tc.consensusObservation, tc.rangeLimitsPerSourceChain)
+			require.Equal(t, tc.expectedOutcome, outc)
+		})
+	}
+}

commit/merkleroot/processor.go

@@ -15,6 +15,10 @@ import (
 	"github.com/smartcontractkit/chainlink-ccip/shared"
 )
 
+// DefaultSeqNumsBatchLimit is the default number of max new messages to scan, we use this value when
+// the config is not set for a specific chain.
+const DefaultSeqNumsBatchLimit = 256
+
 // Processor is the processor responsible for
 // reading next messages and building merkle roots for them,
 // It's setup to use RMN to query which messages to include in the merkle root and ensures

commit/plugin_e2e_test.go

@@ -331,6 +331,11 @@ func setupNode(
 	}
 	ccipReader.EXPECT().NextSeqNum(ctx, sourceChains).Return(offRampNextSeqNums, nil).Maybe()
 
+	for _, ch := range sourceChains {
+		ccipReader.EXPECT().GetExpectedNextSequenceNumber(
+			ctx, ch, destChain).Return(offRampNextSeqNum[ch]+1, nil).Maybe()
+	}
+
 	p := NewPlugin(
 		ctx,
 		nodeID,

internal/libs/cciptypeutil/seqnumrange.go

@@ -0,0 +1,21 @@
+package cciptypeutil
+
+import "github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
+
+// SeqNumRangeLimit limits the range to n elements by truncating the end if necessary.
+func SeqNumRangeLimit(rng ccipocr3.SeqNumRange, n uint64) ccipocr3.SeqNumRange {
+	numElems := rng.End() - rng.Start() + 1
+	if numElems <= 0 {
+		return rng
+	}
+
+	if uint64(numElems) > n {
+		newEnd := rng.Start() + ccipocr3.SeqNum(n) - 1
+		if newEnd > rng.End() { // overflow - do nothing
+			return rng
+		}
+		rng.SetEnd(newEnd)
+	}
+
+	return rng
+}

internal/libs/cciptypeutil/seqnumrange_test.go

@@ -0,0 +1,62 @@
+package cciptypeutil
+
+import (
+	"testing"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
+)
+
+func TestSeqNumRangeLimit(t *testing.T) {
+	testCases := []struct {
+		name string
+		rng  ccipocr3.SeqNumRange
+		n    uint64
+		want ccipocr3.SeqNumRange
+	}{
+		{
+			name: "no truncation",
+			rng:  ccipocr3.NewSeqNumRange(0, 10),
+			n:    11,
+			want: ccipocr3.NewSeqNumRange(0, 10),
+		},
+		{
+			name: "no truncation 2",
+			rng:  ccipocr3.NewSeqNumRange(100, 110),
+			n:    11,
+			want: ccipocr3.NewSeqNumRange(100, 110),
+		},
+		{
+			name: "truncation",
+			rng:  ccipocr3.NewSeqNumRange(0, 10),
+			n:    10,
+			want: ccipocr3.NewSeqNumRange(0, 9),
+		},
+		{
+			name: "truncation 2",
+			rng:  ccipocr3.NewSeqNumRange(100, 110),
+			n:    10,
+			want: ccipocr3.NewSeqNumRange(100, 109),
+		},
+		{
+			name: "empty",
+			rng:  ccipocr3.NewSeqNumRange(0, 0),
+			n:    0,
+			want: ccipocr3.NewSeqNumRange(0, 0),
+		},
+		{
+			name: "wrong range",
+			rng:  ccipocr3.NewSeqNumRange(20, 15),
+			n:    3,
+			want: ccipocr3.NewSeqNumRange(20, 15),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := SeqNumRangeLimit(tc.rng, tc.n)
+			if got != tc.want {
+				t.Errorf("SeqNumRangeLimit(%v, %v) = %v; want %v", tc.rng, tc.n, got, tc.want)
+			}
+		})
+	}
+}