From f8fb3ec79d0868685f1b69610c588f61de97f5de Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Mon, 26 Oct 2026 15:47:04 +0400 Subject: [PATCH 1/8] Copy v1 policies saved under an older country package version translate_legacy_policy refused any v1 policy whose stored api_version differed from the country package version the Public API runs. When a POST /{country}/policy matches an existing row, the route returns that row with the version it was saved under, so every such dual-write copy failed as catalog_or_translation, and so did every older saved policy. The Public API already simulates an old policy with the running country package, so the copy now binds it to the running catalog the same way: a policy copies when every parameter path it sets exists in the running catalog and its periods and values are valid. The version comparison and the country_package_versions argument are gone. The new tests copy the shapes of the two production failures (US policy 96567 saved under 1.633.2, UK policy 97839 saved under 2.89.2) and replace the test that pinned the refusal. The Stage 10 runbook says which policies copy. Co-Authored-By: Claude Opus 5.5 --- ...2plan-1k-142-copy-reused-policies.fixed.md | 1 + docs/migration/stage-10-v2-policies.md | 10 +- .../services/v2/policies/services.py | 13 +- .../services/v2/policies/transformations.py | 14 +- .../integration/test_v2_policy_persistence.py | 5 - tests/unit/v2/test_policy_legacy_copy.py | 206 ++++++++++++++++++ .../unit/v2/test_policy_legacy_translation.py | 18 +- 7 files changed, 240 insertions(+), 27 deletions(-) create mode 100644 changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md create mode 100644 tests/unit/v2/test_policy_legacy_copy.py diff --git a/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md b/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md new file mode 100644 index 000000000..d77679ca8 --- /dev/null +++ b/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md @@ -0,0 +1 @@ +Dual-write now copies a v1 policy to v2 whatever country package version it was saved under, as long as every parameter it sets exists in the running catalog. Before, every policy saved under an older release was refused, which included each `POST /{country}/policy` that matched an existing policy. diff --git a/docs/migration/stage-10-v2-policies.md b/docs/migration/stage-10-v2-policies.md index 252bc940e..59f4e2488 100644 --- a/docs/migration/stage-10-v2-policies.md +++ b/docs/migration/stage-10-v2-policies.md @@ -129,7 +129,12 @@ failure, retained-event processing, idempotent retries, and source/destination rollback using the same MySQL and PostgreSQL engines as the deployed databases. Under this selection, a core-policy mutation commits Cloud SQL first and then -completes its policy transaction in Supabase. A saved-policy mutation commits +completes its policy transaction in Supabase. The copy binds the v1 policy to +the running catalog whatever country package version the v1 row was saved +under (its stored `api_version`), the same way the Public API simulates an old +policy with the running country package. A policy is copied when every +parameter path it sets exists in the running catalog and its periods and values +are valid. A saved-policy mutation commits its source row, incremented revision, and complete event in one Cloud SQL transaction. The same request then processes that source's pending events in revision order and records `processed_at` only after the corresponding @@ -177,7 +182,8 @@ The disposable and route-level tests verify that: - native policy and association create/read/list/update/delete operations use the initialized catalog and Supabase only; -- both newly created and existing v1 policies receive durable mappings; +- both newly created and existing v1 policies receive durable mappings, + including existing policies saved under an older country package version; - v1 saved-policy label updates change association `name`, while v1-only field updates change only the mapping fingerprint; - saved-policy event revisions are processed in ascending order, successful diff --git a/policyengine_api/services/v2/policies/services.py b/policyengine_api/services/v2/policies/services.py index 5c21c124d..e22a1d015 100644 --- a/policyengine_api/services/v2/policies/services.py +++ b/policyengine_api/services/v2/policies/services.py @@ -4,7 +4,7 @@ from typing import TYPE_CHECKING -from policyengine_api.constants import COUNTRY_PACKAGE_VERSIONS, POLICYENGINE_VERSION +from policyengine_api.constants import POLICYENGINE_VERSION from policyengine_api.services.v2.policies.database_connectors.creates import ( create_legacy_policy_mapping, create_parameter_values, @@ -53,7 +53,7 @@ ) if TYPE_CHECKING: - from collections.abc import Callable, Mapping + from collections.abc import Callable from uuid import UUID from sqlmodel import Session @@ -190,13 +190,13 @@ def mirror_legacy_policy_in_session( snapshot: LegacyPolicySnapshot, *, running_policyengine_version: str = POLICYENGINE_VERSION, - country_package_versions: Mapping[str, str] = COUNTRY_PACKAGE_VERSIONS, ) -> LegacyPolicyPersistenceResult: """Copy one committed v1 policy into v2 and map its v1 id to the result. - The policy is translated against the running catalog. Replaying the same v1 - policy is idempotent. Raise LegacyPolicyMappingIntegrityError if the v1 id is - already mapped with a different source hash or to a different v2 policy. + The policy is translated against the running catalog, whichever country + package version it was saved under. Replaying the same v1 policy is + idempotent. Raise LegacyPolicyMappingIntegrityError if the v1 id is already + mapped with a different source hash or to a different v2 policy. """ existing = read_legacy_policy_mapping( session, @@ -227,7 +227,6 @@ def mirror_legacy_policy_in_session( snapshot, selected=selected, parameters=parameters, - country_package_versions=country_package_versions, ) policy_result = create_resolved_policy(session, policy_input) if existing is not None: diff --git a/policyengine_api/services/v2/policies/transformations.py b/policyengine_api/services/v2/policies/transformations.py index 5fceabf51..1d6897338 100644 --- a/policyengine_api/services/v2/policies/transformations.py +++ b/policyengine_api/services/v2/policies/transformations.py @@ -259,18 +259,16 @@ def translate_legacy_policy( *, selected: SelectedCatalog, parameters: Mapping[str, Parameter], - country_package_versions: Mapping[str, str], ) -> ResolvedPolicyCreationInput: """Translate a v1 policy into policy content bound to the running catalog. - Raise LegacyPolicyTranslationError if the v1 policy was saved under another - country package version, names a parameter the catalog lacks, or has - invalid or conflicting periods. + The country package version the v1 policy was saved under is not compared: + the Public API simulates an old policy with the running package, so the copy + binds it to the running catalog the same way. + + Raise LegacyPolicyTranslationError if the v1 policy names a parameter the + catalog lacks, or has invalid or conflicting periods. """ - expected_api_version = country_package_versions.get(snapshot.country_id) - if expected_api_version is None or snapshot.api_version != expected_api_version: - msg = "legacy policy api_version does not match the running country package" - raise LegacyPolicyTranslationError(msg) policy_json = snapshot.policy_json if not isinstance(policy_json, dict): msg = "legacy policy_json must be an object" diff --git a/tests/integration/test_v2_policy_persistence.py b/tests/integration/test_v2_policy_persistence.py index b53538ddd..7597f5c3d 100644 --- a/tests/integration/test_v2_policy_persistence.py +++ b/tests/integration/test_v2_policy_persistence.py @@ -253,13 +253,11 @@ def test_legacy_policy_mapping_is_many_to_one_and_retry_safe() -> None: session, first, running_policyengine_version=version.version, - country_package_versions={"us": "1.0.0"}, ) second_result = mirror_legacy_policy_in_session( session, second, running_policyengine_version=version.version, - country_package_versions={"us": "1.0.0"}, ) with Session(engine) as session, session.begin(): @@ -267,7 +265,6 @@ def test_legacy_policy_mapping_is_many_to_one_and_retry_safe() -> None: session, first, running_policyengine_version="5.2.0", - country_package_versions={"us": "1.0.0"}, ) assert first_result.policy_id == second_result.policy_id == retry.policy_id @@ -310,7 +307,6 @@ def test_changed_hash_for_one_legacy_identity_rolls_back_without_mutation() -> N session, snapshot, running_policyengine_version=version.version, - country_package_versions={"us": "1.0.0"}, ) changed = snapshot.model_copy( @@ -328,7 +324,6 @@ def test_changed_hash_for_one_legacy_identity_rolls_back_without_mutation() -> N session, changed, running_policyengine_version="5.2.0", - country_package_versions={"us": "1.0.0"}, ) with Session(engine) as session: diff --git a/tests/unit/v2/test_policy_legacy_copy.py b/tests/unit/v2/test_policy_legacy_copy.py new file mode 100644 index 000000000..0f2945516 --- /dev/null +++ b/tests/unit/v2/test_policy_legacy_copy.py @@ -0,0 +1,206 @@ +"""Copy committed v1 policies into v2 with the running catalog. + +The catalog and mapping reads run against SQLite. The two inserts use +PostgreSQL's ON CONFLICT ON CONSTRAINT, so the tests record them instead. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from uuid import UUID + +import pytest +from sqlmodel import Session, create_engine + +from policyengine_api.data.v2.models import ( + V2_METADATA, + Parameter, + TaxBenefitModel, + TaxBenefitModelVersion, +) +from policyengine_api.services.v2.policies import services as policy_services +from policyengine_api.services.v2.policies.types import ( + LegacyPolicyPersistenceResult, + LegacyPolicySnapshot, + PolicyCreationResult, + ResolvedPolicyCreationInput, +) + +RUNNING_POLICYENGINE_VERSION = "6.7.5" +COPIED_POLICY_ID = UUID("00000000-0000-0000-0000-000000000042") +MAPPING_ID = UUID("00000000-0000-0000-0000-000000000043") + + +@dataclass +class RecordedCopyWrites: + """The policies and mappings a copy would have inserted, in order.""" + + created_policies: list[ResolvedPolicyCreationInput] = field(default_factory=list) + created_mappings: list[dict[str, object]] = field(default_factory=list) + + +@pytest.fixture +def recorded_writes(monkeypatch) -> RecordedCopyWrites: + writes = RecordedCopyWrites() + + def record_policy(_session, policy_input): + writes.created_policies.append(policy_input) + return PolicyCreationResult(policy_id=COPIED_POLICY_ID, created=True) + + def record_mapping(_session, **mapping_fields): + writes.created_mappings.append(mapping_fields) + return MAPPING_ID + + monkeypatch.setattr(policy_services, "create_resolved_policy", record_policy) + monkeypatch.setattr(policy_services, "create_legacy_policy_mapping", record_mapping) + return writes + + +@pytest.fixture +def catalog_session(): + engine = create_engine("sqlite://") + V2_METADATA.create_all(engine) + session = Session(engine) + try: + yield session + finally: + session.close() + engine.dispose() + + +def _add_running_catalog( + session: Session, *, country_id: str, parameter_names: list[str] +) -> tuple[TaxBenefitModelVersion, dict[str, Parameter]]: + model = TaxBenefitModel(name=f"policyengine-{country_id}") + version = TaxBenefitModelVersion( + model=model, + version=RUNNING_POLICYENGINE_VERSION, + current_law_id=1, + metadata_time_periods=[2026], + ) + parameters = { + name: Parameter(name=name, tax_benefit_model_version=version) + for name in parameter_names + } + session.add(version) + session.add_all(parameters.values()) + session.commit() + return version, parameters + + +def _snapshot( + *, + country_id: str, + legacy_policy_id: int, + api_version: str, + policy_json: dict[str, object], +) -> LegacyPolicySnapshot: + return LegacyPolicySnapshot( + country_id=country_id, + legacy_policy_id=legacy_policy_id, + label=None, + api_version=api_version, + policy_json=policy_json, + source_policy_hash=f"legacy-hash-{legacy_policy_id}", + ) + + +def _copy(session: Session, snapshot: LegacyPolicySnapshot): + return policy_services.mirror_legacy_policy_in_session( + session, + snapshot, + running_policyengine_version=RUNNING_POLICYENGINE_VERSION, + ) + + +@dataclass(frozen=True) +class ReusedProductionPolicy: + """The shape of a v1 policy whose production copy failed.""" + + country_id: str + legacy_policy_id: int + saved_api_version: str + parameter_name: str + value: float + + +# The two production copies that failed after dual-write was turned on: each +# POST matched an existing v1 row saved under a country package release many +# versions older than the one the Public API ran (us 2.2.1, uk 2.130.1). +PRODUCTION_REUSED_POLICIES = [ + pytest.param( + ReusedProductionPolicy("us", 96567, "1.633.2", "gov.dol.minimum_wage", 9), + id="us-96567-saved-under-1.633.2", + ), + pytest.param( + ReusedProductionPolicy("uk", 97839, "2.89.2", "gov.boe.base_rate", 0.05), + id="uk-97839-saved-under-2.89.2", + ), +] + + +@pytest.mark.parametrize("reused_policy", PRODUCTION_REUSED_POLICIES) +def test_policy_saved_under_an_older_country_package_is_copied( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, + reused_policy: ReusedProductionPolicy, +) -> None: + country_id = reused_policy.country_id + parameter_name = reused_policy.parameter_name + version, parameters = _add_running_catalog( + catalog_session, country_id=country_id, parameter_names=[parameter_name] + ) + snapshot = _snapshot( + country_id=country_id, + legacy_policy_id=reused_policy.legacy_policy_id, + api_version=reused_policy.saved_api_version, + policy_json={parameter_name: {"2026-01-01.2100-12-31": reused_policy.value}}, + ) + + result = _copy(catalog_session, snapshot) + + assert result == LegacyPolicyPersistenceResult( + policy_id=COPIED_POLICY_ID, + policy_created=True, + mapping_created=True, + ) + [copied_policy] = recorded_writes.created_policies + assert copied_policy.country_id == country_id + assert copied_policy.tax_benefit_model_version_id == version.id + assert copied_policy.policyengine_version == RUNNING_POLICYENGINE_VERSION + [copied_value] = copied_policy.parameter_values + assert copied_value.parameter_id == parameters[parameter_name].id + assert copied_value.value == reused_policy.value + assert copied_value.start_date == datetime(2026, 1, 1, tzinfo=UTC) + assert copied_value.end_date == datetime(2100, 12, 31, tzinfo=UTC) + assert recorded_writes.created_mappings == [ + { + "country_id": country_id, + "legacy_policy_id": reused_policy.legacy_policy_id, + "source_policy_hash": snapshot.source_policy_hash, + "policy_id": COPIED_POLICY_ID, + } + ] + + +def test_policy_with_a_country_package_the_api_does_not_run_is_copied( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, +) -> None: + # The copy binds content to the running catalog, as the Public API binds an + # old policy to the running country package when it simulates one. + _add_running_catalog( + catalog_session, country_id="us", parameter_names=["gov.example.rate"] + ) + snapshot = _snapshot( + country_id="us", + legacy_policy_id=501, + api_version="99.0.0", + policy_json={"gov.example.rate": {"2026": 0.2}}, + ) + + result = _copy(catalog_session, snapshot) + + assert result.mapping_created is True + assert len(recorded_writes.created_policies) == 1 diff --git a/tests/unit/v2/test_policy_legacy_translation.py b/tests/unit/v2/test_policy_legacy_translation.py index d5cf88753..37e008a1a 100644 --- a/tests/unit/v2/test_policy_legacy_translation.py +++ b/tests/unit/v2/test_policy_legacy_translation.py @@ -98,7 +98,6 @@ def _translate( snapshot, selected=selected, parameters=parameters, - country_package_versions={"us": "1.0.0"}, ) @@ -262,11 +261,20 @@ def test_missing_paths_malformed_periods_and_conflicts_fail( engine.dispose() -def test_country_package_version_must_match_running_release() -> None: - engine, session, _model, _version, _first, _second = _session_and_catalog() +@pytest.mark.parametrize("saved_api_version", ["0.9.0", "1.633.2", "99.0.0"]) +def test_policy_saved_under_another_country_package_translates_to_running_catalog( + saved_api_version: str, +) -> None: + engine, session, _model, version, first, second = _session_and_catalog() try: - with pytest.raises(LegacyPolicyTranslationError, match="api_version"): - _translate(session, _snapshot(api_version="0.9.0")) + translated = _translate(session, _snapshot(api_version=saved_api_version)) + + assert translated.tax_benefit_model_version_id == version.id + assert {value.parameter_id for value in translated.parameter_values} == { + first.id, + second.id, + } + assert translated == _translate(session, _snapshot()) finally: session.close() engine.dispose() From 50e0a4a6aadc1b6fe705260a6158e4b0b658f826 Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Mon, 26 Oct 2026 15:50:10 +0400 Subject: [PATCH 2/8] Reuse a v1 policy's existing v2 copy instead of translating it again mirror_legacy_policy_in_session translated the v1 policy again even when legacy_policy_mappings already held its copy, then checked the result against the mapping. After a country package upgrade the translation could fail as catalog_or_translation, and after a PolicyEngine.py upgrade it bound the content to a new model version, so the content hash changed and the check failed as integrity. Writing the same policy again therefore failed after every upgrade. v1 policies never change, so a mapping whose source hash matches is the copy whatever catalog is running. The copy now returns it at once, with policy_created and mapping_created false. A different source hash is still an integrity error. The MySQL integration test for a historical first-write mapping now expects that mapping to be reused unchanged rather than refused. Co-Authored-By: Claude Opus 5.5 --- ...2plan-1k-142-copy-reused-policies.fixed.md | 2 +- docs/migration/stage-10-v2-policies.md | 4 +- .../services/v2/policies/services.py | 28 +++--- .../test_mysql_policy_dual_write.py | 15 ++-- tests/unit/v2/test_policy_legacy_copy.py | 85 +++++++++++++++++++ 5 files changed, 108 insertions(+), 26 deletions(-) diff --git a/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md b/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md index d77679ca8..5c14f2b01 100644 --- a/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md +++ b/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md @@ -1 +1 @@ -Dual-write now copies a v1 policy to v2 whatever country package version it was saved under, as long as every parameter it sets exists in the running catalog. Before, every policy saved under an older release was refused, which included each `POST /{country}/policy` that matched an existing policy. +Dual-write now copies a v1 policy to v2 whatever country package version it was saved under, as long as every parameter it sets exists in the running catalog. Before, every policy saved under an older release was refused, which included each `POST /{country}/policy` that matched an existing policy. A v1 policy that already has a v2 copy now returns that copy without being translated again, so writing it again after a PolicyEngine.py or country package upgrade no longer fails. diff --git a/docs/migration/stage-10-v2-policies.md b/docs/migration/stage-10-v2-policies.md index 59f4e2488..6993b197f 100644 --- a/docs/migration/stage-10-v2-policies.md +++ b/docs/migration/stage-10-v2-policies.md @@ -152,7 +152,9 @@ its `failure_category`, and the route returns its normal v1 response. A saved-policy event that failed stays pending, and the next mutation of the same saved policy replays it, with earlier retained events first. A core policy whose copy failed is copied again the next time any request writes the same -policy. Destination revision and fingerprint checks make a replay safe if +policy. A core policy that already has a copy returns that copy without being +translated again, so writing it again after a PolicyEngine.py or country +package upgrade does not fail. Destination revision and fingerprint checks make a replay safe if Supabase committed but the Cloud SQL processing marker did not. There is no background processor or reconciliation process. diff --git a/policyengine_api/services/v2/policies/services.py b/policyengine_api/services/v2/policies/services.py index e22a1d015..b3ebd6f1e 100644 --- a/policyengine_api/services/v2/policies/services.py +++ b/policyengine_api/services/v2/policies/services.py @@ -194,9 +194,12 @@ def mirror_legacy_policy_in_session( """Copy one committed v1 policy into v2 and map its v1 id to the result. The policy is translated against the running catalog, whichever country - package version it was saved under. Replaying the same v1 policy is - idempotent. Raise LegacyPolicyMappingIntegrityError if the v1 id is already - mapped with a different source hash or to a different v2 policy. + package version it was saved under. A v1 policy that already has a v2 copy + returns that copy without being translated again, so writing it again after + a PolicyEngine.py or country package upgrade still succeeds. Raise + LegacyPolicyMappingIntegrityError if the v1 id is already mapped with a + different source hash, or a concurrent copy mapped it to a different v2 + policy. """ existing = read_legacy_policy_mapping( session, @@ -205,9 +208,16 @@ def mirror_legacy_policy_in_session( lock=True, ) if existing is not None: + # v1 policies never change, so the same source hash means the mapped + # v2 policy is still the copy, whichever catalog is running now. verify_legacy_policy_mapping( existing, source_policy_hash=snapshot.source_policy_hash ) + return LegacyPolicyPersistenceResult( + policy_id=existing.policy_id, + policy_created=False, + mapping_created=False, + ) selected = read_policy_catalog( session, @@ -229,18 +239,6 @@ def mirror_legacy_policy_in_session( parameters=parameters, ) policy_result = create_resolved_policy(session, policy_input) - if existing is not None: - verify_legacy_policy_mapping( - existing, - source_policy_hash=snapshot.source_policy_hash, - expected_policy_id=policy_result.policy_id, - ) - return LegacyPolicyPersistenceResult( - policy_id=existing.policy_id, - policy_created=False, - mapping_created=False, - ) - mapping_id = create_legacy_policy_mapping( session, country_id=snapshot.country_id, diff --git a/tests/integration/test_mysql_policy_dual_write.py b/tests/integration/test_mysql_policy_dual_write.py index e3a778192..b7e678d05 100644 --- a/tests/integration/test_mysql_policy_dual_write.py +++ b/tests/integration/test_mysql_policy_dual_write.py @@ -48,9 +48,6 @@ NativePolicyCreationInput, PolicyParameterValueInput, ) -from policyengine_api.services.v2.policies.validators import ( - LegacyPolicyMappingIntegrityError, -) from policyengine_api.utils import hash_object INPUT_VALUE = 0.04 + 3 / 1_000_000 @@ -379,7 +376,7 @@ def create_native(value): assert postgres_destination.counts() == (2, 2, 1) -def test_historical_pre_storage_mapping_is_rejected_without_repair( +def test_historical_pre_storage_mapping_is_reused_without_repair( mysql_source, postgres_destination, record_property ) -> None: creation = mysql_source.create() @@ -397,13 +394,13 @@ def test_historical_pre_storage_mapping_is_rejected_without_repair( assert retry.snapshot.legacy_policy_id == historical.legacy_policy_id assert retry.snapshot.policy_json != historical.policy_json - with pytest.raises(PolicyMirrorUnavailableError) as failure: - postgres_destination.mirror(retry) - assert isinstance(failure.value.__cause__, LegacyPolicyMappingIntegrityError) - assert "translated immutable content" in str(failure.value.__cause__) + # A v1 policy that already has a copy is not translated again: the retry + # returns the existing copy and neither repairs nor duplicates it. + reused = postgres_destination.mirror(retry) + assert reused.policy_id == original.policy_id + assert (reused.policy_created, reused.mapping_created) == (False, False) record_property("mysql_legacy_policy_id", creation.policy_id) record_property("postgres_historical_policy_id", str(original.policy_id)) - record_property("retry_error", type(failure.value.__cause__).__name__) assert postgres_destination.counts() == (1, 1, 1) with postgres_destination.sessions() as session: mapping = session.scalar( diff --git a/tests/unit/v2/test_policy_legacy_copy.py b/tests/unit/v2/test_policy_legacy_copy.py index 0f2945516..43cfc12c4 100644 --- a/tests/unit/v2/test_policy_legacy_copy.py +++ b/tests/unit/v2/test_policy_legacy_copy.py @@ -15,6 +15,7 @@ from policyengine_api.data.v2.models import ( V2_METADATA, + LegacyPolicyMapping, Parameter, TaxBenefitModel, TaxBenefitModelVersion, @@ -26,9 +27,13 @@ PolicyCreationResult, ResolvedPolicyCreationInput, ) +from policyengine_api.services.v2.policies.validators import ( + LegacyPolicyMappingIntegrityError, +) RUNNING_POLICYENGINE_VERSION = "6.7.5" COPIED_POLICY_ID = UUID("00000000-0000-0000-0000-000000000042") +EARLIER_COPY_POLICY_ID = UUID("00000000-0000-0000-0000-000000000041") MAPPING_ID = UUID("00000000-0000-0000-0000-000000000043") @@ -204,3 +209,83 @@ def test_policy_with_a_country_package_the_api_does_not_run_is_copied( assert result.mapping_created is True assert len(recorded_writes.created_policies) == 1 + + +def _add_earlier_copy( + session: Session, snapshot: LegacyPolicySnapshot, *, source_policy_hash: str +) -> None: + session.add( + LegacyPolicyMapping( + country_id=snapshot.country_id, + legacy_policy_id=snapshot.legacy_policy_id, + policy_id=EARLIER_COPY_POLICY_ID, + source_policy_hash=source_policy_hash, + ) + ) + session.commit() + + +def _reused_snapshot() -> LegacyPolicySnapshot: + return _snapshot( + country_id="us", + legacy_policy_id=96567, + api_version="1.633.2", + policy_json={"gov.dol.minimum_wage": {"2026-01-01.2100-12-31": 9}}, + ) + + +def test_earlier_copy_is_reused_when_the_running_catalog_is_not_published( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, +) -> None: + # After a PolicyEngine.py upgrade, the new release's catalog may not be + # published yet. Writing the same v1 policy again still succeeds. + snapshot = _reused_snapshot() + _add_earlier_copy( + catalog_session, snapshot, source_policy_hash=snapshot.source_policy_hash + ) + + result = _copy(catalog_session, snapshot) + + assert result == LegacyPolicyPersistenceResult( + policy_id=EARLIER_COPY_POLICY_ID, + policy_created=False, + mapping_created=False, + ) + assert recorded_writes == RecordedCopyWrites() + + +def test_earlier_copy_is_reused_when_the_running_catalog_lacks_its_parameter( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, +) -> None: + # A later release renamed or removed the parameter. The v2 copy made + # before that keeps its own catalog version, so it is still the copy. + _add_running_catalog( + catalog_session, country_id="us", parameter_names=["gov.example.rate"] + ) + snapshot = _reused_snapshot() + _add_earlier_copy( + catalog_session, snapshot, source_policy_hash=snapshot.source_policy_hash + ) + + result = _copy(catalog_session, snapshot) + + assert result.policy_id == EARLIER_COPY_POLICY_ID + assert (result.policy_created, result.mapping_created) == (False, False) + assert recorded_writes == RecordedCopyWrites() + + +def test_earlier_copy_with_another_source_hash_is_an_integrity_error( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, +) -> None: + snapshot = _reused_snapshot() + _add_earlier_copy( + catalog_session, snapshot, source_policy_hash="different-legacy-hash" + ) + + with pytest.raises(LegacyPolicyMappingIntegrityError, match="source hash"): + _copy(catalog_session, snapshot) + + assert recorded_writes == RecordedCopyWrites() From 4403269ea897de3937b1bd2861808b1f2d7b2a09 Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Mon, 26 Oct 2026 15:52:38 +0400 Subject: [PATCH 3/8] Give each untranslatable v1 policy a fixed reason, and share the parameter check LegacyPolicyTranslationError now carries a reason from a fixed set: missing_parameters, invalid_parameter_values, empty_parameter_values, invalid_period or conflicting_values. The message is still never logged. parameters_missing_from_model lists, once each and sorted, the parameter paths a policy sets that a model does not have. The translation uses it for its missing-parameter check, so later code that reports a policy's missing parameters can apply exactly the same rule. A parameter whose period object is empty ({}) sets no value. It used to pass translation and fail later in validate_policy_catalog with a PolicyCatalogValidationError that named the wrong cause. It is still refused, now as empty_parameter_values. Co-Authored-By: Claude Opus 5.5 --- .../services/v2/policies/transformations.py | 47 +++++++---- .../services/v2/policies/validators.py | 43 +++++++++- .../routes/test_policy_dual_write_routes.py | 4 +- .../unit/v2/test_policy_legacy_translation.py | 78 +++++++++++++++---- 4 files changed, 143 insertions(+), 29 deletions(-) diff --git a/policyengine_api/services/v2/policies/transformations.py b/policyengine_api/services/v2/policies/transformations.py index 1d6897338..d4120ed01 100644 --- a/policyengine_api/services/v2/policies/transformations.py +++ b/policyengine_api/services/v2/policies/transformations.py @@ -24,6 +24,8 @@ ) from policyengine_api.services.v2.policies.validators import ( LegacyPolicyTranslationError, + LegacyPolicyTranslationReason, + parameters_missing_from_model, validate_policy_catalog, ) @@ -221,7 +223,9 @@ def _utc_midnight(value: str) -> datetime: parsed = date.fromisoformat(value) except ValueError as error: msg = f"legacy period {value!r} is invalid" - raise LegacyPolicyTranslationError(msg) from error + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.INVALID_PERIOD + ) from error return datetime.combine(parsed, time.min, tzinfo=UTC) @@ -231,14 +235,15 @@ def parse_legacy_period(value: str) -> tuple[datetime, datetime]: Accept a ``start.end`` ISO date range or a PolicyEngine period such as ``2024``. Raise LegacyPolicyTranslationError for anything else. """ + invalid_period = LegacyPolicyTranslationReason.INVALID_PERIOD if not value or value != value.strip(): msg = "legacy period must be non-empty" - raise LegacyPolicyTranslationError(msg) + raise LegacyPolicyTranslationError(msg, reason=invalid_period) if "." in value: parts = value.split(".") if len(parts) != LEGACY_DATE_RANGE_PART_COUNT: msg = f"legacy period {value!r} must contain one date range" - raise LegacyPolicyTranslationError(msg) + raise LegacyPolicyTranslationError(msg, reason=invalid_period) start_date, end_date = (_utc_midnight(item) for item in parts) else: try: @@ -247,10 +252,10 @@ def parse_legacy_period(value: str) -> tuple[datetime, datetime]: end_date = _utc_midnight(str(parsed.stop)) except (TypeError, ValueError) as error: msg = f"legacy period {value!r} is invalid" - raise LegacyPolicyTranslationError(msg) from error + raise LegacyPolicyTranslationError(msg, reason=invalid_period) from error if end_date < start_date: msg = f"legacy period {value!r} ends before it starts" - raise LegacyPolicyTranslationError(msg) + raise LegacyPolicyTranslationError(msg, reason=invalid_period) return start_date, end_date @@ -266,30 +271,42 @@ def translate_legacy_policy( the Public API simulates an old policy with the running package, so the copy binds it to the running catalog the same way. - Raise LegacyPolicyTranslationError if the v1 policy names a parameter the - catalog lacks, or has invalid or conflicting periods. + ``parameters`` holds the running catalog's parameters with the policy's + parameter paths, keyed by path. Raise LegacyPolicyTranslationError, with a + fixed reason, if the v1 policy sets a parameter the catalog lacks, sets no + value for a parameter, or has invalid or conflicting periods. """ policy_json = snapshot.policy_json if not isinstance(policy_json, dict): msg = "legacy policy_json must be an object" raise TypeError(msg) - parameter_names = set(policy_json) - if set(parameters) != parameter_names: + if parameters_missing_from_model(policy_json, parameters): msg = "every legacy parameter path must exist in the running catalog" - raise LegacyPolicyTranslationError(msg) + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.MISSING_PARAMETERS + ) parameter_values: list[PolicyParameterValueInput] = [] - for parameter_name in sorted(parameter_names): + for parameter_name in sorted(policy_json): period_values = policy_json[parameter_name] if type(period_values) is not dict: msg = ( f"legacy parameter {parameter_name!r} must contain period/value entries" ) - raise LegacyPolicyTranslationError(msg) + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.INVALID_PARAMETER_VALUES + ) + if not period_values: + msg = f"legacy parameter {parameter_name!r} sets no values" + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.EMPTY_PARAMETER_VALUES + ) for period_name, value in sorted(period_values.items()): if type(period_name) is not str: msg = f"legacy parameter {parameter_name!r} has a non-string period" - raise LegacyPolicyTranslationError(msg) + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.INVALID_PERIOD + ) start_date, end_date = parse_legacy_period(period_name) parameter_values.append( PolicyParameterValueInput( @@ -307,7 +324,9 @@ def translate_legacy_policy( ) except ValueError as error: msg = "legacy parameter periods or values conflict" - raise LegacyPolicyTranslationError(msg) from error + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.CONFLICTING_VALUES + ) from error return validate_policy_catalog( policy_input, selected=selected, diff --git a/policyengine_api/services/v2/policies/validators.py b/policyengine_api/services/v2/policies/validators.py index 4302d07da..7bf2cf5d5 100644 --- a/policyengine_api/services/v2/policies/validators.py +++ b/policyengine_api/services/v2/policies/validators.py @@ -4,9 +4,11 @@ import math from datetime import UTC, datetime +from enum import StrEnum from typing import TYPE_CHECKING, Any if TYPE_CHECKING: + from collections.abc import Container, Iterable from uuid import UUID from policyengine_api.data.v2.catalog.catalog_selection import SelectedCatalog @@ -41,8 +43,47 @@ class LegacyPolicyMappingIntegrityError(RuntimeError): """Raised when one immutable v1 identity maps inconsistently.""" +class LegacyPolicyTranslationReason(StrEnum): + """Why committed v1 content cannot be translated, as copy failure logs name it.""" + + MISSING_PARAMETERS = "missing_parameters" + INVALID_PARAMETER_VALUES = "invalid_parameter_values" + EMPTY_PARAMETER_VALUES = "empty_parameter_values" + INVALID_PERIOD = "invalid_period" + CONFLICTING_VALUES = "conflicting_values" + + class LegacyPolicyTranslationError(ValueError): - """Raised when committed v1 content cannot be interpreted exactly.""" + """Raised when committed v1 content cannot be interpreted exactly. + + ``reason`` is a fixed code that copy failure logs record; the message is + never logged. + """ + + def __init__(self, message: str, *, reason: LegacyPolicyTranslationReason) -> None: + """Keep the message for debugging and the fixed reason for logs.""" + super().__init__(message) + self.reason = reason + + +def parameters_missing_from_model( + policy_parameter_names: Iterable[str], + model_parameter_names: Container[str], +) -> tuple[str, ...]: + """Return the parameter paths a policy sets that the model does not have. + + Each absent path appears once, in sorted order; an empty tuple means every + path exists. The v2 copy of a v1 policy is refused when it is not empty. + """ + return tuple( + sorted( + { + name + for name in policy_parameter_names + if name not in model_parameter_names + } + ) + ) def require_json_value( diff --git a/tests/unit/routes/test_policy_dual_write_routes.py b/tests/unit/routes/test_policy_dual_write_routes.py index d644a1d29..397b741da 100644 --- a/tests/unit/routes/test_policy_dual_write_routes.py +++ b/tests/unit/routes/test_policy_dual_write_routes.py @@ -16,6 +16,7 @@ from policyengine_api.services.v2.policies.types import LegacyPolicySnapshot from policyengine_api.services.v2.policies.validators import ( LegacyPolicyTranslationError, + LegacyPolicyTranslationReason, ) @@ -224,7 +225,8 @@ def test_destination_failure_returns_the_stored_policy_and_logs_an_error( monkeypatch.setenv("DB_WRITE_POLICY", "dual_write") failing_destination = MagicMock() failing_destination.mirror_legacy_policy.side_effect = LegacyPolicyTranslationError( - "private parameter value" + "private parameter value", + reason=LegacyPolicyTranslationReason.MISSING_PARAMETERS, ) with ( patch( diff --git a/tests/unit/v2/test_policy_legacy_translation.py b/tests/unit/v2/test_policy_legacy_translation.py index 37e008a1a..7540b3349 100644 --- a/tests/unit/v2/test_policy_legacy_translation.py +++ b/tests/unit/v2/test_policy_legacy_translation.py @@ -33,6 +33,8 @@ from policyengine_api.services.v2.policies.types import LegacyPolicySnapshot from policyengine_api.services.v2.policies.validators import ( LegacyPolicyTranslationError, + LegacyPolicyTranslationReason, + parameters_missing_from_model, ) from tests.fixtures.local_v1_database import create_test_v1_schema @@ -236,31 +238,80 @@ def unavailable_mirror(): @pytest.mark.parametrize( - "policy_json", + ("policy_json", "reason"), [ - {"gov.missing": {"2026": 1}}, - {"gov.example.rate": 1}, - {"gov.example.rate": {"not-a-period": 1}}, - { - "gov.example.rate": { - "2026": 1, - "2026-01-01.2026-12-31": 2, - } - }, + pytest.param( + {"gov.missing": {"2026": 1}}, + LegacyPolicyTranslationReason.MISSING_PARAMETERS, + id="unknown-path", + ), + pytest.param( + {"gov.example.rate": {"2026": 1}, "gov.removed": {"2026": 2}}, + LegacyPolicyTranslationReason.MISSING_PARAMETERS, + id="one-of-two-paths-unknown", + ), + pytest.param( + {"gov.example.rate": 1}, + LegacyPolicyTranslationReason.INVALID_PARAMETER_VALUES, + id="value-without-period", + ), + pytest.param( + {"gov.example.rate": {}}, + LegacyPolicyTranslationReason.EMPTY_PARAMETER_VALUES, + id="empty-period-object", + ), + pytest.param( + {"gov.example.rate": {"not-a-period": 1}}, + LegacyPolicyTranslationReason.INVALID_PERIOD, + id="unparseable-period", + ), + pytest.param( + {"gov.example.rate": {".": 1}}, + LegacyPolicyTranslationReason.INVALID_PERIOD, + id="dot-period", + ), + pytest.param( + { + "gov.example.rate": { + "2026": 1, + "2026-01-01.2026-12-31": 2, + } + }, + LegacyPolicyTranslationReason.CONFLICTING_VALUES, + id="two-values-same-start", + ), ], ) -def test_missing_paths_malformed_periods_and_conflicts_fail( +def test_untranslatable_policy_fails_with_a_fixed_reason( policy_json: dict[str, object], + reason: LegacyPolicyTranslationReason, ) -> None: engine, session, _model, _version, _first, _second = _session_and_catalog() try: - with pytest.raises(LegacyPolicyTranslationError): + with pytest.raises(LegacyPolicyTranslationError) as failure: _translate(session, _snapshot(policy_json=policy_json)) + assert failure.value.reason is reason finally: session.close() engine.dispose() +def test_parameters_missing_from_model_lists_each_absent_name_once_in_order() -> None: + model_parameter_names = {"gov.example.rate", "gov.example.amount"} + + assert parameters_missing_from_model( + ["gov.removed.b", "gov.example.rate", "gov.removed.a", "gov.removed.b"], + model_parameter_names, + ) == ("gov.removed.a", "gov.removed.b") + assert parameters_missing_from_model([], model_parameter_names) == () + assert ( + parameters_missing_from_model( + {"gov.example.amount": {"2026": 1}}, model_parameter_names + ) + == () + ) + + @pytest.mark.parametrize("saved_api_version", ["0.9.0", "1.633.2", "99.0.0"]) def test_policy_saved_under_another_country_package_translates_to_running_catalog( saved_api_version: str, @@ -295,8 +346,9 @@ def test_snapshot_rejects_incomplete_or_non_json_committed_fields(changes) -> No def test_reverse_legacy_range_is_rejected() -> None: - with pytest.raises(LegacyPolicyTranslationError, match="ends before"): + with pytest.raises(LegacyPolicyTranslationError, match="ends before") as failure: parse_legacy_period("2026-12-31.2026-01-01") + assert failure.value.reason is LegacyPolicyTranslationReason.INVALID_PERIOD def test_unknown_policyengine_version_never_falls_back() -> None: From d2984699f32cf8f2df99101da6b95f88404e0b71 Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Mon, 26 Oct 2026 15:58:10 +0400 Subject: [PATCH 4/8] Log a fixed failure_reason with each failed policy copy Every failed copy record logged only a broad failure_category, so all three production failures since dual-write was turned on read catalog_or_translation with no way to tell their causes apart from the logs. Each v1_policy_mirror_operations and v1_user_policy_mirror_operations error record now also carries failure_reason, a code taken only from the exception type: the translation's own reason (missing_parameters, invalid_period and so on), catalog_absent or catalog_query_failed, mapping_conflict, database_error, unexpected, and the rest listed in the Stage 10 runbook. A saved policy whose reform policy cannot be loaded logs missing_reform_policy. The records written by the route when no copy was attempted repeat their category as the reason. No policy content or exception text is logged; success records log failure_reason null. Co-Authored-By: Claude Opus 5.5 --- ...2plan-1k-142-copy-reused-policies.fixed.md | 2 +- docs/migration/stage-10-v2-policies.md | 14 +- policyengine_api/routes/policy_routes.py | 2 + policyengine_api/services/policy_mirroring.py | 69 ++++++++++ .../services/user_policy_mirroring.py | 22 +++- .../routes/test_policy_dual_write_routes.py | 2 + .../test_user_policy_dual_write_routes.py | 2 + .../unit/services/test_mirror_log_payloads.py | 5 + tests/unit/services/test_policy_mirroring.py | 121 ++++++++++++++++++ .../services/test_user_policy_mirroring.py | 75 ++++++++++- 10 files changed, 306 insertions(+), 8 deletions(-) diff --git a/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md b/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md index 5c14f2b01..31df2fc62 100644 --- a/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md +++ b/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md @@ -1 +1 @@ -Dual-write now copies a v1 policy to v2 whatever country package version it was saved under, as long as every parameter it sets exists in the running catalog. Before, every policy saved under an older release was refused, which included each `POST /{country}/policy` that matched an existing policy. A v1 policy that already has a v2 copy now returns that copy without being translated again, so writing it again after a PolicyEngine.py or country package upgrade no longer fails. +Dual-write now copies a v1 policy to v2 whatever country package version it was saved under, as long as every parameter it sets exists in the running catalog. Before, every policy saved under an older release was refused, which included each `POST /{country}/policy` that matched an existing policy. A v1 policy that already has a v2 copy now returns that copy without being translated again, so writing it again after a PolicyEngine.py or country package upgrade no longer fails. Each failed policy or saved-policy copy now logs a fixed `failure_reason` next to its `failure_category`, such as `missing_parameters`. diff --git a/docs/migration/stage-10-v2-policies.md b/docs/migration/stage-10-v2-policies.md index 6993b197f..ddef0f6a2 100644 --- a/docs/migration/stage-10-v2-policies.md +++ b/docs/migration/stage-10-v2-policies.md @@ -175,7 +175,19 @@ functions do not implement separate preflight behavior. Monitor structured events with metric names `v1_policy_mirror_operations` and `v1_user_policy_mirror_operations`. Alert on `outcome=error`, grouped by -`failure_category` and country. Events include the legacy integer ID, +`failure_category` and country. Each failed copy also records a fixed +`failure_reason`: + +| `failure_category` | `failure_reason` | +| --- | --- | +| `catalog_or_translation` | `missing_parameters` (the policy sets a parameter the running catalog lacks), `invalid_parameter_values`, `empty_parameter_values`, `invalid_period`, `conflicting_values`, `catalog_absent` (no catalog for the running PolicyEngine.py version), `catalog_query_failed`, `catalog_version_not_found`, `catalog_mismatch` | +| `integrity` | `mapping_conflict`, `saved_policy_mapping_conflict`, `content_hash_collision`, `stored_policy_integrity` | +| `database` | `database_error` | +| `configuration` | `configuration` | +| `unexpected` | `missing_reform_policy` (a saved policy's v1 reform could not be loaded), `unexpected` | +| `missing_mirror_snapshot`, `missing_mirror_revision` | the same value as the category | + +Events include the legacy integer ID, destination UUID when committed, attempted and completed database sources, and duration. They do not include policy JSON, presentation data, database URLs, or credentials. diff --git a/policyengine_api/routes/policy_routes.py b/policyengine_api/routes/policy_routes.py index 1eef717b2..e474341b0 100644 --- a/policyengine_api/routes/policy_routes.py +++ b/policyengine_api/routes/policy_routes.py @@ -90,6 +90,8 @@ def _log_uncopied_v1_write( "request_id": current_request_id(), "outcome": "error", "failure_category": failure_category, + # These failures have no finer cause than their category. + "failure_reason": failure_category, }, severity="ERROR", ) diff --git a/policyengine_api/services/policy_mirroring.py b/policyengine_api/services/policy_mirroring.py index 50bb585ef..dbe0507bb 100644 --- a/policyengine_api/services/policy_mirroring.py +++ b/policyengine_api/services/policy_mirroring.py @@ -5,6 +5,7 @@ import contextlib import time from dataclasses import dataclass +from enum import StrEnum from typing import TYPE_CHECKING, Protocol from sqlalchemy.exc import SQLAlchemyError @@ -18,6 +19,7 @@ from policyengine_api.services.v2.policies.validators import ( LegacyPolicyMappingIntegrityError, LegacyPolicyTranslationError, + LegacyPolicyTranslationReason, PolicyCatalogValidationError, PolicyContentHashCollisionError, PolicyCreationIntegrityError, @@ -86,6 +88,70 @@ def _failure_category(error: Exception) -> str: return "unexpected" +class PolicyCopyFailureReason(StrEnum): + """Fixed reason codes a failed policy or saved-policy copy logs. + + A translation failure logs its LegacyPolicyTranslationReason value instead, + such as ``missing_parameters``. Together the two sets are every value of a + copy record's ``failure_reason``. + """ + + CONFIGURATION = "configuration" + CATALOG_QUERY_FAILED = "catalog_query_failed" + CATALOG_ABSENT = "catalog_absent" + CATALOG_VERSION_NOT_FOUND = "catalog_version_not_found" + CATALOG_MISMATCH = "catalog_mismatch" + MAPPING_CONFLICT = "mapping_conflict" + SAVED_POLICY_MAPPING_CONFLICT = "saved_policy_mapping_conflict" + CONTENT_HASH_COLLISION = "content_hash_collision" + STORED_POLICY_INTEGRITY = "stored_policy_integrity" + MISSING_REFORM_POLICY = "missing_reform_policy" + DATABASE_ERROR = "database_error" + UNEXPECTED = "unexpected" + + +def policy_copy_failure_reason(error: Exception) -> str: + """Return the fixed reason code a failed policy copy logs for this error. + + The code comes only from the error's type, and for a translation error its + fixed reason, so a log record never holds policy content or error text. + """ + return _failure_reason_code(error).value + + +# Checked in order, so the hash collision is matched before its integrity base +# class. +_FAILURE_REASONS_BY_ERROR_TYPE: tuple[ + tuple[type[Exception], PolicyCopyFailureReason], ... +] = ( + (V2ConfigurationError, PolicyCopyFailureReason.CONFIGURATION), + ( + MetadataCatalogVersionNotFoundError, + PolicyCopyFailureReason.CATALOG_VERSION_NOT_FOUND, + ), + (PolicyCatalogValidationError, PolicyCopyFailureReason.CATALOG_MISMATCH), + (LegacyPolicyMappingIntegrityError, PolicyCopyFailureReason.MAPPING_CONFLICT), + (PolicyContentHashCollisionError, PolicyCopyFailureReason.CONTENT_HASH_COLLISION), + (PolicyCreationIntegrityError, PolicyCopyFailureReason.STORED_POLICY_INTEGRITY), + (SQLAlchemyError, PolicyCopyFailureReason.DATABASE_ERROR), +) + + +def _failure_reason_code( + error: Exception, +) -> PolicyCopyFailureReason | LegacyPolicyTranslationReason: + if isinstance(error, LegacyPolicyTranslationError): + return error.reason + if isinstance(error, MetadataCatalogUnavailableError): + if isinstance(error.__cause__, SQLAlchemyError): + return PolicyCopyFailureReason.CATALOG_QUERY_FAILED + return PolicyCopyFailureReason.CATALOG_ABSENT + for error_type, reason in _FAILURE_REASONS_BY_ERROR_TYPE: + if isinstance(error, error_type): + return reason + return PolicyCopyFailureReason.UNEXPECTED + + @dataclass(frozen=True) class _MirrorOutcome: """How one immediate policy mirror ended, as its metric record reports it.""" @@ -94,6 +160,7 @@ class _MirrorOutcome: actual_write_sources: list[str] destination_policy_id: object | None = None failure_category: str | None = None + failure_reason: str | None = None policy_created: bool | None = None mapping_created: bool | None = None @@ -124,6 +191,7 @@ def _log_mirror_event( ), "outcome": mirror_outcome.outcome, "failure_category": mirror_outcome.failure_category, + "failure_reason": mirror_outcome.failure_reason, "policy_created": mirror_outcome.policy_created, "mapping_created": mirror_outcome.mapping_created, "duration_ms": round((time.perf_counter() - started_at) * 1000, 3), @@ -155,6 +223,7 @@ def mirror_policy_after_commit( outcome="error", actual_write_sources=["cloud_sql"], failure_category=_failure_category(error), + failure_reason=policy_copy_failure_reason(error), ), ) msg = "The committed v1 policy could not be mirrored to v2" diff --git a/policyengine_api/services/user_policy_mirroring.py b/policyengine_api/services/user_policy_mirroring.py index e3f14542b..5667bf2a8 100644 --- a/policyengine_api/services/user_policy_mirroring.py +++ b/policyengine_api/services/user_policy_mirroring.py @@ -15,6 +15,10 @@ ) from policyengine_api.data.v2.settings import V2ConfigurationError from policyengine_api.gcp_logging import logger +from policyengine_api.services.policy_mirroring import ( + PolicyCopyFailureReason, + policy_copy_failure_reason, +) from policyengine_api.services.policy_service import PolicyService from policyengine_api.services.user_policy_service import ( PendingUserPolicyMirrorEvent, @@ -60,6 +64,10 @@ class UserPolicyMirrorUnavailableError(RuntimeError): """Raised when a committed saved policy could not be mirrored immediately.""" +class ReformPolicyUnavailableError(UserPolicyMirrorUnavailableError): + """Raised when a saved policy's v1 reform policy cannot be loaded to copy it.""" + + def _default_mirror_factory() -> LegacyUserPolicyMirror: from policyengine_api.data.v2.database import ( get_v2_session_factory, @@ -100,6 +108,14 @@ def _failure_category(error: Exception) -> str: return "unexpected" +def _failure_reason(error: Exception) -> str: + if isinstance(error, LegacyUserPolicyIntegrityError): + return PolicyCopyFailureReason.SAVED_POLICY_MAPPING_CONFLICT.value + if isinstance(error, ReformPolicyUnavailableError): + return PolicyCopyFailureReason.MISSING_REFORM_POLICY.value + return policy_copy_failure_reason(error) + + @dataclass(frozen=True) class UserPolicyMirrorCollaborators: """The services a saved-policy mirror uses; each one left None takes its default. @@ -135,6 +151,7 @@ class _MirrorOutcome: actual_write_sources: list[str] result: LegacyUserPolicyPersistenceResult | None = None failure_category: str | None = None + failure_reason: str | None = None def _log_mirror_operation( @@ -164,6 +181,7 @@ def _log_mirror_operation( ), "outcome": mirror_outcome.outcome, "failure_category": mirror_outcome.failure_category, + "failure_reason": mirror_outcome.failure_reason, "association_created": ( result.association_created if result is not None else None ), @@ -232,6 +250,7 @@ def mirror_user_policy_after_commit( outcome="error", actual_write_sources=["cloud_sql"], failure_category=_failure_category(error), + failure_reason=_failure_reason(error), ), ) msg = "The committed v1 saved policy could not be mirrored to v2" @@ -292,7 +311,7 @@ def process( ) if reform_snapshot is None: msg = "The saved policy references an unavailable reform policy" - raise UserPolicyMirrorUnavailableError(msg) + raise ReformPolicyUnavailableError(msg) result = _run_user_policy_mirror( event.snapshot, reform_snapshot, @@ -367,6 +386,7 @@ def after_processed_commit( ), result=result, failure_category=_failure_category(failure), + failure_reason=_failure_reason(failure), ), ) msg = "The committed v1 saved policy event could not be mirrored to v2" diff --git a/tests/unit/routes/test_policy_dual_write_routes.py b/tests/unit/routes/test_policy_dual_write_routes.py index 397b741da..ef2aebd3e 100644 --- a/tests/unit/routes/test_policy_dual_write_routes.py +++ b/tests/unit/routes/test_policy_dual_write_routes.py @@ -250,6 +250,7 @@ def test_destination_failure_returns_the_stored_policy_and_logs_an_error( assert logged_record["metric_name"] == "v1_policy_mirror_operations" assert logged_record["outcome"] == "error" assert logged_record["failure_category"] == "catalog_or_translation" + assert logged_record["failure_reason"] == "missing_parameters" assert logged_record["legacy_policy_id"] == stored_policy_id assert "private parameter value" not in json.dumps(logged_record) @@ -283,6 +284,7 @@ def test_policy_without_a_mirror_snapshot_is_returned_and_logged( assert logged_record["metric_name"] == "v1_policy_mirror_operations" assert logged_record["outcome"] == "error" assert logged_record["failure_category"] == "missing_mirror_snapshot" + assert logged_record["failure_reason"] == "missing_mirror_snapshot" assert logged_record["country_id"] == "us" assert logged_record["legacy_policy_id"] == 42 assert route_logger.log_struct.call_args.kwargs["severity"] == "ERROR" diff --git a/tests/unit/routes/test_user_policy_dual_write_routes.py b/tests/unit/routes/test_user_policy_dual_write_routes.py index 7233002b4..c36f57436 100644 --- a/tests/unit/routes/test_user_policy_dual_write_routes.py +++ b/tests/unit/routes/test_user_policy_dual_write_routes.py @@ -451,6 +451,7 @@ def test_unexpected_saved_policy_mirror_error_is_logged_and_the_record_returned( assert logged_record["metric_name"] == "v1_user_policy_mirror_operations" assert logged_record["outcome"] == "error" assert logged_record["failure_category"] == "unexpected" + assert logged_record["failure_reason"] == "unexpected" assert logged_record["legacy_user_policy_id"] == 10 assert "secret" not in json.dumps(logged_record) assert route_logger.log_struct.call_args.kwargs["severity"] == "ERROR" @@ -480,6 +481,7 @@ def test_saved_policy_without_a_mirror_revision_is_returned_and_logged( assert logged_record["metric_name"] == "v1_user_policy_mirror_operations" assert logged_record["outcome"] == "error" assert logged_record["failure_category"] == "missing_mirror_revision" + assert logged_record["failure_reason"] == "missing_mirror_revision" assert logged_record["legacy_user_policy_id"] == 10 diff --git a/tests/unit/services/test_mirror_log_payloads.py b/tests/unit/services/test_mirror_log_payloads.py index 493b600de..426ba4169 100644 --- a/tests/unit/services/test_mirror_log_payloads.py +++ b/tests/unit/services/test_mirror_log_payloads.py @@ -204,6 +204,7 @@ def test_policy_mirror_success_logs_one_complete_record() -> None: "destination_policy_id": str(DESTINATION_ID), "outcome": "ok", "failure_category": None, + "failure_reason": None, "policy_created": False, "mapping_created": True, } @@ -235,6 +236,7 @@ def test_policy_mirror_failure_logs_one_complete_record() -> None: "destination_policy_id": None, "outcome": "error", "failure_category": "database", + "failure_reason": "database_error", "policy_created": None, "mapping_created": None, } @@ -285,6 +287,7 @@ def _user_policy_record(**changes: object) -> dict[str, object]: "destination_policy_id": str(DESTINATION_ID), "outcome": "ok", "failure_category": None, + "failure_reason": None, "association_created": False, "association_updated": True, "mapping_created": False, @@ -299,6 +302,7 @@ def _user_policy_record(**changes: object) -> dict[str, object]: destination_policy_id=None, outcome="error", failure_category="database", + failure_reason="database_error", association_created=None, association_updated=None, mapping_created=None, @@ -429,6 +433,7 @@ def test_pending_event_failure_logs_the_active_event_and_unwrapped_cause() -> No legacy_user_policy_id=11, outcome="error", failure_category="database", + failure_reason="database_error", ) assert list(payload) == list(expected) assert payload == expected diff --git a/tests/unit/services/test_policy_mirroring.py b/tests/unit/services/test_policy_mirroring.py index ad2bdaf3f..833802d63 100644 --- a/tests/unit/services/test_policy_mirroring.py +++ b/tests/unit/services/test_policy_mirroring.py @@ -9,6 +9,11 @@ from sqlalchemy.exc import OperationalError from sqlalchemy.exc import TimeoutError as PoolTimeoutError +from policyengine_api.data.v2.catalog.catalog_selection import ( + MetadataCatalogUnavailableError, + MetadataCatalogVersionNotFoundError, +) +from policyengine_api.data.v2.settings import V2ConfigurationError from policyengine_api.services.policy_mirroring import ( PolicyMirrorUnavailableError, mirror_policy_after_commit, @@ -19,6 +24,11 @@ ) from policyengine_api.services.v2.policies.validators import ( LegacyPolicyMappingIntegrityError, + LegacyPolicyTranslationError, + LegacyPolicyTranslationReason, + PolicyCatalogValidationError, + PolicyContentHashCollisionError, + PolicyCreationIntegrityError, ) POLICY_ID = UUID("00000000-0000-0000-0000-000000000010") @@ -63,6 +73,7 @@ def test_success_returns_destination_and_logs_metric_without_policy_content() -> "destination_policy_id": str(POLICY_ID), "outcome": "ok", "failure_category": None, + "failure_reason": None, "policy_created": True, "mapping_created": True, } @@ -123,3 +134,113 @@ def test_failure_is_secret_safe_and_records_cloud_sql_as_only_completed_write( assert payload["destination_policy_id"] is None assert "secret" not in repr(payload) assert "credential" not in repr(payload) + + +def _catalog_query_failure() -> MetadataCatalogUnavailableError: + error = MetadataCatalogUnavailableError("the v2 metadata catalog cannot be queried") + error.__cause__ = OperationalError("select secret", {}, Exception("credential")) + return error + + +def _translation_failure( + reason: LegacyPolicyTranslationReason, +) -> LegacyPolicyTranslationError: + return LegacyPolicyTranslationError("gov.private.secret", reason=reason) + + +@pytest.mark.parametrize( + ("error", "category", "reason"), + [ + pytest.param( + V2ConfigurationError("secret url"), + "configuration", + "configuration", + id="configuration", + ), + pytest.param( + _catalog_query_failure(), + "catalog_or_translation", + "catalog_query_failed", + id="catalog-query-failed", + ), + pytest.param( + MetadataCatalogUnavailableError("catalog absent secret"), + "catalog_or_translation", + "catalog_absent", + id="catalog-absent", + ), + pytest.param( + MetadataCatalogVersionNotFoundError("secret"), + "catalog_or_translation", + "catalog_version_not_found", + id="catalog-version-not-found", + ), + pytest.param( + PolicyCatalogValidationError("secret"), + "catalog_or_translation", + "catalog_mismatch", + id="catalog-mismatch", + ), + *( + pytest.param( + _translation_failure(reason), + "catalog_or_translation", + reason.value, + id=reason.value, + ) + for reason in LegacyPolicyTranslationReason + ), + pytest.param( + LegacyPolicyMappingIntegrityError("secret"), + "integrity", + "mapping_conflict", + id="mapping-conflict", + ), + pytest.param( + PolicyContentHashCollisionError("secret"), + "integrity", + "content_hash_collision", + id="content-hash-collision", + ), + pytest.param( + PolicyCreationIntegrityError("secret"), + "integrity", + "stored_policy_integrity", + id="stored-policy-integrity", + ), + pytest.param( + PoolTimeoutError("pool timeout secret"), + "database", + "database_error", + id="database", + ), + pytest.param( + RuntimeError("policy content secret"), + "unexpected", + "unexpected", + id="unexpected", + ), + ], +) +def test_each_failed_copy_logs_a_fixed_reason_with_its_category( + error: Exception, + category: str, + reason: str, +) -> None: + mirror = MagicMock() + mirror.mirror_legacy_policy.side_effect = error + + with ( + patch("policyengine_api.services.policy_mirroring.logger") as logger, + pytest.raises(PolicyMirrorUnavailableError), + ): + mirror_policy_after_commit(_snapshot(), mirror_factory=lambda: mirror) + + payload = logger.log_struct.call_args.args[0] + assert (payload["failure_category"], payload["failure_reason"]) == ( + category, + reason, + ) + assert type(payload["failure_reason"]) is str + assert "secret" not in repr(payload) + assert "credential" not in repr(payload) diff --git a/tests/unit/services/test_user_policy_mirroring.py b/tests/unit/services/test_user_policy_mirroring.py index df8e7facf..2bb9242f2 100644 --- a/tests/unit/services/test_user_policy_mirroring.py +++ b/tests/unit/services/test_user_policy_mirroring.py @@ -18,8 +18,16 @@ mirror_pending_user_policy_events_after_commit, mirror_user_policy_after_commit, ) -from policyengine_api.services.user_policy_service import UserPolicyService +from policyengine_api.services.user_policy_service import ( + PendingUserPolicyMirrorEvent, + UserPolicyService, +) from policyengine_api.services.v2.policies.types import LegacyPolicySnapshot +from policyengine_api.services.v2.policies.validators import ( + LegacyPolicyMappingIntegrityError, + LegacyPolicyTranslationError, + LegacyPolicyTranslationReason, +) from policyengine_api.services.v2.user_policies.types import ( LegacyUserPolicyPersistenceResult, LegacyUserPolicySnapshot, @@ -131,20 +139,39 @@ def test_logging_failure_does_not_change_successful_mirror_result() -> None: @pytest.mark.parametrize( - ("error", "category"), + ("error", "category", "reason"), [ ( OperationalError("statement secret", {}, Exception("credential")), "database", + "database_error", + ), + (PoolTimeoutError("pool timeout"), "database", "database_error"), + ( + LegacyUserPolicyIntegrityError("mapping conflict"), + "integrity", + "saved_policy_mapping_conflict", ), - (PoolTimeoutError("pool timeout"), "database"), - (LegacyUserPolicyIntegrityError("mapping conflict"), "integrity"), - (RuntimeError("caller data secret"), "unexpected"), + ( + LegacyPolicyMappingIntegrityError("reform mapping conflict"), + "integrity", + "mapping_conflict", + ), + ( + LegacyPolicyTranslationError( + "gov.private.secret", + reason=LegacyPolicyTranslationReason.MISSING_PARAMETERS, + ), + "catalog_or_translation", + "missing_parameters", + ), + (RuntimeError("caller data secret"), "unexpected", "unexpected"), ], ) def test_failure_is_secret_safe_and_reports_only_cloud_sql_completed( error: Exception, category: str, + reason: str, ) -> None: mirror = MagicMock() mirror.mirror_legacy_user_policy.side_effect = error @@ -163,6 +190,7 @@ def test_failure_is_secret_safe_and_reports_only_cloud_sql_completed( payload = logger.log_struct.call_args.args[0] assert payload["actual_write_sources"] == ["cloud_sql"] assert payload["failure_category"] == category + assert payload["failure_reason"] == reason assert payload["destination_association_id"] is None assert "secret" not in repr(payload) assert "credential" not in repr(payload) @@ -268,3 +296,40 @@ def test_event_preparation_failure_logs_error_and_retains_the_event( with orm_session_factory() as session: retained_event = session.scalar(select(UserPolicyMirrorEvent)) assert retained_event.processed_at is None + + +def test_saved_policy_whose_reform_is_missing_logs_its_own_reason() -> None: + event_service = MagicMock() + + def process(_country_id, _legacy_user_policy_id, **callbacks): + return callbacks["processor"]( + PendingUserPolicyMirrorEvent( + event_id=1, + source_revision=1, + event_type="create", + snapshot=_saved(), + changed_fields=frozenset(), + source_fingerprint_sha256="f" * 64, + ) + ) + + event_service.process_pending_mirror_events.side_effect = process + + with ( + patch("policyengine_api.services.user_policy_mirroring.logger") as logger, + pytest.raises(UserPolicyMirrorUnavailableError), + ): + mirror_pending_user_policy_events_after_commit( + "us", + 10, + through_revision=1, + collaborators=UserPolicyMirrorCollaborators( + event_service=event_service, + reform_snapshot_loader=lambda _country_id, _policy_id: None, + mirror_factory=MagicMock(), + ), + ) + + payload = logger.log_struct.call_args.args[0] + assert payload["failure_category"] == "unexpected" + assert payload["failure_reason"] == "missing_reform_policy" From f1de4a81ff1a61b7490e1d0ca2ad54ecc1b28286 Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:20:46 +0400 Subject: [PATCH 5/8] Log too_many_values, not conflicting_values, for a policy over the v2 size limit translate_legacy_policy mapped every ValueError from PolicyCreationInput to conflicting_values. PolicyCreationInput also refuses more than 1,000 dated values, and the v1 route puts no limit on a policy's size, so a large v1 policy logged a conflict that was not there. The count is now checked first and logged as too_many_values; what is left for conflicting_values is two values for one parameter with the same start date. Co-Authored-By: Claude Opus 5.5 --- docs/migration/stage-10-v2-policies.md | 2 +- .../services/v2/policies/transformations.py | 8 ++++++++ .../services/v2/policies/validators.py | 1 + .../unit/v2/test_policy_legacy_translation.py | 20 +++++++++++++++++++ 4 files changed, 30 insertions(+), 1 deletion(-) diff --git a/docs/migration/stage-10-v2-policies.md b/docs/migration/stage-10-v2-policies.md index ddef0f6a2..5a3362790 100644 --- a/docs/migration/stage-10-v2-policies.md +++ b/docs/migration/stage-10-v2-policies.md @@ -180,7 +180,7 @@ Monitor structured events with metric names `v1_policy_mirror_operations` and | `failure_category` | `failure_reason` | | --- | --- | -| `catalog_or_translation` | `missing_parameters` (the policy sets a parameter the running catalog lacks), `invalid_parameter_values`, `empty_parameter_values`, `invalid_period`, `conflicting_values`, `catalog_absent` (no catalog for the running PolicyEngine.py version), `catalog_query_failed`, `catalog_version_not_found`, `catalog_mismatch` | +| `catalog_or_translation` | `missing_parameters` (the policy sets a parameter the running catalog lacks), `invalid_parameter_values`, `empty_parameter_values`, `invalid_period`, `conflicting_values`, `too_many_values` (more than 1,000 dated values), `catalog_absent` (no catalog for the running PolicyEngine.py version), `catalog_query_failed`, `catalog_version_not_found`, `catalog_mismatch` | | `integrity` | `mapping_conflict`, `saved_policy_mapping_conflict`, `content_hash_collision`, `stored_policy_integrity` | | `database` | `database_error` | | `configuration` | `configuration` | diff --git a/policyengine_api/services/v2/policies/transformations.py b/policyengine_api/services/v2/policies/transformations.py index d4120ed01..4a500a262 100644 --- a/policyengine_api/services/v2/policies/transformations.py +++ b/policyengine_api/services/v2/policies/transformations.py @@ -23,6 +23,7 @@ ResolvedPolicyCreationInput, ) from policyengine_api.services.v2.policies.validators import ( + MAXIMUM_POLICY_PARAMETER_VALUES, LegacyPolicyTranslationError, LegacyPolicyTranslationReason, parameters_missing_from_model, @@ -316,6 +317,11 @@ def translate_legacy_policy( end_date=end_date, ) ) + if len(parameter_values) > MAXIMUM_POLICY_PARAMETER_VALUES: + msg = "legacy policy has more values than a v2 policy may hold" + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.TOO_MANY_VALUES + ) try: policy_input = PolicyCreationInput( country_id=snapshot.country_id, @@ -323,6 +329,8 @@ def translate_legacy_policy( parameter_values=parameter_values, ) except ValueError as error: + # The count is checked above, so what remains is two values for one + # parameter with the same start date. msg = "legacy parameter periods or values conflict" raise LegacyPolicyTranslationError( msg, reason=LegacyPolicyTranslationReason.CONFLICTING_VALUES diff --git a/policyengine_api/services/v2/policies/validators.py b/policyengine_api/services/v2/policies/validators.py index 7bf2cf5d5..d2891d062 100644 --- a/policyengine_api/services/v2/policies/validators.py +++ b/policyengine_api/services/v2/policies/validators.py @@ -51,6 +51,7 @@ class LegacyPolicyTranslationReason(StrEnum): EMPTY_PARAMETER_VALUES = "empty_parameter_values" INVALID_PERIOD = "invalid_period" CONFLICTING_VALUES = "conflicting_values" + TOO_MANY_VALUES = "too_many_values" class LegacyPolicyTranslationError(ValueError): diff --git a/tests/unit/v2/test_policy_legacy_translation.py b/tests/unit/v2/test_policy_legacy_translation.py index 7540b3349..453a60891 100644 --- a/tests/unit/v2/test_policy_legacy_translation.py +++ b/tests/unit/v2/test_policy_legacy_translation.py @@ -32,6 +32,7 @@ ) from policyengine_api.services.v2.policies.types import LegacyPolicySnapshot from policyengine_api.services.v2.policies.validators import ( + MAXIMUM_POLICY_PARAMETER_VALUES, LegacyPolicyTranslationError, LegacyPolicyTranslationReason, parameters_missing_from_model, @@ -296,6 +297,25 @@ def test_untranslatable_policy_fails_with_a_fixed_reason( engine.dispose() +def test_policy_with_more_values_than_v2_accepts_fails_as_too_many_values() -> None: + engine, session, _model, _version, _first, _second = _session_and_catalog() + # One more dated value than a v2 policy may hold, with no repeated dates. + too_many_periods = { + f"{2026 + offset}-01-01.{2026 + offset}-12-31": offset + for offset in range(MAXIMUM_POLICY_PARAMETER_VALUES + 1) + } + try: + with pytest.raises(LegacyPolicyTranslationError) as failure: + _translate( + session, + _snapshot(policy_json={"gov.example.rate": too_many_periods}), + ) + assert failure.value.reason is LegacyPolicyTranslationReason.TOO_MANY_VALUES + finally: + session.close() + engine.dispose() + + def test_parameters_missing_from_model_lists_each_absent_name_once_in_order() -> None: model_parameter_names = {"gov.example.rate", "gov.example.amount"} From be5bbbe22bab20de2bafb9dfc1501883e1fa25d0 Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:22:37 +0400 Subject: [PATCH 6/8] Reuse a concurrent copy's mapping by source hash, as a mapping found first is When two first copies of the same v1 policy run at once and this one loses the mapping insert, the copy still required the winner's v2 policy to equal its own. Two copies on revisions with different PolicyEngine.py versions translate to different v2 policies, so the loser logged mapping_conflict for a mapping that is valid under the rule the previous commit introduced: the same source hash makes the mapped v2 policy the copy. The race branch now applies that rule too, and verify_legacy_policy_mapping no longer takes expected_policy_id. The losing copy's own v2 policy row is then committed without a mapping. It is immutable, deduplicated content that nothing refers to. Co-Authored-By: Claude Opus 5.5 --- .../services/v2/policies/services.py | 10 +-- .../services/v2/policies/validators.py | 10 +-- tests/unit/v2/test_policy_legacy_copy.py | 61 +++++++++++++++++++ 3 files changed, 71 insertions(+), 10 deletions(-) diff --git a/policyengine_api/services/v2/policies/services.py b/policyengine_api/services/v2/policies/services.py index b3ebd6f1e..dd7cc2925 100644 --- a/policyengine_api/services/v2/policies/services.py +++ b/policyengine_api/services/v2/policies/services.py @@ -198,8 +198,7 @@ def mirror_legacy_policy_in_session( returns that copy without being translated again, so writing it again after a PolicyEngine.py or country package upgrade still succeeds. Raise LegacyPolicyMappingIntegrityError if the v1 id is already mapped with a - different source hash, or a concurrent copy mapped it to a different v2 - policy. + different source hash. """ existing = read_legacy_policy_mapping( session, @@ -261,10 +260,11 @@ def mirror_legacy_policy_in_session( if concurrent is None: msg = "legacy policy mapping conflict did not resolve to a stored row" raise LegacyPolicyMappingIntegrityError(msg) + # A concurrent copy mapped the policy first. As with a mapping found at the + # start, the same source hash makes its v2 policy the copy, even when it was + # translated with another catalog version than this one. verify_legacy_policy_mapping( - concurrent, - source_policy_hash=snapshot.source_policy_hash, - expected_policy_id=policy_result.policy_id, + concurrent, source_policy_hash=snapshot.source_policy_hash ) return LegacyPolicyPersistenceResult( policy_id=concurrent.policy_id, diff --git a/policyengine_api/services/v2/policies/validators.py b/policyengine_api/services/v2/policies/validators.py index d2891d062..d7317ec1d 100644 --- a/policyengine_api/services/v2/policies/validators.py +++ b/policyengine_api/services/v2/policies/validators.py @@ -179,12 +179,12 @@ def verify_legacy_policy_mapping( mapping: LegacyPolicyMapping, *, source_policy_hash: str, - expected_policy_id: UUID | None = None, ) -> None: - """Validate an existing mapping without performing SQL.""" + """Require an existing mapping to come from the same v1 content, without SQL. + + v1 policies never change, so a mapping with the same source hash is the + copy whichever catalog version translated it. + """ if mapping.source_policy_hash != source_policy_hash: msg = "legacy policy identity was presented with a different source hash" raise LegacyPolicyMappingIntegrityError(msg) - if expected_policy_id is not None and mapping.policy_id != expected_policy_id: - msg = "legacy policy mapping does not match translated immutable content" - raise LegacyPolicyMappingIntegrityError(msg) diff --git a/tests/unit/v2/test_policy_legacy_copy.py b/tests/unit/v2/test_policy_legacy_copy.py index 43cfc12c4..e3ea73d7b 100644 --- a/tests/unit/v2/test_policy_legacy_copy.py +++ b/tests/unit/v2/test_policy_legacy_copy.py @@ -289,3 +289,64 @@ def test_earlier_copy_with_another_source_hash_is_an_integrity_error( _copy(catalog_session, snapshot) assert recorded_writes == RecordedCopyWrites() + + +def _lose_mapping_race(monkeypatch, *, winning_source_hash: str) -> None: + """Make the mapping insert find a row a concurrent copy committed first. + + The replacement returns None, as the insert does when its unique + constraint makes it a no-op. + """ + + def insert_after_concurrent_copy(session, **mapping_fields): + session.add( + LegacyPolicyMapping( + country_id=mapping_fields["country_id"], + legacy_policy_id=mapping_fields["legacy_policy_id"], + policy_id=EARLIER_COPY_POLICY_ID, + source_policy_hash=winning_source_hash, + ) + ) + session.flush() + + monkeypatch.setattr( + policy_services, "create_legacy_policy_mapping", insert_after_concurrent_copy + ) + + +def test_concurrent_copy_on_another_catalog_version_is_reused( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, + monkeypatch, +) -> None: + # Two first copies of the same v1 policy ran at once, for example on two + # revisions with different PolicyEngine.py versions, so each translated + # it to a different v2 policy. The mapping committed first is the copy. + _add_running_catalog( + catalog_session, country_id="us", parameter_names=["gov.dol.minimum_wage"] + ) + snapshot = _reused_snapshot() + _lose_mapping_race(monkeypatch, winning_source_hash=snapshot.source_policy_hash) + + result = _copy(catalog_session, snapshot) + + assert result == LegacyPolicyPersistenceResult( + policy_id=EARLIER_COPY_POLICY_ID, + policy_created=False, + mapping_created=False, + ) + assert len(recorded_writes.created_policies) == 1 + + +def test_concurrent_copy_with_another_source_hash_is_an_integrity_error( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, + monkeypatch, +) -> None: + _add_running_catalog( + catalog_session, country_id="us", parameter_names=["gov.dol.minimum_wage"] + ) + _lose_mapping_race(monkeypatch, winning_source_hash="different-legacy-hash") + + with pytest.raises(LegacyPolicyMappingIntegrityError, match="source hash"): + _copy(catalog_session, _reused_snapshot()) From 9132a0dc98b0741b0668de2b0517bd2cd52bf660 Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:28:30 +0400 Subject: [PATCH 7/8] Classify each failed copy once, into its category and reason together The previous commits classified a failed copy twice: once for failure_category and once for failure_reason, and the saved-policy copy had a third copy of the category rules. classify_policy_copy_failure now returns both from one table of exception types, and the saved-policy copy adds only its two own cases. PolicyCopyFailureReason also holds the route's missing_mirror_snapshot and missing_mirror_revision, so the logged values are typed and listed in one place. Every LegacyUserPolicyIntegrityError is now saved_policy_integrity, because that exception covers more checks than mapping conflicts. LegacyPolicyTranslationReason moves to the policies types module, where the v2 code organization guide puts shared enums. Co-Authored-By: Claude Opus 5.5 --- docs/migration/stage-10-v2-policies.md | 2 +- policyengine_api/routes/policy_routes.py | 13 +- policyengine_api/services/policy_mirroring.py | 130 +++++++++--------- .../services/user_policy_mirroring.py | 71 +++------- .../services/v2/policies/transformations.py | 2 +- .../services/v2/policies/types.py | 14 ++ .../services/v2/policies/validators.py | 13 +- .../routes/test_policy_dual_write_routes.py | 6 +- tests/unit/services/test_policy_mirroring.py | 2 +- .../services/test_user_policy_mirroring.py | 8 +- .../unit/v2/test_policy_legacy_translation.py | 6 +- 11 files changed, 119 insertions(+), 148 deletions(-) diff --git a/docs/migration/stage-10-v2-policies.md b/docs/migration/stage-10-v2-policies.md index 5a3362790..0cfa2c3af 100644 --- a/docs/migration/stage-10-v2-policies.md +++ b/docs/migration/stage-10-v2-policies.md @@ -181,7 +181,7 @@ Monitor structured events with metric names `v1_policy_mirror_operations` and | `failure_category` | `failure_reason` | | --- | --- | | `catalog_or_translation` | `missing_parameters` (the policy sets a parameter the running catalog lacks), `invalid_parameter_values`, `empty_parameter_values`, `invalid_period`, `conflicting_values`, `too_many_values` (more than 1,000 dated values), `catalog_absent` (no catalog for the running PolicyEngine.py version), `catalog_query_failed`, `catalog_version_not_found`, `catalog_mismatch` | -| `integrity` | `mapping_conflict`, `saved_policy_mapping_conflict`, `content_hash_collision`, `stored_policy_integrity` | +| `integrity` | `mapping_conflict`, `content_hash_collision`, `stored_policy_integrity`, `saved_policy_integrity` (any integrity check of a saved-policy copy) | | `database` | `database_error` | | `configuration` | `configuration` | | `unexpected` | `missing_reform_policy` (a saved policy's v1 reform could not be loaded), `unexpected` | diff --git a/policyengine_api/routes/policy_routes.py b/policyengine_api/routes/policy_routes.py index e474341b0..9bb52b4e3 100644 --- a/policyengine_api/routes/policy_routes.py +++ b/policyengine_api/routes/policy_routes.py @@ -21,6 +21,7 @@ from policyengine_api.request_context import current_request_id from policyengine_api.response_factory import _make_error_response from policyengine_api.services.policy_mirroring import ( + PolicyCopyFailureReason, PolicyMirrorUnavailableError, mirror_policy_after_commit, ) @@ -68,7 +69,7 @@ def _log_uncopied_v1_write( legacy_id_field: str, legacy_id: int, country_id: str, - failure_category: str, + failure_reason: PolicyCopyFailureReason, ) -> None: """Record a committed v1 write that has no v2 copy, without failing the request. @@ -89,9 +90,9 @@ def _log_uncopied_v1_write( legacy_id_field: _safe_legacy_integer_id(legacy_id), "request_id": current_request_id(), "outcome": "error", - "failure_category": failure_category, # These failures have no finer cause than their category. - "failure_reason": failure_category, + "failure_category": failure_reason.value, + "failure_reason": failure_reason.value, }, severity="ERROR", ) @@ -113,7 +114,7 @@ def _copy_policy_to_v2( legacy_id_field="legacy_policy_id", legacy_id=policy_id, country_id=country_id, - failure_category="missing_mirror_snapshot", + failure_reason=PolicyCopyFailureReason.MISSING_MIRROR_SNAPSHOT, ) return # mirror_policy_after_commit has already logged the failure and its category. @@ -137,7 +138,7 @@ def _copy_saved_policy_events_to_v2( legacy_id_field="legacy_user_policy_id", legacy_id=legacy_user_policy_id, country_id=country_id, - failure_category="missing_mirror_revision", + failure_reason=PolicyCopyFailureReason.MISSING_MIRROR_REVISION, ) return try: @@ -159,7 +160,7 @@ def _copy_saved_policy_events_to_v2( legacy_id_field="legacy_user_policy_id", legacy_id=legacy_user_policy_id, country_id=country_id, - failure_category="unexpected", + failure_reason=PolicyCopyFailureReason.UNEXPECTED, ) diff --git a/policyengine_api/services/policy_mirroring.py b/policyengine_api/services/policy_mirroring.py index dbe0507bb..26abbcd52 100644 --- a/policyengine_api/services/policy_mirroring.py +++ b/policyengine_api/services/policy_mirroring.py @@ -19,7 +19,6 @@ from policyengine_api.services.v2.policies.validators import ( LegacyPolicyMappingIntegrityError, LegacyPolicyTranslationError, - LegacyPolicyTranslationReason, PolicyCatalogValidationError, PolicyContentHashCollisionError, PolicyCreationIntegrityError, @@ -31,6 +30,7 @@ from policyengine_api.services.v2.policies.types import ( LegacyPolicyPersistenceResult, LegacyPolicySnapshot, + LegacyPolicyTranslationReason, ) @@ -61,39 +61,12 @@ def _default_mirror_factory() -> LegacyPolicyMirror: return V2PolicyService(PolicyDatabaseSession(get_v2_session_factory())) -def _failure_category(error: Exception) -> str: - if isinstance(error, V2ConfigurationError): - return "configuration" - if isinstance( - error, - ( - MetadataCatalogUnavailableError, - MetadataCatalogVersionNotFoundError, - PolicyCatalogValidationError, - LegacyPolicyTranslationError, - ), - ): - return "catalog_or_translation" - if isinstance( - error, - ( - LegacyPolicyMappingIntegrityError, - PolicyContentHashCollisionError, - PolicyCreationIntegrityError, - ), - ): - return "integrity" - if isinstance(error, SQLAlchemyError): - return "database" - return "unexpected" - - class PolicyCopyFailureReason(StrEnum): - """Fixed reason codes a failed policy or saved-policy copy logs. + """Fixed reason codes that failed policy and saved-policy copies log. - A translation failure logs its LegacyPolicyTranslationReason value instead, - such as ``missing_parameters``. Together the two sets are every value of a - copy record's ``failure_reason``. + A translation failure logs its LegacyPolicyTranslationReason instead, such + as ``missing_parameters``. The two sets together are every value of a copy + record's ``failure_reason``. """ CONFIGURATION = "configuration" @@ -102,54 +75,82 @@ class PolicyCopyFailureReason(StrEnum): CATALOG_VERSION_NOT_FOUND = "catalog_version_not_found" CATALOG_MISMATCH = "catalog_mismatch" MAPPING_CONFLICT = "mapping_conflict" - SAVED_POLICY_MAPPING_CONFLICT = "saved_policy_mapping_conflict" CONTENT_HASH_COLLISION = "content_hash_collision" STORED_POLICY_INTEGRITY = "stored_policy_integrity" + SAVED_POLICY_INTEGRITY = "saved_policy_integrity" MISSING_REFORM_POLICY = "missing_reform_policy" + MISSING_MIRROR_SNAPSHOT = "missing_mirror_snapshot" + MISSING_MIRROR_REVISION = "missing_mirror_revision" DATABASE_ERROR = "database_error" UNEXPECTED = "unexpected" -def policy_copy_failure_reason(error: Exception) -> str: - """Return the fixed reason code a failed policy copy logs for this error. +@dataclass(frozen=True) +class PolicyCopyFailure: + """How a failed copy is logged: a broad category and a fixed reason.""" - The code comes only from the error's type, and for a translation error its - fixed reason, so a log record never holds policy content or error text. - """ - return _failure_reason_code(error).value + category: str + reason: PolicyCopyFailureReason | LegacyPolicyTranslationReason # Checked in order, so the hash collision is matched before its integrity base # class. -_FAILURE_REASONS_BY_ERROR_TYPE: tuple[ - tuple[type[Exception], PolicyCopyFailureReason], ... -] = ( - (V2ConfigurationError, PolicyCopyFailureReason.CONFIGURATION), +_FAILURES_BY_ERROR_TYPE: tuple[tuple[type[Exception], PolicyCopyFailure], ...] = ( + ( + V2ConfigurationError, + PolicyCopyFailure("configuration", PolicyCopyFailureReason.CONFIGURATION), + ), ( MetadataCatalogVersionNotFoundError, - PolicyCopyFailureReason.CATALOG_VERSION_NOT_FOUND, + PolicyCopyFailure( + "catalog_or_translation", + PolicyCopyFailureReason.CATALOG_VERSION_NOT_FOUND, + ), + ), + ( + PolicyCatalogValidationError, + PolicyCopyFailure( + "catalog_or_translation", PolicyCopyFailureReason.CATALOG_MISMATCH + ), + ), + ( + LegacyPolicyMappingIntegrityError, + PolicyCopyFailure("integrity", PolicyCopyFailureReason.MAPPING_CONFLICT), + ), + ( + PolicyContentHashCollisionError, + PolicyCopyFailure("integrity", PolicyCopyFailureReason.CONTENT_HASH_COLLISION), + ), + ( + PolicyCreationIntegrityError, + PolicyCopyFailure("integrity", PolicyCopyFailureReason.STORED_POLICY_INTEGRITY), + ), + ( + SQLAlchemyError, + PolicyCopyFailure("database", PolicyCopyFailureReason.DATABASE_ERROR), ), - (PolicyCatalogValidationError, PolicyCopyFailureReason.CATALOG_MISMATCH), - (LegacyPolicyMappingIntegrityError, PolicyCopyFailureReason.MAPPING_CONFLICT), - (PolicyContentHashCollisionError, PolicyCopyFailureReason.CONTENT_HASH_COLLISION), - (PolicyCreationIntegrityError, PolicyCopyFailureReason.STORED_POLICY_INTEGRITY), - (SQLAlchemyError, PolicyCopyFailureReason.DATABASE_ERROR), ) -def _failure_reason_code( - error: Exception, -) -> PolicyCopyFailureReason | LegacyPolicyTranslationReason: +def classify_policy_copy_failure(error: Exception) -> PolicyCopyFailure: + """Return how a failed policy copy logs this error. + + Both fields come only from the error's type, and for a translation error + its fixed reason, so a log record never holds policy content or error text. + """ if isinstance(error, LegacyPolicyTranslationError): - return error.reason + return PolicyCopyFailure("catalog_or_translation", error.reason) if isinstance(error, MetadataCatalogUnavailableError): - if isinstance(error.__cause__, SQLAlchemyError): - return PolicyCopyFailureReason.CATALOG_QUERY_FAILED - return PolicyCopyFailureReason.CATALOG_ABSENT - for error_type, reason in _FAILURE_REASONS_BY_ERROR_TYPE: + reason = ( + PolicyCopyFailureReason.CATALOG_QUERY_FAILED + if isinstance(error.__cause__, SQLAlchemyError) + else PolicyCopyFailureReason.CATALOG_ABSENT + ) + return PolicyCopyFailure("catalog_or_translation", reason) + for error_type, failure in _FAILURES_BY_ERROR_TYPE: if isinstance(error, error_type): - return reason - return PolicyCopyFailureReason.UNEXPECTED + return failure + return PolicyCopyFailure("unexpected", PolicyCopyFailureReason.UNEXPECTED) @dataclass(frozen=True) @@ -159,8 +160,7 @@ class _MirrorOutcome: outcome: str actual_write_sources: list[str] destination_policy_id: object | None = None - failure_category: str | None = None - failure_reason: str | None = None + failure: PolicyCopyFailure | None = None policy_created: bool | None = None mapping_created: bool | None = None @@ -172,6 +172,7 @@ def _log_mirror_event( mirror_outcome: _MirrorOutcome, ) -> None: destination_policy_id = mirror_outcome.destination_policy_id + failure = mirror_outcome.failure # Metric logging must not change the mirror outcome. with contextlib.suppress(Exception): logger.log_struct( @@ -190,8 +191,8 @@ def _log_mirror_event( else None ), "outcome": mirror_outcome.outcome, - "failure_category": mirror_outcome.failure_category, - "failure_reason": mirror_outcome.failure_reason, + "failure_category": failure.category if failure else None, + "failure_reason": failure.reason.value if failure else None, "policy_created": mirror_outcome.policy_created, "mapping_created": mirror_outcome.mapping_created, "duration_ms": round((time.perf_counter() - started_at) * 1000, 3), @@ -222,8 +223,7 @@ def mirror_policy_after_commit( mirror_outcome=_MirrorOutcome( outcome="error", actual_write_sources=["cloud_sql"], - failure_category=_failure_category(error), - failure_reason=policy_copy_failure_reason(error), + failure=classify_policy_copy_failure(error), ), ) msg = "The committed v1 policy could not be mirrored to v2" diff --git a/policyengine_api/services/user_policy_mirroring.py b/policyengine_api/services/user_policy_mirroring.py index 5667bf2a8..ecd69c59b 100644 --- a/policyengine_api/services/user_policy_mirroring.py +++ b/policyengine_api/services/user_policy_mirroring.py @@ -7,30 +7,17 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Protocol -from sqlalchemy.exc import SQLAlchemyError - -from policyengine_api.data.v2.catalog.catalog_selection import ( - MetadataCatalogUnavailableError, - MetadataCatalogVersionNotFoundError, -) -from policyengine_api.data.v2.settings import V2ConfigurationError from policyengine_api.gcp_logging import logger from policyengine_api.services.policy_mirroring import ( + PolicyCopyFailure, PolicyCopyFailureReason, - policy_copy_failure_reason, + classify_policy_copy_failure, ) from policyengine_api.services.policy_service import PolicyService from policyengine_api.services.user_policy_service import ( PendingUserPolicyMirrorEvent, UserPolicyService, ) -from policyengine_api.services.v2.policies.validators import ( - LegacyPolicyMappingIntegrityError, - LegacyPolicyTranslationError, - PolicyCatalogValidationError, - PolicyContentHashCollisionError, - PolicyCreationIntegrityError, -) from policyengine_api.services.v2.user_policies.validators import ( LegacyUserPolicyIntegrityError, ) @@ -80,40 +67,16 @@ def _default_mirror_factory() -> LegacyUserPolicyMirror: return V2UserPolicyService(UserPolicyDatabaseSession(get_v2_session_factory())) -def _failure_category(error: Exception) -> str: - if isinstance(error, V2ConfigurationError): - return "configuration" - if isinstance( - error, - ( - MetadataCatalogUnavailableError, - MetadataCatalogVersionNotFoundError, - PolicyCatalogValidationError, - LegacyPolicyTranslationError, - ), - ): - return "catalog_or_translation" - if isinstance( - error, - ( - LegacyPolicyMappingIntegrityError, - LegacyUserPolicyIntegrityError, - PolicyContentHashCollisionError, - PolicyCreationIntegrityError, - ), - ): - return "integrity" - if isinstance(error, SQLAlchemyError): - return "database" - return "unexpected" - - -def _failure_reason(error: Exception) -> str: +def _classify_failure(error: Exception) -> PolicyCopyFailure: if isinstance(error, LegacyUserPolicyIntegrityError): - return PolicyCopyFailureReason.SAVED_POLICY_MAPPING_CONFLICT.value + return PolicyCopyFailure( + "integrity", PolicyCopyFailureReason.SAVED_POLICY_INTEGRITY + ) if isinstance(error, ReformPolicyUnavailableError): - return PolicyCopyFailureReason.MISSING_REFORM_POLICY.value - return policy_copy_failure_reason(error) + return PolicyCopyFailure( + "unexpected", PolicyCopyFailureReason.MISSING_REFORM_POLICY + ) + return classify_policy_copy_failure(error) @dataclass(frozen=True) @@ -150,8 +113,7 @@ class _MirrorOutcome: outcome: str actual_write_sources: list[str] result: LegacyUserPolicyPersistenceResult | None = None - failure_category: str | None = None - failure_reason: str | None = None + failure: PolicyCopyFailure | None = None def _log_mirror_operation( @@ -159,6 +121,7 @@ def _log_mirror_operation( mirror_outcome: _MirrorOutcome, ) -> None: result = mirror_outcome.result + failure = mirror_outcome.failure # Metric logging must not change the mirror outcome. with contextlib.suppress(Exception): logger.log_struct( @@ -180,8 +143,8 @@ def _log_mirror_operation( str(result.policy_id) if result is not None else None ), "outcome": mirror_outcome.outcome, - "failure_category": mirror_outcome.failure_category, - "failure_reason": mirror_outcome.failure_reason, + "failure_category": failure.category if failure else None, + "failure_reason": failure.reason.value if failure else None, "association_created": ( result.association_created if result is not None else None ), @@ -249,8 +212,7 @@ def mirror_user_policy_after_commit( _MirrorOutcome( outcome="error", actual_write_sources=["cloud_sql"], - failure_category=_failure_category(error), - failure_reason=_failure_reason(error), + failure=_classify_failure(error), ), ) msg = "The committed v1 saved policy could not be mirrored to v2" @@ -385,8 +347,7 @@ def after_processed_commit( ["cloud_sql", "supabase"] if result is not None else ["cloud_sql"] ), result=result, - failure_category=_failure_category(failure), - failure_reason=_failure_reason(failure), + failure=_classify_failure(failure), ), ) msg = "The committed v1 saved policy event could not be mirrored to v2" diff --git a/policyengine_api/services/v2/policies/transformations.py b/policyengine_api/services/v2/policies/transformations.py index 4a500a262..8b20a6c74 100644 --- a/policyengine_api/services/v2/policies/transformations.py +++ b/policyengine_api/services/v2/policies/transformations.py @@ -15,6 +15,7 @@ from policyengine_api.services.v2.policies.types import ( CanonicalPolicyContent, LegacyPolicySnapshot, + LegacyPolicyTranslationReason, PolicyCreationInput, PolicyPage, PolicyParameterValueInput, @@ -25,7 +26,6 @@ from policyengine_api.services.v2.policies.validators import ( MAXIMUM_POLICY_PARAMETER_VALUES, LegacyPolicyTranslationError, - LegacyPolicyTranslationReason, parameters_missing_from_model, validate_policy_catalog, ) diff --git a/policyengine_api/services/v2/policies/types.py b/policyengine_api/services/v2/policies/types.py index 1c6dc15e2..3783d5e82 100644 --- a/policyengine_api/services/v2/policies/types.py +++ b/policyengine_api/services/v2/policies/types.py @@ -4,6 +4,7 @@ from dataclasses import dataclass from datetime import datetime +from enum import StrEnum from typing import Annotated, Any from uuid import UUID @@ -26,6 +27,19 @@ ) StrictJsonValue = Annotated[JsonValue, BeforeValidator(require_json_value)] + + +class LegacyPolicyTranslationReason(StrEnum): + """Why committed v1 content cannot be translated, as copy failure logs name it.""" + + MISSING_PARAMETERS = "missing_parameters" + INVALID_PARAMETER_VALUES = "invalid_parameter_values" + EMPTY_PARAMETER_VALUES = "empty_parameter_values" + INVALID_PERIOD = "invalid_period" + CONFLICTING_VALUES = "conflicting_values" + TOO_MANY_VALUES = "too_many_values" + + UtcDateTime = Annotated[datetime, AfterValidator(normalize_utc)] diff --git a/policyengine_api/services/v2/policies/validators.py b/policyengine_api/services/v2/policies/validators.py index d7317ec1d..ed7a1bbe2 100644 --- a/policyengine_api/services/v2/policies/validators.py +++ b/policyengine_api/services/v2/policies/validators.py @@ -4,7 +4,6 @@ import math from datetime import UTC, datetime -from enum import StrEnum from typing import TYPE_CHECKING, Any if TYPE_CHECKING: @@ -14,6 +13,7 @@ from policyengine_api.data.v2.catalog.catalog_selection import SelectedCatalog from policyengine_api.data.v2.models import LegacyPolicyMapping from policyengine_api.services.v2.policies.types import ( + LegacyPolicyTranslationReason, PolicyCreationInput, ResolvedPolicyCreationInput, ) @@ -43,17 +43,6 @@ class LegacyPolicyMappingIntegrityError(RuntimeError): """Raised when one immutable v1 identity maps inconsistently.""" -class LegacyPolicyTranslationReason(StrEnum): - """Why committed v1 content cannot be translated, as copy failure logs name it.""" - - MISSING_PARAMETERS = "missing_parameters" - INVALID_PARAMETER_VALUES = "invalid_parameter_values" - EMPTY_PARAMETER_VALUES = "empty_parameter_values" - INVALID_PERIOD = "invalid_period" - CONFLICTING_VALUES = "conflicting_values" - TOO_MANY_VALUES = "too_many_values" - - class LegacyPolicyTranslationError(ValueError): """Raised when committed v1 content cannot be interpreted exactly. diff --git a/tests/unit/routes/test_policy_dual_write_routes.py b/tests/unit/routes/test_policy_dual_write_routes.py index ef2aebd3e..87f5a3d52 100644 --- a/tests/unit/routes/test_policy_dual_write_routes.py +++ b/tests/unit/routes/test_policy_dual_write_routes.py @@ -13,10 +13,12 @@ from policyengine_api.routes.policy_routes import policy_bp from policyengine_api.services.policy_mirroring import PolicyMirrorUnavailableError from policyengine_api.services.policy_service import PolicySetResult -from policyengine_api.services.v2.policies.types import LegacyPolicySnapshot +from policyengine_api.services.v2.policies.types import ( + LegacyPolicySnapshot, + LegacyPolicyTranslationReason, +) from policyengine_api.services.v2.policies.validators import ( LegacyPolicyTranslationError, - LegacyPolicyTranslationReason, ) diff --git a/tests/unit/services/test_policy_mirroring.py b/tests/unit/services/test_policy_mirroring.py index 833802d63..8a1d00929 100644 --- a/tests/unit/services/test_policy_mirroring.py +++ b/tests/unit/services/test_policy_mirroring.py @@ -21,11 +21,11 @@ from policyengine_api.services.v2.policies.types import ( LegacyPolicyPersistenceResult, LegacyPolicySnapshot, + LegacyPolicyTranslationReason, ) from policyengine_api.services.v2.policies.validators import ( LegacyPolicyMappingIntegrityError, LegacyPolicyTranslationError, - LegacyPolicyTranslationReason, PolicyCatalogValidationError, PolicyContentHashCollisionError, PolicyCreationIntegrityError, diff --git a/tests/unit/services/test_user_policy_mirroring.py b/tests/unit/services/test_user_policy_mirroring.py index 2bb9242f2..de2e40cb4 100644 --- a/tests/unit/services/test_user_policy_mirroring.py +++ b/tests/unit/services/test_user_policy_mirroring.py @@ -22,11 +22,13 @@ PendingUserPolicyMirrorEvent, UserPolicyService, ) -from policyengine_api.services.v2.policies.types import LegacyPolicySnapshot +from policyengine_api.services.v2.policies.types import ( + LegacyPolicySnapshot, + LegacyPolicyTranslationReason, +) from policyengine_api.services.v2.policies.validators import ( LegacyPolicyMappingIntegrityError, LegacyPolicyTranslationError, - LegacyPolicyTranslationReason, ) from policyengine_api.services.v2.user_policies.types import ( LegacyUserPolicyPersistenceResult, @@ -150,7 +152,7 @@ def test_logging_failure_does_not_change_successful_mirror_result() -> None: ( LegacyUserPolicyIntegrityError("mapping conflict"), "integrity", - "saved_policy_mapping_conflict", + "saved_policy_integrity", ), ( LegacyPolicyMappingIntegrityError("reform mapping conflict"), diff --git a/tests/unit/v2/test_policy_legacy_translation.py b/tests/unit/v2/test_policy_legacy_translation.py index 453a60891..07f69f3ef 100644 --- a/tests/unit/v2/test_policy_legacy_translation.py +++ b/tests/unit/v2/test_policy_legacy_translation.py @@ -30,11 +30,13 @@ parse_legacy_period, translate_legacy_policy, ) -from policyengine_api.services.v2.policies.types import LegacyPolicySnapshot +from policyengine_api.services.v2.policies.types import ( + LegacyPolicySnapshot, + LegacyPolicyTranslationReason, +) from policyengine_api.services.v2.policies.validators import ( MAXIMUM_POLICY_PARAMETER_VALUES, LegacyPolicyTranslationError, - LegacyPolicyTranslationReason, parameters_missing_from_model, ) from tests.fixtures.local_v1_database import create_test_v1_schema From 0d25f44ba57d6660c24e52f3a8ec360d93abdb0d Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:54:09 +0400 Subject: [PATCH 8/8] Type the failure category too, and document a concurrent first copy failure_reason came from enums but failure_category was a plain string repeated across the copy modules and the route. PolicyCopyFailureCategory now holds every category, and the route's two no-copy records and the unexpected case are PolicyCopyFailure constants, so both logged fields come from fixed sets. The comment on the mapping-race branch and the Stage 10 runbook now say what happens when two first copies of one v1 policy run on revisions with different PolicyEngine.py versions: the mapping committed first is the copy, and the other request's v2 policy commits without a mapping, its record reporting policy_created=false. Co-Authored-By: Claude Opus 5.5 --- docs/migration/stage-10-v2-policies.md | 5 +- policyengine_api/routes/policy_routes.py | 18 ++--- policyengine_api/services/policy_mirroring.py | 70 +++++++++++++++---- .../services/user_policy_mirroring.py | 9 ++- .../services/v2/policies/services.py | 5 +- tests/unit/services/test_policy_mirroring.py | 23 ++++++ 6 files changed, 105 insertions(+), 25 deletions(-) diff --git a/docs/migration/stage-10-v2-policies.md b/docs/migration/stage-10-v2-policies.md index 0cfa2c3af..e464e9949 100644 --- a/docs/migration/stage-10-v2-policies.md +++ b/docs/migration/stage-10-v2-policies.md @@ -154,7 +154,10 @@ saved policy replays it, with earlier retained events first. A core policy whose copy failed is copied again the next time any request writes the same policy. A core policy that already has a copy returns that copy without being translated again, so writing it again after a PolicyEngine.py or country -package upgrade does not fail. Destination revision and fingerprint checks make a replay safe if +package upgrade does not fail. When two first copies of the same v1 policy run +at once on revisions with different PolicyEngine.py versions, the mapping +committed first is the copy; the other request's v2 policy commits without a +mapping and its record reports `policy_created=false`. Destination revision and fingerprint checks make a replay safe if Supabase committed but the Cloud SQL processing marker did not. There is no background processor or reconciliation process. diff --git a/policyengine_api/routes/policy_routes.py b/policyengine_api/routes/policy_routes.py index 9bb52b4e3..51694175c 100644 --- a/policyengine_api/routes/policy_routes.py +++ b/policyengine_api/routes/policy_routes.py @@ -21,7 +21,10 @@ from policyengine_api.request_context import current_request_id from policyengine_api.response_factory import _make_error_response from policyengine_api.services.policy_mirroring import ( - PolicyCopyFailureReason, + MISSING_MIRROR_REVISION_FAILURE, + MISSING_MIRROR_SNAPSHOT_FAILURE, + UNEXPECTED_FAILURE, + PolicyCopyFailure, PolicyMirrorUnavailableError, mirror_policy_after_commit, ) @@ -69,7 +72,7 @@ def _log_uncopied_v1_write( legacy_id_field: str, legacy_id: int, country_id: str, - failure_reason: PolicyCopyFailureReason, + failure: PolicyCopyFailure, ) -> None: """Record a committed v1 write that has no v2 copy, without failing the request. @@ -90,9 +93,8 @@ def _log_uncopied_v1_write( legacy_id_field: _safe_legacy_integer_id(legacy_id), "request_id": current_request_id(), "outcome": "error", - # These failures have no finer cause than their category. - "failure_category": failure_reason.value, - "failure_reason": failure_reason.value, + "failure_category": failure.category.value, + "failure_reason": failure.reason.value, }, severity="ERROR", ) @@ -114,7 +116,7 @@ def _copy_policy_to_v2( legacy_id_field="legacy_policy_id", legacy_id=policy_id, country_id=country_id, - failure_reason=PolicyCopyFailureReason.MISSING_MIRROR_SNAPSHOT, + failure=MISSING_MIRROR_SNAPSHOT_FAILURE, ) return # mirror_policy_after_commit has already logged the failure and its category. @@ -138,7 +140,7 @@ def _copy_saved_policy_events_to_v2( legacy_id_field="legacy_user_policy_id", legacy_id=legacy_user_policy_id, country_id=country_id, - failure_reason=PolicyCopyFailureReason.MISSING_MIRROR_REVISION, + failure=MISSING_MIRROR_REVISION_FAILURE, ) return try: @@ -160,7 +162,7 @@ def _copy_saved_policy_events_to_v2( legacy_id_field="legacy_user_policy_id", legacy_id=legacy_user_policy_id, country_id=country_id, - failure_reason=PolicyCopyFailureReason.UNEXPECTED, + failure=UNEXPECTED_FAILURE, ) diff --git a/policyengine_api/services/policy_mirroring.py b/policyengine_api/services/policy_mirroring.py index 26abbcd52..b71dc0d35 100644 --- a/policyengine_api/services/policy_mirroring.py +++ b/policyengine_api/services/policy_mirroring.py @@ -61,6 +61,19 @@ def _default_mirror_factory() -> LegacyPolicyMirror: return V2PolicyService(PolicyDatabaseSession(get_v2_session_factory())) +class PolicyCopyFailureCategory(StrEnum): + """The broad kinds of failed copy, as a copy record's ``failure_category``.""" + + CONFIGURATION = "configuration" + CATALOG_OR_TRANSLATION = "catalog_or_translation" + INTEGRITY = "integrity" + DATABASE = "database" + UNEXPECTED = "unexpected" + # The route logs these when no copy could be attempted. + MISSING_MIRROR_SNAPSHOT = "missing_mirror_snapshot" + MISSING_MIRROR_REVISION = "missing_mirror_revision" + + class PolicyCopyFailureReason(StrEnum): """Fixed reason codes that failed policy and saved-policy copies log. @@ -89,45 +102,74 @@ class PolicyCopyFailureReason(StrEnum): class PolicyCopyFailure: """How a failed copy is logged: a broad category and a fixed reason.""" - category: str + category: PolicyCopyFailureCategory reason: PolicyCopyFailureReason | LegacyPolicyTranslationReason +# What the route logs when it commits a v1 write but cannot attempt a copy. +MISSING_MIRROR_SNAPSHOT_FAILURE = PolicyCopyFailure( + PolicyCopyFailureCategory.MISSING_MIRROR_SNAPSHOT, + PolicyCopyFailureReason.MISSING_MIRROR_SNAPSHOT, +) +MISSING_MIRROR_REVISION_FAILURE = PolicyCopyFailure( + PolicyCopyFailureCategory.MISSING_MIRROR_REVISION, + PolicyCopyFailureReason.MISSING_MIRROR_REVISION, +) +UNEXPECTED_FAILURE = PolicyCopyFailure( + PolicyCopyFailureCategory.UNEXPECTED, PolicyCopyFailureReason.UNEXPECTED +) + + # Checked in order, so the hash collision is matched before its integrity base # class. _FAILURES_BY_ERROR_TYPE: tuple[tuple[type[Exception], PolicyCopyFailure], ...] = ( ( V2ConfigurationError, - PolicyCopyFailure("configuration", PolicyCopyFailureReason.CONFIGURATION), + PolicyCopyFailure( + PolicyCopyFailureCategory.CONFIGURATION, + PolicyCopyFailureReason.CONFIGURATION, + ), ), ( MetadataCatalogVersionNotFoundError, PolicyCopyFailure( - "catalog_or_translation", + PolicyCopyFailureCategory.CATALOG_OR_TRANSLATION, PolicyCopyFailureReason.CATALOG_VERSION_NOT_FOUND, ), ), ( PolicyCatalogValidationError, PolicyCopyFailure( - "catalog_or_translation", PolicyCopyFailureReason.CATALOG_MISMATCH + PolicyCopyFailureCategory.CATALOG_OR_TRANSLATION, + PolicyCopyFailureReason.CATALOG_MISMATCH, ), ), ( LegacyPolicyMappingIntegrityError, - PolicyCopyFailure("integrity", PolicyCopyFailureReason.MAPPING_CONFLICT), + PolicyCopyFailure( + PolicyCopyFailureCategory.INTEGRITY, + PolicyCopyFailureReason.MAPPING_CONFLICT, + ), ), ( PolicyContentHashCollisionError, - PolicyCopyFailure("integrity", PolicyCopyFailureReason.CONTENT_HASH_COLLISION), + PolicyCopyFailure( + PolicyCopyFailureCategory.INTEGRITY, + PolicyCopyFailureReason.CONTENT_HASH_COLLISION, + ), ), ( PolicyCreationIntegrityError, - PolicyCopyFailure("integrity", PolicyCopyFailureReason.STORED_POLICY_INTEGRITY), + PolicyCopyFailure( + PolicyCopyFailureCategory.INTEGRITY, + PolicyCopyFailureReason.STORED_POLICY_INTEGRITY, + ), ), ( SQLAlchemyError, - PolicyCopyFailure("database", PolicyCopyFailureReason.DATABASE_ERROR), + PolicyCopyFailure( + PolicyCopyFailureCategory.DATABASE, PolicyCopyFailureReason.DATABASE_ERROR + ), ), ) @@ -139,18 +181,22 @@ def classify_policy_copy_failure(error: Exception) -> PolicyCopyFailure: its fixed reason, so a log record never holds policy content or error text. """ if isinstance(error, LegacyPolicyTranslationError): - return PolicyCopyFailure("catalog_or_translation", error.reason) + return PolicyCopyFailure( + PolicyCopyFailureCategory.CATALOG_OR_TRANSLATION, error.reason + ) if isinstance(error, MetadataCatalogUnavailableError): reason = ( PolicyCopyFailureReason.CATALOG_QUERY_FAILED if isinstance(error.__cause__, SQLAlchemyError) else PolicyCopyFailureReason.CATALOG_ABSENT ) - return PolicyCopyFailure("catalog_or_translation", reason) + return PolicyCopyFailure( + PolicyCopyFailureCategory.CATALOG_OR_TRANSLATION, reason + ) for error_type, failure in _FAILURES_BY_ERROR_TYPE: if isinstance(error, error_type): return failure - return PolicyCopyFailure("unexpected", PolicyCopyFailureReason.UNEXPECTED) + return UNEXPECTED_FAILURE @dataclass(frozen=True) @@ -191,7 +237,7 @@ def _log_mirror_event( else None ), "outcome": mirror_outcome.outcome, - "failure_category": failure.category if failure else None, + "failure_category": failure.category.value if failure else None, "failure_reason": failure.reason.value if failure else None, "policy_created": mirror_outcome.policy_created, "mapping_created": mirror_outcome.mapping_created, diff --git a/policyengine_api/services/user_policy_mirroring.py b/policyengine_api/services/user_policy_mirroring.py index ecd69c59b..d72d7f7f5 100644 --- a/policyengine_api/services/user_policy_mirroring.py +++ b/policyengine_api/services/user_policy_mirroring.py @@ -10,6 +10,7 @@ from policyengine_api.gcp_logging import logger from policyengine_api.services.policy_mirroring import ( PolicyCopyFailure, + PolicyCopyFailureCategory, PolicyCopyFailureReason, classify_policy_copy_failure, ) @@ -70,11 +71,13 @@ def _default_mirror_factory() -> LegacyUserPolicyMirror: def _classify_failure(error: Exception) -> PolicyCopyFailure: if isinstance(error, LegacyUserPolicyIntegrityError): return PolicyCopyFailure( - "integrity", PolicyCopyFailureReason.SAVED_POLICY_INTEGRITY + PolicyCopyFailureCategory.INTEGRITY, + PolicyCopyFailureReason.SAVED_POLICY_INTEGRITY, ) if isinstance(error, ReformPolicyUnavailableError): return PolicyCopyFailure( - "unexpected", PolicyCopyFailureReason.MISSING_REFORM_POLICY + PolicyCopyFailureCategory.UNEXPECTED, + PolicyCopyFailureReason.MISSING_REFORM_POLICY, ) return classify_policy_copy_failure(error) @@ -143,7 +146,7 @@ def _log_mirror_operation( str(result.policy_id) if result is not None else None ), "outcome": mirror_outcome.outcome, - "failure_category": failure.category if failure else None, + "failure_category": failure.category.value if failure else None, "failure_reason": failure.reason.value if failure else None, "association_created": ( result.association_created if result is not None else None diff --git a/policyengine_api/services/v2/policies/services.py b/policyengine_api/services/v2/policies/services.py index dd7cc2925..eb8250fd1 100644 --- a/policyengine_api/services/v2/policies/services.py +++ b/policyengine_api/services/v2/policies/services.py @@ -262,7 +262,10 @@ def mirror_legacy_policy_in_session( raise LegacyPolicyMappingIntegrityError(msg) # A concurrent copy mapped the policy first. As with a mapping found at the # start, the same source hash makes its v2 policy the copy, even when it was - # translated with another catalog version than this one. + # translated with another catalog version than this one. If this copy's + # translation differed, the v2 policy it inserted commits without a mapping, + # and the result's policy_created is false because it reports the mapped + # policy, not that one. verify_legacy_policy_mapping( concurrent, source_policy_hash=snapshot.source_policy_hash ) diff --git a/tests/unit/services/test_policy_mirroring.py b/tests/unit/services/test_policy_mirroring.py index 8a1d00929..8459e0a0f 100644 --- a/tests/unit/services/test_policy_mirroring.py +++ b/tests/unit/services/test_policy_mirroring.py @@ -15,7 +15,10 @@ ) from policyengine_api.data.v2.settings import V2ConfigurationError from policyengine_api.services.policy_mirroring import ( + PolicyCopyFailureCategory, + PolicyCopyFailureReason, PolicyMirrorUnavailableError, + classify_policy_copy_failure, mirror_policy_after_commit, ) from policyengine_api.services.v2.policies.types import ( @@ -244,3 +247,23 @@ def test_each_failed_copy_logs_a_fixed_reason_with_its_category( assert type(payload["failure_reason"]) is str assert "secret" not in repr(payload) assert "credential" not in repr(payload) + + +@pytest.mark.parametrize( + "error", + [ + V2ConfigurationError("secret url"), + MetadataCatalogUnavailableError("catalog absent"), + _translation_failure(LegacyPolicyTranslationReason.MISSING_PARAMETERS), + LegacyPolicyMappingIntegrityError("secret"), + PoolTimeoutError("pool timeout"), + RuntimeError("policy content secret"), + ], +) +def test_failure_category_and_reason_come_from_fixed_sets(error: Exception) -> None: + failure = classify_policy_copy_failure(error) + + assert isinstance(failure.category, PolicyCopyFailureCategory) + assert isinstance( + failure.reason, (PolicyCopyFailureReason, LegacyPolicyTranslationReason) + )