From 8115e6c142cdda4a1cc4f76b11c3b9b4738c2c24 Mon Sep 17 00:00:00 2001 From: Ning Zhou Date: Fri, 31 Jul 2026 14:06:30 +0800 Subject: [PATCH 1/2] feat: prove retained real feature restart recovery --- .github/workflows/ci.yml | 4 + ..._retained_real_feature_restart_recovery.py | 1239 +++++++++++++++++ data_agent/platform_truth.py | 26 + ..._retained_real_feature_restart_recovery.py | 241 ++++ data_agent/test_platform_truth.py | 7 + ...-retained-real-feature-restart-recovery.md | 81 ++ ...l-feature-restart-recovery-2026-07-31.json | 652 +++++++++ docs/roadmap.md | 4 +- docs/system-of-record-matrix-2026-07-24.md | 12 +- ...-retained-real-feature-restart-recovery.sh | 22 + 10 files changed, 2281 insertions(+), 7 deletions(-) create mode 100644 data_agent/metadata_fabric_retained_real_feature_restart_recovery.py create mode 100644 data_agent/test_metadata_fabric_retained_real_feature_restart_recovery.py create mode 100644 docs/architecture-decisions/adr-071-retained-real-feature-restart-recovery.md create mode 100644 docs/evidence/metadata-fabric-retained-real-feature-restart-recovery-2026-07-31.json create mode 100755 scripts/metadata-fabric-retained-real-feature-restart-recovery.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 74f6bc45..8dee4e41 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -208,6 +208,9 @@ jobs: - name: Validate retained real-feature terminal success evidence run: python -m data_agent.metadata_fabric_retained_real_feature_terminal_success validate + - name: Validate retained real-feature restart recovery evidence + run: python -m data_agent.metadata_fabric_retained_real_feature_restart_recovery validate + - name: Validate Active Metadata consumer deployment boundary run: python -m data_agent.active_metadata_consumer_deployment validate @@ -306,6 +309,7 @@ jobs: data_agent/test_metadata_fabric_real_feature_ingestion.py \ data_agent/test_metadata_fabric_real_feature_ledger_promotion.py \ data_agent/test_metadata_fabric_retained_real_feature_terminal_success.py \ + data_agent/test_metadata_fabric_retained_real_feature_restart_recovery.py \ data_agent/test_metadata_fabric_lineage_delivery.py \ data_agent/test_metadata_fabric_provider_identity.py \ data_agent/test_metadata_fabric_gravitino_identity.py \ diff --git a/data_agent/metadata_fabric_retained_real_feature_restart_recovery.py b/data_agent/metadata_fabric_retained_real_feature_restart_recovery.py new file mode 100644 index 00000000..92682472 --- /dev/null +++ b/data_agent/metadata_fabric_retained_real_feature_restart_recovery.py @@ -0,0 +1,1239 @@ +"""Verify restart recovery of the retained M3-24 real-feature authority chain. + +M3-25 attaches to the exact retained namespace, PVC-backed JDBC/S3 Iceberg +runtime and dedicated GDA Control PostgreSQL database recorded by M3-24. It +does not ingest again or create a new authority record. Instead, it records a +read-only baseline, restarts every retained stateful process in dependency +order, and requires byte-stable material, catalog and ledger readback before +an exact terminal replay. + +This is a bounded local process-restart rehearsal. It does not prove backup +restore, point-in-time recovery, independent failure domains or production +readiness. +""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import subprocess +import time +from collections.abc import Mapping +from datetime import UTC, datetime +from pathlib import Path +from typing import Any +from uuid import UUID + +from pydantic import SecretStr +from sqlalchemy import create_engine, text + +from . import metadata_fabric_object_store_active_metadata_promotion as m321 +from . import metadata_fabric_retained_real_feature_terminal_success as m324 +from . import metadata_fabric_spark_object_store_interoperability as m310 +from .platform_contracts import RunStatus, canonical_json_fingerprint +from .platform_gateway import PlatformGateway + +CONTRACT_SCHEMA = "gda.retained_real_feature_restart_recovery_contract.v1" +OBSERVATION_SCHEMA = "gda.retained_real_feature_restart_recovery_observation.v1" +EVIDENCE_SCHEMA = "gda.retained_real_feature_restart_recovery_evidence.v1" +VALIDATION_SCHEMA = "gda.retained_real_feature_restart_recovery_validation.v1" +SOURCE_EVIDENCE_SHA256 = "d966668b5a2ea57c7a4b2a3bc9824daab9b0128d9f94e515d7be649b145de418" +REPO_ROOT = Path(__file__).resolve().parent.parent +DEFAULT_SOURCE_EVIDENCE_PATH = m324.DEFAULT_EVIDENCE_PATH +DEFAULT_EVIDENCE_PATH = ( + REPO_ROOT + / "docs/evidence/metadata-fabric-retained-real-feature-restart-recovery-2026-07-31.json" +) +DEFAULT_WRAPPER_PATH = ( + REPO_ROOT / "scripts/metadata-fabric-retained-real-feature-restart-recovery.sh" +) +PROVIDER_OBSERVATION_ID = UUID("85bead83-1901-5649-b6e4-fe46d01c9ea9") +EXPECTED_LEDGER_COUNTS = { + "artifacts": 5, + "execution_plans": 1, + "policy_decisions": 1, + "approvals": 1, + "evaluator_evidence": 1, + "attempts": 2, + "quality_results": 1, + "lineage_events": 1, + "run_events": 4, +} +FALSE_CLAIMS = ( + "source_dataset_committed", + "source_absolute_path_committed", + "source_feature_payload_committed", + "new_ingestion_executed", + "new_authority_facts_created", + "persistent_scheduler_verified", + "protected_workload_identity_verified", + "durable_catalog_verified", + "production_object_store_verified", + "production_scheduler_verified", + "production_ingestion_verified", + "production_tenant_attestation_verified", + "backup_restore_verified", + "point_in_time_recovery_verified", + "independent_failure_domains_verified", + "production_restart_recovery_verified", + "oidc_verified", + "tls_verified", + "production_ready", +) + + +class RetainedRealFeatureRestartRecoveryError(RuntimeError): + """The retained real-feature restart/recovery gate failed closed.""" + + +def _mapping(value: Any) -> dict[str, Any]: + return dict(value) if isinstance(value, Mapping) else {} + + +def _load_json_object(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise TypeError("JSON document must be an object") + return value + + +def _file_record(path: Path) -> dict[str, Any]: + try: + payload = path.read_bytes() + relative = str(path.resolve().relative_to(REPO_ROOT)) + except (OSError, ValueError): + return {"path": None, "size_bytes": None, "sha256": None} + return { + "path": relative, + "size_bytes": len(payload), + "sha256": hashlib.sha256(payload).hexdigest(), + } + + +def _run_command(args: list[str], *, label: str, timeout: float = 180) -> str: + try: + completed = subprocess.run( + args, + check=False, + capture_output=True, + text=True, + timeout=timeout, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise RetainedRealFeatureRestartRecoveryError(f"{label} is unavailable") from exc + if completed.returncode != 0: + raise RetainedRealFeatureRestartRecoveryError(f"{label} failed") + return completed.stdout.strip() + + +def _checked_source_evidence( + path: Path = DEFAULT_SOURCE_EVIDENCE_PATH, +) -> dict[str, Any]: + evidence = _load_json_object(path) + errors = m324.validate_evidence(evidence) + if errors or evidence.get("evidence_sha256") != SOURCE_EVIDENCE_SHA256: + raise RetainedRealFeatureRestartRecoveryError( + "checked M3-24 evidence is unavailable or drifted" + ) + return evidence + + +def build_contract_report() -> dict[str, Any]: + errors: list[str] = [] + files = { + "restart_recovery": _file_record(Path(__file__).resolve()), + "wrapper": _file_record(DEFAULT_WRAPPER_PATH), + "terminal_success": _file_record(Path(m324.__file__).resolve()), + "terminal_success_evidence": _file_record(DEFAULT_SOURCE_EVIDENCE_PATH), + } + try: + _checked_source_evidence() + except (OSError, TypeError, ValueError, RetainedRealFeatureRestartRecoveryError): + errors.append("M3-24 checked evidence is unavailable") + if files["wrapper"]["sha256"] is None: + errors.append("M3-25 wrapper is unavailable") + stable = { + "schema": CONTRACT_SCHEMA, + "source_evidence_sha256": SOURCE_EVIDENCE_SHA256, + "files": files, + "requires_same_retention_identity": True, + "requires_ordered_stateful_restart": True, + "requires_stable_namespace_statefulset_service_pvc_identity": True, + "requires_kubernetes_pod_rotation": True, + "requires_control_container_and_volume_identity": True, + "requires_control_process_rotation": True, + "requires_iceberg_snapshot_and_object_continuity": True, + "requires_independent_parquet_re_evaluation": True, + "requires_gravitino_table_readback": True, + "requires_control_ledger_fingerprint_continuity": True, + "requires_exact_terminal_replay_without_new_facts": True, + "creates_new_resource_version_or_run": False, + "retained_local_restart_is_production_recovery": False, + "writes_to_legacy": False, + "errors": errors, + } + return { + **stable, + "status": "valid" if not errors else "invalid", + "contract_sha256": canonical_json_fingerprint(stable), + **{claim: False for claim in FALSE_CLAIMS}, + } + + +def _decode_runtime_material(secret: Mapping[str, Any], key: str, *, label: str) -> SecretStr: + encoded = _mapping(secret.get("data")).get(key) + if not isinstance(encoded, str) or not encoded: + raise RetainedRealFeatureRestartRecoveryError(f"{label} runtime material is unavailable") + try: + value = base64.b64decode(encoded, validate=True).decode("utf-8") + except (ValueError, UnicodeDecodeError) as exc: + raise RetainedRealFeatureRestartRecoveryError( + f"{label} runtime material is invalid" + ) from exc + if not value: + raise RetainedRealFeatureRestartRecoveryError(f"{label} runtime material is empty") + return SecretStr(value) + + +def _read_runtime_materials( + runtime: m310.IsolatedSparkObjectStoreRuntime, +) -> tuple[SecretStr, SecretStr, SecretStr, SecretStr]: + namespace = runtime.profile.cluster.rehearsal_namespace + catalog = runtime.kubectl.get_json( + [ + "-n", + namespace, + "get", + "secret", + "gravitino-persistence-runtime", + ], + label="retained catalog runtime material lookup", + ) + object_store = runtime.kubectl.get_json( + [ + "-n", + namespace, + "get", + "secret", + "metadata-object-store-runtime", + ], + label="retained object-store runtime material lookup", + ) + if catalog is None or object_store is None: + raise RetainedRealFeatureRestartRecoveryError( + "retained runtime material objects are unavailable" + ) + return ( + _decode_runtime_material(catalog, "admin-password", label="catalog admin"), + _decode_runtime_material(catalog, "database-password", label="catalog database"), + _decode_runtime_material(object_store, "access-key-id", label="object store user"), + _decode_runtime_material( + object_store, "secret-access-key", label="object store credential" + ), + ) + + +def _extract_control_password(environment: Any) -> SecretStr: + if not isinstance(environment, list): + raise RetainedRealFeatureRestartRecoveryError("retained control environment is invalid") + values = [ + item.removeprefix("POSTGRES_PASSWORD=") + for item in environment + if isinstance(item, str) and item.startswith("POSTGRES_PASSWORD=") + ] + if len(values) != 1 or not values[0]: + raise RetainedRealFeatureRestartRecoveryError( + "retained control database credential is unavailable" + ) + return SecretStr(values[0]) + + +class RetainedControlAttachment: + """Attach to and restart the identity-bound M3-24 control database.""" + + def __init__( + self, + source: Mapping[str, Any], + retention: m324.RetainedMaterialObservation, + ) -> None: + recorded = _mapping(source.get("control_database")) + self.container_name = str(recorded.get("container_name") or "") + self.volume_name = str(recorded.get("volume_name") or "") + self.host_port = int(recorded.get("host_port") or 0) + self.retention_id = retention.retention_id + self.expires_at = retention.expires_at + expected_container = f"gda-m3-24-control-{self.retention_id.removeprefix('m3-24-')[:24]}" + if ( + self.container_name != expected_container + or self.volume_name != self.container_name + or self.host_port <= 0 + or recorded.get("database_ref") != self.database_ref + ): + raise RetainedRealFeatureRestartRecoveryError( + "retained control database identity does not match M3-24 evidence" + ) + environment = json.loads( + _run_command( + [ + "docker", + "container", + "inspect", + self.container_name, + "--format", + "{{json .Config.Env}}", + ], + label="retained control database environment lookup", + ) + ) + self.password = _extract_control_password(environment) + + @property + def database_ref(self) -> str: + return f"docker:{self.container_name}/postgres" + + @property + def database_url(self) -> str: + return ( + "postgresql://postgres:" + f"{self.password.get_secret_value()}@127.0.0.1:{self.host_port}/postgres" + ) + + def observe(self) -> dict[str, Any]: + state = json.loads( + _run_command( + [ + "docker", + "container", + "inspect", + self.container_name, + "--format", + "{{json .State}}", + ], + label="retained control database state lookup", + ) + ) + container_id = _run_command( + [ + "docker", + "container", + "inspect", + self.container_name, + "--format", + "{{.Id}}", + ], + label="retained control container identity lookup", + ) + labels = json.loads( + _run_command( + [ + "docker", + "container", + "inspect", + self.container_name, + "--format", + "{{json .Config.Labels}}", + ], + label="retained control container labels lookup", + ) + ) + mounts = json.loads( + _run_command( + [ + "docker", + "container", + "inspect", + self.container_name, + "--format", + "{{json .Mounts}}", + ], + label="retained control container mounts lookup", + ) + ) + volume = json.loads( + _run_command( + [ + "docker", + "volume", + "inspect", + self.volume_name, + "--format", + "{{json .}}", + ], + label="retained control volume lookup", + ) + ) + volume_labels = _mapping(volume).get("Labels") + expected_expiry = self.expires_at.isoformat().replace("+00:00", "Z") + expected_labels = { + "gda.retention-id": self.retention_id, + "gda.owner": "team:metadata-platform", + "gda.expires-at": expected_expiry, + } + volume_mounts = [ + _mapping(item) + for item in mounts + if _mapping(item).get("Type") == "volume" + and _mapping(item).get("Destination") == "/var/lib/postgresql/data" + ] + if ( + not isinstance(labels, dict) + or any(labels.get(key) != value for key, value in expected_labels.items()) + or not isinstance(volume_labels, dict) + or any(volume_labels.get(key) != value for key, value in expected_labels.items()) + or len(volume_mounts) != 1 + or volume_mounts[0].get("Name") != self.volume_name + ): + raise RetainedRealFeatureRestartRecoveryError( + "retained control ownership or volume binding drifted" + ) + return { + "database_ref": self.database_ref, + "container_name": self.container_name, + "container_id": container_id, + "container_running": state.get("Running") is True, + "container_status": state.get("Status"), + "process_id": state.get("Pid"), + "started_at": state.get("StartedAt"), + "volume_name": self.volume_name, + "volume_retained": isinstance(volume, dict), + "host_port": self.host_port, + "retention_id": self.retention_id, + "owner": "team:metadata-platform", + "expires_at": expected_expiry, + "credential_material_recorded": False, + } + + def wait_ready(self, timeout_seconds: float = 120) -> None: + deadline = time.monotonic() + timeout_seconds + while time.monotonic() < deadline: + engine = create_engine(self.database_url, pool_pre_ping=True) + try: + with engine.connect() as connection: + connection.execute(text("SELECT 1")).scalar_one() + return + except Exception: + time.sleep(1) + finally: + engine.dispose() + raise RetainedRealFeatureRestartRecoveryError("retained control database did not recover") + + def restart(self) -> dict[str, Any]: + before = self.observe() + _run_command( + ["docker", "restart", self.container_name], + label="retained control database restart", + timeout=180, + ) + self.wait_ready() + return {"before": before, "after": self.observe()} + + +def _attach_runtime( + source: Mapping[str, Any], +) -> tuple[ + m310.IsolatedSparkObjectStoreRuntime, + m324.m322.RealFeatureIngestionProfile, +]: + profile = m324.m322.load_profile() + _, runtime_profile = m324.m322._load_dependencies(profile) + runtime = m310.IsolatedSparkObjectStoreRuntime(runtime_profile) + runtime.gravitino_host_image_id = runtime._inspect_host_image( + runtime_profile.runtime.gravitino_image, + runtime_profile.runtime.gravitino_host_image_id, + "Gravitino", + ) + runtime.spark_host_image_id = runtime._inspect_host_image( + runtime_profile.runtime.spark_image, + runtime_profile.runtime.spark_host_image_id, + "Spark", + ) + runtime.minio_host_image_id = runtime._inspect_host_image( + runtime_profile.runtime.minio_image, + runtime_profile.runtime.minio_host_image_id, + "MinIO", + ) + namespace = runtime_profile.cluster.rehearsal_namespace + schema_object = runtime.kubectl.get_json( + [ + "-n", + namespace, + "get", + "configmap", + "gravitino-persistence-schema", + ], + label="retained catalog schema lookup", + ) + schema_sql = _mapping(_mapping(schema_object).get("data")).get("001-schema.sql") + if not isinstance(schema_sql, str): + raise RetainedRealFeatureRestartRecoveryError("retained catalog schema is unavailable") + runtime.schema_sha256 = hashlib.sha256(schema_sql.encode()).hexdigest() + expected_schema = _mapping(source.get("initial_runtime")).get("source_schema_sha256") + if runtime.schema_sha256 != expected_schema: + raise RetainedRealFeatureRestartRecoveryError("retained catalog schema fingerprint drifted") + return runtime, profile + + +def _observe_runtime( + runtime: m310.IsolatedSparkObjectStoreRuntime, +) -> dict[str, Any]: + observed = runtime.observe_runtime() + namespace = runtime.profile.cluster.rehearsal_namespace + service = runtime.kubectl.get_json( + [ + "-n", + namespace, + "get", + "service", + "gravitino-persistence-postgresql", + ], + label="retained PostgreSQL service observation", + ) + if service is None: + raise RetainedRealFeatureRestartRecoveryError("retained PostgreSQL service is unavailable") + return { + **observed, + "postgresql_service": runtime._service_projection(service), + } + + +def _runtime_stable_projection(value: Mapping[str, Any]) -> dict[str, Any]: + result = { + key: value.get(key) + for key in ( + "context", + "gravitino_host_image_id", + "spark_host_image_id", + "minio_host_image_id", + "namespace", + "service", + "object_store_service", + "postgresql_service", + "iceberg_rest", + "gravitino_jdbc_driver_mounted", + "gravitino_aws_sdk_mounted", + "source_schema_sha256", + ) + if key in value + } + for name in ("postgresql", "object_store", "gravitino"): + workload = _mapping(value.get(name)) + result[name] = {key: nested for key, nested in workload.items() if key not in {"pod_uid"}} + return result + + +def _valid_uuid(value: Any) -> bool: + try: + UUID(str(value)) + except (TypeError, ValueError): + return False + return True + + +def _runtime_continuity_errors( + before: Mapping[str, Any], + after: Mapping[str, Any], + predecessor: Mapping[str, Any], +) -> list[str]: + errors: list[str] = [] + before_stable = _runtime_stable_projection(before) + after_stable = _runtime_stable_projection(after) + predecessor_stable = _runtime_stable_projection(predecessor) + before_without_new_service = { + key: value for key, value in before_stable.items() if key != "postgresql_service" + } + if before_stable != after_stable: + errors.append("retained Kubernetes stable runtime identity changed") + if before_without_new_service != predecessor_stable: + errors.append("retained Kubernetes runtime no longer binds M3-24") + for service_name in ( + "service", + "object_store_service", + "postgresql_service", + ): + old = _mapping(before.get(service_name)) + new = _mapping(after.get(service_name)) + if not _valid_uuid(old.get("uid")) or old != new: + errors.append(f"{service_name} identity changed") + for workload_name in ("postgresql", "object_store", "gravitino"): + old = _mapping(before.get(workload_name)) + new = _mapping(after.get(workload_name)) + if not _valid_uuid(old.get("statefulset_uid")) or old.get("statefulset_uid") != new.get( + "statefulset_uid" + ): + errors.append(f"{workload_name} StatefulSet identity changed") + if ( + not _valid_uuid(old.get("pod_uid")) + or not _valid_uuid(new.get("pod_uid")) + or old.get("pod_uid") == new.get("pod_uid") + ): + errors.append(f"{workload_name} pod did not rotate") + if old.get("ready_replicas") != 1 or new.get("ready_replicas") != 1: + errors.append(f"{workload_name} was not ready around restart") + old_pvc = old.get("pvc") + new_pvc = new.get("pvc") + if old_pvc != new_pvc: + errors.append(f"{workload_name} PVC identity changed") + if isinstance(old_pvc, Mapping) and _mapping(old_pvc).get("phase") != "Bound": + errors.append(f"{workload_name} PVC is not bound") + return errors + + +def _restart_kubernetes_runtime( + runtime: m310.IsolatedSparkObjectStoreRuntime, + before: Mapping[str, Any], +) -> dict[str, Any]: + namespace = runtime.profile.cluster.rehearsal_namespace + order = ( + "statefulset/gravitino-persistence-postgresql", + "statefulset/metadata-object-store", + "statefulset/gravitino-persistence", + ) + for workload in order: + runtime.kubectl.run( + ["-n", namespace, "rollout", "restart", workload], + label=f"M3-25 {workload} restart", + ) + runtime.kubectl.run( + ["-n", namespace, "rollout", "status", workload, "--timeout=10m"], + timeout=660, + label=f"M3-25 {workload} restart rollout", + ) + return { + "order": list(order), + "before": dict(before), + "after": _observe_runtime(runtime), + } + + +def _control_continuity_errors(restart: Mapping[str, Any]) -> list[str]: + errors: list[str] = [] + before = _mapping(restart.get("before")) + after = _mapping(restart.get("after")) + for key in ( + "database_ref", + "container_name", + "container_id", + "volume_name", + "host_port", + "retention_id", + "owner", + "expires_at", + ): + if before.get(key) != after.get(key): + errors.append(f"retained control identity changed: {key}") + if ( + before.get("container_running") is not True + or after.get("container_running") is not True + or before.get("volume_retained") is not True + or after.get("volume_retained") is not True + ): + errors.append("retained control database was not ready around restart") + if ( + not isinstance(before.get("process_id"), int) + or not isinstance(after.get("process_id"), int) + or before.get("process_id") == after.get("process_id") + ): + errors.append("retained control PostgreSQL process did not rotate") + if not before.get("started_at") or before.get("started_at") == after.get("started_at"): + errors.append("retained control PostgreSQL start time did not rotate") + return errors + + +def _source_payload_absent( + runtime: m310.IsolatedSparkObjectStoreRuntime, +) -> bool: + value = runtime.kubectl.get_json( + [ + "-n", + runtime.profile.cluster.rehearsal_namespace, + "get", + "configmap", + "real-feature-ingestion-input", + ], + allow_not_found=True, + label="retained source payload absence probe", + ) + return value is None + + +def _start_forward( + runtime: m310.IsolatedSparkObjectStoreRuntime, + *, + service: str, + target_port: int, +) -> Any: + forward = m321.provider_metrics._PortForward( + kubectl="kubectl", + context=runtime.profile.cluster.context, + namespace=runtime.profile.cluster.rehearsal_namespace, + service=service, + target_port=target_port, + ) + forward.start() + return forward + + +def _material_projection(store: Mapping[str, Any]) -> dict[str, Any]: + latest = _mapping(store.get("latest_metadata")) + return { + "object_count": store.get("object_count"), + "object_inventory_sha256": store.get("object_inventory_sha256"), + "data_file_count": len(store.get("data_keys") or []), + "metadata_file_count": len(store.get("metadata_keys") or []), + "manifest_file_count": len(store.get("manifest_keys") or []), + "metadata_body_sha256": latest.get("body_sha256"), + "snapshot_id": latest.get("current_snapshot_id"), + "schema_id": latest.get("current_schema_id"), + "table_location": latest.get("location"), + "fields": latest.get("fields"), + } + + +def _gravitino_readback( + rehearsal: m321.ObjectStoreProjectionRehearsal, + profile: m324.m322.RealFeatureIngestionProfile, +) -> dict[str, Any]: + status, payload = rehearsal.admin.request( + "GET", + rehearsal._table_path(profile.target), + label="M3-25 retained Gravitino table readback", + ) + projection = m321.durable._table_projection(_mapping(payload)) + return { + "read_status": status, + "table_projection_sha256": canonical_json_fingerprint(projection), + "table_projection": projection, + } + + +def _authority_counts(engine: Any) -> dict[str, int]: + with engine.connect() as connection: + row = ( + connection.execute( + text( + """ + SELECT + (SELECT count(*) FROM gda_control.resource + WHERE tenant_id = :tenant_id) AS resources, + (SELECT count(*) FROM gda_control.resource_version + WHERE tenant_id = :tenant_id) AS resource_versions, + (SELECT count(*) FROM gda_control.platform_definition_version + WHERE tenant_id = :tenant_id) AS definition_versions, + (SELECT count(*) FROM gda_control.platform_run + WHERE tenant_id = :tenant_id) AS platform_runs + """ + ), + {"tenant_id": m324.TENANT}, + ) + .mappings() + .one() + ) + return {key: int(value) for key, value in row.items()} + + +def _observe_control_ledger( + engine: Any, + gateway: PlatformGateway, + source: Mapping[str, Any], + promotion: m324.m323.RunOutputLedgerPromotion, +) -> tuple[dict[str, Any], Any, Any]: + authorization = _mapping(source.get("authorization")) + artifact_ids = ( + UUID(str(authorization["execution_plan_artifact_id"])), + UUID(str(authorization["policy_decision_artifact_id"])), + UUID(str(authorization["approval_artifact_id"])), + promotion.output_artifact.artifact_id, + promotion.quality_evidence_artifact.artifact_id, + ) + run = gateway.get_run(m324.TENANT, m324.RUN_ID) + with gateway._transaction(m324.TENANT) as connection: + observation = gateway._load_observation(connection, m324.TENANT, PROVIDER_OBSERVATION_ID) + output_version = gateway._load_resource_version( + connection, m324.TENANT, m324.OUTPUT_RESOURCE_VERSION_ID + ) + artifacts = [ + gateway._load_artifact(connection, m324.TENANT, artifact_id) + for artifact_id in artifact_ids + ] + quality = gateway._load_quality_result( + connection, + m324.TENANT, + promotion.quality_result.quality_result_id, + ) + lineage = gateway._load_lineage( + connection, + m324.TENANT, + promotion.lineage_event.lineage_event_id, + ) + facts = [run, observation, output_version, *artifacts, quality, lineage] + if any(value is None for value in facts): + raise RetainedRealFeatureRestartRecoveryError("retained control ledger is incomplete") + stable_facts = [value.model_dump(mode="json", by_alias=True) for value in facts] + return ( + { + "ledger_counts": m324._ledger_counts(engine), + "authority_counts": _authority_counts(engine), + "facts_sha256": canonical_json_fingerprint(stable_facts), + "platform_run_status": run.status.value, + "platform_run_state_version": run.state_version, + "provider_observation_id": str(observation.observation_id), + "provider_observation_sha256": observation.observation_sha256, + }, + run, + observation, + ) + + +def _material_errors( + before: Mapping[str, Any], + after: Mapping[str, Any], + retention: m324.RetainedMaterialObservation, +) -> list[str]: + errors: list[str] = [] + if dict(before) != dict(after): + errors.append("retained Iceberg material changed across restart") + expected = { + "object_inventory_sha256": retention.object_inventory_sha256, + "data_file_count": retention.data_file_count, + "metadata_body_sha256": retention.metadata_body_sha256, + "snapshot_id": retention.snapshot_id, + "table_location": retention.storage_uri, + } + if any(before.get(key) != value for key, value in expected.items()): + errors.append("retained Iceberg material no longer binds M3-24") + return errors + + +def build_evidence(observation: Mapping[str, Any]) -> dict[str, Any]: + errors: list[str] = [] + contract = build_contract_report() + if observation.get("schema") != OBSERVATION_SCHEMA: + errors.append("M3-25 observation schema does not match") + if observation.get("contract_sha256") != contract.get("contract_sha256"): + errors.append("M3-25 observation contract binding is stale") + if observation.get("source_evidence_sha256") != SOURCE_EVIDENCE_SHA256: + errors.append("M3-25 predecessor evidence binding drifted") + restart = _mapping(observation.get("kubernetes_restart")) + errors.extend( + _runtime_continuity_errors( + _mapping(restart.get("before")), + _mapping(restart.get("after")), + _mapping(observation.get("m324_initial_runtime")), + ) + ) + errors.extend(_control_continuity_errors(_mapping(observation.get("control_restart")))) + retention = m324.RetainedMaterialObservation.model_validate( + observation.get("retention_observation") + ) + material = _mapping(observation.get("material")) + errors.extend( + _material_errors( + _mapping(material.get("before")), + _mapping(material.get("after")), + retention, + ) + ) + independent = _mapping(observation.get("independent_quality")) + if independent.get("before") != independent.get("after") or independent.get( + "after" + ) != observation.get("m324_independent_quality"): + errors.append("independent Parquet quality changed across restart") + gravitino = _mapping(observation.get("gravitino")) + if ( + gravitino.get("before") != gravitino.get("after") + or _mapping(gravitino.get("after")).get("read_status") != 200 + ): + errors.append("Gravitino table readback changed across restart") + ledger = _mapping(observation.get("control_ledger")) + if not ( + ledger.get("before") == ledger.get("after_restart") == ledger.get("after_terminal_replay") + ): + errors.append("GDA Control ledger changed across restart or replay") + before_ledger = _mapping(ledger.get("before")) + if ( + before_ledger.get("ledger_counts") != EXPECTED_LEDGER_COUNTS + or before_ledger.get("platform_run_status") != "succeeded" + or before_ledger.get("platform_run_state_version") != 3 + or before_ledger.get("provider_observation_id") != str(PROVIDER_OBSERVATION_ID) + ): + errors.append("GDA Control terminal authority no longer matches M3-24") + replay = _mapping(observation.get("terminal_replay")) + if ( + replay.get("promotion_created") is not False + or replay.get("platform_run_status") != "succeeded" + or replay.get("platform_run_state_version") != 3 + ): + errors.append("post-restart terminal replay was not an exact no-op") + for claim in ( + "source_payload_absent_before", + "source_payload_absent_after", + "credential_material_recorded", + "runtime_port_forwards_stopped", + ): + expected = False if claim == "credential_material_recorded" else True + if observation.get(claim) is not expected: + errors.append(f"M3-25 observation boundary failed: {claim}") + stable = { + "schema": EVIDENCE_SCHEMA, + "status": ( + "local_retained_real_feature_restart_recovery_verified" if not errors else "blocked" + ), + "contract_sha256": contract["contract_sha256"], + "source_evidence_sha256": SOURCE_EVIDENCE_SHA256, + "retention_id": retention.retention_id, + "retention_expires_at": retention.expires_at.isoformat().replace("+00:00", "Z"), + "tenant_id": m324.TENANT, + "run_id": str(m324.RUN_ID), + "output_resource_version_id": str(m324.OUTPUT_RESOURCE_VERSION_ID), + "output_content_sha256": retention.output_content_sha256, + "kubernetes_restart": restart, + "control_restart": observation.get("control_restart"), + "material": material, + "gravitino": gravitino, + "independent_quality": independent, + "control_ledger": ledger, + "terminal_replay": replay, + "source_payload_absent_before": observation.get("source_payload_absent_before"), + "source_payload_absent_after": observation.get("source_payload_absent_after"), + "credential_material_recorded": False, + "runtime_port_forwards_stopped": observation.get("runtime_port_forwards_stopped"), + "same_retention_identity_verified": not errors, + "ordered_stateful_restart_verified": not errors, + "kubernetes_runtime_restart_verified": not errors, + "control_database_restart_verified": not errors, + "iceberg_material_continuity_verified": not errors, + "gravitino_catalog_continuity_verified": not errors, + "independent_quality_continuity_verified": not errors, + "control_ledger_continuity_verified": not errors, + "exact_terminal_replay_after_restart_verified": not errors, + "local_retained_real_feature_restart_recovery_verified": not errors, + "writes_to_legacy": False, + **{claim: False for claim in FALSE_CLAIMS}, + "errors": errors, + } + return {**stable, "evidence_sha256": canonical_json_fingerprint(stable)} + + +def validate_evidence(evidence: Mapping[str, Any]) -> list[str]: + errors: list[str] = [] + stable = {key: value for key, value in evidence.items() if key != "evidence_sha256"} + if evidence.get("schema") != EVIDENCE_SCHEMA: + errors.append("M3-25 evidence schema does not match") + if evidence.get("evidence_sha256") != canonical_json_fingerprint(stable): + errors.append("M3-25 evidence fingerprint does not match") + contract = build_contract_report() + if evidence.get("contract_sha256") != contract.get("contract_sha256"): + errors.append("M3-25 contract binding is stale") + if evidence.get("source_evidence_sha256") != SOURCE_EVIDENCE_SHA256: + errors.append("M3-25 source evidence binding drifted") + for claim in ( + "same_retention_identity_verified", + "ordered_stateful_restart_verified", + "kubernetes_runtime_restart_verified", + "control_database_restart_verified", + "iceberg_material_continuity_verified", + "gravitino_catalog_continuity_verified", + "independent_quality_continuity_verified", + "control_ledger_continuity_verified", + "exact_terminal_replay_after_restart_verified", + "local_retained_real_feature_restart_recovery_verified", + "source_payload_absent_before", + "source_payload_absent_after", + "runtime_port_forwards_stopped", + ): + if evidence.get(claim) is not True: + errors.append(f"M3-25 evidence claim is false: {claim}") + if evidence.get("credential_material_recorded") is not False: + errors.append("M3-25 evidence records credential material") + for claim in FALSE_CLAIMS: + if evidence.get(claim) is not False: + errors.append(f"M3-25 evidence may not claim {claim}") + serialized = json.dumps(evidence, ensure_ascii=True, sort_keys=True).lower() + for forbidden in ( + "/users/", + "/home/", + "downloads/", + ".tmp/", + "geometry_wkb_hex", + '"rows"', + "postgres_password", + "password=", + '"password"', + '"secret"', + '"token"', + '"access_key"', + '"access-key"', + ): + if forbidden in serialized: + errors.append("M3-25 evidence contains local, source, or credential material") + break + return errors + + +def build_validation_report(*, evidence_path: Path = DEFAULT_EVIDENCE_PATH) -> dict[str, Any]: + contract = build_contract_report() + errors = list(contract["errors"]) + evidence: dict[str, Any] | None = None + try: + evidence = _load_json_object(evidence_path) + errors.extend(validate_evidence(evidence)) + except (OSError, TypeError, ValueError, RetainedRealFeatureRestartRecoveryError): + errors.append("M3-25 checked evidence is unavailable") + return { + "schema": VALIDATION_SCHEMA, + "status": "valid" if not errors else "invalid", + "contract_sha256": contract["contract_sha256"], + "evidence_sha256": evidence.get("evidence_sha256") if evidence else None, + "errors": errors, + } + + +def _write_json(path: Path, payload: Mapping[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + +def run_live_rehearsal() -> dict[str, Any]: + contract = build_contract_report() + if contract.get("status") != "valid": + raise RetainedRealFeatureRestartRecoveryError("M3-25 static contract is invalid") + source = _checked_source_evidence() + retention = m324.RetainedMaterialObservation.model_validate(source.get("retention_observation")) + if datetime.now(UTC) >= retention.expires_at: + raise RetainedRealFeatureRestartRecoveryError("M3-24 retained material has expired") + runtime, profile = _attach_runtime(source) + admin_material, database_material, object_store_user, object_store_material = ( + _read_runtime_materials(runtime) + ) + if not database_material.get_secret_value(): + raise RetainedRealFeatureRestartRecoveryError("retained catalog database material is empty") + control = RetainedControlAttachment(source, retention) + checked_ingestion = _load_json_object(m324.DEFAULT_SOURCE_EVIDENCE_PATH) + plan = m324.m322.RealFeatureIngestionPlan.model_validate( + _mapping(checked_ingestion.get("observation")).get("plan") + ) + promotion = m324.build_terminal_promotion(checked_ingestion, retention) + before_runtime = _observe_runtime(runtime) + if _mapping(before_runtime.get("namespace")).get("uid") != retention.namespace_uid: + raise RetainedRealFeatureRestartRecoveryError( + "retained namespace identity no longer matches M3-24" + ) + source_absent_before = _source_payload_absent(runtime) + before_control = control.observe() + control.wait_ready() + + object_forward: Any = None + gravitino_forward: Any = None + rehearsal: m321.ObjectStoreProjectionRehearsal | None = None + engine: Any = None + all_forwards_stopped = True + try: + object_forward = _start_forward( + runtime, + service=runtime.profile.runtime.object_store_service, + target_port=runtime.profile.runtime.object_store_service_port, + ) + gravitino_forward = _start_forward( + runtime, + service=runtime.profile.runtime.service, + target_port=runtime.profile.runtime.gravitino_service_port, + ) + endpoint_url = f"http://127.0.0.1:{object_forward.local_port}" + rehearsal = m321.ObjectStoreProjectionRehearsal( + base_url=f"http://127.0.0.1:{gravitino_forward.local_port}/api", + admin_name=profile.identity.service_admin, + admin_material=admin_material, + ) + before_store = m324.m322.observe_ingested_table( + runtime, + profile, + endpoint_url=endpoint_url, + object_store_user=object_store_user, + object_store_material=object_store_material, + ) + before_material = _material_projection(before_store) + before_quality = m324.independently_evaluate_retained_parquet( + runtime, + profile, + plan, + {"projection": source["source_projection"]}, + before_store, + endpoint_url=endpoint_url, + object_store_user=object_store_user, + object_store_material=object_store_material, + ) + before_gravitino = _gravitino_readback(rehearsal, profile) + engine = create_engine(control.database_url, pool_pre_ping=True) + gateway = PlatformGateway(engine) + before_ledger, before_run, _ = _observe_control_ledger(engine, gateway, source, promotion) + if before_run.status != RunStatus.SUCCEEDED or before_run.state_version != 3: + raise RetainedRealFeatureRestartRecoveryError("retained PlatformRun is not succeeded@3") + engine.dispose() + engine = None + rehearsal.close() + rehearsal = None + all_forwards_stopped = bool(gravitino_forward.stop()) and all_forwards_stopped + gravitino_forward = None + all_forwards_stopped = bool(object_forward.stop()) and all_forwards_stopped + object_forward = None + + kubernetes_restart = _restart_kubernetes_runtime(runtime, before_runtime) + control_restart = control.restart() + if control_restart.get("before") != before_control: + raise RetainedRealFeatureRestartRecoveryError( + "control database changed before the scheduled restart" + ) + after_runtime = _mapping(kubernetes_restart.get("after")) + source_absent_after = _source_payload_absent(runtime) + + object_forward = _start_forward( + runtime, + service=runtime.profile.runtime.object_store_service, + target_port=runtime.profile.runtime.object_store_service_port, + ) + gravitino_forward = _start_forward( + runtime, + service=runtime.profile.runtime.service, + target_port=runtime.profile.runtime.gravitino_service_port, + ) + endpoint_url = f"http://127.0.0.1:{object_forward.local_port}" + rehearsal = m321.ObjectStoreProjectionRehearsal( + base_url=f"http://127.0.0.1:{gravitino_forward.local_port}/api", + admin_name=profile.identity.service_admin, + admin_material=admin_material, + ) + after_store = m324.m322.observe_ingested_table( + runtime, + profile, + endpoint_url=endpoint_url, + object_store_user=object_store_user, + object_store_material=object_store_material, + ) + after_material = _material_projection(after_store) + after_quality = m324.independently_evaluate_retained_parquet( + runtime, + profile, + plan, + {"projection": source["source_projection"]}, + after_store, + endpoint_url=endpoint_url, + object_store_user=object_store_user, + object_store_material=object_store_material, + ) + after_gravitino = _gravitino_readback(rehearsal, profile) + engine = create_engine(control.database_url, pool_pre_ping=True) + gateway = PlatformGateway(engine) + after_restart_ledger, _, observation = _observe_control_ledger( + engine, gateway, source, promotion + ) + + def live_probe(observed: m324.RetainedMaterialObservation) -> bool: + return m324._live_material_probe( + observed, + runtime=runtime, + profile=profile, + control=control, + endpoint_url=endpoint_url, + object_store_user=object_store_user, + object_store_material=object_store_material, + ) + + coordinator = m324.RetainedTerminalSuccessCoordinator(gateway, material_probe=live_probe) + replay_promotion, replayed_run = coordinator.finalize( + promotion, + retention, + observation, + ) + after_replay_ledger, _, _ = _observe_control_ledger(engine, gateway, source, promotion) + raw_observation = { + "schema": OBSERVATION_SCHEMA, + "observed_at": datetime.now(UTC).isoformat().replace("+00:00", "Z"), + "contract_sha256": contract["contract_sha256"], + "source_evidence_sha256": SOURCE_EVIDENCE_SHA256, + "retention_observation": retention.model_dump(mode="json", by_alias=True), + "m324_initial_runtime": source["initial_runtime"], + "m324_independent_quality": source["independent_quality"], + "kubernetes_restart": { + "order": kubernetes_restart["order"], + "before": before_runtime, + "after": after_runtime, + }, + "control_restart": control_restart, + "material": {"before": before_material, "after": after_material}, + "gravitino": { + "before": before_gravitino, + "after": after_gravitino, + }, + "independent_quality": { + "before": before_quality, + "after": after_quality, + }, + "control_ledger": { + "before": before_ledger, + "after_restart": after_restart_ledger, + "after_terminal_replay": after_replay_ledger, + }, + "terminal_replay": { + "promotion_created": replay_promotion.created, + "platform_run_status": replayed_run.status.value, + "platform_run_state_version": replayed_run.state_version, + }, + "source_payload_absent_before": source_absent_before, + "source_payload_absent_after": source_absent_after, + "credential_material_recorded": False, + "runtime_port_forwards_stopped": False, + } + finally: + if engine is not None: + engine.dispose() + if rehearsal is not None: + rehearsal.close() + if gravitino_forward is not None: + all_forwards_stopped = bool(gravitino_forward.stop()) and all_forwards_stopped + if object_forward is not None: + all_forwards_stopped = bool(object_forward.stop()) and all_forwards_stopped + raw_observation["runtime_port_forwards_stopped"] = all_forwards_stopped + evidence = build_evidence(raw_observation) + errors = validate_evidence(evidence) + if errors: + raise RetainedRealFeatureRestartRecoveryError( + "M3-25 live evidence failed self-validation: " + "; ".join(errors) + ) + return evidence + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers(dest="command", required=True) + subparsers.add_parser("contract") + validate = subparsers.add_parser("validate") + validate.add_argument("--evidence", type=Path, default=DEFAULT_EVIDENCE_PATH) + live = subparsers.add_parser("live-rehearsal") + live.add_argument("--output", type=Path, default=DEFAULT_EVIDENCE_PATH) + args = parser.parse_args(argv) + if args.command == "contract": + report = build_contract_report() + elif args.command == "validate": + report = build_validation_report(evidence_path=args.evidence) + else: + report = run_live_rehearsal() + _write_json(args.output, report) + print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True)) + return ( + 0 + if report["status"] + in { + "valid", + "local_retained_real_feature_restart_recovery_verified", + } + else 1 + ) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/data_agent/platform_truth.py b/data_agent/platform_truth.py index 2cc729b8..d05179f6 100644 --- a/data_agent/platform_truth.py +++ b/data_agent/platform_truth.py @@ -1064,6 +1064,32 @@ def _config( "scheduler/executor and restart/recovery" ), ), + RuntimeSpec( + "metadata_retained_real_feature_restart_recovery_rehearsal", + "retained_real_feature_restart_recovery_rehearsal", + "governed", + "retained_evidence_durable", + ( + "identity-bound restart of retained namespace/PVC/Iceberg material " + "and dedicated GDA Control PostgreSQL + committed local evidence" + ), + "metadata-platform", + "local_verification_only", + ( + "data_agent/metadata_fabric_retained_real_feature_restart_recovery.py", + "scripts/metadata-fabric-retained-real-feature-restart-recovery.sh", + ), + ( + ( + "data_agent/metadata_fabric_retained_real_feature_restart_recovery.py", + "def run_live_rehearsal", + ), + ), + ( + "Protected production identities/storage/tenant binding, persistent " + "scheduler/executor, backup/PITR and independent failure domains" + ), + ), RuntimeSpec( "datalake_monitor", "monitor_loop", diff --git a/data_agent/test_metadata_fabric_retained_real_feature_restart_recovery.py b/data_agent/test_metadata_fabric_retained_real_feature_restart_recovery.py new file mode 100644 index 00000000..4b1ed888 --- /dev/null +++ b/data_agent/test_metadata_fabric_retained_real_feature_restart_recovery.py @@ -0,0 +1,241 @@ +import base64 +import json +from copy import deepcopy + +import pytest + +from data_agent import metadata_fabric_retained_real_feature_restart_recovery as recovery + + +def _source() -> dict: + return json.loads(recovery.DEFAULT_SOURCE_EVIDENCE_PATH.read_text(encoding="utf-8")) + + +def _runtime_pair() -> tuple[dict, dict]: + before = deepcopy(_source()["initial_runtime"]) + before["postgresql_service"] = { + "name": "gravitino-persistence-postgresql", + "uid": "00000000-0000-4000-8000-000000000251", + "type": "ClusterIP", + "ports": [{"name": "postgresql", "port": 5432}], + } + after = deepcopy(before) + after["postgresql"]["pod_uid"] = "00000000-0000-4000-8000-000000000252" + after["object_store"]["pod_uid"] = "00000000-0000-4000-8000-000000000253" + after["gravitino"]["pod_uid"] = "00000000-0000-4000-8000-000000000254" + return before, after + + +def _observation() -> dict: + source = _source() + retention = source["retention_observation"] + before_runtime, after_runtime = _runtime_pair() + material = { + "object_count": 5, + "object_inventory_sha256": retention["object_inventory_sha256"], + "data_file_count": retention["data_file_count"], + "metadata_file_count": 1, + "manifest_file_count": 3, + "metadata_body_sha256": retention["metadata_body_sha256"], + "snapshot_id": retention["snapshot_id"], + "schema_id": 0, + "table_location": retention["storage_uri"], + "fields": [], + } + ledger = { + "ledger_counts": recovery.EXPECTED_LEDGER_COUNTS, + "authority_counts": { + "resources": 2, + "resource_versions": 2, + "definition_versions": 1, + "platform_runs": 1, + }, + "facts_sha256": "1" * 64, + "platform_run_status": "succeeded", + "platform_run_state_version": 3, + "provider_observation_id": str(recovery.PROVIDER_OBSERVATION_ID), + "provider_observation_sha256": source["provider_observation"]["observation_sha256"], + } + control_before = { + "database_ref": source["control_database"]["database_ref"], + "container_name": source["control_database"]["container_name"], + "container_id": "2" * 64, + "container_running": True, + "container_status": "running", + "process_id": 101, + "started_at": "2026-07-31T04:15:30Z", + "volume_name": source["control_database"]["volume_name"], + "volume_retained": True, + "host_port": source["control_database"]["host_port"], + "retention_id": source["retention_id"], + "owner": "team:metadata-platform", + "expires_at": retention["expires_at"], + "credential_material_recorded": False, + } + control_after = { + **control_before, + "process_id": 202, + "started_at": "2026-07-31T05:00:00Z", + } + gravitino = { + "read_status": 200, + "table_projection_sha256": "3" * 64, + "table_projection": {"name": "cultural_districts"}, + } + return { + "schema": recovery.OBSERVATION_SCHEMA, + "observed_at": "2026-07-31T05:01:00Z", + "contract_sha256": recovery.build_contract_report()["contract_sha256"], + "source_evidence_sha256": recovery.SOURCE_EVIDENCE_SHA256, + "retention_observation": retention, + "m324_initial_runtime": source["initial_runtime"], + "m324_independent_quality": source["independent_quality"], + "kubernetes_restart": { + "order": [ + "statefulset/gravitino-persistence-postgresql", + "statefulset/metadata-object-store", + "statefulset/gravitino-persistence", + ], + "before": before_runtime, + "after": after_runtime, + }, + "control_restart": {"before": control_before, "after": control_after}, + "material": {"before": material, "after": deepcopy(material)}, + "gravitino": {"before": gravitino, "after": deepcopy(gravitino)}, + "independent_quality": { + "before": source["independent_quality"], + "after": deepcopy(source["independent_quality"]), + }, + "control_ledger": { + "before": ledger, + "after_restart": deepcopy(ledger), + "after_terminal_replay": deepcopy(ledger), + }, + "terminal_replay": { + "promotion_created": False, + "platform_run_status": "succeeded", + "platform_run_state_version": 3, + }, + "source_payload_absent_before": True, + "source_payload_absent_after": True, + "credential_material_recorded": False, + "runtime_port_forwards_stopped": True, + } + + +def test_contract_binds_intact_m324_without_production_overclaim(): + contract = recovery.build_contract_report() + + assert contract["status"] == "valid" + assert contract["errors"] == [] + assert contract["source_evidence_sha256"] == recovery.SOURCE_EVIDENCE_SHA256 + assert contract["requires_kubernetes_pod_rotation"] is True + assert contract["requires_control_process_rotation"] is True + assert contract["production_restart_recovery_verified"] is False + assert contract["production_ready"] is False + + +def test_checked_restart_recovery_evidence_is_self_validating(): + evidence = json.loads(recovery.DEFAULT_EVIDENCE_PATH.read_text(encoding="utf-8")) + validation = recovery.build_validation_report() + + assert recovery.validate_evidence(evidence) == [] + assert validation["status"] == "valid" + assert validation["errors"] == [] + assert evidence["contract_sha256"] == validation["contract_sha256"] + assert evidence["evidence_sha256"] == validation["evidence_sha256"] + assert evidence["new_ingestion_executed"] is False + assert evidence["new_authority_facts_created"] is False + + +def test_runtime_continuity_requires_all_pods_to_rotate_and_stable_identity(): + source = _source() + before, after = _runtime_pair() + + assert recovery._runtime_continuity_errors(before, after, source["initial_runtime"]) == [] + + after["object_store"]["pod_uid"] = before["object_store"]["pod_uid"] + after["postgresql"]["pvc"]["uid"] = "00000000-0000-4000-8000-000000000255" + errors = recovery._runtime_continuity_errors(before, after, source["initial_runtime"]) + + assert "object_store pod did not rotate" in errors + assert "retained Kubernetes stable runtime identity changed" in errors + assert "postgresql PVC identity changed" in errors + + +def test_control_continuity_requires_same_container_volume_and_new_process(): + restart = _observation()["control_restart"] + + assert recovery._control_continuity_errors(restart) == [] + + restart["after"]["container_id"] = "4" * 64 + restart["after"]["process_id"] = restart["before"]["process_id"] + restart["after"]["started_at"] = restart["before"]["started_at"] + errors = recovery._control_continuity_errors(restart) + + assert "retained control identity changed: container_id" in errors + assert "retained control PostgreSQL process did not rotate" in errors + assert "retained control PostgreSQL start time did not rotate" in errors + + +def test_runtime_material_decoding_and_control_password_are_memory_only(): + value = recovery._decode_runtime_material( + {"data": {"value": base64.b64encode(b"unit-material").decode()}}, + "value", + label="unit", + ) + control = recovery._extract_control_password( + ["PG_MAJOR=16", "POSTGRES_PASSWORD=control-material"] + ) + + assert value.get_secret_value() == "unit-material" + assert control.get_secret_value() == "control-material" + assert "unit-material" not in repr(value) + with pytest.raises( + recovery.RetainedRealFeatureRestartRecoveryError, + match="credential is unavailable", + ): + recovery._extract_control_password(["PG_MAJOR=16"]) + + +def test_evidence_verifies_and_rejects_drift_overclaim_and_sensitive_fields(): + evidence = recovery.build_evidence(_observation()) + + assert evidence["status"] == ("local_retained_real_feature_restart_recovery_verified") + assert evidence["errors"] == [] + assert recovery.validate_evidence(evidence) == [] + + drifted = deepcopy(evidence) + drifted["material"]["after"]["snapshot_id"] += 1 + assert "M3-25 evidence fingerprint does not match" in recovery.validate_evidence(drifted) + + overclaimed = deepcopy(evidence) + overclaimed["production_ready"] = True + overclaimed["evidence_sha256"] = recovery.canonical_json_fingerprint( + {key: value for key, value in overclaimed.items() if key != "evidence_sha256"} + ) + assert "M3-25 evidence may not claim production_ready" in ( + recovery.validate_evidence(overclaimed) + ) + + sensitive = deepcopy(evidence) + sensitive["diagnostic"] = {"password": "must-not-appear"} + sensitive["evidence_sha256"] = recovery.canonical_json_fingerprint( + {key: value for key, value in sensitive.items() if key != "evidence_sha256"} + ) + assert "M3-25 evidence contains local, source, or credential material" in ( + recovery.validate_evidence(sensitive) + ) + + +def test_build_evidence_fails_closed_on_material_and_ledger_drift(): + observation = _observation() + observation["material"]["after"]["snapshot_id"] += 1 + observation["control_ledger"]["after_restart"]["facts_sha256"] = "0" * 64 + + evidence = recovery.build_evidence(observation) + + assert evidence["status"] == "blocked" + assert "retained Iceberg material changed across restart" in evidence["errors"] + assert "GDA Control ledger changed across restart or replay" in evidence["errors"] + assert evidence["production_ready"] is False diff --git a/data_agent/test_platform_truth.py b/data_agent/test_platform_truth.py index 690b5736..5f823798 100644 --- a/data_agent/test_platform_truth.py +++ b/data_agent/test_platform_truth.py @@ -310,6 +310,13 @@ def test_repository_source_access_and_runtime_baselines_match(): and item["production_role"] == "local_verification_only" for item in static_report["runtime"]["inventory"] ) + assert any( + item["runtime_id"] + == "metadata_retained_real_feature_restart_recovery_rehearsal" + and item["durability"] == "retained_evidence_durable" + and item["production_role"] == "local_verification_only" + for item in static_report["runtime"]["inventory"] + ) def test_runtime_report_detects_unregistered_background_mechanism(tmp_path): diff --git a/docs/architecture-decisions/adr-071-retained-real-feature-restart-recovery.md b/docs/architecture-decisions/adr-071-retained-real-feature-restart-recovery.md new file mode 100644 index 00000000..c44aadd4 --- /dev/null +++ b/docs/architecture-decisions/adr-071-retained-real-feature-restart-recovery.md @@ -0,0 +1,81 @@ +# ADR-071: Retained real-feature restart recovery + +**Status**: Accepted + +**Date**: 2026-07-31 + +**Decision owners**: Data Platform, Metadata Platform, DataOps, Security, Platform Architecture + +**Related decisions**: [ADR-054](adr-054-local-gravitino-jdbc-catalog-restart-continuity.md) | [ADR-067](adr-067-object-store-runtime-bound-active-metadata-promotion.md) | [ADR-070](adr-070-retained-real-feature-terminal-success.md) + +## Context + +M3-24 proved one complete real-data authority chain from authorized DolphinScheduler execution through Spark/Sedona JDBC/S3 Iceberg ingestion, independent spatial quality evaluation, atomic GDA Control promotion and database-authoritative `succeeded@3`. It retained the namespace, PVCs, Iceberg objects and dedicated GDA Control PostgreSQL database for seven days, but explicitly did not prove that those facts survive process restart. + +The retained window creates a bounded opportunity to test recovery against the exact successful authority rather than building a new synthetic runtime. The rehearsal must not re-ingest the Chongqing features, create a new ResourceVersion or Run, repair provider state, change the successful verdict, expose retained credentials, or rewrite M3-24 evidence. + +## Considered options + +### 1. Wait for the production environment + +This avoids another local rehearsal, but leaves the current retained authority untested and wastes its bounded audit window. Production identity and storage attestation are also not yet available. + +### 2. Recreate the runtime and ingest again + +This would test rebuild behavior, not continuity of the successful M3-24 authority. It would create new material and ledger facts, making it impossible to prove exact recovery of the retained state. + +### 3. Restart the retained runtime in place + +This directly tests the current gap. Stable infrastructure identity, rotating process identity, byte-stable data and exact ledger replay can all be checked without creating new authority facts. M3-25 adopts this option. + +## Decision + +### 1. Bind recovery to checked M3-24 evidence + +M3-25 accepts only the checked M3-24 evidence SHA `d966668b5a2ea57c7a4b2a3bc9824daab9b0128d9f94e515d7be649b145de418` and retention ID `m3-24-229740ac50ebb53b`. The namespace UID, expiry, control container and volume, output ResourceVersion, content SHA, snapshot and object inventory must match before any restart. + +Runtime credentials are read only from the retained Kubernetes runtime objects and Docker container environment into `SecretStr`. They are never written to evidence, logs or exceptions. + +### 2. Restart stateful dependencies in order + +The controlled order is PostgreSQL, MinIO object storage, Gravitino, and then the dedicated GDA Control PostgreSQL process. Each Kubernetes rollout must return to one ready replica before the next begins. + +Namespace, StatefulSet, Service, PVC, volume, image and control container identity must remain stable. All three Kubernetes Pod UIDs must change. The GDA Control container ID and volume must remain stable while PostgreSQL PID and `StartedAt` change. + +### 3. Require independent data and catalog continuity + +Before and after restart, direct S3 readback must return the same object inventory SHA, Iceberg metadata body SHA, snapshot ID, schema and single Parquet file. The independent evaluator reopens that Parquet and requires all nine M3-24 spatial quality counts to remain 20, with unchanged row-set and data-body SHA. + +Gravitino admin readback must return the same eight-column table projection, output ResourceVersion, content SHA and provider revision. No repair, table recreation or ingestion is allowed. + +### 4. Keep the successful ledger byte-stable + +The GDA Control authority counts, ledger counts, provider observation, `succeeded@3` state and combined facts SHA must be identical before restart, after restart and after terminal replay. The replay must use the original terminal verdict and return `promotion_created=false`; any additional Artifact, observation, QualityResult, lineage or Run event fails the gate. + +### 5. Cap the claim at local process restart continuity + +The namespace, Kubernetes nodes, PVCs, MinIO and control database remain on one Docker Desktop host. The rehearsal does not restore from backup, exercise PITR, lose a host or storage volume, use protected production identity, or run a persistent scheduler/executor. + +Therefore backup/PITR, independent failure domains, production restart recovery, production ingestion and `production_ready` remain false. + +## Verification + +The M3-25 rehearsal recorded: + +- contract SHA `83ed15ae4eed85e0c261c2b3a04ea2ad559f3deb7b86c7c2f2dedd0cf28d23d0` and evidence SHA `1b5a5ceeadee88868bab6237b3f3280c8b13793cc54193592fec7dbbfdd4e8a6`; +- unchanged namespace, three StatefulSet identities, three Service identities, PostgreSQL/MinIO PVC identities, images, control container ID and control volume; +- PostgreSQL, MinIO and Gravitino Pod UID rotation, plus GDA Control PostgreSQL PID rotation from `2087977` to `2093386`; +- unchanged object inventory SHA `de4a0efed9fdb68f0019b843377f6c8de71664de955130d0dd38e99eccdb8034`, snapshot `8034081021802585202`, 94,603-byte Parquet body SHA `6cc0fc9eaf48f8106f9afe192704c44407c86c9ea119ae20894bf369a8e74779` and row-set SHA `c26ff708f4b6be082327dff63a6a8659420dbc4cab37dea1cac7b40f147512df`; +- nine independent spatial quality counts equal to 20 and unchanged Gravitino projection SHA `f30feb94a5a8280597f331a7f965762bfabb9b82397af49dda05b61ce00bbb1e`; +- unchanged GDA Control facts SHA `5c0b8a58729e551c250b0410bcebdfe3f019f215a50b20d503f094fe1562d8b5`, five Artifacts, two attempt observations, one QualityResult, one LineageEvent, four Run events and `succeeded@3` before restart, after restart and after exact replay; +- absent source payload, stopped port forwards, zero credential material in evidence, no new ingestion and no new authority facts. + +## Consequences + +**Positive**: the retained real Chongqing output, technical catalog and control authority now have one content-bound, failure-closed process restart continuity proof. Recovery is verified against the exact terminal success instead of inferred from provider health. + +**Negative**: the proof is time-bound to the M3-24 retention window and exercises process restart only. It consumes local runtime capacity and cannot support a production durability claim. + +**Next gate**: provide protected production identity, storage and tenant attestation; deploy persistent scheduler/executor and control services; then exercise backup/PITR, host or availability-zone failure, staging scale, Spark/Flink conformance, alerting and runbook recovery. + +**Revisit trigger**: supersede this decision when an approved staging environment can reproduce the same authority chain across independent failure domains with immutable retention, backup/PITR and audited lifecycle automation. diff --git a/docs/evidence/metadata-fabric-retained-real-feature-restart-recovery-2026-07-31.json b/docs/evidence/metadata-fabric-retained-real-feature-restart-recovery-2026-07-31.json new file mode 100644 index 00000000..dc6b3eb3 --- /dev/null +++ b/docs/evidence/metadata-fabric-retained-real-feature-restart-recovery-2026-07-31.json @@ -0,0 +1,652 @@ +{ + "backup_restore_verified": false, + "contract_sha256": "83ed15ae4eed85e0c261c2b3a04ea2ad559f3deb7b86c7c2f2dedd0cf28d23d0", + "control_database_restart_verified": true, + "control_ledger": { + "after_restart": { + "authority_counts": { + "definition_versions": 1, + "platform_runs": 1, + "resource_versions": 3, + "resources": 3 + }, + "facts_sha256": "5c0b8a58729e551c250b0410bcebdfe3f019f215a50b20d503f094fe1562d8b5", + "ledger_counts": { + "approvals": 1, + "artifacts": 5, + "attempts": 2, + "evaluator_evidence": 1, + "execution_plans": 1, + "lineage_events": 1, + "policy_decisions": 1, + "quality_results": 1, + "run_events": 4 + }, + "platform_run_state_version": 3, + "platform_run_status": "succeeded", + "provider_observation_id": "85bead83-1901-5649-b6e4-fe46d01c9ea9", + "provider_observation_sha256": "eed782687756dcddbf7f0966f70902edd037f0fc1e2acf3bb72a860e780d0819" + }, + "after_terminal_replay": { + "authority_counts": { + "definition_versions": 1, + "platform_runs": 1, + "resource_versions": 3, + "resources": 3 + }, + "facts_sha256": "5c0b8a58729e551c250b0410bcebdfe3f019f215a50b20d503f094fe1562d8b5", + "ledger_counts": { + "approvals": 1, + "artifacts": 5, + "attempts": 2, + "evaluator_evidence": 1, + "execution_plans": 1, + "lineage_events": 1, + "policy_decisions": 1, + "quality_results": 1, + "run_events": 4 + }, + "platform_run_state_version": 3, + "platform_run_status": "succeeded", + "provider_observation_id": "85bead83-1901-5649-b6e4-fe46d01c9ea9", + "provider_observation_sha256": "eed782687756dcddbf7f0966f70902edd037f0fc1e2acf3bb72a860e780d0819" + }, + "before": { + "authority_counts": { + "definition_versions": 1, + "platform_runs": 1, + "resource_versions": 3, + "resources": 3 + }, + "facts_sha256": "5c0b8a58729e551c250b0410bcebdfe3f019f215a50b20d503f094fe1562d8b5", + "ledger_counts": { + "approvals": 1, + "artifacts": 5, + "attempts": 2, + "evaluator_evidence": 1, + "execution_plans": 1, + "lineage_events": 1, + "policy_decisions": 1, + "quality_results": 1, + "run_events": 4 + }, + "platform_run_state_version": 3, + "platform_run_status": "succeeded", + "provider_observation_id": "85bead83-1901-5649-b6e4-fe46d01c9ea9", + "provider_observation_sha256": "eed782687756dcddbf7f0966f70902edd037f0fc1e2acf3bb72a860e780d0819" + } + }, + "control_ledger_continuity_verified": true, + "control_restart": { + "after": { + "container_id": "6cacc704c64aa75295341396101f535300f8aadc117cff59fd1cff7717c99494", + "container_name": "gda-m3-24-control-229740ac50ebb53b", + "container_running": true, + "container_status": "running", + "credential_material_recorded": false, + "database_ref": "docker:gda-m3-24-control-229740ac50ebb53b/postgres", + "expires_at": "2026-08-07T04:15:23.082316Z", + "host_port": 64667, + "owner": "team:metadata-platform", + "process_id": 2093386, + "retention_id": "m3-24-229740ac50ebb53b", + "started_at": "2026-07-31T05:43:15.197296672Z", + "volume_name": "gda-m3-24-control-229740ac50ebb53b", + "volume_retained": true + }, + "before": { + "container_id": "6cacc704c64aa75295341396101f535300f8aadc117cff59fd1cff7717c99494", + "container_name": "gda-m3-24-control-229740ac50ebb53b", + "container_running": true, + "container_status": "running", + "credential_material_recorded": false, + "database_ref": "docker:gda-m3-24-control-229740ac50ebb53b/postgres", + "expires_at": "2026-08-07T04:15:23.082316Z", + "host_port": 64667, + "owner": "team:metadata-platform", + "process_id": 2087977, + "retention_id": "m3-24-229740ac50ebb53b", + "started_at": "2026-07-31T05:40:53.703034343Z", + "volume_name": "gda-m3-24-control-229740ac50ebb53b", + "volume_retained": true + } + }, + "credential_material_recorded": false, + "durable_catalog_verified": false, + "errors": [], + "evidence_sha256": "1b5a5ceeadee88868bab6237b3f3280c8b13793cc54193592fec7dbbfdd4e8a6", + "exact_terminal_replay_after_restart_verified": true, + "gravitino": { + "after": { + "read_status": 200, + "table_projection": { + "columns": [ + { + "name": "BSM", + "nullable": false, + "type": "string" + }, + { + "name": "geometry", + "nullable": false, + "type": "binary" + }, + { + "name": "srid", + "nullable": false, + "type": "integer" + }, + { + "name": "min_x", + "nullable": false, + "type": "double" + }, + { + "name": "min_y", + "nullable": false, + "type": "double" + }, + { + "name": "max_x", + "nullable": false, + "type": "double" + }, + { + "name": "max_y", + "nullable": false, + "type": "double" + }, + { + "name": "row_sha256", + "nullable": false, + "type": "string" + } + ], + "content_sha256": "bdc06792e8b935176ee6df6f6f6d4be1535622d54d9b994a778cabfe5a574618", + "name": "cultural_districts", + "provider_revision": "m3-22-real-feature-ingestion-v1", + "resource_urn": "gda://metadata-authorization-local/data_product/chongqing-cultural-districts-iceberg", + "resource_version_id": "a6000000-0000-4000-8000-000000000002" + }, + "table_projection_sha256": "f30feb94a5a8280597f331a7f965762bfabb9b82397af49dda05b61ce00bbb1e" + }, + "before": { + "read_status": 200, + "table_projection": { + "columns": [ + { + "name": "BSM", + "nullable": false, + "type": "string" + }, + { + "name": "geometry", + "nullable": false, + "type": "binary" + }, + { + "name": "srid", + "nullable": false, + "type": "integer" + }, + { + "name": "min_x", + "nullable": false, + "type": "double" + }, + { + "name": "min_y", + "nullable": false, + "type": "double" + }, + { + "name": "max_x", + "nullable": false, + "type": "double" + }, + { + "name": "max_y", + "nullable": false, + "type": "double" + }, + { + "name": "row_sha256", + "nullable": false, + "type": "string" + } + ], + "content_sha256": "bdc06792e8b935176ee6df6f6f6d4be1535622d54d9b994a778cabfe5a574618", + "name": "cultural_districts", + "provider_revision": "m3-22-real-feature-ingestion-v1", + "resource_urn": "gda://metadata-authorization-local/data_product/chongqing-cultural-districts-iceberg", + "resource_version_id": "a6000000-0000-4000-8000-000000000002" + }, + "table_projection_sha256": "f30feb94a5a8280597f331a7f965762bfabb9b82397af49dda05b61ce00bbb1e" + } + }, + "gravitino_catalog_continuity_verified": true, + "iceberg_material_continuity_verified": true, + "independent_failure_domains_verified": false, + "independent_quality": { + "after": { + "data_body_sha256": "6cc0fc9eaf48f8106f9afe192704c44407c86c9ea119ae20894bf369a8e74779", + "data_key_sha256": "fbbb249a176bff74f3ee1b756aad1a5233fa007347f68cf466606ddae78e8402", + "data_size_bytes": 94603, + "feature_payload_recorded": false, + "geometry_values_recorded": false, + "identifier_values_recorded": false, + "metrics": { + "bbox_match_count": 20, + "feature_count": 20, + "geometry_z_count": 20, + "non_empty_geometry_count": 20, + "positive_area_count": 20, + "row_fingerprint_match_count": 20, + "srid_match_count": 20, + "unique_bsm_count": 20, + "valid_geometry_count": 20 + }, + "row_set_sha256": "c26ff708f4b6be082327dff63a6a8659420dbc4cab37dea1cac7b40f147512df" + }, + "before": { + "data_body_sha256": "6cc0fc9eaf48f8106f9afe192704c44407c86c9ea119ae20894bf369a8e74779", + "data_key_sha256": "fbbb249a176bff74f3ee1b756aad1a5233fa007347f68cf466606ddae78e8402", + "data_size_bytes": 94603, + "feature_payload_recorded": false, + "geometry_values_recorded": false, + "identifier_values_recorded": false, + "metrics": { + "bbox_match_count": 20, + "feature_count": 20, + "geometry_z_count": 20, + "non_empty_geometry_count": 20, + "positive_area_count": 20, + "row_fingerprint_match_count": 20, + "srid_match_count": 20, + "unique_bsm_count": 20, + "valid_geometry_count": 20 + }, + "row_set_sha256": "c26ff708f4b6be082327dff63a6a8659420dbc4cab37dea1cac7b40f147512df" + } + }, + "independent_quality_continuity_verified": true, + "kubernetes_restart": { + "after": { + "context": "docker-desktop", + "gravitino": { + "image": "docker.io/gda/gravitino:1.3.0-local-arm64", + "image_id": "sha256:18e24b43be854dabdc13e96b1019eb3dc691d59cc64e411aa6a3cc49225fe2d3", + "node_name": "desktop-worker", + "persistent_volume_claims": [], + "pod_name": "gravitino-persistence-0", + "pod_uid": "76bfc413-582b-4e4e-a830-2436d5a0e7a3", + "pvc": null, + "ready_replicas": 1, + "service_account": "gravitino-persistence", + "service_account_automount_disabled": true, + "statefulset_uid": "a316201a-8ba0-4b27-8ce8-bc27965d9b10" + }, + "gravitino_aws_sdk_mounted": true, + "gravitino_host_image_id": "sha256:d355dc7e92f9e3545d717f3eab2cbdf412115f2b82e1e544d7f6235c1eacd5a5", + "gravitino_jdbc_driver_mounted": true, + "iceberg_rest": { + "aws_sdk_mounted": true, + "image": "docker.io/gda/gravitino:1.3.0-local-arm64", + "image_id": "sha256:18e24b43be854dabdc13e96b1019eb3dc691d59cc64e411aa6a3cc49225fe2d3", + "jdbc_driver_mounted": true, + "path": "/iceberg", + "ready": true + }, + "minio_host_image_id": "sha256:a1ea29fa28355559ef137d71fc570e508a214ec84ff8083e39bc5428980b015e", + "namespace": { + "name": "gda-metadata-spark-object-store", + "uid": "824a5904-70cc-4a85-8503-ca83acbcde16" + }, + "object_store": { + "image": "docker.io/minio/minio:RELEASE.2025-04-22T22-12-26Z", + "image_id": "docker.io/minio/minio@sha256:a1ea29fa28355559ef137d71fc570e508a214ec84ff8083e39bc5428980b015e", + "node_name": "desktop-control-plane", + "persistent_volume_claims": [ + "data-metadata-object-store-0" + ], + "pod_name": "metadata-object-store-0", + "pod_uid": "f56f3a9b-8f57-434d-bcd0-50cca28e0fe6", + "pvc": { + "name": "data-metadata-object-store-0", + "phase": "Bound", + "storage_class": "standard", + "uid": "bddccc8e-1da2-4155-9c8c-a8a23a87b7a4", + "volume_name": "pvc-bddccc8e-1da2-4155-9c8c-a8a23a87b7a4" + }, + "ready_replicas": 1, + "service_account": "metadata-object-store", + "service_account_automount_disabled": true, + "statefulset_uid": "d8e3727b-8812-4c6c-9dc3-248f6f1132cd" + }, + "object_store_service": { + "name": "metadata-object-store", + "ports": [ + { + "name": "api", + "port": 9000 + } + ], + "type": "ClusterIP", + "uid": "47b4845f-6776-40f5-8f31-781c5eea460c" + }, + "postgresql": { + "image": "docker.io/library/postgres:16.10-bookworm", + "image_id": "docker.io/library/postgres@sha256:38471f330eb885e04de130b768d6db4e10469e2311879c7e5c699f6d2d8a1c74", + "node_name": "desktop-worker", + "persistent_volume_claims": [ + "data-gravitino-persistence-postgresql-0" + ], + "pod_name": "gravitino-persistence-postgresql-0", + "pod_uid": "e8c09043-da48-42c0-b7bc-f7188ce71fda", + "pvc": { + "name": "data-gravitino-persistence-postgresql-0", + "phase": "Bound", + "storage_class": "standard", + "uid": "6b4cdb87-5f96-4f65-a571-d8390f6ab405", + "volume_name": "pvc-6b4cdb87-5f96-4f65-a571-d8390f6ab405" + }, + "ready_replicas": 1, + "service_account": "gravitino-persistence-postgresql", + "service_account_automount_disabled": true, + "statefulset_uid": "a2711eb2-a329-4264-8ac8-a6b52b186cbb" + }, + "postgresql_service": { + "name": "gravitino-persistence-postgresql", + "ports": [ + { + "name": "postgresql", + "port": 5432 + } + ], + "type": "ClusterIP", + "uid": "de4f8f3a-65fa-4d9e-bdf4-a4db57b6c51d" + }, + "service": { + "name": "gravitino-persistence", + "ports": [ + { + "name": "http", + "port": 8090 + }, + { + "name": "iceberg-rest", + "port": 9001 + } + ], + "type": "ClusterIP", + "uid": "dd03d369-7873-4514-b6fc-aadbb35ce0ab" + }, + "source_schema_sha256": "7a2d605a677a462ca619dba594ce7ebcf500358345560ad084c1b67a25c722df", + "spark_host_image_id": "sha256:f201367640c7583add224796a629150e63d3859ddd7fe9fd47741662a6d415bb" + }, + "before": { + "context": "docker-desktop", + "gravitino": { + "image": "docker.io/gda/gravitino:1.3.0-local-arm64", + "image_id": "sha256:18e24b43be854dabdc13e96b1019eb3dc691d59cc64e411aa6a3cc49225fe2d3", + "node_name": "desktop-worker", + "persistent_volume_claims": [], + "pod_name": "gravitino-persistence-0", + "pod_uid": "baa6d86a-709c-4cd3-a77d-3f3cff07409c", + "pvc": null, + "ready_replicas": 1, + "service_account": "gravitino-persistence", + "service_account_automount_disabled": true, + "statefulset_uid": "a316201a-8ba0-4b27-8ce8-bc27965d9b10" + }, + "gravitino_aws_sdk_mounted": true, + "gravitino_host_image_id": "sha256:d355dc7e92f9e3545d717f3eab2cbdf412115f2b82e1e544d7f6235c1eacd5a5", + "gravitino_jdbc_driver_mounted": true, + "iceberg_rest": { + "aws_sdk_mounted": true, + "image": "docker.io/gda/gravitino:1.3.0-local-arm64", + "image_id": "sha256:18e24b43be854dabdc13e96b1019eb3dc691d59cc64e411aa6a3cc49225fe2d3", + "jdbc_driver_mounted": true, + "path": "/iceberg", + "ready": true + }, + "minio_host_image_id": "sha256:a1ea29fa28355559ef137d71fc570e508a214ec84ff8083e39bc5428980b015e", + "namespace": { + "name": "gda-metadata-spark-object-store", + "uid": "824a5904-70cc-4a85-8503-ca83acbcde16" + }, + "object_store": { + "image": "docker.io/minio/minio:RELEASE.2025-04-22T22-12-26Z", + "image_id": "docker.io/minio/minio@sha256:a1ea29fa28355559ef137d71fc570e508a214ec84ff8083e39bc5428980b015e", + "node_name": "desktop-control-plane", + "persistent_volume_claims": [ + "data-metadata-object-store-0" + ], + "pod_name": "metadata-object-store-0", + "pod_uid": "17c0ca93-2e1f-4ec8-b2d3-566365812922", + "pvc": { + "name": "data-metadata-object-store-0", + "phase": "Bound", + "storage_class": "standard", + "uid": "bddccc8e-1da2-4155-9c8c-a8a23a87b7a4", + "volume_name": "pvc-bddccc8e-1da2-4155-9c8c-a8a23a87b7a4" + }, + "ready_replicas": 1, + "service_account": "metadata-object-store", + "service_account_automount_disabled": true, + "statefulset_uid": "d8e3727b-8812-4c6c-9dc3-248f6f1132cd" + }, + "object_store_service": { + "name": "metadata-object-store", + "ports": [ + { + "name": "api", + "port": 9000 + } + ], + "type": "ClusterIP", + "uid": "47b4845f-6776-40f5-8f31-781c5eea460c" + }, + "postgresql": { + "image": "docker.io/library/postgres:16.10-bookworm", + "image_id": "docker.io/library/postgres@sha256:38471f330eb885e04de130b768d6db4e10469e2311879c7e5c699f6d2d8a1c74", + "node_name": "desktop-worker", + "persistent_volume_claims": [ + "data-gravitino-persistence-postgresql-0" + ], + "pod_name": "gravitino-persistence-postgresql-0", + "pod_uid": "110029b4-6c93-49a3-8ff3-625d51971243", + "pvc": { + "name": "data-gravitino-persistence-postgresql-0", + "phase": "Bound", + "storage_class": "standard", + "uid": "6b4cdb87-5f96-4f65-a571-d8390f6ab405", + "volume_name": "pvc-6b4cdb87-5f96-4f65-a571-d8390f6ab405" + }, + "ready_replicas": 1, + "service_account": "gravitino-persistence-postgresql", + "service_account_automount_disabled": true, + "statefulset_uid": "a2711eb2-a329-4264-8ac8-a6b52b186cbb" + }, + "postgresql_service": { + "name": "gravitino-persistence-postgresql", + "ports": [ + { + "name": "postgresql", + "port": 5432 + } + ], + "type": "ClusterIP", + "uid": "de4f8f3a-65fa-4d9e-bdf4-a4db57b6c51d" + }, + "service": { + "name": "gravitino-persistence", + "ports": [ + { + "name": "http", + "port": 8090 + }, + { + "name": "iceberg-rest", + "port": 9001 + } + ], + "type": "ClusterIP", + "uid": "dd03d369-7873-4514-b6fc-aadbb35ce0ab" + }, + "source_schema_sha256": "7a2d605a677a462ca619dba594ce7ebcf500358345560ad084c1b67a25c722df", + "spark_host_image_id": "sha256:f201367640c7583add224796a629150e63d3859ddd7fe9fd47741662a6d415bb" + }, + "order": [ + "statefulset/gravitino-persistence-postgresql", + "statefulset/metadata-object-store", + "statefulset/gravitino-persistence" + ] + }, + "kubernetes_runtime_restart_verified": true, + "local_retained_real_feature_restart_recovery_verified": true, + "material": { + "after": { + "data_file_count": 1, + "fields": [ + { + "name": "BSM", + "required": true, + "type": "string" + }, + { + "name": "geometry", + "required": true, + "type": "binary" + }, + { + "name": "srid", + "required": true, + "type": "int" + }, + { + "name": "min_x", + "required": true, + "type": "double" + }, + { + "name": "min_y", + "required": true, + "type": "double" + }, + { + "name": "max_x", + "required": true, + "type": "double" + }, + { + "name": "max_y", + "required": true, + "type": "double" + }, + { + "name": "row_sha256", + "required": true, + "type": "string" + } + ], + "manifest_file_count": 2, + "metadata_body_sha256": "6cceb8ba61378122a989e9b6046c328c517d134ee4513af5697d92c672ab772c", + "metadata_file_count": 2, + "object_count": 5, + "object_inventory_sha256": "de4a0efed9fdb68f0019b843377f6c8de71664de955130d0dd38e99eccdb8034", + "schema_id": 0, + "snapshot_id": 8034081021802585202, + "table_location": "s3://gda-metadata-warehouse/warehouse/cultural_heritage/cultural_districts" + }, + "before": { + "data_file_count": 1, + "fields": [ + { + "name": "BSM", + "required": true, + "type": "string" + }, + { + "name": "geometry", + "required": true, + "type": "binary" + }, + { + "name": "srid", + "required": true, + "type": "int" + }, + { + "name": "min_x", + "required": true, + "type": "double" + }, + { + "name": "min_y", + "required": true, + "type": "double" + }, + { + "name": "max_x", + "required": true, + "type": "double" + }, + { + "name": "max_y", + "required": true, + "type": "double" + }, + { + "name": "row_sha256", + "required": true, + "type": "string" + } + ], + "manifest_file_count": 2, + "metadata_body_sha256": "6cceb8ba61378122a989e9b6046c328c517d134ee4513af5697d92c672ab772c", + "metadata_file_count": 2, + "object_count": 5, + "object_inventory_sha256": "de4a0efed9fdb68f0019b843377f6c8de71664de955130d0dd38e99eccdb8034", + "schema_id": 0, + "snapshot_id": 8034081021802585202, + "table_location": "s3://gda-metadata-warehouse/warehouse/cultural_heritage/cultural_districts" + } + }, + "new_authority_facts_created": false, + "new_ingestion_executed": false, + "oidc_verified": false, + "ordered_stateful_restart_verified": true, + "output_content_sha256": "bdc06792e8b935176ee6df6f6f6d4be1535622d54d9b994a778cabfe5a574618", + "output_resource_version_id": "a6000000-0000-4000-8000-000000000002", + "persistent_scheduler_verified": false, + "point_in_time_recovery_verified": false, + "production_ingestion_verified": false, + "production_object_store_verified": false, + "production_ready": false, + "production_restart_recovery_verified": false, + "production_scheduler_verified": false, + "production_tenant_attestation_verified": false, + "protected_workload_identity_verified": false, + "retention_expires_at": "2026-08-07T04:15:23.082316Z", + "retention_id": "m3-24-229740ac50ebb53b", + "run_id": "a9000000-0000-4000-8000-000000000009", + "runtime_port_forwards_stopped": true, + "same_retention_identity_verified": true, + "schema": "gda.retained_real_feature_restart_recovery_evidence.v1", + "source_absolute_path_committed": false, + "source_dataset_committed": false, + "source_evidence_sha256": "d966668b5a2ea57c7a4b2a3bc9824daab9b0128d9f94e515d7be649b145de418", + "source_feature_payload_committed": false, + "source_payload_absent_after": true, + "source_payload_absent_before": true, + "status": "local_retained_real_feature_restart_recovery_verified", + "tenant_id": "metadata-authorization-local", + "terminal_replay": { + "platform_run_state_version": 3, + "platform_run_status": "succeeded", + "promotion_created": false + }, + "tls_verified": false, + "writes_to_legacy": false +} diff --git a/docs/roadmap.md b/docs/roadmap.md index 55355a04..fc2318b9 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -419,7 +419,7 @@ AR-0 Architecture / Schema / Runtime Truth 当前 Metadata Fabric 证据边界：M1 只读 bridge 合同已验证；ADR-037 至 ADR-046 分别覆盖本地 foundation/recovery/metrics/network-policy 演练与 production readiness contracts；ADR-047 至 ADR-050 已依次建立 deterministic projection plan、本地双 provider replay、tenant-scoped binding ledger 与本地 OpenLineage 幂等 wire delivery；ADR-051 以临时非管理员 OpenMetadata bot 证明项目专用 grant 只有 `table/Create`、`policy/Create` 被 403 拒绝，且 JWT 轮换/吊销后旧值/当前值均返回 401；ADR-052 又在隔离 Gravitino `1.3.0` Basic IdP 中证明 bounded user 的 `USE_CATALOG`、`USE_SCHEMA`、`CREATE_TABLE` 范围、catalog-create 403、密码轮换和用户吊销；ADR-053 将生产 OIDC federation、双 provider integration/workload identity、最小权限、TLS/mTLS、持久 Gravitino catalog、tenant isolation、运营责任和新鲜 protected attestation 冻结为 fail-closed readiness contract。Gravitino `1.3.0` 镜像只发现 Basic IdP，不假设 native OIDC；当前 profile 仍有 40 个外部 blockers 且未提交真实 attestation。`local_openmetadata_minimum_privilege_verified=true` 与 `local_gravitino_minimum_privilege_verified=true` 都只描述各自临时 provider rehearsal；M3-2 ingestion 仍使用 bootstrap admin，Gravitino probe catalog 仍是 memory catalog。因此 `provider_minimum_privilege_verified`、protected workload identity、OIDC、TLS、持久 catalog、生产 ingestion/conformance、生产 lineage receiver、`production_identity_gate_passed` 与 `production_ready` 仍为 false。 -M3-22 至 M3-24 已把一份真实重庆 20-feature EPSG:4490 slice 从受授权 Spark/Sedona + JDBC/S3 Iceberg ingestion，推进到保留 7 天的 staging material、原子 GDA Control `ResourceVersion + 2 Artifacts + QualityResult + LineageEvent` 晋级和数据库裁决的 `succeeded@3`。M3-24 已持久化完整 execution-plan/PolicyDecision/Approval Artifacts，回读真实 DolphinScheduler `SUCCESS`，由独立 evaluator 重开 Parquet 并重算九项空间质量与 row fingerprint，且精确 terminal replay 不新增事实；源 payload ConfigMap 与临时 scheduler 已清理，namespace/PVC/MinIO-Iceberg material 和专用 GDA Control PostgreSQL 受同一 retention ID 约束保留。该结果仍是单开发主机上的 retained local staging rehearsal；生产 identity/storage/tenant attestation、常驻 scheduler/executor、独立故障域、backup/PITR、restart/recovery、staging scale 和完整 Spark/Flink conformance 是下一门槛。边界见 [ADR-070](architecture-decisions/adr-070-retained-real-feature-terminal-success.md)。 +M3-22 至 M3-25 已把一份真实重庆 20-feature EPSG:4490 slice 从受授权 Spark/Sedona + JDBC/S3 Iceberg ingestion，推进到保留 7 天的 staging material、原子 GDA Control `ResourceVersion + 2 Artifacts + QualityResult + LineageEvent` 晋级、数据库裁决的 `succeeded@3` 和保留运行时的受控 restart continuity。M3-24 已持久化完整 execution-plan/PolicyDecision/Approval Artifacts，回读真实 DolphinScheduler `SUCCESS`，由独立 evaluator 重开 Parquet 并重算九项空间质量与 row fingerprint；M3-25 又按 PostgreSQL -> MinIO -> Gravitino -> GDA Control 顺序重启同一运行时，证明稳定 namespace/StatefulSet/Service/PVC/container/volume 身份不变、Pod/PID 轮换，且 Iceberg snapshot/Parquet/Gravitino projection/GDA ledger facts SHA 与 `succeeded@3` 精确 replay 均不漂移、不新增事实。该结果仍是单开发主机上的 retained local staging process-restart rehearsal；生产 identity/storage/tenant attestation、常驻 scheduler/executor、独立故障域、backup/PITR、production restart recovery、staging scale 和完整 Spark/Flink conformance 是下一门槛。边界见 [ADR-070](architecture-decisions/adr-070-retained-real-feature-terminal-success.md) 与 [ADR-071](architecture-decisions/adr-071-retained-real-feature-restart-recovery.md)。 ### AR-2 — Source, Ingestion and Geospatial Lakehouse Vertical Slice（P0） @@ -686,7 +686,7 @@ Golden checks 至少覆盖： 1. 导出所有目标环境 schema/config fingerprint，修复重复 migration ID、checksum 和 fail-open runner。 2. 完成部署、存储、bucket、registry、scheduler/job、API/GIS endpoint、图层/样式/缓存、provider/Gateway、数据资产、消费者和权限事实盘点；部署 OpenMetadata/Gravitino/DolphinScheduler/Temporal sandbox，冻结 owner、version、OIDC、backup/restore 和升级责任。 3. 冻结 ResourceURN、ResourceVersion、PlatformDefinition/PlatformRun/FrameworkAttemptObservation/Artifact/LineageEvent、SubjectContext 与 storage/table/compute provider 最小合同。 -4. 分阶段实现 `gda-metadata-fabric-bridge`：M1 只读 mapping/reconciliation、M2a 本地 foundation/重启连续性、M2b 本地与跨集群恢复、M2c metrics/OTel 故障演练、M2d production readiness contracts，以及 M3-1 至 M3-24 projection、provider identity/interoperability、Active Metadata、真实 feature ingestion、原子 ledger promotion 和 retained terminal success 已验证。下一步以 production identity/storage/tenant attestation 为先决条件，部署常驻 scheduler/executor 与持久 catalog/control/storage，验证 restart/recovery、backup/PITR、独立故障域、staging scale、完整 Spark/Flink conformance 和真实告警/runbook；本地 retained evidence 不计入生产退出门。 +4. 分阶段实现 `gda-metadata-fabric-bridge`：M1 只读 mapping/reconciliation、M2a 本地 foundation/重启连续性、M2b 本地与跨集群恢复、M2c metrics/OTel 故障演练、M2d production readiness contracts，以及 M3-1 至 M3-25 projection、provider identity/interoperability、Active Metadata、真实 feature ingestion、原子 ledger promotion、retained terminal success 和同一保留 authority 的本地进程重启连续性已验证。下一步以 production identity/storage/tenant attestation 为先决条件，部署常驻 scheduler/executor 与持久 catalog/control/storage，验证 backup/PITR、独立故障域、production restart recovery、staging scale、完整 Spark/Flink conformance 和真实告警/runbook；本地 retained evidence 不计入生产退出门。 5. 实现 `gda-orchestration-gateway`、DolphinScheduler process/task/schedule/complement/worker-group、Spark/Flink provider task adapter 和故障注入；不再开发新的 lease/queue/scheduler。 6. 冻结首条地类图斑数据、标准版本、敏感级别、owner、SLO 和 golden result。 7. 冻结 Default Lakehouse、Cloud Managed、Lightweight Integrated profiles；以统一 Run 完成默认 MinIO/Iceberg/Spark/Flink、轻量 PostGIS/DuckDB 和 Azure 代表 adapter 的 conformance smoke。 diff --git a/docs/system-of-record-matrix-2026-07-24.md b/docs/system-of-record-matrix-2026-07-24.md index aa89ce54..ebf9cbf6 100644 --- a/docs/system-of-record-matrix-2026-07-24.md +++ b/docs/system-of-record-matrix-2026-07-24.md @@ -2,9 +2,9 @@ 日期：2026-07-31 -阶段：AR-0 `in_progress`；AR-1 gateway、成功终局 evidence gate、DolphinScheduler adapter sandbox POC、Metadata Fabric M1/M2、M2c-4/M2d-2 production readiness contracts、M3-1 至 M3-24 retained real-feature terminal success 已验证；生产 policy/tenant isolation、生产 identity/object-store attestation、常驻 production consumer/scheduler/executor、restart/recovery、scale/conformance 和生产切换仍 `in_progress` +阶段：AR-0 `in_progress`；AR-1 gateway、成功终局 evidence gate、DolphinScheduler adapter sandbox POC、Metadata Fabric M1/M2、M2c-4/M2d-2 production readiness contracts、M3-1 至 M3-25 retained real-feature restart continuity 已验证；生产 policy/tenant isolation、生产 identity/object-store attestation、常驻 production consumer/scheduler/executor、backup/PITR、独立故障域、production restart recovery、scale/conformance 和生产切换仍 `in_progress` -适用分支：`feat/ar1-metadata-fabric-retained-real-feature-terminal-success` +适用分支：`feat/ar1-metadata-fabric-retained-real-feature-restart-recovery` ## 判定规则 @@ -20,7 +20,7 @@ | SQL schema 历史 | PostgreSQL `schema_migrations`，以完整 migration ID + checksum 为权威 | migration CLI 的 JSON 报告 | 保持现有 ledger；任何 drift fail closed | Data Platform | AR-0，已验证 | | 部署配置策略 | Compose/K8s/进程环境；`platform_truth.CONFIG_SPECS` 定义关键类型与策略；DolphinScheduler worker 与 Active Metadata consumer 均有默认零副本、外部 ConfigMap/Secret 驱动的 Kustomize 模板和静态 validator，前者另有 staging activation preflight | `.env` 仅补默认；脱敏 snapshot、Secret key attestation、未扩容 Deployment 和 `ready_for_activation` 都是观测/模板 | 版本化 DeploymentProfile + secret reference；部署环境始终优先；模板或 preflight 通过都不等于环境已启用 | Platform/SRE/Security | AR-0，部分实现；worker 模板/preflight 本地已验证 | | 环境发布与晋级 | 本地 candidate/registry/provenance/release/live 合同已绑定 publisher、verifier、OCI 和 manifest identity；canonical `main@0182406`、archive refs、三组 active ruleset 与 `staging-provenance` protected environment 已建立，但尚无成功 publisher/verifier 或 deployment | 旧 mainline、feature branch、CI artifact、JSON、离线 report 和合成 `verified_for_staging_apply` 都不能单独成为发布权威；publisher SHA、verifier SHA 与 branch lineage 必须分别验证 | 由受保护 environment 的 DeploymentRevision 绑定 OCI、provenance artifact、release manifest 与全部 live verdict | Platform/SRE/Security/Repository Owner | AR-1 mainline 治理已恢复 -> 首次 GHCR publish/verify -> 真实 staging | -| 后台运行时清单 | `platform_truth.RUNTIME_INVENTORY` 是代码层登记；`gda_control` 已有受控 PlatformRun 写入口；DolphinScheduler managed worker 已登记但尚无生产调用方；Active Metadata consumer 登记为 `activation_request_staging_only`，其 deployment 默认为 0 replicas；M3-24 retained terminal-success rehearsal 与此前 Metadata Fabric rehearsals 均登记为 `local_verification_only` | AST primitive report、worker status JSON、FrameworkAttemptObservation、DolphinScheduler instance state、本地 recovery/metrics/network-policy/catalog/interoperability/failure/outbox/consumer/authorization/delivery/projection/binding/ingestion/promotion/terminal evidence | PlatformRun ledger 唯一登记最终状态；M3-24 的 Run 已由数据库 evidence gate 裁决为 `succeeded@3`，但临时 scheduler/executor、单主机 retained namespace/PVC/MinIO-Iceberg/control DB 和本地 evidence 不得变成生产控制器、catalog/storage/output 或 tenant-isolation 权威 | Platform Architecture | AR-1 adapter/worker、M3-15 至 M3-24 本地控制链和 retained terminal success 已验证；常驻受保护 scheduler/executor、生产 storage/control 与 restart/recovery 待接入 | +| 后台运行时清单 | `platform_truth.RUNTIME_INVENTORY` 是代码层登记；`gda_control` 已有受控 PlatformRun 写入口；DolphinScheduler managed worker 已登记但尚无生产调用方；Active Metadata consumer 登记为 `activation_request_staging_only`，其 deployment 默认为 0 replicas；M3-24 terminal-success 与 M3-25 restart-recovery rehearsal 及此前 Metadata Fabric rehearsals 均登记为 `local_verification_only` | AST primitive report、worker status JSON、FrameworkAttemptObservation、DolphinScheduler instance state、本地 recovery/metrics/network-policy/catalog/interoperability/failure/outbox/consumer/authorization/delivery/projection/binding/ingestion/promotion/terminal/restart evidence | PlatformRun ledger 唯一登记最终状态；M3-24 的 Run 已由数据库 evidence gate 裁决为 `succeeded@3`，M3-25 只证明同一单主机 retained namespace/PVC/MinIO-Iceberg/control DB 的进程重启连续性；两者均不得变成生产控制器、catalog/storage/output、tenant-isolation 或 production recovery 权威 | Platform Architecture | AR-1 adapter/worker、M3-15 至 M3-25 本地控制链、retained terminal success 与进程重启连续性已验证；常驻受保护 scheduler/executor、生产 storage/control、backup/PITR 与独立故障域待接入 | | 原始文件/对象 | 当前 local uploads、S3/MinIO/OBS 均可能被直接写入，权威边界未统一 | 临时上传、下载缓存、预览文件 | Landing object 以 immutable URI + checksum + retention 为权威；本地 scratch 可删除 | Data Platform | AR-2 | | 湖仓表与 snapshot | Iceberg/STAC/S3A 有局部实现，尚无通用发布权威 | STAC item、GeoParquet export | Iceberg catalog snapshot 是分析表版本权威；对象是物理内容，STAC 是发现投影 | Data Platform | AR-2 | | 在线空间数据 | PostGIS 业务表是当前编辑/查询事实，部分临时表混入 | Martin MVT、API JSON、导出文件 | 已批准 DataProductVersion 物化到 PostGIS；不能由瓦片或临时表反向定义产品版本 | GIS/Data Platform | AR-2 -> AR-4 | @@ -68,6 +68,7 @@ 23. M3-22 以 M3-21 candidate 为 predecessor，将同一重庆 bundle 的 20 个真实 EPSG:4490 feature rows 规范化为八列、由精确 PolicyDecision/Approval 授权 Spark/Sedona 写入 JDBC/S3 Iceberg。六项质量计数均为 20，首次执行 `appended/1` 且只有 1 个 snapshot/Parquet，即时 replay 为 `no_op/0`；S3 直读为 1 data + 2 metadata + 2 manifest。输出 ResourceVersion、Artifact、passed QualityResult 与 LineageEvent 只是 path-free candidates，未写 GDA Control，Run 未成功终局；namespace/PV/port-forward 已清理。这不证明生产对象存储、protected identity、完整 engine conformance、生产 ingestion 或 readiness。 24. M3-23 只证明 M3-22 path-free candidates 可在临时 PostgreSQL GDA Control 中按 `ResourceVersion -> output Artifact -> quality evidence Artifact -> QualityResult -> LineageEvent` 单事务追加。缺 authority、半状态、跨租户和 direct mutation 均拒绝，故障注入整笔回滚，精确 replay 不新增；Run 保持 `accepted@0` 且 success finalization 被拒绝。M3-22 material、M3-23 数据库与完整 authorization Artifacts 均未保留或伪造补写；这不证明 persistent authority、terminal success、staging/生产 ingestion 或 readiness。 25. M3-24 以同一真实重庆 20-feature slice 新跑一次受控链：完整 execution-plan/PolicyDecision/Approval 先落 GDA Control，DolphinScheduler Shell task 回调短生命周期 executor 执行 Spark/Sedona JDBC/S3 Iceberg ingestion，真实 `SUCCESS` 只作为 attempt observation；独立 evaluator 重开 retained Parquet 并创建质量 evidence，M3-23 promoter 原子追加 output bundle，既有数据库 finalizer 唯一裁决 `succeeded@3`，精确 terminal replay 不新增。含 BSM/WKB 的 ConfigMap 和临时 scheduler 已删除；namespace/PVC/MinIO-Iceberg material 与专用 GDA Control PostgreSQL 受 retention ID 约束保留 7 天。该单开发主机 rehearsal 不证明生产 identity/storage/tenant attestation、常驻 scheduler/executor、restart/recovery、scale/conformance 或 production readiness。 +26. M3-25 不重新 ingest、不创建新 ResourceVersion/Run，也不改写 M3-24 verdict；它只绑定 M3-24 evidence/retention ID，按 PostgreSQL、MinIO、Gravitino、GDA Control 顺序重启同一保留运行时。三组 Pod UID 与 control PID/StartedAt 必须轮换，namespace/StatefulSet/Service/PVC/image/control container/volume 必须不变；Iceberg inventory/metadata/snapshot、Parquet 九项质量、Gravitino projection 和 GDA Control facts SHA 在重启前后与 exact terminal replay 后必须一致且零新增。该结果只证明单开发主机本地进程重启连续性，不证明 backup/PITR、独立故障域、production restart recovery 或 production readiness。 ## 已建立的 AR-0/AR-1 entry 证据 @@ -115,11 +116,12 @@ - Metadata Fabric M3-22 已由受授权 Spark `3.5.0` + Sedona `1.9.0` 将同一重庆 bundle 的 20 个 EPSG:4490 features 写入 JDBC/S3 Iceberg。六项空间质量计数均为 20，首次 `appended/1`、1 snapshot/Parquet，即时 replay `no_op/0`；S3 直读为 1 data + 2 metadata + 2 manifest。row-set/output/contract/evidence SHA 为 `c26ff708f4b6be082327dff63a6a8659420dbc4cab37dea1cac7b40f147512df` / `bdc06792e8b935176ee6df6f6f6d4be1535622d54d9b994a778cabfe5a574618` / `af211f2d2f4830decb9ffe369cd9e7ec2c9349c2e2c8bd789347a6fdc288e1dc` / `42abd82613eaf28cb53c64280258bc75dba6cf841f9a513a4c801a9f798b9899`。output/quality/lineage 仍是 candidates，runtime/material 已清理，Run 未终局。 - Metadata Fabric M3-23 已把 M3-22 的 output ResourceVersion、output Artifact、quality evidence Artifact、独立 passed QualityResult 与 source-to-output LineageEvent 作为一个 `RunOutputLedgerPromotion`，由独立 promoter 在一个 PlatformGateway 事务中写入真实临时 PostgreSQL，同时保持此前 gateway-bound evidence 指纹不变。缺 authority 先验拒绝；QualityResult 前故障注入后候选计数全为 0；首次 `created=true`、精确 replay `created=false`，最终计数为 `1/2/1/1`。FORCE RLS、最小 grant、跨租户读/直写和八个 direct UPDATE/DELETE 拒绝通过；Run 保持 `accepted@0`，success finalization 被既有 gate 拒绝。promotion/contract/evidence SHA 为 `404b6e4e5d8194f092bd83ef99cbf2d1d727015b926cd438a79eb0210f969a22` / `bd21c81925f66acdfecca5cabd78651f31deab4165da2ccd6900c4e5796e5735` / `f6efea5000791dec1716a8354a8e39a8425b083ca4d409f4bcb61f0e7e03580d`。M3-22 material 与临时数据库已清理，完整 authorization Artifacts 未补写；`output_material_retained=false`、`platform_run_succeeded=false`、`production_ready=false`。 - Metadata Fabric M3-24 已用新的受控执行把同一重庆 slice 推进到 retained terminal success。完整 execution-plan/PolicyDecision/Approval Artifacts 在 dispatch 前持久化；DolphinScheduler `3.4.2` task 返回真实 `SUCCESS`，provider observations 为 2；独立 evaluator 重开 94,603-byte Parquet 并使 feature/unique/non-empty/valid/Z/SRID/positive-area/bbox/row-fingerprint 九项计数均为 20。M3-23 promoter 与 migration 096 finalizer 依次提交 5 Artifacts、1 QualityResult、1 LineageEvent 和 `succeeded@3`，精确 replay 不新增。contract/evidence SHA 为 `9c8f20ca1fb9995530c4e988ced627f665857ecdf0e104bb7d07c4a4a486057a` / `d966668b5a2ea57c7a4b2a3bc9824daab9b0128d9f94e515d7be649b145de418`；retention ID 为 `m3-24-229740ac50ebb53b`，到期 `2026-08-07T04:15:23.082316Z`。source payload 与 scheduler 已清理，retained material/control DB 仍可审计；六项 production claims 与 `production_ready` 均为 `false`。 +- Metadata Fabric M3-25 已在同一 retention ID 和 expiry 内按 PostgreSQL -> MinIO -> Gravitino -> GDA Control 顺序完成进程重启。三组 Pod UID 与 control PID 均轮换，namespace/StatefulSet/Service/PVC/image/control container/volume 均保持；object inventory SHA `de4a0efed9fdb68f0019b843377f6c8de71664de955130d0dd38e99eccdb8034`、snapshot `8034081021802585202`、94,603-byte Parquet body/row-set SHA、九项 20-count 质量、Gravitino projection SHA `f30feb94a5a8280597f331a7f965762bfabb9b82397af49dda05b61ce00bbb1e` 与 GDA Control facts SHA `5c0b8a58729e551c250b0410bcebdfe3f019f215a50b20d503f094fe1562d8b5` 前后不变；terminal replay 为 `created=false`，ledger 计数零新增。contract/evidence SHA 为 `83ed15ae4eed85e0c261c2b3a04ea2ad559f3deb7b86c7c2f2dedd0cf28d23d0` / `1b5a5ceeadee88868bab6237b3f3280c8b13793cc54193592fec7dbbfdd4e8a6`；backup/PITR、独立故障域、production restart recovery 与 `production_ready` 均为 `false`。 ## 下一验收证据 -- M3-21 空表 metadata promotion、M3-22 临时真实 feature ingestion、M3-23 临时 ledger promotion 与 M3-24 retained terminal success 都不计入生产对象存储、持久生产 authority 或 ingestion 退出门，四阶段历史 candidate/evidence 保持不变； -- M3-24 已跨过 retained staging material、完整 authorization、真实 provider success observation、独立 quality evidence provenance 和 `succeeded@3` 门槛；下一步必须由受保护 production identity/storage/tenant attestation 选择生产 provider，并部署常驻 scheduler/executor、持久 catalog/control/storage，验证 restart/recovery、backup/PITR、独立故障域、staging scale 与 Spark/Flink conformance； +- M3-21 空表 metadata promotion、M3-22 临时真实 feature ingestion、M3-23 临时 ledger promotion、M3-24 retained terminal success 与 M3-25 本地进程重启连续性都不计入生产对象存储、持久生产 authority、production recovery 或 ingestion 退出门，五阶段历史 candidate/evidence 保持不变； +- M3-25 已跨过同一 retained authority 的 PostgreSQL/MinIO/Gravitino/control process restart continuity、独立数据/目录/ledger 回读与 terminal replay 零新增门槛；下一步必须由受保护 production identity/storage/tenant attestation 选择生产 provider，并部署常驻 scheduler/executor、持久 catalog/control/storage，验证 backup/PITR、独立故障域、production restart recovery、staging scale 与 Spark/Flink conformance； - 完成首次 application subject publish 与 protected verifier run；当前 mainline、archive refs、ruleset、required reviewer、禁止 bypass 和 environment enable variable 已配置并复核； - 真实 provenance artifact verify、受保护 overlay 的 `verified_for_staging_apply` release report，以及 staging/production 的 schema、config/runtime snapshot、registry/live DeploymentRevision 绑定、release/live artifact attestation 和环境 compare 报告； - staging 的 migration role、应用 login membership、连接池 role/tenant 复位、双租户 API 和 success finalization 运行产物； diff --git a/scripts/metadata-fabric-retained-real-feature-restart-recovery.sh b/scripts/metadata-fabric-retained-real-feature-restart-recovery.sh new file mode 100755 index 00000000..7b627469 --- /dev/null +++ b/scripts/metadata-fabric-retained-real-feature-restart-recovery.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +set -euo pipefail + +repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +common_git_dir="$(git -C "$repo_root" rev-parse --path-format=absolute --git-common-dir 2>/dev/null || true)" +shared_root="" +if [ -n "$common_git_dir" ]; then + shared_root="$(cd "$common_git_dir/.." && pwd)" +fi + +if [ -n "${PYTHON:-}" ]; then + : +elif [ -x "$repo_root/.venv/bin/python" ]; then + PYTHON="$repo_root/.venv/bin/python" +elif [ -n "$shared_root" ] && [ -x "$shared_root/.venv/bin/python" ]; then + PYTHON="$shared_root/.venv/bin/python" +else + PYTHON="python" +fi + +cd "$repo_root" +exec "$PYTHON" -m data_agent.metadata_fabric_retained_real_feature_restart_recovery "$@" From dbb02e15f6f456c40df43e292d68f82b785c2d47 Mon Sep 17 00:00:00 2001 From: Ning Zhou Date: Fri, 31 Jul 2026 14:09:04 +0800 Subject: [PATCH 2/2] ci: run checks for retained restart recovery stack --- .github/workflows/ci.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8dee4e41..407eb9ea 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -39,6 +39,7 @@ on: - feat/ar1-metadata-fabric-object-store-active-metadata-promotion - feat/ar1-metadata-fabric-real-feature-ingestion - feat/ar1-metadata-fabric-real-feature-ledger-promotion + - feat/ar1-metadata-fabric-retained-real-feature-terminal-success env: PYTHON_VERSION: "3.13"