819 lines
29 KiB
Python
Executable file
819 lines
29 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Restore and verify a canonical PostgreSQL dump in an isolated Docker canary."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import signal
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
try:
|
|
from .verify_postgres_parity_manifest import compare_manifests, load_manifest
|
|
except ImportError: # pragma: no cover - direct script execution
|
|
from verify_postgres_parity_manifest import compare_manifests, load_manifest
|
|
|
|
|
|
DEFAULT_IMAGE = "postgres:16-alpine"
|
|
DEFAULT_MANIFEST_SQL = Path(__file__).with_name("postgres_parity_manifest.sql")
|
|
DATA_DIR = "/var/lib/postgresql/data"
|
|
DUMP_DESTINATION = "/tmp/teleo-canonical.dump"
|
|
MANIFEST_DESTINATION = "/tmp/postgres-parity-manifest.sql"
|
|
CANARY_LABEL = "com.livingip.teleo.canonical-rebuild-canary"
|
|
DATABASE_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]{0,62}\Z")
|
|
CONTAINER_PREFIX_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,39}\Z")
|
|
ALLOWED_APPLICATION_ROLES = frozenset({"kb_apply", "kb_review"})
|
|
ROLE_BOOLEAN_FIELDS = (
|
|
"can_login",
|
|
"superuser",
|
|
"create_db",
|
|
"create_role",
|
|
"inherit",
|
|
"replication",
|
|
"bypass_rls",
|
|
)
|
|
KEY_TABLES = (
|
|
"public.claims",
|
|
"public.sources",
|
|
"public.claim_evidence",
|
|
"public.claim_edges",
|
|
"public.reasoning_tools",
|
|
"kb_stage.kb_proposals",
|
|
)
|
|
|
|
|
|
class TerminationRequested(RuntimeError):
|
|
"""Raised after SIGINT/SIGTERM so the normal cleanup path still runs."""
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def validate_custom_dump(path: Path) -> None:
|
|
if not path.is_file():
|
|
raise ValueError(f"custom dump does not exist or is not a regular file: {path}")
|
|
if path.stat().st_size <= 5:
|
|
raise ValueError("custom dump is empty or truncated")
|
|
with path.open("rb") as handle:
|
|
signature = handle.read(5)
|
|
if signature != b"PGDMP":
|
|
raise ValueError("dump is not PostgreSQL custom format (missing PGDMP signature)")
|
|
|
|
|
|
def new_container_name(prefix: str) -> str:
|
|
timestamp = datetime.now(UTC).strftime("%Y%m%d%H%M%S")
|
|
return f"{prefix}-{timestamp}-{uuid.uuid4().hex[:12]}"
|
|
|
|
|
|
def _run(
|
|
command: list[str],
|
|
*,
|
|
timeout: float,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
command,
|
|
capture_output=True,
|
|
check=False,
|
|
text=True,
|
|
timeout=timeout,
|
|
)
|
|
|
|
|
|
def _bounded_error(completed: subprocess.CompletedProcess[str]) -> str:
|
|
message = (completed.stderr or completed.stdout or "no command output").strip()
|
|
return message[-2000:]
|
|
|
|
|
|
def _require_success(completed: subprocess.CompletedProcess[str], action: str) -> str:
|
|
if completed.returncode != 0:
|
|
raise RuntimeError(f"{action} failed (exit {completed.returncode}): {_bounded_error(completed)}")
|
|
return completed.stdout
|
|
|
|
|
|
def _absence_result(completed: subprocess.CompletedProcess[str]) -> tuple[bool | None, str | None]:
|
|
if completed.returncode == 0:
|
|
return False, None
|
|
message = (completed.stderr or completed.stdout or "").strip()
|
|
lowered = message.lower()
|
|
if "no such object" in lowered or "no such container" in lowered:
|
|
return True, None
|
|
return None, message[-2000:] or f"docker inspect exited {completed.returncode}"
|
|
|
|
|
|
def inspect_container_absence(
|
|
docker_bin: str,
|
|
container: str,
|
|
*,
|
|
timeout: float,
|
|
) -> tuple[bool | None, str | None]:
|
|
completed = _run([docker_bin, "container", "inspect", container], timeout=timeout)
|
|
return _absence_result(completed)
|
|
|
|
|
|
def inspect_container_isolation(
|
|
docker_bin: str,
|
|
container: str,
|
|
*,
|
|
expected_image: str,
|
|
expected_label: str,
|
|
timeout: float,
|
|
) -> dict[str, Any]:
|
|
completed = _run([docker_bin, "container", "inspect", container], timeout=timeout)
|
|
raw = _require_success(completed, "container isolation inspection")
|
|
try:
|
|
rows = json.loads(raw)
|
|
observed = rows[0]
|
|
host_config = observed["HostConfig"]
|
|
config = observed["Config"]
|
|
except (IndexError, KeyError, TypeError, json.JSONDecodeError) as exc:
|
|
raise RuntimeError("container isolation inspection returned invalid JSON") from exc
|
|
|
|
network_mode = host_config.get("NetworkMode")
|
|
tmpfs = host_config.get("Tmpfs") or {}
|
|
labels = config.get("Labels") or {}
|
|
image = config.get("Image")
|
|
running = bool((observed.get("State") or {}).get("Running"))
|
|
problems = []
|
|
if network_mode != "none":
|
|
problems.append(f"network_mode={network_mode!r}")
|
|
if DATA_DIR not in tmpfs:
|
|
problems.append("postgres_data_tmpfs_missing")
|
|
if labels.get(CANARY_LABEL) != expected_label:
|
|
problems.append("canary_label_mismatch")
|
|
if image != expected_image:
|
|
problems.append(f"image={image!r}")
|
|
if not running:
|
|
problems.append("container_not_running")
|
|
if problems:
|
|
raise RuntimeError("container isolation check failed: " + ", ".join(problems))
|
|
return {
|
|
"network_mode": network_mode,
|
|
"postgres_data_tmpfs": tmpfs[DATA_DIR],
|
|
"image": image,
|
|
"label": labels[CANARY_LABEL],
|
|
"running": running,
|
|
}
|
|
|
|
|
|
def _psql_command(container: str, database: str, *arguments: str, docker_bin: str) -> list[str]:
|
|
return [
|
|
docker_bin,
|
|
"exec",
|
|
container,
|
|
"psql",
|
|
"-X",
|
|
"-U",
|
|
"postgres",
|
|
"-d",
|
|
database,
|
|
"-Atq",
|
|
"-v",
|
|
"ON_ERROR_STOP=1",
|
|
*arguments,
|
|
]
|
|
|
|
|
|
def wait_for_named_database(
|
|
docker_bin: str,
|
|
container: str,
|
|
database: str,
|
|
*,
|
|
startup_timeout: float,
|
|
command_timeout: float,
|
|
poll_interval: float,
|
|
) -> int:
|
|
deadline = time.monotonic() + startup_timeout
|
|
attempts = 0
|
|
last_error = "no psql attempt completed"
|
|
while True:
|
|
attempts += 1
|
|
remaining = max(0.1, deadline - time.monotonic())
|
|
completed = _run(
|
|
_psql_command(
|
|
container,
|
|
database,
|
|
"-c",
|
|
"select current_database();",
|
|
docker_bin=docker_bin,
|
|
),
|
|
timeout=min(command_timeout, remaining),
|
|
)
|
|
if completed.returncode == 0 and completed.stdout.strip() == database:
|
|
return attempts
|
|
last_error = _bounded_error(completed)
|
|
if time.monotonic() >= deadline:
|
|
raise RuntimeError(
|
|
f"named database did not accept a psql connection within {startup_timeout:g}s: {last_error}"
|
|
)
|
|
time.sleep(min(poll_interval, max(0.0, deadline - time.monotonic())))
|
|
|
|
|
|
def _role_keyword(value: bool, enabled: str, disabled: str) -> str:
|
|
return enabled if value else disabled
|
|
|
|
|
|
def application_role_sql(source_manifest: dict[str, Any]) -> str:
|
|
items = source_manifest["singleton"]["application_roles"].get("items", [])
|
|
if not isinstance(items, list):
|
|
raise ValueError("source manifest application_roles.items must be a list")
|
|
statements = []
|
|
seen: set[str] = set()
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
raise ValueError("source manifest application role must be an object")
|
|
name = item.get("name")
|
|
if name not in ALLOWED_APPLICATION_ROLES:
|
|
raise ValueError(f"source manifest contains non-allowlisted application role: {name!r}")
|
|
if name in seen:
|
|
raise ValueError(f"source manifest contains duplicate application role: {name}")
|
|
seen.add(name)
|
|
for field in ROLE_BOOLEAN_FIELDS:
|
|
if type(item.get(field)) is not bool:
|
|
raise ValueError(f"source manifest application role {name} has invalid {field}")
|
|
attributes = [
|
|
_role_keyword(item["can_login"], "LOGIN", "NOLOGIN"),
|
|
_role_keyword(item["superuser"], "SUPERUSER", "NOSUPERUSER"),
|
|
_role_keyword(item["create_db"], "CREATEDB", "NOCREATEDB"),
|
|
_role_keyword(item["create_role"], "CREATEROLE", "NOCREATEROLE"),
|
|
_role_keyword(item["inherit"], "INHERIT", "NOINHERIT"),
|
|
_role_keyword(item["replication"], "REPLICATION", "NOREPLICATION"),
|
|
_role_keyword(item["bypass_rls"], "BYPASSRLS", "NOBYPASSRLS"),
|
|
]
|
|
statements.append(f'CREATE ROLE "{name}" {" ".join(attributes)};')
|
|
return "\n".join(statements)
|
|
|
|
|
|
def execute_application_role_bootstrap(
|
|
docker_bin: str,
|
|
container: str,
|
|
database: str,
|
|
source_manifest: dict[str, Any],
|
|
*,
|
|
timeout: float,
|
|
) -> list[str]:
|
|
sql = application_role_sql(source_manifest)
|
|
if not sql:
|
|
return []
|
|
completed = _run(
|
|
_psql_command(container, database, "-c", sql, docker_bin=docker_bin),
|
|
timeout=timeout,
|
|
)
|
|
_require_success(completed, "application role bootstrap")
|
|
return sorted(item["name"] for item in source_manifest["singleton"]["application_roles"].get("items", []))
|
|
|
|
|
|
def query_scalar(
|
|
docker_bin: str,
|
|
container: str,
|
|
database: str,
|
|
sql: str,
|
|
*,
|
|
timeout: float,
|
|
) -> str:
|
|
completed = _run(
|
|
_psql_command(container, database, "-c", sql, docker_bin=docker_bin),
|
|
timeout=timeout,
|
|
)
|
|
return _require_success(completed, "PostgreSQL readback").strip()
|
|
|
|
|
|
def collect_key_counts(
|
|
docker_bin: str,
|
|
container: str,
|
|
database: str,
|
|
*,
|
|
timeout: float,
|
|
) -> tuple[dict[str, int], int]:
|
|
counts: dict[str, int] = {}
|
|
for table in KEY_TABLES:
|
|
exists = query_scalar(
|
|
docker_bin,
|
|
container,
|
|
database,
|
|
f"select to_regclass('{table}') is not null;",
|
|
timeout=timeout,
|
|
)
|
|
if exists != "t":
|
|
raise RuntimeError(f"required canonical table is missing after restore: {table}")
|
|
schema, name = table.split(".", 1)
|
|
raw_count = query_scalar(
|
|
docker_bin,
|
|
container,
|
|
database,
|
|
f'select count(*) from "{schema}"."{name}";',
|
|
timeout=timeout,
|
|
)
|
|
try:
|
|
counts[table] = int(raw_count)
|
|
except ValueError as exc:
|
|
raise RuntimeError(f"invalid row count for {table}: {raw_count!r}") from exc
|
|
raw_table_count = query_scalar(
|
|
docker_bin,
|
|
container,
|
|
database,
|
|
"select count(*) from pg_tables where schemaname in ('public', 'kb_stage');",
|
|
timeout=timeout,
|
|
)
|
|
try:
|
|
table_count = int(raw_table_count)
|
|
except ValueError as exc:
|
|
raise RuntimeError(f"invalid canonical schema table count: {raw_table_count!r}") from exc
|
|
return counts, table_count
|
|
|
|
|
|
def run_parity_check(
|
|
docker_bin: str,
|
|
container: str,
|
|
database: str,
|
|
*,
|
|
source_manifest_path: Path,
|
|
source_manifest: dict[str, Any],
|
|
manifest_sql: Path,
|
|
command_timeout: float,
|
|
max_target_query_ms: float,
|
|
max_target_source_ratio: float,
|
|
) -> dict[str, Any]:
|
|
copy_result = _run(
|
|
[docker_bin, "cp", str(manifest_sql), f"{container}:{MANIFEST_DESTINATION}"],
|
|
timeout=command_timeout,
|
|
)
|
|
_require_success(copy_result, "parity manifest copy")
|
|
completed = _run(
|
|
_psql_command(
|
|
container,
|
|
database,
|
|
"-f",
|
|
MANIFEST_DESTINATION,
|
|
docker_bin=docker_bin,
|
|
),
|
|
timeout=command_timeout,
|
|
)
|
|
target_payload = _require_success(completed, "target parity manifest capture")
|
|
with tempfile.TemporaryDirectory(prefix="teleo-canonical-parity-") as temp_dir:
|
|
target_path = Path(temp_dir) / "target-manifest.jsonl"
|
|
target_path.write_text(target_payload, encoding="utf-8")
|
|
target_manifest = load_manifest(target_path)
|
|
problems, details = compare_manifests(
|
|
source_manifest,
|
|
target_manifest,
|
|
max_target_query_ms=max_target_query_ms,
|
|
max_target_source_ratio=max_target_source_ratio,
|
|
)
|
|
target_receipt = {
|
|
"bytes": target_path.stat().st_size,
|
|
"sha256": sha256_file(target_path),
|
|
"line_count": sum(1 for line in target_payload.splitlines() if line.strip()),
|
|
}
|
|
return {
|
|
"requested": True,
|
|
"status": "pass" if not problems else "fail",
|
|
"verifier": "ops.verify_postgres_parity_manifest.compare_manifests",
|
|
"source_manifest": {
|
|
"path": str(source_manifest_path),
|
|
"bytes": source_manifest_path.stat().st_size,
|
|
"sha256": sha256_file(source_manifest_path),
|
|
},
|
|
"target_manifest": target_receipt,
|
|
"problems": problems,
|
|
"details": details,
|
|
}
|
|
|
|
|
|
def cleanup_container(
|
|
docker_bin: str,
|
|
container: str,
|
|
*,
|
|
command_timeout: float,
|
|
poll_interval: float,
|
|
) -> dict[str, Any]:
|
|
cleanup: dict[str, Any] = {
|
|
"attempted": True,
|
|
"container": container,
|
|
"force_remove_attempted": True,
|
|
"container_absent": None,
|
|
"status": "unproven",
|
|
}
|
|
try:
|
|
removed = _run([docker_bin, "rm", "--force", container], timeout=command_timeout)
|
|
cleanup["force_remove_exit_code"] = removed.returncode
|
|
if removed.returncode != 0:
|
|
cleanup["force_remove_message"] = _bounded_error(removed)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
cleanup["force_remove_error"] = str(exc)
|
|
except (KeyboardInterrupt, TerminationRequested) as exc:
|
|
cleanup["interrupted"] = str(exc)
|
|
cleanup["force_remove_error"] = str(exc)
|
|
|
|
last_error = None
|
|
for attempt in range(1, 6):
|
|
cleanup["absence_inspection_attempts"] = attempt
|
|
try:
|
|
absent, error = inspect_container_absence(
|
|
docker_bin,
|
|
container,
|
|
timeout=command_timeout,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
absent, error = None, str(exc)
|
|
except (KeyboardInterrupt, TerminationRequested) as exc:
|
|
cleanup["interrupted"] = str(exc)
|
|
absent, error = None, str(exc)
|
|
if absent is True:
|
|
cleanup["container_absent"] = True
|
|
cleanup["status"] = "pass"
|
|
cleanup.pop("absence_inspection_error", None)
|
|
return cleanup
|
|
cleanup["container_absent"] = absent
|
|
last_error = error
|
|
if error and "interrupted" not in cleanup:
|
|
cleanup["absence_inspection_error"] = error
|
|
break
|
|
if attempt < 5:
|
|
try:
|
|
time.sleep(poll_interval)
|
|
except (KeyboardInterrupt, TerminationRequested) as exc:
|
|
cleanup["interrupted"] = str(exc)
|
|
if last_error:
|
|
cleanup["absence_inspection_error"] = last_error
|
|
cleanup["status"] = "fail" if cleanup["container_absent"] is False else "unproven"
|
|
return cleanup
|
|
|
|
|
|
def _initial_receipt(args: argparse.Namespace, container: str) -> dict[str, Any]:
|
|
return {
|
|
"artifact": "local_canonical_postgres_rebuild_canary",
|
|
"generated_at_utc": datetime.now(UTC).isoformat(),
|
|
"status": "running",
|
|
"database": args.database,
|
|
"dump": {
|
|
"path": str(args.dump.resolve()),
|
|
"bytes": None,
|
|
"sha256": None,
|
|
"custom_format_signature_valid": False,
|
|
},
|
|
"container": {
|
|
"name": container,
|
|
"image": args.image,
|
|
"network_mode_required": "none",
|
|
"postgres_data_tmpfs_required": f"rw,nosuid,nodev,noexec,size={args.tmpfs_mb}m",
|
|
"preexisting_absence_proven": False,
|
|
"started": False,
|
|
"isolation": None,
|
|
},
|
|
"application_roles_bootstrapped": [],
|
|
"key_counts": {},
|
|
"canonical_schema_table_count": None,
|
|
"parity": {
|
|
"requested": args.source_manifest is not None,
|
|
"status": "pending" if args.source_manifest is not None else "not_requested",
|
|
"reason": None if args.source_manifest is not None else "--source-manifest was not supplied",
|
|
},
|
|
"timings_seconds": {},
|
|
"cleanup": {
|
|
"attempted": False,
|
|
"container": container,
|
|
"container_absent": None,
|
|
"status": "pending",
|
|
},
|
|
"safety": {
|
|
"production_database_touched": False,
|
|
"network_access_available_to_container": False,
|
|
"persistent_postgres_storage_used": False,
|
|
},
|
|
"error": None,
|
|
}
|
|
|
|
|
|
def run_canary(args: argparse.Namespace) -> dict[str, Any]:
|
|
started_at = time.monotonic()
|
|
container = new_container_name(args.container_prefix)
|
|
receipt = _initial_receipt(args, container)
|
|
phase = "validation"
|
|
source_manifest: dict[str, Any] | None = None
|
|
primary_error: dict[str, str] | None = None
|
|
|
|
try:
|
|
hash_started = time.monotonic()
|
|
validate_custom_dump(args.dump)
|
|
receipt["dump"].update(
|
|
{
|
|
"bytes": args.dump.stat().st_size,
|
|
"sha256": sha256_file(args.dump),
|
|
"custom_format_signature_valid": True,
|
|
}
|
|
)
|
|
receipt["timings_seconds"]["dump_validation_and_hash"] = round(time.monotonic() - hash_started, 6)
|
|
|
|
if args.source_manifest is not None:
|
|
phase = "source_manifest_validation"
|
|
source_manifest = load_manifest(args.source_manifest)
|
|
|
|
phase = "container_name_preflight"
|
|
absent, absence_error = inspect_container_absence(
|
|
args.docker_bin,
|
|
container,
|
|
timeout=args.command_timeout,
|
|
)
|
|
if absent is not True:
|
|
detail = absence_error or "generated container name unexpectedly already exists"
|
|
raise RuntimeError(f"could not prove generated container name is unused: {detail}")
|
|
receipt["container"]["preexisting_absence_proven"] = True
|
|
|
|
phase = "container_start"
|
|
startup_started = time.monotonic()
|
|
tmpfs_options = f"{DATA_DIR}:rw,nosuid,nodev,noexec,size={args.tmpfs_mb}m"
|
|
run_result = _run(
|
|
[
|
|
args.docker_bin,
|
|
"run",
|
|
"--detach",
|
|
"--rm",
|
|
"--network",
|
|
"none",
|
|
"--name",
|
|
container,
|
|
"--label",
|
|
f"{CANARY_LABEL}={container}",
|
|
"--tmpfs",
|
|
tmpfs_options,
|
|
"--env",
|
|
f"POSTGRES_DB={args.database}",
|
|
"--env",
|
|
"POSTGRES_HOST_AUTH_METHOD=trust",
|
|
args.image,
|
|
],
|
|
timeout=args.command_timeout,
|
|
)
|
|
container_id = _require_success(run_result, "networkless PostgreSQL container start").strip()
|
|
if not container_id:
|
|
raise RuntimeError("networkless PostgreSQL container start returned no container id")
|
|
receipt["container"]["started"] = True
|
|
receipt["container"]["id_prefix"] = container_id[:12]
|
|
receipt["container"]["isolation"] = inspect_container_isolation(
|
|
args.docker_bin,
|
|
container,
|
|
expected_image=args.image,
|
|
expected_label=container,
|
|
timeout=args.command_timeout,
|
|
)
|
|
|
|
phase = "named_database_readiness"
|
|
readiness_attempts = wait_for_named_database(
|
|
args.docker_bin,
|
|
container,
|
|
args.database,
|
|
startup_timeout=args.startup_timeout,
|
|
command_timeout=args.command_timeout,
|
|
poll_interval=args.poll_interval,
|
|
)
|
|
receipt["container"]["named_database_psql_ready"] = True
|
|
receipt["container"]["named_database_psql_attempts"] = readiness_attempts
|
|
receipt["timings_seconds"]["container_start_and_named_database_ready"] = round(
|
|
time.monotonic() - startup_started,
|
|
6,
|
|
)
|
|
|
|
if source_manifest is not None:
|
|
phase = "application_role_bootstrap"
|
|
receipt["application_roles_bootstrapped"] = execute_application_role_bootstrap(
|
|
args.docker_bin,
|
|
container,
|
|
args.database,
|
|
source_manifest,
|
|
timeout=args.command_timeout,
|
|
)
|
|
|
|
phase = "dump_copy"
|
|
copy_result = _run(
|
|
[args.docker_bin, "cp", str(args.dump.resolve()), f"{container}:{DUMP_DESTINATION}"],
|
|
timeout=args.command_timeout,
|
|
)
|
|
_require_success(copy_result, "custom dump copy")
|
|
|
|
phase = "restore"
|
|
restore_started = time.monotonic()
|
|
restore_result = _run(
|
|
[
|
|
args.docker_bin,
|
|
"exec",
|
|
container,
|
|
"pg_restore",
|
|
"-U",
|
|
"postgres",
|
|
"-d",
|
|
args.database,
|
|
"--no-owner",
|
|
"--no-privileges",
|
|
"--exit-on-error",
|
|
DUMP_DESTINATION,
|
|
],
|
|
timeout=args.restore_timeout,
|
|
)
|
|
_require_success(restore_result, "canonical PostgreSQL restore")
|
|
receipt["timings_seconds"]["pg_restore"] = round(time.monotonic() - restore_started, 6)
|
|
|
|
phase = "key_count_readback"
|
|
count_started = time.monotonic()
|
|
key_counts, table_count = collect_key_counts(
|
|
args.docker_bin,
|
|
container,
|
|
args.database,
|
|
timeout=args.command_timeout,
|
|
)
|
|
receipt["key_counts"] = key_counts
|
|
receipt["canonical_schema_table_count"] = table_count
|
|
receipt["timings_seconds"]["key_count_readback"] = round(time.monotonic() - count_started, 6)
|
|
|
|
if source_manifest is not None:
|
|
phase = "parity_manifest_and_compare"
|
|
parity_started = time.monotonic()
|
|
receipt["parity"] = run_parity_check(
|
|
args.docker_bin,
|
|
container,
|
|
args.database,
|
|
source_manifest_path=args.source_manifest,
|
|
source_manifest=source_manifest,
|
|
manifest_sql=args.manifest_sql,
|
|
command_timeout=args.manifest_timeout,
|
|
max_target_query_ms=args.max_target_query_ms,
|
|
max_target_source_ratio=args.max_target_source_ratio,
|
|
)
|
|
receipt["timings_seconds"]["manifest_capture_and_parity_compare"] = round(
|
|
time.monotonic() - parity_started,
|
|
6,
|
|
)
|
|
if receipt["parity"]["status"] != "pass":
|
|
raise RuntimeError("canonical parity verification failed")
|
|
receipt["status"] = "pass"
|
|
except (
|
|
OSError,
|
|
ValueError,
|
|
RuntimeError,
|
|
subprocess.TimeoutExpired,
|
|
json.JSONDecodeError,
|
|
KeyboardInterrupt,
|
|
TerminationRequested,
|
|
) as exc:
|
|
primary_error = {
|
|
"phase": phase,
|
|
"type": type(exc).__name__,
|
|
"message": str(exc)[-2000:],
|
|
}
|
|
receipt["status"] = "fail"
|
|
receipt["error"] = primary_error
|
|
finally:
|
|
cleanup_started = time.monotonic()
|
|
cleanup = cleanup_container(
|
|
args.docker_bin,
|
|
container,
|
|
command_timeout=args.command_timeout,
|
|
poll_interval=args.poll_interval,
|
|
)
|
|
receipt["cleanup"] = cleanup
|
|
receipt["timings_seconds"]["cleanup"] = round(time.monotonic() - cleanup_started, 6)
|
|
receipt["timings_seconds"]["total"] = round(time.monotonic() - started_at, 6)
|
|
if cleanup.get("container_absent") is not True:
|
|
receipt["status"] = "fail"
|
|
cleanup_error = {
|
|
"phase": "cleanup",
|
|
"type": "CleanupProofError",
|
|
"message": "container absence was not independently proven",
|
|
}
|
|
if primary_error is None:
|
|
receipt["error"] = cleanup_error
|
|
else:
|
|
receipt["cleanup_error"] = cleanup_error
|
|
if cleanup.get("interrupted"):
|
|
receipt["status"] = "fail"
|
|
interruption_error = {
|
|
"phase": "cleanup",
|
|
"type": "TerminationRequested",
|
|
"message": cleanup["interrupted"],
|
|
}
|
|
if primary_error is None:
|
|
receipt["error"] = interruption_error
|
|
else:
|
|
receipt["cleanup_interruption"] = interruption_error
|
|
return receipt
|
|
|
|
|
|
def write_receipt(path: Path, payload: dict[str, Any]) -> None:
|
|
path = path.expanduser()
|
|
if not path.is_absolute():
|
|
path = Path.cwd() / path
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path = path.parent.resolve() / path.name
|
|
encoded = (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode()
|
|
temporary = path.parent / f".{path.name}.{uuid.uuid4().hex}.tmp"
|
|
descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
try:
|
|
with os.fdopen(descriptor, "wb") as handle:
|
|
handle.write(encoded)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
os.chmod(path, 0o600)
|
|
finally:
|
|
try:
|
|
temporary.unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--dump", required=True, type=Path, help="PostgreSQL custom-format pg_dump")
|
|
parser.add_argument("--database", default="teleo")
|
|
parser.add_argument("--image", default=DEFAULT_IMAGE)
|
|
parser.add_argument("--container-prefix", default="teleo-canonical-rebuild")
|
|
parser.add_argument("--tmpfs-mb", default=1024, type=int)
|
|
parser.add_argument("--source-manifest", type=Path)
|
|
parser.add_argument("--manifest-sql", default=DEFAULT_MANIFEST_SQL, type=Path)
|
|
parser.add_argument("--max-target-query-ms", default=2000.0, type=float)
|
|
parser.add_argument("--max-target-source-ratio", default=20.0, type=float)
|
|
parser.add_argument("--startup-timeout", default=60.0, type=float)
|
|
parser.add_argument("--restore-timeout", default=300.0, type=float)
|
|
parser.add_argument("--manifest-timeout", default=180.0, type=float)
|
|
parser.add_argument("--command-timeout", default=30.0, type=float)
|
|
parser.add_argument("--poll-interval", default=0.25, type=float)
|
|
parser.add_argument("--docker-bin", default="docker")
|
|
parser.add_argument("--output", required=True, type=Path)
|
|
args = parser.parse_args(argv)
|
|
|
|
if not DATABASE_NAME_RE.fullmatch(args.database):
|
|
parser.error("--database must be a safe PostgreSQL identifier up to 63 characters")
|
|
if not CONTAINER_PREFIX_RE.fullmatch(args.container_prefix):
|
|
parser.error("--container-prefix must be 1-40 Docker-safe characters")
|
|
if not args.image or any(character.isspace() for character in args.image):
|
|
parser.error("--image must be a non-empty Docker image reference without whitespace")
|
|
if not args.docker_bin or any(character.isspace() for character in args.docker_bin):
|
|
parser.error("--docker-bin must be one executable name or path without whitespace")
|
|
if not 64 <= args.tmpfs_mb <= 65536:
|
|
parser.error("--tmpfs-mb must be between 64 and 65536")
|
|
for value, flag in (
|
|
(args.startup_timeout, "--startup-timeout"),
|
|
(args.restore_timeout, "--restore-timeout"),
|
|
(args.manifest_timeout, "--manifest-timeout"),
|
|
(args.command_timeout, "--command-timeout"),
|
|
(args.poll_interval, "--poll-interval"),
|
|
(args.max_target_query_ms, "--max-target-query-ms"),
|
|
(args.max_target_source_ratio, "--max-target-source-ratio"),
|
|
):
|
|
if value <= 0:
|
|
parser.error(f"{flag} must be positive")
|
|
if not args.dump.is_file():
|
|
parser.error("--dump must be an existing regular file")
|
|
if args.source_manifest is not None:
|
|
if not args.source_manifest.is_file():
|
|
parser.error("--source-manifest must be an existing regular file")
|
|
if not args.manifest_sql.is_file():
|
|
parser.error("--manifest-sql must be an existing regular file when parity is requested")
|
|
protected_inputs = {args.dump.resolve(), args.manifest_sql.resolve()}
|
|
if args.source_manifest is not None:
|
|
protected_inputs.add(args.source_manifest.resolve())
|
|
if args.output.resolve() in protected_inputs:
|
|
parser.error("--output must not overwrite a dump or manifest input")
|
|
return args
|
|
|
|
|
|
def _termination_handler(signum: int, _frame: Any) -> None:
|
|
raise TerminationRequested(f"received {signal.Signals(signum).name}")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
previous_handlers = {}
|
|
for watched_signal in (signal.SIGINT, signal.SIGTERM):
|
|
previous_handlers[watched_signal] = signal.getsignal(watched_signal)
|
|
signal.signal(watched_signal, _termination_handler)
|
|
try:
|
|
payload = run_canary(args)
|
|
finally:
|
|
for watched_signal, previous in previous_handlers.items():
|
|
signal.signal(watched_signal, previous)
|
|
|
|
try:
|
|
write_receipt(args.output, payload)
|
|
except OSError as exc:
|
|
payload["status"] = "fail"
|
|
payload["receipt_write_error"] = str(exc)
|
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
|
return 0 if payload["status"] == "pass" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|