| 1 | """Crash-point family: kill -9 the client at time offsets and at every durability boundary."""
|
| 2 |
|
| 3 | from __future__ import annotations
|
| 4 |
|
| 5 | import time
|
| 6 | from pathlib import Path
|
| 7 | from typing import Callable, Optional
|
| 8 |
|
| 9 | from . import HARNESS_ERROR
|
| 10 | from .crashpoints import CrashInstrument, boundary_kinds, count_boundaries
|
| 11 | from .engine import Calibration, Trial, actionable, classify, converge_check, worktree_verdict
|
| 12 | from .fixtures import Fixture
|
| 13 | from .procs import run, run_killed_after
|
| 14 | from .readers import worktree_manifest
|
| 15 |
|
| 16 | TIME_FRACTIONS = (0.02, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 0.97)
|
| 17 | MAX_SYSCALL_POINTS = 16
|
| 18 |
|
| 19 |
|
| 20 | def calibrate(fixture: Fixture, subject_bin: str, work: Path, instrument: Optional[CrashInstrument]) -> Calibration:
|
| 21 | trial = Trial(fixture, subject_bin, work, "calibrate")
|
| 22 | try:
|
| 23 | trial.start_server()
|
| 24 | trial.pre()
|
| 25 | res = trial.adapter.run(trial.op_argv(), trial.op_cwd())
|
| 26 | trial.settle()
|
| 27 | if res.returncode != 0:
|
| 28 | return Calibration(res.elapsed_ms, None, None, None, None, ok=False,
|
| 29 | detail=f"uncrashed op failed rc={res.returncode}: {res.stderr[-300:]}")
|
| 30 | repo = trial.target_repo()
|
| 31 | tree, status = trial.adapter.tree(repo)
|
| 32 | calib = Calibration(res.elapsed_ms, tree, worktree_manifest(repo), trial.adapter.head(repo),
|
| 33 | trial.server_head() if fixture.op in ("push", "clone", "pull_ff", "pull_diverged") else None,
|
| 34 | detail=f"tree_reader={status}")
|
| 35 | finally:
|
| 36 | trial.cleanup()
|
| 37 | if instrument is not None:
|
| 38 | bin_copy = instrument.oak_bin if fixture.kind == "oak" else instrument.git_bin
|
| 39 | if bin_copy is not None:
|
| 40 | trial = Trial(fixture, subject_bin, work, "calibrate-syscalls")
|
| 41 | try:
|
| 42 | trial.start_server()
|
| 43 | trial.pre()
|
| 44 | log = trial.dir / "boundaries.log"
|
| 45 | res = trial.adapter.run(trial.op_argv(bin_override=str(bin_copy)), trial.op_cwd(),
|
| 46 | instrument.env(0, log, fixture.kind))
|
| 47 | calib.boundaries = count_boundaries(log) if res.returncode == 0 else 0
|
| 48 | calib.boundary_kinds = boundary_kinds(log)
|
| 49 | finally:
|
| 50 | trial.cleanup()
|
| 51 | return calib
|
| 52 |
|
| 53 |
|
| 54 | def evaluate(trial: Trial, calib: Calibration, op_res, fault_fired: bool) -> tuple[dict, dict]:
|
| 55 | fx = trial.fixture
|
| 56 | inv: dict[str, Optional[bool]] = {}
|
| 57 | obs: dict = {"op": op_res.brief(400), "fault_fired": fault_fired}
|
| 58 | inv["op_no_hang"] = not op_res.timed_out
|
| 59 | trial.settle()
|
| 60 | repo = trial.target_repo()
|
| 61 | if fx.op == "clone":
|
| 62 | exists = (repo / (".oak" if trial.kind == "oak" else ".git")).exists()
|
| 63 | obs["clone_dest"] = "absent" if not repo.exists() else ("repo" if exists else "partial_without_store")
|
| 64 | inv["repo_opens"] = None
|
| 65 | inv["worktree_consistent"] = None
|
| 66 | if exists and trial.repo_opens(repo):
|
| 67 | inv["repo_opens"] = True
|
| 68 | ok, detail = worktree_verdict(trial, repo, [calib.post_worktree])
|
| 69 | inv["worktree_consistent"] = ok
|
| 70 | obs["worktree"] = detail
|
| 71 | elif repo.exists():
|
| 72 | # A torn clone is fine only if the subject recognises it: its status must
|
| 73 | # refuse with an actionable error (e.g. "remove this directory and re-clone").
|
| 74 | dirty, st = trial.adapter.status_dirty(repo)
|
| 75 | obs["partial_clone_status"] = st.brief(300)
|
| 76 | inv["partial_clone_detected"] = dirty is None and actionable(st)
|
| 77 | else:
|
| 78 | inv["repo_opens"] = trial.repo_opens(repo)
|
| 79 | ok, lost = trial.acked_reachable(repo, fx.meta.get("pre_local_heads", []))
|
| 80 | inv["acked_local_reachable"] = ok
|
| 81 | if lost:
|
| 82 | obs["lost_local_heads"] = lost
|
| 83 | pre_wt = worktree_manifest(trial.fixture.pristine / "client")
|
| 84 | ok, detail = worktree_verdict(trial, repo, [pre_wt, calib.post_worktree])
|
| 85 | inv["worktree_consistent"] = ok
|
| 86 | obs["worktree"] = detail
|
| 87 | if trial.server is not None or trial.kind == "git":
|
| 88 | intended = calib.post_server_head if fx.op == "push" else None
|
| 89 | ok, detail = trial.server_intact(intended)
|
| 90 | inv["server_intact"] = ok
|
| 91 | obs["server"] = detail
|
| 92 | # next operation (no fault)
|
| 93 | if fx.op == "clone":
|
| 94 | nxt = trial.adapter.run(trial.op_argv(dest="cloned2"), trial.op_cwd())
|
| 95 | next_repo = trial.dir / "cloned2"
|
| 96 | if (trial.dir / "cloned").exists() and inv.get("repo_opens"):
|
| 97 | st_dirty, st_res = trial.adapter.status_dirty(trial.dir / "cloned")
|
| 98 | obs["crashed_clone_status"] = {"dirty": st_dirty, "rc": st_res.returncode}
|
| 99 | else:
|
| 100 | nxt = None
|
| 101 | for argv in trial.adapter.pre_argv(fx.op):
|
| 102 | staged = trial.adapter.run(argv, trial.client)
|
| 103 | if staged.returncode != 0:
|
| 104 | nxt = staged # e.g. a stale index.lock blocks staging: judged like the op itself
|
| 105 | break
|
| 106 | if nxt is None:
|
| 107 | nxt = trial.adapter.run(trial.op_argv(), trial.op_cwd())
|
| 108 | next_repo = repo
|
| 109 | obs["next_op"] = nxt.brief(400)
|
| 110 | if nxt.timed_out:
|
| 111 | inv["next_op_ok_or_actionable"] = False
|
| 112 | elif nxt.returncode == 0:
|
| 113 | inv["next_op_ok_or_actionable"] = True
|
| 114 | trial.settle()
|
| 115 | ok, detail = converge_check(trial, calib, next_repo)
|
| 116 | inv["no_false_success"] = ok
|
| 117 | obs["converge"] = detail
|
| 118 | else:
|
| 119 | inv["next_op_ok_or_actionable"] = actionable(nxt)
|
| 120 | obs["recovery_required"] = True
|
| 121 | if fx.op != "clone":
|
| 122 | ok, lost = trial.acked_reachable(repo, fx.meta.get("pre_local_heads", []))
|
| 123 | inv["acked_after_next"] = ok
|
| 124 | if lost:
|
| 125 | obs["lost_after_next"] = lost
|
| 126 | return inv, obs
|
| 127 |
|
| 128 |
|
| 129 | def run_trial(fixture: Fixture, subject_bin: str, work: Path, name: str, calib: Calibration,
|
| 130 | fault: Callable[[Trial], tuple], keep: Callable[[Trial, str], None]) -> dict:
|
| 131 | """fault(trial) -> (op_result, fault_fired, fault_point). Returns a partial row."""
|
| 132 | trial = Trial(fixture, subject_bin, work, name)
|
| 133 | started = time.perf_counter()
|
| 134 | try:
|
| 135 | trial.start_server()
|
| 136 | trial.pre()
|
| 137 | op_res, fired, point = fault(trial)
|
| 138 | inv, obs = evaluate(trial, calib, op_res, fired)
|
| 139 | classification = classify(inv)
|
| 140 | error = None
|
| 141 | except Exception as exc: # noqa: BLE001 - recorded as harness_error, never hidden
|
| 142 | inv, obs, classification, error, point, op_res = {}, {}, HARNESS_ERROR, f"{type(exc).__name__}: {exc}", "?", None
|
| 143 | finally:
|
| 144 | trial.stop_server()
|
| 145 | row = {
|
| 146 | "fault_point": point,
|
| 147 | "invariants": inv,
|
| 148 | "violations": sorted(k for k, v in inv.items() if v is False),
|
| 149 | "observations": obs,
|
| 150 | "classification": classification,
|
| 151 | "elapsed_ms": round(op_res.elapsed_ms if op_res else (time.perf_counter() - started) * 1000.0, 3),
|
| 152 | "command": op_res.argv if op_res else [],
|
| 153 | }
|
| 154 | if error:
|
| 155 | row["harness_error"] = error
|
| 156 | keep(trial, classification)
|
| 157 | return row
|
| 158 |
|
| 159 |
|
| 160 | def time_fault(delay_s: float, argv_fn=None):
|
| 161 | def fault(trial: Trial):
|
| 162 | res = run_killed_after(trial.op_argv(), trial.op_cwd(), trial.adapter.env(), delay_s)
|
| 163 | return res, res.killed_by_harness and not res.timed_out, f"t={delay_s * 1000:.1f}ms"
|
| 164 | return fault
|
| 165 |
|
| 166 |
|
| 167 | def syscall_fault(instrument: CrashInstrument, n: int, kinds: list[str]):
|
| 168 | def fault(trial: Trial):
|
| 169 | bin_copy = instrument.oak_bin if trial.kind == "oak" else instrument.git_bin
|
| 170 | log = trial.dir / "boundaries.log"
|
| 171 | res = trial.adapter.run(trial.op_argv(bin_override=str(bin_copy)), trial.op_cwd(),
|
| 172 | instrument.env(n, log, trial.kind))
|
| 173 | fired = res.returncode == -9
|
| 174 | kind = kinds[n - 1] if 0 < n <= len(kinds) else "?"
|
| 175 | return res, fired, f"before {kind}#{n}"
|
| 176 | return fault
|
| 177 |
|
| 178 |
|
| 179 | def syscall_points(boundaries: int) -> list[int]:
|
| 180 | if boundaries <= MAX_SYSCALL_POINTS:
|
| 181 | return list(range(1, boundaries + 1))
|
| 182 | step = boundaries / MAX_SYSCALL_POINTS
|
| 183 | points = sorted({max(1, int(round(i * step))) for i in range(1, MAX_SYSCALL_POINTS + 1)})
|
| 184 | return points
|