Merge branch 'main' into rs/exec-da-cost

.mockery.yaml

@@ -17,8 +17,6 @@ packages:
     github.com/smartcontractkit/chainlink-ccip/commit/merkleroot/rmn:
         interfaces:
             Controller:
-            PeerGroup:
-            PeerGroupFactory:
             Stream:
     github.com/smartcontractkit/chainlink-ccip/execute/internal/gas:
         interfaces:
@@ -28,6 +26,10 @@ packages:
             CCIPReader:
             PriceReader:
             RMNHome:
+    github.com/smartcontractkit/chainlink-ccip/pkg/peergroup:
+        interfaces:
+            PeerGroupFactory:
+            PeerGroup:
     github.com/smartcontractkit/chainlink-ccip/pkg/contractreader:
         interfaces:
             Extended:

README.md

@@ -1,12 +1,23 @@
+<div style="text-align:center" align="center">
+    <a href="https://chain.link" target="_blank">
+        <img src="https://raw.githubusercontent.com/smartcontractkit/chainlink/develop/docs/logo-chainlink-blue.svg" width="225" alt="Chainlink logo">
+    </a>
+
+[![License](https://img.shields.io/static/v1?label=license&message=BUSL%201.1&color=green)](https://github.com/smartcontractkit/chainlink-ccip/blob/master/LICENSE)
+[![Code Documentation](https://img.shields.io/static/v1?label=code-docs&message=latest&color=blue)](docs/ccip_protocol.md)
+[![API Documentation](https://img.shields.io/static/v1?label=api-docs&message=latest&color=blue)](https://docs.chain.link/ccip)
+</div>
+
 # chainlink-ccip
 
-This is the repo that implements the OCR3 CCIP plugins. This includes the commit and execution plugins.
+This repo contains [OCR3 plugins][ocr3] for CCIP. See the [documentation](docs/ccip_protocol.md) for more.
 
 ## Getting Started
 
 ### Go Version
 
-This repo uses Go 1.22. You can install Go from their [installation page](https://go.dev/doc/install).
+The version of go is specified in the project's [go.mod](go.mod) file.
+You can install Go from their [installation page](https://go.dev/doc/install).
 
 ### Running the Linter
 
@@ -50,3 +61,5 @@ the `ccip-develop` branch of chainlink-ccip. You can do this by:
 5. Go back to your chainlink-ccip PR and re-run the integration test workflow.
 6. Once the integration test passes, merge your chainlink-ccip PR into `ccip-develop`, however do not delete the branch on the remote.
 7. Create a new PR in ccip that points to the newly merged commit in the `ccip-develop` tree and merge that.
+
+[ocr3]: https://github.com/smartcontractkit/libocr/blob/master/offchainreporting2plus/ocr3types/plugin.go#L108

commit/merkleroot/observation.go

@@ -29,6 +29,8 @@ import (
 	"github.com/smartcontractkit/chainlink-ccip/pkg/consts"
 	readerpkg "github.com/smartcontractkit/chainlink-ccip/pkg/reader"
 	cciptypes "github.com/smartcontractkit/chainlink-ccip/pkg/types/ccipocr3"
+
+	ragep2ptypes "github.com/smartcontractkit/libocr/ragep2p/types"
 )
 
 func (w *Processor) ObservationQuorum(
@@ -86,21 +88,18 @@ func (w *Processor) initializeRMNController(ctx context.Context, prevOutcome Out
 		return fmt.Errorf("failed to get RMN nodes info: %w", err)
 	}
 
-	peerIDs := make([]string, 0, len(rmnNodesInfo))
-	for _, node := range rmnNodesInfo {
-		w.lggr.Infow("Adding RMN node to peerIDs", "node", node.PeerID.String())
-		peerIDs = append(peerIDs, node.PeerID.String())
-	}
+	oraclePeerIDs := make([]ragep2ptypes.PeerID, 0, len(w.oracleIDToP2pID))
 	for _, p2pID := range w.oracleIDToP2pID {
 		w.lggr.Infow("Adding oracle node to peerIDs", "p2pID", p2pID.String())
-		peerIDs = append(peerIDs, p2pID.String())
+		oraclePeerIDs = append(oraclePeerIDs, p2pID)
 	}
 
 	if err := w.rmnController.InitConnection(
 		ctx,
 		cciptypes.Bytes32(w.reportingCfg.ConfigDigest),
 		prevOutcome.RMNRemoteCfg.ConfigDigest,
-		peerIDs,
+		oraclePeerIDs,
+		rmnNodesInfo,
 	); err != nil {
 		return fmt.Errorf("failed to init connection to RMN: %w", err)
 	}

commit/merkleroot/observation_test.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/smartcontractkit/libocr/commontypes"
 	"github.com/smartcontractkit/libocr/ragep2p/types"
+	ragep2ptypes "github.com/smartcontractkit/libocr/ragep2p/types"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
@@ -623,13 +624,15 @@ func Test_Processor_initializeRMNController(t *testing.T) {
 		{ID: 1, PeerID: types.PeerID{1, 2, 3}},
 		{ID: 10, PeerID: types.PeerID{1, 2, 31}},
 	}
+	oracleIDs := []ragep2ptypes.PeerID{}
 	rmnHomeReader.EXPECT().GetRMNNodesInfo(cfg.ConfigDigest).Return(rmnNodes, nil)
 
 	rmnController.EXPECT().InitConnection(
 		ctx,
 		cciptypes.Bytes32(p.reportingCfg.ConfigDigest),
 		cfg.ConfigDigest,
-		[]string{rmnNodes[0].PeerID.String(), rmnNodes[1].PeerID.String()},
+		oracleIDs,
+		rmnNodes,
 	).Return(nil)
 
 	err = p.initializeRMNController(ctx, Outcome{RMNRemoteCfg: cfg})

commit/merkleroot/processor.go

@@ -1,14 +1,12 @@
 package merkleroot
 
 import (
-	"errors"
-	"fmt"
-
 	"github.com/smartcontractkit/libocr/commontypes"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
 	libocrtypes "github.com/smartcontractkit/libocr/ragep2p/types"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
 
 	"github.com/smartcontractkit/chainlink-ccip/commit/merkleroot/rmn"
 	"github.com/smartcontractkit/chainlink-ccip/internal/plugincommon"
@@ -85,23 +83,5 @@ func (p *Processor) Close() error {
 		return nil
 	}
 
-	errs := make([]error, 0)
-
-	// close rmn controller
-	if p.rmnController != nil {
-		if err := p.rmnController.Close(); err != nil {
-			errs = append(errs, fmt.Errorf("close RMN controller: %w", err))
-			p.lggr.Errorw("Failed to close RMN controller", "err", err)
-		}
-	}
-
-	// close rmn home reader
-	if p.rmnHomeReader != nil {
-		if err := p.rmnHomeReader.Close(); err != nil {
-			errs = append(errs, fmt.Errorf("close RMNHome reader: %w", err))
-			p.lggr.Errorw("Failed to close RMNHome reader", "err", err)
-		}
-	}
-
-	return errors.Join(errs...)
+	return services.CloseAll(p.rmnController, p.rmnHomeReader)
 }

commit/merkleroot/rmn/controller.go

@@ -19,6 +19,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	chainsel "github.com/smartcontractkit/chain-selectors"
+	ragep2ptypes "github.com/smartcontractkit/libocr/ragep2p/types"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 
@@ -59,7 +60,8 @@ type Controller interface {
 		ctx context.Context,
 		commitConfigDigest cciptypes.Bytes32,
 		rmnHomeConfigDigest cciptypes.Bytes32,
-		peerIDs []string, // union of oraclePeerIDs and rmnNodePeerIDs (oracles required for peer discovery)
+		oraclePeerIDs []ragep2ptypes.PeerID,
+		rmnNodes []rmntypes.HomeNodeInfo,
 	) error
 
 	// Close closes the connection to the generic peer group endpoint and all the underlying streams.
@@ -242,9 +244,10 @@ func (c *controller) InitConnection(
 	ctx context.Context,
 	commitConfigDigest cciptypes.Bytes32,
 	rmnHomeConfigDigest cciptypes.Bytes32,
-	peerIDs []string,
+	oraclePeerIDs []ragep2ptypes.PeerID,
+	rmnNodes []rmntypes.HomeNodeInfo,
 ) error {
-	return c.peerClient.InitConnection(ctx, commitConfigDigest, rmnHomeConfigDigest, peerIDs)
+	return c.peerClient.InitConnection(ctx, commitConfigDigest, rmnHomeConfigDigest, oraclePeerIDs, rmnNodes)
 }
 
 func (c *controller) Close() error {
@@ -263,25 +266,36 @@ func (c *controller) getRmnSignedObservations(
 	requestedNodes := make(map[uint64]mapset.Set[rmntypes.NodeID])                   // sourceChain -> requested rmnNodeIDs
 	requestsPerNode := make(map[rmntypes.NodeID][]*rmnpb.FixedDestLaneUpdateRequest) // grouped requests for each node
 
-	// For each lane update request send an observation request to at most 'F+1' number of rmn nodes.
-	// At this point we can safely assume that we have at least F+1 supporting each source chain.
-	for sourceChain, updateRequest := range updateRequestsPerChain {
-		requestedNodes[sourceChain] = mapset.NewSet[rmntypes.NodeID]()
-		homeChainF, exist := homeFMap[cciptypes.ChainSelector(sourceChain)]
-		if !exist {
-			return nil, fmt.Errorf("no home F for chain %d", sourceChain)
+	// Send to every RMN node all the lane update requests it supports until all chains have a sufficient amount.
+	// of initial observers. Upon timer expiration, additional requests are sent to the rest of the RMN nodes.
+
+	chainsWithEnoughRequests := mapset.NewSet[uint64]()
+	for nodeID := range rmnNodeInfo {
+		if chainsWithEnoughRequests.Cardinality() == len(updateRequestsPerChain) {
+			break // We have enough initial observers for all source chains.
 		}
 
-		for nodeID := range updateRequest.RmnNodes.Iter() {
-			if consensus.Threshold(requestedNodes[sourceChain].Cardinality()) >= consensus.FPlus1(homeChainF) {
-				break // We have enough initial observers for this source chain.
+		for sourceChain, updateRequest := range updateRequestsPerChain {
+			// if this node cannot support the source chain, skip it
+			if !updateRequest.RmnNodes.Contains(nodeID) {
+				continue
 			}
 
-			requestedNodes[sourceChain].Add(nodeID)
-			if _, exists := requestsPerNode[nodeID]; !exists {
-				requestsPerNode[nodeID] = make([]*rmnpb.FixedDestLaneUpdateRequest, 0)
+			// add the node as a requested observer for this source chain
+			if _, ok := requestedNodes[sourceChain]; !ok {
+				requestedNodes[sourceChain] = mapset.NewSet[rmntypes.NodeID]()
 			}
+			requestedNodes[sourceChain].Add(nodeID)
 			requestsPerNode[nodeID] = append(requestsPerNode[nodeID], updateRequest.Data)
+
+			// if we already have enough requests for this source chain, mark it
+			homeChainF, exist := homeFMap[cciptypes.ChainSelector(sourceChain)]
+			if !exist {
+				return nil, fmt.Errorf("no home F for chain %d", sourceChain)
+			}
+			if consensus.Threshold(requestedNodes[sourceChain].Cardinality()) >= consensus.FPlus1(homeChainF) {
+				chainsWithEnoughRequests.Add(sourceChain)
+			}
 		}
 	}
 

commit/merkleroot/rmn/controller_test.go

@@ -18,6 +18,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	chainsel "github.com/smartcontractkit/chain-selectors"
+	ragep2ptypes "github.com/smartcontractkit/libocr/ragep2p/types"
 
 	readerpkg_mock "github.com/smartcontractkit/chainlink-ccip/mocks/pkg/reader"
 
@@ -692,7 +693,12 @@ func (m *mockPeerClient) resetReceivedRequests() {
 	m.receivedRequests = make(map[rmntypes.NodeID][]*rmnpb.Request)
 }
 
-func (m *mockPeerClient) InitConnection(_ context.Context, _ cciptypes.Bytes32, _ cciptypes.Bytes32, _ []string) error {
+func (m *mockPeerClient) InitConnection(
+	_ context.Context,
+	_ cciptypes.Bytes32,
+	_ cciptypes.Bytes32,
+	_ []ragep2ptypes.PeerID,
+	_ []rmntypes.HomeNodeInfo) error {
 	return nil
 }