diff --git a/control_plane/generic_web_deploy_provider_adapter.py b/control_plane/generic_web_deploy_provider_adapter.py new file mode 100644 index 000000000..58ac087aa --- /dev/null +++ b/control_plane/generic_web_deploy_provider_adapter.py @@ -0,0 +1,273 @@ +import hashlib +from dataclasses import dataclass +from pathlib import Path +from typing import cast + +import click + +from control_plane.contracts.product_profile_record import ( + LaunchplaneProductProfileRecord, + ProductLaneProfile, +) +from control_plane.generic_web_deploy_http import ( + GenericWebDeployEnvelope, + execute_generic_web_deploy_result, + should_store_generic_web_deploy_idempotency, +) +from control_plane.provider_operations import ( + ProviderMutationOutcome, + ProviderMutationRejectedError, + ProviderMutationUnknownError, + ProviderObservation, + ProviderOperationLease, + provider_operation_response_payload, + provider_operation_title, +) +from control_plane.workflows.generic_web_deploy import ( + GenericWebDeployStore, + normalize_generic_web_artifact_id, + record_observed_generic_web_deploy, +) +from control_plane.workflows.generic_web_deploy_provider import ( + GenericWebDeployProvider, + GenericWebProviderDeploymentObservation, + GenericWebResolvedDeployTarget, + build_generic_web_provider_reconciliation_key, + build_generic_web_provider_target_key, + default_generic_web_deploy_provider, + generic_web_provider_deployment_succeeded, + resolve_generic_web_provider_reconciliation_target, +) +from control_plane.workflows.odoo_generic_web_post_deploy import ( + generic_web_post_deploy_executor_for_driver_id, +) + + +__all__ = [ + "_GenericWebDeployProviderInspection", + "GenericWebDeployProviderMutationAdapter", +] + + +@dataclass(frozen=True, slots=True) +class _GenericWebDeployProviderInspection: + observation: GenericWebProviderDeploymentObservation + resolved_deploy_target: GenericWebResolvedDeployTarget | None = None + retry_safe: bool = False + identity_matches: bool = True + post_deploy_unobserved: bool = False + + +class GenericWebDeployProviderMutationAdapter: + def __init__( + self, + *, + control_plane_root: Path, + record_store: object, + deploy_request: GenericWebDeployEnvelope, + profile: LaunchplaneProductProfileRecord, + lane: ProductLaneProfile, + trace_id: str, + ) -> None: + self._control_plane_root = control_plane_root + self._record_store = record_store + self._deploy_request = deploy_request + self._profile = profile + self._lane = lane + self._trace_id = trace_id + self._deploy_provider: GenericWebDeployProvider = default_generic_web_deploy_provider() + self._resolved_deploy_target: GenericWebResolvedDeployTarget | None = None + + def _resolve_deploy_target(self) -> GenericWebResolvedDeployTarget: + if self._resolved_deploy_target is None: + deploy = self._deploy_request.deploy + self._resolved_deploy_target = self._deploy_provider.resolve_deploy_target( + control_plane_root=self._control_plane_root, + request_artifact_id=deploy.artifact_id, + request_source_git_ref=deploy.source_git_ref, + request_timeout_seconds=deploy.timeout_seconds, + request_no_cache=deploy.no_cache, + record_store=self._record_store, + profile=self._profile, + lane=self._lane, + normalized_artifact_id=normalize_generic_web_artifact_id( + profile=self._profile, + artifact_id=deploy.artifact_id, + ), + request_deploy_reference=deploy.deploy_reference, + fallback_target_name=f"{self._profile.product}-{self._lane.instance}", + ) + return self._resolved_deploy_target + + def reconciliation_key(self) -> str: + return build_generic_web_provider_reconciliation_key( + self._resolve_deploy_target(), + product=self._profile.product, + ) + + def target_key(self) -> str: + return build_generic_web_provider_target_key(self._resolve_deploy_target()) + + def observe( + self, + provider_operation_key: str, + provider_effect_phase: str, + reconciliation_key: str, + ) -> ProviderObservation: + inspection = self.inspect( + provider_operation_key=provider_operation_key, + provider_effect_phase=provider_effect_phase, + reconciliation_key=reconciliation_key, + ) + observation = inspection.observation + if observation.outcome != "present" or inspection.resolved_deploy_target is None: + return ProviderObservation( + outcome=observation.outcome, + retry_safe=inspection.retry_safe, + ) + resolved_target = inspection.resolved_deploy_target + deployment_record_id = self._deployment_record_id(provider_operation_key) + try: + records, driver_result = record_observed_generic_web_deploy( + record_store=cast(GenericWebDeployStore, self._record_store), + profile=self._profile, + lane=self._lane, + resolved_deploy_target=resolved_target, + observation=observation, + deployment_record_id=deployment_record_id, + post_deploy_unobserved=inspection.post_deploy_unobserved, + ) + except (FileNotFoundError, ValueError, click.ClickException): + return ProviderObservation(outcome="unknown") + terminal_failure = str(driver_result.get("deploy_status", "")).strip() == "fail" + return ProviderObservation( + outcome="present", + response_status_code=502 if terminal_failure else 202, + response_payload=provider_operation_response_payload( + trace_id=self._trace_id, + records=records, + result=driver_result, + ), + ) + + def inspect( + self, + *, + provider_operation_key: str, + provider_effect_phase: str, + reconciliation_key: str, + expected_provider_target_key: str = "", + ) -> _GenericWebDeployProviderInspection: + try: + deploy = self._deploy_request.deploy + resolved_target = resolve_generic_web_provider_reconciliation_target( + reconciliation_key=reconciliation_key, + request_artifact_id=deploy.artifact_id, + request_source_git_ref=deploy.source_git_ref, + request_timeout_seconds=deploy.timeout_seconds, + request_no_cache=deploy.no_cache, + normalized_artifact_id=normalize_generic_web_artifact_id( + profile=self._profile, + artifact_id=deploy.artifact_id, + ), + request_deploy_reference=deploy.deploy_reference, + lane=self._lane, + ) + if ( + expected_provider_target_key.strip() + and build_generic_web_provider_target_key(resolved_target) + != expected_provider_target_key.strip() + ): + return _GenericWebDeployProviderInspection( + observation=GenericWebProviderDeploymentObservation(outcome="unknown"), + identity_matches=False, + ) + self._resolved_deploy_target = resolved_target + observation = self._deploy_provider.observe_artifact_deploy( + control_plane_root=self._control_plane_root, + resolved_deploy_target=resolved_target, + deployment_title=provider_operation_title(provider_operation_key), + ) + except (FileNotFoundError, ValueError, click.ClickException): + return _GenericWebDeployProviderInspection( + observation=GenericWebProviderDeploymentObservation(outcome="unknown") + ) + if observation.outcome != "present": + if observation.outcome == "absent" and provider_effect_phase == "deploy_trigger": + return _GenericWebDeployProviderInspection( + observation=GenericWebProviderDeploymentObservation(outcome="unknown") + ) + return _GenericWebDeployProviderInspection( + observation=observation, + retry_safe=( + observation.outcome == "absent" + and provider_effect_phase in {"", "target_update"} + ), + ) + post_deploy_unobserved = ( + generic_web_provider_deployment_succeeded(observation.deployment_status) + and generic_web_post_deploy_executor_for_driver_id(self._profile.driver_id) is not None + ) + deployment_record_id = self._deployment_record_id(provider_operation_key) + if provider_effect_phase.startswith("post_deploy_"): + read_deployment_record = getattr(self._record_store, "read_deployment_record", None) + if not callable(read_deployment_record): + return _GenericWebDeployProviderInspection( + observation=GenericWebProviderDeploymentObservation(outcome="unknown") + ) + try: + read_deployment_record(deployment_record_id) + except FileNotFoundError: + return _GenericWebDeployProviderInspection( + observation=GenericWebProviderDeploymentObservation(outcome="unknown") + ) + post_deploy_unobserved = False + return _GenericWebDeployProviderInspection( + observation=observation, + resolved_deploy_target=resolved_target, + post_deploy_unobserved=post_deploy_unobserved, + ) + + def _deployment_record_id(self, provider_operation_key: str) -> str: + operation_digest = hashlib.sha256(provider_operation_key.encode("utf-8")).hexdigest()[:24] + return ( + f"deployment-provider-operation-{operation_digest}-" + f"{self._lane.context}-{self._lane.instance}" + ) + + def apply( + self, provider_operation_key: str, lease: ProviderOperationLease + ) -> ProviderMutationOutcome: + try: + records, result = execute_generic_web_deploy_result( + control_plane_root=self._control_plane_root, + record_store=self._record_store, + request=self._deploy_request, + profile=self._profile, + lane=self._lane, + provider_operation_title=provider_operation_title(provider_operation_key), + deployment_record_id=self._deployment_record_id(provider_operation_key), + deploy_provider=self._deploy_provider, + resolved_deploy_target=self._resolve_deploy_target(), + provider_effect_checkpoint=lease.checkpoint_effect, + ) + except (FileNotFoundError, ValueError) as error: + raise ProviderMutationRejectedError(error) + except click.ClickException as error: + raise ProviderMutationUnknownError(str(error)) from error + provider_effect_attempted = result.pop("provider_effect_attempted", False) is True + if str(result.get("deploy_status", "")).strip() == "fail" and provider_effect_attempted: + raise ProviderMutationUnknownError( + str(result.get("error_message", "")).strip() + or "Generic web provider outcome requires reconciliation." + ) + return ProviderMutationOutcome( + response_status_code=202, + response_payload=provider_operation_response_payload( + trace_id=self._trace_id, + records=records, + result=result, + ), + durable=should_store_generic_web_deploy_idempotency(result), + provider_effect_performed=provider_effect_attempted, + ) diff --git a/control_plane/http_routes/generic_web.py b/control_plane/http_routes/generic_web.py index 796d49edf..3cdb840bc 100644 --- a/control_plane/http_routes/generic_web.py +++ b/control_plane/http_routes/generic_web.py @@ -23,9 +23,10 @@ GenericWebDeployEnvelope, GenericWebDeployProductMismatchError, GenericWebDeployRouteDependencyError, - execute_generic_web_deploy_result, resolve_generic_web_deploy_lane, - should_store_generic_web_deploy_idempotency, +) +from control_plane.generic_web_deploy_provider_adapter import ( + GenericWebDeployProviderMutationAdapter, ) from control_plane.generic_web_preview_http import ( GENERIC_WEB_PREVIEW_DESIRED_STATE_ROUTE as _GENERIC_WEB_PREVIEW_DESIRED_STATE_ROUTE, @@ -128,9 +129,7 @@ ) from control_plane.storage.postgres import OutboxWithIdempotencyRequest, PostgresRecordStore from control_plane.workflows.generic_web_deploy import ( - GenericWebDeployStore, normalize_generic_web_artifact_id, - record_observed_generic_web_deploy, ) from control_plane.workflows.generic_web_deploy_provider import ( GenericWebDeployProvider, @@ -138,8 +137,6 @@ build_generic_web_provider_reconciliation_key, build_generic_web_provider_target_key, default_generic_web_deploy_provider, - generic_web_provider_deployment_succeeded, - resolve_generic_web_provider_reconciliation_target, ) from control_plane.workflows.generic_web_preview import ( GenericWebPreviewProfileStore, @@ -147,9 +144,6 @@ preview_pr_number_from_slug, resolve_generic_web_preview_slug, ) -from control_plane.workflows.odoo_generic_web_post_deploy import ( - generic_web_post_deploy_executor_for_driver_id, -) from control_plane.workflows.ship import utc_now_timestamp @@ -162,174 +156,6 @@ ] -class _GenericWebDeployProviderMutationAdapter: - def __init__( - self, - *, - control_plane_root: FilePath, - record_store: object, - deploy_request: GenericWebDeployEnvelope, - profile: LaunchplaneProductProfileRecord, - lane: ProductLaneProfile, - trace_id: str, - ) -> None: - self._control_plane_root = control_plane_root - self._record_store = record_store - self._deploy_request = deploy_request - self._profile = profile - self._lane = lane - self._trace_id = trace_id - self._deploy_provider: GenericWebDeployProvider = default_generic_web_deploy_provider() - self._resolved_deploy_target: GenericWebResolvedDeployTarget | None = None - - def _resolve_deploy_target(self) -> GenericWebResolvedDeployTarget: - if self._resolved_deploy_target is None: - deploy = self._deploy_request.deploy - self._resolved_deploy_target = self._deploy_provider.resolve_deploy_target( - control_plane_root=self._control_plane_root, - request_artifact_id=deploy.artifact_id, - request_source_git_ref=deploy.source_git_ref, - request_timeout_seconds=deploy.timeout_seconds, - request_no_cache=deploy.no_cache, - record_store=self._record_store, - profile=self._profile, - lane=self._lane, - normalized_artifact_id=normalize_generic_web_artifact_id( - profile=self._profile, - artifact_id=deploy.artifact_id, - ), - request_deploy_reference=deploy.deploy_reference, - fallback_target_name=f"{self._profile.product}-{self._lane.instance}", - ) - return self._resolved_deploy_target - - def reconciliation_key(self) -> str: - return build_generic_web_provider_reconciliation_key(self._resolve_deploy_target()) - - def target_key(self) -> str: - return build_generic_web_provider_target_key(self._resolve_deploy_target()) - - def observe( - self, - provider_operation_key: str, - provider_effect_phase: str, - reconciliation_key: str, - ) -> ProviderObservation: - try: - deploy = self._deploy_request.deploy - resolved_target = resolve_generic_web_provider_reconciliation_target( - reconciliation_key=reconciliation_key, - request_artifact_id=deploy.artifact_id, - request_source_git_ref=deploy.source_git_ref, - request_timeout_seconds=deploy.timeout_seconds, - request_no_cache=deploy.no_cache, - normalized_artifact_id=normalize_generic_web_artifact_id( - profile=self._profile, - artifact_id=deploy.artifact_id, - ), - request_deploy_reference=deploy.deploy_reference, - lane=self._lane, - ) - self._resolved_deploy_target = resolved_target - observation = self._deploy_provider.observe_artifact_deploy( - control_plane_root=self._control_plane_root, - resolved_deploy_target=resolved_target, - deployment_title=provider_operation_title(provider_operation_key), - ) - except (FileNotFoundError, ValueError, click.ClickException): - return ProviderObservation(outcome="unknown") - if observation.outcome != "present": - if observation.outcome == "absent" and provider_effect_phase == "deploy_trigger": - return ProviderObservation(outcome="unknown") - return ProviderObservation( - outcome=observation.outcome, - retry_safe=( - observation.outcome == "absent" - and provider_effect_phase in {"", "target_update"} - ), - ) - post_deploy_unobserved = ( - generic_web_provider_deployment_succeeded(observation.deployment_status) - and generic_web_post_deploy_executor_for_driver_id(self._profile.driver_id) is not None - ) - deployment_record_id = self._deployment_record_id(provider_operation_key) - if provider_effect_phase.startswith("post_deploy_"): - read_deployment_record = getattr(self._record_store, "read_deployment_record", None) - if not callable(read_deployment_record): - return ProviderObservation(outcome="unknown") - try: - read_deployment_record(deployment_record_id) - except FileNotFoundError: - return ProviderObservation(outcome="unknown") - post_deploy_unobserved = False - try: - records, driver_result = record_observed_generic_web_deploy( - record_store=cast(GenericWebDeployStore, self._record_store), - profile=self._profile, - lane=self._lane, - resolved_deploy_target=resolved_target, - observation=observation, - deployment_record_id=deployment_record_id, - post_deploy_unobserved=post_deploy_unobserved, - ) - except (FileNotFoundError, ValueError, click.ClickException): - return ProviderObservation(outcome="unknown") - terminal_failure = str(driver_result.get("deploy_status", "")).strip() == "fail" - return ProviderObservation( - outcome="present", - response_status_code=502 if terminal_failure else 202, - response_payload=_provider_operation_response_payload( - trace_id=self._trace_id, - records=records, - result=driver_result, - ), - ) - - def _deployment_record_id(self, provider_operation_key: str) -> str: - operation_digest = hashlib.sha256(provider_operation_key.encode("utf-8")).hexdigest()[:24] - return ( - f"deployment-provider-operation-{operation_digest}-" - f"{self._lane.context}-{self._lane.instance}" - ) - - def apply( - self, provider_operation_key: str, lease: ProviderOperationLease - ) -> ProviderMutationOutcome: - try: - records, result = execute_generic_web_deploy_result( - control_plane_root=self._control_plane_root, - record_store=self._record_store, - request=self._deploy_request, - profile=self._profile, - lane=self._lane, - provider_operation_title=provider_operation_title(provider_operation_key), - deployment_record_id=self._deployment_record_id(provider_operation_key), - deploy_provider=self._deploy_provider, - resolved_deploy_target=self._resolve_deploy_target(), - provider_effect_checkpoint=lease.checkpoint_effect, - ) - except (FileNotFoundError, ValueError) as error: - raise ProviderMutationRejectedError(error) - except click.ClickException as error: - raise ProviderMutationUnknownError(str(error)) from error - provider_effect_attempted = result.pop("provider_effect_attempted", False) is True - if str(result.get("deploy_status", "")).strip() == "fail" and provider_effect_attempted: - raise ProviderMutationUnknownError( - str(result.get("error_message", "")).strip() - or "Generic web provider outcome requires reconciliation." - ) - return ProviderMutationOutcome( - response_status_code=202, - response_payload=_provider_operation_response_payload( - trace_id=self._trace_id, - records=records, - result=result, - ), - durable=should_store_generic_web_deploy_idempotency(result), - provider_effect_performed=provider_effect_attempted, - ) - - class _GenericWebProdPromotionProviderMutationAdapter: def __init__( self, @@ -374,7 +200,10 @@ def resolve_deploy_target(self) -> GenericWebResolvedDeployTarget: return self._resolved_deploy_target def reconciliation_key(self) -> str: - return build_generic_web_provider_reconciliation_key(self.resolve_deploy_target()) + return build_generic_web_provider_reconciliation_key( + self.resolve_deploy_target(), + product=self._profile.product, + ) def target_key(self) -> str: return build_generic_web_provider_target_key(self.resolve_deploy_target()) @@ -1474,7 +1303,7 @@ async def apply_generic_web_deploy( route_path=_GENERIC_WEB_DEPLOY_ROUTE, payload=cast(dict[str, object], raw_payload), ) - adapter = _GenericWebDeployProviderMutationAdapter( + adapter = GenericWebDeployProviderMutationAdapter( control_plane_root=dependencies.control_plane_root, record_store=record_store, deploy_request=deploy_request, diff --git a/control_plane/http_routes/mutation_support.py b/control_plane/http_routes/mutation_support.py index 57a90f310..5c9d4251f 100644 --- a/control_plane/http_routes/mutation_support.py +++ b/control_plane/http_routes/mutation_support.py @@ -11,6 +11,9 @@ GENERIC_WEB_PROD_PROMOTION_ROUTE, GENERIC_WEB_PROD_PROMOTION_WORKFLOW_ROUTE, ) +from control_plane.provider_operations import ( + provider_operation_response_payload as provider_operation_response_payload, +) from control_plane.service_auth import ( GitHubActionsIdentity, GitHubHumanIdentity, @@ -66,19 +69,6 @@ def accepted_evidence_response( ) -def provider_operation_response_payload( - *, - trace_id: str, - records: Mapping[str, object], - result: dict[str, object], -) -> dict[str, object]: - return accepted_evidence_response( - trace_id=trace_id, - records=records, - result=result, - ).model_dump(mode="json", exclude_none=True) - - def idempotency_capable_store(record_store: object) -> IdempotencyCapableStore | None: if callable(getattr(record_store, "read_idempotency_record", None)) and callable( getattr(record_store, "write_idempotency_record", None) diff --git a/control_plane/provider_operations.py b/control_plane/provider_operations.py index b09ccbb1b..f298cb084 100644 --- a/control_plane/provider_operations.py +++ b/control_plane/provider_operations.py @@ -16,6 +16,7 @@ from __future__ import annotations +from collections.abc import Mapping from dataclasses import dataclass, field import hashlib from threading import Event, Lock, Thread @@ -50,6 +51,20 @@ ] +def provider_operation_response_payload( + *, + trace_id: str, + records: Mapping[str, object], + result: dict[str, object], +) -> dict[str, object]: + return { + "status": "accepted", + "trace_id": trace_id, + "records": dict(records), + "result": result, + } + + class ProviderMutationRejectedError(Exception): """Raised by an adapter when the provider effect definitely did not happen. diff --git a/control_plane/storage/postgres.py b/control_plane/storage/postgres.py index f615313c1..386b49d4f 100644 --- a/control_plane/storage/postgres.py +++ b/control_plane/storage/postgres.py @@ -445,6 +445,13 @@ "lease_active", "grace_active", ] +ExistingMutationReservationLookupStatus = Literal[ + "found", + "missing", + "conflict", + "ambiguous", + "hold_unknown", +] class ProductProfileCompareWriteResult(NamedTuple): @@ -483,6 +490,12 @@ class MutationReservationResult(NamedTuple): record: LaunchplaneIdempotencyRecord +class ExistingMutationReservationLookupResult(NamedTuple): + status: ExistingMutationReservationLookupStatus + record: LaunchplaneIdempotencyRecord | None + observed_at: str + + class MutationReservationUpdateResult(NamedTuple): status: MutationReservationUpdateStatus record: LaunchplaneIdempotencyRecord | None = None @@ -4652,6 +4665,147 @@ def read_idempotency_record( return None return self._read_payload(model_type=LaunchplaneIdempotencyRecord, payload=row.payload) + def lookup_existing_mutation_reservation( + self, + *, + route_path: str, + idempotency_key: str, + request_fingerprint: str, + ) -> ExistingMutationReservationLookupResult: + normalized_route_path = route_path.strip() + normalized_idempotency_key = idempotency_key.strip() + normalized_request_fingerprint = request_fingerprint.strip() + if not normalized_route_path: + raise ValueError("Existing mutation lookup requires route_path.") + if not normalized_idempotency_key: + raise ValueError("Existing mutation lookup requires idempotency_key.") + if not normalized_request_fingerprint: + raise ValueError("Existing mutation lookup requires request_fingerprint.") + statement = ( + select(LaunchplaneIdempotencyRow) + .where( + LaunchplaneIdempotencyRow.route_path == normalized_route_path, + LaunchplaneIdempotencyRow.idempotency_key == normalized_idempotency_key, + ) + .order_by(LaunchplaneIdempotencyRow.scope) + .limit(2) + ) + with self._session_factory() as session: + observed_at = self._database_mutation_timestamp(session) + rows = tuple(session.scalars(statement)) + if not rows: + return ExistingMutationReservationLookupResult( + status="missing", + record=None, + observed_at=observed_at, + ) + records: list[LaunchplaneIdempotencyRecord] = [] + for row in rows: + raw_payload = row.payload + if not isinstance(raw_payload, dict) or ( + raw_payload.get("scope") != row.scope + or raw_payload.get("route_path") != row.route_path + or raw_payload.get("idempotency_key") != row.idempotency_key + or raw_payload.get("request_fingerprint") != row.request_fingerprint + ): + return ExistingMutationReservationLookupResult( + status="conflict", + record=None, + observed_at=observed_at, + ) + try: + record = self._read_payload( + model_type=LaunchplaneIdempotencyRecord, + payload=raw_payload, + ) + except ValueError: + if raw_payload.get("state") == "running": + return ExistingMutationReservationLookupResult( + status="hold_unknown", + record=None, + observed_at=observed_at, + ) + return ExistingMutationReservationLookupResult( + status="conflict", + record=None, + observed_at=observed_at, + ) + canonical_payload = record.model_dump(mode="json") + integrity_fields = ( + "record_id", + "scope", + "route_path", + "idempotency_key", + "request_fingerprint", + "state", + "lease_owner", + "lease_expires_at", + "attempt", + "reconciliation_key", + "provider_target_key", + "provider_effect_phase", + "provider_effect_started_at", + "created_at", + "updated_at", + "response_status_code", + "response_trace_id", + "recorded_at", + "response_payload", + ) + row_projection = { + "record_id": row.record_id, + "scope": row.scope, + "route_path": row.route_path, + "idempotency_key": row.idempotency_key, + "request_fingerprint": row.request_fingerprint, + "state": row.state, + "lease_owner": row.lease_owner, + "lease_expires_at": row.lease_expires_at, + "attempt": row.attempt, + "reconciliation_key": row.reconciliation_key, + "provider_target_key": row.provider_target_key, + "created_at": row.created_at, + "updated_at": row.updated_at, + "response_status_code": row.response_status_code, + "response_trace_id": row.response_trace_id, + "recorded_at": row.recorded_at, + } + if any( + raw_payload.get(field_name) != canonical_payload[field_name] + for field_name in integrity_fields + ) or any( + canonical_payload[field_name] != field_value + for field_name, field_value in row_projection.items() + ): + return ExistingMutationReservationLookupResult( + status="conflict", + record=None, + observed_at=observed_at, + ) + records.append(record) + matches = tuple( + record + for record in records + if record.request_fingerprint == normalized_request_fingerprint + ) + if len(matches) != len(records): + return ExistingMutationReservationLookupResult( + status="conflict", + record=None, + observed_at=observed_at, + ) + if len(matches) != 1: + return ExistingMutationReservationLookupResult( + status="ambiguous", + record=None, + observed_at=observed_at, + ) + return ExistingMutationReservationLookupResult( + status="found", + record=matches[0], + observed_at=observed_at, + ) + def prepare_db_only_mutation( self, *, diff --git a/control_plane/workflows/generic_web_deploy_provider.py b/control_plane/workflows/generic_web_deploy_provider.py index 97491497c..3471090ec 100644 --- a/control_plane/workflows/generic_web_deploy_provider.py +++ b/control_plane/workflows/generic_web_deploy_provider.py @@ -47,6 +47,7 @@ class GenericWebProviderReconciliationTarget(BaseModel): model_config = ConfigDict(extra="forbid") schema_version: int = Field(default=1, ge=1) + product: str = "" context: str instance: str provider_id: str @@ -88,9 +89,12 @@ def build_generic_web_provider_target_key( def build_generic_web_provider_reconciliation_key( resolved_deploy_target: GenericWebResolvedDeployTarget, + *, + product: str = "", ) -> str: ship_request = resolved_deploy_target.ship_request snapshot = GenericWebProviderReconciliationTarget( + product=product.strip(), context=ship_request.context, instance=ship_request.instance, provider_id=ship_request.provider_id, @@ -107,6 +111,23 @@ def build_generic_web_provider_reconciliation_key( return f"{_GENERIC_WEB_RECONCILIATION_KEY_PREFIX}{encoded.rstrip('=')}" +def decode_generic_web_provider_reconciliation_target( + reconciliation_key: str, +) -> GenericWebProviderReconciliationTarget: + normalized_key = reconciliation_key.strip() + if not normalized_key.startswith(_GENERIC_WEB_RECONCILIATION_KEY_PREFIX): + raise ValueError("Generic web provider reconciliation key is invalid.") + encoded = normalized_key.removeprefix(_GENERIC_WEB_RECONCILIATION_KEY_PREFIX) + try: + decoded = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)) + snapshot = GenericWebProviderReconciliationTarget.model_validate_json(decoded) + except Exception as error: + raise ValueError("Generic web provider reconciliation key is invalid.") from error + if not snapshot.context.strip() or not snapshot.instance.strip(): + raise ValueError("Generic web provider reconciliation target identity is incomplete.") + return snapshot + + def resolve_generic_web_provider_reconciliation_target( *, reconciliation_key: str, @@ -118,15 +139,7 @@ def resolve_generic_web_provider_reconciliation_target( lane: ProductLaneProfile, request_deploy_reference: str = "", ) -> GenericWebResolvedDeployTarget: - normalized_key = reconciliation_key.strip() - if not normalized_key.startswith(_GENERIC_WEB_RECONCILIATION_KEY_PREFIX): - raise ValueError("Generic web provider reconciliation key is invalid.") - encoded = normalized_key.removeprefix(_GENERIC_WEB_RECONCILIATION_KEY_PREFIX) - try: - decoded = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)) - snapshot = GenericWebProviderReconciliationTarget.model_validate_json(decoded) - except Exception as error: - raise ValueError("Generic web provider reconciliation key is invalid.") from error + snapshot = decode_generic_web_provider_reconciliation_target(reconciliation_key) context_name = lane.context.strip() instance_name = lane.instance.strip() if snapshot.context != context_name or snapshot.instance != instance_name: diff --git a/tests/test_generic_web_deploy.py b/tests/test_generic_web_deploy.py index 9ed9454bb..b0eba2af5 100644 --- a/tests/test_generic_web_deploy.py +++ b/tests/test_generic_web_deploy.py @@ -33,7 +33,9 @@ from control_plane.dokploy import DokploySourceOfTruth, DokployTargetDefinition from control_plane import secrets as control_plane_secrets from control_plane.generic_web_deploy_http import GenericWebDeployEnvelope -from control_plane.http_routes.generic_web import _GenericWebDeployProviderMutationAdapter +from control_plane.generic_web_deploy_provider_adapter import ( + GenericWebDeployProviderMutationAdapter, +) from control_plane.workflows.generic_web_deploy import ( GenericWebDeployRequest, GenericWebDeployStore, @@ -478,8 +480,8 @@ def _provider_mutation_adapter( store: _GenericWebDeployStore, provider: _FakeGenericWebDeployProvider, deploy_request: GenericWebDeployRequest | None = None, - ) -> _GenericWebDeployProviderMutationAdapter: - adapter = _GenericWebDeployProviderMutationAdapter( + ) -> GenericWebDeployProviderMutationAdapter: + adapter = GenericWebDeployProviderMutationAdapter( control_plane_root=Path("."), record_store=store, deploy_request=GenericWebDeployEnvelope( @@ -516,6 +518,41 @@ def test_provider_target_key_is_stable_across_request_timeout_changes(self) -> N long_timeout.reconciliation_key(), ) + def test_provider_inspection_does_not_materialize_observed_records(self) -> None: + profile = _profile() + store = _GenericWebDeployStore(profile) + provider = _FakeGenericWebDeployProvider() + provider.observation = GenericWebProviderDeploymentObservation( + outcome="present", + deployment_status="success", + deployment_id="deployment-provider-1", + started_at="2026-08-16T15:00:00Z", + finished_at="2026-08-16T15:01:00Z", + ) + adapter = self._provider_mutation_adapter( + profile=profile, + store=store, + provider=provider, + ) + reconciliation_key = adapter.reconciliation_key() + provider_target_key = adapter.target_key() + + with patch( + "control_plane.generic_web_deploy_provider_adapter.record_observed_generic_web_deploy" + ) as materialize: + inspection = adapter.inspect( + provider_operation_key="provider-operation:test-inspection", + provider_effect_phase="deploy_trigger", + reconciliation_key=reconciliation_key, + expected_provider_target_key=provider_target_key, + ) + + self.assertEqual(inspection.observation.outcome, "present") + self.assertTrue(inspection.identity_matches) + materialize.assert_not_called() + self.assertEqual(store.deployments, []) + self.assertEqual(store.inventories, []) + def test_normalize_generic_web_artifact_id_qualifies_bare_release_tag(self) -> None: artifact_id = normalize_generic_web_artifact_id( profile=_profile(), diff --git a/tests/test_postgres_store.py b/tests/test_postgres_store.py index 66e1c8328..161032d50 100644 --- a/tests/test_postgres_store.py +++ b/tests/test_postgres_store.py @@ -4278,6 +4278,223 @@ def test_mutation_reservation_replays_conflicts_and_reclaims_expired_lease( self.assertEqual(replayed.record.state, "completed") self.assertEqual(replayed.record.response_trace_id, "trace-mutation-completed") + def test_existing_mutation_lookup_crosses_scope_and_rejects_ambiguity(self) -> None: + with TemporaryDirectory() as temporary_directory_name: + store = PostgresRecordStore( + database_url=_sqlite_database_url( + Path(temporary_directory_name) / "launchplane.sqlite3" + ) + ) + store.ensure_schema() + with patch.object( + store, + "_database_mutation_timestamp", + return_value="2026-08-16T16:00:00Z", + ): + first = store.reserve_mutation( + scope="github-actions:attempt-1", + route_path="/v1/drivers/generic-web/deploy", + idempotency_key="legacy-deploy-key", + request_fingerprint="exact-original-fingerprint", + lease_owner="worker-a", + lease_seconds=300, + ) + found = store.lookup_existing_mutation_reservation( + route_path="/v1/drivers/generic-web/deploy", + idempotency_key="legacy-deploy-key", + request_fingerprint="exact-original-fingerprint", + ) + conflict = store.lookup_existing_mutation_reservation( + route_path="/v1/drivers/generic-web/deploy", + idempotency_key="legacy-deploy-key", + request_fingerprint="different-fingerprint", + ) + store.reserve_mutation( + scope="github-actions:attempt-2", + route_path="/v1/drivers/generic-web/deploy", + idempotency_key="legacy-deploy-key", + request_fingerprint="exact-original-fingerprint", + lease_owner="worker-b", + lease_seconds=300, + ) + ambiguous = store.lookup_existing_mutation_reservation( + route_path="/v1/drivers/generic-web/deploy", + idempotency_key="legacy-deploy-key", + request_fingerprint="exact-original-fingerprint", + ) + store.close() + + self.assertEqual(first.status, "acquired") + self.assertEqual(found.status, "found") + self.assertIsNotNone(found.record) + assert found.record is not None + self.assertEqual(found.record.scope, "github-actions:attempt-1") + self.assertEqual(found.observed_at, "2026-08-16T16:00:00Z") + self.assertEqual(conflict.status, "conflict") + self.assertIsNone(conflict.record) + self.assertEqual(ambiguous.status, "ambiguous") + self.assertIsNone(ambiguous.record) + + def test_existing_mutation_lookup_rejects_payload_projection_drift(self) -> None: + for field_name, corrupt_value, corrupt_row in ( + ("record_id", "corrupt-record", False), + ("scope", "github-actions:corrupt", False), + ("route_path", "/v1/drivers/generic-web/other", False), + ("idempotency_key", "corrupt-key", False), + ("request_fingerprint", "corrupt-fingerprint", False), + ("state", "reconcile_required", True), + ("lease_owner", "worker-corrupt", False), + ("lease_expires_at", "2099-08-16T18:00:00Z", False), + ("attempt", 2, False), + ("reconciliation_key", "reconciliation-corrupt", False), + ("provider_target_key", "provider-target-corrupt", False), + ("provider_effect_phase", " target_update ", False), + ("provider_effect_started_at", "2026-08-16T16:00:00+00:00", False), + ("created_at", "2026-08-16T15:59:59Z", False), + ("updated_at", "2026-08-16T16:00:01Z", False), + ("response_status_code", 201, False), + ("response_trace_id", "trace-corrupt", False), + ("recorded_at", "2026-08-16T16:00:02Z", False), + ("response_payload", "not-a-payload", False), + ): + with ( + self.subTest(field_name=field_name), + TemporaryDirectory() as temporary_directory_name, + ): + store = PostgresRecordStore( + database_url=_sqlite_database_url( + Path(temporary_directory_name) / "launchplane.sqlite3" + ) + ) + store.ensure_schema() + with patch.object( + store, + "_database_mutation_timestamp", + side_effect=( + "2026-08-16T16:00:00Z", + "2026-08-16T16:00:10Z", + "2026-08-16T16:01:00Z", + "2026-08-16T16:02:00Z", + ), + ): + running = store.reserve_mutation( + scope="github-actions:attempt-1", + route_path="/v1/drivers/generic-web/deploy", + idempotency_key="legacy-deploy-key", + request_fingerprint="exact-original-fingerprint", + lease_owner="worker-a", + lease_seconds=300, + reconciliation_key="reconciliation-key", + provider_target_key="provider-target-key", + ).record + checkpointed = store.checkpoint_mutation_provider_effect( + reservation=running, + effect_phase="target_update", + lease_seconds=300, + ) + assert checkpointed.record is not None + completion = complete_launchplane_mutation_reservation( + checkpointed.record, + response_status_code=202, + response_trace_id="trace-completed", + completed_at="2026-08-16T16:01:00Z", + response_payload={"status": "accepted"}, + ) + completed = store.complete_mutation_reservation(completion=completion) + assert completed.record is not None + reservation = completed.record + corrupt_payload = reservation.model_dump(mode="json") + with store._session_factory() as session: + values: dict[str, object] + if corrupt_row: + values = {field_name: corrupt_value} + else: + corrupt_payload[field_name] = corrupt_value + values = {"payload": corrupt_payload} + session.execute( + update(LaunchplaneIdempotencyRow) + .where(LaunchplaneIdempotencyRow.record_id == reservation.record_id) + .values(**values) + ) + session.commit() + + result = store.lookup_existing_mutation_reservation( + route_path="/v1/drivers/generic-web/deploy", + idempotency_key="legacy-deploy-key", + request_fingerprint="exact-original-fingerprint", + ) + store.close() + + self.assertEqual(result.status, "conflict") + self.assertIsNone(result.record) + + def test_existing_mutation_lookup_reads_at_most_two_candidate_payloads(self) -> None: + with TemporaryDirectory() as temporary_directory_name: + store = PostgresRecordStore( + database_url=_sqlite_database_url( + Path(temporary_directory_name) / "launchplane.sqlite3" + ) + ) + store.ensure_schema() + for attempt in range(3): + store.reserve_mutation( + scope=f"github-actions:attempt-{attempt}", + route_path="/v1/drivers/generic-web/deploy", + idempotency_key="legacy-deploy-key", + request_fingerprint="exact-original-fingerprint", + lease_owner=f"worker-{attempt}", + lease_seconds=300, + ) + with patch.object(store, "_read_payload", wraps=store._read_payload) as read_payload: + result = store.lookup_existing_mutation_reservation( + route_path="/v1/drivers/generic-web/deploy", + idempotency_key="legacy-deploy-key", + request_fingerprint="exact-original-fingerprint", + ) + store.close() + + self.assertEqual(result.status, "ambiguous") + self.assertEqual(read_payload.call_count, 2) + + def test_existing_mutation_lookup_holds_on_malformed_running_lease_timestamp(self) -> None: + with TemporaryDirectory() as temporary_directory_name: + store = PostgresRecordStore( + database_url=_sqlite_database_url( + Path(temporary_directory_name) / "launchplane.sqlite3" + ) + ) + store.ensure_schema() + reservation = store.reserve_mutation( + scope="github-actions:attempt-1", + route_path="/v1/drivers/generic-web/deploy", + idempotency_key="legacy-deploy-key", + request_fingerprint="exact-original-fingerprint", + lease_owner="worker-a", + lease_seconds=300, + ).record + corrupt_payload = reservation.model_dump(mode="json") + corrupt_payload["lease_expires_at"] = "not-a-timestamp" + with store._session_factory() as session: + session.execute( + update(LaunchplaneIdempotencyRow) + .where(LaunchplaneIdempotencyRow.record_id == reservation.record_id) + .values( + lease_expires_at="not-a-timestamp", + payload=corrupt_payload, + ) + ) + session.commit() + + result = store.lookup_existing_mutation_reservation( + route_path=reservation.route_path, + idempotency_key=reservation.idempotency_key, + request_fingerprint=reservation.request_fingerprint, + ) + store.close() + + self.assertEqual(result.status, "hold_unknown") + self.assertIsNone(result.record) + def test_active_reconciliation_key_fences_other_idempotency_keys(self) -> None: with TemporaryDirectory() as temporary_directory_name: store = PostgresRecordStore(