fix(streaming): preserve ADO.NET dequeue order

src/AdoNet/Orleans.Streaming.AdoNet/MySQL-Streaming.sql

@@ -318,7 +318,9 @@ FROM
         ON M.ServiceId = B.ServiceId
         AND M.ProviderId = B.ProviderId
         AND M.QueueId = B.QueueId
-        AND M.MessageId = B.MessageId;
+        AND M.MessageId = B.MessageId
+ORDER BY
+    M.MessageId;
 
 DROP TEMPORARY TABLE _Batch;
 

src/AdoNet/Shared/Storage/RelationalOrleansQueries.cs

@@ -464,7 +464,12 @@ internal Task<IList<AdoNetStreamMessage>> GetStreamMessagesAsync(string serviceI
                     EvictionInterval = evictionInterval,
                     EvictionBatchSize = evictionBatchSize
                 },
-                result => result.ToList());
+                result =>
+                {
+                    var messages = result.ToList();
+                    messages.Sort(static (left, right) => left.MessageId.CompareTo(right.MessageId));
+                    return messages;
+                });
         }
 
         /// <summary>

test/Extensions/Orleans.AdoNet.Tests/Streaming/RelationalOrleansQueriesTests.cs

@@ -31,6 +31,9 @@ public MySqlRelationalOrleansQueriesTests() : base(AdoNetInvariants.InvariantNam
     {
         MySqlConnection.ClearAllPools();
     }
+
+    [SkippableFact]
+    public Task RelationalOrleansQueries_OrdersProviderResults() => VerifyProviderResultsAreOrdered();
 }
 
 /// <summary>
@@ -85,8 +88,64 @@ private static byte[] RandomPayload(int size = 1_000_000)
         return payload;
     }
 
+    private async Task<AdoNetStreamMessageAck[]> QueueMessagesAsync(
+        string serviceId,
+        string providerId,
+        string queueId,
+        byte[] payload,
+        int expiryTimeout,
+        int count)
+    {
+        using var semaphore = new SemaphoreSlim(concurrency);
+        return await Task.WhenAll(Enumerable.Range(0, count).Select(async _ =>
+        {
+            await semaphore.WaitAsync();
+            try
+            {
+                return await _queries.QueueStreamMessageAsync(serviceId, providerId, queueId, payload, expiryTimeout);
+            }
+            finally
+            {
+                semaphore.Release();
+            }
+        }));
+    }
+
     public Task DisposeAsync() => Task.CompletedTask;
 
+    protected async Task VerifyProviderResultsAreOrdered()
+    {
+        const string reverseQuery = """
+            SELECT ServiceId, ProviderId, QueueId, MessageId, Dequeued, VisibleOn, ExpiresOn, CreatedOn, ModifiedOn, Payload
+            FROM OrleansStreamMessage
+            WHERE ServiceId = @ServiceId AND ProviderId = @ProviderId AND QueueId = @QueueId
+            ORDER BY MessageId DESC
+            LIMIT @MaxCount
+            """;
+
+        var serviceId = RandomServiceId();
+        var providerId = RandomProviderId();
+        var queueId = RandomQueueId();
+        var payload = new byte[] { 0xFF };
+
+        await QueueMessagesAsync(serviceId, providerId, queueId, payload, 100, 100);
+
+        await _storage.ExecuteAsync(
+            "UPDATE OrleansQuery SET QueryText = @QueryText WHERE QueryKey = 'GetStreamMessagesKey'",
+            command =>
+            {
+                var parameter = command.CreateParameter();
+                parameter.ParameterName = "QueryText";
+                parameter.Value = reverseQuery;
+                command.Parameters.Add(parameter);
+            });
+
+        var queries = await RelationalOrleansQueries.CreateInstance(invariant, _storage.ConnectionString);
+        var messages = await queries.GetStreamMessagesAsync(serviceId, providerId, queueId, 100, 3, 10, 100, 10, 1000);
+
+        Assert.Equal(messages.OrderBy(message => message.MessageId), messages);
+    }
+
     /// <summary>
     /// Tests that a single message is queued.
     /// </summary>
@@ -366,25 +425,22 @@ public async Task RelationalOrleansQueries_DequeuesMessageBatches()
         var serviceId = RandomServiceId();
         var providerId = RandomProviderId();
         var queueId = RandomQueueId();
-        var payload = RandomPayload();
+        var payload = new byte[] { 0xFF };
         var expiryTimeout = 100;
-        var maxCount = 3;
+        var maxCount = 60;
         var maxAttempts = 3;
         var visibilityTimeout = 10;
         var removalTimeout = 100;
         var evictionInterval = 10;
         var evictionBatchSize = 1000;
-        var total = 5;
+        var total = 100;
 
-        // arrange - enqueue five messages
+        // arrange - enqueue messages concurrently
         var beforeQueueing = DateTime.UtcNow.AddSeconds(-1);
-        var acks = await Task.WhenAll(Enumerable
-            .Range(0, total)
-            .Select(i => _queries.QueueStreamMessageAsync(serviceId, providerId, queueId, payload, expiryTimeout))
-            .ToList());
+        var acks = await QueueMessagesAsync(serviceId, providerId, queueId, payload, expiryTimeout, total);
         var afterQueueing = DateTime.UtcNow.AddSeconds(1);
 
-        // act - dequeue three batches of three messages
+        // act - dequeue all messages
         var beforeDequeuing = DateTime.UtcNow.AddSeconds(-1);
         var first = await _queries.GetStreamMessagesAsync(serviceId, providerId, queueId, maxCount, maxAttempts, visibilityTimeout, removalTimeout, evictionInterval, evictionBatchSize);
         var second = await _queries.GetStreamMessagesAsync(serviceId, providerId, queueId, maxCount, maxAttempts, visibilityTimeout, removalTimeout, evictionInterval, evictionBatchSize);
@@ -396,9 +452,11 @@ public async Task RelationalOrleansQueries_DequeuesMessageBatches()
         Assert.Equal(total - maxCount, second.Count);
         Assert.Empty(third);
 
+        var messages = first.Concat(second).Concat(third).ToList();
+        Assert.Equal(messages.OrderBy(message => message.MessageId), messages);
+
         // assert - dequeued messages are consistent with acks
         var ackLookup = acks.ToDictionary(x => (x.ServiceId, x.ProviderId, x.QueueId, x.MessageId));
-        var messages = first.Concat(second).Concat(third).ToList();
         foreach (var message in messages)
         {
             Assert.True(ackLookup.TryGetValue((message.ServiceId, message.ProviderId, message.QueueId, message.MessageId), out var ack), "Ack not found");