feat(stream): support XACKDEL command

[kirito632]

What

This PR introduces the XACKDEL stream command with configurable
PEL cleanup behaviors.

It also fixes an existing issue in XINFO CONSUMERS where consumer metadata
from different groups could leak into the same scan result.

XINFO CONSUMERS Fix

Root cause:

GetConsumerInfo() sets iterate_upper_bound by incrementing only the
stream version while keeping the sub-key prefix unchanged.

Since InternalKey encoding places the version field before the sub-key,
consumer records belonging to other groups can still fall within the
same [version, version+1) iteration range.

Fix:

Extract and validate the group name from each internal key during iteration,
filtering out unrelated consumers.

XACKDEL

XACKDEL key group [KEEPREF|DELREF|ACKED] IDS numids id [id ...]

Deletes stream entries with configurable PEL cleanup behavior.

Supported deletion strategies:

KEEPREF (default):
Delete the stream entry only, without touching existing PEL references.
DELREF:
Delete the stream entry and remove it from all consumer groups' PELs.
ACKED:
Delete the stream entry only if it has been acknowledged by all consumer groups.

Per-ID return codes:

• 1: entry deleted
• 2: entry retained (e.g. ACKED condition not satisfied)
• -1: entry not found or duplicate ID within the same request

Implementation Notes

• Pending-number decrements are aggregated in memory and flushed through
  a single WriteBatch to avoid repeated metadata updates during DELREF operations.
• Duplicate IDs are deduplicated to preserve idempotency and prevent
  stream metadata corruption.
• The originating command name is propagated into
  WriteBatchExtractor so replicas replay the exact
  same deletion semantics.
• Internal stream metadata and PEL delete events are filtered out in
  WriteBatchExtractor to avoid generating invalid
  replicated XDEL commands.

Testing

Added Go integration tests covering:

• multi-group DELREF cleanup
• ACKED blocking semantics
• duplicate-ID idempotency
• invalid syntax handling
• XINFO CONSUMERS group isolation

AI-Assisted Contribution Disclosure

This contribution complies with the ASF AI-assisted contribution guidelines.

AI tools were used to assist with English phrasing and some test scaffolding.
The core C++ implementation, debugging, correctness analysis, and final verification
were independently completed by me.

[jihuayu]

Hi @kirito632. If the core C++ implementation part was also completed by AI, please state that directly as well — no need to worry about it.
As long as you can take responsibility for the final code, we welcome AI-generated code.
We can also exchange some related techniques and workflows.

[kirito632]

Thanks for the clarification.
AI tools were indeed involved not only in wording and test scaffolding, but also in discussing implementation approaches and generating parts of the initial C++ code during development.
That said, I independently reviewed, debugged, modified, and verified the final implementation locally, including the stream deletion semantics, replication handling and metadata updates. I take full responsibility for the final code and behavior in this PR.
I'm happy to exchange related workflows and techniques as well.

[Copilot]

Pull request overview

This PR adds a new Stream command XACKDEL (with multiple PEL cleanup strategies) and fixes XINFO CONSUMERS to avoid leaking consumer metadata across groups when scanning internal keys.

Changes:

• Add XACKDEL command parsing/registration and implement entry deletion with KEEPREF|DELREF|ACKED semantics.
• Fix XINFO CONSUMERS scanning by explicitly validating the group name extracted from each internal key.
• Update replication/migration write-batch extraction to replay XACKDEL semantics and filter internal stream metadata/PEL events; add Go integration tests for the new behaviors.

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
tests/gocase/unit/type/stream/stream_test.go	Adds integration tests for XACKDEL behaviors and XINFO CONSUMERS group isolation.
src/types/redis_stream.h	Declares new helper methods for XACKDEL implementation.
src/types/redis_stream.cc	Implements XACKDEL deletion/PEL cleanup logic and fixes XINFO CONSUMERS group filtering.
src/types/redis_stream_base.h	Introduces enums for delete options and per-ID delete results.
src/storage/batch_extractor.cc	Adjusts stream write-batch extraction to skip internal stream metadata/PEL keys and replay XACKDEL deletes.
src/commands/cmd_stream.cc	Adds XACKDEL command parsing, execution, and registration.

Comments suppressed due to low confidence (1)

src/storage/batch_extractor.cc:282

• In PutCF for the Stream column family, the extracted command is appended to resp_commands_[ns], but ns is never set from the InternalKey. This will bucket stream commands under the empty namespace and can break namespace-aware outputs (e.g., slot migration or command dump). Also, the Stream CF path does not apply the slot_range_ filtering used by other CF handlers, so out-of-range stream updates may be emitted during slot migration.

  } else if (column_family_id == static_cast<uint32_t>(ColumnFamilyID::Stream)) {
    InternalKey ikey(key, is_slot_id_encoded_);
    Slice entry_id_check = ikey.GetSubKey();
    uint64_t delimiter = 0;
    GetFixed64(&entry_id_check, &delimiter);
    if (delimiter == UINT64_MAX) {
      return rocksdb::Status::OK();
    }

    auto s = ExtractStreamAddCommand(is_slot_id_encoded_, key, value, &command_args);
    if (!s.IsOK()) {
      ERROR("Failed to parse write_batch in PutCF. Type=Stream: {}", s.Msg());
      return rocksdb::Status::OK();
    }
  }

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.