chore: merge main into e2e prerequisites

crates/e2e_test/src/kms/mod.rs

@@ -39,9 +39,6 @@ mod kms_edge_cases_test;
 #[cfg(test)]
 mod kms_fault_recovery_test;
 
-#[cfg(test)]
-mod test_runner;
-
 #[cfg(test)]
 mod bucket_default_encryption_test;
 

crates/e2e_test/src/kms/test_runner.rs

@@ -1,499 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-#![allow(dead_code)]
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Unified KMS test suite runner
-//!
-//! This module provides a unified interface for running KMS tests with categorization,
-//! filtering, and comprehensive reporting capabilities.
-
-use crate::common::init_logging;
-use std::time::Instant;
-use tokio::time::{Duration, sleep};
-use tracing::{debug, error, info, warn};
-
-/// Test category for organization and filtering
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub enum TestCategory {
-    CoreFunctionality,
-    MultipartEncryption,
-    EdgeCases,
-    FaultRecovery,
-    Comprehensive,
-    Performance,
-}
-
-impl TestCategory {
-    pub fn as_str(&self) -> &'static str {
-        match self {
-            TestCategory::CoreFunctionality => "core-functionality",
-            TestCategory::MultipartEncryption => "multipart-encryption",
-            TestCategory::EdgeCases => "edge-cases",
-            TestCategory::FaultRecovery => "fault-recovery",
-            TestCategory::Comprehensive => "comprehensive",
-            TestCategory::Performance => "performance",
-        }
-    }
-}
-
-/// Test definition with metadata
-#[derive(Debug, Clone)]
-pub struct TestDefinition {
-    pub name: String,
-    pub description: String,
-    pub category: TestCategory,
-    pub estimated_duration: Duration,
-    pub is_critical: bool,
-}
-
-impl TestDefinition {
-    pub fn new(
-        name: impl Into<String>,
-        description: impl Into<String>,
-        category: TestCategory,
-        estimated_duration: Duration,
-        is_critical: bool,
-    ) -> Self {
-        Self {
-            name: name.into(),
-            description: description.into(),
-            category,
-            estimated_duration,
-            is_critical,
-        }
-    }
-}
-
-/// Test execution result
-#[derive(Debug, Clone)]
-pub struct TestResult {
-    pub test_name: String,
-    pub category: TestCategory,
-    pub success: bool,
-    pub duration: Duration,
-    pub error_message: Option<String>,
-}
-
-impl TestResult {
-    pub fn success(test_name: String, category: TestCategory, duration: Duration) -> Self {
-        Self {
-            test_name,
-            category,
-            success: true,
-            duration,
-            error_message: None,
-        }
-    }
-
-    pub fn failure(test_name: String, category: TestCategory, duration: Duration, error: String) -> Self {
-        Self {
-            test_name,
-            category,
-            success: false,
-            duration,
-            error_message: Some(error),
-        }
-    }
-}
-
-/// Comprehensive test suite configuration
-#[derive(Debug, Clone)]
-pub struct TestSuiteConfig {
-    pub categories: Vec<TestCategory>,
-    pub include_critical_only: bool,
-    pub max_duration: Option<Duration>,
-    pub parallel_execution: bool,
-}
-
-impl Default for TestSuiteConfig {
-    fn default() -> Self {
-        Self {
-            categories: vec![
-                TestCategory::CoreFunctionality,
-                TestCategory::MultipartEncryption,
-                TestCategory::EdgeCases,
-                TestCategory::FaultRecovery,
-                TestCategory::Comprehensive,
-            ],
-            include_critical_only: false,
-            max_duration: None,
-            parallel_execution: false,
-        }
-    }
-}
-
-/// Unified KMS test suite runner
-pub struct KMSTestSuite {
-    tests: Vec<TestDefinition>,
-    config: TestSuiteConfig,
-}
-
-impl KMSTestSuite {
-    /// Create a new test suite with default configuration
-    pub fn new() -> Self {
-        let tests = vec![
-            // Core Functionality Tests
-            TestDefinition::new(
-                "test_local_kms_end_to_end",
-                "End-to-end KMS test with all encryption types",
-                TestCategory::CoreFunctionality,
-                Duration::from_secs(60),
-                true,
-            ),
-            TestDefinition::new(
-                "test_local_kms_key_isolation",
-                "Test KMS key isolation and security",
-                TestCategory::CoreFunctionality,
-                Duration::from_secs(45),
-                true,
-            ),
-            // Multipart Encryption Tests
-            TestDefinition::new(
-                "test_local_kms_multipart_upload",
-                "Test large file multipart upload with encryption",
-                TestCategory::MultipartEncryption,
-                Duration::from_secs(120),
-                true,
-            ),
-            TestDefinition::new(
-                "test_step1_basic_single_file_encryption",
-                "Basic single file encryption test",
-                TestCategory::MultipartEncryption,
-                Duration::from_secs(30),
-                false,
-            ),
-            TestDefinition::new(
-                "test_step2_basic_multipart_upload_without_encryption",
-                "Basic multipart upload without encryption",
-                TestCategory::MultipartEncryption,
-                Duration::from_secs(45),
-                false,
-            ),
-            TestDefinition::new(
-                "test_step3_multipart_upload_with_sse_s3",
-                "Multipart upload with SSE-S3 encryption",
-                TestCategory::MultipartEncryption,
-                Duration::from_secs(60),
-                true,
-            ),
-            TestDefinition::new(
-                "test_step4_large_multipart_upload_with_encryption",
-                "Large file multipart upload with encryption",
-                TestCategory::MultipartEncryption,
-                Duration::from_secs(90),
-                false,
-            ),
-            TestDefinition::new(
-                "test_step5_all_encryption_types_multipart",
-                "All encryption types multipart test",
-                TestCategory::MultipartEncryption,
-                Duration::from_secs(120),
-                true,
-            ),
-            // Edge Cases Tests
-            TestDefinition::new(
-                "test_kms_zero_byte_file_encryption",
-                "Test encryption of zero-byte files",
-                TestCategory::EdgeCases,
-                Duration::from_secs(20),
-                false,
-            ),
-            TestDefinition::new(
-                "test_kms_single_byte_file_encryption",
-                "Test encryption of single-byte files",
-                TestCategory::EdgeCases,
-                Duration::from_secs(20),
-                false,
-            ),
-            TestDefinition::new(
-                "test_kms_multipart_boundary_conditions",
-                "Test multipart upload boundary conditions",
-                TestCategory::EdgeCases,
-                Duration::from_secs(45),
-                false,
-            ),
-            TestDefinition::new(
-                "test_kms_invalid_key_scenarios",
-                "Test invalid key scenarios",
-                TestCategory::EdgeCases,
-                Duration::from_secs(30),
-                false,
-            ),
-            TestDefinition::new(
-                "test_kms_concurrent_encryption",
-                "Test concurrent encryption operations",
-                TestCategory::EdgeCases,
-                Duration::from_secs(60),
-                false,
-            ),
-            TestDefinition::new(
-                "test_kms_key_validation_security",
-                "Test key validation security",
-                TestCategory::EdgeCases,
-                Duration::from_secs(30),
-                false,
-            ),
-            // Fault Recovery Tests
-            TestDefinition::new(
-                "test_kms_key_directory_unavailable",
-                "Test KMS when key directory is unavailable",
-                TestCategory::FaultRecovery,
-                Duration::from_secs(45),
-                false,
-            ),
-            TestDefinition::new(
-                "test_kms_corrupted_key_files",
-                "Test KMS with corrupted key files",
-                TestCategory::FaultRecovery,
-                Duration::from_secs(30),
-                false,
-            ),
-            TestDefinition::new(
-                "test_kms_multipart_upload_interruption",
-                "Test multipart upload interruption recovery",
-                TestCategory::FaultRecovery,
-                Duration::from_secs(60),
-                false,
-            ),
-            TestDefinition::new(
-                "test_kms_resource_constraints",
-                "Test KMS under resource constraints",
-                TestCategory::FaultRecovery,
-                Duration::from_secs(90),
-                false,
-            ),
-            // Comprehensive Tests
-            TestDefinition::new(
-                "test_comprehensive_kms_full_workflow",
-                "Full KMS workflow comprehensive test",
-                TestCategory::Comprehensive,
-                Duration::from_secs(300),
-                true,
-            ),
-            TestDefinition::new(
-                "test_comprehensive_stress_test",
-                "KMS stress test with large datasets",
-                TestCategory::Comprehensive,
-                Duration::from_secs(400),
-                false,
-            ),
-            TestDefinition::new(
-                "test_comprehensive_key_isolation",
-                "Comprehensive key isolation test",
-                TestCategory::Comprehensive,
-                Duration::from_secs(180),
-                false,
-            ),
-            TestDefinition::new(
-                "test_comprehensive_concurrent_operations",
-                "Comprehensive concurrent operations test",
-                TestCategory::Comprehensive,
-                Duration::from_secs(240),
-                false,
-            ),
-            TestDefinition::new(
-                "test_comprehensive_performance_benchmark",
-                "KMS performance benchmark test",
-                TestCategory::Comprehensive,
-                Duration::from_secs(360),
-                false,
-            ),
-        ];
-
-        Self {
-            tests,
-            config: TestSuiteConfig::default(),
-        }
-    }
-
-    /// Configure the test suite
-    pub fn with_config(mut self, config: TestSuiteConfig) -> Self {
-        self.config = config;
-        self
-    }
-
-    /// Filter tests based on category
-    pub fn filter_by_category(&self, category: &TestCategory) -> Vec<&TestDefinition> {
-        self.tests.iter().filter(|test| &test.category == category).collect()
-    }
-
-    /// Filter tests based on criticality
-    pub fn filter_critical_tests(&self) -> Vec<&TestDefinition> {
-        self.tests.iter().filter(|test| test.is_critical).collect()
-    }
-
-    /// Get test summary by category
-    pub fn get_category_summary(&self) -> std::collections::HashMap<TestCategory, Vec<&TestDefinition>> {
-        let mut summary = std::collections::HashMap::new();
-        for test in &self.tests {
-            summary.entry(test.category.clone()).or_insert_with(Vec::new).push(test);
-        }
-        summary
-    }
-
-    /// Run the complete test suite
-    pub async fn run_test_suite(&self) -> Vec<TestResult> {
-        init_logging();
-        info!("🚀 Starting unified KMS test suite");
-
-        let start_time = Instant::now();
-        let mut results = Vec::new();
-
-        // Filter tests based on configuration
-        let tests_to_run: Vec<&TestDefinition> = self
-            .tests
-            .iter()
-            .filter(|test| self.config.categories.contains(&test.category))
-            .filter(|test| !self.config.include_critical_only || test.is_critical)
-            .collect();
-
-        info!("📊 Test plan: {} test(s) scheduled", tests_to_run.len());
-        for (i, test) in tests_to_run.iter().enumerate() {
-            info!("  {}. {} ({})", i + 1, test.name, test.category.as_str());
-        }
-
-        // Execute tests
-        for (i, test_def) in tests_to_run.iter().enumerate() {
-            info!("🧪 Running test {}/{}: {}", i + 1, tests_to_run.len(), test_def.name);
-            info!("   📝 Description: {}", test_def.description);
-            info!("   🏷️  Category: {}", test_def.category.as_str());
-            info!("   ⏱️  Estimated duration: {:?}", test_def.estimated_duration);
-
-            let test_start = Instant::now();
-            let result = self.run_single_test(test_def).await;
-            let test_duration = test_start.elapsed();
-
-            match result {
-                Ok(_) => {
-                    info!("✅ Test passed: {} ({:.2}s)", test_def.name, test_duration.as_secs_f64());
-                    results.push(TestResult::success(test_def.name.clone(), test_def.category.clone(), test_duration));
-                }
-                Err(e) => {
-                    error!("❌ Test failed: {} ({:.2}s): {}", test_def.name, test_duration.as_secs_f64(), e);
-                    results.push(TestResult::failure(
-                        test_def.name.clone(),
-                        test_def.category.clone(),
-                        test_duration,
-                        e.to_string(),
-                    ));
-                }
-            }
-
-            // Add delay between tests to avoid resource conflicts
-            if i < tests_to_run.len() - 1 {
-                debug!("⏸️  Waiting two seconds before the next test...");
-                sleep(Duration::from_secs(2)).await;
-            }
-        }
-
-        let total_duration = start_time.elapsed();
-        self.print_test_summary(&results, total_duration);
-
-        results
-    }
-
-    /// Run a single test by dispatching to the appropriate test function
-    async fn run_single_test(&self, test_def: &TestDefinition) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-        // This is a placeholder for test dispatch logic
-        // In a real implementation, this would dispatch to actual test functions
-        warn!("⚠️  Test '{}' is not implemented in the unified runner; skipping", test_def.name);
-        Ok(())
-    }
-
-    /// Print comprehensive test summary
-    fn print_test_summary(&self, results: &[TestResult], total_duration: Duration) {
-        info!("📊 KMS test suite summary");
-        info!("⏱️  Total duration: {:.2} seconds", total_duration.as_secs_f64());
-        info!("📈 Total tests: {}", results.len());
-
-        let passed = results.iter().filter(|r| r.success).count();
-        let failed = results.iter().filter(|r| !r.success).count();
-
-        info!("✅ Passed: {}", passed);
-        info!("❌ Failed: {}", failed);
-        info!("📊 Success rate: {:.1}%", (passed as f64 / results.len() as f64) * 100.0);
-
-        // Summary by category
-        let mut category_summary: std::collections::HashMap<TestCategory, (usize, usize)> = std::collections::HashMap::new();
-        for result in results {
-            let (total, passed_count) = category_summary.entry(result.category.clone()).or_insert((0, 0));
-            *total += 1;
-            if result.success {
-                *passed_count += 1;
-            }
-        }
-
-        info!("📊 Category summary:");
-        for (category, (total, passed_count)) in category_summary {
-            info!(
-                "  🏷️  {}: {}/{} ({:.1}%)",
-                category.as_str(),
-                passed_count,
-                total,
-                (passed_count as f64 / total as f64) * 100.0
-            );
-        }
-
-        // List failed tests
-        if failed > 0 {
-            warn!("❌ Failing tests:");
-            for result in results.iter().filter(|r| !r.success) {
-                warn!("  - {}: {}", result.test_name, result.error_message.as_deref().unwrap_or("Unknown error"));
-            }
-        }
-    }
-}
-
-/// Quick test suite for critical tests only
-#[tokio::test]
-async fn test_kms_critical_suite() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    let config = TestSuiteConfig {
-        categories: vec![TestCategory::CoreFunctionality, TestCategory::MultipartEncryption],
-        include_critical_only: true,
-        max_duration: Some(Duration::from_secs(600)), // 10 minutes max
-        parallel_execution: false,
-    };
-
-    let suite = KMSTestSuite::new().with_config(config);
-    let results = suite.run_test_suite().await;
-
-    let failed_count = results.iter().filter(|r| !r.success).count();
-    if failed_count > 0 {
-        return Err(format!("Critical test suite failed: {failed_count} tests failed").into());
-    }
-
-    info!("✅ All critical tests passed");
-    Ok(())
-}
-
-/// Full comprehensive test suite
-#[tokio::test]
-async fn test_kms_full_suite() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    let suite = KMSTestSuite::new();
-    let results = suite.run_test_suite().await;
-
-    let total_tests = results.len();
-    let failed_count = results.iter().filter(|r| !r.success).count();
-    let success_rate = ((total_tests - failed_count) as f64 / total_tests as f64) * 100.0;
-
-    info!("📊 Full suite success rate: {:.1}%", success_rate);
-
-    // Allow up to 10% failure rate for non-critical tests
-    if success_rate < 90.0 {
-        return Err(format!("Test suite success rate too low: {success_rate:.1}%").into());
-    }
-
-    info!("✅ Full test suite succeeded");
-    Ok(())
-}

crates/e2e_test/src/object_lambda_test.rs

@@ -16,6 +16,7 @@ use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_lo
 use aws_sdk_s3::primitives::ByteStream;
 use http::header::{CONTENT_TYPE, HOST};
 use reqwest::StatusCode;
+use rustfs_config::{ENV_DRIVE_ACTIVE_CHECK_INTERVAL_SECS, ENV_NOTIFY_ENABLE};
 use rustfs_signer::pre_sign_v4;
 use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS;
 use s3s::Body;
@@ -976,7 +977,8 @@ async fn test_get_object_lambda_rejects_disabled_target() -> Result<(), Box<dyn
     init_logging();
 
     let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    env.start_rustfs_server_with_env(vec![], &[(ENV_NOTIFY_ENABLE, "true")])
+        .await?;
 
     let bucket = "object-lambda-e2e-disabled-target";
     let key = "input.txt";
@@ -992,17 +994,24 @@ async fn test_get_object_lambda_rejects_disabled_target() -> Result<(), Box<dyn
         .send()
         .await?;
 
-    configure_webhook_target_with_key_values(
-        &env,
-        "transformer",
-        vec![
-            ("endpoint", "http://127.0.0.1:9/transform".to_string()),
-            ("auth_token", "secret-token".to_string()),
-            ("enable", "off".to_string()),
-        ],
+    let queue_dir = format!("{}/disabled-target-queue", env.temp_dir);
+    tokio::fs::create_dir_all(&queue_dir).await?;
+    let config_url = format!("{}/rustfs/admin/v3/set-config-kv", env.url);
+    let directive = format!(
+        "notify_webhook:transformer enable=off endpoint=\"http://127.0.0.1:9/transform\" auth_token=\"secret-token\" queue_dir=\"{queue_dir}\""
+    );
+    let disable_response = signed_request(
+        http::Method::PUT,
+        &config_url,
+        &env.access_key,
+        &env.secret_key,
+        Some(directive.into_bytes()),
+        Some("text/plain"),
     )
     .await?;
-    wait_for_target_visibility(&env, "transformer").await?;
+    let disable_status = disable_response.status();
+    let disable_body = disable_response.text().await?;
+    assert_eq!(disable_status, StatusCode::OK, "failed to disable target: {disable_body}");
 
     let lambda_url = format!("{}/{}/{}?lambdaArn={}", env.url, bucket, key, urlencoding::encode(lambda_arn));
     let response = signed_request(http::Method::GET, &lambda_url, &env.access_key, &env.secret_key, None, None).await?;
@@ -1021,7 +1030,8 @@ async fn test_configure_object_lambda_target_rejects_invalid_endpoint() -> Resul
     init_logging();
 
     let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    env.start_rustfs_server_with_env(vec![], &[(ENV_NOTIFY_ENABLE, "true")])
+        .await?;
 
     let bucket = "object-lambda-e2e-invalid-endpoint";
 
@@ -1064,7 +1074,8 @@ async fn test_configure_object_lambda_notify_webhook_rejects_response_header_tim
     init_logging();
 
     let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    env.start_rustfs_server_with_env(vec![], &[(ENV_NOTIFY_ENABLE, "true")])
+        .await?;
 
     let response = send_configure_webhook_target_request(
         &env,
@@ -1173,6 +1184,8 @@ async fn test_listen_notification_fans_in_remote_node_events() -> Result<(), Box
     init_logging();
 
     let mut cluster = RustFSTestClusterEnvironment::new(2).await?;
+    cluster.set_env(ENV_NOTIFY_ENABLE, "true");
+    cluster.set_env(ENV_DRIVE_ACTIVE_CHECK_INTERVAL_SECS, "1");
     cluster.start().await?;
 
     let bucket = "listen-notification-cluster";

crates/e2e_test/src/stale_multipart_cleanup_cluster_test.rs

@@ -15,7 +15,6 @@
 use crate::common::{RustFSTestClusterEnvironment, init_logging};
 use aws_sdk_s3::error::SdkError;
 use aws_sdk_s3::primitives::ByteStream;
-use aws_sdk_s3::types::CompletedMultipartUpload;
 use tokio::time::{Duration, sleep};
 use tracing::info;
 use uuid::Uuid;
@@ -43,32 +42,18 @@ async fn list_parts_reports_missing_upload(
     }
 }
 
-async fn complete_reports_missing_upload(
+async fn multipart_listing_reports_missing_upload(
     client: &aws_sdk_s3::Client,
     bucket: &str,
     key: &str,
     upload_id: &str,
 ) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
-    let result = client
-        .complete_multipart_upload()
-        .bucket(bucket)
-        .key(key)
-        .upload_id(upload_id)
-        .multipart_upload(CompletedMultipartUpload::builder().build())
-        .send()
-        .await;
-    match result {
-        Ok(_) => Ok(false),
-        Err(SdkError::ServiceError(err)) => {
-            let code = err.err().meta().code().unwrap_or("");
-            if code == "NoSuchUpload" {
-                Ok(true)
-            } else {
-                Err(format!("unexpected complete_multipart_upload service error: code={code}, err={err:?}").into())
-            }
-        }
-        Err(err) => Err(format!("unexpected complete_multipart_upload error: {err:?}").into()),
-    }
+    let result = client.list_multipart_uploads().bucket(bucket).prefix(key).send().await?;
+
+    Ok(!result
+        .uploads()
+        .iter()
+        .any(|upload| upload.key() == Some(key) && upload.upload_id() == Some(upload_id)))
 }
 
 async fn wait_for_cleanup_on_all_nodes(
@@ -81,8 +66,8 @@ async fn wait_for_cleanup_on_all_nodes(
         let mut all_cleaned = true;
         for (idx, client) in clients.iter().enumerate() {
             let list_parts_missing = list_parts_reports_missing_upload(client, bucket, key, upload_id).await?;
-            let complete_missing = complete_reports_missing_upload(client, bucket, key, upload_id).await?;
-            if !(list_parts_missing && complete_missing) {
+            let listing_missing = multipart_listing_reports_missing_upload(client, bucket, key, upload_id).await?;
+            if !(list_parts_missing && listing_missing) {
                 info!("stale multipart still visible on node {} at attempt {}", idx, attempt + 1);
                 all_cleaned = false;
                 break;
@@ -146,6 +131,10 @@ async fn test_stale_multipart_cleanup_removes_incomplete_upload_across_cluster()
         1,
         "multipart upload should be visible before background cleanup"
     );
+    assert!(
+        !multipart_listing_reports_missing_upload(&clients[2], CLEANUP_BUCKET, &key, &upload_id).await?,
+        "multipart upload listing should contain the upload before background cleanup"
+    );
 
     wait_for_cleanup_on_all_nodes(&clients, CLEANUP_BUCKET, &key, &upload_id).await?;
 

crates/ecstore/src/runtime/instance.rs

@@ -160,6 +160,10 @@ pub struct InstanceContext {
     /// workers (scanner/heal/tier/lifecycle) without touching another instance.
     /// Replaces the process-global cancel-token static.
     background_cancel_token: OnceLock<CancellationToken>,
+    /// Serializes decommission data-movement operations with cancellation and
+    /// a subsequent restart. Readers are held across one object side effect;
+    /// the transition path takes the writer after cancelling the routine.
+    decommission_operation_gate: Arc<RwLock<()>>,
     /// Resolves object-encryption material at the application boundary.
     object_encryption_resolver: OnceLock<Arc<dyn ObjectEncryptionResolver>>,
     tier_delete_journal_recovery_stores: std::sync::Mutex<HashSet<Uuid>>,
@@ -200,6 +204,7 @@ impl InstanceContext {
             local_disk_set_drives: Arc::new(RwLock::new(Vec::new())),
             bucket_metadata_sys: std::sync::Mutex::new(None),
             background_cancel_token: OnceLock::new(),
+            decommission_operation_gate: Arc::new(RwLock::new(())),
             object_encryption_resolver: OnceLock::new(),
             tier_delete_journal_recovery_stores: std::sync::Mutex::new(HashSet::new()),
             transition_transaction_recovery_stores: std::sync::Mutex::new(HashSet::new()),
@@ -218,6 +223,10 @@ impl InstanceContext {
         self.lock_manager.clone()
     }
 
+    pub(crate) fn decommission_operation_gate(&self) -> Arc<RwLock<()>> {
+        Arc::clone(&self.decommission_operation_gate)
+    }
+
     /// Install the application-owned object-encryption resolver once.
     pub fn set_object_encryption_resolver(
         &self,

crates/ecstore/src/store/rebalance.rs

@@ -859,6 +859,7 @@ fn lifecycle_delete_all_test_failure(phase: crate::object_api::LifecycleDeleteAl
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::bucket::replication::{ReplicationStatusType, VersionPurgeStatusType};
     use crate::config::storageclass::{CLASS_RRS, CLASS_STANDARD, lookup_config_for_pools_without_env};
     use crate::disk::error::DiskError;
     use crate::layout::endpoint::Endpoint;
@@ -1423,6 +1424,14 @@ mod tests {
         }
     }
 
+    fn object_info_with_identity(unix_ts: i64, delete_marker: bool, version_id: Uuid, etag: Option<String>) -> ObjectInfo {
+        ObjectInfo {
+            version_id: Some(version_id),
+            etag,
+            ..object_info_with_mod_time(unix_ts, delete_marker)
+        }
+    }
+
     #[test]
     fn resolve_latest_object_info_candidates_returns_latest_delete_marker() {
         let candidates = vec![
@@ -1446,7 +1455,7 @@ mod tests {
     }
 
     #[test]
-    fn resolve_latest_object_info_candidates_prefers_higher_pool_idx_on_equal_mod_time() {
+    fn resolve_latest_object_info_candidates_prefers_higher_pool_idx_on_equal_mod_time_for_equivalent_candidates() {
         let candidates = vec![
             LatestObjectInfoCandidate {
                 info: Some(object_info_with_mod_time(10, false)),
@@ -1466,6 +1475,382 @@ mod tests {
         assert_eq!(idx, 1);
     }
 
+    #[test]
+    fn resolve_latest_object_info_candidates_keeps_index_fallback_for_fully_equivalent_identities() {
+        let candidates = vec![
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()))),
+                idx: 2,
+                err: None,
+            },
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()))),
+                idx: 7,
+                err: None,
+            },
+        ];
+
+        let (info, idx) = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default())
+            .expect("equivalent replicas must resolve deterministically");
+
+        assert_eq!(idx, 7);
+        assert_eq!(info.version_id, Some(Uuid::from_u128(1)));
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_equal_time_version_id_conflict() {
+        let candidates = vec![
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()))),
+                idx: 0,
+                err: None,
+            },
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(2), Some("etag-a".to_string()))),
+                idx: 1,
+                err: None,
+            },
+        ];
+
+        let err = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default())
+            .expect_err("divergent version ids must not silently resolve to the higher pool index");
+
+        assert_eq!(err, Error::ErasureReadQuorum);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_equal_time_etag_conflict() {
+        let candidates = vec![
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-old".to_string()))),
+                idx: 0,
+                err: None,
+            },
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-new".to_string()))),
+                idx: 1,
+                err: None,
+            },
+        ];
+
+        let err = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default())
+            .expect_err("divergent etags must not silently resolve to the higher pool index");
+
+        assert_eq!(err, Error::ErasureReadQuorum);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_equal_time_delete_marker_conflict() {
+        let candidates = vec![
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), None)),
+                idx: 0,
+                err: None,
+            },
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, true, Uuid::from_u128(1), Some("etag-a".to_string()))),
+                idx: 1,
+                err: None,
+            },
+        ];
+
+        let err = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default())
+            .expect_err("a delete marker tied with a live version must not be masked by the pool index");
+
+        assert_eq!(err, Error::ErasureReadQuorum);
+    }
+
+    fn assert_equal_time_identity_conflict(left: ObjectInfo, right: ObjectInfo) {
+        let err = resolve_latest_object_info_candidates(
+            vec![
+                LatestObjectInfoCandidate {
+                    info: Some(left),
+                    idx: 0,
+                    err: None,
+                },
+                LatestObjectInfoCandidate {
+                    info: Some(right),
+                    idx: 1,
+                    err: None,
+                },
+            ],
+            "bucket",
+            "object",
+            &ObjectOptions::default(),
+        )
+        .expect_err("equal-time identity divergence must fail closed");
+
+        assert_eq!(err, Error::ErasureReadQuorum);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_equal_time_payload_identity_conflicts() {
+        let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+
+        let mut data_dir = base.clone();
+        data_dir.data_dir = Some(Uuid::from_u128(2));
+        assert_equal_time_identity_conflict(base.clone(), data_dir);
+
+        let mut size = base.clone();
+        size.size = 1;
+        assert_equal_time_identity_conflict(base.clone(), size);
+
+        let mut actual_size = base.clone();
+        actual_size.actual_size = 1;
+        assert_equal_time_identity_conflict(base.clone(), actual_size);
+
+        let mut checksum = base.clone();
+        checksum.checksum = Some(bytes::Bytes::from_static(b"checksum"));
+        assert_equal_time_identity_conflict(base.clone(), checksum);
+
+        let mut parts = base.clone();
+        parts.parts = std::sync::Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
+            etag: "part-etag".to_string(),
+            number: 1,
+            size: 1,
+            ..Default::default()
+        }]);
+        assert_equal_time_identity_conflict(base.clone(), parts);
+
+        let mut transition = base;
+        transition.transitioned_object.tier = "tier-a".to_string();
+        assert_equal_time_identity_conflict(
+            object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())),
+            transition,
+        );
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_accepts_internal_metadata_aliases() {
+        let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+        let mut rustfs_alias = base.clone();
+        rustfs_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+            "x-rustfs-internal-compression".to_string(),
+            "zstd".to_string(),
+        )]));
+        let mut minio_alias = base.clone();
+        minio_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+            "X-MINIO-INTERNAL-COMPRESSION".to_string(),
+            "zstd".to_string(),
+        )]));
+
+        let (_, idx) = resolve_latest_object_info_candidates(
+            vec![
+                LatestObjectInfoCandidate {
+                    info: Some(rustfs_alias),
+                    idx: 0,
+                    err: None,
+                },
+                LatestObjectInfoCandidate {
+                    info: Some(minio_alias),
+                    idx: 1,
+                    err: None,
+                },
+            ],
+            "bucket",
+            "object",
+            &ObjectOptions::default(),
+        )
+        .expect("same-value internal aliases should resolve");
+        assert_eq!(idx, 1);
+
+        let mut dual_alias = base.clone();
+        dual_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([
+            ("x-rustfs-internal-compression".to_string(), "zstd".to_string()),
+            ("x-minio-internal-compression".to_string(), "zstd".to_string()),
+        ]));
+        let mut single_alias = base;
+        single_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+            "x-rustfs-internal-compression".to_string(),
+            "zstd".to_string(),
+        )]));
+
+        let (_, idx) = resolve_latest_object_info_candidates(
+            vec![
+                LatestObjectInfoCandidate {
+                    info: Some(dual_alias),
+                    idx: 0,
+                    err: None,
+                },
+                LatestObjectInfoCandidate {
+                    info: Some(single_alias),
+                    idx: 1,
+                    err: None,
+                },
+            ],
+            "bucket",
+            "object",
+            &ObjectOptions::default(),
+        )
+        .expect("dual-key and single-key internal metadata should resolve");
+        assert_eq!(idx, 1);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_different_internal_metadata_alias_values() {
+        let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+        let mut rustfs_alias = base.clone();
+        rustfs_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+            "x-rustfs-internal-compression".to_string(),
+            "zstd".to_string(),
+        )]));
+        let mut minio_alias = base;
+        minio_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+            "x-minio-internal-compression".to_string(),
+            "snappy".to_string(),
+        )]));
+
+        assert_equal_time_identity_conflict(rustfs_alias, minio_alias);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_preserves_dynamic_internal_metadata_identity_case() {
+        for suffix_prefix in ["replication-reset-", "replication-delete-marker-version-"] {
+            let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+            let mut rustfs_alias = base.clone();
+            rustfs_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+                format!(
+                    "X-RUSTFS-INTERNAL-{}{suffix}",
+                    suffix_prefix.to_uppercase(),
+                    suffix = "arn:aws:s3:::Bucket"
+                ),
+                "value".to_string(),
+            )]));
+            let mut minio_alias = base.clone();
+            minio_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+                format!("x-minio-internal-{suffix_prefix}arn:aws:s3:::Bucket"),
+                "value".to_string(),
+            )]));
+
+            let (_, idx) = resolve_latest_object_info_candidates(
+                vec![
+                    LatestObjectInfoCandidate {
+                        info: Some(rustfs_alias.clone()),
+                        idx: 0,
+                        err: None,
+                    },
+                    LatestObjectInfoCandidate {
+                        info: Some(minio_alias),
+                        idx: 1,
+                        err: None,
+                    },
+                ],
+                "bucket",
+                "object",
+                &ObjectOptions::default(),
+            )
+            .expect("dynamic internal aliases with the same target should resolve");
+            assert_eq!(idx, 1);
+
+            let mut different_target_case = base;
+            different_target_case.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+                format!("x-minio-internal-{suffix_prefix}arn:aws:s3:::bucket"),
+                "value".to_string(),
+            )]));
+
+            assert_equal_time_identity_conflict(rustfs_alias, different_target_case);
+        }
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_conflicting_internal_metadata_aliases_in_one_candidate() {
+        let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+        let mut first = base.clone();
+        first.user_defined = std::sync::Arc::new(std::collections::HashMap::from([
+            ("x-rustfs-internal-compression".to_string(), "zstd".to_string()),
+            ("x-minio-internal-compression".to_string(), "snappy".to_string()),
+        ]));
+        let mut second = base;
+        second.user_defined = first.user_defined.clone();
+
+        assert_equal_time_identity_conflict(first, second);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_replication_identity_conflict() {
+        let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+
+        let mut replication = base.clone();
+        replication.replication_status_internal = Some("PENDING".to_string());
+        replication.replication_status = ReplicationStatusType::Pending;
+        assert_equal_time_identity_conflict(base.clone(), replication);
+
+        let mut purge = base.clone();
+        purge.version_purge_status_internal = Some("PENDING".to_string());
+        purge.version_purge_status = VersionPurgeStatusType::Pending;
+        assert_equal_time_identity_conflict(base.clone(), purge);
+
+        let mut decision = base;
+        decision.replication_decision = "replicate".to_string();
+        assert_equal_time_identity_conflict(
+            object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())),
+            decision,
+        );
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_none_vs_unix_epoch_mod_time() {
+        let mut without_mod_time = object_info_with_identity(0, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+        without_mod_time.mod_time = None;
+        let with_unix_epoch = object_info_with_identity(0, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+
+        assert_equal_time_identity_conflict(without_mod_time, with_unix_epoch);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_ignores_older_identity_conflicts() {
+        let latest = object_info_with_identity(20, false, Uuid::from_u128(1), Some("etag-latest".to_string()));
+        let mut older = object_info_with_identity(10, true, Uuid::from_u128(2), Some("etag-old".to_string()));
+        older.data_dir = Some(Uuid::from_u128(2));
+
+        let (info, idx) = resolve_latest_object_info_candidates(
+            vec![
+                LatestObjectInfoCandidate {
+                    info: Some(latest),
+                    idx: 0,
+                    err: None,
+                },
+                LatestObjectInfoCandidate {
+                    info: Some(older),
+                    idx: 9,
+                    err: None,
+                },
+            ],
+            "bucket",
+            "object",
+            &ObjectOptions::default(),
+        )
+        .expect("older identity divergence must not affect the latest candidate");
+
+        assert_eq!(idx, 0);
+        assert_eq!(
+            info.mod_time,
+            Some(OffsetDateTime::from_unix_timestamp(20).expect("operation should succeed"))
+        );
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_ignores_not_found_pools_when_resolving() {
+        let candidates = vec![
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()))),
+                idx: 0,
+                err: None,
+            },
+            LatestObjectInfoCandidate {
+                info: None,
+                idx: 1,
+                err: Some(Error::ObjectNotFound("bucket".to_string(), "object".to_string())),
+            },
+        ];
+
+        let (info, idx) = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default())
+            .expect("not-found pools must not block resolution of found candidates");
+
+        assert_eq!(idx, 0);
+        assert_eq!(info.version_id, Some(Uuid::from_u128(1)));
+    }
+
     #[test]
     fn resolve_latest_object_info_candidates_returns_non_not_found_error() {
         let err = resolve_latest_object_info_candidates(

crates/ecstore/src/store/rebalance/support.rs

@@ -12,10 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::cmp::Ordering;
+use std::collections::HashMap;
 
 use crate::error::{Error, Result, StorageError, is_err_object_not_found, is_err_version_not_found};
 use crate::object_api::{ObjectInfo, ObjectOptions};
+use rustfs_utils::http::metadata_compat::{
+    SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX, SUFFIX_REPLICATION_RESET_ARN_PREFIX,
+    strip_internal_prefix_preserving_case,
+};
 use rustfs_utils::path::decode_dir_object;
 use time::OffsetDateTime;
 
@@ -137,37 +141,158 @@ pub(super) fn rebalance_disk_set_lookup_error(pool_idx: usize, set_idx: usize, p
     ))
 }
 
+fn latest_candidate_mod_time(candidate: &LatestObjectInfoCandidate) -> Option<OffsetDateTime> {
+    candidate
+        .info
+        .as_ref()
+        .map(|info| info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH))
+}
+
+fn same_transition_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool {
+    left.transition_version_state == right.transition_version_state
+        && left.transitioned_object.name == right.transitioned_object.name
+        && left.transitioned_object.version_id == right.transitioned_object.version_id
+        && left.transitioned_object.tier == right.transitioned_object.tier
+        && left.transitioned_object.free_version == right.transitioned_object.free_version
+        && left.transitioned_object.status == right.transitioned_object.status
+}
+
+#[derive(PartialEq, Eq)]
+struct LatestUserDefinedIdentity {
+    internal: HashMap<String, String>,
+    other: HashMap<String, String>,
+}
+
+fn normalize_internal_identity_suffix(key: &str) -> Option<String> {
+    let suffix = strip_internal_prefix_preserving_case(key)?;
+
+    for dynamic_prefix in [
+        SUFFIX_REPLICATION_RESET_ARN_PREFIX,
+        SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX,
+    ] {
+        let prefix_len = dynamic_prefix.len();
+        if let (Some(prefix), Some(remainder)) = (suffix.get(..prefix_len), suffix.get(prefix_len..))
+            && prefix.eq_ignore_ascii_case(dynamic_prefix)
+        {
+            return Some(format!("{dynamic_prefix}{remainder}"));
+        }
+    }
+
+    Some(suffix.to_lowercase())
+}
+
+fn normalize_user_defined_identity(user_defined: &HashMap<String, String>) -> Option<LatestUserDefinedIdentity> {
+    let mut identity = LatestUserDefinedIdentity {
+        internal: HashMap::with_capacity(user_defined.len()),
+        other: HashMap::with_capacity(user_defined.len()),
+    };
+
+    for (key, value) in user_defined {
+        if let Some(suffix) = normalize_internal_identity_suffix(key) {
+            if identity
+                .internal
+                .insert(suffix, value.clone())
+                .is_some_and(|previous| previous != *value)
+            {
+                return None;
+            }
+        } else {
+            identity.other.insert(key.clone(), value.clone());
+        }
+    }
+
+    Some(identity)
+}
+
+fn same_user_defined_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool {
+    match (
+        normalize_user_defined_identity(&left.user_defined),
+        normalize_user_defined_identity(&right.user_defined),
+    ) {
+        (Some(left), Some(right)) => left == right,
+        _ => false,
+    }
+}
+
+/// Pool-specific erasure geometry is intentionally excluded: `get_object_info`
+/// returns each pool's own `data_blocks`/`parity_blocks`, so those values can
+/// differ for the same object version while the selected winner still carries
+/// the chosen pool's layout. `put_object_reader` is also intentionally
+/// excluded because it is a transient request handle that `ObjectInfo::clone`
+/// drops. Every other ObjectInfo field is part of the production-visible
+/// identity and must agree before the pool index can provide a deterministic
+/// tie-break.
+fn same_latest_object_info_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool {
+    left.bucket == right.bucket
+        && left.name == right.name
+        && left.storage_class == right.storage_class
+        && left.mod_time == right.mod_time
+        && left.size == right.size
+        && left.actual_size == right.actual_size
+        && left.is_dir == right.is_dir
+        && same_user_defined_identity(left, right)
+        && left.user_tags == right.user_tags
+        && left.version_id == right.version_id
+        && left.data_dir == right.data_dir
+        && left.delete_marker == right.delete_marker
+        && same_transition_identity(left, right)
+        && left.restore_ongoing == right.restore_ongoing
+        && left.restore_expires == right.restore_expires
+        && left.parts == right.parts
+        && left.is_latest == right.is_latest
+        && left.content_type == right.content_type
+        && left.content_encoding == right.content_encoding
+        && left.expires == right.expires
+        && left.num_versions == right.num_versions
+        && left.successor_mod_time == right.successor_mod_time
+        && left.etag == right.etag
+        && left.inlined == right.inlined
+        && left.metadata_only == right.metadata_only
+        && left.version_only == right.version_only
+        && left.replication_status_internal == right.replication_status_internal
+        && left.replication_status == right.replication_status
+        && left.version_purge_status_internal == right.version_purge_status_internal
+        && left.version_purge_status == right.version_purge_status
+        && left.replication_decision == right.replication_decision
+        && left.checksum == right.checksum
+}
+
 pub(super) fn resolve_latest_object_info_candidates(
-    mut candidates: Vec<LatestObjectInfoCandidate>,
+    candidates: Vec<LatestObjectInfoCandidate>,
     bucket: &str,
     object: &str,
     opts: &ObjectOptions,
 ) -> Result<(ObjectInfo, usize)> {
-    candidates.sort_by(|a, b| {
-        let a_mod = if let Some(info) = &a.info {
-            info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
-        } else {
-            OffsetDateTime::UNIX_EPOCH
+    let latest_mod_time = candidates.iter().filter_map(latest_candidate_mod_time).max();
+
+    if let Some(latest_mod_time) = latest_mod_time {
+        let mut latest_candidates = candidates
+            .into_iter()
+            .filter(|candidate| latest_candidate_mod_time(candidate) == Some(latest_mod_time))
+            .collect::<Vec<_>>();
+
+        latest_candidates.sort_by(|left, right| right.idx.cmp(&left.idx));
+
+        let Some(winner) = latest_candidates.first() else {
+            return Err(Error::ErasureReadQuorum);
+        };
+        let Some(winner_info) = winner.info.as_ref() else {
+            return Err(Error::ErasureReadQuorum);
         };
 
-        let b_mod = if let Some(info) = &b.info {
-            info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
-        } else {
-            OffsetDateTime::UNIX_EPOCH
-        };
-
-        if a_mod == b_mod {
-            return if a.idx < b.idx { Ordering::Greater } else { Ordering::Less };
+        if latest_candidates.iter().skip(1).any(|candidate| {
+            candidate
+                .info
+                .as_ref()
+                .is_none_or(|info| !same_latest_object_info_identity(winner_info, info))
+        }) {
+            return Err(Error::ErasureReadQuorum);
         }
 
-        b_mod.cmp(&a_mod)
-    });
+        return Ok((winner_info.clone(), winner.idx));
+    }
 
     for candidate in candidates {
-        if let Some(info) = candidate.info {
-            return Ok((info, candidate.idx));
-        }
-
         if let Some(err) = candidate.err
             && !is_err_object_not_found(&err)
             && !is_err_version_not_found(&err)

crates/policy/src/policy/policy.rs

@@ -195,7 +195,8 @@ pub struct BucketPolicyArgs<'a> {
 #[derive(Serialize, Deserialize, Clone, Default, Debug)]
 #[serde(deny_unknown_fields)]
 pub struct BucketPolicy {
-    #[serde(default, rename = "Id", skip_serializing_if = "ID::is_empty")]
+    // RUSTFS_COMPAT_TODO(rustfs-6339): accept bucket policies persisted with the legacy "ID" key. Remove after migration tooling rewrites every retained legacy bucket policy.
+    #[serde(default, rename = "Id", alias = "ID", skip_serializing_if = "ID::is_empty")]
     pub id: ID,
     #[serde(rename = "Version")]
     pub version: String,
@@ -2786,7 +2787,7 @@ mod test {
         let parsed: serde_json::Value = serde_json::from_str(&json).expect("Should parse");
 
         // Verify empty fields are omitted
-        assert!(!parsed.as_object().unwrap().contains_key("ID"), "Empty ID should be omitted");
+        assert!(parsed.get("Id").is_none(), "Empty ID should be omitted");
 
         let statement = &parsed["Statement"][0];
         assert!(!statement.as_object().unwrap().contains_key("Sid"), "Empty Sid should be omitted");
@@ -2809,6 +2810,43 @@ mod test {
         assert_eq!(statement["Principal"]["AWS"], "*");
     }
 
+    #[test]
+    fn test_bucket_policy_deserializes_legacy_id() {
+        let legacy_policy = br#"{"ID":"","Version":"2012-10-17","Statement":[{"Sid":"","Effect":"Allow","Principal":{"AWS":["*"]},"Action":["s3:GetObject"],"NotAction":[],"Resource":["arn:aws:s3:::bucket/*"],"NotResource":[],"Condition":{}}]}"#;
+
+        let policy: BucketPolicy =
+            serde_json::from_slice(legacy_policy).expect("bucket policy with legacy ID should deserialize");
+        assert!(policy.id.is_empty());
+        policy.is_valid().expect("legacy bucket policy should remain valid");
+
+        let policy: BucketPolicy = serde_json::from_str(r#"{"ID":"legacy-policy","Version":"2012-10-17","Statement":[]}"#)
+            .expect("non-empty legacy ID should deserialize");
+        assert_eq!(policy.id.0, "legacy-policy");
+
+        let serialized = serde_json::to_value(&policy).expect("bucket policy should serialize");
+        assert_eq!(serialized["Id"], "legacy-policy");
+        assert!(serialized.get("ID").is_none(), "legacy ID spelling should not be serialized");
+    }
+
+    #[test]
+    fn test_bucket_policy_legacy_id_alias_remains_strict() {
+        let unknown_field = r#"{"Version":"2012-10-17","Statement":[],"Unexpected":true}"#;
+        let error =
+            serde_json::from_str::<BucketPolicy>(unknown_field).expect_err("unrelated unknown fields should remain rejected");
+        assert!(
+            error.to_string().contains("unknown field `Unexpected`"),
+            "unexpected deserialization error: {error}"
+        );
+
+        let duplicate_id = r#"{"Id":"current-policy","ID":"legacy-policy","Version":"2012-10-17","Statement":[]}"#;
+        let error = serde_json::from_str::<BucketPolicy>(duplicate_id)
+            .expect_err("canonical and legacy ID fields should not be accepted together");
+        assert!(
+            error.to_string().contains("duplicate field `Id`"),
+            "unexpected deserialization error: {error}"
+        );
+    }
+
     #[test]
     fn test_existing_object_tag_condition_helpers() {
         let identity_policy = Policy::parse_config(