SQS weird behaviour with inflight messages

[asger-noer]

Issue

I experience a problem on the queue that the code below is consuming. I have a producer that creates 8 notifications on SNS with the queue as a destination (AWS console reports the 8 messages publish). Out of the 8 messages I receive 6 (logged right after receiving function). The AWS console reports 2 messages in flight after the other messages have been processed. They never return to the queue, even after the visibility timeout. They are not moved to the DLQ either.

Janky solution

I've experienced that the addition of a sleep of 50ms after deletion resulted in the code working as expected.

time.Sleep(time.Millisecond * 50)

Queue

As per the dashboard below, you can see the publishing of 8 messages, but only deletion of 6.

Stats from the dashboard

Code

type HandlerFunc func(msg types.Message) error

type Queue interface {
	Process(ctx context.Context, handler HandlerFunc)
	Push(ctx context.Context, MessageAttributes MsgAttr, body string) (string, error)
}

type queue struct {
	client            *sqs.Client
	queueName         string
	queueURL          *string
	msgCount          int32
	waitTimeSeconds   int32
	visibilityTimeout int32
}

// NewQueue returns an interface for processing the underlying queue.
func NewQueue(ctx context.Context, queueName string) (Queue, error) {
	ctx, cancel := context.WithTimeout(ctx, time.Second*5)
	defer cancel()

	cfg, err := config.LoadDefaultConfig(ctx,
		config.WithDefaultRegion("eu-central-1"),
	)

	if err != nil {
		return nil, errors.Wrap(err, "loading default config")
	}

	svc := sqs.NewFromConfig(cfg)

	input := &sqs.GetQueueUrlInput{
		QueueName: aws.String(queueName),
	}

	q, err := svc.GetQueueUrl(ctx, input)
	if err != nil {
		return nil, errors.Wrap(err, "finding queue url")
	}

	return &queue{
		client:            svc,
		queueName:         queueName,
		queueURL:          q.QueueUrl,
		msgCount:          1,
		waitTimeSeconds:   10,
		visibilityTimeout: 120,
	}, nil
}


// Process will spawn a goroutine continuesly fetching and processing messages
// from a queue.
func (q *queue) Process(ctx context.Context, handler HandlerFunc) {
	go func() {
		for {
			params := &sqs.ReceiveMessageInput{
				QueueUrl:              q.queueURL,
				MaxNumberOfMessages:   q.msgCount,
				VisibilityTimeout:     q.visibilityTimeout,
				WaitTimeSeconds:       q.waitTimeSeconds,
				MessageAttributeNames: []string{string(types.QueueAttributeNameAll)},
			}

			res, err := q.client.ReceiveMessage(ctx, params)

			log.Err(err).Str("queue", q.queueName).
				Int("message_count", len(res.Messages)).
				Msg("receiving batch messages on SQS")

			for _, msg := range res.Messages {
				// When message is receivec process it with the handler
				err := handler(msg)
				log.Err(err).Str("message_id", *msg.MessageId).Msg("message handled")
				if err != nil {
					// if error is found, keep the message in the queue.
					continue
				}

				params := &sqs.DeleteMessageInput{
					QueueUrl:      q.queueURL,
					ReceiptHandle: msg.ReceiptHandle,
				}

				_, err = q.client.DeleteMessage(ctx, params)
				log.Err(err).Str("message_id", *msg.MessageId).Msg("message deleted")
				if err != nil {
					// if error is found, keep the message in the queue.
					continue
				}
			}
		}
	}()
}

[github-actions[bot]]

Hello! Reopening this discussion to make it searchable.