use crate::bucket::metadata_sys::ObjectLockConfigState;
use crate::disk::endpoint::Endpoint;
use crate::disk::format::FormatV3;
use crate::disk::{DiskAPI, DiskOption, DiskStore, WalkDirOptions, new_disk};
use crate::error::Error;
use crate::io_support::rio::HashReader;
use crate::object_api::{BLOCK_SIZE_V2, ObjectLockConfigSnapshot, ObjectOptions, PutObjReader};
use crate::set_disk::SetDisks;
use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
use crate::storage_api_contracts::multipart::{CompletePart, MultipartOperations as _};
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
use crate::storage_api_contracts::range::HTTPRangeSpec;
use crate::store::init_format::save_format_file;
use http::HeaderMap;
use rustfs_filemeta::{MetacacheReader, MetacacheWriter};
use std::io::Cursor;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::fs;
use tokio::io::AsyncReadExt;
use tokio::sync::RwLock;

/// Returns the backing [`tempfile::TempDir`]s alongside the set so callers keep
/// them alive for the test's duration and the directories are removed on drop.
pub(crate) async fn make_local_set_disks(drive_count: usize, parity_count: usize) -> (Vec<tempfile::TempDir>, Arc<SetDisks>) {
    make_local_set_disks_with_ctx(drive_count, parity_count, crate::runtime::instance::bootstrap_ctx()).await
}

/// Like [`make_local_set_disks`], but binds the set to an explicit instance
/// context. Tests whose assertions depend on ambient runtime state that other
/// tests mutate (e.g. `SetupTypeGuard` flipping the shared setup type to
/// DistErasure from `#[serial]` tests in a different serial group) pass an
/// isolated context here so that state cannot leak into their code path.
pub(crate) async fn make_local_set_disks_with_ctx(
    drive_count: usize,
    parity_count: usize,
    ctx: Arc<crate::runtime::instance::InstanceContext>,
) -> (Vec<tempfile::TempDir>, Arc<SetDisks>) {
    let format = FormatV3::new(1, drive_count);
    let mut dirs = Vec::with_capacity(drive_count);
    let mut endpoints = Vec::with_capacity(drive_count);
    let mut disks = Vec::with_capacity(drive_count);

    for disk_idx in 0..drive_count {
        let dir = tempfile::tempdir().expect("tempdir should be created");
        let mut endpoint =
            Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");
        endpoint.set_pool_index(0);
        endpoint.set_set_index(0);
        endpoint.set_disk_index(disk_idx);

        let disk = new_disk(
            &endpoint,
            &DiskOption {
                cleanup: false,
                health_check: false,
            },
        )
        .await
        .expect("disk should be created");

        let mut disk_format = format.clone();
        disk_format.erasure.this = format.erasure.sets[0][disk_idx];
        save_format_file(&Some(disk.clone()), &Some(disk_format))
            .await
            .expect("format should be saved");

        dirs.push(dir);
        endpoints.push(endpoint);
        disks.push(Some(disk));
    }

    let set_disks = SetDisks::new_with_instance_ctx(
        "ecstore-validation-blackbox".to_string(),
        Arc::new(RwLock::new(disks)),
        drive_count,
        parity_count,
        0,
        0,
        endpoints,
        format,
        Vec::new(),
        ctx,
    )
    .await;

    (dirs, set_disks)
}

async fn first_shard_part_path(disk: &DiskStore, bucket: &str, object: &str) -> PathBuf {
    let object_dir = disk.path().join(bucket).join(object);
    let mut entries = fs::read_dir(&object_dir)
        .await
        .unwrap_or_else(|err| panic!("object data dir should be readable at {object_dir:?}: {err}"));
    while let Some(entry) = entries.next_entry().await.expect("object data dir entry should be readable") {
        if entry
            .file_type()
            .await
            .expect("object data entry type should be readable")
            .is_dir()
        {
            let part_path = entry.path().join("part.1");
            if fs::metadata(&part_path).await.is_ok() {
                return part_path;
            }
        }
    }
    panic!("part.1 shard should exist under {object_dir:?}");
}

async fn shard_part_paths(set_disks: &Arc<SetDisks>, bucket: &str, object: &str) -> Vec<PathBuf> {
    let disks = set_disks.disks.read().await;
    let mut paths = Vec::new();
    for disk in disks.iter().flatten() {
        paths.push(first_shard_part_path(disk, bucket, object).await);
    }
    paths
}

#[tokio::test]
async fn blackbox_put_unknown_actual_size_restores_body_and_records_written_size() {
    let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
    let bucket = "bb-unknown-actual-size";
    let object = "object.bin";
    let payload = (0..(BLOCK_SIZE_V2 + 123))
        .map(|idx| ((idx * 17) % 251) as u8)
        .collect::<Vec<_>>();
    let opts = ObjectOptions {
        no_lock: true,
        ..Default::default()
    };

    set_disks
        .make_bucket(bucket, &MakeBucketOptions::default())
        .await
        .expect("bucket should be created");
    let stream = HashReader::from_stream(Cursor::new(payload.clone()), payload.len() as i64, -1, None, None, false)
        .expect("hash reader should accept unknown actual size");
    let mut reader = PutObjReader::new(stream);
    let written = set_disks
        .put_object(bucket, object, &mut reader, &opts)
        .await
        .expect("object should be written");

    assert_eq!(written.size, payload.len() as i64);
    assert_eq!(written.parts[0].actual_size, payload.len() as i64);

    let mut get_reader = set_disks
        .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
        .await
        .expect("object reader should open");
    let mut restored = Vec::new();
    get_reader
        .stream
        .read_to_end(&mut restored)
        .await
        .expect("object should stream");
    assert_eq!(restored, payload);
}

#[tokio::test]
async fn blackbox_get_restores_body_after_one_shard_file_is_removed() {
    let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
    let bucket = "bb-missing-shard-file";
    let object = "object.bin";
    let payload = (0..(BLOCK_SIZE_V2 + 321))
        .map(|idx| ((idx * 19) % 251) as u8)
        .collect::<Vec<_>>();
    let opts = ObjectOptions {
        no_lock: true,
        ..Default::default()
    };

    set_disks
        .make_bucket(bucket, &MakeBucketOptions::default())
        .await
        .expect("bucket should be created");
    let mut reader = PutObjReader::from_vec(payload.clone());
    set_disks
        .put_object(bucket, object, &mut reader, &opts)
        .await
        .expect("object should be written");

    let disk = {
        let disks = set_disks.disks.read().await;
        disks[0].clone().expect("first disk should exist")
    };
    let shard_path = first_shard_part_path(&disk, bucket, object).await;
    fs::remove_file(&shard_path)
        .await
        .unwrap_or_else(|err| panic!("test shard should be removable at {shard_path:?}: {err}"));

    let mut get_reader = set_disks
        .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
        .await
        .expect("object should remain readable after one shard file is removed");
    let mut restored = Vec::new();
    get_reader
        .stream
        .read_to_end(&mut restored)
        .await
        .expect("degraded object should stream");

    assert_eq!(restored, payload);
}

#[tokio::test]
// Serialized: forces the reader-setup strategy through a process-global env var.
#[serial_test::serial]
async fn blackbox_heal_requests_preserve_repair_scope() {
    use rustfs_common::heal_channel::{
        HealAdmissionResult, HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest, HealRequestSource,
    };

    async fn receive_matching_heal(rx: &mut HealChannelReceiver, bucket: &str, object: &str) -> HealChannelRequest {
        tokio::time::timeout(std::time::Duration::from_secs(30), async {
            loop {
                match rx.recv().await.expect("heal channel should stay open") {
                    HealChannelCommand::Start { request, response_tx }
                        if request.bucket == bucket && request.object_prefix.as_deref() == Some(object) =>
                    {
                        let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
                        break request;
                    }
                    _ => continue,
                }
            }
        })
        .await
        .expect("matching heal request should be submitted")
    }

    // Own the process-global heal channel so the read path's repair submission
    // becomes observable. init_heal_channel() succeeds exactly once per test
    // binary: this must stay the only ecstore unit test that takes the receiver.
    // Submissions from other (non-serial) tests queue in the unbounded channel
    // while this test is setting up, get drained and dropped by the loop below
    // (failing their submitter, which releases their dedup reservation), and
    // fail fast once the receiver drops at test end. Tests that must observe a
    // deterministic channel state serialize under the same serial key.
    let mut heal_rx = rustfs_common::heal_channel::init_heal_channel()
        .expect("this must be the only ecstore test that owns the heal channel receiver");

    // Ordinary PUTs use the same admission channel as read repair. A single
    // rename target failure still satisfies write quorum, so the committed
    // version must be queued for convergence without delaying the PUT ACK.
    let (_put_dirs, put_set) = make_local_set_disks(4, 2).await;
    let put_bucket = "bb-put-partial-convergence";
    let put_object = "object.bin";
    put_set
        .make_bucket(put_bucket, &MakeBucketOptions::default())
        .await
        .expect("PUT bucket should be created");
    let offline_disk = {
        let mut disks = put_set.disks.write().await;
        disks[0].take()
    };
    let mut put_reader = PutObjReader::from_vec(vec![0x42; BLOCK_SIZE_V2 + 1024]);
    let committed = put_set
        .put_object(
            put_bucket,
            put_object,
            &mut put_reader,
            &ObjectOptions {
                no_lock: true,
                versioned: true,
                ..Default::default()
            },
        )
        .await
        .expect("partial ordinary PUT should succeed at write quorum");
    let committed_version = committed
        .version_id
        .expect("versioned PUT should return a version id")
        .to_string();

    let request = tokio::time::timeout(std::time::Duration::from_secs(30), async {
        loop {
            match heal_rx.recv().await.expect("heal channel should stay open") {
                HealChannelCommand::Start { request, response_tx }
                    if request.bucket == put_bucket && request.object_prefix.as_deref() == Some(put_object) =>
                {
                    let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
                    break request;
                }
                HealChannelCommand::Start { response_tx, .. } => {
                    let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
                }
                _ => {}
            }
        }
    })
    .await
    .expect("partial ordinary PUT should enqueue convergence heal");
    assert_eq!(request.object_version_id.as_deref(), Some(committed_version.as_str()));
    assert_eq!(request.pool_index, Some(0));
    assert_eq!(request.set_index, Some(0));

    let duplicate_request = tokio::time::timeout(std::time::Duration::from_millis(100), async {
        loop {
            match heal_rx.recv().await.expect("heal channel should stay open") {
                HealChannelCommand::Start { request, response_tx }
                    if request.bucket == put_bucket && request.object_prefix.as_deref() == Some(put_object) =>
                {
                    let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
                    break Some(request);
                }
                HealChannelCommand::Start { response_tx, .. } => {
                    let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
                }
                _ => {}
            }
        }
    })
    .await
    .ok()
    .flatten();
    assert!(duplicate_request.is_none(), "partial ordinary PUT must enqueue exactly one heal request");

    {
        let mut disks = put_set.disks.write().await;
        disks[0] = offline_disk;
    }

    let healthy_bucket = "bb-put-healthy-convergence";
    put_set
        .make_bucket(healthy_bucket, &MakeBucketOptions::default())
        .await
        .expect("healthy PUT bucket should be created");
    let mut healthy_reader = PutObjReader::from_vec(b"healthy".to_vec());
    put_set
        .put_object(
            healthy_bucket,
            put_object,
            &mut healthy_reader,
            &ObjectOptions {
                no_lock: true,
                ..Default::default()
            },
        )
        .await
        .expect("healthy ordinary PUT should succeed");
    let healthy_request = tokio::time::timeout(std::time::Duration::from_millis(100), async {
        loop {
            match heal_rx.recv().await.expect("heal channel should stay open") {
                HealChannelCommand::Start { request, response_tx }
                    if request.bucket == healthy_bucket && request.object_prefix.as_deref() == Some(put_object) =>
                {
                    let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
                    break Some(request);
                }
                HealChannelCommand::Start { response_tx, .. } => {
                    let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
                }
                _ => {}
            }
        }
    })
    .await
    .ok()
    .flatten();
    assert!(healthy_request.is_none(), "fully converged ordinary PUT must not enqueue heal");

    // Keep data-blocks-first reader setup explicit for this deterministic
    // repair assertion (see ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP in
    // set_disk/core/io_primitives.rs): if a caller opts back into all-shards,
    // the corrupt shard's failed open can lose the setup-quorum race, making
    // the enqueue assertion below a coin flip.
    temp_env::async_with_vars([("RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP", Some("true"))], async {
        let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
        let bucket = "bb-corrupt-shard-repair";
        let object = "object.bin";
        let payload = (0..(BLOCK_SIZE_V2 + 777))
            .map(|idx| ((idx * 29) % 251) as u8)
            .collect::<Vec<_>>();
        let opts = ObjectOptions {
            no_lock: true,
            ..Default::default()
        };

        set_disks
            .make_bucket(bucket, &MakeBucketOptions::default())
            .await
            .expect("bucket should be created");
        let mut reader = PutObjReader::from_vec(payload.clone());
        set_disks
            .put_object(bucket, object, &mut reader, &opts)
            .await
            .expect("object should be written");

        // Corrupt the disk holding DATA shard 1, not blindly disk 0 (which holds
        // a parity shard for this key): a data-blocks-first read must consume the
        // corrupt data shard, making the missing-shard detection deterministic.
        // The write path derives the shard distribution the same way (ops/object.rs).
        let distribution = rustfs_filemeta::FileInfo::new(&format!("{bucket}/{object}"), 2, 2)
            .erasure
            .distribution;
        let corrupt_disk = distribution
            .iter()
            .position(|&shard| shard == 1)
            .expect("distribution should place data shard 1 on a disk");
        let paths = shard_part_paths(&set_disks, bucket, object).await;
        fs::write(&paths[corrupt_disk], b"corrupt shard bytes")
            .await
            .unwrap_or_else(|err| panic!("test shard should be corruptible at {:?}: {err}", paths[corrupt_disk]));

        let mut get_reader = set_disks
            .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
            .await
            .expect("object should remain readable after one corrupt shard");
        let mut restored = Vec::new();
        get_reader
            .stream
            .read_to_end(&mut restored)
            .await
            .expect("corrupt-shard object should stream from parity");

        assert_eq!(restored, payload);

        let request = receive_matching_heal(&mut heal_rx, bucket, object).await;

        assert_eq!(request.source, HealRequestSource::ReadRepair);
        assert_eq!(request.object_prefix.as_deref(), Some(object));
        assert_eq!(request.object_version_id, None, "unversioned PUT must submit repair without a version id");
        assert_eq!(request.pool_index, Some(0));
        assert_eq!(request.set_index, Some(0));
        assert_eq!(request.priority, HealChannelPriority::Low);
        assert_eq!(request.recreate_missing, Some(true));

        let mpu_bucket = "bb-mpu-convergence-heal";
        let partial_object = "partial.bin";
        let suspended_object = "suspended.bin";
        let payload = vec![0x5a; 1 << 20];
        let mpu_opts = ObjectOptions {
            no_lock: true,
            versioned: true,
            ..Default::default()
        };

        set_disks
            .make_bucket(mpu_bucket, &MakeBucketOptions::default())
            .await
            .expect("multipart bucket should be created");

        let stage_upload = async |object: &str| {
            let upload = set_disks
                .new_multipart_upload(mpu_bucket, object, &mpu_opts)
                .await
                .expect("multipart upload should be created");
            let mut reader = PutObjReader::new(
                HashReader::from_stream(
                    Cursor::new(payload.clone()),
                    payload.len() as i64,
                    payload.len() as i64,
                    None,
                    None,
                    false,
                )
                .expect("multipart reader should be constructed"),
            );
            let part = set_disks
                .put_object_part(mpu_bucket, object, &upload.upload_id, 1, &mut reader, &mpu_opts)
                .await
                .expect("multipart part should be written");
            (
                upload.upload_id,
                vec![CompletePart {
                    part_num: part.part_num,
                    etag: part.etag,
                    ..Default::default()
                }],
            )
        };

        let (partial_upload_id, partial_parts) = stage_upload(partial_object).await;
        let offline_disk = {
            let mut disks = set_disks.disks.write().await;
            disks[3].take().expect("fourth disk should be online before completion")
        };
        crate::crash_inject::arm(crate::crash_inject::CrashPoint::MultipartAfterCommitBeforePartsCleanup, partial_object);
        let completed = set_disks
            .clone()
            .complete_multipart_upload(mpu_bucket, partial_object, &partial_upload_id, partial_parts, &mpu_opts)
            .await;
        assert!(
            matches!(completed, Err(Error::Unexpected)),
            "partial multipart completion should reach the post-commit crash point, got {completed:?}"
        );
        crate::crash_inject::disarm(crate::crash_inject::CrashPoint::MultipartAfterCommitBeforePartsCleanup, partial_object);

        let request = receive_matching_heal(&mut heal_rx, mpu_bucket, partial_object).await;

        let completed_version_id = request
            .object_version_id
            .clone()
            .expect("versioned multipart convergence heal must bind a version id");
        assert_eq!(request.pool_index, Some(0));
        assert_eq!(request.set_index, Some(0));
        assert_eq!(request.priority, HealChannelPriority::Normal);

        let duplicate = tokio::time::timeout(std::time::Duration::from_millis(250), async {
            loop {
                match heal_rx.recv().await.expect("heal channel should stay open") {
                    HealChannelCommand::Start { request, .. }
                        if request.bucket == mpu_bucket && request.object_prefix.as_deref() == Some(partial_object) =>
                    {
                        break request;
                    }
                    _ => continue,
                }
            }
        })
        .await;
        assert!(duplicate.is_err(), "partial multipart completion must enqueue exactly one heal request");

        {
            let mut disks = set_disks.disks.write().await;
            disks[3] = Some(offline_disk);
        }
        let committed = set_disks
            .get_object_info(
                mpu_bucket,
                partial_object,
                &ObjectOptions {
                    no_lock: true,
                    versioned: true,
                    version_id: Some(completed_version_id.clone()),
                    ..Default::default()
                },
            )
            .await
            .expect("heal-bound multipart version should be committed and addressable");
        assert_eq!(committed.version_id.map(|version_id| version_id.to_string()), Some(completed_version_id));

        let (suspended_upload_id, suspended_parts) = stage_upload(suspended_object).await;
        let offline_disk = {
            let mut disks = set_disks.disks.write().await;
            disks[3]
                .take()
                .expect("fourth disk should be online before suspended completion")
        };
        let suspended_opts = ObjectOptions {
            no_lock: true,
            version_suspended: true,
            ..Default::default()
        };
        let suspended = set_disks
            .clone()
            .complete_multipart_upload(mpu_bucket, suspended_object, &suspended_upload_id, suspended_parts, &suspended_opts)
            .await
            .expect("suspended multipart completion should succeed at write quorum");
        {
            let mut disks = set_disks.disks.write().await;
            disks[3] = Some(offline_disk);
        }
        assert!(
            suspended.version_id.is_some_and(|version_id| version_id.is_nil()),
            "suspended multipart completion should publish the null version"
        );

        let request = receive_matching_heal(&mut heal_rx, mpu_bucket, suspended_object).await;

        let null_version_id = uuid::Uuid::nil().to_string();
        assert_eq!(request.object_version_id.as_deref(), Some(null_version_id.as_str()));
    })
    .await;
}

#[tokio::test]
async fn blackbox_range_read_restores_exact_slice_with_one_offline_disk() {
    let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
    let bucket = "bb-range-offline-disk";
    let object = "object.bin";
    let payload = (0..(BLOCK_SIZE_V2 + 4096))
        .map(|idx| ((idx * 23) % 251) as u8)
        .collect::<Vec<_>>();
    let range_start = 513usize;
    let range_len = 8192usize;
    let opts = ObjectOptions {
        no_lock: true,
        ..Default::default()
    };

    set_disks
        .make_bucket(bucket, &MakeBucketOptions::default())
        .await
        .expect("bucket should be created");
    let mut reader = PutObjReader::from_vec(payload.clone());
    set_disks
        .put_object(bucket, object, &mut reader, &opts)
        .await
        .expect("object should be written");

    {
        let mut disks = set_disks.disks.write().await;
        disks[2] = None;
    }

    let range = HTTPRangeSpec {
        is_suffix_length: false,
        start: range_start as i64,
        end: (range_start + range_len - 1) as i64,
    };
    let mut get_reader = set_disks
        .get_object_reader(bucket, object, Some(range), HeaderMap::new(), &opts)
        .await
        .expect("range reader should open with one offline disk");
    let mut restored = Vec::new();
    get_reader
        .stream
        .read_to_end(&mut restored)
        .await
        .expect("range body should stream");

    assert_eq!(restored, payload[range_start..range_start + range_len]);
}

#[tokio::test]
async fn blackbox_delete_marker_hides_object_body_without_erasing_prior_version_metadata() {
    let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
    let bucket = "bb-delete-marker-read-negative";
    let object = "object.bin";
    let opts = ObjectOptions {
        no_lock: true,
        version_suspended: true,
        object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::ConfirmedAbsent))),
        ..Default::default()
    };

    set_disks
        .make_bucket(bucket, &MakeBucketOptions::default())
        .await
        .expect("bucket should be created");
    let mut reader = PutObjReader::from_vec(b"body hidden by delete marker".to_vec());
    set_disks
        .put_object(bucket, object, &mut reader, &opts)
        .await
        .expect("object should be written before delete marker");

    let marker = set_disks
        .delete_object(bucket, object, opts.clone())
        .await
        .expect("version suspended delete should create a marker");
    assert!(marker.delete_marker);

    let err = match set_disks
        .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
        .await
    {
        Ok(_) => panic!("delete marker must hide the object body"),
        Err(err) => err,
    };
    assert!(
        matches!(&err, Error::ObjectNotFound(b, o) if b == bucket && o == object),
        "latest delete marker read must map to ObjectNotFound for the requested key, got {err:?}"
    );
}

#[tokio::test]
async fn blackbox_local_disk_walk_dir_emits_metadata_entries_with_prefix_forward_and_limit() {
    let dir = tempfile::tempdir().expect("tempdir should be created");
    let bucket = "bb-local-walk";
    let endpoint = Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");
    let disk = new_disk(
        &endpoint,
        &DiskOption {
            cleanup: false,
            health_check: false,
        },
    )
    .await
    .expect("disk should be created");
    disk.make_volume(bucket).await.expect("bucket volume should be created");

    for object in ["prefix/a", "prefix/b", "prefix/c", "other/d"] {
        disk.write_all(bucket, &format!("{object}/xl.meta"), bytes::Bytes::from_static(b"meta"))
            .await
            .expect("metadata object should be written");
    }

    let (reader, mut writer) = tokio::io::duplex(4096);
    let opts = WalkDirOptions {
        bucket: bucket.to_string(),
        base_dir: "prefix/".to_string(),
        recursive: true,
        forward_to: Some("prefix/b".to_string()),
        limit: 2,
        ..Default::default()
    };

    disk.walk_dir(opts, &mut writer)
        .await
        .expect("walk_dir should stream metadata entries");
    MetacacheWriter::new(&mut writer)
        .close()
        .await
        .expect("metacache stream should close");
    drop(writer);

    let mut reader = MetacacheReader::new(reader);
    let entries = reader.read_all().await.expect("metacache stream should decode");
    let names = entries
        .into_iter()
        .filter(|entry| !entry.metadata.is_empty())
        .map(|entry| entry.name)
        .collect::<Vec<_>>();

    assert_eq!(names, vec!["prefix/b".to_string(), "prefix/c".to_string()]);
}

#[tokio::test]
#[serial_test::serial]
async fn blackbox_issue3031_diag_covers_put_success_cleanup_and_error_summary() {
    temp_env::async_with_vars([("RUSTFS_ISSUE3031_DIAG_ENABLE", Some("true"))], async {
        let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
        let bucket = "bb-issue3031-diag";
        let object = "object.bin";
        let first_payload = b"first diagnostic body".to_vec();
        let second_payload = b"second diagnostic body that replaces the first".to_vec();
        let opts = ObjectOptions {
            no_lock: true,
            ..Default::default()
        };

        set_disks
            .make_bucket(bucket, &MakeBucketOptions::default())
            .await
            .expect("bucket should be created");

        let mut first_reader = PutObjReader::from_vec(first_payload);
        set_disks
            .put_object(bucket, object, &mut first_reader, &opts)
            .await
            .expect("first diagnostic write should succeed");

        let mut second_reader = PutObjReader::from_vec(second_payload.clone());
        set_disks
            .put_object(bucket, object, &mut second_reader, &opts)
            .await
            .expect("overwrite should succeed and report cleanup-present diagnostics");

        let mut get_reader = set_disks
            .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
            .await
            .expect("overwritten object reader should open");
        let mut restored = Vec::new();
        get_reader
            .stream
            .read_to_end(&mut restored)
            .await
            .expect("overwritten object should stream");
        assert_eq!(restored, second_payload);

        let (_failed_dirs, failed_set) = make_local_set_disks(1, 0).await;
        let failed_bucket = "bb-issue3031-fail";
        failed_set
            .make_bucket(failed_bucket, &MakeBucketOptions::default())
            .await
            .expect("failure bucket should be created");
        {
            let mut disks = failed_set.disks.write().await;
            disks[0] = None;
        }
        let mut failed_reader = PutObjReader::from_vec(b"must fail closed".to_vec());
        let err = failed_set
            .put_object(failed_bucket, "object.bin", &mut failed_reader, &opts)
            .await
            .expect_err("offline only disk must fail writer setup under diagnostics");
        assert!(
            matches!(err, Error::InsufficientWriteQuorum(_, _) | Error::ErasureWriteQuorum),
            "diagnostic failure should still report a write-quorum style error, got {err:?}"
        );
    })
    .await;
}

/// rustfs/backlog#1009: the `put_object` old-size backfill must reproduce, for
/// every PUT shape, exactly what the app layer's pre-PUT `get_object_info`
/// lookup would have observed — that is the invariant that lets the app skip
/// the lookup without changing a single usage-accounting number.
mod old_current_size_backfill {
    use super::*;
    use crate::disk::OldCurrentSize;
    use crate::error::is_err_object_not_found;
    use crate::set_disk::SetDisks;

    /// What the pre-PUT lookup would report right now for `object`.
    async fn prelookup_expectation(set_disks: &Arc<SetDisks>, bucket: &str, object: &str) -> OldCurrentSize {
        let opts = ObjectOptions {
            no_lock: true,
            ..Default::default()
        };
        match set_disks.get_object_info(bucket, object, &opts).await {
            Ok(object_info) => OldCurrentSize::Present(object_info.size),
            Err(err) => {
                assert!(is_err_object_not_found(&err), "prelookup expectation hit unexpected error: {err:?}");
                OldCurrentSize::Absent
            }
        }
    }

    async fn put_and_backfill(
        set_disks: &Arc<SetDisks>,
        bucket: &str,
        object: &str,
        len: usize,
        opts: &ObjectOptions,
    ) -> Option<OldCurrentSize> {
        let payload = (0..len).map(|idx| ((idx * 31) % 251) as u8).collect::<Vec<_>>();
        let mut reader = PutObjReader::from_vec(payload);
        let (_object_info, backfill) = set_disks
            .put_object_with_old_current_size(bucket, object, &mut reader, opts)
            .await
            .expect("put_object should succeed");
        backfill
    }

    #[tokio::test]
    async fn backfill_matches_prelookup_for_every_put_shape() {
        let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
        let bucket = "bb-old-current-size";
        set_disks
            .make_bucket(bucket, &MakeBucketOptions::default())
            .await
            .expect("bucket should be created");
        let opts = ObjectOptions {
            no_lock: true,
            ..Default::default()
        };

        // Fresh key (inline-sized): both the lookup and the backfill say Absent.
        let object = "object.bin";
        assert_eq!(prelookup_expectation(&set_disks, bucket, object).await, OldCurrentSize::Absent);
        assert_eq!(
            put_and_backfill(&set_disks, bucket, object, 1024, &opts).await,
            Some(OldCurrentSize::Absent)
        );

        // Unversioned inline overwrite: the previous live size must surface.
        let expected = prelookup_expectation(&set_disks, bucket, object).await;
        assert_eq!(expected, OldCurrentSize::Present(1024));
        assert_eq!(put_and_backfill(&set_disks, bucket, object, 2048, &opts).await, Some(expected));

        // Non-inline overwrite (payload above the inline block threshold).
        let expected = prelookup_expectation(&set_disks, bucket, object).await;
        assert_eq!(expected, OldCurrentSize::Present(2048));
        assert_eq!(
            put_and_backfill(&set_disks, bucket, object, BLOCK_SIZE_V2 + 123, &opts).await,
            Some(expected)
        );

        // Overwriting a non-inline object reports its size back.
        let expected = prelookup_expectation(&set_disks, bucket, object).await;
        assert_eq!(expected, OldCurrentSize::Present((BLOCK_SIZE_V2 + 123) as i64));
        assert_eq!(put_and_backfill(&set_disks, bucket, object, 64, &opts).await, Some(expected));

        // Versioned writes: a new version over a live latest reports the
        // latest's size, not the incoming version's.
        let versioned = "versioned.bin";
        let first_version_opts = ObjectOptions {
            no_lock: true,
            versioned: true,
            version_id: Some(uuid::Uuid::new_v4().to_string()),
            ..Default::default()
        };
        assert_eq!(
            put_and_backfill(&set_disks, bucket, versioned, 111, &first_version_opts).await,
            Some(OldCurrentSize::Absent)
        );
        let expected = prelookup_expectation(&set_disks, bucket, versioned).await;
        assert_eq!(expected, OldCurrentSize::Present(111));
        let second_version_opts = ObjectOptions {
            no_lock: true,
            versioned: true,
            version_id: Some(uuid::Uuid::new_v4().to_string()),
            ..Default::default()
        };
        assert_eq!(
            put_and_backfill(&set_disks, bucket, versioned, 222, &second_version_opts).await,
            Some(expected)
        );

        // Delete-marker latest: today's lookup returns the marker as
        // Ok(size 0) — NOT not-found — and delete-marker creation never
        // decremented objects_count, so the backfill must reproduce
        // Present(0) to keep versioned accounting identical.
        set_disks
            .delete_object(
                bucket,
                versioned,
                ObjectOptions {
                    no_lock: true,
                    versioned: true,
                    object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(
                        ObjectLockConfigState::ConfirmedAbsent,
                    ))),
                    ..Default::default()
                },
            )
            .await
            .expect("versioned delete should create a delete marker");
        let expected = prelookup_expectation(&set_disks, bucket, versioned).await;
        assert_eq!(expected, OldCurrentSize::Present(0));
        let third_version_opts = ObjectOptions {
            no_lock: true,
            versioned: true,
            version_id: Some(uuid::Uuid::new_v4().to_string()),
            ..Default::default()
        };
        assert_eq!(
            put_and_backfill(&set_disks, bucket, versioned, 333, &third_version_opts).await,
            Some(expected)
        );
    }

    /// Sub-quorum divergence through the real set-level fanout: when only 2 of
    /// 4 disks can still decode the destination's xl.meta (write quorum is 3),
    /// the PUT must still commit but the backfill must come back unknown —
    /// never a fabricated `Absent`/`Present` from a below-quorum vote.
    #[tokio::test]
    async fn backfill_is_unknown_below_quorum_but_put_succeeds() {
        let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
        let bucket = "bb-old-current-subquorum";
        set_disks
            .make_bucket(bucket, &MakeBucketOptions::default())
            .await
            .expect("bucket should be created");
        let opts = ObjectOptions {
            no_lock: true,
            ..Default::default()
        };

        let object = "object.bin";
        assert_eq!(
            put_and_backfill(&set_disks, bucket, object, 1024, &opts).await,
            Some(OldCurrentSize::Absent)
        );

        // Corrupt the committed xl.meta on two disks: those observations turn
        // unknown, leaving 2 definite votes < write quorum 3.
        {
            let disks = set_disks.disks.read().await;
            for disk in disks.iter().flatten().take(2) {
                let meta_path = disk.path().join(bucket).join(object).join("xl.meta");
                fs::write(&meta_path, b"not-an-xl-meta")
                    .await
                    .expect("xl.meta should be overwritable with garbage");
            }
        }

        let backfill = put_and_backfill(&set_disks, bucket, object, 2048, &opts).await;
        assert_eq!(backfill, None, "a below-quorum vote must surface as unknown");
    }
}