mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 14:27:10 +00:00
fix(ecstore): recover delete batches with missing versions
Treat an explicit non-null version as a no-op when it is absent from both recovery sources. Keep rejecting missing null versions, delete-marker mismatches and conflicting local generations. Add returning-disk and partial-receipt regressions, extend journal boundary checks, and regenerate the ecstore test-count baseline. Production faults exercise the 24 added assertion sites and 7 semantic checks: 33 faults failed at the expected sites and passed after full restoration. Validation (LANG=en_US.UTF-8, one build job): cargo nextest run --locked -p rustfs-ecstore --lib -E 'test(delete_recovery) | test(quorum_delete_cleanup_coordinator_binds_single_and_batch) | test(crash_delete_retry_batch_reclaims_inline_and_reserved_external)' --test-threads 1 --no-fail-fast Summary [ 34.580s] 69 tests run: 69 passed, 6058 skipped; exit 0 cargo clippy --locked -p rustfs-ecstore --all-targets -- -D warnings Finished `dev` profile [unoptimized + debuginfo] target(s) in 5m 37s; exit 0 All 19 final Quick Checks completed with exit 0: - shellcheck --version && actionlint - cargo fmt --all --check - ./scripts/check_unsafe_code_allowances.sh - ./scripts/check_layer_dependencies.sh - ./scripts/check_architecture_migration_rules.sh - ./scripts/check_logging_guardrails.sh - ./scripts/check_error_other_format_ratchet.sh - ./scripts/check_no_tokio_io_uring.sh - ./scripts/check_extension_schema_boundaries.sh - ./scripts/check_body_cache_whitelist.sh - ./scripts/check_s3s_footprint.sh - ./scripts/check_test_count_ratchet.sh - ./scripts/check_fips_wording.sh - ./scripts/check_embedded_secrets.sh - make script-tests - python3 ./scripts/test_object_generation_protocol_model.py - python3 ./scripts/check_test_wiring.py - ./scripts/check_no_planning_docs.sh - ./scripts/check_uring_lane_lib_only.sh test-count ratchet OK: 52 crates, 22009 tests, 165 ignored match .config/test-count-baseline.txt PASS: cold stampede script self-test model: states=51023, transitions=167520, terminal=732, chosen_candidates=2 multi-slot round schedules: 640 OK: e2e modules, runner selection, fuzz matrices, profiles, scheduled alerts, and Quick Checks are wired
This commit is contained in:
overtrue
@@ -20,7 +20,7 @@ rustfs-connect-symbol-catalog 6 0
|
||||
rustfs-credentials 24 0
|
||||
rustfs-crypto 67 0
|
||||
rustfs-data-usage 54 0
|
||||
rustfs-ecstore 6579 1
|
||||
rustfs-ecstore 6581 1
|
||||
rustfs-extension-schema 23 0
|
||||
rustfs-filemeta 306 0
|
||||
rustfs-heal 856 3
|
||||
|
||||
@@ -2084,7 +2084,7 @@ impl DeleteRecoveryJournal {
|
||||
}
|
||||
let id = request.version_id.unwrap_or_default();
|
||||
let Some(original) = originals.get(&id) else {
|
||||
return false;
|
||||
return !id.is_nil() && !locals.contains_key(&id);
|
||||
};
|
||||
match locals.get(&id) {
|
||||
Some(local) => local == original,
|
||||
@@ -2134,7 +2134,10 @@ impl DeleteRecoveryJournal {
|
||||
};
|
||||
match (local, remote) {
|
||||
(Some(local), Some(remote)) => local == remote,
|
||||
(None, None) => self.original_digest.is_none() && authority.original_digest.is_none(),
|
||||
(None, None) => {
|
||||
(!(request.deleted && request.mark_deleted) && request.version_id.is_some_and(|id| !id.is_nil()))
|
||||
|| (self.original_digest.is_none() && authority.original_digest.is_none())
|
||||
}
|
||||
_ => false,
|
||||
}
|
||||
})
|
||||
@@ -3383,6 +3386,46 @@ mod delete_recovery_adoption_tests {
|
||||
};
|
||||
assert!(journal.matches_adoption_source(&[]), "certified non-null removal can already be absent");
|
||||
assert!(journal.matches_adoption_source(std::slice::from_ref(&source)));
|
||||
let missing = Uuid::new_v4();
|
||||
journal.requests.push(FileInfo {
|
||||
version_id: Some(missing),
|
||||
..Default::default()
|
||||
});
|
||||
journal.original_digest = Some([3; 32]);
|
||||
assert!(
|
||||
journal.matches_prepared_authority(&journal),
|
||||
"an explicit UUID absent from both prepared sources remains a no-op with original metadata"
|
||||
);
|
||||
assert!(journal.matches_adoption_source(&[]), "both non-null versions can remain absent");
|
||||
assert!(
|
||||
journal.matches_adoption_source(std::slice::from_ref(&source)),
|
||||
"a missing non-null version in the batch remains a no-op"
|
||||
);
|
||||
let mut unexpected = source.clone();
|
||||
unexpected.version_id = Some(missing);
|
||||
let mut foreign_preparation = journal.clone();
|
||||
foreign_preparation.original_versions.push(unexpected.clone());
|
||||
assert!(
|
||||
!journal.matches_prepared_authority(&foreign_preparation)
|
||||
&& !foreign_preparation.matches_prepared_authority(&journal),
|
||||
"absence from only one prepared source cannot establish a generation"
|
||||
);
|
||||
assert!(
|
||||
!journal.matches_adoption_source(&[source.clone(), unexpected]),
|
||||
"a version absent from the authority cannot be deleted locally"
|
||||
);
|
||||
for version_id in [None, Some(Uuid::nil())] {
|
||||
journal.requests.last_mut().expect("missing-version request").version_id = version_id;
|
||||
assert!(
|
||||
!journal.matches_prepared_authority(&journal),
|
||||
"a missing null version with original metadata cannot establish its generation"
|
||||
);
|
||||
assert!(
|
||||
!journal.matches_adoption_source(std::slice::from_ref(&source)),
|
||||
"a missing null version cannot establish its generation"
|
||||
);
|
||||
}
|
||||
journal.requests.pop();
|
||||
let mut foreign = source.clone();
|
||||
foreign.metadata_digest = [2; 32];
|
||||
assert!(!journal.matches_adoption_source(std::slice::from_ref(&foreign)));
|
||||
@@ -3398,5 +3441,24 @@ mod delete_recovery_adoption_tests {
|
||||
assert!(journal.matches_adoption_source(std::slice::from_ref(&journal.original_versions[0])));
|
||||
journal.requests.clear();
|
||||
assert!(!journal.matches_adoption_source(&journal.original_versions));
|
||||
|
||||
let mut marker = journal.clone();
|
||||
marker.requests = vec![FileInfo {
|
||||
version_id: Some(id),
|
||||
deleted: true,
|
||||
mark_deleted: true,
|
||||
..Default::default()
|
||||
}];
|
||||
marker.original_versions.clear();
|
||||
marker.original_digest = Some([3; 32]);
|
||||
assert!(
|
||||
!marker.matches_prepared_authority(&marker),
|
||||
"a marker with original metadata requires a source generation"
|
||||
);
|
||||
marker.original_digest = None;
|
||||
assert!(
|
||||
marker.matches_prepared_authority(&marker),
|
||||
"marker preparation without original metadata retains the existing allowance"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1569,6 +1569,86 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn delete_recovery_batch_with_missing_version_recovers_after_preparation_failure() {
|
||||
use crate::disk::RUSTFS_META_TMP_BUCKET;
|
||||
use crate::storage_api_contracts::object::ObjectToDelete;
|
||||
|
||||
let bucket = "recover-batch-missing-version";
|
||||
let (dirs, store, disks) = fixture(bucket).await;
|
||||
let deleted = put(&store, bucket, &vec![48; 4096], true).await.expect("version to delete");
|
||||
let remaining = vec![49; 4096];
|
||||
let retained = put(&store, bucket, &remaining, true).await.expect("version to retain");
|
||||
let missing = Uuid::new_v4();
|
||||
let temporary = dirs[3].path().join(RUSTFS_META_TMP_BUCKET);
|
||||
let saved = dirs[3].path().join("saved-delete-tmp");
|
||||
tokio::fs::rename(&temporary, &saved)
|
||||
.await
|
||||
.expect("save returning disk's staging directory");
|
||||
tokio::fs::write(&temporary, b"block preparation")
|
||||
.await
|
||||
.expect("fail staging after the recovery preflight");
|
||||
|
||||
let (objects, errors) = store
|
||||
.delete_objects(
|
||||
bucket,
|
||||
[deleted, missing]
|
||||
.into_iter()
|
||||
.map(|version| ObjectToDelete {
|
||||
object_name: "object".into(),
|
||||
version_id: Some(version),
|
||||
..Default::default()
|
||||
})
|
||||
.collect(),
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert_eq!(objects.len(), 2);
|
||||
assert!(errors.iter().all(Option::is_none), "batch deletion succeeds at quorum: {errors:?}");
|
||||
assert!(
|
||||
disks[3]
|
||||
.discover_delete_transactions(bucket, Some("object"), None, 100)
|
||||
.await
|
||||
.expect("inspect failed preparation")
|
||||
.transactions
|
||||
.is_empty(),
|
||||
"returning disk must adopt the quorum decision without a local preparation"
|
||||
);
|
||||
tokio::fs::remove_file(&temporary).await.expect("remove preparation failure");
|
||||
tokio::fs::rename(&saved, &temporary)
|
||||
.await
|
||||
.expect("restore returning disk staging");
|
||||
|
||||
let result = store
|
||||
.recover_delete_transactions(bucket, &HealOpts::default())
|
||||
.await
|
||||
.expect("recover batch containing a missing non-null version");
|
||||
assert!(result.detail.contains("blocked=0"), "{}", result.detail);
|
||||
assert_eq!(body(&store, bucket, Some(retained)).await, remaining);
|
||||
for version in [deleted, missing] {
|
||||
assert_eq!(
|
||||
disks[3]
|
||||
.read_version(bucket, bucket, "object", &version.to_string(), &ReadOptions::default())
|
||||
.await
|
||||
.expect_err("returning disk must not retain or create a deleted version"),
|
||||
DiskError::FileVersionNotFound
|
||||
);
|
||||
}
|
||||
for disk in &disks {
|
||||
assert!(
|
||||
disk.discover_delete_transactions(bucket, Some("object"), None, 100)
|
||||
.await
|
||||
.expect("inspect converged batch")
|
||||
.transactions
|
||||
.is_empty(),
|
||||
"every participant must converge and release the decision"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn delete_recovery_dry_run_matches_adoption_of_already_absent_non_null_version() {
|
||||
let bucket = "recover-preview-non-null";
|
||||
@@ -1935,6 +2015,140 @@ mod tests {
|
||||
.expect("no retained version decision may keep the empty bucket nonempty");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn delete_recovery_partial_batch_receipt_handles_missing_versions() {
|
||||
use crate::disk::RUSTFS_META_TMP_BUCKET;
|
||||
use crate::set_disk::ops::object::DeleteCleanupBarrier;
|
||||
use crate::storage_api_contracts::object::ObjectToDelete;
|
||||
|
||||
for include_missing in [false, true] {
|
||||
let bucket = "recover-prepared-batch-missing-versions";
|
||||
let (dirs, store, disks) = fixture(bucket).await;
|
||||
metadata_sys::update_in(
|
||||
&store.ctx,
|
||||
bucket,
|
||||
crate::bucket::metadata::BUCKET_VERSIONING_CONFIG,
|
||||
b"<VersioningConfiguration xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"><Status>Enabled</Status></VersioningConfiguration>".to_vec(),
|
||||
)
|
||||
.await
|
||||
.expect("enable supported versioned bucket");
|
||||
let first = put(&store, bucket, &vec![81; 1024 * 1024], true)
|
||||
.await
|
||||
.expect("first version");
|
||||
let second = put(&store, bucket, &vec![82; 4096], true).await.expect("retained version");
|
||||
let missing = Uuid::new_v4();
|
||||
let barrier = DeleteCleanupBarrier::install(bucket);
|
||||
let deleting = store.clone();
|
||||
let task = tokio::spawn(async move {
|
||||
let mut versions = vec![first];
|
||||
if include_missing {
|
||||
versions.push(missing);
|
||||
}
|
||||
deleting
|
||||
.delete_objects(
|
||||
bucket,
|
||||
versions
|
||||
.into_iter()
|
||||
.map(|version| ObjectToDelete {
|
||||
object_name: "object".into(),
|
||||
version_id: Some(version),
|
||||
..Default::default()
|
||||
})
|
||||
.collect(),
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
});
|
||||
barrier.wait_until_paused().await;
|
||||
let transactions = disks[0]
|
||||
.discover_delete_transactions(bucket, Some("object"), None, 100)
|
||||
.await
|
||||
.expect("discover batch preparation")
|
||||
.transactions;
|
||||
assert_eq!(transactions.len(), 1);
|
||||
let transaction = transactions[0].transaction;
|
||||
for disk in &disks {
|
||||
let state = disk
|
||||
.inspect_delete_transaction(bucket, "object", transaction)
|
||||
.await
|
||||
.expect("inspect prepared batch");
|
||||
assert_eq!(state.state, DeleteRecoveryState::NewPrepared);
|
||||
let journal = state.journal.expect("prepared journal");
|
||||
assert!(journal.original_digest.is_some());
|
||||
assert!(journal.original_versions.iter().all(|v| v.version_id != Some(missing)));
|
||||
}
|
||||
// Fail durable receipt staging on two participants after metadata quorum.
|
||||
for dir in dirs.iter().skip(2) {
|
||||
tokio::fs::rename(dir.path().join(RUSTFS_META_TMP_BUCKET), dir.path().join("saved-receipt-staging"))
|
||||
.await
|
||||
.expect("save receipt staging directory");
|
||||
tokio::fs::write(dir.path().join(RUSTFS_META_TMP_BUCKET), b"receipt staging I/O failure")
|
||||
.await
|
||||
.expect("inject receipt staging failure");
|
||||
}
|
||||
barrier.release();
|
||||
let (_, errors) = task.await.expect("batch deletion must join");
|
||||
drop(barrier);
|
||||
assert_eq!(errors.len(), if include_missing { 2 } else { 1 });
|
||||
assert!(errors.iter().all(Option::is_some), "partial receipts are indeterminate: {errors:?}");
|
||||
let mut receipts = 0;
|
||||
let mut prepared = 0;
|
||||
for disk in &disks {
|
||||
match disk
|
||||
.inspect_delete_transaction(bucket, "object", transaction)
|
||||
.await
|
||||
.expect("partial decision")
|
||||
.state
|
||||
{
|
||||
DeleteRecoveryState::QuorumCommitted => receipts += 1,
|
||||
DeleteRecoveryState::NewPrepared => prepared += 1,
|
||||
state => panic!("unexpected post-publication state: {state:?}"),
|
||||
}
|
||||
}
|
||||
assert_eq!((receipts, prepared), (2, 2));
|
||||
for dir in dirs.iter().skip(2) {
|
||||
tokio::fs::remove_file(dir.path().join(RUSTFS_META_TMP_BUCKET))
|
||||
.await
|
||||
.expect("remove injected staging failure");
|
||||
tokio::fs::rename(dir.path().join("saved-receipt-staging"), dir.path().join(RUSTFS_META_TMP_BUCKET))
|
||||
.await
|
||||
.expect("restore receipt staging");
|
||||
}
|
||||
store
|
||||
.delete_object(
|
||||
bucket,
|
||||
"object",
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
version_id: Some(second.to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("later version deletion succeeds");
|
||||
store
|
||||
.recover_delete_transactions(bucket, &HealOpts::default())
|
||||
.await
|
||||
.expect("certified batch must recover after later version deletion");
|
||||
for disk in &disks {
|
||||
assert!(
|
||||
disk.discover_delete_transactions(bucket, Some("object"), None, 100)
|
||||
.await
|
||||
.expect("settled transaction inventory")
|
||||
.transactions
|
||||
.is_empty()
|
||||
);
|
||||
}
|
||||
store
|
||||
.delete_bucket(bucket, &DeleteBucketOptions::default())
|
||||
.await
|
||||
.expect("settled receipts must not block removal of the empty bucket");
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn delete_recovery_partial_receipt_converges_after_later_version_absence() {
|
||||
let bucket = "recover-partial-chained-versions";
|
||||
|
||||
Reference in New Issue
Block a user