From af009b7cc2f99f7228ef52f58147184f382f369c Mon Sep 17 00:00:00 2001 From: Cosmin Staicu Date: Sat, 24 Oct 2026 08:45:48 +0300 Subject: [PATCH 1/2] fix(broadcast): keep a caught-up consumer group on a quiet stream The maintainer counted a group with consumers as stale once its last delivery was older than MaintainerTrimInterval, without checking whether anything newer had been written. On a stream with no traffic every live group was quarantined, deleted after MaintainerQuarantineInterval and recreated by its reader on NOGROUP. A group whose last-delivered-id has reached the stream's last-generated-id is now healthy. Signed-off-by: Cosmin Staicu --- CHANGELOG.md | 6 ++ .../Redis/RedisStreamHealthMaintainer.cs | 28 +++++-- ...sStreamHealthMaintainerIntegrationTests.cs | 46 +++++++++++ .../Broadcast/RedisStreamTopicMonitorTests.cs | 81 +++++++++++++++++++ 4 files changed, 155 insertions(+), 6 deletions(-) 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/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() { From 124e306485ad7893b60757270e3d15be705025ac Mon Sep 17 00:00:00 2001 From: Cosmin Staicu Date: Sat, 24 Oct 2026 10:29:51 +0300 Subject: [PATCH 2/2] test: hold the reader's retry until the test has seen the failed recreate A_subscriber_joining_while_the_readers_recreate_is_pending_... waited for the state between the failed recreate and the retry, which lasts about one poll interval (50 ms). WaitUntil polls every 20 ms, so a slow runner could miss the whole window. The retry's read now waits until the test has checked that state and the second subscriber has joined. Signed-off-by: Cosmin Staicu --- .../Broadcast/RedisStreamSubjectWriterTests.cs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) 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]