diff --git a/core/capabilities/launcher.go b/core/capabilities/launcher.go index f3eb61d309f..8f592c84645 100644 --- a/core/capabilities/launcher.go +++ b/core/capabilities/launcher.go @@ -556,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 acc433cc527..88110fff51f 100644 --- a/core/capabilities/launcher_test.go +++ b/core/capabilities/launcher_test.go @@ -1578,3 +1578,114 @@ func TestConsensusSignersFor(t *testing.T) { require.ErrorContains(t, err, "multiple consensus capabilities") }) } + +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}} } + 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 +// 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, observedLogs := logger.TestObserved(t, zapcore.DebugLevel) + 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) + + 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) + + 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) +}