gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-24 05:06:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): satisfy checkpoint outcome lint

Browse Source
This commit is contained in:
马登山
2026-08-23 08:14:07 +08:00
parent 276d8e156f
commit 37c5fef252
4 changed files with 81 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/erasure_healer.rs
+44 -44
View File
@@ -15,8 +15,8 @@
use crate::heal::{ use crate::heal::{
progress::{HealProgress, add_bytes, increment_counter}, progress::{HealProgress, add_bytes, increment_counter},
resume::{ resume::{
CheckpointManager, CheckpointObjectOutcome, ReplacementTargetIdentity, ResumeManager, ResumeUtils, compose_key, CheckpointManager, CheckpointObjectOutcome, CheckpointObjectOutcomeRecord, ReplacementTargetIdentity, ResumeManager,
replacement_target_identities_match, ResumeUtils, compose_key, replacement_target_identities_match,
}, },
storage::{HealStorageAPI, next_heal_listing_token}, storage::{HealStorageAPI, next_heal_listing_token},
task::{demote_to_debug_when, is_missing_object_dir_heal_result, take_failure_log_sample}, task::{demote_to_debug_when, is_missing_object_dir_heal_result, take_failure_log_sample},
@@ -909,17 +909,17 @@ impl ErasureSetHealer {
(progress.skipped_new_versions, progress.skipped_ilm_expired, progress.counter_unknown) (progress.skipped_new_versions, progress.skipped_ilm_expired, progress.counter_unknown)
}; };
checkpoint_manager checkpoint_manager
.record_object_outcome( .record_object_outcome(CheckpointObjectOutcomeRecord {
key, object: key,
CheckpointObjectOutcome::Processed, outcome: CheckpointObjectOutcome::Processed,
*successful_objects, successful: *successful_objects,
*failed_objects, failed: *failed_objects,
*skipped_objects, skipped: *skipped_objects,
bytes_processed, bytes: bytes_processed,
skipped_new, skipped_new_versions: skipped_new,
skipped_ilm, skipped_ilm_expired: skipped_ilm,
!counter_ok || counter_unknown, counter_unknown: !counter_ok || counter_unknown,
) })
.await?; .await?;
if !counter_ok || counter_unknown { if !counter_ok || counter_unknown {
resume_manager.mark_counter_unknown().await?; resume_manager.mark_counter_unknown().await?;
@@ -974,17 +974,17 @@ impl ErasureSetHealer {
(progress.skipped_new_versions, progress.skipped_ilm_expired, progress.counter_unknown) (progress.skipped_new_versions, progress.skipped_ilm_expired, progress.counter_unknown)
}; };
checkpoint_manager checkpoint_manager
.record_object_outcome( .record_object_outcome(CheckpointObjectOutcomeRecord {
key, object: key,
CheckpointObjectOutcome::Processed, outcome: CheckpointObjectOutcome::Processed,
*successful_objects, successful: *successful_objects,
*failed_objects, failed: *failed_objects,
*skipped_objects, skipped: *skipped_objects,
bytes_processed, bytes: bytes_processed,
skipped_new, skipped_new_versions: skipped_new,
skipped_ilm, skipped_ilm_expired: skipped_ilm,
!counter_ok || counter_unknown, counter_unknown: !counter_ok || counter_unknown,
) })
.await?; .await?;
if !counter_ok || counter_unknown { if !counter_ok || counter_unknown {
resume_manager.mark_counter_unknown().await?; resume_manager.mark_counter_unknown().await?;
@@ -1204,17 +1204,17 @@ impl ErasureSetHealer {
(progress.counter_unknown, progress.skipped_new_versions, progress.skipped_ilm_expired) (progress.counter_unknown, progress.skipped_new_versions, progress.skipped_ilm_expired)
}; };
checkpoint_manager checkpoint_manager
.record_object_outcome( .record_object_outcome(CheckpointObjectOutcomeRecord {
key, object: key,
checkpoint_outcome, outcome: checkpoint_outcome,
*successful_objects, successful: *successful_objects,
*failed_objects, failed: *failed_objects,
*skipped_objects, skipped: *skipped_objects,
bytes_processed, bytes: bytes_processed,
skipped_new_versions, skipped_new_versions,
skipped_ilm_expired, skipped_ilm_expired,
telemetry_unknown || progress_unknown, counter_unknown: telemetry_unknown || progress_unknown,
) })
.await?; .await?;
if telemetry_unknown || progress_unknown { if telemetry_unknown || progress_unknown {
resume_manager.mark_counter_unknown().await?; resume_manager.mark_counter_unknown().await?;
@@ -2413,17 +2413,17 @@ mod resume_loop_tests {
}, },
); );
env.checkpoint env.checkpoint
.record_object_outcome( .record_object_outcome(CheckpointObjectOutcomeRecord {
compose_key("object", Some("v1")), object: compose_key("object", Some("v1")),
CheckpointObjectOutcome::Failed, outcome: CheckpointObjectOutcome::Failed,
0, successful: 0,
1, failed: 1,
0, skipped: 0,
0, bytes: 0,
0, skipped_new_versions: 0,
0, skipped_ilm_expired: 0,
false, counter_unknown: false,
) })
.await .await
.unwrap(); .unwrap();
env.checkpoint.advance_page(0, 1).await.unwrap(); env.checkpoint.advance_page(0, 1).await.unwrap();
crates/heal/src/heal/resume.rs
+1 -1
View File
@@ -31,7 +31,7 @@ mod checkpoint;
mod replacement; mod replacement;
mod utils; mod utils;
pub use checkpoint::{CheckpointManager, CheckpointObjectOutcome, ResumeCheckpoint}; pub use checkpoint::{CheckpointManager, CheckpointObjectOutcome, CheckpointObjectOutcomeRecord, ResumeCheckpoint};
pub(crate) use replacement::replacement_target_identities_match; pub(crate) use replacement::replacement_target_identities_match;
use replacement::replacement_targets_match_identities; use replacement::replacement_targets_match_identities;
pub use replacement::{ pub use replacement::{
crates/heal/src/heal/resume/checkpoint.rs
+25 -12
View File
@@ -41,6 +41,19 @@ pub enum CheckpointObjectOutcome {
Skipped, Skipped,
} }
#[derive(Debug)]
pub struct CheckpointObjectOutcomeRecord {
pub object: String,
pub outcome: CheckpointObjectOutcome,
pub successful: u64,
pub failed: u64,
pub skipped: u64,
pub bytes: u64,
pub skipped_new_versions: u64,
pub skipped_ilm_expired: u64,
pub counter_unknown: bool,
}
/// resume checkpoint /// resume checkpoint
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResumeCheckpoint { pub struct ResumeCheckpoint {
@@ -389,18 +402,18 @@ impl CheckpointManager {
} }
/// Atomically persist an object's dedup identity with its aggregate result. /// Atomically persist an object's dedup identity with its aggregate result.
pub async fn record_object_outcome( pub async fn record_object_outcome(&self, record: CheckpointObjectOutcomeRecord) -> Result<()> {
&self, let CheckpointObjectOutcomeRecord {
object: String, object,
outcome: CheckpointObjectOutcome, outcome,
successful: u64, successful,
failed: u64, failed,
skipped: u64, skipped,
bytes: u64, bytes,
skipped_new_versions: u64, skipped_new_versions,
skipped_ilm_expired: u64, skipped_ilm_expired,
counter_unknown: bool, counter_unknown,
) -> Result<()> { } = record;
let mut checkpoint = self.checkpoint.write().await; let mut checkpoint = self.checkpoint.write().await;
match outcome { match outcome {
CheckpointObjectOutcome::Processed => checkpoint.add_processed_object(object), CheckpointObjectOutcome::Processed => checkpoint.add_processed_object(object),
crates/heal/src/heal/resume/tests.rs
+11 -11
View File
@@ -1483,17 +1483,17 @@ async fn checkpoint_page_commit_keeps_ledger_until_cursor_is_durable() {
let checkpoint = CheckpointManager::new(disk.clone(), task_id.clone()).await.unwrap(); let checkpoint = CheckpointManager::new(disk.clone(), task_id.clone()).await.unwrap();
checkpoint checkpoint
.record_object_outcome( .record_object_outcome(CheckpointObjectOutcomeRecord {
"bucket/object:v1".to_string(), object: "bucket/object:v1".to_string(),
CheckpointObjectOutcome::Processed, outcome: CheckpointObjectOutcome::Processed,
1, successful: 1,
0, failed: 0,
0, skipped: 0,
128, bytes: 128,
0, skipped_new_versions: 0,
0, skipped_ilm_expired: 0,
false, counter_unknown: false,
) })
.await .await
.unwrap(); .unwrap();
checkpoint.advance_page(0, 1).await.unwrap(); checkpoint.advance_page(0, 1).await.unwrap();