From 8b45bdb3611659636c2951dc646a8c2ce3115ae8 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Wed, 22 Jul 2026 09:25:14 +0200 Subject: [PATCH] fix(dolt): record/alert revision-detection failures + handle cancellation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- app/services/data_import.py | 47 +++++++++++++++--------- tests/unit/test_data_import_framework.py | 32 ++++++++++++++++ 2 files changed, 62 insertions(+), 17 deletions(-) diff --git a/app/services/data_import.py b/app/services/data_import.py index 494ffb3..7fbfb5e 100644 --- a/app/services/data_import.py +++ b/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) - last_rev = await _last_promoted_revision(session, source) - if revision is not None and revision == last_rev: - 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 - + # Record the attempt FIRST — before the external revision probe, the + # most likely failure — so anything below is recorded and alerted and + # never escapes unrecorded. Revision is filled in once detected. 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 diff --git a/tests/unit/test_data_import_framework.py b/tests/unit/test_data_import_framework.py index 9a30141..67b200c 100644 --- a/tests/unit/test_data_import_framework.py +++ b/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