"""Recoverable post-commit retirement of one legacy directory tree.
|
|
The canonical publisher commit and legacy retirement are deliberately separate
|
transactions. This coordinator may run only after the direct stable-path
|
publisher can be verified read-only as IDEMPOTENT_COMMITTED. Its official
|
receipt directory is the durable ownership anchor; directory rename is the
|
only formal legacy mutation.
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
import csv
|
import hashlib
|
import io
|
import json
|
import os
|
import re
|
import sys
|
import uuid
|
from dataclasses import dataclass
|
from pathlib import Path
|
from types import SimpleNamespace
|
from typing import Any
|
|
from . import publisher as base
|
from .direct_stable_path import _validate_direct_config, _verify_replay
|
|
|
SCHEMA = "SHARED_CONTENT_POSTCOMMIT_ARCHIVE_CONFIG_V1"
|
TERMINAL_SCHEMA = "SHARED_CONTENT_POSTCOMMIT_ARCHIVE_TERMINAL_V1"
|
ANCHOR_SCHEMA = "SHARED_CONTENT_POSTCOMMIT_ARCHIVE_ANCHOR_V1"
|
ROLLBACK_SCHEMA = "SHARED_CONTENT_POSTCOMMIT_ARCHIVE_ROLLBACK_CONFIG_V1"
|
ROLLBACK_TERMINAL_SCHEMA = "SHARED_CONTENT_POSTCOMMIT_ARCHIVE_ROLLBACK_TERMINAL_V1"
|
SHA_RE = re.compile(r"^[0-9A-F]{64}$")
|
MAP_HEADER = (
|
"legacy_path", "canonical_target", "archive_path", "legacy_bytes",
|
"legacy_sha256", "canonical_bytes", "canonical_sha256",
|
"prior_acceptance_mode", "prior_acceptance_reference",
|
"prior_acceptance_audit_id", "migration_gate_audit_id",
|
"expected_final_status", "recovery_basis",
|
)
|
|
|
class ArchiveError(Exception):
|
def __init__(self, code: str, detail: str, exit_code: int = 12):
|
super().__init__(f"{code}:{detail}")
|
self.code = code
|
self.detail = detail
|
self.exit_code = exit_code
|
|
|
class InjectedStop(ArchiveError):
|
pass
|
|
|
@dataclass(frozen=True)
|
class Row:
|
legacy_path: str
|
canonical_target: str
|
archive_path: str
|
legacy_bytes: int
|
legacy_sha256: str
|
canonical_bytes: int
|
canonical_sha256: str
|
|
|
@dataclass(frozen=True)
|
class FormalRow:
|
formal_relative_path: str
|
bytes: int
|
sha256: str
|
|
|
@dataclass(frozen=True)
|
class Plan:
|
config: dict[str, Any]
|
config_path: Path
|
config_data: bytes
|
config_sha256: str
|
root: Path
|
identity: dict[str, str]
|
publisher_config: Path
|
publisher_config_data: bytes
|
source_root: Path
|
target_root: Path
|
receipt_root: Path
|
map_path: Path
|
rows: tuple[Row, ...]
|
source_set_sha256: str
|
canonical_set_sha256: str
|
publisher_terminal: dict[str, Any]
|
publisher_formal_rows: tuple[FormalRow, ...]
|
publisher_formal_set_sha256: str
|
publisher_lock_path: Path
|
publisher_ledger_path: Path
|
publisher_attempt_receipt_root: Path
|
publisher_value: dict[str, Any]
|
test_fault: str
|
|
|
def _read(path: Path) -> bytes:
|
try:
|
return base._read_bytes(path)
|
except base.PublisherError as exc:
|
raise ArchiveError(exc.code, exc.detail, exc.exit_code) from exc
|
|
|
def _sha(data: bytes) -> str:
|
return hashlib.sha256(data).hexdigest().upper()
|
|
|
def _canonical(value: Any) -> bytes:
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n"
|
|
|
def _need_object(value: Any, label: str) -> dict[str, Any]:
|
if not isinstance(value, dict):
|
raise ArchiveError("CONFIG_TYPE", label)
|
return value
|
|
|
def _keys(value: dict[str, Any], exact: set[str], label: str) -> None:
|
if set(value) != exact:
|
raise ArchiveError("CONFIG_KEYS", f"{label}:missing={sorted(exact-set(value))};extra={sorted(set(value)-exact)}")
|
|
|
def _string(value: Any, label: str) -> str:
|
if not isinstance(value, str) or not value or value != value.strip():
|
raise ArchiveError("CONFIG_STRING", label)
|
return value
|
|
|
def _sha_value(value: Any, label: str) -> str:
|
text = _string(value, label)
|
if not SHA_RE.fullmatch(text):
|
raise ArchiveError("CONFIG_SHA256", label)
|
return text
|
|
|
def _integer(value: Any, label: str, *, positive: bool = False) -> int:
|
if isinstance(value, bool) or not isinstance(value, int) or (positive and value <= 0):
|
raise ArchiveError("CONFIG_INTEGER", label)
|
return value
|
|
|
def _safe_relative(value: Any, label: str) -> str:
|
text = _string(value, label).replace("\\", "/")
|
path = Path(text)
|
if path.is_absolute() or not path.parts or any(part in ("", ".", "..") for part in path.parts):
|
raise ArchiveError("CONFIG_PATH", label)
|
return "/".join(path.parts)
|
|
|
def _inside(root: Path, relative: str) -> Path:
|
path = root / Path(relative)
|
root_text = os.path.abspath(root).casefold().rstrip("\\/")
|
path_text = os.path.abspath(path).casefold()
|
if path_text == root_text or not path_text.startswith(root_text + os.sep.casefold()):
|
raise ArchiveError("PATH_ESCAPE", relative)
|
return path
|
|
|
def _relative(path: Path, root: Path) -> str:
|
return os.path.relpath(path, root).replace("\\", "/")
|
|
|
def _identity(path: Path) -> tuple[int, str]:
|
data = _read(path)
|
return len(data), _sha(data)
|
|
|
def _ordinary_dir(path: Path, label: str) -> None:
|
try:
|
state = os.lstat(base._long(path))
|
except OSError as exc:
|
raise ArchiveError("DIRECTORY_UNREADABLE", f"{label}:{exc}") from exc
|
attrs = getattr(state, "st_file_attributes", 0)
|
reparse = bool(attrs & getattr(__import__("stat"), "FILE_ATTRIBUTE_REPARSE_POINT", 0x400))
|
if not __import__("stat").S_ISDIR(state.st_mode) or reparse:
|
raise ArchiveError("DIRECTORY_NOT_ORDINARY", label)
|
|
|
def _ancestor_gate(root: Path, path: Path, label: str, *, include_leaf: bool = True) -> None:
|
root_abs = Path(os.path.abspath(root))
|
path_abs = Path(os.path.abspath(path))
|
current = path_abs if include_leaf else path_abs.parent
|
chain: list[Path] = []
|
while True:
|
chain.append(current)
|
if os.path.normcase(str(current)) == os.path.normcase(str(root_abs)):
|
break
|
if current.parent == current:
|
raise ArchiveError("PATH_OUTSIDE_ROOT", label)
|
current = current.parent
|
for item in reversed(chain):
|
if base._exists(item):
|
_ordinary_dir(item, label)
|
|
|
def _walk(root: Path) -> tuple[str, ...]:
|
try:
|
return tuple(sorted(base._walk_files(root), key=str.casefold))
|
except base.PublisherError as exc:
|
raise ArchiveError(exc.code, exc.detail, exc.exit_code) from exc
|
|
|
def _set_digest(rows: tuple[Row, ...], side: str) -> str:
|
values = []
|
for row in rows:
|
if side == "legacy":
|
values.append({"bytes": row.legacy_bytes, "path": row.legacy_path, "sha256": row.legacy_sha256})
|
elif side == "archive":
|
values.append({"bytes": row.legacy_bytes, "path": row.archive_path, "sha256": row.legacy_sha256})
|
else:
|
values.append({"bytes": row.canonical_bytes, "path": row.canonical_target, "sha256": row.canonical_sha256})
|
return _sha(_canonical(values))
|
|
|
def _load_rows(plan_root: Path, map_path: Path, source_root: Path, target_root: Path) -> tuple[Row, ...]:
|
try:
|
reader = csv.DictReader(io.StringIO(_read(map_path).decode("utf-8"), newline=""))
|
except Exception as exc:
|
raise ArchiveError("MAP_PARSE", str(exc)) from exc
|
if tuple(reader.fieldnames or ()) != MAP_HEADER:
|
raise ArchiveError("MAP_HEADER", str(reader.fieldnames))
|
rows: list[Row] = []
|
seen_legacy: set[str] = set()
|
seen_archive: set[str] = set()
|
seen_canonical: set[str] = set()
|
source_prefix = _relative(source_root, plan_root).rstrip("/") + "/"
|
target_prefix = _relative(target_root, plan_root).rstrip("/") + "/"
|
for number, raw in enumerate(reader, start=2):
|
if set(raw) != set(MAP_HEADER) or any(value is None for value in raw.values()):
|
raise ArchiveError("MAP_ROW_SHAPE", str(number))
|
legacy = _safe_relative(raw["legacy_path"], f"map[{number}].legacy_path")
|
archive = _safe_relative(raw["archive_path"], f"map[{number}].archive_path")
|
canonical = _safe_relative(raw["canonical_target"], f"map[{number}].canonical_target")
|
if not legacy.casefold().startswith(source_prefix.casefold()) or not archive.casefold().startswith(target_prefix.casefold()):
|
raise ArchiveError("MAP_ROOT_BINDING", str(number))
|
if legacy.casefold() in seen_legacy or archive.casefold() in seen_archive or canonical.casefold() in seen_canonical:
|
raise ArchiveError("MAP_DUPLICATE", str(number))
|
seen_legacy.add(legacy.casefold()); seen_archive.add(archive.casefold()); seen_canonical.add(canonical.casefold())
|
try:
|
legacy_bytes = int(raw["legacy_bytes"]); canonical_bytes = int(raw["canonical_bytes"])
|
except ValueError as exc:
|
raise ArchiveError("MAP_INTEGER", str(number)) from exc
|
legacy_sha = _sha_value(raw["legacy_sha256"], f"map[{number}].legacy_sha256")
|
canonical_sha = _sha_value(raw["canonical_sha256"], f"map[{number}].canonical_sha256")
|
if legacy_bytes <= 0 or canonical_bytes <= 0:
|
raise ArchiveError("MAP_BYTES", str(number))
|
if raw["expected_final_status"] != "ARCHIVED_AFTER_PUBLISHER_COMMITTED_READBACK":
|
raise ArchiveError("MAP_FINAL_STATUS", str(number))
|
if not raw["recovery_basis"]:
|
raise ArchiveError("MAP_RECOVERY_BASIS", str(number))
|
rows.append(Row(legacy, canonical, archive, legacy_bytes, legacy_sha, canonical_bytes, canonical_sha))
|
if not rows:
|
raise ArchiveError("MAP_EMPTY", "no rows")
|
return tuple(sorted(rows, key=lambda row: row.legacy_path.casefold()))
|
|
|
def _verify_tree(plan: Plan, side: str) -> None:
|
root = plan.source_root if side == "legacy" else plan.target_root
|
expected = {
|
_relative(_inside(plan.root, row.legacy_path if side == "legacy" else row.archive_path), root): row
|
for row in plan.rows
|
}
|
actual = _walk(root)
|
if tuple(sorted(expected, key=str.casefold)) != actual:
|
raise ArchiveError("ARCHIVE_EXACT_SET", f"{side}:expected={len(expected)};actual={len(actual)}", 30)
|
for relative, row in expected.items():
|
path = root / Path(relative)
|
if _identity(path) != (row.legacy_bytes, row.legacy_sha256):
|
raise ArchiveError("ARCHIVE_IDENTITY", f"{side}:{relative}", 30)
|
|
|
def _verify_canonical(plan: Plan) -> None:
|
for row in plan.rows:
|
if _identity(_inside(plan.root, row.canonical_target)) != (row.canonical_bytes, row.canonical_sha256):
|
raise ArchiveError("CANONICAL_IDENTITY", row.canonical_target, 30)
|
|
|
def _publisher_formal_rows(value: dict[str, Any]) -> tuple[FormalRow, ...]:
|
try:
|
raw_rows = value["expected"]["candidate"]["rows"]
|
except (KeyError, TypeError) as exc:
|
raise ArchiveError("PUBLISHER_FORMAL_ROWS", "missing", 30) from exc
|
if not isinstance(raw_rows, list) or not raw_rows:
|
raise ArchiveError("PUBLISHER_FORMAL_ROWS", "not nonempty list", 30)
|
rows: list[FormalRow] = []
|
seen: set[str] = set()
|
for number, raw in enumerate(raw_rows, 1):
|
if not isinstance(raw, dict):
|
raise ArchiveError("PUBLISHER_FORMAL_ROW", str(number), 30)
|
formal = _safe_relative(raw.get("formal_relative_path"), f"publisher.rows[{number}].formal_relative_path")
|
folded = formal.casefold()
|
if folded in seen:
|
raise ArchiveError("PUBLISHER_FORMAL_DUPLICATE", formal, 30)
|
seen.add(folded)
|
rows.append(FormalRow(
|
formal,
|
_integer(raw.get("bytes"), f"publisher.rows[{number}].bytes", positive=True),
|
_sha_value(raw.get("sha256"), f"publisher.rows[{number}].sha256"),
|
))
|
return tuple(sorted(rows, key=lambda row: row.formal_relative_path.casefold()))
|
|
|
def _publisher_formal_digest(rows: tuple[FormalRow, ...]) -> str:
|
return _sha(_canonical([
|
{"bytes": row.bytes, "path": row.formal_relative_path, "sha256": row.sha256}
|
for row in rows
|
]))
|
|
|
def _verify_publisher_formal(plan: Plan) -> None:
|
if base._exists(plan.publisher_lock_path):
|
raise ArchiveError("PUBLISHER_LOCK_PRESENT", str(plan.publisher_lock_path), 30)
|
for row in plan.publisher_formal_rows:
|
if _identity(_inside(plan.root, row.formal_relative_path)) != (row.bytes, row.sha256):
|
raise ArchiveError("PUBLISHER_FORMAL_IDENTITY", row.formal_relative_path, 30)
|
if _publisher_formal_digest(plan.publisher_formal_rows) != plan.publisher_formal_set_sha256:
|
raise ArchiveError("PUBLISHER_FORMAL_SET", "digest", 30)
|
terminal = plan.publisher_terminal
|
states = tuple(terminal.get("event_states", ()))
|
processes = terminal.get("event_process_identities")
|
expected = plan.publisher_value["expected"]
|
ledger_expected = expected["ledger"]
|
ledger_data = _read(plan.publisher_ledger_path)
|
seed_bytes = _integer(ledger_expected["bytes"], "publisher.expected.ledger.bytes", positive=True)
|
if len(ledger_data) <= seed_bytes:
|
raise ArchiveError("PUBLISHER_EVENT_CHAIN", "ledger has no attempt rows", 30)
|
ledger_base_data = ledger_data[:seed_bytes]
|
if _sha(ledger_base_data) != _sha_value(ledger_expected["sha256"], "publisher.expected.ledger.sha256"):
|
raise ArchiveError("PUBLISHER_EVENT_CHAIN", "ledger prefix", 30)
|
if not isinstance(processes, list):
|
raise ArchiveError("PUBLISHER_EVENT_CHAIN", "processes", 30)
|
proxy = SimpleNamespace(
|
operation_root=plan.root,
|
paths={
|
"ledger_path": plan.publisher_ledger_path,
|
"attempt_receipt_root": plan.publisher_attempt_receipt_root,
|
},
|
ledger_base_data=ledger_base_data,
|
next_event_seq=_integer(ledger_expected["next_event_seq"], "publisher.expected.ledger.next_event_seq", positive=True),
|
next_attempt_seq=_integer(ledger_expected["next_attempt_seq"], "publisher.expected.ledger.next_attempt_seq", positive=True),
|
prior_release_set=plan.publisher_value["expected"]["prior"]["release_set_sha256"],
|
identity=plan.publisher_value["identity"],
|
)
|
try:
|
digest, start, end = base._validate_attempt_event_chain(
|
proxy,
|
states,
|
tuple(processes),
|
expected_digest=terminal["attempt_event_chain_sha256"],
|
expected_time_start=terminal["event_time_start_utc"],
|
expected_time_end=terminal["event_time_end_utc"],
|
)
|
except (base.PublisherError, KeyError, TypeError) as exc:
|
raise ArchiveError("PUBLISHER_EVENT_CHAIN", str(exc), 30) from exc
|
if (digest, start, end) != (
|
terminal["attempt_event_chain_sha256"],
|
terminal["event_time_start_utc"],
|
terminal["event_time_end_utc"],
|
):
|
raise ArchiveError("PUBLISHER_EVENT_CHAIN", "terminal mismatch", 30)
|
if base._exists(plan.publisher_lock_path):
|
raise ArchiveError("PUBLISHER_LOCK_PRESENT", str(plan.publisher_lock_path), 30)
|
|
|
def _fault(plan: Plan, name: str) -> None:
|
if plan.test_fault == name:
|
marker = plan.receipt_root.parent / (plan.receipt_root.name + ".test-fault-" + name)
|
if base._exists(marker):
|
return
|
try:
|
base._create_exclusive(marker, b"consumed\n")
|
except base.PublisherError as exc:
|
raise ArchiveError(exc.code, exc.detail, exc.exit_code) from exc
|
raise InjectedStop("INJECTED_PROCESS_RESTART", name, 20)
|
|
|
def _create_exact(path: Path, data: bytes) -> None:
|
if base._exists(path):
|
if _read(path) != data:
|
raise ArchiveError("RECOVERY_REQUIRED_RECEIPT_PARTIAL", _relative(path, path.parent), 30)
|
return
|
try:
|
base._create_exclusive(path, data)
|
except base.PublisherError as exc:
|
raise ArchiveError(exc.code, exc.detail, exc.exit_code) from exc
|
|
|
def _anchor_values(plan: Plan) -> tuple[bytes, bytes, bytes]:
|
contract = _canonical({
|
"archive_id": plan.identity["archive_id"],
|
"attempt_id": plan.identity["attempt_id"],
|
"batch_id": plan.identity["batch_id"],
|
"canonical_set_sha256": plan.canonical_set_sha256,
|
"case_id": plan.identity["case_id"],
|
"config_bytes": len(plan.config_data),
|
"config_sha256": plan.config_sha256,
|
"legacy_map_bytes": len(_read(plan.map_path)),
|
"legacy_map_sha256": _sha(_read(plan.map_path)),
|
"publisher_config_bytes": len(plan.publisher_config_data),
|
"publisher_config_sha256": _sha(plan.publisher_config_data),
|
"publisher_event_chain_sha256": plan.publisher_terminal["attempt_event_chain_sha256"],
|
"publisher_formal_row_count": len(plan.publisher_formal_rows),
|
"publisher_formal_set_sha256": plan.publisher_formal_set_sha256,
|
"publisher_terminal_bytes": _identity(Path(plan.publisher_terminal["_path"]))[0],
|
"publisher_terminal_sha256": _identity(Path(plan.publisher_terminal["_path"]))[1],
|
"row_count": len(plan.rows),
|
"run_id": plan.identity["run_id"],
|
"schema_version": ANCHOR_SCHEMA,
|
"source_root": _relative(plan.source_root, plan.root),
|
"source_set_sha256": plan.source_set_sha256,
|
"target_root": _relative(plan.target_root, plan.root),
|
"task_id": plan.identity["task_id"],
|
})
|
map_data = _read(plan.map_path)
|
manifest = _canonical({
|
"members": [
|
{"bytes": len(contract), "name": "archive_contract.json", "sha256": _sha(contract)},
|
{"bytes": len(map_data), "name": "legacy_case_path_map.csv", "sha256": _sha(map_data)},
|
],
|
"schema_version": ANCHOR_SCHEMA,
|
})
|
return contract, map_data, manifest
|
|
|
def _prepare_anchor(plan: Plan) -> None:
|
if not base._exists(plan.receipt_root):
|
try:
|
os.mkdir(base._long(plan.receipt_root))
|
except FileExistsError:
|
pass
|
except OSError as exc:
|
raise ArchiveError("RECEIPT_ROOT_CREATE", str(exc)) from exc
|
_ordinary_dir(plan.receipt_root, "receipt_root")
|
_fault(plan, "INTERRUPT_AFTER_RECEIPT_ROOT")
|
contract, map_data, manifest = _anchor_values(plan)
|
_create_exact(plan.receipt_root / "archive_contract.json", contract)
|
_fault(plan, "INTERRUPT_AFTER_CONTRACT")
|
_create_exact(plan.receipt_root / "legacy_case_path_map.csv", map_data)
|
_fault(plan, "INTERRUPT_AFTER_MAP")
|
_create_exact(plan.receipt_root / "anchor_manifest.json", manifest)
|
_fault(plan, "INTERRUPT_AFTER_ANCHOR")
|
|
|
def _verify_anchor(plan: Plan, *, allow_terminal: bool) -> None:
|
contract, map_data, manifest = _anchor_values(plan)
|
expected = {
|
"archive_contract.json": contract,
|
"legacy_case_path_map.csv": map_data,
|
"anchor_manifest.json": manifest,
|
}
|
allowed = set(expected)
|
if allow_terminal:
|
allowed |= {"terminal.tmp.json", "terminal.json"}
|
actual = set(_walk(plan.receipt_root))
|
if not set(expected).issubset(actual) or not actual.issubset(allowed):
|
raise ArchiveError("RECOVERY_REQUIRED_RECEIPT_EXACT_SET", f"actual={sorted(actual)}", 30)
|
for name, data in expected.items():
|
if _read(plan.receipt_root / name) != data:
|
raise ArchiveError("RECOVERY_REQUIRED_RECEIPT_BINDING", name, 30)
|
|
|
def _terminal_value(plan: Plan) -> dict[str, Any]:
|
return {
|
"archive_id": plan.identity["archive_id"],
|
"archive_set_sha256": _set_digest(plan.rows, "archive"),
|
"attempt_id": plan.identity["attempt_id"],
|
"batch_id": plan.identity["batch_id"],
|
"canonical_set_sha256": plan.canonical_set_sha256,
|
"case_id": plan.identity["case_id"],
|
"config_bytes": len(plan.config_data),
|
"config_sha256": plan.config_sha256,
|
"exit_code": 0,
|
"legacy_source_present": False,
|
"publisher_event_chain_sha256": plan.publisher_terminal["attempt_event_chain_sha256"],
|
"publisher_formal_row_count": len(plan.publisher_formal_rows),
|
"publisher_formal_set_sha256": plan.publisher_formal_set_sha256,
|
"publisher_status": "COMMITTED",
|
"row_count": len(plan.rows),
|
"run_id": plan.identity["run_id"],
|
"schema_version": TERMINAL_SCHEMA,
|
"status": "ARCHIVED",
|
"target_present": True,
|
"task_id": plan.identity["task_id"],
|
}
|
|
|
def _publish_terminal(plan: Plan) -> dict[str, Any]:
|
expected = _canonical(_terminal_value(plan))
|
temporary = plan.receipt_root / "terminal.tmp.json"
|
terminal = plan.receipt_root / "terminal.json"
|
if base._exists(terminal):
|
if _read(terminal) != expected or base._exists(temporary):
|
raise ArchiveError("RECOVERY_REQUIRED_TERMINAL_BINDING", "terminal/temp", 30)
|
return _terminal_value(plan)
|
_create_exact(temporary, expected)
|
_fault(plan, "INTERRUPT_AFTER_TERMINAL_TEMP")
|
try:
|
os.rename(base._long(temporary), base._long(terminal))
|
except FileExistsError:
|
if _read(terminal) != expected:
|
raise ArchiveError("RECOVERY_REQUIRED_TERMINAL_RACE", "drift", 30)
|
except OSError as exc:
|
raise ArchiveError("TERMINAL_ATOMIC_RENAME", str(exc), 30) from exc
|
if _read(terminal) != expected or base._exists(temporary):
|
raise ArchiveError("RECOVERY_REQUIRED_TERMINAL_READBACK", "mismatch", 30)
|
return _terminal_value(plan)
|
|
|
def _load_plan(config_path: Path) -> Plan:
|
config_data = _read(config_path)
|
try:
|
config = json.loads(config_data.decode("utf-8"), object_pairs_hook=base._pairs_no_duplicates)
|
except Exception as exc:
|
raise ArchiveError("CONFIG_PARSE", str(exc)) from exc
|
config = _need_object(config, "root")
|
_keys(config, {"schema_version", "identity", "roots", "expected", "test_control"}, "root")
|
if config["schema_version"] != SCHEMA:
|
raise ArchiveError("CONFIG_SCHEMA", str(config["schema_version"]))
|
identity = _need_object(config["identity"], "identity")
|
identity_keys = {"task_id", "case_id", "batch_id", "run_id", "attempt_id", "archive_id", "operator", "archive_gate_audit_id"}
|
_keys(identity, identity_keys, "identity")
|
identity = {key: _string(value, f"identity.{key}") for key, value in identity.items()}
|
roots = _need_object(config["roots"], "roots")
|
root_keys = {"operation_root", "resolved_root", "publisher_config_path", "source_root", "target_root", "receipt_root", "legacy_map_path"}
|
_keys(roots, root_keys, "roots")
|
root = Path(_string(roots["operation_root"], "roots.operation_root"))
|
if not root.is_absolute() or os.path.normcase(os.path.abspath(root)) != os.path.normcase(_string(roots["resolved_root"], "roots.resolved_root")):
|
raise ArchiveError("ROOT_IDENTITY", str(root))
|
_ordinary_dir(root, "operation_root")
|
resolved = {key: _inside(root, _safe_relative(roots[key], f"roots.{key}")) for key in root_keys - {"operation_root", "resolved_root"}}
|
source_root = resolved["source_root"]; target_root = resolved["target_root"]
|
receipt_root = resolved["receipt_root"]; map_path = resolved["legacy_map_path"]
|
for left, right in ((source_root, target_root), (source_root, receipt_root), (target_root, receipt_root)):
|
a = os.path.abspath(left).casefold().rstrip("\\/"); b = os.path.abspath(right).casefold().rstrip("\\/")
|
if a == b or a.startswith(b + os.sep.casefold()) or b.startswith(a + os.sep.casefold()):
|
raise ArchiveError("PATH_OVERLAP", f"{left}|{right}")
|
_ancestor_gate(root, source_root, "source_root", include_leaf=base._exists(source_root))
|
_ancestor_gate(root, target_root, "target_root", include_leaf=base._exists(target_root))
|
_ancestor_gate(root, receipt_root, "receipt_root", include_leaf=base._exists(receipt_root))
|
if base._volume(source_root.parent).upper() != base._volume(target_root.parent).upper():
|
raise ArchiveError("CROSS_VOLUME", f"{source_root}|{target_root}")
|
expected = _need_object(config["expected"], "expected")
|
expected_keys = {"publisher_config_bytes", "publisher_config_sha256", "legacy_map_bytes", "legacy_map_sha256", "row_count", "source_set_sha256", "canonical_set_sha256"}
|
_keys(expected, expected_keys, "expected")
|
publisher_config = resolved["publisher_config_path"]
|
publisher_data = _read(publisher_config)
|
if (len(publisher_data), _sha(publisher_data)) != (
|
_integer(expected["publisher_config_bytes"], "expected.publisher_config_bytes", positive=True),
|
_sha_value(expected["publisher_config_sha256"], "expected.publisher_config_sha256"),
|
):
|
raise ArchiveError("PUBLISHER_CONFIG_IDENTITY", str(publisher_config))
|
if _identity(map_path) != (
|
_integer(expected["legacy_map_bytes"], "expected.legacy_map_bytes", positive=True),
|
_sha_value(expected["legacy_map_sha256"], "expected.legacy_map_sha256"),
|
):
|
raise ArchiveError("MAP_IDENTITY", str(map_path))
|
rows = _load_rows(root, map_path, source_root, target_root)
|
if len(rows) != _integer(expected["row_count"], "expected.row_count", positive=True):
|
raise ArchiveError("ROW_COUNT", str(len(rows)))
|
source_digest = _set_digest(rows, "legacy")
|
canonical_digest = _set_digest(rows, "canonical")
|
if source_digest != _sha_value(expected["source_set_sha256"], "expected.source_set_sha256"):
|
raise ArchiveError("SOURCE_SET_DIGEST", source_digest)
|
if canonical_digest != _sha_value(expected["canonical_set_sha256"], "expected.canonical_set_sha256"):
|
raise ArchiveError("CANONICAL_SET_DIGEST", canonical_digest)
|
test_control = _need_object(config["test_control"], "test_control")
|
_keys(test_control, {"environment", "fault"}, "test_control")
|
environment = _string(test_control["environment"], "test_control.environment")
|
fault = _string(test_control["fault"], "test_control.fault")
|
faults = {"NONE", "INTERRUPT_AFTER_RECEIPT_ROOT", "INTERRUPT_AFTER_CONTRACT", "INTERRUPT_AFTER_MAP", "INTERRUPT_AFTER_ANCHOR", "INTERRUPT_BEFORE_RENAME", "INTERRUPT_AFTER_RENAME", "INTERRUPT_AFTER_TERMINAL_TEMP"}
|
if environment not in {"PRODUCTION", "ISOLATED_TEST"} or fault not in faults or (environment == "PRODUCTION" and fault != "NONE"):
|
raise ArchiveError("TEST_CONTROL", f"{environment}:{fault}")
|
try:
|
publisher_config_value = json.loads(publisher_data.decode("utf-8"), object_pairs_hook=base._pairs_no_duplicates)
|
except Exception as exc:
|
raise ArchiveError("PUBLISHER_CONFIG_PARSE", str(exc), 30) from exc
|
if any(publisher_config_value["identity"].get(key) != identity[key] for key in ("task_id", "case_id", "batch_id", "run_id", "attempt_id")):
|
raise ArchiveError("PUBLISHER_IDENTITY_BINDING", "task/case/batch/run/attempt")
|
formal_rows = _publisher_formal_rows(publisher_config_value)
|
formal_set_sha256 = _publisher_formal_digest(formal_rows)
|
publisher_lock_path = _inside(
|
root,
|
_safe_relative(publisher_config_value["roots"]["lock_path"], "publisher.roots.lock_path"),
|
)
|
publisher_ledger_path = _inside(
|
root,
|
_safe_relative(publisher_config_value["roots"]["ledger_path"], "publisher.roots.ledger_path"),
|
)
|
publisher_attempt_receipt_root = _inside(
|
root,
|
_safe_relative(
|
publisher_config_value["roots"]["attempt_receipt_root"],
|
"publisher.roots.attempt_receipt_root",
|
),
|
)
|
terminal_path = _inside(
|
root,
|
_safe_relative(
|
publisher_config_value["roots"]["attempt_receipt_root"],
|
"publisher.roots.attempt_receipt_root",
|
),
|
) / "terminal.json"
|
try:
|
raw_terminal = json.loads(_read(terminal_path).decode("utf-8"), object_pairs_hook=base._pairs_no_duplicates)
|
except Exception as exc:
|
raise ArchiveError("PUBLISHER_TERMINAL_PARSE", str(exc), 30) from exc
|
if (
|
not isinstance(raw_terminal, dict)
|
or raw_terminal.get("status") != "COMMITTED"
|
or raw_terminal.get("exit_code") != 0
|
or raw_terminal.get("config_sha256") != _sha(publisher_data)
|
or not SHA_RE.fullmatch(str(raw_terminal.get("attempt_event_chain_sha256", "")))
|
):
|
raise ArchiveError("PUBLISHER_TERMINAL_BINDING", "status/config/event-chain", 30)
|
publisher_terminal = dict(raw_terminal); publisher_terminal["_path"] = str(terminal_path)
|
return Plan(
|
config, config_path, config_data, _sha(config_data), root, identity,
|
publisher_config, publisher_data, source_root, target_root, receipt_root,
|
map_path, rows, source_digest, canonical_digest, publisher_terminal,
|
formal_rows, formal_set_sha256, publisher_lock_path,
|
publisher_ledger_path, publisher_attempt_receipt_root,
|
publisher_config_value, fault,
|
)
|
|
|
def run(config_path: Path) -> dict[str, Any]:
|
plan = _load_plan(config_path)
|
rollback_root = plan.receipt_root.parent / "rollback" / plan.identity["archive_id"]
|
if base._exists(rollback_root):
|
raise ArchiveError("RECOVERY_REQUIRED_ARCHIVE_ROLLED_BACK", str(rollback_root), 30)
|
source_exists = base._exists(plan.source_root)
|
target_exists = base._exists(plan.target_root)
|
if source_exists and target_exists:
|
raise ArchiveError("RECOVERY_REQUIRED_SOURCE_AND_TARGET", "both present", 30)
|
if not source_exists and not target_exists:
|
raise ArchiveError("RECOVERY_REQUIRED_SOURCE_AND_TARGET", "both absent", 30)
|
if source_exists:
|
try:
|
direct_plan = _validate_direct_config(plan.publisher_value, plan.publisher_config, plan.publisher_config_data)
|
replay = _verify_replay(direct_plan)
|
except base.PublisherError as exc:
|
raise ArchiveError("PUBLISHER_NOT_COMMITTED", f"{exc.code}:{exc.detail}", 30) from exc
|
if replay.get("status") != "IDEMPOTENT_COMMITTED" or replay.get("exit_code") != 0:
|
raise ArchiveError("PUBLISHER_NOT_COMMITTED", str(replay.get("status")), 30)
|
_verify_publisher_formal(plan)
|
_verify_canonical(plan)
|
if source_exists:
|
_verify_tree(plan, "legacy")
|
else:
|
_verify_tree(plan, "archive")
|
_prepare_anchor(plan)
|
_verify_anchor(plan, allow_terminal=True)
|
if source_exists:
|
_fault(plan, "INTERRUPT_BEFORE_RENAME")
|
try:
|
os.rename(base._long(plan.source_root), base._long(plan.target_root))
|
except OSError as exc:
|
raise ArchiveError("ARCHIVE_ATOMIC_RENAME", str(exc), 30) from exc
|
_fault(plan, "INTERRUPT_AFTER_RENAME")
|
if base._exists(plan.source_root) or not base._exists(plan.target_root):
|
raise ArchiveError("RECOVERY_REQUIRED_RENAME_TRUTH", "source/target", 30)
|
_verify_tree(plan, "archive")
|
_verify_publisher_formal(plan)
|
_verify_canonical(plan)
|
value = _publish_terminal(plan)
|
_verify_anchor(plan, allow_terminal=True)
|
if set(_walk(plan.receipt_root)) != {"anchor_manifest.json", "archive_contract.json", "legacy_case_path_map.csv", "terminal.json"}:
|
raise ArchiveError("RECOVERY_REQUIRED_FINAL_RECEIPT_SET", "mismatch", 30)
|
if value["status"] == "ARCHIVED" and base._exists(plan.receipt_root / "terminal.json"):
|
value = dict(value)
|
value["status"] = "IDEMPOTENT_ARCHIVED" if not source_exists else "ARCHIVED"
|
return value
|
|
|
def _rollback_fault(receipt_root: Path, fault: str, name: str) -> None:
|
if fault != name:
|
return
|
marker = receipt_root.parent / (receipt_root.name + ".test-fault-" + name)
|
if base._exists(marker):
|
return
|
try:
|
base._create_exclusive(marker, b"consumed\n")
|
except base.PublisherError as exc:
|
raise ArchiveError(exc.code, exc.detail, exc.exit_code) from exc
|
raise InjectedStop("INJECTED_ROLLBACK_PROCESS_RESTART", name, 20)
|
|
|
def _rollback_config(config_path: Path) -> tuple[Plan, dict[str, Any], bytes, Path, str]:
|
data = _read(config_path)
|
try:
|
value = json.loads(data.decode("utf-8"), object_pairs_hook=base._pairs_no_duplicates)
|
except Exception as exc:
|
raise ArchiveError("ROLLBACK_CONFIG_PARSE", str(exc), 30) from exc
|
value = _need_object(value, "rollback")
|
_keys(value, {"schema_version", "identity", "roots", "binding", "test_control"}, "rollback")
|
if value["schema_version"] != ROLLBACK_SCHEMA:
|
raise ArchiveError("ROLLBACK_CONFIG_SCHEMA", str(value["schema_version"]), 30)
|
identity = _need_object(value["identity"], "rollback.identity")
|
identity_keys = {
|
"task_id", "case_id", "batch_id", "run_id", "attempt_id", "archive_id",
|
"rollback_id", "operator", "rollback_authorization_id", "rollback_audit_id",
|
}
|
_keys(identity, identity_keys, "rollback.identity")
|
identity = {key: _string(item, f"rollback.identity.{key}") for key, item in identity.items()}
|
if not identity["rollback_authorization_id"].startswith("AUTH-") or not identity["rollback_audit_id"].startswith("AUDIT-"):
|
raise ArchiveError("ROLLBACK_AUTHORIZATION_BINDING", "AUTH/AUDIT domain", 30)
|
roots = _need_object(value["roots"], "rollback.roots")
|
_keys(roots, {"operation_root", "resolved_root", "archive_config_path"}, "rollback.roots")
|
root = Path(_string(roots["operation_root"], "rollback.roots.operation_root"))
|
if not root.is_absolute() or os.path.normcase(os.path.abspath(root)) != os.path.normcase(_string(roots["resolved_root"], "rollback.roots.resolved_root")):
|
raise ArchiveError("ROLLBACK_ROOT_IDENTITY", str(root), 30)
|
archive_config = _inside(root, _safe_relative(roots["archive_config_path"], "rollback.roots.archive_config_path"))
|
binding = _need_object(value["binding"], "rollback.binding")
|
_keys(binding, {"archive_config_bytes", "archive_config_sha256", "archive_terminal_bytes", "archive_terminal_sha256"}, "rollback.binding")
|
if _identity(archive_config) != (
|
_integer(binding["archive_config_bytes"], "rollback.binding.archive_config_bytes", positive=True),
|
_sha_value(binding["archive_config_sha256"], "rollback.binding.archive_config_sha256"),
|
):
|
raise ArchiveError("ROLLBACK_ARCHIVE_CONFIG_IDENTITY", str(archive_config), 30)
|
plan = _load_plan(archive_config)
|
for key in ("task_id", "case_id", "batch_id", "run_id", "attempt_id", "archive_id"):
|
if identity[key] != plan.identity[key]:
|
raise ArchiveError("ROLLBACK_IDENTITY_BINDING", key, 30)
|
archive_terminal = plan.receipt_root / "terminal.json"
|
if _identity(archive_terminal) != (
|
_integer(binding["archive_terminal_bytes"], "rollback.binding.archive_terminal_bytes", positive=True),
|
_sha_value(binding["archive_terminal_sha256"], "rollback.binding.archive_terminal_sha256"),
|
):
|
raise ArchiveError("ROLLBACK_ARCHIVE_TERMINAL_IDENTITY", str(archive_terminal), 30)
|
control = _need_object(value["test_control"], "rollback.test_control")
|
_keys(control, {"environment", "fault"}, "rollback.test_control")
|
environment = _string(control["environment"], "rollback.test_control.environment")
|
fault = _string(control["fault"], "rollback.test_control.fault")
|
faults = {"NONE", "INTERRUPT_AFTER_ROLLBACK_ANCHOR", "INTERRUPT_BEFORE_ROLLBACK_RENAME", "INTERRUPT_AFTER_ROLLBACK_RENAME", "INTERRUPT_AFTER_ROLLBACK_TERMINAL_TEMP"}
|
if environment not in {"PRODUCTION", "ISOLATED_TEST"} or fault not in faults or (environment == "PRODUCTION" and fault != "NONE"):
|
raise ArchiveError("ROLLBACK_TEST_CONTROL", f"{environment}:{fault}", 30)
|
receipt_root = plan.receipt_root.parent / "rollback" / plan.identity["archive_id"]
|
_ancestor_gate(plan.root, receipt_root, "rollback_receipt_root", include_leaf=base._exists(receipt_root))
|
return plan, identity, data, receipt_root, fault
|
|
|
def _rollback_values(plan: Plan, identity: dict[str, str], data: bytes) -> tuple[bytes, bytes]:
|
archive_terminal = _read(plan.receipt_root / "terminal.json")
|
contract = _canonical({
|
"archive_config_bytes": len(plan.config_data),
|
"archive_config_sha256": plan.config_sha256,
|
"archive_id": identity["archive_id"],
|
"archive_terminal_bytes": len(archive_terminal),
|
"archive_terminal_sha256": _sha(archive_terminal),
|
"authorization_id": identity["rollback_authorization_id"],
|
"publisher_formal_row_count": len(plan.publisher_formal_rows),
|
"publisher_formal_set_sha256": plan.publisher_formal_set_sha256,
|
"rollback_audit_id": identity["rollback_audit_id"],
|
"rollback_config_bytes": len(data),
|
"rollback_config_sha256": _sha(data),
|
"rollback_id": identity["rollback_id"],
|
"schema_version": ROLLBACK_TERMINAL_SCHEMA,
|
"source_root": _relative(plan.source_root, plan.root),
|
"source_set_sha256": plan.source_set_sha256,
|
"target_root": _relative(plan.target_root, plan.root),
|
})
|
manifest = _canonical({
|
"members": [
|
{"bytes": len(contract), "name": "rollback_contract.json", "sha256": _sha(contract)},
|
{"bytes": len(archive_terminal), "name": "archive_terminal.json", "sha256": _sha(archive_terminal)},
|
],
|
"schema_version": ROLLBACK_TERMINAL_SCHEMA,
|
})
|
return contract, manifest
|
|
|
def _verify_rollback_anchor(plan: Plan, identity: dict[str, str], data: bytes, receipt_root: Path, *, allow_terminal: bool) -> None:
|
contract, manifest = _rollback_values(plan, identity, data)
|
expected = {
|
"archive_terminal.json": _read(plan.receipt_root / "terminal.json"),
|
"rollback_contract.json": contract,
|
"rollback_manifest.json": manifest,
|
}
|
allowed = set(expected) | ({"terminal.json", "terminal.tmp.json"} if allow_terminal else set())
|
actual = set(_walk(receipt_root))
|
if not set(expected).issubset(actual) or not actual.issubset(allowed):
|
raise ArchiveError("RECOVERY_REQUIRED_ROLLBACK_RECEIPT_SET", str(sorted(actual)), 30)
|
for name, payload in expected.items():
|
if _read(receipt_root / name) != payload:
|
raise ArchiveError("RECOVERY_REQUIRED_ROLLBACK_RECEIPT_BINDING", name, 30)
|
|
|
def _rollback_terminal_value(plan: Plan, identity: dict[str, str], data: bytes) -> dict[str, Any]:
|
return {
|
"archive_id": identity["archive_id"],
|
"authorization_id": identity["rollback_authorization_id"],
|
"exit_code": 0,
|
"legacy_source_present": True,
|
"publisher_formal_set_sha256": plan.publisher_formal_set_sha256,
|
"rollback_config_bytes": len(data),
|
"rollback_config_sha256": _sha(data),
|
"rollback_id": identity["rollback_id"],
|
"schema_version": ROLLBACK_TERMINAL_SCHEMA,
|
"status": "LEGACY_RESTORED",
|
"target_present": False,
|
}
|
|
|
def rollback(config_path: Path) -> dict[str, Any]:
|
plan, identity, data, receipt_root, fault = _rollback_config(config_path)
|
source_exists = base._exists(plan.source_root)
|
target_exists = base._exists(plan.target_root)
|
if source_exists == target_exists:
|
raise ArchiveError("RECOVERY_REQUIRED_ROLLBACK_SOURCE_AND_TARGET", f"source={source_exists};target={target_exists}", 30)
|
if not base._exists(receipt_root):
|
try:
|
os.mkdir(base._long(receipt_root))
|
except OSError as exc:
|
raise ArchiveError("ROLLBACK_RECEIPT_ROOT_CREATE", str(exc), 30) from exc
|
contract, manifest = _rollback_values(plan, identity, data)
|
_create_exact(receipt_root / "rollback_contract.json", contract)
|
_create_exact(receipt_root / "archive_terminal.json", _read(plan.receipt_root / "terminal.json"))
|
_create_exact(receipt_root / "rollback_manifest.json", manifest)
|
_verify_rollback_anchor(plan, identity, data, receipt_root, allow_terminal=True)
|
_rollback_fault(receipt_root, fault, "INTERRUPT_AFTER_ROLLBACK_ANCHOR")
|
if target_exists:
|
_verify_anchor(plan, allow_terminal=True)
|
_verify_tree(plan, "archive")
|
_verify_publisher_formal(plan)
|
_rollback_fault(receipt_root, fault, "INTERRUPT_BEFORE_ROLLBACK_RENAME")
|
try:
|
os.rename(base._long(plan.target_root), base._long(plan.source_root))
|
except OSError as exc:
|
raise ArchiveError("ROLLBACK_ATOMIC_RENAME", str(exc), 30) from exc
|
_rollback_fault(receipt_root, fault, "INTERRUPT_AFTER_ROLLBACK_RENAME")
|
if not base._exists(plan.source_root) or base._exists(plan.target_root):
|
raise ArchiveError("RECOVERY_REQUIRED_ROLLBACK_RENAME_TRUTH", "source/target", 30)
|
# A restart after the atomic rollback rename must prove the same complete
|
# publisher/archive preimage as the first rollback attempt before it may
|
# create or replay a PASS-like terminal. The archive directory is gone at
|
# this point, so the durable archive anchor, the 648 formal publisher rows,
|
# the 17-column event chain and the canonical 645-row mapping are the
|
# authoritative bridge between the archived and restored tree states.
|
_verify_anchor(plan, allow_terminal=True)
|
_verify_publisher_formal(plan)
|
_verify_canonical(plan)
|
_verify_tree(plan, "legacy")
|
if base._exists(plan.publisher_lock_path):
|
raise ArchiveError("PUBLISHER_LOCK_PRESENT", str(plan.publisher_lock_path), 30)
|
terminal_value = _rollback_terminal_value(plan, identity, data)
|
terminal_data = _canonical(terminal_value)
|
temporary = receipt_root / "terminal.tmp.json"
|
terminal = receipt_root / "terminal.json"
|
if base._exists(terminal):
|
if _read(terminal) != terminal_data or base._exists(temporary):
|
raise ArchiveError("RECOVERY_REQUIRED_ROLLBACK_TERMINAL", "binding", 30)
|
else:
|
_create_exact(temporary, terminal_data)
|
_rollback_fault(receipt_root, fault, "INTERRUPT_AFTER_ROLLBACK_TERMINAL_TEMP")
|
try:
|
os.rename(base._long(temporary), base._long(terminal))
|
except OSError as exc:
|
raise ArchiveError("ROLLBACK_TERMINAL_RENAME", str(exc), 30) from exc
|
_verify_rollback_anchor(plan, identity, data, receipt_root, allow_terminal=True)
|
if set(_walk(receipt_root)) != {"archive_terminal.json", "rollback_contract.json", "rollback_manifest.json", "terminal.json"}:
|
raise ArchiveError("RECOVERY_REQUIRED_ROLLBACK_FINAL_SET", "mismatch", 30)
|
result = dict(terminal_value)
|
result["status"] = "IDEMPOTENT_LEGACY_RESTORED" if source_exists else "LEGACY_RESTORED"
|
return result
|
|
|
def _failure(exc: ArchiveError, config_path: Path) -> dict[str, Any]:
|
return {
|
"config_path": str(config_path),
|
"detail": exc.detail,
|
"error_code": exc.code,
|
"exit_code": exc.exit_code,
|
"schema_version": TERMINAL_SCHEMA,
|
"status": "RECOVERY_REQUIRED" if exc.exit_code == 30 else "FAIL_CLOSED",
|
}
|
|
|
def main(argv: list[str] | None = None) -> int:
|
parser = argparse.ArgumentParser(description="Recoverable legacy postcommit archive coordinator")
|
group = parser.add_mutually_exclusive_group(required=True)
|
group.add_argument("--config")
|
group.add_argument("--rollback-config")
|
args = parser.parse_args(argv)
|
config_path = Path(args.config or args.rollback_config)
|
try:
|
value = run(config_path) if args.config else rollback(config_path)
|
except ArchiveError as exc:
|
value = _failure(exc, config_path)
|
except Exception as exc:
|
value = _failure(ArchiveError("UNEXPECTED", f"{type(exc).__name__}:{exc}", 30), config_path)
|
sys.stdout.write(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n")
|
return int(value["exit_code"])
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
