cosmin-staicu · Oct 26, 2026 · Oct 24, 2026
@@ -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

@@ -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
 

@@ -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. |

@@ -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,

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

@@ -7,6 +7,7 @@ internal sealed partial class RedisStreamSubjectWriter<T> : IDisposable
     where T : IEvent
 {
     private const string EventInvalid = "Caching." + nameof(RedisStreamSubjectWriter<T>) + "." + nameof(DispatchEventsAsync) + ".InvalidEvent";
+    private const string EventUndecodable = "Caching." + nameof(RedisStreamSubjectWriter<T>) + "." + nameof(DispatchEventsAsync) + ".UndecodableEvent";
     private const string EventReceived = "Caching." + nameof(RedisStreamSubjectWriter<T>) + "." + nameof(DispatchEventsAsync) + ".EventReceived";
     private const string PropTopicKey = "TopicKey";
     private const string PropTransportId = "TransportId";
@@ -26,6 +27,10 @@ internal sealed partial class RedisStreamSubjectWriter<T> : 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);
     }
 
+    /// <summary>Reads the entries Redis gave this consumer without an acknowledgement, such as those of a read that timed out on the client.</summary>
+    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<StreamEntry[]> ReadAsync(RedisValue position) => _redis.Database.StreamReadGroupAsync(
+        _context.Topic,
+        _context.ConsumerGroup,
+        _context.ConsumerName,
+        position,
+        _context.PollBatchSize);
+
     private async Task<bool> ProcessException(Exception ex)
     {
         var id = _lastId;
@@ -181,6 +215,8 @@ private async Task<bool> ProcessException(Exception ex)
             return false;
         }
 
+        _pendingFrom = StreamConstants.PendingMessages;
+
         if (IsUnsupportedCommand(ex))
         {
             if (!_unsupportedCommand)
@@ -216,13 +252,29 @@ private async Task<bool> 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<RedisValue> 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<byte>)@event[_context.FieldName]);
+            T? ev;
+            try
+            {
+                ev = _formatter.Decode((ReadOnlyMemory<byte>)@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<RedisValue> 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<RedisValue> ids)
         LogEventInvalid(ev.Id, _context.Topic, @event.Id);
     }
 
+    /// <summary>Drops an entry no retry would decode: acknowledged, since a bad payload is not a lost invalidation.</summary>
+    private void HandleUndecodableEvent(StreamEntry @event, List<RedisValue> 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);
 

@@ -22,8 +22,17 @@ internal static class StreamConstants
 
     /// <summary>Consecutive failures tolerated at the poll interval before backoff starts growing.</summary>
     internal const int ErrorBackoffThreshold = 3;
+
+    /// <summary>One fetch timeout in this many is logged as a Warning, the rest at Debug.</summary>
+    internal const int TimeoutsPerWarning = 100;
+
+    /// <summary>One undecodable entry in this many is logged as a Warning, the rest at Debug.</summary>
+    internal const int UndecodableEntriesPerWarning = 100;
     internal static readonly RedisValue UndeliveredMessages = ">";
 
+    /// <summary>Read position of the entries delivered to a consumer but not acknowledged.</summary>
+    internal static readonly RedisValue PendingMessages = "0";
+
     /// <summary>Upper bound for the exponential backoff applied after consecutive fetch failures.</summary>
     internal static readonly TimeSpan MaxErrorBackoff = TimeSpan.FromSeconds(30);
 }