diff --git a/ingestify/domain/models/ingestion/ingestion_job.py b/ingestify/domain/models/ingestion/ingestion_job.py index 772d382..c76ee7c 100644 --- a/ingestify/domain/models/ingestion/ingestion_job.py +++ b/ingestify/domain/models/ingestion/ingestion_job.py @@ -24,7 +24,7 @@ DatasetResource, ) from ingestify.domain.models.dataset.dataset import DatasetLastModifiedAtMap -from ingestify.domain.models.task.task_summary import TaskSummary +from ingestify.domain.models.task.task_summary import TaskSummary, Operation from ingestify.exceptions import SaveError, IngestifyError, StopProcessing from ingestify.utils import TaskExecutor, chunker @@ -580,6 +580,19 @@ def _store_async_result( existing_dataset = getattr(dataset_resource, "_existing_dataset", None) + # The source signalled (via mark_failed) that it could not fetch this + # resource: record a FAILED task and store nothing, so it is retried + # next run but stays visible in the summary. Mirror the operation that + # would have run — UPDATE for an existing dataset, CREATE otherwise. + if dataset_resource.fetch_error is not None: + logger.warning( + "Fetch failed for %s: %s", + dataset_identifier, + dataset_resource.fetch_error, + ) + operation = Operation.UPDATE if existing_dataset else Operation.CREATE + return TaskSummary.failed(str(uuid.uuid1()), dataset_identifier, operation) + # Load files that have file_loader or json_content files = {} for file_id, file_resource in dataset_resource.files.items(): diff --git a/ingestify/domain/models/resources/dataset_resource.py b/ingestify/domain/models/resources/dataset_resource.py index c99b502..4c095a6 100644 --- a/ingestify/domain/models/resources/dataset_resource.py +++ b/ingestify/domain/models/resources/dataset_resource.py @@ -56,10 +56,24 @@ class DatasetResource(BaseModel): fetch_policy_config: dict = Field(default_factory=dict) state: DatasetState = Field(default_factory=lambda: DatasetState.COMPLETE) files: dict[str, FileResource] = Field(default_factory=dict) + # Set via mark_failed() when a source (async submit/collect) could not fetch + # this resource; recorded as a FAILED task instead of being stored. + fetch_error: Optional[str] = None post_load_files: Optional[ Callable[["DatasetResource", Dict[str, DraftFile], Optional[Dataset]], None] ] = None + def mark_failed(self, error: Any) -> "DatasetResource": + """Mark this resource as failed to fetch (async submit/collect). + + Yield the marked resource from ``collect()`` and ingestify records it as + a FAILED task. No revision is written, so the resource is retried on the + next run — but the failure is now visible in the ingestion job summary + instead of being silently dropped. + """ + self.fetch_error = str(error) + return self + def run_post_load_files( self, files: Dict[str, DraftFile], existing_dataset: Optional[Dataset] = None ): diff --git a/ingestify/domain/models/task/task_summary.py b/ingestify/domain/models/task/task_summary.py index e2f54bf..924e074 100644 --- a/ingestify/domain/models/task/task_summary.py +++ b/ingestify/domain/models/task/task_summary.py @@ -82,6 +82,27 @@ def update(cls, task_id: str, dataset_identifier: Identifier): def create(cls, task_id: str, dataset_identifier: Identifier): return cls.new(task_id, Operation.CREATE, dataset_identifier) + @classmethod + def failed( + cls, task_id: str, dataset_identifier: Identifier, operation: Operation + ) -> "TaskSummary": + """A FAILED summary for a resource a source could not fetch (async + collect()); nothing was stored, so there is no revision to record. + + ``operation`` reflects what would have happened had the fetch + succeeded: UPDATE for a resource with an existing dataset, CREATE + otherwise.""" + now = utcnow() + task_summary = cls( + task_id=task_id, + started_at=now, + ended_at=now, + operation=operation, + dataset_identifier=dataset_identifier, + ) + task_summary.set_state(TaskState.FAILED) + return task_summary + def set_stats_from_revision(self, revision: Optional["Revision"]): if revision: self.persisted_file_count = len(revision.modified_files) diff --git a/ingestify/tests/test_failed_fetch.py b/ingestify/tests/test_failed_fetch.py new file mode 100644 index 0000000..90e8e28 --- /dev/null +++ b/ingestify/tests/test_failed_fetch.py @@ -0,0 +1,187 @@ +"""A source can signal, in the async submit/collect flow, that it could not +fetch a particular DatasetResource by calling ``dataset_resource.mark_failed()`` +and yielding it from ``collect()``. ingestify then records a FAILED task (no +dataset is stored, so it is retried next run) instead of the failure being +silently dropped. + +These tests assert on what is actually stored in the database. +""" +from typing import Iterator + +from ingestify import Source, DatasetResource +from ingestify.domain.models.dataset.revision import RevisionState +from ingestify.domain.models.ingestion.ingestion_job_summary import IngestionJobState +from ingestify.domain.models.task.task_summary import Operation, TaskState +from ingestify.main import get_dev_engine +from ingestify.utils import utcnow + + +class AsyncSourceWithFailure(Source): + """Async source whose collect() marks a (mutable) set of keywords failed.""" + + provider = "fake_async" + + def __init__(self, name, keywords, failing_keywords=(), capacity=2): + super().__init__(name) + self._keywords = keywords + self.failing_keywords = set(failing_keywords) + self._capacity = capacity + self._submitted = {} + self._in_flight = 0 + # Bumped per successful fetch so a re-fetch yields fresh content (a new + # revision), instead of identical bytes that collapse to NotModified. + self._version = 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()): + if keyword in self.failing_keywords: + resource.mark_failed("API returned PERMISSION_DENIED") + else: + self._version += 1 + resource.add_file( + last_modified=utcnow(), + data_feed_key="data", + data_spec_version="v1", + json_content={"keyword": keyword, "v": self._version}, + ) + del self._submitted[keyword] + self._in_flight -= 1 + yield resource + + def has_pending(self) -> bool: + return self._in_flight > 0 + + +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 _datasets(engine): + return list( + engine.store.get_dataset_collection( + provider="fake_async", dataset_type="keyword" + ) + ) + + +def test_mark_failed_sets_fetch_error(): + dr = DatasetResource( + dataset_resource_id={"keyword": "x"}, + provider="p", + dataset_type="keyword", + name="x", + ) + assert dr.fetch_error is None + assert dr.mark_failed("boom") is dr # chainable + assert dr.fetch_error == "boom" + + +def test_failed_fetch_recorded_as_failed_task(tmp_path): + engine = _dev_engine( + AsyncSourceWithFailure("s", ["a", "b", "c"], failing_keywords={"b"}), + tmp_path, + ) + + engine.run() + + summaries = engine.store.dataset_repository.load_ingestion_job_summaries() + assert len(summaries) == 1 + summary = summaries[0] + + assert summary.state == IngestionJobState.FINISHED + assert summary.successful_tasks == 2 + assert summary.failed_tasks == 1 + + # The one persisted child row is the failed fetch, for the right resource. + # It never existed before, so it is recorded as a failed CREATE. + assert len(summary.task_summaries) == 1 + failed = summary.task_summaries[0] + assert failed.state == TaskState.FAILED + assert failed.operation == Operation.CREATE + assert failed.dataset_identifier["keyword"] == "b" + + +def test_failed_fetch_stores_no_dataset(tmp_path): + engine = _dev_engine( + AsyncSourceWithFailure("s", ["a", "b", "c"], failing_keywords={"b"}), + tmp_path, + ) + + engine.run() + + # Only the two successful keywords produced a dataset; "b" did not. + stored_keywords = {ds.identifier["keyword"] for ds in _datasets(engine)} + assert stored_keywords == {"a", "c"} + + +def test_failed_refetch_leaves_existing_dataset_untouched_and_recovers(tmp_path): + """A failed *refetch* of an existing dataset must not corrupt it: nothing is + stored, the existing revision stays current, the failure is a FAILED UPDATE + task, and the resource still recovers on a later successful run. + + This locks in that mark_failed keeps the existing fetch/retry behaviour + working — a failed fetch is never represented as a bad/failed dataset. + """ + source = AsyncSourceWithFailure("s", ["b"]) + engine = _dev_engine(source, tmp_path) + + # Run 1: succeeds -> dataset "b" exists with a single revision. + engine.run() + dataset = _datasets(engine)[0] + assert len(dataset.revisions) == 1 + + # Force a refetch on the next run by invalidating the current revision. + engine.store.invalidate_revision(dataset, reason="quality check failed") + dataset = _datasets(engine)[0] + assert dataset.current_revision.state == RevisionState.VALIDATION_FAILED + + # Run 2: the refetch fails. The existing dataset must be left exactly as it + # was (still one revision, still VALIDATION_FAILED) and the failure recorded + # as a FAILED UPDATE task (an existing dataset -> UPDATE, not CREATE). + source.failing_keywords = {"b"} + engine.run() + + dataset = _datasets(engine)[0] + assert len(dataset.revisions) == 1, "failed refetch must not add a revision" + assert dataset.current_revision.state == RevisionState.VALIDATION_FAILED + + summaries = engine.store.dataset_repository.load_ingestion_job_summaries() + failed_summaries = [s for s in summaries if s.failed_tasks == 1] + assert len(failed_summaries) == 1 + failed = failed_summaries[0].task_summaries[0] + assert failed.state == TaskState.FAILED + assert failed.operation == Operation.UPDATE + + # Run 3: succeeds again -> the resource recovers with a fresh revision and is + # no longer VALIDATION_FAILED. Retry still works after a failure. + source.failing_keywords = set() + engine.run() + + dataset = _datasets(engine)[0] + assert len(dataset.revisions) == 2 + assert dataset.current_revision.state != RevisionState.VALIDATION_FAILED