[cosmin-staicu]

One PR for the stream reader fixes on main. It folds in #290 (caught-up consumer groups) and #296 (the Limit docs), which are closed.

Pending entries after a failed read

When an XREADGROUP times out on the client, Redis has already moved the group past the entries and put them in the consumer's pending list. The reader only ever read new entries (>) and never read its pending list, so those entries were never dispatched. Their invalidations were lost, and the node kept the old values in its local tier until they expired.

Changes in RedisStreamSubjectWriter:

• Pending read: when it starts, and after every failed read, the reader reads its own pending entries from 0 before going back to >. A full batch continues from its last ID rather than restarting from 0, so an entry that never gets acknowledged is not read in a loop. The consumer name (the SourceUri) stays the same for the writer's whole life, so the pending list it reads is its own, as long as the SourceUri is unique per process.
• Gap check: the pending read is a plain XREADGROUP and doesn't move the group, so it skips the gap check. If it fails, it is handled the same way as any other failed read.
• Deleted entries: a pending entry trimmed or deleted from the stream before it was read comes back with no fields. Its invalidation is lost, and the gap check can't see it because it is behind the group's position. It is acknowledged, logged as a Warning, and reported as missed, which clears the local tier for the topic.
• Undecodable entries: an entry that decodes to null, or whose decode throws, is acknowledged and dropped. It is 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 on every node. Each one is tracked as the Caching.RedisStreamSubjectWriter.DispatchEventsAsync.UndecodableEvent telemetry event and logged as a Warning once per 100, at Debug otherwise.
• Timeouts: a read timeout is logged as a Warning once per 100, at Debug otherwise, instead of as an Error every time.

Not changed: the background poll still uses the connection's request timeout. StackExchange.Redis has no per-command timeout on a shared connection.

Tests:

• Unit tests:
  • pending entries are read first, and only once
  • a failed read is followed by a pending read, both before and after the reader is in use (the script read path)
  • a full pending batch continues from its last ID
  • deleted and undecodable pending entries are acknowledged
  • undecodable entries are tracked, rate-limited, and acknowledged even when the telemetry sink throws
  • timeouts give a Warning once per 100 and never an Error
• RedisStreamSubjectWriterIntegrationTests: entries handed out by a read whose answer was thrown away are dispatched and acknowledged against a real Redis.
• Ten existing tests counted or sequenced every XREADGROUP call. Their read mocks now go through StreamReads.NewOnly, which answers pending reads with nothing. What they assert is unchanged.
• Each weaker version fails at least one test: no pending read at start, no pending read after a failure, re-reading from 0 until empty, deleted entries not acknowledged, timeouts still logged as errors, warning on every undecodable entry.
• The full suite passes on net8.0 and net10.0 with the Redis integration tests on. The Release -warnaserror build is clean.

Caught-up consumer groups on a quiet stream

The stream maintainer counted a consumer group with consumers as stale once its last-delivered-id was older than MaintainerTrimInterval. It never checked 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 the next NOGROUP. That repeated every cycle, with warnings, LogRecreatingTopic, and XGROUP calls that sometimes timed out.

A group whose last-delivered-id has reached the stream's last-generated-id is now healthy, and a quarantine it was under is cleared.

• Both parts of the ID are compared, so a group behind by one entry in the same millisecond still counts as behind.
• If either ID can't be parsed, the old rule applies.
• A caught-up group whose consumer is dead is no longer removed on a quiet stream. It is quarantined and deleted as before once the stream gets a new entry.

Tests:

• Unit tests: a caught-up group is not quarantined; a caught-up group under an elapsed quarantine is released, not deleted; a group behind in the same millisecond is still quarantined; a stream head that can't be read keeps the old rule.
• Integration test against Redis: on a quiet stream, a caught-up group is kept and a group that is behind is deleted.
• The tests fail with the old code, and with a check that compares only milliseconds.
• The full suite passes on net8.0 and net10.0 with the Redis integration tests on. The Release -warnaserror build is clean.

Limit docs

The settings tables in the broadcast guide and the settings reference, and the sample's full settings file, still described RedisStreamsTopicOptions.Limit as the number of entries one read returns. Nothing reads Limit: the reader asks for PollBatchSize entries, and the publisher trims with MaxLength only. #294 fixed the guide's prose. This fixes the three remaining places, so the docs no longer contradict each other.

Docs only.

Flaky test

A_subscriber_joining_while_the_readers_recreate_is_pending_... waited for a state that lasts about one poll interval while WaitUntil polls every 20 ms, so a slow runner could miss it. The reader's retry now waits until the test has seen that state.

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
100.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud