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>,
    /// DAK-6289: whether write-path operations must use the tmp-file + rename atomic
    /// publish dance. TRUE only for the local `Filesystem` backend, where a direct
    /// overwrite is non-atomic and a concurrent reader can observe a truncated file
    /// during the write window (O_TRUNC race, DAK-4545). FALSE for `Memory` and every
    /// object store (`S3`/`Azure`/`Gcs`): their `PUT` is atomic and read-after-write
    /// consistent — a reader sees either the prior object or the fully-written new one,
    /// never a torn read. Object stores have no POSIX rename: OpenDAL's S3 service returns
    /// `Unsupported` for `rename` (verified against live MinIO, DAK-6289), so the old
    /// tmp+rename path ALWAYS fell through to its fallback (delete(tmp) + direct write) =
    /// a wasted PUT(tmp) + DELETE(tmp) per object for zero durability benefit. We write
    /// direct to the final key instead.
    needs_tmp_rename: bool,
    /// DAK-7428: sharded per-namespace async write locks. `upsert`/`delete` do a
    /// metadata read-modify-write (`vector_count`/`dimension`); without serialization
    /// two concurrent writers to the same namespace lost-update the count. A fixed
    /// shard array (bounded memory — no per-namespace map growth — and no lock
    /// poisoning) serializes same-namespace writers while letting different namespaces
    /// proceed in parallel. `ObjectStorage` is the fs/S3 flat backend and the tiered
    /// warm/cold tiers, not the default in-memory hot path, so this serialization is a
    /// correctness win at negligible throughput cost. `Arc`-shared so cloned
    /// `ObjectStorage` handles serialize against the *same* shards (the struct
    /// derives `Clone`).
    namespace_locks: std::sync::Arc<Vec<tokio::sync::Mutex<()>>>,
    /// DAK-7428: count of vector files dropped on read because they were empty or
    /// failed to deserialize. A read that silently skips corrupt files returns
    /// fewer vectors than were stored with no signal to the caller — silent data
    /// loss. This counter (shared across cloned handles) makes corruption
    /// observable via [`ObjectStorage::corruption_count`]; the skip is logged at
    /// error level rather than warn so it trips alerting.
    corruption_count: std::sync::Arc<std::sync::atomic::AtomicU64>,
}

/// Number of write-lock shards. Two namespaces may occasionally share a shard
/// (harmless false serialization); collisions are rare relative to distinct
/// namespaces and the array stays tiny and bounded (DAK-7428).
const NAMESPACE_LOCK_SHARDS: usize = 256;

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,
        };
        let needs_tmp_rename = Self::needs_tmp_rename_for(&config);
        Ok(Self {
            operator,
            fs_root,
            needs_tmp_rename,
            namespace_locks: std::sync::Arc::new(
                (0..NAMESPACE_LOCK_SHARDS)
                    .map(|_| tokio::sync::Mutex::new(()))
                    .collect(),
            ),
            corruption_count: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0)),
        })
    }

    /// DAK-7428: total vector files dropped on read due to emptiness/corruption
    /// since process start (shared across cloned handles). Non-zero means reads
    /// have silently returned fewer vectors than were stored — surface it via
    /// metrics/health so the data loss is not invisible.
    pub fn corruption_count(&self) -> u64 {
        self.corruption_count
            .load(std::sync::atomic::Ordering::Relaxed)
    }

    /// DAK-7428: pick the write-lock shard for `namespace`. Stable within the
    /// process (same namespace → same shard), which is all the serialization needs.
    fn namespace_lock(&self, namespace: &str) -> &tokio::sync::Mutex<()> {
        use std::hash::{Hash, Hasher};
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        namespace.hash(&mut hasher);
        let idx = (hasher.finish() as usize) % self.namespace_locks.len();
        &self.namespace_locks[idx]
    }

    /// DAK-6289: whether `config`'s backend needs the tmp-file + rename atomic publish.
    /// TRUE only for the local `Filesystem` (a direct overwrite there is non-atomic and
    /// a concurrent reader can observe a truncated file — the O_TRUNC race, DAK-4545).
    /// FALSE for `Memory` and every object store (`S3`/`Azure`/`Gcs`), whose `PUT` is
    /// atomic and read-after-write consistent, so we write direct to the final key and
    /// avoid OpenDAL's COPY+DELETE rename on S3.
    fn needs_tmp_rename_for(config: &ObjectStorageConfig) -> bool {
        matches!(config, ObjectStorageConfig::Filesystem { .. })
    }

    /// 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;
            }
        }

        // DAK-7428: fsync the namespace vectors directory so the atomic renames
        // above are durable across power loss. Each file's contents are synced
        // before its rename, but the directory entry created by the rename can
        // still be lost on a crash unless the directory itself is fsynced.
        // Best-effort: a directory-fsync failure must not fail an otherwise
        // successful batch write.
        let dir = vectors_dir;
        let _ = tokio::task::spawn_blocking(move || {
            if let Ok(f) = std::fs::File::open(&dir) {
                let _ = f.sync_all();
            }
        })
        .await;

        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.
    ///
    /// On the local `Filesystem` backend (`needs_tmp_rename == true`) this 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, DAK-4545).
    ///
    /// On object stores and memory (`needs_tmp_rename == false`) the backend's `PUT` is
    /// already atomic and read-after-write consistent, so we write direct to the final
    /// key. This is the DAK-6289 fix: OpenDAL's S3 service has no `rename` (returns
    /// `Unsupported`), so the old tmp+rename always fell through to its fallback —
    /// delete(tmp) + direct write — turning each metadata write into PUT(tmp) +
    /// DELETE(tmp) + PUT(final) for zero durability gain. A direct `PUT` still leaves
    /// readers on the previous version until it completes — no torn-read window exists.
    async fn write_atomic(&self, path: &str, data: Vec<u8>) -> Result<()> {
        if !self.needs_tmp_rename {
            // Atomic-PUT backend (S3/Azure/GCS/Memory): write straight to the final key.
            return self
                .operator
                .write(path, data)
                .await
                .map(|_| ())
                .map_err(|e| DakeraError::Storage(e.to_string()));
        }
        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()
    }

    /// Write raw bytes to `path`, using the same atomic-PUT semantics as metadata
    /// writes. Exposed for backup/restore remote DR (`crate::backup`).
    pub async fn put_object(&self, path: &str, data: Vec<u8>) -> Result<()> {
        self.write_atomic(path, data).await
    }

    /// Read raw bytes at `path`. Returns `Ok(None)` when the object does not exist,
    /// mirroring the `NotFound` handling in `read_namespace_meta`. Exposed for
    /// backup/restore remote DR (`crate::backup`).
    pub async fn get_object(&self, path: &str) -> Result<Option<Vec<u8>>> {
        match self.operator.read(path).await {
            Ok(data) => Ok(Some(data.to_vec())),
            Err(e) if e.kind() == opendal::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(DakeraError::Storage(e.to_string())),
        }
    }
}

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

        // DAK-7428: serialize writers to this namespace across the metadata
        // read-modify-write below so two concurrent upserts can't lost-update
        // `vector_count`/`dimension`. Different namespaces are unaffected.
        let _write_guard = self.namespace_lock(namespace).lock().await;

        // 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();
            // DAK-6289: capture the write strategy once (Copy bool) so each per-vector task
            // chooses direct-PUT vs tmp+rename without touching `self`.
            let needs_tmp_rename = self.needs_tmp_rename;
            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()))?;

                        if needs_tmp_rename {
                            // Local filesystem fallback (only reached here for traversal-unsafe
                            // ids; the safe-id common case takes write_vectors_fs above):
                            // tmp+rename avoids the O_TRUNC torn-read race (DAK-4545).
                            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()))?;
                            }
                        } else {
                            // DAK-6289 hot path: S3/Azure/GCS/Memory `PUT` is atomic and
                            // read-after-write consistent. Write direct to the final key.
                            // The old tmp+rename always degraded to PUT(tmp)+DELETE(tmp)+
                            // PUT(final) on S3 (OpenDAL rename => Unsupported → fallback),
                            // i.e. the payload was PUT twice. Direct write is one PUT with
                            // identical atomicity. Recall-neutral: same key, same bytes.
                            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 corruption = self.corruption_count.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() {
                                corruption.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                                tracing::error!(
                                    vector_id = %id,
                                    "Empty vector file detected, skipping (silent data loss — see corruption_count)"
                                );
                                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) => {
                                    corruption.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                                    tracing::error!(
                                        vector_id = %id,
                                        error = %e,
                                        bytes_len = bytes.len(),
                                        "Corrupted vector file detected, skipping (silent data loss — see corruption_count)"
                                    );
                                    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();
            let corruption = self.corruption_count.clone();
            async move {
                match operator.read(&path).await {
                    Ok(data) => {
                        let bytes = data.to_vec();
                        match serde_json::from_slice::<StoredVector>(&bytes) {
                            Ok(stored) => {
                                let vector: Vector = stored.into();
                                if !vector.is_expired_at(now) {
                                    return Some(vector);
                                }
                                None
                            }
                            Err(e) => {
                                // Previously this deserialize failure was dropped with
                                // no log at all — fully invisible data loss (DAK-7428).
                                corruption.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                                tracing::error!(
                                    path = %path,
                                    error = %e,
                                    bytes_len = bytes.len(),
                                    "Corrupted vector file in namespace scan, skipping (silent data loss — see corruption_count)"
                                );
                                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);
        }

        // DAK-7428: serialize with concurrent upsert/delete on this namespace so the
        // `vector_count` read-modify-write below can't lost-update.
        let _write_guard = self.namespace_lock(namespace).lock().await;

        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()))?;

        // Extract candidate namespace names from paths like "namespaces/myns/".
        let candidates: Vec<String> = entries
            .iter()
            .filter_map(|entry| {
                let ns = entry
                    .path()
                    .strip_prefix("namespaces/")?
                    .trim_end_matches('/');
                if !ns.is_empty() && !ns.contains('/') {
                    Some(ns.to_string())
                } else {
                    None
                }
            })
            .collect();

        // Only include namespaces that actually have metadata (meta.json) — this
        // filters ghost directory entries left after deletion on backends where empty
        // directories persist (local fs, some S3-compatible stores). DAK-7428: probe
        // the candidates CONCURRENTLY; this was a sequential await per namespace, i.e.
        // N serial round trips on S3.
        let checked: Vec<Result<Option<String>>> = stream::iter(candidates)
            .map(|ns| async move { Ok(self.read_namespace_meta(&ns).await?.map(|_| ns)) })
            .buffer_unordered(s3_concurrent_ops())
            .collect()
            .await;

        let mut namespaces = Vec::new();
        for r in checked {
            if let Some(ns) = r? {
                namespaces.push(ns);
            }
        }

        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);
    }

    // DAK-7428: a corrupt vector file must be counted (not silently dropped) so
    // the read-path data loss is observable.
    #[tokio::test]
    async fn corrupt_vector_file_is_counted_and_skipped() {
        let storage = ObjectStorage::memory().unwrap();
        let ns = "test".to_string();
        storage.ensure_namespace(&ns).await.unwrap();
        storage
            .upsert(
                &ns,
                vec![Vector {
                    id: "good".to_string(),
                    values: vec![1.0, 2.0],
                    metadata: None,
                    ttl_seconds: None,
                    expires_at: None,
                }],
            )
            .await
            .unwrap();

        // Inject an undeserializable file directly at a vector path.
        let bad_path = ObjectStorage::vector_path(&ns, "bad");
        storage
            .operator
            .write(&bad_path, b"{ not valid json".to_vec())
            .await
            .unwrap();

        assert_eq!(storage.corruption_count(), 0);

        // Direct get of the corrupt id: skipped and counted.
        let got = storage.get(&ns, &["bad".to_string()]).await.unwrap();
        assert!(got.is_empty());
        assert_eq!(storage.corruption_count(), 1);

        // Namespace scan: corrupt file skipped, valid vector survives, counter grows.
        let all = storage.get_all(&ns).await.unwrap();
        assert_eq!(all.len(), 1);
        assert_eq!(all[0].id, "good");
        assert!(storage.corruption_count() >= 2);
    }

    #[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-6289: S3/object-store direct-write path ───────────────────────────

    // The write strategy is selected once at construction. Only the local filesystem
    // needs the tmp+rename O_TRUNC guard; memory and every object store get the
    // direct-PUT fast path (no COPY+DELETE on S3).
    #[test]
    fn test_write_strategy_selection() {
        // Object stores + memory → direct write (atomic PUT, no COPY+DELETE).
        assert!(!ObjectStorage::needs_tmp_rename_for(
            &ObjectStorageConfig::Memory
        ));
        assert!(!ObjectStorage::needs_tmp_rename_for(
            &ObjectStorageConfig::S3 {
                bucket: "dakera".to_string(),
                region: None,
                endpoint: None,
                access_key_id: None,
                secret_access_key: None,
            }
        ));
        assert!(!ObjectStorage::needs_tmp_rename_for(
            &ObjectStorageConfig::Azure {
                container: "c".to_string(),
                account_name: "acct".to_string(),
                account_key: None,
                sas_token: None,
                endpoint: None,
            }
        ));
        assert!(!ObjectStorage::needs_tmp_rename_for(
            &ObjectStorageConfig::Gcs {
                bucket: "dakera".to_string(),
                credential_path: None,
                endpoint: None,
            }
        ));
        // Local filesystem → tmp+rename (O_TRUNC torn-read guard).
        assert!(ObjectStorage::needs_tmp_rename_for(
            &ObjectStorageConfig::Filesystem {
                root: "/tmp/x".to_string(),
            }
        ));
        // Constructed instances agree with the classifier.
        assert!(!ObjectStorage::memory().unwrap().needs_tmp_rename);
        let tmp = tempfile::tempdir().unwrap();
        assert!(
            ObjectStorage::filesystem(tmp.path().to_str().unwrap())
                .unwrap()
                .needs_tmp_rename
        );
    }

    // End-to-end exercise of the direct-write branch (needs_tmp_rename == false) on a
    // non-filesystem backend: insert, overwrite-in-place, and a mixed insert/update
    // batch must all round-trip with exact count semantics — proving the removal of
    // tmp+rename keeps the read/recall path and vector_count classification intact.
    #[tokio::test]
    async fn test_upsert_direct_write_path_roundtrip() {
        let storage = ObjectStorage::memory().unwrap();
        assert!(!storage.needs_tmp_rename, "memory must take direct path");
        let ns = "direct".to_string();
        storage.ensure_namespace(&ns).await.unwrap();

        // Insert two new vectors via the direct path.
        storage
            .upsert(&ns, vec![make_vector("a", 4), make_vector("b", 4)])
            .await
            .unwrap();
        assert_eq!(storage.count(&ns).await.unwrap(), 2);

        // Overwrite `a` in place with new values (direct PUT to the final key, no rename).
        let a_updated = Vector {
            id: "a".to_string(),
            values: vec![9.0, 9.0, 9.0, 9.0],
            metadata: None,
            ttl_seconds: None,
            expires_at: None,
        };
        storage.upsert(&ns, vec![a_updated]).await.unwrap();
        let got = storage.get(&ns, &["a".to_string()]).await.unwrap();
        assert_eq!(got.len(), 1);
        assert_eq!(got[0].values, vec![9.0, 9.0, 9.0, 9.0]);
        // Overwrite must NOT increment the count.
        assert_eq!(storage.count(&ns).await.unwrap(), 2);

        // Mixed batch: update `b`, insert `c` → count 3, not 4.
        storage
            .upsert(&ns, vec![make_vector("b", 4), make_vector("c", 4)])
            .await
            .unwrap();
        assert_eq!(storage.count(&ns).await.unwrap(), 3);
    }

    // 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)
        );
    }

    // DAK-6289: per-sub-stage profile of ObjectStorage::upsert against a REAL S3/MinIO
    // endpoint. Prod runs DAKERA_STORAGE=s3 (dakera-deploy docker-compose + k8s), whose
    // write path is network-bound and a different shape from the fs 83.6% profiled above.
    //
    // MEASURED (this harness, live MinIO): OpenDAL's S3 service has NO rename capability —
    // `rename` returns Unsupported (a fast CLIENT-side error, not a network COPY). So the OLD
    // upsert always hit its fallback: write(tmp) → rename fails → delete(tmp) + write(final).
    // Net OLD per vector = exists(HEAD) + PUT(tmp) + DELETE(tmp) + PUT(final) — payload PUT
    // TWICE. The fix writes direct = exists(HEAD) + PUT(final). This decomposes the path so
    // the wasted PUT(tmp)+DELETE(tmp) is PROVABLE, and reports BEFORE vs AFTER on one bucket:
    //   - V0            : the REAL upsert() (now the direct-PUT path) end to end
    //   - P_full_old    : exists + write(tmp) + rename-with-fallback (the real BEFORE path)
    //   - P_direct_new  : exists + write(final key)                                  ← AFTER
    //   - P_exists      : HEAD only
    //   - P_put         : PUT only
    //   - P_waste       : write(tmp) + delete(tmp) — the wasted ops the fix removes
    //   - concurrency sweep on the direct-PUT pattern
    //
    // Ignored by default (needs a live endpoint + writes objects). Stand up MinIO and run:
    //   docker run -d -p 9000:9000 -e MINIO_ROOT_USER=minioadmin \
    //     -e MINIO_ROOT_PASSWORD=minioadmin minio/minio server /data
    //   # create bucket `dakera-prof` (mc mb), then:
    //   DAKERA_PROFILE_S3_ENDPOINT=http://127.0.0.1:9000 \
    //   DAKERA_PROFILE_S3_BUCKET=dakera-prof \
    //   AWS_ACCESS_KEY_ID=minioadmin AWS_SECRET_ACCESS_KEY=minioadmin \
    //   DAKERA_PROFILE_S3_REGION=us-east-1 \
    //   cargo test -p dakera-storage --release profile_upsert_substages_s3 -- --ignored --nocapture
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    #[ignore]
    async fn profile_upsert_substages_s3() {
        use std::time::Instant;

        let endpoint = match std::env::var("DAKERA_PROFILE_S3_ENDPOINT") {
            Ok(e) => e,
            Err(_) => {
                eprintln!(
                    "SKIP profile_upsert_substages_s3: set DAKERA_PROFILE_S3_ENDPOINT + \
                     DAKERA_PROFILE_S3_BUCKET (+ AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY) \
                     to a live MinIO/S3 endpoint to run this profile."
                );
                return;
            }
        };
        let bucket = std::env::var("DAKERA_PROFILE_S3_BUCKET")
            .expect("DAKERA_PROFILE_S3_BUCKET required for S3 profile");
        let region = std::env::var("DAKERA_PROFILE_S3_REGION").ok();
        let access_key_id = std::env::var("AWS_ACCESS_KEY_ID").ok();
        let secret_access_key = std::env::var("AWS_SECRET_ACCESS_KEY").ok();
        let n: usize = std::env::var("DAKERA_PROFILE_N")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(100);
        const DIM: usize = 1024; // bge-large-en-v1.5
        let conc = s3_concurrent_ops();

        let s3_cfg = ObjectStorageConfig::S3 {
            bucket: bucket.clone(),
            region: region.clone(),
            endpoint: Some(endpoint.clone()),
            access_key_id: access_key_id.clone(),
            secret_access_key: secret_access_key.clone(),
        };

        // 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 data: Vec<Vec<u8>> = vectors
            .iter()
            .map(|v| serde_json::to_vec(&StoredVector::from(v.clone())).unwrap())
            .collect();
        let payload_bytes = data[0].len();

        // --- V0: authoritative current upsert() (the shipped direct-PUT path) ---
        let s0 = ObjectStorage::new(s3_cfg.clone()).unwrap();
        assert!(!s0.needs_tmp_rename, "S3 must take the direct-PUT path");
        let ns = "locomo-bench-prof-v0".to_string();
        let _ = s0.delete_namespace(&ns).await; // clean slate
        s0.ensure_namespace(&ns).await.unwrap();
        let t = Instant::now();
        let written = s0.upsert(&ns, vectors.clone()).await.unwrap();
        let v0_ms = t.elapsed().as_secs_f64() * 1000.0;
        assert_eq!(written, n);

        // Helper: run a per-vector closure over the S3 operator at `conc`, time it.
        let op = ObjectStorage::build_operator(&s3_cfg).unwrap();
        async fn timed<F, Fut>(op: &Operator, n: usize, conc: usize, f: F) -> f64
        where
            F: Fn(usize, Operator) -> Fut,
            Fut: std::future::Future<Output = ()>,
        {
            let t = Instant::now();
            stream::iter(0..n)
                .map(|i| f(i, op.clone()))
                .buffer_unordered(conc)
                .collect::<Vec<()>>()
                .await;
            t.elapsed().as_secs_f64() * 1000.0
        }
        // Distinct key prefix per phase so phases don't interfere. A nested `fn` (not a
        // closure) so the `move` timing closures below can call it without capturing it.
        fn key(phase: &str, i: usize) -> String {
            format!("namespaces/locomo-bench-prof-{phase}/vectors/prof-{i}.json")
        }

        // P_exists: n HEAD requests.
        let p_exists = timed(&op, n, conc, |i, op| {
            let path = key("exists", i);
            async move {
                let _ = op.exists(&path).await;
            }
        })
        .await;

        // P_put: n direct PUTs (no exists, no tmp, no rename).
        let d = data.clone();
        let p_put = timed(&op, n, conc, move |i, op| {
            let path = key("put", i);
            let bytes = d[i].clone();
            async move {
                op.write(&path, bytes).await.unwrap();
            }
        })
        .await;

        // P_waste: the wasted ops the fix removes — write(tmp) + delete(tmp). On S3,
        // OpenDAL has NO rename capability (`rename` => Unsupported, a fast CLIENT-side
        // error, NO network COPY), so the OLD upsert always fell through to its fallback:
        // delete(tmp) + write(final). Net OLD per vector = PUT(tmp) + DELETE(tmp) + PUT(final)
        // — the payload is PUT twice. This phase isolates the PUT(tmp)+DELETE(tmp) the fix drops.
        let d = data.clone();
        let p_waste = timed(&op, n, conc, move |i, op| {
            let tmp = format!("{}.tmp", key("waste", i));
            let bytes = d[i].clone();
            async move {
                op.write(&tmp, bytes).await.unwrap();
                op.delete(&tmp).await.unwrap();
            }
        })
        .await;

        // P_full_old: the EXACT old per-vector body — exists + write(tmp) + rename-with-fallback.
        // On S3 the rename errors (Unsupported), so this exercises the real fallback path
        // (delete(tmp) + write(final)) that prod actually ran. This is the true BEFORE cost.
        let d = data.clone();
        let p_full_old = timed(&op, n, conc, move |i, op| {
            let path = key("old", i);
            let tmp = format!("{path}.tmp");
            let bytes = d[i].clone();
            async move {
                let _ = op.exists(&path).await;
                let rename_ok = async {
                    op.write(&tmp, bytes.clone()).await?;
                    op.rename(&tmp, &path).await?;
                    Ok::<(), opendal::Error>(())
                }
                .await
                .is_ok();
                if !rename_ok {
                    let _ = op.delete(&tmp).await;
                    op.write(&path, bytes).await.unwrap();
                }
            }
        })
        .await;

        // P_direct_new: exists + write(final key)  (the AFTER per-vector body).
        let d = data.clone();
        let p_direct_new = timed(&op, n, conc, move |i, op| {
            let path = key("new", i);
            let bytes = d[i].clone();
            async move {
                let _ = op.exists(&path).await;
                op.write(&path, bytes).await.unwrap();
            }
        })
        .await;

        // Concurrency sweep on the shipped direct-PUT pattern.
        let mut sweep = Vec::new();
        for &c in &[8usize, 16, 32, 64] {
            let d = data.clone();
            let ms = timed(&op, n, c, move |i, op| {
                let path = format!("namespaces/locomo-bench-prof-sweep{c}/vectors/prof-{i}.json");
                let bytes = d[i].clone();
                async move {
                    op.write(&path, bytes).await.unwrap();
                }
            })
            .await;
            sweep.push((c, ms));
        }

        let rate = |ms: f64| n as f64 / (ms / 1000.0);
        eprintln!("\n=== DAK-6289 storage_upsert sub-stage profile (S3/MinIO backend) ===");
        eprintln!("endpoint={endpoint} bucket={bucket} N={n} dim={DIM} payload={payload_bytes}B concurrency={conc}");
        eprintln!(
            "V0 upsert() REAL (direct PUT): {v0_ms:8.2} ms  {:8.1} mem/s",
            rate(v0_ms)
        );
        eprintln!(
            "P_full_old exists+PUT+rename : {p_full_old:8.2} ms  {:8.1} mem/s   <- BEFORE",
            rate(p_full_old)
        );
        eprintln!(
            "P_direct_new exists+PUT      : {p_direct_new:8.2} ms  {:8.1} mem/s   <- AFTER",
            rate(p_direct_new)
        );
        eprintln!(
            "P_exists (HEAD only)         : {p_exists:8.2} ms  {:8.1} mem/s",
            rate(p_exists)
        );
        eprintln!(
            "P_put (direct PUT only)      : {p_put:8.2} ms  {:8.1} mem/s",
            rate(p_put)
        );
        eprintln!(
            "P_waste (PUT tmp + DELETE)   : {p_waste:8.2} ms  {:8.1} mem/s  (the ops the fix drops)",
            rate(p_waste)
        );
        eprintln!("--- attribution (S3 rename is Unsupported → old path = PUT(tmp)+DELETE(tmp)+PUT(final)) ---");
        eprintln!(
            "wasted-op share of old body  : {:5.1}%  ({p_waste:.1}ms of {p_full_old:.1}ms)",
            100.0 * p_waste / p_full_old.max(0.001)
        );
        eprintln!(
            "=> direct-PUT speedup (old/new): {:.2}x  (end-to-end V0 vs P_full_old: {:.2}x)",
            p_full_old / p_direct_new.max(0.001),
            p_full_old / v0_ms.max(0.001)
        );
        eprintln!("--- concurrency sweep (direct PUT) ---");
        for (c, ms) in &sweep {
            eprintln!("conc={c:4} : {ms:8.2} ms  {:8.1} mem/s", rate(*ms));
        }

        // Best-effort cleanup of every profile object written this run (DAK-2407 bench
        // hygiene). The decomposition phases write raw vector objects with no meta.json,
        // so delete_namespace() would skip them — recursively delete the prefixes directly.
        let mut prefixes: Vec<String> = ["v0", "exists", "put", "waste", "old", "new"]
            .iter()
            .map(|p| format!("namespaces/locomo-bench-prof-{p}/"))
            .collect();
        for &c in &[8usize, 16, 32, 64] {
            prefixes.push(format!("namespaces/locomo-bench-prof-sweep{c}/"));
        }
        for prefix in prefixes {
            let _ = op.delete_with(&prefix).recursive(true).await;
        }
    }
}