WS URL can be optional when LogBroadcaster is disabled

.changeset/silly-lies-boil.md

@@ -0,0 +1,8 @@
+---
+"chainlink": minor
+---
+
+Make websocket URL flag `WSURL` for `EVM.Nodes`, and apply logic so that:
+* If WS URL was not provided, SubscribeFilterLogs should fail with an explicit error
+* If WS URL was not provided LogBroadcaster should be disabled 
+#internal

core/chains/evm/client/config_builder.go

@@ -1,6 +1,7 @@
 package client
 
 import (
+	"errors"
 	"fmt"
 	"net/url"
 	"time"
@@ -15,6 +16,8 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/config/toml"
 )
 
+var ErrLogBroadcasterEnabledWithoutWSURL = errors.New("logbroadcaster cannot be enabled unless all nodes have WSURL provided")
+
 type NodeConfig struct {
 	Name     *string
 	WSURL    *string
@@ -43,12 +46,18 @@ func NewClientConfigs(
 	deathDeclarationDelay time.Duration,
 	noNewFinalizedHeadsThreshold time.Duration,
 	finalizedBlockPollInterval time.Duration,
+	LogBroadcasterEnabled bool,
 
 ) (commonclient.ChainConfig, evmconfig.NodePool, []*toml.Node, error) {
 	nodes, err := parseNodeConfigs(nodeCfgs)
 	if err != nil {
 		return nil, nil, nil, err
 	}
+
+	if err = verifyLogBroadcasterFlag(nodes, LogBroadcasterEnabled); err != nil {
+		return nil, nil, nil, err
+	}
+
 	nodePool := toml.NodePool{
 		SelectionMode:              selectionMode,
 		LeaseDuration:              commonconfig.MustNewDuration(leaseDuration),
@@ -76,18 +85,37 @@ func NewClientConfigs(
 	return chainConfig, nodePoolCfg, nodes, nil
 }
 
+// verifyLogBroadcasterFlag checks node config and return error if LogBroadcaster enabled but some node missing WSURL
+func verifyLogBroadcasterFlag(nodes []*toml.Node, LogBroadcasterEnabled bool) error {
+	if !LogBroadcasterEnabled {
+		return nil
+	}
+	for _, node := range nodes {
+		if node.WSURL == nil || node.WSURL.IsZero() {
+			return ErrLogBroadcasterEnabledWithoutWSURL
+		}
+	}
+	return nil
+}
+
 func parseNodeConfigs(nodeCfgs []NodeConfig) ([]*toml.Node, error) {
 	nodes := make([]*toml.Node, len(nodeCfgs))
 	for i, nodeCfg := range nodeCfgs {
-		if nodeCfg.WSURL == nil || nodeCfg.HTTPURL == nil {
-			return nil, fmt.Errorf("node config [%d]: missing WS or HTTP URL", i)
+		var wsURL, httpURL *commonconfig.URL
+		// wsUrl is optional
+		if nodeCfg.WSURL != nil {
+			wsURL = commonconfig.MustParseURL(*nodeCfg.WSURL)
 		}
-		wsUrl := commonconfig.MustParseURL(*nodeCfg.WSURL)
-		httpUrl := commonconfig.MustParseURL(*nodeCfg.HTTPURL)
+
+		if nodeCfg.HTTPURL == nil {
+			return nil, fmt.Errorf("node config [%d]: HTTP URL", i)
+		}
+
+		httpURL = commonconfig.MustParseURL(*nodeCfg.HTTPURL)
 		node := &toml.Node{
 			Name:     nodeCfg.Name,
-			WSURL:    wsUrl,
-			HTTPURL:  httpUrl,
+			WSURL:    wsURL,
+			HTTPURL:  httpURL,
 			SendOnly: nodeCfg.SendOnly,
 			Order:    nodeCfg.Order,
 		}

core/chains/evm/client/config_builder_test.go

@@ -39,7 +39,7 @@ func TestClientConfigBuilder(t *testing.T) {
 	noNewHeadsThreshold := time.Second
 	chainCfg, nodePool, nodes, err := client.NewClientConfigs(selectionMode, leaseDuration, chainTypeStr, nodeConfigs,
 		pollFailureThreshold, pollInterval, syncThreshold, nodeIsSyncingEnabled, noNewHeadsThreshold, finalityDepth,
-		finalityTagEnabled, finalizedBlockOffset, enforceRepeatableRead, deathDeclarationDelay, noNewFinalizedBlocksThreshold, pollInterval)
+		finalityTagEnabled, finalizedBlockOffset, enforceRepeatableRead, deathDeclarationDelay, noNewFinalizedBlocksThreshold, pollInterval, true)
 	require.NoError(t, err)
 
 	// Validate node pool configs
@@ -90,15 +90,15 @@ func TestNodeConfigs(t *testing.T) {
 		require.Len(t, tomlNodes, len(nodeConfigs))
 	})
 
-	t.Run("parsing missing ws url fails", func(t *testing.T) {
+	t.Run("ws is optional", func(t *testing.T) {
 		nodeConfigs := []client.NodeConfig{
 			{
 				Name:    ptr("foo1"),
 				HTTPURL: ptr("http://foo1.test"),
 			},
 		}
 		_, err := client.ParseTestNodeConfigs(nodeConfigs)
-		require.Error(t, err)
+		require.Nil(t, err)
 	})
 
 	t.Run("parsing missing http url fails", func(t *testing.T) {

core/chains/evm/client/evm_client.go

@@ -20,7 +20,7 @@ func NewEvmClient(cfg evmconfig.NodePool, chainCfg commonclient.ChainConfig, cli
 	var sendonlys []commonclient.SendOnlyNode[*big.Int, RPCClient]
 	largePayloadRPCTimeout, defaultRPCTimeout := getRPCTimeouts(chainType)
 	for i, node := range nodes {
-		if node.SendOnly != nil && *node.SendOnly {
+		if node.WSURL == nil || (node.SendOnly != nil && *node.SendOnly) {
 			rpc := NewRPCClient(lggr, empty, (*url.URL)(node.HTTPURL), *node.Name, int32(i), chainID,
 				commonclient.Secondary, cfg.FinalizedBlockPollInterval(), largePayloadRPCTimeout, defaultRPCTimeout, chainType)
 			sendonly := commonclient.NewSendOnlyNode(lggr, (url.URL)(*node.HTTPURL),

core/chains/evm/client/evm_client_test.go

@@ -40,9 +40,20 @@ func TestNewEvmClient(t *testing.T) {
 	finalityTagEnabled := ptr(true)
 	chainCfg, nodePool, nodes, err := client.NewClientConfigs(selectionMode, leaseDuration, chainTypeStr, nodeConfigs,
 		pollFailureThreshold, pollInterval, syncThreshold, nodeIsSyncingEnabled, noNewHeadsThreshold, finalityDepth,
-		finalityTagEnabled, finalizedBlockOffset, enforceRepeatableRead, deathDeclarationDelay, noNewFinalizedBlocksThreshold, finalizedBlockPollInterval)
+		finalityTagEnabled, finalizedBlockOffset, enforceRepeatableRead, deathDeclarationDelay, noNewFinalizedBlocksThreshold, finalizedBlockPollInterval, true)
 	require.NoError(t, err)
 
-	client := client.NewEvmClient(nodePool, chainCfg, nil, logger.Test(t), testutils.FixtureChainID, nodes, chaintype.ChainType(chainTypeStr))
-	require.NotNil(t, client)
+	evmClient := client.NewEvmClient(nodePool, chainCfg, nil, logger.Test(t), testutils.FixtureChainID, nodes, chaintype.ChainType(chainTypeStr))
+	require.NotNil(t, evmClient)
+
+	nodeConfigs[0].WSURL = nil
+	_, _, _, err = client.NewClientConfigs(selectionMode, leaseDuration, chainTypeStr, nodeConfigs,
+		pollFailureThreshold, pollInterval, syncThreshold, nodeIsSyncingEnabled, noNewHeadsThreshold, finalityDepth,
+		finalityTagEnabled, finalizedBlockOffset, enforceRepeatableRead, deathDeclarationDelay, noNewFinalizedBlocksThreshold, finalizedBlockPollInterval, true)
+	require.Equal(t, client.ErrLogBroadcasterEnabledWithoutWSURL, err)
+
+	_, _, _, err = client.NewClientConfigs(selectionMode, leaseDuration, chainTypeStr, nodeConfigs,
+		pollFailureThreshold, pollInterval, syncThreshold, nodeIsSyncingEnabled, noNewHeadsThreshold, finalityDepth,
+		finalityTagEnabled, finalizedBlockOffset, enforceRepeatableRead, deathDeclarationDelay, noNewFinalizedBlocksThreshold, finalizedBlockPollInterval, false)
+	require.Nil(t, err)
 }

core/chains/evm/client/rpc_client.go

@@ -75,6 +75,7 @@ var (
 			float64(8 * time.Second),
 		},
 	}, []string{"evmChainID", "nodeName", "rpcHost", "isSendOnly", "success", "rpcCallName"})
+	errSubscribeFilterLogsNotAllowedWithoutWS = errors.New("SubscribeFilterLogs is not allowed without ws url")
 )
 
 // RPCClient includes all the necessary generalized RPC methods along with any additional chain-specific methods.
@@ -196,29 +197,28 @@ func (r *rpcClient) Dial(callerCtx context.Context) error {
 	defer cancel()
 
 	promEVMPoolRPCNodeDials.WithLabelValues(r.chainID.String(), r.name).Inc()
-	lggr := r.rpcLog.With("wsuri", r.ws.uri.Redacted())
-	if r.http != nil {
-		lggr = lggr.With("httpuri", r.http.uri.Redacted())
-	}
-	lggr.Debugw("RPC dial: evmclient.Client#dial")
+	lggr := r.rpcLog
+	if r.ws.uri.String() != "" {
+		lggr = lggr.With("wsuri", r.ws.uri.Redacted())
+		wsrpc, err := rpc.DialWebsocket(ctx, r.ws.uri.String(), "")
+		if err != nil {
+			promEVMPoolRPCNodeDialsFailed.WithLabelValues(r.chainID.String(), r.name).Inc()
+			return r.wrapRPCClientError(pkgerrors.Wrapf(err, "error while dialing websocket: %v", r.ws.uri.Redacted()))
+		}
 
-	wsrpc, err := rpc.DialWebsocket(ctx, r.ws.uri.String(), "")
-	if err != nil {
-		promEVMPoolRPCNodeDialsFailed.WithLabelValues(r.chainID.String(), r.name).Inc()
-		return r.wrapRPCClientError(pkgerrors.Wrapf(err, "error while dialing websocket: %v", r.ws.uri.Redacted()))
+		r.ws.rpc = wsrpc
+		r.ws.geth = ethclient.NewClient(wsrpc)
 	}
 
-	r.ws.rpc = wsrpc
-	r.ws.geth = ethclient.NewClient(wsrpc)
-
 	if r.http != nil {
+		lggr = lggr.With("httpuri", r.http.uri.Redacted())
 		if err := r.DialHTTP(); err != nil {
 			return err
 		}
 	}
 
+	lggr.Debugw("RPC dial: evmclient.Client#dial")
 	promEVMPoolRPCNodeDialsSuccess.WithLabelValues(r.chainID.String(), r.name).Inc()
-
 	return nil
 }
 
@@ -1215,6 +1215,9 @@ func (r *rpcClient) ClientVersion(ctx context.Context) (version string, err erro
 }
 
 func (r *rpcClient) SubscribeFilterLogs(ctx context.Context, q ethereum.FilterQuery, ch chan<- types.Log) (_ ethereum.Subscription, err error) {
+	if r.ws.uri.String() == "" {
+		return nil, errSubscribeFilterLogsNotAllowedWithoutWS
+	}
 	ctx, cancel, chStopInFlight, ws, _ := r.acquireQueryCtx(ctx, r.rpcTimeout)
 	defer cancel()
 	lggr := r.newRqLggr().With("q", q)

core/chains/evm/client/rpc_client_test.go

@@ -209,6 +209,14 @@ func TestRPCClient_SubscribeFilterLogs(t *testing.T) {
 	lggr := logger.Test(t)
 	ctx, cancel := context.WithTimeout(tests.Context(t), tests.WaitTimeout(t))
 	defer cancel()
+	t.Run("Failed SubscribeFilterLogs when WSURL is empty", func(t *testing.T) {
+		observedLggr, _ := logger.TestObserved(t, zap.DebugLevel)
+		rpcClient := client.NewRPCClient(observedLggr, url.URL{}, nil, "rpc", 1, chainId, commonclient.Primary, 0, commonclient.QueryTimeout, commonclient.QueryTimeout, "")
+		require.Nil(t, rpcClient.Dial(ctx))
+
+		_, err := rpcClient.SubscribeFilterLogs(ctx, ethereum.FilterQuery{}, make(chan types.Log))
+		require.Error(t, err)
+	})
 	t.Run("Failed SubscribeFilterLogs logs and returns proper error", func(t *testing.T) {
 		server := testutils.NewWSServer(t, chainId, func(reqMethod string, reqParams gjson.Result) (resp testutils.JSONRPCResponse) {
 			return resp

core/chains/evm/config/toml/config.go

@@ -964,19 +964,8 @@ func (n *Node) ValidateConfig() (err error) {
 		err = multierr.Append(err, commonconfig.ErrEmpty{Name: "Name", Msg: "required for all nodes"})
 	}
 
-	var sendOnly bool
-	if n.SendOnly != nil {
-		sendOnly = *n.SendOnly
-	}
-	if n.WSURL == nil {
-		if !sendOnly {
-			err = multierr.Append(err, commonconfig.ErrMissing{Name: "WSURL", Msg: "required for primary nodes"})
-		}
-	} else if n.WSURL.IsZero() {
-		if !sendOnly {
-			err = multierr.Append(err, commonconfig.ErrEmpty{Name: "WSURL", Msg: "required for primary nodes"})
-		}
-	} else {
+	// WSURL can be optional when LogBroadcaster is disabled, if WSURL not provided, rpc will return error for SubscribeFilterLogs
+	if n.WSURL != nil && !n.WSURL.IsZero() {
 		switch n.WSURL.Scheme {
 		case "ws", "wss":
 		default:

core/config/docs/chains-evm.toml

@@ -462,7 +462,7 @@ ObservationGracePeriod = '1s' # Default
 [[EVM.Nodes]]
 # Name is a unique (per-chain) identifier for this node.
 Name = 'foo' # Example
-# WSURL is the WS(S) endpoint for this node. Required for primary nodes.
+# WSURL is the WS(S) endpoint for this node. Optional when `LogBroadcasterEnabled` is `false`
 WSURL = 'wss://web.socket/test' # Example
 # HTTPURL is the HTTP(S) endpoint for this node. Required for all nodes.
 HTTPURL = 'https://foo.web' # Example

core/services/chainlink/config_test.go

@@ -1328,9 +1328,7 @@ func TestConfig_Validate(t *testing.T) {
 				- PriceMax: invalid value (10 gwei): must be greater than or equal to PriceDefault
 				- BlockHistory.BlockHistorySize: invalid value (0): must be greater than or equal to 1 with BlockHistory Mode
 			- Nodes: 2 errors:
-				- 0: 2 errors:
-					- WSURL: missing: required for primary nodes
-					- HTTPURL: missing: required for all nodes
+				- 0.HTTPURL: missing: required for all nodes
 				- 1.HTTPURL: missing: required for all nodes
 		- 1: 10 errors:
 			- ChainType: invalid value (Foo): must not be set with this chain id
@@ -1352,17 +1350,15 @@ func TestConfig_Validate(t *testing.T) {
 			- FinalityDepth: invalid value (0): must be greater than or equal to 1
 			- MinIncomingConfirmations: invalid value (0): must be greater than or equal to 1
 		- 3.Nodes: 5 errors:
-				- 0: 3 errors:
+				- 0: 2 errors:
 					- Name: missing: required for all nodes
-					- WSURL: missing: required for primary nodes
 					- HTTPURL: empty: required for all nodes
 				- 1: 3 errors:
 					- Name: missing: required for all nodes
 					- WSURL: invalid value (http): must be ws or wss
 					- HTTPURL: missing: required for all nodes
-				- 2: 3 errors:
+				- 2: 2 errors:
 					- Name: empty: required for all nodes
-					- WSURL: missing: required for primary nodes
 					- HTTPURL: invalid value (ws): must be http or https
 				- 3.HTTPURL: missing: required for all nodes
 				- 4.HTTPURL: missing: required for all nodes

docs/CONFIG.md

@@ -9002,7 +9002,7 @@ Name is a unique (per-chain) identifier for this node.
 ```toml
 WSURL = 'wss://web.socket/test' # Example
 ```
-WSURL is the WS(S) endpoint for this node. Required for primary nodes.
+WSURL is the WS(S) endpoint for this node. Optional when `LogBroadcasterEnabled` is `false`
 
 ### HTTPURL
 ```toml