Mark job ABORTED on interrupt during metadata/find_datasets

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).

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