From 4bb11ce7dc20bfe873caa56748b37824d8864bc4 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 5 Jun 2026 23:58:16 -0400 Subject: [PATCH] =?UTF-8?q?feat(patreon):=20incremental=20cursor=20checkpo?= =?UTF-8?q?int=20mid-walk=20=E2=80=94=20#705=20step=201=20(#6)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A worker SIGKILL (hard-time-limit backstop) mid-chunk lost the whole chunk's walk — the cursor was only persisted at chunk boundaries by phase 3, so the next tick re-walked from the chunk start. Now the ingester checkpoints _backfill_cursor to the DB at each page boundary (backfill/ recovery only) via an ATOMIC single-key UPDATE (config_overrides::jsonb → jsonb_set('{_backfill_cursor}') → ::json), so it never clobbers operator config or other backfill keys. On a crash the last mid-walk cursor survives → the next chunk resumes near the crash, not the chunk start. phase 3 still writes the final cursor (same value); this is the safety net. Tests: a backfill walk leaves the last page's cursor in the DB (written by the ingester, before any phase 3); a tick never checkpoints. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/app/services/patreon_ingester.py | 33 ++++++++++++++++++- tests/test_patreon_ingester.py | 41 ++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 1 deletion(-) diff --git a/backend/app/services/patreon_ingester.py b/backend/app/services/patreon_ingester.py index 07186b7..5207ff6 100644 --- a/backend/app/services/patreon_ingester.py +++ b/backend/app/services/patreon_ingester.py @@ -44,7 +44,7 @@ import time from collections.abc import Callable from pathlib import Path -from sqlalchemy import select +from sqlalchemy import select, text from sqlalchemy.dialects.postgresql import insert as pg_insert from ..models import PatreonSeenMedia @@ -153,6 +153,9 @@ class PatreonIngester: A client-level failure (drift / auth / network) fails the whole run loud. """ bypass_seen = mode == "recovery" + # Only deep walks checkpoint their cursor mid-flight (plan #705 #6); a + # tick has no resumable backfill state. + checkpoint = mode in ("backfill", "recovery") start = time.monotonic() log_lines: list[str] = [] written: list[str] = [] @@ -213,6 +216,12 @@ class PatreonIngester: # `Cursor:` stdout line to regex back out. if page_cursor and page_cursor != emitted_cursor: emitted_cursor = page_cursor + # plan #705 #6: persist the cursor at each page boundary so a + # worker SIGKILL mid-chunk resumes near the crash, not the + # chunk start. (phase 3 still writes the final cursor — same + # value; this is the crash-safety net.) + if checkpoint: + self._checkpoint_cursor(source_id, emitted_cursor) # Time-box check at the post boundary (coarse, like a gallery-dl # chunk). Backfill/recovery resume from emitted_cursor next chunk. @@ -394,6 +403,28 @@ class PatreonIngester: ).scalars().all() return set(rows) + def _checkpoint_cursor(self, source_id: int, cursor: str) -> None: + """Persist the in-progress backfill cursor mid-walk (plan #705 #6). + + ATOMIC, single-key UPDATE: cast the JSON column to jsonb, set just + `_backfill_cursor`, cast back — so it never clobbers operator config or + the other backfill keys (no read-modify-write race). The in-flight guard + means only this source's one download runs at a time; a concurrent + operator stop is benign (a stray cursor with no `_backfill_state` is + ignored by tick mode and cleared on the next start). + """ + with self.session_factory() as session: + session.execute( + text( + "UPDATE source SET config_overrides = jsonb_set(" + " coalesce(config_overrides::jsonb, '{}'::jsonb)," + " '{_backfill_cursor}', to_jsonb(cast(:cur AS text))" + ")::json WHERE id = :sid" + ), + {"cur": cursor, "sid": source_id}, + ) + session.commit() + def _mark_seen(self, source_id: int, items: list[tuple[str, str]]) -> None: """Idempotent upsert of (filehash, post_id) ledger rows for a page. diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index b557740..e5a0015 100644 --- a/tests/test_patreon_ingester.py +++ b/tests/test_patreon_ingester.py @@ -245,6 +245,47 @@ async def test_backfill_reaches_bottom_signals_complete(source_id, sync_engine, assert result.run_stats["downloaded_count"] == 2 +@pytest.mark.asyncio +async def test_backfill_checkpoints_cursor_mid_walk(source_id, sync_engine, tmp_path, db): + """plan #705 #6: the cursor is persisted to the DB at each page boundary + DURING the walk (not just at the end via phase 3), so a worker SIGKILL + mid-chunk resumes near the crash. After the walk the source carries the last + page's cursor — written by the ingester, not phase 3 (which the unit run + never reaches).""" + pages = [ + (None, [("p1", [_media("p1", 1)])]), + ("CUR2", [("p2", [_media("p2", 1)])]), + ("CUR3", [("p3", [_media("p3", 1)])]), + ] + ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), _FakeDownloader(tmp_path)) + ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="backfill", + ) + co = (await db.execute( + select(Source.config_overrides).where(Source.id == source_id) + )).scalar_one() + assert co.get("_backfill_cursor") == "CUR3" + + +@pytest.mark.asyncio +async def test_tick_does_not_checkpoint_cursor(source_id, sync_engine, tmp_path, db): + """A tick has no resumable backfill state, so it must not write a cursor.""" + ing = _ingester( + sync_engine, tmp_path, + _FakeClient([("CUR2", [("p1", [_media("p1", 1)])])]), + _FakeDownloader(tmp_path), + ) + ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="tick", + ) + co = (await db.execute( + select(Source.config_overrides).where(Source.id == source_id) + )).scalar_one() + assert "_backfill_cursor" not in (co or {}) + + @pytest.mark.asyncio async def test_backfill_budget_cut_returns_partial_with_progress( source_id, sync_engine, tmp_path, monkeypatch,