Coverage for src/local_deep_research/research_library/deletion/utils/cascade_helper.py: 96%
275 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Cascade helper for deletion operations.
4Handles cleanup of related records that don't have proper FK constraints:
5- DocumentChunk (source_id has no FK constraint)
6- Local vector-index files (FAISS)
7- Server-backed vector collections (dropped before their RAGIndex row goes)
8- Filesystem files
9"""
11import time
12from pathlib import Path
13from typing import Any, Dict, Iterable, List, NamedTuple, Optional, Type, Union
15from loguru import logger
16from sqlalchemy.orm import Session
18from ....constants import FILE_PATH_SENTINELS
19from ....database.models.library import (
20 Document,
21 DocumentBlob,
22 DocumentChunk,
23 DocumentCollection,
24 RAGIndex,
25)
26from ....utilities.formatting import short_id
28# A <name>.*.tmp younger than this is assumed to be a concurrent writer's live
29# persist() temp (not yet os.replace()d into place); deleting it would crash
30# that write. Matches FaissVectorStore.persist()'s own stale-temp sweep so both
31# paths agree on when a temp is certainly crash-orphaned rather than in-flight.
32_STALE_TMP_AGE_SECONDS = 3600
35class RemoteVectorStoreDropError(RuntimeError):
36 """A server-backed vector collection could not be dropped.
38 Raised by :meth:`CascadeHelper.delete_rag_indices_for_collection` when a
39 ``RAGIndex`` row belongs to a server-backed (not local-file) store and its
40 collection could not be dropped.
42 This is deliberately an exception rather than a logged warning: it is
43 raised BEFORE any ``RAGIndex`` row is deleted, so it aborts the caller's
44 transaction and leaves ``index_hash`` -- the only handle by which the
45 remote collection can be addressed -- intact. Reporting the deletion as
46 successful here is what turns retained vectors into *unrecoverable*
47 vectors (issue #5765).
49 The message is fixed text plus the provider key, an ``index_hash`` prefix
50 and the *type name* of the underlying error -- never that error's own
51 text, which for a network client can carry a backend URI or credential.
52 For the same reason the underlying error is not chained (``from None``),
53 and callers log :meth:`log_detail` without a traceback.
55 Attributes:
56 retryable: True only when the failure is known to be transient: the
57 backend could not be reached or timed out (``ConnectionError`` /
58 ``TimeoutError``, or an error carrying ``retryable = True`` -- see
59 :meth:`BaseVectorStore.drop_collection`). False for failures a
60 retry cannot fix: no username, a row with no ``index_hash``, an
61 unknown provider, a store with no drop implementation, an
62 authentication/permission error, or anything unclassified.
63 dropped: how many remote collections this call had ALREADY dropped
64 (or found absent) before the failure. Remote drops cannot be made
65 all-or-nothing: when this is non-zero those collections are gone
66 although the caller's rollback restores their ``RAGIndex`` rows
67 (and the collection's ``indexed`` flags), so search over them
68 returns nothing until the operation is retried -- ``drop_collection``
69 is idempotent, so a retried delete completes -- or the collection
70 is force re-indexed. A normal (non-force) re-index does NOT
71 rebuild them, because the restored rows still read as indexed.
72 cause_type: type name of the underlying error, if there was one.
73 """
75 def __init__(
76 self,
77 message: str,
78 *,
79 retryable: bool = False,
80 dropped: int = 0,
81 cause_type: Optional[str] = None,
82 ) -> None:
83 super().__init__(message)
84 self.retryable = retryable
85 self.dropped = dropped
86 self.cause_type = cause_type
88 def log_detail(self) -> str:
89 """A log line for this failure built only from sanitized fields.
91 Callers log this -- at WARNING, with no traceback -- instead of the
92 exception itself: the message is fixed text plus a provider key, an
93 ``index_hash`` prefix and a type name, and nothing else about the
94 underlying error is recorded.
95 """
96 return (
97 f"{self.args[0]} (retryable={self.retryable}, "
98 f"already dropped={self.dropped})"
99 )
102def _is_transient_drop_failure(exc: BaseException) -> bool:
103 """Whether a ``drop_collection`` failure is worth retrying.
105 An explicit boolean ``retryable`` attribute on the error wins, so a
106 backend can classify its own client's exceptions. Otherwise only the
107 built-in transport failures count as transient; everything else
108 (``PermissionError``, ``NotImplementedError``, auth errors, unknown
109 exceptions) is reported as permanent rather than inviting an endless
110 retry loop.
111 """
112 flag = getattr(exc, "retryable", None)
113 if isinstance(flag, bool):
114 return flag
115 return isinstance(exc, (ConnectionError, TimeoutError))
118def remote_drop_failure_message(
119 exc: RemoteVectorStoreDropError, outcome: str
120) -> str:
121 """User-facing text for a failed remote drop.
123 ``outcome`` states what did NOT happen, e.g. ``"the collection was not
124 deleted"``. The text is built only from fixed phrases and counts, never
125 from the underlying error.
126 """
127 if exc.retryable:
128 message = (
129 f"Vector store unreachable — {outcome}. Retry once the vector "
130 "store is available."
131 )
132 else:
133 message = (
134 f"The vector store could not drop the collection — {outcome}. "
135 "Retrying will not help until the vector store configuration or "
136 "permissions are fixed; see the server log."
137 )
138 if exc.dropped:
139 message += (
140 f" {exc.dropped} vector-store collection(s) had already been "
141 "removed before the failure; their index records were kept, so "
142 "search over them returns nothing until this is retried "
143 "successfully or the collection is force re-indexed."
144 )
145 return message
148def read_vector_cleanup_settings_snapshot(settings_manager: Any) -> dict:
149 """The user's settings snapshot for a vector cleanup, failing closed.
151 The cascade resolves each row's provider -- and a server-backed store's
152 connection config -- from this snapshot (#5765). A non-strict
153 ``get_all_settings()`` answers a failed settings query with a
154 defaults-only snapshot, which would silently select the global default
155 provider and connection config: the drop would miss (or target the wrong
156 store), be counted as "already absent", and the rows -- the only handle to
157 the real collection -- would then be deleted.
159 So the read is strict. On failure:
161 * while only local-file providers exist
162 (:func:`~local_deep_research.vector_stores.config.only_local_file_providers_registered`),
163 the snapshot cannot change the cleanup (every row's files are unlinked,
164 nothing is dropped), so this degrades to the non-strict read instead of
165 failing a deletion that would otherwise succeed;
166 * otherwise it raises :class:`RemoteVectorStoreDropError` (not retryable)
167 BEFORE any row is touched, so callers abort exactly as for a failed
168 drop.
170 ``settings_manager`` is only read from. The caller chooses how it is
171 built; a caller inside a transaction must use one that does not seed and
172 commit defaults (``SettingsManager(session, seed_defaults=False)``).
173 """
174 from sqlalchemy.exc import SQLAlchemyError
176 from ....vector_stores import config as vs_config
178 try:
179 return settings_manager.get_all_settings(strict=True)
180 except (SQLAlchemyError, LookupError) as exc:
181 cause_type = type(exc).__name__
182 if vs_config.only_local_file_providers_registered():
183 logger.warning(
184 "Could not read the user's settings ({}); continuing the vector "
185 "cleanup with defaults because only local-file vector stores "
186 "exist",
187 cause_type,
188 )
189 return settings_manager.get_all_settings()
190 raise RemoteVectorStoreDropError(
191 "Could not read the user's settings "
192 f"({cause_type}), so the vector-store provider and its connection "
193 "config are unknown; refusing to delete the RAGIndex records that "
194 "hold the only handle to their vectors.",
195 retryable=False,
196 cause_type=cause_type,
197 )
200class _PlannedIndex(NamedTuple):
201 """One ``RAGIndex`` row with its vector cleanup decided up front."""
203 row: RAGIndex
204 provider: str
205 # None for a local-file store (unlink its files); the store class for a
206 # server-backed one (drop its collection).
207 remote_store: Optional[Type[Any]]
210class CascadeHelper:
211 """Helper class for cleaning up related records during deletion."""
213 @staticmethod
214 def delete_document_chunks(
215 session: Session,
216 document_id: str,
217 collection_name: Optional[str] = None,
218 ) -> int:
219 """
220 Delete DocumentChunks for a document.
222 Since DocumentChunk.source_id has no FK constraint, we must manually
223 clean up chunks when deleting a document.
225 Args:
226 session: Database session
227 document_id: The document ID to delete chunks for
228 collection_name: Optional collection name to limit deletion scope
230 Returns:
231 Number of chunks deleted
232 """
233 query = session.query(DocumentChunk).filter(
234 DocumentChunk.source_id == document_id,
235 DocumentChunk.source_type == "document",
236 )
238 if collection_name:
239 query = query.filter(
240 DocumentChunk.collection_name == collection_name
241 )
243 count = query.delete(synchronize_session=False)
244 logger.debug(
245 f"Deleted {count} chunks for document {short_id(document_id)}"
246 + (f" in collection {collection_name}" if collection_name else "")
247 )
248 return count
250 @staticmethod
251 def delete_collection_chunks(
252 session: Session,
253 collection_name: str,
254 ) -> int:
255 """
256 Delete all DocumentChunks for a collection.
258 Args:
259 session: Database session
260 collection_name: The collection name (e.g., "collection_<uuid>")
262 Returns:
263 Number of chunks deleted
264 """
265 count = (
266 session.query(DocumentChunk)
267 .filter_by(collection_name=collection_name)
268 .delete(synchronize_session=False)
269 )
270 logger.debug(f"Deleted {count} chunks for collection {collection_name}")
271 return count
273 @staticmethod
274 def get_document_blob_size(session: Session, document_id: str) -> int:
275 """
276 Get the size of a document's blob in bytes.
278 Args:
279 session: Database session
280 document_id: The document ID
282 Returns:
283 Size in bytes, or 0 if no blob exists
284 """
285 blob = (
286 session.query(DocumentBlob)
287 .filter_by(document_id=document_id)
288 .first()
289 )
290 if blob and blob.pdf_binary:
291 return len(blob.pdf_binary)
292 return 0
294 @staticmethod
295 def delete_document_blob(session: Session, document_id: str) -> int:
296 """
297 Delete a document's blob record.
299 Note: This is typically handled by CASCADE, but can be called explicitly
300 for blob-only deletion.
302 Args:
303 session: Database session
304 document_id: The document ID
306 Returns:
307 Size of deleted blob in bytes
308 """
309 blob = (
310 session.query(DocumentBlob)
311 .filter_by(document_id=document_id)
312 .first()
313 )
314 if blob:
315 size = len(blob.pdf_binary) if blob.pdf_binary else 0
316 session.delete(blob)
317 logger.debug(
318 f"Deleted blob for document {short_id(document_id)} ({size} bytes)"
319 )
320 return size
321 return 0
323 @staticmethod
324 def _resolved_containment_ok(
325 path: Path,
326 allowed_root: Optional[Union[str, Path]],
327 *,
328 allowed_root_is_resolved: bool = False,
329 ) -> bool:
330 """Return True if `path`'s fully-resolved real location stays under
331 `allowed_root`.
333 Resolution follows every symlink in the chain -- not just the
334 final component -- so this also catches an escape via a symlinked
335 *ancestor* directory (e.g. a library subfolder swapped for a symlink
336 pointing outside the library root between validation and delete).
337 It uses `resolve_allow_missing()`, never a bare non-strict
338 `resolve()`: that one ignores ENAMETOOLONG, so a symlink chain
339 whose resolved form outgrows PATH_MAX could hide an escape.
341 `allowed_root` is resolved here too, unless the caller passes
342 `allowed_root_is_resolved=True`. In that case it was already
343 resolved before the path was validated, and it is compared as
344 given. Resolving it again would let a boundary directory swapped
345 for a symlink after validation move the boundary along with the
346 file.
348 When `allowed_root` is None, no containment boundary is enforced
349 here; callers that can't cheaply supply a root still get the
350 unconditional is_symlink() check the callers of this helper perform
351 on the leaf file itself.
352 """
353 if allowed_root is None:
354 return True
355 from ....security.path_validator import resolve_allow_missing
357 try:
358 resolved_path = resolve_allow_missing(path)
359 if allowed_root_is_resolved:
360 resolved_root = Path(allowed_root)
361 else:
362 resolved_root = resolve_allow_missing(Path(allowed_root))
363 except OSError:
364 # Can't resolve (e.g. a broken component) -- refuse rather than
365 # risk unlinking something outside the intended root.
366 return False
367 return resolved_path.is_relative_to(resolved_root)
369 @staticmethod
370 def delete_filesystem_file(
371 file_path: Optional[str],
372 allowed_root: Optional[Union[str, Path]] = None,
373 *,
374 allowed_root_is_resolved: bool = False,
375 ) -> bool:
376 """
377 Delete a file from the filesystem.
379 Args:
380 file_path: Path to the file (can be relative or absolute)
381 allowed_root: If provided, the file's fully-resolved real path
382 must be contained within this directory or the delete is
383 refused. Defense-in-depth against a symlink planted at (or
384 swapped in for) `file_path` redirecting the unlink to an
385 arbitrary file elsewhere on disk -- see #5481.
386 allowed_root_is_resolved: True when `allowed_root` was resolved
387 before `file_path` was validated, as
388 `get_containment_root_from_settings` returns it. It is then
389 compared as given instead of being resolved again, so a
390 boundary directory swapped for a symlink since then cannot
391 move the boundary along with the file.
393 Returns:
394 True if file was deleted, False otherwise
395 """
396 if not file_path:
397 return False
399 # Skip special path markers
400 if file_path in FILE_PATH_SENTINELS:
401 return False
403 try:
404 path = Path(file_path)
406 # Refuse to unlink through a symlink. This check is unconditional
407 # (doesn't require allowed_root) -- is_file()/unlink() both
408 # follow symlinks, so without it a symlink planted at this exact
409 # path would redirect the delete to whatever it points at.
410 if path.is_symlink():
411 logger.warning(
412 f"Refusing to delete symlinked filesystem file: {file_path}"
413 )
414 return False
416 if not CascadeHelper._resolved_containment_ok(
417 path,
418 allowed_root,
419 allowed_root_is_resolved=allowed_root_is_resolved,
420 ):
421 logger.warning(
422 f"Refusing to delete filesystem file outside allowed root: {file_path}"
423 )
424 return False
426 if path.is_file():
427 # nosemgrep: semgrep.rules.path-traversal-risk, reason: Leaf symlinks are refused and resolved containment is checked above; document callers require an allowed root
428 path.unlink()
429 logger.debug(f"Deleted filesystem file: {file_path}")
430 return True
431 except Exception:
432 logger.exception(f"Failed to delete filesystem file: {file_path}")
433 return False
435 @staticmethod
436 def delete_faiss_index_files(
437 index_path: Optional[str],
438 allowed_root: Optional[Union[str, Path]] = None,
439 ) -> bool:
440 """
441 Delete FAISS index files.
443 FAISS stores indices as .faiss and .pkl files.
445 Args:
446 index_path: Path to the FAISS index file (without extension)
447 allowed_root: If provided, the resolved index family's parent
448 directory must be contained within this directory or the
449 delete is refused. See `delete_filesystem_file` for the
450 symlink-escape rationale.
452 Returns:
453 True if files were deleted, False otherwise
454 """
455 if not index_path:
456 return False
458 try:
459 path = Path(index_path)
461 if not CascadeHelper._resolved_containment_ok(path, allowed_root):
462 logger.warning(
463 f"Refusing to delete FAISS index outside allowed root: {index_path}"
464 )
465 return False
467 deleted_any = False
469 # FAISS index file
470 faiss_file = path.with_suffix(".faiss")
471 if faiss_file.is_symlink():
472 logger.warning(
473 f"Refusing to delete symlinked FAISS index file: {faiss_file}"
474 )
475 elif faiss_file.is_file():
476 # nosemgrep: semgrep.rules.path-traversal-risk, reason: The index base passed the resolved root check and this .faiss leaf is not a symlink
477 faiss_file.unlink()
478 logger.debug(f"Deleted FAISS index file: {faiss_file}")
479 deleted_any = True
481 # Pickle file for metadata (legacy pre-cutover format)
482 pkl_file = path.with_suffix(".pkl")
483 if pkl_file.is_symlink():
484 logger.warning(
485 f"Refusing to delete symlinked FAISS pkl file: {pkl_file}"
486 )
487 elif pkl_file.is_file():
488 # nosemgrep: semgrep.rules.path-traversal-risk, reason: The index base passed the resolved root check and this .pkl leaf is not a symlink
489 pkl_file.unlink()
490 logger.debug(f"Deleted FAISS pkl file: {pkl_file}")
491 deleted_any = True
493 # Migration sidecar (.idmap.json) — the text-free position->uuid map
494 # phase-1 writes and phase-2 consumes. If a collection is deleted
495 # during the .pkl -> .idmap.json cutover window (phase-1 ran, phase-2
496 # hasn't), the sidecar would otherwise be orphaned on disk forever.
497 idmap_file = path.with_suffix(".idmap.json")
498 if idmap_file.is_symlink():
499 logger.warning(
500 f"Refusing to delete symlinked FAISS idmap sidecar: {idmap_file}"
501 )
502 elif idmap_file.is_file():
503 # nosemgrep: semgrep.rules.path-traversal-risk, reason: The index base passed the resolved root check and this id-map leaf is not a symlink
504 idmap_file.unlink()
505 logger.debug(f"Deleted FAISS idmap sidecar: {idmap_file}")
506 deleted_any = True
508 # Sweep crash-orphaned temp files (persist writes <name>.*.tmp) and
509 # quarantined siblings (<name>.corrupt-<ns>) for the whole family,
510 # so deleting a collection doesn't leave them on disk forever — a
511 # leftover .pkl.corrupt-* is quarantined PLAINTEXT.
512 parent = path.parent
513 now = time.time()
514 for base in (faiss_file, pkl_file, idmap_file):
515 # .tmp files are AGE-GATED: FaissVectorStore.persist() writes to
516 # a <name>.*.tmp then os.replace()s it into place, so a live temp
517 # is a concurrent writer's not-yet-renamed file — unlinking it
518 # would crash that persist with FileNotFoundError. Only sweep
519 # temps old enough to be certainly crash-orphaned (mirrors
520 # persist()'s own _STALE_TMP_AGE_SECONDS sweep). A younger temp is
521 # left; a later delete/persist sweeps it once it ages out.
522 # nosemgrep: semgrep.rules.path-traversal-risk, reason: Enumerate generated temp names beside the root-checked index base
523 for stray in parent.glob(f"{base.name}.*.tmp"):
524 try:
525 if stray.is_symlink(): 525 ↛ 526line 525 didn't jump to line 526 because the condition on line 525 was never true
526 logger.warning(
527 f"Refusing to remove symlinked stray index file: {stray}"
528 )
529 continue
530 if now - stray.stat().st_mtime < _STALE_TMP_AGE_SECONDS:
531 continue
532 # nosemgrep: semgrep.rules.path-traversal-risk, reason: Only aged index temp files from the bounded parent glob reach this unlink; symlink leaves are skipped
533 stray.unlink()
534 logger.debug(f"Removed stray temp index file: {stray}")
535 deleted_any = True
536 except FileNotFoundError:
537 # Raced a concurrent writer's os.replace() — already gone.
538 continue
539 except OSError:
540 logger.warning(
541 f"Could not remove stray index file {stray}"
542 )
543 # .corrupt-* are quarantine artefacts (and .pkl.corrupt-* is
544 # PLAINTEXT) — always safe and required to remove, no age gate.
545 # nosemgrep: semgrep.rules.path-traversal-risk, reason: Enumerate generated quarantine names beside the root-checked index base
546 for stray in parent.glob(f"{base.name}.corrupt-*"):
547 try:
548 if stray.is_symlink(): 548 ↛ 549line 548 didn't jump to line 549 because the condition on line 548 was never true
549 logger.warning(
550 f"Refusing to remove symlinked stray index file: {stray}"
551 )
552 continue
553 # nosemgrep: semgrep.rules.path-traversal-risk, reason: Only files from the bounded quarantine glob reach this unlink; symlink leaves are skipped
554 stray.unlink()
555 logger.debug(f"Removed quarantined index file: {stray}")
556 deleted_any = True
557 except FileNotFoundError:
558 continue
559 except OSError:
560 logger.warning(
561 f"Could not remove stray index file {stray}"
562 )
564 return deleted_any
565 except Exception:
566 logger.exception(f"Failed to delete FAISS files for: {index_path}")
567 return False
569 @staticmethod
570 def _row_provider_key(index: RAGIndex) -> Optional[str]:
571 """The provider recorded on ``index`` itself, if the model has one.
573 ``RAGIndex`` has no provider column yet. The first server-backed
574 provider is expected to add one (the open Milvus PR, #5331, adds
575 ``vector_store_provider``), and once it exists it -- not the current
576 setting -- is what built the row's vectors: a user who switched
577 providers can hold rows of both kinds in one collection. Read
578 defensively so this works before and after that column lands.
579 """
580 raw = getattr(index, "vector_store_provider", None)
581 if isinstance(raw, str) and raw.strip():
582 return raw
583 return None
585 @staticmethod
586 def _plan_vector_cleanup(
587 indices: Iterable[RAGIndex],
588 settings_snapshot: Optional[dict],
589 ) -> List[_PlannedIndex]:
590 """Decide, ONCE per call, how each row's vectors are cleaned up.
592 A row's provider is its recorded provider when it has one, else the
593 user's configured provider resolved from ``settings_snapshot``. Doing
594 this in one pass and handing the result to both the drop and the
595 delete loop means the two can never disagree (a provider resolved
596 twice could flip between the reads, skip both the drop and the unlink,
597 and still delete the rows).
598 """
599 from ....vector_stores import config as vs_config
601 default_key: Optional[str] = None
602 store_by_key: Dict[str, Optional[Type[Any]]] = {}
603 plan: List[_PlannedIndex] = []
604 for index in indices:
605 key = CascadeHelper._row_provider_key(index)
606 if key is None:
607 if default_key is None:
608 default_key = vs_config.resolve_provider(settings_snapshot)
609 key = default_key
610 if key not in store_by_key:
611 try:
612 if vs_config.provider_is_local_file(key, settings_snapshot):
613 store_by_key[key] = None
614 else:
615 store_by_key[key] = vs_config.get_vector_store_class(
616 key, settings_snapshot
617 )
618 except ValueError:
619 # Unknown/unregistered provider: we cannot tell whether
620 # there is a remote collection, let alone drop it, so we
621 # must not delete the row that names it.
622 raise RemoteVectorStoreDropError(
623 f"RAGIndex row for collection {index.collection_name} "
624 "uses an unknown vector-store provider; cannot clean "
625 "up its vectors.",
626 retryable=False,
627 ) from None
628 plan.append(_PlannedIndex(index, key, store_by_key[key]))
629 return plan
631 @staticmethod
632 def _drop_planned_remote_collections(
633 plan: List[_PlannedIndex],
634 *,
635 username: Optional[str],
636 settings_snapshot: Optional[dict],
637 ) -> int:
638 """Drop the remote collections of the server-backed rows in ``plan``.
640 See :meth:`drop_remote_vector_collections` for the contract.
641 """
642 remote = [
643 (item, item.remote_store)
644 for item in plan
645 if item.remote_store is not None
646 ]
647 if not remote:
648 return 0
650 if not username:
651 # A server-backed collection's name is derived from the username
652 # plus index_hash; with no username we cannot name -- hence cannot
653 # delete -- the collection. Fail loudly rather than delete the rows
654 # and strand vectors nothing can address.
655 raise RemoteVectorStoreDropError(
656 f"Cannot drop {remote[0][0].provider} vector collections without "
657 "a username; refusing to delete the RAGIndex rows that hold "
658 "the only handle to them.",
659 retryable=False,
660 )
662 # Validate EVERY row's index_hash before dropping anything. This must
663 # stay a separate pass ahead of the drop loop below, not a check
664 # inside it: a bad row sorted after good ones would otherwise let the
665 # loop drop their collections first and only then abort, destroying
666 # collections whose rows survive the rollback. Failing here, before
667 # any drop_collection call, keeps a bad row's abort a true no-op.
668 for item, _store in remote:
669 if not item.row.index_hash:
670 raise RemoteVectorStoreDropError(
671 f"RAGIndex row for collection {item.row.collection_name} "
672 f"has no index_hash; cannot address its {item.provider} "
673 "collection.",
674 retryable=False,
675 )
677 dropped = 0
678 completed = 0
679 for item, store_class in remote:
680 index_hash = item.row.index_hash
681 try:
682 if store_class.drop_collection(
683 username=username,
684 index_hash=index_hash,
685 settings_snapshot=settings_snapshot,
686 ):
687 dropped += 1
688 except Exception as exc:
689 # Includes NotImplementedError: a store that declares itself
690 # server-backed but never implemented a drop path would
691 # otherwise silently orphan every collection it created.
692 #
693 # ``from None``: the cause's text can carry a backend URI or
694 # token, and a chained cause is printed by any traceback
695 # logger. Only its type name is kept.
696 raise RemoteVectorStoreDropError(
697 f"Failed to drop {item.provider} collection for index "
698 f"{short_id(index_hash, 12)}: {type(exc).__name__}. The index "
699 "record was NOT deleted.",
700 retryable=_is_transient_drop_failure(exc),
701 dropped=completed,
702 cause_type=type(exc).__name__,
703 ) from None
704 completed += 1
706 logger.info(
707 f"Dropped {dropped} vector collection(s) "
708 f"({completed - dropped} already absent) before deleting their "
709 "RAGIndex records"
710 )
711 return dropped
713 @staticmethod
714 def drop_remote_vector_collections(
715 indices: Iterable[RAGIndex],
716 *,
717 username: Optional[str],
718 settings_snapshot: Optional[dict] = None,
719 ) -> int:
720 """Drop the server-side collections backing ``indices``.
722 Rows whose provider is a local-file store are skipped -- those are
723 cleaned up by unlinking their index files instead -- so this returns 0
724 for an all-FAISS input. Each row's provider is its recorded provider
725 when the model has one, else the provider configured in
726 ``settings_snapshot`` (see :meth:`_plan_vector_cleanup`).
728 MUST be called while the ``RAGIndex`` rows still exist: the backend
729 collection name is derived from ``(username, index_hash)`` and
730 ``index_hash`` lives on the row. See #5765.
732 Remote drops are not transactional: if the Nth drop fails, the first
733 N-1 collections are already gone. The error's ``dropped`` attribute
734 says how many, so callers can say so instead of "nothing was deleted".
736 Returns:
737 The number of collections actually dropped (an already-absent
738 collection counts as 0 — ``drop_collection`` is idempotent).
740 Raises:
741 RemoteVectorStoreDropError: if a server-backed collection could
742 not be dropped, or cannot be addressed (no ``username``, a row
743 with no ``index_hash``, an unknown provider). Its
744 ``retryable`` attribute says whether a retry can help. The
745 caller must abort — and roll back — rather than delete rows.
746 """
747 # Any iterable, a generator included: the plan is a list built in ONE
748 # pass, and it -- not ``indices`` -- is what the validation pass and
749 # the drop loop both walk. (Walking a generator twice would exhaust
750 # it on the first pass and silently skip the validation guard.)
751 plan = CascadeHelper._plan_vector_cleanup(indices, settings_snapshot)
752 return CascadeHelper._drop_planned_remote_collections(
753 plan, username=username, settings_snapshot=settings_snapshot
754 )
756 @staticmethod
757 def delete_rag_indices_for_collection(
758 session: Session,
759 collection_name: str,
760 *,
761 unlink_files: bool = True,
762 username: Optional[str] = None,
763 settings_snapshot: Optional[dict] = None,
764 ) -> Dict[str, Any]:
765 """
766 Delete RAGIndex records (and their vectors) for a collection.
768 Vector cleanup is provider-aware, decided per row (see
769 :meth:`_plan_vector_cleanup`):
771 * Local-file store (FAISS): the ``.faiss``/``.pkl``/``.idmap.json``
772 files are unlinked — here when ``unlink_files`` is True, otherwise by
773 the caller after its commit (see ``index_paths`` below).
774 * Server-backed store: the remote collection is dropped **before** the
775 ``RAGIndex`` rows are deleted, because its name is derived from
776 ``index_hash`` which only exists on those rows. If a drop fails,
777 :class:`RemoteVectorStoreDropError` propagates before any row is
778 deleted and the caller must roll back. Collections dropped before
779 the failing one stay dropped (see the error's ``dropped``) — see
780 #5765.
782 The remote drops run inside the caller's transaction (they must
783 happen while the rows exist, before they are deleted in that same
784 transaction). If the caller has already written in that transaction,
785 the user's database write lock is held across those network calls, so
786 call this before the transaction's first write where possible: the
787 collection-deletion service does. Collection rebuilds
788 (``collection_rebuild.py``) run only while every registered provider
789 stores local files, so their calls make no network call. Only
790 server-backed stores make network calls here (FAISS never does), and
791 their ``drop_collection`` is expected to bound each call with a
792 client timeout.
794 Args:
795 session: Database session
796 collection_name: Name of the collection (e.g., "collection_<uuid>")
797 unlink_files: When True (default), unlink the FAISS files here.
798 When False, only the RAGIndex rows are deleted and the index
799 paths are returned in ``index_paths`` so the caller can unlink
800 them AFTER its own commit. Unlinking before the commit is
801 unsafe when more DB work follows before the commit: a later
802 rollback restores the RAGIndex rows but cannot restore the
803 already-unlinked files, leaving the collection pointing at a
804 missing index (silently-broken search). Ignored for
805 server-backed rows, which have no files.
806 username: Owner of the indices. Required when a row's provider is
807 server-backed (it is half of the derived remote collection
808 name); unused for local-file providers, which is why it stays
809 optional.
810 settings_snapshot: The user's settings snapshot. Production
811 callers MUST pass it: it selects the configured provider for
812 rows that do not record their own, and carries the connection
813 config ``drop_collection`` needs. Without it the provider
814 falls back to the global default (FAISS).
816 Returns:
817 Dict with deletion results (``index_paths`` lists local-file rows
818 only, and only when ``unlink_files`` is False).
820 Raises:
821 RemoteVectorStoreDropError: a server-backed collection could not be
822 dropped or addressed. Deliberately NOT swallowed: completing
823 the row delete anyway is what makes the leftover vectors
824 unaddressable.
825 """
826 indices = (
827 session.query(RAGIndex)
828 .filter_by(collection_name=collection_name)
829 .all()
830 )
832 # Resolved ONCE; both the drop and the delete loop below read it.
833 plan = CascadeHelper._plan_vector_cleanup(indices, settings_snapshot)
835 # ORDERING IS THE FIX (#5765): drop the remote collections FIRST,
836 # while index_hash is still readable from the rows deleted below. A
837 # failure raises, so the caller's transaction rolls back with the
838 # rows — and therefore the vectors' only name — intact.
839 #
840 # A partial failure (collection 2 of 3 unreachable) leaves the
841 # already-dropped collections gone while their rows (and the
842 # documents' indexed flags) survive the rollback; the error's
843 # ``dropped`` count lets the caller say so. That direction is
844 # recoverable — a retried delete or a force re-index reconciles it
845 # (``drop_collection`` is idempotent) — the opposite order is not.
846 CascadeHelper._drop_planned_remote_collections(
847 plan, username=username, settings_snapshot=settings_snapshot
848 )
850 deleted_indices = 0
851 deleted_files = 0
852 index_paths: List[str] = []
854 # Only needed on the unlink_files=True path (below) -- computed
855 # lazily so the unlink_files=False path (every current external
856 # caller) doesn't pay for it. Same containment root those callers
857 # pass to their own post-commit delete_faiss_index_files() call
858 # (see collection_deletion.py, collection_rebuild.py), computed here too so
859 # unlink_files=True gets the same symlinked-ancestor defense --
860 # currently unreachable (all callers pass unlink_files=False) but
861 # kept in sync to avoid a latent gap if that changes.
862 #
863 # NOTE: this is the COARSE shared rag_indices/ root, not the tighter
864 # per-user rag_indices/<sha256(user)>/ subdir. It is intentionally
865 # NOT narrowed: pre-per-user-scoping (legacy) indexes live DIRECTLY
866 # in this shared root (see
867 # library_rag_service._migrate_legacy_index_files), so a per-user
868 # allowed_root would refuse to delete a legacy-layout index and
869 # orphan its files. Kept consistent with the external callers above.
870 rag_indices_root = None
871 if unlink_files and any(item.remote_store is None for item in plan):
872 from ....config.paths import get_cache_directory
874 rag_indices_root = get_cache_directory() / "rag_indices"
876 for item in plan:
877 index = item.row
878 # index_path is only meaningful for a local-file store. For a
879 # server-backed row it names no file, so neither unlinking it here
880 # nor handing it to the caller's post-commit unlink would do
881 # anything but log a confusing miss — its vectors were already
882 # dropped above.
883 if item.remote_store is None:
884 path = str(index.index_path)
885 if unlink_files:
886 if CascadeHelper.delete_faiss_index_files(
887 path, allowed_root=rag_indices_root
888 ):
889 deleted_files += 1
890 else:
891 index_paths.append(path)
893 session.delete(index)
894 deleted_indices += 1
896 logger.debug(
897 f"Deleted {deleted_indices} RAGIndex records and {deleted_files} "
898 f"index files for collection {collection_name}"
899 )
901 return {
902 "deleted_indices": deleted_indices,
903 "deleted_files": deleted_files,
904 "index_paths": index_paths,
905 }
907 @staticmethod
908 def update_download_tracker(
909 session: Session,
910 document: Document,
911 ) -> bool:
912 """
913 Update DownloadTracker when a document is deleted.
915 The FK has SET NULL, but we also need to update is_downloaded flag.
917 Args:
918 session: Database session
919 document: The document being deleted
921 Returns:
922 True if tracker was updated
923 """
924 if not document.original_url:
925 return False
927 # URL-equivalent trackers can belong to different PDF versions.
928 # Reset only a tracker whose content hash or document link matches,
929 # retaining the legacy URL-hash fallback for the deleted document.
930 from ...utils import find_tracker_for_document
932 try:
933 tracker = find_tracker_for_document(session, document)
935 if tracker:
936 tracker.is_downloaded = False # type: ignore[assignment]
937 tracker.file_path = None # type: ignore[assignment]
938 logger.debug(
939 f"Updated DownloadTracker for document {short_id(document.id)}"
940 )
941 return True
942 except Exception:
943 logger.exception("Failed to update DownloadTracker")
944 return False
946 @staticmethod
947 def count_document_in_collections(
948 session: Session,
949 document_id: str,
950 ) -> int:
951 """
952 Count how many collections a document is in.
954 Args:
955 session: Database session
956 document_id: The document ID
958 Returns:
959 Number of collections the document is in
960 """
961 return (
962 session.query(DocumentCollection)
963 .filter_by(document_id=document_id)
964 .count()
965 )
967 @staticmethod
968 def get_document_collections(
969 session: Session,
970 document_id: str,
971 ) -> List[str]:
972 """
973 Get all collection IDs a document belongs to.
975 Args:
976 session: Database session
977 document_id: The document ID
979 Returns:
980 List of collection IDs
981 """
982 doc_collections = (
983 session.query(DocumentCollection.collection_id)
984 .filter_by(document_id=document_id)
985 .all()
986 )
987 return [dc.collection_id for dc in doc_collections]
989 @staticmethod
990 def delete_document_completely(
991 session: Session,
992 document_id: str,
993 ) -> bool:
994 """
995 Delete a document and all related records using query-based deletes.
997 This avoids ORM cascade issues where SQLAlchemy tries to set
998 DocumentBlob.document_id to NULL (which fails because it's a PK).
1000 Deletes in order:
1001 1. DocumentBlob
1002 2. DocumentCollection links
1003 3. Document itself
1005 Note: DocumentChunks should be deleted separately before calling this,
1006 as they may need collection-specific handling.
1008 Args:
1009 session: Database session
1010 document_id: The document ID to delete
1012 Returns:
1013 True if document was deleted
1014 """
1015 # Delete blob (has document_id as PK, can't be nulled by cascade)
1016 session.query(DocumentBlob).filter_by(document_id=document_id).delete(
1017 synchronize_session=False
1018 )
1020 # Delete collection links
1021 session.query(DocumentCollection).filter_by(
1022 document_id=document_id
1023 ).delete(synchronize_session=False)
1025 # Delete document itself
1026 deleted = (
1027 session.query(Document)
1028 .filter_by(id=document_id)
1029 .delete(synchronize_session=False)
1030 )
1032 if deleted:
1033 logger.debug(f"Deleted document {short_id(document_id)} completely")
1035 return deleted > 0