diff --git a/CHANGELOG.md b/CHANGELOG.md index 98bb2d64..c37024a6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -224,6 +224,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) ### Fixed +- **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. + - **Multi-key calls over keys of several Redis Cluster slots run, one command per slot, instead of throwing `CrossSlotKeysException`.** `GetAsync(CacheKey[])`, `GetCacheEntriesAsync`, the batch `SetAsync` and `RemoveAsync(CacheKey[])` refused such a batch, and with them the batch `GetOrAddAsync`, which reads its keys with `GetCacheEntriesAsync` and stores 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/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/RedisStreamSubjectWriterTests.cs b/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterTests.cs index a0eebef0..a88426a8 100644 --- a/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterTests.cs +++ b/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterTests.cs @@ -1141,12 +1141,15 @@ public async Task A_subscriber_joining_while_the_readers_recreate_is_pending_is_ var reads = 0; var missed = 0; var creates = 0; + // The retry waits for the test: the state between the failed recreate and the retry lasts only a poll interval. + var retry = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); _formatter.Decode(Arg.Any>()).Returns(_ => new TestCacheEvent { Valid = true, Source = new Uri("urn:other") }); _database.StreamReadGroupAsync(_context.Topic, _context.ConsumerGroup, _context.ConsumerName, ">", _context.PollBatchSize) .ReturnsForAnyArgs(_ => Interlocked.Increment(ref reads) switch { 1 => Task.FromResult([new StreamEntry("1-0", [new NameValueEntry(_fieldName, "a")])]), - _ when Volatile.Read(ref creates) < 2 => Task.FromException(new RedisException("NOGROUP No such key or consumer group")), + _ when Volatile.Read(ref creates) == 0 => Task.FromException(new RedisException("NOGROUP No such key or consumer group")), + _ when Volatile.Read(ref creates) == 1 => NogroupOnRetry(), _ => Task.FromResult(Array.Empty()), }); _database.StreamCreateConsumerGroupAsync(_context.Topic, _context.ConsumerGroup, Arg.Any()) @@ -1159,11 +1162,18 @@ _ when Volatile.Read(ref creates) < 2 => Task.FromException(new R (await WaitUntil(() => Volatile.Read(ref missed) == 1 && Volatile.Read(ref creates) == 1)).Should().BeTrue("the group went while in use and its recreate failed"); sut.MarkSubscribed(); + retry.SetResult(); (await WaitUntil(() => Volatile.Read(ref creates) == 2)).Should().BeTrue("the reader retries"); await _database.Received().StreamCreateConsumerGroupAsync(_context.Topic, _context.ConsumerGroup, (RedisValue)"1-0"); Volatile.Read(ref missed).Should().Be(1, "what came since the last read is replayed to the group the reader made, not skipped"); AssertNoSynchronousCommand(); + + async Task NogroupOnRetry() + { + await retry.Task; + throw new RedisException("NOGROUP No such key or consumer group"); + } } [Fact] 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() {