execute: pseudo-delete already executed messages (#888)

* execute: pseudo-delete already executed messages

The plugin was getting stuck on large commit reports because it kept
fully including already executed messages in the observation when it
should have pseudo-deleted those and included messages that are not
executed.

* fix lint

Author: makramkd

execute/observation.go

@@ -3,6 +3,7 @@ package execute
 import (
 	"context"
 	"fmt"
+	"slices"
 	"sort"
 	"time"
 
@@ -339,6 +340,8 @@ func createEmptyMessageWithIDAndSeqNum(msg cciptypes.Message) cciptypes.Message
 //
 // Returns:
 //   - Updated observation containing commit reports, messages, and hashes
+//
+//nolint:gocyclo // TODO: pull out appropriate helpers.
 func (p *Plugin) getMessagesObservation(
 	ctx context.Context,
 	lggr logger.Logger,
@@ -349,7 +352,7 @@ func (p *Plugin) getMessagesObservation(
 	//          These messages will not be executed in the current round, but may be executed in future rounds
 	//          (e.g. if gas prices decrease).
 	if len(previousOutcome.CommitReports) == 0 {
-		lggr.Debug("TODO: No reports to execute. This is expected after a cold start.")
+		lggr.Info("No reports to execute. This is expected after a cold start.")
 		// No reports to execute.
 		// This is expected after a cold start.
 		return observation, nil
@@ -421,10 +424,15 @@ func (p *Plugin) getMessagesObservation(
 		// Process each message in the report and override the empty message and token data if everything fits within
 		// the size limits
 		for _, msg := range msgs {
-			// If msg is not already inflight, add it
+			// If a message is inflight or already executed, don't include it fully in the observation
+			// because its already been transmitted in a previous report or executed onchain.
 			if p.inflightMessageCache.IsInflight(srcChain, msg.Header.MessageID) {
 				continue
 			}
+			if slices.Contains(report.ExecutedMessages, msg.Header.SequenceNumber) {
+				continue
+			}
+
 			seqNum := msg.Header.SequenceNumber
 			messageObs[srcChain][seqNum] = msg
 			tkData[srcChain][seqNum] = p.observeTokenDataForMessage(ctx, lggr, msg)

execute/observation_test.go

@@ -127,11 +127,17 @@ func Test_getMessagesObservation(t *testing.T) {
 	oneByte := make([]byte, 1)
 
 	// Helper functions
-	createCommitData := func(srcChain cciptypes.ChainSelector, from, to cciptypes.SeqNum) exectypes.CommitData {
+	createCommitData := func(
+		srcChain cciptypes.ChainSelector,
+		from,
+		to cciptypes.SeqNum,
+		executedMessages ...cciptypes.SeqNum,
+	) exectypes.CommitData {
 		return exectypes.CommitData{
 			SourceChain:         srcChain,
 			SequenceNumberRange: cciptypes.NewSeqNumRange(from, to),
 			Timestamp:           timestamp,
+			ExecutedMessages:    executedMessages,
 		}
 	}
 
@@ -431,6 +437,46 @@ func Test_getMessagesObservation(t *testing.T) {
 			},
 			expectedError: false,
 		},
+		{
+			name: "executed messages are skipped but hashed",
+			commitData: []exectypes.CommitData{
+				createCommitData(src1, 1, 3, 1, 2), // 1 and 2 are already executed
+			},
+			setupMocks: func(ccipReader *readerpkg_mock.MockCCIPReader,
+				estimateProvider *ccipocr3.MockEstimateProvider,
+				inflightCache *cache.InflightMessageCache,
+				codec *codec_mock.MockExecCodec,
+			) {
+				// Any small size that fits within the max observation size
+				codec.EXPECT().EncodeObservation(mock.Anything).Return(oneByte, nil).Maybe()
+				messages := createMessages(src1, dest, 1, 3)
+
+				ccipReader.On("MsgsBetweenSeqNums", ctx, src1, cciptypes.NewSeqNumRange(1, 3)).
+					Return(messages, nil)
+			},
+			expectedObs: exectypes.Observation{
+				Messages: exectypes.MessageObservations{
+					src1: {
+						// 1 and 2 are pseudo deleted because they are already executed
+						1: NewMessage(1, 1, 0, 0),
+						2: NewMessage(2, 2, 0, 0),
+						3: NewMessage(3, 3, int(src1), int(dest)),
+					},
+				},
+				CommitReports: exectypes.CommitObservations{
+					src1: []exectypes.CommitData{
+						createCommitData(src1, 1, 3, 1, 2), // 1 and 2 are already executed
+					},
+				},
+				Hashes: exectypes.MessageHashes{
+					src1: createHashesMap(1, 3),
+				},
+				TokenData: exectypes.TokenDataObservations{
+					src1: createTokenData(1, 3),
+				},
+			},
+			expectedError: false,
+		},
 		{
 			name: "encoding size exceeded mid report",
 			commitData: []exectypes.CommitData{