From 621c2b8315f676724454a9a297a2992b847c4e8f Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 25 Sep 2026 11:36:17 -0400 Subject: [PATCH] fix: a post record is marked seen only after it is upserted, and an hourly sweep dates the posts killed walks left undated (#4436) ingest_core marked a post's record key seen when it wrote _post.json, but the record reaches the database (and dates the post and its images) only in phase 3. A walk killed before phase 3 left posts the ledger called recorded and the database never dated; ticks early-out long before reaching them again. Keys now join the media in mark_seen_after_import. date_posts_from_records finds each undated native post's record under its artist's folder and upserts it with the post's own source. Hourly on maintenance_long; an empty query once everything is dated. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR --- backend/app/celery_app.py | 7 ++ backend/app/services/ingest_core.py | 11 ++- backend/app/services/post_record_repair.py | 86 ++++++++++++++++++++++ backend/app/tasks/maintenance.py | 26 +++++++ tests/test_celery_routing.py | 1 + tests/test_patreon_ingester.py | 7 +- tests/test_sidecar_import.py | 48 ++++++++++++ 7 files changed, 182 insertions(+), 4 deletions(-) create mode 100644 backend/app/services/post_record_repair.py diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 6843589..63313ed 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -69,6 +69,8 @@ def make_celery() -> Celery: # up behind it (2026-09-24: 7 waiting, "all workers busy for 18 # minutes"). An exact name wins over the glob above. "backend.app.tasks.maintenance.backfill_phash": {"queue": "maintenance_long"}, + # Walks a folder tree per artist with undated posts (#4436). + "backend.app.tasks.maintenance.date_posts_from_records": {"queue": "maintenance_long"}, "backend.app.tasks.backup.*": {"queue": "maintenance_long"}, "backend.app.tasks.admin.*": {"queue": "maintenance_long"}, "backend.app.tasks.library_audit.*": {"queue": "maintenance_long"}, @@ -156,6 +158,11 @@ def make_celery() -> Celery: "schedule": 86400.0, # daily — sweep .part/.partial left by a # download/import killed mid-write (graceful-shutdown fallout) }, + "date-posts-from-records-hourly": { + "task": "backend.app.tasks.maintenance.date_posts_from_records", + "schedule": 3600.0, # an empty query once every native post is + # dated; otherwise upserts the _post.json a killed walk left (#4436) + }, "backfill-phash-daily": { "task": "backend.app.tasks.maintenance.backfill_phash", "schedule": 86400.0, # daily — NULL-only, so a no-op once the diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py index ffc461a..2c8a859 100644 --- a/backend/app/services/ingest_core.py +++ b/backend/app/services/ingest_core.py @@ -289,6 +289,11 @@ class Ingester: # Media handed to phase 3 for import. Marked seen by phase 3 once the # import has run (`mark_seen_after_import`), not here — see there. fetched: list[tuple[str, str]] = [] + # Post-record keys written this walk. Marked with the media, after + # phase 3 has upserted the records — marking them at write time left a + # walk killed before phase 3 with posts the ledger calls recorded that + # the database never dated, and no later tick walks back to them (#4436). + recorded: list[tuple[str, str]] = [] downloaded = 0 errors = 0 quarantined = 0 @@ -342,7 +347,9 @@ class Ingester: written_paths=written, post_record_paths=list(post_records), relink_source_paths=list(relink), - mark_seen_after_import=lambda: self._mark_seen(source_id, fetched), + mark_seen_after_import=lambda: self._mark_seen( + source_id, fetched + recorded, + ), stdout="\n".join(log_lines), stderr="", return_code=return_code, @@ -511,7 +518,7 @@ class Ingester: posts_with_body += 1 if rec.path is not None: post_records.append(str(rec.path)) - self._mark_seen(source_id, [(pkey, ppid)]) + recorded.append((pkey, ppid)) # Per-post handling line in the run stdout (the existing # "Raw stdout" panel) — the downloader already read the # post; we only format its outcome here. post_type beside diff --git a/backend/app/services/post_record_repair.py b/backend/app/services/post_record_repair.py new file mode 100644 index 0000000..7043fd4 --- /dev/null +++ b/backend/app/services/post_record_repair.py @@ -0,0 +1,86 @@ +"""Date the posts whose record reached the disk but never the database (#4436). + +A native walk writes each post's record (`_post.json`, or Discord's +`__post.json`) as it goes, and phase 3 upserts those records +after the walk — which is when the post, and through it its images, get their +date. Until #4436 the walk marked a record's post key seen at write time, so a +walk killed before phase 3 (a restart, a stall) left the post undated with the +ledger saying it was done, and no later tick walked back that far to fix it. + +The record files are still on disk. This finds each undated native post's +record under its artist's folder and upserts it with the post's OWN source — +never the (artist, platform) lookup, which picks the artist's first source and +would misfile a Discord channel's post (#4435). Idempotent: a post that is +already dated is never looked at, and upserting a record twice changes nothing. +""" +from __future__ import annotations + +import json +import logging +from collections import defaultdict +from pathlib import Path + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from ..models import Artist, Post, Source +from ..utils.sidecar import parse_sidecar +from .download_backends import NATIVE_INGESTER_PLATFORMS + +log = logging.getLogger(__name__) + + +def _is_record(path: Path) -> bool: + return path.name == "_post.json" or path.name.endswith("_post.json") + + +def _record_id(path: Path) -> str | None: + try: + data = json.loads(path.read_text("utf-8")) + except (OSError, ValueError): + return None + if not isinstance(data, dict): + return None + return parse_sidecar(data).external_post_id + + +def date_posts_from_records(session: Session, importer, images_root: Path) -> dict: + """Upsert the on-disk record of every undated native post. Returns counts.""" + rows = session.execute( + select(Post.external_post_id, Post.source_id, Post.artist_id, Source.platform) + .join(Source, Source.id == Post.source_id) + .where( + Post.post_date.is_(None), + Post.synthesized_by.is_(None), + Source.platform.in_(NATIVE_INGESTER_PLATFORMS), + ) + ).all() + wanted: dict[tuple[int, str], dict[str, int]] = defaultdict(dict) + for epid, source_id, artist_id, platform in rows: + wanted[(artist_id, platform)][epid] = source_id + + summary = {"undated": len(rows), "dated": 0, "no_record": 0} + for (artist_id, platform), by_epid in wanted.items(): + artist = session.get(Artist, artist_id) + root = Path(images_root) / artist.slug / platform if artist else None + if root is None or not root.is_dir(): + summary["no_record"] += len(by_epid) + continue + found = 0 + for record in root.rglob("*post.json"): + if not _is_record(record): + continue + epid = _record_id(record) + source_id = by_epid.pop(epid, None) if epid else None + if source_id is None: + continue + source = session.get(Source, source_id) + if importer.upsert_post_record(record, artist=artist, source=source): + found += 1 + if not by_epid: + break + summary["dated"] += found + summary["no_record"] += len(by_epid) + if summary["undated"]: + log.info("date_posts_from_records: %s", summary) + return summary diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 31c0fda..51558c0 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -731,6 +731,32 @@ def recover_stalled_download_events() -> int: return events_recovered +@celery.task( + name="backend.app.tasks.maintenance.date_posts_from_records", + soft_time_limit=1500, + time_limit=1800, +) +def date_posts_from_records() -> dict: + """Date undated native posts from the records their walk left on disk + (#4436). Hourly and self-limiting: once every post is dated it is one + empty query.""" + from ..services.importer import Importer + from ..services.post_record_repair import date_posts_from_records as _repair + from ..services.thumbnailer import Thumbnailer + + images_root = IMAGES_ROOT + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + importer = Importer( + session=session, + images_root=images_root, + import_root=images_root, + thumbnailer=Thumbnailer(images_root=images_root), + settings=ImportSettings.load_sync(session), + ) + return _repair(session, importer, images_root) + + @celery.task(name="backend.app.tasks.maintenance.recover_stalled_backup_runs") def recover_stalled_backup_runs() -> int: """Flip BackupRun rows stuck in running/restoring past the hard limit diff --git a/tests/test_celery_routing.py b/tests/test_celery_routing.py index de28fd9..f951594 100644 --- a/tests/test_celery_routing.py +++ b/tests/test_celery_routing.py @@ -28,6 +28,7 @@ def test_quick_maintenance_stays_on_maintenance(): ("backend.app.tasks.translation.translate_posts", "maintenance_long"), ("backend.app.tasks.gpu_queue.enqueue_gpu_backfill", "maintenance"), ("backend.app.tasks.maintenance.backfill_phash", "maintenance_long"), + ("backend.app.tasks.maintenance.date_posts_from_records", "maintenance_long"), ("backend.app.tasks.admin.normalize_tags_task", "maintenance_long"), ("backend.app.tasks.backup.backup_db_task", "maintenance_long"), ("backend.app.tasks.maintenance.vacuum_analyze", "maintenance"), diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index 8d158bc..b986e53 100644 --- a/tests/test_patreon_ingester.py +++ b/tests/test_patreon_ingester.py @@ -943,8 +943,10 @@ async def test_a_run_that_dies_before_import_leaves_its_media_unmarked( _FakeDownloader(tmp_path)) ing.run(source_id=source_id, campaign_id="c1", artist_slug="ingest", url="https://patreon.com/ingest", mode="tick") - # Phase 3 never ran, so `mark_seen_after_import` never did: only the post key. - assert _count_ledger(sync_engine, source_id) == 1 + # Phase 3 never ran, so `mark_seen_after_import` never did: neither the + # media nor the post record is marked (#4436 — the record, marked at write + # time, left the post undated for good once the run died before phase 3). + assert _count_ledger(sync_engine, source_id) == 0 # The next walk finds the file on disk with no record, and imports it. ing2 = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), @@ -954,6 +956,7 @@ async def test_a_run_that_dies_before_import_leaves_its_media_unmarked( assert result.written_paths == [str(tmp_path / "p1_1.jpg")] assert result.files_downloaded == 0 # not fetched again assert "on disk but never imported: p1_1.jpg" in result.stdout + assert len(result.post_record_paths) == 1 # the record is written again result.mark_seen_after_import() assert _count_ledger(sync_engine, source_id) == 2 diff --git a/tests/test_sidecar_import.py b/tests/test_sidecar_import.py index 6149624..a3aec25 100644 --- a/tests/test_sidecar_import.py +++ b/tests/test_sidecar_import.py @@ -415,3 +415,51 @@ def test_post_record_redates_images_linked_before_it(importer, import_layout): assert post.post_date != download_time assert rec.effective_date == post.post_date assert rec.earliest_post_date == post.post_date + + +def test_an_undated_post_is_dated_from_the_record_its_walk_left_on_disk( + importer, import_layout, +): + """#4436: a walk killed before phase 3 wrote the message's record to disk + but never upserted it, and marked it seen, so no later tick fixed it. The + repair finds the record under the artist's folder and dates the post — and + its images — under the post's own source.""" + from backend.app.services.post_record_repair import date_posts_from_records + + import_root, images_root = import_layout + artist = Artist(name="Alice", slug="alice") + importer.session.add(artist) + importer.session.flush() + importer.session.add(Source( + artist_id=artist.id, platform="discord", + url="https://discord.com/channels/1/100", + )) + importer.session.flush() + m = import_root / "Alice" / "20240301_123_01_art.jpg" + _split(m, "v") + _sidecar(m, {"category": "discord", "message_id": "123"}) + r = importer.import_one(m) + assert r.status == "imported" + post = importer.session.execute(select(Post)).scalar_one() + assert post.post_date is None and post.source_id is not None + + channel = images_root / "alice" / "discord" / "rewards" + channel.mkdir(parents=True) + (channel / "20240301_123_post.json").write_text(json.dumps({ + "category": "discord", "message_id": "123", "message": "", + "date": "2024-03-01T18:30:00.000000+00:00", + })) + (channel / "20240301_999_post.json").write_text("not json") # skipped, not fatal + + summary = date_posts_from_records(importer.session, importer, images_root) + + assert summary == {"undated": 1, "dated": 1, "no_record": 0} + importer.session.expire_all() + post = importer.session.execute(select(Post)).scalar_one() + rec = importer.session.get(ImageRecord, r.image_id) + assert post.post_date is not None + assert rec.earliest_post_date == post.post_date + # Nothing left undated: the next sweep is an empty query. + assert date_posts_from_records(importer.session, importer, images_root) == { + "undated": 0, "dated": 0, "no_record": 0, + }