XACKDEL Command - #3466
XACKDEL Command#3466nickiaq wants to merge 7 commits into
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## unstable #3466 +/- ##
============================================
+ Coverage 76.68% 79.87% +3.19%
============================================
Files 162 187 +25
Lines 81021 95554 +14533
============================================
+ Hits 62129 76325 +14196
- Misses 18892 19229 +337
🚀 New features to boost your workflow:
|
|
@zuiderkwast Thanks for updating this to the latest |
I did it just to trigger the "Provenance Guard" CI job. It checks for code from Redis, which is under an incompatible license. It looks green. 👍 I hope we'll get some time to review it in the next weeks. Now, we're finishing the 9.1 release, so this will target the next version. |
|
Thanks for the update. I'll watch out for any review comments to address whenever you have a chance to look. I don't see a 9.2 project on the projects tab, just 10. So it looks like |
madolson
added
major-decision-approved
Initial implementation of XACKDEL command, which combines ACK'ing a stream message with deleting the message and several modes for how to manage other consumer group's PELs. Modes include: - KEEPREF (default): leave references in other consumer groups' PELs - DELREF: delete references from other consumer groups' PELs - ACKED: only delete the message if all consumer groups have ACKED The syntax is as follows: XACKDEL <key> <group> [ KEEPREF | DELREF | ACKED ] IDS <numids> <id> [<id> ...] The response is an array of status codes. One status code for each ID passed in to XACKDEL. The codes are as follows: - -1 for when the message does not exist or is not in the consumer group's PEL. - 1 for acked and deleted - 2 for acked and not deleted (in ACKED mode) Signed-off-by: Nick Iaquinto <git+valkey@iaquinto.io>
Optimize XACKDEL implementation by iterating per-consumer first then per-message for ACKED and DELREF modes. This is faster because looping over consumers is slower than looping over messages. So doing consumers first means we only loop over those once. And it allows skipping the inner-loop iterations once a final response is determined for a message For KEEPREF, there's a separate branch that skips the consumer loop entirely, since KEEPREF only needs to clean up the target group's PEL. Signed-off-by: Nick Iaquinto <git+valkey@iaquinto.io>
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Advanced Run ID: 📒 Files selected for processing (1)
Included review availability: Your plan provides up to 10 included reviews per hour; 8 remain after this review. 📝 WalkthroughWalkthroughAdds the XACKDEL stream command, its command metadata and handler, and unit and replication tests. The command supports KEEPREF, ACKED, and DELREF modes, validates IDs and counts, updates stream and consumer-group state, and returns per-ID status codes. ChangesXACKDEL stream command
Priority: ⬇️ Low Estimated code review effort: 4 (Complex) | ~60 minutes Merge Risk: ⚪ Minimal · up to XACKDEL adds conditional stream acknowledgment and deletion modes with coverage for parsing, consumer-group state changes, and replication behavior. No concrete current-head merge-blocking risk remains. Sequence Diagram(s)sequenceDiagram
participant client
participant xackdelCommand
participant consumerGroupPELs
participant streamState
participant keyspaceNotifications
client->>xackdelCommand: Submit XACKDEL arguments and IDs
xackdelCommand->>consumerGroupPELs: Remove or inspect pending entries
xackdelCommand->>streamState: Delete eligible entries and update stream edges
xackdelCommand->>keyspaceNotifications: Signal stream deletion changes
xackdelCommand->>client: Return per-ID status codes
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
The force push a few seconds ago was just keeping this up with unstable. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/commands/xackdel.json`:
- Around line 48-70: The mode argument definition for XACKDEL is advertising an
extra MODE keyword that does not exist in the handler or Redis 8.2 syntax.
Update the `mode` oneof in the `xackdel` command spec to remove the `token`
value and keep only `KEEPREF`, `DELREF`, and `ACKED`, then regenerate the
generated command artifacts so `XACKDEL_Args` and the docs/definitions match
`xackdelCommand` and no longer expose MODE.
In `@src/t_stream.c`:
- Around line 3799-3804: The XACKDEL path in t_stream.c is currently emitting
the same keyspace notification event as XDEL, so subscribers cannot tell the two
commands apart. Update the notification in the deleted-branch around
signalModifiedKey and notifyKeyspaceEvent to use a dedicated "xackdel" event for
XACKDEL, keeping the rest of the propagation logic unchanged.
In `@tests/unit/type/stream.tcl`:
- Around line 1001-1020: The test comment in the XACKDEL/ACKED scenario is
inconsistent with the actual assertion and should be updated. In the stream test
case around XACKDEL ACKED, replace the stale “despite the -1 response code”
wording with text that matches the asserted return value of 1, so the comments
around XDEL, XPENDING, and XACKDEL all describe the same behavior clearly.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 37dd68bb-cbcf-4401-8293-2c51db6a428b
📒 Files selected for processing (6)
src/commands.defsrc/commands/xackdel.jsonsrc/server.hsrc/t_stream.ctests/unit/type/stream-cgroups.tcltests/unit/type/stream.tcl
- Remove `MODE` token from `XACKDEL` command definition - Correct comment on XACKDEL ACKED test case covering removal of dangling PEL entry Deliberately ignoring the AI review comment on keyspace notifications, because to the best of my digging in the code, it is incorrect. Using keyspace notification `xdel` is valid here, because it is similar to other usage like `del`, which is emitted in many commands not just `DEL`. For example, `GETDEL` emits `del`. So, I'm thinking that `xdel` is the comparable event to emit here to match `XDEL`. Signed-off-by: Nick Iaquinto <git+valkey@iaquinto.io>
|
@valkey-review-bot Please review this change. |
|
I found one correctness issue in the DELREF path where the target group is validated too late; see the inline comment. |
![valkey-review-bot[bot]](https://avatars.githubusercontent.com/in/3633497?s=60&v=4){kind=small}
There was a problem hiding this comment.
Thanks for the patience @nickiaq. Got to this finally.
The code mostly looks good. Have some nitpicks. There is a quite few branching across the code block due to the nature of the command. Need to look at the tests next. Would love additional eyes on this. @soloestoy would you be interested in taking a pass at this?
| if (!streamEntryExists(s, id)) { | ||
| resps[j] = -1; |
There was a problem hiding this comment.
Why do we check if the entry exists after the streamCompareID check? Shouldn't we flip it?
There was a problem hiding this comment.
Initially, I set this up to have the streamCompareID(id, &cg->last_id) check first here because it’s a cheaper check than the other. From what I could tell reviewing the stream state tracking, cg->last_id is the ID of the latest stream message delivered to that consumer group. So I was thinking that means it’s not possible that the stream message is in the PEL of the group if its ID is larger than the latest thing delivered. Whereas streamEntryExists seems to walk the radix tree of stream messages to check if it exists.
But in researching your comment here, I realized the one exception to this is if you run XGROUP SETID to manually move the last_id to before a message in the PEL. In that case, the cheaper check and continue would produce incorrect result because the message in PEL is actually after the last_id.
So I’ve included a fix as you’ve described to be more defensive in commit: 75983f7
- Split `IDS` integer parsing into 2 steps: 1) parse to int, 2) check positive; and use same error message for step 1 as rest of codebase - (Styleguide) Move comments out of conditional expressions to end of line - Move checking PEL before the `cg->last_id` check b/c manually adjusting the `last_id` with `XGROUP SETID` can result in a stream message after the `last_id` that's left in PEL. Signed-off-by: Nick Iaquinto <git+valkey@iaquinto.io>
XACKDEL expects to return -1 for messages not in the target group's PEL. And in this case, we also expect the other groups' PEL's (for DELREF) and the message itself (for ACKED) to remain unmodified/undeleted. To handle this, we need to loop over all messages checking them for the target group first, then we can proceed with the existing code to check the other groups and skip any messages as needed. This addresses review comment valkey-io#3466 (comment) Signed-off-by: Nick Iaquinto <git+valkey@iaquinto.io>
Mirroring the same concern with XDELEX, we need to defer response enqueue-ing until after other modifications to preserve the module keyspace event API contract. See issue valkey-io#3429 and this comment: valkey-io#3467 (comment) Signed-off-by: Nick Iaquinto <git+valkey@iaquinto.io>
|
Thanks for the review! There was lots here to tackle and it was fun digging into the source to figure out some of the more nuanced ones here. I’ve added a note on each review comment above. I had one question (here) and 2 spots where I provided a bit more context (here and here) on what it's trying to do. In addition to the fixes above, I’ve also implemented something to handle this comment (#3467 (comment)) from the XDELEX PR regarding deferring response enqueueing until after message deletion, |
zuiderkwast
left a comment
zuiderkwast
left a comment
There was a problem hiding this comment.
I took a pass, trying to get this reviewed and merged before 9.2 RC1, next week. The logic looks correct. I have only minor comments. It should be close to merge.
If you can refactor to avoid the duplicated code with XDEL, it would be very good. We should not introduce technical dept and then plan to refactor it later, because that often means it never happens.
Responding to [these review comments](valkey-io#3466 (review)), this switches syntax errors to the standard shared helper and simplifies error handling for the mode check and number of ID's matching remaining args. Signed-off-by: Nick Iaquinto <git+valkey@iaquinto.io>
Responding to [these review comments](valkey-io#3466 (review)), this switches syntax errors to the standard shared helper and simplifies error handling for the mode check and number of ID's matching remaining args. Signed-off-by: Nick Iaquinto <git+valkey@iaquinto.io>
0c4618a to
bf72de1
Compare
Responding to [these review comments](valkey-io#3466 (review)), this switches syntax errors to the standard shared helper and simplifies error handling for the mode check and number of ID's matching remaining args. Signed-off-by: Nick Iaquinto <git+valkey@iaquinto.io>
bf72de1 to
1dcd67c
Compare
|
Thanks @zuiderkwast for the review. Also, changed the command JSON to In addition to the commit here, I've taken your larger advice to create a new PR (#4629) that combines this with #3467 |
|
Thanks @zuiderkwast for the help with the reviews. I'm in inline with the feedback to merge these two together and avoid lot's of redundant boilerplate code. |
|
Closing this in favor of #4629! |
4 participants
This PR adds a stream command
XACKDELthat acknowledges one or more messages and (conditionally) deletes them from the stream. The command is especially valuable for teams using Valkey streams as a Kafka alternative. This implementation is compatible with the equivalent command introduced in Redis 8.2.0.For more background on why
XACKDELis necessary, see #2903. This article describes a common pattern when processing a stream in Valkey's current state. To cleanup stream message, a separate "janitor" process tracks each consumer group's progress and periodically callsXTRIMto reclaim space once all messages up to some high water mark have been ack'd by all groups.Another alternative is to use a Lua script that replicates the
XACKDELlogic. Consumer groups can use the script to acknowledge and (conditionally) delete each batch of messages. But, iterating over consumer groups in Lua is slower. And in both the Lua script solution and the janitor process method, there's the risk that the developer introduces a race condition or logic error leading to an unbounded accumulation of stream messages.XACKDELwithACKEDmode makes this janitor process or Lua script unnecessary.For full Redis compatibility, the command supports the same three deletion modes with the same semantics, command syntax, and reply format:
KEEPREF(default, implicit) acks and deletes immediately, leaving PEL references in other groupsACKEDonly deletes once every consumer group has acknowledged or passed the message, making it safe for fan-out stream topologiesDELREFacks, deletes, and forcibly removes PEL entries from all other groupsI've got tests for each mode and possible state included along with other tests for replication, syntax, and other edge cases.
This is my first PR (alongside
XDELEX). And I've tried to follow the style guide & contributing guidelines. Happy to make any corrections if I've missed something.It's possible to create a shared function for parsing the mode & id count between
XACKDELandXDELEX. So I can follow up with a PR to reduce that duplicated code. Or, I can merge these 2 PRs into one.