From 147fe92c45e7cd5f6387ccd33ef10c266027a8b5 Mon Sep 17 00:00:00 2001 From: amit-momin Date: Wed, 7 Oct 2026 14:22:12 -0500 Subject: [PATCH 1/6] Added remote capability dedupe against local capabilities --- core/capabilities/launcher.go | 33 ++++++++++++++++++++++++++++++ core/capabilities/launcher_test.go | 16 +++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/core/capabilities/launcher.go b/core/capabilities/launcher.go index 2fbb8d3b0c5..4a2a75e9f22 100644 --- a/core/capabilities/launcher.go +++ b/core/capabilities/launcher.go @@ -352,6 +352,11 @@ func (w *launcher) onNewRegistry(ctx context.Context, metadataRegistry *registry belongsToAWorkflowDON := len(myWorkflowDONs) > 0 belongsToACapabilityDON := len(myCapabilityDONs) > 0 + // Prefer capabilities hosted locally over remote ones. If this node serves a capability ID as + // part of a capability DON, never set up a remote shim for the same ID. This is a safety mechanism + // to prevent duplicate capability configs taking down the capability altogether. + remoteCapabilityDONs = filterLocalCapabilitiesFromRemote(remoteCapabilityDONs, myCapabilityDONs) + // Prune shims for capabilities/methods/DONs that are no longer wanted, before the create/update // loops below run. This must happen first: if a capability moved to a different DON ID, its old // CombinedClient must be removed from the registry before the replacement is added under the new @@ -413,6 +418,34 @@ func (w *launcher) onNewRegistry(ctx context.Context, metadataRegistry *registry return nil } +// filterLocalCapabilitiesFromRemote returns remoteDONs with every capability ID that is hosted by one +// of myCapabilityDONs removed from their capability configurations. The inputs are not mutated. +// This method still leaves room for the same capability to exist on different remote DONs. +func filterLocalCapabilitiesFromRemote(remoteDONs, myCapabilityDONs []registry.DON) []registry.DON { + local := map[string]struct{}{} + for _, d := range myCapabilityDONs { + for capID := range d.CapabilityConfigurations { + local[capID] = struct{}{} + } + } + if len(local) == 0 { + return remoteDONs + } + filtered := make([]registry.DON, 0, len(remoteDONs)) + for _, d := range remoteDONs { + configs := make(map[string]registry.CapabilityConfiguration, len(d.CapabilityConfigurations)) + for capID, c := range d.CapabilityConfigurations { + if _, isLocal := local[capID]; isLocal { + continue + } + configs[capID] = c + } + d.CapabilityConfigurations = configs + filtered = append(filtered, d) + } + return filtered +} + func filterDONsByFamilies(donList []registry.DON, myDONFamilies []string) []registry.DON { filteredDONs := []registry.DON{} for _, d := range donList { diff --git a/core/capabilities/launcher_test.go b/core/capabilities/launcher_test.go index d7f91800281..f05ad373cc0 100644 --- a/core/capabilities/launcher_test.go +++ b/core/capabilities/launcher_test.go @@ -1450,3 +1450,19 @@ func TestLauncher_ShardIdentityFromConfig(t *testing.T) { require.NoError(t, l.Close()) }) } + +func TestWithoutLocallyHostedCapabilities(t *testing.T) { + cfg := func() regpkg.CapabilityConfiguration { return regpkg.CapabilityConfiguration{Config: []byte{1}} } + mine := regpkg.DON{ID: 2, CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{"evm@1.0.0": cfg()}} + remote := regpkg.DON{ID: 1, CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{"evm@1.0.0": cfg(), "other@1.0.0": cfg()}} + + got := filterLocalCapabilitiesFromRemote([]regpkg.DON{remote}, []regpkg.DON{mine}) + require.Len(t, got, 1) + assert.Equal(t, uint32(1), got[0].ID) + assert.Contains(t, got[0].CapabilityConfigurations, "other@1.0.0") + assert.NotContains(t, got[0].CapabilityConfigurations, "evm@1.0.0") + assert.Contains(t, remote.CapabilityConfigurations, "evm@1.0.0", "input must not be mutated") + + got = filterLocalCapabilitiesFromRemote([]regpkg.DON{remote}, nil) + assert.Len(t, got[0].CapabilityConfigurations, 2) +} From 06b14a4a411920dd5e9d54cefc68b0669eae9d96 Mon Sep 17 00:00:00 2001 From: amit-momin Date: Wed, 7 Oct 2026 14:50:34 -0500 Subject: [PATCH 2/6] Lint --- core/capabilities/launcher_test.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/core/capabilities/launcher_test.go b/core/capabilities/launcher_test.go index f05ad373cc0..292786d9e3c 100644 --- a/core/capabilities/launcher_test.go +++ b/core/capabilities/launcher_test.go @@ -1452,6 +1452,8 @@ func TestLauncher_ShardIdentityFromConfig(t *testing.T) { } func TestWithoutLocallyHostedCapabilities(t *testing.T) { + t.Parallel() + cfg := func() regpkg.CapabilityConfiguration { return regpkg.CapabilityConfiguration{Config: []byte{1}} } mine := regpkg.DON{ID: 2, CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{"evm@1.0.0": cfg()}} remote := regpkg.DON{ID: 1, CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{"evm@1.0.0": cfg(), "other@1.0.0": cfg()}} From 35c5ed2618ae8421ad39878edca8ad604fb3c9f1 Mon Sep 17 00:00:00 2001 From: amit-momin Date: Wed, 21 Oct 2026 17:06:18 -0500 Subject: [PATCH 3/6] Updated test input --- core/capabilities/launcher_test.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/core/capabilities/launcher_test.go b/core/capabilities/launcher_test.go index 8c5c9581f9c..249eb295481 100644 --- a/core/capabilities/launcher_test.go +++ b/core/capabilities/launcher_test.go @@ -1528,15 +1528,15 @@ func TestWithoutLocallyHostedCapabilities(t *testing.T) { t.Parallel() cfg := func() regpkg.CapabilityConfiguration { return regpkg.CapabilityConfiguration{Config: []byte{1}} } - mine := regpkg.DON{ID: 2, CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{"evm@1.0.0": cfg()}} - remote := regpkg.DON{ID: 1, CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{"evm@1.0.0": cfg(), "other@1.0.0": cfg()}} + mine := regpkg.DON{ID: 2, CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{"evm:ChainSelector:5009297550715157269@1.0.0": cfg()}} + remote := regpkg.DON{ID: 1, CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{"evm:ChainSelector:5009297550715157269@1.0.0": cfg(), "evm:ChainSelector:16015286601757825753@1.0.0": cfg()}} got := filterLocalCapabilitiesFromRemote([]regpkg.DON{remote}, []regpkg.DON{mine}) require.Len(t, got, 1) assert.Equal(t, uint32(1), got[0].ID) - assert.Contains(t, got[0].CapabilityConfigurations, "other@1.0.0") - assert.NotContains(t, got[0].CapabilityConfigurations, "evm@1.0.0") - assert.Contains(t, remote.CapabilityConfigurations, "evm@1.0.0", "input must not be mutated") + assert.Contains(t, got[0].CapabilityConfigurations, "evm:ChainSelector:16015286601757825753@1.0.0") + assert.NotContains(t, got[0].CapabilityConfigurations, "evm:ChainSelector:5009297550715157269@1.0.0") + assert.Contains(t, remote.CapabilityConfigurations, "evm:ChainSelector:5009297550715157269@1.0.0", "input must not be mutated") got = filterLocalCapabilitiesFromRemote([]regpkg.DON{remote}, nil) assert.Len(t, got[0].CapabilityConfigurations, 2) From 55efe447e5afe88de95bcd94e1b25ad7e2bcf924 Mon Sep 17 00:00:00 2001 From: amit-momin Date: Thu, 22 Oct 2026 13:54:52 -0500 Subject: [PATCH 4/6] Added test --- core/capabilities/launcher_test.go | 80 ++++++++++++++++++++++++++++++ 1 file changed, 80 insertions(+) diff --git a/core/capabilities/launcher_test.go b/core/capabilities/launcher_test.go index 249eb295481..8ffb7f3752a 100644 --- a/core/capabilities/launcher_test.go +++ b/core/capabilities/launcher_test.go @@ -1541,3 +1541,83 @@ func TestWithoutLocallyHostedCapabilities(t *testing.T) { got = filterLocalCapabilitiesFromRemote([]regpkg.DON{remote}, nil) assert.Len(t, got[0].CapabilityConfigurations, 2) } + +// A chain capability DON mistakenly configured as a workflow DON (AcceptsWorkflows=true) hosts a capability +// locally. When another DON in the same family is later given a config for the same capability, the node must +// keep serving its local capability and must not try to add a remote shim for the duplicate. +func TestLauncher_OnNewRegistry_PrefersLocalCapabilityOverDuplicateInFamily(t *testing.T) { + t.Parallel() + lggr := logger.Test(t) + reg := regpkg.NewRegistry(lggr) + fullTriggerCapID := "streams-trigger@1.0.0" + mt := newMockTrigger(capabilities.MustNewCapabilityInfo( + fullTriggerCapID, + capabilities.CapabilityTypeTrigger, + "streams trigger", + )) + require.NoError(t, reg.Add(t.Context(), mt)) + + dispatcher := remoteMocks.NewDispatcher(t) + + misconfiguredDonNodes, remoteWorkflowDonNodes, capabilityDonNodes := newNodes(4), newNodes(4), newNodes(4) + triggerCapIDHash := RandomUTF8BytesWord() + misconfiguredDonID := uint32(1) + remoteWfDonID := uint32(2) + capDonID := uint32(3) + + triggerCfg, err := proto.Marshal(&capabilitiespb.CapabilityConfig{ + MethodConfigs: map[string]*capabilitiespb.CapabilityMethodConfig{ + "StreamsTrigger": { + RemoteConfig: &capabilitiespb.CapabilityMethodConfig_RemoteTriggerConfig{ + RemoteTriggerConfig: &capabilitiespb.RemoteTriggerConfig{ + RegistrationRefresh: durationpb.New(1 * time.Second), + MinResponsesToAggregate: 3, + }, + }, + }, + }, + }) + require.NoError(t, err) + + localRegistry := buildLocalRegistry() + // The misconfigured DON: hosts the capability but also accepts workflows. + addDON(localRegistry, misconfiguredDonID, 0, 1, true, true, misconfiguredDonNodes, []string{"zone-a"}, 1, [][32]byte{triggerCapIDHash}) + addCapabilityToDON(localRegistry, misconfiguredDonID, fullTriggerCapID, capabilities.CapabilityTypeTrigger, triggerCfg) + // A genuine remote workflow DON that consumes the capability from this node. + addDON(localRegistry, remoteWfDonID, 0, 1, true, true, remoteWorkflowDonNodes, []string{"zone-a"}, 1, nil) + // A second in-family DON gets a config for the same capability. + addDON(localRegistry, capDonID, 0, 1, true, false, capabilityDonNodes, []string{"zone-a"}, 1, [][32]byte{triggerCapIDHash}) + addCapabilityToDON(localRegistry, capDonID, fullTriggerCapID, capabilities.CapabilityTypeTrigger, triggerCfg) + + sharedPeer := mocks.NewSharedPeer(t) + sharedPeer.On("ID").Return(misconfiguredDonNodes[0]) + sharedPeer.On("IsBootstrap").Return(false) + sharedPeer.On("UpdateConnectionsByDONs", mock.Anything, mock.Anything, mock.Anything).Return(nil) + + launcher, err := NewLauncher( + lggr, + sharedPeer, + nil, + dispatcher, + reg, + &mockDonNotifier{}, limits.Factory{}, + false, 0, + ) + require.NoError(t, err) + servicetest.Run(t, launcher) + + // Only the local publisher is set up. A receiver for the duplicate on capDonID would fail the mock. + dispatcher.On("SetReceiverForMethod", fullTriggerCapID, misconfiguredDonID, "StreamsTrigger", mock.AnythingOfType("*remote.triggerPublisher")).Return(nil).Once() + + require.NoError(t, launcher.OnNewRegistry(t.Context(), localRegistry)) + + assert.Empty(t, launcher.cachedShims.combinedClients, "no remote shim for a locally hosted capability") + assert.Len(t, launcher.cachedShims.triggerPublishers, 1) + + got, err := reg.Get(t.Context(), fullTriggerCapID) + require.NoError(t, err) + info, err := got.Info(t.Context()) + require.NoError(t, err) + assert.True(t, info.IsLocal, "the locally hosted capability must stay registered") + assert.Nil(t, info.DON) +} From f32c2155ecb48d2e751419af0e7816045a81cbff Mon Sep 17 00:00:00 2001 From: amit-momin Date: Thu, 22 Oct 2026 14:03:33 -0500 Subject: [PATCH 5/6] Updated test to look for registry add error log --- core/capabilities/launcher_test.go | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/core/capabilities/launcher_test.go b/core/capabilities/launcher_test.go index 8ffb7f3752a..1cae3cdf36c 100644 --- a/core/capabilities/launcher_test.go +++ b/core/capabilities/launcher_test.go @@ -1547,7 +1547,7 @@ func TestWithoutLocallyHostedCapabilities(t *testing.T) { // keep serving its local capability and must not try to add a remote shim for the duplicate. func TestLauncher_OnNewRegistry_PrefersLocalCapabilityOverDuplicateInFamily(t *testing.T) { t.Parallel() - lggr := logger.Test(t) + lggr, observedLogs := logger.TestObserved(t, zapcore.DebugLevel) reg := regpkg.NewRegistry(lggr) fullTriggerCapID := "streams-trigger@1.0.0" mt := newMockTrigger(capabilities.MustNewCapabilityInfo( @@ -1606,11 +1606,16 @@ func TestLauncher_OnNewRegistry_PrefersLocalCapabilityOverDuplicateInFamily(t *t require.NoError(t, err) servicetest.Run(t, launcher) - // Only the local publisher is set up. A receiver for the duplicate on capDonID would fail the mock. - dispatcher.On("SetReceiverForMethod", fullTriggerCapID, misconfiguredDonID, "StreamsTrigger", mock.AnythingOfType("*remote.triggerPublisher")).Return(nil).Once() + dispatcher.On("SetReceiverForMethod", fullTriggerCapID, mock.Anything, "StreamsTrigger", mock.Anything).Return(nil).Maybe() + dispatcher.On("RemoveReceiverForMethod", fullTriggerCapID, mock.Anything, "StreamsTrigger").Return().Maybe() require.NoError(t, launcher.OnNewRegistry(t.Context(), localRegistry)) + // addRemoteCapabilities swallows registry.Add failures (OnNewRegistry still returns nil), so check the + // logged error: a duplicate remote shim fails with "capability already exists" in the registry. + assert.Empty(t, observedLogs.FilterMessage("failed to add remote capability ").All(), + "adding a remote shim for a locally hosted capability must not even be attempted") + assert.Empty(t, observedLogs.FilterLevelExact(zapcore.ErrorLevel).All()) assert.Empty(t, launcher.cachedShims.combinedClients, "no remote shim for a locally hosted capability") assert.Len(t, launcher.cachedShims.triggerPublishers, 1) From ff95fdcfaa009bb94c3476bed4f0d36025a222dd Mon Sep 17 00:00:00 2001 From: amit-momin Date: Fri, 23 Oct 2026 16:53:09 -0500 Subject: [PATCH 6/6] Moved local remote cap dedupe logic to addRemoteCapabilities --- core/capabilities/launcher.go | 40 ++++++------------------------ core/capabilities/launcher_test.go | 34 +++++++++++++++---------- 2 files changed, 28 insertions(+), 46 deletions(-) diff --git a/core/capabilities/launcher.go b/core/capabilities/launcher.go index a1b88f0e5e8..8f592c84645 100644 --- a/core/capabilities/launcher.go +++ b/core/capabilities/launcher.go @@ -440,11 +440,6 @@ func (w *launcher) onNewRegistry(ctx context.Context, metadataRegistry *registry belongsToAWorkflowDON := len(myWorkflowDONs) > 0 belongsToACapabilityDON := len(myCapabilityDONs) > 0 - // Prefer capabilities hosted locally over remote ones. If this node serves a capability ID as - // part of a capability DON, never set up a remote shim for the same ID. This is a safety mechanism - // to prevent duplicate capability configs taking down the capability altogether. - remoteCapabilityDONs = filterLocalCapabilitiesFromRemote(remoteCapabilityDONs, myCapabilityDONs) - // Prune shims for capabilities/methods/DONs that are no longer wanted, before the create/update // loops below run. This must happen first: if a capability moved to a different DON ID, its old // CombinedClient must be removed from the registry before the replacement is added under the new @@ -517,34 +512,6 @@ func (w *launcher) onNewRegistry(ctx context.Context, metadataRegistry *registry return nil } -// filterLocalCapabilitiesFromRemote returns remoteDONs with every capability ID that is hosted by one -// of myCapabilityDONs removed from their capability configurations. The inputs are not mutated. -// This method still leaves room for the same capability to exist on different remote DONs. -func filterLocalCapabilitiesFromRemote(remoteDONs, myCapabilityDONs []registry.DON) []registry.DON { - local := map[string]struct{}{} - for _, d := range myCapabilityDONs { - for capID := range d.CapabilityConfigurations { - local[capID] = struct{}{} - } - } - if len(local) == 0 { - return remoteDONs - } - filtered := make([]registry.DON, 0, len(remoteDONs)) - for _, d := range remoteDONs { - configs := make(map[string]registry.CapabilityConfiguration, len(d.CapabilityConfigurations)) - for capID, c := range d.CapabilityConfigurations { - if _, isLocal := local[capID]; isLocal { - continue - } - configs[capID] = c - } - d.CapabilityConfigurations = configs - filtered = append(filtered, d) - } - return filtered -} - func filterDONsByFamilies(donList []registry.DON, myDONFamilies []string) []registry.DON { filteredDONs := []registry.DON{} for _, d := range donList { @@ -589,6 +556,13 @@ func donFamiliesOverlap(donA, donB []string) bool { // it is best effort to ensure that valid capabilities are added even if some fail func (w *launcher) addRemoteCapabilities(ctx context.Context, myDON, remoteDON registry.DON, localRegistry *registry.RegistryMetadata, offchainMC *offchainMethodConfigs) { for cid, c := range remoteDON.CapabilityConfigurations { + // Prefer capabilities hosted by my own DON over remote ones, so a duplicate remote shim can + // never replace or collide with the local capability. + if _, isLocal := myDON.CapabilityConfigurations[cid]; isLocal { + w.lggr.Warnw("skipping remote capability hosted locally by my DON", "myDON", myDON, "remoteDON", remoteDON, "capabilityID", cid) + w.metrics.recordRemoteCapabilityAdded(ctx, cid, remoteDON.Name, resultSkipped) + continue + } capabilityConfig, err := offchainMC.configuration(cid, remoteDON, c) if err != nil { w.lggr.Errorw("could not unmarshal capability config", "myDON", myDON, "remoteDON", remoteDON, "capabilityID", cid, "error", err) diff --git a/core/capabilities/launcher_test.go b/core/capabilities/launcher_test.go index 6287e522873..88110fff51f 100644 --- a/core/capabilities/launcher_test.go +++ b/core/capabilities/launcher_test.go @@ -1579,22 +1579,30 @@ func TestConsensusSignersFor(t *testing.T) { }) } -func TestWithoutLocallyHostedCapabilities(t *testing.T) { +func TestLauncher_AddRemoteCapabilities_SkipsCapabilitiesHostedByMyDON(t *testing.T) { t.Parallel() + const localCapID = "evm:ChainSelector:5009297550715157269@1.0.0" + const remoteCapID = "evm:ChainSelector:16015286601757825753@1.0.0" cfg := func() regpkg.CapabilityConfiguration { return regpkg.CapabilityConfiguration{Config: []byte{1}} } - mine := regpkg.DON{ID: 2, CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{"evm:ChainSelector:5009297550715157269@1.0.0": cfg()}} - remote := regpkg.DON{ID: 1, CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{"evm:ChainSelector:5009297550715157269@1.0.0": cfg(), "evm:ChainSelector:16015286601757825753@1.0.0": cfg()}} - - got := filterLocalCapabilitiesFromRemote([]regpkg.DON{remote}, []regpkg.DON{mine}) - require.Len(t, got, 1) - assert.Equal(t, uint32(1), got[0].ID) - assert.Contains(t, got[0].CapabilityConfigurations, "evm:ChainSelector:16015286601757825753@1.0.0") - assert.NotContains(t, got[0].CapabilityConfigurations, "evm:ChainSelector:5009297550715157269@1.0.0") - assert.Contains(t, remote.CapabilityConfigurations, "evm:ChainSelector:5009297550715157269@1.0.0", "input must not be mutated") - - got = filterLocalCapabilitiesFromRemote([]regpkg.DON{remote}, nil) - assert.Len(t, got[0].CapabilityConfigurations, 2) + myDON := regpkg.DON{ID: 2, Name: "mine", CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{localCapID: cfg()}} + remote := regpkg.DON{ID: 1, Name: "remote", CapabilityConfigurations: map[string]regpkg.CapabilityConfiguration{localCapID: cfg(), remoteCapID: cfg()}} + + lggr, observedLogs := logger.TestObserved(t, zapcore.WarnLevel) + launcher, err := NewLauncher(lggr, mocks.NewSharedPeer(t), nil, remoteMocks.NewDispatcher(t), regpkg.NewRegistry(lggr), &mockDonNotifier{}, limits.Factory{}, false, 0) + require.NoError(t, err) + + // The local registry has no capabilities, so the non-local capability fails its lookup. The locally + // hosted one must be skipped before any lookup or parsing happens. + launcher.addRemoteCapabilities(t.Context(), myDON, remote, ®pkg.RegistryMetadata{}, nil) + + skipped := observedLogs.FilterMessage("skipping remote capability hosted locally by my DON").All() + require.Len(t, skipped, 1) + assert.Equal(t, localCapID, skipped[0].ContextMap()["capabilityID"]) + for _, e := range observedLogs.FilterLevelExact(zapcore.ErrorLevel).All() { + assert.NotEqual(t, localCapID, e.ContextMap()["capabilityID"], "locally hosted capability must not be processed as remote") + } + assert.Empty(t, launcher.cachedShims.combinedClients) } // A chain capability DON mistakenly configured as a workflow DON (AcceptsWorkflows=true) hosts a capability