Mark job FAILED on any uncaught crash, not just interrupts

.github/workflows/test.yml

@@ -14,7 +14,7 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest, macos-latest]
-        python-version: [3.9, "3.10", "3.11", "3.12"]
+        python-version: ["3.10", "3.11", "3.12"]
 
     steps:
     - uses: actions/checkout@v4
@@ -38,7 +38,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [3.9, "3.10", "3.11", "3.12"]
+        python-version: ["3.10", "3.11", "3.12"]
 
     services:
       mysql:
@@ -110,7 +110,10 @@ jobs:
         pip install -e ".[test]"
     - name: Install Postgres dependencies
       run: |
-        pip install psycopg2-binary
+        # SQLAlchemy 2.1 defaults the `postgresql://` URL to the psycopg (v3)
+        # dialect; older versions default to psycopg2. Install both drivers so the
+        # job works whichever SQLAlchemy resolves.
+        pip install "psycopg[binary]" psycopg2-binary
     - name: Test with pytest
       env:
         INGESTIFY_TEST_DATABASE_URL: postgresql://ingestify:ingestify@localhost:5432/ingestify_test

ingestify/__init__.py

@@ -11,4 +11,4 @@
     from .exceptions import StopProcessing, FatalError
     from .main import debug_source
 
-__version__ = "0.19.1"
+__version__ = "0.19.2"

ingestify/domain/models/ingestion/ingestion_job.py

@@ -318,6 +318,19 @@ def execute(
                 summary.set_aborted()
                 store.save_ingestion_job_summary(summary)
             raise
+        except Exception as e:
+            # Any other crash (an HTTP error from a source, a DB failure, a bug)
+            # can land in a phase the inner handlers do not cover and would
+            # otherwise propagate without ever updating the up-front RUNNING row,
+            # leaving a zombie RUNNING job that monitoring reads as "still running".
+            # Flip it to FAILED and persist directly (same reasoning as the
+            # interrupt handler); if an inner handler already marked the summary,
+            # leave that (richer) one alone.
+            summary = self._current_summary
+            if summary is not None and summary.state == IngestionJobState.RUNNING:
+                summary.set_exception(e)
+                store.save_ingestion_job_summary(summary)
+            raise
         finally:
             run_lock.release()
 

ingestify/tests/test_crash_marks_failed.py

@@ -0,0 +1,64 @@
+"""A crash in a phase the inner handlers do not cover must mark the summary
+FAILED, not leave it stuck at RUNNING.
+
+The async submit/collect loop only handled StopProcessing / interrupts /
+FatalError; an ordinary exception there (e.g. an HTTP 402 from the source API)
+propagated uncaught, so the up-front RUNNING summary was never updated and
+monitoring kept reading the job as "still running". execute() now has a
+top-level backstop that flips any still-RUNNING summary to FAILED on any such
+exception before re-raising.
+"""
+from typing import Iterator
+
+import pytest
+
+from ingestify import DatasetResource, Source
+from ingestify.domain.models.ingestion.ingestion_job_summary import IngestionJobState
+from ingestify.main import get_dev_engine
+from ingestify.utils import utcnow
+
+
+class SourceErrorInSubmit(Source):
+    """Async source (submit/collect) that raises an ordinary exception from
+    submit — mirrors a 402 Payment Required from the upstream API."""
+
+    provider = "fake_async"
+
+    def find_datasets(
+        self, dataset_type, data_spec_versions, dataset_collection_metadata, **kwargs
+    ):
+        yield DatasetResource(
+            dataset_resource_id={"keyword": "alpha"},
+            provider=self.provider,
+            dataset_type="keyword",
+            name="alpha",
+        )
+
+    def submit(self, dataset_resources: Iterator[DatasetResource]) -> bool:
+        raise ValueError("402 Payment Required")
+
+    def collect(self) -> Iterator[DatasetResource]:
+        yield from ()
+
+    def has_pending(self) -> bool:
+        return False
+
+
+def test_crash_in_submit_marks_summary_failed_not_running(tmp_path):
+    engine = get_dev_engine(
+        source=SourceErrorInSubmit("test"),
+        dataset_type="keyword",
+        data_spec_versions={"default": "v1"},
+        dev_dir=str(tmp_path),
+        configure_logging=False,
+    )
+
+    # The crash is a real failure, so it still propagates (the job exits non-zero).
+    with pytest.raises(ValueError, match="402 Payment Required"):
+        engine.run()
+
+    # ...but the up-front RUNNING summary must now be flipped to FAILED, so no
+    # zombie RUNNING row is left behind for monitoring to misread.
+    summaries = engine.store.dataset_repository.load_ingestion_job_summaries()
+    assert len(summaries) == 1
+    assert summaries[0].state == IngestionJobState.FAILED