diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5573f55..41073ce 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,7 +20,7 @@ jobs: uses: actions/checkout@v7 with: &specification-checkout repository: OpenStatSpec/specification - ref: 7edcad38470fffebfe2306f22fdecb6892f8eece + ref: 34141dda023d9e0217c37c232e39f436edfb0746 path: openstatspec-specification - name: Checkout required SPSS engine uses: actions/checkout@v7 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 7595b8e..2c5327e 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -39,7 +39,7 @@ jobs: uses: actions/checkout@v7 with: repository: OpenStatSpec/specification - ref: 7edcad38470fffebfe2306f22fdecb6892f8eece + ref: 34141dda023d9e0217c37c232e39f436edfb0746 path: openstatspec-specification - name: Checkout required SPSS engine uses: actions/checkout@v7 diff --git a/src/openstatspec/sql/capabilities.py b/src/openstatspec/sql/capabilities.py index 0620623..29851ce 100644 --- a/src/openstatspec/sql/capabilities.py +++ b/src/openstatspec/sql/capabilities.py @@ -15,7 +15,7 @@ from .profiles import profile_for_url from ..core import UnsupportedOperationError -SPECIFICATION_COMMIT = "7edcad38470fffebfe2306f22fdecb6892f8eece" +SPECIFICATION_COMMIT = "34141dda023d9e0217c37c232e39f436edfb0746" SPECIFICATION_RELEASE: str | None = None SERVER_POLICIES = { diff --git a/src/openstatspec/sql/workflow.py b/src/openstatspec/sql/workflow.py index b4c41d7..5b2fd97 100644 --- a/src/openstatspec/sql/workflow.py +++ b/src/openstatspec/sql/workflow.py @@ -43,6 +43,7 @@ _SQLITE_MINIMUM_VERSION = (3, 35, 0) _SQLITE_MAXIMUM_VERSION = (4, 0, 0) _TOKEN = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") +_STAGING_PREFIX = "__openstatspec_staging_" _FORBIDDEN = { "ALTER", "ANALYZE", "ATTACH", "CALL", "COPY", "CREATE", "DELETE", "DETACH", "DROP", "EXEC", "EXECUTE", "GRANT", "INSERT", "LOAD", @@ -81,7 +82,9 @@ def all(self) -> tuple[Table, ...]: return tuple(getattr(self, item.name) for item in fields(self)) -def workflow_catalog(metadata: MetaData) -> WorkflowTables: +def workflow_catalog( + metadata: MetaData, *, _include_staging_relation_key: bool = True, +) -> WorkflowTables: core_catalog(metadata) identity = Table( "transformation_profile_identity", metadata, @@ -138,6 +141,10 @@ def workflow_catalog(metadata: MetaData) -> WorkflowTables: Column("status", String(16), nullable=False), Column("executor_identity", String(128), nullable=False), Column("correlation_id", String(36), nullable=False), + *( + [Column("staging_relation_key", String(512))] + if _include_staging_relation_key else [] + ), Column("engine_name", String(64), nullable=False), Column("engine_version", String(128), nullable=False), Column("dialect_profile", String(32), nullable=False), @@ -583,6 +590,78 @@ def _validate_workflow_schema(connection: Any, tables: WorkflowTables) -> None: ) +def _staging_relation_key(relation_name: str) -> str: + return f"sqlite:main.{relation_name}" + + +def _owned_staging_relation_name(value: Any) -> str: + prefix = "sqlite:main." + if not isinstance(value, str) or not value.startswith(prefix): + raise TransformationError( + "reconciliation_ownership_unverified", + "The recorded staging relation key is missing or is not a SQLite main relation.", + ) + relation_name = value[len(prefix):] + if ( + _TOKEN.fullmatch(relation_name) is None + or not relation_name.startswith(_STAGING_PREFIX) + ): + raise TransformationError( + "reconciliation_ownership_unverified", + "The recorded staging relation is outside the profile-owned staging namespace.", + ) + return relation_name + + +def _migrate_staging_relation_key( + connection: Any, tables: WorkflowTables, +) -> None: + """Upgrade pre-recovery v2 without claiming unrecorded staging objects.""" + legacy_tables = workflow_catalog( + MetaData(), _include_staging_relation_key=False + ) + _validate_workflow_schema(connection, legacy_tables) + quote = connection.dialect.identifier_preparer.quote + for trigger_name in _workflow_trigger_sql( + connection, legacy_tables + ): + connection.exec_driver_sql( + f"DROP TRIGGER {quote(trigger_name)}" + ) + + migration_name = "__oss_migrating_transformation_run" + migration_metadata = MetaData() + tables.transformation_version.to_metadata(migration_metadata) + migration_run = tables.transformation_run.to_metadata( + migration_metadata, name=migration_name + ) + migration_run.create(connection) + target_columns = [column.name for column in tables.transformation_run.columns] + target_sql = ", ".join(quote(name) for name in target_columns) + source_sql = ", ".join( + "NULL" if name == "staging_relation_key" else quote(name) + for name in target_columns + ) + connection.exec_driver_sql( + f"INSERT INTO {quote(migration_name)} ({target_sql}) " + f"SELECT {source_sql} FROM {quote(tables.transformation_run.name)}" + ) + connection.exec_driver_sql("PRAGMA defer_foreign_keys = ON") + connection.exec_driver_sql( + f"DROP TABLE {quote(tables.transformation_run.name)}" + ) + connection.exec_driver_sql( + f"ALTER TABLE {quote(migration_name)} " + f"RENAME TO {quote(tables.transformation_run.name)}" + ) + if connection.exec_driver_sql("PRAGMA foreign_key_check").first() is not None: + raise TransformationError( + "profile_incompatible", + "Workflow migration would violate a foreign-key ownership invariant.", + ) + _create_workflow_triggers(connection, tables) + + def create_workflow_catalog(connection: Any, tables: WorkflowTables) -> None: """Create or validate the additive optional-profile catalog.""" if connection.dialect.name != "sqlite": @@ -616,6 +695,13 @@ def create_workflow_catalog(connection: Any, tables: WorkflowTables) -> None: identity = dict(rows[0]) if identity["contract_id"] != PROFILE_ID or identity["core_contract_id"] != CATALOG_CONTRACT_ID or identity["schema_version"] != PROFILE_SCHEMA_VERSION: raise TransformationError("profile_incompatible", "The workflow profile identity is incompatible.") + run_columns = { + str(column["name"]) for column in inspect(connection).get_columns( + tables.transformation_run.name + ) + } + if "staging_relation_key" not in run_columns: + _migrate_staging_relation_key(connection, tables) _validate_workflow_schema(connection, tables) @@ -1478,6 +1564,32 @@ def _validated_parameters( return values, _sha(_canonical_json(hash_envelope)), encoded_by_name +def _next_run_event_ordinal( + connection: Any, tables: WorkflowTables, run_id: str, +) -> int: + ordinals = connection.execute( + select(tables.transformation_event.c.event_ordinal) + .where(tables.transformation_event.c.transformation_run_id == run_id) + ).scalars().all() + return max((int(value) for value in ordinals), default=0) + 1 + + +def _append_run_event( + connection: Any, tables: WorkflowTables, *, run_id: str, + code: str, phase: str, +) -> None: + connection.execute(insert(tables.transformation_event).values( + transformation_event_id=str(uuid4()), transformation_run_id=run_id, + event_ordinal=_next_run_event_ordinal(connection, tables, run_id), + severity="error", event_code=code, execution_phase=phase, + safe_detail_json=_canonical_json({ + "error_code": code, "execution_phase": phase, + "correlation_id_hash": _sha(run_id), + }), + created_at=_now(), + )) + + def _record_failure( engine: Any, tables: WorkflowTables, run_id: str, code: str, phase: str, ) -> None: @@ -1494,16 +1606,90 @@ def _record_failure( raise TransformationError( "publication_failed", "Run failure transition was not started -> failed." ) - connection.execute(insert(tables.transformation_event).values( - transformation_event_id=str(uuid4()), transformation_run_id=run_id, - event_ordinal=1, severity="error", event_code=code, - execution_phase=phase, - safe_detail_json=_canonical_json({ - "error_code": code, "execution_phase": phase, - "correlation_id_hash": _sha(run_id), - }), - created_at=_now(), - )) + _append_run_event( + connection, tables, run_id=run_id, code=code, phase=phase + ) + + +def _record_cleanup_failure( + engine: Any, tables: WorkflowTables, run_id: str, +) -> None: + with engine.begin() as connection: + status = connection.execute( + select(tables.transformation_run.c.status).where( + tables.transformation_run.c.transformation_run_id == run_id + ) + ).scalar_one() + if status != "started": + raise TransformationError( + "publication_failed", + "Cleanup failure may only quarantine a started run.", + ) + _append_run_event( + connection, tables, run_id=run_id, + code="cleanup_failed", phase="cleanup", + ) + + +def _relation_kind(connection: Any, relation_name: str) -> str | None: + inspector = inspect(connection) + if relation_name in inspector.get_view_names(): + return "view" + if relation_name in inspector.get_table_names(): + return "table" + return None + + +def _drop_relation_if_present( + connection: Any, relation_name: str, +) -> None: + relation_kind = _relation_kind(connection, relation_name) + if relation_kind is None: + return + quote = connection.dialect.identifier_preparer.quote + connection.exec_driver_sql( + f"DROP {relation_kind.upper()} {quote(relation_name)}" + ) + + +def _assert_relation_absent(connection: Any, relation_name: str) -> None: + if _relation_kind(connection, relation_name) is not None: + raise TransformationError( + "cleanup_failed", + "The recorded profile-owned relation is still present after cleanup.", + ) + + +def _assert_staging_key_uniquely_owned( + connection: Any, tables: WorkflowTables, run_id: str, relation_key: str, +) -> None: + owners = connection.execute( + select(tables.transformation_run.c.transformation_run_id).where( + tables.transformation_run.c.staging_relation_key == relation_key + ) + ).scalars().all() + if [str(value) for value in owners] != [run_id]: + raise TransformationError( + "reconciliation_ownership_unverified", + "The recorded staging relation key is not uniquely owned by this run.", + ) + + +def _cleanup_execution_relations( + engine: Any, tables: WorkflowTables, run_id: str, +) -> None: + with engine.begin() as connection: + relation_key = connection.execute( + select(tables.transformation_run.c.staging_relation_key).where( + tables.transformation_run.c.transformation_run_id == run_id + ) + ).scalar_one() + staging_name = _owned_staging_relation_name(relation_key) + _assert_staging_key_uniquely_owned( + connection, tables, run_id, relation_key + ) + _drop_relation_if_present(connection, staging_name) + _assert_relation_absent(connection, staging_name) @contextmanager @@ -1787,10 +1973,14 @@ def execute_transformation( "status": "already_exists", } run_id = str(uuid4()) + relation_name = "derived_" + UUID(derived_id).hex + staging_name = _STAGING_PREFIX + UUID(run_id).hex + staging_key = _staging_relation_key(staging_name) connection.execute(insert(tables.transformation_run).values( transformation_run_id=run_id, transformation_version_id=version_id, status="started", executor_identity="openstatspec-python", - correlation_id=run_id, engine_name=connection.dialect.name, + correlation_id=run_id, staging_relation_key=staging_key, + engine_name=connection.dialect.name, engine_version=str(getattr(connection.dialect, "server_version_info", "unknown")), dialect_profile=profile.name, capability_snapshot_json=_canonical_json({"dialect_family": profile.name}), specification_commit=SPECIFICATION_COMMIT or "unreleased", definition_hash=version["definition_hash"], @@ -1815,8 +2005,6 @@ def execute_transformation( snapshot_hash_kind="relation_snapshot", snapshot_hash_algorithm="sha256", snapshot_hash_version="openstatspec-relation-snapshot-v1", )) - relation_name = "derived_" + UUID(derived_id).hex - staging_name = "__oss_stage_" + UUID(run_id).hex schema_hash = _sha(version["output_schema_json"]) phase = "input_validation" try: @@ -1939,6 +2127,9 @@ def execute_transformation( content_hash_policy = "computed" variable_ids: dict[str, str] = {} phase = "publication" + _assert_relation_absent( + connection, _owned_staging_relation_name(staging_key) + ) changed = connection.execute( update(tables.transformation_run) .where( @@ -1999,26 +2190,15 @@ def execute_transformation( )) except Exception as error: code = error.code if isinstance(error, TransformationError) else "execution_failed" - cleanup_error = None try: - with engine.begin() as cleanup: - inspector = inspect(cleanup) - quote = cleanup.dialect.identifier_preparer.quote - for candidate in (staging_name, relation_name): - if candidate in inspector.get_view_names(): - cleanup.exec_driver_sql(f"DROP VIEW {quote(candidate)}") - if inspector.has_table(candidate): - cleanup.exec_driver_sql(f"DROP TABLE {quote(candidate)}") - except Exception as cleanup_exception: - cleanup_error = cleanup_exception - code = "cleanup_failed" - phase = "cleanup" - finally: - _record_failure(engine, tables, run_id, code, phase) - if cleanup_error is not None: + _cleanup_execution_relations(engine, tables, run_id) + except Exception as cleanup_error: + _record_cleanup_failure(engine, tables, run_id) raise TransformationError( - "cleanup_failed", "Profile-owned staging cleanup failed." + "cleanup_failed", + "Profile-owned staging cleanup failed; the run is quarantined.", ) from cleanup_error + _record_failure(engine, tables, run_id, code, phase) raise return { "derived_dataset_id": derived_id, "transformation_run_id": run_id, @@ -2056,7 +2236,7 @@ def derive_dataset( def reconcile_started_runs( *, database_url: str, older_than_seconds: int = 0, ) -> dict[str, Any]: - """Fail interrupted runs and remove only reserved profile staging objects.""" + """Reconcile started runs using only their durable, profile-owned staging key.""" if older_than_seconds < 0: raise TransformationError( "reconciliation_invalid", "older_than_seconds cannot be negative." @@ -2072,7 +2252,7 @@ def reconcile_started_runs( select(tables.transformation_run).where( tables.transformation_run.c.status == "started", tables.transformation_run.c.started_at <= cutoff, - ) + ).order_by(tables.transformation_run.c.transformation_run_id) ).mappings().all() run_ids = {str(row["transformation_run_id"]) for row in runs} published = connection.execute( @@ -2084,40 +2264,36 @@ def reconcile_started_runs( raise TransformationError( "profile_incompatible", "A started run already owns a derived dataset." ) - inspector = inspect(connection) - quote = connection.dialect.identifier_preparer.quote - expected_staging = { - "__oss_stage_" + UUID(run_id).hex for run_id in run_ids - } - staging_relations = expected_staging & set( - inspector.get_table_names() + inspector.get_view_names() - ) - for name in sorted(staging_relations): - if name in inspector.get_view_names(): - connection.exec_driver_sql(f"DROP VIEW {quote(name)}") - else: - connection.exec_driver_sql(f"DROP TABLE {quote(name)}") + for row in runs: run_id = str(row["transformation_run_id"]) - connection.execute( + relation_key = row["staging_relation_key"] + staging_name = _owned_staging_relation_name(relation_key) + _assert_staging_key_uniquely_owned( + connection, tables, run_id, relation_key + ) + _drop_relation_if_present(connection, staging_name) + _assert_relation_absent(connection, staging_name) + changed = connection.execute( update(tables.transformation_run).where( tables.transformation_run.c.transformation_run_id == run_id, tables.transformation_run.c.status == "started", ).values(status="failed", completed_at=_now()) + ).rowcount + if changed != 1: + raise TransformationError( + "publication_failed", + "Reconciliation run transition was not started -> failed.", + ) + _append_run_event( + connection, tables, run_id=run_id, + code="interrupted_run", phase="reconciliation", ) - connection.execute(insert(tables.transformation_event).values( - transformation_event_id=str(uuid4()), transformation_run_id=run_id, - event_ordinal=1, severity="error", event_code="interrupted_run", - execution_phase="reconciliation", - safe_detail_json=_canonical_json({ - "error_code": "interrupted_run", - "execution_phase": "reconciliation", - "correlation_id_hash": _sha(run_id), - }), created_at=_now(), - )) reconciled.append(run_id) - return {"reconciled": len(reconciled), "transformation_run_ids": reconciled} - + return { + "reconciled": len(reconciled), + "transformation_run_ids": reconciled, + } def _next_disposition_ordinal( diff --git a/tests/test_cli.py b/tests/test_cli.py index b8fcfb4..0d9cbaa 100755 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -40,7 +40,7 @@ def test_capability_matrix_is_public_and_cli_matches_engine_boundary(capsys) -> matrix = openstatspec.capability_matrix() assert matrix["specification_status"] == "release_candidate" assert matrix["specification_release"] is None - assert matrix["specification_commit"] == "7edcad38470fffebfe2306f22fdecb6892f8eece" + assert matrix["specification_commit"] == "34141dda023d9e0217c37c232e39f436edfb0746" assert matrix["directions"] == ["import", "export", "semantic_round_trip"] assert matrix["active_connection"] is None assert matrix["engine"]["package"] == "openstatspec-pyspssio" diff --git a/tests/test_sql_workflow.py b/tests/test_sql_workflow.py index 393afe1..a418a70 100644 --- a/tests/test_sql_workflow.py +++ b/tests/test_sql_workflow.py @@ -1,6 +1,7 @@ import hashlib import json import os +from datetime import UTC, datetime from pathlib import Path import rfc8785 @@ -8,16 +9,19 @@ from uuid import UUID import pytest -from sqlalchemy import MetaData, text +from sqlalchemy import MetaData, insert, text +from sqlalchemy.exc import OperationalError from sqlalchemy.dialects import sqlite from sqlalchemy.schema import CreateTable import openstatspec +import openstatspec.sql.workflow as workflow import openstatspec.cli from openstatspec.sql.workflow import ( PROFILE_ID, PROFILE_SCHEMA_VERSION, TransformationError, _definition_hash, - _assert_sqlite_server_version, _workflow_engine, - transformation_capabilities, workflow_catalog, + _assert_sqlite_server_version, _create_workflow_triggers, + _workflow_engine, create_workflow_catalog, transformation_capabilities, + workflow_catalog, ) from openstatspec.sql.wide import create_wide_dataset @@ -138,7 +142,8 @@ def test_materialized_sql_workflow_is_immutable_audited_and_queryable(catalog): PROFILE_ID, "openstatspec-strict-wide-table-v1", ) run_row = connection.execute( - "select status, parameters_hash, input_set_hash from transformation_run" + "select status, parameters_hash, input_set_hash, staging_relation_key " + "from transformation_run" ).fetchone() assert run_row[0] == "succeeded" envelope = connection.execute( @@ -161,6 +166,11 @@ def test_materialized_sql_workflow_is_immutable_audited_and_queryable(catalog): rfc8785.dumps(parameters_document) ).hexdigest() assert len(run_row[2]) == 64 + assert run_row[3].startswith("sqlite:main.__openstatspec_staging_") + assert connection.execute( + "select count(*) from sqlite_master where name = ?", + (run_row[3].removeprefix("sqlite:main."),), + ).fetchone() == (0,) relation = result["physical_relation_name"] assert connection.execute( f'select __row_ordinal, score, grp from "{relation}" order by __row_ordinal' @@ -737,7 +747,7 @@ def test_declared_non_null_output_is_enforced_and_audit_is_redacted(catalog): assert "SELECT" not in event[2] -def test_started_run_and_reserved_staging_are_reconciled(catalog): +def test_started_run_and_recorded_staging_are_reconciled(catalog): url, path, parent_id = catalog registered = openstatspec.register_sql_transformation( database_url=url, parent_dataset_id=parent_id, @@ -747,20 +757,26 @@ def test_started_run_and_reserved_staging_are_reconciled(catalog): }], transformation_name="interrupted", ) run_id = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" + staging_name = "__openstatspec_staging_crash_case" + staging_key = f"sqlite:main.{staging_name}" connection = sqlite3.connect(path) connection.execute("pragma foreign_keys = on") connection.execute( - "insert into transformation_run values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + "insert into transformation_run (" + "transformation_run_id, transformation_version_id, status, " + "executor_identity, correlation_id, staging_relation_key, engine_name, " + "engine_version, dialect_profile, capability_snapshot_json, " + "specification_commit, definition_hash, parameters_hash, input_set_hash, " + "started_at, completed_at" + ") values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", ( run_id, registered["transformation_version_id"], "started", "test", - run_id, "sqlite", "test", "sqlite", "{}", "test", + run_id, staging_key, "sqlite", "test", "sqlite", "{}", "test", registered["definition_hash"], "0" * 64, "1" * 64, "2000-01-01 00:00:00", None, ), ) - connection.execute( - 'create table "__oss_stage_aaaaaaaaaaaa4aaa8aaaaaaaaaaaaaaa" (x integer)' - ) + connection.execute(f'create table "{staging_name}" (x integer)') connection.commit() reconciled = openstatspec.reconcile_sql_transformation_runs(database_url=url) assert reconciled["reconciled"] == 1 @@ -768,8 +784,13 @@ def test_started_run_and_reserved_staging_are_reconciled(catalog): "select status from transformation_run where transformation_run_id = ?", (run_id,) ).fetchone() == ("failed",) assert connection.execute( - "select count(*) from sqlite_master where name like '__oss_stage_%'" + "select count(*) from sqlite_master where name = ?", (staging_name,) ).fetchone() == (0,) + assert connection.execute( + "select event_ordinal, event_code, execution_phase " + "from transformation_event where transformation_run_id = ?", + (run_id,), + ).fetchall() == [(1, "interrupted_run", "reconciliation")] def test_weight_propagation_requires_verified_identity_and_safe_rows(catalog): @@ -1352,3 +1373,365 @@ def test_column_aggregate_with_aggregate_semantics_succeeds(catalog): row_semantics="aggregate", transformation_name="column_aggregate_valid", ) assert registered["version_number"] == 1 + + +def _insert_started_run(connection, registered, run_id, staging_key): + connection.execute( + "insert into transformation_run (" + "transformation_run_id, transformation_version_id, status, " + "executor_identity, correlation_id, staging_relation_key, engine_name, " + "engine_version, dialect_profile, capability_snapshot_json, " + "specification_commit, definition_hash, parameters_hash, input_set_hash, " + "started_at, completed_at" + ") values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + ( + run_id, registered["transformation_version_id"], "started", "test", + run_id, staging_key, "sqlite", "test", "sqlite", "{}", "test", + registered["definition_hash"], "0" * 64, "1" * 64, + "2000-01-01 00:00:00", None, + ), + ) + + +def _create_legacy_workflow_catalog(url, *, with_started_run=False): + engine = _workflow_engine(url, "sqlite") + legacy = workflow_catalog( + MetaData(), _include_staging_relation_key=False + ) + run_id = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" + stage_name = "__oss_stage_bbbbbbbbbbbb4bbb8bbbbbbbbbbbbbbb" + with engine.begin() as connection: + legacy.transformation_profile_identity.metadata.create_all( + connection, tables=list(legacy.all()) + ) + connection.execute(insert( + legacy.transformation_profile_identity + ).values( + profile_identity_key=1, contract_id=PROFILE_ID, + schema_version=PROFILE_SCHEMA_VERSION, + core_contract_id="openstatspec-strict-wide-table-v1", + created_at=datetime.now(UTC), + )) + _create_workflow_triggers(connection, legacy) + if with_started_run: + transformation_id = "cccccccc-cccc-4ccc-8ccc-cccccccccccc" + version_id = "dddddddd-dddd-4ddd-8ddd-dddddddddddd" + definition_hash = "a" * 64 + connection.execute(insert( + legacy.transformation_definition + ).values( + transformation_id=transformation_id, + stable_name="legacy_interrupted", title="Legacy interrupted", + created_at=datetime.now(UTC), + )) + connection.execute(insert( + legacy.transformation_version + ).values( + transformation_version_id=version_id, + transformation_id=transformation_id, version_number=1, + query_sql="SELECT 1", dialect_family="sqlite", + server_version_constraint="supported-profile", + output_mode="materialized", row_semantics="one_to_one", + metadata_policy="declared", deterministic_order_json="[]", + output_schema_json='{"variables":[]}', + definition_hash=definition_hash, + published_at=datetime.now(UTC), + )) + connection.execute(insert(legacy.transformation_run).values( + transformation_run_id=run_id, + transformation_version_id=version_id, status="started", + executor_identity="legacy", correlation_id=run_id, + engine_name="sqlite", engine_version="legacy", + dialect_profile="sqlite", capability_snapshot_json="{}", + specification_commit="legacy", definition_hash=definition_hash, + parameters_hash="0" * 64, input_set_hash="1" * 64, + started_at=datetime(2000, 1, 1, tzinfo=UTC), + )) + connection.exec_driver_sql( + f'CREATE TABLE "{stage_name}" (sentinel INTEGER)' + ) + return engine, run_id, stage_name + + +def test_spec_recovery_cases_are_the_exact_runtime_contract(): + manifest = json.loads( + _workflow_conformance_manifest().read_text(encoding="utf-8") + ) + if "recovery_cases" not in manifest: + pytest.skip("Recovery cases land with the pending specification PR.") + invariants = [ + "no_derived_dataset", + "no_published_output", + "quarantined_staging_not_exposed", + "run_remains_started_while_staging_exists", + "reconciliation_required", + "remove_only_recorded_profile_owned_staging", + "success_forbidden", + ] + assert manifest["recovery_cases"] == [ + { + "id": "cleanup-failure-quarantines-staging", + "trigger": "cleanup_failed", + "initial_status": "started", + "staging_relation_key": + "sqlite:main.__openstatspec_staging_cleanup_case", + "event": {"code": "cleanup_failed", "phase": "cleanup"}, + "invariants": invariants, + "terminal_status_after_reconciliation": "failed", + }, + { + "id": "crash-leaves-quarantined-staging", + "trigger": "process_crash", + "initial_status": "started", + "staging_relation_key": + "sqlite:main.__openstatspec_staging_crash_case", + "event": None, + "invariants": invariants, + "terminal_status_after_reconciliation": "failed", + }, + ] + + +def test_cleanup_failure_quarantines_then_reconciles_exact_recorded_staging( + catalog, monkeypatch, +): + url, path, parent_id = catalog + registered = openstatspec.register_sql_transformation( + database_url=url, parent_dataset_id=parent_id, + query_sql="SELECT score FROM parent ORDER BY score ASC NULLS LAST", + columns=[{ + "name": "score", "storage_kind": "numeric", "source": "score", + }], + transformation_name="cleanup_failure", + ) + real_validate = workflow._validate_order_key + calls = 0 + + def fail_after_staging(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 2: + raise TransformationError( + "output_validation_failed", "Injected staging validation failure." + ) + return real_validate(*args, **kwargs) + + def leave_recorded_staging(_engine, _tables, run_id): + connection = sqlite3.connect(path) + key = connection.execute( + "select staging_relation_key from transformation_run " + "where transformation_run_id = ?", + (run_id,), + ).fetchone()[0] + stage_name = key.removeprefix("sqlite:main.") + connection.execute(f'create table "{stage_name}" (sentinel integer)') + connection.commit() + connection.close() + raise RuntimeError("injected cleanup failure") + + monkeypatch.setattr(workflow, "_validate_order_key", fail_after_staging) + monkeypatch.setattr( + workflow, "_cleanup_execution_relations", leave_recorded_staging + ) + with pytest.raises(TransformationError) as caught: + openstatspec.execute_sql_transformation( + database_url=url, parent_dataset_id=parent_id, + transformation_version_id=registered["transformation_version_id"], + ) + assert caught.value.code == "cleanup_failed" + + connection = sqlite3.connect(path) + run = connection.execute( + "select transformation_run_id, status, completed_at, " + "staging_relation_key from transformation_run" + ).fetchone() + assert run[1:3] == ("started", None) + staging_name = run[3].removeprefix("sqlite:main.") + assert connection.execute( + "select count(*) from sqlite_master where name = ?", (staging_name,) + ).fetchone() == (1,) + assert connection.execute( + "select event_ordinal, event_code, execution_phase " + "from transformation_event" + ).fetchall() == [(1, "cleanup_failed", "cleanup")] + assert connection.execute( + "select count(*) from derived_dataset" + ).fetchone() == (0,) + + reconciled = openstatspec.reconcile_sql_transformation_runs(database_url=url) + assert reconciled["transformation_run_ids"] == [run[0]] + assert connection.execute( + "select status from transformation_run" + ).fetchone() == ("failed",) + assert connection.execute( + "select count(*) from sqlite_master where name = ?", (staging_name,) + ).fetchone() == (0,) + assert connection.execute( + "select event_ordinal, event_code, execution_phase " + "from transformation_event order by event_ordinal" + ).fetchall() == [ + (1, "cleanup_failed", "cleanup"), + (2, "interrupted_run", "reconciliation"), + ] + + +def test_reconciliation_refuses_unowned_recorded_key(catalog): + url, path, parent_id = catalog + registered = openstatspec.register_sql_transformation( + database_url=url, parent_dataset_id=parent_id, + query_sql="SELECT score FROM parent ORDER BY score ASC NULLS LAST", + columns=[{ + "name": "score", "storage_kind": "numeric", "source": "score", + }], + transformation_name="ambiguous_recovery", + ) + connection = sqlite3.connect(path) + connection.execute("pragma foreign_keys = on") + unowned_id = "11111111-1111-4111-8111-111111111111" + _insert_started_run( + connection, registered, unowned_id, "sqlite:main.foreign_table" + ) + connection.execute('create table "foreign_table" (sentinel integer)') + connection.commit() + with pytest.raises(TransformationError) as unowned: + openstatspec.reconcile_sql_transformation_runs(database_url=url) + assert unowned.value.code == "reconciliation_ownership_unverified" + assert connection.execute( + "select status from transformation_run where transformation_run_id = ?", + (unowned_id,), + ).fetchone() == ("started",) + assert connection.execute( + "select count(*) from sqlite_master where name = 'foreign_table'" + ).fetchone() == (1,) + +def test_reconciliation_refuses_duplicate_recorded_keys(catalog): + url, path, parent_id = catalog + registered = openstatspec.register_sql_transformation( + database_url=url, parent_dataset_id=parent_id, + query_sql="SELECT score FROM parent ORDER BY score ASC NULLS LAST", + columns=[{ + "name": "score", "storage_kind": "numeric", "source": "score", + }], + transformation_name="duplicate_recovery", + ) + connection = sqlite3.connect(path) + connection.execute("pragma foreign_keys = on") + shared_name = "__openstatspec_staging_shared" + shared_key = f"sqlite:main.{shared_name}" + first_id = "22222222-2222-4222-8222-222222222222" + second_id = "33333333-3333-4333-8333-333333333333" + _insert_started_run(connection, registered, first_id, shared_key) + _insert_started_run(connection, registered, second_id, shared_key) + connection.execute(f'create table "{shared_name}" (sentinel integer)') + connection.commit() + with pytest.raises(TransformationError) as duplicate: + openstatspec.reconcile_sql_transformation_runs(database_url=url) + assert duplicate.value.code == "reconciliation_ownership_unverified" + assert connection.execute( + "select status from transformation_run " + "where transformation_run_id in (?, ?) order by transformation_run_id", + (first_id, second_id), + ).fetchall() == [("started",), ("started",)] + assert connection.execute( + "select count(*) from sqlite_master where name = ?", (shared_name,) + ).fetchone() == (1,) + + +def test_cleanup_never_drops_preexisting_deterministic_final_name( + catalog, monkeypatch, +): + url, path, parent_id = catalog + registered = openstatspec.register_sql_transformation( + database_url=url, parent_dataset_id=parent_id, + query_sql="SELECT score FROM parent ORDER BY score ASC NULLS LAST", + columns=[{ + "name": "score", "storage_kind": "numeric", "source": "score", + }], + transformation_name="final_name_collision", + ) + collision_id = UUID("44444444-4444-4444-8444-444444444444") + relation_name = "derived_" + collision_id.hex + connection = sqlite3.connect(path) + connection.execute( + f'create table "{relation_name}" (sentinel text)' + ) + connection.execute( + f'insert into "{relation_name}" values ("preserve me")' + ) + connection.commit() + monkeypatch.setattr(workflow, "uuid5", lambda *_args: collision_id) + + with pytest.raises(OperationalError): + openstatspec.execute_sql_transformation( + database_url=url, parent_dataset_id=parent_id, + transformation_version_id=registered["transformation_version_id"], + ) + assert connection.execute( + f'select sentinel from "{relation_name}"' + ).fetchall() == [("preserve me",)] + assert connection.execute( + "select status from transformation_run" + ).fetchone() == ("failed",) + assert connection.execute( + "select count(*) from derived_dataset" + ).fetchone() == (0,) + + +def test_pre_recovery_v2_migration_is_nullable_and_fail_closed(catalog): + url, path, _ = catalog + engine, run_id, stage_name = _create_legacy_workflow_catalog( + url, with_started_run=True + ) + current = workflow_catalog(MetaData()) + with engine.begin() as connection: + create_workflow_catalog(connection, current) + columns = { + row[1]: row for row in sqlite3.connect(path).execute( + "pragma table_info(transformation_run)" + ) + } + assert columns["staging_relation_key"][3] == 0 + connection = sqlite3.connect(path) + assert connection.execute( + "select staging_relation_key from transformation_run " + "where transformation_run_id = ?", + (run_id,), + ).fetchone() == (None,) + with pytest.raises(TransformationError) as caught: + openstatspec.reconcile_sql_transformation_runs(database_url=url) + assert caught.value.code == "reconciliation_ownership_unverified" + assert connection.execute( + "select status from transformation_run where transformation_run_id = ?", + (run_id,), + ).fetchone() == ("started",) + assert connection.execute( + "select count(*) from sqlite_master where name = ?", (stage_name,) + ).fetchone() == (1,) + + +def test_pre_recovery_v2_migration_rolls_back_on_trigger_failure( + catalog, monkeypatch, +): + url, path, _ = catalog + engine, _, _ = _create_legacy_workflow_catalog(url) + current = workflow_catalog(MetaData()) + + def fail_trigger_creation(_connection, _tables): + raise RuntimeError("injected trigger creation failure") + + monkeypatch.setattr( + workflow, "_create_workflow_triggers", fail_trigger_creation + ) + with pytest.raises(RuntimeError, match="injected trigger creation failure"): + with engine.begin() as connection: + create_workflow_catalog(connection, current) + connection = sqlite3.connect(path) + assert "staging_relation_key" not in { + row[1] for row in connection.execute( + "pragma table_info(transformation_run)" + ) + } + assert connection.execute( + "select count(*) from sqlite_master " + "where type = 'trigger' and name = 'oss_transformation_run_update_guard'" + ).fetchone() == (1,)