oracle factory support (#14813)

* oracle factory support

* lint

* comment

* tidy

core/capabilities/integration_tests/framework/don.go

@@ -81,9 +81,16 @@ type DON struct {
 }
 
 func NewDON(ctx context.Context, t *testing.T, lggr logger.Logger, donConfig DonConfiguration,
-	dependentDONs []commoncap.DON, donContext DonContext) *DON {
+	dependentDONs []commoncap.DON, donContext DonContext, supportsOCR bool) *DON {
 	don := &DON{t: t, lggr: lggr.Named(donConfig.name), config: donConfig, capabilitiesRegistry: donContext.capabilityRegistry}
 
+	var newOracleFactoryFn standardcapabilities.NewOracleFactoryFn
+	var libOcr *MockLibOCR
+	if supportsOCR {
+		libOcr = NewMockLibOCR(t, lggr, donConfig.F, 1*time.Second)
+		servicetest.Run(t, libOcr)
+	}
+
 	for i, member := range donConfig.Members {
 		dispatcher := donContext.p2pNetwork.NewDispatcherForNode(member)
 		capabilityRegistry := capabilities.NewRegistry(lggr)
@@ -102,10 +109,15 @@ func NewDON(ctx context.Context, t *testing.T, lggr logger.Logger, donConfig Don
 		}
 		don.nodes = append(don.nodes, cn)
 
+		if supportsOCR {
+			factory := newMockLibOcrOracleFactory(libOcr, donConfig.KeyBundles[i], len(donConfig.Members), int(donConfig.F))
+			newOracleFactoryFn = factory.NewOracleFactory
+		}
+
 		cn.start = func() {
 			node := startNewNode(ctx, t, lggr.Named(donConfig.name+"-"+strconv.Itoa(i)), nodeInfo, donContext.EthBlockchain,
 				donContext.capabilityRegistry.getAddress(), dispatcher,
-				peerWrapper{peer: p2pPeer{member}}, capabilityRegistry,
+				peerWrapper{peer: p2pPeer{member}}, capabilityRegistry, newOracleFactoryFn,
 				donConfig.keys[i], func(c *chainlink.Config) {
 					for _, modifier := range don.nodeConfigModifiers {
 						modifier(c, cn)
@@ -172,7 +184,7 @@ func (d *DON) Start(ctx context.Context, t *testing.T) {
 	}
 
 	if d.addOCR3NonStandardCapability {
-		libocr := NewMockLibOCR(t, d.config.F, 1*time.Second)
+		libocr := NewMockLibOCR(t, d.lggr, d.config.F, 1*time.Second)
 		servicetest.Run(t, libocr)
 
 		for _, node := range d.nodes {
@@ -265,6 +277,7 @@ func startNewNode(ctx context.Context,
 	dispatcher remotetypes.Dispatcher,
 	peerWrapper p2ptypes.PeerWrapper,
 	localCapabilities *capabilities.Registry,
+	newOracleFactoryFn standardcapabilities.NewOracleFactoryFn,
 	keyV2 ethkey.KeyV2,
 	setupCfg func(c *chainlink.Config),
 ) *cltest.TestApplication {
@@ -295,7 +308,7 @@ func startNewNode(ctx context.Context,
 	ethBlockchain.Commit()
 
 	return cltest.NewApplicationWithConfigV2AndKeyOnSimulatedBlockchain(t, config, ethBlockchain.SimulatedBackend, nodeInfo,
-		dispatcher, peerWrapper, localCapabilities, keyV2, lggr)
+		dispatcher, peerWrapper, newOracleFactoryFn, localCapabilities, keyV2, lggr)
 }
 
 // Functions below this point are for adding non-standard capabilities to a DON, deliberately verbose. Eventually these

core/capabilities/integration_tests/framework/mock_libocr.go

@@ -9,28 +9,86 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/require"
+	"github.com/google/uuid"
 
 	"github.com/smartcontractkit/libocr/commontypes"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/types"
 
-	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/consensus/ocr3"
+	coretypes "github.com/smartcontractkit/chainlink-common/pkg/types/core"
+	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/generic"
+
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
 	"github.com/smartcontractkit/chainlink/v2/core/services/keystore/keys/ocr2key"
+
+	"github.com/smartcontractkit/chainlink/v2/core/logger"
 )
 
+type oracleFactoryFactory struct {
+	mockLibOCr *MockLibOCR
+	key        ocr2key.KeyBundle
+	N          int
+	F          int
+}
+
+func newMockLibOcrOracleFactory(mockLibOCr *MockLibOCR, key ocr2key.KeyBundle, N int, F int) *oracleFactoryFactory {
+	return &oracleFactoryFactory{
+		mockLibOCr: mockLibOCr,
+		key:        key,
+		N:          N,
+		F:          F,
+	}
+}
+
+func (o *oracleFactoryFactory) NewOracleFactory(params generic.OracleFactoryParams) (coretypes.OracleFactory, error) {
+	return &mockOracleFactory{o}, nil
+}
+
+type mockOracle struct {
+	*mockOracleFactory
+	args         coretypes.OracleArgs
+	libocrNodeID string
+}
+
+func (m *mockOracle) Start(ctx context.Context) error {
+	plugin, _, err := m.args.ReportingPluginFactoryService.NewReportingPlugin(ctx, ocr3types.ReportingPluginConfig{
+		F: m.F,
+		N: m.N,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to create reporting plugin: %w", err)
+	}
+
+	m.libocrNodeID = m.mockLibOCr.AddNode(plugin, m.args.ContractTransmitter, m.key)
+	return nil
+}
+
+func (m *mockOracle) Close(ctx context.Context) error {
+	m.mockLibOCr.RemoveNode(m.libocrNodeID)
+	return nil
+}
+
+type mockOracleFactory struct {
+	*oracleFactoryFactory
+}
+
+func (m *mockOracleFactory) NewOracle(ctx context.Context, args coretypes.OracleArgs) (coretypes.Oracle, error) {
+	return &mockOracle{mockOracleFactory: m, args: args}, nil
+}
+
 type libocrNode struct {
+	id string
 	ocr3types.ReportingPlugin[[]byte]
-	*ocr3.ContractTransmitter
+	ocr3types.ContractTransmitter[[]byte]
 	key ocr2key.KeyBundle
 }
 
 // MockLibOCR is a mock libocr implementation for testing purposes that simulates libocr protocol rounds without having
 // to setup the libocr network
 type MockLibOCR struct {
 	services.StateMachine
-	t *testing.T
+	t    *testing.T
+	lggr logger.Logger
 
 	nodes                 []*libocrNode
 	f                     uint8
@@ -39,15 +97,17 @@ type MockLibOCR struct {
 	seqNr      uint64
 	outcomeCtx ocr3types.OutcomeContext
 
+	mux    sync.Mutex
 	stopCh services.StopChan
 	wg     sync.WaitGroup
 }
 
-func NewMockLibOCR(t *testing.T, f uint8, protocolRoundInterval time.Duration) *MockLibOCR {
+func NewMockLibOCR(t *testing.T, lggr logger.Logger, f uint8, protocolRoundInterval time.Duration) *MockLibOCR {
 	return &MockLibOCR{
-		t: t,
-		f: f, outcomeCtx: ocr3types.OutcomeContext{
-			SeqNr:           0,
+		t:    t,
+		lggr: lggr,
+		f:    f, outcomeCtx: ocr3types.OutcomeContext{
+			SeqNr:           1,
 			PreviousOutcome: nil,
 			Epoch:           0,
 			Round:           0,
@@ -75,7 +135,7 @@ func (m *MockLibOCR) Start(ctx context.Context) error {
 				case <-ticker.C:
 					err := m.simulateProtocolRound(ctx)
 					if err != nil {
-						require.FailNow(m.t, err.Error())
+						m.lggr.Errorf("simulating protocol round: %v", err)
 					}
 				}
 			}
@@ -92,11 +152,31 @@ func (m *MockLibOCR) Close() error {
 	})
 }
 
-func (m *MockLibOCR) AddNode(plugin ocr3types.ReportingPlugin[[]byte], transmitter *ocr3.ContractTransmitter, key ocr2key.KeyBundle) {
-	m.nodes = append(m.nodes, &libocrNode{plugin, transmitter, key})
+func (m *MockLibOCR) AddNode(plugin ocr3types.ReportingPlugin[[]byte], transmitter ocr3types.ContractTransmitter[[]byte], key ocr2key.KeyBundle) string {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+	node := &libocrNode{uuid.New().String(), plugin, transmitter, key}
+	m.nodes = append(m.nodes, node)
+	return node.id
+}
+
+func (m *MockLibOCR) RemoveNode(id string) {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+
+	var updatedNodes []*libocrNode
+	for _, node := range m.nodes {
+		if node.id != id {
+			updatedNodes = append(updatedNodes, node)
+		}
+	}
+
+	m.nodes = updatedNodes
 }
 
 func (m *MockLibOCR) simulateProtocolRound(ctx context.Context) error {
+	m.mux.Lock()
+	defer m.mux.Unlock()
 	if len(m.nodes) == 0 {
 		return nil
 	}
@@ -114,7 +194,7 @@ func (m *MockLibOCR) simulateProtocolRound(ctx context.Context) error {
 	for oracleID, node := range m.nodes {
 		obs, err2 := node.Observation(ctx, m.outcomeCtx, query)
 		if err2 != nil {
-			return fmt.Errorf("failed to get observation: %w", err)
+			return fmt.Errorf("failed to get observation: %w", err2)
 		}
 
 		observations = append(observations, types.AttributedObservation{
@@ -127,7 +207,7 @@ func (m *MockLibOCR) simulateProtocolRound(ctx context.Context) error {
 	for _, node := range m.nodes {
 		outcome, err2 := node.Outcome(ctx, m.outcomeCtx, query, observations)
 		if err2 != nil {
-			return fmt.Errorf("failed to get outcome: %w", err)
+			return fmt.Errorf("failed to get outcome: %w", err2)
 		}
 
 		if len(outcome) == 0 {

core/capabilities/integration_tests/keystone/setup.go

@@ -56,7 +56,7 @@ func setupKeystoneDons(ctx context.Context, t *testing.T, lggr logger.SugaredLog
 func createKeystoneTriggerDon(ctx context.Context, t *testing.T, lggr logger.SugaredLogger, triggerDonInfo framework.DonConfiguration,
 	donContext framework.DonContext, trigger framework.TriggerFactory) *framework.DON {
 	triggerDon := framework.NewDON(ctx, t, lggr, triggerDonInfo,
-		[]commoncap.DON{}, donContext)
+		[]commoncap.DON{}, donContext, false)
 
 	triggerDon.AddExternalTriggerCapability(trigger)
 	triggerDon.Initialise()
@@ -65,7 +65,7 @@ func createKeystoneTriggerDon(ctx context.Context, t *testing.T, lggr logger.Sug
 
 func createKeystoneWriteTargetDon(ctx context.Context, t *testing.T, lggr logger.SugaredLogger, targetDonInfo framework.DonConfiguration, donContext framework.DonContext, forwarderAddr common.Address) *framework.DON {
 	writeTargetDon := framework.NewDON(ctx, t, lggr, targetDonInfo,
-		[]commoncap.DON{}, donContext)
+		[]commoncap.DON{}, donContext, false)
 	err := writeTargetDon.AddEthereumWriteTargetNonStandardCapability(forwarderAddr)
 	require.NoError(t, err)
 	writeTargetDon.Initialise()
@@ -76,7 +76,7 @@ func createKeystoneWorkflowDon(ctx context.Context, t *testing.T, lggr logger.Su
 	triggerDonInfo framework.DonConfiguration, targetDonInfo framework.DonConfiguration, donContext framework.DonContext) *framework.DON {
 	workflowDon := framework.NewDON(ctx, t, lggr, workflowDonInfo,
 		[]commoncap.DON{triggerDonInfo.DON, targetDonInfo.DON},
-		donContext)
+		donContext, true)
 
 	workflowDon.AddOCR3NonStandardCapability()
 	workflowDon.Initialise()

core/internal/cltest/cltest.go

@@ -37,6 +37,8 @@ import (
 
 	ocrtypes "github.com/smartcontractkit/libocr/offchainreporting/types"
 
+	"github.com/smartcontractkit/chainlink/v2/core/services/standardcapabilities"
+
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities"
 	remotetypes "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types"
 	p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types"
@@ -326,6 +328,15 @@ func NewApplicationWithConfig(t testing.TB, cfg chainlink.GeneralConfig, flagsAn
 		auditLogger = audit.NoopLogger
 	}
 
+	var newOracleFactoryFn standardcapabilities.NewOracleFactoryFn
+	for _, dep := range flagsAndDeps {
+		factoryFn, _ := dep.(standardcapabilities.NewOracleFactoryFn)
+		if factoryFn != nil {
+			newOracleFactoryFn = factoryFn
+			break
+		}
+	}
+
 	var capabilitiesRegistry *capabilities.Registry
 	capabilitiesRegistry = capabilities.NewRegistry(lggr)
 	for _, dep := range flagsAndDeps {
@@ -477,6 +488,7 @@ func NewApplicationWithConfig(t testing.TB, cfg chainlink.GeneralConfig, flagsAn
 		CapabilitiesRegistry:       capabilitiesRegistry,
 		CapabilitiesDispatcher:     dispatcher,
 		CapabilitiesPeerWrapper:    peerWrapper,
+		NewOracleFactoryFn:         newOracleFactoryFn,
 	})
 
 	require.NoError(t, err)

core/services/chainlink/application.go

@@ -26,7 +26,6 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/utils"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/jsonserializable"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/mailbox"
-
 	"github.com/smartcontractkit/chainlink/v2/core/bridges"
 	"github.com/smartcontractkit/chainlink/v2/core/build"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities"
@@ -186,6 +185,7 @@ type ApplicationOpts struct {
 	CapabilitiesRegistry       *capabilities.Registry
 	CapabilitiesDispatcher     remotetypes.Dispatcher
 	CapabilitiesPeerWrapper    p2ptypes.PeerWrapper
+	NewOracleFactoryFn         standardcapabilities.NewOracleFactoryFn
 }
 
 // NewApplication initializes a new store if one is not already
@@ -504,6 +504,7 @@ func NewApplication(opts ApplicationOpts) (Application, error) {
 		gatewayConnectorWrapper,
 		keyStore,
 		peerWrapper,
+		opts.NewOracleFactoryFn,
 	)
 
 	if cfg.OCR().Enabled() {