fix(dolt): record/alert revision-detection failures + handle cancellation

Review fixes to the import-run framework (A1):

1. detect_revision ran outside the failure handler, so a failed revision probe
   (the most likely external failure) escaped unrecorded — violating "every
   attempt is recorded". Now the running row is created FIRST, then
   detect_revision + last-revision lookup + stage + validate + promote all run
   inside the same handler; the row converts to no_op when the revision is
   unchanged. New test covers a detection exception → recorded failed + alert.

2. asyncio.CancelledError (BaseException, not caught by except Exception) left a
   permanent running row on deploy/scheduler shutdown. Now caught explicitly:
   best-effort mark failed, then re-raise the cancellation (never swallowed).
   New test asserts the run is failed and the error re-propagates.

Full suite 682 passed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/services/data_import.py

@@ -25,6 +25,7 @@ table (idempotency queries the last run), no aggregate tables.
 
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import json
 import logging
@@ -170,25 +171,11 @@ async def run_import(
                     logger.info("data_import %s: lock held, skipping", source)
                     return None
 
-            revision = await importer.detect_revision(session)
+            # Record the attempt FIRST — before the external revision probe, the
-            last_rev = await _last_promoted_revision(session, source)
+            # most likely failure — so anything below is recorded and alerted and
-            if revision is not None and revision == last_rev:
+            # never escapes unrecorded. Revision is filled in once detected.
-                run = DataImportRun(
-                    source=source,
-                    revision=revision,
-                    status=STATUS_NO_OP,
-                    started_at=_now(),
-                    completed_at=_now(),
-                )
-                session.add(run)
-                await session.commit()
-                await session.refresh(run)
-                logger.info("data_import %s: no_op (revision %s)", source, revision)
-                return run
-
             run = DataImportRun(
                 source=source,
-                revision=revision,
                 status=STATUS_RUNNING,
                 started_at=_now(),
             )
@@ -197,6 +184,16 @@ async def run_import(
             await session.refresh(run)
 
             try:
+                revision = await importer.detect_revision(session)
+                run.revision = revision
+                last_rev = await _last_promoted_revision(session, source)
+                if revision is not None and revision == last_rev:
+                    run.status = STATUS_NO_OP
+                    run.completed_at = _now()
+                    await session.commit()
+                    logger.info("data_import %s: no_op (revision %s)", source, revision)
+                    return run
+
                 staged = await importer.stage(session)
                 result = await importer.validate(session, staged)
                 run.source_max_date = result.source_max_date
@@ -230,6 +227,22 @@ async def run_import(
                 )
                 return run
 
+            except asyncio.CancelledError:
+                # Deploy / scheduler shutdown: best-effort mark failed so no
+                # ``running`` row lingers, then let the cancellation propagate —
+                # never swallow it.
+                try:
+                    await session.rollback()
+                    run.status = STATUS_FAILED
+                    run.error_details = "cancelled"
+                    run.completed_at = _now()
+                    await session.commit()
+                except BaseException:  # noqa: BLE001 — best-effort during teardown
+                    logger.warning(
+                        "data_import %s: could not record cancellation", source
+                    )
+                raise
+
             except Exception as exc:  # noqa: BLE001 — record + alert, don't crash the job
                 await session.rollback()
                 run.status = STATUS_FAILED

tests/unit/test_data_import_framework.py

@@ -14,6 +14,7 @@ sequenced around the ``run_import`` call.
 
 from __future__ import annotations
 
+import asyncio
 import os
 import tempfile
 from datetime import date, datetime, timezone
@@ -83,6 +84,8 @@ class FakeImporter:
         self.staged_called = True
         if self.raise_in == "stage":
             raise RuntimeError("boom-stage")
+        if self.raise_in == "cancel":
+            raise asyncio.CancelledError()
         return list(range(self.n_rows))
 
     async def validate(self, db, staged):
@@ -199,3 +202,32 @@ async def test_exception_in_promote_rolls_back(engine):
     assert "boom-promote" in (run.error_details or "")
     assert await _count(factory, FundamentalSnapshot) == 3  # partial write rolled back
     assert await _count(factory, SystemEvent) == 1
+
+
+async def test_detection_failure_records_and_alerts(engine):
+    """The external revision probe is the most likely failure — it must produce a
+    recorded failed run + alert, not an unrecorded escaping exception."""
+    factory = _factory(engine)
+    imp = FakeImporter("rev1", raise_in="detect")
+
+    run = await run_import(imp, engine=engine)
+
+    assert run is not None and run.status == STATUS_FAILED
+    assert "boom-detect" in (run.error_details or "")
+    assert imp.staged_called is False
+    assert await _count(factory, FundamentalSnapshot) == 0
+    assert await _count(factory, SystemEvent) == 1  # alerted
+    runs = await _runs(factory)
+    assert len(runs) == 1 and runs[0].status == STATUS_FAILED  # attempt recorded
+
+
+async def test_cancellation_marks_failed_and_reraises(engine):
+    factory = _factory(engine)
+
+    with pytest.raises(asyncio.CancelledError):
+        await run_import(FakeImporter("rev1", raise_in="cancel"), engine=engine)
+
+    runs = await _runs(factory)
+    assert len(runs) == 1 and runs[0].status == STATUS_FAILED  # no lingering running row
+    assert runs[0].error_details == "cancelled"
+    assert await _count(factory, FundamentalSnapshot) == 0