diff --git a/cmake/ProtocyteFunctions.cmake b/cmake/ProtocyteFunctions.cmake index a8146d7..c5df376 100644 --- a/cmake/ProtocyteFunctions.cmake +++ b/cmake/ProtocyteFunctions.cmake @@ -1528,6 +1528,7 @@ function(_protocyte_finalize_output_plans) RESULT_VARIABLE coordinator_result OUTPUT_VARIABLE coordinator_output ERROR_VARIABLE coordinator_error + ECHO_ERROR_VARIABLE ENCODING UTF-8 ) if(NOT "${coordinator_result}" STREQUAL "0") diff --git a/cmake/ProtocyteOutputCoordinator.py b/cmake/ProtocyteOutputCoordinator.py index e999ef6..25f1660 100644 --- a/cmake/ProtocyteOutputCoordinator.py +++ b/cmake/ProtocyteOutputCoordinator.py @@ -27,6 +27,8 @@ import stat import subprocess import sys +import threading +import time from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Any, Mapping, NoReturn, Sequence @@ -61,12 +63,20 @@ def _canonical_json(value: Mapping[str, Any]) -> bytes: return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode() +def _stat_is_link(observed: os.stat_result) -> bool: + return stat.S_ISLNK(observed.st_mode) or ( + os.name == "nt" + and observed.st_reparse_tag == stat.IO_REPARSE_TAG_MOUNT_POINT + ) + + def _is_link(path: Path) -> bool: try: - if path.is_symlink(): - return True - is_junction = getattr(path, "is_junction", None) - return bool(is_junction and is_junction()) + # One non-following observation covers both Windows link kinds. Calling + # is_symlink() and is_junction() separately probes every ancestor twice. + return _stat_is_link(path.lstat()) + except FileNotFoundError: + return False except OSError: return True @@ -170,46 +180,62 @@ def _path_key(path: Path) -> str: def _contains(parent: Path, child: Path) -> bool: - parent_identity = _portable_identity(parent).rstrip("/") - child_identity = _portable_identity(child, leaf_may_be_file=True).rstrip("/") + return _projected_contains( + project_path(parent), project_path(child, leaf_may_be_file=True) + ) + + +def _projected_contains(parent: Path, child: Path) -> bool: + parent_identity = os.fspath(parent).replace("\\", "/").casefold().rstrip("/") + child_identity = os.fspath(child).replace("\\", "/").casefold().rstrip("/") return child_identity == parent_identity or child_identity.startswith( parent_identity + "/" ) def _physical_location_key(path: Path) -> tuple[int, int, tuple[str, ...]]: - projected = project_path(path) + """Identify a freshly projected directory or one of its ancestors. + + Callers must project both comparison inputs first. Re-projecting here walks + the entire ancestor chain again for every step of a containment comparison. + These observations are deliberately local to one comparison, never cached + across comparisons, lock acquisitions, or filesystem mutations. + """ suffix: list[str] = [] - current = projected - while not os.path.lexists(current): - suffix.append(current.name.casefold()) - parent = current.parent - if parent == current: - _fail(f"could not identify physical filesystem location: {path}") - current = parent - try: - observed = current.stat() - except OSError as error: - _fail(f"could not identify physical filesystem location {path}: {error}") + current = path + while True: + try: + # Projection and comparison are separate observations. A formerly + # missing component may now be a file, link, or junction; reject it + # instead of following it or using its identity as a directory. + observed = current.lstat() + if _stat_is_link(observed): + _fail(f"filesystem path contains a symbolic link or junction: {path}") + if not stat.S_ISDIR(observed.st_mode): + _fail(f"filesystem path has a non-directory ancestor: {current}") + break + except FileNotFoundError: + suffix.append(current.name.casefold()) + parent = current.parent + if parent == current: + _fail(f"could not identify physical filesystem location: {path}") + current = parent + except OSError as error: + _fail(f"could not identify physical filesystem location {path}: {error}") return observed.st_dev, observed.st_ino, tuple(reversed(suffix)) -def _same_physical_location(first: Path, second: Path) -> bool: - if first == second: - return True - try: - if os.path.samefile(first, second): - return True - except OSError: - pass - return _physical_location_key(first) == _physical_location_key(second) +def _physically_contains(parent: Path, child: Path) -> bool: + return _projected_physically_contains(project_path(parent), project_path(child)) -def _physically_contains(parent: Path, child: Path) -> bool: - projected_parent = project_path(parent) - current = project_path(child) +def _projected_physically_contains(parent: Path, child: Path) -> bool: + parent_key = _physical_location_key(parent) + current = child while True: - if _same_physical_location(projected_parent, current): + if _same_physical_path(parent, current) or parent_key == _physical_location_key( + current + ): return True next_parent = current.parent if next_parent == current: @@ -218,11 +244,15 @@ def _physically_contains(parent: Path, child: Path) -> bool: def _paths_overlap(first: Path, second: Path) -> bool: + # Each projection checks the whole existing chain for links/junctions. The + # remaining comparisons share only these two freshly validated inputs. + first = project_path(first) + second = project_path(second) return ( - _contains(first, second) - or _contains(second, first) - or _physically_contains(first, second) - or _physically_contains(second, first) + _projected_contains(first, second) + or _projected_contains(second, first) + or _projected_physically_contains(first, second) + or _projected_physically_contains(second, first) ) @@ -510,6 +540,88 @@ def __exit__(self, *_: object) -> None: self._stream = None +class _ReconcileProgress: + """Bounded, path-free stderr progress, including while OS locks block. + + Fast operations stay silent. Long operations emit at most twelve progress + lines and one completion/failure summary. Nothing is written to the claim + token stream on stdout. The reporter observes counters only, not the FS. + """ + + INTERVAL_SECONDS = 5.0 + MAX_REPORTS = 12 + + def __init__(self, current: int, retired: int) -> None: + self.current = current + self.retired = retired + self._claims = 0 + self._states: int | None = None + self._started = time.monotonic() + self._since = self._started + self._phase = "preparation" + self._category = "reconciliation" + self._seconds = {"lock_wait": 0.0, "validation": 0.0, "reconciliation": 0.0} + self._reports = 0 + self._mutex = threading.Lock() + self._stop = threading.Event() + self._thread = threading.Thread(target=self._run, daemon=True) + + def __enter__(self) -> _ReconcileProgress: + self._thread.start() + return self + + def __exit__(self, kind: object, *_: object) -> None: + self._stop.set() + self._thread.join() + with self._mutex: + if self._reports: + self._emit("failed" if kind is not None else "complete") + + def phase(self, phase: str, category: str) -> None: + with self._mutex: + now = time.monotonic() + self._seconds[self._category] += now - self._since + self._phase, self._category, self._since = phase, category, now + + def registry(self, *, states: int | None = None, claim: bool = False) -> None: + with self._mutex: + if states is not None: + self._states = states + if claim: + self._claims += 1 + + def _emit(self, phase: str) -> None: + now = time.monotonic() + seconds = dict(self._seconds) + seconds[self._category] += now - self._since + states = "unknown" if self._states is None else str(self._states) + timings = " ".join(f"{name}={value:.3f}s" for name, value in seconds.items()) + print( + f"Protocyte output coordinator: {phase}; " + f"current_plans={self.current} retired_plans={self.retired} " + f"claims_checked={self._claims} registry_entries={states} " + f"elapsed={now - self._started:.3f}s {timings}", + file=sys.stderr, + flush=True, + ) + + def _report(self) -> bool: + with self._mutex: + if self._reports >= self.MAX_REPORTS: + return False + self._reports += 1 + phase = self._phase + if self._reports == self.MAX_REPORTS: + phase += " (further progress suppressed until completion)" + self._emit(phase) + return True + + def _run(self) -> None: + while not self._stop.wait(self.INTERVAL_SECONDS): + if not self._report(): + return + + @dataclass(frozen=True) class PlanOutput: target: str @@ -770,10 +882,11 @@ def __init__(self, lock_root: Path) -> None: self.lock_root = project_path(_require_absolute(lock_root, "output lock root")) def reconcile(self, plan: Plan) -> str: - state = self._claim(plan) - with FileLock(self._generation_lock(plan.root)): - with FileLock(self._publication_lock(plan.root)): - return self._reconcile_locked(state, plan) + # All claim/plan writers hold the registry lock through plan commit, so + # another reconciliation can safely read each retained record once. + token = self.reconcile_set((), (plan,))[0] + assert token is not None + return token def reconcile_set( self, retired: Sequence[Plan], current: Sequence[Plan] @@ -784,15 +897,23 @@ def reconcile_set( roots = {_path_key(plan.root): plan.root for plan in plans} retiring_root_keys = {_path_key(plan.root) for plan in retired} _durable_mkdir(self.lock_root) - with contextlib.ExitStack() as locks: + with ( + _ReconcileProgress(len(current), len(retired)) as progress, + contextlib.ExitStack() as locks, + ): + progress.phase("waiting for registry lock", "lock_wait") locks.enter_context(FileLock(self.lock_root / "registry.lock")) + progress.phase("waiting for generation locks", "lock_wait") for root in sorted(roots.values(), key=_portable_identity): locks.enter_context(FileLock(self._generation_lock(root))) + progress.phase("waiting for publication locks", "lock_wait") for root in sorted(roots.values(), key=_portable_identity): locks.enter_context(FileLock(self._publication_lock(root))) + progress.phase("validating current plans", "validation") self._validate_plan_set(current) - for plan in current: - self._validate_registry(plan, retiring_root_keys) + progress.phase("validating registry", "validation") + self._validate_registry(current, retiring_root_keys, progress=progress) + progress.phase("reconciling outputs", "reconciliation") states: dict[str, Path] = {} existing_plans: list[Plan] = [] active_retired: set[int] = set() @@ -1119,20 +1240,6 @@ def plan_for_root(self, root: Path) -> Plan: return plan return Plan(claim_root, build_root, str(claim["build_id"]), (), ()) - def _claim(self, plan: Plan) -> Path: - _durable_mkdir(self.lock_root) - state = self._state_directory(plan.root) - with FileLock(self.lock_root / "registry.lock"): - self._validate_registry(plan) - _durable_mkdir(state, anchor=self.lock_root) - claim_path = state / "claim.json" - if claim_path.exists(): - self._load_claim(state, plan) - _sync_directory(state) - return state - self._initialize_claim_locked(state, plan) - return state - def _initialize_claim_locked(self, state: Path, plan: Plan) -> None: _durable_mkdir(state, anchor=self.lock_root) token = secrets.token_hex(32) @@ -1177,32 +1284,41 @@ def _validate_plan_set(plans: Sequence[Plan]) -> None: _fail("current output plan staging overlaps another output root") def _validate_registry( - self, plan: Plan, retiring_root_keys: set[str] | None = None + self, + plans: Sequence[Plan], + retiring_root_keys: set[str] | None = None, + *, + progress: _ReconcileProgress | None = None, ) -> None: + if not plans: + return if retiring_root_keys is None: retiring_root_keys = set() - for target in plan.targets: - if _contains(target.staging, self.lock_root) or _physically_contains( - target.staging, self.lock_root - ): - _fail( - "generation staging contains the output coordinator lock root: " - f"{target.staging} and {self.lock_root}" - ) internal_paths = ( self.lock_root / "roots", self.lock_root / "generation", self.lock_root / "publication", ) - if _paths_overlap(plan.root, self.lock_root) or any( - _paths_overlap(path, plan.root) for path in internal_paths - ): - _fail( - "output root overlaps the output coordinator lock namespace: " - f"{plan.root} and {self.lock_root}" - ) + for plan in plans: + for target in plan.targets: + if _contains(target.staging, self.lock_root) or _physically_contains( + target.staging, self.lock_root + ): + _fail( + "generation staging contains the output coordinator lock root: " + f"{target.staging} and {self.lock_root}" + ) + if _paths_overlap(plan.root, self.lock_root) or any( + _paths_overlap(path, plan.root) for path in internal_paths + ): + _fail( + "output root overlaps the output coordinator lock namespace: " + f"{plan.root} and {self.lock_root}" + ) roots = self.lock_root / "roots" if not os.path.lexists(roots): + if progress is not None: + progress.registry(states=0) return if project_path(roots) != roots: _fail(f"output registry is unsafe or was replaced: {roots}") @@ -1211,6 +1327,8 @@ def _validate_registry( states = sorted(stream, key=lambda entry: entry.name) except OSError as error: _fail(f"could not enumerate output registry {roots}: {error}") + if progress is not None: + progress.registry(states=len(states)) for entry in states: state = roots / entry.name try: @@ -1232,44 +1350,49 @@ def _validate_registry( self._validate_claim_shape(claim, claim_path) claimed_root, _ = self._claim_paths(claim, claim_path) if entry.name != claim["root_key"]: - _fail(f"output claim is stored under the wrong registry key: {claim_path}") + _fail( + f"output claim is stored under the wrong registry key: {claim_path}" + ) recorded_plan = self._load_recorded_plan(state, claim) - if _paths_overlap(claimed_root, plan.root): - if _path_key(claimed_root) != _path_key(plan.root): + if progress is not None: + progress.registry(claim=True) + recorded_targets = recorded_plan.targets if recorded_plan else () + for plan in plans: + same_root_key = claim["root_key"] == _path_key(plan.root) + if _paths_overlap(claimed_root, plan.root) and not same_root_key: _fail( "output root overlaps a root claimed by another build: " f"{plan.root} and {claimed_root}" ) - recorded_targets = recorded_plan.targets if recorded_plan else () - for recorded_target in recorded_targets: - if _paths_overlap(recorded_target.staging, plan.root): - _fail( - "output root overlaps staging reserved by another plan: " - f"{plan.root} and {recorded_target.staging}" - ) - for target in plan.targets: - if _paths_overlap(claimed_root, target.staging): - _fail( - "generation staging overlaps a claimed output root: " - f"{target.staging} and {claimed_root}" - ) for recorded_target in recorded_targets: - if ( - _path_key(plan.root) == _path_key(claimed_root) - and target.identity == recorded_target.identity - ): - continue - if ( - str(claim["root_key"]) in retiring_root_keys - and str(claim["build_id"]) == plan.build_id - and target.identity == recorded_target.identity - ): - continue - if _paths_overlap(recorded_target.staging, target.staging): + if _paths_overlap(recorded_target.staging, plan.root): + _fail( + "output root overlaps staging reserved by another plan: " + f"{plan.root} and {recorded_target.staging}" + ) + for target in plan.targets: + if _paths_overlap(claimed_root, target.staging): _fail( - "generation staging overlaps staging reserved by another plan: " - f"{target.staging} and {recorded_target.staging}" + "generation staging overlaps a claimed output root: " + f"{target.staging} and {claimed_root}" ) + for recorded_target in recorded_targets: + if ( + same_root_key + and target.identity == recorded_target.identity + ): + continue + if ( + str(claim["root_key"]) in retiring_root_keys + and str(claim["build_id"]) == plan.build_id + and target.identity == recorded_target.identity + ): + continue + if _paths_overlap(recorded_target.staging, target.staging): + _fail( + "generation staging overlaps staging reserved by another plan: " + f"{target.staging} and {recorded_target.staging}" + ) def _state_directory(self, root: Path) -> Path: return self.lock_root / "roots" / _path_key(root) diff --git a/docs/wiki/Contributing.md b/docs/wiki/Contributing.md index d5438ed..9e4ce89 100644 --- a/docs/wiki/Contributing.md +++ b/docs/wiki/Contributing.md @@ -54,6 +54,27 @@ for the focused build matrix. Format every touched C++ source or header with `clang-format`. +### Output registry performance + +Run the offline scaling benchmark with Python 3.12 or newer: + +```bash +python tests/benchmark_output_registry.py --claims 0 8 16 --plans 1 4 --repeat 3 +``` + +It creates each retained claim through the real coordinator in a fresh private +registry and reports seed, registry validation, and reconciliation wall/CPU times +separately. It also counts registry scans, recorded-plan loads, and path +projections. It does not generate C++, use the shared registry, or access the +network. The printed temporary directory retains the synthetic state and +`results.json`; use `--work NEW_DIRECTORY` to choose that directory explicitly. + +Use `--baseline-source /path/to/baseline-checkout` for an alternating comparison +against another revision at the same workspace depth. The report includes every +sample and median times. Compare operation counts as well as wall times: Windows +filesystem probes, path depth, caching, and concurrent machine activity affect +timings. The focused regression tests are in `tests/test_output_coordinator.py`. + ## Documentation The canonical wiki source is `docs/wiki/` in the main repository. Edit and diff --git a/docs/wiki/Troubleshooting.md b/docs/wiki/Troubleshooting.md index eeb6c0f..3272f1e 100644 --- a/docs/wiki/Troubleshooting.md +++ b/docs/wiki/Troubleshooting.md @@ -188,6 +188,16 @@ See [CMake API Reference](https://github.com/anthonyprintup/protocyte/wiki/CMake Protocyte generates into private staging and validates outputs before publication. A compiler failure, timeout, or invalid result publishes no output transaction. If publication is interrupted after the durable write-ahead transaction is created, the next configure or build rolls it forward from its durable payloads. +When output reconciliation takes longer than five seconds, CMake shows coordinator +progress on standard error. It distinguishes waiting for registry, generation, or +publication locks from current-plan validation, registry validation, and output +reconciliation. Lines include plan/claim counts and elapsed phase times, without +consumer paths or claim tokens. Fast operations remain silent; long operations +emit at most twelve progress lines and one completion or failure summary. A lock +wait includes preparing and acquiring its lock files. `registry_entries` counts +state directories, including any abandoned initialization directories encountered; +`claims_checked` counts authenticated claim/plan records. + If cleanup refuses an unsafe path or cannot remove it, the diagnostic warns that inert staging data outside `OUT_DIR` may remain. Stop related processes and remove only the named staging path after inspecting it. ## Adding a Proto File Does Not Generate It diff --git a/tests/benchmark_output_registry.py b/tests/benchmark_output_registry.py new file mode 100644 index 0000000..260240f --- /dev/null +++ b/tests/benchmark_output_registry.py @@ -0,0 +1,250 @@ +"""Offline registry scaling benchmark; Python 3.12+, standard library only. + +Each sample seeds real claims into a fresh, explicitly isolated registry. No +generated payloads, shared registry, compiler, network, or source edits are used. +Seed/validation/total reconciliation wall and CPU times are reported separately. +Counters delegate every operation unchanged. Timings include instrumentation. + +Example (run from the repository root): + python tests/benchmark_output_registry.py --claims 0 8 16 --plans 1 4 --repeat 3 + +Pass --baseline-source PATH to compare another checkout (or a directory holding +only cmake/ProtocyteOutputCoordinator.py). Baseline/candidate samples alternate +order, use the same path depth, and never run concurrently. Results and synthetic +state are retained under a fresh temporary directory, or --work NEW_DIRECTORY. +Wall times are descriptive; operation counts are the portable regression gate. +""" + +from __future__ import annotations + +import argparse +import hashlib +import importlib.util +import json +import os +import platform +import statistics +import sys +import tempfile +import time +from pathlib import Path +from types import ModuleType +from typing import Any + + +def _load(source: Path, label: str) -> ModuleType: + script = source / "cmake" / "ProtocyteOutputCoordinator.py" + spec = importlib.util.spec_from_file_location(f"registry_benchmark_{label}", script) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def _plan(module: ModuleType, work: Path, name: str) -> Any: + directory = work / name + build = directory / "build" + build.mkdir(parents=True) + target = hashlib.sha256(name.encode()).hexdigest() + staging = directory / f".protocyte-generation-staging-{target}" + path = directory / "outputs.plan" + + def encode(value: Path | str) -> str: + return os.fspath(value).encode("utf-8").hex() + + path.write_text( + "\n".join( + [ + module.PLAN_HEADER, + "root-hex=" + encode(directory / "generated"), + "build-root-hex=" + encode(build), + "target=" + target + "|" + encode(staging), + "output=" + target + "|" + encode("sample.protocyte.hpp"), + ] + ) + + "\n", + encoding="ascii", + ) + return module.Plan.read(path) + + +def _sample(module: ModuleType, work: Path, claims: int, plans: int) -> dict[str, Any]: + engine = module.OutputCoordinator(work / "isolated-registry") + retained = [_plan(module, work, f"retained-{i:04}") for i in range(claims)] + current = [_plan(module, work, f"current-{i:04}") for i in range(plans)] + wall, cpu = time.perf_counter(), time.process_time() + if retained: + engine.reconcile_set((), retained) + seed_wall = time.perf_counter() - wall + seed_cpu = time.process_time() - cpu + counts = { + "registry_validations": 0, + "registry_scans": 0, + "recorded_plan_loads_in_validation": 0, + "path_projections_in_validation": 0, + "path_projections_total": 0, + } + active = False + validation_wall = validation_cpu = 0.0 + original_validate = engine._validate_registry + original_load = engine._load_recorded_plan + original_project = module.project_path + original_scandir = module.os.scandir + + def validate(*args: Any, **kwargs: Any) -> Any: + nonlocal active, validation_wall, validation_cpu + counts["registry_validations"] += 1 + active = True + start_wall, start_cpu = time.perf_counter(), time.process_time() + try: + return original_validate(*args, **kwargs) + finally: + validation_wall += time.perf_counter() - start_wall + validation_cpu += time.process_time() - start_cpu + active = False + + def load(*args: Any, **kwargs: Any) -> Any: + if active: + counts["recorded_plan_loads_in_validation"] += 1 + return original_load(*args, **kwargs) + + def project(*args: Any, **kwargs: Any) -> Any: + counts["path_projections_total"] += 1 + if active: + counts["path_projections_in_validation"] += 1 + return original_project(*args, **kwargs) + + def scandir(path: Any) -> Any: + if active and path == engine.lock_root / "roots": + counts["registry_scans"] += 1 + return original_scandir(path) + + engine._validate_registry = validate + engine._load_recorded_plan = load + module.project_path = project + module.os.scandir = scandir + wall, cpu = time.perf_counter(), time.process_time() + try: + tokens = engine.reconcile_set((), current) + reconcile_wall = time.perf_counter() - wall + reconcile_cpu = time.process_time() - cpu + finally: + module.project_path = original_project + module.os.scandir = original_scandir + assert len(tokens) == plans and all(tokens) + return { + "retained_claims": claims, + "current_plans": plans, + "seed_wall_seconds": seed_wall, + "seed_cpu_seconds": seed_cpu, + "validation_wall_seconds": validation_wall, + "validation_cpu_seconds": validation_cpu, + "reconcile_wall_seconds": reconcile_wall, + "reconcile_cpu_seconds": reconcile_cpu, + **counts, + "completed_plans": len(tokens), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--source", type=Path, default=Path(__file__).resolve().parents[1] + ) + parser.add_argument("--baseline-source", type=Path) + parser.add_argument("--claims", type=int, nargs="+", default=[0, 8, 16]) + parser.add_argument("--plans", type=int, nargs="+", default=[1, 4]) + parser.add_argument("--repeat", type=int, default=3) + parser.add_argument("--work", type=Path) + args = parser.parse_args() + if ( + any(not 0 <= value <= 500 for value in args.claims) + or any(not 1 <= value <= 16 for value in args.plans) + or not 1 <= args.repeat <= 20 + ): + parser.error("expected 0..500 claims, 1..16 plans, and 1..20 repetitions") + sources = {"candidate": args.source.resolve()} + if args.baseline_source: + sources = {"baseline": args.baseline_source.resolve(), **sources} + for source in sources.values(): + if not (source / "cmake" / "ProtocyteOutputCoordinator.py").is_file(): + parser.error("source must contain cmake/ProtocyteOutputCoordinator.py") + if args.work: + work = args.work.absolute() + work.mkdir(parents=True, exist_ok=False) + else: + work = Path(tempfile.mkdtemp(prefix="protocyte-registry-benchmark-")) + modules = {label: _load(source, label) for label, source in sources.items()} + result: dict[str, Any] = { + "platform": platform.platform(), + "python": platform.python_version(), + "coordinator_sha256": { + label: hashlib.sha256(Path(module.__file__).read_bytes()).hexdigest() + for label, module in modules.items() + }, + "samples": [], + "medians": [], + } + print(f"Synthetic work and results: {work}", flush=True) + labels = list(modules) + for repetition in range(args.repeat): + for claims in args.claims: + for plans in args.plans: + for label in labels if repetition % 2 == 0 else reversed(labels): + sample_work = work / f"r{repetition}-c{claims}-p{plans}" / label + print( + json.dumps( + { + "phase": "seed", + "revision": label, + "claims": claims, + "plans": plans, + "repetition": repetition, + } + ), + flush=True, + ) + sample = { + "revision": label, + "repetition": repetition, + **_sample(modules[label], sample_work, claims, plans), + } + result["samples"].append(sample) + print(json.dumps(sample), flush=True) + (work / "results.json").write_text( + json.dumps(result, indent=2) + "\n", encoding="utf-8" + ) + for claims in args.claims: + for plans in args.plans: + medians = {} + for label in labels: + samples = [ + sample + for sample in result["samples"] + if sample["revision"] == label + and sample["retained_claims"] == claims + and sample["current_plans"] == plans + ] + medians[label] = { + key: statistics.median(sample[key] for sample in samples) + for key in samples[0] + if key.endswith("_seconds") + or key.startswith(("registry_", "path_", "recorded_")) + } + summary = {"retained_claims": claims, "current_plans": plans, **medians} + if "baseline" in medians: + summary["wall_time_reduction_percent"] = 100 * ( + 1 + - medians["candidate"]["reconcile_wall_seconds"] + / medians["baseline"]["reconcile_wall_seconds"] + ) + result["medians"].append(summary) + print(json.dumps(summary), flush=True) + (work / "results.json").write_text( + json.dumps(result, indent=2) + "\n", encoding="utf-8" + ) + + +if __name__ == "__main__": + main() diff --git a/tests/test_output_coordinator.py b/tests/test_output_coordinator.py index b7e66e8..9fe66a5 100644 --- a/tests/test_output_coordinator.py +++ b/tests/test_output_coordinator.py @@ -1522,8 +1522,17 @@ def test_tampered_durable_payload_fails_closed_without_publication( assert not (root / "demo.protocyte.hpp").exists() +@pytest.mark.parametrize( + "operations", + [ + ("reconcile", "reconcile"), + ("reconcile", "reconcile-set"), + ("reconcile-set", "reconcile-set"), + ], +) def test_two_builds_racing_for_one_root_have_exactly_one_winner( tmp_path: Path, + operations: tuple[str, str], ) -> None: root = tmp_path / "generated" lock_root = tmp_path / "locks-v1" @@ -1542,13 +1551,13 @@ def test_two_builds_racing_for_one_root_have_exactly_one_winner( [ sys.executable, str(coordinator.__file__), - "reconcile", + operation, "--lock-root", str(lock_root), - "--plan", + "--plan" if operation == "reconcile" else "--current-plan", str(plan), ] - for plan in plans + for operation, plan in zip(operations, plans) ] processes = [ subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) @@ -1565,6 +1574,298 @@ def test_two_builds_racing_for_one_root_have_exactly_one_winner( assert len(claims) == 1 +@pytest.mark.parametrize("current_count", [1, 4]) +def test_registry_records_are_loaded_once_per_batch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, current_count: int +) -> None: + def plan(name: str) -> object: + return _write_plan( + tmp_path / f"{name}.plan", + tmp_path / name / "generated", + tmp_path / name / "build", + ((_target(name), "sample.protocyte.hpp"),), + ) + + retained = [plan(f"retained-{index}") for index in range(4)] + current = [plan(f"current-{index}") for index in range(current_count)] + engine = coordinator.OutputCoordinator(tmp_path / "locks") + engine.reconcile_set((), retained) + expected = {engine._state_directory(value.root) for value in retained} + loaded: list[Path] = [] + scans: list[Path] = [] + original_load = engine._load_recorded_plan + original_scandir = coordinator.os.scandir + + def load(state: Path, claim: object) -> object: + if state in expected: + loaded.append(state) + return original_load(state, claim) + + def scandir(path: Path) -> object: + if Path(path) == engine.lock_root / "roots": + scans.append(Path(path)) + return original_scandir(path) + + monkeypatch.setattr(engine, "_load_recorded_plan", load) + monkeypatch.setattr(coordinator.os, "scandir", scandir) + assert len(engine.reconcile_set((), current)) == current_count + assert sorted(loaded) == sorted(expected) + assert scans == [engine.lock_root / "roots"] + + +def test_single_reconciliation_holds_registry_lock_through_plan_commit( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + plan = _write_plan( + tmp_path / "plan", + tmp_path / "generated", + tmp_path / "build", + ((_target("demo"), "demo.protocyte.hpp"),), + ) + engine = coordinator.OutputCoordinator(tmp_path / "locks") + original_lock = coordinator.FileLock + original_write = coordinator._atomic_write + held: list[Path] = [] + committed = False + + class RecordingLock(original_lock): + def __enter__(self) -> object: + result = super().__enter__() + held.append(self.path) + return result + + def __exit__(self, *args: object) -> None: + assert held.pop() == self.path + super().__exit__(*args) + + def write(path: Path, content: bytes) -> None: + nonlocal committed + if path.name == "plan.json": + assert held == [ + engine.lock_root / "registry.lock", + engine._generation_lock(plan.root), + engine._publication_lock(plan.root), + ] + committed = True + original_write(path, content) + + monkeypatch.setattr(coordinator, "FileLock", RecordingLock) + monkeypatch.setattr(coordinator, "_atomic_write", write) + engine.reconcile(plan) + assert committed and not held + + +@pytest.mark.parametrize("depth", [0, 8]) +def test_overlap_projects_inputs_once_regardless_of_ancestor_depth( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, depth: int +) -> None: + base = tmp_path.joinpath(*(f"level-{index}" for index in range(depth))) + base.mkdir(parents=True, exist_ok=True) + first, second = base / "first" / "missing", base / "second" / "missing" + projected: list[Path] = [] + original_project = coordinator.project_path + + def project(path: Path, **options: object) -> Path: + projected.append(path) + return original_project(path, **options) + + monkeypatch.setattr(coordinator, "project_path", project) + assert not coordinator._paths_overlap(first, second) + assert projected == [first, second] + + +def test_physical_containment_checks_ancestors_of_missing_paths( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + first, alias = tmp_path / "first", tmp_path / "alias" + first.mkdir() + alias.mkdir() + original_samefile = coordinator.os.path.samefile + + def samefile(left: object, right: object) -> bool: + if {Path(left), Path(right)} == {first, alias}: + return True + return original_samefile(left, right) + + monkeypatch.setattr(coordinator.os.path, "samefile", samefile) + assert coordinator._paths_overlap(first, alias / "missing" / "nested") + assert coordinator._paths_overlap(alias / "missing" / "nested", first) + assert not coordinator._paths_overlap(first, tmp_path / "unrelated") + + +@pytest.mark.parametrize("replacement", ["file", "directory-link"]) +@pytest.mark.parametrize("side", [0, 1]) +def test_overlap_rejects_path_replacement_during_the_comparison( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, replacement: str, side: int +) -> None: + paths = (tmp_path / "first", tmp_path / "second") + outside = tmp_path / "outside" + outside.mkdir() + original_project = coordinator.project_path + replaced = False + + def project(path: Path, **options: object) -> Path: + nonlocal replaced + projected = original_project(path, **options) + if path == paths[side] and not replaced: + replaced = True + if replacement == "file": + path.touch() + else: + _directory_link(path, outside) + return projected + + monkeypatch.setattr(coordinator, "project_path", project) + with pytest.raises( + coordinator.CoordinatorError, match="symbolic link or junction|non-directory" + ): + coordinator._paths_overlap(*paths) + assert replaced + + +def test_missing_suffixes_use_the_physical_identity_of_the_existing_ancestor( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + first, alias = tmp_path / "first", tmp_path / "alias" + first.mkdir() + alias.mkdir() + original_stat = Path.stat + original_lstat = Path.lstat + + def stat_with_alias(path: Path, **options: object) -> os.stat_result: + # Model two non-link directory spellings backed by one filesystem ID. + return original_stat(first if path == alias else path, **options) + + def lstat_with_alias(path: Path) -> os.stat_result: + return original_lstat(first if path == alias else path) + + monkeypatch.setattr(Path, "stat", stat_with_alias) + monkeypatch.setattr(Path, "lstat", lstat_with_alias) + assert coordinator._paths_overlap(first / "missing", alias / "missing" / "child") + assert not coordinator._paths_overlap(first / "missing", alias / "other") + + +def _directory_link(link: Path, destination: Path) -> None: + try: + link.symlink_to(destination, target_is_directory=True) + except OSError as error: + if os.name != "nt": + pytest.skip(f"directory links are unavailable: {error}") + result = subprocess.run( + ["cmd.exe", "/d", "/c", "mklink", "/J", str(link), str(destination)], + capture_output=True, + text=True, + timeout=10, + ) + if result.returncode: + pytest.skip(f"directory links are unavailable: {result.stderr}") + + +def test_registry_comparisons_reject_a_replaced_retained_root( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + retained_root = tmp_path / "retained" / "generated" + retained = _write_plan( + tmp_path / "retained.plan", + retained_root, + tmp_path / "retained-build", + ((_target("retained"), "sample.protocyte.hpp"),), + ) + current = [ + _write_plan( + tmp_path / f"current-{index}.plan", + tmp_path / f"current-{index}" / "generated", + tmp_path / f"current-{index}" / "build", + ((_target(f"current-{index}"), "sample.protocyte.hpp"),), + ) + for index in range(2) + ] + engine = coordinator.OutputCoordinator(tmp_path / "locks") + engine.reconcile(retained) + original_overlap = coordinator._paths_overlap + changed = False + + def overlap(first: Path, second: Path) -> bool: + nonlocal changed + result = original_overlap(first, second) + if not changed and first == retained_root and second == current[0].root: + changed = True + # The previously missing directory is replaced after one comparison. + _directory_link(retained_root, tmp_path) + return result + + monkeypatch.setattr(coordinator, "_paths_overlap", overlap) + with pytest.raises(coordinator.CoordinatorError, match="symbolic link or junction"): + engine.reconcile_set((), current) + assert changed + assert len(list((engine.lock_root / "roots").glob("*/claim.json"))) == 1 + + +def test_reconciliation_revalidates_paths_after_registry_validation( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "generated" + plan = _write_plan( + tmp_path / "plan", + root, + tmp_path / "build", + ((_target("demo"), "demo.protocyte.hpp"),), + ) + engine = coordinator.OutputCoordinator(tmp_path / "locks") + original_validate = engine._validate_registry + + def validate(*args: object, **kwargs: object) -> None: + original_validate(*args, **kwargs) + _directory_link(root, tmp_path) + + monkeypatch.setattr(engine, "_validate_registry", validate) + with pytest.raises(coordinator.CoordinatorError, match="symbolic link or junction"): + engine.reconcile(plan) + assert not list((engine.lock_root / "roots").glob("*/claim.json")) + + +def test_progress_is_bounded_path_free_and_keeps_stdout_clean( + capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(coordinator._ReconcileProgress, "MAX_REPORTS", 3) + with coordinator._ReconcileProgress(4, 1) as progress: + progress.phase("waiting for registry lock", "lock_wait") + assert progress._report() + progress.phase("validating registry", "validation") + progress.registry(states=8, claim=True) + assert progress._report() + progress.phase("reconciling outputs", "reconciliation") + assert progress._report() + for _ in range(10): + assert not progress._report() + output = capsys.readouterr() + assert not output.out + assert len(output.err.splitlines()) == 4 + assert "waiting for registry lock" in output.err + assert "validating registry" in output.err + assert "current_plans=4 retired_plans=1" in output.err + assert "claims_checked=1 registry_entries=8" in output.err + assert "lock_wait=" in output.err and "validation=" in output.err + assert "further progress suppressed" in output.err + assert "complete;" in output.err + + +def test_progress_reports_failure_and_fast_operations_stay_silent( + capsys: pytest.CaptureFixture[str], +) -> None: + with coordinator._ReconcileProgress(1, 0): + pass + assert capsys.readouterr() == ("", "") + with pytest.raises(coordinator.CoordinatorError): + with coordinator._ReconcileProgress(1, 0) as progress: + progress._report() + coordinator._fail("injected failure") + output = capsys.readouterr() + assert not output.out + assert len(output.err.splitlines()) == 2 + assert "failed;" in output.err + + @pytest.mark.skipif(os.name == "nt", reason="requires case-sensitive paths") def test_case_distinct_build_roots_cannot_share_a_claim(tmp_path: Path) -> None: root = tmp_path / "generated" diff --git a/tests/test_output_coordinator_cmake.py b/tests/test_output_coordinator_cmake.py index 12a282f..c6447e6 100644 --- a/tests/test_output_coordinator_cmake.py +++ b/tests/test_output_coordinator_cmake.py @@ -6,6 +6,7 @@ import shutil import subprocess import sys +import time from pathlib import Path import pytest @@ -136,6 +137,73 @@ def test_coordinator_rejects_a_second_build_before_generation(tmp_path: Path) -> assert "different CMake build tree" in diagnostic +def test_configure_surfaces_lock_progress_without_corrupting_claim_tokens( + tmp_path: Path, +) -> None: + source, build, locks = (tmp_path / name for name in ("source", "build", "locks")) + _write_out_dir_owner_project(source, tmp_path / "generated", output_lock_root=locks) + coordinator_script = ( + Path(__file__).resolve().parents[1] / "cmake" / "ProtocyteOutputCoordinator.py" + ) + holder_script = tmp_path / "hold_registry.py" + holder_script.write_text( + "import runpy, sys\n" + "from pathlib import Path\n" + "module = runpy.run_path(sys.argv[1])\n" + "with module['FileLock'](Path(sys.argv[2]) / 'registry.lock'):\n" + " print('locked', flush=True)\n" + " sys.stdin.readline()\n", + encoding="utf-8", + ) + holder = subprocess.Popen( + [sys.executable, str(holder_script), str(coordinator_script), str(locks)], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + configure = None + try: + assert holder.stdout.readline().strip() == "locked" + error_path = tmp_path / "configure.stderr" + # communicate(timeout=...) does not expose partial pipe output on + # Windows. Read a live file to verify progress BEFORE releasing the lock. + with error_path.open("w", encoding="utf-8") as error_stream: + configure = subprocess.Popen( + ["cmake", "-S", str(source), "-B", str(build), "-G", "Ninja"], + stdout=subprocess.PIPE, + stderr=error_stream, + text=True, + ) + deadline = time.monotonic() + 20 + while True: + diagnostic = error_path.read_text(encoding="utf-8") + if "waiting for registry lock" in diagnostic: + break + assert configure.poll() is None, diagnostic + assert time.monotonic() < deadline, diagnostic + time.sleep(0.05) + progress_lines = [ + line for line in diagnostic.splitlines() + if "Protocyte output coordinator:" in line + ] + assert progress_lines and all( + "current_plans=1" in line for line in progress_lines + ) + assert all(str(tmp_path) not in line for line in progress_lines) + holder.communicate(input="\n", timeout=10) + output, _ = configure.communicate(timeout=30) + error = error_path.read_text(encoding="utf-8") + assert configure.returncode == 0, output + error + assert "complete;" in error + assert len(list((locks / "roots").glob("*/claim.json"))) == 1 + finally: + for process in (configure, holder): + if process is not None and process.poll() is None: + process.kill() + process.communicate(timeout=10) + + def test_generation_inventory_is_literal_under_glob_metacharacter_build_path( tmp_path: Path, ) -> None: