Keep Dolt checks and export events transaction-bound

Author: TonisOrmisson

src/openstatspec/sql/wide.py

@@ -579,18 +579,18 @@ def _bound_catalog_transaction(
         try:
             with connection.begin():
                 yield connection
+                after = _capture_dolt_state(
+                    connection, profile_name=profile_name,
+                    audit_relations=audit_relations,
+                )
+                _require_dolt_success_identity(before, after, phase=phase)
         except Exception:
             after = _capture_dolt_state(
                 connection, profile_name=profile_name,
                 audit_relations=audit_relations,
             )
             _dolt_failure_boundary_evidence(before, after)
             raise
-        after = _capture_dolt_state(
-            connection, profile_name=profile_name,
-            audit_relations=audit_relations,
-        )
-        _require_dolt_success_identity(before, after, phase=phase)
 
 
 def dolt_state_snapshot(
@@ -1417,6 +1417,13 @@ def record_export_operation(
                     "source_item": destination,
                     "details": operation_details,
                 },) if operation_details else ()),
+                {
+                    "code": "export-operation-dataset-binding",
+                    "detail": "Export operation dataset identity recorded for audit.",
+                    "severity": "info",
+                    "source_item": destination,
+                    "details": {},
+                },
             ),
         )
     return operation_id
@@ -1435,6 +1442,30 @@ def _export_operation_row(
     return row
 
 
+def _export_operation_dataset_id(
+    connection: Any, normative: Any, operation_id: str,
+) -> str:
+    dataset_ids = set(connection.execute(
+        select(normative.fidelity_event.c.dataset_id)
+        .where(normative.fidelity_event.c.operation_id == operation_id)
+        .where(normative.fidelity_event.c.dataset_id.is_not(None))
+    ).scalars())
+    if len(dataset_ids) != 1:
+        raise UnsupportedOperationError(
+            "The export operation is not bound to exactly one dataset."
+        )
+    dataset_id = str(dataset_ids.pop())
+    if connection.execute(
+        select(normative.dataset.c.dataset_id).where(
+            normative.dataset.c.dataset_id == dataset_id
+        )
+    ).scalar_one_or_none() is None:
+        raise UnsupportedOperationError(
+            "The export operation dataset no longer exists."
+        )
+    return dataset_id
+
+
 def finish_export_operation(
     *, database_url: str, operation_id: str,
     dolt_conformance_source: Any | None = None,
@@ -1504,12 +1535,15 @@ def fail_export_operation(
             raise UnsupportedOperationError(
                 "Only a started export operation can be failed."
             )
+        dataset_id = _export_operation_dataset_id(
+            connection, normative, operation_id,
+        )
         finish_normative_operation(
             connection, normative, operation_id=operation_id, status="failed",
         )
         record_normative_fidelity_events(
             connection, normative, operation_id=operation_id,
-            dataset_id=None, direction="export", events=(event,),
+            dataset_id=dataset_id, direction="export", events=(event,),
         )
 
 
@@ -1529,9 +1563,12 @@ def record_export_backup_retained(
             raise UnsupportedOperationError(
                 "A retained backup warning requires a succeeded export."
             )
+        dataset_id = _export_operation_dataset_id(
+            connection, normative, operation_id,
+        )
         record_normative_fidelity_events(
             connection, normative, operation_id=operation_id,
-            dataset_id=None, direction="export", events=({
+            dataset_id=dataset_id, direction="export", events=({
                 "code": "backup_retained",
                 "detail": "A successful export retained its durable prior-file backup.",
                 "severity": "warning",
@@ -1578,6 +1615,7 @@ def record_export_cleanup_failure(
                 normative.operation.c.operation_id == operation_id
             )
         ).mappings().one_or_none()
+        dataset_id = None
         if row is None:
             failed_at = datetime.now(UTC).replace(tzinfo=None)
             record_normative_operation(
@@ -1590,12 +1628,15 @@ def record_export_cleanup_failure(
                 raise UnsupportedOperationError(
                     "Existing export operation cannot transition to cleanup failure."
                 )
+            dataset_id = _export_operation_dataset_id(
+                connection, normative, operation_id,
+            )
             finish_normative_operation(
                 connection, normative, operation_id=operation_id, status="failed",
             )
         record_normative_fidelity_events(
             connection, normative, operation_id=operation_id,
-            dataset_id=None, direction="export", events=(event,),
+            dataset_id=dataset_id, direction="export", events=(event,),
         )
     return operation_id
 

tests/test_review_transaction_boundary.py

@@ -0,0 +1,117 @@
+"""Regression tests for bound commit checks and export dataset linkage."""
+
+import sqlite3
+
+from sqlalchemy import create_engine
+
+from openstatspec.sql import wide
+
+
+def _variables():
+    return [{
+        "ordinal": 1,
+        "source_name": "name",
+        "physical_name": "name",
+        "storage_kind": "string",
+        "string_width": 8,
+        "label": "",
+        "format": "A8",
+        "measure": "nominal",
+        "alignment": "left",
+        "display_width": 8,
+        "value_labels": "{}",
+        "missing_ranges": "[]",
+    }]
+
+
+def test_dolt_completion_identity_is_checked_before_transaction_commit(
+    monkeypatch,
+):
+    engine = create_engine("sqlite://")
+    state = {
+        "database": "catalog",
+        "active_branch": "main",
+        "head": "abc123",
+    }
+    transaction_states = []
+
+    def capture(connection, **_kwargs):
+        transaction_states.append(connection.in_transaction())
+        return state
+
+    monkeypatch.setattr(wide, "_capture_dolt_state", capture)
+    with wide._bound_catalog_transaction(
+        engine=engine,
+        profile_name="dolt",
+        active={
+            "working_set_binding": {
+                "database": "catalog",
+                "active_branch": "main",
+            },
+        },
+        audit_relations=set(),
+        phase="test",
+    ):
+        pass
+
+    assert transaction_states == [False, True]
+
+
+def test_export_lifecycle_events_remain_linked_to_the_dataset(tmp_path):
+    path = tmp_path / "export-events.sqlite"
+    database_url = f"sqlite:///{path}"
+    wide.create_wide_dataset(
+        database_url=database_url,
+        dataset_id="sample",
+        source_name="sample.sav",
+        source_format="SAV",
+        rows=[{"name": "ok"}],
+        variables=_variables(),
+    )
+
+    failed_id = wide.record_export_operation(
+        database_url=database_url,
+        dataset_id="sample",
+        destination="failed.sav",
+        allowed_fidelity_events=(),
+        terminal=False,
+    )
+    wide.fail_export_operation(
+        database_url=database_url,
+        operation_id=failed_id,
+        failure_details={"reason": "test"},
+    )
+    succeeded_id = wide.record_export_operation(
+        database_url=database_url,
+        dataset_id="sample",
+        destination="succeeded.sav",
+        allowed_fidelity_events=(),
+        terminal=False,
+    )
+    wide.finish_export_operation(
+        database_url=database_url,
+        operation_id=succeeded_id,
+    )
+    wide.record_export_backup_retained(
+        database_url=database_url,
+        operation_id=succeeded_id,
+        destination="succeeded.sav",
+        backup="succeeded.sav.backup",
+        cleanup_error=RuntimeError("test"),
+    )
+
+    events = wide.read_fidelity_events(
+        database_url=database_url,
+        dataset_id="sample",
+    )
+    assert {event["code"] for event in events} == {
+        "backup_retained",
+        "export_failed",
+    }
+    connection = sqlite3.connect(path)
+    assert connection.execute(
+        "select count(*) from fidelity_event "
+        "where operation_id in (?, ?) and dataset_id is null",
+        (failed_id, succeeded_id),
+    ).fetchone() == (0,)
+    connection.close()