fix: avoid bthread mutex deadlock between tx processors and bthread waiters

[liunyl]

Problem

Same bug class as the CkptTsCc hang fixed in #483, found in several more CC requests after the issue reappeared in the field.

A bthread::Mutex/ConditionVariable shared between CcRequest::Execute() and a bthread owner can deadlock under our brpc fork:

1. Execute() runs on the brpc worker main stack (via ProcessModulesTask() → TxProcessor::RunOneRound(), with a fixed 1:1 worker↔shard binding), so a contended lock() there parks the worker as a pthread waiter in the mutex's butex queue.
2. The owner bthread waits on the same mutex/cv. butex_wake prefers bound-bthread waiters, and bound bthreads can only run on their home worker (_bound_rq is never stolen).
3. When an unlock routes the wake to a bthread bound to the blocked worker, the bthread sits in that worker's bound run queue forever while the worker sleeps in butex_wait. The mutex may even end up free — nobody can take it. wait_for timeouts don't help: in the end state no timer is pending.

The dangerous shape is concurrent multi-acquisition (a request broadcast to multiple cores, or one mutex shared by several in-flight requests) combined with a bthread waiter. Single-flight/single-waiter requests are structurally safe.

Fix

Convert the affected requests to the pattern used by the CkptTsCc fix (#483): shard-side completion is tracked with atomics, the waiter polls with bthread_usleep exponential backoff (100us → 100ms), and no bthread::Mutex is shared between tx processors and bthreads.

Request	Change
ActiveTxMaxTsCc (used by ResetStandbySequenceId RPC)	atomic countdown + polling Wait()
WaitableCc (covers OnLeaderStart / SubscribePrimaryNode / UpdateInMemoryClusterConfig / standby RPC handlers, etc.)	blocking Wait() polls; IsFinished/IsError/ErrorCode read atomics; mux_ retained only for the coroutine yield/resume handshake whose waiter is a flush-worker std::thread
ClearCcNodeGroup	atomic counter + done_ flag set after the last core's catalog/range/bucket cleanup
EscalateStandbyCcmCc	atomic countdown + polling Wait()
ClearTxCc	atomic countdown + polling Wait() (was std::mutex; safe today since the waiter is the recovery std::thread, but a latent landmine for any future bthread caller)
DbSizeCc / RemoteDbSizeCc	atomic ref counts; Wait() keeps the give-up-on-remote-after-2s semantics; remote_ref_cnt_ decremented before total_ref_cnt_ so the give-up condition never fires while a local core is pending
UploadBatchCc + CcNodeService::UploadBatch + sk_generator upload path	external coordination state (req_mux/req_cv/finished_req_cnt/req_result/ng_term) replaced with std::atomic references; ValidTermCheck() uses CAS; the UploadBatchClosure callback (runs on a bthread) updates atomics lock-free

Left intentionally unchanged (structurally safe): CheckTxStatusCc, WaitNoNakedBucketRefCc, UploadRangeSlicesCc, UploadBatchSlicesCc (single-flight + single waiter), UpdateCceCkptTsCc and the flush-data coroutine path (waiter is a dedicated std::thread).

Verification

• Full Debug build of eloqkv (with this tx_service) passes: install/build.sh exit 0.
• Recommended runtime coverage: standby subscribe/resubscribe, leader failover/step-down, UploadBatch during bucket migration, and the Redis dbsize command — these are the paths that previously hung.

🤖 Generated with Claude Code

Summary by CodeRabbit

• Refactor
  • Reworked internal completion/error coordination across transaction-control flows to use atomic, lock-free tracking instead of lock/condition-variable synchronization.
  • Updated wait behavior to use bounded exponential-backoff polling, improving stability and avoiding cross-context wait issues.
  • Added explicit Wait() support for clearer completion handling in CC request paths.
• Tests
  • Added coverage for wait completion behavior, error latching, and repeated concurrent completion rounds.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: f0744e4a-e3d8-450a-a9da-2a9d188a7510

📥 Commits

Reviewing files that changed from the base of the PR and between 1448c93 and 302ef9a.

📒 Files selected for processing (10)

• tx_service/include/cc/cc_req_misc.h
• tx_service/include/cc/cc_request.h
• tx_service/include/sk_generator.h
• tx_service/src/cc/cc_req_misc.cpp
• tx_service/src/cc/cc_request.cpp
• tx_service/src/remote/cc_node_service.cpp
• tx_service/src/remote/remote_cc_request.cpp
• tx_service/src/sk_generator.cpp
• tx_service/tests/CMakeLists.txt
• tx_service/tests/CcRequestWait-Test.cpp

✅ Files skipped from review due to trivial changes (1)

• tx_service/tests/CMakeLists.txt

🚧 Files skipped from review as they are similar to previous changes (8)

• tx_service/include/sk_generator.h
• tx_service/src/cc/cc_req_misc.cpp
• tx_service/src/remote/cc_node_service.cpp
• tx_service/src/remote/remote_cc_request.cpp
• tx_service/include/cc/cc_req_misc.h
• tx_service/src/cc/cc_request.cpp
• tx_service/src/sk_generator.cpp
• tx_service/include/cc/cc_request.h

---

Walkthrough

The PR replaces mutex/condition-variable completion coordination with atomic counters and polling waits across CC request types, updates UploadBatch and SK upload flows to use atomic shared state, and adds tests for the new WaitableCc behavior.

Changes

Lock-free CC request synchronization

Layer / File(s)	Summary
WaitableCc atomic completion state tx_service/include/cc/cc_req_misc.h	Moves WaitableCc completion and error tracking to atomic members, updates reset and status accessors to use atomic operations, and adds polling-based Wait().
ClearCcNodeGroup polling completion tx_service/include/cc/cc_req_misc.h, tx_service/src/cc/cc_req_misc.cpp	Replaces ClearCcNodeGroup mutex/CV completion tracking with atomic finish and done state, and implements Wait() as exponential-backoff polling.
Counter-based CC waiters tx_service/include/cc/cc_request.h, tx_service/src/cc/cc_request.cpp	Converts ActiveTxMaxTsCc, ClearTxCc, EscalateStandbyCcmCc, and CkptTsCc to atomic completion counters and adds polling-based Wait() helpers.
UploadBatchCc atomic contract tx_service/include/cc/cc_request.h	Updates UploadBatchCc reset contracts to accept atomic shared state, rewires stored pointers to atomic members, and uses atomic term validation and completion/error publication.
DbSize atomic reference counting tx_service/include/cc/cc_request.h, tx_service/src/cc/cc_request.cpp, tx_service/src/remote/remote_cc_request.cpp	Changes DbSizeCc and RemoteDbSizeCc to atomic reference counters and atomic object-size storage, updates decrement and term checks, and adds bounded polling Wait() plus atomic last-core cleanup.
SK upload atomic coordination tx_service/include/sk_generator.h, tx_service/src/sk_generator.cpp	Updates UploadIndexContext and SendIndexes to use atomic per-node-group terms, atomic finished and result state, and polling waits for local and remote upload completion paths.
RPC handler UploadBatch atomic polling tx_service/src/remote/cc_node_service.cpp	Changes CcNodeService::UploadBatch to pass atomic completion and term state into UploadBatchCc and wait for request completion with exponential-backoff polling.
WaitableCc wait tests tx_service/tests/CMakeLists.txt, tx_service/tests/CcRequestWait-Test.cpp	Adds the Catch2 test target and a new WaitableCc suite covering completion, error latching, reset behavior, and repeated concurrent rounds.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

Possibly related PRs

• eloqdata/tx_service#203: Modifies UploadBatchCc::Reset(...) and the same request-plumbing path that this PR updates to atomic shared state.
• eloqdata/tx_service#436: Touches ActiveTxMaxTsCc, which this PR changes from mutex/CV waiting to atomic completion counters.
• eloqdata/tx_service#454: Overlaps with the UploadBatch/UploadIndex coordination flow that this PR refactors for atomic polling.

Suggested reviewers

• thweetkomputer
• MrGuin

Poem

🐰 I hopped through locks and found a way,

With atomics bright, the counts obey.
I nibbled waits till they were small,
Then polled with care to finish all—
A cozy hop, and done by day.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 0.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly matches the main change: fixing bthread mutex deadlocks by moving CC coordination to atomics and polling waiters.
Description check	✅ Passed	The description is well structured with problem, fix, and verification, and it mentions tests, but it omits the issue-link/RFC reminders from the template.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch fix-cc-request-bthread-mutex-deadlock

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 5

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

tx_service/include/cc/cc_request.h (1)

8424-8452: ⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Make DbSizeCc::term_ atomic before using it as the stale-response fence.

AddRemoteObjSize() now runs lock-free, but it still compares against a plain term_ while IncTerm() mutates the same field for the next round. A late remote reply can now race the next IncTerm() and either slip into the new request or get dropped nondeterministically.

Suggested fix

     int32_t GetTerm()
     {
-        return term_;
+        return term_.load(std::memory_order_acquire);
     }
     void IncTerm()
     {
-        term_++;
+        term_.fetch_add(1, std::memory_order_acq_rel);
     }
@@
-        if (term != term_)
+        if (term != term_.load(std::memory_order_acquire))
         {
             return;
         }
@@
-    int32_t term_{0};
+    std::atomic<int32_t> term_{0};

Also applies to: 8491-8493

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@tx_service/include/cc/cc_req_misc.h`:
- Around line 924-938: AbortCcRequest stores a non-NO_ERROR into the atomic
error_code_, but Execute currently unconditionally stores NO_ERROR on shard
success and can overwrite a prior failure; change Execute(CcShard &ccs) so it
does not overwrite a previously set non-NO_ERROR by using an atomic
compare-exchange (e.g. load expected = CcErrorCode::NO_ERROR and only store
NO_ERROR via error_code_.compare_exchange_strong if the current value is
NO_ERROR) or simply omit writing NO_ERROR altogether, and ensure FinishOne()
behavior remains unchanged (refer to AbortCcRequest, Execute, error_code_,
FinishOne).

In `@tx_service/include/cc/cc_request.h`:
- Around line 8378-8380: total_obj_sizes_ is currently being updated lock-free
via reinterpret_cast to std::atomic_int64_t and calling atomics (e.g.,
fetch_add/load), which violates the C++ atomic object model; change the storage
so each element is genuinely atomic (replace the underlying std::vector<int64_t>
with std::vector<std::atomic<int64_t>> or use std::atomic_ref against properly
aligned storage) or revert to holding the mutex around non-atomic accesses;
remove any reinterpret_cast<std::atomic_int64_t&>(total_obj_sizes_[idx]) usages
and update callsites that call fetch_add/load on those elements (and related
code paths noted around total_ref_cnt_, remote_ref_cnt_ and the ranges
referenced in the review) to use the real atomic type or guarded access
consistently.

In `@tx_service/src/cc/cc_req_misc.cpp`:
- Around line 513-523: The backoff in ClearCcNodeGroup::Wait stops growing one
step short of the intended max_interval (100000us) because the guard uses a
strict '<' against the doubled interval; change the logic so when doubling would
exceed max_interval you set interval_us to max_interval (or use '<=' to allow
reaching the cap) instead of leaving it at the previous value; update the same
pattern in the similar loop at the other occurrence (around lines shown for
1137-1147) and ensure you refer to done_, interval_us and max_interval in your
fix.

In `@tx_service/src/cc/cc_request.cpp`:
- Line 143: In cc_request.cpp update the LOG(WARNING) message that currently
reads "Waitting timeout for dbsize" to correct the typo — replace "Waitting"
with "Waiting" (i.e., change the LOG(WARNING) string literal). Locate the
LOG(WARNING) call emitting that exact message and adjust the string to "Waiting
timeout for dbsize" (or a clearer variant like "Waiting for dbsize timeout") so
log searches and alerts work as expected.

In `@tx_service/src/sk_generator.cpp`:
- Around line 774-776: res_code is being overwritten unconditionally (e.g., when
setting CcErrorCode::ESTABLISH_NODE_CHANNEL_FAILED), which allows a later
success to reset an earlier failure; change writes to res_code inside
UploadIndexInternal/UploadIndexWorker to latch the first non-NO_ERROR by using
an atomic compare-exchange that only transitions from CcErrorCode::NO_ERROR to
the new error and never writes when res_code is already non-NO_ERROR (do not
allow writes that set it back to NO_ERROR); keep finished_req_cnt updates as-is
and ensure this behavior mirrors UploadBatchCc's contract so a later successful
batch cannot clear an earlier failure.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 2706025e-5534-4d78-8441-9a3e711ad87b

📥 Commits

Reviewing files that changed from the base of the PR and between f62f0fa and 9ff9784.

📒 Files selected for processing (8)

• tx_service/include/cc/cc_req_misc.h
• tx_service/include/cc/cc_request.h
• tx_service/include/sk_generator.h
• tx_service/src/cc/cc_req_misc.cpp
• tx_service/src/cc/cc_request.cpp
• tx_service/src/remote/cc_node_service.cpp
• tx_service/src/remote/remote_cc_request.cpp
• tx_service/src/sk_generator.cpp

[liunyl]

@CodeRabbit review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.