No backoff/retry for delete

[EtherZa]

Describe the bug

On a successful handler invocation, the delete operation can still fail due to throttling (or other HTTP error). This results in an otherwise successful message being reprocessed/sent to the DLQ.

Failed to delete message 00000000-0000-0000-0000-000000000000 from queue https://sqs.ap-southeast-2.amazonaws.com/XXXX/XXX.fifo: Request is throttled

Regression Issue

• Select this option if this issue appears to be a regression.

Expected Behavior

Use the IBackoffHandler in SQSMessagePoller.DeleteMessagesAsync to retry delete on a non-fatal exception.

Current Behavior

When API requests are being throttled/other HTTP exception, the request fails and the message is sent back to the queue to be processed again even though the handler has successfully completed invocation.

Reproduction Steps

Exceed the throttle cap on a SQS queue.
'delete' will start to fail, resulting in the same message being processed multiple times and eventually may be redirected to the DLQ.

Failed to delete message 00000000-0000-0000-0000-000000000000 from queue https://sqs.ap-southeast-2.amazonaws.com/XXXX/XXX.fifo: Request is throttled

Possible Solution

Wrap DeleteMessageBatchAsync() in a retry block. Optionally handle exceptions that are not fatal to the poller, but are fatal to the message (eg. already deleted, etc0.

internal class SQSMessagePoller : IMessagePoller, ISQSMessageCommunication
{
    public async Task DeleteMessagesAsync(IEnumerable<MessageEnvelope> messages, CancellationToken token = default)
    {
        ...
        // From:
        var response = await _sqsClient.DeleteMessageBatchAsync(request, token);

        // To:
        var response =
            await _backoffHandler.BackoffAsync(async () =>
            {
                DeleteMessageBatchResponse deleteMessageBatchResponse;
                try
                {
                    deleteMessageBatchResponse = await _sqsClient.DeleteMessageBatchAsync(request, token);
                }
                catch (TaskCanceledException)
                {
                    _logger.LogTrace("Cancellation requested while deleting messages, probably due to shutdown. Returning empty response");
                    return new DeleteMessageBatchResponse();
                }

                return deleteMessageBatchResponse;
            },
            // supply retry exception check instead of base configuration so that exceptions fatal to delete will also avoid a retry
            // ie. message not found/already deleted should not be retried or stop the SQS poller
            ex => _configuration.IsExceptionFatal(ex) || IsDeletedFatalException(ex),  
            token);
        ...
    }
}

Additional Information/Context

No response

AWS.Messaging (or related) package versions

Issue confirmed in source.

Targeted .NET Platform

net8.0

Operating System and version

n/a

[ashishdhingra]

@EtherZa Good afternoon. Thanks for opening the issue. I tried reproducing the issue using standard queue using code below:
.csproj

<Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
    <OutputType>Exe</OutputType>
    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="AWS.Messaging" Version="1.3.0" />
    <PackageReference Include="AWSSDK.SQS" Version="4.0.2.24" />
    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.5" />
    <PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.5" />
    <PackageReference Include="Microsoft.Extensions.Logging" Version="10.0.5" />
    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="10.0.5" />
  </ItemGroup>

</Project>

Program.cs

using Amazon.SQS;
using Amazon.SQS.Model;
using AWS.Messaging;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System.Collections.Concurrent;

namespace SqsDeleteThrottlingRepro
{
    /// <summary>
    /// Test message type for reproduction
    /// </summary>
    public class TestMessage
    {
        public string MessageId { get; set; }
        public string Content { get; set; }
        public DateTime Timestamp { get; set; }
    }

    /// <summary>
    /// Message handler that always succeeds
    /// This simulates successful business logic processing
    /// </summary>
    public class SuccessfulTestMessageHandler : IMessageHandler<TestMessage>
    {
        private readonly ILogger<SuccessfulTestMessageHandler> _logger;
        private readonly ProcessingTracker _tracker;

        public SuccessfulTestMessageHandler(
            ILogger<SuccessfulTestMessageHandler> logger,
            ProcessingTracker tracker)
        {
            _logger = logger;
            _tracker = tracker;
        }

        public async Task<MessageProcessStatus> HandleAsync(
            MessageEnvelope<TestMessage> messageEnvelope,
            CancellationToken token = default)
        {
            var message = messageEnvelope.Message;

            // Track if this message has been processed before
            if (_tracker.HasBeenProcessed(message.MessageId))
            {
                _tracker.IncrementReprocessed();
                _logger.LogWarning(
                    "⚠ Message {MessageId} is being REPROCESSED! This indicates the delete failed.",
                    message.MessageId);
            }
            else
            {
                _tracker.MarkAsProcessed(message.MessageId);
            }

            // Simulate some processing work
            await Task.Delay(10, token);

            _tracker.IncrementProcessed();

            // Handler ALWAYS succeeds - this is key to reproducing the bug
            _logger.LogInformation(
                "✓ Successfully processed message {MessageId}: {Content}",
                message.MessageId,
                message.Content);

            return MessageProcessStatus.Success();
        }
    }

    /// <summary>
    /// Tracks processing statistics to identify the bug
    /// </summary>
    public class ProcessingTracker
    {
        private readonly ConcurrentDictionary<string, bool> _processedMessages = new();
        private int _processedCount = 0;
        private int _reprocessedCount = 0;

        public bool HasBeenProcessed(string messageId)
        {
            return _processedMessages.ContainsKey(messageId);
        }

        public void MarkAsProcessed(string messageId)
        {
            _processedMessages.TryAdd(messageId, true);
        }

        public void IncrementProcessed()
        {
            Interlocked.Increment(ref _processedCount);
        }

        public void IncrementReprocessed()
        {
            Interlocked.Increment(ref _reprocessedCount);
        }

        public (int Processed, int Reprocessed, int Unique) GetStats()
        {
            return (_processedCount, _reprocessedCount, _processedMessages.Count);
        }
    }

    /// <summary>
    /// Main reproduction program
    /// </summary>
    class Program
    {
        static async Task Main(string[] args)
        {
            Console.WriteLine("=== aws-dotnet-messaging Delete Throttling Bug Reproduction ===");
            Console.WriteLine("Issue #319: No backoff/retry for delete operations");
            Console.WriteLine();

            var queueUrl = Environment.GetEnvironmentVariable("SQS_QUEUE_URL")
                ?? throw new InvalidOperationException("SQS_QUEUE_URL environment variable is required");

            var messageCount = int.Parse(
                Environment.GetEnvironmentVariable("MESSAGE_COUNT") ?? "500");

            Console.WriteLine($"Queue URL: {queueUrl}");
            Console.WriteLine($"Messages to send: {messageCount}");
            Console.WriteLine();

            var tracker = new ProcessingTracker();

            // Build the host but don't start it yet
            var host = CreateMessageProcessingHost(queueUrl, tracker);

            // Step 1: Start the host to get the publisher service
            Console.WriteLine("Step 1: Initializing message publisher...");
            await host.StartAsync();

            // Wait for services to initialize
            await Task.Delay(2000);

            // Get the message publisher from DI
            var publisher = host.Services.GetRequiredService<IMessagePublisher>();

            // Step 2: Publish all messages BEFORE processing starts
            Console.WriteLine("Step 2: Publishing all test messages...");

            await SendTestMessagesAsync(publisher, messageCount);
            Console.WriteLine($"✓ All {messageCount} messages published");

            // Give SQS a moment to make messages available
            Console.WriteLine("Waiting for messages to be available in queue...");

            await Task.Delay(5000);

            // Step 3: NOW start processing with high concurrency
            Console.WriteLine("Step 3: Starting message processing with high concurrency...");

            Console.WriteLine("(This will trigger the bug when delete operations are throttled)");
            Console.WriteLine();

            // Let processing run for a set duration
            var processingDuration = TimeSpan.FromMinutes(3);
            Console.WriteLine($"Processing for {processingDuration.TotalMinutes} minutes...");


            await Task.Delay(processingDuration);

            // Step 4: Stop processing
            Console.WriteLine("⏱ Processing time limit reached.Stopping...");

            await host.StopAsync();

            // Step 5: Display results
            Console.WriteLine("=== Results === ");

            var (processed, reprocessed, unique) = tracker.GetStats();

            Console.WriteLine($"Total processing attempts: {processed}");
            Console.WriteLine($"Unique messages processed: {unique}");
            Console.WriteLine($"Messages reprocessed: {reprocessed}");

            // Check DLQ if configured
            var dlqUrl = Environment.GetEnvironmentVariable("DLQ_URL");
            if (!string.IsNullOrEmpty(dlqUrl))
            {
                var dlqCount = await GetDlqMessageCountAsync(dlqUrl);
                Console.WriteLine($"Messages in DLQ: {dlqCount}");

                if (dlqCount > 0)
                {
                    Console.WriteLine($"⚠ WARNING: {dlqCount} successfully processed messages ended up in the DLQ!");
                }
            }

            if (reprocessed > 0)
            {
                Console.WriteLine("✓ BUG REPRODUCED!");

                Console.WriteLine($"  {reprocessed} messages were reprocessed despite successful handling");
                Console.WriteLine("  This indicates delete operations failed (likely due to throttling)");
                Console.WriteLine("  and the SQSMessagePoller.DeleteMessagesAsync does NOT retry with backoff");
            }
            else
            {
                Console.WriteLine("⚠ Bug not reproduced - no reprocessing detected");

                Console.WriteLine("  Try:");
                Console.WriteLine("  - Increasing MESSAGE_COUNT (e.g., 1000+)");
                Console.WriteLine("  - Increasing MaxNumberOfConcurrentMessages (e.g., 100)");
                Console.WriteLine("  - Running multiple test instances simultaneously");
            }

            Console.WriteLine("        Press any key to exit...");

            Console.ReadKey();
        }

        /// <summary>
        /// Create the host with aws-dotnet-messaging configured
        /// </summary>
        static IHost CreateMessageProcessingHost(
            string queueUrl,
            ProcessingTracker tracker)
        {
            return Host.CreateDefaultBuilder()
                .ConfigureServices((context, services) =>
                {
                    // Register the processing tracker
                    services.AddSingleton(tracker);

                    // Configure AWS Messaging with SQS
                    services.AddAWSMessageBus(builder =>
                    {
                        // Configure the message publisher
                        builder.AddSQSPublisher<TestMessage>(queueUrl);

                        // Add SQS poller with high concurrency to trigger throttling
                        builder.AddSQSPoller(queueUrl, options =>
                        {
                            // Very high concurrency to trigger delete throttling
                            options.MaxNumberOfConcurrentMessages = 50;
                        });

                        // Register the message handler
                        builder.AddMessageHandler<SuccessfulTestMessageHandler, TestMessage>();
                    });

                    // Configure logging to see what's happening
                    services.AddLogging(logging =>
                    {
                        logging.AddConsole();
                        logging.SetMinimumLevel(LogLevel.Information);

                        // Increase log level for AWS.Messaging to see delete failures
                        logging.AddFilter("AWS.Messaging", LogLevel.Debug);
                    });
                })
                .Build();
        }

        /// <summary>
        /// Check how many messages ended up in the DLQ
        /// </summary>
        static async Task<int> GetDlqMessageCountAsync(string dlqUrl)
        {
            var sqsClient = new AmazonSQSClient();

            var response = await sqsClient.GetQueueAttributesAsync(new GetQueueAttributesRequest
            {
                QueueUrl = dlqUrl,
                AttributeNames = new List<string> { "ApproximateNumberOfMessages" }
            });

            if (response.Attributes.TryGetValue("ApproximateNumberOfMessages", out var count))
            {
                return int.Parse(count);
            }

            return 0;
        }

        /// <summary>
        /// Send test messages using AWS.Messaging publisher
        /// </summary>
        static async Task SendTestMessagesAsync(IMessagePublisher publisher, int count)
        {
            Console.WriteLine($"Sending {count} messages...");

            for (int i = 0; i < count; i++)
            {
                var message = new TestMessage
                {
                    MessageId = Guid.NewGuid().ToString(),
                    Content = $"Test message {i}",
                    Timestamp = DateTime.UtcNow
                };

                await publisher.PublishAsync(message);

                if ((i + 1) % 10 == 0)
                {
                    Console.Write(".");
                }
            }

            Console.WriteLine();
        }
    }
}

But the issue is not reproducible, perhaps due to high SQS throttling limit.

Do you happen to use FIFO or standard queue and if you have consistent reproduction in your setup?

Needs review with the team.

Thanks,
Ashish

[EtherZa]

Hi @ashishdhingra. Thank you for your response.

I am seeing this issue with FIFO queues (high throughput mode is on) where message ids are predominantly unique, but where a large number of messages are being ingested concurrently. When this occurs, the issue is fairly consistent/predictable.

The below queue has a single poller instance configured to handle a maximum of 20 concurrent messages. It is also ingesting messages in bulk which is adding to the API load.

[ashishdhingra]

@EtherZa Thanks for your input. Somehow, I'm unable to reproduce with below approach.
Created FIFO and DLQ using AWS CLI

# Create FIFO queue
aws sqs create-queue \
  --queue-name throttle-test-queue.fifo \
  --attributes FifoQueue=true,ContentBasedDeduplication=true \
  --region us-east-2

# Create FIFO DLQ
aws sqs create-queue \
  --queue-name throttle-test-dlq.fifo \
  --attributes FifoQueue=true,ContentBasedDeduplication=true \
  --region us-east-2

# Get DLQ ARN
aws sqs get-queue-attributes \
  --queue-url https://sqs.us-east-2.amazonaws.com/YOUR_ACCOUNT/throttle-test-dlq.fifo \
  --attribute-names QueueArn \
  --region us-east-2

# Configure redrive policy (replace YOUR_ACCOUNT with your AWS account ID)
aws sqs set-queue-attributes \
  --queue-url https://sqs.us-east-2.amazonaws.com/YOUR_ACCOUNT/throttle-test-queue.fifo \
  --attributes '{
    "RedrivePolicy": "{\"deadLetterTargetArn\":\"arn:aws:sqs:us-east-2:YOUR_ACCOUNT:throttle-test-dlq.fifo\",\"maxReceiveCount\":\"3\"}"
  }' \
  --region us-east-2

Sample C# test harness
Program.cs

using Amazon.SQS;
using Amazon.SQS.Model;
using AWS.Messaging;
using AWS.Messaging.Publishers.SQS;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System.Collections.Concurrent;

namespace SqsDeleteThrottlingRepro
{
    /// <summary>
    /// Test message type for reproduction
    /// </summary>
    public class TestMessage
    {
        public string MessageId { get; set; }
        public string Content { get; set; }
        public DateTime Timestamp { get; set; }
    }

    /// <summary>
    /// Message handler that always succeeds
    /// This simulates successful business logic processing
    /// </summary>
    public class SuccessfulTestMessageHandler : IMessageHandler<TestMessage>
    {
        private readonly ILogger<SuccessfulTestMessageHandler> _logger;
        private readonly ProcessingTracker _tracker;

        public SuccessfulTestMessageHandler(
            ILogger<SuccessfulTestMessageHandler> logger,
            ProcessingTracker tracker)
        {
            _logger = logger;
            _tracker = tracker;
        }

        public async Task<MessageProcessStatus> HandleAsync(
            MessageEnvelope<TestMessage> messageEnvelope,
            CancellationToken token = default)
        {
            var message = messageEnvelope.Message;

            // Track if this message has been processed before
            if (_tracker.HasBeenProcessed(message.MessageId))
            {
                _tracker.IncrementReprocessed();
                _logger.LogWarning(
                    "⚠ Message {MessageId} is being REPROCESSED! This indicates the delete failed.",
                    message.MessageId);
            }
            else
            {
                _tracker.MarkAsProcessed(message.MessageId);
            }

            // Simulate some processing work (very fast to maximize delete rate)
            await Task.Delay(1, token);

            _tracker.IncrementProcessed();

            // Handler ALWAYS succeeds - this is key to reproducing the bug
            _logger.LogInformation(
                "✓ Successfully processed message {MessageId}: {Content}",
                message.MessageId,
                message.Content);

            return MessageProcessStatus.Success();
        }
    }

    /// <summary>
    /// Tracks processing statistics to identify the bug
    /// </summary>
    public class ProcessingTracker
    {
        private readonly ConcurrentDictionary<string, bool> _processedMessages = new();
        private int _processedCount = 0;
        private int _reprocessedCount = 0;

        public bool HasBeenProcessed(string messageId)
        {
            return _processedMessages.ContainsKey(messageId);
        }

        public void MarkAsProcessed(string messageId)
        {
            _processedMessages.TryAdd(messageId, true);
        }

        public void IncrementProcessed()
        {
            Interlocked.Increment(ref _processedCount);
        }

        public void IncrementReprocessed()
        {
            Interlocked.Increment(ref _reprocessedCount);
        }

        public (int Processed, int Reprocessed, int Unique) GetStats()
        {
            return (_processedCount, _reprocessedCount, _processedMessages.Count);
        }
    }

    /// <summary>
    /// Main reproduction program
    /// </summary>
    class Program
    {
        static async Task Main(string[] args)
        {
            Console.WriteLine("=== aws-dotnet-messaging Delete Throttling Bug Reproduction ===");
            Console.WriteLine("Issue #319: No backoff/retry for delete operations");
            Console.WriteLine("Optimized for FIFO Queue (300 TPS limit)");
            Console.WriteLine();

            var queueUrl = Environment.GetEnvironmentVariable("SQS_QUEUE_URL")
                ?? throw new InvalidOperationException("SQS_QUEUE_URL environment variable is required");

            var messageCount = int.Parse(
                Environment.GetEnvironmentVariable("MESSAGE_COUNT") ?? "1000");

            Console.WriteLine($"Region: us-east-2");
            Console.WriteLine($"Queue URL: {queueUrl}");
            Console.WriteLine($"Messages to send: {messageCount}");
            Console.WriteLine($"Concurrent messages: 20 (customer's configuration)");
            Console.WriteLine();

            var tracker = new ProcessingTracker();

            // Build the host
            var host = CreateMessageProcessingHost(queueUrl, tracker);

            // Step 1: Start the host to initialize services
            Console.WriteLine("Step 1: Initializing SQS publisher...");
            await host.StartAsync();

            // Wait for services to initialize
            await Task.Delay(2000);

            // Get ISQSPublisher from DI (not IMessagePublisher)
            var sqsPublisher = host.Services.GetRequiredService<ISQSPublisher>();

            // Step 2: Publish all messages BEFORE processing starts
            Console.WriteLine("Step 2: Publishing all test messages to FIFO queue...");

            await SendTestMessagesAsync(sqsPublisher, messageCount);
            Console.WriteLine($"✓ All {messageCount} messages published");

            // Give SQS a moment to make messages available
            Console.WriteLine("Waiting for messages to be available in queue...");

            await Task.Delay(5000);

            // Step 3: NOW start processing with high concurrency
            Console.WriteLine("Step 3: Starting message processing...");

            // Let processing run for a set duration
            var processingDuration = TimeSpan.FromMinutes(1);
            Console.WriteLine($"Processing for {processingDuration.TotalMinutes} minutes...");


            await Task.Delay(processingDuration);

            // Step 4: Stop processing
            Console.WriteLine("Processing time limit reached.Stopping...");

            await host.StopAsync();

            // Step 5: Display results
            Console.WriteLine("=== Results === ");

            var (processed, reprocessed, unique) = tracker.GetStats();

            Console.WriteLine($"Total processing attempts: {processed}");
            Console.WriteLine($"Unique messages processed: {unique}");
            Console.WriteLine($"Messages reprocessed: {reprocessed}");

            // Check DLQ if configured
            var dlqUrl = Environment.GetEnvironmentVariable("DLQ_URL");
            if (!string.IsNullOrEmpty(dlqUrl))
            {
                var dlqCount = await GetDlqMessageCountAsync(dlqUrl);
                Console.WriteLine($"Messages in DLQ: {dlqCount}");

                if (dlqCount > 0)
                {
                    Console.WriteLine($"⚠ CRITICAL: {dlqCount} successfully processed messages ended up in the DLQ!");
                }
            }

            if (reprocessed > 0)
            {
                Console.WriteLine("✓ BUG REPRODUCED!");

                Console.WriteLine($"  {reprocessed} messages were reprocessed despite successful handling");
                Console.WriteLine("  Root cause: Delete operations failed and were NOT retried with backoff");
            }
            else
            {
                Console.WriteLine("⚠ Bug not reproduced - try running multiple instances simultaneously");
            }

            Console.WriteLine("Press any key to exit...");
            Console.ReadKey();
        }

        /// <summary>
        /// Create the host with aws-dotnet-messaging configured for FIFO queue
        /// </summary>
        static IHost CreateMessageProcessingHost(
        string queueUrl,
        ProcessingTracker tracker)
        {
            return Host.CreateDefaultBuilder()
                .ConfigureServices((context, services) =>
                {
                    // Register the processing tracker
                    services.AddSingleton(tracker);

                    // Configure AWS Messaging with SQS
                    services.AddAWSMessageBus(builder =>
                    {
                        // Configure the message publisher for FIFO queue
                        builder.AddSQSPublisher<TestMessage>(queueUrl);

                        // Add SQS poller with very high concurrency
                        // This will cause delete operations to exceed FIFO's 300 TPS limit
                        builder.AddSQSPoller(queueUrl, options =>
                        {
                            // Very high concurrency to overwhelm FIFO's 300 TPS limit
                            // With 100 concurrent messages processing in ~10ms each,
                            // we'll generate ~10,000 delete operations per second
                            // This is 33x over the 300 TPS FIFO limit!

                            // Match customer's configuration: 20 concurrent messages
                            options.MaxNumberOfConcurrentMessages = 100;
                        });

                        // Register the message handler
                        builder.AddMessageHandler<SuccessfulTestMessageHandler, TestMessage>();
                    });

                    // Configure logging to see what's happening
                    services.AddLogging(logging =>
                    {
                        logging.AddConsole();
                        logging.SetMinimumLevel(LogLevel.Information);

                        // Increase log level for AWS.Messaging to see delete failures
                        logging.AddFilter("AWS.Messaging", LogLevel.Debug);
                    });
                })
                .Build();
        }

        /// <summary>
        /// Send test messages to FIFO queue
        /// </summary>
        /// <summary>
        /// Send test messages to FIFO queue using ISQSPublisher
        /// </summary>
        static async Task SendTestMessagesAsync(ISQSPublisher sqsPublisher, int count)
        {
            Console.WriteLine($"Sending {count} messages to FIFO queue...");

            // Use multiple message groups to allow parallel processing
            // FIFO queues process messages in order within a group,
            // but different groups can be processed in parallel
            var messageGroups = new[]
            {
                "group-1", "group-2", "group-3", "group-4", "group-5",
                "group-6", "group-7", "group-8", "group-9", "group-10"
            };

            for (int i = 0; i < count; i++)
            {
                var message = new TestMessage
                {
                    MessageId = Guid.NewGuid().ToString(),
                    Content = $"Test message {i}",
                    Timestamp = DateTime.UtcNow
                };

                // Distribute messages across multiple groups for parallel processing
                var messageGroupId = messageGroups[i % messageGroups.Length];

                // Use ISQSPublisher.SendAsync with SQSOptions for FIFO attributes
                await sqsPublisher.SendAsync(message, new SQSOptions
                {
                    MessageGroupId = messageGroupId,
                    MessageDeduplicationId = Guid.NewGuid().ToString()
                });

                if ((i + 1) % 10 == 0)
                {
                    Console.Write(".");
                }
            }

            Console.WriteLine();
        }


        /// <summary>
        /// Check how many messages ended up in the DLQ
        /// </summary>
        static async Task<int> GetDlqMessageCountAsync(string dlqUrl)
        {
            var sqsClient = new AmazonSQSClient(new AmazonSQSConfig
            {
                RegionEndpoint = Amazon.RegionEndpoint.USEast2
            });

            var response = await sqsClient.GetQueueAttributesAsync(new GetQueueAttributesRequest
            {
                QueueUrl = dlqUrl,
                AttributeNames = new List<string> { "ApproximateNumberOfMessages" }
            });

            if (response.Attributes.TryGetValue("ApproximateNumberOfMessages", out var count))
            {
                return int.Parse(count);
            }

            return 0;
        }
    }
}

.csproj

<Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
    <OutputType>Exe</OutputType>
    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="AWS.Messaging" Version="1.3.0" />
    <PackageReference Include="AWSSDK.SQS" Version="4.0.2.24" />
    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.5" />
    <PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.5" />
    <PackageReference Include="Microsoft.Extensions.Logging" Version="10.0.5" />
    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="10.0.5" />
  </ItemGroup>

</Project>

launchSettings.json

{
  "profiles": {
    "SqsDeleteThrottlingRepro": {
      "commandName": "Project",
      "environmentVariables": {
        "SQS_QUEUE_URL": "https://sqs.us-east-2.amazonaws.com/<<ACCOUNTID>>/throttle-test-queue.fifo",
        "AWS_REGION": "us-east-2",
        "MESSAGE_COUNT": "2000",
        "MAX_CONCURRENT_MESSAGES": "100",
        "DLQ_URL": "https://sqs.us-east-2.amazonaws.com/<<ACCOUNTID>>/throttle-test-dlq.fifo"
      }
    }
  }
}

The issue needs investigation as it might happen under high workload as per your report.

Thanks,
Ashish

[EtherZa]

Hi @ashishdhingra.

The reason that you are not reproducing the issue is that the messages are being trickled into the queue. If you seed the queue prior to starting the host, there will be enough messages available to satisfy the poller's MaxNumberOfConcurrentMessages configuration (publishing while consuming will never reach a full load).

ie.

// Build the host
var host = CreateMessageProcessingHost(queueUrl, tracker);


// <----  seed the queue before starting the consumers ---->
var sqsPublisher = host.Services.GetRequiredService<ISQSPublisher>();
await SendTestMessagesAsync(sqsPublisher, messageCount);


// Step 1: Start the host to initialize services
Console.WriteLine("Step 1: Initializing SQS publisher...");
await host.StartAsync();

//// Wait for services to initialize
//await Task.Delay(2000);

//// Get ISQSPublisher from DI (not IMessagePublisher)
//var sqsPublisher = host.Services.GetRequiredService<ISQSPublisher>();

//// Step 2: Publish all messages BEFORE processing starts
//Console.WriteLine("Step 2: Publishing all test messages to FIFO queue...");

//await SendTestMessagesAsync(sqsPublisher, messageCount);
//Console.WriteLine($"✓ All {messageCount} messages published");

//// Give SQS a moment to make messages available
//Console.WriteLine("Waiting for messages to be available in queue...");

//await Task.Delay(5000);

// Step 3: NOW start processing with high concurrency
Console.WriteLine("Step 3: Starting message processing...");

// Let processing run for a set duration
var processingDuration = TimeSpan.FromMinutes(1);
Console.WriteLine($"Processing for {processingDuration.TotalMinutes} minutes...");

For my setup, I have found that running the code with the debugger/instrumentation attached adds enough of a delay for the issue not to be reproduced but running without it raises the issue every time.

(only the region and above change have been made; everything else is as you left it)

[EtherZa]

Having a DeleteMessage per message also exacerbates the issue. For high load systems it would be preferrable to batch deletes to reduce the API load ie. wait for 10 deletes or [configurable] ms before posting a batch DeleteMessage.

As it is right now; each ReceiveMessage of 10 messages, results in 10 DeleteMessages which drives up the service cost and API count by 550% (1 read + 10 delete vs 1 read + 1 bulk delete).

Edit: Added feature request - #327

[ashishdhingra]

Created consolidated backlog item for #319 and #327