From c279b538ebd35e6981c120c18ecffd186ac0cce5 Mon Sep 17 00:00:00 2001 From: Koen Vossen Date: Wed, 28 Oct 2026 21:47:06 +0100 Subject: [PATCH 1/3] Mark job FAILED on any uncaught crash, not just interrupts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The summary is persisted RUNNING up front so a job is observable from the moment it starts. But execute() only caught KeyboardInterrupt/SystemExit (-> ABORTED); the async submit/collect loop only caught StopProcessing / interrupts / FatalError. An ordinary exception in an uncovered phase — e.g. an HTTP 402 from a source API during submit — propagated uncaught, so the up-front RUNNING summary was never updated and monitoring kept reading the job as "still running" forever. Add a top-level backstop in execute(): on any other exception, flip a still-RUNNING summary to FAILED (set_exception) and persist it before re-raising. It only acts when the summary is still RUNNING, so richer inner handlers that already recorded partial results are left untouched. Catch Exception (not BaseException) so GeneratorExit/interrupts are unaffected. Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg --- ingestify/__init__.py | 2 +- .../domain/models/ingestion/ingestion_job.py | 13 ++++ ingestify/tests/test_crash_marks_failed.py | 64 +++++++++++++++++++ 3 files changed, 78 insertions(+), 1 deletion(-) create mode 100644 ingestify/tests/test_crash_marks_failed.py diff --git a/ingestify/__init__.py b/ingestify/__init__.py index 2fc7376..095941f 100644 --- a/ingestify/__init__.py +++ b/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" diff --git a/ingestify/domain/models/ingestion/ingestion_job.py b/ingestify/domain/models/ingestion/ingestion_job.py index e88ef24..7ffca04 100644 --- a/ingestify/domain/models/ingestion/ingestion_job.py +++ b/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() diff --git a/ingestify/tests/test_crash_marks_failed.py b/ingestify/tests/test_crash_marks_failed.py new file mode 100644 index 0000000..a3059b8 --- /dev/null +++ b/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 From 0b36f0844ea8380d72a99fa63a53a379b1c068cd Mon Sep 17 00:00:00 2001 From: Koen Vossen Date: Wed, 28 Oct 2026 21:56:27 +0100 Subject: [PATCH 2/3] Drop Python 3.9 from the CI test matrix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Python 3.9 is end-of-life and setup-python no longer provides it on the GitHub-hosted runners, so those matrix legs fail to start. Test on 3.10–3.12. Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg --- .github/workflows/test.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index f2ca0f1..910da1b 100644 --- a/.github/workflows/test.yml +++ b/.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: From 22d6fd54da33620f5544de2f7831e8cea7f79582 Mon Sep 17 00:00:00 2001 From: Koen Vossen Date: Wed, 28 Oct 2026 22:15:33 +0100 Subject: [PATCH 3/3] Install psycopg v3 in the Postgres CI job SQLAlchemy 2.1 makes psycopg (v3) the default dialect for a plain `postgresql://` URL (psycopg2 was the default before), so the job failed with "No module named 'psycopg'" while only psycopg2-binary was installed. Install both drivers so the job passes whichever SQLAlchemy version resolves. Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg --- .github/workflows/test.yml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 910da1b..fed96ba 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -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