Support "dummy" relayer type

.changeset/tender-carpets-cry.md

@@ -0,0 +1,7 @@
+---
+"chainlink": patch
+---
+
+Add new relayer type "dummy" for testing.
+
+#added

core/chains/evm/logpoller/log_poller.go

@@ -1021,6 +1021,11 @@ func (lp *logPoller) latestBlocks(ctx context.Context) (*evmtypes.Head, int64, e
 		if err != nil {
 			return nil, 0, err
 		}
+		if latestBlock == nil {
+			// Shouldn't happen with a real client, but still better rather to
+			// return error than panic
+			return nil, 0, errors.New("latest block is nil")
+		}
 		// If chain has fewer blocks than finalityDepth, return 0
 		return latestBlock, mathutil.Max(latestBlock.Number-lp.finalityDepth, 0), nil
 	}

core/cmd/shell.go

@@ -184,7 +184,7 @@ func (n ChainlinkAppFactory) NewApplication(ctx context.Context, cfg chainlink.G
 	}
 	// evm always enabled for backward compatibility
 	// TODO BCF-2510 this needs to change in order to clear the path for EVM extraction
-	initOps := []chainlink.CoreRelayerChainInitFunc{chainlink.InitEVM(ctx, relayerFactory, evmFactoryCfg)}
+	initOps := []chainlink.CoreRelayerChainInitFunc{chainlink.InitDummy(ctx, relayerFactory), chainlink.InitEVM(ctx, relayerFactory, evmFactoryCfg)}
 
 	if cfg.CosmosEnabled() {
 		cosmosCfg := chainlink.CosmosFactoryConfig{

core/internal/cltest/cltest.go

@@ -387,7 +387,7 @@ func NewApplicationWithConfig(t testing.TB, cfg chainlink.GeneralConfig, flagsAn
 
 	testCtx := testutils.Context(t)
 	// evm alway enabled for backward compatibility
-	initOps := []chainlink.CoreRelayerChainInitFunc{chainlink.InitEVM(testCtx, relayerFactory, evmOpts)}
+	initOps := []chainlink.CoreRelayerChainInitFunc{chainlink.InitDummy(testCtx, relayerFactory), chainlink.InitEVM(testCtx, relayerFactory, evmOpts)}
 
 	if cfg.CosmosEnabled() {
 		cosmosCfg := chainlink.CosmosFactoryConfig{

core/internal/testutils/testutils.go

@@ -367,15 +367,17 @@ func RequireLogMessage(t *testing.T, observedLogs *observer.ObservedLogs, msg st
 //
 //	observedZapCore, observedLogs := observer.New(zap.DebugLevel)
 //	lggr := logger.TestLogger(t, observedZapCore)
-func WaitForLogMessage(t *testing.T, observedLogs *observer.ObservedLogs, msg string) {
+func WaitForLogMessage(t *testing.T, observedLogs *observer.ObservedLogs, msg string) (le observer.LoggedEntry) {
 	AssertEventually(t, func() bool {
 		for _, l := range observedLogs.All() {
 			if strings.Contains(l.Message, msg) {
+				le = l
 				return true
 			}
 		}
 		return false
 	})
+	return
 }
 
 // WaitForLogMessageCount waits until at least count log message containing the

core/services/chainlink/relayer_chain_interoperators.go

@@ -69,13 +69,19 @@ type ChainsNodesStatuser interface {
 
 var _ RelayerChainInteroperators = &CoreRelayerChainInteroperators{}
 
+type DummyFactory interface {
+	NewDummy(config DummyFactoryConfig) (loop.Relayer, error)
+}
+
 // CoreRelayerChainInteroperators implements [RelayerChainInteroperators]
 // as needed for the core [chainlink.Application]
 type CoreRelayerChainInteroperators struct {
 	mu           sync.Mutex
 	loopRelayers map[types.RelayID]loop.Relayer
 	legacyChains legacyChains
 
+	dummyFactory DummyFactory
+
 	// we keep an explicit list of services because the legacy implementations have more than
 	// just the relayer service
 	srvs []services.ServiceCtx
@@ -98,6 +104,14 @@ func NewCoreRelayerChainInteroperators(initFuncs ...CoreRelayerChainInitFunc) (*
 // CoreRelayerChainInitFunc is a hook in the constructor to create relayers from a factory.
 type CoreRelayerChainInitFunc func(op *CoreRelayerChainInteroperators) error
 
+// InitDummy instantiates a dummy relayer
+func InitDummy(ctx context.Context, factory RelayerFactory) CoreRelayerChainInitFunc {
+	return func(op *CoreRelayerChainInteroperators) error {
+		op.dummyFactory = &factory
+		return nil
+	}
+}
+
 // InitEVM is a option for instantiating evm relayers
 func InitEVM(ctx context.Context, factory RelayerFactory, config EVMFactoryConfig) CoreRelayerChainInitFunc {
 	return func(op *CoreRelayerChainInteroperators) (err error) {
@@ -178,6 +192,16 @@ func (rs *CoreRelayerChainInteroperators) Get(id types.RelayID) (loop.Relayer, e
 	defer rs.mu.Unlock()
 	lr, exist := rs.loopRelayers[id]
 	if !exist {
+		// lazily create dummy relayers
+		if id.Network == "dummy" {
+			var err error
+			lr, err = rs.dummyFactory.NewDummy(DummyFactoryConfig{id.ChainID})
+			if err != nil {
+				return nil, err
+			}
+			rs.loopRelayers[id] = lr
+			return lr, nil
+		}
 		return nil, fmt.Errorf("%w: %s", ErrNoSuchRelayer, id)
 	}
 	return lr, nil

core/services/chainlink/relayer_chain_interoperators_test.go

@@ -198,6 +198,10 @@ func TestCoreRelayerChainInteroperators(t *testing.T) {
 		expectedStarknetNodeCnt    int
 		expectedStarknetRelayerIds []types.RelayID
 
+		expectedDummyChainCnt   int
+		expectedDummyNodeCnt    int
+		expectedDummyRelayerIds []types.RelayID
+
 		expectedCosmosChainCnt   int
 		expectedCosmosNodeCnt    int
 		expectedCosmosRelayerIds []types.RelayID
@@ -368,6 +372,8 @@ func TestCoreRelayerChainInteroperators(t *testing.T) {
 					expectedChainCnt, expectedNodeCnt = tt.expectedSolanaChainCnt, tt.expectedSolanaNodeCnt
 				case relay.NetworkStarkNet:
 					expectedChainCnt, expectedNodeCnt = tt.expectedStarknetChainCnt, tt.expectedStarknetNodeCnt
+				case relay.NetworkDummy:
+					expectedChainCnt, expectedNodeCnt = tt.expectedDummyChainCnt, tt.expectedDummyNodeCnt
 				case relay.NetworkAptos:
 					t.Skip("aptos doesn't need a CoreRelayerChainInteroperator")
 

core/services/chainlink/relayer_factory.go

@@ -26,6 +26,7 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
 	"github.com/smartcontractkit/chainlink/v2/core/services/keystore"
 	corerelay "github.com/smartcontractkit/chainlink/v2/core/services/relay"
+	"github.com/smartcontractkit/chainlink/v2/core/services/relay/dummy"
 	evmrelay "github.com/smartcontractkit/chainlink/v2/core/services/relay/evm"
 	"github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/mercury/wsrpc"
 	"github.com/smartcontractkit/chainlink/v2/plugins"
@@ -39,6 +40,14 @@ type RelayerFactory struct {
 	CapabilitiesRegistry *capabilities.Registry
 }
 
+type DummyFactoryConfig struct {
+	ChainID string
+}
+
+func (r *RelayerFactory) NewDummy(config DummyFactoryConfig) (loop.Relayer, error) {
+	return dummy.NewRelayer(r.Logger, config.ChainID), nil
+}
+
 type EVMFactoryConfig struct {
 	legacyevm.ChainOpts
 	evmrelay.CSAETHKeystore

core/services/llo/bm/dummy_transmitter.go

@@ -2,8 +2,6 @@ package bm
 
 import (
 	"context"
-	"crypto/ed25519"
-	"fmt"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -37,10 +35,10 @@ type transmitter struct {
 	fromAccount string
 }
 
-func NewTransmitter(lggr logger.Logger, fromAccount ed25519.PublicKey) Transmitter {
+func NewTransmitter(lggr logger.Logger, fromAccount string) Transmitter {
 	return &transmitter{
 		lggr.Named("DummyTransmitter"),
-		fmt.Sprintf("%x", fromAccount),
+		fromAccount,
 	}
 }
 

core/services/llo/bm/dummy_transmitter_test.go

@@ -1,7 +1,6 @@
 package bm
 
 import (
-	"crypto/ed25519"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -19,7 +18,7 @@ import (
 
 func Test_DummyTransmitter(t *testing.T) {
 	lggr, observedLogs := logger.TestLoggerObserved(t, zapcore.DebugLevel)
-	tr := NewTransmitter(lggr, ed25519.PublicKey("dummy"))
+	tr := NewTransmitter(lggr, "dummy")
 
 	servicetest.Run(t, tr)
 

core/services/llo/data_source.go

@@ -61,6 +61,8 @@ func (d *dataSource) Observe(ctx context.Context, streamIDs map[llotypes.StreamI
 	sv := make(llo.StreamValues)
 	var mu sync.Mutex
 
+	d.lggr.Debugw("Observing streams", "streamIDs", streamIDs)
+
 	for streamID := range streamIDs {
 		go func(streamID llotypes.StreamID) {
 			defer wg.Done()
@@ -99,5 +101,7 @@ func (d *dataSource) Observe(ctx context.Context, streamIDs map[llotypes.StreamI
 
 	wg.Wait()
 
+	d.lggr.Debugw("Observed streams", "streamIDs", streamIDs, "values", sv)
+
 	return sv, nil
 }

core/services/llo/onchain_config.go

@@ -0,0 +1,21 @@
+package llo
+
+type OnchainConfig struct{}
+
+type OnchainConfigCodec interface {
+	Encode(OnchainConfig) ([]byte, error)
+	Decode([]byte) (OnchainConfig, error)
+}
+
+var _ OnchainConfigCodec = &JSONOnchainConfigCodec{}
+
+// TODO: Replace this with protobuf, if it is actually used for something
+type JSONOnchainConfigCodec struct{}
+
+func (c *JSONOnchainConfigCodec) Encode(OnchainConfig) ([]byte, error) {
+	return nil, nil
+}
+
+func (c *JSONOnchainConfigCodec) Decode([]byte) (OnchainConfig, error) {
+	return OnchainConfig{}, nil
+}

core/services/ocr2/delegate.go

@@ -802,7 +802,7 @@ func (d *Delegate) newServicesMercury(
 		return nil, ErrJobSpecNoRelayer{Err: err, PluginName: "mercury"}
 	}
 	if rid.Network != relay.NetworkEVM {
-		return nil, fmt.Errorf("mercury services: expected EVM relayer got %s", rid.Network)
+		return nil, fmt.Errorf("mercury services: expected EVM relayer got %q", rid.Network)
 	}
 	relayer, err := d.RelayGetter.Get(rid)
 	if err != nil {
@@ -895,9 +895,6 @@ func (d *Delegate) newServicesLLO(
 	if err != nil {
 		return nil, ErrJobSpecNoRelayer{Err: err, PluginName: "streams"}
 	}
-	if rid.Network != relay.NetworkEVM {
-		return nil, fmt.Errorf("streams services: expected EVM relayer got %s", rid.Network)
-	}
 	relayer, err := d.RelayGetter.Get(rid)
 	if err != nil {
 		return nil, ErrRelayNotEnabled{Err: err, Relay: spec.Relay, PluginName: "streams"}
@@ -1079,7 +1076,7 @@ func (d *Delegate) newServicesDKG(
 		return nil, ErrJobSpecNoRelayer{Err: err, PluginName: "DKG"}
 	}
 	if rid.Network != relay.NetworkEVM {
-		return nil, fmt.Errorf("DKG services: expected EVM relayer got %s", rid.Network)
+		return nil, fmt.Errorf("DKG services: expected EVM relayer got %q", rid.Network)
 	}
 
 	chain, err2 := d.legacyChains.Get(rid.ChainID)
@@ -1144,7 +1141,7 @@ func (d *Delegate) newServicesOCR2VRF(
 		return nil, ErrJobSpecNoRelayer{Err: err, PluginName: "VRF"}
 	}
 	if rid.Network != relay.NetworkEVM {
-		return nil, fmt.Errorf("VRF services: expected EVM relayer got %s", rid.Network)
+		return nil, fmt.Errorf("VRF services: expected EVM relayer got %q", rid.Network)
 	}
 	chain, err2 := d.legacyChains.Get(rid.ChainID)
 	if err2 != nil {
@@ -1358,7 +1355,7 @@ func (d *Delegate) newServicesOCR2Keepers21(
 		return nil, ErrJobSpecNoRelayer{Err: err, PluginName: "keeper2"}
 	}
 	if rid.Network != relay.NetworkEVM {
-		return nil, fmt.Errorf("keeper2 services: expected EVM relayer got %s", rid.Network)
+		return nil, fmt.Errorf("keeper2 services: expected EVM relayer got %q", rid.Network)
 	}
 
 	transmitterID := spec.TransmitterID.String
@@ -1511,7 +1508,7 @@ func (d *Delegate) newServicesOCR2Keepers20(
 		return nil, ErrJobSpecNoRelayer{Err: err, PluginName: "keepers2.0"}
 	}
 	if rid.Network != relay.NetworkEVM {
-		return nil, fmt.Errorf("keepers2.0 services: expected EVM relayer got %s", rid.Network)
+		return nil, fmt.Errorf("keepers2.0 services: expected EVM relayer got %q", rid.Network)
 	}
 	chain, err2 := d.legacyChains.Get(rid.ChainID)
 	if err2 != nil {
@@ -1639,7 +1636,7 @@ func (d *Delegate) newServicesOCR2Functions(
 		return nil, ErrJobSpecNoRelayer{Err: err, PluginName: "functions"}
 	}
 	if rid.Network != relay.NetworkEVM {
-		return nil, fmt.Errorf("functions services: expected EVM relayer got %s", rid.Network)
+		return nil, fmt.Errorf("functions services: expected EVM relayer got %q", rid.Network)
 	}
 	chain, err := d.legacyChains.Get(rid.ChainID)
 	if err != nil {

core/services/ocr2/plugins/llo/config/config.go

@@ -26,7 +26,7 @@ type PluginConfig struct {
 	ChannelDefinitionsContractFromBlock int64          `json:"channelDefinitionsContractFromBlock" toml:"channelDefinitionsContractFromBlock"`
 
 	// NOTE: ChannelDefinitions is an override.
-	// If Channe}lDefinitions is specified, values for
+	// If ChannelDefinitions is specified, values for
 	// ChannelDefinitionsContractAddress and
 	// ChannelDefinitionsContractFromBlock will be ignored
 	ChannelDefinitions string `json:"channelDefinitions" toml:"channelDefinitions"`
@@ -41,6 +41,10 @@ type PluginConfig struct {
 	KeyBundleIDs map[string]string `json:"keyBundleIDs" toml:"keyBundleIDs"`
 }
 
+func (p *PluginConfig) Unmarshal(data []byte) error {
+	return json.Unmarshal(data, p)
+}
+
 func (p PluginConfig) Validate() (merr error) {
 	if p.RawServerURL == "" {
 		merr = errors.New("llo: ServerURL must be specified")