Coverage for src/local_deep_research/web/queue/processor_v2.py: 91%
493 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +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 typing import Any, Dict, Optional
12from loguru import logger
13from sqlalchemy.orm import Session
15from ...constants import ResearchStatus
16from ...database.encrypted_db import db_manager
17from ...database.models import (
18 QueuedResearch,
19 ResearchHistory,
20 TaskMetadata,
21 UserActiveResearch,
22)
23from ...database.queue_service import UserQueueService
24from ...database.session_context import get_user_db_session
25from ...database.session_passwords import session_password_store
26from ...exceptions import DuplicateResearchError, SystemAtCapacityError
27from ...security.log_sanitizer import redact_secrets
28from ...notifications.queue_helpers import (
29 send_research_completed_notification_from_session,
30 send_research_failed_notification_from_session,
31)
32from .lifecycle_cleanup import (
33 cleanup_queued_research_state,
34 reconcile_research_queue_status,
35)
36from ..services.research_service import (
37 run_research_process,
38 start_research_process,
39)
41# Retry configuration constants for notification database queries
42MAX_RESEARCH_LOOKUP_RETRIES = 3
43INITIAL_RESEARCH_LOOKUP_DELAY = 0.5 # seconds
44RETRY_BACKOFF_MULTIPLIER = 2
46# Give up on a queued research after this many consecutive spawn failures.
47# Each failure leaves is_processing=False so the next loop tick retries.
48SPAWN_RETRY_LIMIT = 3
51@dataclass(frozen=True, slots=True)
52class QueueSweepResult:
53 can_dispatch: bool
54 cleaned_ids: frozenset[str]
57class QueueProcessorV2:
58 """
59 Processes queued researches using encrypted user databases.
60 This replaces the service.db approach.
61 """
63 def __init__(self, check_interval=10):
64 """
65 Initialize the queue processor.
67 Args:
68 check_interval: How often to check for work (seconds)
69 """
70 self.check_interval = check_interval
71 self.running = False
72 self.thread = None
73 self._loop_iteration = 0
74 # Wakes the loop out of its inter-iteration wait so stop() returns
75 # in milliseconds instead of blocking for up to check_interval.
76 # The test suite stops the processor after every app test
77 # (tests/conftest.py reset_singletons), so a blocking stop() adds
78 # ~check_interval seconds of teardown to each of those tests.
79 self._stop_event = threading.Event()
81 # Per-user settings will be retrieved from each user's database
82 # when processing their queue using SettingsManager
83 logger.info(
84 "Queue processor v2 initialized - will use per-user settings from SettingsManager"
85 )
87 # Track which users we should check
88 self._users_to_check: set[tuple[str, str]] = set()
89 self._users_lock = threading.Lock()
91 # Track pending operations from background threads
92 self.pending_operations = {}
93 self._pending_operations_lock = threading.Lock()
95 # Per-user serialisation for the "count active → start direct"
96 # critical section. Without this the count-then-insert races with
97 # itself for the same user (e.g. two concurrent research submissions
98 # from two browser tabs), and IMMEDIATE isolation was what used to
99 # paper over it at the DB layer.
100 self._user_critical_locks: Dict[str, threading.Lock] = {}
101 self._user_critical_locks_lock = threading.Lock()
103 # Count consecutive spawn failures per research_id. Entries are
104 # popped on success or after hitting SPAWN_RETRY_LIMIT (then the
105 # research is marked FAILED). In-memory is sufficient: a restart
106 # resets the counter and the research gets a fresh N retries,
107 # which is the desired behavior if the underlying system issue
108 # (thread pool, memory) cleared.
109 # Access is guarded by _spawn_retry_counts_lock because the
110 # increment path is a read-modify-write and the loop and direct
111 # request paths can interleave.
112 self._spawn_retry_counts: dict[str, int] = {}
113 self._spawn_retry_counts_lock = threading.Lock()
115 def _get_user_critical_lock(self, username: str) -> threading.Lock:
116 """Get (or lazily create) the per-user lock used to serialise the
117 count-active-and-start-direct critical section for a given user.
118 """
119 with self._user_critical_locks_lock:
120 lock = self._user_critical_locks.get(username)
121 if lock is None: 121 ↛ 124line 121 didn't jump to line 124 because the condition on line 121 was always true
122 lock = threading.Lock()
123 self._user_critical_locks[username] = lock
124 return lock
126 def pop_user_critical_lock(self, username: str) -> None:
127 """Remove the per-user critical-section lock for ``username``.
129 Called from the user-close path so this instance dict doesn't
130 accumulate one entry per username across the process lifetime.
131 The next direct-research submission for that user lazily
132 re-creates the lock if needed — the lock has no state that
133 needs to persist across login/logout.
134 """
135 with self._user_critical_locks_lock:
136 self._user_critical_locks.pop(username, None)
138 def _bump_spawn_retry_count(self, research_id: str) -> int:
139 """Atomically increment and return the spawn-retry counter for
140 ``research_id``. Extracted so tests can exercise the real
141 locked increment path instead of duplicating the lock in the
142 test worker (which would be a tautology).
143 """
144 with self._spawn_retry_counts_lock:
145 attempts = self._spawn_retry_counts.get(research_id, 0) + 1
146 self._spawn_retry_counts[research_id] = attempts
147 return attempts
149 @staticmethod
150 def _commit_with_safe_rollback(db_session: Session, context: str) -> bool:
151 """Commit ``db_session`` with best-effort rollback on failure.
153 Returns ``True`` on success, ``False`` if the commit raised.
154 The failure path logs via ``logger.exception`` and attempts a
155 rollback, itself guarded so a subsequent rollback failure is
156 logged at debug level rather than propagated.
158 Extracted because the ``try: commit / except: log + try:
159 rollback`` idiom repeats at ≥5 sites in this module; inlining
160 hides the defensive structure behind nested ``try`` blocks and
161 makes each callsite longer than the work it describes.
162 """
163 try:
164 db_session.commit()
165 return True
166 except Exception:
167 logger.exception(f"Commit failed: {context}")
168 try:
169 db_session.rollback()
170 except Exception:
171 logger.debug(
172 f"Rollback after commit failure ({context})",
173 exc_info=True,
174 )
175 return False
177 def _delete_queue_row_safely(
178 self, db_session: Session, username: str, research_id: str
179 ) -> None:
180 """Best-effort delete of the ``QueuedResearch`` row for
181 ``(username, research_id)``.
183 Rolls back any pending state first (the session may be in
184 ``PendingRollbackError`` from a failed commit inside
185 ``_start_research``), re-queries the row fresh, deletes it if
186 present, and commits via ``_commit_with_safe_rollback``.
188 Use this for ``DuplicateResearchError`` cleanup where the goal
189 is "drop the queue row regardless of session state." Do NOT
190 use it for paths that need the delete to be atomic with other
191 writes (e.g. the terminal FAILED path bundles
192 ``ResearchHistory.status = FAILED`` with the queue-row delete
193 in a single commit — that stays inline).
194 """
195 try:
196 db_session.rollback()
197 except Exception:
198 logger.debug(
199 f"Rollback before queue-row delete for {research_id}",
200 exc_info=True,
201 )
202 try:
203 fresh_queued = (
204 db_session.query(QueuedResearch)
205 .filter_by(username=username, research_id=research_id)
206 .first()
207 )
208 if fresh_queued:
209 db_session.delete(fresh_queued)
210 self._commit_with_safe_rollback(
211 db_session,
212 f"queue-row delete for research {research_id}",
213 )
214 except Exception:
215 logger.exception(
216 f"Failed to query/delete queue row for {research_id}"
217 )
218 try:
219 db_session.rollback()
220 except Exception:
221 logger.debug(
222 f"Rollback after queue-row delete failure for {research_id}",
223 exc_info=True,
224 )
226 def start(self):
227 """Start the queue processor thread."""
228 if self.running:
229 logger.warning("Queue processor already running")
230 return
232 self.running = True
233 # Re-arm the wait for restart: stop() leaves the event set, and
234 # create_app() restarts this singleton after the test-suite
235 # teardown stops it.
236 self._stop_event.clear()
237 self.thread = threading.Thread(
238 target=self._process_queue_loop, daemon=True
239 )
240 self.thread.start()
241 logger.info("Queue processor v2 started")
243 def stop(self):
244 """Stop the queue processor thread."""
245 self.running = False
246 self._stop_event.set()
247 if self.thread:
248 self.thread.join(timeout=10)
249 logger.info("Queue processor v2 stopped")
251 def notify_user_activity(self, username: str, session_id: str):
252 """
253 Notify that a user has activity and their queue should be checked.
255 Args:
256 username: The username
257 session_id: The Flask session ID (for password access)
258 """
259 with self._users_lock:
260 self._users_to_check.add((username, session_id))
261 logger.debug(f"User {username} added to queue check list")
263 def notify_research_queued(self, username: str, research_id: str, **kwargs):
264 """
265 Notify that a research was queued.
266 In direct mode, this immediately starts the research if slots are available.
267 In queue mode, it adds to the queue.
269 Args:
270 username: The username
271 research_id: The research ID
272 **kwargs: Additional parameters for direct execution (query, mode, etc.)
273 """
274 # Pre-declared so the except handlers below can pass it to
275 # redact_secrets even on paths where it is never assigned.
276 password = None
277 # Check user's queue_mode setting when we have database access
278 if kwargs:
279 session_id = kwargs.get("session_id")
280 if session_id:
281 # Check if we can start it directly
282 password = session_password_store.get_session_password(
283 username, session_id
284 )
285 if password:
286 try:
287 # Open database and check settings + active count
288 engine = db_manager.open_user_database(
289 username, password
290 )
291 if engine: 291 ↛ 369line 291 didn't jump to line 369 because the condition on line 291 was always true
292 with get_user_db_session(username) as db_session:
293 # Get user's settings using SettingsManager
294 from ...settings.manager import SettingsManager
296 settings_manager = SettingsManager(db_session)
298 # Get user's queue_mode setting (env > DB > default)
299 queue_mode = settings_manager.get_setting(
300 "app.queue_mode", "direct"
301 )
303 # Get user's max concurrent setting (env > DB > default)
304 max_concurrent = settings_manager.get_setting(
305 "app.max_concurrent_researches", 3
306 )
308 logger.debug(
309 f"User {username} settings: queue_mode={queue_mode}, "
310 f"max_concurrent={max_concurrent}"
311 )
313 # Only try direct execution if user has queue_mode="direct"
314 if queue_mode == "direct":
315 # Serialise the count→check→start critical
316 # section at the application layer. Two
317 # concurrent submissions for the same user
318 # must not both observe the same active
319 # count and both start — that would exceed
320 # max_concurrent. A per-user Python lock
321 # gives us that atomicity independent of
322 # the DB isolation level.
323 with self._get_user_critical_lock(username):
324 active_count = (
325 db_session.query(UserActiveResearch)
326 .filter_by(
327 username=username,
328 status=ResearchStatus.IN_PROGRESS,
329 )
330 .count()
331 )
333 if active_count < max_concurrent:
334 # We have slots - start directly!
335 logger.info(
336 f"Direct mode: Starting research {research_id} immediately "
337 f"(active: {active_count}/{max_concurrent})"
338 )
340 # Start the research directly
341 self._start_research_directly(
342 username,
343 research_id,
344 password,
345 **kwargs,
346 )
347 return
348 logger.info(
349 f"Direct mode: Max concurrent reached ({active_count}/"
350 f"{max_concurrent}), queueing {research_id}"
351 )
352 else:
353 logger.info(
354 f"User {username} has queue_mode={queue_mode}, "
355 f"queueing research {research_id}"
356 )
357 except Exception as e:
358 # ``password`` is in scope — drop the traceback
359 # chain and redact str(e) so the SQLCipher master
360 # password can't leak via diagnose=True frame
361 # locals (see the generic handler in
362 # _start_research_directly for the full rationale).
363 safe_msg = redact_secrets(str(e), password)
364 logger.warning(
365 f"Error in direct execution for {username}: {safe_msg}"
366 )
368 # Fall back to queue mode (or if direct mode failed)
369 try:
370 with get_user_db_session(username) as session:
371 queue_service = UserQueueService(session)
372 queue_service.add_task_metadata(
373 task_id=research_id,
374 task_type="research",
375 priority=0,
376 )
377 logger.info(
378 f"Research {research_id} queued for user {username}"
379 )
380 except Exception as e:
381 # ``password`` may be bound above — same redaction rationale.
382 safe_msg = redact_secrets(str(e), password)
383 logger.warning(
384 f"Failed to update queue status for {username}: {safe_msg}"
385 )
387 def _start_research_directly(
388 self, username: str, research_id: str, password: str, **kwargs
389 ):
390 """
391 Start a research directly without queueing.
393 Args:
394 username: The username
395 research_id: The research ID
396 password: The user's password
397 **kwargs: Research parameters (query, mode, settings, etc.)
398 """
399 query = kwargs.get("query")
400 mode = kwargs.get("mode")
401 settings_snapshot = kwargs.get("settings_snapshot", {})
403 # Create active research record
404 try:
405 with get_user_db_session(username) as db_session:
406 active_record = UserActiveResearch(
407 username=username,
408 research_id=research_id,
409 status=ResearchStatus.IN_PROGRESS,
410 thread_id="pending",
411 settings_snapshot=settings_snapshot,
412 )
413 db_session.add(active_record)
414 db_session.commit()
416 # Update task status if it exists
417 queue_service = UserQueueService(db_session)
418 queue_service.update_task_status(research_id, "processing")
419 except Exception as e:
420 # ``password`` is a parameter of this method — drop the
421 # traceback chain and redact str(e) (full rationale at the
422 # generic handler below).
423 safe_msg = redact_secrets(str(e), password)
424 logger.warning(
425 f"Failed to create active research record for {research_id}: {safe_msg}"
426 )
427 return
429 # Extract parameters from kwargs
430 model_provider = kwargs.get("model_provider")
431 model = kwargs.get("model")
432 custom_endpoint = kwargs.get("custom_endpoint")
433 search_engine = kwargs.get("search_engine")
435 # Start the research process
436 try:
437 research_thread = start_research_process(
438 research_id,
439 query,
440 mode,
441 run_research_process,
442 username=username,
443 user_password=password,
444 model_provider=model_provider,
445 model=model,
446 custom_endpoint=custom_endpoint,
447 search_engine=search_engine,
448 max_results=kwargs.get("max_results"),
449 time_period=kwargs.get("time_period"),
450 iterations=kwargs.get("iterations"),
451 questions_per_iteration=kwargs.get("questions_per_iteration"),
452 strategy=kwargs.get("strategy", "source-based"),
453 settings_snapshot=settings_snapshot,
454 )
456 # Update thread ID
457 try:
458 with get_user_db_session(username) as db_session:
459 active_record = (
460 db_session.query(UserActiveResearch)
461 .filter_by(username=username, research_id=research_id)
462 .first()
463 )
464 if active_record: 464 ↛ 474line 464 didn't jump to line 474
465 active_record.thread_id = str(research_thread.ident)
466 db_session.commit()
467 except Exception as e:
468 # ``password`` is in scope — same redaction rationale.
469 safe_msg = redact_secrets(str(e), password)
470 logger.warning(
471 f"Failed to update thread ID for {research_id}: {safe_msg}"
472 )
474 logger.info(
475 f"Direct execution: Started research {research_id} for user {username} "
476 f"in thread {research_thread.ident}"
477 )
479 except DuplicateResearchError:
480 # A live thread already owns this research_id. Do NOT delete
481 # the UserActiveResearch row or mark ResearchHistory FAILED —
482 # that state belongs to the live thread, and mutating it
483 # would terminate a running research from the user's
484 # perspective while it keeps executing. Same contract as the
485 # queue processor's dedicated dup branch (#3506).
486 logger.warning(
487 f"Duplicate live thread detected for {research_id} "
488 "in direct mode; leaving state intact"
489 )
490 return
491 except SystemAtCapacityError:
492 # System at concurrent-research capacity in the direct-execution
493 # path. Roll back the IN_PROGRESS active row and mark history
494 # back to QUEUED so the queue processor can pick it up later.
495 logger.info(
496 f"Direct execution hit capacity for {research_id}; re-queueing"
497 )
498 try:
499 with get_user_db_session(username) as db_session:
500 active_record = (
501 db_session.query(UserActiveResearch)
502 .filter_by(username=username, research_id=research_id)
503 .first()
504 )
505 if active_record:
506 db_session.delete(active_record)
507 research_row = (
508 db_session.query(ResearchHistory)
509 .filter_by(id=research_id)
510 .first()
511 )
512 if research_row:
513 research_row.status = ResearchStatus.QUEUED
514 # Bump queued_tasks so _process_user_queue's
515 # `queued_tasks == 0` gate doesn't treat the queue as
516 # empty and strand the QueuedResearch row the submit
517 # path already created. The direct path returns before
518 # the normal add_task_metadata call, so this is the
519 # single, non-double-counting increment;
520 # _start_queued_researches later dispatches the row and
521 # update_task_status() transitions this TaskMetadata
522 # queued->processing (balancing the counter).
523 UserQueueService(db_session).add_task_metadata(
524 task_id=research_id,
525 task_type="research",
526 priority=0,
527 )
528 db_session.commit()
529 except Exception as e:
530 # ``password`` is in scope — same redaction rationale.
531 safe_msg = redact_secrets(str(e), password)
532 logger.warning(
533 f"Cleanup after capacity reject failed for "
534 f"{research_id}; the stale UserActiveResearch row is "
535 f"recovered by reclaim_stale_user_active_research: {safe_msg}"
536 )
537 return
538 except Exception as e:
539 # ``password`` is in lexical scope (function parameter,
540 # passed through to ``start_research_process``). A
541 # SQLAlchemy / requests exception from anywhere in
542 # ``start_research_process`` could carry frame locals that
543 # include the SQLCipher master password (which is
544 # unrecoverable — see TRUST.md §5). Drop the traceback chain
545 # and redact str(e) defensively.
546 safe_msg = redact_secrets(str(e), password)
547 logger.warning(
548 f"Failed to start research {research_id} directly: {safe_msg}"
549 )
550 # Clean up the active record AND mark the research terminal
551 # FAILED so the user-visible state matches reality (no running
552 # thread, not IN_PROGRESS). Same contract as the queue
553 # processor's terminal-failure branch (#3481).
554 try:
555 with get_user_db_session(username) as db_session:
556 active_record = (
557 db_session.query(UserActiveResearch)
558 .filter_by(username=username, research_id=research_id)
559 .first()
560 )
561 if active_record:
562 db_session.delete(active_record)
563 research_row = (
564 db_session.query(ResearchHistory)
565 .filter_by(id=research_id)
566 .first()
567 )
568 if research_row:
569 research_row.status = ResearchStatus.FAILED
570 db_session.commit()
571 except Exception as e2:
572 # ``password`` is in scope — same redaction rationale.
573 safe_msg = redact_secrets(str(e2), password)
574 logger.warning(
575 f"Failed to clean up active research record for {research_id}: {safe_msg}"
576 )
578 def notify_research_completed(
579 self, username: str, research_id: str, user_password: str | None = None
580 ):
581 """
582 Notify that a research completed.
583 Updates the user's queue status in their database.
585 Args:
586 username: The username
587 research_id: The research ID
588 user_password: User password for database access. Required for queue
589 updates and database lookups during notification sending.
590 Optional only because some callers may not have it
591 available, in which case only basic updates occur.
592 """
593 try:
594 # get_user_db_session is already imported at module level (line 19)
595 # It accepts optional password parameter and returns a context manager
596 with get_user_db_session(username, user_password) as session:
597 queue_service = UserQueueService(session)
598 queue_service.update_task_status(
599 research_id, ResearchStatus.COMPLETED
600 )
601 logger.info(
602 f"Research {research_id} completed for user {username}"
603 )
605 # Send notification using helper from notification module
606 send_research_completed_notification_from_session(
607 username=username,
608 research_id=research_id,
609 db_session=session,
610 )
612 except Exception as e:
613 # ``user_password`` is a parameter of this method — drop the
614 # traceback chain and redact str(e) so the SQLCipher master
615 # password can't leak via diagnose=True frame locals.
616 safe_msg = redact_secrets(str(e), user_password)
617 logger.warning(
618 f"Failed to update completion status for {username}: {safe_msg}"
619 )
621 # Auto-convert research to document in History collection.
622 # Documents only — FAISS indexing is triggered separately by the user
623 # via "Index All" on the History page.
624 from ...research_library.search.services.research_history_indexer import (
625 auto_convert_research,
626 )
628 auto_convert_research(username, research_id, db_password=user_password)
630 def notify_research_failed(
631 self,
632 username: str,
633 research_id: str,
634 error_message: str | None = None,
635 user_password: str | None = None,
636 ):
637 """
638 Notify that a research failed.
639 Updates the user's queue status in their database and sends notification.
641 Args:
642 username: The username
643 research_id: The research ID
644 error_message: Optional error message
645 user_password: User password for database access. Required for queue
646 updates and database lookups during notification sending.
647 Optional only because some callers may not have it
648 available, in which case only basic updates occur.
649 """
650 try:
651 # get_user_db_session is already imported at module level (line 19)
652 # It accepts optional password parameter and returns a context manager
653 with get_user_db_session(username, user_password) as session:
654 queue_service = UserQueueService(session)
655 queue_service.update_task_status(
656 research_id,
657 ResearchStatus.FAILED,
658 error_message=error_message,
659 )
660 logger.info(
661 f"Research {research_id} failed for user {username}: "
662 f"{error_message}"
663 )
665 # Send notification using helper from notification module
666 send_research_failed_notification_from_session(
667 username=username,
668 research_id=research_id,
669 error_message=error_message or "Unknown error",
670 db_session=session,
671 )
673 except Exception as e:
674 # ``user_password`` is a parameter of this method — same
675 # redaction rationale as notify_research_completed.
676 safe_msg = redact_secrets(str(e), user_password)
677 logger.warning(
678 f"Failed to update failure status for {username}: {safe_msg}"
679 )
681 def _process_queue_loop(self):
682 """Main loop that processes the queue."""
683 while self.running:
684 try:
685 # Get list of users to check (don't clear immediately)
686 with self._users_lock:
687 users_to_check = list(self._users_to_check)
689 # Process each user's queue
690 users_to_remove = []
691 for user_session in users_to_check:
692 try:
693 username, session_id = user_session
694 # _process_user_queue returns True if queue is empty
695 queue_empty = self._process_user_queue(
696 username, session_id
697 )
698 if queue_empty:
699 users_to_remove.append(user_session)
700 except Exception:
701 logger.exception(
702 f"Error processing queue for {user_session}"
703 )
704 # Don't remove on error - the _process_user_queue method
705 # determines whether to keep checking based on error type
707 # Only remove users whose queues are now empty
708 with self._users_lock:
709 for user_session in users_to_remove:
710 self._users_to_check.discard(user_session)
712 except Exception:
713 logger.exception("Error in queue processor loop")
714 finally:
715 # Clean up thread-local database session after each iteration.
716 # The loop opens a new session each iteration via get_user_db_session();
717 # closing it returns the connection to the shared QueuePool promptly.
718 try:
719 from ...database.thread_local_session import (
720 cleanup_current_thread,
721 cleanup_dead_threads,
722 )
724 cleanup_current_thread()
725 except Exception:
726 logger.debug(
727 "thread-local cleanup on shutdown", exc_info=True
728 )
730 # Periodic dead-thread credential sweep (every ~60s).
731 # One of three sweep trigger points (app_factory
732 # teardown, connection_cleanup scheduler, and here).
733 self._loop_iteration += 1
734 if self._loop_iteration % 6 == 0: # Every ~60s (10s × 6)
735 try:
736 cleanup_dead_threads()
737 except Exception:
738 logger.debug(
739 "periodic dead-thread sweep", exc_info=True
740 )
742 # Event.wait, not time.sleep: stop() must be able to interrupt
743 # this pause, otherwise shutdown blocks for up to
744 # check_interval seconds.
745 self._stop_event.wait(self.check_interval)
747 def _process_user_queue(self, username: str, session_id: str) -> bool:
748 """
749 Process the queue for a specific user.
751 Args:
752 username: The username
753 session_id: The Flask session ID
755 Returns:
756 True if the queue is empty, False if there are still items
757 """
758 # Get the user's password from session store
759 password = session_password_store.get_session_password(
760 username, session_id
761 )
762 if not password:
763 logger.debug(
764 f"No password available for user {username}, skipping queue check"
765 )
766 return True # Remove from checking - session expired
768 # Open the user's encrypted database
769 try:
770 # First ensure the database is open
771 engine = db_manager.open_user_database(username, password)
772 if not engine:
773 logger.error(f"Failed to open database for user {username}")
774 return False # Keep checking - could be temporary DB issue
776 # Get a session and process the queue
777 with get_user_db_session(username, password) as db_session:
778 queue_service = UserQueueService(db_session)
780 sweep_result = self._sweep_missing_parent_queue_rows(
781 db_session, username
782 )
783 if not sweep_result.can_dispatch: 783 ↛ 784line 783 didn't jump to line 784 because the condition on line 783 was never true
784 return False
786 # Get user's settings using SettingsManager
787 from ...settings.manager import SettingsManager
789 settings_manager = SettingsManager(db_session)
791 # Get user's max concurrent setting (env > DB > default)
792 max_concurrent = settings_manager.get_setting(
793 "app.max_concurrent_researches", 3
794 )
796 # Get queue status
797 queue_status = queue_service.get_queue_status() or {
798 "active_tasks": 0,
799 "queued_tasks": 0,
800 }
802 # Calculate available slots
803 available_slots = max_concurrent - queue_status["active_tasks"]
805 if available_slots <= 0:
806 # No slots available, but queue might not be empty
807 return False # Keep checking
809 if queue_status["queued_tasks"] == 0:
810 # Queue is empty
811 return True # Remove from checking
813 logger.info(
814 f"Processing queue for {username}: "
815 f"{queue_status['active_tasks']} active, "
816 f"{queue_status['queued_tasks']} queued, "
817 f"{available_slots} slots available"
818 )
820 # Process queued researches
821 self._start_queued_researches(
822 db_session,
823 queue_service,
824 username,
825 password,
826 available_slots,
827 )
829 # Check if there are still items in queue
830 updated_status = queue_service.get_queue_status() or {
831 "queued_tasks": 0
832 }
833 return bool(updated_status["queued_tasks"] == 0)
835 except Exception as e:
836 # ``password`` (from the session store above) is in scope —
837 # drop the traceback chain and redact str(e).
838 safe_msg = redact_secrets(str(e), password)
839 logger.warning(
840 f"Error processing queue for user {username}: {safe_msg}"
841 )
842 return False # Keep checking - errors might be temporary
844 def _reclaim_stranded_queue_rows(
845 self, db_session: Session, username: str
846 ) -> int:
847 """Reclaim queue rows stranded by a crash or restart.
849 A row is stranded when ``is_processing=True`` but no live thread
850 exists in ``_active_research`` for its ``research_id``. This can
851 happen after a crash/restart between the pre-spawn IN_PROGRESS
852 commit and the queue-row deletion in ``_start_queued_researches``
853 — the row is invisible to the normal ``is_processing=False``
854 query and would never be retried.
856 Reverts only rows whose parent is not database-backed IN_PROGRESS,
857 preserving the worker spawn-grace window. Returns the number of
858 rows reclaimed.
859 """
860 from ..routes.globals import is_research_active
862 stranded = (
863 db_session.query(QueuedResearch)
864 .filter_by(username=username, is_processing=True)
865 .all()
866 )
867 reclaimed = 0
868 for row in stranded:
869 if is_research_active(row.research_id):
870 # A legitimate in-flight claim; don't touch.
871 continue
872 row.is_processing = False
873 research = (
874 db_session.query(ResearchHistory)
875 .filter_by(id=row.research_id)
876 .first()
877 )
878 status_changed = (
879 research is not None
880 and research.status == ResearchStatus.IN_PROGRESS
881 )
882 if status_changed:
883 research.status = ResearchStatus.QUEUED
884 reclaimed += 1
885 logger.warning(
886 f"Reclaimed stranded queue row for research "
887 f"{row.research_id} (user {username}): no live thread, "
888 "resetting is_processing=False"
889 + (" and status=QUEUED" if status_changed else "")
890 )
891 if reclaimed:
892 if not self._commit_with_safe_rollback( 892 ↛ 896line 892 didn't jump to line 896 because the condition on line 892 was never true
893 db_session,
894 f"reclaim of stranded rows for user {username}",
895 ):
896 return 0
897 return reclaimed
899 def _sweep_missing_parent_queue_rows(
900 self, db_session: Session, username: str
901 ) -> QueueSweepResult:
902 queued_orphan_ids = {
903 research_id
904 for (research_id,) in (
905 db_session.query(QueuedResearch.research_id)
906 .outerjoin(
907 ResearchHistory,
908 QueuedResearch.research_id == ResearchHistory.id,
909 )
910 .filter(
911 QueuedResearch.username == username,
912 ResearchHistory.id.is_(None),
913 )
914 .all()
915 )
916 }
917 metadata_orphan_ids = {
918 task_id
919 for (task_id,) in (
920 db_session.query(TaskMetadata.task_id)
921 .outerjoin(
922 ResearchHistory,
923 TaskMetadata.task_id == ResearchHistory.id,
924 )
925 .filter(
926 TaskMetadata.task_type == "research",
927 ResearchHistory.id.is_(None),
928 )
929 .all()
930 )
931 }
932 orphaned_research_ids = queued_orphan_ids | metadata_orphan_ids
933 if not orphaned_research_ids:
934 if reconcile_research_queue_status(db_session):
935 if not self._commit_with_safe_rollback( 935 ↛ 939line 935 didn't jump to line 939 because the condition on line 935 was never true
936 db_session,
937 f"queue status reconciliation for user {username}",
938 ):
939 return QueueSweepResult(False, frozenset())
940 return QueueSweepResult(True, frozenset())
942 cleanup_result = cleanup_queued_research_state(
943 db_session,
944 orphaned_research_ids,
945 include_claimed=True,
946 )
947 if not self._commit_with_safe_rollback(
948 db_session,
949 f"missing-parent queue sweep for user {username}",
950 ):
951 return QueueSweepResult(False, frozenset())
953 with self._spawn_retry_counts_lock:
954 for research_id in cleanup_result.cleaned_ids:
955 self._spawn_retry_counts.pop(research_id, None)
956 return QueueSweepResult(True, cleanup_result.cleaned_ids)
958 def _start_queued_researches(
959 self,
960 db_session: Session,
961 queue_service: UserQueueService,
962 username: str,
963 password: str,
964 available_slots: int,
965 ):
966 """Start queued researches up to available slots."""
967 sweep_result = self._sweep_missing_parent_queue_rows(
968 db_session, username
969 )
970 if not sweep_result.can_dispatch:
971 return
973 # Before picking work, reclaim any rows stranded by a prior
974 # crash — otherwise they are invisible to the is_processing=False
975 # filter below and would never retry.
976 self._reclaim_stranded_queue_rows(db_session, username)
978 # Get queued researches
979 queued = (
980 db_session.query(QueuedResearch)
981 .filter_by(username=username, is_processing=False)
982 .order_by(QueuedResearch.position, QueuedResearch.id)
983 .limit(available_slots)
984 .all()
985 )
987 for queued_research in queued:
988 research_id = queued_research.research_id
989 try:
990 # Atomically claim this item by flipping is_processing from
991 # False to True in a single UPDATE. If another worker has
992 # already claimed it since our SELECT above, the UPDATE will
993 # match zero rows and we skip. Under non-IMMEDIATE isolation
994 # the previous SELECT+assign pattern would race and two
995 # workers could both process the same queued item.
996 claimed = (
997 db_session.query(QueuedResearch)
998 .filter(
999 QueuedResearch.id == queued_research.id,
1000 QueuedResearch.is_processing.is_(False),
1001 db_session.query(ResearchHistory.id)
1002 .filter(
1003 ResearchHistory.id == research_id,
1004 ResearchHistory.status == ResearchStatus.QUEUED,
1005 )
1006 .exists(),
1007 )
1008 .update(
1009 {QueuedResearch.is_processing: True},
1010 synchronize_session=False,
1011 )
1012 )
1013 db_session.commit()
1014 if not claimed:
1015 logger.debug(
1016 f"Queued research {research_id} "
1017 f"already claimed by another worker; skipping"
1018 )
1019 continue
1020 # Refresh local object state now that we hold the claim
1021 db_session.refresh(queued_research)
1023 # Update task status
1024 queue_service.update_task_status(research_id, "processing")
1026 # Start the research
1027 self._start_research(
1028 db_session,
1029 username,
1030 password,
1031 queued_research,
1032 )
1034 # Success — clear any prior spawn-failure count and
1035 # remove the queue row.
1036 with self._spawn_retry_counts_lock:
1037 self._spawn_retry_counts.pop(research_id, None)
1038 db_session.delete(queued_research)
1039 db_session.commit()
1041 logger.info(
1042 f"Started queued research {research_id} for user {username}"
1043 )
1045 except DuplicateResearchError:
1046 # Raised by _start_research when a prior attempt's thread
1047 # is still live, OR when the ResearchHistory row is in a
1048 # non-QUEUED state (IN_PROGRESS from a prior attempt's
1049 # successful pre-spawn commit; terminal COMPLETED /
1050 # FAILED / SUSPENDED from a thread that already finished
1051 # and cleaned up). In every case the correct behavior is
1052 # the same: clear the stale queue row and the retry
1053 # counter, and do NOT fall through to the FAILED/notify
1054 # path — that would terminate-status a live thread or
1055 # emit a false failure for a completed one.
1056 logger.warning(
1057 f"Research {research_id} is already started "
1058 "(live thread or non-QUEUED status); clearing stale "
1059 "queue row"
1060 )
1061 with self._spawn_retry_counts_lock:
1062 self._spawn_retry_counts.pop(research_id, None)
1063 self._delete_queue_row_safely(db_session, username, research_id)
1064 continue
1066 except SystemAtCapacityError:
1067 # System hit the global concurrent-research capacity while
1068 # dispatching this queued item. _start_research already
1069 # reset the ResearchHistory row back to QUEUED before
1070 # re-raising. This is a transient condition, NOT a spawn
1071 # failure, so it must NOT count toward SPAWN_RETRY_LIMIT —
1072 # otherwise a busy system would wrongly mark a perfectly
1073 # valid queued research FAILED after a few ticks. Just
1074 # release the processing claim so the next tick retries.
1075 # Mirrors the dedicated handler in _start_research_directly.
1076 logger.info(
1077 f"System at capacity dispatching queued research "
1078 f"{research_id}; leaving queued for next tick"
1079 )
1080 # Revert the queued->processing claim from
1081 # update_task_status("processing") above. The research stays
1082 # queued for the next tick, so its slot must return to
1083 # queued_tasks rather than leaking into active_tasks on
1084 # every capacity-rejected retry.
1085 queue_service.update_task_status(research_id, "queued")
1086 fresh_queued = (
1087 db_session.query(QueuedResearch)
1088 .filter_by(username=username, research_id=research_id)
1089 .first()
1090 )
1091 if fresh_queued: 1091 ↛ 1098line 1091 didn't jump to line 1098 because the condition on line 1091 was always true
1092 fresh_queued.is_processing = False
1093 self._commit_with_safe_rollback(
1094 db_session,
1095 "is_processing reset after capacity reject for "
1096 f"research {research_id}",
1097 )
1098 continue
1100 except Exception as e:
1101 # ``password`` is a parameter of this method — drop the
1102 # traceback chain and redact str(e).
1103 safe_msg = redact_secrets(str(e), password)
1104 logger.warning(
1105 f"Error starting queued research {research_id}: {safe_msg}"
1106 )
1107 # Session may be in PendingRollbackError state after a
1108 # failed commit inside _start_research.
1109 try:
1110 db_session.rollback()
1111 except Exception as rb_err:
1112 # No exc_info: ``password`` is in this frame and a
1113 # rendered traceback could expose it via
1114 # diagnose=True frame locals.
1115 logger.debug(
1116 "Rollback after start failure: "
1117 f"{redact_secrets(str(rb_err), password)}"
1118 )
1120 attempts = self._bump_spawn_retry_count(research_id)
1122 # Re-query in case rollback expired the ORM object.
1123 fresh_queued = (
1124 db_session.query(QueuedResearch)
1125 .filter_by(username=username, research_id=research_id)
1126 .first()
1127 )
1129 if attempts < SPAWN_RETRY_LIMIT:
1130 # Transient failure — allow the next loop tick to
1131 # retry. _start_research rolls back its own
1132 # IN_PROGRESS write on spawn failure, so the only
1133 # fix-up needed here is resetting is_processing.
1134 logger.warning(
1135 f"Spawn failed for research {research_id} "
1136 f"(attempt {attempts}/{SPAWN_RETRY_LIMIT}), "
1137 "leaving queued for retry"
1138 )
1139 if fresh_queued: 1139 ↛ 1145line 1139 didn't jump to line 1145 because the condition on line 1139 was always true
1140 fresh_queued.is_processing = False
1141 self._commit_with_safe_rollback(
1142 db_session,
1143 f"is_processing reset for research {research_id}",
1144 )
1145 continue
1147 # Exhausted retries — mark terminal FAILED, delete the
1148 # queue row to stop re-dispatch, and notify the user.
1149 # The spawn failure was already logged (redacted) at the
1150 # top of this except block; no need to repeat it here.
1151 logger.warning(
1152 f"Spawn failed for research {research_id} "
1153 f"after {attempts} attempts; marking FAILED"
1154 )
1155 with self._spawn_retry_counts_lock:
1156 self._spawn_retry_counts.pop(research_id, None)
1157 try:
1158 research = (
1159 db_session.query(ResearchHistory)
1160 .filter_by(id=research_id)
1161 .first()
1162 )
1163 if research: 1163 ↛ 1165line 1163 didn't jump to line 1165 because the condition on line 1163 was always true
1164 research.status = ResearchStatus.FAILED
1165 if fresh_queued: 1165 ↛ 1167line 1165 didn't jump to line 1167 because the condition on line 1165 was always true
1166 db_session.delete(fresh_queued)
1167 db_session.commit()
1168 except Exception as e2:
1169 # ``password`` is in scope — same redaction rationale.
1170 safe_msg = redact_secrets(str(e2), password)
1171 logger.warning(
1172 "Failed to persist terminal FAILED state for "
1173 f"research {research_id}: {safe_msg}"
1174 )
1175 try:
1176 db_session.rollback()
1177 except Exception as rb_err:
1178 # No exc_info: same frame-locals rationale as the
1179 # rollback handler above.
1180 logger.debug(
1181 "Rollback after terminal update failure: "
1182 f"{redact_secrets(str(rb_err), password)}"
1183 )
1185 # notify_research_failed opens its own session and
1186 # sends the user notification. Called exactly once
1187 # per research_id because the counter is popped above.
1188 self.notify_research_failed(
1189 username=username,
1190 research_id=research_id,
1191 error_message=(
1192 f"Failed to start research after {attempts} attempts"
1193 ),
1194 user_password=password,
1195 )
1197 def _start_research(
1198 self,
1199 db_session: Session,
1200 username: str,
1201 password: str,
1202 queued_research,
1203 ):
1204 """Start a queued research.
1206 Commits ``ResearchHistory.status = IN_PROGRESS`` BEFORE spawning
1207 the thread. If we did this after, a fast-completing thread
1208 (which opens its own DB session) could write ``COMPLETED`` and
1209 then our post-spawn commit would overwrite that with
1210 ``IN_PROGRESS``, stranding the research as stuck IN_PROGRESS
1211 after it had already finished.
1213 If ``start_research_process`` raises, reset status back to
1214 ``QUEUED`` and re-raise so the caller's 3-strike retry logic
1215 handles it. ``DuplicateResearchError`` is re-raised as-is
1216 because a thread is already running for this research; mutating
1217 status further would be wrong.
1218 """
1219 research_id = queued_research.research_id
1220 research = (
1221 db_session.query(ResearchHistory).filter_by(id=research_id).first()
1222 )
1224 if not research:
1225 raise ValueError(f"Research {research_id} not found")
1227 # Guard against re-entering _start_research on a retry when a
1228 # prior attempt's post-spawn UserActiveResearch commit failed:
1229 # - IN_PROGRESS means the prior thread is (or was) running.
1230 # - COMPLETED/FAILED means the prior thread already finished
1231 # and cleaned itself up out of _active_research, so a bare
1232 # retry would both overwrite the terminal status with
1233 # IN_PROGRESS and then spawn a *second* thread (because
1234 # check_and_start_research sees no live entry), re-running
1235 # the whole research.
1236 # In all three cases the correct behavior is the same: raise
1237 # DuplicateResearchError so the caller's existing dup branch
1238 # deletes the queue row without mutating status or notifying.
1239 if research.status != ResearchStatus.QUEUED:
1240 raise DuplicateResearchError(
1241 f"Research {research_id} is already started "
1242 f"(status={research.status})"
1243 )
1245 # Claim IN_PROGRESS before spawn to close the
1246 # thread-completes-before-parent-commits race.
1247 research.status = ResearchStatus.IN_PROGRESS
1248 db_session.commit()
1250 # Extract settings
1251 settings_snapshot = queued_research.settings_snapshot or {}
1253 # Handle new vs legacy structure
1254 if (
1255 isinstance(settings_snapshot, dict)
1256 and "submission" in settings_snapshot
1257 ):
1258 submission_params = settings_snapshot.get("submission", {})
1259 complete_settings = settings_snapshot.get("settings_snapshot", {})
1260 else:
1261 submission_params = settings_snapshot
1262 # A legacy-flat queued row (enqueued by an older version, pre-
1263 # "submission" wrapper) carries no settings_snapshot. Seed the run's
1264 # primary engine from the submitted search_engine so the worker's
1265 # egress build (resolve_run_primary_engine) doesn't fail closed on
1266 # an empty snapshot and refuse the run.
1267 _legacy_engine = submission_params.get("search_engine")
1268 complete_settings = (
1269 {"search.tool": _legacy_engine} if _legacy_engine else {}
1270 )
1272 try:
1273 research_thread = start_research_process(
1274 research_id,
1275 queued_research.query,
1276 queued_research.mode,
1277 run_research_process,
1278 username=username,
1279 user_password=password, # Pass password for metrics
1280 model_provider=submission_params.get("model_provider"),
1281 model=submission_params.get("model"),
1282 custom_endpoint=submission_params.get("custom_endpoint"),
1283 search_engine=submission_params.get("search_engine"),
1284 max_results=submission_params.get("max_results"),
1285 time_period=submission_params.get("time_period"),
1286 iterations=submission_params.get("iterations"),
1287 questions_per_iteration=submission_params.get(
1288 "questions_per_iteration"
1289 ),
1290 strategy=submission_params.get("strategy", "source-based"),
1291 settings_snapshot=complete_settings,
1292 )
1293 except DuplicateResearchError:
1294 # A live thread already exists for this research_id (e.g.
1295 # previous attempt's post-spawn commit failed). Do NOT
1296 # reset status — that would contradict the running thread.
1297 raise
1298 except SystemAtCapacityError:
1299 # System at concurrent-research capacity. No thread was
1300 # spawned. Reset to QUEUED so the next dispatch tick can try
1301 # again — this is not a permanent spawn failure and should
1302 # NOT count toward SPAWN_RETRY_LIMIT.
1303 logger.info(
1304 f"System at capacity when dispatching {research_id}; "
1305 "re-queueing for next tick"
1306 )
1307 research.status = ResearchStatus.QUEUED
1308 self._commit_with_safe_rollback(
1309 db_session,
1310 f"status reset to QUEUED after capacity reject for research {research_id}",
1311 )
1312 raise
1313 except Exception:
1314 # Genuine spawn failure: no thread exists. Roll back the
1315 # IN_PROGRESS claim so the retry sees a clean QUEUED row.
1316 research.status = ResearchStatus.QUEUED
1317 self._commit_with_safe_rollback(
1318 db_session,
1319 f"status reset to QUEUED after spawn failure for research {research_id}",
1320 )
1321 raise
1323 # Thread is running. Record the active-research row. If this
1324 # commit fails the live thread is unrecorded but still running.
1325 # Raise DuplicateResearchError instead of letting a generic
1326 # exception propagate, so the caller's dup branch cleans up the
1327 # queue row without bumping the retry counter — if we let this
1328 # count as a spawn failure, three consecutive post-spawn commit
1329 # failures (or one at LIMIT-1) would push the counter to
1330 # SPAWN_RETRY_LIMIT and mark a LIVE thread as terminal FAILED.
1331 active_record = UserActiveResearch(
1332 username=username,
1333 research_id=research_id,
1334 status=ResearchStatus.IN_PROGRESS,
1335 thread_id=str(research_thread.ident),
1336 settings_snapshot=queued_research.settings_snapshot,
1337 )
1338 db_session.add(active_record)
1339 if not self._commit_with_safe_rollback(
1340 db_session,
1341 f"UserActiveResearch persist after spawn for research {research_id}",
1342 ):
1343 # Thread is live; the commit failing leaves the UAR row
1344 # unrecorded but the thread running. Raise
1345 # DuplicateResearchError so the caller's dup branch deletes
1346 # the queue row without bumping the retry counter — if we
1347 # let a plain exception count as a spawn failure, a commit
1348 # failure at SPAWN_RETRY_LIMIT - 1 would mark a LIVE thread
1349 # as terminal FAILED.
1350 raise DuplicateResearchError(
1351 f"Research {research_id} thread is live; "
1352 "UserActiveResearch commit failed"
1353 )
1355 def process_user_request(self, username: str, session_id: str) -> int:
1356 """
1357 Process queue for a user during their request.
1358 This is called from request context to check and start queued items.
1360 Returns:
1361 Number of researches started
1362 """
1363 # Pre-declared so the except handler can pass it to redact_secrets
1364 # even if the exception is raised before the retrieve below.
1365 password = None
1366 try:
1367 # Add user to check list
1368 self.notify_user_activity(username, session_id)
1370 # Force immediate check (don't wait for loop)
1371 password = session_password_store.get_session_password(
1372 username, session_id
1373 )
1374 if password:
1375 # Open database and check queue
1376 engine = db_manager.open_user_database(username, password)
1377 if engine: 1377 ↛ 1390line 1377 didn't jump to line 1390 because the condition on line 1377 was always true
1378 with get_user_db_session(username) as db_session:
1379 queue_service = UserQueueService(db_session)
1380 status = queue_service.get_queue_status()
1382 if status and status["queued_tasks"] > 0:
1383 logger.info(
1384 f"User {username} has {status['queued_tasks']} "
1385 f"queued tasks, triggering immediate processing"
1386 )
1387 # Process will happen in background thread
1388 return int(status["queued_tasks"])
1390 return 0
1392 except Exception as e:
1393 # ``password`` may be bound above — drop the traceback chain
1394 # and redact str(e).
1395 safe_msg = redact_secrets(str(e), password)
1396 logger.warning(
1397 f"Error in process_user_request for {username}: {safe_msg}"
1398 )
1399 return 0
1401 def queue_progress_update(
1402 self, username: str, research_id: str, progress: float
1403 ):
1404 """
1405 Queue a progress update that needs database access.
1406 For compatibility with old processor during migration.
1408 Args:
1409 username: The username
1410 research_id: The research ID
1411 progress: The progress value (0-100)
1412 """
1413 # In processor_v2, we can update directly if we have database access
1414 # or queue it for later processing
1415 operation_id = str(uuid.uuid4())
1416 with self._pending_operations_lock:
1417 self.pending_operations[operation_id] = {
1418 "username": username,
1419 "operation_type": "progress_update",
1420 "research_id": research_id,
1421 "progress": progress,
1422 "timestamp": time.time(),
1423 }
1424 logger.debug(
1425 f"Queued progress update for research {research_id}: {progress}%"
1426 )
1428 def queue_error_update(
1429 self,
1430 username: str,
1431 research_id: str,
1432 status: str,
1433 error_message: str,
1434 metadata: Dict[str, Any],
1435 completed_at: str,
1436 report_path: Optional[str] = None,
1437 ):
1438 """
1439 Queue an error status update that needs database access.
1440 For compatibility with old processor during migration.
1442 Args:
1443 username: The username
1444 research_id: The research ID
1445 status: The status to set (failed, suspended, etc.)
1446 error_message: The error message
1447 metadata: Research metadata
1448 completed_at: Completion timestamp
1449 report_path: Optional path to error report
1450 """
1451 operation_id = str(uuid.uuid4())
1452 with self._pending_operations_lock:
1453 self.pending_operations[operation_id] = {
1454 "username": username,
1455 "operation_type": "error_update",
1456 "research_id": research_id,
1457 "status": status,
1458 "error_message": error_message,
1459 "metadata": metadata,
1460 "completed_at": completed_at,
1461 "report_path": report_path,
1462 "timestamp": time.time(),
1463 }
1464 logger.info(
1465 f"Queued error update for research {research_id} with status {status}"
1466 )
1468 def process_pending_operations_for_user(
1469 self, username: str, db_session: Session
1470 ) -> int:
1471 """
1472 Process pending operations for a user when we have database access.
1473 Called from request context where encrypted database is accessible.
1474 For compatibility with old processor during migration.
1476 Args:
1477 username: Username to process operations for
1478 db_session: Active database session for the user
1480 Returns:
1481 Number of operations processed
1482 """
1483 # Find pending operations for this user (with lock)
1484 operations_to_process = []
1485 with self._pending_operations_lock:
1486 for op_id, op_data in list(self.pending_operations.items()):
1487 if op_data["username"] == username:
1488 operations_to_process.append((op_id, op_data))
1489 # Remove immediately to prevent duplicate processing
1490 del self.pending_operations[op_id]
1492 if not operations_to_process:
1493 return 0
1495 processed_count = 0
1497 # Process operations outside the lock (to avoid holding lock during DB operations)
1498 for op_id, op_data in operations_to_process:
1499 try:
1500 operation_type = op_data.get("operation_type")
1502 if operation_type == "progress_update":
1503 # Update progress in database
1504 from ...database.models import ResearchHistory
1506 research = (
1507 db_session.query(ResearchHistory)
1508 .filter_by(id=op_data["research_id"])
1509 .first()
1510 )
1511 if research: 1511 ↛ 1498line 1511 didn't jump to line 1498 because the condition on line 1511 was always true
1512 # Update the progress column directly
1513 research.progress = op_data["progress"]
1514 db_session.commit()
1515 processed_count += 1
1517 elif operation_type == "error_update": 1517 ↛ 1498line 1517 didn't jump to line 1498 because the condition on line 1517 was always true
1518 # Update error status in database
1519 from ...database.models import ResearchHistory
1521 research = (
1522 db_session.query(ResearchHistory)
1523 .filter_by(id=op_data["research_id"])
1524 .first()
1525 )
1526 if research: 1526 ↛ 1498line 1526 didn't jump to line 1498 because the condition on line 1526 was always true
1527 research.status = op_data["status"]
1528 research.error_message = op_data["error_message"]
1529 research.research_meta = op_data["metadata"]
1530 research.completed_at = op_data["completed_at"]
1531 if op_data.get("report_path"):
1532 research.report_path = op_data["report_path"]
1533 db_session.commit()
1534 processed_count += 1
1536 except Exception:
1537 logger.exception(f"Error processing operation {op_id}")
1538 # Rollback to clear the failed transaction state
1539 try:
1540 db_session.rollback()
1541 except Exception:
1542 logger.warning(
1543 f"Failed to rollback after error in operation {op_id}"
1544 )
1546 return processed_count
1549# Global queue processor instance
1550queue_processor = QueueProcessorV2()