FabledCurator/tests/test_maintenance.py

"""Unit tests for the maintenance tasks. Eager-mode Celery so the task
function runs synchronously inside the same DB transaction as the test.
"""

from datetime import UTC, datetime, timedelta

import pytest

from backend.app.celery_app import celery
from backend.app.models import ImportBatch, ImportTask

pytestmark = pytest.mark.integration


@pytest.fixture(autouse=True)
def eager():
    celery.conf.task_always_eager = True
    yield
    celery.conf.task_always_eager = False


def _make_batch(session) -> int:
    batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
    session.add(batch)
    session.flush()
    return batch.id


def test_cleanup_orphaned_temp_files_removes_stale_only(tmp_path, monkeypatch):
    import os

    from backend.app.tasks import maintenance as m

    monkeypatch.setattr(m, "IMAGES_ROOT", tmp_path)
    stale = tmp_path / "artist" / "img.jpg.part"      # killed download → orphan
    stale.parent.mkdir(parents=True)
    stale.write_bytes(b"x")
    old = datetime.now(UTC).timestamp() - 8 * 3600     # older than the 6h guard
    os.utime(stale, (old, old))
    fresh = tmp_path / "in_progress.jpg.partial"       # active download → keep
    fresh.write_bytes(b"x")
    keep = tmp_path / "real.jpg"                        # a real image → keep
    keep.write_bytes(b"x")

    assert m.cleanup_orphaned_temp_files() == 1
    assert not stale.exists()
    assert fresh.exists()
    assert keep.exists()


def test_recover_interrupted_only_old(db_sync, monkeypatch):
    batch_id = _make_batch(db_sync)
    now = datetime.now(UTC)

    # "Fresh" must sit comfortably under whatever STUCK_THRESHOLD_MINUTES
    # currently is (5 min as of 2026-05-24, tightened from 30); 30
    # seconds is well below any reasonable threshold. "Stale" stays at
    # 2 hours so the test remains valid if the threshold ever moves
    # back up.
    fresh = ImportTask(
        batch_id=batch_id, source_path="/import/a.jpg", task_type="media",
        status="processing", started_at=now - timedelta(seconds=30),
    )
    stale = ImportTask(
        batch_id=batch_id, source_path="/import/b.jpg", task_type="media",
        status="processing", started_at=now - timedelta(hours=2),
    )
    db_sync.add_all([fresh, stale])
    db_sync.commit()

    # Isolate the recover task: under eager Celery, the real
    # import_media_file.delay() would run inline against the nonexistent
    # /import/b.jpg and flip the just-requeued row 'queued' -> 'skipped'.
    from backend.app.tasks import import_file

    dispatched: list[int] = []
    monkeypatch.setattr(
        import_file.import_media_file, "delay", dispatched.append
    )

    from backend.app.tasks.maintenance import recover_interrupted_tasks
    recovered = recover_interrupted_tasks.apply().get()
    assert recovered == 1

    db_sync.refresh(fresh)
    db_sync.refresh(stale)
    assert fresh.status == "processing"
    assert stale.status == "queued"
    assert stale.started_at is None
    assert dispatched == [stale.id]


def test_recover_interrupted_sweeps_pending_orphans_to_failed(db_sync, monkeypatch):
    """A scan that creates ImportTask rows but crashes before the second
    pass (transition to 'queued' + .delay()) leaves rows orphaned at
    status='pending'. The sweep flips them to 'failed' so the operator
    can drain via /api/import/retry-failed without thundering-herding.
    Banked 2026-05-25 after operator hit 5490 stuck pending rows.
    """
    from backend.app.tasks import import_file
    monkeypatch.setattr(import_file.import_media_file, "delay", lambda *_: None)

    batch_id = _make_batch(db_sync)
    now = datetime.now(UTC)

    fresh_pending = ImportTask(
        batch_id=batch_id, source_path="/import/fresh.jpg", task_type="media",
        status="pending",
    )
    db_sync.add(fresh_pending)
    db_sync.flush()
    # created_at defaults to now() server-side; fresh row stays untouched.

    # Two stale rows simulating the orphan pile: one 'pending', one
    # 'queued' (scanner crashed AFTER transitioning some rows but
    # before all). Both should sweep.
    stale_pending = ImportTask(
        batch_id=batch_id, source_path="/import/stale1.jpg", task_type="media",
        status="pending",
    )
    stale_queued = ImportTask(
        batch_id=batch_id, source_path="/import/stale2.jpg", task_type="media",
        status="queued",
    )
    db_sync.add_all([stale_pending, stale_queued])
    db_sync.flush()
    # Backdate created_at past the orphan cutoff (30 min).
    from sqlalchemy import update as _upd
    db_sync.execute(
        _upd(ImportTask)
        .where(ImportTask.id.in_([stale_pending.id, stale_queued.id]))
        .values(created_at=now - timedelta(hours=2))
    )
    db_sync.commit()

    from backend.app.tasks.maintenance import recover_interrupted_tasks
    touched = recover_interrupted_tasks.apply().get()
    assert touched == 2

    db_sync.refresh(fresh_pending)
    db_sync.refresh(stale_pending)
    db_sync.refresh(stale_queued)
    assert fresh_pending.status == "pending"  # fresh row untouched
    assert stale_pending.status == "failed"
    assert stale_queued.status == "failed"
    assert "orphan" in (stale_pending.error or "")


def test_recover_interrupted_handles_both_stuck_and_orphans(db_sync, monkeypatch):
    """One sweep tick handles both 'processing' crashes AND
    'pending'/'queued' orphans in a single pass."""
    from backend.app.tasks import import_file
    dispatched: list[int] = []
    monkeypatch.setattr(
        import_file.import_media_file, "delay", dispatched.append
    )

    batch_id = _make_batch(db_sync)
    now = datetime.now(UTC)

    stuck = ImportTask(
        batch_id=batch_id, source_path="/import/stuck.jpg", task_type="media",
        status="processing", started_at=now - timedelta(hours=2),
    )
    orphan = ImportTask(
        batch_id=batch_id, source_path="/import/orphan.jpg", task_type="media",
        status="pending",
    )
    db_sync.add_all([stuck, orphan])
    db_sync.flush()
    from sqlalchemy import update as _upd
    db_sync.execute(
        _upd(ImportTask).where(ImportTask.id == orphan.id)
        .values(created_at=now - timedelta(hours=2))
    )
    db_sync.commit()

    from backend.app.tasks.maintenance import recover_interrupted_tasks
    touched = recover_interrupted_tasks.apply().get()
    assert touched == 2

    db_sync.refresh(stuck)
    db_sync.refresh(orphan)
    assert stuck.status == "queued"
    assert orphan.status == "failed"
    assert dispatched == [stuck.id]  # stuck rows re-enqueue; orphans don't


def test_recover_interrupted_poison_pill_caps_at_max(db_sync, monkeypatch):
    """A stuck row that's already been recovered MAX_RECOVERY_ATTEMPTS-1
    times is marked 'failed' (with a diagnostic) instead of re-queued —
    the circuit breaker against an input that hard-crashes the worker
    every run. Operator-flagged 2026-05-28."""
    from backend.app.tasks import import_file
    from backend.app.tasks.maintenance import (
        MAX_RECOVERY_ATTEMPTS,
        recover_interrupted_tasks,
    )
    dispatched: list[int] = []
    monkeypatch.setattr(
        import_file.import_media_file, "delay", dispatched.append
    )

    batch_id = _make_batch(db_sync)
    now = datetime.now(UTC)

    # At the cap already (recovered MAX-1 times) → fail, don't re-queue.
    poison = ImportTask(
        batch_id=batch_id, source_path="/import/poison.jpg", task_type="media",
        status="processing", started_at=now - timedelta(hours=2),
        recovery_count=MAX_RECOVERY_ATTEMPTS - 1,
    )
    # One recovery short of the cap → re-queue + increment.
    recoverable = ImportTask(
        batch_id=batch_id, source_path="/import/ok.jpg", task_type="media",
        status="processing", started_at=now - timedelta(hours=2),
        recovery_count=MAX_RECOVERY_ATTEMPTS - 2,
    )
    db_sync.add_all([poison, recoverable])
    db_sync.commit()

    touched = recover_interrupted_tasks.apply().get()
    assert touched == 2  # one failed + one re-queued

    db_sync.refresh(poison)
    db_sync.refresh(recoverable)
    assert poison.status == "failed"
    assert "corrupt or" in (poison.error or "")
    assert recoverable.status == "queued"
    assert recoverable.recovery_count == MAX_RECOVERY_ATTEMPTS - 1
    # Only the recoverable row re-enqueues; the poison pill does not.
    assert dispatched == [recoverable.id]


def test_cleanup_old_deletes_finished_old(db_sync):
    batch_id = _make_batch(db_sync)
    now = datetime.now(UTC)

    old_complete = ImportTask(
        batch_id=batch_id, source_path="/import/a.jpg", task_type="media",
        status="complete", finished_at=now - timedelta(days=10),
    )
    recent_complete = ImportTask(
        batch_id=batch_id, source_path="/import/b.jpg", task_type="media",
        status="complete", finished_at=now - timedelta(days=2),
    )
    old_pending = ImportTask(
        batch_id=batch_id, source_path="/import/c.jpg", task_type="media",
        status="pending",
    )
    db_sync.add_all([old_complete, recent_complete, old_pending])
    db_sync.commit()

    from backend.app.tasks.maintenance import cleanup_old_tasks
    deleted = cleanup_old_tasks.apply().get()
    assert deleted == 1

    remaining = {t.source_path for t in db_sync.query(ImportTask).all()}
    assert remaining == {"/import/b.jpg", "/import/c.jpg"}


# --- FC-3i: task_run sweep + retention -----------------------------


def _make_task_run(db_sync, *, status, started_at, finished_at=None,
                   error_type=None, queue="default",
                   task_name="backend.app.tasks.fake.t"):
    from backend.app.models import TaskRun
    row = TaskRun(
        celery_task_id="x",
        queue=queue,
        task_name=task_name,
        target_id=1,
        started_at=started_at,
        finished_at=finished_at,
        duration_ms=1000 if finished_at else None,
        status=status,
        error_type=error_type,
        error_message="x" if status in ("error", "timeout") else None,
    )
    db_sync.add(row)
    db_sync.flush()
    return row.id


def test_recover_stalled_task_runs_flips_old_running_to_error(db_sync):
    from sqlalchemy import select

    from backend.app.models import TaskRun
    from backend.app.tasks.maintenance import recover_stalled_task_runs

    now = datetime.now(UTC)
    stale_id = _make_task_run(
        db_sync, status="running", started_at=now - timedelta(minutes=10),
    )
    db_sync.commit()

    recovered = recover_stalled_task_runs.apply().get()
    assert recovered == 1

    db_sync.expire_all()
    status = db_sync.execute(
        select(TaskRun.status).where(TaskRun.id == stale_id)
    ).scalar_one()
    error_type = db_sync.execute(
        select(TaskRun.error_type).where(TaskRun.id == stale_id)
    ).scalar_one()
    assert status == "error"
    assert error_type == "RecoverySweep"


def test_recover_stalled_task_runs_skips_fresh_running(db_sync):
    from sqlalchemy import select

    from backend.app.models import TaskRun
    from backend.app.tasks.maintenance import recover_stalled_task_runs

    now = datetime.now(UTC)
    fresh_id = _make_task_run(
        db_sync, status="running", started_at=now - timedelta(seconds=30),
    )
    db_sync.commit()

    recovered = recover_stalled_task_runs.apply().get()
    assert recovered == 0

    db_sync.expire_all()
    status = db_sync.execute(
        select(TaskRun.status).where(TaskRun.id == fresh_id)
    ).scalar_one()
    assert status == "running"


def test_recover_stalled_task_runs_ml_queue_uses_longer_threshold(db_sync):
    """ml-queue tasks (embed_image video branch) legitimately run
    past the default 5-min threshold. The sweep must NOT flag an
    ml-queue task that's only been running 10 min — the override
    threshold (QUEUE_STUCK_THRESHOLD_MINUTES["ml"]) protects in-flight
    video tagging. Operator-flagged 2026-05-28 after image 6288 (mp4)
    was marked failed at the 5-min tick mid-run. Read from the table
    rather than restated: the value moved 25 -> 40 in #4432."""
    from sqlalchemy import select

    from backend.app.models import TaskRun
    from backend.app.tasks.maintenance import (
        QUEUE_STUCK_THRESHOLD_MINUTES,
        recover_stalled_task_runs,
    )

    ml_threshold = QUEUE_STUCK_THRESHOLD_MINUTES["ml"]
    now = datetime.now(UTC)
    # 10-min-old ml-queue row: stale by the default 5-min rule but
    # fresh by the ml override. Must survive the sweep.
    ml_fresh_id = _make_task_run(
        db_sync, status="running", queue="ml",
        started_at=now - timedelta(minutes=10),
    )
    # Past even the ml override. Must be flagged.
    ml_stale_id = _make_task_run(
        db_sync, status="running", queue="ml",
        started_at=now - timedelta(minutes=ml_threshold + 5),
    )
    db_sync.commit()

    recovered = recover_stalled_task_runs.apply().get()
    assert recovered == 1

    db_sync.expire_all()
    ml_fresh_status = db_sync.execute(
        select(TaskRun.status).where(TaskRun.id == ml_fresh_id)
    ).scalar_one()
    ml_stale_status = db_sync.execute(
        select(TaskRun.status).where(TaskRun.id == ml_stale_id)
    ).scalar_one()
    assert ml_fresh_status == "running"
    assert ml_stale_status == "error"


def test_recover_stalled_task_runs_download_queue_uses_longer_threshold(db_sync):
    """download_source legitimately walks 5-25 min (Patreon/gallery-dl).
    The 5-min default flagged healthy in-flight walks as phantom
    'RecoverySweep' failures — visible in System Activity but absent from
    the Subscriptions view because the download actually finished ok.
    The 30-min download override (QUEUE_STUCK_THRESHOLD_MINUTES) must
    protect a 10-min-old download row while still flagging a 35-min one.
    Audit 2026-06-10."""
    from sqlalchemy import select

    from backend.app.models import TaskRun
    from backend.app.tasks.maintenance import recover_stalled_task_runs

    now = datetime.now(UTC)
    # 10-min-old download row: stale by the default 5-min rule but fresh
    # by the 30-min download override. Must survive the sweep.
    dl_fresh_id = _make_task_run(
        db_sync, status="running", queue="download",
        task_name="backend.app.tasks.download.download_source",
        started_at=now - timedelta(minutes=10),
    )
    # 35-min-old download row: past even the 30-min override (a genuine
    # hard kill). Must be flagged.
    dl_stale_id = _make_task_run(
        db_sync, status="running", queue="download",
        task_name="backend.app.tasks.download.download_source",
        started_at=now - timedelta(minutes=35),
    )
    db_sync.commit()

    recovered = recover_stalled_task_runs.apply().get()
    assert recovered == 1

    db_sync.expire_all()
    dl_fresh_status = db_sync.execute(
        select(TaskRun.status).where(TaskRun.id == dl_fresh_id)
    ).scalar_one()
    dl_stale_status = db_sync.execute(
        select(TaskRun.status).where(TaskRun.id == dl_stale_id)
    ).scalar_one()
    assert dl_fresh_status == "running"
    assert dl_stale_status == "error"


def test_download_stuck_threshold_exceeds_hard_time_limit():
    """Invariant guard (maintenance.py:112): every queue override MUST be
    ≥ the relevant task's hard time_limit, else the sweep flags in-flight
    work. download_source is the one that regressed — pin it so a future
    DOWNLOAD_HARD_TIME_LIMIT bump can't silently re-break it."""
    from backend.app.tasks.download import DOWNLOAD_HARD_TIME_LIMIT
    from backend.app.tasks.maintenance import QUEUE_STUCK_THRESHOLD_MINUTES

    hard_minutes = DOWNLOAD_HARD_TIME_LIMIT / 60
    assert QUEUE_STUCK_THRESHOLD_MINUTES["download"] >= hard_minutes


def test_recover_stalled_task_runs_archive_task_uses_longer_threshold(db_sync):
    """import_archive_file shares the 'import' queue with fast
    single-file import_media_file, so it gets a per-task-name override
    (40 min) while the import queue keeps its short threshold (10 min
    since #4432; import_media_file's hard limit is 6). A 10-min-old
    archive task-run must survive; a 50-min-old one is flagged.
    Operator-flagged 2026-05-28."""
    from sqlalchemy import select

    from backend.app.models import TaskRun
    from backend.app.tasks.maintenance import recover_stalled_task_runs

    archive_name = "backend.app.tasks.import_file.import_archive_file"
    now = datetime.now(UTC)
    # Fast single-file import on the same queue, 15 min old → flagged
    # by the import queue's 10-min threshold.
    media_id = _make_task_run(
        db_sync, status="running", queue="import",
        task_name="backend.app.tasks.import_file.import_media_file",
        started_at=now - timedelta(minutes=15),
    )
    # Archive on the same queue, 10 min old → survives (40-min override).
    archive_fresh_id = _make_task_run(
        db_sync, status="running", queue="import",
        task_name=archive_name,
        started_at=now - timedelta(minutes=10),
    )
    # Archive 50 min old → past even the 40-min override → flagged.
    archive_stale_id = _make_task_run(
        db_sync, status="running", queue="import",
        task_name=archive_name,
        started_at=now - timedelta(minutes=50),
    )
    db_sync.commit()

    recovered = recover_stalled_task_runs.apply().get()
    assert recovered == 2  # media + stale archive

    db_sync.expire_all()
    def _status(_id):
        return db_sync.execute(
            select(TaskRun.status).where(TaskRun.id == _id)
        ).scalar_one()
    assert _status(media_id) == "error"
    assert _status(archive_fresh_id) == "running"
    assert _status(archive_stale_id) == "error"


def test_recover_stalled_task_runs_external_fetch_uses_longer_threshold(db_sync):
    """fetch_external_link legitimately runs to its 60-min hard limit, but its
    TaskRun records queue='default' (no queue override), so before the
    task-name override (65 min) it fell to the 5-min default and healthy
    in-flight fetches were phantom-flagged 'RecoverySweep' before their own
    timeout/error could surface (operator-flagged 2026-06-17, target 414 swept
    at 6.6min). A 10-min-old row must survive; a 70-min-old one is flagged."""
    from sqlalchemy import select

    from backend.app.models import TaskRun
    from backend.app.tasks.maintenance import recover_stalled_task_runs

    name = "backend.app.tasks.external.fetch_external_link"
    now = datetime.now(UTC)
    # 10-min-old: stale by the default 5-min rule but fresh by the 65-min
    # task-name override. Must survive despite recording queue='default'.
    fresh_id = _make_task_run(
        db_sync, status="running", queue="default", task_name=name,
        started_at=now - timedelta(minutes=10),
    )
    # 70-min-old: past even the 65-min override (a genuine hard kill). Flagged.
    stale_id = _make_task_run(
        db_sync, status="running", queue="default", task_name=name,
        started_at=now - timedelta(minutes=70),
    )
    db_sync.commit()

    recovered = recover_stalled_task_runs.apply().get()
    assert recovered == 1

    db_sync.expire_all()
    def _status(_id):
        return db_sync.execute(
            select(TaskRun.status).where(TaskRun.id == _id)
        ).scalar_one()
    assert _status(fresh_id) == "running"
    assert _status(stale_id) == "error"


def test_external_fetch_stuck_threshold_exceeds_hard_time_limit():
    """Invariant guard (maintenance.py:112): the fetch_external_link task-name
    override MUST be ≥ its hard time_limit, else the sweep flags healthy long
    fetches. Pins it so a future time_limit bump can't silently re-break it."""
    from backend.app.tasks.external import fetch_external_link
    from backend.app.tasks.maintenance import TASK_STUCK_THRESHOLD_MINUTES

    hard_minutes = fetch_external_link.time_limit / 60
    override = TASK_STUCK_THRESHOLD_MINUTES[
        "backend.app.tasks.external.fetch_external_link"
    ]
    assert override >= hard_minutes


def test_prune_task_runs_deletes_ok_older_than_24h(db_sync):
    from sqlalchemy import select

    from backend.app.models import TaskRun
    from backend.app.tasks.maintenance import prune_task_runs

    now = datetime.now(UTC)
    old_id = _make_task_run(
        db_sync, status="ok",
        started_at=now - timedelta(hours=30),
        finished_at=now - timedelta(hours=29),
    )
    recent_id = _make_task_run(
        db_sync, status="ok",
        started_at=now - timedelta(hours=2),
        finished_at=now - timedelta(hours=1),
    )
    db_sync.commit()

    result = prune_task_runs.apply().get()
    assert result["ok_deleted"] == 1

    db_sync.expire_all()
    surviving_ids = set(db_sync.execute(
        select(TaskRun.id).where(TaskRun.id.in_([old_id, recent_id]))
    ).scalars().all())
    assert surviving_ids == {recent_id}


def test_prune_task_runs_deletes_failures_older_than_7d(db_sync):
    from sqlalchemy import select

    from backend.app.models import TaskRun
    from backend.app.tasks.maintenance import prune_task_runs

    now = datetime.now(UTC)
    old_id = _make_task_run(
        db_sync, status="error", error_type="OldError",
        started_at=now - timedelta(days=10),
        finished_at=now - timedelta(days=9),
    )
    # A later run of the same task, so the old failure is not its newest row
    # (the newest is kept for beat — see the test below).
    _make_task_run(
        db_sync, status="ok",
        started_at=now - timedelta(hours=2), finished_at=now - timedelta(hours=1),
    )
    db_sync.commit()

    result = prune_task_runs.apply().get()
    assert result["failures_deleted"] >= 1

    db_sync.expire_all()
    surviving = db_sync.execute(
        select(TaskRun.id).where(TaskRun.id == old_id)
    ).scalar_one_or_none()
    assert surviving is None


def test_prune_task_runs_keeps_each_tasks_newest_row_however_old(db_sync):
    """#4408: beat reads a job's last run from task_run. A weekly job's only
    row is older than the 24h ok-retention, and pruning it would make beat
    think the job never ran and fire it on every restart."""
    from sqlalchemy import select

    from backend.app.models import TaskRun
    from backend.app.tasks.maintenance import prune_task_runs

    now = datetime.now(UTC)
    weekly = "backend.app.tasks.fake.weekly"
    older = _make_task_run(
        db_sync, status="ok", task_name=weekly,
        started_at=now - timedelta(days=14), finished_at=now - timedelta(days=14),
    )
    newest = _make_task_run(
        db_sync, status="ok", task_name=weekly,
        started_at=now - timedelta(days=7), finished_at=now - timedelta(days=7),
    )
    db_sync.commit()

    prune_task_runs.apply().get()

    db_sync.expire_all()
    surviving = set(db_sync.execute(
        select(TaskRun.id).where(TaskRun.task_name == weekly)
    ).scalars().all())
    assert surviving == {newest}
    assert older not in surviving


def test_prune_task_runs_keeps_recent_failures(db_sync):
    from sqlalchemy import select

    from backend.app.models import TaskRun
    from backend.app.tasks.maintenance import prune_task_runs

    now = datetime.now(UTC)
    recent_id = _make_task_run(
        db_sync, status="error", error_type="RecentError",
        started_at=now - timedelta(days=6),
        finished_at=now - timedelta(days=5),
    )
    db_sync.commit()

    prune_task_runs.apply().get()

    db_sync.expire_all()
    surviving = db_sync.execute(
        select(TaskRun.id).where(TaskRun.id == recent_id)
    ).scalar_one_or_none()
    assert surviving == recent_id


def test_prune_task_runs_never_deletes_running(db_sync):
    """Even a 30-day-old running row stays — recovery sweep is the
    mechanism that flips them; prune doesn't touch in-flight."""
    from sqlalchemy import select

    from backend.app.models import TaskRun
    from backend.app.tasks.maintenance import prune_task_runs

    now = datetime.now(UTC)
    ancient_id = _make_task_run(
        db_sync, status="running",
        started_at=now - timedelta(days=30),
    )
    db_sync.commit()

    prune_task_runs.apply().get()

    db_sync.expire_all()
    surviving = db_sync.execute(
        select(TaskRun.id).where(TaskRun.id == ancient_id)
    ).scalar_one_or_none()
    assert surviving == ancient_id


# ---- recover_stalled_download_events ----------------------------------


def _make_source(session, *, slug: str) -> int:
    """Create an Artist + Source pair for the download-recovery tests."""
    from backend.app.models import Artist, Source

    artist = Artist(name=f"Artist {slug}", slug=slug)
    session.add(artist)
    session.flush()
    source = Source(
        artist_id=artist.id, platform="patreon",
        url=f"https://example.com/{slug}",
    )
    session.add(source)
    session.flush()
    return source.id


def test_recover_stalled_download_skips_fresh(db_sync):
    """A pending event whose started_at is under the 30-min threshold is
    left alone — the worker may still legitimately be processing it."""
    from sqlalchemy import select

    from backend.app.models import DownloadEvent, Source
    from backend.app.tasks.maintenance import recover_stalled_download_events

    sid = _make_source(db_sync, slug="fresh")
    now = datetime.now(UTC)
    db_sync.add(DownloadEvent(
        source_id=sid, status="pending", started_at=now - timedelta(seconds=30),
    ))
    db_sync.commit()

    recovered = recover_stalled_download_events.apply().get()

    assert recovered == 0
    db_sync.expire_all()
    status = db_sync.execute(
        select(DownloadEvent.status).where(DownloadEvent.source_id == sid)
    ).scalar_one()
    assert status == "pending"
    failures = db_sync.execute(
        select(Source.consecutive_failures).where(Source.id == sid)
    ).scalar_one()
    assert failures == 0


def test_download_lane_boot_closes_pre_boot_events_without_blaming_the_source(db_sync):
    """#4433: a restart SIGKILLs a walk past its stop grace and strands queued
    ones. At the download lane's boot, everything pending/running from before
    it ends as `skipped`/interrupted — and the source is left as it was, so it
    is due on the next tick instead of backed off as a failure."""
    from sqlalchemy import select

    from backend.app.models import DownloadEvent, Source
    from backend.app.services.download_recovery import (
        DOWNLOAD_INTERRUPTED_MESSAGE,
        interrupt_orphaned_download_events,
    )

    sid = _make_source(db_sync, slug="rebooted")
    booted_at = datetime.now(UTC)
    before = booted_at - timedelta(minutes=5)
    walking = DownloadEvent(
        source_id=sid, status="running", started_at=before,
        metadata_={"live": {"downloaded": 3}},
    )
    queued = DownloadEvent(source_id=sid, status="pending", started_at=before)
    finished = DownloadEvent(source_id=sid, status="ok", started_at=before)
    # Promoted to running after the boot: download_service resets started_at.
    fresh = DownloadEvent(
        source_id=sid, status="running", started_at=booted_at + timedelta(seconds=5),
    )
    db_sync.add_all([walking, queued, finished, fresh])
    db_sync.commit()

    closed = interrupt_orphaned_download_events(db_sync, booted_at=booted_at)
    db_sync.commit()

    assert closed == 2
    db_sync.expire_all()
    for ev in (walking, queued):
        assert ev.status == "skipped"
        assert ev.error == DOWNLOAD_INTERRUPTED_MESSAGE
        assert ev.finished_at is not None
        assert ev.metadata_["error_type"] == "interrupted"
    assert walking.metadata_["live"] == {"downloaded": 3}
    assert finished.status == "ok"
    assert fresh.status == "running"
    src = db_sync.execute(
        select(Source.consecutive_failures, Source.last_error, Source.last_checked_at)
        .where(Source.id == sid)
    ).one()
    assert src.consecutive_failures == 0
    assert src.last_error is None
    assert src.last_checked_at is None


def test_recover_stalled_download_flips_stale_pending(db_sync):
    """A 2-hour-old pending event flips to error AND the source is bumped
    (consecutive_failures, last_error, last_checked_at) so the next scan
    tick can re-queue it (the in-flight guard no longer blocks)."""
    from sqlalchemy import select

    from backend.app.models import DownloadEvent, Source
    from backend.app.tasks.maintenance import recover_stalled_download_events

    sid = _make_source(db_sync, slug="stale-p")
    now = datetime.now(UTC)
    db_sync.add(DownloadEvent(
        source_id=sid, status="pending", started_at=now - timedelta(hours=2),
    ))
    db_sync.commit()

    recovered = recover_stalled_download_events.apply().get()

    assert recovered == 1
    db_sync.expire_all()
    ev_row = db_sync.execute(
        select(
            DownloadEvent.status, DownloadEvent.finished_at, DownloadEvent.error,
        ).where(DownloadEvent.source_id == sid)
    ).one()
    assert ev_row.status == "error"
    assert ev_row.finished_at is not None
    assert "stranded" in ev_row.error
    src_row = db_sync.execute(
        select(
            Source.consecutive_failures, Source.last_error, Source.last_checked_at,
        ).where(Source.id == sid)
    ).one()
    assert src_row.consecutive_failures == 1
    assert "stranded" in src_row.last_error
    assert src_row.last_checked_at is not None


def test_recover_stalled_download_flips_stale_running(db_sync):
    """'running' is the other in-flight state — recovery covers it equally."""
    from sqlalchemy import select

    from backend.app.models import DownloadEvent
    from backend.app.tasks.maintenance import recover_stalled_download_events

    sid = _make_source(db_sync, slug="stale-r")
    now = datetime.now(UTC)
    db_sync.add(DownloadEvent(
        source_id=sid, status="running", started_at=now - timedelta(hours=2),
    ))
    db_sync.commit()

    recovered = recover_stalled_download_events.apply().get()

    assert recovered == 1
    db_sync.expire_all()
    status = db_sync.execute(
        select(DownloadEvent.status).where(DownloadEvent.source_id == sid)
    ).scalar_one()
    assert status == "error"


def test_recover_stalled_download_dedupes_per_source(db_sync):
    """Two stale events on one source bump consecutive_failures ONCE.
    Backoff is exponential on that counter (2^failures), so per-event bumps
    would inflate the next check interval by 2^N for no real reason."""
    from sqlalchemy import select

    from backend.app.models import DownloadEvent, Source
    from backend.app.tasks.maintenance import recover_stalled_download_events

    sid = _make_source(db_sync, slug="dedupe")
    now = datetime.now(UTC)
    db_sync.add_all([
        DownloadEvent(
            source_id=sid, status="pending",
            started_at=now - timedelta(hours=2),
        ),
        DownloadEvent(
            source_id=sid, status="running",
            started_at=now - timedelta(hours=3),
        ),
    ])
    db_sync.commit()

    recovered = recover_stalled_download_events.apply().get()

    assert recovered == 2
    db_sync.expire_all()
    failures = db_sync.execute(
        select(Source.consecutive_failures).where(Source.id == sid)
    ).scalar_one()
    assert failures == 1


def test_vacuum_analyze_runs_over_high_churn_tables():
    """VACUUM (ANALYZE) runs (on its own AUTOCOMMIT connection) and reports the
    tables it touched."""
    from backend.app.tasks.maintenance import VACUUM_TABLES, vacuum_analyze

    result = vacuum_analyze.apply().get()
    assert result["vacuumed"] == list(VACUUM_TABLES)


def test_reclaim_attachments_stuck_threshold_exceeds_hard_time_limit():
    """#883's invariant, applied to the attachment reclaim: a task whose stall
    threshold is under its own hard limit gets phantom-flagged 'RecoverySweep'
    while it is still healthily running."""
    from backend.app.tasks.admin import reclaim_orphaned_attachments_task
    from backend.app.tasks.maintenance import TASK_STUCK_THRESHOLD_MINUTES

    hard_minutes = reclaim_orphaned_attachments_task.time_limit / 60
    override = TASK_STUCK_THRESHOLD_MINUTES[
        "backend.app.tasks.admin.reclaim_orphaned_attachments_task"
    ]
    assert override >= hard_minutes