diff --git a/.env.example b/.env.example
index 3c22283..c99b1d9 100644
--- a/.env.example
+++ b/.env.example
@@ -59,6 +59,7 @@ DB_NAME=gallery_subscriber
REDIS_HOST=redis
REDIS_PORT=6379
+REDIS_DB=0
# ===========================================
# Data Storage Paths
@@ -84,3 +85,6 @@ REDIS_PORT=6379
# Only set these directly for special configurations
# DATABASE_URL=postgresql://gdl:password@localhost:5432/gallery_subscriber
# REDIS_URL=redis://localhost:6379/0
+#
+# IMPORTANT: All services (app, scheduler, worker) MUST use the same Redis database.
+# If you change REDIS_DB, ensure all containers are restarted together.
diff --git a/README.md b/README.md
index 96661f8..06c56d7 100644
--- a/README.md
+++ b/README.md
@@ -92,7 +92,7 @@ docker compose exec app alembic upgrade head
| `PORT` | External port | `8080` |
| `DOWNLOAD_PARALLEL_LIMIT` | Max concurrent downloads | `3` |
| `DOWNLOAD_RATE_LIMIT` | Delay between requests (seconds) | `3.0` |
-| `DEFAULT_CHECK_INTERVAL` | Auto-check interval (seconds) | `3600` |
+| `DEFAULT_CHECK_INTERVAL` | Auto-check interval (seconds) | `28800` (8 hours) |
| `LOG_LEVEL` | Logging verbosity | `INFO` |
See `.env.example` for the complete list with descriptions.
diff --git a/backend/alembic/versions/001_initial_schema.py b/backend/alembic/versions/001_initial_schema.py
index b43fa15..88de8d9 100644
--- a/backend/alembic/versions/001_initial_schema.py
+++ b/backend/alembic/versions/001_initial_schema.py
@@ -27,7 +27,7 @@ def upgrade() -> None:
sa.Column('url', sa.Text(), nullable=False),
sa.Column('enabled', sa.Boolean(), default=True),
sa.Column('priority', sa.Integer(), default=0),
- sa.Column('check_interval', sa.Integer(), default=3600),
+ sa.Column('check_interval', sa.Integer(), default=28800),
sa.Column('last_check', sa.DateTime(), nullable=True),
sa.Column('last_success', sa.DateTime(), nullable=True),
sa.Column('error_count', sa.Integer(), default=0),
@@ -115,7 +115,7 @@ def upgrade() -> None:
('download.parallel_limit', '3'),
('download.rate_limit', '3.0'),
('download.retry_count', '3'),
- ('download.schedule_interval', '3600'),
+ ('download.schedule_interval', '28800'),
('notification.enabled', 'false'),
('notification.webhook_url', 'null')
""")
diff --git a/backend/app/api/downloads.py b/backend/app/api/downloads.py
index 6b2437d..7b32901 100644
--- a/backend/app/api/downloads.py
+++ b/backend/app/api/downloads.py
@@ -207,6 +207,41 @@ async def reset_orphaned_downloads():
return jsonify({"error": str(e)}), 500
+@bp.route("/requeue-stale", methods=["POST"])
+async def requeue_stale_downloads():
+ """Re-queue Celery tasks for downloads stuck in 'queued' status.
+
+ Downloads can get stuck in 'queued' database status if the Celery task
+ was lost (e.g., sent to wrong Redis DB, Redis restart, etc.).
+ This endpoint re-sends the Celery tasks for those jobs.
+ """
+ from app.tasks.maintenance import requeue_stale_jobs, _requeue_all_queued_jobs_async
+
+ data = await request.get_json() or {}
+ threshold_minutes = data.get("threshold_minutes", 0) # 0 = re-queue ALL queued jobs
+
+ try:
+ if threshold_minutes == 0:
+ # Re-queue all queued jobs immediately (synchronous for immediate feedback)
+ result = await _requeue_all_queued_jobs_async()
+ current_app.logger.info(f"Manual re-queue of all queued jobs: {result}")
+ return jsonify({
+ "message": f"Re-queued {result['requeued_count']} queued jobs",
+ "requeued_count": result['requeued_count'],
+ })
+ else:
+ # Queue the task for threshold-based re-queue
+ task = requeue_stale_jobs.delay(threshold_minutes)
+ return jsonify({
+ "message": "Stale job re-queue queued",
+ "task_id": task.id,
+ "threshold_minutes": threshold_minutes,
+ }), 202
+ except Exception as e:
+ current_app.logger.error(f"Failed to re-queue stale jobs: {e}")
+ return jsonify({"error": str(e)}), 500
+
+
@bp.route("/stats", methods=["GET"])
async def get_stats():
"""Get download statistics."""
diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py
index 367f039..0a5ac4d 100644
--- a/backend/app/api/settings.py
+++ b/backend/app/api/settings.py
@@ -118,8 +118,10 @@ async def update_settings():
setting = result.scalar()
if setting:
+ current_app.logger.info(f"Updating setting '{key}': {setting.value} -> {value} (type: {type(value).__name__})")
setting.value = value
else:
+ current_app.logger.info(f"Creating setting '{key}': {value} (type: {type(value).__name__})")
setting = Setting(key=key, value=value)
session.add(setting)
@@ -321,3 +323,49 @@ async def get_system_info():
"download_path": config.download_path,
"config_path": config.config_path,
})
+
+
+@bp.route("/debug/schedule-interval", methods=["GET"])
+async def debug_schedule_interval():
+ """Debug endpoint to verify schedule_interval setting storage and retrieval.
+
+ Returns detailed information about how the setting is stored and what value
+ the worker would read.
+ """
+ async with current_app.db_engine.connect() as conn:
+ session = AsyncSession(bind=conn)
+
+ # Query the raw setting from database
+ result = await session.execute(
+ select(Setting).where(Setting.key == "download.schedule_interval")
+ )
+ setting = result.scalar()
+
+ if setting:
+ raw_value = setting.value
+ db_info = {
+ "found_in_database": True,
+ "raw_value": raw_value,
+ "raw_type": type(raw_value).__name__,
+ "as_int": int(raw_value) if raw_value is not None else None,
+ "updated_at": setting.updated_at.isoformat() if setting.updated_at else None,
+ }
+ else:
+ db_info = {
+ "found_in_database": False,
+ "raw_value": None,
+ "raw_type": None,
+ "as_int": None,
+ "updated_at": None,
+ }
+
+ # What the worker would use
+ fallback_value = DEFAULT_SETTINGS.get("download.schedule_interval", 3600)
+ worker_would_use = db_info["as_int"] if db_info["found_in_database"] else fallback_value
+
+ return jsonify({
+ "database": db_info,
+ "default_settings_value": fallback_value,
+ "worker_would_use": worker_would_use,
+ "worker_would_use_hours": worker_would_use / 3600 if worker_would_use else None,
+ })
diff --git a/backend/app/api/sources.py b/backend/app/api/sources.py
index 662bd89..ca70acd 100644
--- a/backend/app/api/sources.py
+++ b/backend/app/api/sources.py
@@ -105,7 +105,7 @@ async def create_source():
# Get default check interval from database settings
default_interval = await get_setting_value(
- session, "download.schedule_interval", 3600
+ session, "download.schedule_interval", 28800
)
source = Source(
diff --git a/backend/app/api/subscriptions.py b/backend/app/api/subscriptions.py
index b6bd8b6..af38375 100644
--- a/backend/app/api/subscriptions.py
+++ b/backend/app/api/subscriptions.py
@@ -89,7 +89,7 @@ async def create_subscription():
# Get default check interval from database settings
default_interval = await get_setting_value(
- session, "download.schedule_interval", 3600
+ session, "download.schedule_interval", 28800
)
# Add sources if provided
@@ -263,7 +263,7 @@ async def add_source_to_subscription(subscription_id: int):
# Get default check interval from database settings
default_interval = await get_setting_value(
- session, "download.schedule_interval", 3600
+ session, "download.schedule_interval", 28800
)
source = Source(
diff --git a/backend/app/models/setting.py b/backend/app/models/setting.py
index 70ab73e..bea6c1f 100644
--- a/backend/app/models/setting.py
+++ b/backend/app/models/setting.py
@@ -40,7 +40,7 @@ DEFAULT_SETTINGS = {
"download.parallel_limit": 3,
"download.rate_limit": 3.0,
"download.retry_count": 3,
- "download.schedule_interval": 3600,
+ "download.schedule_interval": 28800, # 8 hours
"notification.enabled": False,
"notification.webhook_url": None,
}
diff --git a/backend/app/models/source.py b/backend/app/models/source.py
index ec6e0a6..b5307ac 100644
--- a/backend/app/models/source.py
+++ b/backend/app/models/source.py
@@ -28,7 +28,7 @@ class Source(Base, TimestampMixin):
platform: Mapped[str] = mapped_column(String(50), nullable=False, index=True)
url: Mapped[str] = mapped_column(Text, nullable=False)
enabled: Mapped[bool] = mapped_column(Boolean, default=True)
- check_interval: Mapped[int] = mapped_column(Integer, default=3600) # seconds
+ check_interval: Mapped[int] = mapped_column(Integer, default=28800) # 8 hours in seconds
# Status tracking
last_check: Mapped[Optional[datetime]] = mapped_column(nullable=True)
diff --git a/backend/app/tasks/celery_app.py b/backend/app/tasks/celery_app.py
index 406dde1..9b761ea 100644
--- a/backend/app/tasks/celery_app.py
+++ b/backend/app/tasks/celery_app.py
@@ -9,6 +9,10 @@ from app.config import get_settings
logger = logging.getLogger(__name__)
settings = get_settings()
+# Log Redis configuration at import time (helps diagnose DB mismatch issues)
+_redis_db = settings.redis_url.split('/')[-1] if '/' in settings.redis_url else '0'
+logger.info(f"Celery using Redis: {settings.redis_url} (database: {_redis_db})")
+
def _get_parallel_limit_from_db():
"""Try to get parallel_limit from database, fall back to environment.
@@ -56,7 +60,7 @@ celery_app = Celery(
# Celery configuration
celery_app.conf.update(
task_serializer="json",
- accept_content=["json"],
+ accept_content=["json", "pickle"], # Accept pickle for backward compatibility during upgrades
result_serializer="json",
timezone="UTC",
enable_utc=True,
@@ -64,16 +68,31 @@ celery_app.conf.update(
task_time_limit=3600, # 1 hour max per task
worker_prefetch_multiplier=1, # Process one task at a time
worker_concurrency=parallel_limit,
+ # Task routing - send tasks to specific queues
+ task_routes={
+ # Lightweight scheduling/maintenance tasks go to 'scheduling' queue
+ "app.tasks.downloads.scheduled_check": {"queue": "scheduling"},
+ "app.tasks.downloads.cleanup_old_downloads": {"queue": "scheduling"},
+ "tasks.update_storage_stats": {"queue": "scheduling"},
+ "tasks.reset_orphaned_jobs": {"queue": "scheduling"},
+ "tasks.requeue_stale_jobs": {"queue": "scheduling"},
+ # Heavy download tasks go to 'downloads' queue (default)
+ "app.tasks.downloads.download_source": {"queue": "downloads"},
+ "app.tasks.downloads.process_download": {"queue": "downloads"},
+ },
+ # Default queue for any unrouted tasks
+ task_default_queue="downloads",
)
# Celery Beat schedule for periodic tasks
-# Note: The scheduler runs frequently (every 60s) but only triggers downloads
-# for sources whose individual check_interval has elapsed. Per-source intervals
-# are stored in the database and can be configured via the UI.
+# The scheduler runs periodically to check if any sources are due for download.
+# The actual schedule_interval (e.g., 8 hours) is enforced inside the task
+# by comparing each source's last_check timestamp against the DB setting.
+# 15 minutes is a good balance - responsive enough without excessive overhead.
celery_app.conf.beat_schedule = {
"check-sources-scheduler": {
"task": "app.tasks.downloads.scheduled_check",
- "schedule": 60, # Run every 60 seconds to check for due sources
+ "schedule": 900, # Run every 15 minutes to check for due sources
},
"cleanup-old-downloads-daily": {
"task": "app.tasks.downloads.cleanup_old_downloads",
@@ -87,28 +106,79 @@ celery_app.conf.beat_schedule = {
"task": "tasks.reset_orphaned_jobs",
"schedule": crontab(minute="*/15"), # Every 15 minutes
},
+ "requeue-stale-jobs": {
+ "task": "tasks.requeue_stale_jobs",
+ "schedule": crontab(minute="*/15"), # Every 15 minutes, check for stale queued jobs
+ },
}
-# 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
-
-
+# Worker startup signal - only reset orphaned jobs on scheduler worker
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.
+ Only the scheduler worker resets orphaned jobs to avoid duplicate resets
+ when multiple download workers start.
"""
+ # Check if this worker is consuming from the 'scheduling' queue
+ # sender.app.amqp.queues contains the queues this worker consumes from
try:
- from app.tasks.maintenance import reset_all_running_jobs_sync
- result = reset_all_running_jobs_sync()
- logger.info(f"Worker startup: {result}")
+ queues = [q.name for q in sender.app.amqp.queues.consume_from.values()]
+ is_scheduler = 'scheduling' in queues
+ except Exception:
+ # Fallback: check hostname or other indicators
+ is_scheduler = False
+
+ # Verify Redis connectivity and log configuration
+ try:
+ import redis
+ redis_url = settings.redis_url
+ redis_db = redis_url.split('/')[-1] if '/' in redis_url else '0'
+ r = redis.from_url(redis_url)
+ r.ping()
+ logger.info(f"Redis connectivity verified: {redis_url} (DB: {redis_db})")
+
+ # Check for other workers' markers to detect DB mismatches
+ marker_key = f"gallerysubscriber:worker:{'scheduler' if is_scheduler else 'download'}"
+ r.setex(marker_key, 3600, redis_url) # Expires in 1 hour
+
+ # Check if scheduler marker exists (helps detect if scheduler is on different DB)
+ scheduler_marker = r.get("gallerysubscriber:worker:scheduler")
+ download_marker = r.get("gallerysubscriber:worker:download")
+ if scheduler_marker and download_marker:
+ if scheduler_marker != download_marker:
+ logger.error(f"REDIS DB MISMATCH DETECTED! Scheduler: {scheduler_marker.decode()}, Worker: {download_marker.decode()}")
except Exception as e:
- logger.error(f"Failed to reset orphaned jobs on worker startup: {e}")
+ logger.error(f"Redis connectivity check failed: {e}")
+
+ if is_scheduler:
+ logger.info("Scheduler worker starting - resetting orphaned jobs...")
+ try:
+ from app.tasks.maintenance import reset_all_running_jobs_sync
+ result = reset_all_running_jobs_sync()
+ logger.info(f"Scheduler startup: Reset {result.get('reset_count', 0)} orphaned running jobs")
+ except Exception as e:
+ logger.error(f"Failed to reset orphaned jobs on scheduler startup: {e}")
+
+ # Re-queue any jobs stuck in 'queued' status (Celery tasks may have been lost)
+ logger.info("Scheduler startup: Re-queuing any stale queued jobs...")
+ try:
+ from app.tasks.maintenance import requeue_all_queued_jobs_sync
+ result = requeue_all_queued_jobs_sync()
+ logger.info(f"Scheduler startup: Re-queued {result.get('requeued_count', 0)} queued jobs")
+ except Exception as e:
+ logger.error(f"Failed to re-queue jobs on scheduler startup: {e}")
+
+ # Trigger an immediate scheduled_check on startup (don't wait for beat timer)
+ logger.info("Scheduler startup: Triggering immediate scheduled_check...")
+ try:
+ from app.tasks.downloads import scheduled_check
+ scheduled_check.delay()
+ logger.info("Scheduler startup: scheduled_check task queued")
+ except Exception as e:
+ logger.error(f"Failed to queue initial scheduled_check: {e}")
+ else:
+ logger.info("Download worker starting - skipping orphaned job reset (handled by scheduler)")
diff --git a/backend/app/tasks/downloads.py b/backend/app/tasks/downloads.py
index 6673ed7..280ed9e 100644
--- a/backend/app/tasks/downloads.py
+++ b/backend/app/tasks/downloads.py
@@ -59,10 +59,13 @@ async def _get_db_setting(session: AsyncSession, key: str, default=None):
setting = result.scalar()
if setting:
+ logger.debug(f"DB setting '{key}' found: {setting.value} (type: {type(setting.value).__name__})")
return setting.value
# Fall back to DEFAULT_SETTINGS, then to provided default
- return DEFAULT_SETTINGS.get(key, default)
+ fallback = DEFAULT_SETTINGS.get(key, default)
+ logger.debug(f"DB setting '{key}' not found, using fallback: {fallback}")
+ return fallback
def _get_db_setting_sync(key: str, default=None):
@@ -138,6 +141,28 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic
)
download = result.scalar()
if download:
+ # Guard against race conditions: only proceed if still QUEUED
+ if download.status != DownloadStatus.QUEUED:
+ logger.info(f"Download {download_id} is no longer QUEUED (status: {download.status}), skipping")
+ return {"source_id": source_id, "download_id": download_id, "status": "skipped", "reason": "already_processed"}
+
+ # Also check if source already has another RUNNING download
+ running_check = await session.execute(
+ select(Download.id).where(
+ Download.source_id == source_id,
+ Download.status == DownloadStatus.RUNNING,
+ Download.id != download_id
+ )
+ )
+ running_download_id = running_check.scalar()
+ if running_download_id:
+ # Mark this download as skipped since another is already running
+ download.status = DownloadStatus.SKIPPED
+ download.error_message = f"Skipped - another download ({running_download_id}) already running for this source"
+ await session.commit()
+ logger.info(f"Source {source_id} already has a running download, marked {download_id} as SKIPPED")
+ return {"source_id": source_id, "download_id": download_id, "status": "skipped", "reason": "source_already_running"}
+
download.status = DownloadStatus.RUNNING
download.started_at = utcnow()
else:
@@ -145,6 +170,17 @@ async def _download_source_async(source_id: int, download_id: int = None) -> dic
download = None
if not download_id or download is None:
+ # Check if source already has an active download before creating new one
+ active_check = await session.execute(
+ select(Download.id).where(
+ Download.source_id == source_id,
+ Download.status.in_([DownloadStatus.QUEUED, DownloadStatus.RUNNING])
+ )
+ )
+ if active_check.scalar():
+ logger.info(f"Source {source_id} already has an active download, skipping new download creation")
+ return {"source_id": source_id, "status": "skipped", "reason": "source_already_active"}
+
download = Download(
source_id=source_id,
url=source_url,
@@ -317,10 +353,12 @@ async def _scheduled_check_async() -> dict:
# 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
+ raw_interval = await _get_db_setting(
+ session, "download.schedule_interval", 28800 # Default 8 hours
)
- logger.debug(f"Using global schedule interval: {global_interval} seconds")
+ # Ensure we have an integer value (JSONB might return different types)
+ global_interval = int(raw_interval) if raw_interval is not None else 28800
+ logger.info(f"Schedule check: using interval {global_interval}s (raw value: {raw_interval}, type: {type(raw_interval).__name__})")
# Find sources due for checking:
# - enabled = True
@@ -352,15 +390,21 @@ async def _scheduled_check_async() -> dict:
# Filter by global schedule interval from Settings
# All sources use the same interval - no per-source overrides
due_sources = []
+ not_due_count = 0
for source in sources:
if source.last_check is None:
due_sources.append(source)
+ logger.debug(f"Source {source.id} ({source.platform}) - never checked, marking due")
else:
time_since_check = (now - source.last_check).total_seconds()
if time_since_check >= global_interval:
due_sources.append(source)
+ logger.debug(f"Source {source.id} ({source.platform}) - last check {time_since_check:.0f}s ago >= {global_interval}s, marking due")
+ else:
+ not_due_count += 1
+ logger.debug(f"Source {source.id} ({source.platform}) - last check {time_since_check:.0f}s ago < {global_interval}s, not due yet")
- logger.info(f"Found {len(due_sources)} sources due for checking")
+ logger.info(f"Schedule check: {len(due_sources)} sources due, {not_due_count} sources not due (interval: {global_interval}s)")
# Get source IDs that already have a queued or running download
due_source_ids = [s.id for s in due_sources]
@@ -483,7 +527,14 @@ def scheduled_check():
This is called periodically by Celery Beat.
"""
- return asyncio.run(_scheduled_check_async())
+ logger.info("=== SCHEDULED_CHECK TASK STARTING ===")
+ try:
+ result = asyncio.run(_scheduled_check_async())
+ logger.info(f"=== SCHEDULED_CHECK TASK COMPLETE: {result} ===")
+ return result
+ except Exception as e:
+ logger.exception(f"=== SCHEDULED_CHECK TASK FAILED: {e} ===")
+ raise
@celery_app.task
diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py
index 56286b4..1d9c5ec 100644
--- a/backend/app/tasks/maintenance.py
+++ b/backend/app/tasks/maintenance.py
@@ -23,6 +23,11 @@ settings = get_settings()
# This should be longer than the task_time_limit (1 hour) to account for legitimate long runs
STALE_RUNNING_THRESHOLD_MINUTES = 90
+# Maximum time a job can be in "queued" state before considered lost
+# If a queued job hasn't been picked up in this time, the Celery task was probably lost
+# (e.g., due to Redis DB mismatch or Redis restart)
+STALE_QUEUED_THRESHOLD_MINUTES = 30
+
def get_async_session():
"""Get async session factory for Celery tasks."""
@@ -242,3 +247,164 @@ def reset_all_running_jobs_sync() -> dict:
logger.info("Resetting all running jobs on worker startup...")
result = asyncio.run(_reset_all_running_jobs_async())
return result
+
+
+async def _requeue_stale_queued_jobs_async(threshold_minutes: int = None) -> dict:
+ """Re-queue Celery tasks for downloads stuck in 'queued' status.
+
+ Downloads can get stuck in 'queued' status if the Celery task was lost
+ (e.g., sent to wrong Redis DB, Redis restart, etc.). This function
+ detects these stale queued jobs and re-sends the Celery tasks.
+
+ Only one download per source is re-queued to prevent duplicate work.
+
+ Args:
+ threshold_minutes: Jobs queued longer than this are considered stale.
+ If None, uses STALE_QUEUED_THRESHOLD_MINUTES.
+
+ Returns:
+ Dict with count of re-queued jobs
+ """
+ if threshold_minutes is None:
+ threshold_minutes = STALE_QUEUED_THRESHOLD_MINUTES
+
+ # Import here to avoid circular imports
+ from app.tasks.downloads import download_source
+
+ session_factory, engine = get_async_session()
+
+ try:
+ async with session_factory() as session:
+ now = utcnow()
+ cutoff = now - timedelta(minutes=threshold_minutes)
+
+ # Find queued jobs created before the cutoff, ordered by creation time
+ # so we process older jobs first
+ query = select(Download).where(
+ Download.status == DownloadStatus.QUEUED,
+ Download.created_at < cutoff
+ ).order_by(Download.created_at.asc())
+
+ result = await session.execute(query)
+ stale_jobs = result.scalars().all()
+
+ # Track which sources we've already queued to avoid duplicates
+ queued_source_ids = set()
+ requeued_count = 0
+ skipped_duplicates = 0
+
+ for job in stale_jobs:
+ # Skip if we've already queued a job for this source
+ if job.source_id in queued_source_ids:
+ # Mark duplicate as skipped so it won't be checked again
+ job.status = DownloadStatus.SKIPPED
+ job.error_message = "Duplicate queued job - another job for this source was re-queued"
+ skipped_duplicates += 1
+ logger.info(f"Marked duplicate job {job.id} as SKIPPED (source {job.source_id} already re-queued)")
+ continue
+
+ try:
+ # Re-send the Celery task
+ download_source.delay(job.source_id, download_id=job.id)
+ queued_source_ids.add(job.source_id)
+ requeued_count += 1
+ logger.info(f"Re-queued stale job {job.id} (source_id: {job.source_id}, queued since: {job.created_at})")
+ except Exception as e:
+ logger.error(f"Failed to re-queue job {job.id}: {e}")
+
+ # Commit the status changes for skipped duplicates
+ if skipped_duplicates > 0:
+ await session.commit()
+
+ if requeued_count > 0:
+ logger.info(f"Re-queued {requeued_count} stale queued jobs (marked {skipped_duplicates} duplicates as SKIPPED)")
+ else:
+ logger.debug("No stale queued jobs found")
+
+ return {"requeued_count": requeued_count, "skipped_duplicates": skipped_duplicates, "threshold_minutes": threshold_minutes}
+
+ finally:
+ await _cleanup_engine(engine)
+
+
+async def _requeue_all_queued_jobs_async() -> dict:
+ """Re-queue Celery tasks for ALL downloads in 'queued' status.
+
+ This is useful on scheduler startup to recover from Redis DB mismatch
+ or other issues that caused Celery tasks to be lost.
+
+ Only one download per source is re-queued to prevent duplicate work.
+
+ Returns:
+ Dict with count of re-queued jobs
+ """
+ # Import here to avoid circular imports
+ from app.tasks.downloads import download_source
+
+ session_factory, engine = get_async_session()
+
+ try:
+ async with session_factory() as session:
+ # Find all queued jobs, ordered by creation time (oldest first)
+ query = select(Download).where(
+ Download.status == DownloadStatus.QUEUED
+ ).order_by(Download.created_at.asc())
+ result = await session.execute(query)
+ queued_jobs = result.scalars().all()
+
+ # Track which sources we've already queued to avoid duplicates
+ queued_source_ids = set()
+ requeued_count = 0
+ skipped_duplicates = 0
+
+ for job in queued_jobs:
+ # Skip if we've already queued a job for this source
+ if job.source_id in queued_source_ids:
+ # Mark duplicate as skipped so it won't be checked again
+ job.status = DownloadStatus.SKIPPED
+ job.error_message = "Duplicate queued job - another job for this source was re-queued"
+ skipped_duplicates += 1
+ logger.info(f"Marked duplicate job {job.id} as SKIPPED (source {job.source_id} already re-queued)")
+ continue
+
+ try:
+ download_source.delay(job.source_id, download_id=job.id)
+ queued_source_ids.add(job.source_id)
+ requeued_count += 1
+ logger.info(f"Re-queued job {job.id} (source_id: {job.source_id}) on startup")
+ except Exception as e:
+ logger.error(f"Failed to re-queue job {job.id}: {e}")
+
+ # Commit the status changes for skipped duplicates
+ if skipped_duplicates > 0:
+ await session.commit()
+
+ if requeued_count > 0:
+ logger.info(f"Re-queued {requeued_count} queued jobs on startup (marked {skipped_duplicates} duplicates as SKIPPED)")
+
+ return {"requeued_count": requeued_count, "skipped_duplicates": skipped_duplicates}
+
+ finally:
+ await _cleanup_engine(engine)
+
+
+@celery_app.task(name="tasks.requeue_stale_jobs")
+def requeue_stale_jobs(threshold_minutes: int = None) -> dict:
+ """Celery task to re-queue stale queued jobs.
+
+ Jobs that have been in 'queued' status for longer than the threshold
+ are considered lost (Celery task missing) and re-queued.
+ """
+ logger.info("Checking for stale queued jobs...")
+ result = asyncio.run(_requeue_stale_queued_jobs_async(threshold_minutes))
+ return result
+
+
+def requeue_all_queued_jobs_sync() -> dict:
+ """Synchronous function to re-queue all queued jobs.
+
+ Called on scheduler startup to recover lost Celery tasks.
+ """
+ logger.info("Re-queuing all queued jobs on scheduler startup...")
+ result = asyncio.run(_requeue_all_queued_jobs_async())
+ return result
diff --git a/backend/requirements.txt b/backend/requirements.txt
index 60c1906..b96eddd 100644
--- a/backend/requirements.txt
+++ b/backend/requirements.txt
@@ -1,28 +1,28 @@
# Web framework
-quart==0.19.4
-quart-cors==0.7.0
-hypercorn==0.16.0
-flask>=3.0.0,<3.1.0
+quart
+quart-cors
+hypercorn
+flask
# Database
-sqlalchemy[asyncio]==2.0.25
-asyncpg==0.29.0
-alembic==1.13.1
+sqlalchemy[asyncio]
+asyncpg
+alembic
# Task queue
-celery[redis]==5.3.6
-redis==5.0.1
+celery[redis]
+redis
# Utilities
-pydantic==2.5.3
-pydantic-settings==2.1.0
-python-dotenv==1.0.0
-cryptography==41.0.7
-pyyaml==6.0.1
+pydantic
+pydantic-settings
+python-dotenv
+cryptography
+pyyaml
# gallery-dl
-gallery-dl>=1.31.0
+gallery-dl
# Testing
-pytest==7.4.4
-pytest-asyncio==0.23.3
+pytest
+pytest-asyncio
diff --git a/docker-compose.yml b/docker-compose.yml
index 94402b8..b689fa9 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -16,7 +16,7 @@ services:
environment:
- TZ=${TZ:-America/New_York}
- DATABASE_URL=postgresql://${DB_USER:-gdl}:${DB_PASSWORD}@${DB_HOST:-db}:${DB_PORT:-5432}/${DB_NAME:-gallery_subscriber}
- - REDIS_URL=redis://${REDIS_HOST:-redis}:${REDIS_PORT:-6379}/0
+ - REDIS_URL=redis://${REDIS_HOST:-redis}:${REDIS_PORT:-6379}/${REDIS_DB:-0}
- SECRET_KEY=${SECRET_KEY}
- DOWNLOAD_PARALLEL_LIMIT=${DOWNLOAD_PARALLEL_LIMIT:-3}
- DOWNLOAD_RATE_LIMIT=${DOWNLOAD_RATE_LIMIT:-3.0}
@@ -31,14 +31,14 @@ services:
condition: service_started
restart: unless-stopped
- # Celery worker for background tasks
- celery:
+ # Celery download workers - only process download tasks (heavy work)
+ worker:
image: git.fabledsword.com/bvandeusen/gallerysubscriber:latest
- command: celery -A app.tasks.celery_app worker -B --loglevel=info
+ command: celery -A app.tasks.celery_app worker --queues=downloads --loglevel=info
environment:
- TZ=${TZ:-America/New_York}
- DATABASE_URL=postgresql://${DB_USER:-gdl}:${DB_PASSWORD}@${DB_HOST:-db}:${DB_PORT:-5432}/${DB_NAME:-gallery_subscriber}
- - REDIS_URL=redis://${REDIS_HOST:-redis}:${REDIS_PORT:-6379}/0
+ - REDIS_URL=redis://${REDIS_HOST:-redis}:${REDIS_PORT:-6379}/${REDIS_DB:-0}
- SECRET_KEY=${SECRET_KEY}
- DOWNLOAD_PARALLEL_LIMIT=${DOWNLOAD_PARALLEL_LIMIT:-3}
- DOWNLOAD_RATE_LIMIT=${DOWNLOAD_RATE_LIMIT:-3.0}
@@ -53,6 +53,22 @@ services:
condition: service_started
restart: unless-stopped
+ # Celery scheduler - runs beat + handles lightweight scheduling tasks
+ scheduler:
+ image: git.fabledsword.com/bvandeusen/gallerysubscriber:latest
+ command: celery -A app.tasks.celery_app worker --queues=scheduling --beat --loglevel=info --concurrency=1
+ environment:
+ - TZ=${TZ:-America/New_York}
+ - DATABASE_URL=postgresql://${DB_USER:-gdl}:${DB_PASSWORD}@${DB_HOST:-db}:${DB_PORT:-5432}/${DB_NAME:-gallery_subscriber}
+ - REDIS_URL=redis://${REDIS_HOST:-redis}:${REDIS_PORT:-6379}/${REDIS_DB:-0}
+ - SECRET_KEY=${SECRET_KEY}
+ depends_on:
+ db:
+ condition: service_healthy
+ redis:
+ condition: service_started
+ restart: unless-stopped
+
# PostgreSQL database
db:
image: postgres:15-alpine
diff --git a/frontend/src/views/Downloads.vue b/frontend/src/views/Downloads.vue
index 4992471..e659e0e 100644
--- a/frontend/src/views/Downloads.vue
+++ b/frontend/src/views/Downloads.vue
@@ -143,8 +143,8 @@
| {{ selectedDownload.url }} |
- | Source ID |
- {{ selectedDownload.source_id || 'N/A' }} |
+ Source |
+ {{ getSourceLabel(selectedDownload.source_id) }} |
| Files Downloaded |
@@ -228,7 +228,10 @@ const statusOptions = [
]
const sourceOptions = computed(() =>
- sourcesStore.sources.map(s => ({ title: s.name, value: s.id }))
+ sourcesStore.sources.map(s => ({
+ title: `${s.subscription_name || 'Unknown'}:${s.platform}`,
+ value: s.id
+ }))
)
const headers = [
@@ -308,6 +311,15 @@ function truncateUrl(url) {
return url.substring(0, 47) + '...'
}
+function getSourceLabel(sourceId) {
+ if (!sourceId) return 'N/A'
+ const source = sourcesStore.sources.find(s => s.id === sourceId)
+ if (source) {
+ return `${source.subscription_name || 'Unknown'}:${source.platform}`
+ }
+ return `Source #${sourceId}`
+}
+
function showDetails(download) {
selectedDownload.value = download
detailsDialog.value = true
diff --git a/frontend/src/views/Settings.vue b/frontend/src/views/Settings.vue
index eaef7d8..330ca34 100644
--- a/frontend/src/views/Settings.vue
+++ b/frontend/src/views/Settings.vue
@@ -420,7 +420,7 @@ const sortedAllPackages = computed(() => {
})
const scheduleIntervalHint = computed(() => {
- const seconds = settings.value['download.schedule_interval'] || 3600
+ const seconds = settings.value['download.schedule_interval'] || 28800
if (seconds < 60) return `Check every ${seconds} seconds`
if (seconds < 3600) return `Check every ${Math.round(seconds / 60)} minutes`
if (seconds < 86400) {
diff --git a/frontend/src/views/SourceForm.vue b/frontend/src/views/SourceForm.vue
index 0d29e95..90f323f 100644
--- a/frontend/src/views/SourceForm.vue
+++ b/frontend/src/views/SourceForm.vue
@@ -190,7 +190,7 @@ const formData = ref({
platform: '',
url: '',
enabled: true,
- check_interval: 3600,
+ check_interval: 28800,
metadata: {},
})
diff --git a/summary.md b/summary.md
index 2181d25..4e4a958 100644
--- a/summary.md
+++ b/summary.md
@@ -1,6 +1,6 @@
# GallerySubscriber - Project Summary
-**Updated:** 2026-01-28 (settings integration, orphan cleanup, error categorization fixes)
+**Updated:** 2026-01-29 (task routing, scheduler/worker separation, 8-hour default interval)
## Overview
@@ -95,7 +95,8 @@ GallerySubscriber/
| Component | Entry Point | Command |
|-----------|-------------|---------|
| Backend API | `backend/app/main.py` | `hypercorn app.main:app --bind 0.0.0.0:8080` |
-| Celery Worker | `backend/app/tasks/celery_app.py` | `celery -A app.tasks.celery_app worker -B` |
+| Scheduler | `backend/app/tasks/celery_app.py` | `celery -A app.tasks.celery_app worker --queues=scheduling --beat --concurrency=1` |
+| Download Workers | `backend/app/tasks/celery_app.py` | `celery -A app.tasks.celery_app worker --queues=downloads` |
| Frontend | `frontend/src/main.js` | Built with Vite to `/static` |
## Database Schema
@@ -142,6 +143,7 @@ Download (1) ──→ (many) ContentItem
| `/api/downloads/stats` | GET | Download counts by status/platform |
| `/api/downloads/recent-activity` | GET | Recent failures and completions |
| `/api/downloads/reset-orphaned` | POST | Reset stuck running jobs |
+| `/api/downloads/requeue-stale` | POST | Re-queue lost queued jobs |
| `/api/credentials` | GET, POST | List/upload credentials |
| `/api/credentials/` | DELETE | Remove credential |
| `/api/settings` | GET, PUT | App configuration |
@@ -175,17 +177,18 @@ Download (1) ──→ (many) ContentItem
3. **Service Layer** - GalleryDLService wraps gallery-dl subprocess
4. **Encrypted Credentials** - Fernet symmetric encryption with PBKDF2 key derivation
5. **Global Schedule Interval** - All sources checked at the same interval (configured in Settings)
-6. **Celery Beat** - Runs `scheduled_check` every 60s to trigger due downloads
-7. **Database-Driven Settings** - Worker reads rate_limit, retry_count, parallel_limit from DB
-8. **Orphan Job Cleanup** - Worker startup + periodic task resets stuck "running" jobs
+6. **Task Routing** - Separate queues for scheduling vs download tasks
+7. **Database-Driven Settings** - Worker reads rate_limit, retry_count, parallel_limit, schedule_interval from DB
+8. **Orphan Job Cleanup** - Scheduler startup + periodic task resets stuck "running" jobs
## Docker Services
```yaml
-app: Backend API + Frontend (port 8080)
-celery: Background worker with scheduler
-db: PostgreSQL 15-Alpine
-redis: Redis 7-Alpine
+app: Backend API + Frontend (port 8080)
+scheduler: Celery beat + scheduling worker (lightweight tasks)
+worker: Celery download workers (heavy gallery-dl tasks)
+db: PostgreSQL 15-Alpine
+redis: Redis 7-Alpine (task queue + result backend)
Volumes:
downloads: /data/downloads (downloaded files)
@@ -193,6 +196,39 @@ Volumes:
postgres_data: Database persistence
```
+## Task Routing Architecture
+
+Tasks are routed to separate queues for clean separation of concerns:
+
+```
+┌─────────────────────────────────────────────────────────────────┐
+│ SCHEDULER (1 container) │
+│ - Runs Celery Beat (triggers periodic tasks) │
+│ - 1 worker consuming from "scheduling" queue │
+│ - Handles: scheduled_check, cleanup, storage_stats, etc. │
+│ - Resets orphaned jobs on startup (only here, not workers) │
+└─────────────────────────────────────────────────────────────────┘
+ │ Queues download tasks
+ ▼
+┌─────────────────────────────────────────────────────────────────┐
+│ Redis Queues │
+│ - "scheduling" queue → scheduler only │
+│ - "downloads" queue → download workers only │
+└─────────────────────────────────────────────────────────────────┘
+ │
+ ▼
+┌─────────────────────────────────────────────────────────────────┐
+│ WORKERS (scalable, 1+ containers) │
+│ - Consume from "downloads" queue only │
+│ - Handles: download_source, process_download │
+│ - Heavy work (running gallery-dl) │
+└─────────────────────────────────────────────────────────────────┘
+```
+
+**Task Routing Config** (in `celery_app.py`):
+- `scheduled_check`, `cleanup_old_downloads`, `update_storage_stats`, `reset_orphaned_jobs` → "scheduling" queue
+- `download_source`, `process_download` → "downloads" queue
+
## Common Operations
**Run migrations:**
@@ -207,7 +243,7 @@ docker compose up -d --build
**View logs:**
```bash
-docker compose logs -f app celery
+docker compose logs -f app scheduler worker
```
**Development mode:**
@@ -233,22 +269,39 @@ docker compose -f docker-compose.dev.yml up
5. Download records and content items stored in database
6. Real-time progress broadcast via WebSocket
+## Source Scheduling Rules
+
+A source gets queued for download when ALL of these conditions are met:
+
+1. `source.enabled = True`
+2. `subscription.enabled = True`
+3. `(now - source.last_check) >= schedule_interval` OR `source.last_check is NULL` (never checked)
+4. No existing QUEUED or RUNNING download for this source
+
+**Note:** `last_check` is updated when a download **completes** (success or failure), not when queued. This means the interval is measured from completion to completion.
+
## Scheduled Tasks (Celery Beat)
| Task | Schedule | Purpose |
|------|----------|---------|
-| `scheduled_check` | Every 60s | Check if sources are due (based on global schedule_interval) |
+| `scheduled_check` | Every 15 min | Check if sources are due (based on global schedule_interval) |
| `cleanup_old_downloads` | Daily 3 AM | Remove old download records (30d completed, 7d failed) |
| `update_storage_stats` | Every 30 min | Calculate downloads folder size |
| `reset_orphaned_jobs` | Every 15 min | Reset jobs stuck in "running" > 90 min |
+| `requeue_stale_jobs` | Every 15 min | Re-queue jobs stuck in "queued" > 30 min (lost Celery tasks) |
-**Worker Startup Behavior:** When Celery worker starts, it resets ALL "running" jobs to "queued" (handles crash recovery).
+**Scheduler Startup Behavior:** When the scheduler container starts:
+1. Resets ALL "running" jobs to "queued" (handles crash recovery)
+2. Re-queues ALL "queued" jobs (recovers from Redis DB mismatch or lost tasks)
+3. Triggers an immediate `scheduled_check` (don't wait 15 min for first check)
+
+Download workers do NOT perform these startup tasks to avoid duplicates.
## Settings (Database-Driven)
| Key | Default | Purpose |
|-----|---------|---------|
-| `download.schedule_interval` | 3600 | Seconds between source checks (applies to all sources) |
+| `download.schedule_interval` | 28800 | Seconds between source checks (8 hours, applies to all sources) |
| `download.rate_limit` | 3.0 | Seconds between gallery-dl requests |
| `download.parallel_limit` | 3 | Concurrent Celery workers (requires restart) |
| `download.retry_count` | 3 | Max retries for failed downloads |