From 7e646ee478518a71a8c612d19f621ed60a9272a4 Mon Sep 17 00:00:00 2001 From: Koen Vossen Date: Wed, 12 Aug 2026 12:15:25 +0200 Subject: [PATCH] Mark job ABORTED on interrupt during metadata/find_datasets The KeyboardInterrupt/SystemExit handlers only wrapped the task-execution phase, so a Ctrl-C or Cloud Run SIGTERM during metadata or find_datasets propagated uncaught and left the up-front RUNNING summary as a zombie. execute() now wraps the whole _execute_locked run: any interrupt flips the still-RUNNING summary to ABORTED and persists it directly (not via yield, which a killed consumer may never drain). If an inner task-phase handler already marked the summary, its richer partial-results version is left untouched (state != RUNNING guard). Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg --- .../domain/models/ingestion/ingestion_job.py | 17 ++++ ingestify/tests/test_interrupt_aborts.py | 83 +++++++++++++++++++ 2 files changed, 100 insertions(+) create mode 100644 ingestify/tests/test_interrupt_aborts.py diff --git a/ingestify/domain/models/ingestion/ingestion_job.py b/ingestify/domain/models/ingestion/ingestion_job.py index 84a9eae..e88ef24 100644 --- a/ingestify/domain/models/ingestion/ingestion_job.py +++ b/ingestify/domain/models/ingestion/ingestion_job.py @@ -15,6 +15,7 @@ from ingestify.domain.models.dataset.revision import RevisionSource, SourceType from ingestify.domain.models.ingestion.ingestion_job_summary import ( IngestionJobSummary, + IngestionJobState, ) from ingestify.domain.models.ingestion.ingestion_plan import IngestionPlan from ingestify.domain.models.dataset.events import SelectorSkipped, DatasetSkipped @@ -295,6 +296,7 @@ def execute( # Single-run guard: one job identity must never run in two processes at once # (design: docs/design/single-run-lock.md). The lock is a session-scoped DB lock, # released the instant this process ends. If another run holds it, skip cleanly. + self._current_summary = None run_lock = store.acquire_run_lock(self._job_key()) if run_lock is None: logger.info("Skipping %s: another run holds the lock", self._job_key()) @@ -304,6 +306,18 @@ def execute( return try: yield from self._execute_locked(store, task_executor, summary_map) + except (KeyboardInterrupt, SystemExit): + # An interrupt (Ctrl-C / SIGTERM) can land in any phase — including + # metadata and find_datasets, which the inner task-phase handlers do + # not cover. Flip the up-front RUNNING summary to ABORTED and persist + # it directly (not via yield, which a killed consumer may never drain) + # so no zombie RUNNING row is left behind. If an inner handler already + # marked the summary, leave that (richer, partial-results) one alone. + summary = self._current_summary + if summary is not None and summary.state == IngestionJobState.RUNNING: + summary.set_aborted() + store.save_ingestion_job_summary(summary) + raise finally: run_lock.release() @@ -315,6 +329,9 @@ def _execute_locked( ) -> Iterator[IngestionJobSummary]: is_first_chunk = True ingestion_job_summary = IngestionJobSummary.new(ingestion_job=self) + # Exposed so execute()'s interrupt handler can mark it ABORTED whatever + # phase is running when a Ctrl-C / SIGTERM arrives. + self._current_summary = ingestion_job_summary # Persist the RUNNING row up front so the job is observable from the # moment it starts — and so a record survives even if the run never # reaches its final yield (e.g. an async source that keeps polling). diff --git a/ingestify/tests/test_interrupt_aborts.py b/ingestify/tests/test_interrupt_aborts.py new file mode 100644 index 0000000..d339262 --- /dev/null +++ b/ingestify/tests/test_interrupt_aborts.py @@ -0,0 +1,83 @@ +"""Interrupting a job during discovery must mark its summary ABORTED, not leave +it stuck at RUNNING. + +The KeyboardInterrupt/SystemExit handlers only wrapped the task-execution phase; +an interrupt during metadata or find_datasets (Ctrl-C, or a Cloud Run SIGTERM +while discovering) propagated uncaught and left a zombie RUNNING summary. The +summary is persisted RUNNING up front, so it must be flipped to ABORTED whatever +phase is interrupted. +""" +import pytest + +from ingestify import Source +from ingestify.domain import DataSpecVersionCollection, Selector +from ingestify.domain.models.fetch_policy import FetchPolicy +from ingestify.domain.models.ingestion.ingestion_job_summary import IngestionJobState +from ingestify.domain.models.ingestion.ingestion_plan import IngestionPlan + + +class SourceInterruptInFindDatasets(Source): + """Raises the given interrupt while discovering (before yielding anything), + mirroring a Ctrl-C / SIGTERM landing in find_datasets.""" + + provider = "test_provider" + + def __init__(self, name, exc): + super().__init__(name) + self._exc = exc + + def find_datasets( + self, dataset_type, data_spec_versions, dataset_collection_metadata, **kwargs + ): + raise self._exc + yield # pragma: no cover - makes this a generator + + +class SourceErrorInFindDatasets(Source): + """Fails with an ordinary Exception while discovering.""" + + provider = "test_provider" + + def find_datasets( + self, dataset_type, data_spec_versions, dataset_collection_metadata, **kwargs + ): + raise ValueError("boom") + yield # pragma: no cover - makes this a generator + + +def _setup(engine, source): + dsv = DataSpecVersionCollection.from_dict({"default": {"v1"}}) + engine.add_ingestion_plan( + IngestionPlan( + source=source, + fetch_policy=FetchPolicy(), + dataset_type="test", + selectors=[Selector.build({}, data_spec_versions=dsv)], + data_spec_versions=dsv, + ) + ) + + +@pytest.mark.parametrize("exc", [KeyboardInterrupt, SystemExit]) +def test_interrupt_during_find_datasets_marks_summary_aborted(engine, exc): + _setup(engine, SourceInterruptInFindDatasets("s", exc())) + + with pytest.raises(exc): + engine.run() + + summaries = engine.store.dataset_repository.load_ingestion_job_summaries() + assert len(summaries) == 1 + assert summaries[0].state == IngestionJobState.ABORTED + + +def test_ordinary_exception_in_find_datasets_is_failed_not_aborted(engine): + """Only KeyboardInterrupt/SystemExit map to ABORTED. An ordinary Exception is + a failure, so it must stay FAILED — the interrupt handler must not widen to + catch it.""" + _setup(engine, SourceErrorInFindDatasets("s")) + + engine.run() # ordinary exceptions are recorded, not re-raised + + summaries = engine.store.dataset_repository.load_ingestion_job_summaries() + assert len(summaries) == 1 + assert summaries[0].state == IngestionJobState.FAILED