fix(facade): Use ReplyScope in loop2

[dranikpg]

Fixes #7759

[dranikpg]

augment review

[augmentcode]

🤖 Augment PR Summary

Summary: This PR adjusts the pipelined reply path to ensure reply data lifetimes remain valid while batching replies.

Changes:

• Wraps the entire Connection::ReplyBatch() iteration in a single SinkReplyBuilder::ReplyScope instead of creating/ending scopes per command.
• Defers ReleasePipelinedCommand() calls until after the scope ends, preventing reply references from being invalidated too early.
• Adds a per-command ScopePause for IsSuspendedReply() (coroutine) replies so those replies don’t rely on extended scope lifetimes.
• Tracks how many commands were actually replied to and releases exactly that prefix of the pipeline queue.

Technical Notes: The updated flow keeps reply ordering intact while allowing the reply builder to batch/flush safely without prematurely freeing command-owned reply storage.

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

[augmentcode]

Review completed. No suggestions at this time.

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

[qodo-code-review]

Qodo reviews are paused for this user.

Troubleshooting steps vary by plan Learn more →

On a Teams plan?
Reviews resume once this user has a paid seat and their Git account is linked in Qodo.
Link Git account →

Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center?
These require an Enterprise plan - Contact us
Contact us →

[qodo-free-for-open-source-projects]

PR Summary by Qodo

Fix ReplyBatch lifetime by using a single ReplyScope across IoLoopV2 batching

🐞 Bug fix 🕐 20-40 Minutes

AI Description

• Wrap pipelined replies in a single ReplyScope to keep references valid.
• Delay releasing pipelined commands until after the batch scope completes.
• Add ScopePause handling for suspended replies and stop early on reply errors.

Diagram

graph TD
  A["Connection::ReplyBatch"] --> B["ReplyBuilder batch mode"] --> C["ReplyScope (single)"] --> D["Parsed command list"] --> E{"IsSuspendedReply?"} --> F["SendReply"] --> G["Release replied cmds"]
  E -- "yes" --> H["ScopePause"] --> F
  E -- "no" --> F

Loading

High-Level Assessment

The chosen approach (one ReplyScope for the whole batch, releasing commands only after scope exit) is the smallest change that directly addresses lifetime invalidation risk. Alternatives like deep-copying reply payloads or introducing ref-counted command/reply ownership would be more invasive and harder to validate across the existing pipeline/coroutine reply paths.

Files changed (1) +30 / -14

Bug fix (1) +30 / -14

dragonfly_connection.ccHold a single ReplyScope and defer pipelined command release in ReplyBatch +30/-14

Hold a single ReplyScope and defer pipelined command release in ReplyBatch

• Reworks Connection::ReplyBatch to create one SinkReplyBuilder::ReplyScope for the entire reply loop, preventing reply lifetime invalidation when commands are released. Adds ScopePause for suspended replies and changes error handling to break the loop and return failure after releasing already-replied commands.

src/facade/dragonfly_connection.cc

[augmentcode]

Review completed. No suggestions at this time.

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

[qodo-free-for-open-source-projects]

Code Review by Qodo

🐞 Bugs (1) 📘 Rule violations (0) 📎 Requirement gaps (0) 🔗 Cross-repo conflicts (0) 📜 Skill insights (0)

Context used

✅ Tickets: 🎫 bug(eval) SIGILL when execute EVAL command

✅ Cross-repo context

  Not relevant to this PR: romange/helio

1. ReplyScope triggers early flush 🐞 Bug ➹ Performance

Description

Connection::ReplyBatch wraps multiple command replies in a single SinkReplyBuilder::ReplyScope, so
FinishScope() runs once with total_size_ equal to the whole replied chunk and may call Flush()
inside ReplyBatch. This can defeat IoLoopV2’s intentional deferred flushing/coalescing and increase
send syscalls when the combined reply payload crosses the FinishScope flush threshold.

Code

src/facade/dragonfly_connection.cc[R2913-2931]

+  {
+    SinkReplyBuilder::ReplyScope scope(reply_builder_.get());
+    while (HasInFlightCommands() && parsed_head_->CanReply()) {
+      current_wait_.reset();  // Clear the subscription before moving to the next command
+      auto* cmd = parsed_head_;
+      AdvanceParsedHead(parsed_head_->next);

-    // An in-flight command is considered to be a pipelined command.
-    ReleasePipelinedCommand(cmd);
-    if (reply_builder_->GetError())
-      return false;
+      // Pure coroutine replies don't preserve lifetimes
+      std::optional<SinkReplyBuilder::ScopePause> pause;
+      if (cmd->IsSuspendedReply())
+        pause.emplace(reply_builder_.get());

-    if (!HasInFlightCommands())
-      break;
+      cmd->SendReply();
+      replied++;
+
+      if (reply_builder_->GetError())
+        break;
+    }
  }

Evidence

ReplyBatch() explicitly avoids flushing in IoLoopV2 (if (!ioloop_v2_) reply_builder_->Flush();),
but the new outer ReplyScope will still call FinishScope() on scope exit. FinishScope() calls
Flush() when total_size_ * 2 >= kMaxBufferSize, which sends immediately via sink_->Write(),
reintroducing send syscalls during ReplyBatch() for larger combined batches.

src/facade/dragonfly_connection.cc[2904-2954]
src/facade/reply_builder.cc[171-192]
src/facade/reply_builder.cc[225-235]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`Connection::ReplyBatch()` now replies multiple commands under a single `SinkReplyBuilder::ReplyScope`. Because `ReplyScope::~ReplyScope()` calls `FinishScope()` once for the whole batch, the `total_size_` it evaluates is the *combined* size of all replies in that batch. `FinishScope()` may call `Flush()` when `total_size_ * 2 >= kMaxBufferSize`, which can trigger a `sink_->Write()` syscall inside `ReplyBatch()` even on IoLoopV2, where the code explicitly tries to avoid flushing here.

### Issue Context
- IoLoopV2 comment states flushing is delegated to IoLoopV2 to avoid a syscall per parse chunk.
- `SinkReplyBuilder::FinishScope()` may call `Flush()` based on `total_size_`.

### Fix Focus Areas
- src/facade/dragonfly_connection.cc[2909-2952]
- src/facade/reply_builder.cc[171-192]
- src/facade/reply_builder.cc[225-235]

### Suggested fix
Restructure `ReplyBatch()` so that `ReplyScope` is not allowed to grow unbounded across many replies:
1. Use a per-command (or bounded-size) `ReplyScope` so `FinishScope()` decisions are made on smaller reply segments, reducing the chance of triggering `Flush()` inside `ReplyBatch()`.
2. If you must keep delayed command release for lifetime reasons, consider batching commands in groups (e.g., close/reopen the scope every N commands or when an estimated byte threshold is approached), then release only the commands covered by the closed scope.

This keeps the lifetime-safety benefit (don’t release while refs may be live) while preserving IoLoopV2’s intended flush coalescing behavior.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

[romange]

why did you remove this condition?

[dranikpg]

I made it part of the loop

[romange]

How to validate the improvement?
is there a regression test that covers this?

[dranikpg]

How to validate the improvement?
is there a regression test that covers this?

Yes, #7759

[romange]

Ok, but for the reference, did you run any dfly_bench commands that show before and after?

[dranikpg]

Ok, but for the reference, did you run any dfly_bench commands that show before and after?

Yes, it shows improvement