diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index f2ca0f1..fed96ba 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: @@ -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 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