293 lines
12 KiB
Python
Executable File
293 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Watch, authorize, and safely restart the Blender/WBV2 scheduler process.
|
|
|
|
The watchdog delegates terminal queue advancement to the source-backed
|
|
authorizer. It never edits the queue directly, retries blocked checkpoints, or
|
|
deletes evidence.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from wbv2_runtime import ROOT, SCHEDULER_CHECKPOINT, SCHEDULER_LOCK, lock_owner, queue_pointer, scheduler_args, terminal_pointer
|
|
|
|
|
|
WATCHDOG_LOCK = ROOT / "tests/golden/WBV2/my-tools-watchdog.lock"
|
|
WATCHDOG_PID = ROOT / "tests/golden/WBV2/my-tools-watchdog.pid"
|
|
WATCHDOG_STATUS = ROOT / "tests/golden/WBV2/my-tools-watchdog.json"
|
|
AUTHORIZATION_LOG = ROOT / "tests/golden/WBV2/my-tools-authorization.log"
|
|
|
|
|
|
def now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def write_json(path: Path, value: dict) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
current = json.loads(path.read_text(encoding="utf-8"))
|
|
stable_current = {key: item for key, item in current.items() if key != "updatedAt"}
|
|
stable_next = {key: item for key, item in value.items() if key != "updatedAt"}
|
|
if stable_current == stable_next:
|
|
return
|
|
except (OSError, json.JSONDecodeError):
|
|
pass
|
|
temporary = path.with_suffix(path.suffix + f".tmp-{os.getpid()}")
|
|
temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
temporary.replace(path)
|
|
|
|
|
|
def process_alive(pid: object) -> bool:
|
|
if not isinstance(pid, int) or pid <= 0:
|
|
return False
|
|
try:
|
|
os.kill(pid, 0)
|
|
except ProcessLookupError:
|
|
return False
|
|
except PermissionError:
|
|
return True
|
|
return True
|
|
|
|
|
|
def acquire_lock() -> int:
|
|
try:
|
|
handle = os.open(WATCHDOG_LOCK, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
|
|
except FileExistsError:
|
|
existing = {}
|
|
try:
|
|
existing = json.loads(WATCHDOG_LOCK.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
pass
|
|
owner = existing.get("pid") if isinstance(existing, dict) else None
|
|
if isinstance(owner, int) and not process_alive(owner):
|
|
WATCHDOG_LOCK.unlink(missing_ok=True)
|
|
return acquire_lock()
|
|
raise SystemExit(f"another WBV2 watchdog appears to be running: {WATCHDOG_LOCK}")
|
|
os.write(handle, f"{{\"pid\":{os.getpid()},\"startedAt\":{json.dumps(now())}}}\n".encode())
|
|
return handle
|
|
|
|
|
|
def checkpoint() -> dict:
|
|
try:
|
|
value = json.loads(SCHEDULER_CHECKPOINT.read_text(encoding="utf-8"))
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
return {}
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def checkpoint_task(value: dict) -> str | None:
|
|
for key in ("current", "before", "after"):
|
|
candidate = value.get(key)
|
|
if isinstance(candidate, dict) and isinstance(candidate.get("task"), str):
|
|
return candidate["task"]
|
|
return None
|
|
|
|
|
|
def authorize_terminal_queue() -> tuple[bool, str]:
|
|
"""Ask the repository authorizer for one source-backed successor batch."""
|
|
result = subprocess.run(
|
|
["python3", str(ROOT / "my-tools/authorize_next_wbv2_task.py"), "--write"],
|
|
cwd=ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
AUTHORIZATION_LOG.parent.mkdir(parents=True, exist_ok=True)
|
|
AUTHORIZATION_LOG.write_text(
|
|
f"$ python3 my-tools/authorize_next_wbv2_task.py --write\n{result.stdout}{result.stderr}",
|
|
encoding="utf-8",
|
|
)
|
|
output = f"{result.stdout}{result.stderr}".strip()
|
|
return result.returncode == 0, output
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--interval", type=float, default=10.0)
|
|
parser.add_argument("--restart-delay", type=float, default=5.0)
|
|
parser.add_argument("--max-restarts", type=int, default=0, help="0 means unlimited safe restarts")
|
|
parser.add_argument("--max-iterations", type=int, default=0)
|
|
parser.add_argument("--commit", action="store_true", default=os.getenv("WBV2_AUTO_COMMIT", "1") != "0")
|
|
parser.add_argument("--push", action="store_true", default=os.getenv("WBV2_AUTO_PUSH", "1") != "0")
|
|
parser.add_argument("--push-retries", type=int, default=6)
|
|
parser.add_argument("--auto-authorize", dest="auto_authorize", action="store_true", default=os.getenv("WBV2_AUTO_AUTHORIZE", "1") != "0", help="authorize a source-backed successor when the queue reaches a terminal task")
|
|
parser.add_argument("--no-auto-authorize", dest="auto_authorize", action="store_false")
|
|
parser.add_argument("--wait-on-empty", dest="wait_on_empty", action="store_true", default=os.getenv("WBV2_WAIT_ON_EMPTY", "0") != "0", help="keep the watchdog alive when no source-backed batch is currently configured")
|
|
parser.add_argument("--no-wait-on-empty", dest="wait_on_empty", action="store_false")
|
|
args = parser.parse_args()
|
|
if args.interval <= 0 or args.restart_delay < 0 or args.max_restarts < 0 or args.max_iterations < 0:
|
|
parser.error("interval/delay/limits must be non-negative; interval must be positive")
|
|
if args.push and not args.commit:
|
|
parser.error("--push requires --commit")
|
|
|
|
WATCHDOG_LOCK.parent.mkdir(parents=True, exist_ok=True)
|
|
handle = acquire_lock()
|
|
os.close(handle)
|
|
WATCHDOG_PID.write_text(f"{os.getpid()}\n", encoding="ascii")
|
|
restarts = 0
|
|
child: subprocess.Popen[bytes] | None = None
|
|
stopped = False
|
|
|
|
def stop(signum: int, _frame: object) -> None:
|
|
nonlocal stopped
|
|
stopped = True
|
|
if child and child.poll() is None:
|
|
try:
|
|
os.killpg(child.pid, signum)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
signal.signal(signal.SIGINT, stop)
|
|
signal.signal(signal.SIGTERM, stop)
|
|
try:
|
|
while not stopped:
|
|
pointer = queue_pointer()
|
|
saved_checkpoint = checkpoint()
|
|
state = saved_checkpoint.get("state")
|
|
applies = checkpoint_task(saved_checkpoint) == pointer.get("task")
|
|
if terminal_pointer(pointer) and args.auto_authorize:
|
|
authorized, output = authorize_terminal_queue()
|
|
if authorized:
|
|
write_json(
|
|
WATCHDOG_STATUS,
|
|
{
|
|
"schemaVersion": 1,
|
|
"status": "authorized",
|
|
"reason": "source_backed_successor",
|
|
"task": pointer.get("task"),
|
|
"updatedAt": now(),
|
|
"restartCount": restarts,
|
|
"authorizationLog": str(AUTHORIZATION_LOG.relative_to(ROOT)),
|
|
},
|
|
)
|
|
child = None
|
|
time.sleep(0.2)
|
|
continue
|
|
reason = "no_configured_unresolved_source_batch" if "no configured unresolved Blender source batch remains" in output else "successor_authorization_failed"
|
|
write_json(
|
|
WATCHDOG_STATUS,
|
|
{
|
|
"schemaVersion": 1,
|
|
"status": "stopped" if reason == "no_configured_unresolved_source_batch" else "blocked",
|
|
"reason": reason,
|
|
"task": pointer.get("task"),
|
|
"updatedAt": now(),
|
|
"restartCount": restarts,
|
|
"authorizationLog": str(AUTHORIZATION_LOG.relative_to(ROOT)),
|
|
},
|
|
)
|
|
if reason == "no_configured_unresolved_source_batch" and args.wait_on_empty:
|
|
write_json(
|
|
WATCHDOG_STATUS,
|
|
{
|
|
"schemaVersion": 1,
|
|
"status": "waiting_for_source_batch",
|
|
"reason": reason,
|
|
"task": pointer.get("task"),
|
|
"updatedAt": now(),
|
|
"restartCount": restarts,
|
|
"authorizationLog": str(AUTHORIZATION_LOG.relative_to(ROOT)),
|
|
},
|
|
)
|
|
# Keep the planner heartbeat responsive while no exact
|
|
# source-backed batch is available. Each short cycle
|
|
# retries authorization and records the current pending
|
|
# inventory instead of entering a long silent sleep.
|
|
child = None
|
|
time.sleep(args.interval)
|
|
continue
|
|
return 0 if reason == "no_configured_unresolved_source_batch" else 2
|
|
if terminal_pointer(pointer) or (applies and state in {"complete", "blocked", "stopped"}):
|
|
write_json(
|
|
WATCHDOG_STATUS,
|
|
{
|
|
"schemaVersion": 1,
|
|
"status": "stopped",
|
|
"reason": state or "terminal",
|
|
"task": pointer.get("task"),
|
|
"updatedAt": now(),
|
|
"restartCount": restarts,
|
|
},
|
|
)
|
|
break
|
|
if not lock_owner(SCHEDULER_LOCK)["alive"] and child is None and applies and state in {"started", "running", "preflight"}:
|
|
write_json(
|
|
WATCHDOG_STATUS,
|
|
{
|
|
"schemaVersion": 1,
|
|
"status": "blocked",
|
|
"reason": f"unsafe_scheduler_checkpoint_{state}",
|
|
"task": pointer.get("task"),
|
|
"updatedAt": now(),
|
|
"restartCount": restarts,
|
|
},
|
|
)
|
|
return 2
|
|
if not lock_owner(SCHEDULER_LOCK)["alive"] and (child is None or child.poll() is not None):
|
|
if args.max_restarts and restarts >= args.max_restarts:
|
|
write_json(
|
|
WATCHDOG_STATUS,
|
|
{
|
|
"schemaVersion": 1,
|
|
"status": "blocked",
|
|
"reason": "max_safe_restarts_exceeded",
|
|
"task": pointer.get("task"),
|
|
"updatedAt": now(),
|
|
"restartCount": restarts,
|
|
},
|
|
)
|
|
return 2
|
|
child = subprocess.Popen(
|
|
scheduler_args(commit=args.commit, push=args.push, max_iterations=args.max_iterations),
|
|
cwd=ROOT,
|
|
start_new_session=True,
|
|
)
|
|
restarts += 1
|
|
write_json(
|
|
WATCHDOG_STATUS,
|
|
{
|
|
"schemaVersion": 1,
|
|
"status": "restarted",
|
|
"reason": "scheduler_missing",
|
|
"task": pointer.get("task"),
|
|
"schedulerPid": child.pid,
|
|
"updatedAt": now(),
|
|
"restartCount": restarts,
|
|
},
|
|
)
|
|
time.sleep(args.restart_delay)
|
|
else:
|
|
write_json(
|
|
WATCHDOG_STATUS,
|
|
{
|
|
"schemaVersion": 1,
|
|
"status": "watching",
|
|
"task": pointer.get("task"),
|
|
"schedulerPid": lock_owner(SCHEDULER_LOCK)["pid"],
|
|
"updatedAt": now(),
|
|
"restartCount": restarts,
|
|
},
|
|
)
|
|
time.sleep(args.interval)
|
|
finally:
|
|
if child and child.poll() is None:
|
|
try:
|
|
os.killpg(child.pid, signal.SIGTERM)
|
|
except ProcessLookupError:
|
|
pass
|
|
WATCHDOG_PID.unlink(missing_ok=True)
|
|
WATCHDOG_LOCK.unlink(missing_ok=True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|