From 0679b1ccb6ba3d563048500b2e0b4b20e9f9cdba Mon Sep 17 00:00:00 2001
From: Koen Vossen <info@koenvossen.nl>
Date: Fri, 3 Jul 2026 13:39:40 +0200
Subject: [PATCH 1/4] Persist IngestionJobSummary live (RUNNING) instead of
 only at the end
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Previously the summary row was written exactly once, when the ingestion job
finished. For the async submit/collect flow this means a long-running job
(hours of polling) leaves no trace until completion — and no trace at all if
it never finishes (e.g. a source stuck in has_pending()).

Now the summary is persisted incrementally, for both flows:
- A RUNNING row is written up front, right after the summary is created, so
  the job is observable from the start and a record survives even if the run
  never reaches its final yield.
- A throttled progress snapshot (parent row only, ~every
  PROGRESS_SAVE_INTERVAL=100 tasks) is written after each sync batch and each
  async collect, keeping counters/state current mid-run.
- The final write is unchanged (FINISHED + failed task summaries).

Mechanics:
- IngestionJobSummary.recount(): non-destructive counter recompute, reused by
  _set_ended() (which still truncates to failed tasks on end). duration now
  tolerates ended_at=None for live snapshots.
- save_ingestion_job_summary(..., include_task_summaries=False): parent-only
  upsert so repeated live writes don't re-upsert the growing child rows. The
  upsert is keyed on ingestion_job_summary_id, so live writes update in place.

Tests: test_live_job_summary covers both the sync find_datasets flow and the
async submit/collect flow (RUNNING written first, live progress during the
run, FINISHED last).
---
 ingestify/application/dataset_store.py        |   8 +-
 .../domain/models/ingestion/ingestion_job.py  |  46 +++++
 .../models/ingestion/ingestion_job_summary.py |  16 +-
 .../store/dataset/sqlalchemy/repository.py    |  29 ++-
 ingestify/tests/test_live_job_summary.py      | 165 ++++++++++++++++++
 5 files changed, 252 insertions(+), 12 deletions(-)
 create mode 100644 ingestify/tests/test_live_job_summary.py

diff --git a/ingestify/application/dataset_store.py b/ingestify/application/dataset_store.py
index 8aa0156..79f836b 100644
--- a/ingestify/application/dataset_store.py
+++ b/ingestify/application/dataset_store.py
@@ -191,8 +191,12 @@ def with_file_cache(self):
             self._thread_local.use_file_cache = False
             self._thread_local.file_cache = {}
 
-    def save_ingestion_job_summary(self, ingestion_job_summary):
-        self.dataset_repository.save_ingestion_job_summary(ingestion_job_summary)
+    def save_ingestion_job_summary(
+        self, ingestion_job_summary, include_task_summaries: bool = True
+    ):
+        self.dataset_repository.save_ingestion_job_summary(
+            ingestion_job_summary, include_task_summaries=include_task_summaries
+        )
 
     def get_dataset_last_modified_at_map(
         self, provider: str, dataset_type: str
diff --git a/ingestify/domain/models/ingestion/ingestion_job.py b/ingestify/domain/models/ingestion/ingestion_job.py
index 772d382..78d4413 100644
--- a/ingestify/domain/models/ingestion/ingestion_job.py
+++ b/ingestify/domain/models/ingestion/ingestion_job.py
@@ -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,32 @@ 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.
+
+        Writes the parent summary row only (no child task_summaries), 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, include_task_summaries=False
+        )
 
     def execute(
         self,
@@ -258,6 +289,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")
@@ -462,6 +498,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 +509,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
@@ -563,6 +605,10 @@ def filtered_stream():
                 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)
+
         if ingestion_job_summary.task_count() > 0 or is_first_chunk:
             ingestion_job_summary.set_finished()
             yield ingestion_job_summary
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/infra/store/dataset/sqlalchemy/repository.py b/ingestify/infra/store/dataset/sqlalchemy/repository.py
index e58ddb0..896f833 100644
--- a/ingestify/infra/store/dataset/sqlalchemy/repository.py
+++ b/ingestify/infra/store/dataset/sqlalchemy/repository.py
@@ -739,18 +739,31 @@ def next_identity(self):
         return str(uuid.uuid4())
 
     # TODO: consider moving the IngestionJobSummary methods to a different Repository
-    def save_ingestion_job_summary(self, ingestion_job_summary: IngestionJobSummary):
+    def save_ingestion_job_summary(
+        self,
+        ingestion_job_summary: IngestionJobSummary,
+        include_task_summaries: bool = True,
+    ):
+        """Upsert the summary row (keyed on ingestion_job_summary_id).
+
+        ``include_task_summaries=False`` writes only the parent summary row.
+        Live/progress snapshots use this to cheaply update the counters +
+        state while the job runs, without re-upserting the (potentially large
+        and still-growing) child task_summary rows every time. The final write
+        keeps the default so failed task summaries are persisted.
+        """
         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,
-                }
-            )
+        if include_task_summaries:
+            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,
+                    }
+                )
 
         with self.session_provider.engine.connect() as connection:
             try:
diff --git a/ingestify/tests/test_live_job_summary.py b/ingestify/tests/test_live_job_summary.py
new file mode 100644
index 0000000..eca926e
--- /dev/null
+++ b/ingestify/tests/test_live_job_summary.py
@@ -0,0 +1,165 @@
+"""The IngestionJobSummary is persisted *live*: a RUNNING row is written 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 only written once, at the very end (so a long submit/collect run left no
+trace at all until it finished, and none if it never finished).
+"""
+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 _spy_on_summary_saves(engine):
+    """Record a snapshot of each save_ingestion_job_summary call.
+
+    Snapshots are taken at call time because the summary object is mutated
+    afterwards (recount / _set_ended).
+    """
+    calls = []
+    original = engine.store.save_ingestion_job_summary
+
+    def wrapper(summary, include_task_summaries=True):
+        calls.append(
+            {
+                "state": summary.state,
+                "include_task_summaries": include_task_summaries,
+                "total_tasks": summary.total_tasks,
+                "successful_tasks": summary.successful_tasks,
+                "summary_id": summary.ingestion_job_summary_id,
+            }
+        )
+        return original(summary, include_task_summaries=include_task_summaries)
+
+    engine.store.save_ingestion_job_summary = wrapper
+    return calls
+
+
+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_writes_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)
+    calls = _spy_on_summary_saves(engine)
+
+    engine.run()
+
+    assert calls, "summary was never persisted"
+    # First persisted write is the RUNNING row, parent-only (no task summaries).
+    assert calls[0]["state"] == IngestionJobState.RUNNING
+    assert calls[0]["include_task_summaries"] is False
+    # Final write is FINISHED and carries the task summaries.
+    assert calls[-1]["state"] == IngestionJobState.FINISHED
+    assert calls[-1]["include_task_summaries"] is True
+    assert calls[-1]["successful_tasks"] == 3
+    # A single job -> a single summary id across all writes.
+    assert {c["summary_id"] for c in calls} == {calls[0]["summary_id"]}
+
+
+def test_async_flow_updates_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
+    )
+    calls = _spy_on_summary_saves(engine)
+
+    engine.run()
+
+    # RUNNING row written up front, before any task completed.
+    assert calls[0]["state"] == IngestionJobState.RUNNING
+    assert calls[0]["include_task_summaries"] is False
+    # Progress is written *while polling*, not only at the end: at least one
+    # RUNNING snapshot with tasks already counted precedes the final write.
+    live_progress = [
+        c
+        for c in calls[:-1]
+        if c["state"] == IngestionJobState.RUNNING and c["total_tasks"] > 0
+    ]
+    assert live_progress, "no live progress snapshot was written during collect"
+    # Final write is FINISHED with all tasks accounted for.
+    assert calls[-1]["state"] == IngestionJobState.FINISHED
+    assert calls[-1]["successful_tasks"] == 5

From 22dac810d10185ce42697e8a96b550bd4652f9f3 Mon Sep 17 00:00:00 2001
From: Koen Vossen <info@koenvossen.nl>
Date: Fri, 3 Jul 2026 13:52:51 +0200
Subject: [PATCH 2/4] Persist only FAILED task summaries; assert summaries via
 the database
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Follow-up on the live IngestionJobSummary writes:

- Only FAILED task summaries are ever persisted as child rows (successful,
  ignored and skipped tasks are captured by the counters). This drops the
  include_task_summaries flag entirely: every write — live or final — persists
  the parent row plus any failed task summaries. A successful run writes no
  child rows at all, so live snapshots stay cheap, and failures now show up in
  the database as soon as they happen instead of only at the end.

- test_live_job_summary now asserts on what is actually stored in the database
  (load_ingestion_job_summaries), reading the summaries back after each write
  to prove the RUNNING row + live progress are persisted mid-run (not just that
  save was called).
---
 ingestify/application/dataset_store.py        |   8 +-
 .../domain/models/ingestion/ingestion_job.py  |  15 ++-
 .../store/dataset/sqlalchemy/repository.py    |  36 +++---
 ingestify/tests/test_live_job_summary.py      | 115 ++++++++++--------
 4 files changed, 90 insertions(+), 84 deletions(-)

diff --git a/ingestify/application/dataset_store.py b/ingestify/application/dataset_store.py
index 79f836b..8aa0156 100644
--- a/ingestify/application/dataset_store.py
+++ b/ingestify/application/dataset_store.py
@@ -191,12 +191,8 @@ def with_file_cache(self):
             self._thread_local.use_file_cache = False
             self._thread_local.file_cache = {}
 
-    def save_ingestion_job_summary(
-        self, ingestion_job_summary, include_task_summaries: bool = True
-    ):
-        self.dataset_repository.save_ingestion_job_summary(
-            ingestion_job_summary, include_task_summaries=include_task_summaries
-        )
+    def save_ingestion_job_summary(self, ingestion_job_summary):
+        self.dataset_repository.save_ingestion_job_summary(ingestion_job_summary)
 
     def get_dataset_last_modified_at_map(
         self, provider: str, dataset_type: str
diff --git a/ingestify/domain/models/ingestion/ingestion_job.py b/ingestify/domain/models/ingestion/ingestion_job.py
index 78d4413..6c12c13 100644
--- a/ingestify/domain/models/ingestion/ingestion_job.py
+++ b/ingestify/domain/models/ingestion/ingestion_job.py
@@ -266,20 +266,19 @@ def _save_progress(
     ):
         """Persist a live snapshot of the (still RUNNING) summary.
 
-        Writes the parent summary row only (no child task_summaries), 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.
+        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, include_task_summaries=False
-        )
+        store.save_ingestion_job_summary(ingestion_job_summary)
 
     def execute(
         self,
diff --git a/ingestify/infra/store/dataset/sqlalchemy/repository.py b/ingestify/infra/store/dataset/sqlalchemy/repository.py
index 896f833..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
 
@@ -739,31 +739,27 @@ def next_identity(self):
         return str(uuid.uuid4())
 
     # TODO: consider moving the IngestionJobSummary methods to a different Repository
-    def save_ingestion_job_summary(
-        self,
-        ingestion_job_summary: IngestionJobSummary,
-        include_task_summaries: bool = True,
-    ):
+    def save_ingestion_job_summary(self, ingestion_job_summary: IngestionJobSummary):
         """Upsert the summary row (keyed on ingestion_job_summary_id).
 
-        ``include_task_summaries=False`` writes only the parent summary row.
-        Live/progress snapshots use this to cheaply update the counters +
-        state while the job runs, without re-upserting the (potentially large
-        and still-growing) child task_summary rows every time. The final write
-        keeps the default so failed task summaries are persisted.
+        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 = []
-        if include_task_summaries:
-            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_live_job_summary.py b/ingestify/tests/test_live_job_summary.py
index eca926e..699cef6 100644
--- a/ingestify/tests/test_live_job_summary.py
+++ b/ingestify/tests/test_live_job_summary.py
@@ -1,8 +1,12 @@
-"""The IngestionJobSummary is persisted *live*: a RUNNING row is written 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 only written once, at the very end (so a long submit/collect run left no
-trace at all until it finished, and none if it never finished).
+"""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
 
@@ -14,7 +18,7 @@
 
 
 class SyncSource(Source):
-    """Plain source: find_datasets only (the batch/find_datasets flow)."""
+    """Plain source: find_datasets only (the batch / find_datasets flow)."""
 
     provider = "fake_sync"
 
@@ -86,29 +90,31 @@ def has_pending(self) -> bool:
         return self._in_flight > 0
 
 
-def _spy_on_summary_saves(engine):
-    """Record a snapshot of each save_ingestion_job_summary call.
+def _db_summaries(engine):
+    """Read the ingestion job summaries back from the database."""
+    return engine.store.dataset_repository.load_ingestion_job_summaries()
 
-    Snapshots are taken at call time because the summary object is mutated
-    afterwards (recount / _set_ended).
+
+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.
     """
-    calls = []
+    snapshots = []
     original = engine.store.save_ingestion_job_summary
 
-    def wrapper(summary, include_task_summaries=True):
-        calls.append(
-            {
-                "state": summary.state,
-                "include_task_summaries": include_task_summaries,
-                "total_tasks": summary.total_tasks,
-                "successful_tasks": summary.successful_tasks,
-                "summary_id": summary.ingestion_job_summary_id,
-            }
+    def wrapper(summary):
+        result = original(summary)
+        snapshots.append(
+            [
+                (s.state, s.total_tasks, s.successful_tasks)
+                for s in _db_summaries(engine)
+            ]
         )
-        return original(summary, include_task_summaries=include_task_summaries)
+        return result
 
     engine.store.save_ingestion_job_summary = wrapper
-    return calls
+    return snapshots
 
 
 def _dev_engine(source, tmp_path):
@@ -121,45 +127,54 @@ def _dev_engine(source, tmp_path):
     )
 
 
-def test_sync_flow_writes_running_row_before_finished(tmp_path, monkeypatch):
+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)
-    calls = _spy_on_summary_saves(engine)
+    db_snapshots = _capture_db_after_each_save(engine)
 
     engine.run()
 
-    assert calls, "summary was never persisted"
-    # First persisted write is the RUNNING row, parent-only (no task summaries).
-    assert calls[0]["state"] == IngestionJobState.RUNNING
-    assert calls[0]["include_task_summaries"] is False
-    # Final write is FINISHED and carries the task summaries.
-    assert calls[-1]["state"] == IngestionJobState.FINISHED
-    assert calls[-1]["include_task_summaries"] is True
-    assert calls[-1]["successful_tasks"] == 3
-    # A single job -> a single summary id across all writes.
-    assert {c["summary_id"] for c in calls} == {calls[0]["summary_id"]}
+    # 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_summary_during_collect(tmp_path, monkeypatch):
+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
     )
-    calls = _spy_on_summary_saves(engine)
+    db_snapshots = _capture_db_after_each_save(engine)
 
     engine.run()
 
-    # RUNNING row written up front, before any task completed.
-    assert calls[0]["state"] == IngestionJobState.RUNNING
-    assert calls[0]["include_task_summaries"] is False
-    # Progress is written *while polling*, not only at the end: at least one
-    # RUNNING snapshot with tasks already counted precedes the final write.
-    live_progress = [
-        c
-        for c in calls[:-1]
-        if c["state"] == IngestionJobState.RUNNING and c["total_tasks"] > 0
-    ]
-    assert live_progress, "no live progress snapshot was written during collect"
-    # Final write is FINISHED with all tasks accounted for.
-    assert calls[-1]["state"] == IngestionJobState.FINISHED
-    assert calls[-1]["successful_tasks"] == 5
+    # 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 == []

From 4e9f4bd3e0c44d304513d40c882497c44c6fbe24 Mon Sep 17 00:00:00 2001
From: Koen Vossen <info@koenvossen.nl>
Date: Wed, 8 Jul 2026 11:16:48 +0200
Subject: [PATCH 3/4] 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 <info@koenvossen.nl>
Date: Wed, 8 Jul 2026 12:17:28 +0200
Subject: [PATCH 4/4] 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