diff --git a/backend/app/api/downloads.py b/backend/app/api/downloads.py
index 98fd48b..6b2437d 100644
--- a/backend/app/api/downloads.py
+++ b/backend/app/api/downloads.py
@@ -168,6 +168,45 @@ async def recent_activity():
})
+@bp.route("/reset-orphaned", methods=["POST"])
+async def reset_orphaned_downloads():
+ """Reset orphaned running downloads back to queued.
+
+ Jobs can get stuck in 'running' status if the worker crashes.
+ This endpoint resets them so they can be retried.
+ """
+ from app.tasks.maintenance import reset_orphaned_jobs
+
+ # Allow specifying threshold, default to 0 to reset ALL running jobs
+ data = await request.get_json() or {}
+ threshold_minutes = data.get("threshold_minutes", 0) # 0 = reset all running
+
+ try:
+ if threshold_minutes == 0:
+ # Reset all running jobs immediately (synchronous for immediate feedback)
+ from app.tasks.maintenance import _reset_all_running_jobs_async
+ import asyncio
+
+ # Run in the current event loop
+ result = await _reset_all_running_jobs_async()
+ current_app.logger.info(f"Manual reset of running jobs: {result}")
+ return jsonify({
+ "message": f"Reset {result['reset_count']} orphaned running jobs",
+ "reset_count": result['reset_count'],
+ })
+ else:
+ # Queue the task for threshold-based reset
+ task = reset_orphaned_jobs.delay(threshold_minutes)
+ return jsonify({
+ "message": "Orphaned job reset queued",
+ "task_id": task.id,
+ "threshold_minutes": threshold_minutes,
+ }), 202
+ except Exception as e:
+ current_app.logger.error(f"Failed to reset orphaned jobs: {e}")
+ return jsonify({"error": str(e)}), 500
+
+
@bp.route("/stats", methods=["GET"])
async def get_stats():
"""Get download statistics."""
@@ -208,4 +247,6 @@ async def get_stats():
"completed": status_counts.get(DownloadStatus.COMPLETED, 0),
"failed": status_counts.get(DownloadStatus.FAILED, 0),
"pending": status_counts.get(DownloadStatus.PENDING, 0),
+ "queued": status_counts.get(DownloadStatus.QUEUED, 0),
+ "running": status_counts.get(DownloadStatus.RUNNING, 0),
})
diff --git a/backend/app/services/gallery_dl.py b/backend/app/services/gallery_dl.py
index 0f24b46..0171a71 100644
--- a/backend/app/services/gallery_dl.py
+++ b/backend/app/services/gallery_dl.py
@@ -173,8 +173,15 @@ class GalleryDLService:
},
}
- def __init__(self):
+ def __init__(self, rate_limit: Optional[float] = None):
+ """Initialize the GalleryDL service.
+
+ Args:
+ rate_limit: Override for download rate limit (seconds between requests).
+ If None, uses environment default.
+ """
self.settings = get_settings()
+ self._rate_limit = rate_limit if rate_limit is not None else self.settings.download_rate_limit
self._base_config = self._load_base_config()
def _load_base_config(self) -> dict:
@@ -198,8 +205,8 @@ class GalleryDLService:
"base-directory": self.settings.download_path,
"archive": str(Path(self.settings.config_path) / "archive.sqlite3"),
"skip": True,
- "sleep": self.settings.download_rate_limit,
- "sleep-request": max(1.0, self.settings.download_rate_limit / 2),
+ "sleep": self._rate_limit,
+ "sleep-request": max(1.0, self._rate_limit / 2),
"retries": 3,
"timeout": 30.0,
"verify": True,
@@ -311,19 +318,23 @@ class GalleryDLService:
"""Categorize the error based on output and return code."""
combined = f"{stdout} {stderr}".lower()
- # Check for "no new content" first - skipped files with return code 1 is normal
- if return_code == 1:
- # Check if output indicates content was processed (skipped = already downloaded)
- if "skipping" in combined or "skip" in combined:
- return ErrorType.NO_NEW_CONTENT, "No new content to download"
- # Empty output with return code 1 often means nothing to download
- if not stdout.strip() or "no results" in combined:
+ # Check for "no new content" FIRST for any non-zero return code
+ # Gallery-dl returns 1 when all files are skipped or no new content found
+ # This check must run before other error checks to avoid false positives
+ skip_indicators = ["skipping", "skip", "already exists", "archive"]
+ if any(indicator in combined for indicator in skip_indicators):
+ # If we see skip messages and no actual error messages, it's just "no new content"
+ actual_errors = ["error", "failed", "exception", "traceback"]
+ if not any(err in combined for err in actual_errors):
return ErrorType.NO_NEW_CONTENT, "No new content to download"
- # Auth errors - be more specific to avoid false positives from debug URLs
- # Look for actual error messages, not just keywords that appear in URLs
- auth_patterns = [
- "401",
+ # Empty output with low return codes often means nothing to download
+ if return_code in (0, 1) and not stdout.strip():
+ return ErrorType.NO_NEW_CONTENT, "No new content to download"
+
+ # Auth errors - look for error context, not just HTTP codes that appear in debug logs
+ # HTTP debug lines look like: '"GET /path HTTP/1.1" 401 None' - we need to detect actual errors
+ auth_error_patterns = [
"unauthorized",
"login required",
"please login",
@@ -332,28 +343,84 @@ class GalleryDLService:
"session expired",
"invalid cookie",
"cookies are expired",
+ "cookies have expired",
+ "not logged in",
]
- if any(pattern in combined for pattern in auth_patterns):
+ # Check for 401 specifically as an HTTP error response (not in URLs)
+ # Look for patterns like "401" followed by error context or at end of HTTP log line
+ has_401_error = (
+ '" 401 ' in combined or # HTTP response: "GET /path HTTP/1.1" 401 None
+ "401 unauthorized" in combined or
+ "error 401" in combined or
+ "status: 401" in combined or
+ "status code: 401" in combined
+ )
+ if has_401_error or any(pattern in combined for pattern in auth_error_patterns):
return ErrorType.AUTH_ERROR, "Authentication failed - cookies may be expired or invalid"
- if "429" in combined or "rate limit" in combined or "too many requests" in combined:
+ # Rate limiting - look for actual rate limit errors
+ rate_limit_patterns = [
+ '" 429 ', # HTTP response
+ "rate limit",
+ "too many requests",
+ "ratelimit",
+ "throttl",
+ ]
+ if any(pattern in combined for pattern in rate_limit_patterns):
return ErrorType.RATE_LIMITED, "Rate limited by server - try increasing sleep time"
- if "404" in combined or "not found" in combined:
+ # 404 Not Found - check for HTTP 404 responses, not just "404" anywhere
+ not_found_patterns = [
+ '" 404 ', # HTTP response
+ "error 404",
+ "status: 404",
+ "not found",
+ "does not exist",
+ "no longer available",
+ ]
+ if any(pattern in combined for pattern in not_found_patterns):
return ErrorType.NOT_FOUND, "URL not found - artist may have changed username or deleted content"
- if "timeout" in combined or "connection" in combined or "network" in combined:
+ # Network errors
+ network_error_patterns = [
+ "timeout",
+ "timed out",
+ "connection refused",
+ "connection reset",
+ "network unreachable",
+ "name resolution",
+ "dns",
+ "ssl error",
+ "certificate verify",
+ ]
+ if any(pattern in combined for pattern in network_error_patterns):
return ErrorType.NETWORK_ERROR, "Network error - check internet connection"
- if "403" in combined or "forbidden" in combined or "access denied" in combined:
+ # 403 Forbidden / Access denied
+ access_denied_patterns = [
+ '" 403 ', # HTTP response
+ "error 403",
+ "forbidden",
+ "access denied",
+ "permission denied",
+ "tier required",
+ "pledge required",
+ ]
+ if any(pattern in combined for pattern in access_denied_patterns):
return ErrorType.ACCESS_DENIED, "Access denied - may need higher subscription tier"
- if "http error" in combined:
+ # Generic HTTP errors
+ if "http error" in combined or "httperror" in combined:
return ErrorType.HTTP_ERROR, "HTTP error occurred"
- if "no suitable" in combined or "unsupported" in combined:
+ # Unsupported URL
+ if "no suitable" in combined or "unsupported" in combined or "no extractor" in combined:
return ErrorType.UNSUPPORTED_URL, "URL format not supported by gallery-dl"
+ # If we got here with return code 1 and saw some activity, it might just be no new content
+ if return_code == 1:
+ return ErrorType.NO_NEW_CONTENT, "No new content to download"
+
return ErrorType.UNKNOWN_ERROR, f"Unknown error (return code: {return_code})"
def _count_downloaded_files(self, stdout: str) -> int:
@@ -575,6 +642,6 @@ class GalleryDLService:
"content_types": defaults.get("content_types", ["all"]),
"directory_pattern": defaults.get("directory"),
"filename_pattern": defaults.get("filename"),
- "sleep": self.settings.download_rate_limit,
- "sleep_request": max(1.0, self.settings.download_rate_limit / 2),
+ "sleep": self._rate_limit,
+ "sleep_request": max(1.0, self._rate_limit / 2),
}
diff --git a/backend/app/tasks/celery_app.py b/backend/app/tasks/celery_app.py
index 8aa27d1..406dde1 100644
--- a/backend/app/tasks/celery_app.py
+++ b/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}")
diff --git a/backend/app/tasks/downloads.py b/backend/app/tasks/downloads.py
index 1d79348..6673ed7 100644
--- a/backend/app/tasks/downloads.py
+++ b/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
diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py
index 5934f1a..56286b4 100644
--- a/backend/app/tasks/maintenance.py
+++ b/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
diff --git a/frontend/src/services/api.js b/frontend/src/services/api.js
index d7750fe..5874318 100644
--- a/frontend/src/services/api.js
+++ b/frontend/src/services/api.js
@@ -37,6 +37,7 @@ export const downloadsApi = {
retry: (id) => api.post(`/downloads/${id}/retry`),
stats: (params = {}) => api.get('/downloads/stats', { params }),
recentActivity: (params = {}) => api.get('/downloads/recent-activity', { params }),
+ resetOrphaned: (thresholdMinutes = 0) => api.post('/downloads/reset-orphaned', { threshold_minutes: thresholdMinutes }),
}
// Credentials API
diff --git a/frontend/src/views/Dashboard.vue b/frontend/src/views/Dashboard.vue
index 108a671..1bea648 100644
--- a/frontend/src/views/Dashboard.vue
+++ b/frontend/src/views/Dashboard.vue
@@ -51,8 +51,11 @@