diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py index be418b0..7f7dcab 100644 --- a/backend/app/services/ingest_core.py +++ b/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). diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index 3ca19a5..57a491b 100644 --- a/tests/test_patreon_ingester.py +++ b/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 -------------------------------------------------------------