Add ccip message assertion to e2e tests

contracts/src/v0.8/ccip/test/helpers/receivers/MaybeRevertMessageReceiver.sol

@@ -11,7 +11,13 @@ contract MaybeRevertMessageReceiver is IAny2EVMMessageReceiver, IERC165 {
   error CustomError(bytes err);
 
   event ValueReceived(uint256 amount);
-  event MessageReceived();
+  event MessageReceived(
+    bytes32 messageId,
+    uint64 sourceChainSelector,
+    bytes sender,
+    bytes data,
+    Client.EVMTokenAmount[] destTokenAmounts
+  );
 
   address private s_manager;
   bool public s_toRevert;
@@ -45,13 +51,18 @@ contract MaybeRevertMessageReceiver is IAny2EVMMessageReceiver, IERC165 {
     return interfaceId == type(IAny2EVMMessageReceiver).interfaceId || interfaceId == type(IERC165).interfaceId;
   }
 
-  function ccipReceive(
-    Client.Any2EVMMessage calldata
-  ) external override {
+  function ccipReceive(Client.Any2EVMMessage calldata message) external override {
     if (s_toRevert) {
       revert CustomError(s_err);
     }
-    emit MessageReceived();
+
+    emit MessageReceived(
+      message.messageId,
+      message.sourceChainSelector,
+      message.sender,
+      message.data,
+      message.destTokenAmounts
+    );
   }
 
   receive() external payable {

core/gethwrappers/ccip/generation/generated-wrapper-dependency-versions-do-not-edit.txt

@@ -16,7 +16,7 @@ evm_2_evm_onramp: ../../../contracts/solc/v0.8.24/EVM2EVMOnRamp/EVM2EVMOnRamp.ab
 fee_quoter: ../../../contracts/solc/v0.8.24/FeeQuoter/FeeQuoter.abi ../../../contracts/solc/v0.8.24/FeeQuoter/FeeQuoter.bin 503823a939ff99fe3bdaaef7a89cd4bbe475e260d3921335dbf9c80d4f584b76
 lock_release_token_pool: ../../../contracts/solc/v0.8.24/LockReleaseTokenPool/LockReleaseTokenPool.abi ../../../contracts/solc/v0.8.24/LockReleaseTokenPool/LockReleaseTokenPool.bin 1067f557abeb5570f1da7f050ea982ffad0f35dc064e668a8a0e6af128df490c
 lock_release_token_pool_and_proxy: ../../../contracts/solc/v0.8.24/LockReleaseTokenPoolAndProxy/LockReleaseTokenPoolAndProxy.abi ../../../contracts/solc/v0.8.24/LockReleaseTokenPoolAndProxy/LockReleaseTokenPoolAndProxy.bin e632b08be0fbd1d013e8b3a9d75293d0d532b83071c531ff2be1deec1fa48ec1
-maybe_revert_message_receiver: ../../../contracts/solc/v0.8.24/MaybeRevertMessageReceiver/MaybeRevertMessageReceiver.abi ../../../contracts/solc/v0.8.24/MaybeRevertMessageReceiver/MaybeRevertMessageReceiver.bin d73956c26232ebcc4a5444429fa99cbefed960e323be9b5a24925885c2e477d5
+maybe_revert_message_receiver: ../../../contracts/solc/v0.8.24/MaybeRevertMessageReceiver/MaybeRevertMessageReceiver.abi ../../../contracts/solc/v0.8.24/MaybeRevertMessageReceiver/MaybeRevertMessageReceiver.bin 982f83cce7216832e747614bb25541debffe050a7141e93806e55a69150746da
 message_hasher: ../../../contracts/solc/v0.8.24/MessageHasher/MessageHasher.abi ../../../contracts/solc/v0.8.24/MessageHasher/MessageHasher.bin 0a2661da24147160383ad61d56a258515d1cc07f5e0f471ec5cbb4bccaf82389
 mock_lbtc_token_pool: ../../../contracts/solc/v0.8.24/MockLBTCTokenPool/MockLBTCTokenPool.abi ../../../contracts/solc/v0.8.24/MockLBTCTokenPool/MockLBTCTokenPool.bin 1bb773ea4cd73712c84335de3e56afc2f738c9b96f3417316b5136e47591b5cc
 mock_usdc_token_messenger: ../../../contracts/solc/v0.8.24/MockE2EUSDCTokenMessenger/MockE2EUSDCTokenMessenger.abi ../../../contracts/solc/v0.8.24/MockE2EUSDCTokenMessenger/MockE2EUSDCTokenMessenger.bin d976651d36b33ac2196b32b9d2f4fa6690c6a18d41b621365659fce1c1d1e737

integration-tests/ccip-tests/actions/ccip_helpers.go

@@ -57,6 +57,7 @@
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/evm_2_evm_onramp"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/evm_2_evm_onramp_1_2_0"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/fee_quoter"
+	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/maybe_revert_message_receiver"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/mock_rmn_contract"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/rmn_contract"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/router"
@@ -1888,26 +1889,28 @@
 func (sourceCCIP *SourceCCIPModule) SendRequest(
 	receiver common.Address,
 	gasLimit *big.Int,
-) (common.Hash, time.Duration, *big.Int, error) {
+) (common.Hash, time.Duration, *big.Int, []byte, error) {
 	var d time.Duration
 	destChainSelector, err := chainselectors.SelectorFromChainId(sourceCCIP.DestinationChainId)
 	if err != nil {
-		return common.Hash{}, d, nil, fmt.Errorf("failed getting the chain selector: %w", err)
+		return common.Hash{}, d, nil, nil, fmt.Errorf("failed getting the chain selector: %w", err)
 	}
 	// form the message for transfer
 	msg, err := sourceCCIP.CCIPMsg(receiver, sourceCCIP.Common.AllowOutOfOrder, gasLimit)
 	if err != nil {
-		return common.Hash{}, d, nil, fmt.Errorf("failed forming the ccip msg: %w", err)
+		return common.Hash{}, d, nil, nil, fmt.Errorf("failed forming the ccip msg: %w", err)
 	}
 
+	msgData := msg.Data
+
 	fee, err := sourceCCIP.Common.Router.GetFee(destChainSelector, msg)
 	if err != nil {
 		log.Info().Interface("Msg", msg).Msg("CCIP msg")
 		reason, _ := blockchain.RPCErrorFromError(err)
 		if reason != "" {
-			return common.Hash{}, d, nil, fmt.Errorf("failed getting the fee: %s", reason)
+			return common.Hash{}, d, nil, nil, fmt.Errorf("failed getting the fee: %s", reason)
 		}
-		return common.Hash{}, d, nil, fmt.Errorf("failed getting the fee: %w", err)
+		return common.Hash{}, d, nil, nil, fmt.Errorf("failed getting the fee: %w", err)
 	}
 	log.Info().Str("Fee", fee.String()).Msg("Calculated fee")
 
@@ -1923,7 +1926,7 @@
 			if sendTx != nil {
 				txHash = sendTx.Hash()
 			}
-			return txHash, time.Since(timeNow), nil, fmt.Errorf("failed initiating the transfer ccip-send: %w", err)
+			return txHash, time.Since(timeNow), nil, nil, fmt.Errorf("failed initiating the transfer ccip-send: %w", err)
 		}
 	} else {
 		sendTx, err = sourceCCIP.Common.Router.CCIPSendAndProcessTx(destChainSelector, msg, fee)
@@ -1932,7 +1935,7 @@
 			if sendTx != nil {
 				txHash = sendTx.Hash()
 			}
-			return txHash, time.Since(timeNow), nil, fmt.Errorf("failed initiating the transfer ccip-send: %w", err)
+			return txHash, time.Since(timeNow), nil, nil, fmt.Errorf("failed initiating the transfer ccip-send: %w", err)
 		}
 	}
 
@@ -1941,7 +1944,7 @@
 		Str("Send token transaction", sendTx.Hash().String()).
 		Str("lane", fmt.Sprintf("%s-->%s", sourceCCIP.Common.ChainClient.GetNetworkName(), sourceCCIP.DestNetworkName)).
 		Msg("Sending token")
-	return sendTx.Hash(), time.Since(timeNow), fee, nil
+	return sendTx.Hash(), time.Since(timeNow), fee, msgData, nil
 }
 
 func DefaultSourceCCIPModule(
@@ -1987,6 +1990,7 @@
 	OffRamp                 *contracts.OffRamp
 	ReportAcceptedWatcher   *sync.Map
 	ExecStateChangedWatcher *sync.Map
+	MessageReceivedWatcher  *sync.Map
 	ReportBlessedWatcher    *sync.Map
 	ReportBlessedBySeqNum   *sync.Map
 	NextSeqNumToCommit      *atomic.Uint64
@@ -2697,6 +2701,73 @@
 	}
 }
 
+func (destCCIP *DestCCIPModule) AssertMessageContentMatch(
+	lggr *zerolog.Logger,
+	messageID string,
+	expectedContent []byte,
+	timeout time.Duration,
+	reqStat *testreporters.RequestStat,
+) error {
+	fmt.Println(reqStat)
+	lggr.Info().
+		Str("MsgID", fmt.Sprintf("0x%x", string(messageID))).
+		Str("Timeout", timeout.String()).
+		Msg("Waiting for message content to match")
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	resetTimerCount := 0
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			// Load the message content from the watcher
+			value, ok := destCCIP.MessageReceivedWatcher.Load(messageID)
+			if !ok {
+				continue
+			}
+
+			receivedContent, ok := value.([]uint8)
+			if !ok {
+				lggr.Warn().
+					Str("MsgID", fmt.Sprintf("0x%x", string(messageID))).
+					Msg("Invalid content type in MessageReceivedWatcher")
+				continue
+			}
+
+			// Compare the received content with the expected content
+			if string(receivedContent) == string(expectedContent) {
+				lggr.Info().
+					Str("MessageID 0x%x", string(messageID)).
+					Str("Received Content", string(receivedContent)).
+					Str("Expected Content", string(expectedContent)).
+					Msg("Message received and its content matches the sent content")
+				return nil
+			}
+
+			lggr.Warn().
+				Str("MessageID 0x%x", string(messageID)).
+				Str("Received Content", string(receivedContent)).
+				Str("Expected Content", string(expectedContent)).
+				Msg("Message content mismatch")
+
+		case <-timer.C:
+			// Handle timeout with potential connection issue recovery
+			if destCCIP.Common.IsConnectionRestoredRecently != nil && !destCCIP.Common.IsConnectionRestoredRecently.Load() {
+				if resetTimerCount > 2 {
+					return fmt.Errorf("possible RPC issue - message content did not match for MessageID 0x%x", string(messageID))
+				}
+				timer.Reset(timeout)
+				resetTimerCount++
+				lggr.Info().Int("count of reset", resetTimerCount).Msg("Resetting timer to validate message content match")
+				continue
+			}
+
+			return fmt.Errorf("timeout - message was not received or content did not match for MessageID 0x%x", string(messageID))
+		}
+	}
+}
+
 func DefaultDestinationCCIPModule(
 	logger *zerolog.Logger,
 	testConf *testconfig.CCIPTestGroupConfig,
@@ -2726,6 +2797,7 @@
 		ReportBlessedBySeqNum:   &sync.Map{},
 		ExecStateChangedWatcher: &sync.Map{},
 		ReportAcceptedWatcher:   &sync.Map{},
+		MessageReceivedWatcher:  &sync.Map{},
 	}, nil
 }
 
@@ -2734,6 +2806,7 @@
 	txHash                  string
 	txConfirmationTimestamp time.Time
 	RequestStat             *testreporters.RequestStat
+	MessageData             []byte
 }
 
 func CCIPRequestFromTxHash(txHash common.Hash, chainClient blockchain.EVMClient) (CCIPRequest, *types.Receipt, error) {
@@ -2886,7 +2959,7 @@
 	lane.SentReqs = make(map[common.Hash][]CCIPRequest)
 }
 
-func (lane *CCIPLane) AddToSentReqs(txHash common.Hash, reqStats []*testreporters.RequestStat) (*types.Receipt, error) {
+func (lane *CCIPLane) AddToSentReqs(txHash common.Hash, reqStats []*testreporters.RequestStat, msgData []byte) (*types.Receipt, error) {
 	request, rcpt, err := CCIPRequestFromTxHash(txHash, lane.Source.Common.ChainClient)
 	if err != nil {
 		for _, stat := range reqStats {
@@ -2901,6 +2974,7 @@
 			txHash:                  rcpt.TxHash.Hex(),
 			txConfirmationTimestamp: request.txConfirmationTimestamp,
 			RequestStat:             stat,
+			MessageData:             msgData,
 		})
 		lane.NumberOfReq++
 	}
@@ -2987,7 +3061,7 @@
 		}
 		return fmt.Errorf("failed to send the multicall: %w", err)
 	}
-	rcpt, err := lane.AddToSentReqs(tx.Hash(), reqStats)
+	rcpt, err := lane.AddToSentReqs(tx.Hash(), reqStats, nil)
 	if err != nil {
 		return err
 	}
@@ -3009,7 +3083,7 @@
 func (lane *CCIPLane) SendRequests(noOfRequests int, gasLimit *big.Int) error {
 	for i := 1; i <= noOfRequests; i++ {
 		stat := testreporters.NewCCIPRequestStats(int64(lane.NumberOfReq+i), lane.SourceNetworkName, lane.DestNetworkName)
-		txHash, txConfirmationDur, fee, err := lane.Source.SendRequest(lane.Dest.ReceiverDapp.EthAddress, gasLimit)
+		txHash, txConfirmationDur, fee, msgData, err := lane.Source.SendRequest(lane.Dest.ReceiverDapp.EthAddress, gasLimit)
 		if err != nil {
 			stat.UpdateState(lane.Logger, 0, testreporters.TX, txConfirmationDur, testreporters.Failure, nil)
 			return fmt.Errorf("could not send request: %w", err)
@@ -3026,7 +3100,7 @@
 				noOfTokens++
 			}
 		}
-		_, err = lane.AddToSentReqs(txHash, []*testreporters.RequestStat{stat})
+		_, err = lane.AddToSentReqs(txHash, []*testreporters.RequestStat{stat}, msgData)
 		if err != nil {
 			return err
 		}
@@ -3301,6 +3375,14 @@
 		if opts.phaseExpectedToFail == testreporters.Commit && opts.timeout != 0 {
 			timeout = opts.timeout
 		}
+
+		err = lane.Dest.AssertMessageContentMatch(lane.Logger, string(msgLog.MessageId[:]), []byte(lane.SentReqs[txHash][0].MessageData), timeout, reqStat)
+		if err != nil {
+			return fmt.Errorf("message validation failed: %v", err)
+		} else {
+			log.Info().Msg("Message content validation successful")
+		}
+
 		err = lane.Dest.AssertSeqNumberExecuted(lane.Logger, seqNumber, timeout, sourceLogFinalizedAt, reqStat)
 		if shouldReturn, phaseErr := isPhaseValid(lane.Logger, testreporters.Commit, opts, err); shouldReturn {
 			return phaseErr
@@ -3541,6 +3623,35 @@
 		}
 	}(reportAccSub)
 
+	messageReceivedEvent := make(chan *maybe_revert_message_receiver.MaybeRevertMessageReceiverMessageReceived)
+	messageReceivedSub := event.Resubscribe(DefaultResubscriptionTimeout, func(_ context.Context) (event.Subscription, error) {
+		sub, err := lane.Dest.ReceiverDapp.WatchMessageReceived(nil, messageReceivedEvent)
+		if err != nil {
+			log.Error().Err(err).Msg("error in subscribing to message received event")
+		}
+		return sub, err
+	})
+	if messageReceivedSub == nil {
+		return fmt.Errorf("failed to subscribe to message received event")
+	}
+	go func(sub event.Subscription) {
+		defer sub.Unsubscribe()
+		for {
+			select {
+			case e := <-messageReceivedEvent:
+				messageId := string(e.MessageId[:])
+				messageContent := e.Data
+				messageSender := string(e.Sender[:])
+				log.Info().Msgf("Message event received for message id: 0x%x", messageId)
+				log.Info().Msgf("Message event received with content: %+v", string(messageContent))
+				log.Info().Msgf("Message event received with sender: 0x%x", messageSender[len(messageSender)-20:])
+				lane.Dest.MessageReceivedWatcher.Store(messageId, messageContent)
+			case <-lane.Context.Done():
+				return
+			}
+		}
+	}(messageReceivedSub)
+
 	if lane.Dest.Common.ARM != nil {
 		reportBlessedEvent := make(chan *rmn_contract.RMNContractTaggedRootBlessed)
 		blessedSub := event.Resubscribe(DefaultResubscriptionTimeout, func(_ context.Context) (event.Subscription, error) {

integration-tests/ccip-tests/contracts/contract_models.go

@@ -1099,6 +1099,18 @@ func (rDapp *ReceiverDapp) ToggleRevert(revert bool) error {
 	return rDapp.client.ProcessTransaction(tx)
 }
 
+func (rDapp *ReceiverDapp) WatchMessageReceived(opts *bind.WatchOpts, messageReceivedEvent chan *maybe_revert_message_receiver.MaybeRevertMessageReceiverMessageReceived) (event.Subscription, error) {
+	if rDapp.instance != nil {
+		return rDapp.instance.WatchMessageReceived(opts, messageReceivedEvent)
+	}
+
+	newInstance, err := maybe_revert_message_receiver.NewMaybeRevertMessageReceiver(rDapp.EthAddress, wrappers.MustNewWrappedContractBackend(rDapp.client, nil))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create a new ReceiverDapp contract instance: %w", err)
+	}
+	return newInstance.WatchMessageReceived(opts, messageReceivedEvent)
+}
+
 type InternalTimestampedPackedUint224 struct {
 	Value     *big.Int
 	Timestamp uint32

integration-tests/ccip-tests/load/helper.go

@@ -206,7 +206,7 @@ func (l *LoadArgs) ValidateCurseFollowedByUncurse() {
 		// try to send requests on lanes on which curse is applied on source RMN and the request should revert
 		// data-only transfer is sufficient
 		lane.Source.TransferAmount = []*big.Int{}
-		failedTx, _, _, err := lane.Source.SendRequest(lane.Dest.ReceiverDapp.EthAddress, big.NewInt(actions.DefaultDestinationGasLimit))
+		failedTx, _, _, _, err := lane.Source.SendRequest(lane.Dest.ReceiverDapp.EthAddress, big.NewInt(actions.DefaultDestinationGasLimit))
 		if lane.Source.Common.ChainClient.GetNetworkConfig().MinimumConfirmations > 0 {
 			require.Error(l.t, err)
 		} else {