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

GNUmakefile

@@ -234,6 +234,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

@@ -50,8 +50,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
@@ -72,6 +73,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)
@@ -197,7 +199,6 @@ NOW(),NOW(),$1,'{}',false,$2,$3,0,0,0,0,0,0,0,0,0
 	return spec
 }
 
-
 func MustInsertExternalInitiator(t *testing.T, orm bridges.ORM) (ei bridges.ExternalInitiator) {
 	return MustInsertExternalInitiatorWithOpts(t, orm, ExternalInitiatorOpts{})
 }

core/services/pipeline/bridgeconn/bridge_conn_manager.go

@@ -0,0 +1,171 @@
+package bridgeconn
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	stdErrors "errors"
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/goccy/go-json"
+	"google.golang.org/protobuf/types/known/structpb"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+
+	"github.com/smartcontractkit/chainlink/v2/core/bridges"
+	"github.com/smartcontractkit/chainlink/v2/core/store/models"
+)
+
+//nolint:revive // Interface name matches existing project convention.
+type BridgeConnManager interface {
+	GetObservation(bridge bridges.BridgeType, requestData map[string]any) ([]byte, error)
+}
+
+var (
+	ErrBridgeObservationNotFound = stdErrors.New("bridge observation not found")
+)
+
+// bridgeConnManager is a package-level singleton: one observation cache plus one
+// EAConn registry shared by every pipeline run in the process. It self-initializes
+// lazily as bridges are first used; there is no explicit start/close lifecycle.
+type bridgeConnManager struct {
+	mu    sync.RWMutex
+	cache map[[32]byte][]byte
+
+	connsMu sync.Mutex
+	conns   map[string]*eaConn // bridge name -> EAConn
+	lggr    logger.Logger      // guarded by connsMu; set at most once, from NewBridgeConnManager
+
+	dial eaStreamDialer
+}
+
+var defaultBridgeConnManager BridgeConnManager = &bridgeConnManager{
+	cache: make(map[[32]byte][]byte),
+	conns: make(map[string]*eaConn),
+	lggr:  logger.Nop(),
+	dial:  dialGRPCStream,
+}
+
+// NewBridgeConnManager returns the package-level singleton. Passing a logger sets
+// it on the singleton for use by lazily-created EAConns; it's expected to be
+// called once, from PipelineRunner startup, with all other call sites (fallback
+// construction, tests) using the zero-arg form and getting whatever logger (or
+// the Nop default) is already set.
+func NewBridgeConnManager(lggr ...logger.Logger) BridgeConnManager {
+	m := defaultBridgeConnManager.(*bridgeConnManager)
+	if len(lggr) > 0 && lggr[0] != nil {
+		m.connsMu.Lock()
+		m.lggr = lggr[0]
+		m.connsMu.Unlock()
+	}
+	return m
+}
+
+func (m *bridgeConnManager) GetObservation(bridge bridges.BridgeType, requestData map[string]any) ([]byte, error) {
+	bridgeName := strings.TrimPrefix(bridge.Name.String(), "bridge-")
+	data, err := subscriptionData(requestData)
+	if err != nil {
+		return nil, fmt.Errorf("bridge %q: %w", bridgeName, err)
+	}
+	key, err := bridgeObservationCacheKey(bridgeName, data)
+	if err != nil {
+		return nil, err
+	}
+	m.lggr.Debugw("cache key generated", "key", hex.EncodeToString(key[:]), "bridge", bridgeName, "data", data)
+	subscription, err := structpb.NewStruct(data)
+	if err != nil {
+		return nil, fmt.Errorf("failed to build subscription payload for bridge %q: %w", bridgeName, err)
+	}
+	m.getOrCreateConn(bridgeName, bridge.URL).registerAsset(key, subscription)
+
+	m.mu.RLock()
+	entry, ok := m.cache[key]
+	m.mu.RUnlock()
+	if !ok {
+		return nil, fmt.Errorf("%w for bridge %q", ErrBridgeObservationNotFound, bridgeName)
+	}
+	payload := make([]byte, len(entry))
+	copy(payload, entry)
+	return payload, nil
+}
+
+// PutObservation stores observation bytes under the given payload hash key.
+// It is used by EAConn receiver loops and white-box tests in this package.
+func (m *bridgeConnManager) PutObservation(key [32]byte, observation []byte) {
+	payload := make([]byte, len(observation))
+	copy(payload, observation)
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.cache[key] = payload
+}
+
+// SeedObservation computes the bridge observation key from request data and
+// stores a cache entry. This is intended for tests.
+func (m *bridgeConnManager) SeedObservation(bridge bridges.BridgeType, requestData map[string]any, observation []byte) error {
+	data, err := subscriptionData(requestData)
+	if err != nil {
+		return err
+	}
+	key, err := bridgeObservationCacheKey(strings.TrimPrefix(bridge.Name.String(), "bridge-"), data)
+	if err != nil {
+		return err
+	}
+	m.PutObservation(key, observation)
+	return nil
+}
+
+// getOrCreateConn returns the bridge's persistent EAConn, lazily creating and
+// starting it on first use.
+func (m *bridgeConnManager) getOrCreateConn(bridgeName string, bridgeURL models.WebURL) *eaConn {
+	m.connsMu.Lock()
+	defer m.connsMu.Unlock()
+	if conn, ok := m.conns[bridgeName]; ok {
+		return conn
+	}
+
+	conn := newEAConn(bridgeName, bridgeURL, m)
+	m.conns[bridgeName] = conn
+	conn.start()
+	return conn
+}
+
+var errStreamDialingDisabledForTest = stdErrors.New("EAConn stream dialing disabled for test")
+
+// DisableEAConnDialingForTest replaces the manager's stream dialer with one that
+// fails immediately without any network I/O, for tests that seed the observation
+// cache directly and must not depend on a real streams-adapter connection. It
+// mutates the shared package-level singleton and is intended for test setup only.
+func (m *bridgeConnManager) DisableEAConnDialingForTest() {
+	m.connsMu.Lock()
+	defer m.connsMu.Unlock()
+	m.dial = func(_ context.Context, _ string, _ bool) (eaStreamClient, error) {
+		return nil, errStreamDialingDisabledForTest
+	}
+}
+
+// subscriptionData extracts the inner "data" object from a bridge task's request
+// payload: the only part sent as Subscription.Data and the only part the adapter
+// hashes (see ObservationPayloadHash on the streams-adapter side).
+func subscriptionData(requestData map[string]any) (map[string]any, error) {
+	data, ok := requestData["data"].(map[string]any)
+	if !ok || len(data) == 0 {
+		return nil, stdErrors.New("request data is missing a non-empty \"data\" field required for subscription")
+	}
+	return data, nil
+}
+
+// bridgeObservationCacheKey mirrors the streams-adapter's own ObservationPayloadHash.
+// The adapter is configured with its own adapterName equal to this bridge's name,
+// so payload_hash on an accepted observation equals this same key.
+func bridgeObservationCacheKey(bridgeName string, data map[string]any) ([32]byte, error) {
+	lookupBytes, err := json.Marshal(data)
+	if err != nil {
+		return [32]byte{}, fmt.Errorf("failed to marshal bridge lookup payload: %w", err)
+	}
+	b := make([]byte, 0, len(bridgeName)+len(lookupBytes))
+	b = append(b, bridgeName...)
+	b = append(b, lookupBytes...)
+	return sha256.Sum256(b), nil
+}