|
| 1 | +"""Fail-closed, local consumption of an immutable recovery state ledger. |
| 2 | +
|
| 3 | +This module deliberately has no Google Cloud, GitHub, Cloud Run, broker, or |
| 4 | +order client. The deployment workflow may opt in by supplying a locally |
| 5 | +downloaded ledger file. Without that explicit path, callers retain their |
| 6 | +legacy runtime-target behaviour exactly. |
| 7 | +""" |
| 8 | + |
| 9 | +from __future__ import annotations |
| 10 | + |
| 11 | +import copy |
| 12 | +import json |
| 13 | +from collections.abc import Mapping |
| 14 | +from pathlib import Path |
| 15 | + |
| 16 | +from quant_platform_kit.common.live_continuity import build_live_continuity |
| 17 | +from quant_platform_kit.common.reconciliation_recovery import ReconciliationRecoveryTransitionPlan |
| 18 | + |
| 19 | + |
| 20 | +RECOVERY_STATE_LEDGER_SCHEMA_VERSION = "ibkr_reconciliation_recovery_state_ledger.v1" |
| 21 | +RECOVERY_STATE_LEDGER_PATH_ENV = "IBKR_RECONCILIATION_RECOVERY_STATE_LEDGER_PATH" |
| 22 | + |
| 23 | + |
| 24 | +def _ledger_service_name(ledger: Mapping[str, object]) -> str: |
| 25 | + service_name = ledger.get("service_name") |
| 26 | + if not isinstance(service_name, str) or service_name != service_name.strip() or not 3 <= len(service_name) <= 127: |
| 27 | + raise ValueError("reconciliation recovery state ledger service_name is invalid") |
| 28 | + return service_name |
| 29 | + |
| 30 | + |
| 31 | +def apply_recovery_state_ledger( |
| 32 | + *, |
| 33 | + runtime_target: Mapping[str, object], |
| 34 | + ledger: Mapping[str, object], |
| 35 | +) -> tuple[dict[str, object], dict[str, str]]: |
| 36 | + """Return the one allowed target transition and its five fixed digests. |
| 37 | +
|
| 38 | + A ledger contains only a QPK transition plan. The desired target is never |
| 39 | + supplied by the ledger: it is derived from the current target by changing |
| 40 | + exactly ``live_continuity.state`` after every frozen-baseline precondition |
| 41 | + has been revalidated. |
| 42 | + """ |
| 43 | + |
| 44 | + required = {"schema_version", "recovery_id", "service_name", "transition_plan"} |
| 45 | + if not isinstance(ledger, Mapping) or set(ledger) != required: |
| 46 | + raise ValueError("reconciliation recovery state ledger has invalid fields") |
| 47 | + if ledger.get("schema_version") != RECOVERY_STATE_LEDGER_SCHEMA_VERSION: |
| 48 | + raise ValueError("unsupported reconciliation recovery state ledger schema") |
| 49 | + ledger_service_name = _ledger_service_name(ledger) |
| 50 | + |
| 51 | + raw_plan = ledger.get("transition_plan") |
| 52 | + if not isinstance(raw_plan, Mapping): |
| 53 | + raise ValueError("reconciliation recovery state ledger is missing transition_plan") |
| 54 | + plan = ReconciliationRecoveryTransitionPlan.from_dict(raw_plan) |
| 55 | + if str(ledger.get("recovery_id") or "").strip() != plan.recovery_id: |
| 56 | + raise ValueError("reconciliation recovery state ledger recovery_id mismatch") |
| 57 | + |
| 58 | + target = copy.deepcopy(dict(runtime_target)) |
| 59 | + if str(target.get("platform_id") or "").strip().lower() != "ibkr": |
| 60 | + raise ValueError("reconciliation recovery state ledger only supports ibkr targets") |
| 61 | + if str(target.get("service_name") or "").strip() != ledger_service_name: |
| 62 | + raise ValueError("reconciliation recovery state ledger service_name mismatch") |
| 63 | + continuity_payload = target.get("live_continuity") |
| 64 | + continuity = build_live_continuity(continuity_payload) |
| 65 | + continuity.assert_matches_target(target) |
| 66 | + if continuity.state != plan.expected_live_continuity_state: |
| 67 | + raise ValueError("reconciliation recovery state ledger current continuity state mismatch") |
| 68 | + if continuity.baseline_id != plan.baseline_id: |
| 69 | + raise ValueError("reconciliation recovery state ledger baseline_id mismatch") |
| 70 | + if continuity.baseline_target_sha256 != plan.baseline_target_sha256: |
| 71 | + raise ValueError("reconciliation recovery state ledger baseline digest mismatch") |
| 72 | + |
| 73 | + next_continuity = continuity.to_dict() |
| 74 | + next_continuity["state"] = plan.next_live_continuity_state |
| 75 | + target["live_continuity"] = next_continuity |
| 76 | + return target, dict(plan.expected_digests) |
| 77 | + |
| 78 | + |
| 79 | +def apply_recovery_state_ledger_from_env( |
| 80 | + *, |
| 81 | + runtime_target: Mapping[str, object], |
| 82 | + env: Mapping[str, str], |
| 83 | +) -> tuple[dict[str, object], dict[str, str] | None]: |
| 84 | + """Apply an explicitly supplied local ledger, or retain the original target.""" |
| 85 | + |
| 86 | + raw_path = str(env.get(RECOVERY_STATE_LEDGER_PATH_ENV) or "").strip() |
| 87 | + if not raw_path: |
| 88 | + return dict(runtime_target), None |
| 89 | + try: |
| 90 | + value = json.loads(Path(raw_path).read_text(encoding="utf-8")) |
| 91 | + except OSError as exc: |
| 92 | + raise ValueError("reconciliation recovery state ledger cannot be read") from exc |
| 93 | + except json.JSONDecodeError as exc: |
| 94 | + raise ValueError("reconciliation recovery state ledger is not valid JSON") from exc |
| 95 | + if not isinstance(value, Mapping): |
| 96 | + raise ValueError("reconciliation recovery state ledger must be a JSON object") |
| 97 | + ledger_service_name = _ledger_service_name(value) |
| 98 | + target_service_name = str(runtime_target.get("service_name") or "").strip() |
| 99 | + if target_service_name != ledger_service_name: |
| 100 | + return dict(runtime_target), None |
| 101 | + target, expected_digests = apply_recovery_state_ledger( |
| 102 | + runtime_target=runtime_target, |
| 103 | + ledger=value, |
| 104 | + ) |
| 105 | + return target, expected_digests |
| 106 | + |
| 107 | + |
| 108 | +__all__ = [ |
| 109 | + "RECOVERY_STATE_LEDGER_PATH_ENV", |
| 110 | + "RECOVERY_STATE_LEDGER_SCHEMA_VERSION", |
| 111 | + "apply_recovery_state_ledger", |
| 112 | + "apply_recovery_state_ledger_from_env", |
| 113 | +] |
0 commit comments