[sushanb]

Summary

MutationBatcher::OnBulkApplyDone previously called completion_promise.set_value(...) before acquiring mu_ to decrement outstanding_size_, outstanding_mutations_, num_requests_pending_, and num_outstanding_batches_ and admit queued mutations:

1. Synchronous .then() flow-control deadlock: A synchronous continuation attached to completion_future.then(...) observed un-decremented flow-control counters, blocking follow-up AsyncApply() admission at the batch/mutation limit.
2. Shutdown use-after-free hazard: If a waiting thread or completion continuation destroyed the MutationBatcher as soon as the final completion_future resolved, OnBulkApplyDone subsequently dereferenced this->mu_ and member fields on a destroyed MutationBatcher.

Changes

• Collect (CompletionPromise, Status) pairs during batch result processing, then lock mu_, update flow-control counters, run TryAdmit(cq), release mu_, and satisfy admission_promises and completion_promises without touching this again.
• Track in-flight mutations, outstanding batches, and active completion-callback batches in a std::shared_ptr<WaitState> so AsyncWaitForNoPendingRequests() only resolves after all batch completion callbacks (and any follow-up AsyncApply() work submitted inside them, or concurrent batches completing across CQ threads) have finished, even if a callback destroys MutationBatcher.
• Remove unreachable no_more_pending_promises dead code from MutationBatcher::SatisfyPromises while preserving the symbol for ABI compatibility.
• Add unit tests in google/cloud/bigtable/mutation_batcher_test.cc covering AsyncWaitForNoPendingRequests() across multi-mutation batches, follow-up AsyncApply() inside .then(), concurrent multi-batch completions, flow-control admission inside .then(), and destroying MutationBatcher from completion continuations or waiting threads.

[gemini-code-assist]

Code Review

This pull request modifies MutationBatcher::OnBulkApplyDone to satisfy completion, admission, and pending request promises outside of the mutex lock mu_ and without accessing this afterward. This prevents potential deadlocks or use-after-free issues if a continuation or waiting thread destroys the MutationBatcher instance. Additionally, range-based loops were updated to use explicit types instead of auto to comply with the style guide. Comprehensive unit tests were added to cover various edge cases, including multi-threaded destruction and continuation behavior under flow control limits. I have no further feedback to provide as the changes are well-implemented and adhere to the repository's style guidelines.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 92.50%. Comparing base (5dcdef5) to head (c44e491).
⚠️ Report is 20 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #16659      +/-   ##
==========================================
- Coverage   92.53%   92.50%   -0.03%     
==========================================
  Files        2269     2269              
  Lines      222896   223147     +251     
==========================================
+ Hits       206250   206433     +183     
- Misses      16646    16714      +68     

☔ 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.

[scotthart]

Findings & Edge Cases to Consider

1. [Medium] AsyncWaitForNoPendingRequests() can resolve while a .then() callback submits new work, or while completion_promises are still executing

Because OnBulkApplyDone cannot touch this after satisfying completion_promises (to allow destroying MutationBatcher inside a completion callback), num_requests_pending_ is decremented to 0 and no_more_pending_promises_ is swapped out before admission_promises and completion_promises are satisfied (mutation_batcher.cc:56-85). This introduces three subtle interactions with AsyncWaitForNoPendingRequests():

• Calling AsyncApply() inside .then() when no_more_pending_promises_ was already swapped out:
  If a caller has an active AsyncWaitForNoPendingRequests() future and a completion_future.then(...) (or admission_future.then(...)) callback submits a retry/follow-up mutation via batcher.AsyncApply(cq, ...) (the pattern tested in CompletionContinuationCanApplyAtFlowControlLimit), no_more_pending_promises_ was already swapped into the local no_more_pending_promises vector at line 68 when counters transiently hit 0. At lines 83–85, OnBulkApplyDone will still satisfy no_more_pending_promises even though the newly submitted mutation is now pending/in-flight on cq—which can re-trigger the #2140 premature cq.Shutdown() issue cited in the comment at lines 63–67.
• Concurrent AsyncWaitForNoPendingRequests() call on another thread during promise satisfaction:
  In the canonical usage pattern documented in mutation_batcher.h, the main thread calls batcher.AsyncWaitForNoPendingRequests().get() after submitting all mutations. If that call happens in the window after OnBulkApplyDone unlocks mu_ (line 70) and before completion_promises finish executing (lines 80–82), AsyncWaitForNoPendingRequests() acquires mu_, sees num_requests_pending_ == 0, and immediately returns make_ready_future() while completion_future.then(...) callbacks are still running on the CQ thread.
• Multiple batches in flight across multiple CQ threads (max_batches > 1):
  If two CQ threads execute OnBulkApplyDone for the last two batches concurrently, Thread B can decrement num_requests_pending_ from 1 to 0, swap no_more_pending_promises_, and satisfy no_more_pending_promises while Thread A (which already decremented from 2 to 1 and released mu_) is still executing its completion_promises loop.

2. [Low / Cleanup] Dead no_more_pending_promises_ logic in MutationBatcher::SatisfyPromises

In mutation_batcher.cc, MutationBatcher::SatisfyPromises was previously shared by AsyncApply and OnBulkApplyDone. Now that OnBulkApplyDone inlines its own promise-satisfaction logic, the only remaining caller of SatisfyPromises is AsyncApply, which increments ++num_requests_pending_ at line 84 while holding lk.

• As a result, num_requests_pending_ is always >= 1 when SatisfyPromises is called, making the if (num_requests_pending_ == 0 && num_outstanding_batches_ == 0) block and the no_more_pending_promises loop in SatisfyPromises unreachable dead code (and duplicating the #2140 comment).
• Note: since MutationBatcher is in a GA library and SatisfyPromises is declared in mutation_batcher.h, keeping the MutationBatcher::SatisfyPromises symbol avoids any ABI dump diff (check-api-pr), so simplifying its body in mutation_batcher.cc (or leaving a brief comment if kept for ABI compatibility) would clean up the duplication.

3. [Nit] Multi-mutation batch behavior in CompletionContinuationCanWaitForNoPendingRequests

In OnBulkApplyDone, num_requests_pending_ -= num_mutations decrements the counter for the entire batch before iterating over completion_promises. Consequently, in a batch with multiple mutations (m1, m2), m1's .then() continuation observes AsyncWaitForNoPendingRequests().is_ready() == true even though m2's completion promise has not been satisfied yet. If that is the intended trade-off of updating batch counters upfront, it may be worth noting in a comment.

[scotthart]

See PR level comment.

[sushanb]

Thanks for catching those interactions! Addressed all 3 points in f2fe7b4345:

1. AsyncWaitForNoPendingRequests() synchronization across .then() follow-ups and concurrent batches (WaitState): Introduced a std::shared_ptr<WaitState> (pending_work + promises) held by both MutationBatcher and OnBulkApplyDone:
   • AsyncApply increments pending_work per admitted/queued request, and FlushIfPossible increments pending_work by +1 per in-flight batch.
   • Before unlocking mu_ to satisfy admission_promises and completion_promises, OnBulkApplyDone subtracts num_mutations from pending_work while retaining its +1 batch unit until after all admission_promises and completion_promises finish (never touching this again, preserving safe destruction of MutationBatcher inside callbacks).
   • Once callbacks finish, OnBulkApplyDone decrements the final +1 on wait_state and only satisfies wait_state->promises when pending_work == 0. This ensures:
     • Follow-up AsyncApply() calls inside .then() increment pending_work before promises is checked/swapped (WaitForNoPendingWaitsForFollowUpAppliedInCompletionContinuation).
     • Concurrent AsyncWaitForNoPendingRequests() calls on another thread and concurrent OnBulkApplyDone executions across multiple CQ threads never resolve while any completion callback is still running (WaitForNoPendingAcrossConcurrentBatches).
2. Cleaned up MutationBatcher::SatisfyPromises: Removed the unreachable no_more_pending_promises check/loop and duplicate comment while keeping the MutationBatcher::SatisfyPromises member definition for ABI compatibility.
3. Multi-mutation batch behavior: With pending_work held until the batch's completion_promises loop completes, AsyncWaitForNoPendingRequests() called inside m1's .then() stays unready until m2's callback has also completed (CompletionContinuationCanWaitForNoPendingRequests).

/gemini review

[gemini-code-assist]

Code Review

This pull request refactors the lifetime and synchronization management of MutationBatcher during asynchronous operations. It introduces a shared WaitState structure (held via std::shared_ptr) to track pending work and promises for AsyncWaitForNoPendingRequests(). This allows OnBulkApplyDone to safely satisfy admission and completion promises outside of the main class lock (mu_) and without accessing this, preventing premature destruction or race conditions when user callbacks destroy the MutationBatcher or submit follow-up work. Additionally, several comprehensive unit tests have been added to verify these concurrent behaviors and edge cases. I have no further feedback to provide on these changes.