gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-09 21:56:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): bind segment proof to typed dirty suffix (#7564)

Browse Source 
Track producer identity completeness per pending bucket and retain process coverage across durable acknowledgements. Admit a prior same-process segment proof only for an exact or strictly earlier fully typed generation window.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-09 14:19:09 +08:00
committed by GitHub
parent 14b3cffe8f
commit 6b802a91eb
5 changed files with 229 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/scanner_io.rs
+40 -18
View File
@@ -39,7 +39,9 @@ use s3s::dto::{
BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration, VersioningConfiguration,
};
use sha2::{Digest as _, Sha256};
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
#[cfg(test)]
use std::collections::BTreeSet;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::future::Future;
use std::path::Path;
use std::pin::Pin;
@@ -451,43 +453,46 @@ fn scanner_segment_reuse_baseline_producer_evidence(
let Ok(authoritative) = serde_json::from_slice::<DataUsageInfo>(authoritative_data) else {
return (evidence, false);
};
if complete_scanner_cache_snapshot_plan_digest(&authoritative, baseline_proof, true).is_none()
|| !scanner_snapshot_set_states_have_segment_reuse_activation_proof(
&authoritative,
&evidence,
baseline_proof.expected_sources,
)
{
if complete_scanner_cache_snapshot_plan_digest(&authoritative, baseline_proof, true).is_none() {
return (evidence, false);
}
let Some(cold_zero_walk_oracle) =
scanner_snapshot_set_states_segment_reuse_proof(&authoritative, &evidence, baseline_proof.expected_sources)
else {
return (evidence, false);
};
evidence.durable_producer_identity = true;
evidence.restart_gap_absent = true;
(evidence, true)
(evidence, cold_zero_walk_oracle)
}
fn scanner_snapshot_set_states_have_segment_reuse_activation_proof(
fn scanner_snapshot_set_states_segment_reuse_proof(
snapshot: &DataUsageInfo,
evidence: &DirtyUsageProducerEvidence,
expected_sources: &HashSet<DataUsageCacheSource>,
) -> bool {
) -> Option<bool> {
let mut covered_sources = HashSet::with_capacity(expected_sources.len());
let mut cold_zero_walk_oracle = true;
let all_sets_proved = snapshot.usage_snapshot_set_states.iter().all(|state| {
let (Ok(pool_index), Ok(set_index)) = (usize::try_from(state.pool_index), usize::try_from(state.set_index)) else {
return false;
};
let source = DataUsageCacheSource::new(pool_index, set_index);
state.complete
let proved = state.complete
&& !state.tombstone
&& expected_sources.contains(&source)
&& covered_sources.insert(source)
&& scanner_segment_invalidation_proof_matches(state.segment_invalidation_proof.as_ref(), evidence)
&& state
&& scanner_segment_invalidation_baseline_proof_matches(state.segment_invalidation_proof.as_ref(), evidence);
if proved {
cold_zero_walk_oracle &= state
.segment_invalidation_proof
.as_ref()
.is_some_and(|proof| proof.cold_zero_walk_oracle)
.is_some_and(|proof| proof.cold_zero_walk_oracle);
}
proved
});
all_sets_proved && covered_sources.len() == expected_sources.len()
(all_sets_proved && covered_sources.len() == expected_sources.len()).then_some(cold_zero_walk_oracle)
}
fn scoped_scan_scope_from_dirty_buckets(
@@ -664,13 +669,30 @@ fn scanner_segment_invalidation_proof_matches(
evidence: &DirtyUsageProducerEvidence,
) -> bool {
proof.is_some_and(|proof| {
proof.process_epoch == scanner_activity_epoch()
scanner_segment_invalidation_proof_is_well_formed(proof)
&& proof.generation_start == evidence.generation_start
&& proof.generation_end == evidence.generation_end
&& proof.producer_identity_coverage_complete
})
}
fn scanner_segment_invalidation_baseline_proof_matches(
proof: Option<&crate::DataUsageSegmentInvalidationProof>,
evidence: &DirtyUsageProducerEvidence,
) -> bool {
proof.is_some_and(|proof| {
scanner_segment_invalidation_proof_is_well_formed(proof)
&& ((proof.generation_start == evidence.generation_start && proof.generation_end == evidence.generation_end)
|| proof.generation_end < evidence.generation_start)
})
}
fn scanner_segment_invalidation_proof_is_well_formed(proof: &crate::DataUsageSegmentInvalidationProof) -> bool {
proof.process_epoch == scanner_activity_epoch()
&& proof.generation_start != 0
&& proof.generation_end >= proof.generation_start
&& proof.producer_identity_coverage_complete
}
fn scanner_completed_set_segment_invalidation_proof(
proof: &Option<crate::DataUsageSegmentInvalidationProof>,
cold_zero_walk_reuse_candidate: bool,
crates/scanner/src/scanner_io/dirty_usage.rs
+85 -28
View File
@@ -22,10 +22,11 @@ pub(super) static DIRTY_USAGE_BUCKETS: LazyLock<StdMutex<DirtyUsageBuckets>> = L
// observe a bucket generation without its matching scope and producer evidence.
pub(super) static DIRTY_USAGE_BUCKET_SCOPES: LazyLock<StdMutex<DirtyUsageBucketScopes>> =
LazyLock::new(|| StdMutex::new(HashMap::new()));
// Non-authoritative process-local producer coverage. Any future segment reuse
// activation must bind this to the exact generation window and durable proof.
// Non-authoritative process-local producer coverage. Segment reuse binds this
// per-bucket suffix to an earlier durable proof from the same process epoch.
pub(super) static DIRTY_USAGE_PRODUCER_IDENTITIES: LazyLock<StdMutex<DirtyUsageProducerIdentities>> =
LazyLock::new(|| StdMutex::new(BTreeMap::new()));
pub(super) static DIRTY_USAGE_PRODUCER_COVERAGE: AtomicU64 = AtomicU64::new(0);
pub(super) static DIRTY_USAGE_BUCKET_NOTIFY: LazyLock<Notify> = LazyLock::new(Notify::new);
pub(super) static SCANNER_ACTIVITY_EPOCH: LazyLock<String> = LazyLock::new(|| format!("{:032x}", rand::random::<u128>()));
pub(super) static SCANNER_MAINTENANCE_GENERATION: AtomicU64 = AtomicU64::new(0);
@@ -56,16 +57,16 @@ pub(super) enum DirtyUsageBucketScope {
pub(super) type DirtyUsageBucketScopes = HashMap<String, DirtyUsageBucketScope>;
const MAX_DIRTY_USAGE_TOP_LEVEL_ENTRIES_PER_BUCKET: usize = 128;
pub(super) const MAX_DIRTY_USAGE_TOP_LEVEL_ENTRIES_PER_BUCKET: usize = 128;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct DirtyUsageProducerIdentityState {
first_generation: u64,
last_generation: u64,
fully_identified: bool,
}
pub(super) type DirtyUsageProducerIdentities =
BTreeMap<crate::segment_invalidation::SegmentInvalidationProducerIdentity, DirtyUsageProducerIdentityState>;
pub(super) type DirtyUsageProducerIdentities = BTreeMap<String, DirtyUsageProducerIdentityState>;
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) struct DirtyUsageProducerEvidence {
@@ -126,6 +127,7 @@ pub fn acknowledge_scoped_dirty_usage(
let (cleared, pending) = {
let mut dirty = dirty_usage_buckets();
let mut dirty_scopes = dirty_usage_bucket_scopes();
let mut producer_identities = dirty_usage_producer_identities();
let checked = entries
.iter()
.map(|(guard, generation)| {
@@ -145,6 +147,7 @@ pub fn acknowledge_scoped_dirty_usage(
probe_only,
)?;
if cleared > 0 {
producer_identities.retain(|bucket, _| dirty.contains_key(bucket));
advance_generation(&DIRTY_USAGE_BUCKET_GENERATION);
}
(cleared, dirty.len())
@@ -191,6 +194,7 @@ fn apply_scoped_dirty_usage_ack(
mod scoped_dirty_usage_tests {
use super::*;
use crate::segment_invalidation::SegmentInvalidationProducerIdentity;
use serial_test::serial;
#[test]
fn scoped_dirty_usage_preserves_uncovered_newer_and_replayed_generations() {
@@ -253,6 +257,7 @@ mod scoped_dirty_usage_tests {
}
#[test]
#[serial]
fn dirty_usage_tracks_known_segment_producer_identities_without_authorizing_unknown_sources() {
clear_dirty_usage_buckets_for_tests();
record_dirty_usage_object_from_producer("photos", "hot/object", SegmentInvalidationProducerIdentity::PutObject);
@@ -280,6 +285,19 @@ mod scoped_dirty_usage_tests {
Some(&DirtyUsageBucketScope::WholeBucket),
"an unknown producer keeps the bucket dirty but must not count as producer coverage"
);
let snapshot = snapshot_dirty_usage_buckets(
&[BucketInfo {
name: "photos".to_string(),
created: None,
deleted: None,
versioning: false,
object_locking: false,
}],
dirty_usage_generation(),
);
let evidence = dirty_usage_producer_evidence(&snapshot);
assert!(!evidence.producer_identity_coverage_complete);
assert!(!evidence.generation_window_bound);
clear_dirty_usage_buckets_for_tests();
assert!(dirty_usage_producer_identities_for_tests().is_empty());
@@ -353,7 +371,7 @@ where
let generation = advance_generation(&DIRTY_USAGE_BUCKET_GENERATION);
dirty_buckets.insert(bucket.to_string(), generation);
dirty_scopes.insert(bucket.to_string(), DirtyUsageBucketScope::WholeBucket);
record_segment_invalidation_producer_identities_for_generation(&mut producer_identities, generation, producers);
record_segment_invalidation_producer_identities_for_generation(&mut producer_identities, bucket, generation, producers);
dirty_buckets.len()
};
global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets));
@@ -418,7 +436,7 @@ where
if overflowed {
*scope = DirtyUsageBucketScope::WholeBucket;
}
record_segment_invalidation_producer_identities_for_generation(&mut producer_identities, generation, producers);
record_segment_invalidation_producer_identities_for_generation(&mut producer_identities, bucket, generation, producers);
dirty_buckets.len()
};
global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets));
@@ -428,27 +446,46 @@ where
fn record_segment_invalidation_producer_identities_for_generation<I>(
identities: &mut DirtyUsageProducerIdentities,
bucket: &str,
generation: u64,
producers: I,
) where
I: IntoIterator<Item = crate::segment_invalidation::SegmentInvalidationProducerIdentity>,
{
let mut event_coverage = 0_u64;
let mut fully_identified = true;
for producer in producers {
if producer.producer().is_some() {
identities
.entry(producer)
.and_modify(|state| state.last_generation = state.last_generation.max(generation))
.or_insert(DirtyUsageProducerIdentityState {
first_generation: generation,
last_generation: generation,
});
if let Some(coverage_bit) = producer.production_coverage_bit() {
event_coverage |= coverage_bit;
} else {
fully_identified = false;
}
}
fully_identified &= event_coverage != 0;
DIRTY_USAGE_PRODUCER_COVERAGE.fetch_or(event_coverage, Ordering::AcqRel);
if let Some(state) = identities.get_mut(bucket) {
state.last_generation = state.last_generation.max(generation);
state.fully_identified &= fully_identified;
} else {
identities.insert(
bucket.to_string(),
DirtyUsageProducerIdentityState {
first_generation: generation,
last_generation: generation,
fully_identified,
},
);
}
}
#[cfg(test)]
fn dirty_usage_producer_identities_for_tests() -> BTreeSet<crate::segment_invalidation::SegmentInvalidationProducerIdentity> {
dirty_usage_producer_identities().keys().copied().collect()
let coverage = DIRTY_USAGE_PRODUCER_COVERAGE.load(Ordering::Acquire);
crate::segment_invalidation::SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION
.into_iter()
.filter(|identity| identity.production_coverage_bit().is_some_and(|bit| coverage & bit != 0))
.collect()
}
fn dirty_usage_top_level_entry(object: &str) -> Option<String> {
@@ -531,6 +568,7 @@ pub fn acknowledge_dirty_usage_generation(
let (cleared_buckets, pending_buckets) = {
let mut dirty_buckets = dirty_usage_buckets();
let mut dirty_scopes = dirty_usage_bucket_scopes();
let mut producer_identities = dirty_usage_producer_identities();
let current_generation = DIRTY_USAGE_BUCKET_GENERATION.load(Ordering::Acquire);
if generation == 0 || generation == u64::MAX || current_generation == u64::MAX || generation > current_generation {
return Err(ScannerDirtyUsageAckError::InvalidGeneration);
@@ -539,6 +577,7 @@ pub fn acknowledge_dirty_usage_generation(
let before = dirty_buckets.len();
dirty_buckets.retain(|_, dirty_generation| *dirty_generation > generation);
dirty_scopes.retain(|bucket, _| dirty_buckets.contains_key(bucket));
producer_identities.retain(|bucket, _| dirty_buckets.contains_key(bucket));
let cleared_buckets = before.saturating_sub(dirty_buckets.len());
if cleared_buckets > 0 {
advance_generation(&DIRTY_USAGE_BUCKET_GENERATION);
@@ -562,8 +601,11 @@ pub fn clear_dirty_usage_bucket(bucket: &str) {
let pending_buckets = {
let mut dirty_buckets = dirty_usage_buckets();
let mut dirty_scopes = dirty_usage_bucket_scopes();
let mut producer_identities = dirty_usage_producer_identities();
dirty_buckets.remove(bucket);
dirty_scopes.remove(bucket);
producer_identities.remove(bucket);
DIRTY_USAGE_PRODUCER_COVERAGE.store(0, Ordering::Release);
advance_generation(&DIRTY_USAGE_BUCKET_GENERATION);
dirty_buckets.len()
};
@@ -625,6 +667,7 @@ pub(super) fn clear_dirty_usage_buckets(snapshot: &DirtyUsageBuckets) {
let (cleared_buckets, pending_buckets) = {
let mut dirty_buckets = dirty_usage_buckets();
let mut dirty_scopes = dirty_usage_bucket_scopes();
let mut producer_identities = dirty_usage_producer_identities();
let mut cleared_buckets = 0usize;
for (bucket, generation) in snapshot {
if dirty_buckets.get(bucket).is_some_and(|current| current == generation) {
@@ -634,6 +677,7 @@ pub(super) fn clear_dirty_usage_buckets(snapshot: &DirtyUsageBuckets) {
}
}
if cleared_buckets > 0 {
producer_identities.retain(|bucket, _| dirty_buckets.contains_key(bucket));
advance_generation(&DIRTY_USAGE_BUCKET_GENERATION);
}
(cleared_buckets, dirty_buckets.len())
@@ -724,16 +768,28 @@ pub(super) fn dirty_usage_snapshot_status(snapshot: &DirtyUsageSnapshot) -> Dirt
}
pub(super) fn dirty_usage_producer_evidence(snapshot: &DirtyUsageSnapshot) -> DirtyUsageProducerEvidence {
let generation_window_bound = dirty_usage_snapshot_status(snapshot) == DirtyUsageSnapshotStatus::Current
let snapshot_current = dirty_usage_snapshot_status(snapshot) == DirtyUsageSnapshotStatus::Current
&& snapshot.generation != 0
&& snapshot.generation != u64::MAX;
let identities = dirty_usage_producer_identities()
.iter()
.filter(|(_, state)| state.first_generation <= snapshot.generation)
.map(|(identity, _)| *identity)
.collect::<BTreeSet<_>>();
let producer_identity_coverage_complete =
generation_window_bound && crate::segment_invalidation::complete_segment_invalidation_producers(identities).is_ok();
&& snapshot.generation != u64::MAX
&& !snapshot.buckets.is_empty();
let producer_identities = dirty_usage_producer_identities();
let mut generation_start = u64::MAX;
let mut generation_end = 0;
let producer_identity_coverage_complete = snapshot_current
&& snapshot.buckets.iter().all(|(bucket, generation)| {
producer_identities.get(bucket).is_some_and(|state| {
generation_start = generation_start.min(state.first_generation);
generation_end = generation_end.max(state.last_generation);
state.fully_identified && state.last_generation == *generation
})
})
&& DIRTY_USAGE_PRODUCER_COVERAGE.load(Ordering::Acquire)
& crate::segment_invalidation::SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION_COVERAGE_MASK
== crate::segment_invalidation::SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION_COVERAGE_MASK;
let generation_window_bound = producer_identity_coverage_complete
&& generation_start != u64::MAX
&& generation_end >= generation_start
&& generation_end <= snapshot.generation;
DirtyUsageProducerEvidence {
producer_identity_coverage_complete,
@@ -743,8 +799,8 @@ pub(super) fn dirty_usage_producer_evidence(snapshot: &DirtyUsageSnapshot) -> Di
durable_producer_identity: false,
restart_gap_absent: false,
generation_window_bound,
generation_start: snapshot.generation,
generation_end: snapshot.generation,
generation_start: if generation_window_bound { generation_start } else { 0 },
generation_end: if generation_window_bound { generation_end } else { 0 },
}
}
@@ -758,6 +814,7 @@ pub(crate) fn clear_dirty_usage_buckets_for_tests() {
dirty_usage_buckets().clear();
dirty_usage_bucket_scopes().clear();
dirty_usage_producer_identities().clear();
DIRTY_USAGE_PRODUCER_COVERAGE.store(0, Ordering::Release);
}
#[cfg(test)]
crates/scanner/src/scanner_io/tests.rs
+44 -1
View File
@@ -367,7 +367,48 @@ fn scanner_segment_reuse_activation_replays_cold_durable_baseline() {
assert!(!preflight.scanner_segment_reuse_activated);
assert_eq!(
preflight.fail_closed_blockers().collect::<Vec<_>>(),
vec!["missing_producer_identity", "restart_gap", "missing_cold_zero_walk_oracle"]
vec!["missing_cold_zero_walk_oracle"]
);
clear_dirty_usage_buckets(dirty_usage_snapshot.buckets.as_ref());
record_dirty_usage_object_from_producer("photos", "2027/object", SegmentInvalidationProducerIdentity::PutObject);
let later_dirty_usage_snapshot =
snapshot_dirty_usage_buckets(&[bucket_info("photos"), bucket_info("archive")], dirty_usage_generation());
let preflight = scanner_segment_reuse_activation_preflight_for_baseline(
&later_dirty_usage_snapshot,
false,
ScannerCacheBaselineProof {
authoritative_data: Some(&baseline),
observed_candidate_data: None,
expected_sources: &expected_sources,
leader_epoch: 11,
want_cycle: 8,
scan_plan_digest,
},
);
assert!(preflight.scanner_segment_reuse_activated);
assert_eq!(preflight.fail_closed_blockers().collect::<Vec<_>>(), Vec::<&str>::new());
record_dirty_usage_bucket("photos");
let unidentified_snapshot =
snapshot_dirty_usage_buckets(&[bucket_info("photos"), bucket_info("archive")], dirty_usage_generation());
let preflight = scanner_segment_reuse_activation_preflight_for_baseline(
&unidentified_snapshot,
false,
ScannerCacheBaselineProof {
authoritative_data: Some(&baseline),
observed_candidate_data: None,
expected_sources: &expected_sources,
leader_epoch: 11,
want_cycle: 8,
scan_plan_digest,
},
);
assert!(!preflight.scanner_segment_reuse_activated);
assert!(
preflight
.fail_closed_blockers()
.any(|blocker| blocker == "missing_producer_identity")
);
clear_dirty_usage_buckets_for_tests();
}
@@ -1399,6 +1440,8 @@ fn dirty_usage_producer_evidence_tracks_process_local_coverage_without_durable_r
assert!(evidence.producer_identity_coverage_complete);
assert!(!evidence.durable_producer_identity);
assert!(!evidence.restart_gap_absent);
assert_eq!(evidence.generation_start, snapshot.buckets["photos"]);
assert_eq!(evidence.generation_end, snapshot.buckets["photos"]);
record_dirty_usage_bucket_from_producer("videos", SegmentInvalidationProducerIdentity::PutObject);
let stale_evidence = dirty_usage_producer_evidence(&snapshot);
crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs
+34 -2
View File
@@ -275,10 +275,42 @@ async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks_
run_entry(&store, 1, Some(&hot), false, false, false).await;
let usage = run_entry(&store, 2, Some(&hot), true, true, false).await;
persist_baseline(&store, &usage).await;
acknowledge_dirty_usage_generation(scanner_activity_epoch(), dirty_usage_generation())
.expect("durable whole-cycle publication should acknowledge the initial producer window");
put_and_settle(&store, &hot, "hot-segment/object").await;
record_dirty_usage_object_from_producer(
&hot,
"hot-segment/object",
crate::segment_invalidation::SegmentInvalidationProducerIdentity::PutObject,
);
let usage = run_entry(&store, 3, Some(&hot), true, true, true).await;
assert_eq!(usage.buckets_usage[&hot].objects_count, 1);
assert_eq!(usage.buckets_usage[&hot].objects_count, 2);
assert_eq!(usage.buckets_usage[&cold].objects_count, 1);
assert_eq!(usage.objects_total_count, 2);
assert_eq!(usage.objects_total_count, 3);
persist_baseline(&store, &usage).await;
acknowledge_dirty_usage_generation(scanner_activity_epoch(), dirty_usage_generation())
.expect("durable prefix publication should acknowledge the typed suffix");
for index in 0..=MAX_DIRTY_USAGE_TOP_LEVEL_ENTRIES_PER_BUCKET {
record_dirty_usage_object_from_producer(
&hot,
&format!("overflow-{index}/object"),
crate::segment_invalidation::SegmentInvalidationProducerIdentity::PutObject,
);
}
let usage = run_entry(&store, 4, Some(&hot), true, true, false).await;
assert_eq!(usage.objects_total_count, 3);
persist_baseline(&store, &usage).await;
acknowledge_dirty_usage_generation(scanner_activity_epoch(), dirty_usage_generation())
.expect("durable whole-bucket fallback should acknowledge the overflow window");
record_dirty_usage_object_from_producer(
&hot,
"hot-segment/object",
crate::segment_invalidation::SegmentInvalidationProducerIdentity::Unknown,
);
let usage = run_entry(&store, 5, Some(&hot), true, false, false).await;
assert_eq!(usage.objects_total_count, 3);
clear_dirty_usage_buckets_for_tests();
}
crates/scanner/src/segment_invalidation.rs
+26
View File
@@ -83,6 +83,25 @@ impl SegmentInvalidationProducerIdentity {
Self::DirectoryObject,
];
pub(crate) const REQUIRED_PRODUCTION_COVERAGE_MASK: u64 = (1_u64 << 11) - 1;
pub(crate) const fn production_coverage_bit(self) -> Option<u64> {
match self {
Self::PutObject => Some(1_u64 << 0),
Self::DeleteObject => Some(1_u64 << 1),
Self::DeleteMarker => Some(1_u64 << 2),
Self::CompleteMultipartUpload => Some(1_u64 << 3),
Self::AbortMultipartUpload => Some(1_u64 << 4),
Self::ObjectMetadata => Some(1_u64 << 5),
Self::BucketMetadata => Some(1_u64 << 6),
Self::Replication => Some(1_u64 << 7),
Self::TierTransition => Some(1_u64 << 8),
Self::TierExpiration => Some(1_u64 << 9),
Self::DirectoryObject => Some(1_u64 << 10),
Self::Unknown | Self::TestFixture => None,
}
}
pub fn producer(self) -> Option<SegmentInvalidationProducer> {
match self {
Self::PutObject => Some(SegmentInvalidationProducer::Put),
@@ -400,6 +419,13 @@ mod tests {
#[test]
fn segment_invalidation_producer_identities_must_be_known_and_complete() {
let coverage_mask = SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION
.into_iter()
.filter_map(SegmentInvalidationProducerIdentity::production_coverage_bit)
.fold(0_u64, std::ops::BitOr::bitor);
assert_eq!(coverage_mask, SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION_COVERAGE_MASK);
assert_eq!(SegmentInvalidationProducerIdentity::Unknown.production_coverage_bit(), None);
assert_eq!(SegmentInvalidationProducerIdentity::TestFixture.production_coverage_bit(), None);
assert_eq!(
complete_segment_invalidation_producers(SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION),
Ok(SegmentInvalidationProducer::REQUIRED.into_iter().collect())