fix(broadcast): keep a caught-up consumer group on a quiet stream

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

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

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<RedisStreamHealthMaintainer>.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 });

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<ReadOnlyMemory<byte>>()).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<StreamEntry[]>([new StreamEntry("1-0", [new NameValueEntry(_fieldName, "a")])]),
-                _ when Volatile.Read(ref creates) < 2 => Task.FromException<StreamEntry[]>(new RedisException("NOGROUP No such key or consumer group")),
+                _ when Volatile.Read(ref creates) == 0 => Task.FromException<StreamEntry[]>(new RedisException("NOGROUP No such key or consumer group")),
+                _ when Volatile.Read(ref creates) == 1 => NogroupOnRetry(),
                 _ => Task.FromResult(Array.Empty<StreamEntry>()),
             });
         _database.StreamCreateConsumerGroupAsync(_context.Topic, _context.ConsumerGroup, Arg.Any<RedisValue?>())
@@ -1159,11 +1162,18 @@ _ when Volatile.Read(ref creates) < 2 => Task.FromException<StreamEntry[]>(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<StreamEntry[]> NogroupOnRetry()
+        {
+            await retry.Task;
+            throw new RedisException("NOGROUP No such key or consumer group");
+        }
     }
 
     [Fact]

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<RedisKey>(),
+            Arg.Is<RedisValue>(v => v == g1.Name),
+            Arg.Any<RedisValue>(),
+            Arg.Any<When>(),
+            Arg.Any<CommandFlags>());
+    }
+
+    [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<RedisKey>(), Arg.Any<RedisValue>(), Arg.Any<CommandFlags>());
+        await _database.Received().HashDeleteAsync(
+            Arg.Any<RedisKey>(),
+            Arg.Is<RedisValue>(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<RedisKey>(),
+            Arg.Is<RedisValue>(v => v == g1.Name),
+            Arg.Any<RedisValue>(),
+            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<RedisKey>(),
+            Arg.Is<RedisValue>(v => v == g1.Name),
+            Arg.Any<RedisValue>(),
+            When.Always,
+            CommandFlags.DemandMaster);
+    }
+
     [Fact]
     public async Task Initialize_uses_MaintainerSearchPattern_when_configured()
     {