|
| 1 | +"""Temporary compact observation measurement; no writes between paired reads.""" |
| 2 | +import fcntl as _trace_fcntl |
| 3 | +import json, os, sys, time, uuid |
| 4 | + |
| 5 | +TRACE_LIMIT = 8 * 1024 * 1024 |
| 6 | +ARM = "/proof/diagnostic-arm.json" |
| 7 | +TRACE = "/proof/diagnostic-trace" |
| 8 | + |
| 9 | +def differences(before, after, path=""): |
| 10 | + if isinstance(before, dict) and isinstance(after, dict): |
| 11 | + result = [] |
| 12 | + for key in sorted(set(before) | set(after)): |
| 13 | + child = path+"."+key if path else key |
| 14 | + if key not in before or key not in after: |
| 15 | + result.append({"path": child, "before_present": key in before, "after_present": key in after, |
| 16 | + "before": before.get(key), "after": after.get(key)}) |
| 17 | + else: |
| 18 | + result.extend(differences(before[key], after[key], child)) |
| 19 | + return result |
| 20 | + return [] if before == after else [{"path": path, "before": before, "after": after}] |
| 21 | + |
| 22 | +class Recorder: |
| 23 | + def __init__(self, original, context): |
| 24 | + self.original, self.context = original, context |
| 25 | + self.items, self.sync_start, self.sync_done = [], None, None |
| 26 | + self.arm = None |
| 27 | + if os.path.exists(ARM): |
| 28 | + with open(ARM, encoding="utf8") as source: |
| 29 | + value = json.load(source) |
| 30 | + if context and value["worktreeId"] == context.get("from"): |
| 31 | + self.arm = value |
| 32 | + def snapshot(self, *args): |
| 33 | + began = {"wall_ns": time.time_ns(), "mono_ns": time.monotonic_ns()} |
| 34 | + value = self.original(*args) |
| 35 | + ended = {"wall_ns": time.time_ns(), "mono_ns": time.monotonic_ns()} |
| 36 | + if len(self.items) >= 8: |
| 37 | + raise RuntimeError("DIAGNOSTIC_SNAPSHOT_BOUND") |
| 38 | + self.items.append({"began": began, "ended": ended, "value": value}) |
| 39 | + return value |
| 40 | + def finish(self, ns): |
| 41 | + stable = self.items[-2]["value"] == self.items[-1]["value"] if len(self.items) >= 2 else None |
| 42 | + row = {"version": 1, "observation_id": uuid.uuid4().hex, |
| 43 | + "synthetic": ns.get("_diag_synthetic", False), "armed": self.arm is not None, |
| 44 | + "operation": self.context or {"operation": None, "id": None, "from": None}, |
| 45 | + "reads": ns.get("reads"), "stable": stable, "fsid": ns["fsid"], |
| 46 | + "sync_start_ns": self.sync_start, "sync_done_ns": self.sync_done, |
| 47 | + "finished_mono_ns": time.monotonic_ns(), "finished_wall_ns": time.time_ns(), |
| 48 | + "exception": str(sys.exc_info()[1]) if sys.exc_info()[0] else None} |
| 49 | + # Stable two-read unarmed observations are summaries. Every unequal pair and every armed pair is retained. |
| 50 | + if self.arm is not None or not stable or ns.get("reads", 0) > 2: |
| 51 | + pairs = [] |
| 52 | + for offset in range(0, len(self.items)-1, 2): |
| 53 | + a, b = self.items[offset:offset+2] |
| 54 | + delta = differences(a["value"], b["value"]) |
| 55 | + pairs.append({"index": offset//2+1, "first_began": a["began"], "first_ended": a["ended"], |
| 56 | + "second_began": b["began"], "second_ended": b["ended"], |
| 57 | + "equal": not delta, "first_difference": delta[0]["path"] if delta else None, |
| 58 | + "differences": delta, "first_dropping": a["value"]["dropping"], |
| 59 | + "second_dropping": b["value"]["dropping"], |
| 60 | + "first_subvolumes_count": len(a["value"]["subvolumes"]), |
| 61 | + "second_subvolumes_count": len(b["value"]["subvolumes"])}) |
| 62 | + row["pairs"] = pairs |
| 63 | + if self.arm is not None: |
| 64 | + row.update(arm=self.arm, pool=ns["pool"], helper_pid=os.getpid()) |
| 65 | + data = (json.dumps(row, sort_keys=True, separators=(",", ":"))+"\n").encode() |
| 66 | + os.makedirs(TRACE, mode=0o755, exist_ok=True) |
| 67 | + fd = os.open(os.path.join(TRACE, "observations.jsonl"), os.O_WRONLY|os.O_CREAT|os.O_APPEND|os.O_NOFOLLOW, 0o644) |
| 68 | + try: |
| 69 | + _trace_fcntl.flock(fd, _trace_fcntl.LOCK_EX) |
| 70 | + if os.fstat(fd).st_size+len(data) > TRACE_LIMIT: |
| 71 | + raise RuntimeError("DIAGNOSTIC_TRACE_BUDGET") |
| 72 | + remaining = memoryview(data) |
| 73 | + while remaining: |
| 74 | + written = os.write(fd, remaining) |
| 75 | + if written <= 0: |
| 76 | + raise RuntimeError("DIAGNOSTIC_SHORT_WRITE") |
| 77 | + remaining = remaining[written:] |
| 78 | + os.fchmod(fd, 0o644); os.chmod(TRACE, 0o755) |
| 79 | + os.fsync(fd) |
| 80 | + finally: |
| 81 | + os.close(fd) |
| 82 | + |
| 83 | +_diag_context = json.loads(sys.argv[5]) if len(sys.argv) > 5 else None |
| 84 | +_diag = Recorder(snapshot, _diag_context) |
| 85 | +snapshot = _diag.snapshot |
0 commit comments