forked from chiki2bum2/SniperGold_ML
Chiki Bum
0d5ca3c197
Implements the frozen P3_DATA_ENGINE_V1_SPEC (SHA 84bf0f217ffba51197112a6bbacbcc297058e04b5ca47f0459028fe33e0321e5). Components: engine/ producer (certify, chunkmap, parse, canonical, worker, dispatcher, aggregate, storage, journal, checkpoint, lock, evidence, manifest, run_complete, dataset_builder, cli) + engine/verify independent verifier (vparse, vaggregate, vinvariants, vcompare); headless CLI sniper-data; golden corpus G01-G17; unit/property/adversarial/mutation/legacy-diff/CLI suites. Qualification verdict: QUALIFIED (all 9 mandatory gates pass; independent verifier accepted). Spec, governance record, and legacy checkpoint untouched. G-14 NOT AUTHORIZED honored; no real-data processing, no pilot, no workload 46, no chunk 760 access.
101 lines
No EOL
2.9 KiB
Python
101 lines
No EOL
2.9 KiB
Python
"""Run lock management (spec 21, G-11).
|
|
|
|
Heartbeat refresh every 60 s; lock is stale after 300 s without refresh.
|
|
Default policy REFUSES to take over a lock, even a stale one; release of a
|
|
stale lock requires the explicit human ``--force-release-lock`` flag. The
|
|
Engine never kills or takes over a live process automatically.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
|
|
|
|
class LockError(Exception):
|
|
pass
|
|
|
|
|
|
LOCK_HEARTBEAT_SEC = 60
|
|
LOCK_STALE_SEC = 300
|
|
|
|
|
|
def _read(path):
|
|
try:
|
|
with open(path, "rb") as fh:
|
|
return json.loads(fh.read().decode("utf-8"))
|
|
except FileNotFoundError:
|
|
return None
|
|
except Exception as e:
|
|
raise LockError("unreadable lock file: %s" % e)
|
|
|
|
|
|
def lock_age(info):
|
|
return time.time() - info.get("heartbeat_epoch_s", 0)
|
|
|
|
|
|
def lock_status(path):
|
|
"""Read-only lock inspection. Returns dict or None."""
|
|
if not os.path.exists(path):
|
|
return None
|
|
info = _read(path)
|
|
if info is None:
|
|
return None
|
|
info = dict(info)
|
|
info["_age_sec"] = lock_age(info)
|
|
info["_stale"] = info["_age_sec"] > LOCK_STALE_SEC
|
|
return info
|
|
|
|
|
|
class LockHandle:
|
|
def __init__(self, path, run_id, pid, force_release=False):
|
|
self.path = path
|
|
self.run_id = run_id
|
|
self.pid = pid
|
|
self.force_release = force_release
|
|
self.held = False
|
|
self._last_touch = 0.0
|
|
|
|
def acquire(self):
|
|
existing = _read(self.path)
|
|
now = time.time()
|
|
if existing is not None:
|
|
age = now - existing.get("heartbeat_epoch_s", 0)
|
|
if age <= LOCK_STALE_SEC:
|
|
raise LockError("lock held by live run (pid %s, run %s)"
|
|
% (existing.get("pid"), existing.get("run_id")))
|
|
if not self.force_release:
|
|
raise LockError(
|
|
"stale lock present (age %.0f s); refusing takeover; "
|
|
"explicit --force-release-lock is required (human gate G-11)"
|
|
% age)
|
|
os.makedirs(os.path.dirname(os.path.abspath(self.path)), exist_ok=True)
|
|
self._write(now)
|
|
self.held = True
|
|
|
|
def _write(self, now):
|
|
info = {"pid": self.pid, "run_id": self.run_id,
|
|
"heartbeat_epoch_s": now}
|
|
tmp = self.path + ".tmp"
|
|
with open(tmp, "w", encoding="utf-8") as fh:
|
|
json.dump(info, fh, sort_keys=True, separators=(",", ":"))
|
|
fh.flush()
|
|
os.fsync(fh.fileno())
|
|
os.replace(tmp, self.path)
|
|
|
|
def touch(self):
|
|
if not self.held:
|
|
return
|
|
now = time.time()
|
|
if now - self._last_touch < LOCK_HEARTBEAT_SEC / 2.0:
|
|
return
|
|
self._write(now)
|
|
self._last_touch = now
|
|
|
|
def release(self):
|
|
if not self.held:
|
|
return
|
|
try:
|
|
os.remove(self.path)
|
|
except FileNotFoundError:
|
|
pass
|
|
self.held = False |