diff --git a/src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterReceiver.cs b/src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterReceiver.cs index 77f44eff836..3a6bcdca5af 100644 --- a/src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterReceiver.cs +++ b/src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterReceiver.cs @@ -1,21 +1,60 @@ namespace Orleans.Streaming.AdoNet; +internal interface IStreamMessageQueries +{ + Task> GetStreamMessagesAsync( + string serviceId, + string providerId, + string queueId, + int maxCount, + int maxAttempts, + int visibilityTimeout, + int removalTimeout, + int evictionInterval, + int evictionBatchSize); + + Task> ConfirmStreamMessagesAsync( + string serviceId, + string providerId, + string queueId, + IList messages); + + Task> ReleaseStreamMessagesAsync( + string serviceId, + string providerId, + string queueId, + IList messages); +} + /// /// Receives message batches from an individual queue of an ADO.NET provider. /// -internal partial class AdoNetQueueAdapterReceiver(string providerId, string queueId, AdoNetStreamOptions streamOptions, ClusterOptions clusterOptions, SimpleQueueCacheOptions cacheOptions, RelationalOrleansQueries queries, Serializer serializer, ILogger logger) : IQueueAdapterReceiver +internal partial class AdoNetQueueAdapterReceiver(string providerId, string queueId, AdoNetStreamOptions streamOptions, ClusterOptions clusterOptions, SimpleQueueCacheOptions cacheOptions, IStreamMessageQueries queries, Serializer serializer, ILogger logger) : IQueueAdapterReceiver { private readonly ILogger _logger = logger; + private readonly object _lock = new(); + private readonly Dictionary _pendingMessages = []; /// /// Flags that no further work should be attempted. /// private bool _shutdown; - /// - /// Helps shutdown wait for any outstanding storage operation. - /// - private Task? _outstandingTask; + private int _activeOperations; + private TaskCompletionSource? _operationsCompleted; + + public AdoNetQueueAdapterReceiver( + string providerId, + string queueId, + AdoNetStreamOptions streamOptions, + ClusterOptions clusterOptions, + SimpleQueueCacheOptions cacheOptions, + RelationalOrleansQueries queries, + Serializer serializer, + ILogger logger) + : this(providerId, queueId, streamOptions, clusterOptions, cacheOptions, new RelationalStreamMessageQueries(queries), serializer, logger) + { + } /// /// This receiver does not require initialization. @@ -27,29 +66,61 @@ internal partial class AdoNetQueueAdapterReceiver(string providerId, string queu /// public async Task Shutdown(TimeSpan timeout) { - // disable any further attempts to access storage - _shutdown = true; + Task? operationsCompleted; + lock (_lock) + { + _shutdown = true; + operationsCompleted = _operationsCompleted?.Task; + } - // wait for any outstanding storage operation to complete. - var outstandingTask = _outstandingTask; - if (outstandingTask is not null) + if (operationsCompleted is not null) { try { - await outstandingTask.WaitAsync(timeout); + await operationsCompleted.WaitAsync(timeout); } catch (Exception ex) { LogShutdownFault(ex, clusterOptions.ServiceId, providerId, queueId); + return; } } + + List pending; + lock (_lock) + { + RemoveExpiredPendingMessages(); + if (_pendingMessages.Count == 0) + { + return; + } + + pending = _pendingMessages + .Select(static item => new AdoNetStreamConfirmation(item.Key, item.Value.Dequeued)) + .ToList(); + } + + try + { + var released = await queries.ReleaseStreamMessagesAsync(clusterOptions.ServiceId, providerId, queueId, pending).WaitAsync(timeout); + lock (_lock) + { + foreach (var message in released) + { + _pendingMessages.Remove(message.MessageId); + } + } + } + catch (Exception ex) + { + LogReleaseFailed(ex, clusterOptions.ServiceId, providerId, queueId, pending); + } } /// public async Task> GetQueueMessagesAsync(int maxCount) { - // if shutdown has been called then we refuse further requests gracefully - if (_shutdown) + if (!TryBeginOperation()) { return []; } @@ -59,8 +130,7 @@ public async Task> GetQueueMessagesAsync(int maxCount) try { - // grab a message batch from storage while pinning the task so shutdown can wait for it - var task = queries.GetStreamMessagesAsync( + var messages = await queries.GetStreamMessagesAsync( clusterOptions.ServiceId, providerId, queueId, @@ -71,9 +141,14 @@ public async Task> GetQueueMessagesAsync(int maxCount) streamOptions.EvictionInterval.TotalSecondsCeiling(), streamOptions.EvictionBatchSize); - _outstandingTask = task; - - var messages = await task; + lock (_lock) + { + RemoveExpiredPendingMessages(); + foreach (var message in messages) + { + _pendingMessages[message.MessageId] = new(message.Dequeued, message.ExpiresOn); + } + } // convert the messages into standard batch containers return messages.Select(x => AdoNetBatchContainer.FromMessage(serializer, x)).Cast().ToList(); @@ -85,7 +160,7 @@ public async Task> GetQueueMessagesAsync(int maxCount) } finally { - _outstandingTask = null; + EndOperation(); } } @@ -98,31 +173,122 @@ public async Task MessagesDeliveredAsync(IList messages) return; } + if (!TryBeginOperation()) + { + return; + } + // get the identifiers for the messages to confirm var items = messages.Cast().Select(x => new AdoNetStreamConfirmation(x.SequenceToken.SequenceNumber, x.Dequeued)).ToList(); try { - // execute the confirmation while pinning the task so shutdown can wait for it - var task = queries.ConfirmStreamMessagesAsync(clusterOptions.ServiceId, providerId, queueId, items); - _outstandingTask = task; - try { - await task; + var confirmed = await queries.ConfirmStreamMessagesAsync(clusterOptions.ServiceId, providerId, queueId, items); + var receipts = items.ToDictionary(static item => item.MessageId, static item => item.Dequeued); + lock (_lock) + { + foreach (var message in confirmed) + { + if (receipts.TryGetValue(message.MessageId, out var receipt) + && _pendingMessages.TryGetValue(message.MessageId, out var pending) + && receipt == pending.Dequeued) + { + _pendingMessages.Remove(message.MessageId); + } + } + } } catch (Exception ex) { - LogConfirmationFailed(ex, clusterOptions.ClusterId, providerId, queueId, items); + LogConfirmationFailed(ex, clusterOptions.ServiceId, providerId, queueId, items); throw; } } finally { - _outstandingTask = null; + EndOperation(); } } + private bool TryBeginOperation() + { + lock (_lock) + { + if (_shutdown) + { + return false; + } + + if (_activeOperations++ == 0) + { + _operationsCompleted = new(TaskCreationOptions.RunContinuationsAsynchronously); + } + + return true; + } + } + + private void EndOperation() + { + TaskCompletionSource? operationsCompleted = null; + lock (_lock) + { + if (--_activeOperations == 0) + { + operationsCompleted = _operationsCompleted; + _operationsCompleted = null; + } + } + + operationsCompleted?.TrySetResult(); + } + + private void RemoveExpiredPendingMessages() + { + var now = DateTime.UtcNow; + var expired = _pendingMessages + .Where(message => message.Value.ExpiresOn <= now) + .Select(static message => message.Key) + .ToList(); + foreach (var messageId in expired) + { + _pendingMessages.Remove(messageId); + } + } + + private readonly record struct PendingMessage(int Dequeued, DateTime ExpiresOn); + + private sealed class RelationalStreamMessageQueries(RelationalOrleansQueries queries) : IStreamMessageQueries + { + public Task> GetStreamMessagesAsync( + string serviceId, + string providerId, + string queueId, + int maxCount, + int maxAttempts, + int visibilityTimeout, + int removalTimeout, + int evictionInterval, + int evictionBatchSize) => + queries.GetStreamMessagesAsync(serviceId, providerId, queueId, maxCount, maxAttempts, visibilityTimeout, removalTimeout, evictionInterval, evictionBatchSize); + + public Task> ConfirmStreamMessagesAsync( + string serviceId, + string providerId, + string queueId, + IList messages) => + queries.ConfirmStreamMessagesAsync(serviceId, providerId, queueId, messages); + + public Task> ReleaseStreamMessagesAsync( + string serviceId, + string providerId, + string queueId, + IList messages) => + queries.ReleaseStreamMessagesAsync(serviceId, providerId, queueId, messages); + } + #region Logging [LoggerMessage(1, LogLevel.Error, "Failed to get messages from ({ServiceId}, {ProviderId}, {QueueId})")] @@ -134,5 +300,8 @@ public async Task MessagesDeliveredAsync(IList messages) [LoggerMessage(3, LogLevel.Warning, "Handled fault while shutting down receiver for ({ServiceId}, {ProviderId}, {QueueId})")] private partial void LogShutdownFault(Exception exception, string serviceId, string providerId, string queueId); + [LoggerMessage(4, LogLevel.Warning, "Failed to release messages while shutting down receiver for ({ServiceId}, {ProviderId}, {QueueId}, {@Items})")] + private partial void LogReleaseFailed(Exception exception, string serviceId, string providerId, string queueId, List items); + #endregion Logging } diff --git a/src/AdoNet/Orleans.Streaming.AdoNet/MySQL-Streaming.sql b/src/AdoNet/Orleans.Streaming.AdoNet/MySQL-Streaming.sql index ed709933a54..7caa8b2e4e0 100644 --- a/src/AdoNet/Orleans.Streaming.AdoNet/MySQL-Streaming.sql +++ b/src/AdoNet/Orleans.Streaming.AdoNet/MySQL-Streaming.sql @@ -375,8 +375,8 @@ WHILE LOCATE(_Delimiter1, _Items) > 0 DO SET _Value = SUBSTRING_INDEX(_Items, _Delimiter1, 1); SET _MessageId = CAST(SUBSTRING_INDEX(_Value, _Delimiter2, 1) AS UNSIGNED); - SET _Dequeued = CAST(SUBSTRING_INDEX(_Value, _Delimiter2, -1) AS UNSIGNED); - + SET _Dequeued = CAST(SUBSTRING_INDEX(_Value, _Delimiter2, -1) AS SIGNED); + INSERT INTO _ItemsTable ( ServiceId, @@ -414,7 +414,7 @@ FROM AND M.ProviderId = I.ProviderId AND M.QueueId = I.QueueId AND M.MessageId = I.MessageId - AND M.Dequeued = I.Dequeued + AND M.Dequeued = ABS(I.Dequeued) ORDER BY M.ServiceId, M.ProviderId, @@ -422,14 +422,27 @@ ORDER BY M.MessageId FOR UPDATE; -/* delete the elected batch */ -DELETE M -FROM OrleansStreamMessage AS M -INNER JOIN _Batch AS B - ON M.ServiceId = B.ServiceId - AND M.ProviderId = B.ProviderId - AND M.QueueId = B.QueueId - AND M.MessageId = B.MessageId; +IF EXISTS (SELECT 1 FROM _ItemsTable WHERE Dequeued < 0) THEN + /* negative dequeue receipts release messages for immediate redelivery */ + UPDATE OrleansStreamMessage AS M + INNER JOIN _Batch AS B + ON M.ServiceId = B.ServiceId + AND M.ProviderId = B.ProviderId + AND M.QueueId = B.QueueId + AND M.MessageId = B.MessageId + SET + M.VisibleOn = UTC_TIMESTAMP(6), + M.ModifiedOn = UTC_TIMESTAMP(6); +ELSE + /* delete the elected batch */ + DELETE M + FROM OrleansStreamMessage AS M + INNER JOIN _Batch AS B + ON M.ServiceId = B.ServiceId + AND M.ProviderId = B.ProviderId + AND M.QueueId = B.QueueId + AND M.MessageId = B.MessageId; +END IF; /* return the ack */ SELECT diff --git a/src/AdoNet/Orleans.Streaming.AdoNet/PostgreSQL-Streaming.sql b/src/AdoNet/Orleans.Streaming.AdoNet/PostgreSQL-Streaming.sql index 96d752ef816..817742dab8f 100644 --- a/src/AdoNet/Orleans.Streaming.AdoNet/PostgreSQL-Streaming.sql +++ b/src/AdoNet/Orleans.Streaming.AdoNet/PostgreSQL-Streaming.sql @@ -316,6 +316,7 @@ AS $$ #VARIABLE_CONFLICT USE_COLUMN DECLARE _Count INT; + _Now TIMESTAMP(6) WITHOUT TIME ZONE := CURRENT_TIMESTAMP AT TIME ZONE 'UTC'; BEGIN CREATE TEMP TABLE _ItemsTable @@ -335,6 +336,48 @@ SELECT FROM UNNEST(string_to_array(_Items, '|')) AS Value; +/* negative dequeue receipts release messages for immediate redelivery */ +IF EXISTS (SELECT 1 FROM _ItemsTable WHERE Dequeued < 0) THEN + RETURN QUERY + WITH Batch AS + ( + SELECT + M.* + FROM + OrleansStreamMessage AS M + INNER JOIN _ItemsTable AS I + ON I.MessageId = M.MessageId + AND -I.Dequeued = M.Dequeued + WHERE + ServiceId = _ServiceId + AND ProviderId = _ProviderId + AND QueueId = _QueueId + ORDER BY + ServiceId, + ProviderId, + QueueId, + MessageId + FOR UPDATE + ) + UPDATE OrleansStreamMessage AS M + SET + VisibleOn = _Now, + ModifiedOn = _Now + FROM + Batch AS B + WHERE + M.ServiceId = B.ServiceId + AND M.ProviderId = B.ProviderId + AND M.QueueId = B.QueueId + AND M.MessageId = B.MessageId + RETURNING + M.ServiceId, + M.ProviderId, + M.QueueId, + M.MessageId; + RETURN; +END IF; + RETURN QUERY WITH Batch AS ( diff --git a/src/AdoNet/Orleans.Streaming.AdoNet/SQLServer-Streaming.sql b/src/AdoNet/Orleans.Streaming.AdoNet/SQLServer-Streaming.sql index 5e193dfea45..742792f7102 100644 --- a/src/AdoNet/Orleans.Streaming.AdoNet/SQLServer-Streaming.sql +++ b/src/AdoNet/Orleans.Streaming.AdoNet/SQLServer-Streaming.sql @@ -415,6 +415,47 @@ FROM /* count the number of messages to delete so we can use order by in the next query */ DECLARE @Count INT = (SELECT COUNT(*) FROM @ItemsTable); +/* negative dequeue receipts release messages for immediate redelivery */ +IF EXISTS (SELECT 1 FROM @ItemsTable WHERE Dequeued < 0) +BEGIN + DECLARE @Now DATETIME2(7) = SYSUTCDATETIME(); + + WITH Batch AS + ( + SELECT TOP (@Count) + M.* + FROM + OrleansStreamMessage AS M WITH (UPDLOCK, READPAST, READCOMMITTEDLOCK, ROWLOCK) + WHERE + ServiceId = @ServiceId + AND ProviderId = @ProviderId + AND QueueId = @QueueId + AND EXISTS + ( + SELECT * + FROM @ItemsTable AS I + WHERE I.MessageId = M.MessageId + AND -I.Dequeued = M.Dequeued + ) + ORDER BY + ServiceId, + ProviderId, + QueueId, + MessageId + ) + UPDATE Batch + SET + VisibleOn = @Now, + ModifiedOn = @Now + OUTPUT + Inserted.ServiceId, + Inserted.ProviderId, + Inserted.QueueId, + Inserted.MessageId; + + RETURN; +END; + /* delete messages in the exact same order as the clustered index to avoid deadlocks with other queries */ /* skip messages being changed concurrently since their dequeue receipt may no longer match */ WITH Batch AS diff --git a/src/AdoNet/Shared/Storage/RelationalOrleansQueries.cs b/src/AdoNet/Shared/Storage/RelationalOrleansQueries.cs index 86086c5e77e..a00c90978bb 100644 --- a/src/AdoNet/Shared/Storage/RelationalOrleansQueries.cs +++ b/src/AdoNet/Shared/Storage/RelationalOrleansQueries.cs @@ -495,10 +495,46 @@ internal Task> ConfirmStreamMessagesAsync(str return Task.FromResult>([]); } - // this builds a string in the form "1:2|3:4|5:6" where the first number is the message id and the second is the dequeue counter which acts as a receipt - // while we have more efficient ways of passing this data per RDMS, we use a string here to ensure call compatibility across ADONET providers - // it is the responsibility of the RDMS implementation to parse this string and apply it correctly - var items = messages.Aggregate(new StringBuilder(), (b, m) => b.Append(b.Length > 0 ? "|" : "").Append(m.MessageId).Append(':').Append(m.Dequeued), b => b.ToString()); + return ReadAsync>( + dbStoredQueries.ConfirmStreamMessagesKey, + record => new AdoNetStreamConfirmationAck( + (string)record[nameof(AdoNetStreamConfirmationAck.ServiceId)], + (string)record[nameof(AdoNetStreamConfirmationAck.ProviderId)], + (string)record[nameof(AdoNetStreamConfirmationAck.QueueId)], + (long)record[nameof(AdoNetStreamConfirmationAck.MessageId)]), + command => new DbStoredQueries.Columns(command) + { + ServiceId = serviceId, + ProviderId = providerId, + QueueId = queueId, + Items = FormatStreamConfirmations(messages, release: false) + }, + result => result.ToList()); + } + + /// + /// Makes unconfirmed stream messages immediately available for redelivery. + /// + /// The service identifier. + /// The provider identifier. + /// The queue identifier. + /// The messages to release. + /// A list of released messages. + /// + /// The dequeue counter acts as a receipt so that a stale receiver cannot release a message + /// which has already been dequeued by a new receiver. + /// + internal Task> ReleaseStreamMessagesAsync(string serviceId, string providerId, string queueId, IList messages) + { + ArgumentNullException.ThrowIfNull(serviceId); + ArgumentNullException.ThrowIfNull(providerId); + ArgumentNullException.ThrowIfNull(queueId); + ArgumentNullException.ThrowIfNull(messages); + + if (messages.Count == 0) + { + return Task.FromResult>([]); + } return ReadAsync>( dbStoredQueries.ConfirmStreamMessagesKey, @@ -512,11 +548,22 @@ internal Task> ConfirmStreamMessagesAsync(str ServiceId = serviceId, ProviderId = providerId, QueueId = queueId, - Items = items + Items = FormatStreamConfirmations(messages, release: true) }, result => result.ToList()); } + // Builds a provider-neutral receipt list in the form "1:2|3:4|5:6". + private static string FormatStreamConfirmations(IList messages, bool release) => + messages.Aggregate( + new StringBuilder(), + (builder, message) => builder + .Append(builder.Length > 0 ? "|" : "") + .Append(message.MessageId) + .Append(':') + .Append(release ? -message.Dequeued : message.Dequeued), + static builder => builder.ToString()); + /// /// Applies delivery failure logic to a stream message, such as making the message visible again or moving it to dead letters. /// diff --git a/src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs b/src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs index 444d99b1efd..f6da1b22d42 100644 --- a/src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs +++ b/src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs @@ -360,7 +360,16 @@ private async Task DoHandshakeWithConsumer( if (requestedToken != null) { consumerData.SafeDisposeCursor(logger); - consumerData.Cursor = queueCache.GetCacheCursor(consumerData.StreamId, requestedToken); + try + { + consumerData.Cursor = queueCache.GetCacheCursor(consumerData.StreamId, requestedToken); + } + catch (QueueCacheMissException) when (cacheToken is not null) + { + // A cold stream's triggering batch is the receiver's first available + // message, so resume there if the consumer's prior token was evicted. + consumerData.Cursor = queueCache.GetCacheCursor(consumerData.StreamId, cacheToken); + } } else { @@ -507,6 +516,12 @@ private async Task ReadFromQueue(QueueId myQueueId, IQueueAdapterReceiver? return false; } + // Pause all queue reads so a cold stream's first batch stays pinned until registration completes. + if (pubSubCache.Values.Any(static stream => stream.RegistrationTask is { IsCompleted: false })) + { + return false; + } + var now = _timeProvider.GetUtcNow().UtcDateTime; TagList? tags = null; @@ -746,9 +761,6 @@ async Task RegisterStreamAsync() finally { streamData.RegistrationTask = null; - - // Disposed after all initial subscriber handshakes complete so the first - // batch stays pinned until each subscriber has its own cache cursor. pinCursor?.Dispose(); } } @@ -889,10 +901,20 @@ private async Task RunConsumerCursor(StreamConsumerData consumerData) { consumerData.LastProcessedToken = newToken.Token; consumerData.LastToken = newToken; - IQueueCacheCursor newCursor = queueCache!.GetCacheCursor(consumerData.StreamId, newToken.Token); // queueCache must be non-null here: consumerData.Cursor was only ever populated via queueCache.GetCacheCursor. - // The handshake token points to an already processed event, we need to advance the cursor to - // the next event. - newCursor.MoveNext(); + IQueueCacheCursor newCursor; + try + { + newCursor = queueCache!.GetCacheCursor(consumerData.StreamId, newToken.Token); // queueCache must be non-null here: consumerData.Cursor was only ever populated via queueCache.GetCacheCursor. + // The handshake token points to an already processed event, so advance past it. + newCursor.MoveNext(); + } + catch (QueueCacheMissException) + { + // The current batch is the receiver's first available message. + // Keep it pending when the consumer resumes from an evicted token. + newCursor = queueCache!.GetCacheCursor(consumerData.StreamId, batch.SequenceToken); + } + consumerData.SafeDisposeCursor(logger); consumerData.Cursor = newCursor; } diff --git a/src/Orleans.Streaming/PersistentStreams/StreamConsumerCollection.cs b/src/Orleans.Streaming/PersistentStreams/StreamConsumerCollection.cs index fb197158024..e77c8f21a3a 100644 --- a/src/Orleans.Streaming/PersistentStreams/StreamConsumerCollection.cs +++ b/src/Orleans.Streaming/PersistentStreams/StreamConsumerCollection.cs @@ -69,7 +69,6 @@ public void DisposeAll(ILogger logger) queueData.Clear(); } - public int Count { get { return queueData.Count; } diff --git a/test/Extensions/Orleans.AdoNet.Tests/Streaming/AdoNetQueueAdapterReceiverTests.cs b/test/Extensions/Orleans.AdoNet.Tests/Streaming/AdoNetQueueAdapterReceiverTests.cs index 6dd266e7006..008abe3eb4f 100644 --- a/test/Extensions/Orleans.AdoNet.Tests/Streaming/AdoNetQueueAdapterReceiverTests.cs +++ b/test/Extensions/Orleans.AdoNet.Tests/Streaming/AdoNetQueueAdapterReceiverTests.cs @@ -3,6 +3,7 @@ using Orleans.Configuration; using Orleans.Streaming.AdoNet; using Orleans.Tests.SqlUtils; +using System.Runtime.CompilerServices; using TestExtensions; using UnitTests.General; using static System.String; @@ -10,6 +11,107 @@ namespace Tester.AdoNet.Streaming; +/// +/// Provider-independent lifecycle tests for . +/// +[Collection(TestEnvironmentFixture.DefaultCollection)] +[TestCategory("BVT"), TestCategory("AdoNet"), TestCategory("Streaming")] +[TestProvider("None")] +[TestSuite("BVT")] +[TestArea("Streaming")] +public class AdoNetQueueAdapterReceiverLifecycleTests(TestEnvironmentFixture fixture) +{ + [Fact] + public void AdoNetQueueAdapterReceiver_CanBeCreatedByAdapterFactory() => + RuntimeHelpers.RunClassConstructor(typeof(AdoNetQueueAdapter).TypeHandle); + + [Fact] + public async Task AdoNetQueueAdapterReceiver_Shutdown_WaitsForDequeueBookkeepingBeforeRelease() + { + var serviceId = $"Service-{Guid.NewGuid()}"; + var providerId = $"Provider-{Guid.NewGuid()}"; + var queueId = $"Queue-{Guid.NewGuid()}"; + var clusterOptions = new ClusterOptions { ServiceId = serviceId }; + var streamOptions = new AdoNetStreamOptions + { + VisibilityTimeout = TimeSpan.FromMinutes(5), + EvictionBatchSize = 0 + }; + var cacheOptions = new SimpleQueueCacheOptions(); + var serializer = fixture.Serializer.GetSerializer(); + var logger = NullLogger.Instance; + var payload = serializer.SerializeToArray(new AdoNetBatchContainer(StreamId.Create("MyNamespace", "MyKey"), [new TestModel(1)], null!)); + var now = DateTime.UtcNow; + var message = new AdoNetStreamMessage(serviceId, providerId, queueId, 42, 1, now.AddMinutes(5), now.AddHours(1), now, now, payload); + var dequeueStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var continueDequeue = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var queries = new BlockingStreamMessageQueries(message, dequeueStarted, continueDequeue); + + var receiver = new AdoNetQueueAdapterReceiver(providerId, queueId, streamOptions, clusterOptions, cacheOptions, queries, serializer, logger); + var getTask = receiver.GetQueueMessagesAsync(1); + await dequeueStarted.Task; + + var shutdownTask = receiver.Shutdown(TimeSpan.FromSeconds(10)); + Assert.False(shutdownTask.IsCompleted); + + continueDequeue.SetResult(); + var dequeued = Assert.IsType(Assert.Single(await getTask)); + await shutdownTask; + + Assert.Equal(message.MessageId, dequeued.SequenceToken.SequenceNumber); + var released = Assert.Single(queries.Released); + Assert.Equal(message.MessageId, released.MessageId); + Assert.Equal(message.Dequeued, released.Dequeued); + } + + [GenerateSerializer] + [Alias("Tester.AdoNet.Streaming.AdoNetQueueAdapterReceiverLifecycleTests.TestModel")] + public record TestModel( + [property: Id(0)] int Value); + + private sealed class BlockingStreamMessageQueries( + AdoNetStreamMessage message, + TaskCompletionSource dequeueStarted, + TaskCompletionSource continueDequeue) : IStreamMessageQueries + { + public IList Released { get; private set; } = []; + + public async Task> GetStreamMessagesAsync( + string serviceId, + string providerId, + string queueId, + int maxCount, + int maxAttempts, + int visibilityTimeout, + int removalTimeout, + int evictionInterval, + int evictionBatchSize) + { + dequeueStarted.SetResult(); + await continueDequeue.Task; + return [message]; + } + + public Task> ConfirmStreamMessagesAsync( + string serviceId, + string providerId, + string queueId, + IList messages) => + throw new NotSupportedException(); + + public Task> ReleaseStreamMessagesAsync( + string serviceId, + string providerId, + string queueId, + IList messages) + { + Released = messages.ToList(); + return Task.FromResult>( + [new(serviceId, providerId, queueId, message.MessageId)]); + } + } +} + /// /// Tests for against SQL Server. /// @@ -148,43 +250,80 @@ public async Task AdoNetQueueAdapterReceiver_GetsMessages_ConfirmsMessages() } /// - /// Tests that waits for the outstanding task. + /// Tests that shutting down a receiver immediately releases its unconfirmed messages. /// - /// [SkippableFact] - public async Task AdoNetQueueAdapterReceiver_Shutdown_WaitsForOutstandingTask() + public async Task AdoNetQueueAdapterReceiver_Shutdown_ReleasesUnconfirmedMessages() { - // arrange - receiver - var serviceId = "MyServiceId"; - var clusterOptions = new ClusterOptions + var serviceId = $"Service-{Guid.NewGuid()}"; + var providerId = $"Provider-{Guid.NewGuid()}"; + var queueId = $"Queue-{Guid.NewGuid()}"; + var clusterOptions = new ClusterOptions { ServiceId = serviceId }; + var streamOptions = new AdoNetStreamOptions { - ServiceId = serviceId + Invariant = invariant, + ConnectionString = _storage.ConnectionString, + VisibilityTimeout = TimeSpan.FromMinutes(5), + EvictionBatchSize = 0 }; - var providerId = "MyProviderId"; - var queueId = "MyQueueId"; + var cacheOptions = new SimpleQueueCacheOptions(); + var serializer = _fixture.Serializer.GetSerializer(); + var logger = NullLogger.Instance; + var streamId = StreamId.Create("MyNamespace", "MyKey"); + var payload = serializer.SerializeToArray(new AdoNetBatchContainer(streamId, [new TestModel(1)], null!)); + var ack = await _queries.QueueStreamMessageAsync(serviceId, providerId, queueId, payload, 100); + + var receiver = new AdoNetQueueAdapterReceiver(providerId, queueId, streamOptions, clusterOptions, cacheOptions, _queries, serializer, logger); + var first = Assert.IsType(Assert.Single(await receiver.GetQueueMessagesAsync(1))); + Assert.Equal(1, first.Dequeued); + + await receiver.Shutdown(TimeSpan.FromSeconds(10)); + + var replacement = new AdoNetQueueAdapterReceiver(providerId, queueId, streamOptions, clusterOptions, cacheOptions, _queries, serializer, logger); + var redelivered = Assert.IsType(Assert.Single(await replacement.GetQueueMessagesAsync(1))); + Assert.Equal(ack.MessageId, redelivered.SequenceToken.SequenceNumber); + Assert.Equal(2, redelivered.Dequeued); + await replacement.MessagesDeliveredAsync([redelivered]); + await replacement.Shutdown(TimeSpan.FromSeconds(10)); + } + + /// + /// Tests that waits for the outstanding task. + /// + [SkippableFact] + public async Task AdoNetQueueAdapterReceiver_Shutdown_WaitsForOutstandingTask() + { + var serviceId = $"Service-{Guid.NewGuid()}"; + var providerId = $"Provider-{Guid.NewGuid()}"; + var queueId = $"Queue-{Guid.NewGuid()}"; + var clusterOptions = new ClusterOptions { ServiceId = serviceId }; var streamOptions = new AdoNetStreamOptions { Invariant = invariant, - ConnectionString = _storage.ConnectionString + ConnectionString = _storage.ConnectionString, + VisibilityTimeout = TimeSpan.FromMinutes(5), + EvictionBatchSize = 0 }; var cacheOptions = new SimpleQueueCacheOptions(); var serializer = _fixture.Serializer.GetSerializer(); var logger = NullLogger.Instance; var receiver = new AdoNetQueueAdapterReceiver(providerId, queueId, streamOptions, clusterOptions, cacheOptions, _queries, serializer, logger); - await receiver.Initialize(TimeSpan.FromSeconds(10)); - - // arrange - enqueue a message var payload = serializer.SerializeToArray(new AdoNetBatchContainer(StreamId.Create("MyNamespace", "MyKey"), [new TestModel(1)], null!)); - await _queries.QueueStreamMessageAsync(serviceId, providerId, queueId, payload, 100); + var ack = await _queries.QueueStreamMessageAsync(serviceId, providerId, queueId, payload, 100); - // act - start getting messages from the receiver - var getTask = receiver.GetQueueMessagesAsync(10); - - // act - shutdown the receiver + var getTask = receiver.GetQueueMessagesAsync(1); await receiver.Shutdown(TimeSpan.FromSeconds(10)); - // assert - the outstanding task completes before the shutdown task Assert.True(getTask.IsCompleted); + var first = Assert.IsType(Assert.Single(await getTask)); + Assert.Equal(1, first.Dequeued); + + var replacement = new AdoNetQueueAdapterReceiver(providerId, queueId, streamOptions, clusterOptions, cacheOptions, _queries, serializer, logger); + var redelivered = Assert.IsType(Assert.Single(await replacement.GetQueueMessagesAsync(1))); + Assert.Equal(ack.MessageId, redelivered.SequenceToken.SequenceNumber); + Assert.Equal(2, redelivered.Dequeued); + await replacement.MessagesDeliveredAsync([redelivered]); + await replacement.Shutdown(TimeSpan.FromSeconds(10)); } public Task DisposeAsync() => Task.CompletedTask; diff --git a/test/Extensions/Orleans.AdoNet.Tests/Streaming/RelationalOrleansQueriesTests.cs b/test/Extensions/Orleans.AdoNet.Tests/Streaming/RelationalOrleansQueriesTests.cs index 29b5d851700..42dcd99f1d3 100644 --- a/test/Extensions/Orleans.AdoNet.Tests/Streaming/RelationalOrleansQueriesTests.cs +++ b/test/Extensions/Orleans.AdoNet.Tests/Streaming/RelationalOrleansQueriesTests.cs @@ -619,6 +619,43 @@ public async Task RelationalOrleansQueries_DoesNotDequeueSingleMessageBeforeVisi Assert.Empty(second); } + /// + /// Tests that a message can be released for immediate redelivery using its dequeue receipt. + /// + [SkippableFact] + public async Task RelationalOrleansQueries_ReleasesMessageUsingDequeueReceipt() + { + var serviceId = $"Service-{Guid.NewGuid()}"; + var providerId = $"Provider-{Guid.NewGuid()}"; + var queueId = $"Queue-{Guid.NewGuid()}"; + var payload = RandomPayload(); + var expiryTimeout = 100; + var maxCount = 1; + var maxAttempts = 5; + var visibilityTimeout = 100; + var removalTimeout = 100; + var evictionInterval = 100; + var evictionBatchSize = 0; + + var ack = await _queries.QueueStreamMessageAsync(serviceId, providerId, queueId, payload, expiryTimeout); + var first = Assert.Single(await _queries.GetStreamMessagesAsync(serviceId, providerId, queueId, maxCount, maxAttempts, visibilityTimeout, removalTimeout, evictionInterval, evictionBatchSize)); + var firstReceipt = new AdoNetStreamConfirmation(first.MessageId, first.Dequeued); + + var released = Assert.Single(await _queries.ReleaseStreamMessagesAsync(serviceId, providerId, queueId, [firstReceipt])); + Assert.Equal(ack.MessageId, released.MessageId); + + var second = Assert.Single(await _queries.GetStreamMessagesAsync(serviceId, providerId, queueId, maxCount, maxAttempts, visibilityTimeout, removalTimeout, evictionInterval, evictionBatchSize)); + Assert.Equal(first.Dequeued + 1, second.Dequeued); + + Assert.Empty(await _queries.ReleaseStreamMessagesAsync(serviceId, providerId, queueId, [firstReceipt])); + Assert.Empty(await _queries.GetStreamMessagesAsync(serviceId, providerId, queueId, maxCount, maxAttempts, visibilityTimeout, removalTimeout, evictionInterval, evictionBatchSize)); + + var secondReceipt = new AdoNetStreamConfirmation(second.MessageId, second.Dequeued); + Assert.Single(await _queries.ReleaseStreamMessagesAsync(serviceId, providerId, queueId, [secondReceipt])); + var third = Assert.Single(await _queries.GetStreamMessagesAsync(serviceId, providerId, queueId, maxCount, maxAttempts, visibilityTimeout, removalTimeout, evictionInterval, evictionBatchSize)); + await _queries.ConfirmStreamMessagesAsync(serviceId, providerId, queueId, [new AdoNetStreamConfirmation(third.MessageId, third.Dequeued)]); + } + /// /// Tests that a single message is not dequeued again after expiry /// diff --git a/test/Orleans.Streaming.Tests/StreamingTests/PersistentStreamPullingAgentTests.cs b/test/Orleans.Streaming.Tests/StreamingTests/PersistentStreamPullingAgentTests.cs index 14a3f2d790d..fcd03b9208b 100644 --- a/test/Orleans.Streaming.Tests/StreamingTests/PersistentStreamPullingAgentTests.cs +++ b/test/Orleans.Streaming.Tests/StreamingTests/PersistentStreamPullingAgentTests.cs @@ -60,10 +60,15 @@ public async Task ReadFromQueue_DoesNotWaitForColdStreamRegistration() Assert.True(await readTask, "ReadFromQueue should return true indicating data was read"); + Assert.False(await testAccessor.ReadFromQueue(queueId, receiver, 1)); + await receiver.Received(1).GetQueueMessagesAsync(Arg.Any(), Arg.Any()); + // Completing registration should resolve the tracked task and clear it. registration.SetResult(new HashSet()); await registrationTask; + Assert.True(await testAccessor.ReadFromQueue(queueId, receiver, 1)); Assert.Null(streamData.RegistrationTask); + await receiver.Received(2).GetQueueMessagesAsync(Arg.Any(), Arg.Any()); } [TestSuite("BVT")] diff --git a/test/TestInfrastructure/TestExtensions/Diagnostics/StreamingDiagnosticObserver.cs b/test/TestInfrastructure/TestExtensions/Diagnostics/StreamingDiagnosticObserver.cs index 91ef437da7c..5e398f5e9d9 100644 --- a/test/TestInfrastructure/TestExtensions/Diagnostics/StreamingDiagnosticObserver.cs +++ b/test/TestInfrastructure/TestExtensions/Diagnostics/StreamingDiagnosticObserver.cs @@ -178,13 +178,25 @@ await _events /// public async Task WaitForItemDeliveryCountAsync(StreamId streamId, int expectedCount, string? streamProvider, CancellationToken cancellationToken) { - await _events - .OfType() - .Where(e => MatchesStream(e.StreamId, e.StreamProvider, streamId, streamProvider)) - .Take(expectedCount) - .LastOrDefaultAsync() - .ToTask(cancellationToken) - .ConfigureAwait(false); + var actualCount = 0; + try + { + await _events + .OfType() + .Where(e => MatchesStream(e.StreamId, e.StreamProvider, streamId, streamProvider)) + .Do(_ => Interlocked.Increment(ref actualCount)) + .Take(expectedCount) + .LastOrDefaultAsync() + .ToTask(cancellationToken) + .ConfigureAwait(false); + } + catch (OperationCanceledException exception) when (cancellationToken.IsCancellationRequested) + { + throw new OperationCanceledException( + $"Canceled while waiting for {expectedCount} item deliveries on stream {streamId} from provider '{streamProvider ?? ""}'; observed {Volatile.Read(ref actualCount)}.", + exception, + cancellationToken); + } } /// @@ -192,13 +204,25 @@ await _events /// public async Task WaitForItemDeliveryCountAsync(StreamId streamId, Guid subscriptionId, int expectedCount, string? streamProvider, CancellationToken cancellationToken) { - await _events - .OfType() - .Where(e => MatchesSubscription(e.StreamId, e.SubscriptionId, e.StreamProvider, streamId, subscriptionId, streamProvider)) - .Take(expectedCount) - .LastOrDefaultAsync() - .ToTask(cancellationToken) - .ConfigureAwait(false); + var actualCount = 0; + try + { + await _events + .OfType() + .Where(e => MatchesSubscription(e.StreamId, e.SubscriptionId, e.StreamProvider, streamId, subscriptionId, streamProvider)) + .Do(_ => Interlocked.Increment(ref actualCount)) + .Take(expectedCount) + .LastOrDefaultAsync() + .ToTask(cancellationToken) + .ConfigureAwait(false); + } + catch (OperationCanceledException exception) when (cancellationToken.IsCancellationRequested) + { + throw new OperationCanceledException( + $"Canceled while waiting for {expectedCount} item deliveries on stream {streamId}, subscription {subscriptionId}, from provider '{streamProvider ?? ""}'; observed {Volatile.Read(ref actualCount)}.", + exception, + cancellationToken); + } } ///