fix: allow the previous graph to connect with the new one (#113)

* fix: allow the previous graph to connect with the new one

* make biffer

* test: enable e2e fuzzing

* Change the fuzzer to run for a given time

* Add a fuzzer job in the CI

---------

Co-authored-by: Kerollmops <clement@meilisearch.com>

Author: nnethercott

.github/workflows/rust.yml

@@ -56,3 +56,13 @@ jobs:
         if: always()
         with:
           command: doc
+
+  fuzzer:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v1
+      - uses: dtolnay/rust-toolchain@1.85
+      - name: Run fuzzer
+        env:
+          HANNOY_FUZZ_DURATION_SEC: 1800
+        run: cargo test -- random_read_writes

benches/benchmark.rs

@@ -29,9 +29,9 @@ mod hnsw {
     }
 
     fn create_db_and_fill_with_vecs<const DIM: usize>(
-        env: &Env,
+        env: &'_ Env,
         size: usize,
-    ) -> hannoy::Result<(Writer<Cosine>, RwTxn, Database<Cosine>)> {
+    ) -> hannoy::Result<(Writer<Cosine>, RwTxn<'_>, Database<Cosine>)> {
         let mut wtxn = env.write_txn().unwrap();
 
         let db: Database<Cosine> = env.create_database(&mut wtxn, None)?;

src/hnsw.rs

@@ -185,14 +185,13 @@ impl<'a, D: Distance, const M: usize, const M0: usize> HnswBuilder<'a, D, M, M0>
             Ok(()) as Result<(), Error>
         })?;
 
-        self.maybe_patch_old_links(&lmdb, to_delete, options)?;
+        self.fill_gaps_from_deleted(&lmdb, to_delete, options)?;
 
         // Single-threaded write to lmdb
         options.progress.update(HannoyBuild::WritingTheItems);
         let mut cancellation_index = 0;
 
-        for lvl in 0..=self.max_level {
-            let Some(map) = self.layers.get(lvl) else { break };
+        for (lvl, map) in self.layers.iter().enumerate() {
             let map_guard = map.pin();
 
             for (item_id, node_state) in &map_guard {
@@ -332,7 +331,7 @@ impl<'a, D: Distance, const M: usize, const M0: usize> HnswBuilder<'a, D, M, M0>
     /// the end of indexing we need to merge the old and new links and prune ones pointing to
     /// deleted items.
     /// Algorithm 4 from FreshDiskANN paper.
-    fn maybe_patch_old_links<P>(
+    fn fill_gaps_from_deleted<P>(
         &mut self,
         lmdb: &FrozenReader<D>,
         to_delete: &RoaringBitmap,
@@ -387,6 +386,7 @@ impl<'a, D: Distance, const M: usize, const M0: usize> HnswBuilder<'a, D, M, M0>
             }
             bitmap |= links;
             bitmap -= to_delete;
+            debug_assert!(bitmap.is_disjoint(to_delete));
 
             //  Case 1: Union of [on_disk, current_build, deleted_extension] is small enough
             let thresh = if lvl == 0 { M0 } else { M };
@@ -404,10 +404,8 @@ impl<'a, D: Distance, const M: usize, const M0: usize> HnswBuilder<'a, D, M, M0>
             let curr = &lmdb.get_item(id)?;
 
             for other in bitmap {
-                if let Ok(v) = lmdb.get_item(other) {
-                    let dist = D::distance(curr, &v);
-                    new_links.push((OrderedFloat(dist), other));
-                }
+                let dist = D::distance(curr, &lmdb.get_item(other)?);
+                new_links.push((OrderedFloat(dist), other));
             }
             let pruned = self.robust_prune(new_links, lvl, self.alpha, lmdb)?;
             let _ = map_guard.insert(id, NodeState { links: ArrayVec::from_iter(pruned) });
@@ -451,7 +449,7 @@ impl<'a, D: Distance, const M: usize, const M0: usize> HnswBuilder<'a, D, M, M0>
             Some(node_state) => res.extend(node_state.links.iter().map(|(_, i)| *i)),
             None => {
                 // lazily add this entry so he can get updated later
-                // pinned.insert(item_id, NodeState { links: array_vec![] });
+                pinned.insert(item_id, NodeState { links: array_vec![] });
             }
         }
 

src/tests/fuzz.rs

@@ -1,32 +1,30 @@
+use std::{
+    env::VarError,
+    time::{Duration, Instant},
+};
+
 use crate::{
     distance::Cosine,
+    key::{KeyCodec, Prefix, PrefixCodec},
+    node::{Links, Node, NodeCodec},
+    node_id::NodeMode,
     tests::{create_database_indices_with_items, DatabaseHandle},
     Database, Reader, Writer,
 };
 use arbitrary::{Arbitrary, Unstructured};
 use heed::RoTxn;
-use rand::{self, rngs::StdRng, Rng, SeedableRng};
+use rand::{
+    self,
+    distributions::Uniform,
+    rngs::{StdRng, ThreadRng},
+    Rng, SeedableRng,
+};
 use roaring::RoaringBitmap;
 use tracing::info;
 
-#[derive(Debug)]
-struct Item<const M: usize> {
-    id: u32,
-    data: [f32; M],
-}
-
-impl<'a, const M: usize> Arbitrary<'a> for Item<M> {
-    fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
-        let data: [f32; M] = u.arbitrary()?;
-        let id: u32 = u.arbitrary()?;
-
-        Ok(Item { data, id })
-    }
-}
-
 #[derive(Arbitrary, Debug)]
 enum WriteOp<const M: usize> {
-    Add(Item<M>),
+    Add(u32),
     Del(u32),
 }
 
@@ -38,44 +36,106 @@ fn assert_all_readable<const DIM: usize>(rtxn: &RoTxn, database: Database<Cosine
     assert_eq!(&RoaringBitmap::from_iter(found.into_iter().map(|(id, _)| id)), reader.item_ids())
 }
 
+fn assert_deleted_items_are_gone(
+    rtxn: &RoTxn,
+    database: Database<Cosine>,
+    deleted: &RoaringBitmap,
+) {
+    // assert the reader cannot find any deleted vectors
+    let reader = Reader::<Cosine>::open(&rtxn, 0, database).unwrap();
+    let item_intersection = deleted & reader.item_ids();
+    assert!(item_intersection.is_empty(), "{:?} should be deleted!", item_intersection);
+
+    // now iter over ALL links and assert no connection exists to a deleted item
+    let mut cursor = database
+        .remap_types::<PrefixCodec, NodeCodec<Cosine>>()
+        .prefix_iter(rtxn, &Prefix::links(0))
+        .unwrap()
+        .remap_key_type::<KeyCodec>();
+
+    while let Some((key, node)) = cursor.next().transpose().unwrap() {
+        assert!(
+            !deleted.contains(key.node.item),
+            "the item and its data should be deleted!\n{:?}",
+            &key
+        );
+
+        match key.node.mode {
+            NodeMode::Links => {
+                if let Node::Links(Links { links: links_bitmap }) = node {
+                    let link_intersection = deleted & links_bitmap.as_ref();
+                    assert!(
+                        link_intersection.is_empty(),
+                        "LINKS VIOLATION: {:?} should be empty",
+                        link_intersection
+                    );
+                }
+            }
+            _ => continue,
+        }
+    }
+}
+
 #[test]
-#[ignore = "if working properly this should run infinitely"]
 fn random_read_writes() {
     let seed: u64 = rand::random();
     let mut rng = StdRng::seed_from_u64(seed);
 
-    const DIM: usize = 768;
+    const DIM: usize = 32;
     const NUMEL: usize = 1000;
     const M: usize = 16;
-    const M0: usize = 32;
+    const M0: usize = 768;
 
     let DatabaseHandle { env, database, tempdir: _ } =
         create_database_indices_with_items::<Cosine, DIM, M, M0, _>(0..1, NUMEL, &mut rng);
 
-    for _ in 0.. {
+    let mut deleted = RoaringBitmap::new();
+    let mut vec_rng = rand::thread_rng();
+
+    // util for generating new vectors on the fly
+    fn gen_vec(rng: &mut ThreadRng) -> [f32; DIM] {
+        let unif = Uniform::new(-1.0, 1.0);
+        std::array::from_fn(|_| rng.sample(unif))
+    }
+
+    let duration = match std::env::var("HANNOY_FUZZ_DURATION_SEC") {
+        Ok(value) => Duration::from_secs(value.parse().expect("valid number of seconds")),
+        Err(VarError::NotPresent) => Duration::from_secs(20),
+        Err(VarError::NotUnicode(e)) => panic!("Invalid duration: {e:?}"),
+    };
+
+    let before = Instant::now();
+    while before.elapsed() < duration {
         let rtxn = env.read_txn().unwrap();
         assert_all_readable::<DIM>(&rtxn, database);
+        assert_deleted_items_are_gone(&rtxn, database, &deleted);
+        deleted.clear();
 
         // get batch of write operations and apply them
         info!("WRITING");
         let mut data = [0_u8; 1024 * 1024 * 1];
         rng.fill(&mut data);
         let mut u = Unstructured::new(&data);
-        let ops: Vec<WriteOp<DIM>> = (0..1000).map(|_| u.arbitrary().unwrap()).collect();
+        let ops: Vec<WriteOp<DIM>> = (0..100).map(|_| u.arbitrary().unwrap()).collect();
 
         let mut wtxn = env.write_txn().unwrap();
         let writer = Writer::new(database, 0, DIM);
 
         for op in ops {
             match op {
-                WriteOp::Add(item) => {
-                    let Item { data, id } = item;
+                WriteOp::Add(id) => {
                     let id = id % (NUMEL as u32);
-                    writer.add_item(&mut wtxn, id, &data).unwrap();
+                    let vector = gen_vec(&mut vec_rng);
+                    assert!(vector != [0.0f32; DIM]);
+                    writer.add_item(&mut wtxn, id, &vector).unwrap();
+
+                    // ensure added random id isn't registered in deleted
+                    let _ = deleted.remove(id);
                 }
                 WriteOp::Del(id) => {
                     let id = id % (NUMEL as u32);
                     let _ = writer.del_item(&mut wtxn, id).unwrap();
+                    deleted.insert(id);
                 }
             }
         }