WIP: TXM Status Checking

execute/costlymessages/costly_messages.go

@@ -116,6 +116,10 @@ func (o *observer) Observe(
 
 	costlyMessages := make([]cciptypes.Bytes32, 0)
 	for _, msg := range messages {
+		if msg.IsEmpty() {
+			continue
+		}
+
 		fee, ok := messageFees[msg.Header.MessageID]
 		if !ok {
 			return nil, fmt.Errorf("missing fee for message %s", msg.Header.MessageID)
@@ -335,6 +339,9 @@ func (n *StaticMessageExecCostUSD18Calculator) MessageExecCostUSD18(
 	messageExecCosts := make(map[cciptypes.Bytes32]plugintypes.USD18)
 
 	for _, msg := range messages {
+		if msg.IsEmpty() {
+			continue
+		}
 		cost, ok := n.costs[msg.Header.MessageID]
 		if !ok {
 			return nil, fmt.Errorf("missing exec cost for message %s", msg.Header.MessageID)
@@ -397,6 +404,9 @@ func (c *CCIPMessageFeeUSD18Calculator) MessageFeeUSD18(
 
 	messageFees := make(map[cciptypes.Bytes32]plugintypes.USD18)
 	for _, msg := range messages {
+		if msg.IsEmpty() {
+			continue
+		}
 		feeUSD18 := mathslib.CalculateUsdPerUnitGas(msg.FeeValueJuels.Int, linkPriceUSD.Int)
 		timestamp, ok := messageTimeStamps[msg.Header.MessageID]
 		if !ok {

execute/factory.go

@@ -190,6 +190,7 @@ func (p PluginFactory) NewReportingPlugin(
 			lggr,
 			costlyMessageObserver,
 			metricsReporter,
+			p.chainWriters[p.ocrConfig.Config.ChainSelector].GetTransactionStatus, // dest writer can get status.
 		), ocr3types.ReportingPluginInfo{
 			Name: "CCIPRoleExecute",
 			Limits: ocr3types.ReportingPluginLimits{

execute/observation.go

@@ -8,12 +8,12 @@ import (
 	"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"
+	"github.com/smartcontractkit/chainlink-ccip/execute/report"
 	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"
@@ -273,6 +273,47 @@ func readAllMessages(
 	return messageObs, availableReports, messageTimestamps
 }
 
+// filterMessages according to various caches that we know about.
+func (p *Plugin) filterMessages(
+	ctx context.Context, lggr logger.Logger, messageObs exectypes.MessageObservations,
+) exectypes.MessageObservations {
+
+	// TXM Status checking only supports singleton execution because the msgID is used for the transactionID.
+	// Arbitrary txm checking would be much more complex because we don't know which messages are in the same report.
+	isSingletonExecute := p.offchainCfg.MaxReportMessages == 1 && p.offchainCfg.MaxSingleChainReports == 1
+	if isSingletonExecute {
+		// Reset cache during message observation. It will be referenced again when the report is generated.
+		p.statusCache = report.NewMessageStatusCache(p.statusGetter)
+	}
+	txmStatusChecker := report.NewTXMCheck(p.statusCache, p.offchainCfg.MaxTxmStatusChecks)
+	for chainSelector, msgs := range messageObs {
+		for seqNum, msg := range msgs {
+			// Inflight messages do not need to be observed.
+			if p.inflightMessageCache != nil && p.inflightMessageCache.IsInflight(chainSelector, msg.Header.MessageID) {
+				messageObs[chainSelector][seqNum] = cciptypes.Message{}
+				//delete(messageObs[chainSelector], seqNum)
+				lggr.Infow("skipping message observation - inflight", "msg", msg)
+				continue
+			}
+			// Messages with fatal txm statuses do not need to be observed.
+			if p.statusGetter != nil && isSingletonExecute {
+				status, err := txmStatusChecker(ctx, lggr, msg, 0, exectypes.CommitData{})
+				if err != nil {
+					lggr.Errorw("txm status check error", "msg", msg, "err", err)
+				} else if status != report.None {
+					messageObs[chainSelector][seqNum] = cciptypes.Message{}
+					//delete(messageObs[chainSelector], seqNum)
+					lggr.Infow(fmt.Sprintf("skipping message observation - txm status %s", status),
+						"msg", msg)
+					continue
+				}
+			}
+		}
+	}
+
+	return messageObs
+}
+
 func (p *Plugin) getMessagesObservation(
 	ctx context.Context,
 	lggr logger.Logger,
@@ -296,6 +337,18 @@ func (p *Plugin) getMessagesObservation(
 		previousOutcome.CommitReports,
 	)
 
+	// Compute hashes to allow deleting messages from the observation.
+	hashes, err := exectypes.GetHashes(ctx, messageObs, p.msgHasher)
+	if err != nil {
+		return exectypes.Observation{}, fmt.Errorf("unable to get message hashes: %w", err)
+	}
+
+	// Remove messages which have failed txm statuses.
+	// This is a roundabout way to remove messages from the observation because it will continue to be
+	// executed by other nodes until >F nodes have failed to execute it.
+	// We cannot form consensus on the TXM state because the fatal status is not recorded onchain.
+	messageObs = p.filterMessages(ctx, lggr, messageObs)
+
 	tkData, err1 := p.tokenDataObserver.Observe(ctx, messageObs)
 	if err1 != nil {
 		return exectypes.Observation{}, fmt.Errorf("unable to process token data %w", err1)
@@ -312,11 +365,6 @@ func (p *Plugin) getMessagesObservation(
 		return exectypes.Observation{}, fmt.Errorf("unable to observe costly messages: %w", err)
 	}
 
-	hashes, err := exectypes.GetHashes(ctx, messageObs, p.msgHasher)
-	if err != nil {
-		return exectypes.Observation{}, fmt.Errorf("unable to get message hashes: %w", err)
-	}
-
 	observation.CommitReports = commitReportCache
 	observation.Messages = messageObs
 	observation.Hashes = hashes

execute/observation_test.go

@@ -9,11 +9,13 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/types"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
 
 	"github.com/smartcontractkit/chainlink-ccip/execute/costlymessages"
 	"github.com/smartcontractkit/chainlink-ccip/execute/exectypes"
 	"github.com/smartcontractkit/chainlink-ccip/execute/internal/cache"
+	"github.com/smartcontractkit/chainlink-ccip/execute/report"
 	"github.com/smartcontractkit/chainlink-ccip/execute/tokendata"
 	"github.com/smartcontractkit/chainlink-ccip/internal/mocks"
 	"github.com/smartcontractkit/chainlink-ccip/mocks/internal_/reader"
@@ -104,31 +106,13 @@ func Test_Observation_CacheUpdate(t *testing.T) {
 }
 
 func Test_getMessagesObservation(t *testing.T) {
-	ctx := context.Background()
-
-	// Create mock objects
-	ccipReader := readerpkg_mock.NewMockCCIPReader(t)
-	msgHasher := mocks.NewMessageHasher()
-	tokenDataObserver := tokendata.NoopTokenDataObserver{}
-	costlyMessageObserver := costlymessages.NoopObserver{}
-
-	//emptyMsgHash, err := msgHasher.Hash(ctx, cciptypes.Message{})
-	//require.NoError(t, err)
-	// Set up the plugin with mock objects
-	plugin := &Plugin{
-		lggr:                  mocks.NullLogger,
-		ccipReader:            ccipReader,
-		msgHasher:             msgHasher,
-		tokenDataObserver:     &tokenDataObserver,
-		costlyMessageObserver: &costlyMessageObserver,
-		ocrTypeCodec:          jsonOcrTypeCodec,
-	}
-
 	tests := []struct {
-		name            string
-		previousOutcome exectypes.Outcome
-		expectedObs     exectypes.Observation
-		expectedError   bool
+		name                 string
+		previousOutcome      exectypes.Outcome
+		expectedObs          exectypes.Observation
+		inflightMessageCache func() *cache.InflightMessageCache
+		statusGetter         func() report.StatusGetter
+		expectedError        bool
 	}{
 		{
 			name: "no commit reports",
@@ -230,10 +214,129 @@ func Test_getMessagesObservation(t *testing.T) {
 			},
 			expectedError: false,
 		},
+		{
+			name: "skip inflight messages",
+
+			inflightMessageCache: func() *cache.InflightMessageCache {
+				cache := cache.NewInflightMessageCache(10 * time.Minute)
+				cache.MarkInflight(1, cciptypes.Bytes32{})
+				//cache.MarkInflight(1, )
+				return cache
+			},
+			//inflightMessageCache func() *cache.InflightMessageCache
+			//statusGetter         func() report.StatusGetter
+			previousOutcome: exectypes.Outcome{
+				CommitReports: []exectypes.CommitData{
+					{
+						SourceChain:         1,
+						SequenceNumberRange: cciptypes.NewSeqNumRange(1, 3),
+					},
+				},
+			},
+			expectedObs: exectypes.Observation{
+				CommitReports: exectypes.CommitObservations{
+					1: []exectypes.CommitData{
+						{
+							SourceChain:         1,
+							SequenceNumberRange: cciptypes.NewSeqNumRange(1, 3),
+						},
+					},
+				},
+				Messages: exectypes.MessageObservations{
+					1: {
+						1: cciptypes.Message{}, // skipped message is truncated
+						2: cciptypes.Message{}, // skipped message is truncated
+						3: cciptypes.Message{}, // skipped message is truncated
+					},
+				},
+				CostlyMessages: []cciptypes.Bytes32{},
+				TokenData: exectypes.TokenDataObservations{
+					1: {
+						1: exectypes.NewMessageTokenData(),
+						2: exectypes.NewMessageTokenData(),
+						3: exectypes.NewMessageTokenData(),
+					},
+				},
+			},
+			expectedError: false,
+		},
+		{
+			name: "bad message status",
+
+			statusGetter: func() report.StatusGetter {
+				return func(ctx context.Context, transactionID string) (types.TransactionStatus, error) {
+					return types.Fatal, nil
+				}
+			},
+			previousOutcome: exectypes.Outcome{
+				CommitReports: []exectypes.CommitData{
+					{
+						SourceChain:         1,
+						SequenceNumberRange: cciptypes.NewSeqNumRange(1, 3),
+					},
+				},
+			},
+			expectedObs: exectypes.Observation{
+				CommitReports: exectypes.CommitObservations{
+					1: []exectypes.CommitData{
+						{
+							SourceChain:         1,
+							SequenceNumberRange: cciptypes.NewSeqNumRange(1, 3),
+						},
+					},
+				},
+				Messages: exectypes.MessageObservations{
+					1: {
+						1: cciptypes.Message{}, // skipped message is truncated
+						2: cciptypes.Message{}, // skipped message is truncated
+						3: cciptypes.Message{}, // skipped message is truncated
+					},
+				},
+				CostlyMessages: []cciptypes.Bytes32{},
+				TokenData: exectypes.TokenDataObservations{
+					1: {
+						1: exectypes.NewMessageTokenData(),
+						2: exectypes.NewMessageTokenData(),
+						3: exectypes.NewMessageTokenData(),
+					},
+				},
+			},
+			expectedError: false,
+		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
+
+			// Create mock objects
+			ccipReader := readerpkg_mock.NewMockCCIPReader(t)
+			msgHasher := mocks.NewMessageHasher()
+			tokenDataObserver := tokendata.NoopTokenDataObserver{}
+			costlyMessageObserver := costlymessages.NoopObserver{}
+
+			// Set up the plugin with mock objects
+			plugin := &Plugin{
+				lggr:                  mocks.NullLogger,
+				ccipReader:            ccipReader,
+				msgHasher:             msgHasher,
+				tokenDataObserver:     &tokenDataObserver,
+				costlyMessageObserver: &costlyMessageObserver,
+				ocrTypeCodec:          jsonOcrTypeCodec,
+			}
+
+			if tt.inflightMessageCache != nil {
+				plugin.inflightMessageCache = tt.inflightMessageCache()
+			}
+
+			if tt.statusGetter != nil {
+				plugin.statusGetter = tt.statusGetter()
+
+				// If the status getter is provided, set the offchain configuration to use it.
+				plugin.offchainCfg.MaxSingleChainReports = 1
+				plugin.offchainCfg.MaxReportMessages = 1
+			}
+
 			// Set up mock expectations
 			ccipReader.On("MsgsBetweenSeqNums", ctx, cciptypes.ChainSelector(1),
 				cciptypes.NewSeqNumRange(1, 3)).Return([]cciptypes.Message{

execute/plugin.go

@@ -62,6 +62,7 @@ type Plugin struct {
 	discovery    ContractDiscoveryInterface
 	chainSupport plugincommon.ChainSupport
 	observer     metrics.Reporter
+	statusGetter report.StatusGetter
 
 	oracleIDToP2pID       map[commontypes.OracleID]libocrtypes.PeerID
 	tokenDataObserver     tokendata.TokenDataObserver
@@ -77,6 +78,8 @@ type Plugin struct {
 	commitRootsCache cache.CommitsRootsCache
 	// inflightMessageCache prevents duplicate reports from being sent for the same message.
 	inflightMessageCache inflightMessageCache
+	// statusCache remembers message status to optimize DB lookups. It is reset for each round.
+	statusCache *report.MessageStatusCache
 }
 
 func NewPlugin(
@@ -94,6 +97,7 @@ func NewPlugin(
 	lggr logger.Logger,
 	costlyMessageObserver costlymessages.Observer,
 	metricsReporter metrics.Reporter,
+	getter report.StatusGetter,
 ) *Plugin {
 	lggr.Infow("creating new plugin instance", "p2pID", oracleIDToP2pID[reportingCfg.OracleID])
 
@@ -111,6 +115,7 @@ func NewPlugin(
 		estimateProvider:      estimateProvider,
 		lggr:                  logutil.WithComponent(lggr, "ExecutePlugin"),
 		costlyMessageObserver: costlyMessageObserver,
+		statusGetter:          getter,
 		discovery: discovery.NewContractDiscoveryProcessor(
 			logutil.WithComponent(lggr, "Discovery"),
 			&ccipReader,
@@ -459,6 +464,18 @@ func (p *Plugin) Reports(
 	}
 
 	reportInfo := extractReportInfo(decodedOutcome)
+
+	// Add the TxID if we have a single chain report with a single message.
+	if p.offchainCfg.MaxSingleChainReports == 1 && p.offchainCfg.MaxReportMessages == 1 {
+		if len(decodedOutcome.Report.ChainReports) == 1 {
+			if len(decodedOutcome.Report.ChainReports[0].Messages) == 1 {
+				msg := decodedOutcome.Report.ChainReports[0].Messages[0]
+				reportInfo.TxID = p.statusCache.NextTransactionID(
+					msg.Header.SourceChainSelector, msg.Header.MessageID.String())
+			}
+		}
+	}
+
 	encodedInfo, err := reportInfo.Encode()
 	if err != nil {
 		return nil, err