diff --git a/yuqei-infra/scripts/apply-rc06-publication-batch.py b/yuqei-infra/scripts/apply-rc06-publication-batch.py index 99de5e26..8bfc3cd9 100644 --- a/yuqei-infra/scripts/apply-rc06-publication-batch.py +++ b/yuqei-infra/scripts/apply-rc06-publication-batch.py @@ -59,12 +59,11 @@ SCENE_LEGAL_REVIEW_BUILDER_PATH = SCRIPT_DIR / "build-scene-source-legal-review. APPLY_REQUEST_SCHEMA_VERSION = "yuqei.rc06-online-apply-request.v3" APPLY_RESPONSE_SCHEMA_VERSION = "yuqei.rc06-online-apply-response.v2" -APPLY_STATUS_SCHEMA_VERSION = "yuqei.rc06-online-apply-status.v2" +TRANSACTION_STATUS_SCHEMA_VERSION = "yuqei.rc06-online-transaction-status.v3" ROLLBACK_REQUEST_SCHEMA_VERSION = "yuqei.rc06-online-rollback-request.v2" ROLLBACK_RESPONSE_SCHEMA_VERSION = "yuqei.rc06-online-rollback-response.v2" FINALIZE_REQUEST_SCHEMA_VERSION = "yuqei.rc06-online-finalize-request.v1" FINALIZE_RESPONSE_SCHEMA_VERSION = "yuqei.rc06-online-finalize-response.v1" -FINALIZE_STATUS_SCHEMA_VERSION = "yuqei.rc06-online-finalize-status.v1" EXECUTION_EVIDENCE_SCHEMA_VERSION = "yuqei.rc06-online-apply-evidence.v2" FINAL_EVIDENCE_SCHEMA_VERSION = "yuqei.rc06-finalized-apply-evidence.v1" LEDGER_SCHEMA_VERSION = "yuqei.rc06-local-apply-ledger.v2" @@ -334,7 +333,11 @@ class ApplyLedger: receipt = json.loads(row[2]) self.connection.execute("COMMIT") return "replay", receipt - if row is not None and row[1] in {"dispatching", "verifying"}: + if row is not None and row[1] in { + "dispatching", + "verifying", + "irreversible_failure_no_go", + }: timestamp = now.astimezone(UTC).isoformat(timespec="seconds") self._event( key, @@ -346,11 +349,17 @@ class ApplyLedger: return "recover", None if row is not None: raise ApplyPublicationError("an apply with this idempotency key needs reconciliation") - if not initial_dispatch_authorized: - raise ApplyPublicationError( - "apply challenge is not currently valid for an initial dispatch" - ) timestamp = now.astimezone(UTC).isoformat(timespec="seconds") + if not initial_dispatch_authorized: + self.connection.execute( + "INSERT INTO rc06_apply_operations " + "(idempotency_key, request_sha256, state, updated_at) " + "VALUES (?, ?, 'probing_remote', ?)", + (key, request_sha256, timestamp), + ) + self._event(key, "probing_remote", request_sha256, timestamp) + self.connection.execute("COMMIT") + return "probe", None self.connection.execute( "INSERT INTO rc06_apply_operations " "(idempotency_key, request_sha256, state, updated_at) VALUES (?, ?, 'dispatching', ?) " @@ -503,81 +512,268 @@ def _validate_apply_response( return row -def _validate_online_status( +_TRANSACTION_STATUS_FIELDS = ( + "schema_version", + "idempotency_key", + "request_sha256", + "transaction_id", + "state", + "state_version", + "no_go", + "legal_source_review_binding_sha256", + "environment_id", + "tenant_id", + "transaction_mode", + "cross_database_atomicity", + "publication_item_results", + "publication_item_count", + "runtime_snapshot_dependencies", + "runtime_snapshot_dependency_count", + "participant_evidence", + "online_transaction_receipt", + "online_transaction_receipt_sha256", + "runtime_snapshot_receipt", + "runtime_snapshot_receipt_sha256", + "apply_receipt", + "apply_receipt_sha256", + "finalized_at", + "rollback_response", + "rollback_response_sha256", + "failure_code", + "failure_at", + "failure_evidence_sha256", + "status_updated_at", +) + +_TRANSACTION_STATE_VERSION = { + "prepared": 1, + "rolled_back": 2, + # This is the post-Contract-commit, no-rollback boundary. Replaying the + # exact request may still recover read-back/KMS evidence and advance to 3. + "irreversible_failure": 2, + "committed_and_verified": 3, + "finalized": 4, +} + + +def _validate_transaction_status( status: Any, *, + manifest: Mapping[str, Any], + manifest_raw_sha256: str, + approval: Mapping[str, Any], + approval_raw_sha256: str, + approval_keyring: Mapping[str, Any], + approval_root_public_key: Ed25519PublicKey, + apply_service_public_key: Ed25519PublicKey, + online_service_public_key: Any, + request_sha256: str, key: str, transaction_id: str, legal_review_binding_sha256: str, + legal_review_expires_at: str, environment_id: str, tenant_id: str, - online_transaction_receipt: Mapping[str, Any], - runtime_snapshot_receipt: Mapping[str, Any], - expected_publication_results: list[dict[str, Any]], - expected_runtime_results: list[dict[str, Any]], + minimum_state_version: int = 1, + expected_apply_receipt: Mapping[str, Any] | None = None, ) -> dict[str, Any]: - row = _require_exact( - status, - "online apply status", - ( - "schema_version", - "idempotency_key", - "transaction_id", - "state", - "legal_source_review_binding_sha256", - "environment_id", - "tenant_id", - "transaction_mode", - "cross_database_atomicity", - "publication_item_results", - "publication_item_count", - "runtime_snapshot_dependencies", - "runtime_snapshot_dependency_count", - "online_transaction_receipt_sha256", - "runtime_snapshot_receipt_sha256", - "participant_evidence", - ), - ) + """Validate the one exact forward GET status contract for every client.""" + + row = _require_exact(status, "online transaction status", _TRANSACTION_STATUS_FIELDS) + state = row["state"] + expected_state_version = _TRANSACTION_STATE_VERSION.get(state) if ( - row["schema_version"] != APPLY_STATUS_SCHEMA_VERSION + row["schema_version"] != TRANSACTION_STATUS_SCHEMA_VERSION or row["idempotency_key"] != key + or row["request_sha256"] != request_sha256 or row["transaction_id"] != transaction_id - or row["state"] != "committed_and_verified" + or expected_state_version is None + or not isinstance(row["state_version"], int) + or isinstance(row["state_version"], bool) + or row["state_version"] != expected_state_version + or row["state_version"] < minimum_state_version + or row["no_go"] is not (state != "finalized") or row["legal_source_review_binding_sha256"] != legal_review_binding_sha256 or row["environment_id"] != environment_id or row["tenant_id"] != tenant_id or row["transaction_mode"] != batch.TRANSACTION_MODE or row["cross_database_atomicity"] is not False - or row["publication_item_results"] != expected_publication_results + ): + raise ApplyPublicationError("online transaction status identity or state drift") + batch._require_sha(row["request_sha256"], "online status request SHA-256") + status_updated_at = batch._parse_time( + row["status_updated_at"], "online status_updated_at" + ) + + commit_empty = ( + row["publication_item_results"] == [] + and row["publication_item_count"] == 0 + and row["runtime_snapshot_dependencies"] == [] + and row["runtime_snapshot_dependency_count"] == 0 + and row["participant_evidence"] == [] + and row["online_transaction_receipt"] is None + and row["online_transaction_receipt_sha256"] is None + and row["runtime_snapshot_receipt"] is None + and row["runtime_snapshot_receipt_sha256"] is None + ) + final_empty = ( + row["apply_receipt"] is None + and row["apply_receipt_sha256"] is None + and row["finalized_at"] is None + ) + rollback_empty = ( + row["rollback_response"] is None + and row["rollback_response_sha256"] is None + ) + failure_empty = ( + row["failure_code"] is None + and row["failure_at"] is None + and row["failure_evidence_sha256"] is None + ) + + if state in {"prepared", "irreversible_failure", "rolled_back"} and not commit_empty: + raise ApplyPublicationError("non-committed online status contains commit evidence") + if state != "finalized" and not final_empty: + raise ApplyPublicationError("non-finalized online status contains finalization evidence") + if state != "rolled_back" and not rollback_empty: + raise ApplyPublicationError("non-rollback online status contains rollback evidence") + if state != "irreversible_failure" and not failure_empty: + raise ApplyPublicationError("non-failure online status contains failure evidence") + + if state == "prepared": + if not rollback_empty or not failure_empty: + raise ApplyPublicationError("prepared online status evidence is invalid") + return row + + if state == "rolled_back": + rollback = _validate_rollback_response( + row["rollback_response"], + manifest=manifest, + key=key, + transaction_id=transaction_id, + ) + if row["rollback_response_sha256"] != batch._sha256_json(rollback): + raise ApplyPublicationError("online rollback status hash drift") + return row + + if state == "irreversible_failure": + _require_text(row["failure_code"], "online failure_code", maximum=128) + failure_at = batch._parse_time(row["failure_at"], "online failure_at") + batch._require_sha( + row["failure_evidence_sha256"], "online failure evidence SHA-256" + ) + if failure_at > status_updated_at: + raise ApplyPublicationError("online failure postdates its status update") + return row + + expected_publication_results = batch._expected_apply_item_results(manifest) + expected_runtime_results = expected_runtime_snapshot_results(manifest) + online_transaction_receipt = row["online_transaction_receipt"] + runtime_snapshot_receipt = row["runtime_snapshot_receipt"] + if not isinstance(online_transaction_receipt, dict) or not isinstance( + runtime_snapshot_receipt, dict + ): + raise ApplyPublicationError("committed online status omits signed receipt evidence") + online_receipt = batch.validate_online_transaction_receipt( + manifest, + manifest_raw_sha256, + approval_raw_sha256, + approval, + apply_service_public_key, + online_service_public_key, + online_transaction_receipt, + legal_review_binding_sha256=legal_review_binding_sha256, + legal_review_expires_at=legal_review_expires_at, + environment_id=environment_id, + tenant_id=tenant_id, + ) + batch.validate_runtime_snapshot_receipt( + manifest, + manifest_raw_sha256, + approval_raw_sha256, + approval, + apply_service_public_key, + online_service_public_key, + online_receipt, + runtime_snapshot_receipt, + legal_review_binding_sha256=legal_review_binding_sha256, + legal_review_expires_at=legal_review_expires_at, + environment_id=environment_id, + tenant_id=tenant_id, + ) + if ( + row["publication_item_results"] != expected_publication_results + or row["publication_item_results"] + != online_transaction_receipt["payload"]["item_results"] or row["publication_item_count"] != 74 or row["runtime_snapshot_dependencies"] != expected_runtime_results + or row["runtime_snapshot_dependencies"] + != runtime_snapshot_receipt["payload"]["runtime_snapshot_dependencies"] or row["runtime_snapshot_dependency_count"] != 14 + or row["participant_evidence"] + != online_transaction_receipt["payload"]["participant_evidence"] or row["online_transaction_receipt_sha256"] != batch._sha256_json(online_transaction_receipt) or row["runtime_snapshot_receipt_sha256"] != batch._sha256_json(runtime_snapshot_receipt) - or row["participant_evidence"] - != online_transaction_receipt["payload"]["participant_evidence"] + or not rollback_empty + or not failure_empty ): - raise ApplyPublicationError("online read-after-write hash verification failed") - return row + raise ApplyPublicationError("online read-after-write receipt verification failed") + latest_commit_evidence_at = max( + batch._parse_time( + online_transaction_receipt["payload"]["committed_at"], + "online receipt committed_at", + ), + batch._parse_time( + runtime_snapshot_receipt["payload"]["captured_at"], + "runtime snapshot captured_at", + ), + ) + if status_updated_at < latest_commit_evidence_at: + raise ApplyPublicationError("online status predates its signed commit evidence") + if state == "committed_and_verified": + if not final_empty: + raise ApplyPublicationError("committed status contains finalization evidence") + return row - -def _remote_status_state( - status: Any, - *, - key: str, - transaction_id: str, -) -> str | None: + apply_receipt = row["apply_receipt"] + if not isinstance(apply_receipt, dict): + raise ApplyPublicationError("finalized status omits the signed apply receipt") + batch.validate_apply_receipt( + manifest, + manifest_raw_sha256, + approval, + approval_keyring, + approval_root_public_key, + apply_service_public_key, + online_service_public_key, + approval_raw_sha256, + apply_receipt, + legal_review_binding_sha256=legal_review_binding_sha256, + legal_review_expires_at=legal_review_expires_at, + environment_id=environment_id, + tenant_id=tenant_id, + ) if ( - not isinstance(status, dict) - or status.get("schema_version") != APPLY_STATUS_SCHEMA_VERSION - or status.get("idempotency_key") != key - or status.get("transaction_id") != transaction_id - or not isinstance(status.get("state"), str) + apply_receipt["payload"]["online_transaction_receipt"] + != online_transaction_receipt + or apply_receipt["payload"]["runtime_snapshot_receipt"] + != runtime_snapshot_receipt + or row["apply_receipt_sha256"] != batch._sha256_json(apply_receipt) + or ( + expected_apply_receipt is not None + and apply_receipt != expected_apply_receipt + ) ): - return None - return status["state"] + raise ApplyPublicationError("finalized status receipt binding drift") + finalized_at = batch._parse_time(row["finalized_at"], "online finalized_at") + countersigned_at = batch._parse_time( + apply_receipt["payload"]["countersigned_at"], "apply countersigned_at" + ) + if finalized_at < countersigned_at or status_updated_at < finalized_at: + raise ApplyPublicationError("online finalized status timestamp drift") + return row def _validate_compensation_hashes( @@ -619,37 +815,13 @@ def _validate_compensation_hashes( raise ApplyPublicationError(f"{label} action is unsupported") -def _compensate( - client: JsonHttpClient, +def _validate_rollback_response( + response: Any, *, manifest: Mapping[str, Any], key: str, transaction_id: str, - headers: Mapping[str, str], - manifest_raw_sha256: str, - approval_raw_sha256: str, - legal_review_binding_sha256: str, - environment_id: str, - tenant_id: str, ) -> dict[str, Any]: - request = { - "schema_version": ROLLBACK_REQUEST_SCHEMA_VERSION, - "idempotency_key": key, - "transaction_id": transaction_id, - "batch_manifest_raw_sha256": manifest_raw_sha256, - "approval_receipt_sha256": approval_raw_sha256, - "legal_source_review_binding_sha256": legal_review_binding_sha256, - "environment_id": environment_id, - "tenant_id": tenant_id, - "transaction_mode": batch.TRANSACTION_MODE, - "cross_database_atomicity": False, - } - response = client.request( - "POST", - f"/internal/rc06/publication-transactions/{transaction_id}/rollback", - payload=request, - headers=headers, - ) row = _require_exact( response, "online rollback response", @@ -886,6 +1058,45 @@ def _compensate( return row +def _compensate( + client: JsonHttpClient, + *, + manifest: Mapping[str, Any], + key: str, + transaction_id: str, + headers: Mapping[str, str], + manifest_raw_sha256: str, + approval_raw_sha256: str, + legal_review_binding_sha256: str, + environment_id: str, + tenant_id: str, +) -> dict[str, Any]: + request = { + "schema_version": ROLLBACK_REQUEST_SCHEMA_VERSION, + "idempotency_key": key, + "transaction_id": transaction_id, + "batch_manifest_raw_sha256": manifest_raw_sha256, + "approval_receipt_sha256": approval_raw_sha256, + "legal_source_review_binding_sha256": legal_review_binding_sha256, + "environment_id": environment_id, + "tenant_id": tenant_id, + "transaction_mode": batch.TRANSACTION_MODE, + "cross_database_atomicity": False, + } + response = client.request( + "POST", + f"/internal/rc06/publication-transactions/{transaction_id}/rollback", + payload=request, + headers=headers, + ) + return _validate_rollback_response( + response, + manifest=manifest, + key=key, + transaction_id=transaction_id, + ) + + def _validate_execution_evidence( evidence: Any, *, @@ -968,6 +1179,69 @@ def _validate_execution_evidence( return row +def _execution_from_transaction_status( + status: Mapping[str, Any], + *, + request_sha256: str, + legal_review_expires_at: str, +) -> dict[str, Any]: + if status["state"] not in {"committed_and_verified", "finalized"}: + raise ApplyPublicationError("online status does not contain committed execution evidence") + return { + "schema_version": EXECUTION_EVIDENCE_SCHEMA_VERSION, + "idempotency_key": status["idempotency_key"], + "transaction_id": status["transaction_id"], + "request_sha256": request_sha256, + "legal_source_review_binding_sha256": status[ + "legal_source_review_binding_sha256" + ], + "legal_source_review_expires_at": legal_review_expires_at, + "environment_id": status["environment_id"], + "tenant_id": status["tenant_id"], + "transaction_mode": status["transaction_mode"], + "cross_database_atomicity": status["cross_database_atomicity"], + "publication_item_count": status["publication_item_count"], + "runtime_snapshot_dependency_count": status[ + "runtime_snapshot_dependency_count" + ], + "online_transaction_receipt": status["online_transaction_receipt"], + "runtime_snapshot_receipt": status["runtime_snapshot_receipt"], + } + + +def _finalize_response_from_transaction_status( + status: Mapping[str, Any], +) -> dict[str, Any]: + if status["state"] != "finalized": + raise ApplyPublicationError("online status is not finalized") + return { + "schema_version": FINALIZE_RESPONSE_SCHEMA_VERSION, + "idempotency_key": status["idempotency_key"], + "transaction_id": status["transaction_id"], + "state": "finalized", + "no_go": False, + "apply_receipt_sha256": status["apply_receipt_sha256"], + "finalized_at": status["finalized_at"], + } + + +def _final_evidence_from_transaction_status( + execution: Mapping[str, Any], + status: Mapping[str, Any], +) -> dict[str, Any]: + return { + "schema_version": FINAL_EVIDENCE_SCHEMA_VERSION, + "transaction_mode": batch.TRANSACTION_MODE, + "cross_database_atomicity": False, + "publication_item_count": 74, + "runtime_snapshot_dependency_count": 14, + "online_execution": execution, + "apply_receipt": status["apply_receipt"], + "online_finalize_response": _finalize_response_from_transaction_status(status), + "online_finalize_status": status, + } + + def execute_online_publication_batch( *, manifest: Mapping[str, Any], @@ -1129,7 +1403,69 @@ def execute_online_publication_batch( legal_review_binding_sha256, tenant_id, ) - if reservation == "recover" and not challenge_is_current: + + def validate_remote_status( + value: Any, + *, + minimum_state_version: int = 1, + expected_apply_receipt: Mapping[str, Any] | None = None, + ) -> dict[str, Any]: + return _validate_transaction_status( + value, + manifest=manifest, + manifest_raw_sha256=manifest_raw_sha256, + approval=approval, + approval_raw_sha256=approval_raw_sha256, + approval_keyring=approval_keyring, + approval_root_public_key=approval_root_public_key, + apply_service_public_key=apply_service_public_key, + online_service_public_key=online_service_public_key, + request_sha256=request_sha256, + key=key, + transaction_id=transaction_id, + legal_review_binding_sha256=legal_review_binding_sha256, + legal_review_expires_at=legal_review_expires_at, + environment_id=environment_id, + tenant_id=tenant_id, + minimum_state_version=minimum_state_version, + expected_apply_receipt=expected_apply_receipt, + ) + + def adopt_committed_status(status_row: Mapping[str, Any]) -> dict[str, Any]: + execution = _execution_from_transaction_status( + status_row, + request_sha256=request_sha256, + legal_review_expires_at=legal_review_expires_at, + ) + execution = _validate_execution_evidence( + execution, + manifest=manifest, + manifest_raw_sha256=manifest_raw_sha256, + approval=approval, + approval_raw_sha256=approval_raw_sha256, + apply_service_public_key=apply_service_public_key, + online_service_public_key=online_service_public_key, + legal_review_binding_sha256=legal_review_binding_sha256, + legal_review_expires_at=legal_review_expires_at, + environment_id=environment_id, + tenant_id=tenant_id, + idempotency_key=key, + transaction_id=transaction_id, + ) + if status_row["state"] == "finalized": + final_evidence = _final_evidence_from_transaction_status(execution, status_row) + ledger.state(key, "succeeded", status_row, now, receipt=final_evidence) + else: + ledger.state( + key, + "online_committed_pending_countersign", + status_row, + now, + receipt=execution, + ) + return execution + + if reservation in {"probe", "recover"} and not challenge_is_current: try: reconciliation = client.request( "GET", @@ -1137,11 +1473,7 @@ def execute_online_publication_batch( payload=None, headers=headers, ) - remote_state = _remote_status_state( - reconciliation, - key=key, - transaction_id=transaction_id, - ) + reconciliation = validate_remote_status(reconciliation) except Exception as exc: ledger.state( key, @@ -1151,17 +1483,15 @@ def execute_online_publication_batch( error="expired challenge recovery could not establish remote commit status", ) raise ApplyPublicationError( - "expired challenge cannot be retransmitted until an existing terminal commit is proven" + "apply challenge is not currently valid for an initial dispatch and no existing transaction was proven" ) from exc + remote_state = reconciliation["state"] + if remote_state in {"committed_and_verified", "finalized"}: + return adopt_committed_status(reconciliation) if remote_state == "rolled_back": ledger.state(key, "rolled_back", reconciliation, now) raise ApplyPublicationError("online apply was already rolled back") - if remote_state not in { - "contract_committed", - "evidence_pending", - "committed_and_verified", - "online_committed_pending_countersign", - }: + if remote_state == "prepared": ledger.state( key, "expired_challenge_recovery_no_go", @@ -1172,6 +1502,10 @@ def execute_online_publication_batch( raise ApplyPublicationError( "expired challenge cannot authorize a new or non-terminal retransmission" ) + # irreversible_failure proves that this exact request already crossed + # the no-rollback boundary. Replaying it may only recover its evidence. + if remote_state != "irreversible_failure": + raise ApplyPublicationError("online transaction state is unsupported") dispatch_started = False response: Any = None try: @@ -1231,42 +1565,15 @@ def execute_online_publication_batch( payload=None, headers=headers, ) - status = _validate_online_status( - status, - key=key, - transaction_id=transaction_id, - legal_review_binding_sha256=legal_review_binding_sha256, - environment_id=environment_id, - tenant_id=tenant_id, - online_transaction_receipt=online_receipt, - runtime_snapshot_receipt=runtime_receipt, - expected_publication_results=batch._expected_apply_item_results(manifest), - expected_runtime_results=expected_runtime_snapshot_results(manifest), - ) - evidence = { - "schema_version": EXECUTION_EVIDENCE_SCHEMA_VERSION, - "idempotency_key": key, - "transaction_id": transaction_id, - "request_sha256": request_sha256, - "legal_source_review_binding_sha256": legal_review_binding_sha256, - "legal_source_review_expires_at": legal_review_expires_at, - "environment_id": environment_id, - "tenant_id": tenant_id, - "transaction_mode": batch.TRANSACTION_MODE, - "cross_database_atomicity": False, - "publication_item_count": 74, - "runtime_snapshot_dependency_count": 14, - "online_transaction_receipt": online_receipt, - "runtime_snapshot_receipt": runtime_receipt, - } - ledger.state( - key, - "online_committed_pending_countersign", - status, - now, - receipt=evidence, - ) - return evidence + status = validate_remote_status(status, minimum_state_version=3) + if status["state"] not in {"committed_and_verified", "finalized"}: + raise ApplyPublicationError("online apply did not reach a committed status") + if ( + status["online_transaction_receipt"] != online_receipt + or status["runtime_snapshot_receipt"] != runtime_receipt + ): + raise ApplyPublicationError("online apply response and durable status receipt drift") + return adopt_committed_status(status) except Exception as exc: if not isinstance(exc, ApplyPublicationError): failure = ApplyPublicationError("unexpected online apply failure") @@ -1281,11 +1588,7 @@ def execute_online_publication_batch( payload=None, headers=headers, ) - remote_state = _remote_status_state( - reconciliation, - key=key, - transaction_id=transaction_id, - ) + reconciliation = validate_remote_status(reconciliation) except Exception as reconciliation_exc: ledger.state( key, @@ -1300,26 +1603,24 @@ def execute_online_publication_batch( raise ApplyPublicationError( "online apply evidence failed and commit status is unknown; no rollback was attempted" ) from reconciliation_exc - if remote_state in { - "contract_committed", - "evidence_pending", - "committed_and_verified", - "online_committed_pending_countersign", - }: + remote_state = reconciliation["state"] + if remote_state in {"committed_and_verified", "finalized"}: + return adopt_committed_status(reconciliation) + if remote_state == "irreversible_failure": ledger.state( key, - "committed_evidence_invalid_no_go", + "irreversible_failure_no_go", reconciliation, now, error=str(failure), ) raise ApplyPublicationError( - "online saga is terminally committed but local evidence verification failed; rollback is forbidden" + "online saga crossed the irreversible boundary but evidence recovery is incomplete; rollback is forbidden" ) from failure if remote_state == "rolled_back": ledger.state(key, "rolled_back", reconciliation, now, error=str(failure)) raise failure - if remote_state not in {"prepared", "ai_committed", "failed_before_commit"}: + if remote_state != "prepared": ledger.state( key, "commit_status_unknown", @@ -1405,17 +1706,6 @@ def finalize_publication_batch( idempotency_key=key, transaction_id=transaction_id, ) - state, stored = ledger.load(key, execution["request_sha256"]) - if state not in { - "online_committed_pending_countersign", - "finalize_failed_no_go", - "succeeded", - }: - raise ApplyPublicationError("online saga is not eligible for offline finalization") - if state == "succeeded" and stored.get("schema_version") == FINAL_EVIDENCE_SCHEMA_VERSION: - return stored - if stored != execution: - raise ApplyPublicationError("supplied online evidence differs from the durable ledger") batch.validate_apply_receipt( manifest, manifest_raw_sha256, @@ -1439,6 +1729,67 @@ def finalize_publication_batch( ): raise ApplyPublicationError("offline countersign is not bound to this receipt pair") apply_receipt_sha256 = batch._sha256_json(apply_receipt) + state, stored = ledger.load(key, execution["request_sha256"]) + if state not in { + "online_committed_pending_countersign", + "finalize_failed_no_go", + "succeeded", + }: + raise ApplyPublicationError("online saga is not eligible for offline finalization") + + def validate_remote_status(value: Any, *, minimum_state_version: int) -> dict[str, Any]: + return _validate_transaction_status( + value, + manifest=manifest, + manifest_raw_sha256=manifest_raw_sha256, + approval=approval, + approval_raw_sha256=approval_raw_sha256, + approval_keyring=approval_keyring, + approval_root_public_key=approval_root_public_key, + apply_service_public_key=apply_service_public_key, + online_service_public_key=online_service_public_key, + request_sha256=execution["request_sha256"], + key=key, + transaction_id=transaction_id, + legal_review_binding_sha256=legal_review_binding_sha256, + legal_review_expires_at=legal_review_expires_at, + environment_id=environment_id, + tenant_id=tenant_id, + minimum_state_version=minimum_state_version, + expected_apply_receipt=apply_receipt, + ) + + if state == "succeeded": + stored_final = _require_exact( + stored, + "stored finalized evidence", + ( + "schema_version", + "transaction_mode", + "cross_database_atomicity", + "publication_item_count", + "runtime_snapshot_dependency_count", + "online_execution", + "apply_receipt", + "online_finalize_response", + "online_finalize_status", + ), + ) + if ( + stored_final["schema_version"] != FINAL_EVIDENCE_SCHEMA_VERSION + or stored_final["online_execution"] != execution + or stored_final["apply_receipt"] != apply_receipt + ): + raise ApplyPublicationError("stored finalized evidence differs from the exact replay") + stored_status = validate_remote_status( + stored_final["online_finalize_status"], minimum_state_version=4 + ) + if stored_final != _final_evidence_from_transaction_status(execution, stored_status): + raise ApplyPublicationError("stored finalized evidence is internally inconsistent") + return stored_final + if stored != execution: + raise ApplyPublicationError("supplied online evidence differs from the durable ledger") + finalize_request = { "schema_version": FINALIZE_REQUEST_SCHEMA_VERSION, "idempotency_key": key, @@ -1464,6 +1815,7 @@ def finalize_publication_batch( legal_review_binding_sha256, tenant_id, ) + response_row: dict[str, Any] | None = None try: response = client.request( "POST", @@ -1509,69 +1861,62 @@ def finalize_publication_batch( payload=None, headers=headers, ) - status_row = _require_exact( - status, - "online finalize status", - ( - "schema_version", - "idempotency_key", - "transaction_id", - "state", - "no_go", - "transaction_mode", - "cross_database_atomicity", - "publication_item_count", - "runtime_snapshot_dependency_count", - "online_transaction_receipt_sha256", - "runtime_snapshot_receipt_sha256", - "apply_receipt_sha256", - "finalized_at", - ), - ) - if ( - status_row["schema_version"] != FINALIZE_STATUS_SCHEMA_VERSION - or status_row["idempotency_key"] != key - or status_row["transaction_id"] != transaction_id - or status_row["state"] != "finalized" - or status_row["no_go"] is not False - or status_row["transaction_mode"] != batch.TRANSACTION_MODE - or status_row["cross_database_atomicity"] is not False - or status_row["publication_item_count"] != 74 - or status_row["runtime_snapshot_dependency_count"] != 14 - or status_row["online_transaction_receipt_sha256"] - != finalize_request["online_transaction_receipt_sha256"] - or status_row["runtime_snapshot_receipt_sha256"] - != finalize_request["runtime_snapshot_receipt_sha256"] - or status_row["apply_receipt_sha256"] != apply_receipt_sha256 - or status_row["finalized_at"] != response_row["finalized_at"] - ): - raise ApplyPublicationError("online finalized status verification failed") + status_row = validate_remote_status(status, minimum_state_version=4) + if status_row["state"] != "finalized": + raise ApplyPublicationError("online finalize did not reach finalized status") + if response_row != _finalize_response_from_transaction_status(status_row): + raise ApplyPublicationError("online finalize response and durable status drift") except Exception as exc: failure = ( exc if isinstance(exc, ApplyPublicationError) else ApplyPublicationError("online finalize request failed") ) + try: + reconciliation = client.request( + "GET", + f"/internal/rc06/publication-transactions/{transaction_id}/status", + payload=None, + headers=headers, + ) + reconciliation = validate_remote_status( + reconciliation, minimum_state_version=3 + ) + if reconciliation["state"] == "finalized": + final_evidence = _final_evidence_from_transaction_status( + execution, reconciliation + ) + ledger.state( + key, "succeeded", reconciliation, now, receipt=final_evidence + ) + return final_evidence + if reconciliation["state"] != "committed_and_verified": + raise ApplyPublicationError( + "online finalization reconciliation regressed from committed state" + ) + except Exception as reconciliation_exc: + ledger.state( + key, + "finalize_failed_no_go", + { + "finalize_error": str(failure), + "reconciliation_error": str(reconciliation_exc), + }, + now, + receipt=execution, + error=f"{failure}; finalization status is untrusted", + ) + raise failure from reconciliation_exc ledger.state( key, "finalize_failed_no_go", - {"error": str(failure)}, + {"finalize_error": str(failure), "status": reconciliation}, now, receipt=execution, error=str(failure), ) raise failure - final_evidence = { - "schema_version": FINAL_EVIDENCE_SCHEMA_VERSION, - "transaction_mode": batch.TRANSACTION_MODE, - "cross_database_atomicity": False, - "publication_item_count": 74, - "runtime_snapshot_dependency_count": 14, - "online_execution": execution, - "apply_receipt": apply_receipt, - "online_finalize_response": response_row, - "online_finalize_status": status_row, - } + final_evidence = _final_evidence_from_transaction_status(execution, status_row) ledger.state(key, "succeeded", final_evidence, now, receipt=final_evidence) return final_evidence diff --git a/yuqei-infra/tests/test_rc06_publication_apply.py b/yuqei-infra/tests/test_rc06_publication_apply.py index bbd4f30d..c00bcef0 100644 --- a/yuqei-infra/tests/test_rc06_publication_apply.py +++ b/yuqei-infra/tests/test_rc06_publication_apply.py @@ -495,9 +495,12 @@ class FakePublicationService: self.rollback_calls = 0 self.finalize_calls = 0 self.request: dict | None = None + self.request_sha256: str | None = None self.response: dict | None = None self.state: str | None = None self.finalize_request: dict | None = None + self.finalized_at: str | None = None + self.rollback_response: dict | None = None # The fake online service is deliberately incapable of minting the # offline authority receipt. assert not hasattr(self, "apply_private") @@ -632,8 +635,11 @@ class FakePublicationService: def dispatch(self, request: dict, request_sha256: str) -> tuple[int, dict]: self.apply_calls += 1 if self.response is not None: + if self.state == "irreversible_failure": + self.state = "committed_and_verified" return 200, self.response self.request = request + self.request_sha256 = request_sha256 if self.fail_before_commit: self.state = "prepared" return 409, {"error": "injected failure before any durable commit"} @@ -703,10 +709,11 @@ class FakePublicationService: "unchanged_item_count": 0, } self.state = "rolled_back" - return 200, { + self.rollback_response = { **projection, "rollback_evidence_sha256": apply.batch._sha256_json(projection), } + return 200, self.rollback_response def finalize(self, transaction_id: str, request: dict) -> tuple[int, dict]: self.finalize_calls += 1 @@ -724,6 +731,7 @@ class FakePublicationService: ) + timedelta(seconds=1) ).isoformat(timespec="seconds") + self.finalized_at = finalized_at return 200, { "schema_version": apply.FINALIZE_RESPONSE_SCHEMA_VERSION, "idempotency_key": request["idempotency_key"], @@ -735,51 +743,19 @@ class FakePublicationService: } def status(self, transaction_id: str) -> tuple[int, dict]: - assert self.request is not None - if self.state in {"prepared", "rolled_back"}: - return 200, { - "schema_version": apply.APPLY_STATUS_SCHEMA_VERSION, - "idempotency_key": self.request["idempotency_key"], - "transaction_id": transaction_id, - "state": self.state, - } - assert self.response is not None - online = self.response["online_transaction_receipt"] - runtime = self.response["runtime_snapshot_receipt"] - if self.state == "finalized": - assert self.finalize_request is not None - finalized_at = ( - datetime.fromisoformat( - self.finalize_request["apply_receipt"]["payload"]["countersigned_at"] - ) - + timedelta(seconds=1) - ).isoformat(timespec="seconds") - return 200, { - "schema_version": apply.FINALIZE_STATUS_SCHEMA_VERSION, - "idempotency_key": self.request["idempotency_key"], - "transaction_id": transaction_id, - "state": "finalized", - "no_go": False, - "transaction_mode": apply.batch.TRANSACTION_MODE, - "cross_database_atomicity": False, - "publication_item_count": 74, - "runtime_snapshot_dependency_count": 14, - "online_transaction_receipt_sha256": apply.batch._sha256_json(online), - "runtime_snapshot_receipt_sha256": apply.batch._sha256_json(runtime), - "apply_receipt_sha256": self.finalize_request["apply_receipt_sha256"], - "finalized_at": finalized_at, - } - publication = copy.deepcopy(online["payload"]["item_results"]) - snapshots = copy.deepcopy(runtime["payload"]["runtime_snapshot_dependencies"]) - if self.tamper_status == "publication": - publication[0]["online_content_sha256"] = "f" * 64 - if self.tamper_status == "runtime": - snapshots[0]["online_content_sha256"] = "f" * 64 - return 200, { - "schema_version": apply.APPLY_STATUS_SCHEMA_VERSION, + if self.request is None: + return 404, {"error": "transaction not found"} + assert self.request_sha256 is not None + state = self.state + assert state in apply._TRANSACTION_STATE_VERSION + result = { + "schema_version": apply.TRANSACTION_STATUS_SCHEMA_VERSION, "idempotency_key": self.request["idempotency_key"], + "request_sha256": self.request_sha256, "transaction_id": transaction_id, - "state": "committed_and_verified", + "state": state, + "state_version": apply._TRANSACTION_STATE_VERSION[state], + "no_go": state != "finalized", "legal_source_review_binding_sha256": self.request[ "legal_source_review_binding_sha256" ], @@ -787,14 +763,110 @@ class FakePublicationService: "tenant_id": self.request["tenant_id"], "transaction_mode": apply.batch.TRANSACTION_MODE, "cross_database_atomicity": False, - "publication_item_results": publication, - "publication_item_count": 74, - "runtime_snapshot_dependencies": snapshots, - "runtime_snapshot_dependency_count": 14, - "online_transaction_receipt_sha256": apply.batch._sha256_json(online), - "runtime_snapshot_receipt_sha256": apply.batch._sha256_json(runtime), - "participant_evidence": online["payload"]["participant_evidence"], + "publication_item_results": [], + "publication_item_count": 0, + "runtime_snapshot_dependencies": [], + "runtime_snapshot_dependency_count": 0, + "participant_evidence": [], + "online_transaction_receipt": None, + "online_transaction_receipt_sha256": None, + "runtime_snapshot_receipt": None, + "runtime_snapshot_receipt_sha256": None, + "apply_receipt": None, + "apply_receipt_sha256": None, + "finalized_at": None, + "rollback_response": None, + "rollback_response_sha256": None, + "failure_code": None, + "failure_at": None, + "failure_evidence_sha256": None, + "status_updated_at": self.now.isoformat(timespec="seconds"), } + if state in {"committed_and_verified", "finalized"}: + assert self.response is not None + online = self.response["online_transaction_receipt"] + runtime = self.response["runtime_snapshot_receipt"] + result.update( + { + "publication_item_results": copy.deepcopy( + online["payload"]["item_results"] + ), + "publication_item_count": 74, + "runtime_snapshot_dependencies": copy.deepcopy( + runtime["payload"]["runtime_snapshot_dependencies"] + ), + "runtime_snapshot_dependency_count": 14, + "participant_evidence": copy.deepcopy( + online["payload"]["participant_evidence"] + ), + "online_transaction_receipt": copy.deepcopy(online), + "online_transaction_receipt_sha256": apply.batch._sha256_json( + online + ), + "runtime_snapshot_receipt": copy.deepcopy(runtime), + "runtime_snapshot_receipt_sha256": apply.batch._sha256_json( + runtime + ), + } + ) + if self.tamper_status == "publication": + result["publication_item_results"][0]["online_content_sha256"] = "f" * 64 + if self.tamper_status == "runtime": + result["runtime_snapshot_dependencies"][0]["online_content_sha256"] = ( + "f" * 64 + ) + if state == "finalized": + assert self.finalize_request is not None and self.finalized_at is not None + result.update( + { + "apply_receipt": copy.deepcopy( + self.finalize_request["apply_receipt"] + ), + "apply_receipt_sha256": self.finalize_request[ + "apply_receipt_sha256" + ], + "finalized_at": self.finalized_at, + "status_updated_at": self.finalized_at, + } + ) + elif state == "rolled_back": + assert self.rollback_response is not None + result["rollback_response"] = copy.deepcopy(self.rollback_response) + result["rollback_response_sha256"] = apply.batch._sha256_json( + self.rollback_response + ) + elif state == "irreversible_failure": + failure = { + "code": "contract_committed_evidence_pending", + "failed_at": self.now.isoformat(timespec="seconds"), + "transaction_id": transaction_id, + } + result["failure_code"] = failure["code"] + result["failure_at"] = failure["failed_at"] + result["failure_evidence_sha256"] = apply.batch._sha256_json(failure) + if self.tamper_status == "extra_field": + result["unexpected"] = True + if self.tamper_status == "request_sha256": + result["request_sha256"] = "f" * 64 + if self.tamper_status == "state_version": + result["state_version"] -= 1 + if self.tamper_status == "finalized_apply_sha" and state == "finalized": + result["apply_receipt_sha256"] = "f" * 64 + if self.tamper_status == "state_downgrade" and state == "finalized": + result.update( + { + "state": "committed_and_verified", + "state_version": apply._TRANSACTION_STATE_VERSION[ + "committed_and_verified" + ], + "no_go": True, + "apply_receipt": None, + "apply_receipt_sha256": None, + "finalized_at": None, + "status_updated_at": self.now.isoformat(timespec="seconds"), + } + ) + return 200, result @contextmanager @@ -1218,7 +1290,7 @@ def test_expired_challenge_is_rejected_before_initial_dispatch( ).fetchone()[0] finally: ledger.close() - assert operation_count == 0 + assert operation_count == 1 assert service.apply_calls == 0 @@ -1332,7 +1404,84 @@ def test_restart_recovers_committed_dispatch_after_short_challenge_expires( assert execution["schema_version"] == apply.EXECUTION_EVIDENCE_SCHEMA_VERSION assert state == "online_committed_pending_countersign" + # Status v3 contains the full signed receipt pair, so restart recovery does + # not need to retransmit an already committed apply. + assert service.apply_calls == 1 + assert service.rollback_calls == 0 + + +def test_lost_apply_response_recovers_from_full_status_v3( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC)) + service = _service(material) + + class LoseApplyResponse: + def __init__(self, delegate) -> None: + self.delegate = delegate + self.lost = False + + def request(self, method, path, *, payload, headers): + response = self.delegate.request(method, path, payload=payload, headers=headers) + if method == "POST" and path.endswith("/apply") and not self.lost: + self.lost = True + raise RuntimeError("simulated lost apply response") + return response + + with _serve(service) as url: + client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True) + ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3") + try: + execution = _execute(material, LoseApplyResponse(client), ledger) + state = ledger.connection.execute( + "SELECT state FROM rc06_apply_operations" + ).fetchone()[0] + finally: + ledger.close() + assert execution["schema_version"] == apply.EXECUTION_EVIDENCE_SCHEMA_VERSION + assert state == "online_committed_pending_countersign" + assert service.apply_calls == 1 + assert service.rollback_calls == 0 + + +def test_irreversible_failure_only_replays_exact_apply_to_recover_evidence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC)) + service = _service(material) + + class LoseEvidenceAfterContractCommit: + def __init__(self, delegate) -> None: + self.delegate = delegate + self.failed = False + + def request(self, method, path, *, payload, headers): + response = self.delegate.request(method, path, payload=payload, headers=headers) + if method == "POST" and path.endswith("/apply") and not self.failed: + self.failed = True + service.state = "irreversible_failure" + raise RuntimeError("simulated post-commit KMS failure") + return response + + ledger_path = tmp_path / "ledger.sqlite3" + with _serve(service) as url: + client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True) + ledger = apply.ApplyLedger(ledger_path) + try: + with pytest.raises(apply.ApplyPublicationError, match="rollback is forbidden"): + _execute(material, LoseEvidenceAfterContractCommit(client), ledger) + finally: + ledger.close() + assert service.rollback_calls == 0 + material["now"] += timedelta(minutes=10) + ledger = apply.ApplyLedger(ledger_path) + try: + execution = _execute(material, client, ledger) + finally: + ledger.close() + assert execution["schema_version"] == apply.EXECUTION_EVIDENCE_SCHEMA_VERSION assert service.apply_calls == 2 + assert service.state == "committed_and_verified" assert service.rollback_calls == 0 @@ -1348,14 +1497,17 @@ def test_terminal_commit_verification_failure_never_rolls_back( client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True) ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3") try: - with pytest.raises(apply.ApplyPublicationError, match="rollback is forbidden"): + with pytest.raises( + apply.ApplyPublicationError, + match="commit status is unknown; no rollback was attempted", + ): _execute(material, client, ledger) state = ledger.connection.execute( "SELECT state FROM rc06_apply_operations" ).fetchone()[0] finally: ledger.close() - assert state == "committed_evidence_invalid_no_go" + assert state == "commit_status_unknown" assert service.state == "committed_and_verified" assert service.rollback_calls == 0 @@ -1488,6 +1640,138 @@ def test_finalize_failure_remains_no_go_and_never_rolls_back( assert service.rollback_calls == 0 +def test_lost_finalize_response_recovers_exact_finalized_status( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC)) + service = _service(material) + + class LoseFinalizeResponse: + def __init__(self, delegate) -> None: + self.delegate = delegate + self.lost = False + + def request(self, method, path, *, payload, headers): + response = self.delegate.request(method, path, payload=payload, headers=headers) + if method == "POST" and path.endswith("/finalize") and not self.lost: + self.lost = True + raise RuntimeError("simulated lost finalize response") + return response + + with _serve(service) as url: + client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True) + ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3") + try: + execution = _execute(material, client, ledger) + at = material["now"] + timedelta(minutes=10) + receipt = _countersign(material, execution, at) + final = _finalize( + material, + LoseFinalizeResponse(client), + ledger, + execution, + receipt, + at, + ) + finally: + ledger.close() + assert final["online_finalize_status"]["state"] == "finalized" + assert final["apply_receipt"] == receipt + assert service.finalize_calls == 1 + assert service.rollback_calls == 0 + + +def test_finalized_remote_recovers_after_local_ledger_loss_without_redispatch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC)) + service = _service(material) + with _serve(service) as url: + client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True) + first = apply.ApplyLedger(tmp_path / "first.sqlite3") + try: + execution = _execute(material, client, first) + at = material["now"] + timedelta(minutes=10) + receipt = _countersign(material, execution, at) + original_final = _finalize(material, client, first, execution, receipt, at) + finally: + first.close() + + material["now"] = at + recovered = apply.ApplyLedger(tmp_path / "replacement.sqlite3") + try: + recovered_execution = _execute(material, client, recovered) + recovered_final = _finalize( + material, + client, + recovered, + recovered_execution, + receipt, + at, + ) + finally: + recovered.close() + assert recovered_execution == execution + assert recovered_final == original_final + assert service.apply_calls == 1 + assert service.finalize_calls == 1 + assert service.rollback_calls == 0 + + +@pytest.mark.parametrize("tamper_status", ("state_downgrade", "finalized_apply_sha")) +def test_finalized_status_downgrade_or_receipt_tamper_preserves_apply_evidence( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + tamper_status: str, +) -> None: + material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC)) + service = _service(material) + with _serve(service) as url: + client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True) + ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3") + try: + execution = _execute(material, client, ledger) + service.tamper_status = tamper_status + at = material["now"] + timedelta(minutes=10) + receipt = _countersign(material, execution, at) + with pytest.raises(apply.ApplyPublicationError): + _finalize(material, client, ledger, execution, receipt, at) + state, stored = ledger.load( + material["idempotency_key"], execution["request_sha256"] + ) + finally: + ledger.close() + assert state == "finalize_failed_no_go" + assert stored == execution + assert service.rollback_calls == 0 + + +@pytest.mark.parametrize("tamper_status", ("extra_field", "request_sha256", "state_version")) +def test_status_v3_rejects_exact_field_identity_and_version_tamper( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + tamper_status: str, +) -> None: + material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC)) + service = _service(material, tamper_status=tamper_status) + with _serve(service) as url: + client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True) + ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3") + try: + with pytest.raises( + apply.ApplyPublicationError, + match="commit status is unknown; no rollback was attempted", + ): + _execute(material, client, ledger) + state = ledger.connection.execute( + "SELECT state FROM rc06_apply_operations" + ).fetchone()[0] + finally: + ledger.close() + assert state == "commit_status_unknown" + assert service.rollback_calls == 0 + + @pytest.mark.parametrize("mode", ["missing", "expired", "untrusted_keyring"]) def test_legal_source_review_is_fail_closed_before_http( tmp_path: Path,