diff --git a/.github/workflows/common-warmup-screen.yml b/.github/workflows/common-warmup-screen.yml new file mode 100644 index 0000000..445e6e1 --- /dev/null +++ b/.github/workflows/common-warmup-screen.yml @@ -0,0 +1,41 @@ +name: Common warm-up screen v1 +on: + workflow_dispatch: +permissions: + contents: read +concurrency: + group: common-warmup-screen + cancel-in-progress: false +jobs: + screen: + if: >- + github.repository == 'tappe9/simple-api-benchmark' && + github.ref == format('refs/heads/{0}', github.event.repository.default_branch) && + github.event_name == 'workflow_dispatch' + runs-on: ubuntu-24.04 + timeout-minutes: 120 + env: + PYTHONDONTWRITEBYTECODE: '1' + steps: + - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 + timeout-minutes: 3 + with: + ref: ${{ github.sha }} + fetch-depth: 1 + persist-credentials: false + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 + timeout-minutes: 3 + with: + python-version-file: apps/python-fastapi/.python-version + - name: Run one fixed endpoint-isolated screen, reserving cleanup and upload time + run: timeout --signal=TERM --kill-after=120s 104m python -m benchmark.common_warmup_screen + - name: Retain complete common warm-up screen evidence, including failures + if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 + timeout-minutes: 7 + with: + name: common-warmup-screen-${{ github.run_id }}-${{ github.run_attempt }} + path: .cache/common-warmup-screen/ + include-hidden-files: true + retention-days: 90 + if-no-files-found: error diff --git a/benchmark/common_warmup_environment.py b/benchmark/common_warmup_environment.py new file mode 100644 index 0000000..1aa359d --- /dev/null +++ b/benchmark/common_warmup_environment.py @@ -0,0 +1,620 @@ +"""Evidence-preserving environment for the separate common-warmup screen. + +The original JVM environment keeps its 0..4 protocol. This module deliberately +owns its 0..8 window implementation rather than extending that existing protocol. +All load arguments and sequential CPU/memory sampling follow the common runner. +""" + +import contextlib +import hashlib +import json +import math +import re +import time +from datetime import datetime, timezone +from pathlib import Path + +from .contract_test import read_response +from .environment import ROOT, memory_bytes, validate_processes, validate_state +from .healthcheck import override_text, wait_external_readiness +from .jvm_environment import DiagnosticEnvironment +from .process import execute +from .results import BenchmarkFailure, parse_oha, require, strict_json + + +class CommonWarmupEnvironment(DiagnosticEnvironment): + """One fresh API/PostgreSQL pair; no contract preheating or runtime tuning.""" + + def __init__(self, oha: Path, artifacts: Path, **kwargs): + super().__init__(oha, artifacts, **kwargs) + self.postgres_container = None + self.postgres_identity = None + self.readiness_completed_at = None + self.previous_window_completed_at = None + self.start_attempted = False + + def _failure(self, stage: str, error: BaseException) -> None: + # Shared execute retains only final 4,000 nonzero-exit output characters; + # timeout/interruption output is unavailable. Never mask the first cause. + try: + (self.artifacts / f"{self.implementation}-{stage}-failure.log").write_text( + f"{type(error).__name__}: {error}\n", encoding="utf-8" + ) + try: + output = execute( + self.prefix + + [ + "logs", + "--no-color", + "--timestamps", + "--tail", + "200", + self.implementation, + "postgres", + ], + timeout=15, + ) + except Exception as log_error: + output = f"Container logs unavailable: {type(log_error).__name__}: {log_error}\n" + (self.artifacts / f"{self.implementation}-{stage}-container.log").write_text( + output, encoding="utf-8" + ) + except Exception: + pass + + def build(self, implementation: str) -> None: + self.implementation = implementation + try: + output = execute(self.prefix + ["build", implementation], timeout=900) + (self.artifacts / f"{implementation}-build.log").write_text(output, encoding="utf-8") + except BaseException as error: + with contextlib.suppress(OSError): + (self.artifacts / f"{implementation}-build.log").write_text( + f"{type(error).__name__}: {error}\n", encoding="utf-8" + ) + self._failure("build", error) + raise + + def _append(self, suffix: str, record: dict) -> None: + with (self.artifacts / f"{self.implementation}-{suffix}.jsonl").open( + "a", encoding="utf-8" + ) as output: + output.write(json.dumps(record) + "\n") + output.flush() + + @staticmethod + def _safe_state(value: dict) -> dict: + """Allowlist raw inspection fields; omit Env, health logs and other labels.""" + + def selected(source, fields): + if type(source) is not dict: + return {"invalid_type": type(source).__name__} + return {key: source[key] for key in fields if key in source} + + state = value.get("State", {}) + config = value.get("Config", {}) + host = value.get("HostConfig", {}) + result = selected(value, ("Id", "Image", "RestartCount")) + result["State"] = selected( + state, + ( + "Status", + "Running", + "Restarting", + "Dead", + "OOMKilled", + "StartedAt", + "FinishedAt", + "ExitCode", + ), + ) + if type(state) is dict and "Health" in state: + result["State"]["Health"] = selected(state["Health"], ("Status",)) + result["HostConfig"] = selected(host, ("NanoCpus", "Memory")) + if type(host) is dict: + result["HostConfig"]["RestartPolicy"] = selected( + host.get("RestartPolicy", {}), ("Name",) + ) + if "Tmpfs" in host: + result["HostConfig"]["Tmpfs"] = selected(host["Tmpfs"], ("/var/lib/postgresql",)) + if type(value.get("Mounts")) is list: + result["Mounts"] = [ + selected(mount, ("Type", "Source", "Destination", "RW")) + for mount in value["Mounts"] + if type(mount) is dict + and ( + mount.get("Destination") == "/docker-entrypoint-initdb.d/001-init.sql" + or str(mount.get("Destination", "")).startswith("/var/lib/postgresql") + ) + ] + result["Config"] = {} + if type(config) is dict: + result["Config"]["Labels"] = selected( + config.get("Labels", {}), + ("com.docker.compose.project", "com.docker.compose.service"), + ) + health = config.get("Healthcheck") + result["Config"]["healthcheck_disabled"] = type(health) is dict and health.get( + "Test" + ) == ["NONE"] + return result + + def _inspect_owned(self, container: str, role: str) -> dict: + started = datetime.now(timezone.utc) + record = { + "role": role, + "requested_container_id": container, + "started_at": started.isoformat(), + } + try: + raw = execute(["docker", "inspect", container], timeout=10) + data = strict_json(raw.encode()) + require( + type(data) is list and len(data) == 1 and type(data[0]) is dict, + f"{role} container inspect missing", + ) + record["raw"] = self._safe_state(data[0]) + return data[0] + except BaseException as error: + record["error"] = f"{type(error).__name__}: {error}" + raise + finally: + record["completed_at"] = datetime.now(timezone.utc).isoformat() + try: + self._append("state", record) + except OSError: + if "error" not in record: + raise + + def inspect(self) -> dict: + state = self._inspect_owned(self.container, "api") + require(state.get("Id") == self.container, "API container identity mismatch") + return state + + def check(self) -> None: + state = self.inspect() + validate_state(state, self.project, self.implementation, self.identity) + require( + state.get("Config", {}).get("Healthcheck", {}).get("Test") == ["NONE"], + "measured API healthcheck must be disabled", + ) + + def _check_postgres(self) -> dict: + state = self._inspect_owned(self.postgres_container, "postgres") + try: + status = state["State"] + labels = state["Config"]["Labels"] + require( + state["Id"] == self.postgres_container + and re.fullmatch(r"[0-9a-f]{64}", state["Id"]) is not None, + "PostgreSQL container identity mismatch", + ) + require( + labels["com.docker.compose.project"] == self.project + and labels["com.docker.compose.service"] == "postgres", + "PostgreSQL container belongs to wrong project/service", + ) + require( + status["Running"] is True + and status["Restarting"] is False + and status["Dead"] is False + and status["OOMKilled"] is False + and status["Status"] == "running" + and status.get("Health", {}).get("Status") == "healthy", + "PostgreSQL is not healthy/running or was OOM-killed", + ) + require( + type(state["RestartCount"]) is int and state["RestartCount"] == 0, + "PostgreSQL restart detected", + ) + require( + state["HostConfig"]["RestartPolicy"]["Name"] == "no", + "PostgreSQL restart policy must be no", + ) + require( + type(state["HostConfig"].get("Tmpfs")) is dict + and bool(state["HostConfig"]["Tmpfs"].get("/var/lib/postgresql")), + "PostgreSQL data must use fresh tmpfs", + ) + mounts = state.get("Mounts", []) + require( + type(mounts) is list and all(type(mount) is dict for mount in mounts), + "PostgreSQL mount evidence unavailable", + ) + require( + not any( + str(mount.get("Destination", "")).startswith("/var/lib/postgresql") + and mount.get("Type") != "tmpfs" + for mount in mounts + ), + "PostgreSQL must not use persistent data mounts", + ) + fixture_mounts = [ + mount + for mount in mounts + if mount.get("Destination") == "/docker-entrypoint-initdb.d/001-init.sql" + ] + require( + len(fixture_mounts) == 1 + and fixture_mounts[0].get("Type") == "bind" + and fixture_mounts[0].get("RW") is False, + "PostgreSQL fixture mount must be read-only", + ) + self._parse_started_at(status["StartedAt"]) + identity = (state["Id"], status["StartedAt"]) + require( + self.postgres_identity is None or identity == self.postgres_identity, + "PostgreSQL container identity/start time changed", + ) + self.postgres_identity = identity + return state + except (KeyError, TypeError) as error: + raise BenchmarkFailure(f"incomplete PostgreSQL container state: {error}") from error + + @staticmethod + def _parse_started_at(value: str) -> datetime: + try: + require(type(value) is str and value.endswith("Z"), "missing container start time") + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + require(parsed.tzinfo is not None, "container start time must have timezone") + return parsed + except (ValueError, TypeError) as error: + raise BenchmarkFailure("invalid container start time") from error + + def _readiness_request(self, base_url, path, *, timeout): + record = { + "path": path, + "started_at": datetime.now(timezone.utc).isoformat(), + "timeout_seconds": timeout, + } + try: + response = read_response(base_url, path, timeout=timeout) + record["response"] = { + "status": response.status, + "content_type": response.content_type, + "body": response.body.decode("utf-8", errors="replace"), + } + return response + except BaseException as error: + record["error"] = f"{type(error).__name__}: {error}" + raise + finally: + record["completed_at"] = datetime.now(timezone.utc).isoformat() + try: + self._append("readiness", record) + except OSError: + if "error" not in record: + raise + + def start(self, implementation: str) -> dict: + require(not self.start_attempted, "start requires a fresh environment instance") + self.start_attempted = True + self.implementation = implementation + try: + override = self.artifacts / f"{implementation}-external-readiness.compose.yml" + override.write_text(override_text(implementation), encoding="utf-8") + runtime = self.prefix[:-2] + ["-f", str(override), *self.prefix[-2:]] + output = execute( + runtime + ["up", "--detach", "--wait", "--wait-timeout", "60", "postgres"], + timeout=120, + ) + (self.artifacts / f"{implementation}-postgres-start.log").write_text( + output, encoding="utf-8" + ) + self.postgres_container = execute( + runtime + ["ps", "--quiet", "postgres"], timeout=10 + ).strip() + require( + re.fullmatch(r"[0-9a-f]{64}", self.postgres_container) is not None, + "expected one full PostgreSQL container ID", + ) + pg_state = self._check_postgres() + psql = runtime + [ + "exec", + "-T", + "postgres", + "psql", + "-X", + "-U", + "benchmark", + "-d", + "benchmark", + "--set", + "ON_ERROR_STOP=1", + "-At", + ] + pg_version = execute(psql + ["-c", "SELECT version();"], timeout=15).strip() + require(pg_version.startswith("PostgreSQL "), "PostgreSQL version collection failed") + fixture = execute( + psql + + [ + "-F", + "|", + "-c", + "SELECT id, name, price FROM items WHERE id = 42; SELECT COUNT(*) FROM items;", + ], + timeout=15, + ) + (self.artifacts / f"{implementation}-postgres-fixture.log").write_text( + fixture, encoding="utf-8" + ) + require( + fixture.strip().splitlines() == ["42|Item 42|4200", "1"], + "PostgreSQL benchmark fixture mismatch", + ) + postgres = { + "id": pg_state["Id"], + "image_id": pg_state["Image"], + "container_started_at": pg_state["State"]["StartedAt"], + "restart_count": pg_state["RestartCount"], + "version": pg_version, + "fixture": {"id": 42, "name": "Item 42", "price": 4200, "row_count": 1}, + "tmpfs": self._safe_state(pg_state)["HostConfig"]["Tmpfs"], + "mounts": self._safe_state(pg_state)["Mounts"], + "fixture_source_sha256": hashlib.sha256( + (ROOT / "database/init.sql").read_bytes() + ).hexdigest(), + } + output = execute(runtime + ["up", "--detach", implementation], timeout=120) + (self.artifacts / f"{implementation}-api-start.log").write_text( + output, encoding="utf-8" + ) + self.container = execute( + runtime + ["ps", "--quiet", implementation], timeout=10 + ).strip() + require( + re.fullmatch(r"[0-9a-f]{64}", self.container) is not None, + "expected one full API container ID", + ) + state = self.inspect() + self.identity = validate_state(state, self.project, implementation) + self._parse_started_at(self.identity[1]) + require( + state.get("Config", {}).get("Healthcheck", {}).get("Test") == ["NONE"], + "measured API healthcheck must be disabled", + ) + readiness_started = datetime.now(timezone.utc) + self.readiness = wait_external_readiness( + "http://127.0.0.1:8080", + implementation, + self.check, + timeout_seconds=60.0, + request_timeout=2.0, + reader=self._readiness_request, + ) + self.readiness_completed_at = datetime.now(timezone.utc) + self.readiness.update( + started_at=readiness_started.isoformat(), + completed_at=self.readiness_completed_at.isoformat(), + path="/health", + ) + # Preserve the common runner's ordering: workers may still be starting + # until /health responds, even though the container is already running. + process_list = execute(["docker", "top", self.container, "-eo", "pid,args"], timeout=10) + (self.artifacts / f"{implementation}-processes.log").write_text( + process_list, encoding="utf-8" + ) + validate_processes(state, process_list, allow_health_probe=False) + self._check_postgres() + context = { + "id": self.container, + "image_id": state["Image"], + "command": [state["Path"], *state["Args"]], + "postgresql_version": pg_version, + "postgres": postgres, + "limits": { + "nano_cpus": state["HostConfig"]["NanoCpus"], + "memory_bytes": state["HostConfig"]["Memory"], + }, + "container_started_at": state["State"]["StartedAt"], + "restart_count": state["RestartCount"], + "health_policy": self.health_policy, + "connections": self.connections, + "request_timeout_seconds": self.request_timeout, + "readiness": self.readiness, + } + (self.artifacts / f"{implementation}-context.json").write_text( + json.dumps(context, indent=2) + "\n", encoding="utf-8" + ) + return context + except BaseException as error: + self._failure("start", error) + raise + + def _timing(self, started: datetime, completed: datetime) -> dict: + born = self._parse_started_at(self.identity[1]) + age = (started - born).total_seconds() + require(age >= 0, "container start time is later than host window clock") + return { + "process_age_seconds_at_start": age, + "process_age_seconds_at_completion": (completed - born).total_seconds(), + "readiness_to_window_seconds": (started - self.readiness_completed_at).total_seconds(), + "gap_since_previous_window_seconds": None + if self.previous_window_completed_at is None + else (started - self.previous_window_completed_at).total_seconds(), + } + + def measure(self, endpoint: str, duration: int, index: int) -> dict: + """Use the common oha contract for each diagnostic window, including 0/8.""" + require(type(index) is int and 0 <= index <= 8, "window index must be 0 through 8") + require(type(duration) is int and duration > 0, "window duration must be positive") + require( + endpoint in ("/json", "/db/42", "/cpu"), + "diagnostic endpoint must be /json, /db/42 or /cpu", + ) + label = endpoint.strip("/").replace("/", "-") + f"-window-{index}" + path = self.artifacts / f"{self.implementation}-{label}.json" + require( + self.identity is not None and self.readiness_completed_at is not None, + "start must complete before measurement", + ) + require( + not path.exists() and not path.with_suffix(".window.json").exists(), + "window evidence already exists", + ) + samples = [] + cpu_samples = [] + started = datetime.now(timezone.utc) + clock_started = time.monotonic() + completed = None + elapsed = None + status = "failed" + failure = None + result = None + try: + self._check_postgres() + self.check() + with ( + path.with_suffix(".stats.jsonl").open("w", encoding="utf-8") as stats_log, + path.with_suffix(".memory.jsonl").open("w", encoding="utf-8") as memory_log, + ): + + def sample(): + self.check() + sample_started = datetime.now(timezone.utc) + raw = execute( + [ + "docker", + "stats", + "--no-stream", + "--no-trunc", + "--format", + "{{json .}}", + self.container, + ], + timeout=8, + ) + sample_completed = datetime.now(timezone.utc) + # Write before parsing so malformed samples survive a failure. + stats_log.write( + json.dumps( + { + "started_at": sample_started.isoformat(), + "completed_at": sample_completed.isoformat(), + "raw": raw, + } + ) + + "\n" + ) + stats_log.flush() + memory = memory_bytes(raw, self.container) + cpu_text = strict_json(raw.encode()).get("CPUPerc") + require( + type(cpu_text) is str + and re.fullmatch(r"[0-9]+(?:\.[0-9]+)?%", cpu_text) is not None, + "Docker CPU sample has missing/invalid CPUPerc", + ) + cpu = float(cpu_text[:-1]) + require(math.isfinite(cpu), "Docker CPU sample is not finite") + samples.append(memory) + cpu_samples.append(cpu) + memory_log.write( + json.dumps( + { + "at": sample_completed.isoformat(), + "bytes": memory, + "cpu_percent": cpu, + "container_id": self.container, + } + ) + + "\n" + ) + memory_log.flush() + + started = datetime.now(timezone.utc) + clock_started = time.monotonic() + try: + output = execute( + [ + str(self.oha), + "--no-tui", + "--output-format", + "json", + "--output", + str(path), + "--http-version", + "1.1", + "--redirect", + "0", + "--disable-compression", + "-c", + str(self.connections), + "-z", + f"{duration}s", + "-w", + "-t", + f"{self.request_timeout}s", + "--connect-timeout", + "5s", + "http://127.0.0.1:8080" + endpoint, + ], + timeout=duration + self.request_timeout + 15, + tick=sample, + ) + except BaseException as error: + with contextlib.suppress(OSError): + path.with_suffix(".load.log").write_text( + f"{type(error).__name__}: {error}\n", encoding="utf-8" + ) + raise + finally: + completed = datetime.now(timezone.utc) + elapsed = time.monotonic() - clock_started + path.with_suffix(".load.log").write_text(output, encoding="utf-8") + self.check() + self._check_postgres() + require(bool(samples), "memory/CPU collection produced no samples") + require( + path.is_file() and path.stat().st_size <= 1024 * 1024, + "oha result missing or oversized", + ) + raw = path.read_bytes() + result = parse_oha(raw, duration=duration, request_timeout=self.request_timeout) + result.update( + run=max(index, 1), + window_index=index, + p95_response_time_ms=strict_json(raw)["latencyPercentiles"]["p95"] * 1000, + peak_memory_bytes=max(samples), + memory_samples=len(samples), + cpu_percent_samples=cpu_samples, + started_at=started.isoformat(), + completed_at=completed.isoformat(), + duration_wall_seconds=elapsed, + ) + result.update(self._timing(started, completed)) + status = "complete" + return result + except BaseException as error: + failure = f"{type(error).__name__}: {error}" + self._failure(label, error) + raise + finally: + if completed is None: + completed = datetime.now(timezone.utc) + elapsed = time.monotonic() - clock_started + window = { + "implementation": self.implementation, + "endpoint": endpoint, + "window_index": index, + "requested_duration_seconds": duration, + "started_at": started.isoformat(), + "completed_at": completed.isoformat(), + "duration_wall_seconds": elapsed, + "status": status, + "error": failure, + "memory_samples": len(samples), + "cpu_percent_samples": cpu_samples, + } + try: + window.update(self._timing(started, completed)) + except Exception as timing_error: + # Invalid clocks/state must never replace the failure that stopped load. + window["timing_error"] = f"{type(timing_error).__name__}: {timing_error}" + self.previous_window_completed_at = completed + try: + path.with_suffix(".window.json").write_text( + json.dumps(window, indent=2) + "\n", encoding="utf-8" + ) + except OSError: + if failure is None: + raise diff --git a/benchmark/common_warmup_screen.py b/benchmark/common_warmup_screen.py new file mode 100644 index 0000000..cbf2f78 --- /dev/null +++ b/benchmark/common_warmup_screen.py @@ -0,0 +1,322 @@ +"""Fixed, endpoint-isolated common warm-up screening protocol v1 (not publishable).""" + +import argparse +import contextlib +import copy +import math +import os +import re +import signal +import sys +import uuid +from pathlib import Path +from statistics import median + +from .contract_runner import protect_cleanup +from .contract_test import load_cases, run_contract +from .definition import PROFILE +from .healthcheck import EXTERNAL_READINESS +from .jvm_diagnostic import write_checksums +from .results import atomic_json, number, require +from .run import now + +ROOT = Path(__file__).resolve().parents[1] +CACHE_ROOT = ROOT / ".cache" / "common-warmup-screen" +PROTOCOL_ID = "common-warmup-screen-v1" +MEMBERS = ("java-spring-boot", "node-fastify", "go-gin") +ENDPOINTS = ("/json", "/db/42", "/cpu") +# Each consecutive group covers all three implementations and all three endpoints. +FIRST_ORDER = tuple( + (MEMBERS[i], ENDPOINTS[(i + offset) % 3]) for offset in range(3) for i in range(3) +) +ORDERS = (FIRST_ORDER, FIRST_ORDER[::-1]) +CANDIDATES = (5, 35, 65) +RULES = { + "candidate_cumulative_load_seconds": list(CANDIDATES), + "reference_window_indices": [6, 7, 8], + "rps_range_over_median_max": 0.05, + "p95_range_over_median_max": 0.10, + "rps_relative_reference_difference_max": 0.05, + "p95_relative_reference_difference_max": 0.10, + "reject_strict_monotonic_reference_trend": False, + "numeric_comparison_relative_roundoff_tolerance": 1e-12, + "all_endpoints_implementations_and_both_starts_required": True, +} + + +def validate_directory(path: Path) -> Path: + candidate, root = path.absolute(), CACHE_ROOT.absolute() + require( + candidate != root and candidate.is_relative_to(root), "output must be below screen cache" + ) + require(candidate.resolve().is_relative_to(root.resolve()), "output escapes screen cache") + require(".." not in candidate.parts, "parent traversal in screen path") + for component in (candidate, *candidate.parents): + require(not component.is_symlink(), "symlink in screen output path") + require(not candidate.exists(), "screen attempt already exists; never overwrite or resume") + return candidate + + +def analyze_windows(windows: list[dict]) -> dict: + require(len(windows) == 8, "exactly eight successive windows required") + values = { + field: [number(window[field], field, positive=True) for window in windows] + for field in ("requests_per_second", "p95_response_time_ms") + } + limits = {"requests_per_second": 0.05, "p95_response_time_ms": 0.10} + + def spread(series): + return (max(series) - min(series)) / median(series) + + def within_limit(value, limit): + # Inclusive decimal bounds must not fail on machine-roundoff (e.g. 2.2/2). + # The relative allowance is 1e-12 of the limit, not an extra tolerance band. + return value <= limit or math.isclose(value, limit, rel_tol=1e-12, abs_tol=0.0) + + references = {field: median(series[5:]) for field, series in values.items()} + reference_spreads = {field: spread(series[5:]) for field, series in values.items()} + reference_stable = all( + within_limit(reference_spreads[field], limits[field]) for field in values + ) + monotonic = any(s[5] < s[6] < s[7] or s[5] > s[6] > s[7] for s in values.values()) + candidates = [] + for offset, seconds in enumerate(CANDIDATES): + differences = { + field: [abs(value - references[field]) / references[field] for value in series[offset:]] + for field, series in values.items() + } + spreads = {field: spread(series[offset : offset + 3]) for field, series in values.items()} + supported = reference_stable and all( + within_limit(spreads[field], limits[field]) + and within_limit(max(differences[field]), limits[field]) + for field in values + ) + candidates.append( + { + "cumulative_load_seconds": seconds, + "first_measurement_window_indices": list(range(offset + 1, offset + 4)), + "subsequent_window_indices": list(range(offset + 1, 9)), + "first_three_range_over_median": spreads, + "subsequent_relative_reference_differences": differences, + "screen_supported": supported, + } + ) + return { + "reference_window_indices": [6, 7, 8], + "reference_medians": references, + "reference_range_over_median": reference_spreads, + "reference_stable": reference_stable, + "strict_monotonic_reference_trend": monotonic, + "candidates": candidates, + "screened_candidate_seconds": [ + c["cumulative_load_seconds"] for c in candidates if c["screen_supported"] + ], + } + + +def run_study(directory: Path, factory, *, metadata: dict, contract=run_contract) -> dict: + directory = validate_directory(directory) + for key in ("source_commit", "source_tree"): + require( + type(metadata.get(key)) is str and re.fullmatch(r"[0-9a-f]{40}", metadata[key]), + "missing source identity: " + key, + ) + directory.mkdir(parents=True) + conditions = copy.deepcopy(PROFILE) + conditions.update(runs=8, clean_starts_per_endpoint=2, endpoint_reset=True) + result = { + "schema_version": 1, + "protocol_id": PROTOCOL_ID, + "mode": "common-warmup-screen", + "official": False, + "publishable": False, + "status": "running", + "started_at": now(), + "conditions": conditions, + "criteria": copy.deepcopy(RULES), + "metadata": copy.deepcopy(metadata), + "validations": [ + {"implementation": identifier, "status": "not-started"} for identifier in MEMBERS + ], + "attempts": [ + { + "round": round_number, + "implementation": identifier, + "endpoint": endpoint, + "status": "not-started", + "windows": [], + } + for round_number, order in enumerate(ORDERS, 1) + for identifier, endpoint in order + ], + } + output = directory / "screen.json" + + def persist(): + atomic_json(output, result) + + api_ids, database_ids, validated_images = set(), set(), {} + + def execute_entry(entry, label, *, validation): + identifier = entry["implementation"] + entry.update(status="running", started_at=now()) + persist() + environment = None + failure = None + try: + environment = factory(directory / label) + require(environment.health_policy == EXTERNAL_READINESS, "external readiness required") + require( + environment.connections == 50 and environment.request_timeout == 15, + "fixed load configuration required", + ) + entry["artifact_directory"] = str(environment.artifacts.relative_to(directory)) + persist() + environment.build(identifier) + entry["container"] = environment.start(identifier) + entry["readiness"] = copy.deepcopy(environment.readiness) + persist() + environment.check() + container = entry["container"] + api_id, database_id = container["id"], container["postgres"]["id"] + require( + api_id not in api_ids and database_id not in database_ids, + "fresh API and database identities required", + ) + api_ids.add(api_id) + database_ids.add(database_id) + if not validation: + require( + container["image_id"] == validated_images[identifier], + "image differs from sacrificial contract validation", + ) + if validation: + with ( + (environment.artifacts / "contract.log").open("w") as log, + contextlib.redirect_stdout(log), + ): + checks = contract("http://127.0.0.1:8080", implementation=identifier) + require( + type(checks) is int and checks == 2 * len(load_cases()), + "shared contract incomplete", + ) + entry["contract_checks"] = checks + validated_images[identifier] = container["image_id"] + else: + entry["warmup"] = environment.measure(entry["endpoint"], 5, 0) + persist() + for index in range(1, 9): + entry["windows"].append(environment.measure(entry["endpoint"], 30, index)) + persist() + entry["analysis"] = analyze_windows(entry["windows"]) + environment.check() + persist() + except BaseException as error: + failure = error + entry.update(status="failed", error=f"{type(error).__name__}: {error}") + persist() + raise + finally: + try: + if environment is not None: + with protect_cleanup(): + environment.cleanup() + entry["cleanup"] = "passed" + except BaseException as cleanup_error: + entry.update(status="failed", cleanup="failed", cleanup_error=str(cleanup_error)) + if failure is None: + raise + finally: + entry["completed_at"] = now() + persist() + entry["status"] = "completed" + persist() + + persist() + try: + for index, entry in enumerate(result["validations"], 1): + execute_entry( + entry, f"validation-{index:02d}-{entry['implementation']}", validation=True + ) + for index, entry in enumerate(result["attempts"], 1): + execute_entry(entry, f"trace-{index:02d}-{entry['implementation']}", validation=False) + candidates = [ + seconds + for seconds in CANDIDATES + if all( + seconds in a["analysis"]["screened_candidate_seconds"] for a in result["attempts"] + ) + ] + result["assessment"] = { + "screened_candidate_seconds": candidates, + "outcome": "candidate-for-independent-confirmation" if candidates else "inconclusive", + "official_admission": False, + } + result["status"] = "completed" + except BaseException as error: + result.update(status="failed", error=f"{type(error).__name__}: {error}") + raise + finally: + result["completed_at"] = now() + persist() + write_checksums(directory) + return result + + +def main(argv=None) -> int: + from .common_warmup_environment import CommonWarmupEnvironment + from .environment import provenance, registered_pinned_versions + from .install_oha import ensure_oha + + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--compose", default="docker compose") + args = parser.parse_args(argv) + run_directory = CACHE_ROOT / ("study-" + uuid.uuid4().hex) + CACHE_ROOT.mkdir(parents=True, exist_ok=True) + bootstrap = CACHE_ROOT / (run_directory.name + "-bootstrap.json") + context = { + "protocol_id": PROTOCOL_ID, + "started_at": now(), + "status": "starting", + "official": False, + "publishable": False, + "github_run_id": os.environ.get("GITHUB_RUN_ID"), + "github_run_attempt": os.environ.get("GITHUB_RUN_ATTEMPT"), + } + atomic_json(bootstrap, context) + + def interrupted(signum, _frame): + raise KeyboardInterrupt(f"received signal {signum}") + + previous = {sig: signal.signal(sig, interrupted) for sig in (signal.SIGINT, signal.SIGTERM)} + try: + oha = ensure_oha() + metadata = provenance(oha) + versions = registered_pinned_versions() + metadata["versions"] = {identifier: versions[identifier] for identifier in MEMBERS} + metadata.update( + github_run_id=context["github_run_id"], + github_run_attempt=context["github_run_attempt"], + jit_gc_evidence="Unavailable: unchanged production runtimes; no JIT/GC instrumentation.", + ) + run_study( + run_directory, + lambda root: CommonWarmupEnvironment(oha, root, compose=args.compose), + metadata=metadata, + ) + context["status"] = "completed" + print(f"Common warm-up screen completed: {run_directory}; no official results published.") + return 0 + except (Exception, KeyboardInterrupt) as error: + context.update(status="failed", error=f"{type(error).__name__}: {error}") + print(f"Common warm-up screen failed without retry: {error}", file=sys.stderr) + return 1 + finally: + context["completed_at"] = now() + atomic_json(bootstrap, context) + for sig, handler in previous.items(): + signal.signal(sig, handler) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/docs/COMMON-WARMUP-SCREEN.md b/docs/COMMON-WARMUP-SCREEN.md new file mode 100644 index 0000000..485b19b --- /dev/null +++ b/docs/COMMON-WARMUP-SCREEN.md @@ -0,0 +1,115 @@ +# Common warm-up screen v1 (Issue #73) + +This predeclared, non-publishing diagnostic follows the initial +[JVM diagnostic](JVM-DIAGNOSTIC.md). It does not replace that study or retrospectively +change its interpretation. Protocol ID: `common-warmup-screen-v1`. Java stays +registered-only; official profiles, cohorts, runtime tuning, pins and results are +unchanged. The output is a candidate for independent confirmation, not proof of +steady state or authorization for a nine-implementation comparison. + +## Fixed experiment + +Java / Spring Boot, Node / Fastify and Go / Gin each receive two fresh starts for +**each** of `/json`, `/db/42` and `/cpu`: 18 traces. All traces run sequentially on +one host. Each owns a fresh Compose project, API process and PostgreSQL tmpfs +fixture. Host caches and image-build caches are not reset, and process restarts +are not independent hosts. The first-round order is: + +1. Java JSON, Node DB, Go CPU +2. Java DB, Node CPU, Go JSON +3. Java CPU, Node JSON, Go DB + +The second round reverses all nine entries. This balances simple early/late +positions, but two starts do not eliminate host drift or establish confidence +intervals. All attempts are retained, including failed and unstarted entries. + +Before measurement, one separate sacrificial instance per implementation passes +the full shared two-round correctness contract. Those instances are cleaned up; +measured instances never receive that contract or another endpoint's load. The +measured image must match its validated image, and API/database identities must +be distinct across all instances. Only logged `/health` readiness requests occur +before measured load. They can initialize shared application code: these are not +untouched cold starts. PostgreSQL identity and fixture checks occur before API +startup. The resource caps remain 1 CPU and 512 MiB, pool maximum 10, 50 connections, +HTTP/1.1 and the existing 15-second request timeout. Recurring API healthchecks are +disabled in the owned projects only. No application or dependency tuning occurs. + +Each trace sends five seconds of initial load, then eight 30-second windows. +All 162 invocations (18 warm-ups and 144 windows) are retained. Separate oha +invocations drain requests and have state-check, sampling and orchestration gaps. +Window timestamps, elapsed time, process age and gaps expose this limitation; +these are nominal cumulative-load budgets, not exact process-age intervals or a +single continuous load stream. + +## Predeclared screening rule + +Candidates are 5, 35 and 65 seconds of nominal cumulative load. Their first three +measurement windows are respectively 1–3, 2–4 and 3–5. The reference is always +windows 6–8 (nominal load 155–245 seconds), disjoint from those candidate windows. +There is no search outside this grid, adaptive extension or automatic retry. + +For each trace, all warm-ups/windows must have valid raw evidence, zero transport +or non-200 errors, and passing resource/identity/restart checks. Then: + +- Reference throughput range / median is at most 5%; p95 range / median at most 10% +- Every window from the candidate's first measurement through window 8 is within + ±5% throughput and ±10% p95 of the respective reference median +- The candidate's first-three throughput range / median is at most 5%; its p95 + range / median is at most 10% + +A candidate is screened in only if **all 18 traces** pass: both starts, every +endpoint and all three implementations. Otherwise it remains unsupported by this +screen. No passing candidate means **inconclusive**; it is not proof that a longer +warm-up would solve the problem. Report every window and all candidate outcomes. +Inclusive comparisons allow only machine-roundoff-scale relative error (1e-12 +of the threshold), so decimal boundary values are not rejected accidentally. + +These are practical engineering tolerances, inherited in magnitude from the first +study, not statistically established equivalence margins. A strictly monotonic +reference sequence is reported but does not by itself reject arbitrarily small +noise. This deliberately differs from the first study's strict monotonic rule; +that study and its original failures remain intact. The new rule is fixed before +new data collection. Even a passing candidate needs independent confirmation +using an explicitly specified continuous warm-up, and eventual coverage of all +nine implementations under separately approved conditions. + +## Evidence and limits + +The screen retains source commit/tree, host and pinned version information, lock +hashes, image/container/database identities, fixture checks, readiness probes, +allowlisted state snapshots, process checks, CPU/memory samples, every raw oha +output, UTC timing, contract logs and failure/cleanup evidence. Inspection records +exclude credentials. A checksum manifest is an integrity aid, not a signature or +proof that missing files never existed. + +Docker stats and checks consume host resources. Their overhead is visible in +timing but is not quantified by an uninstrumented comparison. Sampled memory is +not peak RSS. No JIT/GC instrumentation is enabled; neither compilation nor GC can +be established as the cause of a changing window. One JVM process does not mean +one request-handling thread; concurrency semantics remain as documented in the +original guide and implementation sources. + +## Budget, execution and stopping + +The separate **Common warm-up screen v1** workflow is manual-only on the trusted +default branch, with read-only repository permissions and no publication job. +Code merge does not imply permission for future runs. The approved scope is one +additional run, not automatic repetition or a future official comparison. + +Configured load: 18 × (5 + 8 × 30) = 4,410 seconds = 73.5 minutes. Approximately 90–110 +minutes including setup is an estimate, not a completion guarantee. The job cap +is 120 minutes; workload is terminated after 104 minutes with a 120-second kill grace. +Checkout and Python setup each have a 3-minute cap, and artifact upload has a +7-minute cap. Cleanup and upload time are reserved. Failure or time exhaustion +aborts remaining work, journals it and attempts scoped cleanup; there is no rerun +or threshold relaxation. Worker loss, SIGKILL or upload failure can still leave +incomplete evidence, which must be disclosed. + +Evidence lives only in `.cache/common-warmup-screen/`, uploaded even on failure as +`common-warmup-screen--` with 90-day retention. Unique directories +cannot be resumed or overwritten. The schema explicitly marks output +`official: false` and `publishable: false`; official report validation rejects it. +CI tests use synthetic data and do not execute this study's load. Audit raw data +and completeness before recording the outcome in Issue #73. Nine-implementation +confirmation, official-profile changes and result publication require a later +plan and separate approval; they are not included in this run's budget. diff --git a/tests/test_benchmark_common_warmup_environment.py b/tests/test_benchmark_common_warmup_environment.py new file mode 100644 index 0000000..e5fa6e9 --- /dev/null +++ b/tests/test_benchmark_common_warmup_environment.py @@ -0,0 +1,529 @@ +"""Common screening evidence uses command doubles, never starts Docker or load.""" + +import copy +import importlib +import importlib.util +import json +import math +import tempfile +import unittest +from datetime import datetime +from pathlib import Path +from unittest.mock import patch + +from benchmark import environment +from benchmark.contract_test import ContractFailure, Response +from benchmark.results import BenchmarkFailure + +CID = "a" * 64 +PGID = "b" * 64 +FIXTURE = Path(__file__).parent / "fixtures" / "oha-1.16.0" / "timed.json" + + +class CommonWarmupEnvironmentTests(unittest.TestCase): + def setUp(self): + self.assertIsNotNone( + importlib.util.find_spec("benchmark.common_warmup_environment"), + "separate common warmup screening environment is required", + ) + self.module = importlib.import_module("benchmark.common_warmup_environment") + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.env = self.module.CommonWarmupEnvironment(Path("/fake/oha"), Path(self.directory.name)) + self.api = self.state(CID, "go-gin") + self.pg = self.state(PGID, "postgres") + self.pg["State"]["Health"] = {"Status": "healthy", "Log": [{"Output": "secret"}]} + self.pg["HostConfig"]["NanoCpus"] = 0 + self.pg["HostConfig"]["Memory"] = 0 + self.commands = [] + self.requests = [] + self.payload = FIXTURE.read_bytes() + self.stats = json.dumps({"ID": CID, "CPUPerc": "73.25%", "MemUsage": "12MiB / 512MiB"}) + self.fixture_output = "42|Item 42|4200\n1\n" + self.load_failure = None + self.readiness_failure = None + self.api_inspections = 0 + self.pg_inspections = 0 + self.change_after_load = None + self.loaded = False + self.process_list = "PID COMMAND\n42 /go-gin\n" + self.process_requires_ready = False + self.ready_observed = False + + def state(self, cid, service): + value = { + "Id": cid, + "Image": "sha256:" + "c" * 64, + "Path": "/go-gin" if service != "postgres" else "postgres", + "Args": [], + "RestartCount": 0, + "State": { + "Status": "running", + "Running": True, + "Restarting": False, + "Dead": False, + "OOMKilled": False, + "StartedAt": "2026-01-01T00:00:00Z", + }, + "HostConfig": { + "NanoCpus": 1000000000, + "Memory": 536870912, + "RestartPolicy": {"Name": "no"}, + }, + "Config": { + "Labels": { + "com.docker.compose.project": self.env.project, + "com.docker.compose.service": service, + "secret-label": "secret", + }, + "Healthcheck": {"Test": ["NONE"]}, + "Env": ["DATABASE_PASSWORD=secret-never-persist"], + }, + } + if service == "postgres": + value["HostConfig"]["Tmpfs"] = {"/var/lib/postgresql": "rw,size=512m,mode=1777"} + value["Mounts"] = [ + { + "Type": "bind", + "Source": str(environment.ROOT / "database/init.sql"), + "Destination": "/docker-entrypoint-initdb.d/001-init.sql", + "RW": False, + } + ] + return value + + def execute(self, args, *, timeout, tick=None, **kwargs): + self.commands.append((args, timeout)) + if args[:2] == ["docker", "inspect"]: + if args[2] == CID: + self.api_inspections += 1 + value = copy.deepcopy(self.api) + if self.loaded and self.change_after_load: + value["State"].update(self.change_after_load) + return json.dumps([value]) + if args[2] == PGID: + self.pg_inspections += 1 + return json.dumps([self.pg]) + raise AssertionError("unexpected inspect target") + if "ps" in args and "--quiet" in args: + return PGID if args[-1] == "postgres" else CID + if "psql" in args: + return ( + "PostgreSQL 18.6 on test\n" if "SELECT version();" in args else self.fixture_output + ) + if args[:2] == ["docker", "top"]: + if self.process_requires_ready and not self.ready_observed: + return "PID COMMAND\n42 startup-wrapper\n" + return self.process_list + if args[:2] == ["docker", "stats"]: + return self.stats + if "--output" in args: + Path(args[args.index("--output") + 1]).write_bytes(self.payload) + tick() + self.loaded = True + if self.load_failure: + raise self.load_failure + return "oha stdout\n" + if "logs" in args: + return "owned log\n" + return "" + + def reader(self, base_url, path, *, timeout): + self.requests.append((base_url, path, timeout)) + if self.readiness_failure: + raise self.readiness_failure + self.ready_observed = True + return Response(200, "application/json", b'{"status":"ok"}') + + def start(self): + with ( + patch.object(self.module, "execute", side_effect=self.execute), + patch.object(self.module, "read_response", side_effect=self.reader), + ): + return self.env.start("go-gin") + + def measure(self, index=8, endpoint="/json"): + with patch.object(self.module, "execute", side_effect=self.execute): + return self.env.measure(endpoint, 1, index) + + def evidence(self, name): + return self.env.artifacts / ("go-gin-" + name) + + def test_cleanup_and_runtime_policy_remain_inherited_without_tuning(self): + self.assertIs( + self.module.CommonWarmupEnvironment.cleanup, environment.DockerEnvironment.cleanup + ) + self.assertEqual((self.env.connections, self.env.request_timeout), (50, 15)) + self.assertEqual(self.env.health_policy, "external-readiness") + self.assertFalse(self.env.audit_health_events) + for kwargs in ( + {"connections": 1}, + {"request_timeout": 30}, + {"health_policy": "container-healthcheck"}, + {"audit_health_events": True}, + ): + with self.subTest(kwargs=kwargs), self.assertRaises(BenchmarkFailure): + self.module.CommonWarmupEnvironment( + Path("/fake/oha"), Path(self.directory.name), **kwargs + ) + + def test_start_proves_postgres_fixture_before_starting_api_and_only_reads_health(self): + context = self.start() + commands = [args for args, _ in self.commands] + fixture = next( + i for i, args in enumerate(commands) if "psql" in args and "items" in args[-1] + ) + api_start = next( + i for i, args in enumerate(commands) if "up" in args and args[-1] == "go-gin" + ) + self.assertLess(fixture, api_start) + self.assertEqual([path for _, path, _ in self.requests], ["/health"]) + self.assertEqual(context["postgres"]["id"], PGID) + self.assertEqual( + context["postgres"]["fixture"], + {"id": 42, "name": "Item 42", "price": 4200, "row_count": 1}, + ) + self.assertEqual(context["id"], CID) + self.assertEqual(context["readiness"]["attempts"], 1) + self.assertGreaterEqual(context["readiness"]["duration_seconds"], 0) + self.assertEqual(context["limits"], {"nano_cpus": 1000000000, "memory_bytes": 536870912}) + self.assertEqual(self.evidence("postgres-fixture.log").read_text(), self.fixture_output) + self.assertEqual(json.loads(self.evidence("context.json").read_text()), context) + readiness = json.loads(self.evidence("readiness.jsonl").read_text()) + self.assertEqual(readiness["path"], "/health") + self.assertEqual(readiness["response"]["body"], '{"status":"ok"}') + + def test_raw_state_evidence_is_allowlisted_and_keeps_timing_and_identities(self): + self.start() + self.measure() + saved = self.evidence("state.jsonl").read_text() + for forbidden in ( + "secret-never-persist", + "DATABASE_PASSWORD", + '"Env"', + "secret-label", + '"Log"', + ): + self.assertNotIn(forbidden, saved) + records = [json.loads(line) for line in saved.splitlines()] + self.assertEqual({r["role"] for r in records}, {"api", "postgres"}) + self.assertEqual({r["raw"]["Id"] for r in records}, {CID, PGID}) + self.assertTrue(all("StartedAt" in r["raw"]["State"] for r in records)) + self.assertTrue( + all( + datetime.fromisoformat(r["started_at"]) <= datetime.fromisoformat(r["completed_at"]) + for r in records + ) + ) + + def test_zero_and_eight_use_identical_common_load_arguments_and_raw_cpu_memory(self): + self.start() + for index in (0, 8): + result = self.measure(index) + self.assertEqual(result["window_index"], index) + self.assertEqual(result["cpu_percent_samples"], [73.25]) + self.assertEqual(result["peak_memory_bytes"], 12 * 1024**2) + self.assertEqual( + result["p95_response_time_ms"], + json.loads(self.payload)["latencyPercentiles"]["p95"] * 1000, + ) + command, timeout = [call for call in self.commands if "--output" in call[0]][-1] + self.assertEqual( + command, + [ + "/fake/oha", + "--no-tui", + "--output-format", + "json", + "--output", + str(self.evidence(f"json-window-{index}.json")), + "--http-version", + "1.1", + "--redirect", + "0", + "--disable-compression", + "-c", + "50", + "-z", + "1s", + "-w", + "-t", + "15s", + "--connect-timeout", + "5s", + "http://127.0.0.1:8080/json", + ], + ) + self.assertEqual(timeout, 31) + raw = json.loads(self.evidence(f"json-window-{index}.stats.jsonl").read_text()) + self.assertEqual(raw["raw"], self.stats) + self.assertEqual(self.evidence(f"json-window-{index}.json").read_bytes(), self.payload) + self.assertEqual(len(self.requests), 1) + + def test_process_age_readiness_gap_and_between_window_gap_are_explicit(self): + context = self.start() + first = self.measure(0) + second = self.measure(1) + self.assertIsNone(first["gap_since_previous_window_seconds"]) + self.assertGreaterEqual(first["readiness_to_window_seconds"], 0) + expected_gap = ( + datetime.fromisoformat(second["started_at"]) + - datetime.fromisoformat(first["completed_at"]) + ).total_seconds() + self.assertAlmostEqual(second["gap_since_previous_window_seconds"], expected_gap) + expected_age = ( + datetime.fromisoformat(first["started_at"]) + - datetime.fromisoformat(context["container_started_at"].replace("Z", "+00:00")) + ).total_seconds() + self.assertAlmostEqual(first["process_age_seconds_at_start"], expected_age) + self.assertGreaterEqual( + first["process_age_seconds_at_completion"], first["process_age_seconds_at_start"] + ) + saved = json.loads(self.evidence("json-window-0.window.json").read_text()) + for field in ( + "process_age_seconds_at_start", + "process_age_seconds_at_completion", + "gap_since_previous_window_seconds", + "readiness_to_window_seconds", + ): + self.assertEqual(saved[field], first[field]) + + def test_invalid_indexes_duration_or_endpoint_start_no_commands(self): + for index in (-1, 9, True, 1.0): + with self.subTest(index=index), self.assertRaises(BenchmarkFailure): + self.measure(index) + for duration in (0, -1, True, 1.0): + with self.subTest(duration=duration), self.assertRaises(BenchmarkFailure): + self.env.measure("/json", duration, 0) + with self.assertRaises(BenchmarkFailure): + self.measure(endpoint="/../../escape") + self.assertEqual(self.commands, []) + + def test_bad_fixture_preserves_raw_output_and_never_starts_api(self): + self.fixture_output = "42|Wrong|4200\n2\n" + with self.assertRaisesRegex(BenchmarkFailure, "fixture"): + self.start() + self.assertEqual(self.evidence("postgres-fixture.log").read_text(), self.fixture_output) + self.assertFalse(any("up" in args and args[-1] == "go-gin" for args, _ in self.commands)) + self.assertIn("fixture", self.evidence("start-failure.log").read_text()) + + def test_postgres_requires_tmpfs_and_rejects_persistent_data_mount(self): + self.pg["Mounts"].append( + {"Type": "volume", "Destination": "/var/lib/postgresql/18/docker", "RW": True} + ) + with self.assertRaisesRegex(BenchmarkFailure, "persistent"): + self.start() + self.assertFalse(any("up" in args and args[-1] == "go-gin" for args, _ in self.commands)) + + def test_postgres_requires_readonly_fixture_mount(self): + self.pg["Mounts"][0]["RW"] = True + with self.assertRaisesRegex(BenchmarkFailure, "fixture mount"): + self.start() + records = [ + json.loads(line) for line in self.evidence("state.jsonl").read_text().splitlines() + ] + self.assertEqual(records[0]["raw"]["Mounts"][0]["RW"], True) + + def test_postgres_identity_health_restart_or_project_mismatch_fails_closed(self): + for field, value in (("RestartCount", 1), ("Id", "c" * 64)): + with self.subTest(field=field): + old = self.pg[field] + self.pg[field] = value + with self.assertRaises(BenchmarkFailure): + self.start() + self.pg[field] = old + # Every failed attempt owns a fresh environment. + self.env = self.module.CommonWarmupEnvironment( + Path("/fake/oha"), Path(self.directory.name) + ) + self.api = self.state(CID, "go-gin") + self.pg = self.state(PGID, "postgres") + self.pg["State"]["Health"] = {"Status": "healthy"} + self.pg["State"]["Health"] = {"Status": "unhealthy"} + with self.assertRaises(BenchmarkFailure): + self.start() + self.assertTrue(self.evidence("state.jsonl").is_file()) + + def test_api_resource_failure_retains_sanitized_state_before_validation(self): + self.api["HostConfig"]["Memory"] = 0 + with self.assertRaisesRegex(BenchmarkFailure, "512 MiB"): + self.start() + records = [ + json.loads(line) for line in self.evidence("state.jsonl").read_text().splitlines() + ] + self.assertEqual( + [r for r in records if r["role"] == "api"][-1]["raw"]["HostConfig"]["Memory"], 0 + ) + self.assertEqual(self.requests, []) + + def test_initial_api_inspect_must_match_requested_container_id(self): + self.api["Id"] = "d" * 64 + with self.assertRaisesRegex(BenchmarkFailure, "API container identity"): + self.start() + self.assertEqual(self.requests, []) + + def test_enabled_healthcheck_rejected(self): + self.api["Config"]["Healthcheck"]["Test"] = ["CMD", "/go-gin", "healthcheck"] + with self.assertRaisesRegex(BenchmarkFailure, "healthcheck"): + self.start() + self.assertEqual(self.requests, []) + + def test_extra_api_process_rejected_and_process_evidence_retained(self): + self.process_list += "43 unexpected-worker\n" + with self.assertRaisesRegex(BenchmarkFailure, "server process"): + self.start() + self.assertEqual(self.evidence("processes.log").read_text(), self.process_list) + + def test_process_check_waits_for_readiness_so_worker_startup_can_finish(self): + self.process_requires_ready = True + self.start() + self.assertTrue(self.ready_observed) + self.assertEqual(self.evidence("processes.log").read_text(), self.process_list) + + def test_readiness_failure_is_logged_with_no_full_contract_calls(self): + self.readiness_failure = ContractFailure("bad response framing") + with self.assertRaisesRegex(ContractFailure, "bad response framing"): + self.start() + record = json.loads(self.evidence("readiness.jsonl").read_text()) + self.assertEqual(record["path"], "/health") + self.assertIn("bad response framing", record["error"]) + self.assertEqual([p for _, p, _ in self.requests], ["/health"]) + self.assertIn("owned log", self.evidence("start-container.log").read_text()) + + def test_failed_load_retains_stats_oha_timing_and_scoped_logs(self): + self.start() + self.load_failure = BenchmarkFailure("original partial failure") + with self.assertRaisesRegex(BenchmarkFailure, "original partial failure"): + self.measure() + window = json.loads(self.evidence("json-window-8.window.json").read_text()) + self.assertEqual(window["status"], "failed") + self.assertGreaterEqual(window["process_age_seconds_at_start"], 0) + self.assertEqual(self.evidence("json-window-8.json").read_bytes(), self.payload) + self.assertEqual( + json.loads(self.evidence("json-window-8.stats.jsonl").read_text())["raw"], self.stats + ) + self.assertIn( + "original partial failure", self.evidence("json-window-8.load.log").read_text() + ) + self.assertIn("owned log", self.evidence("json-window-8-container.log").read_text()) + for args, timeout in self.commands: + if "logs" in args: + self.assertIn(self.env.project, args) + self.assertLessEqual(timeout, 15) + + def test_invalid_raw_cpu_and_oha_results_are_retained(self): + self.start() + self.stats = json.dumps({"ID": CID, "CPUPerc": "NaN%", "MemUsage": "12MiB / 512MiB"}) + with self.assertRaises(BenchmarkFailure): + self.measure(0) + self.assertEqual( + json.loads(self.evidence("json-window-0.stats.jsonl").read_text())["raw"], self.stats + ) + self.stats = json.dumps({"ID": CID, "CPUPerc": "1%", "MemUsage": "12MiB / 512MiB"}) + value = json.loads(self.payload) + value["latencyPercentiles"]["p95"] = math.inf + self.payload = json.dumps(value).encode() + with self.assertRaises(BenchmarkFailure): + self.measure(1) + self.assertEqual(self.evidence("json-window-1.json").read_bytes(), self.payload) + + def test_restart_after_load_invalidates_retained_window(self): + self.start() + self.change_after_load = {"StartedAt": "2026-01-02T00:00:00Z"} + with self.assertRaisesRegex(BenchmarkFailure, "identity"): + self.measure() + self.assertEqual( + json.loads(self.evidence("json-window-8.window.json").read_text())["status"], "failed" + ) + + def test_postgres_checked_at_boundaries_but_never_inside_stats_sampling(self): + self.start() + before_pg = self.pg_inspections + self.measure() + self.assertEqual(self.pg_inspections - before_pg, 2) + self.assertGreater(self.api_inspections, self.pg_inspections) + + def test_postgres_restart_between_windows_fails_before_another_load(self): + self.start() + self.measure(0) + self.pg["RestartCount"] = 1 + with self.assertRaisesRegex(BenchmarkFailure, "PostgreSQL restart"): + self.measure(1) + self.assertEqual(len([args for args, _ in self.commands if "--output" in args]), 1) + self.assertEqual( + json.loads(self.evidence("json-window-1.window.json").read_text())["status"], "failed" + ) + + def test_invalid_process_age_does_not_mask_original_state_failure(self): + self.start() + self.env.identity = (CID, "2999-01-01T00:00:00Z") + with self.assertRaisesRegex(BenchmarkFailure, "identity/start time changed"): + self.measure() + window = json.loads(self.evidence("json-window-8.window.json").read_text()) + self.assertEqual(window["status"], "failed") + self.assertIn("container start time", window["timing_error"]) + + def test_evidence_disk_failure_does_not_mask_failed_load(self): + self.start() + self.load_failure = BenchmarkFailure("original load failure") + original = Path.write_text + + def fail_write(path, text, **kwargs): + if path.suffix == ".log" or path.name.endswith(".window.json"): + raise OSError("disk full") + return original(path, text, **kwargs) + + with patch.object(Path, "write_text", fail_write): + with self.assertRaisesRegex(BenchmarkFailure, "original load failure"): + self.measure() + + def test_failed_build_retains_error_and_original_cause_when_logs_fail(self): + with patch.object(self.module, "execute", side_effect=BenchmarkFailure("compiler details")): + with self.assertRaisesRegex(BenchmarkFailure, "compiler details"): + self.env.build("go-gin") + self.assertIn("compiler details", self.evidence("build.log").read_text()) + self.assertIn("logs unavailable", self.evidence("build-container.log").read_text()) + + def test_readiness_retries_are_individually_logged(self): + from benchmark.healthcheck import wait_external_readiness + + times = iter(i / 100 for i in range(100)) + + def wait(*args, **kwargs): + return wait_external_readiness( + *args, **kwargs, clock=lambda: next(times), sleep=lambda _: None + ) + + responses = [ + ContractFailure("transport: connection refused"), + Response(200, "application/json", b'{"status":"ok"}'), + ] + with ( + patch.object(self.module, "execute", side_effect=self.execute), + patch.object(self.module, "read_response", side_effect=responses), + patch.object(self.module, "wait_external_readiness", side_effect=wait), + ): + context = self.env.start("go-gin") + records = [ + json.loads(line) for line in self.evidence("readiness.jsonl").read_text().splitlines() + ] + self.assertEqual(context["readiness"]["attempts"], 2) + self.assertEqual(len(records), 2) + self.assertIn("connection refused", records[0]["error"]) + self.assertEqual(records[1]["response"]["status"], 200) + + def test_window_evidence_cannot_be_overwritten_and_start_requires_fresh_instance(self): + self.start() + self.measure() + original = self.evidence("json-window-8.window.json").read_bytes() + count = len(self.commands) + with self.assertRaisesRegex(BenchmarkFailure, "already"): + self.measure() + self.assertEqual(self.evidence("json-window-8.window.json").read_bytes(), original) + self.assertEqual(len(self.commands), count) + with self.assertRaisesRegex(BenchmarkFailure, "fresh"): + self.start() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_benchmark_common_warmup_screen.py b/tests/test_benchmark_common_warmup_screen.py new file mode 100644 index 0000000..1154b8a --- /dev/null +++ b/tests/test_benchmark_common_warmup_screen.py @@ -0,0 +1,298 @@ +"""Synthetic tests for the separately versioned, endpoint-isolated screen.""" + +import importlib +import importlib.util +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from benchmark.contract_test import load_cases +from benchmark.healthcheck import EXTERNAL_READINESS +from benchmark.results import BenchmarkFailure + + +class Environment: + health_policy = EXTERNAL_READINESS + connections = 50 + request_timeout = 15 + + def __init__(self, root, index, fail=None): + self.artifacts = root / "raw" + self.artifacts.mkdir(parents=True) + self.index = index + self.fail = fail + self.events = [] + self.readiness = {"path": "/health", "attempts": 1} + + def build(self, identifier): + self.events.append(("build", identifier)) + if self.fail == "build": + raise BenchmarkFailure("build failed") + + def start(self, identifier): + self.events.append(("start", identifier)) + return { + "id": f"{self.index:064x}", + "image_id": "sha256:" + identifier, + "postgres": {"id": f"{100 + self.index:064x}"}, + } + + def check(self): + self.events.append("check") + + def measure(self, endpoint, duration, index): + self.events.append((endpoint, duration, index)) + if self.fail == "measure": + raise BenchmarkFailure("measure failed") + if self.fail == "interrupt": + raise KeyboardInterrupt() + return { + "requests_per_second": 100, + "p95_response_time_ms": 2, + "successful_requests": 100, + "elapsed_seconds": duration, + } + + def cleanup(self): + self.events.append("cleanup") + if self.fail == "cleanup": + raise BenchmarkFailure("cleanup failed") + + +class ScreenTests(unittest.TestCase): + def setUp(self): + self.assertIsNotNone(importlib.util.find_spec("benchmark.common_warmup_screen")) + self.module = importlib.import_module("benchmark.common_warmup_screen") + temp = tempfile.TemporaryDirectory() + self.addCleanup(temp.cleanup) + self.cache = Path(temp.name) + p = patch.object(self.module, "CACHE_ROOT", self.cache) + p.start() + self.addCleanup(p.stop) + self.output = self.cache / "run" + self.envs = [] + self.contract_envs = [] + self.metadata = {"source_commit": "a" * 40, "source_tree": "b" * 40} + + def run_study(self, fail=None, count=None): + def factory(root): + env = Environment(root, len(self.envs) + 1, fail) + self.envs.append(env) + return env + + def contract(*args, **kwargs): + self.contract_envs.append(self.envs[-1]) + return 2 * len(load_cases()) if count is None else count + + return self.module.run_study( + self.output, factory, metadata=self.metadata, contract=contract + ) + + def test_fixed_schedule_sacrificial_contracts_isolated_endpoints_and_no_publication(self): + result = self.run_study() + self.assertEqual(len(self.envs), 21) + self.assertEqual(self.contract_envs, self.envs[:3]) + self.assertEqual(len(result["validations"]), 3) + self.assertEqual(len(result["attempts"]), 18) + first = [(a["implementation"], a["endpoint"]) for a in result["attempts"][:9]] + second = [(a["implementation"], a["endpoint"]) for a in result["attempts"][9:]] + self.assertEqual(first[::-1], second) + self.assertEqual(len(set(first)), 9) + for env in self.envs: + self.assertEqual(env.events[-1], "cleanup") + for env in self.envs[:3]: + self.assertFalse(any(isinstance(e, tuple) and e[0].startswith("/") for e in env.events)) + for env, attempt in zip(self.envs[3:], result["attempts"]): + loads = [e for e in env.events if isinstance(e, tuple) and e[0].startswith("/")] + self.assertEqual( + loads, [(attempt["endpoint"], 5 if i == 0 else 30, i) for i in range(9)] + ) + self.assertEqual(result["status"], "completed") + self.assertEqual(result["protocol_id"], "common-warmup-screen-v1") + self.assertEqual(result["assessment"]["screened_candidate_seconds"], [5, 35, 65]) + self.assertFalse(result["official"]) + self.assertFalse(result["publishable"]) + self.assertEqual(result, json.loads((self.output / "screen.json").read_text())) + self.assertTrue((self.output / "checksums.json").exists()) + from benchmark.report import validate_report + + with self.assertRaises(BenchmarkFailure): + validate_report(result) + + def test_one_second_round_failure_prevents_common_candidate(self): + envs = [] + + def factory(root): + env = Environment(root, len(envs) + 1) + original = env.measure + + def measure(endpoint, duration, index): + result = original(endpoint, duration, index) + if env.index == 21 and index == 5: + result["requests_per_second"] = 200 + return result + + env.measure = measure + envs.append(env) + return env + + result = self.module.run_study( + self.output, + factory, + metadata=self.metadata, + contract=lambda *a, **k: 2 * len(load_cases()), + ) + self.assertEqual(result["assessment"]["outcome"], "inconclusive") + self.assertEqual(result["assessment"]["screened_candidate_seconds"], []) + self.assertTrue(result["attempts"][0]["analysis"]["screened_candidate_seconds"]) + self.assertFalse(result["attempts"][-1]["analysis"]["screened_candidate_seconds"]) + + def test_failure_stops_and_retains_unstarted_work_without_retry(self): + for fail in ("build", "measure", "cleanup", "interrupt"): + with self.subTest(fail=fail): + self.output = self.cache / fail + self.envs = [] + with self.assertRaises((BenchmarkFailure, KeyboardInterrupt)): + self.run_study(fail=fail) + result = json.loads((self.output / "screen.json").read_text()) + self.assertEqual(result["status"], "failed") + self.assertNotIn("assessment", result) + self.assertLessEqual(len(self.envs), 4) + self.assertEqual(self.envs[-1].events[-1], "cleanup") + self.assertTrue((self.output / "checksums.json").exists()) + self.assertTrue(all(a["status"] == "not-started" for a in result["attempts"][1:])) + + def test_contract_count_fails_before_any_measured_trace(self): + with self.assertRaises(BenchmarkFailure): + self.run_study(count=True) + self.assertEqual(len(self.envs), 1) + self.assertEqual(self.envs[0].events[-1], "cleanup") + + def test_no_overwrite_escape_symlink_or_missing_provenance(self): + self.run_study() + before = (self.output / "screen.json").read_bytes() + with self.assertRaises(BenchmarkFailure): + self.run_study() + self.assertEqual(before, (self.output / "screen.json").read_bytes()) + (self.cache / "link").symlink_to(self.cache, target_is_directory=True) + for path in (self.cache, self.cache / ".." / "outside", self.cache / "link" / "new"): + with self.assertRaises(BenchmarkFailure): + self.module.validate_directory(path) + self.output = self.cache / "missing-source" + self.metadata = {} + with self.assertRaises(BenchmarkFailure): + self.run_study() + + def test_trace_rejects_reused_identity_or_different_validated_image(self): + for kind in ("api", "database", "image"): + with self.subTest(kind=kind): + self.output = self.cache / kind + envs = [] + + def factory(root): + env = Environment(root, len(envs) + 1) + original = env.start + + def start(identifier): + result = original(identifier) + if len(envs) > 3: + if kind == "api": + result["id"] = f"{1:064x}" + elif kind == "database": + result["postgres"]["id"] = f"{101:064x}" + else: + result["image_id"] = "changed" + return result + + env.start = start + envs.append(env) + return env + + with self.assertRaises(BenchmarkFailure): + self.module.run_study( + self.output, + factory, + metadata=self.metadata, + contract=lambda *a, **k: 2 * len(load_cases()), + ) + self.assertEqual(len(envs), 4) + self.assertFalse( + any(isinstance(e, tuple) and e[0].startswith("/") for e in envs[-1].events) + ) + + def test_analysis_exact_candidates_reference_and_all_later_windows(self): + def windows(rps, p95=None): + return [ + {"requests_per_second": r, "p95_response_time_ms": p} + for r, p in zip(rps, p95 or [2] * 8) + ] + + analyze = self.module.analyze_windows + result = analyze(windows([50, 80, 100, 100, 100, 100, 100, 100])) + self.assertEqual(result["screened_candidate_seconds"], [65]) + self.assertEqual(result["reference_window_indices"], [6, 7, 8]) + self.assertFalse( + analyze(windows([100, 100, 100, 100, 120, 100, 100, 100]))["screened_candidate_seconds"] + ) + # Tiny strict monotonic noise is reported, not automatically disqualifying. + self.assertEqual( + analyze(windows([100] * 5 + [99, 100, 101]))["screened_candidate_seconds"], [5, 35, 65] + ) + self.assertTrue( + analyze(windows([100] * 5 + [99, 100, 101]))["strict_monotonic_reference_trend"] + ) + self.assertFalse( + analyze(windows([100] * 8, [2, 2, 2, 2, 3, 2, 2, 2]))["screened_candidate_seconds"] + ) + # Both bands and within-candidate range matter: +/-4% fits bands but range8% fails. + self.assertFalse( + analyze(windows([96, 104, 96, 104, 96, 100, 100, 100]))["screened_candidate_seconds"] + ) + for invalid in ([100] * 7, [100] * 7 + [float("nan")], [100] * 7 + [True]): + with self.assertRaises(BenchmarkFailure): + analyze(windows(invalid)) + + def test_decimal_threshold_roundoff_passes_but_material_excess_fails(self): + def windows(rps, p95): + return [{"requests_per_second": r, "p95_response_time_ms": p} for r, p in zip(rps, p95)] + + analyze = self.module.analyze_windows + for rps, p95 in ( + ([100] * 8, [2.2] * 5 + [2] * 3), + ([1.05] * 5 + [1] * 3, [2] * 8), + ([1] * 5 + [0.975, 1, 1.025], [2] * 5 + [1.9, 2, 2.1]), + ): + self.assertEqual(analyze(windows(rps, p95))["screened_candidate_seconds"], [5, 35, 65]) + self.assertFalse( + analyze(windows([1.05000001] * 5 + [1] * 3, [2] * 8))["screened_candidate_seconds"] + ) + self.assertFalse( + analyze(windows([100] * 8, [2.20000001] * 5 + [2] * 3))["screened_candidate_seconds"] + ) + + def test_analysis_inclusive_thresholds_and_unstable_reference(self): + def windows(values): + return [{"requests_per_second": r, "p95_response_time_ms": 2} for r in values] + + self.assertEqual( + self.module.analyze_windows(windows([95] * 5 + [100] * 3))[ + "screened_candidate_seconds" + ], + [5, 35, 65], + ) + self.assertFalse( + self.module.analyze_windows(windows([94.99] * 5 + [100] * 3))[ + "screened_candidate_seconds" + ] + ) + self.assertFalse( + self.module.analyze_windows(windows([100] * 5 + [90, 100, 110]))[ + "screened_candidate_seconds" + ] + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_workflows.py b/tests/test_workflows.py index f1cebbf..1be82dd 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -26,7 +26,14 @@ def test_only_approved_workflows_use_pinned_actions_and_safe_checkouts(self): paths = list((ROOT / ".github/workflows").glob("*.yml")) self.assertEqual( {p.name for p in paths}, - {"ci.yml", "benchmark.yml", "jvm-diagnostic.yml", "pages.yml", "pages-deploy.yml"}, + { + "ci.yml", + "benchmark.yml", + "jvm-diagnostic.yml", + "common-warmup-screen.yml", + "pages.yml", + "pages-deploy.yml", + }, ) for path in paths: workflow = load(path.name) @@ -101,6 +108,44 @@ def test_jvm_diagnostic_actions_match_existing_pins_and_upload_failures(self): for step in steps: self.assertNotIn("continue-on-error", step) + def test_common_warmup_screen_is_separate_manual_bounded_and_nonpublishing(self): + workflow = load("common-warmup-screen.yml") + self.assertEqual(workflow.get("on"), {"workflow_dispatch": ""}) + self.assertEqual(workflow["permissions"], {"contents": "read"}) + self.assertEqual( + workflow["concurrency"], + {"group": "common-warmup-screen", "cancel-in-progress": "false"}, + ) + self.assertEqual(set(workflow["jobs"]), {"screen"}) + job = workflow["jobs"]["screen"] + previous = load("jvm-diagnostic.yml")["jobs"]["diagnose"] + self.assertEqual(job["if"], previous["if"]) + self.assertEqual(job["timeout-minutes"], "120") + self.assertNotIn("strategy", job) + self.assertNotIn("permissions", job) + self.assertEqual(job["env"], {"PYTHONDONTWRITEBYTECODE": "1"}) + steps = job["steps"] + self.assertEqual(len(steps), 4) + self.assertEqual(steps[:2], previous["steps"][:2]) + self.assertEqual( + steps[2]["run"], + "timeout --signal=TERM --kill-after=120s 104m python -m benchmark.common_warmup_screen", + ) + self.assertEqual(steps[3]["if"], "always()") + self.assertEqual(steps[3]["timeout-minutes"], "7") + self.assertEqual( + steps[3]["with"], + { + "name": "common-warmup-screen-${{ github.run_id }}-${{ github.run_attempt }}", + "path": ".cache/common-warmup-screen/", + "include-hidden-files": "true", + "retention-days": "90", + "if-no-files-found": "error", + }, + ) + for step in steps: + self.assertNotIn("continue-on-error", step) + def test_pr_ci_is_split_read_only_and_never_publishes(self): ci = load("ci.yml") self.assertEqual(ci["permissions"], {"contents": "read"})