perf(push): drop per-entry cache fsync; let long runs profile themselves

The cache fsynced every payload and manifest -- two per entry, ~18 ms, and a
push publishes one entry per question (~22,000 on a WordPress plugin). A
read already rejects any entry whose manifest receipt, payload size or
payload SHA-256 does not check, so a torn write is a miss, never a wrong
answer; the fsyncs bought nothing the verification does not.

The broader problem: three quadratic walls in this replay were found one
per two-hour run, because all were Python bookkeeping between engine calls
and the only record was the engine log. `OUSAST_SAMPLE_PROFILE=<path>`
starts an in-process sampler that rewrites the hottest stacks every 30 s,
so a run killed at its deadline still names where its time went. The host
cannot ptrace the container, hence in-process. The replay harness turns it
on per case.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo

benchmarks/push/transfer_replay.py

@@ -263,7 +263,12 @@ def main() -> int:
         overrun = None
         with stdout_path.open("w") as stdout, stderr_path.open("w") as stderr:
             process = subprocess.Popen(  # noqa: S603
-                command, stdout=stdout, stderr=stderr, text=True, env={**os.environ, "OUSAST_LOG_LEVEL": "INFO"}, start_new_session=True
+                command,
+                stdout=stdout,
+                stderr=stderr,
+                text=True,
+                env={**os.environ, "OUSAST_LOG_LEVEL": "INFO", "OUSAST_SAMPLE_PROFILE": str(out / (case["id"] + ".profile.txt"))},
+                start_new_session=True,
             )
             try:
                 returncode: int | None = process.wait(timeout=args.deadline + 60)

src/openultrasast/cli.py

@@ -14,7 +14,7 @@ from pathlib import Path
 from time import perf_counter
 from typing import TYPE_CHECKING, Any
 
-from . import tool_hunter
+from . import profiling, tool_hunter
 from .benchmark import (
     create_benchmark_run,
     evaluate_benchmark,
@@ -243,6 +243,7 @@ def main(argv: list[str] | None = None) -> int:
     # indistinguishable from a run with nothing to report, which is how four PHP investigations were
     # spent reading a later stage's timeout as the cause.
     _configure_logging()
+    profiling.start()  # only when OUSAST_SAMPLE_PROFILE names a file
     if args.command == "pre-push":
         from .config import PushConfig
         from .push.runner import push, replay

src/openultrasast/profiling.py

@@ -0,0 +1,78 @@
+"""A sampling profiler for long runs, switched on by ``OUSAST_SAMPLE_PROFILE=<path>``.
+
+Named by the 2026-09-23 PHP transfer replay, which found three quadratic walls one per two-hour run: a cache
+that rescanned itself on every write, a relationship list searched linearly, and a fsync per cache entry. Each
+was invisible until a run died on it, because the only record was the engine's own log and all three were
+Python-side bookkeeping BETWEEN engine calls. The host cannot attach a profiler to the container's process, so
+the process profiles itself.
+
+A daemon thread samples the main thread's stack every ``interval`` seconds and rewrites ``path`` every
+``flush`` seconds with the functions that were on the stack most often, cumulative and self. It rewrites rather
+than appends, so a process killed at its deadline still leaves the profile of everything before the kill -- the
+case that matters most, and the one an exit hook cannot serve.
+"""
+
+from __future__ import annotations
+
+import collections
+import contextlib
+import os
+import sys
+import threading
+import time
+from pathlib import Path
+from types import FrameType
+
+ENV = "OUSAST_SAMPLE_PROFILE"
+
+
+def _frames(frame: FrameType | None) -> list[str]:
+    names = []
+    while frame is not None:
+        code = frame.f_code
+        names.append(f"{Path(code.co_filename).name}:{code.co_firstlineno}:{code.co_name}")
+        frame = frame.f_back
+    return names
+
+
+def start(path: str | None = None, *, interval: float = 0.01, flush: float = 30.0) -> threading.Thread | None:
+    """Start sampling the calling thread; ``None`` when no path is configured."""
+    target = path or os.environ.get(ENV, "").strip()
+    if not target:
+        return None
+    main = threading.get_ident()
+    cumulative: collections.Counter[str] = collections.Counter()
+    own: collections.Counter[str] = collections.Counter()
+    started = time.monotonic()
+    samples = 0
+
+    def write() -> None:
+        lines = [f"samples {samples} over {time.monotonic() - started:.0f}s at {interval}s", "", "cumulative (on the stack):"]
+        lines += [f"{count / max(samples, 1):7.1%}  {name}" for name, count in cumulative.most_common(40)]
+        lines += ["", "self (executing):"]
+        lines += [f"{count / max(samples, 1):7.1%}  {name}" for name, count in own.most_common(25)]
+        partial = Path(target + ".partial")
+        partial.write_text("\n".join(lines) + "\n")
+        partial.replace(target)
+
+    def run() -> None:
+        nonlocal samples
+        last = time.monotonic()
+        while True:
+            time.sleep(interval)
+            frame = sys._current_frames().get(main)
+            if frame is None:
+                return
+            stack = _frames(frame)
+            samples += 1
+            own[stack[0]] += 1
+            for name in set(stack):
+                cumulative[name] += 1
+            if time.monotonic() - last >= flush:
+                last = time.monotonic()
+                with contextlib.suppress(OSError):
+                    write()
+
+    thread = threading.Thread(target=run, name="ousast-sampler", daemon=True)
+    thread.start()
+    return thread

src/openultrasast/push/cache.py

@@ -244,12 +244,13 @@ class ArtifactCache:
                     elif graph_bytes(payload, budget.deadline_monotonic, target) != sha:
                         raise ValueError("publication_source_changed")
                     target.chmod(0o600)
-                    with target.open("rb") as stream:
+                    # No fsync. Durability is not what makes an entry safe to read: `_entry` rejects anything
-                        os.fsync(stream.fileno())
+                    # whose manifest receipt, payload size or payload SHA-256 does not check, so a write torn
+                    # by a crash is a MISS, never a wrong answer. Two fsyncs per entry were ~18 ms, and a push
+                    # publishes one entry per question -- about 22,000 on a WordPress plugin, minutes of the
+                    # budget spent making a cache survive power loss that it already survives by verification.
                     with (pending / "manifest.json").open("xb") as stream:
                         stream.write(manifest)
-                        stream.flush()
-                        os.fsync(stream.fileno())
                     (pending / "manifest.json").chmod(0o600)
                     if time.monotonic() >= budget.deadline_monotonic:
                         raise TimeoutError("deadline_exhausted")

tests/test_profiling.py

@@ -0,0 +1,28 @@
+"""The in-process sampler leaves a readable profile behind, including for a process killed mid-run."""
+
+import time
+
+from openultrasast import profiling
+
+
+def _busy_quadratic(n: int) -> int:
+    seen: list[int] = []
+    for i in range(n):
+        if i not in seen:
+            seen.append(i)
+    return len(seen)
+
+
+def test_the_sampler_names_the_function_that_holds_the_time(tmp_path, monkeypatch):
+    target = tmp_path / "profile.txt"
+    monkeypatch.delenv(profiling.ENV, raising=False)
+    assert profiling.start() is None  # off unless asked for
+    assert profiling.start(str(target), interval=0.002, flush=0.2) is not None
+    deadline = time.monotonic() + 1.5
+    while time.monotonic() < deadline:
+        _busy_quadratic(3000)
+    time.sleep(0.3)
+    text = target.read_text()
+    assert text.startswith("samples ")
+    self_section = text.split("self (executing):")[1]
+    assert "_busy_quadratic" in self_section.splitlines()[1], self_section[:300]