341 lines
19 KiB
Python
341 lines
19 KiB
Python
"""Frozen XML delivery primitive for a future ARR source adapter.
|
|
|
|
Not wired to Web, a scheduler, credentials or OHIP. A capture binding records
|
|
provenance; it does NOT certify API-to-report equivalence. Only synthetic XML is
|
|
used in current acceptance tests. Callers must establish that mapping separately.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
import tempfile
|
|
|
|
from arr_ingestion.contracts import (
|
|
ARTIFACT_ROLES, DELIVERY_SCHEMA_VERSION, RESULT_SCHEMA_VERSION,
|
|
DeliveryEnvelope, DATA_RESULT_SCHEMA_VERSION, artifact_roles_for_schema,
|
|
)
|
|
from arr_ingestion.repository import IngestionRepository, JobRegistration
|
|
from arr_ingestion.service import IngestionService
|
|
from arr_ingestion.validation import DeliveryValidator, ProcessorPolicy
|
|
from arr_processing.local import LocalDailyProcessor
|
|
from arr_storage.contracts import CANONICAL_OBJECT_FILENAMES, role_limit
|
|
from arr_storage.filesystem import FilesystemObjectBackend
|
|
from arr_storage.store import ManagedObjectStore
|
|
from arr_web.contracts import validate_upload_filename, validate_xml_payload
|
|
|
|
from . import collect_arr_source as source
|
|
from .audit_arr_capture import protected_read
|
|
from .capture_job import atomic_json, fingerprint, job_lock, private_directory, sync_directory
|
|
|
|
|
|
VERSION = "arr-frozen-xml-handoff/v1"
|
|
DATA_VERSION = "arr-frozen-data-handoff/v1"
|
|
FREEZE_INTENT_VERSION = "arr-freeze-intent/v1"
|
|
SHA256 = re.compile(r"[0-9a-f]{64}")
|
|
require = source.require
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CaptureBinding:
|
|
batch_id: str
|
|
hotel_id: str
|
|
arrival_date: str
|
|
manifest_sha256: str
|
|
adapter_contract: str
|
|
|
|
def validate(self) -> None:
|
|
source.Options(self.arrival_date, self.hotel_id).validate()
|
|
require(isinstance(self.batch_id, str) and bool(re.fullmatch(r"[a-z0-9][a-z0-9_-]{0,63}", self.batch_id)),
|
|
"invalid_batch_id")
|
|
require(isinstance(self.manifest_sha256, str) and bool(SHA256.fullmatch(self.manifest_sha256)),
|
|
"invalid_manifest_pin")
|
|
require(isinstance(self.adapter_contract, str) and bool(re.fullmatch(r"[a-z0-9][a-z0-9._/-]{0,95}", self.adapter_contract)),
|
|
"invalid_adapter_contract")
|
|
|
|
@property
|
|
def job_id(self) -> str:
|
|
# Do not include source hash/rules: a conflicting retry must not choose a new job.
|
|
return "arrbatch-" + fingerprint({"version": VERSION, "service": source.SERVICE,
|
|
"application": source.APPLICATION, "hotel": self.hotel_id, "batch": self.batch_id})[:48]
|
|
|
|
|
|
def _hash(raw: bytes) -> str:
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def _write(path: Path, raw: bytes) -> None:
|
|
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
|
|
with os.fdopen(fd, "wb") as handle:
|
|
handle.write(raw)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
|
|
|
|
def _directory(path: Path) -> None:
|
|
info = path.lstat()
|
|
require(stat.S_ISDIR(info.st_mode) and info.st_uid == os.getuid()
|
|
and stat.S_IMODE(info.st_mode) == 0o700, "unsafe_handoff_directory")
|
|
|
|
|
|
def _document(path: Path) -> dict:
|
|
return source.strict_json(protected_read(path, 1024 * 1024))
|
|
|
|
|
|
def _identity(binding: CaptureBinding, filename: str, payload: bytes, policy: ProcessorPolicy, *, source_role="source_xml") -> dict:
|
|
binding.validate()
|
|
return {"version": DATA_VERSION if source_role == "source_data" else VERSION, "binding": vars(binding), "job_id": binding.job_id,
|
|
"application_id": source.APPLICATION, "service_url": source.SERVICE,
|
|
"uploaded_filename": filename, "source_sha256": _hash(payload), "source_byte_size": len(payload),
|
|
"processor_version": policy.processor_version, "rule_set_sha256": policy.rule_set_sha256}
|
|
|
|
|
|
def _read_package(directory: Path, expected_pin: str | None = None):
|
|
"""Check the separately committed pin, envelope and every artifact before I/O."""
|
|
_directory(directory)
|
|
ready = _document(directory / "ready.json")
|
|
pin = ready.get("manifest_sha256")
|
|
require(isinstance(pin, str) and bool(SHA256.fullmatch(pin)), "invalid_prepared_pin")
|
|
require(expected_pin is None or pin == expected_pin, "prepared_pin_mismatch")
|
|
return _verify_frozen_package(directory, pin)
|
|
|
|
|
|
def _verify_frozen_package(directory: Path, pin: str):
|
|
"""Verify bytes against an independently persisted pin, not a hash inferred from them."""
|
|
_directory(directory)
|
|
identity = _document(directory / "identity.json")
|
|
binding = CaptureBinding(**identity["binding"])
|
|
binding.validate()
|
|
require(identity.get("version") in {VERSION, DATA_VERSION} and identity.get("job_id") == binding.job_id
|
|
and identity.get("application_id") == source.APPLICATION and identity.get("service_url") == source.SERVICE,
|
|
"handoff_identity_mismatch")
|
|
frozen = directory / "frozen"
|
|
_directory(frozen)
|
|
_directory(frozen / "artifacts")
|
|
raw_manifest = protected_read(frozen / "manifest.json", 1024 * 1024)
|
|
require(_hash(raw_manifest) == pin, "prepared_manifest_changed")
|
|
manifest = source.strict_json(raw_manifest)
|
|
require(manifest.get("version") == identity["version"] and fingerprint(manifest.get("identity")) == fingerprint(identity),
|
|
"prepared_identity_changed")
|
|
raw_envelope = protected_read(frozen / "delivery.json", 1024 * 1024)
|
|
require(_hash(raw_envelope) == manifest.get("delivery_sha256"), "prepared_delivery_changed")
|
|
envelope = DeliveryEnvelope.from_dict(source.strict_json(raw_envelope))
|
|
require(envelope.job_id == binding.job_id and envelope.attempt_no == 1
|
|
and envelope.delivery_id == f"local-{binding.job_id}-a1"
|
|
and envelope.processor_version == identity["processor_version"]
|
|
and envelope.rule_set_sha256 == identity["rule_set_sha256"], "prepared_delivery_mismatch")
|
|
expected_files = set()
|
|
artifacts = {}
|
|
for role, ref in envelope.artifacts.items():
|
|
if ref is None:
|
|
continue
|
|
filename = CANONICAL_OBJECT_FILENAMES[role]
|
|
expected_files.add(filename)
|
|
path = frozen / "artifacts" / filename
|
|
raw = protected_read(path, role_limit(role))
|
|
require(_hash(raw) == ref.sha256 and len(raw) == ref.byte_size, "prepared_artifact_changed")
|
|
artifacts[role] = path
|
|
require({path.name for path in (frozen / "artifacts").iterdir()} == expected_files,
|
|
"unexpected_prepared_artifact")
|
|
require({path.name for path in frozen.iterdir()} == {"artifacts", "manifest.json", "delivery.json"},
|
|
"unexpected_prepared_file")
|
|
ref = envelope.source
|
|
require((envelope.result_schema_version == DATA_RESULT_SCHEMA_VERSION) == (identity["version"] == DATA_VERSION), "prepared_source_kind_mismatch")
|
|
require(ref is not None and ref.sha256 == identity["source_sha256"]
|
|
and ref.byte_size == identity["source_byte_size"], "prepared_source_mismatch")
|
|
return identity, envelope, raw_envelope, artifacts, pin
|
|
|
|
|
|
def _freeze_intent_pin(directory: Path, identity: dict) -> str | None:
|
|
path = directory / "freeze-intent.json"
|
|
if not os.path.lexists(path):
|
|
return None
|
|
intent = _document(path)
|
|
require(set(intent) == {"version", "identity_sha256", "manifest_sha256"}
|
|
and intent["version"] == FREEZE_INTENT_VERSION
|
|
and intent["identity_sha256"] == fingerprint(identity), "freeze_intent_context_mismatch")
|
|
pin = intent["manifest_sha256"]
|
|
require(isinstance(pin, str) and bool(SHA256.fullmatch(pin)), "invalid_freeze_intent_pin")
|
|
return pin
|
|
|
|
|
|
def prepare(
|
|
root: Path, binding: CaptureBinding, original_filename: str, payload: bytes,
|
|
policy: ProcessorPolicy, *, processor: LocalDailyProcessor | None = None,
|
|
source_role: str = "source_xml",
|
|
) -> dict:
|
|
"""Freeze source, processor outputs and verified envelope with no remote writes.
|
|
|
|
The caller supplies a separately accepted adapter's XML. This function cannot
|
|
establish that it represents the capture. No production adapter is supplied.
|
|
New packages persist a validated manifest pin before publication, allowing
|
|
restart to finish the ready checkpoint without reprocessing frozen bytes.
|
|
"""
|
|
require(source_role in {"source_xml", "source_data"}, "invalid_source_role")
|
|
schema = DATA_RESULT_SCHEMA_VERSION if source_role == "source_data" else RESULT_SCHEMA_VERSION
|
|
roles = artifact_roles_for_schema(schema)
|
|
source_filename = CANONICAL_OBJECT_FILENAMES[source_role]
|
|
if source_role == "source_xml":
|
|
filename = validate_upload_filename(original_filename)
|
|
validate_xml_payload(payload)
|
|
else:
|
|
filename = original_filename
|
|
require(filename == "ARR.json" and 0 < len(payload) <= role_limit(source_role), "invalid_data_source")
|
|
document = source.strict_json(payload)
|
|
require(document.get("version") == "arr-ohip-data/v1"
|
|
and document.get("report_date") == binding.arrival_date
|
|
and document.get("hotel_id") == binding.hotel_id, "data_source_context_mismatch")
|
|
identity = _identity(binding, filename, payload, policy, source_role=source_role)
|
|
root = Path(root)
|
|
require(not root.resolve().is_relative_to(Path(__file__).resolve().parents[2]),
|
|
"handoff_store_must_be_outside_repository")
|
|
private_directory(root)
|
|
directory = root / binding.job_id
|
|
private_directory(directory)
|
|
sync_directory(root)
|
|
with job_lock(directory):
|
|
identity_path = directory / "identity.json"
|
|
if not os.path.lexists(identity_path):
|
|
require(not os.path.lexists(directory / "frozen") and not os.path.lexists(directory / "ready.json"),
|
|
"orphan_handoff_package")
|
|
atomic_json(identity_path, identity, replace=False)
|
|
require(fingerprint(_document(identity_path)) == fingerprint(identity), "handoff_request_conflict")
|
|
if os.path.lexists(directory / "ready.json"):
|
|
*_, pin = _read_package(directory)
|
|
return {"job_id": binding.job_id, "directory": str(directory), "manifest_sha256": pin,
|
|
"status": "prepared", "reused": True, "source_mapping_verified": False}
|
|
pending_pin = _freeze_intent_pin(directory, identity)
|
|
if os.path.lexists(directory / "frozen"):
|
|
# New packages have a durable pin BEFORE the atomic directory rename.
|
|
# Legacy unpinned packages still fail closed; never adopt their own hash.
|
|
require(pending_pin is not None, "uncommitted_handoff_package")
|
|
_verify_frozen_package(directory, pending_pin)
|
|
sync_directory(directory)
|
|
atomic_json(directory / "ready.json", {"manifest_sha256": pending_pin}, replace=False)
|
|
return {"job_id": binding.job_id, "directory": str(directory), "manifest_sha256": pending_pin,
|
|
"status": "prepared", "reused": True, "source_mapping_verified": False}
|
|
with tempfile.TemporaryDirectory(prefix=".preparing-", dir=directory) as temporary:
|
|
work = Path(temporary)
|
|
candidate = work / "candidate"
|
|
private_directory(candidate)
|
|
files = candidate / "artifacts"
|
|
private_directory(files)
|
|
source_path = files / source_filename
|
|
_write(source_path, payload)
|
|
local_store = ManagedObjectStore(FilesystemObjectBackend(work / "objects", create=True))
|
|
source_object = local_store.upload_committed(job_id=binding.job_id, attempt_no=1,
|
|
role=source_role, source=source_path, original_filename=source_filename)
|
|
materialized = work / "input" / source_filename
|
|
local_store.materialize(source_object.object_key, materialized, source_object.byte_size)
|
|
processed = (processor or LocalDailyProcessor(policy)).run(materialized, work / "outputs")
|
|
references = {role: None for role in roles}
|
|
references[source_role] = source_object.to_artifact_ref().to_dict()
|
|
for role, path in processed.artifacts.items():
|
|
require(role in roles and role not in {source_role, "manual_override_json"},
|
|
"unexpected_processor_artifact")
|
|
stored = local_store.upload_committed(job_id=binding.job_id, attempt_no=1,
|
|
role=role, source=path, original_filename=path.name)
|
|
references[role] = stored.to_artifact_ref().to_dict()
|
|
# Materialize exactly the immutable bytes referenced by validation.
|
|
frozen_path = files / CANONICAL_OBJECT_FILENAMES[role]
|
|
local_store.materialize(stored.object_key, frozen_path, stored.byte_size)
|
|
os.chmod(frozen_path, 0o600)
|
|
with frozen_path.open("rb") as handle:
|
|
os.fsync(handle.fileno())
|
|
envelope = DeliveryEnvelope.from_dict({"delivery_schema_version": DELIVERY_SCHEMA_VERSION,
|
|
"delivery_id": f"local-{binding.job_id}-a1", "job_id": binding.job_id, "attempt_no": 1,
|
|
"status": processed.status, "processor_version": policy.processor_version,
|
|
"rule_set_sha256": policy.rule_set_sha256, "result_schema_version": schema,
|
|
"business_date": processed.business_date.isoformat() if processed.business_date else None,
|
|
"artifacts": references})
|
|
raw_envelope = source.json_bytes(envelope.to_dict())
|
|
verified = DeliveryValidator(local_store, policy).validate(raw_envelope)
|
|
require(verified.envelope.business_date is None
|
|
or verified.envelope.business_date.isoformat() == binding.arrival_date, "handoff_date_mismatch")
|
|
_write(candidate / "delivery.json", raw_envelope)
|
|
manifest = {"version": identity["version"], "identity": identity, "delivery_sha256": _hash(raw_envelope)}
|
|
raw_manifest = source.json_bytes(manifest)
|
|
_write(candidate / "manifest.json", raw_manifest)
|
|
sync_directory(files)
|
|
sync_directory(candidate)
|
|
# Pin only independently validated output. If interrupted before the
|
|
# rename, no package was published and local preparation may rerun.
|
|
# Once frozen exists this intent is immutable and recovery verifies it.
|
|
intent_path = directory / "freeze-intent.json"
|
|
atomic_json(intent_path, {"version": FREEZE_INTENT_VERSION,
|
|
"identity_sha256": fingerprint(identity), "manifest_sha256": _hash(raw_manifest)},
|
|
replace=os.path.lexists(intent_path))
|
|
os.rename(candidate, directory / "frozen")
|
|
sync_directory(directory)
|
|
atomic_json(directory / "ready.json", {"manifest_sha256": _hash(raw_manifest)}, replace=False)
|
|
*_, pin = _read_package(directory)
|
|
return {"job_id": binding.job_id, "directory": str(directory), "manifest_sha256": pin,
|
|
"status": "prepared", "reused": False, "source_mapping_verified": False}
|
|
|
|
|
|
def deliver(
|
|
directory: Path, expected_manifest_sha256: str, *, object_store: ManagedObjectStore,
|
|
repository: IngestionRepository, service: IngestionService,
|
|
expected_binding: CaptureBinding | None = None, expected_policy: ProcessorPolicy | None = None,
|
|
) -> dict:
|
|
"""Deliver the frozen envelope, including after an unknown commit outcome.
|
|
|
|
Dependencies must be explicitly supplied. There is no CLI, runtime wiring or
|
|
credential loading. Never turn an ambiguous exception into a failed DB job:
|
|
retry this same package; PostgreSQL's delivery transaction is authoritative.
|
|
The returned receipt acknowledges this delivery, not later review/job state.
|
|
An orchestrator should supply its frozen capture binding and current policy;
|
|
both are compared with the actual package under the delivery lock before
|
|
any destination writes. Legacy standalone callers can remain pin-only.
|
|
"""
|
|
require(isinstance(expected_manifest_sha256, str) and bool(SHA256.fullmatch(expected_manifest_sha256)),
|
|
"invalid_prepared_pin")
|
|
if expected_binding is not None:
|
|
require(type(expected_binding) is CaptureBinding, "invalid_expected_handoff_binding")
|
|
expected_binding.validate()
|
|
if expected_policy is not None:
|
|
require(type(expected_policy) is ProcessorPolicy, "invalid_expected_handoff_policy")
|
|
directory = Path(directory)
|
|
_directory(directory)
|
|
with job_lock(directory):
|
|
identity, envelope, raw_envelope, artifacts, pin = _read_package(directory, expected_manifest_sha256)
|
|
if expected_binding is not None:
|
|
require(fingerprint(identity["binding"]) == fingerprint(vars(expected_binding)),
|
|
"handoff_expected_binding_mismatch")
|
|
if expected_policy is not None:
|
|
require(identity["processor_version"] == expected_policy.processor_version
|
|
and identity["rule_set_sha256"] == expected_policy.rule_set_sha256,
|
|
"handoff_expected_policy_mismatch")
|
|
# Check every destination before starting any upload.
|
|
for ref in envelope.artifacts.values():
|
|
if ref is not None:
|
|
address = object_store.key_policy.parse(ref.object_key)
|
|
require(address.job_id == envelope.job_id and address.attempt_no == 1
|
|
and address.state == "committed" and address.role == ref.role,
|
|
"handoff_destination_mismatch")
|
|
for role, path in artifacts.items():
|
|
ref = envelope.artifacts[role]
|
|
stored = object_store.upload_committed(job_id=envelope.job_id, attempt_no=1,
|
|
role=role, source=path, original_filename=ref.original_filename,
|
|
expected_sha256=ref.sha256, expected_byte_size=ref.byte_size)
|
|
require(stored.to_artifact_ref() == ref, "uploaded_handoff_artifact_mismatch")
|
|
ref = envelope.source
|
|
repository.register_job(JobRegistration(job_id=envelope.job_id, source=ref,
|
|
processor_version=envelope.processor_version, rule_set_sha256=envelope.rule_set_sha256,
|
|
attempt_no=1, idempotency_key=fingerprint(identity), uploaded_filename=identity["uploaded_filename"]))
|
|
# Queued -> atomic terminal delivery is supported by the repository. A
|
|
# mark_running call would reject retries after an already-committed result.
|
|
outcome = service.ingest(raw_envelope)
|
|
require(outcome.job_id == envelope.job_id, "handoff_outcome_mismatch")
|
|
receipt = {"version": identity["version"], "job_id": outcome.job_id, "manifest_sha256": pin,
|
|
"delivery_id": envelope.delivery_id, "ingestion_status": outcome.status,
|
|
"business_date": outcome.business_date.isoformat() if outcome.business_date else None,
|
|
"daily_version_id": outcome.daily_version_id, "version_no": outcome.version_no,
|
|
"source_mapping_verified": False}
|
|
# This is a convenience receipt, never a substitute for server-side replay.
|
|
atomic_json(directory / "receipt.json", receipt, replace=True)
|
|
return receipt
|