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 c76ee7c..dadfa61 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, Operation -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__) @@ -238,6 +238,11 @@ def __repr__(self): MAX_TASKS_PER_CHUNK = 10_000 +# How many additional tasks must accumulate before a live progress snapshot of +# the IngestionJobSummary is persisted again. Keeps mid-run writes to roughly +# one per PROGRESS_SAVE_INTERVAL tasks instead of one per (possibly tiny) batch. +PROGRESS_SAVE_INTERVAL = 100 + class IngestionJob: def __init__( @@ -249,6 +254,31 @@ def __init__( self.ingestion_job_id = ingestion_job_id self.ingestion_plan = ingestion_plan self.selector = selector + # task_count() at which the summary was last persisted mid-run. + self._last_progress_saved_at = 0 + + def _save_progress( + self, + store: DatasetStore, + ingestion_job_summary: IngestionJobSummary, + *, + force: bool = False, + ): + """Persist a live snapshot of the (still RUNNING) summary. + + Only the parent row plus any FAILED task summaries are written (see + DatasetStore.save_ingestion_job_summary), so it is cheap enough to call + repeatedly. Throttled to roughly every PROGRESS_SAVE_INTERVAL tasks + unless ``force`` is set (used for the initial RUNNING row and when a new + chunk starts). Works for both the sync find_datasets flow and the async + submit/collect flow. + """ + count = ingestion_job_summary.task_count() + if not force and count - self._last_progress_saved_at < PROGRESS_SAVE_INTERVAL: + return + self._last_progress_saved_at = count + ingestion_job_summary.recount() + store.save_ingestion_job_summary(ingestion_job_summary) def execute( self, @@ -258,6 +288,11 @@ def execute( ) -> Iterator[IngestionJobSummary]: is_first_chunk = True ingestion_job_summary = IngestionJobSummary.new(ingestion_job=self) + # 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). + self._last_progress_saved_at = 0 + self._save_progress(store, ingestion_job_summary, force=True) # Process all items in batches. Yield a IngestionJobSummary per batch logger.info("Finding metadata") @@ -302,6 +337,13 @@ def execute( # We need to include the to_batches as that will start the generator batches = to_batches(dataset_resources) + 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 if "Field required" in str(e): @@ -343,6 +385,12 @@ def execute( batch = next(batches) except StopIteration: break + 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") @@ -453,6 +501,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: @@ -462,6 +515,10 @@ def execute( ) ingestion_job_summary.increase_skipped_tasks(skipped_tasks) + # Live snapshot after each batch (throttled) so the summary's + # counters/state stay current while the job runs. + self._save_progress(store, ingestion_job_summary) + if ingestion_job_summary.task_count() >= MAX_TASKS_PER_CHUNK: ingestion_job_summary.set_finished() yield ingestion_job_summary @@ -469,6 +526,8 @@ def execute( # Start a new one is_first_chunk = False ingestion_job_summary = IngestionJobSummary.new(ingestion_job=self) + self._last_progress_saved_at = 0 + self._save_progress(store, ingestion_job_summary, force=True) if ingestion_job_summary.task_count() > 0 or is_first_chunk: # When there is interesting information to store, or there was no data at all, store it @@ -496,6 +555,10 @@ def filtered_stream(): batch = next(batches) except StopIteration: return + 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") ingestion_job_summary.set_exception(e) @@ -556,12 +619,31 @@ 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]) + + # Live snapshot (throttled) so long-running submit/collect jobs + # are observable while polling, instead of only at the very end. + self._save_progress(store, ingestion_job_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/domain/models/ingestion/ingestion_job_summary.py b/ingestify/domain/models/ingestion/ingestion_job_summary.py index f36dcea..c3446be 100644 --- a/ingestify/domain/models/ingestion/ingestion_job_summary.py +++ b/ingestify/domain/models/ingestion/ingestion_job_summary.py @@ -70,7 +70,14 @@ def increase_skipped_tasks(self, skipped_tasks: int): def task_count(self): return len(self.task_summaries) + self.skipped_tasks - def _set_ended(self): + def recount(self): + """Recompute the task counters from the currently collected task + summaries, without ending the job or discarding any summaries. + + Safe to call repeatedly while the job is still RUNNING so a live + snapshot (written mid-run) reflects up-to-date counts. ``_set_ended`` + reuses this before it truncates ``task_summaries``. + """ self.failed_tasks = len( [task for task in self.task_summaries if task.state == TaskState.FAILED] ) @@ -90,6 +97,9 @@ def _set_ended(self): + self.ignored_successful_tasks + self.skipped_tasks ) + + def _set_ended(self): + self.recount() self.ended_at = utcnow() # Only keep failed tasks. Rest isn't interesting @@ -111,7 +121,9 @@ def set_skipped(self): @property def duration(self) -> timedelta: - return self.ended_at - self.started_at + # ended_at is None while the job is still RUNNING (live snapshots); + # measure against "now" so duration stays meaningful mid-run. + return (self.ended_at or utcnow()) - self.started_at def output_report(self): print( 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/infra/store/dataset/sqlalchemy/repository.py b/ingestify/infra/store/dataset/sqlalchemy/repository.py index e58ddb0..00775fb 100644 --- a/ingestify/infra/store/dataset/sqlalchemy/repository.py +++ b/ingestify/infra/store/dataset/sqlalchemy/repository.py @@ -38,7 +38,7 @@ DatasetCollectionMetadata, ) from ingestify.domain.models.ingestion.ingestion_job_summary import IngestionJobSummary -from ingestify.domain.models.task.task_summary import TaskSummary +from ingestify.domain.models.task.task_summary import TaskSummary, TaskState from ingestify.exceptions import IngestifyError from ingestify.utils import get_concurrency, key_from_dict @@ -740,17 +740,26 @@ def next_identity(self): # TODO: consider moving the IngestionJobSummary methods to a different Repository def save_ingestion_job_summary(self, ingestion_job_summary: IngestionJobSummary): + """Upsert the summary row (keyed on ingestion_job_summary_id). + + Only FAILED task summaries are persisted as child rows — successful, + ignored and skipped tasks are captured by the counters, not individual + rows. This keeps repeated live (mid-run) writes cheap (a successful run + writes no child rows at all) while still surfacing failures as soon as + they happen. The upsert is keyed on ingestion_job_summary_id, so live + snapshots update the same row in place. + """ ingestion_job_summary_entities = [ ingestion_job_summary.model_dump(exclude={"task_summaries"}) ] - task_summary_entities = [] - for task_summary in ingestion_job_summary.task_summaries: - task_summary_entities.append( - { - **task_summary.model_dump(), - "ingestion_job_summary_id": ingestion_job_summary.ingestion_job_summary_id, - } - ) + task_summary_entities = [ + { + **task_summary.model_dump(), + "ingestion_job_summary_id": ingestion_job_summary.ingestion_job_summary_id, + } + for task_summary in ingestion_job_summary.task_summaries + if task_summary.state == TaskState.FAILED + ] with self.session_provider.engine.connect() as connection: try: diff --git a/ingestify/tests/test_fatal_error.py b/ingestify/tests/test_fatal_error.py new file mode 100644 index 0000000..b95e615 --- /dev/null +++ b/ingestify/tests/test_fatal_error.py @@ -0,0 +1,181 @@ +"""Tests for FatalError exception.""" +from typing import Iterator +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_job_summary import IngestionJobState +from ingestify.domain.models.ingestion.ingestion_plan import IngestionPlan +from ingestify.exceptions import FatalError, StopProcessing +from ingestify.main import get_dev_engine +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() + + +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 diff --git a/ingestify/tests/test_live_job_summary.py b/ingestify/tests/test_live_job_summary.py new file mode 100644 index 0000000..699cef6 --- /dev/null +++ b/ingestify/tests/test_live_job_summary.py @@ -0,0 +1,180 @@ +"""The IngestionJobSummary is persisted *live*: a RUNNING row is written to the +database as soon as the job starts and updated while it runs — for both the +sync find_datasets flow and the async submit/collect flow. Previously the +summary was written to the database only once, at the very end (so a long +submit/collect run left no trace until it finished, and none at all if it never +finished). + +These tests assert on what is actually stored in the database +(load_ingestion_job_summaries), not just on the in-memory save calls. +""" +from typing import Iterator + +from ingestify import Source, DatasetResource +from ingestify.domain.models.ingestion import ingestion_job as ingestion_job_module +from ingestify.domain.models.ingestion.ingestion_job_summary import IngestionJobState +from ingestify.main import get_dev_engine +from ingestify.utils import utcnow + + +class SyncSource(Source): + """Plain source: find_datasets only (the batch / find_datasets flow).""" + + provider = "fake_sync" + + def __init__(self, name, keywords): + super().__init__(name) + self._keywords = keywords + + 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, + ).add_file( + last_modified=utcnow(), + data_feed_key="data", + data_spec_version="v1", + json_content={"keyword": keyword}, + ) + + +class AsyncSource(Source): + """Source using the submit/collect (async) pattern.""" + + provider = "fake_async" + + def __init__(self, name, keywords, capacity=2): + super().__init__(name) + self._keywords = keywords + self._capacity = capacity + self._submitted = {} + 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 resource in dataset_resources: + self._submitted[resource.dataset_resource_id["keyword"]] = resource + self._in_flight += 1 + if self._in_flight >= self._capacity: + return False + return True + + def collect(self) -> Iterator[DatasetResource]: + for keyword, resource in list(self._submitted.items()): + resource.add_file( + last_modified=utcnow(), + data_feed_key="data", + data_spec_version="v1", + json_content={"keyword": keyword}, + ) + del self._submitted[keyword] + self._in_flight -= 1 + yield resource + + def has_pending(self) -> bool: + return self._in_flight > 0 + + +def _db_summaries(engine): + """Read the ingestion job summaries back from the database.""" + return engine.store.dataset_repository.load_ingestion_job_summaries() + + +def _capture_db_after_each_save(engine): + """After every summary write, read the summaries back FROM THE DATABASE and + snapshot their (state, total_tasks). Lets us assert on what is actually + persisted mid-run — not merely on the in-memory save calls. + """ + snapshots = [] + original = engine.store.save_ingestion_job_summary + + def wrapper(summary): + result = original(summary) + snapshots.append( + [ + (s.state, s.total_tasks, s.successful_tasks) + for s in _db_summaries(engine) + ] + ) + return result + + engine.store.save_ingestion_job_summary = wrapper + return snapshots + + +def _dev_engine(source, tmp_path): + return get_dev_engine( + source=source, + dataset_type="keyword", + data_spec_versions={"default": "v1"}, + dev_dir=str(tmp_path), + configure_logging=False, + ) + + +def test_sync_flow_persists_running_row_before_finished(tmp_path, monkeypatch): + monkeypatch.setattr(ingestion_job_module, "PROGRESS_SAVE_INTERVAL", 1) + engine = _dev_engine(SyncSource("test", ["a", "b", "c"]), tmp_path) + db_snapshots = _capture_db_after_each_save(engine) + + engine.run() + + # The very first thing written to the DB is a RUNNING row. + assert db_snapshots, "nothing was ever written to the database" + first = db_snapshots[0] + assert len(first) == 1 + assert first[0][0] == IngestionJobState.RUNNING + + # End state, read straight from the database. + summaries = _db_summaries(engine) + assert len(summaries) == 1 + summary = summaries[0] + assert summary.state == IngestionJobState.FINISHED + assert summary.successful_tasks == 3 + # Only failed task summaries are persisted; this run had none. + assert summary.task_summaries == [] + + +def test_async_flow_updates_db_summary_during_collect(tmp_path, monkeypatch): + monkeypatch.setattr(ingestion_job_module, "PROGRESS_SAVE_INTERVAL", 1) + engine = _dev_engine( + AsyncSource("test", ["a", "b", "c", "d", "e"], capacity=2), tmp_path + ) + db_snapshots = _capture_db_after_each_save(engine) + + engine.run() + + # RUNNING row is in the database from the start, before any task completed. + assert db_snapshots[0] == [(IngestionJobState.RUNNING, 0, 0)] + + # While polling, the database already shows a RUNNING summary with tasks + # counted — i.e. progress is visible before the job finishes. + seen_live_progress = any( + state == IngestionJobState.RUNNING and total > 0 + for snapshot in db_snapshots[:-1] + for (state, total, _successful) in snapshot + ) + assert seen_live_progress, "database never showed live progress during collect" + + # Final state in the database. + summaries = _db_summaries(engine) + assert len(summaries) == 1 + summary = summaries[0] + assert summary.state == IngestionJobState.FINISHED + assert summary.successful_tasks == 5 + assert summary.task_summaries == []