Merge branch 'main' into overtrue/fix/lifecycle-rule-validation

`NewerNoncurrentVersions` had no lower bound at PUT, and evaluation read a
negative count through `usize::try_from(...).unwrap_or(usize::MAX)`. An
HTTP-accepted rule therefore retained (almost) everything and silently
stopped expiring versions — the one outcome a retention rule must never
produce by accident.

Reject a negative count during validation, and stop reading one as
"retain everything" anywhere it can still arrive from older persistence
or an import: evaluation takes no action for such a rule and says so in a
diagnostic, the batch limit path yields no event, and `Evaluator::eval`
reports a typed corruption error to callers that can surface one.

A count-only noncurrent expiration is a MinIO extension, not an AWS form.
It used to be rejected as an actionless rule and was never executed. It
is now accepted and honoured with the semantics MinIO gives it: the
newest N noncurrent versions are kept and every older one is due as soon
as it became noncurrent. Zero keeps the meaning the batch limit path has
always given it — no count constraint — so a zero-count rule with no age
condition still has no action.

`LifecycleRuleFilter` is an all-`Option` DTO, so the schema constraints
were not checked anywhere: validate at most one top-level predicate, an
`And` that combines at least two, no repeated tag key, tag key/value
limits, non-negative sizes, and `ObjectSizeGreaterThan <
ObjectSizeLessThan`. An empty filter stays valid — AWS documents it as
"every object in the bucket".

Schema-shape violations are reported with a distinct `ErrorKind` so the
S3 boundary answers them with `MalformedXML`; rejected values keep the
`InvalidArgument` this path has always returned.

backlog#2201

.config/make/tests.mak

@@ -31,7 +31,6 @@ script-tests: ## Run shell script tests
 	./scripts/test_object_batch_bench_enhanced.sh
 	./scripts/test_hotpath_warp_ab_gate.sh
 	./scripts/test_hotpath_warp_abba.sh
-	./scripts/test_scanner_validation_harness.sh
 	./scripts/test_exact_1mib_handoff_abba.sh
 	./scripts/test_pinned_paired_abba_bench.sh
 	./scripts/test_manual_transition_runbooks.sh

crates/e2e_test/src/on_demand_migration/concurrency_test.rs

@@ -20,9 +20,10 @@
 //! journal (`count_requests`) carries the assertion in every one of them.
 
 use super::common::{BoxError, OdmTestEnv, RawResponse, SeedObject, start_configured_env};
-use crate::fake_s3_target::Operation;
+use crate::fake_s3_target::{FaultAction, Operation};
 use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
 use bytes::Bytes;
+use futures::{StreamExt, TryStreamExt};
 use std::time::Duration;
 
 type TestResult = Result<(), BoxError>;
@@ -145,14 +146,38 @@ async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients()
     .await?;
 
     let body = payload(128 * 1024);
+    let blocker = "queue/blocker.bin";
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(blocker, body.clone())]);
+    // The one-chunk range completes immediately; its full background pull
+    // occupies the only slot while the remaining requests fill the queue.
+    env.source.inject_for_key(
+        Operation::GetObject,
+        blocker,
+        FaultAction::SlowSendBody {
+            chunk_bytes: 1024,
+            delay: Duration::from_millis(100),
+        },
+        2,
+    );
+    let response = env
+        .raw_object_request(http::Method::GET, bucket, blocker, &[("range", "bytes=0-1023")])
+        .await?;
+    assert_eq!(response.status, 206);
+    assert_eq!(response.body, body.slice(0..1024));
+    env.wait_for_status_counter(bucket, "/inflight_pulls", 1, SETTLE).await?;
+
     let keys: Vec<String> = (0..REQUESTS).map(|index| format!("queue/object-{index:03}.bin")).collect();
     let seeds: Vec<SeedObject> = keys.iter().map(|key| SeedObject::new(key.clone(), body.clone())).collect();
     env.seed_source(SOURCE_BUCKET, &seeds);
 
-    let responses: Vec<RawResponse> = futures::future::try_join_all(
+    // Bound source connections below the fixture's limit while still
+    // submitting all 100 requests to the eight-slot background queue.
+    let responses: Vec<RawResponse> = futures::stream::iter(
         keys.iter()
             .map(|key| env.raw_object_request(http::Method::GET, bucket, key, &[("range", "bytes=0-1023")])),
     )
+    .buffered(16)
+    .try_collect()
     .await?;
     for (key, response) in keys.iter().zip(&responses) {
         assert_eq!(response.status, 206, "{key}: {}", String::from_utf8_lossy(&response.body));
@@ -168,6 +193,15 @@ async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients()
         .wait_for_status_counter(bucket, "/counters/pull_failures_total/queue_full", 1, SETTLE)
         .await?;
     assert!(queue_full > 0, "a 100-deep burst must overflow an 8-slot queue");
+    let queue_full = usize::try_from(queue_full)?;
+    assert!(queue_full <= REQUESTS);
+    env.wait_for_status_counter(
+        bucket,
+        "/counters/pulled_objects_total/background",
+        u64::try_from(REQUESTS + 1 - queue_full)?,
+        SETTLE,
+    )
+    .await?;
 
     let ranged_reads: usize = keys.iter().map(|key| source_get_count(&env, key)).sum();
     assert!(
@@ -175,9 +209,6 @@ async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients()
         "every reader is served from the source: {ranged_reads} GETs for {REQUESTS} readers"
     );
     let dropped = keys.iter().filter(|key| source_get_count(&env, key) == 1).count();
-    assert!(
-        dropped > 0,
-        "the overflowed keys are the ones with no backfill GET, but every key got one"
-    );
+    assert_eq!(dropped, queue_full, "only overflowed keys remain without a background GET");
     Ok(())
 }

crates/e2e_test/src/on_demand_migration/list_through_test.rs

@@ -265,16 +265,13 @@ async fn list_through_rejects_a_tampered_continuation_token() -> TestResult {
     let decoded = String::from_utf8(base64_simd::STANDARD.decode_to_vec(token.as_bytes())?)?;
     assert!(decoded.contains("\"t\":\"odm-list\""), "the merged token is an envelope: {decoded}");
 
-    let tampered = base64_simd::STANDARD.encode_to_string(decoded.replace("\"v\":1", "\"v\":2").as_bytes());
-    let rejected = env
-        .raw_list_objects_v2(bucket, &format!("continuation-token={tampered}"))
-        .await?;
-    assert_eq!(
-        rejected.status,
-        400,
-        "a bumped token version is a client error: {}",
-        String::from_utf8_lossy(&rejected.body)
-    );
+    let tampered = base64_simd::STANDARD.encode_to_string(decoded.replace("\"v\":1", "\"v\":3").as_bytes());
+    assert_ne!(tampered, token, "the test must change the token version");
+    let query = serde_urlencoded::to_string([("continuation-token", tampered.as_str())])?;
+    let rejected = env.raw_list_objects_v2(bucket, &query).await?;
+    let error_body = String::from_utf8_lossy(&rejected.body);
+    assert_eq!(rejected.status, 400, "a bumped token version is a client error: {}", error_body);
+    assert!(error_body.contains("<Code>InvalidArgument</Code>"), "{error_body}");
     Ok(())
 }
 

crates/ecstore/src/api/mod.rs

@@ -89,8 +89,9 @@ pub mod bucket {
         #[allow(clippy::module_inception)]
         pub mod lifecycle {
             pub use crate::bucket::lifecycle::lifecycle::{
-                Event, ExpirationOptions, IlmAction, Lifecycle, LifecycleCalculate, ObjectOpts, RuleValidate,
-                TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions, expected_expiry_time, object_opts_from_object_info,
+                Event, ExpirationOptions, IlmAction, LIFECYCLE_MALFORMED_XML_ERROR_KIND, Lifecycle, LifecycleCalculate,
+                ObjectOpts, RuleValidate, TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions, expected_expiry_time,
+                object_opts_from_object_info,
             };
         }
 

crates/ecstore/src/bucket/lifecycle/core.rs

@@ -15,9 +15,9 @@
 use crate::object_api::ObjectInfo;
 
 pub use rustfs_lifecycle::{
-    Event, ExpirationOptions, IlmAction, Lifecycle, LifecycleCalculate, ObjectOpts, RuleValidate, TRANSITION_COMPLETE,
-    TRANSITION_PENDING, TransitionOptions, abort_incomplete_multipart_upload_due, expected_expiry_time,
-    expiration_action_has_valid_target,
+    Event, ExpirationOptions, IlmAction, LIFECYCLE_MALFORMED_XML_ERROR_KIND, Lifecycle, LifecycleCalculate, ObjectOpts,
+    RuleValidate, TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions, abort_incomplete_multipart_upload_due,
+    expected_expiry_time, expiration_action_has_valid_target,
 };
 
 pub fn object_opts_from_object_info(oi: &ObjectInfo) -> ObjectOpts {

crates/lifecycle/src/core.rs

@@ -18,7 +18,7 @@ use s3s::dto::{
     BucketLifecycleConfiguration, ExpirationStatus, LifecycleExpiration, LifecycleRule, LifecycleRuleFilter,
     NoncurrentVersionTransition, ObjectLockConfiguration, ObjectLockEnabled, RestoreRequest, Transition,
 };
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 use time::macros::offset;
 use time::{self, Duration, OffsetDateTime};
@@ -65,6 +65,66 @@ const ERR_LIFECYCLE_EXPIRED_OBJECT_DELETE_MARKER_WITH_TAGS: &str =
     "Rule with ExpiredObjectDeleteMarker cannot have tags based filtering";
 const ERR_LIFECYCLE_RULE_MUST_HAVE_ACTION: &str = "Rule must have at least one of Expiration, Transition, NoncurrentVersionExpiration, NoncurrentVersionTransition, or DelMarkerExpiration";
 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_TOO_MANY_PREDICATES: &str =
+    "Filter must have at most one of Prefix, Tag, ObjectSizeGreaterThan, ObjectSizeLessThan or And; combine predicates with And";
+const ERR_LIFECYCLE_FILTER_AND_TOO_FEW_PREDICATES: &str = "Filter And must contain at least two 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";
+const ERR_LIFECYCLE_FILTER_SIZE_RANGE: &str = "ObjectSizeGreaterThan must be smaller than ObjectSizeLessThan";
+/// Longest tag key S3 accepts.
+const MAX_TAG_KEY_LEN: usize = 128;
+/// Longest tag value S3 accepts.
+const MAX_TAG_VALUE_LEN: usize = 256;
+
+/// A validation failure that the S3 boundary must answer with `MalformedXML`
+/// rather than `InvalidArgument`: the document does not match the published
+/// schema shape (wrong number of `Filter` predicates, a one-member `And`).
+///
+/// Everything else stays [`std::io::ErrorKind::Other`], which the boundary
+/// already maps to `InvalidArgument`.
+pub const LIFECYCLE_MALFORMED_XML_ERROR_KIND: std::io::ErrorKind = std::io::ErrorKind::InvalidData;
+
+/// A persisted rule that could never have passed validation. Callers that can
+/// report an error surface it; evaluation itself stays fail-closed and takes
+/// no action for the rule.
+pub const LIFECYCLE_CORRUPT_RULE_ERROR_KIND: std::io::ErrorKind = std::io::ErrorKind::InvalidData;
+
+fn malformed_xml_error(message: &'static str) -> std::io::Error {
+    std::io::Error::new(LIFECYCLE_MALFORMED_XML_ERROR_KIND, message)
+}
+
+/// The retention count a rule keeps, or `None` when the persisted value is
+/// negative — a shape PUT validation rejects, so reaching it means the rule
+/// came from older persistence or an import.
+///
+/// A negative count must never be read as "retain everything": that is how an
+/// invalid configuration silently stopped deleting versions (backlog#2201).
+pub fn retained_noncurrent_versions(count: i32) -> Option<usize> {
+    usize::try_from(count).ok()
+}
+
+/// Does any rule carry a retention count that validation would have rejected?
+pub fn lifecycle_has_corrupt_retention_count(lc: &BucketLifecycleConfiguration) -> bool {
+    lc.rules.iter().any(rule_has_corrupt_retention_count)
+}
+
+fn rule_has_corrupt_retention_count(rule: &LifecycleRule) -> bool {
+    let expiration_count = rule
+        .noncurrent_version_expiration
+        .as_ref()
+        .and_then(|expiration| expiration.newer_noncurrent_versions);
+    let transition_counts = rule
+        .noncurrent_version_transitions
+        .iter()
+        .flatten()
+        .filter_map(|transition| transition.newer_noncurrent_versions);
+    expiration_count
+        .into_iter()
+        .chain(transition_counts)
+        .any(|count| retained_noncurrent_versions(count).is_none())
+}
 
 pub use rustfs_scanner_metrics::metrics::IlmAction;
 
@@ -141,6 +201,17 @@ impl RuleValidate for LifecycleRule {
             return Err(std::io::Error::other(ERR_LIFECYCLE_PREFIX_FILTER_CONFLICT));
         }
 
+        if let Some(filter) = self.filter.as_ref() {
+            validate_lifecycle_filter(filter)?;
+        }
+
+        // A negative retention count was accepted and then read as "retain
+        // (almost) everything" during evaluation, so an HTTP-accepted rule
+        // silently stopped deleting versions (backlog#2201).
+        if rule_has_corrupt_retention_count(self) {
+            return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_NEWER_NONCURRENT_VERSIONS));
+        }
+
         // Rule with DelMarkerExpiration cannot have tags based filtering
         let has_tag_filter = self
             .filter
@@ -173,11 +244,14 @@ impl RuleValidate for LifecycleRule {
         // Rule must have at least one action
         let has_expiration = self.expiration.is_some();
         let has_transition = self.transitions.as_ref().is_some_and(|t| !t.is_empty());
-        let has_noncurrent_expiration = self
-            .noncurrent_version_expiration
-            .as_ref()
-            .and_then(|e| e.noncurrent_days)
-            .is_some();
+        // `NewerNoncurrentVersions` on its own is a MinIO extension, not an AWS
+        // form: it keeps the newest N noncurrent versions and expires the rest
+        // with no age condition. RustFS accepts it for MinIO compatibility, so
+        // it has to count as an action here — otherwise a count-only rule was
+        // rejected as actionless (backlog#2201).
+        let has_noncurrent_expiration = self.noncurrent_version_expiration.as_ref().is_some_and(|expiration| {
+            expiration.noncurrent_days.is_some() || expiration.newer_noncurrent_versions.is_some_and(|count| count > 0)
+        });
         let has_noncurrent_transition = self
             .noncurrent_version_transitions
             .as_ref()
@@ -203,6 +277,81 @@ impl RuleValidate for LifecycleRule {
     }
 }
 
+/// Structural validation for `LifecycleRuleFilter`.
+///
+/// The generated DTO is all-`Option`, so the S3 schema constraints have to be
+/// checked here: at most one top-level predicate, an `And` that actually
+/// combines at least two, no repeated tag key, tag key/value limits, and a
+/// coherent non-negative size range (backlog#2201).
+///
+/// A filter with no predicate at all stays valid: AWS documents an empty
+/// `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> {
+    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)?;
+    }
+
+    if let Some(and) = filter.and.as_ref() {
+        let tags = and.tags.as_deref().unwrap_or(&[]);
+        let and_predicates = usize::from(and.prefix.is_some())
+            + tags.len()
+            + usize::from(and.object_size_greater_than.is_some())
+            + usize::from(and.object_size_less_than.is_some());
+        if and_predicates < 2 {
+            return Err(malformed_xml_error(ERR_LIFECYCLE_FILTER_AND_TOO_FEW_PREDICATES));
+        }
+        let mut seen_keys = HashSet::with_capacity(tags.len());
+        for tag in tags {
+            validate_lifecycle_tag(tag)?;
+            let key = tag.key.as_deref().unwrap_or_default();
+            if !seen_keys.insert(key) {
+                return Err(std::io::Error::other(ERR_LIFECYCLE_FILTER_DUPLICATE_TAG_KEY));
+            }
+        }
+        validate_lifecycle_size_bounds(and.object_size_greater_than, and.object_size_less_than)?;
+    }
+
+    validate_lifecycle_size_bounds(filter.object_size_greater_than, filter.object_size_less_than)?;
+
+    Ok(())
+}
+
+/// S3 requires a tag to carry a key and value; both are length-bounded.
+/// The DTO makes both optional, so incomplete tags have to be rejected here
+/// rather than silently matching nothing.
+fn validate_lifecycle_tag(tag: &s3s::dto::Tag) -> Result<(), std::io::Error> {
+    let key = tag.key.as_deref().unwrap_or_default();
+    let Some(value) = tag.value.as_deref() else {
+        return Err(std::io::Error::other(ERR_LIFECYCLE_FILTER_INVALID_TAG));
+    };
+    if key.is_empty() || key.chars().count() > MAX_TAG_KEY_LEN || value.chars().count() > MAX_TAG_VALUE_LEN {
+        return Err(std::io::Error::other(ERR_LIFECYCLE_FILTER_INVALID_TAG));
+    }
+    Ok(())
+}
+
+fn validate_lifecycle_size_bounds(greater_than: Option<i64>, less_than: Option<i64>) -> Result<(), std::io::Error> {
+    if greater_than.is_some_and(|size| size < 0) || less_than.is_some_and(|size| size < 0) {
+        return Err(std::io::Error::other(ERR_LIFECYCLE_FILTER_NEGATIVE_SIZE));
+    }
+    if let (Some(greater_than), Some(less_than)) = (greater_than, less_than)
+        && greater_than >= less_than
+    {
+        return Err(std::io::Error::other(ERR_LIFECYCLE_FILTER_SIZE_RANGE));
+    }
+    Ok(())
+}
+
 fn lifecycle_rule_prefix(rule: &LifecycleRule) -> Option<&str> {
     // Prefer a non-empty legacy prefix; treat an empty legacy prefix as if it were not set
     if let Some(p) = rule.prefix.as_deref()
@@ -293,6 +442,10 @@ impl Lifecycle for BucketLifecycleConfiguration {
                 {
                     return true;
                 }
+                // A positive count is an action on its own (the MinIO count-only
+                // form). Zero means "no count constraint" here, exactly as the
+                // batch limit path reads it, and a negative count is corrupt —
+                // neither makes the rule active (backlog#2201).
                 if let Some(newer_noncurrent_versions) = rule_noncurrent_version_expiration.newer_noncurrent_versions
                     && newer_noncurrent_versions > 0
                 {
@@ -563,6 +716,23 @@ impl Lifecycle for BucketLifecycleConfiguration {
 
         if let Some(ref lc_rules) = self.filter_rules(obj).await {
             for rule in lc_rules.iter() {
+                // A retention count that PUT validation would have rejected can
+                // only come from older persistence or an import. Take no action
+                // for the rule instead of allowing another action on the same
+                // corrupt rule to delete or transition an object (backlog#2201).
+                if rule_has_corrupt_retention_count(rule) {
+                    debug!(
+                        event = EVENT_LIFECYCLE_NONCURRENT_EXPIRY_SKIPPED,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_LIFECYCLE,
+                        object = %obj.name,
+                        rule_id = %rule.id.clone().unwrap_or_default(),
+                        reason = "corrupt_newer_noncurrent_versions",
+                        "Skipped lifecycle evaluation for a rule with an invalid retention count"
+                    );
+                    continue;
+                }
+
                 if obj.is_latest && obj.expired_object_deletemarker() {
                     if let Some(expiration) = rule.expiration.as_ref()
                         && expiration.expired_object_delete_marker.is_some_and(|v| v)
@@ -619,11 +789,18 @@ impl Lifecycle for BucketLifecycleConfiguration {
 
                 if !obj.is_latest
                     && let Some(ref noncurrent_version_expiration) = rule.noncurrent_version_expiration
-                    && let Some(noncurrent_days) = noncurrent_version_expiration.noncurrent_days
+                    && (noncurrent_version_expiration.noncurrent_days.is_some()
+                        || noncurrent_version_expiration
+                            .newer_noncurrent_versions
+                            .is_some_and(|count| count > 0))
                     && noncurrent_version_expiration
                         .newer_noncurrent_versions
                         .is_none_or(|retain| usize::try_from(retain).is_ok_and(|retain| newer_noncurrent_versions >= retain))
                 {
+                    // A count-only rule (MinIO extension) has no age condition:
+                    // every version past the retained count is due as soon as it
+                    // became noncurrent, i.e. zero days after the successor.
+                    let noncurrent_days = noncurrent_version_expiration.noncurrent_days.unwrap_or(0);
                     if let Some(successor_mod_time) = obj.successor_mod_time {
                         let expected_expiry = expected_expiry_time(successor_mod_time, noncurrent_days);
                         if now.unix_timestamp() >= expected_expiry.unix_timestamp() {
@@ -791,15 +968,18 @@ impl Lifecycle for BucketLifecycleConfiguration {
             for rule in filter_rules.iter() {
                 if let Some(ref noncurrent_version_expiration) = rule.noncurrent_version_expiration {
                     return if let Some(newer_noncurrent_versions) = noncurrent_version_expiration.newer_noncurrent_versions {
-                        if newer_noncurrent_versions == 0 {
+                        // Zero means "no count constraint"; a negative count is
+                        // corrupt and must not be read as "retain everything"
+                        // (backlog#2201). Neither yields a limit event.
+                        let Some(retained) = retained_noncurrent_versions(newer_noncurrent_versions).filter(|c| *c > 0) else {
                             continue;
-                        }
+                        };
                         Event {
                             action: IlmAction::DeleteVersionAction,
                             rule_id: rule.id.clone().unwrap_or_default(),
                             noncurrent_days: u32::try_from(noncurrent_version_expiration.noncurrent_days.unwrap_or(0))
                                 .unwrap_or(u32::MAX),
-                            newer_noncurrent_versions: usize::try_from(newer_noncurrent_versions).unwrap_or(usize::MAX),
+                            newer_noncurrent_versions: retained,
                             due: Some(OffsetDateTime::UNIX_EPOCH),
                             storage_class: "".into(),
                         }
@@ -1162,7 +1342,11 @@ mod tests {
     use super::*;
     use metrics_util::MetricKind;
     use metrics_util::debugging::{DebugValue, DebuggingRecorder};
-    use s3s::dto::{LifecycleRuleFilter, TransitionStorageClass};
+    use s3s::dto::{
+        LifecycleRuleAndOperator, LifecycleRuleFilter, NoncurrentVersionExpiration, NoncurrentVersionTransition,
+        TransitionStorageClass,
+    };
+    use s3s::xml::{Deserialize as XmlDeserialize, SerializeContent as XmlSerializeContent};
     use serial_test::serial;
     use std::sync::Arc;
     use time::macros::datetime;
@@ -4183,6 +4367,583 @@ mod tests {
         assert_eq!(event.action, IlmAction::NoneAction);
     }
 
+    // ---- backlog#2201: retention-count and Filter invariants -----------------
+
+    fn rule_with_noncurrent_expiration(expiration: NoncurrentVersionExpiration) -> LifecycleRule {
+        LifecycleRule {
+            status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+            expiration: None,
+            abort_incomplete_multipart_upload: None,
+            del_marker_expiration: None,
+            filter: None,
+            id: Some("noncurrent".to_string()),
+            noncurrent_version_expiration: Some(expiration),
+            noncurrent_version_transitions: None,
+            prefix: None,
+            transitions: None,
+        }
+    }
+
+    fn rule_with_filter(filter: LifecycleRuleFilter) -> LifecycleRule {
+        LifecycleRule {
+            status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+            expiration: Some(LifecycleExpiration {
+                days: Some(1),
+                ..Default::default()
+            }),
+            abort_incomplete_multipart_upload: None,
+            del_marker_expiration: None,
+            filter: Some(filter),
+            id: Some("filtered".to_string()),
+            noncurrent_version_expiration: None,
+            noncurrent_version_transitions: None,
+            prefix: None,
+            transitions: None,
+        }
+    }
+
+    fn config_with_rules(rules: Vec<LifecycleRule>) -> BucketLifecycleConfiguration {
+        BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules,
+        }
+    }
+
+    fn tag(key: &str, value: &str) -> s3s::dto::Tag {
+        s3s::dto::Tag {
+            key: Some(key.to_string()),
+            value: Some(value.to_string()),
+        }
+    }
+
+    #[tokio::test]
+    async fn validate_rejects_negative_newer_noncurrent_versions() {
+        // A negative retention count used to be accepted and then read as
+        // usize::MAX during evaluation, so the rule silently stopped deleting
+        // versions (backlog#2201).
+        let lc = config_with_rules(vec![rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: Some(30),
+            newer_noncurrent_versions: Some(-1),
+        })]);
+
+        let err = lc
+            .validate(&ObjectLockConfiguration::default())
+            .await
+            .expect_err("a negative retention count must be rejected");
+
+        assert_eq!(err.to_string(), ERR_LIFECYCLE_INVALID_NEWER_NONCURRENT_VERSIONS);
+        assert_ne!(err.kind(), LIFECYCLE_MALFORMED_XML_ERROR_KIND, "value errors stay InvalidArgument");
+    }
+
+    #[tokio::test]
+    async fn validate_rejects_negative_newer_noncurrent_versions_on_transition() {
+        let mut rule = rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: Some(30),
+            newer_noncurrent_versions: None,
+        });
+        rule.noncurrent_version_transitions = Some(vec![NoncurrentVersionTransition {
+            newer_noncurrent_versions: Some(-3),
+            noncurrent_days: Some(1),
+            storage_class: Some(TransitionStorageClass::from_static(TransitionStorageClass::GLACIER)),
+        }]);
+
+        // The transition validator already refuses a negative count, and it runs
+        // first, so this pins the rejection rather than the message. The gap
+        // this PR closes is the expiration side, which had no such check.
+        config_with_rules(vec![rule])
+            .validate(&ObjectLockConfiguration::default())
+            .await
+            .expect_err("a negative retention count on a transition must be rejected");
+    }
+
+    #[tokio::test]
+    async fn zero_newer_noncurrent_versions_means_no_count_constraint() {
+        // Zero carries no constraint, matching how the batch limit path has
+        // always read it. Alongside an age condition the rule is valid; on its
+        // own it says nothing, so the rule has no action.
+        config_with_rules(vec![rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: Some(30),
+            newer_noncurrent_versions: Some(0),
+        })])
+        .validate(&ObjectLockConfiguration::default())
+        .await
+        .expect("zero count alongside NoncurrentDays is valid");
+
+        let err = config_with_rules(vec![rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: None,
+            newer_noncurrent_versions: Some(0),
+        })])
+        .validate(&ObjectLockConfiguration::default())
+        .await
+        .expect_err("a zero count on its own is not an action");
+        assert_eq!(err.to_string(), ERR_LIFECYCLE_RULE_MUST_HAVE_ACTION);
+    }
+
+    #[tokio::test]
+    async fn validate_accepts_count_only_noncurrent_expiration() {
+        // MinIO extension: NewerNoncurrentVersions with no NoncurrentDays. It
+        // used to be rejected as an actionless rule (backlog#2201).
+        let lc = config_with_rules(vec![rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: None,
+            newer_noncurrent_versions: Some(2),
+        })]);
+
+        lc.validate(&ObjectLockConfiguration::default())
+            .await
+            .expect("a count-only noncurrent expiration rule is accepted");
+    }
+
+    #[tokio::test]
+    async fn eval_inner_expires_versions_beyond_count_only_retention() {
+        // Count-only rules have no age condition: everything past the retained
+        // count is due as soon as it became noncurrent.
+        let lc = config_with_rules(vec![rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: None,
+            newer_noncurrent_versions: Some(2),
+        })]);
+        let opts = ObjectOpts {
+            name: "obj".to_string(),
+            mod_time: Some(datetime!(2025-01-15 10:30:45 UTC)),
+            successor_mod_time: Some(datetime!(2025-01-15 10:30:45 UTC)),
+            is_latest: false,
+            num_versions: 5,
+            ..Default::default()
+        };
+
+        // Rank 2 is the third-newest noncurrent version: past a retention of 2.
+        let expired = lc.eval_inner(&opts, datetime!(2025-01-15 10:30:46 UTC), 2).await;
+        assert_eq!(expired.action, IlmAction::DeleteVersionAction);
+        assert_eq!(expired.rule_id, "noncurrent");
+
+        // Rank 1 is still within the retained count.
+        let retained = lc.eval_inner(&opts, datetime!(2025-01-15 10:30:46 UTC), 1).await;
+        assert_eq!(retained.action, IlmAction::NoneAction);
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn eval_inner_keeps_age_condition_when_count_and_days_are_set() {
+        // With both set, the count gates which versions are candidates and the
+        // age condition still decides when they are due.
+        with_default_ilm_process_time(|| {});
+        let lc = config_with_rules(vec![rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: Some(10),
+            newer_noncurrent_versions: Some(1),
+        })]);
+        let opts = ObjectOpts {
+            name: "obj".to_string(),
+            mod_time: Some(datetime!(2025-01-01 00:00:00 UTC)),
+            successor_mod_time: Some(datetime!(2025-01-01 00:00:00 UTC)),
+            is_latest: false,
+            num_versions: 3,
+            ..Default::default()
+        };
+
+        let too_young = lc.eval_inner(&opts, datetime!(2025-01-05 00:00:00 UTC), 2).await;
+        assert_eq!(too_young.action, IlmAction::NoneAction, "the age condition still applies");
+
+        let due = lc.eval_inner(&opts, datetime!(2025-01-20 00:00:00 UTC), 2).await;
+        assert_eq!(due.action, IlmAction::DeleteVersionAction);
+    }
+
+    #[tokio::test]
+    async fn eval_inner_takes_no_action_for_a_corrupt_retention_count() {
+        // Reachable only from older persistence or an import; it must not be
+        // read as "retain everything", and it must not delete either.
+        let lc = config_with_rules(vec![rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: Some(1),
+            newer_noncurrent_versions: Some(-1),
+        })]);
+        let opts = ObjectOpts {
+            name: "obj".to_string(),
+            mod_time: Some(datetime!(2025-01-01 00:00:00 UTC)),
+            successor_mod_time: Some(datetime!(2025-01-01 00:00:00 UTC)),
+            is_latest: false,
+            num_versions: 3,
+            ..Default::default()
+        };
+
+        let event = lc.eval_inner(&opts, datetime!(2025-06-01 00:00:00 UTC), 2).await;
+
+        assert_eq!(event.action, IlmAction::NoneAction);
+    }
+
+    #[tokio::test]
+    async fn eval_inner_does_not_expire_latest_object_for_a_corrupt_retention_rule() {
+        let mut rule = rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: Some(1),
+            newer_noncurrent_versions: Some(-1),
+        });
+        rule.expiration = Some(LifecycleExpiration {
+            days: Some(1),
+            ..Default::default()
+        });
+        let lc = config_with_rules(vec![rule]);
+        let opts = ObjectOpts {
+            name: "obj".to_string(),
+            mod_time: Some(datetime!(2025-01-01 00:00:00 UTC)),
+            is_latest: true,
+            ..Default::default()
+        };
+
+        let event = lc.eval_inner(&opts, datetime!(2025-06-01 00:00:00 UTC), 0).await;
+
+        assert_eq!(event.action, IlmAction::NoneAction);
+    }
+
+    #[tokio::test]
+    async fn eval_inner_does_not_delete_latest_marker_for_a_corrupt_retention_rule() {
+        let mut expired_marker_rule = rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: Some(1),
+            newer_noncurrent_versions: Some(-1),
+        });
+        expired_marker_rule.expiration = Some(LifecycleExpiration {
+            expired_object_delete_marker: Some(true),
+            ..Default::default()
+        });
+
+        let mut aged_marker_rule = rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: Some(1),
+            newer_noncurrent_versions: Some(-1),
+        });
+        aged_marker_rule.del_marker_expiration = Some(s3s::dto::DelMarkerExpiration { days: Some(1) });
+
+        for rule in [expired_marker_rule, aged_marker_rule] {
+            let lc = config_with_rules(vec![rule]);
+            let opts = ObjectOpts {
+                name: "obj".to_string(),
+                mod_time: Some(datetime!(2025-01-01 00:00:00 UTC)),
+                version_id: Some(Uuid::new_v4()),
+                is_latest: true,
+                delete_marker: true,
+                num_versions: 1,
+                ..Default::default()
+            };
+
+            let event = lc.eval_inner(&opts, datetime!(2025-06-01 00:00:00 UTC), 0).await;
+
+            assert_eq!(event.action, IlmAction::NoneAction);
+        }
+    }
+
+    #[test]
+    fn corrupt_retention_count_is_detected_on_either_action() {
+        let mut transition_rule = rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: Some(1),
+            newer_noncurrent_versions: Some(0),
+        });
+        transition_rule.noncurrent_version_transitions = Some(vec![NoncurrentVersionTransition {
+            newer_noncurrent_versions: Some(-1),
+            noncurrent_days: Some(1),
+            storage_class: Some(TransitionStorageClass::from_static(TransitionStorageClass::GLACIER)),
+        }]);
+
+        assert!(lifecycle_has_corrupt_retention_count(&config_with_rules(vec![
+            rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+                noncurrent_days: Some(1),
+                newer_noncurrent_versions: Some(-1),
+            })
+        ])));
+        assert!(lifecycle_has_corrupt_retention_count(&config_with_rules(vec![transition_rule])));
+        assert!(!lifecycle_has_corrupt_retention_count(&config_with_rules(vec![
+            rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+                noncurrent_days: Some(1),
+                newer_noncurrent_versions: Some(3),
+            })
+        ])));
+    }
+
+    #[test]
+    fn count_only_rules_are_active_only_for_a_positive_count() {
+        let positive = config_with_rules(vec![rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: None,
+            newer_noncurrent_versions: Some(2),
+        })]);
+        assert!(positive.has_active_rules(""));
+
+        let corrupt = config_with_rules(vec![rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: None,
+            newer_noncurrent_versions: Some(-1),
+        })]);
+        assert!(!corrupt.has_active_rules(""), "a corrupt retention count must not make a rule active");
+    }
+
+    #[tokio::test]
+    async fn noncurrent_versions_expiration_limit_ignores_a_corrupt_count() {
+        // The batch path must not read a negative count as "retain everything".
+        let lc = Arc::new(config_with_rules(vec![rule_with_noncurrent_expiration(NoncurrentVersionExpiration {
+            noncurrent_days: Some(1),
+            newer_noncurrent_versions: Some(-1),
+        })]));
+        let opts = ObjectOpts {
+            name: "obj".to_string(),
+            mod_time: Some(datetime!(2025-01-01 00:00:00 UTC)),
+            is_latest: false,
+            ..Default::default()
+        };
+
+        let event = lc.noncurrent_versions_expiration_limit(&opts).await;
+
+        assert_eq!(event.action, IlmAction::NoneAction);
+        assert_eq!(event.newer_noncurrent_versions, 0);
+    }
+
+    #[tokio::test]
+    async fn validate_covers_filter_invariants() {
+        struct Case {
+            name: &'static str,
+            filter: LifecycleRuleFilter,
+            expected: Option<(&'static str, std::io::ErrorKind)>,
+        }
+
+        let cases = vec![
+            Case {
+                // AWS documents an empty Filter as "every object in the bucket".
+                name: "empty filter applies to all objects",
+                filter: LifecycleRuleFilter::default(),
+                expected: None,
+            },
+            Case {
+                name: "single prefix predicate",
+                filter: LifecycleRuleFilter {
+                    prefix: Some("logs/".to_string()),
+                    ..Default::default()
+                },
+                expected: None,
+            },
+            Case {
+                name: "two top-level predicates",
+                filter: LifecycleRuleFilter {
+                    prefix: Some("logs/".to_string()),
+                    tag: Some(tag("env", "prod")),
+                    ..Default::default()
+                },
+                expected: Some((ERR_LIFECYCLE_FILTER_TOO_MANY_PREDICATES, LIFECYCLE_MALFORMED_XML_ERROR_KIND)),
+            },
+            Case {
+                name: "prefix alongside And",
+                filter: LifecycleRuleFilter {
+                    prefix: Some("logs/".to_string()),
+                    and: Some(LifecycleRuleAndOperator {
+                        prefix: Some("logs/".to_string()),
+                        tags: Some(vec![tag("env", "prod")]),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+                expected: Some((ERR_LIFECYCLE_FILTER_TOO_MANY_PREDICATES, LIFECYCLE_MALFORMED_XML_ERROR_KIND)),
+            },
+            Case {
+                name: "And with a single member",
+                filter: LifecycleRuleFilter {
+                    and: Some(LifecycleRuleAndOperator {
+                        prefix: Some("logs/".to_string()),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+                expected: Some((ERR_LIFECYCLE_FILTER_AND_TOO_FEW_PREDICATES, LIFECYCLE_MALFORMED_XML_ERROR_KIND)),
+            },
+            Case {
+                name: "And with two members",
+                filter: LifecycleRuleFilter {
+                    and: Some(LifecycleRuleAndOperator {
+                        prefix: Some("logs/".to_string()),
+                        tags: Some(vec![tag("env", "prod")]),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+                expected: None,
+            },
+            Case {
+                name: "And with two tags",
+                filter: LifecycleRuleFilter {
+                    and: Some(LifecycleRuleAndOperator {
+                        tags: Some(vec![tag("env", "prod"), tag("team", "storage")]),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+                expected: None,
+            },
+            Case {
+                name: "And repeating a tag key",
+                filter: LifecycleRuleFilter {
+                    and: Some(LifecycleRuleAndOperator {
+                        tags: Some(vec![tag("env", "prod"), tag("env", "dev")]),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+                expected: Some((ERR_LIFECYCLE_FILTER_DUPLICATE_TAG_KEY, std::io::ErrorKind::Other)),
+            },
+            Case {
+                name: "empty tag key",
+                filter: LifecycleRuleFilter {
+                    tag: Some(tag("", "prod")),
+                    ..Default::default()
+                },
+                expected: Some((ERR_LIFECYCLE_FILTER_INVALID_TAG, std::io::ErrorKind::Other)),
+            },
+            Case {
+                name: "missing tag key",
+                filter: LifecycleRuleFilter {
+                    tag: Some(s3s::dto::Tag {
+                        key: None,
+                        value: Some("prod".to_string()),
+                    }),
+                    ..Default::default()
+                },
+                expected: Some((ERR_LIFECYCLE_FILTER_INVALID_TAG, std::io::ErrorKind::Other)),
+            },
+            Case {
+                name: "missing tag value",
+                filter: LifecycleRuleFilter {
+                    tag: Some(s3s::dto::Tag {
+                        key: Some("env".to_string()),
+                        value: None,
+                    }),
+                    ..Default::default()
+                },
+                expected: Some((ERR_LIFECYCLE_FILTER_INVALID_TAG, std::io::ErrorKind::Other)),
+            },
+            Case {
+                name: "empty tag value",
+                filter: LifecycleRuleFilter {
+                    tag: Some(tag("env", "")),
+                    ..Default::default()
+                },
+                expected: None,
+            },
+            Case {
+                name: "tag key at the limit",
+                filter: LifecycleRuleFilter {
+                    tag: Some(tag(&"k".repeat(MAX_TAG_KEY_LEN), "prod")),
+                    ..Default::default()
+                },
+                expected: None,
+            },
+            Case {
+                name: "tag key past the limit",
+                filter: LifecycleRuleFilter {
+                    tag: Some(tag(&"k".repeat(MAX_TAG_KEY_LEN + 1), "prod")),
+                    ..Default::default()
+                },
+                expected: Some((ERR_LIFECYCLE_FILTER_INVALID_TAG, std::io::ErrorKind::Other)),
+            },
+            Case {
+                name: "tag value past the limit",
+                filter: LifecycleRuleFilter {
+                    tag: Some(tag("env", &"v".repeat(MAX_TAG_VALUE_LEN + 1))),
+                    ..Default::default()
+                },
+                expected: Some((ERR_LIFECYCLE_FILTER_INVALID_TAG, std::io::ErrorKind::Other)),
+            },
+            Case {
+                name: "negative ObjectSizeGreaterThan",
+                filter: LifecycleRuleFilter {
+                    object_size_greater_than: Some(-1),
+                    ..Default::default()
+                },
+                expected: Some((ERR_LIFECYCLE_FILTER_NEGATIVE_SIZE, std::io::ErrorKind::Other)),
+            },
+            Case {
+                name: "negative ObjectSizeLessThan",
+                filter: LifecycleRuleFilter {
+                    object_size_less_than: Some(-5),
+                    ..Default::default()
+                },
+                expected: Some((ERR_LIFECYCLE_FILTER_NEGATIVE_SIZE, std::io::ErrorKind::Other)),
+            },
+            Case {
+                name: "inverted size range inside And",
+                filter: LifecycleRuleFilter {
+                    and: Some(LifecycleRuleAndOperator {
+                        object_size_greater_than: Some(100),
+                        object_size_less_than: Some(100),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+                expected: Some((ERR_LIFECYCLE_FILTER_SIZE_RANGE, std::io::ErrorKind::Other)),
+            },
+            Case {
+                name: "valid size range inside And",
+                filter: LifecycleRuleFilter {
+                    and: Some(LifecycleRuleAndOperator {
+                        object_size_greater_than: Some(1),
+                        object_size_less_than: Some(2),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+                expected: None,
+            },
+        ];
+
+        for case in cases {
+            let result = config_with_rules(vec![rule_with_filter(case.filter)])
+                .validate(&ObjectLockConfiguration::default())
+                .await;
+            match (case.expected, result) {
+                (None, Ok(())) => {}
+                (None, Err(err)) => panic!("{}: expected acceptance, got {err}", case.name),
+                (Some((message, _)), Ok(())) => panic!("{}: expected rejection with {message}", case.name),
+                (Some((message, kind)), Err(err)) => {
+                    assert_eq!(err.to_string(), message, "{}", case.name);
+                    assert_eq!(err.kind(), kind, "{}: wrong S3 error category", case.name);
+                }
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn validate_keeps_legacy_prefix_and_filter_mutually_exclusive() {
+        let mut rule = rule_with_filter(LifecycleRuleFilter {
+            prefix: Some("logs/".to_string()),
+            ..Default::default()
+        });
+        rule.prefix = Some("legacy/".to_string());
+
+        let err = config_with_rules(vec![rule])
+            .validate(&ObjectLockConfiguration::default())
+            .await
+            .expect_err("legacy Prefix and Filter cannot both be present");
+
+        assert_eq!(err.to_string(), ERR_LIFECYCLE_PREFIX_FILTER_CONFLICT);
+    }
+
+    #[test]
+    fn count_only_rule_round_trips_through_xml() {
+        // The MinIO count-only form has to survive the wire codec, or the rule
+        // this PR now accepts could not be persisted and read back.
+        let xml = br#"<LifecycleConfiguration><Rule><ID>count-only</ID><Status>Enabled</Status><Filter></Filter><NoncurrentVersionExpiration><NewerNoncurrentVersions>2</NewerNoncurrentVersions></NoncurrentVersionExpiration></Rule></LifecycleConfiguration>"#;
+        let mut deserializer = s3s::xml::Deserializer::new(xml);
+        let parsed =
+            <BucketLifecycleConfiguration as XmlDeserialize>::deserialize(&mut deserializer).expect("count-only XML parses");
+
+        let expiration = parsed.rules[0]
+            .noncurrent_version_expiration
+            .as_ref()
+            .expect("noncurrent expiration is present");
+        assert_eq!(expiration.newer_noncurrent_versions, Some(2));
+        assert_eq!(expiration.noncurrent_days, None);
+
+        let mut buf = Vec::new();
+        let mut serializer = s3s::xml::Serializer::new(&mut buf);
+        XmlSerializeContent::serialize_content(&parsed, &mut serializer).expect("count-only config serializes");
+        let serialized = String::from_utf8(buf).expect("serialized XML is UTF-8");
+        assert!(
+            serialized.contains("<NewerNoncurrentVersions>2</NewerNoncurrentVersions>"),
+            "retention count survives the round trip: {serialized}"
+        );
+        assert!(
+            !serialized.contains("<NoncurrentDays>"),
+            "a count-only rule must not gain an age condition: {serialized}"
+        );
+    }
+
     mod adversarial_regressions {
         use super::*;
         use s3s::dto::NoncurrentVersionExpiration;

crates/lifecycle/src/evaluator.rs

@@ -22,7 +22,10 @@ use rustfs_replication::ReplicationStatusType;
 use rustfs_scanner_metrics::metrics::IlmAction;
 
 use crate::object_lock;
-use crate::{Event, Lifecycle, ObjectOpts, expiration_action_has_valid_target};
+use crate::{
+    Event, LIFECYCLE_CORRUPT_RULE_ERROR_KIND, Lifecycle, ObjectOpts, expiration_action_has_valid_target,
+    lifecycle_has_corrupt_retention_count,
+};
 
 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
 const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle";
@@ -152,6 +155,17 @@ impl Evaluator {
                 format!("number of versions mismatch, expected {}, got {}", objs[0].num_versions, objs.len()),
             ));
         }
+        // PUT validation rejects a negative retention count, so a rule that
+        // carries one came from older persistence or an import. Report it
+        // instead of evaluating a configuration that cannot be honoured;
+        // `eval_inner` independently takes no action for such a rule
+        // (backlog#2201).
+        if lifecycle_has_corrupt_retention_count(&self.policy) {
+            return Err(std::io::Error::new(
+                LIFECYCLE_CORRUPT_RULE_ERROR_KIND,
+                "lifecycle configuration carries a negative 'NewerNoncurrentVersions'",
+            ));
+        }
         Ok(self.eval_inner(objs, OffsetDateTime::now_utc()).await)
     }
 }

docs/operations/scanner-benchmark-runbook.md

@@ -34,142 +34,6 @@ The `scanner` and `heal` subsystems are served by `GetConfigKVHandler` (`rustfs/
 
 ## Test Matrix
 
-### Formal Scanner/Heal ABBA
-
-The `--abba` mode runs five independent scenario cells: `cold-hot`, `fresh-hot`,
-`multi-hot-new`, `running-heal`, and `mrf-replay`. Each scenario runs at least
-three A1/B1/B2/A2 groups for both baseline/candidate with background work on,
-and candidate-only background off/on. A measured leg lasts at least 900
-seconds; the minimum matrix contains 120 legs (30 hours before setup/oracles).
-The existing `performance-ab.yml` supplies the pattern for immutable build
-provenance and failure propagation, but its short Warp workload is not this
-scanner gate. No scheduled workflow starts this matrix automatically.
-
-```bash
-scripts/run_scanner_validation_harness.sh --abba \
-  --manifest scanner-abba.json --adapter /path/to/isolated-deployment-adapter \
-  --out-dir /path/to/new-artifacts --data-root /path/to/new-test-data
-```
-
-Both roots must be new and non-overlapping. Every leg receives a unique data
-directory. The runner checks disk capacity before each leg, never removes data,
-and stops the adapter after success or failure. Retain raw artifacts and inspect
-task ownership before removing any test data. The operator must reserve the
-target machines and map the assigned directory to separate data paths on every
-node; the runner cannot prove remote isolation from local path names.
-
-The manifest has the following JSON contract (all fields are required):
-
-| Field | Value |
-|---|---|
-| `schema`, `evidence` | `1`, and `measured` or `synthetic`. |
-| `rounds`, `duration_seconds`, `min_free_bytes` | 3..10 groups, 900..86400 seconds for measured runs, and the independently estimated free-space reservation in bytes. Synthetic runs may use 1 second. |
-| `baseline`, `candidate` | Each contains executable `binary`, full 40-character `revision`, and verified `sha256`. The runner rehashes binaries before every leg. |
-| `fixed` | `config_sha256`, `dataset_sha256`, `release_flags`, `durability`, `disk_type`, `cache_state`, `load_command`, `resource_isolation`, `topology` (`EC8+4`), and positive `offered_load_ops`. Hashes use 64 lowercase hexadecimal characters. |
-| `oracles` | A map with all five scenario names. Each value contains positive integer `objects`, `versions`, `bytes`, and `sha256` of the independently prepared canonical object/version/content manifest. |
-| `expected_healed_objects` | A map with all five scenario names and independently seeded repair counts. Running-heal and MRF-replay require a positive count. |
-
-Record exact build flags and effective durability settings, not just defaults.
-Use deterministic workload seeds so every isolated leg has the same expected
-object/version/content result. Fix the foreground arrival rate (offered load),
-cache preparation procedure, configuration, and hardware across every leg.
-Do not include credentials in the manifest, adapter output, or saved commands;
-the collector reads `RUSTFS_ACCESS_KEY` and `RUSTFS_SECRET_KEY` from its environment.
-
-#### Deployment Adapter Contract
-
-The runner invokes an executable as `adapter ACTION request.json response.json`
-with no shell evaluation. Actions are separate processes: `prepare`, `measure`,
-`oracle`, and `stop`. Every action must return zero and write a JSON object of
-at most 1 MiB. Logs are kept separately and require an operator-managed disk
-quota. Missing output, timeout, nonzero exit, unknown/missing metrics, zero
-samples, and request errors fail the run. Adapters must terminate their own
-children on failure and `stop` must be idempotent even after partial preparation.
-
-The request contains the fixed manifest fields, selected build, scenario, round,
-leg, comparison (`build` or `background`), background mode (`on` or `off`),
-duration, unique `data_dir`, expected object oracle, and expected repair count.
-Adapter responsibilities:
-
-1. `prepare` deploys the selected binary into an authorized isolated topology,
-   checks actual binary/config/durability, initializes deterministic scenario
-   data and the requested cache state, and returns `{"ready": true}`. For measured
-   runs it also returns `collector` with exactly `alias`, `endpoint`, and
-   comma-separated `metrics_endpoints`; the runner starts the existing scanner
-   collector at 60-second cadence while `measure` runs.
-2. `measure` maintains the fixed offered load for the entire requested duration.
-   `cold-hot` retains cold buckets while mutating a hot bucket; `fresh-hot`
-   creates a bucket after scanner startup; `multi-hot-new` combines several hot
-   buckets with a newly created bucket; `running-heal` applies foreground load
-   during active repair; `mrf-replay` replays independently seeded durable repair
-   work. Capture same-window status for bucket-freshness issue #7108. Actual
-   fault injection and dataset generation belong to the reviewed adapter.
-3. `oracle` independently enumerates all objects and versions, reads and checks
-   their complete bytes, and verifies repairs. Return `complete: true`, integer
-   `errors: 0`, and `actual` matching the manifest's expected oracle. Never copy
-   expected values into a measured oracle or infer completion from empty queues.
-4. `stop` stops task-owned workload/server processes and returns `stopped: true`.
-   Preserve data and artifacts for diagnosis. An adapter may restore previous
-   settings but must not delete arbitrary paths or stop unrelated deployments.
-
-The `measure` response echoes the observed `evidence`, `fixed`, `build`,
-`data_dir`, and `background`, plus `sample_count` (1..3600), `elapsed_seconds`,
-and `metrics`. All metrics must be finite nonnegative numbers: `p99_ms`,
-`throughput_ops`, `rss_bytes`, `cpu_seconds`, `iops`, `rpc_count`,
-`cache_clone_bytes`, `encode_bytes`, `save_bytes`, `oldest_age_seconds`,
-`walk_objects`, `cold_walk_objects`, `healed_objects`, `errors`, and `requests`.
-Requests, throughput, and p99 must be positive; errors must be zero. Repair
-counts must match the manifest when background work is on. Keep underlying
-request samples, counter reset checks, profiler captures, and per-node telemetry
-in the cell artifact directory; aggregate values alone do not establish their
-measurement provenance. Missing production instrumentation is a pending gate,
-not permission to report a fabricated zero.
-
-For P2, `measure.convergence` contains booleans `writes_stopped`,
-`last_mutation_observed`, `first_complete_publication`; numeric
-`last_mutation_time`, `last_mutation_observed_time`, `writes_stopped_time`, `window_start`, `window_end`,
-`budget_available_seconds`, `walk_objects`, and `full_walk_objects`. Times use
-one monotonic clock. The window starts after writes stop and the final mutation
-is observed, and ends at the first complete publication. The reference is an
-independent full walk of the same static namespace. Record available budget
-seconds to interpret elapsed time. During continuing writes, omit this proof
-and report useful-work ratio and justified invalidation/re-scan work separately;
-the runner reports P2 pending and does not impose a fixed cumulative walk bound.
-
-The nightly heal workflow clones **`rustfs/auto-testing`** separately and invokes
-`auto-testing/rustfs_heal_test.sh`; that script is not a local `scripts/test`
-entry point. If an adapter uses it, record and verify the external checkout's
-owner and full commit before use. The current workflow clones the default branch,
-so its contents must not be attributed to a RustFS source SHA.
-
-#### Evidence Gates
-
-`report.json` records each group's verdict and the raw responses remain in their
-cell directories. Candidate/build p99 regression must be at most 5% and
-throughput loss at most 3%; candidate background on/off limits are 10% and 5%.
-P1 requires cold-hot walk reduction of at least the baseline cold-walk share
-times 80%, rather than a fixed 80% reduction for every workload. P2 requires
-candidate post-stop work at most 1.2 times the independent full-walk reference.
-Missing candidate convergence proof yields `inconclusive`. A2/A1 or B2/B1 p99
-or throughput drift above 5% also yields `inconclusive`, with exit code 3.
-Correctness errors and non-noisy performance regressions exit 1. Every group
-must pass; a favorable median cannot hide a failing group.
-
-Synthetic success is explicitly `synthetic_validated`, with `performance:
-pending`. It validates orchestration and gate logic only. It proves no runtime,
-distributed, crash, mixed-version, or performance behavior and cannot close the
-performance acceptance gate. Run the fake-adapter self-tests with:
-
-```bash
-scripts/test_scanner_validation_harness.sh
-```
-
-They cover the complete 120-cell schedule, data isolation, missing builds and
-oracles, zero samples/requests, swallowed request errors, offered-load drift,
-incomplete repairs, missing metrics, noise, and P1/P2/p99 regressions. A real
-deployment adapter and actual ABBA artifacts remain required before any measured
-performance or release claim.
-
 Collect at least two runs on the same RustFS commit and the same workload. Keep hardware, commit, object count, object size, bucket count, scanner-enabled state, and foreground workload constant between runs.
 
 | Run | Purpose | Example scanner settings |

rustfs/src/app/bucket_usecase.rs

@@ -27,8 +27,8 @@ use super::storage_api::bucket_usecase::bucket::target::BucketTarget;
 use super::storage_api::bucket_usecase::bucket::{
     ObjectLockConfigExt as _, VersioningConfigExt as _,
     lifecycle::bucket_lifecycle_ops::{
-        enqueue_expiry_for_existing_objects, enqueue_transition_for_existing_objects, run_stale_multipart_upload_cleanup_once,
-        validate_lifecycle_config, validate_transition_tier,
+        LIFECYCLE_MALFORMED_XML_ERROR_KIND, enqueue_expiry_for_existing_objects, enqueue_transition_for_existing_objects,
+        run_stale_multipart_upload_cleanup_once, validate_lifecycle_config, validate_transition_tier,
     },
     metadata::{
         BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_NOTIFICATION_CONFIG, BUCKET_POLICY_CONFIG,
@@ -1188,6 +1188,21 @@ fn validate_lifecycle_rule_status(rules: &[LifecycleRule]) -> std::result::Resul
     Ok(())
 }
 
+/// Map a lifecycle validation failure onto the S3 error the client should see.
+///
+/// The validator reports a schema-shape violation (a `Filter` with more than
+/// one predicate, a one-member `And`) with
+/// [`LIFECYCLE_MALFORMED_XML_ERROR_KIND`]; AWS answers those with
+/// `MalformedXML`. Everything else is a value the schema allows but S3 refuses,
+/// which stays `InvalidArgument` — the code this path has always returned
+/// (backlog#2201).
+fn lifecycle_validation_error(err: &std::io::Error) -> S3Error {
+    if err.kind() == LIFECYCLE_MALFORMED_XML_ERROR_KIND {
+        return S3Error::with_message(S3ErrorCode::MalformedXML, format!("Malformed XML: {err}"));
+    }
+    s3_error!(InvalidArgument, "{err}")
+}
+
 fn lifecycle_has_transition_rules(config: &BucketLifecycleConfiguration) -> bool {
     config.rules.iter().any(|rule| {
         rule.status == ExpirationStatus::from_static(ExpirationStatus::ENABLED)
@@ -2296,7 +2311,7 @@ impl DefaultBucketUsecase {
         };
 
         if let Err(err) = validate_lifecycle_config(&input_cfg, &rcfg).await {
-            return Err(s3_error!(InvalidArgument, "{err}"));
+            return Err(lifecycle_validation_error(&err));
         }
 
         if let Err(err) = validate_transition_tier(&input_cfg).await {
@@ -4030,6 +4045,70 @@ mod tests {
         assert_eq!(rules[2].id.as_deref(), Some("rule-2"));
     }
 
+    #[tokio::test]
+    async fn put_bucket_lifecycle_validation_errors_keep_their_s3_code() {
+        // The PUT path answers a schema-shape violation with MalformedXML and a
+        // rejected value with InvalidArgument. Both categories are produced by
+        // the real validator here, so the mapping cannot drift from it
+        // (backlog#2201).
+        let malformed = validate_lifecycle_config(
+            &BucketLifecycleConfiguration {
+                expiry_updated_at: None,
+                rules: vec![LifecycleRule {
+                    status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                    expiration: Some(LifecycleExpiration {
+                        days: Some(1),
+                        ..Default::default()
+                    }),
+                    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()),
+                        }),
+                        ..Default::default()
+                    }),
+                    id: Some("two-predicates".to_string()),
+                    noncurrent_version_expiration: None,
+                    noncurrent_version_transitions: None,
+                    prefix: None,
+                    transitions: None,
+                }],
+            },
+            &ObjectLockConfiguration::default(),
+        )
+        .await
+        .expect_err("a Filter with two predicates is a schema violation");
+        assert_eq!(*lifecycle_validation_error(&malformed).code(), S3ErrorCode::MalformedXML);
+
+        let invalid_value = validate_lifecycle_config(
+            &BucketLifecycleConfiguration {
+                expiry_updated_at: None,
+                rules: vec![LifecycleRule {
+                    status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                    expiration: None,
+                    abort_incomplete_multipart_upload: None,
+                    del_marker_expiration: None,
+                    filter: None,
+                    id: Some("negative-count".to_string()),
+                    noncurrent_version_expiration: Some(s3s::dto::NoncurrentVersionExpiration {
+                        noncurrent_days: Some(30),
+                        newer_noncurrent_versions: Some(-1),
+                    }),
+                    noncurrent_version_transitions: None,
+                    prefix: None,
+                    transitions: None,
+                }],
+            },
+            &ObjectLockConfiguration::default(),
+        )
+        .await
+        .expect_err("a negative retention count is rejected");
+        assert_eq!(*lifecycle_validation_error(&invalid_value).code(), S3ErrorCode::InvalidArgument);
+    }
+
     #[test]
     fn validate_lifecycle_rule_status_rejects_invalid_status() {
         let rules = vec![LifecycleRule {

rustfs/src/app/storage_api.rs

@@ -398,6 +398,12 @@ pub(crate) mod bucket {
 
                 lc.validate(lock_config).await
             }
+
+            /// The `std::io::ErrorKind` [`validate_lifecycle_config`] uses for a
+            /// lifecycle document that violates the published schema shape, which
+            /// the S3 boundary answers with `MalformedXML` (backlog#2201).
+            pub(crate) const LIFECYCLE_MALFORMED_XML_ERROR_KIND: std::io::ErrorKind =
+                crate::storage::storage_api::ecstore_bucket::lifecycle::lifecycle::LIFECYCLE_MALFORMED_XML_ERROR_KIND;
         }
 
         pub(crate) mod lifecycle_contract {

scripts/README.md

@@ -54,8 +54,6 @@ their issue closes.
 | `probe.sh` | dev-tool | Probe-style e2e run | `make probe-e2e` |
 | `run_scanner_validation_harness.sh` | dev-tool | Scanner validation harness | `docs/operations/scanner-benchmark-runbook.md` |
 | `test_scanner_validation_harness.sh` | dev-tool | Self-test for the scanner validation harness | — |
-| `scanner_abba.py` | dev-tool | Scanner/heal ABBA orchestration and evidence gates via `run_scanner_validation_harness.sh --abba` | `docs/operations/scanner-benchmark-runbook.md` |
-| `test_scanner_abba.py` | dev-tool | Synthetic ABBA adapter and failure-path tests | `test_scanner_validation_harness.sh` |
 | `test_build_rustfs_options.sh` | dev-tool | Shell test for rustfs build-option wiring | `make test` (script-tests) |
 | `test_entrypoint_credentials.sh` | dev-tool | Container entrypoint credential-handling test | `make test` (script-tests) |
 | `test_helm_chart_version.sh` | dev-tool | Test for `helm_chart_version.sh` | — |

scripts/run_scanner_validation_harness.sh

@@ -1,12 +1,6 @@
 #!/usr/bin/env bash
 set -euo pipefail
 
-if [[ "${1:-}" == "--abba" ]]; then
-  shift
-  SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
-  exec "$SCRIPT_DIR/python_bin.sh" "$SCRIPT_DIR/scanner_abba.py" "$@"
-fi
-
 ALIAS=""
 ENDPOINT=""
 ACCESS_KEY="${RUSTFS_ACCESS_KEY:-}"
@@ -29,7 +23,6 @@ TELEMETRY_PIDS=()
 usage() {
   cat <<'USAGE'
 Usage:
-  scripts/run_scanner_validation_harness.sh --abba --help
   scripts/run_scanner_validation_harness.sh --alias <admin-alias> \
     --endpoint <url> [options]
 

scripts/scanner_abba.py

@@ -1,384 +0,0 @@
-#!/usr/bin/env python3
-"""Run isolated scanner/heal ABBA cells through a deployment-specific adapter."""
-
-import argparse
-from decimal import Decimal
-import hashlib
-import json
-import math
-import os
-from pathlib import Path
-import shutil
-import signal
-import subprocess
-import sys
-import time
-
-SCENARIOS = ("cold-hot", "fresh-hot", "multi-hot-new", "running-heal", "mrf-replay")
-LEGS = ("A1", "B1", "B2", "A2")
-MAX_JSON_BYTES = 1024 * 1024
-METRICS = (
-    "p99_ms", "throughput_ops", "rss_bytes", "cpu_seconds", "iops", "rpc_count",
-    "cache_clone_bytes", "encode_bytes", "save_bytes", "oldest_age_seconds",
-    "walk_objects", "cold_walk_objects", "healed_objects", "errors", "requests",
-)
-REPEATABILITY_LIMIT = Decimal("0.05")
-P2_WORK_MULTIPLE_LIMIT = Decimal("1.2")
-
-
-def require(condition, message):
-    if not condition:
-        raise ValueError(message)
-
-
-def number(value, name, minimum=0):
-    require(type(value) in (float, int) and math.isfinite(value) and value >= minimum,
-            f"invalid {name}")
-    return value
-
-
-def decimal_number(value, name, minimum=0):
-    if isinstance(value, Decimal):
-        require(value.is_finite() and value >= Decimal(str(minimum)), f"invalid {name}")
-        return value
-    number(value, name, minimum)
-    return Decimal(str(value))
-
-
-def ratio(numerator, denominator, name):
-    denominator = decimal_number(denominator, f"{name} denominator")
-    require(denominator > 0, f"invalid {name} denominator")
-    return decimal_number(numerator, name) / denominator
-
-
-def relative_change(current, baseline, name):
-    return ratio(current, baseline, name) - Decimal("1")
-
-
-def repeatability_change(first, second, name):
-    first = decimal_number(first, name)
-    second = decimal_number(second, name)
-    if first == 0 and second == 0:
-        return Decimal("0")
-    if first == 0 or second == 0:
-        return Decimal("Infinity")
-    return abs(second / first - Decimal("1"))
-
-
-def report_number(value):
-    return None if value.is_infinite() else float(value)
-
-
-def digest(path):
-    with Path(path).open("rb") as stream:
-        return hashlib.file_digest(stream, "sha256").hexdigest()
-
-
-def read_json(path):
-    require(path.stat().st_size <= MAX_JSON_BYTES, f"oversized JSON: {path.name}")
-    with path.open() as stream:
-        value = json.load(stream)
-    require(isinstance(value, dict), f"expected JSON object: {path.name}")
-    return value
-
-
-def write_json(path, value):
-    data = json.dumps(value, indent=2, allow_nan=False) + "\n"
-    require(len(data.encode()) <= MAX_JSON_BYTES, "oversized result")
-    path.write_text(data)
-
-
-def sha(value):
-    return isinstance(value, str) and len(value) == 64 and all(c in "0123456789abcdef" for c in value)
-
-
-def validate_manifest(manifest):
-    require(manifest.get("schema") == 1, "unsupported manifest schema")
-    require(manifest.get("evidence") in ("synthetic", "measured"), "missing evidence type")
-    fixed = manifest["fixed"]
-    for key in ("config_sha256", "dataset_sha256"):
-        require(sha(fixed.get(key)), f"invalid fixed.{key}")
-    for key in ("release_flags", "durability", "disk_type", "cache_state", "load_command", "resource_isolation"):
-        require(isinstance(fixed.get(key), str) and fixed[key].strip(), f"missing fixed.{key}")
-    require(fixed.get("topology") == "EC8+4", "formal matrix requires EC8+4")
-    number(fixed.get("offered_load_ops"), "offered load", 1)
-    require(type(manifest.get("rounds")) is int and 3 <= manifest["rounds"] <= 10,
-            "rounds must be 3..10")
-    minimum = 900 if manifest["evidence"] == "measured" else 1
-    require(type(manifest.get("duration_seconds")) is int and
-            minimum <= manifest["duration_seconds"] <= 86400, "invalid duration_seconds")
-    number(manifest.get("min_free_bytes"), "min_free_bytes", 1)
-    for phase in ("baseline", "candidate"):
-        build = manifest[phase]
-        path = Path(build["binary"]).resolve(strict=True)
-        require(path.is_file() and os.access(path, os.X_OK), f"missing executable {phase} build")
-        require(sha(build.get("sha256")) and digest(path) == build["sha256"], f"{phase} binary hash mismatch")
-        require(isinstance(build.get("revision"), str) and len(build["revision"]) == 40 and
-                all(c in "0123456789abcdef" for c in build["revision"]), f"invalid {phase} revision")
-        build["binary"] = str(path)
-    for scenario in SCENARIOS:
-        expected = manifest["oracles"][scenario]
-        for key in ("objects", "versions", "bytes"):
-            require(type(expected.get(key)) is int and expected[key] > 0, f"missing {scenario} oracle {key}")
-        require(sha(expected.get("sha256")), f"missing {scenario} content/version digest")
-        number(manifest["expected_healed_objects"].get(scenario), f"{scenario} expected repairs")
-        if scenario in ("running-heal", "mrf-replay"):
-            require(manifest["expected_healed_objects"][scenario] > 0, f"{scenario} requires repairs")
-
-
-def invoke(adapter, action, request, timeout):
-    """The adapter writes bounded JSON separately; stderr/stdout remain raw evidence."""
-    output = request.parent / f"{action}.json"
-    with (request.parent / f"{action}.log").open("wb") as log:
-        process = subprocess.Popen([str(adapter), action, str(request), str(output)],
-                                   stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
-        try:
-            returncode = process.wait(timeout=timeout)
-            if returncode:
-                raise subprocess.CalledProcessError(returncode, [str(adapter), action])
-        finally:
-            if process.poll() != 0:
-                try:
-                    os.killpg(process.pid, signal.SIGTERM)
-                except ProcessLookupError:
-                    pass
-                try:
-                    process.wait(timeout=10)
-                except subprocess.TimeoutExpired:
-                    os.killpg(process.pid, signal.SIGKILL)
-                    process.wait()
-    return read_json(output)
-
-
-def validate_result(result, request, expected):
-    require(result.get("evidence") == request["evidence"], "adapter evidence type mismatch")
-    require(result.get("fixed") == request["fixed"], "offered load/config/cache/durability drift")
-    require(result.get("build") == request["build"], "deployed build provenance mismatch")
-    require(result.get("data_dir") == request["data_dir"], "adapter data isolation mismatch")
-    require(result.get("background") == request["background"], "background mode mismatch")
-    require(type(result.get("sample_count")) is int and 1 <= result["sample_count"] <= 3600,
-            "sample_count must be 1..3600")
-    number(result.get("elapsed_seconds"), "elapsed_seconds", request["duration_seconds"])
-    metrics = result["metrics"]
-    for key in METRICS:
-        number(metrics.get(key), key)
-    for key in ("requests", "p99_ms", "throughput_ops"):
-        require(metrics[key] > 0, f"zero {key}")
-    require(metrics["errors"] == 0, "workload request errors")
-    require(metrics["cold_walk_objects"] <= metrics["walk_objects"], "cold walk exceeds total walk")
-    require(result.get("oracle") == expected, "object/version/byte oracle mismatch")
-    if request["background"] == "on":
-        require(metrics["walk_objects"] > 0, "zero background walk")
-        require(metrics["healed_objects"] == request["expected_healed_objects"], "incomplete repair oracle")
-        if request["scenario"] in ("running-heal", "mrf-replay"):
-            require(metrics["healed_objects"] > 0, "zero completed repairs")
-    return result
-
-
-def convergence(result):
-    window = result.get("convergence")
-    if not window or window.get("writes_stopped") is not True or window.get("last_mutation_observed") is not True or window.get("first_complete_publication") is not True:
-        return None
-    for key in ("last_mutation_time", "last_mutation_observed_time", "writes_stopped_time", "window_start", "window_end", "walk_objects", "full_walk_objects", "budget_available_seconds"):
-        number(window.get(key), f"convergence.{key}")
-    require(window["last_mutation_time"] <= window["writes_stopped_time"] <= window["window_start"] < window["window_end"],
-            "invalid post-mutation convergence window")
-    require(window["last_mutation_time"] <= window["last_mutation_observed_time"] <= window["window_start"],
-            "convergence started before last mutation was observed")
-    require(window["full_walk_objects"] > 0, "zero full walk reference")
-    require(0 < window["budget_available_seconds"] <= window["window_end"] - window["window_start"],
-            "invalid convergence budget window")
-    return window["walk_objects"] / window["full_walk_objects"]
-
-
-def evaluate(cells):
-    comparisons = []
-    inconclusive = False
-    failed = False
-    for offset in range(0, len(cells), 4):
-        group = cells[offset:offset + 4]
-        require([cell["leg"] for cell in group] == list(LEGS), "incomplete ABBA group")
-        a1, b1, b2, a2 = (cell["result"]["metrics"] for cell in group)
-        control = group[0]["comparison"] == "background"
-        drift = max(abs(relative_change(a2[k], a1[k], k)) for k in ("p99_ms", "throughput_ops"))
-        repeat_drift = max(abs(relative_change(b2[k], b1[k], k)) for k in ("p99_ms", "throughput_ops"))
-        noise = max(drift, repeat_drift) > REPEATABILITY_LIMIT
-        a = {key: (decimal_number(a1[key], key) + decimal_number(a2[key], key)) / Decimal("2") for key in METRICS}
-        b = {key: (decimal_number(b1[key], key) + decimal_number(b2[key], key)) / Decimal("2") for key in METRICS}
-        p99 = relative_change(b["p99_ms"], a["p99_ms"], "p99_ms")
-        throughput = relative_change(b["throughput_ops"], a["throughput_ops"], "throughput_ops")
-        thresholds = {"p99_regression": Decimal("0.10") if control else Decimal("0.05"),
-                      "throughput_loss": Decimal("0.05") if control else Decimal("0.03")}
-        passed = p99 <= thresholds["p99_regression"] and throughput >= -thresholds["throughput_loss"]
-        p1 = None
-        work_drift = None
-        if not control:
-            if group[0]["scenario"] == "cold-hot":
-                require(a["cold_walk_objects"] > 0, "cold-hot baseline has no cold walk samples")
-            work_drift = max(repeatability_change(a1[key], a2[key], key) for key in ("walk_objects", "cold_walk_objects"))
-            work_drift = max(work_drift, *(repeatability_change(b1[key], b2[key], key) for key in ("walk_objects", "cold_walk_objects")))
-            noise |= work_drift > REPEATABILITY_LIMIT
-            required = ratio(a["cold_walk_objects"], a["walk_objects"], "cold walk baseline") * Decimal("0.80")
-            reduction = Decimal("1") - ratio(b["walk_objects"], a["walk_objects"], "walk reduction")
-            p1 = {"required_reduction": float(required), "observed_reduction": float(reduction),
-                  "repeatability_drift": report_number(work_drift)}
-            if group[0]["scenario"] == "cold-hot":
-                # Compare counts before division can round repeating decimal ratios.
-                passed &= a["walk_objects"] - b["walk_objects"] >= a["cold_walk_objects"] * Decimal("0.80")
-        p2 = [convergence(cell["result"]) if cell["background"] == "on" else None for cell in group]
-        candidate_p2 = [value for cell, value in zip(group, p2) if cell["leg"].startswith("B")]
-        p2_pending = any(value is None for value in candidate_p2)
-        passed &= all(ratio(value, 1, "p2 work multiple") <= P2_WORK_MULTIPLE_LIMIT for value in candidate_p2 if value is not None)
-        inconclusive |= noise or p2_pending
-        if not noise and not passed:
-            failed = True
-        comparisons.append({"scenario": group[0]["scenario"], "comparison": group[0]["comparison"],
-                            "round": group[0]["round"], "status": "inconclusive" if noise else ("fail" if not passed else "inconclusive" if p2_pending else "pass"),
-                            "a2_a1_drift": report_number(drift), "b2_b1_drift": report_number(repeat_drift),
-                            "p99_regression": float(p99), "throughput_change": float(throughput),
-                            "thresholds": {key: float(value) for key, value in thresholds.items()},
-                            "p1": p1, "p2_max_work_multiple": float(P2_WORK_MULTIPLE_LIMIT),
-                            "p2_post_stop_work_multiples": p2})
-    return ("fail" if failed else "inconclusive" if inconclusive else "pass"), comparisons
-
-
-def collect_live(prepared, request, request_path, adapter):
-    collector = Path(__file__).with_name("run_scanner_validation_harness.sh")
-    # Only allow connection fields here; the runner owns cadence and output paths.
-    connection = prepared["collector"]
-    require(set(connection) == {"alias", "endpoint", "metrics_endpoints"}, "invalid collector connection")
-    require(all(isinstance(value, str) and value for value in connection.values()), "missing collector endpoint")
-    output = request_path.parent / "telemetry"
-    args = ["bash", str(collector), "--alias", connection["alias"], "--endpoint", connection["endpoint"],
-            "--metrics-endpoints", connection["metrics_endpoints"], "--deployment", "distributed",
-            "--samples", str(request["duration_seconds"] // 60 + 1), "--interval-secs", "60",
-            "--out-dir", str(output)]
-    with (request_path.parent / "collector.log").open("wb") as log:
-        process = subprocess.Popen(args, stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
-        try:
-            started = time.monotonic()
-            result = invoke(adapter, "measure", request_path, request["duration_seconds"] + 300)
-            require(time.monotonic() - started >= request["duration_seconds"], "measurement ended before required window")
-            require(process.wait(timeout=120) == 0, "scanner collector failed")
-            require(output.joinpath("scanner-summary.csv").stat().st_size > 0, "missing collector samples")
-            samples = list((output / "status").glob("scanner-status.*.json"))
-            require(len(samples) == request["duration_seconds"] // 60 + 1, "missing scanner samples")
-            for sample in samples:
-                status = read_json(sample)
-                require(isinstance(status.get("metrics"), dict) and status["metrics"], "invalid scanner status response")
-            heals = list((output / "heal").glob("background-heal-status.*.json"))
-            require(bool(heals), "missing heal samples")
-            for sample in heals:
-                status = read_json(sample)
-                require(isinstance(status.get("healOperations"), dict) and status["healOperations"], "invalid heal status response")
-            metrics = list((output / "metrics").glob("admin-metrics.*.ndjson"))
-            endpoints = [endpoint for endpoint in connection["metrics_endpoints"].split(",") if endpoint]
-            require(metrics and len(metrics) == len(endpoints) * len(samples), "missing distributed metrics samples")
-            for sample in metrics:
-                # The collector requests n=1, so each file contains one final JSON record.
-                status = read_json(sample)
-                require(status.get("errors") == [], "distributed metrics errors")
-                require(status.get("final") is True, "incomplete distributed metrics")
-                hosts = status.get("by_host")
-                require(isinstance(hosts, dict) and hosts, "missing by-host metrics")
-                for host in hosts.values():
-                    require(isinstance(host, dict) and isinstance(host.get("scanner"), dict) and host["scanner"],
-                            "missing per-host scanner metrics")
-            return result
-        finally:
-            # Stop telemetry children as well when measurement fails or times out.
-            try:
-                os.killpg(process.pid, signal.SIGTERM)
-            except ProcessLookupError:
-                pass
-            try:
-                process.wait(timeout=10)
-            except subprocess.TimeoutExpired:
-                os.killpg(process.pid, signal.SIGKILL)
-                process.wait()
-
-
-def run(manifest, adapter, output, data_root):
-    validate_manifest(manifest)
-    require(adapter.is_file() and os.access(adapter, os.X_OK), "missing executable adapter")
-    require(not output.exists() and not data_root.exists(), "output/data root must be new; existing data is preserved")
-    require(output != data_root and output not in data_root.parents and data_root not in output.parents,
-            "output and data roots must not overlap")
-    output.mkdir(parents=True)
-    data_root.mkdir(parents=True)
-    require(shutil.disk_usage(data_root).free >= manifest["min_free_bytes"], "insufficient free disk space")
-    manifest["adapter_sha256"] = digest(adapter)
-    manifest["collector_sha256"] = digest(Path(__file__).with_name("run_scanner_validation_harness.sh"))
-    write_json(output / "manifest.json", manifest)
-    cells = []
-    write_json(output / "report.json", {"status": "incomplete", "performance": "pending"})
-    try:
-        for scenario in SCENARIOS:
-            for comparison in ("build", "background"):
-                for round_id in range(1, manifest["rounds"] + 1):
-                    for leg in LEGS:
-                        phase = "baseline" if comparison == "build" and leg.startswith("A") else "candidate"
-                        background = "off" if comparison == "background" and leg.startswith("A") else "on"
-                        name = f"{scenario}-{comparison}-{round_id}-{leg}"
-                        cell_dir = output / name
-                        cell_dir.mkdir()
-                        data_dir = data_root / name
-                        data_dir.mkdir()
-                        request = {"schema": 1, "scenario": scenario, "comparison": comparison, "round": round_id,
-                                   "leg": leg, "background": background, "build": manifest[phase],
-                                   "evidence": manifest["evidence"], "fixed": manifest["fixed"],
-                                   "duration_seconds": manifest["duration_seconds"], "data_dir": str(data_dir),
-                                   "expected_healed_objects": manifest["expected_healed_objects"][scenario],
-                                   "expected_oracle": manifest["oracles"][scenario]}
-                        require(digest(Path(request["build"]["binary"])) == request["build"]["sha256"], "binary changed during run")
-                        require(digest(adapter) == manifest["adapter_sha256"], "adapter changed during run")
-                        require(shutil.disk_usage(data_root).free >= manifest["min_free_bytes"], "insufficient free disk space")
-                        request_path = cell_dir / "request.json"
-                        write_json(request_path, request)
-                        print(name, flush=True)
-                        try:
-                            prepared = invoke(adapter, "prepare", request_path, 300)
-                            require(prepared.get("ready") is True, "deployment not ready")
-                            if manifest["evidence"] == "measured":
-                                result = collect_live(prepared, request, request_path, adapter)
-                            else:
-                                result = invoke(adapter, "measure", request_path, 300)
-                            # An independent operation must enumerate all object versions and bytes.
-                            oracle = invoke(adapter, "oracle", request_path, 300)
-                            require(oracle.get("complete") is True and oracle.get("errors") == 0, "correctness oracle failed")
-                            require(type(oracle.get("errors")) is int, "invalid oracle error count")
-                            result["oracle"] = oracle["actual"]
-                            validate_result(result, request, request["expected_oracle"])
-                            cells.append({**request, "result": result})
-                        finally:
-                            stopped = invoke(adapter, "stop", request_path, 300)
-                            require(stopped.get("stopped") is True, "adapter failed to stop deployment")
-        status, comparisons = evaluate(cells)
-        synthetic = manifest["evidence"] == "synthetic"
-        report = {"status": "synthetic_validated" if synthetic and status == "pass" else status,
-                  "evidence": manifest["evidence"], "performance": "pending" if synthetic else status,
-                  "cells": len(cells), "comparisons": comparisons}
-        write_json(output / "report.json", report)
-        return 0 if status == "pass" else 3 if status == "inconclusive" else 1
-    except (ValueError, KeyError, OSError, subprocess.SubprocessError) as error:
-        write_json(output / "report.json", {"status": "failed", "performance": "pending",
-                                            "completed_cells": len(cells), "error": str(error)})
-        raise
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--manifest", type=Path, required=True)
-    parser.add_argument("--adapter", type=Path, required=True)
-    parser.add_argument("--out-dir", type=Path, required=True)
-    parser.add_argument("--data-root", type=Path, required=True)
-    args = parser.parse_args()
-    try:
-        return run(read_json(args.manifest), args.adapter.resolve(), args.out_dir.resolve(), args.data_root.resolve())
-    except (ValueError, KeyError, OSError, subprocess.SubprocessError) as error:
-        print(f"ERROR: {error}", file=sys.stderr)
-        return 1
-
-
-if __name__ == "__main__":
-    sys.exit(main())

scripts/test_scanner_abba.py

@@ -1,264 +0,0 @@
-#!/usr/bin/env python3
-"""Synthetic adapter and failure-propagation tests; never start a RustFS server."""
-
-import contextlib
-import copy
-import io
-import json
-import os
-from pathlib import Path
-import subprocess
-import sys
-import tempfile
-import unittest
-from unittest.mock import Mock, patch
-
-import scanner_abba as harness
-
-
-def fake_adapter():
-    action, request_path, output_path = sys.argv[1:]
-    request = harness.read_json(Path(request_path))
-    fault = os.environ.get("SCANNER_ABBA_TEST_FAULT", "")
-    if action == "prepare":
-        result = {"ready": True}
-    elif action == "stop":
-        result = {"stopped": True}
-    elif action == "oracle":
-        if fault == "oracle-exit":
-            return 42
-        if fault == "missing-oracle":
-            return 0
-        result = {"complete": True, "errors": 0, "actual": request["expected_oracle"]}
-        if fault == "oracle-mismatch":
-            result["actual"]["bytes"] += 1
-    else:
-        if fault == "measure-exit":
-            return 42
-        result = {key: request[key] for key in ("evidence", "fixed", "build", "data_dir", "background")}
-        result.update({"sample_count": 10, "elapsed_seconds": request["duration_seconds"],
-                       "metrics": dict.fromkeys(harness.METRICS, 10)})
-        baseline = request["comparison"] == "build" and request["leg"].startswith("A")
-        result["metrics"].update(p99_ms=10, throughput_ops=100, errors=0, requests=100,
-                                 walk_objects=100 if baseline else 20, cold_walk_objects=100 if baseline else 0,
-                                 healed_objects=request["expected_healed_objects"])
-        result["convergence"] = {"writes_stopped": True, "last_mutation_observed": True,
-                                 "first_complete_publication": True, "last_mutation_time": 1,
-                                 "last_mutation_observed_time": 2,
-                                 "writes_stopped_time": 2, "window_start": 2, "window_end": 3,
-                                 "budget_available_seconds": 1, "walk_objects": 110, "full_walk_objects": 100}
-        if fault == "zero-samples":
-            result["sample_count"] = 0
-        elif fault == "request-errors":
-            result["metrics"]["errors"] = 1
-        elif fault == "load-drift":
-            result["fixed"]["offered_load_ops"] += 1
-        elif fault == "noise" and request["leg"] == "A2":
-            result["metrics"]["p99_ms"] = 20
-        elif fault == "zero-requests":
-            result["metrics"]["requests"] = 0
-        elif fault == "no-publication":
-            result["convergence"]["first_complete_publication"] = False
-        elif fault == "p2-regression":
-            result["convergence"]["walk_objects"] = 121
-        elif fault == "latency-regression" and request["leg"].startswith("B"):
-            result["metrics"]["p99_ms"] = 12
-        elif fault == "exact-thresholds" and request["leg"].startswith("B"):
-            result["metrics"].update(p99_ms=10.5, throughput_ops=97)
-        elif fault == "just-over-threshold" and request["comparison"] == "build" and request["leg"].startswith("B"):
-            result["metrics"]["p99_ms"] = 10.500001
-        elif fault == "p1-regression" and not baseline:
-            result["metrics"]["walk_objects"] = 30
-        elif fault in ("p1-exact-fraction", "p1-over-fraction"):
-            result["metrics"].update(walk_objects=9 if baseline else 5 + (fault == "p1-over-fraction"),
-                                     cold_walk_objects=5 if baseline else 0)
-        elif fault == "unstable-p1-control" and request["comparison"] == "build":
-            if request["leg"] == "A1":
-                result["metrics"].update(walk_objects=1000, cold_walk_objects=1000)
-            elif request["leg"] == "A2":
-                result["metrics"].update(walk_objects=10, cold_walk_objects=10)
-            elif request["leg"].startswith("B"):
-                result["metrics"].update(walk_objects=100, cold_walk_objects=0)
-        elif fault == "missing-metric":
-            del result["metrics"]["save_bytes"]
-        elif fault == "incomplete-repair":
-            result["metrics"]["healed_objects"] = 0
-    harness.write_json(Path(output_path), result)
-    return 0
-
-
-class ScannerAbbaTest(unittest.TestCase):
-    def setUp(self):
-        self.temp = tempfile.TemporaryDirectory()
-        self.addCleanup(self.temp.cleanup)
-        self.root = Path(self.temp.name)
-        self.binary = Path(sys.executable).resolve()
-        self.adapter = Path(__file__).resolve()
-        self.manifest = {
-            "schema": 1, "evidence": "synthetic", "rounds": 3, "duration_seconds": 1, "min_free_bytes": 1,
-            "fixed": {"config_sha256": "1" * 64, "dataset_sha256": "2" * 64,
-                      "release_flags": "--release", "durability": "drive-sync=on",
-                      "disk_type": "synthetic", "cache_state": "cold", "load_command": "fake",
-                      "topology": "EC8+4", "offered_load_ops": 100, "resource_isolation": "synthetic"},
-            "oracles": {s: {"objects": 10, "versions": 20, "bytes": 30, "sha256": "3" * 64} for s in harness.SCENARIOS},
-            "expected_healed_objects": {s: 10 for s in harness.SCENARIOS},
-        }
-        build = {"binary": str(self.binary), "sha256": harness.digest(self.binary), "revision": "a" * 40}
-        self.manifest.update(baseline=build.copy(), candidate=build.copy())
-
-    def run_harness(self, fault=""):
-        with patch.dict(os.environ, {"SCANNER_ABBA_TEST_FAULT": fault}), contextlib.redirect_stdout(io.StringIO()):
-            return harness.run(copy.deepcopy(self.manifest), self.adapter, self.root / "out", self.root / "data")
-
-    def test_complete_synthetic_matrix_is_not_performance_evidence(self):
-        self.assertEqual(self.run_harness(), 0)
-        report = harness.read_json(self.root / "out/report.json")
-        self.assertEqual((report["status"], report["performance"], report["cells"]), ("synthetic_validated", "pending", 120))
-        requests = [harness.read_json(path) for path in (self.root / "out").glob("*/request.json")]
-        self.assertEqual(len({r["data_dir"] for r in requests}), 120)
-        for scenario in harness.SCENARIOS:
-            for comparison in ("build", "background"):
-                for round_id in (1, 2, 3):
-                    legs = [r for r in requests if (r["scenario"], r["comparison"], r["round"]) == (scenario, comparison, round_id)]
-                    self.assertEqual({r["leg"] for r in legs}, set(harness.LEGS))
-        self.assertTrue(all(c["p2_max_work_multiple"] == 1.2 for c in report["comparisons"]))
-
-    def test_fail_closed_adapter_and_data_errors(self):
-        for fault in ("measure-exit", "oracle-exit", "missing-oracle", "oracle-mismatch", "zero-samples",
-                      "zero-requests", "request-errors", "load-drift", "missing-metric", "incomplete-repair"):
-            with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory:
-                self.root = Path(directory)
-                with self.assertRaises((ValueError, OSError, subprocess.SubprocessError)):
-                    self.run_harness(fault)
-                report = harness.read_json(self.root / "out/report.json")
-                self.assertEqual(report["status"], "failed")
-                self.assertTrue(list((self.root / "out").glob("*/stop.json")))
-
-    def test_noise_is_inconclusive_and_nonzero(self):
-        with patch.object(harness, "SCENARIOS", ("cold-hot",)):
-            self.assertEqual(self.run_harness("noise"), 3)
-        self.assertEqual(harness.read_json(self.root / "out/report.json")["status"], "inconclusive")
-
-    def test_missing_first_publication_is_inconclusive(self):
-        with patch.object(harness, "SCENARIOS", ("cold-hot",)):
-            self.assertEqual(self.run_harness("no-publication"), 3)
-
-    def test_performance_regressions_fail(self):
-        for fault in ("p1-regression", "p2-regression", "latency-regression"):
-            with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory:
-                self.root = Path(directory)
-                with patch.object(harness, "SCENARIOS", ("cold-hot",)):
-                    self.assertEqual(self.run_harness(fault), 1)
-
-    def test_exact_threshold_boundaries_pass(self):
-        with patch.object(harness, "SCENARIOS", ("cold-hot",)):
-            self.assertEqual(self.run_harness("exact-thresholds"), 0)
-
-    def test_just_over_threshold_fails(self):
-        with patch.object(harness, "SCENARIOS", ("cold-hot",)):
-            self.assertEqual(self.run_harness("just-over-threshold"), 1)
-
-    def test_p1_fractional_boundary(self):
-        for fault, expected in (("p1-exact-fraction", 0), ("p1-over-fraction", 1)):
-            with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory:
-                self.root = Path(directory)
-                with patch.object(harness, "SCENARIOS", ("cold-hot",)):
-                    self.assertEqual(self.run_harness(fault), expected)
-
-    def test_live_collector_rejects_missing_or_failed_node_metrics(self):
-        telemetry = self.root / "telemetry"
-        for name in ("status", "heal", "metrics"):
-            (telemetry / name).mkdir(parents=True)
-        (telemetry / "scanner-summary.csv").write_text("timestamp\n")
-        valid = {"errors": [], "final": True, "by_host": {"node-b:9000": {"scanner": {"objects": 10}}}}
-        for index in range(16):
-            harness.write_json(telemetry / f"status/scanner-status.{index}.json", {"metrics": {"objects": 10}})
-            for node in ("node-a", "node-b"):
-                harness.write_json(telemetry / f"heal/background-heal-status.{node}.{index}.json",
-                                   {"healOperations": {"queueLength": 0}})
-                harness.write_json(telemetry / f"metrics/admin-metrics.{node}.{index}.ndjson",
-                                   {**valid, "by_host": {f"{node}:9000": {"scanner": {"objects": 10}}}})
-        sample = telemetry / "metrics/admin-metrics.node-b.15.ndjson"
-        prepared = {"collector": {"alias": "test", "endpoint": "http://node-a:9000",
-                                  "metrics_endpoints": "http://node-a:9000,http://node-b:9000"}}
-        cases = (
-            ("valid", valid, None),
-            ("missing", None, "missing distributed metrics samples"),
-            ("empty", "", "Expecting value"),
-            ("http-error", {"Code": "AccessDenied"}, "distributed metrics errors"),
-            ("partial-error", {**valid, "errors": ["node unavailable"]}, "distributed metrics errors"),
-            ("unfinished", {**valid, "final": False}, "incomplete distributed metrics"),
-            ("missing-host", {**valid, "by_host": {}}, "missing by-host metrics"),
-            ("missing-scanner", {**valid, "by_host": {"node-a:9000": {}}}, "missing per-host scanner metrics"),
-            ("collector-exit", valid, "scanner collector failed"),
-        )
-        for name, payload, error in cases:
-            with self.subTest(fault=name):
-                if payload is None:
-                    sample.unlink()
-                elif isinstance(payload, str):
-                    sample.write_text(payload)
-                else:
-                    harness.write_json(sample, payload)
-                process = Mock(pid=123, wait=Mock(return_value=1 if name == "collector-exit" else 0))
-                with patch.object(harness.subprocess, "Popen", return_value=process), \
-                        patch.object(harness, "invoke", return_value={"sample_count": 10}), \
-                        patch.object(harness.time, "monotonic", side_effect=(0, 900)), \
-                        patch.object(harness.os, "killpg"):
-                    if error:
-                        with self.assertRaisesRegex(ValueError, error):
-                            harness.collect_live(prepared, {"duration_seconds": 900}, self.root / "request.json", self.adapter)
-                    else:
-                        self.assertEqual(harness.collect_live(prepared, {"duration_seconds": 900},
-                                                             self.root / "request.json", self.adapter), {"sample_count": 10})
-
-    def test_unstable_p1_work_control_is_inconclusive(self):
-        with patch.object(harness, "SCENARIOS", ("cold-hot",)):
-            self.assertEqual(self.run_harness("unstable-p1-control"), 3)
-        comparison = harness.read_json(self.root / "out/report.json")["comparisons"][0]
-        self.assertEqual(comparison["status"], "inconclusive")
-        self.assertGreater(comparison["p1"]["repeatability_drift"], 0.05)
-
-    def test_manifest_rejects_missing_build_or_oracle(self):
-        for section, key in (("baseline", "binary"), ("oracles", "cold-hot")):
-            manifest = copy.deepcopy(self.manifest)
-            del manifest[section][key]
-            with self.subTest(section=section), self.assertRaises((ValueError, KeyError)):
-                harness.validate_manifest(manifest)
-
-    def test_short_measured_window_and_fewer_rounds_rejected(self):
-        self.manifest["evidence"] = "measured"
-        with self.assertRaisesRegex(ValueError, "duration_seconds"):
-            harness.validate_manifest(self.manifest)
-        self.manifest["duration_seconds"] = 900
-        self.manifest["rounds"] = 2
-        with self.assertRaisesRegex(ValueError, "rounds"):
-            harness.validate_manifest(self.manifest)
-
-    def test_existing_data_preserved(self):
-        (self.root / "data").mkdir()
-        marker = self.root / "data/keep"
-        marker.write_text("existing")
-        with self.assertRaisesRegex(ValueError, "preserved"):
-            self.run_harness()
-        self.assertEqual(marker.read_text(), "existing")
-
-    def test_invalid_or_live_write_window_does_not_claim_p2(self):
-        self.assertIsNone(harness.convergence({"convergence": {"writes_stopped": False}}))
-        with self.assertRaises(ValueError):
-            harness.convergence({"convergence": {"writes_stopped": True, "last_mutation_observed": True,
-                                                 "first_complete_publication": True}})
-
-    def test_nan_and_oversized_samples_rejected(self):
-        with self.assertRaises(ValueError):
-            harness.number(float("nan"), "latency")
-        path = self.root / "oversized.json"
-        path.write_bytes(b" " * (harness.MAX_JSON_BYTES + 1))
-        with self.assertRaisesRegex(ValueError, "oversized"):
-            harness.read_json(path)
-
-
-if __name__ == "__main__":
-    if len(sys.argv) == 4 and sys.argv[1] in ("prepare", "measure", "oracle", "stop"):
-        sys.exit(fake_adapter())
-    unittest.main()

scripts/test_scanner_validation_harness.sh

@@ -312,5 +312,3 @@ if PATH="$BIN_DIR:$PATH" "$SCRIPT" --secret-key rustfsadmin >"$secret_arg_log" 2
 fi
 
 grep -q -- 'unknown arg: --secret-key' "$secret_arg_log"
-
-"$ROOT_DIR/scripts/python_bin.sh" "$ROOT_DIR/scripts/test_scanner_abba.py"