mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-06 20:19:14 +00:00
Merge branch 'main' into chore/tier-lint-stage-b
This commit is contained in:
@@ -67,6 +67,7 @@ const ERR_LIFECYCLE_RULE_MUST_HAVE_ACTION: &str = "Rule must have at least one o
|
||||
const ERR_LIFECYCLE_PREFIX_FILTER_CONFLICT: &str = "Legacy Prefix and Filter cannot both be present in a lifecycle rule. Use Filter.Prefix instead of the top-level Prefix element.";
|
||||
const ERR_LIFECYCLE_INVALID_NEWER_NONCURRENT_VERSIONS: &str = "'NewerNoncurrentVersions' must be a non-negative integer";
|
||||
const ERR_LIFECYCLE_FILTER_AND_TOO_FEW_PREDICATES: &str = "Filter And must contain at least two predicates";
|
||||
const ERR_LIFECYCLE_FILTER_TOO_MANY_PREDICATES: &str = "Filter has too many predicates";
|
||||
const ERR_LIFECYCLE_FILTER_DUPLICATE_TAG_KEY: &str = "Filter must not repeat a tag key";
|
||||
const ERR_LIFECYCLE_FILTER_INVALID_TAG: &str = "Tag key must be 1-128 characters and tag value must be at most 256 characters";
|
||||
const ERR_LIFECYCLE_FILTER_NEGATIVE_SIZE: &str = "ObjectSizeGreaterThan and ObjectSizeLessThan must not be negative";
|
||||
@@ -286,11 +287,14 @@ impl RuleValidate for LifecycleRule {
|
||||
/// `Filter` as "applies to every object in the bucket", and rejecting it would
|
||||
/// break the most common way to write an unconditional rule.
|
||||
fn validate_lifecycle_filter(filter: &LifecycleRuleFilter) -> Result<(), std::io::Error> {
|
||||
// AWS S3 allows multiple top-level predicates (Prefix, Tag, ObjectSize*)
|
||||
// as siblings in Filter without an explicit And wrapper — botocore sends
|
||||
// this layout when a rule combines Prefix with Tag. The evaluation code
|
||||
// already treats sibling predicates as implicit AND, so we validate each
|
||||
// predicate individually rather than enforcing a single-predicate limit.
|
||||
let top_level_predicates = usize::from(filter.prefix.is_some())
|
||||
+ usize::from(filter.tag.is_some())
|
||||
+ usize::from(filter.object_size_greater_than.is_some())
|
||||
+ usize::from(filter.object_size_less_than.is_some())
|
||||
+ usize::from(filter.and.is_some());
|
||||
if top_level_predicates > 1 {
|
||||
return Err(malformed_xml_error(ERR_LIFECYCLE_FILTER_TOO_MANY_PREDICATES));
|
||||
}
|
||||
|
||||
if let Some(tag) = filter.tag.as_ref() {
|
||||
validate_lifecycle_tag(tag)?;
|
||||
@@ -4713,7 +4717,7 @@ mod tests {
|
||||
tag: Some(tag("env", "prod")),
|
||||
..Default::default()
|
||||
},
|
||||
expected: None,
|
||||
expected: Some((ERR_LIFECYCLE_FILTER_TOO_MANY_PREDICATES, LIFECYCLE_MALFORMED_XML_ERROR_KIND)),
|
||||
},
|
||||
Case {
|
||||
name: "prefix alongside And",
|
||||
@@ -4726,7 +4730,7 @@ mod tests {
|
||||
}),
|
||||
..Default::default()
|
||||
},
|
||||
expected: None,
|
||||
expected: Some((ERR_LIFECYCLE_FILTER_TOO_MANY_PREDICATES, LIFECYCLE_MALFORMED_XML_ERROR_KIND)),
|
||||
},
|
||||
Case {
|
||||
name: "And with a single member",
|
||||
|
||||
@@ -1910,10 +1910,10 @@ where
|
||||
.as_ref()
|
||||
.map(|(notification_system, grants)| (Arc::clone(notification_system), grants.clone()));
|
||||
let remote_lease_release_safe = Arc::new(AtomicBool::new(true));
|
||||
let mut usage_persist_outcome = match publication_defer_reason {
|
||||
let mut usage_publication_result = match publication_defer_reason {
|
||||
Some(reason) => {
|
||||
drop(receiver);
|
||||
DataUsagePersistOutcome::Deferred(reason)
|
||||
DataUsagePublicationResult::from(DataUsagePersistOutcome::Deferred(reason))
|
||||
}
|
||||
None => {
|
||||
// ScannerIO emits its complete or observational update only after
|
||||
@@ -1928,6 +1928,11 @@ where
|
||||
.as_ref()
|
||||
.map(|(_, grants)| grants.iter().map(|grant| grant.lease.token).collect())
|
||||
.unwrap_or_default();
|
||||
let ack_expectation = scan_result
|
||||
.as_ref()
|
||||
.ok()
|
||||
.filter(|result| result.has_dirty_usage_to_acknowledge())
|
||||
.and_then(ScannerCycleResult::publication_expectation);
|
||||
let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move {
|
||||
store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence(
|
||||
ctx_clone,
|
||||
@@ -1940,6 +1945,7 @@ where
|
||||
remote_lease_deadline,
|
||||
remote_lease_fence,
|
||||
)
|
||||
.with_ack_expectation(ack_expectation)
|
||||
.with_remote_lease_tokens(remote_lease_tokens)
|
||||
.with_lease_release_flag(remote_lease_release_safe_for_task),
|
||||
move || {
|
||||
@@ -1973,7 +1979,7 @@ where
|
||||
error = %err,
|
||||
"Scanner data usage persistence task failed"
|
||||
);
|
||||
DataUsagePersistOutcome::Failed
|
||||
DataUsagePublicationResult::from(DataUsagePersistOutcome::Failed)
|
||||
}
|
||||
DataUsagePersistTaskResult::Cancelled => {
|
||||
debug!(
|
||||
@@ -1985,7 +1991,7 @@ where
|
||||
state = "usage_persist_task_cancelled",
|
||||
"Scanner data usage persistence task cancelled"
|
||||
);
|
||||
DataUsagePersistOutcome::Failed
|
||||
DataUsagePublicationResult::from(DataUsagePersistOutcome::Failed)
|
||||
}
|
||||
DataUsagePersistTaskResult::TimedOut => {
|
||||
error!(
|
||||
@@ -1998,11 +2004,12 @@ where
|
||||
state = "usage_persist_task_timed_out",
|
||||
"Scanner data usage persistence task timed out"
|
||||
);
|
||||
DataUsagePersistOutcome::Failed
|
||||
DataUsagePublicationResult::from(DataUsagePersistOutcome::Failed)
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
let mut usage_persist_outcome = usage_publication_result.outcome();
|
||||
let lease_expired = remote_publication_leases
|
||||
.as_ref()
|
||||
.is_some_and(|(_, grants)| grants.iter().any(|grant| !grant.lease.is_valid()));
|
||||
@@ -2202,8 +2209,9 @@ where
|
||||
};
|
||||
}
|
||||
|
||||
usage_publication_result.restrict_outcome(usage_persist_outcome);
|
||||
let (completion_outcome, scanner_pending_maintenance_work, remote_dirty_usage_acknowledgements) =
|
||||
finalize_scanner_cycle_result(scan_cycle_result, usage_persist_outcome);
|
||||
finalize_scanner_cycle_result(scan_cycle_result, usage_publication_result);
|
||||
let remote_dirty_usage_pending = if remote_dirty_usage_acknowledgements.is_empty() {
|
||||
false
|
||||
} else if let Some(notification_system) = storeapi.scanner_notification_system() {
|
||||
@@ -3437,21 +3445,35 @@ fn scanner_cycle_completion_outcome(
|
||||
|
||||
fn finalize_scanner_cycle_result(
|
||||
scan_cycle_result: crate::scanner_io::ScannerCycleResult,
|
||||
usage_persist_outcome: DataUsagePersistOutcome,
|
||||
publication: DataUsagePublicationResult,
|
||||
) -> (ScannerCycleOutcome, bool, Vec<ScannerDirtyUsageAcknowledgement>) {
|
||||
let (usage_persist_outcome, proof) = publication.into_parts();
|
||||
let completion_outcome = scanner_cycle_completion_outcome_for_result(&scan_cycle_result, usage_persist_outcome);
|
||||
let pending_maintenance_work = scan_cycle_result.has_pending_maintenance_work();
|
||||
let durable_complete_snapshot = scan_cycle_result.status == ScannerCycleStatus::Complete
|
||||
&& matches!(
|
||||
usage_persist_outcome,
|
||||
DataUsagePersistOutcome::Saved | DataUsagePersistOutcome::AlreadyDurable
|
||||
);
|
||||
)
|
||||
&& scan_cycle_result.publication_expectation().as_ref().is_some_and(|expected| {
|
||||
proof
|
||||
.as_ref()
|
||||
.is_some_and(|proof| proof.verified_version_for(expected).is_some())
|
||||
});
|
||||
let pending_maintenance_work = scan_cycle_result.has_pending_maintenance_work()
|
||||
|| (scan_cycle_result.has_dirty_usage_to_acknowledge() && !durable_complete_snapshot);
|
||||
let remote_dirty_usage_acknowledgements = if durable_complete_snapshot {
|
||||
scan_cycle_result.acknowledge_durable_usage()
|
||||
match proof {
|
||||
Some(proof) => scan_cycle_result.acknowledge_durable_usage(&proof),
|
||||
None => Vec::new(),
|
||||
}
|
||||
} else {
|
||||
Vec::new()
|
||||
};
|
||||
(completion_outcome, pending_maintenance_work, remote_dirty_usage_acknowledgements)
|
||||
(
|
||||
completion_outcome,
|
||||
pending_maintenance_work || crate::scanner_io::dirty_usage_buckets_pending(),
|
||||
remote_dirty_usage_acknowledgements,
|
||||
)
|
||||
}
|
||||
|
||||
fn scanner_cycle_completion_outcome_for_result(
|
||||
@@ -3551,6 +3573,7 @@ use activity::*;
|
||||
use backlog::*;
|
||||
use cycle_state::*;
|
||||
use leadership::*;
|
||||
pub(crate) use usage_store::RootPublicationProof;
|
||||
use usage_store::*;
|
||||
|
||||
pub use activity::scanner_topology_digest;
|
||||
|
||||
@@ -36,7 +36,10 @@ use tokio::time::{Duration, advance};
|
||||
|
||||
const TEST_DEFAULT_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60;
|
||||
|
||||
mod quota_reset_preservation;
|
||||
|
||||
mod recovery_control;
|
||||
mod scoped_ack_publication;
|
||||
|
||||
async fn setup_scanner_cycle_store() -> (tempfile::TempDir, Arc<ECStore>) {
|
||||
setup_scanner_cycle_store_with_usage_baseline(true).await
|
||||
@@ -5288,6 +5291,103 @@ async fn scanner_usage_state_reset_resumes_every_cleanup_boundary_without_rewrit
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scanner_usage_state_reset_resumes_real_store_cleanup_boundaries_after_reopen() {
|
||||
let primary_path = DATA_USAGE_OBJ_NAME_PATH.as_str();
|
||||
let cleanup_paths = [
|
||||
format!("{primary_path}.bkp"),
|
||||
LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
|
||||
format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
|
||||
DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str().to_string(),
|
||||
];
|
||||
|
||||
for completed in 0..=cleanup_paths.len() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
let cycle = CurrentCycle {
|
||||
current: 12,
|
||||
next: 42,
|
||||
cycle_completed: vec![Utc::now()],
|
||||
started: Utc::now(),
|
||||
};
|
||||
save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_BLOOM_NAME_PATH.as_str(),
|
||||
encode_scanner_cycle_state(&cycle, 3).expect("cycle state should encode"),
|
||||
)
|
||||
.await
|
||||
.expect("cycle state should persist");
|
||||
let marker = scanner_usage_bootstrap_marker(std::time::SystemTime::UNIX_EPOCH, Some(3));
|
||||
save_config(
|
||||
store.clone(),
|
||||
primary_path,
|
||||
serde_json::to_vec(&marker).expect("usage reset marker should encode"),
|
||||
)
|
||||
.await
|
||||
.expect("usage reset marker should persist");
|
||||
|
||||
for path in cleanup_paths.iter().skip(completed) {
|
||||
let mut usage = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
|
||||
usage.scanner_epoch = Some(1);
|
||||
usage.scanner_cycle = Some(12);
|
||||
save_config(store.clone(), path, serde_json::to_vec(&usage).expect("cleanup slot should encode"))
|
||||
.await
|
||||
.expect("cleanup slot should persist");
|
||||
}
|
||||
for path in ["buckets/quota-reservations/ledger", "buckets/example/incarnation"] {
|
||||
save_config(store.clone(), path, b"retain".to_vec())
|
||||
.await
|
||||
.expect("unrelated state should persist before reopen");
|
||||
}
|
||||
|
||||
let restarted = restart_scanner_cycle_store_from(&store).await;
|
||||
let intent_before = read_config_with_revision(restarted.clone(), primary_path)
|
||||
.await
|
||||
.expect("reopened reset intent should be readable");
|
||||
|
||||
let result = reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), restarted.clone())
|
||||
.await
|
||||
.expect("reopened usage reset should complete");
|
||||
|
||||
assert_eq!(result.leader_epoch, 3, "boundary {completed}");
|
||||
assert_eq!(result.next_cycle, 42, "boundary {completed}");
|
||||
assert_eq!(result.reset_paths.len(), cleanup_paths.len() + 1 - completed, "boundary {completed}");
|
||||
assert_eq!(
|
||||
read_config_with_revision(restarted.clone(), primary_path)
|
||||
.await
|
||||
.expect("completed reset intent should remain readable"),
|
||||
intent_before,
|
||||
"boundary {completed}: resumed cleanup must not rewrite the reset intent"
|
||||
);
|
||||
|
||||
let (floor, state) = persisted_usage_floor_for_startup(restarted.clone(), false)
|
||||
.await
|
||||
.expect("completed reset marker should remain resumable");
|
||||
assert_eq!(floor.leader_epoch, 3, "boundary {completed}");
|
||||
assert_eq!(state, PersistedUsageFloorStartup::BootstrapPending, "boundary {completed}");
|
||||
assert!(
|
||||
persisted_usage_floor(restarted.clone()).await.is_err(),
|
||||
"boundary {completed}: bootstrap marker must not become an authoritative floor"
|
||||
);
|
||||
|
||||
for path in &cleanup_paths {
|
||||
assert!(
|
||||
matches!(read_config(restarted.clone(), path).await, Err(EcstoreError::ConfigNotFound)),
|
||||
"boundary {completed}: reset should remove stale usage slot {path}"
|
||||
);
|
||||
}
|
||||
for path in ["buckets/quota-reservations/ledger", "buckets/example/incarnation"] {
|
||||
assert_eq!(
|
||||
read_config(restarted.clone(), path)
|
||||
.await
|
||||
.expect("unrelated state should survive reopened reset"),
|
||||
b"retain",
|
||||
"boundary {completed}: reset must preserve non-scanner-state config"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scanner_usage_state_reset_stops_usage_fence_after_owner_loss() {
|
||||
let store = Arc::new(MemoryConfigStore::default());
|
||||
@@ -6184,7 +6284,7 @@ async fn coordinator_classifies_an_expired_publication_lease() {
|
||||
.await;
|
||||
|
||||
assert_eq!(
|
||||
outcome,
|
||||
outcome.outcome(),
|
||||
DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded)
|
||||
);
|
||||
assert!(store.put_counts.lock().await.is_empty(), "expired lease must prevent a PUT");
|
||||
@@ -7325,7 +7425,7 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() {
|
||||
|
||||
#[test]
|
||||
#[serial]
|
||||
fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() {
|
||||
fn finalizing_a_saved_enum_without_proof_keeps_dirty_pending() {
|
||||
crate::scanner_io::clear_dirty_usage_bucket("photos");
|
||||
crate::scanner_io::record_dirty_usage_bucket("photos");
|
||||
let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
|
||||
@@ -7337,17 +7437,19 @@ fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() {
|
||||
};
|
||||
let unsaved = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot.clone()))
|
||||
.with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement.clone()]);
|
||||
let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(unsaved, DataUsagePersistOutcome::NoUpdate);
|
||||
let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(unsaved, DataUsagePersistOutcome::NoUpdate.into());
|
||||
assert_eq!(outcome, ScannerCycleOutcome::Failed);
|
||||
assert!(acknowledgements.is_empty());
|
||||
assert!(crate::scanner_io::dirty_usage_buckets_pending());
|
||||
|
||||
let saved = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot))
|
||||
.with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement.clone()]);
|
||||
let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(saved, DataUsagePersistOutcome::Saved);
|
||||
.with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement]);
|
||||
let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(saved, DataUsagePersistOutcome::Saved.into());
|
||||
assert_eq!(outcome, ScannerCycleOutcome::Completed);
|
||||
assert_eq!(acknowledgements, vec![remote_acknowledgement]);
|
||||
assert!(!crate::scanner_io::dirty_usage_buckets_pending());
|
||||
assert!(acknowledgements.is_empty());
|
||||
assert!(pending);
|
||||
assert!(crate::scanner_io::dirty_usage_buckets_pending());
|
||||
crate::scanner_io::clear_dirty_usage_bucket("photos");
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -7359,7 +7461,7 @@ fn finalizing_a_deferred_usage_save_keeps_dirty_work_pending() {
|
||||
let deferred = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot));
|
||||
|
||||
let (outcome, _, acknowledgements) =
|
||||
finalize_scanner_cycle_result(deferred, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
|
||||
finalize_scanner_cycle_result(deferred, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement).into());
|
||||
|
||||
assert_eq!(outcome, ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
|
||||
assert!(acknowledgements.is_empty());
|
||||
@@ -7379,7 +7481,7 @@ fn finalizing_post_scan_observation_advances_partially_without_dirty_ack() {
|
||||
)
|
||||
.with_observational_snapshot_published(true);
|
||||
|
||||
let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(observed, DataUsagePersistOutcome::Saved);
|
||||
let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(observed, DataUsagePersistOutcome::Saved.into());
|
||||
|
||||
assert_eq!(outcome, ScannerCycleOutcome::Partial);
|
||||
assert!(acknowledgements.is_empty());
|
||||
@@ -7415,17 +7517,20 @@ async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
|
||||
|
||||
#[test]
|
||||
#[serial]
|
||||
fn finalizing_an_already_durable_cycle_acknowledges_its_exact_dirty_snapshot() {
|
||||
fn finalizing_an_already_durable_enum_without_proof_keeps_dirty_pending() {
|
||||
crate::scanner_io::clear_dirty_usage_bucket("photos");
|
||||
crate::scanner_io::record_dirty_usage_bucket("photos");
|
||||
let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
|
||||
|
||||
let durable = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot));
|
||||
let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::AlreadyDurable);
|
||||
let (outcome, pending, acknowledgements) =
|
||||
finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::AlreadyDurable.into());
|
||||
|
||||
assert_eq!(outcome, ScannerCycleOutcome::Completed);
|
||||
assert!(acknowledgements.is_empty());
|
||||
assert!(!crate::scanner_io::dirty_usage_buckets_pending());
|
||||
assert!(pending);
|
||||
assert!(crate::scanner_io::dirty_usage_buckets_pending());
|
||||
crate::scanner_io::clear_dirty_usage_bucket("photos");
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -7436,7 +7541,8 @@ fn finalizing_a_prior_same_cycle_snapshot_keeps_new_dirty_work_pending() {
|
||||
let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
|
||||
|
||||
let durable = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot));
|
||||
let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::PriorCycleDurable);
|
||||
let (outcome, _, acknowledgements) =
|
||||
finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::PriorCycleDurable.into());
|
||||
|
||||
assert_eq!(outcome, ScannerCycleOutcome::Completed);
|
||||
assert!(acknowledgements.is_empty());
|
||||
@@ -7452,7 +7558,7 @@ fn finalizing_a_durable_superseded_snapshot_keeps_dirty_work_pending() {
|
||||
let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
|
||||
|
||||
let superseded = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Superseded, Some(dirty_snapshot));
|
||||
let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(superseded, DataUsagePersistOutcome::Saved);
|
||||
let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(superseded, DataUsagePersistOutcome::Saved.into());
|
||||
|
||||
assert_eq!(outcome, ScannerCycleOutcome::Superseded);
|
||||
assert!(acknowledgements.is_empty());
|
||||
@@ -8881,7 +8987,7 @@ fn post_lease_activity_proof_rejects_a_put_tail_that_finished_before_lease_acqui
|
||||
]);
|
||||
let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(
|
||||
result,
|
||||
DataUsagePersistOutcome::Deferred(reason.expect("changed namespace should defer publication")),
|
||||
DataUsagePersistOutcome::Deferred(reason.expect("changed namespace should defer publication")).into(),
|
||||
);
|
||||
assert_eq!(
|
||||
outcome,
|
||||
|
||||
@@ -0,0 +1,148 @@
|
||||
// Copyright 2026 RustFS Team
|
||||
// Licensed under the Apache License, Version 2.0.
|
||||
|
||||
use super::*;
|
||||
use crate::storage_api::owner::ObjectOperations as _;
|
||||
|
||||
const BUCKET: &str = "quota-reset-preservation";
|
||||
const OPERATION: &str = "00000000-0000-0000-0000-000000000002";
|
||||
|
||||
async fn reservation_fixture() -> (tempfile::TempDir, Arc<ECStore>, Uuid, String, Vec<u8>) {
|
||||
let (directory, store) = setup_scanner_cycle_store().await;
|
||||
store
|
||||
.make_bucket(BUCKET, &crate::storage_api::scan::MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create the reservation fixture bucket through its owner");
|
||||
let incarnation = store
|
||||
.bucket_incarnation_id_from_disk(BUCKET)
|
||||
.await
|
||||
.expect("durable bucket incarnation");
|
||||
assert!(!incarnation.is_nil());
|
||||
let path = format!("config/quota-ledger/{BUCKET}.json");
|
||||
let bytes = serde_json::to_vec(&serde_json::json!({
|
||||
"version": 1,
|
||||
"bucket_incarnation": incarnation,
|
||||
"quota_revision_unix_nanos": 1,
|
||||
"accounted_usage": 100,
|
||||
"reservations": {
|
||||
OPERATION: {
|
||||
"object": "pending-object",
|
||||
"old_size": 0,
|
||||
"new_size": 64,
|
||||
"created_at": 1,
|
||||
"pool_index": 0,
|
||||
"set_index": 0,
|
||||
"commit_started": true
|
||||
}
|
||||
}
|
||||
}))
|
||||
.expect("encode the committed reservation fixture");
|
||||
save_config(store.clone(), &path, bytes.clone())
|
||||
.await
|
||||
.expect("persist reservation bytes through the real storage owner");
|
||||
(directory, store, incarnation, path, bytes)
|
||||
}
|
||||
|
||||
async fn assert_reservation_retained(store: &Arc<ECStore>, path: &str, expected: &[u8], incarnation: Uuid) {
|
||||
let bytes = read_config(store.clone(), path)
|
||||
.await
|
||||
.expect("read the actual reservation ledger");
|
||||
assert_eq!(bytes, expected, "scanner reset must not rewrite the reservation ledger");
|
||||
let ledger: serde_json::Value = serde_json::from_slice(&bytes).expect("persisted ledger JSON");
|
||||
assert_eq!(ledger["version"], 1);
|
||||
assert_eq!(ledger["bucket_incarnation"], incarnation.to_string());
|
||||
assert_eq!(ledger["accounted_usage"], 100);
|
||||
let reservations = ledger["reservations"].as_object().expect("reservation map");
|
||||
assert_eq!(reservations.len(), 1);
|
||||
let pending = &reservations[OPERATION];
|
||||
assert_eq!(pending["old_size"], 0);
|
||||
assert_eq!(pending["new_size"], 64);
|
||||
assert_eq!(pending["commit_started"], true);
|
||||
assert_eq!(
|
||||
store
|
||||
.bucket_incarnation_id_from_disk(BUCKET)
|
||||
.await
|
||||
.expect("owner incarnation after restart"),
|
||||
incarnation
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn quota_reset_preservation_survives_storage_owner_reconstruction() {
|
||||
let (_directory, store, incarnation, path, bytes) = reservation_fixture().await;
|
||||
let reset = reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.expect("reset scanner usage through the fenced production entry");
|
||||
assert_eq!(reset.usage_state, "bootstrap-pending");
|
||||
let restarted = restart_scanner_cycle_store_from(&store).await;
|
||||
assert!(
|
||||
!Arc::ptr_eq(&store, &restarted),
|
||||
"the assertion must read through a newly constructed ECStore"
|
||||
);
|
||||
assert_reservation_retained(&restarted, &path, &bytes, incarnation).await;
|
||||
let usage = read_config(restarted.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("read reset usage through the reconstructed owner");
|
||||
let usage: DataUsageInfo = serde_json::from_slice(&usage).expect("bootstrap usage JSON");
|
||||
assert!(data_usage_info_is_bootstrap_pending(&usage));
|
||||
assert!(!data_usage_info_has_persisted_baseline_identity(&usage));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn quota_reset_preservation_unknown_protocol_rejects_put_after_restart() {
|
||||
for quota_shape in ["zero", "null", "missing"] {
|
||||
let (_directory, store, incarnation, path, bytes) = reservation_fixture().await;
|
||||
let mut quota = serde_json::json!({
|
||||
"quota_type": "Hard",
|
||||
"reservation_protocol": 2,
|
||||
"reservation_quota": 1024
|
||||
});
|
||||
match quota_shape {
|
||||
"zero" => quota["quota"] = serde_json::json!(0),
|
||||
"null" => quota["quota"] = serde_json::Value::Null,
|
||||
"missing" => {}
|
||||
_ => unreachable!("fixed quota shapes"),
|
||||
}
|
||||
let unknown_quota = serde_json::to_vec("a).expect("unknown but syntactically valid quota protocol");
|
||||
store
|
||||
.update_bucket_metadata_config(BUCKET, rustfs_config::QUOTA_CONFIG_FILE, unknown_quota)
|
||||
.await
|
||||
.expect("persist a future protocol using the real metadata owner");
|
||||
assert_eq!(
|
||||
store
|
||||
.bucket_incarnation_id_from_disk(BUCKET)
|
||||
.await
|
||||
.expect("same metadata owner incarnation"),
|
||||
incarnation
|
||||
);
|
||||
reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.expect("scanner reset must not change quota metadata");
|
||||
let restarted = restart_scanner_cycle_store_from(&store).await;
|
||||
assert!(!Arc::ptr_eq(&store, &restarted));
|
||||
assert_reservation_retained(&restarted, &path, &bytes, incarnation).await;
|
||||
let mut reader = PutObjReader::from_vec(b"must-not-commit".to_vec());
|
||||
let result = restarted.pools[0].disk_set[0]
|
||||
.put_object(BUCKET, "rejected-object", &mut reader, &ObjectOptions::default())
|
||||
.await;
|
||||
let error = match result {
|
||||
Err(error) => error,
|
||||
Ok(_) => panic!("unknown reservation protocol with quota={quota_shape} must not admit a PUT"),
|
||||
};
|
||||
assert!(
|
||||
matches!(error, EcstoreError::PartMissingOrCorrupt),
|
||||
"unexpected protocol rejection: {error}"
|
||||
);
|
||||
let missing = restarted.pools[0].disk_set[0]
|
||||
.get_object_info(BUCKET, "rejected-object", &ObjectOptions::default())
|
||||
.await
|
||||
.expect_err("the rejected PUT must not create an object");
|
||||
assert!(
|
||||
matches!(missing, EcstoreError::FileNotFound | EcstoreError::ObjectNotFound(_, _)),
|
||||
"object absence must not be confused with another storage failure: {missing}"
|
||||
);
|
||||
assert_reservation_retained(&restarted, &path, &bytes, incarnation).await;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,490 @@
|
||||
// Copyright 2026 RustFS Team
|
||||
// Licensed under the Apache License, Version 2.0.
|
||||
|
||||
use super::super::usage_store::DataUsagePublicationResult;
|
||||
use super::*;
|
||||
use crate::scanner_io::ScannerBucketScanScope;
|
||||
use rustfs_utils::path::path_join_buf;
|
||||
use sha2::Digest;
|
||||
use std::time::SystemTime;
|
||||
|
||||
const PROOF_BUCKET: &str = "publication-proof-bucket";
|
||||
const PROOF_EPOCH: u64 = 7;
|
||||
const PROOF_CYCLE: u64 = 11;
|
||||
|
||||
async fn settle_namespace_commits(store: &ECStore) {
|
||||
tokio::time::timeout(Duration::from_secs(30), async {
|
||||
while store.scanner_data_usage_publication_blocked().await {
|
||||
tokio::time::sleep(Duration::from_millis(1)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("fixture namespace commits must settle before collecting complete coverage");
|
||||
}
|
||||
|
||||
async fn complete_candidate(store: &Arc<ECStore>, cycle: u64) -> (crate::scanner_io::ScannerCycleResult, DataUsageInfo) {
|
||||
settle_namespace_commits(store).await;
|
||||
let ctx = CancellationToken::new();
|
||||
let budget = ScannerCycleBudget::new_with_progress_tracking(
|
||||
&ctx,
|
||||
ScannerCycleBudgetConfig {
|
||||
max_objects: Some(8),
|
||||
..Default::default()
|
||||
},
|
||||
);
|
||||
let (updates, mut receiver) = mpsc::channel(1);
|
||||
let result = crate::scanner_io::nsscanner_with_storage_status_scoped(
|
||||
store.as_ref(),
|
||||
crate::scanner_io::ScannerCycleRequest {
|
||||
ctx,
|
||||
budget,
|
||||
updates,
|
||||
want_cycle: cycle,
|
||||
leader_epoch: PROOF_EPOCH,
|
||||
scan_mode: HealScanMode::Normal,
|
||||
scan_scope: ScannerBucketScanScope::default(),
|
||||
persisted_usage_baseline: None,
|
||||
observed_usage_candidate: None,
|
||||
requires_full_scan: true,
|
||||
service_cohort: None,
|
||||
resolved_scope_observer: None,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("real scanner must produce the fixture candidate");
|
||||
assert_eq!(result.status, ScannerCycleStatus::Complete);
|
||||
let candidate = receiver.recv().await.expect("complete scanner snapshot");
|
||||
assert!(candidate.usage_snapshot_complete);
|
||||
assert_eq!(candidate.usage_snapshot_converged, Some(true));
|
||||
assert_eq!(candidate.scanner_cycle, Some(cycle));
|
||||
assert_eq!(candidate.scanner_epoch, Some(PROOF_EPOCH));
|
||||
(result, candidate)
|
||||
}
|
||||
|
||||
async fn candidate_store() -> (tempfile::TempDir, Arc<ECStore>) {
|
||||
crate::scanner_io::clear_dirty_usage_buckets_for_tests();
|
||||
let (directory, store) = setup_scanner_cycle_store_with_usage_baseline(false).await;
|
||||
store
|
||||
.make_bucket(PROOF_BUCKET, &crate::storage_api::scan::MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create proof fixture bucket through the owner");
|
||||
let mut reader = PutObjReader::from_vec(b"proof".to_vec());
|
||||
store.pools[0].disk_set[0]
|
||||
.put_object(PROOF_BUCKET, "initial", &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("persist fixture object through the owner");
|
||||
crate::scanner_io::record_dirty_usage_bucket(PROOF_BUCKET);
|
||||
settle_namespace_commits(&store).await;
|
||||
(directory, store)
|
||||
}
|
||||
|
||||
async fn read_root(store: &Arc<ECStore>) -> (Option<Vec<u8>>, DataUsageCacheRevision) {
|
||||
read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("read actual v2 root bytes and revision")
|
||||
}
|
||||
|
||||
async fn publish_candidate(
|
||||
store: &Arc<ECStore>,
|
||||
scan: &crate::scanner_io::ScannerCycleResult,
|
||||
candidate: DataUsageInfo,
|
||||
baseline: Option<DataUsagePersistBaseline>,
|
||||
) -> DataUsagePublicationResult {
|
||||
let expectation = scan.publication_expectation();
|
||||
assert!(expectation.is_some(), "only a real complete scan may supply the expectation");
|
||||
let (sender, receiver) = mpsc::channel(1);
|
||||
sender.send(candidate).await.expect("enqueue the real scan candidate");
|
||||
drop(sender);
|
||||
store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence(
|
||||
CancellationToken::new(),
|
||||
store.clone(),
|
||||
receiver,
|
||||
Some(PROOF_EPOCH),
|
||||
baseline,
|
||||
ScannerPublicationFence::new(scan.publication_epoch(), None, None).with_ack_expectation(expectation),
|
||||
|| async { None },
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scoped_ack_publication_companion_only_does_not_authorize_root_ack() {
|
||||
for companion in [
|
||||
format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()),
|
||||
LEGACY_DATA_USAGE_OBJ_NAME_PATH.to_string(),
|
||||
format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
|
||||
] {
|
||||
let (_directory, store) = candidate_store().await;
|
||||
let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
|
||||
let bytes = serde_json::to_vec(&candidate).expect("actual candidate JSON");
|
||||
save_config(store.clone(), &companion, bytes.clone())
|
||||
.await
|
||||
.expect("persist the companion on real disks");
|
||||
let baseline = read_data_usage_persist_baseline(store.clone())
|
||||
.await
|
||||
.expect("companion fallback baseline");
|
||||
assert_eq!(baseline.data.as_deref(), Some(bytes.as_slice()));
|
||||
assert_eq!(baseline.revision, DataUsageCacheRevision::Missing);
|
||||
assert_eq!(read_root(&store).await.0, None);
|
||||
let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
|
||||
|
||||
let publication = publish_candidate(&store, &scan, candidate, Some(baseline)).await;
|
||||
assert_eq!(publication.outcome(), DataUsagePersistOutcome::AlreadyDurable);
|
||||
let (_, pending, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
|
||||
assert!(pending, "unacknowledged durable companion work must remain pending");
|
||||
assert!(acknowledgements.is_empty());
|
||||
assert_eq!(
|
||||
crate::scanner_io::dirty_usage_buckets_for_tests(),
|
||||
dirty,
|
||||
"a companion is not the v2 root target"
|
||||
);
|
||||
assert_eq!(read_root(&store).await, (None, DataUsageCacheRevision::Missing));
|
||||
assert_eq!(read_config(store.clone(), &companion).await.expect("companion retained"), bytes);
|
||||
}
|
||||
crate::scanner_io::clear_dirty_usage_buckets_for_tests();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scoped_ack_publication_actual_root_readback_accepts_semantic_json_equivalence() {
|
||||
let (_directory, store) = candidate_store().await;
|
||||
let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
|
||||
let canonical = serde_json::to_vec(&candidate).expect("candidate encoding");
|
||||
let mut value = serde_json::to_value(&candidate).expect("candidate value");
|
||||
value
|
||||
.as_object_mut()
|
||||
.expect("usage object")
|
||||
.insert("fixture_unknown_field".into(), serde_json::json!({"retained": true}));
|
||||
let different_bytes = serde_json::to_vec_pretty(&value).expect("noncanonical primary JSON");
|
||||
assert_ne!(different_bytes, canonical);
|
||||
assert_eq!(
|
||||
serde_json::from_slice::<DataUsageInfo>(&different_bytes).expect("semantic primary"),
|
||||
candidate
|
||||
);
|
||||
save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), different_bytes.clone())
|
||||
.await
|
||||
.expect("persist actual primary representation");
|
||||
let before = read_root(&store).await;
|
||||
assert!(matches!(&before.1, DataUsageCacheRevision::Etag(etag) if !etag.is_empty()));
|
||||
let baseline = read_data_usage_persist_baseline(store.clone())
|
||||
.await
|
||||
.expect("real primary revision");
|
||||
assert!(crate::scanner_io::dirty_usage_buckets_pending());
|
||||
|
||||
let publication = publish_candidate(&store, &scan, candidate.clone(), Some(baseline)).await;
|
||||
assert_eq!(publication.outcome(), DataUsagePersistOutcome::AlreadyDurable);
|
||||
let (_, proof) = publication.into_parts();
|
||||
let proof = proof.expect("actual primary readback must produce its own root proof");
|
||||
let expected = scan.publication_expectation().expect("real scan expectation");
|
||||
let (etag, raw_digest) = proof.verified_version_for(&expected).expect("proof must bind this candidate");
|
||||
let DataUsageCacheRevision::Etag(expected_etag) = &before.1 else { panic!("actual root ETag") };
|
||||
assert_eq!(etag, expected_etag);
|
||||
let expected_digest: [u8; 32] = sha2::Sha256::digest(&different_bytes).into();
|
||||
assert_eq!(
|
||||
*raw_digest, expected_digest,
|
||||
"proof must record actual bytes, not reserialized candidate bytes"
|
||||
);
|
||||
// Obtain another proof through the same real readback path rather than
|
||||
// fabricating a publication result from the inspected proof above.
|
||||
let publication = publish_candidate(&store, &scan, candidate, None).await;
|
||||
let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
|
||||
assert_eq!(outcome, ScannerCycleOutcome::Completed);
|
||||
assert!(acknowledgements.is_empty(), "the single-node fixture has no remote targets");
|
||||
assert!(
|
||||
!crate::scanner_io::dirty_usage_buckets_pending(),
|
||||
"actual root bytes plus a real revision authorize this scan"
|
||||
);
|
||||
assert_eq!(read_root(&store).await, before, "readback must not rewrite unknown fields or whitespace");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scoped_ack_publication_successful_root_cas_authorizes_its_scan() {
|
||||
let (_directory, store) = candidate_store().await;
|
||||
let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
|
||||
let baseline = read_data_usage_persist_baseline(store.clone())
|
||||
.await
|
||||
.expect("initial root revision");
|
||||
assert_eq!(baseline.revision, DataUsageCacheRevision::Missing);
|
||||
assert!(crate::scanner_io::dirty_usage_buckets_pending());
|
||||
let publication = publish_candidate(&store, &scan, candidate.clone(), Some(baseline)).await;
|
||||
assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved);
|
||||
let (bytes, revision) = read_root(&store).await;
|
||||
assert!(matches!(revision, DataUsageCacheRevision::Etag(etag) if !etag.is_empty()));
|
||||
assert_eq!(
|
||||
serde_json::from_slice::<DataUsageInfo>(&bytes.expect("actual saved root")).expect("root JSON"),
|
||||
candidate
|
||||
);
|
||||
let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
|
||||
assert_eq!(outcome, ScannerCycleOutcome::Completed);
|
||||
assert!(!pending);
|
||||
assert!(acknowledgements.is_empty(), "the single-node fixture has no remote targets");
|
||||
assert!(
|
||||
!crate::scanner_io::dirty_usage_buckets_pending(),
|
||||
"the real root CAS must authorize its matching scan"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scoped_ack_publication_observed_candidate_reuse_requires_a_new_root_proof() {
|
||||
let (_directory, store) = candidate_store().await;
|
||||
let bootstrap = scanner_usage_bootstrap_marker(SystemTime::now(), Some(PROOF_EPOCH));
|
||||
save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_OBJ_NAME_PATH.as_str(),
|
||||
serde_json::to_vec(&bootstrap).expect("bootstrap root encoding"),
|
||||
)
|
||||
.await
|
||||
.expect("persist authoritative bootstrap root");
|
||||
let (prior_scan, mut observed_candidate) = complete_candidate(&store, PROOF_CYCLE).await;
|
||||
// Seed a complete but unconverged observation from real scanner coverage;
|
||||
// the production writer attaches its authoritative baseline identity.
|
||||
observed_candidate.usage_snapshot_converged = Some(false);
|
||||
let observation = publish_candidate(&store, &prior_scan, observed_candidate, None).await;
|
||||
let (outcome, proof) = observation.into_parts();
|
||||
assert_eq!(outcome, DataUsagePersistOutcome::Saved);
|
||||
assert!(proof.is_none(), "an observational write cannot authorize a root ACK");
|
||||
let (root_before, revision_before) = read_root(&store).await;
|
||||
let observed = read_config(store.clone(), DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("read real persisted observation");
|
||||
let ctx = CancellationToken::new();
|
||||
let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
|
||||
let (updates, mut receiver) = mpsc::channel(1);
|
||||
let (observer, selected) = tokio::sync::oneshot::channel();
|
||||
let scan = crate::scanner_io::nsscanner_with_storage_status_scoped(
|
||||
store.as_ref(),
|
||||
crate::scanner_io::ScannerCycleRequest {
|
||||
ctx,
|
||||
budget,
|
||||
updates,
|
||||
want_cycle: PROOF_CYCLE + 1,
|
||||
leader_epoch: PROOF_EPOCH,
|
||||
scan_mode: HealScanMode::Normal,
|
||||
scan_scope: ScannerBucketScanScope::default(),
|
||||
persisted_usage_baseline: root_before.clone().map(Bytes::from),
|
||||
observed_usage_candidate: Some(Bytes::from(observed)),
|
||||
requires_full_scan: false,
|
||||
service_cohort: None,
|
||||
resolved_scope_observer: Some(observer),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("observation-backed scope must run through the real scanner");
|
||||
let scope = selected.await.expect("production resolver decision");
|
||||
assert_eq!(scope.selected_buckets_for_tests(), Some(&HashSet::from([PROOF_BUCKET.to_string()])));
|
||||
assert_eq!(scan.status, ScannerCycleStatus::Complete);
|
||||
let expectation = scan.publication_expectation().expect("reused coverage must be revalidated");
|
||||
assert!(
|
||||
!expectation.same_candidate(&prior_scan.publication_expectation().expect("prior real candidate")),
|
||||
"the observation cannot transfer the previous scan's expectation"
|
||||
);
|
||||
assert_eq!(read_root(&store).await, (root_before, revision_before));
|
||||
assert!(crate::scanner_io::dirty_usage_buckets_pending());
|
||||
let candidate = receiver.recv().await.expect("new validated root candidate");
|
||||
assert_eq!(candidate.scanner_cycle, Some(PROOF_CYCLE + 1));
|
||||
assert_eq!(candidate.usage_snapshot_converged, Some(true));
|
||||
let publication = publish_candidate(&store, &scan, candidate, None).await;
|
||||
assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved);
|
||||
let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
|
||||
assert_eq!(outcome, ScannerCycleOutcome::Completed);
|
||||
assert!(!pending);
|
||||
assert!(acknowledgements.is_empty());
|
||||
assert!(!crate::scanner_io::dirty_usage_buckets_pending());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scoped_ack_publication_stale_root_cas_keeps_dirty_after_bucket_save() {
|
||||
let (_directory, store) = candidate_store().await;
|
||||
let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
|
||||
let mut bucket_cache = DataUsageCache::default();
|
||||
bucket_cache
|
||||
.load(store.pools[0].disk_set[0].clone(), &path_join_buf(&[PROOF_BUCKET, DATA_USAGE_CACHE_NAME]))
|
||||
.await
|
||||
.expect("real bucket checkpoint must be persisted before root publication");
|
||||
assert!(bucket_cache.info.snapshot_complete);
|
||||
assert_eq!(
|
||||
bucket_cache
|
||||
.checked_flatten(PROOF_BUCKET)
|
||||
.expect("persisted bucket root")
|
||||
.objects,
|
||||
1
|
||||
);
|
||||
let stale_baseline = read_data_usage_persist_baseline(store.clone())
|
||||
.await
|
||||
.expect("missing root revision");
|
||||
assert_eq!(stale_baseline.revision, DataUsageCacheRevision::Missing);
|
||||
let mut competing = candidate.clone();
|
||||
competing.scanner_epoch = Some(PROOF_EPOCH + 1);
|
||||
competing.scanner_cycle = Some(PROOF_CYCLE + 1);
|
||||
for state in &mut competing.usage_snapshot_set_states {
|
||||
state.scanner_epoch = Some(PROOF_EPOCH + 1);
|
||||
state.scanner_cycle = Some(PROOF_CYCLE + 1);
|
||||
}
|
||||
let competing_bytes = serde_json::to_vec(&competing).expect("competing root");
|
||||
save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), competing_bytes.clone())
|
||||
.await
|
||||
.expect("another publisher wins the actual root slot");
|
||||
let before = read_root(&store).await;
|
||||
let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
|
||||
|
||||
let publication = publish_candidate(&store, &scan, candidate, Some(stale_baseline)).await;
|
||||
assert_eq!(
|
||||
publication.outcome(),
|
||||
DataUsagePersistOutcome::Current,
|
||||
"the old missing revision loses CAS and reconciles the newer root"
|
||||
);
|
||||
let (_, _, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
|
||||
assert!(acknowledgements.is_empty());
|
||||
assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty);
|
||||
assert_eq!(
|
||||
read_root(&store).await,
|
||||
before,
|
||||
"bucket durability must not authorize replacing the winning root"
|
||||
);
|
||||
crate::scanner_io::clear_dirty_usage_buckets_for_tests();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scoped_ack_publication_cannot_transfer_proof_between_real_scan_results() {
|
||||
let (_directory, store) = candidate_store().await;
|
||||
let (first_scan, first_candidate) = complete_candidate(&store, PROOF_CYCLE).await;
|
||||
let (second_scan, second_candidate) = complete_candidate(&store, PROOF_CYCLE).await;
|
||||
assert_eq!(first_candidate.scanner_epoch, second_candidate.scanner_epoch);
|
||||
assert_eq!(first_candidate.scanner_cycle, second_candidate.scanner_cycle);
|
||||
assert_eq!(first_candidate.objects_total_count, second_candidate.objects_total_count);
|
||||
let baseline = read_data_usage_persist_baseline(store.clone())
|
||||
.await
|
||||
.expect("initial root revision");
|
||||
let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
|
||||
let publication = publish_candidate(&store, &first_scan, first_candidate, Some(baseline)).await;
|
||||
assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved);
|
||||
assert!(read_root(&store).await.0.is_some(), "the first scan really published its root");
|
||||
|
||||
let (_, pending, acknowledgements) = finalize_scanner_cycle_result(second_scan, publication);
|
||||
assert!(pending, "another scan's publication must not finish this scan's dirty maintenance work");
|
||||
assert!(acknowledgements.is_empty());
|
||||
assert_eq!(
|
||||
crate::scanner_io::dirty_usage_buckets_for_tests(),
|
||||
dirty,
|
||||
"same counters and cycle cannot transfer another scan's proof"
|
||||
);
|
||||
|
||||
crate::scanner_io::clear_dirty_usage_buckets_for_tests();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scoped_ack_publication_stale_baseline_cannot_prove_a_replaced_root() {
|
||||
let (_directory, store) = candidate_store().await;
|
||||
let (first_scan, first_candidate) = complete_candidate(&store, PROOF_CYCLE).await;
|
||||
save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_OBJ_NAME_PATH.as_str(),
|
||||
serde_json::to_vec(&first_candidate).expect("first candidate"),
|
||||
)
|
||||
.await
|
||||
.expect("persist the first candidate on real disks");
|
||||
let stale_baseline = read_data_usage_persist_baseline(store.clone())
|
||||
.await
|
||||
.expect("capture the genuine first root revision");
|
||||
let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
|
||||
let mut reader = PutObjReader::from_vec(b"second".to_vec());
|
||||
store.pools[0].disk_set[0]
|
||||
.put_object(PROOF_BUCKET, "second", &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("commit a real namespace change");
|
||||
assert_eq!(
|
||||
crate::scanner_io::dirty_usage_buckets_for_tests(),
|
||||
dirty,
|
||||
"direct storage writes leave this fixture's scanner hint generation unchanged"
|
||||
);
|
||||
let (_, replacement) = complete_candidate(&store, PROOF_CYCLE).await;
|
||||
assert_eq!(first_candidate.scanner_epoch, replacement.scanner_epoch);
|
||||
assert_eq!(first_candidate.scanner_cycle, replacement.scanner_cycle);
|
||||
assert_eq!((first_candidate.objects_total_count, replacement.objects_total_count), (1, 2));
|
||||
save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_OBJ_NAME_PATH.as_str(),
|
||||
serde_json::to_vec(&replacement).expect("replacement candidate"),
|
||||
)
|
||||
.await
|
||||
.expect("publish the replacement root");
|
||||
let current = read_root(&store).await;
|
||||
assert_ne!(current.1, stale_baseline.revision);
|
||||
|
||||
// The supplied baseline still equals candidate A, but the actual target
|
||||
// now contains B. Compatibility's AlreadyDurable outcome is not proof.
|
||||
let publication = publish_candidate(&store, &first_scan, first_candidate, Some(stale_baseline)).await;
|
||||
assert_eq!(publication.outcome(), DataUsagePersistOutcome::AlreadyDurable);
|
||||
let (_, pending, acknowledgements) = finalize_scanner_cycle_result(first_scan, publication);
|
||||
assert!(pending);
|
||||
assert!(acknowledgements.is_empty());
|
||||
assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty);
|
||||
assert_eq!(read_root(&store).await, current);
|
||||
crate::scanner_io::clear_dirty_usage_buckets_for_tests();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scoped_ack_publication_rejects_builder_mutation_after_real_root_publish() {
|
||||
for mutation in ["remote_ack_target", "publication_epoch", "remote_lease_targets"] {
|
||||
let (_directory, store) = candidate_store().await;
|
||||
let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
|
||||
let baseline = read_data_usage_persist_baseline(store.clone())
|
||||
.await
|
||||
.expect("initial root revision");
|
||||
let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
|
||||
let changed_generation = dirty
|
||||
.get(PROOF_BUCKET)
|
||||
.expect("the real scan has dirty work")
|
||||
.checked_add(1)
|
||||
.expect("bounded fixture generation");
|
||||
let changed_epoch = scan
|
||||
.publication_epoch()
|
||||
.expect("real scan publication epoch")
|
||||
.checked_add(1)
|
||||
.expect("bounded fixture epoch");
|
||||
let publication = publish_candidate(&store, &scan, candidate.clone(), Some(baseline)).await;
|
||||
assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved, "{mutation}");
|
||||
let root_before = read_root(&store).await;
|
||||
assert_eq!(
|
||||
serde_json::from_slice::<DataUsageInfo>(root_before.0.as_deref().expect("actual saved root"))
|
||||
.expect("persisted root JSON"),
|
||||
candidate,
|
||||
"{mutation}: the original candidate really reached root storage"
|
||||
);
|
||||
|
||||
let changed = match mutation {
|
||||
"remote_ack_target" => scan.with_remote_dirty_usage_acknowledgements(vec![ScannerDirtyUsageAcknowledgement {
|
||||
host: "proof-peer:9000".to_string(),
|
||||
instance_id: crate::scanner_activity_epoch().to_string(),
|
||||
generation: changed_generation,
|
||||
}]),
|
||||
"publication_epoch" => scan.with_publication_epoch(Some(changed_epoch)),
|
||||
"remote_lease_targets" => scan.with_remote_publication_lease_targets(vec![(
|
||||
"proof-peer:9000".to_string(),
|
||||
crate::scanner_activity_epoch().to_string(),
|
||||
changed_generation,
|
||||
)]),
|
||||
_ => unreachable!("fixed mutation cases"),
|
||||
};
|
||||
let (_, pending, acknowledgements) = finalize_scanner_cycle_result(changed, publication);
|
||||
assert!(
|
||||
acknowledgements.is_empty(),
|
||||
"{mutation}: the old root proof must not authorize changed ACK work"
|
||||
);
|
||||
assert!(pending, "{mutation}: changed maintenance work must remain pending");
|
||||
assert_eq!(
|
||||
crate::scanner_io::dirty_usage_buckets_for_tests(),
|
||||
dirty,
|
||||
"{mutation}: the changed scan must not clear local dirty work"
|
||||
);
|
||||
assert_eq!(read_root(&store).await, root_before, "{mutation}: the durable original root is retained");
|
||||
}
|
||||
crate::scanner_io::clear_dirty_usage_buckets_for_tests();
|
||||
}
|
||||
@@ -34,6 +34,139 @@ pub(super) enum DataUsagePersistOutcome {
|
||||
Failed,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct RootPublicationProof {
|
||||
candidate: crate::scanner_io::ScannerPublicationExpectation,
|
||||
root_version: (String, [u8; 32]),
|
||||
}
|
||||
|
||||
impl RootPublicationProof {
|
||||
pub(crate) fn verified_version_for(
|
||||
&self,
|
||||
expected: &crate::scanner_io::ScannerPublicationExpectation,
|
||||
) -> Option<&(String, [u8; 32])> {
|
||||
self.candidate.same_candidate(expected).then_some(&self.root_version)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(super) struct DataUsagePublicationResult {
|
||||
outcome: DataUsagePersistOutcome,
|
||||
proof: Option<RootPublicationProof>,
|
||||
}
|
||||
|
||||
impl From<DataUsagePersistOutcome> for DataUsagePublicationResult {
|
||||
fn from(outcome: DataUsagePersistOutcome) -> Self {
|
||||
Self { outcome, proof: None }
|
||||
}
|
||||
}
|
||||
|
||||
impl DataUsagePublicationResult {
|
||||
pub(super) fn outcome(&self) -> DataUsagePersistOutcome {
|
||||
self.outcome
|
||||
}
|
||||
pub(super) fn restrict_outcome(&mut self, outcome: DataUsagePersistOutcome) {
|
||||
if outcome != self.outcome {
|
||||
self.proof = None;
|
||||
}
|
||||
self.outcome = outcome;
|
||||
}
|
||||
pub(super) fn into_parts(self) -> (DataUsagePersistOutcome, Option<RootPublicationProof>) {
|
||||
(self.outcome, self.proof)
|
||||
}
|
||||
}
|
||||
|
||||
fn root_ack_write_is_confirmed<T, E>(
|
||||
result: &std::result::Result<T, E>,
|
||||
state: Option<ScannerPublicationCommitState>,
|
||||
written_etag: Option<&str>,
|
||||
) -> bool {
|
||||
result.is_ok() && state == Some(ScannerPublicationCommitState::Committed) && written_etag.is_some_and(|etag| !etag.is_empty())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod root_publication_confirmation_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn root_publication_confirmation_requires_committed_state_and_write_revision() {
|
||||
let saved = Ok::<(), ()>(());
|
||||
for state in [
|
||||
None,
|
||||
Some(ScannerPublicationCommitState::Admitted),
|
||||
Some(ScannerPublicationCommitState::InFlight),
|
||||
Some(ScannerPublicationCommitState::AbortedBeforeCommit),
|
||||
Some(ScannerPublicationCommitState::Indeterminate),
|
||||
] {
|
||||
assert!(!root_ack_write_is_confirmed(&saved, state, Some("revision")));
|
||||
}
|
||||
for etag in [None, Some("")] {
|
||||
assert!(!root_ack_write_is_confirmed(&saved, Some(ScannerPublicationCommitState::Committed), etag));
|
||||
}
|
||||
assert!(root_ack_write_is_confirmed(
|
||||
&saved,
|
||||
Some(ScannerPublicationCommitState::Committed),
|
||||
Some("revision")
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn root_publication_confirmation_does_not_carry_state_across_cas_attempts() {
|
||||
let attempts = [
|
||||
(Err(()), Some(ScannerPublicationCommitState::Committed), Some("first")),
|
||||
(Ok(()), Some(ScannerPublicationCommitState::AbortedBeforeCommit), Some("second")),
|
||||
(Ok(()), None, Some("legacy")),
|
||||
(Ok(()), Some(ScannerPublicationCommitState::Committed), Some("confirmed")),
|
||||
];
|
||||
let confirmations = attempts
|
||||
.iter()
|
||||
.map(|(result, state, etag)| root_ack_write_is_confirmed(result, *state, *etag))
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(confirmations, [false, false, false, true]);
|
||||
}
|
||||
}
|
||||
|
||||
async fn read_root_publication_proof<S: ScannerObjectIO + ScannerConfigObjectDelete>(
|
||||
store: Arc<S>,
|
||||
ctx: &CancellationToken,
|
||||
deadline: tokio::time::Instant,
|
||||
epoch: u64,
|
||||
expected: &crate::scanner_io::ScannerPublicationExpectation,
|
||||
candidate: &DataUsageInfo,
|
||||
written_etag: Option<&str>,
|
||||
) -> Option<RootPublicationProof> {
|
||||
let read = async {
|
||||
let _admission = scanner_publication_admission_for_epoch(store.clone(), epoch).await?;
|
||||
let (bytes, revision) = read_config_with_revision(store, DATA_USAGE_OBJ_NAME_PATH.as_str())
|
||||
.await
|
||||
.ok()?;
|
||||
let bytes = bytes?;
|
||||
let DataUsageCacheRevision::Etag(etag) = revision else {
|
||||
return None;
|
||||
};
|
||||
if etag.is_empty() || written_etag.is_some_and(|written| written != etag) {
|
||||
return None;
|
||||
}
|
||||
let persisted: DataUsageInfo = serde_json::from_slice(&bytes).ok()?;
|
||||
if &persisted != candidate {
|
||||
return None;
|
||||
}
|
||||
let root_digest = Sha256::digest(&bytes).into();
|
||||
if ctx.is_cancelled() || tokio::time::Instant::now() >= deadline {
|
||||
return None;
|
||||
}
|
||||
Some(RootPublicationProof {
|
||||
candidate: expected.clone(),
|
||||
root_version: (etag, root_digest),
|
||||
})
|
||||
};
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = ctx.cancelled() => None,
|
||||
result = tokio::time::timeout_at(deadline, read) => result.ok().flatten(),
|
||||
}
|
||||
}
|
||||
|
||||
fn remote_lease_expired(deadline: Option<std::time::Instant>) -> bool {
|
||||
deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline)
|
||||
}
|
||||
@@ -166,6 +299,7 @@ pub(super) struct ScannerPublicationFence {
|
||||
pub(super) scanner_publication_lease_fence: Option<String>,
|
||||
pub(super) remote_lease_tokens: Vec<Uuid>,
|
||||
pub(super) lease_release_safe: Arc<AtomicBool>,
|
||||
pub(super) ack_expectation: Option<crate::scanner_io::ScannerPublicationExpectation>,
|
||||
}
|
||||
|
||||
impl ScannerPublicationFence {
|
||||
@@ -180,6 +314,7 @@ impl ScannerPublicationFence {
|
||||
scanner_publication_lease_fence,
|
||||
remote_lease_tokens: Vec::new(),
|
||||
lease_release_safe: Arc::new(AtomicBool::new(true)),
|
||||
ack_expectation: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -192,21 +327,26 @@ impl ScannerPublicationFence {
|
||||
self.lease_release_safe = lease_release_safe;
|
||||
self
|
||||
}
|
||||
|
||||
pub(super) fn with_ack_expectation(mut self, expected: Option<crate::scanner_io::ScannerPublicationExpectation>) -> Self {
|
||||
self.ack_expectation = expected;
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(super) enum DataUsagePersistTaskResult {
|
||||
Completed(DataUsagePersistOutcome),
|
||||
pub(super) enum DataUsagePersistTaskResult<T = DataUsagePersistOutcome> {
|
||||
Completed(T),
|
||||
Cancelled,
|
||||
TimedOut,
|
||||
JoinFailed(tokio::task::JoinError),
|
||||
}
|
||||
|
||||
pub(super) async fn wait_for_data_usage_persist_task(
|
||||
pub(super) async fn wait_for_data_usage_persist_task<T>(
|
||||
ctx: &CancellationToken,
|
||||
task: &mut AbortOnDropHandle<DataUsagePersistOutcome>,
|
||||
task: &mut AbortOnDropHandle<T>,
|
||||
timeout: Duration,
|
||||
) -> DataUsagePersistTaskResult {
|
||||
) -> DataUsagePersistTaskResult<T> {
|
||||
tokio::select! {
|
||||
biased;
|
||||
result = &mut *task => match result {
|
||||
@@ -320,6 +460,7 @@ where
|
||||
route_probe,
|
||||
)
|
||||
.await
|
||||
.outcome()
|
||||
}
|
||||
|
||||
pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence<
|
||||
@@ -333,7 +474,7 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel
|
||||
initial_baseline: Option<DataUsagePersistBaseline>,
|
||||
publication_fence: ScannerPublicationFence,
|
||||
route_probe: F,
|
||||
) -> DataUsagePersistOutcome
|
||||
) -> DataUsagePublicationResult
|
||||
where
|
||||
F: Fn() -> Fut + Send + Sync,
|
||||
Fut: Future<Output = Option<ScannerCycleDeferReason>> + Send,
|
||||
@@ -344,11 +485,15 @@ where
|
||||
scanner_publication_lease_fence,
|
||||
remote_lease_tokens,
|
||||
lease_release_safe,
|
||||
ack_expectation,
|
||||
} = publication_fence;
|
||||
let ack_deadline = scanner_publication_scope_deadline(data_usage_persist_timeout(), remote_lease_deadline);
|
||||
let mut outcome = DataUsagePersistOutcome::NoUpdate;
|
||||
let mut proof = None;
|
||||
let mut next_baseline = initial_baseline;
|
||||
|
||||
'updates: while let Some(mut data_usage_info) = receiver.recv().await {
|
||||
proof = None;
|
||||
let _activity_guard = ScannerActivityGuard::new();
|
||||
if ctx.is_cancelled() {
|
||||
break;
|
||||
@@ -523,10 +668,14 @@ where
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let sha256hex = (!data.is_empty()).then(|| hex_simd::encode_to_string(Sha256::digest(&data), hex_simd::AsciiCase::Lower));
|
||||
let data_digest: [u8; 32] = Sha256::digest(&data).into();
|
||||
let sha256hex = (!data.is_empty()).then(|| hex_simd::encode_to_string(data_digest, hex_simd::AsciiCase::Lower));
|
||||
let data = Bytes::from(data);
|
||||
let backup_due = !observational && data_usage_backup_due(&data_usage_info);
|
||||
let mut cas_retry = 0usize;
|
||||
let mut ack_epoch = None;
|
||||
let mut write_confirmed = false;
|
||||
let mut written_etag = None;
|
||||
let save_outcome = loop {
|
||||
if ctx.is_cancelled() {
|
||||
break 'updates;
|
||||
@@ -557,6 +706,7 @@ where
|
||||
} else {
|
||||
None
|
||||
};
|
||||
ack_epoch = Some(publication_epoch_for_save);
|
||||
let (existing_data, revision) = match baseline {
|
||||
Some(baseline) => (baseline.data, baseline.revision),
|
||||
None => match read_config_with_revision(storeapi.clone(), target_path).await {
|
||||
@@ -645,7 +795,7 @@ where
|
||||
}
|
||||
|
||||
let done_save = Metrics::time(Metric::SaveUsage);
|
||||
let save_result = {
|
||||
let (save_result, commit_state) = {
|
||||
let publication_scope = storeapi
|
||||
.scanner_data_usage_publication_commit_scope_with_release_flag(
|
||||
publication_epoch_for_save,
|
||||
@@ -681,24 +831,33 @@ where
|
||||
.await;
|
||||
drop(legacy_publication_admission);
|
||||
if let Some(scope) = publication_scope {
|
||||
match scope.wait_for_completion().await {
|
||||
ScannerPublicationCommitState::Committed | ScannerPublicationCommitState::AbortedBeforeCommit => {
|
||||
save_result
|
||||
}
|
||||
let state = scope.wait_for_completion().await;
|
||||
let result = match state {
|
||||
ScannerPublicationCommitState::Committed => save_result,
|
||||
ScannerPublicationCommitState::AbortedBeforeCommit => save_result,
|
||||
ScannerPublicationCommitState::Indeterminate
|
||||
| ScannerPublicationCommitState::Admitted
|
||||
| ScannerPublicationCommitState::InFlight => Err(EcstoreError::other(
|
||||
"scanner publication commit scope did not reach a safe terminal state",
|
||||
)),
|
||||
}
|
||||
};
|
||||
(result, Some(state))
|
||||
} else {
|
||||
save_result
|
||||
(save_result, None)
|
||||
}
|
||||
};
|
||||
done_save();
|
||||
|
||||
let attempt_confirmed = root_ack_write_is_confirmed(
|
||||
&save_result,
|
||||
commit_state,
|
||||
save_result.as_ref().ok().and_then(|info| info.etag.as_deref()),
|
||||
);
|
||||
|
||||
match save_result {
|
||||
Ok(object_info) => {
|
||||
write_confirmed = attempt_confirmed;
|
||||
written_etag = object_info.etag.as_ref().filter(|etag| !etag.is_empty()).cloned();
|
||||
if !observational {
|
||||
next_baseline = object_info
|
||||
.etag
|
||||
@@ -909,9 +1068,27 @@ where
|
||||
break 'updates;
|
||||
}
|
||||
}
|
||||
if !observational
|
||||
&& data_usage_info.usage_snapshot_converged == Some(true)
|
||||
&& matches!(outcome, DataUsagePersistOutcome::Saved | DataUsagePersistOutcome::AlreadyDurable)
|
||||
&& (outcome == DataUsagePersistOutcome::AlreadyDurable || write_confirmed)
|
||||
&& let (Some(expected), Some(epoch)) = (ack_expectation.as_ref(), ack_epoch)
|
||||
&& expected.matches_encoded_candidate(&data_digest)
|
||||
{
|
||||
proof = read_root_publication_proof(
|
||||
storeapi.clone(),
|
||||
&ctx,
|
||||
ack_deadline,
|
||||
epoch,
|
||||
expected,
|
||||
&data_usage_info,
|
||||
written_etag.as_deref(),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
outcome
|
||||
DataUsagePublicationResult { outcome, proof }
|
||||
}
|
||||
|
||||
async fn cleanup_observed_data_usage_snapshot_for_epoch_and_lease(
|
||||
|
||||
@@ -114,6 +114,11 @@ pub(crate) struct ScannerBucketScanScope {
|
||||
}
|
||||
|
||||
impl ScannerBucketScanScope {
|
||||
#[cfg(test)]
|
||||
pub(crate) fn selected_buckets_for_tests(&self) -> Option<&HashSet<String>> {
|
||||
self.selected_buckets.as_deref()
|
||||
}
|
||||
|
||||
fn is_default(&self) -> bool {
|
||||
self.selected_buckets.is_none() && self.baseline_scan_plan_digest.is_none()
|
||||
}
|
||||
@@ -218,8 +223,8 @@ fn complete_scanner_cache_snapshot_plan_digest(
|
||||
|
||||
fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option<DataUsageScanPlanDigest> {
|
||||
let authoritative = serde_json::from_slice::<DataUsageInfo>(proof.authoritative_data?).ok()?;
|
||||
if let Some(plan_digest) = complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, true) {
|
||||
return Some(plan_digest);
|
||||
if let Some(validated_digest) = complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, true) {
|
||||
return Some(validated_digest);
|
||||
}
|
||||
|
||||
// A complete but superseded observation may reuse its per-set cache only
|
||||
@@ -896,6 +901,7 @@ pub(crate) struct ScannerCycleResult {
|
||||
failed_dirty_usage: bool,
|
||||
pending_maintenance_work: bool,
|
||||
required_cycle_floor: Option<u64>,
|
||||
publication_expectation: Option<ScannerPublicationExpectation>,
|
||||
}
|
||||
|
||||
impl ScannerCycleResult {
|
||||
@@ -911,10 +917,12 @@ impl ScannerCycleResult {
|
||||
failed_dirty_usage: false,
|
||||
pending_maintenance_work: false,
|
||||
required_cycle_floor: None,
|
||||
publication_expectation: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn with_publication_epoch(mut self, publication_epoch: Option<u64>) -> Self {
|
||||
self.publication_expectation = None;
|
||||
self.publication_epoch = publication_epoch;
|
||||
self
|
||||
}
|
||||
@@ -924,6 +932,7 @@ impl ScannerCycleResult {
|
||||
}
|
||||
|
||||
fn with_activity_digest(mut self, activity_digest: [u8; 32]) -> Self {
|
||||
self.publication_expectation = None;
|
||||
self.activity_digest = Some(activity_digest);
|
||||
self
|
||||
}
|
||||
@@ -933,6 +942,7 @@ impl ScannerCycleResult {
|
||||
}
|
||||
|
||||
pub(crate) fn with_observational_snapshot_published(mut self, published: bool) -> Self {
|
||||
self.publication_expectation = None;
|
||||
self.observational_snapshot_published = published;
|
||||
self
|
||||
}
|
||||
@@ -942,16 +952,19 @@ impl ScannerCycleResult {
|
||||
}
|
||||
|
||||
fn with_failed_dirty_usage(mut self, failed_dirty_usage: bool) -> Self {
|
||||
self.publication_expectation = None;
|
||||
self.failed_dirty_usage = failed_dirty_usage;
|
||||
self
|
||||
}
|
||||
|
||||
fn with_pending_maintenance_work(mut self, pending_maintenance_work: bool) -> Self {
|
||||
self.publication_expectation = None;
|
||||
self.pending_maintenance_work = pending_maintenance_work;
|
||||
self
|
||||
}
|
||||
|
||||
fn with_required_cycle_floor(mut self, required_cycle_floor: Option<u64>) -> Self {
|
||||
self.publication_expectation = None;
|
||||
self.required_cycle_floor = required_cycle_floor;
|
||||
self
|
||||
}
|
||||
@@ -960,11 +973,13 @@ impl ScannerCycleResult {
|
||||
mut self,
|
||||
acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
|
||||
) -> Self {
|
||||
self.publication_expectation = None;
|
||||
self.remote_dirty_usage_acknowledgements = acknowledgements;
|
||||
self
|
||||
}
|
||||
|
||||
pub(crate) fn with_remote_publication_lease_targets(mut self, targets: Vec<(String, String, u64)>) -> Self {
|
||||
self.publication_expectation = None;
|
||||
self.remote_publication_lease_targets = targets;
|
||||
self
|
||||
}
|
||||
@@ -973,7 +988,32 @@ impl ScannerCycleResult {
|
||||
&self.remote_publication_lease_targets
|
||||
}
|
||||
|
||||
pub(crate) fn acknowledge_durable_usage(self) -> Vec<crate::scanner::ScannerDirtyUsageAcknowledgement> {
|
||||
pub(crate) fn publication_expectation(&self) -> Option<ScannerPublicationExpectation> {
|
||||
self.publication_expectation.clone()
|
||||
}
|
||||
|
||||
fn with_publication_expectation(mut self, expectation: Option<ScannerPublicationExpectation>) -> Self {
|
||||
// Seal only after all coverage and acknowledgement inputs are final.
|
||||
self.publication_expectation = expectation;
|
||||
self
|
||||
}
|
||||
|
||||
pub(crate) fn acknowledge_durable_usage(
|
||||
self,
|
||||
proof: &crate::scanner::RootPublicationProof,
|
||||
) -> Vec<crate::scanner::ScannerDirtyUsageAcknowledgement> {
|
||||
if self.status != ScannerCycleStatus::Complete
|
||||
|| self
|
||||
.publication_expectation
|
||||
.as_ref()
|
||||
.is_none_or(|expected| proof.verified_version_for(expected).is_none())
|
||||
{
|
||||
return Vec::new();
|
||||
}
|
||||
self.clear_verified_usage()
|
||||
}
|
||||
|
||||
fn clear_verified_usage(self) -> Vec<crate::scanner::ScannerDirtyUsageAcknowledgement> {
|
||||
if let Some(snapshot) = self.dirty_usage_clear {
|
||||
clear_dirty_usage_buckets(&snapshot);
|
||||
}
|
||||
@@ -1013,6 +1053,7 @@ mod publish_gate_tests;
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
pub(crate) use cache::ScannerPublicationExpectation;
|
||||
use cache::*;
|
||||
use dirty_usage::*;
|
||||
use guards::*;
|
||||
|
||||
@@ -308,6 +308,86 @@ impl<'a> ValidatedScannerSnapshot<'a> {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub(crate) struct ScannerPublicationExpectation {
|
||||
candidate: Arc<([u8; 32], DataUsageScanPlanDigest)>,
|
||||
}
|
||||
|
||||
impl ScannerPublicationExpectation {
|
||||
pub(crate) fn matches_encoded_candidate(&self, digest: &[u8; 32]) -> bool {
|
||||
&self.candidate.0 == digest
|
||||
}
|
||||
|
||||
pub(crate) fn same_candidate(&self, other: &Self) -> bool {
|
||||
Arc::ptr_eq(&self.candidate, &other.candidate) && self.candidate.1 == other.candidate.1
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) struct ValidatedUsageCandidate {
|
||||
data: DataUsageInfo,
|
||||
#[cfg(test)]
|
||||
last_update: SystemTime,
|
||||
coverage_digest: DataUsageScanPlanDigest,
|
||||
}
|
||||
|
||||
pub(super) fn empty_namespace_usage_candidate(
|
||||
all_buckets: &[BucketInfo],
|
||||
sources: &HashSet<DataUsageCacheSource>,
|
||||
buckets_by_source: &HashMap<DataUsageCacheSource, Vec<BucketInfo>>,
|
||||
identity: ScannerSnapshotIdentity,
|
||||
) -> Option<ValidatedUsageCandidate> {
|
||||
if !all_buckets.is_empty()
|
||||
|| sources.is_empty()
|
||||
|| sources.len() != buckets_by_source.len()
|
||||
|| sources
|
||||
.iter()
|
||||
.any(|source| buckets_by_source.get(source).is_none_or(|buckets| !buckets.is_empty()))
|
||||
{
|
||||
return None;
|
||||
}
|
||||
let last_update = SystemTime::now();
|
||||
Some(ValidatedUsageCandidate {
|
||||
data: DataUsageInfo {
|
||||
last_update: Some(last_update),
|
||||
scanner_cycle: Some(identity.cycle),
|
||||
scanner_epoch: Some(identity.leader_epoch),
|
||||
usage_snapshot_complete: true,
|
||||
..Default::default()
|
||||
},
|
||||
#[cfg(test)]
|
||||
last_update,
|
||||
coverage_digest: identity.coverage_digest,
|
||||
})
|
||||
}
|
||||
|
||||
impl ValidatedUsageCandidate {
|
||||
pub(super) fn prepare(mut self, status: ScannerCycleStatus) -> (DataUsageInfo, Option<ScannerPublicationExpectation>) {
|
||||
self.data.usage_snapshot_converged = Some(status == ScannerCycleStatus::Complete);
|
||||
let expectation = if status == ScannerCycleStatus::Complete {
|
||||
struct DigestWriter(Sha256);
|
||||
impl std::io::Write for DigestWriter {
|
||||
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
|
||||
self.0.update(bytes);
|
||||
Ok(bytes.len())
|
||||
}
|
||||
fn flush(&mut self) -> std::io::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
let mut writer = DigestWriter(Sha256::new());
|
||||
serde_json::to_writer(&mut writer, &self.data)
|
||||
.ok()
|
||||
.map(|()| ScannerPublicationExpectation {
|
||||
candidate: Arc::new((writer.0.finalize().into(), self.coverage_digest)),
|
||||
})
|
||||
} else {
|
||||
None
|
||||
};
|
||||
(self.data, expectation)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(super) fn completed_data_usage_info(
|
||||
results: &[DataUsageCache],
|
||||
scope: &ScannerSnapshotScope<'_>,
|
||||
@@ -316,6 +396,18 @@ pub(super) fn completed_data_usage_info(
|
||||
budget_elapsed: bool,
|
||||
cancelled: bool,
|
||||
) -> Option<(DataUsageInfo, SystemTime)> {
|
||||
completed_usage_candidate(results, scope, tier_registry_names, bucket_plan_complete, budget_elapsed, cancelled)
|
||||
.map(|candidate| (candidate.data, candidate.last_update))
|
||||
}
|
||||
|
||||
pub(super) fn completed_usage_candidate(
|
||||
results: &[DataUsageCache],
|
||||
scope: &ScannerSnapshotScope<'_>,
|
||||
tier_registry_names: &[String],
|
||||
bucket_plan_complete: bool,
|
||||
budget_elapsed: bool,
|
||||
cancelled: bool,
|
||||
) -> Option<ValidatedUsageCandidate> {
|
||||
if !bucket_plan_complete {
|
||||
return None;
|
||||
}
|
||||
@@ -393,7 +485,12 @@ pub(super) fn completed_data_usage_info(
|
||||
usage_snapshot_set_states,
|
||||
..Default::default()
|
||||
};
|
||||
Some((data_usage_info, merged_last_update))
|
||||
Some(ValidatedUsageCandidate {
|
||||
data: data_usage_info,
|
||||
#[cfg(test)]
|
||||
last_update: merged_last_update,
|
||||
coverage_digest: scope.identity.coverage_digest,
|
||||
})
|
||||
}
|
||||
|
||||
fn tier_accounting_proof_is_publishable(
|
||||
|
||||
@@ -338,12 +338,21 @@ where
|
||||
dirty_usage_status,
|
||||
activity_status,
|
||||
);
|
||||
let empty_usage = DataUsageInfo {
|
||||
last_update: Some(SystemTime::now()),
|
||||
scanner_cycle: Some(want_cycle),
|
||||
usage_snapshot_complete: true,
|
||||
..Default::default()
|
||||
let Some(candidate) = empty_namespace_usage_candidate(
|
||||
&all_buckets,
|
||||
&expected_sources,
|
||||
&buckets_by_source,
|
||||
ScannerSnapshotIdentity {
|
||||
cycle: want_cycle,
|
||||
leader_epoch,
|
||||
plan_digest: scan_plan_digest,
|
||||
coverage_digest: bucket_coverage_digest,
|
||||
tier_registry_generation: Some(tier_registry_generation),
|
||||
},
|
||||
) else {
|
||||
return Ok(ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None).with_publication_epoch(publication_epoch));
|
||||
};
|
||||
let (empty_usage, publication_expectation) = candidate.prepare(status);
|
||||
let observational_snapshot_published = if should_publish_observational_snapshot(status) {
|
||||
publish_observational_snapshot(&updates, empty_usage).await?
|
||||
} else {
|
||||
@@ -366,7 +375,8 @@ where
|
||||
.with_activity_digest(activity_digest)
|
||||
.with_observational_snapshot_published(observational_snapshot_published)
|
||||
.with_remote_publication_lease_targets(remote_publication_lease_targets)
|
||||
.with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements));
|
||||
.with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)
|
||||
.with_publication_expectation(publication_expectation));
|
||||
}
|
||||
|
||||
let total_results = expected_sources.len();
|
||||
@@ -595,7 +605,7 @@ where
|
||||
let (activity_status, remote_publication_lease_targets) =
|
||||
scanner_cycle_activity_status(store, distributed, &activity_before).await;
|
||||
let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::<Vec<_>>();
|
||||
let completed_usage = completed_data_usage_info(
|
||||
let completed_usage = completed_usage_candidate(
|
||||
&results,
|
||||
&ScannerSnapshotScope {
|
||||
sources: &expected_sources,
|
||||
@@ -636,7 +646,10 @@ where
|
||||
dirty_usage_status,
|
||||
activity_status,
|
||||
);
|
||||
let observational_snapshot_published = if let Some((data_usage_info, _)) = completed_usage {
|
||||
let mut publication_expectation = None;
|
||||
let observational_snapshot_published = if let Some(candidate) = completed_usage {
|
||||
let (data_usage_info, expectation) = candidate.prepare(cycle_status);
|
||||
publication_expectation = expectation;
|
||||
if should_publish_observational_snapshot(cycle_status) {
|
||||
publish_observational_snapshot(&updates, data_usage_info).await?
|
||||
} else {
|
||||
@@ -674,5 +687,6 @@ where
|
||||
.with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)
|
||||
.with_failed_dirty_usage(!failed_buckets.is_empty())
|
||||
.with_pending_maintenance_work(pending_maintenance_work)
|
||||
.with_required_cycle_floor(required_cycle_floor))
|
||||
.with_required_cycle_floor(required_cycle_floor)
|
||||
.with_publication_expectation(publication_expectation))
|
||||
}
|
||||
|
||||
@@ -462,6 +462,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure(
|
||||
.await
|
||||
.expect("fixture rename tail should finish before the usage scan");
|
||||
}
|
||||
wait_for_namespace_commit_tails(&store).await;
|
||||
let ctx = CancellationToken::new();
|
||||
let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
|
||||
let (updates, receiver) = mpsc::channel(1);
|
||||
@@ -511,15 +512,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure(
|
||||
.put_object("cold-bucket", "new", &mut reader, &ScannerObjectOptions::default())
|
||||
.await
|
||||
.expect("new cold object should persist");
|
||||
let lock = store.pools[0].disk_set[0]
|
||||
.new_ns_lock("cold-bucket", "new")
|
||||
.await
|
||||
.expect("fixture namespace lock should be created");
|
||||
let _settled = lock
|
||||
.get_write_lock(Duration::from_secs(30))
|
||||
.await
|
||||
.expect("fixture rename tail should finish before the usage scan");
|
||||
drop(_settled);
|
||||
wait_for_namespace_commit_tails(&store).await;
|
||||
record_dirty_usage_bucket("hot-bucket");
|
||||
if scan_mode == HealScanMode::Normal && !requires_full_scan {
|
||||
record_dirty_usage_bucket("cold-bucket");
|
||||
@@ -1011,8 +1004,8 @@ fn dirty_usage_snapshot_clears_a_stably_absent_bucket_after_durable_save() {
|
||||
assert!(dirty_usage_buckets().contains_key("temporarily-omitted"));
|
||||
assert_eq!(dirty_usage_snapshot_status(&snapshot), DirtyUsageSnapshotStatus::Current);
|
||||
|
||||
let acknowledgements = ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(snapshot.buckets.as_ref().clone()))
|
||||
.acknowledge_durable_usage();
|
||||
let acknowledgements =
|
||||
ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(snapshot.buckets.as_ref().clone())).clear_verified_usage();
|
||||
assert!(acknowledgements.is_empty());
|
||||
assert!(!dirty_usage_buckets().contains_key("temporarily-omitted"));
|
||||
clear_dirty_usage_buckets_for_tests();
|
||||
@@ -1118,7 +1111,7 @@ fn dirty_usage_is_acknowledged_only_after_durable_usage_confirmation() {
|
||||
assert!(dirty_usage_buckets().contains_key("photos"));
|
||||
|
||||
let confirmed = ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(snapshot.buckets.as_ref().clone()));
|
||||
let acknowledgements = confirmed.acknowledge_durable_usage();
|
||||
let acknowledgements = confirmed.clear_verified_usage();
|
||||
assert!(acknowledgements.is_empty());
|
||||
assert!(!dirty_usage_buckets().contains_key("photos"));
|
||||
clear_dirty_usage_buckets_for_tests();
|
||||
|
||||
@@ -55,6 +55,14 @@ Standard S3 areas that must not be described as complete:
|
||||
|
||||
`excluded_tests.txt` holds tests that must not block the compatibility gate: vendor-specific or non-portable behavior, and intentionally unsupported product behavior such as ACL authorization.
|
||||
|
||||
## Intentional Deviations From AWS S3
|
||||
|
||||
Object keys are stored as file-system paths under each drive (`{drive}/{bucket}/{object}/xl.meta`), the same layout MinIO uses. The rules below exist to keep that layout unambiguous and are not compatibility gaps to close; clients that need the AWS behavior must adapt on their side.
|
||||
|
||||
| Behavior | RustFS | AWS S3 | Why |
|
||||
|---|---|---|---|
|
||||
| Object key with a `.` or `..` path segment, or an empty segment (`//`), such as `a//b/./c/../d` | `400 InvalidArgument` (`check_object_args` in `crates/ecstore/src/bucket/utils.rs`, mirroring MinIO `IsValidObjectPrefix`) | Accepted as an opaque key | A `..` segment would resolve to a parent directory and `.`/`//` segments would alias other keys on disk; encoding them would change the MinIO-compatible on-disk format. |
|
||||
|
||||
## Update Rule
|
||||
|
||||
When a feature starts passing, move its test entries from `unimplemented_tests.txt` to `implemented_tests.txt` and update the row here in the same PR. Do not change README wording beyond the supported coverage. Handler-level status (missing, stubbed, or diverging endpoints) is tracked in [minio-rustfs-router-compatibility.md](minio-rustfs-router-compatibility.md).
|
||||
|
||||
@@ -470,6 +470,14 @@ async fn authorize_manual_transition_request(req: &S3Request<Body>) -> S3Result<
|
||||
/// The credential pre-check keeps this endpoint family's historical
|
||||
/// missing-credentials message (the shared gate reports "get cred failed") and
|
||||
/// still yields the masked actor every transition audit log records.
|
||||
async fn authorize_recovery_admin_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<String> {
|
||||
if req.credentials.is_none() {
|
||||
return Err(admin_s3_error(AdminS3ErrorCode::InvalidRequest, "authentication required"));
|
||||
}
|
||||
let credentials = authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
|
||||
Ok(recovery_actor_sha256(&credentials))
|
||||
}
|
||||
|
||||
async fn authorize_transition_admin_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<String> {
|
||||
let Some(input_cred) = req.credentials.as_ref() else {
|
||||
return Err(s3_error!(InvalidRequest, "authentication required"));
|
||||
@@ -481,14 +489,6 @@ async fn authorize_transition_admin_request(req: &S3Request<Body>, action: Admin
|
||||
Ok(actor)
|
||||
}
|
||||
|
||||
async fn authorize_recovery_admin_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<String> {
|
||||
if req.credentials.is_none() {
|
||||
return Err(admin_s3_error(AdminS3ErrorCode::InvalidRequest, "authentication required"));
|
||||
}
|
||||
let credentials = authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
|
||||
Ok(recovery_actor_sha256(&credentials))
|
||||
}
|
||||
|
||||
fn transition_transaction_id_from_params(params: &Params<'_, '_>) -> S3Result<Uuid> {
|
||||
Uuid::parse_str(params.get("transaction_id").unwrap_or(""))
|
||||
.map_err(|_| s3_error!(InvalidArgument, "invalid transition transaction id"))
|
||||
@@ -1498,23 +1498,6 @@ impl Operation for ManualTransitionJobCancelHandler {
|
||||
}
|
||||
}
|
||||
|
||||
pub struct TransitionReconcileInspectHandler {}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Operation for TransitionReconcileInspectHandler {
|
||||
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
authorize_transition_admin_request(&req, AdminAction::ListTierAction).await?;
|
||||
let transaction_id = transition_transaction_id_from_params(¶ms)?;
|
||||
let Some(store) = object_store_from_extensions(&req.extensions) else {
|
||||
return Err(s3_error!(InternalError, "object store is not initialized"));
|
||||
};
|
||||
let status = inspect_transition_transaction_for_operator(store, transaction_id)
|
||||
.await
|
||||
.map_err(map_transition_operator_error)?;
|
||||
json_response(StatusCode::OK, &status)
|
||||
}
|
||||
}
|
||||
|
||||
pub struct IlmRecoveryControlListHandler {}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
@@ -1634,6 +1617,23 @@ impl Operation for IlmRecoveryExportDownloadHandler {
|
||||
}
|
||||
}
|
||||
|
||||
pub struct TransitionReconcileInspectHandler {}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Operation for TransitionReconcileInspectHandler {
|
||||
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
authorize_transition_admin_request(&req, AdminAction::ListTierAction).await?;
|
||||
let transaction_id = transition_transaction_id_from_params(¶ms)?;
|
||||
let Some(store) = object_store_from_extensions(&req.extensions) else {
|
||||
return Err(s3_error!(InternalError, "object store is not initialized"));
|
||||
};
|
||||
let status = inspect_transition_transaction_for_operator(store, transaction_id)
|
||||
.await
|
||||
.map_err(map_transition_operator_error)?;
|
||||
json_response(StatusCode::OK, &status)
|
||||
}
|
||||
}
|
||||
|
||||
pub struct TransitionReconcileApplyHandler {}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
@@ -2153,7 +2153,7 @@ mod tests {
|
||||
let inspect = src
|
||||
.split("impl Operation for TransitionReconcileInspectHandler")
|
||||
.nth(1)
|
||||
.and_then(|block| block.split("impl Operation for TransitionReconcileApplyHandler").next())
|
||||
.and_then(|block| block.split("pub struct IlmRecoveryControlListHandler").next())
|
||||
.expect("inspect handler block");
|
||||
assert!(inspect.contains("AdminAction::ListTierAction"));
|
||||
assert!(!inspect.contains("AdminAction::SetTierAction"));
|
||||
@@ -2502,7 +2502,7 @@ mod tests {
|
||||
let wrapper = extract_block_between_markers(
|
||||
production,
|
||||
"async fn authorize_transition_admin_request",
|
||||
"fn transition_transaction_id_from_params",
|
||||
"async fn authorize_recovery_admin_request",
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
|
||||
@@ -4070,14 +4070,13 @@ mod tests {
|
||||
abort_incomplete_multipart_upload: None,
|
||||
del_marker_expiration: None,
|
||||
filter: Some(s3s::dto::LifecycleRuleFilter {
|
||||
prefix: Some("logs/".to_string()),
|
||||
tag: Some(s3s::dto::Tag {
|
||||
key: Some("env".to_string()),
|
||||
value: Some("prod".to_string()),
|
||||
and: Some(s3s::dto::LifecycleRuleAndOperator {
|
||||
prefix: Some("logs/".to_string()),
|
||||
..Default::default()
|
||||
}),
|
||||
..Default::default()
|
||||
}),
|
||||
id: Some("two-predicates".to_string()),
|
||||
id: Some("one-member-and".to_string()),
|
||||
noncurrent_version_expiration: None,
|
||||
noncurrent_version_transitions: None,
|
||||
prefix: None,
|
||||
@@ -4087,7 +4086,7 @@ mod tests {
|
||||
&ObjectLockConfiguration::default(),
|
||||
)
|
||||
.await
|
||||
.expect_err("a Filter with two predicates is a schema violation");
|
||||
.expect_err("a Filter And with one predicate is a schema violation");
|
||||
assert_eq!(*lifecycle_validation_error(&malformed).code(), S3ErrorCode::MalformedXML);
|
||||
|
||||
let invalid_value = validate_lifecycle_config(
|
||||
|
||||
@@ -1,33 +0,0 @@
|
||||
# Legacy Heal Outcome Compatibility
|
||||
|
||||
This fixture executes the real `madmin-go` HTTP decoder and a pinned `mc` binary against synthetic v3 responses. Rust owner, admin adapter, and SDK tests validate the same JSON cases in `crates/madmin/tests/fixtures/heal-outcome-v3.json`. It does not run a storage repair or prove distributed recovery.
|
||||
|
||||
Pinned primary sources:
|
||||
|
||||
- `mc` release `RELEASE.2025-08-13T08-35-41Z`, commit `7394ce0dd2a80935aded936b09fa12cbb3cb8096`: [polling implementation](https://github.com/minio/mc/blob/7394ce0dd2a80935aded936b09fa12cbb3cb8096/cmd/admin-heal-ui.go#L414).
|
||||
- Its `madmin-go/v3` dependency is `v3.0.107-0.20250415152934-4b504b82db63`: [decoder and response type](https://github.com/minio/madmin-go/blob/4b504b82db633e978a57d49443b2be75824244c3/heal-commands.go#L101).
|
||||
|
||||
The old decoder ignores additional JSON fields. The old poller returns success for `finished` without examining `detail`; only `stopped` returns a terminal error. Consequently `completed_with_errors` retains its canonical outcome and complete traversal coverage, but uses legacy summary `stopped`. `completed` describes execution only: unknown storage receipts remain `unknown`, not `repaired`.
|
||||
|
||||
Run from the repository root with an isolated tool cache and binary directory:
|
||||
|
||||
```sh
|
||||
(
|
||||
set -eu
|
||||
compat_dir=$(mktemp -d)
|
||||
trap 'rm -rf "$compat_dir"' EXIT
|
||||
export GOPATH="$compat_dir/gopath" GOMODCACHE="$compat_dir/mod" GOCACHE="$compat_dir/cache" GOBIN="$compat_dir/bin"
|
||||
export CGO_ENABLED=0 GOTOOLCHAIN=local GOMAXPROCS=2
|
||||
go install github.com/minio/mc@v0.0.0-20250813083541-7394ce0dd2a8
|
||||
cd scripts/compat/heal-outcome
|
||||
MC_BINARY="$compat_dir/bin/mc" NO_PROXY=127.0.0.1,localhost go test -mod=readonly -p 2 -count=1 -v ./...
|
||||
)
|
||||
```
|
||||
|
||||
The subshell keeps the calling shell unchanged. `MC_BINARY` is mandatory and its Go build metadata must identify the pinned commit. Each subprocess gets a temporary mc configuration directory and synthetic credentials; it never edits the user's mc configuration. Loopback socket permission is required. The Go tests do not skip unavailable prerequisites.
|
||||
|
||||
Six cases cover completed traversal, unknown repair proof, completed traversal with failures, cancellation, deadline, and untraversable listing. Two receiver cases carry a remote `finished` summary that contradicts an aborted or completed-with-errors outcome. The admin test applies the heal owner's wire validator to each `remoteResponse` and must produce the corresponding public `response`; the old CLI must then exit with an error. Unknown extension fields remain intact. Unknown or missing execution fields cannot validate a successful summary.
|
||||
|
||||
New counters do not replace or reinterpret legacy progress. Outcome is a cumulative snapshot, not a page delta; `sinceSeq` only pages legacy result items. Result cursors and truncation markers remain separate from execution and traversal coverage.
|
||||
|
||||
Two existing CLI limitations remain explicit: this mc does not terminate on `notFound`, and `-f` polling sends `forceStart` together with `clientToken`, a combination the RustFS v3 request contract rejects. The fixture uses the standard non-force polling flow. Neither limitation is hidden by emitting a new summary string or reporting a missing task as completed.
|
||||
@@ -1,137 +0,0 @@
|
||||
// Copyright 2026 RustFS Team
|
||||
// Licensed under the Apache License, Version 2.0.
|
||||
|
||||
package compat_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"debug/buildinfo"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
madmin "github.com/minio/madmin-go/v3"
|
||||
)
|
||||
|
||||
type fixture struct {
|
||||
Name string `json:"name"`
|
||||
CLIExit int `json:"cliExit"`
|
||||
Response json.RawMessage `json:"response"`
|
||||
}
|
||||
|
||||
func fixtures(t *testing.T) []fixture {
|
||||
t.Helper()
|
||||
data, err := os.ReadFile("../../../crates/madmin/tests/fixtures/heal-outcome-v3.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var cases []fixture
|
||||
if err := json.Unmarshal(data, &cases); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(cases) != 8 {
|
||||
t.Fatalf("expected eight owner/receiver-validated fixtures, got %d", len(cases))
|
||||
}
|
||||
return cases
|
||||
}
|
||||
|
||||
func fixtureServer(t *testing.T, response []byte, polls *atomic.Int32) *httptest.Server {
|
||||
t.Helper()
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost || !strings.HasPrefix(r.URL.Path, "/minio/admin/v3/heal/") {
|
||||
t.Errorf("unexpected client request %s %s", r.Method, r.URL.Path)
|
||||
http.Error(w, "unexpected request", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if r.URL.Query().Get("clientToken") == "" {
|
||||
fmt.Fprint(w, `{"clientToken":"fixture-token","clientAddress":"","startTime":"2026-01-01T00:00:00Z"}`)
|
||||
return
|
||||
}
|
||||
polls.Add(1)
|
||||
w.Write(response)
|
||||
}))
|
||||
t.Cleanup(server.Close)
|
||||
return server
|
||||
}
|
||||
|
||||
func TestLegacyMadminDecoder(t *testing.T) {
|
||||
for _, f := range fixtures(t) {
|
||||
t.Run(f.Name, func(t *testing.T) {
|
||||
var polls atomic.Int32
|
||||
server := fixtureServer(t, f.Response, &polls)
|
||||
client, err := madmin.New(strings.TrimPrefix(server.URL, "http://"), "fixture-access", "fixture-secret", false)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
_, status, err := client.Heal(ctx, "bucket", "", madmin.HealOpts{}, "fixture-token", false, false)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var expected struct {
|
||||
Summary string `json:"summary"`
|
||||
Detail string `json:"detail"`
|
||||
}
|
||||
if err := json.Unmarshal(f.Response, &expected); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if status.Summary != expected.Summary || status.FailureDetail != expected.Detail || polls.Load() != 1 {
|
||||
t.Fatalf("decoder changed legacy fields: %+v, polls=%d", status, polls.Load())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestLegacyMCPoll(t *testing.T) {
|
||||
binary := os.Getenv("MC_BINARY")
|
||||
if binary == "" {
|
||||
t.Fatal("MC_BINARY must point to the pinned mc release; this check cannot be skipped")
|
||||
}
|
||||
info, err := buildinfo.ReadFile(binary)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if info.Main.Path != "github.com/minio/mc" || !strings.Contains(info.Main.Version, "7394ce0dd2a8") {
|
||||
t.Fatalf("expected mc RELEASE.2025-08-13T08-35-41Z (7394ce0dd2a8), got %+v", info.Main)
|
||||
}
|
||||
for _, f := range fixtures(t) {
|
||||
t.Run(f.Name, func(t *testing.T) {
|
||||
var polls atomic.Int32
|
||||
server := fixtureServer(t, f.Response, &polls)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
cmd := exec.CommandContext(ctx, binary, "--config-dir", filepath.Join(t.TempDir(), "mc"), "--json", "admin", "heal", "--recursive", "w23/bucket")
|
||||
cmd.Env = append(os.Environ(), "MC_HOST_w23="+strings.Replace(server.URL, "http://", "http://fixture-access:fixture-secret@", 1), "MC_NO_COLOR=1")
|
||||
output, err := cmd.CombinedOutput()
|
||||
if ctx.Err() != nil {
|
||||
t.Fatalf("legacy poll did not terminate: %s", output)
|
||||
}
|
||||
exit := 0
|
||||
if err != nil {
|
||||
var ok bool
|
||||
var status *exec.ExitError
|
||||
status, ok = err.(*exec.ExitError)
|
||||
if !ok {
|
||||
t.Fatal(err)
|
||||
}
|
||||
exit = status.ExitCode()
|
||||
}
|
||||
if exit != f.CLIExit || polls.Load() != 1 {
|
||||
t.Fatalf("exit=%d expected=%d polls=%d output=%s", exit, f.CLIExit, polls.Load(), output)
|
||||
}
|
||||
if f.CLIExit != 0 && !strings.Contains(string(output), "Heal had an error") {
|
||||
t.Fatalf("failure was not the expected legacy terminal result: %s", output)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,42 +0,0 @@
|
||||
module rustfs.local/heal-outcome-compat
|
||||
|
||||
go 1.25.0
|
||||
|
||||
require github.com/minio/madmin-go/v3 v3.0.107-0.20250415152934-4b504b82db63
|
||||
|
||||
require (
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/dustin/go-humanize v1.0.1 // indirect
|
||||
github.com/go-ini/ini v1.67.0 // indirect
|
||||
github.com/go-ole/go-ole v1.3.0 // indirect
|
||||
github.com/goccy/go-json v0.10.5 // indirect
|
||||
github.com/golang-jwt/jwt/v4 v4.5.2 // indirect
|
||||
github.com/golang/protobuf v1.5.4 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.10 // indirect
|
||||
github.com/lufia/plan9stats v0.0.0-20250317134145-8bc96cf8fc35 // indirect
|
||||
github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
|
||||
github.com/minio/md5-simd v1.1.2 // indirect
|
||||
github.com/minio/minio-go/v7 v7.0.90 // indirect
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
|
||||
github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c // indirect
|
||||
github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
|
||||
github.com/prometheus/client_model v0.6.2 // indirect
|
||||
github.com/prometheus/common v0.63.0 // indirect
|
||||
github.com/prometheus/procfs v0.16.0 // indirect
|
||||
github.com/prometheus/prom2json v1.4.2 // indirect
|
||||
github.com/prometheus/prometheus v0.303.0 // indirect
|
||||
github.com/rs/xid v1.6.0 // indirect
|
||||
github.com/safchain/ethtool v0.5.10 // indirect
|
||||
github.com/secure-io/sio-go v0.3.1 // indirect
|
||||
github.com/shirou/gopsutil/v3 v3.24.5 // indirect
|
||||
github.com/shoenig/go-m1cpu v0.1.6 // indirect
|
||||
github.com/tinylib/msgp v1.2.5 // indirect
|
||||
github.com/tklauser/go-sysconf v0.3.15 // indirect
|
||||
github.com/tklauser/numcpus v0.10.0 // indirect
|
||||
github.com/yusufpapurcu/wmi v1.2.4 // indirect
|
||||
golang.org/x/crypto v0.52.0 // indirect
|
||||
golang.org/x/net v0.55.0 // indirect
|
||||
golang.org/x/sync v0.13.0 // indirect
|
||||
golang.org/x/sys v0.45.0 // indirect
|
||||
google.golang.org/protobuf v1.36.6 // indirect
|
||||
)
|
||||
@@ -1,97 +0,0 @@
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
|
||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
|
||||
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
|
||||
github.com/go-ini/ini v1.67.0 h1:z6ZrTEZqSWOTyH2FlglNbNgARyHG8oLW9gMELqKr06A=
|
||||
github.com/go-ini/ini v1.67.0/go.mod h1:ByCAeIL28uOIIG0E3PJtZPDL8WnHpFKFOtgjp+3Ies8=
|
||||
github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
|
||||
github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
|
||||
github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
|
||||
github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
|
||||
github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
|
||||
github.com/golang-jwt/jwt/v4 v4.5.2 h1:YtQM7lnr8iZ+j5q71MGKkNw9Mn7AjHM68uc9g5fXeUI=
|
||||
github.com/golang-jwt/jwt/v4 v4.5.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0=
|
||||
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
|
||||
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
|
||||
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
|
||||
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
|
||||
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
|
||||
github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
|
||||
github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE=
|
||||
github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
|
||||
github.com/lufia/plan9stats v0.0.0-20250317134145-8bc96cf8fc35 h1:PpXWgLPs+Fqr325bN2FD2ISlRRztXibcX6e8f5FR5Dc=
|
||||
github.com/lufia/plan9stats v0.0.0-20250317134145-8bc96cf8fc35/go.mod h1:autxFIvghDt3jPTLoqZ9OZ7s9qTGNAWmYCjVFWPX/zg=
|
||||
github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=
|
||||
github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
|
||||
github.com/minio/madmin-go/v3 v3.0.107-0.20250415152934-4b504b82db63 h1:ktN/FrMuM9sjvjIbPZYRKeHEzBDOXQdpYUDiNO0CutE=
|
||||
github.com/minio/madmin-go/v3 v3.0.107-0.20250415152934-4b504b82db63/go.mod h1:U0bL6ip4yKFwvo0keonUcWFQp0Hd462tOLLeVyPzWmE=
|
||||
github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
|
||||
github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM=
|
||||
github.com/minio/minio-go/v7 v7.0.90 h1:TmSj1083wtAD0kEYTx7a5pFsv3iRYMsOJ6A4crjA1lE=
|
||||
github.com/minio/minio-go/v7 v7.0.90/go.mod h1:uvMUcGrpgeSAAI6+sD3818508nUyMULw94j2Nxku/Go=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
|
||||
github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c h1:dAMKvw0MlJT1GshSTtih8C2gDs04w8dReiOGXrGLNoY=
|
||||
github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM=
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU=
|
||||
github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
|
||||
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
|
||||
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
|
||||
github.com/prometheus/common v0.63.0 h1:YR/EIY1o3mEFP/kZCD7iDMnLPlGyuU2Gb3HIcXnA98k=
|
||||
github.com/prometheus/common v0.63.0/go.mod h1:VVFF/fBIoToEnWRVkYoXEkq3R3paCoxG9PXP74SnV18=
|
||||
github.com/prometheus/procfs v0.16.0 h1:xh6oHhKwnOJKMYiYBDWmkHqQPyiY40sny36Cmx2bbsM=
|
||||
github.com/prometheus/procfs v0.16.0/go.mod h1:8veyXUu3nGP7oaCxhX6yeaM5u4stL2FeMXnCqhDthZg=
|
||||
github.com/prometheus/prom2json v1.4.2 h1:PxCTM+Whqi/eykO1MKsEL0p/zMpxp9ybpsmdFamw6po=
|
||||
github.com/prometheus/prom2json v1.4.2/go.mod h1:zuvPm7u3epZSbXPWHny6G+o8ETgu6eAK3oPr6yFkRWE=
|
||||
github.com/prometheus/prometheus v0.303.0 h1:wsNNsbd4EycMCphYnTmNY9JASBVbp7NWwJna857cGpA=
|
||||
github.com/prometheus/prometheus v0.303.0/go.mod h1:8PMRi+Fk1WzopMDeb0/6hbNs9nV6zgySkU/zds5Lu3o=
|
||||
github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU=
|
||||
github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0=
|
||||
github.com/safchain/ethtool v0.5.10 h1:Im294gZtuf4pSGJRAOGKaASNi3wMeFaGaWuSaomedpc=
|
||||
github.com/safchain/ethtool v0.5.10/go.mod h1:w9jh2Lx7YBR4UwzLkzCmWl85UY0W2uZdd7/DckVE5+c=
|
||||
github.com/secure-io/sio-go v0.3.1 h1:dNvY9awjabXTYGsTF1PiCySl9Ltofk9GA3VdWlo7rRc=
|
||||
github.com/secure-io/sio-go v0.3.1/go.mod h1:+xbkjDzPjwh4Axd07pRKSNriS9SCiYksWnZqdnfpQxs=
|
||||
github.com/shirou/gopsutil/v3 v3.24.5 h1:i0t8kL+kQTvpAYToeuiVk3TgDeKOFioZO3Ztz/iZ9pI=
|
||||
github.com/shirou/gopsutil/v3 v3.24.5/go.mod h1:bsoOS1aStSs9ErQ1WWfxllSeS1K5D+U30r2NfcubMVk=
|
||||
github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM=
|
||||
github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ=
|
||||
github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU=
|
||||
github.com/shoenig/test v0.6.4/go.mod h1:byHiCGXqrVaflBLAMq/srcZIHynQPQgeyvkvXnjqq0k=
|
||||
github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
|
||||
github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
|
||||
github.com/tinylib/msgp v1.2.5 h1:WeQg1whrXRFiZusidTQqzETkRpGjFjcIhW6uqWH09po=
|
||||
github.com/tinylib/msgp v1.2.5/go.mod h1:ykjzy2wzgrlvpDCRc4LA8UXy6D8bzMSuAF3WD57Gok0=
|
||||
github.com/tklauser/go-sysconf v0.3.15 h1:VE89k0criAymJ/Os65CSn1IXaol+1wrsFHEB8Ol49K4=
|
||||
github.com/tklauser/go-sysconf v0.3.15/go.mod h1:Dmjwr6tYFIseJw7a3dRLJfsHAMXZ3nEnL/aZY+0IuI4=
|
||||
github.com/tklauser/numcpus v0.10.0 h1:18njr6LDBk1zuna922MgdjQuJFjrdppsZG60sHGfjso=
|
||||
github.com/tklauser/numcpus v0.10.0/go.mod h1:BiTKazU708GQTYF4mB+cmlpT2Is1gLk7XVuEeem8LsQ=
|
||||
github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
|
||||
github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988=
|
||||
golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc=
|
||||
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8=
|
||||
golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww=
|
||||
golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610=
|
||||
golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
|
||||
golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
|
||||
google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
Reference in New Issue
Block a user