// 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::filemeta::msgp_decode::MAX_MSGP_ELEMENT_SIZE;
use crate::{
    Error, FileInfo, FileInfoOpts, FileInfoVersions, FileMeta, FileMetaShallowVersion, Result, VersionType, get_file_info,
    merge_file_meta_versions, merge_file_meta_versions_with_write_quorum,
};
use arc_swap::ArcSwapOption;
use rmp::Marker;
use serde::{Deserialize, Serialize};
use std::cmp::Ordering;
use std::str::from_utf8;
use std::{
    fmt::Debug,
    future::Future,
    pin::Pin,
    sync::{
        Arc,
        atomic::{AtomicU64, Ordering as AtomicOrdering},
    },
    time::{Duration, SystemTime, UNIX_EPOCH},
};
use time::OffsetDateTime;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::spawn;
use tokio::sync::Mutex;
use tracing::{debug, warn};

const SLASH_SEPARATOR: &str = "/";

#[derive(Clone, Debug, Default)]
pub struct MetadataResolutionParams {
    pub dir_quorum: usize,
    pub obj_quorum: usize,
    pub requested_versions: usize,
    pub bucket: String,
    pub strict: bool,
    pub candidates: Vec<Vec<FileMetaShallowVersion>>,
}

#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
pub struct MetaCacheEntry {
    /// name is the full name of the object including prefixes
    pub name: String,
    /// Metadata. If none is present it is not an object but only a prefix.
    /// Entries without metadata will only be present in non-recursive scans.
    pub metadata: Vec<u8>,

    /// cached contains the metadata if decoded.
    #[serde(skip)]
    pub cached: Option<FileMeta>,

    /// Indicates the entry can be reused and only one reference to metadata is expected.
    pub reusable: bool,
}

impl MetaCacheEntry {
    pub fn marshal_msg(&self) -> Result<Vec<u8>> {
        let mut wr = Vec::new();
        rmp::encode::write_bool(&mut wr, true)?;
        rmp::encode::write_str(&mut wr, &self.name)?;
        rmp::encode::write_bin(&mut wr, &self.metadata)?;
        Ok(wr)
    }

    pub fn is_dir(&self) -> bool {
        self.metadata.is_empty() && self.name.ends_with('/')
    }

    pub fn is_in_dir(&self, dir: &str, separator: &str) -> bool {
        if dir.is_empty() {
            let idx = self.name.find(separator);
            return idx.is_none() || idx.unwrap() == self.name.len() - separator.len();
        }

        let ext = self.name.trim_start_matches(dir);

        if ext.len() != self.name.len() {
            let idx = ext.find(separator);
            return idx.is_none() || idx.unwrap() == ext.len() - separator.len();
        }

        false
    }

    pub fn is_object(&self) -> bool {
        !self.metadata.is_empty()
    }

    pub fn is_object_dir(&self) -> bool {
        !self.metadata.is_empty() && self.name.ends_with(SLASH_SEPARATOR)
    }

    pub fn is_latest_delete_marker(&mut self) -> bool {
        if let Some(cached) = &self.cached {
            if cached.versions.is_empty() {
                return true;
            }
            return cached.versions[0].header.version_type == VersionType::Delete;
        }

        if !FileMeta::is_xl2_v1_format(&self.metadata) {
            return false;
        }

        match FileMeta::is_indexed_meta(&self.metadata) {
            Ok((meta, _inline_data)) => {
                if !meta.is_empty() {
                    return FileMeta::is_latest_delete_marker(meta);
                }
            }
            Err(_) => return true,
        }

        match self.xl_meta() {
            Ok(res) => {
                if res.versions.is_empty() {
                    return true;
                }
                res.versions[0].header.version_type == VersionType::Delete
            }
            Err(_) => true,
        }
    }

    #[tracing::instrument(level = "debug", skip(self))]
    pub fn to_fileinfo(&self, bucket: &str) -> Result<FileInfo> {
        if self.is_dir() {
            return Ok(FileInfo {
                volume: bucket.to_owned(),
                name: self.name.clone(),
                ..Default::default()
            });
        }

        if let Some(fm) = &self.cached {
            if fm.versions.is_empty() {
                return Ok(FileInfo {
                    volume: bucket.to_owned(),
                    name: self.name.clone(),
                    deleted: true,
                    is_latest: true,
                    mod_time: Some(OffsetDateTime::UNIX_EPOCH),
                    ..Default::default()
                });
            }

            let fi = fm.into_fileinfo(bucket, self.name.as_str(), "", false, false, true)?;
            return Ok(fi);
        }

        get_file_info(
            &self.metadata,
            bucket,
            self.name.as_str(),
            "",
            FileInfoOpts {
                data: false,
                include_free_versions: false,
            },
        )
    }

    pub fn file_info_versions(&self, bucket: &str) -> Result<FileInfoVersions> {
        if self.is_dir() {
            return Ok(FileInfoVersions {
                volume: bucket.to_string(),
                name: self.name.clone(),
                versions: vec![FileInfo {
                    volume: bucket.to_string(),
                    name: self.name.clone(),
                    ..Default::default()
                }],
                ..Default::default()
            });
        }

        let mut fm = FileMeta::new();
        fm.unmarshal_msg(&self.metadata)?;
        let mut versions = fm.get_file_info_versions(bucket, self.name.as_str(), false)?;
        versions.free_versions.clear();
        Ok(versions)
    }

    pub fn file_info_versions_with_free_versions(&self, bucket: &str) -> Result<FileInfoVersions> {
        if self.is_dir() {
            return Ok(FileInfoVersions {
                volume: bucket.to_string(),
                name: self.name.clone(),
                versions: vec![FileInfo {
                    volume: bucket.to_string(),
                    name: self.name.clone(),
                    ..Default::default()
                }],
                ..Default::default()
            });
        }

        let mut fm = FileMeta::new();
        fm.unmarshal_msg(&self.metadata)?;
        // `get_file_info_versions(..., false)` is the existing path that
        // separates persisted tier free-version records into `free_versions`.
        fm.get_file_info_versions(bucket, self.name.as_str(), false)
    }

    pub fn matches(&self, other: Option<&MetaCacheEntry>, strict: bool) -> (Option<MetaCacheEntry>, bool) {
        if other.is_none() {
            return (None, false);
        }

        let other = other.unwrap();
        if self.name != other.name {
            if self.name < other.name {
                return (Some(self.clone()), false);
            }
            return (Some(other.clone()), false);
        }

        if other.is_dir() || self.is_dir() {
            if self.is_dir() {
                return (Some(self.clone()), other.is_dir() == self.is_dir());
            }
            return (Some(other.clone()), other.is_dir() == self.is_dir());
        }

        let self_vers = match &self.cached {
            Some(file_meta) => file_meta.clone(),
            None => match FileMeta::load(&self.metadata) {
                Ok(meta) => meta,
                Err(_) => return (None, false),
            },
        };

        let other_vers = match &other.cached {
            Some(file_meta) => file_meta.clone(),
            None => match FileMeta::load(&other.metadata) {
                Ok(meta) => meta,
                Err(_) => return (None, false),
            },
        };

        if self_vers.versions.len() != other_vers.versions.len() {
            match self_vers.latest_mod_time().cmp(&other_vers.latest_mod_time()) {
                Ordering::Greater => return (Some(self.clone()), false),
                Ordering::Less => return (Some(other.clone()), false),
                _ => {}
            }

            if self_vers.versions.len() > other_vers.versions.len() {
                return (Some(self.clone()), false);
            }
            return (Some(other.clone()), false);
        }

        let mut prefer = None;
        for (s_version, o_version) in self_vers.versions.iter().zip(other_vers.versions.iter()) {
            if s_version.header != o_version.header {
                if s_version.header.has_ec() != o_version.header.has_ec() {
                    // One version has EC and the other doesn't - may have been written later.
                    // Compare without considering EC.
                    let (mut a, mut b) = (s_version.header.clone(), o_version.header.clone());
                    (a.ec_n, a.ec_m, b.ec_n, b.ec_m) = (0, 0, 0, 0);
                    if a == b {
                        continue;
                    }
                }

                if !strict && s_version.header.matches_not_strict(&o_version.header) {
                    if prefer.is_none() {
                        if s_version.header.sorts_before(&o_version.header) {
                            prefer = Some(self.clone());
                        } else {
                            prefer = Some(other.clone());
                        }
                    }
                    continue;
                }

                if prefer.is_some() {
                    return (prefer, false);
                }

                if s_version.header.sorts_before(&o_version.header) {
                    return (Some(self.clone()), false);
                }

                return (Some(other.clone()), false);
            }
        }

        if prefer.is_none() {
            prefer = Some(self.clone());
        }

        (prefer, true)
    }

    pub fn xl_meta(&mut self) -> Result<FileMeta> {
        if self.is_dir() {
            return Err(Error::FileNotFound);
        }

        if let Some(meta) = &self.cached {
            Ok(meta.clone())
        } else {
            if self.metadata.is_empty() {
                return Err(Error::FileNotFound);
            }

            let meta = FileMeta::load(&self.metadata)?;
            self.cached = Some(meta.clone());
            Ok(meta)
        }
    }
}

#[derive(Debug, Default)]
pub struct MetaCacheEntries(pub Vec<Option<MetaCacheEntry>>);

impl MetaCacheEntries {
    #[allow(clippy::should_implement_trait)]
    pub fn as_ref(&self) -> &[Option<MetaCacheEntry>] {
        &self.0
    }

    pub fn resolve(&self, params: MetadataResolutionParams) -> Option<MetaCacheEntry> {
        self.resolve_inner(params, false)
    }

    pub fn resolve_with_write_quorum(&self, params: MetadataResolutionParams) -> Option<MetaCacheEntry> {
        self.resolve_inner(params, true)
    }

    /// Resolve the cross-disk UNION of every version present on ANY disk slot.
    ///
    /// This mirrors MinIO's global heal enumeration (cmd/global-heal.go
    /// `healErasureSet` -> `listPathRaw` with `objQuorum = 1` feeding
    /// `mergeEntries`/`mergeXLV2Versions`): at quorum 1 the merge is forced
    /// strict and returns the union of all per-disk version streams, so a version
    /// that survives on FEWER than read-quorum disks is still surfaced for
    /// healing. Read-quorum resolution (`resolve` with `obj_quorum >= 2`) would
    /// silently drop such a sub-quorum version, which is exactly the durability
    /// gap this enumerator closes (backlog#920).
    ///
    /// `bucket` is only used to tag the merged entry; `strict:false` lets the
    /// non-strict header reconciliation collapse equal-but-differently-signed
    /// replicas of the SAME version id while still retaining genuinely distinct
    /// version ids.
    pub fn resolve_union(&self, bucket: &str) -> Option<MetaCacheEntry> {
        self.resolve(MetadataResolutionParams {
            dir_quorum: 1,
            obj_quorum: 1,
            requested_versions: 0,
            bucket: bucket.to_string(),
            strict: false,
            candidates: Vec::new(),
        })
    }

    fn resolve_inner(&self, mut params: MetadataResolutionParams, enforce_write_quorum: bool) -> Option<MetaCacheEntry> {
        if self.0.is_empty() {
            debug!(
                bucket = %params.bucket,
                dir_quorum = params.dir_quorum,
                obj_quorum = params.obj_quorum,
                "metacache resolve skipped because there were no entries to reconcile"
            );
            return None;
        }

        let mut dir_exists = 0;
        let mut selected = None;

        params.candidates.clear();
        let mut objs_agree = 0;
        let mut objs_valid = 0;

        for entry in self.0.iter().flatten() {
            let mut entry = entry.clone();

            debug!(entry = %entry.name, "metacache resolve examining candidate entry");
            if entry.name.is_empty() {
                continue;
            }
            if entry.is_dir() {
                dir_exists += 1;
                selected = Some(entry.clone());
                debug!(entry = %entry.name, "metacache resolve observed directory candidate");
                continue;
            }

            let xl = match entry.xl_meta() {
                Ok(xl) => xl,
                Err(e) => {
                    debug!(entry = %entry.name, error = ?e, "metacache resolve skipped candidate with unreadable xl.meta");
                    continue;
                }
            };

            objs_valid += 1;
            params.candidates.push(xl.versions.clone());

            if selected.is_none() {
                selected = Some(entry.clone());
                objs_agree = 1;
                debug!(entry = %entry.name, "metacache resolve selected initial candidate");
                continue;
            }

            if let (prefer, true) = entry.matches(selected.as_ref(), params.strict) {
                selected = prefer;
                objs_agree += 1;
                debug!(entry = %entry.name, "metacache resolve preferred candidate during reconciliation");
                continue;
            }
        }

        let Some(selected) = selected else {
            debug!(
                bucket = %params.bucket,
                dir_quorum = params.dir_quorum,
                obj_quorum = params.obj_quorum,
                "metacache resolve could not select any candidate entry"
            );
            return None;
        };

        if selected.is_dir() && dir_exists >= params.dir_quorum {
            debug!(
                entry = %selected.name,
                dir_exists,
                dir_quorum = params.dir_quorum,
                "metacache resolve selected directory candidate after quorum reconciliation"
            );
            return Some(selected);
        }

        // If we would never be able to reach the required object quorum.
        if objs_valid < params.obj_quorum {
            debug!(
                objs_valid,
                obj_quorum = params.obj_quorum,
                "metacache resolve did not have enough valid object candidates to satisfy quorum"
            );
            return None;
        }

        if objs_agree == objs_valid {
            let required_quorum = if enforce_write_quorum {
                selected
                    .cached
                    .as_ref()
                    .and_then(|cached| cached.versions.first())
                    .map(|version| version.write_quorum(params.obj_quorum).max(params.obj_quorum))
                    .unwrap_or(params.obj_quorum)
            } else {
                params.obj_quorum
            };

            if objs_agree >= required_quorum {
                debug!(
                    selected = %selected.name,
                    objs_agree,
                    objs_valid,
                    "metacache resolve reused selected candidate because all valid object entries agreed"
                );
                return Some(selected);
            }
        }

        let Some(cached) = selected.cached else {
            debug!(selected = %selected.name, "metacache resolve could not merge because selected candidate had no cached metadata");
            return None;
        };

        let versions = if enforce_write_quorum {
            merge_file_meta_versions_with_write_quorum(
                params.obj_quorum,
                params.strict,
                params.requested_versions,
                &params.candidates,
            )
        } else {
            merge_file_meta_versions(params.obj_quorum, params.strict, params.requested_versions, &params.candidates)
        };
        if versions.is_empty() {
            debug!(
                selected = %selected.name,
                requested_versions = params.requested_versions,
                "metacache resolve produced no merged versions after reconciliation"
            );
            return None;
        }

        let merged_cached = FileMeta {
            meta_ver: cached.meta_ver,
            versions,
            ..Default::default()
        };

        let metadata = match merged_cached.marshal_msg() {
            Ok(meta) => meta,
            Err(e) => {
                warn!(
                    selected = %selected.name,
                    error = ?e,
                    "metacache resolve failed to marshal merged metadata entry"
                );
                return None;
            }
        };

        // Merge if we have disagreement.
        // Create a new merged result.
        let new_selected = MetaCacheEntry {
            name: selected.name.clone(),
            cached: Some(merged_cached),
            reusable: true,
            metadata,
        };

        debug!(
            selected = %new_selected.name,
            candidate_count = params.candidates.len(),
            obj_quorum = params.obj_quorum,
            "metacache resolve produced merged metadata entry after reconciling disagreeing candidates"
        );
        Some(new_selected)
    }

    pub fn first_found(&self) -> (Option<MetaCacheEntry>, usize) {
        (self.0.iter().find(|x| x.is_some()).cloned().unwrap_or_default(), self.0.len())
    }
}

#[derive(Debug, Default)]
pub struct MetaCacheEntriesSortedResult {
    pub entries: Option<MetaCacheEntriesSorted>,
    pub err: Option<Error>,
}

#[derive(Debug, Default)]
pub struct MetaCacheEntriesSorted {
    pub o: MetaCacheEntries,
    pub list_id: Option<String>,
    pub reuse: bool,
    pub last_skipped_entry: Option<String>,
}

impl MetaCacheEntriesSorted {
    pub fn entries(&self) -> Vec<&MetaCacheEntry> {
        let entries: Vec<&MetaCacheEntry> = self.o.0.iter().flatten().collect();
        entries
    }

    pub fn forward_past(&mut self, marker: Option<String>) {
        if let Some(val) = marker
            && let Some(idx) = self.o.0.iter().flatten().position(|v| v.name > val)
        {
            self.o.0 = self.o.0.split_off(idx);
        }
    }
}

const METACACHE_STREAM_VERSION: u8 = 2;

#[derive(Debug)]
pub struct MetacacheWriter<W> {
    wr: W,
    created: bool,
    buf: Vec<u8>,
}

impl<W: AsyncWrite + Unpin> MetacacheWriter<W> {
    pub fn new(wr: W) -> Self {
        Self {
            wr,
            created: false,
            buf: Vec::new(),
        }
    }

    pub async fn flush(&mut self) -> Result<()> {
        self.wr.write_all(&self.buf).await?;
        self.buf.clear();
        Ok(())
    }

    pub async fn init(&mut self) -> Result<()> {
        if !self.created {
            rmp::encode::write_u8(&mut self.buf, METACACHE_STREAM_VERSION).map_err(|e| Error::other(format!("{e:?}")))?;
            self.flush().await?;
            self.created = true;
        }
        Ok(())
    }

    pub async fn write(&mut self, objs: &[MetaCacheEntry]) -> Result<()> {
        if objs.is_empty() {
            return Ok(());
        }

        self.init().await?;

        for obj in objs.iter() {
            if obj.name.is_empty() {
                return Err(Error::other("metacacheWriter: no name"));
            }

            self.write_obj(obj).await?;
        }

        Ok(())
    }

    pub async fn write_obj(&mut self, obj: &MetaCacheEntry) -> Result<()> {
        self.init().await?;

        rmp::encode::write_bool(&mut self.buf, true).map_err(|e| Error::other(format!("{e:?}")))?;
        rmp::encode::write_str(&mut self.buf, &obj.name).map_err(|e| Error::other(format!("{e:?}")))?;
        rmp::encode::write_bin(&mut self.buf, &obj.metadata).map_err(|e| Error::other(format!("{e:?}")))?;
        self.flush().await?;

        Ok(())
    }

    pub async fn close(&mut self) -> Result<()> {
        rmp::encode::write_bool(&mut self.buf, false).map_err(|e| Error::other(format!("{e:?}")))?;
        self.flush().await?;
        Ok(())
    }
}

pub struct MetacacheReader<R> {
    rd: R,
    init: bool,
    err: Option<Error>,
    buf: Vec<u8>,
    offset: usize,
    current: Option<MetaCacheEntry>,
}

impl<R: AsyncRead + Unpin> MetacacheReader<R> {
    pub fn new(rd: R) -> Self {
        Self {
            rd,
            init: false,
            err: None,
            buf: Vec::new(),
            offset: 0,
            current: None,
        }
    }

    pub async fn read_more(&mut self, read_size: usize) -> Result<&[u8]> {
        // `read_size` is usually a length decoded from the stream itself, so
        // it is untrusted: a corrupted length prefix must yield a decode
        // error instead of a huge allocation that aborts the process
        // (see rustfs/rustfs#2715).
        if read_size > MAX_MSGP_ELEMENT_SIZE {
            let err = Error::other(format!(
                "metacache stream corrupt: element length {read_size} exceeds the {MAX_MSGP_ELEMENT_SIZE} byte limit"
            ));
            self.err = Some(err.clone());
            return Err(err);
        }

        let pref = self.offset;
        let ext_size = pref + read_size;

        if self.buf.len() < ext_size {
            let extra = ext_size - self.buf.len();
            if let Err(e) = self.buf.try_reserve(extra) {
                let err = Error::other(format!("metacache stream: buffer allocation of {extra} bytes failed: {e}"));
                self.err = Some(err.clone());
                return Err(err);
            }
            self.buf.resize(ext_size, 0);
        }

        self.rd.read_exact(&mut self.buf[pref..ext_size]).await?;

        self.offset += read_size;

        Ok(&self.buf[pref..ext_size])
    }

    fn reset(&mut self) {
        self.buf.clear();
        self.offset = 0;
    }

    async fn check_init(&mut self) -> Result<()> {
        if !self.init {
            let ver = match rmp::decode::read_u8(&mut self.read_more(2).await?) {
                Ok(res) => res,
                Err(err) => {
                    self.err = Some(Error::other(format!("{err:?}")));
                    0
                }
            };
            match ver {
                1 | 2 => (),
                _ => {
                    self.err = Some(Error::other("invalid version"));
                }
            }

            self.init = true;
        }
        Ok(())
    }

    async fn read_str_len(&mut self) -> Result<u32> {
        let mark = match rmp::decode::read_marker(&mut self.read_more(1).await?) {
            Ok(res) => res,
            Err(err) => {
                let err: Error = err.into();
                self.err = Some(err.clone());
                return Err(err);
            }
        };

        match mark {
            Marker::FixStr(size) => Ok(u32::from(size)),
            Marker::Str8 => Ok(u32::from(self.read_u8().await?)),
            Marker::Str16 => Ok(u32::from(self.read_u16().await?)),
            Marker::Str32 => Ok(self.read_u32().await?),
            _marker => Err(Error::other("str marker err")),
        }
    }

    async fn read_bin_len(&mut self) -> Result<u32> {
        let mark = match rmp::decode::read_marker(&mut self.read_more(1).await?) {
            Ok(res) => res,
            Err(err) => {
                let err: Error = err.into();
                self.err = Some(err.clone());
                return Err(err);
            }
        };

        match mark {
            Marker::Bin8 => Ok(u32::from(self.read_u8().await?)),
            Marker::Bin16 => Ok(u32::from(self.read_u16().await?)),
            Marker::Bin32 => Ok(self.read_u32().await?),
            _ => Err(Error::other("bin marker err")),
        }
    }

    async fn read_u8(&mut self) -> Result<u8> {
        let buf = self.read_more(1).await?;
        Ok(u8::from_be_bytes(buf.try_into().expect("Slice with incorrect length")))
    }

    async fn read_u16(&mut self) -> Result<u16> {
        let buf = self.read_more(2).await?;
        Ok(u16::from_be_bytes(buf.try_into().expect("Slice with incorrect length")))
    }

    async fn read_u32(&mut self) -> Result<u32> {
        let buf = self.read_more(4).await?;
        Ok(u32::from_be_bytes(buf.try_into().expect("Slice with incorrect length")))
    }

    pub async fn skip(&mut self, size: usize) -> Result<()> {
        self.check_init().await?;

        if let Some(err) = &self.err {
            return Err(err.clone());
        }

        let mut n = size;

        if self.current.is_some() {
            n -= 1;
            self.current = None;
        }

        while n > 0 {
            match rmp::decode::read_bool(&mut self.read_more(1).await?) {
                Ok(res) => {
                    if !res {
                        return Ok(());
                    }
                }
                Err(err) => {
                    let err: Error = err.into();
                    self.err = Some(err.clone());
                    return Err(err);
                }
            };

            let l = self.read_str_len().await?;
            let _ = self.read_more(l as usize).await?;
            let l = self.read_bin_len().await?;
            let _ = self.read_more(l as usize).await?;

            n -= 1;
        }

        Ok(())
    }

    pub async fn peek(&mut self) -> Result<Option<MetaCacheEntry>> {
        self.check_init().await?;

        if let Some(err) = &self.err {
            return Err(err.clone());
        }

        match rmp::decode::read_bool(&mut self.read_more(1).await?) {
            Ok(res) => {
                if !res {
                    return Ok(None);
                }
            }
            Err(err) => {
                let err: Error = err.into();
                self.err = Some(err.clone());
                return Err(err);
            }
        };

        let l = self.read_str_len().await?;

        let buf = self.read_more(l as usize).await?;
        let name_buf = buf.to_vec();
        let name = match from_utf8(&name_buf) {
            Ok(decoded) => decoded.to_owned(),
            Err(err) => {
                self.err = Some(Error::other(err.to_string()));
                return Err(Error::other(err.to_string()));
            }
        };

        let l = self.read_bin_len().await?;

        let buf = self.read_more(l as usize).await?;

        let metadata = buf.to_vec();

        self.reset();

        let entry = Some(MetaCacheEntry {
            name,
            metadata,
            cached: None,
            reusable: false,
        });
        self.current = entry.clone();

        Ok(entry)
    }

    pub async fn read_all(&mut self) -> Result<Vec<MetaCacheEntry>> {
        let mut ret = Vec::new();

        loop {
            if let Some(entry) = self.peek().await? {
                ret.push(entry);
                continue;
            }
            break;
        }

        Ok(ret)
    }
}

pub type UpdateFn<T> = Box<dyn Fn() -> Pin<Box<dyn Future<Output = std::io::Result<T>> + Send>> + Send + Sync + 'static>;

#[derive(Clone, Debug, Default)]
pub struct Opts {
    pub return_last_good: bool,
    pub no_wait: bool,
}

pub struct Cache<T: Clone + Debug + Send> {
    update_fn: UpdateFn<T>,
    ttl: Duration,
    opts: Opts,
    val: ArcSwapOption<T>,
    last_update_secs: AtomicU64,
    updating: Arc<Mutex<()>>,
}

impl<T: Clone + Debug + Send + Sync + 'static> Cache<T> {
    pub fn new(update_fn: UpdateFn<T>, ttl: Duration, opts: Opts) -> Self {
        Self {
            update_fn,
            ttl,
            opts,
            val: ArcSwapOption::from(None),
            last_update_secs: AtomicU64::new(0),
            updating: Arc::new(Mutex::new(())),
        }
    }

    pub async fn get(self: Arc<Self>) -> std::io::Result<T> {
        let value = self.get_shared().await?;
        Ok(value.as_ref().clone())
    }

    pub async fn get_shared(self: Arc<Self>) -> std::io::Result<Arc<T>> {
        let now = Self::current_unix_secs();
        let current = self.cached_value();
        if self.age_since_last_update(now) < self.ttl.as_secs()
            && let Some(value) = current.clone()
        {
            return Ok(value);
        }

        if self.opts.no_wait
            && self.age_since_last_update(now) < self.ttl.as_secs().saturating_mul(2)
            && let Some(value) = current
        {
            if let Ok(update_guard) = Arc::clone(&self.updating).try_lock_owned() {
                let this = Arc::clone(&self);
                spawn(async move {
                    let _guard = update_guard;
                    let _ = this.update().await;
                });
            }
            return Ok(value);
        }

        let _guard = self.updating.lock().await;

        let now = Self::current_unix_secs();
        if self.age_since_last_update(now) < self.ttl.as_secs()
            && let Some(value) = self.cached_value()
        {
            return Ok(value);
        }

        self.update().await?;
        self.cached_value()
            .ok_or_else(|| std::io::Error::other("cache update completed without a value"))
    }

    async fn update(&self) -> std::io::Result<()> {
        match (self.update_fn)().await {
            Ok(val) => {
                self.val.store(Some(Arc::new(val)));
                self.last_update_secs.store(Self::current_unix_secs(), AtomicOrdering::SeqCst);
                Ok(())
            }
            Err(err) => {
                if self.opts.return_last_good && self.cached_value().is_some() {
                    return Ok(());
                }

                Err(err)
            }
        }
    }

    fn current_unix_secs() -> u64 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("Time went backwards")
            .as_secs()
    }

    fn age_since_last_update(&self, now_secs: u64) -> u64 {
        now_secs
            .checked_sub(self.last_update_secs.load(AtomicOrdering::SeqCst))
            .unwrap_or(u64::MAX)
    }

    fn cached_value(&self) -> Option<Arc<T>> {
        self.val.load_full()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_data::create_real_xlmeta;
    use crate::{FileMetaVersion, MetaDeleteMarker, TRANSITION_COMPLETE};
    use std::collections::HashMap;
    use std::io::Cursor;
    use std::sync::{
        Arc, Mutex as StdMutex,
        atomic::{AtomicUsize, Ordering},
    };
    use tokio::sync::{Notify, oneshot};
    use uuid::Uuid;

    #[tokio::test]
    async fn test_writer() {
        let mut f = Cursor::new(Vec::new());
        let mut w = MetacacheWriter::new(&mut f);

        let mut objs = Vec::new();
        for i in 0..10 {
            let info = MetaCacheEntry {
                name: format!("item{i}"),
                metadata: vec![0u8, 10],
                cached: None,
                reusable: false,
            };
            objs.push(info);
        }

        w.write(&objs).await.unwrap();
        w.close().await.unwrap();

        let data = f.into_inner();
        let nf = Cursor::new(data);

        let mut r = MetacacheReader::new(nf);
        let nobjs = r.read_all().await.unwrap();

        assert_eq!(objs, nobjs);
    }

    fn corrupt_stream_with_metadata_len(len_marker: u8, len: u32) -> Vec<u8> {
        let mut data = Vec::new();
        rmp::encode::write_u8(&mut data, METACACHE_STREAM_VERSION).unwrap();
        rmp::encode::write_bool(&mut data, true).unwrap();
        rmp::encode::write_str(&mut data, "object").unwrap();
        // Hand-written bin/str length prefix claiming an absurd payload size.
        data.push(len_marker);
        data.extend_from_slice(&len.to_be_bytes());
        data
    }

    /// Regression test for rustfs/rustfs#2715: a corrupted metadata length
    /// prefix must surface as a decode error instead of attempting a huge
    /// allocation that aborts the process.
    #[tokio::test]
    async fn test_reader_rejects_corrupt_bin_length_prefix() {
        // 0xc6 = bin32 marker.
        let data = corrupt_stream_with_metadata_len(0xc6, u32::MAX);
        let mut r = MetacacheReader::new(Cursor::new(data));
        let err = r.peek().await.expect_err("corrupt bin length prefix must fail to decode");
        assert!(err.to_string().contains("exceeds"), "error should mention the exceeded limit, got: {err}");

        // The reader must stay in the error state instead of retrying.
        assert!(r.peek().await.is_err());
    }

    #[tokio::test]
    async fn test_reader_rejects_corrupt_str_length_prefix() {
        let mut data = Vec::new();
        rmp::encode::write_u8(&mut data, METACACHE_STREAM_VERSION).unwrap();
        rmp::encode::write_bool(&mut data, true).unwrap();
        // 0xdb = str32 marker with an absurd object-name length.
        data.push(0xdb);
        data.extend_from_slice(&u32::MAX.to_be_bytes());

        let mut r = MetacacheReader::new(Cursor::new(data));
        let err = r.peek().await.expect_err("corrupt str length prefix must fail to decode");
        assert!(err.to_string().contains("exceeds"), "error should mention the exceeded limit, got: {err}");
    }

    #[tokio::test]
    async fn test_reader_skip_rejects_corrupt_length_prefix() {
        let data = corrupt_stream_with_metadata_len(0xc6, u32::MAX);
        let mut r = MetacacheReader::new(Cursor::new(data));
        assert!(r.skip(1).await.is_err(), "skip over a corrupt length prefix must fail");
    }

    /// Large-but-legitimate records (well above typical sizes, below the
    /// corruption guard) must still round-trip.
    #[tokio::test]
    async fn test_reader_accepts_large_legitimate_metadata() {
        let entry = MetaCacheEntry {
            name: "big-object".to_string(),
            metadata: vec![0xab; 4 << 20],
            cached: None,
            reusable: false,
        };

        let mut f = Cursor::new(Vec::new());
        let mut w = MetacacheWriter::new(&mut f);
        w.write(std::slice::from_ref(&entry)).await.unwrap();
        w.close().await.unwrap();

        let mut r = MetacacheReader::new(Cursor::new(f.into_inner()));
        let decoded = r.read_all().await.unwrap();
        assert_eq!(decoded, vec![entry]);
    }

    #[test]
    fn file_info_versions_with_free_versions_includes_persisted_tier_cleanup_records() {
        let version_id = Uuid::new_v4();
        let remote_version_id = Uuid::new_v4();
        let free_version_id = Uuid::new_v4();
        let mut fm = FileMeta::new();
        fm.add_version(FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            version_id: Some(version_id),
            transition_status: TRANSITION_COMPLETE.to_string(),
            transitioned_objname: "remote/object".to_string(),
            transition_version_id: Some(remote_version_id),
            transition_tier: "WARM".to_string(),
            mod_time: Some(OffsetDateTime::now_utc()),
            ..Default::default()
        })
        .expect("transitioned version should be added");

        let mut delete_fi = FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            version_id: Some(version_id),
            mod_time: Some(OffsetDateTime::now_utc()),
            ..Default::default()
        };
        delete_fi.set_tier_free_version_id(&free_version_id.to_string());
        fm.delete_version(&delete_fi)
            .expect("transitioned delete should create free-version metadata");

        let entry = MetaCacheEntry {
            name: "object".to_string(),
            metadata: fm.marshal_msg().expect("metadata should marshal"),
            ..Default::default()
        };

        let normal = entry.file_info_versions("bucket").expect("normal versions should parse");
        assert!(normal.free_versions.is_empty());

        let with_free = entry
            .file_info_versions_with_free_versions("bucket")
            .expect("versions with free versions should parse");
        assert_eq!(with_free.free_versions.len(), 1);
        assert!(with_free.free_versions[0].tier_free_version());
        assert_eq!(with_free.free_versions[0].transitioned_objname, "remote/object");
        assert_eq!(with_free.free_versions[0].transition_tier, "WARM");
        assert_eq!(with_free.free_versions[0].transition_version_id, Some(remote_version_id));
    }

    #[test]
    fn file_info_versions_excludes_free_versions_from_visible_version_count() {
        let object_version_id = Uuid::new_v4();
        let remote_version_id = Uuid::new_v4();
        let free_version_id = Uuid::new_v4();
        let delete_marker_id = Uuid::new_v4();
        let base_time = OffsetDateTime::now_utc();
        let mut fm = FileMeta::new();

        fm.add_version(FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            version_id: Some(object_version_id),
            transition_status: TRANSITION_COMPLETE.to_string(),
            transitioned_objname: "remote/object".to_string(),
            transition_version_id: Some(remote_version_id),
            transition_tier: "WARM".to_string(),
            mod_time: Some(base_time),
            ..Default::default()
        })
        .expect("transitioned object version should be added");

        let mut delete_fi = FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            version_id: Some(object_version_id),
            mod_time: Some(base_time),
            ..Default::default()
        };
        delete_fi.set_tier_free_version_id(&free_version_id.to_string());
        fm.delete_version(&delete_fi)
            .expect("transitioned delete should create a free-version record");

        fm.add_version(FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            version_id: Some(delete_marker_id),
            deleted: true,
            mod_time: Some(base_time + time::Duration::seconds(1)),
            ..Default::default()
        })
        .expect("delete marker should be added");

        let entry = MetaCacheEntry {
            name: "object".to_string(),
            metadata: fm.marshal_msg().expect("metadata should marshal"),
            ..Default::default()
        };

        let normal = entry.file_info_versions("bucket").expect("normal versions should parse");
        assert_eq!(normal.versions.len(), 1);
        assert!(normal.free_versions.is_empty());
        assert!(normal.versions[0].deleted);
        assert_eq!(normal.versions[0].num_versions, 1);

        let with_free = entry
            .file_info_versions_with_free_versions("bucket")
            .expect("versions with free versions should parse");
        assert_eq!(with_free.versions.len(), 1);
        assert_eq!(with_free.free_versions.len(), 1);
        assert_eq!(with_free.versions[0].num_versions, 1);
        assert_eq!(with_free.free_versions[0].num_versions, 1);
    }

    #[test]
    fn transitioned_delete_persists_recoverable_free_version_after_metadata_roundtrip() {
        let version_id = Uuid::new_v4();
        let remote_version_id = Uuid::new_v4();
        let free_version_id = Uuid::new_v4();
        let mut fm = FileMeta::new();
        fm.add_version(FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            version_id: Some(version_id),
            transition_status: TRANSITION_COMPLETE.to_string(),
            transitioned_objname: "remote/object".to_string(),
            transition_version_id: Some(remote_version_id),
            transition_tier: "WARM".to_string(),
            mod_time: Some(OffsetDateTime::now_utc()),
            ..Default::default()
        })
        .expect("transitioned version should be added");

        let mut delete_fi = FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            version_id: Some(version_id),
            mod_time: Some(OffsetDateTime::now_utc()),
            ..Default::default()
        };
        delete_fi.set_tier_free_version_id(&free_version_id.to_string());

        fm.delete_version(&delete_fi)
            .expect("transitioned delete should persist free-version metadata");
        let encoded = fm.marshal_msg().expect("metadata should marshal");
        let mut decoded = FileMeta::new();
        decoded
            .unmarshal_msg(&encoded)
            .expect("metadata should survive process restart roundtrip");

        let versions = decoded
            .get_file_info_versions("bucket", "object", false)
            .expect("versions with free versions should parse");

        assert!(versions.versions.is_empty());
        assert_eq!(versions.free_versions.len(), 1);
        let free_version = &versions.free_versions[0];
        assert_eq!(free_version.version_id, Some(free_version_id));
        assert!(free_version.tier_free_version());
        assert_eq!(free_version.transitioned_objname, "remote/object");
        assert_eq!(free_version.transition_version_id, Some(remote_version_id));
        assert_eq!(free_version.transition_tier, "WARM");
    }

    #[test]
    fn skip_tier_free_version_does_not_persist_cleanup_record() {
        let version_id = Uuid::new_v4();
        let mut fm = FileMeta::new();
        fm.add_version(FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            version_id: Some(version_id),
            transition_status: TRANSITION_COMPLETE.to_string(),
            transitioned_objname: "remote/object".to_string(),
            transition_version_id: Some(Uuid::new_v4()),
            transition_tier: "WARM".to_string(),
            mod_time: Some(OffsetDateTime::now_utc()),
            ..Default::default()
        })
        .expect("transitioned version should be added");

        let mut delete_fi = FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            version_id: Some(version_id),
            mod_time: Some(OffsetDateTime::now_utc()),
            ..Default::default()
        };
        delete_fi.set_tier_free_version_id(&Uuid::new_v4().to_string());
        delete_fi.set_skip_tier_free_version();

        fm.delete_version(&delete_fi)
            .expect("transitioned delete with skip flag should remove the local version");
        let versions = fm
            .get_file_info_versions("bucket", "object", false)
            .expect("versions should parse after skipped cleanup");

        assert!(versions.free_versions.is_empty());
        assert_eq!(versions.versions.len(), 1);
        assert!(versions.versions[0].deleted);
        assert!(!versions.versions[0].tier_free_version());
    }

    #[test]
    fn worker_style_free_version_delete_removes_persisted_cleanup_record() {
        let version_id = Uuid::new_v4();
        let remote_version_id = Uuid::new_v4();
        let free_version_id = Uuid::new_v4();
        let mut fm = FileMeta::new();
        fm.add_version(FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            version_id: Some(version_id),
            transition_status: TRANSITION_COMPLETE.to_string(),
            transitioned_objname: "remote/object".to_string(),
            transition_version_id: Some(remote_version_id),
            transition_tier: "WARM".to_string(),
            mod_time: Some(OffsetDateTime::now_utc()),
            ..Default::default()
        })
        .expect("transitioned version should be added");

        let mut delete_fi = FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            version_id: Some(version_id),
            mod_time: Some(OffsetDateTime::now_utc()),
            ..Default::default()
        };
        delete_fi.set_tier_free_version_id(&free_version_id.to_string());
        fm.delete_version(&delete_fi)
            .expect("transitioned delete should persist free-version metadata");

        let mut free_delete_fi = FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            version_id: Some(free_version_id),
            deleted: true,
            ..Default::default()
        };
        free_delete_fi.set_tier_free_version();

        fm.delete_version(&free_delete_fi)
            .expect("worker-style free-version delete should remove the cleanup record");
        let versions = fm
            .get_file_info_versions("bucket", "object", false)
            .expect("versions should parse after free-version cleanup");

        assert!(versions.free_versions.is_empty());
        assert_eq!(versions.versions.len(), 1);
        assert!(versions.versions[0].deleted);
        assert!(!versions.versions[0].tier_free_version());
    }

    #[test]
    fn test_resolve_rebuilds_metadata_from_merged_versions() {
        let base_metadata = create_real_xlmeta().expect("base xl.meta");
        let base = FileMeta::load(&base_metadata).expect("load base xl.meta");

        let extra_version = FileMetaVersion {
            version_type: VersionType::Delete,
            object: None,
            delete_marker: Some(MetaDeleteMarker {
                version_id: Some(Uuid::from_u128(0x22222222333344445555666666666666)),
                mod_time: Some(OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp")),
                meta_sys: HashMap::new(),
            }),
            legacy_object: None,
            write_version: 99,
            uses_legacy_checksum: false,
        };

        let extra_shallow = FileMetaShallowVersion::try_from(extra_version).expect("build shallow delete version");

        let mut extended = base.clone();
        extended.versions.insert(0, extra_shallow);

        let base_versions = base.versions.len();
        let extended_versions = extended.versions.len();
        let extended_metadata = extended.marshal_msg().expect("serialize extended xl.meta");

        let resolved = MetaCacheEntries(vec![
            Some(MetaCacheEntry {
                name: "bucket/object".to_string(),
                metadata: extended_metadata,
                cached: Some(extended),
                reusable: false,
            }),
            Some(MetaCacheEntry {
                name: "bucket/object".to_string(),
                metadata: base_metadata,
                cached: Some(base),
                reusable: false,
            }),
        ])
        .resolve(MetadataResolutionParams {
            obj_quorum: 2,
            requested_versions: extended_versions,
            strict: true,
            ..Default::default()
        })
        .expect("merged entry should resolve");

        let cached = resolved.cached.expect("resolved entry should keep merged cached metadata");
        let decoded = FileMeta::load(&resolved.metadata).expect("resolved metadata should decode");

        assert_eq!(cached.versions.len(), base_versions);
        assert_eq!(decoded.versions.len(), base_versions);
        assert_eq!(decoded.versions, cached.versions);
        assert_ne!(extended_versions, cached.versions.len());
    }

    fn metacache_entry_with_mod_time(mod_time: OffsetDateTime, etag: &str) -> MetaCacheEntry {
        let mut metadata = HashMap::new();
        metadata.insert("etag".to_string(), etag.to_string());

        let mut meta = FileMeta::new();
        meta.add_version(FileInfo {
            volume: "bucket".to_string(),
            name: "object".to_string(),
            size: 1,
            mod_time: Some(mod_time),
            metadata,
            ..Default::default()
        })
        .expect("test file metadata should accept object version");
        let encoded = meta.marshal_msg().expect("test file metadata should marshal");

        MetaCacheEntry {
            name: "object".to_string(),
            metadata: encoded,
            cached: Some(meta),
            reusable: false,
        }
    }

    fn metacache_entry_with_erasure(
        mod_time: OffsetDateTime,
        etag: &str,
        data_blocks: usize,
        parity_blocks: usize,
    ) -> MetaCacheEntry {
        let mut metadata = HashMap::new();
        metadata.insert("etag".to_string(), etag.to_string());

        let mut fi = FileInfo::new("object", data_blocks, parity_blocks);
        fi.volume = "bucket".to_string();
        fi.name = "object".to_string();
        fi.size = 1;
        fi.mod_time = Some(mod_time);
        fi.metadata = metadata;

        let mut meta = FileMeta::new();
        meta.add_version(fi).expect("test file metadata should accept object version");
        let encoded = meta.marshal_msg().expect("test file metadata should marshal");

        MetaCacheEntry {
            name: "object".to_string(),
            metadata: encoded,
            cached: Some(meta),
            reusable: false,
        }
    }

    fn metacache_entry_with_erasure_versions(versions: &[(OffsetDateTime, &str, usize, usize)]) -> MetaCacheEntry {
        let mut meta = FileMeta::new();
        for (idx, (mod_time, etag, data_blocks, parity_blocks)) in versions.iter().enumerate() {
            let mut metadata = HashMap::new();
            metadata.insert("etag".to_string(), (*etag).to_string());

            let mut fi = FileInfo::new("object", *data_blocks, *parity_blocks);
            fi.volume = "bucket".to_string();
            fi.name = "object".to_string();
            let version_idx = u128::try_from(idx + 1).expect("test version index should fit u128");
            fi.version_id = Some(Uuid::from_u128(version_idx));
            fi.versioned = true;
            fi.size = 1;
            fi.mod_time = Some(*mod_time);
            fi.metadata = metadata;

            meta.add_version(fi).expect("test file metadata should accept object version");
        }
        let encoded = meta.marshal_msg().expect("test file metadata should marshal");

        MetaCacheEntry {
            name: "object".to_string(),
            metadata: encoded,
            cached: Some(meta),
            reusable: false,
        }
    }

    fn metacache_entry_without_header_ec(mut entry: MetaCacheEntry) -> MetaCacheEntry {
        let mut cached = entry.cached.take().expect("test entry should have cached metadata");
        for version in cached.versions.iter_mut() {
            version.header.ec_m = 0;
            version.header.ec_n = 0;
        }
        entry.metadata = cached.marshal_msg().expect("test file metadata should marshal");
        entry.cached = Some(cached);
        entry
    }

    fn metacache_dir_entry(name: &str) -> MetaCacheEntry {
        MetaCacheEntry {
            name: name.to_string(),
            ..Default::default()
        }
    }

    /// Build an entry holding a single object version with an explicit version id
    /// and mod_time, so a set of these can model DISJOINT per-disk version sets.
    fn metacache_entry_single_version(version_u128: u128, mod_time: OffsetDateTime, etag: &str) -> MetaCacheEntry {
        let mut metadata = HashMap::new();
        metadata.insert("etag".to_string(), etag.to_string());

        let mut fi = FileInfo::new("object", 4, 2);
        fi.volume = "bucket".to_string();
        fi.name = "object".to_string();
        fi.version_id = Some(Uuid::from_u128(version_u128));
        fi.versioned = true;
        fi.size = 1;
        fi.mod_time = Some(mod_time);
        fi.metadata = metadata;

        let mut meta = FileMeta::new();
        meta.add_version(fi).expect("test file metadata should accept object version");
        let encoded = meta.marshal_msg().expect("test file metadata should marshal");

        MetaCacheEntry {
            name: "object".to_string(),
            metadata: encoded,
            cached: Some(meta),
            reusable: false,
        }
    }

    /// backlog#920: `resolve_union` surfaces a version present on a SINGLE slot
    /// among four, while `resolve` at read-quorum (obj_quorum=2) drops it. This
    /// documents the exact sub-quorum durability gap the disk-walk closes.
    #[test]
    fn resolve_union_surfaces_single_disk_version() {
        let t0 = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
        let t1 = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");

        // Slot 0 carries a UNIQUE version (0xBEEF) present nowhere else; the other
        // three slots agree on a shared version (0xCAFE). Read-quorum sees only
        // the shared one; union must see BOTH.
        let unique = metacache_entry_single_version(0xBEEF, t1, "unique-etag");
        let shared_a = metacache_entry_single_version(0xCAFE, t0, "shared-etag");
        let shared_b = metacache_entry_single_version(0xCAFE, t0, "shared-etag");
        let shared_c = metacache_entry_single_version(0xCAFE, t0, "shared-etag");

        let entries = || {
            MetaCacheEntries(vec![
                Some(unique.clone()),
                Some(shared_a.clone()),
                Some(shared_b.clone()),
                Some(shared_c.clone()),
            ])
        };

        let union = entries().resolve_union("bucket").expect("union must resolve an entry");
        let union_meta = union.cached.expect("union entry keeps cached metadata");
        let union_ids: std::collections::HashSet<Option<Uuid>> =
            union_meta.versions.iter().map(|v| v.header.version_id).collect();
        assert!(
            union_ids.contains(&Some(Uuid::from_u128(0xBEEF))),
            "union must surface the single-slot version: {union_ids:?}"
        );
        assert!(
            union_ids.contains(&Some(Uuid::from_u128(0xCAFE))),
            "union must also keep the shared version: {union_ids:?}"
        );

        // Read-quorum resolution (obj_quorum=2) drops the single-slot version.
        let read_quorum = entries()
            .resolve(MetadataResolutionParams {
                obj_quorum: 2,
                dir_quorum: 2,
                strict: false,
                ..Default::default()
            })
            .expect("read-quorum must still resolve the shared version");
        let rq_meta = read_quorum.cached.expect("read-quorum entry keeps cached metadata");
        let rq_ids: std::collections::HashSet<Option<Uuid>> = rq_meta.versions.iter().map(|v| v.header.version_id).collect();
        assert!(
            !rq_ids.contains(&Some(Uuid::from_u128(0xBEEF))),
            "read-quorum must DROP the single-slot version (the gap): {rq_ids:?}"
        );
        assert!(
            rq_ids.contains(&Some(Uuid::from_u128(0xCAFE))),
            "read-quorum must keep the shared version: {rq_ids:?}"
        );
    }

    #[test]
    fn resolve_rejects_partial_latest_and_returns_committed_previous_metadata() {
        let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
        let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");
        let old_entry = metacache_entry_with_mod_time(old_mod_time, "old-etag");
        let new_entry = metacache_entry_with_mod_time(new_mod_time, "new-etag");

        let resolved = MetaCacheEntries(vec![
            Some(new_entry.clone()),
            Some(new_entry),
            Some(old_entry.clone()),
            Some(old_entry.clone()),
            Some(old_entry.clone()),
            Some(old_entry.clone()),
            Some(old_entry),
        ])
        .resolve(MetadataResolutionParams {
            obj_quorum: 5,
            requested_versions: 1,
            bucket: "bucket".to_string(),
            strict: true,
            ..Default::default()
        })
        .expect("previous committed metadata should still satisfy write quorum");

        let info = resolved
            .to_fileinfo("bucket")
            .expect("resolved committed metadata should decode as file info");
        assert_eq!(info.mod_time, Some(old_mod_time));
        assert_eq!(info.metadata.get("etag").map(String::as_str), Some("old-etag"));
    }

    #[test]
    fn resolve_rejects_low_parity_partial_latest_below_required_object_quorum() {
        let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
        let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");
        let old_entry = metacache_entry_with_erasure(old_mod_time, "old-etag", 7, 1);
        let new_entry = metacache_entry_with_erasure(new_mod_time, "new-etag", 7, 1);

        let resolved = MetaCacheEntries(vec![
            Some(new_entry.clone()),
            Some(new_entry.clone()),
            Some(new_entry.clone()),
            Some(new_entry.clone()),
            Some(new_entry),
            Some(old_entry.clone()),
            Some(old_entry),
        ])
        .resolve_with_write_quorum(MetadataResolutionParams {
            obj_quorum: 5,
            requested_versions: 1,
            bucket: "bucket".to_string(),
            strict: true,
            ..Default::default()
        });

        assert!(resolved.is_none());
    }

    #[test]
    fn resolve_rejects_zero_parity_partial_latest_below_required_object_quorum() {
        let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");
        let new_entry = metacache_entry_with_erasure(new_mod_time, "new-etag", 7, 0);

        let resolved = MetaCacheEntries(vec![
            Some(new_entry.clone()),
            Some(new_entry.clone()),
            Some(new_entry.clone()),
            Some(new_entry.clone()),
            Some(new_entry.clone()),
            Some(new_entry),
        ])
        .resolve_with_write_quorum(MetadataResolutionParams {
            obj_quorum: 5,
            requested_versions: 1,
            bucket: "bucket".to_string(),
            strict: true,
            ..Default::default()
        });

        assert!(resolved.is_none());
    }

    #[test]
    fn resolve_accepts_low_parity_latest_at_required_object_quorum() {
        let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");
        let new_entry = metacache_entry_with_erasure(new_mod_time, "new-etag", 7, 1);

        let resolved = MetaCacheEntries(vec![
            Some(new_entry.clone()),
            Some(new_entry.clone()),
            Some(new_entry.clone()),
            Some(new_entry.clone()),
            Some(new_entry.clone()),
            Some(new_entry.clone()),
            Some(new_entry),
        ])
        .resolve_with_write_quorum(MetadataResolutionParams {
            obj_quorum: 5,
            requested_versions: 1,
            bucket: "bucket".to_string(),
            strict: true,
            ..Default::default()
        })
        .expect("latest metadata should resolve after satisfying its own write quorum");

        let info = resolved
            .to_fileinfo("bucket")
            .expect("resolved committed metadata should decode as file info");
        assert_eq!(info.mod_time, Some(new_mod_time));
        assert_eq!(info.metadata.get("etag").map(String::as_str), Some("new-etag"));
    }

    #[test]
    fn resolve_skips_low_parity_partial_latest_and_returns_committed_previous_version() {
        let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
        let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");
        let old_entry = metacache_entry_with_erasure(old_mod_time, "old-etag", 4, 4);
        let new_and_old_entry =
            metacache_entry_with_erasure_versions(&[(old_mod_time, "old-etag", 4, 4), (new_mod_time, "new-etag", 7, 1)]);

        let resolved = MetaCacheEntries(vec![
            Some(new_and_old_entry.clone()),
            Some(new_and_old_entry.clone()),
            Some(new_and_old_entry.clone()),
            Some(new_and_old_entry.clone()),
            Some(new_and_old_entry),
            Some(old_entry.clone()),
            Some(old_entry),
        ])
        .resolve_with_write_quorum(MetadataResolutionParams {
            obj_quorum: 5,
            requested_versions: 1,
            bucket: "bucket".to_string(),
            strict: true,
            ..Default::default()
        })
        .expect("previous committed metadata should resolve after rejecting partial latest");

        let info = resolved
            .to_fileinfo("bucket")
            .expect("resolved committed metadata should decode as file info");
        assert_eq!(info.mod_time, Some(old_mod_time));
        assert_eq!(info.metadata.get("etag").map(String::as_str), Some("old-etag"));
    }

    #[test]
    fn resolve_skips_legacy_header_low_parity_partial_latest_using_payload_quorum() {
        let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
        let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp");
        let old_entry = metacache_entry_with_erasure(old_mod_time, "old-etag", 4, 4);
        let new_and_old_entry = metacache_entry_without_header_ec(metacache_entry_with_erasure_versions(&[
            (old_mod_time, "old-etag", 4, 4),
            (new_mod_time, "new-etag", 7, 1),
        ]));

        let resolved = MetaCacheEntries(vec![
            Some(new_and_old_entry.clone()),
            Some(new_and_old_entry.clone()),
            Some(new_and_old_entry.clone()),
            Some(new_and_old_entry.clone()),
            Some(new_and_old_entry),
            Some(old_entry.clone()),
            Some(old_entry),
        ])
        .resolve_with_write_quorum(MetadataResolutionParams {
            obj_quorum: 5,
            requested_versions: 1,
            bucket: "bucket".to_string(),
            strict: true,
            ..Default::default()
        })
        .expect("previous committed metadata should resolve after rejecting legacy-header partial latest");

        let info = resolved
            .to_fileinfo("bucket")
            .expect("resolved committed metadata should decode as file info");
        assert_eq!(info.mod_time, Some(old_mod_time));
        assert_eq!(info.metadata.get("etag").map(String::as_str), Some("old-etag"));
    }

    #[test]
    fn resolve_rejects_partial_directory_below_dir_quorum() {
        let partial_dir = metacache_dir_entry("prefix/");

        let resolved = MetaCacheEntries(vec![Some(partial_dir.clone()), Some(partial_dir)]).resolve(MetadataResolutionParams {
            dir_quorum: 5,
            obj_quorum: 5,
            bucket: "bucket".to_string(),
            strict: true,
            ..Default::default()
        });

        assert!(resolved.is_none());
    }

    fn build_hashmap_cache(update_size: usize) -> Arc<Cache<HashMap<usize, usize>>> {
        let generation = Arc::new(AtomicUsize::new(0));
        Arc::new(Cache::new(
            Box::new(move || {
                let generation = Arc::clone(&generation);
                Box::pin(async move {
                    let v = generation.fetch_add(1, Ordering::SeqCst);
                    let mut m = HashMap::with_capacity(update_size);
                    for i in 0..update_size {
                        m.insert(i, i ^ v);
                    }
                    Ok(m)
                })
            }),
            Duration::ZERO,
            Opts::default(),
        ))
    }

    async fn run_cache_workload(cache: Arc<Cache<HashMap<usize, usize>>>, workers: usize, rounds: usize, probe_mod: usize) {
        let mut tasks = Vec::with_capacity(workers);
        for worker in 0..workers {
            let cache = Arc::clone(&cache);
            tasks.push(tokio::spawn(async move {
                for round in 0..rounds {
                    let m = Arc::clone(&cache).get().await.expect("cache get should succeed");
                    let key = (worker.wrapping_mul(17).wrapping_add(round)) % probe_mod;
                    assert!(m.contains_key(&key), "expected key {key} to exist");
                }
            }));
        }

        for task in tasks {
            task.await.expect("worker task should not panic");
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
    async fn test_cache_concurrency_smoke() {
        let cache = build_hashmap_cache(2048);
        run_cache_workload(cache, 32, 120, 2048).await;
    }

    #[tokio::test]
    async fn test_cache_get_shared_reuses_fresh_value() {
        let calls = Arc::new(AtomicUsize::new(0));
        let cache = Arc::new(Cache::new(
            Box::new({
                let calls = Arc::clone(&calls);
                move || {
                    let calls = Arc::clone(&calls);
                    Box::pin(async move { Ok(calls.fetch_add(1, Ordering::SeqCst)) })
                }
            }),
            Duration::from_secs(60),
            Opts::default(),
        ));

        let first = Arc::clone(&cache).get_shared().await.expect("prime cache should succeed");
        let second = Arc::clone(&cache).get_shared().await.expect("fresh cache hit should succeed");

        assert!(Arc::ptr_eq(&first, &second));
        assert_eq!(*first, 0);
        assert_eq!(calls.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn test_cache_future_last_update_refreshes_instead_of_underflowing() {
        let calls = Arc::new(AtomicUsize::new(0));
        let cache = Arc::new(Cache::new(
            Box::new({
                let calls = Arc::clone(&calls);
                move || {
                    let calls = Arc::clone(&calls);
                    Box::pin(async move { Ok(calls.fetch_add(1, Ordering::SeqCst)) })
                }
            }),
            Duration::from_secs(60),
            Opts::default(),
        ));

        let prime = Arc::clone(&cache).get().await.expect("prime cache should succeed");
        assert_eq!(prime, 0);

        let now = Cache::<usize>::current_unix_secs();
        cache.last_update_secs.store(now.saturating_add(60), AtomicOrdering::SeqCst);

        let refreshed = Arc::clone(&cache)
            .get()
            .await
            .expect("future timestamp should force refresh instead of underflowing");
        assert_eq!(refreshed, 1);
        assert_eq!(calls.load(Ordering::SeqCst), 2);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_cache_no_wait_returns_stale_and_refreshes_in_background() {
        let calls = Arc::new(AtomicUsize::new(0));
        let (bg_started_tx, bg_started_rx) = oneshot::channel::<()>();
        let (release_bg_tx, release_bg_rx) = oneshot::channel::<()>();
        let bg_started_tx = Arc::new(StdMutex::new(Some(bg_started_tx)));
        let release_bg_rx = Arc::new(StdMutex::new(Some(release_bg_rx)));

        let cache = Arc::new(Cache::new(
            Box::new({
                let calls = Arc::clone(&calls);
                let bg_started_tx = Arc::clone(&bg_started_tx);
                let release_bg_rx = Arc::clone(&release_bg_rx);
                move || {
                    let calls = Arc::clone(&calls);
                    let bg_started_tx = Arc::clone(&bg_started_tx);
                    let release_bg_rx = Arc::clone(&release_bg_rx);
                    Box::pin(async move {
                        let call = calls.fetch_add(1, Ordering::SeqCst);
                        if call == 1 {
                            let tx = { bg_started_tx.lock().expect("start sender lock should not poison").take() };
                            if let Some(tx) = tx {
                                let _ = tx.send(());
                            }
                            let rx = { release_bg_rx.lock().expect("release receiver lock should not poison").take() };
                            if let Some(rx) = rx {
                                let _ = rx.await;
                            }
                        }
                        Ok(call)
                    })
                }
            }),
            Duration::from_secs(1),
            Opts {
                return_last_good: true,
                no_wait: true,
            },
        ));

        let prime = Arc::clone(&cache).get().await.expect("prime cache should succeed");
        assert_eq!(prime, 0);

        let now = Cache::<usize>::current_unix_secs();
        cache.last_update_secs.store(now.saturating_sub(1), AtomicOrdering::SeqCst);

        let stale = tokio::time::timeout(Duration::from_millis(200), Arc::clone(&cache).get())
            .await
            .expect("no_wait path should return without waiting for refresh")
            .expect("stale get should succeed");
        assert_eq!(stale, 0);

        tokio::time::timeout(Duration::from_millis(200), bg_started_rx)
            .await
            .expect("background refresh should start")
            .expect("background start signal should be delivered");

        release_bg_tx.send(()).expect("release signal should be delivered");

        tokio::time::timeout(Duration::from_secs(1), async {
            loop {
                if cache.cached_value().as_deref() == Some(&1) {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("background refresh should complete");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_cache_no_wait_coalesces_background_refreshes() {
        let calls = Arc::new(AtomicUsize::new(0));
        let release_refresh = Arc::new(Notify::new());
        let ttl = Duration::from_secs(60);
        let cache = Arc::new(Cache::new(
            Box::new({
                let calls = Arc::clone(&calls);
                let release_refresh = Arc::clone(&release_refresh);
                move || {
                    let calls = Arc::clone(&calls);
                    let release_refresh = Arc::clone(&release_refresh);
                    Box::pin(async move {
                        let call = calls.fetch_add(1, Ordering::SeqCst);
                        if call > 0 {
                            release_refresh.notified().await;
                        }
                        Ok(call)
                    })
                }
            }),
            ttl,
            Opts {
                return_last_good: true,
                no_wait: true,
            },
        ));

        let prime = Arc::clone(&cache).get().await.expect("prime cache should succeed");
        assert_eq!(prime, 0);

        let now = Cache::<usize>::current_unix_secs();
        cache
            .last_update_secs
            .store(now.saturating_sub(ttl.as_secs()), AtomicOrdering::SeqCst);

        let stale = Arc::clone(&cache).get().await.expect("stale get should succeed");
        assert_eq!(stale, 0);

        tokio::time::timeout(Duration::from_secs(1), async {
            loop {
                if calls.load(Ordering::SeqCst) == 2 {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("background refresh should start");

        let mut readers = Vec::new();
        for _ in 0..8 {
            let cache = Arc::clone(&cache);
            readers.push(tokio::spawn(
                async move { Arc::clone(&cache).get().await.expect("stale get should succeed") },
            ));
        }

        for reader in readers {
            assert_eq!(reader.await.expect("reader task should not panic"), 0);
        }

        tokio::time::sleep(Duration::from_millis(50)).await;
        assert_eq!(calls.load(Ordering::SeqCst), 2);

        release_refresh.notify_waiters();
    }

    #[tokio::test]
    async fn test_cache_return_last_good_on_refresh_error() {
        let calls = Arc::new(AtomicUsize::new(0));
        let cache = Arc::new(Cache::new(
            Box::new({
                let calls = Arc::clone(&calls);
                move || {
                    let calls = Arc::clone(&calls);
                    Box::pin(async move {
                        let call = calls.fetch_add(1, Ordering::SeqCst);
                        if call == 0 {
                            Ok(42usize)
                        } else {
                            Err(std::io::Error::other("refresh failed"))
                        }
                    })
                }
            }),
            Duration::from_secs(1),
            Opts {
                return_last_good: true,
                no_wait: false,
            },
        ));

        let prime = Arc::clone(&cache).get().await.expect("prime cache should succeed");
        assert_eq!(prime, 42);

        let now = Cache::<usize>::current_unix_secs();
        cache.last_update_secs.store(now.saturating_sub(2), AtomicOrdering::SeqCst);

        let stale = Arc::clone(&cache)
            .get()
            .await
            .expect("return_last_good should keep stale value");
        assert_eq!(stale, 42);
        assert_eq!(calls.load(Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn test_cache_refresh_error_without_return_last_good() {
        let calls = Arc::new(AtomicUsize::new(0));
        let cache = Arc::new(Cache::new(
            Box::new({
                let calls = Arc::clone(&calls);
                move || {
                    let calls = Arc::clone(&calls);
                    Box::pin(async move {
                        let call = calls.fetch_add(1, Ordering::SeqCst);
                        if call == 0 {
                            Ok(7usize)
                        } else {
                            Err(std::io::Error::other("refresh failed"))
                        }
                    })
                }
            }),
            Duration::from_secs(1),
            Opts {
                return_last_good: false,
                no_wait: false,
            },
        ));

        let prime = Arc::clone(&cache).get().await.expect("prime cache should succeed");
        assert_eq!(prime, 7);

        let now = Cache::<usize>::current_unix_secs();
        cache.last_update_secs.store(now.saturating_sub(2), AtomicOrdering::SeqCst);

        let err = Arc::clone(&cache)
            .get()
            .await
            .expect_err("refresh error should be propagated when return_last_good is false");
        assert_eq!(err.kind(), std::io::ErrorKind::Other);
        assert_eq!(calls.load(Ordering::SeqCst), 2);
    }

    // ------------------------------------------------------------------
    // backlog#900: metacache boundaries live outside the HTTP CatchPanicLayer
    // (background / listing chains). A corrupt-part xl.meta must yield Err,
    // not panic, so it cannot poison a worker.
    // ------------------------------------------------------------------

    fn corrupt_parts_filemeta() -> FileMeta {
        use crate::{ChecksumAlgo, ErasureAlgo, MetaObject, VersionType};
        let mut fm = FileMeta::new();
        fm.add_version_filemata(FileMetaVersion {
            version_type: VersionType::Object,
            object: Some(MetaObject {
                version_id: Some(Uuid::new_v4()),
                erasure_algorithm: ErasureAlgo::ReedSolomon,
                erasure_m: 2,
                erasure_n: 2,
                erasure_block_size: 1 << 20,
                bitrot_checksum_algo: ChecksumAlgo::HighwayHash,
                part_numbers: vec![1, 2],
                part_sizes: vec![10], // corrupt: shorter than part_numbers
                part_actual_sizes: vec![10, 20],
                mod_time: Some(time::OffsetDateTime::now_utc()),
                ..Default::default()
            }),
            ..Default::default()
        })
        .expect("add version");
        // Round-trip once so the corrupt array survives exactly like on disk.
        FileMeta::load(&fm.marshal_msg().expect("marshal")).expect("load")
    }

    #[test]
    fn metacache_to_fileinfo_returns_err_not_panic_on_corrupt_parts() {
        let entry = MetaCacheEntry {
            name: "obj".to_string(),
            metadata: Vec::new(),
            cached: Some(corrupt_parts_filemeta()),
            reusable: false,
        };
        let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| entry.to_fileinfo("bucket")));
        let inner = caught.expect("to_fileinfo must not panic");
        assert!(matches!(inner, Err(Error::FileCorrupt)), "expected FileCorrupt from metacache boundary");
    }

    #[test]
    fn metacache_file_info_versions_returns_err_not_panic_on_corrupt_parts() {
        // file_info_versions reads self.metadata (not cached), so fill marshaled bytes.
        let bytes = corrupt_parts_filemeta().marshal_msg().expect("marshal");
        let entry = MetaCacheEntry {
            name: "obj".to_string(),
            metadata: bytes,
            cached: None,
            reusable: false,
        };
        let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| entry.file_info_versions("bucket")));
        let inner = caught.expect("file_info_versions must not panic");
        assert!(matches!(inner, Err(Error::FileCorrupt)), "expected FileCorrupt");
    }
}