430 lines
15 KiB
Python
430 lines
15 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 selectors
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import ctypes
|
|
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",
|
|
}
|
|
|
|
|
|
class OrchestratorInterrupted(RuntimeError):
|
|
pass
|
|
|
|
|
|
_owned_process_group: int | None = None
|
|
|
|
|
|
def _forward_termination(signum: int, _frame: object) -> None:
|
|
"""Forward shutdown only to the exactly-owned helper process group."""
|
|
if _owned_process_group is not None:
|
|
try:
|
|
os.killpg(_owned_process_group, signal.SIGTERM)
|
|
except ProcessLookupError:
|
|
pass
|
|
raise OrchestratorInterrupted(f"orchestrator received signal {signum}")
|
|
|
|
|
|
def _child_setup() -> None:
|
|
"""Create an owned group and ask Linux to terminate it if this parent dies."""
|
|
os.setsid()
|
|
libc = ctypes.CDLL(None)
|
|
if libc.prctl(1, signal.SIGTERM) != 0: # PR_SET_PDEATHSIG
|
|
raise OSError("cannot establish parent-death signal")
|
|
|
|
|
|
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:
|
|
with path.open("rb") as stream:
|
|
raw = stream.read(limit + 1)
|
|
if len(raw) > limit:
|
|
raise RuntimeError(f"{path.name} exceeds its bound")
|
|
value = json.loads(raw.decode("utf-8", "strict"))
|
|
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 re.fullmatch(
|
|
re.escape(prefix)
|
|
+ r"[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}",
|
|
identity,
|
|
):
|
|
raise RuntimeError("invalid worker identity")
|
|
if type(value["sessions_reconciled"]) is not 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 read_worker(
|
|
child: subprocess.Popen[bytes], state_path: Path, state: dict, run_id: str, timeout: int
|
|
) -> tuple[dict, bytes]:
|
|
selector = selectors.DefaultSelector()
|
|
assert child.stdout is not None and child.stderr is not None
|
|
selector.register(child.stdout, selectors.EVENT_READ, "stdout")
|
|
selector.register(child.stderr, selectors.EVENT_READ, "stderr")
|
|
buffers = {"stdout": bytearray(), "stderr": bytearray()}
|
|
line_buffer = bytearray()
|
|
final: dict | None = None
|
|
deadline = time.monotonic() + timeout
|
|
allowed = [
|
|
"waiting_android_publication",
|
|
"running",
|
|
"local_import_committed",
|
|
"published",
|
|
"waiting_acknowledgement",
|
|
"peer_consumed",
|
|
]
|
|
last_index = -1
|
|
while selector.get_map():
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise RuntimeError("peer worker timed out")
|
|
for key, _ in selector.select(min(remaining, 0.25)):
|
|
chunk = os.read(key.fileobj.fileno(), 4096)
|
|
if not chunk:
|
|
selector.unregister(key.fileobj)
|
|
continue
|
|
target = buffers[key.data]
|
|
if len(target) + len(chunk) > MAX_REPORT:
|
|
raise RuntimeError(f"peer worker {key.data} exceeds 64 KiB")
|
|
target.extend(chunk)
|
|
if key.data == "stdout":
|
|
line_buffer.extend(chunk)
|
|
while b"\n" in line_buffer:
|
|
raw, _, rest = line_buffer.partition(b"\n")
|
|
line_buffer[:] = rest
|
|
if len(raw) > 8192:
|
|
raise RuntimeError("progress frame exceeds 8 KiB")
|
|
value = json.loads(raw.decode("utf-8", "strict"))
|
|
if value.get("format") == "trainlog-sync-progress":
|
|
if (
|
|
set(value)
|
|
- {
|
|
"format",
|
|
"version",
|
|
"run_id",
|
|
"phase",
|
|
"producer_peer_id",
|
|
"consumer_peer_id",
|
|
"inbound_generation_id",
|
|
"outbound_generation_id",
|
|
"manifest_sha256",
|
|
}
|
|
or value.get("version") != 1
|
|
or value.get("run_id") != run_id
|
|
or value.get("phase") not in allowed
|
|
):
|
|
raise RuntimeError("invalid progress frame")
|
|
index = allowed.index(value["phase"])
|
|
if index < last_index:
|
|
raise RuntimeError("progress phase regressed")
|
|
last_index = index
|
|
update(
|
|
state_path,
|
|
state,
|
|
value["phase"],
|
|
**{
|
|
k: v
|
|
for k, v in value.items()
|
|
if k not in {"format", "version", "run_id", "phase"}
|
|
},
|
|
)
|
|
else:
|
|
final = value
|
|
child.wait()
|
|
if line_buffer.strip():
|
|
raise RuntimeError("unterminated worker frame")
|
|
if child.returncode != 0:
|
|
detail = bytes(buffers["stderr"]).decode("utf-8", "replace").strip()[:MAX_DIAGNOSTIC]
|
|
raise RuntimeError(f"peer worker exited {child.returncode}: {detail}")
|
|
if final is None:
|
|
raise RuntimeError("peer worker omitted final report")
|
|
return validate_result(final, run_id), bytes(buffers["stderr"])
|
|
|
|
|
|
def run(args: argparse.Namespace) -> int:
|
|
global _owned_process_group
|
|
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",
|
|
"expected_peer_id",
|
|
"transport_root",
|
|
"owned_root",
|
|
"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")
|
|
if config["mode"] not in ("directory", "mtp"):
|
|
raise RuntimeError("invalid transport mode")
|
|
for key in ("transport_root", "owned_root"):
|
|
if not isinstance(config[key], str) or not Path(config[key]).is_absolute():
|
|
raise RuntimeError("trusted paths must be absolute")
|
|
if not isinstance(config["expected_peer_id"], str) or not re.fullmatch(
|
|
r"peer_[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}",
|
|
config["expected_peer_id"],
|
|
):
|
|
raise RuntimeError("invalid expected peer identity")
|
|
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, "running", result="running", diagnostic="")
|
|
environment = os.environ.copy()
|
|
environment.update(
|
|
{
|
|
"TRAINLOG_SYNC_RUN_ID": args.run_id,
|
|
"TRAINLOG_SYNC_TRIGGER": state["trigger"],
|
|
"TRAINLOG_SYNC_DATABASE": args.database,
|
|
}
|
|
)
|
|
command = [
|
|
sys.executable,
|
|
str(Path(__file__).with_name("sync_peer_worker.py")),
|
|
"--database",
|
|
args.database,
|
|
"--transport-root",
|
|
config["transport_root"],
|
|
"--owned-root",
|
|
config["owned_root"],
|
|
"--run-id",
|
|
args.run_id,
|
|
"--expected-peer",
|
|
config["expected_peer_id"],
|
|
"--timeout",
|
|
str(timeout),
|
|
"--mode",
|
|
config["mode"],
|
|
]
|
|
if config["mode"] == "mtp":
|
|
adapter = os.environ.get(
|
|
"TRAINLOG_SYNC_MTP_ADAPTER",
|
|
str(
|
|
Path(__file__).resolve().parents[1]
|
|
/ "build/tui/trainlog-generation-mtp-adapter"
|
|
),
|
|
)
|
|
if not Path(adapter).is_absolute() or not os.access(adapter, os.X_OK):
|
|
raise RuntimeError("fixed MTP adapter executable is unavailable")
|
|
command.extend(["--mtp-adapter", adapter])
|
|
child = subprocess.Popen(
|
|
command,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=environment,
|
|
preexec_fn=_child_setup,
|
|
)
|
|
_owned_process_group = child.pid
|
|
try:
|
|
result, _ = read_worker(child, state_path, state, args.run_id, timeout)
|
|
except Exception:
|
|
if child.poll() is None:
|
|
os.killpg(child.pid, signal.SIGTERM)
|
|
try:
|
|
child.wait(5)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(child.pid, signal.SIGKILL)
|
|
child.wait()
|
|
raise
|
|
finally:
|
|
_owned_process_group = None
|
|
update(
|
|
state_path,
|
|
state,
|
|
"completed",
|
|
result="completed",
|
|
diagnostic="",
|
|
inbound_generation_id=result["inbound_generation_id"],
|
|
outbound_generation_id=result["outbound_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"],
|
|
manifest_sha256=result["manifest_sha256"],
|
|
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()
|
|
signal.signal(signal.SIGTERM, _forward_termination)
|
|
signal.signal(signal.SIGINT, _forward_termination)
|
|
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)
|
|
phase = "interrupted" if isinstance(error, OrchestratorInterrupted) else "failed"
|
|
update(
|
|
path,
|
|
state,
|
|
phase,
|
|
result=phase,
|
|
diagnostic=str(error)[:MAX_DIAGNOSTIC],
|
|
finished_at=datetime.now(timezone.utc).isoformat(),
|
|
)
|
|
except Exception:
|
|
pass
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|