Service Bus: GroupId as SessionId, session receive, safe delayed retry, fast emulator drain

README.md

@@ -58,7 +58,6 @@ await messageBus.PublishAsync(new MyMessage { Data = "Hello" });
 | [AWS (S3/SQS/SNS)](https://github.com/FoundatioFx/Foundatio.AWS) | | ✅ | ✅ | ✅ | |
 | [RabbitMQ](https://github.com/FoundatioFx/Foundatio.RabbitMQ) | | | ✅ | | |
 | [Kafka](https://github.com/FoundatioFx/Foundatio.Kafka) | | | ✅ | | |
-| [Minio](https://github.com/FoundatioFx/Foundatio.Minio) | | | | ✅ | |
 | [Aliyun](https://github.com/FoundatioFx/Foundatio.Aliyun) | | | | ✅ | |
 | [SFTP](https://github.com/FoundatioFx/Foundatio.Storage.SshNet) | | | | ✅ | |
 

ServiceBus-Emulator/Config/Config.json

@@ -12,6 +12,16 @@
               "DefaultMessageTimeToLive": "PT1H",
               "DeadLetteringOnMessageExpiration": true
             }
+          },
+          {
+            "Name": "foundatio-test-queue-sessions",
+            "Properties": {
+              "LockDuration": "PT1M",
+              "MaxDeliveryCount": 5,
+              "DefaultMessageTimeToLive": "PT1H",
+              "DeadLetteringOnMessageExpiration": true,
+              "RequiresSession": true
+            }
           }
         ],
         "Topics": [

src/Foundatio.AzureServiceBus/Foundatio.AzureServiceBus.csproj

@@ -4,7 +4,7 @@
     <PackageReference Include="Azure.Messaging.ServiceBus" Version="7.21.0" />
     <PackageReference Include="Azure.Identity" Version="1.21.0" />
 
-    <PackageReference Include="Foundatio" Version="13.0.4" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
+    <PackageReference Include="Foundatio" Version="13.0.5-preview.0.10" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="..\..\..\Foundatio\src\Foundatio\Foundatio.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
   </ItemGroup>
 </Project>

src/Foundatio.AzureServiceBus/Queues/AzureServiceBusQueueEntry.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Linq;
 using Azure.Messaging.ServiceBus;
 using Foundatio.AzureServiceBus.Utility;
@@ -8,6 +9,11 @@ public class AzureServiceBusQueueEntry<T> : QueueEntry<T> where T : class
 {
     public ServiceBusReceivedMessage UnderlyingMessage { get; }
 
+    /// <summary>
+    /// The session receiver that holds this message's session on session-enabled queues; otherwise null.
+    /// </summary>
+    internal ServiceBusSessionReceiver? SessionReceiver { get; init; }
+
     public AzureServiceBusQueueEntry(ServiceBusReceivedMessage message, T value, IQueue<T> queue)
         : base(message.MessageId, message.CorrelationId, value, queue, message.EnqueuedTime.UtcDateTime, GetAttemptCount(message))
     {
@@ -21,17 +27,17 @@ public AzureServiceBusQueueEntry(ServiceBusReceivedMessage message, T value, IQu
         }
 
         UnderlyingMessage = message;
+        GroupId = String.IsNullOrEmpty(message.SessionId) ? null : message.SessionId;
     }
 
     /// <summary>
-    /// Gets the attempt count from the message. Uses the _attempts application property if available (for scheduled retries),
-    /// otherwise falls back to the DeliveryCount.
+    /// Gets the attempt count from the message. A scheduled retry copy stores the attempts of the original in the
+    /// <c>_attempts</c> application property; deliveries of the copy itself (lock expiry or abandon) add its DeliveryCount.
     /// </summary>
     private static int GetAttemptCount(ServiceBusReceivedMessage message)
     {
-        // Check if we have a stored attempt count from a scheduled retry
         if (message.ApplicationProperties.TryGetValue("_attempts", out object? attemptsValue) && attemptsValue is int storedAttempts)
-            return storedAttempts + 1; // Add 1 because this is a new delivery of that retry
+            return storedAttempts + message.DeliveryCount;
 
         // Fall back to delivery count for normal abandon/retry
         return message.DeliveryCount;

src/Foundatio.AzureServiceBus/Queues/AzureServiceBusQueueEntryOptions.cs

@@ -4,5 +4,8 @@ namespace Foundatio.AzureServiceBus.Queues;
 
 public record AzureServiceBusQueueEntryOptions : QueueEntryOptions
 {
+    /// <summary>
+    /// The Service Bus session id. Overrides <see cref="QueueEntryOptions.GroupId"/> when both are set.
+    /// </summary>
     public string? SessionId { get; set; }
 }

src/Foundatio.AzureServiceBus/Queues/AzureServiceBusQueueOptions.cs

@@ -103,7 +103,9 @@ public class AzureServiceBusQueueOptions<T> : SharedQueueOptions<T> where T : cl
     public string? UserMetadata { get; set; }
 
     /// <summary>
-    /// The function to calculate retry delay based on the attempt number.
+    /// The function to calculate retry delay based on the attempt number. A delayed retry is sent as a scheduled copy of the
+    /// message. It is not applied on session-enabled queues, where the copy would be processed after later messages in its
+    /// group, or on queues with duplicate detection, where the copy would be discarded; those queues retry immediately.
     /// </summary>
     public Func<int, TimeSpan> RetryDelay { get; set; } = attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)) + TimeSpan.FromMilliseconds(Random.Shared.Next(0, 100));
 }

tests/Directory.Build.props

@@ -12,7 +12,7 @@
     <PackageReference Include="xunit.runner.visualstudio" Version="4.0.0" />
     <PackageReference Include="GitHubActionsTestLogger" Version="3.0.5" PrivateAssets="All" />
 
-    <PackageReference Include="Foundatio.TestHarness" Version="13.0.4" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
+    <PackageReference Include="Foundatio.TestHarness" Version="13.0.5-preview.0.10" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="..\..\..\Foundatio\src\Foundatio.TestHarness\Foundatio.TestHarness.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
   </ItemGroup>
 </Project>

tests/Foundatio.AzureServiceBus.Tests/Queues/AzureServiceBusQueueTests.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Threading.Tasks;
+using Foundatio.AzureServiceBus.Queues;
 using Foundatio.Queues;
 using Foundatio.Serializer;
 using Foundatio.Tests.Queue;
@@ -40,7 +41,7 @@ public AzureServiceBusQueueTests(ITestOutputHelper output) : base(output)
             _assertStats = false;
     }
 
-    protected override IQueue<SimpleWorkItem>? GetQueue(int retries = 1, TimeSpan? workItemTimeout = null, TimeSpan? retryDelay = null, int[]? retryMultipliers = null, int deadLetterMaxItems = 100, bool runQueueMaintenance = true, TimeProvider? timeProvider = null, ISerializer? serializer = null)
+    protected override IQueue<SimpleWorkItem>? GetQueue(int retries = 1, TimeSpan? workItemTimeout = null, TimeSpan? retryDelay = null, int[]? retryMultipliers = null, int deadLetterMaxItems = 100, bool runQueueMaintenance = true, TimeProvider? timeProvider = null, ISerializer? serializer = null, Func<SimpleWorkItem, string?>? groupIdResolver = null)
     {
         string? connectionString = Configuration.GetConnectionString("AzureServiceBusConnectionString");
         if (String.IsNullOrEmpty(connectionString))
@@ -62,6 +63,9 @@ public AzureServiceBusQueueTests(ITestOutputHelper output) : base(output)
             if (retryDelay.HasValue)
                 o.RetryDelay(_ => retryDelay.Value);
 
+            if (groupIdResolver is not null)
+                o.GroupId(groupIdResolver);
+
             if (!_isEmulator)
             {
                 o.AutoDeleteOnIdle(TimeSpan.FromMinutes(5))
@@ -242,6 +246,18 @@ public override Task DequeueAsync_AfterAbandonWithMutatedValue_ReturnsOriginalVa
         return base.DequeueAsync_AfterAbandonWithMutatedValue_ReturnsOriginalValueAsync();
     }
 
+    [Fact]
+    public override Task AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync()
+    {
+        return base.AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync();
+    }
+
+    [Fact]
+    public override Task AbandonAsync_WithGroupIdAndRetryDelay_PreservesGroupIdOnRetryAsync()
+    {
+        return base.AbandonAsync_WithGroupIdAndRetryDelay_PreservesGroupIdOnRetryAsync();
+    }
+
     [Fact]
     public override Task DequeueAsync_WithDispose_AutoAbandonsEntryAsync()
     {
@@ -318,12 +334,171 @@ public override Task DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync(
         return base.DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync();
     }
 
+    [Fact]
+    public override Task EnqueueAsync_WhenEnqueuingHandlerClearsGroupId_EnqueuesWithoutGroupAsync()
+    {
+        return base.EnqueueAsync_WhenEnqueuingHandlerClearsGroupId_EnqueuesWithoutGroupAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithEmptyGroupId_EnqueuesWithoutGroupAsync()
+    {
+        return base.EnqueueAsync_WithEmptyGroupId_EnqueuesWithoutGroupAsync();
+    }
+
+    [Fact]
+    public async Task EnqueueAsync_WithGroupId_DoesNotClaimGroupedDeliveryAsync()
+    {
+        // Arrange: the queue is not session-enabled and cannot receive from sessions, so GroupId cannot affect delivery
+        using var queue = GetQueue();
+        if (queue is null)
+            return;
+
+        try
+        {
+            await queue.DeleteQueueAsync();
+            await AssertEmptyQueueAsync(queue);
+
+            // Act
+            await queue.EnqueueAsync(new SimpleWorkItem { Data = "no-grouped-delivery" }, new QueueEntryOptions { GroupId = "tenant-123" });
+
+            // Assert
+            Assert.Contains(Log.LogEntries, e => e.LogLevel == LogLevel.Debug && e.Message.Contains("delivery order or fairness"));
+            var entry = await queue.DequeueAsync(TimeSpan.FromSeconds(5));
+            Assert.NotNull(entry);
+            await entry.CompleteAsync();
+        }
+        finally
+        {
+            await CleanupQueueAsync(queue);
+        }
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync()
+    {
+        return base.EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync();
+    }
+
+    [Fact]
+    public async Task EnqueueAsync_WithGroupId_SendsGroupIdAsSessionIdAsync()
+    {
+        // Arrange
+        using var queue = GetQueue();
+        if (queue is null)
+            return;
+
+        try
+        {
+            await queue.DeleteQueueAsync();
+            await AssertEmptyQueueAsync(queue);
+
+            // Act
+            await queue.EnqueueAsync(new SimpleWorkItem { Data = "session-id-test" }, new QueueEntryOptions { GroupId = "tenant-123" });
+
+            // Assert
+            var entry = await queue.DequeueAsync(TimeSpan.FromSeconds(5));
+            var serviceBusEntry = Assert.IsType<AzureServiceBusQueueEntry<SimpleWorkItem>>(entry);
+            Assert.Equal("tenant-123", serviceBusEntry.UnderlyingMessage.SessionId);
+            Assert.Equal("tenant-123", serviceBusEntry.GroupId);
+            await serviceBusEntry.CompleteAsync();
+        }
+        finally
+        {
+            await CleanupQueueAsync(queue);
+        }
+    }
+
     [Fact]
     public override Task EnqueueAsync_WithSerializationError_ThrowsAndLeavesQueueEmptyAsync()
     {
         return base.EnqueueAsync_WithSerializationError_ThrowsAndLeavesQueueEmptyAsync();
     }
 
+    [Fact]
+    public async Task EnqueueAsync_WithSessionIdAndGroupId_UsesSessionIdAsync()
+    {
+        // Arrange
+        using var queue = GetQueue();
+        if (queue is null)
+            return;
+
+        try
+        {
+            await queue.DeleteQueueAsync();
+            await AssertEmptyQueueAsync(queue);
+
+            // Act
+            await queue.EnqueueAsync(new SimpleWorkItem { Data = "session-id-test" },
+                new AzureServiceBusQueueEntryOptions { GroupId = "tenant-123", SessionId = "session-456" });
+
+            // Assert
+            var entry = await queue.DequeueAsync(TimeSpan.FromSeconds(5));
+            var serviceBusEntry = Assert.IsType<AzureServiceBusQueueEntry<SimpleWorkItem>>(entry);
+            Assert.Equal("session-456", serviceBusEntry.UnderlyingMessage.SessionId);
+            Assert.Equal("session-456", serviceBusEntry.GroupId);
+            await serviceBusEntry.CompleteAsync();
+        }
+        finally
+        {
+            await CleanupQueueAsync(queue);
+        }
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithReusedOptions_DoesNotChangeCallerOptionsAsync()
+    {
+        return base.EnqueueAsync_WithReusedOptions_DoesNotChangeCallerOptionsAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WhenResolverReturnsEmpty_EnqueuesWithoutGroupAsync()
+    {
+        return base.EnqueueAsync_WhenResolverReturnsEmpty_EnqueuesWithoutGroupAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WhenResolverReturnsNull_EnqueuesWithoutGroupAsync()
+    {
+        return base.EnqueueAsync_WhenResolverReturnsNull_EnqueuesWithoutGroupAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithActivity_AppliesCorrelationBeforeEnqueuingHandlersAsync()
+    {
+        return base.EnqueueAsync_WithActivity_AppliesCorrelationBeforeEnqueuingHandlersAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithEmptyGroupIdAndResolver_UsesPayloadValueAsync()
+    {
+        return base.EnqueueAsync_WithEmptyGroupIdAndResolver_UsesPayloadValueAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithExplicitGroupId_OverridesResolverAsync()
+    {
+        return base.EnqueueAsync_WithExplicitGroupId_OverridesResolverAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithGroupIdResolver_DoesNotChangeCallerOptionsAsync()
+    {
+        return base.EnqueueAsync_WithGroupIdResolver_DoesNotChangeCallerOptionsAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithGroupIdResolver_UsesPayloadValueAsync()
+    {
+        return base.EnqueueAsync_WithGroupIdResolver_UsesPayloadValueAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithReusedOptionsAndGroupIdResolver_EnqueuesEachMessageAsync()
+    {
+        return base.EnqueueAsync_WithReusedOptionsAndGroupIdResolver_EnqueuesEachMessageAsync();
+    }
+
     [Fact]
     public override Task EnqueueAsync_WithUniqueId_UsesProvidedIdAsync()
     {