Coverage for src/local_deep_research/web/queue/processor_v2.py: 93%
692 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Queue processor v2 - uses encrypted user databases instead of service.db
3Supports both direct execution and queue modes.
4"""
6import threading
7import time
8import uuid
9from dataclasses import dataclass
10from enum import Enum, auto
11from typing import Any, Dict, Optional
13from loguru import logger
14from sqlalchemy import or_, select
15from sqlalchemy.orm import Session
17from ...constants import ResearchStatus
18from ...database.encrypted_db import db_manager
19from ...database.models import (
20 QueuedResearch,
21 ResearchHistory,
22 TaskMetadata,
23 UserActiveResearch,
24)
25from ...database.queue_service import UserQueueService
26from ...database.session_context import get_user_db_session
27from ...database.session_passwords import session_password_store
28from ...exceptions import (
29 DuplicateResearchError,
30 InvalidQueuedResearchOverridesError,
31 SystemAtCapacityError,
32)
33from ...security.log_sanitizer import redact_secrets
34from ...notifications.queue_helpers import (
35 send_research_completed_notification_from_session,
36 send_research_failed_notification_from_session,
37)
38from .lifecycle_cleanup import (
39 cleanup_queued_research_state,
40 reconcile_research_queue_status,
41)
42from ..routes.research_validation import (
43 validate_research_query_length,
44 validate_search_overrides,
45)
46from ..research_state import get_user_research_start_lock
47from ..services.research_service import (
48 _parse_research_metadata,
49 clamp_user_max_concurrent,
50 run_research_process,
51 start_research_process,
52)
54# Retry configuration constants for notification database queries
55MAX_RESEARCH_LOOKUP_RETRIES = 3
56INITIAL_RESEARCH_LOOKUP_DELAY = 0.5 # seconds
57RETRY_BACKOFF_MULTIPLIER = 2
59# Give up on a queued research after this many consecutive spawn failures.
60# Each failure leaves is_processing=False so the next loop tick retries.
61SPAWN_RETRY_LIMIT = 3
64@dataclass(frozen=True, slots=True)
65class QueueSweepResult:
66 can_dispatch: bool
67 cleaned_ids: frozenset[str]
70class _DirectStartOutcome(Enum):
71 """Outcome of the direct-start attempt used by the queue handoff.
73 ``notify_research_queued`` must know whether it should return because the
74 research is running/terminal, or fall through to create queue metadata and
75 register the user for a later dispatch. A bare ``None`` return cannot make
76 that distinction and previously stranded rows on setup and global-capacity
77 failures.
78 """
80 STARTED = auto()
81 ALREADY_RUNNING = auto()
82 QUEUE_FALLBACK = auto()
83 TERMINAL_FAILURE = auto()
86class QueueProcessorV2:
87 """
88 Processes queued researches using encrypted user databases.
89 This replaces the service.db approach.
90 """
92 def __init__(self, check_interval=10):
93 """
94 Initialize the queue processor.
96 Args:
97 check_interval: How often to check for work (seconds)
98 """
99 self.check_interval = check_interval
100 self.running = False
101 self.thread = None
102 self._loop_iteration = 0
103 # Wakes the loop out of its inter-iteration wait so stop() returns
104 # in milliseconds instead of blocking for up to check_interval.
105 # The test suite stops the processor after every app test
106 # (tests/conftest.py reset_singletons), so a blocking stop() adds
107 # ~check_interval seconds of teardown to each of those tests.
108 self._stop_event = threading.Event()
110 # Per-user settings will be retrieved from each user's database
111 # when processing their queue using SettingsManager
112 logger.info(
113 "Queue processor v2 initialized - will use per-user settings from SettingsManager"
114 )
116 # Track which users we should check
117 self._users_to_check: set[tuple[str, str]] = set()
118 self._users_lock = threading.Lock()
120 # Track pending operations from background threads
121 self.pending_operations = {}
122 self._pending_operations_lock = threading.Lock()
124 # Compatibility view of the shared per-user admission locks. The
125 # canonical registry lives in research_state so fresh HTTP starts and
126 # queue replay use the SAME lock for count -> DB claim; a queue-local
127 # lock would still let those two entry points race each other.
128 #
129 # Size is bounded by distinct-usernames-ever-seen, not by session
130 # count. Entries are tiny (~40 bytes) and deliberately retain stable
131 # identity for the process lifetime; see pop_user_critical_lock().
132 self._user_critical_locks: Dict[str, threading.Lock] = {}
133 self._user_critical_locks_lock = threading.Lock()
135 # Count consecutive spawn failures per research_id. Entries are
136 # popped on success or after hitting SPAWN_RETRY_LIMIT (then the
137 # research is marked FAILED). In-memory is sufficient: a restart
138 # resets the counter and the research gets a fresh N retries,
139 # which is the desired behavior if the underlying system issue
140 # (thread pool, memory) cleared.
141 # Access is guarded by _spawn_retry_counts_lock because the
142 # increment path is a read-modify-write and the loop and direct
143 # request paths can interleave.
144 self._spawn_retry_counts: dict[str, int] = {}
145 self._spawn_retry_counts_lock = threading.Lock()
147 def _get_user_critical_lock(self, username: str) -> threading.Lock:
148 """Get (or lazily create) the per-user lock used to serialise the
149 count-active-and-start-direct critical section for a given user.
150 """
151 lock = get_user_research_start_lock(username)
152 with self._user_critical_locks_lock:
153 # Keep the compatibility/introspection view synchronized with the
154 # canonical research_state registry (tests may snapshot/restore
155 # that registry between cases).
156 self._user_critical_locks[username] = lock
157 return lock
159 def pop_user_critical_lock(self, username: str) -> None:
160 """Compatibility no-op: per-user critical locks keep stable identity.
162 A caller receives the lock from ``_get_user_critical_lock`` before it
163 acquires it. Removing even an apparently-unlocked entry therefore has
164 an unavoidable lookup-to-acquire race: another request can retain the
165 old lock while a later request creates and acquires a second one. If a
166 held lock is removed, the race is immediate and the count-to-start
167 critical section is no longer serialised.
169 Keep one tiny lock per distinct username for the process lifetime,
170 matching the deliberately persistent rekey gate in ``research_state``.
171 The method remains as a no-op so existing user-close cleanup callers do
172 not need a special-case import path.
173 """
174 del username
176 def _bump_spawn_retry_count(self, research_id: str) -> int:
177 """Atomically increment and return the spawn-retry counter for
178 ``research_id``. Extracted so tests can exercise the real
179 locked increment path instead of duplicating the lock in the
180 test worker (which would be a tautology).
181 """
182 with self._spawn_retry_counts_lock:
183 attempts = self._spawn_retry_counts.get(research_id, 0) + 1
184 self._spawn_retry_counts[research_id] = attempts
185 return attempts
187 @staticmethod
188 def _commit_with_safe_rollback(db_session: Session, context: str) -> bool:
189 """Commit ``db_session`` with best-effort rollback on failure.
191 Returns ``True`` on success, ``False`` if the commit raised.
192 The failure path logs via ``logger.exception`` and attempts a
193 rollback, itself guarded so a subsequent rollback failure is
194 logged at debug level rather than propagated.
196 Extracted because the ``try: commit / except: log + try:
197 rollback`` idiom repeats at ≥5 sites in this module; inlining
198 hides the defensive structure behind nested ``try`` blocks and
199 makes each callsite longer than the work it describes.
200 """
201 try:
202 db_session.commit()
203 return True
204 except Exception:
205 logger.exception(f"Commit failed: {context}")
206 try:
207 db_session.rollback()
208 except Exception:
209 logger.debug(
210 f"Rollback after commit failure ({context})",
211 exc_info=True,
212 )
213 return False
215 def _delete_queue_row_safely(
216 self, db_session: Session, username: str, research_id: str
217 ) -> None:
218 """Best-effort delete of the ``QueuedResearch`` row for
219 ``(username, research_id)``.
221 Rolls back any pending state first (the session may be in
222 ``PendingRollbackError`` from a failed commit inside
223 ``_start_research``), re-queries the row fresh, deletes it if
224 present, and commits via ``_commit_with_safe_rollback``.
226 Use this for ``DuplicateResearchError`` cleanup where the goal
227 is "drop the queue row regardless of session state." Do NOT
228 use it for paths that need the delete to be atomic with other
229 writes (e.g. the terminal FAILED path bundles
230 ``ResearchHistory.status = FAILED`` with the queue-row delete
231 in a single commit — that stays inline).
232 """
233 try:
234 db_session.rollback()
235 except Exception:
236 logger.debug(
237 f"Rollback before queue-row delete for {research_id}",
238 exc_info=True,
239 )
240 try:
241 fresh_queued = (
242 db_session.query(QueuedResearch)
243 .filter_by(username=username, research_id=research_id)
244 .first()
245 )
246 if fresh_queued:
247 db_session.delete(fresh_queued)
248 self._commit_with_safe_rollback(
249 db_session,
250 f"queue-row delete for research {research_id}",
251 )
252 except Exception:
253 logger.exception(
254 f"Failed to query/delete queue row for {research_id}"
255 )
256 try:
257 db_session.rollback()
258 except Exception:
259 logger.debug(
260 f"Rollback after queue-row delete failure for {research_id}",
261 exc_info=True,
262 )
264 def start(self):
265 """Start the queue processor thread."""
266 if self.running:
267 logger.warning("Queue processor already running")
268 return
270 self.running = True
271 # Re-arm the wait for restart: stop() leaves the event set, and
272 # create_app() restarts this singleton after the test-suite
273 # teardown stops it.
274 self._stop_event.clear()
275 self.thread = threading.Thread(
276 target=self._process_queue_loop, daemon=True
277 )
278 self.thread.start()
279 logger.info("Queue processor v2 started")
281 def stop(self):
282 """Stop the queue processor thread."""
283 self.running = False
284 # Wake the loop immediately rather than waiting for the next
285 # check_interval tick (otherwise shutdown blocks for up to 10s).
286 self._stop_event.set()
287 if self.thread:
288 self.thread.join(timeout=10)
289 logger.info("Queue processor v2 stopped")
291 def notify_user_activity(self, username: str, session_id: str):
292 """
293 Notify that a user has activity and their queue should be checked.
295 Args:
296 username: The username
297 session_id: The Flask session ID (for password access)
298 """
299 with self._users_lock:
300 self._users_to_check.add((username, session_id))
301 logger.debug(f"User {username} added to queue check list")
303 def notify_research_queued(self, username: str, research_id: str, **kwargs):
304 """
305 Notify that a research was queued.
306 In direct mode, this immediately starts the research if slots are available.
307 In queue mode, it adds to the queue.
309 Args:
310 username: The username
311 research_id: The research ID
312 **kwargs: Additional parameters for direct execution (query, mode, etc.)
313 """
314 # Pre-declared so the except handlers below can pass it to
315 # redact_secrets even on paths where it is never assigned.
316 password = None
317 # Check user's queue_mode setting when we have database access
318 if kwargs:
319 session_id = kwargs.get("session_id")
320 if session_id:
321 # Check if we can start it directly
322 password = session_password_store.get_session_password(
323 username, session_id
324 )
325 if password:
326 # TOCTOU guard against post-logout DB resurrection.
327 # Between the read above and open_user_database() below the
328 # user can log out: logout clears the session password store
329 # (auth/routes.py step 2) and THEN closes the DB. Re-opening
330 # here with the captured password would resurrect the user's
331 # decrypted database — and even resume a queued research —
332 # after logout. Re-read the store immediately before the
333 # open; if the credential is gone the user has logged out,
334 # so abort the direct start (``password`` becomes None and
335 # the block below is skipped). Keeping this re-check adjacent
336 # to the open shrinks the residual window to a negligible
337 # remainder — the store clear and the DB open live in
338 # different subsystems and are not jointly lockable without a
339 # fragile cross-subsystem lock. Otherwise adopt the freshest
340 # credential (never the stale captured one) for both the open
341 # and _start_research_directly. On abort we fall through to
342 # the queue-mode path below, which re-reads fresh and no-ops
343 # cleanly when the user is logged out.
344 current_password = (
345 session_password_store.get_session_password(
346 username, session_id
347 )
348 )
349 if current_password is None:
350 logger.info(
351 f"Aborting direct-start DB open for {username}: "
352 "session credential cleared since read (logout?); "
353 "leaving research queued"
354 )
355 password = current_password
356 if password:
357 # The canonical ordering is admission gate -> database.
358 # Fresh starts, logout/rekey, and library initialisation
359 # all follow that order. Opening/querying a pooled session
360 # first and then waiting for the gate can deadlock under
361 # pool exhaustion with a gate holder waiting for a
362 # connection. The inner admission section below is
363 # intentionally retained: the shared gate is reentrant and
364 # that narrower block documents the count -> claim scope.
365 admission_lock = self._get_user_critical_lock(username)
366 admission_lock.acquire()
367 try:
368 # Open database and check settings + active count
369 engine = db_manager.open_user_database(
370 username, password
371 )
372 if engine:
373 with get_user_db_session(username) as db_session:
374 # Get user's settings using SettingsManager
375 from ...settings.manager import SettingsManager
377 settings_manager = SettingsManager(db_session)
379 # Get user's queue_mode setting (env > DB > default)
380 queue_mode = settings_manager.get_setting(
381 "app.queue_mode", "direct"
382 )
384 # Get user's max concurrent setting (env > DB > default),
385 # clamped to the global semaphore ceiling so a stale or
386 # user-inflated value can't monopolize it.
387 max_concurrent = clamp_user_max_concurrent(
388 settings_manager.get_setting(
389 "app.max_concurrent_researches", 3
390 )
391 )
393 logger.debug(
394 f"User {username} settings: queue_mode={queue_mode}, "
395 f"max_concurrent={max_concurrent}"
396 )
398 # Only try direct execution if user has queue_mode="direct"
399 if queue_mode == "direct":
400 # Serialise the count→check→start critical
401 # section at the application layer. Two
402 # concurrent submissions for the same user
403 # must not both observe the same active
404 # count and both start — that would exceed
405 # max_concurrent. A per-user Python lock
406 # gives us that atomicity independent of
407 # the DB isolation level.
408 with self._get_user_critical_lock(username):
409 active_count = (
410 db_session.query(UserActiveResearch)
411 .filter_by(
412 username=username,
413 status=ResearchStatus.IN_PROGRESS,
414 )
415 .count()
416 )
418 if active_count < max_concurrent:
419 # We have slots - start directly!
420 logger.info(
421 f"Direct mode: Starting research {research_id} immediately "
422 f"(active: {active_count}/{max_concurrent})"
423 )
425 # Start the research directly
426 outcome = (
427 self._start_research_directly(
428 username,
429 research_id,
430 password,
431 **kwargs,
432 )
433 )
434 if (
435 outcome
436 is not _DirectStartOutcome.QUEUE_FALLBACK
437 ):
438 return
439 logger.info(
440 "Direct start deferred for "
441 f"{research_id}; queueing for "
442 "a later dispatch"
443 )
444 else:
445 logger.info(
446 "Direct mode: Max concurrent "
447 f"reached ({active_count}/"
448 f"{max_concurrent}), queueing "
449 f"{research_id}"
450 )
451 else:
452 logger.info(
453 f"User {username} has queue_mode={queue_mode}, "
454 f"queueing research {research_id}"
455 )
456 except Exception as e:
457 # ``password`` is in scope — drop the traceback
458 # chain and redact str(e) so the SQLCipher master
459 # password can't leak via diagnose=True frame
460 # locals (see the generic handler in
461 # _start_research_directly for the full rationale).
462 safe_msg = redact_secrets(str(e), password)
463 logger.warning(
464 f"Error in direct execution for {username}: {safe_msg}"
465 )
466 finally:
467 admission_lock.release()
469 # Fall back to queue mode (or if direct mode failed)
470 try:
471 with get_user_db_session(username) as session:
472 queue_service = UserQueueService(session)
473 queue_service.add_task_metadata(
474 task_id=research_id,
475 task_type="research",
476 priority=0,
477 )
478 logger.info(
479 f"Research {research_id} queued for user {username}"
480 )
482 # Register the user with the processing loop. The loop only
483 # dispatches users present in _users_to_check; under Flask that
484 # set was fed by a before_request hook on every authenticated
485 # request (deleted in the FastAPI migration), so without this
486 # call a queued research would never start. The loop keeps
487 # re-checking a registered user every tick until their queue
488 # drains, so this single registration also covers the
489 # at-capacity case: when a running research finishes, the next
490 # tick dispatches the queued one.
491 session_id = kwargs.get("session_id")
492 if session_id:
493 self.notify_user_activity(username, session_id)
494 else:
495 logger.warning(
496 f"Research {research_id} queued for {username} without "
497 "a session_id — the processing loop cannot pick it up "
498 "until the user logs in again"
499 )
500 except Exception as e:
501 # ``password`` may be bound above — same redaction rationale.
502 safe_msg = redact_secrets(str(e), password)
503 logger.warning(
504 f"Failed to update queue status for {username}: {safe_msg}"
505 )
507 def _start_research_directly(
508 self, username: str, research_id: str, password: str, **kwargs
509 ) -> _DirectStartOutcome:
510 """
511 Start a research directly without queueing.
513 Args:
514 username: The username
515 research_id: The research ID
516 password: The user's password
517 **kwargs: Research parameters (query, mode, settings, etc.)
518 """
519 query = kwargs.get("query")
520 mode = kwargs.get("mode")
521 settings_snapshot = kwargs.get("settings_snapshot", {})
523 # Create active research record
524 active_record_persisted = False
525 try:
526 with get_user_db_session(username) as db_session:
527 active_record = UserActiveResearch(
528 username=username,
529 research_id=research_id,
530 status=ResearchStatus.IN_PROGRESS,
531 thread_id="pending",
532 settings_snapshot=settings_snapshot,
533 )
534 db_session.add(active_record)
535 db_session.commit()
536 active_record_persisted = True
538 # Update task status if it exists
539 queue_service = UserQueueService(db_session)
540 queue_service.update_task_status(research_id, "processing")
541 except Exception as e:
542 # ``password`` is a parameter of this method — drop the
543 # traceback chain and redact str(e) (full rationale at the
544 # generic handler below).
545 safe_msg = redact_secrets(str(e), password)
546 logger.warning(
547 f"Failed to create active research record for {research_id}: {safe_msg}"
548 )
549 # The first commit lands before TaskMetadata is transitioned. If
550 # that later update failed, remove the now-stale active row before
551 # asking the caller to use the normal queue fallback.
552 if active_record_persisted:
553 try:
554 with get_user_db_session(username) as db_session:
555 active_record = (
556 db_session.query(UserActiveResearch)
557 .filter_by(
558 username=username, research_id=research_id
559 )
560 .first()
561 )
562 if active_record: 562 ↛ 564line 562 didn't jump to line 564 because the condition on line 562 was always true
563 db_session.delete(active_record)
564 research_row = (
565 db_session.query(ResearchHistory)
566 .filter_by(id=research_id)
567 .first()
568 )
569 if research_row: 569 ↛ 571line 569 didn't jump to line 571 because the condition on line 569 was always true
570 research_row.status = ResearchStatus.QUEUED
571 db_session.commit()
572 except Exception as cleanup_error:
573 safe_msg = redact_secrets(str(cleanup_error), password)
574 logger.warning(
575 "Cleanup after direct-start setup failure failed for "
576 f"{research_id}: {safe_msg}"
577 )
578 return _DirectStartOutcome.QUEUE_FALLBACK
580 # Extract parameters from kwargs
581 model_provider = kwargs.get("model_provider")
582 model = kwargs.get("model")
583 custom_endpoint = kwargs.get("custom_endpoint")
584 search_engine = kwargs.get("search_engine")
586 # Start the research process
587 try:
588 research_thread = start_research_process(
589 research_id,
590 query,
591 mode,
592 run_research_process,
593 username=username,
594 user_password=password,
595 model_provider=model_provider,
596 model=model,
597 custom_endpoint=custom_endpoint,
598 search_engine=search_engine,
599 max_results=kwargs.get("max_results"),
600 time_period=kwargs.get("time_period"),
601 iterations=kwargs.get("iterations"),
602 questions_per_iteration=kwargs.get("questions_per_iteration"),
603 strategy=kwargs.get("strategy", "source-based"),
604 settings_snapshot=settings_snapshot,
605 )
607 # Update thread ID
608 try:
609 with get_user_db_session(username) as db_session:
610 active_record = (
611 db_session.query(UserActiveResearch)
612 .filter_by(username=username, research_id=research_id)
613 .first()
614 )
615 if active_record: 615 ↛ 625line 615 didn't jump to line 625
616 active_record.thread_id = str(research_thread.ident)
617 db_session.commit()
618 except Exception as e:
619 # ``password`` is in scope — same redaction rationale.
620 safe_msg = redact_secrets(str(e), password)
621 logger.warning(
622 f"Failed to update thread ID for {research_id}: {safe_msg}"
623 )
625 logger.info(
626 f"Direct execution: Started research {research_id} for user {username} "
627 f"in thread {research_thread.ident}"
628 )
629 return _DirectStartOutcome.STARTED
631 except DuplicateResearchError:
632 # A live thread already owns this research_id. Do NOT delete
633 # the UserActiveResearch row or mark ResearchHistory FAILED —
634 # that state belongs to the live thread, and mutating it
635 # would terminate a running research from the user's
636 # perspective while it keeps executing. Same contract as the
637 # queue processor's dedicated dup branch (#3506).
638 logger.warning(
639 f"Duplicate live thread detected for {research_id} "
640 "in direct mode; leaving state intact"
641 )
642 return _DirectStartOutcome.ALREADY_RUNNING
643 except SystemAtCapacityError:
644 # System at concurrent-research capacity in the direct-execution
645 # path. Roll back the IN_PROGRESS active row and mark history
646 # back to QUEUED so the queue processor can pick it up later.
647 logger.info(
648 f"Direct execution hit capacity for {research_id}; re-queueing"
649 )
650 try:
651 with get_user_db_session(username) as db_session:
652 active_record = (
653 db_session.query(UserActiveResearch)
654 .filter_by(username=username, research_id=research_id)
655 .first()
656 )
657 if active_record: 657 ↛ 659line 657 didn't jump to line 659 because the condition on line 657 was always true
658 db_session.delete(active_record)
659 research_row = (
660 db_session.query(ResearchHistory)
661 .filter_by(id=research_id)
662 .first()
663 )
664 if research_row: 664 ↛ 666line 664 didn't jump to line 666 because the condition on line 664 was always true
665 research_row.status = ResearchStatus.QUEUED
666 db_session.commit()
667 except Exception as e:
668 # ``password`` is in scope — same redaction rationale.
669 safe_msg = redact_secrets(str(e), password)
670 logger.warning(
671 f"Cleanup after capacity reject failed for "
672 f"{research_id}; the stale UserActiveResearch row is "
673 f"recovered by reclaim_stale_user_active_research: {safe_msg}"
674 )
675 # Let notify_research_queued create TaskMetadata through its one
676 # canonical queue-fallback path and register (username, session_id)
677 # with the processing loop. Handling metadata here used to make
678 # the row look queued while silently omitting that registration.
679 return _DirectStartOutcome.QUEUE_FALLBACK
680 except Exception as e:
681 # ``password`` is in lexical scope (function parameter,
682 # passed through to ``start_research_process``). A
683 # SQLAlchemy / requests exception from anywhere in
684 # ``start_research_process`` could carry frame locals that
685 # include the SQLCipher master password (which is
686 # unrecoverable — see TRUST.md §5). Drop the traceback chain
687 # and redact str(e) defensively.
688 safe_msg = redact_secrets(str(e), password)
689 logger.warning(
690 f"Failed to start research {research_id} directly: {safe_msg}"
691 )
692 # Clean up the active record AND mark the research terminal
693 # FAILED so the user-visible state matches reality (no running
694 # thread, not IN_PROGRESS). Same contract as the queue
695 # processor's terminal-failure branch (#3481).
696 try:
697 with get_user_db_session(username) as db_session:
698 active_record = (
699 db_session.query(UserActiveResearch)
700 .filter_by(username=username, research_id=research_id)
701 .first()
702 )
703 if active_record:
704 db_session.delete(active_record)
705 research_row = (
706 db_session.query(ResearchHistory)
707 .filter_by(id=research_id)
708 .first()
709 )
710 if research_row:
711 research_row.status = ResearchStatus.FAILED
712 # This direct attempt originated from an already-persisted
713 # queue row. A terminal spawn failure must consume that row
714 # and its TaskMetadata atomically with FAILED, otherwise the
715 # non-QUEUED parent makes it permanently undispatchable
716 # while queue counters continue to report it.
717 cleanup_queued_research_state(
718 db_session,
719 [research_id],
720 include_claimed=True,
721 acting_user=username,
722 )
723 db_session.commit()
724 except Exception as e2:
725 # ``password`` is in scope — same redaction rationale.
726 safe_msg = redact_secrets(str(e2), password)
727 logger.warning(
728 f"Failed to clean up active research record for {research_id}: {safe_msg}"
729 )
730 return _DirectStartOutcome.TERMINAL_FAILURE
732 @staticmethod
733 def _drop_active_research_row(session, username: str, research_id: str):
734 """Delete the per-user concurrency-cap row for a finished research.
736 ``UserActiveResearch`` exists to count what a user currently has
737 running. Main deleted finished rows from a ``before_request`` hook
738 (``web/auth/cleanup_middleware.cleanup_completed_research``) that
739 sampled ~1% of requests, walked the user's rows, and dropped any
740 whose thread was no longer active. That hook has no successor
741 under FastAPI, and nothing replaced its *delete*: the spawn-failure
742 branches drop their own row, but neither terminal notification
743 below did, so every normally-completed research left its row
744 behind permanently.
746 The visible effect is not a wrong concurrency count —
747 ``reclaim_stale_user_active_research`` re-derives that at the
748 user's next start, so the cap self-heals — but the rows are never
749 removed, only flipped to FAILED, and only for a user who starts
750 another research. A user who finishes and stops accumulates rows
751 with no purge path anywhere in the codebase, and the ones that do
752 get reclaimed are mislabelled FAILED despite having succeeded.
754 Deleting at the terminal notification is strictly better than
755 main's sampled sweep: it is event-driven, so it cannot miss, and
756 it costs one statement on a path that already holds an open
757 session — no extra connection, no per-request overhead.
759 The caller's session context does not commit on exit, so commit
760 here rather than relying on a later write to flush this one.
761 """
762 from ...database.models import UserActiveResearch
764 try:
765 deleted = (
766 session.query(UserActiveResearch)
767 .filter_by(username=username, research_id=research_id)
768 .delete(synchronize_session=False)
769 )
770 if deleted:
771 session.commit()
772 logger.debug(
773 f"Dropped UserActiveResearch row for {research_id}"
774 )
775 except Exception:
776 # Never let bookkeeping break the completion notification the
777 # user is actually waiting on; the row stays and the next
778 # reclaim sweep will pick it up.
779 logger.exception(
780 f"Could not drop UserActiveResearch row for {research_id}"
781 )
782 # Swallowing is not enough: a failed statement leaves the
783 # session raising PendingRollbackError on everything that
784 # follows, and the caller runs
785 # ``send_research_completed_notification_from_session`` on
786 # *this* session on the very next line. Without the rollback
787 # the swallow would hand that notification a dead session --
788 # exactly what the comment above promises not to do. Main did
789 # the same (web/auth/cleanup_middleware.py, five rollbacks).
790 try:
791 session.rollback()
792 except Exception:
793 # A genuinely exhausted pool can fail the rollback too;
794 # this path is user-visible completion, so still no raise.
795 logger.warning(
796 "Could not roll back after a failed UserActiveResearch "
797 f"delete for {research_id}"
798 )
800 def notify_research_completed(
801 self, username: str, research_id: str, user_password: str | None = None
802 ):
803 """
804 Notify that a research completed.
805 Updates the user's queue status in their database.
807 Args:
808 username: The username
809 research_id: The research ID
810 user_password: User password for database access. Required for queue
811 updates and database lookups during notification sending.
812 Optional only because some callers may not have it
813 available, in which case only basic updates occur.
814 """
815 try:
816 # get_user_db_session is already imported at module level (line 19)
817 # It accepts optional password parameter and returns a context manager
818 with get_user_db_session(username, user_password) as session:
819 queue_service = UserQueueService(session)
820 queue_service.update_task_status(
821 research_id, ResearchStatus.COMPLETED
822 )
823 self._drop_active_research_row(session, username, research_id)
824 logger.info(
825 f"Research {research_id} completed for user {username}"
826 )
828 # Send notification using helper from notification module
829 send_research_completed_notification_from_session(
830 username=username,
831 research_id=research_id,
832 db_session=session,
833 )
835 except Exception as e:
836 # ``user_password`` is a parameter of this method — drop the
837 # traceback chain and redact str(e) so the SQLCipher master
838 # password can't leak via diagnose=True frame locals.
839 safe_msg = redact_secrets(str(e), user_password)
840 logger.warning(
841 f"Failed to update completion status for {username}: {safe_msg}"
842 )
844 # Auto-convert research to document in History collection.
845 # Documents only — FAISS indexing is triggered separately by the user
846 # via "Index All" on the History page.
847 from ...research_library.search.services.research_history_indexer import (
848 auto_convert_research,
849 )
851 auto_convert_research(username, research_id, db_password=user_password)
853 def notify_research_failed(
854 self,
855 username: str,
856 research_id: str,
857 error_message: str | None = None,
858 user_password: str | None = None,
859 ):
860 """
861 Notify that a research failed.
862 Updates the user's queue status in their database and sends notification.
864 Args:
865 username: The username
866 research_id: The research ID
867 error_message: Optional error message
868 user_password: User password for database access. Required for queue
869 updates and database lookups during notification sending.
870 Optional only because some callers may not have it
871 available, in which case only basic updates occur.
872 """
873 try:
874 # get_user_db_session is already imported at module level (line 19)
875 # It accepts optional password parameter and returns a context manager
876 with get_user_db_session(username, user_password) as session:
877 queue_service = UserQueueService(session)
878 queue_service.update_task_status(
879 research_id,
880 ResearchStatus.FAILED,
881 error_message=error_message,
882 )
883 self._drop_active_research_row(session, username, research_id)
884 logger.info(
885 f"Research {research_id} failed for user {username}: "
886 f"{error_message}"
887 )
889 # Send notification using helper from notification module
890 send_research_failed_notification_from_session(
891 username=username,
892 research_id=research_id,
893 error_message=error_message or "Unknown error",
894 db_session=session,
895 )
897 except Exception as e:
898 # ``user_password`` is a parameter of this method — same
899 # redaction rationale as notify_research_completed.
900 safe_msg = redact_secrets(str(e), user_password)
901 logger.warning(
902 f"Failed to update failure status for {username}: {safe_msg}"
903 )
905 def _process_queue_loop(self):
906 """Main loop that processes the queue."""
907 while self.running:
908 try:
909 # Get list of users to check (don't clear immediately)
910 with self._users_lock:
911 users_to_check = list(self._users_to_check)
913 # Process each user's queue
914 users_to_remove = []
915 for user_session in users_to_check:
916 try:
917 username, session_id = user_session
918 # _process_user_queue returns True if queue is empty
919 queue_empty = self._process_user_queue(
920 username, session_id
921 )
922 if queue_empty:
923 users_to_remove.append(user_session)
924 except Exception:
925 logger.exception(
926 f"Error processing queue for user {username}"
927 )
928 # Don't remove on error - the _process_user_queue method
929 # determines whether to keep checking based on error type
931 # Only remove users whose queues are now empty
932 with self._users_lock:
933 for user_session in users_to_remove:
934 self._users_to_check.discard(user_session)
936 # Drain pending operations (progress / error updates queued
937 # by background threads via queue_progress_update /
938 # queue_error_update). Without this, FAILED status is never
939 # persisted unless the user happens to make another HTTP
940 # request that drains it from a request handler.
941 self._drain_pending_operations()
943 except Exception:
944 logger.exception("Error in queue processor loop")
945 finally:
946 # Clean up thread-local database session after each iteration.
947 # The loop opens a new session each iteration via get_user_db_session();
948 # closing it returns the connection to the shared QueuePool promptly.
949 try:
950 from ...database.thread_local_session import (
951 cleanup_current_thread,
952 cleanup_dead_threads,
953 )
955 cleanup_current_thread()
956 except Exception:
957 logger.debug(
958 "thread-local cleanup on shutdown", exc_info=True
959 )
961 # Periodic dead-thread credential sweep (every ~60s).
962 # One of three sweep trigger points (fastapi_app's
963 # post-request cleanup, connection_cleanup scheduler,
964 # and here).
965 self._loop_iteration += 1
966 if self._loop_iteration % 6 == 0: # Every ~60s (10s × 6)
967 try:
968 cleanup_dead_threads()
969 except Exception:
970 logger.debug(
971 "periodic dead-thread sweep", exc_info=True
972 )
974 # Event.wait, not time.sleep: stop() must be able to interrupt
975 # this pause, otherwise shutdown blocks for up to
976 # check_interval seconds.
977 self._stop_event.wait(self.check_interval)
979 def _drain_pending_operations(self) -> None:
980 """Per-iteration drain of pending_operations across users.
982 Background threads (research workers) write into pending_operations
983 when they need to mark a research FAILED or update progress but
984 don't have direct DB access (e.g. password lookup failed). We
985 opportunistically drain by username here using the queue
986 processor's own DB access.
987 """
988 with self._pending_operations_lock:
989 usernames = {
990 op["username"]
991 for op in self.pending_operations.values()
992 if op.get("username")
993 }
994 if not usernames:
995 return
997 for username in usernames:
998 try:
999 with get_user_db_session(username) as db_session:
1000 n = self.process_pending_operations_for_user(
1001 username, db_session
1002 )
1003 if n: 1003 ↛ 997line 1003 didn't jump to line 997
1004 logger.info(
1005 f"Drained {n} pending operation(s) for {username}"
1006 )
1007 except Exception:
1008 # Likely no password available for this user yet — leave the
1009 # ops queued; eviction will reap them after _PENDING_OPS_TTL.
1010 logger.debug(
1011 f"Could not drain pending ops for {username} this tick",
1012 exc_info=True,
1013 )
1015 def _process_user_queue(self, username: str, session_id: str) -> bool:
1016 """
1017 Process the queue for a specific user.
1019 Args:
1020 username: The username
1021 session_id: The Flask session ID
1023 Returns:
1024 True if the queue is empty, False if there are still items
1025 """
1026 # Get the user's password from session store
1027 password = session_password_store.get_session_password(
1028 username, session_id
1029 )
1030 if not password:
1031 logger.debug(
1032 f"No password available for user {username}, skipping queue check"
1033 )
1034 return True # Remove from checking - session expired
1036 # Keep the process-wide lock order consistent with fresh starts and
1037 # logout/rekey: per-user admission gate before database/session work.
1038 # Taking a pooled session first and then blocking on the gate creates a
1039 # resource cycle when the gate holder itself needs a connection.
1040 # The acquire blocks without a timeout on purpose. Direct starts,
1041 # follow-ups and chat send/retry hold this gate until their worker is
1042 # registered (or its spawn-failure cleanup finishes), so a slow DB or
1043 # spawn there delays this loop, and with it queue dispatch and pending
1044 # progress/error writes for every user, until that holder returns. The
1045 # wait is bounded by that start. A timeout that then carried on without
1046 # the gate would let this tick reclaim a committed claim before its
1047 # worker is registered.
1048 admission_lock = self._get_user_critical_lock(username)
1049 admission_lock.acquire()
1051 # Open the user's encrypted database
1052 try:
1053 # First ensure the database is open
1054 engine = db_manager.open_user_database(username, password)
1055 if not engine:
1056 logger.error(f"Failed to open database for user {username}")
1057 return False # Keep checking - could be temporary DB issue
1059 # Get a session and process the queue
1060 with get_user_db_session(username, password) as db_session:
1061 queue_service = UserQueueService(db_session)
1063 sweep_result = self._sweep_missing_parent_queue_rows(
1064 db_session, username
1065 )
1066 if not sweep_result.can_dispatch: 1066 ↛ 1067line 1066 didn't jump to line 1067 because the condition on line 1066 was never true
1067 return False
1069 # Get user's settings using SettingsManager
1070 from ...settings.manager import SettingsManager
1072 settings_manager = SettingsManager(db_session)
1074 # Get user's max concurrent setting (env > DB > default),
1075 # clamped to the global semaphore ceiling so a stale or
1076 # user-inflated value can't monopolize it.
1077 max_concurrent = clamp_user_max_concurrent(
1078 settings_manager.get_setting(
1079 "app.max_concurrent_researches", 3
1080 )
1081 )
1083 # Serialize the live-count -> dispatch handoff with direct
1084 # starts made through this processor. QueueStatus.active_tasks
1085 # only counts tasks that previously travelled through the
1086 # queue, so it misses already-running direct researches and
1087 # cannot enforce the user's actual concurrency limit.
1088 with self._get_user_critical_lock(username):
1089 # A crashed worker can leave an IN_PROGRESS row behind.
1090 # The fresh-submit route performs this same liveness
1091 # reconciliation before counting; queue replay must do so
1092 # as well or a dead row can block the queue forever.
1093 from ..routes.globals import (
1094 reclaim_stale_user_active_research,
1095 )
1097 if reclaim_stale_user_active_research(
1098 db_session, username, logger=logger
1099 ):
1100 db_session.commit()
1102 active_count = (
1103 db_session.query(UserActiveResearch)
1104 .filter_by(
1105 username=username,
1106 status=ResearchStatus.IN_PROGRESS,
1107 )
1108 .count()
1109 )
1110 queue_status = queue_service.get_queue_status() or {
1111 "queued_tasks": 0,
1112 }
1113 available_slots = max_concurrent - active_count
1115 if available_slots <= 0:
1116 # No slots available, but queue might not be empty.
1117 return False # Keep checking
1119 if queue_status["queued_tasks"] == 0 and (
1120 db_session.query(QueuedResearch.id)
1121 .filter_by(username=username)
1122 .first()
1123 is None
1124 ):
1125 # A crashed claim still has a queue row but counts
1126 # as processing, not queued. Let dispatch reclaim it
1127 # before deciding this user has no pending work.
1128 return True # Remove from checking
1130 logger.info(
1131 f"Processing queue for {username}: "
1132 f"{active_count} active, "
1133 f"{queue_status['queued_tasks']} queued, "
1134 f"{available_slots} slots available"
1135 )
1137 # Process queued researches while retaining the same lock
1138 # used for the capacity decision. Otherwise a direct
1139 # notification could consume the last slot between the
1140 # count and the queued spawn.
1141 self._start_queued_researches(
1142 db_session,
1143 queue_service,
1144 username,
1145 password,
1146 available_slots,
1147 )
1149 # Check if there are still items in queue.
1150 updated_status = queue_service.get_queue_status() or {
1151 "queued_tasks": 0
1152 }
1153 return bool(
1154 updated_status["queued_tasks"] == 0
1155 and db_session.query(QueuedResearch.id)
1156 .filter_by(username=username)
1157 .first()
1158 is None
1159 )
1161 except Exception as e:
1162 # ``password`` (from the session store above) is in scope —
1163 # drop the traceback chain and redact str(e).
1164 safe_msg = redact_secrets(str(e), password)
1165 logger.warning(
1166 f"Error processing queue for user {username}: {safe_msg}"
1167 )
1168 return False # Keep checking - errors might be temporary
1169 finally:
1170 admission_lock.release()
1172 def _reclaim_stranded_queue_rows(
1173 self, db_session: Session, username: str
1174 ) -> int:
1175 """Reclaim queue rows stranded by a crash or restart.
1177 A row is stranded when ``is_processing=True`` but no live thread
1178 exists in ``_active_research`` for its ``research_id``. This can
1179 happen after a crash/restart between the pre-spawn IN_PROGRESS
1180 commit and the queue-row deletion in ``_start_queued_researches``
1181 — the row is invisible to the normal ``is_processing=False``
1182 query and would never be retried.
1184 Reclaims the ``ResearchHistory`` status and the
1185 ``TaskMetadata`` claim (``processing`` -> ``queued``) alongside
1186 ``is_processing``, so ``reconcile_research_queue_status`` keeps
1187 reporting the row in ``queued_tasks`` (#6147). Removes leftovers
1188 whose parent is already terminal, including previously released
1189 claims, so they cannot block later queued work. Preserves claims
1190 whose worker is still live. Returns the number of rows recovered.
1191 """
1192 from ..routes.globals import is_research_active
1194 terminal_statuses = (
1195 ResearchStatus.COMPLETED,
1196 ResearchStatus.SUSPENDED,
1197 ResearchStatus.FAILED,
1198 ResearchStatus.ERROR,
1199 )
1200 terminal_research_ids = select(ResearchHistory.id).where(
1201 ResearchHistory.status.in_(terminal_statuses)
1202 )
1203 stranded = (
1204 db_session.query(QueuedResearch)
1205 .filter_by(username=username)
1206 .filter(
1207 QueuedResearch.is_processing.is_(True)
1208 | QueuedResearch.research_id.in_(terminal_research_ids)
1209 )
1210 .all()
1211 )
1212 reclaimed = 0
1213 finished_ids: set[str] = set()
1214 for row in stranded:
1215 if is_research_active(row.research_id):
1216 # A legitimate in-flight claim; don't touch.
1217 continue
1218 research = (
1219 db_session.query(ResearchHistory)
1220 .filter_by(id=row.research_id)
1221 .first()
1222 )
1223 if research is not None and research.status in terminal_statuses:
1224 finished_ids.add(row.research_id)
1225 continue
1226 row.is_processing = False
1227 status_changed = (
1228 research is not None
1229 and research.status == ResearchStatus.IN_PROGRESS
1230 )
1231 if status_changed:
1232 research.status = ResearchStatus.QUEUED
1233 metadata = (
1234 db_session.query(TaskMetadata)
1235 .filter_by(task_id=row.research_id, task_type="research")
1236 .first()
1237 )
1238 metadata_changed = (
1239 metadata is not None and metadata.status == "processing"
1240 )
1241 if metadata_changed:
1242 # A stranded claim has no live worker, so a 'processing'
1243 # TaskMetadata row is stale by definition. Reverting it
1244 # keeps queued_tasks accurate so the dispatch loop does
1245 # not deregister the user before retrying this row
1246 # (#6147).
1247 metadata.status = "queued"
1248 metadata.started_at = None
1249 reclaimed += 1
1250 logger.warning(
1251 f"Reclaimed stranded queue row for research "
1252 f"{row.research_id} (user {username}): no live thread, "
1253 "resetting is_processing=False"
1254 + (" and status=QUEUED" if status_changed else "")
1255 + (
1256 " and TaskMetadata.status=queued"
1257 if metadata_changed
1258 else ""
1259 )
1260 )
1261 if finished_ids:
1262 cleanup_queued_research_state(
1263 db_session,
1264 finished_ids,
1265 include_claimed=True,
1266 acting_user=username,
1267 )
1268 if reclaimed or finished_ids:
1269 reconcile_research_queue_status(db_session, force=True)
1270 if not self._commit_with_safe_rollback(
1271 db_session,
1272 f"reclaim of stranded rows for user {username}",
1273 ):
1274 return 0
1275 # Discard retry bookkeeping only after the database cleanup is
1276 # durable. A rollback must leave the whole claim recoverable.
1277 with self._spawn_retry_counts_lock:
1278 for research_id in finished_ids:
1279 self._spawn_retry_counts.pop(research_id, None)
1280 return reclaimed + len(finished_ids)
1282 def _sweep_missing_parent_queue_rows(
1283 self, db_session: Session, username: str
1284 ) -> QueueSweepResult:
1285 queued_orphan_ids = {
1286 research_id
1287 for (research_id,) in (
1288 db_session.query(QueuedResearch.research_id)
1289 .outerjoin(
1290 ResearchHistory,
1291 QueuedResearch.research_id == ResearchHistory.id,
1292 )
1293 .filter(
1294 QueuedResearch.username == username,
1295 ResearchHistory.id.is_(None),
1296 )
1297 .all()
1298 )
1299 }
1300 # TaskMetadata has no owner column. A row that another user's queue
1301 # entry points at is theirs, so it is left out here, and
1302 # cleanup_queued_research_state leaves it too when given the acting
1303 # user. The queue row stands in for the owner only because
1304 # queued_researches.research_id is UNIQUE; a test in
1305 # test_queued_research_lifecycle_cleanup.py pins that.
1306 metadata_orphan_ids = {
1307 task_id
1308 for (task_id,) in (
1309 db_session.query(TaskMetadata.task_id)
1310 .outerjoin(
1311 ResearchHistory,
1312 TaskMetadata.task_id == ResearchHistory.id,
1313 )
1314 .outerjoin(
1315 QueuedResearch,
1316 TaskMetadata.task_id == QueuedResearch.research_id,
1317 )
1318 .filter(
1319 TaskMetadata.task_type == "research",
1320 ResearchHistory.id.is_(None),
1321 or_(
1322 QueuedResearch.research_id.is_(None),
1323 QueuedResearch.username == username,
1324 ),
1325 )
1326 .all()
1327 )
1328 }
1329 orphaned_research_ids = queued_orphan_ids | metadata_orphan_ids
1330 if not orphaned_research_ids:
1331 if reconcile_research_queue_status(db_session):
1332 if not self._commit_with_safe_rollback( 1332 ↛ 1336line 1332 didn't jump to line 1336 because the condition on line 1332 was never true
1333 db_session,
1334 f"queue status reconciliation for user {username}",
1335 ):
1336 return QueueSweepResult(False, frozenset())
1337 return QueueSweepResult(True, frozenset())
1339 cleanup_result = cleanup_queued_research_state(
1340 db_session,
1341 orphaned_research_ids,
1342 include_claimed=True,
1343 acting_user=username,
1344 )
1345 if not self._commit_with_safe_rollback(
1346 db_session,
1347 f"missing-parent queue sweep for user {username}",
1348 ):
1349 return QueueSweepResult(False, frozenset())
1351 with self._spawn_retry_counts_lock:
1352 for research_id in cleanup_result.cleaned_ids:
1353 self._spawn_retry_counts.pop(research_id, None)
1354 return QueueSweepResult(True, cleanup_result.cleaned_ids)
1356 def _start_queued_researches(
1357 self,
1358 db_session: Session,
1359 queue_service: UserQueueService,
1360 username: str,
1361 password: str,
1362 available_slots: int,
1363 ):
1364 """Start queued researches up to available slots."""
1365 sweep_result = self._sweep_missing_parent_queue_rows(
1366 db_session, username
1367 )
1368 if not sweep_result.can_dispatch:
1369 return
1371 # Before picking work, reclaim any rows stranded by a prior
1372 # crash — otherwise they are invisible to the is_processing=False
1373 # filter below and would never retry.
1374 self._reclaim_stranded_queue_rows(db_session, username)
1376 # Get queued researches
1377 queued = (
1378 db_session.query(QueuedResearch)
1379 .filter_by(username=username, is_processing=False)
1380 .order_by(QueuedResearch.position, QueuedResearch.id)
1381 .limit(available_slots)
1382 .all()
1383 )
1385 for queued_research in queued:
1386 research_id = queued_research.research_id
1387 try:
1388 # Atomically claim this item by flipping is_processing from
1389 # False to True in a single UPDATE. If another worker has
1390 # already claimed it since our SELECT above, the UPDATE will
1391 # match zero rows and we skip. Under non-IMMEDIATE isolation
1392 # the previous SELECT+assign pattern would race and two
1393 # workers could both process the same queued item.
1394 claimed = (
1395 db_session.query(QueuedResearch)
1396 .filter(
1397 QueuedResearch.id == queued_research.id,
1398 QueuedResearch.is_processing.is_(False),
1399 db_session.query(ResearchHistory.id)
1400 .filter(
1401 ResearchHistory.id == research_id,
1402 ResearchHistory.status == ResearchStatus.QUEUED,
1403 )
1404 .exists(),
1405 )
1406 .update(
1407 {QueuedResearch.is_processing: True},
1408 synchronize_session=False,
1409 )
1410 )
1411 db_session.commit()
1412 if not claimed:
1413 logger.debug(
1414 f"Queued research {research_id} "
1415 f"already claimed by another worker; skipping"
1416 )
1417 continue
1418 # Refresh local object state now that we hold the claim
1419 db_session.refresh(queued_research)
1421 # Update task status
1422 queue_service.update_task_status(research_id, "processing")
1424 # Start the research
1425 self._start_research(
1426 db_session,
1427 username,
1428 password,
1429 queued_research,
1430 )
1432 # Success — clear any prior spawn-failure count and
1433 # remove the queue row.
1434 with self._spawn_retry_counts_lock:
1435 self._spawn_retry_counts.pop(research_id, None)
1436 db_session.delete(queued_research)
1437 db_session.commit()
1439 logger.info(
1440 f"Started queued research {research_id} for user {username}"
1441 )
1443 except DuplicateResearchError:
1444 # Raised by _start_research when a prior attempt's thread
1445 # is still live, OR when the ResearchHistory row is in a
1446 # non-QUEUED state (IN_PROGRESS from a prior attempt's
1447 # successful pre-spawn commit; terminal COMPLETED /
1448 # FAILED / SUSPENDED from a thread that already finished
1449 # and cleaned up). In every case the correct behavior is
1450 # the same: clear the stale queue row and the retry
1451 # counter, and do NOT fall through to the FAILED/notify
1452 # path — that would terminate-status a live thread or
1453 # emit a false failure for a completed one.
1454 logger.warning(
1455 f"Research {research_id} is already started "
1456 "(live thread or non-QUEUED status); clearing stale "
1457 "queue row"
1458 )
1459 with self._spawn_retry_counts_lock:
1460 self._spawn_retry_counts.pop(research_id, None)
1461 self._delete_queue_row_safely(db_session, username, research_id)
1462 continue
1464 except SystemAtCapacityError:
1465 # System hit the global concurrent-research capacity while
1466 # dispatching this queued item. _start_research already
1467 # reset the ResearchHistory row back to QUEUED before
1468 # re-raising. This is a transient condition, NOT a spawn
1469 # failure, so it must NOT count toward SPAWN_RETRY_LIMIT —
1470 # otherwise a busy system would wrongly mark a perfectly
1471 # valid queued research FAILED after a few ticks. Just
1472 # release the processing claim so the next tick retries.
1473 # Mirrors the dedicated handler in _start_research_directly.
1474 logger.info(
1475 f"System at capacity dispatching queued research "
1476 f"{research_id}; leaving queued for next tick"
1477 )
1478 # Revert the queued->processing claim from
1479 # update_task_status("processing") above. The research stays
1480 # queued for the next tick, so its slot must return to
1481 # queued_tasks rather than leaking into active_tasks on
1482 # every capacity-rejected retry.
1483 queue_service.update_task_status(research_id, "queued")
1484 fresh_queued = (
1485 db_session.query(QueuedResearch)
1486 .filter_by(username=username, research_id=research_id)
1487 .first()
1488 )
1489 if fresh_queued: 1489 ↛ 1496line 1489 didn't jump to line 1496 because the condition on line 1489 was always true
1490 fresh_queued.is_processing = False
1491 self._commit_with_safe_rollback(
1492 db_session,
1493 "is_processing reset after capacity reject for "
1494 f"research {research_id}",
1495 )
1496 continue
1498 except InvalidQueuedResearchOverridesError as error:
1499 logger.warning(
1500 "Invalid persisted search overrides for queued research "
1501 f"{research_id}; marking FAILED"
1502 )
1503 with self._spawn_retry_counts_lock:
1504 self._spawn_retry_counts.pop(research_id, None)
1506 research = (
1507 db_session.query(ResearchHistory)
1508 .filter_by(id=research_id)
1509 .first()
1510 )
1511 fresh_queued = (
1512 db_session.query(QueuedResearch)
1513 .filter_by(username=username, research_id=research_id)
1514 .first()
1515 )
1516 if research: 1516 ↛ 1518line 1516 didn't jump to line 1518 because the condition on line 1516 was always true
1517 research.status = ResearchStatus.FAILED
1518 if fresh_queued: 1518 ↛ 1520line 1518 didn't jump to line 1520 because the condition on line 1518 was always true
1519 db_session.delete(fresh_queued)
1520 self._commit_with_safe_rollback(
1521 db_session,
1522 "terminal invalid persisted overrides for research "
1523 f"{research_id}",
1524 )
1525 self.notify_research_failed(
1526 username=username,
1527 research_id=research_id,
1528 error_message=str(error),
1529 user_password=password,
1530 )
1531 continue
1533 except Exception as e:
1534 # ``password`` is a parameter of this method — drop the
1535 # traceback chain and redact str(e).
1536 safe_msg = redact_secrets(str(e), password)
1537 logger.warning(
1538 f"Error starting queued research {research_id}: {safe_msg}"
1539 )
1540 # Session may be in PendingRollbackError state after a
1541 # failed commit inside _start_research.
1542 try:
1543 db_session.rollback()
1544 except Exception as rb_err:
1545 # No exc_info: ``password`` is in this frame and a
1546 # rendered traceback could expose it via
1547 # diagnose=True frame locals.
1548 logger.debug(
1549 "Rollback after start failure: "
1550 f"{redact_secrets(str(rb_err), password)}"
1551 )
1553 attempts = self._bump_spawn_retry_count(research_id)
1555 # Re-query in case rollback expired the ORM object.
1556 fresh_queued = (
1557 db_session.query(QueuedResearch)
1558 .filter_by(username=username, research_id=research_id)
1559 .first()
1560 )
1562 if attempts < SPAWN_RETRY_LIMIT:
1563 # Transient failure — allow the next loop tick to
1564 # retry. _start_research rolls back its own
1565 # IN_PROGRESS write on spawn failure, so the fix-ups
1566 # needed here are resetting is_processing and
1567 # reverting the TaskMetadata claim from
1568 # update_task_status("processing") above. Without
1569 # the revert, reconcile_research_queue_status
1570 # under-reports queued_tasks and
1571 # _process_user_queue deregisters the user before
1572 # this row is retried (#6147).
1573 logger.warning(
1574 f"Spawn failed for research {research_id} "
1575 f"(attempt {attempts}/{SPAWN_RETRY_LIMIT}), "
1576 "leaving queued for retry"
1577 )
1578 # Mirror the SystemAtCapacityError branch: revert
1579 # the queued->processing claim so the slot returns
1580 # to queued_tasks instead of leaking into
1581 # active_tasks on every failed attempt.
1582 queue_service.update_task_status(research_id, "queued")
1583 if fresh_queued: 1583 ↛ 1590line 1583 didn't jump to line 1590 because the condition on line 1583 was always true
1584 fresh_queued.is_processing = False
1585 reconcile_research_queue_status(db_session, force=True)
1586 self._commit_with_safe_rollback(
1587 db_session,
1588 f"is_processing reset for research {research_id}",
1589 )
1590 continue
1592 # Exhausted retries — mark terminal FAILED, delete the
1593 # queue row to stop re-dispatch, and notify the user.
1594 # The spawn failure was already logged (redacted) at the
1595 # top of this except block; no need to repeat it here.
1596 logger.warning(
1597 f"Spawn failed for research {research_id} "
1598 f"after {attempts} attempts; marking FAILED"
1599 )
1600 with self._spawn_retry_counts_lock:
1601 self._spawn_retry_counts.pop(research_id, None)
1602 try:
1603 research = (
1604 db_session.query(ResearchHistory)
1605 .filter_by(id=research_id)
1606 .first()
1607 )
1608 if research: 1608 ↛ 1610line 1608 didn't jump to line 1610 because the condition on line 1608 was always true
1609 research.status = ResearchStatus.FAILED
1610 if fresh_queued: 1610 ↛ 1612line 1610 didn't jump to line 1612 because the condition on line 1610 was always true
1611 db_session.delete(fresh_queued)
1612 db_session.commit()
1613 except Exception as e2:
1614 # ``password`` is in scope — same redaction rationale.
1615 safe_msg = redact_secrets(str(e2), password)
1616 logger.warning(
1617 "Failed to persist terminal FAILED state for "
1618 f"research {research_id}: {safe_msg}"
1619 )
1620 try:
1621 db_session.rollback()
1622 except Exception as rb_err:
1623 # No exc_info: same frame-locals rationale as the
1624 # rollback handler above.
1625 logger.debug(
1626 "Rollback after terminal update failure: "
1627 f"{redact_secrets(str(rb_err), password)}"
1628 )
1630 # notify_research_failed opens its own session and
1631 # sends the user notification. Called exactly once
1632 # per research_id because the counter is popped above.
1633 self.notify_research_failed(
1634 username=username,
1635 research_id=research_id,
1636 error_message=(
1637 f"Failed to start research after {attempts} attempts"
1638 ),
1639 user_password=password,
1640 )
1642 def _start_research(
1643 self,
1644 db_session: Session,
1645 username: str,
1646 password: str,
1647 queued_research,
1648 ):
1649 """Start a queued research.
1651 Commits ``ResearchHistory.status = IN_PROGRESS`` BEFORE spawning
1652 the thread. If we did this after, a fast-completing thread
1653 (which opens its own DB session) could write ``COMPLETED`` and
1654 then our post-spawn commit would overwrite that with
1655 ``IN_PROGRESS``, stranding the research as stuck IN_PROGRESS
1656 after it had already finished.
1658 If ``start_research_process`` raises, reset status back to
1659 ``QUEUED`` and re-raise so the caller's 3-strike retry logic
1660 handles it. ``DuplicateResearchError`` is re-raised as-is
1661 because a thread is already running for this research; mutating
1662 status further would be wrong.
1663 """
1664 research_id = queued_research.research_id
1665 research = (
1666 db_session.query(ResearchHistory).filter_by(id=research_id).first()
1667 )
1669 if not research:
1670 raise ValueError(f"Research {research_id} not found")
1672 # Guard against re-entering _start_research on a retry when a
1673 # prior attempt's post-spawn UserActiveResearch commit failed:
1674 # - IN_PROGRESS means the prior thread is (or was) running.
1675 # - COMPLETED/FAILED means the prior thread already finished
1676 # and cleaned itself up out of _active_research, so a bare
1677 # retry would both overwrite the terminal status with
1678 # IN_PROGRESS and then spawn a *second* thread (because
1679 # check_and_start_research sees no live entry), re-running
1680 # the whole research.
1681 # In all three cases the correct behavior is the same: raise
1682 # DuplicateResearchError so the caller's existing dup branch
1683 # deletes the queue row without mutating status or notifying.
1684 if research.status != ResearchStatus.QUEUED:
1685 raise DuplicateResearchError(
1686 f"Research {research_id} is already started "
1687 f"(status={research.status})"
1688 )
1690 # Extract settings
1691 settings_snapshot = queued_research.settings_snapshot or {}
1693 # Handle new vs legacy structure
1694 if (
1695 isinstance(settings_snapshot, dict)
1696 and "submission" in settings_snapshot
1697 ):
1698 submission_params = settings_snapshot.get("submission", {})
1699 complete_settings = settings_snapshot.get("settings_snapshot", {})
1700 allowed_override_keys = (
1701 "model_provider",
1702 "model",
1703 "custom_endpoint",
1704 "search_engine",
1705 "max_results",
1706 "time_period",
1707 "iterations",
1708 "questions_per_iteration",
1709 "strategy",
1710 )
1711 explicit_override_keys = settings_snapshot.get(
1712 "submission_overrides"
1713 )
1714 override_keys = (
1715 explicit_override_keys
1716 if isinstance(explicit_override_keys, list)
1717 else allowed_override_keys
1718 )
1719 runtime_overrides = {
1720 key: submission_params[key]
1721 for key in allowed_override_keys
1722 if key in override_keys and key in submission_params
1723 }
1724 else:
1725 submission_params = settings_snapshot
1726 # A legacy-flat queued row (enqueued by an older version, pre-
1727 # "submission" wrapper) carries no settings_snapshot. Seed the run's
1728 # primary engine from the submitted search_engine so the worker's
1729 # egress build (resolve_run_primary_engine) doesn't fail closed on
1730 # an empty snapshot and refuse the run.
1731 _legacy_engine = submission_params.get("search_engine")
1732 complete_settings = (
1733 {"search.tool": _legacy_engine} if _legacy_engine else {}
1734 )
1735 runtime_overrides = {
1736 "model_provider": submission_params.get("model_provider"),
1737 "model": submission_params.get("model"),
1738 "custom_endpoint": submission_params.get("custom_endpoint"),
1739 "search_engine": submission_params.get("search_engine"),
1740 "max_results": submission_params.get("max_results"),
1741 "time_period": submission_params.get("time_period"),
1742 "iterations": submission_params.get("iterations"),
1743 "questions_per_iteration": submission_params.get(
1744 "questions_per_iteration"
1745 ),
1746 "strategy": submission_params.get("strategy", "source-based"),
1747 }
1749 query_validation_error = validate_research_query_length(
1750 queued_research.query
1751 )
1752 if query_validation_error is not None:
1753 raise InvalidQueuedResearchOverridesError(query_validation_error)
1755 validation_error = validate_search_overrides(runtime_overrides)
1756 if validation_error is not None:
1757 raise InvalidQueuedResearchOverridesError(validation_error)
1759 # Claim IN_PROGRESS before spawn to close the
1760 # thread-completes-before-parent-commits race.
1761 research.status = ResearchStatus.IN_PROGRESS
1762 db_session.commit()
1764 try:
1765 research_thread = start_research_process(
1766 research_id,
1767 queued_research.query,
1768 queued_research.mode,
1769 run_research_process,
1770 username=username,
1771 user_password=password, # Pass password for metrics
1772 settings_snapshot=complete_settings,
1773 **runtime_overrides,
1774 )
1775 except DuplicateResearchError:
1776 # A live thread already exists for this research_id (e.g.
1777 # previous attempt's post-spawn commit failed). Do NOT
1778 # reset status — that would contradict the running thread.
1779 raise
1780 except SystemAtCapacityError:
1781 # System at concurrent-research capacity. No thread was
1782 # spawned. Reset to QUEUED so the next dispatch tick can try
1783 # again — this is not a permanent spawn failure and should
1784 # NOT count toward SPAWN_RETRY_LIMIT.
1785 logger.info(
1786 f"System at capacity when dispatching {research_id}; "
1787 "re-queueing for next tick"
1788 )
1789 research.status = ResearchStatus.QUEUED
1790 self._commit_with_safe_rollback(
1791 db_session,
1792 f"status reset to QUEUED after capacity reject for research {research_id}",
1793 )
1794 raise
1795 except Exception:
1796 # Genuine spawn failure: no thread exists. Roll back the
1797 # IN_PROGRESS claim so the retry sees a clean QUEUED row.
1798 research.status = ResearchStatus.QUEUED
1799 self._commit_with_safe_rollback(
1800 db_session,
1801 f"status reset to QUEUED after spawn failure for research {research_id}",
1802 )
1803 raise
1805 # Thread is running. Record the active-research row. If this
1806 # commit fails the live thread is unrecorded but still running.
1807 # Raise DuplicateResearchError instead of letting a generic
1808 # exception propagate, so the caller's dup branch cleans up the
1809 # queue row without bumping the retry counter — if we let this
1810 # count as a spawn failure, three consecutive post-spawn commit
1811 # failures (or one at LIMIT-1) would push the counter to
1812 # SPAWN_RETRY_LIMIT and mark a LIVE thread as terminal FAILED.
1813 active_record = UserActiveResearch(
1814 username=username,
1815 research_id=research_id,
1816 status=ResearchStatus.IN_PROGRESS,
1817 thread_id=str(research_thread.ident),
1818 settings_snapshot=queued_research.settings_snapshot,
1819 )
1820 db_session.add(active_record)
1821 if not self._commit_with_safe_rollback(
1822 db_session,
1823 f"UserActiveResearch persist after spawn for research {research_id}",
1824 ):
1825 # Thread is live; the commit failing leaves the UAR row
1826 # unrecorded but the thread running. Raise
1827 # DuplicateResearchError so the caller's dup branch deletes
1828 # the queue row without bumping the retry counter — if we
1829 # let a plain exception count as a spawn failure, a commit
1830 # failure at SPAWN_RETRY_LIMIT - 1 would mark a LIVE thread
1831 # as terminal FAILED.
1832 raise DuplicateResearchError(
1833 f"Research {research_id} thread is live; "
1834 "UserActiveResearch commit failed"
1835 )
1837 # Keep pending_operations bounded: if no request comes in to drain
1838 # them (user never navigates after a crash), they would otherwise
1839 # accumulate forever in memory. We drop oldest entries past this cap
1840 # and entries older than this TTL.
1841 _PENDING_OPS_MAX = 10_000
1842 _PENDING_OPS_TTL_SECONDS = 24 * 60 * 60 # 24 hours
1844 def reconcile_orphan_active_research(
1845 self, username: str, db_session
1846 ) -> int:
1847 """Mark stale UserActiveResearch rows as FAILED.
1849 Called after a user logs in (when their DB is open). If the server
1850 was restarted mid-research, rows with status=IN_PROGRESS have a
1851 thread_id pointing to a now-dead thread. Leaving them as IN_PROGRESS
1852 means the dashboard shows "running forever" with no way to cancel.
1854 Only rows whose worker thread is actually DEAD are reconciled. This
1855 runs on EVERY login, so without the liveness guard a user opening a
1856 second tab / logging in on another device while a research is still
1857 running would wrongly flag that live research FAILED — and
1858 delete_attempt could then hard-delete it out from under the running
1859 worker. Mirrors the is_research_thread_alive() guard chat.py and the
1860 research-start reclaim already use.
1862 Returns the number of rows reconciled.
1863 """
1864 from datetime import UTC, datetime
1866 from ..research_state import cleanup_research, is_research_thread_alive
1868 try:
1869 stale = (
1870 db_session.query(UserActiveResearch)
1871 .filter_by(username=username, status=ResearchStatus.IN_PROGRESS)
1872 .all()
1873 )
1874 if not stale:
1875 return 0
1877 count = 0
1878 for record in stale:
1879 # Skip rows whose worker thread is still alive — those are
1880 # genuinely running (e.g. a concurrent login), not orphans.
1881 if is_research_thread_alive(record.research_id):
1882 continue
1883 # Also mark the ResearchHistory row as failed
1884 research = (
1885 db_session.query(ResearchHistory)
1886 .filter_by(id=record.research_id)
1887 .first()
1888 )
1889 if research and research.status == ResearchStatus.IN_PROGRESS: 1889 ↛ 1897line 1889 didn't jump to line 1897 because the condition on line 1889 was always true
1890 research.status = ResearchStatus.FAILED
1891 research.completed_at = datetime.now(UTC).isoformat()
1892 meta = dict(research.research_meta or {})
1893 meta["failure_reason"] = (
1894 "Server restarted while research was in progress"
1895 )
1896 research.research_meta = meta
1897 record.status = ResearchStatus.FAILED
1898 # Drop the dead thread's in-memory entry too, mirroring
1899 # reclaim_stale_user_active_research. Without this the stale
1900 # _active_research entry outlives the DB reconciliation, so
1901 # capacity checks keep counting a research that is already
1902 # FAILED and its termination flag never clears.
1903 cleanup_research(record.research_id)
1904 count += 1
1906 db_session.commit()
1907 logger.info(
1908 f"Reconciled {count} orphan IN_PROGRESS research records for {username}"
1909 )
1910 return count
1911 except Exception:
1912 logger.exception(
1913 f"Failed to reconcile orphan research records for {username}"
1914 )
1915 return 0
1917 def _evict_stale_pending_operations(self) -> None:
1918 """Drop expired and over-capacity entries. Caller must hold the lock."""
1919 now = time.time()
1920 # TTL eviction
1921 expired = [
1922 op_id
1923 for op_id, op in self.pending_operations.items()
1924 if now - op.get("timestamp", now) > self._PENDING_OPS_TTL_SECONDS
1925 ]
1926 for op_id in expired:
1927 del self.pending_operations[op_id]
1928 # Size cap — drop oldest entries if over
1929 overflow = len(self.pending_operations) - self._PENDING_OPS_MAX
1930 if overflow > 0: 1930 ↛ 1931line 1930 didn't jump to line 1931 because the condition on line 1930 was never true
1931 oldest = sorted(
1932 self.pending_operations.items(),
1933 key=lambda kv: kv[1].get("timestamp", 0),
1934 )[:overflow]
1935 for op_id, _ in oldest:
1936 del self.pending_operations[op_id]
1938 def queue_progress_update(
1939 self, username: str, research_id: str, progress: float
1940 ):
1941 """
1942 Queue a progress update that needs database access.
1943 For compatibility with old processor during migration.
1945 Args:
1946 username: The username
1947 research_id: The research ID
1948 progress: The progress value (0-100)
1949 """
1950 # In processor_v2, we can update directly if we have database access
1951 # or queue it for later processing
1952 operation_id = str(uuid.uuid4())
1953 with self._pending_operations_lock:
1954 self.pending_operations[operation_id] = {
1955 "username": username,
1956 "operation_type": "progress_update",
1957 "research_id": research_id,
1958 "progress": progress,
1959 "timestamp": time.time(),
1960 }
1961 self._evict_stale_pending_operations()
1962 logger.debug(
1963 f"Queued progress update for research {research_id}: {progress}%"
1964 )
1966 def queue_error_update(
1967 self,
1968 username: str,
1969 research_id: str,
1970 status: str,
1971 error_message: str,
1972 metadata: Dict[str, Any],
1973 completed_at: str,
1974 report_path: Optional[str] = None,
1975 ):
1976 """
1977 Queue an error status update that needs database access.
1978 For compatibility with old processor during migration.
1980 Args:
1981 username: The username
1982 research_id: The research ID
1983 status: The status to set (failed, suspended, etc.)
1984 error_message: The error message
1985 metadata: Research metadata
1986 completed_at: Completion timestamp
1987 report_path: Optional path to error report
1988 """
1989 operation_id = str(uuid.uuid4())
1990 with self._pending_operations_lock:
1991 self.pending_operations[operation_id] = {
1992 "username": username,
1993 "operation_type": "error_update",
1994 "research_id": research_id,
1995 "status": status,
1996 "error_message": error_message,
1997 "metadata": metadata,
1998 "completed_at": completed_at,
1999 "report_path": report_path,
2000 "timestamp": time.time(),
2001 }
2002 self._evict_stale_pending_operations()
2003 logger.info(
2004 f"Queued error update for research {research_id} with status {status}"
2005 )
2007 def process_pending_operations_for_user(
2008 self, username: str, db_session: Session
2009 ) -> int:
2010 """
2011 Process pending operations for a user when we have database access.
2012 Called from request context where encrypted database is accessible.
2013 For compatibility with old processor during migration.
2015 Args:
2016 username: Username to process operations for
2017 db_session: Active database session for the user
2019 Returns:
2020 Number of operations cleared from the queue
2021 """
2022 # Find pending operations for this user (with lock)
2023 raw_operations = []
2024 with self._pending_operations_lock:
2025 for op_id, op_data in list(self.pending_operations.items()):
2026 if op_data["username"] == username:
2027 raw_operations.append((op_id, op_data))
2028 # Remove immediately to prevent duplicate processing
2029 del self.pending_operations[op_id]
2031 if not raw_operations:
2032 return 0
2034 # Deduplicate and consolidate operations by research_id to minimize DB locks
2035 latest_progress = {}
2036 latest_errors = {}
2037 other_ops = []
2039 for op_id, op_data in raw_operations:
2040 op_type = op_data.get("operation_type")
2041 rid = op_data.get("research_id")
2042 ts = op_data.get("timestamp", 0)
2044 if op_type == "progress_update" and rid:
2045 if rid not in latest_progress or ts >= latest_progress[rid][
2046 1
2047 ].get("timestamp", 0):
2048 latest_progress[rid] = (op_id, op_data)
2049 elif op_type == "error_update" and rid: 2049 ↛ 2055line 2049 didn't jump to line 2055 because the condition on line 2049 was always true
2050 if rid not in latest_errors or ts >= latest_errors[rid][1].get( 2050 ↛ 2039line 2050 didn't jump to line 2039 because the condition on line 2050 was always true
2051 "timestamp", 0
2052 ):
2053 latest_errors[rid] = (op_id, op_data)
2054 else:
2055 other_ops.append((op_id, op_data))
2057 operations_to_process = (
2058 list(latest_progress.values())
2059 + list(latest_errors.values())
2060 + other_ops
2061 )
2063 from sqlalchemy.exc import (
2064 OperationalError,
2065 PendingRollbackError,
2066 TimeoutError,
2067 )
2069 from ...database.models import ResearchHistory
2071 max_retries = 3
2072 backoff = 0.05
2073 max_requeue_attempts = 10
2074 # Track the raw count of operations being flushed so the returned
2075 # count reflects "operations cleared from the queue" rather than
2076 # "deduplicated batches applied". Set after a successful commit
2077 # so retries don't double-count.
2078 processed_count = 0
2080 for attempt in range(max_retries):
2081 try:
2082 # Consolidate DB queries: fetch all relevant ResearchHistory rows in a single batch
2083 rids = list(
2084 {
2085 op_data["research_id"]
2086 for _, op_data in operations_to_process
2087 if op_data.get("research_id")
2088 }
2089 )
2090 if rids: 2090 ↛ 2098line 2090 didn't jump to line 2098 because the condition on line 2090 was always true
2091 researches = (
2092 db_session.query(ResearchHistory)
2093 .filter(ResearchHistory.id.in_(rids))
2094 .all()
2095 )
2096 research_map = {r.id: r for r in researches}
2097 else:
2098 research_map = {}
2100 for op_id, op_data in operations_to_process:
2101 op_type = op_data.get("operation_type")
2102 rid = op_data.get("research_id")
2103 research = research_map.get(rid)
2105 if not research:
2106 continue
2108 if op_type == "progress_update":
2109 research.progress = op_data.get("progress")
2110 elif op_type == "error_update": 2110 ↛ 2100line 2110 didn't jump to line 2100 because the condition on line 2110 was always true
2111 research.status = op_data.get("status", "failed")
2112 research.error_message = op_data.get("error_message")
2113 incoming_meta = op_data.get("metadata")
2114 existing_meta = research.research_meta
2115 if isinstance(incoming_meta, dict): 2115 ↛ 2131line 2115 didn't jump to line 2131 because the condition on line 2115 was always true
2116 # Merge, never wholesale-replace: a bare
2117 # {"terminated_at": ...} from handle_termination
2118 # must not wipe the row's settings_snapshot (and
2119 # submission block) that view/export rendering
2120 # resolves report.uncited_sources_mode from.
2121 # Callers that already merged (the failure path)
2122 # are unaffected: re-merging is idempotent.
2123 # Read legacy JSON-string metadata using the
2124 # same normalization as completion updates.
2125 merged_meta = _parse_research_metadata(
2126 existing_meta
2127 )
2128 merged_meta.update(incoming_meta)
2129 research.research_meta = merged_meta
2130 else:
2131 research.research_meta = incoming_meta
2132 research.completed_at = op_data.get("completed_at")
2133 report_path = op_data.get("report_path")
2134 if report_path:
2135 research.report_path = report_path
2137 # Single commit for the entire batch of consolidated updates.
2138 # Count is set AFTER the commit succeeds so retries don't
2139 # double-count the raw operations.
2140 db_session.commit()
2141 processed_count = len(raw_operations)
2142 break
2144 except (OperationalError, PendingRollbackError, TimeoutError) as e:
2145 try:
2146 db_session.rollback()
2147 except Exception as rollback_err:
2148 logger.debug(f"Queue flush rollback failed: {rollback_err}")
2150 if attempt < max_retries - 1:
2151 logger.debug(
2152 f"Database locked during queue flush for {username} (attempt {attempt + 1}/{max_retries}); retrying in {backoff * 1000:.0f}ms: {type(e).__name__}"
2153 )
2154 time.sleep(backoff)
2155 backoff *= 2
2156 else:
2157 logger.warning(
2158 f"Failed to commit queue operations for {username} after {max_retries} retries: {type(e).__name__}"
2159 )
2160 # Re-queue raw operations so data is not lost, up to max_requeue_attempts
2161 with self._pending_operations_lock:
2162 for op_id, op_data in raw_operations:
2163 retry_count = op_data.get("retry_count", 0) + 1
2164 if retry_count > max_requeue_attempts:
2165 logger.exception(
2166 f"Dropping queued operation {op_id} ({op_data.get('operation_type')}) for user {username} "
2167 f"after exceeding maximum requeue attempts ({max_requeue_attempts})"
2168 )
2169 elif op_id not in self.pending_operations: 2169 ↛ 2162line 2169 didn't jump to line 2162 because the condition on line 2169 was always true
2170 op_data["retry_count"] = retry_count
2171 self.pending_operations[op_id] = op_data
2172 processed_count = 0
2174 except Exception:
2175 logger.exception(
2176 f"Unexpected error committing queue operations for {username}"
2177 )
2178 try:
2179 db_session.rollback()
2180 except Exception as rollback_err:
2181 logger.debug(f"Queue flush rollback failed: {rollback_err}")
2182 # Re-queue raw operations up to max_requeue_attempts
2183 with self._pending_operations_lock:
2184 for op_id, op_data in raw_operations:
2185 retry_count = op_data.get("retry_count", 0) + 1
2186 if retry_count > max_requeue_attempts: 2186 ↛ 2187line 2186 didn't jump to line 2187 because the condition on line 2186 was never true
2187 logger.exception(
2188 f"Dropping queued operation {op_id} ({op_data.get('operation_type')}) for user {username} "
2189 f"after exceeding maximum requeue attempts ({max_requeue_attempts})"
2190 )
2191 elif op_id not in self.pending_operations: 2191 ↛ 2184line 2191 didn't jump to line 2184 because the condition on line 2191 was always true
2192 op_data["retry_count"] = retry_count
2193 self.pending_operations[op_id] = op_data
2194 processed_count = 0
2195 break
2197 return processed_count
2200# Global queue processor instance
2201queue_processor = QueueProcessorV2()