feat(ingester): graceful mid-walk cancel on Stop — B4 cancel (plan #708)

Owning the walk lets Stop interrupt a live backfill chunk instead of letting it
run to its ~14.5-min time-box. ingest_core.run now polls _backfill_state at each
page boundary (a short SELECT, never held across the walk) and bails with PARTIAL
when an operator Stop has popped it. Latched on the first observed "running"
state so a run invoked without one (unit test / stale call) never self-cancels.
Progress is already checkpointed per-page, so a restart resumes from the cursor;
Stop clears it for a clean reset. No UI change — the existing Stop button now
just takes effect immediately.

Tests: _still_running reads the state; a latched run bails PARTIAL at the next
boundary when the state disappears (only the pre-cancel post ran).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/app/services/ingest_core.py

@@ -128,6 +128,8 @@ class Ingester:
         reached_bottom = False
         budget_hit = False
         early_out = False
+        stopped = False  # plan #708 B4: operator hit Stop mid-walk
+        cancel_armed = False  # latched once we observe a live "running" state
 
         def _result(
             *, success: bool, return_code: int,
@@ -179,6 +181,18 @@ class Ingester:
                     # the live posts count alongside it so the badge climbs DURING
                     # the chunk, not only when it ends.
                     if checkpoint:
+                        # plan #708 B4: an operator Stop pops `_backfill_state` —
+                        # bail at the page boundary (progress already checkpointed)
+                        # before more network work, so the live chunk halts
+                        # promptly instead of running to its time-box. LATCH on the
+                        # first observed "running" state, so a run invoked WITHOUT a
+                        # running state (a unit test, or a stale call) never
+                        # spuriously self-cancels. A short SELECT, never held.
+                        if self._still_running(source_id):
+                            cancel_armed = True
+                        elif cancel_armed:
+                            stopped = True
+                            break
                         self._checkpoint_cursor(source_id, emitted_cursor)
                         self._checkpoint_posts(source_id, posts_base + chunk_new_posts)
 
@@ -283,6 +297,16 @@ class Ingester:
             # maps it to a typed error.
             return self._failure_result(exc, _result)
 
+        # plan #708 B4: a Stop already popped the backfill state (incl. cursor +
+        # posts), so don't re-write them — return PARTIAL (reads as "ok/progress",
+        # the lifecycle no-ops since state is gone) instead of a false "complete".
+        if stopped:
+            return _result(
+                success=False, return_code=-1,
+                error_type=ErrorType.PARTIAL,
+                error_message=f"Stopped by operator: {downloaded} file(s) this chunk",
+            )
+
         # Final authoritative posts count for the badge — captures the last page
         # after the last boundary write and the time-box break (plan #704 #5).
         if checkpoint:
@@ -389,6 +413,23 @@ class Ingester:
             )
             session.commit()
 
+    def _still_running(self, source_id: int) -> bool:
+        """True while the source is armed for a deep walk (plan #708 B4).
+
+        An operator Stop (`source_service.stop_backfill`) pops `_backfill_state`,
+        so a False here means "cancel this chunk now". One short SELECT on its own
+        session — never held across the walk
+        ([[db-connection-held-across-subprocess]])."""
+        with self.session_factory() as session:
+            state = session.execute(
+                text(
+                    "SELECT config_overrides::jsonb ->> '_backfill_state' "
+                    "FROM source WHERE id = :sid"
+                ),
+                {"sid": source_id},
+            ).scalar_one_or_none()
+        return state == "running"
+
     def _checkpoint_posts(self, source_id: int, posts: int) -> None:
         """Persist the live backfill posts-processed count mid-walk (plan #704 #5).
 

tests/test_patreon_ingester.py

@@ -382,6 +382,50 @@ async def test_backfill_budget_cut_returns_partial_with_progress(
     assert result.cursor == "CUR2"
 
 
+@pytest.mark.asyncio
+async def test_still_running_reads_backfill_state(source_id, sync_engine, tmp_path, db):
+    """plan #708 B4: _still_running reflects the source's `_backfill_state` — the
+    flag an operator Stop pops to cancel a live chunk."""
+    ing = _ingester(sync_engine, tmp_path, _FakeClient([]), _FakeDownloader(tmp_path))
+    assert ing._still_running(source_id) is False  # absent → not running
+    src = (await db.execute(select(Source).where(Source.id == source_id))).scalar_one()
+    src.config_overrides = {"_backfill_state": "running"}
+    await db.commit()
+    assert ing._still_running(source_id) is True
+
+
+@pytest.mark.asyncio
+async def test_backfill_stops_when_operator_cancels_mid_walk(
+    source_id, sync_engine, tmp_path,
+):
+    """plan #708 B4: with the running state latched, a later page boundary that
+    finds it gone (Stop) bails the chunk with PARTIAL instead of finishing."""
+    pages = [
+        ("CUR2", [("p1", [_media("p1", 1)])]),
+        ("CUR3", [("p2", [_media("p2", 1)])]),
+        ("CUR4", [("p3", [_media("p3", 1)])]),
+    ]
+    downloader = _FakeDownloader(tmp_path)
+    ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), downloader)
+
+    # Latch on the page-2 boundary (running), then "Stop" before page 3.
+    calls = {"n": 0}
+
+    def fake_still_running(_source_id):
+        calls["n"] += 1
+        return calls["n"] == 1
+
+    ing._still_running = fake_still_running
+
+    result = ing.run(
+        source_id=source_id, campaign_id="c1", artist_slug="ingest",
+        url="https://patreon.com/ingest", mode="backfill",
+    )
+    assert result.error_type == ErrorType.PARTIAL
+    assert "Stopped by operator" in result.error_message
+    assert downloader.download_calls == 1  # only p1 ran; cancelled before p2
+
+
 # --- recovery -------------------------------------------------------------