DS-3508/[Data Streams] gRPC transport for EA

GNUmakefile

@@ -232,6 +232,7 @@ codecgen: $(codecgen) ## Install codecgen
 protoc: ## Install protoc
 	core/scripts/install-protoc.sh 29.3 /
 	go install google.golang.org/protobuf/cmd/protoc-gen-go@`go list -m -json google.golang.org/protobuf | jq -r .Version`
+	go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.6.2
 	go install github.com/smartcontractkit/wsrpc/cmd/protoc-gen-go-wsrpc@`go list -m -json github.com/smartcontractkit/wsrpc | jq -r .Version`
 
 .PHONY: telemetry-protobuf

core/bridges/bridge_type.go

@@ -21,6 +21,7 @@ type BridgeTypeRequest struct {
 	URL                    models.WebURL `json:"url"`
 	Confirmations          uint32        `json:"confirmations"`
 	MinimumContractPayment *assets.Link  `json:"minimumContractPayment"`
+	UseConnectionManager   bool          `json:"useConnectionManager"`
 }
 
 // GetID returns the ID of this structure for jsonapi serialization.
@@ -48,6 +49,7 @@ type BridgeTypeAuthentication struct {
 	IncomingToken          string
 	OutgoingToken          string
 	MinimumContractPayment *assets.Link
+	UseConnectionManager   bool `json:"useConnectionManager"`
 }
 
 // BridgeType is used for external adapters and has fields for
@@ -62,6 +64,7 @@ type BridgeType struct {
 	MinimumContractPayment *assets.Link
 	CreatedAt              time.Time
 	UpdatedAt              time.Time
+	UseConnectionManager   bool `json:"useConnectionManager"`
 }
 
 // NewBridgeType returns a bridge type authentication (with plaintext
@@ -84,6 +87,7 @@ func NewBridgeType(btr *BridgeTypeRequest) (*BridgeTypeAuthentication,
 			IncomingToken:          incomingToken,
 			OutgoingToken:          outgoingToken,
 			MinimumContractPayment: btr.MinimumContractPayment,
+			UseConnectionManager:   btr.UseConnectionManager,
 		}, &BridgeType{
 			Name:                   btr.Name,
 			URL:                    btr.URL,
@@ -92,6 +96,7 @@ func NewBridgeType(btr *BridgeTypeRequest) (*BridgeTypeAuthentication,
 			Salt:                   salt,
 			OutgoingToken:          outgoingToken,
 			MinimumContractPayment: btr.MinimumContractPayment,
+			UseConnectionManager:   btr.UseConnectionManager,
 		}, nil
 }
 

core/bridges/orm.go

@@ -124,8 +124,8 @@ func (o *orm) BridgeTypes(ctx context.Context, offset int, limit int) (bridges [
 
 // CreateBridgeType saves the bridge type.
 func (o *orm) CreateBridgeType(ctx context.Context, bt *BridgeType) error {
-	stmt := `INSERT INTO bridge_types (name, url, confirmations, incoming_token_hash, salt, outgoing_token, minimum_contract_payment, created_at, updated_at)
-	VALUES (:name, :url, :confirmations, :incoming_token_hash, :salt, :outgoing_token, :minimum_contract_payment, now(), now())
+	stmt := `INSERT INTO bridge_types (name, url, confirmations, incoming_token_hash, salt, outgoing_token, minimum_contract_payment, use_connection_manager, created_at, updated_at)
+	VALUES (:name, :url, :confirmations, :incoming_token_hash, :salt, :outgoing_token, :minimum_contract_payment, :use_connection_manager, now(), now())
 	RETURNING *;`
 	err := o.transact(ctx, false, func(tx *orm) error {
 		stmt, err := tx.ds.PrepareNamedContext(ctx, stmt)
@@ -141,8 +141,8 @@ func (o *orm) CreateBridgeType(ctx context.Context, bt *BridgeType) error {
 
 // UpdateBridgeType updates the bridge type.
 func (o *orm) UpdateBridgeType(ctx context.Context, bt *BridgeType, btr *BridgeTypeRequest) error {
-	stmt := "UPDATE bridge_types SET url = $1, confirmations = $2, minimum_contract_payment = $3 WHERE name = $4 RETURNING *"
-	err := o.ds.GetContext(ctx, bt, stmt, btr.URL, btr.Confirmations, btr.MinimumContractPayment, bt.Name)
+	stmt := "UPDATE bridge_types SET url = $1, confirmations = $2, minimum_contract_payment = $3, use_connection_manager = $4 WHERE name = $5 RETURNING *"
+	err := o.ds.GetContext(ctx, bt, stmt, btr.URL, btr.Confirmations, btr.MinimumContractPayment, btr.UseConnectionManager, bt.Name)
 
 	return err
 }

core/internal/cltest/factories.go

@@ -49,8 +49,9 @@ func NewPeerID() (id ragep2ptypes.PeerID) {
 }
 
 type BridgeOpts struct {
-	Name string
-	URL  string
+	Name                 string
+	URL                  string
+	UseConnectionManager bool
 }
 
 // NewBridgeType create new bridge type given info slice
@@ -71,6 +72,7 @@ func NewBridgeType(t testing.TB, opts BridgeOpts) (*bridges.BridgeTypeAuthentica
 	} else {
 		btr.URL = WebURL(t, "https://bridge.example.com/api?"+rnd)
 	}
+	btr.UseConnectionManager = opts.UseConnectionManager
 
 	bta, bt, err := bridges.NewBridgeType(btr)
 	require.NoError(t, err)

core/services/ocr2/plugins/llo/helpers_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto"
 	"crypto/ed25519"
+	"crypto/sha256"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -45,6 +46,7 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/services/chainlink"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/validate"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocrbootstrap"
+	"github.com/smartcontractkit/chainlink/v2/core/services/pipeline/bridgeconn/streamspb"
 	"github.com/smartcontractkit/chainlink/v2/core/services/streams"
 	"github.com/smartcontractkit/chainlink/v2/core/store/models"
 	"github.com/smartcontractkit/chainlink/v2/core/utils/testutils/heavyweight"
@@ -462,6 +464,83 @@ func createSingleDecimalCountingBridge(t *testing.T, name string, i int, p decim
 	return bridgeName
 }
 
+// grpcStreamBridgeServer is a minimal stand-in for a streams-adapter EA: it
+// implements the streamspb.StreamServiceServer side of BridgeConnManager's
+// protocol and, for every subscription it receives, immediately replies with a
+// fixed price under the payload hash BridgeConnManager expects to find in its
+// observation cache (see bridgeConnManager.bridgeObservationCacheKey).
+type grpcStreamBridgeServer struct {
+	streamspb.UnimplementedStreamServiceServer
+	bridgeName string // trimmed of the "bridge-" prefix, matching bridgeConnManager's own trimming
+	price      decimal.Decimal
+}
+
+func (s *grpcStreamBridgeServer) Subscribe(stream streamspb.StreamService_SubscribeServer) error {
+	for {
+		req, err := stream.Recv()
+		if err != nil {
+			return nil //nolint:nilerr // client closing the stream ends the test happy-path; no error to report
+		}
+		for _, sub := range req.GetSubscriptions() {
+			hash, err := grpcStreamBridgeObservationHash(s.bridgeName, sub.GetData().AsMap())
+			if err != nil {
+				return err
+			}
+			resp := &streamspb.SubscribeResponse{
+				Timestamp:       time.Now().UTC().Format(time.RFC3339),
+				ObservationJson: fmt.Appendf(nil, `{"result": %s}`, s.price.String()),
+				PayloadHash:     hash[:],
+			}
+			if err := stream.Send(resp); err != nil {
+				return err
+			}
+		}
+	}
+}
+
+// grpcStreamBridgeObservationHash mirrors bridgeConnManager's
+// bridgeObservationCacheKey so this fake EA's responses land in the real cache.
+func grpcStreamBridgeObservationHash(bridgeName string, data map[string]any) ([32]byte, error) {
+	lookupBytes, err := json.Marshal(data)
+	if err != nil {
+		return [32]byte{}, err
+	}
+	b := make([]byte, 0, len(bridgeName)+len(lookupBytes))
+	b = append(b, bridgeName...)
+	b = append(b, lookupBytes...)
+	return sha256.Sum256(b), nil
+}
+
+// createGRPCStreamBridge starts a plaintext gRPC server implementing the
+// streams-adapter side of BridgeConnManager's protocol and registers a bridge
+// with UseConnectionManager=true pointing at it, so BridgeTask.Run reads
+// observations from BridgeConnManager's cache instead of making an HTTP call.
+func createGRPCStreamBridge(t *testing.T, name string, i int, p decimal.Decimal, borm bridges.ORM) (bridgeName string) {
+	ctx := t.Context()
+	trimmedName := fmt.Sprintf("%s-%d", name, i)
+	bridgeName = "bridge-" + trimmedName
+
+	var lc net.ListenConfig
+	lis, err := lc.Listen(ctx, "tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	s := grpc.NewServer()
+	streamspb.RegisterStreamServiceServer(s, &grpcStreamBridgeServer{bridgeName: trimmedName, price: p})
+	go func() {
+		_ = s.Serve(lis)
+	}()
+	t.Cleanup(s.Stop)
+
+	u, err := url.Parse("http://" + lis.Addr().String())
+	require.NoError(t, err)
+	require.NoError(t, borm.CreateBridgeType(ctx, &bridges.BridgeType{
+		Name:                 bridges.BridgeName(bridgeName),
+		URL:                  models.WebURL(*u),
+		UseConnectionManager: true,
+	}))
+	return bridgeName
+}
+
 func createBridge(t *testing.T, bridgeName string, responseJSON string, borm bridges.ORM) {
 	ctx := t.Context()
 	bridge := httptest.NewServer(http.HandlerFunc(func(res http.ResponseWriter, req *http.Request) {

core/services/ocr2/plugins/llo/integration_test.go

@@ -2336,7 +2336,7 @@ func testIntegrationLLOChannelMerging(t *testing.T, ocr31 bool) {
 	appBootstrap, bootstrapPeerID, _, bootstrapKb, _ := setupNode(t, bootstrapNodePort, "bootstrap_llo", backend, bootstrapCSAKey, nil)
 	bootstrapNode := Node{App: appBootstrap, KeyBundle: bootstrapKb}
 
-	t.Run("Channel merging lifecycle with owners and adders", func(t *testing.T) { //nolint:paralleltest // subtest used for documentation
+	t.Run("Channel merging lifecycle with owners and adders", func(t *testing.T) {
 		packetCh := make(chan *packet, 100000)
 		serverKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 2))
 		serverPubKey := serverKey.PublicKey
@@ -2453,7 +2453,7 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi
 		}
 
 		// Scenario 1: Owner adds initial channels
-		t.Run("Owner adds initial channels", func(t *testing.T) { //nolint:paralleltest // subtest used for documentation
+		t.Run("Owner adds initial channels", func(t *testing.T) {
 			channelDefinitions := llotypes.ChannelDefinitions{
 				1: {
 					ReportFormat: llotypes.ReportFormatJSON,
@@ -2513,7 +2513,7 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi
 		})
 
 		// Scenario 2: Adders add new channels
-		t.Run("Adders add new channels", func(t *testing.T) { //nolint:paralleltest // subtest used for documentation
+		t.Run("Adders add new channels", func(t *testing.T) {
 			// Adder1 adds channels
 			adder1Definitions := llotypes.ChannelDefinitions{
 				10: {
@@ -2611,7 +2611,7 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi
 		})
 
 		// Scenario 3: Owner tombstone some channels
-		t.Run("Owner tombstone channels", func(t *testing.T) { //nolint:paralleltest // subtest used for documentation
+		t.Run("Owner tombstone channels", func(t *testing.T) {
 			// Owner updates definitions, add tombstone to channel 2 and 21
 			channelDefinitions := llotypes.ChannelDefinitions{
 				1: {
@@ -2691,7 +2691,7 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi
 		})
 
 		// Scenario 4: Owner overwrites adder channel
-		t.Run("Owner overwrites adder channel", func(t *testing.T) { //nolint:paralleltest // subtest used for documentation
+		t.Run("Owner overwrites adder channel", func(t *testing.T) {
 			// Owner sets a channel definition with same ID as adder1's channel 10
 			channelDefinitions := llotypes.ChannelDefinitions{
 				1: {
@@ -2755,7 +2755,7 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi
 		})
 
 		// Scenario 5: Verify adder cannot remove channels
-		t.Run("Adder cannot remove channels", func(t *testing.T) { //nolint:paralleltest // subtest used for documentation
+		t.Run("Adder cannot remove channels", func(t *testing.T) {
 			// Adder1 tries to set definitions that exclude channel 11 (which they previously added)
 			adder1NewDefinitions := llotypes.ChannelDefinitions{
 				10: {
@@ -3028,6 +3028,132 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi
 		"tombstoned channel's stream should not be observed (no additional bridge calls)")
 }
 
+// TestIntegration_LLO_bridgeConnManagerHappyPath exercises the BridgeConnManager
+// flow end to end: BridgeTask.Run subscribes to a new asset pair on a bridge with
+// UseConnectionManager=true, the fake EA (a real gRPC server implementing
+// streamspb.StreamServiceServer) sends back an observation, and BridgeConnManager
+// caches it for BridgeTask.Run to serve on subsequent pipeline runs, ultimately
+// producing an LLO report carrying that value. Happy path only.
+func TestIntegration_LLO_bridgeConnManagerHappyPath(t *testing.T) {
+	t.Parallel()
+
+	const (
+		salt     = 600
+		donID    = uint32(848484)
+		streamID = uint32(290)
+	)
+
+	offchainConfig := lloprotocol.OffchainConfig{
+		ProtocolVersion:                     1,
+		DefaultMinReportIntervalNanoseconds: uint64(1 * time.Second),
+	}
+
+	clientCSAKeys := make([]csakey.KeyV2, nNodes)
+	clientPubKeys := make([]ed25519.PublicKey, nNodes)
+	for i := range nNodes {
+		k := big.NewInt(int64(salt + i))
+		key := csakey.MustNewV2XXXTestingOnly(k)
+		clientCSAKeys[i] = key
+		clientPubKeys[i] = key.PublicKey
+	}
+
+	steve, backend, configurator, configuratorAddress, _, _, _, _, configStore, configStoreAddress, _, _, _, _ := setupBlockchain(t)
+	fromBlock := 1
+
+	bootstrapCSAKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 1))
+	bootstrapNodePort := freeport.GetOne(t)
+	appBootstrap, bootstrapPeerID, _, bootstrapKb, _ := setupNode(t, bootstrapNodePort, "bootstrap_llo_bridgeconnmanager", backend, bootstrapCSAKey, nil)
+	bootstrapNode := Node{App: appBootstrap, KeyBundle: bootstrapKb}
+
+	packetCh := make(chan *packet, 100000)
+	serverKey := csakey.MustNewV2XXXTestingOnly(big.NewInt(salt - 2))
+	serverPubKey := serverKey.PublicKey
+	srv := NewMercuryServer(t, serverKey, packetCh)
+	serverURL := startMercuryServer(t, srv, clientPubKeys)
+
+	oracles, nodes := setupNodes(t, nNodes, backend, clientCSAKeys, func(c *chainlink.Config) {
+		c.Mercury.Transmitter.Protocol = new(mercurytransmitter.MercuryTransmitterProtocolGRPC)
+	})
+
+	chainID := testutils.SimulatedChainID
+	relayType := "evm"
+	relayConfig := fmt.Sprintf(`
+chainID = "%s"
+fromBlock = %d
+lloDonID = %d
+lloConfigMode = "bluegreen"
+`, chainID, fromBlock, donID)
+	addBootstrapJob(t, bootstrapNode, configuratorAddress, "job-bridgeconnmanager", relayType, relayConfig)
+
+	pluginConfig := fmt.Sprintf(`servers = { "%s" = "%x" }
+donID = %d
+channelDefinitionsContractAddress = "0x%x"
+channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, configStoreAddress, fromBlock)
+
+	price := decimal.NewFromFloat(1234.5)
+	for i, node := range nodes {
+		bridgeName := createGRPCStreamBridge(t, "connmanager", i, price, node.App.BridgeORM())
+		addSingleDecimalStreamJob(t, node, streamID, bridgeName)
+		addLLOJob(
+			t,
+			node,
+			configuratorAddress,
+			bootstrapPeerID,
+			bootstrapNodePort,
+			clientPubKeys[i],
+			"bridgeconnmanager-stream-test",
+			pluginConfig,
+			relayType,
+			relayConfig,
+		)
+	}
+
+	channelDefinitions := llotypes.ChannelDefinitions{
+		1: {
+			ReportFormat: llotypes.ReportFormatJSON,
+			Streams: []llotypes.Stream{
+				{StreamID: streamID, Aggregator: llotypes.AggregatorMedian},
+			},
+		},
+	}
+	url, sha := newChannelDefinitionsServer(t, channelDefinitions)
+	_, err := configStore.SetChannelDefinitions(steve, donID, url, sha)
+	require.NoError(t, err)
+	backend.Commit()
+
+	setProductionConfig(
+		t, donID, steve, backend, configurator, configuratorAddress, nodes,
+		WithOracles(oracles), WithOffchainConfig(offchainConfig),
+	)
+
+	// The fake EA only starts sending observations once BridgeConnManager's
+	// EAConn sends its first subscription snapshot (on a fixed 10s interval), so
+	// this loop tolerates reports that predate the first cached observation and
+	// keeps polling until one carries the expected price.
+	require.Eventually(t, func() bool {
+		pckt, errReceive := receiveWithTimeout(t, packetCh, 2*time.Second)
+		if errReceive != nil {
+			return false
+		}
+		req := pckt.req
+		if req.ReportFormat != uint32(llotypes.ReportFormatJSON) {
+			return false
+		}
+		_, _, r, _, errDecode := (lloreportcodec.JSONReportCodec{}).UnpackDecode(req.Payload)
+		if errDecode != nil || r.ChannelID != 1 {
+			return false
+		}
+		if len(r.Values) != 1 {
+			return false
+		}
+		dv, ok := r.Values[0].(*lloprotocol.Decimal)
+		if !ok {
+			return false
+		}
+		return dv.Decimal().Equal(price)
+	}, reportTimeout, 100*time.Millisecond, "expected a report sourced from the bridge connection manager cache")
+}
+
 func setupNodes(t *testing.T, nNodes int, backend evmtypes.Backend, clientCSAKeys []csakey.KeyV2, f func(*chainlink.Config)) (oracles []confighelper.OracleIdentityExtra, nodes []Node) {
 	ports := freeport.GetN(t, nNodes)
 	for i := range nNodes {