velesdb-core 7.0.0

//! Tests for `HnswIndex` (extracted from index.rs for maintainability)
#![allow(
    clippy::similar_names, // Reason: clippy 1.90 flags idiomatic test bindings (dir/dim, ids/idx)
    clippy::cast_precision_loss,
    clippy::cast_sign_loss,
    clippy::cast_possible_truncation,
    clippy::redundant_closure_for_method_calls
)]

use super::index::{HnswIndex, VacuumError};
use super::params::{HnswParams, SearchQuality};
use crate::distance::DistanceMetric;
use crate::index::VectorIndex;

// =========================================================================
// Parity test for issue #694
// HnswIndex::search_batch_parallel must use ef_search_for_scale on the
// HNSW-only fast path so that single-query and batch-query paths produce
// the same ef value for the same quality profile when dataset_size > 10K.
//
// Note: this test documents and locks in the parity *contract* — that single
// and batch paths return the same top-k for the same quality. With uniformly
// random data the contract is mostly upheld even pre-fix because the candidate
// pool is plentiful at any reasonable ef. Adversarial data (e.g. SIFT1M with
// many near-equidistant points) would surface the asymmetry more clearly.
// We accept the limitation here: the fix is a coherence guarantee, not a
// recall fix; the goal is that future refactors of either path cannot drift
// out of parity without this assertion failing.
// =========================================================================

#[test]
fn test_batch_search_matches_single_query_on_large_dataset_issue_694() {
    use rand::rngs::StdRng;
    use rand::{RngExt, SeedableRng};

    // Arrange: 40K-vector index — strictly above the 10K threshold where
    // ef_search_for_scale starts boosting the ef value. Below 10K both
    // ef_search and ef_search_for_scale return the same number, so the
    // asymmetry only manifests above the threshold.
    //
    // Why 40K + Fast quality: at 40K the scale factor is sqrt(4)=2 (capped),
    // so Fast's base ef doubles with scaling — a 2x change. That gives
    // the candidate set enough wiggle room for the unscaled batch path
    // (pre-fix) to return a different top-k from the scaled single-query path.
    // Below 10K or at smaller scale factors the effect is too small to
    // produce visible top-k divergence on uniformly random data.
    let dim = 16_usize;
    let n = 40_000_usize;
    let mut rng = StdRng::seed_from_u64(0x00C0_FFEE);

    let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();
    let vectors: Vec<(u64, Vec<f32>)> = (0..n)
        .map(|id| {
            let v: Vec<f32> = (0..dim).map(|_| rng.random::<f32>() - 0.5).collect();
            (id as u64, v)
        })
        .collect();
    let refs: Vec<(u64, &[f32])> = vectors.iter().map(|(id, v)| (*id, v.as_slice())).collect();
    index.insert_batch_parallel(refs);
    assert_eq!(index.len(), n);

    // 8 random query vectors
    let queries: Vec<Vec<f32>> = (0..8)
        .map(|_| (0..dim).map(|_| rng.random::<f32>() - 0.5).collect())
        .collect();
    let query_refs: Vec<&[f32]> = queries.iter().map(Vec::as_slice).collect();

    let k = 10_usize;
    // Fast, the lowest named preset on this path, magnifies the asymmetry:
    // the pre-fix batch path ran at the base ef while the single-query path
    // scaled it with the index size. Higher bases (Balanced, Accurate) already
    // saturate the candidate space and hide the bug.
    let quality = SearchQuality::Fast;

    // Act: run both paths
    let single: Vec<Vec<u64>> = queries
        .iter()
        .map(|q| {
            let r = index.search_with_quality(q.as_slice(), k, quality).unwrap();
            r.iter().map(|x| x.id).collect()
        })
        .collect();

    let batch_results = index
        .search_batch_parallel(&query_refs, k, quality)
        .expect("batch search should succeed");
    let batch: Vec<Vec<u64>> = batch_results
        .iter()
        .map(|r| r.iter().map(|x| x.id).collect())
        .collect();

    // Assert: result IDs must match per-query.
    //
    // Pre-fix (issue #694): batch used ef_search(k), single used
    // ef_search_for_scale(k, len). At this test's n=40_000, single ran at
    // twice the base ef and batch at the base ef, so their candidate sets
    // differed and so did their top-k.
    //
    // Post-fix: both paths call ef_search_for_scale(k, self.len()), so the
    // candidate sets are identical and the result lists are identical.
    for (i, (s, b)) in single.iter().zip(batch.iter()).enumerate() {
        assert_eq!(
            s, b,
            "query {i}: batch and single results must match after #694 fix.\nsingle={s:?}\nbatch={b:?}"
        );
    }
}

// =========================================================================
// TDD Tests - Written BEFORE implementation (RED phase)
// =========================================================================

// -------------------------------------------------------------------------
// Vacuum / Maintenance Tests
// -------------------------------------------------------------------------

#[test]
fn test_tombstone_count_empty_index() {
    // Arrange
    let index = HnswIndex::new(64, DistanceMetric::Cosine).unwrap();

    // Act & Assert
    assert_eq!(index.tombstone_count(), 0);
    assert!((index.tombstone_ratio() - 0.0).abs() < f64::EPSILON);
    assert!(!index.needs_vacuum());
}

#[test]
fn test_tombstone_count_after_deletions() {
    // Arrange
    let index = HnswIndex::new(64, DistanceMetric::Cosine).unwrap();

    // Insert 10 vectors
    for i in 0..10 {
        let v: Vec<f32> = (0..64).map(|j| (i + j) as f32 * 0.01).collect();
        index.insert(i as u64, &v);
    }

    // Delete 3 vectors (30%)
    index.remove(1);
    index.remove(3);
    index.remove(5);

    // Assert
    assert_eq!(index.len(), 7);
    assert_eq!(index.tombstone_count(), 3);
    assert!((index.tombstone_ratio() - 0.3).abs() < 0.01);
    assert!(index.needs_vacuum()); // > 20% threshold
}

#[test]
fn test_vacuum_rebuilds_index() {
    // Arrange
    let index = HnswIndex::new(64, DistanceMetric::Cosine).unwrap();

    // Insert 20 vectors
    for i in 0..20 {
        let v: Vec<f32> = (0..64).map(|j| (i + j) as f32 * 0.01).collect();
        index.insert(i as u64, &v);
    }

    // Delete 10 vectors (50% tombstones)
    for i in 0..10 {
        index.remove(i as u64);
    }

    assert_eq!(index.len(), 10);
    assert!(index.needs_vacuum());

    // Act
    let result = index.vacuum();

    // Assert
    assert!(result.is_ok());
    assert_eq!(result.unwrap(), 10);
    assert_eq!(index.len(), 10);
    assert_eq!(index.tombstone_count(), 0);
    assert!(!index.needs_vacuum());
}

#[test]
fn test_vacuum_preserves_search_results() {
    // Arrange
    let index = HnswIndex::new(64, DistanceMetric::Cosine).unwrap();

    // Insert vectors with known patterns
    for i in 0..50 {
        let v: Vec<f32> = (0..64).map(|j| (i * 100 + j) as f32 * 0.001).collect();
        index.insert(i as u64, &v);
    }

    // Delete some vectors
    for i in 0..25 {
        index.remove(i as u64);
    }

    // Query before vacuum
    let query: Vec<f32> = (0..64).map(|j| (30 * 100 + j) as f32 * 0.001).collect();
    let _results_before = index.search(&query, 5);

    // Act
    let _ = index.vacuum();

    // Assert - search still works and returns similar results
    let results_after = index.search(&query, 5);
    assert_eq!(results_after.len(), 5);
    // Results should include vectors 25-49 (the remaining ones)
    for sr in &results_after {
        assert!(sr.id >= 25 && sr.id < 50);
    }
}

#[test]
fn test_drop_after_vacuum_and_reload_is_safe() {
    use tempfile::tempdir;

    let dir = tempdir().expect("Failed to create temp dir");
    {
        let index = HnswIndex::new(32, DistanceMetric::Euclidean).unwrap();

        for i in 0u64..120 {
            let vector: Vec<f32> = (0..32).map(|j| (i + j as u64) as f32 * 0.01).collect();
            index.insert(i, &vector);
        }
        for i in 0u64..40 {
            index.remove(i);
        }

        let rebuilt = index.vacuum().expect("vacuum should succeed");
        assert_eq!(rebuilt, 80);
        index.save(dir.path()).expect("Failed to save");
    } // drop after ManuallyDrop replacement path

    let loaded = HnswIndex::load(dir.path(), 32, DistanceMetric::Euclidean)
        .expect("Failed to load after vacuum+drop");
    let query = vec![0.42; 32];
    let results = loaded.search(&query, 10);
    assert!(!results.is_empty(), "Loaded index should remain searchable");
}

#[test]
fn test_vacuum_fails_with_fast_insert_mode() {
    // Arrange
    let index = HnswIndex::new_fast_insert(64, DistanceMetric::Cosine).unwrap();

    for i in 0..10 {
        let v: Vec<f32> = (0..64).map(|j| (i + j) as f32 * 0.01).collect();
        index.insert(i as u64, &v);
    }

    // Act
    let result = index.vacuum();

    // Assert
    assert!(result.is_err());
    assert_eq!(result.unwrap_err(), VacuumError::VectorStorageDisabled);
}

#[test]
fn test_vacuum_empty_index() {
    // Arrange
    let index = HnswIndex::new(64, DistanceMetric::Cosine).unwrap();

    // Act
    let result = index.vacuum();

    // Assert
    assert!(result.is_ok());
    assert_eq!(result.unwrap(), 0);
}

// -------------------------------------------------------------------------
// Basic Index Tests
// -------------------------------------------------------------------------

#[test]
fn test_hnsw_new_creates_empty_index() {
    // Arrange & Act
    let index = HnswIndex::new(768, DistanceMetric::Cosine).unwrap();

    // Assert
    assert!(index.is_empty());
    assert_eq!(index.len(), 0);
    assert_eq!(index.dimension(), 768);
    assert_eq!(index.metric(), DistanceMetric::Cosine);
}

#[test]
fn test_hnsw_new_turbo_mode() {
    // TDD: Turbo mode uses aggressive params for max insert throughput
    // Target: 5k+ vec/s (vs ~2k/s with auto params)
    let index = HnswIndex::new_turbo(64, DistanceMetric::Cosine).unwrap();

    // Insert vectors - should be faster than standard mode
    for i in 0..100 {
        let v: Vec<f32> = (0..64).map(|j| (i + j) as f32 * 0.01).collect();
        index.insert(i as u64, &v);
    }

    // Assert - basic functionality works
    assert_eq!(index.len(), 100);

    // Search should still work (turbo trades recall for build speed)
    let query: Vec<f32> = (0..64).map(|j| j as f32 * 0.01).collect();
    let results = index.search(&query, 10);
    assert!(!results.is_empty()); // At least some results
}

#[test]
fn test_hnsw_new_fast_insert_mode() {
    // Arrange & Act - fast insert mode turns exact-distance features off
    let index = HnswIndex::new_fast_insert(64, DistanceMetric::Cosine).unwrap();

    // Insert vectors
    for i in 0..100 {
        let v: Vec<f32> = (0..64).map(|j| (i + j) as f32 * 0.01).collect();
        index.insert(i as u64, &v);
    }

    // Assert - basic functionality works
    assert_eq!(index.len(), 100);

    // Search should still work (uses HNSW approximate search)
    let query: Vec<f32> = (0..64).map(|j| j as f32 * 0.01).collect();
    let results = index.search(&query, 10);
    assert_eq!(results.len(), 10);
}

#[test]
fn test_hnsw_insert_single_vector() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    let vector = vec![1.0, 0.0, 0.0];

    // Act
    index.insert(1, &vector);

    // Assert
    assert_eq!(index.len(), 1);
    assert!(!index.is_empty());
}

#[test]
fn test_hnsw_insert_multiple_vectors() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();

    // Act
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.0, 1.0, 0.0]);
    index.insert(3, &[0.0, 0.0, 1.0]);

    // Assert
    assert_eq!(index.len(), 3);
}

#[test]
fn test_hnsw_search_returns_k_nearest() {
    // Arrange - use more vectors to make HNSW more stable
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.9, 0.1, 0.0]); // Similar to 1
    index.insert(3, &[0.0, 1.0, 0.0]); // Different
    index.insert(4, &[0.8, 0.2, 0.0]); // Similar to 1
    index.insert(5, &[0.0, 0.0, 1.0]); // Different

    // Act
    let results = index.search(&[1.0, 0.0, 0.0], 3);

    // Assert - HNSW may return fewer than k results with small datasets
    assert!(
        !results.is_empty() && results.len() <= 3,
        "Should return 1-3 results, got {}",
        results.len()
    );
    // First result should be exact match (id=1) - verify it's in top results
    let top_ids: Vec<u64> = results.iter().map(|sr| sr.id).collect();
    assert!(top_ids.contains(&1), "Exact match should be in top results");
}

#[test]
fn test_hnsw_search_empty_index() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();

    // Act
    let results = index.search(&[1.0, 0.0, 0.0], 10);

    // Assert
    assert!(results.is_empty());
}

#[test]
fn test_hnsw_remove_existing_vector() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.0, 1.0, 0.0]);

    // Act
    let removed = index.remove(1);

    // Assert
    assert!(removed);
    assert_eq!(index.len(), 1);
}

#[test]
fn test_hnsw_remove_nonexistent_vector() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);

    // Act
    let removed = index.remove(999);

    // Assert
    assert!(!removed);
    assert_eq!(index.len(), 1);
}

#[test]
fn test_hnsw_euclidean_metric() {
    // Arrange - use more vectors to avoid HNSW flakiness with tiny datasets
    let index = HnswIndex::new(3, DistanceMetric::Euclidean).unwrap();
    index.insert(1, &[0.0, 0.0, 0.0]);
    index.insert(2, &[1.0, 0.0, 0.0]); // Distance 1
    index.insert(3, &[3.0, 4.0, 0.0]); // Distance 5
    index.insert(4, &[2.0, 0.0, 0.0]); // Distance 2
    index.insert(5, &[0.5, 0.5, 0.0]); // Distance ~0.7

    // Act
    let results = index.search(&[0.0, 0.0, 0.0], 3);

    // Assert - at least get some results, first should be closest
    assert!(!results.is_empty(), "Should return results");
    assert_eq!(results[0].id, 1, "Closest should be exact match");
}

#[test]
fn test_hnsw_dot_product_metric() {
    // Arrange - Use normalized positive vectors for dot product
    // DistDot in hnsw_rs requires non-negative dot products
    // Use more vectors to avoid HNSW flakiness with tiny datasets
    let index = HnswIndex::new(3, DistanceMetric::DotProduct).unwrap();

    // Insert vectors with distinct dot products when queried with [1,0,0]
    index.insert(1, &[1.0, 0.0, 0.0]); // dot=1.0 with query
    index.insert(2, &[0.5, 0.5, 0.5]); // dot=0.5 with query
    index.insert(3, &[0.1, 0.1, 0.1]); // dot=0.1 with query
    index.insert(4, &[0.8, 0.2, 0.0]); // dot=0.8 with query
    index.insert(5, &[0.3, 0.3, 0.3]); // dot=0.3 with query

    // Act - Query with unit vector x
    let query = [1.0, 0.0, 0.0];
    let results = index.search(&query, 3);

    // Assert - at least get some results, first should have highest dot product
    assert!(!results.is_empty(), "Should return results");
    assert_eq!(results[0].id, 1, "Highest dot product should be first");
}

#[test]
fn test_hnsw_insert_wrong_dimension_is_noop() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();

    // Act - dimension mismatch is silently rejected via VectorIndex trait
    index.insert(1, &[1.0, 0.0]); // Wrong dimension

    // Assert - no vector was inserted
    assert_eq!(index.len(), 0, "Wrong-dimension insert should be a no-op");
}

#[test]
fn test_hnsw_search_wrong_dimension_returns_error() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);

    // Act - dimension mismatch returns DimensionMismatch error
    let result = index.search_with_quality(&[1.0, 0.0], 10, SearchQuality::Balanced);
    assert!(result.is_err(), "Should return error on dimension mismatch");

    // The VectorIndex::search trait adapter returns empty Vec on error
    let results = index.search(&[1.0, 0.0], 10);
    assert!(
        results.is_empty(),
        "Trait search should return empty on mismatch"
    );
}

#[test]
fn test_hnsw_duplicate_insert_upserts_vector() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);

    // Act - Insert with same ID performs upsert (replaces the vector)
    index.insert(1, &[0.0, 1.0, 0.0]);

    // Assert
    assert_eq!(index.len(), 1); // Still only one entry

    // Verify the UPDATED vector is indexed (not the original)
    let results = index.search(&[0.0, 1.0, 0.0], 1);
    assert_eq!(results.len(), 1);
    assert_eq!(results[0].id, 1);
    // Score should be ~1.0 (only 1 vector in index, exact match expected)
    assert!(
        results[0].score > 0.9,
        "Updated vector should be indexed, got score {}",
        results[0].score,
    );
}

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

    // Arrange
    let index = Arc::new(HnswIndex::new(3, DistanceMetric::Cosine).unwrap());
    let mut handles = vec![];

    // Act - Insert from multiple threads (unique IDs)
    for i in 0..10 {
        let index_clone = Arc::clone(&index);
        handles.push(thread::spawn(move || {
            #[allow(clippy::cast_precision_loss)]
            index_clone.insert(i, &[i as f32, 0.0, 0.0]);
        }));
    }

    for handle in handles {
        handle.join().expect("Thread panicked");
    }

    // Set searching mode after parallel insertions (required by hnsw_rs)
    index.set_searching_mode();

    // Assert
    assert_eq!(index.len(), 10);
}

#[test]
fn test_hnsw_persistence() {
    use tempfile::tempdir;

    // Arrange
    let dir = tempdir().unwrap();
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.0, 1.0, 0.0]);

    // Act - Save
    index.save(dir.path()).unwrap();

    // Act - Load
    let loaded_index = HnswIndex::load(dir.path(), 3, DistanceMetric::Cosine).unwrap();

    // Assert
    assert_eq!(loaded_index.len(), 2);
    assert_eq!(loaded_index.dimension(), 3);
    assert_eq!(loaded_index.metric(), DistanceMetric::Cosine);

    // Verify search works on loaded index
    let results = loaded_index.search(&[1.0, 0.0, 0.0], 1);
    assert_eq!(results.len(), 1);
    assert_eq!(results[0].id, 1);
}

#[test]
fn test_hnsw_snapshot_without_vectors_file_keeps_vector_features() {
    use tempfile::tempdir;

    // Arrange: a snapshot written by the current format has NO
    // native_vectors.bin — the vectors live inside the graph dump (PERF1).
    let dir = tempdir().unwrap();
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.0, 1.0, 0.0]);
    index.save(dir.path()).unwrap();
    assert!(!dir.path().join("native_vectors.bin").exists());

    // Act: load the snapshot.
    let loaded_index = HnswIndex::load(dir.path(), 3, DistanceMetric::Cosine).unwrap();

    // Assert: full functionality — exact-distance features stay on (vectors are
    // in the graph) and vacuum works.
    assert_eq!(loaded_index.len(), 2);
    assert!(loaded_index.has_vector_storage());
    assert_eq!(loaded_index.vacuum(), Ok(2));
    assert_eq!(loaded_index.len(), 2);
}

#[test]
fn test_hnsw_fast_insert_flag_survives_save_load() {
    use tempfile::tempdir;

    let dir = tempdir().unwrap();
    let index = HnswIndex::new_fast_insert(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.0, 1.0, 0.0]);

    index.save(dir.path()).unwrap();
    assert!(!dir.path().join("native_vectors.bin").exists());

    let loaded = HnswIndex::load(dir.path(), 3, DistanceMetric::Cosine).unwrap();
    assert!(!loaded.has_vector_storage());
    assert_eq!(loaded.vacuum(), Err(VacuumError::VectorStorageDisabled));
}

#[test]
fn test_hnsw_save_removes_stale_legacy_vectors_file() {
    use tempfile::tempdir;

    let dir = tempdir().unwrap();

    // Write a first snapshot, then plant a legacy vectors file as an old
    // binary would have left it.
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.save(dir.path()).unwrap();
    std::fs::write(dir.path().join("native_vectors.bin"), [0_u8; 8]).unwrap();

    // Any save must remove the stale legacy duplicate (its generation could
    // never match the new snapshot's).
    index.insert(2, &[0.0, 1.0, 0.0]);
    index.save(dir.path()).unwrap();

    assert!(!dir.path().join("native_vectors.bin").exists());

    // And the snapshot reloads cleanly.
    let loaded = HnswIndex::load(dir.path(), 3, DistanceMetric::Cosine).unwrap();
    assert_eq!(loaded.len(), 2);
}

#[test]
fn test_hnsw_insert_batch_parallel() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    let vectors: Vec<(u64, Vec<f32>)> = vec![
        (1, vec![1.0, 0.0, 0.0]),
        (2, vec![0.0, 1.0, 0.0]),
        (3, vec![0.0, 0.0, 1.0]),
        (4, vec![0.5, 0.5, 0.0]),
        (5, vec![0.5, 0.0, 0.5]),
    ];

    // Act
    let inserted = index.insert_batch_parallel(vectors.iter().map(|(id, v)| (*id, v.as_slice())));
    index.set_searching_mode();

    // Assert
    assert_eq!(inserted, 5);
    assert_eq!(index.len(), 5);

    // Verify search works
    let results = index.search(&[1.0, 0.0, 0.0], 3);
    assert_eq!(results.len(), 3);
    // ID 1 should be in the top results (exact match)
    // Note: Due to parallel insertion, graph structure may vary
    let result_ids: Vec<u64> = results.iter().map(|r| r.id).collect();
    assert!(result_ids.contains(&1), "ID 1 should be in top 3 results");
}

#[test]
fn test_hnsw_insert_batch_parallel_upserts_duplicates() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();

    // Insert one vector first
    index.insert(1, &[1.0, 0.0, 0.0]);

    // Act - Insert batch with existing ID (upsert) + new ID
    let vectors: Vec<(u64, Vec<f32>)> = vec![
        (1, vec![0.0, 1.0, 0.0]), // Upsert: updates existing id=1
        (2, vec![0.0, 0.0, 1.0]), // New
    ];
    let inserted = index.insert_batch_parallel(vectors.iter().map(|(id, v)| (*id, v.as_slice())));
    index.set_searching_mode();

    // Assert - Both vectors processed (1 upserted + 1 new)
    assert_eq!(inserted, 2);
    assert_eq!(index.len(), 2);
}

#[test]
fn test_hnsw_insert_batch_parallel_empty() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    let vectors: Vec<(u64, &[f32])> = vec![];

    // Act
    let inserted = index.insert_batch_parallel(vectors);

    // Assert
    assert_eq!(inserted, 0);
    assert!(index.is_empty());
}

#[test]
fn test_hnsw_insert_batch_parallel_wrong_dimension_returns_zero() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    let vectors: Vec<(u64, &[f32])> = vec![(1, &[1.0, 0.0])]; // Wrong dim

    // Act - dimension mismatch returns 0 inserted
    let inserted = index.insert_batch_parallel(vectors);
    assert_eq!(inserted, 0, "Wrong-dimension batch should insert nothing");
    assert_eq!(index.len(), 0, "Index should remain empty");
}

// =========================================================================
// HnswIndex with Params Tests
// Note: HnswParams unit tests are in params.rs
// =========================================================================

#[test]
fn test_hnsw_with_params() {
    let params = HnswParams::custom(48, 600, 500_000);
    let index = HnswIndex::with_params(1536, DistanceMetric::Cosine, params).unwrap();

    assert_eq!(index.dimension(), 1536);
    assert!(index.is_empty());
}

// =========================================================================
// SIMD Re-ranking Tests (TDD - RED phase)
// =========================================================================

#[test]
fn test_search_with_rerank_returns_k_results() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.9, 0.1, 0.0]);
    index.insert(3, &[0.8, 0.2, 0.0]);
    index.insert(4, &[0.0, 1.0, 0.0]);
    index.insert(5, &[0.0, 0.0, 1.0]);

    // Act
    let results = index.search_with_rerank(&[1.0, 0.0, 0.0], 3, 5).unwrap();

    // Assert
    assert_eq!(results.len(), 3, "Should return exactly k results");
}

#[test]
#[allow(clippy::cast_precision_loss)]
fn test_search_with_rerank_improves_ranking() {
    // Arrange - vectors with subtle differences
    let index = HnswIndex::new(128, DistanceMetric::Cosine).unwrap();

    // Create vectors with known similarity ordering
    let base: Vec<f32> = (0..128).map(|i| (i as f32 * 0.01).sin()).collect();

    // Slightly modified versions
    let mut v1 = base.clone();
    v1[0] += 0.001; // Very similar

    let mut v2 = base.clone();
    v2[0] += 0.01; // Less similar

    let mut v3 = base.clone();
    v3[0] += 0.1; // Even less similar

    index.insert(1, &v1);
    index.insert(2, &v2);
    index.insert(3, &v3);

    // Act
    let results = index.search_with_rerank(&base, 3, 3).unwrap();

    // Assert - ID 1 should be closest (highest similarity)
    assert_eq!(results[0].id, 1, "Most similar vector should be first");
}

#[test]
fn test_search_with_rerank_handles_rerank_k_greater_than_index_size() {
    // Arrange - use more vectors to avoid HNSW flakiness
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.0, 1.0, 0.0]);
    index.insert(3, &[0.0, 0.0, 1.0]);
    index.insert(4, &[0.5, 0.5, 0.0]);
    index.insert(5, &[0.5, 0.0, 0.5]);

    // Act - rerank_k > index size
    let results = index.search_with_rerank(&[1.0, 0.0, 0.0], 3, 100).unwrap();

    // Assert - should return at least some results
    assert!(!results.is_empty(), "Should return results");
    assert!(results.len() <= 5, "Should not exceed index size");
}

#[test]
#[allow(clippy::cast_precision_loss, clippy::cast_sign_loss)]
fn test_search_with_rerank_uses_simd_distances() {
    // Arrange
    let index = HnswIndex::new(768, DistanceMetric::Cosine).unwrap();

    // Insert 100 vectors
    for i in 0..100_u64 {
        let v: Vec<f32> = (0..768)
            .map(|j| ((i + j as u64) as f32 * 0.01).sin())
            .collect();
        index.insert(i, &v);
    }

    let query: Vec<f32> = (0..768).map(|j| (j as f32 * 0.01).sin()).collect();

    // Act
    let results = index.search_with_rerank(&query, 10, 50).unwrap();

    // Assert - results should have valid distances (SIMD computed)
    // Note: HNSW may return fewer results if graph not fully connected
    assert!(!results.is_empty(), "Should return at least one result");
    for sr in &results {
        assert!(
            sr.score >= -1.0 && sr.score <= 1.0,
            "Cosine should be in [-1, 1]"
        );
    }

    // Results should be sorted by similarity (descending for cosine)
    for i in 1..results.len() {
        assert!(
            results[i - 1].score >= results[i].score,
            "Results should be sorted by similarity descending"
        );
    }
}

#[test]
fn test_search_with_rerank_euclidean_metric() {
    // Arrange
    let index = HnswIndex::new(3, DistanceMetric::Euclidean).unwrap();
    index.insert(1, &[0.0, 0.0, 0.0]);
    index.insert(2, &[1.0, 0.0, 0.0]);
    index.insert(3, &[2.0, 0.0, 0.0]);

    // Act
    let results = index.search_with_rerank(&[0.0, 0.0, 0.0], 3, 3).unwrap();

    // Assert - ID 1 should be closest (smallest distance)
    assert_eq!(results[0].id, 1, "Origin should be closest to itself");
    // For euclidean, smaller is better - results sorted ascending
    for i in 1..results.len() {
        assert!(
            results[i - 1].score <= results[i].score,
            "Euclidean results should be sorted ascending"
        );
    }
}

// =========================================================================
// Memory Leak Fix Tests
// Tests for multi-tenant scenarios and proper Drop behavior
// =========================================================================

#[test]
#[allow(
    clippy::cast_precision_loss,
    clippy::cast_sign_loss,
    clippy::uninlined_format_args
)]
fn test_hnsw_multi_tenant_load_unload() {
    // Arrange - Simulate multi-tenant scenario with multiple load/unload cycles
    // This test verifies that indices can be loaded and dropped without memory leak
    use tempfile::tempdir;

    let dir = tempdir().expect("Failed to create temp dir");

    // Create and save an index
    {
        let index = HnswIndex::new(128, DistanceMetric::Cosine).unwrap();
        for i in 0..100_u64 {
            let v: Vec<f32> = (0..128)
                .map(|j| ((i + j as u64) as f32 * 0.01).sin())
                .collect();
            index.insert(i, &v);
        }
        index.save(dir.path()).expect("Failed to save index");
    }

    // Act - Load and drop multiple times (simulates multi-tenant load/unload)
    for iteration in 0..5 {
        let loaded =
            HnswIndex::load(dir.path(), 128, DistanceMetric::Cosine).expect("Failed to load index");

        // Verify index works correctly
        assert_eq!(
            loaded.len(),
            100,
            "Iteration {}: Should have 100 vectors",
            iteration
        );

        let query: Vec<f32> = (0..128).map(|j| (j as f32 * 0.01).sin()).collect();
        let results = loaded.search(&query, 5);
        // HNSW may return fewer than k results depending on graph connectivity
        assert!(
            !results.is_empty() && results.len() <= 5,
            "Iteration {}: Should return 1-5 results, got {}",
            iteration,
            results.len()
        );

        // Index is dropped here, io_holder should be freed
    }

    // If we get here without crash/hang, memory is being managed correctly
}

#[test]
fn test_hnsw_drop_cleans_up_properly() {
    // Arrange - Create index, verify it can be dropped without issues
    use tempfile::tempdir;

    let dir = tempdir().expect("Failed to create temp dir");

    // Create, save, load, and drop
    {
        let index = HnswIndex::new(64, DistanceMetric::Euclidean).unwrap();
        index.insert(1, &vec![0.5; 64]);
        index.insert(2, &vec![0.3; 64]);
        index.save(dir.path()).expect("Failed to save");
    }

    // Load and immediately drop
    {
        let _loaded =
            HnswIndex::load(dir.path(), 64, DistanceMetric::Euclidean).expect("Failed to load");
        // Dropped here
    }

    // Load again to verify files are still valid after previous drop
    {
        let loaded = HnswIndex::load(dir.path(), 64, DistanceMetric::Euclidean)
            .expect("Failed to load after previous drop");
        assert_eq!(loaded.len(), 2);
    }
}

#[test]
#[allow(clippy::cast_precision_loss, clippy::uninlined_format_args)]
fn test_hnsw_save_load_preserves_all_metrics() {
    use tempfile::tempdir;

    // Test Cosine and Euclidean metrics
    // Note: DotProduct has numerical precision issues in hnsw_rs with certain vectors
    for metric in [DistanceMetric::Cosine, DistanceMetric::Euclidean] {
        let dir = tempdir().expect("Failed to create temp dir");
        let dim = 32;

        // Create varied vectors (not constant) to avoid numerical issues
        let v1: Vec<f32> = (0..dim).map(|i| (i as f32 * 0.1).sin()).collect();
        let v2: Vec<f32> = (0..dim).map(|i| (i as f32 * 0.2).cos()).collect();
        let query: Vec<f32> = (0..dim).map(|i| (i as f32 * 0.15).sin()).collect();

        // Create and save
        {
            let index = HnswIndex::new(dim, metric).unwrap();
            index.insert(1, &v1);
            index.insert(2, &v2);
            index.save(dir.path()).expect("Failed to save");
        }

        // Load and verify
        {
            let loaded = HnswIndex::load(dir.path(), dim, metric).expect("Failed to load");
            assert_eq!(
                loaded.len(),
                2,
                "Metric {:?}: Should have 2 vectors",
                metric
            );
            assert_eq!(loaded.metric(), metric, "Metric should be preserved");
            assert_eq!(loaded.dimension(), dim, "Dimension should be preserved");

            // Verify search works
            let results = loaded.search(&query, 2);
            assert!(
                !results.is_empty(),
                "Metric {:?}: Should return results",
                metric
            );
        }
    }
}

// =========================================================================
// SearchQuality Tests
// =========================================================================

#[test]
fn test_search_quality_fast() {
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    // Insert more vectors for stable HNSW graph (small graphs are non-deterministic)
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.9, 0.1, 0.0]);
    index.insert(3, &[0.8, 0.2, 0.0]);
    index.insert(4, &[0.7, 0.3, 0.0]);
    index.insert(5, &[0.0, 1.0, 0.0]);

    let results = index
        .search_with_quality(&[1.0, 0.0, 0.0], 2, SearchQuality::Fast)
        .unwrap();
    // Fast mode may return fewer results with very small ef_search
    assert!(!results.is_empty(), "Should return at least one result");
    assert!(results.len() <= 2, "Should not exceed requested k");
}

#[test]
fn test_search_quality_balanced() {
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.9, 0.1, 0.0]);

    // Test Balanced quality mode
    let results = index
        .search_with_quality(&[1.0, 0.0, 0.0], 2, SearchQuality::Balanced)
        .unwrap();
    // HNSW may return fewer results for very small indices
    assert!(!results.is_empty(), "Should return at least one result");
    assert_eq!(
        results[0].id, 1,
        "Balanced search should find exact match first"
    );
}

#[test]
fn test_search_quality_custom_ef() {
    // Use more vectors to make HNSW more stable
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.9, 0.1, 0.0]);
    index.insert(3, &[0.8, 0.2, 0.0]);
    index.insert(4, &[0.0, 1.0, 0.0]);
    index.insert(5, &[0.0, 0.0, 1.0]);

    let results = index
        .search_with_quality(&[1.0, 0.0, 0.0], 3, SearchQuality::Custom(512))
        .unwrap();
    assert_eq!(results.len(), 3);
}

// Note: SearchQuality::ef_search unit tests are in params.rs

// =========================================================================
// Edge Cases and Error Handling
// =========================================================================

#[test]
fn test_hnsw_load_nonexistent_path() {
    let result = HnswIndex::load("nonexistent_path_12345", 128, DistanceMetric::Cosine);
    assert!(result.is_err(), "Loading from nonexistent path should fail");
}

#[test]
fn test_hnsw_search_with_rerank_empty_index() {
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    let results = index.search_with_rerank(&[1.0, 0.0, 0.0], 10, 50).unwrap();
    assert!(
        results.is_empty(),
        "Empty index should return empty results"
    );
}

#[test]
fn test_hnsw_search_with_rerank_dot_product() {
    let index = HnswIndex::new(3, DistanceMetric::DotProduct).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.5, 0.5, 0.0]);
    index.insert(3, &[0.0, 1.0, 0.0]);

    let results = index.search_with_rerank(&[1.0, 0.0, 0.0], 3, 3).unwrap();

    // HNSW may return fewer results for very small indices
    assert!(!results.is_empty(), "Should return at least one result");
    // For dot product, ID 1 should have highest score
    assert_eq!(results[0].id, 1, "Highest dot product should be first");
}

#[test]
fn test_hnsw_io_holder_is_none_for_new_index() {
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    assert!(
        index.io_holder.is_none(),
        "new native index must have io_holder == None"
    );
    index.insert(1, &[1.0, 0.0, 0.0]);
    assert!(
        index.io_holder.is_none(),
        "io_holder must stay None after insert (native backend never borrows from disk)"
    );
}

#[test]
#[allow(clippy::cast_precision_loss, clippy::cast_sign_loss)]
fn test_hnsw_large_batch_parallel_insert() {
    let index = HnswIndex::new(128, DistanceMetric::Cosine).unwrap();

    // Create 200 vectors (reduced from 1000 for faster test execution)
    let vectors: Vec<(u64, Vec<f32>)> = (0..200)
        .map(|i| {
            let v: Vec<f32> = (0..128).map(|j| ((i + j) as f32 * 0.001).sin()).collect();
            (i as u64, v)
        })
        .collect();

    let inserted = index.insert_batch_parallel(vectors.iter().map(|(id, v)| (*id, v.as_slice())));
    index.set_searching_mode();

    assert_eq!(inserted, 200, "Should insert 200 vectors");
    assert_eq!(index.len(), 200);

    // Verify search works
    let query: Vec<f32> = (0..128).map(|j| (j as f32 * 0.001).sin()).collect();
    let results = index.search(&query, 10);
    assert_eq!(results.len(), 10);
}

// =========================================================================
// TS-CORE-001: Adaptive Prefetch Tests
// =========================================================================

#[test]
#[allow(clippy::cast_precision_loss, clippy::cast_sign_loss)]
fn test_search_with_rerank_768d_prefetch() {
    // Test adaptive prefetch for 768D vectors (3KB each)
    // prefetch_distance should be 768*4/64 = 48, clamped to 16
    let index = HnswIndex::new(768, DistanceMetric::Cosine).unwrap();

    // Insert 100 vectors
    for i in 0u64..100 {
        let v: Vec<f32> = (0..768)
            .map(|j| ((i + j as u64) as f32 * 0.001).sin())
            .collect();
        index.insert(i, &v);
    }

    let query: Vec<f32> = (0..768).map(|j| (j as f32 * 0.001).sin()).collect();
    let results = index.search_with_rerank(&query, 10, 50).unwrap();

    assert!(!results.is_empty(), "Should return results");
    assert!(results.len() <= 10, "Should not exceed k");
    assert_eq!(
        results[0].id, 0,
        "Self-match: query equals vector 0, should rank top-1"
    );
    for sr in &results {
        assert!(sr.score >= -1.0 && sr.score <= 1.0, "Cosine in [-1,1]");
    }
    for i in 1..results.len() {
        assert!(
            results[i - 1].score >= results[i].score,
            "Sorted descending"
        );
    }
}

#[test]
#[allow(clippy::cast_precision_loss, clippy::cast_sign_loss)]
fn test_search_with_rerank_small_dim_prefetch() {
    // Test adaptive prefetch for small vectors (32D = 128 bytes)
    // prefetch_distance should be 128/64 = 2, clamped to 4 (minimum)
    let index = HnswIndex::new(32, DistanceMetric::Cosine).unwrap();

    for i in 0u64..50 {
        let v: Vec<f32> = (0..32)
            .map(|j| ((i + j as u64) as f32 * 0.01).sin())
            .collect();
        index.insert(i, &v);
    }

    let query: Vec<f32> = (0..32).map(|j| (j as f32 * 0.01).sin()).collect();
    let results = index.search_with_rerank(&query, 5, 20).unwrap();

    assert!(!results.is_empty(), "Should return results");
    assert!(results.len() <= 5, "Should not exceed k");
    // Query equals inserted vector 0 exactly -> id 0 must be the top (exact-match) result.
    assert_eq!(
        results[0].id, 0,
        "Exact-match vector should rank first after rerank"
    );
}

// =========================================================================
// TS-CORE-002: Batch Search Optimization Tests
// =========================================================================

#[test]
#[allow(clippy::cast_precision_loss, clippy::cast_sign_loss)]
fn test_search_batch_parallel_consistency() {
    let index = HnswIndex::new(64, DistanceMetric::Cosine).unwrap();

    // Insert 100 vectors (reduced from 200 for faster test execution)
    for i in 0u64..100 {
        let v: Vec<f32> = (0..64)
            .map(|j| ((i + j as u64) as f32 * 0.01).sin())
            .collect();
        index.insert(i, &v);
    }

    // Create batch queries
    let queries: Vec<Vec<f32>> = (0..10)
        .map(|i| {
            (0..64)
                .map(|j| ((200 + i + j) as f32 * 0.01).sin())
                .collect()
        })
        .collect();
    let query_refs: Vec<&[f32]> = queries.iter().map(Vec::as_slice).collect();

    // Batch search
    let batch_results = index
        .search_batch_parallel(&query_refs, 5, SearchQuality::Balanced)
        .unwrap();

    // Individual searches for comparison
    let individual_results: Vec<Vec<crate::scored_result::ScoredResult>> = queries
        .iter()
        .map(|q| {
            index
                .search_with_quality(q, 5, SearchQuality::Balanced)
                .unwrap()
        })
        .collect();

    // Results should match (same IDs, though order might vary slightly)
    assert_eq!(batch_results.len(), individual_results.len());
    for (batch, individual) in batch_results.iter().zip(&individual_results) {
        assert_eq!(batch.len(), individual.len(), "Result counts should match");
        let batch_ids: std::collections::HashSet<u64> = batch.iter().map(|r| r.id).collect();
        let individual_ids: std::collections::HashSet<u64> =
            individual.iter().map(|r| r.id).collect();
        assert_eq!(
            batch_ids, individual_ids,
            "Batch and individual results must return the same IDs"
        );
    }
}

#[test]
fn test_search_batch_parallel_empty_queries() {
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);

    let queries: Vec<&[f32]> = vec![];
    let results = index
        .search_batch_parallel(&queries, 5, SearchQuality::Fast)
        .unwrap();

    assert!(
        results.is_empty(),
        "Empty queries should return empty results"
    );
}

#[test]
#[allow(clippy::cast_precision_loss, clippy::cast_sign_loss)]
fn test_search_batch_parallel_large_batch() {
    let index = HnswIndex::new(128, DistanceMetric::Cosine).unwrap();

    // Insert 150 vectors (reduced from 500 for faster test execution)
    for i in 0u64..150 {
        let v: Vec<f32> = (0..128)
            .map(|j| ((i + j as u64) as f32 * 0.001).sin())
            .collect();
        index.insert(i, &v);
    }
    index.set_searching_mode();

    // 20 queries batch (reduced from 100 for faster test execution)
    let queries: Vec<Vec<f32>> = (0..20)
        .map(|i| {
            (0..128)
                .map(|j| ((150 + i + j) as f32 * 0.001).sin())
                .collect()
        })
        .collect();
    let query_refs: Vec<&[f32]> = queries.iter().map(Vec::as_slice).collect();

    // Use Balanced for faster test execution
    let results = index
        .search_batch_parallel(&query_refs, 10, SearchQuality::Balanced)
        .unwrap();

    assert_eq!(results.len(), 20, "Should return 20 result sets");
    for result in &results {
        assert_eq!(result.len(), 10, "Each result should have 10 neighbors");
    }
}

// =========================================================================
// Recall Quality Regression Tests
// =========================================================================

#[test]
#[allow(clippy::cast_precision_loss)]
fn test_recall_quality_minimum_threshold() {
    // Ensure recall@10 >= 90% for Accurate quality on small dataset
    let dim = 64;
    let n = 500;
    let k = 10;

    let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();

    // Generate deterministic dataset
    let dataset: Vec<Vec<f32>> = (0..n)
        .map(|i| {
            (0..dim)
                .map(|j| ((i * dim + j) as f32 * 0.001).sin())
                .collect()
        })
        .collect();

    for (idx, vec) in dataset.iter().enumerate() {
        #[allow(clippy::cast_possible_truncation)]
        index.insert(idx as u64, vec);
    }

    // Generate query
    let query: Vec<f32> = (0..dim).map(|j| (j as f32 * 0.001).sin()).collect();

    // Compute ground truth with brute force
    let mut distances: Vec<crate::scored_result::ScoredResult> = dataset
        .iter()
        .enumerate()
        .map(|(idx, vec)| {
            let sim = crate::simd_native::cosine_similarity_native(&query, vec);
            #[allow(clippy::cast_possible_truncation)]
            crate::scored_result::ScoredResult::new(idx as u64, sim)
        })
        .collect();
    distances.sort_by(|a, b| {
        b.score
            .partial_cmp(&a.score)
            .unwrap_or(std::cmp::Ordering::Equal)
    });
    let ground_truth: Vec<u64> = distances.iter().take(k).map(|sr| sr.id).collect();

    // HNSW search
    let results = index
        .search_with_quality(&query, k, SearchQuality::Accurate)
        .unwrap();
    let result_ids: std::collections::HashSet<u64> = results.iter().map(|sr| sr.id).collect();
    let gt_set: std::collections::HashSet<u64> = ground_truth.iter().copied().collect();

    let recall = result_ids.intersection(&gt_set).count() as f64 / k as f64;

    assert!(
        recall >= 0.8,
        "Recall@{k} should be >= 80% for Accurate, got {:.1}%",
        recall * 100.0
    );
}

#[test]
fn test_rerank_latency_target_configuration_roundtrip() {
    let index = HnswIndex::new(32, DistanceMetric::Cosine).unwrap();
    assert_eq!(index.rerank_latency_target_us(), 0);

    index.set_rerank_latency_target_us(250);
    assert_eq!(index.rerank_latency_target_us(), 250);
}

#[test]
fn test_rerank_latency_ema_updates_after_two_stage_search() {
    let index = HnswIndex::new(64, DistanceMetric::Cosine).unwrap();
    index.set_rerank_latency_target_us(1);

    for i in 0u64..1500 {
        let v: Vec<f32> = (0..64)
            .map(|j| ((i * 5 + j as u64) as f32 * 0.0013).sin())
            .collect();
        index.insert(i, &v);
    }

    let query: Vec<f32> = (0..64).map(|j| (j as f32 * 0.009).cos()).collect();
    let _ = index
        .search_with_quality(&query, 20, SearchQuality::Accurate)
        .unwrap();

    assert!(index.rerank_latency_ema_us() > 0);
}

#[test]
fn test_update_rerank_latency_ema_large_current_does_not_overflow() {
    let index = HnswIndex::new(64, DistanceMetric::Cosine).unwrap();

    for i in 0u64..400 {
        let v: Vec<f32> = (0..64)
            .map(|j| ((i * 3 + j as u64) as f32 * 0.0021).sin())
            .collect();
        index.insert(i, &v);
    }

    index
        .rerank_latency_ema_us
        .store(u64::MAX, std::sync::atomic::Ordering::Relaxed);

    let query: Vec<f32> = (0..64).map(|j| (j as f32 * 0.011).cos()).collect();
    let results = index
        .search_with_quality(&query, 20, SearchQuality::Accurate)
        .unwrap();

    assert!(!results.is_empty());

    let ema = index.rerank_latency_ema_us();
    let expected_min = (u64::MAX / 10) * 7;
    assert!(ema >= expected_min, "EMA underflow/wrap detected: {ema}");
}

#[test]
fn test_search_with_quality_custom_ef_uses_high_recall_path_without_regression() {
    let index = HnswIndex::new(64, DistanceMetric::Cosine).unwrap();

    for i in 0u64..2000 {
        let v: Vec<f32> = (0..64)
            .map(|j| ((i + j as u64) as f32 * 0.001).sin())
            .collect();
        index.insert(i, &v);
    }

    let query: Vec<f32> = (0..64).map(|j| (j as f32 * 0.013).cos()).collect();
    let results = index
        .search_with_quality(&query, 20, SearchQuality::Custom(512))
        .unwrap();

    assert!(!results.is_empty());
    assert!(results.len() <= 20);
}

#[test]
fn test_search_with_quality_accurate_stays_stable_on_medium_dataset() {
    let index = HnswIndex::new(128, DistanceMetric::Cosine).unwrap();

    for i in 0u64..5000 {
        let v: Vec<f32> = (0..128)
            .map(|j| ((i * 3 + j as u64) as f32 * 0.0007).sin())
            .collect();
        index.insert(i, &v);
    }

    let query: Vec<f32> = (0..128).map(|j| (j as f32 * 0.007).sin()).collect();
    let results = index
        .search_with_quality(&query, 15, SearchQuality::Accurate)
        .unwrap();

    assert!(!results.is_empty());
    assert!(results.len() <= 15);
}

// =========================================================================
// RF-3: Tests for search_brute_force_buffered (buffer reuse optimization)
// =========================================================================

#[test]
fn test_brute_force_buffered_same_results_as_original() {
    let index = HnswIndex::new(32, DistanceMetric::Cosine).unwrap();

    // Insert vectors
    for i in 0u64..50 {
        let v: Vec<f32> = (0..32)
            .map(|j| ((i + j as u64) as f32 * 0.01).sin())
            .collect();
        index.insert(i, &v);
    }

    let query: Vec<f32> = (0..32).map(|j| (j as f32 * 0.02).cos()).collect();

    // Compare results
    let original = index.search_brute_force(&query, 10).unwrap();
    let buffered = index.search_brute_force_buffered(&query, 10).unwrap();

    assert_eq!(original.len(), buffered.len());
    for (orig, buf) in original.iter().zip(buffered.iter()) {
        assert_eq!(orig.id, buf.id, "IDs should match");
        assert!(
            (orig.score - buf.score).abs() < 1e-6,
            "Distances should match"
        );
    }
}

#[test]
fn test_brute_force_buffered_empty_index() {
    let index = HnswIndex::new(16, DistanceMetric::Euclidean).unwrap();
    let query: Vec<f32> = vec![0.0; 16];

    let results = index.search_brute_force_buffered(&query, 5).unwrap();
    assert!(results.is_empty());
}

#[test]
fn test_brute_force_buffered_all_metrics() {
    for metric in [
        DistanceMetric::Cosine,
        DistanceMetric::Euclidean,
        DistanceMetric::DotProduct,
        DistanceMetric::Hamming,
        DistanceMetric::Jaccard,
    ] {
        let index = HnswIndex::new(8, metric).unwrap();
        index.insert(1, &[1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]);
        index.insert(2, &[0.5, 0.5, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]);
        index.insert(3, &[0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]);

        let results = index
            .search_brute_force_buffered(&[1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], 3)
            .unwrap();
        assert_eq!(results.len(), 3, "Should return 3 results for {metric:?}");
    }
}

#[test]
fn test_brute_force_buffered_repeated_calls_stable() {
    let index = HnswIndex::new(16, DistanceMetric::Cosine).unwrap();

    for i in 0u64..20 {
        let v: Vec<f32> = (0..16)
            .map(|j| ((i + j as u64) as f32 * 0.1).sin())
            .collect();
        index.insert(i, &v);
    }

    let query: Vec<f32> = vec![0.5; 16];

    // Multiple calls should return identical results
    let r1 = index.search_brute_force_buffered(&query, 5).unwrap();
    let r2 = index.search_brute_force_buffered(&query, 5).unwrap();
    let r3 = index.search_brute_force_buffered(&query, 5).unwrap();

    assert_eq!(r1, r2);
    assert_eq!(r2, r3);
}

// =========================================================================
// Regression (#2100): Jaccard is declared a similarity by higher_is_better,
// so transform_score must return `1 - engine_distance` (the similarity),
// exactly like Cosine. Pre-fix it passed the raw `1 - jaccard` distance
// through, so every descending sort/top-k over HNSW scores ranked the
// WORST candidates first and user-visible scores were inverted.
// =========================================================================

/// Dim-4 Jaccard fixture with hand-computed similarities against
/// q=[1,1,0,0]: id1=1.0 (identical), id2=0.5 (Σmin/Σmax=1/2), id3=0.0.
fn jaccard_fixture() -> (HnswIndex, Vec<f32>) {
    let index = HnswIndex::new(4, DistanceMetric::Jaccard).unwrap();
    index.insert(1, &[1.0, 1.0, 0.0, 0.0]);
    index.insert(2, &[1.0, 0.0, 0.0, 0.0]);
    index.insert(3, &[0.0, 0.0, 1.0, 1.0]);
    (index, vec![1.0, 1.0, 0.0, 0.0])
}

#[test]
fn test_jaccard_hnsw_scores_are_similarities() {
    let (index, query) = jaccard_fixture();
    let results = index.search(&query, 3);

    assert_eq!(results.len(), 3);
    assert_eq!(
        results[0].id, 1,
        "exact match must rank first, got {results:?}"
    );
    // Scores must be the Jaccard SIMILARITY (1.0, 0.5, 0.0), not the
    // internal 1-sim traversal distance (0.0, 0.5, 1.0).
    for (result, want) in results.iter().zip([1.0f32, 0.5, 0.0]) {
        assert!(
            (result.score - want).abs() < 1e-6,
            "id {} score {} != similarity {want}",
            result.id,
            result.score
        );
    }
}

#[test]
fn test_jaccard_bitmap_filtered_search_keeps_best_not_worst() {
    let (index, query) = jaccard_fixture();
    let allowed: roaring::RoaringBitmap = [1u32, 2, 3].into_iter().collect();

    let results = index
        .search_with_quality_and_bitmap(&query, 1, SearchQuality::Balanced, &allowed)
        .unwrap();

    // Pre-fix, the descending top-k over raw distances kept id3 (the
    // completely dissimilar vector) and dropped the exact match.
    assert_eq!(results.len(), 1);
    assert_eq!(results[0].id, 1, "top-1 must be the exact match");
    assert!(
        (results[0].score - 1.0).abs() < 1e-6,
        "exact match must score similarity 1.0, got {}",
        results[0].score
    );
}

// =========================================================================
// Stress Tests
// =========================================================================

#[test]
#[allow(clippy::cast_precision_loss, clippy::cast_sign_loss)]
fn test_concurrent_search_stress() {
    use std::sync::Arc;
    use std::thread;

    let index = Arc::new(HnswIndex::new(64, DistanceMetric::Cosine).unwrap());

    // Insert vectors
    for i in 0u64..100 {
        let v: Vec<f32> = (0..64)
            .map(|j| ((i + j as u64) as f32 * 0.01).sin())
            .collect();
        index.insert(i, &v);
    }

    // Spawn multiple search threads
    let handles: Vec<_> = (0..4)
        .map(|t| {
            let idx = Arc::clone(&index);
            thread::spawn(move || {
                for i in 0..50 {
                    let query: Vec<f32> = (0..64)
                        .map(|j| ((t * 100 + i + j) as f32 * 0.01).sin())
                        .collect();
                    let results = idx.search(&query, 5);
                    assert!(!results.is_empty());
                }
            })
        })
        .collect();

    for handle in handles {
        handle.join().expect("Thread panicked");
    }
}

#[test]
fn test_all_distance_metrics_search_with_rerank() {
    for metric in [
        DistanceMetric::Cosine,
        DistanceMetric::Euclidean,
        DistanceMetric::DotProduct,
        DistanceMetric::Hamming,
        DistanceMetric::Jaccard,
    ] {
        let index = HnswIndex::new(8, metric).unwrap();
        index.insert(1, &[1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]);
        index.insert(2, &[0.5, 0.5, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]);
        index.insert(3, &[0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]);

        let results = index
            .search_with_rerank(&[1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], 3, 3)
            .unwrap();

        assert!(
            !results.is_empty(),
            "search_with_rerank should work for {metric:?}"
        );
    }
}

// =========================================================================
// SAFETY: Drop Order Tests for io_holder unsafe invariant
// =========================================================================
//
// These tests verify that the unsafe lifetime extension in HnswIndex::load()
// doesn't cause use-after-free when the index is dropped.
//
// CRITICAL INVARIANT: `inner` (which borrows from io_holder) MUST be dropped
// BEFORE `io_holder`. Our Drop impl ensures this via ManuallyDrop.

#[test]
fn test_drop_safety_loaded_index_no_segfault() {
    // This test verifies that dropping a loaded HnswIndex doesn't segfault.
    // If the Drop order is wrong, this will cause use-after-free.
    use tempfile::tempdir;

    let dir = tempdir().expect("Failed to create temp dir");

    // 1. Create and save an index
    {
        let index = HnswIndex::new(4, DistanceMetric::Cosine).unwrap();
        index.insert(1, &[1.0, 0.0, 0.0, 0.0]);
        index.insert(2, &[0.0, 1.0, 0.0, 0.0]);
        index.insert(3, &[0.0, 0.0, 1.0, 0.0]);
        index.save(dir.path()).expect("Failed to save");
    }

    // 2. Load and drop multiple times to stress test Drop safety
    for _ in 0..5 {
        let loaded =
            HnswIndex::load(dir.path(), 4, DistanceMetric::Cosine).expect("Failed to load");

        // Perform operations that touch the borrowed data
        let results = loaded.search(&[1.0, 0.0, 0.0, 0.0], 2);
        assert!(!results.is_empty(), "Search should return results");

        // Index is dropped here - if Drop order is wrong, this segfaults
    }
}

#[test]
fn test_drop_safety_loaded_index_concurrent_drop() {
    // Stress test: multiple threads loading and dropping indices
    use std::sync::Arc;
    use std::thread;
    use tempfile::tempdir;

    let dir = tempdir().expect("Failed to create temp dir");

    // Create and save an index
    {
        let index = HnswIndex::new(4, DistanceMetric::Cosine).unwrap();
        for i in 0u64..10 {
            let v = vec![i as f32, 0.0, 0.0, 0.0];
            index.insert(i, &v);
        }
        index.save(dir.path()).expect("Failed to save");
    }

    let path = Arc::new(dir.path().to_path_buf());

    // Spawn threads that load, search, and drop
    let handles: Vec<_> = (0..4)
        .map(|_| {
            let p = Arc::clone(&path);
            thread::spawn(move || {
                for _ in 0..3 {
                    let loaded =
                        HnswIndex::load(&*p, 4, DistanceMetric::Cosine).expect("Failed to load");
                    let results = loaded.search(&[1.0, 0.0, 0.0, 0.0], 3);
                    assert!(!results.is_empty());
                    // Drop happens here
                }
            })
        })
        .collect();

    for h in handles {
        h.join().expect("Thread should not panic from Drop");
    }
}

#[test]
fn test_drop_safety_search_after_partial_operations() {
    // Test that search works correctly even with complex operation sequences
    // before drop, ensuring borrowed data is valid until Drop.
    use tempfile::tempdir;

    let dir = tempdir().expect("Failed to create temp dir");

    // Create index with various operations
    {
        let index = HnswIndex::new(8, DistanceMetric::Euclidean).unwrap();
        for i in 0u64..20 {
            let v: Vec<f32> = (0..8).map(|j| (i + j) as f32 * 0.1).collect();
            index.insert(i, &v);
        }
        index.save(dir.path()).expect("Failed to save");
    }

    // Load and perform many operations before drop
    let loaded = HnswIndex::load(dir.path(), 8, DistanceMetric::Euclidean).expect("Failed to load");

    // Multiple searches touching the mmap'd data
    for i in 0..10 {
        let query: Vec<f32> = (0..8).map(|j| (i + j) as f32 * 0.1).collect();
        let results = loaded.search(&query, 5);
        assert!(results.len() <= 5);
    }

    // Batch search
    let queries: Vec<Vec<f32>> = (0..5)
        .map(|i| (0..8).map(|j| (i + j) as f32 * 0.1).collect())
        .collect();
    let query_refs: Vec<&[f32]> = queries.iter().map(|v| v.as_slice()).collect();
    let batch_results = loaded
        .search_batch_parallel(&query_refs, 3, SearchQuality::Balanced)
        .unwrap();
    assert_eq!(batch_results.len(), 5);

    // Drop happens here - all borrowed data must still be valid
    drop(loaded);
}

// =========================================================================
// SEC-1: Stress Test - Drop under heavy concurrent load
// Validates ManuallyDrop + RwLock safety under extreme conditions
// =========================================================================

#[test]
fn test_drop_stress_concurrent_create_destroy_loop() {
    // Stress test: rapidly create/destroy indices while performing operations
    // This tests the ManuallyDrop pattern under pressure
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;

    let success_count = Arc::new(AtomicUsize::new(0));
    let iterations = 50;

    for _ in 0..iterations {
        let success = Arc::clone(&success_count);

        // Create index, perform operations, drop
        let index = Arc::new(HnswIndex::new(16, DistanceMetric::Cosine).unwrap());

        // Spawn readers that will race with drop
        let handles: Vec<_> = (0..4)
            .map(|t| {
                let idx = Arc::clone(&index);
                std::thread::spawn(move || {
                    // Insert some vectors
                    for i in 0..10 {
                        let id = (t * 100 + i) as u64;
                        let v: Vec<f32> = (0..16).map(|j| (id + j) as f32 * 0.01).collect();
                        idx.insert(id, &v);
                    }
                    // Search
                    let q: Vec<f32> = (0..16).map(|i| i as f32 * 0.01).collect();
                    let _ = idx.search(&q, 5);
                })
            })
            .collect();

        // Wait for all threads
        for h in handles {
            h.join().expect("Thread panicked during stress test");
        }

        // Force drop while ensuring all operations completed
        drop(index);
        success.fetch_add(1, Ordering::SeqCst);
    }

    assert_eq!(
        success_count.load(Ordering::SeqCst),
        iterations,
        "All iterations should complete without panic"
    );
}

#[test]
fn test_drop_stress_load_search_destroy_cycle() {
    // Stress test: load from disk, search heavily, destroy - repeated
    use tempfile::tempdir;

    let dir = tempdir().expect("Failed to create temp dir");

    // Create and save initial index
    {
        let index = HnswIndex::new(32, DistanceMetric::Euclidean).unwrap();
        for i in 0u64..100 {
            let v: Vec<f32> = (0..32).map(|j| ((i + j) as f32).sin()).collect();
            index.insert(i, &v);
        }
        index.save(dir.path()).expect("Failed to save");
    }

    // Repeated load/search/destroy cycles
    for cycle in 0..20 {
        let loaded = HnswIndex::load(dir.path(), 32, DistanceMetric::Euclidean)
            .unwrap_or_else(|e| panic!("Cycle {cycle}: Failed to load: {e}"));

        // Heavy search load
        for i in 0..50 {
            let q: Vec<f32> = (0..32).map(|j| ((i + j) as f32).cos()).collect();
            let results = loaded.search(&q, 10);
            assert!(
                results.len() <= 10,
                "Cycle {cycle}: Search returned too many results"
            );
        }

        // Explicit drop to test ManuallyDrop pattern
        drop(loaded);
    }
}

#[test]
fn test_drop_stress_parallel_insert_then_drop() {
    // Stress test: parallel batch insert immediately followed by drop
    // Use Euclidean to avoid cosine normalization requirements
    // Reduced iterations and batch size for faster test execution
    for _ in 0..5 {
        let index = HnswIndex::new(64, DistanceMetric::Euclidean).unwrap();

        // Generate batch data with reasonable magnitude (reduced from 500)
        let batch: Vec<(u64, Vec<f32>)> = (0..100)
            .map(|i| {
                let v: Vec<f32> = (0..64).map(|j| (i + j) as f32 * 0.01).collect();
                (i as u64, v)
            })
            .collect();

        // Parallel insert
        let inserted = index.insert_batch_parallel(batch.iter().map(|(id, v)| (*id, v.as_slice())));
        assert!(inserted > 0, "Should insert at least some vectors");

        // Immediate drop without set_searching_mode
        // This tests that Drop handles partially-initialized state
        drop(index);
    }
}

// =========================================================================
// P1-GPU-1: GPU Batch Search Tests (TDD - Written BEFORE implementation)
// =========================================================================

#[test]
#[cfg(feature = "gpu")]
fn test_search_brute_force_gpu_returns_same_results_as_cpu() {
    // TDD: GPU brute force must return identical results to CPU
    let index = HnswIndex::new(128, DistanceMetric::Cosine).unwrap();

    // Insert test vectors
    for i in 0u64..100 {
        let v: Vec<f32> = (0..128)
            .map(|j| ((i + j as u64) as f32 * 0.01).sin())
            .collect();
        index.insert(i, &v);
    }

    let query: Vec<f32> = (0..128).map(|j| (j as f32 * 0.02).cos()).collect();

    // CPU brute force
    let cpu_results = index.search_brute_force(&query, 10).unwrap();

    // GPU brute force (if available)
    if let Some(gpu_results) = index.search_brute_force_gpu(&query, 10).unwrap() {
        assert_eq!(
            cpu_results.len(),
            gpu_results.len(),
            "Result count mismatch"
        );

        // Verify same IDs returned (order may differ slightly due to floating point)
        let cpu_ids: std::collections::HashSet<u64> = cpu_results.iter().map(|sr| sr.id).collect();
        let gpu_ids: std::collections::HashSet<u64> = gpu_results.iter().map(|sr| sr.id).collect();

        let overlap = cpu_ids.intersection(&gpu_ids).count();
        assert!(
            overlap >= 8,
            "GPU and CPU should return mostly same IDs (got {overlap}/10 overlap)"
        );
    }
}

#[test]
fn test_search_brute_force_gpu_fallback_to_none_without_gpu() {
    // TDD: Without GPU, should return None gracefully
    let index = HnswIndex::new(64, DistanceMetric::Cosine).unwrap();
    index.insert(1, &vec![0.5; 64]);

    let query = vec![0.5; 64];

    // Should not panic, returns Ok(None) if GPU unavailable
    let result = index.search_brute_force_gpu(&query, 5).unwrap();

    #[cfg(not(feature = "gpu"))]
    assert!(result.is_none(), "Should return None without GPU feature");
    #[cfg(feature = "gpu")]
    let _ = result;
}

#[test]
#[cfg(feature = "gpu")]
fn test_brute_force_gpu_euclidean() {
    // RED: GPU brute-force must dispatch Euclidean, not hardcode Cosine.
    let index = HnswIndex::new(4, DistanceMetric::Euclidean).unwrap();

    // Insert vectors at known positions so distances are deterministic.
    // vec0 = origin, vec1 = unit along x, vec2 = 2 units along x.
    index.insert(10, &[0.0, 0.0, 0.0, 0.0]);
    index.insert(20, &[1.0, 0.0, 0.0, 0.0]);
    index.insert(30, &[2.0, 0.0, 0.0, 0.0]);

    let query = [0.0, 0.0, 0.0, 0.0]; // closest to vec0

    if let Some(results) = index.search_brute_force_gpu(&query, 3).unwrap() {
        assert_eq!(results.len(), 3, "Should return 3 results");

        // Euclidean: ascending order (lower distance = better).
        // Expected order: id=10 (dist 0), id=20 (dist 1), id=30 (dist 2).
        assert_eq!(results[0].id, 10, "Closest should be id=10 (at origin)");
        assert_eq!(results[1].id, 20, "Second closest should be id=20");
        assert_eq!(results[2].id, 30, "Farthest should be id=30");

        // Verify actual distances
        assert!(
            results[0].score.abs() < 0.01,
            "Distance to origin should be ~0, got {}",
            results[0].score
        );
        assert!(
            (results[1].score - 1.0).abs() < 0.01,
            "Distance to (1,0,0,0) should be ~1, got {}",
            results[1].score
        );
    }
}

#[test]
#[cfg(feature = "gpu")]
fn test_brute_force_gpu_dot_product() {
    // RED: GPU brute-force must dispatch DotProduct, not hardcode Cosine.
    let index = HnswIndex::new(4, DistanceMetric::DotProduct).unwrap();

    // Insert vectors with different dot products to the query.
    index.insert(10, &[1.0, 0.0, 0.0, 0.0]); // dot with query = 1
    index.insert(20, &[2.0, 0.0, 0.0, 0.0]); // dot with query = 2
    index.insert(30, &[3.0, 0.0, 0.0, 0.0]); // dot with query = 3

    let query = [1.0, 0.0, 0.0, 0.0];

    if let Some(results) = index.search_brute_force_gpu(&query, 3).unwrap() {
        assert_eq!(results.len(), 3, "Should return 3 results");

        // DotProduct: descending order (higher = better).
        // Expected order: id=30 (dot 3), id=20 (dot 2), id=10 (dot 1).
        assert_eq!(
            results[0].id, 30,
            "Highest dot product should be id=30, got id={}",
            results[0].id
        );
        assert_eq!(
            results[1].id, 20,
            "Second highest dot product should be id=20, got id={}",
            results[1].id
        );
        assert_eq!(
            results[2].id, 10,
            "Lowest dot product should be id=10, got id={}",
            results[2].id
        );

        // Verify actual scores
        assert!(
            (results[0].score - 3.0).abs() < 0.01,
            "Dot product with (3,0,0,0) should be ~3, got {}",
            results[0].score
        );
    }
}

// =========================================================================
// FT-2: Property-Based Tests with proptest
// =========================================================================

mod proptest_tests {
    use super::*;
    use proptest::prelude::*;

    /// Strategy for generating valid vector dimensions (reasonable range)
    fn dimension_strategy() -> impl Strategy<Value = usize> {
        8usize..=256
    }

    /// Strategy for generating a random f32 vector of given dimension
    #[allow(dead_code)]
    fn vector_strategy(dim: usize) -> impl Strategy<Value = Vec<f32>> {
        proptest::collection::vec(-1.0f32..1.0, dim)
    }

    proptest! {
        #![proptest_config(ProptestConfig::with_cases(50))]

        /// Property: len() always equals number of successful insertions
        #[test]
        fn prop_len_equals_insertions(
            dim in dimension_strategy(),
            vectors in proptest::collection::vec(
                proptest::collection::vec(-1.0f32..1.0, 8usize..=64),
                1usize..=20
            )
        ) {
            let index = HnswIndex::new(dim, DistanceMetric::Euclidean).unwrap();
            let mut inserted = 0usize;

            for (i, v) in vectors.into_iter().enumerate() {
                if v.len() == dim {
                    index.insert(i as u64, &v);
                    inserted += 1;
                }
            }

            prop_assert_eq!(index.len(), inserted);
        }

        /// Property: search never returns more than k results
        #[test]
        fn prop_search_returns_at_most_k(
            dim in 16usize..=64,
            k in 1usize..=20,
            num_vectors in 5usize..=50
        ) {
            let index = HnswIndex::new(dim, DistanceMetric::Euclidean).unwrap();

            // Insert random vectors
            for i in 0..num_vectors {
                let v: Vec<f32> = (0..dim).map(|j| ((i + j) as f32 * 0.01).sin()).collect();
                index.insert(i as u64, &v);
            }

            let query: Vec<f32> = (0..dim).map(|j| (j as f32 * 0.02).cos()).collect();
            let results = index.search(&query, k);

            prop_assert!(results.len() <= k, "Search returned {} results, expected <= {}", results.len(), k);
        }

        /// Property: brute force search always returns exact results
        #[test]
        fn prop_brute_force_exact(
            dim in 8usize..=32,
            num_vectors in 3usize..=20
        ) {
            let index = HnswIndex::new(dim, DistanceMetric::Euclidean).unwrap();

            // Insert vectors with known distances from origin
            for i in 0..num_vectors {
                let mut v = vec![0.0f32; dim];
                v[0] = i as f32; // Distance from origin = i
                index.insert(i as u64, &v);
            }

            let query = vec![0.0f32; dim];
            let results = index.search_brute_force(&query, 3).unwrap();

            // First result should be id=0 (exact match at origin)
            if !results.is_empty() {
                prop_assert_eq!(results[0].id, 0, "Closest should be id=0 (at origin)");
            }
        }

        /// Property: remove always decreases len or returns false
        #[test]
        fn prop_remove_decreases_len(
            dim in 16usize..=32,
            id_to_remove in 0u64..10
        ) {
            let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();

            // Insert some vectors
            for i in 0u64..10 {
                let v: Vec<f32> = (0..dim).map(|j| ((i + j as u64) as f32 * 0.01).sin()).collect();
                index.insert(i, &v);
            }

            let len_before = index.len();
            let removed = index.remove(id_to_remove);

            if removed {
                prop_assert_eq!(index.len(), len_before - 1);
            } else {
                prop_assert_eq!(index.len(), len_before);
            }
        }

        /// Property: duplicate inserts are idempotent (no increase in len)
        #[test]
        fn prop_duplicate_insert_idempotent(
            dim in 16usize..=32
        ) {
            let index = HnswIndex::new(dim, DistanceMetric::Euclidean).unwrap();
            let v: Vec<f32> = (0..dim).map(|j| j as f32 * 0.1).collect();

            index.insert(42, &v);
            let len_after_first = index.len();

            index.insert(42, &v); // Duplicate
            let len_after_second = index.len();

            prop_assert_eq!(len_after_first, len_after_second, "Duplicate insert should be idempotent");
        }

        /// Property: batch insert count matches individual inserts
        #[test]
        fn prop_batch_insert_count(
            dim in 16usize..=32,
            batch_size in 5usize..=30
        ) {
            let index = HnswIndex::new(dim, DistanceMetric::Euclidean).unwrap();

            let batch: Vec<(u64, Vec<f32>)> = (0..batch_size)
                .map(|i| {
                    let v: Vec<f32> = (0..dim).map(|j| ((i + j) as f32 * 0.01).sin()).collect();
                    (i as u64, v)
                })
                .collect();

            // Use parallel insert (recommended API)
            let count = index.insert_batch_parallel(batch.iter().map(|(id, v)| (*id, v.as_slice())));

            prop_assert_eq!(count, batch_size, "Batch insert count mismatch");
            prop_assert_eq!(index.len(), batch_size, "Index len mismatch after batch");
        }
    }
}

// =========================================================================
// P1: Safety invariant tests for self-referential pattern
// =========================================================================
// NOTE: test_field_order_io_holder_after_inner is in index.rs (requires private field access)

/// Test that `ManuallyDrop` is used correctly for the inner field.
///
/// This verifies that:
/// 1. The inner field uses `ManuallyDrop` (checked by compilation)
/// 2. The custom Drop impl is present and correct
#[test]
fn test_manuallydrop_pattern_integrity() {
    // Create an index and verify it can be dropped without issues
    let index = HnswIndex::new(64, DistanceMetric::Cosine).unwrap();

    // Insert some data to ensure internal state is populated
    for i in 0..10 {
        let v: Vec<f32> = (0..64).map(|j| (i + j) as f32 * 0.01).collect();
        index.insert(i as u64, &v);
    }

    // Index must be fully populated and queryable right up to the drop point.
    assert_eq!(
        index.len(),
        10,
        "all 10 vectors must be present before drop"
    );
    let results = index.search(&vec![0.0f32; 64], 5);
    assert!(
        !results.is_empty() && results.len() <= 5,
        "index must be searchable before drop"
    );

    // Explicit drop exercises the custom Drop impl (unsafe ManuallyDrop::drop on the
    // in-memory native inner). A regression in drop ordering would panic/abort here.
    drop(index);
}

/// Test that loading from disk and dropping works correctly.
///
/// This is the actual use case where the self-referential pattern matters:
/// when loading from disk, `inner` borrows from `io_holder`.
#[test]
fn test_load_and_drop_safety() {
    use tempfile::TempDir;

    let temp_dir = TempDir::new().expect("Failed to create temp dir");
    let path = temp_dir.path();

    // Create, populate, and save an index
    {
        let index = HnswIndex::new(64, DistanceMetric::Cosine).unwrap();
        for i in 0..50 {
            let v: Vec<f32> = (0..64).map(|j| (i + j) as f32 * 0.01).collect();
            index.insert(i as u64, &v);
        }
        index.save(path).expect("Save failed");
    }

    // Load and drop multiple times to stress-test the drop order
    for _ in 0..3 {
        let loaded = HnswIndex::load(path, 64, DistanceMetric::Cosine).expect("Load failed");

        // Verify it works
        let results = loaded.search(&vec![0.0f32; 64], 5);
        assert!(!results.is_empty(), "Search should return results");

        // Drop happens here - critical that inner drops before io_holder
        drop(loaded);
    }
}

// =========================================================================
// Regression: adaptive search spread for distance metrics
// =========================================================================

/// Regression test: `search_adaptive` spread calculation must work correctly
/// for distance-based metrics (Euclidean, Hamming) where results are sorted
/// ascending (best = lowest score).
///
/// Before the fix, the spread formula `(max - min) / min.abs()` produced a
/// negative value for ascending-sorted results, which always compared < 2.0,
/// preventing escalation from ever triggering on distance metrics. The fix
/// uses `(score_a - score_b).abs() / min(|a|, |b|)` which is sign-agnostic.
#[test]
#[allow(clippy::cast_precision_loss)]
fn test_adaptive_search_spread_positive_for_distance_metrics() {
    // Euclidean: distance metric (lower = better, sorted ascending).
    // We need > 100 vectors so brute-force small-collection path is skipped,
    // and we need enough spread so the adaptive threshold (2.0) is exceeded.
    let dim = 32;
    let index = HnswIndex::new(dim, DistanceMetric::Euclidean).unwrap();

    // Insert 150 vectors: a tight cluster near the origin and outliers far away.
    // This creates high spread in search results, which should trigger escalation.
    for i in 0u64..120 {
        // Tight cluster: small perturbation around origin
        let v: Vec<f32> = (0..dim)
            .map(|j| ((i + j as u64) as f32 * 0.001).sin() * 0.1)
            .collect();
        index.insert(i, &v);
    }
    for i in 120u64..150 {
        // Outliers: far from origin (magnitude ~10)
        let v: Vec<f32> = (0..dim)
            .map(|j| 10.0 + (i + j as u64) as f32 * 0.05)
            .collect();
        index.insert(i, &v);
    }

    // Query near the cluster — first results will have low distances (~0.x),
    // last results high distances (~50+), creating a large spread.
    let query: Vec<f32> = vec![0.0; dim];

    // Adaptive search with a narrow min_ef and generous max_ef.
    // If spread calculation is correct and positive, the adaptive path will
    // either escalate or return valid results — either way, the search must
    // complete successfully and return sensible results.
    let results = index
        .search_with_quality(
            &query,
            10,
            SearchQuality::Adaptive {
                min_ef: 32,
                max_ef: 256,
            },
        )
        .unwrap();

    assert!(
        !results.is_empty(),
        "Adaptive Euclidean search should return results"
    );
    assert!(results.len() <= 10, "Should not exceed requested k");

    // Results must be sorted ascending (distance metric: lower = better).
    for pair in results.windows(2) {
        assert!(
            pair[0].score <= pair[1].score + f32::EPSILON,
            "Euclidean results must be sorted ascending: {} > {}",
            pair[0].score,
            pair[1].score,
        );
    }

    // The closest vectors should come from the tight cluster (ids 0..120),
    // not the outliers (ids 120..150).
    let closest_id = results[0].id;
    assert!(
        closest_id < 120,
        "Closest result should be from the tight cluster, got id={closest_id}"
    );
}

/// Regression test: adaptive search works for similarity metrics (Cosine)
/// with spread results — verifying the fix did not regress the happy path.
#[test]
#[allow(clippy::cast_precision_loss)]
fn test_adaptive_search_spread_works_for_similarity_metrics() {
    let dim = 32;
    let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();

    // 150 vectors with varying similarity to the query
    for i in 0u64..150 {
        let v: Vec<f32> = (0..dim)
            .map(|j| ((i * 7 + j as u64) as f32 * 0.013).sin())
            .collect();
        index.insert(i, &v);
    }

    let query: Vec<f32> = (0..dim).map(|j| (j as f32 * 0.013).sin()).collect();

    let results = index
        .search_with_quality(
            &query,
            10,
            SearchQuality::Adaptive {
                min_ef: 32,
                max_ef: 256,
            },
        )
        .unwrap();

    assert!(
        !results.is_empty(),
        "Adaptive Cosine search should return results"
    );
    assert!(results.len() <= 10, "Should not exceed requested k");

    // Results must be sorted descending (similarity metric: higher = better).
    for pair in results.windows(2) {
        assert!(
            pair[0].score >= pair[1].score - f32::EPSILON,
            "Cosine results must be sorted descending: {} < {}",
            pair[0].score,
            pair[1].score,
        );
    }
}

// -------------------------------------------------------------------------
// Upsert Semantics Tests (Issue #371)
// -------------------------------------------------------------------------

#[test]
fn test_insert_same_id_updates_vector() {
    // Arrange: create index with exact-distance features on (default)
    let index = HnswIndex::new(4, DistanceMetric::Cosine).unwrap();

    // Insert id=1 with vector A (pointing along x-axis)
    let vector_a = [1.0, 0.0, 0.0, 0.0];
    index.insert(1, &vector_a);

    // Act: insert id=1 again with vector B (pointing along y-axis, orthogonal to A)
    let vector_b = [0.0, 1.0, 0.0, 0.0];
    index.insert(1, &vector_b);

    // Assert 1: index length must still be 1 (not 2)
    assert_eq!(index.len(), 1, "Upsert must not create duplicate entries");

    // Assert 2: search with query=B should return id=1 with high similarity
    let results = index.search(&vector_b, 1);
    assert_eq!(results.len(), 1, "Should find exactly one result");
    assert_eq!(results[0].id, 1, "Result must be id=1");
    assert!(
        results[0].score > 0.9,
        "Similarity to updated vector B should be > 0.9, got {}",
        results[0].score,
    );
}

// -------------------------------------------------------------------------
// Upsert + Tombstone / Vacuum TDD Cycle 3 Tests
// -------------------------------------------------------------------------

#[test]
fn test_upsert_tombstone_accumulation() {
    // Arrange: 10-dim index so each of 10 IDs gets a unique dominant dimension
    let dim = 10;
    let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();

    // Insert 10 vectors (generation 0)
    for i in 0..10_usize {
        index.insert(i as u64, &make_dominant_vector(dim, i, 0));
    }

    // Assert: no tombstones, 10 active entries
    assert_eq!(
        index.tombstone_count(),
        0,
        "Fresh index should have 0 tombstones"
    );
    assert_eq!(index.len(), 10, "Should have 10 active vectors");

    // Act: update all 10 vectors with new values (generation 1)
    for i in 0..10_usize {
        index.insert(i as u64, &make_dominant_vector(dim, i, 1));
    }

    // Assert: 10 tombstones (old slots are dead), still 10 active
    assert_eq!(
        index.tombstone_count(),
        10,
        "Updating 10 vectors should leave 10 tombstones",
    );
    assert_eq!(index.len(), 10, "Active count must remain 10 after upserts");

    // Tombstone ratio = 10 / 20 = 0.50 => needs_vacuum (threshold 0.20)
    assert!(
        index.needs_vacuum(),
        "Tombstone ratio {:.2} should exceed 0.20 threshold",
        index.tombstone_ratio(),
    );
}

#[test]
fn test_upsert_then_vacuum_cleans() {
    // Arrange: 10-dim index so each of 10 IDs has a unique dominant dimension
    let dim = 10;
    let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();

    // Insert 10 vectors (generation 0)
    for i in 0..10_usize {
        index.insert(i as u64, &make_dominant_vector(dim, i, 0));
    }

    // Update only the first 5 (id=0..4) with generation 1
    for i in 0..5_usize {
        index.insert(i as u64, &make_dominant_vector(dim, i, 1));
    }

    // Pre-vacuum assertions
    assert_eq!(
        index.tombstone_count(),
        5,
        "Updating 5 vectors should create 5 tombstones",
    );
    assert_eq!(index.len(), 10);

    // Act: vacuum
    let rebuilt = index.vacuum().expect("vacuum should succeed");
    assert_eq!(rebuilt, 10, "Vacuum should report 10 active vectors");

    // Post-vacuum: tombstones cleared
    assert_eq!(
        index.tombstone_count(),
        0,
        "Vacuum must eliminate all tombstones",
    );
    assert_eq!(index.len(), 10, "All 10 vectors must survive vacuum");

    // Verify search correctness: each id appears in its own top-1 result
    // Use the *current* generation vector as query
    for i in 0..10_usize {
        let gen = u8::from(i < 5);
        let query = make_dominant_vector(dim, i, gen);
        let results = index.search(&query, 1);
        assert!(
            !results.is_empty(),
            "Search for id={i} returned no results after vacuum",
        );
        assert_eq!(
            results[0].id, i as u64,
            "Top-1 for id={i} should be itself, got id={}",
            results[0].id,
        );
    }
}

#[test]
fn test_batch_upsert_updates_existing() {
    // Arrange: 10-dim index so each of 10 IDs gets a unique dominant dimension
    let dim = 10;
    let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();

    // Insert 10 vectors (generation 0) via batch
    let original_vectors: Vec<(u64, Vec<f32>)> = (0..10)
        .map(|id| (id, make_dominant_vector(dim, id as usize, 0)))
        .collect();

    let refs: Vec<(u64, &[f32])> = original_vectors
        .iter()
        .map(|(id, v)| (*id, v.as_slice()))
        .collect();

    let inserted = index.insert_batch_parallel(refs);
    assert_eq!(inserted, 10, "Should insert all 10 vectors");
    assert_eq!(index.len(), 10, "Index should contain 10 entries");

    // Act: update first 5 (id=0..4) with generation 1 via batch
    let updated_vectors: Vec<(u64, Vec<f32>)> = (0..5)
        .map(|id| (id, make_dominant_vector(dim, id as usize, 1)))
        .collect();

    let update_refs: Vec<(u64, &[f32])> = updated_vectors
        .iter()
        .map(|(id, v)| (*id, v.as_slice()))
        .collect();

    let upserted = index.insert_batch_parallel(update_refs);
    assert_eq!(upserted, 5, "Should upsert all 5 vectors");

    // Assert 1: index length must still be 10 (not 15)
    assert_eq!(
        index.len(),
        10,
        "Batch upsert must not create duplicate entries: expected 10, got {}",
        index.len(),
    );

    // Assert 2: for each updated id (0..4), search with generation 1 vector
    for id in 0..5_u64 {
        let query = make_dominant_vector(dim, id as usize, 1);
        let results = index.search(&query, 1);
        assert_eq!(
            results.len(),
            1,
            "Updated id={id}: should find exactly one result"
        );
        assert_eq!(
            results[0].id, id,
            "Updated id={id}: top-1 result should be the updated vector"
        );
        assert!(
            results[0].score > 0.9,
            "Updated id={id}: similarity to updated vector should be > 0.9, got {}",
            results[0].score,
        );
    }

    // Assert 3: for each non-updated id (5..9), search with generation 0 vector
    for id in 5..10_u64 {
        let query = make_dominant_vector(dim, id as usize, 0);
        let results = index.search(&query, 1);
        assert!(
            !results.is_empty(),
            "Non-updated id={id}: should find results"
        );
        assert_eq!(
            results[0].id, id,
            "Non-updated id={id}: top-1 result should be the original vector"
        );
    }
}

// -------------------------------------------------------------------------
// Upsert TDD Cycle 5: Recall Verification + Edge Cases
// -------------------------------------------------------------------------

/// Verifies that after updating a vector to a different cluster, search
/// finds it in the NEW location and no longer returns it from the OLD cluster.
#[allow(clippy::similar_names)] // origin/target are semantically distinct cluster labels
#[test]
fn test_upsert_search_recall() {
    let index = HnswIndex::new(8, DistanceMetric::Cosine).unwrap();

    // Origin cluster center: dominant component at dim 0
    let origin_center: Vec<f32> = vec![1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0];
    // Target cluster center: dominant component at dim 4
    let target_center: Vec<f32> = vec![0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0];

    // Build deterministic vectors with small perturbation per id.
    // Origin cluster: ids 0..49, base near origin_center
    let origin_vectors: Vec<(u64, Vec<f32>)> = (0..50)
        .map(|i| {
            let mut v = origin_center.clone();
            for (d, component) in v.iter_mut().enumerate() {
                *component += 0.01 * (i * 8 + d) as f32;
            }
            (i as u64, v)
        })
        .collect();

    // Target cluster: ids 50..99, base near target_center
    let target_vectors: Vec<(u64, Vec<f32>)> = (0..50)
        .map(|i| {
            let mut v = target_center.clone();
            for (d, component) in v.iter_mut().enumerate() {
                *component += 0.01 * (i * 8 + d) as f32;
            }
            (i as u64 + 50, v)
        })
        .collect();

    // Insert all 100 vectors via batch
    let all_vectors: Vec<(u64, Vec<f32>)> = origin_vectors
        .iter()
        .chain(target_vectors.iter())
        .map(|(id, v)| (*id, v.clone()))
        .collect();

    let refs: Vec<(u64, &[f32])> = all_vectors
        .iter()
        .map(|(id, v)| (*id, v.as_slice()))
        .collect();

    let inserted = index.insert_batch_parallel(refs);
    assert_eq!(inserted, 100, "Should insert all 100 vectors");

    // Update id=25 (origin cluster) to a target cluster vector
    let moved_vector: Vec<f32> = vec![0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0];
    index.insert(25, &moved_vector);

    assert_eq!(index.len(), 100, "Length must remain 100 after upsert");

    // Search near target cluster center: id=25 should appear (it moved there)
    let results_target = index.search(&target_center, 10);
    let found_in_target = results_target.iter().any(|r| r.id == 25);
    assert!(
        found_in_target,
        "id=25 should appear in target cluster results after upsert, got ids: {:?}",
        results_target.iter().map(|r| r.id).collect::<Vec<_>>(),
    );

    // Search near origin cluster center: id=25 should NOT appear (it left)
    let results_origin = index.search(&origin_center, 10);
    let found_in_origin = results_origin.iter().any(|r| r.id == 25);
    assert!(
        !found_in_origin,
        "id=25 should NOT appear in origin cluster results after upsert, got ids: {:?}",
        results_origin.iter().map(|r| r.id).collect::<Vec<_>>(),
    );
}

/// Verifies that updating the very first vector (likely the HNSW entry point)
/// does not break search for any vector in the index.
#[test]
fn test_upsert_entry_point_vector() {
    let index = HnswIndex::new(4, DistanceMetric::Cosine).unwrap();

    // Insert id=0 first — this becomes the HNSW entry point
    let ep_original = vec![1.0, 0.0, 0.0, 0.0];
    index.insert(0, &ep_original);

    // Update id=0 to a completely different direction
    let ep_updated = vec![0.0, 0.0, 0.0, 1.0];
    index.insert(0, &ep_updated);

    // Insert 20 more vectors with varied, well-separated values.
    // Each vector has a dominant component determined by (id % 4).
    let additional_vectors: Vec<(u64, Vec<f32>)> = (1..=20)
        .map(|i| {
            let dominant_dim = (i as usize) % 4;
            let mut v = vec![0.1_f32; 4];
            v[dominant_dim] += 1.0 + 0.05 * i as f32;
            (i as u64, v)
        })
        .collect();

    let refs: Vec<(u64, &[f32])> = additional_vectors
        .iter()
        .map(|(id, v)| (*id, v.as_slice()))
        .collect();

    let inserted = index.insert_batch_parallel(refs);
    assert_eq!(inserted, 20, "Should insert all 20 vectors");
    assert_eq!(index.len(), 21, "Index should contain 21 entries");

    // Verify: searching for the updated id=0 vector returns id=0 as top-1
    let results_ep = index.search(&ep_updated, 1);
    assert_eq!(results_ep.len(), 1, "Should find at least 1 result for EP");
    assert_eq!(
        results_ep[0].id, 0,
        "Top-1 for updated entry point should be id=0, got id={}",
        results_ep[0].id,
    );

    // Verify: each of the 20 additional vectors is findable as top-1
    for (id, vec) in &additional_vectors {
        let results = index.search(vec.as_slice(), 1);
        assert!(
            !results.is_empty(),
            "Search for id={id} should return results",
        );
        assert_eq!(
            results[0].id, *id,
            "Top-1 for id={id} should be itself, got id={}",
            results[0].id,
        );
    }
}

// -------------------------------------------------------------------------
// Upsert Edge Case Tests (Issue #371)
// -------------------------------------------------------------------------

/// Verifies that batch insert with within-batch duplicate IDs correctly
/// deduplicates: only the last occurrence is reachable, and the count
/// reflects unique IDs (not raw entries). Within-batch duplicates are an
/// unusual pattern but must not corrupt mapping state.
#[test]
fn test_batch_upsert_within_batch_duplicates() {
    let dim = 4;
    let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();

    // Batch with id=1 appearing twice: first [1,0,0,0], then [0,1,0,0]
    let vec_a = vec![1.0_f32, 0.0, 0.0, 0.0];
    let vec_b = vec![0.0_f32, 1.0, 0.0, 0.0];
    let batch: Vec<(u64, &[f32])> = vec![(1, &vec_a), (1, &vec_b), (2, &vec_a)];

    let inserted = index.insert_batch_parallel(batch);
    // Count may include both entries for id=1, but len() must reflect unique IDs
    assert!(
        inserted >= 2,
        "At least 2 entries processed, got {inserted}"
    );
    assert_eq!(index.len(), 2, "Only 2 unique IDs should be mapped");

    // id=1 must be searchable and point to vec_b (last wins)
    let results = index.search(&vec_b, 1);
    assert_eq!(results.len(), 1);
    assert_eq!(
        results[0].id, 1,
        "id=1 must be findable after within-batch upsert"
    );
    assert!(
        results[0].score > 0.9,
        "id=1 should match vec_b, got score {}",
        results[0].score,
    );

    // id=2 must also be searchable
    let results2 = index.search(&vec_a, 1);
    assert_eq!(results2[0].id, 2, "id=2 must be findable");
}

/// Creates a deterministic vector where dimension `dominant_dim` has a
/// large component, making it easily distinguishable in cosine search.
///
/// Shared helper to avoid duplicating the same closure across tests.
fn make_dominant_vector(dim: usize, dominant_dim: usize, gen: u8) -> Vec<f32> {
    let mut v = vec![0.1_f32; dim];
    v[dominant_dim % dim] += 1.0 + f32::from(gen) * 0.5;
    v
}

/// Verifies that insert(id) after remove(id) correctly re-inserts
/// the vector and returns it in search results.
#[test]
fn test_upsert_after_delete() {
    let dim = 8;
    let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();

    // Insert id=1 with dominant dim 0
    let vec_a = make_dominant_vector(dim, 0, 0);
    index.insert(1, &vec_a);
    assert_eq!(index.len(), 1);

    // Remove id=1
    assert!(index.remove(1));
    assert_eq!(index.len(), 0);

    // Re-insert id=1 with dominant dim 4
    let vec_b = make_dominant_vector(dim, 4, 1);
    index.insert(1, &vec_b);
    assert_eq!(index.len(), 1, "Re-inserted id should count as 1 vector");

    // Search should find id=1 near vec_b
    let results = index.search(&vec_b, 1);
    assert_eq!(results.len(), 1);
    assert_eq!(
        results[0].id, 1,
        "After delete + re-insert, search must find re-inserted id=1"
    );
}

/// Verifies that upserting the same id N times leaves exactly N-1
/// tombstones and the index still contains exactly 1 live vector
/// that matches the latest version.
#[test]
fn test_repeated_upsert_accumulates_tombstones() {
    let dim = 8;
    let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();
    let num_upserts: usize = 20;

    for gen in 0..num_upserts {
        let v = make_dominant_vector(dim, gen % dim, gen as u8);
        index.insert(1, &v);
    }

    assert_eq!(
        index.len(),
        1,
        "Repeated upserts must not duplicate entries"
    );
    assert_eq!(
        index.tombstone_count(),
        num_upserts - 1,
        "Each upsert after the first should leave one tombstone"
    );

    // Search must return the latest vector (dominant dim = (num_upserts-1) % dim)
    let latest = make_dominant_vector(dim, (num_upserts - 1) % dim, (num_upserts - 1) as u8);
    let results = index.search(&latest, 1);
    assert_eq!(results.len(), 1);
    assert_eq!(
        results[0].id, 1,
        "Search must return the latest upserted vector"
    );
}

// -------------------------------------------------------------------------
// #2246: an id is mapped to the slot the graph gave its vector
// -------------------------------------------------------------------------

/// A slot the graph filled with no id mapped to it — what another writer's
/// node looks like from here — never captures the next id.
///
/// Before #2246 the mappings predicted slot 1 for id 200 while the graph
/// handed out slot 2, and a correction step had to repair the mapping
/// afterwards. The mapping now names the slot the graph returned.
#[test]
fn test_an_id_is_mapped_to_the_slot_the_graph_gave() {
    let dim = 4;
    let index = HnswIndex::new(dim, DistanceMetric::Euclidean).unwrap();

    index.insert(100, &[1.0, 0.0, 0.0, 0.0]);
    assert_eq!(index.mappings.get_idx(100), Some(0));
    assert_eq!(index.mappings.get_id(0), Some(100));

    let unmapped = index.inner.read().insert(&[0.0, 1.0, 0.0, 0.0]).unwrap();
    assert_eq!(unmapped, 1, "the graph hands out the next slot");

    assert!(index.insert_and_assign(200, &[0.0, 0.0, 1.0, 0.0]));

    assert_eq!(
        index.mappings.get_idx(200),
        Some(2),
        "id 200 must name the slot holding its vector"
    );
    assert_eq!(index.mappings.get_id(2), Some(200));
    assert_eq!(
        index.mappings.get_id(1),
        None,
        "the unmapped slot stays unmapped"
    );
    assert_eq!(index.len(), 2);
}

/// Search resolves an id placed after an unmapped slot to that id.
#[test]
fn test_search_finds_an_id_placed_after_an_unmapped_slot() {
    let dim = 4;
    let index = HnswIndex::new(dim, DistanceMetric::Euclidean).unwrap();

    index.insert(100, &[1.0, 0.0, 0.0, 0.0]);
    index.inner.read().insert(&[0.5, 0.5, 0.0, 0.0]).unwrap();
    index.insert(200, &[0.0, 0.0, 0.0, 1.0]);

    let results = index.search(&[0.0, 0.0, 0.0, 1.0], 1);
    assert_eq!(results.len(), 1);
    assert_eq!(results[0].id, 200, "search must resolve the slot to id 200");
}

/// An id repeated within one batch ends on its last occurrence, and the slot
/// of its first is retired rather than left answering for it.
#[test]
fn test_a_batch_repeating_an_id_keeps_its_last_vector() {
    let index = HnswIndex::new(4, DistanceMetric::Euclidean).unwrap();
    let batch: Vec<(u64, &[f32])> = vec![
        (7, &[1.0, 0.0, 0.0, 0.0]),
        (8, &[0.0, 1.0, 0.0, 0.0]),
        (7, &[0.0, 0.0, 1.0, 0.0]),
    ];
    assert_eq!(index.insert_batch_parallel(batch), 3);

    assert_eq!(index.len(), 2);
    assert_eq!(
        index.mappings.get_idx(7),
        Some(2),
        "id 7 follows its last occurrence"
    );
    assert_eq!(
        index.mappings.get_id(0),
        None,
        "id 7's first slot must be retired"
    );
    let results = index.search(&[0.0, 0.0, 1.0, 0.0], 1);
    assert_eq!(results[0].id, 7);
}

// =========================================================================
// Issue #396: a batch after single inserts — each id follows its slot
// =========================================================================

/// Regression test (#396): a batch after single inserts maps each id to the
/// slot the graph gave its vector.
#[test]
fn test_batch_after_single_insert_mapping_consistency() {
    let index = HnswIndex::new(4, DistanceMetric::Euclidean).unwrap();

    // Single-insert 1 vector: consumes graph node 0
    index.insert(100, &[1.0, 0.0, 0.0, 0.0]);
    assert_eq!(index.len(), 1);

    // Batch-insert 5 vectors; the graph places them after node 0.
    let batch: Vec<(u64, &[f32])> = vec![
        (200, &[0.0, 1.0, 0.0, 0.0]),
        (201, &[0.0, 0.0, 1.0, 0.0]),
        (202, &[0.0, 0.0, 0.0, 1.0]),
        (203, &[0.5, 0.5, 0.0, 0.0]),
        (204, &[0.0, 0.5, 0.5, 0.0]),
    ];
    let inserted = index.insert_batch_parallel(batch);
    assert_eq!(inserted, 5);
    assert_eq!(index.len(), 6);

    // Every external ID must resolve to a valid internal idx, and that idx
    // must map back to the same external ID (bidirectional consistency).
    for &ext_id in &[100u64, 200, 201, 202, 203, 204] {
        let idx = index.mappings.get_idx(ext_id);
        assert!(idx.is_some(), "get_idx({ext_id}) must return Some");
        let reverse = index.mappings.get_id(idx.unwrap());
        assert_eq!(
            reverse,
            Some(ext_id),
            "Reverse mapping for idx {} must be {ext_id}, got {reverse:?}",
            idx.unwrap()
        );
    }

    // Verify search still returns correct results — the original vector
    // should be the nearest to itself.
    let results = index.search(&[1.0, 0.0, 0.0, 0.0], 1);
    assert_eq!(results.len(), 1);
    assert_eq!(results[0].id, 100, "Nearest to [1,0,0,0] must be id=100");
}

/// Regression test: graph vectors readable at graph-assigned IDs.
///
/// The vector each mapping resolves to (in the graph's `ContiguousVectors`)
/// must be the one inserted for that external ID.
#[test]
fn test_batch_insert_vector_storage_uses_assigned_ids() {
    let index = HnswIndex::new(4, DistanceMetric::Euclidean).unwrap();

    // Pre-populate to create a gap between mapping indices and graph node IDs
    for i in 0..3u64 {
        let v = [i as f32, 0.0, 0.0, 0.0];
        index.insert(i, &v);
    }
    assert_eq!(index.len(), 3);

    // Batch-insert 4 vectors
    let batch: Vec<(u64, &[f32])> = vec![
        (10, &[10.0, 0.0, 0.0, 0.0]),
        (11, &[11.0, 0.0, 0.0, 0.0]),
        (12, &[12.0, 0.0, 0.0, 0.0]),
        (13, &[13.0, 0.0, 0.0, 0.0]),
    ];
    let inserted = index.insert_batch_parallel(batch);
    assert_eq!(inserted, 4);

    // For each batch-inserted ID, the graph vector at the mapped idx
    // must exist and match the original vector.
    for ext_id in 10..=13u64 {
        let idx = index.mappings.get_idx(ext_id).expect("mapping must exist");
        let stored = index
            .inner
            .read()
            .with_contiguous_vectors(|cv| cv.get(idx).map(<[f32]>::to_vec));
        assert!(
            stored.is_some(),
            "ContiguousVectors must have a vector at idx {idx} for id {ext_id}"
        );
        let expected_first = ext_id as f32;
        let stored_vec = stored.unwrap();
        assert!(
            (stored_vec[0] - expected_first).abs() < f32::EPSILON,
            "Vector for id {ext_id} at idx {idx}: expected first component {expected_first}, got {}",
            stored_vec[0]
        );
    }
}

/// Regression test: batch upsert (insert + replace) maintains mapping consistency.
///
/// Insert 5 vectors, then batch-upsert 3 of them with new data. The replaced
/// vectors must have updated mappings and the new graph node IDs must be valid.
#[test]
fn test_batch_upsert_mapping_consistency() {
    let index = HnswIndex::new(4, DistanceMetric::Euclidean).unwrap();

    // Initial insert of 5 vectors
    for i in 0..5u64 {
        index.insert(i, &[i as f32, 0.0, 0.0, 0.0]);
    }
    assert_eq!(index.len(), 5);

    // Batch-upsert: update IDs 1, 2, 3 with new vectors
    let batch: Vec<(u64, &[f32])> = vec![
        (1, &[100.0, 0.0, 0.0, 0.0]),
        (2, &[200.0, 0.0, 0.0, 0.0]),
        (3, &[300.0, 0.0, 0.0, 0.0]),
    ];
    let inserted = index.insert_batch_parallel(batch);
    assert_eq!(inserted, 3);
    // Still 5 live IDs (3 replaced, not added)
    assert_eq!(index.len(), 5);

    // Verify all 5 IDs have consistent bidirectional mappings
    for ext_id in 0..5u64 {
        let idx = index.mappings.get_idx(ext_id);
        assert!(idx.is_some(), "get_idx({ext_id}) must return Some");
        let reverse = index.mappings.get_id(idx.unwrap());
        assert_eq!(
            reverse,
            Some(ext_id),
            "Reverse mapping for idx {} must be {ext_id}, got {reverse:?}",
            idx.unwrap()
        );
    }

    // The replaced vectors should be searchable with their new values.
    // Search for [100, 0, 0, 0] — nearest should be id=1.
    let results = index.search(&[100.0, 0.0, 0.0, 0.0], 1);
    assert_eq!(results.len(), 1);
    assert_eq!(results[0].id, 1, "Nearest to [100,0,0,0] must be id=1");
}

// =========================================================================
// Bitmap pre-filter tests for search_with_quality_and_bitmap
// =========================================================================

#[test]
fn test_search_with_quality_and_bitmap_exists() {
    // Arrange: create an index with several vectors
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    for i in 0u64..20 {
        #[allow(clippy::cast_precision_loss)]
        let v = vec![i as f32 * 0.1, 1.0 - i as f32 * 0.05, 0.5];
        index.insert(i, &v);
    }

    // Build a bitmap containing IDs 0..10
    let mut bitmap = roaring::RoaringBitmap::new();
    for i in 0u32..10 {
        bitmap.insert(i);
    }

    // Act
    let results = index
        .search_with_quality_and_bitmap(&[0.5, 0.5, 0.5], 5, SearchQuality::Balanced, &bitmap)
        .unwrap();

    // Assert: returns results, all IDs in bitmap
    assert!(!results.is_empty(), "should return results");
    for sr in &results {
        assert!(
            sr.id < 10,
            "all result IDs should be in bitmap, got id={}",
            sr.id
        );
    }
}

#[test]
fn test_ids_exceeding_u32_max_pass_through() {
    // This test verifies the u32::MAX passthrough logic.
    // Since HnswIndex uses u64 IDs but RoaringBitmap only holds u32,
    // IDs > u32::MAX should pass through unconditionally.
    //
    // We test the filtering logic directly: create an index with a
    // normal ID and verify the bitmap filter works correctly.
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();

    // Insert vectors with normal IDs
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.0, 1.0, 0.0]);
    index.insert(3, &[0.0, 0.0, 1.0]);

    // Bitmap contains only ID 1
    let mut bitmap = roaring::RoaringBitmap::new();
    bitmap.insert(1);

    let results = index
        .search_with_quality_and_bitmap(&[1.0, 0.0, 0.0], 3, SearchQuality::Balanced, &bitmap)
        .unwrap();

    // All results should have id=1 (the only one in bitmap)
    // IDs 2 and 3 are valid u32 but NOT in bitmap, so they're excluded
    for sr in &results {
        assert_eq!(
            sr.id, 1,
            "only id=1 should pass bitmap filter, got {}",
            sr.id
        );
    }
}

#[cfg(test)]
mod bitmap_property_tests {
    use super::*;
    use proptest::prelude::*;

    proptest! {
        #![proptest_config(ProptestConfig::with_cases(100))]

        /// **Validates: Requirements 1.1, 1.2, 2.2**
        ///
        /// Feature: bitmap-prefilter-v2, Property 1: Bitmap filtering invariant
        ///
        /// For any non-empty bitmap and any SearchQuality, all ScoredResult
        /// returned by search_with_quality_and_bitmap have an id present in
        /// the bitmap (via u32 conversion) or an id exceeding u32::MAX.
        #[test]
        fn prop_bitmap_filtering_invariant(
            bitmap_ids in proptest::collection::vec(0u32..200, 1..50),
            quality_idx in 0u8..3,
        ) {
            let dim = 8;
            let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();

            // Insert 200 vectors
            for i in 0u64..200 {
                #[allow(clippy::cast_precision_loss)]
                let v: Vec<f32> = (0..dim).map(|d| ((i + d as u64) as f32 * 0.01).sin()).collect();
                index.insert(i, &v);
            }

            let mut bitmap = roaring::RoaringBitmap::new();
            for &id in &bitmap_ids {
                bitmap.insert(id);
            }

            let quality = match quality_idx {
                0 => SearchQuality::Fast,
                1 => SearchQuality::Balanced,
                _ => SearchQuality::Accurate,
            };

            let query: Vec<f32> = (0..dim).map(|d| (d as f32 * 0.05).cos()).collect();
            let results = index
                .search_with_quality_and_bitmap(&query, 10, quality, &bitmap)
                .unwrap();

            for sr in &results {
                let in_bitmap = u32::try_from(sr.id)
                    .is_ok_and(|id32| bitmap.contains(id32));
                let exceeds_u32 = u32::try_from(sr.id).is_err();
                prop_assert!(
                    in_bitmap || exceeds_u32,
                    "result id={} must be in bitmap or exceed u32::MAX",
                    sr.id
                );
            }
        }
    }
}

// =========================================================================
// Full-scan with bitmap tests
// =========================================================================

#[test]
fn test_full_scan_empty_bitmap_returns_empty() {
    let index = HnswIndex::new(3, DistanceMetric::Cosine).unwrap();
    index.insert(1, &[1.0, 0.0, 0.0]);
    index.insert(2, &[0.0, 1.0, 0.0]);

    let bitmap = roaring::RoaringBitmap::new();
    let results = index
        .full_scan_with_bitmap(&[1.0, 0.0, 0.0], 5, &bitmap)
        .unwrap();
    assert!(
        results.is_empty(),
        "empty bitmap should return empty results"
    );
}

#[test]
#[allow(clippy::cast_precision_loss)]
fn test_full_scan_returns_exact_results() {
    let dim = 8;
    let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();

    // Insert 20 vectors with distinct patterns
    for i in 0u64..20 {
        let v: Vec<f32> = (0..dim)
            .map(|d| ((i + d as u64) as f32 * 0.1).sin())
            .collect();
        index.insert(i, &v);
    }

    // Bitmap: only IDs 0..5
    let mut bitmap = roaring::RoaringBitmap::new();
    for i in 0u32..5 {
        bitmap.insert(i);
    }

    let query: Vec<f32> = (0..dim).map(|d| (d as f32 * 0.1).sin()).collect();
    let results = index.full_scan_with_bitmap(&query, 3, &bitmap).unwrap();

    // Should return at most 3 results, all from bitmap
    assert!(!results.is_empty(), "should return results");
    assert!(results.len() <= 3, "should return at most k results");
    for sr in &results {
        assert!(
            sr.id < 5,
            "all results should be from bitmap, got id={}",
            sr.id
        );
    }

    // First result should be id=0 (exact match with query pattern)
    assert_eq!(results[0].id, 0, "closest vector should be id=0");
}

#[cfg(test)]
mod full_scan_property_tests {
    use super::*;
    use proptest::prelude::*;

    proptest! {
        #![proptest_config(ProptestConfig::with_cases(100))]

        /// **Validates: Requirements 3.5, 5.2, 3.2**
        ///
        /// Feature: bitmap-prefilter-v2, Property 4: Full-scan exactness
        ///
        /// For any set of vectors and any bitmap, the full-scan results are
        /// identical to a naive exhaustive distance computation on the same
        /// bitmap vectors, sorted by the metric.
        #[test]
        #[expect(clippy::significant_drop_tightening)] // Reason: the guard under test is held to the assertion on purpose
        fn prop_full_scan_exactness(
            n_vectors in 5u64..50,
            bitmap_ids in proptest::collection::vec(0u32..50, 1..20),
            k in 1usize..10,
        ) {
            let dim = 4;
            let index = HnswIndex::new(dim, DistanceMetric::Cosine).unwrap();

            // Insert vectors
            for i in 0..n_vectors {
                #[allow(clippy::cast_precision_loss)]
                let v: Vec<f32> = (0..dim)
                    .map(|d| ((i + d as u64) as f32 * 0.07).sin())
                    .collect();
                index.insert(i, &v);
            }

            let mut bitmap = roaring::RoaringBitmap::new();
            for &id in &bitmap_ids {
                if u64::from(id) < n_vectors {
                    bitmap.insert(id);
                }
            }

            if bitmap.is_empty() {
                // Skip: empty bitmap always returns empty
                return Ok(());
            }

            #[allow(clippy::cast_precision_loss)]
            let query: Vec<f32> = (0..dim).map(|d| (d as f32 * 0.05).cos()).collect();

            let results = index
                .full_scan_with_bitmap(&query, k, &bitmap)
                .unwrap();

            // Naive exhaustive computation
            let mut naive: Vec<(u64, f32)> = Vec::new();
            for id32 in &bitmap {
                let id = u64::from(id32);
                if let Some(idx) = index.mappings.get_idx(id) {
                    let inner = index.inner.read();
                    let score = inner.with_contiguous_vectors(|vectors| {
                        vectors.get(idx).map(|v| index.compute_distance(&query, v))
                    });
                    if let Some(s) = score {
                        naive.push((id, s));
                    }
                }
            }

            // Sort naive by cosine (higher is better → descending)
            naive.sort_by(|a, b| b.1.total_cmp(&a.1));
            naive.truncate(k);

            // Compare IDs and scores
            prop_assert_eq!(
                results.len(),
                naive.len(),
                "result count mismatch"
            );
            for (sr, (expected_id, expected_score)) in results.iter().zip(naive.iter()) {
                prop_assert_eq!(sr.id, *expected_id, "ID mismatch");
                prop_assert!(
                    (sr.score - expected_score).abs() < 1e-6,
                    "score mismatch for id={}: got {}, expected {}",
                    sr.id, sr.score, expected_score
                );
            }
        }
    }
}

// =========================================================================
// RaBitQ storage-mode save/load round-trip (quantization wiring)
// =========================================================================

/// `save` must persist the actual backend storage mode (not hardcoded Full)
/// and `load` must rebuild the RaBitQ backend and install `rabitq.idx`.
#[cfg(feature = "persistence")]
#[test]
fn test_hnsw_save_load_roundtrips_rabitq_backend_and_quantizer() {
    use crate::quantization::{RaBitQIndex, StorageMode};

    let dir = tempfile::tempdir().expect("temp dir");
    let dim = 32;
    let mut params = HnswParams::auto(dim);
    params.storage_mode = StorageMode::RaBitQ;
    let index = HnswIndex::with_params(dim, DistanceMetric::Euclidean, params).expect("create");

    let vectors: Vec<Vec<f32>> = (0..200)
        .map(|i| {
            (0..dim)
                .map(|d| ((i * dim + d) as f32 * 0.01).sin())
                .collect()
        })
        .collect();
    for (i, v) in vectors.iter().enumerate() {
        VectorIndex::insert(&index, i as u64, v);
    }

    // Persist a trained quantizer next to the index (what TRAIN QUANTIZER does).
    RaBitQIndex::train(&vectors, 42)
        .expect("train")
        .save(dir.path())
        .expect("save rabitq.idx");
    index.save(dir.path()).expect("save index");

    let loaded = HnswIndex::load(dir.path(), dim, DistanceMetric::Euclidean).expect("load");
    assert!(
        loaded.is_rabitq_quantizer_trained(),
        "persisted storage mode + rabitq.idx must restore a trained RaBitQ backend"
    );

    let results = VectorIndex::search(&loaded, &vectors[42], 5);
    assert_eq!(
        results.first().map(|r| r.id),
        Some(42),
        "self-query must return itself as top-1 after reload"
    );
}

#[test]
fn test_with_params_propagates_alpha_to_native_graph() {
    let params = HnswParams::auto(4).with_alpha(1.5);
    let index = HnswIndex::with_params(4, crate::DistanceMetric::Cosine, params)
        .expect("with_params must succeed with valid alpha");
    let actual = index.inner.read().alpha();
    assert!(
        (actual - 1.5).abs() < f32::EPSILON,
        "alpha 1.5 must propagate to native graph, got {actual}"
    );
}

// =========================================================================
// Resumable adaptive escalation + bitmap under-fill retry (#2077)
// =========================================================================

/// Deterministic pseudo-random unit vector for the resume/retry tests.
fn resume_test_vector(dim: usize, seed: u64) -> Vec<f32> {
    let mut state = seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1;
    let mut v: Vec<f32> = (0..dim)
        .map(|_| {
            state ^= state << 13;
            state ^= state >> 7;
            state ^= state << 17;
            (state as f64 / u64::MAX as f64) as f32 - 0.5
        })
        .collect();
    let norm = v.iter().map(|x| x * x).sum::<f32>().sqrt();
    for x in &mut v {
        *x /= norm;
    }
    v
}

/// The bitmap under-fill retry must widen `ef`, not just `candidates_k`:
/// the graph's result heap is bounded by `ef`, so a same-ef retry re-runs
/// the identical traversal and returns the identical candidate set. This
/// test pins both halves of that statement.
#[test]
fn bitmap_underfill_retry_widens_ef_and_finds_more_matches() {
    const DIM: usize = 64;
    const N: u64 = 1_500;
    const K: usize = 20;

    let index = HnswIndex::new(DIM, DistanceMetric::Cosine).unwrap();
    for id in 0..N {
        index.insert(id, &resume_test_vector(DIM, id + 1));
    }
    let query = resume_test_vector(DIM, 987_654_321);

    // Exact cosine ranking of all ids against the query.
    let mut ranked: Vec<(u64, f32)> = (0..N)
        .map(|id| {
            let v = resume_test_vector(DIM, id + 1);
            let dot: f32 = query.iter().zip(&v).map(|(a, b)| a * b).sum();
            (id, dot)
        })
        .collect();
    ranked.sort_by(|a, b| b.1.total_cmp(&a.1));

    // Allow only a mid-far band: mostly outside the base-ef candidate set,
    // reachable when the traversal digs deeper.
    let allowed: Vec<u64> = ranked[300..380].iter().map(|&(id, _)| id).collect();
    let mut bitmap = roaring::RoaringBitmap::new();
    for &id in &allowed {
        bitmap.insert(u32::try_from(id).unwrap());
    }

    let quality = SearchQuality::Balanced;
    let ef = quality.ef_search_for_scale(K, index.len());
    let candidates_k = K.saturating_mul(4).max(ef);

    // Mechanism pin: doubling candidates_k at the SAME ef returns the exact
    // same survivor set — candidates_k beyond ef only raises a truncation
    // limit the ef-bounded heap never reaches.
    let base = index.search_hnsw_only_filtered(&query, candidates_k, ef, &bitmap);
    let same_ef_double_k = index.search_hnsw_only_filtered(&query, candidates_k * 2, ef, &bitmap);
    let base_ids: Vec<u64> = base.iter().map(|r| r.id).collect();
    let double_k_ids: Vec<u64> = same_ef_double_k.iter().map(|r| r.id).collect();
    assert_eq!(
        base_ids, double_k_ids,
        "same-ef retry must be a no-op: candidates_k is not the exploration knob"
    );

    // The band is chosen so the base pass under-fills — otherwise the retry
    // (the code under test) never runs.
    assert!(
        base.len() < K,
        "test setup: base pass must under-fill (got {} >= K={K})",
        base.len()
    );

    // Fix pin: the full path (base + widened-ef retry) surfaces strictly
    // more allowed matches than the base pass alone.
    let full = index
        .search_with_quality_and_bitmap(&query, K, quality, &bitmap)
        .unwrap();
    assert!(
        full.len() > base.len(),
        "widened-ef retry must find more allowed matches (base {}, full {})",
        base.len(),
        full.len()
    );
}

/// Repeated adaptive searches must be bit-identical: the resumed traversal
/// hands its pooled heaps and visited set back on drop, so any pool
/// pollution (a visited set returned uncleared, a heap kept warm with stale
/// entries) would show up as run-to-run drift here.
#[test]
fn adaptive_resume_is_deterministic_across_pool_reuse() {
    const DIM: usize = 128;
    const N: u64 = 1_200;
    const K: usize = 10;

    let index = HnswIndex::new(DIM, DistanceMetric::Cosine).unwrap();
    // A tight 4-vector pocket around a distinct direction plus random
    // background: the pocket dominates the top of the result list while the
    // background trails far behind, which drives the spread heuristic over
    // its escalation threshold.
    let pocket_dir = resume_test_vector(DIM, 42);
    for id in 0..4u64 {
        let mut v = pocket_dir.clone();
        let noise = resume_test_vector(DIM, 1_000 + id);
        for (x, n) in v.iter_mut().zip(&noise) {
            *x += 0.05 * n;
        }
        let norm = v.iter().map(|x| x * x).sum::<f32>().sqrt();
        for x in &mut v {
            *x /= norm;
        }
        index.insert(id, &v);
    }
    for id in 4..N {
        index.insert(id, &resume_test_vector(DIM, id + 1));
    }

    let quality = SearchQuality::Adaptive {
        min_ef: 32,
        max_ef: 64,
    };
    let first = index.search_with_quality(&pocket_dir, K, quality).unwrap();
    assert!(!first.is_empty());
    for _ in 0..50 {
        let run = index.search_with_quality(&pocket_dir, K, quality).unwrap();
        let first_ids: Vec<u64> = first.iter().map(|r| r.id).collect();
        let run_ids: Vec<u64> = run.iter().map(|r| r.id).collect();
        assert_eq!(
            first_ids, run_ids,
            "adaptive resume drifted across pooled reuse"
        );
    }
}

/// An index built with `new_fast_insert`, a twin built with `new`, both
/// holding the same 100 vectors, and a query.
fn fast_insert_and_twin() -> (HnswIndex, HnswIndex, Vec<f32>) {
    let index = HnswIndex::new_fast_insert(128, DistanceMetric::Cosine).unwrap();
    let twin = HnswIndex::new(128, DistanceMetric::Cosine).unwrap();
    for i in 0u64..100 {
        let v: Vec<f32> = (0..128)
            .map(|j| ((i + j as u64) as f32 * 0.01).sin())
            .collect();
        index.insert(i, &v);
        twin.insert(i, &v);
    }
    let query = (0..128).map(|j| (j as f32 * 0.02).cos()).collect();
    (index, twin, query)
}

/// Exact-distance features off turn the CPU brute-force scan off: it returns
/// nothing for an index built with `new_fast_insert`, while the twin built
/// with `new` answers.
#[test]
fn cpu_brute_force_returns_nothing_with_exact_distance_features_off() {
    let (index, twin, query) = fast_insert_and_twin();
    assert_eq!(
        twin.brute_force_search_parallel(&query, 10).unwrap().len(),
        10
    );
    assert!(index
        .brute_force_search_parallel(&query, 10)
        .unwrap()
        .is_empty());
}

/// Exact-distance features off turn brute force off on the GPU path too: it
/// returns nothing rather than scan an index built with `new_fast_insert`. A
/// twin built with `new` is the positive control, and must answer whenever a
/// GPU is present; without one the test returns early, reported as passed,
/// after printing that the guard went unexercised (shown with `--nocapture`).
#[cfg(feature = "gpu")]
#[test]
fn gpu_brute_force_returns_nothing_with_exact_distance_features_off() {
    let (index, twin, query) = fast_insert_and_twin();
    if !crate::gpu::GpuAccelerator::is_available() {
        eprintln!("GPU unavailable: the exact-distance guard is not exercised here");
        return;
    }
    let scanned = twin.search_brute_force_gpu(&query, 10).unwrap();
    let scanned = scanned.expect("test: with a GPU, the twin built with new() scans");
    assert_eq!(scanned.len(), 10);
    assert!(index.search_brute_force_gpu(&query, 10).unwrap().is_none());
}

/// A delete never lands inside a renumber. The renumber below re-maps as
/// `reorder_for_locality` does, under the write guard; `remove` holds the
/// read guard across its two map writes, so every deleted id stays deleted
/// and every kept id keeps a slot that names it back.
#[test]
fn deletes_racing_a_renumber_stay_deleted() {
    use super::native_inner::Placed;
    use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
    use std::sync::Barrier;

    const IDS: usize = 4_000;
    let index = HnswIndex::new(4, DistanceMetric::Euclidean).unwrap();
    for id in 0..IDS {
        assert_eq!(index.mappings.assign(id as u64, Placed::for_test(id)), None);
    }
    // Swaps each even slot with the next: its own inverse, so an even number
    // of rounds leaves every slot where it started.
    let swap: Vec<usize> = (0..IDS).map(|slot| slot ^ 1).collect();
    let deleting = AtomicBool::new(true);
    let start = Barrier::new(2);
    let rounds = AtomicU32::new(0);
    let mut unmapped = Vec::new();
    let mut stalled = false;
    std::thread::scope(|scope| {
        let renumbers = scope.spawn(|| {
            start.wait();
            while deleting.load(Ordering::Acquire) || rounds.load(Ordering::Acquire) % 2 == 1 {
                let graph = index.inner.write();
                index.mappings.remap_indices(&swap);
                drop(graph);
                rounds.fetch_add(1, Ordering::AcqRel);
                std::thread::yield_now();
            }
        });
        start.wait();
        // Recorded, not asserted: a panic here would leave the renumbering
        // thread looping, and the scope waiting on it for ever. Every 200
        // deletes wait for a renumber to finish, so the renumbering thread is
        // running at every checkpoint, not by scheduling luck.
        for (n, id) in (0..IDS as u64).step_by(2).enumerate() {
            if n % 200 == 0 && !a_round_passes(&rounds, &renumbers) {
                stalled = true;
                break;
            }
            if !index.remove(id) {
                unmapped.push(id);
            }
        }
        deleting.store(false, Ordering::Release);
    });
    assert!(!stalled, "the renumbering thread stopped making rounds");
    assert!(
        unmapped.is_empty(),
        "ids found unmapped when removed: {unmapped:?}"
    );
    for id in 0..IDS as u64 {
        match index.mappings.get_idx(id) {
            Some(slot) => {
                assert_eq!(id % 2, 1, "deleted id {id} is mapped again, at slot {slot}");
                assert_eq!(
                    index.mappings.get_id(slot),
                    Some(id),
                    "slot {slot} names another id"
                );
            }
            None => assert_eq!(id % 2, 0, "kept id {id} lost its slot"),
        }
    }
    // Searches resolve slots through the reverse map: no slot may still name a
    // deleted id, and an even number of swaps leaves each kept id in its own.
    let (_, reverse, _) = index.mappings.as_parts();
    assert_eq!(reverse.len(), IDS / 2, "reverse entries");
    assert!(
        reverse
            .iter()
            .all(|(&slot, &id)| id % 2 == 1 && slot == id as usize),
        "a slot names a deleted id, or a kept id sits outside its own slot"
    );
}

/// Spins, yielding, until `rounds` moves past the value it holds now, and
/// returns `true`; returns `false` once `peer`, the thread making the rounds,
/// has ended without moving it, so the caller stops waiting on a peer that
/// stopped making rounds (see [`seen_before_it_ends`]).
fn a_round_passes<T>(
    rounds: &std::sync::atomic::AtomicU32,
    peer: &std::thread::ScopedJoinHandle<'_, T>,
) -> bool {
    use std::sync::atomic::Ordering;
    let seen = rounds.load(Ordering::Acquire);
    seen_before_it_ends(peer, || rounds.load(Ordering::Acquire) != seen)
}

/// Spins, yielding, until `seen` holds, and returns `true`; returns `false`
/// once `peer`, the thread expected to bring it about, has ended without it.
///
/// Nothing else bounds the wait. A bound on spins is a bound on time, and a
/// loaded machine slows the peer past it: such a bound read a save still
/// running as a stall, and failed `batch_inserts_racing_a_save_reload_consistent`
/// under a full test run. A peer blocked for ever hangs this, as it hangs the
/// scope that joins it.
fn seen_before_it_ends<T>(
    peer: &std::thread::ScopedJoinHandle<'_, T>,
    seen: impl Fn() -> bool,
) -> bool {
    loop {
        // Read before `seen`: whatever the peer did, it did before it ended.
        let ended = peer.is_finished();
        if seen() {
            return true;
        }
        if ended {
            return false;
        }
        std::thread::yield_now();
    }
}

/// A vector no other `(id, version)` pair shares: its first two coordinates
/// are the pair itself, so an exhaustive scan's top-1 on it is unambiguous.
fn racing_vector(id: u64, version: u8) -> Vec<f32> {
    let angle = id as f32 * 0.37;
    vec![id as f32, f32::from(version), angle.sin(), angle.cos()]
}

/// Writes made while `vacuum` rebuilds survive its swap (#2262). One thread
/// vacuums in a loop while this one inserts new ids, upserts old ones and
/// deletes others; afterwards an exhaustive scan sees every write, as if no
/// vacuum had run.
#[test]
fn writes_racing_a_vacuum_survive_it() {
    let index = HnswIndex::new(4, DistanceMetric::Euclidean).unwrap();
    for id in 0..RACED_BASE {
        index.insert(id, &racing_vector(id, 0));
    }
    assert_writes_survive_vacuums(&index);
}

/// As [`writes_racing_a_vacuum_survive_it`], on an SQ8 index with a trained
/// quantizer: the vacuum rebuilds it as SQ8, and copies the writes that race
/// it into that quantized graph, whose inserts encode each vector as they go.
/// Afterwards the index is still SQ8, its quantizer still trained, and every
/// write is there. The codes themselves are not read back: the backend that
/// holds them is private to `native_inner`.
#[cfg(feature = "persistence")]
#[test]
fn writes_racing_a_vacuum_of_an_sq8_index_survive_it() {
    let params = HnswParams {
        storage_mode: crate::StorageMode::SQ8,
        ..HnswParams::auto(4)
    };
    let index = HnswIndex::with_params(4, DistanceMetric::Euclidean, params).unwrap();
    let vectors: Vec<Vec<f32>> = (0..RACED_BASE).map(|id| racing_vector(id, 0)).collect();
    for (id, vector) in (0..RACED_BASE).zip(&vectors) {
        index.insert(id, vector);
    }
    let samples: Vec<&[f32]> = vectors.iter().map(Vec::as_slice).collect();
    let quantizer = crate::index::hnsw::native::ScalarQuantizer::train(&samples).unwrap();
    assert!(
        index
            .inner
            .read()
            .install_trained_sq8(std::sync::Arc::new(quantizer))
            .unwrap(),
        "the index has no SQ8 backend to install the quantizer into"
    );

    assert_writes_survive_vacuums(&index);
    assert_eq!(index.inner.read().storage_mode(), crate::StorageMode::SQ8);
    assert!(
        index.inner.read().is_sq8_quantizer_trained(),
        "the vacuums dropped the quantizer"
    );
}

/// Ids below [`RACED_WRITES`] are upserted during the race, the next
/// `RACED_WRITES` deleted, the next `RACED_WRITES` left alone; `RACED_BASE`
/// onwards are inserted during it.
const RACED_WRITES: u64 = 300;
/// The ids an index holds before [`assert_writes_survive_vacuums`] races it.
const RACED_BASE: u64 = 3 * RACED_WRITES;

/// Races writes against vacuums on `index`, which holds ids `0..RACED_BASE`
/// on their version-0 vectors, and asserts an exhaustive scan sees every write.
fn assert_writes_survive_vacuums(index: &HnswIndex) {
    let vacuums = write_beside_vacuums(index, RACED_WRITES, RACED_BASE);

    let [lost, stale, resurrected] = missed_writes(index, RACED_WRITES, RACED_BASE);
    assert!(
        lost.is_empty() && stale.is_empty() && resurrected.is_empty(),
        "after {vacuums} vacuums: {} inserts lost (first {:?}), {} upserts still on \
         their old vector (first {:?}), {} deletes back in the scan (first {:?})",
        lost.len(),
        lost.first(),
        stale.len(),
        stale.first(),
        resurrected.len(),
        resurrected.first(),
    );
    assert_eq!(index.len(), RACED_BASE as usize, "live ids");
}

/// Makes `writes` rounds of writes on `index` while another thread vacuums it
/// in a loop, and returns how many vacuums finished. Round `n` inserts id
/// `base + n`, upserts id `n` and deletes id `writes + n`. Every 50 rounds
/// wait for a vacuum to finish, so the vacuuming thread is running at every
/// checkpoint, not by scheduling luck.
fn write_beside_vacuums(index: &HnswIndex, writes: u64, base: u64) -> u32 {
    use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
    use std::sync::Barrier;

    let writing = AtomicBool::new(true);
    let start = Barrier::new(2);
    let rounds = AtomicU32::new(0);
    let mut stalled = false;
    let vacuumed = std::thread::scope(|scope| {
        let vacuums = scope.spawn(|| {
            start.wait();
            while writing.load(Ordering::Acquire) {
                index.vacuum()?;
                rounds.fetch_add(1, Ordering::AcqRel);
            }
            Ok::<(), VacuumError>(())
        });
        start.wait();
        // Recorded, not asserted: a panic here would leave the vacuuming
        // thread looping, and the scope waiting on it for ever.
        for n in 0..writes {
            if n % 50 == 0 && !a_round_passes(&rounds, &vacuums) {
                stalled = true;
                break;
            }
            index.insert(base + n, &racing_vector(base + n, 0));
            index.insert(n, &racing_vector(n, 1));
            index.remove(writes + n);
        }
        writing.store(false, Ordering::Release);
        vacuums.join()
    });
    assert_eq!(
        vacuumed.expect("test: the vacuuming thread panicked"),
        Ok(())
    );
    assert!(!stalled, "the vacuuming thread stopped making rounds");
    rounds.load(Ordering::Acquire)
}

/// The writes `write_beside_vacuums(index, writes, base)` made that an
/// exhaustive scan misses, in three lists: inserted ids unmapped or not their
/// own top-1; upserted ids not top-1 on their new vector, or still at distance
/// 0 on their old one; deleted ids still mapped or still scanned.
fn missed_writes(index: &HnswIndex, writes: u64, base: u64) -> [Vec<u64>; 3] {
    let top = |id, version| {
        let hits = index
            .search_brute_force(&racing_vector(id, version), 1)
            .unwrap();
        hits.first().map(|hit| (hit.id, hit.score))
    };
    let on_top = |id, version| top(id, version).map(|(hit, _)| hit) == Some(id);
    let scanned: std::collections::HashSet<u64> = index
        .search_brute_force(&racing_vector(0, 0), (base + writes) as usize)
        .unwrap()
        .iter()
        .map(|hit| hit.id)
        .collect();
    [
        (base..base + writes)
            .filter(|&id| !(index.mappings.contains(id) && on_top(id, 0)))
            .collect(),
        (0..writes)
            .filter(|&id| !on_top(id, 1) || top(id, 0) == Some((id, 0.0)))
            .collect(),
        (writes..2 * writes)
            .filter(|&id| index.mappings.contains(id) || scanned.contains(&id))
            .collect(),
    ]
}

/// Reloads the index saved in `dir` and returns the ids of `ids` whose reloaded
/// vector `own` refuses, with the reload's tombstone count and its dead slots,
/// which must be equal. `own(id, stored)` gets the vector the reload resolves
/// `id` to, `None` when it maps `id` nowhere.
fn check_reload(
    dir: &std::path::Path,
    ids: impl IntoIterator<Item = u64>,
    own: impl Fn(u64, Option<&[f32]>) -> bool,
) -> std::io::Result<(Vec<u64>, (usize, usize))> {
    let reloaded = HnswIndex::load(dir, 4, DistanceMetric::Euclidean)?;
    let counts = (
        reloaded.tombstone_count(),
        reloaded.graph_vector_count() - reloaded.len(),
    );
    let graph = reloaded.inner.read();
    let misresolved = |vectors: &crate::perf_optimizations::ContiguousVectors| {
        ids.into_iter()
            .filter(|&id| {
                let stored = reloaded
                    .mappings
                    .get_idx(id)
                    .and_then(|slot| vectors.get(slot));
                !own(id, stored)
            })
            .collect()
    };
    Ok((graph.with_contiguous_vectors(misresolved), counts))
}

/// A write landing between a save's mappings copy and its graph dump never
/// makes the save name a slot the saved graph lacks (#2262).
///
/// `writes_racing_a_save_reload_consistent` above writes throughout a save
/// and relies on one landing in that window by timing. On a 1200-vector
/// index of four dimensions the dump takes microseconds, so none does: moving
/// the copy after the dump leaves that test green eight runs out of eight.
/// This one puts the write there instead of hoping for it, through the seam
/// `persistence::dump_counted_through_the_window` opens, and is the only test
/// that fails when the two are swapped.
///
/// With the copy first, the injected id is absent from the saved mappings and
/// its slot is in the dumped graph, unnamed: a tombstone, which is consistent.
/// With the dump first, the id is in the mappings and its slot is not in the
/// graph, which is the torn directory the ordering exists to prevent.
#[test]
fn a_write_in_the_save_window_never_names_a_slot_the_graph_lacks() {
    const BASE: u64 = 400;
    const INJECTED: u64 = BASE + 1;
    let dir = tempfile::tempdir().unwrap();
    let index = std::sync::Arc::new(HnswIndex::new(4, DistanceMetric::Euclidean).unwrap());
    for id in 0..BASE {
        index.insert(id, &racing_vector(id, 0));
    }

    let injected = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(INJECTED));
    let window = {
        let (index, injected) = (
            std::sync::Arc::clone(&index),
            std::sync::Arc::clone(&injected),
        );
        crate::index::hnsw::persistence::save_window::install(dir.path(), move || {
            // On a thread of its own: the saving thread is inside its read
            // guard, and a second `read()` on that same thread is what #2343
            // was about. A separate thread takes a reader beside it, which is
            // what a real concurrent write does.
            let id = injected.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
            let index = std::sync::Arc::clone(&index);
            std::thread::spawn(move || index.insert(id, &racing_vector(id, 0)))
                .join()
                .expect("test: the injected write panicked");
        })
    };

    index.save(dir.path()).expect("test: the save");
    drop(window);

    // The ids present before the save must resolve to their own vector; the
    // injected ones may be absent (the save did not name them) but never
    // resolve to something that is not theirs.
    let last_injected = injected.load(std::sync::atomic::Ordering::Acquire);
    let own = |id, stored: Option<&[f32]>| {
        let is_own = stored == Some(racing_vector(id, 0).as_slice());
        if id < BASE {
            is_own
        } else {
            stored.is_none() || is_own
        }
    };
    let (misresolved, (tombstones, dead)) =
        check_reload(dir.path(), 0..last_injected, own).expect("test: the reload");
    assert!(
        misresolved.is_empty(),
        "{} ids resolve to a vector that is not theirs after a write in the save \
         window (first {:?})",
        misresolved.len(),
        misresolved.first()
    );
    assert_eq!(
        tombstones, dead,
        "the reloaded index counts {tombstones} tombstones over {dead} slots no id names"
    );
}

/// A save never straddles a vacuum's swap (#2262): the swap renumbers under
/// the write guard, and a save copies the mappings and dumps the graph under
/// one read guard, so it persists the old numbering or the new, never the one
/// beside the other. Each round races one vacuum, which renumbers every id of
/// a fresh index, against saves in a loop, each into its own directory, so a
/// swap waiting on a save lands right after it; every save is reloaded once
/// the rounds are over.
#[test]
fn a_save_racing_a_vacuum_reloads_consistent() {
    use std::sync::atomic::{AtomicBool, Ordering};

    const IDS: u64 = 1_000;
    const ROUNDS: usize = 10;
    let dir = tempfile::tempdir().unwrap();
    let mut saves = Vec::new();
    for round in 0..ROUNDS {
        let index = HnswIndex::new(4, DistanceMetric::Euclidean).unwrap();
        for id in 0..IDS {
            index.insert(id, &racing_vector(id, 0));
        }
        let vacuumed = AtomicBool::new(false);
        let vacuum = std::thread::scope(|scope| {
            let vacuum = scope.spawn(|| {
                let result = index.vacuum();
                vacuumed.store(true, Ordering::Release);
                result
            });
            // Recorded, not asserted, like every check made while the vacuum
            // runs.
            while !vacuumed.load(Ordering::Acquire) {
                let path = dir.path().join(saves.len().to_string());
                let saved = index.save(&path);
                saves.push((round, path, saved));
            }
            vacuum.join()
        });
        let vacuum = vacuum.expect("test: the vacuuming thread panicked");
        assert_eq!(vacuum, Ok(IDS as usize), "round {round}");
        // The positive control: a fresh index maps id `n` to slot `n`, so a
        // vacuum that left every id there would give the race nothing to show.
        assert!(
            (0..IDS).any(|id| index.mappings.get_idx(id) != Some(id as usize)),
            "round {round}: the vacuum renumbered nothing"
        );
    }
    let made = saves.len();
    assert!(made >= ROUNDS, "{made} saves overlapped a vacuum");
    let inconsistent: Vec<_> = saves
        .into_iter()
        .filter_map(|(round, path, saved)| {
            let own = |id, stored: Option<&[f32]>| stored == Some(racing_vector(id, 0).as_slice());
            match saved.and_then(|()| check_reload(&path, 0..IDS, own)) {
                Ok((ids, (count, dead))) if ids.is_empty() && count == dead => None,
                outcome => Some((round, outcome.map(|(ids, counts)| (ids.len(), counts)))),
            }
        })
        .collect();
    assert!(
        inconsistent.is_empty(),
        "{} of the {made} saves made during a vacuum reloaded inconsistent \
         (round, misresolved ids and (tombstone count, dead slots), or error): \
         {inconsistent:?}",
        inconsistent.len()
    );
}

/// After deletes race a vacuum, the tombstone count is every slot of the new
/// graph that no id names, and the next vacuum leaves none (#2262). `vacuum`
/// is triggered from that count, so a dead slot it misses is one no vacuum is
/// asked to reclaim.
///
/// Each round deletes every id but the first `KEEP` the mappings iterate while
/// one vacuum runs. Its snapshot iterates them in that order and the rebuild
/// places them in it, so the new graph ends on slots of ids deleted during the
/// rebuild; deleting from the last id back makes that tail die first,
/// wherever the delete loop and the snapshot cross.
#[test]
fn deletes_racing_a_vacuum_keep_the_tombstone_count_exact() {
    use std::sync::Barrier;

    const IDS: u64 = 2_000;
    const KEEP: usize = 10;
    const ROUNDS: usize = 5;
    // Per round: (tombstone count, dead slots) after the race, then after
    // the next vacuum.
    let mut counts = Vec::new();
    let mut dead_tails = 0;
    for round in 0..ROUNDS {
        let index = HnswIndex::new(4, DistanceMetric::Euclidean).unwrap();
        for id in 0..IDS {
            index.insert(id, &racing_vector(id, 0));
        }
        let order: Vec<u64> = index.mappings.iter().map(|(id, _)| id).collect();
        let start = Barrier::new(2);
        let (vacuumed, unmapped) = std::thread::scope(|scope| {
            let vacuum = scope.spawn(|| {
                start.wait();
                index.vacuum()
            });
            start.wait();
            let unmapped = order[KEEP..]
                .iter()
                .rev()
                .filter(|&&id| !index.remove(id))
                .count();
            (vacuum.join(), unmapped)
        });
        let vacuumed = vacuumed.expect("test: the vacuuming thread panicked");
        assert!(vacuumed.is_ok(), "round {round}: {vacuumed:?}");
        assert_eq!(
            unmapped, 0,
            "round {round}: ids found unmapped when removed"
        );
        assert_eq!(index.len(), KEEP, "round {round}: live ids");

        let dead = || index.graph_vector_count() - index.len();
        let named_extent = index.mappings.iter().map(|(_, slot)| slot + 1).max();
        dead_tails += usize::from(named_extent < Some(index.graph_vector_count()));
        let raced = (index.tombstone_count(), dead());
        index.vacuum().unwrap();
        counts.push((round, raced, (index.tombstone_count(), dead())));
    }
    // The positive control: a round whose new graph ends on dead slots, the
    // ones a count bounded by the highest named slot leaves out.
    assert!(
        dead_tails > 0,
        "no round left dead slots at the end of the new graph"
    );
    let wrong: Vec<_> = counts
        .iter()
        .filter(|&&(_, (count, dead), after)| count != dead || after != (0, 0))
        .collect();
    assert!(
        wrong.is_empty(),
        "(round, (tombstone count, dead slots) after the race, then after the next \
         vacuum): {wrong:?}"
    );
}

/// Writes racing a save never tear it (#2262): every reload resolves each id
/// to one of its own vectors, and counts every slot no id names as a
/// tombstone. A save that dumps the graph, then reads the mappings one map
/// after the other, saves an id on a slot the saved graph never held when an
/// insert or upsert lands after the dump, and two maps that disagree when a
/// write lands between the two reads. One thread saves in a loop, each save
/// into its own directory, while this one makes the writes
/// `write_beside_vacuums` makes: upserts, deletes and inserts. Every save is
/// reloaded once the writes are over.
#[test]
fn writes_racing_a_save_reload_consistent() {
    // As in `writes_racing_a_vacuum_survive_it`: ids below WRITES are
    // upserted, the next WRITES deleted, the next WRITES left alone; BASE
    // onwards are inserted during the race.
    const WRITES: u64 = 300;
    const BASE: u64 = 3 * WRITES;
    let dir = tempfile::tempdir().unwrap();
    let index = HnswIndex::new(4, DistanceMetric::Euclidean).unwrap();
    for id in 0..BASE {
        index.insert(id, &racing_vector(id, 0));
    }
    let mut stalled = false;
    let saves = save_beside(&index, dir.path(), |a_save_passes| {
        for n in 0..WRITES {
            if n % 50 == 0 && !a_save_passes() {
                stalled = true;
                break;
            }
            index.insert(n, &racing_vector(n, 1));
            index.remove(WRITES + n);
            index.insert(BASE + n, &racing_vector(BASE + n, 0));
        }
    });
    assert!(!stalled, "the saving thread stopped making rounds");

    // Upserted ids stay live throughout, on either vector; deleted and
    // inserted ones may be absent, never on another vector; the rest never
    // move.
    let own = |id, stored: Option<&[f32]>| {
        let is = |version| stored == Some(racing_vector(id, version).as_slice());
        match id / WRITES {
            0 => is(0) || is(1),
            1 | 3 => stored.is_none() || is(0),
            _ => is(0),
        }
    };
    let made = saves.len();
    let torn = torn_saves(saves, &(0..BASE + WRITES), own);
    assert!(
        torn.is_empty(),
        "{} of the {made} saves made during writes reloaded torn: {torn:?}",
        torn.len()
    );
}

/// Batch inserts racing a save never tear its graph files (#2262): every save
/// reloads, each id on its own vector. `file_dump` wrote the vectors file,
/// released the arena's lock, then wrote the graph file. A batch reserves its
/// graph capacity, then waits on the arena's write lock; one waiting when the
/// vectors file was done pushed nodes that file does not hold and linked them
/// into saved nodes while the graph file was written, without the layers lock
/// the graph file's writer holds, and the load refused the save. Vectors of
/// `WIDE` dimensions make the vectors file slow enough to write that a batch
/// is often waiting when it ends.
#[test]
fn batch_inserts_racing_a_save_reload_consistent() {
    const BASE: u64 = 4_000;
    // At least 100: smaller batches go through single inserts.
    const BATCH: u64 = 200;
    const BATCHES: u64 = 40;
    let dir = tempfile::tempdir().unwrap();
    let index = wide_index_with_base(BASE);
    let insert_batch = |first: u64, len: u64| insert_wide_batch(&index, first, len);
    let mut stalled = false;
    let mut placed = 0;
    let saves = save_beside(&index, dir.path(), |a_save_passes| {
        for n in 0..BATCHES {
            if n % 5 == 0 && !a_save_passes() {
                stalled = true;
                break;
            }
            placed += insert_batch(BASE + n * BATCH, BATCH);
        }
    });
    assert!(!stalled, "the saving thread stopped making rounds");
    assert_eq!(u64::try_from(placed), Ok(BATCHES * BATCH), "vectors placed");

    let own = |id, stored: Option<&[f32]>| {
        (id >= BASE && stored.is_none()) || stored == Some(wide_vector(id).as_slice())
    };
    let made = saves.len();
    let torn = torn_saves(saves, &(0..BASE + BATCHES * BATCH), own);
    assert!(
        torn.is_empty(),
        "{} of the {made} saves made during batch inserts reloaded torn: {torn:?}",
        torn.len()
    );
}

/// Dimensions of `wide_vector`.
const WIDE: usize = 128;

/// A `WIDE`-dimensional vector no other id shares: its first coordinate is
/// the id itself.
fn wide_vector(id: u64) -> Vec<f32> {
    let angle = id as f32 * 0.37;
    let mut vector: Vec<f32> = (0..WIDE)
        .map(|i| (angle + f32::from(u16::try_from(i).unwrap())).sin())
        .collect();
    vector[0] = id as f32;
    vector
}

/// Inserts ids `first..first + len` into `index` as one batch, and returns
/// how many it placed.
///
/// `len` is at least 100 in every caller: under that the batch path inserts
/// one vector at a time, and the races these tests drive need the path that
/// pushes every vector, then links them.
fn insert_wide_batch(index: &HnswIndex, first: u64, len: u64) -> usize {
    let batch: Vec<Vec<f32>> = (first..first + len).map(wide_vector).collect();
    index.insert_batch_parallel((first..).zip(batch.iter().map(Vec::as_slice)))
}

/// A `WIDE`-dimensional index holding ids `0..base`, the base the race tests
/// write on top of. Asserts the base went in whole: a race over a short base
/// would prove nothing about the one they mean to run.
fn wide_index_with_base(base: u64) -> HnswIndex {
    let index = HnswIndex::new(WIDE, DistanceMetric::Euclidean).unwrap();
    assert_eq!(
        u64::try_from(insert_wide_batch(&index, 0, base)),
        Ok(base),
        "base placed"
    );
    index
}

/// Saves `index` in a loop, each save into its own directory under `dir`,
/// while `write` runs on this thread, and returns every save once `write`
/// returns. `write` gets a checkpoint that waits for a save to finish and
/// returns `false` if the saving thread ended first. It must record what it
/// checks, not assert it: a panic would leave the saving thread looping, and
/// the scope waiting on it for ever.
fn save_beside(
    index: &HnswIndex,
    dir: &std::path::Path,
    write: impl FnOnce(&dyn Fn() -> bool),
) -> Vec<(std::path::PathBuf, std::io::Result<()>)> {
    use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
    use std::sync::Barrier;

    let writing = AtomicBool::new(true);
    let start = Barrier::new(2);
    let rounds = AtomicU32::new(0);
    let saves = std::thread::scope(|scope| {
        let saver = scope.spawn(|| {
            let mut saves = Vec::new();
            start.wait();
            while writing.load(Ordering::Acquire) {
                let path = dir.join(saves.len().to_string());
                let saved = index.save(&path);
                saves.push((path, saved));
                rounds.fetch_add(1, Ordering::AcqRel);
            }
            saves
        });
        start.wait();
        write(&|| a_round_passes(&rounds, &saver));
        writing.store(false, Ordering::Release);
        saver.join()
    });
    saves.expect("test: the saving thread panicked")
}

/// The saves among `saves` that failed, or whose reload `check_reload` faults
/// over `ids` with `own`, each described by its path and what went wrong.
fn torn_saves(
    saves: Vec<(std::path::PathBuf, std::io::Result<()>)>,
    ids: &std::ops::Range<u64>,
    own: impl Fn(u64, Option<&[f32]>) -> bool + Copy,
) -> Vec<String> {
    saves
        .into_iter()
        .filter_map(|(path, saved)| {
            match saved.and_then(|()| check_reload(&path, ids.clone(), own)) {
                Ok((ids, (count, dead))) if ids.is_empty() && count == dead => None,
                Ok((ids, counts)) => Some(format!(
                    "{}: {} ids misresolved (first {:?}), (tombstone count, dead slots) {counts:?}",
                    path.display(),
                    ids.len(),
                    ids.first()
                )),
                Err(e) => Some(format!("{}: {e}", path.display())),
            }
        })
        .collect()
}

/// A vacuum rebuilds the graph with the index's own parameters, and a save
/// after it persists them (#2262): the rebuild used `HnswParams::auto` for the
/// dimension, whatever M, `ef_construction` and alpha the index was built
/// with.
#[test]
fn vacuum_keeps_the_index_parameters() {
    let built = |index: &HnswIndex| {
        let graph = index.inner.read();
        (
            graph.max_connections(),
            graph.ef_construction(),
            graph.alpha().to_bits(),
        )
    };
    let params = HnswParams {
        max_connections: 12,
        ef_construction: 90,
        alpha: 1.0,
        ..HnswParams::auto(4)
    };
    let expected = (12, 90, 1.0_f32.to_bits());
    let index = HnswIndex::with_params(4, DistanceMetric::Euclidean, params).unwrap();
    for id in 0..300 {
        index.insert(id, &racing_vector(id, 0));
    }
    for id in 0..100 {
        index.remove(id);
    }
    assert_eq!(built(&index), expected, "as built");
    assert_eq!(index.vacuum(), Ok(200));
    assert_eq!(built(&index), expected, "after a vacuum");
    let dir = tempfile::tempdir().unwrap();
    index.save(dir.path()).unwrap();
    let reloaded = HnswIndex::load(dir.path(), 4, DistanceMetric::Euclidean).unwrap();
    assert_eq!(built(&reloaded), expected, "saved after a vacuum");
}

/// A vacuum of an index whose ids are all deleted leaves it empty, with no
/// tombstone, and the next vacuum has nothing to do (#2262). It returned
/// before rebuilding, so every dead slot stayed, and the tombstone count
/// kept asking for a vacuum that reclaimed nothing.
#[test]
fn vacuum_of_an_index_with_no_live_id_empties_it() {
    let index = HnswIndex::new(4, DistanceMetric::Euclidean).unwrap();
    for id in 0..200 {
        index.insert(id, &racing_vector(id, 0));
    }
    let removed = (0..200).filter(|&id| index.remove(id)).count();
    assert_eq!((removed, index.tombstone_count()), (200, 200), "before");
    // (tombstones, graph slots, live ids)
    let state = |index: &HnswIndex| {
        (
            index.tombstone_count(),
            index.graph_vector_count(),
            index.len(),
        )
    };
    assert_eq!(index.vacuum(), Ok(0));
    assert_eq!(state(&index), (0, 0, 0), "after a vacuum");
    let query = racing_vector(7, 0);
    assert!(
        VectorIndex::search(&index, &query, 5).is_empty(),
        "a search of the emptied index"
    );
    assert_eq!(index.vacuum(), Ok(0));
    assert_eq!(state(&index), (0, 0, 0), "after a second vacuum");
    index.insert(500, &racing_vector(500, 0));
    let hits = VectorIndex::search(&index, &racing_vector(500, 0), 1);
    assert_eq!(
        hits.first().map(|hit| hit.id),
        Some(500),
        "a search after an insert"
    );
}

/// Two saves of one index into one directory never mix their files (#2262).
/// A save rewrites its graph file in place and stamps every file with the
/// generation after the one it reads from the directory, so two saves at once
/// wrote one graph file at the same offsets, each from its own reading of
/// neighbor lists a batch was still linking, under one generation. Each round
/// a batch is pushed, and two threads save into one directory while it links;
/// the directory is reloaded after every round.
#[test]
fn saves_racing_into_one_directory_reload_consistent() {
    const BASE: u64 = 2_000;
    // At least 100: the batch path, which pushes every vector, then links.
    const BATCH: u64 = 200;
    const ROUNDS: u64 = 20;
    let dir = tempfile::tempdir().unwrap();
    let index = wide_index_with_base(BASE);
    let insert_batch = |first: u64, len: u64| insert_wide_batch(&index, first, len);
    let own = |id, stored: Option<&[f32]>| {
        (id >= BASE && stored.is_none()) || stored == Some(wide_vector(id).as_slice())
    };
    let (mut torn, mut unseen, mut short) = (Vec::new(), 0, 0);
    for round in 0..ROUNDS {
        let first = BASE + round * BATCH;
        let (seen, saved, placed) =
            two_saves_while_a_batch_links(&index, dir.path(), || insert_batch(first, BATCH));
        unseen += usize::from(!seen);
        short += usize::from(u64::try_from(placed) != Ok(BATCH));
        match saved.and_then(|()| check_reload(dir.path(), 0..first + BATCH, own)) {
            Ok((ids, (count, dead))) if ids.is_empty() && count == dead => {}
            outcome => torn.push(format!(
                "round {round}: {:?}",
                outcome.map(|(ids, counts)| (ids.len(), counts))
            )),
        }
    }
    assert_eq!(
        (unseen, short),
        (0, 0),
        "(rounds whose batch was never seen pushed, batches placed short)"
    );
    assert!(
        torn.is_empty(),
        "{} of the {ROUNDS} rounds left a directory that reloads torn \
         (misresolved ids, (tombstone count, dead slots)), or error: {torn:?}",
        torn.len()
    );
}

/// Runs `batch` on a thread of its own and, once the vectors it pushes are in
/// the arena, saves `index` into `dir` from two threads at once while it links
/// them. Returns whether the push was seen, the two saves' outcome, and what
/// `batch` returned.
fn two_saves_while_a_batch_links(
    index: &HnswIndex,
    dir: &std::path::Path,
    batch: impl FnOnce() -> usize + Send,
) -> (bool, std::io::Result<()>, usize) {
    use std::sync::Barrier;

    let slots = index.graph_vector_count();
    let start = Barrier::new(2);
    let save = || {
        start.wait();
        index.save(dir)
    };
    let (seen, joined, placed) = std::thread::scope(|scope| {
        let batch = scope.spawn(batch);
        let seen = seen_before_it_ends(&batch, || index.graph_vector_count() > slots);
        let (a, b) = (scope.spawn(save), scope.spawn(save));
        (seen, [a.join(), b.join()], batch.join())
    });
    let saved = joined
        .into_iter()
        .try_for_each(|save| save.expect("test: a saving thread panicked"));
    (
        seen,
        saved,
        placed.expect("test: the batch thread panicked"),
    )
}

/// Vectors in the fixtures that reorder. `reorder_for_locality` declines
/// below 1000 vectors (`REORDER_THRESHOLD`, private to
/// `native::graph::reorder`) and returns `None`, which no caller can tell
/// apart from the pass doing nothing, so both tests below clear that gate
/// with margin.
const RENUMBERED: u64 = 1_200;

/// `reorder_for_locality` renumbers the slots, and the mappings move with
/// them.
///
/// Nothing reached the pass before this test: `cargo mutants` replaced the
/// whole body of `HnswIndex::reorder_for_locality` with `Ok(())` and the
/// `--lib` suite stayed green, so the maintenance lock this branch puts in
/// front of it was guarding a function no test made do anything.
///
/// Should `REORDER_THRESHOLD` ever rise past [`RENUMBERED`], the pass starts
/// declining here, `moved` falls to zero, and this test says so rather than
/// passing on a silent skip.
///
/// The lookups at the end are exact (`SearchQuality::Perfect`): the question
/// is whether an id still resolves to its own vector, which a graph search
/// answers only up to its recall.
#[test]
fn reordering_renumbers_the_slots_and_the_mappings_follow() {
    use std::collections::{BTreeMap, BTreeSet};

    let index = wide_index_with_base(RENUMBERED);
    let before: BTreeMap<u64, usize> = index.mappings.iter().collect();

    index.reorder_for_locality().expect("test: reorder");

    let after: BTreeMap<u64, usize> = index.mappings.iter().collect();
    let moved = after
        .iter()
        .filter(|&(id, idx)| before.get(id) != Some(idx))
        .count();
    println!("{moved} of {} mapped ids changed slot", before.len());
    assert!(
        moved > 0,
        "reorder_for_locality left every one of the {} mapped ids on the slot \
         it already held: it renumbered nothing",
        before.len()
    );
    assert_eq!(
        before.keys().copied().collect::<Vec<_>>(),
        after.keys().copied().collect::<Vec<_>>(),
        "the renumbering lost or invented ids"
    );
    assert_eq!(
        before.values().copied().collect::<BTreeSet<usize>>(),
        after.values().copied().collect::<BTreeSet<usize>>(),
        "{moved} of the {} ids moved, but not onto the slots they started \
         from: the renumbering is not a permutation of them",
        before.len()
    );
    // `Perfect` scores every stored vector, so the nearest one to `id`'s own
    // vector is `id` unless the mappings stopped following the slots. A graph
    // search would also miss now and then on an index built by parallel
    // inserts, and that is recall, not what this test is about.
    for id in [0, RENUMBERED / 2, RENUMBERED - 1] {
        let hits = index
            .search_with_quality(&wide_vector(id), 1, SearchQuality::Perfect)
            .expect("test: exact search");
        assert_eq!(
            hits.first().map(|hit| hit.id),
            Some(id),
            "after the renumbering {id}'s own vector no longer finds {id}: \
             the mappings did not follow the slots"
        );
    }
}

/// `reorder_for_locality` waits on the maintenance lock.
///
/// That is the arm of "serializes what renumbers slots" (#2262) no other
/// test drives — the race tests above all enter through `vacuum`. A renumber
/// let through between a vacuum's snapshot and its swap moves the very slots
/// that snapshot recorded.
///
/// The bound on "waits" is measured, not chosen: one unheld reorder of this
/// same index is timed first, and the held one is given `BLOCKED_FACTOR`
/// times that to escape. Without the lock it finishes in about one such
/// unit, so the margin is the factor itself, and it follows the machine
/// instead of a constant written here.
#[test]
fn reorder_waits_for_the_maintenance_lock() {
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::time::Instant;

    /// Unheld reorders of the same index the held one is given to escape in.
    const BLOCKED_FACTOR: u32 = 50;

    let index = wide_index_with_base(RENUMBERED);
    let started = Instant::now();
    index
        .reorder_for_locality()
        .expect("test: the unheld reorder");
    let baseline = started.elapsed();
    let window = BLOCKED_FACTOR * baseline;
    println!("unheld reorder {baseline:?}, window {window:?} ({BLOCKED_FACTOR}x)");
    assert!(
        !window.is_zero(),
        "the unheld reorder was not measurable, so the window it sizes bounds \
         nothing"
    );

    let held = index.lock_maintenance();
    let renumbered = AtomicBool::new(false);
    std::thread::scope(|scope| {
        let blocked = scope.spawn(|| {
            index
                .reorder_for_locality()
                .expect("test: the held reorder");
            renumbered.store(true, Ordering::Release);
        });
        std::thread::sleep(window);
        let escaped = renumbered.load(Ordering::Acquire);
        drop(held);
        blocked.join().expect("test: the reorder thread panicked");
        assert!(
            !escaped,
            "reorder_for_locality renumbered within {window:?} while the \
             maintenance lock was held: it does not take the lock its doc \
             claims, and a vacuum's swap can be renumbered under it"
        );
    });
}