diff --git a/GNUmakefile b/GNUmakefile index a847688f7f8..d81ba831df7 100644 --- a/GNUmakefile +++ b/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 diff --git a/core/bridges/bridge_type.go b/core/bridges/bridge_type.go index b935cfe8318..46ec666e217 100644 --- a/core/bridges/bridge_type.go +++ b/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 } diff --git a/core/bridges/orm.go b/core/bridges/orm.go index b20c95f161d..e315c0bdc64 100644 --- a/core/bridges/orm.go +++ b/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 } diff --git a/core/internal/cltest/factories.go b/core/internal/cltest/factories.go index 8644ac0e308..390044ee970 100644 --- a/core/internal/cltest/factories.go +++ b/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{}) } diff --git a/core/services/pipeline/bridgeconn/bridge_conn_manager.go b/core/services/pipeline/bridgeconn/bridge_conn_manager.go new file mode 100644 index 00000000000..5d7f922199b --- /dev/null +++ b/core/services/pipeline/bridgeconn/bridge_conn_manager.go @@ -0,0 +1,190 @@ +package bridgeconn + +import ( + "context" + "crypto/sha256" + "encoding/hex" + stdErrors "errors" + "fmt" + "strings" + "sync" + "time" + + "github.com/goccy/go-json" + "github.com/jonboulle/clockwork" + "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") + ErrBridgeObservationExpired = stdErrors.New("bridge observation expired") +) + +// observationTTL bounds how long a cached observation may be served before it is +// treated as stale. Hardcoded for now; may become configurable later. +const observationTTL = 5 * time.Second + +// cacheEntry pairs a cached observation with the time it was stored, so +// GetObservation can reject entries older than observationTTL. +type cacheEntry struct { + payload []byte + storedAt time.Time +} + +// 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]cacheEntry + + connsMu sync.Mutex + conns map[string]*eaConn // bridge name -> EAConn + lggr logger.Logger // guarded by connsMu; set at most once, from NewBridgeConnManager + + dial eaStreamDialer + clock clockwork.Clock +} + +var defaultBridgeConnManager BridgeConnManager = &bridgeConnManager{ + cache: make(map[[32]byte]cacheEntry), + conns: make(map[string]*eaConn), + lggr: logger.Nop(), + dial: dialGRPCStream, + clock: clockwork.NewRealClock(), +} + +// 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) + } + if m.clock.Now().Sub(entry.storedAt) > observationTTL { + return nil, fmt.Errorf("%w for bridge %q", ErrBridgeObservationExpired, bridgeName) + } + payload := make([]byte, len(entry.payload)) + copy(payload, entry.payload) + 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] = cacheEntry{payload: payload, storedAt: m.clock.Now()} +} + +// 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 +} diff --git a/core/services/pipeline/bridgeconn/bridge_conn_manager_test.go b/core/services/pipeline/bridgeconn/bridge_conn_manager_test.go new file mode 100644 index 00000000000..c25cf2fa3fd --- /dev/null +++ b/core/services/pipeline/bridgeconn/bridge_conn_manager_test.go @@ -0,0 +1,244 @@ +package bridgeconn + +import ( + "context" + "encoding/hex" + "net/url" + "sync" + "testing" + "time" + + "github.com/jonboulle/clockwork" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "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/services/pipeline/bridgeconn/streamspb" + "github.com/smartcontractkit/chainlink/v2/core/store/models" +) + +func testBridge(t *testing.T, name string) bridges.BridgeType { + t.Helper() + u, err := url.Parse("http://" + name + ".example.invalid:8080") + require.NoError(t, err) + return bridges.BridgeType{ + Name: bridges.BridgeName(name), + URL: models.WebURL(*u), + UseConnectionManager: true, + } +} + +func newTestManager() *bridgeConnManager { + return &bridgeConnManager{ + cache: make(map[[32]byte]cacheEntry), + conns: make(map[string]*eaConn), + lggr: logger.Nop(), + dial: func(_ context.Context, _ string, _ bool) (eaStreamClient, error) { + return nil, errStreamDialingDisabledForTest + }, + clock: clockwork.NewRealClock(), + } +} + +func TestBridgeConnManager_GetOrCreateConn_OnePerBridge(t *testing.T) { + t.Parallel() + m := newTestManager() + bridgeA := testBridge(t, "bridgea") + bridgeB := testBridge(t, "bridgeb") + + var wg sync.WaitGroup + conns := make([]*eaConn, 20) + for idx := range 10 { + wg.Add(2) + go func() { + defer wg.Done() + conns[idx] = m.getOrCreateConn(bridgeA.Name.String(), bridgeA.URL) + }() + go func() { + defer wg.Done() + conns[idx+10] = m.getOrCreateConn(bridgeB.Name.String(), bridgeB.URL) + }() + } + wg.Wait() + + for i := 1; i < 10; i++ { + assert.Same(t, conns[0], conns[i], "all concurrent lookups for bridgeA must return the same EAConn") + } + for i := 11; i < 20; i++ { + assert.Same(t, conns[10], conns[i], "all concurrent lookups for bridgeB must return the same EAConn") + } + assert.NotSame(t, conns[0], conns[10], "separate bridges must get separate EAConns") + + m.connsMu.Lock() + assert.Len(t, m.conns, 2) + m.connsMu.Unlock() +} + +func TestEAConn_RegisterAsset_RefreshAndIdlePrune(t *testing.T) { + t.Parallel() + m := newTestManager() + bridge := testBridge(t, "idlebridge") + conn := newEAConn(bridge.Name.String(), bridge.URL, m) + clock := clockwork.NewFakeClock() + conn.clock = clock + + payload, err := structpb.NewStruct(map[string]any{"endpoint": "crypto"}) + require.NoError(t, err) + + var keyA, keyB [32]byte + keyA[0] = 1 + keyB[0] = 2 + conn.registerAsset(keyA, payload) + conn.registerAsset(keyB, payload) + + // Neither asset is idle yet: both survive a snapshot. + active, req := conn.pruneAndSnapshot() + assert.Len(t, active, 2) + assert.Len(t, req.Subscriptions, 2) + + // Refresh keyA only, then advance past the idle timeout. + clock.Advance(assetIdleTimeout / 2) + conn.registerAsset(keyA, payload) + clock.Advance(assetIdleTimeout/2 + time.Millisecond) + + active, req = conn.pruneAndSnapshot() + assert.Len(t, active, 1) + assert.Len(t, req.Subscriptions, 1) + _, aStillActive := active[keyA] + assert.True(t, aStillActive, "refreshed asset must survive the idle prune") + _, bStillActive := active[keyB] + assert.False(t, bStillActive, "unrefreshed asset must be pruned (indirect unsubscribe)") +} + +func TestEAConn_HandleObservation(t *testing.T) { + t.Parallel() + m := newTestManager() + bridge := testBridge(t, "obsbridge") + conn := newEAConn(bridge.Name.String(), bridge.URL, m) + + payload, err := structpb.NewStruct(map[string]any{"endpoint": "crypto"}) + require.NoError(t, err) + var key [32]byte + key[0] = 42 + conn.registerAsset(key, payload) + + assetKey := hex.EncodeToString(key[:]) + observationsMetric := func() float64 { + return testutil.ToFloat64(promEAConnObservationsTotal.WithLabelValues(conn.bridgeName, assetKey)) + } + + // Registered key: cached, and the per-asset counter is incremented. + conn.handleObservation(&streamspb.SubscribeResponse{ + PayloadHash: key[:], + ObservationJson: []byte(`{"result":"1"}`), + }) + m.mu.RLock() + cached, ok := m.cache[key] + m.mu.RUnlock() + require.True(t, ok) + assert.JSONEq(t, `{"result":"1"}`, string(cached.payload)) + assert.InEpsilon(t, float64(1), observationsMetric(), 0) + + // Unregistered key: discarded, counter unchanged. + var unknownKey [32]byte + unknownKey[0] = 99 + conn.handleObservation(&streamspb.SubscribeResponse{ + PayloadHash: unknownKey[:], + ObservationJson: []byte(`{"result":"2"}`), + }) + m.mu.RLock() + _, unknownCached := m.cache[unknownKey] + m.mu.RUnlock() + assert.False(t, unknownCached, "observation for an unregistered key must not be cached") + assert.InEpsilon(t, float64(1), observationsMetric(), 0, "counter must not increment for an unregistered key") + + // Malformed (wrong-length) payload_hash: discarded, no panic, counter unchanged. + conn.handleObservation(&streamspb.SubscribeResponse{ + PayloadHash: []byte{1, 2, 3}, + ObservationJson: []byte(`{"result":"3"}`), + }) + assert.InEpsilon(t, float64(1), observationsMetric(), 0, "counter must not increment for a malformed payload_hash") +} + +func TestNextBackoff(t *testing.T) { + t.Parallel() + d := reconnectBackoffInitial + seen := make([]time.Duration, 0, 11) + seen = append(seen, d) + for range 10 { + d = nextBackoff(d) + seen = append(seen, d) + } + for i := 1; i < len(seen); i++ { + assert.LessOrEqual(t, seen[i-1], seen[i], "backoff must never decrease") + assert.LessOrEqual(t, seen[i], reconnectBackoffMax, "backoff must never exceed the configured max") + } + assert.Equal(t, reconnectBackoffMax, seen[len(seen)-1], "backoff must saturate at the max") +} + +func TestBridgeConnManager_GetObservation_CacheHitAndMiss(t *testing.T) { + t.Parallel() + m := newTestManager() + bridge := testBridge(t, "cachebridge") + requestData := map[string]any{"data": map[string]any{"endpoint": "crypto"}} + + _, err := m.GetObservation(bridge, requestData) + require.ErrorIs(t, err, ErrBridgeObservationNotFound) + + key, err := bridgeObservationCacheKey(bridge.Name.String(), requestData["data"].(map[string]any)) + require.NoError(t, err) + + m.PutObservation(key, []byte(`{"result":"9700"}`)) + got, err := m.GetObservation(bridge, requestData) + require.NoError(t, err) + assert.JSONEq(t, `{"result":"9700"}`, string(got)) + + // GetObservation must still have registered the asset with the bridge's EAConn. + m.connsMu.Lock() + conn, ok := m.conns[bridge.Name.String()] + m.connsMu.Unlock() + require.True(t, ok) + conn.mu.Lock() + _, registered := conn.assets[key] + conn.mu.Unlock() + assert.True(t, registered) +} + +func TestBridgeConnManager_GetObservation_ExpiresAfterTTL(t *testing.T) { + t.Parallel() + m := newTestManager() + clock := clockwork.NewFakeClock() + m.clock = clock + bridge := testBridge(t, "ttlbridge") + requestData := map[string]any{"data": map[string]any{"endpoint": "crypto"}} + + key, err := bridgeObservationCacheKey(bridge.Name.String(), requestData["data"].(map[string]any)) + require.NoError(t, err) + + m.PutObservation(key, []byte(`{"result":"9700"}`)) + + clock.Advance(observationTTL) + got, err := m.GetObservation(bridge, requestData) + require.NoError(t, err, "an entry at exactly the TTL boundary must not be treated as expired") + assert.JSONEq(t, `{"result":"9700"}`, string(got)) + + clock.Advance(time.Nanosecond) + _, err = m.GetObservation(bridge, requestData) + require.ErrorIs(t, err, ErrBridgeObservationExpired) +} + +func TestBridgeConnManager_GetObservation_MissingDataField(t *testing.T) { + t.Parallel() + m := newTestManager() + bridge := testBridge(t, "nodatabridge") + + _, err := m.GetObservation(bridge, map[string]any{"foo": "bar"}) + require.Error(t, err, "requestData without a \"data\" field must be rejected, not silently subscribed") + + _, err = m.GetObservation(bridge, map[string]any{"data": map[string]any{}}) + assert.Error(t, err, "an empty \"data\" field must be rejected") +} diff --git a/core/services/pipeline/bridgeconn/eaconn.go b/core/services/pipeline/bridgeconn/eaconn.go new file mode 100644 index 00000000000..cc4a9dac88b --- /dev/null +++ b/core/services/pipeline/bridgeconn/eaconn.go @@ -0,0 +1,302 @@ +package bridgeconn + +import ( + "context" + "encoding/hex" + "net/url" + "sync" + "time" + + "github.com/goccy/go-json" + "github.com/jonboulle/clockwork" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "google.golang.org/protobuf/types/known/structpb" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + + "github.com/smartcontractkit/chainlink/v2/core/services/pipeline/bridgeconn/streamspb" + "github.com/smartcontractkit/chainlink/v2/core/store/models" +) + +// Operational timing and reconnect behavior are hardcoded: this manager has no +// configuration section and EAConn takes no configuration interface. +const ( + subscriptionInterval = 10 * time.Second + assetIdleTimeout = 30 * time.Second + reconnectBackoffInitial = time.Second + reconnectBackoffMultiplier = 2.0 + reconnectBackoffMax = time.Minute +) + +// promEAConnObservationsTotal counts accepted gRPC observation messages per bridge +// and asset pair (the hex-encoded bridgeObservationCacheKey). Rate per second is +// derived externally, e.g. via PromQL rate(bridge_eaconn_observations_total[1m]). +var promEAConnObservationsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "bridge_eaconn_observations_total", + Help: "Count of gRPC observation messages accepted per bridge and asset pair", +}, []string{"bridgeName", "assetKey"}) + +// promEAConnTransmitDuration measures the delay between the external adapter receiving +// data from its upstream provider (timestamps.providerDataReceivedUnixMs in the +// observation payload) and this node handling the gRPC message. +var promEAConnTransmitDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Name: "bridge_eaconn_transmit_duration_seconds", + Help: "Delay from the adapter receiving provider data to the node handling the gRPC observation", + Buckets: []float64{ + 0.001, // 1 ms + 0.002, // 2 ms + 0.005, // 5 ms + 0.010, // 10 ms + 0.015, // 15 ms + 0.020, // 20 ms + 0.030, // 30 ms + 0.040, // 40 ms + 0.050, // 50 ms + 0.075, // 75 ms + 0.100, // 100 ms + 0.150, // 150 ms + 0.200, // 200 ms + 0.500, // 500 ms + }, +}, []string{"bridgeName"}) + +// promEAConnTransmitSkewTotal counts observations excluded from the transmit-duration +// histogram because now minus providerDataReceivedUnixMs came out negative, meaning the +// adapter's clock runs ahead of this node's. Skew of tens of milliseconds is normal +// between unsynchronized hosts and exceeds the latency the histogram tries to measure, +// so such samples are dropped rather than folded into the lowest bucket. Compare this +// against bridge_eaconn_observations_total to judge how much of a bridge's traffic the +// histogram actually covers: if this tracks the observation rate, the histogram is empty +// for that bridge and its clocks need synchronizing before the latency data means +// anything. +var promEAConnTransmitSkewTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "bridge_eaconn_transmit_skew_total", + Help: "Count of observations excluded from the transmit-duration histogram due to adapter clock skew", +}, []string{"bridgeName"}) + +// observationTimestamps is the subset of an observation payload EAConn reads for the +// transmit-duration metric; all other payload fields are ignored. +type observationTimestamps struct { + Timestamps struct { + ProviderDataReceivedUnixMs *int64 `json:"providerDataReceivedUnixMs"` + } `json:"timestamps"` +} + +// eaStreamClient is the protobuf-independent contract EAConn depends on for its +// bidirectional observation stream, implemented by grpcStreamClient in production +// and by fakes in tests. +type eaStreamClient interface { + Send(*streamspb.SubscribeRequest) error + Recv() (*streamspb.SubscribeResponse, error) + Close() error +} + +// eaStreamDialer opens a new stream to target (host:port derived from the bridge's +// URL authority). useTLS selects TLS (bridge URL scheme https) versus plaintext gRPC +// (http). +type eaStreamDialer func(ctx context.Context, target string, useTLS bool) (eaStreamClient, error) + +// eaAsset is a bridge's registered asset: an immutable subscription payload plus the +// timestamp it was last requested through GetObservation. +type eaAsset struct { + payload *structpb.Struct + lastUsed time.Time +} + +// eaConn is the single persistent connection owned by BridgeConnManager for one +// bridge name. It sends complete active-asset snapshots on a fixed interval, +// applies indirect unsubscribe by omitting idle assets from the next snapshot, and +// reconnects with fixed exponential backoff on any dial/send/receive failure. +type eaConn struct { + bridgeName string + target string + useTLS bool + dial eaStreamDialer + lggr logger.Logger + manager *bridgeConnManager + clock clockwork.Clock + + mu sync.Mutex + assets map[[32]byte]*eaAsset + + startOnce sync.Once +} + +// newEAConn is called with manager.connsMu already held, so reading manager.lggr +// here is safe without a separate lock. +func newEAConn(bridgeName string, bridgeURL models.WebURL, manager *bridgeConnManager) *eaConn { + u := url.URL(bridgeURL) + return &eaConn{ + bridgeName: bridgeName, + target: u.Host, + useTLS: u.Scheme == "https", + dial: manager.dial, + lggr: logger.With(logger.Named(manager.lggr, "EAConn"), "bridgeName", bridgeName), + manager: manager, + clock: clockwork.NewRealClock(), + assets: make(map[[32]byte]*eaAsset), + } +} + +func (c *eaConn) start() { + c.startOnce.Do(func() { + go c.run(context.Background()) + }) +} + +// registerAsset records or refreshes the lease for key, lazily adding it to the next +// snapshot. Asset changes never trigger an immediate send. +func (c *eaConn) registerAsset(key [32]byte, payload *structpb.Struct) { + c.mu.Lock() + defer c.mu.Unlock() + if asset, ok := c.assets[key]; ok { + asset.lastUsed = c.clock.Now() + return + } + c.assets[key] = &eaAsset{payload: payload, lastUsed: c.clock.Now()} + c.lggr.Debugw("EAConn: registered asset", "key", hex.EncodeToString(key[:])) +} + +// pruneAndSnapshot removes assets idle for at least assetIdleTimeout (indirect +// unsubscribe by omission) and returns a full-snapshot request for what remains. +func (c *eaConn) pruneAndSnapshot() (map[[32]byte]struct{}, *streamspb.SubscribeRequest) { + now := c.clock.Now() + c.mu.Lock() + defer c.mu.Unlock() + for key, asset := range c.assets { + if now.Sub(asset.lastUsed) >= assetIdleTimeout { + delete(c.assets, key) + } + } + keys := make(map[[32]byte]struct{}, len(c.assets)) + req := &streamspb.SubscribeRequest{Subscriptions: make([]*streamspb.Subscription, 0, len(c.assets))} + for key, asset := range c.assets { + keys[key] = struct{}{} + req.Subscriptions = append(req.Subscriptions, &streamspb.Subscription{Data: asset.payload}) + } + return keys, req +} + +// run owns the connect/send/receive/reconnect lifecycle for the process lifetime; +// there is no explicit stop, since this manager has no runner-owned lifecycle. +func (c *eaConn) run(ctx context.Context) { + backoff := reconnectBackoffInitial + for { + stream, err := c.dial(ctx, c.target, c.useTLS) + if err != nil { + c.lggr.Errorw("EAConn: dial failed", "target", c.target, "err", err) + backoff = c.sleepBackoff(backoff) + continue + } + c.lggr.Infow("EAConn: stream opened", "target", c.target) + backoff = reconnectBackoffInitial + + streamErr := c.serve(stream) + _ = stream.Close() + c.lggr.Warnw("EAConn: stream closed, reconnecting", "target", c.target, "err", streamErr) + backoff = c.sleepBackoff(backoff) + } +} + +// serve runs the sender ticker and receiver loop against one open stream until +// either fails, returning the failure. +func (c *eaConn) serve(stream eaStreamClient) error { + errCh := make(chan error, 2) + done := make(chan struct{}) + defer close(done) + + go func() { + errCh <- c.recvLoop(stream, done) + }() + go func() { + errCh <- c.sendLoop(stream, done) + }() + return <-errCh +} + +func (c *eaConn) sendLoop(stream eaStreamClient, done <-chan struct{}) error { + ticker := c.clock.NewTicker(subscriptionInterval) + defer ticker.Stop() + for { + select { + case <-done: + return nil + case <-ticker.Chan(): + _, req := c.pruneAndSnapshot() + if err := stream.Send(req); err != nil { + return err + } + } + } +} + +func (c *eaConn) recvLoop(stream eaStreamClient, done <-chan struct{}) error { + for { + resp, err := stream.Recv() + if err != nil { + return err + } + c.handleObservation(resp) + select { + case <-done: + return nil + default: + } + } +} + +// handleObservation accepts an observation only for a key currently registered to +// this EAConn (payload_hash is expected to equal bridgeObservationCacheKey of the +// subscription payload we sent); unregistered or malformed keys are discarded. +func (c *eaConn) handleObservation(resp *streamspb.SubscribeResponse) { + if len(resp.PayloadHash) != 32 { + c.lggr.Warnw("EAConn: discarding observation with malformed payload_hash", "len", len(resp.PayloadHash)) + return + } + var key [32]byte + copy(key[:], resp.PayloadHash) + + c.mu.Lock() + _, registered := c.assets[key] + c.mu.Unlock() + if !registered { + c.lggr.Debugw("EAConn: discarding observation for unregistered key", "payloadHash", hex.EncodeToString(key[:])) + return + } + promEAConnObservationsTotal.WithLabelValues(c.bridgeName, hex.EncodeToString(key[:])).Inc() + c.observeTransmitDuration(resp.ObservationJson) + c.manager.PutObservation(key, resp.ObservationJson) +} + +// observeTransmitDuration records now minus providerDataReceivedUnixMs. Payloads that +// omit the timestamp are skipped silently; negative durations are counted in +// promEAConnTransmitSkewTotal instead, since they reflect adapter clock skew rather than +// transmission time. Both branches are per-message hot paths at a few thousand +// observations per second, so neither logs: the counters carry the signal. +func (c *eaConn) observeTransmitDuration(observationJSON []byte) { + var obs observationTimestamps + if err := json.Unmarshal(observationJSON, &obs); err != nil { + return + } + if obs.Timestamps.ProviderDataReceivedUnixMs == nil { + return + } + received := time.UnixMilli(*obs.Timestamps.ProviderDataReceivedUnixMs) + d := c.clock.Now().Sub(received) + if d < 0 { + promEAConnTransmitSkewTotal.WithLabelValues(c.bridgeName).Inc() + return + } + promEAConnTransmitDuration.WithLabelValues(c.bridgeName).Observe(d.Seconds()) +} + +func (c *eaConn) sleepBackoff(current time.Duration) time.Duration { + c.clock.Sleep(current) + return nextBackoff(current) +} + +func nextBackoff(current time.Duration) time.Duration { + next := time.Duration(float64(current) * reconnectBackoffMultiplier) + return min(next, reconnectBackoffMax) +} diff --git a/core/services/pipeline/bridgeconn/grpc_stream_client.go b/core/services/pipeline/bridgeconn/grpc_stream_client.go new file mode 100644 index 00000000000..f3aed78a4c7 --- /dev/null +++ b/core/services/pipeline/bridgeconn/grpc_stream_client.go @@ -0,0 +1,51 @@ +package bridgeconn + +import ( + "context" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" + + "github.com/smartcontractkit/chainlink/v2/core/services/pipeline/bridgeconn/streamspb" +) + +// grpcStreamClient adapts streamspb's generated Subscribe stream to eaStreamClient. +type grpcStreamClient struct { + conn *grpc.ClientConn + stream streamspb.StreamService_SubscribeClient +} + +// dialGRPCStream opens a gRPC connection to target and starts the bidirectional +// Subscribe stream. When useTLS is true (bridge URL scheme is https) the connection +// uses TLS with the host's system root CAs; otherwise it is plaintext. Neither mode +// uses application tokens or client certificates. +func dialGRPCStream(ctx context.Context, target string, useTLS bool) (eaStreamClient, error) { + creds := insecure.NewCredentials() + if useTLS { + creds = credentials.NewClientTLSFromCert(nil, "") + } + conn, err := grpc.NewClient(target, grpc.WithTransportCredentials(creds)) + if err != nil { + return nil, err + } + stream, err := streamspb.NewStreamServiceClient(conn).Subscribe(ctx) + if err != nil { + _ = conn.Close() + return nil, err + } + return &grpcStreamClient{conn: conn, stream: stream}, nil +} + +func (c *grpcStreamClient) Send(req *streamspb.SubscribeRequest) error { + return c.stream.Send(req) +} + +func (c *grpcStreamClient) Recv() (*streamspb.SubscribeResponse, error) { + return c.stream.Recv() +} + +func (c *grpcStreamClient) Close() error { + _ = c.stream.CloseSend() + return c.conn.Close() +} diff --git a/core/services/pipeline/bridgeconn/streamspb/generate.go b/core/services/pipeline/bridgeconn/streamspb/generate.go new file mode 100644 index 00000000000..18a73a20300 --- /dev/null +++ b/core/services/pipeline/bridgeconn/streamspb/generate.go @@ -0,0 +1,6 @@ +// Package streamspb holds the generated types for streams.proto (the streams-adapter +// wire contract). +package streamspb + +//go:generate protoc --go_out=. --go_opt=paths=source_relative streams.proto +//go:generate protoc --go-grpc_out=. --go-grpc_opt=paths=source_relative streams.proto diff --git a/core/services/pipeline/bridgeconn/streamspb/streams.pb.go b/core/services/pipeline/bridgeconn/streamspb/streams.pb.go new file mode 100644 index 00000000000..df9dcd7055a --- /dev/null +++ b/core/services/pipeline/bridgeconn/streamspb/streams.pb.go @@ -0,0 +1,243 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc v5.29.3 +// source: streams.proto + +package streamspb + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + structpb "google.golang.org/protobuf/types/known/structpb" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type SubscribeRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Subscriptions []*Subscription `protobuf:"bytes,1,rep,name=subscriptions,proto3" json:"subscriptions,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *SubscribeRequest) Reset() { + *x = SubscribeRequest{} + mi := &file_streams_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *SubscribeRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*SubscribeRequest) ProtoMessage() {} + +func (x *SubscribeRequest) ProtoReflect() protoreflect.Message { + mi := &file_streams_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use SubscribeRequest.ProtoReflect.Descriptor instead. +func (*SubscribeRequest) Descriptor() ([]byte, []int) { + return file_streams_proto_rawDescGZIP(), []int{0} +} + +func (x *SubscribeRequest) GetSubscriptions() []*Subscription { + if x != nil { + return x.Subscriptions + } + return nil +} + +type Subscription struct { + state protoimpl.MessageState `protogen:"open.v1"` + Data *structpb.Struct `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Subscription) Reset() { + *x = Subscription{} + mi := &file_streams_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Subscription) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Subscription) ProtoMessage() {} + +func (x *Subscription) ProtoReflect() protoreflect.Message { + mi := &file_streams_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Subscription.ProtoReflect.Descriptor instead. +func (*Subscription) Descriptor() ([]byte, []int) { + return file_streams_proto_rawDescGZIP(), []int{1} +} + +func (x *Subscription) GetData() *structpb.Struct { + if x != nil { + return x.Data + } + return nil +} + +type SubscribeResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + Timestamp string `protobuf:"bytes,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"` + ObservationJson []byte `protobuf:"bytes,3,opt,name=observation_json,json=observationJson,proto3" json:"observation_json,omitempty"` + PayloadHash []byte `protobuf:"bytes,4,opt,name=payload_hash,json=payloadHash,proto3" json:"payload_hash,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *SubscribeResponse) Reset() { + *x = SubscribeResponse{} + mi := &file_streams_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *SubscribeResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*SubscribeResponse) ProtoMessage() {} + +func (x *SubscribeResponse) ProtoReflect() protoreflect.Message { + mi := &file_streams_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use SubscribeResponse.ProtoReflect.Descriptor instead. +func (*SubscribeResponse) Descriptor() ([]byte, []int) { + return file_streams_proto_rawDescGZIP(), []int{2} +} + +func (x *SubscribeResponse) GetTimestamp() string { + if x != nil { + return x.Timestamp + } + return "" +} + +func (x *SubscribeResponse) GetObservationJson() []byte { + if x != nil { + return x.ObservationJson + } + return nil +} + +func (x *SubscribeResponse) GetPayloadHash() []byte { + if x != nil { + return x.PayloadHash + } + return nil +} + +var File_streams_proto protoreflect.FileDescriptor + +const file_streams_proto_rawDesc = "" + + "\n" + + "\rstreams.proto\x12\n" + + "streams.v1\x1a\x1cgoogle/protobuf/struct.proto\"R\n" + + "\x10SubscribeRequest\x12>\n" + + "\rsubscriptions\x18\x01 \x03(\v2\x18.streams.v1.SubscriptionR\rsubscriptions\";\n" + + "\fSubscription\x12+\n" + + "\x04data\x18\x01 \x01(\v2\x17.google.protobuf.StructR\x04data\"\x7f\n" + + "\x11SubscribeResponse\x12\x1c\n" + + "\ttimestamp\x18\x02 \x01(\tR\ttimestamp\x12)\n" + + "\x10observation_json\x18\x03 \x01(\fR\x0fobservationJson\x12!\n" + + "\fpayload_hash\x18\x04 \x01(\fR\vpayloadHash2_\n" + + "\rStreamService\x12N\n" + + "\tSubscribe\x12\x1c.streams.v1.SubscribeRequest\x1a\x1d.streams.v1.SubscribeResponse\"\x00(\x010\x01BVZTgithub.com/smartcontractkit/chainlink/v2/core/services/pipeline/bridgeconn/streamspbb\x06proto3" + +var ( + file_streams_proto_rawDescOnce sync.Once + file_streams_proto_rawDescData []byte +) + +func file_streams_proto_rawDescGZIP() []byte { + file_streams_proto_rawDescOnce.Do(func() { + file_streams_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_streams_proto_rawDesc), len(file_streams_proto_rawDesc))) + }) + return file_streams_proto_rawDescData +} + +var file_streams_proto_msgTypes = make([]protoimpl.MessageInfo, 3) +var file_streams_proto_goTypes = []any{ + (*SubscribeRequest)(nil), // 0: streams.v1.SubscribeRequest + (*Subscription)(nil), // 1: streams.v1.Subscription + (*SubscribeResponse)(nil), // 2: streams.v1.SubscribeResponse + (*structpb.Struct)(nil), // 3: google.protobuf.Struct +} +var file_streams_proto_depIdxs = []int32{ + 1, // 0: streams.v1.SubscribeRequest.subscriptions:type_name -> streams.v1.Subscription + 3, // 1: streams.v1.Subscription.data:type_name -> google.protobuf.Struct + 0, // 2: streams.v1.StreamService.Subscribe:input_type -> streams.v1.SubscribeRequest + 2, // 3: streams.v1.StreamService.Subscribe:output_type -> streams.v1.SubscribeResponse + 3, // [3:4] is the sub-list for method output_type + 2, // [2:3] is the sub-list for method input_type + 2, // [2:2] is the sub-list for extension type_name + 2, // [2:2] is the sub-list for extension extendee + 0, // [0:2] is the sub-list for field type_name +} + +func init() { file_streams_proto_init() } +func file_streams_proto_init() { + if File_streams_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_streams_proto_rawDesc), len(file_streams_proto_rawDesc)), + NumEnums: 0, + NumMessages: 3, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_streams_proto_goTypes, + DependencyIndexes: file_streams_proto_depIdxs, + MessageInfos: file_streams_proto_msgTypes, + }.Build() + File_streams_proto = out.File + file_streams_proto_goTypes = nil + file_streams_proto_depIdxs = nil +} diff --git a/core/services/pipeline/bridgeconn/streamspb/streams.proto b/core/services/pipeline/bridgeconn/streamspb/streams.proto new file mode 100644 index 00000000000..d25683b9988 --- /dev/null +++ b/core/services/pipeline/bridgeconn/streamspb/streams.proto @@ -0,0 +1,26 @@ +syntax = "proto3"; + +package streams.v1; + +import "google/protobuf/struct.proto"; + +option go_package = "github.com/smartcontractkit/chainlink/v2/core/services/pipeline/bridgeconn/streamspb"; + +service StreamService { + // Each client request replaces the complete subscription set. + rpc Subscribe(stream SubscribeRequest) returns (stream SubscribeResponse) {} +} + +message SubscribeRequest { + repeated Subscription subscriptions = 1; +} + +message Subscription { + google.protobuf.Struct data = 1; +} + +message SubscribeResponse { + string timestamp = 2; + bytes observation_json = 3; + bytes payload_hash = 4; +} diff --git a/core/services/pipeline/bridgeconn/streamspb/streams_grpc.pb.go b/core/services/pipeline/bridgeconn/streamspb/streams_grpc.pb.go new file mode 100644 index 00000000000..3c1a3706ef1 --- /dev/null +++ b/core/services/pipeline/bridgeconn/streamspb/streams_grpc.pb.go @@ -0,0 +1,117 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.6.2 +// - protoc v5.29.3 +// source: streams.proto + +package streamspb + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.64.0 or later. +const _ = grpc.SupportPackageIsVersion9 + +const ( + StreamService_Subscribe_FullMethodName = "/streams.v1.StreamService/Subscribe" +) + +// StreamServiceClient is the client API for StreamService service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +type StreamServiceClient interface { + // Each client request replaces the complete subscription set. + Subscribe(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[SubscribeRequest, SubscribeResponse], error) +} + +type streamServiceClient struct { + cc grpc.ClientConnInterface +} + +func NewStreamServiceClient(cc grpc.ClientConnInterface) StreamServiceClient { + return &streamServiceClient{cc} +} + +func (c *streamServiceClient) Subscribe(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[SubscribeRequest, SubscribeResponse], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &StreamService_ServiceDesc.Streams[0], StreamService_Subscribe_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[SubscribeRequest, SubscribeResponse]{ClientStream: stream} + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StreamService_SubscribeClient = grpc.BidiStreamingClient[SubscribeRequest, SubscribeResponse] + +// StreamServiceServer is the server API for StreamService service. +// All implementations must embed UnimplementedStreamServiceServer +// for forward compatibility. +type StreamServiceServer interface { + // Each client request replaces the complete subscription set. + Subscribe(grpc.BidiStreamingServer[SubscribeRequest, SubscribeResponse]) error + mustEmbedUnimplementedStreamServiceServer() +} + +// UnimplementedStreamServiceServer must be embedded to have +// forward compatible implementations. +// +// NOTE: this should be embedded by value instead of pointer to avoid a nil +// pointer dereference when methods are called. +type UnimplementedStreamServiceServer struct{} + +func (UnimplementedStreamServiceServer) Subscribe(grpc.BidiStreamingServer[SubscribeRequest, SubscribeResponse]) error { + return status.Error(codes.Unimplemented, "method Subscribe not implemented") +} +func (UnimplementedStreamServiceServer) mustEmbedUnimplementedStreamServiceServer() {} +func (UnimplementedStreamServiceServer) testEmbeddedByValue() {} + +// UnsafeStreamServiceServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to StreamServiceServer will +// result in compilation errors. +type UnsafeStreamServiceServer interface { + mustEmbedUnimplementedStreamServiceServer() +} + +func RegisterStreamServiceServer(s grpc.ServiceRegistrar, srv StreamServiceServer) { + // If the following call panics, it indicates UnimplementedStreamServiceServer was + // embedded by pointer and is nil. This will cause panics if an + // unimplemented method is ever invoked, so we test this at initialization + // time to prevent it from happening at runtime later due to I/O. + if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { + t.testEmbeddedByValue() + } + s.RegisterService(&StreamService_ServiceDesc, srv) +} + +func _StreamService_Subscribe_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(StreamServiceServer).Subscribe(&grpc.GenericServerStream[SubscribeRequest, SubscribeResponse]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StreamService_SubscribeServer = grpc.BidiStreamingServer[SubscribeRequest, SubscribeResponse] + +// StreamService_ServiceDesc is the grpc.ServiceDesc for StreamService service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var StreamService_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "streams.v1.StreamService", + HandlerType: (*StreamServiceServer)(nil), + Methods: []grpc.MethodDesc{}, + Streams: []grpc.StreamDesc{ + { + StreamName: "Subscribe", + Handler: _StreamService_Subscribe_Handler, + ServerStreams: true, + ClientStreams: true, + }, + }, + Metadata: "streams.proto", +} diff --git a/core/services/pipeline/helpers_test.go b/core/services/pipeline/helpers_test.go index cef56bf733e..99f3958076e 100644 --- a/core/services/pipeline/helpers_test.go +++ b/core/services/pipeline/helpers_test.go @@ -8,6 +8,7 @@ import ( "github.com/smartcontractkit/chainlink-evm/pkg/chains/legacyevm" "github.com/smartcontractkit/chainlink/v2/core/bridges" + "github.com/smartcontractkit/chainlink/v2/core/services/pipeline/bridgeconn" ) const ( @@ -42,7 +43,11 @@ func (t *BridgeTask) HelperSetDependencies( t.orm = orm t.uuid = id t.httpClient = httpClient - t.specId = specId + t.specID = specId +} + +func (t *BridgeTask) HelperSetBridgeConnManager(bridgeConnManager bridgeconn.BridgeConnManager) { + t.bridgeConnManager = bridgeConnManager } func (t *HTTPTask) HelperSetDependencies(config Config, restrictedHTTPClient, unrestrictedHTTPClient *http.Client) { diff --git a/core/services/pipeline/runner.go b/core/services/pipeline/runner.go index 50ae2d7ac8e..c93a17dfc0b 100644 --- a/core/services/pipeline/runner.go +++ b/core/services/pipeline/runner.go @@ -27,6 +27,7 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/config/env" "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/recovery" + "github.com/smartcontractkit/chainlink/v2/core/services/pipeline/bridgeconn" ) type Runner interface { @@ -69,6 +70,7 @@ type runner struct { lggr logger.Logger httpClient *http.Client unrestrictedHTTPClient *http.Client + bridgeConnManager bridgeconn.BridgeConnManager // test helper runFinished func(*Run) @@ -134,6 +136,7 @@ func NewRunner( lggr: lggr, httpClient: httpClient, unrestrictedHTTPClient: unrestrictedHTTPClient, + bridgeConnManager: bridgeconn.NewBridgeConnManager(lggr), } r.runReaperWorker = commonutils.NewSleeperTask( @@ -344,11 +347,12 @@ func (r *runner) InitializePipeline(spec Spec) (pipeline *Pipeline, err error) { bt.bridgeConfig = r.bridgeConfig // orm added to BridgeTask bt.orm = r.btORM - bt.specId = spec.ID + bt.specID = spec.ID // URL is "safe" because it comes from the node's own database. We // must use the unrestrictedHTTPClient because some node operators // may run external adapters on their own hardware bt.httpClient = r.unrestrictedHTTPClient + bt.bridgeConnManager = r.bridgeConnManager bt.requiredJSONPaths = bt.getRequiredJSONPaths() case TaskTypeETHCall: task.(*ETHCallTask).legacyChains = r.legacyEVMChains diff --git a/core/services/pipeline/task.bridge.go b/core/services/pipeline/task.bridge.go index d19e0993a9d..74c0a30ea77 100644 --- a/core/services/pipeline/task.bridge.go +++ b/core/services/pipeline/task.bridge.go @@ -24,6 +24,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink/v2/core/bridges" + "github.com/smartcontractkit/chainlink/v2/core/services/pipeline/bridgeconn" "github.com/smartcontractkit/chainlink/v2/core/services/pipeline/eautils" ) @@ -85,11 +86,12 @@ type BridgeTask struct { // requiredJSONPaths). When empty or "false", that check is skipped. CheckRequired string `json:"checkRequired"` - specId int32 - orm bridges.ORM - config Config - bridgeConfig BridgeConfig - httpClient *http.Client + specID int32 + orm bridges.ORM + config Config + bridgeConfig BridgeConfig + httpClient *http.Client + bridgeConnManager bridgeconn.BridgeConnManager // requiredJSONPaths is populated in runner.InitializePipeline from strict // downstream jsonparse tasks. When CheckRequired is true and cacheTTL is set, @@ -162,10 +164,69 @@ func (t *BridgeTask) Run(ctx context.Context, lggr logger.Logger, vars Vars, inp overtimeCtx, cancel := overtimeContext(ctx) defer cancel() - url, err := t.getBridgeURLFromName(overtimeCtx, name) + bridge, err := t.getBridgeFromName(overtimeCtx, name) if err != nil { return Result{Error: err}, runInfo } + url := URLParam(bridge.URL) + lookupPayload := make(MapParam) + maps.Copy(lookupPayload, requestData) + + requestCtx, cancel := httpRequestCtx(ctx, t, t.config) + defer cancel() + if bridge.UseConnectionManager { + bridgeConnManager := t.bridgeConnManager + if bridgeConnManager == nil { + bridgeConnManager = bridgeconn.NewBridgeConnManager() + } + start := time.Now() + responseBytes, obsErr := bridgeConnManager.GetObservation(bridge, map[string]any(lookupPayload)) + finish := time.Now() + + statusCode := http.StatusOK + if obsErr != nil { + statusCode = http.StatusGatewayTimeout + } + + if telemetryCh := GetTelemetryCh(ctx); telemetryCh != nil { + requestDataJSON, jsonErr := json.Marshal(lookupPayload) + if jsonErr != nil { + lggr.Warnw("Bridge task: failed to marshal request data for telemetry", "err", jsonErr) + } + bt := &BridgeTelemetry{ + Name: t.Name, + RequestData: requestDataJSON, + ResponseData: responseBytes, + ResponseStatusCode: statusCode, + RequestStartTimestamp: start, + RequestFinishTimestamp: finish, + SpecID: t.specID, + DotID: t.DotID(), + } + if obsErr != nil { + bt.ResponseError = new(string) + *bt.ResponseError = obsErr.Error() + } + + bt.resolveStreamID(t, vars, lggr) + + select { + case telemetryCh <- bt: + default: + lggr.Warn("bridge task: telemetry channel is full, dropping telemetry") + } + } + + if obsErr != nil { + lggr.Debugw("Bridge task: connection manager request failed", + "response", string(responseBytes), + "url", url.String(), + "error", obsErr, + ) + return Result{Error: obsErr}, RunInfo{IsRetryable: true} + } + return Result{Value: string(responseBytes)}, runInfo + } requestDataJSON, err := t.finalizeAndMarshalBridgeRequestData(lggr, vars, inputValues, &requestData, includeInputAtKey) if err != nil { @@ -176,9 +237,6 @@ func (t *BridgeTask) Run(ctx context.Context, lggr logger.Logger, vars Vars, inp "url", url.String(), ) - requestCtx, cancel := httpRequestCtx(ctx, t, t.config) - defer cancel() - var cachedResponse bool responseBytes, statusCode, headers, start, finish, err := makeHTTPRequest(requestCtx, lggr, "POST", url, reqHeaders, requestData, t.httpClient, t.config.DefaultHTTPLimit()) elapsed := finish.Sub(start) @@ -205,7 +263,7 @@ func (t *BridgeTask) Run(ctx context.Context, lggr logger.Logger, vars Vars, inp RequestStartTimestamp: start, RequestFinishTimestamp: finish, LocalCacheHit: cachedResponse, - SpecID: t.specId, + SpecID: t.specID, DotID: t.DotID(), } if err != nil { @@ -252,7 +310,7 @@ func (t *BridgeTask) Run(ctx context.Context, lggr logger.Logger, vars Vars, inp } if !cachedResponse && cacheTTL > 0 { - err := t.orm.UpsertBridgeResponse(overtimeCtx, t.dotID, t.specId, responseBytes) + err := t.orm.UpsertBridgeResponse(overtimeCtx, t.dotID, t.specID, responseBytes) if err != nil { lggr.Errorw("Bridge task: failed to upsert response in bridge cache", "err", err) } @@ -370,7 +428,7 @@ func (t *BridgeTask) resolveFailureOrCache( } //nolint:gosec // disable G115 - cachedBytes, cacheErr := t.orm.GetCachedResponse(ctx, t.dotID, t.specId, time.Duration(cacheTTL)*time.Second) + cachedBytes, cacheErr := t.orm.GetCachedResponse(ctx, t.dotID, t.specID, time.Duration(cacheTTL)*time.Second) if cacheErr != nil { promBridgeCacheErrors.WithLabelValues(t.Name).Inc() if !errors.Is(cacheErr, sql.ErrNoRows) { @@ -408,12 +466,12 @@ func (bt *BridgeTelemetry) resolveStreamID(t *BridgeTask, vars Vars, lggr logger } } -func (t *BridgeTask) getBridgeURLFromName(ctx context.Context, name StringParam) (URLParam, error) { +func (t *BridgeTask) getBridgeFromName(ctx context.Context, name StringParam) (bridges.BridgeType, error) { bt, err := t.orm.FindBridge(ctx, bridges.BridgeName(name)) if err != nil { - return URLParam{}, errors.Wrapf(err, "could not find bridge with name '%s'", name) + return bridges.BridgeType{}, errors.Wrapf(err, "could not find bridge with name '%s'", name) } - return URLParam(bt.URL), nil + return bt, nil } func withRunInfo(request MapParam, meta MapParam) MapParam { diff --git a/core/services/pipeline/task.bridge_test.go b/core/services/pipeline/task.bridge_test.go index 0557d0ee48c..b19b191562d 100644 --- a/core/services/pipeline/task.bridge_test.go +++ b/core/services/pipeline/task.bridge_test.go @@ -34,6 +34,7 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/chainlink" "github.com/smartcontractkit/chainlink/v2/core/services/pipeline" + "github.com/smartcontractkit/chainlink/v2/core/services/pipeline/bridgeconn" "github.com/smartcontractkit/chainlink/v2/core/services/pipeline/eautils" "github.com/smartcontractkit/chainlink/v2/core/store/models" "github.com/smartcontractkit/chainlink/v2/core/utils" @@ -269,6 +270,61 @@ func TestBridgeTask_Happy(t *testing.T) { assert.NotEqual(t, uuid.Nil, btelem.DotID) } +func TestBridgeTask_UsesBridgeConnManagerHappyPath(t *testing.T) { + t.Parallel() + + db := pgtest.NewSqlxDB(t) + cfg := configtest.NewTestGeneralConfig(t) + + var httpCalls atomic.Int32 + s1 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + httpCalls.Add(1) + w.WriteHeader(http.StatusInternalServerError) + })) + defer s1.Close() + + feedURL, err := url.ParseRequestURI(s1.URL) + require.NoError(t, err) + + orm := bridges.NewORM(db) + _, bridge := cltest.MustCreateBridge(t, db, cltest.BridgeOpts{ + URL: feedURL.String(), + UseConnectionManager: true, + }) + + manager := bridgeconn.NewBridgeConnManager() + seedable, ok := manager.(interface { + SeedObservation(bridge bridges.BridgeType, requestData map[string]any, observation []byte) error + DisableEAConnDialingForTest() + }) + require.True(t, ok) + // This test seeds the cache directly and asserts no HTTP calls are made; the + // bridge URL points at a plain httptest server, not a streams-adapter, so real + // EAConn dialing must be disabled to avoid flaky cross-protocol traffic. + seedable.DisableEAConnDialingForTest() + require.NoError(t, seedable.SeedObservation(*bridge, utils.MustUnmarshalToMap(btcUSDPairing), []byte(`{"data":{"result":"9700"}}`))) + + task := pipeline.BridgeTask{ + BaseTask: pipeline.NewBaseTask(0, "bridge", nil, nil, 0), + Name: bridge.Name.String(), + RequestData: btcUSDPairing, + } + c := clhttptest.NewTestLocalOnlyHTTPClient() + trORM := pipeline.NewORM(db, logger.TestLogger(t), cfg.JobPipeline().MaxSuccessfulRuns()) + specID, err := trORM.CreateSpec(t.Context(), pipeline.Pipeline{}, *sqlutil.NewInterval(5 * time.Minute)) + require.NoError(t, err) + task.HelperSetDependencies(cfg.JobPipeline(), cfg.WebServer(), orm, specID, uuid.UUID{}, c) + task.HelperSetBridgeConnManager(manager) + + result, runInfo := task.Run(t.Context(), logger.TestLogger(t), pipeline.NewVarsFrom(nil), nil) + + assert.False(t, runInfo.IsPending) + assert.False(t, runInfo.IsRetryable) + require.NoError(t, result.Error) + assert.JSONEq(t, `{"data":{"result":"9700"}}`, result.Value.(string)) + assert.Equal(t, int32(0), httpCalls.Load()) +} + func TestBridgeTask_HandlesIntermittentFailure(t *testing.T) { t.Parallel() diff --git a/core/store/migrate/migrations/0302_add_use_connection_manager_to_bridge_types.sql b/core/store/migrate/migrations/0302_add_use_connection_manager_to_bridge_types.sql new file mode 100644 index 00000000000..f2e235d8985 --- /dev/null +++ b/core/store/migrate/migrations/0302_add_use_connection_manager_to_bridge_types.sql @@ -0,0 +1,4 @@ +-- +goose Up +ALTER TABLE bridge_types ADD COLUMN use_connection_manager boolean NOT NULL DEFAULT false; +-- +goose Down +ALTER TABLE bridge_types DROP COLUMN use_connection_manager; diff --git a/core/web/presenters/bridges.go b/core/web/presenters/bridges.go index a54718a4308..2c22629724d 100644 --- a/core/web/presenters/bridges.go +++ b/core/web/presenters/bridges.go @@ -18,6 +18,7 @@ type BridgeResource struct { IncomingToken string `json:"incomingToken,omitempty"` OutgoingToken string `json:"outgoingToken"` MinimumContractPayment *assets.Link `json:"minimumContractPayment"` + UseConnectionManager bool `json:"useConnectionManager"` CreatedAt time.Time `json:"createdAt"` } @@ -36,6 +37,7 @@ func NewBridgeResource(b bridges.BridgeType) *BridgeResource { Confirmations: b.Confirmations, OutgoingToken: b.OutgoingToken, MinimumContractPayment: b.MinimumContractPayment, + UseConnectionManager: b.UseConnectionManager, CreatedAt: b.CreatedAt, } } diff --git a/core/web/presenters/bridges_test.go b/core/web/presenters/bridges_test.go index 451698ec929..6f996c411bc 100644 --- a/core/web/presenters/bridges_test.go +++ b/core/web/presenters/bridges_test.go @@ -28,6 +28,7 @@ func TestBridgeResource(t *testing.T) { Confirmations: 1, OutgoingToken: "vjNL7X8Ea6GFJoa6PBsvK2ECzNK3b8IZ", MinimumContractPayment: assets.NewLinkFromJuels(1), + UseConnectionManager: true, CreatedAt: timestamp, } @@ -47,6 +48,7 @@ func TestBridgeResource(t *testing.T) { "confirmations":1, "outgoingToken":"vjNL7X8Ea6GFJoa6PBsvK2ECzNK3b8IZ", "minimumContractPayment":"1", + "useConnectionManager":true, "createdAt":"2000-01-01T00:00:00Z" } } @@ -72,6 +74,7 @@ func TestBridgeResource(t *testing.T) { "incomingToken": "cd+OfGXy3UHEDAlD0y27F6/rJE14X1UI", "outgoingToken":"vjNL7X8Ea6GFJoa6PBsvK2ECzNK3b8IZ", "minimumContractPayment":"1", + "useConnectionManager":true, "createdAt":"2000-01-01T00:00:00Z" } } diff --git a/core/web/resolver/bridge.go b/core/web/resolver/bridge.go index bc9d98d0f8c..c62f37e7dca 100644 --- a/core/web/resolver/bridge.go +++ b/core/web/resolver/bridge.go @@ -59,6 +59,11 @@ func (r *BridgeResolver) MinimumContractPayment() string { return r.bridge.MinimumContractPayment.String() } +// UseConnectionManager resolves the usage of connection manager for the bridge. +func (r *BridgeResolver) UseConnectionManager() bool { + return r.bridge.UseConnectionManager +} + // CreatedAt resolves the bridge's created at field. func (r *BridgeResolver) CreatedAt() graphql.Time { return graphql.Time{Time: r.bridge.CreatedAt} diff --git a/core/web/resolver/mutation.go b/core/web/resolver/mutation.go index 157304399ea..d6e3cae457c 100644 --- a/core/web/resolver/mutation.go +++ b/core/web/resolver/mutation.go @@ -58,6 +58,7 @@ type createBridgeInput struct { URL string Confirmations int32 MinimumContractPayment string + UseConnectionManager *bool } // CreateBridge creates a new bridge. @@ -84,6 +85,7 @@ func (r *Resolver) CreateBridge(ctx context.Context, args struct{ Input createBr URL: webURL, Confirmations: uint32(max(0, args.Input.Confirmations)), MinimumContractPayment: minContractPayment, + UseConnectionManager: args.Input.UseConnectionManager != nil && *args.Input.UseConnectionManager, } bta, bt, err := bridges.NewBridgeType(btr) @@ -451,6 +453,7 @@ type updateBridgeInput struct { URL string Confirmations int32 MinimumContractPayment string + UseConnectionManager *bool } func (r *Resolver) UpdateBridge(ctx context.Context, args struct { @@ -479,6 +482,7 @@ func (r *Resolver) UpdateBridge(ctx context.Context, args struct { URL: webURL, Confirmations: uint32(max(0, args.Input.Confirmations)), MinimumContractPayment: minContractPayment, + UseConnectionManager: args.Input.UseConnectionManager != nil && *args.Input.UseConnectionManager, } taskType, err := bridges.ParseBridgeName(string(args.ID)) diff --git a/core/web/schema/type/bridge.graphql b/core/web/schema/type/bridge.graphql index e24616f3262..f74e0f71155 100644 --- a/core/web/schema/type/bridge.graphql +++ b/core/web/schema/type/bridge.graphql @@ -5,6 +5,7 @@ type Bridge { confirmations: Int! outgoingToken: String! minimumContractPayment: String! + useConnectionManager: Boolean! createdAt: Time! } @@ -23,6 +24,7 @@ input CreateBridgeInput { url: String! confirmations: Int! minimumContractPayment: String! + useConnectionManager: Boolean } # CreateBridgeSuccess defines the success response when creating a bridge @@ -40,6 +42,7 @@ input UpdateBridgeInput { url: String! confirmations: Int! minimumContractPayment: String! + useConnectionManager: Boolean } # UpdateBridgeSuccess defines the success response when updating a bridge