//! DiskLocation: manages volumes on a single disk/directory.
//!
//! Each DiskLocation represents one storage directory containing .dat + .idx files.
//! A Store contains one or more DiskLocations (one per configured directory).
//! Matches Go's storage/disk_location.go.

use std::collections::{HashMap, HashSet};
use std::fs;
use std::io;
use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

use tracing::{info, warn};

use crate::config::MinFreeSpace;
use crate::storage::erasure_coding::ec_bitrot::remove_bitrot_sidecars;
use crate::storage::erasure_coding::ec_shard::{
    DATA_SHARDS_COUNT, ERASURE_CODING_LARGE_BLOCK_SIZE, ERASURE_CODING_SMALL_BLOCK_SIZE,
    EcVolumeShard, ShardId,
};
use crate::storage::erasure_coding::ec_volume::{
    ECJ_COMPACT_TMP_EXT, EcVolume, is_usable_ecx_file,
};
use crate::storage::needle_map::NeedleMapKind;
use crate::storage::super_block::SUPER_BLOCK_SIZE;
use crate::storage::types::*;
use crate::storage::volume::{
    VifVolumeInfo, Volume, VolumeError, VolumeSpec, remove_volume_files, volume_file_name,
};

/// A single disk location managing volumes in one directory.
pub struct DiskLocation {
    pub directory: String,
    pub idx_directory: String,
    pub directory_uuid: String,
    pub disk_type: DiskType,
    pub tags: Vec<String>,
    pub max_volume_count: AtomicI32,
    pub original_max_volume_count: i32,
    volumes: HashMap<VolumeId, Arc<Volume>>,
    ec_volumes: HashMap<VolumeId, EcVolume>,
    pub is_disk_space_low: Arc<AtomicBool>,
    /// Go's `DiskLocation.readOnlyMount`: dir or idx_dir is on a read-only
    /// filesystem. Volumes load read-only and the location offers no free
    /// slots. Fixed at startup.
    pub read_only_mount: bool,
    pub available_space: AtomicU64,
    /// Headroom promised to in-flight copies that statfs cannot see yet.
    /// Probes keep it out of available_space so they cannot re-spend it; a
    /// copy that lands ends its charge, one that fails removes only its own.
    pub reserved_space: AtomicU64,
    // Physical filesystem capacity from the latest check_disk_space probe, reported
    // to the master so balancing can see real disk fullness, not just slot counts.
    pub disk_total_bytes: AtomicU64,
    pub disk_free_bytes: AtomicU64,
    pub min_free_space: MinFreeSpace,
    /// Go's `DiskLocation.loadGuard`: the server's copy-claim set. A scan
    /// skips a claimed vid outright and claims any other through its load,
    /// so a VolumeCopy's .note sweep cannot unlink files it still writes.
    pub(crate) load_guard: Option<Arc<Mutex<HashSet<VolumeId>>>>,
}

/// Keeps `vid` in the copy-claim set through one scan load so a VolumeCopy
/// arriving mid-load sees the claim and backs off — the release half of
/// Go's loadGuard hook.
struct LoadClaim {
    claims: Arc<Mutex<HashSet<VolumeId>>>,
    vid: VolumeId,
}

impl Drop for LoadClaim {
    fn drop(&mut self) {
        self.claims
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .remove(&self.vid);
    }
}

impl DiskLocation {
    const UUID_FILE_NAME: &'static str = "vol_dir.uuid";

    pub fn new(
        directory: &str,
        idx_directory: &str,
        max_volume_count: i32,
        disk_type: DiskType,
        min_free_space: MinFreeSpace,
        tags: Vec<String>,
    ) -> io::Result<Self> {
        fs::create_dir_all(directory)?;

        let idx_dir = if idx_directory.is_empty() {
            directory.to_string()
        } else {
            fs::create_dir_all(idx_directory)?;
            idx_directory.to_string()
        };
        let read_only_mount = is_read_only_mount(directory) || is_read_only_mount(&idx_dir);
        if read_only_mount {
            info!(
                "{} is on a read-only filesystem: its volumes load read-only, no new volumes",
                directory
            );
        }
        let directory_uuid = Self::generate_directory_uuid(directory)?;

        Ok(DiskLocation {
            directory: directory.to_string(),
            idx_directory: idx_dir,
            read_only_mount,
            directory_uuid,
            disk_type,
            tags,
            max_volume_count: AtomicI32::new(max_volume_count),
            original_max_volume_count: max_volume_count,
            volumes: HashMap::new(),
            ec_volumes: HashMap::new(),
            is_disk_space_low: Arc::new(AtomicBool::new(false)),
            available_space: AtomicU64::new(0),
            reserved_space: AtomicU64::new(0),
            disk_total_bytes: AtomicU64::new(0),
            disk_free_bytes: AtomicU64::new(0),
            min_free_space,
            load_guard: None,
        })
    }

    fn generate_directory_uuid(directory: &str) -> io::Result<String> {
        let path = std::path::Path::new(directory).join(Self::UUID_FILE_NAME);
        if path.exists() {
            let existing = fs::read_to_string(&path)?;
            if !existing.trim().is_empty() {
                return Ok(existing);
            }
        }

        let dir_uuid = uuid::Uuid::new_v4().to_string();
        match fs::write(&path, &dir_uuid) {
            Ok(()) => Ok(dir_uuid),
            // The uuid file cannot be written on a read-only mount; derive a
            // stable one from the directory plus its device and host so
            // restarts report the same id and independent disks at the same
            // path differ.
            Err(_) if is_read_only_mount(directory) => {
                use std::hash::{Hash, Hasher};
                let mut h = std::collections::hash_map::DefaultHasher::new();
                directory.hash(&mut h);
                if let Ok(meta) = fs::metadata(directory) {
                    #[cfg(unix)]
                    {
                        use std::os::unix::fs::MetadataExt;
                        meta.dev().hash(&mut h);
                    }
                    #[cfg(not(unix))]
                    {
                        let _ = meta;
                    }
                }
                if let Ok(host) = std::env::var("HOSTNAME") {
                    host.hash(&mut h);
                }
                let derived = (h.finish() as u128) << 64 | h.finish() as u128;
                Ok(uuid::Uuid::from_u128(derived).to_string())
            }
            Err(e) => Err(e),
        }
    }

    // ---- Volume management ----

    /// Go's loadGuard consult: `Ok(_)` lets the load proceed (holding the
    /// vid's claim through it when the gate is installed), `Err` means a
    /// VolumeCopy owns the vid and the scan skips it this pass.
    fn claim_for_load(&self, vid: VolumeId) -> Result<Option<LoadClaim>, ()> {
        match &self.load_guard {
            None => Ok(None),
            Some(claims) => {
                let mut set = claims.lock().unwrap_or_else(|e| e.into_inner());
                if set.contains(&vid) {
                    Err(())
                } else {
                    set.insert(vid);
                    Ok(Some(LoadClaim {
                        claims: claims.clone(),
                        vid,
                    }))
                }
            }
        }
    }

    /// Load existing volumes from the directory.
    ///
    /// Matches Go's `loadExistingVolume`: checks for incomplete volumes (.note file),
    /// validates EC shards before skipping .dat loading, and cleans up stale
    /// compaction temp files (.cpd/.cpx).
    pub fn load_existing_volumes(&mut self, needle_map_kind: NeedleMapKind) -> io::Result<()> {
        // Ensure directory exists
        fs::create_dir_all(&self.directory)?;
        if self.directory != self.idx_directory {
            fs::create_dir_all(&self.idx_directory)?;
        }

        // Recover any interrupted compaction commit before the volume scan. This
        // must run here, not inside the .dat loop: that loop is keyed on .dat
        // files and would miss the marker-only or already-renamed-.idx states a
        // mid-commit crash can leave behind.
        self.reconcile_compact_states();

        // Scan for .dat files
        let entries = fs::read_dir(&self.directory)?;
        let mut dat_files: Vec<(String, VolumeId)> = Vec::new();
        // Every collection claiming an id, in scan order; open_volumes keeps
        // the first that opens.
        let mut to_load: Vec<(VolumeId, Vec<String>)> = Vec::new();
        let mut queued: HashMap<VolumeId, usize> = HashMap::new();
        let mut seen = HashSet::new();

        for entry in entries {
            let entry = entry?;
            let name = entry.file_name().into_string().unwrap_or_default();
            if let Some((collection, vid)) = parse_volume_filename(&name)
                && seen.insert((collection.clone(), vid))
            {
                dat_files.push((collection, vid));
            }
        }

        // Claims for the queued volumes: held until open_volumes has read
        // their indexes, so a VolumeCopy cannot take the vid between this
        // sweep pass and the actual load — the span Go's loadGuard covers.
        let mut held_claims: Vec<LoadClaim> = Vec::new();

        for (collection, vid) in dat_files {
            let volume_name = volume_file_name(&self.directory, &collection, vid);
            let idx_name = volume_file_name(&self.idx_directory, &collection, vid);

            // A copy of this volume may be mid-flight: the sweeps below
            // would unlink files it is still writing, and a register here
            // could slip past its own recheck. Skip while a copy owns the
            // vid and claim it through the load — Go's loadGuard. A vid this
            // scan already queued is claimed by us: run this collection's
            // sweeps without re-claiming and merge it into the entry below.
            let load_claim = if queued.contains_key(&vid) {
                None
            } else {
                match self.claim_for_load(vid) {
                    Ok(claim) => claim,
                    Err(()) => continue,
                }
            };

            // Sweep a leftover empty `.dat` stub (a phantom from the pre-fix
            // loader) before it loads as a phantom volume or blocks startup.
            if remove_empty_ec_dat_stub(&volume_name, &idx_name, vid) {
                continue;
            }

            // If valid EC shards exist (.ecx file present), skip loading .dat
            let ecx_path = format!("{}.ecx", idx_name);
            let ecx_exists = if std::path::Path::new(&ecx_path).exists() {
                true
            } else if self.idx_directory != self.directory {
                // .ecx may have been created before -dir.idx was configured
                let fallback = format!("{}.ecx", volume_name);
                std::path::Path::new(&fallback).exists()
            } else {
                false
            };
            if ecx_exists {
                if self.validate_ec_volume(&collection, vid) {
                    // Valid EC volume — don't load .dat
                    continue;
                } else {
                    warn!(
                        volume_id = vid.0,
                        "EC volume validation failed, removing incomplete EC files"
                    );
                    let _ = self.remove_ec_volume_files(&collection, vid);
                    // Fall through to load .dat file
                }
            }

            // Stale .cpd/.cpx temp files were already rolled forward or back by
            // the reconcile_compact_states pre-pass above.

            // Check for an incomplete volume (.note means a VolumeCopy was
            // interrupted). This runs BELOW the empty-stub sweep and EC
            // validation: when an .ecx for this vid coexists on the disk, the
            // regular and EC volumes share <base>.vif, so removing the
            // incomplete regular copy must keep the .vif (keep_vif=true) or it
            // would strip the EC volume's info file.
            let note_path = format!("{}.note", volume_name);
            if std::path::Path::new(&note_path).exists() {
                let note = fs::read_to_string(&note_path).unwrap_or_default();
                warn!(
                    volume_id = vid.0,
                    "volume was not completed: {}, removing files", note
                );
                // Re-check .ecx now (not the pre-validation ecx_exists): the
                // invalid-EC cleanup above may have removed it, in which case
                // the .vif is no longer shared and must not be preserved.
                let keep_vif = std::path::Path::new(&ecx_path).exists()
                    || (self.idx_directory != self.directory
                        && std::path::Path::new(&format!("{}.ecx", volume_name)).exists());
                remove_volume_files(&volume_name, keep_vif);
                remove_volume_files(&idx_name, keep_vif);
                continue;
            }

            // Skip if already loaded (e.g., from a previous call)
            if self.volumes.contains_key(&vid) {
                continue;
            }

            // Load existing data only; never create a phantom `.dat`. A lone
            // `.vif`/`.idx` (e.g. an EC sidecar whose `.ecx` is on a sibling
            // disk) would otherwise have Volume::new write an 8-byte stub that
            // the sibling-.dat prune deletes real shards against. Remote-tiered
            // volumes also have no local `.dat`, but their `.vif` points at
            // remote files and must still load via the remote path.
            // Plain existence check (not check_dat_file_exists): an empty
            // .dat here is a fresh, needle-less volume that must still load.
            // A stat error other than NotFound counts as present so a
            // transient error doesn't skip a real volume.
            let dat_path = format!("{}.dat", volume_name);
            let dat_missing = matches!(fs::metadata(&dat_path),
                Err(ref e) if e.kind() == io::ErrorKind::NotFound);
            if dat_missing
                && !vif_references_remote_file(&format!("{}.vif", volume_name))
                && !vif_references_remote_file(&format!("{}.vif", idx_name))
            {
                continue;
            }

            match queued.get(&vid) {
                Some(&i) => to_load[i].1.push(collection),
                None => {
                    queued.insert(vid, to_load.len());
                    to_load.push((vid, vec![collection]));
                    held_claims.extend(load_claim);
                }
            }
        }

        for (collection, vid, v) in self.open_volumes(to_load, needle_map_kind) {
            let v = Arc::new(v);
            crate::metrics::VOLUME_GAUGE
                .with_label_values(&[&collection, "volume"])
                .inc();
            self.volumes.insert(vid, v);
        }

        // After regular volumes, auto-discover EC shards on disk so a
        // fresh restart picks up shards without an explicit
        // VolumeEcShardsMount RPC. Mirrors Go's loadExistingVolumes
        // calling loadAllEcShards. Failures here only log — they
        // shouldn't fail the whole disk's startup.
        if let Err(e) = self.load_all_ec_shards() {
            warn!(directory = %self.directory, error = %e, "load_all_ec_shards failed");
        }

        Ok(())
    }

    /// Open the volumes the directory scan selected. Opening one is dominated
    /// by reading its .idx into the needle map, so a disk holding thousands
    /// takes thousands of serial index reads to come up; mirrors Go's
    /// concurrentLoadingVolumes down to the max(cores, 10) worker count, whose
    /// floor keeps a small-core box off one-at-a-time on IO-bound work.
    ///
    /// An id is only spoken for once a volume actually loads, so a corrupt
    /// `colA_5.dat` still leaves `colB_5.dat` a chance.
    fn open_volumes(
        &self,
        to_load: Vec<(VolumeId, Vec<String>)>,
        needle_map_kind: NeedleMapKind,
    ) -> Vec<(String, VolumeId, Volume)> {
        if to_load.is_empty() {
            return Vec::new();
        }
        let workers = std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(1)
            .max(10)
            .min(to_load.len());

        let next = AtomicUsize::new(0);
        let opened = Mutex::new(Vec::with_capacity(to_load.len()));
        std::thread::scope(|scope| {
            for _ in 0..workers {
                scope.spawn(|| {
                    loop {
                        let i = next.fetch_add(1, Ordering::Relaxed);
                        let Some((vid, collections)) = to_load.get(i) else {
                            return;
                        };
                        for collection in collections {
                            // Replica placement and TTL are read back from the
                            // superblock, and a load never preallocates.
                            match Volume::new(
                                &self.directory,
                                &self.idx_directory,
                                *vid,
                                needle_map_kind,
                                &VolumeSpec {
                                    collection,
                                    ..Default::default()
                                },
                            ) {
                                Ok(mut v) => {
                                    v.location_disk_space_low = self.is_disk_space_low.clone();
                                    opened.lock().unwrap().push((collection.clone(), *vid, v));
                                    break;
                                }
                                Err(e) => {
                                    warn!(volume_id = vid.0, error = %e, "failed to load volume");
                                }
                            }
                        }
                    }
                });
            }
        });
        opened.into_inner().unwrap_or_else(|e| e.into_inner())
    }

    /// Directory pre-pass that recovers interrupted compaction commits. Collects
    /// every volume id that still has a .cpc commit marker or a leftover
    /// .cpd/.cpx temp file across the data and idx directories, then rolls the
    /// swap forward (marker present) or back (marker absent) per volume.
    fn reconcile_compact_states(&self) {
        let mut pending: HashSet<(String, VolumeId)> = HashSet::new();
        let mut collect = |dir: &str| {
            if let Ok(entries) = fs::read_dir(dir) {
                for entry in entries.flatten() {
                    let name = entry.file_name().into_string().unwrap_or_default();
                    let stem = name
                        .strip_suffix(".cpc")
                        .or_else(|| name.strip_suffix(".cpd"))
                        .or_else(|| name.strip_suffix(".cpx"));
                    if let Some(stem) = stem
                        && let Some(key) = parse_collection_volume_id(stem)
                    {
                        pending.insert(key);
                    }
                }
            }
        };
        collect(&self.directory);
        if self.idx_directory != self.directory {
            collect(&self.idx_directory);
        }

        for (collection, vid) in pending {
            // On a runtime reload (SIGHUP), an already-loaded volume may be
            // mid-vacuum: its .cpd/.cpx are live, not crash leftovers, and
            // rolling them back would clobber the in-flight compaction. Only
            // reconcile vids not currently loaded; genuine startup recovery
            // runs before any volume is loaded.
            if self.volumes.contains_key(&vid) {
                continue;
            }
            // A copy of this id can be mid-flight at the same basenames:
            // rolling a leftover commit forward would rename compaction
            // temps over the files it is writing. Same skip the load takes.
            let _load_claim = match self.claim_for_load(vid) {
                Ok(claim) => claim,
                Err(()) => continue,
            };
            let v = Volume::new_unloaded(&self.directory, &self.idx_directory, &collection, vid);
            if let Err(e) = v.reconcile_compact_state() {
                warn!(volume_id = vid.0, error = %e, "reconcile interrupted compaction failed");
            }
        }
    }

    /// Reports whether the EC files for (collection, vid) on this disk may be
    /// deleted to reclaim the local .dat. Returns false (delete) only when that
    /// provably loses no data; every ambiguity returns true (keep), since the
    /// shards may be the only copy. Mirrors Go's validateEcVolume.
    fn validate_ec_volume(&self, collection: &str, vid: VolumeId) -> bool {
        let base = volume_file_name(&self.directory, collection, vid);
        let dat_path = format!("{}.dat", base);

        // Custom ratio comes from the volume's own .vif; the server holds no
        // cluster EC config in memory.
        let data_shards =
            ec_data_shards_from_vif(&self.directory, &self.idx_directory, collection, vid);

        // On-disk .dat: an empty <= superblock one is a stub; a transient stat
        // error keeps the shards rather than deleting.
        let mut expected_shard_size: Option<i64> = None;
        let dat_exists = match fs::metadata(&dat_path) {
            Ok(meta) if meta.len() > SUPER_BLOCK_SIZE as u64 => {
                expected_shard_size = Some(calculate_expected_shard_size(
                    meta.len() as i64,
                    data_shards,
                ));
                true
            }
            Ok(_) => false,
            Err(e) if e.kind() == io::ErrorKind::NotFound => false,
            Err(e) => {
                warn!(volume_id = vid.0, error = %e, "cannot stat .dat; keeping EC shards");
                return true;
            }
        };

        let mut shard_count = 0usize;
        let mut actual_shard_size: Option<i64> = None;
        const MAX_SHARD_COUNT: usize = 32;

        for i in 0..MAX_SHARD_COUNT {
            let shard_path = format!("{}.ec{:02}", base, i);
            match fs::metadata(&shard_path) {
                Ok(meta) if meta.len() > 0 => {
                    let size = meta.len() as i64;
                    if let Some(prev) = actual_shard_size {
                        if size != prev {
                            // Inconsistent sizes signal corruption or mixed
                            // generations; not trusted for deletion -> keep.
                            warn!(
                                volume_id = vid.0,
                                shard = i,
                                size,
                                expected = prev,
                                "EC shard size mismatch; keeping shards"
                            );
                            return true;
                        }
                    } else {
                        actual_shard_size = Some(size);
                    }
                    shard_count += 1;
                }
                Err(e) if e.kind() != io::ErrorKind::NotFound => {
                    warn!(volume_id = vid.0, shard = i, error = %e, "cannot stat EC shard; keeping shards");
                    return true;
                }
                _ => {} // not found or zero size — skip
            }
        }

        if !dat_exists {
            return true; // distributed EC; any shard count is valid
        }

        // Reclaim only when it loses no data. Shards smaller than this .dat's
        // full encode are an interrupted encode whose .dat is the complete
        // source -> reclaim. Shards >= expected (valid/distributing EC, or a
        // stale/partial .dat beside larger real shards) may be the only copy -> keep.
        if shard_count == 0 {
            return false;
        }
        if let (Some(actual), Some(expected)) = (actual_shard_size, expected_shard_size)
            && actual < expected
        {
            warn!(
                volume_id = vid.0,
                actual,
                expected,
                "shards smaller than the .dat's full encode; reclaiming the complete .dat"
            );
            return false;
        }
        true
    }

    /// Remove all EC-related files for a volume.
    /// `pub(crate)` so the Store-level cross-disk passes in
    /// `store_ec_reconcile.rs` can call it to scrub partial EC artefacts
    /// when a healthy `.dat` for the same vid lives on a sibling disk
    /// (seaweedfs/seaweedfs#9478).
    pub(crate) fn remove_ec_volume_files(&self, collection: &str, vid: VolumeId) -> io::Result<()> {
        let base = volume_file_name(&self.directory, collection, vid);
        let idx_base = volume_file_name(&self.idx_directory, collection, vid);
        const MAX_SHARD_COUNT: usize = 32;

        // Remove index files from idx directory (.ecx, .ecj, and a compaction
        // tmp a crash may have left beside the .ecj)
        rm_if_present(format!("{}.ecx", idx_base))?;
        rm_if_present(format!("{}.ecj", idx_base))?;
        rm_if_present(format!("{}{}", idx_base, ECJ_COMPACT_TMP_EXT))?;
        // Also try data directory in case .ecx/.ecj were created before -dir.idx was configured
        if self.idx_directory != self.directory {
            rm_if_present(format!("{}.ecx", base))?;
            rm_if_present(format!("{}.ecj", base))?;
            rm_if_present(format!("{}{}", base, ECJ_COMPACT_TMP_EXT))?;
        }

        // Remove all EC shard files (.ec00 ~ .ec31)
        for i in 0..MAX_SHARD_COUNT {
            rm_if_present(format!("{}.ec{:02}", base, i))?;
        }

        // Remove the bitrot checksum sidecars from both the data and idx dirs.
        remove_bitrot_sidecars(&base)?;
        if self.idx_directory != self.directory {
            remove_bitrot_sidecars(&idx_base)?;
        }

        // Staged 2PC generations (<base>.ecNN.v<N>, versioned .ecx/.ecj/.vif)
        // belong to this volume's EC state too; leaving them orphans the files.
        crate::storage::erasure_coding::ec_shard::remove_ec_generation_files(&base, 0)?;
        if self.idx_directory != self.directory {
            crate::storage::erasure_coding::ec_shard::remove_ec_generation_files(&idx_base, 0)?;
        }
        Ok(())
    }

    /// Full-teardown variant: everything remove_ec_volume_files clears, PLUS the
    /// normal-volume <base>.vif on a SHARD-ONLY node. An interrupted shard copy
    /// (which installs shards + .ecx before .vif) could otherwise mount a fresh
    /// generation under the prior run's identity / ratio / dat_file_size carried by
    /// a stale .vif. Gated on .idx absence so a source-volume holder keeps its live
    /// .vif. Reconcile/load-fallback call remove_ec_volume_files directly and
    /// intentionally preserve it, mirroring Go's removeEcVolumeFiles (reconcile) vs
    /// removeStaleEcArtifacts (teardown) split.
    pub(crate) fn remove_ec_volume_files_full_teardown(
        &self,
        collection: &str,
        vid: VolumeId,
    ) -> io::Result<()> {
        self.remove_ec_volume_files(collection, vid)?;
        let base = volume_file_name(&self.directory, collection, vid);
        let idx_base = volume_file_name(&self.idx_directory, collection, vid);
        // try_exists, not exists(): a stat error on a PRESENT .idx must not read as
        // "shard-only" and delete a live source volume's .vif. Only Ok(false) (genuine
        // NotFound) clears it; on a stat error stay conservative and keep the sidecar.
        if matches!(
            std::path::Path::new(&format!("{}.idx", idx_base)).try_exists(),
            Ok(false)
        ) {
            rm_if_present(format!("{}.vif", idx_base))?;
            if self.idx_directory != self.directory
                && matches!(
                    std::path::Path::new(&format!("{}.idx", base)).try_exists(),
                    Ok(false)
                )
            {
                rm_if_present(format!("{}.vif", base))?;
            }
        }
        Ok(())
    }

    /// EC encode generation recorded in this disk's .vif (data dir first, then idx
    /// dir for the split-disk layout). None if no .vif was readable, so the fenced
    /// teardown preserves it (fail-safe). A present .vif with no ec_shard_config
    /// yields Some(0) (a recovered or pre-upgrade live volume), also preserved.
    pub(crate) fn ec_generation_ts_ns(&self, collection: &str, vid: VolumeId) -> Option<i64> {
        for dir in [&self.directory, &self.idx_directory] {
            let vif = format!("{}.vif", volume_file_name(dir, collection, vid));
            if let Ok(s) = fs::read_to_string(&vif)
                && let Ok(vi) = serde_json::from_str::<VifVolumeInfo>(&s)
            {
                return Some(vi.ec_shard_config.map(|c| c.encode_ts_ns).unwrap_or(0));
            }
            if self.directory == self.idx_directory {
                break;
            }
        }
        None
    }

    /// Find a volume by ID. Returns a shared handle like Go's findVolume:
    /// the lookup borrows the map briefly and I/O runs on the handle after.
    pub fn find_volume(&self, vid: VolumeId) -> Option<Arc<Volume>> {
        self.volumes.get(&vid).cloned()
    }

    /// Add a volume to this location.
    pub fn set_volume(&mut self, vid: VolumeId, volume: Volume) {
        let collection = volume.collection.clone();
        if let Some(old) = self.volumes.insert(vid, Arc::new(volume)) {
            // A held handle would keep writing the same files against a stale
            // in-memory index; close it so the stale handle fails closed.
            crate::metrics::VOLUME_GAUGE
                .with_label_values(&[&old.collection, "volume"])
                .dec();
            old.close();
        }
        crate::metrics::VOLUME_GAUGE
            .with_label_values(&[&collection, "volume"])
            .inc();
    }

    /// Create a new volume in this location.
    pub fn create_volume(
        &mut self,
        vid: VolumeId,
        needle_map_kind: NeedleMapKind,
        spec: &VolumeSpec<'_>,
    ) -> Result<(), VolumeError> {
        let mut v = Volume::new(
            &self.directory,
            &self.idx_directory,
            vid,
            needle_map_kind,
            spec,
        )?;
        v.location_disk_space_low = self.is_disk_space_low.clone();
        crate::metrics::VOLUME_GAUGE
            .with_label_values(&[spec.collection, "volume"])
            .inc();
        if let Some(old) = self.volumes.insert(vid, Arc::new(v)) {
            // Same fail-closed rule as set_volume: a stale handle must not
            // keep writing the files the replacement just reopened.
            crate::metrics::VOLUME_GAUGE
                .with_label_values(&[&old.collection, "volume"])
                .dec();
            old.close();
        }
        Ok(())
    }

    /// Remove and close a volume.
    pub fn unload_volume(&mut self, vid: VolumeId) -> Option<Arc<Volume>> {
        if let Some(v) = self.volumes.remove(&vid) {
            crate::metrics::VOLUME_GAUGE
                .with_label_values(&[&v.collection, "volume"])
                .dec();
            v.close();
            Some(v)
        } else {
            None
        }
    }

    /// Remove, close, and delete all files for a volume. When keep_remote_data
    /// is true the cloud-tier object backing the volume is left intact — used
    /// by moves where another server is taking over the same .vif.
    pub fn delete_volume(
        &mut self,
        vid: VolumeId,
        only_empty: bool,
        only_garbage: bool,
        keep_remote_data: bool,
    ) -> Result<(), VolumeError> {
        // Refuse before removing: a refused destroy must leave it mounted.
        if let Some(v) = self.volumes.get(&vid)
            && v.is_compacting()
        {
            return Err(v.compacting_error());
        }
        if let Some(v) = self.volumes.remove(&vid) {
            match v.destroy(only_empty, only_garbage, keep_remote_data) {
                Ok(()) => {
                    crate::metrics::VOLUME_GAUGE
                        .with_label_values(&[&v.collection, "volume"])
                        .dec();
                    Ok(())
                }
                Err(e) => {
                    // A failed destroy left the local files in place, so the
                    // replica stays registered rather than becoming a phantom.
                    self.volumes.insert(vid, v);
                    Err(e)
                }
            }
        } else {
            Err(VolumeError::NotFound)
        }
    }

    /// Delete all volumes in a collection.
    pub fn delete_collection(&mut self, collection: &str) -> Result<(), VolumeError> {
        let vids: Vec<VolumeId> = self
            .volumes
            .iter()
            .filter(|(_, v)| v.collection == collection && !v.is_compacting())
            .map(|(vid, _)| *vid)
            .collect();

        for vid in vids {
            if let Some(v) = self.volumes.remove(&vid) {
                if let Err(e) = v.destroy(false, false, false) {
                    warn!(volume_id = vid.0, error = %e, "delete collection: failed to destroy volume");
                    self.volumes.insert(vid, v);
                } else {
                    crate::metrics::VOLUME_GAUGE
                        .with_label_values(&[&v.collection, "volume"])
                        .dec();
                }
            }
        }

        let ec_vids: Vec<VolumeId> = self
            .ec_volumes
            .iter()
            .filter(|(_, v)| v.collection == collection)
            .map(|(vid, _)| *vid)
            .collect();

        for vid in ec_vids {
            if let Some(mut ec_vol) = self.ec_volumes.remove(&vid) {
                for _ in 0..ec_vol.shard_count() {
                    crate::metrics::VOLUME_GAUGE
                        .with_label_values(&[collection, "ec_shards"])
                        .dec();
                }
                ec_vol.destroy();
            }
        }
        Ok(())
    }

    // ---- Metrics ----

    /// Number of volumes on this disk.
    pub fn volumes_len(&self) -> usize {
        self.volumes.len()
    }

    /// Get all volume IDs, sorted.
    pub fn volume_ids(&self) -> Vec<VolumeId> {
        let mut ids: Vec<VolumeId> = self.volumes.keys().copied().collect();
        ids.sort();
        ids
    }

    /// Iterate over all volumes.
    pub fn iter_volumes(&self) -> impl Iterator<Item = (&VolumeId, &Volume)> {
        self.volumes.iter().map(|(k, v)| (k, v.as_ref()))
    }

    /// Number of free volume slots.
    /// Matches Go's FindFreeLocation formula:
    ///   free = ((MaxVolumeCount - VolumesLen()) * DataShardsCount - EcShardCount()) / DataShardsCount
    pub fn free_volume_count(&self) -> i32 {
        if self.read_only_mount {
            return 0;
        }
        use crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT;
        let max = self.max_volume_count.load(Ordering::Relaxed);
        let free_count = (max as i64 - self.volumes.len() as i64) * DATA_SHARDS_COUNT as i64
            - self.ec_shard_count() as i64;
        let effective_free = free_count / DATA_SHARDS_COUNT as i64;
        if effective_free > 0 {
            effective_free as i32
        } else {
            0
        }
    }

    /// Iterate over all volumes.
    pub fn volumes(&self) -> impl Iterator<Item = (&VolumeId, &Volume)> {
        self.volumes.iter().map(|(k, v)| (k, v.as_ref()))
    }

    /// Sum of unused space in writable volumes (volumeSizeLimit - actual size per volume).
    /// Used by auto-max-volume-count to estimate how many more volumes can fit.
    pub fn unused_space(&self, volume_size_limit: u64) -> u64 {
        let mut unused: u64 = 0;
        for vol in self.volumes.values() {
            if vol.is_read_only() {
                continue;
            }
            // A preallocated volume already holds its space on disk.
            let dat_size = vol
                .dat_file_size()
                .unwrap_or(0)
                .max(vol.dat_allocated_size());
            let idx_size = vol.idx_file_size();
            let used = dat_size + idx_size;
            if volume_size_limit > used {
                unused += volume_size_limit - used;
            }
        }
        unused
    }

    /// Check disk space against min_free_space and update is_disk_space_low.
    pub fn check_disk_space(&self) {
        let (total, free) = get_disk_stats(&self.directory);
        if total == 0 {
            return;
        }
        let used = total.saturating_sub(free);
        let is_low = self.min_free_space.is_low(free, total);
        self.is_disk_space_low.store(is_low, Ordering::Relaxed);
        // Go's DiskLocation.AvailableSpace counts only free space above the
        // -volume.minFreeSpace floor, so space checks stay above it too.
        self.available_space.store(
            self.min_free_space
                .available_space(free, total)
                .saturating_sub(self.reserved_space.load(Ordering::Relaxed)),
            Ordering::Relaxed,
        );
        self.disk_total_bytes.store(total, Ordering::Relaxed);
        self.disk_free_bytes.store(free, Ordering::Relaxed);

        // Update resource gauges
        crate::metrics::RESOURCE_GAUGE
            .with_label_values(&[&self.directory, "all"])
            .set(total as f64);
        crate::metrics::RESOURCE_GAUGE
            .with_label_values(&[&self.directory, "used"])
            .set(used as f64);
        crate::metrics::RESOURCE_GAUGE
            .with_label_values(&[&self.directory, "free"])
            .set(free as f64);
        // "avail" is same as "free" for us (Go subtracts reserved blocks but we use statvfs f_bavail)
        crate::metrics::RESOURCE_GAUGE
            .with_label_values(&[&self.directory, "avail"])
            .set(free as f64);
    }

    // ---- EC volume operations ----

    /// Find an EC volume by ID.
    pub fn find_ec_volume(&self, vid: VolumeId) -> Option<&EcVolume> {
        self.ec_volumes.get(&vid)
    }

    /// Find an EC volume by ID (mutable).
    pub fn find_ec_volume_mut(&mut self, vid: VolumeId) -> Option<&mut EcVolume> {
        self.ec_volumes.get_mut(&vid)
    }

    /// Check if this location has an EC volume.
    pub fn has_ec_volume(&self, vid: VolumeId) -> bool {
        self.ec_volumes.contains_key(&vid)
    }

    /// Reports whether this disk has a sealed `.ecx` index file for the
    /// given (collection, vid). Unlike [`Self::has_ec_volume`] this does
    /// not require the EC volume to be mounted in memory, which makes it
    /// the right primitive for placement decisions during `ec.balance` /
    /// `ec.rebuild` flows where shards may arrive before any
    /// `VolumeEcShardsMount` has happened on the receiving disk. Without
    /// checking the on-disk state, auto-select can split shards from the
    /// `.ecx` that travels with the first shard, which is the source of
    /// the orphan-shard layout reported in seaweedfs/seaweedfs#9212.
    ///
    /// Mirrors `DiskLocation.HasEcxFileOnDisk` in
    /// `weed/storage/disk_location_ec.go`. Skips entries that are
    /// directories so a stray dir named `<collection>_<vid>.ecx` doesn't
    /// register as a present index file. A 0-byte `.ecx` is a corrupt stub
    /// left by a failed EC distribute copy; it must not steer placement
    /// toward this disk, so it counts as absent (Go requires `Size() > 0`).
    pub fn has_ecx_file_on_disk(&self, collection: &str, vid: VolumeId) -> bool {
        let idx_base = volume_file_name(&self.idx_directory, collection, vid);
        if is_usable_ecx_file(&format!("{}.ecx", idx_base)) {
            return true;
        }
        if self.idx_directory != self.directory {
            let data_base = volume_file_name(&self.directory, collection, vid);
            if is_usable_ecx_file(&format!("{}.ecx", data_base)) {
                return true;
            }
        }
        false
    }

    /// Remove an EC volume, returning it.
    pub fn remove_ec_volume(&mut self, vid: VolumeId) -> Option<EcVolume> {
        self.ec_volumes.remove(&vid)
    }

    /// Drop the in-memory EC volume for vid and close its descriptors without
    /// deleting files, so a following unlink frees the inodes instead of
    /// leaving open fds serving the old bytes. Mirrors Go's unloadEcVolume.
    pub fn unload_ec_volume(&mut self, vid: VolumeId) {
        if let Some(mut ec_vol) = self.ec_volumes.remove(&vid) {
            for _ in 0..ec_vol.shard_count() {
                crate::metrics::VOLUME_GAUGE
                    .with_label_values(&[&ec_vol.collection, "ec_shards"])
                    .dec();
            }
            ec_vol.close();
        }
    }

    /// Mount EC shards for a volume on this location.
    ///
    /// `source_disk_type` is the source volume's disk type carried on the
    /// `VolumeEcShardsMount` RPC. When non-empty it overrides the in-memory
    /// EC volume's reported disk type so heartbeats keep reporting under
    /// the source's disk type after EC encoding (#9423). Empty means "use
    /// this location's disk type" — used by disk-scan reload paths that
    /// have no orchestrator context.
    pub fn mount_ec_shards(
        &mut self,
        vid: VolumeId,
        collection: &str,
        shard_ids: &[ShardId],
        source_disk_type: &str,
    ) -> Result<(), VolumeError> {
        let idx_dir = self.idx_directory.clone();
        self.mount_ec_shards_with_idx_dir(vid, collection, shard_ids, &idx_dir, source_disk_type)
    }

    /// Mount EC shards but explicitly specify the idx directory the
    /// EcVolume should pull `.ecx` / `.ecj` / `.vif` from.
    ///
    /// Cross-disk reconcile (mirrors `loadEcShardsWithIdxDir` in
    /// `weed/storage/disk_location_ec.go`): when a volume's shards live
    /// on this disk but the index files were left on a sibling disk
    /// (the seaweedfs/seaweedfs#9212 orphan-shard layout), the
    /// reconciler creates the EcVolume here while pointing it at the
    /// sibling's idx dir. Each shard still ends up registered in this
    /// disk's `ec_volumes` map so heartbeat reporting carries the right
    /// disk_id per shard.
    pub fn mount_ec_shards_with_idx_dir(
        &mut self,
        vid: VolumeId,
        collection: &str,
        shard_ids: &[ShardId],
        idx_dir: &str,
        source_disk_type: &str,
    ) -> Result<(), VolumeError> {
        let dir = self.directory.clone();
        // Avoid the entry().or_insert_with() pattern here: that closure
        // can't return a Result, so any EcVolume::new failure (e.g.
        // .ecx open error, .ecj create error, malformed .vif) would
        // have to panic via unwrap(). Build the EcVolume up front and
        // propagate the error to the caller.
        let created = !self.ec_volumes.contains_key(&vid);
        if created {
            let ec_vol = EcVolume::new(&dir, idx_dir, collection, vid).map_err(VolumeError::Io)?;
            self.ec_volumes.insert(vid, ec_vol);
        }
        let ec_vol = self.ec_volumes.get_mut(&vid).expect("just inserted above");
        // When the orchestrator supplied a source disk type on the Mount
        // RPC, override the EC volume's disk type so heartbeats report
        // under the source volume's disk type (#9423). When the caller
        // passed "" (disk-scan reload, orphan-shard reconcile, restart)
        // we only default to this location's disk type for a fresh EC
        // volume; otherwise we leave whatever a prior RPC mount set in
        // place, so empty-source reloads don't reintroduce the drift.
        if !source_disk_type.is_empty() {
            ec_vol.set_disk_type(DiskType::from_string(source_disk_type));
        } else if ec_vol.shard_count() == 0 {
            ec_vol.set_disk_type(self.disk_type.clone());
        }

        for &shard_id in shard_ids {
            // A mount retry re-listing a shard this volume already holds:
            // keep the existing registration (mirrors Go's AddEcVolumeShard
            // added=false) — re-adding would replace a serving fd and bump
            // the ec_shards gauge without growing the mounted count.
            if ec_vol.has_shard(shard_id) {
                continue;
            }
            let mut shard = EcVolumeShard::new(&dir, collection, vid, shard_id);
            shard.disk_type = ec_vol.disk_type.clone();
            if let Err(e) = ec_vol.add_shard(shard) {
                // The shard was dropped (its descriptors closed) inside the
                // failed add. If this call just created the EcVolume and it
                // holds nothing, remove it too — a zero-shard registration
                // would advertise a mount that serves no data while pinning
                // its descriptors.
                let now_empty = ec_vol.shard_count() == 0;
                if created && now_empty {
                    self.ec_volumes.remove(&vid);
                }
                return Err(VolumeError::Io(e));
            }
            crate::metrics::VOLUME_GAUGE
                .with_label_values(&[collection, "ec_shards"])
                .inc();
        }
        Ok(())
    }

    /// Swap the mounted shard's descriptor for a fresh open of its on-disk
    /// file, keeping the registration — and therefore the master's view —
    /// untouched. Mirrors Go's `DiskLocation.ReopenEcShard`: the rebuild
    /// recreates a shard file under a handle that still serves the detached
    /// inode, and reopening repoints reads at the repaired file without an
    /// unmount/remount heartbeat flip-flop. Readers holding a cloned fd keep
    /// the old inode until they finish — the same contract eviction relies on.
    pub fn reopen_ec_shard(&mut self, vid: VolumeId, shard_id: ShardId) -> Result<bool, VolumeError> {
        let dir = self.directory.clone();
        let Some(ec_vol) = self.ec_volumes.get_mut(&vid) else {
            return Ok(false);
        };
        let Some(existing) = ec_vol
            .shards
            .get_mut(shard_id as usize)
            .and_then(|s| s.as_mut())
        else {
            return Ok(false);
        };
        let mut new_shard = EcVolumeShard::new(&dir, &existing.collection, vid, shard_id);
        new_shard.disk_type = existing.disk_type.clone();
        new_shard.open().map_err(VolumeError::Io)?;
        *existing = new_shard;
        Ok(true)
    }

    /// Unmount EC shards for a volume on this location.
    ///
    /// Only shards that are actually mounted decrement the per-shard
    /// metric — without this guard the gauge would underflow when the
    /// caller passes a shard that lives on a sibling disk
    /// (cross-disk reconcile makes that the common case for the same
    /// `vid` after reconciliation).
    pub fn unmount_ec_shards(&mut self, vid: VolumeId, shard_ids: &[ShardId]) {
        if let Some(ec_vol) = self.ec_volumes.get_mut(&vid) {
            let collection = ec_vol.collection.clone();
            for &shard_id in shard_ids {
                if !ec_vol.has_shard(shard_id) {
                    continue;
                }
                let _ = ec_vol.remove_shard(shard_id);
                crate::metrics::VOLUME_GAUGE
                    .with_label_values(&[&collection, "ec_shards"])
                    .dec();
            }
            if ec_vol.shard_count() == 0 {
                let mut vol = self.ec_volumes.remove(&vid).unwrap();
                vol.close();
            }
        }
    }

    /// Auto-discover .ec?? shard files on disk and mount them.
    ///
    /// Mirrors `loadAllEcShards` in `weed/storage/disk_location_ec.go`.
    /// Reads the data dir (and idx dir, if separate), groups shards by
    /// (collection, vid) using sorted iteration so a volume's shards
    /// land contiguously next to their `.ecx`. When the matching `.ecx`
    /// is encountered, validates and mounts the group.
    ///
    /// Without this, an EC shard on disk is invisible after a Rust
    /// volume-server restart until something explicitly issues
    /// VolumeEcShardsMount — strict superset of seaweedfs/seaweedfs#9212.
    pub fn load_all_ec_shards(&mut self) -> io::Result<()> {
        // Use a HashSet during accumulation so a name appearing in
        // both data dir and idx dir (idempotent legacy layouts can
        // place .ecx in either) is processed exactly once. Without
        // dedup, mount_ec_shards' per-shard metric increment would
        // double-count for those filenames.
        let mut seen: HashSet<String> = HashSet::new();
        let mut entries: Vec<String> = Vec::new();
        // Keep only the shard and index files this scan acts on: a disk of
        // regular volumes has millions of .dat/.idx/.vif names that would
        // otherwise each cost a String here and a slot in the sort below.
        let mut collect = |dir: &str| -> io::Result<()> {
            for ent in fs::read_dir(dir)? {
                let ent = ent?;
                if ent.file_type().map(|ft| ft.is_dir()).unwrap_or(false) {
                    continue;
                }
                let name = ent.file_name().to_string_lossy().into_owned();
                let Some(dot) = name.rfind('.') else {
                    continue;
                };
                let ext = &name[dot..];
                if parse_ec_shard_extension(ext).is_none() && ext != ".ecx" {
                    continue;
                }
                if seen.insert(name.clone()) {
                    entries.push(name);
                }
            }
            Ok(())
        };
        collect(&self.directory)?;
        if self.idx_directory != self.directory {
            collect(&self.idx_directory)?;
        }
        entries.sort();

        let mut same_volume_shards: Vec<(String, ShardId)> = Vec::new(); // (filename, shard_id)
        let mut prev_vid: Option<VolumeId> = None;
        let mut prev_collection: String = String::new();

        for name in entries {
            let Some(dot) = name.rfind('.') else {
                continue;
            };
            let (base, ext) = name.split_at(dot); // ext includes leading '.'
            let Some((collection, vid)) = parse_collection_volume_id(base) else {
                continue;
            };

            if let Some(shard_id) = parse_ec_shard_extension(ext) {
                // Ignore zero-byte shards — same as Go.
                let path = format!("{}/{}", self.directory, name);
                let size = fs::metadata(&path).map(|m| m.len()).unwrap_or(0);
                if size == 0 {
                    continue;
                }

                let same_group = match prev_vid {
                    None => true,
                    Some(p) => p == vid && prev_collection == collection,
                };
                if same_group {
                    same_volume_shards.push((name, shard_id));
                } else {
                    self.check_orphaned_shards(
                        &same_volume_shards,
                        &prev_collection,
                        prev_vid.unwrap_or(VolumeId(0)),
                    );
                    same_volume_shards = vec![(name, shard_id)];
                }
                prev_vid = Some(vid);
                prev_collection = collection;
                continue;
            }

            if ext == ".ecx"
                && prev_vid == Some(vid)
                && prev_collection == collection
                && !same_volume_shards.is_empty()
            {
                let shards = std::mem::take(&mut same_volume_shards);
                self.handle_found_ecx_file(&shards, &collection, vid);
                prev_vid = None;
                prev_collection.clear();
            }
        }

        // Handle any trailing group that never saw a matching .ecx.
        self.check_orphaned_shards(
            &same_volume_shards,
            &prev_collection,
            prev_vid.unwrap_or(VolumeId(0)),
        );

        Ok(())
    }

    /// Validate + mount a (collection, vid) group when its `.ecx` is
    /// found. Mirrors `handleFoundEcxFile` in
    /// `weed/storage/disk_location_ec.go`.
    fn handle_found_ecx_file(
        &mut self,
        shards: &[(String, ShardId)],
        collection: &str,
        vid: VolumeId,
    ) {
        let base = volume_file_name(&self.directory, collection, vid);
        let dat_path = format!("{}.dat", base);
        let dat_exists = check_dat_file_exists(&dat_path);

        if dat_exists && !self.validate_ec_volume(collection, vid) {
            warn!(
                volume_id = vid.0,
                "Incomplete or invalid EC volume: .dat exists but validation failed, cleaning up EC files",
            );
            let _ = self.remove_ec_volume_files(collection, vid);
            return;
        }

        let shard_ids: Vec<ShardId> = shards.iter().map(|(_, sid)| *sid).collect();
        if let Err(e) = self.mount_ec_shards(vid, collection, &shard_ids, "") {
            // A mount failure (corrupt/locked .ecx, EMFILE, transient I/O) is
            // not proof the shards are disposable -- validate_ec_volume already
            // decided they may be the only copy. Release partially-mounted
            // shards (mirror-decrements the metric) but keep the files; never
            // delete on a load error.
            warn!(
                volume_id = vid.0,
                "Failed to load EC shards: {}; keeping files for retry", e,
            );
            self.unmount_ec_shards(vid, &shard_ids);
        }
    }

    /// Mirrors `checkOrphanedShards`: if shards exist on disk without a
    /// matching `.ecx` AND the legacy `.dat` is still present, treat
    /// the encoding as interrupted before `.ecx` was written and clean
    /// up. If `.dat` is gone, leave files alone — they may be
    /// distributed-EC shards waiting for cross-disk reconciliation.
    fn check_orphaned_shards(
        &self,
        shards: &[(String, ShardId)],
        collection: &str,
        vid: VolumeId,
    ) -> bool {
        if shards.is_empty() || vid.0 == 0 {
            return false;
        }
        let base = volume_file_name(&self.directory, collection, vid);
        let dat_path = format!("{}.dat", base);
        if check_dat_file_exists(&dat_path) {
            warn!(
                volume_id = vid.0,
                "Found {} EC shards without .ecx file (incomplete encoding interrupted before .ecx), cleaning up",
                shards.len(),
            );
            let _ = self.remove_ec_volume_files(collection, vid);
            return true;
        }
        false
    }

    /// Total number of EC shards on this location.
    pub fn ec_shard_count(&self) -> usize {
        self.ec_volumes
            .values()
            .map(|ecv| ecv.shards.iter().filter(|s| s.is_some()).count())
            .sum()
    }

    /// Iterate over all EC volumes.
    pub fn ec_volumes(&self) -> impl Iterator<Item = (&VolumeId, &EcVolume)> {
        self.ec_volumes.iter()
    }

    /// Close all volumes.
    pub fn close(&mut self) {
        for v in self.volumes.values_mut() {
            v.close();
        }
        self.volumes.clear();
        for (_, mut ec_vol) in self.ec_volumes.drain() {
            ec_vol.close();
        }
    }
}

/// Mirror of Go's `util.IsReadOnlyMount`: access(2) with W_OK answers EROFS
/// for a filesystem mounted read-only regardless of the permission bits, and
/// also for root, unlike a permission-bits check.
#[cfg(unix)]
pub(crate) fn is_read_only_mount(path: &str) -> bool {
    use std::ffi::CString;
    let c_path = match CString::new(path.as_bytes()) {
        Ok(p) => p,
        Err(_) => return false,
    };
    if unsafe { libc::access(c_path.as_ptr(), libc::W_OK) } == 0 {
        return false;
    }
    io::Error::last_os_error().raw_os_error() == Some(libc::EROFS)
}

#[cfg(not(unix))]
pub(crate) fn is_read_only_mount(_path: &str) -> bool {
    false
}

/// Get total and free disk space for a given path.
/// Returns (total_bytes, free_bytes).
pub fn get_disk_stats(path: &str) -> (u64, u64) {
    #[cfg(unix)]
    {
        use std::ffi::CString;
        let c_path = match CString::new(path) {
            Ok(p) => p,
            Err(_) => return (0, 0),
        };
        // SAFETY: `libc::statvfs` is plain data — integers and reserved
        // padding, no pointers and no restricted niches — so the all-zero
        // value is a valid one for the call to overwrite.
        let mut stat: libc::statvfs = unsafe { std::mem::zeroed() };
        // SAFETY: `c_path` is a live NUL-terminated `CString` that outlives
        // the call, and `&mut stat` is a live, aligned, exclusive pointer the
        // kernel only writes through; the fields are read below only after
        // the call reports success.
        if unsafe { libc::statvfs(c_path.as_ptr(), &mut stat) } != 0 {
            return (0, 0);
        }
        let all = stat.f_blocks as u64 * stat.f_frsize as u64;
        let free = stat.f_bavail as u64 * stat.f_frsize as u64;
        (all, free)
    }
    #[cfg(windows)]
    {
        use std::os::windows::ffi::OsStrExt;

        // Canonicalize so symlinks, `.`/`..` segments, and relative paths
        // resolve to the real location before querying. `\\?\`-prefixed
        // extended-length paths and UNC (`\\?\UNC\...`) are passed through
        // untouched: GetDiskFreeSpaceExW accepts them as-is.
        let canonical = match std::fs::canonicalize(path) {
            Ok(p) => p,
            Err(_) => return (0, 0),
        };
        // UTF-16 with trailing NUL for the Win32 wide-string call.
        let mut wide: Vec<u16> = canonical.as_os_str().encode_wide().collect();
        // UNC directory names must end in a backslash for GetDiskFreeSpaceExW.
        if !wide.ends_with(&[0x5C]) {
            wide.push(0x5C);
        }
        wide.push(0);
        // SAFETY: `wide` is NUL-terminated; the out-params are valid u64
        // writes; the call has no other preconditions.
        unsafe {
            let mut free_available: u64 = 0;
            let mut total: u64 = 0;
            let ok = windows_sys::Win32::Storage::FileSystem::GetDiskFreeSpaceExW(
                wide.as_ptr(),
                &mut free_available,
                &mut total,
                std::ptr::null_mut(),
            );
            if ok == 0 {
                return (0, 0);
            }
            return (total, free_available);
        }
    }
    #[cfg(not(any(unix, windows)))]
    {
        compile_error!("get_disk_stats is implemented for unix and windows only");
    }
}

/// Calculate expected EC shard size from .dat file size.
/// Matches Go's `calculateExpectedShardSize`: large blocks (1GB * data_shards) first,
/// then small blocks (1MB * data_shards) for the remainder.
fn calculate_expected_shard_size(dat_file_size: i64, data_shards: usize) -> i64 {
    let large_batch_size = ERASURE_CODING_LARGE_BLOCK_SIZE as i64 * data_shards as i64;
    let num_large_batches = dat_file_size / large_batch_size;
    let mut shard_size = num_large_batches * ERASURE_CODING_LARGE_BLOCK_SIZE as i64;
    let remaining = dat_file_size - (num_large_batches * large_batch_size);

    if remaining > 0 {
        let small_batch_size = ERASURE_CODING_SMALL_BLOCK_SIZE as i64 * data_shards as i64;
        // Ceiling division
        let num_small_batches = (remaining + small_batch_size - 1) / small_batch_size;
        shard_size += num_small_batches * ERASURE_CODING_SMALL_BLOCK_SIZE as i64;
    }

    shard_size
}

/// Resolve the EC data-shard count from the volume's own `.vif` (the volume
/// server never holds the cluster EC config in memory), checking the data dir
/// then the idx dir. Falls back to the default ratio when no EC `.vif` is found.
/// Remove a file, treating a missing file as success but propagating real errors
/// (permissions, disk full). Mirrors Go's removeFileIfExists so a teardown failure
/// surfaces instead of silently leaving stale artifacts.
fn rm_if_present(path: String) -> io::Result<()> {
    match fs::remove_file(&path) {
        Err(e) if e.kind() != io::ErrorKind::NotFound => Err(e),
        _ => Ok(()),
    }
}

fn ec_data_shards_from_vif(
    directory: &str,
    idx_directory: &str,
    collection: &str,
    vid: VolumeId,
) -> usize {
    for dir in [directory, idx_directory] {
        let vif = format!("{}.vif", volume_file_name(dir, collection, vid));
        if let Some(ds) = fs::read_to_string(&vif)
            .ok()
            .and_then(|s| serde_json::from_str::<VifVolumeInfo>(&s).ok())
            .and_then(|vi| vi.ec_shard_config)
            .map(|c| c.data_shards as usize)
            && ds > 0
        {
            return ds;
        }
        if directory == idx_directory {
            break;
        }
    }
    DATA_SHARDS_COUNT
}

/// Parse a volume filename like "collection_42.dat" or "42.dat" into (collection, VolumeId).
/// Parse a `<collection>_<vid>` or `<vid>` base name into its parts.
/// Mirrors `parseCollectionVolumeId` in
/// `weed/storage/disk_location.go`. Used when iterating raw filenames
/// where the extension has already been stripped.
///
/// `pub(crate)` companion `parse_collection_volume_id_pub` is exposed
/// so the cross-disk reconcile in `store_ec_reconcile.rs` can call it
/// without re-implementing the parser.
pub(crate) fn parse_collection_volume_id_pub(base: &str) -> Option<(String, VolumeId)> {
    parse_collection_volume_id(base)
}

fn parse_collection_volume_id(base: &str) -> Option<(String, VolumeId)> {
    if let Some(pos) = base.rfind('_') {
        let collection = &base[..pos];
        let id_str = &base[pos + 1..];
        let id: u32 = id_str.parse().ok()?;
        Some((collection.to_string(), VolumeId(id)))
    } else {
        let id: u32 = base.parse().ok()?;
        Some((String::new(), VolumeId(id)))
    }
}

/// `pub(crate)` re-export of [`parse_ec_shard_extension`] for the
/// cross-disk reconcile in `store_ec_reconcile.rs`.
pub(crate) fn is_ec_shard_extension(ext: &str) -> Option<ShardId> {
    parse_ec_shard_extension(ext)
}

/// Recognise EC shard extensions `.ec00`–`.ec255` (the ShardId u8
/// range — see `EcVolumeShard`'s typed shard id) and return the
/// shard id. Returns `None` for any other extension.
///
/// Matches the regex `\.ec\d{2,3}` from the Go side, with the same
/// per-shard `id <= 255` clamp Go's loader applies via
/// `strconv.ParseInt(... 10, 64)` followed by the `if shardId < 0 ||
/// shardId > 255` guard. The 3-digit form (`.ec100`–`.ec255`) is
/// retained so the parser can still recognise shards from custom
/// 32+ ratios that fit in a u8 even though OSS only ships 10+4.
fn parse_ec_shard_extension(ext: &str) -> Option<ShardId> {
    let rest = ext.strip_prefix(".ec")?;
    if rest.len() < 2 || rest.len() > 3 {
        return None;
    }
    let id: u32 = rest.parse().ok()?;
    if id > 255 {
        return None;
    }
    ShardId::try_from(id).ok()
}

/// Robust check that a `.dat` with actual data exists. An empty `.dat`
/// (<= a superblock, zero needles) is a leftover stub, not an encode source,
/// and counts as absent so it never justifies deleting shards. Any unexpected
/// stat error (permission, I/O) is treated as "exists" so we don't misclassify
/// local EC as distributed EC. Mirrors `checkDatFileExists` in Go.
fn check_dat_file_exists(path: &str) -> bool {
    match fs::metadata(path) {
        Ok(meta) => meta.len() > SUPER_BLOCK_SIZE as u64,
        Err(e) if e.kind() == io::ErrorKind::NotFound => false,
        Err(_) => true,
    }
}

/// True when a `.vif` references remote-tier files: a remote-only volume
/// that has no local `.dat` but must still load via the remote path,
/// rather than be skipped as a lone EC sidecar.
pub(crate) fn vif_references_remote_file(vif_path: &str) -> bool {
    fs::read_to_string(vif_path)
        .ok()
        .and_then(|s| serde_json::from_str::<VifVolumeInfo>(&s).ok())
        .map(|vif| !vif.files.is_empty())
        .unwrap_or(false)
}

/// True when a `.vif` records an EC shard config, i.e. the volume was
/// erasure-coded. Such a volume keeps no local `.dat`, so an empty `.dat`
/// alongside it is a leftover stub safe to remove.
fn vif_is_ec_volume(vif_path: &str) -> bool {
    fs::read_to_string(vif_path)
        .ok()
        .and_then(|s| serde_json::from_str::<VifVolumeInfo>(&s).ok())
        .map(|vif| vif.ec_shard_config.is_some())
        .unwrap_or(false)
}

/// Remove a leftover empty EC `.dat` stub and return whether one was swept.
///
/// A stub is an empty `.dat` (<= a superblock, i.e. zero needles) whose `.vif`
/// records an EC shard config. An EC volume keeps no local `.dat`, so the stub
/// holds no data — its shards live on other servers. Such stubs (phantoms from
/// the pre-fix loader) otherwise load as phantom empty volumes, and a same-vid
/// stub on two disks blocks startup via the duplicate-vid check. The `.dat` and
/// its empty `.idx` are removed; non-EC empty `.dat` files are left alone. The
/// `.vif` is looked up in both the data and idx directories (which differ only
/// when `-dir.idx` is configured), each read at most once.
fn remove_empty_ec_dat_stub(volume_name: &str, idx_name: &str, vid: VolumeId) -> bool {
    let dat_path = format!("{}.dat", volume_name);
    match fs::metadata(&dat_path) {
        Ok(meta) if meta.len() <= SUPER_BLOCK_SIZE as u64 => {}
        _ => return false,
    }

    let data_vif = format!("{}.vif", volume_name);
    let idx_vif = format!("{}.vif", idx_name);
    let is_ec = vif_is_ec_volume(&data_vif) || (idx_vif != data_vif && vif_is_ec_volume(&idx_vif));
    if !is_ec {
        return false;
    }

    warn!(
        volume_id = vid.0,
        "removing leftover empty .dat stub for EC volume"
    );
    let _ = fs::remove_file(&dat_path);
    let _ = fs::remove_file(format!("{}.idx", idx_name));
    true
}

fn parse_volume_filename(filename: &str) -> Option<(String, VolumeId)> {
    let stem = filename
        .strip_suffix(".dat")
        .or_else(|| filename.strip_suffix(".vif"))
        .or_else(|| filename.strip_suffix(".idx"))?;
    parse_collection_volume_id(stem)
}

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    /// get_disk_stats must report real capacity for a real path on every
    /// platform (Windows included) — consumers treat total==0 as "unknown"
    /// and leave available_space at 0, which breaks volume assignment.
    #[test]
    fn test_get_disk_stats_reports_capacity_for_real_path() {
        let tmp = TempDir::new().unwrap();
        let (total, free) = get_disk_stats(tmp.path().to_str().unwrap());
        assert!(total > 0, "expected total>0, got {total}");
        assert!(free > 0, "expected free>0, got {free}");
    }

    // Mirrors Go's DiskLocation.AvailableSpace: the cached figure a copy is
    // checked against counts only free space above the min-free-space floor.
    #[test]
    fn test_check_disk_space_counts_only_space_above_floor() {
        const FLOOR: u64 = 1 << 20;
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let loc = DiskLocation::new(
            dir,
            dir,
            0,
            DiskType::HardDrive,
            MinFreeSpace::Bytes(FLOOR),
            Vec::new(),
        )
        .unwrap();
        loc.check_disk_space();
        assert_eq!(
            loc.available_space.load(Ordering::Relaxed),
            loc.disk_free_bytes
                .load(Ordering::Relaxed)
                .saturating_sub(FLOOR)
        );

        let loc = DiskLocation::new(
            dir,
            dir,
            0,
            DiskType::HardDrive,
            MinFreeSpace::Bytes(u64::MAX),
            Vec::new(),
        )
        .unwrap();
        loc.check_disk_space();
        assert_eq!(loc.available_space.load(Ordering::Relaxed), 0);
    }

    // Mirrors Go's CheckDiskSpace pending subtraction: an in-flight copy's
    // charge stays out of the cache across a probe, so another copy cannot
    // re-spend the headroom the live reading still shows as free.
    #[test]
    fn test_check_disk_space_keeps_pending_reservation_out_of_cache() {
        const FLOOR: u64 = 1 << 20;
        const RESERVED: u64 = 512 * 1024;
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let loc = DiskLocation::new(
            dir,
            dir,
            0,
            DiskType::HardDrive,
            MinFreeSpace::Bytes(FLOOR),
            Vec::new(),
        )
        .unwrap();
        loc.reserved_space.store(RESERVED, Ordering::Relaxed);
        loc.check_disk_space();
        let expected = loc
            .min_free_space
            .available_space(
                loc.disk_free_bytes.load(Ordering::Relaxed),
                loc.disk_total_bytes.load(Ordering::Relaxed),
            )
            .saturating_sub(RESERVED);
        assert_eq!(loc.available_space.load(Ordering::Relaxed), expected);
    }

    /// When `-dir.idx` is configured the EC `.vif` may live in the idx
    /// directory; the sweep must look there too, not only the data dir.
    #[test]
    fn test_remove_empty_ec_dat_stub_finds_vif_in_idx_dir() {
        let tmp = TempDir::new().unwrap();
        let data = tmp.path().join("data");
        let idx = tmp.path().join("idx");
        std::fs::create_dir_all(&data).unwrap();
        std::fs::create_dir_all(&idx).unwrap();
        let vbase = format!("{}/warp-cal_42", data.to_str().unwrap());
        let ibase = format!("{}/warp-cal_42", idx.to_str().unwrap());
        std::fs::write(format!("{}.dat", vbase), vec![0u8; 8]).unwrap();

        let vif = VifVolumeInfo {
            version: 3,
            ec_shard_config: Some(crate::storage::volume::VifEcShardConfig {
                data_shards: 10,
                parity_shards: 4,
                ..Default::default()
            }),
            ..Default::default()
        };
        std::fs::write(
            format!("{}.vif", ibase),
            serde_json::to_string(&vif).unwrap(),
        )
        .unwrap();

        assert!(
            remove_empty_ec_dat_stub(&vbase, &ibase, VolumeId(42)),
            "EC .vif in the idx dir should be found and the stub removed",
        );
        assert!(!std::path::Path::new(&format!("{}.dat", vbase)).exists());
    }

    // The headline geometry: a stale/partial .dat (e.g. an interrupted decode)
    // smaller than the volume's real source sits next to the full-size shards
    // that are the only copy. validate_ec_volume must keep the shards.
    #[test]
    fn test_validate_ec_volume_partial_dat_next_to_full_shards_keeps() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();
        let base = volume_file_name(dir, "", VolumeId(70));
        let ds = crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT;
        let full = calculate_expected_shard_size(30 * 1024 * 1024, ds);
        for i in 0..ds {
            std::fs::File::create(format!("{}.ec{:02}", base, i))
                .unwrap()
                .set_len(full as u64)
                .unwrap();
        }
        // Partial .dat: bigger than a superblock so it is not swept as a stub,
        // but smaller than what these shards encode.
        std::fs::File::create(format!("{}.dat", base))
            .unwrap()
            .set_len(5 * 1024 * 1024)
            .unwrap();
        assert!(
            loc.validate_ec_volume("", VolumeId(70)),
            "full-size shards beside a smaller (stale/partial) .dat must be kept",
        );
    }

    // The legitimate cleanup: a full source .dat next to shards SMALLER than it
    // would encode (an interrupted encode). The .dat is the complete source, so
    // reclaiming it loses no data.
    #[test]
    fn test_validate_ec_volume_interrupted_encode_reclaims() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();
        let base = volume_file_name(dir, "", VolumeId(71));
        let ds = crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT;
        let dat_size = 30 * 1024 * 1024i64;
        std::fs::File::create(format!("{}.dat", base))
            .unwrap()
            .set_len(dat_size as u64)
            .unwrap();
        let partial = calculate_expected_shard_size(dat_size, ds) / 3;
        assert!(partial > 0);
        for i in 0..ds {
            std::fs::File::create(format!("{}.ec{:02}", base, i))
                .unwrap()
                .set_len(partial as u64)
                .unwrap();
        }
        assert!(
            !loc.validate_ec_volume("", VolumeId(71)),
            "shards smaller than the full source .dat should be reclaimable",
        );
    }

    // The per-disk generation reader behind the fenced EC teardown: a present .vif
    // yields its generation (or 0 with no EC config), a missing .vif is unreadable
    // (preserved, fail-safe), and the idx dir is a fallback for the split-disk layout.
    #[test]
    fn test_ec_generation_ts_ns_reads_and_fences() {
        let tmp = TempDir::new().unwrap();
        let data = tmp.path().join("data");
        let idx = tmp.path().join("idx");
        std::fs::create_dir_all(&data).unwrap();
        std::fs::create_dir_all(&idx).unwrap();
        let loc = DiskLocation::new(
            data.to_str().unwrap(),
            idx.to_str().unwrap(),
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();
        let vid = VolumeId(88);
        let dbase = volume_file_name(data.to_str().unwrap(), "", vid);
        let ibase = volume_file_name(idx.to_str().unwrap(), "", vid);

        // No .vif anywhere -> unreadable (the fenced teardown preserves on None).
        assert_eq!(loc.ec_generation_ts_ns("", vid), None);

        // .vif in the data dir carries its generation.
        let with_gen = VifVolumeInfo {
            version: 3,
            ec_shard_config: Some(crate::storage::volume::VifEcShardConfig {
                data_shards: 10,
                parity_shards: 4,
                encode_ts_ns: 4242,
                ..Default::default()
            }),
            ..Default::default()
        };
        std::fs::write(
            format!("{}.vif", dbase),
            serde_json::to_string(&with_gen).unwrap(),
        )
        .unwrap();
        assert_eq!(loc.ec_generation_ts_ns("", vid), Some(4242));

        // A .vif with no EC config reads as generation 0 (recovered/pre-upgrade live volume).
        std::fs::remove_file(format!("{}.vif", dbase)).unwrap();
        let no_cfg = VifVolumeInfo {
            version: 3,
            ..Default::default()
        };
        std::fs::write(
            format!("{}.vif", dbase),
            serde_json::to_string(&no_cfg).unwrap(),
        )
        .unwrap();
        assert_eq!(loc.ec_generation_ts_ns("", vid), Some(0));

        // idx-dir fallback: only the idx dir holds the .vif.
        std::fs::remove_file(format!("{}.vif", dbase)).unwrap();
        std::fs::write(
            format!("{}.vif", ibase),
            serde_json::to_string(&with_gen).unwrap(),
        )
        .unwrap();
        assert_eq!(loc.ec_generation_ts_ns("", vid), Some(4242));
    }

    #[test]
    fn test_parse_volume_filename() {
        assert_eq!(
            parse_volume_filename("42.dat"),
            Some(("".to_string(), VolumeId(42)))
        );
        assert_eq!(
            parse_volume_filename("pics_7.dat"),
            Some(("pics".to_string(), VolumeId(7)))
        );
        assert_eq!(
            parse_volume_filename("42.vif"),
            Some(("".to_string(), VolumeId(42)))
        );
        assert_eq!(
            parse_volume_filename("pics_7.idx"),
            Some(("pics".to_string(), VolumeId(7)))
        );
        assert_eq!(parse_volume_filename("notadat.idx"), None);
        assert_eq!(parse_volume_filename("bad.dat"), None);
    }

    // A volume mid-copy must not be swept or loaded by a reload: a claimed
    // vid is skipped outright — .note files and all — and an unclaimed vid
    // is claimed through its load, released when the pass ends. Mirrors
    // Go's TestLoadExistingVolumeSkipsACopyInFlight.
    #[test]
    fn test_load_existing_volumes_skips_a_copy_in_flight() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        {
            let mut loc = DiskLocation::new(
                dir,
                dir,
                10,
                DiskType::HardDrive,
                MinFreeSpace::Percent(1.0),
                Vec::new(),
            )
            .unwrap();
            loc.create_volume(VolumeId(9), NeedleMapKind::InMemory, &VolumeSpec::default())
                .unwrap();
            loc.close();
        }
        // The copy's in-progress marker: unclaimed, the sweep unlinks the files.
        fs::write(format!("{dir}/9.note"), "copying from test").unwrap();

        let claims = Arc::new(Mutex::new(HashSet::new()));
        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();
        loc.load_guard = Some(claims.clone());

        claims.lock().unwrap().insert(VolumeId(9));
        loc.load_existing_volumes(NeedleMapKind::InMemory).unwrap();
        assert!(loc.find_volume(VolumeId(9)).is_none());
        assert!(
            std::path::Path::new(&format!("{dir}/9.dat")).exists(),
            "the sweep must not unlink files a VolumeCopy still owns"
        );
        assert!(std::path::Path::new(&format!("{dir}/9.note")).exists());

        claims.lock().unwrap().remove(&VolumeId(9));
        fs::remove_file(format!("{dir}/9.note")).unwrap();
        loc.load_existing_volumes(NeedleMapKind::InMemory).unwrap();
        assert!(loc.find_volume(VolumeId(9)).is_some());
        assert!(
            claims.lock().unwrap().is_empty(),
            "the load's claim must release when the pass ends"
        );
    }

    #[test]
    fn test_disk_location_create_volume() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();

        loc.create_volume(VolumeId(1), NeedleMapKind::InMemory, &VolumeSpec::default())
            .unwrap();

        assert_eq!(loc.volumes_len(), 1);
        assert!(loc.find_volume(VolumeId(1)).is_some());
        assert!(loc.find_volume(VolumeId(99)).is_none());
        assert_eq!(loc.free_volume_count(), 9);
    }

    #[test]
    fn test_disk_location_load_existing() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();

        // Create volumes
        {
            let mut loc = DiskLocation::new(
                dir,
                dir,
                10,
                DiskType::HardDrive,
                MinFreeSpace::Percent(1.0),
                Vec::new(),
            )
            .unwrap();
            loc.create_volume(VolumeId(1), NeedleMapKind::InMemory, &VolumeSpec::default())
                .unwrap();
            loc.create_volume(
                VolumeId(2),
                NeedleMapKind::InMemory,
                &VolumeSpec {
                    collection: "test",
                    ..Default::default()
                },
            )
            .unwrap();
            loc.close();
        }

        // Reload
        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();
        loc.load_existing_volumes(NeedleMapKind::InMemory).unwrap();
        assert_eq!(loc.volumes_len(), 2);

        let ids = loc.volume_ids();
        assert!(ids.contains(&VolumeId(1)));
        assert!(ids.contains(&VolumeId(2)));
    }

    // Two collections can name the same volume id on one disk; a candidate
    // that fails to open must not shadow a good one behind it.
    #[test]
    fn test_open_volumes_falls_back_past_a_corrupt_candidate() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();

        {
            let mut loc = DiskLocation::new(
                dir,
                dir,
                10,
                DiskType::HardDrive,
                MinFreeSpace::Percent(1.0),
                Vec::new(),
            )
            .unwrap();
            loc.create_volume(
                VolumeId(9),
                NeedleMapKind::InMemory,
                &VolumeSpec {
                    collection: "good",
                    ..Default::default()
                },
            )
            .unwrap();
            loc.close();
        }

        // Same id under another collection, unopenable.
        let mut bad = vec![0u8; 16];
        bad[0] = 9; // unsupported version
        std::fs::write(format!("{}/bad_9.dat", dir), &bad).unwrap();

        let loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();
        let opened = loc.open_volumes(
            vec![(VolumeId(9), vec!["bad".to_string(), "good".to_string()])],
            NeedleMapKind::InMemory,
        );

        assert_eq!(opened.len(), 1, "the good candidate should still open");
        assert_eq!(opened[0].0, "good");
        assert_eq!(opened[0].1, VolumeId(9));
    }

    #[test]
    fn test_disk_location_delete_volume() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();

        loc.create_volume(VolumeId(1), NeedleMapKind::InMemory, &VolumeSpec::default())
            .unwrap();
        loc.create_volume(VolumeId(2), NeedleMapKind::InMemory, &VolumeSpec::default())
            .unwrap();
        assert_eq!(loc.volumes_len(), 2);

        loc.delete_volume(VolumeId(1), false, false, false).unwrap();
        assert_eq!(loc.volumes_len(), 1);
        assert!(loc.find_volume(VolumeId(1)).is_none());
    }

    #[test]
    fn test_disk_location_delete_collection() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();

        loc.create_volume(
            VolumeId(1),
            NeedleMapKind::InMemory,
            &VolumeSpec {
                collection: "pics",
                ..Default::default()
            },
        )
        .unwrap();
        loc.create_volume(
            VolumeId(2),
            NeedleMapKind::InMemory,
            &VolumeSpec {
                collection: "pics",
                ..Default::default()
            },
        )
        .unwrap();
        loc.create_volume(
            VolumeId(3),
            NeedleMapKind::InMemory,
            &VolumeSpec {
                collection: "docs",
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(loc.volumes_len(), 3);

        loc.delete_collection("pics").unwrap();
        assert_eq!(loc.volumes_len(), 1);
        assert!(loc.find_volume(VolumeId(3)).is_some());
    }

    /// A 0-byte `.ecx` is the stub a failed EC distribute copy leaves behind.
    /// Go's HasEcxFileOnDisk requires Size() > 0 so the stub cannot pin
    /// placement to a disk that has no usable index.
    #[test]
    fn test_has_ecx_file_on_disk_ignores_zero_byte_stub() {
        let tmp = TempDir::new().unwrap();
        let data = tmp.path().join("data");
        let idx = tmp.path().join("idx");
        fs::create_dir_all(&data).unwrap();
        fs::create_dir_all(&idx).unwrap();
        let loc = DiskLocation::new(
            data.to_str().unwrap(),
            idx.to_str().unwrap(),
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();

        fs::write(idx.join("pics_7.ecx"), b"").unwrap();
        assert!(!loc.has_ecx_file_on_disk("pics", VolumeId(7)));

        // A real index in the data dir still counts, stub or no stub.
        fs::write(data.join("pics_7.ecx"), [0u8; 16]).unwrap();
        assert!(loc.has_ecx_file_on_disk("pics", VolumeId(7)));
    }

    #[test]
    fn test_disk_location_delete_collection_removes_ec_volumes() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();

        let shard_path = format!("{}/pics_7.ec00", dir);
        std::fs::write(&shard_path, b"ec-shard").unwrap();
        // An EC volume needs its .ecx to mount.
        std::fs::write(format!("{}/pics_7.ecx", dir), [0u8; 16]).unwrap();

        loc.mount_ec_shards(VolumeId(7), "pics", &[0], "").unwrap();
        assert!(loc.has_ec_volume(VolumeId(7)));
        assert!(std::path::Path::new(&shard_path).exists());
        assert!(std::path::Path::new(&format!("{}/pics_7.ecj", dir)).exists());

        loc.delete_collection("pics").unwrap();

        assert!(!loc.has_ec_volume(VolumeId(7)));
        assert!(!std::path::Path::new(&shard_path).exists());
        assert!(!std::path::Path::new(&format!("{}/pics_7.ecj", dir)).exists());
    }

    /// #9423: after an RPC `VolumeEcShardsMount` records the source
    /// volume's disk type on the EcVolume, a later empty-source mount
    /// call (disk-scan reload, orphan-shard reconcile, restart) must
    /// not clobber that override back to the physical location's disk
    /// type — otherwise heartbeat reporting drifts back to "hdd" on
    /// every reload.
    #[test]
    fn test_mount_ec_shards_empty_source_preserves_existing_disk_type() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();

        // Plant the first shard so the EC volume can mount, then call
        // mount_ec_shards with source_disk_type="ssd" — simulating the
        // VolumeEcShardsMount RPC path.
        std::fs::write(format!("{}/pics_7.ec00", dir), b"ec-shard").unwrap();
        std::fs::write(format!("{}/pics_7.ecx", dir), [0u8; 16]).unwrap();
        loc.mount_ec_shards(VolumeId(7), "pics", &[0], "ssd")
            .unwrap();
        {
            let ec_vol = loc.find_ec_volume(VolumeId(7)).expect("ec volume mounted");
            assert_eq!(
                ec_vol.disk_type,
                DiskType::Ssd,
                "first RPC mount should have set disk type to source's value",
            );
        }

        // Plant another shard and re-mount with an empty source —
        // matching what a disk-scan reload or orphan-shard reconcile
        // would pass. The previously-recorded "ssd" override must
        // survive.
        std::fs::write(format!("{}/pics_7.ec01", dir), b"ec-shard").unwrap();
        loc.mount_ec_shards(VolumeId(7), "pics", &[1], "").unwrap();
        {
            let ec_vol = loc
                .find_ec_volume(VolumeId(7))
                .expect("ec volume still mounted");
            assert_eq!(
                ec_vol.disk_type,
                DiskType::Ssd,
                "empty-source remount must not reset disk type to the physical location's",
            );
        }
    }

    /// A refused shard (a 0-byte file beside an index with entries) on the
    /// FIRST mount of a volume must not leave the just-created zero-shard
    /// EcVolume registered — it would advertise a mount serving no data,
    /// pin the .ecx/.ecj descriptors, and make placement's mounted tier
    /// prefer this disk. A volume that already holds shards keeps them
    /// (the mount RPC's pre-existing first-error-aborts contract).
    #[test]
    fn test_mount_ec_shards_refused_shard_removes_created_empty_volume() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();

        // An index with one 16-byte entry and a 0-byte shard: the mount is
        // refused, and the EcVolume created for it must be unregistered.
        std::fs::write(format!("{}/pics_9.ecx", dir), [0u8; 16]).unwrap();
        std::fs::write(format!("{}/pics_9.ec00", dir), b"").unwrap();
        let err = loc
            .mount_ec_shards(VolumeId(9), "pics", &[0], "")
            .expect_err("a 0-byte shard beside an index with entries must refuse the mount");
        assert!(
            err.to_string().contains("empty (0 bytes)"),
            "want the empty-shard refusal, got: {}",
            err
        );
        assert!(
            loc.find_ec_volume(VolumeId(9)).is_none(),
            "a refused first mount must not leave a zero-shard EcVolume registered",
        );

        // With a valid shard mounted, a later refused shard keeps the
        // existing registration intact.
        std::fs::write(format!("{}/pics_9.ec01", dir), b"good bytes").unwrap();
        loc.mount_ec_shards(VolumeId(9), "pics", &[1], "").unwrap();
        loc.mount_ec_shards(VolumeId(9), "pics", &[0], "")
            .expect_err("the 0-byte shard stays refused");
        assert_eq!(
            loc.find_ec_volume(VolumeId(9)).map(|v| v.shard_count()),
            Some(1),
            "an existing volume keeps its valid shards when a later shard is refused",
        );
    }

    /// A mount retry re-listing an already mounted shard must keep the
    /// existing registration and not bump the ec_shards gauge — the Rust
    /// twin of Go's AddEcVolumeShard added=false handling.
    #[test]
    fn test_mount_ec_shards_duplicate_keeps_registration_and_gauge() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();

        // A collection name unique to this test: the gauge is process-global
        // and sibling tests running in parallel touch other labels.
        std::fs::write(format!("{}/dupmount_11.ec00", dir), b"shard bytes").unwrap();
        std::fs::write(format!("{}/dupmount_11.ecx", dir), [0u8; 16]).unwrap();
        let gauge = crate::metrics::VOLUME_GAUGE.with_label_values(&["dupmount", "ec_shards"]);
        let before = gauge.get();

        loc.mount_ec_shards(VolumeId(11), "dupmount", &[0], "")
            .unwrap();
        loc.mount_ec_shards(VolumeId(11), "dupmount", &[0], "")
            .expect("a duplicate mount must succeed as a no-op");

        assert_eq!(
            loc.find_ec_volume(VolumeId(11)).map(|v| v.shard_count()),
            Some(1),
        );
        assert_eq!(
            gauge.get(),
            before + 1.0,
            "the duplicate mount must not bump the ec_shards gauge",
        );
    }

    #[test]
    fn test_disk_location_persists_directory_uuid_and_tags() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();

        let loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            vec!["fast".to_string(), "ssd".to_string()],
        )
        .unwrap();
        let directory_uuid = loc.directory_uuid.clone();
        assert_eq!(loc.tags, vec!["fast".to_string(), "ssd".to_string()]);
        drop(loc);

        let reloaded = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(1.0),
            Vec::new(),
        )
        .unwrap();
        assert_eq!(reloaded.directory_uuid, directory_uuid);
    }

    /// Sanity: `parse_ec_shard_extension` accepts `.ec00` through
    /// `.ec999` and rejects anything else.
    #[test]
    fn test_parse_ec_shard_extension() {
        assert_eq!(parse_ec_shard_extension(".ec00"), Some(0));
        assert_eq!(parse_ec_shard_extension(".ec13"), Some(13));
        assert_eq!(parse_ec_shard_extension(".ec999"), None); // > 255
        assert_eq!(parse_ec_shard_extension(".ec255"), Some(255));
        assert_eq!(parse_ec_shard_extension(".ec0"), None); // need 2 digits
        assert_eq!(parse_ec_shard_extension(".ecx"), None);
        assert_eq!(parse_ec_shard_extension(".ecj"), None);
        assert_eq!(parse_ec_shard_extension(".dat"), None);
        assert_eq!(parse_ec_shard_extension("ec00"), None); // missing dot
    }

    /// Auto-load: shards + .ecx on the same disk get mounted into
    /// `ec_volumes` after `load_existing_volumes` finishes. Without
    /// this, EC shards on disk are invisible until something issues
    /// `VolumeEcShardsMount`.
    #[test]
    fn test_load_all_ec_shards_mounts_pairs_with_ecx() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let collection = "grafana-loki";
        let vid = VolumeId(1234);

        // Two shards + .ecx + .ecj + .vif (the post-encode steady state).
        for &sid in &[0u8, 4u8] {
            let path = format!("{}/{}_{}.ec{:02}", dir, collection, vid.0, sid);
            std::fs::write(&path, b"shard data nonempty").unwrap();
        }
        std::fs::write(
            format!("{}/{}_{}.ecx", dir, collection, vid.0),
            vec![0u8; 20],
        )
        .unwrap();
        std::fs::write(format!("{}/{}_{}.ecj", dir, collection, vid.0), b"").unwrap();
        std::fs::write(
            format!("{}/{}_{}.vif", dir, collection, vid.0),
            br#"{"version":3}"#,
        )
        .unwrap();

        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(0.0),
            Vec::new(),
        )
        .unwrap();
        loc.load_existing_volumes(NeedleMapKind::InMemory).unwrap();

        let ec_vol = loc.find_ec_volume(vid).expect("EcVolume should be mounted");
        assert_eq!(ec_vol.shard_count(), 2, "both shards should be loaded");
    }

    /// Orphan shards with no `.ecx` and no `.dat` must stay on disk
    /// untouched (distributed-EC scenario, where the .ecx lives on
    /// another server). Mirrors Go's `checkOrphanedShards` no-op
    /// branch.
    #[test]
    fn test_load_all_ec_shards_keeps_orphan_shards_when_no_dat() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let collection = "grafana-loki";
        let vid = VolumeId(2345);

        // Shards on disk, but no .ecx and no .dat — distributed-EC
        // with the .ecx on a sibling/server.
        let shard_paths: Vec<String> = [0u8, 12u8]
            .iter()
            .map(|sid| format!("{}/{}_{}.ec{:02}", dir, collection, vid.0, sid))
            .collect();
        for p in &shard_paths {
            std::fs::write(p, b"shard data nonempty").unwrap();
        }

        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(0.0),
            Vec::new(),
        )
        .unwrap();
        loc.load_existing_volumes(NeedleMapKind::InMemory).unwrap();

        // No EcVolume should be mounted (no .ecx).
        assert!(loc.find_ec_volume(vid).is_none());
        // But the shard files must still be on disk for cross-disk
        // reconciliation / operator recovery.
        for p in &shard_paths {
            assert!(
                std::path::Path::new(p).exists(),
                "orphan shard {} was destroyed",
                p,
            );
        }
    }

    /// Orphan shards WITH a stale `.dat` are treated as an
    /// interrupted-encoding remnant and cleaned up. Mirrors Go's
    /// `checkOrphanedShards` cleanup branch. Guards a subtle case
    /// where a partial encoding wrote shards before .ecx and a crash
    /// left the .dat behind.
    #[test]
    fn test_load_all_ec_shards_cleans_orphan_shards_when_dat_exists() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let collection = "grafana-loki";
        let vid = VolumeId(3456);

        let shard_paths: Vec<String> = [0u8, 1u8]
            .iter()
            .map(|sid| format!("{}/{}_{}.ec{:02}", dir, collection, vid.0, sid))
            .collect();
        for p in &shard_paths {
            std::fs::write(p, b"shard data nonempty").unwrap();
        }
        // Stale .dat from interrupted encoding.
        let dat_path = format!("{}/{}_{}.dat", dir, collection, vid.0);
        std::fs::write(&dat_path, b"unfinished").unwrap();

        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(0.0),
            Vec::new(),
        )
        .unwrap();
        // load_existing_volumes processes .dat first; ec auto-load
        // runs after and finds the orphans. The .dat itself may or may
        // not be loadable as a regular volume — that's not what this
        // test is about. Just check that the orphan shard files were
        // removed.
        let _ = loc.load_existing_volumes(NeedleMapKind::InMemory);

        for p in &shard_paths {
            assert!(
                !std::path::Path::new(p).exists(),
                "orphan shard {} should have been cleaned up because .dat exists",
                p,
            );
        }
    }

    /// Mirrors Go's TestReopenEcShardReadsReplacedFile: a shard file the
    /// rebuild replaced under its mount keeps serving the old inode through
    /// the open descriptor; reopen_ec_shard must repoint the mounted shard at
    /// the new file without dropping its registration.
    #[test]
    fn test_reopen_ec_shard_reads_replaced_file() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path().to_str().unwrap();
        let mut loc = DiskLocation::new(
            dir,
            dir,
            10,
            DiskType::HardDrive,
            MinFreeSpace::Percent(0.0),
            Vec::new(),
        )
        .unwrap();

        std::fs::write(format!("{}/pics_5.ecx", dir), [0u8; 16]).unwrap();
        let shard_path = format!("{}/pics_5.ec00", dir);
        std::fs::write(&shard_path, b"old bytes").unwrap();
        loc.mount_ec_shards(VolumeId(5), "pics", &[0], "").unwrap();

        let read_shard = |loc: &DiskLocation| -> Vec<u8> {
            let ec_vol = loc.find_ec_volume(VolumeId(5)).expect("ec volume mounted");
            let shard = ec_vol.shards[0].as_ref().expect("shard 0 mounted");
            let mut buf = vec![0u8; shard.file_size() as usize];
            let n = shard.read_at(&mut buf, 0).unwrap();
            buf.truncate(n);
            buf
        };
        assert_eq!(read_shard(&loc), b"old bytes");

        // The rebuild writes a fresh file and renames it over the original:
        // the mounted descriptor is now on the detached inode.
        let rebuilt = format!("{}.rebuilding", shard_path);
        std::fs::write(&rebuilt, b"new bytes").unwrap();
        std::fs::rename(&rebuilt, &shard_path).unwrap();
        assert_eq!(read_shard(&loc), b"old bytes", "mounted shard still reads the replaced inode");

        assert!(loc.reopen_ec_shard(VolumeId(5), 0).unwrap());
        assert_eq!(read_shard(&loc), b"new bytes", "reopened shard should read the regenerated file");

        assert!(
            !loc.reopen_ec_shard(VolumeId(5), 7).unwrap(),
            "a shard that is not mounted should report false"
        );
    }
}