179 lines
9.2 KiB
Python
179 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Durable full-generation synchronization run coordinator.
|
|
|
|
The HTTP adapter starts this executable with fixed arguments. Peer transport is
|
|
selected only by a trusted local configuration file; request JSON can never
|
|
choose commands, paths, databases, capabilities, or causal policy.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import fcntl
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import signal
|
|
import sys
|
|
import tempfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
MAX_CONFIG = 16 * 1024
|
|
MAX_REPORT = 64 * 1024
|
|
MAX_DIAGNOSTIC = 1024
|
|
PHASES = {"requested", "waiting_android_publication", "running",
|
|
"local_import_committed", "published", "waiting_acknowledgement",
|
|
"peer_consumed", "completed", "failed", "interrupted", "explicitly_degraded"}
|
|
|
|
|
|
def canonical(value: object) -> bytes:
|
|
return (json.dumps(value, ensure_ascii=False, sort_keys=True,
|
|
separators=(",", ":")) + "\n").encode()
|
|
|
|
|
|
def atomic_write(path: Path, value: dict) -> None:
|
|
raw = canonical(value)
|
|
if len(raw) > MAX_REPORT:
|
|
raise RuntimeError("run report exceeds 64 KiB")
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
fd, temporary = tempfile.mkstemp(prefix=".sync-run-", dir=path.parent)
|
|
try:
|
|
os.fchmod(fd, 0o600)
|
|
with os.fdopen(fd, "wb") as stream:
|
|
stream.write(raw); stream.flush(); os.fsync(stream.fileno())
|
|
os.replace(temporary, path)
|
|
directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
|
|
try: os.fsync(directory)
|
|
finally: os.close(directory)
|
|
finally:
|
|
try: os.unlink(temporary)
|
|
except FileNotFoundError: pass
|
|
|
|
|
|
def load_json(path: Path, limit: int) -> dict:
|
|
raw = path.read_bytes()
|
|
if len(raw) > limit: raise RuntimeError(f"{path.name} exceeds its bound")
|
|
value = json.loads(raw)
|
|
if not isinstance(value, dict): raise RuntimeError(f"{path.name} must contain an object")
|
|
return value
|
|
|
|
|
|
def update(state_path: Path, state: dict, phase: str, **fields: object) -> None:
|
|
if phase not in PHASES: raise RuntimeError("invalid run phase")
|
|
state.update(fields); state["phase"] = phase
|
|
state["progress_revision"] = int(state.get("progress_revision", 0)) + 1
|
|
state["updated_at"] = datetime.now(timezone.utc).isoformat()
|
|
atomic_write(state_path, state)
|
|
|
|
|
|
def validate_result(value: dict, run_id: str) -> dict:
|
|
required = {"format", "version", "run_id", "producer_peer_id", "consumer_peer_id",
|
|
"inbound_generation_id", "outbound_generation_id", "manifest_sha256",
|
|
"result", "sessions_reconciled", "domains", "drafts", "ai_midpoint", "ai_post_sync"}
|
|
if set(value) != required or value.get("format") != "trainlog-sync-worker-report" or value.get("version") != 1:
|
|
raise RuntimeError("malformed peer worker report")
|
|
if value.get("run_id") != run_id or value.get("result") != "completed":
|
|
raise RuntimeError("peer worker did not complete the correlated run")
|
|
for identity, prefix in ((value["producer_peer_id"], "peer_"),
|
|
(value["consumer_peer_id"], "peer_"),
|
|
(value["inbound_generation_id"], "gen_"),
|
|
(value["outbound_generation_id"], "gen_")):
|
|
if not isinstance(identity, str) or not identity.startswith(prefix) or len(identity) > 64:
|
|
raise RuntimeError("invalid worker identity")
|
|
if not isinstance(value["sessions_reconciled"], int) or value["sessions_reconciled"] < 0:
|
|
raise RuntimeError("invalid sessions_reconciled")
|
|
if not isinstance(value["domains"], dict) or not all(isinstance(k, str) and isinstance(v, bool) for k, v in value["domains"].items()):
|
|
raise RuntimeError("invalid domain coverage")
|
|
if not isinstance(value["drafts"], list) or len(value["drafts"]) > 32:
|
|
raise RuntimeError("invalid draft summary")
|
|
return value
|
|
|
|
|
|
def run(args: argparse.Namespace) -> int:
|
|
state_path, config_path = Path(args.state), Path(args.config)
|
|
state = load_json(state_path, MAX_REPORT)
|
|
config = load_json(config_path, MAX_CONFIG)
|
|
if state.get("run_id") != args.run_id or state.get("request_id") != args.request_id:
|
|
raise RuntimeError("admission identity mismatch")
|
|
if set(config) != {"format", "version", "enabled", "mode", "capabilities", "peer_command", "timeout_seconds"}:
|
|
raise RuntimeError("invalid trusted sync configuration")
|
|
if config["format"] != "trainlog-sync-orchestrator-config" or config["version"] != 1 or config["enabled"] is not True:
|
|
raise RuntimeError("full-generation synchronization is disabled")
|
|
required_caps = {"generation-manifest-v1", "generation-ack-v1", "causal-delete-v1", "mobile-history-v4", "execution-draft-v1"}
|
|
caps = config["capabilities"]
|
|
if not isinstance(caps, list) or not required_caps.issubset(set(caps)):
|
|
update(state_path, state, "explicitly_degraded", result="incompatible",
|
|
missing_capabilities=sorted(required_caps - set(caps if isinstance(caps, list) else [])),
|
|
diagnostic="Peer lacks mandatory full-generation capabilities", finished_at=datetime.now(timezone.utc).isoformat())
|
|
return 3
|
|
command = config["peer_command"]
|
|
if not isinstance(command, list) or not command or len(command) > 16 or not all(isinstance(part, str) and 0 < len(part) <= 4096 for part in command):
|
|
raise RuntimeError("invalid trusted peer command")
|
|
timeout = config["timeout_seconds"]
|
|
if not isinstance(timeout, int) or timeout < 1 or timeout > 900:
|
|
raise RuntimeError("invalid worker timeout")
|
|
# CONTRACT: this is the same XDG data-directory lock used by the legacy
|
|
# C engine, so Web, TUI and daemon admissions cannot overlap.
|
|
lock_path = Path(args.database).parent / "sync.lock"
|
|
lock_fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600)
|
|
try:
|
|
try: fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
update(state_path, state, "failed", result="conflict", diagnostic="Another synchronization owns the common lock", finished_at=datetime.now(timezone.utc).isoformat())
|
|
return 4
|
|
update(state_path, state, "waiting_android_publication", result="running", diagnostic="")
|
|
update(state_path, state, "running")
|
|
environment = os.environ.copy()
|
|
environment.update({"TRAINLOG_SYNC_RUN_ID": args.run_id,
|
|
"TRAINLOG_SYNC_TRIGGER": state["trigger"],
|
|
"TRAINLOG_SYNC_DATABASE": args.database})
|
|
child = subprocess.Popen(command, stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=environment,
|
|
start_new_session=True)
|
|
try:
|
|
stdout, stderr = child.communicate(timeout=timeout)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(child.pid, signal.SIGTERM)
|
|
try: stdout, stderr = child.communicate(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(child.pid, signal.SIGKILL); stdout, stderr = child.communicate()
|
|
raise RuntimeError("peer worker timed out")
|
|
if len(stdout) > MAX_REPORT or len(stderr) > MAX_REPORT:
|
|
raise RuntimeError("peer worker output exceeds 64 KiB")
|
|
if child.returncode != 0:
|
|
detail = stderr.decode("utf-8", "replace").strip()[:MAX_DIAGNOSTIC]
|
|
raise RuntimeError(f"peer worker exited {child.returncode}: {detail}")
|
|
result = validate_result(json.loads(stdout), args.run_id)
|
|
update(state_path, state, "local_import_committed",
|
|
inbound_generation_id=result["inbound_generation_id"],
|
|
producer_peer_id=result["producer_peer_id"],
|
|
consumer_peer_id=result["consumer_peer_id"],
|
|
sessions_reconciled=result["sessions_reconciled"], domains=result["domains"], drafts=result["drafts"])
|
|
update(state_path, state, "published", outbound_generation_id=result["outbound_generation_id"])
|
|
update(state_path, state, "waiting_acknowledgement")
|
|
update(state_path, state, "peer_consumed", manifest_sha256=result["manifest_sha256"])
|
|
update(state_path, state, "completed", result="completed", diagnostic="",
|
|
ai_midpoint=result["ai_midpoint"], ai_post_sync=result["ai_post_sync"],
|
|
finished_at=datetime.now(timezone.utc).isoformat())
|
|
return 0
|
|
finally:
|
|
os.close(lock_fd)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--database", required=True); parser.add_argument("--state", required=True)
|
|
parser.add_argument("--config", required=True); parser.add_argument("--run-id", required=True)
|
|
parser.add_argument("--request-id", required=True)
|
|
args = parser.parse_args()
|
|
try: return run(args)
|
|
except (OSError, ValueError, json.JSONDecodeError, RuntimeError, subprocess.TimeoutExpired) as error:
|
|
try:
|
|
path = Path(args.state); state = load_json(path, MAX_REPORT)
|
|
update(path, state, "failed", result="failed", diagnostic=str(error)[:MAX_DIAGNOSTIC],
|
|
finished_at=datetime.now(timezone.utc).isoformat())
|
|
except Exception: pass
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__": raise SystemExit(main())
|