From 7692f8eb56864217187a3295d593a9f99ebbd3f1 Mon Sep 17 00:00:00 2001 From: sharonrosenfeld Date: Tue, 20 Oct 2026 08:57:55 +0300 Subject: [PATCH] [MOD-18471] Fix SVS marked-deletion GC race --- src/VecSim/algorithms/svs/svs.h | 10 +++++ src/VecSim/algorithms/svs/svs_tiered.h | 20 +++++++--- tests/unit/test_svs_tiered.cpp | 55 ++++++++++++++++++++++++++ 3 files changed, 80 insertions(+), 5 deletions(-) diff --git a/src/VecSim/algorithms/svs/svs.h b/src/VecSim/algorithms/svs/svs.h index 2b998dc11..73a86b196 100644 --- a/src/VecSim/algorithms/svs/svs.h +++ b/src/VecSim/algorithms/svs/svs.h @@ -79,6 +79,7 @@ struct SVSIndexBase // the checks preceding those calls rule out, and it grows its dataset rather than refusing a // write, so without this the code that reports a failed write would go untested. virtual void throwOnNextWriteForTest() = 0; + virtual void setAfterCompactCallbackForTest(std::function callback) = 0; #endif protected: // Index marked deleted vectors counter to initiate reindexing if it exceeds threshold @@ -305,9 +306,13 @@ class SVSIndex : public VecSimIndexAbstract, fl #ifdef BUILD_TESTS public: void throwOnNextWriteForTest() override { throw_on_next_write_ = true; } + void setAfterCompactCallbackForTest(std::function callback) override { + after_compact_callback_ = std::move(callback); + } private: bool throw_on_next_write_ = false; + std::function after_compact_callback_; void maybeThrowForTest() { if (throw_on_next_write_) { throw_on_next_write_ = false; @@ -901,6 +906,11 @@ class SVSIndex : public VecSimIndexAbstract, fl // pending per-label consolidate jobs need not be drained - they become no-ops impl->compact(); } +#ifdef BUILD_TESTS + if (after_compact_callback_) { + after_compact_callback_(); + } +#endif num_marked_deleted.store(0, std::memory_order_relaxed); } diff --git a/src/VecSim/algorithms/svs/svs_tiered.h b/src/VecSim/algorithms/svs/svs_tiered.h index ee66005e0..573a0f240 100644 --- a/src/VecSim/algorithms/svs/svs_tiered.h +++ b/src/VecSim/algorithms/svs/svs_tiered.h @@ -600,7 +600,8 @@ class TieredSVSIndex : public VecSimTieredIndex { * * This static wrapper function performs the following actions: * - Acquires updateJobMutex exclusively to wait for backend update jobs. - * - Acquires mainIndexGuard exclusively to exclude base-class readers during GC. + * - Acquires mainIndexGuard exclusively to exclude base-class readers and backend deletions + * during GC. * - Configures the number of threads for the underlying SVS index update operation. * - Calls the SVSIndex::runGC() method to perform the actual index update. * - Clears the indexGCScheduled flag to allow future scheduling. @@ -980,14 +981,14 @@ class TieredSVSIndex : public VecSimTieredIndex { } // backend index is initialized - we can add the vector directly auto storage_blob = this->frontendIndex->preprocessForStorage(blob); - // Only updateJobMutex is needed here + // Match GC's lock order: updateJobMutex, then mainIndexGuard in the delete helper. std::lock_guard lock(this->updateJobMutex); // Defensive: ensure single-threaded operation for write-in-place mode. // parallelism_ defaults to 1, so this is a no-op in the normal case. svs_index->setParallelism(1); int deleted = 0; if (!this->backendIndex->isMultiValue()) { - deleted = svs_index->deleteVector(label); + deleted = deleteVectorFromBackend(label); if (deleted > 0) svs_index->consolidate({label}); } @@ -1055,7 +1056,7 @@ class TieredSVSIndex : public VecSimTieredIndex { if (!this->backendIndex->isMultiValue() && !defer_until_initialized) { // If we removed the previous vector from both svs and flat in the overwrite // process, we still return 0 (not -1). - auto deleted = svs_index->deleteVector(label); + auto deleted = deleteVectorFromBackend(label); if (deleted > 0) scheduleSVSIndexConsolidate(label); ret = std::max(ret - deleted, 0); @@ -1191,6 +1192,15 @@ class TieredSVSIndex : public VecSimTieredIndex { return ret; } +private: + int deleteVectorFromBackend(labelType label) { + // GC holds mainIndexGuard exclusively through both compact() and the deletion-counter + // reset, so a backend deletion and its counter increment belong wholly to one GC epoch. + std::shared_lock main_lock{this->mainIndexGuard}; + return this->backendIndex->deleteVector(label); + } + +public: int deleteVector(labelType label) override { int ret = 0; @@ -1203,7 +1213,7 @@ class TieredSVSIndex : public VecSimTieredIndex { this->flatIndexGuard.unlock_shared(); } - auto deleted = this->backendIndex->deleteVector(label); + auto deleted = deleteVectorFromBackend(label); if (deleted > 0) { if (this->getWriteMode() == VecSim_WriteInPlace) { GetSVSIndex()->consolidate({label}); diff --git a/tests/unit/test_svs_tiered.cpp b/tests/unit/test_svs_tiered.cpp index 6fabc1cf5..8098045e5 100644 --- a/tests/unit/test_svs_tiered.cpp +++ b/tests/unit/test_svs_tiered.cpp @@ -2814,6 +2814,61 @@ TYPED_TEST(SVSTieredIndexTestBasic, markedDeleted) { ASSERT_EQ(tiered_index->indexLabelCount(), 0); } +TYPED_TEST(SVSTieredIndexTestBasic, deletionAfterCompactWaitsForCounterReset) { + constexpr size_t dim = 4; + constexpr size_t n = 2; + SVSParams params = {.type = TypeParam::get_index_type(), .dim = dim, .metric = VecSimMetric_L2}; + VecSimParams svs_params = CreateParams(params); + auto mock_thread_pool = tieredIndexMock(1); + auto *tiered_index = this->CreateTieredSVSIndex(svs_params, mock_thread_pool, n, n); + ASSERT_INDEX(tiered_index); + + for (labelType label = 0; label < n; ++label) { + GenerateAndAddVector(tiered_index, dim, label); + } + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), n); + + ASSERT_EQ(tiered_index->deleteVector(0), 1); + ASSERT_EQ(tiered_index->GetSVSIndex()->getNumMarkedDeleted(), 1); + + std::future concurrent_delete; + bool deletion_started_after_compact = false; + bool deletion_was_blocked = false; + tiered_index->GetSVSIndex()->setAfterCompactCallbackForTest([&] { + // GC still holds mainIndexGuard exclusively here. A tiered deletion may reach this window + // without updateJobMutex, but its shared mainIndexGuard must wait until after the reset. + concurrent_delete = + std::async(std::launch::async, [&] { return tiered_index->deleteVector(1); }); + deletion_started_after_compact = true; + deletion_was_blocked = concurrent_delete.wait_for(std::chrono::milliseconds(100)) == + std::future_status::timeout; + }); + + VecSimTieredIndex_GC(tiered_index); + while (!deletion_started_after_compact && !mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + + ASSERT_TRUE(deletion_started_after_compact); + ASSERT_TRUE(deletion_was_blocked); + ASSERT_EQ(concurrent_delete.get(), 1); + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + ASSERT_EQ(tiered_index->GetSVSIndex()->getNumMarkedDeleted(), 1); + ASSERT_FALSE(tiered_index->GetSVSIndex()->isLabelExists(1)); + + tiered_index->GetSVSIndex()->setAfterCompactCallbackForTest({}); + VecSimTieredIndex_GC(tiered_index); + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + ASSERT_EQ(tiered_index->GetSVSIndex()->getNumMarkedDeleted(), 0); +} + TYPED_TEST(SVSTieredIndexTestBasic, deleteVectorMulti) { // Create TieredSVS index instance with a mock queue. size_t dim = 4;