Improve P2P primary oplog persistence handling#2977
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces asynchronous oplog writing and a committed watermark model in RedisOpLogStore to improve HA throughput, replacing the synchronous Lua-script-based writes with a multi-worker async queue. It also updates P2PMasterService to support optional async oplog recording for replica additions and refactors oplog cleanup. The review feedback identifies three critical issues in the new Redis store: a retry loop bug in CompleteWrite that can permanently mark successful writes as failed, a silent data loss risk in ReadOpLogSince when reading trimmed sequences, and a thread-blocking issue in UpdateLatestSequenceId where in-flight writes are not properly marked as completed.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| ErrorCode latest_err = AdvanceCommittedLatestUnlocked(ctx); | ||
| if (latest_err != ErrorCode::OK) { | ||
| pending_writes_.push_front(pending); | ||
| async_cv_.notify_one(); | ||
| } |
There was a problem hiding this comment.
If AdvanceCommittedLatestUnlocked fails (returning a non-OK error code), the pending write is pushed back to the front of pending_writes_ to be retried. However, because the entry was already successfully written to Redis during the first attempt, the retried PersistEntryNoLatest call will execute SET NX and fail with REDIS_REPLY_NIL (key already exists). This returns ErrorCode::INVALID_PARAMS, which permanently marks the write as failed, even though the entry was successfully persisted.
To fix this, we should either implement idempotency checks in PersistEntryNoLatest (e.g., treating identical existing content as success) or avoid retrying the write itself when only the watermark advancement failed.
| const uint64_t effective_start = | ||
| std::max(start_sequence_id, trimmed_sequence_id); | ||
| if (effective_start >= latest_sequence_id) { | ||
| return ErrorCode::OK; | ||
| } |
There was a problem hiding this comment.
If start_sequence_id < trimmed_sequence_id, it means the requested sequence range has already been deleted (trimmed) from Redis. Silently taking std::max(start_sequence_id, trimmed_sequence_id) and returning the remaining entries starting from trimmed_sequence_id + 1 will cause the standby to silently miss mutations. This leads to metadata inconsistency and silent data corruption.
Instead, we should return ErrorCode::OPLOG_ENTRY_NOT_FOUND when start_sequence_id < trimmed_sequence_id so the standby knows it has fallen behind the trim horizon and must trigger a full bootstrap/snapshot sync.
| const uint64_t effective_start = | |
| std::max(start_sequence_id, trimmed_sequence_id); | |
| if (effective_start >= latest_sequence_id) { | |
| return ErrorCode::OK; | |
| } | |
| if (start_sequence_id < trimmed_sequence_id) { | |
| LOG(ERROR) << "RedisOpLogStore::ReadOpLogSince: requested sequence has been trimmed" | |
| << ", start_sequence_id=" << start_sequence_id | |
| << ", trimmed_sequence_id=" << trimmed_sequence_id; | |
| return ErrorCode::OPLOG_ENTRY_NOT_FOUND; | |
| } | |
| const uint64_t effective_start = start_sequence_id; | |
| if (effective_start >= latest_sequence_id) { | |
| return ErrorCode::OK; | |
| } |
There was a problem hiding this comment.
We’ll address trim-horizon detection together with standby snapshot/bootstrap recovery in a follow-up change. Added a TODO for now.
| { | ||
| std::lock_guard<std::mutex> async_lock(async_mutex_); | ||
| committed_sequence_id_ = sequence_id; | ||
| persisted_sequences_.erase( | ||
| persisted_sequences_.begin(), | ||
| persisted_sequences_.upper_bound(sequence_id)); | ||
| } |
There was a problem hiding this comment.
When committed_sequence_id_ is advanced administratively via UpdateLatestSequenceId, any matching PendingWrites in inflight_writes_ that have sequence_id <= sequence_id are not marked as done = true or erased. This causes the callers of EnqueueWrite to remain blocked waiting on sync_cv_ until they timeout (30 seconds), even though their writes are now considered committed. Furthermore, if the worker thread later fails to write them, they will be marked as failed, which is incorrect.
We should iterate through inflight_writes_ and mark any writes with sequence_id <= sequence_id as successful and erase them.
{
std::lock_guard<std::mutex> async_lock(async_mutex_);
committed_sequence_id_ = sequence_id;
persisted_sequences_.erase(
persisted_sequences_.begin(),
persisted_sequences_.upper_bound(sequence_id));
for (auto it = inflight_writes_.begin(); it != inflight_writes_.end(); ) {
if (it->first <= sequence_id) {
it->second->done = true;
it->second->result = ErrorCode::OK;
it = inflight_writes_.erase(it);
} else {
++it;
}
}
}There was a problem hiding this comment.
This is a recovery/admin API inherited from OpLogStore and is not used in the normal Redis write path. Added a comment clarifying that it must not run concurrently with writes.
|
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
2 participants
Description
Improve P2P primary metadata OpLog persistence handling.
OpLogManager.Module
mooncake-transfer-engine)mooncake-store)mooncake-ep)mooncake-pg)mooncake-integration)mooncake-p2p-store)mooncake-wheel)mooncake-common)mooncake-rl)Type of Change
How Has This Been Tested?
Built and tested in the local development container with Redis support enabled.
Test commands:
Test results:
Checklist
AI Assistance Disclosure
Codex was used to help rebase the branch, review the changes, run local formatting/build/tests, and prepare the PR description. The submitter reviewed the resulting changes.