fix(scanner): require segment producer identities (#7444)

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs

@@ -3,7 +3,8 @@
 use super::*;
 use crate::segment_invalidation::{
     MAX_SEGMENT_INVALIDATION_BYTES, MAX_SEGMENT_INVALIDATION_ENTRIES, SegmentInvalidationDomain, SegmentInvalidationEnvelope,
-    SegmentInvalidationError, SegmentInvalidationProducer, SegmentInvalidationProof, admit_segment_invalidation,
+    SegmentInvalidationError, SegmentInvalidationProducer, SegmentInvalidationProducerIdentity, SegmentInvalidationProof,
+    admit_segment_invalidation, complete_segment_invalidation_producers,
 };
 use std::collections::BTreeSet;
 
@@ -11,7 +12,8 @@ const MAX_WALK_SAMPLES: usize = 32;
 const MAX_WALK_BYTES: usize = 1024;
 
 fn segment_producers() -> BTreeSet<SegmentInvalidationProducer> {
-    SegmentInvalidationProducer::REQUIRED.into_iter().collect()
+    complete_segment_invalidation_producers(SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION)
+        .expect("fixture should enumerate the complete production producer matrix")
 }
 
 fn segment_envelope() -> SegmentInvalidationEnvelope {

crates/scanner/src/segment_invalidation.rs

@@ -25,6 +25,7 @@ pub enum SegmentInvalidationError {
     ByteLimit,
     InvalidProof,
     InvalidKey,
+    UnknownProducer,
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
@@ -50,6 +51,74 @@ impl SegmentInvalidationProducer {
     ];
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub enum SegmentInvalidationProducerIdentity {
+    PutObject,
+    DeleteObject,
+    DeleteMarker,
+    CompleteMultipartUpload,
+    Replication,
+    TierTransition,
+    TierExpiration,
+    DirectoryObject,
+    Unknown,
+    TestFixture,
+}
+
+impl SegmentInvalidationProducerIdentity {
+    pub const REQUIRED_PRODUCTION: [Self; 8] = [
+        Self::PutObject,
+        Self::DeleteObject,
+        Self::DeleteMarker,
+        Self::CompleteMultipartUpload,
+        Self::Replication,
+        Self::TierTransition,
+        Self::TierExpiration,
+        Self::DirectoryObject,
+    ];
+
+    pub fn producer(self) -> Option<SegmentInvalidationProducer> {
+        match self {
+            Self::PutObject => Some(SegmentInvalidationProducer::Put),
+            Self::DeleteObject => Some(SegmentInvalidationProducer::Delete),
+            Self::DeleteMarker => Some(SegmentInvalidationProducer::DeleteMarker),
+            Self::CompleteMultipartUpload => Some(SegmentInvalidationProducer::Multipart),
+            Self::Replication => Some(SegmentInvalidationProducer::Replication),
+            Self::TierTransition | Self::TierExpiration => Some(SegmentInvalidationProducer::Tier),
+            Self::DirectoryObject => Some(SegmentInvalidationProducer::DirectoryObject),
+            Self::Unknown | Self::TestFixture => None,
+        }
+    }
+}
+
+pub fn complete_segment_invalidation_producers<I>(
+    identities: I,
+) -> Result<BTreeSet<SegmentInvalidationProducer>, SegmentInvalidationError>
+where
+    I: IntoIterator<Item = SegmentInvalidationProducerIdentity>,
+{
+    let mut covered_identities = BTreeSet::new();
+    let mut producers = BTreeSet::new();
+    for identity in identities {
+        let Some(producer) = identity.producer() else {
+            return Err(SegmentInvalidationError::UnknownProducer);
+        };
+        covered_identities.insert(identity);
+        producers.insert(producer);
+    }
+    if SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION
+        .iter()
+        .all(|identity| covered_identities.contains(identity))
+        && SegmentInvalidationProducer::REQUIRED
+            .iter()
+            .all(|producer| producers.contains(producer))
+    {
+        Ok(producers)
+    } else {
+        Err(SegmentInvalidationError::InvalidProof)
+    }
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum SegmentInvalidationDomain {
     LocalSingleSet,
@@ -170,7 +239,8 @@ mod tests {
     use super::*;
 
     fn producers() -> BTreeSet<SegmentInvalidationProducer> {
-        SegmentInvalidationProducer::REQUIRED.into_iter().collect()
+        complete_segment_invalidation_producers(SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION)
+            .expect("production producer matrix should be complete")
     }
 
     fn envelope() -> SegmentInvalidationEnvelope {
@@ -320,6 +390,63 @@ mod tests {
         );
     }
 
+    #[test]
+    fn segment_invalidation_producer_identities_must_be_known_and_complete() {
+        assert_eq!(
+            complete_segment_invalidation_producers(SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION),
+            Ok(SegmentInvalidationProducer::REQUIRED.into_iter().collect())
+        );
+        assert_eq!(
+            complete_segment_invalidation_producers([
+                SegmentInvalidationProducerIdentity::PutObject,
+                SegmentInvalidationProducerIdentity::DeleteObject,
+                SegmentInvalidationProducerIdentity::DeleteMarker,
+                SegmentInvalidationProducerIdentity::CompleteMultipartUpload,
+                SegmentInvalidationProducerIdentity::Replication,
+                SegmentInvalidationProducerIdentity::TierTransition,
+                SegmentInvalidationProducerIdentity::DirectoryObject,
+                SegmentInvalidationProducerIdentity::Unknown,
+            ]),
+            Err(SegmentInvalidationError::UnknownProducer)
+        );
+        assert_eq!(
+            complete_segment_invalidation_producers([
+                SegmentInvalidationProducerIdentity::PutObject,
+                SegmentInvalidationProducerIdentity::DeleteObject,
+                SegmentInvalidationProducerIdentity::DeleteMarker,
+                SegmentInvalidationProducerIdentity::CompleteMultipartUpload,
+                SegmentInvalidationProducerIdentity::Replication,
+                SegmentInvalidationProducerIdentity::TierTransition,
+                SegmentInvalidationProducerIdentity::DirectoryObject,
+                SegmentInvalidationProducerIdentity::TestFixture,
+            ]),
+            Err(SegmentInvalidationError::UnknownProducer)
+        );
+        assert_eq!(
+            complete_segment_invalidation_producers([
+                SegmentInvalidationProducerIdentity::PutObject,
+                SegmentInvalidationProducerIdentity::DeleteObject,
+                SegmentInvalidationProducerIdentity::DeleteMarker,
+                SegmentInvalidationProducerIdentity::Replication,
+                SegmentInvalidationProducerIdentity::TierTransition,
+                SegmentInvalidationProducerIdentity::DirectoryObject,
+            ]),
+            Err(SegmentInvalidationError::InvalidProof)
+        );
+        assert_eq!(
+            complete_segment_invalidation_producers([
+                SegmentInvalidationProducerIdentity::PutObject,
+                SegmentInvalidationProducerIdentity::DeleteObject,
+                SegmentInvalidationProducerIdentity::DeleteMarker,
+                SegmentInvalidationProducerIdentity::CompleteMultipartUpload,
+                SegmentInvalidationProducerIdentity::Replication,
+                SegmentInvalidationProducerIdentity::TierTransition,
+                SegmentInvalidationProducerIdentity::DirectoryObject,
+            ]),
+            Err(SegmentInvalidationError::InvalidProof)
+        );
+    }
+
     #[test]
     fn segment_invalidation_entries_are_bounded_and_key_checked() {
         let envelope = envelope();