use exponential backoff in reconnection code

Author: laser

client.go

@@ -6,8 +6,6 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
-	"math"
-	"math/rand"
 	"net/http"
 	"reflect"
 	"sync/atomic"
@@ -86,7 +84,7 @@ type client struct {
 // NewMergeClient is like NewClient, but allows to specify multiple structs
 // to be filled in the same namespace, using one connection
 func NewMergeClient(addr string, namespace string, outs []interface{}, requestHeader http.Header, opts ...Option) (ClientCloser, error) {
-	var config Config
+	config := defaultConfig
 	for _, o := range opts {
 		o(&config)
 	}
@@ -116,13 +114,13 @@ func NewMergeClient(addr string, namespace string, outs []interface{}, requestHe
 
 	handlers := map[string]rpcHandler{}
 	go (&wsConn{
-		conn:              conn,
-		connFactory:       connFactory,
-		reconnectInterval: config.ReconnectInterval,
-		handler:           handlers,
-		requests:          c.requests,
-		stop:              stop,
-		exiting:           exiting,
+		conn:             conn,
+		connFactory:      connFactory,
+		reconnectBackoff: config.reconnectBackoff,
+		handler:          handlers,
+		requests:         c.requests,
+		stop:             stop,
+		exiting:          exiting,
 	}).handleWsConn(context.TODO())
 
 	for _, handler := range outs {
@@ -386,6 +384,11 @@ func (fn *rpcFunc) handleRpcCall(args []reflect.Value) (results []reflect.Value)
 		}
 	}
 
+	b := backoff{
+		maxDelay: methodMaxRetryDelay,
+		minDelay: methodMinRetryDelay,
+	}
+
 	var resp clientResponse
 	var err error
 	// keep retrying if got a forced closed websocket conn and calling method
@@ -418,30 +421,12 @@ func (fn *rpcFunc) handleRpcCall(args []reflect.Value) (results []reflect.Value)
 			break
 		}
 
-		time.Sleep(backoff(attempt))
+		time.Sleep(b.next(attempt))
 	}
 
 	return fn.processResponse(resp, retVal())
 }
 
-func backoff(attempt int) time.Duration {
-	if attempt < 0 {
-		return methodMinRetryDelay
-	}
-
-	minf := float64(methodMinRetryDelay)
-	durf := minf * math.Pow(1.5, float64(attempt))
-	durf = durf + rand.Float64()*minf
-
-	delay := time.Duration(durf)
-
-	if delay > methodMaxRetryDelay {
-		return methodMaxRetryDelay
-	}
-
-	return delay
-}
-
 func (c *client) makeRpcFunc(f reflect.StructField) (reflect.Value, error) {
 	ftyp := f.Type
 	if ftyp.Kind() != reflect.Func {

options.go

@@ -7,19 +7,25 @@ import (
 )
 
 type Config struct {
-	ReconnectInterval time.Duration
+	reconnectBackoff backoff
 
 	proxyConnFactory func(func() (*websocket.Conn, error)) func() (*websocket.Conn, error) // for testing
 }
 
 var defaultConfig = Config{
-	ReconnectInterval: time.Second * 5,
+	reconnectBackoff: backoff{
+		minDelay: 100 * time.Millisecond,
+		maxDelay: 5 * time.Second,
+	},
 }
 
 type Option func(c *Config)
 
-func WithReconnectInterval(d time.Duration) func(c *Config) {
+func WithReconnectBackoff(minDelay, maxDelay time.Duration) func(c *Config) {
 	return func(c *Config) {
-		c.ReconnectInterval = d
+		c.reconnectBackoff = backoff{
+			minDelay: minDelay,
+			maxDelay: maxDelay,
+		}
 	}
 }

rpc_test.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/gorilla/websocket"
 	logging "github.com/ipfs/go-log/v2"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -62,6 +63,63 @@ func (h *SimpleServerHandler) StringMatch(t TestType, i2 int64) (out TestOut, er
 	return
 }
 
+func TestReconnection(t *testing.T) {
+	var rpcClient struct {
+		Add func(int) error
+	}
+
+	rpcHandler := SimpleServerHandler{}
+
+	rpcServer := NewServer()
+	rpcServer.Register("SimpleServerHandler", &rpcHandler)
+
+	testServ := httptest.NewServer(rpcServer)
+	defer testServ.Close()
+
+	// capture connection attempts for this duration
+	captureDuration := 3 * time.Second
+
+	// run the test until the timer expires
+	timer := time.NewTimer(captureDuration)
+
+	// record the number of connection attempts during this test
+	connectionAttempts := 1
+
+	closer, err := NewMergeClient("ws://"+testServ.Listener.Addr().String(), "SimpleServerHandler", []interface{}{&rpcClient}, nil, func(c *Config) {
+		c.proxyConnFactory = func(f func() (*websocket.Conn, error)) func() (*websocket.Conn, error) {
+			return func() (*websocket.Conn, error) {
+				defer func() {
+					connectionAttempts++
+				}()
+
+				if connectionAttempts > 1 {
+					return nil, errors.New("simulates a failed reconnect attempt")
+				}
+
+				c, err := f()
+				if err != nil {
+					return nil, err
+				}
+
+				// closing the connection here triggers the reconnect logic
+				_ = c.Close()
+
+				return c, nil
+			}
+		}
+	})
+	require.NoError(t, err)
+	defer closer()
+
+	// let the JSON-RPC library attempt to reconnect until the timer runs out
+	<-timer.C
+
+	// do some math
+	attemptsPerSecond := int64(connectionAttempts) / int64(captureDuration/time.Second)
+
+	assert.Less(t, attemptsPerSecond, int64(50))
+}
+
 func TestRPC(t *testing.T) {
 	// setup server
 

util.go

@@ -3,7 +3,10 @@ package jsonrpc
 import (
 	"encoding/json"
 	"fmt"
+	"math"
+	"math/rand"
 	"reflect"
+	"time"
 )
 
 type param struct {
@@ -53,3 +56,26 @@ func processFuncOut(funcType reflect.Type) (valOut int, errOut int, n int) {
 
 	return
 }
+
+type backoff struct {
+	minDelay time.Duration
+	maxDelay time.Duration
+}
+
+func (b *backoff) next(attempt int) time.Duration {
+	if attempt < 0 {
+		return b.minDelay
+	}
+
+	minf := float64(b.minDelay)
+	durf := minf * math.Pow(1.5, float64(attempt))
+	durf = durf + rand.Float64()*minf
+
+	delay := time.Duration(durf)
+
+	if delay > b.maxDelay {
+		return b.maxDelay
+	}
+
+	return delay
+}

websocket.go

@@ -43,13 +43,13 @@ type outChanReg struct {
 
 type wsConn struct {
 	// outside params
-	conn              *websocket.Conn
-	connFactory       func() (*websocket.Conn, error)
-	reconnectInterval time.Duration
-	handler           handlers
-	requests          <-chan clientRequest
-	stop              <-chan struct{}
-	exiting           chan struct{}
+	conn             *websocket.Conn
+	connFactory      func() (*websocket.Conn, error)
+	reconnectBackoff backoff
+	handler          handlers
+	requests         <-chan clientRequest
+	stop             <-chan struct{}
+	exiting          chan struct{}
 
 	// incoming messages
 	incoming    chan io.Reader
@@ -490,13 +490,15 @@ func (c *wsConn) handleWsConn(ctx context.Context) {
 								return
 							}
 
+							attempts := 0
 							var conn *websocket.Conn
 							for conn == nil {
-								time.Sleep(c.reconnectInterval)
+								time.Sleep(c.reconnectBackoff.next(attempts))
 								var err error
 								if conn, err = c.connFactory(); err != nil {
 									log.Debugw("websocket connection retry failed", "error", err)
 								}
+								attempts++
 							}
 
 							c.writeLk.Lock()