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() {