From 4f3d9c48e2864641bfa12e3c311f47a43e288197 Mon Sep 17 00:00:00 2001 From: Ning Zhou Date: Thu, 30 Jul 2026 01:09:58 +0800 Subject: [PATCH] feat(platform): reconcile uncertain Spark commits --- .github/workflows/ci.yml | 5 + ...uncertain-commit-reconciliation.local.yaml | 115 +++ ...c_spark_uncertain_commit_reconciliation.py | 914 ++++++++++++++++++ data_agent/platform_truth.py | 20 + ...c_spark_uncertain_commit_reconciliation.py | 354 +++++++ data_agent/test_platform_truth.py | 6 + ...l-spark-uncertain-commit-reconciliation.md | 34 + ...tain-commit-reconciliation-2026-07-30.json | 437 +++++++++ docs/roadmap-ar0-platform-truth-2026-07-24.md | 5 +- docs/system-of-record-matrix-2026-07-24.md | 9 +- .../probe.py | 368 +++++++ ...c-spark-uncertain-commit-reconciliation.sh | 22 + 12 files changed, 2283 insertions(+), 6 deletions(-) create mode 100644 config/metadata-fabric-spark-uncertain-commit-reconciliation.local.yaml create mode 100644 data_agent/metadata_fabric_spark_uncertain_commit_reconciliation.py create mode 100644 data_agent/test_metadata_fabric_spark_uncertain_commit_reconciliation.py create mode 100644 docs/architecture-decisions/adr-059-local-spark-uncertain-commit-reconciliation.md create mode 100644 docs/evidence/metadata-fabric-spark-uncertain-commit-reconciliation-2026-07-30.json create mode 100644 k8s/metadata-fabric-spark-uncertain-commit-reconciliation/probe.py create mode 100755 scripts/metadata-fabric-spark-uncertain-commit-reconciliation.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1332e3ae..39aa0778 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,6 +28,7 @@ on: - feat/ar1-metadata-fabric-spark-object-store-interoperability - feat/ar1-metadata-fabric-object-store-readiness-gate - feat/ar1-metadata-fabric-spark-commit-failure-recovery + - feat/ar1-metadata-fabric-spark-uncertain-commit-reconciliation env: PYTHON_VERSION: "3.13" @@ -161,6 +162,9 @@ jobs: - name: Validate metadata fabric Spark commit-failure recovery evidence run: python -m data_agent.metadata_fabric_spark_commit_failure_recovery validate + - name: Validate metadata fabric Spark uncertain-commit reconciliation evidence + run: python -m data_agent.metadata_fabric_spark_uncertain_commit_reconciliation validate + - name: Validate DolphinScheduler adapter boundary run: python -m data_agent.dolphinscheduler_adapter validate @@ -224,6 +228,7 @@ jobs: data_agent/test_metadata_fabric_spark_object_store_interoperability.py \ data_agent/test_metadata_fabric_object_store_gate.py \ data_agent/test_metadata_fabric_spark_commit_failure_recovery.py \ + data_agent/test_metadata_fabric_spark_uncertain_commit_reconciliation.py \ data_agent/test_metadata_fabric_otel_failure_rehearsal.py \ data_agent/test_metadata_fabric_otel_metrics.py \ data_agent/test_metadata_fabric_provider_metrics.py \ diff --git a/config/metadata-fabric-spark-uncertain-commit-reconciliation.local.yaml b/config/metadata-fabric-spark-uncertain-commit-reconciliation.local.yaml new file mode 100644 index 00000000..fc8deca2 --- /dev/null +++ b/config/metadata-fabric-spark-uncertain-commit-reconciliation.local.yaml @@ -0,0 +1,115 @@ +schema: gda.metadata_fabric_spark_uncertain_commit_reconciliation_profile.v1 +environment: local_docker_desktop + +cluster: + context: docker-desktop + source_namespace: gda-metadata-sandbox + rehearsal_namespace: gda-metadata-spark-commit-failure + source_schema_configmap: metadata-gravitino-schema-1-3-0 + storage_class: standard + object_store_node: desktop-control-plane + compute_node: desktop-worker + +runtime: + manifest: k8s/metadata-fabric-spark-commit-failure-recovery + gravitino_version: 1.3.0 + gravitino_image: gda/gravitino:1.3.0-local-arm64 + gravitino_host_image_id: sha256:d355dc7e92f9e3545d717f3eab2cbdf412115f2b82e1e544d7f6235c1eacd5a5 + gravitino_kubernetes_image_id: sha256:18e24b43be854dabdc13e96b1019eb3dc691d59cc64e411aa6a3cc49225fe2d3 + iceberg_rest_version: 1.11.0 + postgresql_version: 16.10-bookworm + postgresql_image: postgres:16.10-bookworm + postgresql_image_digest: sha256:38471f330eb885e04de130b768d6db4e10469e2311879c7e5c699f6d2d8a1c74 + spark_version: 3.5.0 + iceberg_spark_runtime_version: 1.6.1 + spark_image: gisdataagent/mmfe-spark-runtime:local + spark_host_image_id: sha256:f201367640c7583add224796a629150e63d3859ddd7fe9fd47741662a6d415bb + spark_kubernetes_image_id: sha256:4a4522bfd4e6d1c6c90a244d0145841fbfbbf21ed16ee29ca8b681b5cec60058 + minio_version: RELEASE.2025-04-22T22-12-26Z + minio_image: minio/minio:RELEASE.2025-04-22T22-12-26Z + minio_host_image_id: sha256:a1ea29fa28355559ef137d71fc570e508a214ec84ff8083e39bc5428980b015e + minio_kubernetes_image_id: sha256:a1ea29fa28355559ef137d71fc570e508a214ec84ff8083e39bc5428980b015e + service: gravitino-persistence + gravitino_service_port: 8090 + iceberg_rest_service_port: 9001 + iceberg_rest_path: /iceberg + object_store_service: metadata-object-store + object_store_service_port: 9000 + spark_job: spark-commit-failure-probe + authenticator: basic + access_control_enabled: true + transport: local_cluster_http + +dependency: + evidence_path: docs/evidence/metadata-fabric-spark-commit-failure-recovery-2026-07-29.json + evidence_fingerprint: 39571cdac1e4043bcfc2d03a73b2b12ff925210daf8ae36bc640b8cb14d89401 + required_claim: local_spark_commit_failure_recovery_verified + +identity: + service_admin: gda-object-store-admin + user: gda-metadata-projection + role: gda-table-projection + material_delivery: runtime_generated_ephemeral_kubernetes_object + +catalog: + provider: lakehouse-iceberg + backend: jdbc + uri: jdbc:postgresql://gravitino-persistence-postgresql:5432/iceberg + jdbc_driver: org.postgresql.Driver + jdbc_driver_source: /opt/gravitino/libs/postgresql-42.7.0.jar + gravitino_jdbc_driver_mount: /opt/gravitino/catalogs/lakehouse-iceberg/libs/postgresql-42.7.0.jar + rest_jdbc_driver_mount: /opt/gravitino/iceberg-rest-server/libs/postgresql-42.7.0.jar + jdbc_initialize: true + warehouse: s3://gda-metadata-warehouse/warehouse + io_impl: org.apache.iceberg.aws.s3.S3FileIO + s3_endpoint: http://metadata-object-store:9000 + s3_region: us-east-1 + s3_path_style_access: true + bucket: gda-metadata-warehouse + object_prefix: warehouse/published/gda_spark_commit_failure_probe/ + authentication_mode: runtime_generated_static_local_material + postgresql_pvc: data-gravitino-persistence-postgresql-0 + object_store_pvc: data-metadata-object-store-0 + interoperability_scope: local_cross_node_s3_commit_failure_recovery + +scope: + metalake: gda_commit_failure + catalog: lakehouse + schema: published + table: gda_spark_commit_failure_probe + denied_catalog: unauthorized_catalog + role_securable_objects: + - full_name: lakehouse + type: CATALOG + privileges: + - name: USE_CATALOG + condition: ALLOW + - full_name: lakehouse.published + type: SCHEMA + privileges: + - name: CREATE_TABLE + condition: ALLOW + - name: USE_SCHEMA + condition: ALLOW + +claims: + local_spark_uncertain_commit_reconciliation_verified: false + local_provider_committed_response_loss_verified: false + local_commit_outcome_readback_verified: false + local_duplicate_resubmission_prevented: false + local_single_visible_commit_verified: false + gravitino_api_metadata_readback_verified: false + local_cross_node_object_store_verified: false + object_store_metadata_verified: false + spark_cancel_verified: false + spark_reconcile_verified: false + spark_lineage_verified: false + persistent_catalog_identity_binding_verified: false + protected_workload_identity_verified: false + oidc_verified: false + tls_verified: false + production_object_store_verified: false + spark_conformance_verified: false + flink_conformance_verified: false + production_ingestion_verified: false + production_ready: false diff --git a/data_agent/metadata_fabric_spark_uncertain_commit_reconciliation.py b/data_agent/metadata_fabric_spark_uncertain_commit_reconciliation.py new file mode 100644 index 00000000..6a59db84 --- /dev/null +++ b/data_agent/metadata_fabric_spark_uncertain_commit_reconciliation.py @@ -0,0 +1,914 @@ +"""Verify local reconciliation after a committed Iceberg response is lost. + +The rehearsal forwards one Spark table commit to Gravitino, receives provider +success, then returns HTTP 504 to Spark instead of the success response. It +uses readback to classify the outcome as committed and does not submit another +write. This remains local Docker Desktop evidence, not production exactly-once. +""" + +from __future__ import annotations + +import argparse +import json +import secrets +import sys +from collections.abc import Mapping +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Literal + +import yaml +from botocore.exceptions import BotoCoreError, ClientError +from pydantic import Field, SecretStr, ValidationError + +from . import metadata_fabric_gravitino_identity as identity +from . import metadata_fabric_gravitino_jdbc_restart as jdbc_restart +from . import metadata_fabric_ingestion_replay as ingestion_replay +from . import metadata_fabric_provider_metrics as provider_metrics +from . import metadata_fabric_recovery_rehearsal as recovery +from . import metadata_fabric_spark_commit_failure_recovery as commit_failure +from . import metadata_fabric_spark_iceberg_rest_interoperability as spark_interop +from . import metadata_fabric_spark_object_store_interoperability as object_interop + + +PROFILE_SCHEMA = "gda.metadata_fabric_spark_uncertain_commit_reconciliation_profile.v1" +CONTRACT_SCHEMA = ( + "gda.metadata_fabric_spark_uncertain_commit_reconciliation_contract.v1" +) +OBSERVATION_SCHEMA = ( + "gda.metadata_fabric_spark_uncertain_commit_reconciliation_observation.v1" +) +EVIDENCE_SCHEMA = ( + "gda.metadata_fabric_spark_uncertain_commit_reconciliation_evidence.v1" +) +VALIDATION_SCHEMA = ( + "gda.metadata_fabric_spark_uncertain_commit_reconciliation_validation.v1" +) + +CONTEXT = commit_failure.CONTEXT +SOURCE_NAMESPACE = commit_failure.SOURCE_NAMESPACE +REHEARSAL_NAMESPACE = commit_failure.REHEARSAL_NAMESPACE +OBJECT_STORE_NODE = commit_failure.OBJECT_STORE_NODE +COMPUTE_NODE = commit_failure.COMPUTE_NODE +DEPENDENCY_EVIDENCE_FINGERPRINT = ( + "39571cdac1e4043bcfc2d03a73b2b12ff925210daf8ae36bc640b8cb14d89401" +) + +REPO_ROOT = Path(__file__).resolve().parent.parent +DEFAULT_PROFILE_PATH = ( + REPO_ROOT + / "config/metadata-fabric-spark-uncertain-commit-reconciliation.local.yaml" +) +DEFAULT_EVIDENCE_PATH = ( + REPO_ROOT + / "docs/evidence/metadata-fabric-spark-uncertain-commit-reconciliation-2026-07-30.json" +) +DEFAULT_WRAPPER_PATH = ( + REPO_ROOT / "scripts/metadata-fabric-spark-uncertain-commit-reconciliation.sh" +) +DEFAULT_PROBE_PATH = ( + REPO_ROOT + / "k8s/metadata-fabric-spark-uncertain-commit-reconciliation/probe.py" +) +BASE_MANIFEST_DIR = REPO_ROOT / "k8s/metadata-fabric-spark-commit-failure-recovery" + +EXPECTED_OBSERVATION_KEYS = commit_failure.EXPECTED_OBSERVATION_KEYS +PRODUCTION_FALSE_CLAIMS = commit_failure.PRODUCTION_FALSE_CLAIMS + + +class MetadataFabricSparkUncertainCommitReconciliationError(RuntimeError): + """The local uncertain-commit reconciliation contract failed closed.""" + + +class DependencyProfile(object_interop._FrozenModel): + evidence_path: Literal[ + "docs/evidence/metadata-fabric-spark-commit-failure-recovery-2026-07-29.json" + ] + evidence_fingerprint: Literal[DEPENDENCY_EVIDENCE_FINGERPRINT] + required_claim: Literal["local_spark_commit_failure_recovery_verified"] + + +class ClaimProfile(object_interop._FrozenModel): + local_spark_uncertain_commit_reconciliation_verified: Literal[False] + local_provider_committed_response_loss_verified: Literal[False] + local_commit_outcome_readback_verified: Literal[False] + local_duplicate_resubmission_prevented: Literal[False] + local_single_visible_commit_verified: Literal[False] + gravitino_api_metadata_readback_verified: Literal[False] + local_cross_node_object_store_verified: Literal[False] + object_store_metadata_verified: Literal[False] + spark_cancel_verified: Literal[False] + spark_reconcile_verified: Literal[False] + spark_lineage_verified: Literal[False] + persistent_catalog_identity_binding_verified: Literal[False] + protected_workload_identity_verified: Literal[False] + oidc_verified: Literal[False] + tls_verified: Literal[False] + production_object_store_verified: Literal[False] + spark_conformance_verified: Literal[False] + flink_conformance_verified: Literal[False] + production_ingestion_verified: Literal[False] + production_ready: Literal[False] + + +class SparkUncertainCommitReconciliationProfile(object_interop._FrozenModel): + schema_name: Literal[PROFILE_SCHEMA] = Field(alias="schema") + environment: Literal["local_docker_desktop"] + cluster: commit_failure.ClusterProfile + runtime: commit_failure.RuntimeProfile + dependency: DependencyProfile + identity: object_interop.IdentityProfile + catalog: commit_failure.CatalogProfile + scope: commit_failure.ScopeProfile + claims: ClaimProfile + + +def _mapping(value: Any) -> Mapping[str, Any]: + return value if isinstance(value, Mapping) else {} + + +def _list(value: Any) -> list[Any]: + return value if isinstance(value, list) else [] + + +def _load_dependency(profile: SparkUncertainCommitReconciliationProfile) -> None: + path = (REPO_ROOT / profile.dependency.evidence_path).resolve() + try: + path.relative_to(REPO_ROOT) + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: + raise MetadataFabricSparkUncertainCommitReconciliationError( + "Spark commit-failure dependency is unavailable" + ) from exc + if not isinstance(value, dict): + raise MetadataFabricSparkUncertainCommitReconciliationError( + "Spark commit-failure dependency is not an object" + ) + if ( + commit_failure.verify_evidence_integrity(value) + or value.get("evidence_fingerprint") + != profile.dependency.evidence_fingerprint + or value.get(profile.dependency.required_claim) is not True + or any(value.get(claim) is not False for claim in PRODUCTION_FALSE_CLAIMS) + ): + raise MetadataFabricSparkUncertainCommitReconciliationError( + "Spark commit-failure dependency does not match" + ) + + +def load_profile( + path: Path = DEFAULT_PROFILE_PATH, +) -> SparkUncertainCommitReconciliationProfile: + try: + raw = yaml.safe_load(path.resolve().read_text(encoding="utf-8")) + if not isinstance(raw, dict): + raise TypeError("profile must be an object") + ingestion_replay._reject_sensitive_fields(raw) + profile = SparkUncertainCommitReconciliationProfile.model_validate(raw) + except (OSError, TypeError, ValueError, ValidationError, yaml.YAMLError) as exc: + raise MetadataFabricSparkUncertainCommitReconciliationError( + f"Spark uncertain-commit profile is invalid: {type(exc).__name__}" + ) from exc + if ( + object_interop._profile_securable_objects(profile) + != identity._expected_securable_objects() + ): + raise MetadataFabricSparkUncertainCommitReconciliationError( + "Spark uncertain-commit role exceeds the bounded table-create scope" + ) + _load_dependency(profile) + return profile + + +def _validate_probe() -> list[str]: + errors = list(commit_failure._validate_manifest()) + try: + probe = DEFAULT_PROBE_PATH.read_text(encoding="utf-8") + except OSError as exc: + return [ + *errors, + f"Spark uncertain-commit probe is invalid: {type(exc).__name__}", + ] + for marker in ( + "post_forward_success_response_drop_http_504", + "injected response loss after provider commit success", + "committed_do_not_resubmit", + '"write_resubmitted": False', + '"provider_commit_forwarded": True', + '"duplicate_resubmission_prevented": True', + "GDA_SPARK_COMMIT_FAILURE_RESULT", + ): + if marker not in probe: + errors.append(f"Spark uncertain-commit probe is missing marker: {marker}") + if probe.count(".writeTo(TABLE)") != 2: + errors.append("Spark uncertain-commit probe write boundary does not match") + if "AWS_SECRET_ACCESS_KEY\"]" not in probe: + errors.append("Spark uncertain-commit probe credential reference is missing") + return errors + + +def build_contract_report( + profile_path: Path = DEFAULT_PROFILE_PATH, + wrapper_path: Path = DEFAULT_WRAPPER_PATH, +) -> dict[str, Any]: + errors: list[str] = [] + profile: SparkUncertainCommitReconciliationProfile | None = None + try: + profile = load_profile(profile_path) + except MetadataFabricSparkUncertainCommitReconciliationError as exc: + errors.append(str(exc)) + errors.extend(_validate_probe()) + try: + wrapper = wrapper_path.resolve().read_text(encoding="utf-8") + for marker in ( + "set -euo pipefail", + "metadata_fabric_spark_uncertain_commit_reconciliation", + ): + if marker not in wrapper: + errors.append(f"Spark uncertain-commit wrapper is missing: {marker}") + except OSError as exc: + errors.append(f"Spark uncertain-commit wrapper is invalid: {type(exc).__name__}") + + paths = [ + Path(__file__).resolve(), + profile_path.resolve(), + wrapper_path.resolve(), + DEFAULT_PROBE_PATH.resolve(), + Path(commit_failure.__file__).resolve(), + ] + paths.extend(sorted(BASE_MANIFEST_DIR.glob("*.yaml"))) + files: dict[str, dict[str, str]] = {} + for path in paths: + if not path.is_file(): + continue + try: + relative = path.relative_to(REPO_ROOT).as_posix() + except ValueError: + relative = path.name + files[relative] = {"path": relative, "sha256": recovery._file_sha256(path)} + + stable = { + "schema": CONTRACT_SCHEMA, + "context": CONTEXT, + "source_namespace": SOURCE_NAMESPACE, + "rehearsal_namespace": REHEARSAL_NAMESPACE, + "object_store_node": OBJECT_STORE_NODE, + "compute_node": COMPUTE_NODE, + "dependency_evidence_fingerprint": DEPENDENCY_EVIDENCE_FINGERPRINT, + "failure_injection": { + "boundary": "iceberg_rest_table_commit_response", + "mode": "post_forward_success_response_drop_http_504", + "scope": "single_spark_driver_loopback_proxy", + "provider_commit_forwarded": True, + "provider_success_response_delivered": False, + }, + "reconciliation": { + "authority": "table_snapshot_row_and_file_readback", + "committed_decision": "committed_do_not_resubmit", + "write_resubmissions": 0, + }, + "required_invariants": { + "visible_snapshot_delta": 1, + "visible_row_delta": 1, + "visible_data_file_delta": 1, + "provider_commit_forward_count": 1, + "provider_success_response_drop_count": 1, + }, + "runtime_image_identity": { + "gravitino_host_image_id": ( + profile.runtime.gravitino_host_image_id if profile else None + ), + "gravitino_kubernetes_image_id": ( + profile.runtime.gravitino_kubernetes_image_id if profile else None + ), + "postgresql_image_digest": ( + profile.runtime.postgresql_image_digest if profile else None + ), + "spark_host_image_id": ( + profile.runtime.spark_host_image_id if profile else None + ), + "spark_kubernetes_image_id": ( + profile.runtime.spark_kubernetes_image_id if profile else None + ), + "minio_host_image_id": ( + profile.runtime.minio_host_image_id if profile else None + ), + "minio_kubernetes_image_id": ( + profile.runtime.minio_kubernetes_image_id if profile else None + ), + }, + "catalog": { + "warehouse": profile.catalog.warehouse if profile else None, + "io_impl": profile.catalog.io_impl if profile else None, + "s3_endpoint": profile.catalog.s3_endpoint if profile else None, + "bucket": profile.catalog.bucket if profile else None, + "object_prefix": profile.catalog.object_prefix if profile else None, + }, + "local_static_contract_verified": not errors, + "local_spark_uncertain_commit_reconciliation_verified": False, + **{claim: False for claim in PRODUCTION_FALSE_CLAIMS}, + "files": files, + "errors": errors, + } + return {**stable, "contract_fingerprint": recovery._canonical_sha256(stable)} + + +def _spark_errors(spark: Mapping[str, Any]) -> list[str]: + errors: list[str] = [] + job = _mapping(spark.get("job")) + pod = _mapping(spark.get("pod")) + result = _mapping(spark.get("result")) + if ( + spark.get("wait_completed") is not True + or spark.get("terminal_condition") != "Complete" + or job.get("name") != "spark-commit-failure-probe" + or job.get("succeeded") != 1 + or job.get("failed") not in {None, 0} + or pod.get("phase") != "Succeeded" + or spark.get("result_line_count") != 1 + or not object_interop._valid_sha256(spark.get("log_sha256")) + or spark.get("log_recorded") is not False + or spark.get("failure_diagnostic") != [] + ): + errors.append("Spark uncertain-commit Job did not complete exactly once") + if ( + pod.get("node_name") != COMPUTE_NODE + or pod.get("service_account") != "spark-commit-failure-probe" + or pod.get("service_account_automount_disabled") is not True + or pod.get("persistent_volume_claims") != [] + or not str(pod.get("image_id") or "").endswith( + commit_failure.SPARK_KUBERNETES_IMAGE_ID + ) + ): + errors.append("Spark uncertain-commit Pod boundary does not match") + if ( + result.get("schema") + != "gda.spark_uncertain_commit_reconciliation_probe_result.v1" + or result.get("spark_version") != "3.5.0" + or result.get("iceberg_runtime") != "1.6.1" + or result.get("catalog_uri") != "http://127.0.0.1:19001/iceberg" + or result.get("catalog_upstream") + != "http://gravitino-persistence:9001/iceberg" + or result.get("warehouse") != "s3://gda-metadata-warehouse/warehouse" + or result.get("object_store_endpoint") + != "http://metadata-object-store:9000" + or result.get("file_io") != "org.apache.iceberg.aws.s3.S3FileIO" + or result.get("table") + != "rest.published.gda_spark_commit_failure_probe" + or result.get("initial_columns") != ["probe_id"] + or result.get("initial_rows") != [] + or result.get("initial_snapshots") != [] + or result.get("material_recorded") is not False + ): + errors.append("Spark uncertain-commit result envelope does not match") + + baseline = _mapping(result.get("baseline")) + attempted = _mapping(result.get("uncertain_attempt")) + reconciled = _mapping(result.get("reconciliation")) + baseline_snapshots = _list(baseline.get("snapshots")) + reconciled_snapshots = _list(reconciled.get("snapshots")) + baseline_files = _list(baseline.get("data_file_paths")) + reconciled_files = _list(reconciled.get("data_file_paths")) + if ( + baseline.get("rows") != ["spark-baseline-a", "spark-baseline-b"] + or len(baseline_snapshots) != 1 + or _mapping(baseline_snapshots[0]).get("parent_id") is not None + or _mapping(baseline_snapshots[0]).get("operation") != "append" + or len(baseline_files) != 1 + ): + errors.append("Spark uncertain-commit baseline does not match") + if ( + attempted.get("exception_observed") is not True + or not isinstance(attempted.get("exception_type"), str) + or not attempted.get("exception_type") + or attempted.get("logical_row") != "spark-uncertain-commit" + ): + errors.append("Spark uncertain-commit attempt was not observed") + if ( + len(reconciled_snapshots) != 2 + or _mapping(reconciled_snapshots[0]) != _mapping(baseline_snapshots[0]) + or _mapping(reconciled_snapshots[1]).get("parent_id") + != _mapping(reconciled_snapshots[0]).get("snapshot_id") + or [_mapping(item).get("operation") for item in reconciled_snapshots] + != ["append", "append"] + or reconciled.get("rows") + != ["spark-baseline-a", "spark-baseline-b", "spark-uncertain-commit"] + or len(reconciled_files) != 2 + or reconciled.get("decision") != "committed_do_not_resubmit" + or reconciled.get("readback_attempts") != 1 + or reconciled.get("write_resubmitted") is not False + ): + errors.append("Spark uncertain commit was not reconciled from readback") + proxy = _mapping(result.get("proxy")) + if ( + proxy.get("forwarded_commit_requests") != 2 + or proxy.get("uncertain_commit_forwarded_requests") != 1 + or proxy.get("provider_success_responses_dropped") != 1 + or proxy.get("suppressed_duplicate_commit_requests") != 1 + or proxy.get("provider_success_status") != 200 + or not isinstance(proxy.get("total_requests"), int) + or proxy.get("total_requests") < 3 + or proxy.get("injection_mode") + != "post_forward_success_response_drop_http_504" + or proxy.get("provider_commit_forwarded") is not True + or proxy.get("loopback_only") is not True + ): + errors.append("Spark uncertain-commit proxy observation does not match") + if ( + result.get("provider_committed_response_loss_verified") is not True + or result.get("commit_outcome_readback_verified") is not True + or result.get("duplicate_resubmission_prevented") is not True + or result.get("single_visible_commit_verified") is not True + or result.get("object_store_data_files_verified") is not True + ): + errors.append("Spark uncertain-commit local claims do not match") + return errors + + +def _object_store_errors( + prepared: Mapping[str, Any], + store: Mapping[str, Any], + spark: Mapping[str, Any], +) -> list[str]: + errors: list[str] = [] + reconciled = _mapping(_mapping(spark.get("result")).get("reconciliation")) + paths = _list(reconciled.get("data_file_paths")) + expected_data_keys = sorted( + str(path).removeprefix("s3://gda-metadata-warehouse/") for path in paths + ) + data_keys = _list(store.get("data_keys")) + metadata_keys = _list(store.get("metadata_keys")) + manifest_keys = _list(store.get("manifest_keys")) + objects = store.get("objects") + object_items = objects if isinstance(objects, list) else [] + inventory_keys = sorted( + str(_mapping(item).get("key") or "") for item in object_items + ) + categorized_keys = sorted( + [ + *(str(item) for item in data_keys), + *(str(item) for item in metadata_keys), + *(str(item) for item in manifest_keys), + ] + ) + if ( + prepared.get("bucket") != "gda-metadata-warehouse" + or prepared.get("head_bucket_verified") is not True + or prepared.get("path_style_access") is not True + or prepared.get("material_recorded") is not False + or store.get("bucket") != "gda-metadata-warehouse" + or store.get("prefix") + != "warehouse/published/gda_spark_commit_failure_probe/" + or data_keys != expected_data_keys + or len(metadata_keys) != 3 + or len(manifest_keys) != 4 + or store.get("object_count") != 9 + or len(object_items) != 9 + or inventory_keys != categorized_keys + or any( + not str(_mapping(item).get("key") or "").startswith( + "warehouse/published/gda_spark_commit_failure_probe/" + ) + or not isinstance(_mapping(item).get("size"), int) + or _mapping(item).get("size") <= 0 + or not _mapping(item).get("etag") + for item in object_items + ) + ): + errors.append("Uncertain-commit object-store inventory does not match") + latest = _mapping(store.get("latest_metadata")) + snapshots = _list(reconciled.get("snapshots")) + expected_snapshot = ( + _mapping(snapshots[-1]).get("snapshot_id") if snapshots else None + ) + if ( + latest.get("location") + != "s3://gda-metadata-warehouse/warehouse/published/gda_spark_commit_failure_probe" + or latest.get("current_snapshot_id") != expected_snapshot + or latest.get("fields") + != [{"name": "probe_id", "required": True, "type": "string"}] + ): + errors.append("Uncertain-commit Iceberg metadata projection does not match") + return errors + + +def build_evidence(observation: Mapping[str, Any]) -> dict[str, Any]: + errors: list[str] = [] + try: + ingestion_replay._reject_sensitive_fields(observation) + except ValueError: + errors.append("Spark uncertain-commit observation contains sensitive material") + if set(observation) != EXPECTED_OBSERVATION_KEYS: + errors.append("Spark uncertain-commit observation inventory does not match") + if observation.get("schema") != OBSERVATION_SCHEMA: + errors.append("Spark uncertain-commit observation schema does not match") + contract = _mapping(observation.get("contract")) + if ( + contract.get("local_static_contract_verified") is not True + or not object_interop._valid_sha256(contract.get("contract_fingerprint")) + or contract.get("dependency_evidence_fingerprint") + != DEPENDENCY_EVIDENCE_FINGERPRINT + ): + errors.append("Spark uncertain-commit contract binding does not match") + + runtime = _mapping(observation.get("runtime")) + runtime_errors = commit_failure._runtime_errors(runtime) + errors.extend(runtime_errors) + prepared = _mapping(observation.get("object_store_prepared")) + pre_spark = _mapping(observation.get("pre_spark")) + pre_table = _mapping(pre_spark.get("table")) + expected_projection = commit_failure._expected_table_projection() + if ( + _mapping(pre_spark.get("authentication")).get("admin_status") != 200 + or _mapping(pre_spark.get("authentication")).get("bounded_status") != 200 + or _mapping(pre_spark.get("catalog")).get("warehouse") + != "s3://gda-metadata-warehouse/warehouse" + or _mapping(pre_spark.get("catalog")).get("io_impl") + != "org.apache.iceberg.aws.s3.S3FileIO" + or pre_table.get("create_status") != 200 + or pre_table.get("read_status") != 200 + or pre_table.get("projection") != expected_projection + or pre_table.get("fingerprint") + != recovery._canonical_sha256(expected_projection) + or pre_spark.get("denied_catalog_create_status") != 403 + ): + errors.append("Gravitino pre-reconciliation boundary does not match") + + spark = _mapping(observation.get("spark")) + spark_errors = _spark_errors(spark) + errors.extend(spark_errors) + post_spark = _mapping(observation.get("post_spark")) + post_table = _mapping(post_spark.get("table")) + api_readback_verified = ( + post_spark.get("authentication_status") == 200 + and post_spark.get("read_status") == 200 + and post_table.get("projection") == expected_projection + and post_table.get("fingerprint") + == recovery._canonical_sha256(expected_projection) + and post_spark.get("denied_catalog_create_status") == 403 + ) + if not api_readback_verified: + errors.append("Gravitino did not read back the reconciled table") + + store_errors = _object_store_errors( + prepared, _mapping(observation.get("object_store")), spark + ) + errors.extend(store_errors) + runtime_checks = _mapping(observation.get("runtime_checks")) + if ( + runtime_checks.get("namespace_delete_completed") is not True + or runtime_checks.get("namespace_absent") is not True + or runtime_checks.get("persistent_volumes_absent") is not True + or runtime_checks.get("provider_objects_retained") is not False + or runtime_checks.get("object_store_objects_retained") is not False + or runtime_checks.get("all_port_forwards_stopped") is not True + or runtime_checks.get("material_recorded") is not False + or runtime_checks.get("kubernetes_service_account_used_for_provider_login") + is not False + ): + errors.append("Spark uncertain-commit rehearsal cleanup is incomplete") + + result = _mapping(spark.get("result")) + readback_verified = ( + not any("not reconciled" in error.lower() for error in spark_errors) + and result.get("commit_outcome_readback_verified") is True + ) + response_loss_verified = ( + not any("proxy observation" in error.lower() for error in spark_errors) + and result.get("provider_committed_response_loss_verified") is True + ) + duplicate_prevented = ( + readback_verified + and _mapping(result.get("reconciliation")).get("write_resubmitted") is False + and result.get("duplicate_resubmission_prevented") is True + ) + single_visible_commit = ( + duplicate_prevented and result.get("single_visible_commit_verified") is True + ) + cross_node_verified = ( + not runtime_errors + and _mapping(runtime.get("object_store")).get("node_name") + == OBJECT_STORE_NODE + and _mapping(runtime.get("gravitino")).get("node_name") == COMPUTE_NODE + and _mapping(spark.get("pod")).get("node_name") == COMPUTE_NODE + ) + verified = not errors + stable = { + "schema": EVIDENCE_SCHEMA, + "observed_at": observation.get("observed_at"), + "local_static_contract_verified": ( + contract.get("local_static_contract_verified") is True + ), + "local_spark_uncertain_commit_reconciliation_verified": verified, + "local_provider_committed_response_loss_verified": response_loss_verified, + "local_commit_outcome_readback_verified": readback_verified, + "local_duplicate_resubmission_prevented": duplicate_prevented, + "local_single_visible_commit_verified": single_visible_commit, + "gravitino_api_metadata_readback_verified": api_readback_verified, + "local_cross_node_object_store_verified": cross_node_verified, + "object_store_metadata_verified": not store_errors, + **{claim: False for claim in PRODUCTION_FALSE_CLAIMS}, + "observation": dict(observation), + "errors": errors, + } + return {**stable, "evidence_fingerprint": recovery._canonical_sha256(stable)} + + +def verify_evidence_integrity(evidence: Mapping[str, Any]) -> list[str]: + errors: list[str] = [] + rebuilt = build_evidence(_mapping(evidence.get("observation"))) + if evidence.get("evidence_fingerprint") != rebuilt.get("evidence_fingerprint"): + errors.append("Spark uncertain-commit evidence fingerprint does not match") + for key, expected in rebuilt.items(): + if key != "evidence_fingerprint" and evidence.get(key) != expected: + errors.append(f"Spark uncertain-commit evidence field drift: {key}") + for claim in PRODUCTION_FALSE_CLAIMS: + if evidence.get(claim) is not False: + errors.append(f"Spark uncertain-commit evidence may not claim {claim}") + return errors + + +class IsolatedSparkUncertainCommitRuntime( + commit_failure.IsolatedSparkCommitFailureRuntime +): + """Reuse the frozen M3-12 runtime and replace only its suspended probe.""" + + def start( + self, + *, + admin_material: SecretStr, + database_material: SecretStr, + object_store_user: SecretStr, + object_store_material: SecretStr, + ) -> dict[str, Any]: + observed = super().start( + admin_material=admin_material, + database_material=database_material, + object_store_user=object_store_user, + object_store_material=object_store_material, + ) + try: + probe = DEFAULT_PROBE_PATH.read_text(encoding="utf-8") + except OSError as exc: + raise MetadataFabricSparkUncertainCommitReconciliationError( + "Spark uncertain-commit probe is unavailable" + ) from exc + patch = json.dumps({"data": {"probe.py": probe}}, ensure_ascii=True) + self.kubectl.run( + [ + "-n", + self.profile.cluster.rehearsal_namespace, + "patch", + "configmap", + self.profile.runtime.spark_job, + "--type=merge", + "-p", + patch, + ], + label="Spark uncertain-commit suspended probe patch", + ) + return observed + + +def run_live_rehearsal( + profile_path: Path = DEFAULT_PROFILE_PATH, +) -> dict[str, Any]: + profile = load_profile(profile_path) + contract = build_contract_report(profile_path) + if contract.get("local_static_contract_verified") is not True: + raise MetadataFabricSparkUncertainCommitReconciliationError( + "Spark uncertain-commit static contract is invalid" + ) + + admin_material = SecretStr(secrets.token_urlsafe(24)) + database_material = SecretStr(secrets.token_urlsafe(24)) + user_material = SecretStr(secrets.token_urlsafe(24)) + object_store_user = SecretStr("gda" + secrets.token_hex(8)) + object_store_material = SecretStr(secrets.token_urlsafe(32)) + runtime = IsolatedSparkUncertainCommitRuntime(profile) + provider_forward: provider_metrics._PortForward | None = None + object_forward: provider_metrics._PortForward | None = None + rehearsal: object_interop.ObjectStoreCatalogRehearsal | None = None + runtime_observation: dict[str, Any] | None = None + prepared: dict[str, Any] | None = None + pre_spark: dict[str, Any] | None = None + spark: dict[str, Any] | None = None + post_spark: dict[str, Any] | None = None + object_store: dict[str, Any] | None = None + provider_forward_stopped = False + object_forward_stopped = False + cleanup: dict[str, Any] = { + "namespace_delete_completed": False, + "namespace_absent": False, + "persistent_volumes_absent": False, + "provider_objects_retained": True, + "object_store_objects_retained": True, + } + try: + runtime_observation = runtime.start( + admin_material=admin_material, + database_material=database_material, + object_store_user=object_store_user, + object_store_material=object_store_material, + ) + object_forward = provider_metrics._PortForward( + kubectl="kubectl", + context=profile.cluster.context, + namespace=profile.cluster.rehearsal_namespace, + service=profile.runtime.object_store_service, + target_port=profile.runtime.object_store_service_port, + ) + object_forward.start() + object_endpoint = f"http://127.0.0.1:{object_forward.local_port}" + prepared = runtime.prepare_object_store( + endpoint_url=object_endpoint, + object_store_user=object_store_user, + object_store_material=object_store_material, + ) + provider_forward = provider_metrics._PortForward( + kubectl="kubectl", + context=profile.cluster.context, + namespace=profile.cluster.rehearsal_namespace, + service=profile.runtime.service, + target_port=profile.runtime.gravitino_service_port, + ) + provider_forward.start() + rehearsal = object_interop.ObjectStoreCatalogRehearsal( + base_url=f"http://127.0.0.1:{provider_forward.local_port}/api", + admin_name=profile.identity.service_admin, + admin_material=admin_material, + ) + pre_spark = rehearsal.bootstrap( + profile, + database_material=database_material, + user_material=user_material, + object_store_user=object_store_user, + object_store_material=object_store_material, + ) + spark = runtime.run_spark_probe() + post_spark = spark_interop._post_spark_readback( + rehearsal, profile, user_material + ) + object_store = runtime.observe_object_store( + endpoint_url=object_endpoint, + object_store_user=object_store_user, + object_store_material=object_store_material, + ) + finally: + if rehearsal is not None: + rehearsal.close() + if provider_forward is not None: + provider_forward_stopped = provider_forward.stop() + if object_forward is not None: + object_forward_stopped = object_forward.stop() + cleanup = runtime.cleanup() + + if any( + value is None + for value in ( + runtime_observation, + prepared, + pre_spark, + spark, + post_spark, + object_store, + ) + ): + raise MetadataFabricSparkUncertainCommitReconciliationError( + "Spark uncertain-commit rehearsal did not produce an outcome" + ) + observation = { + "schema": OBSERVATION_SCHEMA, + "observed_at": datetime.now(UTC).isoformat(), + "contract": { + "contract_fingerprint": contract["contract_fingerprint"], + "local_static_contract_verified": True, + "dependency_evidence_fingerprint": DEPENDENCY_EVIDENCE_FINGERPRINT, + }, + "runtime": runtime_observation, + "object_store_prepared": prepared, + "pre_spark": pre_spark, + "spark": spark, + "post_spark": post_spark, + "object_store": object_store, + "runtime_checks": { + **cleanup, + "all_port_forwards_stopped": ( + provider_forward_stopped and object_forward_stopped + ), + "material_recorded": False, + "kubernetes_service_account_used_for_provider_login": False, + }, + } + return build_evidence(observation) + + +def build_validation_report( + *, + profile_path: Path = DEFAULT_PROFILE_PATH, + evidence_path: Path = DEFAULT_EVIDENCE_PATH, +) -> dict[str, Any]: + contract = build_contract_report(profile_path) + errors = list(contract["errors"]) + evidence: dict[str, Any] | None = None + try: + value = json.loads(evidence_path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise TypeError("evidence must be an object") + evidence = value + errors.extend(verify_evidence_integrity(evidence)) + observed_contract = _mapping( + _mapping(evidence.get("observation")).get("contract") + ).get("contract_fingerprint") + if observed_contract != contract.get("contract_fingerprint"): + errors.append("Spark uncertain-commit evidence contract fingerprint drift") + except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: + errors.append(f"Spark uncertain-commit evidence is invalid: {type(exc).__name__}") + verified = not errors + local_claims = ( + "local_spark_uncertain_commit_reconciliation_verified", + "local_provider_committed_response_loss_verified", + "local_commit_outcome_readback_verified", + "local_duplicate_resubmission_prevented", + "local_single_visible_commit_verified", + "gravitino_api_metadata_readback_verified", + "local_cross_node_object_store_verified", + "object_store_metadata_verified", + ) + return { + "schema": VALIDATION_SCHEMA, + "local_static_contract_verified": contract["local_static_contract_verified"], + **{ + claim: ( + verified and evidence is not None and evidence.get(claim) is True + ) + for claim in local_claims + }, + **{claim: False for claim in PRODUCTION_FALSE_CLAIMS}, + "contract_fingerprint": contract["contract_fingerprint"], + "evidence_fingerprint": ( + evidence.get("evidence_fingerprint") 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=True, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + validate = subparsers.add_parser("validate") + validate.add_argument("--profile", type=Path, default=DEFAULT_PROFILE_PATH) + validate.add_argument("--evidence", type=Path, default=DEFAULT_EVIDENCE_PATH) + rehearse = subparsers.add_parser("rehearse") + rehearse.add_argument("--profile", type=Path, default=DEFAULT_PROFILE_PATH) + rehearse.add_argument("--evidence-out", type=Path, required=True) + verify = subparsers.add_parser("verify") + verify.add_argument("--evidence", type=Path, default=DEFAULT_EVIDENCE_PATH) + args = parser.parse_args(argv) + try: + if args.command == "validate": + report = build_validation_report( + profile_path=args.profile, evidence_path=args.evidence + ) + print(json.dumps(report, ensure_ascii=True, indent=2, sort_keys=True)) + return 0 if not report["errors"] else 1 + if args.command == "verify": + value = json.loads(args.evidence.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise TypeError("evidence must be an object") + errors = verify_evidence_integrity(value) + print(json.dumps({"verified": not errors, "errors": errors}, indent=2)) + return 0 if not errors else 1 + evidence = run_live_rehearsal(args.profile) + _write_json(args.evidence_out, evidence) + print(json.dumps(evidence, ensure_ascii=True, indent=2, sort_keys=True)) + return 0 if not evidence["errors"] else 1 + except ( + BotoCoreError, + ClientError, + KeyError, + OSError, + TypeError, + ValueError, + json.JSONDecodeError, + identity.MetadataFabricGravitinoIdentityError, + jdbc_restart.MetadataFabricGravitinoJdbcRestartError, + object_interop.MetadataFabricSparkObjectStoreInteroperabilityError, + commit_failure.MetadataFabricSparkCommitFailureRecoveryError, + MetadataFabricSparkUncertainCommitReconciliationError, + KeyboardInterrupt, + ) as exc: + print(f"metadata fabric Spark uncertain-commit reconciliation: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/data_agent/platform_truth.py b/data_agent/platform_truth.py index 664a0b6f..b7837e30 100644 --- a/data_agent/platform_truth.py +++ b/data_agent/platform_truth.py @@ -728,6 +728,26 @@ def _config( ), "Protected identity/TLS, production object storage and full Spark/Flink conformance", ), + RuntimeSpec( + "metadata_spark_uncertain_commit_reconciliation_rehearsal", + "spark_uncertain_commit_reconciliation_rehearsal", + "governed", + "evidence_durable", + "committed local Spark/Iceberg uncertain-commit reconciliation evidence", + "metadata-platform", + "local_verification_only", + ( + "data_agent/metadata_fabric_spark_uncertain_commit_reconciliation.py", + "scripts/metadata-fabric-spark-uncertain-commit-reconciliation.sh", + ), + ( + ( + "data_agent/metadata_fabric_spark_uncertain_commit_reconciliation.py", + "class IsolatedSparkUncertainCommitRuntime", + ), + ), + "Protected identity/TLS, production object storage and full Spark/Flink conformance", + ), RuntimeSpec( "datalake_monitor", "monitor_loop", diff --git a/data_agent/test_metadata_fabric_spark_uncertain_commit_reconciliation.py b/data_agent/test_metadata_fabric_spark_uncertain_commit_reconciliation.py new file mode 100644 index 00000000..38e61c1a --- /dev/null +++ b/data_agent/test_metadata_fabric_spark_uncertain_commit_reconciliation.py @@ -0,0 +1,354 @@ +import json +from copy import deepcopy +from pathlib import Path +from types import SimpleNamespace + +import pytest +import yaml +from pydantic import SecretStr + +from data_agent import metadata_fabric_spark_commit_failure_recovery as dependency +from data_agent import metadata_fabric_spark_uncertain_commit_reconciliation as reconcile + + +def _dependency_observation() -> dict: + evidence = json.loads(dependency.DEFAULT_EVIDENCE_PATH.read_text(encoding="utf-8")) + return deepcopy(evidence["observation"]) + + +def _synthetic_observation() -> dict: + observation = _dependency_observation() + observation["schema"] = reconcile.OBSERVATION_SCHEMA + observation["contract"] = { + "contract_fingerprint": reconcile.build_contract_report()[ + "contract_fingerprint" + ], + "local_static_contract_verified": True, + "dependency_evidence_fingerprint": ( + reconcile.DEPENDENCY_EVIDENCE_FINGERPRINT + ), + } + old = observation["spark"]["result"] + final = deepcopy(old["retry"]) + final["rows"] = [ + "spark-baseline-a", + "spark-baseline-b", + "spark-uncertain-commit", + ] + observation["spark"]["result"] = { + "schema": "gda.spark_uncertain_commit_reconciliation_probe_result.v1", + "spark_version": old["spark_version"], + "iceberg_runtime": old["iceberg_runtime"], + "catalog_uri": old["catalog_uri"], + "catalog_upstream": old["catalog_upstream"], + "warehouse": old["warehouse"], + "object_store_endpoint": old["object_store_endpoint"], + "file_io": old["file_io"], + "table": old["table"], + "initial_columns": old["initial_columns"], + "initial_rows": old["initial_rows"], + "initial_snapshots": old["initial_snapshots"], + "baseline": deepcopy(old["baseline"]), + "uncertain_attempt": { + "exception_observed": True, + "exception_type": "Py4JJavaError", + "logical_row": "spark-uncertain-commit", + }, + "reconciliation": { + "decision": "committed_do_not_resubmit", + "readback_attempts": 1, + "write_resubmitted": False, + **final, + }, + "proxy": { + "forwarded_commit_requests": 2, + "uncertain_commit_forwarded_requests": 1, + "provider_success_responses_dropped": 1, + "suppressed_duplicate_commit_requests": 1, + "provider_success_status": 200, + "total_requests": 10, + "injection_mode": "post_forward_success_response_drop_http_504", + "provider_commit_forwarded": True, + "loopback_only": True, + }, + "provider_committed_response_loss_verified": True, + "commit_outcome_readback_verified": True, + "duplicate_resubmission_prevented": True, + "single_visible_commit_verified": True, + "object_store_data_files_verified": True, + "material_recorded": False, + } + return observation + + +def _write_profile(tmp_path: Path, value: dict) -> Path: + path = tmp_path / "profile.yaml" + path.write_text(yaml.safe_dump(value, sort_keys=False), encoding="utf-8") + return path + + +def test_checked_contract_and_evidence_verify_uncertain_commit_reconciliation(): + contract = reconcile.build_contract_report() + validation = reconcile.build_validation_report() + evidence = json.loads(reconcile.DEFAULT_EVIDENCE_PATH.read_text(encoding="utf-8")) + + assert contract["contract_fingerprint"] == ( + "7a8d75a1d6b4558b982c6c3242d8d356c5046955f8aae7a45e5c297b6f4d4132" + ) + assert reconcile.verify_evidence_integrity(evidence) == [] + assert evidence["evidence_fingerprint"] == ( + "d6462fff78d07047311b1f715d5f2c7f08c0ce8fbdd5c8b26a3d95ddc3474786" + ) + assert validation["errors"] == [] + assert validation["local_spark_uncertain_commit_reconciliation_verified"] is True + assert validation["local_provider_committed_response_loss_verified"] is True + assert validation["local_commit_outcome_readback_verified"] is True + assert validation["local_duplicate_resubmission_prevented"] is True + assert validation["local_single_visible_commit_verified"] is True + for claim in reconcile.PRODUCTION_FALSE_CLAIMS: + assert validation[claim] is False + + +def test_checked_evidence_contains_response_loss_readback_and_object_proof(): + evidence = json.loads(reconcile.DEFAULT_EVIDENCE_PATH.read_text(encoding="utf-8")) + observation = evidence["observation"] + result = observation["spark"]["result"] + baseline = result["baseline"] + reconciled = result["reconciliation"] + proxy = result["proxy"] + store = observation["object_store"] + + assert proxy["uncertain_commit_forwarded_requests"] == 1 + assert proxy["provider_success_responses_dropped"] == 1 + assert proxy["suppressed_duplicate_commit_requests"] == 1 + assert proxy["provider_success_status"] == 200 + assert reconciled["decision"] == "committed_do_not_resubmit" + assert reconciled["write_resubmitted"] is False + assert len(reconciled["snapshots"]) == len(baseline["snapshots"]) + 1 + assert reconciled["snapshots"][1]["parent_id"] == ( + reconciled["snapshots"][0]["snapshot_id"] + ) + assert reconciled["rows"] == [ + "spark-baseline-a", + "spark-baseline-b", + "spark-uncertain-commit", + ] + assert store["object_count"] == 9 + assert len(store["data_keys"]) == 2 + assert len(store["metadata_keys"]) == 3 + assert len(store["manifest_keys"]) == 4 + + +def test_static_contract_binds_post_forward_reconciliation(): + contract = reconcile.build_contract_report() + + assert contract["errors"] == [] + assert contract["local_static_contract_verified"] is True + assert contract["failure_injection"] == { + "boundary": "iceberg_rest_table_commit_response", + "mode": "post_forward_success_response_drop_http_504", + "scope": "single_spark_driver_loopback_proxy", + "provider_commit_forwarded": True, + "provider_success_response_delivered": False, + } + assert contract["reconciliation"]["write_resubmissions"] == 0 + assert contract["dependency_evidence_fingerprint"] == ( + dependency.build_validation_report()["evidence_fingerprint"] + ) + for claim in reconcile.PRODUCTION_FALSE_CLAIMS: + assert contract[claim] is False + + +def test_synthetic_observation_proves_one_commit_without_resubmission(): + evidence = reconcile.build_evidence(_synthetic_observation()) + + assert evidence["errors"] == [] + assert evidence["local_spark_uncertain_commit_reconciliation_verified"] is True + assert evidence["local_provider_committed_response_loss_verified"] is True + assert evidence["local_commit_outcome_readback_verified"] is True + assert evidence["local_duplicate_resubmission_prevented"] is True + assert evidence["local_single_visible_commit_verified"] is True + for claim in reconcile.PRODUCTION_FALSE_CLAIMS: + assert evidence[claim] is False + + +@pytest.mark.parametrize( + ("mutate", "expected"), + [ + ( + lambda value: value["spark"]["result"]["reconciliation"].update( + {"write_resubmitted": True} + ), + "not reconciled from readback", + ), + ( + lambda value: value["spark"]["result"]["reconciliation"].update( + {"decision": "retry_write"} + ), + "not reconciled from readback", + ), + ( + lambda value: value["spark"]["result"]["reconciliation"][ + "rows" + ].append("spark-uncertain-commit"), + "not reconciled from readback", + ), + ( + lambda value: value["spark"]["result"]["reconciliation"][ + "snapshots" + ][1].update({"parent_id": 99}), + "not reconciled from readback", + ), + ( + lambda value: value["spark"]["result"]["proxy"].update( + {"provider_success_responses_dropped": 0} + ), + "proxy observation does not match", + ), + ( + lambda value: value["spark"]["result"]["proxy"].update( + {"provider_commit_forwarded": False} + ), + "proxy observation does not match", + ), + ( + lambda value: value["spark"]["result"].update( + {"duplicate_resubmission_prevented": False} + ), + "local claims do not match", + ), + ], +) +def test_evidence_rejects_uncertain_outcome_or_duplicate_drift(mutate, expected): + observation = _synthetic_observation() + mutate(observation) + + evidence = reconcile.build_evidence(observation) + + assert any(expected in error for error in evidence["errors"]) + assert evidence["local_spark_uncertain_commit_reconciliation_verified"] is False + + +@pytest.mark.parametrize( + "mutate", + [ + lambda value: value["spark"]["result"]["reconciliation"].update( + {"rows": 1} + ), + lambda value: value["spark"]["result"]["reconciliation"].update( + {"snapshots": 1} + ), + lambda value: value["spark"]["result"]["reconciliation"].update( + {"data_file_paths": 1} + ), + lambda value: value["object_store"].update({"objects": 1}), + lambda value: value["object_store"].update({"data_keys": 1}), + lambda value: value["object_store"].update({"metadata_keys": 1}), + lambda value: value["object_store"].update({"manifest_keys": 1}), + ], +) +def test_evidence_fails_closed_for_malformed_collection_fields(mutate): + observation = _synthetic_observation() + mutate(observation) + + evidence = reconcile.build_evidence(observation) + + assert evidence["errors"] + assert evidence["local_spark_uncertain_commit_reconciliation_verified"] is False + + +def test_profile_rejects_sensitive_material_and_dependency_drift( + tmp_path, monkeypatch +): + profile = yaml.safe_load(reconcile.DEFAULT_PROFILE_PATH.read_text(encoding="utf-8")) + profile["catalog"]["secret_access_key"] = "must-not-enter-profile" + with pytest.raises( + reconcile.MetadataFabricSparkUncertainCommitReconciliationError, + match="profile is invalid", + ): + reconcile.load_profile(_write_profile(tmp_path, profile)) + + profile = yaml.safe_load(reconcile.DEFAULT_PROFILE_PATH.read_text(encoding="utf-8")) + dependency_value = json.loads( + dependency.DEFAULT_EVIDENCE_PATH.read_text(encoding="utf-8") + ) + dependency_value["production_ready"] = True + dependency_path = ( + tmp_path + / "docs/evidence/metadata-fabric-spark-commit-failure-recovery-2026-07-29.json" + ) + dependency_path.parent.mkdir(parents=True) + dependency_path.write_text(json.dumps(dependency_value), encoding="utf-8") + monkeypatch.setattr(reconcile, "REPO_ROOT", tmp_path) + with pytest.raises( + reconcile.MetadataFabricSparkUncertainCommitReconciliationError, + match="dependency does not match", + ): + reconcile.load_profile(_write_profile(tmp_path, profile)) + + +def test_evidence_integrity_rejects_sensitive_or_production_claims(): + observation = _synthetic_observation() + observation["runtime_secret"] = "must-not-enter-evidence" + evidence = reconcile.build_evidence(observation) + assert any("contains sensitive material" in error for error in evidence["errors"]) + + checked = reconcile.build_evidence(_synthetic_observation()) + forged = deepcopy(checked) + forged["production_ready"] = True + assert reconcile.verify_evidence_integrity(forged) + assert any( + "may not claim production_ready" in error + for error in reconcile.verify_evidence_integrity(forged) + ) + + +def test_runtime_replaces_suspended_probe_before_release(monkeypatch): + profile = reconcile.load_profile() + calls = [] + + monkeypatch.setattr( + dependency.IsolatedSparkCommitFailureRuntime, + "start", + lambda _self, **_kwargs: {"runtime": "observed"}, + ) + runtime = reconcile.IsolatedSparkUncertainCommitRuntime(profile) + + class FakeKubectl: + def run(self, args, **kwargs): + calls.append((args, kwargs)) + return SimpleNamespace(stdout="") + + runtime.kubectl = FakeKubectl() + observed = runtime.start( + admin_material=SecretStr("admin"), + database_material=SecretStr("database"), + object_store_user=SecretStr("object-user"), + object_store_material=SecretStr("object-material"), + ) + + assert observed == {"runtime": "observed"} + assert len(calls) == 1 + args, _kwargs = calls[0] + assert args[2:5] == ["patch", "configmap", profile.runtime.spark_job] + payload = json.loads(args[args.index("-p") + 1]) + assert "committed_do_not_resubmit" in payload["data"]["probe.py"] + + +def test_probe_and_wrapper_are_fail_closed(): + probe = reconcile.DEFAULT_PROBE_PATH.read_text(encoding="utf-8") + wrapper = reconcile.DEFAULT_WRAPPER_PATH.read_text(encoding="utf-8") + + assert probe.count(".writeTo(TABLE)") == 2 + assert "post_forward_success_response_drop_http_504" in probe + assert '"write_resubmitted": False' in probe + assert "set -euo pipefail" in wrapper + assert "metadata_fabric_spark_uncertain_commit_reconciliation" in wrapper + + +def test_validation_without_checked_evidence_fails_closed(tmp_path): + report = reconcile.build_validation_report(evidence_path=tmp_path / "missing.json") + + assert report["errors"] + assert report["local_spark_uncertain_commit_reconciliation_verified"] is False + assert report["spark_reconcile_verified"] is False diff --git a/data_agent/test_platform_truth.py b/data_agent/test_platform_truth.py index 829e34dd..7ff050d6 100644 --- a/data_agent/test_platform_truth.py +++ b/data_agent/test_platform_truth.py @@ -236,6 +236,12 @@ 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_spark_uncertain_commit_reconciliation_rehearsal" + 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-059-local-spark-uncertain-commit-reconciliation.md b/docs/architecture-decisions/adr-059-local-spark-uncertain-commit-reconciliation.md new file mode 100644 index 00000000..526cd65b --- /dev/null +++ b/docs/architecture-decisions/adr-059-local-spark-uncertain-commit-reconciliation.md @@ -0,0 +1,34 @@ +# ADR-059: Local Spark uncertain-commit reconciliation + +- Status: Accepted +- Date: 2026-07-30 +- Scope: local Docker Desktop evidence only +- Depends on: ADR-058 / M3-12 checked evidence + +## Context + +ADR-058 proves that a known pre-forward HTTP 503 changes no visible table state and that one explicit retry produces one commit. It does not cover the harder outcome where Gravitino commits successfully but Spark does not receive the success response. Blindly submitting the logical write again can create a duplicate snapshot and row. + +Iceberg `1.6.1` maps table-commit HTTP `500`, `502`, and `504` responses to `CommitStateUnknownException`. HTTP `503` is a generic service failure instead; a discarded local attempt confirmed that it can trigger staged-file cleanup after the provider has committed. That attempt was cleaned up and is not retained as accepted evidence. + +## Decision + +Reuse the fingerprinted M3-12 PostgreSQL, Gravitino, MinIO, identity, namespace, and suspended Spark Job runtime without modifying its files. Before releasing the Job, replace only its ConfigMap probe with the M3-13 probe. + +The Spark driver loopback proxy forwards one armed table commit to Gravitino. After the provider returns HTTP 200, the proxy drops that response and returns HTTP 504. One transport retry is suppressed with the same unknown-state response. Spark then performs read-only table refresh and verifies the intended row, one child append snapshot, and one additional referenced Parquet. The decision is `committed_do_not_resubmit`; no second logical write is issued. + +## Evidence + +- Baseline: 1 append snapshot, 2 rows, 1 referenced Parquet. +- Uncertain commit: exactly 1 provider forward, provider status 200, exactly 1 dropped success response, and 1 suppressed transport retry. +- Reconciliation: 2 parent-linked append snapshots, 3 rows including `spark-uncertain-commit`, 2 referenced Parquet, `readback_attempts=1`, and `write_resubmitted=false`. +- Direct MinIO inventory: 2 data files, 3 metadata JSON files, and 4 Avro manifest files; 9 objects total. +- Gravitino bounded API readback remains valid; catalog creation remains denied with 403. +- Spark Job completed `1/1`; namespace, both persistent volumes, and port-forwards were removed. +- Contract fingerprint: `7a8d75a1d6b4558b982c6c3242d8d356c5046955f8aae7a45e5c297b6f4d4132` +- Evidence fingerprint: `d6462fff78d07047311b1f715d5f2c7f08c0ce8fbdd5c8b26a3d95ddc3474786` +- Dependency evidence fingerprint: `39571cdac1e4043bcfc2d03a73b2b12ff925210daf8ae36bc640b8cb14d89401` + +## Consequences + +This proves a deterministic local post-forward commit-state-unknown path and a no-resubmit readback decision for one append. It does not provide a durable production reconciliation controller, operation key ledger, crash recovery between exception and readback, concurrent-writer proof, arbitrary mutation conformance, network exactly-once, protected identity/TLS, production object storage, cancel/lineage, Flink, full Spark conformance, production ingestion, or production readiness. The broad `spark_reconcile_verified` and all production claims remain `false`. diff --git a/docs/evidence/metadata-fabric-spark-uncertain-commit-reconciliation-2026-07-30.json b/docs/evidence/metadata-fabric-spark-uncertain-commit-reconciliation-2026-07-30.json new file mode 100644 index 00000000..4920804f --- /dev/null +++ b/docs/evidence/metadata-fabric-spark-uncertain-commit-reconciliation-2026-07-30.json @@ -0,0 +1,437 @@ +{ + "errors": [], + "evidence_fingerprint": "d6462fff78d07047311b1f715d5f2c7f08c0ce8fbdd5c8b26a3d95ddc3474786", + "flink_conformance_verified": false, + "gravitino_api_metadata_readback_verified": true, + "local_commit_outcome_readback_verified": true, + "local_cross_node_object_store_verified": true, + "local_duplicate_resubmission_prevented": true, + "local_provider_committed_response_loss_verified": true, + "local_single_visible_commit_verified": true, + "local_spark_uncertain_commit_reconciliation_verified": true, + "local_static_contract_verified": true, + "object_store_metadata_verified": true, + "observation": { + "contract": { + "contract_fingerprint": "7a8d75a1d6b4558b982c6c3242d8d356c5046955f8aae7a45e5c297b6f4d4132", + "dependency_evidence_fingerprint": "39571cdac1e4043bcfc2d03a73b2b12ff925210daf8ae36bc640b8cb14d89401", + "local_static_contract_verified": true + }, + "object_store": { + "bucket": "gda-metadata-warehouse", + "data_keys": [ + "warehouse/published/gda_spark_commit_failure_probe/data/00000-10-87c2c26d-b266-4473-941f-e81341e71f64-0-00001.parquet", + "warehouse/published/gda_spark_commit_failure_probe/data/00000-2-dd1f8e8c-f3a3-4fca-964c-f864e465d92b-0-00001.parquet" + ], + "latest_metadata": { + "current_schema_id": 0, + "current_snapshot_id": 8106441515576740703, + "fields": [ + { + "name": "probe_id", + "required": true, + "type": "string" + } + ], + "key": "warehouse/published/gda_spark_commit_failure_probe/metadata/00002-f0753421-b20e-49b5-bde5-8073192c10d0.metadata.json", + "location": "s3://gda-metadata-warehouse/warehouse/published/gda_spark_commit_failure_probe" + }, + "manifest_keys": [ + "warehouse/published/gda_spark_commit_failure_probe/metadata/2459b7a2-bdf0-40ce-a2e1-cd508ad11d7f-m0.avro", + "warehouse/published/gda_spark_commit_failure_probe/metadata/4bad994a-d4c5-4823-86ee-cbf8f6fea07f-m0.avro", + "warehouse/published/gda_spark_commit_failure_probe/metadata/snap-8106441515576740703-1-4bad994a-d4c5-4823-86ee-cbf8f6fea07f.avro", + "warehouse/published/gda_spark_commit_failure_probe/metadata/snap-8586350311299081622-1-2459b7a2-bdf0-40ce-a2e1-cd508ad11d7f.avro" + ], + "material_recorded": false, + "metadata_keys": [ + "warehouse/published/gda_spark_commit_failure_probe/metadata/00000-37074b26-5b18-4e41-ad39-a6789a9b14a7.metadata.json", + "warehouse/published/gda_spark_commit_failure_probe/metadata/00001-cdccb127-cb3c-412d-a830-82a924fd746a.metadata.json", + "warehouse/published/gda_spark_commit_failure_probe/metadata/00002-f0753421-b20e-49b5-bde5-8073192c10d0.metadata.json" + ], + "object_count": 9, + "objects": [ + { + "etag": "f9ed6709d7d80764aa0a78791024540c", + "key": "warehouse/published/gda_spark_commit_failure_probe/data/00000-10-87c2c26d-b266-4473-941f-e81341e71f64-0-00001.parquet", + "size": 586 + }, + { + "etag": "b95c2838ca10b35fe904dc1a0ce1e9a6", + "key": "warehouse/published/gda_spark_commit_failure_probe/data/00000-2-dd1f8e8c-f3a3-4fca-964c-f864e465d92b-0-00001.parquet", + "size": 515 + }, + { + "etag": "1f00e5cd84e52b5e15d96e8e78aaa56c", + "key": "warehouse/published/gda_spark_commit_failure_probe/metadata/00000-37074b26-5b18-4e41-ad39-a6789a9b14a7.metadata.json", + "size": 941 + }, + { + "etag": "45d670490b42944cf5616b08b1cad1f8", + "key": "warehouse/published/gda_spark_commit_failure_probe/metadata/00001-cdccb127-cb3c-412d-a830-82a924fd746a.metadata.json", + "size": 2034 + }, + { + "etag": "1faf311755e8289873f172c2ed58c993", + "key": "warehouse/published/gda_spark_commit_failure_probe/metadata/00002-f0753421-b20e-49b5-bde5-8073192c10d0.metadata.json", + "size": 3097 + }, + { + "etag": "e9b8501608ca9c8acbb98868fb078fdd", + "key": "warehouse/published/gda_spark_commit_failure_probe/metadata/2459b7a2-bdf0-40ce-a2e1-cd508ad11d7f-m0.avro", + "size": 6689 + }, + { + "etag": "3962940d4aa2932de4f428702b775c0c", + "key": "warehouse/published/gda_spark_commit_failure_probe/metadata/4bad994a-d4c5-4823-86ee-cbf8f6fea07f-m0.avro", + "size": 6692 + }, + { + "etag": "3d70de7985b5d4ab0fe52115425a7342", + "key": "warehouse/published/gda_spark_commit_failure_probe/metadata/snap-8106441515576740703-1-4bad994a-d4c5-4823-86ee-cbf8f6fea07f.avro", + "size": 4546 + }, + { + "etag": "945a5febec42f06b1c90064a53b57614", + "key": "warehouse/published/gda_spark_commit_failure_probe/metadata/snap-8586350311299081622-1-2459b7a2-bdf0-40ce-a2e1-cd508ad11d7f.avro", + "size": 4474 + } + ], + "prefix": "warehouse/published/gda_spark_commit_failure_probe/" + }, + "object_store_prepared": { + "bucket": "gda-metadata-warehouse", + "created": true, + "head_bucket_verified": true, + "material_recorded": false, + "path_style_access": true, + "region": "us-east-1", + "service": "metadata-object-store" + }, + "observed_at": "2026-07-29T16:48:47.823116+00:00", + "post_spark": { + "authentication_status": 200, + "denied_catalog_create_status": 403, + "read_status": 200, + "table": { + "fingerprint": "ffe896cf5439c074f642dc7c6f9e6817134ac60b04e5b467d4d896b408022244", + "name": "gda_spark_commit_failure_probe", + "projection": { + "columns": [ + { + "name": "probe_id", + "nullable": false, + "type": "string" + } + ], + "name": "gda_spark_commit_failure_probe", + "probe_property": "true" + } + } + }, + "pre_spark": { + "authentication": { + "admin_status": 200, + "bounded_status": 200, + "material_recorded": false + }, + "catalog": { + "backend": "jdbc", + "catalog": "lakehouse", + "io_impl": "org.apache.iceberg.aws.s3.S3FileIO", + "jdbc_initialize": true, + "material_recorded": false, + "metalake": "gda_commit_failure", + "provider": "lakehouse-iceberg", + "s3_endpoint": "http://metadata-object-store:9000", + "s3_path_style_access": true, + "s3_region": "us-east-1", + "schema": "published", + "uri": "jdbc:postgresql://gravitino-persistence-postgresql:5432/iceberg", + "warehouse": "s3://gda-metadata-warehouse/warehouse" + }, + "denied_catalog_create_status": 403, + "role": { + "name": "gda-table-projection", + "securable_objects": [ + { + "fullName": "lakehouse", + "privileges": [ + { + "condition": "ALLOW", + "name": "USE_CATALOG" + } + ], + "type": "CATALOG" + }, + { + "fullName": "lakehouse.published", + "privileges": [ + { + "condition": "ALLOW", + "name": "CREATE_TABLE" + }, + { + "condition": "ALLOW", + "name": "USE_SCHEMA" + } + ], + "type": "SCHEMA" + } + ] + }, + "table": { + "create_status": 200, + "fingerprint": "ffe896cf5439c074f642dc7c6f9e6817134ac60b04e5b467d4d896b408022244", + "name": "gda_spark_commit_failure_probe", + "projection": { + "columns": [ + { + "name": "probe_id", + "nullable": false, + "type": "string" + } + ], + "name": "gda_spark_commit_failure_probe", + "probe_property": "true" + }, + "read_status": 200 + } + }, + "runtime": { + "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": "2c08ade9-4201-4bc9-8ea2-809d6f358df6", + "pvc": null, + "ready_replicas": 1, + "service_account": "gravitino-persistence", + "service_account_automount_disabled": true, + "statefulset_uid": "f99646dc-86d4-406a-8ad2-024b56482e2a" + }, + "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-commit-failure", + "uid": "fd53c94e-8287-457c-9591-cc6a820897d1" + }, + "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": "66306854-79b2-4ff3-a518-6fdf5185aac3", + "pvc": { + "name": "data-metadata-object-store-0", + "phase": "Bound", + "storage_class": "standard", + "uid": "0b962d69-72f7-4790-ac02-f723b5340c17", + "volume_name": "pvc-0b962d69-72f7-4790-ac02-f723b5340c17" + }, + "ready_replicas": 1, + "service_account": "metadata-object-store", + "service_account_automount_disabled": true, + "statefulset_uid": "eba9b1c3-ada2-4493-b7b9-a278de012887" + }, + "object_store_service": { + "name": "metadata-object-store", + "ports": [ + { + "name": "api", + "port": 9000 + } + ], + "type": "ClusterIP", + "uid": "45300871-3f3f-4b17-b50d-2369da70a5b0" + }, + "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": "672cb8b8-2ce8-40cb-8559-fe564cd7327d", + "pvc": { + "name": "data-gravitino-persistence-postgresql-0", + "phase": "Bound", + "storage_class": "standard", + "uid": "fed55abd-1ce2-46cb-ba7c-62f3cdb77617", + "volume_name": "pvc-fed55abd-1ce2-46cb-ba7c-62f3cdb77617" + }, + "ready_replicas": 1, + "service_account": "gravitino-persistence-postgresql", + "service_account_automount_disabled": true, + "statefulset_uid": "c5f308ad-b641-4629-9940-c30bd97fe33e" + }, + "service": { + "name": "gravitino-persistence", + "ports": [ + { + "name": "http", + "port": 8090 + }, + { + "name": "iceberg-rest", + "port": 9001 + } + ], + "type": "ClusterIP", + "uid": "9a68da70-b652-4942-ae1d-1f7348e29d05" + }, + "source_schema_sha256": "7a2d605a677a462ca619dba594ce7ebcf500358345560ad084c1b67a25c722df", + "spark_host_image_id": "sha256:f201367640c7583add224796a629150e63d3859ddd7fe9fd47741662a6d415bb" + }, + "runtime_checks": { + "all_port_forwards_stopped": true, + "kubernetes_service_account_used_for_provider_login": false, + "material_recorded": false, + "namespace_absent": true, + "namespace_delete_completed": true, + "object_store_objects_retained": false, + "persistent_volume_names": [ + "pvc-0b962d69-72f7-4790-ac02-f723b5340c17", + "pvc-fed55abd-1ce2-46cb-ba7c-62f3cdb77617" + ], + "persistent_volumes_absent": true, + "provider_objects_retained": false + }, + "schema": "gda.metadata_fabric_spark_uncertain_commit_reconciliation_observation.v1", + "spark": { + "failure_diagnostic": [], + "job": { + "completion_time": "2026-07-29T16:47:29Z", + "failed": 0, + "name": "spark-commit-failure-probe", + "succeeded": 1, + "uid": "b172d5ed-7e41-4e19-abec-6ba4e7dbacd8" + }, + "log_recorded": false, + "log_sha256": "879a391a2cf653ebffe0149bd7e9e367c3dd22fc47d14b8400f845536194bf2c", + "pod": { + "image": "docker.io/gisdataagent/mmfe-spark-runtime:local", + "image_id": "sha256:4a4522bfd4e6d1c6c90a244d0145841fbfbbf21ed16ee29ca8b681b5cec60058", + "name": "spark-commit-failure-probe-4znvq", + "node_name": "desktop-worker", + "persistent_volume_claims": [], + "phase": "Succeeded", + "service_account": "spark-commit-failure-probe", + "service_account_automount_disabled": true, + "uid": "eb71f5a4-e461-46db-91bb-7491e4bf7c26" + }, + "result": { + "baseline": { + "data_file_paths": [ + "s3://gda-metadata-warehouse/warehouse/published/gda_spark_commit_failure_probe/data/00000-2-dd1f8e8c-f3a3-4fca-964c-f864e465d92b-0-00001.parquet" + ], + "rows": [ + "spark-baseline-a", + "spark-baseline-b" + ], + "snapshots": [ + { + "operation": "append", + "parent_id": null, + "snapshot_id": 8586350311299081622 + } + ] + }, + "catalog_upstream": "http://gravitino-persistence:9001/iceberg", + "catalog_uri": "http://127.0.0.1:19001/iceberg", + "commit_outcome_readback_verified": true, + "duplicate_resubmission_prevented": true, + "file_io": "org.apache.iceberg.aws.s3.S3FileIO", + "iceberg_runtime": "1.6.1", + "initial_columns": [ + "probe_id" + ], + "initial_rows": [], + "initial_snapshots": [], + "material_recorded": false, + "object_store_data_files_verified": true, + "object_store_endpoint": "http://metadata-object-store:9000", + "provider_committed_response_loss_verified": true, + "proxy": { + "forwarded_commit_requests": 2, + "injection_mode": "post_forward_success_response_drop_http_504", + "loopback_only": true, + "provider_commit_forwarded": true, + "provider_success_responses_dropped": 1, + "provider_success_status": 200, + "suppressed_duplicate_commit_requests": 1, + "total_requests": 36, + "uncertain_commit_forwarded_requests": 1 + }, + "reconciliation": { + "data_file_paths": [ + "s3://gda-metadata-warehouse/warehouse/published/gda_spark_commit_failure_probe/data/00000-10-87c2c26d-b266-4473-941f-e81341e71f64-0-00001.parquet", + "s3://gda-metadata-warehouse/warehouse/published/gda_spark_commit_failure_probe/data/00000-2-dd1f8e8c-f3a3-4fca-964c-f864e465d92b-0-00001.parquet" + ], + "decision": "committed_do_not_resubmit", + "readback_attempts": 1, + "rows": [ + "spark-baseline-a", + "spark-baseline-b", + "spark-uncertain-commit" + ], + "snapshots": [ + { + "operation": "append", + "parent_id": null, + "snapshot_id": 8586350311299081622 + }, + { + "operation": "append", + "parent_id": 8586350311299081622, + "snapshot_id": 8106441515576740703 + } + ], + "write_resubmitted": false + }, + "schema": "gda.spark_uncertain_commit_reconciliation_probe_result.v1", + "single_visible_commit_verified": true, + "spark_version": "3.5.0", + "table": "rest.published.gda_spark_commit_failure_probe", + "uncertain_attempt": { + "exception_observed": true, + "exception_type": "Py4JJavaError", + "logical_row": "spark-uncertain-commit" + }, + "warehouse": "s3://gda-metadata-warehouse/warehouse" + }, + "result_line_count": 1, + "terminal_condition": "Complete", + "wait_completed": true + } + }, + "observed_at": "2026-07-29T16:48:47.823116+00:00", + "oidc_verified": false, + "persistent_catalog_identity_binding_verified": false, + "production_ingestion_verified": false, + "production_object_store_verified": false, + "production_ready": false, + "protected_workload_identity_verified": false, + "schema": "gda.metadata_fabric_spark_uncertain_commit_reconciliation_evidence.v1", + "spark_cancel_verified": false, + "spark_conformance_verified": false, + "spark_lineage_verified": false, + "spark_reconcile_verified": false, + "tls_verified": false +} diff --git a/docs/roadmap-ar0-platform-truth-2026-07-24.md b/docs/roadmap-ar0-platform-truth-2026-07-24.md index efcac48e..f88babbc 100644 --- a/docs/roadmap-ar0-platform-truth-2026-07-24.md +++ b/docs/roadmap-ar0-platform-truth-2026-07-24.md @@ -199,7 +199,7 @@ Temporal 继续保持目标组件状态，不在这一包并行接入。OpenMeta 当前完成仅指本地合同、授权 evidence、outbox/callback 代码、数据库成功终局门、托管 worker 代码、默认关闭的部署模板及离线 activation/release preflight、candidate/registry/provenance/artifact-release/live observation evidence gate、合成 golden slice、定向测试、真实 PostgreSQL 16 事务边界和 canonical mainline 治理。`candidate_validated`、`registry_subject_bound`、本地合成 `provenance_verified`、`ready_for_activation`、`ready_for_staging_apply`、`verified_for_staging_apply` 和本地 live collection 都不等于真实镜像已 attested 或 staging 已部署；真实 IAM/OIDC 与 service token 生命周期、首次 GHCR publish/verify、真实 provenance artifact verify、registry-backed live staging revision、worker/callback 扩容运行、golden slice staging 运行链、受保护 release/live evidence provenance、独立 DolphinScheduler metadata PostgreSQL 和真实数据终局证据仍属于 4.7 后续切片。 -### 4.8 Metadata Fabric Bridge M1 + M2 + M3-12（本地 Spark commit-failure recovery 已验证，生产验证待执行） +### 4.8 Metadata Fabric Bridge M1 + M2 + M3-13（本地 Spark uncertain-commit reconcile 已验证，生产验证待执行） 第八块回到 AR-1 的 metadata control plane，以 [ADR-036](architecture-decisions/adr-036-read-only-metadata-fabric-bridge-contract.md) 固定 OpenMetadata + Gravitino + GDA Control Ledger 的首条 table slice： @@ -232,8 +232,9 @@ Temporal 继续保持目标组件状态，不在这一包并行接入。OpenMeta 27. [ADR-056](architecture-decisions/adr-056-local-spark-object-store-interoperability.md) 已移除 Spark/Gravitino 的共享 warehouse PVC：MinIO 在 `desktop-control-plane`，PostgreSQL、Gravitino 和 Spark 在 `desktop-worker`，两端只经 S3-compatible ClusterIP 共享 `s3://gda-metadata-warehouse/warehouse`。Spark 保持两次 append、schema evolution、三行 current state、双 snapshot 和 first-snapshot time travel；Gravitino 回读演进 schema，直接 MinIO 检查确认 2 个 Parquet、4 个 metadata JSON 和 4 个 Avro manifest，并匹配 table location、schema 与 current snapshot。contract fingerprint 为 `9713cdb3040e1b6532489f329aef7ed7b5266e0757551252f537cb83476b4bee`，evidence fingerprint 为 `05844457efb378581fb7fc2e7ed3c706819b2d8fa5a52b2f82577051d38c2cd1`。该结果只证明同一 Docker Desktop 主机/集群内的跨节点 MinIO 互操作，不证明生产云对象存储、protected identity/TLS、故障注入、cancel/reconcile/lineage、Flink 或完整 Spark conformance。 28. [ADR-057](architecture-decisions/adr-057-production-object-store-readiness-gate.md) 已将 M3-10 evidence、S3-compatible provider/account/region/bucket、独立 failure domain、OIDC workload federation、精确八项 S3 permission、TLS/private path、KMS、versioning、cross-region replication、strong read/list consistency、tenant isolation、owner/SLO/runbook 和 26 项 protected attestation check 冻结为 fail-closed profile。当前 profile fingerprint 为 `668e194b3c688307014148391e7f389c9d6e9ca69c95d7b4cc92b4acae93181a`，report fingerprint 为 `85362dd10b7dc565f9fa567673d90b774cdec714bd1e70fb2c3c83c1af48b5ea`，合同有效但 43 项生产输入仍 blocked，全部 production claims 为 `false`。这只是 provider-neutral 决策和验收合同：没有选择、部署或验证 AWS S3、华为云 OBS 或其他生产对象存储；原生非 S3 provider 必须进入新的 conformance slice。 29. [ADR-058](architecture-decisions/adr-058-local-spark-commit-failure-recovery.md) 已在 Spark driver 的 loopback Iceberg REST proxy 中于 provider 转发前注入 HTTP 503。baseline 为 1 个 append snapshot、2 行和 1 个 referenced Parquet；失败调用经过精确 2 次 503 后，snapshot/row/file 均零漂移；对同一 `spark-recovery` 行做一次显式重试后为父子相连的 2 个 append snapshots、3 行和 2 个 referenced Parquet。直接 MinIO inventory 精确为 2 data + 3 metadata + 4 manifest = 9 objects，没有孤儿 data file；namespace、两块 PV 和 port-forward 均清理。contract fingerprint 为 `6d8944ab80246dc65891aa81118cb8b73f7ecad699be9a2af5e62d8260c41002`，evidence fingerprint 为 `39571cdac1e4043bcfc2d03a73b2b12ff925210daf8ae36bc640b8cb14d89401`。该结果只证明已知 pre-forward 失败下的本地原子性和一次显式重试，不证明 uncertain commit reconciliation、网络 exactly-once、生产对象存储或完整 engine conformance。 +30. [ADR-059](architecture-decisions/adr-059-local-spark-uncertain-commit-reconciliation.md) 已将一个 armed commit 转发给 Gravitino，并在 provider 返回 200 后丢弃成功响应、向 Spark 返回 Iceberg `CommitStateUnknownException` 所需的 HTTP 504；一次传输重试被抑制。Spark 不重提逻辑写，而是 readback 得到父子相连的 2 个 append snapshots、3 行和 2 个 referenced Parquet，决策为 `committed_do_not_resubmit`、`write_resubmitted=false`。MinIO inventory 为 2 data + 3 metadata + 4 manifest = 9 objects；Job `Complete 1/1`，namespace、两块 PV 和 port-forward 均清理。contract fingerprint 为 `7a8d75a1d6b4558b982c6c3242d8d356c5046955f8aae7a45e5c297b6f4d4132`，evidence fingerprint 为 `d6462fff78d07047311b1f715d5f2c7f08c0ce8fbdd5c8b26a3d95ddc3474786`。该结果只证明一个本地 append 的确定性 readback/no-resubmit，不证明持久 reconcile controller、并发写、进程崩溃恢复、网络 exactly-once 或生产能力。 -此处 M1 只证明静态合同和只读 HTTP 边界；M2a 只证明本地 live foundation 与 PVC 重挂载连续性；M2b-1/M2b-2 分别限定在同集群新 PVC 和同集群隔离 repository；M2b-3 的 `local_cross_cluster_recovery_verified=true` 只限定在 `local_same_host_distinct_kubernetes_clusters_external_s3_repository`；M2c-1/M2c-2/M2c-3 分别限定本地 provider metrics、临时双周期 OTel 和单 job scrape recovery；M2c-4/M2d-2 只证明 production observability/NetworkPolicy profile 与 attestation 合同可校验；M2d-1 只证明本地两节点 kindnet 的隔离合成流量；M3-1 的 terminal evidence 与 M3-2 的 PolicyDecision/Approval 仍是 deterministic local fixtures。M3-2 只把 projection 写入本地 provider 并证明 retained target 的单次零写入 replay；M3-3 只把该本地 evidence 对应的 binding 写入临时 GDA Control 账本；M3-4 只向无认证 loopback receiver 发送精确 candidate 并验证 503 后幂等恢复；M3-5 只证明 OpenMetadata 在 provider 强制默认 role 之上的项目新增 grant 限定为 `table/Create`，以及本地 JWT 轮换/吊销和越权拒绝；M3-6 只证明隔离 Gravitino Basic IdP 的 bounded table-create、catalog-create 拒绝、登录轮换/吊销和完整清理；M3-7 只证明 pending production identity profile、profile-bound attestation 和派生 claim 的 fail-closed 合同可校验，没有部署或证明真实身份路径；M3-8 只证明同一 Docker Desktop 集群内 Basic 用户、JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明同节点共享 RWO PVC 的 Spark interoperability；M3-10 移除了该共享 PVC，并证明同一 Docker Desktop 主机/集群内 Spark 与 MinIO 的跨节点 S3-compatible 互操作，但不证明生产云对象存储、独立 failure domain、持久 identity binding、Flink 或完整 engine conformance；M3-11 只冻结 provider-neutral production object-store profile、精确 attestation binding 与 fail-closed claims，没有选择 provider、部署 bucket/KMS/policy 或提交真实 attestation；M3-12 只证明同一本地路径的 pre-forward commit failure 不改变可见 table state，随后一次显式重试产生一个新 snapshot/row，且无孤儿 data file，不覆盖 provider 已提交但响应丢失的 uncertain outcome。M3-2 ingestion 仍使用 bootstrap admin，生产持久 binding、ResourceVersion 和 legacy authority 都未写入；生产对象存储、双 provider/生产最小权限、protected workload identity、OIDC、TLS、生产持久 catalog、tenant isolation、真实 receiver/alert/SLO、受保护 provider policy、生产故障注入、source-loss recovery、cancel/reconcile/lineage、完整 Spark/Flink conformance、生产 ingest、四项 production gate 和 `production_ready` 仍为 `false`。 +此处 M1 只证明静态合同和只读 HTTP 边界；M2a 只证明本地 live foundation 与 PVC 重挂载连续性；M2b-1/M2b-2 分别限定在同集群新 PVC 和同集群隔离 repository；M2b-3 的 `local_cross_cluster_recovery_verified=true` 只限定在 `local_same_host_distinct_kubernetes_clusters_external_s3_repository`；M2c-1/M2c-2/M2c-3 分别限定本地 provider metrics、临时双周期 OTel 和单 job scrape recovery；M2c-4/M2d-2 只证明 production observability/NetworkPolicy profile 与 attestation 合同可校验；M2d-1 只证明本地两节点 kindnet 的隔离合成流量；M3-1 的 terminal evidence 与 M3-2 的 PolicyDecision/Approval 仍是 deterministic local fixtures。M3-2 只把 projection 写入本地 provider 并证明 retained target 的单次零写入 replay；M3-3 只把该本地 evidence 对应的 binding 写入临时 GDA Control 账本；M3-4 只向无认证 loopback receiver 发送精确 candidate 并验证 503 后幂等恢复；M3-5 只证明 OpenMetadata 在 provider 强制默认 role 之上的项目新增 grant 限定为 `table/Create`，以及本地 JWT 轮换/吊销和越权拒绝；M3-6 只证明隔离 Gravitino Basic IdP 的 bounded table-create、catalog-create 拒绝、登录轮换/吊销和完整清理；M3-7 只证明 pending production identity profile、profile-bound attestation 和派生 claim 的 fail-closed 合同可校验，没有部署或证明真实身份路径；M3-8 只证明同一 Docker Desktop 集群内 Basic 用户、JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明同节点共享 RWO PVC 的 Spark interoperability；M3-10 移除了该共享 PVC，并证明同一 Docker Desktop 主机/集群内 Spark 与 MinIO 的跨节点 S3-compatible 互操作，但不证明生产云对象存储、独立 failure domain、持久 identity binding、Flink 或完整 engine conformance；M3-11 只冻结 provider-neutral production object-store profile、精确 attestation binding 与 fail-closed claims，没有选择 provider、部署 bucket/KMS/policy 或提交真实 attestation；M3-12 只证明同一本地路径的 pre-forward commit failure 不改变可见 table state，随后一次显式重试产生一个新 snapshot/row，且无孤儿 data file；M3-13 只证明单次本地 append 在 provider 200 响应丢失并映射为 commit-state-unknown 后，可以由即时 table readback 判定 committed 且不重提，不覆盖持久 controller、进程崩溃、并发写或任意 mutation。M3-2 ingestion 仍使用 bootstrap admin，生产持久 binding、ResourceVersion 和 legacy authority 都未写入；生产对象存储、双 provider/生产最小权限、protected workload identity、OIDC、TLS、生产持久 catalog、tenant isolation、真实 receiver/alert/SLO、受保护 provider policy、生产故障注入、source-loss recovery、cancel/reconcile/lineage、完整 Spark/Flink conformance、生产 ingest、四项 production gate 和 `production_ready` 仍为 `false`。 ## 5. 重新评估条件 diff --git a/docs/system-of-record-matrix-2026-07-24.md b/docs/system-of-record-matrix-2026-07-24.md index ab0e3f6a..199c1c3c 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-30 -阶段：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-2、M3-3 local binding ledger、M3-4 local OpenLineage wire delivery、M3-5 local OpenMetadata bounded identity、M3-6 local Gravitino Basic bounded identity、M3-7 production identity readiness contract、M3-8 local Gravitino JDBC restart continuity、M3-9 local Spark/Iceberg REST interoperability、M3-10 local cross-node Spark/object-store interoperability、M3-11 production object-store readiness contract 与 M3-12 local Spark commit-failure recovery 已验证，生产 provider ingestion、生产观测、生产 policy/tenant isolation、生产 identity/object-store attestation 和生产切换仍 `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-2、M3-3 local binding ledger、M3-4 local OpenLineage wire delivery、M3-5 local OpenMetadata bounded identity、M3-6 local Gravitino Basic bounded identity、M3-7 production identity readiness contract、M3-8 local Gravitino JDBC restart continuity、M3-9 local Spark/Iceberg REST interoperability、M3-10 local cross-node Spark/object-store interoperability、M3-11 production object-store readiness contract、M3-12 local Spark commit-failure recovery 与 M3-13 local uncertain-commit reconciliation 已验证，生产 provider ingestion、生产观测、生产 policy/tenant isolation、生产 identity/object-store attestation 和生产切换仍 `in_progress` -适用分支：`feat/ar1-metadata-fabric-spark-commit-failure-recovery` +适用分支：`feat/ar1-metadata-fabric-spark-uncertain-commit-reconciliation` ## 判定规则 @@ -56,7 +56,7 @@ 11. `platform_command_outbox` 只拥有投递状态；callback 只触发 reconcile，不能把 provider payload 直接写成 PlatformRun 状态或平台终局。 12. QualityResult evaluator 必须是 workload，且成功终局中的 evaluator 不能等于 Run workload；该代码级职责分离不替代生产 IAM。 13. `candidate_validated`、`registry_subject_bound`、GitHub provenance action 成功、CI artifact、离线 preflight、未独立 attested 的 live observation JSON 或人工批准都不能单独授权 production；缺少同一 source revision 的 OCI subject 独立验证、registry/live revision/identity/health/golden-slice 绑定及受保护 provenance 时，promotion 必须失败。 -14. Metadata Fabric M1 只允许 OpenMetadata/Gravitino GET；M2 只执行本地 foundation/recovery/metrics/policy 演练或验证 production readiness profile；M3-1 只从 synthetic terminal evidence 生成 plan/candidate；M3-2 只允许 exact local PolicyDecision/Approval 后向本地 provider 写 projection；M3-3 只将同一 source evidence 经 PlatformGateway 写入临时 append-only binding ledger；M3-4 只经 tenant-scoped outbox 向无认证 loopback receiver 投递精确 candidate，并验证 at-least-once + receiver idempotency；M3-5 只证明临时 OpenMetadata bot 在 provider 强制 `DefaultBotRole` 之上的项目新增 grant 是 `table/Create`，并验证 policy-create 拒绝与本地 JWT 轮换/吊销；M3-6 只证明隔离 Gravitino Basic user 的 bounded table-create、catalog-create 拒绝、密码轮换/用户吊销和完整清理；M3-7 只冻结 production identity profile、精确 attestation binding 与 fail-closed 派生 claims，既不部署 identity path，也不提交真实 production attestation；M3-8 只证明 Docker Desktop 单集群中 Basic role、PostgreSQL JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明同节点共享 RWO PVC 的 Spark/Iceberg REST 互操作；M3-10 移除 Spark/Gravitino 共享 warehouse PVC，且只证明同一 Docker Desktop 主机/集群内跨节点 MinIO 的 read/write/schema evolution/snapshot/time travel 与对象级 metadata 一致；M3-11 只冻结 S3-compatible production profile、精确 attestation binding 与 fail-closed claims，既不选择/部署 provider，也不创建 bucket/KMS/policy 或提交真实 attestation；M3-12 只在 Spark driver loopback proxy 中于转发前注入 503，证明失败尝试零可见漂移、随后一次显式重试和直接对象清单无孤儿 data file，不覆盖 provider 已提交但响应丢失的 uncertain outcome。生产对象存储、identity/TLS、受保护环境故障注入、source-loss recovery、cancel/reconcile/lineage 与 Flink 仍未证明。Gravitino `1.3.0` Basic IdP 不算 OIDC，生产必须明确选择并证明 custom OIDC authenticator 或 identity-aware proxy。本地 bootstrap provisioner、Basic IdP、loopback/cluster HTTP、memory/file-backed JDBC catalog、同节点 PVC/MinIO、临时 identity/ledger/outbox、loopback receiver、pending profile、合成 attestation 和 local evidence 都不等于双 provider/生产最小权限、protected workload identity/OIDC、生产持久 catalog/binding、TLS、受保护 OpenLineage receiver、tenant isolation、alert/SLO、生产 ingestion/conformance 或生产写权威。 +14. Metadata Fabric M1 只允许 OpenMetadata/Gravitino GET；M2 只执行本地 foundation/recovery/metrics/policy 演练或验证 production readiness profile；M3-1 只从 synthetic terminal evidence 生成 plan/candidate；M3-2 只允许 exact local PolicyDecision/Approval 后向本地 provider 写 projection；M3-3 只将同一 source evidence 经 PlatformGateway 写入临时 append-only binding ledger；M3-4 只经 tenant-scoped outbox 向无认证 loopback receiver 投递精确 candidate，并验证 at-least-once + receiver idempotency；M3-5 只证明临时 OpenMetadata bot 在 provider 强制 `DefaultBotRole` 之上的项目新增 grant 是 `table/Create`，并验证 policy-create 拒绝与本地 JWT 轮换/吊销；M3-6 只证明隔离 Gravitino Basic user 的 bounded table-create、catalog-create 拒绝、密码轮换/用户吊销和完整清理；M3-7 只冻结 production identity profile、精确 attestation binding 与 fail-closed 派生 claims，既不部署 identity path，也不提交真实 production attestation；M3-8 只证明 Docker Desktop 单集群中 Basic role、PostgreSQL JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明同节点共享 RWO PVC 的 Spark/Iceberg REST 互操作；M3-10 移除 Spark/Gravitino 共享 warehouse PVC，且只证明同一 Docker Desktop 主机/集群内跨节点 MinIO 的 read/write/schema evolution/snapshot/time travel 与对象级 metadata 一致；M3-11 只冻结 S3-compatible production profile、精确 attestation binding 与 fail-closed claims，既不选择/部署 provider，也不创建 bucket/KMS/policy 或提交真实 attestation；M3-12 只在 Spark driver loopback proxy 中于转发前注入 503，证明失败尝试零可见漂移、随后一次显式重试和直接对象清单无孤儿 data file；M3-13 只在 provider 200 响应丢失后以 HTTP 504 触发 commit-state-unknown，并对一个本地 append 做即时 readback/no-resubmit，不构成持久生产 reconcile controller、crash/concurrency proof 或网络 exactly-once。生产对象存储、identity/TLS、受保护环境故障注入、source-loss recovery、cancel/reconcile/lineage 与 Flink 仍未证明。Gravitino `1.3.0` Basic IdP 不算 OIDC，生产必须明确选择并证明 custom OIDC authenticator 或 identity-aware proxy。本地 bootstrap provisioner、Basic IdP、loopback/cluster HTTP、memory/file-backed JDBC catalog、同节点 PVC/MinIO、临时 identity/ledger/outbox、loopback receiver、pending profile、合成 attestation 和 local evidence 都不等于双 provider/生产最小权限、protected workload identity/OIDC、生产持久 catalog/binding、TLS、受保护 OpenLineage receiver、tenant isolation、alert/SLO、生产 ingestion/conformance 或生产写权威。 ## 已建立的 AR-0/AR-1 entry 证据 @@ -92,6 +92,7 @@ - Metadata Fabric M3-10 已将 warehouse 切换为 MinIO S3 API：MinIO 位于 `desktop-control-plane`，PostgreSQL、Gravitino 与 Spark 位于 `desktop-worker`，Spark/Gravitino 均无 warehouse PVC。Spark 保持两次 append、nullable `quality` 演进、三行 current state、双 snapshot 与 first-snapshot time travel；Gravitino 回读演进 schema，直接 MinIO 检查确认 2 个 Parquet、4 个 metadata JSON、4 个 Avro manifest，最新 location/schema/current snapshot 与 Spark 一致，catalog create 前后仍为 403，namespace、两块 PV 和 port-forward 均清理。contract fingerprint 为 `9713cdb3040e1b6532489f329aef7ed7b5266e0757551252f537cb83476b4bee`，evidence fingerprint 为 `05844457efb378581fb7fc2e7ed3c706819b2d8fa5a52b2f82577051d38c2cd1`。这只证明本地同主机/同集群跨节点 S3-compatible interoperability；生产云对象存储、protected identity/OIDC/TLS、故障注入、cancel/reconcile/lineage、Flink、完整 `spark_conformance_verified`、生产 ingestion 与 `production_ready` 仍未证明。 - Metadata Fabric M3-11 已建立 production object-store profile/attestation gate；checked-in profile fingerprint 为 `668e194b3c688307014148391e7f389c9d6e9ca69c95d7b4cc92b4acae93181a`，report fingerprint 为 `85362dd10b7dc565f9fa567673d90b774cdec714bd1e70fb2c3c83c1af48b5ea`，`profile_valid=true`，43 项 provider/identity/transport/encryption/durability/consistency/tenancy/operations 外部输入以 blockers 暴露，`ready_for_protected_verification=false`、`attestation_valid=false`、`production_object_store_gate_passed=false`、`production_ready=false`。该合同绑定 M3-10 evidence，但没有选择或部署 provider；合成完整 attestation 只验证门禁逻辑，不计入生产证据。下一项真实证据是经 owner 批准并物化的 provider profile，以及来自 `production-object-store` 受保护环境、绑定当前 source/profile 并通过全部 26 项检查的 attestation。 - Metadata Fabric M3-12 已在本地 Spark driver loopback Iceberg REST proxy 中于 provider 转发前注入 HTTP 503。baseline 为 1 个 append snapshot、2 行和 1 个 referenced Parquet；失败写经 2 次 503 后 snapshot/row/file 零漂移；对同一逻辑行一次显式重试后为父子相连的 2 个 append snapshots、3 行和 2 个 referenced Parquet。直接 MinIO inventory 精确为 2 data + 3 metadata + 4 manifest = 9 objects，没有孤儿 data file；namespace、两块 PV 和 port-forward 均清理。contract fingerprint 为 `6d8944ab80246dc65891aa81118cb8b73f7ecad699be9a2af5e62d8260c41002`，evidence fingerprint 为 `39571cdac1e4043bcfc2d03a73b2b12ff925210daf8ae36bc640b8cb14d89401`。这只证明已知 pre-forward failure 的本地原子性与一次显式重试，不证明 provider uncertain outcome reconcile、网络 exactly-once、生产对象存储、cancel/lineage、Flink 或完整 Spark conformance。 +- Metadata Fabric M3-13 已在本地 Spark driver loopback proxy 将 armed commit 转发给 Gravitino，并在 provider 200 后丢弃成功响应、返回 HTTP 504；Iceberg 将其映射为 commit-state-unknown，一次传输重试被抑制。Spark 只读 readback 后输出 `committed_do_not_resubmit` 和 `write_resubmitted=false`，最终为父子相连的 2 个 append snapshots、3 行、2 个 referenced Parquet；MinIO 为 2 data + 3 metadata + 4 manifest = 9 objects。Job `Complete 1/1`，namespace、两块 PV 和 port-forward 均清理。contract fingerprint 为 `7a8d75a1d6b4558b982c6c3242d8d356c5046955f8aae7a45e5c297b6f4d4132`，evidence fingerprint 为 `d6462fff78d07047311b1f715d5f2c7f08c0ce8fbdd5c8b26a3d95ddc3474786`。这只证明一个本地 append 的确定性 readback/no-resubmit；持久 controller、crash/concurrency、网络 exactly-once、生产对象存储、cancel/lineage、Flink 和完整 Spark conformance 仍未证明。 ## 下一验收证据 @@ -100,5 +101,5 @@ - staging 的 migration role、应用 login membership、连接池 role/tenant 复位、双租户 API 和 success finalization 运行产物； - DolphinScheduler adapter 的真实 IAM/OIDC、service token provisioning/轮换、provider 最小权限、binding artifact staging 接入、managed outbox worker/provider callback 实际扩容部署、唯一 worker ID、status/lease 故障恢复和无双写证据； - 首条真实图斑链对 golden slice 的 output hash、独立质量结果/evidence、血缘、发布 revision 和 rollback 演练； -- OpenMetadata/Gravitino 的 source host/cluster 外生产 backup account/bucket、已批准 provider profile 与受保护对象存储 attestation、生产对象存储、KMS/TLS/workload identity、PITR/source-loss recovery、RPO/RTO、OIDC、受保护环境 provider NetworkPolicy/tenant isolation、upgrade/rollback、registry provenance、持续 metrics backend/retention/query、真实 alert delivery/SLO owner/runbook，以及受保护 PolicyDecision/Approval、双 provider 最小权限 ingestion、生产持久 binding、受保护 production OpenLineage receiver、无双写 read-back、受保护环境 commit failure injection、provider uncertain outcome reconcile、cancel/reconcile/lineage 和完整 Spark/Flink conformance；M1 fixture、M2 本地 evidence/readiness contracts、M3-1 projection candidate、M3-2 local replay、M3-3 临时 binding ledger、M3-4 loopback delivery、M3-5/M3-6 本地临时 provider identity、M3-7 pending profile/合成 attestation、M3-8 本地 JDBC restart continuity、M3-9 本地同节点 Spark interoperability、M3-10 本地同主机跨节点 MinIO interoperability、M3-11 pending object-store profile/合成 attestation 与 M3-12 local pre-forward commit-failure recovery 均不计入生产退出门； +- OpenMetadata/Gravitino 的 source host/cluster 外生产 backup account/bucket、已批准 provider profile 与受保护对象存储 attestation、生产对象存储、KMS/TLS/workload identity、PITR/source-loss recovery、RPO/RTO、OIDC、受保护环境 provider NetworkPolicy/tenant isolation、upgrade/rollback、registry provenance、持续 metrics backend/retention/query、真实 alert delivery/SLO owner/runbook，以及受保护 PolicyDecision/Approval、双 provider 最小权限 ingestion、生产持久 binding、受保护 production OpenLineage receiver、无双写 read-back、受保护环境 commit failure injection、provider uncertain outcome reconcile、cancel/reconcile/lineage 和完整 Spark/Flink conformance；M1 fixture、M2 本地 evidence/readiness contracts、M3-1 projection candidate、M3-2 local replay、M3-3 临时 binding ledger、M3-4 loopback delivery、M3-5/M3-6 本地临时 provider identity、M3-7 pending profile/合成 attestation、M3-8 本地 JDBC restart continuity、M3-9 本地同节点 Spark interoperability、M3-10 本地同主机跨节点 MinIO interoperability、M3-11 pending object-store profile/合成 attestation、M3-12 local pre-forward commit-failure recovery 与 M3-13 local uncertain-commit readback/no-resubmit 均不计入生产退出门； - DolphinScheduler/Temporal sandbox 的独立数据库、备份恢复、身份、版本和升级责任证明；DolphinScheduler standalone/H2 不计入此退出门。 diff --git a/k8s/metadata-fabric-spark-uncertain-commit-reconciliation/probe.py b/k8s/metadata-fabric-spark-uncertain-commit-reconciliation/probe.py new file mode 100644 index 00000000..42f87931 --- /dev/null +++ b/k8s/metadata-fabric-spark-uncertain-commit-reconciliation/probe.py @@ -0,0 +1,368 @@ +import json +import os +import threading +import urllib.error +import urllib.request +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +from pyspark.sql import SparkSession + + +TABLE = "rest.published.gda_spark_commit_failure_probe" +UPSTREAM = "http://gravitino-persistence:9001" +PROXY = "http://127.0.0.1:19001" +WAREHOUSE = "s3://gda-metadata-warehouse/warehouse" +OBJECT_STORE_ENDPOINT = "http://metadata-object-store:9000" +DATA_PREFIX = WAREHOUSE + "/published/gda_spark_commit_failure_probe/data/" +BASELINE_ROWS = ["spark-baseline-a", "spark-baseline-b"] +UNCERTAIN_ROW = "spark-uncertain-commit" + + +class ProxyState: + def __init__(self): + self.lock = threading.Lock() + self.response_drop_armed = False + self.provider_success_responses_dropped = 0 + self.uncertain_commit_forwarded_requests = 0 + self.suppressed_duplicate_commit_requests = 0 + self.forwarded_commit_requests = 0 + self.total_requests = 0 + self.provider_success_status = None + + +state = ProxyState() + + +def is_table_commit(path, body): + if "/tables/" not in path: + return False + try: + payload = json.loads(body or b"{}") + except json.JSONDecodeError: + return False + return isinstance(payload.get("requirements"), list) and isinstance( + payload.get("updates"), list + ) + + +def commit_state_unknown(message): + return json.dumps( + { + "error": { + "message": message, + "type": "ServiceFailureException", + "code": 504, + } + }, + sort_keys=True, + ).encode("utf-8") + + +class UncertainCommitProxy(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_GET(self): + self.forward() + + def do_HEAD(self): + self.forward() + + def do_POST(self): + self.forward() + + def do_DELETE(self): + self.forward() + + def send_payload(self, status, body, content_type=None): + self.send_response(status) + if content_type: + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(body))) + self.send_header("Connection", "close") + self.end_headers() + if self.command != "HEAD": + self.wfile.write(body) + + def forward(self): + length = int(self.headers.get("Content-Length", "0")) + body = self.rfile.read(length) if length else b"" + commit = self.command == "POST" and is_table_commit(self.path, body) + with state.lock: + state.total_requests += 1 + suppress_duplicate = ( + commit + and state.response_drop_armed + and state.provider_success_responses_dropped == 1 + ) + if suppress_duplicate: + state.suppressed_duplicate_commit_requests += 1 + if suppress_duplicate: + payload = commit_state_unknown( + "uncertain commit response remains unavailable; reconcile before resubmit" + ) + self.send_payload(504, payload, "application/json") + return + + headers = { + key: value + for key, value in self.headers.items() + if key.lower() + not in { + "connection", + "content-length", + "host", + "proxy-connection", + "transfer-encoding", + } + } + request = urllib.request.Request( + UPSTREAM + self.path, + data=body if self.command in {"POST", "PUT", "PATCH"} else None, + headers=headers, + method=self.command, + ) + try: + response = urllib.request.urlopen(request, timeout=30) + status = response.status + response_body = response.read() + content_type = response.headers.get("Content-Type") + except urllib.error.HTTPError as exc: + status = exc.code + response_body = exc.read() + content_type = exc.headers.get("Content-Type") + except urllib.error.URLError: + status = 502 + response_body = json.dumps( + { + "error": { + "message": "catalog upstream unavailable", + "type": "ServiceFailureException", + "code": 502, + } + }, + sort_keys=True, + ).encode("utf-8") + content_type = "application/json" + + drop_success = False + if commit: + with state.lock: + state.forwarded_commit_requests += 1 + drop_success = ( + state.response_drop_armed + and 200 <= status < 300 + and state.provider_success_responses_dropped == 0 + ) + if drop_success: + state.provider_success_responses_dropped = 1 + state.uncertain_commit_forwarded_requests = 1 + state.provider_success_status = status + if drop_success: + payload = commit_state_unknown( + "injected response loss after provider commit success" + ) + self.send_payload(504, payload, "application/json") + return + self.send_payload(status, response_body, content_type) + + def log_message(self, _format, *_args): + return + + +server = ThreadingHTTPServer(("127.0.0.1", 19001), UncertainCommitProxy) +server_thread = threading.Thread(target=server.serve_forever, daemon=True) +server_thread.start() + +spark = ( + SparkSession.builder.appName("gda-spark-uncertain-commit-reconciliation") + .master("local[2]") + .config( + "spark.sql.extensions", + "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions", + ) + .config("spark.jars", "/opt/spark/jars-extra/iceberg-aws-bundle-1.6.1.jar") + .config("spark.sql.catalog.rest", "org.apache.iceberg.spark.SparkCatalog") + .config("spark.sql.catalog.rest.type", "rest") + .config("spark.sql.catalog.rest.uri", PROXY + "/iceberg") + .config( + "spark.sql.catalog.rest.io-impl", + "org.apache.iceberg.aws.s3.S3FileIO", + ) + .config("spark.sql.catalog.rest.s3.endpoint", OBJECT_STORE_ENDPOINT) + .config("spark.sql.catalog.rest.s3.path-style-access", "true") + .config( + "spark.sql.catalog.rest.s3.access-key-id", + os.environ["AWS_ACCESS_KEY_ID"], + ) + .config( + "spark.sql.catalog.rest.s3.secret-access-key", + os.environ["AWS_SECRET_ACCESS_KEY"], + ) + .config("spark.sql.catalog.rest.client.region", "us-east-1") + .config("spark.sql.catalog.rest.cache-enabled", "false") + .config("spark.sql.catalog.rest.rest.client.max-retries", "1") + .config("spark.sql.catalog.rest.commit.retry.num-retries", "0") + .config("spark.sql.catalog.rest.commit.retry.total-timeout-ms", "1000") + .config("spark.sql.shuffle.partitions", "2") + .getOrCreate() +) +spark.sparkContext.setLogLevel("WARN") + + +def snapshots(): + return [ + { + "snapshot_id": row["snapshot_id"], + "parent_id": row["parent_id"], + "operation": row["operation"], + } + for row in spark.sql( + f"SELECT snapshot_id, parent_id, operation FROM {TABLE}.snapshots " + "ORDER BY committed_at, snapshot_id" + ).collect() + ] + + +def rows(): + return [ + row["probe_id"] + for row in spark.sql(f"SELECT probe_id FROM {TABLE} ORDER BY probe_id").collect() + ] + + +def data_files(): + return sorted( + row["file_path"] + for row in spark.sql(f"SELECT file_path FROM {TABLE}.files").collect() + ) + + +try: + initial_columns = spark.table(TABLE).columns + initial_rows = rows() + initial_snapshots = snapshots() + if initial_columns != ["probe_id"] or initial_rows or initial_snapshots: + raise RuntimeError("unexpected Gravitino-created table baseline") + + ( + spark.createDataFrame([(value,) for value in BASELINE_ROWS], ["probe_id"]) + .coalesce(1) + .writeTo(TABLE) + .append() + ) + baseline_snapshots = snapshots() + baseline_rows = rows() + baseline_files = data_files() + if ( + len(baseline_snapshots) != 1 + or baseline_rows != BASELINE_ROWS + or len(baseline_files) != 1 + ): + raise RuntimeError("baseline commit did not produce one visible snapshot") + + exception_type = None + with state.lock: + state.response_drop_armed = True + try: + ( + spark.createDataFrame([(UNCERTAIN_ROW,)], ["probe_id"]) + .coalesce(1) + .writeTo(TABLE) + .append() + ) + except Exception as exc: + exception_type = type(exc).__name__ + finally: + with state.lock: + state.response_drop_armed = False + if exception_type is None: + raise RuntimeError("post-forward response loss was not observed") + + spark.catalog.refreshTable(TABLE) + reconciled_snapshots = snapshots() + reconciled_rows = rows() + reconciled_files = data_files() + expected_rows = sorted(BASELINE_ROWS + [UNCERTAIN_ROW]) + if ( + len(reconciled_snapshots) != 2 + or reconciled_snapshots[1]["parent_id"] + != reconciled_snapshots[0]["snapshot_id"] + or [item["operation"] for item in reconciled_snapshots] + != ["append", "append"] + or reconciled_rows != expected_rows + or len(reconciled_files) != 2 + or not all(path.startswith(DATA_PREFIX) for path in reconciled_files) + ): + raise RuntimeError("readback could not prove the uncertain commit outcome") + + with state.lock: + proxy = { + "forwarded_commit_requests": state.forwarded_commit_requests, + "uncertain_commit_forwarded_requests": ( + state.uncertain_commit_forwarded_requests + ), + "provider_success_responses_dropped": ( + state.provider_success_responses_dropped + ), + "suppressed_duplicate_commit_requests": ( + state.suppressed_duplicate_commit_requests + ), + "provider_success_status": state.provider_success_status, + "total_requests": state.total_requests, + "injection_mode": "post_forward_success_response_drop_http_504", + "provider_commit_forwarded": True, + "loopback_only": True, + } + if ( + proxy["forwarded_commit_requests"] != 2 + or proxy["uncertain_commit_forwarded_requests"] != 1 + or proxy["provider_success_responses_dropped"] != 1 + or proxy["suppressed_duplicate_commit_requests"] != 1 + or proxy["provider_success_status"] != 200 + ): + raise RuntimeError("uncertain commit proxy boundary did not match") + + result = { + "schema": "gda.spark_uncertain_commit_reconciliation_probe_result.v1", + "spark_version": spark.version, + "iceberg_runtime": "1.6.1", + "catalog_uri": PROXY + "/iceberg", + "catalog_upstream": UPSTREAM + "/iceberg", + "warehouse": WAREHOUSE, + "object_store_endpoint": OBJECT_STORE_ENDPOINT, + "file_io": "org.apache.iceberg.aws.s3.S3FileIO", + "table": TABLE, + "initial_columns": initial_columns, + "initial_rows": initial_rows, + "initial_snapshots": initial_snapshots, + "baseline": { + "rows": baseline_rows, + "snapshots": baseline_snapshots, + "data_file_paths": baseline_files, + }, + "uncertain_attempt": { + "exception_observed": True, + "exception_type": exception_type, + "logical_row": UNCERTAIN_ROW, + }, + "reconciliation": { + "decision": "committed_do_not_resubmit", + "readback_attempts": 1, + "write_resubmitted": False, + "rows": reconciled_rows, + "snapshots": reconciled_snapshots, + "data_file_paths": reconciled_files, + }, + "proxy": proxy, + "provider_committed_response_loss_verified": True, + "commit_outcome_readback_verified": True, + "duplicate_resubmission_prevented": True, + "single_visible_commit_verified": True, + "object_store_data_files_verified": True, + "material_recorded": False, + } + print("GDA_SPARK_COMMIT_FAILURE_RESULT=" + json.dumps(result, sort_keys=True)) +finally: + server.shutdown() + server.server_close() + spark.stop() diff --git a/scripts/metadata-fabric-spark-uncertain-commit-reconciliation.sh b/scripts/metadata-fabric-spark-uncertain-commit-reconciliation.sh new file mode 100755 index 00000000..a61f43be --- /dev/null +++ b/scripts/metadata-fabric-spark-uncertain-commit-reconciliation.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +COMMON_GIT_DIR="$(git -C "$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 "$ROOT/.venv/bin/python" ]; then + PYTHON="$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 "$ROOT" +exec "$PYTHON" -m data_agent.metadata_fabric_spark_uncertain_commit_reconciliation "$@"