Minor generic offchain fixes (#769)

Author: dimitris

commit/merkleroot/observation.go

@@ -292,7 +292,7 @@ type Observer interface {
 	// NOTE: Make sure that caller supports the provided chains.
 	ObserveMerkleRoots(ctx context.Context, ranges []plugintypes.ChainRange) []cciptypes.MerkleRootChain
 
-	// ObserveRMNRemoteCfg observes the RMN remote config for the given destination chain.
+	// ObserveRMNRemoteCfg observes the RMN remote config from the configured destination chain.
 	// Check implementation specific details to learn if external calls are made, if values are cached, etc...
 	// NOTE: Make sure that caller supports the destination chain.
 	ObserveRMNRemoteCfg(ctx context.Context) rmntypes.RemoteConfig

execute/plugin_functions.go

@@ -28,11 +28,11 @@ func validateCommitReportsReadingEligibility(
 	supportedChains mapset.Set[cciptypes.ChainSelector],
 	observedData exectypes.CommitObservations,
 ) error {
-	for chainSel := range observedData {
+	for chainSel, observedDataOfChain := range observedData {
 		if !supportedChains.Contains(chainSel) {
 			return fmt.Errorf("observer not allowed to read from chain %d", chainSel)
 		}
-		for _, data := range observedData[chainSel] {
+		for _, data := range observedDataOfChain {
 			if data.SourceChain != chainSel {
 				return fmt.Errorf("invalid observed data, key=%d but data chain=%d",
 					chainSel, data.SourceChain)
@@ -140,12 +140,12 @@ func validateMessagesConformToCommitReports(
 
 	msgsCount := 0
 	for chain, report := range observedData {
-		for _, data := range report {
-			msgsMap, ok := observedMsgs[chain]
-			if !ok {
-				return fmt.Errorf("no messages observed for chain %d, while report was observed", chain)
-			}
+		msgsMap, ok := observedMsgs[chain]
+		if !ok {
+			return fmt.Errorf("no messages observed for chain %d, while report was observed", chain)
+		}
 
+		for _, data := range report {
 			for seqNum := data.SequenceNumberRange.Start(); seqNum <= data.SequenceNumberRange.End(); seqNum++ {
 				_, ok = msgsMap[seqNum]
 				if !ok {

pkg/reader/ccip.go

@@ -196,10 +196,7 @@ func (r *ccipChainReader) CommitReportsGTETimestamp(
 		"ts", ts,
 		"limit", limit*2)
 
-	reports, err := r.processCommitReports(iter, ts, limit)
-	if err != nil {
-		return nil, err
-	}
+	reports := r.processCommitReports(iter, ts, limit)
 
 	lggr.Debugw("decoded commit reports", "reports", reports)
 
@@ -234,9 +231,11 @@ func (r *ccipChainReader) queryCommitReports(
 	)
 }
 
+// processCommitReports decodes the commit reports from the query results
+// and returns the ones that can be properly parsed and validated.
 func (r *ccipChainReader) processCommitReports(
 	iter []types.Sequence, ts time.Time, limit int,
-) ([]plugintypes2.CommitPluginReportWithMeta, error) {
+) []plugintypes2.CommitPluginReportWithMeta {
 	lggr := r.lggr
 	reports := make([]plugintypes2.CommitPluginReportWithMeta, 0)
 	for _, item := range iter {
@@ -254,14 +253,17 @@ func (r *ccipChainReader) processCommitReports(
 		}
 		allMerkleRoots := append(ev.BlessedMerkleRoots, ev.UnblessedMerkleRoots...)
 		blessedMerkleRoots, unblessedMerkleRoots := r.processMerkleRoots(allMerkleRoots, isBlessed)
+
 		priceUpdates, err := r.processPriceUpdates(ev.PriceUpdates)
 		if err != nil {
-			return nil, err
+			lggr.Errorw("failed to process price updates", "err", err, "priceUpdates", ev.PriceUpdates)
+			continue
 		}
 
 		blockNum, err := strconv.ParseUint(item.Head.Height, 10, 64)
 		if err != nil {
-			return nil, fmt.Errorf("failed to parse block number %s: %w", item.Head.Height, err)
+			lggr.Errorw("failed to parse block number", "blockNum", item.Head.Height, "err", err)
+			continue
 		}
 
 		reports = append(reports, plugintypes2.CommitPluginReportWithMeta{
@@ -278,9 +280,9 @@ func (r *ccipChainReader) processCommitReports(
 	lggr.Debugw("decoded commit reports", "reports", reports)
 
 	if len(reports) < limit {
-		return reports, nil
+		return reports
 	}
-	return reports[:limit], nil
+	return reports[:limit]
 }
 
 func (r *ccipChainReader) processMerkleRoots(
@@ -514,6 +516,8 @@ func (r *ccipChainReader) MsgsBetweenSeqNums(
 	if err != nil {
 		return nil, fmt.Errorf("get onRamp address after query: %w", err)
 	}
+
+	// Ensure the onRamp address hasn't changed during the query.
 	if !bytes.Equal(onRampAddress, onRampAddressAfterQuery) {
 		return nil, fmt.Errorf("onRamp address has changed from %s to %s", onRampAddress, onRampAddressAfterQuery)
 	}