Copy reused and older v1 policies to v2, reuse existing copies, and log why a copy fails

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. 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`.

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
@@ -147,7 +152,12 @@ 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. 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.
 
@@ -168,7 +178,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`, `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`, `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` |
+| `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.
@@ -177,7 +199,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

policyengine_api/routes/policy_routes.py

@@ -21,6 +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 (
+    MISSING_MIRROR_REVISION_FAILURE,
+    MISSING_MIRROR_SNAPSHOT_FAILURE,
+    UNEXPECTED_FAILURE,
+    PolicyCopyFailure,
     PolicyMirrorUnavailableError,
     mirror_policy_after_commit,
 )
@@ -68,7 +72,7 @@ def _log_uncopied_v1_write(
     legacy_id_field: str,
     legacy_id: int,
     country_id: str,
-    failure_category: str,
+    failure: PolicyCopyFailure,
 ) -> None:
     """Record a committed v1 write that has no v2 copy, without failing the request.
 
@@ -89,7 +93,8 @@ 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,
+                "failure_category": failure.category.value,
+                "failure_reason": failure.reason.value,
             },
             severity="ERROR",
         )
@@ -111,7 +116,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=MISSING_MIRROR_SNAPSHOT_FAILURE,
         )
         return
     # mirror_policy_after_commit has already logged the failure and its category.
@@ -135,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_category="missing_mirror_revision",
+            failure=MISSING_MIRROR_REVISION_FAILURE,
         )
         return
     try:
@@ -157,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_category="unexpected",
+            failure=UNEXPECTED_FAILURE,
         )
 
 

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
@@ -29,6 +30,7 @@
     from policyengine_api.services.v2.policies.types import (
         LegacyPolicyPersistenceResult,
         LegacyPolicySnapshot,
+        LegacyPolicyTranslationReason,
     )
 
 
@@ -59,31 +61,142 @@ 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,
+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.
+
+    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"
+    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"
+    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"
+
+
+@dataclass(frozen=True)
+class PolicyCopyFailure:
+    """How a failed copy is logged: a broad category and a fixed reason."""
+
+    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(
+            PolicyCopyFailureCategory.CONFIGURATION,
+            PolicyCopyFailureReason.CONFIGURATION,
+        ),
+    ),
+    (
+        MetadataCatalogVersionNotFoundError,
+        PolicyCopyFailure(
+            PolicyCopyFailureCategory.CATALOG_OR_TRANSLATION,
+            PolicyCopyFailureReason.CATALOG_VERSION_NOT_FOUND,
+        ),
+    ),
+    (
+        PolicyCatalogValidationError,
+        PolicyCopyFailure(
+            PolicyCopyFailureCategory.CATALOG_OR_TRANSLATION,
+            PolicyCopyFailureReason.CATALOG_MISMATCH,
         ),
-    ):
-        return "catalog_or_translation"
-    if isinstance(
-        error,
-        (
-            LegacyPolicyMappingIntegrityError,
-            PolicyContentHashCollisionError,
-            PolicyCreationIntegrityError,
+    ),
+    (
+        LegacyPolicyMappingIntegrityError,
+        PolicyCopyFailure(
+            PolicyCopyFailureCategory.INTEGRITY,
+            PolicyCopyFailureReason.MAPPING_CONFLICT,
         ),
-    ):
-        return "integrity"
-    if isinstance(error, SQLAlchemyError):
-        return "database"
-    return "unexpected"
+    ),
+    (
+        PolicyContentHashCollisionError,
+        PolicyCopyFailure(
+            PolicyCopyFailureCategory.INTEGRITY,
+            PolicyCopyFailureReason.CONTENT_HASH_COLLISION,
+        ),
+    ),
+    (
+        PolicyCreationIntegrityError,
+        PolicyCopyFailure(
+            PolicyCopyFailureCategory.INTEGRITY,
+            PolicyCopyFailureReason.STORED_POLICY_INTEGRITY,
+        ),
+    ),
+    (
+        SQLAlchemyError,
+        PolicyCopyFailure(
+            PolicyCopyFailureCategory.DATABASE, PolicyCopyFailureReason.DATABASE_ERROR
+        ),
+    ),
+)
+
+
+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 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(
+            PolicyCopyFailureCategory.CATALOG_OR_TRANSLATION, reason
+        )
+    for error_type, failure in _FAILURES_BY_ERROR_TYPE:
+        if isinstance(error, error_type):
+            return failure
+    return UNEXPECTED_FAILURE
 
 
 @dataclass(frozen=True)
@@ -93,7 +206,7 @@ class _MirrorOutcome:
     outcome: str
     actual_write_sources: list[str]
     destination_policy_id: object | None = None
-    failure_category: str | None = None
+    failure: PolicyCopyFailure | None = None
     policy_created: bool | None = None
     mapping_created: bool | None = None
 
@@ -105,6 +218,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(
@@ -123,7 +237,8 @@ def _log_mirror_event(
                     else None
                 ),
                 "outcome": mirror_outcome.outcome,
-                "failure_category": mirror_outcome.failure_category,
+                "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,
                 "duration_ms": round((time.perf_counter() - started_at) * 1000, 3),
@@ -154,7 +269,7 @@ def mirror_policy_after_commit(
             mirror_outcome=_MirrorOutcome(
                 outcome="error",
                 actual_write_sources=["cloud_sql"],
-                failure_category=_failure_category(error),
+                failure=classify_policy_copy_failure(error),
             ),
         )
         msg = "The committed v1 policy could not be mirrored to v2"