Coverage for src/local_deep_research/database/research_cascade.py: 95%
266 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""Explicit deletion of research rows and their foreign-key descendants.
3Legacy SQLite databases retain the FK rules from table creation, and several
4current download edges use NO ACTION. Derive a child-before-parent plan from
5model metadata instead of relying on database cascades. The survival policy
6clears links to chat, library, sync and synthesis rows that outlive a research.
7Shared download trackers can move to another research's resource before the
8plan runs, preserving their URL-level attempt and duplicate history.
9Documents referenced by a surviving research likewise move to that resource;
10their storage, collection links and indexing state remain attached to the same
11document ID. Both ownership fields move together, including on legacy schemas.
13Rows without an FK path to research_history, including document_chunks, are
14outside this plan. Report files and UserActiveResearch remain caller concerns.
15All writes use a nested savepoint; the caller owns the outer transaction.
16"""
18from loguru import logger
19from sqlalchemy import (
20 not_,
21 select,
22 text,
23 tuple_,
24)
25from sqlalchemy.orm import Session
27from ..constants import ResearchStatus
28from .models import (
29 Base,
30 Document,
31 DocumentChunk,
32 DocumentCollection,
33 DownloadDuplicates,
34 DownloadTracker,
35 QueuedResearch,
36 ResearchHistory,
37 ResearchResource,
38)
39from ..utilities.download_url import get_url_hash, get_url_hash_candidates
42def collect_research_document_rag_targets(
43 db_session: Session, research_id: str
44) -> list[tuple[str, list[str]]]:
45 """Snapshot owned documents with chunk rows before the research cascade.
47 ``DocumentChunk.source_id`` has no foreign key. The cascade deletes the
48 owning Document and its collection links, but must leave chunk rows until
49 the post-commit vector purge has used their FAISS IDs. Include linked
50 collections as well as those named by chunks, since an interrupted index
51 can leave the two sets different.
52 """
53 # Legacy SQLite SELECTs do not open a database transaction. Discovery
54 # must share the cascade's snapshot, or a newly committed document can
55 # be deleted without appearing in these post-commit cleanup targets.
56 _ensure_outer_db_transaction(db_session)
57 chunk_rows = (
58 db_session.query(Document.id, DocumentChunk.collection_name)
59 .join(
60 DocumentChunk,
61 (DocumentChunk.source_id == Document.id)
62 & (DocumentChunk.source_type == "document"),
63 )
64 .filter(Document.research_id == research_id)
65 .distinct()
66 .all()
67 )
68 targets: dict[str, set[str]] = {}
69 for document_id, collection_name in chunk_rows:
70 collections = targets.setdefault(document_id, set())
71 if collection_name and collection_name.startswith("collection_"): 71 ↛ 69line 71 didn't jump to line 69 because the condition on line 71 was always true
72 collections.add(collection_name.removeprefix("collection_"))
74 if not targets:
75 return []
77 links = (
78 db_session.query(
79 DocumentCollection.document_id, DocumentCollection.collection_id
80 )
81 .join(Document, Document.id == DocumentCollection.document_id)
82 .filter(Document.research_id == research_id)
83 .all()
84 )
85 for document_id, collection_id in links:
86 if document_id in targets: 86 ↛ 85line 86 didn't jump to line 85 because the condition on line 86 was always true
87 targets[document_id].add(collection_id)
89 return [
90 (document_id, sorted(collections))
91 for document_id, collections in sorted(targets.items())
92 ]
95_CASCADE_ROOT_TABLE = "research_history"
97# (child_table, child_column) pairs whose rows SURVIVE a research
98# delete with the link cleared. Every other edge reached by the
99# metadata walk becomes a bulk DELETE. Keep this list to exactly the
100# cases where deletion would destroy data owned by someone else:
101# chat history outlives the research it discusses; library, sync and
102# synthesis rows outlive any single research.
103_CASCADE_SET_NULL = frozenset(
104 {
105 ("chat_messages", "research_id"),
106 ("documents", "resource_id"),
107 ("research_resources", "document_id"),
108 ("download_tracker", "library_document_id"),
109 ("note_syntheses", "result_document_id"),
110 ("zotero_item_map", "document_id"),
111 }
112)
114_cascade_plan_cache = None
117def _research_cascade_plan():
118 """Derive ``(set_null_steps, delete_steps, table_to_class)`` from
119 ``Base.metadata``.
121 Each step is ``(child_table, child_column, parent_table,
122 parent_column)``. Delete steps additionally carry their runtime
123 order: deepest tables first, children before the parents they
124 reference (topological over the reached subgraph). The result is
125 cached — model metadata is fixed per process.
126 """
127 global _cascade_plan_cache
128 if _cascade_plan_cache is not None:
129 return _cascade_plan_cache
131 table_to_class = {}
132 for mapper in Base.registry.mappers:
133 cls = mapper.class_
134 table_name = getattr(cls.__table__, "name", None)
135 if table_name: 135 ↛ 132line 135 didn't jump to line 132 because the condition on line 135 was always true
136 if table_name in table_to_class: 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true
137 raise AssertionError(
138 f"cascade planner: table {table_name!r} maps to two classes"
139 )
140 table_to_class[table_name] = cls
142 children_of: dict = {}
143 for table in Base.metadata.tables.values():
144 for fk in table.foreign_keys:
145 children_of.setdefault(fk.column.table.name, []).append(
146 (table.name, fk.parent.name, fk.column.name)
147 )
149 reached_edges = []
150 levels = {_CASCADE_ROOT_TABLE: 0}
151 queue = [_CASCADE_ROOT_TABLE]
152 seen_edges = set()
153 while queue:
154 parent = queue.pop(0)
155 for child, column, parent_column in sorted(children_of.get(parent, [])):
156 key = (child, column, parent, parent_column)
157 if key in seen_edges: 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true
158 continue
159 seen_edges.add(key)
160 if child not in table_to_class: 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true
161 raise AssertionError(
162 "cascade planner: table "
163 f"{child!r} references {parent!r} but has no "
164 "mapped class — map it or exclude it explicitly"
165 )
166 reached_edges.append(key)
167 if child not in levels:
168 levels[child] = levels[parent] + 1
169 queue.append(child)
171 # A table's rows are deleted only when a deleted parent references
172 # them through a non-surviving edge. Tables reached solely through
173 # ``_CASCADE_SET_NULL`` edges (chat history, syntheses, sync state)
174 # survive: edges targeting them need no step, since a row that is
175 # never deleted cannot block anything.
176 deleted_tables = {_CASCADE_ROOT_TABLE}
177 changed = True
178 while changed:
179 changed = False
180 for child, column, parent, _parent_column in reached_edges:
181 if (
182 parent in deleted_tables
183 and (child, column) not in _CASCADE_SET_NULL
184 and child not in deleted_tables
185 ):
186 deleted_tables.add(child)
187 changed = True
189 set_null_steps = [
190 key for key in reached_edges if (key[0], key[1]) in _CASCADE_SET_NULL
191 ]
192 delete_steps = [
193 key
194 for key in reached_edges
195 if (key[0], key[1]) not in _CASCADE_SET_NULL
196 and key[2] in deleted_tables
197 ]
199 # Order deletes so every child goes before the parents it
200 # references (Kahn's algorithm, table-name tie-break for
201 # determinism). The planner asserts acyclicity: a future cyclic
202 # FK must fail loudly here, not as a runtime IntegrityError.
203 delete_tables = {step[0] for step in delete_steps}
204 waiting_on = {table: set() for table in delete_tables}
205 for child, _column, parent, _parent_column in delete_steps:
206 if parent in waiting_on:
207 waiting_on[parent].add(child)
208 ordered_tables: list = []
209 ready = sorted(
210 table for table, blockers in waiting_on.items() if not blockers
211 )
212 while ready:
213 table = ready.pop(0)
214 ordered_tables.append(table)
215 for other, blockers in waiting_on.items():
216 if table in blockers:
217 blockers.discard(table)
218 if not blockers and other not in ordered_tables:
219 ready.append(other)
220 ready.sort()
221 if len(ordered_tables) != len(waiting_on): 221 ↛ 222line 221 didn't jump to line 222 because the condition on line 221 was never true
222 raise AssertionError(
223 "cascade planner: cyclic FKs among "
224 f"{sorted(waiting_on)} — cannot order deletes"
225 )
226 table_rank = {table: rank for rank, table in enumerate(ordered_tables)}
227 delete_steps.sort(key=lambda step: (table_rank[step[0]], step[0]))
229 _cascade_plan_cache = (set_null_steps, delete_steps, table_to_class)
230 return _cascade_plan_cache
233def _ensure_outer_db_transaction(db_session: Session) -> None:
234 """Start a DB-level transaction when none exists yet.
236 SQLite (and SQLCipher) in legacy transaction mode — the mode this
237 app uses on every production connection (``isolation_level=""``,
238 DEFERRED) — only opens a database transaction implicitly before
239 DML. Pure ``SELECT``s leave the driver with no transaction even
240 though the SQLAlchemy ``Session`` reports ``in_transaction() ==
241 True``. Calling ``Session.begin_nested()`` in that state issues a
242 standalone ``SAVEPOINT`` outside any ``BEGIN``; its ``RELEASE``
243 then commits immediately instead of joining the caller's
244 transaction, so a later ``db_session.rollback()`` cannot restore
245 the rows (``clear_history`` ends up partly applied after a later
246 failure). See the SQLAlchemy SQLite dialect docs on legacy
247 transaction mode with the pysqlite driver.
249 Probing the driver-level ``in_transaction`` flag and issuing a
250 plain (DEFERRED) ``BEGIN`` when it is ``False`` makes the
251 following ``begin_nested()`` a true nested savepoint: ``RELEASE``
252 keeps the outer transaction open and the caller still owns
253 ``commit`` / ``rollback``. DEFERRED (not IMMEDIATE) preserves the
254 read-concurrency property production relies on (SELECTs take only
255 a SHARED lock). If the transaction cannot be established, refuse
256 deletion rather than risk using a stale snapshot or committing early.
257 ``session_context.telemetry_savepoint`` reads the same driver flag
258 but deliberately issues no ``BEGIN``: best-effort telemetry joins a
259 transaction that is already open and otherwise commits on its own.
260 Keep the two probes in step.
261 """
262 connection = db_session.connection()
263 raw = getattr(
264 getattr(connection, "connection", None), "driver_connection", None
265 )
266 if raw is None: 266 ↛ 267line 266 didn't jump to line 267 because the condition on line 266 was never true
267 raise RuntimeError("research cascade: driver_connection unavailable")
268 in_txn = getattr(raw, "in_transaction", None)
269 if in_txn is None: 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true
270 raise RuntimeError(
271 "research cascade: driver does not expose in_transaction"
272 )
273 is_in_txn = in_txn() if callable(in_txn) else bool(in_txn)
274 if not is_in_txn:
275 db_session.execute(text("BEGIN"))
278# Bound on bind parameters per IN list; SQLite builds before 3.32 cap a
279# statement at 999.
280_IN_CHUNK = 500
283def _chunks(values):
284 values = list(values)
285 for start in range(0, len(values), _IN_CHUNK):
286 yield values[start : start + _IN_CHUNK]
289def _owned_trackers(db_session: Session, research_ids):
290 """``(id, url, url_hash)`` of trackers whose first resource belongs to
291 one of ``research_ids``."""
292 owned = []
293 for chunk in _chunks(research_ids):
294 owned.extend(
295 db_session.query(
296 DownloadTracker.id,
297 DownloadTracker.url,
298 DownloadTracker.url_hash,
299 )
300 .filter(
301 DownloadTracker.first_resource_id.in_(
302 select(ResearchResource.id).where(
303 ResearchResource.research_id.in_(chunk)
304 )
305 )
306 )
307 .all()
308 )
309 return owned
312def _tracker_references(db_session: Session, trackers):
313 """Map each tracker id to the sorted ``(resource_id, research_id)`` of
314 every existing resource that references it.
316 A resource references a tracker when its URL is canonically equivalent
317 to the tracker's URL, when its URL has the tracker's ``url_hash`` under the
318 download service's current or legacy normalization (the service reuses
319 trackers this way without recording a duplicate), or when a
320 ``DownloadDuplicates`` row links it. Stored URLs and hashes are only
321 read, never rewritten.
323 Cost: ``research_resources.url`` has no index and the normalization
324 cannot use one, so the URL match scans that table once for the whole
325 ``trackers`` set. Stream rows in bounded batches and check the same
326 hash candidates as download lookups in Python. The old SQL base filter
327 cannot express canonical URL identity (userinfo, ports, fragments and
328 tracking parameters), and could omit a surviving reference.
329 """
330 if not trackers:
331 return {}
332 references: dict = {tracker_id: set() for tracker_id, _u, _h in trackers}
333 by_hash = {url_hash: tracker_id for tracker_id, _url, url_hash in trackers}
334 by_url: dict = {}
335 by_canonical_hash: dict = {}
336 for tracker_id, url, _url_hash in trackers:
337 by_url.setdefault(url, []).append(tracker_id)
338 # A canonical resource URL cannot reconstruct a legacy hash of the
339 # original decorated URL. Normalize both URLs without re-keying rows.
340 # Several legacy trackers can represent the same canonical URL.
341 by_canonical_hash.setdefault(get_url_hash(url), []).append(tracker_id)
343 for chunk in _chunks(by_hash):
344 for url_hash, resource_id, research_id in (
345 db_session.query(
346 DownloadDuplicates.url_hash,
347 ResearchResource.id,
348 ResearchResource.research_id,
349 )
350 .join(
351 ResearchResource,
352 ResearchResource.id == DownloadDuplicates.resource_id,
353 )
354 .join(
355 ResearchHistory,
356 ResearchHistory.id == ResearchResource.research_id,
357 )
358 .filter(DownloadDuplicates.url_hash.in_(chunk))
359 ):
360 references[by_hash[url_hash]].add((resource_id, research_id))
362 url_column = ResearchResource.url
363 candidates = (
364 db_session.query(
365 ResearchResource.id, ResearchResource.research_id, url_column
366 )
367 .join(
368 ResearchHistory, ResearchHistory.id == ResearchResource.research_id
369 )
370 .filter(url_column.isnot(None))
371 )
372 for resource_id, research_id, url in candidates.yield_per(_IN_CHUNK):
373 tracker_ids = set(by_url.get(url, ()))
374 url_hashes = get_url_hash_candidates(url)
375 tracker_ids.update(by_canonical_hash.get(url_hashes[0], ()))
376 for url_hash in url_hashes:
377 tracker_id = by_hash.get(url_hash)
378 if tracker_id is not None:
379 tracker_ids.add(tracker_id)
380 for tracker_id in tracker_ids:
381 references[tracker_id].add((resource_id, research_id))
382 return {tracker_id: sorted(refs) for tracker_id, refs in references.items()}
385class SharedDownloadIndex:
386 """Shared-download references for a batch of research deletes.
388 ``clear_history`` deletes many researches in one transaction. Scanning
389 ``research_resources`` once per tracker-owning research would repeat a
390 full-table scan per research while holding the user database's write
391 lock. :meth:`build` collects, in one scan, every resource that
392 references a tracker owned by the batch. Trackers only move to
393 resources that reference them, so this covers every tracker a batch
394 member can own while the batch runs.
396 Each delete then hands a tracker to the lowest-id referencing resource
397 of a research that is neither being deleted nor already deleted
398 (:meth:`mark_deleted`). This is the choice a fresh per-research scan
399 would make, so a tracker handed to a later batch member moves again
400 when that member is deleted, and stays with it when its guarded delete
401 is refused. Researches outside the batch, including protected ones,
402 are never deleted, so they stay eligible. A tracker the index has not
403 seen (for example one created after :meth:`build`) is looked up with
404 its own scan.
405 """
407 def __init__(self):
408 self._references: dict = {}
409 self._first_live: dict = {}
410 self._deleted: set = set()
412 @classmethod
413 def build(cls, db_session: Session, research_ids) -> "SharedDownloadIndex":
414 # The snapshot must belong to the same database transaction as the
415 # deletes. Legacy SQLite SELECTs alone do not begin a transaction.
416 _ensure_outer_db_transaction(db_session)
417 index = cls()
418 owned = _owned_trackers(db_session, list(dict.fromkeys(research_ids)))
419 index._references = _tracker_references(db_session, owned)
420 return index
422 def mark_deleted(self, research_id: str) -> None:
423 self._deleted.add(research_id)
425 def replacements(self, db_session: Session, research_id: str) -> dict:
426 """Map each tracker this research owns to its surviving resource.
428 Returns early, after one tracker query, when the research owns no
429 tracker. Trackers without a surviving reference are left out and
430 deleted with the research.
431 """
432 owned = _owned_trackers(db_session, [research_id])
433 if not owned:
434 return {}
435 unseen = [row for row in owned if row[0] not in self._references]
436 if unseen:
437 self._references.update(_tracker_references(db_session, unseen))
438 replacements = {}
439 for tracker_id, _url, _url_hash in owned:
440 refs = self._references[tracker_id]
441 start = self._first_live.get(tracker_id, 0)
442 # Deleted researches never come back; skip them for good.
443 while start < len(refs) and refs[start][1] in self._deleted:
444 start += 1
445 self._first_live[tracker_id] = start
446 for resource_id, ref_research_id in refs[start:]:
447 if (
448 ref_research_id != research_id
449 and ref_research_id not in self._deleted
450 ):
451 replacements[tracker_id] = resource_id
452 break
453 return replacements
456def _shared_download_replacements(db_session: Session, research_id: str):
457 """Choose a surviving resource for each tracker owned by this research.
459 One-off form of :meth:`SharedDownloadIndex.replacements`: no scan when
460 the research owns no tracker, otherwise one scan of
461 ``research_resources``.
462 """
463 return SharedDownloadIndex().replacements(db_session, research_id)
466def _shared_document_replacements(
467 db_session: Session, research_id: str
468) -> dict[str, tuple[int, str]]:
469 """Choose a surviving explicit reference for each shared document.
471 Resource URLs may differ for identical downloaded bytes. The canonical
472 document link is the evidence of sharing, not URL equality. Choose the
473 lowest resource ID deterministically and keep resource/research ownership
474 aligned, preserving ADR-0012's protection on legacy resource-CASCADE DBs.
475 """
476 references = (
477 db_session.query(
478 Document.id, ResearchResource.id, ResearchResource.research_id
479 )
480 .join(ResearchResource, ResearchResource.document_id == Document.id)
481 .join(
482 ResearchHistory, ResearchHistory.id == ResearchResource.research_id
483 )
484 .filter(
485 Document.research_id == research_id,
486 ResearchResource.research_id != research_id,
487 )
488 .order_by(ResearchResource.id)
489 .all()
490 )
491 replacements = {}
492 for document_id, resource_id, surviving_research_id in references:
493 replacements.setdefault(
494 document_id, (resource_id, surviving_research_id)
495 )
496 return replacements
499def delete_research_with_dependents(
500 db_session: Session,
501 research_id: str,
502 *,
503 require_status_not_in_progress: bool = True,
504 require_no_claimed_queue_row: bool = True,
505 shared_downloads: SharedDownloadIndex | None = None,
506) -> bool:
507 """Delete a research and its owned FK descendants atomically.
509 Callers deleting research outside the wired entry points must also
510 collect the per-research RAG targets inside the same transaction and
511 run ``DocumentDeletionService.purge_deleted_research_documents``
512 after the guarded commit (never after a refusal or rollback) — see
513 ``web/routers/research.py`` for the pattern. Skipping it leaves
514 orphaned, still-searchable vector and chunk rows behind.
516 URL-level download trackers move to a surviving resource when one exists;
517 their shared attempt history and other resources' duplicate links survive.
518 Shared documents transfer both owner fields to a surviving resource,
519 preserving their ID, stored PDF, collections and existing index entries.
520 Deleting the last referencing research retains ordinary cascade behavior;
521 explicit global document deletion remains a separate operation.
522 A caller deleting many researches in one transaction passes one
523 ``shared_downloads`` index built for the whole set
524 (:meth:`SharedDownloadIndex.build`); the helper records each successful
525 delete in it. Without one, a research that owns a tracker pays its own
526 scan of ``research_resources``.
528 All child-table deletes and the parent delete are issued against the
529 same SQLAlchemy session without an intervening commit, so the row
530 is removed in one transaction. The two ``require_*`` flags gate the
531 parent DELETE on the same predicates the original ``#5074`` guard
532 embedded inline:
534 * ``require_status_not_in_progress`` — refuses rows still
535 ``IN_PROGRESS`` (a worker owns them).
536 * ``require_no_claimed_queue_row`` — refuses rows whose
537 ``QueuedResearch`` entry has ``is_processing = True`` (a worker is
538 about to start or has not finished writing back the row).
540 When either flag is ``False`` the corresponding predicate is dropped
541 entirely. Both default to ``True`` so the helper applies the full
542 guard by default — callers that genuinely want an unconditional
543 cascade (e.g. the chat service's per-attempt teardown, where a
544 pre-loaded liveness check has already ruled out a live worker) opt
545 out explicitly. The predicates run **inside** the parent ``DELETE``
546 so a concurrent claim that lands between the helper's read and its
547 delete is observed by SQL and the row is preserved.
549 Atomicity per research: child writes run inside a ``SAVEPOINT``
550 (``Session.begin_nested``). When the parent ``DELETE`` deletes zero
551 rows — ghost row or a guard refusal — the savepoint is rolled back
552 so the ``False`` return leaves parent *and* children untouched. A
553 ``True`` return releases the savepoint into the caller's outer
554 transaction (the caller still owns ``commit`` / ``rollback``). This
555 is what makes ``clear_history``'s multi-row loop safe: one
556 protected row's rollback does not discard its siblings' deletes.
558 The reference lookup and savepoint share a real database transaction.
559 In legacy transaction mode (production SQLCipher/SQLite with
560 ``isolation_level=""``), SELECTs do not open one, so a DEFERRED
561 ``BEGIN`` precedes Phase A (and the batch index's initial lookup).
562 This both prevents a standalone ``RELEASE`` from committing early
563 and prevents a concurrent new reference from being silently missed.
564 If a WAL writer changes the snapshot, upgrading it to a write fails;
565 the caller must roll back and retry with a fresh index.
567 The helper performs an explicit, application-level cascade driven
568 by ``_research_cascade_plan()`` (derived from ``Base.metadata``,
569 so new plain-data tables are picked up with no code change). See
570 the module docstring for why we do not rely on the DB-level
571 ``ON DELETE CASCADE``. Caller is responsible for the surrounding
572 transaction (``commit`` / ``rollback``).
574 Returns ``True`` iff the ``research_history`` row was deleted.
575 """
576 _ensure_outer_db_transaction(db_session)
577 set_null_steps, delete_steps, table_to_class = _research_cascade_plan()
579 def _pk_columns(table):
580 return [table.c[name] for name in table.primary_key.columns.keys()]
582 def _pk_filter(table, pk_values):
583 pk_cols = _pk_columns(table)
584 if len(pk_cols) == 1: 584 ↛ 586line 584 didn't jump to line 586 because the condition on line 584 was always true
585 return pk_cols[0].in_([row[0] for row in pk_values])
586 return tuple_(*pk_cols).in_(pk_values)
588 # Phase A: resolve scopes top-down. ``scopes[table]`` holds the PK
589 # tuples reachable from the named research; each step's parent
590 # values are the referenced-column values over those rows. Edges
591 # resolve in rounds (a step resolves once its parent table's scope
592 # is final), so same-level parents — e.g. ``note_references``
593 # scoped by both ``research_history`` and ``documents`` — work.
594 # Read-only SELECTs only; no writes happen before Phase B.
595 root_table = table_to_class[_CASCADE_ROOT_TABLE].__table__
596 scopes = {
597 _CASCADE_ROOT_TABLE: db_session.query(*_pk_columns(root_table))
598 .filter(root_table.c.id == research_id)
599 .all()
600 }
601 # Ghost row fast-path: no writes, no savepoint.
602 if not scopes[_CASCADE_ROOT_TABLE]:
603 return False
605 if shared_downloads is None:
606 shared_downloads = SharedDownloadIndex()
607 download_replacements = shared_downloads.replacements(
608 db_session, research_id
609 )
610 document_replacements = _shared_document_replacements(
611 db_session, research_id
612 )
613 edge_parent_values: dict = {}
615 def _resolve_step(child, column, parent, parent_column):
616 parent_table = table_to_class[parent].__table__
617 parent_pks = scopes.get(parent, [])
618 if not parent_pks:
619 edge_parent_values[(child, column, parent, parent_column)] = []
620 return []
621 values = [
622 value
623 for (value,) in db_session.query(parent_table.c[parent_column])
624 .filter(_pk_filter(parent_table, parent_pks))
625 .distinct()
626 .all()
627 if value is not None
628 ]
629 edge_parent_values[(child, column, parent, parent_column)] = values
630 return values
632 # A table's scope is final once every DELETE edge into it has
633 # resolved (SET NULL edges never extend a scope — nulled rows
634 # survive, so nothing traverses from them). DELETE edges form a
635 # DAG (asserted by the planner), and every such edge is processed
636 # parent-first, so this loop always terminates; the AssertionError
637 # below is a safety net, not an expected path.
638 all_steps = set_null_steps + delete_steps
639 delete_step_set = set(delete_steps)
640 resolved_tables = {_CASCADE_ROOT_TABLE}
641 remaining_steps = set(all_steps)
642 while remaining_steps:
643 resolvable = sorted(
644 step for step in remaining_steps if step[2] in resolved_tables
645 )
646 if not resolvable: 646 ↛ 647line 646 didn't jump to line 647 because the condition on line 646 was never true
647 raise AssertionError(
648 "cascade planner: unresolvable scope chain from "
649 f"{sorted(remaining_steps)} — cyclic FKs?"
650 )
651 for step in resolvable:
652 child, column, parent, parent_column = step
653 values = _resolve_step(child, column, parent, parent_column)
654 if step in delete_step_set and values:
655 child_table = table_to_class[child].__table__
656 children = db_session.query(*_pk_columns(child_table)).filter(
657 child_table.c[column].in_(values)
658 )
659 if child == "download_tracker" and download_replacements:
660 # These trackers will move before deleting resources.
661 # Exclude them from descendant scopes so their shared
662 # attempts and other researches' duplicates survive.
663 children = children.filter(
664 child_table.c.id.notin_(download_replacements)
665 )
666 if child == "documents" and document_replacements:
667 # Exclude retained documents before computing descendant
668 # scopes, or blobs/collection links would still be deleted
669 # even though the document's owner changes below.
670 children = children.filter(
671 child_table.c.id.notin_(document_replacements)
672 )
673 scopes.setdefault(child, []).extend(
674 tuple(row) for row in children.all()
675 )
676 remaining_steps.discard(step)
677 # Mark a table resolved once every DELETE edge into it has
678 # resolved. NB: this must scan tables, not remaining steps —
679 # a table whose edges all resolved in an earlier round is no
680 # longer in ``remaining_steps`` but still needs marking so
681 # deeper edges become resolvable.
682 done_steps = set(all_steps) - remaining_steps
683 for table in {step[0] for step in delete_steps}:
684 if table not in resolved_tables and all(
685 incoming in done_steps
686 for incoming in delete_steps
687 if incoming[0] == table
688 ):
689 resolved_tables.add(table)
691 # Phases B–D run inside a SAVEPOINT so a zero-row parent DELETE
692 # (ghost or guard refusal) rolls back the child writes for *this*
693 # research only. Without this, ``clear_history`` would commit the
694 # orphaned children of a skipped row along with its siblings.
695 # Phase A already started the outer DB transaction, so RELEASE
696 # cannot commit ahead of the caller's commit / rollback.
697 savepoint = db_session.begin_nested()
698 try:
699 for document_id, (
700 resource_id,
701 owner_id,
702 ) in document_replacements.items():
703 db_session.query(Document).filter(
704 Document.id == document_id, Document.research_id == research_id
705 ).update(
706 {"resource_id": resource_id, "research_id": owner_id},
707 synchronize_session=False,
708 )
710 # Keep URL-level tracking with a surviving research. These updates
711 # share the savepoint with the cascade, so a refused parent delete
712 # restores the original ownership along with all child rows.
713 for tracker_id, resource_id in download_replacements.items():
714 db_session.query(DownloadTracker).filter(
715 DownloadTracker.id == tracker_id
716 ).update(
717 {"first_resource_id": resource_id}, synchronize_session=False
718 )
720 # Phase B: clear the surviving links first (all such columns are
721 # nullable). Nulling before deleting means the deletes below never
722 # meet a row that still points at them, on any schema.
723 for child, column, parent, parent_column in set_null_steps:
724 values = edge_parent_values[(child, column, parent, parent_column)]
725 if not values:
726 continue
727 cls = table_to_class[child]
728 db_session.query(cls).filter(
729 getattr(cls, column).in_(values)
730 ).update({column: None}, synchronize_session=False)
732 # Phase C: bulk deletes in planner order (deepest first), each
733 # scoped to its own edge. Scopes were fully resolved in Phase A,
734 # so no reads are needed here.
735 for child, column, parent, parent_column in delete_steps:
736 values = edge_parent_values[(child, column, parent, parent_column)]
737 if not values:
738 continue
739 cls = table_to_class[child]
740 child_table = cls.__table__
741 db_session.query(cls).filter(
742 child_table.c[column].in_(values)
743 ).delete(synchronize_session=False)
745 # Phase D: finally, the parent row itself. The ``#5074``
746 # predicates are applied INSIDE this DELETE so a concurrent queue
747 # claim or status write observed by the row sees them atomically —
748 # mirroring the old ``DELETE ... WHERE id = ? AND status !=
749 # IN_PROGRESS AND ~claimed_queue_row`` shape. A zero-row return
750 # means the row did not exist or one of the guards blocked it;
751 # callers that map the latter to a 400 must check the row
752 # externally (the helper deliberately does not distinguish).
753 parent_filters = [ResearchHistory.id == research_id]
754 if require_status_not_in_progress:
755 parent_filters.append(
756 ResearchHistory.status != ResearchStatus.IN_PROGRESS
757 )
758 if require_no_claimed_queue_row:
759 claimed_queue_row = db_session.query(QueuedResearch.id).filter(
760 QueuedResearch.research_id == research_id,
761 QueuedResearch.is_processing.is_(True),
762 )
763 parent_filters.append(not_(claimed_queue_row.exists()))
765 deleted = (
766 db_session.query(ResearchHistory)
767 .filter(*parent_filters)
768 .delete(synchronize_session=False)
769 )
771 if not deleted:
772 # Guard refused (or ghost raced away): undo this research's
773 # child writes, keep the outer transaction for siblings.
774 savepoint.rollback()
775 return False
776 savepoint.commit()
777 shared_downloads.mark_deleted(research_id)
778 return True
779 except Exception:
780 try:
781 savepoint.rollback()
782 except Exception as rollback_exc:
783 logger.debug(
784 f"research cascade savepoint rollback failed: {rollback_exc}"
785 )
786 raise