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.
 

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,