MultiNode Soak Testing (#894)

* Add defaults

* Add latest block methods

* Address comments

* lint

* Fix lint overflow issues

* Update transaction_sender.go

* Fix lint

* Validate node config

* Update toml.go

* Add SendOnly nodes

* Use pointers on config

* Add test outlines

* Use test context

* Use configured selection mode

* Set defaults

* lint

* Add nil check

* Add client test

* Add subscription test

* tidy

* Fix imports

* Update chain_test.go

* Update multinode.go

* Add comments

* Update multinode.go

* Wrap multinode config

* Fix imports

* Update .golangci.yml

* Use MultiNode

* Add multinode to txm

* Use MultiNode

* Update chain.go

* Update balance_test.go

* Add retries

* Fix head

* Update client.go

* lint

* lint

* Use MultiNode TxSender

* Update txm_internal_test.go

* Address comments

* Remove total difficulty

* Register polling subs

* Extract MultiNodeClient

* Remove caching changes

* Undo cache changes

* Fix tests

* Update chain.go

* Fix variables

* Move classify errors

* Fix imports

* lint

* Update txm_internal_test.go

* Update txm_internal_test.go

* lint

* Fix error classification

* Update txm_internal_test.go

* Update multinode_client.go

* lint

* Update classify_errors.go

* Update classify_errors.go

* Add tests

* Add test coverage

* lint

* Add dial comment

* CTF bump for image build

* Update pkg/solana/client/multinode_client.go

Co-authored-by: Dmytro Haidashenko <[email protected]>

* Update txm.go

* Create loader

* Update transaction_sender.go

* Fix tests

* Update txm_internal_test.go

* lint

* Update txm.go

* Add ctx

* Fix imports

* Add SendTxResult to TxSender

* Update chain_test.go

* Enable MultiNode

* Move error classification

* Add MultiNode config

* Use loader

* Update multinode.go

* Update multinode.go

* Use loader in txm tests

* lint

* Update testconfig.go

* Update loader

* Use single RPC

* Fix tests

* lint

* Use default thresholds

* Address comments

* Update classify_errors.go

* Update testconfig.go

* Update errors

* lint

* Fix SendTransaction

* Update chain.go

* Update sendTx

* Fix ctx issues

* Enable multiple RPCs in soak tests

* Update defaults for testing

* Add health check tags

* Increase sync threshold

* Validate heads

* Use latestChainInfo

* Fix AliveLoop bug

* Update configurations

* Update transaction_sender.go

* Get chain info

* Update ctx

* Update transaction_sender.go

* Update transaction_sender.go

* Increase tx timeout

* Update transaction_sender.go

* Update ctx

* Add timer

* Update transaction_sender.go

* Update transaction_sender.go

* Update testconfig.go

* Fix ctx

* Remove debug logging

* Update run_soak_test.sh

* lint

* Add debugging logs

* Fix ctx cancel

* Fix ctx cancel

* Fix DoAll ctx

* Remove debugging logs

* Remove logs

* defer reportWg

* Add result ctx logging

* log on close

* Update transaction_sender.go

* add cancel func

* Update transaction_sender.go

* Update transaction_sender.go

* Add ctx to reportSendTxAnomalies

* Update comments

* Fix comments

* Address comments

* lint

* lint

* Pass context

* Update node_lifecycle.go

* Use get reader function

* Make rpcurls plural

* Fix reader getters

* lint

* fix imports

* Update transaction_sender.go

* Remove TxError

* Rename getReader

* lint

* Update chain_test.go

* Update transmissions_cache.go

* Update run_soak_test.sh

* Fix deprecated method

* Clean up getReader

* Use AccountReader

---------

Co-authored-by: Damjan Smickovski <[email protected]>
Co-authored-by: Dmytro Haidashenko <[email protected]>

integration-tests/common/common.go

@@ -51,7 +51,7 @@ type TestEnvDetails struct {
 type ChainDetails struct {
 	ChainName             string
 	ChainID               string
-	RPCUrl                string
+	RPCUrls               []string
 	RPCURLExternal        string
 	WSURLExternal         string
 	ProgramAddresses      *chainConfig.ProgramAddresses
@@ -116,9 +116,9 @@ func New(testConfig *tc.TestConfig) *Common {
 		config = chainConfig.DevnetConfig()
 		privateKeyString = *testConfig.Common.PrivateKey
 
-		if *testConfig.Common.RPCURL != "" {
-			config.RPCUrl = *testConfig.Common.RPCURL
-			config.WSUrl = *testConfig.Common.WsURL
+		if len(*testConfig.Common.RPCURLs) > 0 {
+			config.RPCUrls = *testConfig.Common.RPCURLs
+			config.WSUrls = *testConfig.Common.WsURLs
 			config.ProgramAddresses = &chainConfig.ProgramAddresses{
 				OCR2:             *testConfig.SolanaConfig.OCR2ProgramID,
 				AccessController: *testConfig.SolanaConfig.AccessControllerProgramID,
@@ -130,7 +130,7 @@ func New(testConfig *tc.TestConfig) *Common {
 	c = &Common{
 		ChainDetails: &ChainDetails{
 			ChainID:          config.ChainID,
-			RPCUrl:           config.RPCUrl,
+			RPCUrls:          config.RPCUrls,
 			ChainName:        config.ChainName,
 			ProgramAddresses: config.ProgramAddresses,
 		},
@@ -146,7 +146,7 @@ func New(testConfig *tc.TestConfig) *Common {
 	}
 	// provide getters for TestConfig (pointers to chain details)
 	c.TestConfig.GetChainID = func() string { return c.ChainDetails.ChainID }
-	c.TestConfig.GetURL = func() string { return c.ChainDetails.RPCUrl }
+	c.TestConfig.GetURL = func() []string { return c.ChainDetails.RPCUrls }
 
 	return c
 }
@@ -298,7 +298,7 @@ func (c *Common) CreateJobsForContract(contractNodeInfo *ContractNodeInfo) error
 		bootstrapNodeInternalIP = contractNodeInfo.BootstrapNode.InternalIP()
 	}
 	relayConfig := job.JSONConfig{
-		"nodeEndpointHTTP": c.ChainDetails.RPCUrl,
+		"nodeEndpointHTTP": c.ChainDetails.RPCUrls,
 		"ocr2ProgramID":    contractNodeInfo.OCR2.ProgramAddress(),
 		"transmissionsID":  contractNodeInfo.Store.TransmissionsAddress(),
 		"storeProgramID":   contractNodeInfo.Store.ProgramAddress(),

integration-tests/common/test_common.go

@@ -118,9 +118,9 @@ func (m *OCRv2TestState) DeployCluster(contractsDir string) {
 		m.Common.ChainDetails.WSURLExternal = m.Common.Env.URLs["sol"][1]
 
 		if *m.Config.TestConfig.Common.Network == "devnet" {
-			m.Common.ChainDetails.RPCUrl = *m.Config.TestConfig.Common.RPCURL
-			m.Common.ChainDetails.RPCURLExternal = *m.Config.TestConfig.Common.RPCURL
-			m.Common.ChainDetails.WSURLExternal = *m.Config.TestConfig.Common.WsURL
+			m.Common.ChainDetails.RPCUrls = *m.Config.TestConfig.Common.RPCURLs
+			m.Common.ChainDetails.RPCURLExternal = (*m.Config.TestConfig.Common.RPCURLs)[0]
+			m.Common.ChainDetails.WSURLExternal = (*m.Config.TestConfig.Common.WsURLs)[0]
 		}
 
 		m.Common.ChainDetails.MockserverURLInternal = m.Common.Env.URLs["qa_mock_adapter_internal"][0]
@@ -133,14 +133,14 @@ func (m *OCRv2TestState) DeployCluster(contractsDir string) {
 		require.NoError(m.Config.T, err)
 
 		// Setting the External RPC url for Gauntlet
-		m.Common.ChainDetails.RPCUrl = sol.InternalHTTPURL
+		m.Common.ChainDetails.RPCUrls = []string{sol.InternalHTTPURL}
 		m.Common.ChainDetails.RPCURLExternal = sol.ExternalHTTPURL
 		m.Common.ChainDetails.WSURLExternal = sol.ExternalWsURL
 
 		if *m.Config.TestConfig.Common.Network == "devnet" {
-			m.Common.ChainDetails.RPCUrl = *m.Config.TestConfig.Common.RPCURL
-			m.Common.ChainDetails.RPCURLExternal = *m.Config.TestConfig.Common.RPCURL
-			m.Common.ChainDetails.WSURLExternal = *m.Config.TestConfig.Common.WsURL
+			m.Common.ChainDetails.RPCUrls = *m.Config.TestConfig.Common.RPCURLs
+			m.Common.ChainDetails.RPCURLExternal = (*m.Config.TestConfig.Common.RPCURLs)[0]
+			m.Common.ChainDetails.WSURLExternal = (*m.Config.TestConfig.Common.WsURLs)[0]
 		}
 
 		b, err := test_env.NewCLTestEnvBuilder().
@@ -273,7 +273,7 @@ func (m *OCRv2TestState) CreateJobs() {
 	require.NoError(m.Config.T, err, "Error connecting to websocket client")
 
 	relayConfig := job.JSONConfig{
-		"nodeEndpointHTTP": m.Common.ChainDetails.RPCUrl,
+		"nodeEndpointHTTP": m.Common.ChainDetails.RPCUrls,
 		"ocr2ProgramID":    m.Common.ChainDetails.ProgramAddresses.OCR2,
 		"transmissionsID":  m.Gauntlet.FeedAddress,
 		"storeProgramID":   m.Common.ChainDetails.ProgramAddresses.Store,

integration-tests/config/config.go

@@ -3,8 +3,8 @@ package config
 type Config struct {
 	ChainName        string
 	ChainID          string
-	RPCUrl           string
-	WSUrl            string
+	RPCUrls          []string
+	WSUrls           []string
 	ProgramAddresses *ProgramAddresses
 	PrivateKey       string
 }
@@ -20,8 +20,8 @@ func DevnetConfig() *Config {
 		ChainName: "solana",
 		ChainID:   "devnet",
 		// Will be overridden if set in toml
-		RPCUrl: "https://api.devnet.solana.com",
-		WSUrl:  "wss://api.devnet.solana.com/",
+		RPCUrls: []string{"https://api.devnet.solana.com"},
+		WSUrls:  []string{"wss://api.devnet.solana.com/"},
 	}
 }
 
@@ -30,8 +30,8 @@ func LocalNetConfig() *Config {
 		ChainName: "solana",
 		ChainID:   "localnet",
 		// Will be overridden if set in toml
-		RPCUrl: "http://sol:8899",
-		WSUrl:  "ws://sol:8900",
+		RPCUrls: []string{"http://sol:8899"},
+		WSUrls:  []string{"ws://sol:8900"},
 		ProgramAddresses: &ProgramAddresses{
 			OCR2:             "E3j24rx12SyVsG6quKuZPbQqZPkhAUCh8Uek4XrKYD2x",
 			AccessController: "2ckhep7Mvy1dExenBqpcdevhRu7CLuuctMcx7G9mWEvo",

integration-tests/scripts/run_soak_test.sh

@@ -0,0 +1,59 @@
+#!/bin/bash
+
+NODE_VERSION=18
+
+cd ../smoke || exit
+
+echo "Switching to required Node.js version $NODE_VERSION..."
+export NVM_DIR="$HOME/.nvm"
+[ -s "$NVM_DIR/nvm.sh" ] && \. "$NVM_DIR/nvm.sh"
+nvm use $NODE_VERSION
+
+echo "Initializing soak test..."
+terminated_by_script=false
+while IFS= read -r line; do
+    echo "$line"
+    # Check if the line contains the target string
+    if echo "$line" | grep -q "ocr2:inspect:responses"; then
+        # Send SIGINT (Ctrl+C) to the 'go test' process
+        sudo pkill -INT -P $$ go 2>/dev/null
+        terminated_by_script=true
+        break
+    fi
+done < <(sudo go test -timeout 24h -count=1 -run TestSolanaOCRV2Smoke/embedded -test.timeout 30m 2>&1)
+
+# Capture the PID of the background process
+READER_PID=$!
+
+# Start a background timer (sleeps for 15 minutes, then sends SIGALRM to the script)
+( sleep 900 && kill -s ALRM $$ ) &
+TIMER_PID=$!
+
+# Set a trap to catch the SIGALRM signal for timeout
+trap 'on_timeout' ALRM
+
+# Function to handle timeout
+on_timeout() {
+    echo "Error: failed to start soak test: timeout exceeded (15 minutes)."
+    # Send SIGINT to the 'go test' process
+    pkill -INT -P $$ go 2>/dev/null
+    # Clean up
+    kill "$TIMER_PID" 2>/dev/null
+    kill "$READER_PID" 2>/dev/null
+    exit 1
+}
+
+# Wait for the reader process to finish
+wait "$READER_PID"
+EXIT_STATUS=$?
+
+# Clean up: kill the timer process if it's still running
+kill "$TIMER_PID" 2>/dev/null
+
+if [ "$terminated_by_script" = true ]; then
+    echo "Soak test started successfully"
+    exit 0
+else
+    echo "Soak test failed to start"
+    exit 1
+fi

integration-tests/solclient/solclient.go

@@ -481,7 +481,7 @@ func SendFunds(senderPrivateKey string, receiverPublicKey string, lamports uint6
 	accountTo := solana.MustPublicKeyFromBase58(receiverPublicKey)
 
 	// Get recent blockhash
-	recent, err := rpcClient.GetRecentBlockhash(context.Background(), rpc.CommitmentFinalized)
+	recent, err := rpcClient.GetLatestBlockhash(context.Background(), rpc.CommitmentFinalized)
 	if err != nil {
 		return err
 	}

integration-tests/solclient/store.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink-solana/contracts/generated/store"
 	relaySol "github.com/smartcontractkit/chainlink-solana/pkg/solana"
+	"github.com/smartcontractkit/chainlink-solana/pkg/solana/client"
 )
 
 type Store struct {
@@ -19,7 +20,8 @@ type Store struct {
 }
 
 func (m *Store) GetLatestRoundData() (uint64, uint64, uint64, error) {
-	a, _, err := relaySol.GetLatestTransmission(context.Background(), m.Client.RPC, m.Feed.PublicKey(), rpc.CommitmentConfirmed)
+	getReader := func() (client.AccountReader, error) { return m.Client.RPC, nil }
+	a, _, err := relaySol.GetLatestTransmission(context.Background(), getReader, m.Feed.PublicKey(), rpc.CommitmentConfirmed)
 	if err != nil {
 		return 0, 0, 0, err
 	}

integration-tests/testconfig/testconfig.go

@@ -44,7 +44,7 @@ type TestConfig struct {
 
 	// getter funcs for passing parameters
 	GetChainID func() string
-	GetURL     func() string
+	GetURL     func() []string
 }
 
 const (
@@ -188,22 +188,22 @@ func (c *TestConfig) ReadFromEnvVar() error {
 		c.Network.RpcWsUrls = rpcWsUrls
 	}
 
-	commonRPCURL := ctf_config.MustReadEnvVar_String(E2E_TEST_COMMON_RPC_URL_ENV)
-	if commonRPCURL != "" {
+	commonRPCURL := ctf_config.MustReadEnvVar_Strings(E2E_TEST_COMMON_RPC_URL_ENV, ",")
+	if len(commonRPCURL) > 0 {
 		if c.Common == nil {
 			c.Common = &Common{}
 		}
-		logger.Info().Msgf("Using %s env var to override Common.RPCURL", E2E_TEST_COMMON_RPC_URL_ENV)
-		c.Common.RPCURL = &commonRPCURL
+		logger.Info().Msgf("Using %s env var to override Common.RPCURLs", E2E_TEST_COMMON_RPC_URL_ENV)
+		c.Common.RPCURLs = &commonRPCURL
 	}
 
-	commonWSURL := ctf_config.MustReadEnvVar_String(E2E_TEST_COMMON_WS_URL_ENV)
-	if commonWSURL != "" {
+	commonWSURL := ctf_config.MustReadEnvVar_Strings(E2E_TEST_COMMON_WS_URL_ENV, ",")
+	if len(commonWSURL) > 0 {
 		if c.Common == nil {
 			c.Common = &Common{}
 		}
-		logger.Info().Msgf("Using %s env var to override Common.WsURL", E2E_TEST_COMMON_WS_URL_ENV)
-		c.Common.WsURL = &commonWSURL
+		logger.Info().Msgf("Using %s env var to override Common.WsURLs", E2E_TEST_COMMON_WS_URL_ENV)
+		c.Common.WsURLs = &commonWSURL
 	}
 
 	commonPrivateKey := ctf_config.MustReadEnvVar_String(E2E_TEST_COMMON_PRIVATE_KEY_ENV)
@@ -256,24 +256,44 @@ func (c *TestConfig) GetNodeConfig() *ctf_config.NodeConfig {
 }
 
 func (c *TestConfig) GetNodeConfigTOML() (string, error) {
-	var chainID, url string
+	var chainID string
+	var url []string
 	if c.GetChainID != nil {
 		chainID = c.GetChainID()
 	}
 	if c.GetURL != nil {
 		url = c.GetURL()
 	}
 
-	solConfig := solcfg.TOMLConfig{
-		Enabled: ptr.Ptr(true),
-		ChainID: ptr.Ptr(chainID),
-		Nodes: []*solcfg.Node{
-			{
-				Name: ptr.Ptr("primary"),
-				URL:  config.MustParseURL(url),
-			},
+	mnConfig := solcfg.MultiNodeConfig{
+		MultiNode: solcfg.MultiNode{
+			Enabled:       ptr.Ptr(true),
+			SyncThreshold: ptr.Ptr(uint32(170)),
 		},
 	}
+	mnConfig.SetDefaults()
+
+	var nodes []*solcfg.Node
+	for i, u := range url {
+		nodes = append(nodes, &solcfg.Node{
+			Name: ptr.Ptr(fmt.Sprintf("primary-%d", i)),
+			URL:  config.MustParseURL(u),
+		})
+	}
+
+	chainCfg := solcfg.Chain{
+		// Increase timeout for TransactionSender
+		TxTimeout: config.MustNewDuration(2 * time.Minute),
+	}
+	chainCfg.SetDefaults()
+
+	solConfig := solcfg.TOMLConfig{
+		Enabled:   ptr.Ptr(true),
+		ChainID:   ptr.Ptr(chainID),
+		Nodes:     nodes,
+		MultiNode: mnConfig,
+		Chain:     chainCfg,
+	}
 	baseConfig := node.NewBaseConfig()
 	baseConfig.Solana = solcfg.TOMLConfigs{
 		&solConfig,
@@ -357,12 +377,12 @@ type Common struct {
 	InsideK8s *bool   `toml:"inside_k8"`
 	User      *string `toml:"user"`
 	// if rpc requires api key to be passed as an HTTP header
-	RPCURL             *string `toml:"-"`
-	WsURL              *string `toml:"-"`
-	PrivateKey         *string `toml:"-"`
-	Stateful           *bool   `toml:"stateful_db"`
-	InternalDockerRepo *string `toml:"internal_docker_repo"`
-	DevnetImage        *string `toml:"devnet_image"`
+	RPCURLs            *[]string `toml:"-"`
+	WsURLs             *[]string `toml:"-"`
+	PrivateKey         *string   `toml:"-"`
+	Stateful           *bool     `toml:"stateful_db"`
+	InternalDockerRepo *string   `toml:"internal_docker_repo"`
+	DevnetImage        *string   `toml:"devnet_image"`
 }
 
 type SolanaConfig struct {
@@ -410,10 +430,10 @@ func (c *Common) Validate() error {
 		if c.PrivateKey == nil {
 			return fmt.Errorf("private_key must be set")
 		}
-		if c.RPCURL == nil {
+		if c.RPCURLs == nil {
 			return fmt.Errorf("rpc_url must be set")
 		}
-		if c.WsURL == nil {
+		if c.WsURLs == nil {
 			return fmt.Errorf("rpc_url must be set")
 		}
 

pkg/monitoring/chain_reader.go

@@ -7,6 +7,7 @@ import (
 	"github.com/gagliardetto/solana-go/rpc"
 
 	pkgSolana "github.com/smartcontractkit/chainlink-solana/pkg/solana"
+	"github.com/smartcontractkit/chainlink-solana/pkg/solana/client"
 )
 
 //go:generate mockery --name ChainReader --output ./mocks/
@@ -31,11 +32,13 @@ type chainReader struct {
 }
 
 func (c *chainReader) GetState(ctx context.Context, account solana.PublicKey, commitment rpc.CommitmentType) (state pkgSolana.State, blockHeight uint64, err error) {
-	return pkgSolana.GetState(ctx, c.client, account, commitment)
+	getReader := func() (client.AccountReader, error) { return c.client, nil }
+	return pkgSolana.GetState(ctx, getReader, account, commitment)
 }
 
 func (c *chainReader) GetLatestTransmission(ctx context.Context, account solana.PublicKey, commitment rpc.CommitmentType) (answer pkgSolana.Answer, blockHeight uint64, err error) {
-	return pkgSolana.GetLatestTransmission(ctx, c.client, account, commitment)
+	getReader := func() (client.AccountReader, error) { return c.client, nil }
+	return pkgSolana.GetLatestTransmission(ctx, getReader, account, commitment)
 }
 
 func (c *chainReader) GetTokenAccountBalance(ctx context.Context, account solana.PublicKey, commitment rpc.CommitmentType) (out *rpc.GetTokenAccountBalanceResult, err error) {