Fixed network increase backoff concurrency issue

.github/workflows/build.yml

@@ -12,7 +12,7 @@ on:
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
   cancel-in-progress: true
-    
+
 jobs:
     build:
         name: Build
@@ -62,12 +62,12 @@ jobs:
 
             - name: Tests Unit
               if: success()
-              run: go test -tags="unit" -timeout 9999s -v -coverprofile=unit.out -covermode=atomic
+              run: go test -tags="unit" -timeout 9999s -v -coverprofile=unit.out -covermode=atomic -race
 
             - name: Tests Integration
               if: success()
               run: go test -tags="e2e" -timeout 9999s -v -coverprofile=e2e.out -covermode=atomic
-            
+
             - name: Tests Testnets
               if: success()
               env:

address_book_query_e2e_test.go

@@ -24,38 +24,42 @@ package hedera
  */
 
 import (
+	"github.com/stretchr/testify/require"
 	"os"
 	"testing"
-	"time"
-
-	"github.com/stretchr/testify/require"
 )
 
 func TestIntegrationAddressBookQueryUpdateAll(t *testing.T) {
-	client := ClientForPreviewnet()
-	// There are some limitation on requests: unexpected HTTP status code received from server: 429 (Too Many Requests)
-	time.Sleep(time.Second * 5)
+	client, err := ClientFromConfig([]byte(`{"network":"previewnet"}`))
+	require.NoError(t, err)
+	client.SetMirrorNetwork(previewnetMirror)
+
 	previewnet, err := NewAddressBookQuery().
 		SetFileID(FileIDForAddressBook()).
+		SetMaxAttempts(5).
 		Execute(client)
 	require.NoError(t, err)
 	require.Greater(t, len(previewnet.NodeAddresses), 0)
 
-	client = ClientForTestnet()
-	// There are some limitation on requests: unexpected HTTP status code received from server: 429 (Too Many Requests)
-	// Testnet specifically has a more aggresive rate limit
-	time.Sleep(time.Second * 20)
+	client, err = ClientFromConfig([]byte(`{"network":"testnet"}`))
+	require.NoError(t, err)
+	client.SetMirrorNetwork(testnetMirror)
+
 	testnet, err := NewAddressBookQuery().
 		SetFileID(FileIDForAddressBook()).
+		SetMaxAttempts(5).
 		Execute(client)
+
 	require.NoError(t, err)
 	require.Greater(t, len(testnet.NodeAddresses), 0)
 
-	client = ClientForMainnet()
-	// There are some limitation on requests: unexpected HTTP status code received from server: 429 (Too Many Requests)
-	time.Sleep(time.Second * 5)
+	client, err = ClientFromConfig([]byte(`{"network":"mainnet"}`))
+	require.NoError(t, err)
+	client.SetMirrorNetwork(mainnetMirror)
+
 	mainnet, err := NewAddressBookQuery().
 		SetFileID(FileIDForAddressBook()).
+		SetMaxAttempts(5).
 		Execute(client)
 	require.NoError(t, err)
 	require.Greater(t, len(mainnet.NodeAddresses), 0)

managed_network.go

@@ -24,13 +24,15 @@ import (
 	"crypto/rand"
 	"math"
 	"math/big"
+	"sync"
 	"time"
 )
 
 type _ManagedNetwork struct {
 	network                map[string][]_IManagedNode
 	nodes                  []_IManagedNode
 	healthyNodes           []_IManagedNode
+	healthyNodesMutex      *sync.RWMutex
 	maxNodeAttempts        int
 	minBackoff             time.Duration
 	maxBackoff             time.Duration
@@ -48,6 +50,7 @@ func _NewManagedNetwork() _ManagedNetwork {
 		network:                map[string][]_IManagedNode{},
 		nodes:                  []_IManagedNode{},
 		healthyNodes:           []_IManagedNode{},
+		healthyNodesMutex:      &sync.RWMutex{},
 		maxNodeAttempts:        -1,
 		minBackoff:             8 * time.Second,
 		maxBackoff:             1 * time.Hour,
@@ -111,6 +114,9 @@ func (this *_ManagedNetwork) _SetNetwork(network map[string]_IManagedNode) error
 func (this *_ManagedNetwork) _ReadmitNodes() {
 	now := time.Now()
 
+	this.healthyNodesMutex.Lock()
+	defer this.healthyNodesMutex.Unlock()
+
 	if this.earliestReadmitTime.Before(now) {
 		nextEarliestReadmitTime := now.Add(this.maxNodeReadmitPeriod)
 
@@ -189,6 +195,8 @@ func (this *_ManagedNetwork) _SetMinBackoff(minBackoff time.Duration) {
 
 func (this *_ManagedNetwork) _GetNode() _IManagedNode {
 	this._ReadmitNodes()
+	this.healthyNodesMutex.RLock()
+	defer this.healthyNodesMutex.RUnlock()
 
 	if len(this.healthyNodes) == 0 {
 		panic("failed to find a healthy working node")

mirror_node.go

@@ -163,8 +163,8 @@ func (node *_MirrorNode) _GetNetworkServiceClient() (*mirror.NetworkServiceClien
 	}
 
 	var kacp = keepalive.ClientParameters{
-		Time:                10 * time.Second,
-		Timeout:             time.Second,
+		Time:                time.Minute,
+		Timeout:             20 * time.Second,
 		PermitWithoutStream: true,
 	}
 

mock_test.go

@@ -475,13 +475,14 @@ func NewMockClientAndServer(allNodeResponses [][]interface{}) (*Client, *MockSer
 	for i, responses := range allNodeResponses {
 		responses := responses
 
+		serverReady := make(chan bool)
 		nodeAccountID := AccountID{Account: uint64(3 + i)}
 		go func() {
 			servers[i] = NewMockServer(responses)
+			serverReady <- true
 		}()
 
-		for servers[i] == nil {
-		}
+		<-serverReady
 
 		network[servers[i].listener.Addr().String()] = nodeAccountID
 		mirrorNetwork[i] = servers[i].listener.Addr().String()

network.go

@@ -64,9 +64,11 @@ func (network *_Network) _GetNetwork() map[string]AccountID {
 }
 
 func (network *_Network) _IncreaseBackoff(node *_Node) {
+	network.healthyNodesMutex.Lock()
+	defer network.healthyNodesMutex.Unlock()
 	node._IncreaseBackoff()
 
-	index := 0
+	index := -1
 	for i, healthyNode := range network.healthyNodes {
 		if node == healthyNode {
 			index = i
@@ -130,8 +132,11 @@ func (network *_Network) _SetLedgerID(id LedgerID) {
 
 func (network *_Network) _GetNodeAccountIDsForExecute() []AccountID { //nolint
 	nodes := make([]AccountID, 0)
+	nodesForTransaction := network._GetNumberOfNodesForTransaction()
 
-	for i := 0; i < network._GetNumberOfNodesForTransaction(); i++ {
+	network.healthyNodesMutex.RLock()
+	defer network.healthyNodesMutex.RUnlock()
+	for i := 0; i < nodesForTransaction; i++ {
 		nodes = append(nodes, network.healthyNodes[i].(*_Node).accountID)
 	}
 

network_unit_test.go

@@ -24,11 +24,22 @@ package hedera
  */
 
 import (
-	"testing"
-
 	"github.com/stretchr/testify/require"
+	"sync"
+	"testing"
+	"time"
 )
 
+func newNetworkMockNodes() map[string]AccountID {
+	nodes := make(map[string]AccountID, 2)
+	nodes["0.testnet.hedera.com:50211"] = AccountID{0, 0, 3, nil, nil, nil}
+	nodes["1.testnet.hedera.com:50211"] = AccountID{0, 0, 4, nil, nil, nil}
+	nodes["2.testnet.hedera.com:50211"] = AccountID{0, 0, 5, nil, nil, nil}
+	nodes["3.testnet.hedera.com:50211"] = AccountID{0, 0, 6, nil, nil, nil}
+	nodes["4.testnet.hedera.com:50211"] = AccountID{0, 0, 7, nil, nil, nil}
+	return nodes
+}
+
 func TestUnitNetworkAddressBookGetsSet(t *testing.T) {
 	t.Parallel()
 
@@ -43,3 +54,59 @@ func TestUnitNetworkAddressBookGetsSet(t *testing.T) {
 
 	require.True(t, network.addressBook != nil)
 }
+
+func TestUnitNetworkIncreaseBackoffConcurrent(t *testing.T) {
+	t.Parallel()
+
+	network := _NewNetwork()
+	nodes := newNetworkMockNodes()
+	err := network.SetNetwork(nodes)
+	require.NoError(t, err)
+
+	node := network._GetNode()
+	require.NotNil(t, node)
+
+	numThreads := 20
+	var wg sync.WaitGroup
+	wg.Add(numThreads)
+	for i := 0; i < numThreads; i++ {
+		go func() {
+			network._IncreaseBackoff(node)
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+
+	require.Equal(t, len(nodes)-1, len(network.healthyNodes))
+}
+
+func TestUnitConcurrentGetNodeReadmit(t *testing.T) {
+	t.Parallel()
+
+	network := _NewNetwork()
+	nodes := newNetworkMockNodes()
+	err := network.SetNetwork(nodes)
+	network._SetMinNodeReadmitPeriod(0)
+	network._SetMaxNodeReadmitPeriod(0)
+	require.NoError(t, err)
+
+	for _, node := range network.nodes {
+		node._SetMaxBackoff(-1 * time.Minute)
+	}
+
+	numThreads := 3
+	var wg sync.WaitGroup
+	wg.Add(numThreads)
+	for i := 0; i < numThreads; i++ {
+		go func() {
+			for i := 0; i < 20; i++ {
+				node := network._GetNode()
+				network._IncreaseBackoff(node)
+			}
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+	network._ReadmitNodes()
+	require.Equal(t, len(nodes), len(network.healthyNodes))
+}