bvandeusen
e35fb1edf7
The scan tick (scan.py:_tick_due_sources_async) inserts DownloadEvent(status='pending') and fires download_source.delay(). If the task dies before finalizing the event — worker OOM/SIGKILL, lost task, or a gallery-dl that didn't unwind on the 1200s hard time_limit — the event stays in-flight forever. Every later tick then skips the source via the in-flight guard (scan.py:168), so Source.last_checked_at is never written and the operator sees "last check never" in the Subscriptions health column, permanently. cleanup_old_download_events only prunes terminal events (by design); no existing sweep covered the pending/running case. Operator confirmed 2026-05-29 with a diagnostic query: all 43 "never checked" sources were stranded behind stale in-flight events (eligible_stuck_inflight = 43, every other bucket zero). New recover_stalled_download_events task (Beat every 5 min): - Flips DownloadEvent rows pending/running > 30 min (10 min past the download_source 1200s hard kill, so legitimately-running tasks are never touched) to status='error' with a sentinel message. - Bumps each affected Source's consecutive_failures ONCE per source — backoff is 2^N on that counter so per-event bumps would needlessly inflate the next interval — sets last_error, stamps last_checked_at. UPDATE...RETURNING source_id avoids a SELECT-then-UPDATE-WHERE-IN that would hit the psycopg 65535-param ceiling on a large strand pile. Net: the 43 currently-stranded sources unstick on the first sweep after deploy, their health dots flip amber instead of unchecked, and the next scan tick re-queues them. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
118 lines
4.6 KiB
Python
118 lines
4.6 KiB
Python
"""Celery configuration with separate queue lanes.
|
|
|
|
Queues:
|
|
import — filesystem import path (FC-2)
|
|
ml — WD14 + SigLIP inference (FC-2; runs in the ml-worker image)
|
|
thumbnail — image and video thumbnail generation (FC-2)
|
|
download — gallery-dl tasks (FC-3)
|
|
scan — periodic source checks (FC-3) — kept separate so long imports
|
|
don't starve the scheduler
|
|
maintenance — pHash recomputation, centroid rebuild, etc. (FC-2/FC-3)
|
|
default — anything not explicitly routed
|
|
"""
|
|
|
|
from celery import Celery
|
|
|
|
from .config import get_config
|
|
|
|
|
|
def make_celery() -> Celery:
|
|
cfg = get_config()
|
|
app = Celery(
|
|
"fabledcurator",
|
|
broker=cfg.celery_broker_url,
|
|
backend=cfg.celery_result_backend,
|
|
include=[
|
|
"backend.app.tasks.smoke",
|
|
"backend.app.tasks.scan",
|
|
"backend.app.tasks.import_file",
|
|
"backend.app.tasks.thumbnail",
|
|
"backend.app.tasks.maintenance",
|
|
"backend.app.tasks.ml",
|
|
"backend.app.tasks.download",
|
|
"backend.app.tasks.backup",
|
|
"backend.app.tasks.admin",
|
|
"backend.app.tasks.library_audit",
|
|
],
|
|
)
|
|
app.conf.update(
|
|
task_default_queue="default",
|
|
task_routes={
|
|
"backend.app.tasks.import_file.*": {"queue": "import"},
|
|
"backend.app.tasks.ml.*": {"queue": "ml"},
|
|
"backend.app.tasks.thumbnail.*": {"queue": "thumbnail"},
|
|
"backend.app.tasks.download.*": {"queue": "download"},
|
|
"backend.app.tasks.scan.*": {"queue": "scan"},
|
|
"backend.app.tasks.maintenance.*": {"queue": "maintenance"},
|
|
"backend.app.tasks.backup.*": {"queue": "maintenance"},
|
|
"backend.app.tasks.admin.*": {"queue": "maintenance"},
|
|
"backend.app.tasks.library_audit.*": {"queue": "maintenance"},
|
|
},
|
|
# Heavy ML tasks need fair dispatch — see ImageRepo's precedent.
|
|
task_acks_late=True,
|
|
worker_prefetch_multiplier=1,
|
|
broker_connection_retry_on_startup=True,
|
|
beat_schedule={
|
|
"recover-interrupted-tasks": {
|
|
"task": "backend.app.tasks.maintenance.recover_interrupted_tasks",
|
|
"schedule": 300.0, # every 5 minutes
|
|
},
|
|
"cleanup-old-tasks": {
|
|
"task": "backend.app.tasks.maintenance.cleanup_old_tasks",
|
|
"schedule": 86400.0, # daily
|
|
},
|
|
"ml-backfill-daily": {
|
|
"task": "backend.app.tasks.ml.backfill",
|
|
"schedule": 86400.0,
|
|
},
|
|
"recompute-centroids-daily": {
|
|
"task": "backend.app.tasks.ml.recompute_centroids",
|
|
"schedule": 86400.0,
|
|
},
|
|
"apply-allowlist-sweep-daily": {
|
|
"task": "backend.app.tasks.ml.apply_allowlist_tags",
|
|
"schedule": 86400.0,
|
|
},
|
|
"integrity-verify-weekly": {
|
|
"task": "backend.app.tasks.maintenance.verify_integrity",
|
|
"schedule": 604800.0, # weekly
|
|
},
|
|
"fc3d-tick-due-sources": {
|
|
"task": "backend.app.tasks.scan.tick_due_sources",
|
|
"schedule": 60.0, # every minute
|
|
},
|
|
"fc3d-cleanup-download-events": {
|
|
"task": "backend.app.tasks.maintenance.cleanup_old_download_events",
|
|
"schedule": 86400.0, # daily
|
|
},
|
|
"recover-stalled-download-events": {
|
|
"task": "backend.app.tasks.maintenance.recover_stalled_download_events",
|
|
"schedule": 300.0, # every 5 min, matches recover-interrupted-tasks
|
|
},
|
|
"recover-stalled-task-runs": {
|
|
"task": "backend.app.tasks.maintenance.recover_stalled_task_runs",
|
|
"schedule": 300.0, # every 5 min, matches recover-interrupted-tasks
|
|
},
|
|
"prune-task-runs": {
|
|
"task": "backend.app.tasks.maintenance.prune_task_runs",
|
|
"schedule": 86400.0, # daily
|
|
},
|
|
"fc3h-backup-db-nightly": {
|
|
"task": "backend.app.tasks.backup.backup_db_nightly",
|
|
"schedule": 3600.0, # hourly tick; task self-gates on configured UTC hour
|
|
},
|
|
"fc3h-prune-backups": {
|
|
"task": "backend.app.tasks.backup.prune_backups",
|
|
"schedule": 86400.0, # daily
|
|
},
|
|
},
|
|
timezone="UTC",
|
|
)
|
|
# FC-3i: register task_run signal handlers (side-effect import).
|
|
from . import celery_signals # noqa: F401
|
|
|
|
return app
|
|
|
|
|
|
celery = make_celery()
|