fixing multinode state transition logic by register polling subscription in rpc client

.changeset/moody-rules-agree.md

@@ -0,0 +1,7 @@
+---
+"chainlink": patch
+---
+
+- register polling subscription logic in rpc client so when node unhealthy new susbcription will be used
+- 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.
+#fixed

core/chains/evm/client/rpc_client.go

@@ -496,11 +496,30 @@ func (r *rpcClient) SubscribeNewHead(ctx context.Context, channel chan<- *evmtyp
 	if r.newHeadsPollInterval > 0 {
 		interval := r.newHeadsPollInterval
 		timeout := interval
-		poller, _ := commonclient.NewPoller[*evmtypes.Head](interval, r.latestBlock, timeout, r.rpcLog)
+		poller, pollerCh := commonclient.NewPoller[*evmtypes.Head](interval, r.latestBlock, timeout, r.rpcLog)
 		if err = poller.Start(ctx); err != nil {
 			return nil, err
 		}
 
+		// NOTE this is a temporary special treatment for SubscribeNewHead before we refactor head tracker to use SubscribeToHeads
+		// as we need to forward new head from the poller channel to the channel passed from caller.
+		go func() {
+			for head := range pollerCh {
+				select {
+				case channel <- head:
+					// forwarding new head to the channel passed from caller
+				case <-poller.Err():
+					// return as poller returns error
+					return
+				}
+			}
+		}()
+
+		err = r.registerSub(&poller, chStopInFlight)
+		if err != nil {
+			return nil, err
+		}
+
 		lggr.Debugf("Polling new heads over http")
 		return &poller, nil
 	}
@@ -547,6 +566,11 @@ func (r *rpcClient) SubscribeToHeads(ctx context.Context) (ch <-chan *evmtypes.H
 			return nil, nil, err
 		}
 
+		err = r.registerSub(&poller, chStopInFlight)
+		if err != nil {
+			return nil, nil, err
+		}
+
 		lggr.Debugf("Polling new heads over http")
 		return channel, &poller, nil
 	}
@@ -579,6 +603,8 @@ func (r *rpcClient) SubscribeToHeads(ctx context.Context) (ch <-chan *evmtypes.H
 }
 
 func (r *rpcClient) SubscribeToFinalizedHeads(ctx context.Context) (<-chan *evmtypes.Head, commontypes.Subscription, error) {
+	ctx, cancel, chStopInFlight, _, _ := r.acquireQueryCtx(ctx, r.rpcTimeout)
+	defer cancel()
 	interval := r.finalizedBlockPollInterval
 	if interval == 0 {
 		return nil, nil, errors.New("FinalizedBlockPollInterval is 0")
@@ -588,6 +614,12 @@ func (r *rpcClient) SubscribeToFinalizedHeads(ctx context.Context) (<-chan *evmt
 	if err := poller.Start(ctx); err != nil {
 		return nil, nil, err
 	}
+
+	err := r.registerSub(&poller, chStopInFlight)
+	if err != nil {
+		return nil, nil, err
+	}
+
 	return channel, &poller, nil
 }
 

core/chains/evm/client/rpc_client_test.go

@@ -184,6 +184,49 @@ func TestRPCClient_SubscribeNewHead(t *testing.T) {
 		require.ErrorContains(t, err, "RPCClient returned error (rpc)")
 		tests.AssertLogEventually(t, observed, "evmclient.Client#EthSubscribe RPC call failure")
 	})
+	t.Run("Closed rpc client should remove existing SubscribeNewHead subscription", func(t *testing.T) {
+		server := testutils.NewWSServer(t, chainId, serverCallBack)
+		wsURL := server.WSURL()
+
+		rpc := client.NewRPCClient(lggr, *wsURL, nil, "rpc", 1, chainId, commonclient.Primary, 0, 0, commonclient.QueryTimeout, commonclient.QueryTimeout, "")
+		require.NoError(t, rpc.Dial(ctx))
+
+		ch := make(chan *evmtypes.Head)
+		_, err := rpc.SubscribeNewHead(tests.Context(t), ch)
+		require.NoError(t, err)
+		require.Equal(t, int32(1), rpc.SubscribersCount())
+		rpc.DisconnectAll()
+		require.NoError(t, rpc.Dial(ctx))
+		require.Equal(t, int32(0), rpc.SubscribersCount())
+	})
+	t.Run("Closed rpc client should remove existing SubscribeToHeads subscription", func(t *testing.T) {
+		server := testutils.NewWSServer(t, chainId, serverCallBack)
+		wsURL := server.WSURL()
+
+		rpc := client.NewRPCClient(lggr, *wsURL, nil, "rpc", 1, chainId, commonclient.Primary, 0, 0, commonclient.QueryTimeout, commonclient.QueryTimeout, "")
+		require.NoError(t, rpc.Dial(ctx))
+
+		_, _, err := rpc.SubscribeToHeads(tests.Context(t))
+		require.NoError(t, err)
+		require.Equal(t, int32(1), rpc.SubscribersCount())
+		rpc.DisconnectAll()
+		require.NoError(t, rpc.Dial(ctx))
+		require.Equal(t, int32(0), rpc.SubscribersCount())
+	})
+	//t.Run("Closed rpc client should remove existing SubscribeToFinalizedHeads subscription", func(t *testing.T) {
+	//	server := testutils.NewWSServer(t, chainId, serverCallBack)
+	//	wsURL := server.WSURL()
+	//
+	//	rpc := client.NewRPCClient(lggr, *wsURL, &url.URL{}, "rpc", 1, chainId, commonclient.Primary, 1, 0, commonclient.QueryTimeout, commonclient.QueryTimeout, "")
+	//	require.NoError(t, rpc.Dial(ctx))
+	//
+	//	_, _, err := rpc.SubscribeToFinalizedHeads(tests.Context(t))
+	//	require.NoError(t, err)
+	//	require.Equal(t, int32(1), rpc.SubscribersCount())
+	//	rpc.DisconnectAll()
+	//	require.NoError(t, rpc.Dial(ctx))
+	//	require.Equal(t, int32(0), rpc.SubscribersCount())
+	//})
 	t.Run("Subscription error is properly wrapper", func(t *testing.T) {
 		server := testutils.NewWSServer(t, chainId, serverCallBack)
 		wsURL := server.WSURL()