Repository navigation
fix(broadcast): keep a caught-up consumer group on a quiet stream - #290
Closed
cosmin-staicu wants to merge 2 commits into
Closed
cosmin-staicu wants to merge 2 commits into
cosmin-staicu wants to merge 2 commits into
Conversation
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 <cosmin.staicu@uipath.com>
cosmin-staicu
requested review from
alinahornet,
cosminvlad,
florin-munteanu-uipath,
litheon,
lucianaparaschivei and
razvalex
as code owners
October 24, 2026 07:11
…eate 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 <cosmin.staicu@uipath.com>
|
This was referenced Oct 24, 2026
Member
Author
|
Folded into #291. |
cosmin-staicu
added a commit
that referenced
this pull request
Oct 26, 2026
…caught-up groups (2.1) (#293) One PR for the stream reader fixes on 2.1.x, the backport of #291. It folds in #292 (caught-up consumer groups) and #297 (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. - **Deleted entries:** a pending entry trimmed or deleted from the stream before it was read comes back with no fields. Its invalidation is lost. It is acknowledged and logged as a Warning; 2.1 has no missed-message signal to raise. - **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 - 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. - 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. The undecodable-entry rate limit was red-checked on #291. - The full suite passes on net8.0 and net10.0 with the Redis integration tests on. The Release `-warnaserror` build is clean. Differences from #291, where 2.1's code differs: - 2.1 has no gap-check script, so new entries are always read with a plain `XREADGROUP`, and the pending read has no gap-check state to keep in step. - No existing 2.1 test counted every read, so 2.1 needs no `StreamReads` helper. ## 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. The code is the same as #290. Only the CHANGELOG entry differs, because it goes into 2.1's own Unreleased section. ## `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 (main) fixed the guide's prose. This fixes the three remaining places, so the docs no longer contradict each other. Docs only. Signed-off-by: Cosmin Staicu <cosmin.staicu@uipath.com>
cosmin-staicu
added a commit
that referenced
this pull request
Oct 26, 2026
…caught-up groups (#291) 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. Signed-off-by: Cosmin Staicu <cosmin.staicu@uipath.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
The stream maintainer counted a consumer group with consumers as stale once its
last-delivered-idwas older thanMaintainerTrimInterval. It never checked whether anything newer had been written. On a stream with no traffic, every live group was quarantined, deleted afterMaintainerQuarantineInterval, and recreated by its reader on the nextNOGROUP. That repeated every cycle, with warnings,LogRecreatingTopic, andXGROUPcalls that sometimes timed out.A group whose
last-delivered-idhas reached the stream'slast-generated-idis now healthy, and a quarantine it was under is cleared.Tests:
-warnaserrorbuild is clean.