// 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 crate::disk::disk_store::get_drive_walkdir_stall_timeout;
use crate::disk::error::DiskError;
use crate::disk::{self, DiskAPI, DiskStore, WalkDirOptions};
use futures::future::join_all;
use metrics::counter;
use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetacacheReader, is_io_eof};
use std::{
    collections::VecDeque,
    future::Future,
    pin::Pin,
    sync::{Arc, OnceLock},
    time::Duration,
};
use tokio::io::AsyncRead;
use tokio::spawn;
use tokio::time::timeout;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, warn};

const LOG_COMPONENT_ECSTORE: &str = "ecstore";
const LOG_SUBSYSTEM_METACACHE: &str = "metacache";
const EVENT_METACACHE_LISTING: &str = "metacache_listing";

pub type AgreedFn = Box<dyn Fn(MetaCacheEntry) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
pub type PartialFn =
    Box<dyn Fn(MetaCacheEntries, &[Option<DiskError>]) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
type FinishedFn = Box<dyn Fn(&[Option<DiskError>]) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;

#[derive(Debug)]
enum PeekOutcome {
    Ready(Option<MetaCacheEntry>),
    Error(rustfs_filemeta::Error),
    TimedOut,
}

async fn peek_with_timeout<R: AsyncRead + Unpin>(reader: &mut MetacacheReader<R>, timeout_duration: Duration) -> PeekOutcome {
    match timeout(timeout_duration, reader.peek()).await {
        Ok(Ok(entry)) => PeekOutcome::Ready(entry),
        Ok(Err(err)) => PeekOutcome::Error(err),
        Err(_) => PeekOutcome::TimedOut,
    }
}

fn is_missing_path_error(err: &DiskError) -> bool {
    matches!(err, DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound)
}

#[cfg(test)]
#[derive(Clone)]
pub(crate) enum TestReaderBehavior {
    Eof,
    Stall,
    IgnoreCancel,
    ProducerError(DiskError),
    PartialThenTimeout(Vec<MetaCacheEntry>),
}

#[derive(Default)]
pub struct ListPathRawOptions {
    pub disks: Vec<Option<DiskStore>>,
    pub fallback_disks: Vec<Option<DiskStore>>,
    pub bucket: String,
    pub path: String,
    pub recursive: bool,
    pub filter_prefix: Option<String>,
    pub forward_to: Option<String>,
    pub min_disks: usize,
    pub report_not_found: bool,
    pub per_disk_limit: i32,
    pub skip_walkdir_total_timeout: bool,
    pub agreed: Option<AgreedFn>,
    pub partial: Option<PartialFn>,
    pub finished: Option<FinishedFn>,
    #[cfg(test)]
    pub(crate) test_reader_behaviors: Vec<TestReaderBehavior>,
    #[cfg(test)]
    pub(crate) peek_timeout: Option<Duration>,
    // pub agreed: Option<Arc<dyn Fn(MetaCacheEntry) + Send + Sync>>,
    // pub partial: Option<Arc<dyn Fn(MetaCacheEntries, &[Option<Error>]) + Send + Sync>>,
    // pub finished: Option<Arc<dyn Fn(&[Option<Error>]) + Send + Sync>>,
}

impl Clone for ListPathRawOptions {
    fn clone(&self) -> Self {
        Self {
            disks: self.disks.clone(),
            fallback_disks: self.fallback_disks.clone(),
            bucket: self.bucket.clone(),
            path: self.path.clone(),
            recursive: self.recursive,
            filter_prefix: self.filter_prefix.clone(),
            forward_to: self.forward_to.clone(),
            min_disks: self.min_disks,
            report_not_found: self.report_not_found,
            per_disk_limit: self.per_disk_limit,
            skip_walkdir_total_timeout: self.skip_walkdir_total_timeout,
            #[cfg(test)]
            test_reader_behaviors: self.test_reader_behaviors.clone(),
            #[cfg(test)]
            peek_timeout: self.peek_timeout,
            ..Default::default()
        }
    }
}

pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> disk::error::Result<()> {
    if opts.disks.is_empty() {
        return Err(DiskError::ErasureReadQuorum);
    }

    let log_bucket = opts.bucket.clone();
    let log_path = opts.path.clone();

    let mut jobs: Vec<tokio::task::JoinHandle<std::result::Result<(), DiskError>>> = Vec::new();
    let mut readers = Vec::with_capacity(opts.disks.len());
    let fds = opts.fallback_disks.iter().flatten().cloned().collect::<VecDeque<_>>();
    let max_disk_failures = opts.disks.len().saturating_sub(opts.min_disks);
    let producer_errs: Arc<[OnceLock<DiskError>]> = (0..opts.disks.len()).map(|_| OnceLock::new()).collect::<Vec<_>>().into();

    let cancel_rx = CancellationToken::new();

    for (disk_idx, disk) in opts.disks.iter().enumerate() {
        let opdisk = disk.clone();
        let opts_clone = opts.clone();
        let mut fds_clone = fds.clone();
        let cancel_rx_clone = cancel_rx.clone();
        let producer_errs_clone = producer_errs.clone();
        let (rd, wr) = tokio::io::duplex(64);
        readers.push(MetacacheReader::new(rd));
        jobs.push(spawn(async move {
            #[cfg(test)]
            if let Some(behavior) = opts_clone.test_reader_behaviors.get(disk_idx).cloned() {
                match behavior {
                    TestReaderBehavior::Eof => return Ok(()),
                    TestReaderBehavior::Stall => {
                        let _held_writer = wr;
                        cancel_rx_clone.cancelled().await;
                        return Ok(());
                    }
                    TestReaderBehavior::IgnoreCancel => {
                        let _held_writer = wr;
                        std::future::pending::<()>().await;
                        return Ok(());
                    }
                    TestReaderBehavior::ProducerError(err) => {
                        record_producer_error(&producer_errs_clone, disk_idx, &err);
                        return Err(err);
                    }
                    TestReaderBehavior::PartialThenTimeout(entries) => {
                        let mut wr = wr;
                        let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr);
                        let err = DiskError::Timeout;
                        record_producer_error(&producer_errs_clone, disk_idx, &err);
                        let _ = out.write(&entries).await;
                        drop(out);
                        return Err(err);
                    }
                }
            }

            let mut wr = wr;
            let wakl_opts = WalkDirOptions {
                bucket: opts_clone.bucket.clone(),
                base_dir: opts_clone.path.clone(),
                recursive: opts_clone.recursive,
                report_notfound: opts_clone.report_not_found,
                filter_prefix: opts_clone.filter_prefix.clone(),
                forward_to: opts_clone.forward_to.clone(),
                limit: opts_clone.per_disk_limit,
                skip_total_timeout: opts_clone.skip_walkdir_total_timeout,
                ..Default::default()
            };

            let mut need_fallback = false;
            let mut last_err = None;
            if let Some(disk) = opdisk {
                let primary_walk_started = std::time::Instant::now();
                match disk.walk_dir(wakl_opts, &mut wr).await {
                    Ok(_res) => {
                        rustfs_io_metrics::record_stage_duration(
                            "metacache_walk_dir_primary",
                            primary_walk_started.elapsed().as_secs_f64() * 1000.0,
                        );
                    }
                    Err(err) => {
                        rustfs_io_metrics::record_stage_duration(
                            "metacache_walk_dir_primary_failed",
                            primary_walk_started.elapsed().as_secs_f64() * 1000.0,
                        );
                        if is_missing_path_error(&err) {
                            debug!(
                                event = EVENT_METACACHE_LISTING,
                                component = LOG_COMPONENT_ECSTORE,
                                subsystem = LOG_SUBSYSTEM_METACACHE,
                                bucket = %opts_clone.bucket,
                                path = %opts_clone.path,
                                disk_index = disk_idx,
                                state = "walk_dir_missing_path",
                                error = ?err,
                                "Metacache walk_dir missing path skipped"
                            );
                        } else {
                            warn!(
                                event = EVENT_METACACHE_LISTING,
                                component = LOG_COMPONENT_ECSTORE,
                                subsystem = LOG_SUBSYSTEM_METACACHE,
                                bucket = %opts_clone.bucket,
                                path = %opts_clone.path,
                                disk_index = disk_idx,
                                state = "walk_dir_failed",
                                error = ?err,
                                "Metacache walk_dir failed"
                            );
                        }
                        last_err = Some(err);
                        need_fallback = true;
                    }
                }
            } else {
                last_err = Some(DiskError::DiskNotFound);
                need_fallback = true;
            }

            if cancel_rx_clone.is_cancelled() {
                // warn!("list_path_raw: cancel_rx_clone.is_cancelled()");
                return Ok(());
            }

            while need_fallback {
                let mut disk_op = None;
                while let Some(disk) = fds_clone.pop_front() {
                    if disk.is_online().await {
                        disk_op = Some(disk);
                        break;
                    }
                }

                let Some(disk) = disk_op else {
                    let err = last_err.unwrap_or(DiskError::DiskNotFound);
                    if is_missing_path_error(&err) {
                        debug!(
                            event = EVENT_METACACHE_LISTING,
                            component = LOG_COMPONENT_ECSTORE,
                            subsystem = LOG_SUBSYSTEM_METACACHE,
                            bucket = %opts_clone.bucket,
                            path = %opts_clone.path,
                            disk_index = disk_idx,
                            state = "fallback_disk_missing_for_path",
                            error = ?err,
                            "Metacache fallback disk unavailable for missing path"
                        );
                    } else {
                        warn!(
                            event = EVENT_METACACHE_LISTING,
                            component = LOG_COMPONENT_ECSTORE,
                            subsystem = LOG_SUBSYSTEM_METACACHE,
                            bucket = %opts_clone.bucket,
                            path = %opts_clone.path,
                            disk_index = disk_idx,
                            state = "fallback_disk_missing",
                            error = ?err,
                            "Metacache fallback disk missing"
                        );
                    }
                    record_producer_error(&producer_errs_clone, disk_idx, &err);
                    return Err(err);
                };

                let fallback_walk_started = std::time::Instant::now();
                match disk
                    .as_ref()
                    .walk_dir(
                        WalkDirOptions {
                            bucket: opts_clone.bucket.clone(),
                            base_dir: opts_clone.path.clone(),
                            recursive: opts_clone.recursive,
                            report_notfound: opts_clone.report_not_found,
                            filter_prefix: opts_clone.filter_prefix.clone(),
                            forward_to: opts_clone.forward_to.clone(),
                            limit: opts_clone.per_disk_limit,
                            skip_total_timeout: opts_clone.skip_walkdir_total_timeout,
                            ..Default::default()
                        },
                        &mut wr,
                    )
                    .await
                {
                    Ok(_r) => {
                        rustfs_io_metrics::record_stage_duration(
                            "metacache_walk_dir_fallback",
                            fallback_walk_started.elapsed().as_secs_f64() * 1000.0,
                        );
                        need_fallback = false;
                        last_err = None;
                    }
                    Err(err) => {
                        rustfs_io_metrics::record_stage_duration(
                            "metacache_walk_dir_fallback_failed",
                            fallback_walk_started.elapsed().as_secs_f64() * 1000.0,
                        );
                        if is_missing_path_error(&err) {
                            debug!(
                                event = EVENT_METACACHE_LISTING,
                                component = LOG_COMPONENT_ECSTORE,
                                subsystem = LOG_SUBSYSTEM_METACACHE,
                                bucket = %opts_clone.bucket,
                                path = %opts_clone.path,
                                disk_index = disk_idx,
                                state = "fallback_walk_dir_missing_path",
                                error = ?err,
                                "Metacache fallback walk_dir missing path skipped"
                            );
                        } else {
                            error!(
                                event = EVENT_METACACHE_LISTING,
                                component = LOG_COMPONENT_ECSTORE,
                                subsystem = LOG_SUBSYSTEM_METACACHE,
                                bucket = %opts_clone.bucket,
                                path = %opts_clone.path,
                                disk_index = disk_idx,
                                state = "fallback_walk_dir_failed",
                                error = ?err,
                                "Metacache fallback walk_dir failed"
                            );
                        }
                        last_err = Some(err);
                    }
                }
            }

            if need_fallback {
                return Err(last_err.unwrap_or(DiskError::DiskNotFound));
            }

            // warn!("list_path_raw: while need_fallback done");
            Ok(())
        }));
    }

    let revjob = spawn(async move {
        #[cfg(test)]
        let peek_timeout = opts.peek_timeout.unwrap_or_else(get_drive_walkdir_stall_timeout);
        #[cfg(not(test))]
        let peek_timeout = get_drive_walkdir_stall_timeout();
        let mut errs: Vec<Option<DiskError>> = Vec::with_capacity(readers.len());
        for _ in 0..readers.len() {
            errs.push(None);
        }

        loop {
            let mut current = MetaCacheEntry::default();

            // warn!(
            //     "list_path_raw: loop start, bucket: {}, path: {}, current: {:?}",
            //     opts.bucket, opts.path, &current.name
            // );

            if rx.is_cancelled() {
                return Err(DiskError::other("canceled"));
            }

            let mut top_entries: Vec<Option<MetaCacheEntry>> = vec![None; readers.len()];

            let mut at_eof = 0;
            let mut fnf = 0;
            let mut vnf = 0;
            let mut has_err = 0;
            let mut agree = 0;

            for (i, r) in readers.iter_mut().enumerate() {
                if errs[i].is_some() {
                    has_err += 1;
                    continue;
                }

                let entry = match peek_with_timeout(r, peek_timeout).await {
                    PeekOutcome::Ready(res) => {
                        if let Some(entry) = res {
                            // info!("read entry disk: {}, name: {}", i, entry.name);
                            entry
                        } else {
                            if let Some(err) = producer_error(&producer_errs, i) {
                                has_err += 1;
                                errs[i] = Some(err);
                                continue;
                            }
                            // eof
                            at_eof += 1;
                            // warn!("list_path_raw: peek eof, disk: {}", i);
                            continue;
                        }
                    }
                    PeekOutcome::Error(err) => {
                        if let Some(err) = producer_error(&producer_errs, i) {
                            has_err += 1;
                            errs[i] = Some(err);
                            continue;
                        }

                        if err == rustfs_filemeta::Error::Unexpected {
                            at_eof += 1;
                            // warn!("list_path_raw: peek err eof, disk: {}", i);
                            continue;
                        }

                        // warn!("list_path_raw: peek err00, err: {:?}", err);

                        if is_io_eof(&err) {
                            at_eof += 1;
                            // warn!("list_path_raw: peek eof, disk: {}", i);
                            continue;
                        }

                        if err == rustfs_filemeta::Error::FileNotFound {
                            at_eof += 1;
                            fnf += 1;
                            // warn!("list_path_raw: peek fnf, disk: {}", i);
                            continue;
                        } else if err == rustfs_filemeta::Error::VolumeNotFound {
                            at_eof += 1;
                            fnf += 1;
                            vnf += 1;
                            // warn!("list_path_raw: peek vnf, disk: {}", i);
                            continue;
                        } else {
                            has_err += 1;
                            errs[i] = Some(err.into());
                            // warn!("list_path_raw: peek err, disk: {}", i);
                            continue;
                        }
                    }
                    PeekOutcome::TimedOut => {
                        has_err += 1;
                        errs[i] = Some(DiskError::Timeout);
                        let endpoint = opts
                            .disks
                            .get(i)
                            .and_then(|disk| disk.as_ref().map(|disk| disk.endpoint().to_string()))
                            .unwrap_or_else(|| "missing".to_string());
                        counter!(
                            "rustfs_list_path_raw_stall_total",
                            "drive" => endpoint.clone()
                        )
                        .increment(1);
                        warn!(
                            event = EVENT_METACACHE_LISTING,
                            component = LOG_COMPONENT_ECSTORE,
                            subsystem = LOG_SUBSYSTEM_METACACHE,
                            drive = %endpoint,
                            bucket = %opts.bucket,
                            path = %opts.path,
                            timeout_ms = peek_timeout.as_millis(),
                            state = "peek_timed_out",
                            "Metacache reader peek timed out"
                        );
                        let (detached_rd, write_half) = tokio::io::duplex(1);
                        drop(write_half);
                        *r = MetacacheReader::new(detached_rd);
                        continue;
                    }
                };

                // warn!("list_path_raw: loop entry: {:?}, disk: {}", &entry.name, i);

                // If no current, add it.
                if current.name.is_empty() {
                    top_entries[i] = Some(entry.clone());
                    current = entry;
                    agree += 1;

                    continue;
                }
                // If exact match, we agree.
                if let (_, true) = current.matches(Some(&entry), true) {
                    top_entries[i] = Some(entry);
                    agree += 1;

                    continue;
                }
                // If only the name matches we didn't agree, but add it for resolution.
                if entry.name == current.name {
                    top_entries[i] = Some(entry);
                    continue;
                }
                // We got different entries
                if entry.name > current.name {
                    continue;
                }

                for item in top_entries.iter_mut().take(i) {
                    *item = None;
                }

                agree = 1;
                top_entries[i] = Some(entry.clone());
                current = entry;
            }

            if vnf > 0 && vnf >= (readers.len() - opts.min_disks) {
                // warn!("list_path_raw: vnf > 0 && vnf >= (readers.len() - opts.min_disks) break");
                return Err(DiskError::VolumeNotFound);
            }

            if fnf > 0 && fnf >= (readers.len() - opts.min_disks) {
                // warn!("list_path_raw: fnf > 0 && fnf >= (readers.len() - opts.min_disks) break");
                return Err(DiskError::FileNotFound);
            }

            if has_err > 0 && has_err > opts.disks.len() - opts.min_disks {
                if let Some(finished_fn) = opts.finished.as_ref() {
                    finished_fn(&errs).await;
                }
                if errs.iter().flatten().any(|err| *err == DiskError::Timeout) {
                    return Err(DiskError::Timeout);
                }
                let mut err_iter = errs.iter().flatten();
                if let Some(err) = err_iter.next()
                    && err_iter.next().is_none()
                {
                    return Err(err.clone());
                }
                let mut combined_err = Vec::new();
                errs.iter().zip(opts.disks.iter()).for_each(|(err, disk)| match (err, disk) {
                    (Some(err), Some(disk)) => {
                        combined_err.push(format!("drive {} returned: {}", disk.to_string(), err));
                    }
                    (Some(err), None) => {
                        combined_err.push(err.to_string());
                    }
                    _ => {}
                });

                error!(
                    event = EVENT_METACACHE_LISTING,
                    component = LOG_COMPONENT_ECSTORE,
                    subsystem = LOG_SUBSYSTEM_METACACHE,
                    bucket = %opts.bucket,
                    path = %opts.path,
                    state = "quorum_failed",
                    error = %combined_err.join(", "),
                    "Metacache listing quorum failed"
                );
                return Err(DiskError::other(combined_err.join(", ")));
            }

            // Break if all at EOF or error.
            if at_eof + has_err == readers.len() {
                if has_err > 0
                    && let Some(finished_fn) = opts.finished.as_ref()
                {
                    finished_fn(&errs).await;
                }
                // All remaining readers are either at EOF or failed. Earlier logic
                // returned Timeout here for even a single stalled drive, despite
                // `has_err` being within the tolerated drive-failure budget. That
                // makes small distributed listings fail once healthy quorum readers
                // reach EOF but one remote walk stream is slow/stalled. Only the
                // `has_err > opts.disks.len() - opts.min_disks` branch above should
                // turn tolerated reader failures into request failures.
                // error!("list_path_raw: at_eof + has_err == readers.len() break {:?}", &errs);
                break;
            }

            if agree == readers.len() {
                for r in readers.iter_mut() {
                    let _ = r.skip(1).await;
                }

                if let Some(agreed_fn) = opts.agreed.as_ref() {
                    // warn!("list_path_raw: agreed_fn start, current: {:?}", &current.name);
                    agreed_fn(current).await;
                    // warn!("list_path_raw: agreed_fn done");
                }

                continue;
            }

            // warn!("list_path_raw: skip start, current: {:?}", &current.name);

            for (i, r) in readers.iter_mut().enumerate() {
                if top_entries[i].is_some() {
                    let _ = r.skip(1).await;
                }
            }

            if let Some(partial_fn) = opts.partial.as_ref() {
                partial_fn(MetaCacheEntries(top_entries), &errs).await;
            }
        }
        Ok(())
    });

    let merge_started = std::time::Instant::now();
    if let Err(err) = revjob.await.map_err(std::io::Error::other)? {
        rustfs_io_metrics::record_stage_duration("metacache_merge_failed", merge_started.elapsed().as_secs_f64() * 1000.0);
        if is_missing_path_error(&err) {
            debug!(
                event = EVENT_METACACHE_LISTING,
                component = LOG_COMPONENT_ECSTORE,
                subsystem = LOG_SUBSYSTEM_METACACHE,
                bucket = %log_bucket,
                path = %log_path,
                state = "merge_job_missing_path",
                error = ?err,
                "Metacache merge job missing path skipped"
            );
        } else {
            error!(
                event = EVENT_METACACHE_LISTING,
                component = LOG_COMPONENT_ECSTORE,
                subsystem = LOG_SUBSYSTEM_METACACHE,
                bucket = %log_bucket,
                path = %log_path,
                state = "merge_job_failed",
                error = ?err,
                "Metacache merge job failed"
            );
        }
        cancel_rx.cancel();
        for job in jobs {
            job.abort();
        }

        return Err(err);
    }
    rustfs_io_metrics::record_stage_duration("metacache_merge", merge_started.elapsed().as_secs_f64() * 1000.0);

    // The merge consumer can finish successfully before every producer finishes
    // (for example after reaching EOF quorum while a tolerated drive is stalled,
    // or after the requested listing limit is satisfied). Cancel remaining walk
    // jobs before aborting them so list calls do not wait for slow remote streams.
    cancel_rx.cancel();

    for job in jobs.iter() {
        if !job.is_finished() {
            job.abort();
        }
    }

    let results = join_all(jobs).await;
    let mut job_errs = Vec::new();
    for result in results {
        match result {
            Ok(Ok(())) => {}
            Ok(Err(err)) => {
                if is_missing_path_error(&err) {
                    debug!(
                        event = EVENT_METACACHE_LISTING,
                        component = LOG_COMPONENT_ECSTORE,
                        subsystem = LOG_SUBSYSTEM_METACACHE,
                        state = "producer_missing_path",
                        error = ?err,
                        "Metacache producer missing path"
                    );
                } else {
                    error!(
                        event = EVENT_METACACHE_LISTING,
                        component = LOG_COMPONENT_ECSTORE,
                        subsystem = LOG_SUBSYSTEM_METACACHE,
                        state = "producer_failed",
                        error = ?err,
                        "Metacache producer failed"
                    );
                }
                job_errs.push(err);
            }
            Err(err) => {
                if err.is_cancelled() {
                    continue;
                }
                error!(
                    event = EVENT_METACACHE_LISTING,
                    component = LOG_COMPONENT_ECSTORE,
                    subsystem = LOG_SUBSYSTEM_METACACHE,
                    state = "producer_join_failed",
                    error = ?err,
                    "Metacache producer join failed"
                );
                job_errs.push(err.into());
            }
        }
    }

    if job_errs.len() > max_disk_failures {
        return Err(job_errs.remove(0));
    }

    // warn!("list_path_raw: done");
    Ok(())
}

#[inline]
fn record_producer_error(producer_errs: &[OnceLock<DiskError>], idx: usize, err: &DiskError) {
    let _ = producer_errs[idx].set(err.clone());
}

#[inline]
fn producer_error(producer_errs: &[OnceLock<DiskError>], idx: usize) -> Option<DiskError> {
    producer_errs[idx].get().cloned()
}

#[cfg(test)]
mod tests {
    use super::*;
    use rustfs_filemeta::MetacacheWriter;
    use std::sync::Mutex;

    #[tokio::test]
    async fn list_path_raw_empty_disks_returns_read_quorum() {
        let err = list_path_raw(CancellationToken::new(), ListPathRawOptions::default())
            .await
            .expect_err("empty drive list should fail");

        assert_eq!(err, DiskError::ErasureReadQuorum);
    }

    #[test]
    fn missing_path_error_classification_excludes_actionable_failures() {
        assert!(is_missing_path_error(&DiskError::FileNotFound));
        assert!(is_missing_path_error(&DiskError::FileVersionNotFound));
        assert!(is_missing_path_error(&DiskError::VolumeNotFound));
        assert!(!is_missing_path_error(&DiskError::Timeout));
        assert!(!is_missing_path_error(&DiskError::DiskNotFound));
        assert!(!is_missing_path_error(&DiskError::FileAccessDenied));
    }

    #[tokio::test]
    async fn list_path_raw_returns_timeout_when_reader_stalls_before_completion() {
        let err = list_path_raw(
            CancellationToken::new(),
            ListPathRawOptions {
                disks: vec![None, None],
                min_disks: 2,
                test_reader_behaviors: vec![TestReaderBehavior::Stall, TestReaderBehavior::Eof],
                peek_timeout: Some(Duration::from_millis(20)),
                ..Default::default()
            },
        )
        .await
        .expect_err("stalled reader should fail when read quorum cannot be met");

        assert_eq!(err, DiskError::Timeout);
    }

    #[tokio::test]
    async fn list_path_raw_tolerates_stalled_reader_after_quorum_eof() {
        list_path_raw(
            CancellationToken::new(),
            ListPathRawOptions {
                disks: vec![None, None, None],
                min_disks: 2,
                test_reader_behaviors: vec![TestReaderBehavior::Eof, TestReaderBehavior::Eof, TestReaderBehavior::Stall],
                peek_timeout: Some(Duration::from_millis(20)),
                ..Default::default()
            },
        )
        .await
        .expect("listing should complete when healthy quorum reached EOF and only a tolerated drive stalled");
    }

    #[tokio::test]
    async fn list_path_raw_aborts_unresponsive_producer_after_quorum_eof() {
        let result = timeout(
            Duration::from_millis(200),
            list_path_raw(
                CancellationToken::new(),
                ListPathRawOptions {
                    disks: vec![None, None, None],
                    min_disks: 2,
                    test_reader_behaviors: vec![
                        TestReaderBehavior::Eof,
                        TestReaderBehavior::Eof,
                        TestReaderBehavior::IgnoreCancel,
                    ],
                    peek_timeout: Some(Duration::from_millis(20)),
                    ..Default::default()
                },
            ),
        )
        .await;

        let listing = result.expect("list_path_raw should abort unresponsive producer instead of hanging");
        assert!(listing.is_ok());
    }

    #[tokio::test]
    async fn list_path_raw_returns_timeout_when_producer_fails_after_partial_entry() {
        let seen = Arc::new(Mutex::new(Vec::new()));
        let seen_clone = seen.clone();

        let err = list_path_raw(
            CancellationToken::new(),
            ListPathRawOptions {
                disks: vec![None],
                min_disks: 1,
                test_reader_behaviors: vec![TestReaderBehavior::PartialThenTimeout(vec![MetaCacheEntry {
                    name: "bucket/object".to_string(),
                    metadata: vec![1, 2, 3],
                    cached: None,
                    reusable: false,
                }])],
                agreed: Some(Box::new(move |entry: MetaCacheEntry| {
                    let seen = seen_clone.clone();
                    Box::pin(async move {
                        seen.lock().expect("seen mutex poisoned").push(entry.name);
                    })
                })),
                ..Default::default()
            },
        )
        .await
        .expect_err("producer timeout after partial output must fail the listing");

        assert_eq!(err, DiskError::Timeout);
        assert_eq!(seen.lock().expect("seen mutex poisoned").as_slice(), &["bucket/object".to_string()]);
    }

    #[tokio::test]
    async fn peek_with_timeout_times_out_on_silent_reader() {
        let (_writer, reader) = tokio::io::duplex(64);
        let mut reader = MetacacheReader::new(reader);

        let outcome = peek_with_timeout(&mut reader, Duration::from_millis(20)).await;
        assert!(matches!(outcome, PeekOutcome::TimedOut));
    }

    #[tokio::test]
    async fn peek_with_timeout_reads_entry_before_deadline() {
        let (reader, writer) = tokio::io::duplex(256);
        let mut metacache_reader = MetacacheReader::new(reader);

        tokio::spawn(async move {
            let mut writer = MetacacheWriter::new(writer);
            let entry = MetaCacheEntry {
                name: "bucket/object".to_string(),
                metadata: vec![1, 2, 3],
                cached: None,
                reusable: false,
            };
            writer.write(&[entry]).await.expect("entry should be written");
            writer.close().await.expect("writer should close");
        });

        let outcome = peek_with_timeout(&mut metacache_reader, Duration::from_secs(1)).await;
        match outcome {
            PeekOutcome::Ready(Some(entry)) => assert_eq!(entry.name, "bucket/object"),
            other => panic!("expected ready entry, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn list_path_raw_propagates_producer_access_denied() {
        let err = list_path_raw(
            CancellationToken::new(),
            ListPathRawOptions {
                disks: vec![None],
                min_disks: 1,
                test_reader_behaviors: vec![TestReaderBehavior::ProducerError(DiskError::FileAccessDenied)],
                ..Default::default()
            },
        )
        .await
        .expect_err("producer access failure must not be treated as an empty listing");

        assert_eq!(err, DiskError::FileAccessDenied);
    }
}