[kalragauri]

When a Flush() completes on AsyncWriterConnectionResumed or AsyncWriterConnectionBuffered, OnQuery() delegates to SetFlushed(). SetFlushed() satisfies pending flush promises (pending_flush_promises_) and notifies Write() calls that are blocked because the buffer reached the high watermark (flush_handlers_).

Previously, SetFlushed() satisfied the flush promises while the writer was still marked busy (state_ == State::kWriting in WriterConnectionResumed, writing_ == true in WriterConnectionBuffered). OnQuery() only restarted the write loop after SetFlushed() returned. As a result, any operation (Flush, Write, Finalize, or Close) chained synchronously from a Flush().then(...) continuation saw a busy writer and was queued instead of being dispatched before the callback returned.

Running callbacks while the writer is idle exposed related issues, all raised in review:

• A chained Write() could cancel a queued flush. HandleNewData() recalculated flush_ from the buffer size alone. A small Write() issued while another Flush() was still pending set flush_ = false, which turned the queued flush into a plain write and left its future unsatisfied. On the resume path the reordering above newly exposed this, because the write loop used to restart before the callbacks ran.
• High-watermark waiters were released too early. SetFlushed() released every blocked Write() after any flush, even when the buffer was still at or above the low watermark. It also cleared flush_ once no Flush() calls were pending, regardless of how much data remained buffered.
• A chained operation that completed inline could overtake earlier notifications. When one server flush satisfied several Flush() futures (for example Flush(data) followed by Flush({})), a Flush() chained from the first continuation and completed inline by the underlying connection re-entered SetFlushed() and satisfied its own future before the remaining ones from the outer call.
• A chained operation that failed permanently could trigger one more RPC. StartWriting() did not check resume_status_. If a chained operation failed inline with a permanent error, SetError() left the data in the buffer, and the StartWriting() call after the callbacks sent it again on the failed connection. In WriterConnectionResumed the second failure also overwrote resume_status_ (observed as CANCELLED: already finished).

Changes

writer_connection_resumed.cc and writer_connection_buffered.cc (identical changes):

• SetFlushed()
  • Collects the handlers to notify and the satisfied flush promises, updates flush_, and marks the writer idle (state_ = State::kIdle / writing_ = false), all while holding mu_.
  • Hands them to NotifyAndStartWriting(), which delivers them with mu_ released before restarting the write loop, so chained operations see an idle writer and are dispatched immediately.
  • Releases high-watermark waiters only once the buffer is below the low watermark or empty (ClearHandlersIfEmpty() instead of ClearHandlers()).
  • Sets flush_ = NeedsFlush(lk), so the writer keeps flushing while waiters are still blocked.
• OnQuery() (is_resume branch): applies the same idle-then-notify-then-restart sequence through NotifyAndStartWriting() when notifying waiters on stream resume.
• New NotifyAndStartWriting() helper: appends the handlers and flush promises to a pending_notifications_ queue and, unless a delivery loop is already running (notifying_), drains the queue in FIFO order with the lock released between callbacks, then calls StartWriting(). A nested call from a callback whose chained operation completed inline only appends to the queue, so earlier notifications are delivered first and the recursion depth is bounded. Chained operations are still dispatched immediately; only their notifications wait their turn.
• StartWriting(): returns early when resume_status_ is not OK, so no RPC is issued on a connection that already failed permanently.
• New NeedsFlush() helper: returns true while Flush() calls are pending, or while the buffer is non-empty and at or above the low watermark. HandleNewData() and SetFlushed() both use it, so a small Write() can no longer clear flush_ while a flush is pending.
• ClearHandlersIfEmpty(): an empty buffer now always releases waiters. With a low watermark of 0 the previous size >= lwm check was always true, so waiters were never released.

Call Flow: Before vs. After

Before this PR

sequenceDiagram
    participant Impl as Underlying Stream (impl_)
    participant Writer as Buffered / Resumed Writer
    participant User as User f1.then() Callback

    Impl->>Writer: Flush(data1) completes -> OnQuery() -> SetFlushed()
    Note over Writer: Writer is still busy (kWriting / writing_ == true)<br/>Unlocks mu_
    Writer->>User: f1.set_value(OK)
    activate User
    User->>Writer: Flush(data2)
    Note over Writer: HandleNewData() queues data2<br/>StartWriting() sees writer busy -> returns early
    Writer-->>User: returns future f2 (data2 NOT dispatched yet)
    User-->>Writer: callback returns
    deactivate User
    Note over Writer: OnQuery() restarts the write loop after SetFlushed() returns
    Writer->>Impl: impl_->Flush(data2) (dispatched after callback returns)

Loading

After this PR

sequenceDiagram
    participant Impl as Underlying Stream (impl_)
    participant Writer as Buffered / Resumed Writer
    participant User as User f1.then() Callback

    Impl->>Writer: Flush(data1) completes -> OnQuery() -> SetFlushed()
    Note over Writer: Marks writer IDLE under mu_ (kIdle / writing_ = false)<br/>Unlocks mu_
    Writer->>User: f1.set_value(OK)
    activate User
    User->>Writer: Flush(data2)
    Note over Writer: HandleNewData() queues data2<br/>StartWriting() sees writer idle -> dispatches immediately
    Writer->>Impl: impl_->Flush(data2) (in flight before callback returns)
    Writer-->>User: returns future f2
    User-->>Writer: callback returns
    deactivate User
    Note over Writer: StartWriting() sees writer busy -> no-op

Loading

[gemini-code-assist]

Code Review

This pull request refactors the asynchronous buffered and resumed writer connections to mark the writer as idle under the lock before releasing it and notifying handlers, ensuring chained operations are dispatched immediately. It also adds corresponding unit tests to verify these callback behaviors. The review feedback points out a potential data race and crash in the newly added cross-thread flush tests, where worker.join() can be called before the continuation thread finishes initializing the worker thread object, and suggests waiting for the callback to complete before joining.

[codecov]

Codecov Report

❌ Patch coverage is 99.91708% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 92.39%. Comparing base (6285140) to head (c6ffd9a).

Files with missing lines	Patch %	Lines
...e/internal/async/writer_connection_resumed_test.cc	99.83%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #16504      +/-   ##
==========================================
+ Coverage   92.34%   92.39%   +0.04%     
==========================================
  Files        2262     2262              
  Lines      217209   218382    +1173     
==========================================
+ Hits       200592   201771    +1179     
+ Misses      16617    16611       -6     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[v-pratap]

Thanks for addressing the earlier comments; the NeedsFlush() / ClearHandlersIfEmpty() changes and the new HWM/resume tests look good.

Because the writer is now idle while callbacks run, I found two new re-entrancy paths. Both apply to the buffered and resumed writers and are covered in the inline comments with repro tests. Each test fails on this PR and passes with the main-branch versions of the two .cc files:

1. An earlier Flush() future can complete after a later flush chained from a callback that completes inline.
2. After a permanent error raised inline from a callback, the trailing StartWriting() sends one more RPC on the failed connection.