Capability Registry Config Provider

core/scripts/go.mod

@@ -47,7 +47,7 @@ require (
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chainlink-automation v0.8.1
 	github.com/smartcontractkit/chainlink-ccip v0.1.1-solana.0.20251128020529-88d93b01d749
-	github.com/smartcontractkit/chainlink-common v0.9.6-0.20251125103916-0b41e73b80c4
+	github.com/smartcontractkit/chainlink-common v0.9.6-0.20251208161008-e78201796d79
 	github.com/smartcontractkit/chainlink-data-streams v0.1.7-0.20251123170926-313d8827bd6f
 	github.com/smartcontractkit/chainlink-deployments-framework v0.70.0
 	github.com/smartcontractkit/chainlink-evm v0.3.4-0.20251201175512-af04e919ebfb

core/scripts/go.sum

@@ -1634,8 +1634,8 @@ github.com/smartcontractkit/chainlink-ccip/deployment v0.0.0-20251027185542-babb
 github.com/smartcontractkit/chainlink-ccip/deployment v0.0.0-20251027185542-babb09e5363e/go.mod h1:IaoLCQE1miX3iUlQNxOPcVrXrshcO/YsFpxnFuhG9DM=
 github.com/smartcontractkit/chainlink-ccv v0.0.0-20251208105135-980e1424951b h1:HKz0L4mLH5FlINK+4Wc5cLNYTAe64sVsS4q3Srcvp9U=
 github.com/smartcontractkit/chainlink-ccv v0.0.0-20251208105135-980e1424951b/go.mod h1:Ysd/qkofD0bepk29RS7Q4ZlVDd4yAHXucYsp5gAy6AE=
-github.com/smartcontractkit/chainlink-common v0.9.6-0.20251125103916-0b41e73b80c4 h1:2LHrlWAStA2oRcDKrJ9lKOShC9an2Pkis2BwY8J7gDw=
-github.com/smartcontractkit/chainlink-common v0.9.6-0.20251125103916-0b41e73b80c4/go.mod h1:uRnGLHKo56QYaPk93z0NRAIgv115lh72rzG40CiE1Mk=
+github.com/smartcontractkit/chainlink-common v0.9.6-0.20251208161008-e78201796d79 h1:j46QW8Tf90rbWYN9EwXb26wQrsvDW+AR2Qhtx/+G0fo=
+github.com/smartcontractkit/chainlink-common v0.9.6-0.20251208161008-e78201796d79/go.mod h1:uRnGLHKo56QYaPk93z0NRAIgv115lh72rzG40CiE1Mk=
 github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.10 h1:FJAFgXS9oqASnkS03RE1HQwYQQxrO4l46O5JSzxqLgg=
 github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.10/go.mod h1:oiDa54M0FwxevWwyAX773lwdWvFYYlYHHQV1LQ5HpWY=
 github.com/smartcontractkit/chainlink-common/pkg/monitoring v0.0.0-20250415235644-8703639403c7 h1:9wh1G+WbXwPVqf0cfSRSgwIcaXTQgvYezylEAfwmrbw=

core/services/chainlink/application.go

@@ -722,6 +722,7 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err
 				Relayers:                       relayChainInterops,
 				MailMon:                        mailMon,
 				CapabilitiesRegistry:           opts.CapabilitiesRegistry,
+				RegistrySyncer:                 creServices.capabilityRegistrySyncer,
 				DonTimeStore:                   opts.DonTimeStore,
 				RetirementReportCache:          opts.RetirementReportCache,
 				GatewayConnectorServiceWrapper: creServices.gatewayConnectorWrapper,
@@ -902,8 +903,8 @@ type CREServices struct {
 	// srvs are all the services that are created, including those that are explicitly exposed
 	srvs []services.ServiceCtx
 
-	workflowRegistrySyncer syncerV2.WorkflowRegistrySyncer
-
+	workflowRegistrySyncer   syncerV2.WorkflowRegistrySyncer
+	capabilityRegistrySyncer registrysyncerV2.RegistrySyncer
 	// orgResolver provides realtime workflow owner --> orgID resolution
 	orgResolver orgresolver.OrgResolver
 }
@@ -969,6 +970,7 @@ func newCREServices(
 		srvcs = append(srvcs, gatewayConnectorWrapper)
 	}
 
+	var capRegSyncer registrysyncerV2.RegistrySyncer
 	var orgResolver orgresolver.OrgResolver
 	if cfg.CRE().Linking().URL() != "" {
 		var wrChainDetails chainselectors.ChainDetails
@@ -1110,6 +1112,7 @@ func newCREServices(
 
 				registrySyncer.AddListener(wfLauncher)
 				srvcs = append(srvcs, wfLauncher, registrySyncer)
+				capRegSyncer = registrySyncer
 			case 2:
 				registrySyncer, err := registrysyncerV2.New(
 					globalLogger,
@@ -1124,6 +1127,7 @@ func newCREServices(
 
 				registrySyncer.AddListener(wfLauncher)
 				srvcs = append(srvcs, wfLauncher, registrySyncer)
+				capRegSyncer = registrySyncer
 			default:
 				return nil, fmt.Errorf("could not configure capability registry syncer with version: %d", externalRegistryVersion.Major())
 			}
@@ -1315,13 +1319,14 @@ func newCREServices(
 		opts.CapabilitiesRegistry.SetLocalRegistry(&capabilities.TestMetadataRegistry{})
 	}
 	return &CREServices{
-		workflowRateLimiter:     workflowRateLimiter,
-		workflowLimits:          workflowLimits,
-		gatewayConnectorWrapper: gatewayConnectorWrapper,
-		getPeerID:               getPeerID,
-		srvs:                    srvcs,
-		workflowRegistrySyncer:  workflowRegistrySyncerV2,
-		orgResolver:             orgResolver,
+		workflowRateLimiter:      workflowRateLimiter,
+		workflowLimits:           workflowLimits,
+		gatewayConnectorWrapper:  gatewayConnectorWrapper,
+		getPeerID:                getPeerID,
+		srvs:                     srvcs,
+		workflowRegistrySyncer:   workflowRegistrySyncerV2,
+		capabilityRegistrySyncer: capRegSyncer,
+		orgResolver:              orgResolver,
 	}, nil
 }
 

core/services/ocr2/delegate.go

@@ -84,6 +84,7 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/validate"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocrcommon"
 	"github.com/smartcontractkit/chainlink/v2/core/services/pipeline"
+	"github.com/smartcontractkit/chainlink/v2/core/services/registrysyncer"
 	"github.com/smartcontractkit/chainlink/v2/core/services/relay"
 	evmrelay "github.com/smartcontractkit/chainlink/v2/core/services/relay/evm"
 	functionsRelay "github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/functions"
@@ -145,6 +146,7 @@ type Delegate struct {
 
 	legacyChains                   legacyevm.LegacyChainContainer // legacy: use relayers instead
 	capabilitiesRegistry           core.CapabilitiesRegistry
+	registrySyncer                 registrysyncer.RegistrySyncer
 	dontimeStore                   *dontime.Store
 	gatewayConnectorServiceWrapper *gatewayconnector.ServiceWrapper
 	WorkflowRegistrySyncer         syncerV2.WorkflowRegistrySyncer
@@ -257,6 +259,7 @@ type DelegateOpts struct {
 	Relayers                       RelayGetter
 	MailMon                        *mailbox.Monitor
 	CapabilitiesRegistry           core.CapabilitiesRegistry
+	RegistrySyncer                 registrysyncer.RegistrySyncer
 	DonTimeStore                   *dontime.Store
 	RetirementReportCache          retirement.RetirementReportCache
 	GatewayConnectorServiceWrapper *gatewayconnector.ServiceWrapper
@@ -878,6 +881,27 @@ func (d *Delegate) newDonTimePlugin(
 		return nil, ErrRelayNotEnabled{Err: err, Relay: spec.Relay, PluginName: "dontime"}
 	}
 
+	var capRegConfigTracker *relay.CapRegConfigProvider
+	if true { // TODO: Check if {job spec?} specifies that config is stored in Cap Reg
+		localNode, err := d.capabilitiesRegistry.LocalNode(ctx)
+		if err != nil {
+			return nil, errors.New("Failed to get local node from cap reg")
+		}
+		// TODO: Will this work to get the DonID for tracking config in Cap Reg?
+		donID := localNode.WorkflowDON.ID
+
+		capName := string(jb.OCR2OracleSpec.PluginType) // TODO: Can we get Cap Name from spec or where?
+		capRegConfigTracker, err := relay.NewCapRegConfigProvider(ctx, lggr, donID, capName)
+		if err != nil {
+			return nil, err
+		}
+		d.registrySyncer.AddListener(capRegConfigTracker) // Subscribe to Cap Reg changes
+		err = d.registrySyncer.Sync(ctx, false)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	provider, err := relayer.NewPluginProvider(ctx, types.RelayArgs{
 		ExternalJobID: jb.ExternalJobID,
 		JobID:         jb.ID,
@@ -887,8 +911,9 @@ func (d *Delegate) newDonTimePlugin(
 		RelayConfig:   spec.RelayConfig.Bytes(),
 		ProviderType:  string(types.DonTimePlugin),
 	}, types.PluginArgs{
-		TransmitterID: spec.TransmitterID.String,
-		PluginConfig:  spec.PluginConfig.Bytes(),
+		TransmitterID:       spec.TransmitterID.String,
+		PluginConfig:        spec.PluginConfig.Bytes(),
+		CapRegConfigTracker: capRegConfigTracker,
 	})
 	if err != nil {
 		return nil, err

core/services/registrysyncer/local_registry.go

@@ -137,12 +137,12 @@ type NodeInfo struct {
 
 type LocalRegistry struct {
 	core.UnimplementedCapabilitiesRegistryMetadata
-
-	Logger            logger.Logger
-	GetPeerID         func() (types.PeerID, error)
-	IDsToDONs         map[DonID]DON
-	IDsToNodes        map[types.PeerID]NodeInfo
-	IDsToCapabilities map[string]Capability
+	LastSyncedBlockHeight string
+	Logger                logger.Logger
+	GetPeerID             func() (types.PeerID, error)
+	IDsToDONs             map[DonID]DON
+	IDsToNodes            map[types.PeerID]NodeInfo
+	IDsToCapabilities     map[string]Capability
 }
 
 func NewLocalRegistry(

core/services/registrysyncer/syncer.go

@@ -3,6 +3,7 @@ package registrysyncer
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -251,10 +252,13 @@ func (s *registrySyncer) importOnchainRegistry(ctx context.Context) (*LocalRegis
 
 	nodes := []kcr.INodeInfoProviderNodeInfo{}
 
-	err = s.reader.GetLatestValue(ctx, s.capabilitiesContract.ReadIdentifier("getNodes"), primitives.Unconfirmed, nil, &nodes)
+	head, err := s.reader.GetLatestValueWithHeadData(ctx, s.capabilitiesContract.ReadIdentifier("getNodes"), primitives.Unconfirmed, nil, &nodes)
 	if err != nil {
 		return nil, err
 	}
+	if head != nil {
+		return nil, errors.New("Received nil head")
+	}
 
 	idsToNodes := map[p2ptypes.PeerID]NodeInfo{}
 	for _, node := range nodes {
@@ -285,11 +289,12 @@ func (s *registrySyncer) importOnchainRegistry(ctx context.Context) (*LocalRegis
 	}
 
 	return &LocalRegistry{
-		Logger:            s.lggr,
-		GetPeerID:         s.getPeerID,
-		IDsToDONs:         idsToDONs,
-		IDsToCapabilities: idsToCapabilities,
-		IDsToNodes:        idsToNodes,
+		Logger:                s.lggr,
+		LastSyncedBlockHeight: head.Height,
+		GetPeerID:             s.getPeerID,
+		IDsToDONs:             idsToDONs,
+		IDsToCapabilities:     idsToCapabilities,
+		IDsToNodes:            idsToNodes,
 	}, nil
 }
 

core/services/relay/cap_reg_config_provider.go

@@ -0,0 +1,133 @@
+package relay
+
+import (
+	"bytes"
+	"context"
+	"github.com/pkg/errors"
+	"strconv"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
+	"github.com/smartcontractkit/chainlink/v2/core/services/registrysyncer"
+	ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types"
+)
+
+var _ ocrtypes.ContractConfigTracker = &CapRegConfigProvider{}
+
+// CapRegConfigProvider subscribes to the registrySyncer for on-chain changes in the Capability Registry.
+// Parses config from on-chain Capability Config.
+type CapRegConfigProvider struct {
+	services.StateMachine
+	lggr logger.Logger
+
+	lastSyncedBlockHeight string
+	localConfig           ocrtypes.ContractConfig
+	donID                 registrysyncer.DonID
+	capability            string
+	initialSync           bool
+}
+
+func NewCapRegConfigProvider(ctx context.Context, lggr logger.Logger, donID uint32, capability string) (*CapRegConfigProvider, error) {
+	return newCapRegConfigProvider(ctx, lggr, donID, capability)
+}
+
+func newCapRegConfigProvider(ctx context.Context, lggr logger.Logger, donID uint32, capability string) (*CapRegConfigProvider, error) {
+	return &CapRegConfigProvider{
+		lggr:       logger.Named(lggr, "ConfigPoller"),
+		donID:      registrysyncer.DonID(donID),
+		capability: capability,
+		// localConfig will be updated once sync is called on registry syncer
+		localConfig: ocrtypes.ContractConfig{},
+		initialSync: false,
+	}, nil
+}
+
+// Subscribes to registry syncer for config changes
+var _ registrysyncer.Listener = &CapRegConfigProvider{}
+
+func (cp *CapRegConfigProvider) OnNewRegistry(ctx context.Context, registry *registrysyncer.LocalRegistry) error {
+	if registry == nil {
+		return errors.New("registry is nil")
+	}
+	cp.initialSync = true
+
+	don, ok := registry.IDsToDONs[cp.donID]
+	if !ok {
+		cp.lggr.Warnw("DON not found in registry", "donID", cp.donID)
+		return nil
+	}
+
+	capConfig, ok := don.CapabilityConfigurations[cp.capability]
+	if !ok {
+		cp.lggr.Warnw("Capability not found for DON", "donID", cp.donID, "capability", cp.capability)
+		return nil
+	}
+
+	// This config is on-chain in the Capability Registry
+	newOnChainConfig := capConfig.Config
+	cp.lastSyncedBlockHeight = registry.LastSyncedBlockHeight
+
+	// TODO: Do we unmarshal newOnChainConfig into ocrtypes.ContractConfig or is that just the OnchainConfig?
+	// TODO: If so, how do we obtain the rest of the information?
+	cp.localConfig = ocrtypes.ContractConfig{
+		ConfigDigest:          ocrtypes.ConfigDigest{},
+		ConfigCount:           0,
+		Signers:               nil,
+		Transmitters:          nil,
+		F:                     don.F,
+		OnchainConfig:         nil, // TODO: Is newOnChainConfig just this part?
+		OffchainConfigVersion: 0,
+		OffchainConfig:        nil,
+	}
+
+	if !bytes.Equal(newOnChainConfig, cp.localConfig.OnchainConfig) {
+		cp.lggr.Infow("capability config updated", "donID", cp.donID, "capability", cp.capability)
+		cp.localConfig.OnchainConfig = newOnChainConfig
+	}
+	return nil
+}
+
+// LatestConfigDetails returns the latest config details from the logs
+func (cp *CapRegConfigProvider) LatestConfigDetails(ctx context.Context) (changedInBlock uint64, configDigest ocrtypes.ConfigDigest, err error) {
+	if !cp.initialSync {
+		return 0, ocrtypes.ConfigDigest{}, errors.New("Config Provider has not been synced yet")
+	}
+	blockHeight, err := cp.LatestBlockHeight(ctx)
+	if err != nil {
+		return 0, ocrtypes.ConfigDigest{}, err
+	}
+	// TODO: Implement Config Digest...
+	return blockHeight, ocrtypes.ConfigDigest{}, errors.New("Unimplemented")
+}
+
+// LatestConfig returns the latest config from the logs on a certain block
+func (cp *CapRegConfigProvider) LatestConfig(ctx context.Context, changedInBlock uint64) (ocrtypes.ContractConfig, error) {
+	if !cp.initialSync {
+		return ocrtypes.ContractConfig{}, errors.New("Config Provider has not been synced yet")
+	}
+	latestConfigSet := ocrtypes.ContractConfig{
+		ConfigDigest:          cp.localConfig.ConfigDigest,
+		ConfigCount:           cp.localConfig.ConfigCount,
+		Signers:               cp.localConfig.Signers,
+		Transmitters:          cp.localConfig.Transmitters,
+		F:                     cp.localConfig.F,
+		OnchainConfig:         cp.localConfig.OnchainConfig,
+		OffchainConfigVersion: cp.localConfig.OffchainConfigVersion,
+		OffchainConfig:        cp.localConfig.OffchainConfig,
+	}
+	cp.lggr.Infow("LatestConfig", "latestConfig", latestConfigSet)
+	return latestConfigSet, nil
+}
+
+// LatestBlockHeight returns the latest block height from the logs
+func (cp *CapRegConfigProvider) LatestBlockHeight(_ context.Context) (blockHeight uint64, err error) {
+	blockHeight, err = strconv.ParseUint(cp.lastSyncedBlockHeight, 10, 64)
+	if err != nil {
+		return 0, err
+	}
+	return blockHeight, err
+}
+
+func (cp *CapRegConfigProvider) Notify() <-chan struct{} {
+	return nil
+}

core/services/relay/evm/cap_reg_config_poller.go

@@ -0,0 +1,30 @@
+package evm
+
+import (
+	"context"
+
+	"github.com/smartcontractkit/chainlink-evm/pkg/config"
+	ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types"
+)
+
+var _ config.ConfigPoller = &CapRegConfigPoller{}
+
+func NewCapRegConfigPoller(configTracker ocrtypes.ContractConfigTracker) *CapRegConfigPoller {
+	return &CapRegConfigPoller{
+		ContractConfigTracker: configTracker,
+	}
+}
+
+type CapRegConfigPoller struct {
+	ocrtypes.ContractConfigTracker
+}
+
+func (cp *CapRegConfigPoller) Start() {}
+
+func (cp *CapRegConfigPoller) Close() error {
+	return nil
+}
+
+func (cp *CapRegConfigPoller) Replay(ctx context.Context, fromBlock int64) error {
+	return nil
+}