From c7a259a63b269a9f842a9c5013e4001acfa08156 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 27 Jan 2026 18:05:54 -0500 Subject: [PATCH] correcting worker db connection issue --- backend/app/tasks/downloads.py | 221 ++++++++++++++++++--------------- 1 file changed, 123 insertions(+), 98 deletions(-) diff --git a/backend/app/tasks/downloads.py b/backend/app/tasks/downloads.py index ffffe71..1d79348 100644 --- a/backend/app/tasks/downloads.py +++ b/backend/app/tasks/downloads.py @@ -44,6 +44,9 @@ async def _cleanup_engine(engine): async def _download_source_async(source_id: int, download_id: int = None) -> dict: """Async implementation of source download. + Uses separate DB sessions for setup and result writing to avoid + connection timeouts during long-running gallery-dl downloads. + Args: source_id: ID of the source to download download_id: Optional ID of a pre-created Download record (QUEUED state) @@ -51,6 +54,7 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic Returns: Dict with download results """ + # --- Phase 1: Setup (read source info, mark as RUNNING, get credentials) --- session_factory, engine = get_async_session() try: @@ -69,7 +73,6 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic if not source.enabled: logger.info(f"Source {source.subscription.name}/{source.platform} is disabled, skipping") - # If there's a queued download record, mark it as skipped if download_id: dl_result = await session.execute( select(Download).where(Download.id == download_id) @@ -80,8 +83,12 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic await session.commit() return {"source_id": source_id, "status": "skipped", "reason": "disabled"} + # Capture values we need after the session closes subscription_name = source.subscription.name - logger.info(f"Starting download for {subscription_name}/{source.platform}") + source_url = source.url + source_platform = source.platform + source_metadata = dict(source.metadata_ or {}) + logger.info(f"Starting download for {subscription_name}/{source_platform}") # Use existing download record or create a new one if download_id: @@ -98,8 +105,8 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic if not download_id or download is None: download = Download( - source_id=source.id, - url=source.url, + source_id=source_id, + url=source_url, status=DownloadStatus.RUNNING, started_at=utcnow(), ) @@ -107,131 +114,149 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic await session.commit() await session.refresh(download) + actual_download_id = download.id # Broadcast download started event publish_event("download.started", { - "download_id": download.id, - "source_id": source.id, + "download_id": actual_download_id, + "source_id": source_id, "subscription_name": subscription_name, - "platform": source.platform, - "url": source.url, + "platform": source_platform, + "url": source_url, }) - try: - # Get credentials - cred_manager = CredentialManager(session) - cookies_path = None - auth_token = None + # Get credentials before closing session + cred_manager = CredentialManager(session) + cookies_path = None + auth_token = None - # Discord uses token auth, others use cookies - if source.platform == "discord": - auth_token = await cred_manager.get_token(source.platform) - if not auth_token: - logger.warning(f"No token for {source.platform}, download may fail") - else: - cookies_path = await cred_manager.get_cookies_path(source.platform) - if not cookies_path and source.platform in ["patreon", "subscribestar"]: - logger.warning(f"No credentials for {source.platform}, download may fail") + if source_platform == "discord": + auth_token = await cred_manager.get_token(source_platform) + if not auth_token: + logger.warning(f"No token for {source_platform}, download may fail") + else: + cookies_path = await cred_manager.get_cookies_path(source_platform) + if not cookies_path and source_platform in ["patreon", "subscribestar"]: + logger.warning(f"No credentials for {source_platform}, download may fail") - # Build source config from metadata - source_config = SourceConfig.from_dict(source.metadata_ or {}) + # Session is now closed - DB connection released + finally: + await _cleanup_engine(engine) - # Execute download - gdl_service = GalleryDLService() - dl_result = await gdl_service.download( - url=source.url, - subscription_name=subscription_name, - platform=source.platform, - source_config=source_config, - cookies_path=cookies_path, - auth_token=auth_token, - ) + # --- Phase 2: Execute download (no DB connection held) --- + source_config = SourceConfig.from_dict(source_metadata) + dl_result = None + dl_error = None - # Update download record - download.completed_at = utcnow() - download.file_count = dl_result.files_downloaded - download.metadata_ = { - "stdout": dl_result.stdout[:10000] if dl_result.stdout else None, # Truncate - "stderr": dl_result.stderr[:5000] if dl_result.stderr else None, - "duration_seconds": dl_result.duration_seconds, - } + try: + gdl_service = GalleryDLService() + dl_result = await gdl_service.download( + url=source_url, + subscription_name=subscription_name, + platform=source_platform, + source_config=source_config, + cookies_path=cookies_path, + auth_token=auth_token, + ) + except Exception as e: + logger.exception(f"Unexpected error downloading {subscription_name}/{source_platform}: {e}") + dl_error = e - if dl_result.success: - download.status = DownloadStatus.COMPLETED - source.last_success = utcnow() - source.error_count = 0 - logger.info(f"Download completed for {subscription_name}/{source.platform}: {dl_result.files_downloaded} files") + # --- Phase 3: Write results (fresh DB session) --- + session_factory2, engine2 = get_async_session() - # Broadcast download completed event - publish_event("download.completed", { - "download_id": download.id, - "source_id": source.id, - "subscription_name": subscription_name, - "platform": source.platform, - "file_count": dl_result.files_downloaded, - "duration_seconds": dl_result.duration_seconds, - }) - else: - download.status = DownloadStatus.FAILED - download.error_type = dl_result.error_type.value if dl_result.error_type else None - download.error_message = dl_result.error_message - source.error_count = (source.error_count or 0) + 1 - logger.warning(f"Download failed for {subscription_name}/{source.platform}: {dl_result.error_message}") + try: + async with session_factory2() as session: + # Re-fetch download and source with fresh session + download = (await session.execute( + select(Download).where(Download.id == actual_download_id) + )).scalar() + source = (await session.execute( + select(Source).where(Source.id == source_id) + )).scalar() - # Broadcast download failed event - publish_event("download.failed", { - "download_id": download.id, - "source_id": source.id, - "subscription_name": subscription_name, - "platform": source.platform, - "error_type": dl_result.error_type.value if dl_result.error_type else None, - "error_message": dl_result.error_message, - }) - - # Update source last_check - source.last_check = utcnow() - await session.commit() - - return { - "source_id": source_id, - "download_id": download.id, - "status": "completed" if dl_result.success else "failed", - "files_downloaded": dl_result.files_downloaded, - "error_type": dl_result.error_type.value if dl_result.error_type else None, - "error_message": dl_result.error_message, - "duration_seconds": dl_result.duration_seconds, - } - - except Exception as e: - logger.exception(f"Unexpected error downloading {subscription_name}/{source.platform}: {e}") + if not download or not source: + logger.error(f"Could not re-fetch download {actual_download_id} or source {source_id} for result update") + return {"source_id": source_id, "status": "error", "error": "Lost DB records"} + if dl_error: download.status = DownloadStatus.FAILED download.error_type = DBErrorType.UNKNOWN_ERROR - download.error_message = str(e) + download.error_message = str(dl_error) download.completed_at = utcnow() source.error_count = (source.error_count or 0) + 1 source.last_check = utcnow() - await session.commit() - # Broadcast download failed event publish_event("download.failed", { - "download_id": download.id, - "source_id": source.id, + "download_id": actual_download_id, + "source_id": source_id, "subscription_name": subscription_name, - "platform": source.platform, + "platform": source_platform, "error_type": "unknown_error", - "error_message": str(e), + "error_message": str(dl_error), }) return { "source_id": source_id, - "download_id": download.id, + "download_id": actual_download_id, "status": "error", - "error": str(e), + "error": str(dl_error), } + + # Update download record with results + download.completed_at = utcnow() + download.file_count = dl_result.files_downloaded + download.metadata_ = { + "stdout": dl_result.stdout[:10000] if dl_result.stdout else None, + "stderr": dl_result.stderr[:5000] if dl_result.stderr else None, + "duration_seconds": dl_result.duration_seconds, + } + + if dl_result.success: + download.status = DownloadStatus.COMPLETED + source.last_success = utcnow() + source.error_count = 0 + logger.info(f"Download completed for {subscription_name}/{source_platform}: {dl_result.files_downloaded} files") + + publish_event("download.completed", { + "download_id": actual_download_id, + "source_id": source_id, + "subscription_name": subscription_name, + "platform": source_platform, + "file_count": dl_result.files_downloaded, + "duration_seconds": dl_result.duration_seconds, + }) + else: + download.status = DownloadStatus.FAILED + download.error_type = dl_result.error_type.value if dl_result.error_type else None + download.error_message = dl_result.error_message + source.error_count = (source.error_count or 0) + 1 + logger.warning(f"Download failed for {subscription_name}/{source_platform}: {dl_result.error_message}") + + publish_event("download.failed", { + "download_id": actual_download_id, + "source_id": source_id, + "subscription_name": subscription_name, + "platform": source_platform, + "error_type": dl_result.error_type.value if dl_result.error_type else None, + "error_message": dl_result.error_message, + }) + + source.last_check = utcnow() + await session.commit() + + return { + "source_id": source_id, + "download_id": actual_download_id, + "status": "completed" if dl_result.success else "failed", + "files_downloaded": dl_result.files_downloaded, + "error_type": dl_result.error_type.value if dl_result.error_type else None, + "error_message": dl_result.error_message, + "duration_seconds": dl_result.duration_seconds, + } finally: - await _cleanup_engine(engine) + await _cleanup_engine(engine2) async def _scheduled_check_async() -> dict: