From 99701e9f52eaafecccd73b3a2a35bf66423dd4f9 Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 4 Aug 2026 07:22:19 +0800 Subject: [PATCH] fix(replication): fence journal snapshots with CAS (#5674) --- .mimocode/skills/issue-triage/SKILL.md | 116 ++++++++++++++ .mimocode/skills/pr-review/SKILL.md | 147 ++++++++++++++++++ .../bucket/replication/replication_pool.rs | 97 +++++++++--- 3 files changed, 335 insertions(+), 25 deletions(-) create mode 100644 .mimocode/skills/issue-triage/SKILL.md create mode 100644 .mimocode/skills/pr-review/SKILL.md diff --git a/.mimocode/skills/issue-triage/SKILL.md b/.mimocode/skills/issue-triage/SKILL.md new file mode 100644 index 000000000..d6976255d --- /dev/null +++ b/.mimocode/skills/issue-triage/SKILL.md @@ -0,0 +1,116 @@ +--- +name: issue-triage +description: Triage a GitHub issue — determine if it is already fixed, needs implementation, or should be closed. Searches related commits and PRs, verifies implementation status, and posts a triage comment or closes the issue. Use when the user provides an issue URL and asks whether it can be closed or needs work. +--- + +# Issue Triage + +Use this skill when the user provides a GitHub issue URL and asks "can this be closed?", "is this already implemented?", "check completion status", or similar triage questions. + +## Workflow + +### 1. Fetch issue context + +```bash +gh issue view --repo --json title,body,state,comments,labels,updatedAt +``` + +Read the issue body to understand what was requested. Extract: +- The specific feature/fix/behavior described. +- Any linked PRs or commits mentioned in the body or comments. +- Any checklist items or sub-issues. + +### 2. Search for related work + +Search git history for commits referencing the issue: +```bash +git log --oneline --all --grep="" | head -30 +``` + +Search for related PRs: +```bash +gh pr list --search "fixes # OR closes # OR #" --state all --json number,title,state,mergedAt +``` + +If the issue mentions specific PRs, check their status: +```bash +gh pr view --json state,mergedAt,title +``` + +### 3. Verify implementation + +For each linked or related PR that is merged, verify the fix is actually present on the current main branch: +```bash +git log --oneline main | grep -i "" +# or +git log --oneline main --grep="" +``` + +If the issue describes a specific defect, check the relevant code to confirm the fix is in place: +```bash +grep -n "" crates//src/.rs +``` + +For issues with checklists, verify each item individually. If sub-items are tracked as separate issues, check those too: +```bash +gh issue view --repo --json state +``` + +### 4. Determine verdict + +- **All items fixed and merged**: Close with a summary comment listing what was fixed and which PRs. +- **Some items fixed, some remaining**: Comment with status of each item. Do not close. +- **Not yet implemented**: Comment with a summary of what remains. Do not close. +- **Superseded or no longer relevant**: Close with explanation. + +### 5. Take action + +Close with comment: +```bash +gh issue close --repo --comment "" +``` + +Comment without closing: +```bash +gh issue comment --repo --body-file /tmp/triage.md +``` + +Update issue labels if needed: +```bash +gh issue edit --repo --add-label "completed" --remove-label "needs-triage" +``` + +Always use `--body-file` for multiline content, never inline `--body`. + +### 6. Handle multi-issue batches + +When the user asks to check multiple issues (e.g., "check all issues by user X" or "scan backlog for closable issues"): +1. List the issues: `gh issue list --repo --author --state open --json number,title,updatedAt` +2. For each issue, run steps 1-5 above. +3. Report a summary table of all triaged issues with verdicts. + +## Output format + +### Issue Triage: # — + +**State**: OPEN / CLOSED +**Linked PRs**: <list with merge status> + +#### Assessment +<what was requested vs what is implemented> + +#### Verdict +- Close — all items resolved by <PR list> +- Keep open — <remaining items> +- Not started — <what needs to be done> + +#### Action taken +- Closed with comment / Commented / No action + +## Notes + +- The user may ask in Chinese ("是否可以关闭", "检查完成情况"); respond in the same language. +- When closing, always include a summary of what was fixed and which PRs resolved it — this creates a useful audit trail. +- For issues in `rustfs/backlog`, use `--repo rustfs/backlog`. +- For issues in `rustfs/rustfs`, use `--repo rustfs/rustfs`. +- If the issue has sub-issues (GitHub sub-issues API), check each one's state before declaring the parent complete. diff --git a/.mimocode/skills/pr-review/SKILL.md b/.mimocode/skills/pr-review/SKILL.md new file mode 100644 index 000000000..824590681 --- /dev/null +++ b/.mimocode/skills/pr-review/SKILL.md @@ -0,0 +1,147 @@ +--- +name: pr-review +description: Review a GitHub PR end-to-end from a URL or number — fetch metadata, inspect the diff, run multi-role adversarial review, check CI status, and post the review comment. Use when the user provides a PR link and asks to review it. +--- + +# PR Review + +Use this skill when the user provides a GitHub PR URL or number and asks to review it. This covers the full review lifecycle: data gathering, code review, CI verification, and posting the result. + +## Prerequisites + +- Read `AGENTS.md` for the repository's adversarial validation policy and change-style rules. +- The `adversarial-validation` skill handles the review role playbooks; this skill orchestrates the workflow around it. + +## Workflow + +### 1. Gather PR context + +```bash +gh pr view <N> --json title,author,state,body,additions,deletions,changedFiles,commits,baseRefName,headRefName +gh pr diff <N> --name-only +``` + +Read the PR body and linked issues to understand the change's purpose. If the PR references an issue, fetch that too: +```bash +gh issue view <ISSUE> --json title,body,state +``` + +### 2. Fetch the diff and classify the change + +```bash +git fetch origin pull/<N>/head:pr-<N> +git diff main...pr-<N> --stat +``` + +Classify the change by risk tier (per AGENTS.md): +- **Exempt**: docs/comments/instruction-only, formatting, typos. +- **Mechanical**: renames, file moves, test-only or tooling changes. +- **Standard** (default): any behavior change. +- **High risk**: locking, erasure coding, quorum/heal, replication, multipart, RPC, lifecycle/tiering, metadata formats, persistence/fsync, IAM/KMS/auth, on-disk/on-wire formats, S3 API-visible behavior. + +### 3. Cluster changed files and delegate review + +Group the changed files into logical clusters (by crate or functional area). For each cluster, spawn a subagent with a focused review prompt that includes: +- The cluster's changed files and their diffs. +- The applicable adversarial role probes (from the `adversarial-validation` skill). +- The repository's AGENTS.md rules relevant to that domain. + +For standard-tier changes: correctness adversary + simplicity adversary + test-coverage skeptic, plus every role whose domain the diff touches. +For high-risk changes: run all seven roles. + +Each subagent must produce findings (concrete failure scenario with file:line) or a null report ("attacked X, Y, Z — no break found"). + +### 4. Check CI status + +```bash +gh pr checks <N> +``` + +If any checks fail, investigate: +```bash +gh run view --log-failed --job=<JOB_ID> +``` + +Determine whether failures are pre-existing (on main), flaky, or caused by the PR. + +### 5. Synthesize findings + +Combine all subagent findings into a structured review: +- **Summary**: one-paragraph overview of the change and overall assessment. +- **Findings**: each finding with severity (critical/major/minor/nit), file:line, concrete failure scenario, and suggested fix. +- **CI status**: pass/fail with notes on any failures. +- **Verdict**: APPROVE, REQUEST_CHANGES, or COMMENT. + +### 6. Post the review + +Write the review body to a temp file and post via CLI: +```bash +# Request changes +gh pr review <N> --request-changes --body-file /tmp/pr_review.md + +# Approve +gh pr review <N> --approve --body-file /tmp/pr_review.md + +# Comment only (no verdict) +gh pr review <N> --comment --body-file /tmp/pr_review.md +``` + +For inline comments on specific lines, use the GitHub API: +```bash +cat > /tmp/pr_review.json <<'EOF' +{ + "body": "review body", + "event": "REQUEST_CHANGES", + "comments": [ + { + "path": "crates/foo/src/bar.rs", + "line": 42, + "body": "finding description" + } + ] +} +EOF +gh api --method POST /repos/{owner}/{repo}/pulls/<N>/reviews --input /tmp/pr_review.json +``` + +Always use `--body-file` or `--input`, never inline multiline `--body`. + +### 7. Handle follow-up + +If the review requests changes: +- Monitor for new commits: `gh pr view <N> --json commits` +- Re-review changed files only: `git diff pr-<N>..origin/pull/<N>/head` +- Update the review when findings are addressed. + +If CI was failing due to pre-existing main breakage: +- Comment on the PR noting the failure is pre-existing. +- Suggest updating the branch: `gh pr update-branch <N>` + +## Output format + +### PR Review: #<N> — <title> + +**Author**: <author> +**Risk tier**: exempt | mechanical | standard | high-risk +**Changed files**: <count> across <cluster count> clusters + +#### Summary +<one-paragraph overview> + +#### Findings +| Severity | Location | Finding | +|----------|----------|---------| +| critical | file:line | concrete failure scenario | + +#### CI Status +- All checks pass / Failing: <details> + +#### Verdict +APPROVE / REQUEST_CHANGES / COMMENT + +## Notes + +- The user may ask for review in Chinese; respond in the same language but keep the review body in English per AGENTS.md rules. +- When the user asks for "多角色对抗 review", run the full adversarial validation protocol — this skill's step 3 covers that. +- If the PR is from a fork, check `maintainerCanModify` before attempting to push fixes. +- For very large PRs (>50 files), cluster aggressively and delegate in parallel to keep review time reasonable. diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index cb6546c76..78b945670 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -534,7 +534,6 @@ where result => return result, } } - Err(EcstoreError::other("force-delete journal update retries exhausted")) } @@ -587,6 +586,71 @@ fn ensure_force_delete_journal_lock_held(lock_lost: bool) -> Result<(), EcstoreE Ok(()) } +async fn write_mrf_journal_snapshot<S: ReplicationStorage>( + storage: Arc<S>, + desired: &[MrfReplicateEntry], +) -> Result<(), EcstoreError> { + let file = ReplicationMetadataStore::MRF_REPLICATION_FILE; + let mut merged = desired.to_vec(); + let mut saw_conflict = false; + for _attempt in 0..=FORCE_DELETE_INTENT_CAS_RETRIES { + let lock = storage + .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), file) + .await?; + let guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?; + let current = ReplicationConfigStore::read_no_lock_with_metadata(storage.clone(), file).await; + let etag = match current { + Ok((data, object_info)) => { + if saw_conflict { + let current = decode_mrf_file(&data)?; + for entry in current { + if !merged.iter().any(|existing| mrf_entries_same(existing, &entry)) { + merged.push(entry); + } + } + } + object_info.etag + } + Err(EcstoreError::ConfigNotFound) => None, + Err(err) => return Err(err), + }; + if guard.is_lock_lost() { + return Err(EcstoreError::other("MRF journal namespace lock was lost before commit")); + } + let preconditions = match etag.filter(|value| !value.trim().is_empty()) { + Some(etag) => HTTPPreconditions { + if_match: Some(etag), + ..Default::default() + }, + None => HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }, + }; + let data = if merged.is_empty() { + Vec::new() + } else { + encode_mrf_file(&merged)? + }; + match ReplicationConfigStore::save_conditional(storage.clone(), file, data, preconditions).await { + Ok(()) => return Ok(()), + Err(EcstoreError::PreconditionFailed) => saw_conflict = true, + Err(err) => return Err(err), + } + } + Err(EcstoreError::PreconditionFailed) +} + +fn mrf_entries_same(left: &MrfReplicateEntry, right: &MrfReplicateEntry) -> bool { + left.bucket == right.bucket + && left.object == right.object + && left.version_id == right.version_id + && left.op == right.op + && left.target_arns == right.target_arns + && left.force_delete_id == right.force_delete_id + && left.delete_marker_version_id == right.delete_marker_version_id +} + #[derive(Debug, thiserror::Error)] #[error("replication resync {active_resync_id} is already active for {bucket}/{arn}")] struct ResyncActiveConflictError { @@ -2253,34 +2317,19 @@ fn dec_mrf_entries(stats: &ReplicationStats, entries: &[MrfReplicateEntry]) { /// Returns the flush duration on success; on failure logs the error and returns `None`. /// Callers must NOT clear their in-memory buffer on `None` so the next tick /// can retry — otherwise a transient storage error permanently drops the batch. -async fn flush_mrf_to_disk<S: ReplicationObjectIO>(entries: &[MrfReplicateEntry], storage: &Arc<S>) -> Option<u64> { +async fn flush_mrf_to_disk<S: ReplicationStorage>(entries: &[MrfReplicateEntry], storage: &Arc<S>) -> Option<u64> { let started = Instant::now(); - match encode_mrf_file(entries) { - Ok(data) => { - if let Err(e) = - ReplicationConfigStore::save(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE, data).await - { - let duration_millis = duration_millis_u64(started.elapsed()); - observe_mrf_flush_failure(duration_millis); - warn!( - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION, - count = entries.len(), - error = %e, - "Failed to flush MRF entries to disk" - ); - return None; - } - Some(duration_millis_u64(started.elapsed())) - } + match write_mrf_journal_snapshot(storage.clone(), entries).await { + Ok(()) => Some(duration_millis_u64(started.elapsed())), Err(e) => { - observe_mrf_flush_failure(0); + let duration_millis = duration_millis_u64(started.elapsed()); + observe_mrf_flush_failure(duration_millis); warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, count = entries.len(), error = %e, - "Failed to encode MRF entries for disk flush" + "Failed to flush MRF entries to disk" ); None } @@ -2329,9 +2378,7 @@ async fn recover_corrupt_mrf_generation<S: ReplicationStorage>( digest: mrf_payload_digest(&data), entry_count: entries.len(), }); - if let Err(error) = - ReplicationConfigStore::save_no_lock(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE, data).await - { + if let Err(error) = write_mrf_journal_snapshot(storage.clone(), &entries).await { observe_mrf_flush_failure(duration_millis_u64(started.elapsed())); warn!( component = LOG_COMPONENT_ECSTORE,