Improve backend error tolerance (#33)

* Update SQS test to reflect resizable concurrency pool signature

* Improve SQS broker reliability when backend returns errors

Originally, the SQS broker created a goroutine with an infinite loop. That
goroutine would poll SQS and put any received messages on a channel. A separate
goroutine would poll that channel, spawning a goroutine for any received items.
A separate channel reported errors between these different goroutines. In
theory, all the goroutines would listen for channel closure and error channel
items to handle errors. Unfortunately, if the backend reports errors then
polling the queue would quickly stop, because in this case the currency pool
wasn't properly returned.

To address this, the multiple goroutines are collapsed. When the initial
pollying goroutine receives a message, it immediately spawns a goroutine to
process that message, rather than using a channel and intermediate goroutine.
In addition, it always returns to the concurrency pool immediately after either
encountering an error or processing a message. This allows polling and
processing to continue even if the backend reports an error.

Author: mwudka

v1/brokers/sqs/sqs.go

@@ -66,48 +66,57 @@ func (b *Broker) StartConsuming(consumerTag string, concurrency iface.Resizeable
 	//save it so that it can be used later when attempting to delete task
 	b.queueUrl = qURL
 
-	deliveries := make(chan *awssqs.ReceiveMessageOutput)
-
 	b.stopReceivingChan = make(chan int)
 	b.receivingWG.Add(1)
 
-	go func() {
-		defer b.receivingWG.Done()
-
-		log.INFO.Printf("[*] Waiting for messages on queue: %s. To exit press CTRL+C\n", *qURL)
-
-		pool := concurrency.Pool()
+	defer b.receivingWG.Done()
 
-		for {
-			select {
-			// A way to stop this goroutine from b.StopConsuming
-			case <-b.stopReceivingChan:
-				close(deliveries)
+	log.INFO.Printf("[*] Waiting for messages on queue: %s. To exit press CTRL+C\n", *qURL)
 
-				return
-			case <-pool:
-				output, err := b.receiveMessage(qURL)
-				if err == nil && len(output.Messages) > 0 {
-					deliveries <- output
+	pool := concurrency.Pool()
 
-				} else {
-					if err != nil {
-						log.ERROR.Printf("Queue consume error on %s: %s", *qURL, err)
+	errorsChan := make(chan error)
+	defer func() {
+		// There could be an outstanding consumeOne call in a goroutine that will return an error that
+		// get put on the channel, so make sure there are no oustanding consumeOne calls before closing the channel
+		b.processingWG.Wait()
+		close(errorsChan)
+	}()
 
-						// Avoid repeating this
-						if strings.Contains(err.Error(), "AWS.SimpleQueueService.NonExistentQueue") {
-							time.Sleep(30 * time.Second)
-						}
+	for {
+		select {
+		case workerError := <-errorsChan:
+			return b.GetRetry(), workerError
+		// A way to stop this goroutine from b.StopConsuming
+		case <-b.stopReceivingChan:
+			// If someone called b.stopReceivingChannel, they are trying to shut down the process
+			// so we don't want to retry the StartConsuming call
+			return false, nil
+		case <-pool:
+			output, err := b.receiveMessage(qURL)
+			if err == nil && len(output.Messages) > 0 {
+				b.processingWG.Add(1)
+				go func() {
+					consumeError := b.consumeOne(output, taskProcessor)
+					if consumeError != nil {
+						errorsChan <- consumeError
 					}
-					//return back to pool right away
 					concurrency.Return()
+					b.processingWG.Done()
+				}()
+			} else {
+				if err != nil {
+					log.ERROR.Printf("Queue consume error on %s: %s", *qURL, err)
+
+					// Avoid repeating this
+					if strings.Contains(err.Error(), "AWS.SimpleQueueService.NonExistentQueue") {
+						time.Sleep(30 * time.Second)
+					}
 				}
+				//return back to pool right away
+				concurrency.Return()
 			}
 		}
-	}()
-
-	if err := b.consume(deliveries, taskProcessor, concurrency); err != nil {
-		return b.GetRetry(), err
 	}
 
 	return b.GetRetry(), nil
@@ -231,22 +240,6 @@ func restrictVisibilityTimeoutDelay(delay time.Duration, receivedAt time.Time) t
 	return delay
 }
 
-// consume is a method which keeps consuming deliveries from a channel, until there is an error or a stop signal
-func (b *Broker) consume(deliveries <-chan *awssqs.ReceiveMessageOutput, taskProcessor iface.TaskProcessor, concurrency iface.ResizeablePool) error {
-
-	errorsChan := make(chan error)
-
-	for {
-		whetherContinue, err := b.consumeDeliveries(deliveries, taskProcessor, concurrency, errorsChan)
-		if err != nil {
-			return err
-		}
-		if whetherContinue == false {
-			return nil
-		}
-	}
-}
-
 // consumeOne is a method consumes a delivery. If a delivery was consumed successfully, it will be deleted from AWS SQS
 func (b *Broker) consumeOne(delivery *awssqs.ReceiveMessageOutput, taskProcessor iface.TaskProcessor) error {
 	if len(delivery.Messages) == 0 {
@@ -390,34 +383,6 @@ func (b *Broker) initializePool(pool chan struct{}, concurrency int) {
 	}
 }
 
-// consumeDeliveries is a method consuming deliveries from deliveries channel
-func (b *Broker) consumeDeliveries(deliveries <-chan *awssqs.ReceiveMessageOutput, taskProcessor iface.TaskProcessor, concurrency iface.ResizeablePool, errorsChan chan error) (bool, error) {
-	select {
-	case err := <-errorsChan:
-		return false, err
-	case d := <-deliveries:
-
-		b.processingWG.Add(1)
-
-		// Consume the task inside a goroutine so multiple tasks
-		// can be processed concurrently
-		go func() {
-
-			if err := b.consumeOne(d, taskProcessor); err != nil {
-				errorsChan <- err
-			}
-
-			b.processingWG.Done()
-
-			// give worker back to pool
-			concurrency.Return()
-		}()
-	case <-b.GetStopChan():
-		return false, nil
-	}
-	return true, nil
-}
-
 // continueReceivingMessages is a method returns a continue signal
 func (b *Broker) continueReceivingMessages(qURL *string, deliveries chan *awssqs.ReceiveMessageOutput) (bool, error) {
 	select {
@@ -432,7 +397,9 @@ func (b *Broker) continueReceivingMessages(qURL *string, deliveries chan *awssqs
 		if len(output.Messages) == 0 {
 			return true, nil
 		}
-		go func() { deliveries <- output }()
+		go func() {
+			deliveries <- output
+		}()
 	}
 	return true, nil
 }

v1/brokers/sqs/sqs_export_test.go

@@ -147,7 +147,8 @@ func NewTestErrorBroker() *Broker {
 }
 
 func (b *Broker) ConsumeForTest(deliveries <-chan *awssqs.ReceiveMessageOutput, taskProcessor iface.TaskProcessor, concurrency iface.ResizeablePool) error {
-	return b.consume(deliveries, taskProcessor, concurrency)
+	panic("not implemented")
+	return nil
 }
 
 func (b *Broker) ConsumeOneForTest(delivery *awssqs.ReceiveMessageOutput, taskProcessor iface.TaskProcessor) error {
@@ -171,11 +172,13 @@ func (b *Broker) InitializePoolForTest(pool chan struct{}, concurrency int) {
 }
 
 func (b *Broker) ConsumeDeliveriesForTest(deliveries <-chan *awssqs.ReceiveMessageOutput, taskProcessor iface.TaskProcessor, concurrency iface.ResizeablePool, errorsChan chan error) (bool, error) {
-	return b.consumeDeliveries(deliveries, taskProcessor, concurrency, errorsChan)
+	panic("not implemented")
+	return false, nil
 }
 
 func (b *Broker) ContinueReceivingMessagesForTest(qURL *string, deliveries chan *awssqs.ReceiveMessageOutput) (bool, error) {
-	return b.continueReceivingMessages(qURL, deliveries)
+	panic("not implemented")
+	return false, nil
 }
 
 func (b *Broker) StopReceivingForTest() {

v1/brokers/sqs/sqs_test.go

@@ -102,7 +102,7 @@ func TestPrivateFunc_consume(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	concurrency := common.NewResizablePool(1)
+	concurrency, _ := common.NewResizablePool(1)
 	wk := server1.NewWorker("sms_worker", 0)
 	deliveries := make(chan *awssqs.ReceiveMessageOutput)
 	outputCopy := *receiveMessageOutput
@@ -204,7 +204,7 @@ func TestPrivateFunc_receiveMessage(t *testing.T) {
 
 func TestPrivateFunc_consumeDeliveries(t *testing.T) {
 
-	concurrency := common.NewResizablePool(0)
+	concurrency, _ := common.NewResizablePool(0)
 	errorsChan := make(chan error)
 	deliveries := make(chan *awssqs.ReceiveMessageOutput)
 	server1, err := machinery.NewServer(cnf)
@@ -319,7 +319,7 @@ func TestPrivateFunc_consumeWithConcurrency(t *testing.T) {
 	broker.SetRegisteredTaskNames([]string{"test-task"})
 	assert.NoError(t, err)
 
-	concurrency := common.NewResizablePool(1)
+	concurrency, _ := common.NewResizablePool(1)
 	wk := server1.NewWorker("sms_worker", 1)
 	deliveries := make(chan *awssqs.ReceiveMessageOutput)
 	outputCopy := *receiveMessageOutput