Exponential retry (#66)

* Implement exponential backoff for retries

* do not use exp backoff if MaxReconnectWait is set to 0

* solve the deadlock

* add test for MaxReconnectWait and exponential backoff

* update documentation with default values

---------

Co-authored-by: sonali.bhavsar <sonali.bhavsar@espressif.com>

Author: alovak

README.md

@@ -264,6 +264,7 @@ by the server. It does it using `ConnectionClosedHandler`.
 Following options are supported:
 
 * `ReconnectWait` sets the time to wait after first re-connect attempt
+* `MaxReconnectWait` specifies the maximum duration to wait between reconnection attempts, serving as the upper bound for exponential backoff; if set to zero, there's no exponential backoff and ReconnectWait is used for each retry.
 * `ErrorHandler` is called in a goroutine with the errors that can't be returned to the caller (from other goroutines)
 * `MinConnections` is the number of connections required to be established when we connect the pool
 * `ConnectionsFilter` is a function to filter connections in the pool for `Get`, `IsDegraded` or `IsUp` methods

go.sum

@@ -5,12 +5,6 @@ github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/mediocregopher/radix.v2 v0.0.0-20181115013041-b67df6e626f9/go.mod h1:fLRUbhbSd5Px2yKUaGYYPltlyxi1guJz1vCmo1RQL50=
-github.com/moov-io/iso8583 v0.18.3 h1:RYr/zoNxeRcLpJ1VjJ8e8yC0fRK5+xFqKqpnyMTz3+A=
-github.com/moov-io/iso8583 v0.18.3/go.mod h1:QPxDQTxKJPjECpK8vCsh7TpwvKyIrw3tXN/lrdATzCA=
-github.com/moov-io/iso8583 v0.19.0 h1:pIPWsBCydjN6SEoFUWuc4YKlY8e+NzKrnRvx7cfxCdo=
-github.com/moov-io/iso8583 v0.19.0/go.mod h1:QPxDQTxKJPjECpK8vCsh7TpwvKyIrw3tXN/lrdATzCA=
-github.com/moov-io/iso8583 v0.19.1 h1:401h9LFZYxsG/rDc3uHQxZn/rZoAhSMs4WQrnZPoGI8=
-github.com/moov-io/iso8583 v0.19.1/go.mod h1:QPxDQTxKJPjECpK8vCsh7TpwvKyIrw3tXN/lrdATzCA=
 github.com/moov-io/iso8583 v0.19.2 h1:xV5kY0t7UpN/d/polVdNt4Rj6ub4eZuTSTarSqC2NtQ=
 github.com/moov-io/iso8583 v0.19.2/go.mod h1:QPxDQTxKJPjECpK8vCsh7TpwvKyIrw3tXN/lrdATzCA=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

options.go

@@ -13,17 +13,21 @@ import (
 
 type Options struct {
 	// ConnectTimeout sets the timeout for establishing new connections.
+	// The default is 10 seconds.
 	ConnectTimeout time.Duration
 
-	// SendTimeout sets the timeout for a Send operation
+	// SendTimeout sets the timeout for a Send operation.
+	// The default is 30 seconds.
 	SendTimeout time.Duration
 
 	// IdleTime is the period at which the client will be sending ping
-	// message to the server
+	// message to the server.
+	// The default is 5 seconds.
 	IdleTime time.Duration
 
 	// ReadTimeout is the maximum time between read events before the
-	// ReadTimeoutHandler is called
+	// ReadTimeoutHandler is called.
+	// The default is 60 seconds.
 	ReadTimeout time.Duration
 
 	// PingHandler is called when no message was sent during idle time

pool.go

@@ -168,19 +168,23 @@ func (p *Pool) handleClosedConnection(closedConn *Connection) {
 func (p *Pool) recreateConnection(closedConn *Connection) {
 	defer p.wg.Done()
 
-	var conn *Connection
-	var err error
+	reconnectTime := p.Opts.ReconnectWait
+
 	for {
 		select {
-		case <-time.After(p.Opts.ReconnectWait):
-			// Wait before recreating connection
+		case <-time.After(reconnectTime):
+			if p.Opts.MaxReconnectWait != 0 {
+				reconnectTime *= 2
+				if reconnectTime > p.Opts.MaxReconnectWait {
+					reconnectTime = p.Opts.MaxReconnectWait
+				}
+			}
 		case <-p.Done():
-			// If pool is closed, let's get out of here
+			// if pool is closed, let's get out of here
 			return
 		}
 
-		// Recreate connection
-		conn, err = p.Factory(closedConn.addr)
+		conn, err := p.Factory(closedConn.addr)
 		if err != nil {
 			p.handleError(fmt.Errorf("failed to re-create connection for %s: %w", closedConn.addr, err))
 			return
@@ -190,17 +194,17 @@ func (p *Pool) recreateConnection(closedConn *Connection) {
 		// recreate goroutine to create new connection for the same address
 		conn.SetOptions(ConnectionClosedHandler(p.handleClosedConnection))
 
-		err = conn.Connect()
-		if err == nil {
-			break
+		// if we successfully reconnected, add connection to the pool and return
+		if err = conn.Connect(); err == nil {
+			p.mu.Lock()
+			p.connections = append(p.connections, conn)
+			p.mu.Unlock()
+
+			return
 		}
 
 		p.handleError(fmt.Errorf("failed to reconnect to %s: %w", conn.addr, err))
 	}
-
-	p.mu.Lock()
-	p.connections = append(p.connections, conn)
-	p.mu.Unlock()
 }
 
 // Close closes all connections in the pool

pool_options.go

@@ -8,25 +8,38 @@ type PoolOption func(*PoolOptions) error
 
 type PoolOptions struct {
 	// ReconnectWait sets the time to wait after first re-connect attempt
+	// The default is 5 seconds
 	ReconnectWait time.Duration
 
+	// MaxReconnectWait specifies the maximum duration to wait between
+	// reconnection attempts, serving as the upper bound for exponential
+	// backoff.
+	// A zero value means no exponential backoff and ReconnectWait is used
+	// for each retry.
+	// The default is 0.
+	MaxReconnectWait time.Duration
+
 	// ErrorHandler is called in a goroutine with the errors that can't be
 	// returned to the caller
 	ErrorHandler func(err error)
 
 	// MinConnections is the number of connections required to be established when
-	// we connect the pool
+	// we connect the pool.
+	// A zero value means that Pool will not return error on `Connect` and will
+	// try to connect to all the addresses in the pool.
+	// The default is 1.
 	MinConnections int
 
-	// ConntionsFilter is a function to filter connections in the pool
+	// ConnectionsFilter is a function to filter connections in the pool
 	// when Get() is called
 	ConnectionsFilter func(*Connection) bool
 }
 
 func GetDefaultPoolOptions() PoolOptions {
 	return PoolOptions{
-		ReconnectWait:  5 * time.Second,
-		MinConnections: 1,
+		ReconnectWait:    5 * time.Second,
+		MaxReconnectWait: 0,
+		MinConnections:   1,
 	}
 }
 
@@ -37,6 +50,13 @@ func PoolReconnectWait(rw time.Duration) PoolOption {
 	}
 }
 
+func PoolMaxReconnectWait(rw time.Duration) PoolOption {
+	return func(opts *PoolOptions) error {
+		opts.MaxReconnectWait = rw
+		return nil
+	}
+}
+
 func PoolMinConnections(n int) PoolOption {
 	return func(opts *PoolOptions) error {
 		opts.MinConnections = n

pool_test.go

@@ -73,7 +73,11 @@ func TestPool(t *testing.T) {
 	// And the pool of connections
 	// one of our tests
 	reconnectWait := 500 * time.Millisecond
-	pool, err := connection.NewPool(factory, addrs, connection.PoolReconnectWait(reconnectWait))
+	pool, err := connection.NewPool(
+		factory,
+		addrs,
+		connection.PoolReconnectWait(reconnectWait),
+	)
 	require.NoError(t, err)
 	defer pool.Close()
 
@@ -146,6 +150,59 @@ func TestPool(t *testing.T) {
 		}, 2000*time.Millisecond, 50*time.Millisecond, "expect to have one less connection")
 	})
 
+	t.Run("when MaxReconnectWait is set reconnect wait time will exponentially increase", func(t *testing.T) {
+		// Context: MaxReconnectWait is set to 400ms and initial
+		// ReconnectWait is 100ms. The server is offline, so the pool
+		// will try to reconnect endlessly with exponential backoff
+		// until we close the pool. We will wait for 850ms before
+		// checking the number of reconnects. Expected number of
+		// reconnects within 850ms is 3. The sequence of reconnect
+		// waits is: 100ms, 200ms, 400ms (total 700ms). Next reconnect
+		// would take 400ms, which exceeds 1 second. Thus, 3 reconnect
+		// attempts are expected.
+
+		// Counter for connection attempts
+		var connectAttempts atomic.Int32
+
+		// Factory method will be called on each connection attempt
+		factory := func(addr string) (*connection.Connection, error) {
+			// Increment connection attempts counter
+			connectAttempts.Add(1)
+
+			c, err := connection.New(
+				addr,
+				testSpec,
+				readMessageLength,
+				writeMessageLength,
+			)
+			if err != nil {
+				return nil, fmt.Errorf("building iso 8583 connection: %w", err)
+			}
+
+			return c, nil
+		}
+
+		pool, err := connection.NewPool(
+			factory,
+			[]string{"no-live-server-address"}, // connect only to the first server
+			connection.PoolReconnectWait(100*time.Millisecond),
+			connection.PoolMaxReconnectWait(400*time.Millisecond),
+			// let pool start even without connections (it will start reconnecting)
+			connection.PoolMinConnections(0),
+		)
+		require.NoError(t, err)
+
+		// There should be no error even if we try to connect to a non-existing server
+		// as the min connections is set to 0
+		err = pool.Connect()
+		require.NoError(t, err)
+		defer pool.Close()
+
+		time.Sleep(850 * time.Millisecond)
+
+		require.Equal(t, int32(4), connectAttempts.Load(), "expected 4 connection attempts (3 reconnects + 1 initial connect)")
+	})
+
 	t.Run("Close() closes all connections", func(t *testing.T) {
 		require.NotZero(t, pool.Connections())
 

server/server.go

@@ -181,10 +181,10 @@ func (s *Server) Close() {
 		s.ln.Close()
 	}
 
-	s.wg.Wait()
-
 	s.isClosed = true
 	s.mu.Unlock()
+
+	s.wg.Wait()
 }
 
 func (s *Server) handleConnection(conn net.Conn) error {