a3s_box_runtime/volume/

store.rs

//! Persistent storage for volume configurations.
//!
//! Volumes are stored as JSON in `~/.a3s/volumes.json` with atomic writes
//! (write to tmp file, then rename) to prevent corruption.
//! Volume data is stored under `~/.a3s/volumes/<name>/`.

use a3s_box_core::error::{BoxError, Result};
use a3s_box_core::volume::VolumeConfig;
use std::collections::HashMap;
use std::path::{Path, PathBuf};

/// Persistent store for volume configurations.
#[derive(Debug)]
pub struct VolumeStore {
    /// Path to the JSON file.
    path: PathBuf,
    /// Base directory for volume data (~/.a3s/volumes/).
    volumes_dir: PathBuf,
}

/// Serializable wrapper for the volumes file.
#[derive(Debug, serde::Serialize, serde::Deserialize, Default)]
struct VolumesFile {
    volumes: HashMap<String, VolumeConfig>,
}

const ANONYMOUS_LABEL: &str = "anonymous";
const ANONYMOUS_KIND_LABEL: &str = "a3s.box.volume.kind";
const ANONYMOUS_KIND: &str = "anonymous-v1";
const ANONYMOUS_OWNER_LABEL: &str = "a3s.box.volume.owner";

impl VolumeStore {
    /// Create a new store at the given path.
    pub fn new(path: impl Into<PathBuf>, volumes_dir: impl Into<PathBuf>) -> Self {
        Self {
            path: path.into(),
            volumes_dir: volumes_dir.into(),
        }
    }

    /// Create a store at the default location (`~/.a3s/volumes.json`).
    pub fn default_path() -> Result<Self> {
        let home = a3s_box_core::dirs_home();
        Ok(Self::new(home.join("volumes.json"), home.join("volumes")))
    }

    /// Load all volumes from disk.
    pub fn load(&self) -> Result<HashMap<String, VolumeConfig>> {
        if !self.path.exists() {
            return Ok(HashMap::new());
        }

        let data = std::fs::read_to_string(&self.path).map_err(|e| {
            BoxError::ConfigError(format!(
                "failed to read volumes file {}: {}",
                self.path.display(),
                e
            ))
        })?;

        // A corrupt/old-schema volumes file must not brick the runtime: quarantine
        // it and start from an empty set (create repopulates) rather than failing
        // every volume operation. Mirrors the boxes.json hardening.
        let file: VolumesFile = match serde_json::from_str(&data) {
            Ok(f) => f,
            Err(e) => {
                let preserved = crate::store_io::quarantine_label(&self.path);
                tracing::warn!(
                    "volumes file {} is corrupt ({e}); preserved a copy at {preserved} \
                     and started from an empty volume set",
                    self.path.display(),
                );
                return Ok(HashMap::new());
            }
        };

        Ok(file.volumes)
    }

    /// Save all volumes to disk (atomic write).
    pub fn save(&self, volumes: &HashMap<String, VolumeConfig>) -> Result<()> {
        if let Some(parent) = self.path.parent() {
            std::fs::create_dir_all(parent).map_err(|e| {
                BoxError::ConfigError(format!(
                    "failed to create directory {}: {}",
                    parent.display(),
                    e
                ))
            })?;
        }

        let file = VolumesFile {
            volumes: volumes.clone(),
        };

        let json = serde_json::to_string_pretty(&file).map_err(|e| {
            BoxError::SerializationError(format!("failed to serialize volumes: {}", e))
        })?;

        let tmp_path = self.path.with_extension("json.tmp");
        std::fs::write(&tmp_path, &json).map_err(|e| {
            BoxError::ConfigError(format!(
                "failed to write tmp file {}: {}",
                tmp_path.display(),
                e
            ))
        })?;

        std::fs::rename(&tmp_path, &self.path).map_err(|e| {
            BoxError::ConfigError(format!(
                "failed to rename {} → {}: {}",
                tmp_path.display(),
                self.path.display(),
                e
            ))
        })?;

        Ok(())
    }

    /// Run `f` over the volume map under a cross-process advisory lock,
    /// re-loading fresh from disk inside the lock and saving the result.
    ///
    /// `create`/`remove`/`update`/`modify`/`get_or_create` all funnel through
    /// here so concurrent `a3s-box` processes cannot lose each other's writes.
    /// The atomic tmp+rename in `save` only prevents a *torn* read — two
    /// processes that both load, mutate a different entry, and save would still
    /// clobber one update (and, for attach/detach, silently drop a volume's
    /// `in_use_by` entry, letting `prune`/`remove` delete data a live box still
    /// has mounted). `save` itself stays lock-free: the guard is held here for
    /// the whole load → mutate → save, and the lock is non-reentrant.
    fn with_write_lock<F, R>(&self, f: F) -> Result<R>
    where
        F: FnOnce(&mut HashMap<String, VolumeConfig>) -> Result<R>,
    {
        let _lock = crate::file_lock::FileLock::acquire(&self.path).map_err(|e| {
            BoxError::ConfigError(format!(
                "failed to lock volumes file {}: {e}",
                self.path.display()
            ))
        })?;
        let mut volumes = self.load()?;
        let r = f(&mut volumes)?;
        self.save(&volumes)?;
        Ok(r)
    }

    /// Get a single volume by name.
    pub fn get(&self, name: &str) -> Result<Option<VolumeConfig>> {
        let volumes = self.load()?;
        Ok(volumes.get(name).cloned())
    }

    /// Create a new named volume. Returns the host mount point path.
    ///
    /// Creates the volume data directory under `~/.a3s/volumes/<name>/`.
    /// Errors if the name already exists (use [`Self::get_or_create`] for the
    /// idempotent auto-create on the `run -v name:/path` path).
    pub fn create(&self, config: VolumeConfig) -> Result<VolumeConfig> {
        self.with_write_lock(|volumes| {
            if volumes.contains_key(&config.name) {
                return Err(BoxError::ConfigError(format!(
                    "volume '{}' already exists",
                    config.name
                )));
            }
            self.materialize(config, volumes)
        })
    }

    /// Return the existing volume, or create it if absent — atomic under the
    /// cross-process lock. Two concurrent first-time `run -v name:/path` then
    /// share one volume instead of one racing to an "already exists" error.
    pub fn get_or_create(&self, config: VolumeConfig) -> Result<VolumeConfig> {
        self.with_write_lock(|volumes| {
            if let Some(existing) = volumes.get(&config.name) {
                return Ok(existing.clone());
            }
            self.materialize(config, volumes)
        })
    }

    /// Atomically create or reclaim one Box-owned anonymous volume.
    ///
    /// Anonymous identities are single-owner capabilities. An existing named
    /// volume, a volume owned by another execution, or metadata pointing away
    /// from the canonical managed directory all fail closed without mutation.
    pub(crate) fn claim_anonymous(&self, name: &str, owner: &str) -> Result<(VolumeConfig, bool)> {
        validate_anonymous_identity(name, owner)?;

        let expected_mount_point = self.volumes_dir.join(name).to_string_lossy().into_owned();
        self.with_write_lock(|volumes| {
            if let Some(existing) = volumes.get_mut(name) {
                validate_anonymous_config(existing, name, owner, &expected_mount_point)?;
                ensure_managed_volume_directory(Path::new(&expected_mount_point))?;
                existing.attach(owner);
                existing
                    .labels
                    .insert(ANONYMOUS_KIND_LABEL.to_string(), ANONYMOUS_KIND.to_string());
                existing
                    .labels
                    .insert(ANONYMOUS_OWNER_LABEL.to_string(), owner.to_string());
                return Ok((existing.clone(), false));
            }

            let mut config = VolumeConfig::new(name, "");
            config
                .labels
                .insert(ANONYMOUS_LABEL.to_string(), "true".to_string());
            config
                .labels
                .insert(ANONYMOUS_KIND_LABEL.to_string(), ANONYMOUS_KIND.to_string());
            config
                .labels
                .insert(ANONYMOUS_OWNER_LABEL.to_string(), owner.to_string());
            config.attach(owner);
            self.materialize(config, volumes)
                .map(|created| (created, true))
        })
    }

    /// Remove only the anonymous volume capability owned by `owner`.
    ///
    /// The directory is removed while the metadata lock is held, before the
    /// atomic metadata update. If a previous claim created the deterministic
    /// directory but crashed before publishing metadata, the durable Box
    /// record can use this operation to remove that orphan safely.
    pub fn remove_anonymous(&self, name: &str, owner: &str) -> Result<bool> {
        validate_anonymous_identity(name, owner)?;
        let expected_mount_point = self.volumes_dir.join(name).to_string_lossy().into_owned();
        self.with_write_lock(|volumes| {
            let existed = if let Some(existing) = volumes.get(name) {
                validate_anonymous_config(existing, name, owner, &expected_mount_point)?;
                true
            } else {
                false
            };

            remove_managed_volume_path(Path::new(&expected_mount_point))?;
            if existed {
                volumes.remove(name);
            }
            Ok(existed)
        })
    }

    /// Create the volume's data directory, set its mount point, and insert it
    /// into `volumes`. Caller must already hold the write lock.
    fn materialize(
        &self,
        mut config: VolumeConfig,
        volumes: &mut HashMap<String, VolumeConfig>,
    ) -> Result<VolumeConfig> {
        let vol_dir = self.volumes_dir.join(&config.name);
        ensure_managed_volume_directory(&vol_dir)?;
        config.mount_point = vol_dir.to_string_lossy().into_owned();
        volumes.insert(config.name.clone(), config.clone());
        Ok(config)
    }

    /// Remove a volume by name. Returns error if in use.
    pub fn remove(&self, name: &str, force: bool) -> Result<VolumeConfig> {
        let config = self.with_write_lock(|volumes| {
            let config = volumes
                .remove(name)
                .ok_or_else(|| BoxError::ConfigError(format!("volume '{}' not found", name)))?;

            if config.is_in_use() && !force {
                // Put it back
                volumes.insert(name.to_string(), config.clone());
                return Err(BoxError::ConfigError(format!(
                    "volume '{}' is in use by {} box(es); use --force to remove",
                    name,
                    config.in_use_by.len()
                )));
            }
            Ok(config)
        })?;

        // Remove the data directory outside the lock; it is keyed by name and
        // the removal is idempotent.
        let vol_dir = self.volumes_dir.join(name);
        if vol_dir.exists() {
            std::fs::remove_dir_all(&vol_dir).ok();
        }

        Ok(config)
    }

    /// List all volumes.
    pub fn list(&self) -> Result<Vec<VolumeConfig>> {
        let volumes = self.load()?;
        Ok(volumes.into_values().collect())
    }

    /// Replace a volume's config wholesale under the cross-process lock.
    ///
    /// For attach/detach prefer [`Self::modify`]: a `get` → mutate → `update`
    /// reads outside the lock and would lose a concurrent update made between
    /// the two calls.
    pub fn update(&self, config: &VolumeConfig) -> Result<()> {
        self.with_write_lock(|volumes| {
            if !volumes.contains_key(&config.name) {
                return Err(BoxError::ConfigError(format!(
                    "volume '{}' not found",
                    config.name
                )));
            }
            volumes.insert(config.name.clone(), config.clone());
            Ok(())
        })
    }

    /// Atomically mutate one volume's config under the cross-process lock.
    ///
    /// Re-reads the current entry inside the lock so concurrent attach/detach
    /// accumulate correctly — the canonical fix for the split `get` → mutate →
    /// `update` race that could drop a volume's `in_use_by` entry. Returns
    /// `false` if the volume does not exist.
    pub fn modify<F>(&self, name: &str, f: F) -> Result<bool>
    where
        F: FnOnce(&mut VolumeConfig),
    {
        self.with_write_lock(|volumes| match volumes.get_mut(name) {
            Some(config) => {
                f(config);
                Ok(true)
            }
            None => Ok(false),
        })
    }

    /// Remove all volumes that are not in use. Returns names of removed volumes.
    pub fn prune(&self) -> Result<Vec<String>> {
        let volumes = self.load()?;
        let mut pruned = Vec::new();

        for (name, config) in &volumes {
            if !config.is_in_use() {
                pruned.push(name.clone());
            }
        }

        for name in &pruned {
            self.remove(name, false).ok();
        }

        Ok(pruned)
    }

    /// Get the volume data directory for a named volume.
    pub fn volume_dir(&self, name: &str) -> PathBuf {
        self.volumes_dir.join(name)
    }

    /// Get the store file path.
    pub fn path(&self) -> &Path {
        &self.path
    }
}

fn valid_anonymous_volume_name(name: &str) -> bool {
    name.starts_with("anon_")
        && name.len() <= 128
        && name
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
}

fn validate_anonymous_identity(name: &str, owner: &str) -> Result<()> {
    if !valid_anonymous_volume_name(name) {
        return Err(BoxError::ConfigError(format!(
            "invalid anonymous volume name {name:?}"
        )));
    }
    if owner.is_empty() || owner.contains('\0') {
        return Err(BoxError::ConfigError(
            "anonymous volume owner must be a non-empty execution identity".to_string(),
        ));
    }
    Ok(())
}

fn validate_anonymous_config(
    config: &VolumeConfig,
    name: &str,
    owner: &str,
    expected_mount_point: &str,
) -> Result<()> {
    if config.labels.get(ANONYMOUS_LABEL).map(String::as_str) != Some("true") {
        return Err(BoxError::ConfigError(format!(
            "volume {name:?} is not an anonymous volume"
        )));
    }
    if config.driver != "local" {
        return Err(BoxError::ConfigError(format!(
            "anonymous volume {name:?} does not use the local driver"
        )));
    }
    if config.mount_point != expected_mount_point {
        return Err(BoxError::ConfigError(format!(
            "anonymous volume {name:?} does not use its canonical managed directory"
        )));
    }
    let exact_owner = config.in_use_by.len() == 1
        && config
            .in_use_by
            .first()
            .is_some_and(|current| current == owner);
    if config.in_use_by.iter().any(|current| current != owner) {
        return Err(BoxError::ConfigError(format!(
            "anonymous volume {name:?} is owned by another execution"
        )));
    }
    match config.labels.get(ANONYMOUS_KIND_LABEL).map(String::as_str) {
        Some(ANONYMOUS_KIND)
            if exact_owner
                && config
                    .labels
                    .get(ANONYMOUS_OWNER_LABEL)
                    .is_some_and(|current| current == owner) =>
        {
            Ok(())
        }
        None if exact_owner => Ok(()),
        _ => Err(BoxError::ConfigError(format!(
            "anonymous volume {name:?} has no compatible ownership contract"
        ))),
    }
}

fn ensure_managed_volume_directory(path: &Path) -> Result<()> {
    match std::fs::symlink_metadata(path) {
        Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_dir() => {
            return Err(BoxError::ConfigError(format!(
                "managed volume path {} is not a directory",
                path.display()
            )))
        }
        Ok(_) => return Ok(()),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
        Err(error) => {
            return Err(BoxError::ConfigError(format!(
                "failed to inspect volume directory {}: {error}",
                path.display()
            )))
        }
    }
    std::fs::create_dir_all(path).map_err(|error| {
        BoxError::ConfigError(format!(
            "failed to create volume directory {}: {error}",
            path.display()
        ))
    })?;
    let metadata = std::fs::symlink_metadata(path).map_err(|error| {
        BoxError::ConfigError(format!(
            "failed to verify volume directory {}: {error}",
            path.display()
        ))
    })?;
    if metadata.file_type().is_symlink() || !metadata.is_dir() {
        return Err(BoxError::ConfigError(format!(
            "managed volume path {} is not a directory",
            path.display()
        )));
    }
    Ok(())
}

fn remove_managed_volume_path(path: &Path) -> Result<()> {
    let metadata = match std::fs::symlink_metadata(path) {
        Ok(metadata) => metadata,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
        Err(error) => return Err(BoxError::IoError(error)),
    };
    if metadata.is_dir() && !metadata.file_type().is_symlink() {
        std::fs::remove_dir_all(path).map_err(BoxError::IoError)
    } else {
        std::fs::remove_file(path).map_err(BoxError::IoError)
    }
}

impl a3s_box_core::traits::VolumeStoreBackend for VolumeStore {
    fn get(&self, name: &str) -> Result<Option<VolumeConfig>> {
        self.get(name)
    }

    fn create(&self, config: VolumeConfig) -> Result<VolumeConfig> {
        self.create(config)
    }

    fn remove(&self, name: &str, force: bool) -> Result<VolumeConfig> {
        self.remove(name, force)
    }

    fn list(&self) -> Result<Vec<VolumeConfig>> {
        self.list()
    }

    fn update(&self, config: &VolumeConfig) -> Result<()> {
        self.update(config)
    }

    fn prune(&self) -> Result<Vec<String>> {
        self.prune()
    }
}

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

    fn temp_store() -> (tempfile::TempDir, VolumeStore) {
        let dir = tempfile::tempdir().unwrap();
        let store = VolumeStore::new(dir.path().join("volumes.json"), dir.path().join("volumes"));
        (dir, store)
    }

    #[test]
    fn test_load_empty() {
        let (_dir, store) = temp_store();
        let volumes = store.load().unwrap();
        assert!(volumes.is_empty());
    }

    #[test]
    fn test_create_and_load() {
        let (_dir, store) = temp_store();
        let vol = VolumeConfig::new("mydata", "");
        store.create(vol).unwrap();

        let volumes = store.load().unwrap();
        assert_eq!(volumes.len(), 1);
        assert!(volumes.contains_key("mydata"));
    }

    #[test]
    fn test_create_sets_mount_point() {
        let (_dir, store) = temp_store();
        let vol = VolumeConfig::new("mydata", "");
        let created = store.create(vol).unwrap();

        assert!(created.mount_point.contains("mydata"));
        assert!(PathBuf::from(&created.mount_point).exists());
    }

    #[test]
    fn test_create_duplicate() {
        let (_dir, store) = temp_store();
        let v1 = VolumeConfig::new("mydata", "");
        let v2 = VolumeConfig::new("mydata", "");

        store.create(v1).unwrap();
        assert!(store.create(v2).is_err());
    }

    #[test]
    fn test_get_existing() {
        let (_dir, store) = temp_store();
        store.create(VolumeConfig::new("mydata", "")).unwrap();

        let found = store.get("mydata").unwrap();
        assert!(found.is_some());
        assert_eq!(found.unwrap().name, "mydata");
    }

    #[test]
    fn test_get_nonexistent() {
        let (_dir, store) = temp_store();
        let found = store.get("nope").unwrap();
        assert!(found.is_none());
    }

    #[test]
    fn test_remove() {
        let (_dir, store) = temp_store();
        store.create(VolumeConfig::new("mydata", "")).unwrap();

        let removed = store.remove("mydata", false).unwrap();
        assert_eq!(removed.name, "mydata");

        let volumes = store.load().unwrap();
        assert!(volumes.is_empty());
    }

    #[test]
    fn test_remove_nonexistent() {
        let (_dir, store) = temp_store();
        assert!(store.remove("nope", false).is_err());
    }

    #[test]
    fn test_remove_in_use_fails() {
        let (_dir, store) = temp_store();
        let mut vol = VolumeConfig::new("mydata", "");
        vol.attach("box-1");
        // Manually insert since create() doesn't set in_use_by
        let created = store.create(VolumeConfig::new("mydata", "")).unwrap();
        let mut updated = created;
        updated.attach("box-1");
        store.update(&updated).unwrap();

        assert!(store.remove("mydata", false).is_err());
    }

    #[test]
    fn test_remove_in_use_force() {
        let (_dir, store) = temp_store();
        let created = store.create(VolumeConfig::new("mydata", "")).unwrap();
        let mut updated = created;
        updated.attach("box-1");
        store.update(&updated).unwrap();

        let removed = store.remove("mydata", true).unwrap();
        assert_eq!(removed.name, "mydata");
    }

    #[test]
    fn test_list() {
        let (_dir, store) = temp_store();
        store.create(VolumeConfig::new("vol1", "")).unwrap();
        store.create(VolumeConfig::new("vol2", "")).unwrap();

        let list = store.list().unwrap();
        assert_eq!(list.len(), 2);
    }

    #[test]
    fn test_update() {
        let (_dir, store) = temp_store();
        let created = store.create(VolumeConfig::new("mydata", "")).unwrap();

        let mut updated = created;
        updated.attach("box-1");
        store.update(&updated).unwrap();

        let loaded = store.get("mydata").unwrap().unwrap();
        assert_eq!(loaded.in_use_by, vec!["box-1"]);
    }

    #[test]
    fn test_update_nonexistent() {
        let (_dir, store) = temp_store();
        let vol = VolumeConfig::new("nope", "/tmp");
        assert!(store.update(&vol).is_err());
    }

    #[test]
    fn test_prune() {
        let (_dir, store) = temp_store();
        store.create(VolumeConfig::new("unused1", "")).unwrap();
        store.create(VolumeConfig::new("unused2", "")).unwrap();

        let created = store.create(VolumeConfig::new("in_use", "")).unwrap();
        let mut updated = created;
        updated.attach("box-1");
        store.update(&updated).unwrap();

        let pruned = store.prune().unwrap();
        assert_eq!(pruned.len(), 2);
        assert!(pruned.contains(&"unused1".to_string()));
        assert!(pruned.contains(&"unused2".to_string()));

        // in_use should remain
        let remaining = store.list().unwrap();
        assert_eq!(remaining.len(), 1);
        assert_eq!(remaining[0].name, "in_use");
    }

    #[test]
    fn test_atomic_write() {
        let (_dir, store) = temp_store();
        store.create(VolumeConfig::new("mydata", "")).unwrap();

        let data = std::fs::read_to_string(store.path()).unwrap();
        let _: serde_json::Value = serde_json::from_str(&data).unwrap();

        let tmp = store.path().with_extension("json.tmp");
        assert!(!tmp.exists());
    }

    #[test]
    fn test_creates_parent_directory() {
        let dir = tempfile::tempdir().unwrap();
        let store = VolumeStore::new(
            dir.path().join("subdir").join("volumes.json"),
            dir.path().join("subdir").join("volumes"),
        );

        store.create(VolumeConfig::new("mydata", "")).unwrap();
        assert!(store.path().exists());
    }

    #[test]
    fn test_remove_cleans_up_directory() {
        let (_dir, store) = temp_store();
        let created = store.create(VolumeConfig::new("mydata", "")).unwrap();
        let vol_dir = PathBuf::from(&created.mount_point);
        assert!(vol_dir.exists());

        store.remove("mydata", false).unwrap();
        assert!(!vol_dir.exists());
    }

    #[test]
    fn test_get_or_create_is_idempotent() {
        let (_dir, store) = temp_store();
        let first = store
            .get_or_create(VolumeConfig::new("shared", ""))
            .unwrap();
        let second = store
            .get_or_create(VolumeConfig::new("shared", ""))
            .unwrap();
        assert_eq!(first.mount_point, second.mount_point);
        assert_eq!(store.list().unwrap().len(), 1);
    }

    #[test]
    fn test_get_or_create_preserves_existing_config() {
        let (_dir, store) = temp_store();
        let created = store
            .get_or_create(VolumeConfig::with_size_limit("shared", "", 4096))
            .unwrap();
        let mut updated = created;
        updated.attach("box-1");
        store.update(&updated).unwrap();

        let reused = store
            .get_or_create(VolumeConfig::with_size_limit("shared", "/ignored", 8192))
            .unwrap();

        assert_eq!(reused.size_limit, 4096);
        assert_eq!(reused.in_use_by, vec!["box-1"]);
        assert!(PathBuf::from(&reused.mount_point).exists());
    }

    #[test]
    fn claim_anonymous_is_idempotent_for_the_exact_owner() {
        let (_dir, store) = temp_store();

        let (first, first_created) = store.claim_anonymous("anon_owned", "box-owner").unwrap();
        let (second, second_created) = store.claim_anonymous("anon_owned", "box-owner").unwrap();

        assert!(first_created);
        assert!(!second_created);
        assert_eq!(first.mount_point, second.mount_point);
        assert_eq!(second.in_use_by, vec!["box-owner"]);
        assert_eq!(
            second.labels.get(ANONYMOUS_LABEL).map(String::as_str),
            Some("true")
        );
        assert_eq!(
            second.labels.get(ANONYMOUS_KIND_LABEL).map(String::as_str),
            Some(ANONYMOUS_KIND)
        );
        assert_eq!(
            second.labels.get(ANONYMOUS_OWNER_LABEL).map(String::as_str),
            Some("box-owner")
        );
    }

    #[test]
    fn claim_anonymous_upgrades_an_exact_owner_legacy_volume() {
        let (_dir, store) = temp_store();
        let mut legacy = VolumeConfig::new("anon_legacy", "");
        legacy
            .labels
            .insert(ANONYMOUS_LABEL.to_string(), "true".to_string());
        legacy.attach("box-owner");
        store.create(legacy).unwrap();

        let (claimed, created) = store.claim_anonymous("anon_legacy", "box-owner").unwrap();

        assert!(!created);
        assert_eq!(
            claimed.labels.get(ANONYMOUS_KIND_LABEL).map(String::as_str),
            Some(ANONYMOUS_KIND)
        );
        assert_eq!(
            claimed
                .labels
                .get(ANONYMOUS_OWNER_LABEL)
                .map(String::as_str),
            Some("box-owner")
        );
    }

    #[test]
    fn claim_anonymous_rejects_named_volume_collision_without_mutation() {
        let (_dir, store) = temp_store();
        let named = store
            .create(VolumeConfig::new("anon_collision", ""))
            .unwrap();

        let error = store
            .claim_anonymous("anon_collision", "box-owner")
            .expect_err("a named volume must never become anonymously owned");

        assert!(error.to_string().contains("not an anonymous volume"));
        assert_eq!(
            store.get("anon_collision").unwrap().unwrap().in_use_by,
            named.in_use_by
        );
    }

    #[test]
    fn claim_anonymous_rejects_a_different_owner_without_mutation() {
        let (_dir, store) = temp_store();
        store.claim_anonymous("anon_owned", "box-one").unwrap();

        let error = store
            .claim_anonymous("anon_owned", "box-two")
            .expect_err("anonymous volumes have exactly one Box owner");

        assert!(error.to_string().contains("owned by another execution"));
        assert_eq!(
            store.get("anon_owned").unwrap().unwrap().in_use_by,
            vec!["box-one"]
        );
    }

    #[test]
    fn claim_anonymous_rejects_unsafe_identity_before_creating_a_directory() {
        let (dir, store) = temp_store();

        let error = store
            .claim_anonymous("../escaped", "box-owner")
            .expect_err("managed identities may not escape the volume root");

        assert!(error.to_string().contains("invalid anonymous volume name"));
        assert!(!dir.path().join("escaped").exists());
        assert!(store.load().unwrap().is_empty());
    }

    #[test]
    fn concurrent_anonymous_claims_publish_one_exact_owner() {
        use std::sync::Arc;
        use std::thread;

        let dir = tempfile::tempdir().unwrap();
        let store = Arc::new(VolumeStore::new(
            dir.path().join("volumes.json"),
            dir.path().join("volumes"),
        ));
        let handles = (0..16)
            .map(|_| {
                let store = Arc::clone(&store);
                thread::spawn(move || {
                    store
                        .claim_anonymous("anon_concurrent", "box-owner")
                        .unwrap()
                        .1
                })
            })
            .collect::<Vec<_>>();

        let created = handles
            .into_iter()
            .map(|handle| usize::from(handle.join().unwrap()))
            .sum::<usize>();
        let claimed = store.get("anon_concurrent").unwrap().unwrap();

        assert_eq!(created, 1);
        assert_eq!(claimed.in_use_by, vec!["box-owner"]);
        assert_eq!(store.list().unwrap().len(), 1);
    }

    #[test]
    fn remove_anonymous_requires_and_removes_the_exact_owner() {
        let (_dir, store) = temp_store();
        let (claimed, _) = store.claim_anonymous("anon_owned", "box-owner").unwrap();

        let removed = store.remove_anonymous("anon_owned", "box-owner").unwrap();

        assert!(removed);
        assert!(store.get("anon_owned").unwrap().is_none());
        assert!(!PathBuf::from(claimed.mount_point).exists());
    }

    #[test]
    fn remove_anonymous_rejects_named_and_different_owner_collisions() {
        let (_dir, store) = temp_store();
        let named = store.create(VolumeConfig::new("anon_named", "")).unwrap();
        store.claim_anonymous("anon_owned", "box-one").unwrap();

        let named_error = store
            .remove_anonymous("anon_named", "box-one")
            .expect_err("named volume metadata must fail closed");
        let owner_error = store
            .remove_anonymous("anon_owned", "box-two")
            .expect_err("another owner must fail closed");

        assert!(named_error.to_string().contains("not an anonymous volume"));
        assert!(owner_error
            .to_string()
            .contains("owned by another execution"));
        assert!(PathBuf::from(named.mount_point).exists());
        assert!(store.get("anon_owned").unwrap().is_some());
    }

    #[test]
    fn remove_anonymous_cleans_an_unpublished_deterministic_directory() {
        let (_dir, store) = temp_store();
        let orphan = store.volume_dir("anon_unpublished");
        std::fs::create_dir_all(&orphan).unwrap();
        std::fs::write(orphan.join("partial"), b"partial").unwrap();

        let removed = store
            .remove_anonymous("anon_unpublished", "box-owner")
            .unwrap();

        assert!(!removed);
        assert!(!orphan.exists());
        assert!(store.get("anon_unpublished").unwrap().is_none());
    }

    #[test]
    fn reclaim_anonymous_recreates_a_missing_owned_directory() {
        let (_dir, store) = temp_store();
        let (claimed, _) = store.claim_anonymous("anon_owned", "box-owner").unwrap();
        std::fs::remove_dir_all(&claimed.mount_point).unwrap();

        let (reclaimed, created) = store.claim_anonymous("anon_owned", "box-owner").unwrap();

        assert!(!created);
        assert!(PathBuf::from(reclaimed.mount_point).is_dir());
    }

    #[test]
    fn test_modify_missing_returns_false() {
        let (_dir, store) = temp_store();
        assert!(!store.modify("nope", |c| c.attach("box-1")).unwrap());
    }

    #[test]
    fn test_modify_existing_returns_true_and_persists() {
        let (_dir, store) = temp_store();
        store.create(VolumeConfig::new("shared", "")).unwrap();

        let modified = store.modify("shared", |c| c.attach("box-1")).unwrap();

        assert!(modified);
        assert_eq!(
            store.get("shared").unwrap().unwrap().in_use_by,
            vec!["box-1"]
        );
    }

    #[test]
    fn test_volume_dir_joins_name_under_base_directory() {
        let (dir, store) = temp_store();
        assert_eq!(
            store.volume_dir("mydata"),
            dir.path().join("volumes").join("mydata")
        );
    }

    #[test]
    fn test_volume_store_backend_trait_dispatch() {
        let (_dir, store) = temp_store();
        let backend: &dyn a3s_box_core::traits::VolumeStoreBackend = &store;

        let created = backend.create(VolumeConfig::new("shared", "")).unwrap();
        assert_eq!(created.name, "shared");
        assert!(backend.get("shared").unwrap().is_some());

        let mut updated = created;
        updated.attach("box-1");
        backend.update(&updated).unwrap();
        assert_eq!(backend.list().unwrap().len(), 1);

        let err = backend.remove("shared", false).unwrap_err().to_string();
        assert!(err.contains("is in use"));

        let removed = backend.remove("shared", true).unwrap();
        assert_eq!(removed.name, "shared");
        assert!(backend.list().unwrap().is_empty());
    }

    // The advisory lock is per-open-file-description, so separate
    // FileLock::acquire calls serialize even across threads in one process —
    // which is exactly what lets this exercise the lost-update fix in-process.
    #[test]
    fn concurrent_attaches_accumulate_without_lost_update() {
        use std::sync::Arc;
        use std::thread;

        let dir = tempfile::tempdir().unwrap();
        let store = Arc::new(VolumeStore::new(
            dir.path().join("volumes.json"),
            dir.path().join("volumes"),
        ));
        store.create(VolumeConfig::new("shared", "")).unwrap();

        let n = 16;
        let handles: Vec<_> = (0..n)
            .map(|i| {
                let store = Arc::clone(&store);
                thread::spawn(move || {
                    store
                        .modify("shared", |c| c.attach(&format!("box-{i}")))
                        .unwrap();
                })
            })
            .collect();
        for h in handles {
            h.join().unwrap();
        }

        let cfg = store.get("shared").unwrap().unwrap();
        assert_eq!(
            cfg.in_use_by.len(),
            n,
            "every concurrent attach must persist (no lost update): {:?}",
            cfg.in_use_by
        );
        for i in 0..n {
            assert!(cfg.in_use_by.contains(&format!("box-{i}")));
        }
    }

    #[test]
    fn concurrent_creates_persist_every_volume() {
        use std::sync::Arc;
        use std::thread;

        let dir = tempfile::tempdir().unwrap();
        let store = Arc::new(VolumeStore::new(
            dir.path().join("volumes.json"),
            dir.path().join("volumes"),
        ));

        let n = 16;
        let handles: Vec<_> = (0..n)
            .map(|i| {
                let store = Arc::clone(&store);
                thread::spawn(move || {
                    store
                        .create(VolumeConfig::new(&format!("vol-{i}"), ""))
                        .unwrap();
                })
            })
            .collect();
        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(
            store.list().unwrap().len(),
            n,
            "every concurrent create must persist (no lost update)"
        );
    }

    #[test]
    fn corrupt_volumes_file_is_quarantined_not_fatal() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("volumes.json");
        std::fs::write(&path, "{ not valid json").unwrap();
        let store = VolumeStore::new(path.clone(), dir.path().join("volumes"));

        // load() must succeed (empty) instead of erroring every volume op.
        assert!(store.load().unwrap().is_empty());
        let quarantined = std::fs::read_dir(dir.path())
            .unwrap()
            .filter_map(|e| e.ok())
            .any(|e| {
                e.file_name()
                    .to_string_lossy()
                    .contains("volumes.json.corrupt-")
            });
        assert!(quarantined, "corrupt volumes.json must be quarantined");
    }
}