From ef042054cda2d7963e63b92631583c885dcb560f Mon Sep 17 00:00:00 2001 From: Chris Busillo Date: Sun, 16 Aug 2026 16:48:10 -0400 Subject: [PATCH] Add digest-bound generic web recovery apply --- .../contracts/generic_web_deploy_recovery.py | 28 + .../generic_web_deploy_provider_adapter.py | 11 + .../generic_web_deploy_recovery_http.py | 1027 ++++++++++++----- control_plane/http_routes/generic_web.py | 35 + control_plane/provider_operations.py | 62 + docs/operations.md | 28 +- frontend/generated/openapi-canonical.json | 252 ++++ tests/test_generic_web_deploy_recovery.py | 479 ++++++++ tests/test_http_write_route_registrars.py | 29 +- tests/test_provider_operations.py | 135 +++ 10 files changed, 1774 insertions(+), 312 deletions(-) diff --git a/control_plane/contracts/generic_web_deploy_recovery.py b/control_plane/contracts/generic_web_deploy_recovery.py index 1f40dbb01..9fa8e0e3c 100644 --- a/control_plane/contracts/generic_web_deploy_recovery.py +++ b/control_plane/contracts/generic_web_deploy_recovery.py @@ -51,6 +51,15 @@ def _validate_request(self) -> GenericWebDeployRecoveryDryRunRequest: return self +class GenericWebDeployRecoveryApplyRequest(GenericWebDeployRecoveryDryRunRequest): + expected_recovery_digest: str = Field(pattern=r"^[0-9a-f]{64}$") + + @model_validator(mode="after") + def _validate_apply_request(self) -> "GenericWebDeployRecoveryApplyRequest": + self.expected_recovery_digest = self.expected_recovery_digest.strip() + return self + + class GenericWebDeployRecoveryDryRunResponse(BaseModel): model_config = ConfigDict(extra="forbid") @@ -76,6 +85,25 @@ class GenericWebDeployRecoveryDryRunResponse(BaseModel): recovery_digest: str = Field(pattern=r"^[0-9a-f]{64}$") +class GenericWebDeployRecoveryApplyResponse(BaseModel): + model_config = ConfigDict(extra="forbid") + + schema_version: Literal[1] = 1 + status: Literal["accepted"] = "accepted" + mode: Literal["apply"] = "apply" + trace_id: str + product: str + context: str + instance: str + reservation_state: Literal["completed", "reconcile_required"] + reservation_attempt: int = Field(ge=1) + recovery_action: GenericWebDeployRecoveryAction + recovery_digest: str = Field(pattern=r"^[0-9a-f]{64}$") + provider_outcome: GenericWebDeployRecoveryProviderOutcome + provider_status: str = Field(default="", max_length=128) + retry_safe: bool + + def generic_web_deploy_recovery_identifier_sha256(value: str) -> str: return hashlib.sha256(value.strip().encode("utf-8")).hexdigest() diff --git a/control_plane/generic_web_deploy_provider_adapter.py b/control_plane/generic_web_deploy_provider_adapter.py index 58ac087aa..91c152a51 100644 --- a/control_plane/generic_web_deploy_provider_adapter.py +++ b/control_plane/generic_web_deploy_provider_adapter.py @@ -119,6 +119,17 @@ def observe( provider_effect_phase=provider_effect_phase, reconciliation_key=reconciliation_key, ) + return self.provider_observation_from_inspection( + provider_operation_key=provider_operation_key, + inspection=inspection, + ) + + def provider_observation_from_inspection( + self, + *, + provider_operation_key: str, + inspection: _GenericWebDeployProviderInspection, + ) -> ProviderObservation: observation = inspection.observation if observation.outcome != "present" or inspection.resolved_deploy_target is None: return ProviderObservation( diff --git a/control_plane/generic_web_deploy_recovery_http.py b/control_plane/generic_web_deploy_recovery_http.py index 2e2e716a6..dedee959e 100644 --- a/control_plane/generic_web_deploy_recovery_http.py +++ b/control_plane/generic_web_deploy_recovery_http.py @@ -9,13 +9,18 @@ from control_plane.contracts.generic_web_deploy_recovery import ( GenericWebDeployRecoveryAction, + GenericWebDeployRecoveryApplyRequest, + GenericWebDeployRecoveryApplyResponse, GenericWebDeployRecoveryDryRunRequest, GenericWebDeployRecoveryDryRunResponse, GenericWebDeployRecoveryProviderOutcome, build_generic_web_deploy_recovery_digest, generic_web_deploy_recovery_identifier_sha256, ) -from control_plane.contracts.idempotency_record import parse_launchplane_mutation_timestamp +from control_plane.contracts.idempotency_record import ( + LaunchplaneIdempotencyRecord, + parse_launchplane_mutation_timestamp, +) from control_plane.generic_web_deploy_http import ( GENERIC_WEB_DEPLOY_ROUTE, GenericWebDeployProductMismatchError, @@ -26,7 +31,13 @@ GenericWebDeployProviderMutationAdapter, ) from control_plane.http_routes.support import AuthorizationAllows, HttpErrorFactory -from control_plane.provider_operations import build_provider_operation_key +from control_plane.provider_operations import ( + ProviderMutationOutcome, + ProviderObservation, + ProviderOperationLease, + build_provider_operation_key, + resume_acquired_provider_operation, +) from control_plane.service_auth import AuthorizationTarget, LaunchplaneIdentity from control_plane.storage.postgres import PostgresRecordStore from control_plane.workflows.generic_web_deploy import ( @@ -41,10 +52,13 @@ GENERIC_WEB_DEPLOY_RECOVERY_DRY_RUN_ROUTE = "/v1/admin/generic-web/deploy-recovery/dry-run" +GENERIC_WEB_DEPLOY_RECOVERY_APPLY_ROUTE = "/v1/admin/generic-web/deploy-recovery/apply" __all__ = [ "GENERIC_WEB_DEPLOY_RECOVERY_DRY_RUN_ROUTE", + "GENERIC_WEB_DEPLOY_RECOVERY_APPLY_ROUTE", "GenericWebDeployRecoveryDependencies", + "build_generic_web_deploy_recovery_apply_handler", "build_generic_web_deploy_recovery_dry_run_handler", ] @@ -64,370 +78,771 @@ def _bounded_recovery_value(value: object, *, limit: int = 128) -> str: return str(value or "").strip()[:limit] -def build_generic_web_deploy_recovery_dry_run_handler( - *, dependencies: GenericWebDeployRecoveryDependencies -) -> Callable[..., Any]: - async def dry_run_generic_web_deploy_recovery( - request: Request, - recovery_request: GenericWebDeployRecoveryDryRunRequest, - identity: Annotated[LaunchplaneIdentity, Depends(dependencies.read_write_identity)], - record_store: Annotated[object, Depends(dependencies.get_record_store)], - idempotency_key: Annotated[str, Header(alias="Idempotency-Key")], - ) -> GenericWebDeployRecoveryDryRunResponse: - trace_id = dependencies.next_trace_id() +@dataclass(frozen=True, slots=True) +class _GenericWebDeployRecoveryInspection: + request: GenericWebDeployRecoveryDryRunRequest + product: str + context: str + instance: str + reservation: LaunchplaneIdempotencyRecord + observed_at: str + original_fingerprint: str + idempotency_key: str + provider_operation_key: str + provider_outcome: GenericWebDeployRecoveryProviderOutcome + provider_status: str + retry_safe: bool + proposed_action: GenericWebDeployRecoveryAction + provider_observation_payload: dict[str, object] + adapter: GenericWebDeployProviderMutationAdapter | None = None + provider_inspection: Any | None = None + + @property + def recovery_digest(self) -> str: + return build_generic_web_deploy_recovery_digest(self.digest_payload()) + + def digest_payload(self) -> dict[str, object]: + return { + "schema_version": 1, + "request": self.request.model_dump( + mode="json", + exclude={"expected_recovery_digest"}, + ), + "original_route": GENERIC_WEB_DEPLOY_ROUTE, + "idempotency_key": self.idempotency_key, + "request_fingerprint": self.original_fingerprint, + "reservation": self.reservation.model_dump(mode="json"), + "provider_operation_key": self.provider_operation_key, + "provider_observation": self.provider_observation_payload, + "retry_safe": self.retry_safe, + "proposed_action": self.proposed_action, + } + + def dry_run_response(self) -> GenericWebDeployRecoveryDryRunResponse: + return GenericWebDeployRecoveryDryRunResponse( + product=self.product, + context=self.context, + instance=self.instance, + reservation_state=self.reservation.state, + reservation_attempt=self.reservation.attempt, + reservation_created_at=self.reservation.created_at, + reservation_updated_at=self.reservation.updated_at, + reservation_lease_expires_at=self.reservation.lease_expires_at, + observed_at=self.observed_at, + reconciliation_key_sha256=generic_web_deploy_recovery_identifier_sha256( + self.reservation.reconciliation_key + ), + provider_target_key_sha256=generic_web_deploy_recovery_identifier_sha256( + self.reservation.provider_target_key + ), + provider_effect_phase=_bounded_recovery_value(self.reservation.provider_effect_phase), + provider_outcome=self.provider_outcome, + provider_status=self.provider_status, + retry_safe=self.retry_safe, + proposed_action=self.proposed_action, + recovery_digest=self.recovery_digest, + ) + + +class _GenericWebRecoveryApplyAdapter: + def __init__( + self, + *, + delegate: GenericWebDeployProviderMutationAdapter, + recovery_metadata: dict[str, object], + ) -> None: + self._delegate = delegate + self._recovery_metadata = recovery_metadata + + def target_key(self) -> str: + return self._delegate.target_key() + + def reconciliation_key(self) -> str: + return self._delegate.reconciliation_key() + + def observe( + self, + provider_operation_key: str, + provider_effect_phase: str, + reconciliation_key: str, + ) -> ProviderObservation: + return self._delegate.observe( + provider_operation_key, + provider_effect_phase, + reconciliation_key, + ) + + def apply( + self, provider_operation_key: str, lease: ProviderOperationLease + ) -> ProviderMutationOutcome: + outcome = self._delegate.apply(provider_operation_key, lease) + response_payload = dict(outcome.response_payload) + response_payload["recovery"] = self._recovery_metadata + return ProviderMutationOutcome( + response_status_code=outcome.response_status_code, + response_payload=response_payload, + durable=outcome.durable, + provider_effect_performed=outcome.provider_effect_performed, + ) + + +def _stored_recovery_metadata( + reservation: LaunchplaneIdempotencyRecord, +) -> tuple[str, GenericWebDeployRecoveryAction] | None: + stored_recovery = reservation.response_payload.get("recovery") + if not isinstance(stored_recovery, dict) or set(stored_recovery) != { + "schema_version", + "recovery_digest", + "recovery_action", + }: + return None + if stored_recovery.get("schema_version") != 1: + return None + digest = str(stored_recovery.get("recovery_digest") or "").strip() + action = str(stored_recovery.get("recovery_action") or "").strip() + if ( + len(digest) != 64 + or any(character not in "0123456789abcdef" for character in digest) + or action + not in { + "replay_completed", + "wait_for_active_lease", + "adopt_observed", + "retry_original_operation", + "hold_unknown", + } + ): + return None + return digest, cast(GenericWebDeployRecoveryAction, action) + + +def _recovery_metadata( + inspection: _GenericWebDeployRecoveryInspection, +) -> dict[str, object]: + return { + "schema_version": 1, + "recovery_digest": inspection.recovery_digest, + "recovery_action": inspection.proposed_action, + } + + +def _apply_response( + *, + trace_id: str, + inspection: _GenericWebDeployRecoveryInspection, + reservation: LaunchplaneIdempotencyRecord, +) -> GenericWebDeployRecoveryApplyResponse: + metadata = _stored_recovery_metadata(reservation) + action = inspection.proposed_action + digest = inspection.recovery_digest + if metadata is not None: + digest, action = metadata + if reservation.state not in {"completed", "reconcile_required"}: + raise RuntimeError("Generic web recovery apply response requires terminal state.") + return GenericWebDeployRecoveryApplyResponse( + trace_id=trace_id, + product=inspection.product, + context=inspection.context, + instance=inspection.instance, + reservation_state=reservation.state, + reservation_attempt=reservation.attempt, + recovery_action=action, + recovery_digest=digest, + provider_outcome=inspection.provider_outcome, + provider_status=inspection.provider_status, + retry_safe=inspection.retry_safe, + ) + + +def _apply_replay_response( + *, + trace_id: str, + request: GenericWebDeployRecoveryApplyRequest, + reservation: LaunchplaneIdempotencyRecord, + context: str, +) -> GenericWebDeployRecoveryApplyResponse: + metadata = _stored_recovery_metadata(reservation) + if metadata is None: + raise RuntimeError("Recovered generic web reservation replay requires metadata.") + digest, action = metadata + return GenericWebDeployRecoveryApplyResponse( + trace_id=trace_id, + product=request.product, + context=context, + instance=request.instance, + reservation_state="completed", + reservation_attempt=reservation.attempt, + recovery_action=action, + recovery_digest=digest, + provider_outcome="not_inspected", + provider_status="", + retry_safe=False, + ) + + +def _transition_expired_running_recovery( + *, + inspection: _GenericWebDeployRecoveryInspection, + store: PostgresRecordStore, + trace_id: str, + dependencies: GenericWebDeployRecoveryDependencies, +) -> _GenericWebDeployRecoveryInspection: + if inspection.reservation.state != "running": + return inspection + transition = store.mark_mutation_reconcile_required( + reservation=inspection.reservation, + reconciliation_key=inspection.reservation.reconciliation_key, + ) + if transition.status != "updated" or transition.record is None: + raise dependencies.http_error( + status_code=409, + trace_id=trace_id, + code="reservation_changed", + message="Generic web deploy recovery reservation changed before apply.", + ) + return _GenericWebDeployRecoveryInspection( + request=inspection.request, + product=inspection.product, + context=inspection.context, + instance=inspection.instance, + reservation=transition.record, + observed_at=inspection.observed_at, + original_fingerprint=inspection.original_fingerprint, + idempotency_key=inspection.idempotency_key, + provider_operation_key=inspection.provider_operation_key, + provider_outcome=inspection.provider_outcome, + provider_status=inspection.provider_status, + retry_safe=inspection.retry_safe, + proposed_action=inspection.proposed_action, + provider_observation_payload=inspection.provider_observation_payload, + adapter=inspection.adapter, + provider_inspection=inspection.provider_inspection, + ) + + +async def _inspect_generic_web_deploy_recovery( + *, + request: Request, + recovery_request: GenericWebDeployRecoveryDryRunRequest, + identity: LaunchplaneIdentity, + record_store: object, + idempotency_key: str, + trace_id: str, + dependencies: GenericWebDeployRecoveryDependencies, +) -> _GenericWebDeployRecoveryInspection: + try: + profile, lane = resolve_generic_web_deploy_lane( + record_store=record_store, + product=recovery_request.product, + instance=recovery_request.instance, + ) + except GenericWebDeployRouteDependencyError as error: + raise dependencies.http_error( + status_code=503, + trace_id=trace_id, + code="storage_unavailable", + message="Generic web deploy recovery requires database-backed profile storage.", + ) from error + except GenericWebDeployProductMismatchError as error: + raise dependencies.http_error( + status_code=403, + trace_id=trace_id, + code="product_driver_mismatch", + message="Product is not configured for generic web deploy recovery.", + ) from error + except (ValueError, click.ClickException) as error: + raise dependencies.http_error( + status_code=400, + trace_id=trace_id, + code="invalid_request", + message="Request could not be completed.", + ) from error + if not dependencies.authorization_allows( + identity=identity, + action="generic_web_deploy.execute", + product=profile.product, + context=lane.context, + target=AuthorizationTarget(scope="instance", instances=(lane.instance,)), + ): + raise dependencies.http_error( + status_code=403, + trace_id=trace_id, + code="authorization_denied", + message=( + "Identity cannot inspect generic web deploy recovery for the requested " + "product/context." + ), + ) + normalized_key = idempotency_key.strip() + if not normalized_key: + raise dependencies.http_error( + status_code=400, + trace_id=trace_id, + code="idempotency_key_required", + message="Generic web deploy recovery requires an Idempotency-Key header.", + ) + if not isinstance(record_store, PostgresRecordStore): + raise dependencies.http_error( + status_code=503, + trace_id=trace_id, + code="database_storage_required", + message="Generic web deploy recovery requires database storage.", + ) + raw_payload = await request.json() + original_payload = raw_payload.get("original_deploy") if isinstance(raw_payload, dict) else None + if not isinstance(original_payload, dict): + raise dependencies.http_error( + status_code=400, + trace_id=trace_id, + code="invalid_request", + message="Generic web deploy recovery requires the exact original deploy payload.", + ) + original_fingerprint = dependencies.idempotency_request_fingerprint( + route_path=GENERIC_WEB_DEPLOY_ROUTE, + payload=cast(dict[str, object], original_payload), + ) + lookup = record_store.lookup_existing_mutation_reservation( + route_path=GENERIC_WEB_DEPLOY_ROUTE, + idempotency_key=normalized_key, + request_fingerprint=original_fingerprint, + ) + if lookup.status == "missing": + raise dependencies.http_error( + status_code=404, + trace_id=trace_id, + code="reservation_not_found", + message="No matching generic web deploy reservation exists.", + ) + if lookup.status in {"conflict", "ambiguous"}: + raise dependencies.http_error( + status_code=409, + trace_id=trace_id, + code=f"reservation_{lookup.status}", + message="Generic web deploy recovery reservation identity is not unique and exact.", + ) + if lookup.status == "hold_unknown": + raise dependencies.http_error( + status_code=409, + trace_id=trace_id, + code="hold_unknown", + message="Generic web deploy recovery lease timing is not authoritative.", + ) + reservation = lookup.record + if reservation is None: + raise dependencies.http_error( + status_code=503, + trace_id=trace_id, + code="reservation_lookup_failed", + message="Generic web deploy recovery could not inspect the reservation.", + ) + if ( + reservation.route_path != GENERIC_WEB_DEPLOY_ROUTE + or reservation.idempotency_key != normalized_key + or reservation.request_fingerprint != original_fingerprint + ): + raise dependencies.http_error( + status_code=409, + trace_id=trace_id, + code="reservation_conflict", + message="Generic web deploy recovery reservation identity is not exact.", + ) + + provider_outcome: GenericWebDeployRecoveryProviderOutcome = "not_inspected" + provider_status = "" + retry_safe = False + proposed_action: GenericWebDeployRecoveryAction = "hold_unknown" + provider_operation_key = "" + provider_observation_payload: dict[str, object] = {"outcome": provider_outcome} + operation_product = recovery_request.original_deploy.product.strip() + operation_context = "" + operation_instance = recovery_request.original_deploy.deploy.instance.strip() + authoritative_lane = lane + + if reservation.state == "completed": + stored_result = reservation.response_payload.get("result") try: - profile, lane = resolve_generic_web_deploy_lane( - record_store=record_store, - product=recovery_request.product, - instance=recovery_request.instance, - ) - except GenericWebDeployRouteDependencyError as error: - raise dependencies.http_error( - status_code=503, - trace_id=trace_id, - code="storage_unavailable", - message="Generic web deploy recovery requires database-backed profile storage.", - ) from error - except GenericWebDeployProductMismatchError as error: + completed_result = GenericWebDeployResult.model_validate(stored_result) + except ValidationError as error: raise dependencies.http_error( - status_code=403, - trace_id=trace_id, - code="product_driver_mismatch", - message="Product is not configured for generic web deploy recovery.", - ) from error - except (ValueError, click.ClickException) as error: - raise dependencies.http_error( - status_code=400, + status_code=409, trace_id=trace_id, - code="invalid_request", - message="Request could not be completed.", + code="reservation_target_conflict", + message="Completed generic web deploy recovery context is not authoritative.", ) from error - if not dependencies.authorization_allows( - identity=identity, - action="generic_web_deploy.execute", - product=profile.product, - context=lane.context, - target=AuthorizationTarget(scope="instance", instances=(lane.instance,)), + completed_product = completed_result.product.strip() + completed_context = completed_result.context.strip() + completed_instance = completed_result.instance.strip() + if ( + not completed_product + or not completed_context + or not completed_instance + or completed_product != operation_product + or completed_instance != operation_instance ): raise dependencies.http_error( - status_code=403, - trace_id=trace_id, - code="authorization_denied", - message=( - "Identity cannot inspect generic web deploy recovery for the requested " - "product/context." - ), - ) - normalized_key = idempotency_key.strip() - if not normalized_key: - raise dependencies.http_error( - status_code=400, - trace_id=trace_id, - code="idempotency_key_required", - message="Generic web deploy recovery requires an Idempotency-Key header.", - ) - if not isinstance(record_store, PostgresRecordStore): - raise dependencies.http_error( - status_code=503, + status_code=409, trace_id=trace_id, - code="database_storage_required", - message="Generic web deploy recovery requires database storage.", + code="reservation_target_conflict", + message="Completed generic web deploy recovery context is not authoritative.", ) - raw_payload = await request.json() - original_payload = ( - raw_payload.get("original_deploy") if isinstance(raw_payload, dict) else None - ) - if not isinstance(original_payload, dict): + operation_context = completed_context + proposed_action = "replay_completed" + provider_status = _bounded_recovery_value(completed_result.deploy_status) + else: + if not reservation.reconciliation_key or not reservation.provider_target_key: raise dependencies.http_error( - status_code=400, + status_code=409, trace_id=trace_id, - code="invalid_request", - message="Generic web deploy recovery requires the exact original deploy payload.", + code="reservation_target_conflict", + message="Stored generic web deploy recovery target identity is incomplete.", ) - original_fingerprint = dependencies.idempotency_request_fingerprint( - route_path=GENERIC_WEB_DEPLOY_ROUTE, - payload=cast(dict[str, object], original_payload), - ) - lookup = record_store.lookup_existing_mutation_reservation( - route_path=GENERIC_WEB_DEPLOY_ROUTE, - idempotency_key=normalized_key, - request_fingerprint=original_fingerprint, - ) - if lookup.status == "missing": - raise dependencies.http_error( - status_code=404, - trace_id=trace_id, - code="reservation_not_found", - message="No matching generic web deploy reservation exists.", + try: + stored_target = decode_generic_web_provider_reconciliation_target( + reservation.reconciliation_key ) - if lookup.status in {"conflict", "ambiguous"}: + except ValueError as error: raise dependencies.http_error( status_code=409, trace_id=trace_id, - code=f"reservation_{lookup.status}", - message="Generic web deploy recovery reservation identity is not unique and exact.", - ) - if lookup.status == "hold_unknown": + code="reservation_target_conflict", + message="Stored generic web deploy recovery target identity is invalid.", + ) from error + operation_context = stored_target.context.strip() + stored_instance = stored_target.instance.strip() + stored_product = stored_target.product.strip() + legacy_snapshot_without_product = "product" not in stored_target.model_fields_set + if ( + not operation_context + or (not legacy_snapshot_without_product and stored_product != operation_product) + or stored_instance != operation_instance + ): raise dependencies.http_error( status_code=409, trace_id=trace_id, - code="hold_unknown", - message="Generic web deploy recovery lease timing is not authoritative.", + code="reservation_target_conflict", + message="Stored generic web deploy recovery target identity conflicts.", ) - reservation = lookup.record - if reservation is None: + authoritative_lane = lane.model_copy( + update={"context": operation_context, "instance": stored_instance} + ) + try: + resolved_stored_target = resolve_generic_web_provider_reconciliation_target( + reconciliation_key=reservation.reconciliation_key, + request_artifact_id=recovery_request.original_deploy.deploy.artifact_id, + request_source_git_ref=recovery_request.original_deploy.deploy.source_git_ref, + request_timeout_seconds=recovery_request.original_deploy.deploy.timeout_seconds, + request_no_cache=recovery_request.original_deploy.deploy.no_cache, + normalized_artifact_id=normalize_generic_web_artifact_id( + profile=profile, + artifact_id=recovery_request.original_deploy.deploy.artifact_id, + ), + request_deploy_reference=recovery_request.original_deploy.deploy.deploy_reference, + lane=authoritative_lane, + ) + except (ValueError, click.ClickException) as error: raise dependencies.http_error( - status_code=503, + status_code=409, trace_id=trace_id, - code="reservation_lookup_failed", - message="Generic web deploy recovery could not inspect the reservation.", - ) + code="reservation_target_conflict", + message="Stored generic web deploy recovery target identity is invalid.", + ) from error if ( - reservation.route_path != GENERIC_WEB_DEPLOY_ROUTE - or reservation.idempotency_key != normalized_key - or reservation.request_fingerprint != original_fingerprint + resolved_stored_target.ship_request.context != operation_context + or resolved_stored_target.ship_request.instance != stored_instance + or build_generic_web_provider_target_key(resolved_stored_target) + != reservation.provider_target_key ): raise dependencies.http_error( status_code=409, trace_id=trace_id, - code="reservation_conflict", - message="Generic web deploy recovery reservation identity is not exact.", + code="reservation_target_conflict", + message="Stored generic web deploy recovery target identity conflicts.", ) - provider_outcome: GenericWebDeployRecoveryProviderOutcome = "not_inspected" - provider_status = "" - retry_safe = False - proposed_action: GenericWebDeployRecoveryAction = "hold_unknown" - provider_operation_key = "" - provider_observation_payload: dict[str, object] = {"outcome": provider_outcome} - operation_product = recovery_request.original_deploy.product.strip() - operation_context = "" - operation_instance = recovery_request.original_deploy.deploy.instance.strip() - authoritative_lane = lane - if reservation.state == "completed": - stored_result = reservation.response_payload.get("result") + if not dependencies.authorization_allows( + identity=identity, + action="generic_web_deploy.execute", + product=operation_product, + context=operation_context, + target=AuthorizationTarget(scope="instance", instances=(operation_instance,)), + ): + raise dependencies.http_error( + status_code=403, + trace_id=trace_id, + code="authorization_denied", + message=( + "Identity cannot inspect generic web deploy recovery for the stored " + "product/context." + ), + ) + + adapter = None + provider_inspection = None + if reservation.state != "completed": + lease_is_active = False + if reservation.state == "running": try: - completed_result = GenericWebDeployResult.model_validate(stored_result) - except ValidationError as error: + lease_is_active = parse_launchplane_mutation_timestamp( + reservation.lease_expires_at, + field_name="lease_expires_at", + ) > parse_launchplane_mutation_timestamp( + lookup.observed_at, + field_name="observed_at", + ) + except ValueError as error: raise dependencies.http_error( status_code=409, trace_id=trace_id, - code="reservation_target_conflict", - message="Completed generic web deploy recovery context is not authoritative.", + code="hold_unknown", + message="Generic web deploy recovery lease timing is not authoritative.", ) from error - completed_product = completed_result.product.strip() - completed_context = completed_result.context.strip() - completed_instance = completed_result.instance.strip() - if ( - not completed_product - or not completed_context - or not completed_instance - or completed_product != operation_product - or completed_instance != operation_instance - ): + if lease_is_active: + proposed_action = "wait_for_active_lease" + if reservation.state != "running" or not lease_is_active: + try: + provider_operation_key = build_provider_operation_key( + scope=reservation.scope, + route_path=reservation.route_path, + idempotency_key=reservation.idempotency_key, + request_fingerprint=reservation.request_fingerprint, + reconciliation_key=reservation.reconciliation_key, + ) + except ValueError as error: raise dependencies.http_error( status_code=409, trace_id=trace_id, code="reservation_target_conflict", - message="Completed generic web deploy recovery context is not authoritative.", - ) - operation_context = completed_context - proposed_action = "replay_completed" - provider_status = _bounded_recovery_value(completed_result.deploy_status) - else: - if not reservation.reconciliation_key or not reservation.provider_target_key: + message="Generic web deploy recovery reservation identity is incomplete.", + ) from error + adapter = GenericWebDeployProviderMutationAdapter( + control_plane_root=dependencies.control_plane_root, + record_store=record_store, + deploy_request=recovery_request.original_deploy, + profile=profile, + lane=authoritative_lane, + trace_id=trace_id, + ) + provider_inspection = adapter.inspect( + provider_operation_key=provider_operation_key, + provider_effect_phase=reservation.provider_effect_phase, + reconciliation_key=reservation.reconciliation_key, + expected_provider_target_key=reservation.provider_target_key, + ) + if not provider_inspection.identity_matches: raise dependencies.http_error( status_code=409, trace_id=trace_id, code="reservation_target_conflict", - message="Stored generic web deploy recovery target identity is incomplete.", + message="Stored generic web deploy recovery target identities conflict.", ) - try: - stored_target = decode_generic_web_provider_reconciliation_target( - reservation.reconciliation_key + provider_outcome = provider_inspection.observation.outcome + provider_status = _bounded_recovery_value( + provider_inspection.observation.deployment_status + ) + retry_safe = provider_inspection.retry_safe + provider_observation_payload = provider_inspection.observation.model_dump(mode="json") + if provider_outcome == "present": + proposed_action = "adopt_observed" + elif provider_outcome == "absent" and retry_safe: + proposed_action = "retry_original_operation" + else: + proposed_action = "hold_unknown" + + return _GenericWebDeployRecoveryInspection( + request=recovery_request, + product=operation_product, + context=operation_context, + instance=operation_instance, + reservation=reservation, + observed_at=lookup.observed_at, + original_fingerprint=original_fingerprint, + idempotency_key=normalized_key, + provider_operation_key=provider_operation_key, + provider_outcome=provider_outcome, + provider_status=provider_status, + retry_safe=retry_safe, + proposed_action=proposed_action, + provider_observation_payload=provider_observation_payload, + adapter=adapter, + provider_inspection=provider_inspection, + ) + + +def build_generic_web_deploy_recovery_dry_run_handler( + *, dependencies: GenericWebDeployRecoveryDependencies +) -> Callable[..., Any]: + async def dry_run_generic_web_deploy_recovery( + request: Request, + recovery_request: GenericWebDeployRecoveryDryRunRequest, + identity: Annotated[LaunchplaneIdentity, Depends(dependencies.read_write_identity)], + record_store: Annotated[object, Depends(dependencies.get_record_store)], + idempotency_key: Annotated[str, Header(alias="Idempotency-Key")], + ) -> GenericWebDeployRecoveryDryRunResponse: + trace_id = dependencies.next_trace_id() + inspection = await _inspect_generic_web_deploy_recovery( + request=request, + recovery_request=recovery_request, + identity=identity, + record_store=record_store, + idempotency_key=idempotency_key, + trace_id=trace_id, + dependencies=dependencies, + ) + return inspection.dry_run_response() + + return dry_run_generic_web_deploy_recovery + + +def build_generic_web_deploy_recovery_apply_handler( + *, dependencies: GenericWebDeployRecoveryDependencies +) -> Callable[..., Any]: + async def apply_generic_web_deploy_recovery( + request: Request, + recovery_request: GenericWebDeployRecoveryApplyRequest, + identity: Annotated[LaunchplaneIdentity, Depends(dependencies.read_write_identity)], + record_store: Annotated[object, Depends(dependencies.get_record_store)], + idempotency_key: Annotated[str, Header(alias="Idempotency-Key")], + ) -> GenericWebDeployRecoveryApplyResponse: + trace_id = dependencies.next_trace_id() + inspection = await _inspect_generic_web_deploy_recovery( + request=request, + recovery_request=recovery_request, + identity=identity, + record_store=record_store, + idempotency_key=idempotency_key, + trace_id=trace_id, + dependencies=dependencies, + ) + if not isinstance(record_store, PostgresRecordStore): + raise dependencies.http_error( + status_code=503, + trace_id=trace_id, + code="database_storage_required", + message="Generic web deploy recovery apply requires database storage.", + ) + if inspection.reservation.state == "completed": + metadata = _stored_recovery_metadata(inspection.reservation) + if metadata is not None and metadata[0] == recovery_request.expected_recovery_digest: + return _apply_replay_response( + trace_id=trace_id, + request=recovery_request, + reservation=inspection.reservation, + context=inspection.context, ) - except ValueError as error: + if inspection.recovery_digest != recovery_request.expected_recovery_digest: raise dependencies.http_error( status_code=409, trace_id=trace_id, - code="reservation_target_conflict", - message="Stored generic web deploy recovery target identity is invalid.", - ) from error - operation_context = stored_target.context.strip() - stored_instance = stored_target.instance.strip() - stored_product = stored_target.product.strip() - legacy_snapshot_without_product = "product" not in stored_target.model_fields_set - if ( - not operation_context - or (not legacy_snapshot_without_product and stored_product != operation_product) - or stored_instance != operation_instance - ): + code="stale_recovery_digest", + message="Reviewed generic web deploy recovery digest no longer matches.", + ) + return _apply_response( + trace_id=trace_id, + inspection=inspection, + reservation=inspection.reservation, + ) + if inspection.recovery_digest != recovery_request.expected_recovery_digest: + raise dependencies.http_error( + status_code=409, + trace_id=trace_id, + code="stale_recovery_digest", + message="Reviewed generic web deploy recovery digest no longer matches.", + ) + if inspection.proposed_action not in {"adopt_observed", "retry_original_operation"}: + raise dependencies.http_error( + status_code=409, + trace_id=trace_id, + code="recovery_not_actionable", + message="Generic web deploy recovery inspection is not safely actionable.", + ) + if inspection.adapter is None: + raise RuntimeError("Generic web deploy recovery apply requires an adapter.") + recovery_metadata = _recovery_metadata(inspection) + action_inspection = _transition_expired_running_recovery( + inspection=inspection, + store=record_store, + trace_id=trace_id, + dependencies=dependencies, + ) + + if inspection.proposed_action == "adopt_observed": + if action_inspection.provider_inspection is None: + raise RuntimeError("Generic web deploy recovery adoption requires inspection.") + provider_observation = inspection.adapter.provider_observation_from_inspection( + provider_operation_key=inspection.provider_operation_key, + inspection=action_inspection.provider_inspection, + ) + if provider_observation.outcome != "present": raise dependencies.http_error( status_code=409, trace_id=trace_id, - code="reservation_target_conflict", - message="Stored generic web deploy recovery target identity conflicts.", + code="recovery_not_actionable", + message="Generic web deploy recovery provider evidence is not adoptable.", ) - authoritative_lane = lane.model_copy( - update={"context": operation_context, "instance": stored_instance} + response_payload = dict(provider_observation.response_payload) + response_payload["recovery"] = recovery_metadata + adoption = record_store.adopt_reconciled_mutation( + reservation=action_inspection.reservation, + response_status_code=provider_observation.response_status_code, + response_trace_id=trace_id, + response_payload=response_payload, ) - try: - resolved_stored_target = resolve_generic_web_provider_reconciliation_target( - reconciliation_key=reservation.reconciliation_key, - request_artifact_id=recovery_request.original_deploy.deploy.artifact_id, - request_source_git_ref=recovery_request.original_deploy.deploy.source_git_ref, - request_timeout_seconds=recovery_request.original_deploy.deploy.timeout_seconds, - request_no_cache=recovery_request.original_deploy.deploy.no_cache, - normalized_artifact_id=normalize_generic_web_artifact_id( - profile=profile, - artifact_id=recovery_request.original_deploy.deploy.artifact_id, - ), - request_deploy_reference=( - recovery_request.original_deploy.deploy.deploy_reference - ), - lane=authoritative_lane, - ) - except (ValueError, click.ClickException) as error: - raise dependencies.http_error( - status_code=409, + if adoption.status == "replayed" and adoption.record is not None: + return _apply_response( trace_id=trace_id, - code="reservation_target_conflict", - message="Stored generic web deploy recovery target identity is invalid.", - ) from error - if ( - resolved_stored_target.ship_request.context != operation_context - or resolved_stored_target.ship_request.instance != stored_instance - or build_generic_web_provider_target_key(resolved_stored_target) - != reservation.provider_target_key - ): + inspection=inspection, + reservation=adoption.record, + ) + if adoption.status != "adopted" or adoption.record is None: raise dependencies.http_error( status_code=409, trace_id=trace_id, - code="reservation_target_conflict", - message="Stored generic web deploy recovery target identity conflicts.", + code="reservation_changed", + message="Generic web deploy recovery reservation changed before adoption.", ) + return _apply_response( + trace_id=trace_id, + inspection=inspection, + reservation=adoption.record, + ) - if not dependencies.authorization_allows( - identity=identity, - action="generic_web_deploy.execute", - product=operation_product, - context=operation_context, - target=AuthorizationTarget(scope="instance", instances=(operation_instance,)), - ): + retry = record_store.retry_reconciled_mutation( + reservation=action_inspection.reservation, + lease_owner=trace_id, + ) + if retry.status == "replayed" and retry.record is not None: + return _apply_response( + trace_id=trace_id, + inspection=inspection, + reservation=retry.record, + ) + if retry.status != "acquired" or retry.record is None: raise dependencies.http_error( - status_code=403, + status_code=409, trace_id=trace_id, - code="authorization_denied", - message=( - "Identity cannot inspect generic web deploy recovery for the stored " - "product/context." - ), + code="reservation_changed", + message="Generic web deploy recovery reservation changed before retry.", ) - - if reservation.state != "completed": - lease_is_active = False - if reservation.state == "running": - try: - lease_is_active = parse_launchplane_mutation_timestamp( - reservation.lease_expires_at, - field_name="lease_expires_at", - ) > parse_launchplane_mutation_timestamp( - lookup.observed_at, - field_name="observed_at", - ) - except ValueError as error: - raise dependencies.http_error( - status_code=409, - trace_id=trace_id, - code="hold_unknown", - message="Generic web deploy recovery lease timing is not authoritative.", - ) from error - if lease_is_active: - proposed_action = "wait_for_active_lease" - if reservation.state != "running" or not lease_is_active: - try: - provider_operation_key = build_provider_operation_key( - scope=reservation.scope, - route_path=reservation.route_path, - idempotency_key=reservation.idempotency_key, - request_fingerprint=reservation.request_fingerprint, - reconciliation_key=reservation.reconciliation_key, - ) - except ValueError as error: - raise dependencies.http_error( - status_code=409, - trace_id=trace_id, - code="reservation_target_conflict", - message="Generic web deploy recovery reservation identity is incomplete.", - ) from error - adapter = GenericWebDeployProviderMutationAdapter( - control_plane_root=dependencies.control_plane_root, - record_store=record_store, - deploy_request=recovery_request.original_deploy, - profile=profile, - lane=authoritative_lane, - trace_id=trace_id, - ) - inspection = adapter.inspect( - provider_operation_key=provider_operation_key, - provider_effect_phase=reservation.provider_effect_phase, - reconciliation_key=reservation.reconciliation_key, - expected_provider_target_key=reservation.provider_target_key, - ) - if not inspection.identity_matches: - raise dependencies.http_error( - status_code=409, - trace_id=trace_id, - code="reservation_target_conflict", - message="Stored generic web deploy recovery target identities conflict.", - ) - provider_outcome = inspection.observation.outcome - provider_status = _bounded_recovery_value(inspection.observation.deployment_status) - retry_safe = inspection.retry_safe - provider_observation_payload = inspection.observation.model_dump(mode="json") - if provider_outcome == "present": - proposed_action = "adopt_observed" - elif provider_outcome == "absent" and retry_safe: - proposed_action = "retry_original_operation" - else: - proposed_action = "hold_unknown" - - digest_payload: dict[str, object] = { - "schema_version": 1, - "mode": "dry-run", - "request": recovery_request.model_dump(mode="json"), - "original_route": GENERIC_WEB_DEPLOY_ROUTE, - "idempotency_key": normalized_key, - "request_fingerprint": original_fingerprint, - "reservation": reservation.model_dump(mode="json"), - "observed_at": lookup.observed_at, - "provider_operation_key": provider_operation_key, - "provider_observation": provider_observation_payload, - "retry_safe": retry_safe, - "proposed_action": proposed_action, - } - return GenericWebDeployRecoveryDryRunResponse( - product=operation_product, - context=operation_context, - instance=operation_instance, - reservation_state=reservation.state, - reservation_attempt=reservation.attempt, - reservation_created_at=reservation.created_at, - reservation_updated_at=reservation.updated_at, - reservation_lease_expires_at=reservation.lease_expires_at, - observed_at=lookup.observed_at, - reconciliation_key_sha256=generic_web_deploy_recovery_identifier_sha256( - reservation.reconciliation_key + result = resume_acquired_provider_operation( + store=record_store, + reservation=retry.record, + response_trace_id=trace_id, + adapter=_GenericWebRecoveryApplyAdapter( + delegate=inspection.adapter, + recovery_metadata=recovery_metadata, ), - provider_target_key_sha256=generic_web_deploy_recovery_identifier_sha256( - reservation.provider_target_key - ), - provider_effect_phase=_bounded_recovery_value(reservation.provider_effect_phase), - provider_outcome=provider_outcome, - provider_status=provider_status, - retry_safe=retry_safe, - proposed_action=proposed_action, - recovery_digest=build_generic_web_deploy_recovery_digest(digest_payload), + ) + if result.status in {"completed", "adopted", "replayed"} and result.record is not None: + return _apply_response( + trace_id=trace_id, + inspection=inspection, + reservation=result.record, + ) + raise dependencies.http_error( + status_code=409, + trace_id=trace_id, + code="mutation_reconciliation_required", + message="Generic web deploy recovery retry requires reconciliation.", ) - return dry_run_generic_web_deploy_recovery + return apply_generic_web_deploy_recovery diff --git a/control_plane/http_routes/generic_web.py b/control_plane/http_routes/generic_web.py index 45bbe71d9..22d2f2c30 100644 --- a/control_plane/http_routes/generic_web.py +++ b/control_plane/http_routes/generic_web.py @@ -12,6 +12,7 @@ from starlette.exceptions import HTTPException as StarletteHTTPException from control_plane.contracts.generic_web_deploy_recovery import ( + GenericWebDeployRecoveryApplyResponse, GenericWebDeployRecoveryDryRunResponse, ) from control_plane.contracts.idempotency_record import LaunchplaneIdempotencyRecord @@ -32,8 +33,10 @@ GenericWebDeployProviderMutationAdapter, ) from control_plane.generic_web_deploy_recovery_http import ( + GENERIC_WEB_DEPLOY_RECOVERY_APPLY_ROUTE as _GENERIC_WEB_DEPLOY_RECOVERY_APPLY_ROUTE, GENERIC_WEB_DEPLOY_RECOVERY_DRY_RUN_ROUTE as _GENERIC_WEB_DEPLOY_RECOVERY_DRY_RUN_ROUTE, GenericWebDeployRecoveryDependencies, + build_generic_web_deploy_recovery_apply_handler, build_generic_web_deploy_recovery_dry_run_handler, ) from control_plane.generic_web_preview_http import ( @@ -310,6 +313,7 @@ class GenericWebWriteRouteHandlers: apply_generic_web_preview_destroy: Callable[..., Any] apply_generic_web_deploy: Callable[..., Any] dry_run_generic_web_deploy_recovery: Callable[..., Any] + apply_generic_web_deploy_recovery: Callable[..., Any] apply_generic_web_prod_promotion: Callable[..., Any] dispatch_generic_web_prod_promotion_workflow: Callable[..., Any] apply_generic_web_stable_verification: Callable[..., Any] @@ -353,6 +357,17 @@ def build_generic_web_write_route_handlers( idempotency_request_fingerprint=dependencies.idempotency_request_fingerprint, ) ) + apply_generic_web_deploy_recovery = build_generic_web_deploy_recovery_apply_handler( + dependencies=GenericWebDeployRecoveryDependencies( + read_write_identity=dependencies.read_write_identity, + get_record_store=dependencies.get_record_store, + next_trace_id=dependencies.next_trace_id, + authorization_allows=dependencies.authorization_allows, + http_error=dependencies.http_error, + control_plane_root=dependencies.control_plane_root, + idempotency_request_fingerprint=dependencies.idempotency_request_fingerprint, + ) + ) def manager_preview_pr_number( *, @@ -1949,6 +1964,7 @@ async def apply_generic_web_preview_verification( apply_generic_web_preview_destroy=apply_generic_web_preview_destroy, apply_generic_web_deploy=apply_generic_web_deploy, dry_run_generic_web_deploy_recovery=dry_run_generic_web_deploy_recovery, + apply_generic_web_deploy_recovery=apply_generic_web_deploy_recovery, apply_generic_web_prod_promotion=apply_generic_web_prod_promotion, dispatch_generic_web_prod_promotion_workflow=dispatch_generic_web_prod_promotion_workflow, apply_generic_web_stable_verification=apply_generic_web_stable_verification, @@ -2147,6 +2163,25 @@ def register_generic_web_write_routes( }, ) + app.add_api_route( + _GENERIC_WEB_DEPLOY_RECOVERY_APPLY_ROUTE, + handlers.apply_generic_web_deploy_recovery, + methods=["POST"], + status_code=202, + response_model=GenericWebDeployRecoveryApplyResponse, + response_model_exclude_none=True, + operation_id="apply_generic_web_deploy_recovery", + summary="Apply a reviewed generic web deploy recovery", + responses={ + 400: {"model": dependencies.error_response_model}, + 401: {"model": dependencies.error_response_model}, + 403: {"model": dependencies.error_response_model}, + 404: {"model": dependencies.error_response_model}, + 409: {"model": dependencies.error_response_model}, + 503: {"model": dependencies.error_response_model}, + }, + ) + app.add_api_route( _GENERIC_WEB_PROD_PROMOTION_ROUTE, handlers.apply_generic_web_prod_promotion, diff --git a/control_plane/provider_operations.py b/control_plane/provider_operations.py index f298cb084..f851bb0b9 100644 --- a/control_plane/provider_operations.py +++ b/control_plane/provider_operations.py @@ -486,6 +486,55 @@ def reserve() -> MutationReservationResult: ) +def resume_acquired_provider_operation( + *, + store: DurableProviderOperationStore, + reservation: LaunchplaneIdempotencyRecord, + response_trace_id: str, + adapter: DurableProviderMutationAdapter, + lease_seconds: int = 300, + heartbeat_interval_seconds: float | None = None, +) -> DurableProviderOperationResult: + if reservation.state != "running": + raise ValueError("Provider operation resume requires a running reservation.") + if not reservation.reconciliation_key: + raise ValueError("Provider operation resume requires a reconciliation key.") + if adapter.reconciliation_key().strip() != reservation.reconciliation_key: + raise ValueError("Provider operation resume reconciliation identity does not match.") + if adapter.target_key().strip() != reservation.provider_target_key: + raise ValueError("Provider operation resume target identity does not match.") + normalized_response_trace_id = response_trace_id.strip() + if not normalized_response_trace_id: + raise ValueError("Provider operation resume requires a response trace id.") + if lease_seconds < 1: + raise ValueError("Durable provider operation leases must be positive.") + resolved_heartbeat_interval = heartbeat_interval_seconds + if resolved_heartbeat_interval is None: + resolved_heartbeat_interval = max(0.1, min(30.0, lease_seconds / 3)) + if resolved_heartbeat_interval <= 0: + raise ValueError("Durable provider operation heartbeat intervals must be positive.") + if resolved_heartbeat_interval >= lease_seconds: + raise ValueError("Durable provider operation heartbeats must run before lease expiry.") + provider_operation_key = build_provider_operation_key( + scope=reservation.scope, + route_path=reservation.route_path, + idempotency_key=reservation.idempotency_key, + request_fingerprint=reservation.request_fingerprint, + reconciliation_key=reservation.reconciliation_key, + ) + return _apply_acquired( + store=store, + adapter=adapter, + reservation=reservation, + reconciliation_key=reservation.reconciliation_key, + provider_operation_key=provider_operation_key, + response_trace_id=normalized_response_trace_id, + lease_seconds=lease_seconds, + heartbeat_interval_seconds=resolved_heartbeat_interval, + release_pre_effect_failures=False, + ) + + def _apply_acquired( *, store: DurableProviderOperationStore, @@ -496,6 +545,7 @@ def _apply_acquired( response_trace_id: str, lease_seconds: int, heartbeat_interval_seconds: float, + release_pre_effect_failures: bool = True, ) -> DurableProviderOperationResult: heartbeat = _ReservationHeartbeat( store=store, @@ -521,6 +571,12 @@ def _apply_acquired( reservation=current_reservation, reconciliation_key=reconciliation_key, ) + if not release_pre_effect_failures: + return _mark_reconcile_required( + store=store, + reservation=current_reservation, + reconciliation_key=reconciliation_key, + ) release = store.release_reserved_mutation(reservation=current_reservation) if release.status == "released": raise rejection.cause @@ -561,6 +617,12 @@ def _apply_acquired( reservation=current_reservation, reconciliation_key=reconciliation_key, ) + if not release_pre_effect_failures: + return _mark_reconcile_required( + store=store, + reservation=current_reservation, + reconciliation_key=reconciliation_key, + ) release = store.release_reserved_mutation(reservation=current_reservation) if release.status != "released": return _mark_reconcile_required( diff --git a/docs/operations.md b/docs/operations.md index 85c30ee2f..2dc8d5a8e 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -366,9 +366,9 @@ former replaces the previous process-local apply lock. Both require an explaining that cleanup won. Provider read failures remain fail-closed. Do not reproduce this transition with direct SQL or provider-side deletion. -### Generic-web deploy recovery dry-run +### Generic-web deploy recovery dry-run/apply -Stage 1 recovery for legacy generic-web deploys is read-only. Operators call +Legacy generic-web deploy recovery starts read-only. Operators call `POST /v1/admin/generic-web/deploy-recovery/dry-run` with the exact original `GenericWebDeployEnvelope` under `original_deploy`, the original `Idempotency-Key`, the product and instance, and a non-empty reason. The service @@ -398,9 +398,27 @@ timestamps, hashed identifiers, provider outcome/status, retry safety, one of Raw scopes, idempotency keys, reconciliation keys, provider-target keys, original payloads, target URLs, and provider payloads are never returned. -There is no recovery apply endpoint in Stage 1. A later stage must bind any -mutation to reviewed dry-run evidence and revalidate the reservation atomically; -until then, the recovery digest is evidence only and authorizes no mutation. +Stage 2 apply is explicit and digest-gated. Operators call +`POST /v1/admin/generic-web/deploy-recovery/apply` with the same request body as +the dry-run plus `expected_recovery_digest`. The service recomputes a fresh +inspection and requires an exact digest match before it writes anything. The +digest intentionally excludes `observed_at`, so routine database observation +time changes do not stale a reviewed recovery; all reservation identity, +provider classification, action, request, and bounded provider observation +inputs remain part of the digest. + +Apply never creates a new reservation, releases a reservation, or supersedes a +provider target. If the reviewed reservation is an expired `running` mutation, +apply first CAS-transitions that exact reservation to `reconcile_required` via +the durable mutation store, then uses the existing exact-match adoption or retry +methods. `adopt_observed` stores deployment/inventory evidence from the reviewed +inspection without re-observing the provider. `retry_original_operation` resumes +the already-acquired reservation path and preserves `reconcile_required` for +pre-effect rejection, lease loss, uncertain, or non-durable outcomes. Successful +apply preserves the original deploy response evidence and adds only bounded +recovery metadata, `recovery_digest` and `recovery_action`, so a lost-response +apply retry can replay safely without exposing raw scope, key, target, or +provider payload data. ## Target Launchplane Ingress diff --git a/frontend/generated/openapi-canonical.json b/frontend/generated/openapi-canonical.json index 0407d6032..a8f8cfe17 100644 --- a/frontend/generated/openapi-canonical.json +++ b/frontend/generated/openapi-canonical.json @@ -6499,6 +6499,149 @@ "title": "GenericWebDeployEnvelope", "type": "object" }, + "GenericWebDeployRecoveryApplyRequest": { + "additionalProperties": false, + "properties": { + "expected_recovery_digest": { + "pattern": "^[0-9a-f]{64}$", + "title": "Expected Recovery Digest", + "type": "string" + }, + "instance": { + "title": "Instance", + "type": "string" + }, + "original_deploy": { + "$ref": "#/components/schemas/GenericWebDeployEnvelope" + }, + "product": { + "title": "Product", + "type": "string" + }, + "reason": { + "maxLength": 1000, + "title": "Reason", + "type": "string" + }, + "schema_version": { + "const": 1, + "default": 1, + "title": "Schema Version", + "type": "integer" + } + }, + "required": [ + "product", + "instance", + "original_deploy", + "reason", + "expected_recovery_digest" + ], + "title": "GenericWebDeployRecoveryApplyRequest", + "type": "object" + }, + "GenericWebDeployRecoveryApplyResponse": { + "additionalProperties": false, + "properties": { + "context": { + "title": "Context", + "type": "string" + }, + "instance": { + "title": "Instance", + "type": "string" + }, + "mode": { + "const": "apply", + "default": "apply", + "title": "Mode", + "type": "string" + }, + "product": { + "title": "Product", + "type": "string" + }, + "provider_outcome": { + "enum": [ + "present", + "absent", + "unknown", + "not_inspected" + ], + "title": "Provider Outcome", + "type": "string" + }, + "provider_status": { + "default": "", + "maxLength": 128, + "title": "Provider Status", + "type": "string" + }, + "recovery_action": { + "enum": [ + "replay_completed", + "wait_for_active_lease", + "adopt_observed", + "retry_original_operation", + "hold_unknown" + ], + "title": "Recovery Action", + "type": "string" + }, + "recovery_digest": { + "pattern": "^[0-9a-f]{64}$", + "title": "Recovery Digest", + "type": "string" + }, + "reservation_attempt": { + "minimum": 1.0, + "title": "Reservation Attempt", + "type": "integer" + }, + "reservation_state": { + "enum": [ + "completed", + "reconcile_required" + ], + "title": "Reservation State", + "type": "string" + }, + "retry_safe": { + "title": "Retry Safe", + "type": "boolean" + }, + "schema_version": { + "const": 1, + "default": 1, + "title": "Schema Version", + "type": "integer" + }, + "status": { + "const": "accepted", + "default": "accepted", + "title": "Status", + "type": "string" + }, + "trace_id": { + "title": "Trace Id", + "type": "string" + } + }, + "required": [ + "trace_id", + "product", + "context", + "instance", + "reservation_state", + "reservation_attempt", + "recovery_action", + "recovery_digest", + "provider_outcome", + "retry_safe" + ], + "title": "GenericWebDeployRecoveryApplyResponse", + "type": "object" + }, "GenericWebDeployRecoveryDryRunRequest": { "additionalProperties": false, "properties": { @@ -33909,6 +34052,115 @@ "summary": "Logout human auth session" } }, + "/v1/admin/generic-web/deploy-recovery/apply": { + "post": { + "operationId": "apply_generic_web_deploy_recovery", + "parameters": [ + { + "in": "header", + "name": "Idempotency-Key", + "required": true, + "schema": { + "title": "Idempotency-Key", + "type": "string" + } + }, + { + "in": "header", + "name": "Authorization", + "required": false, + "schema": { + "default": "", + "title": "Authorization", + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/GenericWebDeployRecoveryApplyRequest" + } + } + }, + "required": true + }, + "responses": { + "202": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/GenericWebDeployRecoveryApplyResponse" + } + } + }, + "description": "Successful Response" + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/LaunchplaneErrorResponse" + } + } + }, + "description": "Bad Request" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/LaunchplaneErrorResponse" + } + } + }, + "description": "Unauthorized" + }, + "403": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/LaunchplaneErrorResponse" + } + } + }, + "description": "Forbidden" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/LaunchplaneErrorResponse" + } + } + }, + "description": "Not Found" + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/LaunchplaneErrorResponse" + } + } + }, + "description": "Conflict" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/LaunchplaneErrorResponse" + } + } + }, + "description": "Service Unavailable" + } + }, + "summary": "Apply a reviewed generic web deploy recovery" + } + }, "/v1/admin/generic-web/deploy-recovery/dry-run": { "post": { "operationId": "dry_run_generic_web_deploy_recovery", diff --git a/tests/test_generic_web_deploy_recovery.py b/tests/test_generic_web_deploy_recovery.py index bb8672d81..d9878d3e3 100644 --- a/tests/test_generic_web_deploy_recovery.py +++ b/tests/test_generic_web_deploy_recovery.py @@ -33,6 +33,7 @@ _RECOVERY_ROUTE = "/v1/admin/generic-web/deploy-recovery/dry-run" +_RECOVERY_APPLY_ROUTE = "/v1/admin/generic-web/deploy-recovery/apply" _OPERATOR_TOKEN = "local-operator-token" @@ -84,6 +85,31 @@ def _invoke_recovery( ) +def _invoke_recovery_apply( + app: Any, + *, + original_deploy: dict[str, object], + idempotency_key: str, + reason: str, + expected_recovery_digest: str, +) -> tuple[int, dict[str, Any]]: + return _invoke_app( + app, + method="POST", + path=_RECOVERY_APPLY_ROUTE, + authorization=f"Bearer {_OPERATOR_TOKEN}", + payload={ + "schema_version": 1, + "product": "sellyouroutboard", + "instance": "testing", + "original_deploy": original_deploy, + "reason": reason, + "expected_recovery_digest": expected_recovery_digest, + }, + headers={"Idempotency-Key": idempotency_key}, + ) + + def _generic_web_deploy_result( *, deployment_record_id: str = "deployment-syo-testing", @@ -341,6 +367,13 @@ def test_generic_web_deploy_recovery_dry_run_replays_without_writes(self) -> Non idempotency_key=idempotency_key, reason="Inspect the legacy reservation before recovery.", ) + apply_status, apply_payload = _invoke_recovery_apply( + app, + original_deploy=original_deploy, + idempotency_key=idempotency_key, + reason="Inspect the legacy reservation before recovery.", + expected_recovery_digest=payload["recovery_digest"], + ) after = store.read_idempotency_record( scope=reservation.scope, @@ -352,6 +385,9 @@ def test_generic_web_deploy_recovery_dry_run_replays_without_writes(self) -> Non store.close() self.assertEqual(status_code, 200) + self.assertEqual(apply_status, 202) + self.assertEqual(apply_payload["recovery_action"], "replay_completed") + self.assertEqual(apply_payload["recovery_digest"], payload["recovery_digest"]) self.assertEqual(payload["proposed_action"], "replay_completed") self.assertEqual(payload["provider_outcome"], "not_inspected") self.assertEqual(payload["provider_status"], "pass") @@ -721,6 +757,449 @@ def test_generic_web_deploy_recovery_holds_on_provider_read_uncertainty(self) -> self.assertEqual(payload["proposed_action"], "hold_unknown") self.assertEqual(provider.observation_calls, 1) + def test_generic_web_deploy_recovery_digest_ignores_observed_at(self) -> None: + with TemporaryDirectory() as temporary_directory_name: + root = Path(temporary_directory_name) + store = PostgresRecordStore( + database_url=_sqlite_database_url(root / "launchplane.sqlite3") + ) + store.ensure_schema() + store.write_product_profile_record( + LaunchplaneProductProfileRecord.model_validate(_product_profile_payload()) + ) + original_deploy = _generic_web_recovery_original_deploy() + reservation = _write_generic_web_recovery_reservation( + store, + _generic_web_recovery_reservation( + original_deploy=original_deploy, + idempotency_key="stable-digest", + ), + ) + provider = _RecoveryObservationProvider( + GenericWebProviderDeploymentObservation(outcome="absent") + ) + app = _create_recovery_app(root=root, store=store) + lookup_a = ExistingMutationReservationLookupResult( + status="found", + record=reservation, + observed_at="2026-08-16T17:00:00Z", + ) + lookup_b = ExistingMutationReservationLookupResult( + status="found", + record=reservation, + observed_at="2026-08-16T17:05:00Z", + ) + with ( + patch.object( + store, + "lookup_existing_mutation_reservation", + side_effect=(lookup_a, lookup_b), + ), + patch( + "control_plane.generic_web_deploy_provider_adapter." + "default_generic_web_deploy_provider", + return_value=provider, + ), + ): + first_status, first_payload = _invoke_recovery( + app, + original_deploy=original_deploy, + idempotency_key=reservation.idempotency_key, + reason="Check digest stability.", + ) + second_status, second_payload = _invoke_recovery( + app, + original_deploy=original_deploy, + idempotency_key=reservation.idempotency_key, + reason="Check digest stability.", + ) + store.close() + + self.assertEqual(first_status, 200) + self.assertEqual(second_status, 200) + self.assertNotEqual(first_payload["observed_at"], second_payload["observed_at"]) + self.assertEqual(first_payload["recovery_digest"], second_payload["recovery_digest"]) + + def test_generic_web_deploy_recovery_apply_rejects_stale_digest(self) -> None: + with TemporaryDirectory() as temporary_directory_name: + root = Path(temporary_directory_name) + store = PostgresRecordStore( + database_url=_sqlite_database_url(root / "launchplane.sqlite3") + ) + store.ensure_schema() + store.write_product_profile_record( + LaunchplaneProductProfileRecord.model_validate(_product_profile_payload()) + ) + original_deploy = _generic_web_recovery_original_deploy() + reservation = _write_generic_web_recovery_reservation( + store, + _generic_web_recovery_reservation( + original_deploy=original_deploy, + idempotency_key="stale-digest", + ), + ) + provider = _RecoveryObservationProvider( + GenericWebProviderDeploymentObservation(outcome="absent") + ) + app = _create_recovery_app(root=root, store=store) + with patch( + "control_plane.generic_web_deploy_provider_adapter." + "default_generic_web_deploy_provider", + return_value=provider, + ): + status_code, payload = _invoke_recovery_apply( + app, + original_deploy=original_deploy, + idempotency_key=reservation.idempotency_key, + reason="Reject stale recovery digest.", + expected_recovery_digest="0" * 64, + ) + stored = store.read_idempotency_record( + scope=reservation.scope, + route_path=reservation.route_path, + idempotency_key=reservation.idempotency_key, + ) + store.close() + + self.assertEqual(status_code, 409) + self.assertEqual(payload["error"]["code"], "stale_recovery_digest") + self.assertIsNotNone(stored) + assert stored is not None + self.assertEqual(stored.state, "reconcile_required") + + def test_generic_web_deploy_recovery_apply_adopts_observed_without_raw_leakage( + self, + ) -> None: + with TemporaryDirectory() as temporary_directory_name: + root = Path(temporary_directory_name) + store = PostgresRecordStore( + database_url=_sqlite_database_url(root / "launchplane.sqlite3") + ) + store.ensure_schema() + store.write_product_profile_record( + LaunchplaneProductProfileRecord.model_validate(_product_profile_payload()) + ) + original_deploy = _generic_web_recovery_original_deploy() + reservation = _write_generic_web_recovery_reservation( + store, + _generic_web_recovery_reservation( + original_deploy=original_deploy, + idempotency_key="adopt-observed", + ), + ) + provider = _RecoveryObservationProvider( + GenericWebProviderDeploymentObservation( + outcome="present", + deployment_status="success", + deployment_id="provider-deployment-123", + started_at="2026-08-15T12:00:00Z", + finished_at="2026-08-15T12:05:00Z", + ) + ) + app = _create_recovery_app(root=root, store=store) + with patch( + "control_plane.generic_web_deploy_provider_adapter." + "default_generic_web_deploy_provider", + return_value=provider, + ): + reason = "Review observed provider recovery." + dry_status, dry_payload = _invoke_recovery( + app, + original_deploy=original_deploy, + idempotency_key=reservation.idempotency_key, + reason=reason, + ) + with ( + patch( + "control_plane.generic_web_deploy_provider_adapter." + "default_generic_web_deploy_provider", + return_value=provider, + ), + patch.object(store, "reserve_mutation", side_effect=AssertionError("reserve")), + patch.object( + store, + "release_reserved_mutation", + side_effect=AssertionError("release"), + ), + patch.object( + store, + "supersede_expired_reconciled_mutation_and_reserve", + side_effect=AssertionError("supersession"), + ), + ): + apply_status, apply_payload = _invoke_recovery_apply( + app, + original_deploy=original_deploy, + idempotency_key=reservation.idempotency_key, + reason=reason, + expected_recovery_digest=dry_payload["recovery_digest"], + ) + stored = store.read_idempotency_record( + scope=reservation.scope, + route_path=reservation.route_path, + idempotency_key=reservation.idempotency_key, + ) + deployments = store.list_deployment_records() + store.close() + + self.assertEqual(dry_status, 200) + self.assertEqual(apply_status, 202) + self.assertEqual(apply_payload["recovery_action"], "adopt_observed") + self.assertEqual(apply_payload["recovery_digest"], dry_payload["recovery_digest"]) + self.assertEqual(provider.observation_calls, 2) + self.assertIsNotNone(stored) + assert stored is not None + self.assertEqual(stored.state, "completed") + self.assertEqual( + stored.response_payload["recovery"], + { + "schema_version": 1, + "recovery_digest": dry_payload["recovery_digest"], + "recovery_action": "adopt_observed", + }, + ) + self.assertEqual(stored.response_payload["result"]["deploy_status"], "pass") + self.assertTrue(stored.response_payload["records"]["deployment_record_id"]) + self.assertEqual(len(deployments), 1) + serialized = json.dumps(apply_payload, sort_keys=True) + self.assertNotIn(reservation.scope, serialized) + self.assertNotIn(reservation.idempotency_key, serialized) + self.assertNotIn("provider-deployment-123", serialized) + + def test_generic_web_deploy_recovery_apply_retries_and_replays_recovered_metadata( + self, + ) -> None: + with TemporaryDirectory() as temporary_directory_name: + root = Path(temporary_directory_name) + store = PostgresRecordStore( + database_url=_sqlite_database_url(root / "launchplane.sqlite3") + ) + store.ensure_schema() + store.write_product_profile_record( + LaunchplaneProductProfileRecord.model_validate(_product_profile_payload()) + ) + original_deploy = _generic_web_recovery_original_deploy() + reservation = _write_generic_web_recovery_reservation( + store, + _generic_web_recovery_reservation( + original_deploy=original_deploy, + idempotency_key="retry-original", + ), + ) + provider = _RecoveryObservationProvider( + GenericWebProviderDeploymentObservation(outcome="absent") + ) + app = _create_recovery_app(root=root, store=store) + with patch( + "control_plane.generic_web_deploy_provider_adapter." + "default_generic_web_deploy_provider", + return_value=provider, + ): + reason = "Review retry recovery." + dry_status, dry_payload = _invoke_recovery( + app, + original_deploy=original_deploy, + idempotency_key=reservation.idempotency_key, + reason=reason, + ) + with ( + patch( + "control_plane.generic_web_deploy_provider_adapter." + "default_generic_web_deploy_provider", + return_value=provider, + ), + patch( + "control_plane.generic_web_deploy_provider_adapter." + "execute_generic_web_deploy_result", + return_value=( + {"deployment_record_id": "deployment-retry-recovery"}, + { + "deployment_record_id": "deployment-retry-recovery", + "deploy_status": "pass", + "deploy_started_at": "2026-08-15T12:10:00Z", + "deploy_finished_at": "2026-08-15T12:15:00Z", + "product": "sellyouroutboard", + "context": "sellyouroutboard-testing", + "instance": "testing", + "post_deploy_status": "skipped", + "provider_effect_attempted": False, + }, + ), + ), + patch.object(store, "reserve_mutation", side_effect=AssertionError("reserve")), + patch.object( + store, + "release_reserved_mutation", + side_effect=AssertionError("release"), + ), + ): + apply_status, apply_payload = _invoke_recovery_apply( + app, + original_deploy=original_deploy, + idempotency_key=reservation.idempotency_key, + reason=reason, + expected_recovery_digest=dry_payload["recovery_digest"], + ) + replay_status, replay_payload = _invoke_recovery_apply( + app, + original_deploy=original_deploy, + idempotency_key=reservation.idempotency_key, + reason=reason, + expected_recovery_digest=dry_payload["recovery_digest"], + ) + stored = store.read_idempotency_record( + scope=reservation.scope, + route_path=reservation.route_path, + idempotency_key=reservation.idempotency_key, + ) + store.close() + + self.assertEqual(dry_status, 200) + self.assertEqual(apply_status, 202) + self.assertEqual(replay_status, 202) + self.assertEqual(apply_payload["recovery_action"], "retry_original_operation") + self.assertEqual(replay_payload["recovery_action"], "retry_original_operation") + self.assertEqual(replay_payload["recovery_digest"], dry_payload["recovery_digest"]) + self.assertIsNotNone(stored) + assert stored is not None + self.assertEqual(stored.state, "completed") + self.assertEqual( + stored.response_payload["recovery"], + { + "schema_version": 1, + "recovery_digest": dry_payload["recovery_digest"], + "recovery_action": "retry_original_operation", + }, + ) + self.assertEqual( + stored.response_payload["result"]["deployment_record_id"], + "deployment-retry-recovery", + ) + + def test_generic_web_deploy_recovery_apply_transitions_expired_running_before_retry( + self, + ) -> None: + with TemporaryDirectory() as temporary_directory_name: + root = Path(temporary_directory_name) + store = PostgresRecordStore( + database_url=_sqlite_database_url(root / "launchplane.sqlite3") + ) + store.ensure_schema() + store.write_product_profile_record( + LaunchplaneProductProfileRecord.model_validate(_product_profile_payload()) + ) + original_deploy = _generic_web_recovery_original_deploy() + reservation = _write_generic_web_recovery_reservation( + store, + _generic_web_recovery_reservation( + original_deploy=original_deploy, + idempotency_key="expired-running-retry", + state="running", + ), + ) + provider = _RecoveryObservationProvider( + GenericWebProviderDeploymentObservation(outcome="absent") + ) + app = _create_recovery_app(root=root, store=store) + lookup = ExistingMutationReservationLookupResult( + status="found", + record=reservation, + observed_at=reservation.lease_expires_at, + ) + with ( + patch.object(store, "lookup_existing_mutation_reservation", return_value=lookup), + patch( + "control_plane.generic_web_deploy_provider_adapter." + "default_generic_web_deploy_provider", + return_value=provider, + ), + ): + reason = "Review expired running retry recovery." + dry_status, dry_payload = _invoke_recovery( + app, + original_deploy=original_deploy, + idempotency_key=reservation.idempotency_key, + reason=reason, + ) + provider.observation = GenericWebProviderDeploymentObservation(outcome="absent") + transitions: list[str] = [] + original_mark_reconcile = store.mark_mutation_reconcile_required + original_retry_reconciled = store.retry_reconciled_mutation + + def mark_reconcile_side_effect( + *, + reservation: LaunchplaneIdempotencyRecord, + reconciliation_key: str, + ) -> object: + transitions.append("mark") + return original_mark_reconcile( + reservation=reservation, + reconciliation_key=reconciliation_key, + ) + + def retry_reconciled_side_effect( + *, + reservation: LaunchplaneIdempotencyRecord, + lease_owner: str, + lease_seconds: int = 300, + ) -> object: + transitions.append("retry") + return original_retry_reconciled( + reservation=reservation, + lease_owner=lease_owner, + lease_seconds=lease_seconds, + ) + + with ( + patch.object(store, "lookup_existing_mutation_reservation", return_value=lookup), + patch( + "control_plane.generic_web_deploy_provider_adapter." + "default_generic_web_deploy_provider", + return_value=provider, + ), + patch.object( + store, + "mark_mutation_reconcile_required", + side_effect=mark_reconcile_side_effect, + ), + patch.object( + store, + "retry_reconciled_mutation", + side_effect=retry_reconciled_side_effect, + ), + patch( + "control_plane.generic_web_deploy_provider_adapter." + "execute_generic_web_deploy_result", + return_value=( + {"deployment_record_id": "deployment-expired-running-recovery"}, + { + "deployment_record_id": "deployment-expired-running-recovery", + "deploy_status": "pass", + "deploy_started_at": "2026-08-15T12:20:00Z", + "deploy_finished_at": "2026-08-15T12:25:00Z", + "product": "sellyouroutboard", + "context": "sellyouroutboard-testing", + "instance": "testing", + "post_deploy_status": "skipped", + "provider_effect_attempted": False, + }, + ), + ), + ): + apply_status, apply_payload = _invoke_recovery_apply( + app, + original_deploy=original_deploy, + idempotency_key=reservation.idempotency_key, + reason=reason, + expected_recovery_digest=dry_payload["recovery_digest"], + ) + store.close() + + self.assertEqual(dry_status, 200) + self.assertEqual(apply_status, 202) + self.assertEqual(apply_payload["recovery_action"], "retry_original_operation") + self.assertEqual(transitions[:2], ["mark", "retry"]) + def test_generic_web_deploy_recovery_rejects_stored_target_identity_before_provider_read( self, ) -> None: diff --git a/tests/test_http_write_route_registrars.py b/tests/test_http_write_route_registrars.py index 1c5da0ae6..05176e236 100644 --- a/tests/test_http_write_route_registrars.py +++ b/tests/test_http_write_route_registrars.py @@ -38,6 +38,27 @@ def test_generic_web_deploy_recovery_openapi_contract(self) -> None: for status_code in ("400", "401", "403", "404", "409", "503"): self.assertIn(status_code, route["responses"]) + def test_generic_web_deploy_recovery_apply_openapi_contract(self) -> None: + route = self.app.openapi()["paths"]["/v1/admin/generic-web/deploy-recovery/apply"]["post"] + + self.assertEqual(route["operationId"], "apply_generic_web_deploy_recovery") + self.assertEqual( + route["requestBody"]["content"]["application/json"]["schema"]["$ref"], + "#/components/schemas/GenericWebDeployRecoveryApplyRequest", + ) + self.assertEqual( + route["responses"]["202"]["content"]["application/json"]["schema"]["$ref"], + "#/components/schemas/GenericWebDeployRecoveryApplyResponse", + ) + idempotency_header = next( + parameter + for parameter in route["parameters"] + if parameter["in"] == "header" and parameter["name"] == "Idempotency-Key" + ) + self.assertTrue(idempotency_header["required"]) + for status_code in ("400", "401", "403", "404", "409", "503"): + self.assertIn(status_code, route["responses"]) + def test_evidence_write_routes_preserve_contracts_and_ownership(self) -> None: expected_routes = ( ("/v1/evidence/backup-gates", "write_backup_gate_evidence"), @@ -123,6 +144,11 @@ def test_generic_web_write_routes_preserve_contracts_and_ownership(self) -> None "dry_run_generic_web_deploy_recovery", "GenericWebDeployRecoveryDryRunResponse", ), + ( + "/v1/admin/generic-web/deploy-recovery/apply", + "apply_generic_web_deploy_recovery", + "GenericWebDeployRecoveryApplyResponse", + ), ( "/v1/drivers/generic-web/prod-promotion", "apply_generic_web_prod_promotion", @@ -172,7 +198,7 @@ def test_generic_web_write_routes_preserve_contracts_and_ownership(self) -> None self.assertEqual(route.response_model.__name__, response_model) expected_module = ( "control_plane.generic_web_deploy_recovery_http" - if route.path == "/v1/admin/generic-web/deploy-recovery/dry-run" + if route.path.startswith("/v1/admin/generic-web/deploy-recovery/") else "control_plane.http_routes.generic_web" ) self.assertEqual(route.endpoint.__module__, expected_module) @@ -188,6 +214,7 @@ def test_generic_web_write_routes_preserve_interleaved_route_order(self) -> None ("POST", "/v1/drivers/generic-web/preview-destroy"), ("POST", "/v1/drivers/generic-web/deploy"), ("POST", "/v1/admin/generic-web/deploy-recovery/dry-run"), + ("POST", "/v1/admin/generic-web/deploy-recovery/apply"), ("POST", "/v1/drivers/generic-web/prod-promotion"), ("POST", "/v1/drivers/generic-web/prod-promotion-workflow"), ("POST", "/v1/drivers/generic-web/stable-verification"), diff --git a/tests/test_provider_operations.py b/tests/test_provider_operations.py index 5bb71b0dd..8a04b7206 100644 --- a/tests/test_provider_operations.py +++ b/tests/test_provider_operations.py @@ -16,6 +16,7 @@ ProviderOperationLease, ProviderTargetSupersession, run_durable_provider_operation, + resume_acquired_provider_operation, ) from control_plane.contracts.idempotency_record import LaunchplaneIdempotencyRecord from control_plane.storage.postgres import PostgresRecordStore @@ -168,6 +169,140 @@ def test_fresh_acquire_applies_once_and_completes(self) -> None: self.assertEqual(len(adapter.provider_operation_keys), 1) self.assertTrue(adapter.provider_operation_keys[0].startswith("provider-operation:")) + def test_resume_acquired_running_reservation_completes_without_reserve_or_release( + self, + ) -> None: + with TemporaryDirectory() as directory: + fixture = _StoreFixture(directory) + adapter = _FakeAdapter() + reservation = fixture.store.reserve_mutation( + scope=_SCOPE, + route_path=_ROUTE, + idempotency_key=_KEY, + request_fingerprint=_FINGERPRINT, + lease_owner="recovery-instance", + lease_seconds=300, + reconciliation_key=_RECONCILIATION_KEY, + provider_target_key=adapter.target_key(), + ).record + with ( + patch.object( + fixture.store, "reserve_mutation", side_effect=AssertionError("reserve") + ), + patch.object( + fixture.store, + "release_reserved_mutation", + side_effect=AssertionError("release"), + ), + ): + result = resume_acquired_provider_operation( + store=fixture.store, + reservation=reservation, + response_trace_id="recovery-trace", + adapter=adapter, + ) + + self.assertEqual(result.status, "completed") + self.assertEqual(adapter.apply_calls, 1) + self.assertEqual(fixture.stored().state, "completed") + + def test_resume_acquired_preserves_reconcile_required_on_pre_effect_rejection( + self, + ) -> None: + with TemporaryDirectory() as directory: + fixture = _StoreFixture(directory) + adapter = _FakeAdapter( + apply_error=ProviderMutationRejectedError(ValueError("invalid request")) + ) + reservation = fixture.store.reserve_mutation( + scope=_SCOPE, + route_path=_ROUTE, + idempotency_key=_KEY, + request_fingerprint=_FINGERPRINT, + lease_owner="recovery-instance", + lease_seconds=300, + reconciliation_key=_RECONCILIATION_KEY, + provider_target_key=adapter.target_key(), + ).record + with patch.object( + fixture.store, + "release_reserved_mutation", + side_effect=AssertionError("release"), + ): + result = resume_acquired_provider_operation( + store=fixture.store, + reservation=reservation, + response_trace_id="recovery-trace", + adapter=adapter, + ) + + self.assertEqual(result.status, "reconcile_required") + self.assertEqual(fixture.stored().state, "reconcile_required") + + def test_resume_acquired_preserves_reconcile_required_on_non_durable_outcome( + self, + ) -> None: + with TemporaryDirectory() as directory: + fixture = _StoreFixture(directory) + adapter = _FakeAdapter( + apply_outcome=ProviderMutationOutcome( + response_status_code=202, + response_payload={"status": "accepted"}, + durable=False, + provider_effect_performed=False, + ) + ) + reservation = fixture.store.reserve_mutation( + scope=_SCOPE, + route_path=_ROUTE, + idempotency_key=_KEY, + request_fingerprint=_FINGERPRINT, + lease_owner="recovery-instance", + lease_seconds=300, + reconciliation_key=_RECONCILIATION_KEY, + provider_target_key=adapter.target_key(), + ).record + with patch.object( + fixture.store, + "release_reserved_mutation", + side_effect=AssertionError("release"), + ): + result = resume_acquired_provider_operation( + store=fixture.store, + reservation=reservation, + response_trace_id="recovery-trace", + adapter=adapter, + ) + + self.assertEqual(result.status, "reconcile_required") + self.assertEqual(fixture.stored().state, "reconcile_required") + + def test_resume_acquired_rejects_adapter_identity_drift(self) -> None: + with TemporaryDirectory() as directory: + fixture = _StoreFixture(directory) + adapter = _FakeAdapter() + reservation = fixture.store.reserve_mutation( + scope=_SCOPE, + route_path=_ROUTE, + idempotency_key=_KEY, + request_fingerprint=_FINGERPRINT, + lease_owner="recovery-instance", + lease_seconds=300, + reconciliation_key=_RECONCILIATION_KEY, + provider_target_key=adapter.target_key(), + ).record + mismatched = reservation.model_copy( + update={"provider_target_key": "provider-target:other"} + ) + + with self.assertRaisesRegex(ValueError, "target identity does not match"): + resume_acquired_provider_operation( + store=fixture.store, + reservation=mismatched, + response_trace_id="recovery-trace", + adapter=adapter, + ) + def test_long_apply_renews_lease_and_completes_with_latest_reservation(self) -> None: with TemporaryDirectory() as directory: fixture = _StoreFixture(directory)