// 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
//
// 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.

use super::storage_api::test::bucket::{
    lifecycle,
    metadata::{BUCKET_LIFECYCLE_CONFIG, OBJECT_LOCK_CONFIG},
    metadata_sys,
};
use super::storage_api::test::contract::{
    bucket::{BucketOperations, BucketOptions, MakeBucketOptions},
    list::ListOperations as _,
    multipart::MultipartOperations as _,
    namespace::NamespaceLocking as _,
    object::{ObjectIO as _, ObjectOperations as _},
};
use super::storage_api::test::ecfs::FS;
use super::storage_api::test::object_utils::to_s3s_etag;
use super::storage_api::test::runtime::{MockWarmBackend, MockWarmOp, register_mock_tier as register_mock_tier_util};
use super::storage_api::test::{
    ECStore, Endpoint, EndpointServerPools, Endpoints, PoolEndpoints, StorageObjectInfo as ObjectInfo,
    StorageObjectOptions as ObjectOptions, StoragePutObjReader as PutObjReader, install_all_v6_fleet_capability_proof,
};
use super::{multipart_usecase::DefaultMultipartUsecase, object_usecase::DefaultObjectUsecase};
use crate::app::bucket_usecase::DefaultBucketUsecase;
use crate::app::runtime_sources::current_tier_config_handle;
use bytes::Bytes;
use futures::FutureExt;
use futures::stream;
use http::{Extensions, HeaderMap, HeaderValue, Method, Uri, header::IF_NONE_MATCH};
use rustfs_config::{ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT};
use rustfs_object_capacity::capacity_manager::{HybridStrategyConfig, create_isolated_manager};
use rustfs_utils::http::{AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, insert_header};
use rustfs_utils::path::encode_dir_object;
use s3s::{S3Request, dto::*};
use serial_test::serial;
use std::{
    convert::Infallible,
    env, fs as stdfs,
    path::PathBuf,
    sync::{Arc, Once, OnceLock},
    time::Duration,
};
use tokio::fs;
use tokio::io::AsyncReadExt;
use tokio::sync::Barrier;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

static GLOBAL_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>)> = OnceLock::new();
static INIT: Once = Once::new();
const TRANSITION_WAIT_TIMEOUT: Duration = Duration::from_secs(15);
const RESTORE_SUSPENDED_WAIT_TIMEOUT: Duration = Duration::from_secs(15);
const ENV_GET_CODEC_STREAMING_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_ENABLE";
const ENV_GET_CODEC_STREAMING_ROLLOUT: &str = "RUSTFS_GET_CODEC_STREAMING_ROLLOUT";
const ENV_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED";
const ENV_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED";
const ENV_GET_CODEC_STREAMING_MIN_SIZE: &str = "RUSTFS_GET_CODEC_STREAMING_MIN_SIZE";
const RUSTFS_EXPECTED_CURRENT_VERSION_ID: &str = "x-rustfs-expected-current-version-id";

fn init_tracing() {
    INIT.call_once(|| {});
}

async fn setup_test_env() -> (Vec<PathBuf>, Arc<ECStore>) {
    init_tracing();

    if let Some((paths, ecstore)) = GLOBAL_ENV.get() {
        return (paths.clone(), ecstore.clone());
    }

    // Production startup installs this proof through the notification-system
    // fleet probe. This isolated single-node harness constructs ECStore
    // directly, so model that completed all-v6 probe before any journal permit
    // or background expiry worker can exist.
    install_all_v6_fleet_capability_proof();

    let test_base_dir = format!("/tmp/rustfs_app_lifecycle_test_{}", Uuid::new_v4());
    let temp_dir = PathBuf::from(&test_base_dir);
    if temp_dir.exists() {
        fs::remove_dir_all(&temp_dir).await.ok();
    }
    fs::create_dir_all(&temp_dir).await.unwrap();

    let disk_paths = vec![
        temp_dir.join("disk1"),
        temp_dir.join("disk2"),
        temp_dir.join("disk3"),
        temp_dir.join("disk4"),
    ];

    for disk_path in &disk_paths {
        fs::create_dir_all(disk_path).await.unwrap();
    }

    let mut endpoints = Vec::new();
    for (i, disk_path) in disk_paths.iter().enumerate() {
        let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap();
        endpoint.set_pool_index(0);
        endpoint.set_set_index(0);
        endpoint.set_disk_index(i);
        endpoints.push(endpoint);
    }

    let pool_endpoints = PoolEndpoints {
        legacy: false,
        set_count: 1,
        drives_per_set: 4,
        endpoints: Endpoints::from(endpoints),
        cmd_line: "test".to_string(),
        platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
    };

    let endpoint_pools = EndpointServerPools(vec![pool_endpoints]);

    super::storage_api::test::runtime::init_local_disks(endpoint_pools.clone())
        .await
        .unwrap();

    let server_addr: std::net::SocketAddr = "127.0.0.1:9003".parse().unwrap();
    let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new())
        .await
        .unwrap();

    let buckets_list = ecstore
        .list_bucket(&BucketOptions {
            no_metadata: true,
            ..Default::default()
        })
        .await
        .unwrap();
    let buckets = buckets_list.into_iter().map(|v| v.name).collect();
    metadata_sys::init_bucket_metadata_sys(ecstore.clone(), buckets).await;

    lifecycle::bucket_lifecycle_ops::init_background_expiry(ecstore.clone()).await;

    // The usecases resolve the object store through the ambient AppContext
    // since the backlog#1052 InstanceContext migration; without one installed
    // every store-touching usecase call fails with InternalError "Not init".
    // Publish a default-interface context wrapping this test store.
    crate::app::runtime_sources::install_test_app_context(ecstore.clone()).await;

    let _ = GLOBAL_ENV.set((disk_paths.clone(), ecstore.clone()));

    (disk_paths, ecstore)
}

async fn create_test_bucket(ecstore: &Arc<ECStore>, bucket_name: &str) {
    (**ecstore)
        .make_bucket(
            bucket_name,
            &MakeBucketOptions {
                versioning_enabled: true,
                ..Default::default()
            },
        )
        .await
        .expect("Failed to create test bucket");
}

async fn suspend_test_bucket(bucket: &str) {
    DefaultBucketUsecase::from_global()
        .execute_put_bucket_versioning(build_request(
            PutBucketVersioningInput::builder()
                .bucket(bucket.to_string())
                .versioning_configuration(VersioningConfiguration {
                    status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::SUSPENDED)),
                    ..Default::default()
                })
                .build()
                .expect("suspended versioning request should build"),
            Method::PUT,
        ))
        .await
        .expect("bucket versioning should be suspended");
}

async fn upload_test_object(ecstore: &Arc<ECStore>, bucket: &str, object: &str, data: &[u8]) -> ObjectInfo {
    let mut reader = PutObjReader::from_vec(data.to_vec());
    (**ecstore)
        .put_object(bucket, object, &mut reader, &ObjectOptions::default())
        .await
        .expect("Failed to upload test object")
}

async fn transition_uploaded_object_directly(
    ecstore: &Arc<ECStore>,
    bucket: &str,
    object: &str,
    tier_name: &str,
    uploaded: &ObjectInfo,
) -> ObjectInfo {
    let transition_opts = ObjectOptions {
        transition: lifecycle::lifecycle_contract::TransitionOptions {
            status: lifecycle::lifecycle_contract::TRANSITION_PENDING.to_string(),
            tier: tier_name.to_string(),
            etag: uploaded.etag.clone().unwrap_or_default(),
            ..Default::default()
        },
        version_id: uploaded.version_id.map(|version| version.to_string()),
        versioned: uploaded.version_id.is_some(),
        mod_time: uploaded.mod_time,
        ..Default::default()
    };
    ecstore
        .transition_object(bucket, object, &transition_opts)
        .await
        .expect("Failed to transition object directly");
    wait_for_transition(ecstore, bucket, object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("object should transition before restore assertions")
}

async fn set_bucket_lifecycle_transition_with_tier(
    bucket_name: &str,
    storage_class: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let lifecycle_xml = format!(
        r#"<?xml version="1.0" encoding="UTF-8"?>
<LifecycleConfiguration>
    <Rule>
        <ID>test-rule</ID>
        <Status>Enabled</Status>
        <Filter>
            <Prefix>test/</Prefix>
        </Filter>
        <Transition>
          <Days>0</Days>
          <StorageClass>{storage_class}</StorageClass>
        </Transition>
    </Rule>
</LifecycleConfiguration>"#
    );

    metadata_sys::update(bucket_name, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes()).await?;
    Ok(())
}

async fn set_bucket_object_lock_enabled(bucket_name: &str) -> Result<(), Box<dyn std::error::Error>> {
    let object_lock_xml = r#"<?xml version="1.0" encoding="UTF-8"?>
<ObjectLockConfiguration>
    <ObjectLockEnabled>Enabled</ObjectLockEnabled>
</ObjectLockConfiguration>"#;
    metadata_sys::update(bucket_name, OBJECT_LOCK_CONFIG, object_lock_xml.as_bytes().to_vec()).await?;
    Ok(())
}

fn expiration_lifecycle_configuration(prefix: &str) -> BucketLifecycleConfiguration {
    BucketLifecycleConfiguration {
        expiry_updated_at: None,
        rules: vec![LifecycleRule {
            status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
            abort_incomplete_multipart_upload: None,
            del_marker_expiration: None,
            expiration: Some(LifecycleExpiration {
                date: Some(Timestamp::from(
                    time::OffsetDateTime::now_utc()
                        .replace_time(time::Time::MIDNIGHT)
                        .saturating_sub(time::Duration::days(1)),
                )),
                days: None,
                expired_object_delete_marker: None,
                ..Default::default()
            }),
            filter: Some(LifecycleRuleFilter {
                and: None,
                object_size_greater_than: None,
                object_size_less_than: None,
                prefix: Some(prefix.to_string()),
                tag: None,
                ..Default::default()
            }),
            id: Some("expire-existing".to_string()),
            noncurrent_version_expiration: None,
            noncurrent_version_transitions: None,
            prefix: None,
            transitions: None,
        }],
    }
}

fn expired_object_delete_marker_lifecycle_configuration() -> BucketLifecycleConfiguration {
    BucketLifecycleConfiguration {
        expiry_updated_at: None,
        rules: vec![LifecycleRule {
            status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
            abort_incomplete_multipart_upload: None,
            del_marker_expiration: None,
            expiration: Some(LifecycleExpiration {
                expired_object_delete_marker: Some(true),
                ..Default::default()
            }),
            filter: Some(LifecycleRuleFilter {
                prefix: Some("test/".to_string()),
                ..Default::default()
            }),
            id: Some("expired-delete-marker".to_string()),
            noncurrent_version_expiration: None,
            noncurrent_version_transitions: None,
            prefix: None,
            transitions: None,
        }],
    }
}

fn del_marker_expiration_lifecycle_configuration(days: i32) -> BucketLifecycleConfiguration {
    BucketLifecycleConfiguration {
        expiry_updated_at: None,
        rules: vec![LifecycleRule {
            status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
            abort_incomplete_multipart_upload: None,
            del_marker_expiration: Some(DelMarkerExpiration { days: Some(days) }),
            expiration: Some(LifecycleExpiration {
                days: Some(30),
                ..Default::default()
            }),
            filter: Some(LifecycleRuleFilter {
                prefix: Some("test/".to_string()),
                ..Default::default()
            }),
            id: Some("del-marker-expiration".to_string()),
            noncurrent_version_expiration: None,
            noncurrent_version_transitions: None,
            prefix: None,
            transitions: None,
        }],
    }
}

/// Register the shared [`MockWarmBackend`] into this instance's tier config
/// manager. Thin wrapper over the shared `register_mock_tier` helper
/// (rustfs/backlog#1148 ilm-6).
async fn register_mock_tier(tier_name: &str) -> MockWarmBackend {
    register_mock_tier_util(&current_tier_config_handle(), tier_name).await
}

async fn wait_for_transition(ecstore: &Arc<ECStore>, bucket: &str, object: &str, timeout: Duration) -> Option<ObjectInfo> {
    let deadline = tokio::time::Instant::now() + timeout;

    loop {
        if let Ok(info) = (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await
            && info.transitioned_object.status == "complete"
        {
            return Some(info);
        }

        if tokio::time::Instant::now() >= deadline {
            return None;
        }

        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

async fn wait_for_restore_completion(
    ecstore: &Arc<ECStore>,
    backend: &MockWarmBackend,
    bucket: &str,
    object: &str,
    timeout: Duration,
) -> Result<ObjectInfo, String> {
    let deadline = tokio::time::Instant::now() + timeout;
    let mut last_state = None;

    loop {
        if tokio::time::Instant::now() >= deadline {
            let tier_gets = backend.get_count().await;
            let op_log = backend.op_log().await;
            return Err(format!(
                "restore copy-back should complete within {timeout:?}; tier_gets={tier_gets}, op_log={op_log:?}; last observed state: {}",
                last_state.unwrap_or_else(|| "no object info observed".to_string())
            ));
        }

        match (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
            Ok(info) => {
                if !info.restore_ongoing && info.restore_expires.is_some() {
                    return Ok(info);
                }
                last_state = Some(format!(
                    "restore_ongoing={}, restore_expires={:?}, transitioned_status={}",
                    info.restore_ongoing, info.restore_expires, info.transitioned_object.status
                ));
            }
            Err(err) => {
                last_state = Some(format!("get_object_info failed: {err}"));
            }
        }

        tokio::time::sleep(Duration::from_millis(500)).await;
    }
}

// SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio
// runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the
// `env::set_var` / `env::remove_var` window.
#[allow(unsafe_code)]
async fn with_forced_immediate_enqueue_timeout<F, Fut>(test_fn: F)
where
    F: FnOnce() -> Fut,
    Fut: std::future::Future<Output = ()>,
{
    let original = env::var_os(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT);
    unsafe {
        env::set_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT, "1");
    }
    let result = std::panic::AssertUnwindSafe(test_fn()).catch_unwind().await;
    match original {
        Some(value) => unsafe {
            env::set_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT, value);
        },
        None => unsafe {
            env::remove_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT);
        },
    }
    if let Err(err) = result {
        std::panic::resume_unwind(err);
    }
}

async fn with_get_codec_streaming_remote_probe_env<F, Fut>(test_fn: F)
where
    F: FnOnce() -> Fut,
    Fut: std::future::Future<Output = ()>,
{
    let metrics_was_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
    rustfs_io_metrics::set_get_stage_metrics_enabled(true);
    let result = std::panic::AssertUnwindSafe(temp_env::async_with_vars(
        [
            (ENV_GET_CODEC_STREAMING_ENABLE, Some("true")),
            (ENV_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
            (ENV_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
            (ENV_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
            (ENV_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
        ],
        test_fn(),
    ))
    .catch_unwind()
    .await;
    rustfs_io_metrics::set_get_stage_metrics_enabled(metrics_was_enabled);
    if let Err(err) = result {
        std::panic::resume_unwind(err);
    }
}

async fn wait_for_object_absence(ecstore: &Arc<ECStore>, bucket: &str, object: &str, timeout: Duration) -> bool {
    let deadline = tokio::time::Instant::now() + timeout;

    loop {
        if ecstore
            .get_object_info(bucket, object, &ObjectOptions::default())
            .await
            .is_err()
        {
            return true;
        }

        if tokio::time::Instant::now() >= deadline {
            return false;
        }

        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

async fn wait_for_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &str, timeout: Duration) -> bool {
    let deadline = tokio::time::Instant::now() + timeout;

    loop {
        if let Ok(info) = ecstore.get_object_info(bucket, object, &ObjectOptions::default()).await
            && info.delete_marker
        {
            return true;
        }

        if tokio::time::Instant::now() >= deadline {
            return false;
        }

        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

fn build_request<T>(input: T, method: Method) -> S3Request<T> {
    S3Request {
        input,
        method,
        uri: Uri::from_static("/"),
        headers: HeaderMap::new(),
        extensions: Extensions::new(),
        credentials: None,
        region: None,
        service: None,
        trailing_headers: None,
    }
}

fn streaming_blob_from_bytes(data: &[u8]) -> StreamingBlob {
    let body = Bytes::copy_from_slice(data);
    StreamingBlob::wrap::<_, Infallible>(stream::once(async move { Ok(body) }))
}

async fn read_object_bytes(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> Vec<u8> {
    let mut reader = (**ecstore)
        .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
        .await
        .expect("Failed to read object");
    let mut buf = Vec::new();
    reader
        .stream
        .read_to_end(&mut buf)
        .await
        .expect("Failed to drain object reader");
    buf
}

async fn live_object_version_count(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> usize {
    let versions = ecstore
        .clone()
        .list_object_versions(bucket, object, None, None, None, 1000)
        .await
        .expect("Failed to list object versions");

    versions
        .objects
        .iter()
        .filter(|info| info.name == object && !info.delete_marker)
        .count()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn put_object_if_none_match_existing_object_returns_precondition_failed() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let fs = FS::new();
    let usecase = DefaultObjectUsecase::from_global();

    let bucket = format!("test-put-if-none-match-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/object.txt";
    let initial_payload = b"initial conditional put payload";
    let replacement_payload = b"replacement conditional put payload";

    create_test_bucket(&ecstore, bucket.as_str()).await;

    let initial_input = PutObjectInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .body(Some(streaming_blob_from_bytes(initial_payload)))
        .content_length(Some(initial_payload.len() as i64))
        .build()
        .unwrap();
    Box::pin(usecase.execute_put_object(&fs, build_request(initial_input, Method::PUT)))
        .await
        .expect("Failed to upload initial object through usecase");

    let existing_info = ecstore
        .get_object_info(bucket.as_str(), object, &ObjectOptions::default())
        .await
        .expect("Failed to fetch existing object info");
    let existing_etag = existing_info.etag.expect("existing object should have an ETag");

    let replacement_input = PutObjectInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .body(Some(streaming_blob_from_bytes(replacement_payload)))
        .content_length(Some(replacement_payload.len() as i64))
        .build()
        .unwrap();
    let mut req = build_request(replacement_input, Method::PUT);
    req.headers
        .insert(IF_NONE_MATCH, HeaderValue::from_str(existing_etag.as_str()).unwrap());

    let err = Box::pin(usecase.execute_put_object(&fs, req)).await.unwrap_err();

    assert_eq!(err.code(), &s3s::S3ErrorCode::PreconditionFailed);
    assert_eq!(
        read_object_bytes(&ecstore, bucket.as_str(), object).await,
        initial_payload,
        "failed conditional PutObject must not overwrite the current object"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn copy_object_if_none_match_existing_destination_returns_precondition_failed() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();

    let src_bucket = format!("test-copy-if-none-match-src-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let dst_bucket = format!("test-copy-if-none-match-dst-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let src_object = "test/source.txt";
    let dst_object = "test/destination.txt";
    let src_payload = b"conditional copy source payload";
    let dst_payload = b"conditional copy destination payload";

    create_test_bucket(&ecstore, src_bucket.as_str()).await;
    create_test_bucket(&ecstore, dst_bucket.as_str()).await;
    let _ = upload_test_object(&ecstore, src_bucket.as_str(), src_object, src_payload).await;
    let _ = upload_test_object(&ecstore, dst_bucket.as_str(), dst_object, dst_payload).await;

    let dst_info = ecstore
        .get_object_info(dst_bucket.as_str(), dst_object, &ObjectOptions::default())
        .await
        .expect("Failed to fetch destination object info");
    let dst_etag = dst_info.etag.expect("destination object should have an ETag");

    let copy_input = CopyObjectInput::builder()
        .copy_source(CopySource::Bucket {
            bucket: src_bucket.clone().into(),
            key: src_object.to_string().into(),
            version_id: None,
        })
        .bucket(dst_bucket.clone())
        .key(dst_object.to_string())
        .build()
        .unwrap();
    let mut req = build_request(copy_input, Method::PUT);
    req.headers
        .insert(IF_NONE_MATCH, HeaderValue::from_str(dst_etag.as_str()).unwrap());

    let err = Box::pin(usecase.execute_copy_object(req)).await.unwrap_err();

    assert_eq!(err.code(), &s3s::S3ErrorCode::PreconditionFailed);
    assert_eq!(
        read_object_bytes(&ecstore, dst_bucket.as_str(), dst_object).await,
        dst_payload,
        "failed conditional CopyObject must not overwrite the current destination object"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state object-version integration test: runs serialized in the CI ILM Integration (serial) lane"]
async fn undo_copy_requires_the_observed_current_version() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let fs = FS::new();
    let usecase = DefaultObjectUsecase::from_global();
    let bucket = format!("test-undo-copy-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/object.txt";
    create_test_bucket(&ecstore, bucket.as_str()).await;

    let put = |payload: &'static [u8]| {
        PutObjectInput::builder()
            .bucket(bucket.clone())
            .key(object.to_string())
            .body(Some(streaming_blob_from_bytes(payload)))
            .content_length(Some(payload.len() as i64))
            .build()
            .unwrap()
    };
    let original = Box::pin(usecase.execute_put_object(&fs, build_request(put(b"original"), Method::PUT)))
        .await
        .expect("initial version should be written")
        .output
        .version_id
        .expect("versioned PUT should return a version ID");
    let observed_current = Box::pin(usecase.execute_put_object(&fs, build_request(put(b"replacement"), Method::PUT)))
        .await
        .expect("replacement version should be written")
        .output
        .version_id
        .expect("versioned PUT should return a version ID");

    let copy = || {
        CopyObjectInput::builder()
            .copy_source(CopySource::Bucket {
                bucket: bucket.clone().into(),
                key: object.to_string().into(),
                version_id: Some(original.clone().into()),
            })
            .bucket(bucket.clone())
            .key(object.to_string())
            .build()
            .unwrap()
    };
    let mut stale_req = build_request(copy(), Method::PUT);
    stale_req.headers.insert(
        RUSTFS_EXPECTED_CURRENT_VERSION_ID,
        HeaderValue::from_str(&Uuid::new_v4().to_string()).unwrap(),
    );
    let versions_before = live_object_version_count(&ecstore, bucket.as_str(), object).await;
    let err = Box::pin(usecase.execute_copy_object(stale_req))
        .await
        .expect_err("stale destination must fail");
    assert_eq!(err.code(), &s3s::S3ErrorCode::PreconditionFailed);
    assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), object).await, versions_before);
    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, b"replacement");

    let mut matching_req = build_request(copy(), Method::PUT);
    matching_req
        .headers
        .insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_str(&observed_current).unwrap());
    Box::pin(usecase.execute_copy_object(matching_req))
        .await
        .expect("matching destination version should permit undo copy");
    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, b"original");
    assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), object).await, versions_before + 1);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state object-version integration test: runs serialized in the CI ILM Integration (serial) lane"]
async fn undo_delete_only_removes_the_current_matching_delete_marker() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let fs = FS::new();
    let usecase = DefaultObjectUsecase::from_global();
    let bucket = format!("test-undo-delete-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/object.txt";
    create_test_bucket(&ecstore, bucket.as_str()).await;

    let put = PutObjectInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .body(Some(streaming_blob_from_bytes(b"visible again")))
        .content_length(Some(13))
        .build()
        .unwrap();
    Box::pin(usecase.execute_put_object(&fs, build_request(put, Method::PUT)))
        .await
        .expect("object version should be written");
    let marker = Box::pin(
        usecase.execute_delete_object(build_request(
            DeleteObjectInput::builder()
                .bucket(bucket.clone())
                .key(object.to_string())
                .build()
                .unwrap(),
            Method::DELETE,
        )),
    )
    .await
    .expect("delete marker should be created")
    .output
    .version_id
    .expect("versioned delete should return the marker version");

    let delete_marker = || {
        DeleteObjectInput::builder()
            .bucket(bucket.clone())
            .key(object.to_string())
            .version_id(Some(marker.clone()))
            .build()
            .unwrap()
    };
    let mut stale_req = build_request(delete_marker(), Method::DELETE);
    stale_req.headers.insert(
        RUSTFS_EXPECTED_CURRENT_VERSION_ID,
        HeaderValue::from_str(&Uuid::new_v4().to_string()).unwrap(),
    );
    let err = Box::pin(usecase.execute_delete_object(stale_req))
        .await
        .expect_err("mismatched delete marker must fail");
    assert_eq!(err.code(), &s3s::S3ErrorCode::PreconditionFailed);

    let mut matching_req = build_request(delete_marker(), Method::DELETE);
    matching_req
        .headers
        .insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_str(&marker).unwrap());
    Box::pin(usecase.execute_delete_object(matching_req))
        .await
        .expect("current matching delete marker should be removed");
    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, b"visible again");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
#[ignore = "global-state object-version integration test: runs serialized in the CI ILM Integration (serial) lane"]
async fn undo_copy_serializes_a_concurrent_put_until_after_commit() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let bucket = format!("test-undo-race-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/object.txt";
    create_test_bucket(&ecstore, bucket.as_str()).await;

    let put = |payload: &'static [u8]| {
        PutObjectInput::builder()
            .bucket(bucket.clone())
            .key(object.to_string())
            .body(Some(streaming_blob_from_bytes(payload)))
            .content_length(Some(payload.len() as i64))
            .build()
            .unwrap()
    };
    let fs = FS::new();
    let usecase = DefaultObjectUsecase::from_global();
    let original = Box::pin(usecase.execute_put_object(&fs, build_request(put(b"original"), Method::PUT)))
        .await
        .expect("original should be written")
        .output
        .version_id
        .expect("original should be versioned");
    let observed = Box::pin(usecase.execute_put_object(&fs, build_request(put(b"observed"), Method::PUT)))
        .await
        .expect("observed should be written")
        .output
        .version_id
        .expect("observed should be versioned");
    let encoded_object = encode_dir_object(object);
    let lock = ecstore
        .new_ns_lock(bucket.as_str(), &encoded_object)
        .await
        .expect("namespace lock should be available");
    let blocker = lock
        .get_write_lock(Duration::from_secs(10))
        .await
        .expect("test should acquire blocker");

    let mut undo_req = build_request(
        CopyObjectInput::builder()
            .copy_source(CopySource::Bucket {
                bucket: bucket.clone().into(),
                key: object.to_string().into(),
                version_id: Some(original.into()),
            })
            .bucket(bucket.clone())
            .key(object.to_string())
            .build()
            .unwrap(),
        Method::PUT,
    );
    undo_req
        .headers
        .insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_str(&observed).unwrap());

    let undo = tokio::spawn(async move {
        let usecase = DefaultObjectUsecase::from_global();
        Box::pin(usecase.execute_copy_object(undo_req)).await
    });
    tokio::task::yield_now().await;
    let put = PutObjectInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .body(Some(streaming_blob_from_bytes(b"concurrent")))
        .content_length(Some(10))
        .build()
        .unwrap();
    let concurrent_put = tokio::spawn(async move {
        let fs = FS::new();
        let usecase = DefaultObjectUsecase::from_global();
        Box::pin(usecase.execute_put_object(&fs, build_request(put, Method::PUT))).await
    });

    drop(blocker);
    let undo_result = undo.await.expect("undo task should not panic");
    let put_result = concurrent_put.await.expect("PUT task should not panic");
    assert!(put_result.is_ok(), "ordinary PUT should eventually succeed: {put_result:?}");
    if let Err(err) = undo_result {
        assert_eq!(err.code(), &s3s::S3ErrorCode::PreconditionFailed);
    }
    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, b"concurrent");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn copy_object_allows_new_version_for_locked_destination_but_blocks_explicit_overwrite() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let fs = FS::new();
    let usecase = DefaultObjectUsecase::from_global();

    let bucket = format!("test-copy-object-lock-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let src_object = "test/source.txt";
    let dst_object = "test/destination.txt";
    let locked_payload = b"locked destination payload";
    let source_payload = b"copy source payload";

    create_test_bucket(&ecstore, bucket.as_str()).await;
    set_bucket_object_lock_enabled(bucket.as_str())
        .await
        .expect("Failed to enable object lock for bucket");
    let _ = upload_test_object(&ecstore, bucket.as_str(), src_object, source_payload).await;

    let retain_until = Timestamp::from(time::OffsetDateTime::now_utc().saturating_add(time::Duration::days(1)));
    let locked_input = PutObjectInput::builder()
        .bucket(bucket.clone())
        .key(dst_object.to_string())
        .body(Some(streaming_blob_from_bytes(locked_payload)))
        .content_length(Some(locked_payload.len() as i64))
        .object_lock_mode(Some(ObjectLockMode::from_static(ObjectLockMode::COMPLIANCE)))
        .object_lock_retain_until_date(Some(retain_until))
        .build()
        .unwrap();

    Box::pin(usecase.execute_put_object(&fs, build_request(locked_input, Method::PUT)))
        .await
        .expect("Failed to upload locked destination object");

    let locked_info = ecstore
        .get_object_info(bucket.as_str(), dst_object, &ObjectOptions::default())
        .await
        .expect("Failed to fetch locked destination object info");
    let locked_version_id = locked_info
        .version_id
        .expect("locked destination should have a version ID")
        .to_string();

    let copy_input = CopyObjectInput::builder()
        .copy_source(CopySource::Bucket {
            bucket: bucket.clone().into(),
            key: src_object.to_string().into(),
            version_id: None,
        })
        .bucket(bucket.clone())
        .key(dst_object.to_string())
        .build()
        .unwrap();

    let copy_output = Box::pin(usecase.execute_copy_object(build_request(copy_input, Method::PUT)))
        .await
        .expect("CopyObject should create a new version over a locked current version");
    let copied_version_id = copy_output
        .output
        .version_id
        .expect("versioned CopyObject should return the created version ID");

    assert_ne!(copied_version_id, locked_version_id);
    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), dst_object).await, source_payload);
    assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), dst_object).await, 2);

    let explicit_overwrite_input = CopyObjectInput::builder()
        .copy_source(CopySource::Bucket {
            bucket: bucket.clone().into(),
            key: src_object.to_string().into(),
            version_id: None,
        })
        .bucket(bucket.clone())
        .key(dst_object.to_string())
        .version_id(Some(locked_version_id))
        .build()
        .unwrap();

    let err = Box::pin(usecase.execute_copy_object(build_request(explicit_overwrite_input, Method::PUT)))
        .await
        .expect_err("explicit CopyObject overwrite of a locked version should be blocked");

    assert_eq!(err.code(), &s3s::S3ErrorCode::AccessDenied);
    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), dst_object).await, source_payload);
    assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), dst_object).await, 2);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn concurrent_reverse_copy_object_does_not_deadlock_with_reader_locks() {
    temp_env::async_with_vars([(ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, Some("false"))], async {
        let (_disk_paths, ecstore) = setup_test_env().await;

        let bucket = format!("test-reverse-copy-{}", &Uuid::new_v4().simple().to_string()[..8]);
        let object_a = "test/a.bin";
        let object_b = "test/b.bin";
        let payload_a = vec![b'a'; 4 * 1024 * 1024];
        let payload_b = vec![b'b'; 4 * 1024 * 1024];

        create_test_bucket(&ecstore, bucket.as_str()).await;
        let _ = upload_test_object(&ecstore, bucket.as_str(), object_a, &payload_a).await;
        let _ = upload_test_object(&ecstore, bucket.as_str(), object_b, &payload_b).await;

        let a_to_b_input = CopyObjectInput::builder()
            .copy_source(CopySource::Bucket {
                bucket: bucket.clone().into(),
                key: object_a.to_string().into(),
                version_id: None,
            })
            .bucket(bucket.clone())
            .key(object_b.to_string())
            .build()
            .unwrap();
        let b_to_a_input = CopyObjectInput::builder()
            .copy_source(CopySource::Bucket {
                bucket: bucket.clone().into(),
                key: object_b.to_string().into(),
                version_id: None,
            })
            .bucket(bucket.clone())
            .key(object_a.to_string())
            .build()
            .unwrap();

        let start = Arc::new(Barrier::new(3));
        let start_a = start.clone();
        let a_to_b = tokio::spawn(async move {
            start_a.wait().await;
            let usecase = DefaultObjectUsecase::from_global();
            Box::pin(usecase.execute_copy_object(build_request(a_to_b_input, Method::PUT))).await
        });
        let start_b = start.clone();
        let b_to_a = tokio::spawn(async move {
            start_b.wait().await;
            let usecase = DefaultObjectUsecase::from_global();
            Box::pin(usecase.execute_copy_object(build_request(b_to_a_input, Method::PUT))).await
        });

        start.wait().await;

        let (a_to_b_result, b_to_a_result) = tokio::time::timeout(Duration::from_secs(10), async {
            let (a_to_b_result, b_to_a_result) = tokio::join!(a_to_b, b_to_a);
            (
                a_to_b_result.expect("A-to-B CopyObject task should not panic"),
                b_to_a_result.expect("B-to-A CopyObject task should not panic"),
            )
        })
        .await
        .expect("reverse CopyObject operations should not deadlock");

        a_to_b_result.expect("A-to-B CopyObject should succeed");
        b_to_a_result.expect("B-to-A CopyObject should succeed");

        let final_a = read_object_bytes(&ecstore, bucket.as_str(), object_a).await;
        let final_b = read_object_bytes(&ecstore, bucket.as_str(), object_b).await;
        assert!(
            final_a == payload_a || final_a == payload_b,
            "object A must contain a complete copied payload"
        );
        assert!(
            final_b == payload_a || final_b == payload_b,
            "object B must contain a complete copied payload"
        );
    })
    .await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn put_and_copy_object_transition_immediately_via_usecases() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let fs = FS::new();
    let usecase = DefaultObjectUsecase::from_global();

    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;

    let put_bucket = format!("test-api-put-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let put_object = "test/object.txt";
    let put_payload = b"Hello, immediate transition through put API!";

    create_test_bucket(&ecstore, put_bucket.as_str()).await;
    set_bucket_lifecycle_transition_with_tier(put_bucket.as_str(), &tier_name)
        .await
        .expect("Failed to set lifecycle configuration");

    let put_input = PutObjectInput::builder()
        .bucket(put_bucket.clone())
        .key(put_object.to_string())
        .body(Some(streaming_blob_from_bytes(put_payload)))
        .content_length(Some(put_payload.len() as i64))
        .build()
        .unwrap();

    Box::pin(usecase.execute_put_object(&fs, build_request(put_input, Method::PUT)))
        .await
        .expect("Failed to put object through usecase");

    let put_info = wait_for_transition(&ecstore, put_bucket.as_str(), put_object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("object should transition immediately after put usecase");

    assert_eq!(put_info.transitioned_object.status, "complete");
    assert_eq!(put_info.transitioned_object.tier, tier_name);
    assert!(backend.contains(&put_info.transitioned_object.name).await);

    let src_bucket = format!("test-api-copy-src-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let dst_bucket = format!("test-api-copy-dst-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let src_object = "test/source.txt";
    let dst_object = "test/copied.txt";
    let copy_payload = b"copy object immediate transition through copy API";

    create_test_bucket(&ecstore, src_bucket.as_str()).await;
    create_test_bucket(&ecstore, dst_bucket.as_str()).await;
    set_bucket_lifecycle_transition_with_tier(dst_bucket.as_str(), &tier_name)
        .await
        .expect("Failed to set destination lifecycle configuration");
    let _ = upload_test_object(&ecstore, src_bucket.as_str(), src_object, copy_payload).await;

    let copy_input = CopyObjectInput::builder()
        .copy_source(CopySource::Bucket {
            bucket: src_bucket.clone().into(),
            key: src_object.to_string().into(),
            version_id: None,
        })
        .bucket(dst_bucket.clone())
        .key(dst_object.to_string())
        .build()
        .unwrap();

    Box::pin(usecase.execute_copy_object(build_request(copy_input, Method::PUT)))
        .await
        .expect("Failed to copy object through usecase");

    let copy_info = wait_for_transition(&ecstore, dst_bucket.as_str(), dst_object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("copied object should transition immediately after copy usecase");

    assert_eq!(copy_info.transitioned_object.status, "complete");
    assert_eq!(copy_info.transitioned_object.tier, tier_name);
    assert!(backend.contains(&copy_info.transitioned_object.name).await);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn complete_multipart_upload_transitions_immediately_via_usecase() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultMultipartUsecase::from_global();

    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;

    let bucket = format!("test-api-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/multipart.txt";
    let payload = b"multipart immediate transition through complete API";

    create_test_bucket(&ecstore, bucket.as_str()).await;
    set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name)
        .await
        .expect("Failed to set lifecycle configuration");

    let upload = ecstore
        .new_multipart_upload(bucket.as_str(), object, &ObjectOptions::default())
        .await
        .expect("Failed to create multipart upload");

    let mut reader = PutObjReader::from_vec(payload.to_vec());
    let uploaded_part = ecstore
        .put_object_part(bucket.as_str(), object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default())
        .await
        .expect("Failed to upload multipart part");

    let complete_input = CompleteMultipartUploadInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .upload_id(upload.upload_id.clone())
        .multipart_upload(Some(CompletedMultipartUpload {
            parts: Some(vec![CompletedPart {
                part_number: Some(1),
                e_tag: uploaded_part.etag.clone().map(|etag| to_s3s_etag(&etag)),
                ..Default::default()
            }]),
        }))
        .build()
        .unwrap();

    let complete_output = Box::pin(usecase.execute_complete_multipart_upload(build_request(complete_input, Method::POST)))
        .await
        .expect("Failed to complete multipart upload through usecase");
    assert!(
        complete_output.output.version_id.is_some(),
        "versioned CompleteMultipartUpload should return the created version ID"
    );

    let info = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("multipart object should transition immediately after complete usecase");

    assert_eq!(info.transitioned_object.status, "complete");
    assert_eq!(info.transitioned_object.tier, tier_name);
    assert!(backend.contains(&info.transitioned_object.name).await);
}

#[test]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
fn get_transitioned_object_uses_remote_codec_fallback_path() {
    // CI's default test stack overflows this deep async path; run it on a dedicated large-stack thread.
    std::thread::Builder::new()
        .name("lifecycle-transition-remote-codec-fallback".to_string())
        .stack_size(16 * 1024 * 1024)
        .spawn(|| {
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("remote codec fallback test runtime should build");

            runtime.block_on(async {
                with_get_codec_streaming_remote_probe_env(|| async {
                    let (_disk_paths, ecstore) = setup_test_env().await;

                    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
                    let backend = register_mock_tier(&tier_name).await;

                    let bucket = format!("test-api-get-remote-{}", &Uuid::new_v4().simple().to_string()[..8]);
                    let object = "test/remote-codec-fallback.txt";
                    let payload: Vec<u8> = (0..(1024 * 1024))
                        .map(|index| u8::try_from(index % 251).expect("payload byte fits in u8"))
                        .collect();

                    create_test_bucket(&ecstore, bucket.as_str()).await;

                    let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
                    let transition_opts = ObjectOptions {
                        transition: lifecycle::lifecycle_contract::TransitionOptions {
                            status: lifecycle::lifecycle_contract::TRANSITION_PENDING.to_string(),
                            tier: tier_name.clone(),
                            etag: uploaded.etag.clone().unwrap_or_default(),
                            ..Default::default()
                        },
                        version_id: uploaded.version_id.map(|version| version.to_string()),
                        versioned: true,
                        mod_time: uploaded.mod_time,
                        ..Default::default()
                    };
                    ecstore
                        .transition_object(bucket.as_str(), object, &transition_opts)
                        .await
                        .expect("Failed to transition object directly");

                    let transitioned = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
                        .await
                        .expect("object should transition before remote fallback GET");

                    assert_eq!(transitioned.transitioned_object.status, "complete");
                    assert_eq!(transitioned.transitioned_object.tier, tier_name);
                    assert!(!transitioned.transitioned_object.name.is_empty());
                    assert!(backend.contains(&transitioned.transitioned_object.name).await);

                    let actual = read_object_bytes(&ecstore, bucket.as_str(), object).await;
                    assert_eq!(actual, payload);
                })
                .await;
            });
        })
        .expect("remote codec fallback test thread should spawn")
        .join()
        .expect("remote codec fallback test thread should not panic");
}

/// Regression test for rustfs/rustfs#4827: a duplicate transition task that runs
/// after the winning transition completed must return early without a second
/// tier upload and without clobbering the winner's remote object / metadata.
///
/// The duplicate models the residual race left after the in-flight enqueue dedup
/// (#4839): a stale-snapshot task admitted after the winner released its claim.
/// On an unversioned bucket the duplicate's metadata re-read is eligible for the
/// GET metadata cache; before the fix the cache still held the pre-transition
/// entry (transition_object never invalidated it after persisting), so the
/// TRANSITION_COMPLETE early-return never fired and the duplicate streamed the
/// already-deleted local data to the tier again (NotFound reader errors +
/// rejected duplicate tier PUT).
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn duplicate_transition_task_skips_already_transitioned_version() {
    let (_disk_paths, ecstore) = setup_test_env().await;

    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;

    let bucket = format!("test-api-dup-transition-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/duplicate-transition.txt";
    let payload = b"duplicate transition tasks must not re-upload";

    // Unversioned bucket: the duplicate's re-read goes through the GET metadata
    // cache (versioned reads bypass it), which is the stale-read window under test.
    (*ecstore)
        .make_bucket(bucket.as_str(), &MakeBucketOptions::default())
        .await
        .expect("Failed to create unversioned test bucket");

    let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, payload).await;
    let transition_opts = ObjectOptions {
        transition: lifecycle::lifecycle_contract::TransitionOptions {
            status: lifecycle::lifecycle_contract::TRANSITION_PENDING.to_string(),
            tier: tier_name.clone(),
            etag: uploaded.etag.clone().unwrap_or_default(),
            ..Default::default()
        },
        version_id: uploaded.version_id.map(|version| version.to_string()),
        mod_time: uploaded.mod_time,
        ..Default::default()
    };

    // Winner: transitions the object, persists transition_status=complete, and
    // deletes the local data.
    ecstore
        .transition_object(bucket.as_str(), object, &transition_opts)
        .await
        .expect("winning transition should succeed");

    let puts_after_winner: Vec<String> = backend
        .op_log()
        .await
        .into_iter()
        .filter_map(|op| match op {
            MockWarmOp::Put { object, .. } => Some(object),
            _ => None,
        })
        .collect();
    assert_eq!(puts_after_winner.len(), 1, "winner must upload exactly one tier object");
    let winner_remote_object = puts_after_winner[0].clone();

    // Duplicate task with the same stale snapshot (same mod_time / etag), run
    // immediately so the pre-fix stale metadata-cache entry is still live. It
    // must observe transition_status=complete and return Ok without uploading.
    ecstore
        .transition_object(bucket.as_str(), object, &transition_opts)
        .await
        .expect("duplicate transition must return early without error");

    let puts_after_duplicate = backend
        .op_log()
        .await
        .into_iter()
        .filter(|op| matches!(op, MockWarmOp::Put { .. }))
        .count();
    assert_eq!(puts_after_duplicate, 1, "duplicate transition must not attempt a second tier upload");

    // The winner's remote object and local metadata must be intact.
    assert!(backend.contains(&winner_remote_object).await);
    let info = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("object must remain transitioned after duplicate task");
    assert_eq!(info.transitioned_object.status, "complete");
    assert_eq!(info.transitioned_object.tier, tier_name);
    assert_eq!(
        info.transitioned_object.name, winner_remote_object,
        "duplicate transition must not repoint metadata at a new remote object"
    );
    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, payload);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn transition_rejects_stale_etag_before_remote_upload() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;
    let bucket = format!("test-api-transition-stale-etag-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/transition-stale-etag.txt";
    (*ecstore)
        .make_bucket(bucket.as_str(), &MakeBucketOptions::default())
        .await
        .expect("unversioned test bucket should be created");
    let original = upload_test_object(&ecstore, bucket.as_str(), object, b"original transition candidate").await;
    let replacement_payload = b"replacement committed before the stale transition starts";
    let replacement = upload_test_object(&ecstore, bucket.as_str(), object, replacement_payload).await;
    let transition_opts = ObjectOptions {
        transition: lifecycle::lifecycle_contract::TransitionOptions {
            status: lifecycle::lifecycle_contract::TRANSITION_PENDING.to_string(),
            tier: tier_name,
            etag: original.etag.clone().unwrap_or_default(),
            ..Default::default()
        },
        version_id: replacement.version_id.map(|version| version.to_string()),
        mod_time: replacement.mod_time,
        ..Default::default()
    };

    ecstore
        .transition_object(bucket.as_str(), object, &transition_opts)
        .await
        .expect_err("stale ETag must reject the transition before remote upload");
    assert_eq!(backend.put_count().await, 0, "rejected stale work must not upload a remote candidate");
    assert_eq!(
        read_object_bytes(&ecstore, bucket.as_str(), object).await,
        replacement_payload,
        "the replacement object must remain local and readable"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn transition_commit_rejects_replaced_source() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;
    let bucket = format!("test-api-transition-replace-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/transition-replaced-source.txt";
    let original_payload = b"transition source that must not replace a concurrent overwrite";
    let replacement_payload = b"concurrent replacement must remain the authoritative local object";

    (*ecstore)
        .make_bucket(bucket.as_str(), &MakeBucketOptions::default())
        .await
        .expect("unversioned test bucket should be created");
    let original = upload_test_object(&ecstore, bucket.as_str(), object, original_payload).await;
    let transition_opts = ObjectOptions {
        transition: lifecycle::lifecycle_contract::TransitionOptions {
            status: lifecycle::lifecycle_contract::TRANSITION_PENDING.to_string(),
            tier: tier_name.clone(),
            etag: original.etag.clone().unwrap_or_default(),
            ..Default::default()
        },
        version_id: original.version_id.map(|version| version.to_string()),
        mod_time: original.mod_time,
        ..Default::default()
    };

    let put_barrier = backend.arm_put_barrier().await;
    let transition_store = Arc::clone(&ecstore);
    let transition_bucket = bucket.clone();
    let transition = tokio::spawn(async move {
        transition_store
            .transition_object(&transition_bucket, object, &transition_opts)
            .await
    });
    put_barrier.wait_until_paused().await;

    upload_test_object(&ecstore, bucket.as_str(), object, replacement_payload).await;
    put_barrier.release();

    transition
        .await
        .expect("transition task should not panic")
        .expect_err("transition must reject a source replaced during remote upload");
    assert_eq!(
        backend.put_count().await,
        1,
        "the stale transition should upload exactly one remote candidate"
    );
    assert_eq!(
        backend.remove_count().await,
        1,
        "the uncommitted remote candidate must be removed after identity revalidation fails"
    );
    assert_eq!(
        backend.remove_versions().await,
        backend.put_versions().await,
        "cleanup must delete the exact object version returned by the tier PUT"
    );
    assert_eq!(
        backend.object_count().await,
        0,
        "the rejected transition must not leave a remote candidate"
    );
    assert_eq!(
        read_object_bytes(&ecstore, bucket.as_str(), object).await,
        replacement_payload,
        "the replacement object must remain readable and unmodified"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn cancelled_transition_waiting_for_prepared_reader_cleans_remote() {
    temp_env::async_with_vars([(ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, Some("false"))], async {
        let (_disk_paths, ecstore) = setup_test_env().await;
        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
        let backend = register_mock_tier(&tier_name).await;
        let bucket = format!("test-api-transition-reader-lock-{}", &Uuid::new_v4().simple().to_string()[..8]);
        let object = "test/transition-reader-lock.txt";
        let payload = b"prepared reader must keep transition commit behind the namespace lock".repeat(1024);

        (*ecstore)
            .make_bucket(bucket.as_str(), &MakeBucketOptions::default())
            .await
            .expect("unversioned test bucket should be created");
        let original = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
        let transition_opts = ObjectOptions {
            transition: lifecycle::lifecycle_contract::TransitionOptions {
                status: lifecycle::lifecycle_contract::TRANSITION_PENDING.to_string(),
                tier: tier_name,
                etag: original.etag.clone().unwrap_or_default(),
                ..Default::default()
            },
            version_id: original.version_id.map(|version| version.to_string()),
            mod_time: original.mod_time,
            ..Default::default()
        };

        let put_barrier = backend.arm_put_barrier().await;
        let transition_store = Arc::clone(&ecstore);
        let transition_bucket = bucket.clone();
        let mut transition = tokio::spawn(async move {
            transition_store
                .transition_object(&transition_bucket, object, &transition_opts)
                .await
        });
        put_barrier.wait_until_paused().await;

        let prepared_reader = ecstore
            .prepare_get_object_reader(bucket.as_str(), object, None, HeaderMap::new(), &ObjectOptions::default())
            .await
            .expect("prepared reader should resolve while the local source still exists");
        put_barrier.release();
        assert!(
            tokio::time::timeout(Duration::from_millis(200), &mut transition)
                .await
                .is_err(),
            "transition commit must wait while a prepared reader owns the namespace read lock"
        );

        transition.abort();
        assert!(
            transition
                .await
                .expect_err("transition task should be cancelled")
                .is_cancelled(),
            "transition cancellation should not be reported as a task panic"
        );
        drop(prepared_reader);
        assert_eq!(backend.put_count().await, 1, "the transition should upload exactly one remote candidate");
        assert!(
            backend.wait_for_object_count(0, TRANSITION_WAIT_TIMEOUT).await,
            "cancelling while the commit waits for the namespace lock must clean the exact uploaded candidate"
        );
        assert_eq!(backend.remove_versions().await, backend.put_versions().await);
        assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, payload);
    })
    .await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn delete_transitioned_object_removes_remote_tier_copy_via_usecase() {
    use super::storage_api::test::ReqInfo;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();

    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;

    let bucket = format!("test-api-delete-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/object.txt";
    let payload = b"delete transitioned object through delete API";

    create_test_bucket(&ecstore, bucket.as_str()).await;
    set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name)
        .await
        .expect("Failed to set lifecycle configuration");
    let _ = upload_test_object(&ecstore, bucket.as_str(), object, payload).await;

    lifecycle::bucket_lifecycle_ops::enqueue_transition_for_existing_objects(ecstore.clone(), bucket.as_str())
        .await
        .expect("Failed to enqueue transitioned object");

    let transitioned = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("object should transition before delete usecase runs");
    let remote_object = transitioned.transitioned_object.name.clone();

    assert!(backend.contains(&remote_object).await);

    let mut req = build_request(
        DeleteObjectInput::builder()
            .bucket(bucket.clone())
            .key(object.to_string())
            .build()
            .unwrap(),
        Method::DELETE,
    );
    insert_header(&mut req.headers, SUFFIX_FORCE_DELETE, "true");
    req.extensions.insert(ReqInfo {
        cred: Some(rustfs_credentials::Credentials::default()),
        is_owner: true,
        ..Default::default()
    });

    Box::pin(usecase.execute_delete_object(req))
        .await
        .expect("Failed to delete object through usecase");

    assert!(
        wait_for_object_absence(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT).await,
        "object should be removed from hot tier after delete usecase"
    );

    assert!(
        backend.wait_for_remote_absence(&remote_object, TRANSITION_WAIT_TIMEOUT).await,
        "transitioned object should be removed from remote tier after delete usecase"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn lifecycle_transition_marks_dirty_disks_for_capacity_manager() {
    let (disk_paths, ecstore) = setup_test_env().await;
    let manager = create_isolated_manager(HybridStrategyConfig::default());
    let _ = manager.get_dirty_disks().await;

    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let _backend = register_mock_tier(&tier_name).await;

    let bucket = format!("test-capacity-transition-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/object.txt";
    let payload = b"transition should mark dirty scope";

    create_test_bucket(&ecstore, bucket.as_str()).await;
    set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name)
        .await
        .expect("Failed to set lifecycle configuration");
    let _ = upload_test_object(&ecstore, bucket.as_str(), object, payload).await;

    lifecycle::bucket_lifecycle_ops::enqueue_transition_for_existing_objects(ecstore.clone(), bucket.as_str())
        .await
        .expect("Failed to enqueue transitioned object");

    let _ = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("object should transition before dirty scope assertion");

    let dirty_disks = manager.get_dirty_disks().await;
    assert_eq!(dirty_disks.len(), disk_paths.len());

    let actual_paths: std::collections::HashSet<_> = dirty_disks
        .into_iter()
        .map(|disk| stdfs::canonicalize(&disk.drive_path).unwrap().to_string_lossy().into_owned())
        .collect();
    let expected_paths: std::collections::HashSet<_> = disk_paths
        .iter()
        .map(|path| stdfs::canonicalize(path).unwrap().to_string_lossy().into_owned())
        .collect();
    assert_eq!(actual_paths, expected_paths);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn immediate_transition_timeout_eventually_completes_via_compensation() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;

    let bucket = format!("test-compensation-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/object.txt";
    let payload = b"transition compensation should eventually complete";

    create_test_bucket(&ecstore, bucket.as_str()).await;
    set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name)
        .await
        .expect("Failed to set lifecycle configuration");

    with_forced_immediate_enqueue_timeout(|| async {
        let _ = upload_test_object(&ecstore, bucket.as_str(), object, payload).await;
    })
    .await;

    let info = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("object should eventually transition after compensation backfill");

    assert_eq!(info.transitioned_object.status, "complete");
    assert_eq!(info.transitioned_object.tier, tier_name);
    assert!(backend.contains(&info.transitioned_object.name).await);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn compensation_driven_copy_still_completes_transition() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();

    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;

    let src_bucket = format!("test-comp-copy-src-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let dst_bucket = format!("test-comp-copy-dst-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let src_object = "test/source.txt";
    let dst_object = "test/copied.txt";
    let payload = b"copy object should still transition after compensation";

    create_test_bucket(&ecstore, src_bucket.as_str()).await;
    create_test_bucket(&ecstore, dst_bucket.as_str()).await;
    set_bucket_lifecycle_transition_with_tier(dst_bucket.as_str(), &tier_name)
        .await
        .expect("Failed to set destination lifecycle configuration");
    let _ = upload_test_object(&ecstore, src_bucket.as_str(), src_object, payload).await;

    let copy_input = CopyObjectInput::builder()
        .copy_source(CopySource::Bucket {
            bucket: src_bucket.clone().into(),
            key: src_object.to_string().into(),
            version_id: None,
        })
        .bucket(dst_bucket.clone())
        .key(dst_object.to_string())
        .build()
        .unwrap();

    with_forced_immediate_enqueue_timeout(|| async {
        Box::pin(usecase.execute_copy_object(build_request(copy_input, Method::PUT)))
            .await
            .expect("Failed to copy object through usecase");
    })
    .await;

    let info = wait_for_transition(&ecstore, dst_bucket.as_str(), dst_object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("copied object should eventually transition after compensation backfill");

    assert_eq!(info.transitioned_object.status, "complete");
    assert_eq!(info.transitioned_object.tier, tier_name);
    assert!(backend.contains(&info.transitioned_object.name).await);
}

#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
#[test]
fn compensation_driven_complete_multipart_upload_still_transitions() {
    std::thread::Builder::new()
        .name("lifecycle-compensation-complete-multipart".to_string())
        .stack_size(32 * 1024 * 1024)
        .spawn(|| {
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("compensation multipart test runtime should build");

            runtime.block_on(compensation_driven_complete_multipart_upload_still_transitions_inner());
        })
        .expect("compensation multipart test thread should spawn")
        .join()
        .expect("compensation multipart test thread should finish");
}

async fn compensation_driven_complete_multipart_upload_still_transitions_inner() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultMultipartUsecase::from_global();

    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;

    let bucket = format!("test-comp-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/multipart.txt";
    let payload = b"multipart should still transition after compensation";

    create_test_bucket(&ecstore, bucket.as_str()).await;
    set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name)
        .await
        .expect("Failed to set lifecycle configuration");

    let upload = ecstore
        .new_multipart_upload(bucket.as_str(), object, &ObjectOptions::default())
        .await
        .expect("Failed to create multipart upload");

    let mut reader = PutObjReader::from_vec(payload.to_vec());
    let uploaded_part = ecstore
        .put_object_part(bucket.as_str(), object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default())
        .await
        .expect("Failed to upload multipart part");

    let complete_input = CompleteMultipartUploadInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .upload_id(upload.upload_id.clone())
        .multipart_upload(Some(CompletedMultipartUpload {
            parts: Some(vec![CompletedPart {
                part_number: Some(1),
                e_tag: uploaded_part.etag.clone().map(|etag| to_s3s_etag(&etag)),
                ..Default::default()
            }]),
        }))
        .build()
        .unwrap();

    with_forced_immediate_enqueue_timeout(|| async {
        Box::pin(usecase.execute_complete_multipart_upload(build_request(complete_input, Method::POST)))
            .await
            .expect("Failed to complete multipart upload through usecase");
    })
    .await;

    let info = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("multipart object should eventually transition after compensation backfill");

    assert_eq!(info.transitioned_object.status, "complete");
    assert_eq!(info.transitioned_object.tier, tier_name);
    assert!(backend.contains(&info.transitioned_object.name).await);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn compensation_driven_transition_still_cleans_remote_tier_on_delete() {
    use super::storage_api::test::ReqInfo;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();

    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;

    let bucket = format!("test-compensation-delete-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/object.txt";
    let payload = b"compensation should still preserve delete cleanup";

    create_test_bucket(&ecstore, bucket.as_str()).await;
    set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name)
        .await
        .expect("Failed to set lifecycle configuration");

    with_forced_immediate_enqueue_timeout(|| async {
        let _ = upload_test_object(&ecstore, bucket.as_str(), object, payload).await;
    })
    .await;

    let transitioned = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("object should eventually transition after compensation backfill");
    let remote_object = transitioned.transitioned_object.name.clone();

    assert!(backend.contains(&remote_object).await);

    let mut req = build_request(
        DeleteObjectInput::builder()
            .bucket(bucket.clone())
            .key(object.to_string())
            .build()
            .unwrap(),
        Method::DELETE,
    );
    insert_header(&mut req.headers, SUFFIX_FORCE_DELETE, "true");
    req.extensions.insert(ReqInfo {
        cred: Some(rustfs_credentials::Credentials::default()),
        is_owner: true,
        ..Default::default()
    });

    Box::pin(usecase.execute_delete_object(req))
        .await
        .expect("Failed to delete object through usecase after compensation-driven transition");

    assert!(
        wait_for_object_absence(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT).await,
        "object should be removed from hot tier after delete usecase"
    );

    assert!(
        backend.wait_for_remote_absence(&remote_object, TRANSITION_WAIT_TIMEOUT).await,
        "transitioned object should be removed from remote tier after delete usecase"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn compensation_driven_versioned_delete_still_creates_delete_marker() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();

    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;

    let bucket = format!("test-comp-versioned-delete-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/object.txt";
    let payload = b"versioned delete should preserve transitioned remote version behind delete marker";

    create_test_bucket(&ecstore, bucket.as_str()).await;
    set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name)
        .await
        .expect("Failed to set lifecycle configuration");

    with_forced_immediate_enqueue_timeout(|| async {
        let _ = upload_test_object(&ecstore, bucket.as_str(), object, payload).await;
    })
    .await;

    let transitioned = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("object should eventually transition after compensation backfill");
    let remote_object = transitioned.transitioned_object.name.clone();

    assert!(backend.contains(&remote_object).await);

    let req = build_request(
        DeleteObjectInput::builder()
            .bucket(bucket.clone())
            .key(object.to_string())
            .build()
            .unwrap(),
        Method::DELETE,
    );

    Box::pin(usecase.execute_delete_object(req))
        .await
        .expect("Failed to issue versioned delete after compensation-driven transition");

    assert!(
        wait_for_delete_marker(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT).await,
        "versioned delete should create a delete marker after compensation-driven transition"
    );
    assert!(
        backend.contains(&remote_object).await,
        "creating a delete marker should not remove the transitioned remote object version"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn compensation_driven_delete_marker_still_honors_lifecycle_cleanup() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();
    let bucket_usecase = DefaultBucketUsecase::from_global();

    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;

    let bucket = format!("test-comp-del-marker-cleanup-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/object.txt";
    let payload = b"delete marker lifecycle should still clean up after compensation-driven transition";

    create_test_bucket(&ecstore, bucket.as_str()).await;
    set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name)
        .await
        .expect("Failed to set transition lifecycle configuration");

    with_forced_immediate_enqueue_timeout(|| async {
        let _ = upload_test_object(&ecstore, bucket.as_str(), object, payload).await;
    })
    .await;

    let transitioned = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
        .await
        .expect("object should eventually transition after compensation backfill");
    let remote_object = transitioned.transitioned_object.name.clone();

    assert!(backend.contains(&remote_object).await);

    let req = build_request(
        DeleteObjectInput::builder()
            .bucket(bucket.clone())
            .key(object.to_string())
            .build()
            .unwrap(),
        Method::DELETE,
    );

    Box::pin(usecase.execute_delete_object(req))
        .await
        .expect("Failed to issue versioned delete after compensation-driven transition");

    assert!(
        wait_for_delete_marker(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT).await,
        "versioned delete should create a delete marker before lifecycle cleanup"
    );
    assert!(
        backend.contains(&remote_object).await,
        "delete marker creation should keep the transitioned remote object version"
    );

    let req = build_request(
        PutBucketLifecycleConfigurationInput::builder()
            .bucket(bucket.clone())
            .lifecycle_configuration(Some(expiration_lifecycle_configuration("test/")))
            .build()
            .unwrap(),
        Method::PUT,
    );
    bucket_usecase
        .execute_put_bucket_lifecycle_configuration(req)
        .await
        .expect("Failed to update lifecycle configuration for delete marker cleanup");

    assert!(
        wait_for_delete_marker(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT).await,
        "delete marker should remain visible after lifecycle update until cleanup completes"
    );
    assert!(
        backend.contains(&remote_object).await,
        "delete marker lifecycle cleanup should not remove the transitioned remote object version"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn put_bucket_lifecycle_configuration_expires_existing_objects() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultBucketUsecase::from_global();

    let bucket = format!("test-api-expire-existing-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/existing.txt";
    let payload = b"expire existing object after lifecycle update";

    create_test_bucket(&ecstore, bucket.as_str()).await;
    let _ = upload_test_object(&ecstore, bucket.as_str(), object, payload).await;

    let req = build_request(
        PutBucketLifecycleConfigurationInput::builder()
            .bucket(bucket.clone())
            .lifecycle_configuration(Some(expiration_lifecycle_configuration("test/")))
            .build()
            .unwrap(),
        Method::PUT,
    );

    usecase
        .execute_put_bucket_lifecycle_configuration(req)
        .await
        .expect("Failed to update lifecycle configuration");

    assert!(
        wait_for_delete_marker(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT).await,
        "existing object should be lifecycle-deleted after lifecycle update"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn put_bucket_lifecycle_configuration_allows_expired_delete_marker_on_object_lock_bucket() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultBucketUsecase::from_global();

    let bucket = format!("test-lock-expired-del-marker-{}", &Uuid::new_v4().simple().to_string()[..8]);
    create_test_bucket(&ecstore, bucket.as_str()).await;
    set_bucket_object_lock_enabled(bucket.as_str())
        .await
        .expect("Failed to enable object lock for bucket");

    let req = build_request(
        PutBucketLifecycleConfigurationInput::builder()
            .bucket(bucket.clone())
            .lifecycle_configuration(Some(expired_object_delete_marker_lifecycle_configuration()))
            .build()
            .unwrap(),
        Method::PUT,
    );

    usecase
        .execute_put_bucket_lifecycle_configuration(req)
        .await
        .expect("ExpiredObjectDeleteMarker should be accepted on object-lock bucket");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn put_bucket_lifecycle_configuration_rejects_del_marker_expiration_on_object_lock_bucket() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultBucketUsecase::from_global();

    let bucket = format!("test-lock-del-marker-exp-{}", &Uuid::new_v4().simple().to_string()[..8]);
    create_test_bucket(&ecstore, bucket.as_str()).await;
    set_bucket_object_lock_enabled(bucket.as_str())
        .await
        .expect("Failed to enable object lock for bucket");

    let req = build_request(
        PutBucketLifecycleConfigurationInput::builder()
            .bucket(bucket.clone())
            .lifecycle_configuration(Some(del_marker_expiration_lifecycle_configuration(1)))
            .build()
            .unwrap(),
        Method::PUT,
    );

    let err = usecase
        .execute_put_bucket_lifecycle_configuration(req)
        .await
        .expect_err("DelMarkerExpiration must be rejected on object-lock bucket");
    assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidArgument);
    let message = err.message().unwrap_or_default();
    assert!(
        message.contains(
            "ExpiredObjectAllVersions element and DelMarkerExpiration action cannot be used on an object locked bucket"
        ),
        "unexpected error message: {message}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn put_bucket_lifecycle_configuration_rejects_zero_day_del_marker_expiration_on_object_lock_bucket() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultBucketUsecase::from_global();

    let bucket = format!("test-lock-del-marker-exp0-{}", &Uuid::new_v4().simple().to_string()[..8]);
    create_test_bucket(&ecstore, bucket.as_str()).await;
    set_bucket_object_lock_enabled(bucket.as_str())
        .await
        .expect("Failed to enable object lock for bucket");

    let req = build_request(
        PutBucketLifecycleConfigurationInput::builder()
            .bucket(bucket.clone())
            .lifecycle_configuration(Some(del_marker_expiration_lifecycle_configuration(0)))
            .build()
            .unwrap(),
        Method::PUT,
    );

    let err = usecase
        .execute_put_bucket_lifecycle_configuration(req)
        .await
        .expect_err("DelMarkerExpiration with zero days must be rejected on object-lock bucket");
    assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidArgument);
    let message = err.message().unwrap_or_default();
    assert!(
        message.contains("Days must be a positive integer with DelMarkerExpiration"),
        "unexpected error message: {message}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn put_bucket_lifecycle_configuration_rejects_zero_day_expiration() {
    // S3 compatibility (ceph s3-tests `test_lifecycle_expiration_days0`): a rule with
    // Expiration{Days:0} must be rejected with InvalidArgument through the real
    // PutBucketLifecycleConfiguration handler, not merely at the crates/lifecycle
    // io::Error layer. No object lock required.
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultBucketUsecase::from_global();

    let bucket = format!("test-zero-day-exp-{}", &Uuid::new_v4().simple().to_string()[..8]);
    create_test_bucket(&ecstore, bucket.as_str()).await;

    let lifecycle = BucketLifecycleConfiguration {
        expiry_updated_at: None,
        rules: vec![LifecycleRule {
            status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
            abort_incomplete_multipart_upload: None,
            del_marker_expiration: None,
            expiration: Some(LifecycleExpiration {
                days: Some(0),
                ..Default::default()
            }),
            filter: Some(LifecycleRuleFilter {
                prefix: Some("days0/".to_string()),
                ..Default::default()
            }),
            id: Some("expire-zero-days".to_string()),
            noncurrent_version_expiration: None,
            noncurrent_version_transitions: None,
            prefix: None,
            transitions: None,
        }],
    };

    let req = build_request(
        PutBucketLifecycleConfigurationInput::builder()
            .bucket(bucket.clone())
            .lifecycle_configuration(Some(lifecycle))
            .build()
            .unwrap(),
        Method::PUT,
    );

    let err = usecase
        .execute_put_bucket_lifecycle_configuration(req)
        .await
        .expect_err("zero-day expiration must be rejected with InvalidArgument");
    assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidArgument);
    let message = err.message().unwrap_or_default();
    assert!(
        message.contains("'Days' for Expiration action must be a positive integer"),
        "unexpected error message: {message}"
    );
}

/// backlog#1148 ilm-8: the RestoreObject API surface on a transitioned object.
///
/// POST restore(days=1) is accepted and flips the object to
/// `x-amz-restore: ongoing-request="true"` while the mock tier GET barrier
/// proves the background copy-back has reached the remote read; a second POST
/// during that window is rejected with 409 `RestoreAlreadyInProgress`.
/// Synchronous SetDisks transition tests cover copy-back completion, restore
/// metadata, and local byte-identical reads.
///
/// Re-enabled in the serial lane by backlog#1304: the accept path now flips
/// the ongoing flag under a short compare-and-set guard and the copy-back
/// runs without the #4877 whole-copy-back lock, so the mid-restore
/// `ongoing-request="true"` read and the fast 409 rejection asserted here are
/// the implemented contract.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-8)"]
async fn restore_object_usecase_reports_ongoing_conflict() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();

    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;

    let bucket = format!("test-api-restore-{}", &Uuid::new_v4().simple().to_string()[..8]);
    // Keep the object under the shared ILM test prefix even though this setup
    // transitions it directly; it keeps diagnostics aligned with sibling tests.
    let object = "test/restore/api-object.bin";
    let payload: Vec<u8> = (0..128 * 1024).map(|i| (i % 251) as u8).collect();

    create_test_bucket(&ecstore, bucket.as_str()).await;
    let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
    assert!(uploaded.version_id.is_none(), "fixture must create the null version");
    let _ = transition_uploaded_object_directly(&ecstore, bucket.as_str(), object, &tier_name, &uploaded).await;
    backend.clear_op_log().await;
    let get_barrier = backend.arm_get_barrier().await;

    let restore_request = || RestoreRequest {
        days: Some(1),
        description: None,
        glacier_job_parameters: None,
        output_location: None,
        select_parameters: None,
        tier: None,
        type_: None,
    };
    let restore_input = || {
        RestoreObjectInput::builder()
            .bucket(bucket.clone())
            .key(object.to_string())
            .restore_request(Some(restore_request()))
            .build()
            .unwrap()
    };

    Box::pin(usecase.execute_restore_object(build_request(restore_input(), Method::POST)))
        .await
        .expect("restore request should be accepted");

    get_barrier.wait_until_paused().await;

    // The barrier proves the detached copy-back reached the tier GET and is
    // still paused, so the ongoing state and conflict rejection are not timing
    // assumptions about task scheduling.
    let ongoing = ecstore
        .get_object_info(bucket.as_str(), object, &ObjectOptions::default())
        .await
        .expect("Failed to load object info during restore");
    assert!(
        ongoing.restore_ongoing,
        "x-amz-restore must report ongoing-request=true while the copy-back tier GET is paused"
    );

    // A second restore while one is in flight is rejected.
    let err = Box::pin(usecase.execute_restore_object(build_request(restore_input(), Method::POST)))
        .await
        .expect_err("a second restore while one is ongoing must be rejected");
    assert_eq!(
        err.code(),
        &s3s::S3ErrorCode::RestoreAlreadyInProgress,
        "unexpected rejection for a repeated restore: {err:?}"
    );

    get_barrier.release();
}

/// rustfs/backlog#1337: cancellation after the ongoing metadata commit but
/// before detached worker creation must not strand the object forever. The
/// replacement POST proves the abandoned v1 generation has no live worker,
/// atomically supersedes it, and performs exactly one remote copy-back.
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1337"]
#[test]
fn restore_object_usecase_recovers_cancelled_post_commit_generation() {
    std::thread::Builder::new()
        .name("lifecycle-restore-orphan-recovery".to_string())
        .stack_size(32 * 1024 * 1024)
        .spawn(|| {
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("orphan restore recovery test runtime should build");
            runtime.block_on(restore_object_usecase_recovers_cancelled_post_commit_generation_inner());
        })
        .expect("orphan restore recovery test thread should spawn")
        .join()
        .expect("orphan restore recovery test thread should finish");
}

async fn restore_object_usecase_recovers_cancelled_post_commit_generation_inner() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;
    let bucket = format!("test-api-restore-orphan-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/restore/orphaned-generation.bin";
    let payload: Vec<u8> = (0..128 * 1024).map(|i| (i % 251) as u8).collect();

    create_test_bucket(&ecstore, bucket.as_str()).await;
    let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
    let _ = transition_uploaded_object_directly(&ecstore, bucket.as_str(), object, &tier_name, &uploaded).await;
    backend.clear_op_log().await;
    let tier_gets_before_restore = backend.get_count().await;

    let restore_input = || {
        RestoreObjectInput::builder()
            .bucket(bucket.clone())
            .key(object.to_string())
            .restore_request(Some(RestoreRequest {
                days: Some(1),
                description: None,
                glacier_job_parameters: None,
                output_location: None,
                select_parameters: None,
                tier: None,
                type_: None,
            }))
            .build()
            .expect("restore request should build")
    };

    let commit_barrier = crate::app::object::RestoreStatusCommitBarrier::install(bucket.as_str(), object);
    let first_input = restore_input();
    let first = tokio::spawn(async move {
        DefaultObjectUsecase::from_global()
            .execute_restore_object(build_request(first_input, Method::POST))
            .await
    });
    commit_barrier.wait_until_paused().await;

    first.abort();
    assert!(first.await.expect_err("the first request must be cancelled").is_cancelled());
    drop(commit_barrier);

    let orphaned = ecstore
        .get_object_info(bucket.as_str(), object, &ObjectOptions::default())
        .await
        .expect("the committed orphan generation should remain readable");
    assert!(orphaned.restore_ongoing, "the interrupted request must have committed ongoing=true");
    assert_eq!(
        rustfs_utils::http::get_consistent_str(orphaned.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_WORKER_LOCK,),
        Some(rustfs_utils::http::RESTORE_WORKER_LOCK_PROTOCOL_V1),
        "recoverable generations must advertise the worker-lock protocol"
    );
    let orphaned_operation_id =
        rustfs_utils::http::get_consistent_str(orphaned.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_OPERATION_ID)
            .and_then(|value| Uuid::parse_str(value).ok())
            .expect("recoverable generations must persist a non-nil operation id");
    assert!(!orphaned_operation_id.is_nil());

    DefaultObjectUsecase::from_global()
        .execute_restore_object(build_request(restore_input(), Method::POST))
        .await
        .expect("a new POST must supersede the committed generation whose worker lock was released");

    let completed = wait_for_restore_completion(&ecstore, &backend, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
        .await
        .unwrap_or_else(|err| panic!("{err}"));
    assert!(!completed.restore_ongoing);
    assert!(completed.restore_expires.is_some());
    assert!(
        rustfs_utils::http::get_str(completed.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_OPERATION_ID,).is_none(),
        "completed replacement restore must consume its operation id"
    );
    assert!(
        rustfs_utils::http::get_str(completed.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_WORKER_LOCK,).is_none(),
        "completed replacement restore must consume its liveness marker"
    );
    assert_eq!(
        backend.get_count().await - tier_gets_before_restore,
        1,
        "the cancelled pre-spawn generation must not issue a tier GET"
    );
}

#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#4879"]
#[test]
fn restore_object_usecase_completes_suspended_null_version_in_place() {
    std::thread::Builder::new()
        .name("lifecycle-restore-suspended-null".to_string())
        .stack_size(32 * 1024 * 1024)
        .spawn(|| {
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("suspended null-version restore test runtime should build");

            runtime.block_on(restore_object_usecase_completes_suspended_null_version_in_place_inner());
        })
        .expect("suspended null-version restore test thread should spawn")
        .join()
        .expect("suspended null-version restore test thread should finish");
}

async fn restore_object_usecase_completes_suspended_null_version_in_place_inner() {
    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();
    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;
    let bucket = format!("test-api-restore-suspended-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/restore/suspended-null.bin";
    let payload: Vec<u8> = (0..128 * 1024).map(|i| (i % 251) as u8).collect();

    create_test_bucket(&ecstore, bucket.as_str()).await;
    let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
    assert!(uploaded.version_id.is_none(), "fixture must create the null version");
    let _ = transition_uploaded_object_directly(&ecstore, bucket.as_str(), object, &tier_name, &uploaded).await;
    suspend_test_bucket(bucket.as_str()).await;
    backend.clear_op_log().await;
    let tier_gets_before_restore = backend.get_count().await;
    let get_barrier = backend.arm_get_barrier().await;

    Box::pin(
        usecase.execute_restore_object(build_request(
            RestoreObjectInput::builder()
                .bucket(bucket.clone())
                .key(object.to_string())
                .restore_request(Some(RestoreRequest {
                    days: Some(1),
                    description: None,
                    glacier_job_parameters: None,
                    output_location: None,
                    select_parameters: None,
                    tier: None,
                    type_: None,
                }))
                .build()
                .expect("restore request should build"),
            Method::POST,
        )),
    )
    .await
    .expect("suspended null-version restore should be accepted");

    get_barrier.wait_until_paused().await;
    get_barrier.release();
    let completed = wait_for_restore_completion(&ecstore, &backend, bucket.as_str(), object, RESTORE_SUSPENDED_WAIT_TIMEOUT)
        .await
        .unwrap_or_else(|err| panic!("{err}"));

    assert!(!completed.restore_ongoing, "the original null version must complete in place");
    assert!(completed.restore_expires.is_some(), "completed restore must carry an expiry");
    assert!(
        completed.version_id.is_none() || completed.version_id.is_some_and(|version_id| version_id.is_nil()),
        "suspended restore must remain on the null version"
    );
    assert_eq!(
        live_object_version_count(&ecstore, bucket.as_str(), object).await,
        1,
        "suspended restore must not create a UUID version"
    );
    assert_eq!(
        backend.get_count().await - tier_gets_before_restore,
        1,
        "suspended restore copy-back must fetch the tier exactly once"
    );
}

/// backlog#1304: the restore-accept compare-and-set itself, under real
/// concurrency. Two POST ?restore for the same transitioned object race each
/// other from separate tasks; the accept guard must let exactly one through —
/// the winner flips `ongoing-request="true"`, the loser must read that flag
/// and be rejected with 409 `RestoreAlreadyInProgress` — and the tier must
/// see exactly one copy-back GET, never two. Without the accept guard both
/// requests can observe ongoing=false and double-start the copy-back (the
/// #4859 race), which this test catches via the tier GET count.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-8)"]
async fn restore_object_usecase_accepts_exactly_one_of_two_concurrent_restores() {
    let (_disk_paths, ecstore) = setup_test_env().await;

    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
    let backend = register_mock_tier(&tier_name).await;

    let bucket = format!("test-api-restore-cas-{}", &Uuid::new_v4().simple().to_string()[..8]);
    // Keep the object under the shared ILM test prefix for diagnostics parity.
    let object = "test/restore/cas-object.bin";
    let payload: Vec<u8> = (0..128 * 1024).map(|i| (i % 251) as u8).collect();

    create_test_bucket(&ecstore, bucket.as_str()).await;
    let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
    let _ = transition_uploaded_object_directly(&ecstore, bucket.as_str(), object, &tier_name, &uploaded).await;
    backend.clear_op_log().await;

    // Hold the accepted copy-back at the tier GET until both accept attempts
    // return, so the loser cannot observe an already-restored object.
    let get_barrier = backend.arm_get_barrier().await;

    let tier_gets_before_restore = backend.get_count().await;

    let restore_input = || {
        RestoreObjectInput::builder()
            .bucket(bucket.clone())
            .key(object.to_string())
            .restore_request(Some(RestoreRequest {
                days: Some(1),
                description: None,
                glacier_job_parameters: None,
                output_location: None,
                select_parameters: None,
                tier: None,
                type_: None,
            }))
            .build()
            .unwrap()
    };
    let spawn_restore = |input: RestoreObjectInput| {
        tokio::spawn(async move {
            let usecase = DefaultObjectUsecase::from_global();
            Box::pin(usecase.execute_restore_object(build_request(input, Method::POST))).await
        })
    };

    let (first, second) = tokio::join!(spawn_restore(restore_input()), spawn_restore(restore_input()));
    let results = [
        first.expect("first concurrent restore task panicked"),
        second.expect("second concurrent restore task panicked"),
    ];

    let accepted = results.iter().filter(|result| result.is_ok()).count();
    assert_eq!(accepted, 1, "exactly one of two concurrent restores must be accepted, got {accepted}");
    let rejection = results
        .iter()
        .find_map(|result| result.as_ref().err())
        .expect("the losing concurrent restore must surface an error");
    assert_eq!(
        rejection.code(),
        &s3s::S3ErrorCode::RestoreAlreadyInProgress,
        "the losing concurrent restore must be rejected as already in progress: {rejection:?}"
    );

    get_barrier.wait_until_paused().await;
    get_barrier.release();

    // This test is scoped to the accept CAS: completion, expiry metadata, and
    // local restored GET service are covered by the single-request restore test
    // above. Here it is enough to prove exactly one copy-back was admitted.
    let expected_tier_gets = tier_gets_before_restore + 1;
    let deadline = tokio::time::Instant::now() + TRANSITION_WAIT_TIMEOUT;
    loop {
        let actual_tier_gets = backend.get_count().await;
        if actual_tier_gets >= expected_tier_gets {
            assert_eq!(
                actual_tier_gets - tier_gets_before_restore,
                1,
                "two concurrent restore requests must trigger exactly one tier copy-back GET"
            );
            break;
        }
        if tokio::time::Instant::now() >= deadline {
            let op_log = backend.op_log().await;
            panic!(
                "mock tier should record exactly one restore GET within {TRANSITION_WAIT_TIMEOUT:?}; \
                 tier_gets_before_restore={tier_gets_before_restore}, actual_tier_gets={actual_tier_gets}, op_log={op_log:?}"
            );
        }

        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

/// rustfs/backlog#1320: a single PUT must compute the replication decision
/// exactly once. Historically the PUT path computed `must_replicate_object`
/// twice with the same inputs — once before commit to persist the pending
/// replication metadata, and again after commit to drive the schedule. Besides
/// repeating the versioning/config/target traversal on the hot path, a
/// replication-config hot update between the two computations could split the
/// two phases (pending-without-schedule or schedule-without-pending). The fix
/// reuses one immutable decision for both phases; this test observes the
/// computation counter and fails if a second computation is reintroduced.
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state usecase integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn put_object_computes_replication_decision_exactly_once() {
    use super::storage_api::object_usecase::bucket::replication::MUST_REPLICATE_OBJECT_CALLS;
    use std::sync::atomic::Ordering;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let fs = FS::new();
    let usecase = DefaultObjectUsecase::from_global();

    let bucket = format!("test-repl-decision-once-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "test/object.txt";
    let payload = b"replication decision single-compute payload";

    create_test_bucket(&ecstore, bucket.as_str()).await;

    // Reset immediately before the single PUT so the counter reflects only this
    // request. `#[serial]` guarantees no other usecase PUT runs concurrently, so
    // no unrelated `must_replicate_object` call can perturb the count.
    MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst);

    let input = PutObjectInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .body(Some(streaming_blob_from_bytes(payload)))
        .content_length(Some(payload.len() as i64))
        .build()
        .unwrap();
    Box::pin(usecase.execute_put_object(&fs, build_request(input, Method::PUT)))
        .await
        .expect("Failed to upload object through usecase");

    let calls = MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst);
    assert_eq!(
        calls, 1,
        "a single PUT must compute the replication decision exactly once; reverting to a \
         pre-commit + post-commit recompute makes this observe 2 (rustfs/backlog#1320)"
    );
}

/// CopyObject creates a new object on the destination key, so it must join
/// bucket replication exactly like PutObject: compute one replication decision
/// that drives both the persisted pending marker and the post-commit schedule
/// (MinIO CopyObjectHandler parity). A count of 0 means a copied object is
/// never scheduled for replication and — because the copy path clones the
/// source metadata wholesale — the destination silently inherits the source's
/// replication bookkeeping, faking a COMPLETED/REPLICA state for an object
/// that never replicated. The second half of this test pins the metadata
/// cleanup (MinIO filterReplicationStatusMetadata parity).
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state usecase integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn copy_object_computes_replication_decision_and_strips_stale_status() {
    use super::storage_api::object_usecase::bucket::replication::MUST_REPLICATE_OBJECT_CALLS;
    use rustfs_utils::http::{
        SUFFIX_REPLICA_STATUS, SUFFIX_REPLICA_TIMESTAMP, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, get_str,
        insert_str,
    };
    use std::collections::HashMap;
    use std::sync::atomic::Ordering;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();

    let bucket = format!("test-copy-repl-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let src_object = "copy/source.txt";
    let dst_object = "copy/destination.txt";
    let payload = b"copy replication decision payload";

    create_test_bucket(&ecstore, bucket.as_str()).await;

    // Seed the source object with stale replication bookkeeping, as if it had
    // already replicated elsewhere (internal status/timestamp under both
    // compatibility prefixes, replica state, and the surfaced x-amz header).
    let mut stale_metadata = HashMap::new();
    insert_str(
        &mut stale_metadata,
        SUFFIX_REPLICATION_STATUS,
        "arn:minio:replication::stale:dst=COMPLETED;".to_string(),
    );
    insert_str(&mut stale_metadata, SUFFIX_REPLICATION_TIMESTAMP, "2024-01-01T00:00:00Z".to_string());
    insert_str(&mut stale_metadata, SUFFIX_REPLICA_STATUS, "REPLICA".to_string());
    insert_str(&mut stale_metadata, SUFFIX_REPLICA_TIMESTAMP, "2024-01-01T00:00:00Z".to_string());
    stale_metadata.insert(AMZ_BUCKET_REPLICATION_STATUS.to_string(), "COMPLETED".to_string());
    let mut reader = PutObjReader::from_vec(payload.to_vec());
    (*ecstore)
        .put_object(
            bucket.as_str(),
            src_object,
            &mut reader,
            &ObjectOptions {
                user_defined: stale_metadata,
                ..Default::default()
            },
        )
        .await
        .expect("Failed to upload source object with stale replication metadata");

    MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst);

    let copy_input = CopyObjectInput::builder()
        .copy_source(CopySource::Bucket {
            bucket: bucket.clone().into(),
            key: src_object.into(),
            version_id: None,
        })
        .bucket(bucket.clone())
        .key(dst_object.to_string())
        .build()
        .unwrap();
    Box::pin(usecase.execute_copy_object(build_request(copy_input, Method::PUT)))
        .await
        .expect("Failed to copy object through usecase");

    assert_eq!(
        MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst),
        1,
        "CopyObject must compute the replication decision exactly once; 0 means the copied \
         object never enters bucket replication (P0-6, MinIO CopyObjectHandler parity)"
    );

    let copied = ecstore
        .get_object_info(bucket.as_str(), dst_object, &ObjectOptions::default())
        .await
        .expect("Failed to read copied destination object info");
    assert!(
        get_str(&copied.user_defined, SUFFIX_REPLICATION_STATUS).is_none(),
        "destination must not inherit the source's replication status; got {:?}",
        get_str(&copied.user_defined, SUFFIX_REPLICATION_STATUS)
    );
    assert!(
        get_str(&copied.user_defined, SUFFIX_REPLICATION_TIMESTAMP).is_none(),
        "destination must not inherit the source's replication timestamp"
    );
    assert!(
        get_str(&copied.user_defined, SUFFIX_REPLICA_STATUS).is_none(),
        "destination must not inherit the source's replica status"
    );
    assert!(
        !copied
            .user_defined
            .keys()
            .any(|k| k.eq_ignore_ascii_case(AMZ_BUCKET_REPLICATION_STATUS)),
        "destination must not inherit the surfaced x-amz-replication-status header key"
    );
}

/// Snowball auto-extract writes each archive member as an independent object,
/// so each member must join bucket replication like a regular PUT (MinIO
/// PutObjectExtract parity). The archive holds two file entries; a count other
/// than 2 means extracted objects bypass the replication decision entirely and
/// stay local forever.
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state usecase integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn put_object_extract_computes_replication_decision_per_entry() {
    use super::storage_api::object_usecase::bucket::replication::MUST_REPLICATE_OBJECT_CALLS;
    use super::storage_api::test::ReqInfo;
    use super::storage_api::test::bucket::metadata::BUCKET_POLICY_CONFIG;
    use std::io::Cursor;
    use std::sync::atomic::Ordering;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();

    let bucket = format!("test-extract-repl-{}", &Uuid::new_v4().simple().to_string()[..8]);
    create_test_bucket(&ecstore, bucket.as_str()).await;

    // The extract path re-authorizes every extracted entry internally. The test
    // harness has no IAM system, so allow anonymous PutObject via bucket policy.
    let policy_json = serde_json::json!({
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Principal": "*",
                "Action": ["s3:PutObject"],
                "Resource": [format!("arn:aws:s3:::{bucket}/*")]
            }
        ]
    })
    .to_string();
    metadata_sys::update(bucket.as_str(), BUCKET_POLICY_CONFIG, policy_json.into_bytes())
        .await
        .expect("Failed to install anonymous put bucket policy");

    let mut builder = tokio_tar::Builder::new(Cursor::new(Vec::new()));
    for (path, data) in [
        ("member-one.txt", b"first member payload".as_slice()),
        ("nested/member-two.txt", b"second member payload".as_slice()),
    ] {
        let mut header = tokio_tar::Header::new_gnu();
        header.set_size(data.len() as u64);
        header.set_mode(0o644);
        header.set_cksum();
        builder
            .append_data(&mut header, path, Cursor::new(data))
            .await
            .expect("Failed to append archive member");
    }
    let archive = builder.into_inner().await.expect("Failed to finish archive").into_inner();

    MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst);

    let input = PutObjectInput::builder()
        .bucket(bucket.clone())
        .key("archive.tar".to_string())
        .body(Some(streaming_blob_from_bytes(&archive)))
        .content_length(Some(archive.len() as i64))
        .build()
        .unwrap();
    let mut req = build_request(input, Method::PUT);
    req.extensions.insert(ReqInfo::default());
    Box::pin(usecase.execute_put_object_extract(req))
        .await
        .expect("Failed to extract archive through usecase");

    assert_eq!(
        MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst),
        2,
        "snowball auto-extract must compute one replication decision per extracted member \
         object; 0 means extracted objects bypass bucket replication (P0-6 companion, MinIO \
         PutObjectExtract parity)"
    );
}

/// The object-lock handlers do not go through the object PUT path, so they must schedule
/// replication themselves. Without it a retention or legal hold applied after upload stays
/// local and the replica keeps its previous, unprotected state (a WORM object that is still
/// deletable on the peer). This observes the shared decision counter: each handler must
/// compute a replication decision exactly once, which is the step that drives both the
/// persisted pending marker and the schedule.
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state usecase integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
async fn object_lock_handlers_schedule_replication() {
    use super::storage_api::object_usecase::bucket::replication::MUST_REPLICATE_OBJECT_CALLS;
    use s3s::S3;
    use std::sync::atomic::Ordering;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let fs = FS::new();

    let bucket = format!("test-lock-repl-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "locked/object.txt";

    (*ecstore)
        .make_bucket(
            bucket.as_str(),
            &MakeBucketOptions {
                lock_enabled: true,
                versioning_enabled: true,
                ..Default::default()
            },
        )
        .await
        .expect("Failed to create object-lock bucket");
    upload_test_object(&ecstore, bucket.as_str(), object, b"worm payload").await;

    // PutObjectRetention must compute a replication decision.
    MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst);
    let retention_input = PutObjectRetentionInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .retention(Some(ObjectLockRetention {
            mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::COMPLIANCE)),
            retain_until_date: Some(time::macros::datetime!(2030-01-01 00:00:00 UTC).into()),
        }))
        .build()
        .unwrap();
    fs.put_object_retention(build_request(retention_input, Method::PUT))
        .await
        .expect("put_object_retention should succeed");
    assert_eq!(
        MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst),
        1,
        "PutObjectRetention must compute the replication decision exactly once; 0 means the \
         retention change never replicates and the peer copy stays unprotected"
    );

    // PutObjectLegalHold must do the same.
    MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst);
    let legal_hold_input = PutObjectLegalHoldInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .legal_hold(Some(ObjectLockLegalHold {
            status: Some(ObjectLockLegalHoldStatus::from_static(ObjectLockLegalHoldStatus::ON)),
        }))
        .build()
        .unwrap();
    fs.put_object_legal_hold(build_request(legal_hold_input, Method::PUT))
        .await
        .expect("put_object_legal_hold should succeed");
    assert_eq!(
        MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst),
        1,
        "PutObjectLegalHold must compute the replication decision exactly once; 0 means the \
         legal hold never replicates and the peer copy stays deletable"
    );
}

/// Regression pin for the DeleteObjects hot path: the bucket's versioning
/// configuration must be resolved exactly once per request — not once per
/// key, which used to cost up to 1000 identical metadata lookups per call.
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"]
async fn delete_objects_resolves_bucket_versioning_once_per_request() {
    use super::storage_api::test::bucket::replication::DELETE_CONFIG_SNAPSHOT_LOADS;
    use super::storage_api::test::{ReqInfo, VERSIONING_CONFIG_LOOKUPS};
    use std::sync::atomic::Ordering;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();

    let bucket = format!("test-delobjs-vers-once-{}", &Uuid::new_v4().simple().to_string()[..8]);
    create_test_bucket(&ecstore, bucket.as_str()).await;

    let keys = ["batch/a.txt", "batch/b.txt", "batch/c.txt"];
    for key in keys {
        upload_test_object(&ecstore, bucket.as_str(), key, b"delete objects versioning payload").await;
    }

    let input = DeleteObjectsInput::builder()
        .bucket(bucket.clone())
        .delete(Delete {
            objects: keys
                .iter()
                .map(|key| ObjectIdentifier {
                    key: key.to_string(),
                    version_id: None,
                    ..Default::default()
                })
                .collect(),
            quiet: None,
        })
        .build()
        .expect("delete objects input should build");

    let mut req = build_request(input, Method::POST);
    req.extensions.insert(ReqInfo {
        cred: Some(rustfs_credentials::Credentials::default()),
        is_owner: true,
        ..Default::default()
    });

    let lookups_before = VERSIONING_CONFIG_LOOKUPS.load(Ordering::SeqCst);
    let snapshots_before = DELETE_CONFIG_SNAPSHOT_LOADS.load(Ordering::SeqCst);
    let response = usecase
        .execute_delete_objects(req)
        .await
        .expect("delete objects should succeed");

    let output = response.output;
    let errors = output.errors.unwrap_or_default();
    assert!(errors.is_empty(), "DeleteObjects reported per-key errors: {errors:?}");
    assert_eq!(
        output.deleted.unwrap_or_default().len(),
        keys.len(),
        "every requested key should be deleted"
    );

    let lookups = VERSIONING_CONFIG_LOOKUPS.load(Ordering::SeqCst) - lookups_before;
    let snapshots = DELETE_CONFIG_SNAPSHOT_LOADS.load(Ordering::SeqCst) - snapshots_before;
    assert_eq!(
        snapshots,
        1,
        "DeleteObjects must resolve delete admission config exactly once per request; got {snapshots} snapshots for {} keys",
        keys.len()
    );
    assert_eq!(lookups, 0, "DeleteObjects must not fall back to legacy per-key versioning lookups");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"]
async fn delete_objects_treats_empty_directory_version_as_absent() {
    use super::storage_api::test::ReqInfo;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();
    let bucket = format!("test-delobjs-empty-version-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "directory/";
    create_test_bucket(&ecstore, &bucket).await;
    upload_test_object(&ecstore, &bucket, object, b"").await;

    let input = DeleteObjectsInput::builder()
        .bucket(bucket)
        .delete(Delete {
            objects: vec![ObjectIdentifier {
                key: object.to_string(),
                version_id: Some(" \t ".to_string()),
                ..Default::default()
            }],
            quiet: None,
        })
        .build()
        .expect("delete objects input should build");
    let mut req = build_request(input, Method::POST);
    req.extensions.insert(ReqInfo {
        cred: Some(rustfs_credentials::Credentials::default()),
        is_owner: true,
        ..Default::default()
    });

    let response = usecase
        .execute_delete_objects(req)
        .await
        .expect("empty VersionId should be treated as an absent version");
    let deleted = response
        .output
        .deleted
        .expect("delete response should contain the directory entry");

    assert_eq!(deleted.len(), 1);
    assert_eq!(
        deleted[0].delete_marker, None,
        "the synthetic null version must be purged, not hidden by a new marker"
    );
    assert_eq!(deleted[0].version_id, None, "the synthetic null sentinel must stay off the wire");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"]
async fn delete_objects_preserves_explicit_null_directory_version() {
    use super::storage_api::test::ReqInfo;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();
    let bucket = format!("test-delobjs-null-version-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "directory/";
    create_test_bucket(&ecstore, &bucket).await;
    upload_test_object(&ecstore, &bucket, object, b"").await;

    let input = DeleteObjectsInput::builder()
        .bucket(bucket)
        .delete(Delete {
            objects: vec![ObjectIdentifier {
                key: object.to_string(),
                version_id: Some("null".to_string()),
                ..Default::default()
            }],
            quiet: None,
        })
        .build()
        .expect("delete objects input should build");
    let mut req = build_request(input, Method::POST);
    req.extensions.insert(ReqInfo {
        cred: Some(rustfs_credentials::Credentials::default()),
        is_owner: true,
        ..Default::default()
    });

    let response = usecase
        .execute_delete_objects(req)
        .await
        .expect("explicit null VersionId should delete the directory's null version");
    let deleted = response
        .output
        .deleted
        .expect("delete response should contain the directory entry");

    assert_eq!(deleted.len(), 1);
    assert_eq!(deleted[0].delete_marker, None);
    assert_eq!(deleted[0].version_id.as_deref(), Some("null"));
}

async fn install_malformed_versioning_config(bucket: &str) {
    use super::storage_api::test::{get_global_bucket_metadata_sys, set_bucket_metadata};

    let sys = get_global_bucket_metadata_sys().expect("bucket metadata system should be initialized");
    let metadata = {
        let sys = sys.read().await;
        sys.get(bucket)
            .await
            .expect("bucket metadata should be cached before corruption injection")
    };
    let mut metadata = (*metadata).clone();
    metadata.versioning_config_xml = b"<VersioningConfiguration>".to_vec();
    metadata.versioning_config = None;
    set_bucket_metadata(bucket.to_string(), metadata)
        .await
        .expect("malformed versioning metadata should be installed for the request test");
}

async fn install_malformed_replication_config(bucket: &str) {
    use super::storage_api::test::{get_global_bucket_metadata_sys, set_bucket_metadata};

    let sys = get_global_bucket_metadata_sys().expect("bucket metadata system should be initialized");
    let metadata = {
        let sys = sys.read().await;
        sys.get(bucket)
            .await
            .expect("bucket metadata should be cached before corruption injection")
    };
    let mut metadata = (*metadata).clone();
    metadata.replication_config_xml = b"<ReplicationConfiguration>".to_vec();
    metadata.replication_config = None;
    set_bucket_metadata(bucket.to_string(), metadata)
        .await
        .expect("malformed replication metadata should be installed for the request test");
}

async fn install_delete_replication_config(
    bucket: &str,
    delete_enabled: bool,
    replica_modifications: bool,
    required_tag: Option<(&str, &str)>,
) {
    use super::storage_api::test::{bucket::utils::serialize, get_global_bucket_metadata_sys, set_bucket_metadata};

    let sys = get_global_bucket_metadata_sys().expect("bucket metadata system should be initialized");
    let metadata = {
        let sys = sys.read().await;
        sys.get(bucket)
            .await
            .expect("bucket metadata should be cached before replication config injection")
    };
    let mut metadata = (*metadata).clone();
    let target = "arn:aws:s3:::target-bucket";
    let delete_status = if delete_enabled {
        DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED)
    } else {
        DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED)
    };

    metadata.versioning_config_xml = b"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>".to_vec();
    metadata.versioning_config = Some(VersioningConfiguration {
        status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
        ..Default::default()
    });
    let filter = required_tag.map(|(key, value)| ReplicationRuleFilter {
        tag: Some(Tag {
            key: Some(key.to_string()),
            value: Some(value.to_string()),
        }),
        ..Default::default()
    });
    let source_selection_criteria = replica_modifications.then(|| SourceSelectionCriteria {
        replica_modifications: Some(ReplicaModifications {
            status: ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED),
        }),
        sse_kms_encrypted_objects: None,
    });
    let replication_config = ReplicationConfiguration {
        role: String::new(),
        rules: vec![ReplicationRule {
            delete_marker_replication: Some(DeleteMarkerReplication {
                status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)),
            }),
            delete_replication: Some(DeleteReplication { status: delete_status }),
            destination: Destination {
                bucket: target.to_string(),
                ..Default::default()
            },
            existing_object_replication: None,
            filter,
            id: Some("delete".to_string()),
            prefix: Some(String::new()),
            priority: Some(1),
            source_selection_criteria,
            status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
        }],
    };
    metadata.replication_config_xml = serialize(&replication_config).expect("replication test config should serialize");
    metadata.replication_config = Some(replication_config);
    set_bucket_metadata(bucket.to_string(), metadata)
        .await
        .expect("delete replication metadata should be installed for the request test");
}

#[tokio::test]
#[serial]
async fn delete_object_versioning_config_failure_leaves_latest_object_intact() {
    use super::storage_api::test::ReqInfo;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();
    let bucket = format!("test-delete-versioning-fail-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "object.txt";
    let payload = b"local delete must not commit before versioning admission";
    create_test_bucket(&ecstore, &bucket).await;
    upload_test_object(&ecstore, &bucket, object, payload).await;
    install_malformed_versioning_config(&bucket).await;

    let input = DeleteObjectInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .build()
        .expect("delete input should build");
    let mut req = build_request(input, Method::DELETE);
    req.extensions.insert(ReqInfo {
        cred: Some(rustfs_credentials::Credentials::default()),
        is_owner: true,
        ..Default::default()
    });

    let err = usecase
        .execute_delete_object(req)
        .await
        .expect_err("versioning config failure must reject DeleteObject");

    assert_eq!(err.code(), &s3s::S3ErrorCode::InternalError);
    assert_eq!(read_object_bytes(&ecstore, &bucket, object).await, payload);
}

#[tokio::test]
#[serial]
async fn delete_object_replication_config_failure_leaves_latest_object_intact() {
    use super::storage_api::test::ReqInfo;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();
    let bucket = format!("test-delete-repl-fail-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "object.txt";
    let payload = b"local delete must not commit before replication admission";
    create_test_bucket(&ecstore, &bucket).await;
    upload_test_object(&ecstore, &bucket, object, payload).await;
    install_malformed_replication_config(&bucket).await;

    let input = DeleteObjectInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .build()
        .expect("delete input should build");
    let mut req = build_request(input, Method::DELETE);
    req.extensions.insert(ReqInfo {
        cred: Some(rustfs_credentials::Credentials::default()),
        is_owner: true,
        ..Default::default()
    });

    let err = usecase
        .execute_delete_object(req)
        .await
        .expect_err("replication config failure must reject DeleteObject");

    assert_eq!(err.code(), &s3s::S3ErrorCode::InternalError);
    assert_eq!(read_object_bytes(&ecstore, &bucket, object).await, payload);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn delete_object_uses_one_replication_config_generation() {
    use super::object_usecase::install_delete_snapshot_test_hook;
    use super::storage_api::test::ReqInfo;
    use super::storage_api::test::bucket::replication::take_scheduled_replication_deletes;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();
    let bucket = format!("test-delete-config-generation-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "object.txt";
    create_test_bucket(&ecstore, &bucket).await;

    let mut reader = PutObjReader::from_vec(b"generation-bound delete".to_vec());
    let uploaded = ecstore
        .put_object(
            &bucket,
            object,
            &mut reader,
            &ObjectOptions {
                versioned: true,
                ..Default::default()
            },
        )
        .await
        .expect("versioned object should be uploaded");
    let version_id = uploaded.version_id.expect("versioned upload should return a version ID");
    let version_id_string = version_id.to_string();

    install_delete_replication_config(&bucket, true, false, None).await;
    assert!(take_scheduled_replication_deletes().is_empty());

    let loaded = Arc::new(Barrier::new(2));
    let resume = Arc::new(Barrier::new(2));
    install_delete_snapshot_test_hook(bucket.clone(), Arc::clone(&loaded), Arc::clone(&resume));

    let input = DeleteObjectInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .version_id(Some(version_id_string.clone()))
        .build()
        .expect("version delete input should build");
    let mut req = build_request(input, Method::DELETE);
    req.extensions.insert(ReqInfo {
        cred: Some(rustfs_credentials::Credentials::default()),
        is_owner: true,
        ..Default::default()
    });

    let update_config = async {
        loaded.wait().await;
        install_delete_replication_config(&bucket, false, false, None).await;
        resume.wait().await;
    };
    let (response, ()) = tokio::join!(usecase.execute_delete_object(req), update_config);
    let response = response.expect("delete admitted by the first config generation should succeed");

    assert_eq!(response.output.version_id.as_deref(), Some(version_id_string.as_str()));
    let scheduled = take_scheduled_replication_deletes();
    assert_eq!(scheduled.len(), 1, "the first config generation must control post-delete queueing");
    assert_eq!(scheduled[0].version_id, Some(version_id));
    let state = scheduled[0]
        .replication_state
        .as_ref()
        .expect("admitted version purge should carry replication state");
    assert_eq!(
        state.version_purge_status_internal.as_deref(),
        Some("arn:aws:s3:::target-bucket=PENDING;")
    );
}

#[tokio::test]
#[serial]
async fn replica_delete_uses_authorized_source_version_and_its_tags() {
    use super::object_usecase::install_delete_source_test_hook;
    use super::storage_api::test::ReqInfo;
    use super::storage_api::test::bucket::replication::{ReplicationStatusType, take_scheduled_replication_deletes};
    use rustfs_utils::http::headers::AMZ_OBJECT_TAGGING;
    use rustfs_utils::http::{SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID};
    use std::collections::HashMap;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();
    let bucket = format!("test-replica-source-version-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "object.txt";
    create_test_bucket(&ecstore, &bucket).await;
    install_delete_replication_config(&bucket, true, true, Some(("generation", "updated"))).await;

    let mut old_opts = ObjectOptions {
        versioned: true,
        user_defined: HashMap::from([(AMZ_OBJECT_TAGGING.to_string(), "generation=old".to_string())]),
        ..Default::default()
    };
    old_opts.set_replica_status(ReplicationStatusType::Replica);
    let mut old_reader = PutObjReader::from_vec(b"old replica version".to_vec());
    let old = ecstore
        .put_object(&bucket, object, &mut old_reader, &old_opts)
        .await
        .expect("old replica version should be uploaded");
    let old_version_id = old.version_id.expect("old replica version should have a version ID");
    let old_version_id_string = old_version_id.to_string();

    let mut latest_opts = ObjectOptions {
        versioned: true,
        user_defined: HashMap::from([(AMZ_OBJECT_TAGGING.to_string(), "generation=latest".to_string())]),
        ..Default::default()
    };
    latest_opts.set_replica_status(ReplicationStatusType::Replica);
    let mut latest_reader = PutObjReader::from_vec(b"latest replica version".to_vec());
    let latest = ecstore
        .put_object(&bucket, object, &mut latest_reader, &latest_opts)
        .await
        .expect("latest replica version should be uploaded");
    assert_ne!(latest.version_id, Some(old_version_id));
    assert!(take_scheduled_replication_deletes().is_empty());

    let source_loaded = Arc::new(Barrier::new(2));
    let resume_delete = Arc::new(Barrier::new(2));
    install_delete_source_test_hook(bucket.clone(), Arc::clone(&source_loaded), Arc::clone(&resume_delete));

    let input = DeleteObjectInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .build()
        .expect("replica delete input should build");
    let mut req = build_request(input, Method::DELETE);
    req.headers
        .insert(AMZ_BUCKET_REPLICATION_STATUS, HeaderValue::from_static("REPLICA"));
    insert_header(&mut req.headers, SUFFIX_SOURCE_VERSION_ID, old_version_id_string.clone());
    insert_header(&mut req.headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
    req.extensions.insert(ReqInfo {
        cred: Some(rustfs_credentials::Credentials::default()),
        is_owner: true,
        ..Default::default()
    });

    let update_tags = async {
        source_loaded.wait().await;
        ecstore
            .put_object_tags(
                &bucket,
                object,
                "generation=updated",
                &ObjectOptions {
                    versioned: true,
                    version_id: Some(old_version_id_string.clone()),
                    ..Default::default()
                },
            )
            .await
            .expect("tag update ordered before the locked delete should succeed");
        resume_delete.wait().await;
    };
    let (response, ()) = tokio::join!(usecase.execute_delete_object(req), update_tags);
    let response = response.expect("authorized replica version delete should succeed");

    assert_eq!(response.output.version_id.as_deref(), Some(old_version_id_string.as_str()));
    let scheduled = take_scheduled_replication_deletes();
    assert_eq!(
        scheduled.len(),
        1,
        "the tag committed after the advisory pre-read must control downstream fanout"
    );
    assert_eq!(scheduled[0].version_id, Some(old_version_id));
    assert_eq!(
        scheduled[0]
            .replication_state
            .as_ref()
            .and_then(|state| state.version_purge_status_internal.as_deref()),
        Some("arn:aws:s3:::target-bucket=PENDING;")
    );
    assert_eq!(read_object_bytes(&ecstore, &bucket, object).await, b"latest replica version");
}

#[tokio::test]
#[serial]
async fn force_delete_replication_config_failure_leaves_prefix_intact() {
    use super::storage_api::test::ReqInfo;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();
    let bucket = format!("test-force-delete-repl-fail-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let prefix = "prefix/";
    let object = "prefix/object.txt";
    let payload = b"force delete must validate replication metadata before deleting descendants";
    create_test_bucket(&ecstore, &bucket).await;
    upload_test_object(&ecstore, &bucket, object, payload).await;
    install_malformed_replication_config(&bucket).await;

    let input = DeleteObjectInput::builder()
        .bucket(bucket.clone())
        .key(prefix.to_string())
        .build()
        .expect("delete input should build");
    let mut req = build_request(input, Method::DELETE);
    insert_header(&mut req.headers, SUFFIX_FORCE_DELETE, "true");
    req.extensions.insert(ReqInfo {
        cred: Some(rustfs_credentials::Credentials::default()),
        is_owner: true,
        ..Default::default()
    });

    let err = usecase
        .execute_delete_object(req)
        .await
        .expect_err("replication config failure must reject force delete");

    assert_eq!(err.code(), &s3s::S3ErrorCode::InternalError);
    assert_eq!(read_object_bytes(&ecstore, &bucket, object).await, payload);
}

#[tokio::test]
#[serial]
async fn replica_delete_replication_config_failure_leaves_object_intact() {
    use super::storage_api::test::ReqInfo;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();
    let bucket = format!("test-replica-delete-repl-fail-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "object.txt";
    let payload = b"incoming replica delete must use the admitted request snapshot";
    create_test_bucket(&ecstore, &bucket).await;
    upload_test_object(&ecstore, &bucket, object, payload).await;
    install_malformed_replication_config(&bucket).await;

    let input = DeleteObjectInput::builder()
        .bucket(bucket.clone())
        .key(object.to_string())
        .build()
        .expect("delete input should build");
    let mut req = build_request(input, Method::DELETE);
    req.headers
        .insert(AMZ_BUCKET_REPLICATION_STATUS, HeaderValue::from_static("REPLICA"));
    req.extensions.insert(ReqInfo {
        cred: Some(rustfs_credentials::Credentials::default()),
        is_owner: true,
        ..Default::default()
    });

    let err = usecase
        .execute_delete_object(req)
        .await
        .expect_err("replication config failure must reject incoming replica delete");

    assert_eq!(err.code(), &s3s::S3ErrorCode::InternalError);
    assert_eq!(read_object_bytes(&ecstore, &bucket, object).await, payload);
}

#[tokio::test]
#[serial]
async fn delete_objects_replication_config_failure_does_not_call_storage_delete() {
    use super::storage_api::test::ReqInfo;

    let (_disk_paths, ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();
    let bucket = format!("test-delete-repls-fail-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let object = "object.txt";
    let payload = b"batch delete must fail closed on replication metadata errors";
    create_test_bucket(&ecstore, &bucket).await;
    upload_test_object(&ecstore, &bucket, object, payload).await;
    install_malformed_replication_config(&bucket).await;

    let input = DeleteObjectsInput::builder()
        .bucket(bucket.clone())
        .delete(Delete {
            objects: vec![ObjectIdentifier {
                key: object.to_string(),
                version_id: None,
                ..Default::default()
            }],
            quiet: None,
        })
        .build()
        .expect("delete objects input should build");
    let mut req = build_request(input, Method::POST);
    req.extensions.insert(ReqInfo {
        cred: Some(rustfs_credentials::Credentials::default()),
        is_owner: true,
        ..Default::default()
    });

    let err = usecase
        .execute_delete_objects(req)
        .await
        .expect_err("replication config failure must reject DeleteObjects");

    assert_eq!(err.code(), &s3s::S3ErrorCode::InternalError);
    assert_eq!(read_object_bytes(&ecstore, &bucket, object).await, payload);
}

/// Regression pin for the nonexistent-bucket fast path: GET/HEAD/DeleteObject
/// must decide NoSuchBucket from the bucket-existence check, before any bucket
/// versioning lookup runs — and cheap request-shape validation still wins over
/// bucket existence for GET.
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"]
async fn nonexistent_bucket_fails_before_versioning_lookup_on_get_head_delete() {
    use super::storage_api::test::VERSIONING_CONFIG_LOOKUPS;
    use std::sync::atomic::Ordering;

    let (_disk_paths, _ecstore) = setup_test_env().await;
    let usecase = DefaultObjectUsecase::from_global();

    // Never created.
    let bucket = format!("test-missing-{}", &Uuid::new_v4().simple().to_string()[..8]);

    // Request-shape validation keeps precedence over bucket existence: GET with
    // both range and partNumber is InvalidArgument even on a missing bucket.
    let get_invalid = GetObjectInput::builder()
        .bucket(bucket.clone())
        .key("k.txt".to_string())
        .range(Some(Range::Int { first: 0, last: Some(1) }))
        .part_number(Some(1))
        .build()
        .expect("get input should build");
    let err = Box::pin(usecase.execute_get_object(build_request(get_invalid, Method::GET)))
        .await
        .expect_err("range+partNumber must be rejected");
    assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidArgument);

    let lookups_before = VERSIONING_CONFIG_LOOKUPS.load(Ordering::SeqCst);

    let get = GetObjectInput::builder()
        .bucket(bucket.clone())
        .key("k.txt".to_string())
        .build()
        .expect("get input should build");
    let err = Box::pin(usecase.execute_get_object(build_request(get, Method::GET)))
        .await
        .expect_err("GET on a missing bucket must fail");
    assert_eq!(err.code(), &s3s::S3ErrorCode::NoSuchBucket);

    // Bucket existence beats versionId-format validation (which lives in the
    // opts builders) for HEAD and DeleteObject.
    let head = HeadObjectInput::builder()
        .bucket(bucket.clone())
        .key("k.txt".to_string())
        .version_id(Some("not-a-uuid".to_string()))
        .build()
        .expect("head input should build");
    let err = Box::pin(usecase.execute_head_object(build_request(head, Method::HEAD)))
        .await
        .expect_err("HEAD on a missing bucket must fail");
    assert_eq!(err.code(), &s3s::S3ErrorCode::NoSuchBucket);

    let del = DeleteObjectInput::builder()
        .bucket(bucket.clone())
        .key("k.txt".to_string())
        .version_id(Some("not-a-uuid".to_string()))
        .build()
        .expect("delete input should build");
    let err = Box::pin(usecase.execute_delete_object(build_request(del, Method::DELETE)))
        .await
        .expect_err("DeleteObject on a missing bucket must fail");
    assert_eq!(err.code(), &s3s::S3ErrorCode::NoSuchBucket);

    let lookups = VERSIONING_CONFIG_LOOKUPS.load(Ordering::SeqCst) - lookups_before;
    assert_eq!(
        lookups, 0,
        "requests naming a nonexistent bucket must fail before any bucket versioning lookup; got {lookups}"
    );
}
