diff --git a/alembic/versions/0117_external_link_started_at.py b/alembic/versions/0117_external_link_started_at.py new file mode 100644 index 0000000..8d38dab --- /dev/null +++ b/alembic/versions/0117_external_link_started_at.py @@ -0,0 +1,33 @@ +"""external_link.started_at: when a fetch claimed the link. + +A worker killed mid-fetch (the Celery hard limit, an OOM, a restart) never +reaches the code that moves its link out of 'downloading'. The external-link +sweep returns such a link to the retry path once it has been 'downloading' +longer than the task's hard limit, which needs the claim time. + +A row in 'downloading' with no claim time is treated as stale: no worker +survives the deploy that adds this column. + +Revision ID: 0117 +Revises: 0116 +Create Date: 2026-11-05 + +""" +import sqlalchemy as sa +from alembic import op + +revision = "0117" +down_revision = "0116" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "external_link", + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("external_link", "started_at") diff --git a/backend/app/models/external_link.py b/backend/app/models/external_link.py index 4882125..b83aa8a 100644 --- a/backend/app/models/external_link.py +++ b/backend/app/models/external_link.py @@ -86,6 +86,11 @@ class ExternalLink(Base): created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) + # When the current fetch claimed the link ('downloading'). The sweep uses it + # to recover a link whose worker died mid-fetch. + started_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) completed_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True ) diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index 167a1e0..1b0e948 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -369,11 +369,8 @@ class DownloadService: if do_embed: embed_image.delay(img_id) elif result.status == "attached": - # Non-media or extracted archive captured as PostAttachment - # (FC-2d-iii). The canonical copy lives in the attachments - # store; the original download path is now redundant — - # mirror duplicate_hash cleanup so we don't keep two copies. - # Operator-flagged 2026-06-02 (Lustria OST zip). + # Non-media, or an archive with no new images, captured as a + # PostAttachment; the attachments store holds the copy. import_summary["attached"] += 1 # An archive captured WITHOUT extracting any image carries a # reason on result.error — record it so the event explains the @@ -390,35 +387,13 @@ class DownloadService: bytes_downloaded += path.stat().st_size # noqa: ASYNC240 except OSError: pass - try: - path.unlink(missing_ok=True) # noqa: ASYNC240 - except OSError: - pass - elif result.status == "skipped" and result.skip_reason and result.skip_reason.value in ( - "duplicate_hash", "duplicate_phash", - ): - import_summary["skipped"] += 1 - try: - path.unlink(missing_ok=True) # noqa: ASYNC240 - except OSError: - pass elif result.status == "skipped": - # Soft skip (too_small, too_transparent, invalid_image) — - # the file just didn't qualify, not a download/ingest - # failure. Don't flag the run as error; the file stays - # on disk for operator inspection. + # A duplicate, or a file the import filters rejected. Neither + # is a download failure, so the run is not flagged. import_summary["skipped"] += 1 elif result.status == "failed": - # Hard failure (today only: archive probe crash/timeout). - # The original archive sits in /images/ as an orphan; the - # filesystem scanner would re-import and re-crash on the - # same file, so delete the source file and surface the - # error in import_summary. Audit 2026-06-02. + # Hard failure: an archive probe crashed or timed out. import_summary["errors"] += 1 - try: - path.unlink(missing_ok=True) # noqa: ASYNC240 - except OSError: - pass elif result.status == "refreshed": # Currently unreachable from attach_in_place (the download # path never runs in deep=True mode), but the importer's @@ -427,6 +402,10 @@ class DownloadService: import_summary["attached"] += 1 else: import_summary["errors"] += 1 + # Remove the download when the library no longer needs it at this + # path: an attachment or archive now in the store, a duplicate, a + # rejected file. One policy for every download path. + self.importer.discard_after_attach(path, result) # Post-only records: media-less posts (pure-text) the native ingester # captured so the artist archive is complete. Upsert each (keyed on diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py index 9e28d07..9715053 100644 --- a/backend/app/services/importer.py +++ b/backend/app/services/importer.py @@ -1401,6 +1401,47 @@ class Importer: self.session.commit() return ImportResult(status="imported", image_id=record.id) + def discard_after_attach(self, path: Path, result: ImportResult) -> bool: + """Delete a downloaded file that `attach_in_place` has finished with, + when keeping it would only leave a second copy or a rejected file on + disk. Returns True when the file was removed. Both download paths (a + subscription walk's import step and the external-link worker) apply + this one policy: + + - attached: the attachments store holds the canonical copy. + - imported or superseded ARCHIVE: its new images are in the library and + the archive itself is in the attachments store. + - duplicate skip: the library already holds the content. + - soft skip (too small, too transparent, single colour, not a readable + image): the import filters rejected it. A file left on disk with no + record is picked up by the next walk and imported again; the seen + ledger is what stops it being downloaded again. + - failed: the archive probe crashed, and a retry would crash again. + + An imported or superseded media file is kept: it IS the record's file. + So is any path an ImageRecord points at, whatever the status, because a + re-fetch can write identical bytes over a record's own file.""" + if result.status in ("imported", "superseded"): + redundant = is_archive(path) + else: + redundant = result.status in ("attached", "skipped", "failed") + if not redundant: + return False + owned = self.session.execute( + select(ImageRecord.id).where( + ImageRecord.path.in_({str(path), str(path.resolve())}) + ).limit(1) + ).first() + if owned is not None: + log.info("keeping %s: it is image %s's file", path.name, owned.id) + return False + try: + path.unlink(missing_ok=True) + except OSError as exc: + log.warning("could not remove %s: %s", path, exc) + return False + return True + def _upsert_artist(self, name: str) -> Artist: slug = slugify(name) artist = self.session.execute( diff --git a/backend/app/tasks/external.py b/backend/app/tasks/external.py index 327d89b..a3ac844 100644 --- a/backend/app/tasks/external.py +++ b/backend/app/tasks/external.py @@ -24,7 +24,7 @@ from datetime import UTC, datetime, timedelta from pathlib import Path import redis -from sqlalchemy import delete, select, update +from sqlalchemy import delete, or_, select, update from ..celery_app import celery from ..config import get_config @@ -54,6 +54,9 @@ _RETENTION_DAYS = 30 # than one fetch would have taken. _LOCK_PREFIX = "fc:extdl_lock:" _LOCK_TTL = 2400 +# A link 'downloading' this long has outlived fetch_external_link's hard limit +# (time_limit=3600), so its worker is gone and the sweep recovers it. +_STALL_MINUTES = 65 _SERIALIZE_COUNTDOWN = 120 _MAX_SERIALIZE_WAITS = 30 @@ -106,7 +109,7 @@ def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict: ExternalLink.id == link_id, ExternalLink.status.in_(("pending", "failed")), ) - .values(status="downloading") + .values(status="downloading", started_at=datetime.now(UTC)) .returning(ExternalLink.id) ).first() session.commit() @@ -180,6 +183,10 @@ def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict: / str(post.external_post_id) / str(link_id) ) try: + # Clear what an earlier, interrupted attempt left behind, so this fetch + # starts from the link's imported files only. + with SessionLocal() as session: + _discard_unrecorded(session, post_dir) result = fetch_external(host, url, post_dir) with SessionLocal() as session: link = session.get(ExternalLink, link_id) @@ -190,6 +197,7 @@ def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict: ) _record_failure(link, attempts, result.error or "fetch failed") session.commit() + _discard_unrecorded(session, post_dir) return {"link_id": link_id, "error": result.error} log.info( @@ -231,6 +239,7 @@ def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict: if link is not None and link.status == "downloading": _record_failure(link, attempts, f"{type(exc).__name__}: {exc}") session.commit() + _discard_unrecorded(session, post_dir) raise finally: if lock is not None: @@ -287,10 +296,9 @@ def _route_files(session, files, post, platform, artist, source) -> list[int]: post.id, f.name, result.status, len(ids), ) elif result.status == "attached": - # Non-art / archive captured as a PostAttachment (copied into the - # store) — drop the on-disk original. An archive captured WITHOUT - # extracting any image carries the reason (the recurring "zip but no - # images" symptom, #718) — surface it loudly. + # Non-art, or an archive with no new images, captured as a + # PostAttachment. An archive captured WITHOUT extracting any image + # carries the reason, so the "zip but no images" case is visible. if result.error: log.warning( "extdl archive captured UNEXTRACTED: post=%s file=%s reason=%s", @@ -298,51 +306,91 @@ def _route_files(session, files, post, platform, artist, source) -> list[int]: ) else: log.info("extdl attachment captured: post=%s file=%s", post.id, f.name) - f.unlink(missing_ok=True) elif result.status == "skipped": reason = result.skip_reason.value if result.skip_reason else None log.info( "extdl skipped: post=%s file=%s reason=%s", post.id, f.name, reason, ) - if reason in ("duplicate_hash", "duplicate_phash"): - # Path-safe unlink: external files are imported IN PLACE, so `f` - # can BE the canonical record's backing file (same content - # re-fetched / a colliding name). Only delete `f` when it is NOT - # the existing record's file — otherwise we orphan the record and - # playback 404s. (#859) - canonical = None - if result.image_id is not None: - rec = session.get(ImageRecord, result.image_id) - if rec is not None: - canonical = (IMAGES_ROOT / rec.path).resolve() - if canonical != f.resolve(): - f.unlink(missing_ok=True) # a redundant copy — safe to drop - else: - log.info( - "extdl keep: %s IS record %s's canonical file — not unlinking", - f.name, result.image_id, - ) elif result.status == "failed": log.warning( "extdl import FAILED: post=%s file=%s error=%s", post.id, f.name, result.error, ) - f.unlink(missing_ok=True) else: # 'refreshed' or any future status — work happened, no new images log.info( "extdl import: post=%s file=%s status=%s", post.id, f.name, result.status, ) + # Same policy as the subscription download path: drop the file unless + # the library needs it at this path. A file imported in place is kept. + importer.discard_after_attach(f, result) # The synthesized sidecar has done its job — don't litter the tree. Path(str(f) + ".json").unlink(missing_ok=True) return image_ids +def _discard_unrecorded(session, link_dir: Path) -> int: + """Delete every file in a link's staging dir that no ImageRecord points at, + then the dirs left empty. The dir belongs to one link, so anything in it + without a record is a partial or abandoned download from an earlier + attempt (megatools' `.megatmp.*`, a killed transfer, a rejected file). + Images imported in place stay. Returns the number of files removed.""" + if not link_dir.is_dir(): + return 0 + files = [p for p in link_dir.rglob("*") if p.is_file()] + if not files: + return 0 + recorded = set(session.execute( + select(ImageRecord.path).where( + ImageRecord.path.in_([str(p) for p in files]) + ) + ).scalars()) + removed = 0 + for p in files: + if str(p) in recorded: + continue + try: + p.unlink(missing_ok=True) + removed += 1 + except OSError as exc: + log.warning("extdl could not remove %s: %s", p, exc) + for d in sorted((p for p in link_dir.rglob("*") if p.is_dir()), reverse=True): + try: + d.rmdir() + except OSError: + pass # not empty: it holds an imported file + try: + link_dir.rmdir() + except OSError: + pass + if removed: + log.info("extdl removed %d leftover file(s) from %s", removed, link_dir) + return removed + + @celery.task(name="backend.app.tasks.external.sweep_external_links") def sweep_external_links() -> dict: """Enqueue a bounded batch of actionable links (pending, or failed below the - dead-letter threshold). Driven by the post-download hook and a beat tick.""" + dead-letter threshold). Driven by the post-download hook and a beat tick. + + First it recovers links whose worker died mid-fetch: a link 'downloading' + for longer than the task's hard limit has no worker left, so it counts as a + failed attempt and goes back to the retry path (dead-lettered at the + threshold like any other failure). The next attempt clears its leftovers.""" SessionLocal = _sync_session_factory() with SessionLocal() as session: + now = datetime.now(UTC) + stale = session.execute( + select(ExternalLink).where( + ExternalLink.status == "downloading", + or_( + ExternalLink.started_at.is_(None), + ExternalLink.started_at < now - timedelta(minutes=_STALL_MINUTES), + ), + ) + ).scalars().all() + for link in stale: + _record_failure(link, link.attempts, "worker stopped mid-fetch") + session.commit() ids = session.execute( select(ExternalLink.id).where( ExternalLink.status.in_(("pending", "failed")), @@ -351,7 +399,7 @@ def sweep_external_links() -> dict: ).scalars().all() for link_id in ids: fetch_external_link.delay(link_id) - return {"enqueued": len(ids)} + return {"enqueued": len(ids), "recovered": len(stale)} @celery.task(name="backend.app.tasks.external.recover_external_links") diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 51558c0..1a60b94 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -88,8 +88,18 @@ TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600 # 7 days # keeps it from deleting an in-flight download's staging file mid-write. IMAGES_ROOT = Path("/images") TEMP_STAGING_SUFFIXES = (".part", ".partial") +# megatools stages a Mega download as `.megatmp.` in the target dir. +TEMP_STAGING_PREFIXES = (".megatmp.",) ORPHAN_TEMP_MIN_AGE_HOURS = 6 + +def is_temp_staging_file(path: Path) -> bool: + """True for a downloader's staging file: one that never becomes library + content under that name, so a stale one is safe to delete.""" + return path.suffix in TEMP_STAGING_SUFFIXES or path.name.startswith( + TEMP_STAGING_PREFIXES + ) + # Audit 2026-06-02: per-entity recovery sweep thresholds. Each must be # > the entity's longest legitimate runtime (its task's time_limit + a # small buffer) so the sweep never flags in-flight work. @@ -373,16 +383,17 @@ def cleanup_old_tasks() -> int: @celery.task(name="backend.app.tasks.maintenance.cleanup_orphaned_temp_files") def cleanup_orphaned_temp_files() -> int: - """Delete orphaned .part/.partial staging files under the images root, left by - a download/import killed mid-write. Only removes files older than - ORPHAN_TEMP_MIN_AGE_HOURS so an in-flight download's staging file is never - pulled out from under it. Returns the count removed.""" + """Delete orphaned staging files (.part, .partial, megatools' .megatmp.*) + under the images root, left by a download/import killed mid-write. Only + removes files older than ORPHAN_TEMP_MIN_AGE_HOURS so an in-flight + download's staging file is never pulled out from under it. Returns the + count removed.""" if not IMAGES_ROOT.is_dir(): return 0 cutoff = datetime.now(UTC).timestamp() - ORPHAN_TEMP_MIN_AGE_HOURS * 3600 removed = 0 for path in IMAGES_ROOT.rglob("*"): - if path.suffix not in TEMP_STAGING_SUFFIXES or not path.is_file(): + if not is_temp_staging_file(path) or not path.is_file(): continue try: if path.stat().st_mtime >= cutoff: diff --git a/tests/test_download_service.py b/tests/test_download_service.py index b8f1dc2..8c12f88 100644 --- a/tests/test_download_service.py +++ b/tests/test_download_service.py @@ -193,6 +193,69 @@ async def test_download_source_attaches_written_files( assert imported_when_marked == [2] +@pytest.mark.asyncio +async def test_import_step_keeps_only_the_files_the_library_needs_on_disk( + db, db_sync, tmp_path, seed_artist_and_source, +): + """After the import step, a downloaded image the library took stays where it + is, while an archive whose images were extracted (the archive is in the + attachments store) and an image the import filters rejected are removed, so + nothing is left on disk without a record for a later walk to import again.""" + import zipfile + + from PIL import Image + + from backend.app.models import PostAttachment + from backend.app.services.download_service import DownloadService + from backend.app.services.importer import Importer + + _artist, source = seed_artist_and_source + images_root = tmp_path / "images" + post_dir = images_root / "alice" / "patreon" / "post" + + kept = post_dir / "a.jpg" + _make_jpg(kept, split="h") + member = tmp_path / "member.jpg" + _make_jpg(member, split="v") + archive = post_dir / "pack.zip" + with zipfile.ZipFile(archive, "w") as z: + z.writestr("art.jpg", member.read_bytes()) + tiny = post_dir / "tiny.png" + Image.new("RGB", (16, 16), (200, 30, 30)).save(tiny) # below min_width + + result = _make_fake_dl_result( + success=True, written_paths=[str(kept), str(archive), str(tiny)], + files_downloaded=3, + ) + sync_settings = db_sync.execute( + select(ImportSettings).where(ImportSettings.id == 1) + ).scalar_one() + sync_settings.phash_threshold = 0 + sync_settings.min_width = 32 + importer = Importer( + session=db_sync, images_root=images_root, import_root=images_root, + thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings, + ) + crypto = CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True) + svc = DownloadService( + async_session=db, sync_session=db_sync, gdl=_fake_gdl_with_result(result), + importer=importer, cred_service=CredentialService(db, crypto), + ) + _stub_patreon_ingester(svc, result) + await svc.download_source(source.id) + + assert kept.exists() + assert not archive.exists() + assert not tiny.exists() + # The archive and its image are both still in the library. + assert db_sync.execute( + select(func.count(PostAttachment.id)).where( + PostAttachment.original_filename == "pack.zip" + ) + ).scalar_one() == 1 + assert db_sync.execute(select(func.count(ImageRecord.id))).scalar_one() == 2 + + @pytest.mark.asyncio async def test_download_source_skipped_when_disabled( db, db_sync, tmp_path, seed_artist_and_source, diff --git a/tests/test_external_worker.py b/tests/test_external_worker.py index 7d0eb59..01f7b49 100644 --- a/tests/test_external_worker.py +++ b/tests/test_external_worker.py @@ -251,3 +251,154 @@ def test_downloaded_archive_gets_provenance_and_tagging(db_sync, tmp_path, monke # Tagging + thumbnails queued for exactly those member images. assert set(tagged) == member_ids assert set(thumbed) == member_ids + + +def _link_dir(tmp_path, host, link_id): + return tmp_path / f"ext-{host}" / "patreon" / "external" / f"EXT-{host}" / str(link_id) + + +def test_extracted_archive_is_not_left_in_the_link_dir(db_sync, tmp_path, monkeypatch): + """An archive whose images were extracted is in the attachments store, so + the downloaded copy is removed instead of sitting beside it.""" + _, link = _seed(db_sync, host="mega") + link_id = link.id + monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path) + monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis()) + written = [] + + def fake_fetch(host, url, dest_dir, **kwargs): + dest_dir.mkdir(parents=True, exist_ok=True) + zpath = dest_dir / "pack.zip" + with zipfile.ZipFile(zpath, "w") as z: + z.writestr("art.jpg", _structured_jpeg_bytes()) + written.append(zpath) + return FetchResult(files=[zpath], bytes=zpath.stat().st_size) + + monkeypatch.setattr(ext, "fetch_external", fake_fetch) + import backend.app.tasks.ml as ml_mod + import backend.app.tasks.thumbnail as thumb_mod + monkeypatch.setattr(ml_mod.embed_image, "delay", lambda i: None) + monkeypatch.setattr(thumb_mod.generate_thumbnail, "delay", lambda i: None) + + out = ext.fetch_external_link(link_id) + assert out["images"] >= 1 + assert not written[0].exists() + db_sync.expire_all() + assert db_sync.execute( + select(func.count(PostAttachment.id)) + .where(PostAttachment.original_filename == "pack.zip") + ).scalar_one() == 1 + + +def test_a_failed_fetch_leaves_no_partial_files(db_sync, tmp_path, monkeypatch): + """A fetch that fails midway (megatools leaves `.megatmp.*` behind) clears + the link's staging dir, so a dead link leaves nothing on disk.""" + _, link = _seed(db_sync, host="mega") + link_id = link.id + monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path) + monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis()) + + def fake_fetch(host, url, dest_dir, **kwargs): + dest_dir.mkdir(parents=True, exist_ok=True) + (dest_dir / ".megatmp.AbCd").write_bytes(b"half a game build") + return FetchResult(error="megatools dl exit 1: timed out") + + monkeypatch.setattr(ext, "fetch_external", fake_fetch) + ext.fetch_external_link(link_id) + + link_dir = _link_dir(tmp_path, "mega", link_id) + assert not (link_dir / ".megatmp.AbCd").exists() + db_sync.expire_all() + assert db_sync.get(ExternalLink, link_id).status == "failed" + + +def test_a_retry_clears_what_a_killed_attempt_left(db_sync, tmp_path, monkeypatch): + """Leftovers from an attempt whose worker was killed are removed before the + next fetch starts; an image imported in place by an earlier attempt stays.""" + _, link = _seed(db_sync, host="pixeldrain") + link_id = link.id + monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path) + monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis()) + link_dir = _link_dir(tmp_path, "pixeldrain", link_id) + link_dir.mkdir(parents=True) + leftover = link_dir / "build.zip.part" + leftover.write_bytes(b"partial") + imported = link_dir / "kept.jpg" + imported.write_bytes(_structured_jpeg_bytes()) + db_sync.add(ImageRecord( + path=str(imported), sha256="ext-kept".ljust(64, "x"), + size_bytes=imported.stat().st_size, mime="image/jpeg", + width=256, height=256, origin="downloaded", integrity_status="ok", + )) + db_sync.commit() + seen_at_fetch = [] + + def fake_fetch(host, url, dest_dir, **kwargs): + seen_at_fetch.extend(sorted(p.name for p in dest_dir.iterdir())) + return FetchResult(error="HTTP 503") + + monkeypatch.setattr(ext, "fetch_external", fake_fetch) + ext.fetch_external_link(link_id) + + assert seen_at_fetch == ["kept.jpg"] + assert imported.exists() + assert not leftover.exists() + + +def test_a_rejected_download_is_removed(db_sync, tmp_path, monkeypatch): + """A downloaded image the import filters reject has no record, so it is + removed rather than left in the link's dir.""" + from backend.app.models import ImportSettings + + _, link = _seed(db_sync, host="dropbox") + link_id = link.id + settings = db_sync.execute( + select(ImportSettings).where(ImportSettings.id == 1) + ).scalar_one() + settings.min_width = 512 + db_sync.commit() + monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path) + monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis()) + written = [] + + def fake_fetch(host, url, dest_dir, **kwargs): + dest_dir.mkdir(parents=True, exist_ok=True) + f = dest_dir / "small.jpg" # 256px wide, below min_width + f.write_bytes(_structured_jpeg_bytes()) + written.append(f) + return FetchResult(files=[f], bytes=f.stat().st_size) + + monkeypatch.setattr(ext, "fetch_external", fake_fetch) + out = ext.fetch_external_link(link_id) + + assert out["images"] == 0 + assert not written[0].exists() + db_sync.expire_all() + assert db_sync.get(ExternalLink, link_id).status == "downloaded" + + +def test_sweep_recovers_a_link_whose_worker_died(db_sync, monkeypatch): + """A link 'downloading' past the task's hard limit (or with no claim time) + counts as a failed attempt and is retried; a fresh claim is left alone.""" + from datetime import UTC, datetime, timedelta + + _, stale = _seed(db_sync, host="mega", status="downloading") + _, unclaimed = _seed(db_sync, host="gdrive", status="downloading") + _, fresh = _seed(db_sync, host="dropbox", status="downloading") + now = datetime.now(UTC) + stale.started_at = now - timedelta(minutes=ext._STALL_MINUTES + 5) + fresh.started_at = now - timedelta(minutes=5) + db_sync.commit() + stale_id, unclaimed_id, fresh_id = stale.id, unclaimed.id, fresh.id + + enqueued = [] + monkeypatch.setattr(ext.fetch_external_link, "delay", lambda lid: enqueued.append(lid)) + out = ext.sweep_external_links() + + assert out["recovered"] == 2 + assert set(enqueued) == {stale_id, unclaimed_id} + db_sync.expire_all() + for lid in (stale_id, unclaimed_id): + row = db_sync.get(ExternalLink, lid) + assert (row.status, row.attempts) == ("failed", 1) + assert db_sync.get(ExternalLink, fresh_id).status == "downloading" diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index c9df368..8982c96 100644 --- a/tests/test_maintenance.py +++ b/tests/test_maintenance.py @@ -41,9 +41,13 @@ def test_cleanup_orphaned_temp_files_removes_stale_only(tmp_path, monkeypatch): fresh.write_bytes(b"x") keep = tmp_path / "real.jpg" # a real image → keep keep.write_bytes(b"x") + mega = tmp_path / "artist" / ".megatmp.AbCd123" # abandoned megatools stage + mega.write_bytes(b"x") + os.utime(mega, (old, old)) - assert m.cleanup_orphaned_temp_files() == 1 + assert m.cleanup_orphaned_temp_files() == 2 assert not stale.exists() + assert not mega.exists() assert fresh.exists() assert keep.exists()