From 92dc2242f88f8b40713a71cbd10b723ba4beddd1 Mon Sep 17 00:00:00 2001
From: Zareian <zareian.comp@gmail.com>
Date: Tue, 15 Sep 2026 23:00:09 +0330
Subject: [PATCH] take the points back too when an upsert fails
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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

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

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

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

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

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 rag/services/ingest.py       | 15 ++++++++++-
 rag/services/qdrant_store.py | 16 +++++++++++
 rag/tests.py                 | 51 ++++++++++++++++++++++++++++++++++++
 3 files changed, 81 insertions(+), 1 deletion(-)

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)