Implement new EVM Multinode

common/client/multi_node_test.go

@@ -17,44 +17,34 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
 
-	"github.com/smartcontractkit/chainlink/v2/common/config"
 	"github.com/smartcontractkit/chainlink/v2/common/types"
 )
 
-type multiNodeRPCClient RPC[types.ID, *big.Int, Hashable, Hashable, any, Hashable, any, any,
-	types.Receipt[Hashable, Hashable], Hashable, types.Head[Hashable], any]
+type multiNodeRPCClient RPCClient[types.ID, types.Head[Hashable]]
 
 type testMultiNode struct {
-	*multiNode[types.ID, *big.Int, Hashable, Hashable, any, Hashable, any, any,
-		types.Receipt[Hashable, Hashable], Hashable, types.Head[Hashable], multiNodeRPCClient, any]
+	*multiNode[types.ID, Hashable, types.Head[Hashable], multiNodeRPCClient]
 }
 
 type multiNodeOpts struct {
-	logger              logger.Logger
-	selectionMode       string
-	leaseDuration       time.Duration
-	noNewHeadsThreshold time.Duration
-	nodes               []Node[types.ID, types.Head[Hashable], multiNodeRPCClient]
-	sendonlys           []SendOnlyNode[types.ID, multiNodeRPCClient]
-	chainID             types.ID
-	chainType           config.ChainType
-	chainFamily         string
-	classifySendTxError func(tx any, err error) SendTxReturnCode
-	sendTxSoftTimeout   time.Duration
+	logger        logger.Logger
+	selectionMode string
+	leaseDuration time.Duration
+	nodes         []Node[types.ID, types.Head[Hashable], multiNodeRPCClient]
+	sendonlys     []Node[types.ID, types.Head[Hashable], multiNodeRPCClient]
+	chainID       types.ID
+	chainFamily   string
 }
 
 func newTestMultiNode(t *testing.T, opts multiNodeOpts) testMultiNode {
 	if opts.logger == nil {
 		opts.logger = logger.Test(t)
 	}
 
-	result := NewMultiNode[types.ID, *big.Int, Hashable, Hashable, any, Hashable, any, any,
-		types.Receipt[Hashable, Hashable], Hashable, types.Head[Hashable], multiNodeRPCClient, any](opts.logger,
-		opts.selectionMode, opts.leaseDuration, opts.noNewHeadsThreshold, opts.nodes, opts.sendonlys,
-		opts.chainID, opts.chainType, opts.chainFamily, opts.classifySendTxError, opts.sendTxSoftTimeout)
+	result := NewMultiNode[types.ID, Hashable, types.Head[Hashable], multiNodeRPCClient](
+		opts.logger, opts.selectionMode, opts.leaseDuration, opts.nodes, opts.sendonlys, opts.chainID, opts.chainFamily)
 	return testMultiNode{
-		result.(*multiNode[types.ID, *big.Int, Hashable, Hashable, any, Hashable, any, any,
-			types.Receipt[Hashable, Hashable], Hashable, types.Head[Hashable], multiNodeRPCClient, any]),
+		result.(*multiNode[types.ID, Hashable, types.Head[Hashable], multiNodeRPCClient]),
 	}
 }
 

common/client/node.go

@@ -57,32 +57,47 @@ type ChainConfig interface {
 	ChainType() commonconfig.ChainType
 }
 
+// ChainInfo - represents RPC’s view of the chain
+type ChainInfo struct {
+	// BlockNumber - block number of the most recent block observed by the Node
+	BlockNumber int64
+	// BlockDifficulty - difficulty of the most recent block observed by the Node
+	BlockDifficulty *big.Int
+	// LatestFinalizedBlock - block number of the most recently finalized block
+	LatestFinalizedBlock int64
+}
+
 //go:generate mockery --quiet --name Node --structname mockNode --filename "mock_node_test.go" --inpackage --case=underscore
 type Node[
 	CHAIN_ID types.ID,
 	HEAD Head,
-	RPC NodeClient[CHAIN_ID, HEAD],
+	RPC_CLIENT RPCClient[CHAIN_ID, HEAD],
 ] interface {
-	// State returns nodeState
+	// State returns health state of the underlying RPC
 	State() nodeState
-	// StateAndLatest returns nodeState with the latest received block number & total difficulty.
-	StateAndLatest() (nodeState, int64, *big.Int)
+	// StateAndLatest returns health state with the latest received block number & total difficulty.
+	StateAndLatest() (nodeState, ChainInfo)
 	// Name is a unique identifier for this node.
 	Name() string
+	// String - returns string representation of the node, useful for debugging (name + URLS used to connect to the RPC)
 	String() string
-	RPC() RPC
-	SubscribersCount() int32
-	UnsubscribeAllExceptAliveLoop()
+	// RPC - returns the underlying RPC_CLIENT
+	RPC() RPC_CLIENT
+	// UnsubscribeAll - terminates all client subscriptions. Called by MultiNode to trigger clients to resubscribe to
+	// new best RPC
+	UnsubscribeAll()
 	ConfiguredChainID() CHAIN_ID
+	// Order - returns priority order configured for the RPC
 	Order() int32
+	// Start - starts health checks
 	Start(context.Context) error
 	Close() error
 }
 
 type node[
 	CHAIN_ID types.ID,
 	HEAD Head,
-	RPC NodeClient[CHAIN_ID, HEAD],
+	RPC_CLIENT RPCClient[CHAIN_ID, HEAD],
 ] struct {
 	services.StateMachine
 	lfcLog      logger.Logger
@@ -97,7 +112,7 @@ type node[
 	ws   url.URL
 	http *url.URL
 
-	rpc RPC
+	rpc RPC_CLIENT
 
 	stateMu sync.RWMutex // protects state* fields
 	state   nodeState
@@ -123,7 +138,7 @@ type node[
 func NewNode[
 	CHAIN_ID types.ID,
 	HEAD Head,
-	RPC NodeClient[CHAIN_ID, HEAD],
+	RPC_CLIENT RPCClient[CHAIN_ID, HEAD],
 ](
 	nodeCfg NodeConfig,
 	chainCfg ChainConfig,
@@ -134,10 +149,10 @@ func NewNode[
 	id int32,
 	chainID CHAIN_ID,
 	nodeOrder int32,
-	rpc RPC,
+	rpc RPC_CLIENT,
 	chainFamily string,
-) Node[CHAIN_ID, HEAD, RPC] {
-	n := new(node[CHAIN_ID, HEAD, RPC])
+) Node[CHAIN_ID, HEAD, RPC_CLIENT] {
+	n := new(node[CHAIN_ID, HEAD, RPC_CLIENT])
 	n.name = name
 	n.id = id
 	n.chainID = chainID
@@ -164,39 +179,35 @@ func NewNode[
 	return n
 }
 
-func (n *node[CHAIN_ID, HEAD, RPC]) String() string {
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) String() string {
 	s := fmt.Sprintf("(%s)%s:%s", Primary.String(), n.name, n.ws.String())
 	if n.http != nil {
 		s = s + fmt.Sprintf(":%s", n.http.String())
 	}
 	return s
 }
 
-func (n *node[CHAIN_ID, HEAD, RPC]) ConfiguredChainID() (chainID CHAIN_ID) {
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) ConfiguredChainID() (chainID CHAIN_ID) {
 	return n.chainID
 }
 
-func (n *node[CHAIN_ID, HEAD, RPC]) Name() string {
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) Name() string {
 	return n.name
 }
 
-func (n *node[CHAIN_ID, HEAD, RPC]) RPC() RPC {
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) RPC() RPC_CLIENT {
 	return n.rpc
 }
 
-func (n *node[CHAIN_ID, HEAD, RPC]) SubscribersCount() int32 {
-	return n.rpc.SubscribersCount()
-}
-
-func (n *node[CHAIN_ID, HEAD, RPC]) UnsubscribeAllExceptAliveLoop() {
-	n.rpc.UnsubscribeAllExceptAliveLoop()
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) UnsubscribeAll() {
+	n.rpc.UnsubscribeAllExcept()
 }
 
-func (n *node[CHAIN_ID, HEAD, RPC]) Close() error {
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) Close() error {
 	return n.StopOnce(n.name, n.close)
 }
 
-func (n *node[CHAIN_ID, HEAD, RPC]) close() error {
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) close() error {
 	defer func() {
 		n.wg.Wait()
 		n.rpc.Close()
@@ -214,7 +225,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) close() error {
 // Should only be called once in a node's lifecycle
 // Return value is necessary to conform to interface but this will never
 // actually return an error.
-func (n *node[CHAIN_ID, HEAD, RPC]) Start(startCtx context.Context) error {
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) Start(startCtx context.Context) error {
 	return n.StartOnce(n.name, func() error {
 		n.start(startCtx)
 		return nil
@@ -226,7 +237,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) Start(startCtx context.Context) error {
 // Not thread-safe.
 // Node lifecycle is synchronous: only one goroutine should be running at a
 // time.
-func (n *node[CHAIN_ID, HEAD, RPC]) start(startCtx context.Context) {
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) start(startCtx context.Context) {
 	if n.state != nodeStateUndialed {
 		panic(fmt.Sprintf("cannot dial node with state %v", n.state))
 	}
@@ -245,7 +256,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) start(startCtx context.Context) {
 // verifyChainID checks that connection to the node matches the given chain ID
 // Not thread-safe
 // Pure verifyChainID: does not mutate node "state" field.
-func (n *node[CHAIN_ID, HEAD, RPC]) verifyChainID(callerCtx context.Context, lggr logger.Logger) nodeState {
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) verifyChainID(callerCtx context.Context, lggr logger.Logger) nodeState {
 	promPoolRPCNodeVerifies.WithLabelValues(n.chainFamily, n.chainID.String(), n.name).Inc()
 	promFailed := func() {
 		promPoolRPCNodeVerifiesFailed.WithLabelValues(n.chainFamily, n.chainID.String(), n.name).Inc()
@@ -288,7 +299,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) verifyChainID(callerCtx context.Context, lgg
 
 // createVerifiedConn - establishes new connection with the RPC and verifies that it's valid: chainID matches, and it's not syncing.
 // Returns desired state if one of the verifications fails. Otherwise, returns nodeStateAlive.
-func (n *node[CHAIN_ID, HEAD, RPC]) createVerifiedConn(ctx context.Context, lggr logger.Logger) nodeState {
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) createVerifiedConn(ctx context.Context, lggr logger.Logger) nodeState {
 	if err := n.rpc.Dial(ctx); err != nil {
 		n.lfcLog.Errorw("Dial failed: Node is unreachable", "err", err, "nodeState", n.State())
 		return nodeStateUnreachable
@@ -299,7 +310,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) createVerifiedConn(ctx context.Context, lggr
 
 // verifyConn - verifies that current connection is valid: chainID matches, and it's not syncing.
 // Returns desired state if one of the verifications fails. Otherwise, returns nodeStateAlive.
-func (n *node[CHAIN_ID, HEAD, RPC]) verifyConn(ctx context.Context, lggr logger.Logger) nodeState {
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) verifyConn(ctx context.Context, lggr logger.Logger) nodeState {
 	state := n.verifyChainID(ctx, lggr)
 	if state != nodeStateAlive {
 		return state
@@ -321,13 +332,6 @@ func (n *node[CHAIN_ID, HEAD, RPC]) verifyConn(ctx context.Context, lggr logger.
 	return nodeStateAlive
 }
 
-// disconnectAll disconnects all clients connected to the node
-// WARNING: NOT THREAD-SAFE
-// This must be called from within the n.stateMu lock
-func (n *node[CHAIN_ID, HEAD, RPC]) disconnectAll() {
-	n.rpc.DisconnectAll()
-}
-
-func (n *node[CHAIN_ID, HEAD, RPC]) Order() int32 {
+func (n *node[CHAIN_ID, HEAD, RPC_CLIENT]) Order() int32 {
 	return n.order
 }

common/client/node_fsm.go

@@ -120,10 +120,13 @@ func (n *node[CHAIN_ID, HEAD, RPC]) State() nodeState {
 	return n.state
 }
 
-func (n *node[CHAIN_ID, HEAD, RPC]) StateAndLatest() (nodeState, int64, *big.Int) {
+func (n *node[CHAIN_ID, HEAD, RPC]) StateAndLatest() (nodeState, ChainInfo) {
 	n.stateMu.RLock()
 	defer n.stateMu.RUnlock()
-	return n.state, n.stateLatestBlockNumber, n.stateLatestTotalDifficulty
+	return n.state, ChainInfo{
+		BlockNumber:          n.stateLatestBlockNumber,
+		BlockDifficulty:      n.stateLatestTotalDifficulty,
+		LatestFinalizedBlock: n.stateLatestFinalizedBlockNumber}
 }
 
 // setState is only used by internal state management methods.
@@ -209,7 +212,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) transitionToOutOfSync(fn func()) {
 	}
 	switch n.state {
 	case nodeStateAlive:
-		n.disconnectAll()
+		n.UnsubscribeAll()
 		n.state = nodeStateOutOfSync
 	default:
 		panic(transitionFail(n.state, nodeStateOutOfSync))
@@ -234,7 +237,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) transitionToUnreachable(fn func()) {
 	}
 	switch n.state {
 	case nodeStateUndialed, nodeStateDialed, nodeStateAlive, nodeStateOutOfSync, nodeStateInvalidChainID, nodeStateSyncing:
-		n.disconnectAll()
+		n.UnsubscribeAll()
 		n.state = nodeStateUnreachable
 	default:
 		panic(transitionFail(n.state, nodeStateUnreachable))
@@ -277,7 +280,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) transitionToInvalidChainID(fn func()) {
 	}
 	switch n.state {
 	case nodeStateDialed, nodeStateOutOfSync, nodeStateSyncing:
-		n.disconnectAll()
+		n.UnsubscribeAll()
 		n.state = nodeStateInvalidChainID
 	default:
 		panic(transitionFail(n.state, nodeStateInvalidChainID))
@@ -302,7 +305,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) transitionToSyncing(fn func()) {
 	}
 	switch n.state {
 	case nodeStateDialed, nodeStateOutOfSync, nodeStateInvalidChainID:
-		n.disconnectAll()
+		n.UnsubscribeAll()
 		n.state = nodeStateSyncing
 	default:
 		panic(transitionFail(n.state, nodeStateSyncing))