[queue] create PendingCount (to replace Lag) #160
Merged
Conversation
This file contains 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
This adds a new PendingCount command which returns the aggregate number of pending messages in a queue. ----- I tried working with Lag but it has the problem that it is fundamentally heuristic, based on keeping a track of two counters: - how many messages have ever been added to the stream - how many messages have ever been read by the given consumer group "Lag" is then the difference between these counters - the number of messages that have been added to the stream but not read by the consumer group. Unfortunately, these counters can desync horribly if you ever XDEL a message before it gets read by a consumer group. This is because deleting a message does not decrement the "added to stream" counter, nor does it increment the "read by consumer group" counter. Currently we rely on being able to XDEL messages. Fixing that would be nontrivial. But we can measure the size of the PEL with XPENDING, and we can measure the length of the stream with XLEN, so we can calculate lag as PendingCount() - Len().
This should be |
evilstreak
approved these changes
Dec 9, 2024
There was a problem hiding this comment.
Thinking about your PR description, this has the opposite problem in that it requires we do call XDEL on everything after we're done with it. But since that's the behaviour of our system currently, and that behaviour is relied on by our autoscaler for working out how many pods to run, I think that's fine!
er.. yes |
philandstuff
added a commit
that referenced
this pull request
Dec 10, 2024
This is just like #160, but we calculate length and pending count atomically. Before this change, I was subtracting the pending count from the length to get a "waiting messages" metric; but as these numbers could not be measured atomically I sometimes saw negative waiting messages. This changes PendingCount() into Stats() which returns both length and pending count.
philandstuff
added a commit
that referenced
this pull request
Dec 10, 2024
This is just like #160, but we calculate length and pending count atomically. Before this change, I was subtracting the pending count from the length to get a "waiting messages" metric; but as these numbers could not be measured atomically I sometimes saw negative waiting messages. This changes PendingCount() into Stats() which returns both length and pending count.
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.
This adds a new PendingCount command which returns the aggregate number of pending messages in a queue.
I tried working with Lag but it has the problem that it is fundamentally heuristic, based on keeping a track of two counters:
"Lag" is then the difference between these counters - the number of messages that have been added to the stream but not read by the consumer group.
Unfortunately, these counters can desync horribly if you ever XDEL a message before it gets read by a consumer group. This is because deleting a message does not decrement the "added to stream" counter, nor does it increment the "read by consumer group" counter.
Currently we rely on being able to XDEL messages. Fixing that would be nontrivial.
But we can measure the size of the PEL with XPENDING, and we can measure the length of the stream with XLEN, so we can calculate lag as Len() - PendingCount().