diff --git a/src/VecSim/algorithms/hnsw/hnsw.h b/src/VecSim/algorithms/hnsw/hnsw.h index 2c1fae87f..581bca7e7 100644 --- a/src/VecSim/algorithms/hnsw/hnsw.h +++ b/src/VecSim/algorithms/hnsw/hnsw.h @@ -205,7 +205,7 @@ class HNSWIndex : public VecSimIndexAbstract, idType id) const; void emplaceToHeap(vecsim_stl::abstract_priority_queue &heap, DistType dist, idType id) const; - void removeAndSwap(idType internalId); + void swapWithLast(idType removedId); size_t getVectorRelativeIndex(idType id) const { return id % this->blockSize; } @@ -279,6 +279,24 @@ class HNSWIndex : public VecSimIndexAbstract, void unmarkInProcess(idType internalId); HNSWAddVectorState storeNewElement(labelType label, const void *vector_data); void removeAndSwapMarkDeletedElement(idType internalId); + void removeFromGraph(idType internalId); + // Whether `level_data` holds a link to `id`. + static bool hasLink(const ElementLevelData &level_data, idType id) { + for (size_t i = 0; i < level_data.getNumLinks(); i++) { + if (level_data.getLinkAtPos(i) == id) { + return true; + } + } + return false; + } + // Take a marked-deleted element out of the graph entirely: drop every edge going out of it and + // every edge coming into it, at every level, keeping the bookkeeping of the other side + // consistent. To be called once all the repair jobs created for its deletion are done and + // before its swap job disposes of it, so that from that point no element refers to it and it + // refers to no element. + // Takes the per-element links locks (one at a time), so holding the main index guard for shared + // ownership is enough. + void isolateDeletedElement(idType internalId); void repairNodeConnections(idType node_id, size_t level); // For prefetching only. const ElementMetaData *getMetaDataAddress(idType internal_id) const { @@ -1644,20 +1662,89 @@ HNSWIndex::~HNSWIndex() { */ template -void HNSWIndex::removeAndSwap(idType internalId) { +void HNSWIndex::isolateDeletedElement(idType internalId) { + assert(isMarkedDeleted(internalId) && "Only a marked-deleted element may be isolated"); + auto element = getGraphDataByInternalId(internalId); + for (size_t level = 0; level <= element->toplevel; level++) { + // Collect the elements this one shares an edge with at this level, in either direction. No + // edge can be added to a deleted element, so this set only shrinks from here on (a repair + // job of another element may still remove an edge concurrently). + lockNodeLinks(internalId); + ElementLevelData &level_data = getElementLevelData(element, level); + auto others = level_data.copyLinks(); + others.insert(others.end(), level_data.getIncomingEdges().begin(), + level_data.getIncomingEdges().end()); + unlockNodeLinks(internalId); + + for (idType other_id : others) { + // Remove both sides of the edge as one atomic step, holding the two elements' locks in + // ascending id order (as every other multi-lock site here does, to avoid deadlocks). + // Doing it mutually keeps the "an edge is recorded on exactly one side" invariant true + // at every observable point, which is what lets `mutuallyRemoveNeighborAtPos` tell a + // bidirectional edge from a unidirectional one by the record alone. + idType first = std::min(internalId, other_id); + idType second = std::max(internalId, other_id); + lockNodeLinks(first); + lockNodeLinks(second); + + ElementLevelData &other = getElementLevelData(other_id, level); + bool points_to_other = hasLink(level_data, other_id); + bool other_points_here = hasLink(other, internalId); + + if (points_to_other && other_points_here) { + // Bidirectional, so neither side recorded it as an incoming edge - just drop both + // links. Only two deleted elements can still point at each other at this stage: + // neither of them gets a repair job for the other's deletion. + assert(isMarkedDeleted(other_id) && + "a live element still points to a fully repaired deleted element"); + level_data.removeLink(other_id); + other.removeLink(internalId); + } else if (points_to_other) { + // Unidirectional out - the other side recorded it as an incoming edge. + level_data.removeLink(other_id); + bool res = other.removeIncomingUnidirectionalEdgeIfExists(internalId); + (void)res; + assert(res && "The edge should be in the incoming unidirectional edges"); + } else if (other_points_here) { + // Unidirectional in - recorded as an incoming edge here. As above, at this stage it + // can only come from another deleted element. + assert(isMarkedDeleted(other_id) && + "a live element still points to a fully repaired deleted element"); + other.removeLink(internalId); + bool res = level_data.removeIncomingUnidirectionalEdgeIfExists(other_id); + (void)res; + assert(res && "The edge should be in the incoming unidirectional edges"); + } + // Else the edge is already gone - a repair job of another element got to it first. + + unlockNodeLinks(second); + unlockNodeLinks(first); + } + + lockNodeLinks(internalId); + assert(level_data.getNumLinks() == 0 && level_data.getIncomingEdges().empty() && + "the element should have no edge left at this level"); + unlockNodeLinks(internalId); + } +} + +template +void HNSWIndex::removeFromGraph(idType internalId) { // Sanity check - the id to remove cannot be the entry point, as it should have been replaced // upon marking it as deleted. assert(entrypointNode != internalId); auto element = getGraphDataByInternalId(internalId); - // Remove the deleted id form the relevant incoming edges sets in which it appears. + // Remove the deleted id form the relevant incoming edges sets in which it appears. For an + // asynchronously deleted element there is nothing to walk here: `isolateDeletedElement` already + // took all of its edges out when its last repair job completed. for (size_t level = 0; level <= element->toplevel; level++) { ElementLevelData &cur_level = getElementLevelData(element, level); for (size_t i = 0; i < cur_level.getNumLinks(); i++) { ElementLevelData &neighbour = getElementLevelData(cur_level.getLinkAtPos(i), level); - // Note that in case of in-place delete, we might have not accounted for this edge in + // Note that in case of in-place delete, we might have not accounted for this edge // in the unidirectional edges, since there is no point in keeping it there temporarily - // (we know we will get here and remove this deleted id permanently). + // . (We know we will get here and remove this deleted id permanently.) // However, upon asynchronous delete, this should always succeed since we do update // the incoming edges in the mutual update even for deleted elements. bool res = neighbour.removeIncomingUnidirectionalEdgeIfExists(internalId); @@ -1673,16 +1760,19 @@ void HNSWIndex::removeAndSwap(idType internalId) { // We can say now that the element has removed completely from index. --curElementCount; +} +template +void HNSWIndex::swapWithLast(idType removedId) { // Get the last element's metadata and data. - // If we are deleting the last element, we already destroyed it's metadata. + // If we are deleting the last element, we already destroyed its metadata. auto *last_element_data = getDataByInternalId(curElementCount); DataBlock &last_gd_block = graphDataBlocks.back(); auto last_element = (ElementGraphData *)last_gd_block.removeAndFetchLastElement(); // Swap the last id with the deleted one, and invalidate the last id data. - if (curElementCount != internalId) { - SwapLastIdWithDeletedId(internalId, last_element, last_element_data); + if (curElementCount != removedId) { + SwapLastIdWithDeletedId(removedId, last_element, last_element_data); } // If we need to free a complete block and there is at least one block between the @@ -1693,7 +1783,8 @@ void HNSWIndex::removeAndSwap(idType internalId) { template void HNSWIndex::removeAndSwapMarkDeletedElement(idType internalId) { - removeAndSwap(internalId); + removeFromGraph(internalId); + swapWithLast(internalId); // element is permanently removed from the index, it is no longer counted as marked deleted. --numMarkedDeleted; } @@ -1756,7 +1847,8 @@ void HNSWIndex::removeVectorInPlace(const idType element_int } // Finally, remove the element from the index and make a swap with the last internal id to // avoid fragmentation and reclaim memory when needed. - removeAndSwap(element_internal_id); + removeFromGraph(element_internal_id); + swapWithLast(element_internal_id); } // Store the new element in the global data structures and keep the new state. In multithreaded diff --git a/src/VecSim/algorithms/hnsw/hnsw_tiered.h b/src/VecSim/algorithms/hnsw/hnsw_tiered.h index a4d5e08e4..b90ba8e69 100644 --- a/src/VecSim/algorithms/hnsw/hnsw_tiered.h +++ b/src/VecSim/algorithms/hnsw/hnsw_tiered.h @@ -104,7 +104,8 @@ class TieredHNSWIndex : public VecSimTieredIndex { // To be executed synchronously upon deleting a vector, doesn't require a wrapper. Main HNSW // lock is assumed to be held exclusive here. - void executeSwapJob(idType deleted_id, vecsim_stl::vector &idsToRemove); + void fixJobsAfterSwap(idType deleted_id, vecsim_stl::vector &idsToRemove); + void invalidateRepairJobs(idType deleted_id); // Execute the ready swap jobs, run no more than 'maxSwapsToRun' jobs (run all of them for -1). void executeReadySwapJobs(size_t maxSwapsToRun = -1); @@ -128,6 +129,12 @@ class TieredHNSWIndex : public VecSimTieredIndex { // while *HNSW shared lock is held* (shared locked). int deleteLabelFromHNSW(labelType label); + // Take a deleted element out of the graph once the last repair job created for its deletion is + // done - no edge in or out of it is left, so the swap job that disposes of it later only has to + // reclaim its slot. Called in the repair context, while the main index guard is held for shared + // ownership and `idToRepairJobsGuard` is not held. + void isolateRepairedElement(idType deleted_id); + // Insert a single vector to HNSW. This can be called in both write modes - insert async and // in-place. For the async mode, we have to release the flat index guard that is held for shared // ownership (we do it right after we update the HNSW global data and receive the new state). @@ -283,23 +290,11 @@ void TieredHNSWIndex::executeRepairJobWrapper(AsyncJob *job) } template -void TieredHNSWIndex::executeSwapJob(idType deleted_id, - vecsim_stl::vector &idsToRemove) { - // Get the id that was last and was had been swapped with the job's deleted id. +void TieredHNSWIndex::fixJobsAfterSwap( + idType deleted_id, vecsim_stl::vector &idsToRemove) { + // Get the id that was last and had been swapped with the job's deleted id. idType prev_last_id = this->getHNSWIndex()->indexSize(); - // Invalidate repair jobs for the disposed id (if exist), and update the associated swap jobs. - if (idToRepairJobs.find(deleted_id) != idToRepairJobs.end()) { - for (auto &job_it : idToRepairJobs.at(deleted_id)) { - job_it->node_id = this->setAndSaveInvalidJob(job_it); - for (auto &swap_job_it : job_it->associatedSwapJobs) { - if (swap_job_it->atomicDecreasePendingJobsNum() == 0) { - readySwapJobs++; - } - } - } - idToRepairJobs.erase(deleted_id); - } // Swap the ids in the pending jobs for the current last id (if exist). if (idToRepairJobs.find(prev_last_id) != idToRepairJobs.end()) { for (auto &job_it : idToRepairJobs.at(prev_last_id)) { @@ -324,11 +319,34 @@ void TieredHNSWIndex::executeSwapJob(idType deleted_id, } } +template +void TieredHNSWIndex::invalidateRepairJobs(idType deleted_id) { + // Invalidate repair jobs for the disposed id (if exist), and update the associated swap jobs. + if (idToRepairJobs.find(deleted_id) == idToRepairJobs.end()) { + return; + } + + for (auto &job_it : idToRepairJobs.at(deleted_id)) { + job_it->node_id = this->setAndSaveInvalidJob(job_it); + for (auto &swap_job_it : job_it->associatedSwapJobs) { + if (swap_job_it->atomicDecreasePendingJobsNum() == 0) { + readySwapJobs++; + } + } + } + idToRepairJobs.erase(deleted_id); +} + template HNSWIndex *TieredHNSWIndex::getHNSWIndex() const { return dynamic_cast *>(this->backendIndex); } +template +void TieredHNSWIndex::isolateRepairedElement(idType deleted_id) { + this->getHNSWIndex()->isolateDeletedElement(deleted_id); +} + template void TieredHNSWIndex::executeReadySwapJobs(size_t maxJobsToRun) { @@ -342,10 +360,12 @@ void TieredHNSWIndex::executeReadySwapJobs(size_t maxJobsToR idsToRemove.reserve(idToSwapJob.size()); for (auto &it : idToSwapJob) { auto *swap_job = it.second; + // Swap job is ready for execution - execute and delete it. if (swap_job->pending_repair_jobs_counter.load() == 0) { - // Swap job is ready for execution - execute and delete it. - this->getHNSWIndex()->removeAndSwapMarkDeletedElement(swap_job->deleted_id); - this->executeSwapJob(swap_job->deleted_id, idsToRemove); + auto deleted_id = swap_job->deleted_id; + this->getHNSWIndex()->removeAndSwapMarkDeletedElement(deleted_id); + this->invalidateRepairJobs(deleted_id); + this->fixJobsAfterSwap(deleted_id, idsToRemove); delete swap_job; } if (maxJobsToRun > 0 && idsToRemove.size() >= maxJobsToRun) { @@ -415,6 +435,13 @@ int TieredHNSWIndex::deleteLabelFromHNSW(labelType label) { } this->idToRepairJobsGuard.unlock(); + if (incomingEdges.size() == 0) { + // No repair job will ever run for this element, so this is already the point at which + // it can be taken out of the graph (outside the repair jobs guard, as isolating takes + // the per-element links locks). + this->isolateRepairedElement(id); + } + this->submitJobs(repair_jobs); // Insert the swap job into the swap jobs lookup (for fast update in case that the // node id is changed due to swap job). @@ -533,7 +560,8 @@ int TieredHNSWIndex::deleteLabelFromHNSWInplace(labelType la // Get the id in every iteration, since the ids can be swapped in every iteration. idType id = hnsw_index->getElementIds(label).at(id_ind); hnsw_index->removeVectorInPlace(id); - this->executeSwapJob(id, idsToRemove); + this->invalidateRepairJobs(id); + this->fixJobsAfterSwap(id, idsToRemove); } hnsw_index->removeLabel(label); for (idType id : idsToRemove) { @@ -639,14 +667,29 @@ void TieredHNSWIndex::executeRepairJob(HNSWRepairJob *job) { *it = repair_jobs.back(); repair_jobs.pop_back(); } + this->idToRepairJobsGuard.unlock(); + + hnsw_index->repairNodeConnections(job->node_id, job->level); + + // Account for this job only now that its repair has actually been performed. Decreasing the + // counter beforehand would let a swap job be seen as ready, and its element isolated, while + // an element still points to it from a repair that has not run yet. + vecsim_stl::vector fully_repaired_ids(this->allocator); + this->idToRepairJobsGuard.lock(); for (auto &it : job->associatedSwapJobs) { if (it->atomicDecreasePendingJobsNum() == 0) { readySwapJobs++; + fully_repaired_ids.push_back(it->deleted_id); } } this->idToRepairJobsGuard.unlock(); - hnsw_index->repairNodeConnections(job->node_id, job->level); + // These deleted elements have no pending repair job left, so nothing points to them anymore. + // Take them out of the graph entirely, leaving no edge in or out for the swap job to deal with. + // Done outside the repair jobs guard, as isolating takes the per-element links locks. + for (idType deleted_id : fully_repaired_ids) { + this->isolateRepairedElement(deleted_id); + } this->mainIndexGuard.unlock_shared(); } diff --git a/src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h b/src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h index 21f99f8f5..d4d5cd999 100644 --- a/src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h +++ b/src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h @@ -24,6 +24,7 @@ INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_deleteVectorAndRepairAsync_Test) INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_alternateInsertDeleteAsync_Test) INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_swapJobBasic_Test) INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_swapJobBasic2_Test) +INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_invalidRepairJobOnSwap_Test) INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_deleteVectorsAndSwapSync_Test) INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_BatchIterator_Test) INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTest_BatchIteratorAdvanced_Test) diff --git a/tests/unit/test_hnsw_tiered.cpp b/tests/unit/test_hnsw_tiered.cpp index 21f504177..a0927790b 100644 --- a/tests/unit/test_hnsw_tiered.cpp +++ b/tests/unit/test_hnsw_tiered.cpp @@ -1888,18 +1888,18 @@ TYPED_TEST(HNSWTieredIndexTest, swapJobBasic2) { ASSERT_EQ(mock_thread_pool.jobQ.front().job->jobType, HNSW_REPAIR_NODE_CONNECTIONS_JOB); mock_thread_pool.thread_iteration(); EXPECT_EQ(tiered_index->idToSwapJob.at(0)->pending_repair_jobs_counter.load(), 0); - // Delete 2, expect to create two repair job pending from 0 and 1. Also, expect that swap - // job for 0 will be executed, so that 2 and 0 are swapped. Then, we should have only 1 - // pending repair job for the "new" 0 - for deleting the old 1->2, while the second job for - // deleting the old 0->2 is invalid and reduced from the pending repair jobs counter. + // Delete 2. Only the 1->2 edge is left to repair: 0 was taken out of the graph when its own + // repair jobs completed, so it no longer points to 2 and no 0->2 job is created. Also, expect + // that swap job for 0 will be executed, so that 2 and 0 are swapped - the single pending job is + // then 1->0 (originally 1->2). EXPECT_EQ(tiered_index->deleteVector(2), 1); EXPECT_EQ(tiered_index->indexSize(), 2); EXPECT_EQ(tiered_index->getHNSWIndex()->getNumMarkedDeleted(), 1); EXPECT_EQ(tiered_index->statisticInfo().numberOfMarkedDeleted, 1); EXPECT_EQ(tiered_index->idToSwapJob.at(0)->pending_repair_jobs_counter.load(), 1); - EXPECT_EQ(mock_thread_pool.jobQ.size(), 2); - // The first repair job should remove 1->0 (originally was 1->2). + EXPECT_EQ(mock_thread_pool.jobQ.size(), 1); + // The repair job should remove 1->0 (originally was 1->2). ASSERT_EQ(mock_thread_pool.jobQ.front().job->jobType, HNSW_REPAIR_NODE_CONNECTIONS_JOB); ASSERT_EQ(reinterpret_cast(mock_thread_pool.jobQ.front().job)->node_id, 1); ASSERT_EQ(reinterpret_cast(mock_thread_pool.jobQ.front().job) @@ -1908,34 +1908,15 @@ TYPED_TEST(HNSWTieredIndexTest, swapJobBasic2) { 0); mock_thread_pool.thread_iteration(); EXPECT_EQ(tiered_index->idToSwapJob.at(0)->pending_repair_jobs_counter.load(), 0); - // The second repair job is invalid due to the removal of (the original) 0. - ASSERT_EQ(mock_thread_pool.jobQ.front().job->jobType, HNSW_REPAIR_NODE_CONNECTIONS_JOB); - ASSERT_EQ(mock_thread_pool.jobQ.front().job->isValid, false); - ASSERT_EQ(reinterpret_cast(mock_thread_pool.jobQ.front().job)->node_id, - invalid_jobs_counter++); - ASSERT_EQ(reinterpret_cast(mock_thread_pool.jobQ.front().job) - ->associatedSwapJobs[0] - ->deleted_id, - 0); - mock_thread_pool.thread_iteration(); - // Delete 1, that should still have 0->1 edge that should be repaired. This should cause - // the swap and removal of 0 (that has no more pending jobs at that point) - so that 1 would - // get id 0, and then the new 0 should have no pending repair jobs. + // Delete 1. The only other element left (the "new" 0, which is the old 2) is deleted and was + // already taken out of the graph, so nothing points to 1 and no u->1 job is created. + // Its swap job is therefore ready right away, and the swap and removal of the previous 0 is + // triggered - so that 1 gets id 0. EXPECT_EQ(tiered_index->deleteVector(1), 1); - EXPECT_EQ(mock_thread_pool.jobQ.size(), 1); + EXPECT_EQ(mock_thread_pool.jobQ.size(), 0); EXPECT_EQ(tiered_index->idToSwapJob.size(), 1); EXPECT_EQ(tiered_index->idToSwapJob.at(0)->deleted_id, 0); EXPECT_EQ(tiered_index->idToSwapJob.at(0)->pending_repair_jobs_counter.load(), 0); - // The repair job is invalid due to the removal of (the previous) 0. - ASSERT_EQ(mock_thread_pool.jobQ.front().job->jobType, HNSW_REPAIR_NODE_CONNECTIONS_JOB); - ASSERT_EQ(mock_thread_pool.jobQ.front().job->isValid, false); - ASSERT_EQ(reinterpret_cast(mock_thread_pool.jobQ.front().job)->node_id, - invalid_jobs_counter); - ASSERT_EQ(reinterpret_cast(mock_thread_pool.jobQ.front().job) - ->associatedSwapJobs[0] - ->deleted_id, - 0); - mock_thread_pool.thread_iteration(); EXPECT_EQ(tiered_index->indexSize(), 1); EXPECT_EQ(tiered_index->getHNSWIndex()->getNumMarkedDeleted(), 1); EXPECT_EQ(tiered_index->statisticInfo().numberOfMarkedDeleted, 1); @@ -1949,6 +1930,80 @@ TYPED_TEST(HNSWTieredIndexTest, swapJobBasic2) { EXPECT_EQ(tiered_index->statisticInfo().numberOfMarkedDeleted, 0); } +// Covers the invalidation of a pending repair job whose node is disposed of by a swap job. A repair +// job is denoted below as the edge it removes: u->v is the job that repairs u's connections after +// its neighbour v was deleted. +// A deleted element is taken out of the graph as soon as the jobs of *its own* deletion are done, +// so for a job on it to still be pending when it is disposed of, that job has to belong to +// *another* element's deletion: 0 is deleted first, then 1 is deleted while 0 still points to it +// (registering a 0->1 job), and only then the 1->0 and 2->0 jobs complete and 0 is swapped out. +TYPED_TEST(HNSWTieredIndexTest, invalidRepairJobOnSwap) { + size_t dim = 4; + HNSWParams params = {.type = TypeParam::get_index_type(), + .dim = dim, + .metric = VecSimMetric_L2, + .multi = TypeParam::isMulti()}; + VecSimParams hnsw_params = CreateParams(params); + auto mock_thread_pool = tieredIndexMock(); + // Threshold of 1, so that a ready swap job is executed at the first opportunity. + auto *tiered_index = this->CreateTieredHNSWIndex(hnsw_params, mock_thread_pool, 1); + + // Insert 3 vectors directly into HNSW, expect to have a fully connected graph. + for (size_t i = 0; i < 3; i++) { + GenerateAndAddVector(tiered_index->backendIndex, dim, i, i); + } + + // Delete 0 - a u->0 job is created for every (u, level) pair that points to it, that is 1->0 + // and 2->0. Note that the number of jobs depends on the levels the elements got (and that jobs + // of the same (u, level) are merged), so the counters are compared to each other rather than to + // fixed values below. + EXPECT_EQ(tiered_index->deleteVector(0), 1); + ASSERT_GT(mock_thread_pool.jobQ.size(), 0); + ASSERT_GT(tiered_index->idToSwapJob.at(0)->pending_repair_jobs_counter.load(), 0); + + // Delete 1 before those jobs run. 0 is deleted but still connected (its own repairs are + // pending), so a 0->1 job is created here and stays pending. + EXPECT_EQ(tiered_index->deleteVector(1), 1); + ASSERT_TRUE(tiered_index->idToRepairJobs.contains(0)); + + // Execute the 1->0 and 2->0 jobs, so that 0 has no pending repair job left and is taken out of + // the graph, making its swap job ready. The 0->1 job, which belongs to 1's swap job, is still + // queued. + while (tiered_index->idToSwapJob.at(0)->pending_repair_jobs_counter.load() > 0) { + ASSERT_GT(mock_thread_pool.jobQ.size(), 0); + mock_thread_pool.thread_iteration(); + } + ASSERT_TRUE(tiered_index->idToRepairJobs.contains(0)); + ASSERT_EQ(tiered_index->invalidJobs.size(), 0); + int pending_for_1 = tiered_index->idToSwapJob.at(1)->pending_repair_jobs_counter.load(); + ASSERT_GT(pending_for_1, 0); + + // Dispose of 0. The pending 0->1 job has to be invalidated, and 1's swap job should stop + // waiting for it. + tiered_index->runGC(); + EXPECT_EQ(tiered_index->indexSize(), 2); + EXPECT_EQ(tiered_index->invalidJobs.size(), 1); + EXPECT_EQ(tiered_index->idToSwapJob.at(1)->pending_repair_jobs_counter.load(), + pending_for_1 - 1); + + // Drain the remaining jobs: the invalidated 0->1 job is disposed of without being executed, and + // the 2->1 job (whose node id was renamed by the swap above) completes 1's repairs - so 1 is + // taken out of the graph as well. + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + EXPECT_EQ(tiered_index->invalidJobs.size(), 0); + EXPECT_EQ(tiered_index->idToSwapJob.at(1)->pending_repair_jobs_counter.load(), 0); + + // Disposing of 1 as well leaves a single element in the index, with a valid graph. + tiered_index->runGC(); + EXPECT_EQ(tiered_index->indexSize(), 1); + EXPECT_EQ(tiered_index->getHNSWIndex()->getNumMarkedDeleted(), 0); + auto state = tiered_index->getHNSWIndex()->checkIntegrity(); + EXPECT_EQ(state.valid_state, true); + EXPECT_EQ(state.connections_to_repair, 0); +} + // A set of lambdas that determine whether a vector should be inserted to the // HNSW index (returns true) or to the flat index (returns false). inline constexpr std::array, 11> lambdas = {{