mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-25 21:46:50 +00:00
fix(heal): canonicalize checkpoint integrity digest
This commit is contained in:
@@ -615,7 +615,18 @@ impl CheckpointManager {
|
|||||||
fn serialize_without_digest(checkpoint: &ResumeCheckpoint) -> Result<Vec<u8>> {
|
fn serialize_without_digest(checkpoint: &ResumeCheckpoint) -> Result<Vec<u8>> {
|
||||||
let mut unsigned = checkpoint.clone();
|
let mut unsigned = checkpoint.clone();
|
||||||
unsigned.integrity_digest = None;
|
unsigned.integrity_digest = None;
|
||||||
serde_json::to_vec(&unsigned).map_err(|e| Error::TaskExecutionFailed {
|
let mut value = serde_json::to_value(&unsigned).map_err(|e| Error::TaskExecutionFailed {
|
||||||
|
message: format!("Failed to serialize checkpoint: {e}"),
|
||||||
|
})?;
|
||||||
|
for field in ["processed_objects", "failed_objects", "skipped_objects"] {
|
||||||
|
let Some(values) = value.get_mut(field).and_then(serde_json::Value::as_array_mut) else {
|
||||||
|
return Err(Error::TaskExecutionFailed {
|
||||||
|
message: format!("Failed to canonicalize checkpoint field: {field}"),
|
||||||
|
});
|
||||||
|
};
|
||||||
|
values.sort_by(|left, right| left.as_str().cmp(&right.as_str()));
|
||||||
|
}
|
||||||
|
serde_json::to_vec(&value).map_err(|e| Error::TaskExecutionFailed {
|
||||||
message: format!("Failed to serialize checkpoint: {e}"),
|
message: format!("Failed to serialize checkpoint: {e}"),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1786,6 +1786,24 @@ async fn checkpoint_integrity_survives_missing_legacy_sidecar() {
|
|||||||
temp_dir.close().unwrap();
|
temp_dir.close().unwrap();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn checkpoint_integrity_survives_multi_object_reload() {
|
||||||
|
let (temp_dir, disk) = schema_test_disk().await;
|
||||||
|
let task_id = ResumeUtils::generate_task_id();
|
||||||
|
let manager = CheckpointManager::new(disk.clone(), task_id.clone()).await.unwrap();
|
||||||
|
for index in 0..32 {
|
||||||
|
manager.add_processed_object(format!("processed-{index}")).await.unwrap();
|
||||||
|
manager.add_failed_object(format!("failed-{index}")).await.unwrap();
|
||||||
|
manager.add_skipped_object(format!("skipped-{index}")).await.unwrap();
|
||||||
|
}
|
||||||
|
manager.update_position(2, 9).await.unwrap();
|
||||||
|
|
||||||
|
CheckpointManager::load_from_disk(disk, &task_id)
|
||||||
|
.await
|
||||||
|
.expect("a healthy multi-object checkpoint must survive reload");
|
||||||
|
temp_dir.close().unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn new_checkpoint_manager_rebuilds_an_empty_snapshot() {
|
async fn new_checkpoint_manager_rebuilds_an_empty_snapshot() {
|
||||||
let (temp_dir, disk) = schema_test_disk().await;
|
let (temp_dir, disk) = schema_test_disk().await;
|
||||||
|
|||||||
Reference in New Issue
Block a user