diff --git a/rag/services/ingest.py b/rag/services/ingest.py index a46162b..bedbabc 100644 --- a/rag/services/ingest.py +++ b/rag/services/ingest.py @@ -19,7 +19,7 @@ from .config import build_collection_name, resolve_grade_title from .embedding import embed_text from .known_collections import is_math_subject, resolve_math_chapter 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 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) ]) 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() + try: + delete_points(collection, point_ids) + except Exception: + logger.exception( + 'rag: پاک‌سازیِ %d نقطه‌ی نیمه‌نوشته در %r هم شکست خورد؛ ' + 'ممکن است نقطه‌ی بی‌صاحب بماند.', len(point_ids), collection, + ) logger.exception( 'rag: upsert به collection %r شکست خورد؛ %d ردیفِ همین دسته پس گرفته شد', collection, len(created), diff --git a/rag/services/qdrant_store.py b/rag/services/qdrant_store.py index 68bdea2..0fa1ed6 100644 --- a/rag/services/qdrant_store.py +++ b/rag/services/qdrant_store.py @@ -38,6 +38,22 @@ def upsert_points(collection: str, points: list[dict]) -> int: 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, lesson_filter_key: str | None = None, limit: int = 4) -> list[dict]: """ diff --git a/rag/tests.py b/rag/tests.py index 8a5083d..972bb67 100644 --- a/rag/tests.py +++ b/rag/tests.py @@ -860,3 +860,54 @@ class ReviewFixRegressionTests(TestCase): # هیچ ردیفی نباید بماند، وگرنه _done_pages آن صفحه را انجام‌شده می‌بیند # و chunkها برای همیشه در DB می‌مانند بدونِ اینکه به Qdrant برسند. 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)