chore(facade): Track async_op_start_cycle_ in the V2 dispatch loop

[mkaruza]

ProcessControlMessages (the V2 dispatch_q_ loop) ran AsyncOperations without setting
async_op_start_cycle_, so check in IsAsyncOpTimedOut was always false. Set/clear it
around the visit, matching the existing pattern in V1 ProcessAdminMessage.

Closes #7897

[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(facade): Track async_op_start_cycle_ in V2 dispatch loop

🐞 Bug fix 🕐 Less than 10 minutes

---

AI Description

• Track control-path async operation start time in the V2 dispatch queue loop.
• Fix slow-subscriber timeout detection by making IsAsyncOpOverdue() observe active sends.
• Align V2 behavior with the existing V1/AsyncFiber start-cycle tracking pattern.

Diagram

graph TD
  X(["Publisher"]) --> Y["SendPubMessageAsync"] --> Q[("dispatch_q_")] --> L["ProcessControlMessages (V2)"] --> S["Set start cycle"] --> V["AsyncOperations visit"] --> C["Clear start cycle"]
  Y --> Z{"IsAsyncOpOverdue?"} -->|"true"| W["RequestPubsubClose"]
  S -. "updates" .-> Z

  subgraph Legend
    direction LR
    _a(["Actor"]) ~~~ _q[("Queue/State")] ~~~ _d{"Decision"} ~~~ _p["Process"]
  end

Loading

High-Level Assessment

The following are alternative approaches to this PR:

1. RAII/absl::Cleanup guard for start-cycle tracking

• ➕ Ensures async_op_start_cycle_ is always cleared even on early returns/exceptions
• ➕ Reduces risk of future edits forgetting to reset the field
• ➖ Slightly more indirection for a simple set/clear
• ➖ May be unnecessary if exceptions/early exits are not possible in this path

2. Move tracking into AsyncOperations visitor implementation

• ➕ Centralizes timing logic so callers (V1/V2 loops) can’t diverge
• ➕ Easier to keep consistent across all dispatch sites
• ➖ Harder to scope timing precisely for only the blocking portion of certain operations
• ➖ May require broader refactoring and careful auditing of all visitor entry points

Recommendation: The PR’s approach is the right minimal fix: it matches the established V1/AsyncFiber pattern and restores correctness for V2 without refactoring call sites. If this timing field gains more uses, consider an RAII guard to make the set/reset pattern harder to accidentally break.

Files changed (1) +2 / -0

Bug fix (1) +2 / -0

dragonfly_connection.ccSet/clear async_op_start_cycle_ around V2 control-message dispatch +2/-0

Set/clear async_op_start_cycle_ around V2 control-message dispatch

• In Connection::ProcessControlMessages (V2 dispatch_q_ draining loop), record async_op_start_cycle_ before executing an AsyncOperations visitor and clear it afterward. This makes IsAsyncOpOverdue() meaningful during V2 control-path operations (e.g., blocked Pub/Sub sends), enabling slow-subscriber timeout protection to trigger correctly.

src/facade/dragonfly_connection.cc

[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 🎫 Failing test: test_pubsub_slow_subscriber_closed

✅ Cross-repo context

  Not relevant to this PR: romange/helio

---

🔴 Action Required

1. Timeout misses deferred flush 🐞 Bug ≡ Correctness

Description

ProcessControlMessages clears async_op_start_cycle_ immediately after std::visit, but IoLoopV2 can
later FlushReplies() outside that scope while async_op_start_cycle_ is 0. If the flush parks on a
slow socket, SendPubMessageAsync’s IsAsyncOpOverdue() gate will stay false during the blocked send
and may fail to trigger RequestPubsubClose/forced_disconnect.

Code

src/facade/dragonfly_connection.cc[R1756-1758]

+    async_op_start_cycle_ = base::CycleClock::Now();
    std::visit(AsyncOperations{reply_builder_.get(), this}, msg.handle);
+    async_op_start_cycle_ = 0;

Evidence

The PR change resets async_op_start_cycle_ right after each control-message visit, but V2 explicitly
flushes deferred replies later via FlushReplies() in the idle park. Since IsAsyncOpOverdue() returns
false when async_op_start_cycle_ is 0 and SendPubMessageAsync relies on it to call
RequestPubsubClose(), a parked flush outside ProcessControlMessages can’t be observed by the timeout
gate. The reply builder already tracks send start time (send_time_cycles_) precisely during
sink_->Write, which can be used to time the real blocked send.

src/facade/dragonfly_connection.cc[1724-1762]
src/facade/dragonfly_connection.cc[2380-2384]
src/facade/dragonfly_connection.cc[2432-2446]
src/facade/dragonfly_connection.cc[3680-3696]
src/facade/reply_builder.h[108-115]
src/facade/reply_builder.cc[171-222]
docs/pub-sub.md[415-437]

Agent prompt

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

## Issue description
`async_op_start_cycle_` is used to decide when a Pub/Sub send is overdue, but in IoLoopV2 the actual socket write can happen in `FlushReplies()` (idle park / error / migrate paths), after `ProcessControlMessages()` has already reset `async_op_start_cycle_` to 0. This makes `IsAsyncOpOverdue()` return false during a blocked `FlushReplies()` send.

## Issue Context
- `ProcessControlMessages()` batches and does not necessarily flush; it sets `async_op_start_cycle_` only around `std::visit` and then resets it.
- IoLoopV2 later calls `FlushReplies()` (which calls `SinkReplyBuilder::Flush()` -> `Send()` -> `sink_->Write(...)`) while `async_op_start_cycle_` is 0.
- `SendPubMessageAsync()` uses `IsAsyncOpOverdue()` as the timeout gate for slow-subscriber close.

## Fix Focus Areas
- src/facade/dragonfly_connection.cc[1724-1762]
- src/facade/dragonfly_connection.cc[2380-2446]
- src/facade/dragonfly_connection.cc[3680-3705]
- src/facade/reply_builder.h[104-115]
- src/facade/reply_builder.cc[171-222]

## Suggested direction
Prefer timing the *actual* blocked send rather than the visitor scope:
- Option A (localized): Change `Connection::IsAsyncOpOverdue()` to return true when `reply_builder_->IsSendActive()` and `Now() - reply_builder_->GetLastSendTimeCycles() >= timeout_cycles` (optionally keep the existing `async_op_start_cycle_` check as a fallback).
- Option B: Set/clear `async_op_start_cycle_` around `FlushReplies()` calls in IoLoopV2 paths that can flush Pub/Sub pushes (idle park, io_ec_, migrate), so the timeout window spans the real parked write.

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

---

---

Powered by Qodo

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

🔴 Action Required

1. Timeout misses deferred flush 🐞 Bug ≡ Correctness

ProcessControlMessages clears async_op_start_cycle_ immediately after std::visit, but IoLoopV2 can
later FlushReplies() outside that scope while async_op_start_cycle_ is 0. If the flush parks on a
slow socket, SendPubMessageAsync’s IsAsyncOpOverdue() gate will stay false during the blocked send
and may fail to trigger RequestPubsubClose/forced_disconnect.

Agent Prompt

## Issue description
`async_op_start_cycle_` is used to decide when a Pub/Sub send is overdue, but in IoLoopV2 the actual socket write can happen in `FlushReplies()` (idle park / error / migrate paths), after `ProcessControlMessages()` has already reset `async_op_start_cycle_` to 0. This makes `IsAsyncOpOverdue()` return false during a blocked `FlushReplies()` send.

## Issue Context
- `ProcessControlMessages()` batches and does not necessarily flush; it sets `async_op_start_cycle_` only around `std::visit` and then resets it.
- IoLoopV2 later calls `FlushReplies()` (which calls `SinkReplyBuilder::Flush()` -> `Send()` -> `sink_->Write(...)`) while `async_op_start_cycle_` is 0.
- `SendPubMessageAsync()` uses `IsAsyncOpOverdue()` as the timeout gate for slow-subscriber close.

## Fix Focus Areas
- src/facade/dragonfly_connection.cc[1724-1762]
- src/facade/dragonfly_connection.cc[2380-2446]
- src/facade/dragonfly_connection.cc[3680-3705]
- src/facade/reply_builder.h[104-115]
- src/facade/reply_builder.cc[171-222]

## Suggested direction
Prefer timing the *actual* blocked send rather than the visitor scope:
- Option A (localized): Change `Connection::IsAsyncOpOverdue()` to return true when `reply_builder_->IsSendActive()` and `Now() - reply_builder_->GetLastSendTimeCycles() >= timeout_cycles` (optionally keep the existing `async_op_start_cycle_` check as a fallback).
- Option B: Set/clear `async_op_start_cycle_` around `FlushReplies()` calls in IoLoopV2 paths that can flush Pub/Sub pushes (idle park, io_ec_, migrate), so the timeout window spans the real parked write.

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

[augmentcode]

🤖 Augment PR Summary

Summary: V2 ProcessControlMessages now records async_op_start_cycle_ while visiting control messages and clears it afterward.

Why: Enables IsAsyncOpOverdue/slow-subscriber protection for Pub/Sub in the V2 dispatch loop, matching the existing V1 path.

_{🤖 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.

[glevkovich]

LGTM, consider openning a follow up on https://github.com/dragonflydb/dragonfly/pull/7902/changes#r3622273930 for V2