take the points back too when an upsert fails

بررسیِ collectionها بعد از ingestِ پایه نهم نشان داد Qdrant ۲۸۷ نقطه دارد ولی
دیتابیس ۲۷۹ ردیف. هشت نقطه‌ی اضافه، همه از صفحه‌ی ۱۲۵ — همان صفحه‌ای که دو بار
تایم‌اوت داد و بار سوم گرفت. سه مسئله سه نسخه داشتند و دو نقطه هم بی‌صاحب بود.

علت: `upsert` با `wait=True` ممکن است نوشته باشد و فقط جوابش تایم‌اوت شده
باشد. مسیرِ جبران فقط ردیف‌های DB را پس می‌گرفت، نه نقطه‌ها؛ و چون هر اجرای
دوباره `qdrant_point_id` تازه می‌سازد، هر retry یک نسخه‌ی تکراری اضافه می‌کرد.

اثرش در بازیابی دیده می‌شود نه در لاگ: top-k با سه نسخه از یک مسئله پر می‌شود
و بقیه‌ی متنِ مرتبط بیرون می‌ماند — یعنی دستیار بافتِ کمتری می‌بیند و هیچ‌کس
هم متوجه نمی‌شود.

حالا در صورتِ شکست، نقطه‌ها هم پاک می‌شوند. اگر خودِ پاک‌سازی هم شکست بخورد
لاگ می‌شود ولی خطای اصلی پوشانده نمی‌شود، چون صفحه باید ناتمام بماند تا اجرای
بعدی دوباره سراغش برود.

داده‌ی موجود هم تمیز شد: ۸ نقطه‌ی بی‌صاحب پاک شدند و حالا ۲۷۹ = ۲۷۹.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-15 23:00:09 +03:30
parent 1131569b76
commit 92dc2242f8
3 changed files with 81 additions and 1 deletions

View File

@@ -19,7 +19,7 @@ from .config import build_collection_name, resolve_grade_title
from .embedding import embed_text from .embedding import embed_text
from .known_collections import is_math_subject, resolve_math_chapter from .known_collections import is_math_subject, resolve_math_chapter
from . import pdf as pdf_service from . import pdf as pdf_service
from .qdrant_store import ensure_collection, upsert_points from .qdrant_store import delete_points, ensure_collection, upsert_points
from .textbooks import Textbook, get_textbook from .textbooks import Textbook, get_textbook
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -140,7 +140,20 @@ def _persist_and_upsert(rows: list[dict], *, lesson_id, grade_id, subject,
for c, v in zip(created, vectors) for c, v in zip(created, vectors)
]) ])
except Exception: except Exception:
# ⚠️ نقطه‌ها را هم پس می‌گیریم، نه فقط ردیف‌ها. upsert با wait=True ممکن
# است نوشته باشد و فقط جوابش تایم‌اوت شده باشد؛ آن‌وقت پاک کردنِ صرفِ
# ردیف‌ها نقطه‌ی بی‌صاحب جا می‌گذارد، و چون اجرای بعدی شناسه‌ی تازه
# می‌سازد، هر تلاشِ دوباره یک نسخه‌ی تکراری اضافه می‌کند. روی صفحه‌ی ۱۲۵
# کتابِ نهم همین شد: سه نسخه از یک مسئله و دو نقطه‌ی یتیم.
point_ids = [str(c.qdrant_point_id) for c in created]
LessonChunk.objects.filter(pk__in=[c.pk for c in created]).delete() LessonChunk.objects.filter(pk__in=[c.pk for c in created]).delete()
try:
delete_points(collection, point_ids)
except Exception:
logger.exception(
'rag: پاک‌سازیِ %d نقطه‌ی نیمه‌نوشته در %r هم شکست خورد؛ '
'ممکن است نقطه‌ی بی‌صاحب بماند.', len(point_ids), collection,
)
logger.exception( logger.exception(
'rag: upsert به collection %r شکست خورد؛ %d ردیفِ همین دسته پس گرفته شد', 'rag: upsert به collection %r شکست خورد؛ %d ردیفِ همین دسته پس گرفته شد',
collection, len(created), collection, len(created),

View File

@@ -38,6 +38,22 @@ def upsert_points(collection: str, points: list[dict]) -> int:
return len(structs) return len(structs)
def delete_points(collection: str, point_ids: list[str]) -> int:
"""
پاک‌کردنِ نقاط با شناسه — برای جبرانِ upsertِ نیمه‌کاره.
`wait=True` در upsert یعنی تایم‌اوتِ کلاینت لزوماً به معنیِ ننوشتنِ Qdrant
نیست؛ ممکن است نوشته باشد و فقط جوابش دیر رسیده باشد. در آن حالت ردیف‌های
DB پس گرفته می‌شوند و نقطه‌ها بی‌صاحب می‌مانند. این تابع همان را جمع می‌کند.
"""
if not point_ids:
return 0
client = clients.get_qdrant_client()
client.delete(collection_name=collection,
points_selector=[str(i) for i in point_ids], wait=True)
return len(point_ids)
def search(collection: str, query_vector: list[float], *, lesson_filter_value=None, def search(collection: str, query_vector: list[float], *, lesson_filter_value=None,
lesson_filter_key: str | None = None, limit: int = 4) -> list[dict]: lesson_filter_key: str | None = None, limit: int = 4) -> list[dict]:
""" """

View File

@@ -860,3 +860,54 @@ class ReviewFixRegressionTests(TestCase):
# هیچ ردیفی نباید بماند، وگرنه _done_pages آن صفحه را انجام‌شده می‌بیند # هیچ ردیفی نباید بماند، وگرنه _done_pages آن صفحه را انجام‌شده می‌بیند
# و chunkها برای همیشه در DB می‌مانند بدونِ اینکه به Qdrant برسند. # و chunkها برای همیشه در DB می‌مانند بدونِ اینکه به Qdrant برسند.
self.assertEqual(LessonChunk.objects.filter(source_page=42).count(), 0) self.assertEqual(LessonChunk.objects.filter(source_page=42).count(), 0)
class IngestUpsertRollbackTests(TestCase):
"""
وقتی نوشتن در Qdrant شکست می‌خورد، هیچ ردِ نیمه‌کاره‌ای نباید بماند.
چرا این تست هست: upsert با wait=True ممکن است واقعاً نوشته باشد و فقط
جوابش تایم‌اوت شده باشد. تا امروز در آن حالت فقط ردیف‌های DB پس گرفته
می‌شدند و نقطه‌ها در Qdrant می‌ماندند؛ و چون هر اجرای دوباره شناسه‌ی تازه
می‌سازد، هر retry یک نسخه‌ی تکراری اضافه می‌کرد. روی صفحه‌ی ۱۲۵ کتابِ
ریاضی نهم همین اتفاق افتاد: سه نسخه از یک مسئله و دو نقطه‌ی بی‌صاحب.
"""
@override_settings(**RAG_ON)
def test_failed_upsert_removes_both_the_rows_and_the_points(self):
from .services.ingest import _persist_and_upsert
with patch('rag.services.ingest.ensure_collection'), \
patch('rag.services.ingest.embed_text', return_value=[0.0] * 8), \
patch('rag.services.ingest.upsert_points', side_effect=TimeoutError('boom')), \
patch('rag.services.ingest.delete_points') as delete_mock:
with self.assertRaises(TimeoutError):
_persist_and_upsert(
[{'order': 1, 'original_text': 'x', 'summary': 'خلاصه', 'source_page': 125}],
lesson_id=1, grade_id=1, subject='ریاضی', lesson_number=1,
chapter=7, title='عبارت‌های گویا', collection='ریاضی نهم',
)
self.assertEqual(LessonChunk.objects.filter(collection='ریاضی نهم').count(), 0)
delete_mock.assert_called_once()
collection, point_ids = delete_mock.call_args[0]
self.assertEqual(collection, 'ریاضی نهم')
self.assertEqual(len(point_ids), 1, 'شناسه‌ی همان نقطه باید برای پاک شدن برود.')
@override_settings(**RAG_ON)
def test_a_failure_while_cleaning_up_does_not_hide_the_original_error(self):
"""پاک‌سازیِ ناموفق نباید خطای اصلی را بپوشاند؛ صفحه باید ناتمام بماند."""
from .services.ingest import _persist_and_upsert
with patch('rag.services.ingest.ensure_collection'), \
patch('rag.services.ingest.embed_text', return_value=[0.0] * 8), \
patch('rag.services.ingest.upsert_points', side_effect=TimeoutError('boom')), \
patch('rag.services.ingest.delete_points', side_effect=RuntimeError('qdrant down')):
with self.assertRaises(TimeoutError):
_persist_and_upsert(
[{'order': 1, 'original_text': 'x', 'summary': 'خلاصه', 'source_page': 9}],
lesson_id=1, grade_id=1, subject='ریاضی', lesson_number=1,
chapter=1, title='مجموعه‌ها', collection='ریاضی نهم',
)
self.assertEqual(LessonChunk.objects.filter(collection='ریاضی نهم').count(), 0)