Add DatasetResource.mark_failed for async fetch failures

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():

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
     ):

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)

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