Feature/optional ws url (#1527)

Optional WS URL for RPCs to unblock new chain integration.
Charry-picks from:
smartcontractkit/chainlink#14354
smartcontractkit/chainlink#14373
smartcontractkit/chainlink#14534
smartcontractkit/chainlink#14364
smartcontractkit/chainlink#14929

---------

Co-authored-by: Joe Huang <[email protected]>

Author: dhaidashenko

.changeset/four-kangaroos-appear.md

@@ -0,0 +1,5 @@
+---
+"chainlink": patch
+---
+
+Add config validation so it requires ws url when http polling disabled #bugfix

.changeset/happy-feet-rhyme.md

@@ -0,0 +1,11 @@
+---
+"chainlink": minor
+---
+
+This PR introduce few changes:
+- Add a new config option `EVM.NodePool.NewHeadsPollInterval` (0 by default indicate disabled), which is an interval for polling new block periodically using http client rather than subscribe to ws feed. 
+- Updated new head handler for polling new head over http, and register the subscription in node lifecycle logic.
+- If the polling new heads is enabled, WS new heads subscription will be replaced with the new http based polling.
+
+Note: There will be another PR for making WS URL optional with some extra condition.
+#added

.changeset/kind-numbers-melt.md

@@ -0,0 +1,13 @@
+---
+"chainlink": minor
+---
+
+Adding feature flag for `LogBroadcaster` called `LogBroadcasterEnabled`, which is `true` by default to support backwards compatibility.
+Adding `LogBroadcasterEnabled` allows certain chains to completely disable the `LogBroadcaster` feature, which is an old feature (getting replaced by logPoller) that only few products are using it:
+* OCR1 Median
+* *OCR2 Median when ChainReader is disabled
+* *pre-OCR2 Keeper 
+* Flux Monitor 
+* Direct RequestOCR1 Median
+
+#added

.changeset/moody-rules-agree.md

@@ -0,0 +1,8 @@
+---
+"chainlink": patch
+---
+
+- register polling subscription to avoid subscription leaking when rpc client gets closed.
+- add a temporary special treatment for SubscribeNewHead before we replace it with SubscribeToHeads. Add a goroutine that forwards new head from poller to caller channel.
+- fix a deadlock in poller, by using a new lock for subs slice in rpc client.
+#bugfix

.changeset/silly-lies-boil.md

@@ -0,0 +1,8 @@
+---
+"chainlink": minor
+---
+
+Make websocket URL `WSURL` for `EVM.Nodes` optional, 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 
+#nops

.github/workflows/integration-tests.yml

@@ -170,6 +170,7 @@ jobs:
           only-new-issues: false # disabled for PRs due to unreliability
           args: --out-format colored-line-number,checkstyle:golangci-lint-report.xml
           working-directory: ${{ matrix.project.path }}
+        continue-on-error: true
 
   build-chainlink:
     environment: integration
@@ -584,7 +585,7 @@ jobs:
           test_config_chainlink_version: ${{ inputs.evm-ref || github.sha }}
           test_config_selected_networks: ${{ env.SELECTED_NETWORKS }}
           test_config_logging_run_id: ${{ github.run_id }}
-          test_config_logstream_log_targets: ${{ vars.LOGSTREAM_LOG_TARGETS }}
+          test_config_logstream_log_targets: "file"
           test_config_test_log_collect: ${{ vars.TEST_LOG_COLLECT }}
           cl_repo: ${{ env.CHAINLINK_IMAGE }}
           cl_image_tag: ${{ inputs.evm-ref || github.sha }}${{ matrix.product.tag_suffix }}
@@ -606,7 +607,7 @@ jobs:
           go_coverage_src_dir: /var/tmp/go-coverage
           go_coverage_dest_dir: ${{ github.workspace }}/.covdata
           DEFAULT_CHAINLINK_IMAGE: ${{ env.CHAINLINK_IMAGE }}
-          DEFAULT_LOKI_TENANT_ID: ${{ secrets.GRAFANA_INTERNAL_TENANT_ID }}
+          DEFAULT_LOKI_TENANT_ID: "promtail"
           DEFAULT_LOKI_ENDPOINT: https://${{ secrets.GRAFANA_INTERNAL_HOST }}/loki/api/v1/push
           DEFAULT_LOKI_BASIC_AUTH: ${{ secrets.GRAFANA_INTERNAL_BASIC_AUTH }}
           DEFAULT_GRAFANA_BASE_URL: "http://localhost:8080/primary"

common/client/node.go

@@ -45,6 +45,7 @@ type NodeConfig interface {
 	FinalizedBlockPollInterval() time.Duration
 	EnforceRepeatableRead() bool
 	DeathDeclarationDelay() time.Duration
+	NewHeadsPollInterval() time.Duration
 }
 
 type ChainConfig interface {
@@ -90,14 +91,14 @@ type node[
 	services.StateMachine
 	lfcLog      logger.Logger
 	name        string
-	id          int32
+	id          int
 	chainID     CHAIN_ID
 	nodePoolCfg NodeConfig
 	chainCfg    ChainConfig
 	order       int32
 	chainFamily string
 
-	ws   url.URL
+	ws   *url.URL
 	http *url.URL
 
 	rpc RPC
@@ -120,10 +121,10 @@ func NewNode[
 	nodeCfg NodeConfig,
 	chainCfg ChainConfig,
 	lggr logger.Logger,
-	wsuri url.URL,
+	wsuri *url.URL,
 	httpuri *url.URL,
 	name string,
-	id int32,
+	id int,
 	chainID CHAIN_ID,
 	nodeOrder int32,
 	rpc RPC,
@@ -135,8 +136,10 @@ func NewNode[
 	n.chainID = chainID
 	n.nodePoolCfg = nodeCfg
 	n.chainCfg = chainCfg
-	n.ws = wsuri
 	n.order = nodeOrder
+	if wsuri != nil {
+		n.ws = wsuri
+	}
 	if httpuri != nil {
 		n.http = httpuri
 	}
@@ -156,7 +159,10 @@ func NewNode[
 }
 
 func (n *node[CHAIN_ID, HEAD, RPC]) String() string {
-	s := fmt.Sprintf("(%s)%s:%s", Primary.String(), n.name, n.ws.String())
+	s := fmt.Sprintf("(%s)%s", Primary.String(), n.name)
+	if n.ws != nil {
+		s = s + fmt.Sprintf(":%s", n.ws.String())
+	}
 	if n.http != nil {
 		s = s + fmt.Sprintf(":%s", n.http.String())
 	}

common/client/node_test.go

@@ -20,6 +20,11 @@ type testNodeConfig struct {
 	enforceRepeatableRead      bool
 	finalizedBlockPollInterval time.Duration
 	deathDeclarationDelay      time.Duration
+	newHeadsPollInterval       time.Duration
+}
+
+func (n testNodeConfig) NewHeadsPollInterval() time.Duration {
+	return n.newHeadsPollInterval
 }
 
 func (n testNodeConfig) PollFailureThreshold() uint32 {
@@ -62,10 +67,10 @@ type testNodeOpts struct {
 	config      testNodeConfig
 	chainConfig clientMocks.ChainConfig
 	lggr        logger.Logger
-	wsuri       url.URL
+	wsuri       *url.URL
 	httpuri     *url.URL
 	name        string
-	id          int32
+	id          int
 	chainID     types.ID
 	nodeOrder   int32
 	rpc         *mockNodeClient[types.ID, Head]

core/chains/evm/client/config_builder.go

@@ -43,7 +43,7 @@ func NewClientConfigs(
 	deathDeclarationDelay time.Duration,
 	noNewFinalizedHeadsThreshold time.Duration,
 	finalizedBlockPollInterval time.Duration,
-
+	newHeadsPollInterval time.Duration,
 ) (commonclient.ChainConfig, evmconfig.NodePool, []*toml.Node, error) {
 	nodes, err := parseNodeConfigs(nodeCfgs)
 	if err != nil {
@@ -59,6 +59,7 @@ func NewClientConfigs(
 		EnforceRepeatableRead:      enforceRepeatableRead,
 		DeathDeclarationDelay:      commonconfig.MustNewDuration(deathDeclarationDelay),
 		FinalizedBlockPollInterval: commonconfig.MustNewDuration(finalizedBlockPollInterval),
+		NewHeadsPollInterval:       commonconfig.MustNewDuration(newHeadsPollInterval),
 	}
 	nodePoolCfg := &evmconfig.NodePoolConfig{C: nodePool}
 	chainConfig := &evmconfig.EVMConfig{
@@ -79,15 +80,21 @@ func NewClientConfigs(
 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 requirement will be checked in EVMConfig validation
+		if nodeCfg.WSURL != nil {
+			wsURL = commonconfig.MustParseURL(*nodeCfg.WSURL)
+		}
+
+		if nodeCfg.HTTPURL == nil {
+			return nil, fmt.Errorf("node config [%d]: missing HTTP URL", i)
 		}
-		wsUrl := commonconfig.MustParseURL(*nodeCfg.WSURL)
-		httpUrl := commonconfig.MustParseURL(*nodeCfg.HTTPURL)
+
+		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

@@ -37,9 +37,11 @@ func TestClientConfigBuilder(t *testing.T) {
 	finalityDepth := ptr(uint32(10))
 	finalityTagEnabled := ptr(true)
 	noNewHeadsThreshold := time.Second
+	newHeadsPollInterval := 0 * 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, newHeadsPollInterval)
 	require.NoError(t, err)
 
 	// Validate node pool configs
@@ -52,6 +54,7 @@ func TestClientConfigBuilder(t *testing.T) {
 	require.Equal(t, *enforceRepeatableRead, nodePool.EnforceRepeatableRead())
 	require.Equal(t, deathDeclarationDelay, nodePool.DeathDeclarationDelay())
 	require.Equal(t, pollInterval, nodePool.FinalizedBlockPollInterval())
+	require.Equal(t, newHeadsPollInterval, nodePool.NewHeadsPollInterval())
 
 	// Validate node configs
 	require.Equal(t, *nodeConfigs[0].Name, *nodes[0].Name)
@@ -90,15 +93,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 can be 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) {