From 4e9f4bd3e0c44d304513d40c882497c44c6fbe24 Mon Sep 17 00:00:00 2001 From: Koen Vossen Date: Wed, 8 Jul 2026 11:16:48 +0200 Subject: [PATCH 1/2] Add FatalError: fail the run loudly (exit 1) on unrecoverable source errors StopProcessing stops a run gracefully (exit 2). FatalError is its counterpart for unrecoverable, non-retryable problems (e.g. a deactivated account): the source/loader raises it, ingestify propagates it out of find_datasets/collect instead of swallowing it as a skipped find_datasets, and the CLI exits non-zero so schedulers surface the failure. Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg --- ingestify/__init__.py | 2 +- ingestify/cmdline.py | 5 +- .../domain/models/ingestion/ingestion_job.py | 15 +++- ingestify/exceptions.py | 17 ++++ ingestify/tests/test_fatal_error.py | 87 +++++++++++++++++++ 5 files changed, 123 insertions(+), 3 deletions(-) create mode 100644 ingestify/tests/test_fatal_error.py diff --git a/ingestify/__init__.py b/ingestify/__init__.py index 8798985..94e59ab 100644 --- a/ingestify/__init__.py +++ b/ingestify/__init__.py @@ -8,7 +8,7 @@ from .infra import retrieve_http from .source_base import Source, DatasetResource from .domain.models.fetch_policy import FetchPolicy - from .exceptions import StopProcessing + from .exceptions import StopProcessing, FatalError from .main import debug_source __version__ = "0.17.1" diff --git a/ingestify/cmdline.py b/ingestify/cmdline.py index b432eda..5b4dffb 100644 --- a/ingestify/cmdline.py +++ b/ingestify/cmdline.py @@ -7,7 +7,7 @@ import click from dotenv import find_dotenv, load_dotenv -from ingestify.exceptions import ConfigurationError, StopProcessing +from ingestify.exceptions import ConfigurationError, StopProcessing, FatalError from ingestify.main import get_engine from ingestify import __version__ @@ -121,6 +121,9 @@ def run( except StopProcessing as e: logger.warning(f"Stopped early: {e}") sys.exit(e.exit_code) + except FatalError as e: + logger.error(f"Fatal error: {e}") + sys.exit(e.exit_code) logger.info("Done") diff --git a/ingestify/domain/models/ingestion/ingestion_job.py b/ingestify/domain/models/ingestion/ingestion_job.py index 772d382..ea64ed9 100644 --- a/ingestify/domain/models/ingestion/ingestion_job.py +++ b/ingestify/domain/models/ingestion/ingestion_job.py @@ -25,7 +25,7 @@ ) from ingestify.domain.models.dataset.dataset import DatasetLastModifiedAtMap from ingestify.domain.models.task.task_summary import TaskSummary -from ingestify.exceptions import SaveError, IngestifyError, StopProcessing +from ingestify.exceptions import SaveError, IngestifyError, StopProcessing, FatalError from ingestify.utils import TaskExecutor, chunker logger = logging.getLogger(__name__) @@ -302,6 +302,11 @@ def execute( # We need to include the to_batches as that will start the generator batches = to_batches(dataset_resources) + except (StopProcessing, FatalError): + # Propagate: the run must stop gracefully (StopProcessing) or fail + # loudly (FatalError). Neither should be swallowed as a skipped + # find_datasets. + raise except ValidationError as e: # Make sure to pass this to the highest level as this means the Source is wrong if "Field required" in str(e): @@ -343,6 +348,10 @@ def execute( batch = next(batches) except StopIteration: break + except (StopProcessing, FatalError): + # Propagate stop/fail signals instead of swallowing them as a + # failed batch fetch. + raise except Exception as e: logger.exception("Failed to fetch next batch") @@ -496,6 +505,10 @@ def filtered_stream(): batch = next(batches) except StopIteration: return + except (StopProcessing, FatalError): + # Propagate stop/fail signals instead of swallowing them as + # a failed batch fetch. + raise except Exception as e: logger.exception("Failed to fetch next batch") ingestion_job_summary.set_exception(e) diff --git a/ingestify/exceptions.py b/ingestify/exceptions.py index cd311a9..011e989 100644 --- a/ingestify/exceptions.py +++ b/ingestify/exceptions.py @@ -26,3 +26,20 @@ class StopProcessing(IngestifyError): """ exit_code = 2 + + +class FatalError(IngestifyError): + """Raised by a source or loader to signal that the run cannot continue and + must fail loudly. Unlike StopProcessing, this is NOT recoverable by + retrying later — e.g. a deactivated account or an otherwise misconfigured + source. + + Already-processed datasets are preserved, but the run aborts with a + non-zero exit so schedulers surface it as a failure instead of a silent + success. It propagates out of find_datasets/collect rather than being + swallowed as a skipped find_datasets. + + Exit code: 1. + """ + + exit_code = 1 diff --git a/ingestify/tests/test_fatal_error.py b/ingestify/tests/test_fatal_error.py new file mode 100644 index 0000000..d0ee92b --- /dev/null +++ b/ingestify/tests/test_fatal_error.py @@ -0,0 +1,87 @@ +"""Tests for FatalError exception.""" +from unittest.mock import patch + +import pytest + +from ingestify import Source, DatasetResource +from ingestify.domain import DataSpecVersionCollection, DraftFile, Selector +from ingestify.domain.models.fetch_policy import FetchPolicy +from ingestify.domain.models.ingestion.ingestion_plan import IngestionPlan +from ingestify.exceptions import FatalError +from ingestify.utils import utcnow + + +def good_loader(file_resource, current_file, **kwargs): + return DraftFile.from_input("data", data_feed_key="f1") + + +class SourceFatalInFindDatasets(Source): + """Source whose find_datasets fails fatally before yielding anything + (mirrors an account-level authorization error surfacing in discovery).""" + + provider = "test_provider" + + def find_datasets( + self, dataset_type, data_spec_versions, dataset_collection_metadata, **kwargs + ): + raise FatalError("account deactivated") + yield # pragma: no cover - makes this a generator + + +class SourceFatalMidStream(Source): + """Source that yields 2 good datasets, then fails fatally.""" + + provider = "test_provider" + + def find_datasets( + self, dataset_type, data_spec_versions, dataset_collection_metadata, **kwargs + ): + for i in range(2): + r = DatasetResource( + dataset_resource_id={"item_id": i}, + provider=self.provider, + dataset_type="test", + name=f"item-{i}", + ) + r.add_file( + last_modified=utcnow(), + data_feed_key="f1", + data_spec_version="v1", + file_loader=good_loader, + ) + yield r + raise FatalError("account deactivated") + + +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, + ) + ) + + +def test_fatal_error_has_exit_code(): + assert FatalError.exit_code == 1 + + +def test_fatal_error_in_find_datasets_propagates(engine): + """FatalError raised in find_datasets propagates out of engine.run() + instead of being swallowed as a skipped find_datasets.""" + _setup(engine, SourceFatalInFindDatasets("s")) + + with pytest.raises(FatalError, match="deactivated"): + engine.run() + + +def test_fatal_error_mid_stream_propagates(engine): + """FatalError raised after some datasets still propagates.""" + _setup(engine, SourceFatalMidStream("s")) + + with pytest.raises(FatalError, match="deactivated"): + engine.run() From d81cc68ac9b699e371863077ab7450dc1a6f73e8 Mon Sep 17 00:00:00 2001 From: Koen Vossen Date: Wed, 8 Jul 2026 12:17:28 +0200 Subject: [PATCH 2/2] Persist the job summary on FatalError/StopProcessing; keep StopProcessing controlled MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two fixes on top of the FatalError work: 1. FatalError now persists the summary (as FAILED) before aborting. The loader has no finally to save an in-progress summary, so a bare re-raise lost the record entirely. It now set_exception + yields (so the loader stores it) + re-raises, at each point FatalError can surface. 2. The async submit/collect path had NO stop/fail handling at all: a StopProcessing (e.g. quota) or FatalError raised while collecting propagated without ever persisting the summary. Added a handler mirroring the sync path so StopProcessing saves a FINISHED summary (controlled stop) and FatalError a FAILED one. StopProcessing keeps its own controlled-stop semantics and is no longer folded into FatalError's re-raise clauses — only FatalError forces the hard stop. Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg --- .../domain/models/ingestion/ingestion_job.py | 54 ++++++++--- ingestify/tests/test_fatal_error.py | 96 ++++++++++++++++++- 2 files changed, 134 insertions(+), 16 deletions(-) diff --git a/ingestify/domain/models/ingestion/ingestion_job.py b/ingestify/domain/models/ingestion/ingestion_job.py index 372aa25..b51dae3 100644 --- a/ingestify/domain/models/ingestion/ingestion_job.py +++ b/ingestify/domain/models/ingestion/ingestion_job.py @@ -302,10 +302,12 @@ def execute( # We need to include the to_batches as that will start the generator batches = to_batches(dataset_resources) - except (StopProcessing, FatalError): - # Propagate: the run must stop gracefully (StopProcessing) or fail - # loudly (FatalError). Neither should be swallowed as a skipped - # find_datasets. + except FatalError as e: + # Persist the failure so it isn't lost, then abort — not swallowed as + # a skipped find_datasets. (StopProcessing keeps its own controlled + # stop and is not intercepted here.) + ingestion_job_summary.set_exception(e) + yield ingestion_job_summary raise except ValidationError as e: # Make sure to pass this to the highest level as this means the Source is wrong @@ -348,9 +350,11 @@ def execute( batch = next(batches) except StopIteration: break - except (StopProcessing, FatalError): - # Propagate stop/fail signals instead of swallowing them as a + except FatalError as e: + # Persist the failure, then abort, instead of swallowing it as a # failed batch fetch. + ingestion_job_summary.set_exception(e) + yield ingestion_job_summary raise except Exception as e: logger.exception("Failed to fetch next batch") @@ -462,6 +466,11 @@ def execute( ingestion_job_summary.set_finished() yield ingestion_job_summary raise + except FatalError as e: + logger.error("Fatal error — saving summary and aborting") + ingestion_job_summary.set_exception(e) + yield ingestion_job_summary + raise ingestion_job_summary.add_task_summaries(results) else: @@ -505,9 +514,9 @@ def filtered_stream(): batch = next(batches) except StopIteration: return - except (StopProcessing, FatalError): - # Propagate stop/fail signals instead of swallowing them as - # a failed batch fetch. + except FatalError: + # Propagate to the submit/collect handler (which persists the + # summary); don't swallow it as a failed batch fetch. raise except Exception as e: logger.exception("Failed to fetch next batch") @@ -569,12 +578,27 @@ def filtered_stream(): resources = filtered_stream() done = False - while not done or source.has_pending(): - done = source.submit(resources) - - for dataset_resource in source.collect(): - task_summary = self._store_async_result(dataset_resource, store) - ingestion_job_summary.add_task_summaries([task_summary]) + # Unlike the sync path, submit/collect had no stop/fail handling: a + # StopProcessing (e.g. quota) or FatalError raised while collecting would + # propagate without ever persisting the summary, losing the record. Mirror + # the sync handler here so both are saved before aborting. + try: + while not done or source.has_pending(): + done = source.submit(resources) + + for dataset_resource in source.collect(): + task_summary = self._store_async_result(dataset_resource, store) + ingestion_job_summary.add_task_summaries([task_summary]) + except StopProcessing: + logger.info("StopProcessing raised — saving partial results and stopping") + ingestion_job_summary.set_finished() + yield ingestion_job_summary + raise + except FatalError as e: + logger.error("Fatal error — saving summary and aborting") + ingestion_job_summary.set_exception(e) + yield ingestion_job_summary + raise if ingestion_job_summary.task_count() > 0 or is_first_chunk: ingestion_job_summary.set_finished() diff --git a/ingestify/tests/test_fatal_error.py b/ingestify/tests/test_fatal_error.py index d0ee92b..b95e615 100644 --- a/ingestify/tests/test_fatal_error.py +++ b/ingestify/tests/test_fatal_error.py @@ -1,4 +1,5 @@ """Tests for FatalError exception.""" +from typing import Iterator from unittest.mock import patch import pytest @@ -6,8 +7,10 @@ from ingestify import Source, DatasetResource from ingestify.domain import DataSpecVersionCollection, DraftFile, 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 -from ingestify.exceptions import FatalError +from ingestify.exceptions import FatalError, StopProcessing +from ingestify.main import get_dev_engine from ingestify.utils import utcnow @@ -85,3 +88,94 @@ def test_fatal_error_mid_stream_propagates(engine): with pytest.raises(FatalError, match="deactivated"): engine.run() + + +def test_fatal_error_persists_failed_summary(engine): + """FatalError must persist the summary (as FAILED) before aborting, so the + failure isn't lost — the loader has no finally to save it otherwise.""" + _setup(engine, SourceFatalInFindDatasets("s")) + + with patch.object(engine.store, "save_ingestion_job_summary") as mock_save: + with pytest.raises(FatalError): + engine.run() + + assert mock_save.call_count >= 1, "summary was never saved on FatalError" + saved = mock_save.call_args[0][0] + assert saved.state == IngestionJobState.FAILED + + +# --- async submit/collect path ------------------------------------------------- + + +class _AsyncSource(Source): + """Async source that raises `error` from collect() after submitting.""" + + provider = "fake_async" + + def __init__(self, name, keywords, error): + super().__init__(name) + self._keywords = keywords + self._error = error + self._in_flight = 0 + + def find_datasets( + self, dataset_type, data_spec_versions, dataset_collection_metadata, **kwargs + ): + for keyword in self._keywords: + yield DatasetResource( + dataset_resource_id={"keyword": keyword}, + provider=self.provider, + dataset_type="keyword", + name=keyword, + ) + + def submit(self, dataset_resources: Iterator[DatasetResource]) -> bool: + for _ in dataset_resources: + self._in_flight += 1 + return True + + def collect(self) -> Iterator[DatasetResource]: + raise self._error + yield # pragma: no cover - makes this a generator + + def has_pending(self) -> bool: + return self._in_flight > 0 + + +def _run_async(source, tmp_path): + engine = get_dev_engine( + source=source, + dataset_type="keyword", + data_spec_versions={"default": "v1"}, + dev_dir=str(tmp_path), + configure_logging=False, + ) + return engine + + +def test_fatal_error_in_async_collect_persists_failed_summary(tmp_path): + """A FatalError raised while collecting (the submit/collect path bigintel + uses) still persists a FAILED summary before aborting.""" + engine = _run_async(_AsyncSource("s", ["a", "b"], FatalError("boom")), tmp_path) + + with pytest.raises(FatalError): + engine.run() + + summaries = engine.store.dataset_repository.load_ingestion_job_summaries() + assert len(summaries) == 1 + assert summaries[0].state == IngestionJobState.FAILED + + +def test_stop_processing_in_async_collect_persists_summary(tmp_path): + """StopProcessing while collecting is a controlled stop: the summary is + persisted (FINISHED) instead of being lost, mirroring the sync path.""" + engine = _run_async( + _AsyncSource("s", ["a", "b"], StopProcessing("quota")), tmp_path + ) + + with pytest.raises(StopProcessing): + engine.run() + + summaries = engine.store.dataset_repository.load_ingestion_job_summaries() + assert len(summaries) == 1 + assert summaries[0].state == IngestionJobState.FINISHED