Merge branch 'develop' into fix/ccip_misc_fixes

.changeset/fuzzy-hairs-appear.md

@@ -0,0 +1,5 @@
+---
+"chainlink": patch
+---
+
+Prometheus observability layer added to OCR3 Reporting Plugins #internal

.github/e2e-tests.yml

@@ -948,19 +948,6 @@ runner-test-matrix:
       E2E_TEST_SELECTED_NETWORK: SIMULATED_1,SIMULATED_2
       E2E_JD_VERSION: 0.6.0
 
-  - id: smoke/ccip/ccip_batching_test.go:*
-    path: integration-tests/smoke/ccip/ccip_batching_test.go
-    test_env_type: docker
-    runs_on: ubuntu-latest
-    triggers:
-      - PR E2E Core Tests
-      - Nightly E2E Tests
-    test_cmd: cd integration-tests/ && go test smoke/ccip/ccip_batching_test.go -timeout 12m -test.parallel=1 -count=1 -json
-    pyroscope_env: ci-smoke-ccipv1_6-evm-simulated
-    test_env_vars:
-      E2E_TEST_SELECTED_NETWORK: SIMULATED_1,SIMULATED_2,SIMULATED_3
-      E2E_JD_VERSION: 0.6.0
-
   - id: smoke/ccip/ccip_token_transfer_test.go:*
     path: integration-tests/smoke/ccip/ccip_token_transfer_test.go
     test_env_type: docker

.github/integration-in-memory-tests.yml

@@ -31,6 +31,14 @@ runner-test-matrix:
     triggers:
       - PR Integration CCIP Tests
     test_cmd: cd integration-tests/smoke/ccip && go test ccip_fee_boosting_test.go -timeout 12m -test.parallel=2 -count=1 -json
+
+  - id: smoke/ccip/ccip_batching_test.go:*
+    path: integration-tests/smoke/ccip/ccip_batching_test.go
+    test_env_type: in-memory
+    runs_on: ubuntu-latest
+    triggers:
+      - PR Integration CCIP Tests
+    test_cmd: cd integration-tests/smoke/ccip && go test ccip_batching_test.go -timeout 12m -test.parallel=2 -count=1 -json
 
   - id: contracts/ccipreader_test.go:*
     path: integration-tests/contracts/ccipreader_test.go

.github/workflows/changeset.yml

@@ -182,7 +182,7 @@ jobs:
               - added|modified: 'contracts/.changeset/*.md'
 
       - name: Setup node
-        uses: ./.github/actions/setup-node
+        uses: ./.github/actions/setup-nodejs
         if: ${{ steps.files-changed.outputs.contracts-changeset == 'true' }}
 
       - name: Validate changeset files

contracts/src/v0.8/ccip/capability/CCIPHome.sol

@@ -62,7 +62,7 @@ import {EnumerableSet} from "../../vendor/openzeppelin-solidity/v5.0.2/contracts
 ///       │    Active   │    revokeCandidate │  Candidate  │◄───────────┐
 ///       │    [1,0]    │◄───────────────────┤    [1,1]    │────────────┘
 ///       │             ├───────────────────►│             │
-///       └─────────────┘    setSecondary    └─────────────┘
+///       └─────────────┘    setCandidate    └─────────────┘
 ///
 contract CCIPHome is Ownable2StepMsgSender, ITypeAndVersion, ICapabilityConfiguration, IERC165 {
   using EnumerableSet for EnumerableSet.UintSet;

contracts/src/v0.8/ccip/rmn/RMNHome.sol

@@ -54,7 +54,7 @@ import {Ownable2StepMsgSender} from "../../shared/access/Ownable2StepMsgSender.s
 ///       │    Active   │    revokeCandidate │  Candidate  │◄───────────┐
 ///       │    [1,0]    │◄───────────────────┤    [1,1]    │────────────┘
 ///       │             ├───────────────────►│             │
-///       └─────────────┘    setSecondary    └─────────────┘
+///       └─────────────┘    setCandidate    └─────────────┘
 ///
 contract RMNHome is Ownable2StepMsgSender, ITypeAndVersion {
   event ConfigSet(bytes32 indexed configDigest, uint32 version, StaticConfig staticConfig, DynamicConfig dynamicConfig);

core/capabilities/ccip/oraclecreator/plugin.go

@@ -22,6 +22,7 @@ import (
 	cctypes "github.com/smartcontractkit/chainlink/v2/core/capabilities/ccip/types"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/assets"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/config/toml"
+	"github.com/smartcontractkit/chainlink/v2/core/services/ocr3/promwrapper"
 
 	"github.com/smartcontractkit/libocr/commontypes"
 	libocr3 "github.com/smartcontractkit/libocr/offchainreporting2plus"
@@ -229,6 +230,12 @@ func (i *pluginOracleCreator) createFactoryAndTransmitter(
 ) (ocr3types.ReportingPluginFactory[[]byte], ocr3types.ContractTransmitter[[]byte], error) {
 	var factory ocr3types.ReportingPluginFactory[[]byte]
 	var transmitter ocr3types.ContractTransmitter[[]byte]
+
+	chainID, err := chainsel.GetChainIDFromSelector(uint64(config.Config.ChainSelector))
+	if err != nil {
+		return nil, nil, fmt.Errorf("unsupported chain selector %d %w", config.Config.ChainSelector, err)
+	}
+
 	if config.Config.PluginType == uint8(cctypes.PluginTypeCCIPCommit) {
 		if !i.peerWrapper.IsStarted() {
 			return nil, nil, fmt.Errorf("peer wrapper is not started")
@@ -263,6 +270,7 @@ func (i *pluginOracleCreator) createFactoryAndTransmitter(
 			rmnPeerClient,
 			rmnCrypto,
 		)
+		factory = promwrapper.NewReportingPluginFactory[[]byte](factory, chainID, "CCIPCommit")
 		transmitter = ocrimpls.NewCommitContractTransmitter[[]byte](destChainWriter,
 			ocrtypes.Account(destFromAccounts[0]),
 			hexutil.Encode(config.Config.OfframpAddress), // TODO: this works for evm only, how about non-evm?
@@ -283,6 +291,7 @@ func (i *pluginOracleCreator) createFactoryAndTransmitter(
 			contractReaders,
 			chainWriters,
 		)
+		factory = promwrapper.NewReportingPluginFactory[[]byte](factory, chainID, "CCIPExec")
 		transmitter = ocrimpls.NewExecContractTransmitter[[]byte](destChainWriter,
 			ocrtypes.Account(destFromAccounts[0]),
 			hexutil.Encode(config.Config.OfframpAddress), // TODO: this works for evm only, how about non-evm?

core/capabilities/compute/compute.go

@@ -273,9 +273,19 @@ func (c *Compute) worker(ctx context.Context) {
 }
 
 func (c *Compute) Close() error {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
 	c.modules.close()
 	close(c.stopCh)
+
+	err := c.registry.Remove(ctx, CapabilityIDCompute)
+	if err != nil {
+		return err
+	}
+
 	c.wg.Wait()
+
 	return nil
 }
 

core/capabilities/registry.go

@@ -193,6 +193,21 @@ func (r *Registry) Add(ctx context.Context, c capabilities.BaseCapability) error
 	return nil
 }
 
+// Add adds a capability to the registry.
+func (r *Registry) Remove(ctx context.Context, id string) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	_, ok := r.m[id]
+	if !ok {
+		return fmt.Errorf("unable to remove, capability not found: %s", id)
+	}
+
+	delete(r.m, id)
+	r.lggr.Infow("capability removed", "id", id)
+	return nil
+}
+
 // NewRegistry returns a new Registry.
 func NewRegistry(lggr logger.Logger) *Registry {
 	return &Registry{

core/capabilities/webapi/target/target.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
@@ -57,6 +58,12 @@ func (c *Capability) Start(ctx context.Context) error {
 }
 
 func (c *Capability) Close() error {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	err := c.registry.Remove(ctx, c.capabilityInfo.ID)
+	if err != nil {
+		return err
+	}
 	return nil
 }
 

core/capabilities/webapi/trigger/trigger.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	ethCommon "github.com/ethereum/go-ethereum/common"
 
@@ -256,6 +257,12 @@ func (h *triggerConnectorHandler) Start(ctx context.Context) error {
 }
 func (h *triggerConnectorHandler) Close() error {
 	return h.StopOnce("GatewayConnectorServiceWrapper", func() error {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		err := h.registry.Remove(ctx, h.ID)
+		if err != nil {
+			return err
+		}
 		return nil
 	})
 }

core/cmd/shell.go

@@ -53,6 +53,7 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/mercury/wsrpc/cache"
 	"github.com/smartcontractkit/chainlink/v2/core/services/versioning"
 	"github.com/smartcontractkit/chainlink/v2/core/services/webhook"
+	"github.com/smartcontractkit/chainlink/v2/core/services/workflows"
 	"github.com/smartcontractkit/chainlink/v2/core/sessions"
 	"github.com/smartcontractkit/chainlink/v2/core/static"
 	"github.com/smartcontractkit/chainlink/v2/core/store/migrate"
@@ -111,6 +112,10 @@ func initGlobals(cfgProm config.Prometheus, cfgTracing config.Tracing, cfgTeleme
 				AuthPublicKeyHex:         csaPubKeyHex,
 				AuthHeaders:              beholderAuthHeaders,
 			}
+			// note: due to the OTEL specification, all histogram buckets
+			// must be defined when the beholder client is created
+			clientCfg.MetricViews = append(clientCfg.MetricViews, workflows.MetricViews()...)
+
 			if tracingCfg.Enabled {
 				clientCfg.TraceSpanExporter, err = tracingCfg.NewSpanExporter()
 				if err != nil {

core/scripts/go.mod

@@ -2,6 +2,8 @@ module github.com/smartcontractkit/chainlink/core/scripts
 
 go 1.23.3
 
+toolchain go1.23.4
+
 // Make sure we're working with the latest chainlink libs
 replace github.com/smartcontractkit/chainlink/v2 => ../../
 
@@ -24,7 +26,7 @@ require (
 	github.com/prometheus/client_golang v1.20.5
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chainlink-automation v0.8.1
-	github.com/smartcontractkit/chainlink-common v0.3.1-0.20241127162636-07aa781ee1f4
+	github.com/smartcontractkit/chainlink-common v0.3.1-0.20241206011233-b6684ee6508f
 	github.com/smartcontractkit/chainlink/deployment v0.0.0-00010101000000-000000000000
 	github.com/smartcontractkit/chainlink/v2 v2.14.0-mercury-20240807.0.20241106193309-5560cd76211a
 	github.com/smartcontractkit/libocr v0.0.0-20241007185508-adbe57025f12
@@ -304,6 +306,7 @@ require (
 	github.com/smartcontractkit/chainlink-protos/orchestrator v0.3.2 // indirect
 	github.com/smartcontractkit/chainlink-solana v1.1.1-0.20241204153209-c3a71b0eef99 // indirect
 	github.com/smartcontractkit/chainlink-starknet/relayer v0.1.1-0.20241202202529-2033490e77b8 // indirect
+	github.com/smartcontractkit/chainlink-testing-framework/lib v1.50.13 // indirect
 	github.com/smartcontractkit/grpc-proxy v0.0.0-20240830132753-a7e17fec5ab7 // indirect
 	github.com/smartcontractkit/tdh2/go/ocr2/decryptionplugin v0.0.0-20241009055228-33d0c0bf38de // indirect
 	github.com/smartcontractkit/tdh2/go/tdh2 v0.0.0-20241009055228-33d0c0bf38de // indirect

core/scripts/go.sum

@@ -1142,8 +1142,8 @@ github.com/smartcontractkit/chainlink-automation v0.8.1 h1:sTc9LKpBvcKPc1JDYAmgB
 github.com/smartcontractkit/chainlink-automation v0.8.1/go.mod h1:Iij36PvWZ6blrdC5A/nrQUBuf3MH3JvsBB9sSyc9W08=
 github.com/smartcontractkit/chainlink-ccip v0.0.0-20241204015713-8956bb614e9e h1:GnM6ZWV6vlk2+n6c6o+v/R1LtXzBGVVx7r37nt/h6Uc=
 github.com/smartcontractkit/chainlink-ccip v0.0.0-20241204015713-8956bb614e9e/go.mod h1:80vGBbOfertJig0xFKsRfm+i17FkjdKkk1dAaGE45Os=
-github.com/smartcontractkit/chainlink-common v0.3.1-0.20241127162636-07aa781ee1f4 h1:atCZ1jol7a+tdtgU/wNqXgliBun5H7BjGBicGL8Tj6o=
-github.com/smartcontractkit/chainlink-common v0.3.1-0.20241127162636-07aa781ee1f4/go.mod h1:bQktEJf7sJ0U3SmIcXvbGUox7SmXcnSEZ4kUbT8R5Nk=
+github.com/smartcontractkit/chainlink-common v0.3.1-0.20241206011233-b6684ee6508f h1:hH+cAG2zt+WK4I2m572LXAnAJg3wtGEAwzBKR8FiXo8=
+github.com/smartcontractkit/chainlink-common v0.3.1-0.20241206011233-b6684ee6508f/go.mod h1:bQktEJf7sJ0U3SmIcXvbGUox7SmXcnSEZ4kUbT8R5Nk=
 github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241202195413-82468150ac1e h1:PRoeby6ZlTuTkv2f+7tVU4+zboTfRzI+beECynF4JQ0=
 github.com/smartcontractkit/chainlink-cosmos v0.5.2-0.20241202195413-82468150ac1e/go.mod h1:mUh5/woemsVaHgTorA080hrYmO3syBCmPdnWc/5dOqk=
 github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20241202141438-a90db35252db h1:N1RH1hSr2ACzOFc9hkCcjE8pRBTdcU3p8nsTJByaLes=
@@ -1158,6 +1158,8 @@ github.com/smartcontractkit/chainlink-solana v1.1.1-0.20241204153209-c3a71b0eef9
 github.com/smartcontractkit/chainlink-solana v1.1.1-0.20241204153209-c3a71b0eef99/go.mod h1:p8aUDfJeley6oer7y+Ucd3edOtRlMTnWg3mN6rhaLWo=
 github.com/smartcontractkit/chainlink-starknet/relayer v0.1.1-0.20241202202529-2033490e77b8 h1:tNS7U9lrxkFvEuyxQv11HHOiV9LPDGC9wYEy+yM/Jv4=
 github.com/smartcontractkit/chainlink-starknet/relayer v0.1.1-0.20241202202529-2033490e77b8/go.mod h1:EBrEgcdIbwepqguClkv8Ohy7CbyWSJaE4EC9aBJlQK0=
+github.com/smartcontractkit/chainlink-testing-framework/lib v1.50.13 h1:T0kbw07Vb6xUyA9MIJZfErMgWseWi1zf7cYvRpoq7ug=
+github.com/smartcontractkit/chainlink-testing-framework/lib v1.50.13/go.mod h1:1CKUOzoK+Ga19WuhRH9pxZ+qUUnrlIx108VEA6qSzeQ=
 github.com/smartcontractkit/grpc-proxy v0.0.0-20240830132753-a7e17fec5ab7 h1:12ijqMM9tvYVEm+nR826WsrNi6zCKpwBhuApq127wHs=
 github.com/smartcontractkit/grpc-proxy v0.0.0-20240830132753-a7e17fec5ab7/go.mod h1:FX7/bVdoep147QQhsOPkYsPEXhGZjeYx6lBSaSXtZOA=
 github.com/smartcontractkit/libocr v0.0.0-20241007185508-adbe57025f12 h1:NzZGjaqez21I3DU7objl3xExTH4fxYvzTqar8DC6360=

core/services/chainlink/application.go

@@ -300,18 +300,19 @@ func NewApplication(opts ApplicationOpts) (Application, error) {
 					return nil, fmt.Errorf("expected 1 key, got %d", len(keys))
 				}
 
-				fetcher := syncer.NewFetcherService(globalLogger, gatewayConnectorWrapper)
+				lggr := globalLogger.Named("WorkflowRegistrySyncer")
+				fetcher := syncer.NewFetcherService(lggr, gatewayConnectorWrapper)
 
-				eventHandler := syncer.NewEventHandler(globalLogger, syncer.NewWorkflowRegistryDS(opts.DS, globalLogger),
-					fetcher.Fetch, workflowstore.NewDBStore(opts.DS, globalLogger, clockwork.NewRealClock()), opts.CapabilitiesRegistry,
+				eventHandler := syncer.NewEventHandler(lggr, syncer.NewWorkflowRegistryDS(opts.DS, globalLogger),
+					fetcher.Fetch, workflowstore.NewDBStore(opts.DS, lggr, clockwork.NewRealClock()), opts.CapabilitiesRegistry,
 					custmsg.NewLabeler(), clockwork.NewRealClock(), keys[0])
 
-				loader := syncer.NewWorkflowRegistryContractLoader(globalLogger, cfg.Capabilities().WorkflowRegistry().Address(), func(ctx context.Context, bytes []byte) (syncer.ContractReader, error) {
+				loader := syncer.NewWorkflowRegistryContractLoader(lggr, cfg.Capabilities().WorkflowRegistry().Address(), func(ctx context.Context, bytes []byte) (syncer.ContractReader, error) {
 					return relayer.NewContractReader(ctx, bytes)
 				}, eventHandler)
 
 				globalLogger.Debugw("Creating WorkflowRegistrySyncer")
-				wfSyncer := syncer.NewWorkflowRegistry(globalLogger, func(ctx context.Context, bytes []byte) (syncer.ContractReader, error) {
+				wfSyncer := syncer.NewWorkflowRegistry(lggr, func(ctx context.Context, bytes []byte) (syncer.ContractReader, error) {
 					return relayer.NewContractReader(ctx, bytes)
 				}, cfg.Capabilities().WorkflowRegistry().Address(),
 					syncer.WorkflowEventPollerConfig{

core/services/ocr3/promwrapper/factory.go

@@ -0,0 +1,42 @@
+package promwrapper
+
+import (
+	"context"
+
+	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
+)
+
+var _ ocr3types.ReportingPluginFactory[any] = &ReportingPluginFactory[any]{}
+
+type ReportingPluginFactory[RI any] struct {
+	origin  ocr3types.ReportingPluginFactory[RI]
+	chainID string
+	plugin  string
+}
+
+func NewReportingPluginFactory[RI any](
+	origin ocr3types.ReportingPluginFactory[RI],
+	chainID string,
+	plugin string,
+) *ReportingPluginFactory[RI] {
+	return &ReportingPluginFactory[RI]{
+		origin:  origin,
+		chainID: chainID,
+		plugin:  plugin,
+	}
+}
+
+func (r ReportingPluginFactory[RI]) NewReportingPlugin(ctx context.Context, config ocr3types.ReportingPluginConfig) (ocr3types.ReportingPlugin[RI], ocr3types.ReportingPluginInfo, error) {
+	plugin, info, err := r.origin.NewReportingPlugin(ctx, config)
+	if err != nil {
+		return nil, ocr3types.ReportingPluginInfo{}, err
+	}
+	wrapped := newReportingPlugin(
+		plugin,
+		r.chainID,
+		r.plugin,
+		promOCR3ReportsGenerated,
+		promOCR3Durations,
+	)
+	return wrapped, info, err
+}

core/services/ocr3/promwrapper/factory_test.go

@@ -0,0 +1,41 @@
+package promwrapper
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
+)
+
+func Test_WrapperFactory(t *testing.T) {
+	validFactory := NewReportingPluginFactory(fakeFactory[uint]{}, "solana", "plugin")
+	failingFactory := NewReportingPluginFactory(fakeFactory[uint]{err: errors.New("error")}, "123", "plugin")
+
+	plugin, _, err := validFactory.NewReportingPlugin(tests.Context(t), ocr3types.ReportingPluginConfig{})
+	require.NoError(t, err)
+
+	_, err = plugin.Outcome(tests.Context(t), ocr3types.OutcomeContext{}, nil, nil)
+	require.NoError(t, err)
+
+	require.Equal(t, 1, counterFromHistogramByLabels(t, promOCR3Durations, "solana", "plugin", "outcome", "true"))
+	require.Equal(t, 0, counterFromHistogramByLabels(t, promOCR3Durations, "solana", "plugin", "outcome", "false"))
+
+	_, _, err = failingFactory.NewReportingPlugin(tests.Context(t), ocr3types.ReportingPluginConfig{})
+	require.Error(t, err)
+}
+
+type fakeFactory[RI any] struct {
+	err error
+}
+
+func (f fakeFactory[RI]) NewReportingPlugin(context.Context, ocr3types.ReportingPluginConfig) (ocr3types.ReportingPlugin[RI], ocr3types.ReportingPluginInfo, error) {
+	if f.err != nil {
+		return nil, ocr3types.ReportingPluginInfo{}, f.err
+	}
+	return fakePlugin[RI]{}, ocr3types.ReportingPluginInfo{}, nil
+}