PRT-554 adding listener with retry to RPC Consumer (#376)

* adding listener with retry * fixing merge conflict * adding provider reconnect retry mechanism * recover from address already in use error
lavanet · Mar 29, 2023 · 6317b10 · 6317b10
1 parent 646ee7b
commit 6317b10
Show file tree

Hide file tree

Showing 8 changed files with 35 additions and 43 deletions.
diff --git a/docs/static/openapi.yml b/docs/static/openapi.yml
@@ -29674,16 +29674,6 @@ paths:
           in: query
           required: false
           type: boolean
-        - name: pagination.reverse
-          description: >-
-            reverse is set to true if results are to be returned in the
-            descending order.
-
-
-            Since: cosmos-sdk 0.43
-          in: query
-          required: false
-          type: boolean
       tags:
         - Query
   '/lavanet/lava/epochstorage/fixated_params/{index}':
@@ -29943,16 +29933,6 @@ paths:
           in: query
           required: false
           type: boolean
-        - name: pagination.reverse
-          description: >-
-            reverse is set to true if results are to be returned in the
-            descending order.
-
-
-            Since: cosmos-sdk 0.43
-          in: query
-          required: false
-          type: boolean
       tags:
         - Query
   '/lavanet/lava/epochstorage/stake_storage/{index}':

diff --git a/protocol/chainlib/common.go b/protocol/chainlib/common.go
@@ -3,6 +3,7 @@ package chainlib
 import (
 	"encoding/json"
 	"fmt"
+	"net"
 	"net/url"
 	"regexp"
 	"strings"
@@ -19,6 +20,7 @@ import (
 
 const (
 	ContextUserValueKeyDappID = "dappID"
+	RetryListeningInterval    = 10 // seconds
 )
 
 type BaseChainParser struct {
@@ -201,6 +203,28 @@ func verifyTendermintEndpoint(endpoints []common.NodeUrl) (websocketEndpoint com
 	return websocketEndpoint, httpEndpoint
 }
 
+func ListenWithRetry(app *fiber.App, address string) {
+	for {
+		err := app.Listen(address)
+		if err != nil {
+			utils.LavaFormatError("app.Listen(listenAddr)", err)
+		}
+		time.Sleep(RetryListeningInterval * time.Second)
+	}
+}
+
+func GetListenerWithRetryGrpc(protocol string, addr string) net.Listener {
+	for {
+		lis, err := net.Listen(protocol, addr)
+		if err == nil {
+			return lis
+		}
+		utils.LavaFormatError("failure setting up listener, net.Listen(protocol, addr)", err, utils.Attribute{Key: "listenAddr", Value: addr})
+		time.Sleep(RetryListeningInterval * time.Second)
+		utils.LavaFormatWarning("Attempting connection retry", nil)
+	}
+}
+
 func GetApiInterfaceFromServiceApi(serviceApi *spectypes.ServiceApi, connectionType string) *spectypes.ApiInterface {
 	var apiInterface *spectypes.ApiInterface = nil
 	for i := range serviceApi.ApiInterfaces {

diff --git a/protocol/chainlib/grpc.go b/protocol/chainlib/grpc.go
@@ -6,7 +6,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"net"
 	"net/http"
 	"strings"
 	"sync"
@@ -203,10 +202,7 @@ func (apil *GrpcChainListener) Serve(ctx context.Context) {
 
 	utils.LavaFormatInfo("gRPC PortalStart")
 
-	lis, err := net.Listen("tcp", apil.endpoint.NetworkAddress)
-	if err != nil {
-		utils.LavaFormatFatal("provider failure setting up listener", err, utils.Attribute{Key: "listenAddr", Value: apil.endpoint.NetworkAddress})
-	}
+	lis := GetListenerWithRetryGrpc("tcp", apil.endpoint.NetworkAddress)
 	apiInterface := apil.endpoint.ApiInterface
 	sendRelayCallback := func(ctx context.Context, method string, reqBody []byte) ([]byte, error) {
 		ctx = utils.WithUniqueIdentifier(ctx, utils.GenerateUniqueIdentifier())
@@ -215,7 +211,7 @@ func (apil *GrpcChainListener) Serve(ctx context.Context) {
 		utils.LavaFormatInfo("GRPC Got Relay ", utils.Attribute{Key: "GUID", Value: ctx}, utils.Attribute{Key: "method", Value: method})
 		var relayReply *pairingtypes.RelayReply
 		metricsData := metrics.NewRelayAnalytics("NoDappID", apil.endpoint.ChainID, apiInterface)
-		relayReply, _, err = apil.relaySender.SendRelay(ctx, method, string(reqBody), "", "NoDappID", metricsData)
+		relayReply, _, err := apil.relaySender.SendRelay(ctx, method, string(reqBody), "", "NoDappID", metricsData)
 		go apil.logger.AddMetricForGrpc(metricsData, err, &metadataValues)
 
 		if err != nil {

diff --git a/protocol/chainlib/jsonRPC.go b/protocol/chainlib/jsonRPC.go
@@ -336,10 +336,7 @@ func (apil *JsonRPCChainListener) Serve(ctx context.Context) {
 	})
 
 	// Go
-	err := app.Listen(apil.endpoint.NetworkAddress)
-	if err != nil {
-		utils.LavaFormatError("app.Listen(listenAddr)", err)
-	}
+	ListenWithRetry(app, apil.endpoint.NetworkAddress)
 }
 
 type JrpcChainProxy struct {

diff --git a/protocol/chainlib/rest.go b/protocol/chainlib/rest.go
@@ -294,10 +294,7 @@ func (apil *RestChainListener) Serve(ctx context.Context) {
 	})
 
 	// Go
-	err := app.Listen(apil.endpoint.NetworkAddress)
-	if err != nil {
-		utils.LavaFormatError("app.Listen(listenAddr)", err)
-	}
+	ListenWithRetry(app, apil.endpoint.NetworkAddress)
 }
 
 type RestChainProxy struct {

diff --git a/protocol/chainlib/tendermintRPC.go b/protocol/chainlib/tendermintRPC.go
@@ -417,10 +417,7 @@ func (apil *TendermintRpcChainListener) Serve(ctx context.Context) {
 	})
 	//
 	// Go
-	err := app.Listen(apil.endpoint.NetworkAddress)
-	if err != nil {
-		utils.LavaFormatError("app.Listen(listenAddr)", err)
-	}
+	ListenWithRetry(app, apil.endpoint.NetworkAddress)
 }
 
 type tendermintRpcChainProxy struct {

diff --git a/protocol/rpcprovider/provider_listener.go b/protocol/rpcprovider/provider_listener.go
@@ -3,11 +3,11 @@ package rpcprovider
 import (
 	"context"
 	"errors"
-	"net"
 	"net/http"
 	"strings"
 	"sync"
 
+	"github.com/lavanet/lava/protocol/chainlib"
 	"github.com/lavanet/lava/protocol/lavasession"
 
 	"github.com/improbable-eng/grpc-web/go/grpcweb"
@@ -53,10 +53,7 @@ func NewProviderListener(ctx context.Context, networkAddress string) *ProviderLi
 	pl := &ProviderListener{networkAddress: networkAddress}
 
 	// GRPC
-	lis, err := net.Listen("tcp", networkAddress)
-	if err != nil {
-		utils.LavaFormatFatal("provider failure setting up listener", err, utils.Attribute{Key: "listenAddr", Value: networkAddress})
-	}
+	lis := chainlib.GetListenerWithRetryGrpc("tcp", networkAddress)
 	grpcServer := grpc.NewServer()
 
 	wrappedServer := grpcweb.WrapServer(grpcServer)

diff --git a/scripts/pre_setups/init_lava_only_test.sh b/scripts/pre_setups/init_lava_only_test.sh
@@ -3,6 +3,10 @@ __dir=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
 source "$__dir"/../useful_commands.sh
 . "${__dir}"/../vars/variables.sh
 
+LOGS_DIR=${__dir}/../../testutil/debugging/logs
+mkdir -p $LOGS_DIR
+rm $LOGS_DIR/*.log
+
 killall screen
 screen -wipe
 GASPRICE="0.000000001ulava"