fix(indexing): release write lock during reindex Finalizing phase

src/core/indexing/index_project_runner.py

@@ -23,6 +23,7 @@
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Callable, Dict, Optional, Set
 
@@ -168,6 +169,44 @@ def _sync_table_ref(self, new_table):
         if hasattr(self, '_db_writer') and self._db_writer is not None:
             self._db_writer.table = new_table
 
+    @contextmanager
+    def _suspend_write_lock(self):
+        """Temporarily releases the global write-lock (held by this same thread
+        via ``begin_write()``) for the duration of a ``with`` block.
+
+        KI 2026-08-28: ``_safe_ivf_index()`` (LanceDB ``optimize()`` /
+        ``create_index()``) is invoked from inside the ``begin_write()``
+        context that ``run()`` keeps for the WHOLE reindex. Running those
+        heavy operations under the global lock deadlocks the "Finalizing"
+        phase (both processes at 0% CPU, job never completes). Releasing
+        the lock around them removes the contention.
+
+        Safe because the reindex guard (``is_reindexing()``) — not this lock —
+        is what blocks concurrent search during reindex.
+        """
+        _dbm = self.db_manager
+        if _dbm is None:
+            yield
+            return
+        _wl = _dbm.begin_write()  # returns the same lock object as run() holds
+        # The outer `with begin_write()` holds this RLock exactly once in this
+        # thread. Release all nesting levels so optimize/create_index run free.
+        _released = 0
+        while True:
+            try:
+                _wl.release()
+                _released += 1
+            except RuntimeError:
+                break
+        try:
+            yield
+        finally:
+            for _ in range(_released):
+                try:
+                    _wl.acquire()
+                except RuntimeError:
+                    pass
+
     def run(
         self,
         project_path: Path,
@@ -493,22 +532,49 @@ def _safe_prune(self, prune_deleted_files: Callable, current_files_on_disk: Set[
                 return 0
         return 0
 
-    def _safe_ivf_index(self):
-        """IVF index с self-healing от Not Found."""
+    def _safe_ivf_index(self, timeout: float = 300.0) -> None:
+        """IVF index с self-healing от Not Found.
+
+        KI 2026-08-28: этот метод вызывается ИЗ-ПОД begin_write() (который
+        держит глобальный _write_lock весь reindex). Вызывающий ОБЯЗАН
+        отпустить lock вокруг этого вызова (см. _suspend_write_lock в run()),
+        иначе optimize()/create_index() deadlock'уют в фазе Finalizing.
+
+        ``create_index`` жёстко ограничен ``timeout``'ом: даже если LanceDB
+        зависнет, фаза Finalizing гарантированно завершится и job не висит
+        вечно (индекс — non-critical, поиск работает и без него).
+        """
+        # KI 2026-08-28: тяжёлые LanceDB-операции НЕ должны выполняться под
+        # глобальным _write_lock (который держит run() весь reindex) — иначе
+        # deadlock в фазе Finalizing. Отпускаем lock на время optimize/create_index.
+        with self._suspend_write_lock():
+            self._run_ivf_index_body(timeout=timeout)
+
+
+    def _run_ivf_index_body(self, timeout: float = 300.0) -> None:
         # Phase 1: optimize
         def _safe_optimize():
             for attempt in range(2):
                 try:
                     _opt_ex = ThreadPoolExecutor(max_workers=1)
                     try:
-                        _opt_ex.submit(self.table.optimize).result(timeout=300)
+                        _opt_ex.submit(self.table.optimize).result(timeout=timeout)
                     except Exception as _opt_to:
                         logger.warning(
-                            f"Table optimize exceeded 300s timeout "
-                            f"(continuing to wait for completion): {_opt_to}"
+                            f"Table optimize exceeded {timeout:.0f}s timeout "
+                            f"(continuing without blocking the job): {_opt_to}"
                         )
+                        # Не ждём вечно завершения optimize в фоне — job не должен висеть.
+                        try:
+                            _opt_ex.shutdown(wait=False)
+                        except Exception:
+                            pass
+                        return True
                     finally:
-                        _opt_ex.shutdown(wait=True)
+                        try:
+                            _opt_ex.shutdown(wait=True)
+                        except Exception:
+                            pass
                     return True
                 except Exception as e:
                     if attempt == 0 and self._reset_table_if_not_found(e, "optimize", attempt):
@@ -531,38 +597,50 @@ def _safe_optimize():
             logger.debug(f"Drop old index (non-critical): {_drop_err}")
 
         # Phase 2: create IVF_FLAT index
+        def _create_index_once(self=self) -> None:
+            try:
+                self.table.create_index(
+                    "vector",
+                    index_type="IVF_FLAT",
+                    metric="cosine",
+                    replace=True,
+                )
+                logger.info("IVF_FLAT index created")
+            except TypeError:
+                # Fallback to legacy positional API (< 0.33)
+                self.table.create_index(
+                    metric="cosine", vector_column_name="vector",
+                    index_type="IVF_FLAT", replace=True,
+                )
+                logger.info("IVF_FLAT index created (legacy API)")
+
         def _safe_create_index():
+            from concurrent.futures import TimeoutError as _FuturesTimeout
+
             for attempt in range(2):
                 try:
-                    self.table.create_index(
-                        "vector",
-                        index_type="IVF_FLAT",
-                        metric="cosine",
-                        replace=True,
-                    )
-                    logger.info("IVF_FLAT index created")
-                    return True
-                except TypeError:
-                    # Fallback to legacy positional API (< 0.33)
+                    _ci_ex = ThreadPoolExecutor(max_workers=1)
                     try:
-                        self.table.create_index(
-                            metric="cosine", vector_column_name="vector",
-                            index_type="IVF_FLAT", replace=True,
+                        _ci_ex.submit(_create_index_once).result(timeout=timeout)
+                    except _FuturesTimeout:
+                        logger.warning(
+                            f"create_index exceeded {timeout:.0f}s timeout — "
+                            f"proceeding WITHOUT vector index (non-critical)"
                         )
-                        logger.info("IVF_FLAT index created (legacy API)")
-                        return True
-                    except Exception as _legacy_e:
-                        logger.warning(f"Legacy create_index failed: {_legacy_e}")
-                        raise
+                        return
+                    finally:
+                        try:
+                            _ci_ex.shutdown(wait=False)
+                        except Exception:
+                            pass
+                    return
                 except Exception as e:
                     if attempt == 0 and self._reset_table_if_not_found(e, "create_index", attempt):
                         continue
                     logger.warning(f"create_index failed (non-critical): {e}")
-                    return False
-            return False
+                    return
 
-        if not _safe_create_index():
-            logger.warning("create_index failed after retries")
+        _safe_create_index()
 
 
 def threading_lock_context():

tests/test_reindex_finalizing_deadlock.py

@@ -0,0 +1,126 @@
+"""Regression guard for KI 2026-08-28: full reindex hangs in "Finalizing".
+
+Root cause: ``IndexProjectRunner.run()`` holds ``db_manager.begin_write()``
+(the global ``_write_lock`` RLock) for the ENTIRE job, including the heavy
+LanceDB ``optimize()`` / ``create_index()`` calls in ``_safe_ivf_index()``.
+Running those under the lock deadlocks the Finalizing phase (both processes
+at 0% CPU, job never completes, ETA grows forever).
+
+These tests pin the fix:
+  * optimize/create_index must run with the write lock RELEASED;
+  * a hanging create_index must not block the job past a bounded timeout.
+"""
+
+import threading
+import time
+from pathlib import Path
+
+from src.core.indexing.index_project_runner import IndexProjectRunner
+
+
+class _FakeDBM:
+    """Minimal db_manager: begin_write() returns a real threading.Lock."""
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self.reindex = False
+
+    def begin_write(self):
+        return self._lock
+
+    def is_reindexing(self):
+        return self.reindex
+
+
+class _LockRecorderTable:
+    """Fake LanceDB table that records whether the write lock is held
+    at the moment optimize()/create_index() are invoked."""
+
+    def __init__(self, lock):
+        self.lock = lock
+        self.optimize_called = False
+        self.create_called = False
+        self.lock_held_at_optimize = None
+        self.lock_held_at_create = None
+
+    def count_rows(self):
+        return 5000  # force the >1000 branch
+
+    def optimize(self):
+        self.optimize_called = True
+        self.lock_held_at_optimize = self.lock.locked()
+        return None
+
+    def list_indices(self):
+        return []
+
+    def create_index(self, *args, **kwargs):
+        self.create_called = True
+        self.lock_held_at_create = self.lock.locked()
+
+
+class _SlowTable:
+    """Fake table whose create_index() blocks longer than the timeout."""
+
+    def count_rows(self):
+        return 5000
+
+    def optimize(self):
+        return None
+
+    def list_indices(self):
+        return []
+
+    def create_index(self, *args, **kwargs):
+        time.sleep(10)  # simulate a hung LanceDB index build
+
+
+def _make_runner(table, dbm):
+    return IndexProjectRunner(
+        parse_file_only=lambda *a, **k: None,
+        write_file_records=lambda *a, **k: False,
+        embedder=None,
+        file_guard=None,
+        searcher=None,
+        table=table,
+        path_manager=None,
+        project_path=Path("/tmp/mscb_reindex_test"),
+        db_manager=dbm,
+        db_writer=None,
+    )
+
+
+def test_safe_ivf_index_runs_with_write_lock_released():
+    """optimize()/create_index() MUST execute with the global write lock
+    FREE (i.e. NOT held by run()'s begin_write()). Reproduces the deadlock
+    root cause: if the lock were held here, the Finalizing phase freezes."""
+    dbm = _FakeDBM()
+    table = _LockRecorderTable(dbm._lock)
+    runner = _make_runner(table, dbm)
+
+    # Simulate run() holding the lock for the whole job.
+    with dbm.begin_write():
+        runner._safe_ivf_index(timeout=5)
+
+    assert table.optimize_called, "optimize() was never called"
+    assert table.create_called, "create_index() was never called"
+    assert table.lock_held_at_optimize is False, (
+        "optimize() ran UNDER the write lock — Finalizing deadlock risk"
+    )
+    assert table.lock_held_at_create is False, (
+        "create_index() ran UNDER the write lock — Finalizing deadlock risk"
+    )
+
+
+def test_safe_ivf_index_create_index_timeout_does_not_hang():
+    """A hung create_index() must not block the job past the timeout.
+    Verifies the Finalizing phase actually completes even if LanceDB stalls."""
+    dbm = _FakeDBM()
+    table = _SlowTable()
+    runner = _make_runner(table, dbm)
+
+    t0 = time.perf_counter()
+    runner._safe_ivf_index(timeout=1)  # create_index sleeps 10s
+    dt = time.perf_counter() - t0
+
+    assert dt < 3.0, f"create_index hung the Finalizing phase for {dt:.1f}s"