storage/

object.rs

//! Object Storage Backend using OpenDAL
//!
//! Provides storage abstraction for multiple cloud storage backends:
//! - S3-compatible (AWS S3, MinIO, DigitalOcean Spaces, etc.)
//! - Azure Blob Storage
//! - Google Cloud Storage (GCS)
//! - Local filesystem
//! - In-memory (for testing)

use async_trait::async_trait;
use common::{DakeraError, NamespaceId, Result, Vector, VectorId};
use futures_util::stream::{self, StreamExt};
use opendal::{layers::RetryLayer, services, Operator};
use serde::{Deserialize, Serialize};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

static TMP_COUNTER: AtomicU64 = AtomicU64::new(0);

use crate::traits::{IndexStorage, IndexType, VectorStorage};

fn s3_concurrent_ops() -> usize {
    std::env::var("DAKERA_S3_CONCURRENT_OPS")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(16)
}

/// Serializable namespace metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
struct NamespaceMetadata {
    dimension: Option<usize>,
    vector_count: usize,
    created_at: u64,
    updated_at: u64,
}

/// Serializable vector data for storage
#[derive(Debug, Clone, Serialize, Deserialize)]
struct StoredVector {
    id: String,
    values: Vec<f32>,
    metadata: Option<serde_json::Value>,
}

impl From<Vector> for StoredVector {
    fn from(v: Vector) -> Self {
        Self {
            id: v.id,
            values: v.values,
            metadata: v.metadata,
        }
    }
}

impl From<StoredVector> for Vector {
    fn from(v: StoredVector) -> Self {
        Self {
            id: v.id,
            values: v.values,
            metadata: v.metadata,
            ttl_seconds: None,
            expires_at: None,
        }
    }
}

/// Object storage backend configuration
#[derive(Debug, Clone, Default)]
pub enum ObjectStorageConfig {
    /// In-memory storage (for testing)
    #[default]
    Memory,
    /// Local filesystem storage
    Filesystem { root: String },
    /// S3-compatible storage (S3, MinIO, etc.)
    S3 {
        bucket: String,
        region: Option<String>,
        endpoint: Option<String>,
        access_key_id: Option<String>,
        secret_access_key: Option<String>,
    },
    /// Azure Blob Storage
    Azure {
        container: String,
        account_name: String,
        account_key: Option<String>,
        sas_token: Option<String>,
        endpoint: Option<String>,
    },
    /// Google Cloud Storage
    Gcs {
        bucket: String,
        credential_path: Option<String>,
        endpoint: Option<String>,
    },
}

/// Object storage backend using OpenDAL
#[derive(Clone)]
pub struct ObjectStorage {
    operator: Operator,
    /// DAK-6287: local filesystem root, retained ONLY when the backend is a real
    /// local filesystem. Lets `upsert` take a raw-`std::fs` fast write path that
    /// bypasses OpenDAL's per-op overhead (~9× faster on the same disk; profiled).
    /// `None` for memory/S3/Azure/GCS — those keep the OpenDAL path.
    fs_root: Option<std::path::PathBuf>,
}

impl ObjectStorage {
    /// Create a new object storage with the given configuration
    pub fn new(config: ObjectStorageConfig) -> Result<Self> {
        let operator = Self::build_operator(&config)?;
        let fs_root = match &config {
            ObjectStorageConfig::Filesystem { root } => Some(std::path::PathBuf::from(root)),
            _ => None,
        };
        Ok(Self { operator, fs_root })
    }

    /// DAK-6287: true when `s` is safe as a single filesystem path segment (non-empty,
    /// no separators, no parent refs, no NUL). Gates the raw-fs fast write path so a
    /// crafted namespace or vector id cannot escape the storage root via path traversal.
    fn is_fs_safe_segment(s: &str) -> bool {
        !s.is_empty()
            && s != "."
            && s != ".."
            && !s.contains('/')
            && !s.contains('\\')
            && !s.contains('\0')
    }

    /// DAK-6287: raw-`std::fs` batch write for the local Filesystem backend. Each vector is
    /// written to `<root>/namespaces/<ns>/vectors/<id>.json` via tmp+fsync+rename, matching
    /// the OpenDAL on-disk layout byte-for-byte so the read/recall path is unchanged. The
    /// directory is created once per batch; each file is fsync'd before its atomic rename, so
    /// crash-consistency is >= the OpenDAL path. Bypasses OpenDAL's per-op overhead, which the
    /// storage_upsert profile showed dominates (~9× faster on the same disk). Callers MUST
    /// ensure `namespace` and every vector id pass `is_fs_safe_segment`.
    ///
    /// Returns the number of NEW inserts (ids not already present on disk before the batch),
    /// so the caller can maintain `meta.vector_count` without a separate O(namespace) listing
    /// (DAK-6299). Classification is a single cheap raw-`std::fs` stat per file (O(batch),
    /// ~1% of upsert cost per the storage_upsert profile), done inside the per-file write
    /// task so it adds no extra round-trip pass over the namespace.
    async fn write_vectors_fs(
        &self,
        root: &std::path::Path,
        namespace: &NamespaceId,
        vectors: Vec<Vector>,
    ) -> Result<usize> {
        let vectors_dir = root.join(format!("namespaces/{}/vectors", namespace));
        tokio::fs::create_dir_all(&vectors_dir)
            .await
            .map_err(|e| DakeraError::Storage(e.to_string()))?;

        let results: Vec<Result<bool>> = stream::iter(vectors)
            .map(|vector| {
                let dir = vectors_dir.clone();
                async move {
                    let id = vector.id.clone();
                    let stored: StoredVector = vector.into();
                    let data = serde_json::to_vec(&stored)
                        .map_err(|e| DakeraError::Storage(e.to_string()))?;
                    let final_path = dir.join(format!("{}.json", id));
                    let seq = TMP_COUNTER.fetch_add(1, Ordering::Relaxed);
                    // Leading dot keeps the temp file out of the `*.json` id listing.
                    let tmp_path = dir.join(format!(".{}.tmp.{}", id, seq));
                    tokio::task::spawn_blocking(move || -> Result<bool> {
                        use std::io::Write;
                        // Cheap per-file stat classifies insert-vs-update (matches the prior
                        // !exists semantics) without scanning the whole namespace.
                        let is_new = !final_path.exists();
                        let mut f = std::fs::File::create(&tmp_path)
                            .map_err(|e| DakeraError::Storage(e.to_string()))?;
                        f.write_all(&data)
                            .map_err(|e| DakeraError::Storage(e.to_string()))?;
                        // Flush contents before the atomic publish so a crash can never
                        // expose a half-written vector file (>= OpenDAL durability).
                        f.sync_all()
                            .map_err(|e| DakeraError::Storage(e.to_string()))?;
                        drop(f);
                        if let Err(e) = std::fs::rename(&tmp_path, &final_path) {
                            let _ = std::fs::remove_file(&tmp_path);
                            return Err(DakeraError::Storage(e.to_string()));
                        }
                        Ok(is_new)
                    })
                    .await
                    .map_err(|e| DakeraError::Storage(e.to_string()))?
                }
            })
            .buffer_unordered(s3_concurrent_ops())
            .collect()
            .await;

        let mut new_inserts = 0usize;
        for r in results {
            if r? {
                new_inserts += 1;
            }
        }
        Ok(new_inserts)
    }

    /// Create in-memory storage (for testing)
    pub fn memory() -> Result<Self> {
        Self::new(ObjectStorageConfig::Memory)
    }

    /// Create filesystem storage
    pub fn filesystem(root: impl Into<String>) -> Result<Self> {
        Self::new(ObjectStorageConfig::Filesystem { root: root.into() })
    }

    /// Create S3 storage
    pub fn s3(bucket: impl Into<String>) -> Result<Self> {
        Self::new(ObjectStorageConfig::S3 {
            bucket: bucket.into(),
            region: None,
            endpoint: None,
            access_key_id: None,
            secret_access_key: None,
        })
    }

    /// Create Azure Blob storage
    pub fn azure(container: impl Into<String>, account_name: impl Into<String>) -> Result<Self> {
        Self::new(ObjectStorageConfig::Azure {
            container: container.into(),
            account_name: account_name.into(),
            account_key: None,
            sas_token: None,
            endpoint: None,
        })
    }

    /// Create Google Cloud Storage
    pub fn gcs(bucket: impl Into<String>) -> Result<Self> {
        Self::new(ObjectStorageConfig::Gcs {
            bucket: bucket.into(),
            credential_path: None,
            endpoint: None,
        })
    }

    pub fn build_operator(config: &ObjectStorageConfig) -> Result<Operator> {
        match config {
            ObjectStorageConfig::Memory => {
                let builder = services::Memory::default();
                Operator::new(builder)
                    .map(|op| op.finish())
                    .map_err(|e| DakeraError::Storage(e.to_string()))
            }
            ObjectStorageConfig::Filesystem { root } => {
                let builder = services::Fs::default().root(root);
                Operator::new(builder)
                    .map(|op| op.layer(Self::retry_layer()).finish())
                    .map_err(|e| DakeraError::Storage(e.to_string()))
            }
            ObjectStorageConfig::S3 {
                bucket,
                region,
                endpoint,
                access_key_id,
                secret_access_key,
            } => {
                let mut builder = services::S3::default().bucket(bucket);

                if let Some(region) = region {
                    builder = builder.region(region);
                }
                if let Some(endpoint) = endpoint {
                    builder = builder.endpoint(endpoint);
                }
                if let Some(key) = access_key_id {
                    builder = builder.access_key_id(key);
                }
                if let Some(secret) = secret_access_key {
                    builder = builder.secret_access_key(secret);
                }

                Operator::new(builder)
                    .map(|op| op.layer(Self::retry_layer()).finish())
                    .map_err(|e| DakeraError::Storage(e.to_string()))
            }
            ObjectStorageConfig::Azure {
                container,
                account_name,
                account_key,
                sas_token,
                endpoint,
            } => {
                let mut builder = services::Azblob::default()
                    .container(container)
                    .account_name(account_name);

                if let Some(key) = account_key {
                    builder = builder.account_key(key);
                }
                if let Some(token) = sas_token {
                    builder = builder.sas_token(token);
                }
                if let Some(endpoint) = endpoint {
                    builder = builder.endpoint(endpoint);
                }

                Operator::new(builder)
                    .map(|op| op.layer(Self::retry_layer()).finish())
                    .map_err(|e| DakeraError::Storage(e.to_string()))
            }
            ObjectStorageConfig::Gcs {
                bucket,
                credential_path,
                endpoint,
            } => {
                let mut builder = services::Gcs::default().bucket(bucket);

                if let Some(cred_path) = credential_path {
                    builder = builder.credential_path(cred_path);
                }
                if let Some(endpoint) = endpoint {
                    builder = builder.endpoint(endpoint);
                }

                Operator::new(builder)
                    .map(|op| op.layer(Self::retry_layer()).finish())
                    .map_err(|e| DakeraError::Storage(e.to_string()))
            }
        }
    }

    fn retry_layer() -> RetryLayer {
        // DAK-3430: default capped from 10 → 3 so a MinIO 429 storm fails fast
        // (≤3×60s ≈ 3min) rather than blocking production for 10min. Override
        // with DAKERA_S3_MAX_RETRIES=10 for high-resilience deployments.
        let max_times: usize = std::env::var("DAKERA_S3_MAX_RETRIES")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(3);
        let min_delay_ms: u64 = std::env::var("DAKERA_S3_RETRY_MIN_DELAY_MS")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(500);
        let max_delay_secs: u64 = std::env::var("DAKERA_S3_RETRY_MAX_DELAY_SECS")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(60);

        tracing::info!(
            max_times,
            min_delay_ms,
            max_delay_secs,
            "S3 retry layer configured"
        );

        RetryLayer::new()
            .with_max_times(max_times)
            .with_min_delay(Duration::from_millis(min_delay_ms))
            .with_max_delay(Duration::from_secs(max_delay_secs))
            .with_jitter()
            .with_factor(2.0)
    }

    /// Get the path for a vector
    fn vector_path(namespace: &str, vector_id: &str) -> String {
        format!("namespaces/{}/vectors/{}.json", namespace, vector_id)
    }

    /// Get the path for namespace metadata
    fn namespace_meta_path(namespace: &str) -> String {
        format!("namespaces/{}/meta.json", namespace)
    }

    /// Get the path prefix for all vectors in a namespace
    fn namespace_vectors_prefix(namespace: &str) -> String {
        format!("namespaces/{}/vectors/", namespace)
    }

    /// Read namespace metadata
    async fn read_namespace_meta(&self, namespace: &str) -> Result<Option<NamespaceMetadata>> {
        let path = Self::namespace_meta_path(namespace);
        match self.operator.read(&path).await {
            Ok(data) => {
                let bytes = data.to_vec();
                if bytes.is_empty() {
                    tracing::warn!(
                        namespace = %namespace,
                        path = %path,
                        "Empty namespace metadata file detected, treating as missing"
                    );
                    return Ok(None);
                }
                match serde_json::from_slice(&bytes) {
                    Ok(meta) => Ok(Some(meta)),
                    Err(e) => {
                        tracing::warn!(
                            namespace = %namespace,
                            path = %path,
                            error = %e,
                            bytes_len = bytes.len(),
                            "Corrupted namespace metadata, treating as missing and will be recreated"
                        );
                        Ok(None)
                    }
                }
            }
            Err(e) if e.kind() == opendal::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(DakeraError::Storage(e.to_string())),
        }
    }

    /// Write namespace metadata atomically.
    ///
    /// Writes to a `.tmp` file first, then renames to the final path. On POSIX filesystems
    async fn write_namespace_meta(&self, namespace: &str, meta: &NamespaceMetadata) -> Result<()> {
        let path = Self::namespace_meta_path(namespace);
        let data = serde_json::to_vec(meta).map_err(|e| DakeraError::Storage(e.to_string()))?;
        self.write_atomic(&path, data).await
    }

    /// Write data to `path` atomically via tmp+rename (DAK-4545).
    ///
    /// Writes to a `.tmp` file first, then renames to the final path. On POSIX filesystems
    /// rename(2) is atomic within the same device, preventing concurrent readers from seeing
    /// a truncated/empty file during the write window (O_TRUNC race). For object-store
    /// backends that don't support rename (e.g. S3 without copy-delete), falls back to a
    /// direct write — S3 PUT is itself atomic (readers get previous version until PUT completes).
    async fn write_atomic(&self, path: &str, data: Vec<u8>) -> Result<()> {
        let seq = TMP_COUNTER.fetch_add(1, Ordering::Relaxed);
        let tmp_path = format!("{}.tmp.{}.{}", path, Self::now(), seq);
        let rename_result = async {
            self.operator
                .write(&tmp_path, data.clone())
                .await
                .map_err(|e| DakeraError::Storage(e.to_string()))?;
            self.operator
                .rename(&tmp_path, path)
                .await
                .map_err(|e| DakeraError::Storage(e.to_string()))?;
            Ok::<(), DakeraError>(())
        }
        .await;
        if let Err(e) = rename_result {
            // Backend doesn't support rename: fall back to direct write.
            tracing::debug!(path = %path, error = %e, "atomic rename failed, falling back to direct write");
            let _ = self.operator.delete(&tmp_path).await;
            self.operator
                .write(path, data)
                .await
                .map_err(|e| DakeraError::Storage(e.to_string()))?;
        }
        Ok(())
    }

    /// Get current timestamp
    fn now() -> u64 {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs()
    }
}

#[async_trait]
impl VectorStorage for ObjectStorage {
    async fn upsert(&self, namespace: &NamespaceId, vectors: Vec<Vector>) -> Result<usize> {
        if vectors.is_empty() {
            return Ok(0);
        }

        // Get or create namespace metadata
        let mut meta = self
            .read_namespace_meta(namespace)
            .await?
            .unwrap_or_else(|| NamespaceMetadata {
                dimension: None,
                vector_count: 0,
                created_at: Self::now(),
                updated_at: Self::now(),
            });

        // Validate dimensions
        let first_dim = vectors[0].values.len();
        if let Some(dim) = meta.dimension {
            for v in &vectors {
                if v.values.len() != dim {
                    return Err(DakeraError::DimensionMismatch {
                        expected: dim,
                        actual: v.values.len(),
                    });
                }
            }
        } else {
            meta.dimension = Some(first_dim);
        }

        // DAK-5553: parallelize per-vector write to eliminate sequential I/O bottleneck.
        // Buffer up to s3_concurrent_ops() (default 16) concurrent writes at a time.
        //
        // DAK-6299: classify insert-vs-update (to maintain meta.vector_count) with O(batch)
        // per-vector existence checks, NOT a per-batch full-namespace list(). PR#612 replaced
        // N per-vector exists() stats with ONE operator.list(vectors_prefix), but that list
        // returns the ENTIRE namespace = O(namespace_size) per batch => O(N^2/batch) over a
        // full ingest, and it ran for ALL backends (incl S3, where it is a paginated LIST of
        // every key). Full-scale measurement (DAK-6297, fs, 2675 mem, one namespace) showed
        // the list inverted the fast path: 1.43x at 220 mem -> 0.55x at 2675 mem. The
        // profile that motivated PR#612 (DAK-6287) showed the per-vector stat was only ~1%
        // of upsert cost, so removing it bought ~nothing while the list it introduced is the
        // regression. Here: on the fs fast path a cheap raw std::fs stat per file (O(batch))
        // classifies new-vs-existing; on the OpenDAL path a per-vector exists() (O(batch)
        // HEAD/stat, the pre-PR#612 DAK-5553 semantics) does the same. No full-namespace
        // scan. Write-path only, recall-neutral: vector file format, the read/recall path,
        // and the tmp+rename atomic-publish durability guarantee are all unchanged.
        let total = vectors.len();

        // DAK-6287: on the local Filesystem backend, write via raw std::fs instead of
        // OpenDAL. Profiled on an x64 runner (storage_upsert decomposition): the per-vector
        // cost is dominated by OpenDAL's fs-writer overhead, NOT fsync/stat/rename — raw
        // write+fsync reaches ~5800 mem/s vs OpenDAL ~650 mem/s on the same disk. The fast
        // path keeps the EXACT on-disk layout (namespaces/<ns>/vectors/<id>.json), atomic
        // tmp+rename publish, and adds an explicit per-file fsync, so the read/recall path is
        // byte-identical and durability is >= the OpenDAL path. Guarded: only taken when the
        // namespace and every id are filesystem-safe segments (no `/`, `\`, `..`, NUL) to
        // prevent path traversal; otherwise it falls back to the OpenDAL path below.
        let fs_root = if Self::is_fs_safe_segment(namespace)
            && vectors.iter().all(|v| Self::is_fs_safe_segment(&v.id))
        {
            self.fs_root.clone()
        } else {
            None
        };

        let new_inserts = if let Some(root) = fs_root {
            // Fast path classifies inserts via a per-file stat inside the write loop.
            self.write_vectors_fs(&root, namespace, vectors).await?
        } else {
            let op = self.operator.clone();
            let ns = namespace.clone();
            let results: Vec<Result<bool>> = stream::iter(vectors)
                .map(|vector| {
                    let op = op.clone();
                    let ns = ns.clone();
                    async move {
                        let path = ObjectStorage::vector_path(&ns, &vector.id);
                        let stored: StoredVector = vector.into();
                        let data = serde_json::to_vec(&stored)
                            .map_err(|e| DakeraError::Storage(e.to_string()))?;

                        // O(1) per vector: a single exists() (HEAD on S3) classifies
                        // insert-vs-update without scanning the namespace.
                        let exists = op
                            .exists(&path)
                            .await
                            .map_err(|e| DakeraError::Storage(e.to_string()))?;

                        // Atomic write: tmp+rename, fall back to direct write if unsupported.
                        let seq = TMP_COUNTER.fetch_add(1, Ordering::Relaxed);
                        let now_secs = std::time::SystemTime::now()
                            .duration_since(std::time::UNIX_EPOCH)
                            .unwrap_or_default()
                            .as_secs();
                        let tmp_path = format!("{}.tmp.{}.{}", path, now_secs, seq);
                        let rename_result = async {
                            op.write(&tmp_path, data.clone())
                                .await
                                .map_err(|e| DakeraError::Storage(e.to_string()))?;
                            op.rename(&tmp_path, &path)
                                .await
                                .map_err(|e| DakeraError::Storage(e.to_string()))?;
                            Ok::<(), DakeraError>(())
                        }
                        .await;
                        if let Err(e) = rename_result {
                            tracing::debug!(
                                path = %path,
                                error = %e,
                                "atomic rename failed, falling back to direct write"
                            );
                            let _ = op.delete(&tmp_path).await;
                            op.write(&path, data)
                                .await
                                .map_err(|e| DakeraError::Storage(e.to_string()))?;
                        }

                        Ok::<bool, DakeraError>(!exists)
                    }
                })
                .buffer_unordered(s3_concurrent_ops())
                .collect()
                .await;

            let mut count = 0usize;
            for r in results {
                if r? {
                    count += 1;
                }
            }
            count
        };

        meta.vector_count += new_inserts;
        meta.updated_at = Self::now();
        self.write_namespace_meta(namespace, &meta).await?;

        tracing::debug!(
            namespace = namespace,
            upserted = total,
            "Upserted vectors to object storage"
        );

        Ok(total)
    }

    async fn get(&self, namespace: &NamespaceId, ids: &[VectorId]) -> Result<Vec<Vector>> {
        if ids.is_empty() {
            return Ok(Vec::new());
        }

        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();

        let read_tasks: Vec<_> = ids
            .iter()
            .map(|id| {
                let operator = self.operator.clone();
                let path = Self::vector_path(namespace, id);
                let id = id.clone();
                async move {
                    match operator.read(&path).await {
                        Ok(data) => {
                            let bytes = data.to_vec();
                            if bytes.is_empty() {
                                tracing::warn!(
                                    vector_id = %id,
                                    "Empty vector file detected, skipping"
                                );
                                return Ok(None);
                            }
                            match serde_json::from_slice::<StoredVector>(&bytes) {
                                Ok(stored) => {
                                    let vector: Vector = stored.into();
                                    if !vector.is_expired_at(now) {
                                        Ok(Some(vector))
                                    } else {
                                        Ok(None)
                                    }
                                }
                                Err(e) => {
                                    tracing::warn!(
                                        vector_id = %id,
                                        error = %e,
                                        bytes_len = bytes.len(),
                                        "Corrupted vector file detected, skipping"
                                    );
                                    Ok(None)
                                }
                            }
                        }
                        Err(e) if e.kind() == opendal::ErrorKind::NotFound => Ok(None),
                        Err(e) => Err(DakeraError::Storage(e.to_string())),
                    }
                }
            })
            .collect();

        let results: Vec<Result<Option<Vector>>> = stream::iter(read_tasks)
            .buffer_unordered(s3_concurrent_ops())
            .collect()
            .await;

        let mut vectors = Vec::with_capacity(ids.len());
        for result in results {
            if let Some(v) = result? {
                vectors.push(v);
            }
        }
        Ok(vectors)
    }

    async fn get_all(&self, namespace: &NamespaceId) -> Result<Vec<Vector>> {
        let prefix = Self::namespace_vectors_prefix(namespace);

        let entries = self
            .operator
            .list(&prefix)
            .await
            .map_err(|e| DakeraError::Storage(e.to_string()))?;

        let json_paths: Vec<String> = entries
            .into_iter()
            .filter(|e| e.path().ends_with(".json"))
            .map(|e| e.path().to_string())
            .collect();

        if json_paths.is_empty() {
            return Ok(Vec::new());
        }

        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();

        let results: Vec<Option<Vector>> = stream::iter(json_paths.into_iter().map(|path| {
            let operator = self.operator.clone();
            async move {
                match operator.read(&path).await {
                    Ok(data) => {
                        let bytes = data.to_vec();
                        if let Ok(stored) = serde_json::from_slice::<StoredVector>(&bytes) {
                            let vector: Vector = stored.into();
                            if !vector.is_expired_at(now) {
                                return Some(vector);
                            }
                        }
                        None
                    }
                    Err(e) => {
                        tracing::warn!(path = %path, error = %e, "Failed to read vector");
                        None
                    }
                }
            }
        }))
        .buffer_unordered(s3_concurrent_ops())
        .collect()
        .await;

        Ok(results.into_iter().flatten().collect())
    }

    async fn delete(&self, namespace: &NamespaceId, ids: &[VectorId]) -> Result<usize> {
        if ids.is_empty() {
            return Ok(0);
        }

        let delete_tasks: Vec<_> = ids
            .iter()
            .map(|id| {
                let operator = self.operator.clone();
                let path = Self::vector_path(namespace, id);
                async move {
                    let exists = operator
                        .exists(&path)
                        .await
                        .map_err(|e| DakeraError::Storage(e.to_string()))?;
                    if exists {
                        match operator.delete(&path).await {
                            Ok(_) => Ok(true),
                            Err(e) if e.kind() == opendal::ErrorKind::NotFound => Ok(false),
                            Err(e) => Err(DakeraError::Storage(e.to_string())),
                        }
                    } else {
                        Ok(false)
                    }
                }
            })
            .collect();

        let results: Vec<Result<bool>> = stream::iter(delete_tasks)
            .buffer_unordered(s3_concurrent_ops())
            .collect()
            .await;

        let mut deleted = 0;
        for result in results {
            if result? {
                deleted += 1;
            }
        }

        // Update metadata
        if deleted > 0 {
            if let Some(mut meta) = self.read_namespace_meta(namespace).await? {
                meta.vector_count = meta.vector_count.saturating_sub(deleted);
                meta.updated_at = Self::now();
                self.write_namespace_meta(namespace, &meta).await?;
            }
        }

        Ok(deleted)
    }

    async fn namespace_exists(&self, namespace: &NamespaceId) -> Result<bool> {
        Ok(self.read_namespace_meta(namespace).await?.is_some())
    }

    async fn ensure_namespace(&self, namespace: &NamespaceId) -> Result<()> {
        if self.read_namespace_meta(namespace).await?.is_none() {
            let meta = NamespaceMetadata {
                dimension: None,
                vector_count: 0,
                created_at: Self::now(),
                updated_at: Self::now(),
            };
            self.write_namespace_meta(namespace, &meta).await?;
        }
        Ok(())
    }

    async fn dimension(&self, namespace: &NamespaceId) -> Result<Option<usize>> {
        Ok(self
            .read_namespace_meta(namespace)
            .await?
            .and_then(|m| m.dimension))
    }

    async fn count(&self, namespace: &NamespaceId) -> Result<usize> {
        Ok(self
            .read_namespace_meta(namespace)
            .await?
            .map(|m| m.vector_count)
            .unwrap_or(0))
    }

    async fn list_namespaces(&self) -> Result<Vec<NamespaceId>> {
        let entries = self
            .operator
            .list("namespaces/")
            .await
            .map_err(|e| DakeraError::Storage(e.to_string()))?;

        let mut namespaces = Vec::new();
        for entry in entries {
            let path = entry.path();
            // Extract namespace name from path like "namespaces/myns/"
            if let Some(ns) = path.strip_prefix("namespaces/") {
                let ns = ns.trim_end_matches('/');
                if !ns.is_empty() && !ns.contains('/') {
                    // Only include namespaces that actually have metadata (meta.json).
                    // This filters out ghost directory entries left behind after
                    // namespace deletion on backends where empty directories persist
                    // (e.g., local filesystem, some S3-compatible stores).
                    if self.read_namespace_meta(ns).await?.is_some() {
                        namespaces.push(ns.to_string());
                    }
                }
            }
        }

        Ok(namespaces)
    }

    async fn delete_namespace(&self, namespace: &NamespaceId) -> Result<bool> {
        // Check if namespace exists
        if !self.namespace_exists(namespace).await? {
            return Ok(false);
        }

        // Recursively remove everything under the namespace prefix.
        // This deletes vectors, metadata, indexes, AND directory entries,
        // preventing ghost namespaces from appearing in list_namespaces().
        let prefix = format!("namespaces/{}/", namespace);
        self.operator
            .delete_with(&prefix)
            .recursive(true)
            .await
            .map_err(|e| DakeraError::Storage(e.to_string()))?;

        tracing::debug!(
            namespace = namespace,
            "Deleted namespace from object storage"
        );

        Ok(true)
    }

    async fn cleanup_expired(&self, _namespace: &NamespaceId) -> Result<usize> {
        // Object storage doesn't track TTL internally - cleanup handled by application layer
        // StoredVector doesn't persist TTL fields, so nothing to clean up here
        Ok(0)
    }

    async fn cleanup_all_expired(&self) -> Result<usize> {
        // Object storage doesn't track TTL internally - cleanup handled by application layer
        Ok(0)
    }
}

#[async_trait]
impl IndexStorage for ObjectStorage {
    async fn save_index(
        &self,
        namespace: &NamespaceId,
        index_type: IndexType,
        data: Vec<u8>,
    ) -> Result<()> {
        let path = format!(
            "namespaces/{}/indexes/{}.bin",
            namespace,
            index_type.as_str()
        );
        self.operator
            .write(&path, data)
            .await
            .map_err(|e| DakeraError::Storage(e.to_string()))?;

        tracing::debug!(
            namespace = namespace,
            index_type = index_type.as_str(),
            "Saved index to object storage"
        );
        Ok(())
    }

    async fn load_index(
        &self,
        namespace: &NamespaceId,
        index_type: IndexType,
    ) -> Result<Option<Vec<u8>>> {
        let path = format!(
            "namespaces/{}/indexes/{}.bin",
            namespace,
            index_type.as_str()
        );
        match self.operator.read(&path).await {
            Ok(data) => {
                tracing::debug!(
                    namespace = namespace,
                    index_type = index_type.as_str(),
                    size = data.len(),
                    "Loaded index from object storage"
                );
                Ok(Some(data.to_vec()))
            }
            Err(e) if e.kind() == opendal::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(DakeraError::Storage(e.to_string())),
        }
    }

    async fn delete_index(&self, namespace: &NamespaceId, index_type: IndexType) -> Result<bool> {
        let path = format!(
            "namespaces/{}/indexes/{}.bin",
            namespace,
            index_type.as_str()
        );
        let exists = self
            .operator
            .exists(&path)
            .await
            .map_err(|e| DakeraError::Storage(e.to_string()))?;

        if exists {
            self.operator
                .delete(&path)
                .await
                .map_err(|e| DakeraError::Storage(e.to_string()))?;
            tracing::debug!(
                namespace = namespace,
                index_type = index_type.as_str(),
                "Deleted index from object storage"
            );
        }
        Ok(exists)
    }

    async fn index_exists(&self, namespace: &NamespaceId, index_type: IndexType) -> Result<bool> {
        let path = format!(
            "namespaces/{}/indexes/{}.bin",
            namespace,
            index_type.as_str()
        );
        self.operator
            .exists(&path)
            .await
            .map_err(|e| DakeraError::Storage(e.to_string()))
    }

    async fn list_indexes(&self, namespace: &NamespaceId) -> Result<Vec<IndexType>> {
        let prefix = format!("namespaces/{}/indexes/", namespace);
        let entries = self
            .operator
            .list(&prefix)
            .await
            .map_err(|e| DakeraError::Storage(e.to_string()))?;

        let mut indexes = Vec::new();
        for entry in entries {
            let path = entry.path();
            if path.ends_with(".bin") {
                // Extract index type from filename
                if let Some(filename) = path.strip_prefix(&prefix) {
                    let name = filename.trim_end_matches(".bin");
                    match name {
                        "hnsw" => indexes.push(IndexType::Hnsw),
                        "pq" => indexes.push(IndexType::Pq),
                        "ivf" => indexes.push(IndexType::Ivf),
                        "spfresh" => indexes.push(IndexType::SpFresh),
                        "fulltext" => indexes.push(IndexType::FullText),
                        _ => {} // Ignore unknown index types
                    }
                }
            }
        }

        Ok(indexes)
    }
}

/// Create an OpenDAL operator from configuration without constructing a full ObjectStorage.
/// Useful for lightweight S3 access (e.g., BackupManager metadata persistence).
pub fn create_operator(config: &ObjectStorageConfig) -> Result<Operator> {
    ObjectStorage::build_operator(config)
}

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

    #[tokio::test]
    async fn test_object_storage_memory() {
        let storage = ObjectStorage::memory().unwrap();
        let namespace = "test".to_string();

        // Ensure namespace
        storage.ensure_namespace(&namespace).await.unwrap();
        assert!(storage.namespace_exists(&namespace).await.unwrap());

        // Insert vectors
        let vectors = vec![
            Vector {
                id: "v1".to_string(),
                values: vec![1.0, 2.0, 3.0],
                metadata: None,
                ttl_seconds: None,
                expires_at: None,
            },
            Vector {
                id: "v2".to_string(),
                values: vec![4.0, 5.0, 6.0],
                metadata: Some(serde_json::json!({"key": "value"})),
                ttl_seconds: None,
                expires_at: None,
            },
        ];

        let count = storage.upsert(&namespace, vectors).await.unwrap();
        assert_eq!(count, 2);

        // Get single vector
        let results = storage.get(&namespace, &["v1".to_string()]).await.unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].id, "v1");
        assert_eq!(results[0].values, vec![1.0, 2.0, 3.0]);

        // Get all vectors
        let all = storage.get_all(&namespace).await.unwrap();
        assert_eq!(all.len(), 2);

        // Count
        assert_eq!(storage.count(&namespace).await.unwrap(), 2);

        // Delete
        let deleted = storage
            .delete(&namespace, &["v1".to_string()])
            .await
            .unwrap();
        assert_eq!(deleted, 1);
        assert!(storage
            .get(&namespace, &["v1".to_string()])
            .await
            .unwrap()
            .is_empty());
        assert_eq!(storage.count(&namespace).await.unwrap(), 1);
    }

    #[tokio::test]
    async fn test_object_storage_dimension_mismatch() {
        let storage = ObjectStorage::memory().unwrap();
        let namespace = "test".to_string();
        storage.ensure_namespace(&namespace).await.unwrap();

        // Insert first vector
        let v1 = vec![Vector {
            id: "v1".to_string(),
            values: vec![1.0, 2.0, 3.0],
            metadata: None,
            ttl_seconds: None,
            expires_at: None,
        }];
        storage.upsert(&namespace, v1).await.unwrap();

        // Try to insert vector with different dimension
        let v2 = vec![Vector {
            id: "v2".to_string(),
            values: vec![1.0, 2.0], // Wrong dimension
            metadata: None,
            ttl_seconds: None,
            expires_at: None,
        }];
        let result = storage.upsert(&namespace, v2).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_object_storage_upsert() {
        let storage = ObjectStorage::memory().unwrap();
        let namespace = "test".to_string();
        storage.ensure_namespace(&namespace).await.unwrap();

        // Insert
        let v1 = vec![Vector {
            id: "v1".to_string(),
            values: vec![1.0, 2.0],
            metadata: None,
            ttl_seconds: None,
            expires_at: None,
        }];
        storage.upsert(&namespace, v1).await.unwrap();

        // Upsert (update)
        let v1_updated = vec![Vector {
            id: "v1".to_string(),
            values: vec![3.0, 4.0],
            metadata: None,
            ttl_seconds: None,
            expires_at: None,
        }];
        storage.upsert(&namespace, v1_updated).await.unwrap();

        // Verify update
        let results = storage.get(&namespace, &["v1".to_string()]).await.unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].values, vec![3.0, 4.0]);

        // Count should still be 1
        assert_eq!(storage.count(&namespace).await.unwrap(), 1);
    }

    #[tokio::test]
    async fn test_index_storage() {
        let storage = ObjectStorage::memory().unwrap();
        let namespace = "test_index".to_string();

        // Initially no indexes
        assert!(!storage
            .index_exists(&namespace, IndexType::Hnsw)
            .await
            .unwrap());

        // Save an index
        let index_data = b"fake hnsw index data for testing".to_vec();
        storage
            .save_index(&namespace, IndexType::Hnsw, index_data.clone())
            .await
            .unwrap();

        // Check it exists
        assert!(storage
            .index_exists(&namespace, IndexType::Hnsw)
            .await
            .unwrap());
        assert!(!storage
            .index_exists(&namespace, IndexType::Pq)
            .await
            .unwrap());

        // Load it back
        let loaded = storage
            .load_index(&namespace, IndexType::Hnsw)
            .await
            .unwrap();
        assert!(loaded.is_some());
        assert_eq!(loaded.unwrap(), index_data);

        // Save another index type
        let pq_data = b"fake pq index data".to_vec();
        storage
            .save_index(&namespace, IndexType::Pq, pq_data)
            .await
            .unwrap();

        // List indexes
        let indexes = storage.list_indexes(&namespace).await.unwrap();
        assert_eq!(indexes.len(), 2);
        assert!(indexes.contains(&IndexType::Hnsw));
        assert!(indexes.contains(&IndexType::Pq));

        // Delete index
        let deleted = storage
            .delete_index(&namespace, IndexType::Hnsw)
            .await
            .unwrap();
        assert!(deleted);
        assert!(!storage
            .index_exists(&namespace, IndexType::Hnsw)
            .await
            .unwrap());

        // Delete non-existent
        let deleted = storage
            .delete_index(&namespace, IndexType::Hnsw)
            .await
            .unwrap();
        assert!(!deleted);

        // Load non-existent
        let loaded = storage
            .load_index(&namespace, IndexType::Hnsw)
            .await
            .unwrap();
        assert!(loaded.is_none());
    }

    // ── DAK-5553: parallel upsert correctness ─────────────────────────────────

    fn make_vector(id: &str, dim: usize) -> Vector {
        Vector {
            id: id.to_string(),
            values: vec![0.1; dim],
            metadata: None,
            ttl_seconds: None,
            expires_at: None,
        }
    }

    #[tokio::test]
    async fn test_upsert_batch_parallel_all_new() {
        let storage = ObjectStorage::memory().unwrap();
        let ns = "batch_all_new".to_string();
        storage.ensure_namespace(&ns).await.unwrap();

        // Insert 50 unique vectors in one batch call
        let vectors: Vec<Vector> = (0..50).map(|i| make_vector(&format!("v{i}"), 4)).collect();
        let count = storage.upsert(&ns, vectors).await.unwrap();

        assert_eq!(count, 50);
        assert_eq!(storage.count(&ns).await.unwrap(), 50);
    }

    #[tokio::test]
    async fn test_upsert_batch_parallel_idempotent_count() {
        let storage = ObjectStorage::memory().unwrap();
        let ns = "batch_idempotent".to_string();
        storage.ensure_namespace(&ns).await.unwrap();

        let vectors: Vec<Vector> = (0..10).map(|i| make_vector(&format!("v{i}"), 4)).collect();
        storage.upsert(&ns, vectors.clone()).await.unwrap();

        // Upsert the same IDs again — vector_count must not double
        storage.upsert(&ns, vectors).await.unwrap();
        assert_eq!(storage.count(&ns).await.unwrap(), 10);
    }

    #[tokio::test]
    async fn test_upsert_batch_parallel_empty() {
        let storage = ObjectStorage::memory().unwrap();
        let ns = "batch_empty".to_string();
        storage.ensure_namespace(&ns).await.unwrap();

        let count = storage.upsert(&ns, vec![]).await.unwrap();
        assert_eq!(count, 0);
        assert_eq!(storage.count(&ns).await.unwrap(), 0);
    }

    #[tokio::test]
    async fn test_upsert_batch_parallel_large_batch() {
        // Verify that a batch exceeding the default concurrency window (16) processes all items
        let storage = ObjectStorage::memory().unwrap();
        let ns = "batch_large".to_string();
        storage.ensure_namespace(&ns).await.unwrap();

        let vectors: Vec<Vector> = (0..200).map(|i| make_vector(&format!("v{i}"), 8)).collect();
        let count = storage.upsert(&ns, vectors).await.unwrap();

        assert_eq!(count, 200);
        assert_eq!(storage.count(&ns).await.unwrap(), 200);
    }

    // DAK-6287: the per-vector exists() stat was replaced by a single directory list()
    // to classify insert-vs-update for vector_count. Prove a batch mixing updates of
    // existing ids with new inserts produces the exact count (only new ids increment).
    #[tokio::test]
    async fn test_upsert_count_mixed_insert_update_batch() {
        let storage = ObjectStorage::memory().unwrap();
        let ns = "mixed".to_string();
        storage.ensure_namespace(&ns).await.unwrap();

        // Seed two vectors.
        storage
            .upsert(&ns, vec![make_vector("v0", 4), make_vector("v1", 4)])
            .await
            .unwrap();
        assert_eq!(storage.count(&ns).await.unwrap(), 2);

        // Batch updates v1 (existing) and inserts v2, v3 (new) => count must be 4, not 5.
        storage
            .upsert(
                &ns,
                vec![
                    make_vector("v1", 4),
                    make_vector("v2", 4),
                    make_vector("v3", 4),
                ],
            )
            .await
            .unwrap();
        assert_eq!(storage.count(&ns).await.unwrap(), 4);

        // Re-upserting only existing ids must not change the count.
        storage
            .upsert(&ns, vec![make_vector("v0", 4), make_vector("v3", 4)])
            .await
            .unwrap();
        assert_eq!(storage.count(&ns).await.unwrap(), 4);
    }

    // DAK-6287: exercise the raw-fs fast write path (real Filesystem backend) end to end —
    // it must produce the same on-disk layout the OpenDAL read path expects, keep exact
    // count semantics, and safely fall back for traversal-unsafe ids without escaping root.
    #[tokio::test]
    async fn test_upsert_fs_fast_path_roundtrip() {
        let tmp = tempfile::tempdir().unwrap();
        let storage = ObjectStorage::filesystem(tmp.path().to_str().unwrap()).unwrap();
        let ns = "fsfast".to_string();
        storage.ensure_namespace(&ns).await.unwrap();

        // Insert via the fast path.
        storage
            .upsert(
                &ns,
                vec![
                    make_vector("a", 4),
                    make_vector("b", 4),
                    make_vector("c", 4),
                ],
            )
            .await
            .unwrap();
        assert_eq!(storage.count(&ns).await.unwrap(), 3);

        // Read back through OpenDAL: proves the fast-path files are where the read path looks.
        let got = storage
            .get(&ns, &["a".to_string(), "b".to_string()])
            .await
            .unwrap();
        assert_eq!(got.len(), 2);

        // Update b (existing) + insert d (new) in one batch => count 4, not 5.
        storage
            .upsert(&ns, vec![make_vector("b", 4), make_vector("d", 4)])
            .await
            .unwrap();
        assert_eq!(storage.count(&ns).await.unwrap(), 4);

        // Guard: a batch containing a traversal-unsafe id must not panic or escape the
        // storage root; it falls back to the OpenDAL path and the safe sibling persists.
        storage
            .upsert(
                &ns,
                vec![make_vector("safe1", 4), make_vector("../escape", 4)],
            )
            .await
            .unwrap();
        let got = storage.get(&ns, &["safe1".to_string()]).await.unwrap();
        assert_eq!(got.len(), 1);
        // Nothing was written above the storage root.
        assert!(!tmp.path().parent().unwrap().join("escape.json").exists());
    }

    // DAK-6299: ingest pattern — many small batches into ONE growing namespace. This is the
    // O(N^2) scenario the removed per-batch full-namespace list() created. vector_count must
    // stay exact when classified incrementally per batch (fs fast path: per-file stat).
    // Each batch overlaps the previous one by one id (an update), so only the genuinely-new
    // id may increment the count.
    #[tokio::test]
    async fn test_upsert_incremental_count_growing_namespace_fs() {
        let tmp = tempfile::tempdir().unwrap();
        let storage = ObjectStorage::filesystem(tmp.path().to_str().unwrap()).unwrap();
        let ns = "grow".to_string();
        storage.ensure_namespace(&ns).await.unwrap();

        const BATCHES: usize = 40;
        for b in 0..BATCHES {
            // Batch b writes ids {b, b+1}: id `b` is an update of the prior batch's new id,
            // id `b+1` is the only genuinely new vector => count grows by exactly 1 per batch.
            let new_id = format!("v{}", b + 1);
            let upd_id = format!("v{}", b);
            storage
                .upsert(&ns, vec![make_vector(&upd_id, 4), make_vector(&new_id, 4)])
                .await
                .unwrap();
            // After batch b we have ids v0..=v(b+1) => b+2 distinct vectors.
            assert_eq!(storage.count(&ns).await.unwrap(), b + 2);
        }
        assert_eq!(storage.count(&ns).await.unwrap(), BATCHES + 1);
    }

    // DAK-6299: same incremental-count contract on the OpenDAL path (Memory backend), which
    // classifies inserts via a per-vector exists() rather than the fs stat.
    #[tokio::test]
    async fn test_upsert_incremental_count_growing_namespace_opendal() {
        let storage = ObjectStorage::memory().unwrap();
        let ns = "grow-od".to_string();
        storage.ensure_namespace(&ns).await.unwrap();

        const BATCHES: usize = 30;
        for b in 0..BATCHES {
            let new_id = format!("v{}", b + 1);
            let upd_id = format!("v{}", b);
            storage
                .upsert(&ns, vec![make_vector(&upd_id, 4), make_vector(&new_id, 4)])
                .await
                .unwrap();
            assert_eq!(storage.count(&ns).await.unwrap(), b + 2);
        }
        assert_eq!(storage.count(&ns).await.unwrap(), BATCHES + 1);
    }

    // DAK-6287: per-sub-stage profile of ObjectStorage::upsert on the fs backend.
    //
    // Post-tier (DAKERA_TIERED=1, embedding deferred) the per-stage ingest profile
    // (PR#609) names storage_upsert as the dominant stage (83.6% of the tiered
    // pipeline). This test decomposes that stage into its three per-vector fs ops
    // (exists stat / write-tmp / rename) plus candidate fixes, all at the production
    // concurrency window, so the dominant SUB-stage is provable rather than assumed.
    //
    // Ignored by default (writes thousands of files); run explicitly:
    //   cargo test -p dakera-storage --release profile_upsert_substages -- --ignored --nocapture
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    #[ignore]
    async fn profile_upsert_substages() {
        use std::time::Instant;
        const N: usize = 200;
        const DIM: usize = 1024; // bge-large-en-v1.5
        let conc = s3_concurrent_ops();

        // Realistic payload: 1024-d vector + LoCoMo-shaped metadata.
        let make = |i: usize| -> Vector {
            Vector {
                id: format!("prof-{i}"),
                values: vec![0.0123_f32; DIM],
                metadata: Some(serde_json::json!({
                    "_embedding_kind": "static",
                    "tags": ["dia_1", "speaker_a", "session_3", "2026-06-03"],
                    "importance": 0.7,
                    "session_id": format!("sess-{}", i % 8),
                })),
                ttl_seconds: None,
                expires_at: None,
            }
        };
        let vectors: Vec<Vector> = (0..N).map(make).collect();
        let payload_bytes = serde_json::to_vec(&StoredVector::from(vectors[0].clone()))
            .unwrap()
            .len();

        // --- V0: authoritative current upsert() (real code path) ---
        let tmp0 = tempfile::tempdir().unwrap();
        let s0 = ObjectStorage::filesystem(tmp0.path().to_str().unwrap()).unwrap();
        let ns = "prof".to_string();
        s0.ensure_namespace(&ns).await.unwrap();
        let t = Instant::now();
        let n = s0.upsert(&ns, vectors.clone()).await.unwrap();
        let v0_ms = t.elapsed().as_secs_f64() * 1000.0;
        assert_eq!(n, N);

        // Helper: run a per-vector closure over a fresh fs operator at `conc` and time it.
        async fn timed<F, Fut>(label: &str, n: usize, conc: usize, f: F) -> f64
        where
            F: Fn(usize, Operator) -> Fut,
            Fut: std::future::Future<Output = ()>,
        {
            let tmp = tempfile::tempdir().unwrap();
            let op = ObjectStorage::filesystem(tmp.path().to_str().unwrap())
                .unwrap()
                .operator;
            let t = Instant::now();
            stream::iter(0..n)
                .map(|i| {
                    let op = op.clone();
                    f(i, op)
                })
                .buffer_unordered(conc)
                .collect::<Vec<()>>()
                .await;
            let ms = t.elapsed().as_secs_f64() * 1000.0;
            // keep tmp alive until after timing
            drop(tmp);
            let _ = label;
            ms
        }

        let data: Vec<Vec<u8>> = vectors
            .iter()
            .map(|v| serde_json::to_vec(&StoredVector::from(v.clone())).unwrap())
            .collect();
        let path_of = |i: usize| ObjectStorage::vector_path(&ns, &format!("prof-{i}"));

        // P_exists: 200 stats only (the exists() round-trip upsert does per vector).
        let d = data.clone();
        let p_exists = timed("exists", N, conc, move |i, op| {
            let path = path_of(i);
            let _ = &d;
            async move {
                let _ = op.exists(&path).await;
            }
        })
        .await;

        // P_write_direct: 200 direct writes (no tmp, no rename, no exists).
        let d = data.clone();
        let p_write_direct = timed("write_direct", N, conc, move |i, op| {
            let path = path_of(i);
            let bytes = d[i].clone();
            async move {
                op.write(&path, bytes).await.unwrap();
            }
        })
        .await;

        // P_write_tmp_rename: 200 × (write tmp + rename) — the atomic write, no exists.
        let d = data.clone();
        let p_tmp_rename = timed("write_tmp_rename", N, conc, move |i, op| {
            let path = path_of(i);
            let bytes = d[i].clone();
            async move {
                let seq = TMP_COUNTER.fetch_add(1, Ordering::Relaxed);
                let tmp_path = format!("{path}.tmp.{seq}");
                op.write(&tmp_path, bytes).await.unwrap();
                op.rename(&tmp_path, &path).await.unwrap();
            }
        })
        .await;

        // P_full: exists + write tmp + rename (mirror of current upsert per-vector body).
        let d = data.clone();
        let p_full = timed("exists+tmp+rename", N, conc, move |i, op| {
            let path = path_of(i);
            let bytes = d[i].clone();
            async move {
                let _ = op.exists(&path).await;
                let seq = TMP_COUNTER.fetch_add(1, Ordering::Relaxed);
                let tmp_path = format!("{path}.tmp.{seq}");
                op.write(&tmp_path, bytes).await.unwrap();
                op.rename(&tmp_path, &path).await.unwrap();
            }
        })
        .await;

        // --- Concurrency sweep on the real write pattern: does throughput scale with
        // parallelism (I/O-latency-bound) or plateau (CPU/serialized)? ---
        let mut sweep = Vec::new();
        for &c in &[8usize, 16, 32, 64, 128] {
            let d = data.clone();
            let ms = timed("sweep", N, c, move |i, op| {
                let path = path_of(i);
                let bytes = d[i].clone();
                async move {
                    let seq = TMP_COUNTER.fetch_add(1, Ordering::Relaxed);
                    let tmp_path = format!("{path}.tmp.{seq}");
                    op.write(&tmp_path, bytes).await.unwrap();
                    op.rename(&tmp_path, &path).await.unwrap();
                }
            })
            .await;
            sweep.push((c, ms));
        }

        // --- fsync isolation: raw std::fs write vs write+sync_all at conc 16, to test
        // whether the per-file durability flush is the dominant write cost (the issue's
        // "disable per-write sync" lever). ---
        async fn raw_fs(n: usize, conc: usize, bytes: usize, do_sync: bool) -> f64 {
            let dir = tempfile::tempdir().unwrap();
            let root = dir.path().to_path_buf();
            let payload = vec![0u8; bytes];
            let t = Instant::now();
            stream::iter(0..n)
                .map(|i| {
                    let root = root.clone();
                    let payload = payload.clone();
                    async move {
                        tokio::task::spawn_blocking(move || {
                            use std::io::Write;
                            let p = root.join(format!("f{i}.bin"));
                            let mut f = std::fs::File::create(&p).unwrap();
                            f.write_all(&payload).unwrap();
                            if do_sync {
                                f.sync_all().unwrap();
                            }
                        })
                        .await
                        .unwrap();
                    }
                })
                .buffer_unordered(conc)
                .collect::<Vec<()>>()
                .await;
            let ms = t.elapsed().as_secs_f64() * 1000.0;
            drop(dir);
            ms
        }
        let raw_nosync = raw_fs(N, 16, payload_bytes, false).await;
        let raw_sync = raw_fs(N, 16, payload_bytes, true).await;

        let rate = |ms: f64| N as f64 / (ms / 1000.0);
        eprintln!("\n=== DAK-6287 storage_upsert sub-stage profile (fs backend) ===");
        eprintln!("N={N} dim={DIM} payload={payload_bytes}B concurrency={conc}");
        eprintln!(
            "V0 upsert() REAL path      : {v0_ms:8.2} ms  {:8.1} mem/s",
            rate(v0_ms)
        );
        eprintln!(
            "P_full exists+tmp+rename   : {p_full:8.2} ms  {:8.1} mem/s",
            rate(p_full)
        );
        eprintln!(
            "P_exists (stat only)       : {p_exists:8.2} ms  {:8.1} mem/s",
            rate(p_exists)
        );
        eprintln!(
            "P_write_tmp_rename         : {p_tmp_rename:8.2} ms  {:8.1} mem/s",
            rate(p_tmp_rename)
        );
        eprintln!(
            "P_write_direct             : {p_write_direct:8.2} ms  {:8.1} mem/s",
            rate(p_write_direct)
        );
        eprintln!("--- attribution (of P_full) ---");
        eprintln!("exists share   : {:5.1}%", 100.0 * p_exists / p_full);
        eprintln!("rename overhead: {:5.1}%  (tmp+rename {p_tmp_rename:.1} vs direct {p_write_direct:.1})",
            100.0 * (p_tmp_rename - p_write_direct).max(0.0) / p_full);
        eprintln!("--- concurrency sweep (write tmp+rename) ---");
        for (c, ms) in &sweep {
            eprintln!("conc={c:4} : {ms:8.2} ms  {:8.1} mem/s", rate(*ms));
        }
        eprintln!("--- fsync isolation (raw std::fs, conc=16, {payload_bytes}B) ---");
        eprintln!(
            "write no-sync : {raw_nosync:8.2} ms  {:8.1} mem/s",
            rate(raw_nosync)
        );
        eprintln!(
            "write+sync_all: {raw_sync:8.2} ms  {:8.1} mem/s",
            rate(raw_sync)
        );
        eprintln!(
            "=> fsync cost factor: {:.1}x",
            raw_sync / raw_nosync.max(0.001)
        );
    }
}