mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-09 05:36:24 +00:00
feat(scanner): implement V2 evidence preflights (#7523)
* feat(scanner): wire dirty usage producer identities Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): add segment activation preflight proof Keep scanner segment reuse behind a structured activation preflight so release evidence can prove the production gate remains disabled until every producer, generation, overflow, cold-oracle, and distributed invalidation check is satisfied. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * feat(scanner): expose distributed invalidation evidence Record an explicit distributed segment invalidation evidence summary when remote dirty usage snapshots are bound to the current activity window and the authenticated scoped ACK capability probe succeeds. Reject peer dirty usage snapshots that contradict the peer activity pending bit so scoped ACKs fail closed instead of clearing an unadvertised remote mutation. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
@@ -97,9 +97,9 @@ pub use scanner::{
|
||||
pub use scanner_io::{
|
||||
ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState,
|
||||
acknowledge_dirty_usage_generation, acknowledge_scoped_dirty_usage, clear_dirty_usage_bucket, record_dirty_usage_bucket,
|
||||
record_dirty_usage_bucket_from_producer, record_dirty_usage_object, record_dirty_usage_object_from_producer,
|
||||
record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state,
|
||||
scanner_maintenance_generation,
|
||||
record_dirty_usage_bucket_from_producer, record_dirty_usage_bucket_from_producers, record_dirty_usage_object,
|
||||
record_dirty_usage_object_from_producer, record_scanner_maintenance_change, scanner_activity_epoch,
|
||||
scanner_dirty_usage_snapshot, scanner_dirty_usage_state, scanner_maintenance_generation,
|
||||
};
|
||||
pub use segment_invalidation::SegmentInvalidationProducerIdentity;
|
||||
pub use sleeper::{DynamicSleeper, SCANNER_IDLE_MODE, SCANNER_SLEEPER};
|
||||
|
||||
@@ -98,6 +98,24 @@ const METRIC_SCANNER_SET_SCANS_QUEUED: &str = "rustfs_scanner_set_scans_queued";
|
||||
const METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE: &str = "rustfs_scanner_disk_bucket_scans_active";
|
||||
const METRIC_SCANNER_DISK_BUCKET_SCANS_QUEUED: &str = "rustfs_scanner_disk_bucket_scans_queued";
|
||||
|
||||
pub(crate) const SCANNER_SEGMENT_ACTIVATION_PROOF_INPUTS: [&str; 7] = [
|
||||
"source",
|
||||
"bucket_incarnation",
|
||||
"key_format",
|
||||
"baseline_scan_plan_digest",
|
||||
"process_epoch",
|
||||
"generation_window",
|
||||
"producer_identities",
|
||||
];
|
||||
pub(crate) const SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS: [&str; 6] = [
|
||||
"missing_producer_identity",
|
||||
"restart_gap",
|
||||
"generation_gap",
|
||||
"overflow",
|
||||
"missing_cold_zero_walk_oracle",
|
||||
"distributed_without_peer_invalidation",
|
||||
];
|
||||
|
||||
pub type DirtyUsageBuckets = HashMap<String, u64>;
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
@@ -163,11 +181,51 @@ struct ScannerPeerDirtyUsageExpectation {
|
||||
struct VerifiedRemoteDirtyUsage {
|
||||
dirty_buckets: HashSet<String>,
|
||||
acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
|
||||
peer_count: usize,
|
||||
dirty_peer_count: usize,
|
||||
}
|
||||
|
||||
struct ScannerBucketScopeResolutionResult {
|
||||
scope: ScannerBucketScanScope,
|
||||
remote_dirty_usage_acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
|
||||
distributed_segment_invalidation_evidence: Option<DistributedSegmentInvalidationEvidence>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub(crate) struct DistributedSegmentInvalidationEvidence {
|
||||
pub(crate) invalidation_domain: crate::segment_invalidation::SegmentInvalidationDomain,
|
||||
pub(crate) distributed_ec_invalidation: bool,
|
||||
pub(crate) peer_count: usize,
|
||||
pub(crate) dirty_peer_count: usize,
|
||||
pub(crate) same_window_remote_proof: bool,
|
||||
pub(crate) all_peers_bound_to_generation_window: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
|
||||
pub(crate) struct ScannerSegmentReuseActivationProof {
|
||||
pub(crate) production_activation: bool,
|
||||
pub(crate) durable_producer_identity: bool,
|
||||
pub(crate) restart_gap_absent: bool,
|
||||
pub(crate) generation_window_bound: bool,
|
||||
pub(crate) overflow_absent: bool,
|
||||
pub(crate) cold_zero_walk_oracle: bool,
|
||||
pub(crate) distributed_peer_invalidation: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub(crate) struct ScannerSegmentReuseActivationPreflight {
|
||||
pub(crate) production_activation: bool,
|
||||
pub(crate) scanner_segment_reuse_activated: bool,
|
||||
pub(crate) proof_inputs: &'static [&'static str],
|
||||
pub(crate) fail_closed_checks: &'static [&'static str],
|
||||
pub(crate) fail_closed_blockers: [Option<&'static str>; 6],
|
||||
}
|
||||
|
||||
impl ScannerSegmentReuseActivationPreflight {
|
||||
#[cfg(test)]
|
||||
pub(crate) fn fail_closed_blockers(&self) -> impl Iterator<Item = &'static str> + '_ {
|
||||
self.fail_closed_blockers.iter().filter_map(|blocker| *blocker)
|
||||
}
|
||||
}
|
||||
|
||||
fn verified_remote_dirty_usage(
|
||||
@@ -191,6 +249,7 @@ fn verified_remote_dirty_usage(
|
||||
|| !snapshot.complete
|
||||
|| snapshot.pending_bucket_count != u64::try_from(snapshot.buckets.len()).unwrap_or(u64::MAX)
|
||||
|| (expected.pending && snapshot.pending_bucket_count == 0)
|
||||
|| (!expected.pending && snapshot.pending_bucket_count != 0)
|
||||
{
|
||||
return None;
|
||||
}
|
||||
@@ -216,9 +275,13 @@ fn verified_remote_dirty_usage(
|
||||
}
|
||||
}
|
||||
|
||||
let peer_count = received_peers.len();
|
||||
let dirty_peer_count = acknowledgements.len();
|
||||
(received_peers.len() == expected_peers.len()).then_some(VerifiedRemoteDirtyUsage {
|
||||
dirty_buckets,
|
||||
acknowledgements,
|
||||
peer_count,
|
||||
dirty_peer_count,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -244,8 +307,11 @@ fn resolve_remote_dirty_usage_scope(
|
||||
let default_result = |scope: ScannerBucketScanScope| ScannerBucketScopeResolutionResult {
|
||||
scope,
|
||||
remote_dirty_usage_acknowledgements: Vec::new(),
|
||||
distributed_segment_invalidation_evidence: None,
|
||||
};
|
||||
|
||||
let peer_count = remote_dirty_usage.peer_count;
|
||||
let dirty_peer_count = remote_dirty_usage.dirty_peer_count;
|
||||
dirty_buckets.extend(remote_dirty_usage.dirty_buckets);
|
||||
// Peer snapshots contribute bucket names only; the local prefix scopes
|
||||
// would narrow a bucket a peer dirtied elsewhere, so the merged scope
|
||||
@@ -283,10 +349,21 @@ fn resolve_remote_dirty_usage_scope(
|
||||
if scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(&scoped_acknowledgements) {
|
||||
return default_result(ScannerBucketScanScope::default());
|
||||
}
|
||||
let has_scoped_acknowledgements = !scoped_acknowledgements.is_empty();
|
||||
|
||||
ScannerBucketScopeResolutionResult {
|
||||
scope,
|
||||
remote_dirty_usage_acknowledgements: scoped_acknowledgements,
|
||||
distributed_segment_invalidation_evidence: (dirty_peer_count > 0 && has_scoped_acknowledgements).then_some(
|
||||
DistributedSegmentInvalidationEvidence {
|
||||
invalidation_domain: crate::segment_invalidation::SegmentInvalidationDomain::DistributedEc,
|
||||
distributed_ec_invalidation: true,
|
||||
peer_count,
|
||||
dirty_peer_count,
|
||||
same_window_remote_proof: true,
|
||||
all_peers_bound_to_generation_window: true,
|
||||
},
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -406,10 +483,39 @@ fn scoped_scan_scope_from_dirty_buckets(
|
||||
ScannerBucketScanScope::from_dirty_buckets(selected_buckets, selected_bucket_prefixes, baseline_scan_plan_digest)
|
||||
}
|
||||
|
||||
fn scanner_segment_reuse_activated() -> bool {
|
||||
fn scanner_segment_reuse_activation_preflight() -> ScannerSegmentReuseActivationPreflight {
|
||||
// Production segment reuse stays disabled until a durable mutation-stream
|
||||
// proof satisfies the segment invalidation contract.
|
||||
false
|
||||
scanner_segment_reuse_activation_preflight_from_proof(ScannerSegmentReuseActivationProof::default())
|
||||
}
|
||||
|
||||
fn scanner_segment_reuse_activation_preflight_from_proof(
|
||||
proof: ScannerSegmentReuseActivationProof,
|
||||
) -> ScannerSegmentReuseActivationPreflight {
|
||||
ScannerSegmentReuseActivationPreflight {
|
||||
production_activation: proof.production_activation,
|
||||
scanner_segment_reuse_activated: proof.production_activation
|
||||
&& proof.durable_producer_identity
|
||||
&& proof.restart_gap_absent
|
||||
&& proof.generation_window_bound
|
||||
&& proof.overflow_absent
|
||||
&& proof.cold_zero_walk_oracle
|
||||
&& proof.distributed_peer_invalidation,
|
||||
proof_inputs: &SCANNER_SEGMENT_ACTIVATION_PROOF_INPUTS,
|
||||
fail_closed_checks: &SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS,
|
||||
fail_closed_blockers: [
|
||||
(!proof.durable_producer_identity).then_some("missing_producer_identity"),
|
||||
(!proof.restart_gap_absent).then_some("restart_gap"),
|
||||
(!proof.generation_window_bound).then_some("generation_gap"),
|
||||
(!proof.overflow_absent).then_some("overflow"),
|
||||
(!proof.cold_zero_walk_oracle).then_some("missing_cold_zero_walk_oracle"),
|
||||
(!proof.distributed_peer_invalidation).then_some("distributed_without_peer_invalidation"),
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
fn scanner_segment_reuse_activated() -> bool {
|
||||
scanner_segment_reuse_activation_preflight().scanner_segment_reuse_activated
|
||||
}
|
||||
|
||||
pub(crate) fn is_scanner_metadata_corrupt_error(err: &StorageError) -> bool {
|
||||
@@ -1056,6 +1162,7 @@ pub(crate) struct ScannerCycleResult {
|
||||
observational_snapshot_published: bool,
|
||||
dirty_usage_clear: Option<DirtyUsageBuckets>,
|
||||
remote_dirty_usage_acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
|
||||
distributed_segment_invalidation_evidence: Option<DistributedSegmentInvalidationEvidence>,
|
||||
remote_publication_lease_targets: Vec<(String, String, u64)>,
|
||||
failed_dirty_usage: bool,
|
||||
pending_maintenance_work: bool,
|
||||
@@ -1072,6 +1179,7 @@ impl ScannerCycleResult {
|
||||
observational_snapshot_published: false,
|
||||
dirty_usage_clear,
|
||||
remote_dirty_usage_acknowledgements: Vec::new(),
|
||||
distributed_segment_invalidation_evidence: None,
|
||||
remote_publication_lease_targets: Vec::new(),
|
||||
failed_dirty_usage: false,
|
||||
pending_maintenance_work: false,
|
||||
@@ -1137,6 +1245,15 @@ impl ScannerCycleResult {
|
||||
self
|
||||
}
|
||||
|
||||
fn with_distributed_segment_invalidation_evidence(
|
||||
mut self,
|
||||
evidence: Option<DistributedSegmentInvalidationEvidence>,
|
||||
) -> Self {
|
||||
self.publication_expectation = None;
|
||||
self.distributed_segment_invalidation_evidence = evidence;
|
||||
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;
|
||||
@@ -1224,9 +1341,9 @@ pub(crate) use cache::{
|
||||
pub use dirty_usage::{
|
||||
ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState,
|
||||
acknowledge_dirty_usage_generation, acknowledge_scoped_dirty_usage, clear_dirty_usage_bucket, record_dirty_usage_bucket,
|
||||
record_dirty_usage_bucket_from_producer, record_dirty_usage_object, record_dirty_usage_object_from_producer,
|
||||
record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state,
|
||||
scanner_maintenance_generation,
|
||||
record_dirty_usage_bucket_from_producer, record_dirty_usage_bucket_from_producers, record_dirty_usage_object,
|
||||
record_dirty_usage_object_from_producer, record_scanner_maintenance_change, scanner_activity_epoch,
|
||||
scanner_dirty_usage_snapshot, scanner_dirty_usage_state, scanner_maintenance_generation,
|
||||
};
|
||||
#[cfg(test)]
|
||||
pub(crate) use dirty_usage::{clear_dirty_usage_buckets_for_tests, dirty_usage_buckets_for_tests};
|
||||
|
||||
@@ -226,12 +226,21 @@ mod scoped_dirty_usage_tests {
|
||||
record_dirty_usage_object_from_producer("photos", "hot/object", SegmentInvalidationProducerIdentity::PutObject);
|
||||
record_dirty_usage_object_from_producer("photos", "archive/object", SegmentInvalidationProducerIdentity::DeleteObject);
|
||||
record_dirty_usage_bucket_from_producer("photos", SegmentInvalidationProducerIdentity::Unknown);
|
||||
record_dirty_usage_bucket_from_producers(
|
||||
"photos",
|
||||
[
|
||||
SegmentInvalidationProducerIdentity::DeleteMarker,
|
||||
SegmentInvalidationProducerIdentity::AbortMultipartUpload,
|
||||
],
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
dirty_usage_producer_identities_for_tests(),
|
||||
BTreeSet::from([
|
||||
SegmentInvalidationProducerIdentity::PutObject,
|
||||
SegmentInvalidationProducerIdentity::DeleteObject
|
||||
SegmentInvalidationProducerIdentity::DeleteObject,
|
||||
SegmentInvalidationProducerIdentity::DeleteMarker,
|
||||
SegmentInvalidationProducerIdentity::AbortMultipartUpload
|
||||
])
|
||||
);
|
||||
assert_eq!(
|
||||
@@ -292,6 +301,18 @@ pub fn record_dirty_usage_bucket_from_producer(
|
||||
record_dirty_usage_bucket_inner(bucket);
|
||||
}
|
||||
|
||||
pub fn record_dirty_usage_bucket_from_producers<I>(bucket: &str, producers: I)
|
||||
where
|
||||
I: IntoIterator<Item = crate::segment_invalidation::SegmentInvalidationProducerIdentity>,
|
||||
{
|
||||
if bucket.is_empty() {
|
||||
return;
|
||||
}
|
||||
|
||||
record_segment_invalidation_producer_identities(producers);
|
||||
record_dirty_usage_bucket_inner(bucket);
|
||||
}
|
||||
|
||||
fn record_dirty_usage_bucket_inner(bucket: &str) {
|
||||
let pending_buckets = {
|
||||
let mut dirty_buckets = dirty_usage_buckets();
|
||||
@@ -367,8 +388,18 @@ fn record_dirty_usage_object_inner(bucket: &str, object: &str) {
|
||||
}
|
||||
|
||||
fn record_segment_invalidation_producer_identity(producer: crate::segment_invalidation::SegmentInvalidationProducerIdentity) {
|
||||
if producer.producer().is_some() {
|
||||
dirty_usage_producer_identities().insert(producer);
|
||||
record_segment_invalidation_producer_identities([producer]);
|
||||
}
|
||||
|
||||
fn record_segment_invalidation_producer_identities<I>(producers: I)
|
||||
where
|
||||
I: IntoIterator<Item = crate::segment_invalidation::SegmentInvalidationProducerIdentity>,
|
||||
{
|
||||
let mut identities = dirty_usage_producer_identities();
|
||||
for producer in producers {
|
||||
if producer.producer().is_some() {
|
||||
identities.insert(producer);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -122,6 +122,7 @@ where
|
||||
let default_result = |scope: ScannerBucketScanScope| ScannerBucketScopeResolutionResult {
|
||||
scope,
|
||||
remote_dirty_usage_acknowledgements: Vec::new(),
|
||||
distributed_segment_invalidation_evidence: None,
|
||||
};
|
||||
if resolution.requires_full_scan {
|
||||
return default_result(ScannerBucketScanScope::default());
|
||||
@@ -408,6 +409,7 @@ where
|
||||
)
|
||||
.await;
|
||||
let remote_dirty_usage_acknowledgements = scope_resolution.remote_dirty_usage_acknowledgements;
|
||||
let distributed_segment_invalidation_evidence = scope_resolution.distributed_segment_invalidation_evidence;
|
||||
let scan_scope = scope_resolution.scope;
|
||||
#[cfg(test)]
|
||||
if let Some(observer) = resolved_scope_observer {
|
||||
@@ -783,6 +785,7 @@ where
|
||||
.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_distributed_segment_invalidation_evidence(distributed_segment_invalidation_evidence)
|
||||
.with_failed_dirty_usage(!failed_buckets.is_empty())
|
||||
.with_pending_maintenance_work(pending_maintenance_work)
|
||||
.with_required_cycle_floor(required_cycle_floor)
|
||||
|
||||
@@ -85,6 +85,78 @@ fn scanner_activity_preflight_defers_a_temporarily_offline_peer() {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_segment_reuse_activation_preflight_reports_release_gate_inputs() {
|
||||
let preflight = scanner_segment_reuse_activation_preflight();
|
||||
|
||||
assert!(!preflight.production_activation);
|
||||
assert!(!preflight.scanner_segment_reuse_activated);
|
||||
assert!(!scanner_segment_reuse_activated());
|
||||
assert_eq!(preflight.proof_inputs, SCANNER_SEGMENT_ACTIVATION_PROOF_INPUTS);
|
||||
assert_eq!(preflight.fail_closed_checks, SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS);
|
||||
assert_eq!(
|
||||
preflight.fail_closed_blockers().collect::<Vec<_>>(),
|
||||
SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_segment_reuse_activation_requires_every_preflight_proof() {
|
||||
let complete_proof = ScannerSegmentReuseActivationProof {
|
||||
production_activation: true,
|
||||
durable_producer_identity: true,
|
||||
restart_gap_absent: true,
|
||||
generation_window_bound: true,
|
||||
overflow_absent: true,
|
||||
cold_zero_walk_oracle: true,
|
||||
distributed_peer_invalidation: true,
|
||||
};
|
||||
|
||||
let mut production_disabled = complete_proof;
|
||||
production_disabled.production_activation = false;
|
||||
let preflight = scanner_segment_reuse_activation_preflight_from_proof(production_disabled);
|
||||
assert!(!preflight.production_activation);
|
||||
assert!(!preflight.scanner_segment_reuse_activated);
|
||||
assert_eq!(preflight.fail_closed_blockers().collect::<Vec<_>>(), Vec::<&str>::new());
|
||||
|
||||
let preflight = scanner_segment_reuse_activation_preflight_from_proof(complete_proof);
|
||||
assert!(preflight.production_activation);
|
||||
assert!(preflight.scanner_segment_reuse_activated);
|
||||
assert_eq!(preflight.fail_closed_blockers().collect::<Vec<_>>(), Vec::<&str>::new());
|
||||
|
||||
let mut missing_identity = complete_proof;
|
||||
missing_identity.durable_producer_identity = false;
|
||||
assert_segment_reuse_activation_blocked_by(missing_identity, "missing_producer_identity");
|
||||
|
||||
let mut restart_gap = complete_proof;
|
||||
restart_gap.restart_gap_absent = false;
|
||||
assert_segment_reuse_activation_blocked_by(restart_gap, "restart_gap");
|
||||
|
||||
let mut generation_gap = complete_proof;
|
||||
generation_gap.generation_window_bound = false;
|
||||
assert_segment_reuse_activation_blocked_by(generation_gap, "generation_gap");
|
||||
|
||||
let mut overflow = complete_proof;
|
||||
overflow.overflow_absent = false;
|
||||
assert_segment_reuse_activation_blocked_by(overflow, "overflow");
|
||||
|
||||
let mut missing_cold_oracle = complete_proof;
|
||||
missing_cold_oracle.cold_zero_walk_oracle = false;
|
||||
assert_segment_reuse_activation_blocked_by(missing_cold_oracle, "missing_cold_zero_walk_oracle");
|
||||
|
||||
let mut missing_distributed_invalidation = complete_proof;
|
||||
missing_distributed_invalidation.distributed_peer_invalidation = false;
|
||||
assert_segment_reuse_activation_blocked_by(missing_distributed_invalidation, "distributed_without_peer_invalidation");
|
||||
}
|
||||
|
||||
fn assert_segment_reuse_activation_blocked_by(proof: ScannerSegmentReuseActivationProof, blocker: &'static str) {
|
||||
let preflight = scanner_segment_reuse_activation_preflight_from_proof(proof);
|
||||
|
||||
assert!(preflight.production_activation);
|
||||
assert!(!preflight.scanner_segment_reuse_activated);
|
||||
assert_eq!(preflight.fail_closed_blockers().collect::<Vec<_>>(), vec![blocker]);
|
||||
}
|
||||
|
||||
async fn setup_two_pool_scanner_store() -> (tempfile::TempDir, Arc<ECStore>) {
|
||||
init_ecstore_config_for_scanner_tests();
|
||||
let temp_dir = tempfile::tempdir().expect("multi-pool scanner test directory should be created");
|
||||
@@ -1781,6 +1853,18 @@ fn remote_dirty_usage_invalidates_local_prefix_hints_until_distributed_proof_exi
|
||||
"peer dirty state is not a distributed segment invalidation proof"
|
||||
);
|
||||
assert_eq!(distributed.remote_dirty_usage_acknowledgements.len(), 1);
|
||||
let evidence = distributed
|
||||
.distributed_segment_invalidation_evidence
|
||||
.expect("same-window peer snapshot and scoped ACK capability form distributed evidence");
|
||||
assert_eq!(evidence.peer_count, 1);
|
||||
assert_eq!(evidence.dirty_peer_count, 1);
|
||||
assert_eq!(
|
||||
evidence.invalidation_domain,
|
||||
crate::segment_invalidation::SegmentInvalidationDomain::DistributedEc
|
||||
);
|
||||
assert!(evidence.distributed_ec_invalidation);
|
||||
assert!(evidence.same_window_remote_proof);
|
||||
assert!(evidence.all_peers_bound_to_generation_window);
|
||||
}
|
||||
|
||||
fn peer_dirty_usage_snapshot(
|
||||
@@ -1827,7 +1911,7 @@ fn verified_remote_dirty_usage_buckets_merges_only_complete_current_snapshots()
|
||||
ScannerPeerDirtyUsageExpectation {
|
||||
instance_id: "instance-b".to_string(),
|
||||
generation: 3,
|
||||
pending: false,
|
||||
pending: true,
|
||||
},
|
||||
),
|
||||
]);
|
||||
@@ -1874,10 +1958,36 @@ fn verified_remote_dirty_usage_buckets_merges_only_complete_current_snapshots()
|
||||
},
|
||||
},
|
||||
],
|
||||
peer_count: 2,
|
||||
dirty_peer_count: 2,
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn verified_remote_dirty_usage_rejects_peer_snapshot_that_contradicts_activity_pending_state() {
|
||||
let expected_peers = HashMap::from([(
|
||||
"node-a:9000".to_string(),
|
||||
ScannerPeerDirtyUsageExpectation {
|
||||
instance_id: "instance-a".to_string(),
|
||||
generation: 7,
|
||||
pending: false,
|
||||
},
|
||||
)]);
|
||||
|
||||
assert!(
|
||||
verified_remote_dirty_usage(
|
||||
&expected_peers,
|
||||
vec![(
|
||||
"node-a:9000".to_string(),
|
||||
peer_dirty_usage_snapshot("instance-a", 7, true, &[("photos", 7)]),
|
||||
)],
|
||||
)
|
||||
.is_none(),
|
||||
"a clean activity window cannot authorize a dirty peer snapshot or scoped ACK"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_scoped_dirty_usage_ack_cost_threshold_is_single_protocol_batch() {
|
||||
let acknowledgement = |entry_count: usize| crate::scanner::ScannerDirtyUsageAcknowledgement {
|
||||
@@ -1951,6 +2061,7 @@ fn remote_dirty_usage_scope_resolution_falls_back_when_ack_batch_exceeds_thresho
|
||||
result.remote_dirty_usage_acknowledgements.is_empty(),
|
||||
"full-scan fallback must not send a scoped ACK that peers would reject or split"
|
||||
);
|
||||
assert!(result.distributed_segment_invalidation_evidence.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -2092,6 +2203,11 @@ async fn distributed_scoped_scan_falls_back_when_remote_scoped_ack_capability_is
|
||||
|
||||
assert_eq!(result.scope.selected_buckets.as_deref(), expected_buckets.as_ref());
|
||||
assert_eq!(result.remote_dirty_usage_acknowledgements.len(), expected_ack_count);
|
||||
assert_eq!(
|
||||
result.distributed_segment_invalidation_evidence.is_some(),
|
||||
capability,
|
||||
"distributed evidence requires an authenticated scoped ACK capability probe"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -57,6 +57,9 @@ pub enum SegmentInvalidationProducerIdentity {
|
||||
DeleteObject,
|
||||
DeleteMarker,
|
||||
CompleteMultipartUpload,
|
||||
AbortMultipartUpload,
|
||||
ObjectMetadata,
|
||||
BucketMetadata,
|
||||
Replication,
|
||||
TierTransition,
|
||||
TierExpiration,
|
||||
@@ -66,11 +69,14 @@ pub enum SegmentInvalidationProducerIdentity {
|
||||
}
|
||||
|
||||
impl SegmentInvalidationProducerIdentity {
|
||||
pub const REQUIRED_PRODUCTION: [Self; 8] = [
|
||||
pub const REQUIRED_PRODUCTION: [Self; 11] = [
|
||||
Self::PutObject,
|
||||
Self::DeleteObject,
|
||||
Self::DeleteMarker,
|
||||
Self::CompleteMultipartUpload,
|
||||
Self::AbortMultipartUpload,
|
||||
Self::ObjectMetadata,
|
||||
Self::BucketMetadata,
|
||||
Self::Replication,
|
||||
Self::TierTransition,
|
||||
Self::TierExpiration,
|
||||
@@ -82,7 +88,9 @@ impl SegmentInvalidationProducerIdentity {
|
||||
Self::PutObject => Some(SegmentInvalidationProducer::Put),
|
||||
Self::DeleteObject => Some(SegmentInvalidationProducer::Delete),
|
||||
Self::DeleteMarker => Some(SegmentInvalidationProducer::DeleteMarker),
|
||||
Self::CompleteMultipartUpload => Some(SegmentInvalidationProducer::Multipart),
|
||||
Self::CompleteMultipartUpload | Self::AbortMultipartUpload => Some(SegmentInvalidationProducer::Multipart),
|
||||
Self::ObjectMetadata => Some(SegmentInvalidationProducer::Put),
|
||||
Self::BucketMetadata => Some(SegmentInvalidationProducer::DirectoryObject),
|
||||
Self::Replication => Some(SegmentInvalidationProducer::Replication),
|
||||
Self::TierTransition | Self::TierExpiration => Some(SegmentInvalidationProducer::Tier),
|
||||
Self::DirectoryObject => Some(SegmentInvalidationProducer::DirectoryObject),
|
||||
@@ -402,8 +410,12 @@ mod tests {
|
||||
SegmentInvalidationProducerIdentity::DeleteObject,
|
||||
SegmentInvalidationProducerIdentity::DeleteMarker,
|
||||
SegmentInvalidationProducerIdentity::CompleteMultipartUpload,
|
||||
SegmentInvalidationProducerIdentity::AbortMultipartUpload,
|
||||
SegmentInvalidationProducerIdentity::ObjectMetadata,
|
||||
SegmentInvalidationProducerIdentity::BucketMetadata,
|
||||
SegmentInvalidationProducerIdentity::Replication,
|
||||
SegmentInvalidationProducerIdentity::TierTransition,
|
||||
SegmentInvalidationProducerIdentity::TierExpiration,
|
||||
SegmentInvalidationProducerIdentity::DirectoryObject,
|
||||
SegmentInvalidationProducerIdentity::Unknown,
|
||||
]),
|
||||
@@ -415,8 +427,12 @@ mod tests {
|
||||
SegmentInvalidationProducerIdentity::DeleteObject,
|
||||
SegmentInvalidationProducerIdentity::DeleteMarker,
|
||||
SegmentInvalidationProducerIdentity::CompleteMultipartUpload,
|
||||
SegmentInvalidationProducerIdentity::AbortMultipartUpload,
|
||||
SegmentInvalidationProducerIdentity::ObjectMetadata,
|
||||
SegmentInvalidationProducerIdentity::BucketMetadata,
|
||||
SegmentInvalidationProducerIdentity::Replication,
|
||||
SegmentInvalidationProducerIdentity::TierTransition,
|
||||
SegmentInvalidationProducerIdentity::TierExpiration,
|
||||
SegmentInvalidationProducerIdentity::DirectoryObject,
|
||||
SegmentInvalidationProducerIdentity::TestFixture,
|
||||
]),
|
||||
@@ -427,6 +443,7 @@ mod tests {
|
||||
SegmentInvalidationProducerIdentity::PutObject,
|
||||
SegmentInvalidationProducerIdentity::DeleteObject,
|
||||
SegmentInvalidationProducerIdentity::DeleteMarker,
|
||||
SegmentInvalidationProducerIdentity::CompleteMultipartUpload,
|
||||
SegmentInvalidationProducerIdentity::Replication,
|
||||
SegmentInvalidationProducerIdentity::TierTransition,
|
||||
SegmentInvalidationProducerIdentity::DirectoryObject,
|
||||
@@ -439,6 +456,9 @@ mod tests {
|
||||
SegmentInvalidationProducerIdentity::DeleteObject,
|
||||
SegmentInvalidationProducerIdentity::DeleteMarker,
|
||||
SegmentInvalidationProducerIdentity::CompleteMultipartUpload,
|
||||
SegmentInvalidationProducerIdentity::AbortMultipartUpload,
|
||||
SegmentInvalidationProducerIdentity::ObjectMetadata,
|
||||
SegmentInvalidationProducerIdentity::BucketMetadata,
|
||||
SegmentInvalidationProducerIdentity::Replication,
|
||||
SegmentInvalidationProducerIdentity::TierTransition,
|
||||
SegmentInvalidationProducerIdentity::DirectoryObject,
|
||||
|
||||
@@ -1389,7 +1389,10 @@ impl DefaultBucketUsecase {
|
||||
counter!("rustfs_create_bucket_total").increment(1);
|
||||
let result = Ok(S3Response::new(output));
|
||||
let _ = helper.complete(&result);
|
||||
rustfs_scanner::record_dirty_usage_bucket(&bucket);
|
||||
rustfs_scanner::record_dirty_usage_bucket_from_producer(
|
||||
&bucket,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata,
|
||||
);
|
||||
result
|
||||
}
|
||||
|
||||
@@ -1781,7 +1784,10 @@ impl DefaultBucketUsecase {
|
||||
warn!(bucket = %bucket, error = ?err, "site replication bucket tagging delete hook failed");
|
||||
}
|
||||
|
||||
rustfs_scanner::record_dirty_usage_bucket(&bucket);
|
||||
rustfs_scanner::record_dirty_usage_bucket_from_producer(
|
||||
&bucket,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata,
|
||||
);
|
||||
Ok(S3Response::new(DeleteBucketTaggingOutput {}))
|
||||
}
|
||||
|
||||
@@ -2698,7 +2704,10 @@ impl DefaultBucketUsecase {
|
||||
warn!(bucket = %bucket, error = ?err, "site replication bucket tagging hook failed");
|
||||
}
|
||||
|
||||
rustfs_scanner::record_dirty_usage_bucket(&bucket);
|
||||
rustfs_scanner::record_dirty_usage_bucket_from_producer(
|
||||
&bucket,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata,
|
||||
);
|
||||
Ok(S3Response::new(PutBucketTaggingOutput::default()))
|
||||
}
|
||||
|
||||
@@ -2733,7 +2742,10 @@ impl DefaultBucketUsecase {
|
||||
warn!(bucket = %bucket, error = ?err, "site replication bucket versioning hook failed");
|
||||
}
|
||||
|
||||
rustfs_scanner::record_dirty_usage_bucket(&bucket);
|
||||
rustfs_scanner::record_dirty_usage_bucket_from_producer(
|
||||
&bucket,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata,
|
||||
);
|
||||
Ok(S3Response::new(PutBucketVersioningOutput {}))
|
||||
}
|
||||
|
||||
|
||||
@@ -532,7 +532,11 @@ impl DefaultMultipartUsecase {
|
||||
.await
|
||||
{
|
||||
Ok(_) => {
|
||||
rustfs_scanner::record_dirty_usage_object(&bucket, &key);
|
||||
rustfs_scanner::record_dirty_usage_object_from_producer(
|
||||
&bucket,
|
||||
&key,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::AbortMultipartUpload,
|
||||
);
|
||||
Ok(S3Response::new(AbortMultipartUploadOutput { ..Default::default() }))
|
||||
}
|
||||
Err(err) => {
|
||||
@@ -802,7 +806,11 @@ impl DefaultMultipartUsecase {
|
||||
schedule_object_replication(obj_info.clone(), store, completion_replication_decision).await;
|
||||
}
|
||||
|
||||
rustfs_scanner::record_dirty_usage_bucket(&bucket);
|
||||
rustfs_scanner::record_dirty_usage_object_from_producer(
|
||||
&bucket,
|
||||
&key,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::CompleteMultipartUpload,
|
||||
);
|
||||
Ok::<_, S3Error>(obj_info)
|
||||
}
|
||||
});
|
||||
|
||||
@@ -797,6 +797,18 @@ impl DefaultObjectUsecase {
|
||||
let resp_elements =
|
||||
build_event_resp_elements(&S3Response::new(DeleteObjectsOutput::default()), &request_context.request_id);
|
||||
let deleted_any = delete_results.iter().any(|result| result.delete_object.is_some());
|
||||
let delete_producers = delete_results
|
||||
.iter()
|
||||
.filter_map(|result| {
|
||||
result.delete_object.as_ref().map(|deleted_object| {
|
||||
if deleted_object.delete_marker && result.requested_version_id.is_none() {
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::DeleteMarker
|
||||
} else {
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::DeleteObject
|
||||
}
|
||||
})
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
let notify_bucket = bucket.clone();
|
||||
spawn_background_with_context(Some(request_context), async move {
|
||||
let _activity_guard = DeleteTailActivityGuard::new(DeleteTailStage::Notify);
|
||||
@@ -838,7 +850,7 @@ impl DefaultObjectUsecase {
|
||||
let result = Ok(S3Response::new(output));
|
||||
let _ = helper.complete(&result);
|
||||
if deleted_any {
|
||||
rustfs_scanner::record_dirty_usage_bucket(&bucket);
|
||||
rustfs_scanner::record_dirty_usage_bucket_from_producers(&bucket, delete_producers);
|
||||
}
|
||||
// Record write operation for capacity management (inline to avoid per-request tokio::spawn overhead)
|
||||
let manager = get_capacity_manager();
|
||||
@@ -1101,7 +1113,10 @@ impl DefaultObjectUsecase {
|
||||
let manager = get_capacity_manager();
|
||||
manager.record_write_operation().await;
|
||||
let _ = helper.complete(&result);
|
||||
rustfs_scanner::record_dirty_usage_bucket(&bucket);
|
||||
rustfs_scanner::record_dirty_usage_bucket_from_producer(
|
||||
&bucket,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::DeleteObject,
|
||||
);
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
@@ -627,6 +627,7 @@ struct ExtractPreparedMember<R = ExtractStagedBody> {
|
||||
write_plan: WritePlan,
|
||||
opts: ObjectOptions,
|
||||
replication: ReplicateDecision,
|
||||
producer_identity: rustfs_scanner::SegmentInvalidationProducerIdentity,
|
||||
staging_permit: OwnedSemaphorePermit,
|
||||
member_permit: OwnedSemaphorePermit,
|
||||
}
|
||||
@@ -765,6 +766,7 @@ where
|
||||
write_plan,
|
||||
opts,
|
||||
replication,
|
||||
producer_identity,
|
||||
staging_permit,
|
||||
member_permit,
|
||||
} = member;
|
||||
@@ -793,7 +795,7 @@ where
|
||||
// the scanner before its post-store awaits, then retains the lifecycle slot
|
||||
// through quota, cache, replication, and event construction.
|
||||
if !context.wrote_any_entry.swap(true, Ordering::AcqRel) {
|
||||
rustfs_scanner::record_dirty_usage_bucket(&context.bucket);
|
||||
rustfs_scanner::record_dirty_usage_bucket_from_producer(&context.bucket, producer_identity);
|
||||
}
|
||||
let success =
|
||||
complete_extract_member_post_commit(context, key, opts, replication, obj_info, backfilled_old_current_size).await;
|
||||
@@ -2603,6 +2605,7 @@ impl DefaultObjectUsecase {
|
||||
write_plan,
|
||||
opts,
|
||||
replication,
|
||||
producer_identity: rustfs_scanner::SegmentInvalidationProducerIdentity::DirectoryObject,
|
||||
staging_permit,
|
||||
member_permit,
|
||||
},
|
||||
@@ -2627,6 +2630,7 @@ impl DefaultObjectUsecase {
|
||||
write_plan,
|
||||
opts,
|
||||
replication,
|
||||
producer_identity: rustfs_scanner::SegmentInvalidationProducerIdentity::PutObject,
|
||||
staging_permit,
|
||||
member_permit,
|
||||
},
|
||||
@@ -2656,6 +2660,11 @@ impl DefaultObjectUsecase {
|
||||
write_plan,
|
||||
opts,
|
||||
replication,
|
||||
producer_identity: if is_dir {
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::DirectoryObject
|
||||
} else {
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::PutObject
|
||||
},
|
||||
staging_permit,
|
||||
member_permit,
|
||||
});
|
||||
|
||||
@@ -733,7 +733,11 @@ impl DefaultObjectUsecase {
|
||||
)
|
||||
.await
|
||||
.map_err(ApiError::from)?;
|
||||
rustfs_scanner::record_dirty_usage_bucket(bucket);
|
||||
rustfs_scanner::record_dirty_usage_object_from_producer(
|
||||
bucket,
|
||||
key,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::AbortMultipartUpload,
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -420,7 +420,11 @@ impl DefaultObjectUsecase {
|
||||
)
|
||||
.await
|
||||
.map_err(ApiError::from)?;
|
||||
rustfs_scanner::record_dirty_usage_object(&bucket, &object);
|
||||
rustfs_scanner::record_dirty_usage_object_from_producer(
|
||||
&bucket,
|
||||
&object,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::TierTransition,
|
||||
);
|
||||
#[cfg(test)]
|
||||
maybe_pause_after_restore_status_commit(&bucket, &object).await;
|
||||
drop(superseded_worker_guard.take());
|
||||
@@ -494,7 +498,11 @@ impl DefaultObjectUsecase {
|
||||
err.to_string()
|
||||
);
|
||||
} else {
|
||||
rustfs_scanner::record_dirty_usage_object(&bucket_clone, &object_clone);
|
||||
rustfs_scanner::record_dirty_usage_object_from_producer(
|
||||
&bucket_clone,
|
||||
&object_clone,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::TierTransition,
|
||||
);
|
||||
debug!(bucket = %bucket_clone, object = %object_clone, "Transitioned object restored");
|
||||
}
|
||||
});
|
||||
|
||||
@@ -730,7 +730,11 @@ impl S3 for FS {
|
||||
|
||||
let result = Ok(S3Response::new(DeleteObjectTaggingOutput { version_id }));
|
||||
let _ = helper.complete(&result);
|
||||
rustfs_scanner::record_dirty_usage_bucket(&bucket);
|
||||
rustfs_scanner::record_dirty_usage_object_from_producer(
|
||||
&bucket,
|
||||
&object,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::ObjectMetadata,
|
||||
);
|
||||
let duration = start_time.elapsed();
|
||||
histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "delete").record(duration.as_secs_f64());
|
||||
result
|
||||
@@ -1629,7 +1633,11 @@ impl S3 for FS {
|
||||
|
||||
let result = Ok(S3Response::new(output));
|
||||
let _ = helper.complete(&result);
|
||||
rustfs_scanner::record_dirty_usage_bucket(&bucket);
|
||||
rustfs_scanner::record_dirty_usage_object_from_producer(
|
||||
&bucket,
|
||||
&key,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::ObjectMetadata,
|
||||
);
|
||||
result
|
||||
}
|
||||
|
||||
@@ -1733,7 +1741,10 @@ impl S3 for FS {
|
||||
);
|
||||
}
|
||||
|
||||
rustfs_scanner::record_dirty_usage_bucket(&bucket);
|
||||
rustfs_scanner::record_dirty_usage_bucket_from_producer(
|
||||
&bucket,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata,
|
||||
);
|
||||
Ok(S3Response::new(PutObjectLockConfigurationOutput::default()))
|
||||
}
|
||||
|
||||
@@ -1849,7 +1860,11 @@ impl S3 for FS {
|
||||
|
||||
let result = Ok(S3Response::new(output));
|
||||
let _ = helper.complete(&result);
|
||||
rustfs_scanner::record_dirty_usage_bucket(&bucket);
|
||||
rustfs_scanner::record_dirty_usage_object_from_producer(
|
||||
&bucket,
|
||||
&key,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::ObjectMetadata,
|
||||
);
|
||||
result
|
||||
}
|
||||
|
||||
@@ -1959,7 +1974,11 @@ impl S3 for FS {
|
||||
version_id: req.input.version_id.clone(),
|
||||
}));
|
||||
let _ = helper.complete(&result);
|
||||
rustfs_scanner::record_dirty_usage_bucket(&bucket);
|
||||
rustfs_scanner::record_dirty_usage_object_from_producer(
|
||||
&bucket,
|
||||
&object,
|
||||
rustfs_scanner::SegmentInvalidationProducerIdentity::ObjectMetadata,
|
||||
);
|
||||
let duration = start_time.elapsed();
|
||||
histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "put").record(duration.as_secs_f64());
|
||||
result
|
||||
|
||||
Reference in New Issue
Block a user