From 0d80a8e3243d6c02e5894918b688833ae56aa1f6 Mon Sep 17 00:00:00 2001 From: Cosmin Staicu Date: Sat, 24 Oct 2026 13:23:32 +0300 Subject: [PATCH] fix(broadcast): recover a stream consumer's pending entries and keep caught-up groups (2.1) A timed-out stream read lost its messages. Redis moves the consumer group past the entries of an XREADGROUP and puts them in the consumer's pending list even when the answer never reaches the client, and the reader only ever read new entries (">"), so those invalidations were never applied. The reader now reads its own pending entries (from "0", then on from the last one) when it starts and after every failed read, before reading new entries again. - A pending entry trimmed or deleted from the stream before it was read is acknowledged and, once acknowledged, logged as a Warning. 2.1 has no missed-message signal to raise. - An entry that cannot be decoded is acknowledged and dropped, not reported as missed: a bad payload is not a lost invalidation. It is tracked as the UndecodableEvent telemetry event and logged as a Warning once per 100. - A read timeout is logged as a Warning once per 100, at Debug otherwise, instead of as an Error each time. The stream maintainer no longer deletes the consumer group of a live subscriber on a quiet stream. A group with consumers counted as stale once its last delivery was older than MaintainerTrimInterval, even when nothing newer had been written. A group whose last-delivered-id has reached the stream's last-generated-id is now healthy, and a quarantine it was under is cleared. The broadcast docs and the sample settings no longer describe Limit as capping a read: neither the reader nor the publisher uses it. Signed-off-by: Cosmin Staicu --- CHANGELOG.md | 20 ++ docs/how-to/broadcast.md | 8 +- docs/reference/settings.md | 2 +- .../appsettings.all.json | 2 +- .../Redis/RedisStreamHealthMaintainer.cs | 28 +- .../Redis/RedisStreamSubjectWriter.cs | 122 +++++++- .../Broadcast/Redis/StreamConstants.cs | 9 + ...sStreamHealthMaintainerIntegrationTests.cs | 46 +++ ...edisStreamSubjectWriterIntegrationTests.cs | 85 +++++ .../RedisStreamSubjectWriterTests.cs | 292 +++++++++++++++++- .../Broadcast/RedisStreamTopicMonitorTests.cs | 81 +++++ 11 files changed, 672 insertions(+), 23 deletions(-) create mode 100644 tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterIntegrationTests.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d49051e..e01aadcf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,26 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) ## [Unreleased] +### Fixed + +- **A Redis Streams read that times out no longer loses its invalidations.** Redis hands the entries of a read to the + consumer and moves the group past them even when the answer never reaches the client, and the reader only ever asked + for new entries, so those stayed pending and their invalidations were never applied: the node kept the old values + until they expired. The reader now reads its own pending entries when it starts and after every failed read, before + going back to new ones. A pending entry deleted from the stream before it was read is acknowledged and logged as a + Warning, since its invalidation is lost. A read timeout is logged as a + Warning once per 100, and at Debug otherwise, rather than as an Error every time. An entry that cannot be decoded is + acknowledged and dropped, not reported as a missed message: a bad payload is not a lost invalidation, and reporting it + would let one bad writer clear the local tier of every node. Each one is tracked as the + `Caching.RedisStreamSubjectWriter.DispatchEventsAsync.UndecodableEvent` telemetry event and logged as a Warning once + per 100. + +- **The stream maintainer no longer deletes the consumer group of a live subscriber on a quiet stream.** A group with + consumers counted as stale once its last delivery was older than `MaintainerTrimInterval`, even when nothing newer had + been written, so on a stream with no traffic every live group was quarantined and then deleted after + `MaintainerQuarantineInterval`, and its reader recreated it on the next `NOGROUP`. A group whose last delivery is the + stream's last entry is now healthy, and a quarantine it was under is cleared. + ### Changed - **`RedisStreamsTopicOptions.PollBatchSize` defaults to 1024 instead of 4096.** It is the `COUNT` of each diff --git a/docs/how-to/broadcast.md b/docs/how-to/broadcast.md index 0db278e0..e11b5707 100644 --- a/docs/how-to/broadcast.md +++ b/docs/how-to/broadcast.md @@ -79,7 +79,7 @@ See [reference/settings.md](../reference/settings.md) for the full set. The tabl | --- | --- | --- | --- | | `Enabled` | `true` | No | App-wide provider toggle. | | `MaxLength` | `32768` | Yes | `XADD MAXLEN ~` cap (server-side trim). | -| `Limit` | `1024` | Yes | Max entries per `XREADGROUP` call. | +| `Limit` | `1024` | Yes | Not used by the stream reader or the publisher. | | `PollBatchSize` | `1024` | Yes | Entries one `XREADGROUP` asks for. | | `FieldName` | `"event"` | Yes | Redis stream field name for the payload. | | `PollInterval` | `250 ms` | Yes | `XREADGROUP` cadence. | @@ -116,9 +116,9 @@ See [reference/settings.md](../reference/settings.md) for the full set. The tabl | `SubscriberTimeout` | multiplexer timeout | Yes | Resubscribe interval. | | `SubscriberDueTime` | half of timeout | Yes | First-subscribe delay. | -### `Limit` vs `MaxLength` vs `PollBatchSize` +### `MaxLength` vs `PollBatchSize` -All three touch how many entries move through the consumer pipeline, but at different layers: +Both touch how many entries move through the consumer pipeline, but at different layers (`Limit` is not used): - `MaxLength` (default 32768) is a server-side cap — `XADD MAXLEN ~` trims the stream to at most this many entries. Tune it to bound Redis memory usage on the stream key. - `Limit` (default 1024) is not used by the stream reader or the publisher. @@ -212,7 +212,7 @@ Some options are read at provider construction or by singleton hosted services. - `MaintainerQuarantineInterval` — quarantine window for empty consumer groups; read per maintainer cycle. - `MaintainerSearchPattern` — SCAN pattern used to discover streams; resolved once at `Initialize`. -Everything else (`MaxLength`, `Limit`, `PollBatchSize`, `PollInterval`, `ConsumerCapacity`, `FullMode`, `SlowObserverThreshold`, every `Notify*` field, etc.) is honored per-topic. +Everything else (`MaxLength`, `PollBatchSize`, `PollInterval`, `ConsumerCapacity`, `FullMode`, `SlowObserverThreshold`, every `Notify*` field, etc.) is honored per-topic. ## Notify doorbell diff --git a/docs/reference/settings.md b/docs/reference/settings.md index e01b9633..043a5202 100644 --- a/docs/reference/settings.md +++ b/docs/reference/settings.md @@ -108,7 +108,7 @@ App-wide-only fields (per-topic `Topics[]` entries ignore these): `Enabled`, `Co |---|---|---|---|---| | `Enabled` | `bool` | `true` | App-wide only | Enable/disable the Redis Streams provider for the whole app; per-topic override is ignored. | | `MaxLength` | `long?` | `32768` | App-wide / Per-topic | Max entries in a stream before approximate trimming (`MAXLEN ~`); `null` = unlimited. | -| `Limit` | `long?` | `1024` | App-wide / Per-topic | Max entries returned per `XREAD` call. | +| `Limit` | `long?` | `1024` | App-wide / Per-topic | Not used by the stream reader or the publisher. | | `PollBatchSize` | `int` | `1024` | App-wide / Per-topic | Entries one `XREADGROUP` asks for. | | `FieldName` | `string` | `"event"` | App-wide / Per-topic | Stream entry field name used for the serialized event payload. | | `PollInterval` | `TimeSpan` | `00:00:00.250` | App-wide / Per-topic | Time between poll cycles when no notify signal is received. | diff --git a/samples/UiPath.Caching.Sample/appsettings.all.json b/samples/UiPath.Caching.Sample/appsettings.all.json index 6a02864e..0ec2d2b3 100644 --- a/samples/UiPath.Caching.Sample/appsettings.all.json +++ b/samples/UiPath.Caching.Sample/appsettings.all.json @@ -149,7 +149,7 @@ "Enabled": true, // MaxLength: max entries in a stream before trimming (MAXLEN ~); null = unlimited "MaxLength": 32768, - // Limit: max entries returned per XREAD call + // Limit: not used by the stream reader or the publisher "Limit": 1024, // PollBatchSize: max entries fetched per poll cycle "PollBatchSize": 1024, diff --git a/src/UiPath.Caching/Broadcast/Redis/RedisStreamHealthMaintainer.cs b/src/UiPath.Caching/Broadcast/Redis/RedisStreamHealthMaintainer.cs index cf3bbfa7..9ae4c754 100644 --- a/src/UiPath.Caching/Broadcast/Redis/RedisStreamHealthMaintainer.cs +++ b/src/UiPath.Caching/Broadcast/Redis/RedisStreamHealthMaintainer.cs @@ -188,6 +188,21 @@ private static bool TryParseDeliveredIdToDatetimeOffset(string? entryId, out Dat } } + private static bool IsCaughtUp(string? lastDeliveredId, RedisValue lastGeneratedId) => + TryParseEntryId(lastDeliveredId, out var delivered) + && TryParseEntryId(lastGeneratedId, out var generated) + && delivered.CompareTo(generated) >= 0; + + private static bool TryParseEntryId(string? entryId, out (ulong Milliseconds, ulong Sequence) id) + { + id = default; + var span = entryId.AsSpan(); + var separator = span.IndexOf('-'); + return separator > 0 + && ulong.TryParse(span[..separator], NumberStyles.None, CultureInfo.InvariantCulture, out id.Milliseconds) + && ulong.TryParse(span[(separator + 1)..], NumberStyles.None, CultureInfo.InvariantCulture, out id.Sequence); + } + private static List ConnectedPrimaries(IConnectionMultiplexer? multiplexer) => multiplexer?.GetServers().Where(static server => !server.IsReplica && server.IsConnected).ToList() ?? []; @@ -287,7 +302,7 @@ private async Task CheckStreamWithGroupsAsync(StreamContext context, DateTimeOff foreach (var groupInfo in groupInfos) { - await CheckStreamGroupAsync(context, groupInfo, minOffset, cancellationToken).ConfigureAwait(false); + await CheckStreamGroupAsync(context, groupInfo, minOffset, streamInfo.LastGeneratedId, cancellationToken).ConfigureAwait(false); } if (_supportsXtrimMinId.Value && TryParseDeliveredIdToDatetimeOffset(streamInfo.LastGeneratedId, out var dateTimeOffset) && dateTimeOffset > minOffset) @@ -319,7 +334,7 @@ private async Task CheckEmptyStreamAsync(StreamContext context, StreamInfo strea } } - private async Task CheckStreamGroupAsync(StreamContext context, StreamGroupInfo groupInfo, DateTimeOffset minOffset, CancellationToken cancellationToken) + private async Task CheckStreamGroupAsync(StreamContext context, StreamGroupInfo groupInfo, DateTimeOffset minOffset, RedisValue lastGeneratedId, CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); @@ -331,16 +346,17 @@ private async Task CheckStreamGroupAsync(StreamContext context, StreamGroupInfo } else { - await CheckStreamGroupWithConsumersAsync(context, groupInfo, minOffset, cancellationToken).ConfigureAwait(false); + await CheckStreamGroupWithConsumersAsync(context, groupInfo, minOffset, lastGeneratedId, cancellationToken).ConfigureAwait(false); } } - private async Task CheckStreamGroupWithConsumersAsync(StreamContext context, StreamGroupInfo groupInfo, DateTimeOffset minOffset, CancellationToken cancellationToken) + private async Task CheckStreamGroupWithConsumersAsync(StreamContext context, StreamGroupInfo groupInfo, DateTimeOffset minOffset, RedisValue lastGeneratedId, CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); - // Check if the consumer group's last-delivered-id is older than minOffset - if (TryParseDeliveredIdToDatetimeOffset(groupInfo.LastDeliveredId, out var lastDeliveredOffset) && + // A group at the head of a quiet stream has nothing left to read, however old its last delivery. + if (!IsCaughtUp(groupInfo.LastDeliveredId, lastGeneratedId) && + TryParseDeliveredIdToDatetimeOffset(groupInfo.LastDeliveredId, out var lastDeliveredOffset) && lastDeliveredOffset.HasValue && lastDeliveredOffset.Value < minOffset) { diff --git a/src/UiPath.Caching/Broadcast/Redis/RedisStreamSubjectWriter.cs b/src/UiPath.Caching/Broadcast/Redis/RedisStreamSubjectWriter.cs index 9e5803dd..848c07b8 100644 --- a/src/UiPath.Caching/Broadcast/Redis/RedisStreamSubjectWriter.cs +++ b/src/UiPath.Caching/Broadcast/Redis/RedisStreamSubjectWriter.cs @@ -7,6 +7,7 @@ internal sealed partial class RedisStreamSubjectWriter : IDisposable where T : IEvent { private const string EventInvalid = "Caching." + nameof(RedisStreamSubjectWriter) + "." + nameof(DispatchEventsAsync) + ".InvalidEvent"; + private const string EventUndecodable = "Caching." + nameof(RedisStreamSubjectWriter) + "." + nameof(DispatchEventsAsync) + ".UndecodableEvent"; private const string EventReceived = "Caching." + nameof(RedisStreamSubjectWriter) + "." + nameof(DispatchEventsAsync) + ".EventReceived"; private const string PropTopicKey = "TopicKey"; private const string PropTransportId = "TransportId"; @@ -26,6 +27,10 @@ internal sealed partial class RedisStreamSubjectWriter : IDisposable private bool _disposed; private RedisValue _lastId = StreamPosition.NewMessages; + // Where the next read of this consumer's pending entries starts; null once they are all read. + private RedisValue? _pendingFrom = StreamConstants.PendingMessages; + private long _timeouts; + private long _undecodable; private int _consecutiveFailures; private volatile bool _unsupportedCommand; @@ -149,12 +154,13 @@ private async Task FetchBatch() return; } - StreamEntry[] events = await _redis.Database.StreamReadGroupAsync( - _context.Topic, - _context.ConsumerGroup, - _context.ConsumerName, - StreamConstants.UndeliveredMessages, - _context.PollBatchSize).ConfigureAwait(false); + if (_pendingFrom is { } from) + { + await FetchPendingBatch(from).ConfigureAwait(false); + return; + } + + StreamEntry[] events = await ReadAsync(StreamConstants.UndeliveredMessages).ConfigureAwait(false); _consecutiveFailures = 0; if (_unsupportedCommand) @@ -172,6 +178,34 @@ private async Task FetchBatch() await _waiter.WaitAsync(_cancelationToken).ConfigureAwait(false); } + /// Reads the entries Redis gave this consumer without an acknowledgement, such as those of a read that timed out on the client. + private async Task FetchPendingBatch(RedisValue from) + { + StreamEntry[] events = await ReadAsync(from).ConfigureAwait(false); + if (events.Length > 0) + { + await DispatchEventsAsync(events).ConfigureAwait(false); + } + + // Given to this consumer, then trimmed or deleted before it read them: their invalidations are lost. Told only once + // acknowledged, so a failed acknowledgement's retry does not tell it again. + var deleted = events.Count(static e => !e.IsNull && e.Values is not { Length: > 0 }); + if (deleted > 0) + { + LogPendingEntriesDeleted(_context.Topic, deleted); + } + + // Read on from the last entry, not from the start: an entry left unacknowledged would be read again and again. + _pendingFrom = events.Length < _context.PollBatchSize || events[^1].Id.IsNull ? null : (RedisValue?)events[^1].Id; + } + + private Task ReadAsync(RedisValue position) => _redis.Database.StreamReadGroupAsync( + _context.Topic, + _context.ConsumerGroup, + _context.ConsumerName, + position, + _context.PollBatchSize); + private async Task ProcessException(Exception ex) { var id = _lastId; @@ -181,6 +215,8 @@ private async Task ProcessException(Exception ex) return false; } + _pendingFrom = StreamConstants.PendingMessages; + if (IsUnsupportedCommand(ex)) { if (!_unsupportedCommand) @@ -216,13 +252,29 @@ private async Task ProcessException(Exception ex) } else { - LogFetchLoopError(ex); + LogFetchFailed(ex); } await BackoffAsync().ConfigureAwait(false); return true; } + private void LogFetchFailed(Exception ex) + { + if (ex is not TimeoutException) + { + LogFetchLoopError(ex); + } + else if (++_timeouts is var timeouts && timeouts % StreamConstants.TimeoutsPerWarning == 1) + { + LogFetchTimedOut(ex, _context.Topic, timeouts); + } + else + { + LogFetchTimedOutAgain(ex, _context.Topic); + } + } + private Task BackoffAsync() { var failures = ++_consecutiveFailures; @@ -277,11 +329,29 @@ private ValueTask ProcessEvent(StreamEntry @event, List ids) return default; } + // A pending entry since deleted from the stream comes back without fields, and would stay pending for good. + if (@event.Values is not { Length: > 0 }) + { + ids.Add(@event.Id); + return default; + } + try { - var ev = _formatter.Decode((ReadOnlyMemory)@event[_context.FieldName]); + T? ev; + try + { + ev = _formatter.Decode((ReadOnlyMemory)@event[_context.FieldName]); + } + catch (Exception ex) + { + HandleUndecodableEvent(@event, ids, ex); + return default; + } + if (ev is null) { + HandleUndecodableEvent(@event, ids, null); return default; } ev.AttachTransportId(@event.Id); @@ -307,6 +377,7 @@ private ValueTask ProcessEvent(StreamEntry @event, List ids) catch (Exception ex) { LogOnMessageError(ex, _context.Topic); + ids.Add(@event.Id); return default; } } @@ -353,6 +424,26 @@ private void HandleInvalidEvent(T ev, StreamEntry @event, List ids) LogEventInvalid(ev.Id, _context.Topic, @event.Id); } + /// Drops an entry no retry would decode: acknowledged, since a bad payload is not a lost invalidation. + private void HandleUndecodableEvent(StreamEntry @event, List ids, Exception? ex) + { + // Acknowledged first, so a throw below cannot leave the entry pending. + ids.Add(@event.Id); + _cachingTelemetryProvider.TryTrackEvent(EventUndecodable, + [ + new(PropTopicKey, _context.Topic.ToString()), + new(PropTransportId, @event.Id.ToString()), + ]); + if (++_undecodable is var count && count % StreamConstants.UndecodableEntriesPerWarning == 1) + { + LogEventUndecodable(ex, _context.Topic, @event.Id, count); + } + else + { + LogEventUndecodableAgain(ex, _context.Topic, @event.Id); + } + } + private void TraceReceipt(T ev) { LogEventReceived(ev.Id, _context.Topic, ev.TransportId); @@ -389,6 +480,12 @@ private void TraceReceipt(T ev) [LoggerMessage(Level = LogLevel.Error, Message = "Fetch events loop error")] private partial void LogFetchLoopError(Exception ex); + [LoggerMessage(Level = LogLevel.Warning, Message = "Reading topic {Topic} timed out ({Timeouts} times so far); the entries of a read that timed out are read again from the pending list.")] + private partial void LogFetchTimedOut(Exception ex, RedisKey topic, long timeouts); + + [LoggerMessage(Level = LogLevel.Debug, Message = "Reading topic {Topic} timed out")] + private partial void LogFetchTimedOutAgain(Exception ex, RedisKey topic); + [LoggerMessage( Level = LogLevel.Critical, Message = "Redis Streams reads are not supported by the server for topic {Topic}. This node will not receive cross-node cache invalidations over Streams. Retrying slowly in case the connection moves to a server that supports XREADGROUP; to fix it now switch to Pub/Sub (DefaultTopic: RedisPubSub) or disable broadcast.")] @@ -406,6 +503,15 @@ private void TraceReceipt(T ev) [LoggerMessage(Level = LogLevel.Warning, Message = "Event invalid. Id {EventId} Topic : {Topic}, StreamId : {StreamId}")] private partial void LogEventInvalid(string? eventId, RedisKey topic, RedisValue streamId); + [LoggerMessage(Level = LogLevel.Warning, Message = "Entry {StreamId} of topic {Topic} could not be decoded and was dropped ({Count} so far).")] + private partial void LogEventUndecodable(Exception? ex, RedisKey topic, RedisValue streamId, long count); + + [LoggerMessage(Level = LogLevel.Debug, Message = "Entry {StreamId} of topic {Topic} could not be decoded and was dropped")] + private partial void LogEventUndecodableAgain(Exception? ex, RedisKey topic, RedisValue streamId); + + [LoggerMessage(Level = LogLevel.Warning, Message = "Topic {Topic}: {Count} pending entries were deleted from the stream before this consumer read them; their invalidations are lost.")] + private partial void LogPendingEntriesDeleted(RedisKey topic, int count); + [LoggerMessage(Level = LogLevel.Error, Message = "OnMessage error. Topic : {Topic}")] private partial void LogOnMessageError(Exception ex, RedisKey topic); diff --git a/src/UiPath.Caching/Broadcast/Redis/StreamConstants.cs b/src/UiPath.Caching/Broadcast/Redis/StreamConstants.cs index 850e66e8..cbf1970b 100644 --- a/src/UiPath.Caching/Broadcast/Redis/StreamConstants.cs +++ b/src/UiPath.Caching/Broadcast/Redis/StreamConstants.cs @@ -22,8 +22,17 @@ internal static class StreamConstants /// Consecutive failures tolerated at the poll interval before backoff starts growing. internal const int ErrorBackoffThreshold = 3; + + /// One fetch timeout in this many is logged as a Warning, the rest at Debug. + internal const int TimeoutsPerWarning = 100; + + /// One undecodable entry in this many is logged as a Warning, the rest at Debug. + internal const int UndecodableEntriesPerWarning = 100; internal static readonly RedisValue UndeliveredMessages = ">"; + /// Read position of the entries delivered to a consumer but not acknowledged. + internal static readonly RedisValue PendingMessages = "0"; + /// Upper bound for the exponential backoff applied after consecutive fetch failures. internal static readonly TimeSpan MaxErrorBackoff = TimeSpan.FromSeconds(30); } diff --git a/tests/UiPath.Caching.Tests/Broadcast/RedisStreamHealthMaintainerIntegrationTests.cs b/tests/UiPath.Caching.Tests/Broadcast/RedisStreamHealthMaintainerIntegrationTests.cs index 544e440e..3ce03f49 100644 --- a/tests/UiPath.Caching.Tests/Broadcast/RedisStreamHealthMaintainerIntegrationTests.cs +++ b/tests/UiPath.Caching.Tests/Broadcast/RedisStreamHealthMaintainerIntegrationTests.cs @@ -189,6 +189,52 @@ public async Task A_stream_in_a_non_default_database_is_discovered_through_a_con } } + [Fact] + public async Task On_a_quiet_stream_a_caught_up_group_is_kept_and_a_group_behind_is_deleted() + { + Assert.SkipUnless(fixture.Enabled, "Set RUN_REDIS_INTEGRATION_TESTS=1 (Docker required) to run."); + + var cacheOptions = new CacheOptions { Enabled = true, AppShortName = $"it{Guid.NewGuid():N}"[..10] }; + var streamKey = new PrefixStrategy(RedisKeyspaces.Streams, cacheOptions).GetRedisKey("topic"); + var streamOptions = new RedisStreamsTopicOptions { MaintainerEnabled = true }; + var quiet = DateTimeOffset.UtcNow.Subtract(streamOptions.MaintainerTrimInterval).Subtract(TimeSpan.FromMinutes(5)); + + using var connector = NewConnector(); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + var db = connector.Database; + await db.StreamAddAsync(streamKey, "field", "value", $"{quiet.ToUnixTimeMilliseconds()}-0"); + await db.StreamAddAsync(streamKey, "field", "value", $"{quiet.ToUnixTimeMilliseconds()}-1"); + await db.StreamCreateConsumerGroupAsync(streamKey, "caught-up", "0-0"); + await db.StreamReadGroupAsync(streamKey, "caught-up", "pod", StreamPosition.NewMessages); + await db.StreamCreateConsumerGroupAsync(streamKey, "behind", "0-0"); + await db.StreamReadGroupAsync(streamKey, "behind", "pod", StreamPosition.NewMessages, count: 1); + var maintainer = new RedisStreamHealthMaintainer( + connector, + NullTelemetryProvider.Instance, + Options.Create(streamOptions), + Options.Create(new RedisCacheOptions { Enabled = true }), + Options.Create(cacheOptions), + NullLogger.Instance, + TimeProvider.System); + maintainer.Initialize(); + var quarantineKey = maintainer.QuarantineKey(streamKey); + + try + { + var elapsed = DateTimeOffset.UtcNow.Subtract(streamOptions.MaintainerQuarantineInterval).Subtract(TimeSpan.FromMinutes(1)).ToString("O"); + await db.HashSetAsync(quarantineKey, [new("caught-up", elapsed), new("behind", elapsed)]); + + await maintainer.CheckStreamsAsync(TestContext.Current.CancellationToken); + + (await db.StreamGroupInfoAsync(streamKey)).Select(g => g.Name).Should().Equal("caught-up"); + } + finally + { + await db.KeyDeleteAsync(quarantineKey); + await db.KeyDeleteAsync(streamKey); + } + } + private RedisConnector NewConnector() { var options = Options.Create(new RedisConnectionOptions { ConnectionString = fixture.ConnectionString, EnableHangDetection = false }); diff --git a/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterIntegrationTests.cs b/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterIntegrationTests.cs new file mode 100644 index 00000000..16d1edd3 --- /dev/null +++ b/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterIntegrationTests.cs @@ -0,0 +1,85 @@ +using System.Threading.Channels; +using Microsoft.Extensions.Logging.Abstractions; +using StackExchange.Redis; +using UiPath.Caching.Telemetry; +using UiPath.Caching.Tests.Redis; + +namespace UiPath.Caching.Tests.Broadcast; + +[Collection("RedisIntegration")] +[Trait("Category", "Integration")] +public class RedisStreamSubjectWriterIntegrationTests(RedisContainerFixture fixture) +{ + [Fact] + public async Task Entries_of_a_read_whose_answer_was_lost_are_dispatched_and_acknowledged() + { + Assert.SkipUnless(fixture.Enabled, "Set RUN_REDIS_INTEGRATION_TESTS=1 (Docker required) to run."); + + RedisKey stream = $"it:pending:{Guid.NewGuid():N}"; + RedisValue group = "group"; + RedisValue consumer = "consumer"; + RedisValue field = "event"; + using var connector = NewConnector(); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + var db = connector.Database; + using var stop = new CancellationTokenSource(); + + try + { + await db.StreamCreateConsumerGroupAsync(stream, group, StreamPosition.NewMessages, createStream: true); + await db.StreamAddAsync(stream, field, "first"); + await db.StreamAddAsync(stream, field, "second"); + var deleted = await db.StreamAddAsync(stream, field, "deleted"); + // A read whose answer never reached the client: Redis moved the group on and made the entries pending. + (await db.StreamReadGroupAsync(stream, group, consumer, StreamConstants.UndeliveredMessages, 10)).Should().HaveCount(3); + await db.StreamDeleteAsync(stream, [deleted]); + + var formatter = Substitute.For>(); + formatter.Decode(Arg.Any>()) + .Returns(call => call.Arg>().IsEmpty ? null : new TestCacheEvent { Valid = true, Source = new Uri("urn:other-source") }); + var connectionState = Substitute.For(); + connectionState.IsConnected.Returns(true); + var channel = Channel.CreateUnbounded(); + var pollInterval = TimeSpan.FromMilliseconds(50); + var context = new RedisStreamContext(stream, field, consumer, group, new Uri("urn:this-source"), 10, pollInterval, false, false); + + using var sut = new RedisStreamSubjectWriter( + context, + connectionState, + connector, + channel.Writer, + formatter, + NullLogger.Instance, + NullTelemetryProvider.Instance, + NullRedisProfiler.Instance, + new TimedFetchWaiter(pollInterval), + stop.Token); + + for (var i = 0; i < 2; i++) + { + await channel.Reader.ReadAsync(TestContext.Current.CancellationToken).AsTask().WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + } + + var deadline = DateTime.UtcNow.AddSeconds(10); + while ((await db.StreamPendingAsync(stream, group)).PendingMessageCount > 0 && DateTime.UtcNow < deadline) + { + await Task.Delay(20, TestContext.Current.CancellationToken); + } + + (await db.StreamPendingAsync(stream, group)).PendingMessageCount.Should().Be(0, "the deleted entry is acknowledged too"); + } + finally + { + await stop.CancelAsync(); + await db.KeyDeleteAsync(stream); + } + } + + private RedisConnector NewConnector() + { + var options = Options.Create(new RedisConnectionOptions { ConnectionString = fixture.ConnectionString, EnableHangDetection = false }); + var optionsProvider = new RedisConfigurationOptionsProvider(NullLoggerFactory.Instance, options); + var factory = new ConnectionMultiplexerFactory(options, NullRedisProfiler.Instance); + return new RedisConnector(NullTelemetryProvider.Instance, optionsProvider, factory, options); + } +} diff --git a/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterTests.cs b/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterTests.cs index d291480c..53c2fa06 100644 --- a/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterTests.cs +++ b/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterTests.cs @@ -577,6 +577,276 @@ public async Task Dispatch_failure_logs_error_with_event_and_stream_ids() await _database.DidNotReceiveWithAnyArgs().StreamAcknowledgeAsync(default, default, default(RedisValue[])!); } + [Fact] + public async Task Start_reads_the_consumers_pending_entries_once_before_new_ones() + { + var channel = Channel.CreateUnbounded(); + var pendingId = _fixture.Create(); + var positions = new ConcurrentQueue(); + var ev = new TestCacheEvent { Valid = true, Source = new Uri("urn:other-source") }; + _formatter.Decode(Arg.Any>()).Returns(ev); + // Unacknowledged on purpose: the pending entry stays where it is, so reading the pending list again would find it again. + _database.StreamReadGroupAsync(_context.Topic, _context.ConsumerGroup, _context.ConsumerName, ">", _context.PollBatchSize) + .ReturnsForAnyArgs(call => Read(positions, call, position => position == ">" ? [] : [Entry(pendingId)])); + + using var sut = CreateSut(channel.Writer, _logger, pollBatchSize: 2); + + var read = await channel.Reader.ReadAsync(TestContext.Current.CancellationToken).AsTask().WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + (await WaitUntil(() => positions.Count(p => p == ">") >= 3)).Should().BeTrue(); + _cancellationTokenSource.Cancel(); + await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + + read.Should().BeSameAs(ev); + positions.First().Should().Be("0", "entries delivered to this consumer before it started are read before new ones"); + positions.Count(p => p == "0").Should().Be(1, "a pending entry that stays pending must not be read over and over"); + } + + [Fact] + public async Task A_failed_read_is_followed_by_a_read_of_the_pending_entries() + { + var channel = Channel.CreateUnbounded(); + var lostId = _fixture.Create(); + var positions = new ConcurrentQueue(); + var failed = 0; + var ev = new TestCacheEvent { Valid = true, Source = new Uri("urn:other-source") }; + _formatter.Decode(Arg.Any>()).Returns(ev); + _database.StreamReadGroupAsync(_context.Topic, _context.ConsumerGroup, _context.ConsumerName, ">", _context.PollBatchSize) + .ReturnsForAnyArgs(call => Read(positions, call, position => position switch + { + ">" when Interlocked.Exchange(ref failed, 1) == 0 => throw Timeout(), + ">" => [], + // The entries the timed-out read was given. + _ when Volatile.Read(ref failed) == 1 => [Entry(lostId)], + _ => [], + })); + + using var sut = CreateSut(channel.Writer, _logger, pollBatchSize: 2); + + var read = await channel.Reader.ReadAsync(TestContext.Current.CancellationToken).AsTask().WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + _cancellationTokenSource.Cancel(); + await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + + read.Should().BeSameAs(ev); + positions.Take(3).Should().Equal("0", ">", "0"); + } + + [Fact] + public async Task A_full_batch_of_pending_entries_is_followed_by_the_entries_after_it() + { + var channel = Channel.CreateUnbounded(); + var positions = new ConcurrentQueue(); + _formatter.Decode(Arg.Any>()).Returns(new TestCacheEvent { Valid = true, Source = new Uri("urn:other-source") }); + _database.StreamReadGroupAsync(_context.Topic, _context.ConsumerGroup, _context.ConsumerName, ">", _context.PollBatchSize) + .ReturnsForAnyArgs(call => Read(positions, call, position => position switch + { + "0" => [Entry("1-0"), Entry("2-0")], + "2-0" => [Entry("3-0")], + _ => [], + })); + + using var sut = CreateSut(channel.Writer, _logger, pollBatchSize: 2); + + (await WaitUntil(() => positions.Contains(">"))).Should().BeTrue(); + _cancellationTokenSource.Cancel(); + await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + + positions.Should().StartWith(["0", "2-0", ">"]); + channel.Reader.Count.Should().Be(3); + } + + [Fact] + public async Task A_pending_entry_deleted_from_the_stream_is_acknowledged() + { + var acked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _database.StreamAcknowledgeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(call => { acked.TrySetResult(call.Arg()); return Task.FromResult(1L); }); + var positions = new ConcurrentQueue(); + var id = _fixture.Create(); + _database.StreamReadGroupAsync(_context.Topic, _context.ConsumerGroup, _context.ConsumerName, ">", _context.PollBatchSize) + .ReturnsForAnyArgs(call => Read(positions, call, position => position == "0" ? [new StreamEntry(id, [])] : [])); + var telemetry = new RecordingTelemetryProvider(); + var recordingLogger = new RecordingLogger(); + + using var sut = CreateSut(Channel.CreateUnbounded().Writer, recordingLogger, telemetry, pollBatchSize: 2); + + var ids = await acked.Task.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + _cancellationTokenSource.Cancel(); + await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + + ids.Select(v => v.ToString()).Should().Equal(id); + _formatter.DidNotReceiveWithAnyArgs().Decode(default); + telemetry.Events.Should().NotContain(e => e.Name.EndsWith(".UndecodableEvent", StringComparison.Ordinal), "a deleted entry is not a bad payload"); + recordingLogger.Records.Should().ContainSingle(r => r.Level == LogLevel.Warning && r.Message.Contains("deleted", StringComparison.Ordinal)); + } + + [Fact] + public async Task Read_timeouts_log_a_warning_once_per_hundred_and_never_an_error() + { + var recordingLogger = new RecordingLogger(); + var timeouts = 0; + _database.StreamReadGroupAsync(_context.Topic, _context.ConsumerGroup, _context.ConsumerName, ">", _context.PollBatchSize) + .ReturnsForAnyArgs(call => Read(new ConcurrentQueue(), call, position => + // Every other new-entry read times out; the successes between keep the backoff from growing. + position == ">" && Interlocked.Increment(ref timeouts) % 2 == 1 ? throw Timeout() : [])); + + using var sut = CreateSut(Channel.CreateUnbounded().Writer, recordingLogger, pollInterval: TimeSpan.FromMilliseconds(1)); + + (await WaitUntil(() => Volatile.Read(ref timeouts) >= 2 * 101 + 2)).Should().BeTrue(); + _cancellationTokenSource.Cancel(); + try { await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); } catch (OperationCanceledException) { } + + recordingLogger.Records.Should().NotContain(r => r.Level >= LogLevel.Error); + recordingLogger.Records.Count(r => r.Level == LogLevel.Warning && r.Message.Contains("timed out", StringComparison.Ordinal)) + .Should().Be(2, "the 1st and the 101st timeouts are reported"); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task An_entry_that_cannot_be_decoded_is_acknowledged(bool decodeThrows) + { + // Decoding the same bytes fails again, so left pending it would be read on every pass of the pending list. + var acked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _database.StreamAcknowledgeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(call => { acked.TrySetResult(call.Arg()); return Task.FromResult(1L); }); + _formatter.Decode(Arg.Any>()).Returns(_ => decodeThrows ? throw new FormatException("bad payload") : null); + var id = _fixture.Create(); + SetupSingleBatch([new StreamEntry(id, [new NameValueEntry(_fieldName, _fixture.Create())])]); + + using var sut = CreateSut(Channel.CreateUnbounded().Writer, _logger); + + var ids = await acked.Task.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + _cancellationTokenSource.Cancel(); + await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + + ids.Select(v => v.ToString()).Should().Equal(id); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task An_entry_that_cannot_be_decoded_is_tracked_and_logged_as_a_warning(bool decodeThrows) + { + var acked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _database.StreamAcknowledgeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(call => { acked.TrySetResult(call.Arg()); return Task.FromResult(1L); }); + _formatter.Decode(Arg.Any>()).Returns(_ => decodeThrows ? throw new FormatException("bad payload") : null); + var id = _fixture.Create(); + SetupSingleBatch([new StreamEntry(id, [new NameValueEntry(_fieldName, _fixture.Create())])]); + var telemetry = new RecordingTelemetryProvider(); + var recordingLogger = new RecordingLogger(); + + using var sut = CreateSut(Channel.CreateUnbounded().Writer, recordingLogger, telemetry); + + await acked.Task.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + _cancellationTokenSource.Cancel(); + await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + + telemetry.Events.Should().ContainSingle(e => e.Name == "Caching.RedisStreamSubjectWriter.DispatchEventsAsync.UndecodableEvent") + .Which.Properties.Should().Contain("TopicKey", _context.Topic.ToString()).And.Contain("TransportId", id); + recordingLogger.Records.Should().ContainSingle(r => r.Level == LogLevel.Warning && r.Message.Contains(id, StringComparison.Ordinal)); + recordingLogger.Records.Should().NotContain(r => r.Level >= LogLevel.Error, "a bad entry is dropped, not a failure of the reader"); + } + + [Fact] + public async Task Undecodable_entries_are_each_tracked_but_logged_as_a_warning_once_per_hundred() + { + var acked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _database.StreamAcknowledgeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(call => { acked.TrySetResult(call.Arg()); return Task.FromResult(1L); }); + _formatter.Decode(Arg.Any>()).Returns((ICacheEvent?)null); + SetupSingleBatch([.. Enumerable.Range(1, 101).Select(i => new StreamEntry($"{i}-0", [new NameValueEntry(_fieldName, _fixture.Create())]))]); + var telemetry = new RecordingTelemetryProvider(); + var recordingLogger = new RecordingLogger(); + + using var sut = CreateSut(Channel.CreateUnbounded().Writer, recordingLogger, telemetry); + + (await acked.Task.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken)).Should().HaveCount(101); + _cancellationTokenSource.Cancel(); + await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + + telemetry.Events.Count(e => e.Name.EndsWith(".UndecodableEvent", StringComparison.Ordinal)).Should().Be(101); + recordingLogger.Records.Count(r => r.Level == LogLevel.Warning).Should().Be(2, "the 1st and the 101st are reported"); + } + + [Fact] + public async Task An_undecodable_entry_is_acknowledged_when_the_telemetry_sink_refuses_the_record() + { + var acked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _database.StreamAcknowledgeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(call => { acked.TrySetResult(call.Arg()); return Task.FromResult(1L); }); + _formatter.Decode(Arg.Any>()).Returns((ICacheEvent?)null); + var id = _fixture.Create(); + SetupSingleBatch([new StreamEntry(id, [new NameValueEntry(_fieldName, _fixture.Create())])]); + var telemetry = new RefusingTelemetryProvider("Caching.RedisStreamSubjectWriter.DispatchEventsAsync.UndecodableEvent"); + var recordingLogger = new RecordingLogger(); + + using var sut = CreateSut(Channel.CreateUnbounded().Writer, recordingLogger, telemetry); + + var ids = await acked.Task.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + _cancellationTokenSource.Cancel(); + await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + + ids.Select(v => v.ToString()).Should().Equal(id); + telemetry.Exceptions.Should().Contain(RefusingTelemetryProvider.Failure); + } + + [Fact] + public async Task A_pending_batch_with_a_deleted_entry_warns_once_and_still_dispatches_the_rest() + { + var acked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _database.StreamAcknowledgeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(call => { acked.TrySetResult(call.Arg()); return Task.FromResult(1L); }); + var channel = Channel.CreateUnbounded(); + var ev = new TestCacheEvent { Valid = true, Source = new Uri("urn:other-source") }; + _formatter.Decode(Arg.Any>()).Returns(ev); + _database.StreamReadGroupAsync(_context.Topic, _context.ConsumerGroup, _context.ConsumerName, ">", _context.PollBatchSize) + .ReturnsForAnyArgs(call => Read(new ConcurrentQueue(), call, position => position == "0" ? [new StreamEntry("1-0", []), Entry("2-0")] : [])); + var recordingLogger = new RecordingLogger(); + + using var sut = CreateSut(channel.Writer, recordingLogger, pollBatchSize: 4); + + var ids = await acked.Task.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + _cancellationTokenSource.Cancel(); + await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + + ids.Select(v => v.ToString()).Should().BeEquivalentTo(["1-0", "2-0"]); + recordingLogger.Records.Count(r => r.Level == LogLevel.Warning && r.Message.Contains("deleted", StringComparison.Ordinal)).Should().Be(1); + channel.Reader.TryRead(out var read).Should().BeTrue(); + read.Should().BeSameAs(ev); + } + + [Fact] + public async Task A_deleted_pending_entry_is_logged_once_after_its_acknowledgement_succeeds() + { + var acks = 0; + var acked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _database.StreamAcknowledgeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(_ => + { + if (Interlocked.Increment(ref acks) == 1) + { + return Task.FromException(Timeout()); + } + + acked.TrySetResult(); + return Task.FromResult(1L); + }); + _database.StreamReadGroupAsync(_context.Topic, _context.ConsumerGroup, _context.ConsumerName, ">", _context.PollBatchSize) + .ReturnsForAnyArgs(call => Read(new ConcurrentQueue(), call, position => position == "0" ? [new StreamEntry("1-0", [])] : [])); + var recordingLogger = new RecordingLogger(); + + using var sut = CreateSut(Channel.CreateUnbounded().Writer, recordingLogger, pollBatchSize: 4); + + await acked.Task.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + (await WaitUntil(() => recordingLogger.Records.Any(r => r.Message.Contains("deleted", StringComparison.Ordinal)))).Should().BeTrue(); + await Task.Delay(100, TestContext.Current.CancellationToken); + _cancellationTokenSource.Cancel(); + try { await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); } catch (OperationCanceledException) { } + + recordingLogger.Records.Count(r => r.Message.Contains("deleted", StringComparison.Ordinal)).Should().Be(1, "the first acknowledgement failed, so only the pass whose acknowledgement succeeded logs"); + } + public ValueTask DisposeAsync() { return ValueTask.CompletedTask; @@ -628,6 +898,15 @@ private static async Task WaitUntil(Func condition) private static RedisServerException UnknownCommandError(string message) => new(RedisErrorKind.UnknownCommand, CommandFlags.None, message); + private static RedisTimeoutException Timeout() => new(CommandFlags.None, "Timeout performing XREADGROUP", CommandStatus.Sent); + + private static Task Read(ConcurrentQueue positions, NSubstitute.Core.CallInfo call, Func answer) + { + var position = call.ArgAt(3).ToString()!; + positions.Enqueue(position); + return Task.FromResult(answer(position)); + } + private void SetupSingleBatch(StreamEntry[] entries) { var emitted = 0; @@ -635,14 +914,21 @@ private void SetupSingleBatch(StreamEntry[] entries) .ReturnsForAnyArgs(_ => ++emitted == 1 ? entries : []); } - private RedisStreamSubjectWriter CreateSut(ChannelWriter writer, ILogger logger, ICachingTelemetryProvider? telemetry = null) + private StreamEntry Entry(string id) => new(id, [new NameValueEntry(_fieldName, _fixture.Create())]); + + private RedisStreamSubjectWriter CreateSut(ChannelWriter writer, ILogger logger, ICachingTelemetryProvider? telemetry = null, int? pollBatchSize = null, TimeSpan? pollInterval = null) { var connectionState = _fixture.Create(); connectionState.IsConnected.Returns(true); var redis = _fixture.Create(); redis.Database.Returns(_database); + var context = _context with + { + PollBatchSize = pollBatchSize ?? _context.PollBatchSize, + PollInterval = pollInterval ?? _context.PollInterval, + }; return new RedisStreamSubjectWriter( - _context, + context, connectionState, redis, writer, @@ -650,7 +936,7 @@ private RedisStreamSubjectWriter CreateSut(ChannelWriter(), _fixture.Create(), - new TimedFetchWaiter(_pollInterval), + new TimedFetchWaiter(context.PollInterval), _cancellationTokenSource.Token); } diff --git a/tests/UiPath.Caching.Tests/Broadcast/RedisStreamTopicMonitorTests.cs b/tests/UiPath.Caching.Tests/Broadcast/RedisStreamTopicMonitorTests.cs index 5c260280..e6b5138f 100644 --- a/tests/UiPath.Caching.Tests/Broadcast/RedisStreamTopicMonitorTests.cs +++ b/tests/UiPath.Caching.Tests/Broadcast/RedisStreamTopicMonitorTests.cs @@ -330,6 +330,87 @@ await _database.Received().HashDeleteAsync( CommandFlags.DemandMaster); } + [Fact] + public async Task Group_with_consumers_caught_up_on_a_quiet_stream_is_not_quarantined() + { + var quiet = DateTimeOffset.UtcNow.Subtract(_streamOptions.MaintainerTrimInterval).Subtract(TimeSpan.FromMinutes(5)); + var g1 = GenerateGroupInfo(quiet, consumerCount: 1); + _streamGroupInfos = [g1]; + _lastGeneratedId = g1.LastDeliveredId!; + _quarantineValue = RedisValue.Null; + + Sut.Initialize(); + await Sut.CheckStreamsAsync(testContextAccessor.Current.CancellationToken); + + await _database.DidNotReceive().HashSetAsync( + Arg.Any(), + Arg.Is(v => v == g1.Name), + Arg.Any(), + Arg.Any(), + Arg.Any()); + } + + [Fact] + public async Task Group_with_consumers_caught_up_on_a_quiet_stream_leaves_quarantine_instead_of_being_deleted() + { + var quiet = DateTimeOffset.UtcNow.Subtract(_streamOptions.MaintainerTrimInterval).Subtract(TimeSpan.FromMinutes(5)); + var g1 = GenerateGroupInfo(quiet, consumerCount: 1); + _streamGroupInfos = [g1]; + _lastGeneratedId = g1.LastDeliveredId!; + _quarantineValue = DateTimeOffset.UtcNow.Subtract(_streamOptions.MaintainerQuarantineInterval).Subtract(TimeSpan.FromMinutes(1)).ToString("O"); + + Sut.Initialize(); + await Sut.CheckStreamsAsync(testContextAccessor.Current.CancellationToken); + + await _database.DidNotReceive().StreamDeleteConsumerGroupAsync(Arg.Any(), Arg.Any(), Arg.Any()); + await _database.Received().HashDeleteAsync( + Arg.Any(), + Arg.Is(v => v == g1.Name), + CommandFlags.DemandMaster); + } + + [Fact] + public async Task Group_with_consumers_behind_by_an_entry_of_the_same_millisecond_is_still_quarantined() + { + var stale = DateTimeOffset.UtcNow.Subtract(_streamOptions.MaintainerTrimInterval).Subtract(TimeSpan.FromMinutes(5)); + var g1 = GenerateGroupInfo(stale, consumerCount: 1); + _streamGroupInfos = [g1]; + _lastGeneratedId = $"{stale.ToUnixTimeMilliseconds()}-1"; + _quarantineValue = RedisValue.Null; + + Sut.Initialize(); + await Sut.CheckStreamsAsync(testContextAccessor.Current.CancellationToken); + + await _database.Received().HashSetAsync( + Arg.Any(), + Arg.Is(v => v == g1.Name), + Arg.Any(), + When.Always, + CommandFlags.DemandMaster); + } + + [Theory] + [InlineData("")] + [InlineData("not-an-id")] + public async Task Group_with_consumers_and_stale_LastDeliveredId_is_quarantined_when_the_stream_head_is_unreadable(string lastGeneratedId) + { + var stale = DateTimeOffset.UtcNow.Subtract(_streamOptions.MaintainerTrimInterval).Subtract(TimeSpan.FromMinutes(5)); + var g1 = GenerateGroupInfo(stale, consumerCount: 1); + _streamGroupInfos = [g1]; + _lastGeneratedId = lastGeneratedId; + _quarantineValue = RedisValue.Null; + + Sut.Initialize(); + await Sut.CheckStreamsAsync(testContextAccessor.Current.CancellationToken); + + await _database.Received().HashSetAsync( + Arg.Any(), + Arg.Is(v => v == g1.Name), + Arg.Any(), + When.Always, + CommandFlags.DemandMaster); + } + [Fact] public async Task Initialize_uses_MaintainerSearchPattern_when_configured() {