Files
FabledCurator/backend/app/celery_app.py
T
bvandeusenandClaude Opus 5 11a01a9686
CI / lint (push) Successful in 2s
CI / extension-version (push) Successful in 3s
Build images / sign-extension (push) Successful in 3s
Build images / build-agent (push) Successful in 6s
CI / frontend-build (push) Successful in 26s
CI / backend-lint-and-test (push) Successful in 33s
Build images / build-web (push) Successful in 59s
Build images / smoke-web (push) Skipped
Build images / build-ml (push) Successful in 1m55s
Build images / promote (push) Skipped
CI / integration (push) Successful in 2m22s
revert: remove the placement reconciler — it manufactured the problem it solved
Milestone #421 built a sweep that compared each image's `artist_id` to the
name of the directory holding its file, and called every mismatch a misplaced
image. It reported 33,789 of 63,605 as wrongly filed. That number described
the comparison, not the library.

What it actually was:

  32,475  (97.1%)  one artist's own folder, spelled differently
                   — Telepurte/ vs telepurte/. Same artist, same art.
     657  ( 2.0%)  loose at the images root
     328  ( 1.0%)  in a folder named after a different artist

And the 1% did not mean what the tool assumed either. `ImageProvenance`
records the post and source every file was downloaded from — the
authoritative answer, which the tool never consulted. Querying it for all 328:

    144  provenance agrees with the record  (move would be right)
     87  provenance agrees with the FOLDER  (the record is wrong; move wrong)
     53  provenance names SEVERAL artists   (no single correct folder)
     41  no provenance at all
      3  agrees with neither

So the sweep would have misfiled or arbitrarily picked for ~41% of the only
set it was really needed for. The system already knew where each file came
from; the tool inferred it from a column and a directory name instead.

Operator, 2026-09-21: *"the current system consistently records where items
are and where they came from this is just complicating something works and
doesn't need fixing."* Correct on both counts.

Removed: the service, the tasks, the model and migration 0099's table, the
/api/cleanup/layout and /placement/* endpoints, the Maintenance card and its
store actions, and the tests. 0100 drops the table (rule #22 — no legacy).

KEPT deliberately, per the operator:
- `utils.paths.canonical_subdir` — new filesystem imports derive their
  directory from the artist's slug, matching what the downloader always did.
  Not part of this tool; removing it would be churn that fixes nothing.
- The 327 files run 1 moved (InsoUwu/ -> insouwu/). Same artist either way,
  and the gallery renders them correctly.
- Everything from #4223 (three-gate dedup, 256-bit pHash) and #4234 (backup
  credential exclusion). Those fixed problems that were actually reported.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR
2026-09-21 18:17:13 -04:00

335 lines
17 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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 — recovery sweeps, pHash backfill, GPU-queue coordination, etc.
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.gpu_queue",
"backend.app.tasks.download",
"backend.app.tasks.external",
"backend.app.tasks.backup",
"backend.app.tasks.admin",
"backend.app.tasks.library_audit",
"backend.app.tasks.translation",
],
)
app.conf.update(
task_default_queue="default",
task_routes={
"backend.app.tasks.import_file.*": {"queue": "import"},
"backend.app.tasks.ml.*": {"queue": "ml"},
# GPU-queue coordination (backfill enqueues, orphan recovery,
# reprocess) is pure DB work — it rides the maintenance quick lane
# so the GPU agent pipeline works even on stacks that drop the
# (now-optional, B3) ml-worker container entirely.
"backend.app.tasks.gpu_queue.*": {"queue": "maintenance"},
"backend.app.tasks.thumbnail.*": {"queue": "thumbnail"},
"backend.app.tasks.download.*": {"queue": "download"},
# External file-host fetches are downloads — same lane (they can run
# long, but the download worker already tolerates long backfills).
"backend.app.tasks.external.*": {"queue": "download"},
"backend.app.tasks.scan.*": {"queue": "scan"},
# `maintenance` is the QUICK lane — recovery sweeps, vacuum, cleanup
# (concurrency-1 on the scheduler). The long one-shots (DB backups,
# library audits, admin maintenance: normalize/re-extract/cascade-
# delete) run on a SEPARATE `maintenance_long` lane + worker so they
# can never starve the quick self-healing sweeps (operator-flagged
# 2026-06-07: a 2h audit blocked vacuum/backup/normalize for hours).
"backend.app.tasks.maintenance.*": {"queue": "maintenance"},
"backend.app.tasks.backup.*": {"queue": "maintenance_long"},
"backend.app.tasks.admin.*": {"queue": "maintenance_long"},
"backend.app.tasks.library_audit.*": {"queue": "maintenance_long"},
# Translation backfill hits the LLM (~16s/item) → the long lane so it
# never starves the quick self-healing sweeps (#143).
"backend.app.tasks.translation.*": {"queue": "maintenance_long"},
},
# Heavy ML tasks need fair dispatch — see ImageRepo's precedent.
task_acks_late=True,
# Deploy graceful-shutdown safety: with acks_late, a task killed because
# it outran the container's stop-grace window (SIGKILL) is re-queued
# rather than silently lost. Safe because our long tasks are idempotent +
# chunked (translation per-post commit, downloads terminal-status, audits
# chunk) and the 5-min recovery sweeps re-drive anything left non-terminal
# — a re-run resumes cleanly and never corrupts. No redeliver-loop risk:
# heavy GPU work is tombstoned via gpu_queue, not run inline in a worker.
task_reject_on_worker_lost=True,
worker_prefetch_multiplier=1,
# Broker resilience (2026-06-24): a swarm overlay-network blip after a
# redeploy left Redis healthy but transiently unreachable, and a worker
# starting in that window crash-looped on the initial broker connect
# (kombu OperationalError) instead of waiting it out — needing a manual
# Redis reset to recover. Retry the broker FOREVER (None) on startup and
# at runtime so a transient outage self-heals when routing returns,
# rather than the worker exiting.
broker_connection_retry_on_startup=True,
broker_connection_retry=True,
broker_connection_max_retries=None,
# Redis-transport socket options (apply to the BROKER connection): a
# short connect timeout + TCP keepalive so a dead/blocked socket is
# noticed and retried, and a periodic health check that proactively
# reconnects a live worker through a network hiccup.
broker_transport_options={
"socket_connect_timeout": 5,
"socket_timeout": 30,
"socket_keepalive": True,
"retry_on_timeout": True,
"health_check_interval": 30,
},
# Same hardening for the Redis RESULT backend (separate connection pool).
redis_socket_connect_timeout=5,
redis_socket_timeout=30,
redis_socket_keepalive=True,
redis_retry_on_timeout=True,
redis_backend_health_check_interval=30,
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
},
"cleanup-orphaned-temp-files": {
"task": "backend.app.tasks.maintenance.cleanup_orphaned_temp_files",
"schedule": 86400.0, # daily — sweep .part/.partial left by a
# download/import killed mid-write (graceful-shutdown fallout)
},
"backfill-phash-daily": {
"task": "backend.app.tasks.maintenance.backfill_phash",
"schedule": 86400.0, # daily — NULL-only, so a no-op once the
# library is hashed. This is what makes migration 0098's
# re-hash happen on its own: 0098 NULLs every phash, and
# without a scheduled refill the library would sit
# dedup-disabled until someone ran a deep scan (#4223).
},
"train-heads-nightly": {
"task": "backend.app.tasks.ml.scheduled_train_heads",
"schedule": 86400.0, # passive cadence; manual retrain stays available
},
"refresh-character-prototypes": {
"task": "backend.app.tasks.ml.refresh_character_prototypes",
"schedule": 900.0, # ~15 min; cheap global-gate no-op when idle (#1317)
},
"reconcile-character-prototypes-nightly": {
"task": "backend.app.tasks.ml.refresh_character_prototypes",
"schedule": 86400.0, # nightly FULL reconcile (belt-and-suspenders)
"args": (True,), # full=True
},
"apply-head-tags-daily": {
"task": "backend.app.tasks.ml.scheduled_apply_head_tags",
"schedule": 86400.0, # no-op unless head_auto_apply_enabled
},
"recover-orphaned-gpu-jobs": {
"task": "backend.app.tasks.gpu_queue.recover_orphaned_gpu_jobs",
"schedule": 60.0, # quick pickup of work a dead agent orphaned
},
"triage-gpu-errors": {
"task": "backend.app.tasks.maintenance.triage_gpu_errors",
"schedule": 900.0, # probe errored jobs' files → defect/file_ok
},
"enqueue-ccip-backfill-hourly": {
"task": "backend.app.tasks.gpu_queue.enqueue_gpu_backfill",
"schedule": 3600.0, # auto-feed NEW images; errored are
"args": ("ccip",), # tombstoned — retry is the button only
},
"enqueue-siglip-backfill-daily": {
"task": "backend.app.tasks.gpu_queue.enqueue_gpu_backfill",
"schedule": 86400.0, # drain the concept-crop back-catalogue
"args": ("siglip",), # (errored are tombstoned, not retried)
},
"enqueue-embed-backfill-daily": {
"task": "backend.app.tasks.gpu_queue.enqueue_gpu_backfill",
"schedule": 86400.0, # whole-image re-embed under the current
"args": ("embed",), # model (an operator swap) drains via agent
},
"ccip-auto-apply-daily": {
"task": "backend.app.tasks.ml.scheduled_ccip_auto_apply",
"schedule": 86400.0, # no-op unless ccip_auto_apply_enabled
},
"retract-auto-tags-daily": {
"task": "backend.app.tasks.ml.scheduled_retract_auto_tags",
"schedule": 86400.0, # soft auto-apply: drop auto-tags now below
# their threshold (m139); no-op unless the auto-apply switch is on
},
"presentation-auto-apply-daily": {
"task": "backend.app.tasks.ml.scheduled_presentation_auto_apply",
"schedule": 86400.0, # auto-hide banner chrome (#141);
# no-op unless presentation_auto_apply_enabled
},
"process-auto-apply-daily": {
"task": "backend.app.tasks.ml.scheduled_process_auto_apply",
"schedule": 86400.0, # auto-tag wip/editor process art (#1464);
# no-op unless process_auto_apply_enabled (opt-in)
},
"soft-wip-conflict-audit-daily": {
"task": "backend.app.tasks.ml.scheduled_soft_wip_conflict_audit",
"schedule": 86400.0, # flag ring-loud soft-WIP (sketch/doodle) tags
# for review (#1474); no-op with no content heads
},
"prune-presentation-reviews-daily": {
"task": "backend.app.tasks.ml.prune_presentation_reviews",
"schedule": 86400.0, # retention: drop resolved review flags >30d
},
"translate-posts-8h": {
"task": "backend.app.tasks.translation.translate_posts",
"schedule": 28800.0, # every 8h: steady-state cadence for the
# trickle of newly-imported posts (no-op unless translation
# configured + healthy). One bounded 300-chunk per fire — the
# one-time backlog drains via the "Translate now" button
# (drain=True, run-until-done), not this sweep.
},
"snapshot-head-metrics-daily": {
"task": "backend.app.tasks.maintenance.snapshot_head_metrics",
"schedule": 86400.0,
},
"group-discord-drops-hourly": {
"task": "backend.app.tasks.maintenance.group_discord_drops",
"schedule": 3600.0, # hourly. Not daily: the grouping signal is
# the SigLIP embedding, which lands asynchronously AFTER import
# (#388 E2), so this sweep is what picks up a drop once its
# vectors have caught up. No-op unless discord_grouping_enabled.
},
"match-post-associations-hourly": {
"task": "backend.app.tasks.maintenance.match_post_associations",
"schedule": 3600.0, # hourly, and AFTER the grouper's own cadence
# by construction: a pair cannot be proposed until the drop it
# points at exists as a grouping (#388 E5). No-op unless
# discord_link_enabled.
},
"sync-memberships-daily": {
"task": "backend.app.tasks.maintenance.sync_memberships",
"schedule": 86400.0, # daily — memberships change on a BILLING
# cycle, not a download cadence (#387 C3). No-op per platform
# when the client lacks the seam or no credential exists.
},
"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
},
"vacuum-analyze": {
"task": "backend.app.tasks.maintenance.vacuum_analyze",
"schedule": 604800.0, # weekly — reclaim dead-tuple bloat + refresh stats
},
"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
},
# Audit 2026-06-02 — three new per-entity recovery sweeps.
# Each runs every 5 min like the other recover_stalled_*
# sweeps; each is a no-op when nothing is stuck.
"recover-stalled-backup-runs": {
"task": "backend.app.tasks.maintenance.recover_stalled_backup_runs",
"schedule": 300.0,
},
"recover-stalled-library-audit-runs": {
"task": "backend.app.tasks.maintenance.recover_stalled_library_audit_runs",
"schedule": 300.0,
},
"recover-stalled-head-training-runs": {
"task": "backend.app.tasks.maintenance.recover_stalled_head_training_runs",
"schedule": 300.0,
},
"recover-stalled-head-auto-apply-runs": {
"task": "backend.app.tasks.maintenance.recover_stalled_head_auto_apply_runs",
"schedule": 300.0,
},
"recover-stalled-import-batches": {
"task": "backend.app.tasks.maintenance.recover_stalled_import_batches",
"schedule": 300.0,
},
# Audit 2026-06-02 — daily retention for two entities
# whose terminal rows otherwise accumulate forever.
"prune-library-audit-runs": {
"task": "backend.app.tasks.maintenance.prune_library_audit_runs",
"schedule": 86400.0,
},
"prune-import-batches": {
"task": "backend.app.tasks.maintenance.prune_import_batches",
"schedule": 86400.0,
},
# Audit 2026-06-02 — backfill_thumbnails's docstring claimed
# "periodic Beat" but the entry was never registered, so the
# library got no self-healing thumbnail repair; only the
# manual admin-UI button fired it. Daily cadence is gentle
# (the task is idempotent and only enqueues regen for rows
# whose stored thumbnails are missing or corrupt).
"backfill-thumbnails-daily": {
"task": "backend.app.tasks.thumbnail.backfill_thumbnails",
"schedule": 86400.0,
},
# External file-host downloads (#830): a steady sweep catches links
# the post-download hook missed (worker down, etc.); recovery re-tries
# dead links daily; retention prunes long-dead rows.
"extdl-sweep": {
"task": "backend.app.tasks.external.sweep_external_links",
"schedule": 600.0, # every 10 min
},
"extdl-recover-daily": {
"task": "backend.app.tasks.external.recover_external_links",
"schedule": 86400.0,
},
"extdl-prune-daily": {
"task": "backend.app.tasks.external.prune_external_links",
"schedule": 86400.0,
},
},
timezone="UTC",
)
# FC-3i: register task_run signal handlers (side-effect import).
from . import celery_signals # noqa: F401
return app
celery = make_celery()