From eda84ccb581bc9084439d39fd6b774f4b6c3c0fb Mon Sep 17 00:00:00 2001 From: Bo Chen Date: Sun, 12 Jul 2026 09:44:46 +0800 Subject: [PATCH 1/2] Add RocksDB native compaction policy task --- 2.0/README.md | 8 + .../config.yaml | 49 + .../docker/build_images.sh | 27 + .../docker/judge/Dockerfile | 82 + .../evaluate.sh | 19 + .../evaluator.py | 934 +++++++++ .../harbor/app/make_submission.sh | 42 + .../harbor/app/validate_solution_patch.py | 229 +++ .../judge/harness/lsm_policy_harness.cc | 1706 +++++++++++++++++ .../rocksdb_native_compaction_policy/readme | 57 + .../reference.cpp | 62 + .../reference.patch | 62 + .../test_evaluator.py | 421 ++++ 13 files changed, 3698 insertions(+) create mode 100644 2.0/problems/rocksdb_native_compaction_policy/config.yaml create mode 100644 2.0/problems/rocksdb_native_compaction_policy/docker/build_images.sh create mode 100644 2.0/problems/rocksdb_native_compaction_policy/docker/judge/Dockerfile create mode 100644 2.0/problems/rocksdb_native_compaction_policy/evaluate.sh create mode 100644 2.0/problems/rocksdb_native_compaction_policy/evaluator.py create mode 100644 2.0/problems/rocksdb_native_compaction_policy/harbor/app/make_submission.sh create mode 100644 2.0/problems/rocksdb_native_compaction_policy/harbor/app/validate_solution_patch.py create mode 100644 2.0/problems/rocksdb_native_compaction_policy/judge/harness/lsm_policy_harness.cc create mode 100644 2.0/problems/rocksdb_native_compaction_policy/readme create mode 100644 2.0/problems/rocksdb_native_compaction_policy/reference.cpp create mode 100644 2.0/problems/rocksdb_native_compaction_policy/reference.patch create mode 100644 2.0/problems/rocksdb_native_compaction_policy/test_evaluator.py diff --git a/2.0/README.md b/2.0/README.md index 628e3f314..289ca6a00 100644 --- a/2.0/README.md +++ b/2.0/README.md @@ -47,6 +47,14 @@ applies the submitted patch to a clean skeleton, runs a hidden arena against multiple baseline bot families, and scores by mean baseline win rate with a small faster-win tiebreak. The online generals.io service is not used. +## RocksDB Native Compaction Policy + +This systems problem asks agents to patch the leveled compaction picker in a +pinned RocksDB checkout. Its problem ID is `rocksdb_native_compaction_policy`. +The judge runs native RocksDB workloads, checks snapshots and full database +contents, and scores paired improvements in write/read/space amplification and +compaction debt against unmodified RocksDB. + ## vLLM LLM-Serving Optimization This systems problem asks agents to patch a clean upstream vLLM checkout to diff --git a/2.0/problems/rocksdb_native_compaction_policy/config.yaml b/2.0/problems/rocksdb_native_compaction_policy/config.yaml new file mode 100644 index 000000000..dba8a83de --- /dev/null +++ b/2.0/problems/rocksdb_native_compaction_policy/config.yaml @@ -0,0 +1,49 @@ +tag: systems +runtime: + language: cpp + timeout_seconds: 10800 + environment: "Patch a pinned RocksDB v10.10.1 checkout; native correctness and compaction-cost judge" + apt_packages: + - bash + - build-essential + - ca-certificates + - git + - libbz2-dev + - libgflags-dev + - liblz4-dev + - libsnappy-dev + - libzstd-dev + - zlib1g-dev + docker: + image: python:3.12-slim-bookworm + judge_image: frontiercs/rocksdb-native-compaction-judge:experimental-v10.10.1-task2 + visible_inputs: + - source: /opt/rocksdb-clean + destination: /app/rocksdb +environment: + cpus: 8 + memory_mb: 16384 + storage_mb: 32768 + build_timeout_seconds: 7200 +evaluation: + schema_version: rocksdb-native-compaction-v2 + public_suite_id: rocksdb-native-public-v2 + final_suite_id: rocksdb-native-final-v2 + rocksdb_commit: "4595a5e95ae8525c42e172a054435782b3479c57" + feedback_cases: + - {seed: 1101, profile: smoke} + - {seed: 1202, profile: l0_pressure} + - {seed: 1303, profile: range_snapshot} + - {seed: 1404, profile: scanmix} + - {seed: 1505, profile: multi_cf} + - {seed: 1606, profile: time_series} + - {seed: 1707, profile: difficulty} + - {seed: 1808, profile: overlap_rewrite} + build_timeout_seconds: 7200 + run_timeout_seconds: 1800 + build_jobs: 3 +submission: + kind: file + path: /app/solution.patch + allow_empty: true + max_queue_size: 2 diff --git a/2.0/problems/rocksdb_native_compaction_policy/docker/build_images.sh b/2.0/problems/rocksdb_native_compaction_policy/docker/build_images.sh new file mode 100644 index 000000000..79fa000bf --- /dev/null +++ b/2.0/problems/rocksdb_native_compaction_policy/docker/build_images.sh @@ -0,0 +1,27 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROCKSDB_COMMIT="${ROCKSDB_COMMIT:-4595a5e95ae8525c42e172a054435782b3479c57}" +ROCKSDB_ARCHIVE_SHA256="${ROCKSDB_ARCHIVE_SHA256:-9469e7d24644e298134d0464ec2d398a9b91cc401102009ec1ecde0383485d6c}" +ROCKSDB_BUILD_JOBS="${ROCKSDB_BUILD_JOBS:-3}" +JUDGE_TAG="${JUDGE_TAG:-frontiercs/rocksdb-native-compaction-judge:experimental-v10.10.1-task2}" +PUSH_IMAGE="${PUSH_IMAGE:-0}" +PLATFORMS="${PLATFORMS:-linux/amd64,linux/arm64}" +PROBLEM_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd) + +BUILD_ARGS=( + --build-arg "ROCKSDB_COMMIT=${ROCKSDB_COMMIT}" \ + --build-arg "ROCKSDB_ARCHIVE_SHA256=${ROCKSDB_ARCHIVE_SHA256}" \ + --build-arg "ROCKSDB_BUILD_JOBS=${ROCKSDB_BUILD_JOBS}" \ + -f "${PROBLEM_DIR}/docker/judge/Dockerfile" \ + -t "${JUDGE_TAG}" \ + "${PROBLEM_DIR}" +) + +if [[ "${PUSH_IMAGE}" == 1 ]]; then + docker buildx build --platform "${PLATFORMS}" --push "${BUILD_ARGS[@]}" + echo "Published ${JUDGE_TAG} for ${PLATFORMS}" +else + DOCKER_BUILDKIT=1 docker build "${BUILD_ARGS[@]}" + echo "Built ${JUDGE_TAG} for the local platform" +fi diff --git a/2.0/problems/rocksdb_native_compaction_policy/docker/judge/Dockerfile b/2.0/problems/rocksdb_native_compaction_policy/docker/judge/Dockerfile new file mode 100644 index 000000000..6cf6d118a --- /dev/null +++ b/2.0/problems/rocksdb_native_compaction_policy/docker/judge/Dockerfile @@ -0,0 +1,82 @@ +# syntax=docker/dockerfile:1.7 +FROM python:3.12-slim-bookworm + +ARG ROCKSDB_COMMIT=4595a5e95ae8525c42e172a054435782b3479c57 +ARG ROCKSDB_ARCHIVE_SHA256=9469e7d24644e298134d0464ec2d398a9b91cc401102009ec1ecde0383485d6c +ARG ROCKSDB_BUILD_JOBS=3 +ARG DEBIAN_FRONTEND=noninteractive + +RUN apt-get update && \ + apt-get install -y --no-install-recommends \ + bash \ + build-essential \ + ca-certificates \ + cmake \ + git \ + ninja-build \ + pkg-config \ + libgflags-dev \ + libsnappy-dev \ + zlib1g-dev \ + libbz2-dev \ + liblz4-dev \ + libzstd-dev \ + python3 && \ + rm -rf /var/lib/apt/lists/* + +# Prebuild one tree for incremental candidate builds, then copy it for vanilla. +RUN python3 - "${ROCKSDB_COMMIT}" "${ROCKSDB_ARCHIVE_SHA256}" <<'PY' +import hashlib +import io +import shutil +import sys +import tarfile +import time +import urllib.request +from pathlib import Path + +commit, expected_sha256 = sys.argv[1:] +url = f"https://codeload.github.com/facebook/rocksdb/tar.gz/{commit}" +for attempt in range(5): + try: + request = urllib.request.Request(url, headers={"User-Agent": "Frontier-CS"}) + with urllib.request.urlopen(request, timeout=30) as response: + archive = response.read() + actual_sha256 = hashlib.sha256(archive).hexdigest() + if actual_sha256 != expected_sha256: + raise ValueError(f"archive sha256 mismatch: {actual_sha256}") + break + except Exception: + if attempt == 4: + raise + time.sleep(3 * (attempt + 1)) + +extract_dir = Path("/tmp/rocksdb-extract") +shutil.rmtree(extract_dir, ignore_errors=True) +extract_dir.mkdir(parents=True) +with tarfile.open(fileobj=io.BytesIO(archive), mode="r:gz") as source: + source.extractall(extract_dir, filter="data") +roots = list(extract_dir.iterdir()) +if len(roots) != 1 or not roots[0].is_dir(): + raise RuntimeError("unexpected RocksDB archive layout") +shutil.move(str(roots[0]), "/opt/rocksdb-clean") +shutil.rmtree(extract_dir) +PY + +RUN printf '%s\n' "${ROCKSDB_COMMIT}" \ + > /opt/rocksdb-clean/.frontier-upstream-commit && \ + git -C /opt/rocksdb-clean init -q && \ + git -C /opt/rocksdb-clean config user.name "Frontier-CS" && \ + git -C /opt/rocksdb-clean config user.email frontier-cs@example.invalid && \ + git -C /opt/rocksdb-clean add -A && \ + GIT_AUTHOR_DATE=2026-01-01T00:00:00Z \ + GIT_COMMITTER_DATE=2026-01-01T00:00:00Z \ + git -C /opt/rocksdb-clean commit -q -m "Pinned RocksDB ${ROCKSDB_COMMIT}" && \ + cd /opt/rocksdb-clean && \ + PORTABLE=1 DEBUG_LEVEL=0 DISABLE_WARNING_AS_ERROR=1 \ + make -j"${ROCKSDB_BUILD_JOBS}" static_lib && \ + cp -a /opt/rocksdb-clean /opt/rocksdb-vanilla + +COPY judge/harness/lsm_policy_harness.cc /judge/harness/lsm_policy_harness.cc + +WORKDIR /judge diff --git a/2.0/problems/rocksdb_native_compaction_policy/evaluate.sh b/2.0/problems/rocksdb_native_compaction_policy/evaluate.sh new file mode 100644 index 000000000..7234ee295 --- /dev/null +++ b/2.0/problems/rocksdb_native_compaction_policy/evaluate.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) + +if [[ $# -gt 0 ]]; then + exec python3 "$SCRIPT_DIR/evaluator.py" "$1" +fi + +for candidate in \ + /work/execution_env/solution_env/solution.patch \ + /work/execution_env/solution_env/solution.cpp; do + if [[ -f "$candidate" ]]; then + exec python3 "$SCRIPT_DIR/evaluator.py" "$candidate" + fi +done + +echo "Error: solution artifact is missing" >&2 +exit 1 diff --git a/2.0/problems/rocksdb_native_compaction_policy/evaluator.py b/2.0/problems/rocksdb_native_compaction_policy/evaluator.py new file mode 100644 index 000000000..ad9441960 --- /dev/null +++ b/2.0/problems/rocksdb_native_compaction_policy/evaluator.py @@ -0,0 +1,934 @@ +"""Judge for the RocksDB native compaction-policy task.""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import re +import secrets +import shlex +import shutil +import statistics +import subprocess +import sys +import time +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + + +TASK_CONFIG_PATH = Path(os.environ.get("TASK_CONFIG_PATH", "/judge/task_config.json")) +DEFAULT_COMMIT = "4595a5e95ae8525c42e172a054435782b3479c57" +SOURCE_COMMIT_MARKER = ".frontier-upstream-commit" +MAX_PATCH_BYTES = 256 * 1024 +MAX_CHANGED_FILES = 5 + +ALLOWLIST = frozenset( + { + "db/compaction/compaction_picker.cc", + "db/compaction/compaction_picker.h", + "db/compaction/compaction_picker_level.cc", + "db/compaction/compaction_picker_level.h", + "db/version_set.cc", + } +) + +KNOWN_PROFILES = frozenset( + { + "smoke", + "l0_pressure", + "range_snapshot", + "scanmix", + "multi_cf", + "time_series", + "difficulty", + "overlap_rewrite", + "full", + } +) + +FINAL_PROFILES = ( + "l0_pressure", + "range_snapshot", + "scanmix", + "multi_cf", + "time_series", + "difficulty", + "overlap_rewrite", +) + +DEFAULT_FINAL_SUITE_KEY = "rocksdb-native-compaction-v2-development" + +STRUCTURAL_MARKERS = ( + "new file mode", + "deleted file mode", + "rename from", + "rename to", + "copy from", + "copy to", + "old mode ", + "new mode ", + "GIT binary patch", + "Subproject commit", +) + +FORBIDDEN_IDENTIFIERS = frozenset( + { + "getenv", + "secure_getenv", + "system", + "popen", + "fork", + "execve", + "dlopen", + "dlsym", + "readlink", + "realpath", + "getcwd", + "clock_gettime", + "rdtsc", + "rdtscp", + "thread_local", + "argv", + "argc", + "GetEnv", + "GetFileSystem", + "NowMicros", + "NowNanos", + "NowSeconds", + "SleepForMicroseconds", + "GetCurrentTime", + "getpid", + "getppid", + "syscall", + "timespec_get", + "gettimeofday", + "getrusage", + "uname", + "sysinfo", + "sched_getcpu", + "gethostname", + "getuid", + "geteuid", + "environ", + "srand", + "getrandom", + "arc4random", + "random_device", + } +) + +FORBIDDEN_CALL_IDENTIFIERS = frozenset({"clock"}) + +FORBIDDEN_FRAGMENTS = ( + "/proc", + "/sys", + "frontier", + "harbor", + "submission", + "harness", + "profile", + "case-file", + "std::chrono", + "system_clock", + "steady_clock", + "high_resolution_clock", + "std::random_device", + "std::filesystem", + "program_invocation_name", + "ioptions_.clock", + "mutable_db_options_.clock", + "Env::", + "FileSystem::", + "std::ifstream", + "std::ofstream", +) + +TOKEN_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") +PREPROCESSOR_RE = re.compile(r"^\s*(?:#|%:|\?\?=)\s*[A-Za-z_][A-Za-z0-9_]*") + + +def _load_config() -> dict[str, Any]: + try: + payload = json.loads(TASK_CONFIG_PATH.read_text(encoding="utf-8")) + return payload if isinstance(payload, dict) else {} + except Exception: + pass + + local = Path(__file__).with_name("config.yaml") + if not local.exists(): + return {} + cases = [ + {"seed": int(seed), "profile": profile} + for seed, profile in re.findall( + r"\{\s*seed:\s*(\d+)\s*,\s*profile:\s*([A-Za-z0-9_]+)\s*\}", + local.read_text(encoding="utf-8"), + ) + ] + return {"evaluation": {"feedback_cases": cases}} + + +CONFIG = _load_config() +EVALUATION_CONFIG = ( + CONFIG.get("evaluation", {}) + if isinstance(CONFIG.get("evaluation"), dict) + else {} +) + + +def _config_int(name: str, default: int) -> int: + value = os.environ.get(name) + if value is None: + value = EVALUATION_CONFIG.get(name.lower(), default) + try: + return int(value) + except (TypeError, ValueError): + return default + + +PINNED_COMMIT = str(EVALUATION_CONFIG.get("rocksdb_commit", DEFAULT_COMMIT)) +VANILLA_DIR = Path(os.environ.get("ROCKSDB_VANILLA_DIR", "/opt/rocksdb-vanilla")) +WARM_DIR = Path(os.environ.get("ROCKSDB_WARM_DIR", "/opt/rocksdb-clean")) +HARNESS_SRC = Path( + os.environ.get("HARNESS_SRC", "/judge/harness/lsm_policy_harness.cc") +) +WORK_DIR = Path(os.environ.get("WORK_DIR", "/work")) +BUILD_TIMEOUT = _config_int("BUILD_TIMEOUT_SECONDS", 7200) +RUN_TIMEOUT = _config_int("RUN_TIMEOUT_SECONDS", 1800) +BUILD_JOBS = _config_int("BUILD_JOBS", 3) +COMPRESS_LINK = ["-lz", "-lbz2", "-llz4", "-lzstd", "-lsnappy"] + +_FINAL_CASES: tuple[tuple[int, str], ...] | None = None + + +@dataclass +class PatchFile: + diff_old: str + diff_new: str + old_path: str = "" + new_path: str = "" + hunks: int = 0 + added: list[str] = field(default_factory=list) + removed: list[str] = field(default_factory=list) + + @property + def path(self) -> str: + return self.new_path or self.diff_new + + +def _strip_prefix(path: str) -> str: + path = path.strip().split("\t", 1)[0] + if path.startswith("a/") or path.startswith("b/"): + return path[2:] + return path + + +def _unsafe_path(path: str) -> bool: + return ( + not path + or path.startswith("/") + or "\\" in path + or "\x00" in path + or ".." in Path(path).parts + ) + + +def _diff_paths(line: str) -> tuple[str, str]: + try: + parts = shlex.split(line) + except ValueError as exc: + raise ValueError("malformed diff header") from exc + if len(parts) != 4 or parts[:2] != ["diff", "--git"]: + raise ValueError("malformed diff header") + return _strip_prefix(parts[2]), _strip_prefix(parts[3]) + + +def _parse_patch(text: str) -> list[PatchFile]: + files: list[PatchFile] = [] + current: PatchFile | None = None + for line in text.splitlines(): + if line.startswith("diff --git "): + old, new = _diff_paths(line) + current = PatchFile(old, new) + files.append(current) + continue + if current is None: + continue + if line.startswith("--- "): + current.old_path = _strip_prefix(line[4:]) + elif line.startswith("+++ "): + current.new_path = _strip_prefix(line[4:]) + elif line.startswith("@@ "): + current.hunks += 1 + elif line.startswith("+") and not line.startswith("+++ "): + current.added.append(line[1:]) + elif line.startswith("-") and not line.startswith("--- "): + current.removed.append(line[1:]) + return files + + +def _forbidden_added_line(line: str) -> str | None: + if PREPROCESSOR_RE.search(line): + return "preprocessor directive" + lowered = line.lower() + for fragment in FORBIDDEN_FRAGMENTS: + if fragment.lower() in lowered: + return fragment + identifiers = set(TOKEN_RE.findall(line)) + for identifier in FORBIDDEN_IDENTIFIERS: + if identifier in identifiers: + return identifier + for identifier in FORBIDDEN_CALL_IDENTIFIERS: + if re.search(rf"\b{re.escape(identifier)}\s*\(", line): + return f"{identifier}()" + return None + + +def _validate_patch_text( + text: str, + *, + allow_empty: bool, +) -> tuple[bool, str, dict[str, Any]]: + metrics: dict[str, Any] = { + "valid_patch": 0, + "patch_bytes": len(text.encode("utf-8", errors="replace")), + } + if metrics["patch_bytes"] > MAX_PATCH_BYTES: + return False, "patch exceeds 256 KiB", metrics + if not text.strip(): + if not allow_empty: + return False, "empty post-apply diff", metrics + metrics.update({"valid_patch": 1, "empty_patch": 1, "changed_files": []}) + return True, "empty no-op patch", metrics + for marker in STRUCTURAL_MARKERS: + if marker in text: + return False, f"forbidden structural change: {marker.strip()}", metrics + + try: + files = _parse_patch(text) + except ValueError as exc: + return False, str(exc), metrics + if not files: + return False, "submission is not a git patch", metrics + if len(files) > MAX_CHANGED_FILES: + return False, "patch changes too many files", metrics + + changed: list[str] = [] + for patch_file in files: + path = patch_file.path + if patch_file.diff_old != patch_file.diff_new: + return False, "renames are not allowed", metrics + if patch_file.old_path != patch_file.diff_old: + return False, f"old patch path mismatch for {path}", metrics + if patch_file.new_path != patch_file.diff_new: + return False, f"new patch path mismatch for {path}", metrics + if _unsafe_path(path): + return False, f"unsafe patch path: {path}", metrics + if path not in ALLOWLIST: + return False, f"changed file is outside the editable surface: {path}", metrics + if patch_file.hunks == 0 or not (patch_file.added or patch_file.removed): + return False, f"patch for {path} has no changes", metrics + for line in patch_file.added: + reason = _forbidden_added_line(line) + if reason: + return False, f"forbidden added code in {path}: {reason}", metrics + changed.append(path) + + metrics.update( + { + "valid_patch": 1, + "empty_patch": 0, + "changed_files": sorted(set(changed)), + } + ) + return True, "patch accepted by static policy", metrics + + +def validate_patch(path: Path) -> tuple[bool, str, dict[str, Any]]: + try: + text = path.read_text(encoding="utf-8", errors="replace") + except OSError: + return False, "solution patch is unreadable", {"valid_patch": 0} + return _validate_patch_text(text, allow_empty=True) + + +def _run( + command: list[str], + *, + cwd: Path | None = None, + timeout: int, + env: dict[str, str] | None = None, +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + command, + cwd=str(cwd) if cwd else None, + timeout=timeout, + env=env, + text=True, + capture_output=True, + check=True, + ) + + +def _source_commit(source: Path) -> str: + return source.joinpath(SOURCE_COMMIT_MARKER).read_text(encoding="utf-8").strip() + + +def _ensure_sources() -> None: + for source in (VANILLA_DIR, WARM_DIR): + if not source.joinpath(".git").is_dir(): + raise RuntimeError("pinned RocksDB source is missing") + if _source_commit(source) != PINNED_COMMIT: + raise RuntimeError("RocksDB source is not at the pinned commit") + if not source.joinpath("librocksdb.a").is_file(): + raise RuntimeError("prebuilt RocksDB static library is missing") + if not HARNESS_SRC.is_file(): + raise RuntimeError("judge harness source is missing") + + +def _compile_harness(source: Path, output: Path) -> None: + _run( + [ + os.environ.get("CXX", "g++"), + "-O2", + "-DNDEBUG", + "-std=c++20", + "-fno-rtti", + f"-I{source / 'include'}", + f"-I{source}", + str(HARNESS_SRC), + str(source / "librocksdb.a"), + "-o", + str(output), + "-lpthread", + "-ldl", + *COMPRESS_LINK, + ], + timeout=600, + ) + + +def _restore_warm_source() -> None: + dirty = _run( + ["git", "diff", "--name-only", "--", *sorted(ALLOWLIST)], + cwd=WARM_DIR, + timeout=120, + ).stdout.splitlines() + if dirty: + _run( + ["git", "checkout", "HEAD", "--", *dirty], + cwd=WARM_DIR, + timeout=300, + ) + + +def _build_candidate(patch_path: Path) -> Path: + _ensure_sources() + _restore_warm_source() + text = patch_path.read_text(encoding="utf-8", errors="replace") + if text.strip(): + _run(["git", "apply", "--check", str(patch_path)], cwd=WARM_DIR, timeout=120) + _run(["git", "apply", str(patch_path)], cwd=WARM_DIR, timeout=120) + actual = _run( + ["git", "diff", "--binary", "--no-color"], + cwd=WARM_DIR, + timeout=120, + ).stdout + ok, message, _ = _validate_patch_text(actual, allow_empty=False) + if not ok: + raise RuntimeError(message) + + build_env = { + "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"), + "HOME": "/tmp", + "LANG": "C", + "LC_ALL": "C", + "PORTABLE": "1", + "DEBUG_LEVEL": "0", + "DISABLE_WARNING_AS_ERROR": "1", + } + _run( + ["make", f"-j{BUILD_JOBS}", "static_lib"], + cwd=WARM_DIR, + timeout=BUILD_TIMEOUT, + env=build_env, + ) + output = WORK_DIR / "candidate_harness" + _compile_harness(WARM_DIR, output) + return output + + +def _ensure_vanilla_harness() -> Path: + _ensure_sources() + output = WORK_DIR / "vanilla_harness" + if not output.exists(): + _compile_harness(VANILLA_DIR, output) + return output + + +def _harness_env() -> dict[str, str]: + return { + "PATH": "/usr/bin:/bin", + "HOME": "/nonexistent", + "LANG": "C", + "LC_ALL": "C", + } + + +def _run_harness( + binary: Path, + trusted_inspector: Path, + seed: int, + profile: str, +) -> dict[str, Any]: + run_id = secrets.token_hex(12) + db_dir = WORK_DIR / f"db_{run_id}" + output = WORK_DIR / f"result_{run_id}.json" + trusted_output = WORK_DIR / f"trusted_{run_id}.json" + case_file = WORK_DIR / f"case_{run_id}.txt" + case_file.write_text(f"seed={seed}\nprofile={profile}\n", encoding="utf-8") + try: + started = time.monotonic() + _run( + [ + str(binary), + "--db", + str(db_dir), + "--out", + str(output), + "--case-file", + str(case_file), + ], + timeout=RUN_TIMEOUT, + env=_harness_env(), + ) + payload = json.loads(output.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError("harness result is not an object") + workload_seconds = time.monotonic() - started + trusted_started = time.monotonic() + _run( + [ + str(trusted_inspector), + "--trusted-residual", + "--db", + str(db_dir), + "--out", + str(trusted_output), + "--seed", + str(seed), + "--profile", + profile, + ], + timeout=RUN_TIMEOUT, + env=_harness_env(), + ) + trusted = json.loads(trusted_output.read_text(encoding="utf-8")) + if not isinstance(trusted, dict) or not trusted.get("ok"): + raise ValueError("trusted drain failed") + for key in ( + "compaction_output_bytes", + "final_table_bytes", + "final_files", + "l0_files", + "upper_level_files", + "upper_level_bytes", + "passes", + "digest_entries", + "digest_bytes", + ): + payload[f"trusted_{key}"] = trusted.get(key) + payload["_wall_seconds"] = workload_seconds + payload["_trusted_wall_seconds"] = time.monotonic() - trusted_started + return payload + finally: + shutil.rmtree(db_dir, ignore_errors=True) + output.unlink(missing_ok=True) + trusted_output.unlink(missing_ok=True) + case_file.unlink(missing_ok=True) + + +def _number(run: dict[str, Any], key: str, *, minimum: float = 0.0) -> float: + try: + value = float(run[key]) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError(f"missing metric: {key}") from exc + if not math.isfinite(value) or value < minimum: + raise ValueError(f"invalid metric: {key}") + return value + + +def _objective_metrics(run: dict[str, Any]) -> dict[str, float]: + user_bytes = _number(run, "user_put_bytes", minimum=1.0) + policy_drain_bytes = _number(run, "drain_compaction_bytes") + residual_bytes = _number(run, "trusted_compaction_output_bytes") + effective_drain_bytes = policy_drain_bytes + residual_bytes + table_bytes = _number(run, "table_created_bytes") + denominator = max(user_bytes, 64.0 * 1024 * 1024) + return { + "write_amp": (table_bytes + residual_bytes) / user_bytes, + "read_amp": _number(run, "read_amp"), + "space_amp": _number(run, "pre_drain_space_amp", minimum=1e-9), + "debt": 1.0 + effective_drain_bytes / denominator, + } + + +METRIC_WEIGHTS = { + "write_amp": 0.40, + "read_amp": 0.25, + "space_amp": 0.20, + "debt": 0.15, +} + +METRIC_FLOORS = { + "write_amp": 0.05, + "read_amp": 0.05, + "space_amp": 0.05, + "debt": 0.0, +} + +MIN_ROBUST_GAIN = 1.005 +TARGET_ROBUST_GAIN = 1.20 +PROFILE_IMPROVEMENT_GAIN = MIN_ROBUST_GAIN +PROFILE_REGRESSION_GAIN = 0.98 +MIN_IMPROVED_PROFILE_FRACTION = 0.40 + + +def _paired_case( + vanilla_binary: Path, + candidate_binary: Path, + seed: int, + profile: str, +) -> dict[str, Any]: + ratio_samples = {name: [] for name in METRIC_WEIGHTS} + with ThreadPoolExecutor(max_workers=2) as executor: + futures = { + "vanilla": executor.submit( + _run_harness, vanilla_binary, vanilla_binary, seed, profile + ), + "candidate": executor.submit( + _run_harness, candidate_binary, vanilla_binary, seed, profile + ), + } + runs = {label: future.result() for label, future in futures.items()} + + vanilla = runs["vanilla"] + candidate = runs["candidate"] + if not vanilla.get("ok") or not vanilla.get("correctness_ok"): + raise RuntimeError("vanilla harness failed") + if not candidate.get("ok") or not candidate.get("correctness_ok"): + return {"ok": False, "reason": "candidate correctness or runtime failure"} + if candidate.get("base_fingerprint") != vanilla.get("base_fingerprint"): + return {"ok": False, "reason": "candidate changed the fixed base load"} + + vanilla_wall = _number(vanilla, "_wall_seconds", minimum=1e-6) + candidate_wall = _number(candidate, "_wall_seconds", minimum=1e-6) + if candidate_wall > vanilla_wall * 1.50 + 5.0: + return {"ok": False, "reason": "candidate exceeded the runtime guard"} + vanilla_stall = _number(vanilla, "stall_micros") + candidate_stall = _number(candidate, "stall_micros") + if candidate_stall > vanilla_stall * 1.25 + 10_000.0: + return {"ok": False, "reason": "candidate exceeded the stall guard"} + + vanilla_metrics = _objective_metrics(vanilla) + candidate_metrics = _objective_metrics(candidate) + for name in METRIC_WEIGHTS: + floor = METRIC_FLOORS[name] + ratio = (vanilla_metrics[name] + floor) / (candidate_metrics[name] + floor) + ratio_samples[name].append(ratio) + + ratios: dict[str, float] = {} + weighted_log_gain = 0.0 + for name, weight in METRIC_WEIGHTS.items(): + ratio = math.exp(statistics.mean(math.log(value) for value in ratio_samples[name])) + ratios[name] = ratio + weighted_log_gain += weight * math.log(min(2.0, max(0.5, ratio))) + return { + "ok": True, + "profile": profile, + "gain": math.exp(weighted_log_gain), + "ratios": ratios, + "candidate_intra_l0": int(candidate.get("intra_l0_compactions", 0)), + "vanilla_intra_l0": int(vanilla.get("intra_l0_compactions", 0)), + "runtime_ratio": candidate_wall / vanilla_wall, + } + + +def _score_cases(cases: list[dict[str, Any]]) -> tuple[float, float, dict[str, Any]]: + if not cases: + raise ValueError("no scored cases") + logs = [math.log(float(case["gain"])) for case in cases] + mean_log = statistics.mean(logs) + stderr = statistics.stdev(logs) / math.sqrt(len(logs)) if len(logs) > 1 else 0.0 + robust_log = mean_log - 0.10 * stderr + composite_gain = math.exp(mean_log) + robust_gain = math.exp(robust_log) + worst_gain = min(float(case["gain"]) for case in cases) + component_floor = min( + float(ratio) for case in cases for ratio in case["ratios"].values() + ) + profile_logs: dict[str, list[float]] = {} + for case, log_gain in zip(cases, logs): + profile_logs.setdefault(str(case.get("profile", "")), []).append(log_gain) + seed_spreads = [ + max(values) - min(values) + for values in profile_logs.values() + if len(values) > 1 + ] + profile_gains = { + profile: math.exp(statistics.mean(values)) + for profile, values in profile_logs.items() + } + required_improved_profiles = max( + 2, + math.ceil(len(profile_gains) * MIN_IMPROVED_PROFILE_FRACTION), + ) + improved_profile_count = sum( + gain >= PROFILE_IMPROVEMENT_GAIN for gain in profile_gains.values() + ) + material_regression_count = sum( + gain < PROFILE_REGRESSION_GAIN for gain in profile_gains.values() + ) + + unbounded = 100.0 * ( + robust_log - math.log(MIN_ROBUST_GAIN) + ) / math.log(TARGET_ROBUST_GAIN / MIN_ROBUST_GAIN) + bounded = max(0.0, min(100.0, unbounded)) + if ( + improved_profile_count < required_improved_profiles + or material_regression_count > 1 + or worst_gain < 0.75 + or component_floor <= 0.50 + ): + bounded = 0.0 + elif component_floor < 0.67: + bounded = min(bounded, 10.0) + elif component_floor < 0.80: + bounded = min(bounded, 25.0) + elif worst_gain < 0.90: + bounded = min(bounded, 25.0) + + metrics = { + "composite_gain": round(composite_gain, 5), + "robust_gain": round(robust_gain, 5), + "worst_case_gain": round(worst_gain, 5), + "component_floor": round(component_floor, 5), + "profile_count": len(profile_gains), + "improved_profile_count": improved_profile_count, + "required_improved_profile_count": required_improved_profiles, + "material_regression_count": material_regression_count, + "mean_seed_log_spread": round(statistics.mean(seed_spreads), 5) + if seed_spreads + else 0.0, + "max_seed_log_spread": round(max(seed_spreads), 5) if seed_spreads else 0.0, + "case_count": len(cases), + "intra_l0_delta": round( + sum( + case["candidate_intra_l0"] - case["vanilla_intra_l0"] + for case in cases + ) + / len(cases), + 3, + ), + "max_runtime_ratio": round( + max(float(case.get("runtime_ratio", 1.0)) for case in cases), 3 + ), + } + return bounded, unbounded, metrics + + +def _feedback_cases() -> tuple[tuple[int, str], ...]: + raw = EVALUATION_CONFIG.get("feedback_cases", []) + cases: list[tuple[int, str]] = [] + if isinstance(raw, list): + for item in raw: + if not isinstance(item, dict): + continue + seed = int(item.get("seed", 0)) + profile = str(item.get("profile", "")) + if seed > 0 and profile in KNOWN_PROFILES: + cases.append((seed, profile)) + return tuple(cases) or ( + (1101, "smoke"), + (1202, "l0_pressure"), + (1303, "range_snapshot"), + (1404, "scanmix"), + (1505, "multi_cf"), + (1606, "time_series"), + (1707, "difficulty"), + (1808, "overlap_rewrite"), + ) + + +def _generate_final_cases() -> tuple[tuple[int, str], ...]: + suite_key = os.environ.get( + "FRONTIER_LSM_FINAL_SUITE_KEY", DEFAULT_FINAL_SUITE_KEY + ) + cases: list[tuple[int, str]] = [] + used_seeds: set[int] = set() + for profile in FINAL_PROFILES: + for replica in range(2): + counter = 0 + while True: + material = f"{suite_key}:{profile}:{replica}:{counter}".encode() + seed = int.from_bytes(hashlib.sha256(material).digest()[:8], "big") + seed &= (1 << 63) - 1 + seed = seed or 1 + if seed not in used_seeds: + break + counter += 1 + used_seeds.add(seed) + cases.append((seed, profile)) + return tuple(cases) + + +def _is_final_role() -> bool: + return os.environ.get("FRONTIER_SUBMISSION_ROLE", "agent") == "final" + + +def _cases_for_role() -> tuple[tuple[int, str], ...]: + global _FINAL_CASES + if not _is_final_role(): + return _feedback_cases() + if _FINAL_CASES is None: + _FINAL_CASES = _generate_final_cases() + return _FINAL_CASES + + +def _public_metrics(metrics: dict[str, Any]) -> dict[str, Any]: + keys = ( + "valid_patch", + "empty_patch", + "build_ok", + "changed_files_count", + "case_count", + "composite_gain", + "robust_gain", + "worst_case_gain", + "component_floor", + "profile_count", + "improved_profile_count", + "required_improved_profile_count", + "material_regression_count", + "mean_seed_log_spread", + "max_seed_log_spread", + "intra_l0_delta", + "max_runtime_ratio", + "score_band", + ) + return {key: metrics[key] for key in keys if key in metrics} + + +def _invalid(message: str, metrics: dict[str, Any] | None = None): + payload = dict(metrics or {}) + payload.setdefault("valid_patch", 0) + return 0.0, 0.0, message, _public_metrics(payload) + + +def _band(score: float) -> str: + if score <= 0: + return "baseline" + if score < 25: + return "measurable" + if score < 60: + return "broad" + if score < 85: + return "strong" + return "frontier" + + +def full_evaluation( + patch_path: Path, + metrics: dict[str, Any], +) -> tuple[float, float, str, dict[str, Any]]: + WORK_DIR.mkdir(parents=True, exist_ok=True) + candidate = _build_candidate(patch_path) + vanilla = _ensure_vanilla_harness() + metrics["build_ok"] = 1 + + results: list[dict[str, Any]] = [] + for seed, profile in _cases_for_role(): + result = _paired_case(vanilla, candidate, seed, profile) + if not result.get("ok"): + return _invalid(str(result.get("reason", "candidate run failed")), metrics) + results.append(result) + + scored_results = [result for result in results if result["profile"] != "smoke"] + score, score_unbounded, score_metrics = _score_cases(scored_results or results) + metrics.update(score_metrics) + metrics["score_band"] = _band(score) + metrics["changed_files_count"] = len(metrics.get("changed_files", [])) + return ( + score, + score_unbounded, + f"scored; band={metrics['score_band']}", + _public_metrics(metrics), + ) + + +def evaluate(solution_path: str) -> tuple[float, float, str, dict[str, Any]]: + patch_path = Path(solution_path) + ok, message, metrics = validate_patch(patch_path) + if not ok: + return _invalid(message, metrics) + metrics["changed_files_count"] = len(metrics.get("changed_files", [])) + if metrics.get("empty_patch"): + metrics["score_band"] = "baseline" + return 0.0, 0.0, "empty no-op baseline", _public_metrics(metrics) + try: + return full_evaluation(patch_path, metrics) + except subprocess.TimeoutExpired: + return _invalid("build or benchmark timed out", metrics) + except subprocess.CalledProcessError as exc: + command = exc.cmd[0] if isinstance(exc.cmd, (list, tuple)) else str(exc.cmd) + metrics["failed_command"] = Path(str(command)).name + return _invalid("build, patch apply, or benchmark command failed", metrics) + except Exception as exc: + metrics["error_type"] = type(exc).__name__ + return _invalid("evaluation failed", metrics) + + +def prepare() -> dict[str, Any]: + global _FINAL_CASES + WORK_DIR.mkdir(parents=True, exist_ok=True) + _FINAL_CASES = _generate_final_cases() + _ensure_vanilla_harness() + suite_commitment = hashlib.sha256( + json.dumps(_FINAL_CASES, separators=(",", ":")).encode() + ).hexdigest() + return { + "vanilla_harness": "ready", + "final_suite_id": str( + EVALUATION_CONFIG.get("final_suite_id", "rocksdb-native-v1") + ), + "final_suite_commitment": suite_commitment, + "final_case_count": len(_FINAL_CASES), + } + + +def main(argv: list[str]) -> int: + if len(argv) != 2: + print("usage: evaluator.py SOLUTION_PATCH", file=sys.stderr) + return 2 + score, score_unbounded, message, metrics = evaluate(argv[1]) + print( + json.dumps( + { + "score": score, + "score_unbounded": score_unbounded, + "message": message, + "metrics": metrics, + }, + sort_keys=True, + ), + file=sys.stderr, + ) + print(f"{score:.12f} {score_unbounded:.12f}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv)) diff --git a/2.0/problems/rocksdb_native_compaction_policy/harbor/app/make_submission.sh b/2.0/problems/rocksdb_native_compaction_policy/harbor/app/make_submission.sh new file mode 100644 index 000000000..f56edddf2 --- /dev/null +++ b/2.0/problems/rocksdb_native_compaction_policy/harbor/app/make_submission.sh @@ -0,0 +1,42 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROCKSDB_DIR="${ROCKSDB_DIR:-/app/rocksdb}" +OUT="${1:-/app/solution.patch}" +ALLOWLIST=( + db/compaction/compaction_picker.cc + db/compaction/compaction_picker.h + db/compaction/compaction_picker_level.cc + db/compaction/compaction_picker_level.h + db/version_set.cc +) + +if [[ ! -d "$ROCKSDB_DIR/.git" ]]; then + echo "RocksDB checkout not found at $ROCKSDB_DIR" >&2 + exit 2 +fi + +declare -A allowed=() +for path in "${ALLOWLIST[@]}"; do + allowed["$path"]=1 +done + +outside=() +while IFS= read -r -d '' path; do + [[ ${allowed[$path]+yes} ]] || outside+=("$path") +done < <( + { + git -C "$ROCKSDB_DIR" diff --name-only -z HEAD -- + git -C "$ROCKSDB_DIR" ls-files --others --exclude-standard -z + } +) + +if (( ${#outside[@]} )); then + echo "ERROR: checkout contains changes outside the editable surface:" >&2 + printf ' %q\n' "${outside[@]}" >&2 + exit 3 +fi + +git -C "$ROCKSDB_DIR" diff --binary --no-color HEAD -- "${ALLOWLIST[@]}" > "$OUT" +python3 /app/validate_solution_patch.py "$OUT" +echo "Wrote $OUT ($(wc -c < "$OUT" | tr -d ' ') bytes)." diff --git a/2.0/problems/rocksdb_native_compaction_policy/harbor/app/validate_solution_patch.py b/2.0/problems/rocksdb_native_compaction_policy/harbor/app/validate_solution_patch.py new file mode 100644 index 000000000..5ab5c7cee --- /dev/null +++ b/2.0/problems/rocksdb_native_compaction_policy/harbor/app/validate_solution_patch.py @@ -0,0 +1,229 @@ +#!/usr/bin/env python3 +"""Fast agent-side validation for the RocksDB patch submission.""" + +from __future__ import annotations + +import re +import shlex +import sys +from pathlib import Path + + +MAX_PATCH_BYTES = 256 * 1024 +MAX_CHANGED_FILES = 5 +ALLOWLIST = frozenset( + { + "db/compaction/compaction_picker.cc", + "db/compaction/compaction_picker.h", + "db/compaction/compaction_picker_level.cc", + "db/compaction/compaction_picker_level.h", + "db/version_set.cc", + } +) +STRUCTURAL_MARKERS = ( + "new file mode", + "deleted file mode", + "rename from", + "rename to", + "copy from", + "copy to", + "old mode ", + "new mode ", + "GIT binary patch", + "Subproject commit", +) +FORBIDDEN_IDENTIFIERS = frozenset( + { + "getenv", + "secure_getenv", + "system", + "popen", + "fork", + "execve", + "dlopen", + "dlsym", + "readlink", + "realpath", + "getcwd", + "clock_gettime", + "rdtsc", + "rdtscp", + "thread_local", + "argv", + "argc", + "GetEnv", + "GetFileSystem", + "NowMicros", + "NowNanos", + "NowSeconds", + "SleepForMicroseconds", + "GetCurrentTime", + "getpid", + "getppid", + "syscall", + "timespec_get", + "gettimeofday", + "getrusage", + "uname", + "sysinfo", + "sched_getcpu", + "gethostname", + "getuid", + "geteuid", + "environ", + "srand", + "getrandom", + "arc4random", + "random_device", + } +) +FORBIDDEN_CALL_IDENTIFIERS = frozenset({"clock"}) +FORBIDDEN_FRAGMENTS = ( + "/proc", + "/sys", + "frontier", + "harbor", + "submission", + "harness", + "profile", + "case-file", + "std::chrono", + "system_clock", + "steady_clock", + "high_resolution_clock", + "std::random_device", + "std::filesystem", + "program_invocation_name", + "ioptions_.clock", + "mutable_db_options_.clock", + "Env::", + "FileSystem::", + "std::ifstream", + "std::ofstream", +) +TOKEN_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") +PREPROCESSOR_RE = re.compile(r"^\s*(?:#|%:|\?\?=)\s*[A-Za-z_][A-Za-z0-9_]*") + + +def fail(message: str) -> None: + print(f"ERROR: invalid solution patch: {message}", file=sys.stderr) + raise SystemExit(7) + + +def strip_prefix(path: str) -> str: + path = path.strip().split("\t", 1)[0] + if path.startswith("a/") or path.startswith("b/"): + return path[2:] + return path + + +def unsafe_path(path: str) -> bool: + return ( + not path + or path.startswith("/") + or "\\" in path + or "\x00" in path + or ".." in Path(path).parts + ) + + +def diff_paths(line: str) -> tuple[str, str]: + try: + parts = shlex.split(line) + except ValueError: + fail("malformed diff header") + if len(parts) != 4 or parts[:2] != ["diff", "--git"]: + fail("malformed diff header") + return strip_prefix(parts[2]), strip_prefix(parts[3]) + + +def forbidden_added_line(line: str) -> str | None: + if PREPROCESSOR_RE.search(line): + return "preprocessor directive" + lowered = line.lower() + for fragment in FORBIDDEN_FRAGMENTS: + if fragment.lower() in lowered: + return fragment + identifiers = set(TOKEN_RE.findall(line)) + for identifier in FORBIDDEN_IDENTIFIERS: + if identifier in identifiers: + return identifier + for identifier in FORBIDDEN_CALL_IDENTIFIERS: + if re.search(rf"\b{re.escape(identifier)}\s*\(", line): + return f"{identifier}()" + return None + + +def validate(text: str) -> None: + if len(text.encode("utf-8", errors="replace")) > MAX_PATCH_BYTES: + fail("patch exceeds 256 KiB") + if not text.strip(): + return + for marker in STRUCTURAL_MARKERS: + if marker in text: + fail(f"forbidden structural change: {marker.strip()}") + + current_path = "" + old_path = "" + new_path = "" + hunks = 0 + changes = 0 + additions = 0 + removals = 0 + changed_files: list[str] = [] + + def finish_file() -> None: + if not current_path: + return + if old_path != current_path or new_path != current_path: + fail(f"patch path mismatch for {current_path}") + if hunks == 0 or changes == 0: + fail(f"patch for {current_path} has no changes") + + for raw_line in text.splitlines(): + if raw_line.startswith("diff --git "): + finish_file() + diff_old, diff_new = diff_paths(raw_line) + if diff_old != diff_new: + fail("renames are not allowed") + if unsafe_path(diff_new): + fail(f"unsafe patch path: {diff_new}") + if diff_new not in ALLOWLIST: + fail(f"changed file is outside the editable surface: {diff_new}") + current_path = diff_new + old_path = new_path = "" + hunks = changes = additions = removals = 0 + changed_files.append(current_path) + elif raw_line.startswith("--- "): + old_path = strip_prefix(raw_line[4:]) + elif raw_line.startswith("+++ "): + new_path = strip_prefix(raw_line[4:]) + elif raw_line.startswith("@@ "): + hunks += 1 + elif raw_line.startswith("+") and not raw_line.startswith("+++ "): + changes += 1 + additions += 1 + reason = forbidden_added_line(raw_line[1:]) + if reason: + fail(f"forbidden added code in {current_path}: {reason}") + elif raw_line.startswith("-") and not raw_line.startswith("--- "): + changes += 1 + removals += 1 + + finish_file() + if not changed_files: + fail("submission is not a git patch") + if len(changed_files) > MAX_CHANGED_FILES: + fail("patch changes too many files") + + +def main(argv: list[str]) -> int: + if len(argv) != 2: + print("usage: validate_solution_patch.py SOLUTION_PATCH", file=sys.stderr) + return 2 + validate(Path(argv[1]).read_text(encoding="utf-8", errors="replace")) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv)) diff --git a/2.0/problems/rocksdb_native_compaction_policy/judge/harness/lsm_policy_harness.cc b/2.0/problems/rocksdb_native_compaction_policy/judge/harness/lsm_policy_harness.cc new file mode 100644 index 000000000..6c5bd35a6 --- /dev/null +++ b/2.0/problems/rocksdb_native_compaction_policy/judge/harness/lsm_policy_harness.cc @@ -0,0 +1,1706 @@ +// Runs identical deterministic workloads against vanilla and candidate +// RocksDB builds. + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace fs = std::filesystem; +using Clock = std::chrono::steady_clock; + +// Helpers + +static inline uint64_t SplitMix64(uint64_t& x) { + uint64_t z = (x += 0x9e3779b97f4a7c15ULL); + z = (z ^ (z >> 30)) * 0xbf58476d1ce4e5b9ULL; + z = (z ^ (z >> 27)) * 0x94d049bb133111ebULL; + return z ^ (z >> 31); +} + +static inline uint64_t Rng(uint64_t seed, uint32_t phase, uint64_t op_index, + uint64_t stream) { + uint64_t x = seed ^ (0x9e3779b97f4a7c15ULL * (phase + 1)); + x ^= op_index * 0xbf58476d1ce4e5b9ULL; + x ^= stream * 0x94d049bb133111ebULL; + return SplitMix64(x); +} + +// 128-bit value hash (two independent FNV-1a streams). +static inline std::pair Hash128(const char* p, size_t n) { + uint64_t h1 = 1469598103934665603ULL; + uint64_t h2 = 1099511628211ULL ^ 0xff51afd7ed558ccdULL; + for (size_t i = 0; i < n; ++i) { + unsigned char c = static_cast(p[i]); + h1 = (h1 ^ c) * 1099511628211ULL; + h2 = (h2 ^ c) * 0x100000001b3ULL; + h2 = (h2 << 7) | (h2 >> 57); + } + return {h1, h2}; +} +static inline std::pair Hash128(const std::string& s) { + return Hash128(s.data(), s.size()); +} + +// Per-key token folded into an order-independent XOR aggregate. +static inline uint64_t KeyToken(uint64_t id, uint32_t size, uint64_t h1, + uint64_t h2) { + uint64_t x = id * 0x9e3779b97f4a7c15ULL; + x ^= (uint64_t(size) + 0x100) * 0xbf58476d1ce4e5b9ULL; + x ^= h1; + x ^= (h2 << 1) | (h2 >> 63); + return SplitMix64(x); +} + +static inline std::string MakeKey(uint64_t key_id, uint64_t seed) { + std::string k(16, '\0'); + uint64_t salt = key_id ^ (seed * 0x9e3779b97f4a7c15ULL); + for (int i = 0; i < 8; ++i) k[i] = char((key_id >> (56 - 8 * i)) & 0xff); + for (int i = 0; i < 8; ++i) k[8 + i] = char((salt >> (56 - 8 * i)) & 0xff); + return k; +} +static inline uint64_t KeyIdFromSlice(const char* p, size_t n) { + uint64_t id = 0; + for (int i = 0; i < 8 && size_t(i) < n; ++i) + id = (id << 8) | static_cast(p[i]); + return id; +} + +static inline std::string MakeValue(uint64_t seed, uint32_t phase, + uint64_t op_index, uint64_t key_id, + size_t size) { + std::string v(size, '\0'); + uint64_t x = seed ^ key_id ^ (uint64_t(phase) << 48) ^ op_index; + for (size_t i = 0; i < size; ++i) v[i] = char(SplitMix64(x) & 0xff); + return v; +} + +static double Percentile(std::vector& xs, double p) { + if (xs.empty()) return 0.0; + std::sort(xs.begin(), xs.end()); + double idx = p * (static_cast(xs.size()) - 1.0); + size_t lo = static_cast(idx); + size_t hi = std::min(lo + 1, xs.size() - 1); + double frac = idx - static_cast(lo); + return xs[lo] * (1.0 - frac) + xs[hi] * frac; +} + +static std::string JsonEscape(const std::string& s) { + std::ostringstream out; + for (unsigned char c : s) { + switch (c) { + case '\\': + out << "\\\\"; + break; + case '"': + out << "\\\""; + break; + case '\b': + out << "\\b"; + break; + case '\f': + out << "\\f"; + break; + case '\n': + out << "\\n"; + break; + case '\r': + out << "\\r"; + break; + case '\t': + out << "\\t"; + break; + default: + if (c < 0x20) { + out << "\\u"; + const char* hex = "0123456789abcdef"; + out << "00" << hex[(c >> 4) & 0xf] << hex[c & 0xf]; + } else { + out << char(c); + } + } + } + return out.str(); +} + +// Event listener + +class MetricsListener final : public rocksdb::EventListener { + public: + std::atomic flush_output_bytes{0}; + std::atomic compaction_output_bytes{0}; + std::atomic table_created_bytes{0}; // sanity cross-check + std::atomic compaction_events{0}; // diagnostic: # compactions + std::atomic intra_l0_compactions{0}; + std::atomic failed_compactions{0}; + std::atomic background_errors{0}; + + void OnFlushCompleted(rocksdb::DB*, + const rocksdb::FlushJobInfo& info) override { + uint64_t bytes = 0; + std::error_code ec; + if (!info.file_path.empty()) { + bytes = static_cast(fs::file_size(info.file_path, ec)); + if (ec) bytes = 0; + } + if (bytes == 0) { + const auto& p = info.table_properties; + bytes = p.data_size + p.index_size + p.filter_size; + } + flush_output_bytes.fetch_add(bytes, std::memory_order_relaxed); + } + void OnCompactionCompleted(rocksdb::DB*, + const rocksdb::CompactionJobInfo& info) override { + if (!info.status.ok()) { + failed_compactions.fetch_add(1, std::memory_order_relaxed); + return; + } + compaction_events.fetch_add(1, std::memory_order_relaxed); + if (info.base_input_level == 0 && info.output_level == 0) { + intra_l0_compactions.fetch_add(1, std::memory_order_relaxed); + } + compaction_output_bytes.fetch_add(info.stats.total_output_bytes, + std::memory_order_relaxed); + } + void OnBackgroundError(rocksdb::BackgroundErrorReason, + rocksdb::Status* bg_error) override { + if (bg_error && !bg_error->ok()) { + background_errors.fetch_add(1, std::memory_order_relaxed); + } + } + void OnTableFileCreated(const rocksdb::TableFileCreationInfo& info) override { + table_created_bytes.fetch_add(static_cast(info.file_size), + std::memory_order_relaxed); + } + + void Reset() { + flush_output_bytes.store(0, std::memory_order_relaxed); + compaction_output_bytes.store(0, std::memory_order_relaxed); + table_created_bytes.store(0, std::memory_order_relaxed); + compaction_events.store(0, std::memory_order_relaxed); + intra_l0_compactions.store(0, std::memory_order_relaxed); + failed_compactions.store(0, std::memory_order_relaxed); + background_errors.store(0, std::memory_order_relaxed); + } +}; + +// Workload profiles + +struct Profile { + std::string name; + uint64_t U; + uint64_t p0, p1, p2, p3, p4, p5_reads, p5_scans, window; + size_t wbuf_mib, tfs_mib, lbase_mib, maxcomp_mib, cache_mib; + int l0_trigger, l0_slow, l0_stop, lmult, bg_jobs, num_levels, num_cfs; +}; + +static Profile EmptyProfile() { + return {"", 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}; +} + +static uint64_t ScaleCount(uint64_t value, uint64_t num, uint64_t den) { + return std::max(1, (value * num + den - 1) / den); +} + +static Profile ClampProfile(Profile p) { + p.wbuf_mib = std::max(2, p.wbuf_mib); + p.tfs_mib = std::max(2, p.tfs_mib); + p.lbase_mib = std::max(8, p.lbase_mib); + p.maxcomp_mib = std::max(16, p.maxcomp_mib); + p.cache_mib = std::max(16, p.cache_mib); + p.l0_trigger = std::max(2, p.l0_trigger); + p.l0_slow = std::max(p.l0_trigger + 3, p.l0_slow); + p.l0_stop = std::max(p.l0_slow + 4, p.l0_stop); + p.lmult = std::max(2, p.lmult); + p.bg_jobs = std::max(1, p.bg_jobs); + p.num_levels = std::max(4, p.num_levels); + p.num_cfs = std::max(1, p.num_cfs); + return p; +} + +static Profile DeriveCaseProfile(Profile p, uint64_t seed) { + if (p.name == "smoke") return p; + uint64_t r = Rng(seed, 9, 0, 601); + switch (r % 5) { + case 0: + p.wbuf_mib = std::max(2, p.wbuf_mib / 2); + p.tfs_mib = std::max(2, p.tfs_mib / 2); + p.l0_trigger = std::max(2, p.l0_trigger - 1); + break; + case 1: + p.lbase_mib = std::max(16, p.lbase_mib / 2); + p.maxcomp_mib = std::max(16, p.maxcomp_mib / 2); + p.p5_scans = ScaleCount(p.p5_scans, 5, 4); + break; + case 2: + p.cache_mib = std::max(16, p.cache_mib / 2); + p.p5_reads = ScaleCount(p.p5_reads, 6, 5); + break; + case 3: + p.window = std::max(1024, p.window / 2); + p.p3 = ScaleCount(p.p3, 7, 6); + p.p4 = ScaleCount(p.p4, 7, 6); + break; + case 4: + p.bg_jobs = std::max(1, p.bg_jobs - 1); + p.p2 = ScaleCount(p.p2, 6, 5); + break; + } + uint64_t skew = 90 + ((r >> 8) % 21); // 0.90x .. 1.10x + p.p1 = ScaleCount(p.p1, skew, 100); + p.p2 = ScaleCount(p.p2, 200 - skew, 100); + p.p4 = ScaleCount(p.p4, 95 + ((r >> 16) % 16), 100); + return ClampProfile(p); +} + +static Profile GetProfile(const std::string& name, bool* ok = nullptr) { + if (ok) *ok = true; + if (name == "full") + return {"full", 524288, 70000, 60000, 85000, 70000, 80000, 25000, + 2500, 4096, 4, 4, 16, 64, 48, 3, + 9, 18, 6, 2, 6, 1}; + if (name == "smoke") + return {"smoke", 262144, 40000, 40000, 70000, 50000, 60000, 20000, + 1000, 2048, 4, 4, 16, 64, 32, 4, + 10, 20, 4, 2, 6, 1}; + if (name == "l0_pressure") + return {"l0_pressure", + 1048576, + 120000, + 180000, + 300000, + 160000, + 180000, + 50000, + 3000, + 4096, + 4, + 4, + 16, + 64, + 64, + 3, + 8, + 16, + 4, + 2, + 6, + 1}; + if (name == "range_snapshot") + return {"range_snapshot", + 1048576, + 140000, + 120000, + 320000, + 160000, + 220000, + 70000, + 3500, + 8192, + 8, + 8, + 32, + 128, + 64, + 4, + 10, + 20, + 8, + 3, + 6, + 1}; + if (name == "scanmix") + return {"scanmix", 1572864, 180000, 120000, 180000, 220000, 300000, 140000, + 5000, 16384, 8, 8, 32, 128, 96, 4, + 10, 20, 8, 3, 6, 1}; + if (name == "multi_cf") + return {"multi_cf", 524288, 90000, 120000, 170000, 130000, 160000, 60000, + 2500, 4096, 4, 4, 16, 64, 64, 4, + 10, 20, 4, 2, 6, 3}; + if (name == "time_series") + return {"time_series", + 2097152, + 200000, + 180000, + 240000, + 220000, + 240000, + 80000, + 6000, + 8192, + 8, + 8, + 32, + 128, + 64, + 4, + 10, + 20, + 8, + 3, + 6, + 1}; + if (name == "difficulty") + return { + "difficulty", 1048576, 160000, 140000, 260000, 180000, 220000, 80000, + 4000, 8192, 4, 4, 16, 64, 64, 4, + 10, 20, 4, 2, 6, 1}; + if (name == "overlap_rewrite") + return {"overlap_rewrite", + 1048576, + 160000, + 144000, + 144000, + 120000, + 120000, + 60000, + 25000, + 4096, + 8, + 4, + 16, + 64, + 64, + 4, + 10, + 20, + 4, + 1, + 6, + 1}; + if (ok) *ok = false; + return EmptyProfile(); +} + +static bool IsHiddenOrHardProfile(const std::string& name) { + return name == "difficulty" || name == "full"; +} + +static rocksdb::Options MakeOptions(const Profile& pf, + std::shared_ptr listener) { + rocksdb::Options o; + o.create_if_missing = true; + o.compaction_style = rocksdb::kCompactionStyleLevel; + o.compression = rocksdb::kNoCompression; + o.num_levels = pf.num_levels; + o.write_buffer_size = pf.wbuf_mib << 20; + o.max_write_buffer_number = 4; + o.min_write_buffer_number_to_merge = 1; + o.level0_file_num_compaction_trigger = pf.l0_trigger; + o.level0_slowdown_writes_trigger = pf.l0_slow; + o.level0_stop_writes_trigger = pf.l0_stop; + o.target_file_size_base = pf.tfs_mib << 20; + o.target_file_size_multiplier = 1; + o.max_bytes_for_level_base = pf.lbase_mib << 20; + o.max_bytes_for_level_multiplier = pf.lmult; + o.max_compaction_bytes = pf.maxcomp_mib << 20; + o.max_background_jobs = 1; + o.max_subcompactions = 1; + o.use_direct_reads = false; + o.use_direct_io_for_flush_and_compaction = false; + o.statistics = rocksdb::CreateDBStatistics(); + if (listener) o.listeners.emplace_back(listener); + rocksdb::BlockBasedTableOptions topt; + topt.block_cache = rocksdb::NewLRUCache(pf.cache_mib << 20); + topt.read_amp_bytes_per_bit = 16; + o.table_factory.reset(rocksdb::NewBlockBasedTableFactory(topt)); + return o; +} + +struct DataDigest { + uint64_t h1 = 1469598103934665603ULL; + uint64_t h2 = 1099511628211ULL ^ 0xff51afd7ed558ccdULL; + uint64_t entries = 0; + uint64_t bytes = 0; + bool ok = true; +}; + +static void DigestBytes(DataDigest* digest, const char* data, size_t size) { + const uint64_t encoded_size = static_cast(size); + const char* size_bytes = reinterpret_cast(&encoded_size); + for (size_t i = 0; i < sizeof(encoded_size); ++i) { + const unsigned char c = static_cast(size_bytes[i]); + digest->h1 = (digest->h1 ^ c) * 1099511628211ULL; + digest->h2 = (digest->h2 ^ c) * 0x100000001b3ULL; + digest->h2 = (digest->h2 << 7) | (digest->h2 >> 57); + } + for (size_t i = 0; i < size; ++i) { + const unsigned char c = static_cast(data[i]); + digest->h1 = (digest->h1 ^ c) * 1099511628211ULL; + digest->h2 = (digest->h2 ^ c) * 0x100000001b3ULL; + digest->h2 = (digest->h2 << 7) | (digest->h2 >> 57); + } +} + +static DataDigest CollectDataDigest( + rocksdb::DB* db, const std::vector& handles, + const std::vector& cf_names) { + DataDigest digest; + rocksdb::ReadOptions read_options; + for (size_t cf = 0; cf < handles.size(); ++cf) { + DigestBytes(&digest, cf_names[cf].data(), cf_names[cf].size()); + std::unique_ptr it( + db->NewIterator(read_options, handles[cf])); + for (it->SeekToFirst(); it->Valid(); it->Next()) { + DigestBytes(&digest, it->key().data(), it->key().size()); + DigestBytes(&digest, it->value().data(), it->value().size()); + ++digest.entries; + digest.bytes += it->key().size() + it->value().size(); + } + if (!it->status().ok()) digest.ok = false; + } + return digest; +} + +static int RunTrustedResidual(const std::string& db_dir, + const std::string& out_path, const Profile& pf) { + auto write_failure = [&](const std::string& reason) { + std::ofstream out(out_path); + out << "{\"ok\":false,\"fail_reason\":\"" << JsonEscape(reason) << "\"}\n"; + }; + + auto listener = std::make_shared(); + rocksdb::Options options = MakeOptions(pf, listener); + options.create_if_missing = false; + options.disable_auto_compactions = true; + std::vector cf_names; + rocksdb::Status status = rocksdb::DB::ListColumnFamilies( + rocksdb::DBOptions(options), db_dir, &cf_names); + if (!status.ok() || cf_names.empty()) { + write_failure("list-column-families: " + status.ToString()); + return 0; + } + + std::vector descriptors; + descriptors.reserve(cf_names.size()); + for (const auto& name : cf_names) { + descriptors.emplace_back(name, rocksdb::ColumnFamilyOptions(options)); + } + std::vector handles; + rocksdb::DB* db = nullptr; +#if ROCKSDB_VERSION_GE(11, 0, 0) + std::unique_ptr db_owner; + status = rocksdb::DB::Open(rocksdb::DBOptions(options), db_dir, descriptors, + &handles, &db_owner); + db = db_owner.get(); +#else + status = rocksdb::DB::Open(rocksdb::DBOptions(options), db_dir, descriptors, + &handles, &db); +#endif + if (!status.ok()) { + write_failure("open-trusted-drain: " + status.ToString()); + return 0; + } + + bool ok = true; + std::string fail_reason; + const DataDigest before_digest = CollectDataDigest(db, handles, cf_names); + if (!before_digest.ok) { + ok = false; + fail_reason = "pre-residual-iterator-failed"; + } + + bool converged = false; + int passes = 0; + uint64_t previous_output = 0; + for (int pass = 0; pass < 3 && ok; ++pass) { + status = db->EnableAutoCompaction(handles); + if (!status.ok()) { + ok = false; + fail_reason = "enable-auto-compaction: " + status.ToString(); + break; + } + rocksdb::WaitForCompactOptions wait; + wait.flush = false; + status = db->WaitForCompact(wait); + if (!status.ok()) { + ok = false; + fail_reason = "wait-trusted-residual: " + status.ToString(); + break; + } + for (auto* handle : handles) { + status = db->SetOptions(handle, {{"disable_auto_compactions", "true"}}); + if (!status.ok()) { + ok = false; + fail_reason = "disable-auto-compaction: " + status.ToString(); + break; + } + } + if (!ok) break; + ++passes; + const uint64_t current_output = + listener->compaction_output_bytes.load(std::memory_order_relaxed); + if (current_output == previous_output) { + converged = true; + break; + } + previous_output = current_output; + } + if (ok && !converged) { + ok = false; + fail_reason = "trusted-residual-did-not-converge"; + } + + uint64_t l0_files = 0; + uint64_t upper_level_files = 0; + uint64_t upper_level_bytes = 0; + uint64_t final_files = 0; + std::vector metadata; + db->GetAllColumnFamilyMetaData(&metadata); + for (const auto& cf : metadata) { + for (const auto& level : cf.levels) { + final_files += level.files.size(); + if (level.level == 0) l0_files += level.files.size(); + if (level.level < pf.num_levels - 1) { + upper_level_files += level.files.size(); + upper_level_bytes += level.size; + } + } + } + const uint64_t compaction_bytes = + listener->compaction_output_bytes.load(std::memory_order_relaxed); + const uint64_t failed_compactions = + listener->failed_compactions.load(std::memory_order_relaxed); + const uint64_t background_errors = + listener->background_errors.load(std::memory_order_relaxed); + const DataDigest after_digest = CollectDataDigest(db, handles, cf_names); + if (ok && (!after_digest.ok || before_digest.h1 != after_digest.h1 || + before_digest.h2 != after_digest.h2 || + before_digest.entries != after_digest.entries || + before_digest.bytes != after_digest.bytes)) { + ok = false; + fail_reason = "trusted-residual-changed-logical-data"; + } + + for (auto* handle : handles) db->DestroyColumnFamilyHandle(handle); +#if ROCKSDB_VERSION_GE(11, 0, 0) + db_owner.reset(); +#else + delete db; +#endif + + uint64_t table_bytes = 0; + std::error_code walk_ec; + for (const auto& entry : fs::directory_iterator(db_dir, walk_ec)) { + if (walk_ec) break; + if (!entry.is_regular_file(walk_ec) || walk_ec) continue; + const auto extension = entry.path().extension(); + if (extension != ".sst" && extension != ".blob") continue; + table_bytes += static_cast(entry.file_size(walk_ec)); + if (walk_ec) break; + } + if (walk_ec && ok) { + ok = false; + fail_reason = "table-file-size: " + walk_ec.message(); + } + if ((failed_compactions != 0 || background_errors != 0) && ok) { + ok = false; + fail_reason = "trusted-drain-background-error"; + } + + std::ofstream out(out_path); + out << "{\n \"ok\": " << (ok ? "true" : "false") << ",\n"; + out << " \"fail_reason\": \"" << JsonEscape(fail_reason) << "\",\n"; + out << " \"compaction_output_bytes\": " << compaction_bytes << ",\n"; + out << " \"final_table_bytes\": " << table_bytes << ",\n"; + out << " \"final_files\": " << final_files << ",\n"; + out << " \"l0_files\": " << l0_files << ",\n"; + out << " \"upper_level_files\": " << upper_level_files << ",\n"; + out << " \"upper_level_bytes\": " << upper_level_bytes << ",\n"; + out << " \"passes\": " << passes << ",\n"; + out << " \"digest_entries\": " << after_digest.entries << ",\n"; + out << " \"digest_bytes\": " << after_digest.bytes << "\n}\n"; + return 0; +} + +// Workload sampling + +static uint64_t MovingHotCenter(uint64_t seed, uint32_t phase, + uint64_t op_index, uint64_t U, + uint64_t window) { + const uint64_t period = 50000; + uint64_t steps = op_index / period; + uint64_t center = Rng(seed, phase, 0, 99) % U; + uint64_t span = std::max(window, U / 64); + for (uint64_t i = 0; i < steps; ++i) { + int64_t delta = + int64_t(Rng(seed, phase, i, 100) % (2 * span)) - int64_t(span); + int64_t c = (int64_t(center) + delta) % int64_t(U); + if (c < 0) c += int64_t(U); + center = uint64_t(c); + } + return center; +} +static uint64_t SampleHot(uint64_t seed, uint32_t phase, uint64_t op_index, + uint64_t center, uint64_t window, uint64_t U) { + uint64_t off = Rng(seed, phase, op_index, 10) % std::max(1, window); + uint64_t start = (center + U - window / 2) % U; + return (start + off) % U; +} +static uint64_t SampleHeavyTail(uint64_t seed, uint32_t phase, + uint64_t op_index, uint64_t U) { + uint64_t r1 = Rng(seed, phase, op_index, 20) % U; + uint64_t r2 = Rng(seed, phase, op_index, 21) % U; + uint64_t r3 = Rng(seed, phase, op_index, 22) % U; + return std::min(r1, std::min(r2, r3)); +} +static size_t PickValueSize(uint64_t seed, uint32_t phase, uint64_t op_index, + const std::vector>& mix) { + int r = int(Rng(seed, phase, op_index, 30) % 100), acc = 0; + for (auto& m : mix) { + acc += m.first; + if (r < acc) return m.second; + } + return mix.back().second; +} + +struct OracleEntry { + uint32_t size = 0; + uint64_t h1 = 0, h2 = 0; + bool present = false; +}; + +struct SnapshotSample { + size_t cf = 0; + uint64_t key_id = 0; + bool present = false; + uint32_t size = 0; + uint64_t h1 = 0, h2 = 0; +}; + +struct LsmMetrics { + uint64_t pending = 0; + uint64_t lsm_bytes = 0; + uint64_t l0_base_overlap_bytes = 0; + uint64_t active_memtable_bytes = 0; + uint64_t active_memtable_entries = 0; + uint64_t immutable_memtables = 0; + uint64_t fingerprint = 1469598103934665603ULL; + uint64_t level_bytes[6] = {0, 0, 0, 0, 0, 0}; + uint64_t level_files[6] = {0, 0, 0, 0, 0, 0}; + std::vector l0_files_by_cf; + uint64_t column_families = 0; +}; + +static void LoadCaseFile(const std::string& path, uint64_t* seed, + std::string* profile_name) { + std::ifstream in(path); + std::string line; + while (std::getline(in, line)) { + const size_t eq = line.find('='); + if (eq == std::string::npos) continue; + const std::string key = line.substr(0, eq); + const std::string value = line.substr(eq + 1); + if (key == "seed") + *seed = std::stoull(value); + else if (key == "profile") + *profile_name = value; + } +} + +int main(int argc, char** argv) { + std::string db_dir, out_path, case_file, profile_name = "difficulty"; + uint64_t seed = 1; + bool trusted_drain = false; + for (int i = 1; i < argc; ++i) { + std::string a = argv[i]; + auto nxt = [&]() -> std::string { return (i + 1 < argc) ? argv[++i] : ""; }; + if (a == "--db") + db_dir = nxt(); + else if (a == "--seed") + seed = std::stoull(nxt()); + else if (a == "--profile") + profile_name = nxt(); + else if (a == "--out") + out_path = nxt(); + else if (a == "--case-file") + case_file = nxt(); + else if (a == "--trusted-residual") + trusted_drain = true; + } + if (db_dir.empty() || out_path.empty()) { + std::fprintf(stderr, "usage: --db DIR --out JSON [--case-file PATH]\n"); + return 2; + } + if (!case_file.empty()) { + LoadCaseFile(case_file, &seed, &profile_name); + std::remove(case_file.c_str()); + } + + bool profile_ok = false; + const Profile base_pf = GetProfile(profile_name, &profile_ok); + if (!profile_ok) { + std::ofstream o(out_path); + o << "{\"ok\":false,\"fail_reason\":\"unknown-profile: " + << JsonEscape(profile_name) << "\"}\n"; + return 0; + } + const Profile pf = DeriveCaseProfile(base_pf, seed); + if (trusted_drain) return RunTrustedResidual(db_dir, out_path, pf); + const uint64_t U = pf.U; + const bool hidden_or_hard_profile = IsHiddenOrHardProfile(profile_name); + const uint64_t mix_bits = hidden_or_hard_profile ? Rng(seed, 8, 0, 713) : 0; + const bool mixed_range_snapshot = + hidden_or_hard_profile && ((mix_bits % 3) != 1); + const bool mixed_scan = hidden_or_hard_profile && (((mix_bits / 3) % 3) != 1); + const bool mixed_time_series = + hidden_or_hard_profile && + ((((mix_bits / 9) % 4) == 0) || profile_name == "full"); + const bool snapshot_profile = + profile_name == "range_snapshot" || profile_name == "multi_cf" || + profile_name == "full" || mixed_range_snapshot || + (hidden_or_hard_profile && pf.num_cfs > 1 && ((mix_bits >> 5) & 1)); + const bool range_profile = profile_name == "range_snapshot" || + profile_name == "full" || mixed_range_snapshot; + const bool scan_profile = + profile_name == "scanmix" || profile_name == "full" || mixed_scan; + const bool time_series_profile = profile_name == "time_series" || + profile_name == "full" || mixed_time_series; + const bool overlap_rewrite_profile = profile_name == "overlap_rewrite"; + + std::error_code ec; + fs::remove_all(db_dir, ec); + fs::create_directories(db_dir, ec); + + auto listener = std::make_shared(); + rocksdb::Options options = MakeOptions(pf, listener); + rocksdb::DB* db = nullptr; +#if ROCKSDB_VERSION_GE(11, 0, 0) + std::unique_ptr db_owner; + rocksdb::Status s = rocksdb::DB::Open(options, db_dir, &db_owner); + db = db_owner.get(); +#else + rocksdb::Status s = rocksdb::DB::Open(options, db_dir, &db); +#endif + if (!s.ok()) { + std::ofstream o(out_path); + o << "{\"ok\":false,\"fail_reason\":\"open: " << JsonEscape(s.ToString()) + << "\"}\n"; + return 0; + } + std::vector cf_handles; + std::vector owned_cf_handles; + cf_handles.push_back(db->DefaultColumnFamily()); + rocksdb::ColumnFamilyOptions cf_opts(options); + for (int i = 1; i < pf.num_cfs; ++i) { + rocksdb::ColumnFamilyHandle* h = nullptr; + rocksdb::Status cfs = + db->CreateColumnFamily(cf_opts, "cf" + std::to_string(i), &h); + if (!cfs.ok()) { + std::ofstream o(out_path); + o << "{\"ok\":false,\"fail_reason\":\"create-cf: " + << JsonEscape(cfs.ToString()) << "\"}\n"; +#if ROCKSDB_VERSION_GE(11, 0, 0) + db_owner.reset(); +#else + delete db; +#endif + return 0; + } + cf_handles.push_back(h); + owned_cf_handles.push_back(h); + } + std::vector> oracle(cf_handles.size(), + std::vector(U)); + + rocksdb::WriteOptions wo; + wo.disableWAL = + true; // measuring compaction policy; clean close before reopen + rocksdb::ReadOptions ro; + + std::vector get_lat_us, scan_lat_us; + uint64_t user_put_bytes = 0; + int64_t live = 0; + bool correctness_ok = true; + std::string fail_reason; + const rocksdb::Snapshot* held_snapshot = nullptr; + std::vector snapshot_samples; + std::vector> modified_since_snapshot; + + auto fail = [&](const std::string& reason) { + if (correctness_ok) { + correctness_ok = false; + fail_reason = reason; + } + }; + auto check_background_errors = [&](const std::string& where) { + uint64_t bg_count = 0; + if (!db->GetIntProperty(rocksdb::DB::Properties::kBackgroundErrors, + &bg_count)) { + fail("property-background-errors: " + where); + } else if (bg_count > 0) { + fail("background-errors: " + where + ": " + std::to_string(bg_count)); + } + uint64_t listener_bg = + listener->background_errors.load(std::memory_order_relaxed); + if (listener_bg > 0) { + fail("listener-background-error: " + where + ": " + + std::to_string(listener_bg)); + } + uint64_t failed_compactions = + listener->failed_compactions.load(std::memory_order_relaxed); + if (failed_compactions > 0) { + fail("compaction-error: " + where + ": " + + std::to_string(failed_compactions)); + } + }; + auto set_auto_compactions = [&](bool enabled, const std::string& where) { + for (auto* h : cf_handles) { + rocksdb::Status changed; + if (enabled) { + changed = db->SetOptions( + h, {{"level0_slowdown_writes_trigger", std::to_string(pf.l0_slow)}, + {"level0_stop_writes_trigger", std::to_string(pf.l0_stop)}}); + } else { + changed = + db->SetOptions(h, {{"disable_auto_compactions", "true"}, + {"level0_slowdown_writes_trigger", "1000000"}, + {"level0_stop_writes_trigger", "1000000"}}); + } + if (!changed.ok()) { + fail("set-compaction-mode-" + where + ": " + changed.ToString()); + } + } + if (enabled) { + rocksdb::Status status = db->EnableAutoCompaction(cf_handles); + if (!status.ok()) { + fail("enable-auto-compaction-" + where + ": " + status.ToString()); + } + } + }; + auto flush_memtables = [&](const std::string& where) { + rocksdb::FlushOptions flush; + flush.wait = true; + for (auto* h : cf_handles) { + rocksdb::Status status = db->Flush(flush, h); + if (!status.ok()) fail("flush-" + where + ": " + status.ToString()); + } + }; + auto run_compaction_cycle = [&](const std::string& where) { + flush_memtables(where); + set_auto_compactions(true, where); + rocksdb::WaitForCompactOptions wait; + wait.flush = true; + rocksdb::Status status = db->WaitForCompact(wait); + if (!status.ok()) fail("wait-" + where + ": " + status.ToString()); + check_background_errors(where); + set_auto_compactions(false, where); + }; + auto cf_for = [&](uint32_t ph, uint64_t op, uint64_t kid) -> size_t { + if (cf_handles.size() == 1) return 0; + return size_t(Rng(seed, ph, op ^ (kid * 1315423911ULL), 77) % + cf_handles.size()); + }; + + auto put = [&](uint32_t ph, uint64_t op, uint64_t kid, size_t vsz) { + size_t cf = cf_for(ph, op, kid); + std::string k = MakeKey(kid, seed), v = MakeValue(seed, ph, op, kid, vsz); + rocksdb::Status st = db->Put(wo, cf_handles[cf], k, v); + if (!st.ok()) { + fail("put: " + st.ToString()); + return; + } + auto& e = oracle[cf][kid]; + if (e.present) + live += int64_t(vsz) - int64_t(e.size); + else + live += int64_t(16 + vsz); + auto h = Hash128(v); + e.present = true; + e.size = uint32_t(vsz); + e.h1 = h.first; + e.h2 = h.second; + if (!modified_since_snapshot.empty()) modified_since_snapshot[cf][kid] = 1; + user_put_bytes += k.size() + v.size(); + }; + auto del = [&](uint32_t ph, uint64_t op, uint64_t kid) { + size_t cf = cf_for(ph, op, kid); + rocksdb::Status st = db->Delete(wo, cf_handles[cf], MakeKey(kid, seed)); + if (!st.ok()) { + fail("delete: " + st.ToString()); + return; + } + auto& e = oracle[cf][kid]; + if (e.present) live -= int64_t(16 + e.size); + e.present = false; + e.size = 0; + if (!modified_since_snapshot.empty()) modified_since_snapshot[cf][kid] = 1; + }; + auto delrange = [&](uint32_t ph, uint64_t op, uint64_t b, uint64_t e2) { + if (e2 > U) e2 = U; + if (b >= e2) return; + size_t cf = cf_for(ph, op, b); + rocksdb::Status st = db->DeleteRange(wo, cf_handles[cf], MakeKey(b, seed), + MakeKey(e2, seed)); + if (!st.ok()) { + fail("delete-range: " + st.ToString()); + return; + } + size_t targeted = 0; + if (held_snapshot && !modified_since_snapshot.empty()) { + for (uint64_t k = b; + k < e2 && targeted < 8 && snapshot_samples.size() < 8192; ++k) { + const auto& e = oracle[cf][k]; + if (!modified_since_snapshot[cf][k] && e.present) { + SnapshotSample sample; + sample.cf = cf; + sample.key_id = k; + sample.present = true; + sample.size = e.size; + sample.h1 = e.h1; + sample.h2 = e.h2; + snapshot_samples.push_back(sample); + ++targeted; + } + } + } + for (uint64_t k = b; k < e2; ++k) { + auto& e = oracle[cf][k]; + if (e.present) live -= int64_t(16 + e.size); + e.present = false; + e.size = 0; + if (!modified_since_snapshot.empty()) modified_since_snapshot[cf][k] = 1; + } + }; + auto timed_get = [&](uint32_t ph, uint64_t op, uint64_t kid) { + size_t cf = cf_for(ph, op, kid); + std::string got; + auto t0 = Clock::now(); + rocksdb::Status st = db->Get(ro, cf_handles[cf], MakeKey(kid, seed), &got); + get_lat_us.push_back( + std::chrono::duration(Clock::now() - t0).count()); + if (!st.ok() && !st.IsNotFound()) fail("get: " + st.ToString()); + }; + auto timed_scan = [&](uint32_t ph, uint64_t op, uint64_t start, int len) { + size_t cf = cf_for(ph, op, start); + auto t0 = Clock::now(); + std::unique_ptr it(db->NewIterator(ro, cf_handles[cf])); + it->Seek(MakeKey(start, seed)); + int c = 0; + while (it->Valid() && c < len) { + it->Next(); + ++c; + } + scan_lat_us.push_back( + std::chrono::duration(Clock::now() - t0).count()); + if (!it->status().ok()) fail("scan-iter: " + it->status().ToString()); + }; + + std::vector> mix1 = {{75, 400}, {20, 1024}, {5, 4096}}; + std::vector> mix2 = {{90, 128}, {10, 2048}}; + std::vector> mix4 = {{80, 128}, {15, 1024}, {5, 4096}}; + + double ops_per_sec[5] = {0, 0, 0, 0, 0}; + auto capture_snapshot = [&]() { + held_snapshot = db->GetSnapshot(); + snapshot_samples.clear(); + modified_since_snapshot.assign(cf_handles.size(), + std::vector(U, 0)); + for (size_t cf = 0; cf < cf_handles.size(); ++cf) { + for (uint64_t i = 0; i < 256; ++i) { + uint64_t kid = Rng(seed, 6, i + cf * 1000, 44) % U; + const auto& e = oracle[cf][kid]; + SnapshotSample sample; + sample.cf = cf; + sample.key_id = kid; + sample.present = e.present; + sample.size = e.size; + sample.h1 = e.h1; + sample.h2 = e.h2; + snapshot_samples.push_back(sample); + } + } + }; + + auto verify_snapshot = [&](bool release_after) { + if (!held_snapshot) return; + rocksdb::ReadOptions sro = ro; + sro.snapshot = held_snapshot; + for (const auto& sample : snapshot_samples) { + std::string got; + rocksdb::Status st = db->Get(sro, cf_handles[sample.cf], + MakeKey(sample.key_id, seed), &got); + if (sample.present) { + auto h = Hash128(got); + if (!st.ok() || got.size() != sample.size || h.first != sample.h1 || + h.second != sample.h2) { + fail("snapshot-mismatch present"); + break; + } + } else if (!st.IsNotFound()) { + fail("snapshot-mismatch deleted"); + break; + } + } + if (release_after) { + db->ReleaseSnapshot(held_snapshot); + held_snapshot = nullptr; + } + }; + + auto collect_lsm = [&]() { + LsmMetrics m; + for (auto* h : cf_handles) { + uint64_t value = 0; + if (!db->GetIntProperty(h, "rocksdb.estimate-pending-compaction-bytes", + &value)) + fail("property-pending"); + m.pending += value; + value = 0; + if (!db->GetIntProperty(h, "rocksdb.cur-size-active-mem-table", &value)) + fail("property-active-memtable"); + m.active_memtable_bytes += value; + value = 0; + if (!db->GetIntProperty(h, "rocksdb.num-entries-active-mem-table", + &value)) + fail("property-active-memtable-entries"); + m.active_memtable_entries += value; + value = 0; + if (!db->GetIntProperty(h, "rocksdb.num-immutable-mem-table", &value)) + fail("property-immutable-memtable"); + m.immutable_memtables += value; + } + + std::vector metadata; + db->GetAllColumnFamilyMetaData(&metadata); + m.column_families = metadata.size(); + if (metadata.size() != cf_handles.size()) { + fail("metadata-column-family-count"); + } + for (const auto& cf : metadata) { + auto mix_u64 = [&m](uint64_t value) { + m.fingerprint ^= value + 0x9e3779b97f4a7c15ULL + (m.fingerprint << 6) + + (m.fingerprint >> 2); + }; + auto mix_string = [&m](const std::string& value) { + for (unsigned char c : value) { + m.fingerprint = (m.fingerprint ^ c) * 1099511628211ULL; + } + }; + mix_string(cf.name); + m.lsm_bytes += cf.size + cf.blob_file_size; + const rocksdb::LevelMetaData* l0 = nullptr; + const rocksdb::LevelMetaData* base = nullptr; + for (const auto& level : cf.levels) { + if (level.level < 0 || level.level >= 6) continue; + mix_u64(static_cast(level.level)); + mix_u64(level.size); + m.level_bytes[level.level] += level.size; + m.level_files[level.level] += level.files.size(); + for (const auto& file : level.files) { + mix_u64(file.size); + mix_string(file.smallestkey); + mix_string(file.largestkey); + } + if (level.level == 0) l0 = &level; + if (level.level > 0 && base == nullptr && !level.files.empty()) { + base = &level; + } + } + m.l0_files_by_cf.push_back(l0 == nullptr ? 0 : l0->files.size()); + if (l0 != nullptr && base != nullptr) { + auto user_key = [](const std::string& internal_key) { + return internal_key.substr(0, internal_key.size() >= 8 + ? internal_key.size() - 8 + : internal_key.size()); + }; + for (const auto& base_file : base->files) { + const std::string base_smallest = user_key(base_file.smallestkey); + const std::string base_largest = user_key(base_file.largestkey); + for (const auto& l0_file : l0->files) { + const std::string l0_smallest = user_key(l0_file.smallestkey); + const std::string l0_largest = user_key(l0_file.largestkey); + if (base_smallest <= l0_largest && l0_smallest <= base_largest) { + m.l0_base_overlap_bytes += base_file.size; + break; + } + } + } + } + } + return m; + }; + + auto same_lsm = [](const LsmMetrics& a, const LsmMetrics& b) { + if (a.pending != b.pending || a.lsm_bytes != b.lsm_bytes || + a.l0_base_overlap_bytes != b.l0_base_overlap_bytes || + a.active_memtable_bytes != b.active_memtable_bytes || + a.active_memtable_entries != b.active_memtable_entries || + a.immutable_memtables != b.immutable_memtables || + a.fingerprint != b.fingerprint || + a.l0_files_by_cf != b.l0_files_by_cf || + a.column_families != b.column_families) { + return false; + } + for (int level = 0; level < 6; ++level) { + if (a.level_bytes[level] != b.level_bytes[level] || + a.level_files[level] != b.level_files[level]) { + return false; + } + } + return true; + }; + + auto collect_paused_lsm = [&](const std::string& where) { + LsmMetrics first; + rocksdb::Status paused = db->PauseBackgroundWork(); + if (!paused.ok()) { + fail("pause-background-" + where + ": " + paused.ToString()); + return first; + } + first = collect_lsm(); + const LsmMetrics second = collect_lsm(); + if (!same_lsm(first, second)) { + fail("unstable-metadata-" + where); + } + rocksdb::Status resumed = db->ContinueBackgroundWork(); + if (!resumed.ok()) { + fail("resume-background-" + where + ": " + resumed.ToString()); + } + return second; + }; + + auto percentile_copy = [](const std::vector& xs, double p) { + std::vector copy(xs); + return Percentile(copy, p); + }; + + auto table_file_bytes = [&]() { + uint64_t bytes = 0; + std::error_code walk_ec; + for (const auto& entry : fs::directory_iterator(db_dir, walk_ec)) { + if (walk_ec) break; + if (!entry.is_regular_file(walk_ec) || walk_ec) continue; + const auto extension = entry.path().extension(); + if (extension != ".sst" && extension != ".blob") continue; + bytes += static_cast(entry.file_size(walk_ec)); + if (walk_ec) break; + } + if (walk_ec) fail("table-file-size: " + walk_ec.message()); + return bytes; + }; + + uint64_t base_fingerprint = 0; + // Phase 0: build a quiescent starting LSM, then reset all scored counters. + { + set_auto_compactions(false, "phase0"); + if (overlap_rewrite_profile) { + const uint64_t band_width = U / 4; + for (uint64_t op = 0; op < pf.p0; ++op) { + const uint64_t draw = Rng(seed, 0, op, 101) % 100; + const uint64_t band = draw < 50 ? 0 : 1 + (Rng(seed, 0, op, 102) % 3); + const uint64_t kid = + band * band_width + (Rng(seed, 0, op, 103) % band_width); + put(0, op, kid, band == 0 ? 1024 : 128); + } + } else { + uint64_t step = 2654435761ULL % U; + if (step == 0) step = 1; + uint64_t cur = Rng(seed, 0, 0, 1) % U; + for (uint64_t op = 0; op < pf.p0; ++op) { + cur = (cur + step) % U; + put(0, op, cur, 400); + } + } + rocksdb::FlushOptions fo; + fo.wait = true; + for (auto* h : cf_handles) { + rocksdb::Status fst = db->Flush(fo, h); + if (!fst.ok()) fail("flush-phase0: " + fst.ToString()); + } + rocksdb::CompactRangeOptions compact; + compact.change_level = true; + compact.target_level = pf.num_levels - 1; + for (auto* h : cf_handles) { + rocksdb::Status cst = db->CompactRange(compact, h, nullptr, nullptr); + if (!cst.ok()) fail("compact-phase0: " + cst.ToString()); + } + rocksdb::WaitForCompactOptions wfc; + wfc.flush = true; + rocksdb::Status wst = db->WaitForCompact(wfc); + if (!wst.ok()) fail("wait-phase0: " + wst.ToString()); + check_background_errors("phase0"); + const LsmMetrics base = collect_paused_lsm("phase0"); + base_fingerprint = base.fingerprint; + uint64_t base_files = 0; + for (int level = 1; level < 6; ++level) { + base_files += base.level_files[level]; + } + if (base_files == 0) fail("phase0-base-level-empty"); + listener->Reset(); + rocksdb::Status reset = options.statistics->Reset(); + if (!reset.ok()) fail("reset-statistics: " + reset.ToString()); + user_put_bytes = 0; + } + // Phase 1: uniform ingest + mild overwrite. + { + uint32_t ph = 1; + auto t0 = Clock::now(); + const uint64_t append_base = Rng(seed, ph, 0, 62) % U; + for (uint64_t op = 0; op < pf.p1; ++op) { + if (overlap_rewrite_profile) { + const uint64_t batch_size = 12000; + const uint64_t band = ((op / batch_size) + 1) % 4; + const uint64_t band_width = U / 4; + const uint64_t kid = + band * band_width + (Rng(seed, ph, op, 104) % band_width); + put(ph, op, kid, 400); + if ((op + 1) % batch_size == 0) { + flush_memtables("phase1-band"); + } + continue; + } + uint64_t kid = + time_series_profile + ? ((append_base + op + (Rng(seed, ph, op, 63) % 64)) % U) + : (Rng(seed, ph, op, 2) % U); + put(ph, op, kid, + PickValueSize(seed, ph, op, time_series_profile ? mix4 : mix1)); + } + ops_per_sec[1] = + double(pf.p1) / + std::max(1e-6, + std::chrono::duration(Clock::now() - t0).count()); + check_background_errors("phase1"); + } + run_compaction_cycle("phase1"); + // Phase 2: moving hot window. + { + uint32_t ph = 2; + auto t0 = Clock::now(); + for (uint64_t op = 0; op < pf.p2; ++op) { + if (overlap_rewrite_profile) { + const uint64_t batch_size = 12000; + const uint64_t band = ((op / batch_size) + 2) % 4; + const uint64_t band_width = U / 4; + const uint64_t kid = + band * band_width + (Rng(seed, ph, op, 105) % band_width); + put(ph, op, kid, 400); + if ((op + 1) % batch_size == 0) { + flush_memtables("phase2-band"); + } + continue; + } + if (time_series_profile) { + uint64_t tail = (pf.p0 + op * 3 + (Rng(seed, ph, op, 61) % 256)) % U; + uint64_t cl_ts = Rng(seed, ph, op, 62) % 100; + if (cl_ts < 78) { + put(ph, op, tail, PickValueSize(seed, ph, op, mix4)); + } else if (cl_ts < 93) { + timed_get(ph, op, + Rng(seed, ph, op, 63) % std::max(1, U / 8)); + } else { + timed_scan(ph, op, + Rng(seed, ph, op, 64) % std::max(1, U / 6), + scan_profile ? 300 : 120); + } + continue; + } + uint64_t center = MovingHotCenter(seed, ph, op, U, pf.window); + uint64_t kid = (Rng(seed, ph, op, 3) % 100 < 80) + ? SampleHot(seed, ph, op, center, pf.window, U) + : SampleHeavyTail(seed, ph, op, U); + uint64_t cl = Rng(seed, ph, op, 4) % 100; + if (cl < 70) + put(ph, op, kid, PickValueSize(seed, ph, op, mix2)); + else if (cl < 90) + timed_get(ph, op, kid); + else + del(ph, op, kid); + } + ops_per_sec[2] = + double(pf.p2) / + std::max(1e-6, + std::chrono::duration(Clock::now() - t0).count()); + check_background_errors("phase2"); + } + run_compaction_cycle("phase2"); + if (snapshot_profile) { + capture_snapshot(); + } + // Phase 3: tombstone burst. + { + uint32_t ph = 3; + auto t0 = Clock::now(); + for (uint64_t op = 0; op < pf.p3; ++op) { + uint64_t kid; + if (Rng(seed, ph, op, 6) % 100 < 70) { + uint64_t past = Rng(seed, ph, op, 7) % std::max(1, pf.p2); + kid = SampleHot(seed, ph, op, + MovingHotCenter(seed, 2, past, U, pf.window), pf.window, + U); + } else + kid = Rng(seed, ph, op, 8) % U; + uint64_t cl = Rng(seed, ph, op, 5) % 100; + if (cl < 45) { + if (Rng(seed, ph, op, 9) % 100 < 90) + del(ph, op, kid); + else { + uint64_t span = range_profile ? 128 + (Rng(seed, ph, op, 11) % 3969) + : 32 + (Rng(seed, ph, op, 11) % 481); + delrange(ph, op, kid, kid + span); + } + } else if (cl < 75) + put(ph, op, kid, 400); + else if (cl < 95) + timed_get(ph, op, kid); + else + timed_scan(ph, op, kid, scan_profile ? 400 : 100); + } + ops_per_sec[3] = + double(pf.p3) / + std::max(1e-6, + std::chrono::duration(Clock::now() - t0).count()); + check_background_errors("phase3"); + } + run_compaction_cycle("phase3"); + // Phase 4: abrupt skew shift. + { + uint32_t ph = 4; + uint64_t new_center = Rng(seed, ph, 0, 50) % U; + auto t0 = Clock::now(); + for (uint64_t op = 0; op < pf.p4; ++op) { + if (time_series_profile) { + uint64_t tail = + (pf.p0 + pf.p1 + pf.p2 + op * 5 + (Rng(seed, ph, op, 71) % 512)) % + U; + uint64_t cl_ts = Rng(seed, ph, op, 72) % 100; + if (cl_ts < 70) { + put(ph, op, tail, PickValueSize(seed, ph, op, mix4)); + } else if (cl_ts < 90) { + timed_get(ph, op, + Rng(seed, ph, op, 73) % std::max(1, U / 5)); + } else { + timed_scan(ph, op, + Rng(seed, ph, op, 74) % std::max(1, U / 4), + scan_profile ? 500 : 160); + } + continue; + } + uint64_t bucket = Rng(seed, ph, op, 12) % 100, kid; + if (bucket < 60) + kid = SampleHot(seed, ph, op, new_center, pf.window, U); + else if (bucket < 85) { + uint64_t past = Rng(seed, ph, op, 13) % std::max(1, pf.p2); + kid = SampleHot(seed, ph, op, + MovingHotCenter(seed, 2, past, U, pf.window), pf.window, + U); + } else + kid = SampleHeavyTail(seed, ph, op, U); + uint64_t cl = Rng(seed, ph, op, 14) % 100; + if (cl < 65) + put(ph, op, kid, PickValueSize(seed, ph, op, mix4)); + else if (cl < 90) + timed_get(ph, op, kid); + else + timed_scan(ph, op, kid, scan_profile ? 500 : 100); + } + ops_per_sec[4] = + double(pf.p4) / + std::max(1e-6, + std::chrono::duration(Clock::now() - t0).count()); + check_background_errors("phase4"); + } + verify_snapshot(false); + check_background_errors("post-phase4"); + + flush_memtables("pre-drain"); + const LsmMetrics pre_lsm = collect_paused_lsm("pre-drain"); + const uint64_t pre_drain_table_bytes = table_file_bytes(); + if (pre_lsm.active_memtable_entries != 0 || + pre_lsm.immutable_memtables != 0) { + fail("pre-drain-memtable-not-empty"); + } + uint64_t live_pre = 0; + for (const auto& cf_oracle : oracle) + for (const auto& e : cf_oracle) + if (e.present) live_pre += 16 + e.size; + double pre_denom = double(std::max(live_pre, 64ull << 20)); + double pre_drain_space_amp = double(pre_drain_table_bytes) / pre_denom; + double pre_drain_get_p99 = percentile_copy(get_lat_us, 0.99); + double pre_drain_scan_p99 = percentile_copy(scan_lat_us, 0.99); + + // Phase 5: drain + read audit. + LsmMetrics final_lsm; + double drain_ms = 0.0; + const uint64_t drain_compaction_before = + listener->compaction_output_bytes.load(std::memory_order_relaxed); + { + auto drain_t0 = Clock::now(); + run_compaction_cycle("final"); + drain_ms = + std::chrono::duration(Clock::now() - drain_t0) + .count(); + verify_snapshot(true); + uint32_t ph = 5; + uint64_t every = std::max(1, pf.p5_reads / 20000); + for (uint64_t i = 0; i < pf.p5_reads; ++i) { + uint64_t kid = (time_series_profile && (Rng(seed, ph, i, 88) % 100 < 65)) + ? (Rng(seed, ph, i, 89) % std::max(1, U / 6)) + : (Rng(seed, ph, i, 1) % U); + size_t cf = cf_for(ph, i, kid); + std::string k = MakeKey(kid, seed), got; + auto t0 = Clock::now(); + rocksdb::Status st = db->Get(ro, cf_handles[cf], k, &got); + get_lat_us.push_back( + std::chrono::duration(Clock::now() - t0).count()); + if (i % every == 0) { + auto& e = oracle[cf][kid]; + if (e.present) { + auto h = Hash128(got); + if (!st.ok() || got.size() != e.size || h.first != e.h1 || + h.second != e.h2) { + fail("get-mismatch present"); + } + } else if (!st.IsNotFound()) { + fail("get-mismatch deleted"); + } + } else if (!st.ok() && !st.IsNotFound()) { + fail("get-audit: " + st.ToString()); + } + } + for (uint64_t i = 0; i < pf.p5_scans; ++i) { + uint64_t start = + (time_series_profile && (Rng(seed, ph, i, 90) % 100 < 70)) + ? (Rng(seed, ph, i, 91) % std::max(1, U / 5)) + : (Rng(seed, ph, i, 2) % U); + timed_scan(ph, i, start, scan_profile ? 500 : 100); + } + } + final_lsm = collect_paused_lsm("final"); + const uint64_t drain_compaction_bytes = + listener->compaction_output_bytes.load(std::memory_order_relaxed) - + drain_compaction_before; + + uint64_t flush_bytes = listener->flush_output_bytes.load(); + uint64_t comp_bytes = listener->compaction_output_bytes.load(); + uint64_t table_bytes = listener->table_created_bytes.load(); + uint64_t read_amp_useful = options.statistics->getTickerCount( + rocksdb::READ_AMP_ESTIMATE_USEFUL_BYTES); + uint64_t read_amp_total = + options.statistics->getTickerCount(rocksdb::READ_AMP_TOTAL_READ_BYTES); + double read_amp = + double(read_amp_total) / double(std::max(1, read_amp_useful)); + uint64_t stall_micros = + options.statistics->getTickerCount(rocksdb::STALL_MICROS); + + uint64_t oracle_count = 0, oracle_size_sum = 0; + for (size_t cf = 0; cf < oracle.size(); ++cf) { + for (uint64_t id = 0; id < U; ++id) { + const auto& e = oracle[cf][id]; + if (!e.present) continue; + ++oracle_count; + oracle_size_sum += e.size; + } + } + + // Reopen + full iterator check against the exact oracle entry for each key. + if (held_snapshot) { + db->ReleaseSnapshot(held_snapshot); + held_snapshot = nullptr; + } + for (auto* h : owned_cf_handles) db->DestroyColumnFamilyHandle(h); + owned_cf_handles.clear(); +#if ROCKSDB_VERSION_GE(11, 0, 0) + db_owner.reset(); +#else + delete db; + db = nullptr; +#endif + db = nullptr; + rocksdb::Options ropts = MakeOptions(pf, nullptr); + std::vector reopen_handles; + rocksdb::Status rs; +#if ROCKSDB_VERSION_GE(11, 0, 0) + std::unique_ptr reopen_owner; +#endif + if (pf.num_cfs > 1) { + std::vector descs; + descs.emplace_back(rocksdb::kDefaultColumnFamilyName, + rocksdb::ColumnFamilyOptions(ropts)); + for (int i = 1; i < pf.num_cfs; ++i) + descs.emplace_back("cf" + std::to_string(i), + rocksdb::ColumnFamilyOptions(ropts)); +#if ROCKSDB_VERSION_GE(11, 0, 0) + rs = rocksdb::DB::Open(rocksdb::DBOptions(ropts), db_dir, descs, + &reopen_handles, &reopen_owner); + db = reopen_owner.get(); +#else + rs = rocksdb::DB::Open(rocksdb::DBOptions(ropts), db_dir, descs, + &reopen_handles, &db); +#endif + } else { +#if ROCKSDB_VERSION_GE(11, 0, 0) + rs = rocksdb::DB::Open(ropts, db_dir, &reopen_owner); + db = reopen_owner.get(); +#else + rs = rocksdb::DB::Open(ropts, db_dir, &db); +#endif + if (rs.ok()) reopen_handles.push_back(db->DefaultColumnFamily()); + } + if (!rs.ok()) { + fail("reopen: " + rs.ToString()); + } else { + uint64_t db_count = 0, db_size_sum = 0; + std::vector> seen(oracle.size(), + std::vector(U, uint8_t{0})); + for (size_t cf = 0; cf < reopen_handles.size(); ++cf) { + if (cf >= oracle.size()) { + fail("iterator unexpected column family"); + break; + } + std::unique_ptr it( + db->NewIterator(ro, reopen_handles[cf])); + for (it->SeekToFirst(); it->Valid(); it->Next()) { + uint64_t id = KeyIdFromSlice(it->key().data(), it->key().size()); + if (id >= U) { + fail("iterator unexpected key id"); + continue; + } + std::string expected_key = MakeKey(id, seed); + if (it->key().compare(rocksdb::Slice(expected_key)) != 0) { + fail("iterator key mismatch"); + continue; + } + const auto& e = oracle[cf][id]; + if (!e.present) { + fail("iterator unexpected live key"); + continue; + } + if (seen[cf][id]) { + fail("iterator duplicate key"); + continue; + } + auto v = it->value(); + auto h = Hash128(v.data(), v.size()); + if (v.size() != e.size || h.first != e.h1 || h.second != e.h2) { + fail("iterator value mismatch"); + continue; + } + seen[cf][id] = 1; + ++db_count; + db_size_sum += uint32_t(v.size()); + } + if (!it->status().ok()) { + fail("iter status: " + it->status().ToString()); + } + } + for (size_t cf = 0; cf < oracle.size(); ++cf) { + for (uint64_t id = 0; id < U; ++id) { + if (oracle[cf][id].present && !seen[cf][id]) { + fail("iterator missing live key"); + break; + } + } + } + if (db_count != oracle_count || db_size_sum != oracle_size_sum) { + fail("full-iterator mismatch (count " + std::to_string(db_count) + "/" + + std::to_string(oracle_count) + ")"); + } + } + if (db && pf.num_cfs > 1) { + for (auto* h : reopen_handles) db->DestroyColumnFamilyHandle(h); + } +#if ROCKSDB_VERSION_GE(11, 0, 0) + reopen_owner.reset(); +#else + if (db) delete db; +#endif + + uint64_t live_final = 0; + for (const auto& cf_oracle : oracle) + for (const auto& e : cf_oracle) + if (e.present) live_final += 16 + e.size; + double denom = double(std::max(live_final, 64ull << 20)); + const uint64_t final_table_bytes = table_file_bytes(); + double final_space_amp = double(final_table_bytes) / denom; + double lsm_waf = + double(table_bytes) / double(std::max(1, user_put_bytes)); + + std::ofstream o(out_path); + o.setf(std::ios::fixed); + o << "{\n \"ok\": true,\n"; + o << " \"correctness_ok\": " << (correctness_ok ? "true" : "false") << ",\n"; + o << " \"fail_reason\": \"" << JsonEscape(fail_reason) << "\",\n"; + o << " \"phase1_ops_per_sec\": " << ops_per_sec[1] << ",\n"; + o << " \"phase2_ops_per_sec\": " << ops_per_sec[2] << ",\n"; + o << " \"phase3_ops_per_sec\": " << ops_per_sec[3] << ",\n"; + o << " \"phase4_ops_per_sec\": " << ops_per_sec[4] << ",\n"; + o << " \"lsm_waf\": " << lsm_waf << ",\n"; + o << " \"get_p99_us\": " << Percentile(get_lat_us, 0.99) << ",\n"; + o << " \"scan_p99_us\": " << Percentile(scan_lat_us, 0.99) << ",\n"; + o << " \"pre_drain_get_p99_us\": " << pre_drain_get_p99 << ",\n"; + o << " \"pre_drain_scan_p99_us\": " << pre_drain_scan_p99 << ",\n"; + o << " \"pre_drain_space_amp\": " << pre_drain_space_amp << ",\n"; + o << " \"pre_drain_lsm_bytes\": " << pre_drain_table_bytes << ",\n"; + o << " \"pre_drain_l0_base_overlap_bytes\": " + << pre_lsm.l0_base_overlap_bytes << ",\n"; + o << " \"pre_drain_pending_compaction_bytes\": " << pre_lsm.pending << ",\n"; + o << " \"pre_drain_l0_files\": " << pre_lsm.level_files[0] << ",\n"; + o << " \"pre_drain_l0_files_by_cf\": ["; + for (size_t i = 0; i < pre_lsm.l0_files_by_cf.size(); ++i) { + if (i > 0) o << ","; + o << pre_lsm.l0_files_by_cf[i]; + } + o << "],\n"; + o << " \"pre_drain_level_files\": [" << pre_lsm.level_files[0] << "," + << pre_lsm.level_files[1] << "," << pre_lsm.level_files[2] << "," + << pre_lsm.level_files[3] << "," << pre_lsm.level_files[4] << "," + << pre_lsm.level_files[5] << "],\n"; + o << " \"pre_drain_level_bytes\": [" << pre_lsm.level_bytes[0] << "," + << pre_lsm.level_bytes[1] << "," << pre_lsm.level_bytes[2] << "," + << pre_lsm.level_bytes[3] << "," << pre_lsm.level_bytes[4] << "," + << pre_lsm.level_bytes[5] << "],\n"; + o << " \"drain_ms\": " << drain_ms << ",\n"; + o << " \"drain_compaction_bytes\": " << drain_compaction_bytes << ",\n"; + o << " \"final_space_amp\": " << final_space_amp << ",\n"; + o << " \"final_lsm_bytes\": " << final_table_bytes << ",\n"; + o << " \"pending_compaction_bytes\": " << final_lsm.pending << ",\n"; + o << " \"l0_files\": " << final_lsm.level_files[0] << ",\n"; + o << " \"level_files\": [" << final_lsm.level_files[0] << "," + << final_lsm.level_files[1] << "," << final_lsm.level_files[2] << "," + << final_lsm.level_files[3] << "," << final_lsm.level_files[4] << "," + << final_lsm.level_files[5] << "],\n"; + o << " \"column_families\": " << final_lsm.column_families << ",\n"; + o << " \"base_fingerprint\": " << base_fingerprint << ",\n"; + o << " \"l0_trigger\": " << pf.l0_trigger << ",\n"; + o << " \"compaction_events\": " << listener->compaction_events.load() + << ",\n"; + o << " \"intra_l0_compactions\": " << listener->intra_l0_compactions.load() + << ",\n"; + o << " \"failed_compactions\": " << listener->failed_compactions.load() + << ",\n"; + o << " \"background_errors\": " << listener->background_errors.load() + << ",\n"; + o << " \"user_put_bytes\": " << user_put_bytes << ",\n"; + o << " \"live_logical_bytes\": " << live_final << ",\n"; + o << " \"flush_output_bytes\": " << flush_bytes << ",\n"; + o << " \"compaction_output_bytes\": " << comp_bytes << ",\n"; + o << " \"table_created_bytes\": " << table_bytes << ",\n"; + o << " \"read_amp\": " << read_amp << ",\n"; + o << " \"stall_micros\": " << stall_micros << "\n}\n"; + o.close(); + + return 0; +} diff --git a/2.0/problems/rocksdb_native_compaction_policy/readme b/2.0/problems/rocksdb_native_compaction_policy/readme new file mode 100644 index 000000000..683bbd272 --- /dev/null +++ b/2.0/problems/rocksdb_native_compaction_policy/readme @@ -0,0 +1,57 @@ +RocksDB Native Compaction Policy + +Goal + +Improve leveled compaction selection in RocksDB v10.10.1 while preserving database correctness. The workspace contains the pinned source tree at /app/rocksdb. The judge applies your patch to a clean checkout at commit 4595a5e95ae8525c42e172a054435782b3479c57, rebuilds RocksDB, and compares it with the unmodified build. + +Workload + +The judge runs native RocksDB workloads with changing write, point-read, scan, range-delete, snapshot, time-series, and multi-column-family phases. Options such as write-buffer size, L0 thresholds, level sizes, value sizes, and cache size vary by case. Leveled compaction is always used; universal and FIFO compaction are outside this task. + +Feedback uses one fixed development case per workload family plus a smoke case. Final verification uses two fixed judge-derived seeds per family. Final seeds are not included in the agent workspace or task configuration. + +Submission + +Submit /app/solution.patch. After editing the checkout, run: + + bash /app/make_submission.sh + bash /app/submit.sh + +make_submission.sh rejects changes outside the editable surface instead of silently omitting them. An empty patch is a valid zero-score baseline. + +Editable surface + + db/compaction/compaction_picker.cc + db/compaction/compaction_picker.h + db/compaction/compaction_picker_level.cc + db/compaction/compaction_picker_level.h + db/version_set.cc + +The task covers leveled compaction selection: choosing levels and files, computing file priority, handling L0 pressure, intra-L0 decisions, marked files, tombstone-driven picks, and picker expansion. Output-file cutting is not part of the editable surface. + +Correctness + +Correctness is a hard gate. The candidate must build and complete every case without crash, timeout, deadlock, or background error. The harness checks point reads, range deletes, held snapshots, column families, database reopen, and a complete iterator comparison against its logical oracle. + +Patches may not inspect judge identity, paths, environment variables, process state, clocks, profile names, or infrastructure details. New preprocessor directives and changes outside the five listed files are rejected. Submitted binaries and local benchmark output are ignored. + +Scoring + +Each case runs one isolated vanilla/candidate pair concurrently on the same deterministic operation stream. Final verification uses two seeds per workload family. The case objective is a weighted geometric mean of lower-is-better ratios: + + 40% write amplification + 25% read amplification + 20% pre-drain space amplification + 15% trusted compaction output required after the policy run + +The initial database load is compacted through a fixed manual path, fingerprinted, and excluded from scored counters. A candidate that changes this base state is invalid. Later writes and compactions run in fixed phase-boundary cycles so each picker decision starts from a reproducible state. Pre-drain memtables are flushed, actual table-file bytes are measured, and metadata is captured while background work is paused. After each policy run closes, an unmodified judge binary reopens the database and runs the normal vanilla policy until an additional pass produces no compaction output. It verifies the logical data before and after this residual drain. Trusted residual output is added to write amplification, and the policy plus residual drain is scored separately as 1 + output bytes divided by the larger of user-write bytes and 64 MiB, so deferred work cannot lower the measured cost. Final score uses the mean paired log improvement with a small cross-case dispersion penalty. Robust gains at or below 1.005x are treated as measurement noise and earn zero; a robust 1.20x aggregate reaches 100. A positive score requires at least 40% and at least two workload families to improve by 0.5% or more, and at most one family may regress by more than 2%. Severe per-case or per-metric regressions reduce or cap the score. Extreme runtime or stall regressions are validity guards; otherwise wall-clock throughput, latency, and stall time are diagnostics, not score terms. + +Feedback exposes validity, build status, aggregate gain, worst-case gain, component floor, workload breadth counts, average intra-L0 decision delta per case, case count, and a coarse score band. It does not expose per-case metrics, seeds, or final profile order. + +Resources + + vCPUs: 8 + memory: 16 GiB + storage: 32 GiB + build timeout: 7200 seconds + per-run timeout: 1800 seconds diff --git a/2.0/problems/rocksdb_native_compaction_policy/reference.cpp b/2.0/problems/rocksdb_native_compaction_policy/reference.cpp new file mode 100644 index 000000000..77f06be59 --- /dev/null +++ b/2.0/problems/rocksdb_native_compaction_policy/reference.cpp @@ -0,0 +1,62 @@ +diff --git a/db/compaction/compaction_picker.cc b/db/compaction/compaction_picker.cc +index b92a507..d885b70 100644 +--- a/db/compaction/compaction_picker.cc ++++ b/db/compaction/compaction_picker.cc +@@ -551,9 +551,54 @@ bool CompactionPicker::SetupOtherInputs( + assert(!expanded_output_level_inputs.empty()); + if (!AreFilesInCompaction(expanded_output_level_inputs.files) && + ExpandInputsToCleanCut(cf_name, vstorage, +- &expanded_output_level_inputs) && +- expanded_output_level_inputs.size() == output_level_inputs->size()) { +- expand_inputs = true; ++ &expanded_output_level_inputs)) { ++ expanded_inputs_size = TotalFileSize(expanded_inputs.files); ++ if (expanded_output_level_inputs.size() == ++ output_level_inputs->size()) { ++ expand_inputs = true; ++ } else if (input_level + 2 == vstorage->num_levels()) { ++ CompactionInputFiles added_inputs; ++ added_inputs.level = input_level; ++ for (auto* file : expanded_inputs.files) { ++ if (std::find(inputs->files.begin(), inputs->files.end(), file) == ++ inputs->files.end()) { ++ added_inputs.files.push_back(file); ++ } ++ } ++ if (!added_inputs.empty()) { ++ InternalKey added_start, added_limit; ++ GetRange(added_inputs, &added_start, &added_limit); ++ CompactionInputFiles added_output_inputs; ++ added_output_inputs.level = output_level; ++ vstorage->GetOverlappingInputs(output_level, &added_start, ++ &added_limit, ++ &added_output_inputs.files); ++ if (!added_output_inputs.empty() && ++ !AreFilesInCompaction(added_output_inputs.files) && ++ ExpandInputsToCleanCut(cf_name, vstorage, ++ &added_output_inputs)) { ++ const uint64_t added_inputs_size = ++ TotalFileSize(added_inputs.files); ++ const uint64_t added_output_size = ++ TotalFileSize(added_output_inputs.files); ++ const uint64_t expanded_output_size = ++ TotalFileSize(expanded_output_level_inputs.files); ++ const long double separate_write_bytes = ++ static_cast(inputs_size) + ++ output_level_inputs_size + added_inputs_size + ++ added_output_size; ++ const long double combined_write_bytes = ++ static_cast(expanded_inputs_size) + ++ expanded_output_size; ++ if (combined_write_bytes * 10.0L <= separate_write_bytes * 9.0L && ++ expanded_output_size <= ++ MultiplyCheckOverflow(expanded_inputs_size, 2.0)) { ++ expand_inputs = true; ++ output_level_inputs->files = expanded_output_level_inputs.files; ++ } ++ } ++ } ++ } + } + } + if (!expand_inputs) { diff --git a/2.0/problems/rocksdb_native_compaction_policy/reference.patch b/2.0/problems/rocksdb_native_compaction_policy/reference.patch new file mode 100644 index 000000000..77f06be59 --- /dev/null +++ b/2.0/problems/rocksdb_native_compaction_policy/reference.patch @@ -0,0 +1,62 @@ +diff --git a/db/compaction/compaction_picker.cc b/db/compaction/compaction_picker.cc +index b92a507..d885b70 100644 +--- a/db/compaction/compaction_picker.cc ++++ b/db/compaction/compaction_picker.cc +@@ -551,9 +551,54 @@ bool CompactionPicker::SetupOtherInputs( + assert(!expanded_output_level_inputs.empty()); + if (!AreFilesInCompaction(expanded_output_level_inputs.files) && + ExpandInputsToCleanCut(cf_name, vstorage, +- &expanded_output_level_inputs) && +- expanded_output_level_inputs.size() == output_level_inputs->size()) { +- expand_inputs = true; ++ &expanded_output_level_inputs)) { ++ expanded_inputs_size = TotalFileSize(expanded_inputs.files); ++ if (expanded_output_level_inputs.size() == ++ output_level_inputs->size()) { ++ expand_inputs = true; ++ } else if (input_level + 2 == vstorage->num_levels()) { ++ CompactionInputFiles added_inputs; ++ added_inputs.level = input_level; ++ for (auto* file : expanded_inputs.files) { ++ if (std::find(inputs->files.begin(), inputs->files.end(), file) == ++ inputs->files.end()) { ++ added_inputs.files.push_back(file); ++ } ++ } ++ if (!added_inputs.empty()) { ++ InternalKey added_start, added_limit; ++ GetRange(added_inputs, &added_start, &added_limit); ++ CompactionInputFiles added_output_inputs; ++ added_output_inputs.level = output_level; ++ vstorage->GetOverlappingInputs(output_level, &added_start, ++ &added_limit, ++ &added_output_inputs.files); ++ if (!added_output_inputs.empty() && ++ !AreFilesInCompaction(added_output_inputs.files) && ++ ExpandInputsToCleanCut(cf_name, vstorage, ++ &added_output_inputs)) { ++ const uint64_t added_inputs_size = ++ TotalFileSize(added_inputs.files); ++ const uint64_t added_output_size = ++ TotalFileSize(added_output_inputs.files); ++ const uint64_t expanded_output_size = ++ TotalFileSize(expanded_output_level_inputs.files); ++ const long double separate_write_bytes = ++ static_cast(inputs_size) + ++ output_level_inputs_size + added_inputs_size + ++ added_output_size; ++ const long double combined_write_bytes = ++ static_cast(expanded_inputs_size) + ++ expanded_output_size; ++ if (combined_write_bytes * 10.0L <= separate_write_bytes * 9.0L && ++ expanded_output_size <= ++ MultiplyCheckOverflow(expanded_inputs_size, 2.0)) { ++ expand_inputs = true; ++ output_level_inputs->files = expanded_output_level_inputs.files; ++ } ++ } ++ } ++ } + } + } + if (!expand_inputs) { diff --git a/2.0/problems/rocksdb_native_compaction_policy/test_evaluator.py b/2.0/problems/rocksdb_native_compaction_policy/test_evaluator.py new file mode 100644 index 000000000..979e66ab8 --- /dev/null +++ b/2.0/problems/rocksdb_native_compaction_policy/test_evaluator.py @@ -0,0 +1,421 @@ +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + + +PROBLEM_DIR = Path(__file__).parent +SPEC = importlib.util.spec_from_file_location( + "rocksdb_native_compaction_evaluator", + PROBLEM_DIR / "evaluator.py", +) +assert SPEC is not None and SPEC.loader is not None +EVALUATOR = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = EVALUATOR +SPEC.loader.exec_module(EVALUATOR) + +VALIDATOR_SPEC = importlib.util.spec_from_file_location( + "rocksdb_native_compaction_validator", + PROBLEM_DIR / "harbor" / "app" / "validate_solution_patch.py", +) +assert VALIDATOR_SPEC is not None and VALIDATOR_SPEC.loader is not None +VALIDATOR = importlib.util.module_from_spec(VALIDATOR_SPEC) +VALIDATOR_SPEC.loader.exec_module(VALIDATOR) + + +def _case(gain: float, *, profile: str, component: float = 1.0) -> dict: + return { + "gain": gain, + "profile": profile, + "ratios": {name: component for name in EVALUATOR.METRIC_WEIGHTS}, + "candidate_intra_l0": 0, + "vanilla_intra_l0": 0, + } + + +def _cases(gains: list[float], *, component: float = 1.0) -> list[dict]: + return [ + _case(gain, profile=f"profile_{index}", component=component) + for index, gain in enumerate(gains) + ] + + +def _agent_accepts(text: str) -> bool: + try: + VALIDATOR.validate(text) + except SystemExit: + return False + return True + + +def test_reference_patch_is_valid() -> None: + ok, message, metrics = EVALUATOR.validate_patch(PROBLEM_DIR / "reference.patch") + assert ok, message + assert metrics["changed_files"] == [ + "db/compaction/compaction_picker.cc" + ] + assert (PROBLEM_DIR / "reference.cpp").read_bytes() == ( + PROBLEM_DIR / "reference.patch" + ).read_bytes() + + +def test_no_improvement_scores_zero() -> None: + score, unbounded, _ = EVALUATOR._score_cases(_cases([1.0] * 6)) + assert score == 0.0 + assert unbounded < 0.0 + + +def test_gain_below_half_percent_is_inside_noise_floor() -> None: + score, unbounded, _ = EVALUATOR._score_cases(_cases([1.004] * 6)) + assert score == 0.0 + assert unbounded < 0.0 + + +def test_gain_at_half_percent_floor_scores_zero() -> None: + score, unbounded, _ = EVALUATOR._score_cases(_cases([1.005] * 6)) + assert score == 0.0 + assert abs(unbounded) < 1e-9 + + +def test_single_profile_spike_scores_zero() -> None: + cases = _cases([1.10] + [1.0] * 5) + score, _, metrics = EVALUATOR._score_cases(cases) + assert score == 0.0 + assert metrics["improved_profile_count"] == 1 + assert metrics["required_improved_profile_count"] == 3 + assert metrics["worst_case_gain"] == 1.0 + + +def test_multi_profile_improvement_can_score() -> None: + cases = _cases([1.04, 1.04, 1.04, 1.0, 1.0, 1.0]) + score, _, metrics = EVALUATOR._score_cases(cases) + assert score > 0.0 + assert metrics["improved_profile_count"] == 3 + + +def test_broad_target_improvement_reaches_full_score() -> None: + score, unbounded, _ = EVALUATOR._score_cases(_cases([1.20] * 6)) + assert score > 99.9 + assert unbounded > 99.9 + + +def test_severe_component_regression_hard_fails() -> None: + cases = _cases([1.10] * 5) + [_case(1.0, profile="profile_5", component=0.49)] + score, _, metrics = EVALUATOR._score_cases(cases) + assert score == 0.0 + assert metrics["component_floor"] == 0.49 + + +def test_component_regression_caps_are_graduated() -> None: + for component, expected_cap in ((0.50, 0.0), (0.51, 10.0), (0.67, 25.0)): + cases = _cases([2.0] * 6, component=component) + score, _, _ = EVALUATOR._score_cases(cases) + assert score <= expected_cap + + +def test_multiple_material_profile_regressions_score_zero() -> None: + cases = _cases([1.10, 1.10, 1.10, 1.10, 0.97, 0.97]) + score, _, metrics = EVALUATOR._score_cases(cases) + assert score == 0.0 + assert metrics["material_regression_count"] == 2 + + +def test_final_cases_are_stable_per_suite_key(monkeypatch) -> None: + monkeypatch.setenv("FRONTIER_LSM_FINAL_SUITE_KEY", "suite-a") + first = EVALUATOR._generate_final_cases() + second = EVALUATOR._generate_final_cases() + assert sorted(profile for _, profile in first) == sorted( + profile for profile in EVALUATOR.FINAL_PROFILES for _ in range(2) + ) + assert sorted(profile for _, profile in second) == sorted( + profile for profile in EVALUATOR.FINAL_PROFILES for _ in range(2) + ) + assert len({seed for seed, _ in first}) == len(first) + assert all(seed > 0 for seed, _ in first) + assert first == second + + monkeypatch.setenv("FRONTIER_LSM_FINAL_SUITE_KEY", "suite-b") + assert EVALUATOR._generate_final_cases() != first + + +def test_feedback_covers_every_final_profile() -> None: + feedback_profiles = { + profile for _, profile in EVALUATOR._feedback_cases() if profile != "smoke" + } + assert feedback_profiles == set(EVALUATOR.FINAL_PROFILES) + + +def test_agent_and_judge_patch_policies_match() -> None: + assert VALIDATOR.ALLOWLIST == EVALUATOR.ALLOWLIST + assert VALIDATOR.MAX_CHANGED_FILES == EVALUATOR.MAX_CHANGED_FILES + assert VALIDATOR.STRUCTURAL_MARKERS == EVALUATOR.STRUCTURAL_MARKERS + assert VALIDATOR.FORBIDDEN_IDENTIFIERS == EVALUATOR.FORBIDDEN_IDENTIFIERS + assert ( + VALIDATOR.FORBIDDEN_CALL_IDENTIFIERS + == EVALUATOR.FORBIDDEN_CALL_IDENTIFIERS + ) + assert VALIDATOR.FORBIDDEN_FRAGMENTS == EVALUATOR.FORBIDDEN_FRAGMENTS + + reference = (PROBLEM_DIR / "reference.patch").read_text() + repeated = "\n".join(reference for _ in range(EVALUATOR.MAX_CHANGED_FILES + 1)) + outside = reference.replace( + "db/compaction/compaction_picker.cc", "db/db_impl/db_impl.cc" + ) + assert outside != reference + for text in ("", reference, repeated, outside): + judge_ok, _, _ = EVALUATOR._validate_patch_text(text, allow_empty=True) + assert _agent_accepts(text) == judge_ok + + anchor = "+ expanded_inputs_size = TotalFileSize(expanded_inputs.files);" + for added_line in ( + "+ const auto probe = ioptions_.clock->NowMicros();", + "+ const auto probe = clock();", + "+ const auto probe = gettimeofday(nullptr, nullptr);", + "+ const auto probe = std::random_device{}();", + ): + probe = reference.replace(anchor, f"{added_line}\n{anchor}") + assert probe != reference + judge_ok, _, _ = EVALUATOR._validate_patch_text(probe, allow_empty=True) + assert not judge_ok + assert not _agent_accepts(probe) + + +def test_patch_policy_allows_removal_only_changes() -> None: + text = """diff --git a/db/compaction/compaction_picker.cc b/db/compaction/compaction_picker.cc +--- a/db/compaction/compaction_picker.cc ++++ b/db/compaction/compaction_picker.cc +@@ -1 +0,0 @@ +-obsolete_policy_branch(); +""" + judge_ok, message, _ = EVALUATOR._validate_patch_text(text, allow_empty=True) + assert judge_ok, message + assert _agent_accepts(text) + + +def test_patch_policy_allows_legitimate_picker_context_names() -> None: + reference = (PROBLEM_DIR / "reference.patch").read_text() + anchor = "+ expanded_inputs_size = TotalFileSize(expanded_inputs.files);" + additions = "\n".join( + ( + "+ const auto path_count = ioptions_.db_paths.size();", + "+ const uint64_t seed = path_count;", + "+ SystemClock* clock = nullptr;", + ) + ) + probe = reference.replace(anchor, f"{additions}\n{anchor}") + assert probe != reference + judge_ok, message, _ = EVALUATOR._validate_patch_text(probe, allow_empty=True) + assert judge_ok, message + assert _agent_accepts(probe) + + +def test_read_amplification_instrumentation_is_enabled() -> None: + harness = (PROBLEM_DIR / "judge" / "harness" / "lsm_policy_harness.cc").read_text() + assert "topt.read_amp_bytes_per_bit = 16;" in harness + + +def test_case_arms_run_once_in_parallel_pair(monkeypatch) -> None: + calls: list[str] = [] + + def fake_run( + binary: Path, trusted_inspector: Path, seed: int, profile: str + ) -> dict: + calls.append(binary.name) + return { + "ok": True, + "correctness_ok": True, + "_wall_seconds": 1.0, + "user_put_bytes": 1 << 30, + "drain_compaction_bytes": 0, + "trusted_compaction_output_bytes": 0, + "trusted_upper_level_files": 0, + "table_created_bytes": 2 << 30, + "l0_trigger": 4, + "pre_drain_pending_compaction_bytes": 0, + "pre_drain_l0_files": 2, + "pre_drain_l0_files_by_cf": [2], + "pre_drain_level_files": [2, 1, 1, 1, 1, 1], + "column_families": 1, + "base_fingerprint": 123, + "lsm_waf": 2.0, + "read_amp": 1.5, + "pre_drain_space_amp": 1.2, + "stall_micros": 1000, + "intra_l0_compactions": 1, + } + + monkeypatch.setattr(EVALUATOR, "_run_harness", fake_run) + result = EVALUATOR._paired_case( + Path("vanilla"), Path("candidate"), 1202, "l0_pressure" + ) + assert result["ok"] + assert sorted(calls) == ["candidate", "vanilla"] + + +def test_feedback_case_runs_one_stable_pair(monkeypatch) -> None: + calls: list[str] = [] + + def fake_run( + binary: Path, trusted_inspector: Path, seed: int, profile: str + ) -> dict: + calls.append(binary.name) + return { + "ok": True, + "correctness_ok": True, + "_wall_seconds": 1.0, + "user_put_bytes": 1 << 30, + "drain_compaction_bytes": 0, + "trusted_compaction_output_bytes": 0, + "trusted_upper_level_files": 0, + "table_created_bytes": 2 << 30, + "l0_trigger": 4, + "column_families": 1, + "pre_drain_pending_compaction_bytes": 0, + "pre_drain_l0_files": 2, + "pre_drain_l0_files_by_cf": [2], + "lsm_waf": 2.0, + "read_amp": 1.5, + "pre_drain_space_amp": 1.2, + "stall_micros": 0, + "intra_l0_compactions": 1, + "base_fingerprint": 123, + } + + monkeypatch.setattr(EVALUATOR, "_run_harness", fake_run) + monkeypatch.delenv("FRONTIER_SUBMISSION_ROLE", raising=False) + result = EVALUATOR._paired_case( + Path("vanilla"), Path("candidate"), 1202, "l0_pressure" + ) + assert result["ok"] + assert sorted(calls) == ["candidate", "vanilla"] + + +def test_debt_uses_measured_drain_output() -> None: + base = { + "user_put_bytes": 1 << 30, + "drain_compaction_bytes": 0, + "trusted_compaction_output_bytes": 1 << 28, + "table_created_bytes": 2 << 30, + "l0_trigger": 4, + "column_families": 1, + "pre_drain_pending_compaction_bytes": 0, + "pre_drain_l0_files": 4, + "pre_drain_l0_files_by_cf": [4], + "pre_drain_level_files": [4, 1, 1, 1, 1, 1], + "lsm_waf": 2.0, + "read_amp": 1.5, + "pre_drain_space_amp": 1.2, + "stall_micros": 1000, + } + moved = dict(base) + moved["pre_drain_l0_files"] = 3 + moved["pre_drain_l0_files_by_cf"] = [3] + moved["pre_drain_level_files"] = [3, 2, 1, 1, 1, 1] + assert EVALUATOR._objective_metrics(base)["debt"] == EVALUATOR._objective_metrics( + moved + )["debt"] + + +def test_more_drain_output_increases_debt() -> None: + low = { + "user_put_bytes": 1 << 30, + "drain_compaction_bytes": 0, + "trusted_compaction_output_bytes": 1 << 27, + "table_created_bytes": 2 << 30, + "l0_trigger": 4, + "column_families": 3, + "pre_drain_pending_compaction_bytes": 0, + "pre_drain_l0_files": 12, + "pre_drain_l0_files_by_cf": [12, 0, 0], + "lsm_waf": 2.0, + "read_amp": 1.5, + "pre_drain_space_amp": 1.2, + } + high = dict(low) + high["trusted_compaction_output_bytes"] = 1 << 29 + assert EVALUATOR._objective_metrics(high)["debt"] > EVALUATOR._objective_metrics( + low + )["debt"] + + +def test_debt_cost_has_explicit_neutral_point_and_scale() -> None: + run = { + "user_put_bytes": 1 << 30, + "drain_compaction_bytes": 0, + "trusted_compaction_output_bytes": 0, + "table_created_bytes": 2 << 30, + "read_amp": 1.5, + "pre_drain_space_amp": 1.2, + } + assert EVALUATOR._objective_metrics(run)["debt"] == 1.0 + run["trusted_compaction_output_bytes"] = 1 << 28 + assert EVALUATOR._objective_metrics(run)["debt"] == 1.25 + + +def test_debt_cost_uses_64_mib_denominator_floor() -> None: + run = { + "user_put_bytes": 1 << 25, + "drain_compaction_bytes": 0, + "trusted_compaction_output_bytes": 1 << 26, + "table_created_bytes": 1 << 27, + "read_amp": 1.5, + "pre_drain_space_amp": 1.2, + } + assert EVALUATOR._objective_metrics(run)["debt"] == 2.0 + + +def test_debt_pair_ratio_is_total_cost_ratio() -> None: + base = { + "user_put_bytes": 1 << 30, + "drain_compaction_bytes": 0, + "table_created_bytes": 2 << 30, + "read_amp": 1.5, + "pre_drain_space_amp": 1.2, + } + vanilla = dict(base, trusted_compaction_output_bytes=1 << 26) + candidate = dict(base, trusted_compaction_output_bytes=3 << 26) + ratio = EVALUATOR._objective_metrics(vanilla)["debt"] / EVALUATOR._objective_metrics( + candidate + )["debt"] + assert abs(ratio - 17 / 19) < 1e-12 + + +def test_observed_overlap_debt_regression_is_not_catastrophic() -> None: + base = { + "user_put_bytes": 187_597_280, + "drain_compaction_bytes": 0, + "table_created_bytes": 400_000_000, + "read_amp": 1.5, + "pre_drain_space_amp": 1.2, + } + vanilla = dict(base, trusted_compaction_output_bytes=11_740_781) + candidate = dict(base, trusted_compaction_output_bytes=35_169_986) + ratio = EVALUATOR._objective_metrics(vanilla)["debt"] / EVALUATOR._objective_metrics( + candidate + )["debt"] + assert abs(ratio - 0.8948265361) < 1e-9 + assert ratio > 0.80 + + +def test_harness_uses_quiescent_scoring_boundaries() -> None: + harness = (PROBLEM_DIR / "judge" / "harness" / "lsm_policy_harness.cc").read_text() + assert 'listener->Reset();' in harness + assert 'options.statistics->Reset()' in harness + assert 'PauseBackgroundWork()' in harness + assert 'GetAllColumnFamilyMetaData' in harness + assert '--trusted-residual' in harness + assert 'db->EnableAutoCompaction(handles)' in harness + assert 'trusted-residual-did-not-converge' in harness + assert 'CollectDataDigest' in harness + assert 'SpaceSampler' not in harness + + +def test_judge_image_uses_a_verified_source_archive() -> None: + dockerfile = (PROBLEM_DIR / "docker" / "judge" / "Dockerfile").read_text() + assert "9469e7d24644e298134d0464ec2d398a9b91cc401102009ec1ecde0383485d6c" in dockerfile + assert "codeload.github.com/facebook/rocksdb/tar.gz/{commit}" in dockerfile + assert ".frontier-upstream-commit" in dockerfile + assert "hashlib.sha256(archive).hexdigest()" in dockerfile + assert "git clone" not in dockerfile + assert EVALUATOR.SOURCE_COMMIT_MARKER == ".frontier-upstream-commit" From 6b3989af90b67f482657e9938e5daa8227a3cb28 Mon Sep 17 00:00:00 2001 From: Bo Chen Date: Wed, 15 Jul 2026 01:28:58 +0800 Subject: [PATCH 2/2] Rank invalid submissions below valid ones and recalibrate the score scale Address review feedback on #156: - _invalid() now returns score_unbounded = -1e6 so Harbor's (score, score_unbounded) best-artifact selection can never prefer an invalid submission over a valid or empty one; the empty no-op patch keeps (0.0, 0.0) as the documented neutral baseline. - TARGET_ROBUST_GAIN 1.20 -> 1.017: the reference patch (robust_gain ~1.011 on the final suite) lands near 50 instead of 3.32. The 1.005 noise deadband and every validity/breadth/regression gate are unchanged, so the zero-vs-positive frontier is identical. - readme scoring section updated; evaluator tests extended 26 -> 31. Co-Authored-By: Claude Fable 5 --- .../evaluator.py | 12 +++- .../rocksdb_native_compaction_policy/readme | 2 +- .../test_evaluator.py | 56 ++++++++++++++++++- 3 files changed, 64 insertions(+), 6 deletions(-) diff --git a/2.0/problems/rocksdb_native_compaction_policy/evaluator.py b/2.0/problems/rocksdb_native_compaction_policy/evaluator.py index ad9441960..985ac497b 100644 --- a/2.0/problems/rocksdb_native_compaction_policy/evaluator.py +++ b/2.0/problems/rocksdb_native_compaction_policy/evaluator.py @@ -26,6 +26,11 @@ MAX_PATCH_BYTES = 256 * 1024 MAX_CHANGED_FILES = 5 +# Harbor keeps the best iterative artifact by (score, score_unbounded), and +# valid below-floor runs report small negative unbounded scores, so invalid +# submissions must carry a far lower sentinel than any valid run can reach. +INVALID_SCORE_UNBOUNDED = -1.0e6 + ALLOWLIST = frozenset( { "db/compaction/compaction_picker.cc", @@ -591,7 +596,10 @@ def _objective_metrics(run: dict[str, Any]) -> dict[str, float]: } MIN_ROBUST_GAIN = 1.005 -TARGET_ROBUST_GAIN = 1.20 +# Saturation anchor for 100 points, calibrated so the reference patch +# (robust_gain ~1.011 on the final suite) lands near 50 while keeping +# headroom above it before the bounded score saturates. +TARGET_ROBUST_GAIN = 1.017 PROFILE_IMPROVEMENT_GAIN = MIN_ROBUST_GAIN PROFILE_REGRESSION_GAIN = 0.98 MIN_IMPROVED_PROFILE_FRACTION = 0.40 @@ -825,7 +833,7 @@ def _public_metrics(metrics: dict[str, Any]) -> dict[str, Any]: def _invalid(message: str, metrics: dict[str, Any] | None = None): payload = dict(metrics or {}) payload.setdefault("valid_patch", 0) - return 0.0, 0.0, message, _public_metrics(payload) + return 0.0, INVALID_SCORE_UNBOUNDED, message, _public_metrics(payload) def _band(score: float) -> str: diff --git a/2.0/problems/rocksdb_native_compaction_policy/readme b/2.0/problems/rocksdb_native_compaction_policy/readme index 683bbd272..68032c05f 100644 --- a/2.0/problems/rocksdb_native_compaction_policy/readme +++ b/2.0/problems/rocksdb_native_compaction_policy/readme @@ -44,7 +44,7 @@ Each case runs one isolated vanilla/candidate pair concurrently on the same dete 20% pre-drain space amplification 15% trusted compaction output required after the policy run -The initial database load is compacted through a fixed manual path, fingerprinted, and excluded from scored counters. A candidate that changes this base state is invalid. Later writes and compactions run in fixed phase-boundary cycles so each picker decision starts from a reproducible state. Pre-drain memtables are flushed, actual table-file bytes are measured, and metadata is captured while background work is paused. After each policy run closes, an unmodified judge binary reopens the database and runs the normal vanilla policy until an additional pass produces no compaction output. It verifies the logical data before and after this residual drain. Trusted residual output is added to write amplification, and the policy plus residual drain is scored separately as 1 + output bytes divided by the larger of user-write bytes and 64 MiB, so deferred work cannot lower the measured cost. Final score uses the mean paired log improvement with a small cross-case dispersion penalty. Robust gains at or below 1.005x are treated as measurement noise and earn zero; a robust 1.20x aggregate reaches 100. A positive score requires at least 40% and at least two workload families to improve by 0.5% or more, and at most one family may regress by more than 2%. Severe per-case or per-metric regressions reduce or cap the score. Extreme runtime or stall regressions are validity guards; otherwise wall-clock throughput, latency, and stall time are diagnostics, not score terms. +The initial database load is compacted through a fixed manual path, fingerprinted, and excluded from scored counters. A candidate that changes this base state is invalid. Later writes and compactions run in fixed phase-boundary cycles so each picker decision starts from a reproducible state. Pre-drain memtables are flushed, actual table-file bytes are measured, and metadata is captured while background work is paused. After each policy run closes, an unmodified judge binary reopens the database and runs the normal vanilla policy until an additional pass produces no compaction output. It verifies the logical data before and after this residual drain. Trusted residual output is added to write amplification, and the policy plus residual drain is scored separately as 1 + output bytes divided by the larger of user-write bytes and 64 MiB, so deferred work cannot lower the measured cost. Final score uses the mean paired log improvement with a small cross-case dispersion penalty. Robust gains at or below 1.005x are treated as measurement noise and earn zero; a robust 1.017x aggregate reaches 100. Invalid or failed submissions score zero and report a strongly negative unbounded score, so they always rank below valid submissions. A positive score requires at least 40% and at least two workload families to improve by 0.5% or more, and at most one family may regress by more than 2%. Severe per-case or per-metric regressions reduce or cap the score. Extreme runtime or stall regressions are validity guards; otherwise wall-clock throughput, latency, and stall time are diagnostics, not score terms. Feedback exposes validity, build status, aggregate gain, worst-case gain, component floor, workload breadth counts, average intra-L0 decision delta per case, case count, and a coarse score band. It does not expose per-case metrics, seeds, or final profile order. diff --git a/2.0/problems/rocksdb_native_compaction_policy/test_evaluator.py b/2.0/problems/rocksdb_native_compaction_policy/test_evaluator.py index 979e66ab8..da4d0a2a4 100644 --- a/2.0/problems/rocksdb_native_compaction_policy/test_evaluator.py +++ b/2.0/problems/rocksdb_native_compaction_policy/test_evaluator.py @@ -78,6 +78,48 @@ def test_gain_at_half_percent_floor_scores_zero() -> None: assert abs(unbounded) < 1e-9 +def test_reference_level_gain_lands_midscale() -> None: + score, unbounded, _ = EVALUATOR._score_cases(_cases([1.011] * 6)) + assert 45.0 < score < 55.0 + assert abs(score - unbounded) < 1e-9 + + +def test_invalid_submission_ranks_below_any_valid_run() -> None: + score, unbounded, _, metrics = EVALUATOR._invalid("boom") + assert score == 0.0 + assert unbounded == EVALUATOR.INVALID_SCORE_UNBOUNDED + assert metrics["valid_patch"] == 0 + + _, worst_valid_unbounded, _ = EVALUATOR._score_cases(_cases([0.5] * 6)) + assert worst_valid_unbounded > EVALUATOR.INVALID_SCORE_UNBOUNDED + assert (0.0, worst_valid_unbounded) > (0.0, EVALUATOR.INVALID_SCORE_UNBOUNDED) + assert (0.0, 0.0) > (0.0, worst_valid_unbounded) + + +def test_evaluate_reports_sentinel_for_invalid_patch(tmp_path) -> None: + reference = (PROBLEM_DIR / "reference.patch").read_text() + outside = reference.replace( + "db/compaction/compaction_picker.cc", "db/db_impl/db_impl.cc" + ) + patch = tmp_path / "solution.patch" + patch.write_text(outside) + score, unbounded, message, metrics = EVALUATOR.evaluate(str(patch)) + assert score == 0.0 + assert unbounded == EVALUATOR.INVALID_SCORE_UNBOUNDED + assert "outside the editable surface" in message + assert metrics["valid_patch"] == 0 + + +def test_evaluate_keeps_empty_patch_as_neutral_baseline(tmp_path) -> None: + patch = tmp_path / "solution.patch" + patch.write_text("") + score, unbounded, message, metrics = EVALUATOR.evaluate(str(patch)) + assert score == 0.0 + assert unbounded == 0.0 + assert metrics["empty_patch"] == 1 + assert unbounded > EVALUATOR.INVALID_SCORE_UNBOUNDED + + def test_single_profile_spike_scores_zero() -> None: cases = _cases([1.10] + [1.0] * 5) score, _, metrics = EVALUATOR._score_cases(cases) @@ -94,10 +136,18 @@ def test_multi_profile_improvement_can_score() -> None: assert metrics["improved_profile_count"] == 3 -def test_broad_target_improvement_reaches_full_score() -> None: - score, unbounded, _ = EVALUATOR._score_cases(_cases([1.20] * 6)) +def test_target_gain_reaches_full_score() -> None: + score, unbounded, _ = EVALUATOR._score_cases( + _cases([EVALUATOR.TARGET_ROBUST_GAIN] * 6) + ) assert score > 99.9 - assert unbounded > 99.9 + assert abs(unbounded - 100.0) < 1e-6 + + +def test_gains_far_above_target_saturate_bounded_score() -> None: + score, unbounded, _ = EVALUATOR._score_cases(_cases([1.20] * 6)) + assert score == 100.0 + assert unbounded > 100.0 def test_severe_component_regression_hard_fails() -> None: