fix: xreadgroup is a write command #6492

kostasrim · 2026-01-28T20:06:25Z

Summary: This PR treats XREADGROUP as a write/journaled command and adds explicit journaling to replicate its side effects.

Changes:

Extends stream read records with delivery_time to support journaling PEL delivery metadata.
Updates OpRange/OpRead paths to capture delivery time when creating PEL entries for consumer groups.
Adds JournalXReadGroupIfNeeded to journal XCLAIM (when not NOACK) and XGROUP SETID with ENTRIESREAD when reading new messages (ID >).
Extends XGROUP SETID implementation to optionally update entries_read, and adds command parsing for ENTRIESREAD.
Adjusts XREADGROUP command registration flags to be JOURNALED with NO_AUTOJOURNAL (manual journaling).
Adds replication tests validating journal output and replica XINFO GROUPS state across NOACK/PEL and blocking/non-blocking paths.

Notes: The replication strategy emulates Redis/Valkey behavior by translating XREADGROUP effects into explicit journal entries (XCLAIM + XGROUP SETID) to keep consumer-group state consistent on replicas.

resolves #6493

Signed-off-by: Kostas Kyrimis <[email protected]>

Copilot

Pull request overview

Adjusts Dragonfly’s handling of XREADGROUP to behave as a write/journaled command and adds tests to validate correct replication/journaling behavior (including ENTRIESREAD propagation).

Changes:

Marks XREADGROUP as CO::JOURNALED | CO::NO_AUTOJOURNAL and adds manual journaling logic for XREADGROUP side effects.
Adds support for parsing/setting XGROUP SETID ... ENTRIESREAD <n> to keep group counters consistent across replicas.
Adds replication and monitor-based tests covering XREADGROUP journaling/replication (NOACK vs PEL, blocking vs non-blocking).

Reviewed changes

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

File	Description
tests/dragonfly/replication_test.py	Adds monitor expectations for `XREADGROUP` journaling and a new replication test ensuring group state matches between master/replica.
src/server/stream_family.cc	Implements manual journaling for `XREADGROUP`, tracks delivery time, and propagates `ENTRIESREAD` via `XGROUP SETID`.

Comments suppressed due to low confidence (1)

src/server/stream_family.cc:2554

In XReadBlock(), the key variable is no longer assigned when a wake key is found (the previous key = *wake_key; was removed). The reply path uses SendStreamRecords(key, *result), so blocking XREAD/XREADGROUP responses will be sent with an empty/incorrect stream key. Restore setting key (e.g., to *wake_key) alongside computing result.

      range_opts.noack = opts->noack;

      result = OpRange(t->GetOpArgs(shard), *wake_key, range_opts);
      if (result) {
        JournalXReadGroupIfNeeded(t->GetOpArgs(shard), *opts, *result, *wake_key);
      }
    }
    return OpStatus::OK;
  };
  tx->Execute(std::move(range_cb), true);

  if (result) {
    SinkReplyBuilder::ReplyAggregator agg(rb);
    if (opts->read_group && rb->IsResp3()) {
      rb->StartCollection(1, CollectionType::MAP);
    } else {
      rb->StartArray(1);
      rb->StartArray(2);
    }
    return StreamReplies{rb}.SendStreamRecords(key, *result);
  } else if (result.status() == OpStatus::INVALID_VALUE) {

src/server/stream_family.cc

tests/dragonfly/replication_test.py

augmentcode · 2026-01-28T20:13:01Z

🤖 Augment PR Summary

Summary: This PR treats XREADGROUP as a write/journaled command and adds explicit journaling to replicate its side effects.

Changes:

Extends stream read records with delivery_time to support journaling PEL delivery metadata.
Updates OpRange/OpRead paths to capture delivery time when creating PEL entries for consumer groups.
Adds JournalXReadGroupIfNeeded to journal XCLAIM (when not NOACK) and XGROUP SETID with ENTRIESREAD when reading new messages (ID >).
Extends XGROUP SETID implementation to optionally update entries_read, and adds command parsing for ENTRIESREAD.
Adjusts XREADGROUP command registration flags to be JOURNALED with NO_AUTOJOURNAL (manual journaling).
Adds replication tests validating journal output and replica XINFO GROUPS state across NOACK/PEL and blocking/non-blocking paths.

Technical Notes: The replication strategy emulates Redis/Valkey behavior by translating XREADGROUP effects into explicit journal entries (XCLAIM + XGROUP SETID) to keep consumer-group state consistent on replicas.

_{🤖 Was this summary useful? React with 👍 or 👎}

augmentcode

Review completed. 5 suggestions posted.

Comment augment review to trigger a new review at any time.

src/server/stream_family.cc

kostasrim · 2026-01-28T20:17:04Z

src/server/stream_family.cc


    OpResult<RecordVec> range_res;

+    // TODO server history IS PER stream. Fix this incorrect behaviour


Signed-off-by: Kostas Kyrimis <[email protected]>

Copilot

Pull request overview

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

Comments suppressed due to low confidence (1)

src/server/stream_family.cc:806

OpRange no longer short-circuits when opts.count == 0. ParseReadArgsOrReply allows COUNT 0, so this path can now scan and return entries instead of returning an empty result, changing command semantics and potentially causing unexpected work.

Restore the count == 0 early-return (and ensure it also avoids any XREADGROUP side-effects like PEL/NACK creation).

  RecordVec result;

  streamIterator si;
  int64_t numfields;
  streamID id;
  stream* s = (stream*)cobj->RObjPtr();
  streamID sstart = opts.start.val, send = opts.end.val;

  streamIteratorStart(&si, s, &sstart, &send, opts.is_rev);
  while (streamIteratorGetID(&si, &id, &numfields)) {

src/server/stream_family.cc

Signed-off-by: Kostas Kyrimis <[email protected]>

Copilot

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 4 comments.

src/server/stream_family.cc

Copilot

Pull request overview

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

src/server/stream_family.cc

tests/dragonfly/replication_test.py

Signed-off-by: Kostas Kyrimis <[email protected]>

Copilot

Pull request overview

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

Comments suppressed due to low confidence (1)

src/server/stream_family.cc:934

In OpRangeFromConsumerPEL, the code updates nack->delivery_time/delivery_count, but the returned Record doesn’t carry the updated delivery metadata (Record::delivery_time remains 0, and there’s no delivery-count field). If XREADGROUP history reads need to be journaled (or otherwise observed outside the shard), this loses the information required to reproduce PEL delivery metadata on replicas. Consider propagating delivery time/count into the returned record (or journaling directly from the nack).

      Record rec;
      rec.id = id;
      result.push_back(rec);
    } else {
      streamNACK* nack = static_cast<streamNACK*>(ri.data);
      nack->delivery_time = op_args.db_cntx.time_now_ms;
      nack->delivery_count++;
      result.push_back(std::move(op_result.value()[0]));

Copilot · 2026-01-29T12:53:01Z

src/server/stream_family.cc

+  if (!opts.serve_history) {
+    // Reading NEW messages (ID = ">")
+    auto journal_xgroup = [&opts, op_args](const auto& records, std::string_view key) {
+      if (!records.empty()) {
+        const auto& sitem = opts.stream_ids.at(key);
+        auto id = absl::StrCat(records.back().id.ms, "-", records.back().id.seq);
+        auto entries_read = absl::StrCat(sitem.group->entries_read);
+        CmdArgVec journal_args = {"SETID", key, opts.group_name, id, "ENTRIESREAD", entries_read};
+        RecordJournal(op_args, "XGROUP"sv, journal_args);
+      }
+    };
+    for (auto& record : records) {
+      if (!opts.noack) {
+        auto id = absl::StrCat(record.id.ms, "-", record.id.seq);
+        auto deliv_time = absl::StrCat(record.delivery_time);
+        CmdArgVec journal_args = {key, opts.group_name, opts.consumer_name, "0",
+                                  id,  "TIME",          deliv_time,         "RETRYCOUNT",
+                                  "1", "FORCE",         "JUSTID",           "LASTID",
+                                  id};
+
+        RecordJournal(op_args, "XCLAIM"sv, journal_args);
+      }
+    }
+    journal_xgroup(records, key);
+  }


JournalXReadGroupIfNeeded skips journaling entirely when opts.serve_history is true, but OpRangeFromConsumerPEL still mutates PEL metadata (delivery_time/delivery_count). That means XREADGROUP reads from the PEL (non->) will not replicate these side effects, and replicas can diverge from the master’s pending-entry delivery metadata. Consider journaling an XCLAIM (with correct TIME/RETRYCOUNT) for history reads as well, while still only emitting XGROUP SETID ... ENTRIESREAD when reading new messages (>).

Interesting! I thought we did not propagate the consumer but I was wrong. Also there are some small corner cases! Will adress the shortly

Ok, so this is what I was talking about consumers. Valkey and Redis handles this slightly differently. Either way I will follow up on a different PR - it's getting too large and the case is not very interesting

src/server/stream_family.cc

romange

Looks good. there are some small comments that you still can fix.
also - seems it can benefit from unit tests as well, not only the replication.

kostasrim · 2026-01-30T07:36:41Z

Looks good. there are some small comments that you still can fix. also - seems it can benefit from unit tests as well, not only the replication.

Can you plz clarify the following:

What did I miss from small comments ?
The only comments I did not address is (a) batching of journal changes (and I did that on purpose) (b) the consumer not being replicated via XGROUPCREATE under one flow which I can I will follow up on a separate PR (It's 260 lines already -- I don't want to push it)
"benefit from unit tests as well ?"
We have unit tests for xreadgroup in stream_family_test.cc so I am not really sure what you want to achieve with a unit here ? Do you mean rewritting the pytests as units or ? Furthermore, I am planning to improve our unit tests around xreadgroup but not on this PR: see XREADGROUP sticks if reading from history #6494

Copilot

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

Copilot · 2026-01-30T08:00:21Z

src/server/stream_family.cc

+  if (is_write_command) {
    group = streamLookupCG(s, WrapSds(opts->group_name));
    if (!group)
      return facade::ErrorReply{


In the XREADGROUP path, FindOrAddConsumer() (a few lines below) can return nullptr on OOM. That pointer is stored into requested_sitem.consumer and later OpRange dereferences opts.consumer when NOACK is not set, which can crash. Add an explicit null check after FindOrAddConsumer() and return an OUT_OF_MEMORY error reply instead of proceeding.

romange

You changed the semantics of the commands. Is it possible to capture these changes via unit tests? i.e. if you create a consumer - check that a consumer is created. if you updated the metadata - check that the metadata is updated.

kostasrim · 2026-01-30T08:26:59Z

You changed the semantics of the commands. Is it possible to capture these changes via unit tests? i.e. if you create a consumer - check that a consumer is created. if you updated the metadata - check that the metadata is updated.

Yes this is done implicitly by the replication test + the XINFO command which checks if the side effect propagates. I did not change or affect consumer creation at all.

The main change for master, is that we properly call FindMutable on write paths which updates the dbtable via the post updater but I don't see why we would need a unit for this 🤷

I am happy to add anything -- just trying not to be redundant with tests at least

kostasrim added 3 commits January 28, 2026 22:02

fix: xreadgroup is a write command

85cab30

Signed-off-by: Kostas Kyrimis <[email protected]>

Merge branch 'main' into kpr5

efed268

fix

e28b9d0

Copilot AI review requested due to automatic review settings January 28, 2026 20:06

Copilot started reviewing on behalf of kostasrim January 28, 2026 20:07 View session

kostasrim self-assigned this Jan 28, 2026

Copilot AI reviewed Jan 28, 2026

View reviewed changes

augmentcode bot reviewed Jan 28, 2026

View reviewed changes

kostasrim commented Jan 28, 2026

View reviewed changes

kostasrim added 2 commits January 28, 2026 22:22

wording

77515c9

fix: replace FindReadOnly with FindMutable on write paths

1179c3d

Signed-off-by: Kostas Kyrimis <[email protected]>

Copilot AI review requested due to automatic review settings January 29, 2026 10:58

Copilot started reviewing on behalf of kostasrim January 29, 2026 10:58 View session

Copilot AI reviewed Jan 29, 2026

View reviewed changes

src/server/stream_family.cc Show resolved Hide resolved

src/server/stream_family.cc Outdated Show resolved Hide resolved

src/server/stream_family.cc Outdated Show resolved Hide resolved

kostasrim added 2 commits January 29, 2026 13:31

polish

f2d080f

Signed-off-by: Kostas Kyrimis <[email protected]>

Merge branch 'main' into kpr5

de20c2f

Copilot AI review requested due to automatic review settings January 29, 2026 11:39

Copilot started reviewing on behalf of kostasrim January 29, 2026 11:39 View session

Copilot AI reviewed Jan 29, 2026

View reviewed changes

src/server/stream_family.cc Show resolved Hide resolved

src/server/stream_family.cc Outdated Show resolved Hide resolved

src/server/stream_family.cc Outdated Show resolved Hide resolved

src/server/stream_family.cc Show resolved Hide resolved

comments

56df24b

kostasrim requested a review from Copilot January 29, 2026 11:55

Copilot started reviewing on behalf of kostasrim January 29, 2026 11:56 View session

Copilot AI reviewed Jan 29, 2026

View reviewed changes

src/server/stream_family.cc Outdated Show resolved Hide resolved

src/server/stream_family.cc Show resolved Hide resolved

tests/dragonfly/replication_test.py Outdated Show resolved Hide resolved

comments

c85557e

Signed-off-by: Kostas Kyrimis <[email protected]>

kostasrim requested a review from Copilot January 29, 2026 12:45

Copilot started reviewing on behalf of kostasrim January 29, 2026 12:46 View session

Copilot AI reviewed Jan 29, 2026

View reviewed changes

small fix

7588a31

kostasrim requested a review from romange January 29, 2026 16:28

romange reviewed Jan 29, 2026

View reviewed changes

Merge branch 'main' into kpr5

6d3cbf0

Copilot AI review requested due to automatic review settings January 30, 2026 07:50

Copilot started reviewing on behalf of kostasrim January 30, 2026 07:50 View session

Copilot AI reviewed Jan 30, 2026

View reviewed changes

romange reviewed Jan 30, 2026

View reviewed changes


		OpResult<RecordVec> range_res;

		// TODO server history IS PER stream. Fix this incorrect behaviour

fix: xreadgroup is a write command #6492

Are you sure you want to change the base?

fix: xreadgroup is a write command #6492

Conversation

kostasrim commented Jan 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

augmentcode bot commented Jan 28, 2026

Uh oh!

augmentcode bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

kostasrim Jan 28, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI Jan 29, 2026

Choose a reason for hiding this comment

Uh oh!

kostasrim Jan 29, 2026

Choose a reason for hiding this comment

Uh oh!

kostasrim Jan 29, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

romange left a comment

Choose a reason for hiding this comment

Uh oh!

kostasrim commented Jan 30, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI Jan 30, 2026

Choose a reason for hiding this comment

Uh oh!

romange left a comment

kostasrim commented Jan 28, 2026 •

edited

Loading