From 2c12b8d0fccd35ce46ccb74057b8eea963d5c74d Mon Sep 17 00:00:00 2001 From: rtmalikian Date: Thu, 2 Jul 2026 12:09:49 -0700 Subject: [PATCH] fix: add stability check to wait_for_vector_indexing to prevent race condition Shards briefly report READY + vectorQueueSize=0 while objectCount has not yet caught up to the inserted objects. This causes wait_for_vector_indexing() to return too early. Fix: After detecting readiness, wait 0.5s and re-verify. If the shard is no longer ready, continue waiting. Fixes #1412 Signed-off-by: rtmalikian --- weaviate/collections/batch/batch_wrapper.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/weaviate/collections/batch/batch_wrapper.py b/weaviate/collections/batch/batch_wrapper.py index a3a3598d6..2b14c06f5 100644 --- a/weaviate/collections/batch/batch_wrapper.py +++ b/weaviate/collections/batch/batch_wrapper.py @@ -85,6 +85,13 @@ def wait_for_vector_indexing( logger.debug("Waiting for async indexing to finish...") time.sleep(0.25) waiting_count += 1 + # Stability check: wait briefly and re-verify to prevent race condition + # where shards report READY + vectorQueueSize=0 while objectCount + # has not yet caught up to the inserted objects. + # See: https://github.com/weaviate/weaviate-python-client/issues/1412 + time.sleep(0.5) + if not self.__is_ready(how_many_failures, shards): + return self.wait_for_vector_indexing(shards, how_many_failures) logger.debug("Async indexing finished!") def __get_shards_readiness(self, shard: Shard) -> List[bool]: @@ -186,6 +193,13 @@ async def wait_for_vector_indexing( logger.debug("Waiting for async indexing to finish...") await asyncio.sleep(0.25) waiting_count += 1 + # Stability check: wait briefly and re-verify to prevent race condition + # where shards report READY + vectorQueueSize=0 while objectCount + # has not yet caught up to the inserted objects. + # See: https://github.com/weaviate/weaviate-python-client/issues/1412 + await asyncio.sleep(0.5) + if not await self.__is_ready(how_many_failures, shards): + return await self.wait_for_vector_indexing(shards, how_many_failures) logger.debug("Async indexing finished!") async def __get_shards_readiness(self, shard: Shard) -> List[bool]: