Merge #37

37: Backoff mechanism r=48d90782 a=48d90782



Co-authored-by: Valery Piashchynski <[email protected]>

Author: bors[bot]

.gitignore

@@ -6,4 +6,6 @@ vendor/
 clover.xml
 .DS_Store
 tests/.rr-test.yml
-go.sum
+go.sum
+psr-worker.php
+.rr-sample.yaml

README.md

@@ -5,6 +5,7 @@
 [![Go Report Card](https://goreportcard.com/badge/github.com/spiral/jobs)](https://goreportcard.com/report/github.com/spiral/jobs)
 [![Codecov](https://codecov.io/gh/spiral/jobs/branch/master/graph/badge.svg)](https://codecov.io/gh/spiral/jobs/)
 
+
 ## Documentation
   * [Installation and Configuration](https://spiral.dev/docs/queue-configuration)
   * [Console Commands](https://spiral.dev/docs/queue-commands)

broker/amqp/conn.go

@@ -2,14 +2,14 @@ package amqp
 
 import (
 	"fmt"
+	"github.com/cenkalti/backoff/v4"
 	"github.com/streadway/amqp"
 	"sync"
 	"time"
 )
 
 // manages set of AMQP channels
 type chanPool struct {
-	tout      time.Duration
 	mu        sync.Mutex
 	conn      *amqp.Connection
 	channels  map[string]*channel
@@ -36,7 +36,6 @@ func newConn(dial dialer, tout time.Duration) (*chanPool, error) {
 	}
 
 	cp := &chanPool{
-		tout:      tout,
 		conn:      conn,
 		channels:  make(map[string]*channel),
 		wait:      make(chan interface{}),
@@ -112,46 +111,48 @@ func (cp *chanPool) watch(dial dialer, errors chan *amqp.Error) {
 			cp.channels = nil
 			cp.mu.Unlock()
 
-			conn, errChan := cp.reconnect(dial)
-			if conn == nil {
+			// initialize the backoff
+			expb := backoff.NewExponentialBackOff()
+			expb.MaxInterval = DefaultMaxInterval
+
+			//reconnect function
+			reconnect := func() error {
+				cp.mu.Lock()
+				defer cp.mu.Unlock()
+				conn, err := dial()
+				if err != nil {
+					// still failing
+					fmt.Println(fmt.Sprintf("error during the amqp dialing, %s", err.Error()))
+					return err
+				}
+
+				// TODO ADD LOGGING
+				fmt.Println("------amqp successfully redialed------")
+
+				// here we are reconnected
+				// replace the connection
+				cp.conn = conn
+				// re-init the channels
+				cp.channels = make(map[string]*channel)
+				errors = cp.conn.NotifyClose(make(chan *amqp.Error))
+				return nil
+			}
+
+
+			errb := backoff.Retry(reconnect, expb)
+			if errb != nil {
+				fmt.Println(fmt.Sprintf("backoff Retry error, %s", errb.Error()))
+				// reconnection failed
 				cp.mu.Lock()
 				close(cp.connected)
 				cp.mu.Unlock()
-
-				// interrupted
 				return
 			}
-
-			cp.mu.Lock()
-			cp.conn = conn
-			cp.channels = make(map[string]*channel)
-			errors = errChan
-			cp.mu.Unlock()
-
 			close(cp.connected)
 		}
 	}
 }
 
-func (cp *chanPool) reconnect(dial dialer) (conn *amqp.Connection, errors chan *amqp.Error) {
-	for {
-		select {
-		case <-cp.wait:
-			// connection has been cancelled is not possible
-			return nil, nil
-
-		case <-time.NewTimer(cp.tout).C:
-			conn, err := dial()
-			if err != nil {
-				// still failing
-				continue
-			}
-
-			return conn, conn.NotifyClose(make(chan *amqp.Error))
-		}
-	}
-}
-
 // channel allocates new channel on amqp connection
 func (cp *chanPool) channel(name string) (*channel, error) {
 	cp.mu.Lock()
@@ -161,7 +162,7 @@ func (cp *chanPool) channel(name string) (*channel, error) {
 	if dead {
 		// wait for connection restoration (doubled the timeout duration)
 		select {
-		case <-time.NewTimer(cp.tout * 2).C:
+		case <-time.NewTimer(DefaultMaxInterval * 2).C:
 			return nil, fmt.Errorf("connection is dead")
 		case <-cp.connected:
 			// connected

broker/amqp/constants.go

@@ -0,0 +1,6 @@
+package amqp
+
+import "time"
+
+// DefaultMaxInterval is the max reconnect time interval
+const DefaultMaxInterval = 30 * time.Second

broker/beanstalk/conn.go

@@ -3,6 +3,7 @@ package beanstalk
 import (
 	"fmt"
 	"github.com/beanstalkd/go-beanstalk"
+	"github.com/cenkalti/backoff/v4"
 	"strings"
 	"sync"
 	"time"
@@ -114,15 +115,30 @@ func (cn *conn) watch(network, addr string) {
 		select {
 		case <-cn.dead:
 			// try to reconnect
-			if conn, err := beanstalk.Dial(network, addr); err == nil {
+			// TODO add logging here
+			expb := backoff.NewExponentialBackOff()
+			expb.MaxInterval = time.Second * 5
+
+			reconnect := func() error {
+				conn, err := beanstalk.Dial(network, addr)
+				if err != nil {
+					fmt.Println(fmt.Sprintf("redial: error during the beanstalk dialing, %s", err.Error()))
+					return err
+				}
+
+				// TODO ADD LOGGING
+				fmt.Println("------beanstalk successfully redialed------")
+
 				cn.conn = conn
 				cn.free <- nil
-				continue
+				return nil
 			}
 
-			// retry later
-			time.Sleep(cn.tout)
-			cn.dead <- nil
+			err := backoff.Retry(reconnect, expb)
+			if err != nil {
+				fmt.Println(fmt.Sprintf("redial failed: %s", err.Error()))
+				cn.dead <- nil
+			}
 
 		case <-cn.stop:
 			cn.lock.L.Lock()

broker/beanstalk/constants.go

@@ -0,0 +1,6 @@
+package beanstalk
+
+import "time"
+
+// DefaultMaxInterval is the max reconnect time interval
+const DefaultMaxInterval = 30 * time.Second

broker/sqs/job.go

@@ -19,7 +19,7 @@ var jobAttributes = []*string{
 
 // pack job metadata into headers
 func pack(url *string, j *jobs.Job) *sqs.SendMessageInput {
-	msg := &sqs.SendMessageInput{
+	return &sqs.SendMessageInput{
 		QueueUrl:     url,
 		DelaySeconds: aws.Int64(int64(j.Options.Delay)),
 		MessageBody:  aws.String(j.Payload),
@@ -31,8 +31,6 @@ func pack(url *string, j *jobs.Job) *sqs.SendMessageInput {
 			"rr-retryDelay":  {DataType: aws.String("Number"), StringValue: awsDuration(j.Options.RetryDuration())},
 		},
 	}
-
-	return msg
 }
 
 // unpack restores jobs.Options

go.mod

@@ -6,6 +6,8 @@ require (
 	github.com/aws/aws-sdk-go v1.16.14
 	github.com/beanstalkd/go-beanstalk v0.0.0-20180822062812-53ecdaa3bcfb
 	github.com/buger/goterm v0.0.0-20181115115552-c206103e1f37
+	github.com/cenkalti/backoff v2.2.1+incompatible // indirect
+	github.com/cenkalti/backoff/v4 v4.0.0
 	github.com/coreos/go-etcd v2.0.0+incompatible // indirect
 	github.com/cpuguy83/go-md2man v1.0.10 // indirect
 	github.com/dustin/go-humanize v1.0.0