|
@@ -4548,3 +4548,1091 @@ class DataRuleRepository:
|
|
|
"dataflow references rules without published execution plans"
|
|
"dataflow references rules without published execution plans"
|
|
|
)
|
|
)
|
|
|
return standards, rules
|
|
return standards, rules
|
|
|
|
|
+
|
|
|
|
|
+ # ------------------------------------------------------------------
|
|
|
|
|
+ # Governed Data Factory deployment lifecycle (M5)
|
|
|
|
|
+ # ------------------------------------------------------------------
|
|
|
|
|
+
|
|
|
|
|
+ @staticmethod
|
|
|
|
|
+ def _deployment_result(row) -> dict[str, Any]:
|
|
|
|
|
+ result = dict(row)
|
|
|
|
|
+ for key in (
|
|
|
|
|
+ "id",
|
|
|
|
|
+ "dataflow_version_id",
|
|
|
|
|
+ "dataflow_uid",
|
|
|
|
|
+ "workflow_version_id",
|
|
|
|
|
+ "schedule_plan_id",
|
|
|
|
|
+ "canary_evidence_id",
|
|
|
|
|
+ "previous_active_deployment_id",
|
|
|
|
|
+ "created_by",
|
|
|
|
|
+ "activated_by",
|
|
|
|
|
+ "correlation_id",
|
|
|
|
|
+ ):
|
|
|
|
|
+ if result.get(key) is not None:
|
|
|
|
|
+ result[key] = str(result[key])
|
|
|
|
|
+ for key, label, default in (
|
|
|
|
|
+ ("package", "deployment package", {}),
|
|
|
|
|
+ ("standard_version_ids", "standard version ids", []),
|
|
|
|
|
+ ("rule_version_ids", "rule version ids", []),
|
|
|
|
|
+ ("binding_snapshot", "binding snapshot", {}),
|
|
|
|
|
+ ("schema_snapshots", "schema snapshots", {}),
|
|
|
|
|
+ ("physical_plan_hashes", "physical plan hashes", []),
|
|
|
|
|
+ ("workflow_spec", "workflow spec", {}),
|
|
|
|
|
+ ("schedule_snapshot", "schedule snapshot", {}),
|
|
|
|
|
+ ("deploy_receipt", "deploy receipt", {}),
|
|
|
|
|
+ ):
|
|
|
|
|
+ value = result.get(key)
|
|
|
|
|
+ if value is None:
|
|
|
|
|
+ result[key] = copy.deepcopy(default)
|
|
|
|
|
+ elif isinstance(default, dict):
|
|
|
|
|
+ result[key] = _object(value, label)
|
|
|
|
|
+ else:
|
|
|
|
|
+ result[key] = _array(value, label)
|
|
|
|
|
+ for key in (
|
|
|
|
|
+ "created_at",
|
|
|
|
|
+ "updated_at",
|
|
|
|
|
+ "activated_at",
|
|
|
|
|
+ "superseded_at",
|
|
|
|
|
+ "rolled_back_at",
|
|
|
|
|
+ ):
|
|
|
|
|
+ value = result.get(key)
|
|
|
|
|
+ if value is not None and hasattr(value, "isoformat"):
|
|
|
|
|
+ result[key] = value.isoformat()
|
|
|
|
|
+ return result
|
|
|
|
|
+
|
|
|
|
|
+ @staticmethod
|
|
|
|
|
+ def _deployment_select() -> str:
|
|
|
|
|
+ return (
|
|
|
|
|
+ "SELECT d.id::text AS id, "
|
|
|
|
|
+ "d.dataflow_version_id::text AS dataflow_version_id, "
|
|
|
|
|
+ "v.version_no, d.dataflow_uid::text AS dataflow_uid, "
|
|
|
|
|
+ "d.environment, d.workflow_version_id::text AS workflow_version_id, "
|
|
|
|
|
+ "d.schedule_plan_id::text AS schedule_plan_id, "
|
|
|
|
|
+ "d.deployment_config, d.status, d.package, d.package_hash, "
|
|
|
|
|
+ "d.standard_version_ids, d.rule_version_ids, "
|
|
|
|
|
+ "d.binding_snapshot, d.binding_hash, d.schema_snapshots, "
|
|
|
|
|
+ "d.schema_snapshot_hash, d.physical_plan_hashes, "
|
|
|
|
|
+ "d.workflow_spec, d.workflow_spec_hash, "
|
|
|
|
|
+ "d.schedule_snapshot, d.schedule_hash, "
|
|
|
|
|
+ "d.engine_namespace, d.engine_definition_id, "
|
|
|
|
|
+ "d.engine_revision, d.engine_definition_hash, d.deploy_receipt, "
|
|
|
|
|
+ "d.canary_evidence_id::text AS canary_evidence_id, "
|
|
|
|
|
+ "d.previous_active_deployment_id::text "
|
|
|
|
|
+ "AS previous_active_deployment_id, "
|
|
|
|
|
+ "d.created_by::text AS created_by, d.create_reason, "
|
|
|
|
|
+ "d.create_idempotency_key, d.correlation_id::text AS correlation_id, "
|
|
|
|
|
+ "d.lock_version, d.failure_code, "
|
|
|
|
|
+ "d.activated_by::text AS activated_by, d.activated_at, "
|
|
|
|
|
+ "d.superseded_at, d.rolled_back_at, d.created_at, d.updated_at "
|
|
|
|
|
+ "FROM public.dataflow_deployments d "
|
|
|
|
|
+ "JOIN public.dataflow_versions v ON v.id = d.dataflow_version_id "
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ def load_deployable_release(
|
|
|
|
|
+ self, dataflow_version_id: str, *, environment: str | None = None
|
|
|
|
|
+ ) -> dict[str, Any]:
|
|
|
|
|
+ version_id = _uid(dataflow_version_id, "dataflow_version_id")
|
|
|
|
|
+ if environment not in {"development", "test", "production"}:
|
|
|
|
|
+ raise ValueError("deployment environment is invalid")
|
|
|
|
|
+ version = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "SELECT id::text AS id, dataflow_uid::text AS dataflow_uid, "
|
|
|
|
|
+ "version_no, status, package, package_hash "
|
|
|
|
|
+ "FROM public.dataflow_versions "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid) AND status = 'released'"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"id": version_id},
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .one_or_none()
|
|
|
|
|
+ )
|
|
|
|
|
+ if version is None:
|
|
|
|
|
+ raise ValueError("released dataflow version was not found")
|
|
|
|
|
+ package = _object(version["package"], "production-line package")
|
|
|
|
|
+ package_hash = _digest(version["package_hash"], "package_hash")
|
|
|
|
|
+ if package.get("package_hash") != package_hash:
|
|
|
|
|
+ raise ValueError("production-line package has drifted")
|
|
|
|
|
+
|
|
|
|
|
+ # Task 7 creates a server-owned preparatory deployment so its physical
|
|
|
|
|
+ # plans can be compiled and tested against exact bindings. M5 promotes
|
|
|
|
|
+ # that same identity instead of copying plan identifiers into another
|
|
|
|
|
+ # deployment where Runner linkage would no longer attest.
|
|
|
|
|
+ context_rows = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "SELECT d.id::text AS deployment_id, "
|
|
|
|
|
+ "p.id::text AS plan_id, p.plan_hash, "
|
|
|
|
|
+ "ib.id::text AS input_binding_id, "
|
|
|
|
|
+ "ib.logical_ref AS input_logical_ref, "
|
|
|
|
|
+ "ib.data_source_uid::text AS input_data_source_uid, "
|
|
|
|
|
+ "ib.object_kind AS input_object_kind, "
|
|
|
|
|
+ "ib.object_ref AS input_object_ref, "
|
|
|
|
|
+ "ib.access_mode AS input_access_mode, "
|
|
|
|
|
+ "ib.write_mode AS input_write_mode, "
|
|
|
|
|
+ "ib.dialect AS input_dialect, "
|
|
|
|
|
+ "ib.binding_hash AS input_binding_hash, "
|
|
|
|
|
+ "ob.id::text AS output_binding_id, "
|
|
|
|
|
+ "ob.logical_ref AS output_logical_ref, "
|
|
|
|
|
+ "ob.data_source_uid::text AS output_data_source_uid, "
|
|
|
|
|
+ "ob.object_kind AS output_object_kind, "
|
|
|
|
|
+ "ob.object_ref AS output_object_ref, "
|
|
|
|
|
+ "ob.access_mode AS output_access_mode, "
|
|
|
|
|
+ "ob.write_mode AS output_write_mode, "
|
|
|
|
|
+ "ob.dialect AS output_dialect, "
|
|
|
|
|
+ "ob.binding_hash AS output_binding_hash, "
|
|
|
|
|
+ "ins.id::text AS input_snapshot_id, "
|
|
|
|
|
+ "ins.schema_hash AS input_schema_hash, "
|
|
|
|
|
+ "ins.schema_ref AS input_schema_ref, "
|
|
|
|
|
+ "ins.fields AS input_fields, "
|
|
|
|
|
+ "outs.id::text AS output_snapshot_id, "
|
|
|
|
|
+ "outs.schema_hash AS output_schema_hash, "
|
|
|
|
|
+ "outs.schema_ref AS output_schema_ref, "
|
|
|
|
|
+ "outs.fields AS output_fields "
|
|
|
|
|
+ "FROM public.dataflow_deployments d "
|
|
|
|
|
+ "JOIN public.dataflow_component_bindings cb "
|
|
|
|
|
+ "ON cb.dataflow_version_id = d.dataflow_version_id "
|
|
|
|
|
+ "JOIN public.rule_execution_plans p "
|
|
|
|
|
+ "ON p.component_binding_id = cb.id "
|
|
|
|
|
+ "AND p.status = 'published' "
|
|
|
|
|
+ "JOIN public.dataflow_dataset_bindings ib "
|
|
|
|
|
+ "ON ib.id = CAST(p.plan->>'input_binding_id' AS uuid) "
|
|
|
|
|
+ "AND ib.dataflow_deployment_id = d.id "
|
|
|
|
|
+ "JOIN public.data_schema_snapshots ins "
|
|
|
|
|
+ "ON ins.id = ib.schema_snapshot_id "
|
|
|
|
|
+ "JOIN public.dataflow_dataset_bindings ob "
|
|
|
|
|
+ "ON ob.id = CAST(p.plan->>'output_binding_id' AS uuid) "
|
|
|
|
|
+ "AND ob.dataflow_deployment_id = d.id "
|
|
|
|
|
+ "JOIN public.data_schema_snapshots outs "
|
|
|
|
|
+ "ON outs.id = ob.schema_snapshot_id "
|
|
|
|
|
+ "WHERE d.dataflow_version_id = CAST(:version_id AS uuid) "
|
|
|
|
|
+ "AND d.environment = :environment "
|
|
|
|
|
+ "AND d.package IS NULL "
|
|
|
|
|
+ "AND EXISTS (SELECT 1 FROM public.rule_compile_evidence ce "
|
|
|
|
|
+ "WHERE ce.rule_execution_plan_id = p.id "
|
|
|
|
|
+ "AND ce.status = 'success' "
|
|
|
|
|
+ "AND ce.legacy_untrusted = FALSE "
|
|
|
|
|
+ "AND ce.plan_hash = p.plan_hash) "
|
|
|
|
|
+ "AND EXISTS (SELECT 1 FROM public.rule_test_evidence te "
|
|
|
|
|
+ "WHERE te.rule_execution_plan_id = p.id "
|
|
|
|
|
+ "AND te.status = 'success' "
|
|
|
|
|
+ "AND te.legacy_untrusted = FALSE "
|
|
|
|
|
+ "AND te.plan_hash = p.plan_hash) "
|
|
|
|
|
+ "ORDER BY cb.order_no, cb.id "
|
|
|
|
|
+ "/* deployable_release_physical_context */"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"version_id": version_id, "environment": environment},
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .all()
|
|
|
|
|
+ )
|
|
|
|
|
+ component_count = int(
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "SELECT COUNT(*) FROM public.dataflow_component_bindings "
|
|
|
|
|
+ "WHERE dataflow_version_id = CAST(:id AS uuid)"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"id": version_id},
|
|
|
|
|
+ ).scalar_one()
|
|
|
|
|
+ )
|
|
|
|
|
+ if not context_rows or len(context_rows) != component_count:
|
|
|
|
|
+ raise ValueError(
|
|
|
|
|
+ "every production-line component requires a published "
|
|
|
|
|
+ "physical execution plan"
|
|
|
|
|
+ )
|
|
|
|
|
+ deployment_ids = {str(row["deployment_id"]) for row in context_rows}
|
|
|
|
|
+ if len(deployment_ids) != 1:
|
|
|
|
|
+ raise ValueError(
|
|
|
|
|
+ "physical execution plans do not share one deployment"
|
|
|
|
|
+ )
|
|
|
|
|
+ first = context_rows[0]
|
|
|
|
|
+ last = context_rows[-1]
|
|
|
|
|
+ return {
|
|
|
|
|
+ "id": str(version["id"]),
|
|
|
|
|
+ "dataflow_uid": str(version["dataflow_uid"]),
|
|
|
|
|
+ "version_no": int(version["version_no"]),
|
|
|
|
|
+ "status": str(version["status"]),
|
|
|
|
|
+ "package": package,
|
|
|
|
|
+ "package_hash": package_hash,
|
|
|
|
|
+ "preparatory_deployment_id": next(iter(deployment_ids)),
|
|
|
|
|
+ "physical_plan_hashes": [
|
|
|
|
|
+ str(row["plan_hash"]) for row in context_rows
|
|
|
|
|
+ ],
|
|
|
|
|
+ "binding_snapshot": {
|
|
|
|
|
+ "input": {
|
|
|
|
|
+ "binding_id": str(first["input_binding_id"]),
|
|
|
|
|
+ "logical_ref": str(first["input_logical_ref"]),
|
|
|
|
|
+ "data_source_uid": str(first["input_data_source_uid"]),
|
|
|
|
|
+ "object_kind": str(first["input_object_kind"]),
|
|
|
|
|
+ "object_ref": str(first["input_object_ref"]),
|
|
|
|
|
+ "schema_snapshot_id": str(first["input_snapshot_id"]),
|
|
|
|
|
+ "schema_hash": str(first["input_schema_hash"]),
|
|
|
|
|
+ "binding_hash": str(first["input_binding_hash"]),
|
|
|
|
|
+ "access_mode": str(first["input_access_mode"]),
|
|
|
|
|
+ "dialect": str(first["input_dialect"]),
|
|
|
|
|
+ },
|
|
|
|
|
+ "output": {
|
|
|
|
|
+ "binding_id": str(last["output_binding_id"]),
|
|
|
|
|
+ "logical_ref": str(last["output_logical_ref"]),
|
|
|
|
|
+ "data_source_uid": str(last["output_data_source_uid"]),
|
|
|
|
|
+ "object_kind": str(last["output_object_kind"]),
|
|
|
|
|
+ "object_ref": str(last["output_object_ref"]),
|
|
|
|
|
+ "schema_snapshot_id": str(last["output_snapshot_id"]),
|
|
|
|
|
+ "schema_hash": str(last["output_schema_hash"]),
|
|
|
|
|
+ "binding_hash": str(last["output_binding_hash"]),
|
|
|
|
|
+ "access_mode": str(last["output_access_mode"]),
|
|
|
|
|
+ "write_mode": str(last["output_write_mode"]),
|
|
|
|
|
+ "dialect": str(last["output_dialect"]),
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ "schema_snapshots": {
|
|
|
|
|
+ "input": {
|
|
|
|
|
+ "id": str(first["input_snapshot_id"]),
|
|
|
|
|
+ "schema_ref": str(first["input_schema_ref"]),
|
|
|
|
|
+ "schema_hash": str(first["input_schema_hash"]),
|
|
|
|
|
+ "fields": _array(first["input_fields"], "input fields"),
|
|
|
|
|
+ },
|
|
|
|
|
+ "output": {
|
|
|
|
|
+ "id": str(last["output_snapshot_id"]),
|
|
|
|
|
+ "schema_ref": str(last["output_schema_ref"]),
|
|
|
|
|
+ "schema_hash": str(last["output_schema_hash"]),
|
|
|
|
|
+ "fields": _array(last["output_fields"], "output fields"),
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ def create_deployment(self, snapshot: dict[str, Any]) -> dict[str, Any]:
|
|
|
|
|
+ value = _bounded_object(
|
|
|
|
|
+ snapshot, "deployment snapshot", maximum_bytes=2_000_000
|
|
|
|
|
+ )
|
|
|
|
|
+ deployment_id = _uid(
|
|
|
|
|
+ value.get("preparatory_deployment_id"),
|
|
|
|
|
+ "preparatory_deployment_id",
|
|
|
|
|
+ )
|
|
|
|
|
+ params = {
|
|
|
|
|
+ "id": deployment_id,
|
|
|
|
|
+ "version_id": _uid(
|
|
|
|
|
+ value["dataflow_version_id"], "dataflow_version_id"
|
|
|
|
|
+ ),
|
|
|
|
|
+ "dataflow_uid": _uid(value["dataflow_uid"], "dataflow_uid"),
|
|
|
|
|
+ "environment": _text(value["environment"], "environment", 20),
|
|
|
|
|
+ "package": _json(value["package"]),
|
|
|
|
|
+ "package_hash": _digest(value["package_hash"], "package_hash"),
|
|
|
|
|
+ "standard_version_ids": _json(value["standard_version_ids"]),
|
|
|
|
|
+ "rule_version_ids": _json(value["rule_version_ids"]),
|
|
|
|
|
+ "binding_snapshot": _json(value["binding_snapshot"]),
|
|
|
|
|
+ "binding_hash": _digest(value["binding_hash"], "binding_hash"),
|
|
|
|
|
+ "schema_snapshots": _json(value["schema_snapshots"]),
|
|
|
|
|
+ "schema_snapshot_hash": _digest(
|
|
|
|
|
+ value["schema_snapshot_hash"], "schema_snapshot_hash"
|
|
|
|
|
+ ),
|
|
|
|
|
+ "physical_plan_hashes": _json(value["physical_plan_hashes"]),
|
|
|
|
|
+ "workflow_spec": _json(value["workflow_spec"]),
|
|
|
|
|
+ "workflow_spec_hash": _digest(
|
|
|
|
|
+ value["workflow_spec_hash"], "workflow_spec_hash"
|
|
|
|
|
+ ),
|
|
|
|
|
+ "schedule_snapshot": _json(value["schedule_snapshot"]),
|
|
|
|
|
+ "schedule_hash": _digest(value["schedule_hash"], "schedule_hash"),
|
|
|
|
|
+ "actor_uid": _uid(value["actor_uid"], "actor_uid"),
|
|
|
|
|
+ "reason": _text(value["reason"], "reason", 1000),
|
|
|
|
|
+ "idempotency_key": _text(
|
|
|
|
|
+ value["idempotency_key"], "idempotency_key", 200
|
|
|
|
|
+ ),
|
|
|
|
|
+ "correlation_id": _uid(
|
|
|
|
|
+ value["correlation_id"], "correlation_id"
|
|
|
|
|
+ ),
|
|
|
|
|
+ }
|
|
|
|
|
+ updated = self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "UPDATE public.dataflow_deployments SET "
|
|
|
|
|
+ "dataflow_uid = CAST(:dataflow_uid AS uuid), "
|
|
|
|
|
+ "package = CAST(:package AS jsonb), package_hash = :package_hash, "
|
|
|
|
|
+ "standard_version_ids = CAST(:standard_version_ids AS jsonb), "
|
|
|
|
|
+ "rule_version_ids = CAST(:rule_version_ids AS jsonb), "
|
|
|
|
|
+ "binding_snapshot = CAST(:binding_snapshot AS jsonb), "
|
|
|
|
|
+ "binding_hash = :binding_hash, "
|
|
|
|
|
+ "schema_snapshots = CAST(:schema_snapshots AS jsonb), "
|
|
|
|
|
+ "schema_snapshot_hash = :schema_snapshot_hash, "
|
|
|
|
|
+ "physical_plan_hashes = CAST(:physical_plan_hashes AS jsonb), "
|
|
|
|
|
+ "workflow_spec = CAST(:workflow_spec AS jsonb), "
|
|
|
|
|
+ "workflow_spec_hash = :workflow_spec_hash, "
|
|
|
|
|
+ "schedule_snapshot = CAST(:schedule_snapshot AS jsonb), "
|
|
|
|
|
+ "schedule_hash = :schedule_hash, "
|
|
|
|
|
+ "created_by = CAST(:actor_uid AS uuid), "
|
|
|
|
|
+ "create_reason = :reason, "
|
|
|
|
|
+ "create_idempotency_key = :idempotency_key, "
|
|
|
|
|
+ "correlation_id = CAST(:correlation_id AS uuid), "
|
|
|
|
|
+ "status = 'draft', lock_version = lock_version + 1, "
|
|
|
|
|
+ "updated_at = CURRENT_TIMESTAMP "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid) "
|
|
|
|
|
+ "AND dataflow_version_id = CAST(:version_id AS uuid) "
|
|
|
|
|
+ "AND environment = :environment "
|
|
|
|
|
+ "AND package IS NULL "
|
|
|
|
|
+ "RETURNING id"
|
|
|
|
|
+ ),
|
|
|
|
|
+ params,
|
|
|
|
|
+ )
|
|
|
|
|
+ if getattr(updated, "rowcount", 0) != 1:
|
|
|
|
|
+ replay = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ self._deployment_select()
|
|
|
|
|
+ + "WHERE d.created_by = CAST(:actor_uid AS uuid) "
|
|
|
|
|
+ "AND d.environment = :environment "
|
|
|
|
|
+ "AND d.create_idempotency_key = :idempotency_key"
|
|
|
|
|
+ ),
|
|
|
|
|
+ params,
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .one_or_none()
|
|
|
|
|
+ )
|
|
|
|
|
+ if replay is None:
|
|
|
|
|
+ raise ValueError(
|
|
|
|
|
+ "preparatory deployment is unavailable or already claimed"
|
|
|
|
|
+ )
|
|
|
|
|
+ if (
|
|
|
|
|
+ str(replay["dataflow_version_id"]) != params["version_id"]
|
|
|
|
|
+ or str(replay["package_hash"]) != params["package_hash"]
|
|
|
|
|
+ or str(replay["binding_hash"]) != params["binding_hash"]
|
|
|
|
|
+ or str(replay["schedule_hash"]) != params["schedule_hash"]
|
|
|
|
|
+ ):
|
|
|
|
|
+ raise ValueError(
|
|
|
|
|
+ "deployment idempotency key has conflicting content"
|
|
|
|
|
+ )
|
|
|
|
|
+ return self._deployment_result(replay)
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "INSERT INTO public.dataflow_deployment_transitions "
|
|
|
|
|
+ "(id, deployment_id, from_status, to_status, actor_uid, "
|
|
|
|
|
+ "reason, correlation_id, detail) VALUES "
|
|
|
|
|
+ "(CAST(:transition_id AS uuid), CAST(:id AS uuid), "
|
|
|
|
|
+ "'disabled', 'draft', CAST(:actor_uid AS uuid), :reason, "
|
|
|
|
|
+ "CAST(:correlation_id AS uuid), "
|
|
|
|
|
+ "'{\"source\":\"preparatory_physical_gate\"}'::jsonb)"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {**params, "transition_id": new_governance_uid()},
|
|
|
|
|
+ )
|
|
|
|
|
+ return self.get_deployment(deployment_id)
|
|
|
|
|
+
|
|
|
|
|
+ def get_deployment(
|
|
|
|
|
+ self, deployment_id: str, *, for_update: bool = False
|
|
|
|
|
+ ) -> dict[str, Any]:
|
|
|
|
|
+ suffix = " FOR UPDATE OF d" if for_update else ""
|
|
|
|
|
+ row = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ self._deployment_select()
|
|
|
|
|
+ + "WHERE d.id = CAST(:id AS uuid)"
|
|
|
|
|
+ + suffix
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"id": _uid(deployment_id, "deployment_id")},
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .one_or_none()
|
|
|
|
|
+ )
|
|
|
|
|
+ if row is None or row["package"] is None:
|
|
|
|
|
+ raise ValueError("governed deployment was not found")
|
|
|
|
|
+ return self._deployment_result(row)
|
|
|
|
|
+
|
|
|
|
|
+ def claim_operation(
|
|
|
|
|
+ self,
|
|
|
|
|
+ deployment_id: str,
|
|
|
|
|
+ action: str,
|
|
|
|
|
+ idempotency_key: str,
|
|
|
|
|
+ actor_uid: str,
|
|
|
|
|
+ correlation_id: str,
|
|
|
|
|
+ reason: str,
|
|
|
|
|
+ request_payload: dict[str, Any],
|
|
|
|
|
+ ) -> tuple[bool, dict[str, Any]]:
|
|
|
|
|
+ if action not in {
|
|
|
|
|
+ "deploy_disabled",
|
|
|
|
|
+ "run_canary",
|
|
|
|
|
+ "activate",
|
|
|
|
|
+ "rollback",
|
|
|
|
|
+ }:
|
|
|
|
|
+ raise ValueError("deployment operation is invalid")
|
|
|
|
|
+ params = {
|
|
|
|
|
+ "id": new_governance_uid(),
|
|
|
|
|
+ "deployment_id": _uid(deployment_id, "deployment_id"),
|
|
|
|
|
+ "action": action,
|
|
|
|
|
+ "key": _text(idempotency_key, "idempotency_key", 200),
|
|
|
|
|
+ "actor_uid": _uid(actor_uid, "actor_uid"),
|
|
|
|
|
+ "correlation_id": _uid(correlation_id, "correlation_id"),
|
|
|
|
|
+ "reason": _text(reason, "reason", 1000),
|
|
|
|
|
+ "request": _json(
|
|
|
|
|
+ _bounded_object(
|
|
|
|
|
+ request_payload,
|
|
|
|
|
+ "deployment operation request",
|
|
|
|
|
+ maximum_bytes=131_072,
|
|
|
|
|
+ )
|
|
|
|
|
+ ),
|
|
|
|
|
+ "request_hash": _canonical_hash(request_payload),
|
|
|
|
|
+ }
|
|
|
|
|
+ row = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "INSERT INTO public.dataflow_deployment_operations "
|
|
|
|
|
+ "(id, deployment_id, action, idempotency_key, status, "
|
|
|
|
|
+ "actor_uid, correlation_id, reason, request, request_hash) VALUES "
|
|
|
|
|
+ "(CAST(:id AS uuid), CAST(:deployment_id AS uuid), "
|
|
|
|
|
+ ":action, :key, 'claimed', CAST(:actor_uid AS uuid), "
|
|
|
|
|
+ "CAST(:correlation_id AS uuid), :reason, "
|
|
|
|
|
+ "CAST(:request AS jsonb), :request_hash) "
|
|
|
|
|
+ "ON CONFLICT (deployment_id, action, idempotency_key) "
|
|
|
|
|
+ "DO NOTHING RETURNING id::text AS id, deployment_id::text, "
|
|
|
|
|
+ "action, status, actor_uid::text, correlation_id::text, "
|
|
|
|
|
+ "reason, request, request_hash, result, result_hash, error_code"
|
|
|
|
|
+ ),
|
|
|
|
|
+ params,
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .one_or_none()
|
|
|
|
|
+ )
|
|
|
|
|
+ claimed = row is not None
|
|
|
|
|
+ if row is None:
|
|
|
|
|
+ row = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "SELECT id::text AS id, deployment_id::text, action, "
|
|
|
|
|
+ "status, actor_uid::text, correlation_id::text, reason, "
|
|
|
|
|
+ "request, request_hash, result, result_hash, error_code "
|
|
|
|
|
+ "FROM public.dataflow_deployment_operations "
|
|
|
|
|
+ "WHERE deployment_id = CAST(:deployment_id AS uuid) "
|
|
|
|
|
+ "AND action = :action AND idempotency_key = :key"
|
|
|
|
|
+ ),
|
|
|
|
|
+ params,
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .one()
|
|
|
|
|
+ )
|
|
|
|
|
+ result = dict(row)
|
|
|
|
|
+ result["request"] = _object(
|
|
|
|
|
+ result["request"], "deployment operation request"
|
|
|
|
|
+ )
|
|
|
|
|
+ if _canonical_hash(result["request"]) != result.get("request_hash"):
|
|
|
|
|
+ raise ValueError("deployment operation request has drifted")
|
|
|
|
|
+ if (
|
|
|
|
|
+ str(result.get("actor_uid")) != params["actor_uid"]
|
|
|
|
|
+ or str(result.get("request_hash")) != params["request_hash"]
|
|
|
|
|
+ or str(result.get("reason")) != params["reason"]
|
|
|
|
|
+ ):
|
|
|
|
|
+ raise ValueError(
|
|
|
|
|
+ "deployment idempotency key is bound to another actor or request"
|
|
|
|
|
+ )
|
|
|
|
|
+ if result.get("result") is not None:
|
|
|
|
|
+ result["result"] = _object(
|
|
|
|
|
+ result["result"], "deployment operation result"
|
|
|
|
|
+ )
|
|
|
|
|
+ if _canonical_hash(result["result"]) != result.get("result_hash"):
|
|
|
|
|
+ raise ValueError("deployment operation result has drifted")
|
|
|
|
|
+ return claimed, result
|
|
|
|
|
+
|
|
|
|
|
+ def commit(self) -> None:
|
|
|
|
|
+ self.session.commit()
|
|
|
|
|
+
|
|
|
|
|
+ def get_operation(
|
|
|
|
|
+ self, deployment_id: str, action: str, idempotency_key: str
|
|
|
|
|
+ ) -> dict[str, Any]:
|
|
|
|
|
+ row = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "SELECT id::text AS id, deployment_id::text AS deployment_id, "
|
|
|
|
|
+ "action, status, actor_uid::text AS actor_uid, "
|
|
|
|
|
+ "correlation_id::text AS correlation_id, reason, request, "
|
|
|
|
|
+ "request_hash, result, result_hash, error_code "
|
|
|
|
|
+ "FROM public.dataflow_deployment_operations "
|
|
|
|
|
+ "WHERE deployment_id = CAST(:deployment_id AS uuid) "
|
|
|
|
|
+ "AND action = :action AND idempotency_key = :key"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {
|
|
|
|
|
+ "deployment_id": _uid(
|
|
|
|
|
+ deployment_id, "deployment_id"
|
|
|
|
|
+ ),
|
|
|
|
|
+ "action": _text(action, "action", 30),
|
|
|
|
|
+ "key": _text(
|
|
|
|
|
+ idempotency_key, "idempotency_key", 200
|
|
|
|
|
+ ),
|
|
|
|
|
+ },
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .one_or_none()
|
|
|
|
|
+ )
|
|
|
|
|
+ if row is None:
|
|
|
|
|
+ raise ValueError("deployment operation was not found")
|
|
|
|
|
+ result = dict(row)
|
|
|
|
|
+ result["request"] = _object(
|
|
|
|
|
+ result["request"], "deployment operation request"
|
|
|
|
|
+ )
|
|
|
|
|
+ if _canonical_hash(result["request"]) != result["request_hash"]:
|
|
|
|
|
+ raise ValueError("deployment operation request has drifted")
|
|
|
|
|
+ if result.get("result") is not None:
|
|
|
|
|
+ result["result"] = _object(
|
|
|
|
|
+ result["result"], "deployment operation result"
|
|
|
|
|
+ )
|
|
|
|
|
+ if _canonical_hash(result["result"]) != result["result_hash"]:
|
|
|
|
|
+ raise ValueError("deployment operation result has drifted")
|
|
|
|
|
+ return result
|
|
|
|
|
+
|
|
|
|
|
+ def reclaim_unknown_operation(self, operation_id: str) -> None:
|
|
|
|
|
+ updated = self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "UPDATE public.dataflow_deployment_operations "
|
|
|
|
|
+ "SET status = 'claimed', error_code = NULL, "
|
|
|
|
|
+ "updated_at = CURRENT_TIMESTAMP "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid) AND status = 'unknown'"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"id": _uid(operation_id, "operation_id")},
|
|
|
|
|
+ )
|
|
|
|
|
+ if getattr(updated, "rowcount", 0) != 1:
|
|
|
|
|
+ raise ValueError("unknown operation changed concurrently")
|
|
|
|
|
+
|
|
|
|
|
+ def fail_operation(self, operation_id: str, error_code: str) -> None:
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "UPDATE public.dataflow_deployment_operations "
|
|
|
|
|
+ "SET status = 'failed', error_code = :error_code, "
|
|
|
|
|
+ "failed_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid) AND status = 'claimed'"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": _uid(operation_id, "operation_id"),
|
|
|
|
|
+ "error_code": _text(error_code, "error_code", 100),
|
|
|
|
|
+ },
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ def mark_operation_unknown(
|
|
|
|
|
+ self, operation_id: str, error_code: str
|
|
|
|
|
+ ) -> None:
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "UPDATE public.dataflow_deployment_operations "
|
|
|
|
|
+ "SET status = 'unknown', error_code = :error_code, "
|
|
|
|
|
+ "updated_at = CURRENT_TIMESTAMP "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid) AND status = 'claimed'"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": _uid(operation_id, "operation_id"),
|
|
|
|
|
+ "error_code": _text(error_code, "error_code", 100),
|
|
|
|
|
+ },
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ def _complete_deployment_operation(
|
|
|
|
|
+ self, operation_id: str, result: dict[str, Any]
|
|
|
|
|
+ ) -> None:
|
|
|
|
|
+ operation = _uid(operation_id, "operation_id")
|
|
|
|
|
+ normalized = _bounded_object(result, "operation result")
|
|
|
|
|
+ updated = self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "UPDATE public.dataflow_deployment_operations "
|
|
|
|
|
+ "SET status = 'completed', result = CAST(:result AS jsonb), "
|
|
|
|
|
+ "result_hash = :result_hash, completed_at = CURRENT_TIMESTAMP, "
|
|
|
|
|
+ "updated_at = CURRENT_TIMESTAMP "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid) AND status = 'claimed'"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": operation,
|
|
|
|
|
+ "result": _json(normalized),
|
|
|
|
|
+ "result_hash": _canonical_hash(normalized),
|
|
|
|
|
+ },
|
|
|
|
|
+ )
|
|
|
|
|
+ if getattr(updated, "rowcount", 0) != 1:
|
|
|
|
|
+ raise ValueError("deployment operation state changed concurrently")
|
|
|
|
|
+
|
|
|
|
|
+ def _operation_context(self, operation_id: str):
|
|
|
|
|
+ return (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "SELECT actor_uid::text AS actor_uid, "
|
|
|
|
|
+ "correlation_id::text AS correlation_id, reason "
|
|
|
|
|
+ "FROM public.dataflow_deployment_operations "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid)"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"id": _uid(operation_id, "operation_id")},
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .one()
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ def _record_deployment_transition(
|
|
|
|
|
+ self,
|
|
|
|
|
+ deployment_id: str,
|
|
|
|
|
+ operation_id: str,
|
|
|
|
|
+ from_status: str,
|
|
|
|
|
+ to_status: str,
|
|
|
|
|
+ detail: dict[str, Any],
|
|
|
|
|
+ ) -> None:
|
|
|
|
|
+ operation = self._operation_context(operation_id)
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "INSERT INTO public.dataflow_deployment_transitions "
|
|
|
|
|
+ "(id, deployment_id, operation_id, from_status, to_status, "
|
|
|
|
|
+ "actor_uid, reason, correlation_id, detail) VALUES "
|
|
|
|
|
+ "(CAST(:id AS uuid), CAST(:deployment_id AS uuid), "
|
|
|
|
|
+ "CAST(:operation_id AS uuid), :from_status, :to_status, "
|
|
|
|
|
+ "CAST(:actor_uid AS uuid), :reason, "
|
|
|
|
|
+ "CAST(:correlation_id AS uuid), CAST(:detail AS jsonb))"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": new_governance_uid(),
|
|
|
|
|
+ "deployment_id": _uid(deployment_id, "deployment_id"),
|
|
|
|
|
+ "operation_id": _uid(operation_id, "operation_id"),
|
|
|
|
|
+ "from_status": from_status,
|
|
|
|
|
+ "to_status": to_status,
|
|
|
|
|
+ "actor_uid": operation["actor_uid"],
|
|
|
|
|
+ "reason": operation["reason"],
|
|
|
|
|
+ "correlation_id": operation["correlation_id"],
|
|
|
|
|
+ "detail": _json(detail),
|
|
|
|
|
+ },
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ def mark_disabled(
|
|
|
|
|
+ self,
|
|
|
|
|
+ deployment_id: str,
|
|
|
|
|
+ expected_version: int,
|
|
|
|
|
+ engine: dict[str, Any],
|
|
|
|
|
+ operation_id: str,
|
|
|
|
|
+ ) -> dict[str, Any]:
|
|
|
|
|
+ deployment = _uid(deployment_id, "deployment_id")
|
|
|
|
|
+ current = self.get_deployment(deployment, for_update=True)
|
|
|
|
|
+ if (
|
|
|
|
|
+ current["lock_version"] != expected_version
|
|
|
|
|
+ or current["status"] not in {"draft", "failed"}
|
|
|
|
|
+ ):
|
|
|
|
|
+ raise ValueError("deployment state changed concurrently")
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "UPDATE public.dataflow_deployments SET status = 'disabled', "
|
|
|
|
|
+ "engine_namespace = :namespace, "
|
|
|
|
|
+ "engine_definition_id = :definition_id, "
|
|
|
|
|
+ "engine_revision = :revision, "
|
|
|
|
|
+ "engine_definition_hash = :definition_hash, "
|
|
|
|
|
+ "deploy_receipt = CAST(:receipt AS jsonb), "
|
|
|
|
|
+ "failure_code = NULL, lock_version = lock_version + 1, "
|
|
|
|
|
+ "updated_at = CURRENT_TIMESTAMP "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid) AND lock_version = :version"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": deployment,
|
|
|
|
|
+ "version": expected_version,
|
|
|
|
|
+ "namespace": _text(engine["namespace"], "namespace", 200),
|
|
|
|
|
+ "definition_id": _text(
|
|
|
|
|
+ engine["definition_id"], "definition_id", 200
|
|
|
|
|
+ ),
|
|
|
|
|
+ "revision": (
|
|
|
|
|
+ None
|
|
|
|
|
+ if engine.get("revision") is None
|
|
|
|
|
+ else str(engine["revision"])[:200]
|
|
|
|
|
+ ),
|
|
|
|
|
+ "definition_hash": _digest(
|
|
|
|
|
+ engine["definition_hash"], "definition_hash"
|
|
|
|
|
+ ),
|
|
|
|
|
+ "receipt": _json(
|
|
|
|
|
+ _bounded_object(engine.get("receipt") or {}, "deploy receipt")
|
|
|
|
|
+ ),
|
|
|
|
|
+ },
|
|
|
|
|
+ )
|
|
|
|
|
+ result = self.get_deployment(deployment)
|
|
|
|
|
+ self._record_deployment_transition(
|
|
|
|
|
+ deployment,
|
|
|
|
|
+ operation_id,
|
|
|
|
|
+ current["status"],
|
|
|
|
|
+ "disabled",
|
|
|
|
|
+ {
|
|
|
|
|
+ "engine_namespace": result["engine_namespace"],
|
|
|
|
|
+ "engine_definition_id": result["engine_definition_id"],
|
|
|
|
|
+ "engine_definition_hash": result["engine_definition_hash"],
|
|
|
|
|
+ },
|
|
|
|
|
+ )
|
|
|
|
|
+ self._complete_deployment_operation(operation_id, result)
|
|
|
|
|
+ return result
|
|
|
|
|
+
|
|
|
|
|
+ def record_canary(
|
|
|
|
|
+ self,
|
|
|
|
|
+ deployment_id: str,
|
|
|
|
|
+ expected_version: int,
|
|
|
|
|
+ evidence: dict[str, Any],
|
|
|
|
|
+ operation_id: str,
|
|
|
|
|
+ ) -> dict[str, Any]:
|
|
|
|
|
+ deployment = _uid(deployment_id, "deployment_id")
|
|
|
|
|
+ current = self.get_deployment(deployment, for_update=True)
|
|
|
|
|
+ if (
|
|
|
|
|
+ current["lock_version"] != expected_version
|
|
|
|
|
+ or current["status"] not in {"disabled", "canary"}
|
|
|
|
|
+ ):
|
|
|
|
|
+ raise ValueError("deployment state changed concurrently")
|
|
|
|
|
+ value = _bounded_object(evidence, "canary evidence")
|
|
|
|
|
+ params = {
|
|
|
|
|
+ **value,
|
|
|
|
|
+ "physical_plan_hashes": _json(value["physical_plan_hashes"]),
|
|
|
|
|
+ "engine_response": _json(value.get("engine_response") or {}),
|
|
|
|
|
+ }
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "INSERT INTO public.dataflow_canary_evidence "
|
|
|
|
|
+ "(id, deployment_id, execution_id, status, package_hash, "
|
|
|
|
|
+ "binding_hash, schema_snapshot_hash, physical_plan_hashes, "
|
|
|
|
|
+ "workflow_spec_hash, schedule_hash, engine_definition_hash, "
|
|
|
|
|
+ "verified_by, engine_response, expires_at) VALUES "
|
|
|
|
|
+ "(CAST(:id AS uuid), CAST(:deployment_id AS uuid), "
|
|
|
|
|
+ ":execution_id, :status, :package_hash, :binding_hash, "
|
|
|
|
|
+ ":schema_snapshot_hash, CAST(:physical_plan_hashes AS jsonb), "
|
|
|
|
|
+ ":workflow_spec_hash, :schedule_hash, "
|
|
|
|
|
+ ":engine_definition_hash, CAST(:verified_by AS uuid), "
|
|
|
|
|
+ "CAST(:engine_response AS jsonb), CAST(:expires_at AS timestamptz))"
|
|
|
|
|
+ ),
|
|
|
|
|
+ params,
|
|
|
|
|
+ )
|
|
|
|
|
+ next_status = "canary" if value["status"] == "passed" else current["status"]
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "UPDATE public.dataflow_deployments SET status = :status, "
|
|
|
|
|
+ "canary_evidence_id = CASE WHEN :passed THEN CAST(:evidence_id AS uuid) "
|
|
|
|
|
+ "ELSE canary_evidence_id END, "
|
|
|
|
|
+ "lock_version = lock_version + 1, updated_at = CURRENT_TIMESTAMP "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid) AND lock_version = :version"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": deployment,
|
|
|
|
|
+ "version": expected_version,
|
|
|
|
|
+ "status": next_status,
|
|
|
|
|
+ "passed": value["status"] == "passed",
|
|
|
|
|
+ "evidence_id": value["id"],
|
|
|
|
|
+ },
|
|
|
|
|
+ )
|
|
|
|
|
+ result = self.get_canary_evidence(value["id"])
|
|
|
|
|
+ self._record_deployment_transition(
|
|
|
|
|
+ deployment,
|
|
|
|
|
+ operation_id,
|
|
|
|
|
+ current["status"],
|
|
|
|
|
+ next_status,
|
|
|
|
|
+ {
|
|
|
|
|
+ "canary_evidence_id": result["id"],
|
|
|
|
|
+ "execution_id": result["execution_id"],
|
|
|
|
|
+ "evidence_status": result["status"],
|
|
|
|
|
+ },
|
|
|
|
|
+ )
|
|
|
|
|
+ self._complete_deployment_operation(operation_id, result)
|
|
|
|
|
+ return result
|
|
|
|
|
+
|
|
|
|
|
+ def get_canary_evidence(self, evidence_id: str) -> dict[str, Any]:
|
|
|
|
|
+ row = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "SELECT id::text AS id, deployment_id::text AS deployment_id, "
|
|
|
|
|
+ "execution_id, status, package_hash, binding_hash, "
|
|
|
|
|
+ "schema_snapshot_hash, physical_plan_hashes, "
|
|
|
|
|
+ "workflow_spec_hash, schedule_hash, engine_definition_hash, "
|
|
|
|
|
+ "verified_by::text AS verified_by, engine_response, "
|
|
|
|
|
+ "expires_at, created_at "
|
|
|
|
|
+ "FROM public.dataflow_canary_evidence "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid)"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"id": _uid(evidence_id, "evidence_id")},
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .one_or_none()
|
|
|
|
|
+ )
|
|
|
|
|
+ if row is None:
|
|
|
|
|
+ raise ValueError("canary evidence was not found")
|
|
|
|
|
+ result = dict(row)
|
|
|
|
|
+ result["physical_plan_hashes"] = _array(
|
|
|
|
|
+ result["physical_plan_hashes"], "physical plan hashes"
|
|
|
|
|
+ )
|
|
|
|
|
+ result["engine_response"] = _object(
|
|
|
|
|
+ result["engine_response"], "engine response"
|
|
|
|
|
+ )
|
|
|
|
|
+ for key in ("expires_at", "created_at"):
|
|
|
|
|
+ if hasattr(result[key], "isoformat"):
|
|
|
|
|
+ result[key] = result[key].isoformat()
|
|
|
|
|
+ return result
|
|
|
|
|
+
|
|
|
|
|
+ def activate_atomic(
|
|
|
|
|
+ self,
|
|
|
|
|
+ deployment_id: str,
|
|
|
|
|
+ expected_version: int,
|
|
|
|
|
+ evidence_id: str,
|
|
|
|
|
+ actor_uid: str,
|
|
|
|
|
+ operation_id: str,
|
|
|
|
|
+ ) -> dict[str, Any]:
|
|
|
|
|
+ deployment = _uid(deployment_id, "deployment_id")
|
|
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
|
|
+ current = self.get_deployment(deployment, for_update=True)
|
|
|
|
|
+ if (
|
|
|
|
|
+ current["lock_version"] != expected_version
|
|
|
|
|
+ or current["status"] != "canary"
|
|
|
|
|
+ ):
|
|
|
|
|
+ raise ValueError("deployment state changed concurrently")
|
|
|
|
|
+ previous = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ self._deployment_select()
|
|
|
|
|
+ + "WHERE d.dataflow_uid = CAST(:uid AS uuid) "
|
|
|
|
|
+ "AND d.environment = :environment AND d.status = 'active' "
|
|
|
|
|
+ "AND d.id <> CAST(:id AS uuid) FOR UPDATE OF d"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {
|
|
|
|
|
+ "uid": current["dataflow_uid"],
|
|
|
|
|
+ "environment": current["environment"],
|
|
|
|
|
+ "id": deployment,
|
|
|
|
|
+ },
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .one_or_none()
|
|
|
|
|
+ )
|
|
|
|
|
+ previous_id = None
|
|
|
|
|
+ if previous is not None:
|
|
|
|
|
+ previous_id = str(previous["id"])
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "UPDATE public.dataflow_deployments "
|
|
|
|
|
+ "SET status = 'superseded', superseded_at = CURRENT_TIMESTAMP, "
|
|
|
|
|
+ "lock_version = lock_version + 1, updated_at = CURRENT_TIMESTAMP "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid) AND status = 'active'"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"id": previous_id},
|
|
|
|
|
+ )
|
|
|
|
|
+ updated = self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "UPDATE public.dataflow_deployments SET status = 'active', "
|
|
|
|
|
+ "previous_active_deployment_id = CAST(:previous_id AS uuid), "
|
|
|
|
|
+ "canary_evidence_id = CAST(:evidence_id AS uuid), "
|
|
|
|
|
+ "activated_by = CAST(:actor_uid AS uuid), "
|
|
|
|
|
+ "activated_at = CURRENT_TIMESTAMP, "
|
|
|
|
|
+ "lock_version = lock_version + 1, updated_at = CURRENT_TIMESTAMP "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid) AND lock_version = :version "
|
|
|
|
|
+ "AND status = 'canary'"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": deployment,
|
|
|
|
|
+ "version": expected_version,
|
|
|
|
|
+ "previous_id": previous_id,
|
|
|
|
|
+ "evidence_id": _uid(evidence_id, "evidence_id"),
|
|
|
|
|
+ "actor_uid": actor,
|
|
|
|
|
+ },
|
|
|
|
|
+ )
|
|
|
|
|
+ if getattr(updated, "rowcount", 0) != 1:
|
|
|
|
|
+ raise ValueError("deployment state changed concurrently")
|
|
|
|
|
+ result = self.get_deployment(deployment)
|
|
|
|
|
+ if previous_id:
|
|
|
|
|
+ self._record_deployment_transition(
|
|
|
|
|
+ previous_id,
|
|
|
|
|
+ operation_id,
|
|
|
|
|
+ "active",
|
|
|
|
|
+ "superseded",
|
|
|
|
|
+ {"superseded_by": deployment},
|
|
|
|
|
+ )
|
|
|
|
|
+ self._record_deployment_transition(
|
|
|
|
|
+ deployment,
|
|
|
|
|
+ operation_id,
|
|
|
|
|
+ "canary",
|
|
|
|
|
+ "active",
|
|
|
|
|
+ {
|
|
|
|
|
+ "canary_evidence_id": evidence_id,
|
|
|
|
|
+ "previous_active_deployment_id": previous_id,
|
|
|
|
|
+ },
|
|
|
|
|
+ )
|
|
|
|
|
+ self._complete_deployment_operation(operation_id, result)
|
|
|
|
|
+ return result
|
|
|
|
|
+
|
|
|
|
|
+ def rollback_atomic(
|
|
|
|
|
+ self,
|
|
|
|
|
+ deployment_id: str,
|
|
|
|
|
+ expected_version: int,
|
|
|
|
|
+ actor_uid: str,
|
|
|
|
|
+ operation_id: str,
|
|
|
|
|
+ ) -> dict[str, Any]:
|
|
|
|
|
+ deployment = _uid(deployment_id, "deployment_id")
|
|
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
|
|
+ current = self.get_deployment(deployment, for_update=True)
|
|
|
|
|
+ if (
|
|
|
|
|
+ current["lock_version"] != expected_version
|
|
|
|
|
+ or current["status"] not in {"active", "failed"}
|
|
|
|
|
+ or not current["previous_active_deployment_id"]
|
|
|
|
|
+ ):
|
|
|
|
|
+ raise ValueError("deployment state changed concurrently")
|
|
|
|
|
+ previous = self.get_deployment(
|
|
|
|
|
+ current["previous_active_deployment_id"], for_update=True
|
|
|
|
|
+ )
|
|
|
|
|
+ if (
|
|
|
|
|
+ previous["status"] != "superseded"
|
|
|
|
|
+ or previous["dataflow_uid"] != current["dataflow_uid"]
|
|
|
|
|
+ or previous["environment"] != current["environment"]
|
|
|
|
|
+ ):
|
|
|
|
|
+ raise ValueError("previous deployment snapshot is not restorable")
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "UPDATE public.dataflow_deployments "
|
|
|
|
|
+ "SET status = 'rolled_back', rolled_back_at = CURRENT_TIMESTAMP, "
|
|
|
|
|
+ "lock_version = lock_version + 1, updated_at = CURRENT_TIMESTAMP "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid) AND lock_version = :version"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"id": deployment, "version": expected_version},
|
|
|
|
|
+ )
|
|
|
|
|
+ restored = self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "UPDATE public.dataflow_deployments "
|
|
|
|
|
+ "SET status = 'active', activated_by = CAST(:actor_uid AS uuid), "
|
|
|
|
|
+ "activated_at = CURRENT_TIMESTAMP, superseded_at = NULL, "
|
|
|
|
|
+ "lock_version = lock_version + 1, updated_at = CURRENT_TIMESTAMP "
|
|
|
|
|
+ "WHERE id = CAST(:id AS uuid) AND status = 'superseded'"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"id": previous["id"], "actor_uid": actor},
|
|
|
|
|
+ )
|
|
|
|
|
+ if getattr(restored, "rowcount", 0) != 1:
|
|
|
|
|
+ raise ValueError("previous deployment state changed concurrently")
|
|
|
|
|
+ rolled_back = self.get_deployment(deployment)
|
|
|
|
|
+ active = self.get_deployment(previous["id"])
|
|
|
|
|
+ self._record_deployment_transition(
|
|
|
|
|
+ deployment,
|
|
|
|
|
+ operation_id,
|
|
|
|
|
+ current["status"],
|
|
|
|
|
+ "rolled_back",
|
|
|
|
|
+ {"restored_deployment_id": active["id"]},
|
|
|
|
|
+ )
|
|
|
|
|
+ self._record_deployment_transition(
|
|
|
|
|
+ active["id"],
|
|
|
|
|
+ operation_id,
|
|
|
|
|
+ "superseded",
|
|
|
|
|
+ "active",
|
|
|
|
|
+ {"rollback_source_deployment_id": deployment},
|
|
|
|
|
+ )
|
|
|
|
|
+ result = {"rolled_back": rolled_back, "active": active}
|
|
|
|
|
+ self._complete_deployment_operation(operation_id, result)
|
|
|
|
|
+ return result
|
|
|
|
|
+
|
|
|
|
|
+ def list_deployments(
|
|
|
|
|
+ self, environment: str | None = None
|
|
|
|
|
+ ) -> list[dict[str, Any]]:
|
|
|
|
|
+ where = "WHERE d.package IS NOT NULL "
|
|
|
|
|
+ params: dict[str, Any] = {}
|
|
|
|
|
+ if environment is not None:
|
|
|
|
|
+ where += "AND d.environment = :environment "
|
|
|
|
|
+ params["environment"] = environment
|
|
|
|
|
+ rows = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ self._deployment_select()
|
|
|
|
|
+ + where
|
|
|
|
|
+ + "ORDER BY d.updated_at DESC, d.id"
|
|
|
|
|
+ ),
|
|
|
|
|
+ params,
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .all()
|
|
|
|
|
+ )
|
|
|
|
|
+ items = [self._deployment_result(row) for row in rows]
|
|
|
|
|
+ if not items:
|
|
|
|
|
+ return items
|
|
|
|
|
+ deployment_ids = [item["id"] for item in items]
|
|
|
|
|
+ transitions = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "SELECT id::text AS id, deployment_id::text AS deployment_id, "
|
|
|
|
|
+ "from_status, to_status, reason, detail, created_at "
|
|
|
|
|
+ "FROM public.dataflow_deployment_transitions "
|
|
|
|
|
+ "WHERE deployment_id = ANY(CAST(:ids AS uuid[])) "
|
|
|
|
|
+ "ORDER BY created_at DESC, id DESC"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"ids": deployment_ids},
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .all()
|
|
|
|
|
+ )
|
|
|
|
|
+ unknown = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "SELECT deployment_id::text AS deployment_id, action, "
|
|
|
|
|
+ "idempotency_key, error_code, updated_at "
|
|
|
|
|
+ "FROM public.dataflow_deployment_operations "
|
|
|
|
|
+ "WHERE deployment_id = ANY(CAST(:ids AS uuid[])) "
|
|
|
|
|
+ "AND status = 'unknown' ORDER BY updated_at DESC"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"ids": deployment_ids},
|
|
|
|
|
+ )
|
|
|
|
|
+ .mappings()
|
|
|
|
|
+ .all()
|
|
|
|
|
+ )
|
|
|
|
|
+ by_transition = {deployment_id: [] for deployment_id in deployment_ids}
|
|
|
|
|
+ for row in transitions:
|
|
|
|
|
+ event = dict(row)
|
|
|
|
|
+ event["id"] = str(event["id"])
|
|
|
|
|
+ event["deployment_id"] = str(event["deployment_id"])
|
|
|
|
|
+ event["detail"] = _object(event["detail"], "transition detail")
|
|
|
|
|
+ if hasattr(event["created_at"], "isoformat"):
|
|
|
|
|
+ event["created_at"] = event["created_at"].isoformat()
|
|
|
|
|
+ by_transition[event["deployment_id"]].append(event)
|
|
|
|
|
+ by_unknown = {deployment_id: [] for deployment_id in deployment_ids}
|
|
|
|
|
+ for row in unknown:
|
|
|
|
|
+ value = dict(row)
|
|
|
|
|
+ value["deployment_id"] = str(value["deployment_id"])
|
|
|
|
|
+ if hasattr(value["updated_at"], "isoformat"):
|
|
|
|
|
+ value["updated_at"] = value["updated_at"].isoformat()
|
|
|
|
|
+ by_unknown[value["deployment_id"]].append(value)
|
|
|
|
|
+ active_by_scope = {
|
|
|
|
|
+ (item["dataflow_uid"], item["environment"]): item
|
|
|
|
|
+ for item in items
|
|
|
|
|
+ if item["status"] == "active"
|
|
|
|
|
+ }
|
|
|
|
|
+ by_id = {item["id"]: item for item in items}
|
|
|
|
|
+ for item in items:
|
|
|
|
|
+ item["transition_history"] = by_transition[item["id"]]
|
|
|
|
|
+ item["unknown_operations"] = by_unknown[item["id"]]
|
|
|
|
|
+ active = active_by_scope.get(
|
|
|
|
|
+ (item["dataflow_uid"], item["environment"])
|
|
|
|
|
+ )
|
|
|
|
|
+ item["current_active"] = (
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": active["id"],
|
|
|
|
|
+ "version_no": active["version_no"],
|
|
|
|
|
+ "package_hash": active["package_hash"],
|
|
|
|
|
+ }
|
|
|
|
|
+ if active
|
|
|
|
|
+ else None
|
|
|
|
|
+ )
|
|
|
|
|
+ previous = by_id.get(item.get("previous_active_deployment_id"))
|
|
|
|
|
+ item["previous_active"] = (
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": previous["id"],
|
|
|
|
|
+ "version_no": previous["version_no"],
|
|
|
|
|
+ "package_hash": previous["package_hash"],
|
|
|
|
|
+ "binding_hash": previous["binding_hash"],
|
|
|
|
|
+ }
|
|
|
|
|
+ if previous
|
|
|
|
|
+ else None
|
|
|
|
|
+ )
|
|
|
|
|
+ return items
|
|
|
|
|
+
|
|
|
|
|
+ def list_deployable_releases(
|
|
|
|
|
+ self, *, environment: str, limit: int = 50
|
|
|
|
|
+ ) -> list[dict[str, Any]]:
|
|
|
|
|
+ if environment not in {"development", "test", "production"}:
|
|
|
|
|
+ raise ValueError("deployment environment is invalid")
|
|
|
|
|
+ bounded_limit = int(limit)
|
|
|
|
|
+ if bounded_limit < 1 or bounded_limit > 100:
|
|
|
|
|
+ raise ValueError("candidate limit is invalid")
|
|
|
|
|
+ rows = (
|
|
|
|
|
+ self.session.execute(
|
|
|
|
|
+ text(
|
|
|
|
|
+ "SELECT DISTINCT v.id::text AS id "
|
|
|
|
|
+ "FROM public.dataflow_versions v "
|
|
|
|
|
+ "JOIN public.dataflow_deployments d "
|
|
|
|
|
+ "ON d.dataflow_version_id = v.id "
|
|
|
|
|
+ "WHERE v.status = 'released' AND v.package IS NOT NULL "
|
|
|
|
|
+ "AND d.environment = :environment AND d.package IS NULL "
|
|
|
|
|
+ "ORDER BY v.id DESC LIMIT :limit"
|
|
|
|
|
+ ),
|
|
|
|
|
+ {"environment": environment, "limit": bounded_limit},
|
|
|
|
|
+ )
|
|
|
|
|
+ .scalars()
|
|
|
|
|
+ .all()
|
|
|
|
|
+ )
|
|
|
|
|
+ candidates = []
|
|
|
|
|
+ for version_id in rows:
|
|
|
|
|
+ try:
|
|
|
|
|
+ candidate = self.load_deployable_release(
|
|
|
|
|
+ str(version_id), environment=environment
|
|
|
|
|
+ )
|
|
|
|
|
+ except ValueError:
|
|
|
|
|
+ continue
|
|
|
|
|
+ candidates.append(candidate)
|
|
|
|
|
+ return candidates
|