ongoing polish in queue handling

backend/app/tasks/celery_app.py

@@ -1,12 +1,51 @@
 """Celery application configuration."""
 
+import logging
 from celery import Celery
 from celery.schedules import crontab
 
 from app.config import get_settings
 
+logger = logging.getLogger(__name__)
 settings = get_settings()
 
+
+def _get_parallel_limit_from_db():
+    """Try to get parallel_limit from database, fall back to environment.
+
+    This is called once at Celery startup. Changes require worker restart.
+    """
+    try:
+        import asyncio
+        from sqlalchemy import select
+        from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
+
+        async def _fetch():
+            engine = create_async_engine(settings.async_database_url, echo=False)
+            session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+            try:
+                async with session_factory() as session:
+                    from app.models.setting import Setting, DEFAULT_SETTINGS
+                    result = await session.execute(
+                        select(Setting).where(Setting.key == "download.parallel_limit")
+                    )
+                    setting = result.scalar()
+                    if setting:
+                        return setting.value
+                    return DEFAULT_SETTINGS.get("download.parallel_limit", settings.download_parallel_limit)
+            finally:
+                await engine.dispose()
+
+        return asyncio.run(_fetch())
+    except Exception as e:
+        logger.warning(f"Could not read parallel_limit from database, using environment: {e}")
+        return settings.download_parallel_limit
+
+
+# Get parallel limit (database takes precedence over environment)
+parallel_limit = _get_parallel_limit_from_db()
+logger.info(f"Celery worker concurrency set to: {parallel_limit}")
+
 celery_app = Celery(
     "gallery_subscriber",
     broker=settings.redis_url,
@@ -24,7 +63,7 @@ celery_app.conf.update(
     task_track_started=True,
     task_time_limit=3600,  # 1 hour max per task
     worker_prefetch_multiplier=1,  # Process one task at a time
-    worker_concurrency=settings.download_parallel_limit,
+    worker_concurrency=parallel_limit,
 )
 
 # Celery Beat schedule for periodic tasks
@@ -44,4 +83,32 @@ celery_app.conf.beat_schedule = {
         "task": "tasks.update_storage_stats",
         "schedule": crontab(minute="*/30"),  # Every 30 minutes
     },
+    "reset-orphaned-jobs": {
+        "task": "tasks.reset_orphaned_jobs",
+        "schedule": crontab(minute="*/15"),  # Every 15 minutes
+    },
 }
+
+
+# Worker startup signal - reset any orphaned running jobs
+@celery_app.on_after_configure.connect
+def setup_worker_startup(sender, **kwargs):
+    """Called after Celery is configured but before workers start processing."""
+    pass  # Placeholder - actual reset happens in worker_ready signal
+
+
+from celery.signals import worker_ready
+
+@worker_ready.connect
+def on_worker_ready(sender, **kwargs):
+    """Called when a worker is ready to accept tasks.
+
+    Resets any jobs stuck in 'running' status from previous runs.
+    This handles the case where workers crashed or were restarted.
+    """
+    try:
+        from app.tasks.maintenance import reset_all_running_jobs_sync
+        result = reset_all_running_jobs_sync()
+        logger.info(f"Worker startup: {result}")
+    except Exception as e:
+        logger.error(f"Failed to reset orphaned jobs on worker startup: {e}")

backend/app/tasks/downloads.py

@@ -15,6 +15,7 @@ from app.models.base import utcnow
 from app.models.source import Source
 from app.models.subscription import Subscription
 from app.models.download import Download, DownloadStatus, ErrorType as DBErrorType
+from app.models.setting import Setting, DEFAULT_SETTINGS
 from app.services.gallery_dl import GalleryDLService, SourceConfig, DownloadResult
 from app.services.credential_manager import CredentialManager
 from app.events import publish_event
@@ -41,6 +42,46 @@ async def _cleanup_engine(engine):
     await engine.dispose()
 
 
+async def _get_db_setting(session: AsyncSession, key: str, default=None):
+    """Get a setting value from the database, falling back to defaults.
+
+    Args:
+        session: Database session
+        key: Setting key (e.g., 'download.rate_limit')
+        default: Fallback if not in database or DEFAULT_SETTINGS
+
+    Returns:
+        The setting value
+    """
+    result = await session.execute(
+        select(Setting).where(Setting.key == key)
+    )
+    setting = result.scalar()
+
+    if setting:
+        return setting.value
+
+    # Fall back to DEFAULT_SETTINGS, then to provided default
+    return DEFAULT_SETTINGS.get(key, default)
+
+
+def _get_db_setting_sync(key: str, default=None):
+    """Synchronously get a setting value from the database.
+
+    This is used for getting settings in Celery task retry logic.
+    Creates a fresh connection for each call.
+    """
+    async def _fetch():
+        session_factory, engine = get_async_session()
+        try:
+            async with session_factory() as session:
+                return await _get_db_setting(session, key, default)
+        finally:
+            await _cleanup_engine(engine)
+
+    return asyncio.run(_fetch())
+
+
 async def _download_source_async(source_id: int, download_id: int = None) -> dict:
     """Async implementation of source download.
 
@@ -139,6 +180,9 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic
                 if not cookies_path and source_platform in ["patreon", "subscribestar"]:
                     logger.warning(f"No credentials for {source_platform}, download may fail")
 
+            # Get rate limit from database settings
+            db_rate_limit = await _get_db_setting(session, "download.rate_limit", settings.download_rate_limit)
+
         # Session is now closed - DB connection released
     finally:
         await _cleanup_engine(engine)
@@ -149,7 +193,7 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic
     dl_error = None
 
     try:
-        gdl_service = GalleryDLService()
+        gdl_service = GalleryDLService(rate_limit=db_rate_limit)
         dl_result = await gdl_service.download(
             url=source_url,
             subscription_name=subscription_name,
@@ -271,9 +315,16 @@ async def _scheduled_check_async() -> dict:
         async with session_factory() as session:
             now = utcnow()
 
+            # Get global schedule interval from database settings
+            # This is the primary interval that controls how often sources are checked
+            global_interval = await _get_db_setting(
+                session, "download.schedule_interval", 3600  # Default 1 hour
+            )
+            logger.debug(f"Using global schedule interval: {global_interval} seconds")
+
             # Find sources due for checking:
             # - enabled = True
-            # - last_check is NULL OR (now - last_check) > check_interval
+            # - last_check is NULL OR (now - last_check) > global_interval
             # Order by subscription priority, then by when last checked
             query = (
                 select(Source)
@@ -298,14 +349,15 @@ async def _scheduled_check_async() -> dict:
             result = await session.execute(query)
             sources = result.scalars().all()
 
-            # Filter by check_interval
+            # Filter by global schedule interval from Settings
+            # All sources use the same interval - no per-source overrides
             due_sources = []
             for source in sources:
                 if source.last_check is None:
                     due_sources.append(source)
                 else:
                     time_since_check = (now - source.last_check).total_seconds()
-                    if time_since_check >= source.check_interval:
+                    if time_since_check >= global_interval:
                         due_sources.append(source)
 
             logger.info(f"Found {len(due_sources)} sources due for checking")
@@ -403,7 +455,7 @@ async def _cleanup_old_downloads_async() -> dict:
 
 # Celery tasks that wrap async functions
 
-@celery_app.task(bind=True, max_retries=3)
+@celery_app.task(bind=True)
 def download_source(self, source_id: int, download_id: int = None):
     """Download new content from a single source.
 
@@ -415,8 +467,14 @@ def download_source(self, source_id: int, download_id: int = None):
         return asyncio.run(_download_source_async(source_id, download_id=download_id))
     except Exception as e:
         logger.exception(f"Task failed for source {source_id}: {e}")
-        # Retry with exponential backoff
-        raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries))
+        # Get retry count from database settings
+        max_retries = _get_db_setting_sync("download.retry_count", 3)
+        if self.request.retries < max_retries:
+            # Retry with exponential backoff
+            raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries), max_retries=max_retries)
+        else:
+            logger.error(f"Max retries ({max_retries}) exceeded for source {source_id}")
+            raise
 
 
 @celery_app.task
@@ -437,7 +495,7 @@ def cleanup_old_downloads():
     return asyncio.run(_cleanup_old_downloads_async())
 
 
-@celery_app.task(bind=True, max_retries=2)
+@celery_app.task(bind=True)
 def process_download(self, download_id: int):
     """Process a single download record (for retries).
 
@@ -469,4 +527,10 @@ def process_download(self, download_id: int):
         return asyncio.run(_process())
     except Exception as e:
         logger.exception(f"Task failed for download {download_id}: {e}")
-        raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries))
+        # Get retry count from database settings
+        max_retries = _get_db_setting_sync("download.retry_count", 3)
+        if self.request.retries < max_retries:
+            raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries), max_retries=max_retries)
+        else:
+            logger.error(f"Max retries ({max_retries}) exceeded for download {download_id}")
+            raise

backend/app/tasks/maintenance.py

@@ -3,20 +3,26 @@
 import asyncio
 import logging
 import os
-from datetime import datetime
+from datetime import datetime, timedelta
 from pathlib import Path
 
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 
 from app.config import get_settings
 from app.tasks.celery_app import celery_app
 from app.models.setting import Setting, STORAGE_STATS_SETTING
+from app.models.download import Download, DownloadStatus
+from app.models.base import utcnow
 
 logger = logging.getLogger(__name__)
 
 settings = get_settings()
 
+# Maximum time a job can be in "running" state before considered orphaned
+# This should be longer than the task_time_limit (1 hour) to account for legitimate long runs
+STALE_RUNNING_THRESHOLD_MINUTES = 90
+
 
 def get_async_session():
     """Get async session factory for Celery tasks."""
@@ -125,3 +131,114 @@ def update_storage_stats() -> dict:
     result = asyncio.run(_update_storage_stats_async())
     logger.info(f"Storage stats updated: {result}")
     return result
+
+
+async def _reset_orphaned_running_jobs_async(threshold_minutes: int = None) -> dict:
+    """Reset jobs stuck in 'running' status back to 'queued'.
+
+    Jobs can get stuck in 'running' status if the worker crashes or restarts
+    while processing them. This function detects and resets these orphaned jobs.
+
+    Args:
+        threshold_minutes: Jobs running longer than this are considered orphaned.
+                          If None, uses STALE_RUNNING_THRESHOLD_MINUTES.
+
+    Returns:
+        Dict with count of reset jobs
+    """
+    if threshold_minutes is None:
+        threshold_minutes = STALE_RUNNING_THRESHOLD_MINUTES
+
+    session_factory, engine = get_async_session()
+
+    try:
+        async with session_factory() as session:
+            now = utcnow()
+            cutoff = now - timedelta(minutes=threshold_minutes)
+
+            # Find running jobs that started before the cutoff
+            # OR running jobs with no started_at (shouldn't happen but handle it)
+            query = select(Download).where(
+                Download.status == DownloadStatus.RUNNING,
+                (Download.started_at < cutoff) | (Download.started_at == None)
+            )
+
+            result = await session.execute(query)
+            orphaned_jobs = result.scalars().all()
+
+            reset_count = 0
+            for job in orphaned_jobs:
+                job.status = DownloadStatus.QUEUED
+                job.started_at = None
+                job.error_message = f"Reset from orphaned running state (was running since {job.started_at})"
+                reset_count += 1
+                logger.info(f"Reset orphaned job {job.id} (URL: {job.url[:50]}...) to QUEUED")
+
+            if reset_count > 0:
+                await session.commit()
+                logger.info(f"Reset {reset_count} orphaned running jobs to QUEUED")
+            else:
+                logger.debug("No orphaned running jobs found")
+
+            return {"reset_count": reset_count, "threshold_minutes": threshold_minutes}
+
+    finally:
+        await _cleanup_engine(engine)
+
+
+async def _reset_all_running_jobs_async() -> dict:
+    """Reset ALL jobs in 'running' status back to 'queued'.
+
+    This is useful on worker startup to clean up any jobs that were
+    interrupted by a previous shutdown/crash.
+
+    Returns:
+        Dict with count of reset jobs
+    """
+    session_factory, engine = get_async_session()
+
+    try:
+        async with session_factory() as session:
+            # Find all running jobs
+            query = select(Download).where(Download.status == DownloadStatus.RUNNING)
+            result = await session.execute(query)
+            running_jobs = result.scalars().all()
+
+            reset_count = 0
+            for job in running_jobs:
+                job.status = DownloadStatus.QUEUED
+                job.started_at = None
+                job.error_message = "Reset on worker startup"
+                reset_count += 1
+                logger.info(f"Reset running job {job.id} to QUEUED on startup")
+
+            if reset_count > 0:
+                await session.commit()
+                logger.info(f"Reset {reset_count} running jobs to QUEUED on worker startup")
+
+            return {"reset_count": reset_count}
+
+    finally:
+        await _cleanup_engine(engine)
+
+
+@celery_app.task(name="tasks.reset_orphaned_jobs")
+def reset_orphaned_jobs(threshold_minutes: int = None) -> dict:
+    """Celery task to reset orphaned running jobs.
+
+    Jobs that have been in 'running' status for longer than the threshold
+    are considered orphaned (worker probably crashed) and reset to 'queued'.
+    """
+    logger.info("Checking for orphaned running jobs...")
+    result = asyncio.run(_reset_orphaned_running_jobs_async(threshold_minutes))
+    return result
+
+
+def reset_all_running_jobs_sync() -> dict:
+    """Synchronous function to reset all running jobs.
+
+    Called on worker startup before Celery is fully initialized.
+    """
+    logger.info("Resetting all running jobs on worker startup...")
+    result = asyncio.run(_reset_all_running_jobs_async())
+    return result